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) and AsyncQueue::move_pop(value_type& obj) methods provide the
 * popped 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 pushing and popping from different threads.  It is
 * therefore useful for passing data between threads, perhaps in
 * response to a signal being emitted from a Notifier object.  Data
 * can be pushed or pulled by move constructor/move assignment
 * operator or by means of a std::unique_ptr object, SharedLockPtr
 * object or an IntrusivePtr object referencing data derived from
 * IntrusiveLockCounter.
 *
 * AsyncQueueDispatch is a class which has blocking pop() and
 * move_pop() methods, 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 in the kind of use mentioned above they are unlikely to
 * hold many objects but they can be changed to, say, a std::deque
 * object by specifying it as the second template parameter.
 */

#ifndef CGU_ASYNC_QUEUE_H
#define CGU_ASYNC_QUEUE_H

#include <queue>
#include <list>
#include <exception>
#include <utility>    // for std::move and std::forward
#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 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) and AsyncQueue::move_pop(value_type& obj) methods provide the
 * popped 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 pushing and popping from different threads.  It is
 * therefore useful for passing data between threads, perhaps in
 * response to a signal being emitted from a Notifier object.  Data
 * can be pushed or pulled by move constructor/move assignment
 * operator or by means of a std::unique_ptr object, SharedLockPtr
 * object or an IntrusivePtr object referencing data derived from
 * IntrusiveLockCounter.
 *
 * By default the queue uses a std::list object as its container
 * because in the kind of use mentioned above it is unlikely to hold
 * many objects but it can be changed to, say, a std::deque object by
 * specifying it as the second template parameter.
 *
 * 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:
  mutable Thread::Mutex mutex;
  std::queue<T, Container> q;

// this 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 and the copy constructor, move constructor, copy
 * 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 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, move constructor, assignment
 * operator or move assignment operator of the queue item might throw.
 */
  void push(const value_type& obj) {
    Thread::Mutex::Lock lock{mutex};
    q.push(obj);
  }

/**
 * 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 and the copy constructor, move constructor, copy
 * 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 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, move constructor, assignment
 * operator or move assignment operator of the queue item might throw.
 *
 * Since 2.0.0-rc5
 */
  void push(value_type&& obj) {
    Thread::Mutex::Lock lock{mutex};
    q.push(std::move(obj));
  }

/**
 * Pushes an item onto the queue by constructing it in place: that is,
 * by passing to this method the item's constructor's arguments,
 * rather than the item itself.  This method has strong exception
 * safety if the container is a std::list or std::deque container (the
 * default is std::list).  (Technically, for a std::deque container,
 * emplace() only offers the same exception guarantees as does push(),
 * namely only the basic guarantee where a copy or move of the queue
 * item throws during the call, but the purpose of emplace is to
 * construct in place and any reasonable implementation will not copy
 * or move the queue item.)  It is thread safe.
 * @param args The constructor arguments for 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 item's constructor (including any of its
 * constructor arguments) might throw when constructing the item.
 * @note The constructor of the item pushed onto the queue must not
 * access any of the methods of the same queue object, or a deadlock
 * might occur.
 *
 * Since 2.0.0-rc5
 */
  template<class... Args>
  void emplace(Args&&... args) {
    Thread::Mutex::Lock lock{mutex};
    q.emplace(std::forward<Args>(args)...);
  }

/**
 * 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 copy 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 popped 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 using the contained type's move
 * assignment operator, if it has one.  This method is identical to
 * the pop() method if that type has no move assignment operator.
 * 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 and the
 * move assignment operator of a contained item has strong exception
 * safety.  It is thread safe.  Use this method in preference to the
 * pop() method if it is known that the contained items' move
 * assignment operator does not throw or is strongly exception safe,
 * or if the use case does not require strong exception safety.  This
 * method must be used in place of the pop() method if the contained
 * type has a move assignment operator but no copy assignment operator
 * (such as a std::unique_ptr object).
 * @param obj A value type reference to which the item at the front of
 * the queue will be move assigned.
 * @exception AsyncQueuePopError If the queue is empty when a pop is
 * attempted, this method will throw AsyncQueuePopError.  It might
 * also throw if the move assignment operator of the queue item might
 * throw, or if it has no move assignment operator and its copy
 * assignment operator throws.  In order to complete pop operations
 * atomically under a single lock and to retain strong exception
 * safety, the object into which the popped data is to be placed is
 * passed as an argument by reference (this avoids a move 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).
 *
 * Since 2.0.11
 */
  void move_pop(value_type& obj) {
    Thread::Mutex::Lock lock{mutex};
    if (q.empty()) throw AsyncQueuePopError();
    obj = std::move(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 2.0.8
 */
  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++11 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 2.0.8
 */
  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};
      q.swap(other.q);
    }
  }

