Merge pull request #2918 from alicevision/dev/delayChunkEvaluation

[core] Add support for dynamic chunks

Author: cbentejac

.git-blame-ignore-revs

@@ -1,3 +1,7 @@
+# Linting: Remove trailing whitespaces
+04a425decc1b80f0c67e8c4c98c0062d73836684
+# [core] Linting: Remove all trailing whitespaces
+8be302115edca60c93b1e97de3f457d91c271666
 # [tests] Linting: Remove trailing whitespaces
 5fe886b6b08fa19082dc0e1bf837fa34c2e2de2d
 # [core] Linting: Remove remaining trailing whitespaces

bin/meshroom_compute

@@ -3,6 +3,7 @@ import argparse
 import logging
 import os
 import sys
+from typing import NoReturn
 
 try:
     import meshroom
@@ -16,7 +17,7 @@ meshroom.setupEnvironment()
 
 import meshroom.core
 import meshroom.core.graph
-from meshroom.core.node import Status, ExecMode
+from meshroom.core.node import Status
 
 
 parser = argparse.ArgumentParser(description='Execute a Graph of processes.')
@@ -63,12 +64,28 @@ else:
 
 meshroom.core.initPlugins()
 meshroom.core.initNodes()
+meshroom.core.initSubmitters()
 
 graph = meshroom.core.graph.loadGraph(args.graphFile)
 if args.cache:
     graph.cacheDir = args.cache
 graph.update()
 
+
+def killRunningJob(node) -> NoReturn:
+    """ Kills current job and try to avoid job restarting """
+    jobInfo = node.nodeStatus.jobInfo
+    submitterName = jobInfo.get("submitterName")
+    if not submitterName:
+        sys.exit(meshroom.MeshroomExitStatus.ERROR_NO_RETRY)
+    from meshroom.core import submitters
+    for subName, sub in submitters.items():
+        if submitterName == subName:
+            sub.killRunningJob()
+            break
+    sys.exit(meshroom.MeshroomExitStatus.ERROR_NO_RETRY)
+
+
 if args.node:
     node = graph.findNode(args.node)
     submittedStatuses = [Status.RUNNING]
@@ -83,6 +100,15 @@ if args.node:
         # If running as "extern", the task is supposed to have the status SUBMITTED.
         # If not running as "extern", the SUBMITTED status should generate a warning.
         submittedStatuses.append(Status.SUBMITTED)
+
+    if not node._chunksCreated:
+        print(f"Error: Node {node} has been submitted before chunks have been created." \
+               "See file: \"{node.nodeStatusFile}\".")
+        sys.exit(-1)
+
+    if node._isInputNode():
+        print(f"InputNode: No computation to do.")
+
     if not args.forceStatus and not args.forceCompute:
         if args.iteration != -1:
             chunks = [node.chunks[args.iteration]]
@@ -91,10 +117,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:
@@ -105,8 +132,14 @@ if args.node:
     node.preprocess()
     if args.iteration != -1:
         chunk = node.chunks[args.iteration]
+        if chunk._status.status == Status.STOPPED:
+            print(f"Chunk {chunk} : status is STOPPED")
+            killRunningJob(node)
         chunk.process(args.forceCompute, args.inCurrentEnv)
     else:
+        if node.nodeStatus.status == Status.STOPPED:
+            print(f"Node {node} : status is STOPPED")
+            killRunningJob(node)
         node.process(args.forceCompute, args.inCurrentEnv)
     node.postprocess()
     node.restoreLogger()

bin/meshroom_createChunks

