doc/core/benchmarks/linereceiver.py

   1 import math, time
   2
   3 from twisted.protocols import basic
   4
   5 class CollectingLineReceiver(basic.LineReceiver):
   6     def __init__(self):
   7         self.lines = []
   8         self.lineReceived = self.lines.append
   9
  10 def deliver(proto, chunks):
  11     map(proto.dataReceived, chunks)
  12
  13 def benchmark(chunkSize, lineLength, numLines):
  14     bytes = ('x' * lineLength + '\r\n') * numLines
  15     chunkCount = len(bytes) / chunkSize + 1
  16     chunks = []
  17     for n in xrange(chunkCount):
  18         chunks.append(bytes[n*chunkSize:(n+1)*chunkSize])
  19     assert ''.join(chunks) == bytes, (chunks, bytes)
  20     p = CollectingLineReceiver()
  21
  22     before = time.clock()
  23     deliver(p, chunks)
  24     after = time.clock()
  25
  26     assert bytes.splitlines() == p.lines, (bytes.splitlines(), p.lines)
  27
  28     print 'chunkSize:', chunkSize,
  29     print 'lineLength:', lineLength,
  30     print 'numLines:', numLines,
  31     print 'CPU Time: ', after - before
  32
  33
  34
  35 def main():
  36     for numLines in 100, 1000:
  37         for lineLength in (10, 100, 1000):
  38             for chunkSize in (1, 500, 5000):
  39                 benchmark(chunkSize, lineLength, numLines)
  40
  41     for numLines in 10000, 50000:
  42         for lineLength in (1000, 2000):
  43             for chunkSize in (51, 500, 5000):
  44                 benchmark(chunkSize, lineLength, numLines)
  45
  46 if __name__ == '__main__':
  47     main()