Add locking mechanism and refactor task execution methods

pandajedi/jediorder/TaskCommando.py

@@ -24,7 +24,7 @@
 class TaskCommando(JediKnight):
     # constructor
     def __init__(self, commuChannel, taskBufferIF, ddmIF, vos, prodSourceLabels):
-        self.vos = self.parseInit(vos)
+        self.vos = self.parseInit(vos)        
         self.prodSourceLabels = self.parseInit(prodSourceLabels)
         self.pid = f"{socket.getfqdn().split('.')[0]}-{os.getpid()}-dog"
         JediKnight.__init__(self, commuChannel, taskBufferIF, ddmIF, logger)
@@ -44,6 +44,11 @@ def start(self):
                 for vo in self.vos:
                     # loop over all sourceLabels
                     for prodSourceLabel in self.prodSourceLabels:
+                        # lock process
+                        get_lock = self.taskBufferIF.lockProcess_JEDI(vo, prodSourceLabel, None, None, None, self.__class__.__name__, self.pid, timeLimit=1)
+                        if not get_lock:
+                            tmpLog.debug(f"failed to get lock for vo={vo} label={prodSourceLabel}")
+                            continue
                         # get the list of tasks to exec command
                         tmpList = self.taskBufferIF.getTasksToExecCommand_JEDI(vo, prodSourceLabel)
                         if tmpList is None:
@@ -62,6 +67,8 @@ def start(self):
                                 thr.start()
                             # join
                             threadPool.join()
+                        # unlock process
+                        self.taskBufferIF.unlockProcess_JEDI(vo, prodSourceLabel, None, None, None, self.__class__.__name__, self.pid)
                 tmpLog.debug("done")
             except Exception as e:
                 tmpLog.error(f"failed in {self.__class__.__name__}.start() with {str(e)} {traceback.format_exc()}")

pandaserver/taskbuffer/db_proxy_mods/misc_standalone_module.py

@@ -4136,14 +4136,12 @@ def extendSandboxLifetime_JEDI(self, jedi_taskid, file_name):
     def lockProcess_JEDI(self, vo, prodSourceLabel, cloud, workqueue_id, resource_name, component, pid, forceOption, timeLimit):
         comment = " /* JediDBProxy.lockProcess_JEDI */"
         # defaults
-        if cloud is None:
-            cloud = "default"
-        if workqueue_id is None:
-            workqueue_id = 0
-        if resource_name is None:
-            resource_name = "default"
-        if component is None:
-            component = "default"
+        vo = "default" if vo is None else vo
+        prodSourceLabel = "default" if prodSourceLabel is None else prodSourceLabel
+        cloud = "default" if cloud is None else cloud
+        workqueue_id = 0 if workqueue_id is None else workqueue_id
+        resource_name = "default" if resource_name is None else resource_name
+        component = "default" if component is None else component
         tmpLog = self.create_tagged_logger(
             comment, f"vo={vo} label={prodSourceLabel} cloud={cloud} queue={workqueue_id} resource_type={resource_name} component={component} pid={pid}"
         )
@@ -4227,14 +4225,12 @@ def lockProcess_JEDI(self, vo, prodSourceLabel, cloud, workqueue_id, resource_na
     def unlockProcess_JEDI(self, vo, prodSourceLabel, cloud, workqueue_id, resource_name, component, pid):
         comment = " /* JediDBProxy.unlockProcess_JEDI */"
         # defaults
-        if cloud is None:
-            cloud = "default"
-        if workqueue_id is None:
-            workqueue_id = 0
-        if resource_name is None:
-            resource_name = "default"
-        if component is None:
-            component = "default"
+        vo = "default" if vo is None else vo
+        prodSourceLabel = "default" if prodSourceLabel is None else prodSourceLabel
+        cloud = "default" if cloud is None else cloud
+        workqueue_id = 0 if workqueue_id is None else workqueue_id
+        resource_name = "default" if resource_name is None else resource_name
+        component = "default" if component is None else component
         tmpLog = self.create_tagged_logger(
             comment, f"vo={vo} label={prodSourceLabel} cloud={cloud} queue={workqueue_id} resource_type={resource_name} component={component} pid={pid}"
         )