<s> import unittest <EOL> import time <EOL> import logging <EOL> import traceback <EOL> import ufora . native . Cumulus as CumulusNative <EOL> import ufora . FORA . python . FORA as FORA <EOL> import ufora . cumulus . test . InMemoryCumulusSimulation as InMemoryCumulusSimulation <EOL> import ufora . distributed . S3 . InMemoryS3Interface as InMemoryS3Interface <EOL> import ufora . distributed . S3 . ActualS3Interface as ActualS3Interface <EOL> import ufora . native . CallbackScheduler as CallbackScheduler <EOL> import ufora . test . PerformanceTestReporter as PerformanceTestReporter <EOL> import ufora . FORA . python . Runtime as Runtime <EOL> import uuid <EOL> import os <EOL> import ufora . distributed . Storage . S3ObjectStore as S3ObjectStore <EOL> import ufora . config . Setup as Setup <EOL> callbackScheduler = CallbackScheduler . singletonForTesting ( ) <EOL> class BigboxPerformanceTest ( unittest . TestCase ) : <EOL> def getTestDataBucket ( self ) : <EOL> aws_az_key = '<STR_LIT>' <EOL> bucketName = '<STR_LIT>' <EOL> if aws_az_key in os . environ : <EOL> az = os . environ [ aws_az_key ] <EOL> if az is not '<STR_LIT>' : <EOL> region = az [ : - <NUM_LIT:1> ] <EOL> bucketName += '<STR_LIT:->' + region <EOL> logging . info ( "<STR_LIT>" , az , region ) <EOL> else : <EOL> logging . info ( "<STR_LIT>" ) <EOL> return bucketName <EOL> def computeUsingSeveralWorkers ( self , * args , ** kwds ) : <EOL> return InMemoryCumulusSimulation . computeUsingSeveralWorkers ( * args , ** kwds ) <EOL> @ staticmethod <EOL> def createObjectStore ( s3Service ) : <EOL> return S3ObjectStore . S3ObjectStore ( <EOL> s3Service , <EOL> Setup . config ( ) . userDataS3Bucket , <EOL> prefix = "<STR_LIT>" <EOL> ) <EOL> def downloadTaxiData ( self , <EOL> filecount , <EOL> parse = False , <EOL> workers = <NUM_LIT:1> , <EOL> threadsPerWorker = <NUM_LIT:30> , <EOL> downloaderThreads = <NUM_LIT:8> ) : <EOL> s3 = ActualS3Interface . ActualS3InterfaceFactory ( ) <EOL> bucketName = self . getTestDataBucket ( ) <EOL> result , simulation = self . computeUsingSeveralWorkers ( <EOL> "<STR_LIT>" , <EOL> s3 , <EOL> workers , <EOL> memoryLimitMb = <NUM_LIT> * <NUM_LIT> / workers , <EOL> threadCount = threadsPerWorker , <EOL> returnSimulation = True , <EOL> ioTaskThreadOverride = downloaderThreads , <EOL> useInMemoryCache = False , <EOL> objectStore = self . createObjectStore ( s3 ) <EOL> ) <EOL> try : <EOL> dsText = ( <EOL> """<STR_LIT>""" + "<STR_LIT:+>" . join ( [ <EOL> '<STR_LIT>' % ( bucketName , ix ) for ix in range ( <NUM_LIT:1> , filecount + <NUM_LIT:1> ) <EOL> ] ) + "<STR_LIT:;>" <EOL> ) <EOL> text = dsText + "<STR_LIT>" <EOL> downloadTimeStart = time . time ( ) <EOL> result = simulation . compute ( text , timeout = <NUM_LIT> ) <EOL> self . assertTrue ( result . isResult ( ) ) <EOL> downloadTimeEnd = time . time ( ) <EOL> ds , dsSum , bytecount = result . asResult . result <EOL> if parse : <EOL> parseTimeStart = time . time ( ) <EOL> result = simulation . compute ( "<STR_LIT>" , timeout = <NUM_LIT> , ds = ds ) <EOL> parseTimeEnd = time . time ( ) <EOL> self . assertTrue ( result . isResult ( ) ) <EOL> PerformanceTestReporter . recordTest ( <EOL> "<STR_LIT>" + str ( filecount ) , <EOL> parseTimeEnd - parseTimeStart , <EOL> None <EOL> ) <EOL> else : <EOL> bytecount = bytecount . pyval <EOL> PerformanceTestReporter . recordTest ( <EOL> "<STR_LIT>" + str ( filecount ) , <EOL> ( downloadTimeEnd - downloadTimeStart ) / ( bytecount / <NUM_LIT> / <NUM_LIT> / <NUM_LIT> ) , <EOL> None <EOL> ) <EOL> finally : <EOL> simulation . teardown ( ) <EOL> def test_parseTaxiData_1 ( self ) : <EOL> self . downloadTaxiData ( <NUM_LIT:1> , parse = True ) <EOL> def test_downloadTaxiData_1 ( self ) : <EOL> self . downloadTaxiData ( <NUM_LIT:1> ) <EOL> def test_downloadTaxiData_2 ( self ) : <EOL> self . downloadTaxiData ( <NUM_LIT:2> ) <EOL> def test_downloadTaxiData_6 ( self ) : <EOL> self . downloadTaxiData ( <NUM_LIT:6> ) <EOL> def dataCreationTest ( self , totalMB , workers , threadsPerWorker , testName ) : <EOL> s3 = InMemoryS3Interface . InMemoryS3InterfaceFactory ( ) <EOL> text = """<STR_LIT>""" % ( totalMB * <NUM_LIT> * <NUM_LIT> / <NUM_LIT:8> ) <EOL> t0 = time . time ( ) <EOL> result , simulation = self . computeUsingSeveralWorkers ( <EOL> "<STR_LIT>" , <EOL> s3 , <EOL> workers , <EOL> timeout = <NUM_LIT> , <EOL> memoryLimitMb = <NUM_LIT> * <NUM_LIT> / workers , <EOL> threadCount = threadsPerWorker , <EOL> returnSimulation = True , <EOL> useInMemoryCache = False <EOL> ) <EOL> try : <EOL> t0 = time . time ( ) <EOL> result = simulation . compute ( text , timeout = <NUM_LIT> ) <EOL> totalTimeToReturnResult = time . time ( ) - t0 <EOL> PerformanceTestReporter . recordTest ( testName , totalTimeToReturnResult , None ) <EOL> finally : <EOL> simulation . teardown ( ) <EOL> def test_createData_4000_1_30_threads ( self ) : <EOL> self . dataCreationTest ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:30> , "<STR_LIT>" ) <EOL> def test_createData_4000_1_2_threads ( self ) : <EOL> self . dataCreationTest ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:2> , "<STR_LIT>" ) <EOL> def test_createData_40000_1_30_threads ( self ) : <EOL> self . dataCreationTest ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:30> , "<STR_LIT>" ) <EOL> def test_createData_40000_1_30_threads_2 ( self ) : <EOL> self . dataCreationTest ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:30> , "<STR_LIT>" ) <EOL> def test_createData_40000_1_2_threads ( self ) : <EOL> self . dataCreationTest ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:2> , "<STR_LIT>" ) <EOL> @ PerformanceTestReporter . PerfTest ( "<STR_LIT>" ) <EOL> def test_byteToStringAndBackInDifferentPatterns ( self ) : <EOL> s3 = ActualS3Interface . ActualS3InterfaceFactory ( ) <EOL> setupText = ( <EOL> """<STR_LIT>""" <EOL> ) <EOL> setupResults , simulation = self . computeUsingSeveralWorkers ( <EOL> setupText , <EOL> s3 , <EOL> <NUM_LIT:1> , <EOL> memoryLimitMb = <NUM_LIT> * <NUM_LIT> , <EOL> threadCount = <NUM_LIT:30> , <EOL> returnSimulation = True , <EOL> ioTaskThreadOverride = <NUM_LIT:8> , <EOL> useInMemoryCache = False , <EOL> timeout = <NUM_LIT:30> , <EOL> objectStore = self . createObjectStore ( s3 ) <EOL> ) <EOL> try : <EOL> ds , dat , datSum = setupResults . asResult . result <EOL> t0 = time . time ( ) <EOL> result = simulation . compute ( <EOL> "<STR_LIT>" , <EOL> timeout = <NUM_LIT> , <EOL> ds = ds , <EOL> dat = dat , <EOL> datSum = datSum <EOL> ) <EOL> PerformanceTestReporter . recordTest ( <EOL> "<STR_LIT>" , <EOL> time . time ( ) - t0 , <EOL> None <EOL> ) <EOL> t0 = time . time ( ) <EOL> result = simulation . compute ( <EOL> "<STR_LIT>" , <EOL> timeout = <NUM_LIT> , <EOL> ds = ds , <EOL> dat = dat , <EOL> datSum = datSum <EOL> ) <EOL> PerformanceTestReporter . recordTest ( <EOL> "<STR_LIT>" , <EOL> time . time ( ) - t0 , <EOL> None <EOL> ) <EOL> finally : <EOL> simulation . teardown ( ) <EOL> @ PerformanceTestReporter . PerfTest ( "<STR_LIT>" ) <EOL> def test_write_to_s3_3mb ( self ) : <EOL> self . writeToS3Test ( <NUM_LIT> ) <EOL> @ PerformanceTestReporter . PerfTest ( "<STR_LIT>" ) <EOL> def test_write_to_s3_50mb ( self ) : <EOL> self . writeToS3Test ( <NUM_LIT:50> * <NUM_LIT:1000> * <NUM_LIT:1000> ) <EOL> @ PerformanceTestReporter . PerfTest ( "<STR_LIT>" ) <EOL> def test_write_to_s3_200mb_multibox ( self ) : <EOL> self . writeToS3Test ( <NUM_LIT:200> * <NUM_LIT:1000> * <NUM_LIT:1000> , workers = <NUM_LIT:4> , memoryLimitMb = <NUM_LIT:100> , threadCount = <NUM_LIT:1> ) <EOL> @ PerformanceTestReporter . PerfTest ( "<STR_LIT>" ) <EOL> def test_write_to_s3_2gb_multibox ( self ) : <EOL> self . writeToS3Test ( <NUM_LIT:1000> * <NUM_LIT:1000> * <NUM_LIT:1000> , workers = <NUM_LIT:4> , memoryLimitMb = <NUM_LIT:1000> , threadCount = <NUM_LIT:2> ) <EOL> def writeToS3Test ( self , bytecount , pageSizeOverride = <NUM_LIT> * <NUM_LIT> , workers = <NUM_LIT:1> , memoryLimitMb = <NUM_LIT> * <NUM_LIT> , threadCount = <NUM_LIT:30> ) : <EOL> text = """<STR_LIT>""" <EOL> s3 = ActualS3Interface . ActualS3InterfaceFactory ( ) <EOL> keyGuid = "<STR_LIT>" + str ( uuid . uuid4 ( ) ) <EOL> try : <EOL> setupText = text . replace ( '<STR_LIT>' , str ( bytecount ) ) <EOL> setupResults , simulation = self . computeUsingSeveralWorkers ( <EOL> setupText , <EOL> s3 , <EOL> workers , <EOL> memoryLimitMb = memoryLimitMb , <EOL> threadCount = threadCount , <EOL> returnSimulation = True , <EOL> ioTaskThreadOverride = <NUM_LIT:8> , <EOL> useInMemoryCache = False , <EOL> timeout = <NUM_LIT:30> , <EOL> objectStore = self . createObjectStore ( s3 ) , <EOL> pageSizeOverride = pageSizeOverride <EOL> ) <EOL> result = simulation . executeExternalIoTask ( <EOL> CumulusNative . ExternalIoTask . WriteCharBigvecToS3 ( <EOL> setupResults . asResult . result . getVectorBigvecGuid ( ) , <EOL> CumulusNative . S3KeyAndCredentials ( <EOL> self . getTestDataBucket ( ) , <EOL> keyGuid , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> ) <EOL> ) , <EOL> timeout = <NUM_LIT> <EOL> ) <EOL> self . assertTrue ( result . isSuccess ( ) , result ) <EOL> assert s3 ( ) . getKeySize ( self . getTestDataBucket ( ) , keyGuid ) == bytecount <EOL> finally : <EOL> try : <EOL> s3 ( ) . deleteKey ( self . getTestDataBucket ( ) , keyGuid ) <EOL> except : <EOL> logging . warn ( "<STR_LIT>" , traceback . format_exc ( ) ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> import ufora . config . Mainline as Mainline <EOL> Mainline . UnitTestMainline ( [ FORA , Runtime ] ) </s>
