asyncutils.rwlocks

Readers-writer locks with different fairness policies, applicable in different situations.

Classes

AgingRWLock
CoercedMethod	Interpret any callable as a regular function in a class body so that access on instance returns something like a bound method.
FairPriorityRWLock	At the same priority level, writers are preferred over readers.
FairRWLock	Readers-writer lock that maintains FIFO order for both readers and writers indiscriminately.
PriorityRWLock
RWLock	Common base class for all readers-writer locks.
ReadPreferredRWLock	Readers-writer lock preferring readers, which risks writer starvation.
WritePreferredPriorityRWLock	Writers at any priority level are preferred over readers at any priority level.
WritePreferredRWLock	Readers-writer lock preferring writers, which risks reader starvation.

Module Contents

class asyncutils.rwlocks.AgingRWLock

Bases: PriorityRWLock

A readers-writer lock with a priority policy multiplying the current number of attempts to acquire the reading or writing lock,

counted per corresponding task, by the respective factor given at construction, to obtain the priority.

• rf, the read priority factor, defaults to context.AGING_RWLOCK_DEFAULT_READ_PRIORITY_FACTOR.

• wf, the write priority factor, defaults to context.AGING_RWLOCK_DEFAULT_WRITE_PRIORITY_FACTOR.

reading(priority: int = ...) → asyncutils._internal.types.RWLockCM

Return an async context manager for reading access. It is recommended to implement this by decorating an async generator function with @contextlib.asynccontextmanager.

setup() → None

Set up the internal state of the lock.

writing(priority: int = ...) → asyncutils._internal.types.RWLockCM

Return an async context manager for writing access. It is recommended to implement this by decorating an async generator function with @contextlib.asynccontextmanager.

property cur_unsuccessful_reads: int

The current number of unsuccessful attempts to acquire the reading lock, counted across all tasks.

property cur_unsuccessful_writes: int

The current number of unsuccessful attempts to acquire the writing lock, counted across all tasks.

class asyncutils.rwlocks.CoercedMethod[T, R, **P](func: collections.abc.Callable[Concatenate[R, P], T], /)

Interpret any callable as a regular function in a class body so that access on instance returns something like a bound method.

__get__(instance: R, owner: type[R] | None = ..., /) → collections.abc.Callable[P, T]

Return the ‘bound method’. The descriptor itself is hidden and cannot be accessed on the class it is defined in, only instances of.

__getattr__(name: str, /) → Any

__set_name__(owner: type[R], name: str, /) → None

class asyncutils.rwlocks.FairPriorityRWLock

Bases: PriorityRWLock

At the same priority level, writers are preferred over readers.

Whether instantiating this class gives a PriorityRWLock unfair to readers by default depends on context.RWLOCK_DEFAULT_PREFER_WRITERS.

class asyncutils.rwlocks.FairRWLock

Bases: RWLock

Readers-writer lock that maintains FIFO order for both readers and writers indiscriminately.

Return a WritePreferredRWLock if prefer_writers is True, otherwise a ReadPreferredRWLock. context.RWLOCK_DEFAULT_PREFER_WRITERS becomes the value of prefer_writers when it is not passed.

reading() → asyncutils._internal.types.RWLockCM

Return an async context manager for reading access. It is recommended to implement this by decorating an async generator function with @contextlib.asynccontextmanager.

setup() → None

Set up the internal state of the lock.

writing() → asyncutils._internal.types.RWLockCM

Return an async context manager for writing access. It is recommended to implement this by decorating an async generator function with @contextlib.asynccontextmanager.

class asyncutils.rwlocks.PriorityRWLock

Bases: RWLock

Common base class of FairPriorityRWLock and WritePreferredPriorityRWLock.

Lower priority levels are preferred, and the default priority is 0, as in other patterns in this module related to priority.

Whether instantiating this class gives a PriorityRWLock unfair to readers by default depends on context.RWLOCK_DEFAULT_PREFER_WRITERS.

