New notion of local isolated computation for python nodes using meshroom_compute

Reoganization
- BaseNode: is the base class for all nodes
- Node: is now dedicated to python nodes, with the implentation directly
in the process function
- CommandLineNode: dedicated to generate and run external command lines

Author: fabiencastan

bin/meshroom_compute

@@ -16,7 +16,7 @@ meshroom.setupEnvironment()
 
 import meshroom.core
 import meshroom.core.graph
-from meshroom.core.node import Status
+from meshroom.core.node import Status, ExecMode
 
 
 parser = argparse.ArgumentParser(description='Execute a Graph of processes.')
@@ -26,6 +26,8 @@ parser.add_argument('--node', metavar='NODE_NAME', type=str,
                     help='Process the node. It will generate an error if the dependencies are not already computed.')
 parser.add_argument('--toNode', metavar='NODE_NAME', type=str,
                     help='Process the node with its dependencies.')
+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.',
@@ -81,7 +83,11 @@ if args.node:
             chunks = node.chunks
         for chunk in chunks:
             if chunk.status.status in submittedStatuses:
-                print('Warning: Node is already submitted with status "{}". See file: "{}"'.format(chunk.status.status.name, chunk.statusFile))
+                # 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.status.execMode == ExecMode.LOCAL_ISOLATED and not chunk.status.sessionUid and chunk.status.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}')
                 # sys.exit(-1)
 
     if args.extern:
@@ -91,9 +97,9 @@ if args.node:
     node.preprocess()
     if args.iteration != -1:
         chunk = node.chunks[args.iteration]
-        chunk.process(args.forceCompute)
+        chunk.process(args.forceCompute, args.inCurrentEnv)
     else:
-        node.process(args.forceCompute)
+        node.process(args.forceCompute, args.inCurrentEnv)
     node.postprocess()
 else:
     if args.iteration != -1:

meshroom/core/desc/__init__.py

@@ -21,36 +21,9 @@
 )
 from .node import (
     AVCommandLineNode,
+    BaseNode,
     CommandLineNode,
     InitNode,
     InputNode,
     Node,
 )
-
-__all__ = [
-    # attribute
-    "Attribute",
-    "BoolParam",
-    "ChoiceParam",
-    "ColorParam",
-    "File",
-    "FloatParam",
-    "GroupAttribute",
-    "IntParam",
-    "ListAttribute",
-    "PushButtonParam",
-    "StringParam",
-    # computation
-    "DynamicNodeSize",
-    "Level",
-    "MultiDynamicNodeSize",
-    "Parallelization",
-    "Range",
-    "StaticNodeSize",
-    # node
-    "AVCommandLineNode",
-    "CommandLineNode",
-    "InitNode",
-    "InputNode",
-    "Node",
-]

meshroom/core/desc/node.py

@@ -1,16 +1,34 @@
 from inspect import getfile
 from pathlib import Path
+import logging
 import os
 import psutil
 import shlex
+import shutil
+import sys
 
 from .computation import Level, StaticNodeSize
 from .attribute import StringParam, ColorParam
+from .process import ProcessEnvironment, _MESHROOM_COMPUTE
+from .config import getProcessEnvironment
 
 from meshroom.core import cgroup
 
 
-class Node(object):
+def isNodeSaved(node):
+    """Returns whether a node is identical to its serialized counterpart in the current graph file."""
+    filepath = node.graph.filepath
+    if not filepath:
+        return False
+
+    from meshroom.core.graph import loadGraph
+    graphSaved = loadGraph(filepath)
+    nodeSaved = graphSaved.node(node.name)
+    if nodeSaved is None:
+        return False
+    return nodeSaved._uid == node._uid
+
+class BaseNode(object):
     """
     """
     cpu = Level.NORMAL
@@ -62,7 +80,7 @@ class Node(object):
     category = 'Other'
 
     def __init__(self):
-        super(Node, self).__init__()
+        super(BaseNode, self).__init__()
         self.hasDynamicOutputAttribute = any(output.isDynamicValue for output in self.outputs)
         self.sourceCodeFolder = Path(getfile(self.__class__)).parent.resolve().as_posix()
 
@@ -113,13 +131,102 @@ def postprocess(self, node):
         pass
 
     def stopProcess(self, chunk):
