<s> import gzip <EOL> import Queue <EOL> import logging <EOL> import time <EOL> import threading <EOL> import ufora . FORA . python . FORA as FORA <EOL> import ufora . native . FORA as ForaNative <EOL> import ufora . native . Cumulus as CumulusNative <EOL> import ufora . native . Hash as HashNative <EOL> import ufora . util . Teardown as Teardown <EOL> import ufora . util . CodeCoverage as CodeCoverage <EOL> import cPickle as pickle <EOL> import ufora . native . CallbackScheduler as CallbackScheduler <EOL> callbackScheduler = CallbackScheduler . singletonForTesting ( ) <EOL> IVC = ForaNative . ImplValContainer <EOL> callIVC = ForaNative . makeSymbol ( "<STR_LIT>" ) <EOL> expensiveCalculationText = """<STR_LIT>""" <EOL> def makeComputationDefinitionFromIVCs ( * args ) : <EOL> return CumulusNative . ComputationDefinition . Root ( <EOL> CumulusNative . ImmutableTreeVectorOfComputationDefinitionTerm ( <EOL> [ CumulusNative . ComputationDefinitionTerm . Value ( ForaNative . ImplValContainer ( x ) , None ) <EOL> for x in args ] <EOL> ) <EOL> ) <EOL> def foraBrownian ( x , depth ) : <EOL> brownianIVC = FORA . extractImplValContainer ( FORA . eval ( "<STR_LIT>" ) ) <EOL> return makeComputationDefinitionFromIVCs ( <EOL> ( brownianIVC , callIVC , IVC ( x ) , IVC ( depth ) ) <EOL> ) <EOL> class CumulusTestCases ( object ) : <EOL> def evaluateWithGateway ( self , computationDefinition , timeout = <NUM_LIT> ) : <EOL> computationId = self . gateway . requestComputation ( computationDefinition ) <EOL> try : <EOL> return self . gateway . finalResponses . get ( timeout = <NUM_LIT> ) <EOL> finally : <EOL> self . gateway . deprioritizeComputation ( computationId ) <EOL> @ Teardown . Teardown <EOL> def test_vecWithinVec ( self ) : <EOL> self . desirePublisher . desireNumberOfWorkers ( <NUM_LIT:4> , blocking = True ) <EOL> expr = FORA . extractImplValContainer ( <EOL> FORA . eval ( <EOL> """<STR_LIT>""" <EOL> ) <EOL> ) <EOL> response = self . evaluateWithGateway ( <EOL> makeComputationDefinitionFromIVCs ( <EOL> expr , <EOL> ForaNative . makeSymbol ( "<STR_LIT>" ) <EOL> ) <EOL> ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . isResult ( ) ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . asResult . result . pyval == '<STR_LIT>' , response [ <NUM_LIT:1> ] . asResult . result . pyval ) <EOL> @ Teardown . Teardown <EOL> def test_mixedVecOfNothingAndFloat ( self ) : <EOL> self . desirePublisher . desireNumberOfWorkers ( <NUM_LIT:4> , blocking = True ) <EOL> expr = FORA . extractImplValContainer ( <EOL> FORA . eval ( <EOL> """<STR_LIT>""" <EOL> ) <EOL> ) <EOL> response = self . evaluateWithGateway ( <EOL> makeComputationDefinitionFromIVCs ( <EOL> expr , <EOL> ForaNative . makeSymbol ( "<STR_LIT>" ) <EOL> ) <EOL> ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . isResult ( ) ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . asResult . result . pyval == <NUM_LIT> , response [ <NUM_LIT:1> ] . asResult . result . pyval ) <EOL> @ Teardown . Teardown <EOL> def DISABLEDtest_repeatedCalculationOfSomethingComplex ( self ) : <EOL> self . desirePublisher . desireNumberOfWorkers ( <NUM_LIT:1> , blocking = True ) <EOL> expr = FORA . extractImplValContainer ( FORA . eval ( expensiveCalculationText ) ) <EOL> def calculateTime ( count ) : <EOL> t0 = time . time ( ) <EOL> response = self . evaluateWithGateway ( <EOL> makeComputationDefinitionFromIVCs ( <EOL> expr , <EOL> ForaNative . makeSymbol ( "<STR_LIT>" ) , <EOL> ForaNative . ImplValContainer ( count ) <EOL> ) <EOL> ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . isResult ( ) ) <EOL> stats = response [ <NUM_LIT:2> ] <EOL> return ( stats . timeSpentInInterpreter , stats . timeSpentInCompiler , time . time ( ) - t0 ) <EOL> perSecondValues = [ ] <EOL> for base in [ <NUM_LIT> ] : <EOL> for ix in range ( <NUM_LIT:10> ) : <EOL> ratio = <NUM_LIT:1.0> + ix <EOL> interp , compiled , wallClock = calculateTime ( base * ratio ) <EOL> perSecondValues . append ( ( ( base * ratio ) / wallClock / <NUM_LIT> ) ) <EOL> print base * ratio , "<STR_LIT>" , wallClock , "<STR_LIT>" , "<STR_LIT>" % ( ( base * ratio ) / wallClock / <NUM_LIT> ) , "<STR_LIT>" <EOL> fastest = max ( perSecondValues ) <EOL> saturationPoint = <NUM_LIT:0> <EOL> while saturationPoint < len ( perSecondValues ) and perSecondValues [ saturationPoint ] < fastest / <NUM_LIT:2> : <EOL> saturationPoint += <NUM_LIT:1> <EOL> if saturationPoint < len ( perSecondValues ) : <EOL> slowest = min ( perSecondValues [ saturationPoint : ] ) <EOL> else : <EOL> slowest = <NUM_LIT:0.0> <EOL> self . assertTrue ( slowest > fastest / <NUM_LIT> , "<STR_LIT>" % perSecondValues ) <EOL> @ Teardown . Teardown <EOL> def test_vectorOfPagedVectorApplyWithDropping ( self ) : <EOL> self . desirePublisher . desireNumberOfWorkers ( <NUM_LIT:3> , blocking = True ) <EOL> expr = FORA . extractImplValContainer ( <EOL> FORA . eval ( <EOL> """<STR_LIT>""" <EOL> ) <EOL> ) <EOL> computation1 = self . gateway . requestComputation ( <EOL> makeComputationDefinitionFromIVCs ( <EOL> expr , <EOL> ForaNative . makeSymbol ( "<STR_LIT>" ) <EOL> ) <EOL> ) <EOL> try : <EOL> response = self . gateway . finalResponses . get ( timeout = <NUM_LIT> ) <EOL> except Queue . Empty : <EOL> response = None <EOL> self . assertTrue ( response is not None , response ) <EOL> self . gateway . deprioritizeComputation ( computation1 ) <EOL> self . desirePublisher . desireNumberOfWorkers ( <NUM_LIT:2> , blocking = True ) <EOL> expr2 = FORA . extractImplValContainer ( <EOL> FORA . eval ( <EOL> """<STR_LIT>""" <EOL> ) <EOL> ) <EOL> computation2 = self . gateway . requestComputation ( <EOL> makeComputationDefinitionFromIVCs ( <EOL> expr2 , <EOL> ForaNative . makeSymbol ( "<STR_LIT>" ) <EOL> ) <EOL> ) <EOL> try : <EOL> response = self . gateway . finalResponses . get ( timeout = <NUM_LIT> ) <EOL> except Queue . Empty : <EOL> response = None <EOL> self . assertTrue ( response is not None ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . isResult ( ) ) <EOL> self . gateway . deprioritizeComputation ( computation2 ) <EOL> @ Teardown . Teardown <EOL> def test_sortALargeVectorWithFourWorkers ( self ) : <EOL> self . desirePublisher . desireNumberOfWorkers ( <NUM_LIT:4> , blocking = True ) <EOL> expr = FORA . extractImplValContainer ( <EOL> FORA . eval ( <EOL> """<STR_LIT>""" <EOL> ) <EOL> ) <EOL> try : <EOL> response = self . evaluateWithGateway ( <EOL> makeComputationDefinitionFromIVCs ( <EOL> expr , <EOL> ForaNative . makeSymbol ( "<STR_LIT>" ) <EOL> ) , <EOL> timeout = <NUM_LIT> <EOL> ) <EOL> except Queue . Empty : <EOL> response = None <EOL> if response is None : <EOL> try : <EOL> dumpFun = self . dumpSchedulerEventStreams <EOL> dumpFun ( ) <EOL> except : <EOL> logging . warn ( "<STR_LIT>" ) ; <EOL> self . assertTrue ( response is not None ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . isResult ( ) ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . asResult . result . pyval == '<STR_LIT>' , response [ <NUM_LIT:1> ] . asResult . result . pyval ) <EOL> @ Teardown . Teardown <EOL> def test_bigSumWithAdding ( self ) : <EOL> self . desirePublisher . desireNumberOfWorkers ( <NUM_LIT:1> , blocking = True ) <EOL> expr = FORA . extractImplValContainer ( <EOL> FORA . eval ( <EOL> """<STR_LIT>""" <EOL> ) <EOL> ) <EOL> computationId = self . gateway . requestComputation ( <EOL> makeComputationDefinitionFromIVCs ( <EOL> expr , <EOL> ForaNative . makeSymbol ( "<STR_LIT>" ) <EOL> ) <EOL> ) <EOL> time . sleep ( <NUM_LIT:5> ) <EOL> self . desirePublisher . desireNumberOfWorkers ( <NUM_LIT:2> , blocking = True ) <EOL> try : <EOL> response = self . gateway . finalResponses . get ( timeout = <NUM_LIT> ) <EOL> except Queue . Empty : <EOL> response = None <EOL> if response is None : <EOL> try : <EOL> dumpFun = self . dumpSchedulerEventStreams <EOL> dumpFun ( ) <EOL> except : <EOL> logging . warn ( "<STR_LIT>" ) ; <EOL> self . assertTrue ( response is not None ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . isResult ( ) ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . asResult . result . pyval == '<STR_LIT>' , response [ <NUM_LIT:1> ] . asResult . result . pyval ) <EOL> self . gateway . deprioritizeComputation ( computationId ) <EOL> @ Teardown . Teardown <EOL> def test_manyDuplicateCachecallsAndAdding ( self ) : <EOL> self . desirePublisher . desireNumberOfWorkers ( <NUM_LIT:2> , blocking = True ) <EOL> expr = FORA . extractImplValContainer ( <EOL> FORA . eval ( <EOL> """<STR_LIT>""" <EOL> ) <EOL> ) <EOL> computationId = self . gateway . requestComputation ( <EOL> makeComputationDefinitionFromIVCs ( <EOL> expr , <EOL> ForaNative . makeSymbol ( "<STR_LIT>" ) <EOL> ) <EOL> ) <EOL> time . sleep ( <NUM_LIT:5> ) <EOL> self . desirePublisher . desireNumberOfWorkers ( <NUM_LIT:4> , blocking = True ) <EOL> try : <EOL> response = self . gateway . finalResponses . get ( timeout = <NUM_LIT> ) <EOL> except Queue . Empty : <EOL> response = None <EOL> self . assertTrue ( response is not None ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . isResult ( ) ) <EOL> self . gateway . deprioritizeComputation ( computationId ) <EOL> @ Teardown . Teardown <EOL> def test_sortALargeVectorWithAdding ( self ) : <EOL> self . desirePublisher . desireNumberOfWorkers ( <NUM_LIT:2> , blocking = True ) <EOL> expr = FORA . extractImplValContainer ( <EOL> FORA . eval ( <EOL> """<STR_LIT>""" <EOL> ) <EOL> ) <EOL> computationId = self . gateway . requestComputation ( <EOL> makeComputationDefinitionFromIVCs ( <EOL> expr , <EOL> ForaNative . makeSymbol ( "<STR_LIT>" ) <EOL> ) <EOL> ) <EOL> time . sleep ( <NUM_LIT:10> ) <EOL> self . desirePublisher . desireNumberOfWorkers ( <NUM_LIT:4> , blocking = True ) <EOL> try : <EOL> response = self . gateway . finalResponses . get ( timeout = <NUM_LIT> ) <EOL> except Queue . Empty : <EOL> response = None <EOL> self . gateway . deprioritizeComputation ( computationId ) <EOL> if response is None : <EOL> try : <EOL> dumpFun = self . dumpSchedulerEventStreams <EOL> dumpFun ( ) <EOL> except : <EOL> logging . warn ( "<STR_LIT>" ) ; <EOL> self . assertTrue ( response is not None ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . isResult ( ) ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . asResult . result . pyval == '<STR_LIT>' , response [ <NUM_LIT:1> ] . asResult . result . pyval ) <EOL> @ Teardown . Teardown <EOL> def test_sortALargeVectorWithRemoving ( self ) : <EOL> self . desirePublisher . desireNumberOfWorkers ( <NUM_LIT:4> , blocking = True ) <EOL> expr = FORA . extractImplValContainer ( <EOL> FORA . eval ( <EOL> """<STR_LIT>""" <EOL> ) <EOL> ) <EOL> computationId = self . gateway . requestComputation ( <EOL> makeComputationDefinitionFromIVCs ( <EOL> expr , <EOL> ForaNative . makeSymbol ( "<STR_LIT>" ) <EOL> ) <EOL> ) <EOL> time . sleep ( <NUM_LIT:10> ) <EOL> self . desirePublisher . desireNumberOfWorkers ( <NUM_LIT:3> , blocking = True ) <EOL> try : <EOL> response = self . gateway . finalResponses . get ( timeout = <NUM_LIT> ) <EOL> except Queue . Empty : <EOL> response = None <EOL> self . gateway . deprioritizeComputation ( computationId ) <EOL> if response is None : <EOL> try : <EOL> dumpFun = self . dumpSchedulerEventStreams <EOL> dumpFun ( ) <EOL> except : <EOL> logging . warn ( "<STR_LIT>" ) ; <EOL> self . assertTrue ( response is not None ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . isResult ( ) ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . asResult . result . pyval == '<STR_LIT>' , response [ <NUM_LIT:1> ] . asResult . result . pyval ) <EOL> @ Teardown . Teardown <EOL> def test_vectorApplyWithAdding ( self ) : <EOL> self . desirePublisher . desireNumberOfWorkers ( <NUM_LIT:2> , blocking = True ) <EOL> expr = FORA . extractImplValContainer ( <EOL> FORA . eval ( <EOL> """<STR_LIT>""" <EOL> ) <EOL> ) <EOL> computationId = self . gateway . requestComputation ( <EOL> makeComputationDefinitionFromIVCs ( <EOL> expr , <EOL> ForaNative . makeSymbol ( "<STR_LIT>" ) <EOL> ) <EOL> ) <EOL> try : <EOL> response = self . gateway . finalResponses . get ( timeout = <NUM_LIT> ) <EOL> except Queue . Empty : <EOL> response = None <EOL> self . assertTrue ( response is None , response ) <EOL> self . desirePublisher . desireNumberOfWorkers ( <NUM_LIT:3> , blocking = True ) <EOL> response = self . gateway . finalResponses . get ( timeout = <NUM_LIT> ) <EOL> self . gateway . deprioritizeComputation ( computationId ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . isResult ( ) ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . asResult . result . pyval == '<STR_LIT>' , response [ <NUM_LIT:1> ] . asResult . result . pyval ) <EOL> @ Teardown . Teardown <EOL> def test_expensive_brownian ( self ) : <EOL> self . desirePublisher . desireNumberOfWorkers ( <NUM_LIT:2> , blocking = True ) <EOL> expr = FORA . extractImplValContainer ( <EOL> FORA . eval ( <EOL> """<STR_LIT>""" <EOL> ) <EOL> ) <EOL> response = self . evaluateWithGateway ( <EOL> makeComputationDefinitionFromIVCs ( <EOL> expr , <EOL> ForaNative . makeSymbol ( "<STR_LIT>" ) <EOL> ) <EOL> ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . isResult ( ) ) <EOL> @ Teardown . Teardown <EOL> def test_simple_computation ( self ) : <EOL> self . desirePublisher . desireNumberOfWorkers ( <NUM_LIT:2> , blocking = True ) <EOL> simpleFunc = FORA . extractImplValContainer ( FORA . eval ( "<STR_LIT>" ) ) <EOL> response = self . evaluateWithGateway ( <EOL> makeComputationDefinitionFromIVCs ( <EOL> simpleFunc , <EOL> ForaNative . makeSymbol ( "<STR_LIT>" ) <EOL> ) <EOL> ) <EOL> self . assertEqual ( <NUM_LIT:3> , response [ <NUM_LIT:1> ] . asResult . result . pyval ) <EOL> @ Teardown . Teardown <EOL> def test_sortManySmallVectors ( self ) : <EOL> self . desirePublisher . desireNumberOfWorkers ( <NUM_LIT:4> , blocking = True ) <EOL> expr = FORA . extractImplValContainer ( <EOL> FORA . eval ( <EOL> """<STR_LIT>""" <EOL> ) <EOL> ) <EOL> response = self . evaluateWithGateway ( <EOL> makeComputationDefinitionFromIVCs ( <EOL> expr , <EOL> ForaNative . makeSymbol ( "<STR_LIT>" ) <EOL> ) <EOL> ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . isResult ( ) ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . asResult . result . pyval == True , response [ <NUM_LIT:1> ] . asResult . result . pyval ) <EOL> @ Teardown . Teardown <EOL> def test_basic_sum_1_worker ( self ) : <EOL> self . desirePublisher . desireNumberOfWorkers ( <NUM_LIT:1> , blocking = True ) <EOL> expr = FORA . extractImplValContainer ( <EOL> FORA . eval ( <EOL> """<STR_LIT>""" <EOL> ) <EOL> ) <EOL> response = self . evaluateWithGateway ( <EOL> makeComputationDefinitionFromIVCs ( <EOL> expr , <EOL> ForaNative . makeSymbol ( "<STR_LIT>" ) <EOL> ) <EOL> ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . isResult ( ) ) <EOL> @ Teardown . Teardown <EOL> def test_basic_sum_2_workers ( self ) : <EOL> self . desirePublisher . desireNumberOfWorkers ( <NUM_LIT:2> , blocking = True ) <EOL> expr = FORA . extractImplValContainer ( <EOL> FORA . eval ( <EOL> """<STR_LIT>""" <EOL> ) <EOL> ) <EOL> response = self . evaluateWithGateway ( <EOL> makeComputationDefinitionFromIVCs ( <EOL> expr , <EOL> ForaNative . makeSymbol ( "<STR_LIT>" ) <EOL> ) <EOL> ) <EOL> self . assertTrue ( response [ <NUM_LIT:1> ] . isResult ( ) ) </s>
