Fixed an issue where, on single-core machines, the acceptThread could

get stuck due to the messageThread not yielding and hogging the fdsMtx
lock. This caused long waiting times for new connections to be
accept()ed.

CollectAgent/simplemqttserverthread.cpp

@@ -87,6 +87,20 @@ void* SimpleMQTTServerThread::launch(void *selfPtr)
   return NULL;
 }
 
+void SimpleMQTTServerThread::yield()
+{
+  /*
+   * Yield this thread's execution to give way to other threads.
+   * This is being used in the SimpleMQTTServer to allow the
+   * accept thread acquiring the fdsMtx lock. Otherwise, in
+   * particular on single-CPU systems (or when the acceptThread
+   * and the messageThread share a core), waiting connections
+   * may have to wait a long time before the lock will be given
+   * to the acceptThread.
+   */
+  pthread_yield();
+}
+
 void SimpleMQTTServerAcceptThread::run()
 {
 #ifdef SimpleMQTTVerbose
@@ -145,12 +159,29 @@ void SimpleMQTTServerAcceptThread::run()
           if (newsock != -1) {
               int opt = fcntl(newsock, F_GETFL, 0);
               if (opt == -1 || fcntl(newsock, F_SETFL, opt | O_NONBLOCK)==-1) {
+#ifdef SimpleMQTTVerbose
+          coutMtx.lock();
+          cout << "Setting socket to non-blocking in thread (" << this << ") failed for socket " << newsock << "...\n";
+          coutMtx.unlock();
+#endif
                   close(newsock);
               }
               else {
+#ifdef SimpleMQTTVerbose
+          coutMtx.lock();
+          cout << "Successfully set socket " << newsock << " nonblocing in thread (" << this << ")...\n";
+          coutMtx.unlock();
+#endif
                   mt->assignConnection(newsock);
               }
           }
+          else {
+#ifdef SimpleMQTTVerbose
+          coutMtx.lock();
+          cout << "Accept() in thread (" << this << ") returned -1...\n";
+          coutMtx.unlock();
+#endif
+          }
       }
   }
 }
@@ -206,6 +237,14 @@ void SimpleMQTTServerMessageThread::run()
       numfds = poll(fds, SimpleMQTTConnectionsPerThread, SimpleMQTTPollTimeout);
       fdsMtx.unlock();
 
+      /*
+       * Allow other threads to run (avoid instantaneous
+       * re-acquisition of the fdsMtx lock in the next
+       * iteration while connections are pending in the
+       * acceptThread).
+       */
+      yield();
+
       if (numfds == -1)
         throw new system_error(errno, system_category(), "Error in poll().");
 

CollectAgent/simplemqttserverthread.h

@@ -17,6 +17,7 @@ protected:
   bool terminate;
 
   static void* launch(void* thisPtr);
+  static void yield();
   virtual void run() = 0;
 
 public: