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 30There are only two ways to have a program on a single processor do "more 31than one thing at a time". Multi-threaded programming is the simplest and 32most popular way to do it, but there is another very different technique, 33that lets you have nearly all the advantages of multi-threading, without 34actually using multiple threads. it's really only practical if your program 35is largely I/O bound. If your program is CPU bound, then pre-emptive 36scheduled threads are probably what you really need. Network servers are 37rarely CPU-bound, however. 38 39If your operating system supports the select() system call in its I/O 40library (and nearly all do), then you can use it to juggle multiple 41communication channels at once; doing other work while your I/O is taking 42place in the "background." Although this strategy can seem strange and 43complex, especially at first, it is in many ways easier to understand and 44control than multi-threaded programming. The module documented here solves 45many of the difficult problems for you, making the task of building 46sophisticated high-performance network servers and clients a snap. 47""" 48 49import select 50import socket 51import sys 52import time 53import warnings 54 55import os 56from 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 63try: 64 socket_map 65except NameError: 66 socket_map = {} 67 68def _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 76class ExitNow(Exception): 77 pass 78 79_reraised_exceptions = (ExitNow, KeyboardInterrupt, SystemExit) 80 81def read(obj): 82 try: 83 obj.handle_read_event() 84 except _reraised_exceptions: 85 raise 86 except: 87 obj.handle_error() 88 89def write(obj): 90 try: 91 obj.handle_write_event() 92 except _reraised_exceptions: 93 raise 94 except: 95 obj.handle_error() 96 97def _exception(obj): 98 try: 99 obj.handle_expt_event() 100 except _reraised_exceptions: 101 raise 102 except: 103 obj.handle_error() 104 105def 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 125def 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 169def 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 201poll3 = poll2 # Alias for backward compatibility 202 203def 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 221class 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 525class 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 552def 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 572def 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 603if 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() 656