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)
python类devpoll()的实例源码
def __init__(self):
super(DevpollSelector, self).__init__()
self._devpoll = select.devpoll()
def _wrap_poll(self, timeout=None):
""" Wrapper function for select.poll.poll() so that
_syscall_wrapper can work with only seconds. """
if timeout is not None:
if timeout <= 0:
timeout = 0
else:
# select.devpoll.poll() has a resolution of 1 millisecond,
# round away from zero to wait *at least* timeout seconds.
timeout = math.ceil(timeout * 1000)
result = self._devpoll.poll(timeout)
return result
def __init__(self):
super(DevpollSelector, self).__init__()
self._devpoll = select.devpoll()
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)
def __init__(self):
super(DevpollSelector, self).__init__()
self._devpoll = select.devpoll()
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)
def __init__(self):
super(DevpollSelector, self).__init__()
self._devpoll = select.devpoll()
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)
def __init__(self):
super(DevpollSelector, self).__init__()
self._devpoll = select.devpoll()
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)
def __init__(self):
super(DevpollSelector, self).__init__()
self._devpoll = select.devpoll()
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)
def test_fd_non_inheritable(self):
devpoll = select.devpoll()
self.addCleanup(devpoll.close)
self.assertEqual(os.get_inheritable(devpoll.fileno()), False)
def test_events_mask_overflow(self):
pollster = select.devpoll()
w, r = os.pipe()
pollster.register(w)
# Issue #17919
self.assertRaises(OverflowError, pollster.register, 0, -1)
self.assertRaises(OverflowError, pollster.register, 0, 1 << 64)
self.assertRaises(OverflowError, pollster.modify, 1, -1)
self.assertRaises(OverflowError, pollster.modify, 1, 1 << 64)
def __init__(self):
super(DevpollSelector, self).__init__()
self._devpoll = select.devpoll()
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 OSError as exc:
if exc.errno == EINTR:
return ready
else:
raise
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._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)
def __init__(self):
super(DevpollSelector, self).__init__()
self._devpoll = select.devpoll()
def __init__(self):
super(DevpollSelector, self).__init__()
self._devpoll = select.devpoll()
