894 lines
27 KiB
Python
894 lines
27 KiB
Python
import errno
|
|
import heapq
|
|
import os
|
|
import select
|
|
import socket
|
|
import sys
|
|
import threading
|
|
import time
|
|
import traceback
|
|
import warnings
|
|
|
|
from .log import config_logging
|
|
from .log import debug
|
|
from .log import is_logging_configured
|
|
from .log import logger
|
|
|
|
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter('ignore', DeprecationWarning)
|
|
import asynchat
|
|
import asyncore
|
|
|
|
timer = getattr(time, 'monotonic', time.time)
|
|
_read = asyncore.read
|
|
_write = asyncore.write
|
|
|
|
_ERRNOS_DISCONNECTED = {
|
|
errno.ECONNRESET,
|
|
errno.ENOTCONN,
|
|
errno.ESHUTDOWN,
|
|
errno.ECONNABORTED,
|
|
errno.EPIPE,
|
|
errno.EBADF,
|
|
errno.ETIMEDOUT,
|
|
}
|
|
if hasattr(errno, "WSAECONNRESET"):
|
|
_ERRNOS_DISCONNECTED.add(errno.WSAECONNRESET)
|
|
if hasattr(errno, "WSAECONNABORTED"):
|
|
_ERRNOS_DISCONNECTED.add(errno.WSAECONNABORTED)
|
|
|
|
_ERRNOS_RETRY = {errno.EAGAIN, errno.EWOULDBLOCK}
|
|
if hasattr(errno, "WSAEWOULDBLOCK"):
|
|
_ERRNOS_RETRY.add(errno.WSAEWOULDBLOCK)
|
|
|
|
|
|
class RetryError(Exception):
|
|
pass
|
|
|
|
|
|
# ===================================================================
|
|
# --- scheduler
|
|
# ===================================================================
|
|
|
|
|
|
class _Scheduler:
|
|
def __init__(self):
|
|
self._tasks = []
|
|
self._cancellations = 0
|
|
|
|
def poll(self):
|
|
now = timer()
|
|
calls = []
|
|
while self._tasks:
|
|
if now < self._tasks[0].timeout:
|
|
break
|
|
call = heapq.heappop(self._tasks)
|
|
if call.cancelled:
|
|
self._cancellations -= 1
|
|
else:
|
|
calls.append(call)
|
|
|
|
for call in calls:
|
|
if call._repush:
|
|
heapq.heappush(self._tasks, call)
|
|
call._repush = False
|
|
continue
|
|
try:
|
|
call.call()
|
|
except Exception:
|
|
logger.error(traceback.format_exc())
|
|
|
|
if self._cancellations > 512 and self._cancellations > (
|
|
len(self._tasks) >> 1
|
|
):
|
|
debug(f"re-heapifying {self._cancellations} cancelled tasks")
|
|
self.reheapify()
|
|
|
|
try:
|
|
return max(0, self._tasks[0].timeout - now)
|
|
except IndexError:
|
|
pass
|
|
|
|
def register(self, what):
|
|
heapq.heappush(self._tasks, what)
|
|
|
|
def unregister(self, what):
|
|
self._cancellations += 1
|
|
|
|
def reheapify(self):
|
|
self._cancellations = 0
|
|
self._tasks = [x for x in self._tasks if not x.cancelled]
|
|
heapq.heapify(self._tasks)
|
|
|
|
def close(self):
|
|
for x in self._tasks:
|
|
try:
|
|
if not x.cancelled:
|
|
x.cancel()
|
|
except Exception:
|
|
logger.error(traceback.format_exc())
|
|
del self._tasks[:]
|
|
self._cancellations = 0
|
|
|
|
|
|
class _CallLater:
|
|
|
|
__slots__ = (
|
|
'_args',
|
|
'_delay',
|
|
'_errback',
|
|
'_kwargs',
|
|
'_repush',
|
|
'_sched',
|
|
'_target',
|
|
'cancelled',
|
|
'timeout',
|
|
)
|
|
|
|
def __init__(self, seconds, target, *args, **kwargs):
|
|
assert callable(target), f"{target} is not callable"
|
|
assert (
|
|
sys.maxsize >= seconds >= 0
|
|
), f"{seconds} is not greater than or equal to 0 seconds"
|
|
self._delay = seconds
|
|
self._target = target
|
|
self._args = args
|
|
self._kwargs = kwargs
|
|
self._errback = kwargs.pop('_errback', None)
|
|
self._sched = kwargs.pop('_scheduler')
|
|
self._repush = False
|
|
# seconds from the epoch at which to call the function
|
|
if not seconds:
|
|
self.timeout = 0
|
|
else:
|
|
self.timeout = timer() + self._delay
|
|
self.cancelled = False
|
|
self._sched.register(self)
|
|
|
|
def __lt__(self, other):
|
|
return self.timeout < other.timeout
|
|
|
|
def __le__(self, other):
|
|
return self.timeout <= other.timeout
|
|
|
|
def __repr__(self):
|
|
if self._target is None:
|
|
sig = object.__repr__(self)
|
|
else:
|
|
sig = repr(self._target)
|
|
sig += ' args=%s, kwargs=%s, cancelled=%s, secs=%s' % ( # noqa: UP031
|
|
self._args or '[]',
|
|
self._kwargs or '{}',
|
|
self.cancelled,
|
|
self._delay,
|
|
)
|
|
return f'<{sig}>'
|
|
|
|
__str__ = __repr__
|
|
|
|
def _post_call(self, exc):
|
|
if not self.cancelled:
|
|
self.cancel()
|
|
|
|
def call(self):
|
|
assert not self.cancelled, "already cancelled"
|
|
exc = None
|
|
try:
|
|
self._target(*self._args, **self._kwargs)
|
|
except Exception as _:
|
|
exc = _
|
|
if self._errback is not None:
|
|
self._errback()
|
|
else:
|
|
raise
|
|
finally:
|
|
self._post_call(exc)
|
|
|
|
def reset(self):
|
|
assert not self.cancelled, "already cancelled"
|
|
self.timeout = timer() + self._delay
|
|
self._repush = True
|
|
|
|
def cancel(self):
|
|
if not self.cancelled:
|
|
self.cancelled = True
|
|
self._target = self._args = self._kwargs = self._errback = None
|
|
self._sched.unregister(self)
|
|
|
|
|
|
class _CallEvery(_CallLater):
|
|
|
|
def _post_call(self, exc):
|
|
if not self.cancelled:
|
|
if exc:
|
|
self.cancel()
|
|
else:
|
|
self.timeout = timer() + self._delay
|
|
self._sched.register(self)
|
|
|
|
|
|
class _IOLoop:
|
|
|
|
READ = 1
|
|
WRITE = 2
|
|
_instance = None
|
|
_lock = threading.Lock()
|
|
_started_once = False
|
|
|
|
def __init__(self):
|
|
self.socket_map = {}
|
|
self.sched = _Scheduler()
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
self.close()
|
|
|
|
def __repr__(self):
|
|
status = [self.__class__.__module__ + "." + self.__class__.__name__]
|
|
status.append(
|
|
f"(fds={len(self.socket_map)}, tasks={len(self.sched._tasks)})"
|
|
)
|
|
return '<%s at %#x>' % (' '.join(status), id(self)) # noqa: UP031
|
|
|
|
__str__ = __repr__
|
|
|
|
@classmethod
|
|
def instance(cls):
|
|
if cls._instance is None:
|
|
with cls._lock:
|
|
if cls._instance is None:
|
|
cls._instance = cls()
|
|
return cls._instance
|
|
|
|
@classmethod
|
|
def factory(cls):
|
|
return cls()
|
|
|
|
def register(self, fd, instance, events):
|
|
raise NotImplementedError('must be implemented in subclass')
|
|
|
|
def unregister(self, fd):
|
|
raise NotImplementedError('must be implemented in subclass')
|
|
|
|
def modify(self, fd, events):
|
|
raise NotImplementedError('must be implemented in subclass')
|
|
|
|
def poll(self, timeout):
|
|
raise NotImplementedError('must be implemented in subclass')
|
|
|
|
def loop(self, timeout=None, blocking=True):
|
|
if not _IOLoop._started_once:
|
|
_IOLoop._started_once = True
|
|
if not is_logging_configured():
|
|
# If we get to this point it means the user hasn't
|
|
# configured logging. We want to log by default so
|
|
# we configure logging ourselves so that it will
|
|
# print to stderr.
|
|
config_logging()
|
|
|
|
if blocking:
|
|
# localize variable access to minimize overhead
|
|
poll = self.poll
|
|
socket_map = self.socket_map
|
|
sched_poll = self.sched.poll
|
|
|
|
if timeout is not None:
|
|
while socket_map:
|
|
poll(timeout)
|
|
sched_poll()
|
|
else:
|
|
soonest_timeout = None
|
|
while socket_map:
|
|
poll(soonest_timeout)
|
|
soonest_timeout = sched_poll()
|
|
else:
|
|
sched = self.sched
|
|
if self.socket_map:
|
|
self.poll(timeout)
|
|
if sched._tasks:
|
|
return sched.poll()
|
|
|
|
def call_later(self, seconds, target, *args, **kwargs):
|
|
kwargs['_scheduler'] = self.sched
|
|
return _CallLater(seconds, target, *args, **kwargs)
|
|
|
|
def call_every(self, seconds, target, *args, **kwargs):
|
|
kwargs['_scheduler'] = self.sched
|
|
return _CallEvery(seconds, target, *args, **kwargs)
|
|
|
|
def close(self):
|
|
debug("closing IOLoop", self)
|
|
self.__class__._instance = None
|
|
|
|
# free connections
|
|
instances = sorted(self.socket_map.values(), key=lambda x: x._fileno)
|
|
for inst in instances:
|
|
try:
|
|
inst.close()
|
|
except OSError as err:
|
|
if err.errno != errno.EBADF:
|
|
logger.error(traceback.format_exc())
|
|
except Exception:
|
|
logger.error(traceback.format_exc())
|
|
self.socket_map.clear()
|
|
|
|
# free scheduled functions
|
|
self.sched.close()
|
|
|
|
class Select(_IOLoop):
|
|
|
|
def __init__(self):
|
|
_IOLoop.__init__(self)
|
|
self._r = []
|
|
self._w = []
|
|
|
|
def register(self, fd, instance, events):
|
|
if fd not in self.socket_map:
|
|
self.socket_map[fd] = instance
|
|
if events & self.READ:
|
|
self._r.append(fd)
|
|
if events & self.WRITE:
|
|
self._w.append(fd)
|
|
|
|
def unregister(self, fd):
|
|
try:
|
|
del self.socket_map[fd]
|
|
except KeyError:
|
|
debug("call: unregister(); fd was no longer in socket_map", self)
|
|
for ls in (self._r, self._w):
|
|
try:
|
|
ls.remove(fd)
|
|
except ValueError:
|
|
pass
|
|
|
|
def modify(self, fd, events):
|
|
inst = self.socket_map.get(fd)
|
|
if inst is not None:
|
|
self.unregister(fd)
|
|
self.register(fd, inst, events)
|
|
else:
|
|
debug("call: modify(); fd was no longer in socket_map", self)
|
|
|
|
def poll(self, timeout):
|
|
try:
|
|
r, w, _ = select.select(self._r, self._w, [], timeout)
|
|
except InterruptedError:
|
|
return
|
|
|
|
smap_get = self.socket_map.get
|
|
for fd in r:
|
|
obj = smap_get(fd)
|
|
if obj is None or not obj.readable():
|
|
continue
|
|
_read(obj)
|
|
for fd in w:
|
|
obj = smap_get(fd)
|
|
if obj is None or not obj.writable():
|
|
continue
|
|
_write(obj)
|
|
|
|
class _BasePollEpoll(_IOLoop):
|
|
|
|
def __init__(self):
|
|
_IOLoop.__init__(self)
|
|
self._poller = self._poller()
|
|
|
|
def register(self, fd, instance, events):
|
|
try:
|
|
self._poller.register(fd, events)
|
|
except FileExistsError:
|
|
debug("call: register(); poller raised EEXIST; ignored", self)
|
|
self.socket_map[fd] = instance
|
|
|
|
def unregister(self, fd):
|
|
try:
|
|
del self.socket_map[fd]
|
|
except KeyError:
|
|
debug("call: unregister(); fd was no longer in socket_map", self)
|
|
else:
|
|
try:
|
|
self._poller.unregister(fd)
|
|
except OSError as err:
|
|
if err.errno in (errno.ENOENT, errno.EBADF):
|
|
debug(
|
|
f"call: unregister(); poller returned {err!r};"
|
|
" ignoring it",
|
|
self,
|
|
)
|
|
else:
|
|
raise
|
|
|
|
def modify(self, fd, events):
|
|
try:
|
|
self._poller.modify(fd, events)
|
|
except FileNotFoundError:
|
|
if fd in self.socket_map:
|
|
instance = self.socket_map[fd]
|
|
self.register(fd, instance, events)
|
|
else:
|
|
raise
|
|
|
|
def poll(self, timeout):
|
|
if timeout is None:
|
|
timeout = -1
|
|
try:
|
|
events = self._poller.poll(timeout)
|
|
except InterruptedError:
|
|
return
|
|
smap_get = self.socket_map.get
|
|
for fd, event in events:
|
|
inst = smap_get(fd)
|
|
if inst is None:
|
|
continue
|
|
if event & self._ERROR and not event & self.READ:
|
|
inst.handle_close()
|
|
else:
|
|
if event & self.READ and inst.readable():
|
|
_read(inst)
|
|
if event & self.WRITE and inst.writable():
|
|
_write(inst)
|
|
|
|
if hasattr(select, 'poll'):
|
|
|
|
class Poll(_BasePollEpoll):
|
|
|
|
READ = select.POLLIN
|
|
WRITE = select.POLLOUT
|
|
_ERROR = select.POLLERR | select.POLLHUP | select.POLLNVAL
|
|
_poller = select.poll
|
|
|
|
def modify(self, fd, events):
|
|
inst = self.socket_map[fd]
|
|
self.unregister(fd)
|
|
self.register(fd, inst, events)
|
|
|
|
def poll(self, timeout):
|
|
# poll() timeout is expressed in milliseconds
|
|
if timeout is not None:
|
|
timeout = int(timeout * 1000)
|
|
_BasePollEpoll.poll(self, timeout)
|
|
|
|
if hasattr(select, 'devpoll'):
|
|
|
|
class DevPoll(_BasePollEpoll):
|
|
|
|
READ = select.POLLIN
|
|
WRITE = select.POLLOUT
|
|
_ERROR = select.POLLERR | select.POLLHUP | select.POLLNVAL
|
|
_poller = select.devpoll
|
|
|
|
if hasattr(select.devpoll, 'fileno'):
|
|
|
|
def fileno(self):
|
|
return self._poller.fileno()
|
|
|
|
def modify(self, fd, events):
|
|
inst = self.socket_map[fd]
|
|
self.unregister(fd)
|
|
self.register(fd, inst, events)
|
|
|
|
def poll(self, timeout):
|
|
if timeout is not None:
|
|
timeout = int(timeout * 1000)
|
|
_BasePollEpoll.poll(self, timeout)
|
|
|
|
if hasattr(select.devpoll, 'close'):
|
|
|
|
def close(self):
|
|
_IOLoop.close(self)
|
|
self._poller.close()
|
|
|
|
if hasattr(select, 'epoll'):
|
|
|
|
class Epoll(_BasePollEpoll):
|
|
|
|
READ = select.EPOLLIN
|
|
WRITE = select.EPOLLOUT
|
|
_ERROR = select.EPOLLERR | select.EPOLLHUP
|
|
_poller = select.epoll
|
|
|
|
def fileno(self):
|
|
return self._poller.fileno()
|
|
|
|
def close(self):
|
|
_IOLoop.close(self)
|
|
self._poller.close()
|
|
|
|
if hasattr(select, 'kqueue'):
|
|
|
|
class Kqueue(_IOLoop):
|
|
|
|
def __init__(self):
|
|
_IOLoop.__init__(self)
|
|
self._kqueue = select.kqueue()
|
|
self._active = {}
|
|
|
|
def fileno(self):
|
|
return self._kqueue.fileno()
|
|
|
|
def close(self):
|
|
_IOLoop.close(self)
|
|
self._kqueue.close()
|
|
|
|
def register(self, fd, instance, events):
|
|
self.socket_map[fd] = instance
|
|
try:
|
|
self._control(fd, events, select.KQ_EV_ADD)
|
|
except FileExistsError:
|
|
debug("call: register(); poller raised EEXIST; ignored", self)
|
|
self._active[fd] = events
|
|
|
|
def unregister(self, fd):
|
|
try:
|
|
del self.socket_map[fd]
|
|
events = self._active.pop(fd)
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
try:
|
|
self._control(fd, events, select.KQ_EV_DELETE)
|
|
except OSError as err:
|
|
if err.errno in (errno.ENOENT, errno.EBADF):
|
|
debug(
|
|
f"call: unregister(); poller returned {err!r};"
|
|
" ignoring it",
|
|
self,
|
|
)
|
|
else:
|
|
raise
|
|
|
|
def modify(self, fd, events):
|
|
instance = self.socket_map[fd]
|
|
self.unregister(fd)
|
|
self.register(fd, instance, events)
|
|
|
|
def _control(self, fd, events, flags):
|
|
kevents = []
|
|
if events & self.WRITE:
|
|
kevents.append(
|
|
select.kevent(
|
|
fd, filter=select.KQ_FILTER_WRITE, flags=flags
|
|
)
|
|
)
|
|
if events & self.READ or not kevents:
|
|
# always read when there is not a write
|
|
kevents.append(
|
|
select.kevent(
|
|
fd, filter=select.KQ_FILTER_READ, flags=flags
|
|
)
|
|
)
|
|
# even though control() takes a list, it seems to return
|
|
# EINVAL on Mac OS X (10.6) when there is more than one
|
|
# event in the list
|
|
for kevent in kevents:
|
|
self._kqueue.control([kevent], 0)
|
|
|
|
# localize variable access to minimize overhead
|
|
def poll(
|
|
self,
|
|
timeout,
|
|
_len=len,
|
|
_READ=select.KQ_FILTER_READ,
|
|
_WRITE=select.KQ_FILTER_WRITE,
|
|
_EOF=select.KQ_EV_EOF,
|
|
_ERROR=select.KQ_EV_ERROR,
|
|
):
|
|
try:
|
|
kevents = self._kqueue.control(
|
|
None, _len(self.socket_map), timeout
|
|
)
|
|
except InterruptedError:
|
|
return
|
|
for kevent in kevents:
|
|
inst = self.socket_map.get(kevent.ident)
|
|
if inst is None:
|
|
continue
|
|
if kevent.filter == _READ and inst.readable():
|
|
_read(inst)
|
|
if kevent.filter == _WRITE:
|
|
if kevent.flags & _EOF:
|
|
# If an asynchronous connection is refused,
|
|
# kqueue returns a write event with the EOF
|
|
# flag set.
|
|
# Note that for read events, EOF may be returned
|
|
# before all data has been consumed from the
|
|
# socket buffer, so we only check for EOF on
|
|
# write events.
|
|
inst.handle_close()
|
|
elif inst.writable():
|
|
_write(inst)
|
|
if kevent.flags & _ERROR:
|
|
inst.handle_close()
|
|
|
|
|
|
# ===================================================================
|
|
# --- choose the better poller for this platform
|
|
# ===================================================================
|
|
|
|
if hasattr(select, 'epoll'): # epoll() - Linux
|
|
IOLoop = Epoll
|
|
elif hasattr(select, 'kqueue'): # kqueue() - BSD / OSX
|
|
IOLoop = Kqueue
|
|
elif hasattr(select, 'devpoll'): # /dev/poll - Solaris
|
|
IOLoop = DevPoll
|
|
elif hasattr(select, 'poll'): # poll() - POSIX
|
|
IOLoop = Poll
|
|
else: # select() - POSIX and Windows
|
|
IOLoop = Select
|
|
|
|
|
|
class AsyncChat(asynchat.async_chat):
|
|
|
|
|
|
def __init__(self, sock=None, ioloop=None):
|
|
self.ioloop = ioloop or IOLoop.instance()
|
|
self._wanted_io_events = self.ioloop.READ
|
|
self._current_io_events = self.ioloop.READ
|
|
self._closed = False
|
|
self._closing = False
|
|
self._fileno = sock.fileno() if sock else None
|
|
self._tasks = []
|
|
asynchat.async_chat.__init__(self, sock)
|
|
|
|
# --- IO loop related methods
|
|
|
|
def add_channel(self, map=None, events=None):
|
|
assert self._fileno, repr(self._fileno)
|
|
events = events if events is not None else self.ioloop.READ
|
|
self.ioloop.register(self._fileno, self, events)
|
|
self._wanted_io_events = events
|
|
self._current_io_events = events
|
|
|
|
def del_channel(self, map=None):
|
|
if self._fileno is not None:
|
|
self.ioloop.unregister(self._fileno)
|
|
|
|
def modify_ioloop_events(self, events, logdebug=False):
|
|
if not self._closed:
|
|
assert self._fileno, repr(self._fileno)
|
|
if self._fileno not in self.ioloop.socket_map:
|
|
debug(
|
|
"call: modify_ioloop_events(), fd was no longer in "
|
|
"socket_map, had to register() it again",
|
|
inst=self,
|
|
)
|
|
self.add_channel(events=events)
|
|
elif events != self._current_io_events:
|
|
if logdebug:
|
|
if events == self.ioloop.READ:
|
|
ev = "R"
|
|
elif events == self.ioloop.WRITE:
|
|
ev = "W"
|
|
elif events == self.ioloop.READ | self.ioloop.WRITE:
|
|
ev = "RW"
|
|
else:
|
|
ev = events
|
|
debug(
|
|
f"call: IOLoop.modify(); setting {ev!r} IO events",
|
|
self,
|
|
)
|
|
self.ioloop.modify(self._fileno, events)
|
|
self._current_io_events = events
|
|
else:
|
|
debug(
|
|
"call: modify_ioloop_events(), handler had already been "
|
|
"close()d, skipping modify()",
|
|
inst=self,
|
|
)
|
|
|
|
def call_later(self, seconds, target, *args, **kwargs):
|
|
if '_errback' not in kwargs and hasattr(self, 'handle_error'):
|
|
kwargs['_errback'] = self.handle_error
|
|
callback = self.ioloop.call_later(seconds, target, *args, **kwargs)
|
|
self._tasks.append(callback)
|
|
return callback
|
|
|
|
def connect(self, addr):
|
|
self.modify_ioloop_events(self.ioloop.WRITE)
|
|
asynchat.async_chat.connect(self, addr)
|
|
|
|
def connect_af_unspecified(self, addr, source_address=None):
|
|
assert self.socket is None
|
|
host, port = addr
|
|
err = "getaddrinfo() returned an empty list"
|
|
info = socket.getaddrinfo(
|
|
host,
|
|
port,
|
|
socket.AF_UNSPEC,
|
|
socket.SOCK_STREAM,
|
|
0,
|
|
socket.AI_PASSIVE,
|
|
)
|
|
for res in info:
|
|
self.socket = None
|
|
af, socktype, _proto, _canonname, _sa = res
|
|
try:
|
|
self.create_socket(af, socktype)
|
|
if source_address:
|
|
if source_address[0].startswith('::ffff:'):
|
|
source_address = (
|
|
source_address[0][7:],
|
|
source_address[1],
|
|
)
|
|
self.bind(source_address)
|
|
self.connect((host, port))
|
|
except OSError as _:
|
|
err = _
|
|
if self.socket is not None:
|
|
self.socket.close()
|
|
self.del_channel()
|
|
self.socket = None
|
|
continue
|
|
break
|
|
if self.socket is None:
|
|
self.del_channel()
|
|
raise OSError(err)
|
|
return af
|
|
|
|
def send(self, data):
|
|
try:
|
|
return self.socket.send(data)
|
|
except OSError as err:
|
|
debug(f"call: send(), err: {err}", inst=self)
|
|
if err.errno in _ERRNOS_RETRY:
|
|
return 0
|
|
elif err.errno in _ERRNOS_DISCONNECTED:
|
|
self.handle_close()
|
|
return 0
|
|
else:
|
|
raise
|
|
|
|
def recv(self, buffer_size):
|
|
try:
|
|
data = self.socket.recv(buffer_size)
|
|
except OSError as err:
|
|
debug(f"call: recv(), err: {err}", inst=self)
|
|
if err.errno in _ERRNOS_DISCONNECTED:
|
|
self.handle_close()
|
|
return b''
|
|
elif err.errno in _ERRNOS_RETRY:
|
|
raise RetryError
|
|
else:
|
|
raise
|
|
else:
|
|
if not data:
|
|
# a closed connection is indicated by signaling
|
|
# a read condition, and having recv() return 0.
|
|
self.handle_close()
|
|
return b''
|
|
else:
|
|
return data
|
|
|
|
def handle_read(self):
|
|
try:
|
|
asynchat.async_chat.handle_read(self)
|
|
except RetryError:
|
|
# This can be raised by (the overridden) recv().
|
|
pass
|
|
|
|
def initiate_send(self):
|
|
asynchat.async_chat.initiate_send(self)
|
|
if not self._closed:
|
|
# if there's still data to send we want to be ready
|
|
# for writing, else we're only interested in reading
|
|
if not self.producer_fifo:
|
|
wanted = self.ioloop.READ
|
|
else:
|
|
# In FTPHandler, we also want to listen for user input
|
|
# hence the READ. DTPHandler has its own initiate_send()
|
|
# which will either READ or WRITE.
|
|
wanted = self.ioloop.READ | self.ioloop.WRITE
|
|
if self._wanted_io_events != wanted:
|
|
self.ioloop.modify(self._fileno, wanted)
|
|
self._wanted_io_events = wanted
|
|
else:
|
|
debug(
|
|
"call: initiate_send(); called with no connection", inst=self
|
|
)
|
|
|
|
def close_when_done(self):
|
|
if len(self.producer_fifo) == 0:
|
|
self.handle_close()
|
|
else:
|
|
self._closing = True
|
|
asynchat.async_chat.close_when_done(self)
|
|
|
|
def close(self):
|
|
if not self._closed:
|
|
self._closed = True
|
|
try:
|
|
asynchat.async_chat.close(self)
|
|
finally:
|
|
for fun in self._tasks:
|
|
try:
|
|
fun.cancel()
|
|
except Exception:
|
|
logger.error(traceback.format_exc())
|
|
self._tasks = []
|
|
self._closed = True
|
|
self._closing = False
|
|
self.connected = False
|
|
|
|
|
|
class Connector(AsyncChat):
|
|
def add_channel(self, map=None, events=None):
|
|
AsyncChat.add_channel(self, map=map, events=self.ioloop.WRITE)
|
|
|
|
|
|
class Acceptor(AsyncChat):
|
|
def add_channel(self, map=None, events=None):
|
|
AsyncChat.add_channel(self, map=map, events=self.ioloop.READ)
|
|
|
|
def bind_af_unspecified(self, addr):
|
|
assert self.socket is None
|
|
host, port = addr
|
|
if not host:
|
|
host = None
|
|
err = "getaddrinfo() returned an empty list"
|
|
info = socket.getaddrinfo(
|
|
host,
|
|
port,
|
|
socket.AF_UNSPEC,
|
|
socket.SOCK_STREAM,
|
|
0,
|
|
socket.AI_PASSIVE,
|
|
)
|
|
for res in info:
|
|
self.socket = None
|
|
self.del_channel()
|
|
af, socktype, _proto, _canonname, sa = res
|
|
try:
|
|
self.create_socket(af, socktype)
|
|
self.set_reuse_addr()
|
|
self.bind(sa)
|
|
except OSError as _:
|
|
err = _
|
|
if self.socket is not None:
|
|
self.socket.close()
|
|
self.del_channel()
|
|
self.socket = None
|
|
continue
|
|
break
|
|
if self.socket is None:
|
|
self.del_channel()
|
|
raise OSError(err)
|
|
return af
|
|
|
|
def listen(self, num):
|
|
AsyncChat.listen(self, num)
|
|
try:
|
|
if isinstance(self.ioloop, Kqueue):
|
|
self.ioloop.modify(self._fileno, self.ioloop.READ)
|
|
except NameError:
|
|
pass
|
|
|
|
def handle_accept(self):
|
|
try:
|
|
sock, addr = self.accept()
|
|
except TypeError:
|
|
debug("call: handle_accept(); accept() returned None", self)
|
|
return
|
|
except OSError as err:
|
|
if err.errno != errno.ECONNABORTED:
|
|
raise
|
|
else:
|
|
debug(
|
|
"call: handle_accept(); accept() returned ECONNABORTED",
|
|
self,
|
|
)
|
|
else:
|
|
if addr is not None:
|
|
self.handle_accepted(sock, addr)
|
|
|
|
def handle_accepted(self, sock, addr):
|
|
sock.close()
|
|
self.log_info('unhandled accepted event', 'warning')
|
|
|
|
if (os.name in ('nt', 'ce')) or (sys.platform == 'cygwin'):
|
|
|
|
def set_reuse_addr(self):
|
|
pass
|