diff --git a/pandajedi/jediorder/TaskCommando.py b/pandajedi/jediorder/TaskCommando.py index e5a6ab71e..f5f69e708 100644 --- a/pandajedi/jediorder/TaskCommando.py +++ b/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()}") diff --git a/pandaserver/taskbuffer/db_proxy_mods/misc_standalone_module.py b/pandaserver/taskbuffer/db_proxy_mods/misc_standalone_module.py index 9d278fc72..c9840580b 100644 --- a/pandaserver/taskbuffer/db_proxy_mods/misc_standalone_module.py +++ b/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}" )