[core] graph : Manage nodeStatus file monitoring

Author: Alxiice

meshroom/core/desc/node.py

@@ -2,7 +2,6 @@
 from inspect import getfile
 from pathlib import Path
 import logging
-import os
 import psutil
 import shlex
 import shutil
@@ -29,6 +28,7 @@ def __init__(self):
         signal.signal(signal.SIGTERM, self.exit)
 
     def addSubprocess(self, process):
+        print(f"[ExitCleanup] (addSubprocess) register subprocess {process}")
         self._subprocesses.append(process)
 
     def exit(self, signum, frame):
@@ -40,7 +40,7 @@ def exit(self, signum, frame):
                     proc.wait(timeout=5)
             except subprocess.TimeoutExpired:
                 proc.kill()
-        raise RuntimeError("Process has been killed")
+        sys.exit(0)
 
 exitCleanup = ExitCleanup()
 

meshroom/core/graph.py

@@ -1478,6 +1478,9 @@ def update(self):
             self.dirtyTopology = False
 
         self.updated.emit()
+    
+    def updateMonitoredFiles(self):
+        self.statusUpdated.emit()
 
     def markNodesDirty(self, fromNode):
         """
@@ -1601,6 +1604,7 @@ def setVerbose(self, v):
     cacheDirChanged = Signal()
     cacheDir = Property(str, cacheDir.fget, cacheDir.fset, notify=cacheDirChanged)
     updated = Signal()
+    statusUpdated = Signal()
     canComputeLeavesChanged = Signal()
     canComputeLeaves = Property(bool, lambda self: self._canComputeLeaves, notify=canComputeLeavesChanged)
 

meshroom/core/node.py

@@ -50,6 +50,7 @@ class Status(Enum):
     STOPPED = 4
     KILLED = 5
     SUCCESS = 6
+    # SKIPPED = 7
     INPUT = 7  # Special status for input nodes
 
 
@@ -94,12 +95,6 @@ def setNode(self, node):
         self.nodeName = node.name
         self.setNodeType(node)
 
-    def initStartCompute(self):
-        pass  # TODO
-
-    def initIsolatedCompute(self):
-        pass  # TODO
-
     def initExternSubmit(self):
         """
         When submitting a node, we reset the status information to ensure that we do not keep
@@ -125,12 +120,10 @@ def initLocalSubmit(self):
         self.status = Status.SUBMITTED
         self.execMode = ExecMode.LOCAL
 
-    def initEndCompute(self):
-        pass  # TODO
-
     def setComputationStatusToInheritChunks(self):
+        # TODO
         self.status: Status = Status.NONE
-    
+
     def setNodeType(self, node):
         """
         Set the node type and package information from the given node.
@@ -525,6 +518,15 @@ def logger(self):
 
     def getExecModeName(self):
         return self._status.execMode.name
+    
+    def shouldMonitorChanges(self):
+        """ Check whether we should monitor changes in minimal mode 
+        Only chunks that are run externally or local_isolated should be monitored,
+        when run locally, status changes are already notified.
+        Chunks with an ERROR status may be re-submitted externally and should thus still be monitored
+        """
+        return (self.isExtern() and self._status.status in (Status.SUBMITTED, Status.RUNNING, Status.ERROR)) or \
+               (self.node.getMrNodeType() == MrNodeType.NODE and self._status.status in (Status.SUBMITTED, Status.RUNNING))
 
     def updateStatusFromCache(self):
         """
