Performance comparison

[JoseAntonioTorre]

These past few days I have been working with this library and first of all, I wanted to thank the entire team for all the work done.

I have been testing the library to simulate 10,000 producers sending messages every second, half a second, and every 100ms. I have encountered some problems with my complete program, so I preferred to do some simple benchmarks to see if I am misusing the library or if these are the actual limits.

First, I tried paho-mqtt using the following code (obtained from the internet). I get about 20,000 messages sent per second.

import paho.mqtt.client as mqtt
from time import time, sleep
import random


BROKER = "localhost"
NB_MESSAGES = 100000
PAYLOAD_LEN = 128
TOPIC = 'a/b/c/d'

rcpt_counter = 0


def on_disconnect(client, userdata, rc):
    elapsed = time() - T0
    print('sending', NB_MESSAGES / elapsed, 'messages per sec')

def on_message(client, userdata, msg):
    global T1, rcpt_counter
    rcpt_counter += 1
    if rcpt_counter % 1000 == 0:
        T2 = time()
        print('  receiveing', 1000 / (T2 - T1), 'messages per sec')
        T1 = T2

p = mqtt.Client()
p.on_disconnect = on_disconnect
p.connect(BROKER)
p.loop_start()

c = mqtt.Client()
c.connect(BROKER)
c.on_message = on_message
c.subscribe(TOPIC)
c.loop_start()

#prepare some random data
data = [''.join(chr(random.getrandbits(8)) for _ in range(PAYLOAD_LEN))
        for _ in range(NB_MESSAGES)]

T0 = T1 = time()

for i in range(NB_MESSAGES):
    p.publish(TOPIC, data[i])

p.disconnect()

sleep(20)
c.disconnect()

On the other hand, with the following benchmark I did using async-mqtt, I get about 10k messages per second.

import asyncio
from asyncio_mqtt import Client
import logging
import time

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

messages = 0

async def monitor_task():
    start_time = 0
    while True:
        await asyncio.sleep(1)
        global messages
        elapsed = time.time() - start_time
        logger.debug(f"Messages: {messages/elapsed} per second")
        messages = 0
        start_time = time.time()

async def publisher(client):
    await client.publish("test", "Hello World!")
    global messages
    messages += 1
    logger.debug("Messages: " + messages)

async def main():
    async with Client("localhost") as client:
        while True:
            tasks = set()
            for i in range(1000):
                task = publisher(client)
                tasks.add(task)

            await asyncio.gather(*tasks, return_exceptions=True)

async def bootstrap():
    tasks = [monitor_task(), main()]
    await asyncio.gather(*tasks, return_exceptions=True)


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

In the production code, I have also observed that asyncio, in debug mode, complains that the function on_socket_register_write callback takes too much time (0.4 seconds, 400ms), specifically in client.loop_write(). Could it be that this function is blocking asyncio?

I hope everything is clear.

[frederikaalund]

Hi Jose. Thanks for opening this discussion. Always nice with a code example to reproduce the issue with. 👍 It's an interesting benchmark!

First of: Python (in 2023) is about correctness and maintainability. Not performance. If you want performance, start with another programming language.
With that out of the way, let's try to understand what's going on here. :)

I ran the code on my machine with a local mosquitto MQTT server, python 3.10, and using Ubuntu 20.04. Roughly, I got

• paho-mqtt: 45K msg/s
• asyncio-mqtt: 11K msg/s

Quite the difference. I changed the publishers "batch" size from 1000 to 100 (for i in range(100)) and got:

• asyncio-mqtt: 18K msg/s

It's always a challenge to find the optimal batch size. Depends on the overhead of async scheduling and the task itself. To set things in perspective, try to comment out await client.publish(...). This yield around 200K msg/s. That's the performance limit. We're actually not far from it (only a factor of 10). Let's get closer to the limit.

Next, I switch from the built-in asyncio event loop to uvloop. That gives:

• asyncio-mqtt: 21K msg/s

Not much of a difference but still better.

Now I cheat: I go into the asyncio-mqtt source code and comment out the "wait until the publish succeeds" logic inside Client.publish. This gives:

• asyncio-mqtt: 43K msg/s

That seems to be the main culprit! We are now very close to raw paho-mqtt.

It also showcases one of the key differences between asyncio-mqtt and paho-mqtt: asyncio-mqtt waits for acknowledgement from the server before it proceeds. paho-mqtt uses a fire-and-forget approach.

