Lib/asyncore.py

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