async_queue.h

Go to the documentation of this file.

/* Copyright (C) 2006 to 2013 Chris Vine
 
The library comprised in this file or of which this file is part is
distributed by Chris Vine under the GNU Lesser General Public
License as follows:
 
   This library is free software; you can redistribute it and/or
   modify it under the terms of the GNU Lesser General Public License
   as published by the Free Software Foundation; either version 2.1 of
   the License, or (at your option) any later version.
 
   This library 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 GNU
   Lesser General Public License, version 2.1, for more details.
 
   You should have received a copy of the GNU Lesser General Public
   License, version 2.1, along with this library (see the file LGPL.TXT
   which came with this source code package in the c++-gtk-utils
   sub-directory); if not, write to the Free Software Foundation, Inc.,
   51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 
However, it is not intended that the object code of a program whose
source code instantiates a template from this file or uses macros or
inline functions (of any length) should by reason only of that
instantiation or use be subject to the restrictions of use in the GNU
Lesser General Public License.  With that in mind, the words "and
macros, inline functions and instantiations of templates (of any
length)" shall be treated as substituted for the words "and small
macros and small inline functions (ten lines or less in length)" in
the fourth paragraph of section 5 of that licence.  This does not
affect any other reason why object code may be subject to the
restrictions in that licence (nor for the avoidance of doubt does it
affect the application of section 2 of that licence to modifications
of the source code in this file).
 
*/
 
/**
 * @file async_queue.h
 * @brief This file provides thread-safe asynchronous queue classes.
 *
 * AsyncQueue is a class which provides some of the functionality of a
 * std::queue object (but note that the AsyncQueue::pop(value_type&
 * obj) method provides the pop()ed element by reference - see the
 * comments on that method for the reason), except that it has mutex
 * locking of the data container so as to permit push()ing and
 * pop()ing from different threads.  It is therefore useful for
 * passing data between threads, perhaps in response to a signal being
 * emitted from a Notifier object.
 *
 * AsyncQueueDispatch is a class which has a blocking pop() method,
 * which allows it to be waited on by a dedicated event/message
 * dispatching thread for incoming work (represented by the data
 * pushed onto the queue).  In the same way, it can be used to
 * implement thread pools, by having threads in the pool waiting on
 * the queue.
 *
 * By default the queues use a std::list object as their container
 * because when adding an item to the queue all allocation can take
 * place outside the queue object's mutex.  However, for types which
 * have low overhead copy constructors, this can be changed to, say, a
 * std::deque object by specifying it as the second template
 * parameter.
 *
 * If data pushed and popped from a queue are held by a reference
 * counted smart pointer, the reference count must be thread-safe,
 * such as by using SharedLockPtr or IntrusiveLockCounter.
 */
 
#ifndef CGU_ASYNC_QUEUE_H
#define CGU_ASYNC_QUEUE_H
 
#include <queue>
#include <list>
#include <exception>
#include <algorithm>  // for std::swap
#include <time.h>
#include <c++-gtk-utils/mutex.h>
#include <c++-gtk-utils/thread.h>
#include <c++-gtk-utils/cgu_config.h>
 
#ifdef CGU_USE_SCHED_YIELD
#include <sched.h>
#else
#include <unistd.h>
#endif
 
