tornado.locks – Synchronization primitives

New in version 4.2.

Coordinate coroutines with synchronization primitives analogous to those the standard library provides to threads.

(Note that these primitives are not actually thread-safe and cannot be used in place of those from the standard library–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:

from tornado import gen
from tornado.ioloop import IOLoop
from tornado.locks import Condition

condition = Condition()

@gen.coroutine
def waiter():
    print("I'll wait right here")
    yield condition.wait()  # Yield a Future.
    print("I'm done waiting")

@gen.coroutine
def notifier():
    print("About to notify")
    condition.notify()
    print("Done notifying")

@gen.coroutine
def runner():
    # Yield two Futures; wait for waiter() and notifier() to finish.
    yield [waiter(), notifier()]

IOLoop.current().run_sync(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.
yield condition.wait(timeout=io_loop.time() + 1)

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

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

The method raises tornado.gen.TimeoutError if there’s no notification before the deadline.

wait(timeout=None)

Wait for notify.

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

notify(n=1)

Wake n waiters.

notify_all()

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:

from tornado import gen
from tornado.ioloop import IOLoop
from tornado.locks import Event

event = Event()

@gen.coroutine
def waiter():
    print("Waiting for event")
    yield event.wait()
    print("Not waiting this time")
    yield event.wait()
    print("Done")

@gen.coroutine
def setter():
    print("About to set the event")
    event.set()

@gen.coroutine
def runner():
    yield [waiter(), setter()]

IOLoop.current().run_sync(runner)

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

is_set()

Return True if the internal flag is true.

set()

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

Calling wait once the flag is set will not block.

clear()

Reset the internal flag to False.

Calls to wait will block until set is called.

wait(timeout=None)

Block until the internal flag is true.

Returns a Future, which raises tornado.gen.TimeoutError after a timeout.

Semaphore

class tornado.locks.Semaphore(value=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:

from tornado import gen
from tornado.ioloop import IOLoop
from tornado.locks import Semaphore

sem = Semaphore(2)

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

@gen.coroutine
def runner():
    # Join all workers.
    yield [worker(i) for i in range(3)]

IOLoop.current().run_sync(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.

acquire is a context manager, so worker could 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)

In Python 3.5, the semaphore itself 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)

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

release()

Increment the counter and wake one waiter.

acquire(timeout=None)

Decrement the counter. Returns a Future.

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

BoundedSemaphore

class tornado.locks.BoundedSemaphore(value=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()

Increment the counter and wake one waiter.

acquire(timeout=None)

Decrement the counter. Returns a Future.

Block if the counter is zero and wait for a release. The Future 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.

acquire supports the context manager protocol in all Python versions:

>>> from tornado import gen, locks
>>> lock = locks.Lock()
>>>
>>> @gen.coroutine
... def f():
...    with (yield lock.acquire()):
...        # Do something holding the lock.
...        pass
...
...    # Now the lock is released.

In Python 3.5, Lock also supports the async context manager protocol. Note that in this case there is no acquire, because async with includes both the yield and the acquire (just as it does with threading.Lock):

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

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

acquire(timeout=None)

Attempt to lock. Returns a Future.

Returns a Future, which raises tornado.gen.TimeoutError after a timeout.

release()

Unlock.

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

If not locked, raise a RuntimeError.