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
4from collections import deque
5from collections.abc import Callable
6from dataclasses import dataclass
7from types import TracebackType
8from typing import TypeVar
10from ..lowlevel import checkpoint_if_cancelled
11from ._eventloop import get_async_backend
12from ._exceptions import BusyResourceError, NoEventLoopError
13from ._tasks import CancelScope
14from ._testing import TaskInfo, get_current_task
16T = TypeVar("T")
19@dataclass(frozen=True)
20class EventStatistics:
21 """
22 :ivar int tasks_waiting: number of tasks waiting on :meth:`~.Event.wait`
23 """
25 tasks_waiting: int
28@dataclass(frozen=True)
29class CapacityLimiterStatistics:
30 """
31 :ivar int borrowed_tokens: number of tokens currently borrowed by tasks
32 :ivar float total_tokens: total number of available tokens
33 :ivar tuple borrowers: tasks or other objects currently holding tokens borrowed from
34 this limiter
35 :ivar int tasks_waiting: number of tasks waiting on
36 :meth:`~.CapacityLimiter.acquire` or
37 :meth:`~.CapacityLimiter.acquire_on_behalf_of`
38 """
40 borrowed_tokens: int
41 total_tokens: float
42 borrowers: tuple[object, ...]
43 tasks_waiting: int
46@dataclass(frozen=True)
47class LockStatistics:
48 """
49 :ivar bool locked: flag indicating if this lock is locked or not
50 :ivar ~anyio.TaskInfo owner: task currently holding the lock (or ``None`` if the
51 lock is not held by any task)
52 :ivar int tasks_waiting: number of tasks waiting on :meth:`~.Lock.acquire`
53 """
55 locked: bool
56 owner: TaskInfo | None
57 tasks_waiting: int
60@dataclass(frozen=True)
61class ConditionStatistics:
62 """
63 :ivar int tasks_waiting: number of tasks blocked on :meth:`~.Condition.wait`
64 :ivar ~anyio.LockStatistics lock_statistics: statistics of the underlying
65 :class:`~.Lock`
66 """
68 tasks_waiting: int
69 lock_statistics: LockStatistics
72@dataclass(frozen=True)
73class SemaphoreStatistics:
74 """
75 :ivar int tasks_waiting: number of tasks waiting on :meth:`~.Semaphore.acquire`
77 """
79 tasks_waiting: int
82class Event:
83 __slots__ = ("__weakref__",)
85 def __new__(cls) -> Event:
86 try:
87 return get_async_backend().create_event()
88 except NoEventLoopError:
89 return EventAdapter()
91 def set(self) -> None:
92 """Set the flag, notifying all listeners."""
93 raise NotImplementedError
95 def is_set(self) -> bool:
96 """Return ``True`` if the flag is set, ``False`` if not."""
97 raise NotImplementedError
99 async def wait(self) -> None:
100 """
101 Wait until the flag has been set.
103 If the flag has already been set when this method is called, it returns
104 immediately.
106 """
107 raise NotImplementedError
109 def statistics(self) -> EventStatistics:
110 """Return statistics about the current state of this event."""
111 raise NotImplementedError
114class EventAdapter(Event):
115 __slots__ = "_internal_event", "_is_set"
117 def __new__(cls) -> EventAdapter:
118 return object.__new__(cls)
120 def __init__(self) -> None:
121 self._internal_event: Event | None = None
122 self._is_set = False
124 @property
125 def _event(self) -> Event:
126 if self._internal_event is None:
127 self._internal_event = get_async_backend().create_event()
128 if self._is_set:
129 self._internal_event.set()
131 return self._internal_event
133 def set(self) -> None:
134 if self._internal_event is None:
135 self._is_set = True
136 else:
137 self._event.set()
139 def is_set(self) -> bool:
140 if self._internal_event is None:
141 return self._is_set
143 return self._internal_event.is_set()
145 async def wait(self) -> None:
146 await self._event.wait()
148 def statistics(self) -> EventStatistics:
149 if self._internal_event is None:
150 return EventStatistics(tasks_waiting=0)
152 return self._internal_event.statistics()
155class Lock:
156 __slots__ = ("__weakref__",)
158 def __new__(cls, *, fast_acquire: bool = False) -> Lock:
159 try:
160 return get_async_backend().create_lock(fast_acquire=fast_acquire)
161 except NoEventLoopError:
162 return LockAdapter(fast_acquire=fast_acquire)
164 async def __aenter__(self) -> None:
165 await self.acquire()
167 async def __aexit__(
168 self,
169 exc_type: type[BaseException] | None,
170 exc_val: BaseException | None,
171 exc_tb: TracebackType | None,
172 ) -> None:
173 self.release()
175 async def acquire(self) -> None:
176 """Acquire the lock."""
177 raise NotImplementedError
179 def acquire_nowait(self) -> None:
180 """
181 Acquire the lock, without blocking.
183 :raises ~anyio.WouldBlock: if the operation would block
185 """
186 raise NotImplementedError
188 def release(self) -> None:
189 """Release the lock."""
190 raise NotImplementedError
192 def locked(self) -> bool:
193 """Return True if the lock is currently held."""
194 raise NotImplementedError
196 def statistics(self) -> LockStatistics:
197 """
198 Return statistics about the current state of this lock.
200 .. versionadded:: 3.0
201 """
202 raise NotImplementedError
205class LockAdapter(Lock):
206 __slots__ = "_internal_lock", "_fast_acquire"
208 def __new__(cls, *, fast_acquire: bool = False) -> LockAdapter:
209 return object.__new__(cls)
211 def __init__(self, *, fast_acquire: bool = False):
212 self._internal_lock: Lock | None = None
213 self._fast_acquire = fast_acquire
215 @property
216 def _lock(self) -> Lock:
217 if self._internal_lock is None:
218 self._internal_lock = get_async_backend().create_lock(
219 fast_acquire=self._fast_acquire
220 )
222 return self._internal_lock
224 async def __aenter__(self) -> None:
225 await self._lock.acquire()
227 async def __aexit__(
228 self,
229 exc_type: type[BaseException] | None,
230 exc_val: BaseException | None,
231 exc_tb: TracebackType | None,
232 ) -> None:
233 if self._internal_lock is not None:
234 self._internal_lock.release()
236 async def acquire(self) -> None:
237 """Acquire the lock."""
238 await self._lock.acquire()
240 def acquire_nowait(self) -> None:
241 """
242 Acquire the lock, without blocking.
244 :raises ~anyio.WouldBlock: if the operation would block
246 """
247 self._lock.acquire_nowait()
249 def release(self) -> None:
250 """Release the lock."""
251 self._lock.release()
253 def locked(self) -> bool:
254 """Return True if the lock is currently held."""
255 return self._lock.locked()
257 def statistics(self) -> LockStatistics:
258 """
259 Return statistics about the current state of this lock.
261 .. versionadded:: 3.0
263 """
264 if self._internal_lock is None:
265 return LockStatistics(False, None, 0)
267 return self._internal_lock.statistics()
270class Condition:
271 __slots__ = "__weakref__", "_owner_task", "_lock", "_waiters"
273 def __init__(self, lock: Lock | None = None):
274 self._owner_task: TaskInfo | None = None
275 self._lock = lock or Lock()
276 self._waiters: deque[Event] = deque()
278 async def __aenter__(self) -> None:
279 await self.acquire()
281 async def __aexit__(
282 self,
283 exc_type: type[BaseException] | None,
284 exc_val: BaseException | None,
285 exc_tb: TracebackType | None,
286 ) -> None:
287 self.release()
289 def _check_acquired(self) -> None:
290 if self._owner_task != get_current_task():
291 raise RuntimeError("The current task is not holding the underlying lock")
293 async def acquire(self) -> None:
294 """Acquire the underlying lock."""
295 await self._lock.acquire()
296 self._owner_task = get_current_task()
298 def acquire_nowait(self) -> None:
299 """
300 Acquire the underlying lock, without blocking.
302 :raises ~anyio.WouldBlock: if the operation would block
304 """
305 self._lock.acquire_nowait()
306 self._owner_task = get_current_task()
308 def release(self) -> None:
309 """Release the underlying lock."""
310 self._lock.release()
312 def locked(self) -> bool:
313 """Return True if the lock is set."""
314 return self._lock.locked()
316 def notify(self, n: int = 1) -> None:
317 """Notify exactly n listeners."""
318 self._check_acquired()
319 for _ in range(n):
320 try:
321 event = self._waiters.popleft()
322 except IndexError:
323 break
325 event.set()
327 def notify_all(self) -> None:
328 """Notify all the listeners."""
329 self._check_acquired()
330 for event in self._waiters:
331 event.set()
333 self._waiters.clear()
335 async def wait(self) -> None:
336 """Wait for a notification."""
337 await checkpoint_if_cancelled()
338 self._check_acquired()
339 event = Event()
340 self._waiters.append(event)
341 self.release()
342 try:
343 await event.wait()
344 except BaseException:
345 if not event.is_set():
346 self._waiters.remove(event)
347 elif self._waiters:
348 # This task was notified by could not act on it, so pass
349 # it on to the next task
350 self._waiters.popleft().set()
352 raise
353 finally:
354 with CancelScope(shield=True):
355 await self.acquire()
357 async def wait_for(self, predicate: Callable[[], T]) -> T:
358 """
359 Wait until a predicate becomes true.
361 :param predicate: a callable that returns a truthy value when the condition is
362 met
363 :return: the result of the predicate
365 .. versionadded:: 4.11.0
367 """
368 while not (result := predicate()):
369 await self.wait()
371 return result
373 def statistics(self) -> ConditionStatistics:
374 """
375 Return statistics about the current state of this condition.
377 .. versionadded:: 3.0
378 """
379 return ConditionStatistics(len(self._waiters), self._lock.statistics())
382class Semaphore:
383 __slots__ = "__weakref__", "_fast_acquire"
385 def __new__(
386 cls,
387 initial_value: int,
388 *,
389 max_value: int | None = None,
390 fast_acquire: bool = False,
391 ) -> Semaphore:
392 try:
393 return get_async_backend().create_semaphore(
394 initial_value, max_value=max_value, fast_acquire=fast_acquire
395 )
396 except NoEventLoopError:
397 return SemaphoreAdapter(initial_value, max_value=max_value)
399 def __init__(
400 self,
401 initial_value: int,
402 *,
403 max_value: int | None = None,
404 fast_acquire: bool = False,
405 ):
406 if not isinstance(initial_value, int):
407 raise TypeError("initial_value must be an integer")
408 if initial_value < 0:
409 raise ValueError("initial_value must be >= 0")
410 if max_value is not None:
411 if not isinstance(max_value, int):
412 raise TypeError("max_value must be an integer or None")
413 if max_value < initial_value:
414 raise ValueError(
415 "max_value must be equal to or higher than initial_value"
416 )
418 self._fast_acquire = fast_acquire
420 async def __aenter__(self) -> Semaphore:
421 await self.acquire()
422 return self
424 async def __aexit__(
425 self,
426 exc_type: type[BaseException] | None,
427 exc_val: BaseException | None,
428 exc_tb: TracebackType | None,
429 ) -> None:
430 self.release()
432 async def acquire(self) -> None:
433 """Decrement the semaphore value, blocking if necessary."""
434 raise NotImplementedError
436 def acquire_nowait(self) -> None:
437 """
438 Acquire the underlying lock, without blocking.
440 :raises ~anyio.WouldBlock: if the operation would block
442 """
443 raise NotImplementedError
445 def release(self) -> None:
446 """Increment the semaphore value."""
447 raise NotImplementedError
449 @property
450 def value(self) -> int:
451 """The current value of the semaphore."""
452 raise NotImplementedError
454 @property
455 def max_value(self) -> int | None:
456 """The maximum value of the semaphore."""
457 raise NotImplementedError
459 def statistics(self) -> SemaphoreStatistics:
460 """
461 Return statistics about the current state of this semaphore.
463 .. versionadded:: 3.0
464 """
465 raise NotImplementedError
468class SemaphoreAdapter(Semaphore):
469 __slots__ = "_internal_semaphore", "_initial_value", "_max_value"
471 def __new__(
472 cls,
473 initial_value: int,
474 *,
475 max_value: int | None = None,
476 fast_acquire: bool = False,
477 ) -> SemaphoreAdapter:
478 return object.__new__(cls)
480 def __init__(
481 self,
482 initial_value: int,
483 *,
484 max_value: int | None = None,
485 fast_acquire: bool = False,
486 ) -> None:
487 super().__init__(initial_value, max_value=max_value, fast_acquire=fast_acquire)
488 self._internal_semaphore: Semaphore | None = None
489 self._initial_value = initial_value
490 self._max_value = max_value
492 @property
493 def _semaphore(self) -> Semaphore:
494 if self._internal_semaphore is None:
495 self._internal_semaphore = get_async_backend().create_semaphore(
496 self._initial_value, max_value=self._max_value
497 )
499 return self._internal_semaphore
501 async def acquire(self) -> None:
502 await self._semaphore.acquire()
504 def acquire_nowait(self) -> None:
505 self._semaphore.acquire_nowait()
507 def release(self) -> None:
508 self._semaphore.release()
510 @property
511 def value(self) -> int:
512 if self._internal_semaphore is None:
513 return self._initial_value
515 return self._semaphore.value
517 @property
518 def max_value(self) -> int | None:
519 return self._max_value
521 def statistics(self) -> SemaphoreStatistics:
522 if self._internal_semaphore is None:
523 return SemaphoreStatistics(tasks_waiting=0)
525 return self._semaphore.statistics()
528class CapacityLimiter:
529 __slots__ = ("__weakref__",)
531 def __new__(cls, total_tokens: float) -> CapacityLimiter:
532 try:
533 return get_async_backend().create_capacity_limiter(total_tokens)
534 except NoEventLoopError:
535 return CapacityLimiterAdapter(total_tokens)
537 async def __aenter__(self) -> None:
538 raise NotImplementedError
540 async def __aexit__(
541 self,
542 exc_type: type[BaseException] | None,
543 exc_val: BaseException | None,
544 exc_tb: TracebackType | None,
545 ) -> None:
546 raise NotImplementedError
548 @property
549 def total_tokens(self) -> float:
550 """
551 The total number of tokens available for borrowing.
553 This is a read-write property. If the total number of tokens is increased, the
554 proportionate number of tasks waiting on this limiter will be granted their
555 tokens.
557 .. versionchanged:: 3.0
558 The property is now writable.
559 .. versionchanged:: 4.12
560 The value can now be set to 0.
562 """
563 raise NotImplementedError
565 @total_tokens.setter
566 def total_tokens(self, value: float) -> None:
567 raise NotImplementedError
569 @property
570 def borrowed_tokens(self) -> int:
571 """The number of tokens that have currently been borrowed."""
572 raise NotImplementedError
574 @property
575 def available_tokens(self) -> float:
576 """The number of tokens currently available to be borrowed"""
577 raise NotImplementedError
579 def acquire_nowait(self) -> None:
580 """
581 Acquire a token for the current task without waiting for one to become
582 available.
584 :raises ~anyio.WouldBlock: if there are no tokens available for borrowing
586 """
587 raise NotImplementedError
589 def acquire_on_behalf_of_nowait(self, borrower: object) -> None:
590 """
591 Acquire a token without waiting for one to become available.
593 :param borrower: the entity borrowing a token
594 :raises ~anyio.WouldBlock: if there are no tokens available for borrowing
596 """
597 raise NotImplementedError
599 async def acquire(self) -> None:
600 """
601 Acquire a token for the current task, waiting if necessary for one to become
602 available.
604 """
605 raise NotImplementedError
607 async def acquire_on_behalf_of(self, borrower: object) -> None:
608 """
609 Acquire a token, waiting if necessary for one to become available.
611 :param borrower: the entity borrowing a token
613 """
614 raise NotImplementedError
616 def release(self) -> None:
617 """
618 Release the token held by the current task.
620 :raises RuntimeError: if the current task has not borrowed a token from this
621 limiter.
623 """
624 raise NotImplementedError
626 def release_on_behalf_of(self, borrower: object) -> None:
627 """
628 Release the token held by the given borrower.
630 :raises RuntimeError: if the borrower has not borrowed a token from this
631 limiter.
633 """
634 raise NotImplementedError
636 def statistics(self) -> CapacityLimiterStatistics:
637 """
638 Return statistics about the current state of this limiter.
640 .. versionadded:: 3.0
642 """
643 raise NotImplementedError
646class CapacityLimiterAdapter(CapacityLimiter):
647 __slots__ = "_internal_limiter", "_total_tokens"
649 def __new__(cls, total_tokens: float) -> CapacityLimiterAdapter:
650 return object.__new__(cls)
652 def __init__(self, total_tokens: float) -> None:
653 self._internal_limiter: CapacityLimiter | None = None
654 self.total_tokens = total_tokens
656 @property
657 def _limiter(self) -> CapacityLimiter:
658 if self._internal_limiter is None:
659 self._internal_limiter = get_async_backend().create_capacity_limiter(
660 self._total_tokens
661 )
663 return self._internal_limiter
665 async def __aenter__(self) -> None:
666 await self._limiter.__aenter__()
668 async def __aexit__(
669 self,
670 exc_type: type[BaseException] | None,
671 exc_val: BaseException | None,
672 exc_tb: TracebackType | None,
673 ) -> None:
674 return await self._limiter.__aexit__(exc_type, exc_val, exc_tb)
676 @property
677 def total_tokens(self) -> float:
678 if self._internal_limiter is None:
679 return self._total_tokens
681 return self._internal_limiter.total_tokens
683 @total_tokens.setter
684 def total_tokens(self, value: float) -> None:
685 if not isinstance(value, int) and not math.isinf(value):
686 raise TypeError("total_tokens must be an int or math.inf")
687 elif value < 0:
688 raise ValueError("total_tokens must be >= 0")
690 if self._internal_limiter is None:
691 self._total_tokens = value
692 return
694 self._limiter.total_tokens = value
696 @property
697 def borrowed_tokens(self) -> int:
698 if self._internal_limiter is None:
699 return 0
701 return self._internal_limiter.borrowed_tokens
703 @property
704 def available_tokens(self) -> float:
705 if self._internal_limiter is None:
706 return self._total_tokens
708 return self._internal_limiter.available_tokens
710 def acquire_nowait(self) -> None:
711 self._limiter.acquire_nowait()
713 def acquire_on_behalf_of_nowait(self, borrower: object) -> None:
714 self._limiter.acquire_on_behalf_of_nowait(borrower)
716 async def acquire(self) -> None:
717 await self._limiter.acquire()
719 async def acquire_on_behalf_of(self, borrower: object) -> None:
720 await self._limiter.acquire_on_behalf_of(borrower)
722 def release(self) -> None:
723 self._limiter.release()
725 def release_on_behalf_of(self, borrower: object) -> None:
726 self._limiter.release_on_behalf_of(borrower)
728 def statistics(self) -> CapacityLimiterStatistics:
729 if self._internal_limiter is None:
730 return CapacityLimiterStatistics(
731 borrowed_tokens=0,
732 total_tokens=self.total_tokens,
733 borrowers=(),
734 tasks_waiting=0,
735 )
737 return self._internal_limiter.statistics()
740class ResourceGuard:
741 """
742 A context manager for ensuring that a resource is only used by a single task at a
743 time.
745 Entering this context manager while the previous has not exited it yet will trigger
746 :exc:`BusyResourceError`.
748 :param action: the action to guard against (visible in the :exc:`BusyResourceError`
749 when triggered, e.g. "Another task is already {action} this resource")
751 .. versionadded:: 4.1
752 """
754 __slots__ = "__weakref__", "action", "_guarded"
756 def __init__(self, action: str = "using"):
757 self.action: str = action
758 self._guarded = False
760 def __enter__(self) -> None:
761 if self._guarded:
762 raise BusyResourceError(self.action)
764 self._guarded = True
766 def __exit__(
767 self,
768 exc_type: type[BaseException] | None,
769 exc_val: BaseException | None,
770 exc_tb: TracebackType | None,
771 ) -> None:
772 self._guarded = False