@@ -838,6 +840,7 @@ def __init__(self, nodeType: str, position: Position = None, parent: BaseObject
 
         self._nodeStatus: NodeStatusData = NodeStatusData(self._name, nodeType, self.packageName,
                                                           self.packageVersion, self.getMrNodeType())
+        self.nodeStatusFileLastModTime = -1
 
         self.globalStatusChanged.connect(self.updateDuplicatesStatusAndLocked)
 
@@ -1352,6 +1355,15 @@ def isExtern(self):
         interrupted, its execution mode will always be local, even if computations resume
         externally.
         """
+        if not self._chunksCreated:
+            if self._nodeStatus.execMode == ExecMode.EXTERN:
+                return True
+            # TODO : use sessionID
+            # elif self._nodeStatus.execMode == ExecMode.LOCAL:
+            #     if self._nodeStatus.status in (Status.SUBMITTED, Status.RUNNING):
+            #         return meshroom.core.sessionUid not in (self._nodeStatus.submitterSessionUid, self._nodeStatus.sessionUid)
+            #     return False
+            return False
         if len(self._chunks) == 0:
             return False
         return any(chunk.isExtern() for chunk in self._chunks)
@@ -1512,11 +1524,23 @@ def sourceCodeFolder(self):
     @property
     def nodeStatusFile(self):
         return os.path.join(self.graph.cacheDir, self.internalFolder, "nodeStatus")
+    
+    def shouldMonitorChanges(self):
+        """ Check whether we should monitor changes in minimal mode 
+        Only chunks that are run externally or local_isolated should be monitored,
+        when run locally, status changes are already notified.
+        Chunks with an ERROR status may be re-submitted externally and should thus still be monitored
+        """
+        if self._chunksCreated:
+            # Only monitor when chunks are not created (in this case monitor chunk status files instead)
+            return False
+        return (self.isExtern() and self._nodeStatus.status in (Status.SUBMITTED, Status.RUNNING, Status.ERROR)) or \
+               (self.getMrNodeType() == MrNodeType.NODE and self._nodeStatus.status in (Status.SUBMITTED, Status.RUNNING))
 
     def updateNodeStatusFromCache(self):
         """
         Update node status based on status file content/existence.
-        # TODO : integrate statusFileLastModTime ?
+        # TODO : integrate nodeStatusFileLastModTime ?
         Returns True if a change on the chunk setup has been detected
         """
         chunksRangeHasChanged = False
@@ -1525,8 +1549,10 @@ def updateNodeStatusFromCache(self):
             self._nodeStatus.loadFromCache(self.nodeStatusFile)
             if self._nodeStatus.chunks != oldChunkSetup:
                 chunksRangeHasChanged = True
+            self.nodeStatusFileLastModTime = os.path.getmtime(self.nodeStatusFile)
         else:
             # No status file => reset status to Status.None
+            self.nodeStatusFileLastModTime = -1
             self._nodeStatus.reset()
         self._nodeStatus.setNodeType(self)
         return chunksRangeHasChanged

meshroom/core/taskManager.py

@@ -479,7 +479,9 @@ def submit(self, graph, submitter=None, toNodes=None, submitLabel="{projectName}
                                 ))
 
         # Update task manager's lists
+        print("[TaskManager] (submit) updateNodes")
         self.updateNodes()
+        print("[TaskManager] (submit) graph.update")
         graph.update()
 
         # Check dependencies of toNodes
@@ -500,6 +502,13 @@ def submit(self, graph, submitter=None, toNodes=None, submitLabel="{projectName}
         self.checkCompatibilityNodes(graph, nodesToProcess, "SUBMITTING")  # name of the context is important for QML
         self.checkDuplicates(nodesToProcess, "SUBMITTING")  # name of the context is important for QML
 
+        # Update nodes status
+        for node in nodesToProcess:
+            node.destroyed.connect(lambda obj=None, name=node.name: self.onNodeDestroyed(obj, name))
+            node.initStatusOnSubmit()
+        print("[TaskManager] (submit) graph.updateMonitoredFiles")
+        graph.updateMonitoredFiles()
+            
         flowEdges = graph.flowEdges(startNodes=toNodes)
         edgesToProcess = set(edgesToProcess).intersection(flowEdges)
 
@@ -508,11 +517,13 @@ def submit(self, graph, submitter=None, toNodes=None, submitLabel="{projectName}
 
         try:
             res = sub.submit(nodesToProcess, edgesToProcess, graph.filepath, submitLabel=submitLabel)
-            for node in nodesToProcess:
-                node.destroyed.connect(lambda obj=None, name=node.name: self.onNodeDestroyed(obj, name))
-                node.initStatusOnSubmit()  # update node status
+            if res:
                 if isinstance(res, BaseSubmittedJob):
                     jobManager.addJob(res, nodesToProcess)
+            else:
+                for node in nodesToProcess:
+                    # TODO : Notify the node that there was an issue on submit
+                    pass
             self._nodes.update(nodesToProcess)
             self._nodesExtern.extend(nodesToProcess)