Improve documentation to use `Broker` as consumer

[MuriloScarpaSitonio]

I want to use taskiq as a consumer. From this comment, I was advised to use taskiq.receiver.Receiver to accomplish this. I think I've figured how to make this work (thanks to the great code quality!), but I'm not 100% sure that I'm in the correct path. Please, take a look at the snippet bellow:

import asyncio
import json
from typing import AsyncGenerator
from uuid import uuid4

from taskiq.brokers.inmemory_broker import InMemoryBroker
from taskiq.receiver import Receiver


class ListenerInMemoryBroker(InMemoryBroker):
    async def listen(self) -> AsyncGenerator[str, None]:
        for i in range(1, 4):
            await asyncio.sleep(i)
            yield json.dumps(
                {
                    "task_id": str(uuid4()),
                    "task_name": "my_task",
                    "labels": {},
                    "args": [i, "a" * i],
                    "kwargs": {"i": i},
                }
            )


broker = ListenerInMemoryBroker()


@broker.task(task_name="my_task")
async def my_task(n: int, a: str, i: int) -> None:
    print(f"{n=}, {n.__class__=}")
    print(f"{a=}, {a.__class__=}")
    print(f"{i=}, {i.__class__=}")


async def main() -> None:
    consumer = Receiver(broker=broker, max_async_tasks=100)
    await consumer.listen()


if __name__ == "__main__":
    asyncio.run(main())

From this dummy implementation, I assumed the following:

• Messages must be produced with the schema defined at taskig.message.TaskiqMessage;
• task_name must be explicit set;

And also left me with some doubts:

• What's the usage of max_async_tasks and max_prefetch?

With that said, I think this important use case deserves better documentation. If anyone can confirm my assumptions, answer my queries, and point out where I can add a section in the docs about this, I'd be happy to do so!

[vvanglro]

I want to process a message from a given topic.

I'm not sure what that means?

AioKafkaBroker passes in the specified topic to consume messages.

from taskiq_aio_kafka import AioKafkaBroker

broker = AioKafkaBroker(bootstrap_servers="localhost", kafka_topic="test_topic")

@broker.task
async def test() -> None:
    print("The best task ever!")

Or rewrite listen after inheritance and judge the subject of the message

from taskiq_aio_kafka import AioKafkaBroker

class CustomAioKafkaBroker(AioKafkaBroker):
    async def listen(
            self,
        ) -> AsyncGenerator[bytes, None]:
            """Listen to topic.
    
            This function starts listen to topic and
            yields every new message.
    
            :yields: parsed broker message.
            :raises ValueError: if no aiokafka_consumer or startup wasn't called.
            """
            if not self._is_consumer_started:
                raise ValueError("Please run startup before listening.")
    
            async for raw_kafka_message in self._aiokafka_consumer:  # type: ignore
                if raw_kafka_message.topic == "test_topic":    # Pseudocode
                   yield raw_kafka_message.value
                
broker = CustomAioKafkaBroker(bootstrap_servers="localhost", kafka_topic="test_topic")

max_prefetch: Limit the number of messages accepted by the local queue. #127

max_async_tasks : The maximum number of concurrent processing tasks per worker process.

[s3rius]

Hi, @MuriloScarpaSitonio, thanks for warm words. I'm not sure what you're trying to achieve.

So, both of your points are half true.

• Yes, messages are just formatted TaskiqMessage objects. We use formatters to dump and parse messages. So, you can set any formatter your like, when creating a broker. By default we use JSONFormatter.

   from taskiq.formatters.json_formatter import JSONFormatter
   broker = MyBroker()
   broker.formatter = JSONFormatter()

  You should yield message formatted with formatter. For your example it should be something like:

       for i in range(1, 4):
           await asyncio.sleep(i)
   		yield self.formatter.dumps(
   			TaskiqMessage(
   				task_id=uuid4().hex, 
   				task_name=my_task.task_name,
   				labels={}, 
   				args=[], 
   				kwargs={}
   		)).message

• You don't need to assign task names explicitly every time, unless it breaks something. We generate task names based on python path to function. If it doesn't work for you, then you should set it explicitly.

Receiver

Receiver is a class, that helps you to start listen to messages for a specific broker. You can use it in your code, for example, if you want to start multiple brokers at the same time. Or you don't want to run taskiq worker command and want to listen inside the same application (maybe on a different thread or process).

But if you just want to listen to incoming messages, I advise you to spin a taskiq-worker program. It has automatic healing for dead processes and can run multiple consuming processes.

taskiq worker `my_project.my_module:broker`

You can read more information in Distributed run section or in Cli|Worker section in documentation.

Also, you can see all arguments and options by running:

