Manage socket entirely in thread, control w/ atomic flags

Author: ethanbb

Source/Plugins/NetworkEvents/NetworkEvents.cpp

@@ -41,37 +41,23 @@ CriticalSection NetworkEvents::sharedContextLock{};
 NetworkEvents::NetworkEvents()
     : GenericProcessor  ("Network Events")
     , Thread            ("NetworkThread")
-    , restart           (false)
-    , changeResponder   (false)
-    , urlport           (0)
+    , makeNewSocket     (true)
+    , requestedPort     (5556)
+    , boundPort         (0)
 {
     setProcessorType (PROCESSOR_TYPE_SOURCE);
 
-    if (!setNewListeningPort(5556))
-    {
-        // resort to choosing a port automatically
-        setNewListeningPort(0);
-    }
-
     startThread();
 
     sendSampleCount = false; // disable updating the continuous buffer sample counts,
     // since this processor only sends events
 }
 
 
-bool NetworkEvents::setNewListeningPort(uint16 port)
+void NetworkEvents::setNewListeningPort(uint16 port)
 {
-    ScopedPointer<Responder> newResponder = Responder::makeResponder(port);
-    if (newResponder == nullptr)
-    {
-        return false;
-    }
-
-    ScopedLock nrLock(nextResponderLock);
-    nextResponder = newResponder;
-    changeResponder = true;
-    return true;
+    requestedPort = port;
+    makeNewSocket = true;
 }
 
 
@@ -85,25 +71,15 @@ NetworkEvents::~NetworkEvents()
 }
 
 
-String NetworkEvents::getPortString() const
+String NetworkEvents::getCurrPortString() const
 {
-#ifdef ZEROMQ
-    uint16 port = urlport;
-    if (port == 0)
-    {
-        return "<no cxn>";
-    }
-
-    return String(port);
-#else
-    return "<no zeromq>";
-#endif
+    return getPortString(boundPort);
 }
 
 
 void NetworkEvents::restartConnection()
 {
-    restart = true;
+    makeNewSocket = true;
 }
 
 
@@ -260,51 +236,58 @@ void NetworkEvents::process (AudioSampleBuffer& buffer)
 void NetworkEvents::run()
 {
 #ifdef ZEROMQ
-    ScopedPointer<Responder> responder;
-    char buffer[MAX_MESSAGE_LENGTH];
-    bool connected = false;
+    HeapBlock<char> buffer(MAX_MESSAGE_LENGTH);
+
+    // responder should always be valid (bound to a port) if it is non-null
+    ScopedPointer<Responder> responder(new Responder(0)); // use any available port as default
+    if (responder->isValid())
+    {
+        boundPort = responder->getBoundPort();
+    }
+    else
+    { 
+        responder = nullptr;
+        boundPort = 0;
+    }
+
+    // purposely don't call updatePortString - makeNewSocket will be true on startup,
+    // so wait to try the requested port (5556) before updating the editor.
 
     while (!threadShouldExit())
     {
-        // reopen connection if necessary
-        if (restart)
+        // change socket if necessary
+        while (makeNewSocket.exchange(false))
         {
-            restart = false; // the other thread only sets restart to true, so it's not a race condition
-            responder = nullptr; // destroy old one, which frees the port
-            responder = Responder::makeResponder(urlport);
+            uint16 nextPort = requestedPort; // (maybe the newly entered port on the editor text box)
+            if (nextPort > 0 && nextPort == boundPort) // i.e. this is a restart
+            {
+                responder = nullptr; // destroy old one, which frees the port
+                boundPort = 0;
+            }
 
-            if (responder != nullptr)
+            if (nextPort == 0)
             {
-                connected = true;
-                updatePort(responder->getBoundPort());
+                CoreServices::sendStatusMessage("NetworkEvents: Selecting port automatically");
             }
-        }
 
-        // switch to new responder if necessary
-        if (changeResponder)
-        {
-            const ScopedLock nrLock(nextResponderLock);
-            if (nextResponder != nullptr)
+            ScopedPointer<Responder> newResponder(new Responder(nextPort));
+            if (newResponder->isValid())
             {
-                changeResponder = false; // this is also controlled by the nextResponderLock
-                responder = nextResponder;
-                connected = true;
-                updatePort(responder->getBoundPort());
+                // replace the current socket with the newly created socket
+                responder = newResponder;
+                boundPort = responder->getBoundPort();
             }
             else
             {
-                jassertfalse; // huh? a new responder should be available
+                newResponder->reportErr("Failed to connect to port " + String(nextPort));
             }
+
+            updatePortString(boundPort);
         }
 
         // if we don't have a vaild (connected) socket, keep looping until we do
         if (responder == nullptr)
         {
-            if (connected)
-            {
-                connected = false;
-                updatePort(0);
-            }
             wait(100);
             continue;
         }
@@ -367,7 +350,9 @@ void NetworkEvents::setEnabledState (bool newState)
 void NetworkEvents::saveCustomParametersToXml (XmlElement* parentElement)
 {
     XmlElement* mainNode = parentElement->createNewChildElement ("NETWORKEVENTS");
-    mainNode->setAttribute ("port", urlport);
+    uint16 currBoundPort = boundPort;
+    // save the actual bound port if any, otherwise the last attempted port.
+    mainNode->setAttribute ("port", currBoundPort ? currBoundPort : requestedPort.load());
 }
 
 
@@ -411,14 +396,13 @@ StringPairArray NetworkEvents::parseNetworkMessage(StringRef msg)
 }
 
 
-void NetworkEvents::updatePort(uint16 port)
+void NetworkEvents::updatePortString(uint16 port)
 {
-    urlport = port;
     auto ed = static_cast<NetworkEventsEditor*>(getEditor());
     if (ed)
     {
         const MessageManagerLock mmLock;
-        ed->setPortText(getPortString());
+        ed->setPortText(getPortString(port));
     }
 }
 
@@ -429,6 +413,21 @@ String NetworkEvents::getEndpoint(uint16 port)
 }
 
 