/**
 * The copy 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 This 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 2.0.8
 */
  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};
      q.swap(temp);
    }
    return *this;
  }

/**
 * This move assignment operator is thread safe as regards the
 * assignee (the object moved to), but no synchronization is carried
 * out with respect to the rvalue assignor/movant.  This is because
 * temporaries are only visible and accessible in the thread carrying
 * out the move operation and synchronization for them would represent
 * pointless overhead.  In a case where the user uses std::move to
 * force a move from a named object, and that named object's lifetime
 * is managed by (or the object is otherwise accessed by) a different
 * thread than the one making the move, the user must carry out her
 * own synchronization with respect to that different thread, both to
 * ensure that a consistent view of the the named object is obtained
 * and because that object will be mutated by the move.  This method
 * invokes std::queue's move assignment operator, and therefore has
 * the same exception safety as the standard library's implementation
 * of that operator.  It will not normally throw unless a custom
 * allocator is used which throws on move assignment, or the
 * destructor of a contained item throws.
 * @param rhs The assignor/movant.
 * @return The AsyncQueue object after move assignment.
 *
 * Since 2.0.8
 */
  AsyncQueue& operator=(AsyncQueue&& rhs) {
    Thread::Mutex::Lock lock{mutex};
    q = std::move(rhs.q);
    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 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.)
 */
  AsyncQueue() = default;

/**
 * As regards thread safety, the move constructor does not synchronize
 * with respect to the initializing rvalue.  This is because
 * temporaries are only visible and accessible in the thread carrying
 * out the move operation and synchronization for them would represent
 * pointless overhead.  In a case where a user uses std::move to force
 * a move from a named object, and that named object's lifetime is
 * managed by (or the object is otherwise accessed by) a different
 * thread than the one making the move, the user must carry out her
 * own synchronization with respect to that different thread, both to
 * ensure that a consistent view of the the named object is obtained
 * and because that object will be mutated by the move.
 * @param rhs The AsyncQueue object to be moved.
 * @exception Thread::MutexError The move constructor might throw
 * Thread::MutexError if initialization of the contained mutex fails.
 * If it does so this move constructor is strongly exception safe (if
 * it is thrown, the object passed as an argument will be unchanged).
 * (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.)  It might also throw if the queue's container type's
 * move constructor might throw, but it should not do that unless a
 * custom allocator is in use.
 *
 * Since 2.0.8
 */
  AsyncQueue(AsyncQueue&& rhs): q(std::move(rhs.q)) {}

/**
 * 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 2.0.8
 */
  // 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++11 1.9/10 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 AsyncResult
 *
 * AsyncQueueDispatch is a class which has blocking pop_dispatch()
 * and move_pop_dispatch() methods, 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 in the kind of use mentioned above it is unlikely to hold
 * many objects but it can be changed to, say, a std::deque object by
 * specifying it as the second template parameter.
 *
 * 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:
  mutable Thread::Mutex mutex;
  Thread::Cond cond;
  std::queue<T, Container> q;

// this 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 and the copy constructor, move constructor, copy
 * 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 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, move constructor, assignment
 * operator or move assignment operator of the queue item might throw.
 */
  void push(const value_type& obj) {
    Thread::Mutex::Lock lock{mutex};
    q.push(obj);
    cond.signal();
  }

/**
 * 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 and the copy constructor, move constructor, copy
 * 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 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, move constructor, assignment
 * operator or move assignment operator of the queue item might throw.
 *
 * Since 2.0.0-rc5
 */
  void push(value_type&& obj) {
    Thread::Mutex::Lock lock{mutex};
    q.push(std::move(obj));
    cond.signal();
  }

