paho.mqtt.python icon indicating copy to clipboard operation
paho.mqtt.python copied to clipboard

Published 20 hours ago • verified publisher icon eclipse

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

Open maurerle opened this issue 1 year ago • 4 comments

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 https://github.com/OFFIS-DAI/mango/issues/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`

maurerle avatar Dec 01 '24 21:12 maurerle

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 avatar Dec 02 '24 07:12 maurerle

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 https://github.com/eclipse-paho/paho.mqtt.python/blob/d45de3737879cfe7a6acc361631fa5cb1ef584bb/src/paho/mqtt/client.py#L1663

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..

maurerle avatar Dec 02 '24 07:12 maurerle

Hi,

I didn't yet tried to reproduce you issue. But a delay shouldn't cause a deadlock (that would means the deadlock exists, but it's just much less likely).

The only possible issue I see by quickly reading you code:

  • on the first code, you call client.loop_stop after the "ping" message were send and don't wait anything for the pong message. You could perfectly have all messages being sent by your publish function and those message could still be in the broken. The broker could not yet started to send back to message. Anyway it could means a pong packet can't be send (loop is stopped), but I would expect wait_for_publish to terminate (with an exception)
  • still on the first code, you call subscribe which just send the subscription packet, the broker might not yet received and processed it before you start sending message. It could be better to wait to suback package (on_subscribe callback). Note: I'm not 100% sure this is really an issue, because you first message is sent on the same socket, so the broker will necessary receive the subscription packet before your first message. The point I'm not sure is: does MQTT protocol say anything about order of processing of received packet, is the broker allowed to process subscription "asynchronously" or not ? I think mosquitto handle them synchronously
  • on the 2nd code, it might be invalid to do a blocking call inside a callback: it might be invalid to call wait_for_publish from a callback. I'll need to confirm this by looking at paho-mqtt code, but IIRC callbacks are run on the network loop, so if you block them you block the network loop (which is not great). If you block them because you wait for a event on the network you have a dead-lock (but as you commented "using a threadpool didn't help" so there is another issue).

So I don't see good reason to explain your deadlock. I'll try to reproduce with your sample.

About TCP nodelay, the described behaviour you give match what I expected from TCP no-delay:

  • without TCP no-delay (which is the default on most OS I think): when an application (mosquitto, or paho-mqtt) write to a TCP socket, the operating system (the socket implementation) will not immediately send data on the network. This is done, so if you are doing lots of small write (e.g. send lots of small message), on the network they get aggregated into a single network packet.
  • with TCP no-delay, this delay is removed. It means on write to a TCP socket cause one (or more) network packet and the OS will no longer aggregate multiple application write into a single network packet.
  • Therefore TCP no-delay is a trade-off between reducing network usage versus reducing latency.

PierreF avatar Mar 04 '25 22:03 PierreF

Hi

There is a good reason why your code deadlocks and the thread pool did not help.

As Pierre said, the network thread calls your callbacks. So waiting in callbacks will block the network thread. You may observe that pool.apply(..) is blocking the network thread and wait_for_publish(..) is blocking the worker thread - causing a deadlock.

This deadlock will occur for QoS > 0; otherwise wait_for_publish(..) will not block.

rubenbaer avatar Mar 18 '25 02:03 rubenbaer