doc/conch/benchmarks/buffering_mixin.py

   1 # Copyright (c) Twisted Matrix Laboratories.
   2 # See LICENSE for details.
   3
   4 """
   5 Benchmarks comparing the write performance of a "normal" Protocol instance
   6 and an instance of a Protocol class which has had L{twisted.conch.mixin}'s
   7 L{BufferingMixin<twisted.conch.mixin.BufferingMixin>} mixed in to perform
   8 Nagle-like write coalescing.
   9 """
  10
  11 from sys import stdout
  12 from pprint import pprint
  13 from time import time
  14
  15 from twisted.python.usage import Options
  16 from twisted.python.log import startLogging
  17
  18 from twisted.internet.protocol import ServerFactory, Protocol, ClientCreator
  19 from twisted.internet.defer import Deferred
  20 from twisted.internet import reactor
  21
  22 from twisted.conch.mixin import BufferingMixin
  23
  24
  25 class BufferingBenchmark(Options):
  26     """
  27     Options for configuring the execution parameters of a benchmark run.
  28     """
  29
  30     optParameters = [
  31         ('scale', 's', '1',
  32          'Work multiplier (bigger takes longer, might resist noise better)')]
  33
  34     def postOptions(self):
  35         self['scale'] = int(self['scale'])
  36
  37
  38
  39 class ServerProtocol(Protocol):
  40     """
  41     A silent protocol which only waits for a particular amount of input and
  42     then fires a Deferred.
  43     """
  44     def __init__(self, expected, finished):
  45         self.expected = expected
  46         self.finished = finished
  47
  48
  49     def dataReceived(self, bytes):
  50         self.expected -= len(bytes)
  51         if self.expected == 0:
  52             finished, self.finished = self.finished, None
  53             finished.callback(None)
  54
  55
  56
  57 class BufferingProtocol(Protocol, BufferingMixin):
  58     """
  59     A protocol which uses the buffering mixin to provide a write method.
  60     """
  61
  62
  63
  64 class UnbufferingProtocol(Protocol):
  65     """
  66     A protocol which provides a naive write method which simply passes through
  67     to the transport.
  68     """
  69
  70     def connectionMade(self):
  71         """
  72         Bind write to the transport's write method and flush to a no-op
  73         function in order to provide the same API as is provided by
  74         BufferingProtocol.
  75         """
  76         self.write = self.transport.write
  77         self.flush = lambda: None
  78
  79
  80
  81 def _write(proto, byteCount):
  82     write = proto.write
  83     flush = proto.flush
  84
  85     for i in range(byteCount):
  86         write('x')
  87     flush()
  88
  89
  90
  91 def _benchmark(byteCount, clientProtocol):
  92     result = {}
  93     finished = Deferred()
  94     def cbFinished(ignored):
  95         result[u'disconnected'] = time()
  96         result[u'duration'] = result[u'disconnected'] - result[u'connected']
  97         return result
  98     finished.addCallback(cbFinished)
  99
 100     f = ServerFactory()
 101     f.protocol = lambda: ServerProtocol(byteCount, finished)
 102     server = reactor.listenTCP(0, f)
 103
 104     f2 = ClientCreator(reactor, clientProtocol)
 105     proto = f2.connectTCP('127.0.0.1', server.getHost().port)
 106     def connected(proto):
 107         result[u'connected'] = time()
 108         return proto
 109     proto.addCallback(connected)
 110     proto.addCallback(_write, byteCount)
 111     return finished
 112
 113
 114
 115 def _benchmarkBuffered(byteCount):
 116     return _benchmark(byteCount, BufferingProtocol)
 117
 118
 119
 120 def _benchmarkUnbuffered(byteCount):
 121     return _benchmark(byteCount, UnbufferingProtocol)
 122
 123
 124
 125 def benchmark(scale=1):
 126     """
 127     Benchmark and return information regarding the relative performance of a
 128     protocol which does not use the buffering mixin and a protocol which
 129     does.
 130
 131     @type scale: C{int}
 132     @param scale: A multipler to the amount of work to perform
 133
 134     @return: A Deferred which will fire with a dictionary mapping each of
 135     the two unicode strings C{u'buffered'} and C{u'unbuffered'} to
 136     dictionaries describing the performance of a protocol of each type.
 137     These value dictionaries will map the unicode strings C{u'connected'}
 138     and C{u'disconnected'} to the times at which each of those events
 139     occurred and C{u'duration'} two the difference between these two values.
 140     """
 141     overallResult = {}
 142
 143     byteCount = 1024
 144
 145     bufferedDeferred = _benchmarkBuffered(byteCount * scale)
 146     def didBuffered(bufferedResult):
 147         overallResult[u'buffered'] = bufferedResult
 148         unbufferedDeferred =  _benchmarkUnbuffered(byteCount * scale)
 149         def didUnbuffered(unbufferedResult):
 150             overallResult[u'unbuffered'] = unbufferedResult
 151             return overallResult
 152         unbufferedDeferred.addCallback(didUnbuffered)
 153         return unbufferedDeferred
 154     bufferedDeferred.addCallback(didBuffered)
 155     return bufferedDeferred
 156
 157
 158
 159 def main(args=None):
 160     """
 161     Perform a single benchmark run, starting and stopping the reactor and
 162     logging system as necessary.
 163     """
 164     startLogging(stdout)
 165
 166     options = BufferingBenchmark()
 167     options.parseOptions(args)
 168
 169     d = benchmark(options['scale'])
 170     def cbBenchmark(result):
 171         pprint(result)
 172     def ebBenchmark(err):
 173         print err.getTraceback()
 174     d.addCallbacks(cbBenchmark, ebBenchmark)
 175     def stopReactor(ign):
 176         reactor.stop()
 177     d.addBoth(stopReactor)
 178     reactor.run()
 179
 180
 181 if __name__ == '__main__':
 182     main()