max-prefetch argument doesn't work as expected

[bashirmindee]

Thank you for this amazing project.

The max-prefetch argument doesn't work as expected. If there are tasks to be consumed, the prefetch queue will be bigger than the argument max-prefetch by 1.

Look at this part of the code in the receiver.prefetcher here; and consider the max-prefetch to be 0.

 while True:
 if finish_event.is_set():
 break
 try:
 await self.sem_prefetch.acquire()
 if (
 self.max_tasks_to_execute
 and fetched_tasks >= self.max_tasks_to_execute
 ):
 logger.info("Max number of tasks executed.")
 break
 # Here we wait for the message to be fetched,
 # but we make it with timeout so it can be interrupted
 done, _ = await asyncio.wait({current_message}, timeout=0.3)
 # If the message is not fetched, we release the semaphore
 # and continue the loop. So it will check if finished event was set.
 if not done:
 self.sem_prefetch.release()
 continue
 # We're done, so now we need to check
 # whether task has returned an error.
 message = current_message.result()
 ## << consider the case where we already consumed a message from the previous iteration
 ## << and so message is an Ackable instance
 current_message = asyncio.create_task(iterator.__anext__()) 
 ## << when the interpreter arrives here, the task to consume another message is created but not awaited
 ## << so we don't have control over when it will start consuming via the broker.
 ## << usually the broker will have enough time to consume a request and it will yield its value after the next await (line done, _ = await asyncio.wait({current_message}, timeout=0.3)
 fetched_tasks += 1
 await queue.put(message)
 except (asyncio.CancelledError, StopAsyncIteration):
 break

I think we shouldn't create a task an wait for it separately like we currently do.
we should something like:

 while True:
 if finish_event.is_set():
 break
 try:
 await self.sem_prefetch.acquire()
 if (
 self.max_tasks_to_execute
 and fetched_tasks >= self.max_tasks_to_execute
 ):
 logger.info("Max number of tasks executed.")
 break
 # Here we wait for the message to be fetched,
 # but we make it with timeout so it can be interrupted
 ## << create current task and consume it right away
 current_message = asyncio.create_task(iterator.__anext__()) # type: ignore
 done, _ = await asyncio.wait({current_message}, timeout=0.3)
 # If the message is not fetched, we release the semaphore
 # and continue the loop. So it will check if finished event was set.
 if not done:
 self.sem_prefetch.release()
 continue
 # We're done, so now we need to check
 # whether task has returned an error.
 message = current_message.result()
 
 fetched_tasks += 1
 await queue.put(message)
 except (asyncio.CancelledError, StopAsyncIteration):
 break

I can do a PR and do a proper fix if you agree

[s3rius]

That's an interesting idea. I guess it can simplify the code.

Can you prepare the PR? I'll test it out locally against different scenarios.