1 files changed, 114 insertions, 135 deletions

diff --git a/src/ctx.cpp b/src/ctx.cpp
index 397f692..91157a5 100644
--- a/src/ctx.cpp
+++ b/src/ctx.cpp
@@ -24,7 +24,6 @@
 
 #include "ctx.hpp"
 #include "socket_base.hpp"
-#include "app_thread.hpp"
 #include "io_thread.hpp"
 #include "platform.hpp"
 #include "err.hpp"
@@ -32,11 +31,12 @@
 
 #if defined ZMQ_HAVE_WINDOWS
 #include "windows.h"
+#else
+#include "unistd.h"
 #endif
 
 zmq::ctx_t::ctx_t (uint32_t io_threads_) :
-    sockets (0),
-    terminated (false)
+    no_sockets_notify (false)
 {
 #ifdef ZMQ_HAVE_WINDOWS
     //  Intialise Windows sockets. Note that WSAStartup can be called multiple
@@ -50,44 +50,32 @@ zmq::ctx_t::ctx_t (uint32_t io_threads_) :
 #endif
 
     //  Initialise the array of signalers.
-    signalers_count = max_app_threads + io_threads_;
-    signalers = (signaler_t**) malloc (sizeof (signaler_t*) * signalers_count);
-    zmq_assert (signalers);
-    memset (signalers, 0, sizeof (signaler_t*) * signalers_count);
+    slot_count = max_sockets + io_threads_;
+    slots = (signaler_t**) malloc (sizeof (signaler_t*) * slot_count);
+    zmq_assert (slots);
 
     //  Create I/O thread objects and launch them.
     for (uint32_t i = 0; i != io_threads_; i++) {
         io_thread_t *io_thread = new (std::nothrow) io_thread_t (this, i);
         zmq_assert (io_thread);
         io_threads.push_back (io_thread);
-        signalers [i] = io_thread->get_signaler ();
+        slots [i] = io_thread->get_signaler ();
         io_thread->start ();
     }
-}
-
-int zmq::ctx_t::term ()
-{
-    //  First send stop command to application threads so that any
-    //  blocking calls are interrupted.
-    for (app_threads_t::size_type i = 0; i != app_threads.size (); i++)
-        app_threads [i].app_thread->stop ();
-
-    //  Then mark context as terminated.
-    term_sync.lock ();
-    zmq_assert (!terminated);
-    terminated = true;
-    bool destroy = (sockets == 0);
-    term_sync.unlock ();
-    
-    //  If there are no sockets open, destroy the context immediately.
-    if (destroy)
-        delete this;
 
-    return 0;
+    //  In the unused part of the slot array, create a list of empty slots.
+    for (uint32_t i = slot_count - 1; i >= io_threads_; i--) {
+        empty_slots.push_back (i);
+        slots [i] = NULL;
+    }
 }
 
 zmq::ctx_t::~ctx_t ()
 {
+    //  Check that there are no remaining open or zombie sockets.
+    zmq_assert (sockets.empty ());
+    zmq_assert (zombies.empty ());
+
     //  Ask I/O threads to terminate. If stop signal wasn't sent to I/O
     //  thread subsequent invocation of destructor would hang-up.
     for (io_threads_t::size_type i = 0; i != io_threads.size (); i++)
@@ -97,18 +85,10 @@ zmq::ctx_t::~ctx_t ()
     for (io_threads_t::size_type i = 0; i != io_threads.size (); i++)
         delete io_threads [i];
 
-    //  Close all application theads, sockets, io_objects etc.
-    for (app_threads_t::size_type i = 0; i != app_threads.size (); i++)
-        delete app_threads [i].app_thread;
-
-    //  Deallocate all the orphaned pipes.
-    while (!pipes.empty ())
-        delete *pipes.begin ();
-
-    //  Deallocate the array of pointers to signalers. No special work is
+    //  Deallocate the array of slot. No special work is
     //  needed as signalers themselves were deallocated with their
-    //  corresponding (app_/io_) thread objects.
-    free (signalers);
+    //  corresponding io_thread/socket objects.
+    free (slots);
     
 #ifdef ZMQ_HAVE_WINDOWS
     //  On Windows, uninitialise socket layer.
@@ -117,110 +97,113 @@ zmq::ctx_t::~ctx_t ()
 #endif
 }
 
