Lib/asyncore.py

   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 exceptions
  50 import select
  51 import socket
  52 import sys
  53 import types
  54
  55 import os
  56 if os.name == 'nt':
  57     EWOULDBLOCK = 10035
  58     EINPROGRESS = 10036
  59     EALREADY    = 10037
  60     ECONNRESET  = 10054
  61     ENOTCONN    = 10057
  62     ESHUTDOWN   = 10058
  63 else:
  64     from errno import EALREADY, EINPROGRESS, EWOULDBLOCK, ECONNRESET, ENOTCONN, ESHUTDOWN
  65
  66 try:
  67     socket_map
  68 except NameError:
  69     socket_map = {}
  70
  71 class ExitNow (exceptions.Exception):
  72     pass
  73
  74 DEBUG = 0
  75
  76 def poll (timeout=0.0, map=None):
  77     global DEBUG
  78     if map is None:
  79         map = socket_map
  80     if map:
  81         r = []; w = []; e = []
  82         for fd, obj in map.items():
  83             if obj.readable():
  84                 r.append (fd)
  85             if obj.writable():
  86                 w.append (fd)
  87         r,w,e = select.select (r,w,e, timeout)
  88
  89         if DEBUG:
  90             print r,w,e
  91
  92         for fd in r:
  93             try:
  94                 obj = map[fd]
  95                 try:
  96                     obj.handle_read_event()
  97                 except ExitNow:
  98                     raise ExitNow
  99                 except:
 100                     obj.handle_error()
 101             except KeyError:
 102                 pass
 103
 104         for fd in w:
 105             try:
 106                 obj = map[fd]
 107                 try:
 108                     obj.handle_write_event()
 109                 except ExitNow:
 110                     raise ExitNow
 111                 except:
 112                     obj.handle_error()
 113             except KeyError:
 114                 pass
 115
 116 def poll2 (timeout=0.0, map=None):
 117     import poll
 118     if map is None:
 119         map=socket_map
 120     # timeout is in milliseconds
 121     timeout = int(timeout*1000)
 122     if map:
 123         l = []
 124         for fd, obj in map.items():
 125             flags = 0
 126             if obj.readable():
 127                 flags = poll.POLLIN
 128             if obj.writable():
 129                 flags = flags | poll.POLLOUT
 130             if flags:
 131                 l.append ((fd, flags))
 132         r = poll.poll (l, timeout)
 133         for fd, flags in r:
 134             try:
 135                 obj = map[fd]
 136                 try:
 137                     if (flags  & poll.POLLIN):
 138                         obj.handle_read_event()
 139                     if (flags & poll.POLLOUT):
 140                         obj.handle_write_event()
 141                 except ExitNow:
 142                     raise ExitNow
 143                 except:
 144                     obj.handle_error()
 145             except KeyError:
 146                 pass
 147
 148 def poll3 (timeout=0.0, map=None):
 149     # Use the poll() support added to the select module in Python 2.0
 150     if map is None:
 151         map=socket_map
 152     # timeout is in milliseconds
 153     timeout = int(timeout*1000)
 154     pollster = select.poll()
 155     if map:
 156         l = []
 157         for fd, obj in map.items():
 158             flags = 0
 159             if obj.readable():
 160                 flags = select.POLLIN
 161             if obj.writable():
 162                 flags = flags | select.POLLOUT
 163             if flags:
 164                 pollster.register(fd, flags)
 165         r = pollster.poll (timeout)
 166         for fd, flags in r:
 167             try:
 168                 obj = map[fd]
 169                 try:
 170                     if (flags  & select.POLLIN):
 171                         obj.handle_read_event()
 172                     if (flags & select.POLLOUT):
 173                         obj.handle_write_event()
 174                 except ExitNow:
 175                     raise ExitNow
 176                 except:
 177                     obj.handle_error()
 178             except KeyError:
 179                 pass
 180
 181 def loop (timeout=30.0, use_poll=0, map=None):
 182
 183     if map is None:
 184         map=socket_map
 185
 186     if use_poll:
 187         if hasattr (select, 'poll'):
 188             poll_fun = poll3
 189         else:
 190             poll_fun = poll2
 191     else:
 192         poll_fun = poll
 193
 194     while map:
 195         poll_fun (timeout, map)
 196
 197 class dispatcher:
 198     debug = 0
 199     connected = 0
 200     accepting = 0
 201     closing = 0
 202     addr = None
 203
 204     def __init__ (self, sock=None, map=None):
 205         if sock:
 206             self.set_socket (sock, map)
 207             # I think it should inherit this anyway
 208             self.socket.setblocking (0)
 209             self.connected = 1
 210
 211     def __repr__ (self):
 212         try:
 213             status = []
 214             if self.accepting and self.addr:
 215                 status.append ('listening')
 216             elif self.connected:
 217                 status.append ('connected')
 218             if self.addr:
 219                 if self.addr == types.TupleType:
 220                     status.append ('%s:%d' % self.addr)
 221                 else:
 222                     status.append (self.addr)
 223             return '<%s %s at %x>' % (self.__class__.__name__,
 224                                       ' '.join (status), id (self))
 225         except:
 226             pass
 227
 228         try:
 229             ar = repr (self.addr)
 230         except AttributeError:
 231             ar = 'no self.addr!'
 232
 233         return '<__repr__() failed for %s instance at %x (addr=%s)>' % \
 234                (self.__class__.__name__, id (self), ar)
 235
 236     def add_channel (self, map=None):
 237         #self.log_info ('adding channel %s' % self)
 238         if map is None:
 239             map=socket_map
 240         map [self._fileno] = self
 241
 242     def del_channel (self, map=None):
 243         fd = self._fileno
 244         if map is None:
 245             map=socket_map
 246         if map.has_key (fd):
 247             #self.log_info ('closing channel %d:%s' % (fd, self))
 248             del map [fd]
 249
 250     def create_socket (self, family, type):
 251         self.family_and_type = family, type
 252         self.socket = socket.socket (family, type)
 253         self.socket.setblocking(0)
 254         self._fileno = self.socket.fileno()
 255         self.add_channel()
 256
 257     def set_socket (self, sock, map=None):
 258         self.__dict__['socket'] = sock
 259         self._fileno = sock.fileno()
 260         self.add_channel (map)
 261
 262     def set_reuse_addr (self):
 263         # try to re-use a server port if possible
 264         try:
 265             self.socket.setsockopt (
 266                 socket.SOL_SOCKET, socket.SO_REUSEADDR,
 267                 self.socket.getsockopt (socket.SOL_SOCKET, socket.SO_REUSEADDR) | 1
 268                 )
 269         except socket.error:
 270             pass
 271
 272     # ==================================================
 273     # predicates for select()
 274     # these are used as filters for the lists of sockets
 275     # to pass to select().
 276     # ==================================================
 277
 278     def readable (self):
 279         return 1
 280
 281     if os.name == 'mac':
 282         # The macintosh will select a listening socket for
 283         # write if you let it.  What might this mean?
 284         def writable (self):
 285             return not self.accepting
 286     else:
 287         def writable (self):
 288             return 1
 289
 290     # ==================================================
 291     # socket object methods.
 292     # ==================================================
 293
 294     def listen (self, num):
 295         self.accepting = 1
 296         if os.name == 'nt' and num > 5:
 297             num = 1
 298         return self.socket.listen (num)
 299
 300     def bind (self, addr):
 301         self.addr = addr
 302         return self.socket.bind (addr)
 303
 304     def connect (self, address):
 305         self.connected = 0
 306         # XXX why not use connect_ex?
 307         try:
 308             self.socket.connect (address)
 309         except socket.error, why:
 310             if why[0] in (EINPROGRESS, EALREADY, EWOULDBLOCK):
 311                 return
 312             else:
 313                 raise socket.error, why
 314         self.connected = 1
 315         self.handle_connect()
 316
 317     def accept (self):
 318         try:
 319             conn, addr = self.socket.accept()
 320             return conn, addr
 321         except socket.error, why:
 322             if why[0] == EWOULDBLOCK:
 323                 pass
 324             else:
 325                 raise socket.error, why
 326
 327     def send (self, data):
 328         try:
 329             result = self.socket.send (data)
 330             return result
 331         except socket.error, why:
 332             if why[0] == EWOULDBLOCK:
 333                 return 0
 334             else:
 335                 raise socket.error, why
 336             return 0
 337
 338     def recv (self, buffer_size):
 339         try:
 340             data = self.socket.recv (buffer_size)
 341             if not data:
 342                 # a closed connection is indicated by signaling
 343                 # a read condition, and having recv() return 0.
 344                 self.handle_close()
 345                 return ''
 346             else:
 347                 return data
 348         except socket.error, why:
 349             # winsock sometimes throws ENOTCONN
 350             if why[0] in [ECONNRESET, ENOTCONN, ESHUTDOWN]:
 351                 self.handle_close()
 352                 return ''
 353             else:
 354                 raise socket.error, why
 355
 356     def close (self):
 357         self.del_channel()
 358         self.socket.close()
 359
 360     # cheap inheritance, used to pass all other attribute
 361     # references to the underlying socket object.
 362     def __getattr__ (self, attr):
 363         return getattr (self.socket, attr)
 364
 365     # log and log_info maybe overriden to provide more sophisitcated
 366     # logging and warning methods. In general, log is for 'hit' logging
 367     # and 'log_info' is for informational, warning and error logging.
 368
 369     def log (self, message):
 370         sys.stderr.write ('log: %s\n' % str(message))
 371
 372     def log_info (self, message, type='info'):
 373         if __debug__ or type != 'info':
 374             print '%s: %s' % (type, message)
 375
 376     def handle_read_event (self):
 377         if self.accepting:
 378             # for an accepting socket, getting a read implies
 379             # that we are connected
 380             if not self.connected:
 381                 self.connected = 1
 382             self.handle_accept()
 383         elif not self.connected:
 384             self.handle_connect()
 385             self.connected = 1
 386             self.handle_read()
 387         else:
 388             self.handle_read()
 389
 390     def handle_write_event (self):
 391         # getting a write implies that we are connected
 392         if not self.connected:
 393             self.handle_connect()
 394             self.connected = 1
 395         self.handle_write()
 396
 397     def handle_expt_event (self):
 398         self.handle_expt()
 399
 400     def handle_error (self):
 401         (file,fun,line), t, v, tbinfo = compact_traceback()
 402
 403         # sometimes a user repr method will crash.
 404         try:
 405             self_repr = repr (self)
 406         except:
 407             self_repr = '<__repr__ (self) failed for object at %0x>' % id(self)
 408
 409         self.log_info (
 410             'uncaptured python exception, closing channel %s (%s:%s %s)' % (
 411                 self_repr,
 412                 t,
 413                 v,
 414                 tbinfo
 415                 ),
 416             'error'
 417             )
 418         self.close()
 419
 420     def handle_expt (self):
 421         self.log_info ('unhandled exception', 'warning')
 422
 423     def handle_read (self):
 424         self.log_info ('unhandled read event', 'warning')
 425
 426     def handle_write (self):
 427         self.log_info ('unhandled write event', 'warning')
 428
 429     def handle_connect (self):
 430         self.log_info ('unhandled connect event', 'warning')
 431
 432     def handle_accept (self):
 433         self.log_info ('unhandled accept event', 'warning')
 434
 435     def handle_close (self):
 436         self.log_info ('unhandled close event', 'warning')
 437         self.close()
 438
 439 # ---------------------------------------------------------------------------
 440 # adds simple buffered output capability, useful for simple clients.
 441 # [for more sophisticated usage use asynchat.async_chat]
 442 # ---------------------------------------------------------------------------
 443
 444 class dispatcher_with_send (dispatcher):
 445     def __init__ (self, sock=None):
 446         dispatcher.__init__ (self, sock)
 447         self.out_buffer = ''
 448
 449     def initiate_send (self):
 450         num_sent = 0
 451         num_sent = dispatcher.send (self, self.out_buffer[:512])
 452         self.out_buffer = self.out_buffer[num_sent:]
 453
 454     def handle_write (self):
 455         self.initiate_send()
 456
 457     def writable (self):
 458         return (not self.connected) or len(self.out_buffer)
 459
 460     def send (self, data):
 461         if self.debug:
 462             self.log_info ('sending %s' % repr(data))
 463         self.out_buffer = self.out_buffer + data
 464         self.initiate_send()
 465
 466 # ---------------------------------------------------------------------------
 467 # used for debugging.
 468 # ---------------------------------------------------------------------------
 469
 470 def compact_traceback ():
 471     t,v,tb = sys.exc_info()
 472     tbinfo = []
 473     while 1:
 474         tbinfo.append ((
 475             tb.tb_frame.f_code.co_filename,
 476             tb.tb_frame.f_code.co_name,
 477             str(tb.tb_lineno)
 478             ))
 479         tb = tb.tb_next
 480         if not tb:
 481             break
 482
 483     # just to be safe
 484     del tb
 485
 486     file, function, line = tbinfo[-1]
 487     info = '[' + '] ['.join(map(lambda x: '|'.join(x), tbinfo)) + ']'
 488     return (file, function, line), t, v, info
 489
 490 def close_all (map=None):
 491     if map is None:
 492         map=socket_map
 493     for x in map.values():
 494         x.socket.close()
 495     map.clear()
 496
 497 # Asynchronous File I/O:
 498 #
 499 # After a little research (reading man pages on various unixen, and
 500 # digging through the linux kernel), I've determined that select()
 501 # isn't meant for doing doing asynchronous file i/o.
 502 # Heartening, though - reading linux/mm/filemap.c shows that linux
 503 # supports asynchronous read-ahead.  So _MOST_ of the time, the data
 504 # will be sitting in memory for us already when we go to read it.
 505 #
 506 # What other OS's (besides NT) support async file i/o?  [VMS?]
 507 #
 508 # Regardless, this is useful for pipes, and stdin/stdout...
 509
 510 import os
 511 if os.name == 'posix':
 512     import fcntl
 513
 514     class file_wrapper:
 515         # here we override just enough to make a file
 516         # look like a socket for the purposes of asyncore.
 517         def __init__ (self, fd):
 518             self.fd = fd
 519
 520         def recv (self, *args):
 521             return apply (os.read, (self.fd,)+args)
 522
 523         def send (self, *args):
 524             return apply (os.write, (self.fd,)+args)
 525
 526         read = recv
 527         write = send
 528
 529         def close (self):
 530             return os.close (self.fd)
 531
 532         def fileno (self):
 533             return self.fd
 534
 535     class file_dispatcher (dispatcher):
 536         def __init__ (self, fd):
 537             dispatcher.__init__ (self)
 538             self.connected = 1
 539             # set it to non-blocking mode
 540             flags = fcntl.fcntl (fd, fcntl.F_GETFL, 0)
 541             flags = flags | os.O_NONBLOCK
 542             fcntl.fcntl (fd, fcntl.F_SETFL, flags)
 543             self.set_file (fd)
 544
 545         def set_file (self, fd):
 546             self._fileno = fd
 547             self.socket = file_wrapper (fd)
 548             self.add_channel()