[mosquitto-dev] Can one process have multiple user threads, each with it

["Costescu, Nick P Collins" <Nicolae.Costescu@xxxxxxxxxxx>]

Hi Everyone,

 

Thank you for creating mosquitto, I’ve really enjoyed using it.

 

We’ve built the threaded version of 2.0.18 x64 for Windows with pthreads 2.9.1. Seems to work fine with loop_start()/loop_stop() for the most part, but I get some unexpected fail to publish errors, so I would very much appreciate a sanity check of my assumptions (including order of operations).

 

Each instance of my MqttClient class has its own struct mosquitto * which it allocates and uses. This is the order of operations:

• oneTimeInit() (see below – call mosquitto_lib_init() only if this is the first MqttClient instance (thread) to call this). Increment a ref count which tracks the # of times this function has been called.
• mosquitto_new() & set options (use v5, disable Nagle’s)
• set callbacks
• connect
• loop_start()
• do work until we’re done
• disconnect
• loop_stop()
• oneTimeShutdown() (see below – call mosquitto_lib_cleanup() only if this is the last MqttClient instance (thread) to call this. Determine by seeing if the ref count set above is down to 0)

 

I don’t use mosquitto_threaded_set() because I’m using loop_start().

 

I’m using 2 of these MqttClient objects and I’ve verified that each has its own separate network IO thread started by loop_start() by printing the thread ID from the on message callback.

 

My questions are:

1. Is this order of operations correct?
2. Is libmosquitto designed so that you can have multiple threads, each with their own client connection (own struct mosquito* independently created and connected and own loop_start()) in one process?  
3. What is the correct way to do the one-time library init and cleanup in this situation?

Thanks!

Nick

 

Here are my oneTimeInit() and oneTimeShutdown() functions, for reference:

 

void MqttClient::oneTimeInit()

{

   lock_guard<Lock> m(s_libraryLock);

   if (s_libraryInitialized == 0)

   {

      if (mosquitto_lib_init() != MOSQ_ERR_SUCCESS)

      {

         throw runtime_error("mosquitto_lib_init failed.");

      }

   }

   s_libraryInitialized++;

}

 

void MqttClient::oneTimeShutdown()

{

   lock_guard<Lock> m(s_libraryLock);

   s_libraryInitialized--;

   if (s_libraryInitialized == 0)

   {

      mosquitto_lib_cleanup();

   }

}

----------

Nicolae P. Costescu, Ph.D. | Pr. Software Engineer | Mission Systems

COLLINS AEROSPACE

22640 Davis Dr., MS 291-100, Sterling, VA 20164 U.S.A

Tel: +1 703 880 5648

nicolae.costescu@xxxxxxxxxxx | collinsaerospace.com