does not work --max-prefetch=1

[Randommist]

I have the following task. I need to have one queue and several workers that take a task from this queue. The conditions are as follows: when a worker takes a task, he cannot do anything else except this task, including taking other tasks. When other tasks arrive, I assume that other workers will pick them up.

code example

class MyMiddleware(TaskiqMiddleware):
 def startup(self) -> None:
 print("RUN STARTUP")

 def shutdown(self) -> None:
 print("RUN SHUTDOWN")

 def pre_execute(self, message: "TaskiqMessage") -> TaskiqMessage:
 print("BEFORE EXECUTE")
 return message

 def post_execute(self, message: "TaskiqMessage", result: "TaskiqResult[Any]") -> None:
 print("POST EXECUTE")

 def post_save(self, message: "TaskiqMessage", result: "TaskiqResult[Any]") -> None:
 print("POST SAVE")


redis_async_result = RedisAsyncResultBackend(
redis_url="redis://localhost:6379"
)

broker = ListQueueBroker(url="redis://localhost:6379", ).with_result_backend(redis_async_result).with_middlewares(MyMiddleware())

@broker.task(task_name="add_one")
async def add_one(value: int) -> float:
print("START WORK")
await asyncio.sleep(15)
res = value +1
print(f"Result: {res}")
return res

before starting, I pre-loaded 3 tasks into the queue
I start 2 workers at the same time, but you can see how the first worker took all the tasks and started executing them

This also doesn't work in rabbitmq

P.S. In general, I have a task of the following nature. There is a certain resource and I need to parse it. For this, I have, say, 3 accounts. The problem is that I cannot simultaneously do two parsing tasks using the same account. The tasks also take a long time (~3 minutes), so I plan to make many workers (in this case, 3) who will own their own account and, as far as possible, take tasks. But another approach is also possible when we have one worker owning several accounts and simultaneously performing tasks using different accounts. But here the difficulties arise in that he can take, say, 5 tasks for himself, but he can only perform 3, and it would be better if 2 tasks were in the queue so that another worker could possibly take them. In general, I would like to hear how you would solve this problem? Thank you!

[Randommist]

For me, the problem was solved by using RabbitMQ with the qos parameter. It turns out that it is responsible for the amount of prefetch.

broker = (
AioPikaBroker(qos=1)
.with_result_backend(RedisAsyncResultBackend(redis_url="redis://localhost:6379"))
)

but I still don't understand what --max-prefetch does then, since in my tests it didn't make any difference

[s3rius]

Hello! Here's an explanation of why this happens:

This issue is specific to RabbitMQ's underlying implementation and the aio-pika library. While it's not necessarily a problem, it might break some expectations.

Essentially, the prefetch-count parameter creates a queue at the worker level, limiting the number of messages coming from the broker. However, aio-pika prefetches messages before they are given to the worker. As a result, you may still see prefetched messages even if the max-prefetch parameter is set to 1.

Updating the qos settings will instruct aio-pika not to prefetch messages and to delegate the task to the taskiq worker entirely.

Hope that has helped.