selectors.py 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588
  1. # SPDX-License-Identifier: MIT
  2. # Backport of selectors.py from Python 3.5+ to support Python < 3.4
  3. # Also has the behavior specified in PEP 475 which is to retry syscalls
  4. # in the case of an EINTR error. This module is required because selectors34
  5. # does not follow this behavior and instead returns that no dile descriptor
  6. # events have occurred rather than retry the syscall. The decision to drop
  7. # support for select.devpoll is made to maintain 100% test coverage.
  8. import errno
  9. import math
  10. import select
  11. import socket
  12. import sys
  13. import time
  14. from collections import namedtuple
  15. try:
  16. from collections import Mapping
  17. except ImportError:
  18. from collections.abc import Mapping
  19. try:
  20. monotonic = time.monotonic
  21. except (AttributeError, ImportError): # Python 3.3<
  22. monotonic = time.time
  23. EVENT_READ = (1 << 0)
  24. EVENT_WRITE = (1 << 1)
  25. HAS_SELECT = True # Variable that shows whether the platform has a selector.
  26. _SYSCALL_SENTINEL = object() # Sentinel in case a system call returns None.
  27. _DEFAULT_SELECTOR = None
  28. class SelectorError(Exception):
  29. def __init__(self, errcode):
  30. super(SelectorError, self).__init__()
  31. self.errno = errcode
  32. def __repr__(self):
  33. return "<SelectorError errno={0}>".format(self.errno)
  34. def __str__(self):
  35. return self.__repr__()
  36. def _fileobj_to_fd(fileobj):
  37. """ Return a file descriptor from a file object. If
  38. given an integer will simply return that integer back. """
  39. if isinstance(fileobj, int):
  40. fd = fileobj
  41. else:
  42. try:
  43. fd = int(fileobj.fileno())
  44. except (AttributeError, TypeError, ValueError):
  45. raise ValueError("Invalid file object: {0!r}".format(fileobj))
  46. if fd < 0:
  47. raise ValueError("Invalid file descriptor: {0}".format(fd))
  48. return fd
  49. # Determine which function to use to wrap system calls because Python 3.5+
  50. # already handles the case when system calls are interrupted.
  51. if sys.version_info >= (3, 5):
  52. def _syscall_wrapper(func, _, *args, **kwargs):
  53. """ This is the short-circuit version of the below logic
  54. because in Python 3.5+ all system calls automatically restart
  55. and recalculate their timeouts. """
  56. try:
  57. return func(*args, **kwargs)
  58. except (OSError, IOError, select.error) as e:
  59. errcode = None
  60. if hasattr(e, "errno"):
  61. errcode = e.errno
  62. raise SelectorError(errcode)
  63. else:
  64. def _syscall_wrapper(func, recalc_timeout, *args, **kwargs):
  65. """ Wrapper function for syscalls that could fail due to EINTR.
  66. All functions should be retried if there is time left in the timeout
  67. in accordance with PEP 475. """
  68. timeout = kwargs.get("timeout", None)
  69. if timeout is None:
  70. expires = None
  71. recalc_timeout = False
  72. else:
  73. timeout = float(timeout)
  74. if timeout < 0.0: # Timeout less than 0 treated as no timeout.
  75. expires = None
  76. else:
  77. expires = monotonic() + timeout
  78. args = list(args)
  79. if recalc_timeout and "timeout" not in kwargs:
  80. raise ValueError(
  81. "Timeout must be in args or kwargs to be recalculated")
  82. result = _SYSCALL_SENTINEL
  83. while result is _SYSCALL_SENTINEL:
  84. try:
  85. result = func(*args, **kwargs)
  86. # OSError is thrown by select.select
  87. # IOError is thrown by select.epoll.poll
  88. # select.error is thrown by select.poll.poll
  89. # Aren't we thankful for Python 3.x rework for exceptions?
  90. except (OSError, IOError, select.error) as e:
  91. # select.error wasn't a subclass of OSError in the past.
  92. errcode = None
  93. if hasattr(e, "errno"):
  94. errcode = e.errno
  95. elif hasattr(e, "args"):
  96. errcode = e.args[0]
  97. # Also test for the Windows equivalent of EINTR.
  98. is_interrupt = (errcode == errno.EINTR or (hasattr(errno, "WSAEINTR") and
  99. errcode == errno.WSAEINTR))
  100. if is_interrupt:
  101. if expires is not None:
  102. current_time = monotonic()
  103. if current_time > expires:
  104. raise OSError(errno=errno.ETIMEDOUT)
  105. if recalc_timeout:
  106. if "timeout" in kwargs:
  107. kwargs["timeout"] = expires - current_time
  108. continue
  109. if errcode:
  110. raise SelectorError(errcode)
  111. else:
  112. raise
  113. return result
  114. SelectorKey = namedtuple('SelectorKey', ['fileobj', 'fd', 'events', 'data'])
  115. class _SelectorMapping(Mapping):
  116. """ Mapping of file objects to selector keys """
  117. def __init__(self, selector):
  118. self._selector = selector
  119. def __len__(self):
  120. return len(self._selector._fd_to_key)
  121. def __getitem__(self, fileobj):
  122. try:
  123. fd = self._selector._fileobj_lookup(fileobj)
  124. return self._selector._fd_to_key[fd]
  125. except KeyError:
  126. raise KeyError("{0!r} is not registered.".format(fileobj))
  127. def __iter__(self):
  128. return iter(self._selector._fd_to_key)
  129. class BaseSelector(object):
  130. """ Abstract Selector class
  131. A selector supports registering file objects to be monitored
  132. for specific I/O events.
  133. A file object is a file descriptor or any object with a
  134. `fileno()` method. An arbitrary object can be attached to the
  135. file object which can be used for example to store context info,
  136. a callback, etc.
  137. A selector can use various implementations (select(), poll(), epoll(),
  138. and kqueue()) depending on the platform. The 'DefaultSelector' class uses
  139. the most efficient implementation for the current platform.
  140. """
  141. def __init__(self):
  142. # Maps file descriptors to keys.
  143. self._fd_to_key = {}
  144. # Read-only mapping returned by get_map()
  145. self._map = _SelectorMapping(self)
  146. def _fileobj_lookup(self, fileobj):
  147. """ Return a file descriptor from a file object.
  148. This wraps _fileobj_to_fd() to do an exhaustive
  149. search in case the object is invalid but we still
  150. have it in our map. Used by unregister() so we can
  151. unregister an object that was previously registered
  152. even if it is closed. It is also used by _SelectorMapping
  153. """
  154. try:
  155. return _fileobj_to_fd(fileobj)
  156. except ValueError:
  157. # Search through all our mapped keys.
  158. for key in self._fd_to_key.values():
  159. if key.fileobj is fileobj:
  160. return key.fd
  161. # Raise ValueError after all.
  162. raise
  163. def register(self, fileobj, events, data=None):
  164. """ Register a file object for a set of events to monitor. """
  165. if (not events) or (events & ~(EVENT_READ | EVENT_WRITE)):
  166. raise ValueError("Invalid events: {0!r}".format(events))
  167. key = SelectorKey(fileobj, self._fileobj_lookup(fileobj), events, data)
  168. if key.fd in self._fd_to_key:
  169. raise KeyError("{0!r} (FD {1}) is already registered"
  170. .format(fileobj, key.fd))
  171. self._fd_to_key[key.fd] = key
  172. return key
  173. def unregister(self, fileobj):
  174. """ Unregister a file object from being monitored. """
  175. try:
  176. key = self._fd_to_key.pop(self._fileobj_lookup(fileobj))
  177. except KeyError:
  178. raise KeyError("{0!r} is not registered".format(fileobj))
  179. # Getting the fileno of a closed socket on Windows errors with EBADF.
  180. except socket.error as e: # Platform-specific: Windows.
  181. if e.errno != errno.EBADF:
  182. raise
  183. else:
  184. for key in self._fd_to_key.values():
  185. if key.fileobj is fileobj:
  186. self._fd_to_key.pop(key.fd)
  187. break
  188. else:
  189. raise KeyError("{0!r} is not registered".format(fileobj))
  190. return key
  191. def modify(self, fileobj, events, data=None):
  192. """ Change a registered file object monitored events and data. """
  193. # NOTE: Some subclasses optimize this operation even further.
  194. try:
  195. key = self._fd_to_key[self._fileobj_lookup(fileobj)]
  196. except KeyError:
  197. raise KeyError("{0!r} is not registered".format(fileobj))
  198. if events != key.events:
  199. self.unregister(fileobj)
  200. key = self.register(fileobj, events, data)
  201. elif data != key.data:
  202. # Use a shortcut to update the data.
  203. key = key._replace(data=data)
  204. self._fd_to_key[key.fd] = key
  205. return key
  206. def select(self, timeout=None):
  207. """ Perform the actual selection until some monitored file objects
  208. are ready or the timeout expires. """
  209. raise NotImplementedError()
  210. def close(self):
  211. """ Close the selector. This must be called to ensure that all
  212. underlying resources are freed. """
  213. self._fd_to_key.clear()
  214. self._map = None
  215. def get_key(self, fileobj):
  216. """ Return the key associated with a registered file object. """
  217. mapping = self.get_map()
  218. if mapping is None:
  219. raise RuntimeError("Selector is closed")
  220. try:
  221. return mapping[fileobj]
  222. except KeyError:
  223. raise KeyError("{0!r} is not registered".format(fileobj))
  224. def get_map(self):
  225. """ Return a mapping of file objects to selector keys """
  226. return self._map
  227. def _key_from_fd(self, fd):
  228. """ Return the key associated to a given file descriptor
  229. Return None if it is not found. """
  230. try:
  231. return self._fd_to_key[fd]
  232. except KeyError:
  233. return None
  234. def __enter__(self):
  235. return self
  236. def __exit__(self, *args):
  237. self.close()
  238. # Almost all platforms have select.select()
  239. if hasattr(select, "select"):
  240. class SelectSelector(BaseSelector):
  241. """ Select-based selector. """
  242. def __init__(self):
  243. super(SelectSelector, self).__init__()
  244. self._readers = set()
  245. self._writers = set()
  246. def register(self, fileobj, events, data=None):
  247. key = super(SelectSelector, self).register(fileobj, events, data)
  248. if events & EVENT_READ:
  249. self._readers.add(key.fd)
  250. if events & EVENT_WRITE:
  251. self._writers.add(key.fd)
  252. return key
  253. def unregister(self, fileobj):
  254. key = super(SelectSelector, self).unregister(fileobj)
  255. self._readers.discard(key.fd)
  256. self._writers.discard(key.fd)
  257. return key
  258. def _select(self, r, w, timeout=None):
  259. """ Wrapper for select.select because timeout is a positional arg """
  260. return select.select(r, w, [], timeout)
  261. def select(self, timeout=None):
  262. # Selecting on empty lists on Windows errors out.
  263. if not len(self._readers) and not len(self._writers):
  264. return []
  265. timeout = None if timeout is None else max(timeout, 0.0)
  266. ready = []
  267. r, w, _ = _syscall_wrapper(self._select, True, self._readers,
  268. self._writers, timeout)
  269. r = set(r)
  270. w = set(w)
  271. for fd in r | w:
  272. events = 0
  273. if fd in r:
  274. events |= EVENT_READ
  275. if fd in w:
  276. events |= EVENT_WRITE
  277. key = self._key_from_fd(fd)
  278. if key:
  279. ready.append((key, events & key.events))
  280. return ready
  281. if hasattr(select, "poll"):
  282. class PollSelector(BaseSelector):
  283. """ Poll-based selector """
  284. def __init__(self):
  285. super(PollSelector, self).__init__()
  286. self._poll = select.poll()
  287. def register(self, fileobj, events, data=None):
  288. key = super(PollSelector, self).register(fileobj, events, data)
  289. event_mask = 0
  290. if events & EVENT_READ:
  291. event_mask |= select.POLLIN
  292. if events & EVENT_WRITE:
  293. event_mask |= select.POLLOUT
  294. self._poll.register(key.fd, event_mask)
  295. return key
  296. def unregister(self, fileobj):
  297. key = super(PollSelector, self).unregister(fileobj)
  298. self._poll.unregister(key.fd)
  299. return key
  300. def _wrap_poll(self, timeout=None):
  301. """ Wrapper function for select.poll.poll() so that
  302. _syscall_wrapper can work with only seconds. """
  303. if timeout is not None:
  304. if timeout <= 0:
  305. timeout = 0
  306. else:
  307. # select.poll.poll() has a resolution of 1 millisecond,
  308. # round away from zero to wait *at least* timeout seconds.
  309. timeout = math.ceil(timeout * 1e3)
  310. result = self._poll.poll(timeout)
  311. return result
  312. def select(self, timeout=None):
  313. ready = []
  314. fd_events = _syscall_wrapper(self._wrap_poll, True, timeout=timeout)
  315. for fd, event_mask in fd_events:
  316. events = 0
  317. if event_mask & ~select.POLLIN:
  318. events |= EVENT_WRITE
  319. if event_mask & ~select.POLLOUT:
  320. events |= EVENT_READ
  321. key = self._key_from_fd(fd)
  322. if key:
  323. ready.append((key, events & key.events))
  324. return ready
  325. if hasattr(select, "epoll"):
  326. class EpollSelector(BaseSelector):
  327. """ Epoll-based selector """
  328. def __init__(self):
  329. super(EpollSelector, self).__init__()
  330. self._epoll = select.epoll()
  331. def fileno(self):
  332. return self._epoll.fileno()
  333. def register(self, fileobj, events, data=None):
  334. key = super(EpollSelector, self).register(fileobj, events, data)
  335. events_mask = 0
  336. if events & EVENT_READ:
  337. events_mask |= select.EPOLLIN
  338. if events & EVENT_WRITE:
  339. events_mask |= select.EPOLLOUT
  340. _syscall_wrapper(self._epoll.register, False, key.fd, events_mask)
  341. return key
  342. def unregister(self, fileobj):
  343. key = super(EpollSelector, self).unregister(fileobj)
  344. try:
  345. _syscall_wrapper(self._epoll.unregister, False, key.fd)
  346. except SelectorError:
  347. # This can occur when the fd was closed since registry.
  348. pass
  349. return key
  350. def select(self, timeout=None):
  351. if timeout is not None:
  352. if timeout <= 0:
  353. timeout = 0.0
  354. else:
  355. # select.epoll.poll() has a resolution of 1 millisecond
  356. # but luckily takes seconds so we don't need a wrapper
  357. # like PollSelector. Just for better rounding.
  358. timeout = math.ceil(timeout * 1e3) * 1e-3
  359. timeout = float(timeout)
  360. else:
  361. timeout = -1.0 # epoll.poll() must have a float.
  362. # We always want at least 1 to ensure that select can be called
  363. # with no file descriptors registered. Otherwise will fail.
  364. max_events = max(len(self._fd_to_key), 1)
  365. ready = []
  366. fd_events = _syscall_wrapper(self._epoll.poll, True,
  367. timeout=timeout,
  368. maxevents=max_events)
  369. for fd, event_mask in fd_events:
  370. events = 0
  371. if event_mask & ~select.EPOLLIN:
  372. events |= EVENT_WRITE
  373. if event_mask & ~select.EPOLLOUT:
  374. events |= EVENT_READ
  375. key = self._key_from_fd(fd)
  376. if key:
  377. ready.append((key, events & key.events))
  378. return ready
  379. def close(self):
  380. self._epoll.close()
  381. super(EpollSelector, self).close()
  382. if hasattr(select, "kqueue"):
  383. class KqueueSelector(BaseSelector):
  384. """ Kqueue / Kevent-based selector """
  385. def __init__(self):
  386. super(KqueueSelector, self).__init__()
  387. self._kqueue = select.kqueue()
  388. def fileno(self):
  389. return self._kqueue.fileno()
  390. def register(self, fileobj, events, data=None):
  391. key = super(KqueueSelector, self).register(fileobj, events, data)
  392. if events & EVENT_READ:
  393. kevent = select.kevent(key.fd,
  394. select.KQ_FILTER_READ,
  395. select.KQ_EV_ADD)
  396. _syscall_wrapper(self._kqueue.control, False, [kevent], 0, 0)
  397. if events & EVENT_WRITE:
  398. kevent = select.kevent(key.fd,
  399. select.KQ_FILTER_WRITE,
  400. select.KQ_EV_ADD)
  401. _syscall_wrapper(self._kqueue.control, False, [kevent], 0, 0)
  402. return key
  403. def unregister(self, fileobj):
  404. key = super(KqueueSelector, self).unregister(fileobj)
  405. if key.events & EVENT_READ:
  406. kevent = select.kevent(key.fd,
  407. select.KQ_FILTER_READ,
  408. select.KQ_EV_DELETE)
  409. try:
  410. _syscall_wrapper(self._kqueue.control, False, [kevent], 0, 0)
  411. except SelectorError:
  412. pass
  413. if key.events & EVENT_WRITE:
  414. kevent = select.kevent(key.fd,
  415. select.KQ_FILTER_WRITE,
  416. select.KQ_EV_DELETE)
  417. try:
  418. _syscall_wrapper(self._kqueue.control, False, [kevent], 0, 0)
  419. except SelectorError:
  420. pass
  421. return key
  422. def select(self, timeout=None):
  423. if timeout is not None:
  424. timeout = max(timeout, 0)
  425. max_events = len(self._fd_to_key) * 2
  426. ready_fds = {}
  427. kevent_list = _syscall_wrapper(self._kqueue.control, True,
  428. None, max_events, timeout)
  429. for kevent in kevent_list:
  430. fd = kevent.ident
  431. event_mask = kevent.filter
  432. events = 0
  433. if event_mask == select.KQ_FILTER_READ:
  434. events |= EVENT_READ
  435. if event_mask == select.KQ_FILTER_WRITE:
  436. events |= EVENT_WRITE
  437. key = self._key_from_fd(fd)
  438. if key:
  439. if key.fd not in ready_fds:
  440. ready_fds[key.fd] = (key, events & key.events)
  441. else:
  442. old_events = ready_fds[key.fd][1]
  443. ready_fds[key.fd] = (key, (events | old_events) & key.events)
  444. return list(ready_fds.values())
  445. def close(self):
  446. self._kqueue.close()
  447. super(KqueueSelector, self).close()
  448. if not hasattr(select, 'select'): # Platform-specific: AppEngine
  449. HAS_SELECT = False
  450. def _can_allocate(struct):
  451. """ Checks that select structs can be allocated by the underlying
  452. operating system, not just advertised by the select module. We don't
  453. check select() because we'll be hopeful that most platforms that
  454. don't have it available will not advertise it. (ie: GAE) """
  455. try:
  456. # select.poll() objects won't fail until used.
  457. if struct == 'poll':
  458. p = select.poll()
  459. p.poll(0)
  460. # All others will fail on allocation.
  461. else:
  462. getattr(select, struct)().close()
  463. return True
  464. except (OSError, AttributeError) as e:
  465. return False
  466. # Choose the best implementation, roughly:
  467. # kqueue == epoll > poll > select. Devpoll not supported. (See above)
  468. # select() also can't accept a FD > FD_SETSIZE (usually around 1024)
  469. def DefaultSelector():
  470. """ This function serves as a first call for DefaultSelector to
  471. detect if the select module is being monkey-patched incorrectly
  472. by eventlet, greenlet, and preserve proper behavior. """
  473. global _DEFAULT_SELECTOR
  474. if _DEFAULT_SELECTOR is None:
  475. if _can_allocate('kqueue'):
  476. _DEFAULT_SELECTOR = KqueueSelector
  477. elif _can_allocate('epoll'):
  478. _DEFAULT_SELECTOR = EpollSelector
  479. elif _can_allocate('poll'):
  480. _DEFAULT_SELECTOR = PollSelector
  481. elif hasattr(select, 'select'):
  482. _DEFAULT_SELECTOR = SelectSelector
  483. else: # Platform-specific: AppEngine
  484. raise ValueError('Platform does not have a selector')
  485. return _DEFAULT_SELECTOR()