Merge branch 'deepmodeling:master' into master

Author: felix5572

dpdispatcher/dp_cloud_server.py

@@ -90,7 +90,7 @@ def do_submit(self, job):
         input_data['command'] = f"bash {job.script_file_name}"
         # input_data['backward_files'] = self._gen_backward_files_list(job)
         if self.context.remote_profile.get('program_id') is None:
-            warnings.warn('program_id will be compulsory in the future.')
+            warnings.warn('program_id is compulsory.')
         job_id, group_id = self.api.job_create(
             job_type=input_data['job_type'],
             oss_path=input_data['job_resources'],
@@ -124,13 +124,13 @@ def check_status(self, job):
         try:
             dp_job_status = check_return["status"]
         except IndexError as e:
-            dlog.error(f"cannot find job information in check_return. job {job.job_id}. check_return:{check_return}; retry one more time after 60 seconds")
+            dlog.error(f"cannot find job information in bohrium for job {job.job_id}. check_return:{check_return}; retry one more time after 60 seconds")
             time.sleep(60)
             retry_return = self.api.get_tasks(job_id, group_id)
             try:
                 dp_job_status = retry_return["status"]
             except IndexError as e:
-                raise RuntimeError(f"cannot find job information in dpcloudserver's database for job {job.job_id} {check_return} {retry_return}")
+                raise RuntimeError(f"cannot find job information in bohrium for job {job.job_id} {check_return} {retry_return}")
 
         job_state = self.map_dp_job_state(dp_job_status)
         if job_state == JobStatus.finished:

dpdispatcher/dp_cloud_server_context.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python
 # coding: utf-8
 # %%
+import time
 import uuid
 
 from dargs.dargs import Argument
@@ -14,7 +15,10 @@
 from .dpcloudserver import zip_file
 import shutil
 import tqdm
+
 # from zip_file import zip_files
+from .dpcloudserver.config import ALI_OSS_BUCKET_URL
+
 DP_CLOUD_SERVER_HOME_DIR = os.path.join(
     os.path.expanduser('~'),
     '.dpdispatcher/',
@@ -23,14 +27,15 @@
 ENDPOINT = 'http://oss-cn-shenzhen.aliyuncs.com'
 BUCKET_NAME = 'dpcloudserver'
 
+
 class DpCloudServerContext(BaseContext):
-    def __init__ (self,
-        local_root,
-        remote_root=None,
-        remote_profile={},
-        *args,
-        **kwargs,
-    ):
+    def __init__(self,
+                 local_root,
+                 remote_root=None,
+                 remote_profile={},
+                 *args,
+                 **kwargs,
+                 ):
         self.init_local_root = local_root
         self.init_remote_root = remote_root
         self.temp_local_root = os.path.abspath(local_root)
@@ -83,6 +88,43 @@ def _gen_oss_path(self, job, zip_filename):
             setattr(job, 'upload_path', path)
             return path
 
+    def upload_job(self, job, common_files=None):
+        MAX_RETRY = 3
+        if common_files is None:
+            common_files = []
+        self.machine.gen_local_script(job)
+        zip_filename = job.job_hash + '.zip'
+        oss_task_zip = self._gen_oss_path(job, zip_filename)
+        zip_task_file = os.path.join(self.local_root, zip_filename)
+
+        upload_file_list = [job.script_file_name, ]
+        upload_file_list.extend(common_files)
+
+        for task in job.job_task_list:
+            for file in task.forward_files:
+                upload_file_list.append(
+                    os.path.join(
+                        task.task_work_path, file
+                    )
+                )
+
+        upload_zip = zip_file.zip_file_list(
+            self.local_root,
+            zip_task_file,
+            file_list=upload_file_list
+        )
+        result = self.api.upload(oss_task_zip, upload_zip, ENDPOINT, BUCKET_NAME)
+        retry_count = 0
+        while True:
+            if self.api.check_file_has_uploaded(ALI_OSS_BUCKET_URL + oss_task_zip):
+                self._backup(self.local_root, upload_zip)
+                break
+            elif retry_count < MAX_RETRY:
+                time.sleep(1 + retry_count)
+                retry_count += 1
+            else:
+                raise ValueError(f"upload retried excess {MAX_RETRY} terminate.")
+
     def upload(self, submission):
         # oss_task_dir = os.path.join('%s/%s/%s.zip' % ('indicate', file_uuid, file_uuid))
         # zip_filename = submission.submission_hash + '.zip'
@@ -100,30 +142,8 @@ def upload(self, submission):
         if len(job_to_be_uploaded) == 0:
             dlog.info("all job has been uploaded, continue")
             return result
-        for job in tqdm.tqdm(job_to_be_uploaded, desc="Uploading to Lebesgue", bar_format=bar_format):
-            self.machine.gen_local_script(job)
-            zip_filename = job.job_hash + '.zip'
-            oss_task_zip = self._gen_oss_path(job, zip_filename)
-            zip_task_file = os.path.join(self.local_root, zip_filename)
-
-            upload_file_list = [job.script_file_name, ]
-            upload_file_list.extend(submission.forward_common_files)
-
-            for task in job.job_task_list:
-                for file in task.forward_files:
-                    upload_file_list.append(
-                        os.path.join(
-                            task.task_work_path, file
-                        )
-                    )
-
-            upload_zip = zip_file.zip_file_list(
-                self.local_root,
-                zip_task_file,
-                file_list=upload_file_list
-            )
-            result = self.api.upload(oss_task_zip, upload_zip, ENDPOINT, BUCKET_NAME)
-            self._backup(self.local_root, upload_zip)
+        for job in tqdm.tqdm(job_to_be_uploaded, desc="Uploading to Lebesgue", bar_format=bar_format, leave=False):
+            self.upload_job(job, submission.forward_common_files)
         return result
         # return oss_task_zip
         # api.upload(self.oss_task_dir, zip_task_file)
@@ -151,7 +171,8 @@ def download(self, submission):
                         job_hash = job_hashs[each['task_id']]
                     job_infos[job_hash] = each
         bar_format = "{l_bar}{bar}| {n:.02f}/{total:.02f} %  [{elapsed}<{remaining}, {rate_fmt}{postfix}]"
-        for job_hash, info in tqdm.tqdm(job_infos.items(), desc="Validating download file from Lebesgue", bar_format=bar_format):
+        for job_hash, info in tqdm.tqdm(job_infos.items(), desc="Validating download file from Lebesgue",
+                                        bar_format=bar_format, leave=False):
             result_filename = job_hash + '_back.zip'
             target_result_zip = os.path.join(self.local_root, result_filename)
             if self._check_if_job_has_already_downloaded(target_result_zip, self.local_root):
@@ -234,7 +255,7 @@ def clean(self):
     #         retcode = cmd_pipes['stdout'].channel.recv_exit_status()
     #         return retcode, cmd_pipes['stdout'], cmd_pipes['stderr']
 
-    def kill(self, cmd_pipes) :
+    def kill(self, cmd_pipes):
         pass
 
     @classmethod
@@ -251,11 +272,12 @@ def machine_subfields(cls) -> List[Argument]:
             Argument("email", str, optional=False, doc="Email"),
             Argument("password", str, optional=False, doc="Password"),
             Argument("program_id", int, optional=False, doc="Program ID"),
+            Argument("keep_backup", bool, optional=True, doc="keep download and upload zip"),
             Argument("input_data", dict, optional=False, doc="Configuration of job"),
         ], doc=doc_remote_profile)]
 
 
 class LebesgueContext(DpCloudServerContext):
     pass
 
