async_channel.h

Go to the documentation of this file.

/* Copyright (C) 2016 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_channel.h
 * @brief This file provides a "channel" class for inter-thread communication.
 *
 * AsyncChannel is similar to the AsyncQueueDispatch class, in that it
 * provides a means of sending data between threads.  Producer threads
 * push data onto the queue and consumer threads pop them off in a
 * thread safe way, and if there are no data in the channel any
 * consumer thread will block until a producer thread pushes an item
 * onto it.  However, unlike the AsyncQueueDispatch class, it has a
 * fixed maximum size as part of its type, which may be any size
 * greater than 0, and if the number of data items still in the
 * channel is such as to make the channel full, then producer threads
 * will block on the channel until a consumer thread pops an item from
 * it.
 *
 * It therefore provides for back pressure on producer threads which
 * will automatically prevent the channel being overwhelmed by
 * producer threads pushing more items onto the queue than consumer
 * threads have the capacity to take off it.
 *
 * AsyncChannel is useful where this feature is important, and an
 * AsyncChannel object can be used with any number of producer threads
 * and consumer threads.  However, under heavy contention with complex
 * data item types AsyncQueueDispatch objects will usually be faster.
 * Under lower contention and with simpler data types, an AsyncChannel
 * object may be faster as it can benefit from its fixed buffer size
 * (the AsyncChannel implementation uses a circular buffer with
 * in-buffer construction of data items).
 *
 * This class is available from version 1.2.42.
 */
 
#ifndef CGU_ASYNC_CHANNEL_H
#define CGU_ASYNC_CHANNEL_H
 
#include <new>        // for std::bad_alloc
#include <cstddef>    // for std::size_t
#include <cstdlib>    // for std::malloc
 
#include <pthread.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 {
 
// function for pthread_cleanup_push() in AsyncChannel::push() and
// AsyncChannel::pop() methods
extern "C" {
inline void cgu_async_channel_waiters_dec(void* arg) {
  --(*static_cast<std::size_t*>(arg));
}
} // extern "C"
 
/**
 * @class AsyncChannel async_channel.h c++-gtk-utils/async_channel.h
 * @brief A thread-safe "channel" class for inter-thread communication.
 * @sa AsyncQueue AsyncQueueDispatch AsyncChannel AsyncResult
 *
 * AsyncChannel is similar to the AsyncQueueDispatch class, in that it
 * provides a means of sending data between threads.  Producer threads
 * push data onto the queue and consumer threads pop them off in a
 * thread safe way, and if there are no data in the channel any
 * consumer thread will block until a producer thread pushes an item
 * onto it.  However, unlike the AsyncQueueDispatch class, it has a
 * fixed maximum size as part of its type, which may be any size
 * greater than 0, and if the number of data items still in the
 * channel is such as to make the channel full, then producer threads
 * will block on the channel until a consumer thread pops an item from
 * it.
 *
 * It therefore provides for back pressure on producer threads which
 * will automatically prevent the channel being overwhelmed by
 * producer threads pushing more items onto the queue than consumer
 * threads have the capacity to take off it.
 *
 * AsyncChannel is useful where this feature is important, and an
 * AsyncChannel object can be used with any number of producer threads
 * and consumer threads.  However, under heavy contention with complex
 * data item types AsyncQueueDispatch objects will usually be faster.
 * Under lower contention and with simpler data types, an AsyncChannel
 * object may be faster as it can benefit from its fixed buffer size
 * (the AsyncChannel implementation uses a circular buffer with
 * in-buffer construction of data items).
 *
 * AsyncChannel objects are instantiated with firstly a template type
 * 'T' and secondly a template integer value 'n'.  'T' is the type of
 * the data items to be placed on the queue.  'n' is the size of the
 * channel, which as mentioned may be any size greater than 0.
 *
 * This class is available from version 1.2.42.
 */
 
