<s> """<STR_LIT>""" <EOL> import time <EOL> import unittest <EOL> import threading <EOL> import ufora . distributed . SharedState . ComputedGraph . SharedStateSynchronizer as SharedStateSynchronizer <EOL> from ufora . distributed . SharedState . tests . SharedStateTestHarness import SharedStateTestHarness <EOL> import ufora . BackendGateway . ComputedGraph . ComputedGraph as ComputedGraph <EOL> import ufora . BackendGateway . ComputedGraph . ComputedGraphTestHarness as ComputedGraphTestHarness <EOL> import ufora . native . Cumulus as CumulusNative <EOL> import ufora . native . Hash as HashNative <EOL> import ufora . native . TCMalloc as TCMallocNative <EOL> HashSet = HashNative . ImmutableTreeSetOfHash <EOL> sha1 = HashNative . Hash . sha1 <EOL> import ufora . native . CallbackScheduler as CallbackScheduler <EOL> callbackScheduler = CallbackScheduler . singletonForTesting ( ) <EOL> class TestPersistentCacheIndex ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . sharedState = SharedStateTestHarness ( inMemory = True ) <EOL> self . synchronizer = SharedStateSynchronizer . SharedStateSynchronizer ( ) <EOL> self . synchronizer . attachView ( self . sharedState . newView ( ) ) <EOL> self . synchronizer . __enter__ ( ) <EOL> def tearDown ( self ) : <EOL> self . synchronizer . __exit__ ( None , None , None ) <EOL> def waitForSync ( self , predicate , * synchronizers ) : <EOL> if not synchronizers : <EOL> synchronizers = [ self . synchronizer ] <EOL> ComputedGraph . assertHasGraph ( ) <EOL> passes = <NUM_LIT:0> <EOL> while passes < <NUM_LIT:100> : <EOL> for s in synchronizers : <EOL> s . update ( ) <EOL> if predicate ( ) : <EOL> break <EOL> time . sleep ( <NUM_LIT> ) <EOL> passes += <NUM_LIT:1> <EOL> self . assertTrue ( predicate ( ) ) <EOL> @ ComputedGraphTestHarness . UnderHarness <EOL> def test_persistentCacheUnderLoad ( self ) : <EOL> cppView1 = CumulusNative . PersistentCacheIndex ( <EOL> self . sharedState . newView ( ) , <EOL> callbackScheduler <EOL> ) <EOL> t0 = time . time ( ) <EOL> for index in range ( <NUM_LIT> ) : <EOL> if index % <NUM_LIT:1000> == <NUM_LIT:0> and index > <NUM_LIT:0> : <EOL> print index , ( time . time ( ) - t0 ) / ( index / <NUM_LIT> ) , "<STR_LIT>" <EOL> cppView1 . addPage ( sha1 ( "<STR_LIT>" + str ( index ) ) , HashSet ( ) , <NUM_LIT:1> , sha1 ( "<STR_LIT>" ) ) <EOL> print "<STR_LIT>" , time . time ( ) - t0 , "<STR_LIT>" <EOL> t1 = time . time ( ) <EOL> bytes0 = TCMallocNative . getBytesUsed ( ) <EOL> cppView2 = CumulusNative . PersistentCacheIndex ( <EOL> self . sharedState . newView ( ) , <EOL> callbackScheduler <EOL> ) <EOL> while cppView2 . totalBytesInCache ( ) < <NUM_LIT> : <EOL> time . sleep ( <NUM_LIT> ) <EOL> print cppView2 . totalBytesInCache ( ) <EOL> print "<STR_LIT>" , time . time ( ) - t1 , "<STR_LIT>" , ( TCMallocNative . getBytesUsed ( ) - bytes0 ) / <NUM_LIT> / <NUM_LIT> , "<STR_LIT>" <EOL> @ ComputedGraphTestHarness . UnderHarness <EOL> def test_circularPageReferencesAreInvalid ( self ) : <EOL> self . circularPageReferenceTest ( True ) <EOL> @ ComputedGraphTestHarness . UnderHarness <EOL> def test_noncircularPageReferencesAreValid ( self ) : <EOL> self . circularPageReferenceTest ( False ) <EOL> def circularPageReferenceTest ( self , shouldBeInvalid ) : <EOL> cppView1 = CumulusNative . PersistentCacheIndex ( <EOL> self . sharedState . newView ( ) , <EOL> callbackScheduler <EOL> ) <EOL> computationId = CumulusNative . ComputationId . Root ( <EOL> sha1 ( "<STR_LIT>" ) <EOL> ) <EOL> checkpointRequest = CumulusNative . CheckpointRequest ( <NUM_LIT:0.0> , True , computationId ) <EOL> cppView1 . addBigvec ( sha1 ( "<STR_LIT>" ) , HashSet ( ) + sha1 ( "<STR_LIT>" ) , <NUM_LIT:2> , sha1 ( "<STR_LIT>" ) ) <EOL> cppView1 . addPage ( sha1 ( "<STR_LIT>" ) , ( HashSet ( ) + sha1 ( "<STR_LIT>" ) ) if shouldBeInvalid else HashSet ( ) , <NUM_LIT:1> , sha1 ( "<STR_LIT>" ) ) <EOL> cppView1 . addCheckpointFile ( checkpointRequest , sha1 ( "<STR_LIT:file>" ) , HashSet ( ) + sha1 ( "<STR_LIT>" ) , <NUM_LIT:2> , sha1 ( "<STR_LIT>" ) ) <EOL> cppView1 . addCheckpoint ( checkpointRequest , HashSet ( ) + sha1 ( "<STR_LIT:file>" ) , <NUM_LIT:2> , sha1 ( "<STR_LIT>" ) , True , <NUM_LIT:1.0> , HashSet ( ) ) <EOL> self . assertTrue ( <EOL> len ( cppView1 . computeInvalidObjects ( ) ) == <EOL> ( <NUM_LIT:4> if shouldBeInvalid else <NUM_LIT:0> ) , <EOL> "<STR_LIT>" % ( len ( cppView1 . computeInvalidObjects ( ) ) , ( <NUM_LIT:4> if shouldBeInvalid else <NUM_LIT:0> ) ) <EOL> ) <EOL> @ ComputedGraphTestHarness . UnderHarness <EOL> def test_orphanedPageIsCollected ( self ) : <EOL> cppView1 = CumulusNative . PersistentCacheIndex ( <EOL> self . sharedState . newView ( ) , <EOL> callbackScheduler <EOL> ) <EOL> cppView1 . addPage ( sha1 ( "<STR_LIT>" ) , HashSet ( ) , <NUM_LIT:1> , sha1 ( "<STR_LIT>" ) ) <EOL> self . assertTrue ( len ( cppView1 . computeInvalidObjects ( ) ) == <NUM_LIT:1> ) <EOL> @ ComputedGraphTestHarness . UnderHarness <EOL> def test_basicPersistentCache ( self ) : <EOL> cppView1 = CumulusNative . PersistentCacheIndex ( <EOL> self . sharedState . newView ( ) , <EOL> callbackScheduler <EOL> ) <EOL> cppView2 = CumulusNative . PersistentCacheIndex ( <EOL> self . sharedState . newView ( ) , <EOL> callbackScheduler <EOL> ) <EOL> cppView1 . addPage ( sha1 ( "<STR_LIT>" ) , HashSet ( ) , <NUM_LIT:1> , sha1 ( "<STR_LIT>" ) ) <EOL> cppView1 . addBigvec ( sha1 ( "<STR_LIT>" ) , HashSet ( ) + sha1 ( "<STR_LIT>" ) , <NUM_LIT:2> , sha1 ( "<STR_LIT>" ) ) <EOL> cppView1 . addPage ( sha1 ( "<STR_LIT>" ) , HashSet ( ) + sha1 ( "<STR_LIT>" ) , <NUM_LIT:3> , sha1 ( "<STR_LIT>" ) ) <EOL> cppView1 . addBigvec ( sha1 ( "<STR_LIT>" ) , HashSet ( ) + sha1 ( "<STR_LIT>" ) , <NUM_LIT:4> , sha1 ( "<STR_LIT>" ) ) <EOL> self . assertEqual ( cppView1 . totalBytesInCache ( ) , <NUM_LIT:10> ) <EOL> def seesEverything ( ) : <EOL> if not cppView2 . pageExists ( sha1 ( "<STR_LIT>" ) ) : <EOL> return False <EOL> if not cppView2 . pageExists ( sha1 ( "<STR_LIT>" ) ) : <EOL> return False <EOL> if not cppView2 . bigvecExists ( sha1 ( "<STR_LIT>" ) ) : <EOL> return False <EOL> if not cppView2 . bigvecExists ( sha1 ( "<STR_LIT>" ) ) : <EOL> return False <EOL> return True <EOL> self . waitForSync ( seesEverything ) <EOL> for view in [ cppView1 , cppView2 ] : <EOL> self . assertEqual ( view . pageBytecount ( sha1 ( "<STR_LIT>" ) ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( view . bigvecBytecount ( sha1 ( "<STR_LIT>" ) ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( view . pageBytecount ( sha1 ( "<STR_LIT>" ) ) , <NUM_LIT:3> ) <EOL> self . assertEqual ( view . bigvecBytecount ( sha1 ( "<STR_LIT>" ) ) , <NUM_LIT:4> ) <EOL> self . assertEqual ( view . totalBytesInCache ( ) , <NUM_LIT:10> ) <EOL> @ ComputedGraphTestHarness . UnderHarness <EOL> def test_writing_while_disconnected ( self ) : <EOL> currentView = [ self . sharedState . newView ( ) ] <EOL> cppView1 = CumulusNative . PersistentCacheIndex ( <EOL> currentView [ <NUM_LIT:0> ] , <EOL> callbackScheduler <EOL> ) <EOL> def writeInLoop ( ) : <EOL> for ix in range ( <NUM_LIT:100> ) : <EOL> time . sleep ( <NUM_LIT> ) <EOL> cppView1 . addPage ( sha1 ( "<STR_LIT>" + str ( ix ) ) , HashSet ( ) , ix , sha1 ( "<STR_LIT>" ) ) <EOL> thread1 = threading . Thread ( target = writeInLoop ) <EOL> thread1 . start ( ) <EOL> def disconnectAndReconnectInLoop ( ) : <EOL> ix = <NUM_LIT:0> <EOL> while thread1 . isAlive ( ) : <EOL> ix += <NUM_LIT:1> <EOL> time . sleep ( <NUM_LIT> ) <EOL> currentView [ <NUM_LIT:0> ] . disconnect ( ) <EOL> currentView [ <NUM_LIT:0> ] = self . sharedState . newView ( ) <EOL> cppView1 . resetView ( currentView [ <NUM_LIT:0> ] ) <EOL> thread2 = threading . Thread ( target = disconnectAndReconnectInLoop ) <EOL> thread2 . start ( ) <EOL> thread1 . join ( ) <EOL> thread2 . join ( ) <EOL> self . assertTrue ( cppView1 . timesViewReconnected ( ) > <NUM_LIT:10> ) <EOL> cppView2 = CumulusNative . PersistentCacheIndex ( <EOL> self . sharedState . newView ( ) , <EOL> callbackScheduler <EOL> ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> count1 = <NUM_LIT:0> <EOL> count2 = <NUM_LIT:0> <EOL> for ix in range ( <NUM_LIT:100> ) : <EOL> if cppView1 . pageExists ( sha1 ( "<STR_LIT>" + str ( ix ) ) ) : <EOL> count1 += <NUM_LIT:1> <EOL> if cppView2 . pageExists ( sha1 ( "<STR_LIT>" + str ( ix ) ) ) : <EOL> count2 += <NUM_LIT:1> <EOL> self . assertTrue ( count1 == <NUM_LIT:100> and count2 == <NUM_LIT:100> , ( count1 , count2 ) ) </s>
