events.py 30 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739
  1. # Copyright (C) 2023 The Qt Company Ltd.
  2. # SPDX-License-Identifier: LicenseRef-Qt-Commercial OR LGPL-3.0-only OR GPL-2.0-only OR GPL-3.0-only
  3. from __future__ import annotations
  4. from PySide6.QtCore import (QCoreApplication, QDateTime, QDeadlineTimer,
  5. QEventLoop, QObject, QTimer, QThread, Slot)
  6. from . import futures
  7. from . import tasks
  8. from typing import Any, Callable, TypeVar
  9. import asyncio
  10. import collections.abc
  11. import concurrent.futures
  12. import contextvars
  13. import enum
  14. import os
  15. import signal
  16. import socket
  17. import subprocess
  18. import warnings
  19. __all__ = [
  20. "QAsyncioEventLoopPolicy", "QAsyncioEventLoop",
  21. "QAsyncioHandle", "QAsyncioTimerHandle",
  22. ]
  23. from typing import TYPE_CHECKING
  24. _T = TypeVar("_T")
  25. if TYPE_CHECKING:
  26. try:
  27. from typing import TypeVarTuple, Unpack
  28. except ImportError:
  29. from typing_extensions import TypeVarTuple, Unpack # type: ignore
  30. _Ts = TypeVarTuple("_Ts")
  31. Context = contextvars.Context # type: ignore
  32. else:
  33. _Ts = None # type: ignore
  34. Context = contextvars.Context
  35. class QAsyncioExecutorWrapper(QObject):
  36. """
  37. Executors in asyncio allow running synchronous code in a separate thread or
  38. process without blocking the event loop or interrupting the asynchronous
  39. program flow. Callables are scheduled for execution by calling submit() or
  40. map() on an executor object.
  41. Executors require a bit of extra work for QtAsyncio, as we can't use
  42. naked Python threads; instead, we must make sure that the thread created
  43. by executor.submit() has an event loop. This is achieved by not submitting
  44. the callable directly, but a small wrapper that attaches a QEventLoop to
  45. the executor thread, and then creates a zero-delay singleshot timer to push
  46. the actual callable for the executor into this new event loop.
  47. """
  48. def __init__(self, func: Callable[[Unpack[_Ts]], Any], *args: Unpack[_Ts]) -> None:
  49. super().__init__()
  50. self._loop: QEventLoop
  51. self._func = func
  52. self._args = args
  53. self._result: Any = None
  54. self._exception: BaseException | None = None
  55. def _cb(self):
  56. try:
  57. # Call the synchronous callable that we submitted with submit() or
  58. # map().
  59. self._result = self._func(*self._args)
  60. except BaseException as e:
  61. self._exception = e
  62. self._loop.exit()
  63. def do(self) -> Any:
  64. # This creates a new event loop and dispatcher for the thread, if not
  65. # already created.
  66. self._loop = QEventLoop()
  67. asyncio.events._set_running_loop(self._loop)
  68. # The do() function will always be executed from the new executor
  69. # thread and never from outside, so using the overload without the
  70. # context argument is sufficient.
  71. QTimer.singleShot(0, lambda: self._cb())
  72. self._loop.exec()
  73. if self._exception is not None:
  74. raise self._exception
  75. return self._result
  76. def exit(self):
  77. self._loop.exit()
  78. class QAsyncioEventLoopPolicy(asyncio.AbstractEventLoopPolicy):
  79. """
  80. Event loop policies are expected to be deprecated with Python 3.13, with
  81. subsequent removal in Python 3.15. At that point, part of the current
  82. logic of the QAsyncioEventLoopPolicy constructor will have to be moved
  83. to QtAsyncio.run() and/or to a loop factory class (to be provided as an
  84. argument to asyncio.run()). In particular, this concerns the logic of
  85. setting up the QCoreApplication and the SIGINT handler.
  86. More details:
  87. https://discuss.python.org/t/removing-the-asyncio-policy-system-asyncio-set-event-loop-policy-in-python-3-15/37553
  88. """
  89. def __init__(self,
  90. quit_qapp: bool = True,
  91. handle_sigint: bool = False) -> None:
  92. super().__init__()
  93. self._application = QCoreApplication.instance() or QCoreApplication()
  94. # Configure whether the QCoreApplication at the core of QtAsyncio
  95. # should be shut down when asyncio finishes. A special case where one
  96. # would want to disable this is test suites that want to reuse a single
  97. # QCoreApplication instance across all unit tests, which would fail if
  98. # this instance is shut down every time.
  99. self._quit_qapp = quit_qapp
  100. self._event_loop: asyncio.AbstractEventLoop | None = None
  101. if handle_sigint:
  102. signal.signal(signal.SIGINT, signal.SIG_DFL)
  103. def get_event_loop(self) -> asyncio.AbstractEventLoop:
  104. if self._event_loop is None:
  105. self._event_loop = QAsyncioEventLoop(self._application, quit_qapp=self._quit_qapp)
  106. return self._event_loop
  107. def set_event_loop(self, loop: asyncio.AbstractEventLoop | None) -> None:
  108. self._event_loop = loop
  109. def new_event_loop(self) -> asyncio.AbstractEventLoop:
  110. return QAsyncioEventLoop(self._application, quit_qapp=self._quit_qapp)
  111. def get_child_watcher(self) -> "asyncio.AbstractChildWatcher":
  112. raise DeprecationWarning("Child watchers are deprecated since Python 3.12")
  113. def set_child_watcher(self, watcher: "asyncio.AbstractChildWatcher") -> None:
  114. raise DeprecationWarning("Child watchers are deprecated since Python 3.12")
  115. class QAsyncioEventLoop(asyncio.BaseEventLoop, QObject):
  116. """
  117. Implements the asyncio API:
  118. https://docs.python.org/3/library/asyncio-eventloop.html
  119. """
  120. class ShutDownThread(QThread):
  121. """
  122. Used to shut down the default executor when calling
  123. shutdown_default_executor(). As the executor is a ThreadPoolExecutor,
  124. it must be shut down in a separate thread as all the threads from the
  125. thread pool must join, which we want to do without blocking the event
  126. loop.
  127. """
  128. def __init__(self, future: futures.QAsyncioFuture, loop: "QAsyncioEventLoop") -> None:
  129. super().__init__()
  130. self._future = future
  131. self._loop = loop
  132. self.started.connect(self.shutdown)
  133. def run(self) -> None:
  134. pass
  135. def shutdown(self) -> None:
  136. try:
  137. self._loop._default_executor.shutdown(wait=True)
  138. if not self._loop.is_closed():
  139. self._loop.call_soon_threadsafe(self._future.set_result, None)
  140. except Exception as e:
  141. if not self._loop.is_closed():
  142. self._loop.call_soon_threadsafe(self._future.set_exception, e)
  143. def __init__(self,
  144. application: QCoreApplication, quit_qapp: bool = True) -> None:
  145. asyncio.BaseEventLoop.__init__(self)
  146. QObject.__init__(self)
  147. self._application: QCoreApplication = application
  148. # Configure whether the QCoreApplication at the core of QtAsyncio
  149. # should be shut down when asyncio finishes. A special case where one
  150. # would want to disable this is test suites that want to reuse a single
  151. # QCoreApplication instance across all unit tests, which would fail if
  152. # this instance is shut down every time.
  153. self._quit_qapp = quit_qapp
  154. self._thread = QThread.currentThread()
  155. self._closed = False
  156. # These two flags are used to determine whether the loop was stopped
  157. # from inside the loop (i.e., coroutine or callback called stop()) or
  158. # from outside the loop (i.e., the QApplication is being shut down, for
  159. # example, by the user closing the window or by calling
  160. # QApplication.quit()). The different cases can trigger slightly
  161. # different behaviors (see the comments where the flags are used).
  162. # There are two variables for this as in a third case the loop is still
  163. # running and both flags are False.
  164. self._quit_from_inside = False
  165. self._quit_from_outside = False
  166. # A set of all asynchronous generators that are currently running.
  167. self._asyncgens: set[collections.abc.AsyncGenerator] = set()
  168. # Starting with Python 3.11, this must be an instance of
  169. # ThreadPoolExecutor.
  170. self._default_executor = concurrent.futures.ThreadPoolExecutor()
  171. # The exception handler, if set with set_exception_handler(). The
  172. # exception handler is currently called in two places: One, if an
  173. # asynchonrous generator raises an exception when closed, and two, if
  174. # an exception is raised during the execution of a task. Currently, the
  175. # default exception handler just prints the exception to the console.
  176. self._exception_handler: Callable | None = self.default_exception_handler
  177. # The task factory, if set with set_task_factory(). Otherwise, a new
  178. # task is created with the QAsyncioTask constructor.
  179. self._task_factory: Callable | None = None
  180. # The future that is currently being awaited with run_until_complete().
  181. self._future_to_complete: futures.QAsyncioFuture | None = None
  182. self._debug = bool(os.getenv("PYTHONASYNCIODEBUG", False))
  183. self._application.aboutToQuit.connect(self._about_to_quit_cb)
  184. # Running and stopping the loop
  185. def _run_until_complete_cb(self, future: futures.QAsyncioFuture) -> None:
  186. """
  187. A callback that stops the loop when the future is done, used when
  188. running the loop with run_until_complete().
  189. """
  190. if not future.cancelled():
  191. if isinstance(future.exception(), (SystemExit, KeyboardInterrupt)):
  192. return
  193. future.get_loop().stop()
  194. def run_until_complete(self,
  195. future: futures.QAsyncioFuture) -> Any: # type: ignore[override]
  196. if self.is_closed():
  197. raise RuntimeError("Event loop is closed")
  198. if self.is_running():
  199. raise RuntimeError("Event loop is already running")
  200. arg_was_coro = not asyncio.futures.isfuture(future)
  201. future = asyncio.tasks.ensure_future(future, loop=self) # type: ignore[assignment]
  202. future.add_done_callback(self._run_until_complete_cb)
  203. self._future_to_complete = future
  204. try:
  205. self.run_forever()
  206. except Exception as e:
  207. if arg_was_coro and future.done() and not future.cancelled():
  208. future.exception()
  209. raise e
  210. finally:
  211. future.remove_done_callback(self._run_until_complete_cb)
  212. if not future.done():
  213. raise RuntimeError("Event loop stopped before Future completed")
  214. return future.result()
  215. def run_forever(self) -> None:
  216. if self.is_closed():
  217. raise RuntimeError("Event loop is closed")
  218. if self.is_running():
  219. raise RuntimeError("Event loop is already running")
  220. asyncio.events._set_running_loop(self)
  221. self._application.exec()
  222. asyncio.events._set_running_loop(None)
  223. def _about_to_quit_cb(self):
  224. """ A callback for the aboutToQuit signal of the QCoreApplication. """
  225. if not self._quit_from_inside:
  226. # If the aboutToQuit signal is emitted, the user is closing the
  227. # application window or calling QApplication.quit(). In this case,
  228. # we want to close the event loop, and we consider this a quit from
  229. # outside the loop.
  230. self._quit_from_outside = True
  231. self.close()
  232. def stop(self) -> None:
  233. if self._future_to_complete is not None:
  234. if self._future_to_complete.done():
  235. self._future_to_complete = None
  236. else:
  237. # Do not stop the loop if there is a future still being awaited
  238. # with run_until_complete().
  239. return
  240. self._quit_from_inside = True
  241. # The user might want to keep the QApplication running after the event
  242. # event loop finishes, which they can control with the quit_qapp
  243. # argument.
  244. if self._quit_qapp:
  245. self._application.quit()
  246. def is_running(self) -> bool:
  247. return self._thread.loopLevel() > 0
  248. def is_closed(self) -> bool:
  249. return self._closed
  250. def close(self) -> None:
  251. if self.is_running() and not self._quit_from_outside:
  252. raise RuntimeError("Cannot close a running event loop")
  253. if self.is_closed():
  254. return
  255. if self._default_executor is not None:
  256. self._default_executor.shutdown(wait=False)
  257. self._closed = True
  258. async def shutdown_asyncgens(self) -> None:
  259. if not len(self._asyncgens):
  260. return
  261. results = await asyncio.tasks.gather(
  262. *[asyncgen.aclose() for asyncgen in self._asyncgens],
  263. return_exceptions=True)
  264. for result, asyncgen in zip(results, self._asyncgens):
  265. if isinstance(result, Exception):
  266. self.call_exception_handler({
  267. "message": f"Closing asynchronous generator {asyncgen}"
  268. f"raised an exception",
  269. "exception": result,
  270. "asyncgen": asyncgen})
  271. self._asyncgens.clear()
  272. async def shutdown_default_executor(self, # type: ignore[override]
  273. timeout: int | float | None = None) -> None:
  274. shutdown_successful = False
  275. if timeout is not None:
  276. deadline_timer = QDeadlineTimer(int(timeout * 1000))
  277. else:
  278. deadline_timer = QDeadlineTimer(QDeadlineTimer.ForeverConstant.Forever)
  279. if self._default_executor is None:
  280. return
  281. future = self.create_future()
  282. thread = QAsyncioEventLoop.ShutDownThread(future, self)
  283. thread.start()
  284. try:
  285. await future
  286. finally:
  287. shutdown_successful = thread.wait(deadline_timer)
  288. if timeout is not None and not shutdown_successful:
  289. warnings.warn(
  290. f"Could not shutdown the default executor within {timeout} seconds",
  291. RuntimeWarning, stacklevel=2)
  292. self._default_executor.shutdown(wait=False)
  293. # Scheduling callbacks
  294. def _call_soon_impl(self, callback: Callable[[Unpack[_Ts]], object], *args: Unpack[_Ts],
  295. context: Context | None = None,
  296. is_threadsafe: bool | None = False) -> asyncio.Handle:
  297. return self._call_later_impl(0, callback, *args, context=context,
  298. is_threadsafe=is_threadsafe)
  299. def call_soon(self, callback: Callable[[Unpack[_Ts]], object], *args: Unpack[_Ts],
  300. context: Context | None = None) -> asyncio.Handle:
  301. return self._call_soon_impl(callback, *args, context=context, is_threadsafe=False)
  302. def call_soon_threadsafe(self, callback: Callable[[Unpack[_Ts]], object], *args: Unpack[_Ts],
  303. context: Context | None = None) -> asyncio.Handle:
  304. if self.is_closed():
  305. raise RuntimeError("Event loop is closed")
  306. if context is None:
  307. context = contextvars.copy_context()
  308. return self._call_soon_impl(callback, *args, context=context, is_threadsafe=True)
  309. def _call_later_impl(self, delay: float, callback: Callable[[Unpack[_Ts]], object],
  310. *args: Unpack[_Ts], context: Context | None = None,
  311. is_threadsafe: bool | None = False) -> asyncio.TimerHandle:
  312. if not isinstance(delay, (int, float)):
  313. raise TypeError("delay must be an int or float")
  314. return self._call_at_impl(self.time() + delay, callback, *args,
  315. context=context, is_threadsafe=is_threadsafe)
  316. def call_later(self, delay: float, callback: Callable[[Unpack[_Ts]], object],
  317. *args: Unpack[_Ts], context: Context | None = None) -> asyncio.TimerHandle:
  318. return self._call_later_impl(delay, callback, *args, context=context, is_threadsafe=False)
  319. def _call_at_impl(self, when: float, callback: Callable[[Unpack[_Ts]], object],
  320. *args: Unpack[_Ts], context: Context | None = None,
  321. is_threadsafe: bool | None = False) -> asyncio.TimerHandle:
  322. """ All call_at() and call_later() methods map to this method. """
  323. if not isinstance(when, (int, float)):
  324. raise TypeError("when must be an int or float")
  325. return QAsyncioTimerHandle(when, callback, args, self, context, is_threadsafe=is_threadsafe)
  326. def call_at(self, when: float, callback: Callable[[Unpack[_Ts]], object],
  327. *args: Unpack[_Ts], context: Context | None = None) -> asyncio.TimerHandle:
  328. return self._call_at_impl(when, callback, *args, context=context, is_threadsafe=False)
  329. def time(self) -> float:
  330. return QDateTime.currentMSecsSinceEpoch() / 1000.0
  331. # Creating Futures and Tasks
  332. def create_future(self) -> futures.QAsyncioFuture: # type: ignore[override]
  333. return futures.QAsyncioFuture(loop=self)
  334. def create_task(self, # type: ignore[override]
  335. coro: collections.abc.Generator | collections.abc.Coroutine,
  336. *, name: str | None = None,
  337. context: contextvars.Context | None = None) -> tasks.QAsyncioTask:
  338. if self._task_factory is None:
  339. task = tasks.QAsyncioTask(coro, loop=self, name=name, context=context)
  340. else:
  341. task = self._task_factory(self, coro, context=context)
  342. task.set_name(name)
  343. return task
  344. def set_task_factory(self, factory: Callable | None) -> None:
  345. if factory is not None and not callable(factory):
  346. raise TypeError("The task factory must be a callable or None")
  347. self._task_factory = factory
  348. def get_task_factory(self) -> Callable | None:
  349. return self._task_factory
  350. # Opening network connections
  351. async def create_connection(
  352. self, protocol_factory, host=None, port=None,
  353. *, ssl=None, family=0, proto=0,
  354. flags=0, sock=None, local_addr=None,
  355. server_hostname=None,
  356. ssl_handshake_timeout=None,
  357. ssl_shutdown_timeout=None,
  358. happy_eyeballs_delay=None, interleave=None):
  359. raise NotImplementedError("QAsyncioEventLoop.create_connection() is not implemented yet")
  360. async def create_datagram_endpoint(self, protocol_factory,
  361. local_addr=None, remote_addr=None, *,
  362. family=0, proto=0, flags=0,
  363. reuse_address=None, reuse_port=None,
  364. allow_broadcast=None, sock=None):
  365. raise NotImplementedError(
  366. "QAsyncioEventLoop.create_datagram_endpoint() is not implemented yet")
  367. async def create_unix_connection(
  368. self, protocol_factory, path=None, *,
  369. ssl=None, sock=None,
  370. server_hostname=None,
  371. ssl_handshake_timeout=None,
  372. ssl_shutdown_timeout=None):
  373. raise NotImplementedError(
  374. "QAsyncioEventLoop.create_unix_connection() is not implemented yet")
  375. # Creating network servers
  376. async def create_server(
  377. self, protocol_factory, host=None, port=None,
  378. *, family=socket.AF_UNSPEC,
  379. flags=socket.AI_PASSIVE, sock=None, backlog=100,
  380. ssl=None, reuse_address=None, reuse_port=None,
  381. ssl_handshake_timeout=None,
  382. ssl_shutdown_timeout=None,
  383. start_serving=True):
  384. raise NotImplementedError("QAsyncioEventLoop.create_server() is not implemented yet")
  385. async def create_unix_server(
  386. self, protocol_factory, path=None, *,
  387. sock=None, backlog=100, ssl=None,
  388. ssl_handshake_timeout=None,
  389. ssl_shutdown_timeout=None,
  390. start_serving=True):
  391. raise NotImplementedError("QAsyncioEventLoop.create_unix_server() is not implemented yet")
  392. async def connect_accepted_socket(
  393. self, protocol_factory, sock,
  394. *, ssl=None,
  395. ssl_handshake_timeout=None,
  396. ssl_shutdown_timeout=None):
  397. raise NotImplementedError(
  398. "QAsyncioEventLoop.connect_accepted_socket() is not implemented yet")
  399. # Transferring files
  400. async def sendfile(self, transport, file, offset=0, count=None,
  401. *, fallback=True):
  402. raise NotImplementedError("QAsyncioEventLoop.sendfile() is not implemented yet")
  403. # TLS Upgrade
  404. async def start_tls(self, transport, protocol, sslcontext, *,
  405. server_side=False,
  406. server_hostname=None,
  407. ssl_handshake_timeout=None,
  408. ssl_shutdown_timeout=None):
  409. raise NotImplementedError("QAsyncioEventLoop.start_tls() is not implemented yet")
  410. # Watching file descriptors
  411. def add_reader(self, fd, callback, *args):
  412. raise NotImplementedError("QAsyncioEventLoop.add_reader() is not implemented yet")
  413. def remove_reader(self, fd):
  414. raise NotImplementedError("QAsyncioEventLoop.remove_reader() is not implemented yet")
  415. def add_writer(self, fd, callback, *args):
  416. raise NotImplementedError("QAsyncioEventLoop.add_writer() is not implemented yet")
  417. def remove_writer(self, fd):
  418. raise NotImplementedError("QAsyncioEventLoop.remove_writer() is not implemented yet")
  419. # Working with socket objects directly
  420. async def sock_recv(self, sock, nbytes):
  421. raise NotImplementedError("QAsyncioEventLoop.sock_recv() is not implemented yet")
  422. async def sock_recv_into(self, sock, buf):
  423. raise NotImplementedError("QAsyncioEventLoop.sock_recv_into() is not implemented yet")
  424. async def sock_recvfrom(self, sock, bufsize):
  425. raise NotImplementedError("QAsyncioEventLoop.sock_recvfrom() is not implemented yet")
  426. async def sock_recvfrom_into(self, sock, buf, nbytes=0):
  427. raise NotImplementedError("QAsyncioEventLoop.sock_recvfrom_into() is not implemented yet")
  428. async def sock_sendall(self, sock, data):
  429. raise NotImplementedError("QAsyncioEventLoop.sock_sendall() is not implemented yet")
  430. async def sock_sendto(self, sock, data, address):
  431. raise NotImplementedError("QAsyncioEventLoop.sock_sendto() is not implemented yet")
  432. async def sock_connect(self, sock, address):
  433. raise NotImplementedError("QAsyncioEventLoop.sock_connect() is not implemented yet")
  434. async def sock_accept(self, sock):
  435. raise NotImplementedError("QAsyncioEventLoop.sock_accept() is not implemented yet")
  436. async def sock_sendfile(self, sock, file, offset=0, count=None, *,
  437. fallback=None):
  438. raise NotImplementedError("QAsyncioEventLoop.sock_sendfile() is not implemented yet")
  439. # DNS
  440. async def getaddrinfo(self, host, port, *,
  441. family=0, type=0, proto=0, flags=0):
  442. raise NotImplementedError("QAsyncioEventLoop.getaddrinfo() is not implemented yet")
  443. async def getnameinfo(self, sockaddr, flags=0):
  444. raise NotImplementedError("QAsyncioEventLoop.getnameinfo() is not implemented yet")
  445. # Working with pipes
  446. async def connect_read_pipe(self, protocol_factory, pipe):
  447. raise NotImplementedError("QAsyncioEventLoop.connect_read_pipe() is not implemented yet")
  448. async def connect_write_pipe(self, protocol_factory, pipe):
  449. raise NotImplementedError("QAsyncioEventLoop.connect_write_pipe() is not implemented yet")
  450. # Unix signals
  451. def add_signal_handler(self, sig, callback, *args):
  452. raise NotImplementedError("QAsyncioEventLoop.add_signal_handler() is not implemented yet")
  453. def remove_signal_handler(self, sig):
  454. raise NotImplementedError(
  455. "QAsyncioEventLoop.remove_signal_handler() is not implemented yet")
  456. # Executing code in thread or process pools
  457. def run_in_executor(self, executor: concurrent.futures.ThreadPoolExecutor | None,
  458. func: Callable[[Unpack[_Ts]], _T],
  459. *args: Unpack[_Ts]) -> asyncio.Future[_T]:
  460. if self.is_closed():
  461. raise RuntimeError("Event loop is closed")
  462. if executor is None:
  463. executor = self._default_executor
  464. # Executors require a bit of extra work for QtAsyncio, as we can't use
  465. # naked Python threads; instead, we must make sure that the thread
  466. # created by executor.submit() has an event loop. This is achieved by
  467. # not submitting the callable directly, but a small wrapper that
  468. # attaches a QEventLoop to the executor thread, and then pushes the
  469. # actual callable for the executor into this new event loop.
  470. wrapper = QAsyncioExecutorWrapper(func, *args)
  471. return asyncio.futures.wrap_future(executor.submit(wrapper.do), loop=self)
  472. def set_default_executor(self,
  473. executor: concurrent.futures.ThreadPoolExecutor | None) -> None:
  474. if not isinstance(executor, concurrent.futures.ThreadPoolExecutor):
  475. raise TypeError("The executor must be a ThreadPoolExecutor")
  476. self._default_executor = executor
  477. # Error Handling API
  478. def set_exception_handler(self, handler: Callable | None) -> None:
  479. if handler is not None and not callable(handler):
  480. raise TypeError("The handler must be a callable or None")
  481. self._exception_handler = handler
  482. def get_exception_handler(self) -> Callable | None:
  483. return self._exception_handler
  484. def default_exception_handler(self, context: dict[str, Any]) -> None:
  485. # TODO
  486. if context["message"]:
  487. print(f"{context['message']} from task {context['task']._name},"
  488. "read the following traceback:")
  489. print(context["traceback"])
  490. def call_exception_handler(self, context: dict[str, Any]) -> None:
  491. if self._exception_handler is not None:
  492. self._exception_handler(context)
  493. # Enabling debug mode
  494. def get_debug(self) -> bool:
  495. # TODO: Part of the asyncio API but currently unused. More details:
  496. # https://docs.python.org/3/library/asyncio-dev.html#asyncio-debug-mode
  497. return self._debug
  498. def set_debug(self, enabled: bool) -> None:
  499. self._debug = enabled
  500. # Running subprocesses
  501. async def subprocess_exec(self, protocol_factory, *args,
  502. stdin=subprocess.PIPE,
  503. stdout=subprocess.PIPE,
  504. stderr=subprocess.PIPE,
  505. **kwargs):
  506. raise NotImplementedError("QAsyncioEventLoop.subprocess_exec() is not implemented yet")
  507. async def subprocess_shell(self, protocol_factory, cmd, *,
  508. stdin=subprocess.PIPE,
  509. stdout=subprocess.PIPE,
  510. stderr=subprocess.PIPE,
  511. **kwargs):
  512. raise NotImplementedError("QAsyncioEventLoop.subprocess_shell() is not implemented yet")
  513. class QAsyncioHandle():
  514. """
  515. The handle enqueues a callback to be executed by the event loop, and allows
  516. for this callback to be cancelled before it is executed. This callback will
  517. typically execute the step function for a task. This makes the handle one
  518. of the main components of asyncio.
  519. """
  520. class HandleState(enum.Enum):
  521. PENDING = enum.auto()
  522. CANCELLED = enum.auto()
  523. DONE = enum.auto()
  524. def __init__(self, callback: Callable, args: tuple,
  525. loop: QAsyncioEventLoop, context: contextvars.Context | None,
  526. is_threadsafe: bool | None = False) -> None:
  527. self._callback = callback
  528. self._cb_args = args # renamed from _args to avoid conflict with TimerHandle._args
  529. self._loop = loop
  530. self._context = context
  531. self._is_threadsafe = is_threadsafe
  532. self._timeout = 0
  533. self._state = QAsyncioHandle.HandleState.PENDING
  534. self._start()
  535. def _start(self) -> None:
  536. self._schedule_event(self._timeout, lambda: self._cb())
  537. def _schedule_event(self, timeout: int, func: Callable) -> None:
  538. # Do not schedule events from asyncio when the app is quit from outside
  539. # the event loop, as this would cause events to be enqueued after the
  540. # event loop was destroyed.
  541. if not self._loop.is_closed() and not self._loop._quit_from_outside:
  542. if self._is_threadsafe:
  543. # This singleShot overload will push func into self._loop
  544. # instead of the current thread's loop. This allows scheduling
  545. # a callback from a different thread, which is necessary for
  546. # thread-safety.
  547. # https://docs.python.org/3/library/asyncio-dev.html#asyncio-multithreading
  548. QTimer.singleShot(timeout, self._loop, func)
  549. else:
  550. QTimer.singleShot(timeout, func)
  551. @Slot()
  552. def _cb(self) -> None:
  553. """
  554. A slot, enqueued into the event loop, that wraps around the actual
  555. callback, typically the step function of a task.
  556. """
  557. if self._state == QAsyncioHandle.HandleState.PENDING:
  558. if self._context is not None:
  559. self._context.run(self._callback, *self._cb_args)
  560. else:
  561. self._callback(*self._cb_args)
  562. self._state = QAsyncioHandle.HandleState.DONE
  563. def cancel(self) -> None:
  564. if self._state == QAsyncioHandle.HandleState.PENDING:
  565. # The old timer that was created in _start will still trigger but
  566. # _cb won't do anything, therefore the callback is effectively
  567. # cancelled.
  568. self._state = QAsyncioHandle.HandleState.CANCELLED
  569. def cancelled(self) -> bool:
  570. return self._state == QAsyncioHandle.HandleState.CANCELLED
  571. class QAsyncioTimerHandle(QAsyncioHandle, asyncio.TimerHandle):
  572. def __init__(self, when: float, callback: Callable, args: tuple,
  573. loop: QAsyncioEventLoop, context: contextvars.Context | None,
  574. is_threadsafe: bool | None = False) -> None:
  575. QAsyncioHandle.__init__(self, callback, args, loop, context, is_threadsafe)
  576. self._when = when
  577. time = self._loop.time()
  578. # PYSIDE-2644: Timeouts should be rounded up or down instead of only up
  579. # as happens with int(). Otherwise, a timeout of e.g. 0.9 would be
  580. # handled as 0, where 1 would be more appropriate.
  581. self._timeout = round(max(self._when - time, 0) * 1000)
  582. QAsyncioHandle._start(self)
  583. def _start(self) -> None:
  584. """
  585. Overridden so that timer.start() is only called once at the end of the
  586. constructor for both QtHandle and QtTimerHandle.
  587. """
  588. pass
  589. def when(self) -> float:
  590. return self._when