asyncore.py

# -*- Mode: Python -*-
# Id: asyncore.py,v 2.51 2000/09/07 22:29:26 rushing Exp
# Author: Sam Rushing <rushing@nightmare.com>

# ======================================================================
# Copyright 1996 by Sam Rushing
#
# All Rights Reserved
#
# Permission to use, copy, modify, and distribute this software and
# its documentation for any purpose and without fee is hereby
# granted, provided that the above copyright notice appear in all
# copies and that both that copyright notice and this permission
# notice appear in supporting documentation, and that the name of Sam
# Rushing not be used in advertising or publicity pertaining to
# distribution of the software without specific, written prior
# permission.
#
# SAM RUSHING DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE,
# INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS, IN
# NO EVENT SHALL SAM RUSHING BE LIABLE FOR ANY SPECIAL, INDIRECT OR
# CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS
# OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT,
# NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
# CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
# ======================================================================

"""Basic infrastructure for asynchronous socket service clients and servers.

There are only two ways to have a program on a single processor do "more
than one thing at a time". Multi-threaded programming is the simplest and
most popular way to do it, but there is another very different technique,
that lets you have nearly all the advantages of multi-threading, without
actually using multiple threads. it's really only practical if your program
is largely I/O bound. If your program is CPU bound, then pre-emptive
scheduled threads are probably what you really need. Network servers are
rarely CPU-bound, however.

If your operating system supports the select() system call in its I/O
library (and nearly all do), then you can use it to juggle multiple
communication channels at once; doing other work while your I/O is taking
place in the "background." Although this strategy can seem strange and
complex, especially at first, it is in many ways easier to understand and
control than multi-threaded programming. The module documented here solves
many of the difficult problems for you, making the task of building
sophisticated high-performance network servers and clients a snap.
"""

importselect
importsocket
importsys
importtime
importwarnings

importos
fromerrnoimportEALREADY, EINPROGRESS, EWOULDBLOCK, ECONNRESET, EINVAL, \
ENOTCONN, ESHUTDOWN, EISCONN, EBADF, ECONNABORTED, EPIPE, EAGAIN, \
errorcode

_DISCONNECTED=frozenset({ECONNRESET, ENOTCONN, ESHUTDOWN, ECONNABORTED, EPIPE,
EBADF})

try:
socket_map
exceptNameError:
socket_map= {}

def_strerror(err):
try:
returnos.strerror(err)
except (ValueError, OverflowError, NameError):
iferrinerrorcode:
returnerrorcode[err]
return"Unknown error %s"%err

classExitNow(Exception):
pass

_reraised_exceptions= (ExitNow, KeyboardInterrupt, SystemExit)

defread(obj):
try:
obj.handle_read_event()
except_reraised_exceptions:
raise
except:
obj.handle_error()

defwrite(obj):
try:
obj.handle_write_event()
except_reraised_exceptions:
raise
except:
obj.handle_error()

def_exception(obj):
try:
obj.handle_expt_event()
except_reraised_exceptions:
raise
except:
obj.handle_error()

defreadwrite(obj, flags):
try:
ifflags&select.POLLIN:
obj.handle_read_event()
ifflags&select.POLLOUT:
obj.handle_write_event()
ifflags&select.POLLPRI:
obj.handle_expt_event()
ifflags& (select.POLLHUP|select.POLLERR|select.POLLNVAL):
obj.handle_close()
exceptOSErrorase:
ife.args[0] notin_DISCONNECTED:
obj.handle_error()
else:
obj.handle_close()
except_reraised_exceptions:
raise
except:
obj.handle_error()

defpoll(timeout=0.0, map=None):
ifmapisNone:
map=socket_map
ifmap:
r= []; w= []; e= []
forfd, objinlist(map.items()):
is_r=obj.readable()
is_w=obj.writable()
ifis_r:
r.append(fd)
# accepting sockets should not be writable
ifis_wandnotobj.accepting:
w.append(fd)
ifis_roris_w:
e.append(fd)
if [] ==r==w==e:
time.sleep(timeout)
return