namespace Cgu {
 
/**
 * @class AsyncQueuePopError async_queue.h c++-gtk-utils/async_queue.h
 * @brief An exception thrown if calling pop() on a AsyncQueue or
 * AsyncQueueDispatch object fails because the queue is empty.
 * @sa AsyncQueue AsyncQueueDispatch
 */
 
struct AsyncQueuePopError: public std::exception {
  virtual const char* what() const throw() {return "AsyncQueuePopError: popping from empty AsyncQueue object\n";}
};
 
 
/**
 * @class AsyncQueue async_queue.h c++-gtk-utils/async_queue.h
 * @brief A thread-safe asynchronous queue.
 * @sa AsyncQueueDispatch AsyncChannel AsyncResult
 *
 * AsyncQueue is a class which provides some of the functionality of a
 * std::queue object (but note that the AsyncQueue::pop(value_type&
 * obj) method provides the pop()ed element by reference - see the
 * comments on that method for the reason), except that it has mutex
 * locking of the data container so as to permit push()ing and
 * pop()ing from different threads.  It is therefore useful for
 * passing data between threads, perhaps in response to a signal being
 * emitted from a Notifier object.
 *
 * By default the queue uses a std::list object as its container
 * because when adding an item to the queue all allocation can take
 * place outside the queue object's mutex.  However, for types which
 * have low overhead copy constructors, this can be changed to, say, a
 * std::deque object by specifying it as the second template
 * parameter.
 *
 * If data pushed and popped from the queue are held by a reference
 * counted smart pointer, the reference count must be thread-safe,
 * such as by using SharedLockPtr or IntrusiveLockCounter.
 *
 * If the library is installed using the
 * \--with-glib-memory-slices-compat or
 * \--with-glib-memory-slices-no-compat configuration options, any
 * AsyncQueue objects constructed on free store will be constructed in
 * glib memory slices.  This does not affect the queue container
 * itself: to change the allocator of the C++ container, a custom
 * allocator type can be provided when the AsyncQueue object is
 * instantiated offering the standard allocator interface.  If glib
 * memory slices are not used or no AsyncQueue objects are constructed
 * on free store, it is not necessary to call g_thread_init() before
 * manipulating or using an AsyncQueue object in multiple threads, but
 * prior to glib version 2.32 glib itself (and thus glib memory
 * slices) are not thread safe unless that function has been called.
 */
 
template <class T, class Container = std::list<T> > class AsyncQueue {
public:
  typedef typename Container::value_type value_type;
  typedef typename Container::size_type size_type;
  typedef Container container_type;
private:
  std::queue<T, Container> q;
  mutable Thread::Mutex mutex;
 
// TODO: at the next ABI break make this method explicitly static
// this method won't throw: it is for the user to ensure the arguments
// do not refer to the same mutex object
  void lock2(Thread::Mutex& m1, Thread::Mutex& m2) {
    m1.lock();
    for(;;) {
      if (!m2.trylock()) {
    return;
      }
      m1.unlock();
      // spin nicely
#ifdef CGU_USE_SCHED_YIELD
      sched_yield();
#else
      usleep(10);
#endif
      m1.lock();
    }
  }
public:
/**
 * Pushes an item onto the queue.  This method has strong exception
 * safety if the container is a std::list or std::deque container (the
 * default is std::list), except that if std::deque is used as the
 * container under C++11 and the copy constructor, move constructor,
 * assignment operator or move assignment operator of the queue item
 * throws, it only gives the basic exception guarantee (and the basic
 * guarantee is not given by std::deque under C++11 if the queue
 * item's move constructor throws and it uses a non-default allocator
 * which does not provide for it to be CopyInsertable).  It is thread
 * safe.
 * @param obj The item to be pushed onto the queue.
 * @exception std::bad_alloc The method might throw std::bad_alloc if
 * memory is exhausted and the system throws in that case.  It might
 * also throw if the copy constructor or assignment operator of the
 * queue item might throw (or under C++11 if its move constructor or
 * move assignment operator throws).
 */
  void push(const value_type& obj) {
    Thread::Mutex::Lock lock(mutex);
    q.push(obj);
  }
 
/**
 * Pops an item from the queue.  This method has strong exception
 * safety if the container is a std::deque or std::list container (the
 * default is std::list), provided the destructor of a contained item
 * does not throw.  It is thread safe.
 * @param obj A value type reference to which the item at the front of
 * the queue will be assigned.
 * @exception AsyncQueuePopError If the queue is empty when a pop is
 * attempted, this method will throw AsyncQueuePopError.  It might
 * also throw if the assignment operator of the queue item might
 * throw.  In order to complete pop() operations atomically under a
 * single lock and to retain strong exception safety, the object into
 * which the pop()ed data is to be placed is passed as an argument by
 * reference (this avoids a copy from a temporary object after the
 * data has been extracted from the queue, which would occur if the
 * item extracted were returned by value).  It might also throw if the
 * destructor of the queue item might throw (but that should never
 * happen), or if the empty() method of the container type throws
 * (which would not happen on any sane implementation).
 */
  void pop(value_type& obj) {
    Thread::Mutex::Lock lock(mutex);
    if (q.empty()) throw AsyncQueuePopError();
    obj = q.front();
    q.pop();
  }
 
/**
 * Discards the item at the front of the queue.  This method has
 * strong exception safety if the container is a std::deque or
 * std::list container (the default is std::list), provided the
 * destructor of a contained item does not throw.  It is thread safe.
 * @exception AsyncQueuePopError If the queue is empty when a pop is
 * attempted, this method will throw AsyncQueuePopError.  It might
 * also throw if the destructor of the queue item might throw (but
 * that should never happen), or if the empty() method of the
 * container type throws (which would not happen on any sane
 * implementation).
 */
  void pop() {
    Thread::Mutex::Lock lock(mutex);
    if (q.empty()) throw AsyncQueuePopError();
    q.pop();
  }
 
/**
 * @return Whether the queue is empty.  It will not throw assuming
 * that the empty() method of the container type does not throw, as it
 * will not on any sane implementation.
 * @note This method is thread safe, but the return value may not be
 * valid if another thread has pushed to or popped from the queue
 * before the value returned by the method is acted on.  It is
 * provided as a utility, but may not be meaningful, depending on the
 * intended usage.
 */
  bool empty() const {
    Thread::Mutex::Lock lock(mutex);
    return q.empty();
  }
 
/**
 * @return The number of items currently in the queue.  It will not
 * throw assuming that the size() method of the container type does
 * not throw, as it will not on any sane implementation.
 * @note This method is thread safe, but the return value may not be
 * valid if another thread has pushed to or popped from the queue
 * before the value returned by the method is acted on.  It is
 * provided as a utility, but may not be meaningful, depending on the
 * intended usage.
 *
 * Since 1.2.22
 */
  size_type size() const {
    Thread::Mutex::Lock lock(mutex);
    return q.size();
  }
 
/**
 * Swaps the contents of 'this' and 'other'.  It will not throw
 * assuming that the swap method of the container type does not throw
 * (which the C++ standard requires not to happen with the standard
 * sequence containers).  It is thread safe and the swap is
 * thread-wise atomic.  A non-class function
 * Cgu::swap(Cgu::AsyncQueue&, Cgu::AsyncQueue&) method is also
 * provided which will call this method.
 * @param other The object to be swapped with this one.
 *
 * Since 1.2.22
 */
  void swap(AsyncQueue& other) {
    if (this != &other) {
      lock2(mutex, other.mutex); // doesn't throw
      Thread::Mutex::Lock l1(mutex, Thread::locked);
      Thread::Mutex::Lock l2(other.mutex, Thread::locked);
      std::swap(q, other.q);
    }
  }
 
/**
 * The assignment operator is strongly exception safe with the
 * standard sequence containers (it uses copy and swap).  It is also
 * thread safe, as it safely locks both the assignor's and assignee's
 * mutex to provide a thread-wise atomic assignment.
 * @param rhs The assignor.
 * @return The AsyncQueue object after assignment.
 * @exception std::bad_alloc The copy constructor of the queue's
 * container type, and so this assignment operator, might throw
 * std::bad_alloc if memory is exhausted and the system throws in that
 * case.  This assignment operator will also throw if the copy
 * constructor of the queue's container type throws any other
 * exceptions, including if any copy or move constructor or copy or
 * move assignment operator of a contained item throws.
 * @exception Thread::MutexError The assignment operator might throw
 * Thread::MutexError if initialization of a transitional object's
 * contained mutex fails.  (It is often not worth checking for this,
 * as it means either memory is exhausted or pthread has run out of
 * other resources to create new mutexes.)
 *
 * Since 1.2.22
 */
  AsyncQueue& operator=(const AsyncQueue& rhs) {
    if (this != &rhs) {
      lock2(mutex, rhs.mutex); // doesn't throw
      Thread::Mutex::Lock l1(mutex, Thread::locked);
      Thread::Mutex::Lock l2(rhs.mutex, Thread::locked);
      std::queue<T, Container> temp(rhs.q);
      std::swap(q, temp);
    }
    return *this;
  }
 
/**
 * @exception std::bad_alloc The default constructor might throw
 * std::bad_alloc if memory is exhausted and the system throws in that
 * case.
 * @exception Thread::MutexError The default constructor might throw
 * Thread::MutexError if initialisation of the contained mutex fails.
 * (It is often not worth checking for this, as it means either memory
 * is exhausted or pthread has run out of other resources to create
 * new mutexes.)
*/
  AsyncQueue() {}
 
/**
 * The copy constructor is thread safe, as it locks the initializing
 * object's mutex to obtain a consistent view of it.
 * @param rhs The AsyncQueue object to be copied.
 * @exception std::bad_alloc The copy constructor of the queue's
 * container type, and so this constructor, might throw std::bad_alloc
 * if memory is exhausted and the system throws in that case.  It will
 * also throw if the copy constructor of the queue's container type
 * throws any other exceptions, including if any copy or move
 * constructor or copy or move assignment operator of a contained item
 * throws.
 * @exception Thread::MutexError The copy constructor might throw
 * Thread::MutexError if initialization of the contained mutex fails.
 * (It is often not worth checking for this, as it means either memory
 * is exhausted or pthread has run out of other resources to create
 * new mutexes.)
 *
 * Since 1.2.22
 */
  // we use the comma operator here to lock the mutex and call the
  // copy constructor: the lock will be retained until the end of the
  // full expression in which it is lexically situated, namely until
  // the end of q's constructor - see C++98 1.9/12 and 12.2/3
  AsyncQueue(const AsyncQueue& rhs): q((Thread::Mutex::Lock(rhs.mutex), rhs.q)) {}
 
/**
 * The destructor does not throw unless the destructor of a contained
 * item throws.  It is thread safe (any thread may delete the
 * AsyncQueue object).
 */
  ~AsyncQueue() {
    // lock and unlock the mutex in the destructor so that we have an
    // acquire operation to ensure that when the std::queue object is
    // destroyed memory is synchronised, so any thread may destroy the
    // AsyncQueue object
    Thread::Mutex::Lock lock(mutex);
  }
 
/* Only has effect if --with-glib-memory-slices-compat or
 * --with-glib-memory-slices-no-compat option picked */
  CGU_GLIB_MEMORY_SLICES_FUNCS
};
 
/**
 * @class AsyncQueueDispatch async_queue.h c++-gtk-utils/async_queue.h
 * @brief A thread-safe asynchronous queue with a blocking pop()
 * method.
 * @sa AsyncQueue AsyncChannel AsyncResult
 *
 * AsyncQueueDispatch is similar to the AsyncQueue class, except that
 * it has a blocking pop_dispatch() method, which allows it to be
 * waited on by a dedicated event/message dispatching thread for
 * incoming work (represented by the data pushed onto the queue).  In
 * the same way, it can be used to implement thread pools, by having
 * threads in the pool waiting on the queue.  The AsyncResult class
 * can be useful for passing results between threads in conjunction
 * with AsyncQueueDispatch (the documentation on AsyncResult gives an
 * example).
 *
 * By default the queue uses a std::list object as its container
 * because when adding an item to the queue all allocation can take
 * place outside the queue object's mutex.  However, for types which
 * have low overhead copy constructors, this can be changed to, say, a
 * std::deque object by specifying it as the second template
 * parameter.
 *
 * If data pushed and popped from the queue are held by a reference
 * counted smart pointer, the reference count must be thread-safe,
 * such as by using SharedLockPtr or IntrusiveLockCounter.
 *
 * If the library is installed using the
 * \--with-glib-memory-slices-compat or
 * \--with-glib-memory-slices-no-compat configuration options, any
 * AsyncQueueDispatch objects constructed on free store will be
 * constructed in glib memory slices.  This does not affect the queue
 * container itself: to change the allocator of the C++ container, a
 * custom allocator type can be provided when the AsyncQueueDispatch
 * object is instantiated offering the standard allocator interface.
 * If glib memory slices are not used or no AsyncQueueDispatch objects
 * are constructed on free store, it is not necessary to call
 * g_thread_init() before manipulating or using an AsyncQueueDispatch
 * object in multiple threads, but prior to glib version 2.32 glib
 * itself (and thus glib memory slices) are not thread safe unless
 * that function has been called.
 */
 
template <class T, class Container = std::list<T> > class AsyncQueueDispatch {
public:
  typedef typename Container::value_type value_type;
  typedef typename Container::size_type size_type;
  typedef Container container_type;
private:
  std::queue<T, Container> q;
  mutable Thread::Mutex mutex;
  Thread::Cond cond;
 
// TODO: at the next ABI break make this method explicitly static
// this method won't throw: it is for the user to ensure the arguments
// do not refer to the same mutex object
  void lock2(Thread::Mutex& m1, Thread::Mutex& m2) {
    m1.lock();
    for(;;) {
      if (!m2.trylock()) {
    return;
      }
      m1.unlock();
      // spin nicely
#ifdef CGU_USE_SCHED_YIELD
      sched_yield();
#else
      usleep(10);
#endif
      m1.lock();
    }
  }
public:
/**
 * Pushes an item onto the queue.  This method has strong exception
 * safety if the container is a std::list or std::deque container (the
 * default is std::list), except that if std::deque is used as the
 * container under C++11 and the copy constructor, move constructor,
 * assignment operator or move assignment operator of the queue item
 * throws, it only gives the basic exception guarantee (and the basic
 * guarantee is not given by std::deque under C++11 if the queue
 * item's move constructor throws and it uses a non-default allocator
 * which does not provide for it to be CopyInsertable).  It is thread
 * safe.
 * @param obj The item to be pushed onto the queue.
 * @exception std::bad_alloc The method might throw std::bad_alloc if
 * memory is exhausted and the system throws in that case.  It might
 * also throw if the copy constructor or assignment operator of the
 * queue item might throw (or under C++11 if its move constructor or
 * move assignment operator throws).
 */
  void push(const value_type& obj) {
    Thread::Mutex::Lock lock(mutex);
    q.push(obj);
    cond.signal();
  }
 
/**
 * Pops an item from the queue.  This method has strong exception
 * safety if the container is a std::deque or std::list container (the
 * default is std::list), provided the destructor of a contained item
 * does not throw.  It is thread safe.
 * @param obj A value type reference to which the item at the front of
 * the queue will be assigned.
 * @exception AsyncQueuePopError If the queue is empty when a pop is
 * attempted, this method will throw AsyncQueuePopError.  It might
 * also throw if the assignment operator of the queue item might
 * throw.  In order to complete pop() operations atomically under a
 * single lock and to retain strong exception safety, the object into
 * which the pop()ed data is to be placed is passed as an argument by
 * reference (this avoids a copy from a temporary object after the
 * data has been extracted from the queue, which would occur if the
 * item extracted were returned by value).  It might also throw if the
 * destructor of the queue item might throw (but that should never
 * happen), or if the empty() method of the container type throws
 * (which would not happen on any sane implementation).
 */
  void pop(value_type& obj) {
    Thread::Mutex::Lock lock(mutex);
    if (q.empty()) throw AsyncQueuePopError();
    obj = q.front();
    q.pop();
  }
 
/**
 * Pops an item from the queue.  If the queue is empty, it will block
 * until an item becomes available.  If it blocks, the wait comprises
 * a cancellation point.  This method is cancellation safe if the
 * stack unwinds on cancellation, as cancellation is blocked while the
 * queue is being operated on after coming out of a wait.  This method
 * has strong exception safety if the container is a std::deque or
 * std::list container (the default is std::list), provided the
 * destructor of a contained item does not throw.  It is thread safe.
 * @param obj A value type reference to which the item at the front of
 * the queue will be assigned.  This method might throw if the
 * assignment operator of the queue item might throw.  In order to
 * complete pop() operations atomically under a single lock and to
 * retain strong exception safety, the object into which the pop()ed
 * data is to be placed is passed as an argument by reference (this
 * avoids a copy from a temporary object after the data has been
 * extracted from the queue, which would occur if the item extracted
 * were returned by value).  It might also throw if the destructor of
 * the queue item might throw (but that should never happen), or if
 * the empty() method of the container type throws (which would not
 * happen on any sane implementation).
 */
  void pop_dispatch(value_type& obj) {
    Thread::Mutex::Lock lock(mutex);
    while (q.empty()) cond.wait(mutex);
    Thread::CancelBlock b;
    obj = q.front();
    q.pop();
  }    
 
/**
 * Pops an item from the queue.  If the queue is empty, it will block
 * until an item becomes available or until the timeout expires.  If
 * it blocks, the wait comprises a cancellation point.  This method is
 * cancellation safe if the stack unwinds on cancellation, as
 * cancellation is blocked while the queue is being operated on after
 * coming out of a wait.  This method has strong exception safety if
 * the container is a std::deque or std::list container (the default
 * is std::list), provided the destructor of a contained item does not
 * throw.  It is thread safe.
 * @param obj A value type reference to which the item at the front of
 * the queue will be assigned.  This method might throw if the
 * assignment operator of the queue item might throw.  In order to
 * complete pop() operations atomically under a single lock and to
 * retain strong exception safety, the object into which the pop()ed
 * data is to be placed is passed as an argument by reference (this
 * avoids a copy from a temporary object after the data has been
 * extracted from the queue, which would occur if the item extracted
 * were returned by value).  It might also throw if the destructor of
 * the queue item might throw (but that should never happen), or if
 * the empty() method of the container type throws (which would not
 * happen on any sane implementation).
 * @param millisec The timeout interval, in milliseconds.
 * @return If the timeout expires without an item becoming available,
 * the method will return true.  If an item from the queue is
 * extracted, it returns false.
 */
  bool pop_timed_dispatch(value_type& obj, unsigned int millisec) {
    timespec ts;
    Thread::Cond::get_abs_time(ts, millisec);
    Thread::Mutex::Lock lock(mutex);
    while (q.empty()) {
      if (cond.timed_wait(mutex, ts)) return true;
    }
    Thread::CancelBlock b;
    obj = q.front();
    q.pop();
    return false;
  }
 
/**
 * Discards the item at the front of the queue.  This method has
 * strong exception safety if the container is a std::deque or
 * std::list container (the default is std::list), provided the
 * destructor of a contained item does not throw.  It is thread safe.
 * @exception AsyncQueuePopError If the queue is empty when a pop is
 * attempted, this method will throw AsyncQueuePopError.  It might
 * also throw if the destructor of the queue item might throw (but
 * that should never happen), or if the empty() method of the
 * container type throws (which would not happen on any sane
 * implementation).
 */
  void pop() {
    Thread::Mutex::Lock lock(mutex);
    if (q.empty()) throw AsyncQueuePopError();
    q.pop();
  }
 
/**
 * @return Whether the queue is empty.  It will not throw assuming
 * that the empty() method of the container type does not throw, as it
 * will not on any sane implementation.
 * @note This method is thread safe, but the return value may not be
 * valid if another thread has pushed to or popped from the queue
 * before the value returned by the method is acted on.  It is
 * provided as a utility, but may not be meaningful, depending on the
 * intended usage.
 */
  bool empty() const {
    Thread::Mutex::Lock lock(mutex);
    return q.empty();
  }
 
/**
 * @return The number of items currently in the queue.  It will not
 * throw assuming that the size() method of the container type does
 * not throw, as it will not on any sane implementation.
 * @note This method is thread safe, but the return value may not be
 * valid if another thread has pushed to or popped from the queue
 * before the value returned by the method is acted on.  It is
 * provided as a utility, but may not be meaningful, depending on the
 * intended usage.
 *
 * Since 1.2.22
 */
  size_type size() const {
    Thread::Mutex::Lock lock(mutex);
    return q.size();
  }
 
/**
 * Swaps the contents of 'this' and 'other'.  It will not throw
 * assuming that the swap method of the container type does not throw
 * (which the C++ standard requires not to happen with the standard
 * sequence containers).  It is thread safe and the swap is
 * thread-wise atomic.  A non-class function
 * Cgu::swap(Cgu::AsyncQueueDispatch&, Cgu::AsyncQueueDispatch&)
 * method is also provided which will call this method.
 * @param other The object to be swapped with this one.
 * @note An object swapped does not, by virtue of the swap, inherit
 * any threads waiting on the other one.  However if threads were
 * waiting on a swapped object prior to the swap, and it acquires
 * items by virtue of the swap, the waiting threads will unblock and
 * extract those items.
 *
 * Since 1.2.22
 */
  void swap(AsyncQueueDispatch& other) {
    if (this != &other) {
      lock2(mutex, other.mutex); // doesn't throw
      Thread::Mutex::Lock l1(mutex, Thread::locked);
      Thread::Mutex::Lock l2(other.mutex, Thread::locked);
      std::swap(q, other.q);
      if (!q.empty()) cond.broadcast();
      if (!other.q.empty()) other.cond.broadcast();
    }
  }
 
/**
 * The assignment operator is strongly exception safe with the
 * standard sequence containers (it uses copy and swap).  It is also
 * thread safe, as it safely locks both the assignor's and assignee's
 * mutex to provide a thread-wise atomic assignment.
 * @param rhs The assignor.
 * @return The AsyncQueueDispatch object after assignment.
 * @exception std::bad_alloc The copy constructor of the queue's
 * container type, and so this assignment operator, might throw
 * std::bad_alloc if memory is exhausted and the system throws in that
 * case.  This assignment operator will also throw if the copy
 * constructor of the queue's container type throws any other
 * exceptions, including if any copy or move constructor or copy or
 * move assignment operator of a contained item throws.
 * @exception Thread::MutexError The assignment operator might throw
 * Thread::MutexError if initialization of a transitional object's
 * contained mutex fails.  (It is often not worth checking for this,
 * as it means either memory is exhausted or pthread has run out of
 * other resources to create new mutexes.)
 * @exception Thread::CondError The assignment operator might throw
 * this exception if initialisation of a transitional object's
 * contained condition variable fails.  (It is often not worth
 * checking for this, as it means either memory is exhausted or
 * pthread has run out of other resources to create new condition
 * variables.)
 * @note The assignee does not, by virtue of the assignment, inherit
 * any threads waiting on the assignor.  However, if prior to the
 * assignment threads were waiting on the assignee and the assignee
 * acquires items from the assignor as a result of the assignment, the
 * waiting threads will unblock and extract those items.
 *
 * Since 1.2.22
 */
  AsyncQueueDispatch& operator=(const AsyncQueueDispatch& rhs) {
    if (this != &rhs) {
      lock2(mutex, rhs.mutex); // doesn't throw
      Thread::Mutex::Lock l1(mutex, Thread::locked);
      Thread::Mutex::Lock l2(rhs.mutex, Thread::locked);
      std::queue<T, Container> temp(rhs.q);
      std::swap(q, temp);
      if (!q.empty()) cond.broadcast();
    }
    return *this;
  }
 
/**
 * @exception std::bad_alloc The default constructor might throw this
 * exception if memory is exhausted and the system throws in that
 * case.
 * @exception Thread::MutexError The default constructor might throw
 * this exception if initialisation of the contained mutex fails.  (It
 * is often not worth checking for this, as it means either memory is
 * exhausted or pthread has run out of other resources to create new
 * mutexes.)
 * @exception Thread::CondError The default constructor might throw
 * this exception if initialisation of the contained condition
 * variable fails.  (It is often not worth checking for this, as it
 * means either memory is exhausted or pthread has run out of other
 * resources to create new condition variables.)
 */
  AsyncQueueDispatch() {}
 
/**
 * The copy constructor is thread safe, as it locks the initializing
 * object's mutex to obtain a consistent view of it.
 * @param rhs The AsyncQueue object to be copied.
 * @exception std::bad_alloc The copy constructor of the queue's
 * container type, and so this constructor, might throw std::bad_alloc
 * if memory is exhausted and the system throws in that case.  It will
 * also throw if the copy constructor of the queue's container type
 * throws any other exceptions, including if any copy or move
 * constructor or copy or move assignment operator of a contained item
 * throws.
 * @exception Thread::MutexError The copy constructor might throw
 * Thread::MutexError if initialization of the contained mutex fails.
 * (It is often not worth checking for this, as it means either memory
 * is exhausted or pthread has run out of other resources to create
 * new mutexes.)
 * @exception Thread::CondError The copy constructor might throw this
 * exception if initialisation of the contained condition variable
 * fails.  (It is often not worth checking for this, as it means
 * either memory is exhausted or pthread has run out of other
 * resources to create new condition variables.)
 *
 * Since 1.2.22
 */
  // we use the comma operator here to lock the mutex and call the
  // copy constructor: the lock will be retained until the end of the
  // full expression in which it is lexically situated, namely until
  // the end of q's constructor - see C++98 1.9/12 and 12.2/3
  AsyncQueueDispatch(const AsyncQueueDispatch& rhs):
                        q((Thread::Mutex::Lock(rhs.mutex), rhs.q)) {}
 
/**
 * The destructor does not throw unless the destructor of a contained
 * item throws.  It is thread safe (any thread may delete the
 * AsyncQueueDispatch object).  Destroying an AsyncQueueDispatch
 * object on which another thread is currently blocked results in
 * undefined behavior.
 */
  ~AsyncQueueDispatch() {
    // lock and unlock the mutex in the destructor so that we have an
    // acquire operation to ensure that when the std::queue object is
    // destroyed memory is synchronised, so any thread may destroy the
    // AsyncQueueDispatch object
    Thread::Mutex::Lock lock(mutex);
  }
 
/* Only has effect if --with-glib-memory-slices-compat or
 * --with-glib-memory-slices-no-compat option picked */
  CGU_GLIB_MEMORY_SLICES_FUNCS
};
 
/**
 * Swaps the contents of two AsyncQueue objects.  It will not throw
 * assuming that the swap method of the container type does not throw
 * (which the C++ standard requires not to happen with the standard
 * sequence containers).  It is thread safe and the swap is
 * thread-wise atomic.
 * @param q1 An object to be swapped with the other.
 * @param q2 An object to be swapped with the other.
 * @note Calling std::swap on AsyncQueue objects is thread safe but
 * does not provide a thread-wise atomic swap (the swapped objects may
 * not be mirror images if during the execution of std::swap's default
 * algorithm one of them has been modified), although in many cases
 * that doesn't matter.  If swap() is called without a namespace
 * qualifier, argument dependent look-up will pick this one correctly.
 *
 * Since 1.2.22
 */
template <class T, class Container>
void swap(Cgu::AsyncQueue<T, Container>& q1,
      Cgu::AsyncQueue<T, Container>& q2) {
  q1.swap(q2);
}
 
/**
 * Swaps the contents of two AsyncQueueDispatch objects.  It will not
 * throw assuming that the swap method of the container type does not
 * throw (which the C++ standard requires not to happen with the
 * standard sequence containers).  It is thread safe and the swap is
 * thread-wise atomic.
 * @param q1 An object to be swapped with the other.
 * @param q2 An object to be swapped with the other.
 * @note 1. An object swapped does not, by virtue of the swap, inherit
 * any threads waiting on the other one.  However if threads were
 * waiting on a swapped object prior to the swap, and it acquires
 * items by virtue of the swap, the waiting threads will unblock and
 * extract those items.
 * @note 2. Calling std::swap on AsyncQueueDispatch objects is thread
 * safe but does not provide a thread-wise atomic swap (the swapped
 * objects may not be mirror images if during the execution of
 * std::swap's default algorithm one of them has been modified),
 * although in many cases that doesn't matter.  If swap() is called
 * without a namespace qualifier, argument dependent look-up will pick
 * this one correctly.
 *
 * Since 1.2.22
 */
template <class T, class Container>
void swap(Cgu::AsyncQueueDispatch<T, Container>& q1,
      Cgu::AsyncQueueDispatch<T, Container>& q2) {
  q1.swap(q2);
}
 
#if defined(CGU_USE_INHERITABLE_QUEUE) && !defined(DOXYGEN_PARSING)
 
/* This is a specialization of AsyncQueue for std::list objects, which
   uses std::list::splice() to push a new item on the queue.  This
   means that allocation for the push occurs outside the mutex, so
   reducing contention (a tip from a talk by Sean Parent of Adobe).
   This is first available in version 1.2.32.
 */
template <class T, class Allocator>
class AsyncQueue<T, std::list<T, Allocator> > {
public:
  typedef std::list<T, Allocator> Container;
  typedef typename Container::value_type value_type;
  typedef typename Container::size_type size_type;
  typedef Container container_type;
private:
  // 23.2.3.1 of C++98 requires std::queue to have a protected
  // container member called 'c' for the purposes of derivation.  This
  // specialisation will have the same binary layout as the
  // unspecialized version on any practical implementation: all we do
  // is add a splice_end() member
  class Q: public std::queue<T, Container> {
   public:
    void splice_end(Container& lst) {
      this->c.splice(this->c.end(), lst);
    }
  } q;
  mutable Thread::Mutex mutex;
 
// TODO: at the next ABI break make this method explicitly static
  void lock2(Thread::Mutex& m1, Thread::Mutex& m2) {
    m1.lock();
    for(;;) {
      if (!m2.trylock()) {
    return;
      }
      m1.unlock();
      // spin nicely
#ifdef CGU_USE_SCHED_YIELD
      sched_yield();
#else
      usleep(10);
#endif
      m1.lock();
    }
  }
public:
  void push(const value_type& obj) {
    Container temp;
    temp.push_back(obj);
    Thread::Mutex::Lock lock(mutex);
    q.splice_end(temp);
  }
 