+String NetworkEvents::getPortString(uint16 port)
+{
+#ifdef ZEROMQ
+    if (port == 0)
+    {
+        return "<no cxn>";
+    }
+
+    return String(port);
+#else
+    return "<no zeromq>";
+#endif
+}
+
+
 /*** ZMQContext ***/
 
 NetworkEvents::ZMQContext::ZMQContext(const ScopedLock& lock)
@@ -467,9 +466,10 @@ void* NetworkEvents::ZMQContext::createSocket()
 const int NetworkEvents::Responder::RECV_TIMEOUT_MS = 100;
 
 NetworkEvents::Responder::Responder(uint16 port)
-    : socket(nullptr)
-    , boundPort(0)
-    , lastErrno(0)
+    : socket    (nullptr)
+    , valid     (false)
+    , boundPort (0)
+    , lastErrno (0)
 {
     {
         ScopedLock lock(sharedContextLock);
@@ -510,7 +510,6 @@ NetworkEvents::Responder::Responder(uint16 port)
     // if requested port was 0, find out which port was actually used
     if (port == 0)
     {
-        CoreServices::sendStatusMessage("NetworkEvents: Selecting port automatically");
         const size_t BUF_LEN = 32;
         size_t len = BUF_LEN;
         char endpoint[BUF_LEN];
@@ -521,34 +520,15 @@ NetworkEvents::Responder::Responder(uint16 port)
         }
 
         port = String(endpoint).getTrailingIntValue();
-        jassert(port > 0);
     }
 
+    jassert(port > 0);
+    valid = true;
     boundPort = port;
 #endif
 }
 
 
-ScopedPointer<NetworkEvents::Responder> NetworkEvents::Responder::makeResponder(uint16 port)
-{
-    ScopedPointer<Responder> socketPtr = new Responder(port);
-    uint16 boundPort = socketPtr->getBoundPort();
-    if (boundPort == 0)
-    {
-        socketPtr->reportErr("Failed to connect to port " + String(port));
-        return nullptr;
-    }
-
-    if (port != 0 && boundPort != port)
-    {
-        jassertfalse; // huh?
-        return nullptr;
-    }
-    return socketPtr;
-}
-
-
-
 NetworkEvents::Responder::~Responder()
 {
 #ifdef ZEROMQ
@@ -584,6 +564,12 @@ void NetworkEvents::Responder::reportErr(const String& message) const
 };
 
 