-        raise NotImplementedError('No stopProcess implementation on node: {}'.format(chunk.node.name))
+        logging.warning(f'No stopProcess implementation on node: {chunk.node.name}')
 
     def processChunk(self, chunk):
         raise NotImplementedError(f'No processChunk implementation on node: "{chunk.node.name}"')
 
+    def executeChunkCommandLine(self, chunk, cmd, env=None):
+        try:
+            with open(chunk.logFile, 'w') as logF:
+                chunk.status.commandLine = cmd
+                chunk.saveStatusFile()
+                cmdList = shlex.split(cmd)
+                # Resolve executable to full path
+                prog = shutil.which(cmdList[0], path=env.get('PATH') if env else None)
+
+                print(f"Starting Process for '{chunk.node.name}'")
+                print(f' - commandLine: {cmd}')
+                print(f' - logFile: {chunk.logFile}')
+                if prog:
+                    cmdList[0] = prog
+                    print(f' - command full path: {prog}')
+
+                # Change the process group to avoid Meshroom main process being killed if the subprocess
+                # gets terminated by the user or an Out Of Memory (OOM kill).
+                if sys.platform == "win32":
+                    platformArgs = {"creationflags": psutil.CREATE_NEW_PROCESS_GROUP}
+                    # Note: DETACHED_PROCESS means fully detached process.
+                    # We don't want a fully detached process to ensure that if Meshroom is killed,
+                    # the subprocesses are killed too.
+                else:
+                    platformArgs = {"start_new_session": True}
+                    # Note: "preexec_fn"=os.setsid is the old way before python-3.2
+
+                chunk.subprocess = psutil.Popen(
+                    cmdList,
+                    stdout=logF,
+                    stderr=logF,
+                    cwd=chunk.node.internalFolder,
+                    env=env,
+                    **platformArgs,
+                )
+
+                if hasattr(chunk, "statThread"):
+                    # We only have a statThread if the node is running in the current process
+                    # and not in a dedicated environment/process.
+                    chunk.statThread.proc = chunk.subprocess
+
+                stdout, stderr = chunk.subprocess.communicate()
 
-class InputNode(Node):
+                chunk.status.returnCode = chunk.subprocess.returncode
+
+                if chunk.subprocess.returncode and chunk.subprocess.returncode < 0:
+                    signal_num = -chunk.subprocess.returncode
+                    logF.write(f"Process was killed by signal: {signal_num}")
+                    try:
+                        status = chunk.subprocess.status()
+                        logF.write(f"Process status: {status}")
+                    except:
+                        pass
+
+            if chunk.subprocess.returncode != 0:
+                with open(chunk.logFile, 'r') as logF:
+                    logContent = ''.join(logF.readlines())
+                raise RuntimeError('Error on node "{}":\nLog:\n{}'.format(chunk.name, logContent))
+        finally:
+            chunk.subprocess = None
+
+    def stopProcess(self, chunk):
+        # The same node could exists several times in the graph and
+        # only one would have the running subprocess; ignore all others
+        if not chunk.subprocess:
+            print(f"[{chunk.node.name}] stopProcess: no subprocess")
+            return
+
+        # Retrieve process tree
+        processes = chunk.subprocess.children(recursive=True) + [chunk.subprocess]
+        logging.debug(f"[{chunk.node.name}] Processes to stop: {len(processes)}")
+        for process in processes:
+            try:
+                # With terminate, the process has a chance to handle cleanup
+                process.terminate()
+            except psutil.NoSuchProcess:
+                pass
+
+        # If it is still running, force kill it
+        for process in processes:
+            try:
+                # Use is_running() instead of poll() as we use a psutil.Process object
+                if process.is_running():  # Check if process is still alive
+                    process.kill()  # Forcefully kill it
+            except psutil.NoSuchProcess:
+                logging.info(f"[{chunk.node.name}] Process already terminated.")
+            except psutil.AccessDenied:
+                logging.info(f"[{chunk.node.name}] Permission denied to kill the process.")
+
+
+class InputNode(BaseNode):
     """
     Node that does not need to be processed, it is just a placeholder for inputs.
     """
@@ -130,7 +237,24 @@ def processChunk(self, chunk):
         pass
 
 
