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