Issue #16853: Add new selectors module.
This commit is contained in:
Charles-François Natali
parent
af722bf9cf
commit
243d8d85de
|
|
@ -21,6 +21,7 @@ multitasking). Here's an overview:
|
|||
sched.rst
|
||||
queue.rst
|
||||
select.rst
|
||||
selectors.rst
|
||||
|
||||
|
||||
The following are support modules for some of the above services:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,231 @@
|
|||
:mod:`selectors` -- High-level I/O multiplexing
|
||||
===============================================
|
||||
|
||||
.. module:: selectors
|
||||
:synopsis: High-level I/O multiplexing.
|
||||
|
||||
.. versionadded:: 3.4
|
||||
|
||||
|
||||
Introduction
|
||||
------------
|
||||
|
||||
This module allows high-level and efficient I/O multiplexing, built upon the
|
||||
:mod:`select` module primitives. Users are encouraged to use this module
|
||||
instead, unless they want precise control over the OS-level primitives used.
|
||||
|
||||
It defines a :class:`BaseSelector` abstract base class, along with several
|
||||
concrete implementations (:class:`KqueueSelector`, :class:`EpollSelector`...),
|
||||
that can be used to wait for I/O readiness notification on multiple file
|
||||
objects. In the following, "file object" refers to any object with a
|
||||
:meth:`fileno()` method, or a raw file descriptor. See :term:`file object`.
|
||||
|
||||
:class:`DefaultSelector` is an alias to the most efficient implementation
|
||||
available on the current platform: this should be the default choice for most
|
||||
users.
|
||||
|
||||
.. note::
|
||||
The type of file objects supported depends on the platform: on Windows,
|
||||
sockets are supported, but not pipes, whereas on Unix, both are supported
|
||||
(some other types may be supported as well, such as fifos or special file
|
||||
devices).
|
||||
|
||||
.. seealso::
|
||||
|
||||
:mod:`select`
|
||||
Low-level I/O multiplexing module.
|
||||
|
||||
|
||||
Classes
|
||||
-------
|
||||
|
||||
Classes hierarchy::
|
||||
|
||||
BaseSelector
|
||||
+-- SelectSelector
|
||||
+-- PollSelector
|
||||
+-- EpollSelector
|
||||
+-- KqueueSelector
|
||||
|
||||
|
||||
In the following, *events* is a bitwise mask indicating which I/O events should
|
||||
be waited for on a given file object. It can be a combination of the constants
|
||||
below:
|
||||
|
||||
+-----------------------+-----------------------------------------------+
|
||||
| Constant | Meaning |
|
||||
+=======================+===============================================+
|
||||
| :const:`EVENT_READ` | Available for read |
|
||||
+-----------------------+-----------------------------------------------+
|
||||
| :const:`EVENT_WRITE` | Available for write |
|
||||
+-----------------------+-----------------------------------------------+
|
||||
|
||||
|
||||
.. class:: SelectorKey
|
||||
|
||||
A :class:`SelectorKey` is a :class:`~collections.namedtuple` used to
|
||||
associate a file object to its underlying file decriptor, selected event
|
||||
mask and attached data. It is returned by several :class:`BaseSelector`
|
||||
methods.
|
||||
|
||||
.. attribute:: fileobj
|
||||
|
||||
File object registered.
|
||||
|
||||
.. attribute:: fd
|
||||
|
||||
Underlying file descriptor.
|
||||
|
||||
.. attribute:: events
|
||||
|
||||
Events that must be waited for this file object.
|
||||
|
||||
.. attribute:: data
|
||||
|
||||
Optional opaque data associated to this file object: for example, this
|
||||
could be used to store per-client session.
|
||||
|
||||
|
||||
.. class:: BaseSelector
|
||||
|
||||
A :class:`BaseSelector` is used to wait for I/O event readiness on multiple
|
||||
file objects. It supports file stream registration, unregistration, and a
|
||||
method to wait for I/O events on those streams, with an optional timeout.
|
||||
It's an abstract base class, so cannot be instantiated. Use
|
||||
:class:`DefaultSelector` instead, or one of :class:`SelectSelector`,
|
||||
:class:`KqueueSelector` etc. if you want to specifically use an
|
||||
implementation, and your platform supports it.
|
||||
:class:`BaseSelector` and its concrete implementations support the
|
||||
:term:`context manager` protocol.
|
||||
|
||||
.. method:: register(fileobj, events, data=None)
|
||||
|
||||
Register a file object for selection, monitoring it for I/O events.
|
||||
|
||||
*fileobj* is the file object to monitor.
|
||||
*events* is a bitwise mask of events to monitor.
|
||||
*data* is an opaque object.
|
||||
|
||||
This returns a new :class:`SelectorKey` instance, or raises a
|
||||
:exc:`ValueError` in case of invalid event mask or file descriptor, or
|
||||
:exc:`KeyError` if the file object is already registered.
|
||||
|
||||
.. method:: unregister(fileobj)
|
||||
|
||||
Unregister a file object from selection, removing it from monitoring. A
|
||||
file object shall be unregistered prior to being closed.
|
||||
|
||||
*fileobj* must be a file object previously registered.
|
||||
|
||||
This returns the associated :class:`SelectorKey` instance, or raises a
|
||||
:exc:`KeyError` if the file object is not registered.
|
||||
|
||||
.. method:: modify(fileobj, events, data=None)
|
||||
|
||||
Change a registered file object monitored events or attached data.
|
||||
|
||||
This is equivalent to :meth:`BaseSelector.unregister(fileobj)` followed
|
||||
by :meth:`BaseSelector.register(fileobj, events, data)`, except that it
|
||||
can be implemented more efficiently.
|
||||
|
||||
This returns a new :class:`SelectorKey` instance, or raises a
|
||||
:exc:`ValueError` in case of invalid event mask or file descriptor, or
|
||||
:exc:`KeyError` if the file object is not registered.
|
||||
|
||||
.. method:: select(timeout=None)
|
||||
|
||||
Wait until some registered file objects become ready, or the timeout
|
||||
expires.
|
||||
|
||||
If ``timeout > 0``, this specifies the maximum wait time, in seconds.
|
||||
If ``timeout <= 0``, the call won't block, and will report the currently
|
||||
ready file objects.
|
||||
If *timeout* is ``None``, the call will block until a monitored file object
|
||||
becomes ready.
|
||||
|
||||
This returns a list of ``(key, events)`` tuple, one for each ready file
|
||||
object.
|
||||
|
||||
*key* is the :class:`SelectorKey` instance corresponding to a ready file
|
||||
object.
|
||||
*events* is a bitmask of events ready on this file object.
|
||||
|
||||
.. method:: close()
|
||||
|
||||
Close the selector.
|
||||
|
||||
This must be called to make sure that any underlying resource is freed.
|
||||
The selector shall not be used once it has been closed.
|
||||
|
||||
.. method:: get_key(fileobj)
|
||||
|
||||
Return the key associated to a registered file object.
|
||||
|
||||
This returns the :class:`SelectorKey` instance associated to this file
|
||||
object, or raises :exc:`KeyError` if the file object is not registered.
|
||||
|
||||
|
||||
.. class:: DefaultSelector()
|
||||
|
||||
The default selector class, using the most efficient implementation
|
||||
available on the current platform. This should be the default choice for
|
||||
most users.
|
||||
|
||||
|
||||
.. class:: SelectSelector()
|
||||
|
||||
:func:`select.select`-based selector.
|
||||
|
||||
|
||||
.. class:: PollSelector()
|
||||
|
||||
:func:`select.poll`-based selector.
|
||||
|
||||
|
||||
.. class:: EpollSelector()
|
||||
|
||||
:func:`select.epoll`-based selector.
|
||||
|
||||
.. method:: fileno()
|
||||
|
||||
This returns the file descriptor used by the underlying
|
||||
:func:`select.epoll` object.
|
||||
|
||||
|
||||
.. class:: KqueueSelector()
|
||||
|
||||
:func:`select.kqueue`-based selector.
|
||||
|
||||
.. method:: fileno()
|
||||
|
||||
This returns the file descriptor used by the underlying
|
||||
:func:`select.kqueue` object.
|
||||
|
||||
|
||||
Examples of selector usage::
|
||||
|
||||
>>> import selectors
|
||||
>>> import socket
|
||||
>>>
|
||||
>>> s = selectors.DefaultSelector()
|
||||
>>> r, w = socket.socketpair()
|
||||
>>>
|
||||
>>> s.register(r, selectors.EVENT_READ)
|
||||
SelectorKey(fileobj=<socket.socket fd=4, family=1, type=1, proto=0>, fd=4, events=1, data=None)
|
||||
>>> s.register(w, selectors.EVENT_WRITE)
|
||||
SelectorKey(fileobj=<socket.socket fd=5, family=1, type=1, proto=0>, fd=5, events=2, data=None)
|
||||
>>>
|
||||
>>> print(s.select())
|
||||
[(SelectorKey(fileobj=<socket.socket fd=5, family=1, type=1, proto=0>, fd=5, events=2, data=None), 2)]
|
||||
>>>
|
||||
>>> for key, events in s.select():
|
||||
... if events & selectors.EVENT_WRITE:
|
||||
... key.fileobj.send(b'spam')
|
||||
...
|
||||
4
|
||||
>>> for key, events in s.select():
|
||||
... if events & selectors.EVENT_READ:
|
||||
... print(key.fileobj.recv(1024))
|
||||
...
|
||||
b'spam'
|
||||
>>> s.close()
|
||||
|
|
@ -174,10 +174,11 @@ Some smaller changes made to the core Python language are:
|
|||
New Modules
|
||||
===========
|
||||
|
||||
.. module name
|
||||
.. -----------
|
||||
selectors
|
||||
---------
|
||||
|
||||
* None yet.
|
||||
The new :mod:`selectors` module allows high-level and efficient I/O
|
||||
multiplexing, built upon the :mod:`select` module primitives.
|
||||
|
||||
|
||||
Improved Modules
|
||||
|
|
|
|||
|
|
@ -0,0 +1,405 @@
|
|||
"""Selectors module.
|
||||
|
||||
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
|
||||
import functools
|
||||
import select
|
||||
import sys
|
||||
|
||||
|
||||
# generic events, that must be mapped to implementation-specific ones
|
||||
EVENT_READ = (1 << 0)
|
||||
EVENT_WRITE = (1 << 1)
|
||||
|
||||
|
||||
def _fileobj_to_fd(fileobj):
|
||||
"""Return a file descriptor from a file object.
|
||||
|
||||
Parameters:
|
||||
fileobj -- file object or file descriptor
|
||||
|
||||
Returns:
|
||||
corresponding file descriptor
|
||||
"""
|
||||
if isinstance(fileobj, int):
|
||||
fd = fileobj
|
||||
else:
|
||||
try:
|
||||
fd = int(fileobj.fileno())
|
||||
except (AttributeError, TypeError, ValueError):
|
||||
raise ValueError("Invalid file object: "
|
||||
"{!r}".format(fileobj)) from None
|
||||
if fd < 0:
|
||||
raise ValueError("Invalid file descriptor: {}".format(fd))
|
||||
return fd
|
||||
|
||||
|
||||
SelectorKey = namedtuple('SelectorKey', ['fileobj', 'fd', 'events', 'data'])
|
||||
"""Object used to associate a file object to its backing file descriptor,
|
||||
selected event mask and attached data."""
|
||||
|
||||
|
||||
class BaseSelector(metaclass=ABCMeta):
|
||||
"""Base selector class.
|
||||
|
||||
A selector supports registering file objects to be monitored for specific
|
||||
I/O events.
|
||||
|
||||
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
|
||||
performant implementation on the current platform.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
# this maps file descriptors to keys
|
||||
self._fd_to_key = {}
|
||||
|
||||
def register(self, fileobj, events, data=None):
|
||||
"""Register a file object.
|
||||
|
||||
Parameters:
|
||||
fileobj -- file object or file descriptor
|
||||
events -- events to monitor (bitwise mask of EVENT_READ|EVENT_WRITE)
|
||||
data -- attached data
|
||||
|
||||
Returns:
|
||||
SelectorKey instance
|
||||
"""
|
||||
if (not events) or (events & ~(EVENT_READ | EVENT_WRITE)):
|
||||
raise ValueError("Invalid events: {!r}".format(events))
|
||||
|
||||
key = SelectorKey(fileobj, _fileobj_to_fd(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
|
||||
return key
|
||||
|
||||
def unregister(self, fileobj):
|
||||
"""Unregister a file object.
|
||||
|
||||
Parameters:
|
||||
fileobj -- file object or file descriptor
|
||||
|
||||
Returns:
|
||||
SelectorKey instance
|
||||
"""
|
||||
try:
|
||||
key = self._fd_to_key.pop(_fileobj_to_fd(fileobj))
|
||||
except KeyError:
|
||||
raise KeyError("{!r} is not registered".format(fileobj)) from None
|
||||
return key
|
||||
|
||||
def modify(self, fileobj, events, data=None):
|
||||
"""Change a registered file object monitored events or attached data.
|
||||
|
||||
Parameters:
|
||||
fileobj -- file object or file descriptor
|
||||
events -- events to monitor (bitwise mask of EVENT_READ|EVENT_WRITE)
|
||||
data -- attached data
|
||||
|
||||
Returns:
|
||||
SelectorKey instance
|
||||
"""
|
||||
# TODO: Subclasses can probably optimize this even further.
|
||||
try:
|
||||
key = self._fd_to_key[_fileobj_to_fd(fileobj)]
|
||||
except KeyError:
|
||||
raise KeyError("{!r} is not registered".format(fileobj)) from None
|
||||
if events != key.events or data != key.data:
|
||||
# TODO: If only the data changed, use a shortcut that only
|
||||
# updates the data.
|
||||
self.unregister(fileobj)
|
||||
return self.register(fileobj, events, data)
|
||||
else:
|
||||
return key
|
||||
|
||||
@abstractmethod
|
||||
def select(self, timeout=None):
|
||||
"""Perform the actual selection, until some monitored file objects are
|
||||
ready or a timeout expires.
|
||||
|
||||
Parameters:
|
||||
timeout -- if timeout > 0, this specifies the maximum wait time, in
|
||||
seconds
|
||||
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
|
||||
|
||||
Returns:
|
||||
list of (key, events) for ready file objects
|
||||
`events` is a bitwise mask of EVENT_READ|EVENT_WRITE
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
def close(self):
|
||||
"""Close the selector.
|
||||
|
||||
This must be called to make sure that any underlying resource is freed.
|
||||
"""
|
||||
self._fd_to_key.clear()
|
||||
|
||||
def get_key(self, fileobj):
|
||||
"""Return the key associated to a registered file object.
|
||||
|
||||
Returns:
|
||||
SelectorKey for this file object
|
||||
"""
|
||||
try:
|
||||
return self._fd_to_key[_fileobj_to_fd(fileobj)]
|
||||
except KeyError:
|
||||
raise KeyError("{!r} is not registered".format(fileobj)) from None
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *args):
|
||||
self.close()
|
||||
|
||||
def _key_from_fd(self, fd):
|
||||
"""Return the key associated to a given file descriptor.
|
||||
|
||||
Parameters:
|
||||
fd -- file descriptor
|
||||
|
||||
Returns:
|
||||
corresponding key, or None if not found
|
||||
"""
|
||||
try:
|
||||
return self._fd_to_key[fd]
|
||||
except KeyError:
|
||||
return None
|
||||
|
||||
|
||||
class SelectSelector(BaseSelector):
|
||||
"""Select-based selector."""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self._readers = set()
|
||||
self._writers = set()
|
||||
|
||||
def register(self, fileobj, events, data=None):
|
||||
key = super().register(fileobj, events, data)
|
||||
if events & EVENT_READ:
|
||||
self._readers.add(key.fd)
|
||||
if events & EVENT_WRITE:
|
||||
self._writers.add(key.fd)
|
||||
return key
|
||||
|
||||
def unregister(self, fileobj):
|
||||
key = super().unregister(fileobj)
|
||||
self._readers.discard(key.fd)
|
||||
self._writers.discard(key.fd)
|
||||
return key
|
||||
|
||||
if sys.platform == 'win32':
|
||||
def _select(self, r, w, _, timeout=None):
|
||||
r, w, x = select.select(r, w, w, timeout)
|
||||
return r, w + x, []
|
||||
else:
|
||||
_select = select.select
|
||||
|
||||
def select(self, timeout=None):
|
||||
timeout = None if timeout is None else max(timeout, 0)
|
||||
ready = []
|
||||
try:
|
||||
r, w, _ = self._select(self._readers, self._writers, [], timeout)
|
||||
except InterruptedError:
|
||||
return ready
|
||||
r = set(r)
|
||||
w = set(w)
|
||||
for fd in r | w:
|
||||
events = 0
|
||||
if fd in r:
|
||||
events |= EVENT_READ
|
||||
if fd in w:
|
||||
events |= EVENT_WRITE
|
||||
|
||||
key = self._key_from_fd(fd)
|
||||
if key:
|
||||
ready.append((key, events & key.events))
|
||||
return ready
|
||||
|
||||
|
||||
if hasattr(select, 'poll'):
|
||||
|
||||
class PollSelector(BaseSelector):
|
||||
"""Poll-based selector."""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self._poll = select.poll()
|
||||
|
||||
def register(self, fileobj, events, data=None):
|
||||
key = super().register(fileobj, events, data)
|
||||
poll_events = 0
|
||||
if events & EVENT_READ:
|
||||
poll_events |= select.POLLIN
|
||||
if events & EVENT_WRITE:
|
||||
poll_events |= select.POLLOUT
|
||||
self._poll.register(key.fd, poll_events)
|
||||
return key
|
||||
|
||||
def unregister(self, fileobj):
|
||||
key = super().unregister(fileobj)
|
||||
self._poll.unregister(key.fd)
|
||||
return key
|
||||
|
||||
def select(self, timeout=None):
|
||||
timeout = None if timeout is None else max(int(1000 * timeout), 0)
|
||||
ready = []
|
||||
try:
|
||||
fd_event_list = self._poll.poll(timeout)
|
||||
except InterruptedError:
|
||||
return ready
|
||||
for fd, event in fd_event_list:
|
||||
events = 0
|
||||
if event & ~select.POLLIN:
|
||||
events |= EVENT_WRITE
|
||||
if event & ~select.POLLOUT:
|
||||
events |= EVENT_READ
|
||||
|
||||
key = self._key_from_fd(fd)
|
||||
if key:
|
||||
ready.append((key, events & key.events))
|
||||
return ready
|
||||
|
||||
|
||||
if hasattr(select, 'epoll'):
|
||||
|
||||
class EpollSelector(BaseSelector):
|
||||
"""Epoll-based selector."""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self._epoll = select.epoll()
|
||||
|
||||
def fileno(self):
|
||||
return self._epoll.fileno()
|
||||
|
||||
def register(self, fileobj, events, data=None):
|
||||
key = super().register(fileobj, events, data)
|
||||
epoll_events = 0
|
||||
if events & EVENT_READ:
|
||||
epoll_events |= select.EPOLLIN
|
||||
if events & EVENT_WRITE:
|
||||
epoll_events |= select.EPOLLOUT
|
||||
self._epoll.register(key.fd, epoll_events)
|
||||
return key
|
||||
|
||||
def unregister(self, fileobj):
|
||||
key = super().unregister(fileobj)
|
||||
self._epoll.unregister(key.fd)
|
||||
return key
|
||||
|
||||
def select(self, timeout=None):
|
||||
timeout = -1 if timeout is None else max(timeout, 0)
|
||||
max_ev = len(self._fd_to_key)
|
||||
ready = []
|
||||
try:
|
||||
fd_event_list = self._epoll.poll(timeout, max_ev)
|
||||
except InterruptedError:
|
||||
return ready
|
||||
for fd, event in fd_event_list:
|
||||
events = 0
|
||||
if event & ~select.EPOLLIN:
|
||||
events |= EVENT_WRITE
|
||||
if event & ~select.EPOLLOUT:
|
||||
events |= EVENT_READ
|
||||
|
||||
key = self._key_from_fd(fd)
|
||||
if key:
|
||||
ready.append((key, events & key.events))
|
||||
return ready
|
||||
|
||||
def close(self):
|
||||
super().close()
|
||||
self._epoll.close()
|
||||
|
||||
|
||||
if hasattr(select, 'kqueue'):
|
||||
|
||||
class KqueueSelector(BaseSelector):
|
||||
"""Kqueue-based selector."""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self._kqueue = select.kqueue()
|
||||
|
||||
def fileno(self):
|
||||
return self._kqueue.fileno()
|
||||
|
||||
def register(self, fileobj, events, data=None):
|
||||
key = super().register(fileobj, events, data)
|
||||
if events & EVENT_READ:
|
||||
kev = select.kevent(key.fd, select.KQ_FILTER_READ,
|
||||
select.KQ_EV_ADD)
|
||||
self._kqueue.control([kev], 0, 0)
|
||||
if events & EVENT_WRITE:
|
||||
kev = select.kevent(key.fd, select.KQ_FILTER_WRITE,
|
||||
select.KQ_EV_ADD)
|
||||
self._kqueue.control([kev], 0, 0)
|
||||
return key
|
||||
|
||||
def unregister(self, fileobj):
|
||||
key = super().unregister(fileobj)
|
||||
if key.events & EVENT_READ:
|
||||
kev = select.kevent(key.fd, select.KQ_FILTER_READ,
|
||||
select.KQ_EV_DELETE)
|
||||
self._kqueue.control([kev], 0, 0)
|
||||
if key.events & EVENT_WRITE:
|
||||
kev = select.kevent(key.fd, select.KQ_FILTER_WRITE,
|
||||
select.KQ_EV_DELETE)
|
||||
self._kqueue.control([kev], 0, 0)
|
||||
return key
|
||||
|
||||
def select(self, timeout=None):
|
||||
timeout = None if timeout is None else max(timeout, 0)
|
||||
max_ev = len(self._fd_to_key)
|
||||
ready = []
|
||||
try:
|
||||
kev_list = self._kqueue.control(None, max_ev, timeout)
|
||||
except InterruptedError:
|
||||
return ready
|
||||
for kev in kev_list:
|
||||
fd = kev.ident
|
||||
flag = kev.filter
|
||||
events = 0
|
||||
if flag == select.KQ_FILTER_READ:
|
||||
events |= EVENT_READ
|
||||
if flag == select.KQ_FILTER_WRITE:
|
||||
events |= EVENT_WRITE
|
||||
|
||||
key = self._key_from_fd(fd)
|
||||
if key:
|
||||
ready.append((key, events & key.events))
|
||||
return ready
|
||||
|
||||
def close(self):
|
||||
super().close()
|
||||
self._kqueue.close()
|
||||
|
||||
|
||||
# Choose the best implementation: roughly, epoll|kqueue > poll > select.
|
||||
# select() also can't accept a FD > FD_SETSIZE (usually around 1024)
|
||||
if 'KqueueSelector' in globals():
|
||||
DefaultSelector = KqueueSelector
|
||||
elif 'EpollSelector' in globals():
|
||||
DefaultSelector = EpollSelector
|
||||
elif 'PollSelector' in globals():
|
||||
DefaultSelector = PollSelector
|
||||
else:
|
||||
DefaultSelector = SelectSelector
|
||||
|
|
@ -0,0 +1,390 @@
|
|||
import errno
|
||||
import random
|
||||
import selectors
|
||||
import signal
|
||||
import socket
|
||||
from test import support
|
||||
from time import sleep
|
||||
import unittest
|
||||
try:
|
||||
from time import monotonic as time
|
||||
except ImportError:
|
||||
from time import time as time
|
||||
try:
|
||||
import resource
|
||||
except ImportError:
|
||||
resource = None
|
||||
|
||||
|
||||
if hasattr(socket, 'socketpair'):
|
||||
socketpair = socket.socketpair
|
||||
else:
|
||||
def socketpair(family=socket.AF_INET, type=socket.SOCK_STREAM, proto=0):
|
||||
with socket.socket(family, type, proto) as l:
|
||||
l.bind((support.HOST, 0))
|
||||
l.listen(3)
|
||||
c = socket.socket(family, type, proto)
|
||||
try:
|
||||
c.connect(l.getsockname())
|
||||
caddr = c.getsockname()
|
||||
while True:
|
||||
a, addr = l.accept()
|
||||
# check that we've got the correct client
|
||||
if addr == caddr:
|
||||
return c, a
|
||||
a.close()
|
||||
except OSError:
|
||||
c.close()
|
||||
raise
|
||||
|
||||
|
||||
def find_ready_matching(ready, flag):
|
||||
match = []
|
||||
for key, events in ready:
|
||||
if events & flag:
|
||||
match.append(key.fileobj)
|
||||
return match
|
||||
|
||||
|
||||
class BaseSelectorTestCase(unittest.TestCase):
|
||||
|
||||
def test_register(self):
|
||||
s = self.SELECTOR()
|
||||
self.addCleanup(s.close)
|
||||
|
||||
rd, wr = socketpair()
|
||||
self.addCleanup(rd.close)
|
||||
self.addCleanup(wr.close)
|
||||
|
||||
key = s.register(rd, selectors.EVENT_READ, "data")
|
||||
self.assertIsInstance(key, selectors.SelectorKey)
|
||||
self.assertEqual(key.fileobj, rd)
|
||||
self.assertEqual(key.fd, rd.fileno())
|
||||
self.assertEqual(key.events, selectors.EVENT_READ)
|
||||
self.assertEqual(key.data, "data")
|
||||
|
||||
# register an unknown event
|
||||
self.assertRaises(ValueError, s.register, 0, 999999)
|
||||
|
||||
# register an invalid FD
|
||||
self.assertRaises(ValueError, s.register, -10, selectors.EVENT_READ)
|
||||
|
||||
# register twice
|
||||
self.assertRaises(KeyError, s.register, rd, selectors.EVENT_READ)
|
||||
|
||||
# register the same FD, but with a different object
|
||||
self.assertRaises(KeyError, s.register, rd.fileno(),
|
||||
selectors.EVENT_READ)
|
||||
|
||||
def test_unregister(self):
|
||||
s = self.SELECTOR()
|
||||
self.addCleanup(s.close)
|
||||
|
||||
rd, wr = socketpair()
|
||||
self.addCleanup(rd.close)
|
||||
self.addCleanup(wr.close)
|
||||
|
||||
s.register(rd, selectors.EVENT_READ)
|
||||
s.unregister(rd)
|
||||
|
||||
# unregister an unknown file obj
|
||||
self.assertRaises(KeyError, s.unregister, 999999)
|
||||
|
||||
# unregister twice
|
||||
self.assertRaises(KeyError, s.unregister, rd)
|
||||
|
||||
def test_modify(self):
|
||||
s = self.SELECTOR()
|
||||
self.addCleanup(s.close)
|
||||
|
||||
rd, wr = socketpair()
|
||||
self.addCleanup(rd.close)
|
||||
self.addCleanup(wr.close)
|
||||
|
||||
key = s.register(rd, selectors.EVENT_READ)
|
||||
|
||||
# modify events
|
||||
key2 = s.modify(rd, selectors.EVENT_WRITE)
|
||||
self.assertNotEqual(key.events, key2.events)
|
||||
self.assertEqual(key2, s.get_key(rd))
|
||||
|
||||
s.unregister(rd)
|
||||
|
||||
# modify data
|
||||
d1 = object()
|
||||
d2 = object()
|
||||
|
||||
key = s.register(rd, selectors.EVENT_READ, d1)
|
||||
key2 = s.modify(rd, selectors.EVENT_READ, d2)
|
||||
self.assertEqual(key.events, key2.events)
|
||||
self.assertNotEqual(key.data, key2.data)
|
||||
self.assertEqual(key2, s.get_key(rd))
|
||||
self.assertEqual(key2.data, d2)
|
||||
|
||||
# modify unknown file obj
|
||||
self.assertRaises(KeyError, s.modify, 999999, selectors.EVENT_READ)
|
||||
|
||||
def test_close(self):
|
||||
s = self.SELECTOR()
|
||||
self.addCleanup(s.close)
|
||||
|
||||
rd, wr = socketpair()
|
||||
self.addCleanup(rd.close)
|
||||
self.addCleanup(wr.close)
|
||||
|
||||
s.register(rd, selectors.EVENT_READ)
|
||||
s.register(wr, selectors.EVENT_WRITE)
|
||||
|
||||
s.close()
|
||||
self.assertRaises(KeyError, s.get_key, rd)
|
||||
self.assertRaises(KeyError, s.get_key, wr)
|
||||
|
||||
def test_get_key(self):
|
||||
s = self.SELECTOR()
|
||||
self.addCleanup(s.close)
|
||||
|
||||
rd, wr = socketpair()
|
||||
self.addCleanup(rd.close)
|
||||
self.addCleanup(wr.close)
|
||||
|
||||
key = s.register(rd, selectors.EVENT_READ, "data")
|
||||
self.assertEqual(key, s.get_key(rd))
|
||||
|
||||
# unknown file obj
|
||||
self.assertRaises(KeyError, s.get_key, 999999)
|
||||
|
||||
def test_select(self):
|
||||
s = self.SELECTOR()
|
||||
self.addCleanup(s.close)
|
||||
|
||||
rd, wr = socketpair()
|
||||
self.addCleanup(rd.close)
|
||||
self.addCleanup(wr.close)
|
||||
|
||||
s.register(rd, selectors.EVENT_READ)
|
||||
wr_key = s.register(wr, selectors.EVENT_WRITE)
|
||||
|
||||
result = s.select()
|
||||
for key, events in result:
|
||||
self.assertTrue(isinstance(key, selectors.SelectorKey))
|
||||
self.assertTrue(events)
|
||||
self.assertFalse(events & ~(selectors.EVENT_READ |
|
||||
selectors.EVENT_WRITE))
|
||||
|
||||
self.assertEqual([(wr_key, selectors.EVENT_WRITE)], result)
|
||||
|
||||
def test_context_manager(self):
|
||||
s = self.SELECTOR()
|
||||
self.addCleanup(s.close)
|
||||
|
||||
rd, wr = socketpair()
|
||||
self.addCleanup(rd.close)
|
||||
self.addCleanup(wr.close)
|
||||
|
||||
with s as sel:
|
||||
sel.register(rd, selectors.EVENT_READ)
|
||||
sel.register(wr, selectors.EVENT_WRITE)
|
||||
|
||||
self.assertRaises(KeyError, s.get_key, rd)
|
||||
self.assertRaises(KeyError, s.get_key, wr)
|
||||
|
||||
def test_fileno(self):
|
||||
s = self.SELECTOR()
|
||||
self.addCleanup(s.close)
|
||||
|
||||
if hasattr(s, 'fileno'):
|
||||
fd = s.fileno()
|
||||
self.assertTrue(isinstance(fd, int))
|
||||
self.assertGreaterEqual(fd, 0)
|
||||
|
||||
def test_selector(self):
|
||||
s = self.SELECTOR()
|
||||
self.addCleanup(s.close)
|
||||
|
||||
NUM_SOCKETS = 12
|
||||
MSG = b" This is a test."
|
||||
MSG_LEN = len(MSG)
|
||||
readers = []
|
||||
writers = []
|
||||
r2w = {}
|
||||
w2r = {}
|
||||
|
||||
for i in range(NUM_SOCKETS):
|
||||
rd, wr = socketpair()
|
||||
self.addCleanup(rd.close)
|
||||
self.addCleanup(wr.close)
|
||||
s.register(rd, selectors.EVENT_READ)
|
||||
s.register(wr, selectors.EVENT_WRITE)
|
||||
readers.append(rd)
|
||||
writers.append(wr)
|
||||
r2w[rd] = wr
|
||||
w2r[wr] = rd
|
||||
|
||||
bufs = []
|
||||
|
||||
while writers:
|
||||
ready = s.select()
|
||||
ready_writers = find_ready_matching(ready, selectors.EVENT_WRITE)
|
||||
if not ready_writers:
|
||||
self.fail("no sockets ready for writing")
|
||||
wr = random.choice(ready_writers)
|
||||
wr.send(MSG)
|
||||
|
||||
for i in range(10):
|
||||
ready = s.select()
|
||||
ready_readers = find_ready_matching(ready,
|
||||
selectors.EVENT_READ)
|
||||
if ready_readers:
|
||||
break
|
||||
# there might be a delay between the write to the write end and
|
||||
# the read end is reported ready
|
||||
sleep(0.1)
|
||||
else:
|
||||
self.fail("no sockets ready for reading")
|
||||
self.assertEqual([w2r[wr]], ready_readers)
|
||||
rd = ready_readers[0]
|
||||
buf = rd.recv(MSG_LEN)
|
||||
self.assertEqual(len(buf), MSG_LEN)
|
||||
bufs.append(buf)
|
||||
s.unregister(r2w[rd])
|
||||
s.unregister(rd)
|
||||
writers.remove(r2w[rd])
|
||||
|
||||
self.assertEqual(bufs, [MSG] * NUM_SOCKETS)
|
||||
|
||||
def test_timeout(self):
|
||||
s = self.SELECTOR()
|
||||
self.addCleanup(s.close)
|
||||
|
||||
rd, wr = socketpair()
|
||||
self.addCleanup(rd.close)
|
||||
self.addCleanup(wr.close)
|
||||
|
||||
s.register(wr, selectors.EVENT_WRITE)
|
||||
t = time()
|
||||
self.assertEqual(1, len(s.select(0)))
|
||||
self.assertEqual(1, len(s.select(-1)))
|
||||
self.assertTrue(time() - t < 0.5)
|
||||
|
||||
s.unregister(wr)
|
||||
s.register(rd, selectors.EVENT_READ)
|
||||
t = time()
|
||||
self.assertFalse(s.select(0))
|
||||
self.assertFalse(s.select(-1))
|
||||
self.assertTrue(time() - t < 0.5)
|
||||
|
||||
t = time()
|
||||
self.assertFalse(s.select(1))
|
||||
self.assertTrue(0.5 < time() - t < 1.5)
|
||||
|
||||
@unittest.skipUnless(hasattr(signal, "alarm"),
|
||||
"signal.alarm() required for this test")
|
||||
def test_interrupted_retry(self):
|
||||
s = self.SELECTOR()
|
||||
self.addCleanup(s.close)
|
||||
|
||||
rd, wr = socketpair()
|
||||
self.addCleanup(rd.close)
|
||||
self.addCleanup(wr.close)
|
||||
|
||||
orig_alrm_handler = signal.signal(signal.SIGALRM, lambda *args: None)
|
||||
self.addCleanup(signal.signal, signal.SIGALRM, orig_alrm_handler)
|
||||
self.addCleanup(signal.alarm, 0)
|
||||
|
||||
signal.alarm(1)
|
||||
|
||||
s.register(rd, selectors.EVENT_READ)
|
||||
t = time()
|
||||
self.assertFalse(s.select(2))
|
||||
self.assertLess(time() - t, 2.5)
|
||||
|
||||
|
||||
class ScalableSelectorMixIn:
|
||||
|
||||
@support.requires_mac_ver(10, 5)
|
||||
@unittest.skipUnless(resource, "Test needs resource module")
|
||||
def test_above_fd_setsize(self):
|
||||
# A scalable implementation should have no problem with more than
|
||||
# FD_SETSIZE file descriptors. Since we don't know the value, we just
|
||||
# try to set the soft RLIMIT_NOFILE to the hard RLIMIT_NOFILE ceiling.
|
||||
soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
|
||||
try:
|
||||
resource.setrlimit(resource.RLIMIT_NOFILE, (hard, hard))
|
||||
self.addCleanup(resource.setrlimit, resource.RLIMIT_NOFILE,
|
||||
(soft, hard))
|
||||
NUM_FDS = hard
|
||||
except OSError:
|
||||
NUM_FDS = soft
|
||||
|
||||
# guard for already allocated FDs (stdin, stdout...)
|
||||
NUM_FDS -= 32
|
||||
|
||||
s = self.SELECTOR()
|
||||
self.addCleanup(s.close)
|
||||
|
||||
for i in range(NUM_FDS // 2):
|
||||
try:
|
||||
rd, wr = socketpair()
|
||||
except OSError:
|
||||
# too many FDs, skip - note that we should only catch EMFILE
|
||||
# here, but apparently *BSD and Solaris can fail upon connect()
|
||||
# or bind() with EADDRNOTAVAIL, so let's be safe
|
||||
self.skipTest("FD limit reached")
|
||||
|
||||
self.addCleanup(rd.close)
|
||||
self.addCleanup(wr.close)
|
||||
|
||||
try:
|
||||
s.register(rd, selectors.EVENT_READ)
|
||||
s.register(wr, selectors.EVENT_WRITE)
|
||||
except OSError as e:
|
||||
if e.errno == errno.ENOSPC:
|
||||
# this can be raised by epoll if we go over
|
||||
# fs.epoll.max_user_watches sysctl
|
||||
self.skipTest("FD limit reached")
|
||||
raise
|
||||
|
||||
self.assertEqual(NUM_FDS // 2, len(s.select()))
|
||||
|
||||
|
||||
class DefaultSelectorTestCase(BaseSelectorTestCase):
|
||||
|
||||
SELECTOR = selectors.DefaultSelector
|
||||
|
||||
|
||||
class SelectSelectorTestCase(BaseSelectorTestCase):
|
||||
|
||||
SELECTOR = selectors.SelectSelector
|
||||
|
||||
|
||||
@unittest.skipUnless(hasattr(selectors, 'PollSelector'),
|
||||
"Test needs selectors.PollSelector")
|
||||
class PollSelectorTestCase(BaseSelectorTestCase, ScalableSelectorMixIn):
|
||||
|
||||
SELECTOR = getattr(selectors, 'PollSelector', None)
|
||||
|
||||
|
||||
@unittest.skipUnless(hasattr(selectors, 'EpollSelector'),
|
||||
"Test needs selectors.EpollSelector")
|
||||
class EpollSelectorTestCase(BaseSelectorTestCase, ScalableSelectorMixIn):
|
||||
|
||||
SELECTOR = getattr(selectors, 'EpollSelector', None)
|
||||
|
||||
|
||||
@unittest.skipUnless(hasattr(selectors, 'KqueueSelector'),
|
||||
"Test needs selectors.KqueueSelector)")
|
||||
class KqueueSelectorTestCase(BaseSelectorTestCase, ScalableSelectorMixIn):
|
||||
|
||||
SELECTOR = getattr(selectors, 'KqueueSelector', None)
|
||||
|
||||
|
||||
def test_main():
|
||||
tests = [DefaultSelectorTestCase, SelectSelectorTestCase,
|
||||
PollSelectorTestCase, EpollSelectorTestCase,
|
||||
KqueueSelectorTestCase]
|
||||
support.run_unittest(*tests)
|
||||
support.reap_children()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_main()
|
||||
Loading…
Reference in New Issue