tornado.locks – Synchronization primitives

New in version 4.2.

Coordinate coroutines with synchronization primitives analogous to those the standard library provides to threads. These classes are very similar to those provided in the standard library’s asyncio package.

Warning

Note that these primitives are not actually thread-safe and cannot be used in place of those from the standard library’s threading module–they are meant to coordinate Tornado coroutines in a single-threaded app, not to protect shared objects in a multithreaded app.

Condition

class tornado.locks.Condition

A condition allows one or more coroutines to wait until notified.

Like a standard threading.Condition, but does not need an underlying lock that is acquired and released.

With a Condition, coroutines can wait to be notified by other coroutines:

import asyncio
from tornado import gen
from tornado.locks import Condition

condition = Condition()

async def waiter():
    print("I'll wait right here")
    await condition.wait()
    print("I'm done waiting")

async def notifier():
    print("About to notify")
    condition.notify()
    print("Done notifying")

async def runner():
    # Wait for waiter() and notifier() in parallel
    await gen.multi([waiter(), notifier()])

asyncio.run(runner())

I'll wait right here
About to notify
Done notifying
I'm done waiting

wait takes an optional timeout argument, which is either an absolute timestamp:

io_loop = IOLoop.current()

# Wait up to 1 second for a notification.
await condition.wait(timeout=io_loop.time() + 1)

…or a datetime.timedelta for a timeout relative to the current time:

# Wait up to 1 second.
await condition.wait(timeout=datetime.timedelta(seconds=1))

The method returns False if there’s no notification before the deadline.

Changed in version 5.0: Previously, waiters could be notified synchronously from within notify. Now, the notification will always be received on the next iteration of the IOLoop.

wait(timeout: Optional[Union[float, timedelta]] = None) → Awaitable[bool]

Wait for notify.

Returns a Future that resolves True if the condition is notified, or False after a timeout.

notify(n: int = 1) → None

Wake n waiters.

notify_all() → None

Wake all waiters.

Event

class tornado.locks.Event

An event blocks coroutines until its internal flag is set to True.

Similar to threading.Event.

A coroutine can wait for an event to be set. Once it is set, calls to yield event.wait() will not block unless the event has been cleared:

import asyncio
from tornado import gen
from tornado.locks import Event

event = Event()

async def waiter():
    print("Waiting for event")
    await event.wait()
    print("Not waiting this time")
    await event.wait()
    print("Done")

async def setter():
    print("About to set the event")
    event.set()

async def runner():
    await gen.multi([waiter(), setter()])

asyncio.run(runner())

Waiting for event
About to set the event
Not waiting this time
Done

is_set() → bool

Return True if the internal flag is true.

set() → None

Set the internal flag to True. All waiters are awakened.

Calling wait once the flag is set will not block.

clear() → None

Reset the internal flag to False.

Calls to wait will block until set is called.

wait(timeout: Optional[Union[float, timedelta]] = None) → Awaitable[None]

Block until the internal flag is true.

Returns an awaitable, which raises tornado.util.TimeoutError after a timeout.

Semaphore

class tornado.locks.Semaphore(value: int = 1)

A lock that can be acquired a fixed number of times before blocking.

A Semaphore manages a counter representing the number of release calls minus the number of acquire calls, plus an initial value. The acquire method blocks if necessary until it can return without making the counter negative.

Semaphores limit access to a shared resource. To allow access for two workers at a time:

import asyncio
from tornado import gen
from tornado.locks import Semaphore

sem = Semaphore(2)

async def worker(worker_id):
    await sem.acquire()
    try:
        print("Worker %d is working" % worker_id)
        await use_some_resource()
    finally:
        print("Worker %d is done" % worker_id)
        sem.release()

async def runner():
    # Join all workers.
    await gen.multi([worker(i) for i in range(3)])

asyncio.run(runner())

Worker 0 is working
Worker 1 is working
Worker 0 is done
Worker 2 is working
Worker 1 is done
Worker 2 is done

Workers 0 and 1 are allowed to run concurrently, but worker 2 waits until the semaphore has been released once, by worker 0.

The semaphore can be used as an async context manager:

async def worker(worker_id):
    async with sem:
        print("Worker %d is working" % worker_id)
        await use_some_resource()

    # Now the semaphore has been released.
    print("Worker %d is done" % worker_id)

For compatibility with older versions of Python, acquire is a context manager, so worker could also be written as:

@gen.coroutine
def worker(worker_id):
    with (yield sem.acquire()):
        print("Worker %d is working" % worker_id)
        yield use_some_resource()

    # Now the semaphore has been released.
    print("Worker %d is done" % worker_id)

Changed in version 4.3: Added async with support in Python 3.5.

release() → None

Increment the counter and wake one waiter.

acquire(timeout: Optional[Union[float, timedelta]] = None) → Awaitable[_ReleasingContextManager]

Decrement the counter. Returns an awaitable.

Block if the counter is zero and wait for a release. The awaitable raises TimeoutError after the deadline.

BoundedSemaphore

class tornado.locks.BoundedSemaphore(value: int = 1)

A semaphore that prevents release() being called too many times.

If release would increment the semaphore’s value past the initial value, it raises ValueError. Semaphores are mostly used to guard resources with limited capacity, so a semaphore released too many times is a sign of a bug.

release() → None

Increment the counter and wake one waiter.

acquire(timeout: Optional[Union[float, timedelta]] = None) → Awaitable[_ReleasingContextManager]

Decrement the counter. Returns an awaitable.

Block if the counter is zero and wait for a release. The awaitable raises TimeoutError after the deadline.

Lock

class tornado.locks.Lock

A lock for coroutines.

A Lock begins unlocked, and acquire locks it immediately. While it is locked, a coroutine that yields acquire waits until another coroutine calls release.

Releasing an unlocked lock raises RuntimeError.

A Lock can be used as an async context manager with the async with statement:

>>> from tornado import locks
>>> lock = locks.Lock()
>>>
>>> async def f():
...    async with lock:
...        # Do something holding the lock.
...        pass
...
...    # Now the lock is released.

For compatibility with older versions of Python, the acquire method asynchronously returns a regular context manager:

>>> async def f2():
...    with (yield lock.acquire()):
...        # Do something holding the lock.
...        pass
...
...    # Now the lock is released.

Changed in version 4.3: Added async with support in Python 3.5.

acquire(timeout: Optional[Union[float, timedelta]] = None) → Awaitable[_ReleasingContextManager]

Attempt to lock. Returns an awaitable.

Returns an awaitable, which raises tornado.util.TimeoutError after a timeout.

release() → None

Unlock.

The first coroutine in line waiting for acquire gets the lock.

If not locked, raise a RuntimeError.