[issue1736190] asyncore/asynchat patches

Lib/asynchat.py

 # -*- Mode: Python; tab-width: 4 -*-
 #       Id: asynchat.py,v 2.26 2000/09/07 22:29:26 rushing Exp
 #       Author: Sam Rushing <rushing@nightmare.com>
 
 # ======================================================================
 # Copyright 1996 by Sam Rushing
 #
 #                         All Rights Reserved
 #
 # Permission to use, copy, modify, and distribute this software and
@@ skipping 42 matching lines @@
 class async_chat (asyncore.dispatcher):
     """This is an abstract class.  You must derive from this class, and add
     the two methods collect_incoming_data() and found_terminator()"""
 
     # these are overridable defaults
 
     ac_in_buffer_size       = 4096
     ac_out_buffer_size      = 4096
 
     def __init__ (self, conn=None):
+        # for string terminator matching
         self.ac_in_buffer = ''
-        self.ac_out_buffer = ''
-        self.producer_fifo = fifo()
+        
+        # we use a list here rather than cStringIO for a few reasons...
+        # del lst[:] is faster than sio.truncate(0)
+        # lst = [] is faster than sio.truncate(0)
+        # cStringIO will be gaining unicode support in py3k, which
+        # will negatively affect the performance of bytes compared to
+        # a ''.join() equivalent
+        self.incoming = []
+        
+        # we toss the use of the "simple producer" and replace it with
+        # a pure deque, which the original fifo was a wrapping of
+        self.producer_fifo = deque()
         asyncore.dispatcher.__init__ (self, conn)
 
     def collect_incoming_data(self, data):
-        raise NotImplementedError, "must be implemented in subclass"
+        raise NotImplementedError("must be implemented in subclass")
+    
+    def _collect_incoming_data(self, data):
+        self.incoming.append(data)
+    
+    def _get_data(self):
+        d = ''.join(self.incoming)
+        del self.incoming[:]
+        return d
 
     def found_terminator(self):
-        raise NotImplementedError, "must be implemented in subclass"
+        raise NotImplementedError("must be implemented in subclass")
 
     def set_terminator (self, term):
         "Set the input delimiter.  Can be a fixed string of any length, an integer, or None"
         self.terminator = term
 
     def get_terminator (self):
         return self.terminator
 
     # grab some more data from the socket,
     # throw it to the collector method,
     # check for the terminator,
     # if found, transition to the next state.
 
     def handle_read (self):
 
         try:
             data = self.recv (self.ac_in_buffer_size)
         except socket.error, why:
             self.handle_error()
             return
 
         self.ac_in_buffer = self.ac_in_buffer + data
 
         # Continue to search for self.terminator in self.ac_in_buffer,
         # while calling self.collect_incoming_data.  The while loop
         # is necessary because we might read several data+terminator
-        # combos with a single recv(1024).
+        # combos with a single recv(4096).
 
         while self.ac_in_buffer:
             lb = len(self.ac_in_buffer)
             terminator = self.get_terminator()
             if not terminator:
                 # no terminator, collect it all
                 self.collect_incoming_data (self.ac_in_buffer)
                 self.ac_in_buffer = ''
             elif isinstance(terminator, int) or isinstance(terminator, long):
                 # numeric terminator
@@ skipping 33 matching lines @@
                             # we found a prefix, collect up to the prefix
                             self.collect_incoming_data (self.ac_in_buffer[:-index])
                             self.ac_in_buffer = self.ac_in_buffer[-index:]
                         break
                     else:
                         # no prefix, collect it all
                         self.collect_incoming_data (self.ac_in_buffer)
                         self.ac_in_buffer = ''
 
     def handle_write (self):
-        self.initiate_send ()
+        self.initiate_send()
 
     def handle_close (self):
         self.close()
 
     def push (self, data):
-        self.producer_fifo.push (simple_producer (data))
+        sabs = self.ac_out_buffer_size
+        if len(data) > sabs:
+            for i in xrange(0, len(data), sabs):
+                self.producer_fifo.append(data[i:i+sabs])
+        else:
+            self.producer_fifo.append(data)
         self.initiate_send()
