tornado.locks – Synchronization primitives

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:

condition = locks.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()]

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

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

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

# Wait up to 1 second.
yield condition.wait(deadline=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:

event = locks.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()]

io_loop.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:

sem = locks.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)]

io_loop.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)

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.