Provide an async get() function for a theading based queue

user_2401309 0
Viewed 32

I have a simple queue that uses a threading.Condition to deal with consumers that wait at an empty queue.

I want this queue also to be consumed by async code.

That's why I added a get_async method.

The problem is now in the consumer below. It uses asyncio.wait_for since I wanted to see whether my queue is safe to be cancelled. Now the observation is, that after I induce the cancellation, my consumer always runs into a timeout.

I guess that this is caused by the CancellationError that prevents the self.condition from being released properly.

Can someone give me a hint what's going wrong here?

import threading
import time
import asyncio

SENTINEL = object()


class Queue:
    def __init__(self) -> None:
        self.items = []
        self.condition = threading.Condition()

    def put(self, item):
        with self.condition:
            self.items.append(item)
            self.condition.notify()

    def get(self):
        with self.condition:
            while not self.items:
                self.condition.wait()
            return self.items.pop(0)

    async def get_async(self):
        try:
            loop = asyncio.get_running_loop()
            return await loop.run_in_executor(None, lambda: self.get())
        except asyncio.CancelledError:
            print("I've been cancelled and I think the self.condition is now broken")


async def consumer(n: int, queue: Queue):
    while True:
        try:
            # if I leave out the "wait_for" or make it longer than my sleep(2) below,
            # it works
            item = await asyncio.wait_for(queue.get_async(), 5)
            if item is SENTINEL:
                print(f"consumer {n=} got SENTINEL")
                queue.put(item)
                break
            else:
                print(f"consumer {n=} got {item=}")
        except asyncio.TimeoutError:
            print("got timeout")


def producer(queue: Queue):
    queue.put("foo")
    queue.put("bar")
    time.sleep(2)
    queue.put("baz")
    queue.put(SENTINEL)


q = Queue()
t = threading.Thread(target=producer, args=(q,))
t.start()

asyncio.run(consumer(0, q))

t.join()

With a "long enough" timeout (or without wait_for at all) the output is

consumer n=0 got item='foo'
consumer n=0 got item='bar'
consumer n=0 got item='baz'
consumer n=0 got SENTINEL

When I induce the cancellation, it's

consumer n=0 got item='foo'
consumer n=0 got item='bar'
got timeout
got timeout
got timeout
got timeout
got timeout
(and so on)

It should be like:

consumer n=0 got item='foo'
consumer n=0 got item='bar'
got timeout
got timeout
got timeout
got timeout
consumer n=0 got item='baz'
consumer n=0 got SENTINEL
python-asyncio python-multithreading python
1 Answers

Cancelling an "async thread" only cancels waiting for the thread. It does not cancel what the thread is doing.

As the threads run self.get(), they will consume an item as soon as it is available. Cancelling the "async thread" part merely means that the async code proceeds and discards the item provided by self.get().
As a result, cancelling one thread (1 < timeout < 2) discards the 'baz' item and cancelling two threads (timeout < 1) also discards the SENTINEL – trapping the consumer in an infinite loop.

Instead of synchronising and consuming an item between async/thread code, only synchronise async/thread code and handle item consumption explicitly. This allows to safely ignore the first part but cancel the latter.

A simple albeit inefficient approach is to a) notify_all when an item becomes available and b) await condition.wait() via threads.

class Queue:
    def __init__(self) -> None:
        self.items = []
        self.condition = threading.Condition()

    def put(self, item):
        with self.condition:
            self.items.append(item)
            # wake up *everyone* who is waiting – they check for items anyway
            self.condition.notify_all()

    async def get_async(self):
        def wait():
            with self.condition:
                self.condition.wait()
        try:
            loop = asyncio.get_running_loop()
            while True:
                while not self.items:
                    await loop.run_in_executor(None, lambda: wait())
                # another thread may get the item inbetween
                # be defensive and retry on failure
                try:
                    return self.items.pop(0)
                except IndexError:
                    continue
        except asyncio.CancelledError:
            print("I've been cancelled while waiting for an item")

    ...
user_53499160
Related