Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Roundtrips take at least 40ms or are not in the right order when publishing in subscribe #874

Open
maurerle opened this issue Dec 1, 2024 · 2 comments
Labels
Status: Available No one has claimed responsibility for resolving this issue.

Comments

@maurerle
Copy link

maurerle commented Dec 1, 2024

Bug Description

I need to respond to a received message by publishing another message.
I do not need QoS, but I do need to have the correct message ordering.

In when using an agent-based tool mango-agents, round trip time of MQTT connections is always at least 40ms - even when using QoS=0

An example of this is shown in OFFIS-DAI/mango#142

I tried reproducing this with plain paho-mqtt and got that far:

This does work with QoS = 1 set, in which case it takes 40ms for each roundtrip.
(I see that QoS=1 itself has a roundtrip, though this does not explain 40ms of waiting).
With QoS=0, the messages are not received in the correct order.
Adding wait_for_publish deadlocks the application.
Usage of the Threadpool did not help either.

# python 3.11-3.13
import logging
import random
from multiprocessing.pool import ThreadPool

import paho.mqtt.client as paho

logger = logging.getLogger(__name__)

myQOS = 1
broker = "localhost"
port = 1883
topic = "python/mqtt"
# Generate a Client ID with the publish prefix.
client_id = f"publish-{random.randint(0, 1000)}"

pool = ThreadPool(processes=4)

def connect_mqtt(topic):
    def on_connect(client, userdata, flags, rc, pa):
        if rc == 0:
            logger.info("Connected to MQTT Broker!")
        else:
            logger.error("Failed to connect, return code %d\n", rc)
        
    def thread_publish(client, topic, payload):
        info = client.publish("no topic", payload, qos=myQOS)
        # wait for publish does deadlock here
        #info.wait_for_publish()

    def on_message(client, userdata, message: paho.MQTTMessage):
        logger.info("got message %s", message.payload)
        # create a thread pool where we submit the publishing
        # and wait for the result
        pool.apply(thread_publish, args=(client, topic, message.payload))
        #info = client.publish("no topic", message.payload, qos=myQOS)

    client = paho.Client(paho.CallbackAPIVersion.VERSION2, client_id)
    client.on_connect = on_connect
    client.on_message = on_message
    client.connect(broker, port)
    client.subscribe(topic)

    return client


def publish(client, topic, max_count=100):
    msg_count = 1
    while True:
        msg = f"messages: {msg_count}"
        result = client.publish(topic, msg, qos=myQOS)
        result.wait_for_publish()
        # result: [0, 1]
        status = result[0]
        if status == 0:
            logger.info("Send `%s` to topic `%s`", msg, topic)
        else:
            print(f"Failed to send message to topic {topic}")
        msg_count += 1
        if msg_count > max_count:
            break


def run():
    client = connect_mqtt(topic)
    client.loop_start()
    #client.subscribe("no topic")

    publish(client, topic)

    # this helps when QoS is 0 to wait until all is received
    import time
    time.sleep(1)

    client.loop_stop()


if __name__ == "__main__":
    logging.basicConfig(format="%(asctime)s;%(levelname)s;%(message)s", level="INFO")
    run()

Reproduction