@@ -0,0 +1,145 @@
+#!/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:
+    # If meshroom module is not in the PYTHONPATH, add our root using the relative path
+    import pathlib
+    meshroomRootFolder = pathlib.Path(__file__).parent.parent.resolve()
+    sys.path.append(meshroomRootFolder)
+    import meshroom
+meshroom.setupEnvironment()
+
+import meshroom.core
+import meshroom.core.graph
+from meshroom.core import submitters
+from meshroom.core.submitter import SubmitterOptionsEnum
+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.',
+                    action='store_true')
+parser.add_argument('--forceStatus', help='Force computation if status is RUNNING or SUBMITTED.',
+                    action='store_true')
+parser.add_argument('--forceCompute', help='Compute in all cases even if already computed.',
+                    action='store_true')
+parser.add_argument('--extern', help='Use this option when you compute externally after submission to a render farm from meshroom.',
+                    action='store_true')
+parser.add_argument('--cache', metavar='FOLDER', type=str,
+                    default=None,
+                    help='Override the cache folder')
+parser.add_argument('-v', '--verbose',
+                    help='Set the verbosity level for logging:\n'
+                            '  - fatal: Show only critical errors.\n'
+                            '  - error: Show errors only.\n'
+                            '  - warning: Show warnings and errors.\n'
+                            '  - info: Show standard informational messages.\n'
+                            '  - debug: Show detailed debug information.\n'
+                            '  - trace: Show all messages, including trace-level details.',
+                    default=os.environ.get('MESHROOM_VERBOSE', 'info'),
+                    choices=['fatal', 'error', 'warning', 'info', 'debug', 'trace'])
+
+args = parser.parse_args()
+
+# For extern computation, we want to focus on the node computation log.
+# So, we avoid polluting the log with general warning about plugins, versions of nodes in file, etc.
+logging.getLogger().setLevel(level=logging.INFO)
+
+meshroom.core.initPlugins()
+meshroom.core.initNodes()
+meshroom.core.initSubmitters()  # Required to spool child job
+
+graph = meshroom.core.graph.loadGraph(args.graphFile)
+if args.cache:
+    graph.cacheDir = args.cache
+graph.update()
+
+# Execute the node
+node = graph.findNode(args.node)
+submittedStatuses = [Status.RUNNING]
+
+# Find submitter
+submitter = None
+# It's required if we want to spool chunks on different machines
+for subName, sub in submitters.items():
+    if args.submitter == subName:
+        submitter = sub
+        break
+
+if node._nodeStatus.status in (Status.STOPPED, Status.KILLED):
+    logging.error("Node status is STOPPED or KILLED.")
+    if submitter:
+        submitter.killRunningJob()
+    sys.exit(meshroom.MeshroomExitStatus.ERROR_NO_RETRY)
+
+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()
+    # Set the chunks statuses
+    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 = []
+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(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 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}, computeSessionUid: {chunk.status.computeSessionUid}, " \
+                f"submitterSessionUid: {node._nodeStatus.submitterSessionUid}")
+
+if chunksToProcess:
+    node.prepareLogger()
+    node.preprocess()
+    for chunk in chunksToProcess:
+        logging.info(f"[MeshroomCreateChunks] process chunk {chunk}")
+        chunk.process(args.forceCompute, args.inCurrentEnv)
+    node.postprocess()
+    node.restoreLogger()
+else:
+    logging.info(f"[MeshroomCreateChunks] -> create job to process chunks {node.chunks}")
+    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/__init__.py

@@ -1,4 +1,4 @@
-from enum import Enum
+from enum import Enum, IntEnum
 import logging
 import os
 import sys
@@ -75,6 +75,18 @@ def logToRoot(message, *args, **kwargs):
 logging.getLogger().setLevel(logStringToPython[os.environ.get('MESHROOM_VERBOSE', 'warning')])
 
 