  void pop(value_type& obj) {
    Thread::Mutex::Lock lock(mutex);
    if (q.empty()) throw AsyncQueuePopError();
    obj = q.front();
    q.pop();
  }
 
  void pop() {
    Thread::Mutex::Lock lock(mutex);
    if (q.empty()) throw AsyncQueuePopError();
    q.pop();
  }
 
  bool empty() const {
    Thread::Mutex::Lock lock(mutex);
    return q.empty();
  }
 
  size_type size() const {
    Thread::Mutex::Lock lock(mutex);
    return q.size();
  }
 
  void swap(AsyncQueue& other) {
    if (this != &other) {
      lock2(mutex, other.mutex); // doesn't throw
      Thread::Mutex::Lock l1(mutex, Thread::locked);
      Thread::Mutex::Lock l2(other.mutex, Thread::locked);
      std::swap(q, other.q);
    }
  }
 
  AsyncQueue& operator=(const AsyncQueue& rhs) {
    if (this != &rhs) {
      lock2(mutex, rhs.mutex); // doesn't throw
      Thread::Mutex::Lock l1(mutex, Thread::locked);
      Thread::Mutex::Lock l2(rhs.mutex, Thread::locked);
      Q temp(rhs.q);
      std::swap(q, temp);
    }
    return *this;
  }
 
