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 return self._lock.locked()
263 def statistics(self) -> LockStatistics:
264 """
265 Return statistics about the current state of this lock.
267 .. versionadded:: 3.0
269 """
270 if self._internal_lock is None:
271 return LockStatistics(False, None, 0)
273 return self._internal_lock.statistics()
276class Condition:
277 __slots__ = "__weakref__", "_lock", "_owner_task", "_waiters"
279 def __init__(self, lock: Lock | None = None):
280 self._owner_task: TaskInfo | None = None
281 self._lock = lock or Lock()
282 self._waiters: deque[Event] = deque()
284 async def __aenter__(self) -> None:
285 await self.acquire()
287 async def __aexit__(
288 self,
289 exc_type: type[BaseException] | None,
290 exc_val: BaseException | None,
291 exc_tb: TracebackType | None,
292 ) -> None:
293 self.release()
295 def _check_acquired(self) -> None:
296 if self._owner_task != get_current_task():
297 raise RuntimeError("The current task is not holding the underlying lock")
299 async def acquire(self) -> None:
300 """Acquire the underlying lock."""
301 await self._lock.acquire()
302 self._owner_task = get_current_task()
304 def acquire_nowait(self) -> None:
305 """
306 Acquire the underlying lock, without blocking.
308 :raises ~anyio.WouldBlock: if the operation would block
310 """
311 self._lock.acquire_nowait()
312 self._owner_task = get_current_task()
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, max_value=self._max_value
503 )
505 return self._internal_semaphore
507 async def acquire(self) -> None:
508 await self._semaphore.acquire()
510 def acquire_nowait(self) -> None:
511 self._semaphore.acquire_nowait()
513 def release(self) -> None:
514 self._semaphore.release()
516 @property
517 def value(self) -> int:
518 if self._internal_semaphore is None:
519 return self._initial_value
521 return self._semaphore.value
523 @property
524 def max_value(self) -> int | None:
525 return self._max_value
527 def statistics(self) -> SemaphoreStatistics:
528 if self._internal_semaphore is None:
529 return SemaphoreStatistics(tasks_waiting=0)
531 return self._semaphore.statistics()
534class CapacityLimiter:
535 __slots__ = ("__weakref__",)
537 def __new__(cls, total_tokens: float) -> CapacityLimiter:
538 try:
539 return get_async_backend().create_capacity_limiter(total_tokens)
540 except NoEventLoopError:
541 return CapacityLimiterAdapter(total_tokens)
543 async def __aenter__(self) -> None:
544 raise NotImplementedError
546 async def __aexit__(
547 self,
548 exc_type: type[BaseException] | None,
549 exc_val: BaseException | None,
550 exc_tb: TracebackType | None,
551 ) -> None:
552 raise NotImplementedError
554 @property
555 def total_tokens(self) -> float:
556 """
557 The total number of tokens available for borrowing.
559 This is a read-write property. If the total number of tokens is increased, the
560 proportionate number of tasks waiting on this limiter will be granted their
561 tokens.
563 .. versionchanged:: 3.0
564 The property is now writable.
565 .. versionchanged:: 4.12
566 The value can now be set to 0.
568 """
569 raise NotImplementedError
571 @total_tokens.setter
572 def total_tokens(self, value: float) -> None:
573 raise NotImplementedError
575 @property
576 def borrowed_tokens(self) -> int:
577 """The number of tokens that have currently been borrowed."""
578 raise NotImplementedError
580 @property
581 def available_tokens(self) -> float:
582 """The number of tokens currently available to be borrowed"""
583 raise NotImplementedError
585 def acquire_nowait(self) -> None:
586 """
587 Acquire a token for the current task without waiting for one to become
588 available.
590 :raises ~anyio.WouldBlock: if there are no tokens available for borrowing
592 """
593 raise NotImplementedError
595 def acquire_on_behalf_of_nowait(self, borrower: object) -> None:
596 """
597 Acquire a token without waiting for one to become available.
599 :param borrower: the entity borrowing a token
600 :raises ~anyio.WouldBlock: if there are no tokens available for borrowing
602 """
603 raise NotImplementedError
605 async def acquire(self) -> None:
606 """
607 Acquire a token for the current task, waiting if necessary for one to become
608 available.
610 """
611 raise NotImplementedError
613 async def acquire_on_behalf_of(self, borrower: object) -> None:
614 """
615 Acquire a token, waiting if necessary for one to become available.
617 :param borrower: the entity borrowing a token
619 """
620 raise NotImplementedError
622 def release(self) -> None:
623 """
624 Release the token held by the current task.
626 :raises RuntimeError: if the current task has not borrowed a token from this
627 limiter.
629 """
630 raise NotImplementedError
632 def release_on_behalf_of(self, borrower: object) -> None:
633 """
634 Release the token held by the given borrower.
636 :raises RuntimeError: if the borrower has not borrowed a token from this
637 limiter.
639 """
640 raise NotImplementedError
642 def statistics(self) -> CapacityLimiterStatistics:
643 """
644 Return statistics about the current state of this limiter.
646 .. versionadded:: 3.0
648 """
649 raise NotImplementedError
652class CapacityLimiterAdapter(CapacityLimiter):
653 __slots__ = "_internal_limiter", "_total_tokens"
655 def __new__(cls, total_tokens: float) -> Self:
656 return object.__new__(cls)
658 def __init__(self, total_tokens: float) -> None:
659 self._internal_limiter: CapacityLimiter | None = None
660 self.total_tokens = total_tokens
662 @property
663 def _limiter(self) -> CapacityLimiter:
664 if self._internal_limiter is None:
665 self._internal_limiter = get_async_backend().create_capacity_limiter(
666 self._total_tokens
667 )
669 return self._internal_limiter
671 async def __aenter__(self) -> None:
672 await self._limiter.__aenter__()
674 async def __aexit__(
675 self,
676 exc_type: type[BaseException] | None,
677 exc_val: BaseException | None,
678 exc_tb: TracebackType | None,
679 ) -> None:
680 return await self._limiter.__aexit__(exc_type, exc_val, exc_tb)
682 @property
683 def total_tokens(self) -> float:
684 if self._internal_limiter is None:
685 return self._total_tokens
687 return self._internal_limiter.total_tokens
689 @total_tokens.setter
690 def total_tokens(self, value: float) -> None:
691 if not isinstance(value, int) and not math.isinf(value):
692 raise TypeError("total_tokens must be an int or math.inf")
693 elif value < 0:
694 raise ValueError("total_tokens must be >= 0")
696 if self._internal_limiter is None:
697 self._total_tokens = value
698 return
700 self._limiter.total_tokens = value
702 @property
703 def borrowed_tokens(self) -> int:
704 if self._internal_limiter is None:
705 return 0
707 return self._internal_limiter.borrowed_tokens
709 @property
710 def available_tokens(self) -> float:
711 if self._internal_limiter is None:
712 return self._total_tokens
714 return self._internal_limiter.available_tokens
716 def acquire_nowait(self) -> None:
717 self._limiter.acquire_nowait()
719 def acquire_on_behalf_of_nowait(self, borrower: object) -> None:
720 self._limiter.acquire_on_behalf_of_nowait(borrower)
722 async def acquire(self) -> None:
723 await self._limiter.acquire()
725 async def acquire_on_behalf_of(self, borrower: object) -> None:
726 await self._limiter.acquire_on_behalf_of(borrower)
728 def release(self) -> None:
729 self._limiter.release()
731 def release_on_behalf_of(self, borrower: object) -> None:
732 self._limiter.release_on_behalf_of(borrower)
734 def statistics(self) -> CapacityLimiterStatistics:
735 if self._internal_limiter is None:
736 return CapacityLimiterStatistics(
737 borrowed_tokens=0,
738 total_tokens=self.total_tokens,
739 borrowers=(),
740 tasks_waiting=0,
741 )
743 return self._internal_limiter.statistics()
746class ResourceGuard:
747 """
748 A context manager for ensuring that a resource is only used by a single task at a
749 time.
751 Entering this context manager while the previous has not exited it yet will trigger
752 :exc:`BusyResourceError`.
754 :param action: the action to guard against (visible in the :exc:`BusyResourceError`
755 when triggered, e.g. "Another task is already {action} this resource")
757 .. versionadded:: 4.1
758 """
760 __slots__ = "__weakref__", "_guarded", "action"
762 def __init__(self, action: str = "using"):
763 self.action: str = action
764 self._guarded = False
766 def __enter__(self) -> None:
767 if self._guarded:
768 raise BusyResourceError(self.action)
770 self._guarded = True
772 def __exit__(
773 self,
774 exc_type: type[BaseException] | None,
775 exc_val: BaseException | None,
776 exc_tb: TracebackType | None,
777 ) -> None:
778 self._guarded = False