/**
 * Pushes an item onto the queue by constructing it in place: that is,
 * by passing to this method the item's constructor's arguments,
 * rather than the item itself.  This method has strong exception
 * safety if the container is a std::list or std::deque container (the
 * default is std::list).  (Technically, for a std::deque container,
 * emplace() only offers the same exception guarantees as does push(),
 * namely only the basic guarantee where a copy or move of the queue
 * item throws during the call, but the purpose of emplace is to
 * construct in place and any reasonable implementation will not copy
 * or move the queue item.)  It is thread safe.
 * @param args The constructor arguments for 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 item's constructor (including any of its
 * constructor arguments) might throw when constructing the item.
 * @note The constructor of the item pushed onto the queue must not
 * access any of the methods of the same queue object, or a deadlock
 * might occur.
 *
 * Since 2.0.0-rc5
 */
  template<class... Args>
  void emplace(Args&&... args) {
    Thread::Mutex::Lock lock{mutex};
    q.emplace(std::forward<Args>(args)...);
    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 copy 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 popped 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 using the contained type's move
 * assignment operator, if it has one.  This method is identical to
 * the pop() method if that type has no move assignment operator.
 * 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 and the
 * move assignment operator of a contained item has strong exception
 * safety.  It is thread safe.  Use this method in preference to the
 * pop() method if it is known that the contained items' move
 * assignment operator does not throw or is strongly exception safe,
 * or if the use case does not require strong exception safety.  This
 * method must be used in place of the pop() method if the contained
 * type has a move assignment operator but no copy assignment operator
 * (such as a std::unique_ptr object).
 * @param obj A value type reference to which the item at the front of
 * the queue will be move assigned.
 * @exception AsyncQueuePopError If the queue is empty when a pop is
 * attempted, this method will throw AsyncQueuePopError.  It might
 * also throw if the move assignment operator of the queue item might
 * throw, or if it has no move assignment operator and its copy
 * assignment operator throws.  In order to complete pop operations
 * atomically under a single lock and to retain strong exception
 * safety, the object into which the popped data is to be placed is
 * passed as an argument by reference (this avoids a move 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).
 *
 * Since 2.0.11
 */
  void move_pop(value_type& obj) {
    Thread::Mutex::Lock lock{mutex};
    if (q.empty()) throw AsyncQueuePopError();
    obj = std::move(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 copy
 * 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 popped
 * 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 using the contained type's move
 * assignment operator, if it has one (this method is identical to the
 * pop_dispatch() method if that type has no move assignment
 * operator).  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 and the move assignment operator of a
 * contained item has strong exception safety.  It is thread safe.
 * Use this method in preference to the pop_dispatch() method if it is
 * known that the contained items' move assignment operator does not
 * throw or is strongly exception safe, or if the use case does not
 * require strong exception safety.  This method must be used in place
 * of the pop_dispatch() method if the contained type has a move
 * assignment operator but no copy assignment operator (such as a
 * std::unique_ptr object).
 * @param obj A value type reference to which the item at the front of
 * the queue will be move assigned.  This method might throw if the
 * move assignment operator of the queue item might throw, or if it
 * has no move assignment operator and its copy assignment operator
 * throws.  In order to complete pop operations atomically under a
 * single lock and to retain strong exception safety, the object into
 * which the popped data is to be placed is passed as an argument by
 * reference (this avoids a move 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).
 *
 * Since 2.0.11
 */
  void move_pop_dispatch(value_type& obj) {
    Thread::Mutex::Lock lock{mutex};
    while (q.empty()) cond.wait(mutex);
    Thread::CancelBlock b;
    obj = std::move(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 copy
 * 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 popped
 * 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;
  }

/**
 * Pops an item from the queue using the contained type's move
 * assignment operator, if it has one (this method is identical to the
 * pop_timed_dispatch() method if that type has no move assignment
 * operator).  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 and the
 * move assignment operator of a contained item has strong exception
 * safety.  It is thread safe.  Use this method in preference to the
 * pop_timed_dispatch() method if it is known that the contained
 * items' move assignment operator does not throw or is strongly
 * exception safe, or if the use case does not require strong
 * exception safety.  This method must be used in place of the
 * pop_timed_dispatch() method if the contained type has a move
 * assignment operator but no copy assignment operator (such as a
 * std::unique_ptr object).
 * @param obj A value type reference to which the item at the front of
 * the queue will be move assigned.  This method might throw if the
 * move assignment operator of the queue item might throw, or if it
 * has no move assignment operator and its copy assignment operator
 * throws.  In order to complete pop operations atomically under a
 * single lock and to retain strong exception safety, the object into
 * which the popped data is to be placed is passed as an argument by
 * reference (this avoids a move 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.
 *
 * Since 2.0.11
 */
  bool move_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 = std::move(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 2.0.8
 */
  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++11 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.
 * @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 2.0.8
 */
  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};
      q.swap(other.q);
      if (!q.empty()) cond.broadcast();
      if (!other.q.empty()) other.cond.broadcast();
    }
  }

/**
 * The copy 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 This 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 This assignment operator might throw
 * Thread::CondError 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 2.0.8
 */
  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};
      q.swap(temp);
      if (!q.empty()) cond.broadcast();
    }
    return *this;
  }

