asyncore.py (14082B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\stdlib\asyncore.py 2 import select 3 import socket 4 import sys 5 import time 6 import warnings 7 import os 8 from errno import EALREADY, EINPROGRESS, EWOULDBLOCK, ECONNRESET, EINVAL, ENOTCONN, ESHUTDOWN, EINTR, EISCONN, EBADF, ECONNABORTED, errorcode 9 try: 10 socket_map 11 except NameError: 12 socket_map = {} 13 14 def _strerror(err): 15 try: 16 return os.strerror(err) 17 except (ValueError, OverflowError, NameError): 18 if err in errorcode: 19 return errorcode[err] 20 return 'Unknown error %s' % err 21 22 23 class ExitNow(Exception): 24 pass 25 26 27 _reraised_exceptions = (ExitNow, KeyboardInterrupt, SystemExit) 28 29 def read(obj): 30 try: 31 obj.handle_read_event() 32 except _reraised_exceptions: 33 raise 34 except: 35 obj.handle_error() 36 37 38 def write(obj): 39 try: 40 obj.handle_write_event() 41 except _reraised_exceptions: 42 raise 43 except: 44 obj.handle_error() 45 46 47 def _exception(obj): 48 try: 49 obj.handle_expt_event() 50 except _reraised_exceptions: 51 raise 52 except: 53 obj.handle_error() 54 55 56 def readwrite(obj, flags): 57 try: 58 if flags & select.POLLIN: 59 obj.handle_read_event() 60 if flags & select.POLLOUT: 61 obj.handle_write_event() 62 if flags & select.POLLPRI: 63 obj.handle_expt_event() 64 if flags & (select.POLLHUP | select.POLLERR | select.POLLNVAL): 65 obj.handle_close() 66 except socket.error as e: 67 if e.args[0] not in (EBADF, 68 ECONNRESET, 69 ENOTCONN, 70 ESHUTDOWN, 71 ECONNABORTED): 72 obj.handle_error() 73 else: 74 obj.handle_close() 75 except _reraised_exceptions: 76 raise 77 except: 78 obj.handle_error() 79 80 81 def poll(timeout = 0.0, map = None): 82 if map is None: 83 map = socket_map 84 if map: 85 r = [] 86 w = [] 87 e = [] 88 for fd, obj in map.items(): 89 is_r = obj.readable() 90 is_w = obj.writable() 91 if is_r: 92 r.append(fd) 93 if is_w: 94 w.append(fd) 95 if is_r or is_w: 96 e.append(fd) 97 98 if [] == r == w == e: 99 time.sleep(timeout) 100 return 101 try: 102 r, w, e = select.select(r, w, e, timeout) 103 except select.error as err: 104 if err.args[0] != EINTR: 105 raise 106 else: 107 return 108 109 for fd in r: 110 obj = map.get(fd) 111 if obj is None: 112 continue 113 read(obj) 114 115 for fd in w: 116 obj = map.get(fd) 117 if obj is None: 118 continue 119 write(obj) 120 121 for fd in e: 122 obj = map.get(fd) 123 if obj is None: 124 continue 125 _exception(obj) 126 127 128 def poll2(timeout = 0.0, map = None): 129 if map is None: 130 map = socket_map 131 if timeout is not None: 132 timeout = int(timeout * 1000) 133 pollster = select.poll() 134 if map: 135 for fd, obj in map.items(): 136 flags = 0 137 if obj.readable(): 138 flags |= select.POLLIN | select.POLLPRI 139 if obj.writable(): 140 flags |= select.POLLOUT 141 if flags: 142 flags |= select.POLLERR | select.POLLHUP | select.POLLNVAL 143 pollster.register(fd, flags) 144 145 try: 146 r = pollster.poll(timeout) 147 except select.error as err: 148 if err.args[0] != EINTR: 149 raise 150 r = [] 151 152 for fd, flags in r: 153 obj = map.get(fd) 154 if obj is None: 155 continue 156 readwrite(obj, flags) 157 158 159 poll3 = poll2 160 161 def loop(timeout = 30.0, use_poll = False, map = None, count = None): 162 if map is None: 163 map = socket_map 164 if use_poll and hasattr(select, 'poll'): 165 poll_fun = poll2 166 else: 167 poll_fun = poll 168 if count is None: 169 while map: 170 poll_fun(timeout, map) 171 172 else: 173 while map and count > 0: 174 poll_fun(timeout, map) 175 count = count - 1 176 177 178 class dispatcher(): 179 debug = False 180 connected = False 181 accepting = False 182 closing = False 183 addr = None 184 ignore_log_types = frozenset(['warning']) 185 186 def __init__(self, sock = None, map = None): 187 if map is None: 188 self._map = socket_map 189 else: 190 self._map = map 191 self._fileno = None 192 if sock: 193 sock.setblocking(0) 194 self.set_socket(sock, map) 195 self.connected = True 196 try: 197 self.addr = sock.getpeername() 198 except socket.error as err: 199 if err.args[0] == ENOTCONN: 200 self.connected = False 201 else: 202 self.del_channel(map) 203 raise 204 205 else: 206 self.socket = None 207 208 def __repr__(self): 209 status = [self.__class__.__module__ + '.' + self.__class__.__name__] 210 if self.accepting and self.addr: 211 status.append('listening') 212 elif self.connected: 213 status.append('connected') 214 if self.addr is not None: 215 try: 216 status.append('%s:%d' % self.addr) 217 except TypeError: 218 status.append(repr(self.addr)) 219 220 return '<%s at %#x>' % (' '.join(status), id(self)) 221 222 __str__ = __repr__ 223 224 def add_channel(self, map = None): 225 if map is None: 226 map = self._map 227 map[self._fileno] = self 228 229 def del_channel(self, map = None): 230 fd = self._fileno 231 if map is None: 232 map = self._map 233 if fd in map: 234 del map[fd] 235 self._fileno = None 236 237 def create_socket(self, family, type): 238 self.family_and_type = (family, type) 239 sock = socket.socket(family, type) 240 sock.setblocking(0) 241 self.set_socket(sock) 242 243 def set_socket(self, sock, map = None): 244 self.socket = sock 245 self._fileno = sock.fileno() 246 self.add_channel(map) 247 248 def set_reuse_addr(self): 249 try: 250 self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) | 1) 251 except socket.error: 252 pass 253 254 def readable(self): 255 return True 256 257 def writable(self): 258 return True 259 260 def listen(self, num): 261 self.accepting = True 262 if os.name == 'nt' and num > 5: 263 num = 5 264 return self.socket.listen(num) 265 266 def bind(self, addr): 267 self.addr = addr 268 return self.socket.bind(addr) 269 270 def connect(self, address): 271 self.connected = False 272 err = self.socket.connect_ex(address) 273 if err in (EINPROGRESS, EALREADY, EWOULDBLOCK) or err == EINVAL and os.name in ('nt', 'ce'): 274 return 275 if err in (0, EISCONN): 276 self.addr = address 277 self.handle_connect_event() 278 else: 279 raise socket.error(err, errorcode[err]) 280 281 def accept(self): 282 try: 283 conn, addr = self.socket.accept() 284 except TypeError: 285 return None 286 except socket.error as why: 287 if why.args[0] in (EWOULDBLOCK, ECONNABORTED): 288 return None 289 raise 290 else: 291 return (conn, addr) 292 293 def send(self, data): 294 try: 295 result = self.socket.send(data) 296 return result 297 except socket.error as why: 298 if why.args[0] == EWOULDBLOCK: 299 return 0 300 if why.args[0] in (ECONNRESET, 301 ENOTCONN, 302 ESHUTDOWN, 303 ECONNABORTED): 304 self.handle_close() 305 return 0 306 raise 307 308 def recv(self, buffer_size): 309 try: 310 data = self.socket.recv(buffer_size) 311 if not data: 312 self.handle_close() 313 return '' 314 return data 315 except socket.error as why: 316 if why.args[0] in [ECONNRESET, 317 ENOTCONN, 318 ESHUTDOWN, 319 ECONNABORTED]: 320 self.handle_close() 321 return '' 322 raise 323 324 def close(self): 325 self.connected = False 326 self.accepting = False 327 self.del_channel() 328 try: 329 self.socket.close() 330 except socket.error as why: 331 if why.args[0] not in (ENOTCONN, EBADF): 332 raise 333 334 def __getattr__(self, attr): 335 try: 336 retattr = getattr(self.socket, attr) 337 except AttributeError: 338 raise AttributeError("%s instance has no attribute '%s'" % (self.__class__.__name__, attr)) 339 else: 340 msg = '%(me)s.%(attr)s is deprecated. Use %(me)s.socket.%(attr)s instead.' % {'me': self.__class__.__name__, 341 'attr': attr} 342 warnings.warn(msg, DeprecationWarning, stacklevel=2) 343 return retattr 344 345 def log(self, message): 346 sys.stderr.write('log: %s\n' % str(message)) 347 348 def log_info(self, message, type = 'info'): 349 if type not in self.ignore_log_types: 350 print '%s: %s' % (type, message) 351 352 def handle_read_event(self): 353 if self.accepting: 354 self.handle_accept() 355 elif not self.connected: 356 self.handle_connect_event() 357 self.handle_read() 358 else: 359 self.handle_read() 360 361 def handle_connect_event(self): 362 err = self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR) 363 if err != 0: 364 raise socket.error(err, _strerror(err)) 365 self.handle_connect() 366 self.connected = True 367 368 def handle_write_event(self): 369 if self.accepting: 370 return 371 if not self.connected: 372 err = self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR) 373 if err != 0: 374 raise socket.error(err, _strerror(err)) 375 self.handle_connect_event() 376 self.handle_write() 377 378 def handle_expt_event(self): 379 err = self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR) 380 if err != 0: 381 self.handle_close() 382 else: 383 self.handle_expt() 384 385 def handle_error(self): 386 nil, t, v, tbinfo = compact_traceback() 387 try: 388 self_repr = repr(self) 389 except: 390 self_repr = '<__repr__(self) failed for object at %0x>' % id(self) 391 392 self.log_info('uncaptured python exception, closing channel %s (%s:%s %s)' % (self_repr, 393 t, 394 v, 395 tbinfo), 'error') 396 self.handle_close() 397 398 def handle_expt(self): 399 self.log_info('unhandled incoming priority event', 'warning') 400 401 def handle_read(self): 402 self.log_info('unhandled read event', 'warning') 403 404 def handle_write(self): 405 self.log_info('unhandled write event', 'warning') 406 407 def handle_connect(self): 408 self.log_info('unhandled connect event', 'warning') 409 410 def handle_accept(self): 411 self.log_info('unhandled accept event', 'warning') 412 413 def handle_close(self): 414 self.log_info('unhandled close event', 'warning') 415 self.close() 416 417 418 class dispatcher_with_send(dispatcher): 419 420 def __init__(self, sock = None, map = None): 421 dispatcher.__init__(self, sock, map) 422 self.out_buffer = '' 423 424 def initiate_send(self): 425 num_sent = 0 426 num_sent = dispatcher.send(self, self.out_buffer[:512]) 427 self.out_buffer = self.out_buffer[num_sent:] 428 429 def handle_write(self): 430 self.initiate_send() 431 432 def writable(self): 433 return not self.connected or len(self.out_buffer) 434 435 def send(self, data): 436 if self.debug: 437 self.log_info('sending %s' % repr(data)) 438 self.out_buffer = self.out_buffer + data 439 self.initiate_send() 440 441 442 def compact_traceback(): 443 t, v, tb = sys.exc_info() 444 tbinfo = [] 445 if not tb: 446 raise AssertionError('traceback does not exist') 447 while tb: 448 tbinfo.append((tb.tb_frame.f_code.co_filename, tb.tb_frame.f_code.co_name, str(tb.tb_lineno))) 449 tb = tb.tb_next 450 451 del tb 452 file, function, line = tbinfo[-1] 453 info = ' '.join([ '[%s|%s|%s]' % x for x in tbinfo ]) 454 return ((file, function, line), 455 t, 456 v, 457 info) 458 459 460 def close_all(map = None, ignore_all = False): 461 if map is None: 462 map = socket_map 463 for x in map.values(): 464 try: 465 x.close() 466 except OSError as x: 467 if x.args[0] == EBADF: 468 pass 469 elif not ignore_all: 470 raise 471 except _reraised_exceptions: 472 raise 473 except: 474 if not ignore_all: 475 raise 476 477 map.clear() 478 479 480 if os.name == 'posix': 481 import fcntl 482 483 class file_wrapper(): 484 485 def __init__(self, fd): 486 self.fd = os.dup(fd) 487 488 def recv(self, *args): 489 return os.read(self.fd, *args) 490 491 def send(self, *args): 492 return os.write(self.fd, *args) 493 494 def getsockopt(self, level, optname, buflen = None): 495 if level == socket.SOL_SOCKET and optname == socket.SO_ERROR and not buflen: 496 return 0 497 raise NotImplementedError('Only asyncore specific behaviour implemented.') 498 499 read = recv 500 write = send 501 502 def close(self): 503 os.close(self.fd) 504 505 def fileno(self): 506 return self.fd 507 508 509 class file_dispatcher(dispatcher): 510 511 def __init__(self, fd, map = None): 512 dispatcher.__init__(self, None, map) 513 self.connected = True 514 try: 515 fd = fd.fileno() 516 except AttributeError: 517 pass 518 519 self.set_file(fd) 520 flags = fcntl.fcntl(fd, fcntl.F_GETFL, 0) 521 flags = flags | os.O_NONBLOCK 522 fcntl.fcntl(fd, fcntl.F_SETFL, flags) 523 524 def set_file(self, fd): 525 self.socket = file_wrapper(fd) 526 self._fileno = self.socket.fileno() 527 self.add_channel()