reading(priority: int = ...) → asyncutils._internal.types.RWLockCM

Return an async context manager for reading access. It is recommended to implement this by decorating an async generator function with @contextlib.asynccontextmanager.

setup() → None

Set up the internal state of the lock.

writing(priority: int = ...) → asyncutils._internal.types.RWLockCM

Return an async context manager for writing access. It is recommended to implement this by decorating an async generator function with @contextlib.asynccontextmanager.

class asyncutils.rwlocks.RWLock

Bases: abc.ABC

Common base class for all readers-writer locks.

Return a WritePreferredRWLock if prefer_writers is True, otherwise a ReadPreferredRWLock. context.RWLOCK_DEFAULT_PREFER_WRITERS becomes the value of prefer_writers when it is not passed.

classmethod lock[T, **P](f: collections.abc.Callable[P, collections.abc.Awaitable[T]], /) → asyncutils._internal.types.RWLockRV[T, P]

Create a lock and register a reader.

locked() → bool

Whether the lock is currently held by a writer.

reader[T, **P](f: collections.abc.Callable[P, collections.abc.Awaitable[T]], /) → asyncutils._internal.types.RWLockRV[T, P]

A decorator wrapping a function returning an awaitable in an async reading context. reader() and writer() methods are attached to the returned async callable.

abstractmethod reading() → asyncutils._internal.types.RWLockCM

Return an async context manager for reading access. It is recommended to implement this by decorating an async generator function with @contextlib.asynccontextmanager.

abstractmethod setup() → None

Set up the internal state of the lock.

writer[T, **P](f: collections.abc.Callable[P, collections.abc.Awaitable[T]], /) → asyncutils._internal.types.RWLockRV[T, P]

A decorator wrapping a function returning an awaitable in an async writing context. reader() and writer() methods are attached to the returned async callable.

abstractmethod writing() → asyncutils._internal.types.RWLockCM

Return an async context manager for writing access. It is recommended to implement this by decorating an async generator function with @contextlib.asynccontextmanager.

class asyncutils.rwlocks.ReadPreferredRWLock

Bases: RWLock

Readers-writer lock preferring readers, which risks writer starvation.

Return a WritePreferredRWLock if prefer_writers is True, otherwise a ReadPreferredRWLock. context.RWLOCK_DEFAULT_PREFER_WRITERS becomes the value of prefer_writers when it is not passed.

locked() → bool

Whether the lock is currently held by a writer.

reading() → asyncutils._internal.types.RWLockCM

Return an async context manager for reading access. It is recommended to implement this by decorating an async generator function with @contextlib.asynccontextmanager.

setup() → None

Set up the internal state of the lock.

writing() → asyncutils._internal.types.RWLockCM

Return an async context manager for writing access. It is recommended to implement this by decorating an async generator function with @contextlib.asynccontextmanager.

class asyncutils.rwlocks.WritePreferredPriorityRWLock

Bases: PriorityRWLock

Writers at any priority level are preferred over readers at any priority level.

Whether instantiating this class gives a PriorityRWLock unfair to readers by default depends on context.RWLOCK_DEFAULT_PREFER_WRITERS.

class asyncutils.rwlocks.WritePreferredRWLock

Bases: RWLock

Readers-writer lock preferring writers, which risks reader starvation.

Return a WritePreferredRWLock if prefer_writers is True, otherwise a ReadPreferredRWLock. context.RWLOCK_DEFAULT_PREFER_WRITERS becomes the value of prefer_writers when it is not passed.

reading() → asyncutils._internal.types.RWLockCM

Return an async context manager for reading access. It is recommended to implement this by decorating an async generator function with @contextlib.asynccontextmanager.

setup() → None

Set up the internal state of the lock.

writing() → asyncutils._internal.types.RWLockCM

Return an async context manager for writing access. It is recommended to implement this by decorating an async generator function with @contextlib.asynccontextmanager.