The big question is now: Why is asyncio-mqtt just about on par with paho-mqtt in terms of performance? Why are we still far from the performance limit of 200K msgs/s? Isn't async/await supposed to be fast? The answer is usually yes, but only if you use async I/O in depth. We (asyncio-mqtt) do not. We simply call paho-mqtt and let paho deal with the I/O. This means that underneath everything goes through python's sync socket interface. That limits the potential concurrency. We'll never be faster than paho itself.

So we use asyncio-mqtt and not just paho-mqtt? Well, use whatever you want. :) Don't expect asyncio-mqtt to give you extra performance due to concurrent I/O (it's all paho behind the scenes). asyncio-mqtt gives you safety on top of paho. It proceeds carefully and always makes sure that your requests (publish, subscribe, etc.) gets an acknowledgement. This cautious approach costs in terms of performance. That is what you see in your benchmarks.

I hope that helps explain it.

[JoseAntonioTorre]

Hello @frederikaalund , thank you very much for your response. More or less, I have been conducting the same experiments that you have done, and I have come to the same conclusions. Indeed, what has the most impact is the message confirmation, and I suppose the broker's own speed comes into play there. I also tried the changes with uvloop, but I left that for later as "fine-tuning."

One of the things I realized during these tests is that the messages were being sent in batches. Let me explain, in the production code, I have about 10,000 objects that perform instant sending each one in their own task. I tried to put a print before the await publish and another one after. What I observed is that until all the messages before the await were printed in "mosquitto_sub," I could not see the messages.

Looking at the code in detail, I suppose this is because the method responsible for sending the messages is the loop_write() that we invoke from the on_socket_register_write method. Therefore, these calls are "queued" at the end inside the event_loop. The problem is that if the number of tasks is high, then the latencies start to accumulate, and the time from await publish to loop_write call grows. This is not a problem with async-mqtt, but with the asyncio paradigm itself. I suppose we are reaching the limit of concurrent tasks, and in the end, we are only using one core.

My question is whether it could be interesting to handle the mqtt loop in a separate thread. In this way, we would not be limited by asyncio, and we would have the opportunity to attend to the internal loop of paho-mqtt, as long as it allows us the GIL.

I did some tests by opening the async-mqtt client in another thread and creating a wrapper around publish that enqueues in a queue that is serviced from the other thread, and indeed, the mqtt messages are seen to be interleaved between the print before the await and the print after.

[frederikaalund]

Great to hear that it helped. :)

I suppose we are reaching the limit of concurrent tasks, and in the end, we are only using one core.

It's more about OS-level concurrency than hardware-based parallelism (here is a great thread on that topic). asyncio-mqtt uses paho-mqtt. paho-mqtt uses socket. socket is a sync interface (no concurrency).

If we want OS-level concurrency, we need to use an actual async interface. E.g.: anyio.open_connection/loop.create_connection instead of socket. That's what, e.g., gmqtt does.

In a perfect world, we would have:

• A safe interface (like asyncio-mqtt)
• An actual async I/O backend (like gmqtt)
• The maturity of paho-mqtt

I don't think such a library exists yet. When such a library appears, I'll gladly link to that as a replacement for asyncio-mqtt!

[JoseAntonioTorre]

Also, I'm not sure if it will help, but I have observed that when I have a high load of concurrent tasks sending messages, the on_socket_register_write function takes more than 100 ms to process and therefore acts as a blocking call for the asyncio loop. Furthermore, I get a warning message indicating that the function blocked the loop for too long. I have measured before and after and sporadically obtain times of 100 to 200 ms.

def _on_socket_register_write(
        self, client: mqtt.Client, userdata: Any, sock: _PahoSocket
    ) -> None:
        def callback() -> None:
            # client.loop_write() may raise an exception, such as BadPipe. It's
            # usually a sign that the underlaying connection broke, therefore we
            # disconnect straight away
            try:
                start = time.time()
                client.loop_write()
                print("loop_write", time.time() - a)
            except Exception as exc:
                if not self._disconnected.done():
                    self._disconnected.set_exception(exc)

        # paho-mqtt may call this function from the executor thread on which we've called
        # `self._client.connect()` (see [3]), so we can't do most operations on
        # self._loop directly.
        self._loop.call_soon_threadsafe(self._loop.add_writer, sock, callback)