[Python-checkins] cpython: Issue #16853: Add new selectors module.
charles-francois.natali
python-checkins at python.org
Wed Sep 4 19:05:46 CEST 2013
http://hg.python.org/cpython/rev/e4d45315c38c
changeset: 85523:e4d45315c38c
user: Charles-François Natali <cf.natali at gmail.com>
date: Wed Sep 04 19:02:49 2013 +0200
summary:
Issue #16853: Add new selectors module.
files:
Doc/library/concurrency.rst | 1 +
Doc/library/selectors.rst | 231 +++++++++++++
Doc/whatsnew/3.4.rst | 7 +-
Lib/selectors.py | 405 ++++++++++++++++++++++++
Lib/test/test_selectors.py | 390 +++++++++++++++++++++++
Misc/NEWS | 2 +
6 files changed, 1033 insertions(+), 3 deletions(-)
diff --git a/Doc/library/concurrency.rst b/Doc/library/concurrency.rst
--- a/Doc/library/concurrency.rst
+++ b/Doc/library/concurrency.rst
@@ -21,6 +21,7 @@
sched.rst
queue.rst
select.rst
+ selectors.rst
The following are support modules for some of the above services:
diff --git a/Doc/library/selectors.rst b/Doc/library/selectors.rst
new file mode 100644
--- /dev/null
+++ b/Doc/library/selectors.rst
@@ -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()
diff --git a/Doc/whatsnew/3.4.rst b/Doc/whatsnew/3.4.rst
--- a/Doc/whatsnew/3.4.rst
+++ b/Doc/whatsnew/3.4.rst
@@ -174,10 +174,11 @@
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
diff --git a/Lib/selectors.py b/Lib/selectors.py
new file mode 100644
--- /dev/null
+++ b/Lib/selectors.py
@@ -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
diff --git a/Lib/test/test_selectors.py b/Lib/test/test_selectors.py
new file mode 100644
--- /dev/null
+++ b/Lib/test/test_selectors.py
@@ -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
+
+
+ at unittest.skipUnless(hasattr(selectors, 'PollSelector'),
+ "Test needs selectors.PollSelector")
+class PollSelectorTestCase(BaseSelectorTestCase, ScalableSelectorMixIn):
+
+ SELECTOR = getattr(selectors, 'PollSelector', None)
+
+
+ at unittest.skipUnless(hasattr(selectors, 'EpollSelector'),
+ "Test needs selectors.EpollSelector")
+class EpollSelectorTestCase(BaseSelectorTestCase, ScalableSelectorMixIn):
+
+ SELECTOR = getattr(selectors, 'EpollSelector', None)
+
+
+ at 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()
diff --git a/Misc/NEWS b/Misc/NEWS
--- a/Misc/NEWS
+++ b/Misc/NEWS
@@ -54,6 +54,8 @@
Library
-------
+- Issue #16853: Add new selectors module.
+
- Issue #18882: Add threading.main_thread() function.
- Issue #18901: The sunau getparams method now returns a namedtuple rather than
--
Repository URL: http://hg.python.org/cpython
More information about the Python-checkins
mailing list