/**
 * This move assignment operator is thread safe as regards the
 * assignee (the object moved to), but no synchronization is carried
 * out with respect to the rvalue assignor/movant.  This is because
 * temporaries are only visible and accessible in the thread carrying
 * out the move operation and synchronization for them would represent
 * pointless overhead.  In a case where the user uses std::move to
 * force a move from a named object, and that named object's lifetime
 * is managed by (or the object is otherwise accessed by) a different
 * thread than the one making the move, the user must carry out her
 * own synchronization with respect to that different thread, both to
 * ensure that a consistent view of the the named object is obtained
 * and because that object will be mutated by the move.  This method
 * invokes std::queue's move assignment operator, and therefore has
 * the same exception safety as the standard library's implementation
 * of that operator.  It will not normally throw unless a custom
 * allocator is used which throws on move assignment, or the
 * destructor of a contained item throws.
 * @param rhs The assignor/movant.
 * @return The AsyncQueueDispatch object after move assignment.
 * @note The assignee does not, by virtue of the move, inherit any
 * threads waiting on the assignor/movant.  However, if prior to the
 * move threads were waiting on the assignee and the assignee acquires
 * items from the assignor/movant as a result of the move, from
 * version 2.0.9 the waiting threads will unblock and extract those
 * items (such unblocking on move assignment did not happen with
 * version 2.0.8, which was a bug).
 *
 * Since 2.0.8
 */
  AsyncQueueDispatch& operator=(AsyncQueueDispatch&& rhs) {
    Thread::Mutex::Lock lock{mutex};
    q = std::move(rhs.q);
    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() = default;

/**
 * As regards thread safety, the move constructor does not synchronize
 * with respect to the initializing rvalue.  This is because
 * temporaries are only visible and accessible in the thread carrying
 * out the move operation and synchronization for them would represent
 * pointless overhead.  In a case where a user uses std::move to force
 * a move from a named object, and that named object's lifetime is
 * managed by (or the object is otherwise accessed by) a different
 * thread than the one making the move, the user must carry out her
 * own synchronization with respect to that different thread, both to
 * ensure that a consistent view of the the named object is obtained
 * and because that object will be mutated by the move.
 * @param rhs The AsyncQueueDispatch object to be moved.
 * @exception Thread::MutexError The move constructor might throw
 * Thread::MutexError if initialization of the contained mutex fails.
 * If it does so this move constructor is strongly exception safe (if
 * it is thrown, the object passed as an argument will be unchanged).
 * (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 move constructor might throw
 * Thread::CondError exception if initialisation of the contained
 * condition variable fails.  If it does so this move constructor is
 * strongly exception safe (if it is thrown, the object passed as an
 * argument will be unchanged).  (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 move constructor might also throw if the queue's
 * container type's move constructor might throw, but it should not do
 * that unless a custom allocator is in use.
 *
 * Since 2.0.8
 */
  AsyncQueueDispatch(AsyncQueueDispatch&& rhs): q(std::move(rhs.q)) {}

/**
 * The copy constructor is thread safe, as it locks the initializing
 * object's mutex to obtain a consistent view of it.
 * @param rhs The AsyncQueueDispatch 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 2.0.8
 */
  // 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++11 1.9/10 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++11 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 2.0.8
 */
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++11 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 2.0.8
 */
template <class T, class Container>
void swap(Cgu::AsyncQueueDispatch<T, Container>& q1,
          Cgu::AsyncQueueDispatch<T, Container>& q2) {
  q1.swap(q2);
}

} // namespace Cgu

#endif