  AsyncQueue() {}
 
  AsyncQueue(const AsyncQueue& rhs): q((Thread::Mutex::Lock(rhs.mutex), rhs.q)) {}
 
  ~AsyncQueue() {
    Thread::Mutex::Lock lock(mutex);
  }
 
  CGU_GLIB_MEMORY_SLICES_FUNCS
};
 
/* This is a specialization of AsyncQueueDispatch for std::list
   objects, which uses std::list::splice() to push a new item on the
   queue.  This means that allocation for the push occurs outside the
   mutex, so reducing contention (a tip from a talk by Sean Parent of
   Adobe).  This is first available in version 1.2.32.
 */
template <class T, class Allocator>
class AsyncQueueDispatch<T, std::list<T, Allocator> > {
public:
  typedef std::list<T, Allocator> Container;
  typedef typename Container::value_type value_type;
  typedef typename Container::size_type size_type;
  typedef Container container_type;
private:
  // 23.2.3.1 of C++98 requires std::queue to have a protected
  // container member called 'c' for the purposes of derivation.  This
  // specialisation will have the same binary layout as the
  // unspecialized version on any practical implementation: all we do
  // is add a splice_end() member
  class Q: public std::queue<T, Container> {
   public:
    void splice_end(Container& lst) {
      this->c.splice(this->c.end(), lst);
    }
  } q;
  mutable Thread::Mutex mutex;
  Thread::Cond cond;
 
// TODO: at the next ABI break make this method explicitly static
  void lock2(Thread::Mutex& m1, Thread::Mutex& m2) {
    m1.lock();
    for(;;) {
      if (!m2.trylock()) {
    return;
      }
      m1.unlock();
      // spin nicely
#ifdef CGU_USE_SCHED_YIELD
      sched_yield();
#else
      usleep(10);
#endif
      m1.lock();
    }
  }
public:
  void push(const value_type& obj) {
    Container temp;
    temp.push_back(obj);
    Thread::Mutex::Lock lock(mutex);
    q.splice_end(temp);
    cond.signal();
  }
 
