AppPkg/Applications/Python/Python-2.7.2/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 select
  50 import socket
  51 import sys
  52 import time
  53 import warnings
  54
  55 import os
  56 from errno import EALREADY, EINPROGRESS, EWOULDBLOCK, ECONNRESET, EINVAL, \
  57      ENOTCONN, ESHUTDOWN, EINTR, EISCONN, EBADF, ECONNABORTED, EPIPE, EAGAIN, \
  58      errorcode
  59
  60 _DISCONNECTED = frozenset((ECONNRESET, ENOTCONN, ESHUTDOWN, ECONNABORTED, EPIPE,
  61                            EBADF))
  62
  63 try:
  64     socket_map
  65 except NameError:
  66     socket_map = {}
  67
  68 def _strerror(err):
  69     try:
  70         return os.strerror(err)
  71     except (ValueError, OverflowError, NameError):
  72         if err in errorcode:
  73             return errorcode[err]
  74         return "Unknown error %s" %err
  75
  76 class ExitNow(Exception):
  77     pass
  78
  79 _reraised_exceptions = (ExitNow, KeyboardInterrupt, SystemExit)
  80
  81 def read(obj):
  82     try:
  83         obj.handle_read_event()
  84     except _reraised_exceptions:
  85         raise
  86     except:
  87         obj.handle_error()
  88
  89 def write(obj):
  90     try:
  91         obj.handle_write_event()
  92     except _reraised_exceptions:
  93         raise
  94     except:
  95         obj.handle_error()
  96
  97 def _exception(obj):
  98     try:
  99         obj.handle_expt_event()
 100     except _reraised_exceptions:
 101         raise
 102     except:
 103         obj.handle_error()
 104
 105 def readwrite(obj, flags):
 106     try:
 107         if flags & select.POLLIN:
 108             obj.handle_read_event()
 109         if flags & select.POLLOUT:
 110             obj.handle_write_event()
 111         if flags & select.POLLPRI:
 112             obj.handle_expt_event()
 113         if flags & (select.POLLHUP | select.POLLERR | select.POLLNVAL):
 114             obj.handle_close()
 115     except socket.error, e:
 116         if e.args[0] not in _DISCONNECTED:
 117             obj.handle_error()
 118         else:
 119             obj.handle_close()
 120     except _reraised_exceptions:
 121         raise
 122     except:
 123         obj.handle_error()
 124
 125 def poll(timeout=0.0, map=None):
 126     if map is None:
 127         map = socket_map
 128     if map:
 129         r = []; w = []; e = []
 130         for fd, obj in map.items():
 131             is_r = obj.readable()
 132             is_w = obj.writable()
 133             if is_r:
 134                 r.append(fd)
 135             if is_w:
 136                 w.append(fd)
 137             if is_r or is_w:
 138                 e.append(fd)
 139         if [] == r == w == e:
 140             time.sleep(timeout)
 141             return
 142
 143         try:
 144             r, w, e = select.select(r, w, e, timeout)
 145         except select.error, err:
 146             if err.args[0] != EINTR:
 147                 raise
 148             else:
 149                 return
 150
 151         for fd in r:
 152             obj = map.get(fd)
 153             if obj is None:
 154                 continue
 155             read(obj)
 156
 157         for fd in w:
 158             obj = map.get(fd)
 159             if obj is None:
 160                 continue
 161             write(obj)
 162
 163         for fd in e:
 164             obj = map.get(fd)
 165             if obj is None:
 166                 continue
 167             _exception(obj)
 168
 169 def poll2(timeout=0.0, map=None):
 170     # Use the poll() support added to the select module in Python 2.0
 171     if map is None:
 172         map = socket_map
 173     if timeout is not None:
 174         # timeout is in milliseconds
 175         timeout = int(timeout*1000)
 176     pollster = select.poll()
 177     if map:
 178         for fd, obj in map.items():
 179             flags = 0
 180             if obj.readable():
 181                 flags |= select.POLLIN | select.POLLPRI
 182             if obj.writable():
 183                 flags |= select.POLLOUT
 184             if flags:
 185                 # Only check for exceptions if object was either readable
 186                 # or writable.
 187                 flags |= select.POLLERR | select.POLLHUP | select.POLLNVAL
 188                 pollster.register(fd, flags)
 189         try:
 190             r = pollster.poll(timeout)
 191         except select.error, err:
 192             if err.args[0] != EINTR:
 193                 raise
 194             r = []
 195         for fd, flags in r:
 196             obj = map.get(fd)
 197             if obj is None:
 198                 continue
 199             readwrite(obj, flags)
 200
 201 poll3 = poll2                           # Alias for backward compatibility
 202
 203 def loop(timeout=30.0, use_poll=False, map=None, count=None):
 204     if map is None:
 205         map = socket_map
 206
 207     if use_poll and hasattr(select, 'poll'):
 208         poll_fun = poll2
 209     else:
 210         poll_fun = poll
 211
 212     if count is None:
 213         while map:
 214             poll_fun(timeout, map)
 215
 216     else:
 217         while map and count > 0:
 218             poll_fun(timeout, map)
 219             count = count - 1
 220
 221 class dispatcher:
 222
 223     debug = False
 224     connected = False
 225     accepting = False
 226     closing = False
 227     addr = None
 228     ignore_log_types = frozenset(['warning'])
 229
 230     def __init__(self, sock=None, map=None):
 231         if map is None:
 232             self._map = socket_map
 233         else:
 234             self._map = map
 235
 236         self._fileno = None
 237
 238         if sock:
 239             # Set to nonblocking just to make sure for cases where we
 240             # get a socket from a blocking source.
 241             sock.setblocking(0)
 242             self.set_socket(sock, map)
 243             self.connected = True
 244             # The constructor no longer requires that the socket
 245             # passed be connected.
 246             try:
 247                 self.addr = sock.getpeername()
 248             except socket.error, err:
 249                 if err.args[0] == ENOTCONN:
 250                     # To handle the case where we got an unconnected
 251                     # socket.
 252                     self.connected = False
 253                 else:
 254                     # The socket is broken in some unknown way, alert
 255                     # the user and remove it from the map (to prevent
 256                     # polling of broken sockets).
 257                     self.del_channel(map)
 258                     raise
 259         else:
 260             self.socket = None
 261
 262     def __repr__(self):
 263         status = [self.__class__.__module__+"."+self.__class__.__name__]
 264         if self.accepting and self.addr:
 265             status.append('listening')
 266         elif self.connected:
 267             status.append('connected')
 268         if self.addr is not None:
 269             try:
 270                 status.append('%s:%d' % self.addr)
 271             except TypeError:
 272                 status.append(repr(self.addr))
 273         return '<%s at %#x>' % (' '.join(status), id(self))
 274
 275     __str__ = __repr__
 276
 277     def add_channel(self, map=None):
 278         #self.log_info('adding channel %s' % self)
 279         if map is None:
 280             map = self._map
 281         map[self._fileno] = self
 282
 283     def del_channel(self, map=None):
 284         fd = self._fileno
 285         if map is None:
 286             map = self._map
 287         if fd in map:
 288             #self.log_info('closing channel %d:%s' % (fd, self))
 289             del map[fd]
 290         self._fileno = None
 291
 292     def create_socket(self, family, type):
 293         self.family_and_type = family, type
 294         sock = socket.socket(family, type)
 295         sock.setblocking(0)
 296         self.set_socket(sock)
 297
 298     def set_socket(self, sock, map=None):
 299         self.socket = sock
 300 ##        self.__dict__['socket'] = sock
 301         self._fileno = sock.fileno()
 302         self.add_channel(map)
 303
 304     def set_reuse_addr(self):
 305         # try to re-use a server port if possible
 306         try:
 307             self.socket.setsockopt(
 308                 socket.SOL_SOCKET, socket.SO_REUSEADDR,
 309                 self.socket.getsockopt(socket.SOL_SOCKET,
 310                                        socket.SO_REUSEADDR) | 1
 311                 )
 312         except socket.error:
 313             pass
 314
 315     # ==================================================
 316     # predicates for select()
 317     # these are used as filters for the lists of sockets
 318     # to pass to select().
 319     # ==================================================
 320
 321     def readable(self):
 322         return True
 323
 324     def writable(self):
 325         return True
 326
 327     # ==================================================
 328     # socket object methods.
 329     # ==================================================
 330
 331     def listen(self, num):
 332         self.accepting = True
 333         if os.name == 'nt' and num > 5:
 334             num = 5
 335         return self.socket.listen(num)
 336
 337     def bind(self, addr):
 338         self.addr = addr
 339         return self.socket.bind(addr)
 340
 341     def connect(self, address):
 342         self.connected = False
 343         err = self.socket.connect_ex(address)
 344         if err in (EINPROGRESS, EALREADY, EWOULDBLOCK) \
 345         or err == EINVAL and os.name in ('nt', 'ce'):
 346             return
 347         if err in (0, EISCONN):
 348             self.addr = address
 349             self.handle_connect_event()
 350         else:
 351             raise socket.error(err, errorcode[err])
 352
 353     def accept(self):
 354         # XXX can return either an address pair or None
 355         try:
 356             conn, addr = self.socket.accept()
 357         except TypeError:
 358             return None
 359         except socket.error as why:
 360             if why.args[0] in (EWOULDBLOCK, ECONNABORTED, EAGAIN):
 361                 return None
 362             else:
 363                 raise
 364         else:
 365             return conn, addr
 366
 367     def send(self, data):
 368         try:
 369             result = self.socket.send(data)
 370             return result
 371         except socket.error, why:
 372             if why.args[0] == EWOULDBLOCK:
 373                 return 0
 374             elif why.args[0] in _DISCONNECTED:
 375                 self.handle_close()
 376                 return 0
 377             else:
 378                 raise
 379
 380     def recv(self, buffer_size):
 381         try:
 382             data = self.socket.recv(buffer_size)
 383             if not data:
 384                 # a closed connection is indicated by signaling
 385                 # a read condition, and having recv() return 0.
 386                 self.handle_close()
 387                 return ''
 388             else:
 389                 return data
 390         except socket.error, why:
 391             # winsock sometimes throws ENOTCONN
 392             if why.args[0] in _DISCONNECTED:
 393                 self.handle_close()
 394                 return ''
 395             else:
 396                 raise
 397
 398     def close(self):
 399         self.connected = False
 400         self.accepting = False
 401         self.del_channel()
 402         try:
 403             self.socket.close()
 404         except socket.error, why:
 405             if why.args[0] not in (ENOTCONN, EBADF):
 406                 raise
 407
 408     # cheap inheritance, used to pass all other attribute
 409     # references to the underlying socket object.
 410     def __getattr__(self, attr):
 411         try:
 412             retattr = getattr(self.socket, attr)
 413         except AttributeError:
 414             raise AttributeError("%s instance has no attribute '%s'"
 415                                  %(self.__class__.__name__, attr))
 416         else:
 417             msg = "%(me)s.%(attr)s is deprecated. Use %(me)s.socket.%(attr)s " \
 418                   "instead." % {'me': self.__class__.__name__, 'attr':attr}
 419             warnings.warn(msg, DeprecationWarning, stacklevel=2)
 420             return retattr
 421
 422     # log and log_info may be overridden to provide more sophisticated
 423     # logging and warning methods. In general, log is for 'hit' logging
 424     # and 'log_info' is for informational, warning and error logging.
 425
 426     def log(self, message):
 427         sys.stderr.write('log: %s\n' % str(message))
 428
 429     def log_info(self, message, type='info'):
 430         if type not in self.ignore_log_types:
 431             print '%s: %s' % (type, message)
 432
 433     def handle_read_event(self):
 434         if self.accepting:
 435             # accepting sockets are never connected, they "spawn" new
 436             # sockets that are connected
 437             self.handle_accept()
 438         elif not self.connected:
 439             self.handle_connect_event()
 440             self.handle_read()
 441         else:
 442             self.handle_read()
 443
 444     def handle_connect_event(self):
 445         err = self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
 446         if err != 0:
 447             raise socket.error(err, _strerror(err))
 448         self.handle_connect()
 449         self.connected = True
 450
 451     def handle_write_event(self):
 452         if self.accepting:
 453             # Accepting sockets shouldn't get a write event.
 454             # We will pretend it didn't happen.
 455             return
 456
 457         if not self.connected:
 458             #check for errors
 459             err = self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
 460             if err != 0:
 461                 raise socket.error(err, _strerror(err))
 462
 463             self.handle_connect_event()
 464         self.handle_write()
 465
 466     def handle_expt_event(self):
 467         # handle_expt_event() is called if there might be an error on the
 468         # socket, or if there is OOB data
 469         # check for the error condition first
 470         err = self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
 471         if err != 0:
 472             # we can get here when select.select() says that there is an
 473             # exceptional condition on the socket
 474             # since there is an error, we'll go ahead and close the socket
 475             # like we would in a subclassed handle_read() that received no
 476             # data
 477             self.handle_close()
 478         else:
 479             self.handle_expt()
 480
 481     def handle_error(self):
 482         nil, t, v, tbinfo = compact_traceback()
 483
 484         # sometimes a user repr method will crash.
 485         try:
 486             self_repr = repr(self)
 487         except:
 488             self_repr = '<__repr__(self) failed for object at %0x>' % id(self)
 489
 490         self.log_info(
 491             'uncaptured python exception, closing channel %s (%s:%s %s)' % (
 492                 self_repr,
 493                 t,
 494                 v,
 495                 tbinfo
 496                 ),
 497             'error'
 498             )
 499         self.handle_close()
 500
 501     def handle_expt(self):
 502         self.log_info('unhandled incoming priority event', 'warning')
 503
 504     def handle_read(self):
 505         self.log_info('unhandled read event', 'warning')
 506
 507     def handle_write(self):
 508         self.log_info('unhandled write event', 'warning')
 509
 510     def handle_connect(self):
 511         self.log_info('unhandled connect event', 'warning')
 512
 513     def handle_accept(self):
 514         self.log_info('unhandled accept event', 'warning')
 515
 516     def handle_close(self):
 517         self.log_info('unhandled close event', 'warning')
 518         self.close()
 519
 520 # ---------------------------------------------------------------------------
 521 # adds simple buffered output capability, useful for simple clients.
 522 # [for more sophisticated usage use asynchat.async_chat]
 523 # ---------------------------------------------------------------------------
 524
 525 class dispatcher_with_send(dispatcher):
 526
 527     def __init__(self, sock=None, map=None):
 528         dispatcher.__init__(self, sock, map)
 529         self.out_buffer = ''
 530
 531     def initiate_send(self):
 532         num_sent = 0
 533         num_sent = dispatcher.send(self, self.out_buffer[:512])
 534         self.out_buffer = self.out_buffer[num_sent:]
 535
 536     def handle_write(self):
 537         self.initiate_send()
 538
 539     def writable(self):
 540         return (not self.connected) or len(self.out_buffer)
 541
 542     def send(self, data):
 543         if self.debug:
 544             self.log_info('sending %s' % repr(data))
 545         self.out_buffer = self.out_buffer + data
 546         self.initiate_send()
 547
 548 # ---------------------------------------------------------------------------
 549 # used for debugging.
 550 # ---------------------------------------------------------------------------
 551
 552 def compact_traceback():
 553     t, v, tb = sys.exc_info()
 554     tbinfo = []
 555     if not tb: # Must have a traceback
 556         raise AssertionError("traceback does not exist")
 557     while tb:
 558         tbinfo.append((
 559             tb.tb_frame.f_code.co_filename,
 560             tb.tb_frame.f_code.co_name,
 561             str(tb.tb_lineno)
 562             ))
 563         tb = tb.tb_next
 564
 565     # just to be safe
 566     del tb
 567
 568     file, function, line = tbinfo[-1]
 569     info = ' '.join(['[%s|%s|%s]' % x for x in tbinfo])
 570     return (file, function, line), t, v, info
 571
 572 def close_all(map=None, ignore_all=False):
 573     if map is None:
 574         map = socket_map
 575     for x in map.values():
 576         try:
 577             x.close()
 578         except OSError, x:
 579             if x.args[0] == EBADF:
 580                 pass
 581             elif not ignore_all:
 582                 raise
 583         except _reraised_exceptions:
 584             raise
 585         except:
 586             if not ignore_all:
 587                 raise
 588     map.clear()
 589
 590 # Asynchronous File I/O:
 591 #
 592 # After a little research (reading man pages on various unixen, and
 593 # digging through the linux kernel), I've determined that select()
 594 # isn't meant for doing asynchronous file i/o.
 595 # Heartening, though - reading linux/mm/filemap.c shows that linux
 596 # supports asynchronous read-ahead.  So _MOST_ of the time, the data
 597 # will be sitting in memory for us already when we go to read it.
 598 #
 599 # What other OS's (besides NT) support async file i/o?  [VMS?]
 600 #
 601 # Regardless, this is useful for pipes, and stdin/stdout...
 602
 603 if os.name == 'posix':
 604     import fcntl
 605
 606     class file_wrapper:
 607         # Here we override just enough to make a file
 608         # look like a socket for the purposes of asyncore.
 609         # The passed fd is automatically os.dup()'d
 610
 611         def __init__(self, fd):
 612             self.fd = os.dup(fd)
 613
 614         def recv(self, *args):
 615             return os.read(self.fd, *args)
 616
 617         def send(self, *args):
 618             return os.write(self.fd, *args)
 619
 620         def getsockopt(self, level, optname, buflen=None):
 621             if (level == socket.SOL_SOCKET and
 622                 optname == socket.SO_ERROR and
 623                 not buflen):
 624                 return 0
 625             raise NotImplementedError("Only asyncore specific behaviour "
 626                                       "implemented.")
 627
 628         read = recv
 629         write = send
 630
 631         def close(self):
 632             os.close(self.fd)
 633
 634         def fileno(self):
 635             return self.fd
 636
 637     class file_dispatcher(dispatcher):
 638
 639         def __init__(self, fd, map=None):
 640             dispatcher.__init__(self, None, map)
 641             self.connected = True
 642             try:
 643                 fd = fd.fileno()
 644             except AttributeError:
 645                 pass
 646             self.set_file(fd)
 647             # set it to non-blocking mode
 648             flags = fcntl.fcntl(fd, fcntl.F_GETFL, 0)
 649             flags = flags | os.O_NONBLOCK
 650             fcntl.fcntl(fd, fcntl.F_SETFL, flags)
 651
 652         def set_file(self, fd):
 653             self.socket = file_wrapper(fd)
 654             self._fileno = self.socket.fileno()
 655             self.add_channel()