Ticket #2611: faster-string-receivers-2611.patch

doc/core/benchmarks/receivers_fragmentation.py

@@ +1 @@
+#!/usr/bin/env python
+
+"""
+Benchmark to test the performance of
+basic.(LineOnlyReceiver|LineReceiver|IntNStringReceiver) when receiving
+very fragmented packets.
+
+This benchmark was created to verify that the patch reducing the time
+complexity of these protocols from O(n^2) to O(n) worked correctly, and
+to make sure that the non-pathological cases didn't result in any
+performance degradation.
+
+This benchmark makes each tested protocol receive N messages, each split
+into X packets of M bytes, with X varying.
+
+Run with args: <N> <M> <min(X)> <max(X)> <step>.
+
+The output is: <X> <runtime(LineOnlyReceiver)> <runtime(LineReceiver)> <runtime(IntNStringReceiver)>
+"""
+
+import time
+import sys
+import struct
+
+from twisted.protocols import basic
+from twisted.internet import protocol
+from twisted.test.test_protocols import StringIOWithoutClosing
+
+class NoopLineOnlyReceiver(basic.LineOnlyReceiver):
+    MAX_LENGTH = 100000
+    delimiter = '\n'
+
+    def lineReceived(self, _):
+        pass
+
+class NoopLineReceiver(basic.LineReceiver):
+    MAX_LENGTH = 100000
+    delimiter = '\n'
+
+    def lineReceived(self, _):
+        pass
+
+class NoopInt32Receiver(basic.Int32StringReceiver):
+    MAX_LENGTH = 100000
+
+    def stringReceived(self, _):
+        pass
+
+def run_lineonly_iteration(num_lines, pkts_per_line, packet_size):
+    packet = 'a'*packet_size
+    packet_with_delimiter = packet + '\n'
+
+    t = StringIOWithoutClosing()
+    a = NoopLineOnlyReceiver()
+    a.makeConnection(protocol.FileWrapper(t))
+
+    start = time.time()
+    for _ in xrange(num_lines):
+        for _ in xrange(pkts_per_line-1):
+            a.dataReceived(packet)
+        a.dataReceived(packet_with_delimiter)
+    stop = time.time()
+    return stop - start
+
+def run_line_iteration(num_lines, pkts_per_line, packet_size):
+    packet = 'a'*packet_size
+    packet_with_delimiter = packet + '\n'
+
+    t = StringIOWithoutClosing()
+    a = NoopLineReceiver()
+    a.makeConnection(protocol.FileWrapper(t))
+
+    start = time.time()
+    for _ in xrange(num_lines):
+        for _ in xrange(pkts_per_line-1):
+            a.dataReceived(packet)
+        a.dataReceived(packet_with_delimiter)
+    stop = time.time()
+    return stop - start
+
+def run_int32_iteration(num_lines, pkts_per_msg, packet_size):
+    packet = 'a'*packet_size
+    packet_with_prefix = struct.pack('!I', pkts_per_msg*packet_size) + packet
+
+    t = StringIOWithoutClosing()
+    a = NoopInt32Receiver()
+    a.makeConnection(protocol.FileWrapper(t))
+
+    start = time.time()
+    for _ in xrange(num_lines):
+        for c in packet_with_prefix:
+            a.dataReceived(c)
+        for _ in xrange(pkts_per_msg-1):
+            a.dataReceived(packet)
+    stop = time.time()
+    return stop - start
+
+def run_over_range_of_num_pkts(num_lines, pkt_size, min, max, step):
+    for num_pkts in xrange(min, max, step):
+        t = run_lineonly_iteration(num_lines, num_pkts, pkt_size)
+        t2 = run_line_iteration(num_lines, num_pkts, pkt_size)
+        t3 = run_int32_iteration(num_lines, num_pkts, pkt_size)
+        print "%d %f %f %f" % (num_pkts, t, t2, t3)
+        sys.stdout.flush()
+
+def main():
+    run_over_range_of_num_pkts(*[int(x) for x in sys.argv[1:]])
+
+if __name__ == '__main__':
+    main()

twisted/protocols/basic.py

@@ -428 +428 @@
 
     @cvar delimiter: The line-ending delimiter to use. By default this is
                      '\\r\\n'.
+    @type delimiter: C{str}
+
     @cvar MAX_LENGTH: The maximum length of a line to allow (If a
                       sent line is longer than this, the connection is dropped).
                       Default is 16384.
