[issue1736190] asyncore/asynchat patches

Lib/asyncore.py

 # -*- Mode: Python -*-
 #   Id: asyncore.py,v 2.51 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 35 matching lines @@
 sophisticated high-performance network servers and clients a snap.
 """
 
 import select
 import socket
 import sys
 import time
 
 import os
 from errno import EALREADY, EINPROGRESS, EWOULDBLOCK, ECONNRESET, \
-     ENOTCONN, ESHUTDOWN, EINTR, EISCONN, errorcode
+     ENOTCONN, ESHUTDOWN, EINTR, EISCONN, EBADF, ECONNABORTED, errorcode
 
 try:
     socket_map
 except NameError:
     socket_map = {}
+
+def _strerror(err):
+    res = os.strerror(err)
+    if res == 'Unknown error':
+        res = errorcode[err]
+    return res
 
 class ExitNow(Exception):
     pass
 
 def read(obj):
     try:
         obj.handle_read_event()
-    except ExitNow:
+    except (ExitNow, KeyboardInterrupt, SystemExit):
         raise
     except:
         obj.handle_error()
 
 def write(obj):
     try:
         obj.handle_write_event()
-    except ExitNow:
+    except (ExitNow, KeyboardInterrupt, SystemExit):
         raise
     except:
         obj.handle_error()
 
-def _exception (obj):
+def _exception(obj):
     try:
         obj.handle_expt_event()
-    except ExitNow:
+    except (ExitNow, KeyboardInterrupt, SystemExit):
         raise
     except:
         obj.handle_error()
 
 def readwrite(obj, flags):
     try:
         if flags & (select.POLLIN | select.POLLPRI):
             obj.handle_read_event()
         if flags & select.POLLOUT:
             obj.handle_write_event()
         if flags & (select.POLLERR | select.POLLHUP | select.POLLNVAL):
             obj.handle_expt_event()
-    except ExitNow:
+    except (ExitNow, KeyboardInterrupt, SystemExit):
         raise
     except:
         obj.handle_error()
 
 def poll(timeout=0.0, map=None):
     if map is None:
         map = socket_map
     if map:
         r = []; w = []; e = []
         for fd, obj in map.items():
             is_r = obj.readable()
             is_w = obj.writable()
             if is_r:
                 r.append(fd)
             if is_w:
                 w.append(fd)
             if is_r or is_w:
                 e.append(fd)
         if [] == r == w == e:
             time.sleep(timeout)
-        else:
-            try:
-                r, w, e = select.select(r, w, e, timeout)
-            except select.error, err:
-                if err[0] != EINTR:
-                    raise
-                else:
-                    return
+            return
+        
+        try:
+            r, w, e = select.select(r, w, e, timeout)
+        except select.error, err:
+            if err[0] != EINTR:
+                raise
+            else:
+                return
 
         for fd in r:
             obj = map.get(fd)
             if obj is None:
                 continue
             read(obj)
 
         for fd in w:
             obj = map.get(fd)
             if obj is None:
@@ skipping 64 matching lines @@
     connected = False
     accepting = False
     closing = False
     addr = None
 
     def __init__(self, sock=None, map=None):
         if map is None:
             self._map = socket_map
         else:
             self._map = map
-
+        
+        self._fileno = None
+        
         if sock:
+            # Set to nonblocking just to make sure for cases where we
+            # get a socket from a blocking source.
+            sock.setblocking(0)
             self.set_socket(sock, map)
-            # I think it should inherit this anyway
-            self.socket.setblocking(0)
             self.connected = True
-            # XXX Does the constructor require that the socket passed
-            # be connected?
+            # The constructor no longer requires that the socket
+            # passed be connected.
             try:
                 self.addr = sock.getpeername()
             except socket.error:
-                # The addr isn't crucial
-                pass
+                if err[0] == ENOTCONN:
+                    # To handle the case where we got an unconnected
+                    # socket.
+                    self.connected = False
+                else:
+                    # The socket is broken in some unknown way, alert
+                    # the user and remove it from the map (to prevent
+                    # polling of broken sockets).
+                    self.del_channel(map)
+                    raise
         else:
             self.socket = None
 
     def __repr__(self):
         status = [self.__class__.__module__+"."+self.__class__.__name__]
         if self.accepting and self.addr:
             status.append('listening')
         elif self.connected:
             status.append('connected')
         if self.addr is not None:
@@ skipping 13 matching lines @@
         fd = self._fileno
         if map is None:
             map = self._map
         if map.has_key(fd):
             #self.log_info('closing channel %d:%s' % (fd, self))
             del map[fd]
         self._fileno = None
 
     def create_socket(self, family, type):
         self.family_and_type = family, type
-        self.socket = socket.socket(family, type)
-        self.socket.setblocking(0)
-        self._fileno = self.socket.fileno()
-        self.add_channel()
+        sock = socket.socket(family, type)
+        sock.setblocking(0)
+        self.set_socket(sock)
 
     def set_socket(self, sock, map=None):
         self.socket = sock
 ##        self.__dict__['socket'] = sock
         self._fileno = sock.fileno()
         self.add_channel(map)
 
     def set_reuse_addr(self):
         # try to re-use a server port if possible
         try:
@@ skipping 17 matching lines @@
     def writable(self):
         return True
 
     # ==================================================
     # socket object methods.
     # ==================================================
 
     def listen(self, num):
         self.accepting = True
         if os.name == 'nt' and num > 5:
-            num = 1
+            num = 5
         return self.socket.listen(num)
 
     def bind(self, addr):
         self.addr = addr
         return self.socket.bind(addr)
 
     def connect(self, address):
         self.connected = False
         err = self.socket.connect_ex(address)
         # XXX Should interpret Winsock return values
         if err in (EINPROGRESS, EALREADY, EWOULDBLOCK):
             return
         if err in (0, EISCONN):
             self.addr = address
-            self.connected = True
-            self.handle_connect()
+            self.handle_connect_event()
         else:
-            raise socket.error, (err, errorcode[err])
+            raise socket.error(err, errorcode[err])
 
     def accept(self):
         # XXX can return either an address pair or None
         try:
             conn, addr = self.socket.accept()
             return conn, addr
         except socket.error, why:
             if why[0] == EWOULDBLOCK:
                 pass
             else:
                 raise
 
     def send(self, data):
         try:
             result = self.socket.send(data)
             return result
         except socket.error, why:
             if why[0] == EWOULDBLOCK:
                 return 0
+            elif why[0] in (ECONNRESET, ENOTCONN, ESHUTDOWN, ECONNABORTED):
+                self.handle_close()
+                return 0
             else:
                 raise
-            return 0
 
     def recv(self, buffer_size):
         try:
             data = self.socket.recv(buffer_size)
             if not data:
                 # a closed connection is indicated by signaling
                 # a read condition, and having recv() return 0.
                 self.handle_close()
                 return ''
             else:
                 return data
         except socket.error, why:
             # winsock sometimes throws ENOTCONN
-            if why[0] in [ECONNRESET, ENOTCONN, ESHUTDOWN]:
+            if why[0] in [ECONNRESET, ENOTCONN, ESHUTDOWN, ECONNABORTED]:
                 self.handle_close()
                 return ''
             else:
                 raise
 
     def close(self):
+        self.connected = False
+        self.accepting = False
         self.del_channel()
-        self.socket.close()
+        try:
+            self.socket.close()
+        except socket.error, why:
+            if why[0] not in (ENOTCONN, EBADF):
+                raise
 
     # cheap inheritance, used to pass all other attribute
     # references to the underlying socket object.
     def __getattr__(self, attr):
         return getattr(self.socket, attr)
 
     # log and log_info may be overridden to provide more sophisticated
     # logging and warning methods. In general, log is for 'hit' logging
     # and 'log_info' is for informational, warning and error logging.
 
     def log(self, message):
         sys.stderr.write('log: %s\n' % str(message))
 
     def log_info(self, message, type='info'):
         if __debug__ or type != 'info':
             print '%s: %s' % (type, message)
 
     def handle_read_event(self):
         if self.accepting:
-            # for an accepting socket, getting a read implies
-            # that we are connected
-            if not self.connected:
-                self.connected = True
+            # accepting sockets are never connected, they "spawn" new
+            # sockets that are connected
             self.handle_accept()
         elif not self.connected:
-            self.handle_connect()
-            self.connected = True
+            self.handle_connect_event()
             self.handle_read()
         else:
             self.handle_read()
 
+    def handle_connect_event(self):
+        self.connected = True
+        self.handle_connect()
+
     def handle_write_event(self):
-        # getting a write implies that we are connected
+        if self.accepting:
+            # Accepting sockets shouldn't get a write event.
+            # We will pretend it didn't happen.
+            return
+        
         if not self.connected:
-            self.handle_connect()
-            self.connected = True
+            #check for errors
+            err = self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
+            if err != 0:
+                raise socket.error(err, strerror(err))
+            
+            self.handle_connect_event()
         self.handle_write()
 
     def handle_expt_event(self):
-        self.handle_expt()
+        # if the handle_expt is the same default worthless method,
+        # we'll not even bother calling it, we'll instead generate
+        # a useful error
+        x = True
+        try:
+            y1 = self.__class__.handle_expt.im_func
+            y2 = dispatcher.handle_expt.im_func
+            x = y1 is y2
+        except AttributeError:
+            pass
+        
+        if x:
+            err = self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
+            msg = _strerror(err)
+            
+            raise socket.error(err, msg)
+        else:
+            self.handle_expt()
 
     def handle_error(self):
         nil, t, v, tbinfo = compact_traceback()
 
         # sometimes a user repr method will crash.
         try:
             self_repr = repr(self)
         except:
             self_repr = '<__repr__(self) failed for object at %0x>' % id(self)
 
@@ skipping 55 matching lines @@
         self.out_buffer = self.out_buffer + data
         self.initiate_send()
 
 # ---------------------------------------------------------------------------
 # used for debugging.
 # ---------------------------------------------------------------------------
 
 def compact_traceback():
     t, v, tb = sys.exc_info()
     tbinfo = []
-    assert tb # Must have a traceback
+    if not tb: # Must have a traceback
+        raise AssertionError("traceback does not exist")
     while tb:
         tbinfo.append((
             tb.tb_frame.f_code.co_filename,
             tb.tb_frame.f_code.co_name,
             str(tb.tb_lineno)
             ))
         tb = tb.tb_next
 
     # just to be safe
     del tb
 
     file, function, line = tbinfo[-1]
     info = ' '.join(['[%s|%s|%s]' % x for x in tbinfo])
     return (file, function, line), t, v, info
 
-def close_all(map=None):
+def close_all(map=None, ignore_all=False):
     if map is None:
         map = socket_map
     for x in map.values():
-        x.socket.close()
+        try:
+            x.close()
+        except OSError, x:
+            if x[0] == EBADF:
+                pass
+            elif not ignore_all:
+                raise
+        except (ExitNow, KeyboardInterrupt, SystemExit):
+            raise
+        except:
+            if not ignore_all:
+                raise
     map.clear()
 
 # Asynchronous File I/O:
 #
 # After a little research (reading man pages on various unixen, and
 # digging through the linux kernel), I've determined that select()
 # isn't meant for doing asynchronous file i/o.
 # Heartening, though - reading linux/mm/filemap.c shows that linux
 # supports asynchronous read-ahead.  So _MOST_ of the time, the data
 # will be sitting in memory for us already when we go to read it.
 #
 # What other OS's (besides NT) support async file i/o?  [VMS?]
 #
 # Regardless, this is useful for pipes, and stdin/stdout...
 
 if os.name == 'posix':
     import fcntl
 
     class file_wrapper:
-        # here we override just enough to make a file
+        # Here we override just enough to make a file
         # look like a socket for the purposes of asyncore.
+        # The passed fd is automatically os.dup()'d
 
         def __init__(self, fd):
-            self.fd = fd
+            self.fd = os.dup(fd)
 
         def recv(self, *args):
             return os.read(self.fd, *args)
 
         def send(self, *args):
             return os.write(self.fd, *args)
 
         read = recv
         write = send
 
         def close(self):
             os.close(self.fd)
 
         def fileno(self):
             return self.fd
 
     class file_dispatcher(dispatcher):
 
         def __init__(self, fd, map=None):
             dispatcher.__init__(self, None, map)
             self.connected = True
+            try:
+                fd = fd.fileno()
+            except AttributeError:
+                pass
             self.set_file(fd)
             # set it to non-blocking mode
             flags = fcntl.fcntl(fd, fcntl.F_GETFL, 0)
             flags = flags | os.O_NONBLOCK
             fcntl.fcntl(fd, fcntl.F_SETFL, flags)
 
         def set_file(self, fd):
             self._fileno = fd
             self.socket = file_wrapper(fd)
             self.add_channel()