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         ESHUTDOWN       = 10058
  41 else:
  42         from errno import EALREADY, EINPROGRESS, EWOULDBLOCK, ECONNRESET, ENOTCONN, ESHUTDOWN
  43
  44 socket_map = {}
  45
  46 def poll (timeout=0.0):
  47         if socket_map:
  48                 r = []; w = []; e = []
  49                 for s in socket_map.keys():
  50                         if s.readable():
  51                                 r.append (s)
  52                         if s.writable():
  53                                 w.append (s)
  54
  55                 (r,w,e) = select.select (r,w,e, timeout)
  56
  57                 for x in r:
  58                         try:
  59                                 x.handle_read_event()
  60                         except:
  61                                 x.handle_error()
  62                 for x in w:
  63                         try:
  64                                 x.handle_write_event()
  65                         except:
  66                                 x.handle_error()
  67
  68 def poll2 (timeout=0.0):
  69         import poll
  70         # timeout is in milliseconds
  71         timeout = int(timeout*1000)
  72         if socket_map:
  73                 fd_map = {}
  74                 for s in socket_map.keys():
  75                         fd_map[s.fileno()] = s
  76                 l = []
  77                 for fd, s in fd_map.items():
  78                         flags = 0
  79                         if s.readable():
  80                                 flags = poll.POLLIN
  81                         if s.writable():
  82                                 flags = flags | poll.POLLOUT
  83                         if flags:
  84                                 l.append (fd, flags)
  85                 r = poll.poll (l, timeout)
  86                 for fd, flags in r:
  87                         s = fd_map[fd]
  88                         try:
  89                                 if (flags & poll.POLLIN):
  90                                         s.handle_read_event()
  91                                 if (flags & poll.POLLOUT):
  92                                         s.handle_write_event()
  93                                 if (flags & poll.POLLERR):
  94                                         s.handle_expt_event()
  95                         except:
  96                                 s.handle_error()
  97
  98
  99 def loop (timeout=30.0, use_poll=0):
 100
 101         if use_poll:
 102                 poll_fun = poll2
 103         else:
 104                 poll_fun = poll
 105
 106         while socket_map:
 107                 poll_fun (timeout)
 108
 109 class dispatcher:
 110         debug = 0
 111         connected = 0
 112         accepting = 0
 113         closing = 0
 114         addr = None
 115
 116         def __init__ (self, sock=None):
 117                 if sock:
 118                         self.set_socket (sock)
 119                         # I think it should inherit this anyway
 120                         self.socket.setblocking (0)
 121                         self.connected = 1
 122
 123         def __repr__ (self):
 124                 try:
 125                         status = []
 126                         if self.accepting and self.addr:
 127                                 status.append ('listening')
 128                         elif self.connected:
 129                                 status.append ('connected')
 130                         if self.addr:
 131                                 status.append ('%s:%d' % self.addr)
 132                         return '<%s %s at %x>' % (
 133                                 self.__class__.__name__,
 134                                 string.join (status, ' '),
 135                                 id(self)
 136                                 )
 137                 except:
 138                         try:
 139                                 ar = repr(self.addr)
 140                         except:
 141                                 ar = 'no self.addr!'
 142
 143                         return '<__repr__ (self) failed for object at %x (addr=%s)>' % (id(self),ar)
 144
 145         def add_channel (self):
 146                 if __debug__:
 147                         self.log ('adding channel %s' % self)
 148                 socket_map [self] = 1
 149
 150         def del_channel (self):
 151                 if socket_map.has_key (self):
 152                         if __debug__:
 153                                 self.log ('closing channel %d:%s' % (self.fileno(), self))
 154                         del socket_map [self]
 155
 156         def create_socket (self, family, type):
 157                 self.family_and_type = family, type
 158                 self.socket = socket.socket (family, type)
 159                 self.socket.setblocking(0)
 160                 self.add_channel()
 161
 162         def set_socket (self, socket):
 163                 # This is done so we can be called safely from __init__
 164                 self.__dict__['socket'] = socket
 165                 self.add_channel()
 166
 167         def set_reuse_addr (self):
 168                 # try to re-use a server port if possible
 169                 try:
 170                         self.socket.setsockopt (
 171                                 socket.SOL_SOCKET, socket.SO_REUSEADDR,
 172                                 self.socket.getsockopt (socket.SOL_SOCKET, socket.SO_REUSEADDR) | 1
 173                                 )
 174                 except:
 175                         pass
 176
 177         # ==================================================
 178         # predicates for select()
 179         # these are used as filters for the lists of sockets
 180         # to pass to select().
 181         # ==================================================
 182
 183         def readable (self):
 184                 return 1
 185
 186         if os.name == 'mac':
 187                 # The macintosh will select a listening socket for
 188                 # write if you let it.  What might this mean?
 189                 def writable (self):
 190                         return not self.accepting
 191         else:
 192                 def writable (self):
 193                         return 1
 194
 195         # ==================================================
 196         # socket object methods.
 197         # ==================================================
 198
 199         def listen (self, num):
 200                 self.accepting = 1
 201                 if os.name == 'nt' and num > 5:
 202                         num = 1
 203                 return self.socket.listen (num)
 204
 205         def bind (self, addr):
 206                 self.addr = addr
 207                 return self.socket.bind (addr)
 208
 209         def connect (self, address):
 210                 self.connected = 0
 211                 try:
 212                         self.socket.connect (address)
 213                 except socket.error, why:
 214                         if why[0] in (EINPROGRESS, EALREADY, EWOULDBLOCK):
 215                                 return
 216                         else:
 217                                 raise socket.error, why
 218                 self.connected = 1
 219                 self.handle_connect()
 220
 221         def accept (self):
 222                 try:
 223                         conn, addr = self.socket.accept()
 224                         return conn, addr
 225                 except socket.error, why:
 226                         if why[0] == EWOULDBLOCK:
 227                                 pass
 228                         else:
 229                                 raise socket.error, why
 230
 231         def send (self, data):
 232                 try:
 233                         result = self.socket.send (data)
 234                         return result
 235                 except socket.error, why:
 236                         if why[0] == EWOULDBLOCK:
 237                                 return 0
 238                         else:
 239                                 raise socket.error, why
 240                         return 0
 241
 242         def recv (self, buffer_size):
 243                 try:
 244                         data = self.socket.recv (buffer_size)
 245                         if not data:
 246                                 # a closed connection is indicated by signaling
 247                                 # a read condition, and having recv() return 0.
 248                                 self.handle_close()
 249                                 return ''
 250                         else:
 251                                 return data
 252                 except socket.error, why:
 253                         # winsock sometimes throws ENOTCONN
 254                         if why[0] in [ECONNRESET, ENOTCONN, ESHUTDOWN]:
 255                                 self.handle_close()
 256                                 return ''
 257                         else:
 258                                 raise socket.error, why
 259
 260         def close (self):
 261                 self.del_channel()
 262                 self.socket.close()
 263
 264         # cheap inheritance, used to pass all other attribute
 265         # references to the underlying socket object.
 266         # NOTE: this may be removed soon for performance reasons.
 267         def __getattr__ (self, attr):
 268                 return getattr (self.socket, attr)
 269
 270         def log (self, message):
 271                 print 'log:', message
 272
 273         def handle_read_event (self):
 274                 if self.accepting:
 275                         # for an accepting socket, getting a read implies
 276                         # that we are connected
 277                         if not self.connected:
 278                                 self.connected = 1
 279                         self.handle_accept()
 280                 elif not self.connected:
 281                         self.handle_connect()
 282                         self.connected = 1
 283                         self.handle_read()
 284                 else:
 285                         self.handle_read()
 286
 287         def handle_write_event (self):
 288                 # getting a write implies that we are connected
 289                 if not self.connected:
 290                         self.handle_connect()
 291                         self.connected = 1
 292                 self.handle_write()
 293
 294         def handle_expt_event (self):
 295                 self.handle_expt()
 296
 297         def handle_error (self):
 298                 (file,fun,line), t, v, tbinfo = compact_traceback()
 299
 300                 # sometimes a user repr method will crash.
 301                 try:
 302                         self_repr = repr (self)
 303                 except:
 304                         self_repr = '<__repr__ (self) failed for object at %0x>' % id(self)
 305
 306                 print (
 307                         'uncaptured python exception, closing channel %s (%s:%s %s)' % (
 308                                 self_repr,
 309                                 t,
 310                                 v,
 311                                 tbinfo
 312                                 )
 313                         )
 314                 self.close()
 315
 316         def handle_expt (self):
 317                 if __debug__:
 318                         self.log ('unhandled exception')
 319
 320         def handle_read (self):
 321                 if __debug__:
 322                         self.log ('unhandled read event')
 323
 324         def handle_write (self):
 325                 if __debug__:
 326                         self.log ('unhandled write event')
 327
 328         def handle_connect (self):
 329                 if __debug__:
 330                         self.log ('unhandled connect event')
 331
 332         def handle_accept (self):
 333                 if __debug__:
 334                         self.log ('unhandled accept event')
 335
 336         def handle_close (self):
 337                 if __debug__:
 338                         self.log ('unhandled close event')
 339                 self.close()
 340
 341 # ---------------------------------------------------------------------------
 342 # adds simple buffered output capability, useful for simple clients.
 343 # [for more sophisticated usage use asynchat.async_chat]
 344 # ---------------------------------------------------------------------------
 345
 346 class dispatcher_with_send (dispatcher):
 347         def __init__ (self, sock=None):
 348                 dispatcher.__init__ (self, sock)
 349                 self.out_buffer = ''
 350
 351         def initiate_send (self):
 352                 num_sent = 0
 353                 num_sent = dispatcher.send (self, self.out_buffer[:512])
 354                 self.out_buffer = self.out_buffer[num_sent:]
 355
 356         def handle_write (self):
 357                 self.initiate_send()
 358
 359         def writable (self):
 360                 return (not self.connected) or len(self.out_buffer)
 361
 362         def send (self, data):
 363                 if self.debug:
 364                         self.log ('sending %s' % repr(data))
 365                 self.out_buffer = self.out_buffer + data
 366                 self.initiate_send()
 367
 368 # ---------------------------------------------------------------------------
 369 # used for debugging.
 370 # ---------------------------------------------------------------------------
 371
 372 def compact_traceback ():
 373         t,v,tb = sys.exc_info()
 374         tbinfo = []
 375         while 1:
 376                 tbinfo.append (
 377                         tb.tb_frame.f_code.co_filename,
 378                         tb.tb_frame.f_code.co_name,
 379                         str(tb.tb_lineno)
 380                         )
 381                 tb = tb.tb_next
 382                 if not tb:
 383                         break
 384
 385         # just to be safe
 386         del tb
 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), t, v, 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