asyncutils.rwlocks

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

Classes

AgingRWLock	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.
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
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 AGING_RWLOCK_DEFAULT_READ_PRIORITY_FACTOR.

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

reading(priority: int = ...) → asyncutils._internal.prots.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.prots.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 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. RWLOCK_DEFAULT_PREFER_WRITERS becomes the value of prefer_writers when it is not passed.

reading() → asyncutils._internal.prots.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.prots.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 AgingRWLock, 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 RWLOCK_DEFAULT_PREFER_WRITERS.

reading(priority: int = ...) → asyncutils._internal.prots.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.prots.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.

If you would like to subclass this, you must implement reading(), writing() and setup().

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

is_reading() → bool

Whether the lock is currently held by a reader. Default implementation returns whether the _nr attribute is greater than 0, assuming that slot holds the number of readers.

is_writing() → bool

Whether the lock is currently held by a writer. Default implementation returns the value of the _wa attribute.

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

Create a lock and register a reader.

locked() → bool

A writer can enter at this moment only if False is returned.

reader[T, **P](f: collections.abc.Callable[P, collections.abc.Awaitable[T]], /) → asyncutils._internal.prots.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.prots.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.prots.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.prots.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. RWLOCK_DEFAULT_PREFER_WRITERS becomes the value of prefer_writers when it is not passed.

is_reading() → bool

Whether the lock is currently held by a reader. Default implementation returns whether the _nr attribute is greater than 0, assuming that slot holds the number of readers.

is_writing() → bool

Whether the lock is currently held by a writer. Default implementation returns the value of the _wa attribute.

locked() → bool

A writer can enter at this moment only if False is returned.

reading() → asyncutils._internal.prots.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.prots.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 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. RWLOCK_DEFAULT_PREFER_WRITERS becomes the value of prefer_writers when it is not passed.

reading() → asyncutils._internal.prots.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.prots.RWLockCM

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