  void pop(value_type& obj) {
    Thread::Mutex::Lock lock(mutex);
    if (q.empty()) throw AsyncQueuePopError();
    obj = q.front();
    q.pop();
  }
 
  void pop_dispatch(value_type& obj) {
    Thread::Mutex::Lock lock(mutex);
    while (q.empty()) cond.wait(mutex);
    Thread::CancelBlock b;
    obj = q.front();
    q.pop();
  }    
 
  bool pop_timed_dispatch(value_type& obj, unsigned int millisec) {
    timespec ts;
    Thread::Cond::get_abs_time(ts, millisec);
    Thread::Mutex::Lock lock(mutex);
    while (q.empty()) {
      if (cond.timed_wait(mutex, ts)) return true;
    }
    Thread::CancelBlock b;
    obj = q.front();
    q.pop();
    return false;
  }
 
  void pop() {
    Thread::Mutex::Lock lock(mutex);
    if (q.empty()) throw AsyncQueuePopError();
    q.pop();
  }
 
  bool empty() const {
    Thread::Mutex::Lock lock(mutex);
    return q.empty();
  }
 
  size_type size() const {
    Thread::Mutex::Lock lock(mutex);
    return q.size();
  }
 
  void swap(AsyncQueueDispatch& other) {
    if (this != &other) {
      lock2(mutex, other.mutex); // doesn't throw
      Thread::Mutex::Lock l1(mutex, Thread::locked);
      Thread::Mutex::Lock l2(other.mutex, Thread::locked);
      std::swap(q, other.q);
      if (!q.empty()) cond.broadcast();
      if (!other.q.empty()) other.cond.broadcast();
    }
  }
 
  AsyncQueueDispatch& operator=(const AsyncQueueDispatch& rhs) {
    if (this != &rhs) {
      lock2(mutex, rhs.mutex); // doesn't throw
      Thread::Mutex::Lock l1(mutex, Thread::locked);
      Thread::Mutex::Lock l2(rhs.mutex, Thread::locked);
      Q temp(rhs.q);
      std::swap(q, temp);
      if (!q.empty()) cond.broadcast();
    }
    return *this;
  }
 
  AsyncQueueDispatch() {}
 
  AsyncQueueDispatch(const AsyncQueueDispatch& rhs):
                        q((Thread::Mutex::Lock(rhs.mutex), rhs.q)) {}
 
  ~AsyncQueueDispatch() {
    Thread::Mutex::Lock lock(mutex);
  }
 
  CGU_GLIB_MEMORY_SLICES_FUNCS
};
 
#endif // CGU_USE_INHERITABLE_QUEUE
 
} // namespace Cgu
 
#endif