-zmq::socket_base_t *zmq::ctx_t::create_socket (int type_)
+int zmq::ctx_t::term ()
 {
-    app_threads_sync.lock ();
-
-    //  Find whether the calling thread has app_thread_t object associated
-    //  already. At the same time find an unused app_thread_t so that it can
-    //  be used if there's no associated object for the calling thread.
-    //  Check whether thread ID is already assigned. If so, return it.
-    app_threads_t::size_type unused = app_threads.size ();
-    app_threads_t::size_type current;
-    for (current = 0; current != app_threads.size (); current++) {
-        if (app_threads [current].associated &&
-              thread_t::equal (thread_t::id (), app_threads [current].tid))
-            break;
-        if (!app_threads [current].associated)
-            unused = current;
+    //  First send stop command to sockets so that any
+    //  blocking calls are interrupted.
+    for (sockets_t::size_type i = 0; i != sockets.size (); i++)
+        sockets [i]->stop ();
+
+    //  Find out whether there are any open sockets to care about.
+    //  If so, sleep till they are closed. Note that we can use
+    //  no_sockets_notify safely out of the critical section as once set
+    //  its value is never changed again.
+    slot_sync.lock ();
+    if (!sockets.empty ())
+        no_sockets_notify = true;
+    slot_sync.unlock ();
+    if (no_sockets_notify)
+        no_sockets_sync.wait ();
+
+    //  At this point there's only one application thread (this one) remaining.
+    //  We don't even have to synchronise access to data.
+    zmq_assert (sockets.empty ());
+
+    //  Get rid of remaining zombie sockets.
+    while (!zombies.empty ()) {
+        dezombify ();
+
+        //  Sleep for 1ms not to end up busy-looping in the case the I/O threads
+        //  are still busy sending data. We can possibly add a grand poll here
+        //  (polling for fds associated with all the zombie sockets), but it's
+        //  probably not worth of implementing it.
+#if defined ZMQ_HAVE_WINDOWS
+        Sleep (1);
+#else
+        usleep (1000);
+#endif
     }
 
-    //  If no app_thread_t is associated with the calling thread,
-    //  associate it with one of the unused app_thread_t objects.
-    if (current == app_threads.size ()) {
+    //  Deallocate the resources.
+    delete this;
 
-        //  If all the existing app_threads are already used, create one more.
-        if (unused == app_threads.size ()) {
-
-            //  If max_app_threads limit was reached, return error.
-            if (app_threads.size () == max_app_threads) {
-                app_threads_sync.unlock ();
-                errno = EMTHREAD;
-                return NULL;
-            }
+    return 0;
+}
 
-            //  Create the new application thread proxy object.
-            app_thread_info_t info;
-            memset (&info, 0, sizeof (info));
-            info.associated = false;
-            info.app_thread = new (std::nothrow) app_thread_t (this,
-                io_threads.size () + app_threads.size ());
-            zmq_assert (info.app_thread);
-            signalers [io_threads.size () + app_threads.size ()] =
-                info.app_thread->get_signaler ();
-            app_threads.push_back (info);
-        }
+zmq::socket_base_t *zmq::ctx_t::create_socket (int type_)
+{
+    slot_sync.lock ();
 
-        //  Incidentally, this works both when there is an unused app_thread
-        //  and when a new one is created.
-        current = unused;
+    //  Free the slots, if possible.
+    dezombify ();
 
-        //  Associate the selected app_thread with the OS thread.
-        app_threads [current].associated = true;
-        app_threads [current].tid = thread_t::id ();
+    //  If max_sockets limit was reached, return error.
+    if (empty_slots.empty ()) {
+        slot_sync.unlock ();
+        errno = EMFILE;
+        return NULL;
     }
 
-    app_thread_t *thread = app_threads [current].app_thread;
-    app_threads_sync.unlock ();
+    //  Choose a slot for the socket.
+    uint32_t slot = empty_slots.back ();
+    empty_slots.pop_back ();
 
-    socket_base_t *s = thread->create_socket (type_);
-    if (!s)
+    //  Create the socket and register its signaler.
+    socket_base_t *s = socket_base_t::create (type_, this, slot);
+    if (!s) {
+        empty_slots.push_back (slot);
+        slot_sync.unlock ();
         return NULL;
+    }
+    sockets.push_back (s);
+    slots [slot] = s->get_signaler ();
 
-    term_sync.lock ();
-    sockets++;
-    term_sync.unlock ();
+    slot_sync.unlock ();
 
     return s;
 }
 
-void zmq::ctx_t::destroy_socket ()
+void zmq::ctx_t::zombify (socket_base_t *socket_)
 {
-    //  If zmq_term was already called and there are no more sockets,
-    //  terminate the whole 0MQ infrastructure.
-    term_sync.lock ();
-    zmq_assert (sockets > 0);
-    sockets--;
-    bool destroy = (sockets == 0 && terminated);
-    term_sync.unlock ();
-
-    if (destroy)
-       delete this;
-}
+    //  Zombification of socket basically means that its ownership is tranferred
+    //  from the application that created it to the context.
 
-void zmq::ctx_t::no_sockets (app_thread_t *thread_)
-{
-    app_threads_sync.lock ();
-    app_threads_t::size_type i;
-    for (i = 0; i != app_threads.size (); i++)
-        if (app_threads [i].app_thread == thread_) {
-            app_threads [i].associated = false;
-            break;
-        }
-    zmq_assert (i != app_threads.size ());
-    app_threads_sync.unlock ();
+    //  Note that the lock provides the memory barrier needed to migrate
+    //  zombie-to-be socket from it's native thread to shared data area
+    //  synchronised by slot_sync.
+    slot_sync.lock ();
+    sockets.erase (socket_);
+    zombies.push_back (socket_);
+
+    //  Try to get rid of at least some zombie sockets at this point.
+    dezombify ();
+
+    //  If shutdown thread is interested in notification about no more
+    //  open sockets, notify it now.
+    if (sockets.empty () && no_sockets_notify)
+        no_sockets_sync.post ();
+
+    slot_sync.unlock ();
 }
 
-void zmq::ctx_t::send_command (uint32_t destination_,
-    const command_t &command_)
+void zmq::ctx_t::send_command (uint32_t slot_, const command_t &command_)
 {
-    signalers [destination_]->send (command_);
+    slots [slot_]->send (command_);
 }
 
-bool zmq::ctx_t::recv_command (uint32_t thread_slot_,
-    command_t *command_, bool block_)
+bool zmq::ctx_t::recv_command (uint32_t slot_, command_t *command_, bool block_)
 {
-    return signalers [thread_slot_]->recv (command_, block_);
+    return slots [slot_]->recv (command_, block_);
 }
 
 zmq::io_thread_t *zmq::ctx_t::choose_io_thread (uint64_t affinity_)
@@ -242,22 +225,6 @@ zmq::io_thread_t *zmq::ctx_t::choose_io_thread (uint64_t affinity_)
     return io_threads [result];
 }
 
-void zmq::ctx_t::register_pipe (class pipe_t *pipe_)
-{
-    pipes_sync.lock ();
-    bool inserted = pipes.insert (pipe_).second;
-    zmq_assert (inserted);
-    pipes_sync.unlock ();
-}
-
-void zmq::ctx_t::unregister_pipe (class pipe_t *pipe_)
-{
-    pipes_sync.lock ();
-    pipes_t::size_type erased = pipes.erase (pipe_);
-    zmq_assert (erased == 1);
-    pipes_sync.unlock ();
-}
-
 int zmq::ctx_t::register_endpoint (const char *addr_,
     socket_base_t *socket_)
 {
@@ -315,3 +282,15 @@ zmq::socket_base_t *zmq::ctx_t::find_endpoint (const char *addr_)
      return endpoint;
 }
 
+void zmq::ctx_t::dezombify ()
+{
+    //  Try to dezombify each zombie in the list.
+    for (zombies_t::size_type i = 0; i != zombies.size ();)
+        if (zombies [i]->dezombify ()) {
+            empty_slots.push_back (zombies [i]->get_slot ());
+            zombies.erase (zombies [i]);
+        }
+        else
+            i++;
+}
+