[code] submitter : fix issues in dynamic chunks & submitting

Author: Alxiice

bin/meshroom_compute

@@ -91,10 +91,11 @@ if args.node:
         for chunk in chunks:
             if chunk.status.status in submittedStatuses:
                 # Particular case for the local isolated, the node status is set to RUNNING by the submitter directly.
-                # We ensure that no other instance has started to compute, by checking that the sessionUid is empty.
-                if chunk.node.getMrNodeType() == meshroom.core.MrNodeType.NODE and not chunk.status.sessionUid and chunk.status.submitterSessionUid:
+                # We ensure that no other instance has started to compute, by checking that the computeSessionUid is empty.
+                if chunk.node.getMrNodeType() == meshroom.core.MrNodeType.NODE and \
+                    not chunk.status.computeSessionUid and node._nodeStatus.submitterSessionUid:
                     continue
-                print(f'Warning: Node is already submitted with status "{chunk.status.status.name}". See file: "{chunk.statusFile}". ExecMode: {chunk.status.execMode.name}, SessionUid: {chunk.status.sessionUid}, submitterSessionUid: {chunk.status.submitterSessionUid}')
+                print(f'Warning: Node is already submitted with status "{chunk.status.status.name}". See file: "{chunk.statusFile}". ExecMode: {chunk.status.execMode.name}, computeSessionUid: {chunk.status.computeSessionUid}, submitterSessionUid: {node._nodeStatus.submitterSessionUid}')
                 # sys.exit(-1)
 
     if args.extern:

bin/meshroom_createChunks

@@ -78,6 +78,10 @@ graph.update()
 node = graph.findNode(args.node)
 submittedStatuses = [Status.RUNNING]
 
+if node._nodeStatus.status in (Status.STOPPED, Status.KILLED):
+    logging.error("Node status is STOPPED or KILLED.")
+    sys.exit(666)
+
 
 # Find job
 submitter = None
@@ -87,12 +91,17 @@ for subName, sub in submitters.items():
         submitter = sub
         break
 
-if not (hasattr(node, '_chunksCreated') and node._chunksCreated):
+if not node._chunksCreated:
     # Create node chunks
     # Once created we don't have to do it again even if we relaunch the job
-    node._createChunks()
+    node.createChunks()
     # Set the chunks statuses
-    node.initStatusOnSubmit()
+    for chunk in node._chunks:
+        if args.forceCompute or chunk._status.status != Status.SUCCESS:
+            hasChunkToLaunch = True
+            chunk._status.setNode(node)
+            chunk._status.initExternSubmit()
+            chunk.upgradeStatusFile()
 
 # Get chunks to process in the current process
 chunksToProcess = []
@@ -103,20 +112,21 @@ else:
     # Cannot retrieve job -> execute process serially
     chunksToProcess = node.chunks
 
-logging.info("[MeshroomCreateChunks] Chunks to process here :", chunksToProcess)
+logging.info(f"[MeshroomCreateChunks] Chunks to process here : {chunksToProcess}")
 
 if not args.forceStatus and not args.forceCompute:
     for chunk in chunksToProcess:
         if chunk.status.status in submittedStatuses:
             # Particular case for the local isolated, the node status is set to RUNNING by the submitter directly.
-            # We ensure that no other instance has started to compute, by checking that the sessionUid is empty.
-            if chunk.node.getMrNodeType() == meshroom.core.MrNodeType.NODE and not chunk.status.sessionUid and chunk.status.submitterSessionUid:
+            # We ensure that no other instance has started to compute, by checking that the sessicomputeSessionUidonUid is empty.
+            if chunk.node.getMrNodeType() == meshroom.core.MrNodeType.NODE and \
+                not chunk.status.computeSessionUid and node._nodeStatus.submitterSessionUid:
                 continue
             logging.warning(
                 f"[MeshroomCreateChunks] Node is already submitted with status " \
                 f"\"{chunk.status.status.name}\". See file: \"{chunk.statusFile}\". " \
-                f"ExecMode: {chunk.status.execMode.name}, SessionUid: {chunk.status.sessionUid}, " \
-                f"submitterSessionUid: {chunk.status.submitterSessionUid}")
+                f"ExecMode: {chunk.status.execMode.name}, computeSessionUid: {chunk.status.computeSessionUid}, " \
+                f"submitterSessionUid: {node._nodeStatus.submitterSessionUid}")
             # sys.exit(-1)
 
 if chunksToProcess:

meshroom/core/desc/node.py

@@ -23,17 +23,21 @@
 
 # Handle cleanup
 class ExitCleanup:
+    """
+    Make sure we kill child subprocesses when the main process exits receive SIGTERM.
+    """
+
     def __init__(self):
         self._subprocesses = []
         signal.signal(signal.SIGTERM, self.exit)
-    
+
     def addSubprocess(self, process):
-        print(f"[ExitCleanup] (addSubprocess) register subprocess {process}")
+        logging.debug(f"[ExitCleanup] Register subprocess {process}")
         self._subprocesses.append(process)
-    
+
     def exit(self, signum, frame):
         for proc in self._subprocesses:
-            print(f"[ExitCleanup] (exit) kill subprocess {proc}")
+            logging.debug(f"[ExitCleanup] Kill subprocess {proc}")
             try:
                 if proc.is_running():
                     proc.terminate()
@@ -307,8 +311,6 @@ def processChunkInEnvironment(self, chunk):
 
         if len(chunk.node.getChunks()) > 1:
             meshroomComputeCmd += f" --iteration {chunk.range.iteration}"
-        
-        print(f"(processChunkInEnvironment) meshroomComputeCmd={meshroomComputeCmd}")
 
         runtimeEnv = chunk.node.nodeDesc.plugin.runtimeEnv
         cmdPrefix = chunk.node.nodeDesc.plugin.commandPrefix

meshroom/core/graph.py

@@ -21,6 +21,7 @@
 from meshroom.core.node import BaseNode, Status, Node, CompatibilityNode
 from meshroom.core.nodeFactory import nodeFactory
 from meshroom.core.mtyping import PathLike
+from meshroom.core.submitter import BaseSubmittedJob, jobManager
 
 # Replace default encoder to support Enums
 
@@ -1669,7 +1670,7 @@ def executeGraph(graph, toNodes=None, forceCompute=False, forceStatus=False):
     graph.save()
 
     for node in nodes:
-        node.beginSequence(forceCompute)
+        node.initStatusOnCompute(forceCompute)
 
     for n, node in enumerate(nodes):
         try:
@@ -1715,11 +1716,18 @@ def submitGraph(graph, submitter, toNodes=None, submitLabel="{projectName}"):
         raise RuntimeError("Unknown Submitter: '{submitter}'. Available submitters are: '{allSubmitters}'.".format(
             submitter=submitter, allSubmitters=str(meshroom.core.submitters.keys())))
 
+    for node in nodesToProcess:
+        node.initStatusOnSubmit()
+
     try:
         res = sub.submit(nodesToProcess, edgesToProcess, graph.filepath, submitLabel=submitLabel)
         if res:
+            if isinstance(res, BaseSubmittedJob):
+                jobManager.addJob(res, nodesToProcess)
+        else:
             for node in nodesToProcess:
-                node.initStatusOnSubmit()  # update node status
+                # TODO : Notify the node that there was an issue on submit
+                pass
     except Exception as e:
         logging.error(f"Error on submit : {e}")