r, w, e=select.select(r, w, e, timeout)

forfdinr:
obj=map.get(fd)
ifobjisNone:
continue
read(obj)

forfdinw:
obj=map.get(fd)
ifobjisNone:
continue
write(obj)

forfdine:
obj=map.get(fd)
ifobjisNone:
continue
_exception(obj)

defpoll2(timeout=0.0, map=None):
# Use the poll() support added to the select module in Python 2.0
ifmapisNone:
map=socket_map
iftimeoutisnotNone:
# timeout is in milliseconds
timeout=int(timeout*1000)
pollster=select.poll()
ifmap:
forfd, objinlist(map.items()):
flags=0
ifobj.readable():
flags|=select.POLLIN|select.POLLPRI
# accepting sockets should not be writable
ifobj.writable() andnotobj.accepting:
flags|=select.POLLOUT
ifflags:
pollster.register(fd, flags)

r=pollster.poll(timeout)
forfd, flagsinr:
obj=map.get(fd)
ifobjisNone:
continue
readwrite(obj, flags)

poll3=poll2# Alias for backward compatibility

defloop(timeout=30.0, use_poll=False, map=None, count=None):
ifmapisNone:
map=socket_map

ifuse_pollandhasattr(select, 'poll'):
poll_fun=poll2
else:
poll_fun=poll

ifcountisNone:
whilemap:
poll_fun(timeout, map)

else:
whilemapandcount>0:
poll_fun(timeout, map)
count=count-1

classdispatcher:

debug=False
connected=False
accepting=False
connecting=False
closing=False
addr=None
ignore_log_types=frozenset({'warning'})

def__init__(self, sock=None, map=None):
ifmapisNone:
self._map=socket_map
else:
self._map=map

self._fileno=None

ifsock:
# Set to nonblocking just to make sure for cases where we
# get a socket from a blocking source.
sock.setblocking(0)
self.set_socket(sock, map)
self.connected=True
# The constructor no longer requires that the socket
# passed be connected.
try:
self.addr=sock.getpeername()
exceptOSErroraserr:
iferr.args[0] in (ENOTCONN, EINVAL):
# To handle the case where we got an unconnected
# socket.
self.connected=False
else:
# The socket is broken in some unknown way, alert
# the user and remove it from the map (to prevent
# polling of broken sockets).
self.del_channel(map)
raise
else:
self.socket=None

def__repr__(self):
status= [self.__class__.__module__+"."+self.__class__.__qualname__]
ifself.acceptingandself.addr:
status.append('listening')
elifself.connected:
status.append('connected')
ifself.addrisnotNone:
try:
status.append('%s:%d'%self.addr)
exceptTypeError:
status.append(repr(self.addr))
return'<%s at %#x>'% (' '.join(status), id(self))

__str__=__repr__

defadd_channel(self, map=None):
#self.log_info('adding channel %s' % self)
ifmapisNone:
map=self._map
map[self._fileno] =self

defdel_channel(self, map=None):
fd=self._fileno
ifmapisNone:
map=self._map
iffdinmap:
#self.log_info('closing channel %d:%s' % (fd, self))
delmap[fd]
self._fileno=None

defcreate_socket(self, family=socket.AF_INET, type=socket.SOCK_STREAM):
self.family_and_type=family, type
sock=socket.socket(family, type)
sock.setblocking(0)
self.set_socket(sock)

defset_socket(self, sock, map=None):
self.socket=sock
self._fileno=sock.fileno()
self.add_channel(map)

defset_reuse_addr(self):
# try to re-use a server port if possible
try:
self.socket.setsockopt(
socket.SOL_SOCKET, socket.SO_REUSEADDR,
self.socket.getsockopt(socket.SOL_SOCKET,
socket.SO_REUSEADDR) |1
 )
exceptOSError:
pass

# ==================================================
# predicates for select()
# these are used as filters for the lists of sockets
# to pass to select().
# ==================================================

defreadable(self):
returnTrue

defwritable(self):
returnTrue

