diff options
| -rw-r--r-- | src/app_thread.cpp | 20 | ||||
| -rw-r--r-- | src/config.hpp | 5 | ||||
| -rw-r--r-- | src/dispatcher.cpp | 71 | ||||
| -rw-r--r-- | src/dispatcher.hpp | 30 | ||||
| -rw-r--r-- | src/io_thread.cpp | 17 | ||||
| -rw-r--r-- | src/object.cpp | 6 | ||||
| -rw-r--r-- | src/pipe.hpp | 2 | ||||
| -rw-r--r-- | src/signaler.cpp | 224 | ||||
| -rw-r--r-- | src/signaler.hpp | 45 | ||||
| -rw-r--r-- | src/ypipe.hpp | 83 |
10 files changed, 174 insertions, 329 deletions
diff --git a/src/app_thread.cpp b/src/app_thread.cpp index bd5f5e3..1c06337 100644 --- a/src/app_thread.cpp +++ b/src/app_thread.cpp @@ -82,9 +82,12 @@ zmq::signaler_t *zmq::app_thread_t::get_signaler () bool zmq::app_thread_t::process_commands (bool block_, bool throttle_) { - uint32_t signal; - if (block_) - signal = signaler.poll (); + bool received; + command_t cmd; + if (block_) { + received = signaler.recv (&cmd, true); + zmq_assert (received); + } else { #if defined ZMQ_DELAY_COMMANDS @@ -117,15 +120,12 @@ bool zmq::app_thread_t::process_commands (bool block_, bool throttle_) #endif // Check whether there are any commands pending for this thread. - signal = signaler.check (); + received = signaler.recv (&cmd, false); } - // Process all the commands from the signaling source if there is one. - if (signal != signaler_t::no_signal) { - command_t cmd; - while (get_dispatcher ()->read (signal, get_thread_slot (), &cmd)) - cmd.destination->process_command (cmd); - } + // Process the command, if any. + if (received) + cmd.destination->process_command (cmd); return !terminated; } diff --git a/src/config.hpp b/src/config.hpp index 99c9d86..5b51a08 100644 --- a/src/config.hpp +++ b/src/config.hpp @@ -32,11 +32,6 @@ namespace zmq // memory allocation by approximately 99.6% message_pipe_granularity = 256, - // Number of new commands in command pipe needed to trigger new memory - // allocation. The number should be kept low to decrease the memory - // footprint of dispatcher. - command_pipe_granularity = 4, - // Number of signals that can be read by the signaler // using a single system call. signal_buffer_size = 8, diff --git a/src/dispatcher.cpp b/src/dispatcher.cpp index b1ba11f..db06d32 100644 --- a/src/dispatcher.cpp +++ b/src/dispatcher.cpp @@ -18,6 +18,7 @@ */ #include <new> +#include <string.h> #include "../include/zmq.h" @@ -48,35 +49,30 @@ zmq::dispatcher_t::dispatcher_t (uint32_t app_threads_, uint32_t io_threads_) : HIBYTE (wsa_data.wVersion) == 2); #endif - // Create application thread proxies. - for (uint32_t i = 0; i != app_threads_; i++) { - app_thread_info_t info; - info.associated = false; - info.app_thread = new (std::nothrow) app_thread_t (this, i); - zmq_assert (info.app_thread); - app_threads.push_back (info); - signalers.push_back (info.app_thread->get_signaler ()); - } + // Initialise the array of signalers. + signalers_count = app_threads_ + io_threads_; + signalers = (signaler_t**) malloc (sizeof (signaler_t*) * signalers_count); + zmq_assert (signalers); + memset (signalers, 0, sizeof (signaler_t*) * signalers_count); // Create I/O thread objects. for (uint32_t i = 0; i != io_threads_; i++) { - io_thread_t *io_thread = new (std::nothrow) io_thread_t (this, - i + app_threads_); + io_thread_t *io_thread = new (std::nothrow) io_thread_t (this, i); zmq_assert (io_thread); io_threads.push_back (io_thread); - signalers.push_back (io_thread->get_signaler ()); + signalers [i] = io_thread->get_signaler (); } - // Create the administrative thread. Nothing special is needed. NULL - // is used instead of signaler given that as for now, administrative - // thread doesn't receive any commands. The only thing it is used for - // is sending 'stop' command to I/O threads on shutdown. - signalers.push_back (NULL); - - // Create command pipe matrix. - command_pipes = new (std::nothrow) command_pipe_t [signalers.size () * - signalers.size ()]; - zmq_assert (command_pipes); + // Create application thread proxies. + for (uint32_t i = 0; i != app_threads_; i++) { + app_thread_info_t info; + info.associated = false; + info.app_thread = new (std::nothrow) app_thread_t (this, + i + io_threads_); + zmq_assert (info.app_thread); + app_threads.push_back (info); + signalers [i + io_threads_] = info.app_thread->get_signaler (); + } // Launch I/O threads. for (uint32_t i = 0; i != io_threads_; i++) @@ -123,12 +119,11 @@ zmq::dispatcher_t::~dispatcher_t () while (!pipes.empty ()) delete *pipes.begin (); - // TODO: Deallocate any commands still in the pipes. Keep in mind that - // simple reading from a pipe and deallocating commands won't do as - // command pipe has template parameter D set to true, meaning that - // read may return false even if there are still commands in the pipe. - delete [] command_pipes; - + // Deallocate the array of pointers to signalers. No special work is + // needed as signalers themselves were deallocated with their + // corresponding (app_/io_) thread objects. + free (signalers); + #ifdef ZMQ_HAVE_WINDOWS // On Windows, uninitialise socket layer. int rc = WSACleanup (); @@ -136,11 +131,6 @@ zmq::dispatcher_t::~dispatcher_t () #endif } -uint32_t zmq::dispatcher_t::thread_slot_count () -{ - return (uint32_t) signalers.size (); -} - zmq::socket_base_t *zmq::dispatcher_t::create_socket (int type_) { app_threads_sync.lock (); @@ -213,21 +203,16 @@ void zmq::dispatcher_t::no_sockets (app_thread_t *thread_) app_threads_sync.unlock (); } -void zmq::dispatcher_t::write (uint32_t source_, uint32_t destination_, +void zmq::dispatcher_t::send_command (uint32_t destination_, const command_t &command_) { - command_pipe_t &pipe = - command_pipes [source_ * signalers.size () + destination_]; - pipe.write (command_); - if (!pipe.flush ()) - signalers [destination_]->signal (source_); + signalers [destination_]->send (command_); } -bool zmq::dispatcher_t::read (uint32_t source_, uint32_t destination_, - command_t *command_) +bool zmq::dispatcher_t::recv_command (uint32_t thread_slot_, + command_t *command_, bool block_) { - return command_pipes [source_ * signalers.size () + - destination_].read (command_); + return signalers [thread_slot_]->recv (command_, block_); } zmq::io_thread_t *zmq::dispatcher_t::choose_io_thread (uint64_t affinity_) diff --git a/src/dispatcher.hpp b/src/dispatcher.hpp index 0a1ed5c..0986e8a 100644 --- a/src/dispatcher.hpp +++ b/src/dispatcher.hpp @@ -27,7 +27,6 @@ #include "signaler.hpp" #include "ypipe.hpp" -#include "command.hpp" #include "config.hpp" #include "mutex.hpp" #include "stdint.hpp" @@ -69,19 +68,12 @@ namespace zmq // should disassociate the object from the current OS thread. void no_sockets (class app_thread_t *thread_); - // Returns number of thread slots in the dispatcher. To be used by - // individual threads to find out how many distinct signals can be - // received. - uint32_t thread_slot_count (); + // Send command to the destination thread. + void send_command (uint32_t destination_, const command_t &command_); - // Send command from the source to the destination. - void write (uint32_t source_, uint32_t destination_, - const command_t &command_); - - // Receive command from the source. Returns false if there is no - // command available. - bool read (uint32_t source_, uint32_t destination_, - command_t *command_); + // Receive command from another thread. + bool recv_command (uint32_t thread_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). @@ -126,15 +118,9 @@ namespace zmq typedef std::vector <class io_thread_t*> io_threads_t; io_threads_t io_threads; - // Signalers for both application and I/O threads. - std::vector <signaler_t*> signalers; - - // Pipe to hold the commands. - typedef ypipe_t <command_t, true, - command_pipe_granularity> command_pipe_t; - - // NxN matrix of command pipes. - command_pipe_t *command_pipes; + // 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 diff --git a/src/io_thread.cpp b/src/io_thread.cpp index b38f58c..92c314a 100644 --- a/src/io_thread.cpp +++ b/src/io_thread.cpp @@ -22,10 +22,8 @@ #include "../include/zmq.h" #include "io_thread.hpp" -#include "command.hpp" #include "platform.hpp" #include "err.hpp" -#include "command.hpp" #include "dispatcher.hpp" zmq::io_thread_t::io_thread_t (dispatcher_t *dispatcher_, @@ -67,17 +65,18 @@ int zmq::io_thread_t::get_load () void zmq::io_thread_t::in_event () { + // TODO: Do we want to limit number of commands I/O thread can + // process in a single go? + while (true) { - // Get the next signal. - uint32_t signal = signaler.check (); - if (signal == signaler_t::no_signal) + // Get the next command. If there is none, exit. + command_t cmd; + if (!signaler.recv (&cmd, false)) break; - // Process all the commands from the thread that sent the signal. - command_t cmd; - while (get_dispatcher ()->read (signal, get_thread_slot (), &cmd)) - cmd.destination->process_command (cmd); + // Process the command. + cmd.destination->process_command (cmd); } } diff --git a/src/object.cpp b/src/object.cpp index 38eb1f8..c5c89cb 100644 --- a/src/object.cpp +++ b/src/object.cpp @@ -157,11 +157,10 @@ void zmq::object_t::send_stop () { // 'stop' command goes always from administrative thread to // the current object. - uint32_t admin_thread_id = dispatcher->thread_slot_count () - 1; command_t cmd; cmd.destination = this; cmd.type = command_t::stop; - dispatcher->write (admin_thread_id, thread_slot, cmd); + dispatcher->send_command (thread_slot, cmd); } void zmq::object_t::send_plug (owned_t *destination_, bool inc_seqnum_) @@ -370,7 +369,6 @@ void zmq::object_t::process_seqnum () void zmq::object_t::send_command (command_t &cmd_) { - uint32_t destination_thread_slot = cmd_.destination->get_thread_slot (); - dispatcher->write (thread_slot, destination_thread_slot, cmd_); + dispatcher->send_command (cmd_.destination->get_thread_slot (), cmd_); } diff --git a/src/pipe.hpp b/src/pipe.hpp index a4569e3..b0428b5 100644 --- a/src/pipe.hpp +++ b/src/pipe.hpp @@ -145,7 +145,7 @@ namespace zmq }; // Message pipe. - class pipe_t : public ypipe_t <zmq_msg_t, false, message_pipe_granularity> + class pipe_t : public ypipe_t <zmq_msg_t, message_pipe_granularity> { public: diff --git a/src/signaler.cpp b/src/signaler.cpp index e1fa6a3..0fa43f7 100644 --- a/src/signaler.cpp +++ b/src/signaler.cpp @@ -30,52 +30,9 @@ #else #include <unistd.h> #include <fcntl.h> +#include <limits.h> #endif -const uint32_t zmq::signaler_t::no_signal = 0xffffffff; - -uint32_t zmq::signaler_t::poll () -{ - // Return next signal. - if (current != count) { - uint32_t result = buffer [current]; - current++; - return result; - } - - // If there is no signal buffered, poll for new signals. - xpoll (); - - // Return first signal. - zmq_assert (current != count); - uint32_t result = buffer [current]; - current++; - return result; -} - -uint32_t zmq::signaler_t::check () -{ - // Return next signal. - if (current != count) { - uint32_t result = buffer [current]; - current++; - return result; - } - - // If there is no signal buffered, check whether more signals - // can be obtained. - xcheck (); - - // Return first signal if any. - if (current != count) { - uint32_t result = buffer [current]; - current++; - return result; - } - - return no_signal; -} - zmq::fd_t zmq::signaler_t::get_fd () { return r; @@ -84,8 +41,6 @@ zmq::fd_t zmq::signaler_t::get_fd () #if defined ZMQ_HAVE_WINDOWS zmq::signaler_t::signaler_t () : - current (0), - count (0) { // Windows have no 'socketpair' function. CreatePipe is no good as pipe // handles cannot be polled on. Here we create the socketpair by hand. @@ -146,51 +101,49 @@ zmq::signaler_t::~signaler_t () wsa_assert (rc != SOCKET_ERROR); } -void zmq::signaler_t::signal (uint32_t signal_) +void zmq::signaler_t::send (const command_t &cmd_) { // TODO: Note that send is a blocking operation. // How should we behave if the signal cannot be written to the signaler? - int rc = send (w, (char*) &signal_, sizeof (signal_), 0); + // Even worse: What if half of a command is written? + int rc = send (w, (char*) &cmd_, sizeof (command_t), 0); win_assert (rc != SOCKET_ERROR); - zmq_assert (rc == sizeof (signal_)); + zmq_assert (rc == sizeof (command_t)); } -void zmq::signaler_t::xpoll () +bool zmq::signaler_t::recv (command_t *cmd_, bool block_) { - // Switch to blocking mode. - unsigned long argp = 0; - int rc = ioctlsocket (r, FIONBIO, &argp); - wsa_assert (rc != SOCKET_ERROR); + if (block_) { - // Get the signals. Given that we are in the blocking mode now, - // there should be at least a single signal returned. - xcheck (); - zmq_assert (current != count); + // Switch to blocking mode. + unsigned long argp = 0; + int rc = ioctlsocket (r, FIONBIO, &argp); + wsa_assert (rc != SOCKET_ERROR); + } - // Switch back to non-blocking mode. - argp = 1; - rc = ioctlsocket (r, FIONBIO, &argp); - wsa_assert (rc != SOCKET_ERROR); -} + bool result; + int nbytes = recv (r, (char*) cmd_, sizeof (command_t), 0); + if (nbytes == -1 && WSAGetLastError () == WSAEWOULDBLOCK) { + result = false; + } + else { + wsa_assert (nbytes != -1); -void zmq::signaler_t::xcheck () -{ - int nbytes = recv (r, (char*) buffer, sizeof (buffer), 0); + // Check whether we haven't got half of a signal. + zmq_assert (nbytes % sizeof (uint32_t) == 0); - // No signals are available. - if (nbytes == -1 && WSAGetLastError () == WSAEWOULDBLOCK) { - current = 0; - count = 0; - return; + result = true; } - wsa_assert (nbytes != -1); + if (block_) { - // Check whether we haven't got half of a signal. - zmq_assert (nbytes % sizeof (uint32_t) == 0); + // Switch back to non-blocking mode. + unsigned long argp = 1; + int rc = ioctlsocket (r, FIONBIO, &argp); + wsa_assert (rc != SOCKET_ERROR); + } - current = 0; - count = nbytes / sizeof (uint32_t); + return result; } #elif defined ZMQ_HAVE_HPUX || defined ZMQ_HAVE_AIX @@ -198,9 +151,7 @@ void zmq::signaler_t::xcheck () #include <sys/types.h> #include <sys/socket.h> -zmq::signaler_t::signaler_t () : - current (0), - count (0) +zmq::signaler_t::signaler_t () { int sv [2]; int rc = socketpair (AF_UNIX, SOCK_STREAM, 0, sv); @@ -222,49 +173,50 @@ zmq::signaler_t::~signaler_t () close (r); } -void zmq::signaler_t::signal (uint32_t signal_) +void zmq::signaler_t::send (const command_t &cmd_) { - ssize_t nbytes = send (w, &signal_, sizeof (signal_), 0); + ssize_t nbytes = send (w, &cmd_, sizeof (command_t), 0); errno_assert (nbytes != -1); - zmq_assert (nbytes == sizeof (signal_); + zmq_assert (nbytes == sizeof (command_t)); } -void zmq::signaler_t::xpoll () +bool zmq::signaler_t::recv (command_t &cmd_, bool block_) { - // Set the reader to blocking mode. - int flags = fcntl (r, F_GETFL, 0); - if (flags == -1) - flags = 0; - int rc = fcntl (r, F_SETFL, flags & ~O_NONBLOCK); - errno_assert (rc != -1); + if (block_) { + + // Set the reader to blocking mode. + int flags = fcntl (r, F_GETFL, 0); + if (flags == -1) + flags = 0; + int rc = fcntl (r, F_SETFL, flags & ~O_NONBLOCK); + errno_assert (rc != -1); + } - // Poll for events. - xcheck (); - zmq_assert (current != count); + bool result; + ssize_t nbytes = recv (r, buffer, sizeof (command_t), 0); + if (nbytes == -1 && errno == EAGAIN) { + result = false; + } + else { + zmq_assert (nbytes != -1); - // Set the reader to non-blocking mode. - flags = fcntl (r, F_GETFL, 0); - if (flags == -1) - flags = 0; - rc = fcntl (r, F_SETFL, flags | O_NONBLOCK); - errno_assert (rc != -1); -} + // Check whether we haven't got half of command. + zmq_assert (nbytes == sizeof (command_t)); -void zmq::signaler_t::xcheck () -{ - ssize_t nbytes = recv (r, buffer, sizeof (buffer), 0); - if (nbytes == -1 && errno == EAGAIN) { - current = 0; - count = 0; - return; + result = true; } - zmq_assert (nbytes != -1); - // Check whether we haven't got half of a signal. - zmq_assert (nbytes % sizeof (uint32_t) == 0); + if (block_) + + // Set the reader to non-blocking mode. + int flags = fcntl (r, F_GETFL, 0); + if (flags == -1) + flags = 0; + int rc = fcntl (r, F_SETFL, flags | O_NONBLOCK); + errno_assert (rc != -1); + } - current = 0; - count = nbytes / sizeof (uint32_t); + return result; } #else @@ -272,10 +224,13 @@ void zmq::signaler_t::xcheck () #include <sys/types.h> #include <sys/socket.h> -zmq::signaler_t::signaler_t () : - current (0), - count (0) +zmq::signaler_t::signaler_t () { + // Make sure that command can be written to the socket in atomic fashion. + // If this wasn't guaranteed, commands from different threads would be + // interleaved. + zmq_assert (sizeof (command_t) <= PIPE_BUF); + int sv [2]; int rc = socketpair (AF_UNIX, SOCK_STREAM, 0, sv); errno_assert (rc == 0); @@ -289,42 +244,33 @@ zmq::signaler_t::~signaler_t () close (r); } -void zmq::signaler_t::signal (uint32_t signal_) +void zmq::signaler_t::send (const command_t &cmd_) { // TODO: Note that send is a blocking operation. - // How should we behave if the signal cannot be written to the signaler? - ssize_t nbytes = send (w, &signal_, sizeof (signal_), 0); + // How should we behave if the command cannot be written to the signaler? + ssize_t nbytes = ::send (w, &cmd_, sizeof (command_t), 0); errno_assert (nbytes != -1); - zmq_assert (nbytes == sizeof (signal_)); + + // This should never happen as we've already checked that command size is + // less than PIPE_BUF. + zmq_assert (nbytes == sizeof (command_t)); } -void zmq::signaler_t::xpoll () +bool zmq::signaler_t::recv (command_t *cmd_, bool block_) { - ssize_t nbytes = recv (r, buffer, sizeof (buffer), 0); - errno_assert (nbytes != -1); - - // Check whether we haven't got half of a signal. - zmq_assert (nbytes % sizeof (uint32_t) == 0); + ssize_t nbytes = ::recv (r, cmd_, sizeof (command_t), + block_ ? 0 : MSG_DONTWAIT); - current = 0; - count = nbytes / sizeof (uint32_t); -} + // If there's no signal available return false. + if (nbytes == -1 && errno == EAGAIN) + return false; -void zmq::signaler_t::xcheck () -{ - ssize_t nbytes = recv (r, buffer, 64, MSG_DONTWAIT); - if (nbytes == -1 && errno == EAGAIN) { - current = 0; - count = 0; - return; - } errno_assert (nbytes != -1); - // Check whether we haven't got half of a signal. - zmq_assert (nbytes % sizeof (uint32_t) == 0); + // Check whether we haven't got half of command. + zmq_assert (nbytes == sizeof (command_t)); - current = 0; - count = nbytes / sizeof (uint32_t); + return true; } #endif diff --git a/src/signaler.hpp b/src/signaler.hpp index 5509894..64a1899 100644 --- a/src/signaler.hpp +++ b/src/signaler.hpp @@ -26,15 +26,11 @@ #include "fd.hpp" #include "stdint.hpp" #include "config.hpp" +#include "command.hpp" namespace zmq { - // This object can be used to send individual signals from one thread to - // another. The specific of this pipe is that it has associated file - // descriptor and so it can be polled on. Same signal cannot be sent twice - // unless signals are retrieved by the reader side in the meantime. - class signaler_t { public: @@ -42,44 +38,29 @@ namespace zmq signaler_t (); ~signaler_t (); - static const uint32_t no_signal; - - void signal (uint32_t signal_); - uint32_t poll (); - uint32_t check (); fd_t get_fd (); - + void send (const command_t &cmd_); + bool recv (command_t *cmd_, bool block_); + private: - void xpoll (); - void xcheck (); - #if defined ZMQ_HAVE_OPENVMS - // Whilst OpenVMS supports socketpair - it maps to AF_INET only. - // Further, it does not set the socket options TCP_NODELAY and - // TCP_NODELACK which can lead to performance problems. We'll - // overload the socketpair function for this class. - // - // The bug will be fixed in V5.6 ECO4 and beyond. In the - // meantime, we'll create the socket pair manually. - static int socketpair (int domain_, int type_, int protocol_, - int sv_ [2]); + // Whilst OpenVMS supports socketpair - it maps to AF_INET only. + // Further, it does not set the socket options TCP_NODELAY and + // TCP_NODELACK which can lead to performance problems. We'll + // overload the socketpair function for this class. + // + // The bug will be fixed in V5.6 ECO4 and beyond. In the + // meantime, we'll create the socket pair manually. + static int socketpair (int domain_, int type_, int protocol_, + int sv_ [2]); #endif // Write & read end of the socketpair. fd_t w; fd_t r; - // Signal buffer. - uint32_t buffer [signal_buffer_size]; - - // Position of the next signal in the buffer to return to the user. - size_t current; - - // Number of signals in the signal buffer. - size_t count; - // Disable copying of fd_signeler object. signaler_t (const signaler_t&); void operator = (const signaler_t&); diff --git a/src/ypipe.hpp b/src/ypipe.hpp index 83ae6a7..445b487 100644 --- a/src/ypipe.hpp +++ b/src/ypipe.hpp @@ -30,31 +30,24 @@ namespace zmq // Lock-free queue implementation. // Only a single thread can read from the pipe at any specific moment. // Only a single thread can write to the pipe at any specific moment. - // // T is the type of the object in the queue. - // If the template parameter D is set to true, it is quaranteed that - // the pipe will die in a finite time (so that you can swich to some - // other task). If D is set to false, reading from the pipe may result - // in an infinite cycle (if the pipe is continuosly fed by new elements). - // N is granularity of the pipe (how many elements have to be inserted - // till actual memory allocation is required). - - template <typename T, bool D, int N> class ypipe_t + // N is granularity of the pipe, i.e. how many messages are needed to + // perform next memory allocation. + + template <typename T, int N> class ypipe_t { public: - // Initialises the pipe. In D scenario it is created in dead state. - // Otherwise it's alive. - inline ypipe_t () : - stop (false) + // Initialises the pipe. + inline ypipe_t () { // Insert terminator element into the queue. queue.push (); - // Let all the pointers to point to the terminator + // Let all the pointers to point to the terminator. // (unless pipe is dead, in which case c is set to NULL). r = w = &queue.back (); - c.set (D ? NULL : &queue.back ()); + c.set (&queue.back ()); } // Following function (write) deliberately copies uninitialised data @@ -125,50 +118,17 @@ namespace zmq return true; // There's no prefetched value, so let us prefetch more values. - // (Note that D is a template parameter. Becaue of that one of - // the following branches will be completely optimised away - // by the compiler.) - if (D) { - - // If one prefetch was already done since last sleeping, - // don't do a new one, rather ask caller to go asleep. - if (stop) { - stop = false; - return false; - } - - // Get new items. Perform the operation in atomic fashion. - r = c.xchg (NULL); - - // If there are no elements prefetched, exit and go asleep. - // During pipe's lifetime r should never be NULL, however, - // during pipe shutdown when retrieving messages from it - // to deallocate them, this can happen. - if (&queue.front () == r || !r) { - stop = false; - return false; - } - - // We want to do only a single prefetch in D scenario - // before going asleep. Thus, we set stop variable to true - // so that we can return false next time the prefetch is - // attempted. - stop = true; - } - else { - - // Prefetching in non-D scenario is to simply retrieve the - // pointer from c in atomic fashion. If there are no - // items to prefetch, set c to NULL (using compare-and-swap). - r = c.cas (&queue.front (), NULL); - - // If there are no elements prefetched, exit. - // During pipe's lifetime r should never be NULL, however, - // it can happen during pipe shutdown when messages - // are being deallocated. - if (&queue.front () == r || !r) - return false; - } + // Prefetching is to simply retrieve the + // pointer from c in atomic fashion. If there are no + // items to prefetch, set c to NULL (using compare-and-swap). + r = c.cas (&queue.front (), NULL); + + // If there are no elements prefetched, exit. + // During pipe's lifetime r should never be NULL, however, + // it can happen during pipe shutdown when messages + // are being deallocated. + if (&queue.front () == r || !r) + return false; // There was at least one value prefetched. return true; @@ -211,11 +171,6 @@ namespace zmq // atomic operations. atomic_ptr_t <T> c; - // Used only if 'D' template parameter is set to true. If true, - // prefetch was already done since last sleeping and the reader - // should go asleep instead of prefetching once more. - bool stop; - // Disable copying of ypipe object. ypipe_t (const ypipe_t&); void operator = (const ypipe_t&); |