-#%%
+# %%

dpdispatcher/dpcloudserver/api.py

@@ -13,7 +13,9 @@
 from dpdispatcher import dlog
 
 from .retcode import RETCODE
-from .config import HTTP_TIME_OUT, API_HOST
+from .config import HTTP_TIME_OUT, API_HOST, API_LOGGER_STACK_INFO
+
+ENABLE_STACK = True if API_LOGGER_STACK_INFO else False
 
 
 class API:
@@ -35,12 +37,13 @@ def get(self, url, params, retry=0):
                     headers=headers
                 )
             except Exception as e:
-                dlog.error(f"request error {e}")
+                dlog.error(f"request error {e}", stack_info=ENABLE_STACK)
                 continue
             if ret.ok:
                 break
             else:
-                dlog.error(f"request error status_code:{ret.status_code} reason: {ret.reason} body: \n{ret.text}")
+                dlog.error(f"request error status_code:{ret.status_code} reason: {ret.reason} body: \n{ret.text}",
+                           stack_info=ENABLE_STACK)
                 time.sleep(retry_count * 10)
         if ret is None:
             raise ConnectionError("request fail")
@@ -69,7 +72,7 @@ def post(self, url, params, retry=0):
                     headers=headers
                 )
             except Exception as e:
-                dlog.error(f"request error {e}")
+                dlog.error(f"request error {e}", stack_info=ENABLE_STACK)
                 continue
             if ret.ok:
                 break
