This module allows high-level and efficient I/O multiplexing, built upon the
`select` module primitives.
from abc import ABCMeta, abstractmethod
from collections import namedtuple
from collections.abc import Mapping
# generic events, that must be mapped to implementation-specific ones
def _fileobj_to_fd(fileobj):
"""Return a file descriptor from a file object.
fileobj -- file object or file descriptor
corresponding file descriptor
ValueError if the object is invalid
if isinstance(fileobj, int):
fd = int(fileobj.fileno())
except (AttributeError, TypeError, ValueError):
raise ValueError("Invalid file object: "
"{!r}".format(fileobj)) from None
raise ValueError("Invalid file descriptor: {}".format(fd))
SelectorKey = namedtuple('SelectorKey', ['fileobj', 'fd', 'events', 'data'])
SelectorKey.__doc__ = """SelectorKey(fileobj, fd, events, data)
Object used to associate a file object to its backing
file descriptor, selected event mask, and attached data.
if sys.version_info >= (3, 5):
SelectorKey.fileobj.__doc__ = 'File object registered.'
SelectorKey.fd.__doc__ = 'Underlying file descriptor.'
SelectorKey.events.__doc__ = 'Events that must be waited for on this file object.'
SelectorKey.data.__doc__ = ('''Optional opaque data associated to this file object.
For example, this could be used to store a per-client session ID.''')
class _SelectorMapping(Mapping):
"""Mapping of file objects to selector keys."""
def __init__(self, selector):
self._selector = selector
return len(self._selector._fd_to_key)
def __getitem__(self, fileobj):
fd = self._selector._fileobj_lookup(fileobj)
return self._selector._fd_to_key[fd]
raise KeyError("{!r} is not registered".format(fileobj)) from None
return iter(self._selector._fd_to_key)
class BaseSelector(metaclass=ABCMeta):
"""Selector abstract base class.
A selector supports registering file objects to be monitored for specific
A file object is a file descriptor or any object with a `fileno()` method.
An arbitrary object can be attached to the file object, which can be used
for example to store context information, a callback, etc.
A selector can use various implementations (select(), poll(), epoll()...)
depending on the platform. The default `Selector` class uses the most
efficient implementation on the current platform.
def register(self, fileobj, events, data=None):
"""Register a file object.
fileobj -- file object or file descriptor
events -- events to monitor (bitwise mask of EVENT_READ|EVENT_WRITE)
ValueError if events is invalid
KeyError if fileobj is already registered
OSError if fileobj is closed or otherwise is unacceptable to
the underlying system call (if a system call is made)
OSError may or may not be raised
raise NotImplementedError
def unregister(self, fileobj):
"""Unregister a file object.
fileobj -- file object or file descriptor
KeyError if fileobj is not registered
If fileobj is registered but has since been closed this does
*not* raise OSError (even if the wrapped syscall does)
raise NotImplementedError
def modify(self, fileobj, events, data=None):
"""Change a registered file object monitored events or attached data.
fileobj -- file object or file descriptor
events -- events to monitor (bitwise mask of EVENT_READ|EVENT_WRITE)
Anything that unregister() or register() raises
return self.register(fileobj, events, data)
def select(self, timeout=None):
"""Perform the actual selection, until some monitored file objects are
ready or a timeout expires.
timeout -- if timeout > 0, this specifies the maximum wait time, in
if timeout <= 0, the select() call won't block, and will
report the currently ready file objects
if timeout is None, select() will block until a monitored
file object becomes ready
list of (key, events) for ready file objects
`events` is a bitwise mask of EVENT_READ|EVENT_WRITE
raise NotImplementedError
This must be called to make sure that any underlying resource is freed.
def get_key(self, fileobj):
"""Return the key associated to a registered file object.
SelectorKey for this file object
raise RuntimeError('Selector is closed')
raise KeyError("{!r} is not registered".format(fileobj)) from None
"""Return a mapping of file objects to selector keys."""
raise NotImplementedError
def __exit__(self, *args):
class _BaseSelectorImpl(BaseSelector):
"""Base selector implementation."""
# this maps file descriptors to keys
# read-only mapping returned by get_map()
self._map = _SelectorMapping(self)
def _fileobj_lookup(self, fileobj):
"""Return a file descriptor from a file object.
This wraps _fileobj_to_fd() to do an exhaustive search in case
the object is invalid but we still have it in our map. This
is used by unregister() so we can unregister an object that
was previously registered even if it is closed. It is also
used by _SelectorMapping.
return _fileobj_to_fd(fileobj)
# Do an exhaustive search.
for key in self._fd_to_key.values():
if key.fileobj is fileobj:
# Raise ValueError after all.
def register(self, fileobj, events, data=None):
if (not events) or (events & ~(EVENT_READ | EVENT_WRITE)):
raise ValueError("Invalid events: {!r}".format(events))
key = SelectorKey(fileobj, self._fileobj_lookup(fileobj), events, data)
if key.fd in self._fd_to_key:
raise KeyError("{!r} (FD {}) is already registered"
.format(fileobj, key.fd))
self._fd_to_key[key.fd] = key
def unregister(self, fileobj):
key = self._fd_to_key.pop(self._fileobj_lookup(fileobj))
raise KeyError("{!r} is not registered".format(fileobj)) from None
def modify(self, fileobj, events, data=None):
key = self._fd_to_key[self._fileobj_lookup(fileobj)]
raise KeyError("{!r} is not registered".format(fileobj)) from None
key = self.register(fileobj, events, data)
# Use a shortcut to update the data.
key = key._replace(data=data)
self._fd_to_key[key.fd] = key
def _key_from_fd(self, fd):
"""Return the key associated to a given file descriptor.
corresponding key, or None if not found
return self._fd_to_key[fd]
class SelectSelector(_BaseSelectorImpl):
"""Select-based selector."""
def register(self, fileobj, events, data=None):
key = super().register(fileobj, events, data)
self._readers.add(key.fd)
self._writers.add(key.fd)
def unregister(self, fileobj):
key = super().unregister(fileobj)
self._readers.discard(key.fd)
self._writers.discard(key.fd)
if sys.platform == 'win32':
def _select(self, r, w, _, timeout=None):
r, w, x = select.select(r, w, w, timeout)
def select(self, timeout=None):
timeout = None if timeout is None else max(timeout, 0)
r, w, _ = self._select(self._readers, self._writers, [], timeout)
key = self._key_from_fd(fd)
ready.append((key, events & key.events))
class _PollLikeSelector(_BaseSelectorImpl):
"""Base class shared between poll, epoll and devpoll selectors."""
self._selector = self._selector_cls()
def register(self, fileobj, events, data=None):
key = super().register(fileobj, events, data)
poller_events |= self._EVENT_READ
poller_events |= self._EVENT_WRITE
self._selector.register(key.fd, poller_events)
super().unregister(fileobj)
def unregister(self, fileobj):
key = super().unregister(fileobj)
self._selector.unregister(key.fd)
# This can happen if the FD was closed since it
def modify(self, fileobj, events, data=None):
key = self._fd_to_key[self._fileobj_lookup(fileobj)]
raise KeyError(f"{fileobj!r} is not registered") from None
selector_events |= self._EVENT_READ
selector_events |= self._EVENT_WRITE
self._selector.modify(key.fd, selector_events)
super().unregister(fileobj)
key = key._replace(events=events, data=data)
self._fd_to_key[key.fd] = key
def select(self, timeout=None):
# This is shared between poll() and epoll().
# epoll() has a different signature and handling of timeout parameter.
# poll() has a resolution of 1 millisecond, round away from
# zero to wait *at least* timeout seconds.
timeout = math.ceil(timeout * 1e3)
fd_event_list = self._selector.poll(timeout)
for fd, event in fd_event_list:
if event & ~self._EVENT_READ:
if event & ~self._EVENT_WRITE:
key = self._key_from_fd(fd)
ready.append((key, events & key.events))
if hasattr(select, 'poll'):
class PollSelector(_PollLikeSelector):
"""Poll-based selector."""
_selector_cls = select.poll
_EVENT_READ = select.POLLIN
_EVENT_WRITE = select.POLLOUT
if hasattr(select, 'epoll'):
class EpollSelector(_PollLikeSelector):
"""Epoll-based selector."""
_selector_cls = select.epoll
_EVENT_READ = select.EPOLLIN
_EVENT_WRITE = select.EPOLLOUT
return self._selector.fileno()
def select(self, timeout=None):
# epoll_wait() has a resolution of 1 millisecond, round away
# from zero to wait *at least* timeout seconds.
timeout = math.ceil(timeout * 1e3) * 1e-3
# epoll_wait() expects `maxevents` to be greater than zero;
# we want to make sure that `select()` can be called when no
max_ev = max(len(self._fd_to_key), 1)
fd_event_list = self._selector.poll(timeout, max_ev)
for fd, event in fd_event_list:
if event & ~select.EPOLLIN:
if event & ~select.EPOLLOUT:
key = self._key_from_fd(fd)
ready.append((key, events & key.events))
if hasattr(select, 'devpoll'):
class DevpollSelector(_PollLikeSelector):
"""Solaris /dev/poll selector."""
_selector_cls = select.devpoll
_EVENT_READ = select.POLLIN
_EVENT_WRITE = select.POLLOUT
return self._selector.fileno()
It is recommended that you Edit text format, this type of Fix handles quite a lot in one request
