Skip to content


Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
tree: 37fd1a77a6
Fetching contributors…

Cannot retrieve contributors at this time

227 lines (186 sloc) 8.25 kb
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
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 "atomic_ptr.hpp"
#include "yqueue.hpp"
#include "platform.hpp"
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
// Initialises the pipe. In D scenario it is created in dead state.
// Otherwise it's alive.
inline ypipe_t () :
stop (false)
// Insert terminator element into the queue.
queue.push ();
// 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 ());
// Following function (write) deliberately copies uninitialised data
// when used with zmq_msg. Initialising the VSM body for
// non-VSM messages won't be good for performance.
#pragma message save
#pragma message disable(UNINIT)
// Write an item to the pipe. Don't flush it yet.
inline void write (const T &value_)
// Place the value to the queue, add new terminator element.
queue.back () = value_;
queue.push ();
#pragma message restore
// Pop an unflushed message from the pipe. Returns true is such
// message exists, false otherwise.
inline bool unwrite (T *value_)
if (w == &queue.back ())
return false;
queue.unpush ();
*value_ = queue.back ();
return true;
// Flush the messages into the pipe. Returns false if the reader
// thread is sleeping. In that case, caller is obliged to wake the
// reader up before using the pipe again.
inline bool flush ()
// If there are no un-flushed items, do nothing.
if (w == &queue.back ())
return true;
// Try to set 'c' to 'back'
if (c.cas (w, &queue.back ()) != w) {
// Compare-and-swap was unseccessful because 'c' is NULL.
// This means that the reader is asleep. Therefore we don't
// care about thread-safeness and update c in non-atomic
// manner. We'll return false to let the caller know
// that reader is sleeping.
c.set (&queue.back ());
w = &queue.back ();
return false;
// Reader is alive. Nothing special to do now. Just move
// the 'first un-flushed item' pointer to the end of the queue.
w = &queue.back ();
return true;
// Check whether item is available for reading.
inline bool check_read ()
// Was the value prefetched already? If so, return.
if (&queue.front () != r)
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;
// There was at least one value prefetched.
return true;
// Reads an item from the pipe. Returns false if there is no value.
// available.
inline bool read (T *value_)
// Try to prefetch a value.
if (!check_read ())
return false;
// There was at least one value prefetched.
// Return it to the caller.
*value_ = queue.front ();
queue.pop ();
return true;
// Allocation-efficient queue to store pipe items.
// Front of the queue points to the first prefetched item, back of
// the pipe points to last un-flushed item. Front is used only by
// reader thread, while back is used only by writer thread.
yqueue_t <T, N> queue;
// Points to the first un-flushed item. This variable is used
// exclusively by writer thread.
T *w;
// Points to the first un-prefetched item. This variable is used
// exclusively by reader thread.
T *r;
// The single point of contention between writer and reader thread.
// Points past the last flushed item. If it is NULL,
// reader is asleep. This pointer should be always accessed using
// 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&);
Jump to Line
Something went wrong with that request. Please try again.