+    @type MAX_LENGTH: C{int}
+
+    @ivar _buffer: A list of data pieces received and buffered, waiting
+                   for a delimiter.
+    @type _buffer: C{list} of C{str}
+
+    @ivar _buffer_size: The amount of data stored in L{_buffer}.
+    @type _buffer_size: C{int}
     """
-    _buffer = ''
+    _buffer = None
+    _buffer_size = 0
     delimiter = '\r\n'
     MAX_LENGTH = 16384
 
@@ -440 +451 @@
         """
         Translates bytes into lines, and calls lineReceived.
         """
-        lines  = (self._buffer+data).split(self.delimiter)
-        self._buffer = lines.pop(-1)
+        if self._buffer is None:
+            self._buffer = []
+        lines = data.split(self.delimiter)
+        trailing = lines.pop()
+
         for line in lines:
             if self.transport.disconnecting:
                 # this is necessary because the transport may be told to lose
@@ -449 +463 @@
                 # important to disregard all the lines in that packet following
                 # the one that told it to close.
                 return
+            self._buffer.append(line)
+            line = ''.join(self._buffer)
+            self._buffer, self._buffer_size = [], 0
+
             if len(line) > self.MAX_LENGTH:
                 return self.lineLengthExceeded(line)
             else:
                 self.lineReceived(line)
-        if len(self._buffer) > self.MAX_LENGTH:
-            return self.lineLengthExceeded(self._buffer)
+
+        self._buffer.append(trailing)
+        self._buffer_size += len(trailing)
+        if self._buffer_size > self.MAX_LENGTH:
+            return self.lineLengthExceeded(''.join(self._buffer))
 
 
     def lineReceived(self, line):
@@ -519 +540 @@
 
     @cvar delimiter: The line-ending delimiter to use. By default this is
                      '\\r\\n'.
+    @type delimiter: C{str}
+
     @cvar MAX_LENGTH: The maximum length of a line to allow (If a
                       sent line is longer than this, the connection is dropped).
                       Default is 16384.
+    @type MAX_LENGTH: C{int}
+
+    @ivar _buffer: A list of data pieces received and buffered. Either
+                   an incomplete line, or raw data when the protocol is
+                   paused.
+    @type _buffer: C{list} of C{str}
+
+    @ivar _buffer_size: The amount of data stored in L{_buffer}.
+    @type _buffer_size: C{int}
+
+    @ivar _was_paused: True if the protocol was paused the last time
+                       L{dataReceived} was called. This is needed to
+                       detect unpausing and trigger processing of
+                       buffered data.
+    @type _was_paused: C{bool}
     """
     line_mode = 1
-    __buffer = ''
+    _buffer = None
+    _buffer_size = 0
+    _was_paused = False
     delimiter = '\r\n'
     MAX_LENGTH = 16384
 
+    def _appendToLineBuffer(self, data):
+        self._buffer.append(data)
+        self._buffer_size += len(data)
+
     def clearLineBuffer(self):
         """
         Clear buffered data.
@@ -535 +579 @@
         @return: All of the cleared buffered data.
         @rtype: C{str}
         """
-        b = self.__buffer
-        self.__buffer = ""
-        return b
-
+        b = self._buffer
+        self._buffer = []
+        self._buffer_size = 0
+        return b        # XXX this probably isn't a string.
 
     def dataReceived(self, data):
         """
@@ -546 +590 @@
         Translates bytes into lines, and calls lineReceived (or
         rawDataReceived, depending on mode.)
         """
-        self.__buffer = self.__buffer+data
+        if self._buffer is None:
+            self._buffer = []
+
+        if self.paused:
+            self._was_paused = True
+            self._appendToLineBuffer(data)
+            return
+        elif self._was_paused:
+            # This is the call of dataReceived that follows an
+            # unpausing. The buffer has been accumulating data without
+            # processing it, so we need to send it all back through the
+            # pipes.
+            self._was_paused = False
+            self._buffer.append(data)
+            buf = self._buffer
+            self.clearLineBuffer()
+            for chunk in buf:
+                why = self.dataReceived(chunk)
+                if why or self.transport and self.transport.disconnecting:
+                    return why
+            return
+        self._was_paused = False
+
         while self.line_mode and not self.paused:
             try:
-                line, self.__buffer = self.__buffer.split(self.delimiter, 1)
+                line, data = data.split(self.delimiter, 1)
             except ValueError:
-                if len(self.__buffer) > self.MAX_LENGTH:
-                    line, self.__buffer = self.__buffer, ''
+                self._appendToLineBuffer(data)
+                if self._buffer_size > self.MAX_LENGTH:
+                    line = ''.join(self._buffer)
+                    self.clearLineBuffer()
                     return self.lineLengthExceeded(line)
                 break
             else:
-                linelength = len(line)
-                if linelength > self.MAX_LENGTH:
-                    exceeded = line + self.__buffer
-                    self.__buffer = ''
-                    return self.lineLengthExceeded(exceeded)
+                self._buffer.append(line)
+                line = ''.join(self._buffer)
+                self.clearLineBuffer()
+                if len(line) > self.MAX_LENGTH:
+                    return self.lineLengthExceeded(line)
                 why = self.lineReceived(line)
                 if why or self.transport and self.transport.disconnecting:
                     return why
