From 05d908492dc382941fc633ad7082b5bd86e84e67 Mon Sep 17 00:00:00 2001 From: Martin Sustrik Date: Fri, 6 Aug 2010 17:49:37 +0200 Subject: WIP: Socket migration between threads, new zmq_close() semantics Sockets may now be migrated between OS threads; sockets may not be used by more than one thread at any time. To migrate a socket to another thread the caller must ensure that a full memory barrier is called before using the socket from the target thread. The new zmq_close() semantics implement the behaviour discussed at: http://lists.zeromq.org/pipermail/zeromq-dev/2010-July/004244.html Specifically, zmq_close() is now deterministic and while it still returns immediately, it does not discard any data that may still be queued for sending. Further, zmq_term() will now block until all outstanding data has been sent. TODO: Many bugs have been introduced, needs testing. Further, SO_LINGER or an equivalent mechanism (possibly a configurable timeout to zmq_term()) needs to be implemented. --- src/Makefile.am | 15 +- src/app_thread.cpp | 195 --------------------- src/app_thread.hpp | 88 ---------- src/config.hpp | 5 +- src/ctx.cpp | 249 ++++++++++++-------------- src/ctx.hpp | 94 ++++------ src/fq.cpp | 29 +-- src/fq.hpp | 17 +- src/i_endpoint.hpp | 43 ----- src/io_thread.cpp | 5 +- src/io_thread.hpp | 2 +- src/lb.cpp | 21 ++- src/lb.hpp | 13 +- src/object.cpp | 29 ++- src/object.hpp | 14 +- src/owned.cpp | 10 +- src/owned.hpp | 9 +- src/pair.cpp | 72 ++++---- src/pair.hpp | 26 ++- src/pipe.cpp | 250 +++++++++++++++----------- src/pipe.hpp | 111 +++++++----- src/pub.cpp | 62 +++---- src/pub.hpp | 21 ++- src/pull.cpp | 48 +---- src/pull.hpp | 16 +- src/push.cpp | 50 +----- src/push.hpp | 16 +- src/rep.cpp | 81 ++++++--- src/rep.hpp | 30 ++-- src/req.cpp | 65 ++++--- src/req.hpp | 32 ++-- src/semaphore.hpp | 135 ++++++++++++++ src/session.cpp | 62 +++---- src/session.hpp | 26 ++- src/socket_base.cpp | 495 +++++++++++++++++++++++++++++++++++++--------------- src/socket_base.hpp | 100 +++++++---- src/sub.cpp | 40 +---- src/sub.hpp | 11 +- src/thread.cpp | 20 --- src/thread.hpp | 9 - src/xrep.cpp | 65 +++---- src/xrep.hpp | 26 ++- src/xreq.cpp | 37 +--- src/xreq.hpp | 10 +- src/zmq.cpp | 7 +- src/zmq_encoder.cpp | 2 +- 46 files changed, 1392 insertions(+), 1371 deletions(-) delete mode 100644 src/app_thread.cpp delete mode 100644 src/app_thread.hpp delete mode 100644 src/i_endpoint.hpp create mode 100644 src/semaphore.hpp (limited to 'src') diff --git a/src/Makefile.am b/src/Makefile.am index 19a80d0..937372f 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -49,7 +49,7 @@ endif nodist_libzmq_la_SOURCES = $(pgm_sources) -libzmq_la_SOURCES = app_thread.hpp \ +libzmq_la_SOURCES = \ atomic_counter.hpp \ atomic_ptr.hpp \ blob.hpp \ @@ -58,7 +58,6 @@ libzmq_la_SOURCES = app_thread.hpp \ ctx.hpp \ decoder.hpp \ devpoll.hpp \ - push.hpp \ encoder.hpp \ epoll.hpp \ err.hpp \ @@ -69,7 +68,6 @@ libzmq_la_SOURCES = app_thread.hpp \ io_object.hpp \ io_thread.hpp \ ip.hpp \ - i_endpoint.hpp \ i_engine.hpp \ i_poll_events.hpp \ kqueue.hpp \ @@ -91,10 +89,13 @@ libzmq_la_SOURCES = app_thread.hpp \ pair.hpp \ prefix_tree.hpp \ pub.hpp \ + pull.hpp \ + push.hpp \ queue.hpp \ rep.hpp \ req.hpp \ select.hpp \ + semaphore.hpp \ session.hpp \ signaler.hpp \ socket_base.hpp \ @@ -105,7 +106,6 @@ libzmq_la_SOURCES = app_thread.hpp \ tcp_listener.hpp \ tcp_socket.hpp \ thread.hpp \ - pull.hpp \ uuid.hpp \ windows.hpp \ wire.hpp \ @@ -121,11 +121,9 @@ libzmq_la_SOURCES = app_thread.hpp \ zmq_engine.hpp \ zmq_init.hpp \ zmq_listener.hpp \ - app_thread.cpp \ command.cpp \ ctx.cpp \ devpoll.cpp \ - push.cpp \ epoll.cpp \ err.cpp \ forwarder.cpp \ @@ -139,13 +137,15 @@ libzmq_la_SOURCES = app_thread.hpp \ object.cpp \ options.cpp \ owned.cpp \ + pair.cpp \ pgm_receiver.cpp \ pgm_sender.cpp \ pgm_socket.cpp \ - pair.cpp \ prefix_tree.cpp \ pipe.cpp \ poll.cpp \ + pull.cpp \ + push.cpp \ pub.cpp \ queue.cpp \ rep.cpp \ @@ -160,7 +160,6 @@ libzmq_la_SOURCES = app_thread.hpp \ tcp_listener.cpp \ tcp_socket.cpp \ thread.cpp \ - pull.cpp \ uuid.cpp \ xrep.cpp \ xreq.cpp \ diff --git a/src/app_thread.cpp b/src/app_thread.cpp deleted file mode 100644 index ac59464..0000000 --- a/src/app_thread.cpp +++ /dev/null @@ -1,195 +0,0 @@ -/* - Copyright (c) 2007-2010 iMatix Corporation - - This file is part of 0MQ. - - 0MQ is free software; you can redistribute it and/or modify it under - the terms of the Lesser GNU General Public License as published by - the Free Software Foundation; either version 3 of the License, or - (at your option) any later version. - - 0MQ is distributed in the hope that it will be useful, - but WITHOUT ANY WARRANTY; without even the implied warranty of - MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - Lesser GNU General Public License for more details. - - You should have received a copy of the Lesser GNU General Public License - along with this program. If not, see . -*/ - -#include -#include - -#include "../include/zmq.h" - -#include "platform.hpp" - -#if defined ZMQ_HAVE_WINDOWS -#include "windows.hpp" -#if defined _MSC_VER -#include -#endif -#else -#include -#endif - -#include "app_thread.hpp" -#include "ctx.hpp" -#include "err.hpp" -#include "pipe.hpp" -#include "config.hpp" -#include "socket_base.hpp" -#include "pair.hpp" -#include "pub.hpp" -#include "sub.hpp" -#include "req.hpp" -#include "rep.hpp" -#include "xreq.hpp" -#include "xrep.hpp" -#include "pull.hpp" -#include "push.hpp" - -// If the RDTSC is available we use it to prevent excessive -// polling for commands. The nice thing here is that it will work on any -// system with x86 architecture and gcc or MSVC compiler. -#if (defined __GNUC__ && (defined __i386__ || defined __x86_64__)) ||\ - (defined _MSC_VER && (defined _M_IX86 || defined _M_X64)) -#define ZMQ_DELAY_COMMANDS -#endif - -zmq::app_thread_t::app_thread_t (ctx_t *ctx_, - uint32_t thread_slot_) : - object_t (ctx_, thread_slot_), - last_processing_time (0), - terminated (false) -{ -} - -zmq::app_thread_t::~app_thread_t () -{ - zmq_assert (sockets.empty ()); -} - -void zmq::app_thread_t::stop () -{ - send_stop (); -} - -zmq::signaler_t *zmq::app_thread_t::get_signaler () -{ - return &signaler; -} - -bool zmq::app_thread_t::process_commands (bool block_, bool throttle_) -{ - bool received; - command_t cmd; - if (block_) { - received = signaler.recv (&cmd, true); - zmq_assert (received); - } - else { - -#if defined ZMQ_DELAY_COMMANDS - // Optimised version of command processing - it doesn't have to check - // for incoming commands each time. It does so only if certain time - // elapsed since last command processing. Command delay varies - // depending on CPU speed: It's ~1ms on 3GHz CPU, ~2ms on 1.5GHz CPU - // etc. The optimisation makes sense only on platforms where getting - // a timestamp is a very cheap operation (tens of nanoseconds). - if (throttle_) { - - // Get timestamp counter. -#if defined __GNUC__ - uint32_t low; - uint32_t high; - __asm__ volatile ("rdtsc" : "=a" (low), "=d" (high)); - uint64_t current_time = (uint64_t) high << 32 | low; -#elif defined _MSC_VER - uint64_t current_time = __rdtsc (); -#else -#error -#endif - - // Check whether certain time have elapsed since last command - // processing. - if (current_time - last_processing_time <= max_command_delay) - return !terminated; - last_processing_time = current_time; - } -#endif - - // Check whether there are any commands pending for this thread. - received = signaler.recv (&cmd, false); - } - - // Process all the commands available at the moment. - while (received) { - cmd.destination->process_command (cmd); - received = signaler.recv (&cmd, false); - } - - return !terminated; -} - -zmq::socket_base_t *zmq::app_thread_t::create_socket (int type_) -{ - socket_base_t *s = NULL; - switch (type_) { - case ZMQ_PAIR: - s = new (std::nothrow) pair_t (this); - break; - case ZMQ_PUB: - s = new (std::nothrow) pub_t (this); - break; - case ZMQ_SUB: - s = new (std::nothrow) sub_t (this); - break; - case ZMQ_REQ: - s = new (std::nothrow) req_t (this); - break; - case ZMQ_REP: - s = new (std::nothrow) rep_t (this); - break; - case ZMQ_XREQ: - s = new (std::nothrow) xreq_t (this); - break; - case ZMQ_XREP: - s = new (std::nothrow) xrep_t (this); - break; - case ZMQ_PULL: - s = new (std::nothrow) pull_t (this); - break; - case ZMQ_PUSH: - s = new (std::nothrow) push_t (this); - break; - default: - if (sockets.empty ()) - get_ctx ()->no_sockets (this); - errno = EINVAL; - return NULL; - } - zmq_assert (s); - - sockets.push_back (s); - - return s; -} - -void zmq::app_thread_t::remove_socket (socket_base_t *socket_) -{ - sockets.erase (socket_); - if (sockets.empty ()) - get_ctx ()->no_sockets (this); -} - -void zmq::app_thread_t::process_stop () -{ - terminated = true; -} - -bool zmq::app_thread_t::is_terminated () -{ - return terminated; -} - diff --git a/src/app_thread.hpp b/src/app_thread.hpp deleted file mode 100644 index f0deaab..0000000 --- a/src/app_thread.hpp +++ /dev/null @@ -1,88 +0,0 @@ -/* - Copyright (c) 2007-2010 iMatix Corporation - - This file is part of 0MQ. - - 0MQ is free software; you can redistribute it and/or modify it under - the terms of the Lesser GNU General Public License as published by - the Free Software Foundation; either version 3 of the License, or - (at your option) any later version. - - 0MQ is distributed in the hope that it will be useful, - but WITHOUT ANY WARRANTY; without even the implied warranty of - MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - Lesser GNU General Public License for more details. - - You should have received a copy of the Lesser GNU General Public License - along with this program. If not, see . -*/ - -#ifndef __ZMQ_APP_THREAD_HPP_INCLUDED__ -#define __ZMQ_APP_THREAD_HPP_INCLUDED__ - -#include - -#include "stdint.hpp" -#include "object.hpp" -#include "yarray.hpp" -#include "signaler.hpp" - -namespace zmq -{ - - class app_thread_t : public object_t - { - public: - - app_thread_t (class ctx_t *ctx_, uint32_t thread_slot_); - - ~app_thread_t (); - - // Interrupt blocking call if the app thread is stuck in one. - // This function is is called from a different thread! - void stop (); - - // Returns signaler associated with this application thread. - signaler_t *get_signaler (); - - // Processes commands sent to this thread (if any). If 'block' is - // set to true, returns only after at least one command was processed. - // If throttle argument is true, commands are processed at most once - // in a predefined time period. The function returns false is the - // associated context was terminated, true otherwise. - bool process_commands (bool block_, bool throttle_); - - // Create a socket of a specified type. - class socket_base_t *create_socket (int type_); - - // Unregister the socket from the app_thread (called by socket itself). - void remove_socket (class socket_base_t *socket_); - - // Returns true is the associated context was already terminated. - bool is_terminated (); - - private: - - // Command handlers. - void process_stop (); - - // All the sockets created from this application thread. - typedef yarray_t sockets_t; - sockets_t sockets; - - // App thread's signaler object. - signaler_t signaler; - - // Timestamp of when commands were processed the last time. - uint64_t last_processing_time; - - // If true, 'stop' command was already received. - bool terminated; - - app_thread_t (const app_thread_t&); - void operator = (const app_thread_t&); - }; - -} - -#endif diff --git a/src/config.hpp b/src/config.hpp index 2c0ac2d..83e612a 100644 --- a/src/config.hpp +++ b/src/config.hpp @@ -27,9 +27,8 @@ namespace zmq enum { - // Maximal number of OS threads that can own 0MQ sockets - // at the same time. - max_app_threads = 512, + // Maximum number of sockets that can be opened at the same time. + max_sockets = 512, // Number of new messages in message pipe needed to trigger new memory // allocation. Setting this parameter to 256 decreases the impact of 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++; +} + diff --git a/src/ctx.hpp b/src/ctx.hpp index c96a923..cb9a2d9 100644 --- a/src/ctx.hpp +++ b/src/ctx.hpp @@ -26,7 +26,9 @@ #include #include "signaler.hpp" +#include "semaphore.hpp" #include "ypipe.hpp" +#include "yarray.hpp" #include "config.hpp" #include "mutex.hpp" #include "stdint.hpp" @@ -55,29 +57,19 @@ namespace zmq // Create a socket. class socket_base_t *create_socket (int type_); - // Destroy a socket. - void destroy_socket (); + // Make socket a zombie. + void zombify (socket_base_t *socket_); - // Called by app_thread_t when it has no more sockets. The function - // should disassociate the object from the current OS thread. - void no_sockets (class app_thread_t *thread_); + // Send command to the destination slot. + void send_command (uint32_t slot_, const command_t &command_); - // Send command to the destination thread. - void send_command (uint32_t destination_, const command_t &command_); - - // Receive command from another thread. - bool recv_command (uint32_t thread_slot_, command_t *command_, - bool block_); + // Receive command from the source slot. + bool recv_command (uint32_t slot_, command_t *command_, bool block_); // Returns the I/O thread that is the least busy at the moment. // Taskset specifies which I/O threads are eligible (0 = all). class io_thread_t *choose_io_thread (uint64_t taskset_); - // All pipes are registered with the context so that even the - // orphaned pipes can be deallocated on the terminal shutdown. - void register_pipe (class pipe_t *pipe_); - void unregister_pipe (class pipe_t *pipe_); - // Management of inproc endpoints. int register_endpoint (const char *addr_, class socket_base_t *socket_); void unregister_endpoints (class socket_base_t *socket_); @@ -87,57 +79,45 @@ namespace zmq ~ctx_t (); - struct app_thread_info_t - { - // If false, 0MQ application thread is free, there's no associated - // OS thread. - bool associated; + // Sockets belonging to this context. + typedef yarray_t sockets_t; + sockets_t sockets; + + // Array of sockets that were already closed but not yet deallocated. + // These sockets still have some pipes and I/O objects attached. + typedef yarray_t zombies_t; + zombies_t zombies; + + // List of unused slots. + typedef std::vector emtpy_slots_t; + emtpy_slots_t empty_slots; - // ID of the associated OS thread. If 'associated' is false, - // this field contains bogus data. - thread_t::id_t tid; + // If true, shutdown thread wants to be informed when there are no + // more open sockets. Do so by posting no_sockets_sync semaphore. + // Note that this variable is synchronised by slot_sync mutex. + bool no_sockets_notify; - // Pointer to the 0MQ application thread object. - class app_thread_t *app_thread; - }; + // Object used by zmq_term to wait while all the sockets are closed + // by different application threads. + semaphore_t no_sockets_sync; - // Application threads. - typedef std::vector app_threads_t; - app_threads_t app_threads; + // Synchronisation of accesses to global slot-related data: + // sockets, zombies, empty_slots, terminated. It also synchronises + // access to zombie sockets as such (as oposed to slots) and provides + // a memory barrier to ensure that all CPU cores see the same data. + mutex_t slot_sync; - // Synchronisation of accesses to shared application thread data. - mutex_t app_threads_sync; + // This function attempts to deallocate as many zombie sockets as + // possible. It must be called within a slot_sync critical section. + void dezombify (); // I/O threads. typedef std::vector io_threads_t; io_threads_t io_threads; // Array of pointers to signalers for both application and I/O threads. - int signalers_count; - signaler_t **signalers; - - // As pipes may reside in orphaned state in particular moments - // of the pipe shutdown process, i.e. neither pipe reader nor - // pipe writer hold reference to the pipe, we have to hold references - // to all pipes in context so that we can deallocate them - // during terminal shutdown even though it conincides with the - // pipe being in the orphaned state. - typedef std::set pipes_t; - pipes_t pipes; - - // Synchronisation of access to the pipes repository. - mutex_t pipes_sync; - - // Number of sockets alive. - int sockets; - - // If true, zmq_term was already called. When last socket is closed - // the whole 0MQ infrastructure should be deallocated. - bool terminated; - - // Synchronisation of access to the termination data (socket count - // and 'terminated' flag). - mutex_t term_sync; + uint32_t slot_count; + signaler_t **slots; // List of inproc endpoints within this context. typedef std::map endpoints_t; diff --git a/src/fq.cpp b/src/fq.cpp index 9028853..48a7029 100644 --- a/src/fq.cpp +++ b/src/fq.cpp @@ -32,18 +32,19 @@ zmq::fq_t::fq_t () : zmq::fq_t::~fq_t () { - for (pipes_t::size_type i = 0; i != pipes.size (); i++) - pipes [i]->term (); + zmq_assert (pipes.empty ()); } void zmq::fq_t::attach (reader_t *pipe_) { + pipe_->set_event_sink (this); + pipes.push_back (pipe_); pipes.swap (active, pipes.size () - 1); active++; } -void zmq::fq_t::detach (reader_t *pipe_) +void zmq::fq_t::terminated (reader_t *pipe_) { zmq_assert (!more || pipes [current] != pipe_); @@ -57,16 +58,18 @@ void zmq::fq_t::detach (reader_t *pipe_) pipes.erase (pipe_); } -void zmq::fq_t::kill (reader_t *pipe_) +bool zmq::fq_t::has_pipes () { - // Move the pipe to the list of inactive pipes. - active--; - if (current == active) - current = 0; - pipes.swap (pipes.index (pipe_), active); + return !pipes.empty (); +} + +void zmq::fq_t::term_pipes () +{ + for (pipes_t::size_type i = 0; i != pipes.size (); i++) + pipes [i]->terminate (); } -void zmq::fq_t::revive (reader_t *pipe_) +void zmq::fq_t::activated (reader_t *pipe_) { // Move the pipe to the list of active pipes. pipes.swap (pipes.index (pipe_), active); @@ -98,6 +101,12 @@ int zmq::fq_t::recv (zmq_msg_t *msg_, int flags_) } return 0; } + else { + active--; + pipes.swap (current, active); + if (current == active) + current = 0; + } } // No message is available. Initialise the output parameter diff --git a/src/fq.hpp b/src/fq.hpp index 5c699ee..2e09809 100644 --- a/src/fq.hpp +++ b/src/fq.hpp @@ -21,6 +21,7 @@ #define __ZMQ_FQ_HPP_INCLUDED__ #include "yarray.hpp" +#include "pipe.hpp" namespace zmq { @@ -28,24 +29,28 @@ namespace zmq // Class manages a set of inbound pipes. On receive it performs fair // queueing (RFC970) so that senders gone berserk won't cause denial of // service for decent senders. - class fq_t + class fq_t : public i_reader_events { public: fq_t (); ~fq_t (); - void attach (class reader_t *pipe_); - void detach (class reader_t *pipe_); - void kill (class reader_t *pipe_); - void revive (class reader_t *pipe_); + void attach (reader_t *pipe_); + bool has_pipes (); + void term_pipes (); + int recv (zmq_msg_t *msg_, int flags_); bool has_in (); + // i_reader_events implementation. + void activated (reader_t *pipe_); + void terminated (reader_t *pipe_); + private: // Inbound pipes. - typedef yarray_t pipes_t; + typedef yarray_t pipes_t; pipes_t pipes; // Number of active pipes. All the active pipes are located at the diff --git a/src/i_endpoint.hpp b/src/i_endpoint.hpp deleted file mode 100644 index 0d14224..0000000 --- a/src/i_endpoint.hpp +++ /dev/null @@ -1,43 +0,0 @@ -/* - Copyright (c) 2007-2010 iMatix Corporation - - This file is part of 0MQ. - - 0MQ is free software; you can redistribute it and/or modify it under - the terms of the Lesser GNU General Public License as published by - the Free Software Foundation; either version 3 of the License, or - (at your option) any later version. - - 0MQ is distributed in the hope that it will be useful, - but WITHOUT ANY WARRANTY; without even the implied warranty of - MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - Lesser GNU General Public License for more details. - - You should have received a copy of the Lesser GNU General Public License - along with this program. If not, see . -*/ - -#ifndef __ZMQ_I_ENDPOINT_HPP_INCLUDED__ -#define __ZMQ_I_ENDPOINT_HPP_INCLUDED__ - -#include "blob.hpp" - -namespace zmq -{ - - struct i_endpoint - { - virtual ~i_endpoint () {} - - virtual void attach_pipes (class reader_t *inpipe_, - class writer_t *outpipe_, const blob_t &peer_identity_) = 0; - virtual void detach_inpipe (class reader_t *pipe_) = 0; - virtual void detach_outpipe (class writer_t *pipe_) = 0; - virtual void kill (class reader_t *pipe_) = 0; - virtual void revive (class reader_t *pipe_) = 0; - virtual void revive (class writer_t *pipe_) = 0; - }; - -} - -#endif diff --git a/src/io_thread.cpp b/src/io_thread.cpp index fac6961..3d202cf 100644 --- a/src/io_thread.cpp +++ b/src/io_thread.cpp @@ -26,9 +26,8 @@ #include "err.hpp" #include "ctx.hpp" -zmq::io_thread_t::io_thread_t (ctx_t *ctx_, - uint32_t thread_slot_) : - object_t (ctx_, thread_slot_) +zmq::io_thread_t::io_thread_t (ctx_t *ctx_, uint32_t slot_) : + object_t (ctx_, slot_) { poller = new (std::nothrow) poller_t; zmq_assert (poller); diff --git a/src/io_thread.hpp b/src/io_thread.hpp index 3d832c0..9e7c2ea 100644 --- a/src/io_thread.hpp +++ b/src/io_thread.hpp @@ -38,7 +38,7 @@ namespace zmq { public: - io_thread_t (class ctx_t *ctx_, uint32_t thread_slot_); + io_thread_t (class ctx_t *ctx_, uint32_t slot_); // Clean-up. If the thread was started, it's neccessary to call 'stop' // before invoking destructor. Otherwise the destructor would hang up. diff --git a/src/lb.cpp b/src/lb.cpp index ca93ba2..ccfaaae 100644 --- a/src/lb.cpp +++ b/src/lb.cpp @@ -32,19 +32,27 @@ zmq::lb_t::lb_t () : zmq::lb_t::~lb_t () { - for (pipes_t::size_type i = 0; i != pipes.size (); i++) - pipes [i]->term (); + zmq_assert (pipes.empty ()); } void zmq::lb_t::attach (writer_t *pipe_) { + pipe_->set_event_sink (this); + pipes.push_back (pipe_); pipes.swap (active, pipes.size () - 1); active++; } -void zmq::lb_t::detach (writer_t *pipe_) +void zmq::lb_t::term_pipes () { + for (pipes_t::size_type i = 0; i != pipes.size (); i++) + pipes [i]->terminate (); +} + +void zmq::lb_t::terminated (writer_t *pipe_) +{ + // ??? zmq_assert (!more || pipes [current] != pipe_); // Remove the pipe from the list; adjust number of active pipes @@ -57,7 +65,12 @@ void zmq::lb_t::detach (writer_t *pipe_) pipes.erase (pipe_); } -void zmq::lb_t::revive (writer_t *pipe_) +bool zmq::lb_t::has_pipes () +{ + return !pipes.empty (); +} + +void zmq::lb_t::activated (writer_t *pipe_) { // Move the pipe to the list of active pipes. pipes.swap (pipes.index (pipe_), active); diff --git a/src/lb.hpp b/src/lb.hpp index 526a727..e69385e 100644 --- a/src/lb.hpp +++ b/src/lb.hpp @@ -21,25 +21,30 @@ #define __ZMQ_LB_HPP_INCLUDED__ #include "yarray.hpp" +#include "pipe.hpp" namespace zmq { // Class manages a set of outbound pipes. On send it load balances // messages fairly among the pipes. - class lb_t + class lb_t : public i_writer_events { public: lb_t (); ~lb_t (); - void attach (class writer_t *pipe_); - void detach (class writer_t *pipe_); - void revive (class writer_t *pipe_); + void attach (writer_t *pipe_); + void term_pipes (); + bool has_pipes (); int send (zmq_msg_t *msg_, int flags_); bool has_out (); + // i_writer_events interface implementation. + void activated (writer_t *pipe_); + void terminated (writer_t *pipe_); + private: // List of outbound pipes. diff --git a/src/object.cpp b/src/object.cpp index 324450f..cdb177f 100644 --- a/src/object.cpp +++ b/src/object.cpp @@ -28,15 +28,15 @@ #include "session.hpp" #include "socket_base.hpp" -zmq::object_t::object_t (ctx_t *ctx_, uint32_t thread_slot_) : +zmq::object_t::object_t (ctx_t *ctx_, uint32_t slot_) : ctx (ctx_), - thread_slot (thread_slot_) + slot (slot_) { } zmq::object_t::object_t (object_t *parent_) : ctx (parent_->ctx), - thread_slot (parent_->thread_slot) + slot (parent_->slot) { } @@ -44,9 +44,9 @@ zmq::object_t::~object_t () { } -uint32_t zmq::object_t::get_thread_slot () +uint32_t zmq::object_t::get_slot () { - return thread_slot; + return slot; } zmq::ctx_t *zmq::object_t::get_ctx () @@ -123,16 +123,6 @@ void zmq::object_t::process_command (command_t &cmd_) deallocate_command (&cmd_); } -void zmq::object_t::register_pipe (class pipe_t *pipe_) -{ - ctx->register_pipe (pipe_); -} - -void zmq::object_t::unregister_pipe (class pipe_t *pipe_) -{ - ctx->unregister_pipe (pipe_); -} - int zmq::object_t::register_endpoint (const char *addr_, socket_base_t *socket_) { return ctx->register_endpoint (addr_, socket_); @@ -153,6 +143,11 @@ zmq::io_thread_t *zmq::object_t::choose_io_thread (uint64_t taskset_) return ctx->choose_io_thread (taskset_); } +void zmq::object_t::zombify (socket_base_t *socket_) +{ + ctx->zombify (socket_); +} + void zmq::object_t::send_stop () { // 'stop' command goes always from administrative thread to @@ -160,7 +155,7 @@ void zmq::object_t::send_stop () command_t cmd; cmd.destination = this; cmd.type = command_t::stop; - ctx->send_command (thread_slot, cmd); + ctx->send_command (slot, cmd); } void zmq::object_t::send_plug (owned_t *destination_, bool inc_seqnum_) @@ -369,6 +364,6 @@ void zmq::object_t::process_seqnum () void zmq::object_t::send_command (command_t &cmd_) { - ctx->send_command (cmd_.destination->get_thread_slot (), cmd_); + ctx->send_command (cmd_.destination->get_slot (), cmd_); } diff --git a/src/object.hpp b/src/object.hpp index a38b0a6..c75a95a 100644 --- a/src/object.hpp +++ b/src/object.hpp @@ -32,18 +32,14 @@ namespace zmq { public: - object_t (class ctx_t *ctx_, uint32_t thread_slot_); + object_t (class ctx_t *ctx_, uint32_t slot_); object_t (object_t *parent_); virtual ~object_t (); - uint32_t get_thread_slot (); + uint32_t get_slot (); ctx_t *get_ctx (); void process_command (struct command_t &cmd_); - // Allow pipe to access corresponding context functions. - void register_pipe (class pipe_t *pipe_); - void unregister_pipe (class pipe_t *pipe_); - protected: // Using following function, socket is able to access global @@ -55,6 +51,10 @@ namespace zmq // Chooses least loaded I/O thread. class io_thread_t *choose_io_thread (uint64_t taskset_); + // Zombify particular socket. In other words, pass the ownership to + // the context. + void zombify (class socket_base_t *socket_); + // Derived object can use these functions to send commands // to other objects. void send_stop (); @@ -105,7 +105,7 @@ namespace zmq class ctx_t *ctx; // Slot ID of the thread the object belongs to. - uint32_t thread_slot; + uint32_t slot; void send_command (command_t &cmd_); diff --git a/src/owned.cpp b/src/owned.cpp index d6be444..7d1cf5e 100644 --- a/src/owned.cpp +++ b/src/owned.cpp @@ -35,7 +35,7 @@ zmq::owned_t::~owned_t () void zmq::owned_t::inc_seqnum () { - // NB: This function may be called from a different thread! + // This function may be called from a different thread! sent_seqnum.add (1); } @@ -62,10 +62,16 @@ void zmq::owned_t::finalise () { // If termination request was already received and there are no more // commands to wait for, terminate the object. - if (shutting_down && processed_seqnum == sent_seqnum.get ()) { + if (shutting_down && processed_seqnum == sent_seqnum.get () + && is_terminable ()) { process_unplug (); send_term_ack (owner); delete this; } } +bool zmq::owned_t::is_terminable () +{ + return true; +} + diff --git a/src/owned.hpp b/src/owned.hpp index 91189a1..80cf42f 100644 --- a/src/owned.hpp +++ b/src/owned.hpp @@ -45,6 +45,13 @@ namespace zmq protected: + // A mechanism allowing derived owned objects to postpone the + // termination process. Default implementation defines no such delay. + // Note that the derived object has to call finalise method when the + // delay is over. + virtual bool is_terminable (); + void finalise (); + // Ask owner socket to terminate this object. void term (); @@ -69,8 +76,6 @@ namespace zmq void process_term (); void process_seqnum (); - void finalise (); - // Sequence number of the last command sent to this object. atomic_counter_t sent_seqnum; diff --git a/src/pair.cpp b/src/pair.cpp index 3872b28..1ff2e1a 100644 --- a/src/pair.cpp +++ b/src/pair.cpp @@ -23,11 +23,12 @@ #include "err.hpp" #include "pipe.hpp" -zmq::pair_t::pair_t (class app_thread_t *parent_) : - socket_base_t (parent_), +zmq::pair_t::pair_t (class ctx_t *parent_, uint32_t slot_) : + socket_base_t (parent_, slot_), inpipe (NULL), outpipe (NULL), - alive (true) + inpipe_alive (false), + outpipe_alive (false) { options.requires_in = true; options.requires_out = true; @@ -35,56 +36,61 @@ zmq::pair_t::pair_t (class app_thread_t *parent_) : zmq::pair_t::~pair_t () { - if (inpipe) - inpipe->term (); - if (outpipe) - outpipe->term (); + zmq_assert (!inpipe); + zmq_assert (!outpipe); } void zmq::pair_t::xattach_pipes (class reader_t *inpipe_, class writer_t *outpipe_, const blob_t &peer_identity_) { zmq_assert (!inpipe && !outpipe); + inpipe = inpipe_; + inpipe_alive = true; + inpipe->set_event_sink (this); + outpipe = outpipe_; outpipe_alive = true; + outpipe->set_event_sink (this); } -void zmq::pair_t::xdetach_inpipe (class reader_t *pipe_) +void zmq::pair_t::terminated (class reader_t *pipe_) { zmq_assert (pipe_ == inpipe); inpipe = NULL; + inpipe_alive = false; } -void zmq::pair_t::xdetach_outpipe (class writer_t *pipe_) +void zmq::pair_t::terminated (class writer_t *pipe_) { zmq_assert (pipe_ == outpipe); outpipe = NULL; + outpipe_alive = false; } -void zmq::pair_t::xkill (class reader_t *pipe_) +void zmq::pair_t::xterm_pipes () { - zmq_assert (alive); - alive = false; + if (inpipe) + inpipe->terminate (); + if (outpipe) + outpipe->terminate (); } -void zmq::pair_t::xrevive (class reader_t *pipe_) +bool zmq::pair_t::xhas_pipes () { - zmq_assert (!alive); - alive = true; + return inpipe != NULL || outpipe != NULL; } -void zmq::pair_t::xrevive (class writer_t *pipe_) +void zmq::pair_t::activated (class reader_t *pipe_) { - zmq_assert (!outpipe_alive); - outpipe_alive = true; + zmq_assert (!inpipe_alive); + inpipe_alive = true; } -int zmq::pair_t::xsetsockopt (int option_, const void *optval_, - size_t optvallen_) +void zmq::pair_t::activated (class writer_t *pipe_) { - errno = EINVAL; - return -1; + zmq_assert (!outpipe_alive); + outpipe_alive = true; } int zmq::pair_t::xsend (zmq_msg_t *msg_, int flags_) @@ -100,7 +106,8 @@ int zmq::pair_t::xsend (zmq_msg_t *msg_, int flags_) return -1; } - outpipe->flush (); + if (!(flags_ & ZMQ_SNDMORE)) + outpipe->flush (); // Detach the original message from the data buffer. int rc = zmq_msg_init (msg_); @@ -114,9 +121,12 @@ int zmq::pair_t::xrecv (zmq_msg_t *msg_, int flags_) // Deallocate old content of the message. zmq_msg_close (msg_); - if (!alive || !inpipe || !inpipe->read (msg_)) { - // No message is available. Initialise the output parameter - // to be a 0-byte message. + if (!inpipe_alive || !inpipe || !inpipe->read (msg_)) { + + // No message is available. + inpipe_alive = false; + + // Initialise the output parameter to be a 0-byte message. zmq_msg_init (msg_); errno = EAGAIN; return -1; @@ -126,14 +136,16 @@ int zmq::pair_t::xrecv (zmq_msg_t *msg_, int flags_) bool zmq::pair_t::xhas_in () { - if (alive && inpipe && inpipe->check_read ()) - return true; - return false; + if (!inpipe || !inpipe_alive) + return false; + + inpipe_alive = inpipe->check_read (); + return inpipe_alive; } bool zmq::pair_t::xhas_out () { - if (outpipe == NULL || !outpipe_alive) + if (!outpipe || !outpipe_alive) return false; outpipe_alive = outpipe->check_write (); diff --git a/src/pair.hpp b/src/pair.hpp index aea249f..0c484d7 100644 --- a/src/pair.hpp +++ b/src/pair.hpp @@ -21,37 +21,45 @@ #define __ZMQ_PAIR_HPP_INCLUDED__ #include "socket_base.hpp" +#include "pipe.hpp" namespace zmq { - class pair_t : public socket_base_t + class pair_t : + public socket_base_t, + public i_reader_events, + public i_writer_events { public: - pair_t (class app_thread_t *parent_); + pair_t (class ctx_t *parent_, uint32_t slot_); ~pair_t (); // Overloads of functions from socket_base_t. void xattach_pipes (class reader_t *inpipe_, class writer_t *outpipe_, const blob_t &peer_identity_); - void xdetach_inpipe (class reader_t *pipe_); - void xdetach_outpipe (class writer_t *pipe_); - void xkill (class reader_t *pipe_); - void xrevive (class reader_t *pipe_); - void xrevive (class writer_t *pipe_); - int xsetsockopt (int option_, const void *optval_, size_t optvallen_); + void xterm_pipes (); + bool xhas_pipes (); int xsend (zmq_msg_t *msg_, int flags_); int xrecv (zmq_msg_t *msg_, int flags_); bool xhas_in (); bool xhas_out (); + // i_reader_events interface implementation. + void activated (class reader_t *pipe_); + void terminated (class reader_t *pipe_); + + // i_writer_events interface implementation. + void activated (class writer_t *pipe_); + void terminated (class writer_t *pipe_); + private: class reader_t *inpipe; class writer_t *outpipe; - bool alive; + bool inpipe_alive; bool outpipe_alive; pair_t (const pair_t&); diff --git a/src/pipe.cpp b/src/pipe.cpp index 200beb0..1903422 100644 --- a/src/pipe.cpp +++ b/src/pipe.cpp @@ -17,31 +17,54 @@ along with this program. If not, see . */ +#include + #include "../include/zmq.h" #include "pipe.hpp" +#include "likely.hpp" -zmq::reader_t::reader_t (object_t *parent_, uint64_t lwm_) : +zmq::reader_t::reader_t (object_t *parent_, pipe_t *pipe_, + uint64_t lwm_) : object_t (parent_), - pipe (NULL), - peer (NULL), + pipe (pipe_), + writer (NULL), lwm (lwm_), msgs_read (0), - endpoint (NULL) -{} + sink (NULL), + terminating (false) +{ + // Note that writer is not set here. Writer will inform reader about its + // address once it is created (via set_writer method). +} + +void zmq::reader_t::set_writer (writer_t *writer_) +{ + zmq_assert (!writer); + writer = writer_; +} zmq::reader_t::~reader_t () { - if (pipe) - unregister_pipe (pipe); + // Pipe as such is owned and deallocated by reader object. + // The point is that reader processes the last step of terminal + // handshaking (term_ack). + zmq_assert (pipe); + + // First delete all the unread messages in the pipe. We have to do it by + // hand because zmq_msg_t is a POD, not a class, so there's no associated + // destructor. + zmq_msg_t msg; + while (pipe->read (&msg)) + zmq_msg_close (&msg); + + delete pipe; } -void zmq::reader_t::set_pipe (pipe_t *pipe_) +void zmq::reader_t::set_event_sink (i_reader_events *sink_) { - zmq_assert (!pipe); - pipe = pipe_; - peer = &pipe->writer; - register_pipe (pipe); + zmq_assert (!sink); + sink = sink_; } bool zmq::reader_t::is_delimiter (zmq_msg_t &msg_) @@ -53,19 +76,20 @@ bool zmq::reader_t::is_delimiter (zmq_msg_t &msg_) bool zmq::reader_t::check_read () { + if (unlikely (terminating)) + return false; + // Check if there's an item in the pipe. // If not, deactivate the pipe. if (!pipe->check_read ()) { - endpoint->kill (this); + terminate (); return false; } // If the next item in the pipe is message delimiter, // initiate its termination. if (pipe->probe (is_delimiter)) { - if (endpoint) - endpoint->detach_inpipe (this); - term (); + terminate (); return false; } @@ -74,17 +98,16 @@ bool zmq::reader_t::check_read () bool zmq::reader_t::read (zmq_msg_t *msg_) { - if (!pipe->read (msg_)) { - endpoint->kill (this); + if (unlikely (terminating)) + return false; + + if (!pipe->read (msg_)) return false; - } // If delimiter was read, start termination process of the pipe. unsigned char *offset = 0; if (msg_->content == (void*) (offset + ZMQ_DELIMITER)) { - if (endpoint) - endpoint->detach_inpipe (this); - term (); + terminate (); return false; } @@ -92,51 +115,64 @@ bool zmq::reader_t::read (zmq_msg_t *msg_) msgs_read++; if (lwm > 0 && msgs_read % lwm == 0) - send_reader_info (peer, msgs_read); + send_reader_info (writer, msgs_read); return true; } -void zmq::reader_t::set_endpoint (i_endpoint *endpoint_) +void zmq::reader_t::terminate () { - endpoint = endpoint_; + // If termination was already started by the peer, do nothing. + if (terminating) + return; + + terminating = true; + send_pipe_term (writer); } -void zmq::reader_t::term () +bool zmq::reader_t::is_terminating () { - endpoint = NULL; - send_pipe_term (peer); + return terminating; } void zmq::reader_t::process_revive () { - // Beacuse of command throttling mechanism, incoming termination request - // may not have been processed before subsequent send. - // In that case endpoint is NULL. - if (endpoint) - endpoint->revive (this); + // Forward the event to the sink (either socket or session). + sink->activated (this); } void zmq::reader_t::process_pipe_term_ack () { - peer = NULL; - delete pipe; + // At this point writer may already be deallocated. + // For safety's sake drop the reference to it. + writer = NULL; + + // Notify owner about the termination. + zmq_assert (sink); + sink->terminated (this); + + // Deallocate resources. + delete this; } -zmq::writer_t::writer_t (object_t *parent_, +zmq::writer_t::writer_t (object_t *parent_, pipe_t *pipe_, reader_t *reader_, uint64_t hwm_, int64_t swap_size_) : object_t (parent_), - pipe (NULL), - peer (NULL), + pipe (pipe_), + reader (reader_), hwm (hwm_), msgs_read (0), msgs_written (0), msg_store (NULL), extra_msg_flag (false), stalled (false), - pending_close (false), - endpoint (NULL) + sink (NULL), + terminating (false), + pending_close (false) { + // Inform reader about the writer. + reader->set_writer (this); + if (swap_size_ > 0) { msg_store = new (std::nothrow) msg_store_t (swap_size_); if (msg_store != NULL) { @@ -148,11 +184,6 @@ zmq::writer_t::writer_t (object_t *parent_, } } -void zmq::writer_t::set_endpoint (i_endpoint *endpoint_) -{ - endpoint = endpoint_; -} - zmq::writer_t::~writer_t () { if (extra_msg_flag) @@ -161,15 +192,17 @@ zmq::writer_t::~writer_t () delete msg_store; } -void zmq::writer_t::set_pipe (pipe_t *pipe_) +void zmq::writer_t::set_event_sink (i_writer_events *sink_) { - zmq_assert (!pipe); - pipe = pipe_; - peer = &pipe->reader; + zmq_assert (!sink); + sink = sink_; } bool zmq::writer_t::check_write () { + if (terminating) + return false; + if (pipe_full () && (msg_store == NULL || msg_store->full () || extra_msg_flag)) { stalled = true; return false; @@ -180,6 +213,9 @@ bool zmq::writer_t::check_write () bool zmq::writer_t::write (zmq_msg_t *msg_) { + if (terminating) + return false; + if (!check_write ()) return false; @@ -216,23 +252,27 @@ void zmq::writer_t::rollback () while (pipe->unwrite (&msg)) { zmq_assert (msg.flags & ZMQ_MSG_MORE); zmq_msg_close (&msg); + msgs_written--; } - if (stalled && endpoint != NULL && check_write ()) { + if (stalled && check_write ()) { stalled = false; - endpoint->revive (this); + zmq_assert (sink); + sink->activated (this); } } void zmq::writer_t::flush () { if (!pipe->flush ()) - send_revive (peer); + send_revive (reader); } -void zmq::writer_t::term () +void zmq::writer_t::terminate () { - endpoint = NULL; + // Prevent double termination. + if (terminating) + return; // Rollback any unfinished messages. rollback (); @@ -293,71 +333,69 @@ void zmq::writer_t::process_reader_info (uint64_t msgs_read_) flush (); } - if (stalled && endpoint != NULL) { + if (stalled) { stalled = false; - endpoint->revive (this); + zmq_assert (sink); + sink->activated (this); } } void zmq::writer_t::process_pipe_term () { - if (endpoint) - endpoint->detach_outpipe (this); + send_pipe_term_ack (reader); - reader_t *p = peer; - peer = NULL; - send_pipe_term_ack (p); -} + // The above command allows reader to deallocate itself and the pipe. + // For safety's sake we'll drop the pointers here. + reader = NULL; + pipe = NULL; -bool zmq::writer_t::pipe_full () -{ - return hwm > 0 && msgs_written - msgs_read == hwm; -} + // Notify owner about the termination. + zmq_assert (sink); + sink->terminated (this); -zmq::pipe_t::pipe_t (object_t *reader_parent_, object_t *writer_parent_, - uint64_t hwm_, int64_t swap_size_) : - reader (reader_parent_, compute_lwm (hwm_)), - writer (writer_parent_, hwm_, swap_size_) -{ - reader.set_pipe (this); - writer.set_pipe (this); + // Deallocate the resources. + delete this; } -zmq::pipe_t::~pipe_t () +bool zmq::writer_t::pipe_full () { - // Deallocate all the unread messages in the pipe. We have to do it by - // hand because zmq_msg_t is a POD, not a class, so there's no associated - // destructor. - zmq_msg_t msg; - while (read (&msg)) - zmq_msg_close (&msg); + return hwm > 0 && msgs_written - msgs_read == hwm; } -uint64_t zmq::pipe_t::compute_lwm (uint64_t hwm_) +void zmq::create_pipe (object_t *reader_parent_, object_t *writer_parent_, + uint64_t hwm_, int64_t swap_size_, reader_t **reader_, writer_t **writer_) { - // Following point should be taken into consideration when computing - // low watermark: - // - // 1. LWM has to be less than HWM. - // 2. LWM cannot be set to very low value (such as zero) as after filling - // the queue it would start to refill only after all the messages are - // read from it and thus unnecessarily hold the progress back. - // 3. LWM cannot be set to very high value (such as HWM-1) as it would - // result in lock-step filling of the queue - if a single message is read - // from a full queue, writer thread is resumed to write exactly one - // message to the queue and go back to sleep immediately. This would - // result in low performance. - // - // Given the 3. it would be good to keep HWM and LWM as far apart as - // possible to reduce the thread switching overhead to almost zero, - // say HWM-LWM should be 500 (max_wm_delta). - // - // That done, we still we have to account for the cases where HWM<500 thus - // driving LWM to negative numbers. Let's make LWM 1/2 of HWM in such cases. - - if (hwm_ > max_wm_delta * 2) - return hwm_ - max_wm_delta; - else - return (hwm_ + 1) / 2; + // First compute the low water mark. Following point should be taken + // into consideration: + // + // 1. LWM has to be less than HWM. + // 2. LWM cannot be set to very low value (such as zero) as after filling + // the queue it would start to refill only after all the messages are + // read from it and thus unnecessarily hold the progress back. + // 3. LWM cannot be set to very high value (such as HWM-1) as it would + // result in lock-step filling of the queue - if a single message is + // read from a full queue, writer thread is resumed to write exactly one + // message to the queue and go back to sleep immediately. This would + // result in low performance. + // + // Given the 3. it would be good to keep HWM and LWM as far apart as + // possible to reduce the thread switching overhead to almost zero, + // say HWM-LWM should be max_wm_delta. + // + // That done, we still we have to account for the cases where + // HWM < max_wm_delta thus driving LWM to negative numbers. + // Let's make LWM 1/2 of HWM in such cases. + uint64_t lwm = (hwm_ > max_wm_delta * 2) ? + hwm_ - max_wm_delta : (hwm_ + 1) / 2; + + // Create all three objects pipe consists of: the pipe per se, reader and + // writer. The pipe will be handled by reader and writer, its never passed + // to the user. Reader and writer are returned to the user. + pipe_t *pipe = new (std::nothrow) pipe_t (); + zmq_assert (pipe); + *reader_ = new (std::nothrow) reader_t (reader_parent_, pipe, lwm); + zmq_assert (*reader_); + *writer_ = new (std::nothrow) writer_t (writer_parent_, pipe, *reader_, + hwm_, swap_size_); + zmq_assert (*writer_); } - diff --git a/src/pipe.hpp b/src/pipe.hpp index ece678a..34c5600 100644 --- a/src/pipe.hpp +++ b/src/pipe.hpp @@ -23,7 +23,6 @@ #include "../include/zmq.h" #include "stdint.hpp" -#include "i_endpoint.hpp" #include "yarray_item.hpp" #include "ypipe.hpp" #include "msg_store.hpp" @@ -33,15 +32,31 @@ namespace zmq { + // The shutdown mechanism for pipe works as follows: Either endpoint + // (or even both of them) can ask pipe to terminate by calling 'terminate' + // method. Pipe then terminates in asynchronous manner. When the part of + // the shutdown tied to the endpoint is done it triggers 'terminated' + // event. When endpoint processes the event and returns, associated + // reader/writer object is deallocated. + + typedef ypipe_t pipe_t; + + struct i_reader_events + { + virtual void terminated (class reader_t *pipe_) = 0; + virtual void activated (class reader_t *pipe_) = 0; + }; + class reader_t : public object_t, public yarray_item_t { - public: + friend void zmq::create_pipe (object_t*, object_t*, uint64_t, + int64_t, reader_t**, writer_t**); + friend class writer_t; - reader_t (class object_t *parent_, uint64_t lwm_); - ~reader_t (); + public: - void set_pipe (class pipe_t *pipe_); - void set_endpoint (i_endpoint *endpoint_); + // Specifies the object to get events from the reader. + void set_event_sink (i_reader_events *endpoint_); // Returns true if there is at least one message to read in the pipe. bool check_read (); @@ -50,10 +65,20 @@ namespace zmq bool read (zmq_msg_t *msg_); // Ask pipe to terminate. - void term (); + void terminate (); + + // Returns true if the pipe is already terminating + // (say if delimiter was already read). + bool is_terminating (); private: + reader_t (class object_t *parent_, pipe_t *pipe_, uint64_t lwm_); + ~reader_t (); + + // To be called only by writer itself! + void set_writer (class writer_t *writer_); + // Command handlers. void process_revive (); void process_pipe_term_ack (); @@ -62,10 +87,10 @@ namespace zmq static bool is_delimiter (zmq_msg_t &msg_); // The underlying pipe. - class pipe_t *pipe; + pipe_t *pipe; // Pipe writer associated with the other side of the pipe. - class writer_t *peer; + class writer_t *writer; // Low watermark for in-memory storage (in bytes). uint64_t lwm; @@ -73,22 +98,32 @@ namespace zmq // Number of messages read so far. uint64_t msgs_read; - // Endpoint (either session or socket) the pipe is attached to. - i_endpoint *endpoint; + // Sink for the events (either the socket of the session). + i_reader_events *sink; + + // True is 'terminate' method was called or delimiter + // was read from the pipe. + bool terminating; reader_t (const reader_t&); void operator = (const reader_t&); }; + struct i_writer_events + { + virtual void terminated (class writer_t *pipe_) = 0; + virtual void activated (class writer_t *pipe_) = 0; + }; + class writer_t : public object_t, public yarray_item_t { - public: + friend void zmq::create_pipe (object_t*, object_t*, uint64_t, + int64_t, reader_t**, writer_t**); - writer_t (class object_t *parent_, uint64_t hwm_, int64_t swap_size_); - ~writer_t (); + public: - void set_pipe (class pipe_t *pipe_); - void set_endpoint (i_endpoint *endpoint_); + // Specifies the object to get events from the writer. + void set_event_sink (i_writer_events *endpoint_); // Checks whether a message can be written to the pipe. // If writing the message would cause high watermark to be @@ -106,10 +141,14 @@ namespace zmq void flush (); // Ask pipe to terminate. - void term (); + voi