fix uploading to lebesgue

Author: KZHIWEI

dpdispatcher/dp_cloud_server.py

@@ -119,13 +119,13 @@ def check_status(self, job):
         # print("api",self.api_version,self.input_data.get('job_group_id'),job.job_id)
         check_return = self.api.get_tasks(job_id,group_id)
         try:
-            dp_job_status = check_return[0]["status"]
+            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")
             time.sleep(60)
             retry_return = self.api.get_tasks(job_id, group_id)
             try:
-                dp_job_status = retry_return[0]["status"]
+                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}")
 

dpdispatcher/dp_cloud_server_context.py

@@ -91,7 +91,14 @@ def upload(self, submission):
         # zip_path = "/home/felix/workplace/22_dpdispatcher/dpdispatcher-yfb/dpdispatcher/dpcloudserver/t.txt"
         # zip_path = self.local_root
         bar_format = "{l_bar}{bar}| {n:.02f}/{total:.02f} %  [{elapsed}<{remaining}, {rate_fmt}{postfix}]"
-        for job in tqdm.tqdm(submission.belonging_jobs, desc="Uploading to Lebesgue", bar_format=bar_format):
+        job_to_be_uploaded = []
+        result = None
+        for job in submission.belonging_jobs:
+            if not self.api.check_job_has_uploaded(job.job_id):
+                job_to_be_uploaded.append(job)
+        if len(job_to_be_uploaded) == 0:
+            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)
@@ -143,7 +150,7 @@ 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="Downloading to Lebesgue", bar_format=bar_format):
+        for job_hash, info in tqdm.tqdm(job_infos.items(), desc="Downloading from Lebesgue", bar_format=bar_format):
             result_filename = job_hash + '_back.zip'
             target_result_zip = os.path.join(self.local_root, result_filename)
             self.api.download_from_url(info['result_url'], target_result_zip)

dpdispatcher/dpcloudserver/api.py

@@ -210,10 +210,10 @@ def get_tasks(self, job_id, group_id, page=1, per_page=10):
         )
         for each in ret['items']:
             if job_id == each["task_id"]:
-                return [each]
+                return each
         if len(ret['items']) != 0:
             return self.get_tasks(job_id, group_id, page=page + 1)
-        return []
+        return None
 
     def get_tasks_list(self, group_id, per_page=30):
         result = []
@@ -233,4 +233,22 @@ def get_tasks_list(self, group_id, per_page=30):
             page += 1
         return result
 
+    def check_job_has_uploaded(self, job_id):
+        try:
+            if not job_id:
+                return False
+            if 'job_group_id' in job_id:
+                ids = job_id.split(":job_group_id:")
+                job_id, _ = int(ids[0]), int(ids[1])
+            ret = self.get(f'data/job/{job_id}', {})
+            if len(ret) == 0:
+                return False
+            if ret.get('input_data'):
+                return True
+            else:
+                return False
+        except ValueError as e:
+            dlog.error(e)
+            return False
+
 # %%