-
+    
     def push_with_producer (self, producer):
-        self.producer_fifo.push (producer)
+        self.producer_fifo.append(producer)
         self.initiate_send()
-
+    
     def readable (self):
         "predicate for inclusion in the readable for select()"
-        return (len(self.ac_in_buffer) <= self.ac_in_buffer_size)
+        # cannot use the old predicate, it violates the claim of the
+        # set_terminator method.
+        
+        # return (len(self.ac_in_buffer) <= self.ac_in_buffer_size)
+        return 1 
 
     def writable (self):
         "predicate for inclusion in the writable for select()"
-        # return len(self.ac_out_buffer) or len(self.producer_fifo) or (not self.connected)
-        # this is about twice as fast, though not as clear.
-        return not (
-                (self.ac_out_buffer == '') and
-                self.producer_fifo.is_empty() and
-                self.connected
-                )
+        return self.producer_fifo or (not self.connected)
 
     def close_when_done (self):
         "automatically close this channel once the outgoing queue is empty"
-        self.producer_fifo.push (None)
+        self.producer_fifo.append(None)
 
-    # refill the outgoing buffer by calling the more() method
-    # of the first producer in the queue
-    def refill_buffer (self):
-        while 1:
-            if len(self.producer_fifo):
-                p = self.producer_fifo.first()
-                # a 'None' in the producer fifo is a sentinel,
-                # telling us to close the channel.
-                if p is None:
-                    if not self.ac_out_buffer:
-                        self.producer_fifo.pop()
-                        self.close()
+    def initiate_send(self):
+        while self.producer_fifo and self.connected:
+            first = self.producer_fifo[0]
+            # handle empty string/buffer or None entry
+            if not first:
+                del self.producer_fifo[0]
+                if first is None:
+                    self.handle_close()
                     return
-                elif isinstance(p, str):
-                    self.producer_fifo.pop()
-                    self.ac_out_buffer = self.ac_out_buffer + p
-                    return
-                data = p.more()
+
+            # handle classic producer behavior
+            obs = self.ac_out_buffer_size
+            try:
+                data = buffer(first, 0, obs)
+            except TypeError:
+                data = first.more()
                 if data:
-                    self.ac_out_buffer = self.ac_out_buffer + data
-                    return
+                    self.producer_fifo.appendleft(data)
                 else:
-                    self.producer_fifo.pop()
-            else:
+                    del self.producer_fifo[0]
+                continue
+
+            # send the data
+            try:
+                num_sent = self.send(data)
+            except socket.error:
+                self.handle_error()
                 return
 
-    def initiate_send (self):
-        obs = self.ac_out_buffer_size
-        # try to refill the buffer
-        if (len (self.ac_out_buffer) < obs):
-            self.refill_buffer()
-
-        if self.ac_out_buffer and self.connected:
-            # try to send the buffer
-            try:
-                num_sent = self.send (self.ac_out_buffer[:obs])
-                if num_sent:
-                    self.ac_out_buffer = self.ac_out_buffer[num_sent:]
-
-            except socket.error, why:
-                self.handle_error()
-                return
+            if num_sent:
+                if num_sent < len(data) or obs < len(first):
+                    self.producer_fifo[0] = first[num_sent:]
+                else:
+                    del self.producer_fifo[0]
+            # we tried to send some actual data
+            return
 
     def discard_buffers (self):
         # Emergencies only!
         self.ac_in_buffer = ''
-        self.ac_out_buffer = ''
-        while self.producer_fifo:
-            self.producer_fifo.pop()
-
+        del self.incoming[:]
+        self.producer_fifo.clear()
 
 class simple_producer:
 
     def __init__ (self, data, buffer_size=512):
         self.data = data
         self.buffer_size = buffer_size
 
     def more (self):
         if len (self.data) > self.buffer_size:
             result = self.data[:self.buffer_size]
@@ skipping 42 matching lines @@
 # new python:   28961/s
 # old python:   18307/s
 # re:        12820/s
 # regex:     14035/s
 
 def find_prefix_at_end (haystack, needle):
     l = len(needle) - 1
     while l and not haystack.endswith(needle[:l]):
         l -= 1
     return l