# ==================================================
# socket object methods.
# ==================================================

deflisten(self, num):
self.accepting=True
ifos.name=='nt'andnum>5:
num=5
returnself.socket.listen(num)

defbind(self, addr):
self.addr=addr
returnself.socket.bind(addr)

defconnect(self, address):
self.connected=False
self.connecting=True
err=self.socket.connect_ex(address)
iferrin (EINPROGRESS, EALREADY, EWOULDBLOCK) \
orerr==EINVALandos.name=='nt':
self.addr=address
return
iferrin (0, EISCONN):
self.addr=address
self.handle_connect_event()
else:
raiseOSError(err, errorcode[err])

defaccept(self):
# XXX can return either an address pair or None
try:
conn, addr=self.socket.accept()
exceptTypeError:
returnNone
exceptOSErroraswhy:
ifwhy.args[0] in (EWOULDBLOCK, ECONNABORTED, EAGAIN):
returnNone
else:
raise
else:
returnconn, addr

defsend(self, data):
try:
result=self.socket.send(data)
returnresult
exceptOSErroraswhy:
ifwhy.args[0] ==EWOULDBLOCK:
return0
elifwhy.args[0] in_DISCONNECTED:
self.handle_close()
return0
else:
raise

defrecv(self, buffer_size):
try:
data=self.socket.recv(buffer_size)
ifnotdata:
# a closed connection is indicated by signaling
# a read condition, and having recv() return 0.
self.handle_close()
returnb''
else:
returndata
exceptOSErroraswhy:
# winsock sometimes raises ENOTCONN
ifwhy.args[0] in_DISCONNECTED:
self.handle_close()
returnb''
else:
raise

defclose(self):
self.connected=False
self.accepting=False
self.connecting=False
self.del_channel()
ifself.socketisnotNone:
try:
self.socket.close()
exceptOSErroraswhy:
ifwhy.args[0] notin (ENOTCONN, EBADF):
raise

# log and log_info may be overridden to provide more sophisticated
# logging and warning methods. In general, log is for 'hit' logging
# and 'log_info' is for informational, warning and error logging.

deflog(self, message):
sys.stderr.write('log: %s\n'%str(message))

deflog_info(self, message, type='info'):
iftypenotinself.ignore_log_types:
print('%s: %s'% (type, message))

defhandle_read_event(self):
ifself.accepting:
# accepting sockets are never connected, they "spawn" new
# sockets that are connected
self.handle_accept()
elifnotself.connected:
ifself.connecting:
self.handle_connect_event()
self.handle_read()
else:
self.handle_read()

defhandle_connect_event(self):
err=self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
iferr!=0:
raiseOSError(err, _strerror(err))
self.handle_connect()
self.connected=True
self.connecting=False

defhandle_write_event(self):
ifself.accepting:
# Accepting sockets shouldn't get a write event.
# We will pretend it didn't happen.
return

ifnotself.connected:
ifself.connecting:
self.handle_connect_event()
self.handle_write()

defhandle_expt_event(self):
# handle_expt_event() is called if there might be an error on the
# socket, or if there is OOB data
# check for the error condition first
err=self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
iferr!=0:
# we can get here when select.select() says that there is an
# exceptional condition on the socket
# since there is an error, we'll go ahead and close the socket
# like we would in a subclassed handle_read() that received no
# data
self.handle_close()
else:
self.handle_expt()

defhandle_error(self):
nil, t, v, tbinfo=compact_traceback()

# sometimes a user repr method will crash.
try:
self_repr=repr(self)
except:
self_repr='<__repr__(self) failed for object at %0x>'%id(self)

self.log_info(
'uncaptured python exception, closing channel %s (%s:%s %s)'% (
self_repr,
t,
v,
tbinfo
 ),
'error'
 )
self.handle_close()

defhandle_expt(self):
self.log_info('unhandled incoming priority event', 'warning')

defhandle_read(self):
self.log_info('unhandled read event', 'warning')

defhandle_write(self):
self.log_info('unhandled write event', 'warning')