Please provide detailed steps showing how to replicate the issue (it's difficult to fix an issue we cannot replicate).
If errors are output then include the full error (including any stack trace).

Most issues should include a minimal example that
demonstrates the issue (ideally one that can be run without modification, i.e. runnable code using a public broker).

Environment

  • Python version: 3.12.6 (any of 3.11-3.13 has this issue)
  • Library version: 2.1.0
  • Operating system (including version): Debian 13, trixie
  • MQTT server (name, version, configuration, hosting details): local mosquitto, dockerized:
listener 1883
allow_anonymous true

max_keepalive 3600

Logs

For many issues, especially when you cannot provide code to replicate the issue, it's helpful to include logs. Please
consider including:

2024-12-01 22:37:06,386;INFO;Connected to MQTT Broker!
2024-12-01 22:37:06,386;INFO;got message b'1'
2024-12-01 22:37:06,387;INFO;got message b'messages: 1'
2024-12-01 22:37:06,387;INFO;Send `messages: 1` to topic `python/mqtt`
2024-12-01 22:37:06,430;INFO;got message b'messages: 2'
2024-12-01 22:37:06,431;INFO;Send `messages: 2` to topic `python/mqtt`
2024-12-01 22:37:06,431;INFO;got message b'messages: 3'
2024-12-01 22:37:06,432;INFO;Send `messages: 3` to topic `python/mqtt`
2024-12-01 22:37:06,478;INFO;got message b'messages: 4'
2024-12-01 22:37:06,480;INFO;Send `messages: 4` to topic `python/mqtt`
2024-12-01 22:37:06,481;INFO;got message b'messages: 5'
2024-12-01 22:37:06,482;INFO;Send `messages: 5` to topic `python/mqtt`
2024-12-01 22:37:06,530;INFO;got message b'messages: 6'
2024-12-01 22:37:06,532;INFO;Send `messages: 6` to topic `python/mqtt`
2024-12-01 22:37:06,533;INFO;got message b'messages: 7'
2024-12-01 22:37:06,535;INFO;Send `messages: 7` to topic `python/mqtt`
2024-12-01 22:37:06,578;INFO;got message b'messages: 8'
2024-12-01 22:37:06,580;INFO;Send `messages: 8` to topic `python/mqtt`
2024-12-01 22:37:06,581;INFO;got message b'messages: 9'
2024-12-01 22:37:06,582;INFO;Send `messages: 9` to topic `python/mqtt`
2024-12-01 22:37:06,626;INFO;got message b'messages: 10'
2024-12-01 22:37:06,628;INFO;Send `messages: 10` to topic `python/mqtt`
2024-12-01 22:37:06,629;INFO;got message b'messages: 11'
2024-12-01 22:37:06,631;INFO;Send `messages: 11` to topic `python/mqtt`
2024-12-01 22:37:06,674;INFO;got message b'messages: 12'
2024-12-01 22:37:06,676;INFO;Send `messages: 12` to topic `python/mqtt`
2024-12-01 22:37:06,677;INFO;got message b'messages: 13'
2024-12-01 22:37:06,679;INFO;Send `messages: 13` to topic `python/mqtt`
2024-12-01 22:37:06,722;INFO;got message b'messages: 14'
2024-12-01 22:37:06,724;INFO;Send `messages: 14` to topic `python/mqtt`
2024-12-01 22:37:06,725;INFO;got message b'messages: 15'
2024-12-01 22:37:06,727;INFO;Send `messages: 15` to topic `python/mqtt`
2024-12-01 22:37:06,770;INFO;got message b'messages: 16'
2024-12-01 22:37:06,772;INFO;Send `messages: 16` to topic `python/mqtt`
2024-12-01 22:37:06,773;INFO;got message b'messages: 17'
2024-12-01 22:37:06,775;INFO;Send `messages: 17` to topic `python/mqtt`
2024-12-01 22:37:06,822;INFO;got message b'messages: 18'
2024-12-01 22:37:06,824;INFO;Send `messages: 18` to topic `python/mqtt`
2024-12-01 22:37:06,825;INFO;got message b'messages: 19'
2024-12-01 22:37:06,827;INFO;Send `messages: 19` to topic `python/mqtt`
2024-12-01 22:37:06,870;INFO;got message b'messages: 20'
2024-12-01 22:37:06,872;INFO;Send `messages: 20` to topic `python/mqtt`
2024-12-01 22:37:06,873;INFO;got message b'messages: 21'
2024-12-01 22:37:06,874;INFO;Send `messages: 21` to topic `python/mqtt`
2024-12-01 22:37:06,918;INFO;got message b'messages: 22'
2024-12-01 22:37:06,920;INFO;Send `messages: 22` to topic `python/mqtt`
2024-12-01 22:37:06,921;INFO;got message b'messages: 23'
2024-12-01 22:37:06,922;INFO;Send `messages: 23` to topic `python/mqtt`
2024-12-01 22:37:06,966;INFO;got message b'messages: 24'
2024-12-01 22:37:06,968;INFO;Send `messages: 24` to topic `python/mqtt`
2024-12-01 22:37:06,969;INFO;got message b'messages: 25'
2024-12-01 22:37:06,970;INFO;Send `messages: 25` to topic `python/mqtt`
2024-12-01 22:37:07,014;INFO;got message b'messages: 26'
2024-12-01 22:37:07,015;INFO;Send `messages: 26` to topic `python/mqtt`
2024-12-01 22:37:07,017;INFO;got message b'messages: 27'
2024-12-01 22:37:07,018;INFO;Send `messages: 27` to topic `python/mqtt`
2024-12-01 22:37:07,062;INFO;got message b'messages: 28'
2024-12-01 22:37:07,064;INFO;Send `messages: 28` to topic `python/mqtt`
2024-12-01 22:37:07,065;INFO;got message b'messages: 29'
2024-12-01 22:37:07,067;INFO;Send `messages: 29` to topic `python/mqtt`
2024-12-01 22:37:07,110;INFO;got message b'messages: 30'
2024-12-01 22:37:07,111;INFO;Send `messages: 30` to topic `python/mqtt`
2024-12-01 22:37:07,113;INFO;got message b'messages: 31'
2024-12-01 22:37:07,114;INFO;Send `messages: 31` to topic `python/mqtt`
2024-12-01 22:37:07,158;INFO;got message b'messages: 32'
2024-12-01 22:37:07,160;INFO;Send `messages: 32` to topic `python/mqtt`
2024-12-01 22:37:07,161;INFO;got message b'messages: 33'
2024-12-01 22:37:07,162;INFO;Send `messages: 33` to topic `python/mqtt`
2024-12-01 22:37:07,210;INFO;got message b'messages: 34'
2024-12-01 22:37:07,211;INFO;Send `messages: 34` to topic `python/mqtt`
2024-12-01 22:37:07,212;INFO;got message b'messages: 35'
2024-12-01 22:37:07,214;INFO;Send `messages: 35` to topic `python/mqtt`
2024-12-01 22:37:07,258;INFO;got message b'messages: 36'
2024-12-01 22:37:07,260;INFO;Send `messages: 36` to topic `python/mqtt`
2024-12-01 22:37:07,261;INFO;got message b'messages: 37'
2024-12-01 22:37:07,262;INFO;Send `messages: 37` to topic `python/mqtt`
2024-12-01 22:37:07,310;INFO;got message b'messages: 38'
2024-12-01 22:37:07,312;INFO;Send `messages: 38` to topic `python/mqtt`
2024-12-01 22:37:07,313;INFO;got message b'messages: 39'
2024-12-01 22:37:07,315;INFO;Send `messages: 39` to topic `python/mqtt`
2024-12-01 22:37:07,358;INFO;got message b'messages: 40'
2024-12-01 22:37:07,360;INFO;Send `messages: 40` to topic `python/mqtt`
2024-12-01 22:37:07,361;INFO;got message b'messages: 41'
2024-12-01 22:37:07,362;INFO;Send `messages: 41` to topic `python/mqtt`
2024-12-01 22:37:07,406;INFO;got message b'messages: 42'
2024-12-01 22:37:07,408;INFO;Send `messages: 42` to topic `python/mqtt`
2024-12-01 22:37:07,409;INFO;got message b'messages: 43'
2024-12-01 22:37:07,410;INFO;Send `messages: 43` to topic `python/mqtt`
2024-12-01 22:37:07,454;INFO;got message b'messages: 44'
2024-12-01 22:37:07,456;INFO;Send `messages: 44` to topic `python/mqtt`
2024-12-01 22:37:07,457;INFO;got message b'messages: 45'
2024-12-01 22:37:07,458;INFO;Send `messages: 45` to topic `python/mqtt`
2024-12-01 22:37:07,502;INFO;got message b'messages: 46'
2024-12-01 22:37:07,504;INFO;Send `messages: 46` to topic `python/mqtt`
2024-12-01 22:37:07,505;INFO;got message b'messages: 47'
2024-12-01 22:37:07,507;INFO;Send `messages: 47` to topic `python/mqtt`
2024-12-01 22:37:07,554;INFO;got message b'messages: 48'
2024-12-01 22:37:07,556;INFO;Send `messages: 48` to topic `python/mqtt`
2024-12-01 22:37:07,557;INFO;got message b'messages: 49'
2024-12-01 22:37:07,558;INFO;Send `messages: 49` to topic `python/mqtt`
2024-12-01 22:37:07,602;INFO;got message b'messages: 50'
2024-12-01 22:37:07,603;INFO;Send `messages: 50` to topic `python/mqtt`
2024-12-01 22:37:07,604;INFO;got message b'messages: 51'
2024-12-01 22:37:07,606;INFO;Send `messages: 51` to topic `python/mqtt`
2024-12-01 22:37:07,650;INFO;got message b'messages: 52'
2024-12-01 22:37:07,652;INFO;Send `messages: 52` to topic `python/mqtt`
2024-12-01 22:37:07,653;INFO;got message b'messages: 53'
2024-12-01 22:37:07,654;INFO;Send `messages: 53` to topic `python/mqtt`
2024-12-01 22:37:07,698;INFO;got message b'messages: 54'
2024-12-01 22:37:07,699;INFO;Send `messages: 54` to topic `python/mqtt`
2024-12-01 22:37:07,700;INFO;got message b'messages: 55'
2024-12-01 22:37:07,702;INFO;Send `messages: 55` to topic `python/mqtt`
2024-12-01 22:37:07,746;INFO;got message b'messages: 56'
2024-12-01 22:37:07,747;INFO;Send `messages: 56` to topic `python/mqtt`
2024-12-01 22:37:07,749;INFO;got message b'messages: 57'
2024-12-01 22:37:07,750;INFO;Send `messages: 57` to topic `python/mqtt`
2024-12-01 22:37:07,794;INFO;got message b'messages: 58'
2024-12-01 22:37:07,796;INFO;Send `messages: 58` to topic `python/mqtt`
2024-12-01 22:37:07,797;INFO;got message b'messages: 59'
2024-12-01 22:37:07,799;INFO;Send `messages: 59` to topic `python/mqtt`
2024-12-01 22:37:07,842;INFO;got message b'messages: 60'
2024-12-01 22:37:07,844;INFO;Send `messages: 60` to topic `python/mqtt`
2024-12-01 22:37:07,845;INFO;got message b'messages: 61'
2024-12-01 22:37:07,846;INFO;Send `messages: 61` to topic `python/mqtt`
2024-12-01 22:37:07,894;INFO;got message b'messages: 62'
2024-12-01 22:37:07,896;INFO;Send `messages: 62` to topic `python/mqtt`
2024-12-01 22:37:07,897;INFO;got message b'messages: 63'
2024-12-01 22:37:07,898;INFO;Send `messages: 63` to topic `python/mqtt`
2024-12-01 22:37:07,942;INFO;got message b'messages: 64'
2024-12-01 22:37:07,944;INFO;Send `messages: 64` to topic `python/mqtt`
2024-12-01 22:37:07,944;INFO;got message b'messages: 65'
2024-12-01 22:37:07,946;INFO;Send `messages: 65` to topic `python/mqtt`
2024-12-01 22:37:07,990;INFO;got message b'messages: 66'
2024-12-01 22:37:07,992;INFO;Send `messages: 66` to topic `python/mqtt`
2024-12-01 22:37:07,993;INFO;got message b'messages: 67'
2024-12-01 22:37:07,994;INFO;Send `messages: 67` to topic `python/mqtt`
2024-12-01 22:37:08,038;INFO;got message b'messages: 68'
2024-12-01 22:37:08,040;INFO;Send `messages: 68` to topic `python/mqtt`
2024-12-01 22:37:08,041;INFO;got message b'messages: 69'
2024-12-01 22:37:08,042;INFO;Send `messages: 69` to topic `python/mqtt`
2024-12-01 22:37:08,086;INFO;got message b'messages: 70'
2024-12-01 22:37:08,088;INFO;Send `messages: 70` to topic `python/mqtt`
2024-12-01 22:37:08,089;INFO;got message b'messages: 71'
2024-12-01 22:37:08,091;INFO;Send `messages: 71` to topic `python/mqtt`
2024-12-01 22:37:08,134;INFO;got message b'messages: 72'
2024-12-01 22:37:08,135;INFO;Send `messages: 72` to topic `python/mqtt`
2024-12-01 22:37:08,137;INFO;got message b'messages: 73'
2024-12-01 22:37:08,138;INFO;Send `messages: 73` to topic `python/mqtt`
2024-12-01 22:37:08,182;INFO;got message b'messages: 74'
2024-12-01 22:37:08,183;INFO;Send `messages: 74` to topic `python/mqtt`
2024-12-01 22:37:08,184;INFO;got message b'messages: 75'
2024-12-01 22:37:08,186;INFO;Send `messages: 75` to topic `python/mqtt`
2024-12-01 22:37:08,234;INFO;got message b'messages: 76'
2024-12-01 22:37:08,236;INFO;Send `messages: 76` to topic `python/mqtt`
2024-12-01 22:37:08,237;INFO;got message b'messages: 77'
2024-12-01 22:37:08,238;INFO;Send `messages: 77` to topic `python/mqtt`
2024-12-01 22:37:08,282;INFO;got message b'messages: 78'
2024-12-01 22:37:08,283;INFO;Send `messages: 78` to topic `python/mqtt`
2024-12-01 22:37:08,284;INFO;got message b'messages: 79'
2024-12-01 22:37:08,286;INFO;Send `messages: 79` to topic `python/mqtt`
2024-12-01 22:37:08,334;INFO;got message b'messages: 80'
2024-12-01 22:37:08,335;INFO;Send `messages: 80` to topic `python/mqtt`
2024-12-01 22:37:08,336;INFO;got message b'messages: 81'
2024-12-01 22:37:08,338;INFO;Send `messages: 81` to topic `python/mqtt`
2024-12-01 22:37:08,386;INFO;got message b'messages: 82'
2024-12-01 22:37:08,388;INFO;Send `messages: 82` to topic `python/mqtt`
2024-12-01 22:37:08,389;INFO;got message b'messages: 83'
2024-12-01 22:37:08,390;INFO;Send `messages: 83` to topic `python/mqtt`
2024-12-01 22:37:08,434;INFO;got message b'messages: 84'
2024-12-01 22:37:08,436;INFO;Send `messages: 84` to topic `python/mqtt`
2024-12-01 22:37:08,437;INFO;got message b'messages: 85'
2024-12-01 22:37:08,438;INFO;Send `messages: 85` to topic `python/mqtt`
2024-12-01 22:37:08,482;INFO;got message b'messages: 86'
2024-12-01 22:37:08,483;INFO;Send `messages: 86` to topic `python/mqtt`
2024-12-01 22:37:08,485;INFO;got message b'messages: 87'
2024-12-01 22:37:08,486;INFO;Send `messages: 87` to topic `python/mqtt`
2024-12-01 22:37:08,534;INFO;got message b'messages: 88'
2024-12-01 22:37:08,535;INFO;Send `messages: 88` to topic `python/mqtt`
2024-12-01 22:37:08,537;INFO;got message b'messages: 89'
2024-12-01 22:37:08,539;INFO;Send `messages: 89` to topic `python/mqtt`
2024-12-01 22:37:08,582;INFO;got message b'messages: 90'
2024-12-01 22:37:08,583;INFO;Send `messages: 90` to topic `python/mqtt`
2024-12-01 22:37:08,584;INFO;got message b'messages: 91'
2024-12-01 22:37:08,586;INFO;Send `messages: 91` to topic `python/mqtt`
2024-12-01 22:37:08,630;INFO;got message b'messages: 92'
2024-12-01 22:37:08,632;INFO;Send `messages: 92` to topic `python/mqtt`
2024-12-01 22:37:08,633;INFO;got message b'messages: 93'
2024-12-01 22:37:08,634;INFO;Send `messages: 93` to topic `python/mqtt`
2024-12-01 22:37:08,678;INFO;got message b'messages: 94'
2024-12-01 22:37:08,679;INFO;Send `messages: 94` to topic `python/mqtt`
2024-12-01 22:37:08,681;INFO;got message b'messages: 95'
2024-12-01 22:37:08,682;INFO;Send `messages: 95` to topic `python/mqtt`
2024-12-01 22:37:08,726;INFO;got message b'messages: 96'
2024-12-01 22:37:08,728;INFO;Send `messages: 96` to topic `python/mqtt`
2024-12-01 22:37:08,729;INFO;got message b'messages: 97'
2024-12-01 22:37:08,730;INFO;Send `messages: 97` to topic `python/mqtt`
2024-12-01 22:37:08,778;INFO;got message b'messages: 98'
2024-12-01 22:37:08,780;INFO;Send `messages: 98` to topic `python/mqtt`
2024-12-01 22:37:08,781;INFO;got message b'messages: 99'
2024-12-01 22:37:08,783;INFO;Send `messages: 99` to topic `python/mqtt`
2024-12-01 22:37:08,826;INFO;got message b'messages: 100'
2024-12-01 22:37:08,827;INFO;Send `messages: 100` to topic `python/mqtt`
@github-actions github-actions bot added the Status: Available No one has claimed responsibility for resolving this issue. label Dec 1, 2024
@maurerle
Copy link
Author

maurerle commented Dec 2, 2024

Interesting:

When using the below code, which only sends the first message from the main loop and then lets the on_message and publish method do its ping pong, no deadlock happens, when QoS=0.
Though the 40ms wait time appear as well if QoS >0:

# python 3.11-3.13
import logging
import random
from multiprocessing.pool import ThreadPool

import paho.mqtt.client as paho

logger = logging.getLogger(__name__)

myQOS = 1
broker = "localhost"
port = 1883
topic = "python/mqtt"
# Generate a Client ID with the publish prefix.
client_id = f"publish-{random.randint(0, 1000)}"

pool = ThreadPool(processes=4)

def connect_mqtt(topic):
    def on_connect(client, userdata, flags, rc, pa):
        if rc == 0:
            logger.info("Connected to MQTT Broker!")
        else:
            logger.error("Failed to connect, return code %d\n", rc)
        
    def thread_publish(client, topic, payload: int):
        number = int(payload) + 1
        if number < 100:
            info = client.publish(topic, number, qos=myQOS)
            logger.info("did send message %s", number)
        # wait for publish does deadlock here
        #info.wait_for_publish()

    def on_message(client, userdata, message: paho.MQTTMessage):
        logger.info("got message %s", message.payload)
        
        payload = message.payload
        # using a threadpool here did not help either
        # pool.apply(thread_publish, args=(client, topic, message.payload))
        number = int(payload) + 1
        if number < 100:
            info = client.publish(topic, number, qos=myQOS)
            # wait for publish does deadlock here
            #info.wait_for_publish()
            logger.info("did send message %s", number)

    client = paho.Client(paho.CallbackAPIVersion.VERSION2, client_id)
    client.on_connect = on_connect
    client.on_message = on_message
    client.connect(broker, port)
    client.subscribe(topic)

    return client


def run():
    client = connect_mqtt(topic)
    client.loop_start()
    #client.subscribe("no topic")

    result = client.publish(topic, 1, qos=myQOS)
    result.wait_for_publish()

    # this helps when QoS is 0 to wait until all is received
    import time
    time.sleep(1)

    client.loop_stop()


if __name__ == "__main__":
    logging.basicConfig(format="%(asctime)s;%(levelname)s;%(message)s", level="INFO")
    run()

@maurerle
Copy link
Author

maurerle commented Dec 2, 2024

So I don't know why the code in the issue description blocks and deadlocks when result.wait_for_publish() is set in publish(), but I know now, why the 40ms wait happens:

Adding logging.error("i am here") around

socklist = select.select(rlist, wlist, [], timeout)

shows that the python client waits for the socket io to become available from the MQTT server.
So this is not a problem in the paho-mqtt-python library.

Taking a look at the MQTT server, we can see that this does not occur if set_tcp_nodelay = True is set in the mosquitto.conf

I am not sure if this is one of the programming errors related around set_tcp_nodelay or if this is a valid case.
I am leaving this open for now, though I found a solution for my demands..

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
Status: Available No one has claimed responsibility for resolving this issue.
Projects
None yet
Development

No branches or pull requests

1 participant