Queues & Locks

import asyncio

from tornado.queues import Queue

async def demo_basic():
    shelf_jobs = Queue(maxsize=2)           # at most 2 waiting items
    done = []

    async def worker(name):
        async for title in shelf_jobs:      # never ends by itself
            try:
                await asyncio.sleep(0.01)   # pretend to do I/O
                done.append((name, title))
            finally:
                shelf_jobs.task_done()      # always, even on error

    workers = [asyncio.create_task(worker(f"w{i}")) for i in range(3)]
    for title in ["Dune", "Emma", "Ulysses", "Kim", "Ivanhoe"]:
        await shelf_jobs.put(title)         # waits whenever 2 are queued
    await shelf_jobs.join(timeout=None)     # every put has a task_done
    for w in workers:                       # workers loop forever: stop them
        w.cancel()
    await asyncio.gather(*workers, return_exceptions=True)
    return sorted(t for _, t in done)

assert asyncio.run(demo_basic()) == ["Dune", "Emma", "Ivanhoe", "Kim", "Ulysses"]
print("processed all five titles")

import asyncio

from tornado.queues import LifoQueue, PriorityQueue

async def demo_ordering():
    pq = PriorityQueue()
    pq.put_nowait((10, "bulk re-index"))
    pq.put_nowait((0, "reader refresh"))
    pq.put_nowait((1, "new book"))
    by_priority = [(await pq.get())[1] for _ in range(3)]

    lq = LifoQueue()
    for n in (3, 2, 1):
        lq.put_nowait(n)
    newest_first = [await lq.get() for _ in range(3)]
    return by_priority, newest_first

by_priority, newest_first = asyncio.run(demo_ordering())
assert by_priority == ["reader refresh", "new book", "bulk re-index"]
assert newest_first == [1, 2, 3]
print(by_priority, newest_first)

import asyncio
import datetime

from tornado.queues import Queue, QueueEmpty, QueueFull
from tornado.util import TimeoutError as TornadoTimeoutError

async def demo_timeouts():
    q = Queue(maxsize=1)
    q.put_nowait("first")
    try:
        q.put_nowait("second")              # full: fails immediately
    except QueueFull:
        outcome_full = "QueueFull"

    try:
        await q.put("second", timeout=datetime.timedelta(milliseconds=50))
    except TornadoTimeoutError:
        outcome_put = "put timed out"

    assert q.get_nowait() == "first"
    try:
        q.get_nowait()
    except QueueEmpty:
        outcome_empty = "QueueEmpty"

    try:
        await q.get(timeout=datetime.timedelta(milliseconds=50))
    except TornadoTimeoutError:
        outcome_get = "get timed out"
    return outcome_full, outcome_put, outcome_empty, outcome_get

print(asyncio.run(demo_timeouts()))
assert TornadoTimeoutError is TimeoutError

import json
import unittest

from tornado.queues import Queue, QueueFull
from tornado.testing import AsyncHTTPTestCase
from tornado.web import Application, RequestHandler

class AddBookHandler(RequestHandler):
    def post(self):
        jobs = self.application.settings["jobs"]
        try:
            jobs.put_nowait(self.get_argument("isbn"))
        except QueueFull:
            self.set_status(503)
            self.set_header("Retry-After", "5")
            self.write({"error": "busy"})
            return
        self.set_status(202)
        self.write({"queued": jobs.qsize()})

class AddBookTest(AsyncHTTPTestCase):
    def get_app(self):
        self.jobs = Queue(maxsize=2)
        return Application([(r"/books", AddBookHandler)], jobs=self.jobs)

    def test_accepts_then_pushes_back(self):
        for n in (1, 2):
            r = self.fetch("/books", method="POST", body=f"isbn=97800{n}")
            self.assertEqual(r.code, 202)
        r = self.fetch("/books", method="POST", body="isbn=978003")
        self.assertEqual(r.code, 503)
        self.assertEqual(r.headers["Retry-After"], "5")
        self.assertEqual(self.jobs.qsize(), 2)

