[FIXED] create dependency between concurrent tasks in Python asyncio

Issue

I have two tasks in a consumer/producer relationship, separated by a asyncio.Queue. If the producer task fails, I'd like the consumer task to also fail as soon as possible, and not wait indefinitely on the queue. The consumer task can be created(spawned) independently from the producer task.

In general terms, I'd like to implement a dependency between two tasks, such that the failure of one is also the failure of the other, while keeping those two tasks concurrent(i.e. one will not await the other directly).

What kind of solutions(e.g. patterns) could be used here?

Basically, I'm thinking of erlang's "links".

I think it may be possible to implement something similar using callbacks, i.e. asyncio.Task.add_done_callback

Thanks!

Solution

From the comment:

The behavior I'm trying to avoid is the consumer being oblivious to the producer's death and waiting indefinitely on the queue. I want the consumer to be notified of the producer's death, and have a chance to react. or just fail, and that even while it's also waiting on the queue.

Other than the answer presented by Yigal, another way is to set up a third task that monitors the two and cancels one when the other one finishes. This can be generalized to any two tasks:

async def cancel_when_done(source, target):
    assert isinstance(source, asyncio.Task)
    assert isinstance(target, asyncio.Task)
    try:
        await source
    except:
        # SOURCE is a task which we expect to be awaited by someone else
        pass
    target.cancel()

Now when setting up the producer and the consumer, you can link them with the above function. For example:

async def producer(q):
    for i in itertools.count():
        await q.put(i)
        await asyncio.sleep(.2)
        if i == 7:
            1/0

async def consumer(q):
    while True:
        val = await q.get()
        print('got', val)

async def main():
    loop = asyncio.get_event_loop()
    queue = asyncio.Queue()
    p = loop.create_task(producer(queue))
    c = loop.create_task(consumer(queue))
    loop.create_task(cancel_when_done(p, c))
    await asyncio.gather(p, c)

asyncio.get_event_loop().run_until_complete(main())

Answered By - user4815162342