[Python] Gracefully terminate `asyncio` program in Python with a full queue

The Problem


I have a simplified example of an asynchronous program (Python 3.9) that is not working when exceptions are raised in futures, and I am looking for ways to gracefully terminate it.

In particular, when the number of failed requests exceeds the sum of the queue size and the number of workers, I end up in a situation where the queue is full (see COMMENT1 in the code snippet below), and I can no longer submit new requests, nor fail gracefully. I am looking for advice how to avoid this situation by raising the fact that an exception occurred and shutting the entire program down. In this specific case, I do not need to reraise the specific errors of worker as those are logged. Thank you in advance!

import asyncio

HEALTHY_REQUESTS = 10
MAX_WORKERS = 3
QUEUE_SIZE = 19
# REQUESTS >= HEALTHY_REQUESTS + (QUEUE_SIZE + MAX_WORKERS) + 1 ---> full queue, program stalls because queue is full
REQUESTS = HEALTHY_REQUESTS + (QUEUE_SIZE + MAX_WORKERS) + 2


async def task_producer(requests):

print(f'Starting task producer with: {requests}')
request_queue = asyncio.Queue(maxsize=QUEUE_SIZE) # starting tasks

print('Initializing futures of the queue')
request_workers = [asyncio.ensure_future(request_consumer(request_queue)) for _ in range(MAX_WORKERS)]

# Submit requests
for req in requests:
print(f'Putting request {req} into queue, {request_queue.qsize()}')
# COMMENT1: we get stuck here if
# REQUESTS = HEALTHY_REQUESTS + (QUEUE_SIZE + MAX_WORKERS) + 2 or more.
# We get stuck later if there is one less,
# but I think we can ignore this fortunate case where we could still do something
await request_queue.put(req)

# Wait for all requests to come back
await request_queue.put(None)
await asyncio.wait(request_workers)

print('Getting results from the parser')
return 0


async def request_consumer(request_queue):
print('request consumer started')
while True:
request_info = await request_queue.get()
await asyncio.sleep(0.5)
if request_info is None:
await request_queue.put(None)
break
elif request_info > HEALTHY_REQUESTS:
raise RuntimeError(f'Breaking thing in make requests with request={request_info}')
print(f'{request_info} - request consumer is finished with request')


if __name__ == '__main__':
requests = list(range(REQUESTS))
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
producer = loop.create_task(task_producer(requests))
producer_result = loop.run_until_complete(producer)
loop.close()
print(f'Output of producer: {producer_result}')


Solution Attempts

Attempt 1


Tried adding a check for exceptions raised in the worker futures, which in turn I put in the for loop and before putting a request (see COMMENT1). See the code snippet of the worker status check is presented below.

This works in the contrived example above in a sense, that it helps terminating the loop, but the program still gets stuck when putting None afterwards. In the actual program, this does not work the same. Somehow the queue gets filled before this check has a chance to return False.

def workers_are_healthy(request_workers):
for worker in request_workers:
try:
exception = worker.exception()
except asyncio.InvalidStateError:
continue
if isinstance(exception, Exception):
return False
return True

Attempt 2


Found a something that works, but I really do not like it... If I create an instance variable in the class where this code is packaged in, (effectively a global variable in the example above,) I can use it in a separate task to stop the program. In particular, I create another task, which then is passed into loop.run_until_complete. Inside of this task, I check every few seconds what the value of the global variable is. If an error were to occur inside of a specific worker (request_workers), I change its value, and hope that this works. Feels very wrong...

Continue reading...