Merge pull request #278 from tne-lab/shared-resource-zmq

Use SharedResourcePointers for ZeroMQ contexts

Author: aacuevas

Source/Plugins/EventBroadcaster/EventBroadcaster.cpp

@@ -11,23 +11,16 @@
 #include "EventBroadcaster.h"
 #include "EventBroadcasterEditor.h"
 
-EventBroadcaster::ZMQContext* EventBroadcaster::sharedContext = nullptr;
-CriticalSection EventBroadcaster::sharedContextLock{};
-
-EventBroadcaster::ZMQContext::ZMQContext(const ScopedLock& lock)
+EventBroadcaster::ZMQContext::ZMQContext()
 #ifdef ZEROMQ
     : context(zmq_ctx_new())
 #endif
-{
-    sharedContext = this;
-}
+{}
 
 // ZMQContext is a ReferenceCountedObject with a pointer in each instance's 
 // socket pointer, so this only happens when the last instance is destroyed.
 EventBroadcaster::ZMQContext::~ZMQContext()
 {
-    ScopedLock lock(sharedContextLock);
-    sharedContext = nullptr;
 #ifdef ZEROMQ
     zmq_ctx_destroy(context);
 #endif
@@ -44,18 +37,6 @@ void* EventBroadcaster::ZMQContext::createZMQSocket()
 EventBroadcaster::ZMQSocketPtr::ZMQSocketPtr()
     : std::unique_ptr<void, decltype(&closeZMQSocket)>(nullptr, &closeZMQSocket)
 {
-    ScopedLock lock(sharedContextLock);
-    if (sharedContext == nullptr)
-    {
-        // first one, create the context
-        context = new ZMQContext(lock);
-    }
-    else
-    {
-        // use already-created context
-        context = sharedContext;
-    }
-
 #ifdef ZEROMQ
     reset(context->createZMQSocket());
 #endif

Source/Plugins/EventBroadcaster/EventBroadcaster.h

@@ -44,11 +44,11 @@ class EventBroadcaster : public GenericProcessor
 
 
 private:
-    class ZMQContext : public ReferenceCountedObject
+    class ZMQContext
     {
     public:
-        ZMQContext(const ScopedLock& lock);
-        ~ZMQContext() override;
+        ZMQContext();
+        ~ZMQContext();
         void* createZMQSocket();
     private:
         void* context;
@@ -62,7 +62,7 @@ class EventBroadcaster : public GenericProcessor
         ZMQSocketPtr();
         ~ZMQSocketPtr();
     private:
-        ReferenceCountedObjectPtr<ZMQContext> context;
+        SharedResourcePointer<ZMQContext> context;
     };
 
     int unbindZMQSocket();
@@ -73,11 +73,6 @@ class EventBroadcaster : public GenericProcessor
     // called from getListeningPort() depending on success/failure of ZMQ operations
     void reportActualListeningPort(int 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;
     ZMQSocketPtr zmqSocket;
     int listeningPort;
 };

Source/Plugins/NetworkEvents/NetworkEvents.cpp

@@ -35,9 +35,6 @@ const int MAX_MESSAGE_LENGTH = 64000;
     #include <unistd.h>
 #endif
 
-NetworkEvents::ZMQContext* NetworkEvents::sharedContext = nullptr;
-CriticalSection NetworkEvents::sharedContextLock{};
-
 NetworkEvents::NetworkEvents()
     : GenericProcessor  ("Network Events")
     , Thread            ("NetworkThread")
@@ -79,6 +76,7 @@ String NetworkEvents::getCurrPortString() const
 
 void NetworkEvents::restartConnection()
 {
+    requestedPort = boundPort.load();
     makeNewSocket = true;
 }
 
@@ -430,21 +428,15 @@ String NetworkEvents::getPortString(uint16 port)
 
 /*** ZMQContext ***/
 
-NetworkEvents::ZMQContext::ZMQContext(const ScopedLock& lock)
+NetworkEvents::ZMQContext::ZMQContext()
 #ifdef ZEROMQ
     : context(zmq_ctx_new())
 #endif
-{
-    // sharedContextLock should already be held here
-    sharedContext = this;
-}
+{}
 
-// ZMQContext is a ReferenceCountedObject with a pointer in each instance's 
-// socket pointer, so this only happens when the last instance is destroyed.
+// only happens when the last SharedResourcePointer is destroyed.
 NetworkEvents::ZMQContext::~ZMQContext()
 {
-    ScopedLock lock(sharedContextLock);
-    sharedContext = nullptr;
 #ifdef ZEROMQ
     zmq_ctx_destroy(context);
 #endif
@@ -471,20 +463,6 @@ NetworkEvents::Responder::Responder(uint16 port)
     , boundPort (0)
     , lastErrno (0)
 {
-    {
-        ScopedLock lock(sharedContextLock);
-        if (sharedContext == nullptr)
-        {
-            // first one, create the context
-            context = new ZMQContext(lock);
-        }
-        else
-        {
-            // use already-created context
-            context = sharedContext;
-        }
-    }
-
 #ifdef ZEROMQ
     socket = context->createSocket();
     if (!socket)

Source/Plugins/NetworkEvents/NetworkEvents.h

@@ -90,15 +90,12 @@ class NetworkEvents : public GenericProcessor
         int64 timestamp;
     };
 
-    class ZMQContext : public ReferenceCountedObject
+    class ZMQContext
     {
     public:
-        ZMQContext(const ScopedLock& lock);
-        ~ZMQContext() override;
+        ZMQContext();
+        ~ZMQContext();
         void* createSocket();
-
-        typedef ReferenceCountedObjectPtr<ZMQContext> Ptr;
-
     private:
         void* context;
 
@@ -134,7 +131,7 @@ class NetworkEvents : public GenericProcessor
         int send(StringRef response);
 
     private:
-        ZMQContext::Ptr context;
+        SharedResourcePointer<ZMQContext> context;
         void* socket;
         bool valid;
         uint16 boundPort;
@@ -161,12 +158,6 @@ class NetworkEvents : public GenericProcessor
     // 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;
-
     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