/*
 * We have to use Thread::Cond::broadcast() and not
 * Thread::Cond::signal(), because it is possible to have at any one
 * time both a producer and a consumer waiting on the AsyncChannel's
 * condition variable.  This can create a "thundering herd" problem if
 * there are a large number of threads waiting on the channel, but it
 * is within the range of the acceptable.  As an example of the issue
 * requiring this approach, take this case:
 *
 * Let there be a channel which has a capacity of one.  Let the
 * channel already have an item in it from some past producer.  In
 * addition, let two producer threads be currently in a cond-wait,
 * waiting for the channel to become empty in order to put an item in
 * it.
 *
 * A first consumer thread starts to remove the item in the channel.
 * It acquires the channel's mutex without blocking because it is not
 * locked.  It sees there is an item in the channel so does not begin
 * a cond-wait.  Meanwhile, immediately after the first consumer
 * thread acquires the mutex a second consumer thread tries to obtain
 * an item from the channel and thus will block when trying to acquire
 * the locked mutex.
 *
 * The first consumer thread then removes the item from the channel's
 * queue and signals the cond-var - which causes one of the producer
 * threads to wake-up and block on trying to acquire the mutex ("the
 * first producer").  So for the producers, we now have the first
 * producer contending on the mutex after being so awoken and the
 * second producer still waiting on the cond-var.  And we have two
 * threads now contending on the mutex - the thread of the second
 * consumer and the thread of the first producer.  One of these
 * threads will acquire the mutex when the first consumer releases it
 * after signalling the cond-var, and it is unspecified which one.
 *
 * If it is the second consumer, it will find the channel empty and
 * thus enter a cond-wait (so we now have the second producer and the
 * second consumer waiting on the same cond-var) and release the
 * mutex.  This will cause the first producer to acquire the mutex,
 * which will then add the item to the channel, signal the cond-var
 * (which will cause either the second producer or second consumer to
 * awaken), and release the mutex.  If it is the second producer which
 * awakens, its thread will acquire the mutex, find the channel full,
 * enter a cond-wait again and release the mutex.  Now we are stuffed
 * because there is nothing to awaken the remaining consumer even
 * though there is something in the channel.
 */
