asyncio.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376
  1. """An asyncio-based implementation of the file lock."""
  2. from __future__ import annotations
  3. import asyncio
  4. import contextlib
  5. import logging
  6. import os
  7. import time
  8. from dataclasses import dataclass
  9. from inspect import iscoroutinefunction
  10. from threading import local
  11. from typing import TYPE_CHECKING, Any, NoReturn, cast
  12. from ._api import _UNSET_FILE_MODE, BaseFileLock, FileLockContext, FileLockMeta
  13. from ._error import Timeout
  14. from ._soft import SoftFileLock
  15. from ._unix import UnixFileLock
  16. from ._windows import WindowsFileLock
  17. if TYPE_CHECKING:
  18. import sys
  19. from collections.abc import Callable
  20. from concurrent import futures
  21. from types import TracebackType
  22. if sys.version_info >= (3, 11): # pragma: no cover (py311+)
  23. from typing import Self
  24. else: # pragma: no cover (<py311)
  25. from typing_extensions import Self
  26. _LOGGER = logging.getLogger("filelock")
  27. @dataclass
  28. class AsyncFileLockContext(FileLockContext):
  29. """A dataclass which holds the context for a ``BaseAsyncFileLock`` object."""
  30. #: Whether run in executor
  31. run_in_executor: bool = True
  32. #: The executor
  33. executor: futures.Executor | None = None
  34. #: The loop
  35. loop: asyncio.AbstractEventLoop | None = None
  36. class AsyncThreadLocalFileContext(AsyncFileLockContext, local):
  37. """A thread local version of the ``FileLockContext`` class."""
  38. class AsyncAcquireReturnProxy:
  39. """A context-aware object that will release the lock file when exiting."""
  40. def __init__(self, lock: BaseAsyncFileLock) -> None: # noqa: D107
  41. self.lock = lock
  42. async def __aenter__(self) -> BaseAsyncFileLock: # noqa: D105
  43. return self.lock
  44. async def __aexit__( # noqa: D105
  45. self,
  46. exc_type: type[BaseException] | None,
  47. exc_value: BaseException | None,
  48. traceback: TracebackType | None,
  49. ) -> None:
  50. await self.lock.release()
  51. class AsyncFileLockMeta(FileLockMeta):
  52. def __call__( # ty: ignore[invalid-method-override] # noqa: PLR0913
  53. cls, # noqa: N805
  54. lock_file: str | os.PathLike[str],
  55. timeout: float = -1,
  56. mode: int = _UNSET_FILE_MODE,
  57. thread_local: bool = False, # noqa: FBT001, FBT002
  58. *,
  59. blocking: bool = True,
  60. is_singleton: bool = False,
  61. poll_interval: float = 0.05,
  62. lifetime: float | None = None,
  63. loop: asyncio.AbstractEventLoop | None = None,
  64. run_in_executor: bool = True,
  65. executor: futures.Executor | None = None,
  66. ) -> BaseAsyncFileLock:
  67. if thread_local and run_in_executor:
  68. msg = "run_in_executor is not supported when thread_local is True"
  69. raise ValueError(msg)
  70. instance = super().__call__(
  71. lock_file=lock_file,
  72. timeout=timeout,
  73. mode=mode,
  74. thread_local=thread_local,
  75. blocking=blocking,
  76. is_singleton=is_singleton,
  77. poll_interval=poll_interval,
  78. lifetime=lifetime,
  79. loop=loop,
  80. run_in_executor=run_in_executor,
  81. executor=executor,
  82. )
  83. return cast("BaseAsyncFileLock", instance)
  84. class BaseAsyncFileLock(BaseFileLock, metaclass=AsyncFileLockMeta):
  85. """
  86. Base class for asynchronous file locks.
  87. .. versionadded:: 3.15.0
  88. """
  89. def __init__( # noqa: PLR0913
  90. self,
  91. lock_file: str | os.PathLike[str],
  92. timeout: float = -1,
  93. mode: int = _UNSET_FILE_MODE,
  94. thread_local: bool = False, # noqa: FBT001, FBT002
  95. *,
  96. blocking: bool = True,
  97. is_singleton: bool = False,
  98. poll_interval: float = 0.05,
  99. lifetime: float | None = None,
  100. loop: asyncio.AbstractEventLoop | None = None,
  101. run_in_executor: bool = True,
  102. executor: futures.Executor | None = None,
  103. ) -> None:
  104. """
  105. Create a new lock object.
  106. :param lock_file: path to the file
  107. :param timeout: default timeout when acquiring the lock, in seconds. It will be used as fallback value in the
  108. acquire method, if no timeout value (``None``) is given. If you want to disable the timeout, set it to a
  109. negative value. A timeout of 0 means that there is exactly one attempt to acquire the file lock.
  110. :param mode: file permissions for the lockfile. When not specified, the OS controls permissions via umask and
  111. default ACLs, preserving POSIX default ACL inheritance in shared directories.
  112. :param thread_local: Whether this object's internal context should be thread local or not. If this is set to
  113. ``False`` then the lock will be reentrant across threads.
  114. :param blocking: whether the lock should be blocking or not
  115. :param is_singleton: If this is set to ``True`` then only one instance of this class will be created per lock
  116. file. This is useful if you want to use the lock object for reentrant locking without needing to pass the
  117. same object around.
  118. :param poll_interval: default interval for polling the lock file, in seconds. It will be used as fallback value
  119. in the acquire method, if no poll_interval value (``None``) is given.
  120. :param lifetime: maximum time in seconds a lock can be held before it is considered expired. When set, a waiting
  121. process will break a lock whose file modification time is older than ``lifetime`` seconds. ``None`` (the
  122. default) means locks never expire.
  123. :param loop: The event loop to use. If not specified, the running event loop will be used.
  124. :param run_in_executor: If this is set to ``True`` then the lock will be acquired in an executor.
  125. :param executor: The executor to use. If not specified, the default executor will be used.
  126. """
  127. self._is_thread_local = thread_local
  128. self._is_singleton = is_singleton
  129. # Create the context. Note that external code should not work with the context directly and should instead use
  130. # properties of this class.
  131. kwargs: dict[str, Any] = {
  132. "lock_file": os.fspath(lock_file),
  133. "timeout": timeout,
  134. "mode": mode,
  135. "blocking": blocking,
  136. "poll_interval": poll_interval,
  137. "lifetime": lifetime,
  138. "loop": loop,
  139. "run_in_executor": run_in_executor,
  140. "executor": executor,
  141. }
  142. self._context: AsyncFileLockContext = (AsyncThreadLocalFileContext if thread_local else AsyncFileLockContext)(
  143. **kwargs
  144. )
  145. @property
  146. def run_in_executor(self) -> bool:
  147. """:returns: whether run in executor."""
  148. return self._context.run_in_executor
  149. @property
  150. def executor(self) -> futures.Executor | None:
  151. """:returns: the executor."""
  152. return self._context.executor
  153. @executor.setter
  154. def executor(self, value: futures.Executor | None) -> None: # pragma: no cover
  155. """
  156. Change the executor.
  157. :param futures.Executor | None value: the new executor or ``None``
  158. """
  159. self._context.executor = value
  160. @property
  161. def loop(self) -> asyncio.AbstractEventLoop | None:
  162. """:returns: the event loop."""
  163. return self._context.loop
  164. async def acquire( # ty: ignore[invalid-method-override]
  165. self,
  166. timeout: float | None = None,
  167. poll_interval: float | None = None,
  168. *,
  169. blocking: bool | None = None,
  170. cancel_check: Callable[[], bool] | None = None,
  171. ) -> AsyncAcquireReturnProxy:
  172. """
  173. Try to acquire the file lock.
  174. :param timeout: maximum wait time for acquiring the lock, ``None`` means use the default
  175. :attr:`~BaseFileLock.timeout` is and if ``timeout < 0``, there is no timeout and this method will block
  176. until the lock could be acquired
  177. :param poll_interval: interval of trying to acquire the lock file, ``None`` means use the default
  178. :attr:`~BaseFileLock.poll_interval`
  179. :param blocking: defaults to True. If False, function will return immediately if it cannot obtain a lock on the
  180. first attempt. Otherwise, this method will block until the timeout expires or the lock is acquired.
  181. :param cancel_check: a callable returning ``True`` when the acquisition should be canceled. Checked on each poll
  182. iteration. When triggered, raises :class:`~Timeout` just like an expired timeout.
  183. :returns: a context object that will unlock the file when the context is exited
  184. :raises Timeout: if fails to acquire lock within the timeout period
  185. .. code-block:: python
  186. # You can use this method in the context manager (recommended)
  187. with lock.acquire():
  188. pass
  189. # Or use an equivalent try-finally construct:
  190. lock.acquire()
  191. try:
  192. pass
  193. finally:
  194. lock.release()
  195. """
  196. # Use the default timeout, if no timeout is provided.
  197. if timeout is None:
  198. timeout = self._context.timeout
  199. if blocking is None:
  200. blocking = self._context.blocking
  201. if poll_interval is None:
  202. poll_interval = self._context.poll_interval
  203. # Increment the number right at the beginning. We can still undo it, if something fails.
  204. self._context.lock_counter += 1
  205. lock_id = id(self)
  206. lock_filename = self.lock_file
  207. start_time = time.perf_counter()
  208. try:
  209. while True:
  210. if not self.is_locked:
  211. self._try_break_expired_lock()
  212. _LOGGER.debug("Attempting to acquire lock %s on %s", lock_id, lock_filename)
  213. await self._run_internal_method(self._acquire)
  214. if self.is_locked:
  215. _LOGGER.debug("Lock %s acquired on %s", lock_id, lock_filename)
  216. break
  217. if self._check_give_up(
  218. lock_id,
  219. lock_filename,
  220. blocking=blocking,
  221. cancel_check=cancel_check,
  222. timeout=timeout,
  223. start_time=start_time,
  224. ):
  225. raise Timeout(lock_filename) # noqa: TRY301
  226. msg = "Lock %s not acquired on %s, waiting %s seconds ..."
  227. _LOGGER.debug(msg, lock_id, lock_filename, poll_interval)
  228. await asyncio.sleep(poll_interval)
  229. except BaseException: # Something did go wrong, so decrement the counter.
  230. self._context.lock_counter = max(0, self._context.lock_counter - 1)
  231. raise
  232. return AsyncAcquireReturnProxy(lock=self)
  233. async def release(self, force: bool = False) -> None: # ty: ignore[invalid-method-override] # noqa: FBT001, FBT002
  234. """
  235. Release the file lock. The lock is only completely released when the lock counter reaches 0. The lock file
  236. itself is not automatically deleted.
  237. :param force: If true, the lock counter is ignored and the lock is released in every case.
  238. """
  239. if self.is_locked:
  240. self._context.lock_counter -= 1
  241. if self._context.lock_counter == 0 or force:
  242. lock_id, lock_filename = id(self), self.lock_file
  243. _LOGGER.debug("Attempting to release lock %s on %s", lock_id, lock_filename)
  244. await self._run_internal_method(self._release)
  245. self._context.lock_counter = 0
  246. _LOGGER.debug("Lock %s released on %s", lock_id, lock_filename)
  247. async def _run_internal_method(self, method: Callable[[], Any]) -> None:
  248. if iscoroutinefunction(method):
  249. await method()
  250. elif self.run_in_executor:
  251. loop = self.loop or asyncio.get_running_loop()
  252. await loop.run_in_executor(self.executor, method)
  253. else:
  254. method()
  255. def __enter__(self) -> NoReturn:
  256. """
  257. Replace old __enter__ method to avoid using it.
  258. NOTE: DO NOT USE `with` FOR ASYNCIO LOCKS, USE `async with` INSTEAD.
  259. :returns: none
  260. :rtype: NoReturn
  261. """
  262. msg = "Do not use `with` for asyncio locks, use `async with` instead."
  263. raise NotImplementedError(msg)
  264. async def __aenter__(self) -> Self:
  265. """
  266. Acquire the lock.
  267. :returns: the lock object
  268. """
  269. await self.acquire()
  270. return self
  271. async def __aexit__(
  272. self,
  273. exc_type: type[BaseException] | None,
  274. exc_value: BaseException | None,
  275. traceback: TracebackType | None,
  276. ) -> None:
  277. """
  278. Release the lock.
  279. :param exc_type: the exception type if raised
  280. :param exc_value: the exception value if raised
  281. :param traceback: the exception traceback if raised
  282. """
  283. await self.release()
  284. def __del__(self) -> None:
  285. """Called when the lock object is deleted."""
  286. with contextlib.suppress(RuntimeError):
  287. loop = self.loop or asyncio.get_running_loop()
  288. if not loop.is_running(): # pragma: no cover
  289. loop.run_until_complete(self.release(force=True))
  290. else:
  291. loop.create_task(self.release(force=True))
  292. class AsyncSoftFileLock(SoftFileLock, BaseAsyncFileLock):
  293. """Simply watches the existence of the lock file."""
  294. class AsyncUnixFileLock(UnixFileLock, BaseAsyncFileLock):
  295. """Uses the :func:`fcntl.flock` to hard lock the lock file on unix systems."""
  296. class AsyncWindowsFileLock(WindowsFileLock, BaseAsyncFileLock):
  297. """Uses the :func:`msvcrt.locking` to hard lock the lock file on windows systems."""
  298. __all__ = [
  299. "AsyncAcquireReturnProxy",
  300. "AsyncSoftFileLock",
  301. "AsyncUnixFileLock",
  302. "AsyncWindowsFileLock",
  303. "BaseAsyncFileLock",
  304. ]