Destroy ZMQ context before program exits

Author: ethanbb

Source/Plugins/EventBroadcaster/EventBroadcaster.cpp

@@ -11,15 +11,60 @@
 #include "EventBroadcaster.h"
 #include "EventBroadcasterEditor.h"
 
-std::shared_ptr<void> EventBroadcaster::getZMQContext() {
-    // Note: C++11 guarantees that initialization of static local variables occurs exactly once, even
-    // if multiple threads attempt to initialize the same static local variable concurrently.
+EventBroadcaster::ZMQContext* EventBroadcaster::sharedContext = nullptr;
+CriticalSection EventBroadcaster::sharedContextLock{};
+
+EventBroadcaster::ZMQContext::ZMQContext(const ScopedLock& lock)
 #ifdef ZEROMQ
-    static const std::shared_ptr<void> ctx(zmq_ctx_new(), zmq_ctx_destroy);
-#else
-    static const std::shared_ptr<void> ctx;
+    : 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
+}
+
+void* EventBroadcaster::ZMQContext::createZMQSocket()
+{
+#ifdef ZEROMQ
+    jassert(context != nullptr);
+    return zmq_socket(context, ZMQ_PUB);
 #endif
-    return ctx;
+}
+
+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
+}
+
+EventBroadcaster::ZMQSocketPtr::~ZMQSocketPtr()
+{
+    // close the socket before the context might get destroyed.
+    reset(nullptr);
 }
 
 int EventBroadcaster::unbindZMQSocket()
@@ -70,8 +115,6 @@ void EventBroadcaster::reportActualListeningPort(int port)
 
 EventBroadcaster::EventBroadcaster()
     : GenericProcessor  ("Event Broadcaster")
-    , zmqContext        (getZMQContext())
-    , zmqSocket         (nullptr)
     , listeningPort     (0)
 {
     setProcessorType (PROCESSOR_TYPE_SINK);
@@ -105,7 +148,7 @@ int EventBroadcaster::setListeningPort(int port, bool forceRestart)
 #ifdef ZEROMQ
         // unbind current socket (if any) to free up port
         unbindZMQSocket();
-        zmqSocketPtr newSocket(zmq_socket(zmqContext.get(), ZMQ_PUB));
+        ZMQSocketPtr newSocket;
         auto editor = static_cast<EventBroadcasterEditor*>(getEditor());
         int status = 0;
 

Source/Plugins/EventBroadcaster/EventBroadcaster.h

@@ -24,7 +24,6 @@
 
 #include <memory>
 
-
 class EventBroadcaster : public GenericProcessor
 {
 public:
@@ -45,27 +44,42 @@ class EventBroadcaster : public GenericProcessor
 
 
 private:
-	void sendEvent(const MidiMessage& event, float eventSampleRate) const;
-    static std::shared_ptr<void> getZMQContext();
+    class ZMQContext : public ReferenceCountedObject
+    {
+    public:
+        ZMQContext(const ScopedLock& lock);
+        ~ZMQContext() override;
+        void* createZMQSocket();
+    private:
+        void* context;
+    };
+
+    static void closeZMQSocket(void* socket);
+
+    class ZMQSocketPtr : public std::unique_ptr<void, decltype(&closeZMQSocket)>
+    {
+    public:
+        ZMQSocketPtr();
+        ~ZMQSocketPtr();
+    private:
+        ReferenceCountedObjectPtr<ZMQContext> context;
+    };
+    
     int unbindZMQSocket();
     int rebindZMQSocket();
-    static void closeZMQSocket (void* socket);
+
+	void sendEvent(const MidiMessage& event, float eventSampleRate) const;
     static String getEndpoint(int port);
     // called from getListeningPort() depending on success/failure of ZMQ operations
     void reportActualListeningPort(int port);
 
-    // encapuslate closing sockets when their pointers go out of scope
-    struct zmqSocketPtr : public std::unique_ptr<void, decltype(&closeZMQSocket)>
-    {
-        zmqSocketPtr(void* ptr) 
-            : std::unique_ptr<void, decltype(&closeZMQSocket)>(ptr, &closeZMQSocket)
-        {}
-    };
-
-    const std::shared_ptr<void> zmqContext;
-    zmqSocketPtr zmqSocket;
+    // 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;
-
 };