_eventloop.py 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409
  1. from __future__ import annotations
  2. import math
  3. import sys
  4. from abc import ABCMeta, abstractmethod
  5. from collections.abc import AsyncIterator, Awaitable, Callable, Sequence
  6. from contextlib import AbstractContextManager
  7. from os import PathLike
  8. from signal import Signals
  9. from socket import AddressFamily, SocketKind, socket
  10. from typing import (
  11. IO,
  12. TYPE_CHECKING,
  13. Any,
  14. TypeAlias,
  15. TypeVar,
  16. overload,
  17. )
  18. if sys.version_info >= (3, 11):
  19. from typing import TypeVarTuple, Unpack
  20. else:
  21. from typing_extensions import TypeVarTuple, Unpack
  22. if TYPE_CHECKING:
  23. from _typeshed import FileDescriptorLike
  24. from .._core._synchronization import CapacityLimiter, Event, Lock, Semaphore
  25. from .._core._tasks import CancelScope
  26. from .._core._testing import TaskInfo
  27. from ._sockets import (
  28. ConnectedUDPSocket,
  29. ConnectedUNIXDatagramSocket,
  30. IPSockAddrType,
  31. SocketListener,
  32. SocketStream,
  33. UDPSocket,
  34. UNIXDatagramSocket,
  35. UNIXSocketStream,
  36. )
  37. from ._subprocesses import Process
  38. from ._tasks import TaskGroup
  39. from ._testing import TestRunner
  40. T_Retval = TypeVar("T_Retval")
  41. PosArgsT = TypeVarTuple("PosArgsT")
  42. StrOrBytesPath: TypeAlias = str | bytes | PathLike[str] | PathLike[bytes]
  43. class AsyncBackend(metaclass=ABCMeta):
  44. @classmethod
  45. @abstractmethod
  46. def run(
  47. cls,
  48. func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval]],
  49. args: tuple[Unpack[PosArgsT]],
  50. kwargs: dict[str, Any],
  51. options: dict[str, Any],
  52. ) -> T_Retval:
  53. """
  54. Run the given coroutine function in an asynchronous event loop.
  55. The current thread must not be already running an event loop.
  56. :param func: a coroutine function
  57. :param args: positional arguments to ``func``
  58. :param kwargs: positional arguments to ``func``
  59. :param options: keyword arguments to call the backend ``run()`` implementation
  60. with
  61. :return: the return value of the coroutine function
  62. """
  63. @classmethod
  64. @abstractmethod
  65. def current_token(cls) -> object:
  66. """
  67. Return an object that allows other threads to run code inside the event loop.
  68. :return: a token object, specific to the event loop running in the current
  69. thread
  70. """
  71. @classmethod
  72. @abstractmethod
  73. def current_time(cls) -> float:
  74. """
  75. Return the current value of the event loop's internal clock.
  76. :return: the clock value (seconds)
  77. """
  78. @classmethod
  79. @abstractmethod
  80. def cancelled_exception_class(cls) -> type[BaseException]:
  81. """Return the exception class that is raised in a task if it's cancelled."""
  82. @classmethod
  83. @abstractmethod
  84. async def checkpoint(cls) -> None:
  85. """
  86. Check if the task has been cancelled, and allow rescheduling of other tasks.
  87. This is effectively the same as running :meth:`checkpoint_if_cancelled` and then
  88. :meth:`cancel_shielded_checkpoint`.
  89. """
  90. @classmethod
  91. async def checkpoint_if_cancelled(cls) -> None:
  92. """
  93. Check if the current task group has been cancelled.
  94. This will check if the task has been cancelled, but will not allow other tasks
  95. to be scheduled if not.
  96. """
  97. if cls.current_effective_deadline() == -math.inf:
  98. await cls.checkpoint()
  99. @classmethod
  100. async def cancel_shielded_checkpoint(cls) -> None:
  101. """
  102. Allow the rescheduling of other tasks.
  103. This will give other tasks the opportunity to run, but without checking if the
  104. current task group has been cancelled, unlike with :meth:`checkpoint`.
  105. """
  106. with cls.create_cancel_scope(shield=True):
  107. await cls.sleep(0)
  108. @classmethod
  109. @abstractmethod
  110. async def sleep(cls, delay: float) -> None:
  111. """
  112. Pause the current task for the specified duration.
  113. :param delay: the duration, in seconds
  114. """
  115. @classmethod
  116. @abstractmethod
  117. def create_cancel_scope(
  118. cls, *, deadline: float = math.inf, shield: bool = False
  119. ) -> CancelScope:
  120. pass
  121. @classmethod
  122. @abstractmethod
  123. def current_effective_deadline(cls) -> float:
  124. """
  125. Return the nearest deadline among all the cancel scopes effective for the
  126. current task.
  127. :return:
  128. - a clock value from the event loop's internal clock
  129. - ``inf`` if there is no deadline in effect
  130. - ``-inf`` if the current scope has been cancelled
  131. :rtype: float
  132. """
  133. @classmethod
  134. @abstractmethod
  135. def create_task_group(cls) -> TaskGroup:
  136. pass
  137. @classmethod
  138. @abstractmethod
  139. def create_event(cls) -> Event:
  140. pass
  141. @classmethod
  142. @abstractmethod
  143. def create_lock(cls, *, fast_acquire: bool) -> Lock:
  144. pass
  145. @classmethod
  146. @abstractmethod
  147. def create_semaphore(
  148. cls,
  149. initial_value: int,
  150. *,
  151. max_value: int | None = None,
  152. fast_acquire: bool = False,
  153. ) -> Semaphore:
  154. pass
  155. @classmethod
  156. @abstractmethod
  157. def create_capacity_limiter(cls, total_tokens: float) -> CapacityLimiter:
  158. pass
  159. @classmethod
  160. @abstractmethod
  161. async def run_sync_in_worker_thread(
  162. cls,
  163. func: Callable[[Unpack[PosArgsT]], T_Retval],
  164. args: tuple[Unpack[PosArgsT]],
  165. abandon_on_cancel: bool = False,
  166. limiter: CapacityLimiter | None = None,
  167. ) -> T_Retval:
  168. pass
  169. @classmethod
  170. @abstractmethod
  171. def check_cancelled(cls) -> None:
  172. pass
  173. @classmethod
  174. @abstractmethod
  175. def run_async_from_thread(
  176. cls,
  177. func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval]],
  178. args: tuple[Unpack[PosArgsT]],
  179. token: object,
  180. ) -> T_Retval:
  181. pass
  182. @classmethod
  183. @abstractmethod
  184. def run_sync_from_thread(
  185. cls,
  186. func: Callable[[Unpack[PosArgsT]], T_Retval],
  187. args: tuple[Unpack[PosArgsT]],
  188. token: object,
  189. ) -> T_Retval:
  190. pass
  191. @classmethod
  192. @abstractmethod
  193. async def open_process(
  194. cls,
  195. command: StrOrBytesPath | Sequence[StrOrBytesPath],
  196. *,
  197. stdin: int | IO[Any] | None,
  198. stdout: int | IO[Any] | None,
  199. stderr: int | IO[Any] | None,
  200. **kwargs: Any,
  201. ) -> Process:
  202. pass
  203. @classmethod
  204. @abstractmethod
  205. def setup_process_pool_exit_at_shutdown(cls, workers: set[Process]) -> None:
  206. pass
  207. @classmethod
  208. @abstractmethod
  209. async def connect_tcp(
  210. cls, host: str, port: int, local_address: IPSockAddrType | None = None
  211. ) -> SocketStream:
  212. pass
  213. @classmethod
  214. @abstractmethod
  215. async def connect_unix(cls, path: str | bytes) -> UNIXSocketStream:
  216. pass
  217. @classmethod
  218. @abstractmethod
  219. def create_tcp_listener(cls, sock: socket) -> SocketListener:
  220. pass
  221. @classmethod
  222. @abstractmethod
  223. def create_unix_listener(cls, sock: socket) -> SocketListener:
  224. pass
  225. @classmethod
  226. @abstractmethod
  227. async def create_udp_socket(
  228. cls,
  229. family: AddressFamily,
  230. local_address: IPSockAddrType | None,
  231. remote_address: IPSockAddrType | None,
  232. reuse_port: bool,
  233. ) -> UDPSocket | ConnectedUDPSocket:
  234. pass
  235. @classmethod
  236. @overload
  237. async def create_unix_datagram_socket(
  238. cls, raw_socket: socket, remote_path: None
  239. ) -> UNIXDatagramSocket: ...
  240. @classmethod
  241. @overload
  242. async def create_unix_datagram_socket(
  243. cls, raw_socket: socket, remote_path: str | bytes
  244. ) -> ConnectedUNIXDatagramSocket: ...
  245. @classmethod
  246. @abstractmethod
  247. async def create_unix_datagram_socket(
  248. cls, raw_socket: socket, remote_path: str | bytes | None
  249. ) -> UNIXDatagramSocket | ConnectedUNIXDatagramSocket:
  250. pass
  251. @classmethod
  252. @abstractmethod
  253. async def getaddrinfo(
  254. cls,
  255. host: bytes | str | None,
  256. port: str | int | None,
  257. *,
  258. family: int | AddressFamily = 0,
  259. type: int | SocketKind = 0,
  260. proto: int = 0,
  261. flags: int = 0,
  262. ) -> Sequence[
  263. tuple[
  264. AddressFamily,
  265. SocketKind,
  266. int,
  267. str,
  268. tuple[str, int] | tuple[str, int, int, int] | tuple[int, bytes],
  269. ]
  270. ]:
  271. pass
  272. @classmethod
  273. @abstractmethod
  274. async def getnameinfo(
  275. cls, sockaddr: IPSockAddrType, flags: int = 0
  276. ) -> tuple[str, str]:
  277. pass
  278. @classmethod
  279. @abstractmethod
  280. async def wait_readable(cls, obj: FileDescriptorLike) -> None:
  281. pass
  282. @classmethod
  283. @abstractmethod
  284. async def wait_writable(cls, obj: FileDescriptorLike) -> None:
  285. pass
  286. @classmethod
  287. @abstractmethod
  288. def notify_closing(cls, obj: FileDescriptorLike) -> None:
  289. pass
  290. @classmethod
  291. @abstractmethod
  292. async def wrap_listener_socket(cls, sock: socket) -> SocketListener:
  293. pass
  294. @classmethod
  295. @abstractmethod
  296. async def wrap_stream_socket(cls, sock: socket) -> SocketStream:
  297. pass
  298. @classmethod
  299. @abstractmethod
  300. async def wrap_unix_stream_socket(cls, sock: socket) -> UNIXSocketStream:
  301. pass
  302. @classmethod
  303. @abstractmethod
  304. async def wrap_udp_socket(cls, sock: socket) -> UDPSocket:
  305. pass
  306. @classmethod
  307. @abstractmethod
  308. async def wrap_connected_udp_socket(cls, sock: socket) -> ConnectedUDPSocket:
  309. pass
  310. @classmethod
  311. @abstractmethod
  312. async def wrap_unix_datagram_socket(cls, sock: socket) -> UNIXDatagramSocket:
  313. pass
  314. @classmethod
  315. @abstractmethod
  316. async def wrap_connected_unix_datagram_socket(
  317. cls, sock: socket
  318. ) -> ConnectedUNIXDatagramSocket:
  319. pass
  320. @classmethod
  321. @abstractmethod
  322. def current_default_thread_limiter(cls) -> CapacityLimiter:
  323. pass
  324. @classmethod
  325. @abstractmethod
  326. def open_signal_receiver(
  327. cls, *signals: Signals
  328. ) -> AbstractContextManager[AsyncIterator[Signals]]:
  329. pass
  330. @classmethod
  331. @abstractmethod
  332. def get_current_task(cls) -> TaskInfo:
  333. pass
  334. @classmethod
  335. @abstractmethod
  336. def get_running_tasks(cls) -> Sequence[TaskInfo]:
  337. pass
  338. @classmethod
  339. @abstractmethod
  340. async def wait_all_tasks_blocked(cls) -> None:
  341. pass
  342. @classmethod
  343. @abstractmethod
  344. def create_test_runner(cls, options: dict[str, Any]) -> TestRunner:
  345. pass