我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用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 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 DefaultSelector(): """ This function serves as a first call for DefaultSelector to detect if the select module is being monkey-patched incorrectly by eventlet, greenlet, and preserve proper behavior. """ global _DEFAULT_SELECTOR if _DEFAULT_SELECTOR is None: if platform.python_implementation() == 'Jython': # Platform-specific: Jython _DEFAULT_SELECTOR = JythonSelectSelector elif _can_allocate('kqueue'): _DEFAULT_SELECTOR = KqueueSelector elif _can_allocate('devpoll'): _DEFAULT_SELECTOR = DevpollSelector elif _can_allocate('epoll'): _DEFAULT_SELECTOR = EpollSelector elif _can_allocate('poll'): _DEFAULT_SELECTOR = PollSelector elif hasattr(select, 'select'): _DEFAULT_SELECTOR = SelectSelector else: # Platform-specific: AppEngine raise RuntimeError('Platform does not have a selector.') return _DEFAULT_SELECTOR()
def test_timeout_overflow(self): pollster = select.devpoll() w, r = os.pipe() pollster.register(w) pollster.poll(-1) self.assertRaises(OverflowError, pollster.poll, -2) self.assertRaises(OverflowError, pollster.poll, -1 << 31) self.assertRaises(OverflowError, pollster.poll, -1 << 64) pollster.poll(0) pollster.poll(1) pollster.poll(1 << 30) self.assertRaises(OverflowError, pollster.poll, 1 << 31) self.assertRaises(OverflowError, pollster.poll, 1 << 63) self.assertRaises(OverflowError, pollster.poll, 1 << 64)
def test_close(self): open_file = open(__file__, "rb") self.addCleanup(open_file.close) fd = open_file.fileno() devpoll = select.devpoll() # test fileno() method and closed attribute self.assertIsInstance(devpoll.fileno(), int) self.assertFalse(devpoll.closed) # test close() devpoll.close() self.assertTrue(devpoll.closed) self.assertRaises(ValueError, devpoll.fileno) # close() can be called more than once devpoll.close() # operations must fail with ValueError("I/O operation on closed ...") self.assertRaises(ValueError, devpoll.modify, fd, select.POLLIN) self.assertRaises(ValueError, devpoll.poll) self.assertRaises(ValueError, devpoll.register, fd, fd, select.POLLIN) self.assertRaises(ValueError, devpoll.unregister, fd)
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().__init__() self._devpoll = select.devpoll()
def close(self): self._kqueue.close() super().close() # Choose the best implementation, roughly: # epoll|kqueue|devpoll > poll > select. # select() also can't accept a FD > FD_SETSIZE (usually around 1024)
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 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 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