+                elif self.paused:
+                    self._was_paused = True
+                    self._appendToLineBuffer(data)
         else:
             if not self.paused:
-                data=self.__buffer
-                self.__buffer=''
+                self._buffer.append(data)
+                data = ''.join(self._buffer)
+                self.clearLineBuffer()
                 if data:
                     return self.rawDataReceived(data)
 
@@ -622 +694 @@
         @param line: The line to send, not including the delimiter.
         @type line: C{str}
         """
-        return self.transport.write(line + self.delimiter)
+        return self.transport.writeSequence((line, self.delimiter))
 
 
     def lineLengthExceeded(self, line):
@@ -651 +723 @@
     """
     Generic class for length prefixed protocols.
 
-    @ivar recvd: buffer holding received data when splitted.
-    @type recvd: C{str}
-
     @ivar structFormat: format used for struct packing/unpacking. Define it in
         subclass.
     @type structFormat: C{str}
@@ -661 +730 @@
     @ivar prefixLength: length of the prefix, in bytes. Define it in subclass,
         using C{struct.calcsize(structFormat)}
     @type prefixLength: C{int}
+
+    @ivar _buffer: A list of data pieces received and buffered.
+    @type _buffer: C{list} of C{str}
+
+    @ivar _buffer_size: The amount of data stored in L{_buffer}.
+    @type _buffer_size: C{int}
+
+    @ivar _packet_length: The total length of the packet currently being
+                          buffered up, including the prefix. The length
+                          will be None until a full prefix can be
+                          decoded.
+    @type _packet_length: C{int} or C{None}
+
+    @ivar recvd: A string representation of C{_buffer}. This is a
+                 deprecated attribute, and will fire a
+                 DeprecationWarning when accessed. Be very careful if
+                 you write to this attribute, no sanity checking is
+                 performed on the contents being injected into the
+                 buffer.
+    @type recvd: C{str}
     """
     MAX_LENGTH = 99999
-    recvd = ""
+
+    _buffer = None
+    _buffer_size = 0
+    _packet_length = None
+
+    def _appendToBuffer(self, data):
+        self._buffer.append(data)
+        self._buffer_size += len(data)
+
+
+    def _clearBuffer(self):
+        """Clear buffered data."""
+        self._buffer, self._buffer_size = [], 0
+
 
     def stringReceived(self, string):
         """
@@ -688 +790 @@
         self.transport.loseConnection()
 
 
-    def dataReceived(self, recd):
+    def dataReceived(self, data):
         """
         Convert int prefixed strings into calls to stringReceived.
         """
-        self.recvd = self.recvd + recd
-        while len(self.recvd) >= self.prefixLength and not self.paused:
-            length ,= struct.unpack(
-                self.structFormat, self.recvd[:self.prefixLength])
-            if length > self.MAX_LENGTH:
-                self.lengthLimitExceeded(length)
-                return
-            if len(self.recvd) < length + self.prefixLength:
-                break
-            packet = self.recvd[self.prefixLength:length + self.prefixLength]
-            self.recvd = self.recvd[length + self.prefixLength:]
-            self.stringReceived(packet)
+        if self._buffer is None:
+            self._buffer = []
+        self._appendToBuffer(data)
+
+        while not self.paused:
+            if self._packet_length is None:
+                if self._buffer_size < self.prefixLength:
+                    return
+                data = ''.join(self._buffer)
+                self._packet_length ,= struct.unpack(
+                    self.structFormat, data[:self.prefixLength])
+                if self._packet_length > self.MAX_LENGTH:
+                    self._clearBuffer()
+                    return self.lengthLimitExceeded(self._packet_length)
+                self._packet_length += self.prefixLength
+                self._buffer = [data]
+
+            if self._packet_length is not None:
+                if self._buffer_size < self._packet_length:
+                    return
+                data = ''.join(self._buffer)
+                packet = data[self.prefixLength:self._packet_length]
+                self._clearBuffer()
+                self._appendToBuffer(data[self._packet_length:])
+                self._packet_length = None
+                self.stringReceived(packet)
+
 
 
     def sendString(self, string):
