source: python/trunk/Lib/asyncore.py

Last change on this file was 391, checked in by dmik, 11 years ago

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