template <class T, std::size_t n> class AsyncChannel {
public:
  typedef T value_type;
  typedef std::size_t size_type;
private:
  mutable Thread::Mutex mutex;
  Thread::Cond cond;
  size_type size;  // number of available items in channel
  size_type idx;   // index of first available item in channel (this
           // value is meaningless when size == 0)
  size_type waiters;
  enum Status {normal, closed, destructing} status;
  T* buf;
 
// this class cannot be copied
  AsyncChannel(const AsyncChannel&);
  AsyncChannel& operator=(const AsyncChannel&);
public:
/**
 * Closes the channel.  This means that (i) any threads blocking on a
 * full channel with a call to push() will unblock with a false return
 * value, and any call to that method after the closure will return
 * immediately with a false return value, (ii) any threads blocking on
 * an empty channel with a call to pop() will unblock with a false
 * return value, (iii) any data items remaining in the channel which
 * were pushed to the channel prior to the closure of the channel can
 * be popped after that closure by further calls to pop(), which will
 * return normally with a true return value, and (iv) after any such
 * remaining data items have been removed, any subsequent calls to
 * pop() will return with a false return value.
 *
 * If called more than once, this method will do nothing.
 *
 * This method will not throw.  It is thread safe - any thread may
 * call it.
 *
 * One of the main purposes of this method is to enable a producer
 * thread to inform a consumer thread that nothing more will be put in
 * the channel by it for the consumer: for such cases, as mentioned
 * once everything pushed to the channel prior to its closure has been
 * extracted from the channel by pop() calls, any further pop() calls
 * will return false.  At that point the consumer thread can abandon
 * and destroy the AsyncChannel object.
 *
 * Since 1.2.42
 */
  void close() {
    Thread::Mutex::Lock lock(mutex);
    if (status == normal) {
      status = closed;
      cond.broadcast();
    }
  }
 
/**
 * Pushes an item onto the channel. This method will only throw if the
 * copy constructor of the pushed item throws, and has strong
 * exception safety in such a case.  It is thread safe.
 *
 * If the number of items already in the channel is equal to the size
 * of the channel, then this method blocks until either room becomes
 * available for the item by virtue of another thread calling the
 * pop() method, or another thread calls the close() method.  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 channel is being operated on
 * after coming out of a wait.
 *
 * @param obj The item to be pushed onto the channel.
 * @return If the push succeeds (whether after blocking or not
 * blocking) this method returns true.  If this method unblocks
 * because the channel has been closed, or any subsequent calls to
 * this method are made after the channel has been closed, this method
 * returns false.
 *
 * Since 1.2.42
 */
  bool push(const value_type& obj) {
    bool waiting = false;
    Thread::Mutex::Lock lock(mutex);
    if (status != normal) return false;
 
    // the only function call that could be a cancellation point
    // within this pthread_cleanup_push/pop block is the call to
    // Cond::wait() - if there is a cancellation request while
    // blocking in that call we need to decrement the waiters count
    // with a cancellation handler.  The push is efficient enough not
    // to have to unlock the mutex - the cleanup_push/pop block
    // enables the cleanup macros to construct all the cleanup datum
    // on the function stack set up at function entry
    pthread_cleanup_push(cgu_async_channel_waiters_dec, &this->waiters);
    while (size >= n && status == normal) {
      if (!waiting) {
    ++waiters;
    waiting = true;
      }
      cond.wait(mutex); // cancellation point
    }
    Thread::CancelBlock b;
    pthread_cleanup_pop(false);
    // we need to keep a local copy of 'status' because as soon as
    // 'waiters' is decremented the AsyncChannel object could be
    // destroyed if status == destructing
    Status local_status = status;
    if (waiting) --waiters;
    if (local_status != normal) return false;
    // next is the index of the next available space in the channel
    size_type next = (idx + size) % n;
    new (static_cast<void*>(buf + next)) T(obj);
    ++size;
    cond.broadcast();
    return true;
  }
 
/**
 * Pops an item from the channel using the item type's copy assignment
 * operator.  This method will only throw if that operator throws or
 * the contained item's destructor throws.  It has strong exception
 * safety, provided the destructor of the contained item does not
 * throw.  It is thread safe.
 *
 * If the channel is empty, then this method blocks until either an
 * item becomes available by virtue of another thread calling the
 * push() method, or another thread calls the close() method.  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 channel is being operated on
 * after coming out of a wait.
 *
 * @param obj A value type reference to which the item at the front of
 * the channel will be copy assigned.
 * @return If the pop succeeds (whether after blocking or not
 * blocking) this method returns true.  If this method unblocks
 * because the channel has been closed or any subsequent calls to this
 * method are made, and there are no remaining items in the channel,
 * this method returns false.
 *
 * Since 1.2.42
 */
  bool pop(value_type& obj) {
    bool waiting = false;
    Thread::Mutex::Lock lock(mutex);
 
    // the only function call that could be a cancellation point
    // within this pthread_cleanup_push/pop block is the call to
    // Cond::wait() - if there is a cancellation request while
    // blocking in that call we need to decrement the waiters count
    // with a cancellation handler.  The push is efficient enough not
    // to have to unlock the mutex - the cleanup_push/pop block
    // enables the cleanup macros to construct all the cleanup datum
    // on the function stack set up at function entry
    pthread_cleanup_push(cgu_async_channel_waiters_dec, &this->waiters);
    while (!size && status == normal) {
      if (!waiting) {
    ++waiters;
    waiting = true;
      }
      cond.wait(mutex);
    }
    Thread::CancelBlock b;
    pthread_cleanup_pop(false);
 
    if (status == destructing) {
      // decrementing waiters must be the last thing we do as it might
      // cause the destructor to return
      if (waiting) --waiters;
      return false;
    }
    else if (size) { // status == normal, or status == closed && size > 0
      size_type old_idx = idx;
      obj = buf[old_idx];
      ++idx;
      if (idx == n) idx = 0;
      --size;
      try {
    buf[old_idx].~T();
      }
      catch (...) {
    // we might as well keep the AsyncChannel object's waiters in
    // a workable state even if the item's destructor throws
    if (waiting) --waiters;
    cond.broadcast();
    throw;
      }
      if (waiting) --waiters;
      cond.broadcast();
      return true;
    }
    else { // status == closed and size == 0
      if (waiting) --waiters;
      return false;
    }
  }
 
/**
 * AsyncChannel objects are instantiated with firstly a template type
 * 'T' and secondly a template integer value 'n'.  'T' is the type of
 * the data items to be placed on the queue.  'n' is the size of the
 * channel, which must be greater than 0.  However, a circular buffer
 * for that size will be allocated at construction time by this
 * default constructor, so the given size should not be unnecessarily
 * large.  Where a large AsyncChannel object would be required,
 * consider using AsyncQueueDispatch instead, which sizes itself
 * dynamically.
 *
 * @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.)
 *
 * Since 1.2.42
 */
  AsyncChannel(): size(0), idx(0), waiters(0), status(normal),
                  buf(static_cast<T*>(std::malloc(sizeof(T) * n))) {
    if (!buf) throw std::bad_alloc();
  }
 
 /**
 * The destructor does not throw unless the destructor of a data item
 * in the channel throws.  It is thread safe (any thread may delete
 * the AsyncChannel object).
 *
 * It is not an error for a thread to destroy the AsyncChannel object
 * and so invoke this destructor while another thread is blocking on
 * it: instead the destructor will release any blocking threads.  The
 * destructor will not return until all threads (if any) blocking on
 * the AsyncChannel object have been released.
 *
 * Since 1.2.42
 */
  ~AsyncChannel() {
    mutex.lock();
    status = destructing;
    mutex.unlock();
    cond.broadcast();
 
    // since all a waiting thread does upon status == destructing is
    // to unblock and return, it is more efficient to spin gracefully
    // until 'waiters' is 0 instead of starting another condition
    // variable wait
    for (;;) {
      mutex.lock();
      if (waiters) {
    mutex.unlock();
#ifdef CGU_USE_SCHED_YIELD
    sched_yield();
#else
    usleep(10);
#endif
      }
      else {
    mutex.unlock();
    break;
      }
    }
    while (size) {
      buf[idx].~T();
      ++idx;
      if (idx == n) idx = 0;
      --size;
    }
    std::free(buf);
  }
 
/* Only has effect if --with-glib-memory-slices-compat or
 * --with-glib-memory-slices-no-compat option picked */
  CGU_GLIB_MEMORY_SLICES_FUNCS
};
 
