<s> import unittest <EOL> import time <EOL> import logging <EOL> import threading <EOL> import uuid <EOL> import shutil <EOL> import os <EOL> import tempfile <EOL> import ufora . FORA . python . FORA as FORA <EOL> import ufora . distributed . S3 . ActualS3Interface as ActualS3Interface <EOL> import ufora . cumulus . test . InMemoryCumulusSimulation as InMemoryCumulusSimulation <EOL> import ufora . native . CallbackScheduler as CallbackScheduler <EOL> import ufora . test . PerformanceTestReporter as PerformanceTestReporter <EOL> import ufora . FORA . python . Runtime as Runtime <EOL> import ufora . native . Cumulus as CumulusNative <EOL> import ufora . native . FORA as ForaNative <EOL> import ufora . native . Hash as HashNative <EOL> import ufora . distributed . Storage . S3ObjectStore as S3ObjectStore <EOL> import ufora . config . Setup as Setup <EOL> callbackScheduler = CallbackScheduler . singletonForTesting ( ) <EOL> class BigboxDiskPerformanceTest ( unittest . TestCase ) : <EOL> def test_disk_scans ( self ) : <EOL> s3 = ActualS3Interface . ActualS3InterfaceFactory ( ) <EOL> objectStore = S3ObjectStore . S3ObjectStore ( <EOL> s3 , <EOL> Setup . config ( ) . userDataS3Bucket , <EOL> prefix = "<STR_LIT>" <EOL> ) <EOL> _ , simulation = InMemoryCumulusSimulation . computeUsingSeveralWorkers ( <EOL> "<STR_LIT>" , <EOL> s3 , <EOL> <NUM_LIT:1> , <EOL> memoryLimitMb = <NUM_LIT:1> * <NUM_LIT> , <EOL> threadCount = <NUM_LIT:30> , <EOL> returnSimulation = True , <EOL> ioTaskThreadOverride = <NUM_LIT:8> , <EOL> objectStore = objectStore , <EOL> useInMemoryCache = False <EOL> ) <EOL> try : <EOL> gigabytes = <NUM_LIT:8> <EOL> t0 = time . time ( ) <EOL> resultVectors = [ ] <EOL> for ix in range ( gigabytes ) : <EOL> result = simulation . compute ( "<STR_LIT>" % ix , timeout = <NUM_LIT> ) <EOL> resultVectors . append ( result . asResult . result ) <EOL> t1 = time . time ( ) <EOL> intResults = [ ] <EOL> for vec in resultVectors : <EOL> result = simulation . compute ( "<STR_LIT>" , timeout = <NUM_LIT> , v = vec ) <EOL> intResults . append ( result . asResult . result . pyval ) <EOL> self . assertTrue ( len ( intResults ) == gigabytes ) <EOL> PerformanceTestReporter . recordTest ( <EOL> "<STR_LIT>" , <EOL> t1 - t0 , <EOL> None <EOL> ) <EOL> PerformanceTestReporter . recordTest ( <EOL> "<STR_LIT>" % gigabytes , <EOL> time . time ( ) - t0 , <EOL> None <EOL> ) <EOL> finally : <EOL> simulation . teardown ( ) <EOL> def diskThroughputTest ( self , gb ) : <EOL> if os . getenv ( "<STR_LIT>" ) is None : <EOL> dataDir = tempfile . mkdtemp ( ) <EOL> else : <EOL> dataDir = os . getenv ( "<STR_LIT>" ) <EOL> dataDir = os . path . join ( dataDir , str ( uuid . uuid4 ( ) ) ) <EOL> diskCache = CumulusNative . DiskOfflineCache ( <EOL> callbackScheduler , <EOL> dataDir , <EOL> <NUM_LIT:100> * <NUM_LIT> * <NUM_LIT> * <NUM_LIT> , <EOL> <NUM_LIT> <EOL> ) <EOL> fiftyMegabytes = ForaNative . encodeStringInSerializedObject ( "<STR_LIT:U+0020>" * <NUM_LIT> * <NUM_LIT> * <NUM_LIT:50> ) <EOL> logging . info ( "<STR_LIT>" , dataDir ) <EOL> try : <EOL> t0 = time . time ( ) <EOL> for ix in range ( gb * <NUM_LIT:20> ) : <EOL> diskCache . store ( <EOL> ForaNative . PageId ( HashNative . Hash . sha1 ( str ( ix ) ) , <NUM_LIT:50> * <NUM_LIT> * <NUM_LIT> , <NUM_LIT:50> * <NUM_LIT> * <NUM_LIT> ) , <EOL> fiftyMegabytes <EOL> ) <EOL> PerformanceTestReporter . recordTest ( <EOL> "<STR_LIT>" % gb , <EOL> time . time ( ) - t0 , <EOL> None <EOL> ) <EOL> t0 = time . time ( ) <EOL> for ix in range ( gb * <NUM_LIT:20> ) : <EOL> diskCache . loadIfExists ( <EOL> ForaNative . PageId ( HashNative . Hash . sha1 ( str ( ix ) ) , <NUM_LIT:50> * <NUM_LIT> * <NUM_LIT> , <NUM_LIT:50> * <NUM_LIT> * <NUM_LIT> ) <EOL> ) <EOL> PerformanceTestReporter . recordTest ( <EOL> "<STR_LIT>" % gb , <EOL> time . time ( ) - t0 , <EOL> None <EOL> ) <EOL> finally : <EOL> shutil . rmtree ( dataDir ) <EOL> def test_disk_read_and_write_perf ( self ) : <EOL> if os . getenv ( "<STR_LIT>" ) is None : <EOL> dataDir = tempfile . mkdtemp ( ) <EOL> else : <EOL> dataDir = os . getenv ( "<STR_LIT>" ) <EOL> dataDir = os . path . join ( dataDir , str ( uuid . uuid4 ( ) ) ) <EOL> diskCache = CumulusNative . DiskOfflineCache ( <EOL> callbackScheduler , <EOL> dataDir , <EOL> <NUM_LIT:100> * <NUM_LIT> * <NUM_LIT> * <NUM_LIT> , <EOL> <NUM_LIT> <EOL> ) <EOL> try : <EOL> fiftyMegabytes = ForaNative . encodeStringInSerializedObject ( "<STR_LIT:U+0020>" * <NUM_LIT> * <NUM_LIT> * <NUM_LIT:50> ) <EOL> logging . info ( "<STR_LIT>" , dataDir ) <EOL> storedPageID = ForaNative . PageId ( HashNative . Hash . sha1 ( "<STR_LIT>" ) , <NUM_LIT:50> * <NUM_LIT> * <NUM_LIT> , <NUM_LIT:50> * <NUM_LIT> * <NUM_LIT> ) <EOL> diskCache . store ( storedPageID , fiftyMegabytes ) <EOL> t0 = time . time ( ) <EOL> TOTAL_SECONDS = <NUM_LIT> <EOL> totalReadBytes = [ <NUM_LIT:0> ] <EOL> totalWriteBytes = [ <NUM_LIT:0> ] <EOL> def readerThread ( ) : <EOL> while time . time ( ) - t0 < TOTAL_SECONDS : <EOL> diskCache . loadIfExists ( storedPageID ) <EOL> totalReadBytes [ <NUM_LIT:0> ] += <NUM_LIT:50> <EOL> def writerThread ( ) : <EOL> ix = <NUM_LIT:0> <EOL> while time . time ( ) - t0 < TOTAL_SECONDS : <EOL> ix += <NUM_LIT:1> <EOL> diskCache . store ( <EOL> ForaNative . PageId ( HashNative . Hash . sha1 ( str ( ix ) ) , <NUM_LIT:50> * <NUM_LIT> * <NUM_LIT> , <NUM_LIT:50> * <NUM_LIT> * <NUM_LIT> ) , <EOL> fiftyMegabytes <EOL> ) <EOL> totalWriteBytes [ <NUM_LIT:0> ] += <NUM_LIT:50> <EOL> threads = [ <EOL> threading . Thread ( target = readerThread ) , <EOL> threading . Thread ( target = writerThread ) <EOL> ] <EOL> for t in threads : <EOL> t . start ( ) <EOL> for t in threads : <EOL> t . join ( ) <EOL> PerformanceTestReporter . recordTest ( <EOL> "<STR_LIT>" , <EOL> <NUM_LIT> / ( totalWriteBytes [ <NUM_LIT:0> ] / ( time . time ( ) - t0 ) ) , <EOL> None <EOL> ) <EOL> PerformanceTestReporter . recordTest ( <EOL> "<STR_LIT>" , <EOL> <NUM_LIT> / ( totalReadBytes [ <NUM_LIT:0> ] / ( time . time ( ) - t0 ) ) , <EOL> None <EOL> ) <EOL> finally : <EOL> shutil . rmtree ( dataDir ) <EOL> def test_disk_throughput ( self ) : <EOL> self . diskThroughputTest ( <NUM_LIT:1> ) <EOL> self . diskThroughputTest ( <NUM_LIT:2> ) <EOL> self . diskThroughputTest ( <NUM_LIT:4> ) <EOL> self . diskThroughputTest ( <NUM_LIT:8> ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> import ufora . config . Mainline as Mainline <EOL> Mainline . UnitTestMainline ( [ FORA , Runtime ] ) </s>