result = unittest.TextTestRunner(verbosity=0).run(
    unittest.defaultTestLoader.loadTestsFromTestCase(AddBookTest))
assert result.wasSuccessful()

import asyncio

from tornado.locks import Lock

async def demo_lock():
    counter = {"unlocked": 0, "locked": 0}
    lock = Lock()

    async def bump_unlocked():
        value = counter["unlocked"]
        await asyncio.sleep(0)              # another coroutine runs here
        counter["unlocked"] = value + 1

    async def bump_locked():
        async with lock:
            value = counter["locked"]
            await asyncio.sleep(0)
            counter["locked"] = value + 1

    await asyncio.gather(*[bump_unlocked() for _ in range(10)])
    await asyncio.gather(*[bump_locked() for _ in range(10)])
    return counter

print(counter := asyncio.run(demo_lock()))
assert counter["unlocked"] < 10 and counter["locked"] == 10

import asyncio
import datetime

from tornado.locks import Event
from tornado.util import TimeoutError as TornadoTimeoutError

async def demo_event():
    cache_ready = Event()

    async def warm_cache():
        await asyncio.sleep(0.02)
        cache_ready.set()                   # wakes every waiter at once

    async def handler(n):
        await cache_ready.wait(timeout=datetime.timedelta(seconds=1))
        return n

    asyncio.create_task(warm_cache())
    served = await asyncio.gather(*[handler(n) for n in range(3)])

    never = Event()
    try:
        await never.wait(timeout=datetime.timedelta(milliseconds=30))
        timed_out = False
    except TornadoTimeoutError:
        timed_out = True
    return served, cache_ready.is_set(), timed_out

served, is_set, timed_out = asyncio.run(demo_event())
assert served == [0, 1, 2] and is_set and timed_out
print(served, is_set, timed_out)

import asyncio
import datetime

from tornado.locks import Condition

async def demo_condition():
    new_reviews = Condition()
    reviews = []

    async def long_poll(seen):
        while len(reviews) <= seen:
            notified = await new_reviews.wait(
                timeout=datetime.timedelta(milliseconds=500))
            if not notified:
                return "timeout"            # wait() returns False, no exception
        return reviews[seen]

    pollers = [asyncio.create_task(long_poll(0)) for _ in range(3)]
    await asyncio.sleep(0.02)
    reviews.append("Great book")
    new_reviews.notify_all()
    got = await asyncio.gather(*pollers)

    quiet = await long_poll(1)              # no review arrives
    return got, quiet

print(asyncio.run(demo_condition()))

import asyncio
import datetime

from tornado.locks import BoundedSemaphore, Lock, Semaphore
from tornado.util import TimeoutError as TornadoTimeoutError

async def demo_semaphore():
    sem = Semaphore(2)
    running = peak = 0

    async def job():
        nonlocal running, peak
        async with sem:
            running += 1
            peak = max(peak, running)
            await asyncio.sleep(0.01)
            running -= 1

    await asyncio.gather(*[job() for _ in range(8)])

    leaky = Semaphore(1)
    leaky.release()                         # accepted: limit is now 2
    bounded = BoundedSemaphore(1)
    try:
        bounded.release()                   # more releases than acquires
    except ValueError as exc:
        bounded_error = str(exc)

    lock = Lock()
    try:
        lock.release()                      # not locked
    except RuntimeError:
        lock_error = "RuntimeError"

    await lock.acquire()
    try:
        await lock.acquire(timeout=datetime.timedelta(milliseconds=30))
    except TornadoTimeoutError:
        reacquire = "timed out (locks are not re-entrant)"
    try:
        with lock:                          # wrong: must be `async with`
            pass
    except RuntimeError:
        sync_with = "RuntimeError"
    return peak, bounded_error, lock_error, reacquire, sync_with

result = asyncio.run(demo_semaphore())
assert result[0] == 2
print(result)