1 # -*- Mode: Python -*-

     2 #   Id: asyncore.py,v 2.51 2000/09/07 22:29:26 rushing Exp

     3 #   Author: Sam Rushing <rushing@nightmare.com>

     4 

     5 # ======================================================================

     6 # Copyright 1996 by Sam Rushing

     7 #

     8 #                         All Rights Reserved

     9 #

    10 # Permission to use, copy, modify, and distribute this software and

    11 # its documentation for any purpose and without fee is hereby

    12 # granted, provided that the above copyright notice appear in all

    13 # copies and that both that copyright notice and this permission

    14 # notice appear in supporting documentation, and that the name of Sam

    15 # Rushing not be used in advertising or publicity pertaining to

    16 # distribution of the software without specific, written prior

    17 # permission.

    18 #

    19 # SAM RUSHING DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE,

    20 # INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS, IN

    21 # NO EVENT SHALL SAM RUSHING BE LIABLE FOR ANY SPECIAL, INDIRECT OR

    22 # CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS

    23 # OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT,

    24 # NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN

    25 # CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

    26 # ======================================================================

    27 

    28 """Basic infrastructure for asynchronous socket service clients and servers.

    29 

    30 There are only two ways to have a program on a single processor do "more

    31 than one thing at a time".  Multi-threaded programming is the simplest and

    32 most popular way to do it, but there is another very different technique,

    33 that lets you have nearly all the advantages of multi-threading, without

    34 actually using multiple threads. it's really only practical if your program

    35 is largely I/O bound. If your program is CPU bound, then pre-emptive

    36 scheduled threads are probably what you really need. Network servers are

    37 rarely CPU-bound, however.

    38 

    39 If your operating system supports the select() system call in its I/O

    40 library (and nearly all do), then you can use it to juggle multiple

    41 communication channels at once; doing other work while your I/O is taking

    42 place in the "background."  Although this strategy can seem strange and

    43 complex, especially at first, it is in many ways easier to understand and

    44 control than multi-threaded programming. The module documented here solves

    45 many of the difficult problems for you, making the task of building

    46 sophisticated high-performance network servers and clients a snap.

    47 """

    48 

    49 import select

    50 import socket

    51 import sys

    52 import time

    53 

    54 import os

    55 from errno import EALREADY, EINPROGRESS, EWOULDBLOCK, ECONNRESET, \

    56      ENOTCONN, ESHUTDOWN, EINTR, EISCONN, EBADF, ECONNABORTED, errorcode

    57 

    58 try:

    59     socket_map

    60 except NameError:

    61     socket_map = {}

    62 

    63 def _strerror(err):

    64     res = os.strerror(err)

    65     if res == 'Unknown error':

    66         res = errorcode[err]

    67     return res

    68 

    69 class ExitNow(Exception):

    70     pass

    71 

    72 def read(obj):

    73     try:

    74         obj.handle_read_event()

    75     except (ExitNow, KeyboardInterrupt, SystemExit):

    76         raise

    77     except:

    78         obj.handle_error()

    79 

    80 def write(obj):

    81     try:

    82         obj.handle_write_event()

    83     except (ExitNow, KeyboardInterrupt, SystemExit):

    84         raise

    85     except:

    86         obj.handle_error()

    87 

    88 def _exception(obj):

    89     try:

    90         obj.handle_expt_event()

    91     except (ExitNow, KeyboardInterrupt, SystemExit):

    92         raise

    93     except:

    94         obj.handle_error()

    95 

    96 def readwrite(obj, flags):

    97     try:

    98         if flags & (select.POLLIN | select.POLLPRI):

    99             obj.handle_read_event()

   100         if flags & select.POLLOUT:

   101             obj.handle_write_event()

   102         if flags & (select.POLLERR | select.POLLNVAL):

   103             obj.handle_expt_event()

   104         if flags & select.POLLHUP:

   105             obj.handle_close()

   106     except (ExitNow, KeyboardInterrupt, SystemExit):

   107         raise

   108     except:

   109         obj.handle_error()

   110 

   111 def poll(timeout=0.0, map=None):

   112     if map is None:

   113         map = socket_map

   114     if map:

   115         r = []; w = []; e = []

   116         for fd, obj in map.items():

   117             is_r = obj.readable()

   118             is_w = obj.writable()

   119             if is_r:

   120                 r.append(fd)

   121             if is_w:

   122                 w.append(fd)

   123             if is_r or is_w:

   124                 e.append(fd)

   125         if [] == r == w == e:

   126             time.sleep(timeout)

   127             return

   128 

   129         try:

   130             r, w, e = select.select(r, w, e, timeout)

   131         except select.error, err:

   132             if err.args[0] != EINTR:

   133                 raise

   134             else:

   135                 return

   136 

   137         for fd in r:

   138             obj = map.get(fd)

   139             if obj is None:

   140                 continue

   141             read(obj)

   142 

   143         for fd in w:

   144             obj = map.get(fd)

   145             if obj is None:

   146                 continue

   147             write(obj)

   148 

   149         for fd in e:

   150             obj = map.get(fd)

   151             if obj is None:

   152                 continue

   153             _exception(obj)

   154 

   155 def poll2(timeout=0.0, map=None):

   156     # Use the poll() support added to the select module in Python 2.0

   157     if map is None:

   158         map = socket_map

   159     if timeout is not None:

   160         # timeout is in milliseconds

   161         timeout = int(timeout*1000)

   162     pollster = select.poll()

   163     if map:

   164         for fd, obj in map.items():

   165             flags = 0

   166             if obj.readable():

   167                 flags |= select.POLLIN | select.POLLPRI

   168             if obj.writable():

   169                 flags |= select.POLLOUT

   170             if flags:

   171                 # Only check for exceptions if object was either readable

   172                 # or writable.

   173                 flags |= select.POLLERR | select.POLLHUP | select.POLLNVAL

   174                 pollster.register(fd, flags)

   175         try:

   176             r = pollster.poll(timeout)

   177         except select.error, err:

   178             if err.args[0] != EINTR:

   179                 raise

   180             r = []

   181         for fd, flags in r:

   182             obj = map.get(fd)

   183             if obj is None:

   184                 continue

   185             readwrite(obj, flags)

   186 

   187 poll3 = poll2                           # Alias for backward compatibility

   188 

   189 def loop(timeout=30.0, use_poll=False, map=None, count=None):

   190     if map is None:

   191         map = socket_map

   192 

   193     if use_poll and hasattr(select, 'poll'):

   194         poll_fun = poll2

   195     else:

   196         poll_fun = poll

   197 

   198     if count is None:

   199         while map:

   200             poll_fun(timeout, map)

   201 

   202     else:

   203         while map and count > 0:

   204             poll_fun(timeout, map)

   205             count = count - 1

   206 

   207 class dispatcher:

   208 

   209     debug = False

   210     connected = False

   211     accepting = False

   212     closing = False

   213     addr = None

   214 

   215     def __init__(self, sock=None, map=None):

   216         if map is None:

   217             self._map = socket_map

   218         else:

   219             self._map = map

   220 

   221         self._fileno = None

   222 

   223         if sock:

   224             # Set to nonblocking just to make sure for cases where we

   225             # get a socket from a blocking source.

   226             sock.setblocking(0)

   227             self.set_socket(sock, map)

   228             self.connected = True

   229             # The constructor no longer requires that the socket

   230             # passed be connected.

   231             try:

   232                 self.addr = sock.getpeername()

   233             except socket.error, err:

   234                 if err.args[0] == ENOTCONN:

   235                     # To handle the case where we got an unconnected

   236                     # socket.

   237                     self.connected = False

   238                 else:

   239                     # The socket is broken in some unknown way, alert

   240                     # the user and remove it from the map (to prevent

   241                     # polling of broken sockets).

   242                     self.del_channel(map)

   243                     raise

   244         else:

   245             self.socket = None

   246 

   247     def __repr__(self):

   248         status = [self.__class__.__module__+"."+self.__class__.__name__]

   249         if self.accepting and self.addr:

   250             status.append('listening')

   251         elif self.connected:

   252             status.append('connected')

   253         if self.addr is not None:

   254             try:

   255                 status.append('%s:%d' % self.addr)

   256             except TypeError:

   257                 status.append(repr(self.addr))

   258         return '<%s at %#x>' % (' '.join(status), id(self))

   259 

   260     def add_channel(self, map=None):

   261         #self.log_info('adding channel %s' % self)

   262         if map is None:

   263             map = self._map

   264         map[self._fileno] = self

   265 

   266     def del_channel(self, map=None):

   267         fd = self._fileno

   268         if map is None:

   269             map = self._map

   270         if fd in map:

   271             #self.log_info('closing channel %d:%s' % (fd, self))

   272             del map[fd]

   273         self._fileno = None

   274 

   275     def create_socket(self, family, type):

   276         self.family_and_type = family, type

   277         sock = socket.socket(family, type)

   278         sock.setblocking(0)

   279         self.set_socket(sock)

   280 

   281     def set_socket(self, sock, map=None):

   282         self.socket = sock

   283 ##        self.__dict__['socket'] = sock

   284         self._fileno = sock.fileno()

   285         self.add_channel(map)

   286 

   287     def set_reuse_addr(self):

   288         # try to re-use a server port if possible

   289         try:

   290             self.socket.setsockopt(

   291                 socket.SOL_SOCKET, socket.SO_REUSEADDR,

   292                 self.socket.getsockopt(socket.SOL_SOCKET,

   293                                        socket.SO_REUSEADDR) | 1

   294                 )

   295         except socket.error:

   296             pass

   297 

   298     # ==================================================

   299     # predicates for select()

   300     # these are used as filters for the lists of sockets

   301     # to pass to select().

   302     # ==================================================

   303 

   304     def readable(self):

   305         return True

   306 

   307     def writable(self):

   308         return True

   309 

   310     # ==================================================

   311     # socket object methods.

   312     # ==================================================

   313 

   314     def listen(self, num):

   315         self.accepting = True

   316         if os.name == 'nt' and num > 5:

   317             num = 5

   318         return self.socket.listen(num)

   319 

   320     def bind(self, addr):

   321         self.addr = addr

   322         return self.socket.bind(addr)

   323 

   324     def connect(self, address):

   325         self.connected = False

   326         err = self.socket.connect_ex(address)

   327         # XXX Should interpret Winsock return values

   328         if err in (EINPROGRESS, EALREADY, EWOULDBLOCK):

   329             return

   330         if err in (0, EISCONN):

   331             self.addr = address

   332             self.handle_connect_event()

   333         else:

   334             raise socket.error(err, errorcode[err])

   335 

   336     def accept(self):

   337         # XXX can return either an address pair or None

   338         try:

   339             conn, addr = self.socket.accept()

   340             return conn, addr

   341         except socket.error, why:

   342             if why.args[0] == EWOULDBLOCK:

   343                 pass

   344             else:

   345                 raise

   346 

   347     def send(self, data):

   348         try:

   349             result = self.socket.send(data)

   350             return result

   351         except socket.error, why:

   352             if why.args[0] == EWOULDBLOCK:

   353                 return 0

   354             elif why.args[0] in (ECONNRESET, ENOTCONN, ESHUTDOWN, ECONNABORTED):

   355                 self.handle_close()

   356                 return 0

   357             else:

   358                 raise

   359 

   360     def recv(self, buffer_size):

   361         try:

   362             data = self.socket.recv(buffer_size)

   363             if not data:

   364                 # a closed connection is indicated by signaling

   365                 # a read condition, and having recv() return 0.

   366                 self.handle_close()

   367                 return ''

   368             else:

   369                 return data

   370         except socket.error, why:

   371             # winsock sometimes throws ENOTCONN

   372             if why.args[0] in [ECONNRESET, ENOTCONN, ESHUTDOWN, ECONNABORTED]:

   373                 self.handle_close()

   374                 return ''

   375             else:

   376                 raise

   377 

   378     def close(self):

   379         self.connected = False

   380         self.accepting = False

   381         self.del_channel()

   382         try:

   383             self.socket.close()

   384         except socket.error, why:

   385             if why.args[0] not in (ENOTCONN, EBADF):

   386                 raise

   387 

   388     # cheap inheritance, used to pass all other attribute

   389     # references to the underlying socket object.

   390     def __getattr__(self, attr):

   391         return getattr(self.socket, attr)

   392 

   393     # log and log_info may be overridden to provide more sophisticated

   394     # logging and warning methods. In general, log is for 'hit' logging

   395     # and 'log_info' is for informational, warning and error logging.

   396 

   397     def log(self, message):

   398         sys.stderr.write('log: %s\n' % str(message))

   399 

   400     def log_info(self, message, type='info'):

   401         if __debug__ or type != 'info':

   402             print '%s: %s' % (type, message)

   403 

   404     def handle_read_event(self):

   405         if self.accepting:

   406             # accepting sockets are never connected, they "spawn" new

   407             # sockets that are connected

   408             self.handle_accept()

   409         elif not self.connected:

   410             self.handle_connect_event()

   411             self.handle_read()

   412         else:

   413             self.handle_read()

   414 

   415     def handle_connect_event(self):

   416         self.connected = True

   417         self.handle_connect()

   418 

   419     def handle_write_event(self):

   420         if self.accepting:

   421             # Accepting sockets shouldn't get a write event.

   422             # We will pretend it didn't happen.

   423             return

   424 

   425         if not self.connected:

   426             #check for errors

   427             err = self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)

   428             if err != 0:

   429                 raise socket.error(err, _strerror(err))

   430 

   431             self.handle_connect_event()

   432         self.handle_write()

   433 

   434     def handle_expt_event(self):

   435         # if the handle_expt is the same default worthless method,

   436         # we'll not even bother calling it, we'll instead generate

   437         # a useful error

   438         x = True

   439         try:

   440             y1 = self.__class__.handle_expt.im_func

   441             y2 = dispatcher.handle_expt.im_func

   442             x = y1 is y2

   443         except AttributeError:

   444             pass

   445 

   446         if x:

   447             err = self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)

   448             msg = _strerror(err)

   449 

   450             raise socket.error(err, msg)

   451         else:

   452             self.handle_expt()

   453 

   454     def handle_error(self):

   455         nil, t, v, tbinfo = compact_traceback()

   456 

   457         # sometimes a user repr method will crash.

   458         try:

   459             self_repr = repr(self)

   460         except:

   461             self_repr = '<__repr__(self) failed for object at %0x>' % id(self)

   462 

   463         self.log_info(

   464             'uncaptured python exception, closing channel %s (%s:%s %s)' % (

   465                 self_repr,

   466                 t,

   467                 v,

   468                 tbinfo

   469                 ),

   470             'error'

   471             )

   472         self.handle_close()

   473 

   474     def handle_expt(self):

   475         self.log_info('unhandled exception', 'warning')

   476 

   477     def handle_read(self):

   478         self.log_info('unhandled read event', 'warning')

   479 

   480     def handle_write(self):

   481         self.log_info('unhandled write event', 'warning')

   482 

   483     def handle_connect(self):

   484         self.log_info('unhandled connect event', 'warning')

   485 

   486     def handle_accept(self):

   487         self.log_info('unhandled accept event', 'warning')

   488 

   489     def handle_close(self):

   490         self.log_info('unhandled close event', 'warning')

   491         self.close()

   492 

   493 # ---------------------------------------------------------------------------

   494 # adds simple buffered output capability, useful for simple clients.

   495 # [for more sophisticated usage use asynchat.async_chat]

   496 # ---------------------------------------------------------------------------

   497 

   498 class dispatcher_with_send(dispatcher):

   499 

   500     def __init__(self, sock=None, map=None):

   501         dispatcher.__init__(self, sock, map)

   502         self.out_buffer = ''

   503 

   504     def initiate_send(self):

   505         num_sent = 0

   506         num_sent = dispatcher.send(self, self.out_buffer[:512])

   507         self.out_buffer = self.out_buffer[num_sent:]

   508 

   509     def handle_write(self):

   510         self.initiate_send()

   511 

   512     def writable(self):

   513         return (not self.connected) or len(self.out_buffer)

   514 

   515     def send(self, data):

   516         if self.debug:

   517             self.log_info('sending %s' % repr(data))

   518         self.out_buffer = self.out_buffer + data

   519         self.initiate_send()

   520 

   521 # ---------------------------------------------------------------------------

   522 # used for debugging.

   523 # ---------------------------------------------------------------------------

   524 

   525 def compact_traceback():

   526     t, v, tb = sys.exc_info()

   527     tbinfo = []

   528     if not tb: # Must have a traceback

   529         raise AssertionError("traceback does not exist")

   530     while tb:

   531         tbinfo.append((

   532             tb.tb_frame.f_code.co_filename,

   533             tb.tb_frame.f_code.co_name,

   534             str(tb.tb_lineno)

   535             ))

   536         tb = tb.tb_next

   537 

   538     # just to be safe

   539     del tb

   540 

   541     file, function, line = tbinfo[-1]

   542     info = ' '.join(['[%s|%s|%s]' % x for x in tbinfo])

   543     return (file, function, line), t, v, info

   544 

   545 def close_all(map=None, ignore_all=False):

   546     if map is None:

   547         map = socket_map

   548     for x in map.values():

   549         try:

   550             x.close()

   551         except OSError, x:

   552             if x.args[0] == EBADF:

   553                 pass

   554             elif not ignore_all:

   555                 raise

   556         except (ExitNow, KeyboardInterrupt, SystemExit):

   557             raise

   558         except:

   559             if not ignore_all:

   560                 raise

   561     map.clear()

   562 

   563 # Asynchronous File I/O:

   564 #

   565 # After a little research (reading man pages on various unixen, and

   566 # digging through the linux kernel), I've determined that select()

   567 # isn't meant for doing asynchronous file i/o.

   568 # Heartening, though - reading linux/mm/filemap.c shows that linux

   569 # supports asynchronous read-ahead.  So _MOST_ of the time, the data

   570 # will be sitting in memory for us already when we go to read it.

   571 #

   572 # What other OS's (besides NT) support async file i/o?  [VMS?]

   573 #

   574 # Regardless, this is useful for pipes, and stdin/stdout...

   575 

   576 if os.name == 'posix':

   577     import fcntl

   578 

   579     class file_wrapper:

   580         # Here we override just enough to make a file

   581         # look like a socket for the purposes of asyncore.

   582         # The passed fd is automatically os.dup()'d

   583 

   584         def __init__(self, fd):

   585             self.fd = os.dup(fd)

   586 

   587         def recv(self, *args):

   588             return os.read(self.fd, *args)

   589 

   590         def send(self, *args):

   591             return os.write(self.fd, *args)

   592 

   593         read = recv

   594         write = send

   595 

   596         def close(self):

   597             os.close(self.fd)

   598 

   599         def fileno(self):

   600             return self.fd

   601 

   602     class file_dispatcher(dispatcher):

   603 

   604         def __init__(self, fd, map=None):

   605             dispatcher.__init__(self, None, map)

   606             self.connected = True

   607             try:

   608                 fd = fd.fileno()

   609             except AttributeError:

   610                 pass

   611             self.set_file(fd)

   612             # set it to non-blocking mode

   613             flags = fcntl.fcntl(fd, fcntl.F_GETFL, 0)

   614             flags = flags | os.O_NONBLOCK

   615             fcntl.fcntl(fd, fcntl.F_SETFL, flags)

   616 

   617         def set_file(self, fd):

   618             self.socket = file_wrapper(fd)

   619             self._fileno = self.socket.fileno()

   620             self.add_channel()