defhandle_connect(self):
self.log_info('unhandled connect event', 'warning')

defhandle_accept(self):
pair=self.accept()
ifpairisnotNone:
self.handle_accepted(*pair)

defhandle_accepted(self, sock, addr):
sock.close()
self.log_info('unhandled accepted event', 'warning')

defhandle_close(self):
self.log_info('unhandled close event', 'warning')
self.close()

# ---------------------------------------------------------------------------
# adds simple buffered output capability, useful for simple clients.
# [for more sophisticated usage use asynchat.async_chat]
# ---------------------------------------------------------------------------

classdispatcher_with_send(dispatcher):

def__init__(self, sock=None, map=None):
dispatcher.__init__(self, sock, map)
self.out_buffer=b''

definitiate_send(self):
num_sent=0
num_sent=dispatcher.send(self, self.out_buffer[:65536])
self.out_buffer=self.out_buffer[num_sent:]

defhandle_write(self):
self.initiate_send()

defwritable(self):
return (notself.connected) orlen(self.out_buffer)

defsend(self, data):
ifself.debug:
self.log_info('sending %s'%repr(data))
self.out_buffer=self.out_buffer+data
self.initiate_send()

# ---------------------------------------------------------------------------
# used for debugging.
# ---------------------------------------------------------------------------

defcompact_traceback():
t, v, tb=sys.exc_info()
tbinfo= []
ifnottb: # Must have a traceback
raiseAssertionError("traceback does not exist")
whiletb:
tbinfo.append((
tb.tb_frame.f_code.co_filename,
tb.tb_frame.f_code.co_name,
str(tb.tb_lineno)
 ))
tb=tb.tb_next

# just to be safe
deltb

file, function, line=tbinfo[-1]
info=' '.join(['[%s|%s|%s]'%xforxintbinfo])
return (file, function, line), t, v, info

defclose_all(map=None, ignore_all=False):
ifmapisNone:
map=socket_map
forxinlist(map.values()):
try:
x.close()
exceptOSErrorasx:
ifx.args[0] ==EBADF:
pass
elifnotignore_all:
raise
except_reraised_exceptions:
raise
except:
ifnotignore_all:
raise
map.clear()

# Asynchronous File I/O:
#
# After a little research (reading man pages on various unixen, and
# digging through the linux kernel), I've determined that select()
# isn't meant for doing asynchronous file i/o.
# Heartening, though - reading linux/mm/filemap.c shows that linux
# supports asynchronous read-ahead. So _MOST_ of the time, the data
# will be sitting in memory for us already when we go to read it.
#
# What other OS's (besides NT) support async file i/o? [VMS?]
#
# Regardless, this is useful for pipes, and stdin/stdout...

ifos.name=='posix':
classfile_wrapper:
# Here we override just enough to make a file
# look like a socket for the purposes of asyncore.
# The passed fd is automatically os.dup()'d

def__init__(self, fd):
self.fd=os.dup(fd)

def__del__(self):
ifself.fd>=0:
warnings.warn("unclosed file %r"%self, ResourceWarning,
source=self)
self.close()

defrecv(self, *args):
returnos.read(self.fd, *args)

defsend(self, *args):
returnos.write(self.fd, *args)

defgetsockopt(self, level, optname, buflen=None):
if (level==socket.SOL_SOCKETand
optname==socket.SO_ERRORand
notbuflen):
return0
raiseNotImplementedError("Only asyncore specific behaviour "
"implemented.")

read=recv
write=send

defclose(self):
ifself.fd<0:
return
fd=self.fd
self.fd=-1
os.close(fd)

deffileno(self):
returnself.fd

classfile_dispatcher(dispatcher):

def__init__(self, fd, map=None):
dispatcher.__init__(self, None, map)
self.connected=True
try:
fd=fd.fileno()
exceptAttributeError:
pass
self.set_file(fd)
# set it to non-blocking mode
os.set_blocking(fd, False)

defset_file(self, fd):
self.socket=file_wrapper(fd)
self._fileno=self.socket.fileno()
self.add_channel()