Merge pull request #392 from HSF/dev

Dev

Author: wguanicedew

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

@@ -383,7 +383,8 @@ def depend_on(self, work):
 
                 for input_d in inputs_dependency:
                     task_name = input_d['task']
-                    dependency_tasks.add(task_name)
+                    if task_name not in dependency_tasks:
+                        dependency_tasks.add(task_name)
             self.dependency_tasks = list(dependency_tasks)
 
         if work.task_name in self.dependency_tasks:
@@ -881,7 +882,7 @@ def submit_panda_task(self, processing):
                                 task_param['processingType'] = p_type
                                 self.logger.warn(msg)
                         else:
-                            if task_param['site']:
+                            if 'site' in task_param and task_param['site']:
                                 for q in queue_processing_type:
                                     if task_param['site'] in q or q in task_param['site']:
                                         p_type = queue_processing_type[q]

main/lib/idds/agents/common/eventbus/msgeventbusbackend.py

@@ -423,6 +423,9 @@ def send(self, event):
                 if self.debug:
                     self.logger.debug("MsgEventBusBackend send event: %s" % req)
 
+                if not self.manager_socket or self.manager_socket.closed:
+                    self.init_msg_channel()
+
                 self.manager_socket.send_string(req)
                 if self.manager_socket.poll(self.socket_timeout * 1000):
                     reply = self.manager_socket.recv_string()
@@ -465,6 +468,9 @@ def send_bulk(self, events):
                 if self.debug:
                     self.logger.debug("MsgEventBusBackend send bulk event: %s" % req)
 
+                if not self.manager_socket or self.manager_socket.closed:
+                    self.init_msg_channel()
+
                 self.manager_socket.send_string(req)
                 if self.manager_socket.poll(self.socket_timeout * 1000):
                     reply = self.manager_socket.recv_string()
@@ -510,6 +516,10 @@ def get(self, event_type, num_events=1, wait=0):
 
                 if self.debug:
                     self.logger.debug("MsgEventBusBackend get event: %s" % req)
+
+                if not self.manager_socket or self.manager_socket.closed:
+                    self.init_msg_channel()
+
                 self.manager_socket.send_string(req)
 
                 if self.manager_socket.poll(10 * 1000):