Merge pull request #410 from HSF/dev

Dev

Author: wguanicedew

doma/lib/idds/doma/workflowv2/domapandawork.py

@@ -165,16 +165,58 @@ def get_site(self):
             return self.queue
         return self.task_cloud
 
+    @property
+    def num_inputs(self):
+        num = self.get_metadata_item('num_inputs', None)
+        return num
+
+    @num_inputs.setter
+    def num_inputs(self, value):
+        self.add_metadata_item('num_inputs', value)
+
+    @property
+    def num_dependencies(self):
+        num = self.get_metadata_item('num_dependencies', None)
+        return num
+
+    @num_dependencies.setter
+    def num_dependencies(self, value):
+        self.add_metadata_item('num_dependencies', value)
+
+    def count_dependencies(self, data):
+        if self.num_dependencies is not None and self.num_inputs is not None:
+            return self.num_inputs, self.num_dependencies
+
+        num_inputs = 0
+        num_dependencies = 0
+        try:
+            for item in data:
+                # item_name = item['name']
+                inputs_dependency = item["dependencies"]
+                num_dependencies += len(inputs_dependency)
+                num_inputs += 1
+        except Exception as ex:
+            self.logger.warn(f"Failed to count dependencies: {ex}")
+        return num_inputs, num_dependencies
+
     @property
     def dependency_map(self):
         if self.should_unzip('_dependency_map'):
             data = self.unzip_data(self._dependency_map)
+            num_inputs, num_dependencies = self.count_dependencies(data)
+            self.num_inputs = num_inputs
+            self.num_dependencies = num_dependencies
             return data
 
+        num_inputs, num_dependencies = self.count_dependencies(self._dependency_map)
+        self.num_inputs = num_inputs
+        self.num_dependencies = num_dependencies
         return self._dependency_map
 
     @dependency_map.setter
     def dependency_map(self, value):
+        num_dependencies = 0
+        num_inputs = 0
         if value:
             if type(value) not in [list, tuple]:
                 raise exceptions.IDDSException("dependency_map should be a list or tuple")
@@ -184,6 +226,8 @@ def dependency_map(self, value):
                 if len(item_name) > self.max_name_length:
                     raise exceptions.IDDSException("The file name is long (%s), which is bigger than the maximum name length (%s)" % (len(item_name), self.max_name_length))
                 inputs_dependency = item["dependencies"]
+                num_inputs += 1
+                num_dependencies += len(inputs_dependency)
                 if item_name not in item_names:
                     item_names[item_name] = item
                 else:
@@ -201,6 +245,9 @@ def dependency_map(self, value):
 
         self._dependency_map = value
 
+        self.num_inputs = num_inputs
+        self.num_dependencies = num_dependencies
+
         if self.es:
             self.construct_es_files()
 
@@ -1458,7 +1505,7 @@ def get_event_job(self, sub_map_id, panda_jobs, job_set_events):
             ret_job = item.get('job', None)
         return ret_event, ret_job
 
-    def get_update_contents(self, unterminated_jobs_status, input_output_maps, contents_ext, job_info_maps, abort=False, log_prefix=''):
+    def get_update_contents(self, unterminated_jobs_status, input_output_maps, contents_ext, job_info_maps, abort=False, terminated_status=False, log_prefix=''):
         inputname_to_map_id_outputs = {}
         for map_id in input_output_maps:
             inputs = input_output_maps[map_id]['inputs']
