Lib/asyncore.py

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