[core] Start updating taskmanager and submitter for new chunk process

Author: Alxiice

bin/meshroom_createChunks

@@ -1,9 +1,16 @@
 #!/usr/bin/env python
+
+"""
+This is a script used to wrap the process of processing a node on the farm
+It will handle chunk creation and create all the jobs for these chunks
+If the submitter cannot create chunks, then it will process the chunks serially
+in the current process
+"""
+
 import argparse
 import logging
 import os
 import sys
-
 try:
     import meshroom
 except Exception:
@@ -25,6 +32,9 @@ from meshroom.core.node import Status
 parser = argparse.ArgumentParser(description='Execute a Graph of processes.')
 parser.add_argument('graphFile', metavar='GRAPHFILE.mg', type=str,
                     help='Filepath to a graph file.')
+
+parser.add_argument('--submitter', type=str, required=True,
+                    help='Name of the submitter used to create the job.')
 parser.add_argument('--node', metavar='NODE_NAME', type=str, required=True,
                     help='Process the node. It will generate an error if the dependencies are not already computed.')
 parser.add_argument('--inCurrentEnv', help='Execute process in current env without creating a dedicated runtime environment.',
@@ -70,47 +80,31 @@ submittedStatuses = [Status.RUNNING]
 
 
 # Find job
-submitter, job = None, None
+submitter = None
 # It's required if we want to spool chunks on different machines
-jobInfos = node._nodeStatus.jobInfos
-submitterName, jid = jobInfos.get("submitterName"), jobInfos.get("jid")
 for subName, sub in submitters.items():
-    if not sub._options.includes(SubmitterOptionsEnum.RETRIEVE):
-        continue
-    if submitterName == subName:
+    if args.submitter == subName:
         submitter = sub
-        job = jobManager.retreiveJob(sub, jid)
         break
 
 if not (hasattr(node, '_chunksCreated') and node._chunksCreated):
     # Create node chunks
     # Once created we don't have to do it again even if we relaunch the job
     node._createChunks()
-    # TODO
-    # Notify that the chunk is submitted
-    # for chunk in node._chunks:
-    #     chunk._status.initExternSubmit()
+    # Set the chunks statuses
+    node.initStatusOnSubmit()
 
 # Get chunks to process in the current process
 chunksToProcess = []
-if job:
-    if not job.submitterOptions.includes(SubmitterOptionsEnum.EDIT_TASKS):
+if submitter:
+    if not submitter._options.includes(SubmitterOptionsEnum.EDIT_TASKS):
         chunksToProcess = node.chunks
 else:
     # Cannot retrieve job -> execute process serially
     chunksToProcess = node.chunks
 
 logging.info("[MeshroomCreateChunks] Chunks to process here :", chunksToProcess)
 
-# Identify if the current process 
-# is a running chunk or if we need to spool the chunk job
-# 
-
-# - create map from job/no des
-# - if we are on a task process
-# - do like before
-pass
-
 if not args.forceStatus and not args.forceCompute:
     for chunk in chunksToProcess:
         if chunk.status.status in submittedStatuses:
@@ -135,8 +129,8 @@ if chunksToProcess:
     node.restoreLogger()
 else:
     logging.info(f"[MeshroomCreateChunks] -> create job to process chunks {node.chunks}")
-    job.addChunkTask(node, graphFile=args.graphFile, cache=args.cache, 
-                     forceStatus=args.forceStatus, forceCompute=args.forceCompute)
+    submitter.createChunkTask(node, graphFile=args.graphFile, cache=args.cache, 
+                              forceStatus=args.forceStatus, forceCompute=args.forceCompute)
 
 # Restore the log level
 logging.getLogger().setLevel(meshroom.logStringToPython[args.verbose])

meshroom/core/graph.py

@@ -1081,7 +1081,6 @@ def dfsToProcess(self, startNodes=None):
              visited nodes and edges that are not already computed (node.status != SUCCESS).
              The order is defined by the visit and finishVertex event.
         """
-        print(f"[Graph] (dfsToProcess) startNodes={startNodes}")
         nodes = []
         edges = []
         visitor = Visitor(reverse=False, dependenciesOnly=True)

meshroom/core/node.py

@@ -40,7 +40,6 @@ def renameWritingToFinalPath(writingFilepath: str, filepath: str) -> str:
                 pass
     os.rename(writingFilepath, filepath)
 
-
 class Status(Enum):
     """
     """
@@ -75,24 +74,20 @@ def __init__(self, nodeName='', nodeType='', packageName='', packageVersion='',
         self.nodeType: str = nodeType
         self.packageName: str = packageName
         self.packageVersion: str = packageVersion
-        self.mrNodeType: MrNodeType = mrNodeType
-        self.chunks: NodeChunkSetup = None
+        self.mrNodeType: str = mrNodeType
+        self.reset()
+
+    def reset(self):
+        self.resetChunkInfos()
         self.resetDynamicValues()
 
+    def resetChunkInfos(self):
+        self.chunks: NodeChunkSetup = None
+
     def resetDynamicValues(self):
-        self.statusInheritedFromChunks: bool = False
         self.status: Status = Status.NONE
         self.execMode: ExecMode = ExecMode.NONE
         self.jobInfos: dict = {}
-        # TODO : move all possible infos from ChunkStatusData to this place
-
-    def reset(self):
-        self.nodeName: str = ""
-        self.nodeType: str = ""
-        self.packageName: str = ""
-        self.packageVersion: str = ""
-        self.mrNodeType: str = MrNodeType.NONE
-        self.resetDynamicValues()
 
     def setNode(self, node):
         """ Set the node information from one node instance. """
@@ -113,6 +108,13 @@ def initExternSubmit(self):
         self.resetDynamicValues()
         self.status = Status.SUBMITTED
         self.execMode = ExecMode.EXTERN
+    
+    def setJob(self, jid, submitterName):
+        """ Set Job infos on the node so that """
+        self.jobInfos = {
+            "jid": str(jid),
+            "submitterName": str(submitterName),
+        }
 
     def initLocalSubmit(self):
         """
@@ -127,15 +129,8 @@ def initEndCompute(self):
         pass  # TODO
 
     def setComputationStatusToInheritChunks(self):
-        self.statusInheritedFromChunks = True
         self.status: Status = Status.NONE
 
-    def setJob(self, jid, submitterName):
-        self.jobInfos = {
-            "jid": str(jid),
-            "submitterName": str(submitterName),
-        }
-
     def setNodeType(self, node):
         """
         Set the node type and package information from the given node.
@@ -149,38 +144,39 @@ def setNodeType(self, node):
     def toDict(self):
         keys = list(self.__slots__) or []
         d = {key:getattr(self, key) for key in keys}
-        chunks = [-1, -1, -1]
+        chunks = None
         if self.chunks:
             chunks = list(self.chunks)
         d["chunks"] = chunks
         return d
 
-    def fromDict(self, d):
+    def updateFromDict(self, d):
         self.reset()
         self.mrNodeType = d.pop("mrNodeType", MrNodeType.NONE)
         if not isinstance(self.mrNodeType, MrNodeType):
             self.mrNodeType = MrNodeType[self.mrNodeType]
-        self.chunks = NodeChunkSetup(*d.pop("chunks", [-1, -1, -1]))
+        if "chunks" in d:
+            chunks = d.pop("chunks")
+            if chunks:
+                self.chunks = NodeChunkSetup(*chunks)
         if "status" in d:
             statusName = d.pop("status")
             self.status: Status = Status[statusName]
         if "execMode" in d:
             execModeName = d.pop("execMode")
-            self.execMode = Status[execModeName]
+            self.execMode = ExecMode[execModeName]
         for _key, _value in d.items():
             if _key in self.__slots__:
                 setattr(self, _key, _value)
 
     def loadFromCache(self, statusFile):
-        print(f"[NodeStatusData](loadFromCache) {statusFile}")
         self.reset()
         try:
             with open(statusFile) as jsonFile:
                 statusData = json.load(jsonFile)
-            print(f"statusData :\n{statusData}")
-            self.fromDict(statusData)
+            self.updateFromDict(statusData)
         except Exception as e:
-            logging.warning(f"[NodeStatusData] (loadFromCache) {self.nodeName}: Error while loading status file {statusFile}: {e}")
+            logging.warning(f"(loadFromCache) {self.nodeName}: Error while loading status file {statusFile}: {e}")
             self.reset()
 
     @property 
@@ -534,7 +530,6 @@ def updateStatusFromCache(self):
         """
         Update chunk status based on status file content/existence.
         """
-        print(f"updateStatusFromCache {self}")
         # TODO : If this is a placeholder chunk
         # Then we shouldn't do anything here
 
@@ -651,7 +646,6 @@ def isFinished(self):
         return self._status.status == Status.SUCCESS
 
     def process(self, forceCompute=False, inCurrentEnv=False):
-        print(f"[NodeChunk] (process) {self}")
         if not forceCompute and self._status.status == Status.SUCCESS:
             logging.info(f"Node chunk already computed: {self.name}")
             return
@@ -698,7 +692,6 @@ def process(self, forceCompute=False, inCurrentEnv=False):
             self.statThread.join()
             self.statistics = stats.Statistics()
             del runningProcesses[self.name]
-        print(f"[NodeChunk] (process) -> done")
 
 
     def _processInIsolatedEnvironment(self):
@@ -1529,9 +1522,6 @@ def updateNodeStatusFromCache(self):
         # TODO : integrate statusFileLastModTime ?
         Returns True if a change on the chunk setup has been detected
         """
-        if self.label == "E":
-            chunksInfos = [f"<{hex(id(c._status))}, {c._status.status.name}>" for c in self._chunks]
-            print(f"[BaseNode] (updateNodeStatusFromCache) E ({self}) ({chunksInfos})")
         chunksRangeHasChanged = False
         if os.path.exists(self.nodeStatusFile):
             oldChunkSetup = self._nodeStatus.chunks
@@ -1548,7 +1538,6 @@ def updateStatusFromCache(self):
         """
         Update node status based on status file content/existence.
         """
-        print(f"updateStatusFromCache {self}")
         # Update nodeStatus from cache
         chunkChanged = self.updateNodeStatusFromCache()
         # Create chunks if we found info on them on the node cache
@@ -1580,13 +1569,11 @@ def saveNodeStatusFile(self):
         renameWritingToFinalPath(statusFilepathWriting, statusFilepath)
 
     def setJobId(self, jid, submitterName):
-        print("[BaseNode] (setJobId)", self.label, "->", jid, f"({submitterName})")
         self._nodeStatus.setJob(jid, submitterName)
         self.saveNodeStatusFile()
 
     def initStatusOnSubmit(self, forceCompute=False):
         """ Prepare chunks status when the node is in a graph that was submitted """
-        print(f"[BaseNode] (initStatusOnSubmit) {self.label} ({self})")
         if not self._chunksCreated:
             self._nodeStatus.setNode(self)  # ???
             self._nodeStatus.initExternSubmit()
@@ -1600,7 +1587,6 @@ def initStatusOnSubmit(self, forceCompute=False):
                 chunk.upgradeStatusFile()
 
     def beginSequence(self, forceCompute=False):
-        print(f"[BaseNode] (beginSequence) {self.label} ({self})")
         if not self._chunksCreated:
             self._nodeStatus.setNode(self)  # ???
             self._nodeStatus.initLocalSubmit()
@@ -1721,7 +1707,9 @@ def endSequence(self):
 
     def stopComputation(self):
         """ Stop the computation of this node. """
+        print(f"[BaseNode] (stopComputation) {self}")
         for chunk in self._chunks.values():
+            print(f"[BaseNode] (stopComputation) stop chunk {chunk}")
             chunk.stopProcess()
 
     def getGlobalStatus(self):
@@ -1807,7 +1795,6 @@ def getLocked(self):
         return self._locked
 
     def setLocked(self, lock):
-        print(f"(setLocked) {self.label} ({self}) -> {lock}")
         if self._locked == lock:
             return
         self._locked = lock
@@ -1816,17 +1803,14 @@ def setLocked(self, lock):
     @Slot()
     def updateDuplicatesStatusAndLocked(self):
         """ Update status of duplicate nodes without any latency and update locked. """
-        print(f"[BaseNode] (updateDuplicatesStatusAndLocked) {self}")
         if self.isMainNode():
             for node in self._duplicates:
                 node.updateStatusFromCache()
 
             self.updateLocked()
 
     def updateLocked(self):
-        print(f"(updateLocked) {self.label} ({self})")
         currentStatus = self.getGlobalStatus()
-        print(f"  currentStatus: {currentStatus}")
 
         lockedStatus = (Status.RUNNING, Status.SUBMITTED)
 
@@ -1850,7 +1834,6 @@ def updateLocked(self):
         if not self._locked and currentStatus == Status.SUCCESS:
             return
 
-        print(f"  a")
         if currentStatus == Status.SUCCESS:
             # At this moment, the node is necessarily locked because of previous if statement
             inputNodes = self.getInputNodes(recursive=True, dependenciesOnly=True)
@@ -1869,14 +1852,12 @@ def updateLocked(self):
                     node.setLocked(False)
             return
         elif currentStatus in lockedStatus and self.isMainNode():
-            print(f"  b")
             self.setLocked(True)
             inputNodes = self.getInputNodes(recursive=True, dependenciesOnly=True)
             for node in inputNodes:
                 node.setLocked(True)
             return
 
-        print(f"  c")
         self.setLocked(False)
 
     def updateDuplicates(self, nodesPerUid):
@@ -2178,15 +2159,13 @@ def _resetChunks(self):
         """
         if isinstance(self.nodeDesc, desc.InputNode):
             return
-        print("[Node] (_resetChunks)", self.label)
         # Disconnect signals
         for chunk in self._chunks:
-            print(f"[Node] (_resetChunks) -> remove {chunk}")
             chunk.statusChanged.disconnect(self.globalStatusChanged)
         # Empty list
         self._chunks.setObjectList([])
         # Clear cache
-
+        self._nodeStatus.resetChunkInfos()
         # Recreate list with reset values (1 chunk or the static size)
         if not self.isParallelized:
             self._chunksCreated = True
@@ -2198,8 +2177,6 @@ def _resetChunks(self):
             self._chunksCreated = False
             self.setSize(0)
         # Create chunks when possible 
-        print(f"[Node] (_resetChunks) <{self.label}|{self._uid[:5] + ('..' if len(self._uid) > 5 else '')}> ({self.size})", end="")
-        print(f" parallel={1 if self.isParallelized else 0} chunksCreated={1 if self._chunksCreated else 0}", end="")
         if self._chunksCreated and self.isParallelized:
             try:
                 ranges = self.nodeDesc.parallelization.getRanges(self)
@@ -2216,7 +2193,6 @@ def _resetChunks(self):
             self._chunks[0].statusChanged.connect(self.globalStatusChanged)
         else:
             self._chunks.setObjectList([])
-        print(f" -> ranges={[_c.range for _c in self._chunks]}")
         self.chunksCreatedChanged.emit()
         self.chunksChanged.emit()
         self.globalStatusChanged.emit()
@@ -2251,14 +2227,14 @@ def __createChunks(self, ranges):
 
     def _createChunksFromCache(self):
         """Create chunks when a node cache exists"""
-        print("[Node] (_createChunksFromCache)", self.label, end="")
+        # print("[Node] (_createChunksFromCache)", self.label, end="")
         try:
             # Get size from cache
             size = self._nodeStatus.nbChunks
             self.setSize(size)
             ranges = self._nodeStatus.getChunkRanges()
             self.__createChunks(ranges)
-            print(" -> ranges =", [_c.range for _c in self._chunks])
+            # print(" -> ranges =", [_c.range for _c in self._chunks])
         except Exception as e:
             logging.error(f"Failed to create chunks for {self.name}")
             self._chunks.clear()