+++ /dev/null
-"""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, Mapping
-import math
-import select
-import sys
-
-from .py33_exceptions import wrap_error, InterruptedError
-from .compat import integer_types
-
-
-# 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
-
- Raises:
- ValueError if the object is invalid
- """
- if isinstance(fileobj, integer_types):
- fd = fileobj
- else:
- try:
- fd = int(fileobj.fileno())
- except (AttributeError, TypeError, ValueError):
- raise ValueError("Invalid file object: "
- "{0!r}".format(fileobj))
- if fd < 0:
- raise ValueError("Invalid file descriptor: {0}".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 _SelectorMapping(Mapping):
- """Mapping of file objects to selector keys."""
-
- def __init__(self, selector):
- self._selector = selector
-
- def __len__(self):
- return len(self._selector._fd_to_key)
-
- def __getitem__(self, fileobj):
- try:
- fd = self._selector._fileobj_lookup(fileobj)
- return self._selector._fd_to_key[fd]
- except KeyError:
- raise KeyError("{0!r} is not registered".format(fileobj))
-
- def __iter__(self):
- return iter(self._selector._fd_to_key)
-
-
-class BaseSelector(object):
- """Selector abstract base 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
- efficient implementation on the current platform.
- """
- __metaclass__ = ABCMeta
-
- @abstractmethod
- 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
-
- Raises:
- 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)
-
- Note:
- OSError may or may not be raised
- """
- raise NotImplementedError
-
- @abstractmethod
- def unregister(self, fileobj):
- """Unregister a file object.
-
- Parameters:
- fileobj -- file object or file descriptor
-
- Returns:
- SelectorKey instance
-
- Raises:
- KeyError if fileobj is not registered
-
- Note:
- 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.
-
- Parameters:
- fileobj -- file object or file descriptor
- events -- events to monitor (bitwise mask of EVENT_READ|EVENT_WRITE)
- data -- attached data
-
- Returns:
- SelectorKey instance
-
- Raises:
- Anything that unregister() or register() raises
- """
- self.unregister(fileobj)
- return self.register(fileobj, events, data)
-
- @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.
- """
- pass
-
- def get_key(self, fileobj):
- """Return the key associated to a registered file object.
-
- Returns:
- SelectorKey for this file object
- """
- mapping = self.get_map()
- if mapping is None:
- raise RuntimeError('Selector is closed')
- try:
- return mapping[fileobj]
- except KeyError:
- raise KeyError("{0!r} is not registered".format(fileobj))
-
- @abstractmethod
- def get_map(self):
- """Return a mapping of file objects to selector keys."""
- raise NotImplementedError
-
- def __enter__(self):
- return self
-
- def __exit__(self, *args):
- self.close()
-
-
-class _BaseSelectorImpl(BaseSelector):
- """Base selector implementation."""
-
- def __init__(self):
- # this maps file descriptors to keys
- self._fd_to_key = {}
- # 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.
- """
- try:
- return _fileobj_to_fd(fileobj)
- except ValueError:
- # Do an exhaustive search.
- for key in self._fd_to_key.values():
- if key.fileobj is fileobj:
- return key.fd
- # Raise ValueError after all.
- raise
-
- def register(self, fileobj, events, data=None):
- if (not events) or (events & ~(EVENT_READ | EVENT_WRITE)):
- raise ValueError("Invalid events: {0!r}".format(events))
-
- key = SelectorKey(fileobj, self._fileobj_lookup(fileobj), events, data)
-
- if key.fd in self._fd_to_key:
- raise KeyError("{0!r} (FD {1}) is already registered"
- .format(fileobj, key.fd))
-
- self._fd_to_key[key.fd] = key
- return key
-
- def unregister(self, fileobj):
- try:
- key = self._fd_to_key.pop(self._fileobj_lookup(fileobj))
- except KeyError:
- raise KeyError("{0!r} is not registered".format(fileobj))
- return key
-
- def modify(self, fileobj, events, data=None):
- # TODO: Subclasses can probably optimize this even further.
- try:
- key = self._fd_to_key[self._fileobj_lookup(fileobj)]
- except KeyError:
- raise KeyError("{0!r} is not registered".format(fileobj))
- if events != key.events:
- self.unregister(fileobj)
- key = self.register(fileobj, events, data)
- elif data != key.data:
- # Use a shortcut to update the data.
- key = key._replace(data=data)
- self._fd_to_key[key.fd] = key
- return key
-
- def close(self):
- self._fd_to_key.clear()
- self._map = None
-
- def get_map(self):
- return self._map
-
- 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(_BaseSelectorImpl):
- """Select-based selector."""
-
- def __init__(self):
- super(SelectSelector, self).__init__()
- self._readers = set()
- self._writers = set()
-
- def register(self, fileobj, events, data=None):
- key = super(SelectSelector, self).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(SelectSelector, self).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:
- def _select(self, r, w, x, timeout=None):
- return select.select(r, w, x, timeout)
-
- def select(self, timeout=None):
- timeout = None if timeout is None else max(timeout, 0)
- ready = []
- try:
- r, w, _ = wrap_error(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(_BaseSelectorImpl):
- """Poll-based selector."""
-
- def __init__(self):
- super(PollSelector, self).__init__()
- self._poll = select.poll()
-
- def register(self, fileobj, events, data=None):
- key = super(PollSelector, self).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(PollSelector, self).unregister(fileobj)
- self._poll.unregister(key.fd)
- return key
-
- def select(self, timeout=None):
- if timeout is None:
- timeout = None
- elif timeout <= 0:
- timeout = 0
- else:
- # poll() has a resolution of 1 millisecond, round away from
- # zero to wait *at least* timeout seconds.
- timeout = int(math.ceil(timeout * 1e3))
- ready = []
- try:
- fd_event_list = wrap_error(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(_BaseSelectorImpl):
- """Epoll-based selector."""
-
- def __init__(self):
- super(EpollSelector, self).__init__()
- self._epoll = select.epoll()
-
- def fileno(self):
- return self._epoll.fileno()
-
- def register(self, fileobj, events, data=None):
- key = super(EpollSelector, self).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(EpollSelector, self).unregister(fileobj)
- try:
- self._epoll.unregister(key.fd)
- except IOError:
- # This can happen if the FD was closed since it
- # was registered.
- pass
- return key
-
- def select(self, timeout=None):
- if timeout is None:
- timeout = -1
- elif timeout <= 0:
- timeout = 0
- else:
- # 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
- # FD is registered.
- max_ev = max(len(self._fd_to_key), 1)
-
- ready = []
- try:
- fd_event_list = wrap_error(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):
- self._epoll.close()
- super(EpollSelector, self).close()
-
-
-if hasattr(select, 'devpoll'):
-
- class DevpollSelector(_BaseSelectorImpl):
- """Solaris /dev/poll selector."""
-
- def __init__(self):
- super(DevpollSelector, self).__init__()
- self._devpoll = select.devpoll()
-
- def fileno(self):
- return self._devpoll.fileno()
-
- def register(self, fileobj, events, data=None):
- key = super(DevpollSelector, self).register(fileobj, events, data)
- poll_events = 0
- if events & EVENT_READ:
- poll_events |= select.POLLIN
- if events & EVENT_WRITE:
- poll_events |= select.POLLOUT
- self._devpoll.register(key.fd, poll_events)
- return key
-
- def unregister(self, fileobj):
- key = super(DevpollSelector, self).unregister(fileobj)
- self._devpoll.unregister(key.fd)
- return key
-
- def select(self, timeout=None):
- if timeout is None:
- timeout = None
- elif timeout <= 0:
- timeout = 0
- else:
- # devpoll() has a resolution of 1 millisecond, round away from
- # zero to wait *at least* timeout seconds.
- timeout = math.ceil(timeout * 1e3)
- ready = []
- try:
- fd_event_list = self._devpoll.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
-
- def close(self):
- self._devpoll.close()
- super(DevpollSelector, self).close()
-
-
-if hasattr(select, 'kqueue'):
-
- class KqueueSelector(_BaseSelectorImpl):
- """Kqueue-based selector."""
-
- def __init__(self):
- super(KqueueSelector, self).__init__()
- self._kqueue = select.kqueue()
-
- def fileno(self):
- return self._kqueue.fileno()
-
- def register(self, fileobj, events, data=None):
- key = super(KqueueSelector, self).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(KqueueSelector, self).unregister(fileobj)
- if key.events & EVENT_READ:
- kev = select.kevent(key.fd, select.KQ_FILTER_READ,
- select.KQ_EV_DELETE)
- try:
- self._kqueue.control([kev], 0, 0)
- except OSError:
- # This can happen if the FD was closed since it
- # was registered.
- pass
- if key.events & EVENT_WRITE:
- kev = select.kevent(key.fd, select.KQ_FILTER_WRITE,
- select.KQ_EV_DELETE)
- try:
- self._kqueue.control([kev], 0, 0)
- except OSError:
- # See comment above.
- pass
- 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 = wrap_error(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):
- self._kqueue.close()
- super(KqueueSelector, self).close()
-
-
-# Choose the best implementation, roughly:
-# epoll|kqueue|devpoll > 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 'DevpollSelector' in globals():
- DefaultSelector = DevpollSelector
-elif 'PollSelector' in globals():
- DefaultSelector = PollSelector
-else:
- DefaultSelector = SelectSelector
projects / dotfiles / .git / blobdiff