<s> from time import time <EOL> from collections import defaultdict <EOL> from weakref import WeakKeyDictionary <EOL> from octopus . dispatcher . model . enums import * <EOL> from octopus . dispatcher . model import Task <EOL> from . import models <EOL> import logging <EOL> LOGGER = logging . getLogger ( "<STR_LIT>" ) <EOL> class NoRenderNodeAvailable ( BaseException ) : <EOL> '''<STR_LIT>''' <EOL> class NoLicenseAvailableForTask ( BaseException ) : <EOL> '''<STR_LIT>''' <EOL> class DependencyListField ( models . Field ) : <EOL> def to_json ( self , node ) : <EOL> return [ [ dep . id , statusList ] for ( dep , statusList ) in node . dependencies ] <EOL> class PoolShareDictField ( models . Field ) : <EOL> def to_json ( self , instance ) : <EOL> return [ [ poolShare . id , poolShare . pool . name ] for poolShare in instance . poolShares . values ( ) ] <EOL> class AdditionnalPoolShareDictField ( models . Field ) : <EOL> def to_json ( self , instance ) : <EOL> return [ [ ps . id , ps . pool . name ] for ps in instance . additionnalPoolShares . values ( ) ] <EOL> class FolderNodeChildrenField ( models . Field ) : <EOL> def to_json ( self , instance ) : <EOL> return [ child . id for child in instance . children ] <EOL> class BaseNode ( models . Model ) : <EOL> dispatcher = None <EOL> name = models . StringField ( ) <EOL> parent = models . ModelField ( allow_null = True ) <EOL> user = models . StringField ( ) <EOL> priority = models . IntegerField ( ) <EOL> dispatchKey = models . FloatField ( ) <EOL> maxRN = models . IntegerField ( ) <EOL> updateTime = models . FloatField ( ) <EOL> poolShares = PoolShareDictField ( ) <EOL> additionnalPoolShares = AdditionnalPoolShareDictField ( ) <EOL> completion = models . FloatField ( ) <EOL> status = models . IntegerField ( ) <EOL> creationTime = models . FloatField ( ) <EOL> startTime = models . FloatField ( allow_null = True ) <EOL> updateTime = models . FloatField ( allow_null = True ) <EOL> endTime = models . FloatField ( allow_null = True ) <EOL> dependencies = DependencyListField ( ) <EOL> averageTimeByFrame = models . FloatField ( allow_null = True ) <EOL> minTimeByFrame = models . FloatField ( allow_null = True ) <EOL> maxTimeByFrame = models . FloatField ( allow_null = True ) <EOL> timer = models . FloatField ( allow_null = True ) <EOL> @ property <EOL> def tags ( self ) : <EOL> return { } <EOL> def __init__ ( self , id , name , parent , user , priority , dispatchKey , maxRN , <EOL> creationTime = None , startTime = None , <EOL> updateTime = None , endTime = None , <EOL> status = NODE_READY ) : <EOL> '''<STR_LIT>''' <EOL> if not self . dispatcher : <EOL> from octopus . dispatcher . dispatcher import Dispatcher <EOL> self . dispatcher = Dispatcher ( None ) <EOL> self . __dict__ [ '<STR_LIT>' ] = None <EOL> models . Model . __init__ ( self ) <EOL> self . id = int ( id ) if id is not None else None <EOL> self . name = str ( name ) <EOL> self . parent = parent <EOL> self . user = str ( user ) <EOL> self . priority = int ( priority ) <EOL> self . dispatchKey = int ( dispatchKey ) <EOL> self . maxRN = int ( maxRN ) <EOL> self . optimalMaxRN = <NUM_LIT:0> <EOL> self . allocatedRN = <NUM_LIT:0> <EOL> self . poolShares = WeakKeyDictionary ( ) <EOL> self . additionnalPoolShares = WeakKeyDictionary ( ) <EOL> self . completion = <NUM_LIT:1.0> <EOL> self . status = status <EOL> self . creationTime = time ( ) if not creationTime else creationTime <EOL> self . startTime = startTime <EOL> self . updateTime = updateTime <EOL> self . endTime = endTime <EOL> self . dependencies = [ ] <EOL> self . reverseDependencies = [ ] <EOL> self . lastDependenciesSatisfaction = False <EOL> self . lastDependenciesSatisfactionDispatchCycle = - <NUM_LIT:1> <EOL> self . readyCommandCount = <NUM_LIT:0> <EOL> self . doneCommandCount = <NUM_LIT:0> <EOL> self . commandCount = <NUM_LIT:0> <EOL> self . averageTimeByFrameList = [ ] <EOL> self . averageTimeByFrame = <NUM_LIT:0.0> <EOL> self . minTimeByFrame = <NUM_LIT:0.0> <EOL> self . maxTimeByFrame = <NUM_LIT:0.0> <EOL> self . timer = None <EOL> def mainPoolShare ( self ) : <EOL> return self . poolShares . values ( ) [ <NUM_LIT:0> ] <EOL> def mainPool ( self ) : <EOL> return self . poolShares . keys ( ) [ <NUM_LIT:0> ] <EOL> def to_json ( self ) : <EOL> base = super ( BaseNode , self ) . to_json ( ) <EOL> base [ "<STR_LIT>" ] = self . allocatedRN <EOL> base [ "<STR_LIT>" ] = self . optimalMaxRN <EOL> base [ "<STR_LIT>" ] = self . tags . copy ( ) <EOL> base [ "<STR_LIT>" ] = self . readyCommandCount <EOL> base [ "<STR_LIT>" ] = self . doneCommandCount <EOL> base [ "<STR_LIT>" ] = self . commandCount <EOL> return base <EOL> def addDependency ( self , node , acceptedStatus ) : <EOL> if not acceptedStatus : <EOL> return <EOL> if self is node : <EOL> return <EOL> self . status = NODE_BLOCKED <EOL> val = [ node , acceptedStatus ] <EOL> if not val in self . dependencies : <EOL> self . dependencies . append ( val ) <EOL> if self not in node . reverseDependencies : <EOL> node . reverseDependencies . append ( self ) <EOL> def checkDependenciesSatisfaction ( self ) : <EOL> if self . dispatcher . cycle == self . lastDependenciesSatisfactionDispatchCycle : <EOL> return self . lastDependenciesSatisfaction <EOL> self . lastDependenciesSatisfaction = True <EOL> self . lastDependenciesSatisfactionDispatchCycle = self . dispatcher . cycle <EOL> for node , acceptedStatus in self . dependencies : <EOL> if node . status not in acceptedStatus : <EOL> self . lastDependenciesSatisfaction = False <EOL> break <EOL> else : <EOL> if self . parent is not None : <EOL> self . lastDependenciesSatisfaction = self . parent . checkDependenciesSatisfaction ( ) <EOL> return self . lastDependenciesSatisfaction <EOL> def __new__ ( cls , * args , ** kwargs ) : <EOL> obj = super ( BaseNode , cls ) . __new__ ( cls ) <EOL> obj . _parent_value = None <EOL> obj . invalidated = True <EOL> return obj <EOL> def __setattr__ ( self , name , value ) : <EOL> if name == '<STR_LIT>' : <EOL> self . setParentValue ( value ) <EOL> super ( BaseNode , self ) . __setattr__ ( name , value ) <EOL> def setParentValue ( self , parent ) : <EOL> if self . parent is parent : <EOL> return <EOL> if self . parent : <EOL> self . parent . removeChild ( self , False ) <EOL> if parent : <EOL> parent . addChild ( self , False ) <EOL> self . __dict__ [ '<STR_LIT>' ] = parent <EOL> def dispatchIterator ( self ) : <EOL> raise NotImplementedError <EOL> def updateAllocation ( self ) : <EOL> '''<STR_LIT>''' <EOL> self . allocatedRN = <NUM_LIT:0> <EOL> for currPoolShare in self . poolShares . values ( ) : <EOL> self . maxRN = currPoolShare . maxRN <EOL> self . allocatedRN += currPoolShare . allocatedRN <EOL> for ps in self . additionnalPoolShares . values ( ) : <EOL> self . allocatedRN += ps . allocatedRN <EOL> def updateCompletionAndStatus ( self ) : <EOL> raise NotImplementedError <EOL> def __repr__ ( self ) : <EOL> nodes = [ self ] <EOL> parent = self . parent <EOL> while parent is not None : <EOL> nodes . insert ( <NUM_LIT:0> , parent ) <EOL> parent = parent . parent <EOL> names = [ node . name for node in nodes ] <EOL> return "<STR_LIT>" % ( self . name , "<STR_LIT:/>" . join ( names ) ) <EOL> def __str__ ( self ) : <EOL> return "<STR_LIT>" % ( self . name , self . maxRN , self . allocatedRN ) <EOL> parent_value = property ( lambda self : self . _parent_value , setParentValue ) <EOL> def invalidate ( self ) : <EOL> self . invalidated = True <EOL> while self . parent and not self . parent . invalidated : <EOL> self . parent . invalidated = True <EOL> self = self . parent <EOL> class FolderNode ( BaseNode ) : <EOL> strategy = models . StrategyField ( ) <EOL> taskGroup = models . ModelField ( allow_null = True ) <EOL> children = FolderNodeChildrenField ( ) <EOL> @ property <EOL> def tags ( self ) : <EOL> return self . taskGroup . tags if self . taskGroup else { } <EOL> def __init__ ( self , id , name , parent , user , priority , dispatchKey , maxRN , strategy , creationTime = None , startTime = None , updateTime = None , endTime = None , status = NODE_DONE , taskGroup = None ) : <EOL> BaseNode . __init__ ( self , id , name , parent , user , priority , dispatchKey , maxRN , creationTime , startTime , updateTime , endTime , status ) <EOL> self . children = [ ] <EOL> self . strategy = strategy <EOL> self . taskGroup = taskGroup <EOL> if taskGroup is not None : <EOL> self . timer = taskGroup . timer <EOL> def addChild ( self , child , setParent = True ) : <EOL> if child . parent is not self and setParent : <EOL> child . parent = self <EOL> else : <EOL> self . children . append ( child ) <EOL> self . fireChildAddedEvent ( child ) <EOL> def removeChild ( self , child , setParent = True ) : <EOL> if child . parent is self and setParent : <EOL> child . parent = None <EOL> else : <EOL> self . children . remove ( child ) <EOL> self . fireChildRemovedEvent ( child ) <EOL> def fireChildAddedEvent ( self , child ) : <EOL> self . invalidate ( ) <EOL> for l in self . changeListeners : <EOL> try : <EOL> l . onChildAddedEvent ( self , child ) <EOL> except AttributeError : <EOL> pass <EOL> def fireChildRemovedEvent ( self , child ) : <EOL> self . invalidate ( ) <EOL> for l in self . changeListeners : <EOL> try : <EOL> l . onChildRemovedEvent ( self , child ) <EOL> except AttributeError : <EOL> pass <EOL> def cmdIterator ( self ) : <EOL> for child in self . children : <EOL> for command in child . cmdIterator ( ) : <EOL> yield command <EOL> def dispatchIterator ( self , stopFunc , ep = None ) : <EOL> if ep is None : <EOL> ep = self <EOL> if self . readyCommandCount == <NUM_LIT:0> : <EOL> return <EOL> self . strategy . update ( self , ep ) <EOL> for child in self . children : <EOL> try : <EOL> for assignment in child . dispatchIterator ( stopFunc , ep ) : <EOL> node , command = assignment <EOL> self . strategy . on_assignment ( self , child , node ) <EOL> yield assignment <EOL> except NoRenderNodeAvailable : <EOL> return <EOL> except NoLicenseAvailableForTask : <EOL> LOGGER . info ( "<STR_LIT>" % self . name ) <EOL> continue <EOL> if stopFunc ( ) : <EOL> return <EOL> def updateCompletionAndStatus ( self ) : <EOL> """<STR_LIT>""" <EOL> self . updateAllocation ( ) <EOL> if not self . invalidated : <EOL> return <EOL> if not self . children : <EOL> completion = <NUM_LIT:1.0> <EOL> status = NODE_DONE <EOL> else : <EOL> self . readyCommandCount = <NUM_LIT:0> <EOL> self . doneCommandCount = <NUM_LIT:0> <EOL> self . commandCount = <NUM_LIT:0> <EOL> completion = <NUM_LIT:0.0> <EOL> status = defaultdict ( int ) <EOL> for child in self . children : <EOL> child . updateCompletionAndStatus ( ) <EOL> completion += child . completion <EOL> status [ child . status ] += <NUM_LIT:1> <EOL> self . readyCommandCount += child . readyCommandCount <EOL> self . doneCommandCount += child . doneCommandCount <EOL> self . commandCount += child . commandCount <EOL> if hasattr ( self , "<STR_LIT>" ) and int ( self . commandCount ) != <NUM_LIT:0> : <EOL> self . completion = self . doneCommandCount / float ( self . commandCount ) <EOL> else : <EOL> self . completion = completion / len ( self . children ) <EOL> if NODE_PAUSED in status : <EOL> self . status = NODE_PAUSED <EOL> elif NODE_ERROR in status : <EOL> self . status = NODE_ERROR <EOL> elif NODE_RUNNING in status : <EOL> self . status = NODE_RUNNING <EOL> elif NODE_READY in status : <EOL> self . status = NODE_READY <EOL> elif NODE_BLOCKED in status : <EOL> self . status = NODE_BLOCKED <EOL> elif NODE_CANCELED in status : <EOL> self . status = NODE_CANCELED <EOL> else : <EOL> self . completion = <NUM_LIT:1.0> <EOL> self . status = NODE_DONE <EOL> times = [ childNode . creationTime for childNode in self . children if childNode . creationTime is not None ] <EOL> if times : <EOL> self . creationTime = min ( times ) <EOL> if self . taskGroup and ( self . taskGroup . creationTime is None or self . taskGroup . creationTime > self . creationTime ) : <EOL> self . taskGroup . creationTime = self . creationTime <EOL> times = [ childNode . startTime for childNode in self . children if childNode . startTime is not None ] <EOL> if times : <EOL> self . startTime = min ( times ) <EOL> if self . taskGroup and ( self . taskGroup . startTime is None or self . taskGroup . startTime > self . startTime ) : <EOL> self . taskGroup . startTime = self . startTime <EOL> times = [ childNode . updateTime for childNode in self . children if childNode . updateTime is not None ] <EOL> if times : <EOL> self . updateTime = max ( times ) <EOL> if self . taskGroup and ( self . taskGroup . updateTime is None or self . taskGroup . updateTime > self . updateTime ) : <EOL> self . taskGroup . updateTime = self . updateTime <EOL> if isFinalNodeStatus ( self . status ) : <EOL> times = [ childNode . endTime for childNode in self . children if childNode . endTime is not None ] <EOL> if times : <EOL> self . endTime = max ( times ) <EOL> if self . taskGroup and ( self . taskGroup . endTime is None or <EOL> self . taskGroup . endTime > self . taskGroup . endTime ) : <EOL> self . taskGroup . endTime = self . endTime <EOL> else : <EOL> self . endTime = None <EOL> if self . taskGroup : <EOL> self . taskGroup . endTime = None <EOL> self . invalidated = False <EOL> if self . taskGroup : <EOL> self . timer = self . taskGroup . timer <EOL> self . taskGroup . updateStatusAndCompletion ( ) <EOL> def setPaused ( self , paused ) : <EOL> for child in self . children : <EOL> child . setPaused ( paused ) <EOL> def resetCompletion ( self ) : <EOL> self . completion = <NUM_LIT:0> <EOL> for child in self . children : <EOL> child . resetCompletion ( ) <EOL> def setStatus ( self , pStatus , pCascadeUpdate = False ) : <EOL> '''<STR_LIT>''' <EOL> if pCascadeUpdate : <EOL> for dependingNode in self . reverseDependencies : <EOL> dependingNode . setStatus ( pStatus , pCascadeUpdate ) <EOL> for child in self . children : <EOL> child . setStatus ( pStatus , pCascadeUpdate ) <EOL> self . status = pStatus <EOL> return True <EOL> def setMaxAttempt ( self , maxAttempt ) : <EOL> '''<STR_LIT:U+0020>''' <EOL> globalResult = True <EOL> for child in self . children : <EOL> res = child . setMaxAttempt ( maxAttempt ) <EOL> if res is False : <EOL> globalResult = False <EOL> self . dispatcher . dispatchTree . toModifyElements . append ( self ) <EOL> return globalResult <EOL> class TaskNode ( BaseNode ) : <EOL> task = models . ModelField ( ) <EOL> paused = models . BooleanField ( ) <EOL> maxAttempt = models . IntegerField ( ) <EOL> @ property <EOL> def tags ( self ) : <EOL> if self . task is not None : <EOL> return self . task . tags <EOL> return None <EOL> def __init__ ( self , id , name , parent , user , priority , dispatchKey , maxRN , task , creationTime = None , startTime = None , updateTime = None , endTime = None , status = NODE_BLOCKED , paused = False , maxAttempt = <NUM_LIT:1> ) : <EOL> '''<STR_LIT>''' <EOL> BaseNode . __init__ ( self , id , name , parent , user , priority , dispatchKey , maxRN , creationTime , startTime , updateTime , endTime , status ) <EOL> self . task = task <EOL> self . paused = paused <EOL> self . maxAttempt = int ( maxAttempt ) <EOL> self . commmandCount = <NUM_LIT:0> <EOL> if task is not None : <EOL> self . timer = task . timer <EOL> self . maxAttempt = int ( task . maxAttempt ) <EOL> self . commandCount = len ( task . commands ) <EOL> def cmdIterator ( self ) : <EOL> for command in self . task . commands : <EOL> yield command <EOL> def dispatchIterator ( self , stopFunc , ep = None ) : <EOL> if ep is None : <EOL> ep = self <EOL> if self . readyCommandCount == <NUM_LIT:0> : <EOL> return <EOL> if self . paused : <EOL> return <EOL> sorted ( self . task . commands , key = lambda x : x . id ) <EOL> for command in self . task . commands : <EOL> if command . status != CMD_READY : <EOL> continue <EOL> renderNode = self . reserve_rendernode ( command , ep ) <EOL> if not renderNode : <EOL> return <EOL> self . readyCommandCount -= <NUM_LIT:1> <EOL> tmp_ep = ep <EOL> while tmp_ep : <EOL> tmp_ep . readyCommandCount -= <NUM_LIT:1> <EOL> tmp_ep = tmp_ep . parent <EOL> yield ( renderNode , command ) <EOL> def reserve_rendernode ( self , command , ep ) : <EOL> '''<STR_LIT>''' <EOL> if ep is None : <EOL> ep = self <EOL> for poolshare in [ poolShare for poolShare in ep . poolShares . values ( ) if poolShare . hasRenderNodesAvailable ( ) ] : <EOL> rnList = sorted ( poolshare . pool . renderNodes , key = lambda rn : rn . performance , reverse = True ) <EOL> for rendernode in rnList : <EOL> if rendernode . isAvailable ( ) and rendernode . canRun ( command ) : <EOL> if rendernode . reserveLicense ( command , self . dispatcher . licenseManager ) : <EOL> rendernode . addAssignment ( command ) <EOL> return rendernode <EOL> else : <EOL> raise NoLicenseAvailableForTask <EOL> if not [ poolShare for poolShare in ep . poolShares . values ( ) if poolShare . hasRenderNodesAvailable ( ) ] : <EOL> raise NoRenderNodeAvailable <EOL> return None <EOL> def updateCompletionAndStatus ( self ) : <EOL> '''<STR_LIT>''' <EOL> self . updateAllocation ( ) <EOL> if not self . invalidated : <EOL> return <EOL> if self . task is None : <EOL> self . status = NODE_CANCELED <EOL> return <EOL> completion = <NUM_LIT:0.0> <EOL> status = defaultdict ( int ) <EOL> self . readyCommandCount = <NUM_LIT:0> <EOL> self . doneCommandCount = <NUM_LIT:0> <EOL> self . commandCount = len ( self . task . commands ) <EOL> for command in self . task . commands : <EOL> completion += command . completion <EOL> status [ command . status ] += <NUM_LIT:1> <EOL> if command . status == CMD_READY : <EOL> self . readyCommandCount += <NUM_LIT:1> <EOL> if command . status == CMD_DONE : <EOL> self . doneCommandCount += <NUM_LIT:1> <EOL> if self . task . commands : <EOL> self . completion = completion / len ( self . task . commands ) <EOL> else : <EOL> self . completion = <NUM_LIT:1.0> <EOL> if CMD_CANCELED in status : <EOL> self . status = NODE_CANCELED <EOL> elif self . paused : <EOL> self . status = NODE_PAUSED <EOL> elif CMD_ERROR in status : <EOL> self . status = NODE_ERROR <EOL> elif CMD_TIMEOUT in status : <EOL> self . status = NODE_ERROR <EOL> elif CMD_RUNNING in status : <EOL> self . status = NODE_RUNNING <EOL> elif CMD_ASSIGNED in status : <EOL> self . status = NODE_READY <EOL> elif CMD_FINISHING in status : <EOL> self . status = NODE_RUNNING <EOL> elif CMD_READY in status : <EOL> self . status = NODE_READY <EOL> elif CMD_BLOCKED in status : <EOL> self . status = NODE_BLOCKED <EOL> else : <EOL> self . completion = <NUM_LIT:1.0> <EOL> self . status = NODE_DONE <EOL> times = [ command . creationTime for command in self . task . commands if command . creationTime is not None ] <EOL> if times : <EOL> self . creationTime = min ( times ) <EOL> times = [ command . startTime for command in self . task . commands if command . startTime is not None ] <EOL> if times : <EOL> self . startTime = min ( times ) <EOL> times = [ command . updateTime for command in self . task . commands if command . updateTime is not None ] <EOL> if times : <EOL> self . updateTime = max ( times ) <EOL> if self . status == NODE_DONE : <EOL> times = [ command . endTime for command in self . task . commands if command . endTime is not None ] <EOL> if times : <EOL> self . endTime = max ( times ) <EOL> else : <EOL> self . endTime = None <EOL> self . task . status = self . status <EOL> self . task . completion = self . completion <EOL> self . task . creationTime = self . creationTime <EOL> self . task . startTime = self . startTime <EOL> self . task . updateTime = self . updateTime <EOL> self . task . endTime = self . endTime <EOL> self . timer = self . task . timer <EOL> self . invalidated = False <EOL> def checkDependenciesSatisfaction ( self ) : <EOL> taskNodes = [ taskNode <EOL> for taskNode in self . dispatcher . dispatchTree . nodes . values ( ) <EOL> if isinstance ( taskNode , TaskNode ) and taskNode . task == self . task ] <EOL> return all ( BaseNode . checkDependenciesSatisfaction ( taskNode ) for taskNode in taskNodes ) <EOL> def setPaused ( self , paused ) : <EOL> if self . status != NODE_DONE : <EOL> self . paused = paused <EOL> if self . status == NODE_PAUSED and not paused : <EOL> self . status = NODE_READY <EOL> self . invalidate ( ) <EOL> def setMaxAttempt ( self , maxAttempt ) : <EOL> '''<STR_LIT:U+0020>''' <EOL> if not isinstance ( self . task , Task ) : <EOL> return False <EOL> self . task . maxAttempt = maxAttempt <EOL> self . maxAttempt = maxAttempt <EOL> self . dispatcher . dispatchTree . toModifyElements . append ( self . task ) <EOL> self . dispatcher . dispatchTree . toModifyElements . append ( self ) <EOL> return True <EOL> def resetCompletion ( self ) : <EOL> self . completion = <NUM_LIT:0> <EOL> for command in self . task . commands : <EOL> command . completion = <NUM_LIT:0> <EOL> def setStatus ( self , pStatus , pCascadeUpdate = False ) : <EOL> '''<STR_LIT>''' <EOL> if pCascadeUpdate : <EOL> for dependingNode in self . reverseDependencies : <EOL> dependingNode . setStatus ( pStatus , pCascadeUpdate ) <EOL> if pStatus == NODE_CANCELED and self . status != NODE_DONE : <EOL> for command in self . task . commands : <EOL> command . cancel ( ) <EOL> elif pStatus == NODE_READY and self . status != NODE_RUNNING : <EOL> if any ( isRunningStatus ( command . status ) for command in self . task . commands ) : <EOL> return False <EOL> for command in self . task . commands : <EOL> command . setReadyStatus ( ) <EOL> elif pStatus in ( NODE_DONE , NODE_ERROR , NODE_BLOCKED , NODE_RUNNING ) : <EOL> return False <EOL> return True </s>
