[core] pre/post tasks : add submitter tests

Author: Alxiice

meshroom/core/node.py

@@ -837,8 +837,10 @@ def __init__(self, nodeType: str, position: Position = None, parent: BaseObject
         self.graph = None
         self.dirty: bool = True  # whether this node's outputs must be re-evaluated on next Graph update
         self._chunks: list[NodeChunk] = ListModel(parent=self)
-        self._preprocessChunk = NodeChunk(self, desc.Range(ChunkIndex.PREPROCESS)) if self.nodeDesc._hasPreprocess else None
-        self._postprocessChunk = NodeChunk(self, desc.Range(ChunkIndex.POSTPROCESS)) if self.nodeDesc._hasPostprocess else None
+        self._preprocessChunk = NodeChunk(self, desc.Range(ChunkIndex.PREPROCESS)) if \
+            self.nodeDesc and self.nodeDesc._hasPreprocess else None
+        self._postprocessChunk = NodeChunk(self, desc.Range(ChunkIndex.POSTPROCESS)) if \
+            self.nodeDesc and self.nodeDesc._hasPostprocess else None
         self._chunksCreated = False  # Only initialize chunks on compute
         self._chunkPlaceholder: list[NodeChunk] = ListModel(parent=self)  # Placeholder chunk for nodes with dynamic ones
         self._uid: str = uid
@@ -1714,10 +1716,10 @@ def processIteration(self, iteration):
 
     def preprocess(self, forceCompute=False, inCurrentEnv=False):
         """ Prepare the node processing """
-        self.prepareLogger(ChunkIndex.PREPROCESS)
         if self.nodeDesc._hasPreprocess:
+            self.prepareLogger(ChunkIndex.PREPROCESS)
             self._preprocessChunk.process(forceCompute, inCurrentEnv)
-        self.restoreLogger()
+            self.restoreLogger()
 
     def process(self, forceCompute=False, inCurrentEnv=False):
         for chunk in self._chunks:
@@ -1728,10 +1730,10 @@ def postprocess(self, forceCompute=False, inCurrentEnv=False):
         Invoke the post process on Client Node to execute after the processing on the
         node is completed
         """
-        self.prepareLogger(ChunkIndex.POSTPROCESS)
         if self.nodeDesc._hasPostprocess:
+            self.prepareLogger(ChunkIndex.POSTPROCESS)
             self._postprocessChunk.process(forceCompute, inCurrentEnv)
-        self.restoreLogger()
+            self.restoreLogger()
 
     def getLogHandlers(self):
         return self._handlers

meshroom/core/submitter.py

@@ -104,7 +104,7 @@ def __repr__(self):
 
 
 class OrderedNode:
-    """ A task = a node"""
+    """ Intermediate structure used to order tasks """
 
     def __init__(self, node, dependencies=None):
         # node can be None for placeholder tasks (task that don't do anything else than regrouping dependencies)
@@ -214,32 +214,32 @@ def createNodeTasks(self, orderedNode: OrderedNode, parentTask: OrderedTask):
         """ Create tasks corresponding to a node and link them correctly.
         Also link them to the parent task, and recursively create children tasks.
         """
-        print(f"* (createNodeTasks) node {orderedNode.node._name}, parent {parentTask.node}")
+        logger.debug(f"* (createNodeTasks) node {orderedNode.node._name}, parent {parentTask.node}")
         # Check if task has already been created
         visited = (nodeUid:=orderedNode.node._uid) in self._nodeUidToLastTask
         if visited:
-            print("  -> is visited")
+            logger.debug("  -> is visited")
             # If task is already created simply create the connection
             lastTask = self._nodeUidToLastTask[nodeUid]
             parentTask.addDependency(lastTask)
             return
         # Create node tasks
         if orderedNode.isPlaceholder:
-            print("  -> is placeholder")
+            logger.debug("  -> is placeholder")
             task = OrderedTask(OrderedTaskType.PLACEHOLDER, orderedNode)
             firstTask = lastTask = task
         else:
             lastTask = firstTask = None
             # Create pre/post tasks if needed
             if orderedNode.hasPostprocess:
-                print("  -> postprocess")
+                logger.debug("  -> postprocess")
                 lastTask = OrderedTask(OrderedTaskType.POSTPROCESS, orderedNode)
             if orderedNode.hasPreprocess:
-                print("  -> preprocess")
+                logger.debug("  -> preprocess")
                 firstTask = OrderedTask(OrderedTaskType.PREPROCESS, orderedNode)
             # Process
             if orderedNode.isExpanding:
-                print("  -> is expanding")
+                logger.debug("  -> is expanding")
                 expandingTask = OrderedTask(OrderedTaskType.EXPANDING, orderedNode)
                 if lastTask:
                     lastTask.addDependency(expandingTask)
@@ -250,7 +250,7 @@ def createNodeTasks(self, orderedNode: OrderedNode, parentTask: OrderedTask):
                 else:
                     firstTask = expandingTask
             else:
-                print("  -> has chunks :", orderedNode.chunksIterations)
+                logger.debug(f"  -> has chunks : {orderedNode.chunksIterations}")
                 # Handle 0 chunks case
                 if len(orderedNode.chunksIterations) == 0:
                     if firstTask and lastTask:
@@ -267,19 +267,19 @@ def createNodeTasks(self, orderedNode: OrderedNode, parentTask: OrderedTask):
                     lastTask = lastTask if lastTask else OrderedTask(OrderedTaskType.PLACEHOLDER, orderedNode)
                     firstTask = firstTask if firstTask else OrderedTask(OrderedTaskType.PLACEHOLDER, orderedNode)
                     for iteration in orderedNode.chunksIterations:
-                        print(f"    - chunk {iteration}")
+                        logger.debug(f"    - chunk {iteration}")
                         chunkTask = OrderedTask(OrderedTaskType.CHUNK, orderedNode, iteration=iteration)
                         lastTask.addDependency(chunkTask)
                         chunkTask.addDependency(firstTask)
         # Add parent dependency
         parentTask.addDependency(lastTask)
         # Create children
         for n in orderedNode.dependencies:
-            print("  -> create deps", n)
+            logger.debug(f"  -> create deps {n}")
             self.createNodeTasks(n, firstTask)
         # Add the last task to execute for this node to _nodeUidToLastTask
         self._nodeUidToLastTask[nodeUid] = lastTask
-        print(f"  -> done {orderedNode.node._name}")
+        logger.debug(f"  -> done {orderedNode.node._name}")
 
     def _orderTasks(self):
         """ Use the nodesByLevel info to create all tasks to send to the submitter """
@@ -292,7 +292,7 @@ def _orderTasks(self):
     def display(self, task:OrderedTask=None, level=0):
         if task is None:
             task = self.rootTask
-        print(f"{' '*4*level}[{level:02d}] {task}")
+        logger.debug(f"{' '*4*level}[{level:02d}] {task}")
         for child in task.dependencies:
             self.display(child, level+1)
 

tests/plugins/meshroom/pluginSubmitter/PluginSubmitter.py

@@ -8,7 +8,7 @@
 LOGGER = logging.getLogger("TestSubmit")
 
 
-class PluginSubmitterA(desc.BaseNode):
+class PluginSubmitterA(desc.Node):
     """
     Test process no parallelization
     """
@@ -58,27 +58,19 @@ class PluginSubmitterB(PluginSubmitterA):
     size = desc.StaticNodeSize(2)
     parallelization = desc.Parallelization(blockSize=1)
 
+    def postprocess(self, node):
+        LOGGER.info(f"> PluginSubmitterB postprocess Done")
+
 
 class PluginSubmitterC(PluginSubmitterA):
     """
     Test process with parallelization and dynamic node size
     """
     size = desc.DynamicNodeSize("nbChunks")
     parallelization = desc.Parallelization(blockSize=1)
-
-
-class PluginSubmitterAPrePost(PluginSubmitterA):
+    
     def preprocess(self, node):
-        LOGGER.info(f"> Done")
-    def postprocess(self, node):
-        LOGGER.info(f"> Done")
-
+        LOGGER.info(f"> PluginSubmitterC preprocess Done")
 
-class PluginSubmitterBPrePost(PluginSubmitterB):
     def postprocess(self, node):
-        LOGGER.info(f"> Done")
-
-
-class PluginSubmitterCPrePost(PluginSubmitterC):
-    def postprocess(self, node):
-        LOGGER.info(f"> Done")
+        LOGGER.info(f"> PluginSubmitterC postprocess Done")

tests/test_compute.py

@@ -13,7 +13,7 @@
 
 from meshroom.core.graph import Graph, loadGraph
 from meshroom.core import desc, pluginManager, loadClassesNodes
-from meshroom.core.node import Status
+from meshroom.core.node import Status, ChunkIndex
 from meshroom.core.plugins import Plugin
 from .utils import registerNodeDesc, unregisterNodeDesc
 
@@ -25,7 +25,8 @@ def executeChunks(node, size):
     logFiles = {}
     node.preprocess()
     for chunkIndex in range(size):
-        iteration = chunkIndex if size > 1 else -1
+        iteration = chunkIndex if size >= 0 else ChunkIndex.NONE
+        node.prepareLogger(iteration)
         logFileName = f"{chunkIndex}.log"
         logFile = Path(node.internalFolder) / logFileName
         logFiles[chunkIndex] = logFile

tests/test_submit.py

@@ -17,10 +17,15 @@
 from meshroom.core.node import Node, Status
 from meshroom.core.submitter import BaseSubmitter
 from meshroom.core.submitter import jobManager
+from meshroom.core.submitter import OrderedTask, OrderedTasks, OrderedTaskType
 from meshroom.submitters.localFarmSubmitter import LocalFarmSubmitter, LocalFarmJob
 
 from localfarm.localFarmLauncher import FarmLauncher
 
+import logging
+from meshroom.core.submitter import logger
+logger.setLevel(logging.DEBUG)
+
 
 IS_LINUX = (platform == "linux" or platform == "linux2")
 
@@ -133,26 +138,68 @@ def registerNode(self, name):
         registerNodeDesc(nodeType)
         return nodeType.__name__
 
-    def addNewNode(self, graph, name, nodeParams):
+    def addNewNode(self, graph, name, nodeParams=None):
         nodeTypeName = self.registerNode(name)
-        if nodeParams:
-            node = graph.addNewNode(nodeTypeName, **nodeParams)
-        else:
-            node = graph.addNewNode(nodeTypeName)
+        nodeParams = nodeParams or {}
+        node = graph.addNewNode(nodeTypeName, **nodeParams)
         return node
 
-    def test_buildTaskGraph(self):
+    def test_orderTasks(self):
+        """
+        phd=placeholder
+        chk=chunk
+                                                                    *" [B chk_0] "* 
+        [phd (start_A)] - [A chks] - [phd (end_A)] - [phd (start_B)]               [B post] - [C pre] - [C expand] - [C post] - [phd (root)]
+                                                                    *_ [B chk_1] _* 
+        """
         graph = Graph("")
         # Add nodes
-        nodeA = self.addNewNode(graph, "PluginSubmitter"+"A"+"PrePost", nodeParams={})
-        nodeB = self.addNewNode(graph, "PluginSubmitter"+"B"+"PrePost", nodeParams={"inputs": [nodeA.output]})
-        nodeC = self.addNewNode(graph, "PluginSubmitter"+"C"+"PrePost", nodeParams={"inputs": [nodeB.output]})
-        # Submit
+        nodeA = self.addNewNode(graph, "PluginSubmitter"+"A", nodeParams={})
+        nodeB = self.addNewNode(graph, "PluginSubmitter"+"B", nodeParams={"inputs": [nodeA.output]})
+        nodeC = self.addNewNode(graph, "PluginSubmitter"+"C", nodeParams={"inputs": [nodeB.output]})
+        # Order tasks
         submitter = get_submitter()
         nodes, edges = graph.dfsOnFinish(startNodes=[nodeC])
-        print(nodes, edges)
-        res = submitter.submit(nodes, edges, "")
-        print("res", res)
+        orderedTasks = OrderedTasks(nodes, edges)
+        # === Test result ===
+        def checkTask(task, taskType, nbDependencies):
+            assert task.taskType == taskType
+            assert len(task.dependencies) == nbDependencies
+        # root
+        rootTask = orderedTasks.rootTask
+        checkTask(rootTask, OrderedTaskType.PLACEHOLDER, 1)
+        # C (post)
+        task: OrderedTask = rootTask.dependencies[0]
+        checkTask(task, OrderedTaskType.POSTPROCESS, 1)
+        # C (expand)
+        task: OrderedTask = task.dependencies[0]
+        checkTask(task, OrderedTaskType.EXPANDING, 1)
+        # C (pre)
+        task: OrderedTask = task.dependencies[0]
+        checkTask(task, OrderedTaskType.PREPROCESS, 1)
+        # B (post)
+        task: OrderedTask = task.dependencies[0]
+        checkTask(task, OrderedTaskType.POSTPROCESS, 2)
+        # B (chunks)
+        task_0: OrderedTask = task.dependencies[0]
+        task_1: OrderedTask = task.dependencies[1]
+        checkTask(task_0, OrderedTaskType.CHUNK, 1)
+        checkTask(task_1, OrderedTaskType.CHUNK, 1)
+        assert (task_0.iteration, task_1.iteration) == (0, 1)
+        assert task_0.dependencies[0] == task_1.dependencies[0]
+        # B (pre)
+        task: OrderedTask = task_0.dependencies[0]
+        checkTask(task, OrderedTaskType.PLACEHOLDER, 1)
+        # A (post)
+        task: OrderedTask = task.dependencies[0]
+        checkTask(task, OrderedTaskType.PLACEHOLDER, 1)
+        # A (chunks)
+        task: OrderedTask = task.dependencies[0]
+        checkTask(task, OrderedTaskType.CHUNK, 1)
+        assert task.iteration == -1
+        # A (pre)
+        task: OrderedTask = task.dependencies[0]
+        checkTask(task, OrderedTaskType.PLACEHOLDER, 0)
 
     def test_submitNoParallel(self, tmp_path):
         graph = Graph("")