+class MeshroomExitStatus(IntEnum):
+    """ In case we want to catch some special case from the parent process
+    We could use 3-125 for custom exist codes :
+    https://tldp.org/LDP/abs/html/exitcodes.html
+    """
+    SUCCESS = 0
+    ERROR = 1
+    # In some farm tools jobs are automatically re-tried, 
+    # using ERROR_NO_RETRY will try to prevent that
+    ERROR_NO_RETRY = -999  # It's actually -999 % 256 => 25
+
+
 def setupEnvironment(backend=Backend.STANDALONE):
     """
     Setup environment for Meshroom to work in a prebuilt, standalone configuration.

meshroom/core/__init__.py

@@ -119,7 +119,7 @@ def loadClasses(folder: str, packageName: str, classType: type) -> list[type]:
                         classes.append(p)
             except Exception as exc:
                 if classType == BaseSubmitter:
-                    logging.warning(f" Could not load submitter {pluginName} from package '{package.__name__}'")
+                    logging.warning(f" Could not load submitter {pluginName} from package '{package.__name__}'\n{exc}")
                 else:
                     tb = traceback.extract_tb(exc.__traceback__)
                     last_call = tb[-1]

meshroom/core/desc/computation.py

@@ -1,13 +1,15 @@
 import math
-from enum import Enum
+from enum import IntEnum
 
 from .attribute import ListAttribute, IntParam
 
 
-class Level(Enum):
+class Level(IntEnum):
+    SCRIPT=-1
     NONE = 0
     NORMAL = 1
     INTENSIVE = 2
+    EXTREME = 3
 
 
 class Range:
@@ -46,6 +48,9 @@ def toDict(self):
             "rangeBlocksCount": self.nbBlocks
             }
 
+    def __repr__(self):
+        return f"<Range {self.iteration}({self.blockSize})/{self.nbBlocks}({self.fullSize})>"
+
 
 class Parallelization:
     def __init__(self, staticNbBlocks=0, blockSize=0):

meshroom/core/desc/node.py

@@ -5,6 +5,8 @@
 import shlex
 import shutil
 import sys
+import signal
+import subprocess
 
 import psutil
 
@@ -20,6 +22,34 @@
 _MESHROOM_COMPUTE_DEPS = ["psutil"]
 
 
+# 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):
+        logging.debug(f"[ExitCleanup] Register subprocess {process}")
+        self._subprocesses.append(process)
+
+    def exit(self, signum, frame):
+        for proc in self._subprocesses:
+            logging.debug(f"[ExitCleanup] Kill subprocess {proc}")
+            try:
+                if proc.is_running():
+                    proc.terminate()
+                    proc.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                proc.kill()
+        sys.exit(0)
+
+exitCleanup = ExitCleanup()
+
+
 class MrNodeType(enum.Enum):
     NONE = enum.auto()
     BASENODE = enum.auto()
@@ -90,6 +120,9 @@ class BaseNode(object):
     documentation = ""
     category = "Other"
     plugin = None
+    # Licenses required to run the plugin
+    # Only used to select machines on the farm when the node is submitted
+    _licenses = []
 
     def __init__(self):
         super(BaseNode, self).__init__()
@@ -158,7 +191,7 @@ def processChunk(self, chunk):
 
     def executeChunkCommandLine(self, chunk, cmd, env=None):
         try:
-            with open(chunk.logFile, 'a') as logF:
+            with open(chunk.getLogFile(), 'a') as logF:
                 chunk.status.commandLine = cmd
                 chunk.saveStatusFile()
                 cmdList = shlex.split(cmd)
@@ -167,7 +200,7 @@ def executeChunkCommandLine(self, chunk, cmd, env=None):
 
                 print(f"Starting Process for '{chunk.node.name}'")
                 print(f" - commandLine: {cmd}")
-                print(f" - logFile: {chunk.logFile}")
+                print(f" - logFile: {chunk.getLogFile()}")
                 if prog:
                     cmdList[0] = Path(prog).as_posix()
                     print(f" - command full path: {cmdList[0]}")
@@ -193,6 +226,7 @@ def executeChunkCommandLine(self, chunk, cmd, env=None):
                     text=True,
                     **platformArgs,
                 )
+                exitCleanup.addSubprocess(chunk.subprocess)
 
                 if hasattr(chunk, "statThread"):
                     # We only have a statThread if the node is running in the current process
@@ -213,7 +247,7 @@ def executeChunkCommandLine(self, chunk, cmd, env=None):
                         pass
 
             if chunk.subprocess.returncode != 0:
-                with open(chunk.logFile, "r") as logF:
+                with open(chunk.getLogFile(), "r") as logF:
                     logContent = "".join(logF.readlines())
                 raise RuntimeError(f'Error on node "{chunk.name}":\nLog:\n{logContent}')
         finally: