Merge pull request #5553 from chrisburr/executor-fixes

[rel-v7r2] Fixes to executor infrastructure

Author: fstagni

src/DIRAC/Core/Base/ExecutorMindHandler.py

@@ -124,6 +124,7 @@ def __sendTask(self, taskId, taskObj, eId, eType):
         taskStub = result["Value"]
         result = self.srv_msgCreate("ProcessTask")
         if not result["OK"]:
+            gLogger.error("Failed to create message for", "%s %r" % (taskId, result))
             return result
         msgObj = result["Value"]
         msgObj.taskId = taskId

src/DIRAC/Core/DISET/MessageClient.py

@@ -83,14 +83,12 @@ def __cbDisconnect(self, trid):
         self.__trid = False
         try:
             self.__transport.close()
-        except BaseException:
+        except Exception:
             pass
         for cb in self.__specialCallbacks["drop"]:
             try:
                 cb(self)
-            except SystemExit:
-                raise
-            except BaseException:
+            except Exception:
                 gLogger.exception("Exception while processing disconnect callbacks")
 
     def __cbRecvMsg(self, trid, msgObj):
@@ -107,7 +105,7 @@ def __cbRecvMsg(self, trid, msgObj):
                 # If no specific callback but a generic one, return the generic one
                 if msgName not in self.__callbacks:
                     return result
-            except BaseException:
+            except Exception:
                 gLogger.exception("Exception while processing callbacks", msgObj.getName())
         if msgName not in self.__callbacks:
             return S_ERROR("Unexpected message")
@@ -117,7 +115,7 @@ def __cbRecvMsg(self, trid, msgObj):
                 gLogger.error("Callback for message does not return S_OK/S_ERROR", msgName)
                 return S_ERROR("No response")
             return result
-        except BaseException:
+        except Exception:
             gLogger.exception("Exception while processing callbacks", msgName)
         return S_ERROR("No response")
 
@@ -128,6 +126,7 @@ def sendMessage(self, msgObj):
         if not self.__trid:
             result = self.connect()
             if not result["OK"]:
+                gLogger.error("Failed connect for sending", "%r" % msgObj)
                 return result
         return self.__msgBroker.sendMessage(self.__trid, msgObj)
 

src/DIRAC/Core/DISET/private/MessageBroker.py

@@ -165,12 +165,8 @@ def __listenAutoReceiveConnections(self):
 
             try:
                 events = sel.select(timeout=1)
-            except (socket.error, select.error):
-                # TODO: When can this happen?
-                time.sleep(0.001)
-                continue
-            except Exception as e:
-                gLogger.exception("Exception while selecting persistent connections", lException=e)
+            except Exception:
+                gLogger.exception("Exception while selecting persistent connections")
                 continue
 
             for key, event in events:
@@ -205,7 +201,7 @@ def __receiveMsgDataAndQueue(self, trid):
 
     def __processIncomingData(self, trid, receivedResult):
         # If keep alive, return OK
-        if "keepAlive" in receivedResult and receivedResult["keepAlive"]:
+        if receivedResult.get("keepAlive"):
             return S_OK()
         # If idle read return
         self.__trInOutLock.acquire()

src/DIRAC/Core/Utilities/ExecutorDispatcher.py

@@ -103,7 +103,8 @@ def getIdleExecutor(self, eType):
         idleId = None
         maxFreeSlots = 0
         try:
-            for eId in self.__typeToId[eType]:
+            # Work on a copy of self.__typeToId[eType] to race conditions causing it's size to change while iterating
+            for eId in list(self.__typeToId[eType]):
                 freeSlots = self.freeSlots(eId)
                 if freeSlots > maxFreeSlots:
                     maxFreeSlots = freeSlots
@@ -425,7 +426,7 @@ def addExecutor(self, eId, eTypes, maxTasks=1):
             self.__fillExecutors(eType)
 
     def removeExecutor(self, eId):