@@ -717 +834 @@
         if len(string) >= 2 ** (8 * self.prefixLength):
             raise StringTooLongError(
                 "Try to send %s bytes whereas maximum is %s" % (
-                len(string), 2 ** (8 * self.prefixLength)))
-        self.transport.write(
-            struct.pack(self.structFormat, len(string)) + string)
+                len(data), 2 ** (8 * self.prefixLength)))
+        self.transport.writeSequence(
+            (struct.pack(self.structFormat, len(data)), data))
+
+    def _getRecvd(self):
+        warnings.warn(
+            "IntNStringReceiver.recvd deprecated since Twisted 8.3",
+            DeprecationWarning, 2)
+        if not self._buffer: # None or empty
+            return ''
+        elif len(self._buffer) > 1:
+            self._buffer = ''.join(self._buffer)
+        return self._buffer[0]
+
+    def _setRecvd(self, data):
+        warnings.warn("The recvd attribute of IntNStringReceiver is deprecated",
+                      DeprecationWarning)
+        self._buffer = [data]
+        self._buffer_size = len(data)
+
+    recvd = property(_getRecvd, _setRecvd)
 
 
 

twisted/test/test_protocols.py

@@ -195 +195 @@
               'len 20', 'foo 123', '0123456789\n012345678',
               'len 0', 'foo 5', '', '67890', 'len 1', 'a']
 
+    def _iotest(self, delimiter):
+        data = delimiter.join(self.output) + delimiter
+        for packetSize in range(1, 11):
+            transport = proto_helpers.StringTransport()
+            protocol = LineTester()
+            protocol.delimiter = delimiter
+            protocol.makeConnection(transport)
+            for i in range(len(data) / packetSize + 1):
+                bytes = self.buffer[i * packetSize:(i + 1) * packetSize]
+                protocol.dataReceived(bytes)
+            self.assertEqual(protocol.received, self.output)
+
+
     def testBuffer(self):
         """
         Test buffering for different packet size, checking received matches
@@ -210 +223 @@
             self.assertEqual(self.output, a.received)
 
 
+    def test_multibyteDelimiter(self):
+        """
+        L{LineReceiver.delimiter} may be multiple bytes.
+        """
+        self._iotest('\1\2\3')
+
+
     pause_buf = 'twiddle1\ntwiddle2\npause\ntwiddle3\n'
 
     pause_output1 = ['twiddle1', 'twiddle2', 'pause']
@@ -750 +770 @@
         self.assertEqual(r.received, [])
 
 
+    def _verifyRecvdDeprecation(self, warnings):
+        self.assertEqual(warnings[0]['category'], DeprecationWarning)
+        self.assertEqual(
+            warnings[0]['message'],
+            "IntNStringReceiver.recvd deprecated since Twisted 8.3")
+        self.assertEqual(len(warnings), 1)
+
+
+    def test_recvdInStringReceived(self):
+        """
+        During a call to C{stringReceived}, the protocol's C{recvd} attribute
+        is a C{str} containing any bytes already received for the next string.
+        """
+        buffered = []
+        protocol = self.getProtocol()
+        def invasiveStringReceived(string):
+            buffered.append(protocol.recvd)
+        protocol.stringReceived = invasiveStringReceived
+        protocol.dataReceived(
+            struct.pack(protocol.structFormat, 1) + 'x' + 'abc')
+        self.assertEqual(buffered, ['abc'])
+        warnings = self.flushWarnings(
+            offendingFunctions=[invasiveStringReceived])
+        self._verifyRecvdDeprecation(warnings)
+
+
+    def test_recvdBeforeDataReceived(self):
+        """
+        Before any calls to C{dataReceived}, the protocol's C{recvd} attribute
+        is C{""}.
+        """
+        protocol = self.getProtocol()
+        self.assertEqual(protocol.recvd, "")
+        warnings = self.flushWarnings(
+            offendingFunctions=[self.test_recvdBeforeDataReceived])
+        self._verifyRecvdDeprecation(warnings)
+
+
+    def test_recvdSetInStringReceived(self):
+        """
+        If C{recvd} is changed during a call to the protocol's
+        C{stringReceived}, the previous value is forgotten and the new value is
+        parsed instead.
+        """
+        received = []
+        protocol = self.getProtocol()
+        def invasiveStringReceived(string):
+            received.append(string)
+            protocol.recvd = nextValues.pop(0) + 'garbage'
+        protocol.stringReceived = invasiveStringReceived
+        nextValues = [
+            struct.pack(protocol.structFormat, 1) + '1',
+            struct.pack(protocol.structFormat, 3) + 'abc',
+            struct.pack(protocol.structFormat, 5) + 'hello',
+            '']
+        protocol.dataReceived(nextValues.pop(0) + 'garbage')
+        self.assertEqual(received, ['1', 'abc', 'hello'])
+        warnings = self.flushWarnings()
+        self.assertEqual(warnings[0]['category'], DeprecationWarning)
+        self.assertEqual(
+            warnings[0]['message'],
+            "Setting IntNStringReceiver.recvd deprecated since Twisted 8.3")
+
 
 class TestInt32(TestMixin, basic.Int32StringReceiver):
     """