#ifndef DOXYGEN_PARSING
 
/* This is a specialization of AsyncChannel when instantiated with a
   size of 1.  This specialization allows a number of optimizations
   for that case.
*/
template <class T>
class AsyncChannel<T, 1> {
public:
  typedef T value_type;
  typedef std::size_t size_type;
private:
  mutable Thread::Mutex mutex;
  Thread::Cond cond;
  size_type waiters;
  bool full;
  enum Status {normal, closed, destructing} status;
  T* datum;
 
  AsyncChannel(const AsyncChannel&);
  AsyncChannel& operator=(const AsyncChannel&);
public:
  void close() {
    Thread::Mutex::Lock lock(mutex);
    if (status == normal) {
      status = closed;
      cond.broadcast();
    }
  }
 
  bool push(const value_type& obj) {
    bool waiting = false;
    Thread::Mutex::Lock lock(mutex);
    if (status != normal) return false;
 
    // the only function call that could be a cancellation point
    // within this pthread_cleanup_push/pop block is the call to
    // Cond::wait() - if there is a cancellation request while
    // blocking in that call we need to decrement the waiters count
    // with a cancellation handler.  The push is efficient enough not
    // to have to unlock the mutex - the cleanup_push/pop block
    // enables the cleanup macros to construct all the cleanup datum
    // on the function stack set up at function entry
    pthread_cleanup_push(cgu_async_channel_waiters_dec, &this->waiters);
    while (full && status == normal) {
      if (!waiting) {
    ++waiters;
    waiting = true;
      }
      cond.wait(mutex); // cancellation point
    }
    Thread::CancelBlock b;
    pthread_cleanup_pop(false);
    // we need to keep a local copy of 'status' because as soon as
    // 'waiters' is decremented the AsyncChannel object could be
    // destroyed if status == destructing
    Status local_status = status;
    if (waiting) --waiters;
    if (local_status != normal) return false;
    new (static_cast<void*>(datum)) T(obj);
    full = true;
    cond.broadcast();
    return true;
  }
 
  bool pop(value_type& obj) {
    bool waiting = false;
    Thread::Mutex::Lock lock(mutex);
 
    // the only function call that could be a cancellation point
    // within this pthread_cleanup_push/pop block is the call to
    // Cond::wait() - if there is a cancellation request while
    // blocking in that call we need to decrement the waiters count
    // with a cancellation handler.  The push is efficient enough not
    // to have to unlock the mutex - the cleanup_push/pop block
    // enables the cleanup macros to construct all the cleanup datum
    // on the function stack set up at function entry
    pthread_cleanup_push(cgu_async_channel_waiters_dec, &this->waiters);
    while (!full && status == normal) {
      if (!waiting) {
    ++waiters;
    waiting = true;
      }
      cond.wait(mutex);
    }
    Thread::CancelBlock b;
    pthread_cleanup_pop(false);
 
    if (status == destructing) {
      // decrementing waiters must be the last thing we do as it might
      // cause the destructor to return
      if (waiting) --waiters;
      return false;
    }
    else if (full) { // status == normal, or status == closed && full
      obj = *datum;
      full = false;
      try {
    datum->~T();
      }
      catch (...) {
    // we might as well keep the AsyncChannel object's waiters in
    // a workable state even if the item's destructor throws
    if (waiting) --waiters;
    cond.broadcast();
    throw;
      }
      if (waiting) --waiters;
      cond.broadcast();
      return true;
    }
    else { // status == closed and !full
      if (waiting) --waiters;
      return false;
    }
  }
 
  AsyncChannel(): waiters(0), full(false), status(normal),
                  datum(static_cast<T*>(std::malloc(sizeof(T)))) {
    if (!datum) throw std::bad_alloc();
  }
 
  ~AsyncChannel() {
    mutex.lock();
    status = destructing;
    mutex.unlock();
    cond.broadcast();
 
    // since all a waiting thread does upon status == destructing is
    // to unblock and return, it is more efficient to spin gracefully
    // until 'waiters' is 0 instead of starting another condition
    // variable wait
    for (;;) {
      mutex.lock();
      if (waiters) {
    mutex.unlock();
#ifdef CGU_USE_SCHED_YIELD
    sched_yield();
#else
    usleep(10);
#endif
      }
      else {
    mutex.unlock();
    break;
      }
    }
    if (full) datum->~T();
    std::free(datum);
  }
 
  CGU_GLIB_MEMORY_SLICES_FUNCS
};
 
#endif // DOXYGEN_PARSING
 
} // namespace Cgu
 
#endif // CGU_ASYNC_CHANNEL_H