#NEWLINE# def computeOneNode( self , node ) : #NEWLINE# #INDENT# "push 'node' one step further in its computation requirements\n\n        self.intermediates_[node] either contains a list of values to be computed\n        or an execution context\n        " #NEWLINE# if ( self . intermediates_ [ node ] is None ) : #NEWLINE# #INDENT# context = self . grabContext ( ) #NEWLINE# with self . contextEnterer ( context ) : #NEWLINE# #INDENT# context . resetInterruptState ( ) #NEWLINE# if isinstance ( node , tuple ) : #NEWLINE# #INDENT# with freestoreLock : #NEWLINE# #INDENT# context . placeInEvaluationStateWithoutRenamingMutableVectors ( * node ) #NEWLINE# #UNINDENT# context . resume ( ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if isinstance ( node , FORANative . PausedComputation ) : #NEWLINE# #INDENT# context . resumePausedComputation ( node ) #NEWLINE# context . resetInterruptState ( ) #NEWLINE# context . resume ( ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# assert False , ( "don't know what to do with node of type %s" % node ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# self . intermediates_ [ node ] = context #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if isinstance ( self . intermediates_ [ node ] , FORANative . ExecutionContext ) : #NEWLINE# #INDENT# context = self . intermediates_ [ node ] #NEWLINE# req = context . getCacheRequest ( ) #NEWLINE# if CacheSemantics . isVectorCacheLoadRequest ( req ) : #NEWLINE# #INDENT# with self . contextEnterer ( context ) : #NEWLINE# #INDENT# context . resetInterruptState ( ) #NEWLINE# context . resume ( ComputationResult_ . Result ( ImplValContainer_ ( ) , ImplValContainer_ ( ) ) ) #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# if CacheSemantics . isCacheRequestWithResult ( req ) : #NEWLINE# #INDENT# result = CacheSemantics . getCacheRequestComputationResult ( req ) #NEWLINE# with self . contextEnterer ( context ) : #NEWLINE# #INDENT# context . resetInterruptState ( ) #NEWLINE# context . resume ( result ) #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# cacheCalls = [ x . extractApplyTuple ( ) for x in CacheSemantics . processCacheCall ( req ) ] #NEWLINE# res = [ ] #NEWLINE# exception = None #NEWLINE# for t in cacheCalls : #NEWLINE# #INDENT# assert ( t in self . finishedValues_ ) , ( "Couldn't find result for: %s in %s" % ( t , '\n' . join ( [ str ( x ) for x in self . finishedValues_ . keys ( ) ] ) ) ) #NEWLINE# if self . finishedValues_ [ t ] . isException ( ) : #NEWLINE# #INDENT# if ( exception is None ) : #NEWLINE# #INDENT# exception = self . finishedValues_ [ t ] #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# res . append ( self . finishedValues_ [ t ] . asResult . result ) #NEWLINE# #UNINDENT# #UNINDENT# with self . contextEnterer ( context ) : #NEWLINE# #INDENT# if exception : #NEWLINE# #INDENT# context . resetInterruptState ( ) #NEWLINE# context . resume ( exception ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# context . resetInterruptState ( ) #NEWLINE# context . resume ( ComputationResult_ . Result ( ImplValContainer_ ( tuple ( res ) ) , ImplValContainer_ ( ) ) ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# ( splitResult , splitComputationLog ) = self . intermediates_ [ node ] #NEWLINE# for slotComputation in splitResult . submittedComputations ( ) : #NEWLINE# #INDENT# assert ( slotComputation in self . finishedValues_ ) #NEWLINE# value = self . finishedValues_ [ slotComputation ] #NEWLINE# if value . isFailure ( ) : #NEWLINE# #INDENT# self . finishNode_ ( node , value ) #NEWLINE# return #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# splitResult . futuresSplitResult . slotCompleted ( cacheCalls . computationsToSlotIndices [ slotComputation ] , value ) #NEWLINE# del splitResult . computationsToSlotIndices [ slotComputation ] #NEWLINE# #UNINDENT# #UNINDENT# submittableFutures = splitResult . futuresSplitResult . indicesOfSubmittableFutures ( ) #NEWLINE# if ( len ( submittableFutures ) == 0 ) : #NEWLINE# #INDENT# context = self . grabContext ( ) #NEWLINE# toResumeWith = splitResult . futuresSplitResult . getFinalResult ( ) #NEWLINE# context . resumePausedComputation ( toResumeWith ) #NEWLINE# context . resetInterruptState ( ) #NEWLINE# self . intermediates_ [ node ] = context #NEWLINE# with self . contextEnterer ( context ) : #NEWLINE# #INDENT# context . resume ( ) #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# with self . lock_ : #NEWLINE# #INDENT# futuresSplitResult = splitResult . futuresSplitResult #NEWLINE# ( isFinished , result ) = self . findMeatyPausedComputations ( futuresSplitResult ) #NEWLINE# if ( not isFinished ) : #NEWLINE# #INDENT# splitResult = self . computeIntermediatesForSplitResult ( node , futuresSplitResult , result ) #NEWLINE# self . intermediates_ [ node ] = ( splitResult , [ ] ) #NEWLINE# return #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# toResume = result #NEWLINE# context = self . grabContext ( ) #NEWLINE# context . resumePausedComputation ( toResume ) #NEWLINE# context . resetInterruptState ( ) #NEWLINE# self . intermediates_ [ node ] = context #NEWLINE# with self . contextEnterer ( context ) : #NEWLINE# #INDENT# context . resume ( ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# while True : #NEWLINE# #INDENT# if context . isFinished ( ) : #NEWLINE# #INDENT# result = context . getFinishedResult ( ) #NEWLINE# self . checkContextBackIn ( context ) #NEWLINE# self . finishNode_ ( node , result ) #NEWLINE# return #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if context . isVectorLoad ( ) : #NEWLINE# #INDENT# for vectorToLoad in context . getVectorLoadAsVDIDs ( ) : #NEWLINE# #INDENT# toLoad = None #NEWLINE# loaded = False #NEWLINE# if ( self . offlineCache_ is not None ) : #NEWLINE# #INDENT# toLoad = self . offlineCache_ . loadIfExists ( vectorToLoad . page ) #NEWLINE# if ( toLoad is not None ) : #NEWLINE# #INDENT# self . vdm_ . loadSerializedVectorPage ( vectorToLoad . page , toLoad ) #NEWLINE# loaded = True #NEWLINE# #UNINDENT# #UNINDENT# if ( ( not loaded ) and vectorToLoad . isExternal ( ) ) : #NEWLINE# #INDENT# PythonIoTasks . loadExternalDataset ( getCurrentS3Interface ( ) , vectorToLoad , self . vdm_ , self . inProcessDownloader ) #NEWLINE# loaded = True #NEWLINE# #UNINDENT# assert loaded , ( 'lost the definition for VDID: %s' % vectorToLoad ) #NEWLINE# #UNINDENT# with self . contextEnterer ( context ) : #NEWLINE# #INDENT# context . resetInterruptState ( ) #NEWLINE# context . resume ( ) #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# if context . isInterrupted ( ) : #NEWLINE# #INDENT# toResume = None #NEWLINE# if self . checkShouldSplit ( context ) : #NEWLINE# #INDENT# futuresSplitResult = context . splitWithFutures ( ) #NEWLINE# if ( futuresSplitResult is not None ) : #NEWLINE# #INDENT# with self . lock_ : #NEWLINE# #INDENT# futuresSplitResult . disallowRepeatNodes ( ) #NEWLINE# ( isFinished , result ) = self . findMeatyPausedComputations ( futuresSplitResult ) #NEWLINE# if ( not isFinished ) : #NEWLINE# #INDENT# splitResult = self . computeIntermediatesForSplitResult ( node , futuresSplitResult , result ) #NEWLINE# self . intermediates_ [ node ] = ( splitResult , context . getComputationLog ( ) ) #NEWLINE# self . checkContextBackIn ( context ) #NEWLINE# return #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# toResume = result #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# with self . contextEnterer ( context ) : #NEWLINE# #INDENT# if ( toResume is not None ) : #NEWLINE# #INDENT# context . resumePausedComputation ( toResume ) #NEWLINE# #UNINDENT# context . resetInterruptState ( ) #NEWLINE# context . resume ( ) #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# if context . isCacheRequest ( ) : #NEWLINE# #INDENT# req = context . getCacheRequest ( ) #NEWLINE# deps = set ( ) #NEWLINE# if CacheSemantics . isVectorCacheLoadRequest ( req ) : #NEWLINE# #INDENT# pass #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if CacheSemantics . isCacheRequestWithResult ( req ) : #NEWLINE# #INDENT# pass #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# cacheCalls = [ x . extractApplyTuple ( ) for x in CacheSemantics . processCacheCall ( req ) ] #NEWLINE# with self . lock_ : #NEWLINE# #INDENT# for t in cacheCalls : #NEWLINE# #INDENT# if ( ( t not in self . finishedValues_ ) and ( t not in self . intermediates_ ) ) : #NEWLINE# #INDENT# self . intermediates_ [ t ] = None #NEWLINE# self . completable_ . put ( t ) #NEWLINE# self . watchers_ [ t ] = threading . Event ( ) #NEWLINE# #UNINDENT# if ( t not in self . finishedValues_ ) : #NEWLINE# #INDENT# deps . add ( t ) #NEWLINE# #UNINDENT# #UNINDENT# self . dependencies_ [ node ] = deps #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# if ( not deps ) : #NEWLINE# #INDENT# with self . lock_ : #NEWLINE# #INDENT# self . completable_ . put ( node ) #NEWLINE# #UNINDENT# #UNINDENT# return