+bool NetworkEvents::Responder::isValid() const
+{
+    return valid;
+}
+
+
 uint16 NetworkEvents::Responder::getBoundPort() const
 {
     return boundPort;

Source/Plugins/NetworkEvents/NetworkEvents.h

@@ -39,7 +39,7 @@
 
 #include <list>
 #include <queue>
-
+#include <atomic>
 
 /**
  Sends incoming TCP/IP messages from 0MQ to the events buffer
@@ -76,11 +76,10 @@ class NetworkEvents : public GenericProcessor
     void run() override;
 
     // passing 0 corresponds to wildcard ("*") and picks any available port
-    // returns true on success, false on failure
-    bool setNewListeningPort (uint16 port);
+    void setNewListeningPort (uint16 port);
 
     // gets a string for the editor's port input to reflect current urlport
-    String getPortString() const;
+    String getCurrPortString() const;
 
     void restartConnection();
 
@@ -110,8 +109,9 @@ class NetworkEvents : public GenericProcessor
     class Responder
     {
     public:
-        // tries to create a responder and bind to given port; returns nullptr on failure.
-        static ScopedPointer<Responder> makeResponder(uint16 port);
+        // creates socket from given context and tries to bind to port.
+        // if port is 0, chooses an available ephemeral port.
+        Responder(uint16 port);
         ~Responder();
 
         // returns the latest errno value
@@ -120,7 +120,9 @@ class NetworkEvents : public GenericProcessor
         // output last error on stdout and status bar, including the passed message
         void reportErr(const String& message) const;
 
-        // returns the port if the socket was successfully bound to one.
+        bool isValid() const;
+
+        // returns the port if the socket was successfully bound to one, else 0
         // if not, or if the socket is invalid, returns 0.
         uint16 getBoundPort() const;
 
@@ -132,13 +134,10 @@ class NetworkEvents : public GenericProcessor
         int send(StringRef response);
 
     private:
-        // creates socket from given context and tries to bind to port.
-        // if port is 0, chooses an available ephemeral port.
-        Responder(uint16 port);
-
         ZMQContext::Ptr context;
         void* socket;
-        uint16 boundPort; // 0 indicates not bound
+        bool valid;
+        uint16 boundPort;
         int lastErrno;
 
         static const int RECV_TIMEOUT_MS;
@@ -153,28 +152,24 @@ class NetworkEvents : public GenericProcessor
     //* Split network message into name/value pairs (name1=val1 name2=val2 etc) */
     StringPairArray parseNetworkMessage(StringRef msg);
 
-    // updates urlport and the port input on the editor (0 indicates not connected)
-    void updatePort(uint16 port);
+    // updates urlport and the port input on the editor (< 0 indicates not connected)
+    void updatePortString(uint16 port);
 
     // get an endpoint url for the given port (using 0 to represent *)
     static String getEndpoint(uint16 port);
 
+    // get a representation of the given port for use on the editor
+    static String getPortString(uint16 port);
 
     // share a "dumb" pointer that doesn't take part in reference counting.
     // want the context to be terminated by the time the static members are
     // destroyed (see: https://github.com/zeromq/libzmq/issues/1708)
     static ZMQContext* sharedContext;
     static CriticalSection sharedContextLock;
 
-    // To switch ports, a new socket is created and (if successful) assigned to this pointer,
-    // and then the thread will switch to using this socket at the next opportunity.
-    ScopedPointer<Responder> nextResponder;
-    CriticalSection nextResponderLock;
-
-    bool restart;
-    bool changeResponder;
-
-    uint16 urlport;   // 0 indicates not connected
+    std::atomic<bool> makeNewSocket;   // port change or restart needed (depending on requestedPort)
+    std::atomic<uint16> requestedPort; // never set by the thread; 0 means any free port
+    std::atomic<uint16> boundPort;     // only set by the thread; 0 means no connection
 
     std::queue<StringTS> networkMessagesQueue;
     CriticalSection queueLock;