-class CommandLineNode(Node):
+class Node(BaseNode):
+
+    def __init__(self):
+        super(Node, self).__init__()
+
+    def processChunkInEnvironment(self, chunk):
+        if not isNodeSaved(chunk.node):
+            raise RuntimeError("File must be saved before computing in isolated environment.")
+
+        meshroomComputeCmd = f"python {_MESHROOM_COMPUTE} {chunk.node.graph.filepath} --node {chunk.node.name} --extern --inCurrentEnv"
+        if len(chunk.node.getChunks()) > 1:
+            meshroomComputeCmd += f" --iteration {chunk.range.iteration}"
+
+        runtimeEnv = None
+        self.executeChunkCommandLine(chunk, meshroomComputeCmd, env=runtimeEnv)
+
+
+class CommandLineNode(BaseNode):
     """
     """
     commandLine = ''  # need to be defined on the node
@@ -143,63 +267,25 @@ def __init__(self):
     def buildCommandLine(self, chunk):
 
         cmdPrefix = ''
-        # If rez available in env, we use it
-        if "REZ_ENV" in os.environ and chunk.node.packageVersion:
-            # If the node package is already in the environment, we don't need a new dedicated rez environment
-            alreadyInEnv = os.environ.get("REZ_{}_VERSION".format(chunk.node.packageName.upper()),
-                                          "").startswith(chunk.node.packageVersion)
-            if not alreadyInEnv:
-                cmdPrefix = '{rez} {packageFullName} -- '.format(rez=os.environ.get("REZ_ENV"),
-                                                                 packageFullName=chunk.node.packageFullName)
+        # # If rez available in env, we use it
+        # if "REZ_ENV" in os.environ and chunk.node.packageVersion:
+        #     # If the node package is already in the environment, we don't need a new dedicated rez environment
+        #     alreadyInEnv = os.environ.get("REZ_{}_VERSION".format(chunk.node.packageName.upper()),
+        #                                   "").startswith(chunk.node.packageVersion)
+        #     if not alreadyInEnv:
+        #         cmdPrefix = '{rez} {packageFullName} -- '.format(rez=os.environ.get("REZ_ENV"),
+        #                                                          packageFullName=chunk.node.packageFullName)
 
         cmdSuffix = ''
         if chunk.node.isParallelized and chunk.node.size > 1:
             cmdSuffix = ' ' + self.commandLineRange.format(**chunk.range.toDict())
 
         return cmdPrefix + chunk.node.nodeDesc.commandLine.format(**chunk.node._cmdVars) + cmdSuffix
 
-    def stopProcess(self, chunk):
-        # The same node could exists several times in the graph and
-        # only one would have the running subprocess; ignore all others
-        if not hasattr(chunk, "subprocess"):
-            return
-        if chunk.subprocess:
-            # Kill process tree
-            processes = chunk.subprocess.children(recursive=True) + [chunk.subprocess]
-            try:
-                for process in processes:
-                    process.terminate()
-            except psutil.NoSuchProcess:
-                pass
-
     def processChunk(self, chunk):
-        try:
-            with open(chunk.logFile, 'w') as logF:
-                cmd = self.buildCommandLine(chunk)
-                chunk.status.commandLine = cmd
-                chunk.saveStatusFile()
-                print(' - commandLine: {}'.format(cmd))
-                print(' - logFile: {}'.format(chunk.logFile))
-                chunk.subprocess = psutil.Popen(shlex.split(cmd), stdout=logF, stderr=logF, cwd=chunk.node.internalFolder)
-
-                # Store process static info into the status file
-                # chunk.status.env = node.proc.environ()
-                # chunk.status.createTime = node.proc.create_time()
-
-                chunk.statThread.proc = chunk.subprocess
-                stdout, stderr = chunk.subprocess.communicate()
-                chunk.subprocess.wait()
-
-                chunk.status.returnCode = chunk.subprocess.returncode
-
-            if chunk.subprocess.returncode != 0:
-                with open(chunk.logFile, 'r') as logF:
-                    logContent = ''.join(logF.readlines())
-                raise RuntimeError('Error on node "{}":\nLog:\n{}'.format(chunk.name, logContent))
-        except Exception:
-            raise
-        finally:
-            chunk.subprocess = None
+        cmd = self.buildCommandLine(chunk)
+        # TODO: Setup runtime env
+        self.executeChunkCommandLine(chunk, cmd)
 
 
 # Specific command line node for AliceVision apps
@@ -282,3 +368,4 @@ def setAttributes(self, node, attributesDict):
         for attr in attributesDict:
             if node.hasAttribute(attr):
                 node.attribute(attr).value = attributesDict[attr]
+