Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import math
4import sys
5from collections import deque
6from collections.abc import Callable
7from dataclasses import dataclass
8from types import TracebackType
9from typing import TypeVar
11from ..lowlevel import checkpoint_if_cancelled
12from ._eventloop import get_async_backend
13from ._exceptions import BusyResourceError, NoEventLoopError
14from ._tasks import CancelScope
15from ._testing import TaskInfo, get_current_task
17if sys.version_info >= (3, 11):
18 from typing import Self
19else:
20 from typing_extensions import Self
22T = TypeVar("T")
25@dataclass(frozen=True)
26class EventStatistics:
27 """
28 :ivar int tasks_waiting: number of tasks waiting on :meth:`~.Event.wait`
29 """
31 tasks_waiting: int
34@dataclass(frozen=True)
35class CapacityLimiterStatistics:
36 """
37 :ivar int borrowed_tokens: number of tokens currently borrowed by tasks
38 :ivar float total_tokens: total number of available tokens
39 :ivar tuple borrowers: tasks or other objects currently holding tokens borrowed from
40 this limiter
41 :ivar int tasks_waiting: number of tasks waiting on
42 :meth:`~.CapacityLimiter.acquire` or
43 :meth:`~.CapacityLimiter.acquire_on_behalf_of`
44 """
46 borrowed_tokens: int
47 total_tokens: float
48 borrowers: tuple[object, ...]
49 tasks_waiting: int
52@dataclass(frozen=True)
53class LockStatistics:
54 """
55 :ivar bool locked: flag indicating if this lock is locked or not
56 :ivar ~anyio.TaskInfo owner: task currently holding the lock (or ``None`` if the
57 lock is not held by any task)
58 :ivar int tasks_waiting: number of tasks waiting on :meth:`~.Lock.acquire`
59 """
61 locked: bool
62 owner: TaskInfo | None
63 tasks_waiting: int
66@dataclass(frozen=True)
67class ConditionStatistics:
68 """
69 :ivar int tasks_waiting: number of tasks blocked on :meth:`~.Condition.wait`
70 :ivar ~anyio.LockStatistics lock_statistics: statistics of the underlying
71 :class:`~.Lock`
72 """
74 tasks_waiting: int
75 lock_statistics: LockStatistics
78@dataclass(frozen=True)
79class SemaphoreStatistics:
80 """
81 :ivar int tasks_waiting: number of tasks waiting on :meth:`~.Semaphore.acquire`
83 """
85 tasks_waiting: int
88class Event:
89 __slots__ = ("__weakref__",)
91 def __new__(cls) -> Event:
92 try:
93 return get_async_backend().create_event()
94 except NoEventLoopError:
95 return EventAdapter()
97 def set(self) -> None:
98 """Set the flag, notifying all listeners."""
99 raise NotImplementedError
101 def is_set(self) -> bool:
102 """Return ``True`` if the flag is set, ``False`` if not."""
103 raise NotImplementedError
105 async def wait(self) -> None:
106 """
107 Wait until the flag has been set.
109 If the flag has already been set when this method is called, it returns
110 immediately.
112 """
113 raise NotImplementedError
115 def statistics(self) -> EventStatistics:
116 """Return statistics about the current state of this event."""
117 raise NotImplementedError
120class EventAdapter(Event):
121 __slots__ = "_internal_event", "_is_set"
123 def __new__(cls) -> Self:
124 return object.__new__(cls)
126 def __init__(self) -> None:
127 self._internal_event: Event | None = None
128 self._is_set = False
130 @property
131 def _event(self) -> Event:
132 if self._internal_event is None:
133 self._internal_event = get_async_backend().create_event()
134 if self._is_set:
135 self._internal_event.set()
137 return self._internal_event
139 def set(self) -> None:
140 if self._internal_event is None:
141 self._is_set = True
142 else:
143 self._event.set()
145 def is_set(self) -> bool:
146 if self._internal_event is None:
147 return self._is_set
149 return self._internal_event.is_set()
151 async def wait(self) -> None:
152 await self._event.wait()
154 def statistics(self) -> EventStatistics:
155 if self._internal_event is None:
156 return EventStatistics(tasks_waiting=0)
158 return self._internal_event.statistics()
161class Lock:
162 __slots__ = ("__weakref__",)
164 def __new__(cls, *, fast_acquire: bool = False) -> Lock:
165 try:
166 return get_async_backend().create_lock(fast_acquire=fast_acquire)
167 except NoEventLoopError:
168 return LockAdapter(fast_acquire=fast_acquire)
170 async def __aenter__(self) -> None:
171 await self.acquire()
173 async def __aexit__(
174 self,
175 exc_type: type[BaseException] | None,
176 exc_val: BaseException | None,
177 exc_tb: TracebackType | None,
178 ) -> None:
179 self.release()
181 async def acquire(self) -> None:
182 """Acquire the lock."""
183 raise NotImplementedError
185 def acquire_nowait(self) -> None:
186 """
187 Acquire the lock, without blocking.
189 :raises ~anyio.WouldBlock: if the operation would block
191 """
192 raise NotImplementedError
194 def release(self) -> None:
195 """Release the lock."""
196 raise NotImplementedError
198 def locked(self) -> bool:
199 """Return True if the lock is currently held."""
200 raise NotImplementedError
202 def statistics(self) -> LockStatistics:
203 """
204 Return statistics about the current state of this lock.
206 .. versionadded:: 3.0
207 """
208 raise NotImplementedError
211class LockAdapter(Lock):
212 __slots__ = "_fast_acquire", "_internal_lock"
214 def __new__(cls, *, fast_acquire: bool = False) -> Self:
215 return object.__new__(cls)
217 def __init__(self, *, fast_acquire: bool = False):
218 self._internal_lock: Lock | None = None
219 self._fast_acquire = fast_acquire
221 @property
222 def _lock(self) -> Lock:
223 if self._internal_lock is None:
224 self._internal_lock = get_async_backend().create_lock(
225 fast_acquire=self._fast_acquire
226 )
228 return self._internal_lock
230 async def __aenter__(self) -> None:
231 await self._lock.acquire()
233 async def __aexit__(
234 self,
235 exc_type: type[BaseException] | None,
236 exc_val: BaseException | None,
237 exc_tb: TracebackType | None,
238 ) -> None:
239 if self._internal_lock is not None:
240 self._internal_lock.release()
242 async def acquire(self) -> None:
243 """Acquire the lock."""
244 await self._lock.acquire()
246 def acquire_nowait(self) -> None:
247 """
248 Acquire the lock, without blocking.
250 :raises ~anyio.WouldBlock: if the operation would block
252 """
253 self._lock.acquire_nowait()
255 def release(self) -> None:
256 """Release the lock."""
257 self._lock.release()
259 def locked(self) -> bool:
260 """Return True if the lock is currently held."""
261 if self._internal_lock is None:
262 return False
264 return self._internal_lock.locked()
266 def statistics(self) -> LockStatistics:
267 """
268 Return statistics about the current state of this lock.
270 .. versionadded:: 3.0
272 """
273 if self._internal_lock is None:
274 return LockStatistics(False, None, 0)
276 return self._internal_lock.statistics()
279class Condition:
280 __slots__ = "__weakref__", "_lock", "_waiters"
282 def __init__(self, lock: Lock | None = None):
283 self._lock = lock or Lock()
284 self._waiters: deque[Event] = deque()
286 async def __aenter__(self) -> None:
287 await self.acquire()
289 async def __aexit__(
290 self,
291 exc_type: type[BaseException] | None,
292 exc_val: BaseException | None,
293 exc_tb: TracebackType | None,
294 ) -> None:
295 self.release()
297 def _check_acquired(self) -> None:
298 if self._lock.statistics().owner != get_current_task():
299 raise RuntimeError("The current task is not holding the underlying lock")
301 async def acquire(self) -> None:
302 """Acquire the underlying lock."""
303 await self._lock.acquire()
305 def acquire_nowait(self) -> None:
306 """
307 Acquire the underlying lock, without blocking.
309 :raises ~anyio.WouldBlock: if the operation would block
311 """
312 self._lock.acquire_nowait()
314 def release(self) -> None:
315 """Release the underlying lock."""
316 self._lock.release()
318 def locked(self) -> bool:
319 """Return True if the lock is set."""
320 return self._lock.locked()
322 def notify(self, n: int = 1) -> None:
323 """Notify exactly n listeners."""
324 self._check_acquired()
325 for _ in range(n):
326 try:
327 event = self._waiters.popleft()
328 except IndexError:
329 break
331 event.set()
333 def notify_all(self) -> None:
334 """Notify all the listeners."""
335 self._check_acquired()
336 for event in self._waiters:
337 event.set()
339 self._waiters.clear()
341 async def wait(self) -> None:
342 """Wait for a notification."""
343 await checkpoint_if_cancelled()
344 self._check_acquired()
345 event = Event()
346 self._waiters.append(event)
347 self.release()
348 try:
349 await event.wait()
350 except BaseException:
351 if not event.is_set():
352 self._waiters.remove(event)
353 elif self._waiters:
354 # This task was notified by could not act on it, so pass
355 # it on to the next task
356 self._waiters.popleft().set()
358 raise
359 finally:
360 with CancelScope(shield=True):
361 await self.acquire()
363 async def wait_for(self, predicate: Callable[[], T]) -> T:
364 """
365 Wait until a predicate becomes true.
367 :param predicate: a callable that returns a truthy value when the condition is
368 met
369 :return: the result of the predicate
371 .. versionadded:: 4.11.0
373 """
374 while not (result := predicate()):
375 await self.wait()
377 return result
379 def statistics(self) -> ConditionStatistics:
380 """
381 Return statistics about the current state of this condition.
383 .. versionadded:: 3.0
384 """
385 return ConditionStatistics(len(self._waiters), self._lock.statistics())
388class Semaphore:
389 __slots__ = "__weakref__", "_fast_acquire"
391 def __new__(
392 cls,
393 initial_value: int,
394 *,
395 max_value: int | None = None,
396 fast_acquire: bool = False,
397 ) -> Semaphore:
398 try:
399 return get_async_backend().create_semaphore(
400 initial_value, max_value=max_value, fast_acquire=fast_acquire
401 )
402 except NoEventLoopError:
403 return SemaphoreAdapter(initial_value, max_value=max_value)
405 def __init__(
406 self,
407 initial_value: int,
408 *,
409 max_value: int | None = None,
410 fast_acquire: bool = False,
411 ):
412 if not isinstance(initial_value, int):
413 raise TypeError("initial_value must be an integer")
414 if initial_value < 0:
415 raise ValueError("initial_value must be >= 0")
416 if max_value is not None:
417 if not isinstance(max_value, int):
418 raise TypeError("max_value must be an integer or None")
419 if max_value < initial_value:
420 raise ValueError(
421 "max_value must be equal to or higher than initial_value"
422 )
424 self._fast_acquire = fast_acquire
426 async def __aenter__(self) -> Self:
427 await self.acquire()
428 return self
430 async def __aexit__(
431 self,
432 exc_type: type[BaseException] | None,
433 exc_val: BaseException | None,
434 exc_tb: TracebackType | None,
435 ) -> None:
436 self.release()
438 async def acquire(self) -> None:
439 """Decrement the semaphore value, blocking if necessary."""
440 raise NotImplementedError
442 def acquire_nowait(self) -> None:
443 """
444 Acquire the underlying lock, without blocking.
446 :raises ~anyio.WouldBlock: if the operation would block
448 """
449 raise NotImplementedError
451 def release(self) -> None:
452 """Increment the semaphore value."""
453 raise NotImplementedError
455 @property
456 def value(self) -> int:
457 """The current value of the semaphore."""
458 raise NotImplementedError
460 @property
461 def max_value(self) -> int | None:
462 """The maximum value of the semaphore."""
463 raise NotImplementedError
465 def statistics(self) -> SemaphoreStatistics:
466 """
467 Return statistics about the current state of this semaphore.
469 .. versionadded:: 3.0
470 """
471 raise NotImplementedError
474class SemaphoreAdapter(Semaphore):
475 __slots__ = "_initial_value", "_internal_semaphore", "_max_value"
477 def __new__(
478 cls,
479 initial_value: int,
480 *,
481 max_value: int | None = None,
482 fast_acquire: bool = False,
483 ) -> Self:
484 return object.__new__(cls)
486 def __init__(
487 self,
488 initial_value: int,
489 *,
490 max_value: int | None = None,
491 fast_acquire: bool = False,
492 ) -> None:
493 super().__init__(initial_value, max_value=max_value, fast_acquire=fast_acquire)
494 self._internal_semaphore: Semaphore | None = None
495 self._initial_value = initial_value
496 self._max_value = max_value
498 @property
499 def _semaphore(self) -> Semaphore:
500 if self._internal_semaphore is None:
501 self._internal_semaphore = get_async_backend().create_semaphore(
502 self._initial_value,
503 max_value=self._max_value,
504 fast_acquire=self._fast_acquire,
505 )
507 return self._internal_semaphore
509 async def acquire(self) -> None:
510 await self._semaphore.acquire()
512 def acquire_nowait(self) -> None:
513 self._semaphore.acquire_nowait()
515 def release(self) -> None:
516 self._semaphore.release()
518 @property
519 def value(self) -> int:
520 if self._internal_semaphore is None:
521 return self._initial_value
523 return self._semaphore.value
525 @property
526 def max_value(self) -> int | None:
527 return self._max_value
529 def statistics(self) -> SemaphoreStatistics:
530 if self._internal_semaphore is None:
531 return SemaphoreStatistics(tasks_waiting=0)
533 return self._semaphore.statistics()
536class CapacityLimiter:
537 __slots__ = ("__weakref__",)
539 def __new__(cls, total_tokens: float) -> CapacityLimiter:
540 try:
541 return get_async_backend().create_capacity_limiter(total_tokens)
542 except NoEventLoopError:
543 return CapacityLimiterAdapter(total_tokens)
545 async def __aenter__(self) -> None:
546 raise NotImplementedError
548 async def __aexit__(
549 self,
550 exc_type: type[BaseException] | None,
551 exc_val: BaseException | None,
552 exc_tb: TracebackType | None,
553 ) -> None:
554 raise NotImplementedError
556 @property
557 def total_tokens(self) -> float:
558 """
559 The total number of tokens available for borrowing.
561 This is a read-write property. If the total number of tokens is increased, the
562 proportionate number of tasks waiting on this limiter will be granted their
563 tokens.
565 .. versionchanged:: 3.0
566 The property is now writable.
567 .. versionchanged:: 4.12
568 The value can now be set to 0.
570 """
571 raise NotImplementedError
573 @total_tokens.setter
574 def total_tokens(self, value: float) -> None:
575 raise NotImplementedError
577 @property
578 def borrowed_tokens(self) -> int:
579 """The number of tokens that have currently been borrowed."""
580 raise NotImplementedError
582 @property
583 def available_tokens(self) -> float:
584 """The number of tokens currently available to be borrowed"""
585 raise NotImplementedError
587 def acquire_nowait(self) -> None:
588 """
589 Acquire a token for the current task without waiting for one to become
590 available.
592 :raises ~anyio.WouldBlock: if there are no tokens available for borrowing
594 """
595 raise NotImplementedError
597 def acquire_on_behalf_of_nowait(self, borrower: object) -> None:
598 """
599 Acquire a token without waiting for one to become available.
601 :param borrower: the entity borrowing a token
602 :raises ~anyio.WouldBlock: if there are no tokens available for borrowing
604 """
605 raise NotImplementedError
607 async def acquire(self) -> None:
608 """
609 Acquire a token for the current task, waiting if necessary for one to become
610 available.
612 """
613 raise NotImplementedError
615 async def acquire_on_behalf_of(self, borrower: object) -> None:
616 """
617 Acquire a token, waiting if necessary for one to become available.
619 :param borrower: the entity borrowing a token
621 """
622 raise NotImplementedError
624 def release(self) -> None:
625 """
626 Release the token held by the current task.
628 :raises RuntimeError: if the current task has not borrowed a token from this
629 limiter.
631 """
632 raise NotImplementedError
634 def release_on_behalf_of(self, borrower: object) -> None:
635 """
636 Release the token held by the given borrower.
638 :raises RuntimeError: if the borrower has not borrowed a token from this
639 limiter.
641 """
642 raise NotImplementedError
644 def statistics(self) -> CapacityLimiterStatistics:
645 """
646 Return statistics about the current state of this limiter.
648 .. versionadded:: 3.0
650 """
651 raise NotImplementedError
654class CapacityLimiterAdapter(CapacityLimiter):
655 __slots__ = "_internal_limiter", "_total_tokens"
657 def __new__(cls, total_tokens: float) -> Self:
658 return object.__new__(cls)
660 def __init__(self, total_tokens: float) -> None:
661 self._internal_limiter: CapacityLimiter | None = None
662 self.total_tokens = total_tokens
664 @property
665 def _limiter(self) -> CapacityLimiter:
666 if self._internal_limiter is None:
667 self._internal_limiter = get_async_backend().create_capacity_limiter(
668 self._total_tokens
669 )
671 return self._internal_limiter
673 async def __aenter__(self) -> None:
674 await self._limiter.__aenter__()
676 async def __aexit__(
677 self,
678 exc_type: type[BaseException] | None,
679 exc_val: BaseException | None,
680 exc_tb: TracebackType | None,
681 ) -> None:
682 return await self._limiter.__aexit__(exc_type, exc_val, exc_tb)
684 @property
685 def total_tokens(self) -> float:
686 if self._internal_limiter is None:
687 return self._total_tokens
689 return self._internal_limiter.total_tokens
691 @total_tokens.setter
692 def total_tokens(self, value: float) -> None:
693 if not isinstance(value, int) and not math.isinf(value):
694 raise TypeError("total_tokens must be an int or math.inf")
695 elif value < 0:
696 raise ValueError("total_tokens must be >= 0")
698 if self._internal_limiter is None:
699 self._total_tokens = value
700 return
702 self._limiter.total_tokens = value
704 @property
705 def borrowed_tokens(self) -> int:
706 if self._internal_limiter is None:
707 return 0
709 return self._internal_limiter.borrowed_tokens
711 @property
712 def available_tokens(self) -> float:
713 if self._internal_limiter is None:
714 return self._total_tokens
716 return self._internal_limiter.available_tokens
718 def acquire_nowait(self) -> None:
719 self._limiter.acquire_nowait()
721 def acquire_on_behalf_of_nowait(self, borrower: object) -> None:
722 self._limiter.acquire_on_behalf_of_nowait(borrower)
724 async def acquire(self) -> None:
725 await self._limiter.acquire()
727 async def acquire_on_behalf_of(self, borrower: object) -> None:
728 await self._limiter.acquire_on_behalf_of(borrower)
730 def release(self) -> None:
731 self._limiter.release()
733 def release_on_behalf_of(self, borrower: object) -> None:
734 self._limiter.release_on_behalf_of(borrower)
736 def statistics(self) -> CapacityLimiterStatistics:
737 if self._internal_limiter is None:
738 return CapacityLimiterStatistics(
739 borrowed_tokens=0,
740 total_tokens=self.total_tokens,
741 borrowers=(),
742 tasks_waiting=0,
743 )
745 return self._internal_limiter.statistics()
748class ResourceGuard:
749 """
750 A context manager for ensuring that a resource is only used by a single task at a
751 time.
753 Entering this context manager while the previous has not exited it yet will trigger
754 :exc:`BusyResourceError`.
756 :param action: the action to guard against (visible in the :exc:`BusyResourceError`
757 when triggered, e.g. "Another task is already {action} this resource")
759 .. versionadded:: 4.1
760 """
762 __slots__ = "__weakref__", "_guarded", "action"
764 def __init__(self, action: str = "using"):
765 self.action: str = action
766 self._guarded = False
768 def __enter__(self) -> None:
769 if self._guarded:
770 raise BusyResourceError(self.action)
772 self._guarded = True
774 def __exit__(
775 self,
776 exc_type: type[BaseException] | None,
777 exc_val: BaseException | None,
778 exc_tb: TracebackType | None,
779 ) -> None:
780 self._guarded = False