-        self.__log.verbose("Removing executor %s" % eId)
+        self.__log.info("Removing executor %s" % eId)
         self.__executorsLock.acquire()
         try:
             if eId not in self.__idMap:
@@ -439,8 +440,8 @@ def removeExecutor(self, eId):
                     eTask = self.__tasks[taskId]
                 except KeyError:
                     # Task already removed
-                    pass
-                if eTask.eType:
+                    eTask = None
+                if eTask and eTask.eType:
                     self.__queues.pushTask(eTask.eType, taskId, ahead=True)
                 else:
                     self.__dispatchTask(taskId)
@@ -844,6 +845,8 @@ def __msgTaskToExecutor(self, taskId, eId, eType):
             errMsg = "Send task callback did not send back an S_OK/S_ERROR structure"
             self.__log.fatal(errMsg)
             raise ValueError(errMsg)
+        if not result["OK"]:
+            self.__log.error("Failed to cbSendTask", "%r" % result)
 
 
 class UnrecoverableTaskException(Exception):

src/DIRAC/WorkloadManagementSystem/Client/JobState/JobState.py

@@ -37,10 +37,10 @@ def reset(self):
     def checkDBAccess(cls):
         # Init DB if there
         if not JobState.__db.checked:
-            JobState.__db.checked = True
             JobState.__db.jobDB = JobDB()
             JobState.__db.logDB = JobLoggingDB()
             JobState.__db.tqDB = TaskQueueDB()
+            JobState.__db.checked = True
 
     def __init__(self, jid, source="Unknown"):
         self.__jid = jid

src/DIRAC/WorkloadManagementSystem/DB/JobLoggingDB.py

@@ -36,11 +36,11 @@ def __init__(self):
     def addLoggingRecord(
         self,
         jobID,
-        status="idem",
-        minorStatus="idem",
-        applicationStatus="idem",
+        status=None,
+        minorStatus=None,
+        applicationStatus=None,
         date=None,
-        source="Unknown",
+        source=None,
         minor=None,
         application=None,
     ):
@@ -59,6 +59,11 @@ def addLoggingRecord(
         if application:
             applicationStatus = application
 
+        status = status or "idem"
+        minorStatus = minorStatus or "idem"
+        applicationStatus = applicationStatus or "idem"
+        source = source or "Unknown"
+
         event = "status/minor/app=%s/%s/%s" % (status, minorStatus, applicationStatus)
         self.log.info("Adding record for job ", str(jobID) + ": '" + event + "' from " + source)
 

src/DIRAC/WorkloadManagementSystem/Service/OptimizationMindHandler.py

@@ -201,15 +201,20 @@ def exec_deserializeTask(cls, taskStub):
 
     @classmethod
     def exec_taskError(cls, jid, cachedJobState, errorMsg):
-        result = cachedJobState.commitChanges()
-        if not result["OK"]:
-            cls.log.error("Cannot write changes to job %s: %s" % (jid, result["Message"]))
+        if cachedJobState:
+            result = cachedJobState.commitChanges()
+            if not result["OK"]:
+                cls.log.error("Cannot write changes to job %s: %s" % (jid, result["Message"]))
+        else:
+            cls.log.error(
+                "Called exec_taskError with",
+                "jid=%r cachedJobState=%r errorMsg=%r" % (jid, cachedJobState, errorMsg),
+            )
         jobState = JobState(jid)
         result = jobState.getStatus()
-        if result["OK"]:
-            if result["Value"][0].lower() == "failed":
-                return S_OK()
-        else:
+        if not result["OK"]:
             cls.log.error("Could not get status of job %s: %s" % (jid, result["Message "]))
+        elif result["Value"][0].lower() == "failed":
+            return S_OK()
         cls.log.notice("Job %s: Setting to Failed|%s" % (jid, errorMsg))
         return jobState.setStatus("Failed", errorMsg, source="OptimizationMindHandler")