@@ -132,7 +135,7 @@ def download_from_url(self, url, save_file):
                     stream=True
                 )
             except Exception as e:
-                dlog.error(f"request error {e}")
+                dlog.error(f"request error {e}", stack_info=ENABLE_STACK)
                 continue
             if ret.ok:
                 break
@@ -147,7 +150,6 @@ def download_from_url(self, url, save_file):
                     f.write(chunk)
             ret.close()
 
-
     def upload(self, oss_task_zip, zip_task_file, endpoint, bucket_name):
         dlog.debug(f"debug: upload: oss_task_zip:{oss_task_zip}; zip_task_file:{zip_task_file}")
         bucket = self._get_oss_bucket(endpoint, bucket_name)
@@ -170,7 +172,6 @@ def upload(self, oss_task_zip, zip_task_file, endpoint, bucket_name):
         # print('debug:upload_result:', result, dir())
         return result
 
-
     def job_create(self, job_type, oss_path, input_data, program_id=None, group_id=None):
         post_data = {
             'job_type': job_type,
@@ -244,11 +245,23 @@ def check_job_has_uploaded(self, job_id):
             if len(ret) == 0:
                 return False
             if ret.get('input_data'):
-                return True
+                return self.check_file_has_uploaded(ret.get('input_data'))
             else:
                 return False
         except ValueError as e:
-            dlog.error(e)
+            dlog.error(e, stack_info=ENABLE_STACK)
+            return False
+
+    def check_file_has_uploaded(self, file_url):
+        try:
+            if not file_url:
+                return False
+            resp = requests.head(file_url)
+            if resp.ok:
+                return True
+            return False
+        except Exception as e:
+            dlog.error(e, stack_info=ENABLE_STACK)
             return False
 
     def get_job_result_url(self, job_id):
@@ -264,7 +277,7 @@ def get_job_result_url(self, job_id):
             else:
                 return None
         except ValueError as e:
-            dlog.error(e)
+            dlog.error(e, stack_info=ENABLE_STACK)
             return None
 
 # %%

dpdispatcher/dpcloudserver/config.py

@@ -1,8 +1,8 @@
 import os
 HTTP_TIME_OUT = 30
 
-API_HOST = os.environ.get('DPDISPATCHER_LEBESGUE_API_HOST', "https://lebesgue.dp.tech")
-
+API_HOST = os.environ.get('DPDISPATCHER_LEBESGUE_API_HOST', "https://bohrium.dp.tech/")
+API_LOGGER_STACK_INFO = os.environ.get('API_LOGGER_STACK_INFO', "")
 ALI_STS_ENDPOINT = os.environ.get('DPDISPATCHER_LEBESGUE_ALI_STS_ENDPOINT', 'http://oss-cn-shenzhen.aliyuncs.com')
 ALI_STS_BUCKET_NAME = os.environ.get('DPDISPATCHER_LEBESGUE_ALI_STS_BUCKET_NAME', "dpcloudserver")
 ALI_OSS_BUCKET_URL = os.environ.get('DPDISPATCHER_LEBESGUE_ALI_OSS_BUCKET_URL', "https://dpcloudserver.oss-cn-shenzhen.aliyuncs.com/")

dpdispatcher/machine.py

@@ -334,7 +334,7 @@ def arginfo(cls):
             Argument("batch_type", str, optional=False, doc=doc_batch_type),
             # TODO: add default to local_root and remote_root after refactor the code
             Argument("local_root", [str, None], optional=False, doc=doc_local_root),
-            Argument("remote_root", str, optional=True, doc=doc_remote_root),
+            Argument("remote_root", [str, None], optional=True, doc=doc_remote_root),
             Argument("clean_asynchronously", bool, optional=True, default=False, doc=doc_clean_asynchronously),
         ]
 

