<s> from twisted . test import proto_helpers , test_protocols <EOL> import math <EOL> import time <EOL> import sys <EOL> import os <EOL> import gc <EOL> NETSTRING_PREFIX_TEMPLATE = "<STR_LIT>" <EOL> NETSTRING_POSTFIX = "<STR_LIT:U+002C>" <EOL> USAGE = """<STR_LIT>""" <EOL> class PerformanceTester ( object ) : <EOL> """<STR_LIT>""" <EOL> headers = [ ] <EOL> lineFormat = "<STR_LIT>" <EOL> performanceData = { } <EOL> def __init__ ( self , filename ) : <EOL> """<STR_LIT>""" <EOL> if os . path . isfile ( filename ) : <EOL> response = raw_input ( ( "<STR_LIT>" <EOL> "<STR_LIT>" ) % filename ) <EOL> if response . lower ( ) != "<STR_LIT:y>" : <EOL> print "<STR_LIT>" <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> self . filename = filename <EOL> def testPerformance ( self , number ) : <EOL> """<STR_LIT>""" <EOL> for iteration in xrange ( number ) : <EOL> self . performTest ( iteration ) <EOL> def performTest ( self , iteration ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def createReport ( self ) : <EOL> """<STR_LIT>""" <EOL> self . outputFile = open ( self . filename , "<STR_LIT:w>" ) <EOL> self . writeHeader ( ) <EOL> self . writePerformanceData ( ) <EOL> self . writeLineSeparator ( ) <EOL> print "<STR_LIT>" % self . filename <EOL> def writeHeader ( self ) : <EOL> """<STR_LIT>""" <EOL> self . writeLineSeparator ( ) <EOL> self . outputFile . write ( "<STR_LIT>" % ( "<STR_LIT>" . join ( self . headers ) , ) ) <EOL> self . writeLineSeparator ( ) <EOL> def writeLineSeparator ( self ) : <EOL> """<STR_LIT>""" <EOL> dashes = ( "<STR_LIT:->" * ( len ( header ) + <NUM_LIT:2> ) for header in self . headers ) <EOL> self . outputFile . write ( "<STR_LIT>" % "<STR_LIT:+>" . join ( dashes ) ) <EOL> def writePerformanceData ( self ) : <EOL> """<STR_LIT>""" <EOL> for combination , elapsed in sorted ( self . performanceData . iteritems ( ) ) : <EOL> totalSize , chunkSize , numberOfChunks = combination <EOL> self . outputFile . write ( self . lineFormat % <EOL> ( totalSize , chunkSize , numberOfChunks , <EOL> elapsed ) ) <EOL> class NetstringPerformanceTester ( PerformanceTester ) : <EOL> """<STR_LIT>""" <EOL> headers = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ] <EOL> lineFormat = ( "<STR_LIT>" % <EOL> tuple ( [ len ( header ) for header in headers ] ) ) <EOL> def __init__ ( self , filename ) : <EOL> """<STR_LIT>""" <EOL> PerformanceTester . __init__ ( self , filename ) <EOL> transport = proto_helpers . StringTransport ( ) <EOL> self . netstringReceiver = test_protocols . TestNetstring ( ) <EOL> self . netstringReceiver . makeConnection ( transport ) <EOL> def performTest ( self , number ) : <EOL> """<STR_LIT>""" <EOL> chunkSize = <NUM_LIT:2> ** number <EOL> numberOfChunks = chunkSize <EOL> while numberOfChunks : <EOL> self . testCombination ( chunkSize , numberOfChunks ) <EOL> numberOfChunks = numberOfChunks // <NUM_LIT:2> <EOL> def testCombination ( self , chunkSize , numberOfChunks ) : <EOL> """<STR_LIT>""" <EOL> chunk , dataSize = self . configureCombination ( chunkSize , numberOfChunks ) <EOL> elapsed = self . receiveData ( chunk , numberOfChunks , dataSize ) <EOL> key = ( chunkSize , numberOfChunks , dataSize ) <EOL> self . performanceData [ key ] = elapsed <EOL> def configureCombination ( self , chunkSize , numberOfChunks ) : <EOL> """<STR_LIT>""" <EOL> chunk = "<STR_LIT:a>" * chunkSize <EOL> dataSize = chunkSize * numberOfChunks <EOL> self . netstringReceiver . MAX_LENGTH = dataSize <EOL> numberOfDigits = math . ceil ( math . log10 ( dataSize ) ) + <NUM_LIT:1> <EOL> return chunk , dataSize <EOL> def receiveData ( self , chunk , numberOfChunks , dataSize ) : <EOL> dr = self . netstringReceiver . dataReceived <EOL> now = time . time ( ) <EOL> dr ( NETSTRING_PREFIX_TEMPLATE % ( dataSize , ) ) <EOL> for idx in xrange ( numberOfChunks ) : <EOL> dr ( chunk ) <EOL> dr ( NETSTRING_POSTFIX ) <EOL> elapsed = time . time ( ) - now <EOL> assert self . netstringReceiver . received , "<STR_LIT>" <EOL> return elapsed <EOL> def disableGarbageCollector ( ) : <EOL> gc . disable ( ) <EOL> print '<STR_LIT>' <EOL> def main ( number , filename ) : <EOL> disableGarbageCollector ( ) <EOL> npt = NetstringPerformanceTester ( filename ) <EOL> npt . testPerformance ( int ( number ) ) <EOL> npt . createReport ( ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> if len ( sys . argv ) < <NUM_LIT:3> : <EOL> print USAGE % sys . argv [ <NUM_LIT:0> ] <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> main ( * sys . argv [ <NUM_LIT:1> : <NUM_LIT:3> ] ) </s>
