3pp-mirror/Meshroom
1
0
Fork 0
mirror of https://github.com/alicevision/Meshroom.git synced 2025-07-16 16:25:20 +02:00
Code Issues Projects Releases Packages Wiki Activity

Make Task Manager with Nodes submitted to a render farm

Browse source
This commit is contained in:
Lee Geertsen 2019-08-27 17:46:12 +02:00 • committed by Yann Lanthony
parent c00db25c23
commit 01974c23ec
No known key found for this signature in database
GPG key ID: 519FAE6DF7A70642
5 changed files with 108 additions and 57 deletions
Download patch file Download diff file Expand all files Collapse all files

119
meshroom/core/taskManager.py
View file

@ -2,6 +2,7 @@ import logging
from threading import Thread
from enum import Enum
import meshroom
from meshroom.common import BaseObject, DictModel, Property
class State(Enum):
@ -18,44 +19,50 @@ class TaskThread(Thread):
self._manager = manager
self.forceCompute = False
def isRunning(self):
return self._state == State.RUNNING
def run(self):
self._state = State.RUNNING
for n, node in enumerate(self._manager._nodesToProcess):
if not node.isFinishedOrRunning():
#try:
multiChunks = len(node.chunks) > 1
for c, chunk in enumerate(node.chunks):
if multiChunks:
print('\n[{node}/{nbNodes}]({chunk}/{nbChunks}) {nodeName}'.format(
node=n + 1, nbNodes=len(self._manager._nodesToProcess),
chunk=c + 1, nbChunks=len(node.chunks), nodeName=node.nodeType))
else:
print('\n[{node}/{nbNodes}] {nodeName}'.format(
node=n + 1, nbNodes=len(self._manager._nodesToProcess), nodeName=node.nodeType))
multiChunks = len(node.chunks) > 1
for c, chunk in enumerate(node.chunks):
if multiChunks:
print('\n[{node}/{nbNodes}]({chunk}/{nbChunks}) {nodeName}'.format(
node=n + 1, nbNodes=len(self._manager._nodesToProcess),
chunk=c + 1, nbChunks=len(node.chunks), nodeName=node.nodeType))
else:
print('\n[{node}/{nbNodes}] {nodeName}'.format(
node=n + 1, nbNodes=len(self._manager._nodesToProcess), nodeName=node.nodeType))
if not chunk.isFinishedOrRunning() and self._state == State.RUNNING:
try:
chunk.process(self.forceCompute)
if not chunk.isFinishedOrRunning() and self._state == State.RUNNING:
try:
chunk.process(self.forceCompute)
except Exception as e:
if chunk.isStopped():
self._state = State.STOPPED
self._manager._graph.clearSubmittedNodes()
self._manager._nodesToProcess.clear()
except Exception as e:
if chunk.isStopped():
self._state = State.STOPPED
self._manager._graph.clearSubmittedNodes()
self._manager._nodesToProcess.clear()
else:
logging.error("Error on node computation: {}".format(e))
nodesToDelete, _ = self._manager._graph.nodesFromNode(node)
else:
logging.error("Error on node computation: {}".format(e))
nodesToDelete, _ = self._manager._graph.nodesFromNode(node)
for nodeD in nodesToDelete:
if nodeD != node:
for nodeD in nodesToDelete:
if nodeD != node:
try:
self._manager._nodesToProcess.remove(nodeD)
nodeD.clearSubmittedChunks()
self._state = State.DEAD
except:
# Node already removed (for instance a global clear of _nodesToProcess)
pass
nodeD.clearSubmittedChunks()
self._manager._nodesToProcess.clear()
self._state = State.DEAD
self._manager._uigraph.computeStatusChanged.emit()
class TaskManager(BaseObject):
@ -63,16 +70,30 @@ class TaskManager(BaseObject):
super(TaskManager, self).__init__(parent)
self._nodes = DictModel(keyAttrName='name', parent=self)
self._nodesToProcess = DictModel(keyAttrName='name', parent=self)
self._nodesExtern = DictModel(keyAttrName='name', parent=self)
self._graph = None
self._thread = TaskThread(self)
def addNodes(self, graph=None, toNodes=None, forceCompute=False, forceStatus=False):
def compute(self, graph=None, toNodes=None, uigraph=None, forceCompute=False, forceStatus=False):
self._graph = graph
logging.info(self._thread._state)
self._uigraph = uigraph
if self._thread._state in (State.IDLE, State.DEAD, State.ERROR, State.STOPPED):
self._nodes.clear()
try:
self._nodes.clear()
except:
print("Task Manager nodes already empty")
externEmpty = True
for node in self._nodesExtern:
if node.isAlreadySubmitted():
externEmpty = False
break
if not externEmpty:
self._nodes.update(self._nodesExtern)
else:
self._nodesExtern.clear()
if forceCompute:
nodes, edges = graph.dfsOnFinish(startNodes=toNodes)
@ -88,7 +109,10 @@ class TaskManager(BaseObject):
for node in nodes:
node.beginSequence(forceCompute)
self._nodes.update(nodes)
try:
self._nodes.update(nodes)
except:
print("nodes already added to Task Manager")
self._nodesToProcess.update(nodes)
if self._thread._state == State.IDLE:
@ -97,6 +121,39 @@ class TaskManager(BaseObject):
self._thread = TaskThread(self)
self._thread.start()
def submit(self, graph=None, submitter=None, toNodes=None):
if self._thread._state in (State.IDLE, State.DEAD, State.ERROR, State.STOPPED):
self._nodes.clear()
externEmpty = True
for node in self._nodesExtern:
if node.isAlreadySubmitted():
externEmpty = False
break
if not externEmpty:
self._nodes.update(self._nodesExtern)
else:
self._nodesExtern.clear()
nodesToProcess, edgesToProcess = graph.dfsToProcess(startNodes=toNodes)
flowEdges = graph.flowEdges(startNodes=toNodes)
edgesToProcess = set(edgesToProcess).intersection(flowEdges)
sub = meshroom.core.submitters.get(submitter, None)
if sub is None:
raise RuntimeError("Unknown Submitter : " + submitter)
try:
res = sub.submit(nodesToProcess, edgesToProcess, graph.filepath)
if res:
for node in nodesToProcess:
node.submit() # update node status
self._nodes.update(nodesToProcess)
self._nodesExtern.update(nodesToProcess)
except Exception as e:
logging.error("Error on submit : {}".format(e))
nodes = Property(BaseObject, lambda self: self._nodes, constant=True)
def getAlreadySubmittedChunks(nodes):