dpdispatcher/pbs.py

@@ -114,3 +114,16 @@ def check_status(self, job):
                 return JobStatus.terminated
         else :
             return JobStatus.unknown
+
+    def gen_script_header(self, job):
+        # ref: https://support.adaptivecomputing.com/wp-content/uploads/2021/02/torque/torque.htm#topics/torque/2-jobs/requestingRes.htm
+        resources = job.resources
+        pbs_script_header_dict= {}
+        pbs_script_header_dict['select_node_line']="#PBS -l nodes={number_node}:ppn={cpu_per_node}".format(
+            number_node=resources.number_node, cpu_per_node=resources.cpu_per_node
+        )
+        if (resources.gpu_per_node != 0):
+            pbs_script_header_dict['select_node_line'] += ":gpus={gpu_per_node}".format(gpu_per_node=resources.gpu_per_node)
+        pbs_script_header_dict['queue_name_line']="#PBS -q {queue_name}".format(queue_name=resources.queue_name)
+        pbs_script_header = pbs_script_header_template.format(**pbs_script_header_dict)
+        return pbs_script_header

dpdispatcher/submission.py

@@ -130,13 +130,13 @@ def serialize(self, if_static=False):
 
     def register_task(self, task):
         if self.belonging_jobs:
-            raise RuntimeError("Not allowed to register tasks after generating jobs."
+            raise RuntimeError("Not allowed to register tasks after generating jobs. "
                     "submission hash error {self}".format(self=self))
         self.belonging_tasks.append(task)
 
     def register_task_list(self, task_list):
         if self.belonging_jobs:
-            raise RuntimeError("Not allowed to register tasks after generating jobs."
+            raise RuntimeError("Not allowed to register tasks after generating jobs. "
                     "submission hash error {self}".format(self=self))
         self.belonging_tasks.extend(task_list)
 
@@ -245,8 +245,8 @@ def handle_unexpected_submission_state(self):
                 f"Meet errors will handle unexpected submission state.\n"
                 f"Debug information: remote_root=={self.machine.context.remote_root}.\n"
                 f"Debug information: submission_hash=={self.submission_hash}.\n"
-                f"Please check the dirs and scripts in remote_root"
-                f"The job information mentioned above may help"
+                f"Please check the dirs and scripts in remote_root. "
+                f"The job information mentioned above may help."
             ) from e
 
     # not used here, submitting job is in handle_unexpected_submission_state.
@@ -840,10 +840,10 @@ def arginfo(detail_kwargs=True):
         strategy_format = Argument("strategy", dict, strategy_args, optional=True, doc=doc_strategy)
 
         resources_args = [
-            Argument("number_node", int, optional=False, doc=doc_number_node),
-            Argument("cpu_per_node", int, optional=False, doc=doc_cpu_per_node),
-            Argument("gpu_per_node", int, optional=False, doc=doc_gpu_per_node),
-            Argument("queue_name", str, optional=False, doc=doc_queue_name),
+            Argument("number_node", int, optional=True, doc=doc_number_node, default=1),
+            Argument("cpu_per_node", int, optional=True, doc=doc_cpu_per_node, default=1),
+            Argument("gpu_per_node", int, optional=True, doc=doc_gpu_per_node, default=0),
+            Argument("queue_name", str, optional=True, doc=doc_queue_name, default=""),
             Argument("group_size", int, optional=False, doc=doc_group_size),
 
             Argument("custom_flags", list, optional=True, doc=doc_custom_flags),