@@ -1781,7 +1828,7 @@ def get_update_contents(self, unterminated_jobs_status, input_output_maps, conte
                                     update_content_ext['exe_exit_diag'] = event_error_diag
                                 update_contents_ext.append(update_content_ext)
 
-        if abort:
+        if abort or terminated_status:
             for map_id in input_output_maps:
                 outputs = input_output_maps[map_id]['outputs']
                 for content in outputs:
@@ -1847,7 +1894,11 @@ def poll_panda_task(self, processing=None, input_output_maps=None, contents_ext=
                     abort_status = False
                     if processing_status in [ProcessingStatus.Cancelled]:
                         abort_status = True
-                    ret_contents = self.get_update_contents(unterminated_jobs_status, input_output_maps, contents_ext, job_info_maps, abort=abort_status, log_prefix=log_prefix)
+                    terminated_status = False
+                    if processing_status in [ProcessingStatus.Cancelled, ProcessingStatus.Failed, ProcessingStatus.Broken]:
+                        terminated_status = True
+                    ret_contents = self.get_update_contents(unterminated_jobs_status, input_output_maps, contents_ext, job_info_maps,
+                                                            abort=abort_status, terminated_status=terminated_status, log_prefix=log_prefix)
                     updated_contents, update_contents_full, new_contents_ext, update_contents_ext = ret_contents
 
                     return processing_status, updated_contents, update_contents_full, new_contents_ext, update_contents_ext

main/lib/idds/agents/carrier/poller.py

@@ -603,7 +603,9 @@ def process_update_processing(self, event):
 
     def clean_locks(self):
         self.logger.info("clean locking")
-        core_processings.clean_locking(time_period=self.locking_period)
+        health_items = self.get_health_items()
+        min_request_id = BaseAgent.min_request_id
+        core_processings.clean_locking(health_items=health_items, min_request_id=min_request_id, time_period=None)
 
     def init_event_function_map(self):
         self.event_func_map = {

main/lib/idds/agents/carrier/utils.py

@@ -184,7 +184,7 @@ def get_collection_ids(collections):
     return coll_ids
 
 
-def get_input_output_maps(transform_id, work):
+def get_input_output_maps(transform_id, work, with_deps=True):
     # link collections
     input_collections = work.get_input_collections()
     output_collections = work.get_output_collections()
@@ -202,7 +202,8 @@ def get_input_output_maps(transform_id, work):
                                                                                input_coll_ids=input_coll_ids,
                                                                                output_coll_ids=output_coll_ids,
                                                                                log_coll_ids=log_coll_ids,
-                                                                               with_sub_map_id=work.with_sub_map_id())
+                                                                               with_sub_map_id=work.with_sub_map_id(),
+                                                                               with_deps=with_deps)
 
     # work_name_to_coll_map = core_transforms.get_work_name_to_coll_map(request_id=transform['request_id'])
     # work.set_work_name_to_coll_map(work_name_to_coll_map)
@@ -603,7 +604,7 @@ def handle_new_processing(processing, agent_attributes, func_site_to_cloud=None,
             update_collections.append(u_coll)
 
     if proc.submitted_at:
-        input_output_maps = get_input_output_maps(transform_id, work)
+        input_output_maps = get_input_output_maps(transform_id, work, with_deps=False)
         new_input_output_maps = work.get_new_input_output_maps(input_output_maps)
         request_id = processing['request_id']
         transform_id = processing['transform_id']
@@ -700,7 +701,7 @@ def get_input_output_sub_maps(inputs, outputs, inputs_dependency, logs=[]):
     return input_output_sub_maps
 
 
-def get_updated_contents_by_input_output_maps(input_output_maps=None, terminated=False, max_updates_per_round=2000, logger=None, log_prefix=''):
+def get_updated_contents_by_input_output_maps(input_output_maps=None, terminated=False, max_updates_per_round=2000, with_deps=False, logger=None, log_prefix=''):
     updated_contents, updated_contents_full_input, updated_contents_full_output = [], [], []
     updated_contents_full_input_deps = []
     new_update_contents = []
@@ -762,10 +763,12 @@ def get_updated_contents_by_input_output_maps(input_output_maps=None, terminated
             inputs_dependency_sub = input_output_sub_maps[sub_map_id]['inputs_dependency']
 
             input_content_update_status = None
-            if is_all_contents_available(inputs_dependency_sub):
-                input_content_update_status = ContentStatus.Available
-            elif is_all_contents_terminated(inputs_dependency_sub, terminated):
-                input_content_update_status = ContentStatus.Missing
+            if with_deps:
+                # if deps are not loaded. This part should not be executed. Otherwise it will release all jobs
+                if is_all_contents_available(inputs_dependency_sub):
+                    input_content_update_status = ContentStatus.Available
+                elif is_all_contents_terminated(inputs_dependency_sub, terminated):
+                    input_content_update_status = ContentStatus.Missing
             if input_content_update_status:
                 for content in inputs_sub:
                     if content['substatus'] != input_content_update_status:
@@ -1187,7 +1190,7 @@ def handle_update_processing(processing, agent_attributes, max_updates_per_round
     work = proc.work
     work.set_agent_attributes(agent_attributes, processing)
 
-    input_output_maps = get_input_output_maps(transform_id, work)
+    input_output_maps = get_input_output_maps(transform_id, work, with_deps=False)
     logger.debug(log_prefix + "get_input_output_maps: len: %s" % len(input_output_maps))
     logger.debug(log_prefix + "get_input_output_maps.keys[:3]: %s" % str(list(input_output_maps.keys())[:3]))
 
@@ -1414,11 +1417,35 @@ def handle_trigger_processing(processing, agent_attributes, trigger_new_updates=
     request_id = processing['request_id']
     transform_id = processing['transform_id']
     workload_id = processing['workload_id']
+    processing_id = processing['processing_id']
 
     proc = processing['processing_metadata']['processing']
     work = proc.work
     work.set_agent_attributes(agent_attributes, processing)
 
+    num_dependencies = None
+    num_inputs = None
+    default_input_dep_page_size = 500
+    min_input_dep_page_size = 100
+    max_dependencies = 5000
+    try:
+        num_inputs = work.num_inputs
+        num_dependencies = work.num_dependencies
+        if num_inputs is not None and num_dependencies is not None and num_dependencies > 0:
+            input_dep_page_size = int(max_dependencies * num_inputs / num_dependencies)
+            if input_dep_page_size < default_input_dep_page_size:
+                default_input_dep_page_size = input_dep_page_size
+                log_info = f"input_dep_page_size ({input_dep_page_size}) is smaller than default_input_dep_page_size ({default_input_dep_page_size}),"
+                log_info = "update default_input_dep_page_size from input_dep_page_size"
+                logger.info(log_info)
+            if default_input_dep_page_size < min_input_dep_page_size:
+                log_info = f"default_input_dep_page_size ({default_input_dep_page_size}) is smaller than min_input_dep_page_size ({min_input_dep_page_size}),"
+                log_info = "update default_input_dep_page_size from min_input_dep_page_size"
+                logger.info(log_info)
+                default_input_dep_page_size = min_input_dep_page_size
+    except Exception as ex:
+        logger.warn(f"request_id ({request_id}) transform_id ({transform_id}) processing_id ({processing_id}) fails to get num_dependencies: {ex}")
+
     if (not work.use_dependency_to_release_jobs()) or workload_id is None:
         return processing['substatus'], [], [], {}, {}, {}, [], [], has_updates
     else:
@@ -1467,6 +1494,7 @@ def handle_trigger_processing(processing, agent_attributes, trigger_new_updates=
         core_catalog.delete_contents_update(request_id=request_id, transform_id=transform_id, fetch=True)
         logger.debug(log_prefix + "sync contents_update to contents done")
 
+        """
         logger.debug(log_prefix + "update_contents_from_others_by_dep_id")
         # core_catalog.update_contents_from_others_by_dep_id(request_id=request_id, transform_id=transform_id)
         to_triggered_contents = core_catalog.get_update_contents_from_others_by_dep_id(request_id=request_id, transform_id=transform_id)
@@ -1490,19 +1518,38 @@ def handle_trigger_processing(processing, agent_attributes, trigger_new_updates=
             wait_futures_finish(ret_futures, "update_contents_from_others_by_dep_id", logger, log_prefix)
 
         logger.debug(log_prefix + "update_contents_from_others_by_dep_id done")
+        """
 
-        input_output_maps = get_input_output_maps(transform_id, work)
-        logger.debug(log_prefix + "input_output_maps.keys[:2]: %s" % str(list(input_output_maps.keys())[:2]))
+        logger.debug(log_prefix + "update_contents_from_others_by_dep_id_pages")
+        status_not_to_check = [ContentStatus.Available, ContentStatus.FakeAvailable,
+                               ContentStatus.FinalFailed, ContentStatus.Missing]
+        core_catalog.update_contents_from_others_by_dep_id_pages(request_id=request_id, transform_id=transform_id,
+                                                                 page_size=1000, status_not_to_check=status_not_to_check)
+        logger.debug(log_prefix + "update_contents_from_others_by_dep_id_pages done")
 
         terminated_processing = False
         terminated_status = [ProcessingStatus.Finished, ProcessingStatus.Failed, ProcessingStatus.SubFinished,
                              ProcessingStatus.Terminating, ProcessingStatus.Cancelled]
         if processing['status'] in terminated_status or processing['substatus'] in terminated_status:
             terminated_processing = True
 
+        logger.debug(log_prefix + "update_input_contents_by_dependency_pages")
+        status_not_to_check = [ContentStatus.Available, ContentStatus.FakeAvailable,
+                               ContentStatus.FinalFailed, ContentStatus.Missing]
+        core_catalog.update_input_contents_by_dependency_pages(request_id=request_id, transform_id=transform_id,
+                                                               page_size=default_input_dep_page_size,
+                                                               terminated=terminated_processing,
+                                                               batch_size=1000, status_not_to_check=status_not_to_check)
+        logger.debug(log_prefix + "update_input_contents_by_dependency_pages done")
+
+        with_deps = False
+        input_output_maps = get_input_output_maps(transform_id, work, with_deps=with_deps)
+        logger.debug(log_prefix + "input_output_maps.keys[:2]: %s" % str(list(input_output_maps.keys())[:2]))
+
         updated_contents_ret_chunks = get_updated_contents_by_input_output_maps(input_output_maps=input_output_maps,
                                                                                 terminated=terminated_processing,
                                                                                 max_updates_per_round=max_updates_per_round,
+                                                                                with_deps=with_deps,
                                                                                 logger=logger,
                                                                                 log_prefix=log_prefix)
 
@@ -1913,7 +1960,7 @@ def handle_messages_processing(messages, logger=None, log_prefix='', update_proc
 def sync_collection_status(request_id, transform_id, workload_id, work, input_output_maps=None,
                            close_collection=False, force_close_collection=False, abort=False, terminate=False):
     if input_output_maps is None:
-        input_output_maps = get_input_output_maps(transform_id, work)
+        input_output_maps = get_input_output_maps(transform_id, work, with_deps=False)
 
     all_updates_flushed = True
     coll_status = {}
@@ -2116,7 +2163,7 @@ def sync_processing(processing, agent_attributes, terminate=False, abort=False,
     work.set_agent_attributes(agent_attributes, processing)
 
     messages = []
-    input_output_maps = get_input_output_maps(transform_id, work)
+    input_output_maps = get_input_output_maps(transform_id, work, with_deps=False)
     if processing['substatus'] in terminated_status or processing['substatus'] in terminated_status:
         terminate = True
     update_collections, all_updates_flushed, msgs = sync_collection_status(request_id, transform_id, workload_id, work,
@@ -2225,7 +2272,7 @@ def handle_resume_processing(processing, agent_attributes, logger=None, log_pref
                         'substatus': CollectionStatus.Open}
         update_collections.append(u_collection)
 
-    input_output_maps = get_input_output_maps(transform_id, work)
+    input_output_maps = get_input_output_maps(transform_id, work, with_deps=False)
     update_contents = reactive_contents(request_id, transform_id, workload_id, work, input_output_maps)
 
     processing['status'] = ProcessingStatus.Running

main/lib/idds/agents/clerk/clerk.py

@@ -1890,7 +1890,9 @@ def process_resume_request(self, event):
 
     def clean_locks(self):
         self.logger.info("clean locking")
-        core_requests.clean_locking()
+        health_items = self.get_health_items()
+        min_request_id = BaseAgent.min_request_id
+        core_requests.clean_locking(health_items=health_items, min_request_id=min_request_id, time_period=None)
 
     def init_event_function_map(self):
         self.event_func_map = {
@@ -1942,7 +1944,7 @@ def run(self):
             self.add_task(task)
             task = self.create_task(task_func=self.clean_min_request_id, task_output_queue=None, task_args=tuple(), task_kwargs={}, delay_time=3600, priority=1)
             self.add_task(task)
-            task = self.create_task(task_func=self.clean_locks, task_output_queue=None, task_args=tuple(), task_kwargs={}, delay_time=1800, priority=1)
+            task = self.create_task(task_func=self.clean_locks, task_output_queue=None, task_args=tuple(), task_kwargs={}, delay_time=60, priority=1)
             self.add_task(task)
 
             self.execute()