taskiq worker --help

[MuriloScarpaSitonio]

I apologize for the confusion. I didn't realize that what I want to accomplish is already implemented in a "normal" worker (through taskiq worker command).

I think I've understood the max_async_tasks argument: it's essentially the number of messages a worker can process concurrently. If I have only one worker and my queue receives 100 messages simultaneously, the worker can handle them all without idle time because it'll spawn 100 asynchronous process (because the default value is 100), correct?

However, I'm not sure if I've understood max_prefetch. From my understanding (at least in celery), the prefetch is a limit for the number of tasks a worker can reserve, in memory, for itself. The default value (zero) seems to me that the worker will keep consuming messages even if other workers are available to process them sooner. Am I missing something or max_prefetch=0 means something else? Would you agree that for short-running tasks this number should be large while for while for long-running tasks this number should be one?

[s3rius]

Hi, @MuriloScarpaSitonio, you're right about max_async_tasks. Basically it's a number of how many tasks can run concurrently. We created this number to avoid slow processing and loop overloading with tasks.

The max_prefetch option, is a bit complicated, but I'll try to explain using an example. Imagine you have set max_prefetch to 25, max_async_tasks to 100 and use only one worker process to listen to messages. If you send 150 messages, then 100 messages are going to be processed immediately, 25 messages are going to be stored in memory, and when worker decide to get more tasks to put them in the loop, it is going to take tasks from the memory instead of fetching them from network. It's used to reduce latency, so you can read messages from queue in advance while your loop is busy running n other tasks, where n is equal to max_async_tasks.

However, if you set max_prefetch to 0 no messages are going to be read in advance and when worker will ask for more messages, it's going to read them from network.

[MuriloScarpaSitonio]

Thank you very much for the clear explanation, @s3rius!! It was really helpful.

From what you've said the only "point of attention" is to not set a very big value for max_prefetch. Have you ran any kind of tests to asset a reasonable value? Maybe using the same as max_async_tasks is a good choice?

[s3rius]

Personally I haven't got so many tasks that it required advanced fetching. I would suggest to set it to quater of the valuemax_async_tasks. It should be a good fit for many systems.

Also, it depends on what tasks do you have. If you have lots of small tasks, you can increase this number. If you have lots of long-running tasks, you may want to reduce it, because for you it would be better that workers grab messages from distributed queue when they're ready to process them.

If you have many small tasks along with many long-running tasks, I guess it would be better to disable this parameter completely, because you may face the situation, when worker processes 100 long-running tasks and prefetched 25 small tasks. Because of that processing of your small tasks is going to be delayed.

Of course these rules doesn't apply to systems where you can easily scale out workers, it's more about systems with lots of tasks and when it's critical to have minimum number of worker processes.

[s3rius]

Also. Since this issue called "Improve documentation". Can you please tell us how we can improve our documentation so it would be easier for new users to start enjoying working with taskiq.

And if you find your experience with taskiq pleasant, you can help us improve, by hitting the star button, giving us feedback here in the discussion or becoming a sponsor (Github will give you a cool sponsor emblem on your profile).

[vvanglro]

@s3rius As it stands, the messages in the local queue will be lost after worker restarts, right? (using any messaging middleware)

[s3rius]

@vvanglro, it depends. It won't be lost if your queue supports message acknowledgement.

For example, take a look at this line in aio-pika broker implementation for RabbitMQ. It returns ackable message and if ack is not called in some configured time, the message is considered as not delivered, so it would be returned back to the RabbitMQ.

Of course there's a possible drawback of using short timeouts for acknowledges, because your message can be prefetched and stored in memory, but if it's not going to be executed before ack timeouts, you may receive the same message in another worker process.

[vvanglro]

Yes, sorry, the latest code has added ack. I missed it.😂

[MuriloScarpaSitonio]

Also. Since this issue called "Improve documentation". Can you please tell us how we can improve our documentation so it would be easier for new users to start enjoying working with taskiq.

I think that I got confused because the documentation is heavy-focused on the task-dispatching-side (i.e. my_task.kiq(...)) of systems that use messaging for communication. However, I acknowledge that the docs are actually implicity obvious (ok I know that sounds weird) that what I was trying to accomplish (a system that does not dispatch tasks but rather only consumes them) is already handled by the CLI command.

Maybe add an example where the message is dispatched from the terminal (e.g. using redis-cli) and a python code to simply process the message?! As I said, this is already implicit written in the docs so not sure if there's anything to improve really.

[s3rius]

Maybe a video explanation can do a better job?

Just example of how to start using taskiq with some existing fastapi application.

[MuriloScarpaSitonio]

Sounds good!