FiFoQueue.hh

/*
This file is part of the BIAS library (Basic ImageAlgorithmS).

Copyright (C) 2003-2009    (see file CONTACT for details)
  Multimediale Systeme der Informationsverarbeitung
  Institut fuer Informatik
  Christian-Albrechts-Universitaet Kiel


BIAS 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.

BIAS 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 for more details.

You should have received a copy of the GNU Lesser General Public License
along with BIAS; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*/
#ifndef __FiFoQueue_hh
#define __FiFoQueue_hh

#include <bias_config.h>
#include <Base/Debug/Error.hh>

#ifndef BIAS_HAVE_PTHREADS
#  error pthreads need to be activated for FiFo-queue
#endif

#include <pthread.h>

#define __BIAS_FIFOQUEUE_DEFAULT_SIZE__ 10

namespace BIAS {

  /**
   * @class  FiFoQueue
   * @brief  This class implements a templated standard FiFo-queue for inter
   *         process communication (arbitrary number of readers/writers), e.g.
   *         between GUIs and underlying processes. It can be used as synced or
   *         unsynced.
   *         FiFo: Oldest item will be return at first. Items once fetched are
   *         'deleted' from the queue.
   *         synced:   process is block until a free slot available for writing
   *                   -> AppendSynced
   *                   or an item is available for read
   *                   -> FetchSynced
   *         unsynced: no item is returned from empty buffer when reading and
   *                   oldest item gets 'lost' when buffer is full during
   *                   writing
   *         !!! ATTENTION: WHEN TEMPLATED OVER POINTERS DATA 'BEHIND' !!!
   *         !!! THE POINTER WON'T DELETED WHEN BUFFER IS FULL DURING  !!!
   *         !!! WRITING                                               !!!
   * @author BIAS
   */
  template<class T> class FiFoQueue {

  public:
    /** Default constructor. Creates a queue with defined default size
        __BIAS_FIFOQUEUE_DEFAULT_SIZE__ and initializes mutex-/condition-
        variables
     */
    FiFoQueue() : length_(__BIAS_FIFOQUEUE_DEFAULT_SIZE__), head_(0),
                  tail_(1%__BIAS_FIFOQUEUE_DEFAULT_SIZE__), numItems_(0)
    {
      data_ = new T[__BIAS_FIFOQUEUE_DEFAULT_SIZE__];

      //initialize trigger/lock
      triggerCondFull_ = new pthread_cond_t();
      pthread_cond_init(triggerCondFull_, NULL);
      triggerCondEmpty_ = new pthread_cond_t();
      pthread_cond_init(triggerCondEmpty_, NULL);
      syncMutex_ = new pthread_mutex_t();
      pthread_mutex_init(syncMutex_, NULL);
    }

    /** Constructor. Creates a queue with given size and initializes 
        mutex-/condition-variables
     */
    FiFoQueue(unsigned length) :
      length_(length), head_(0), tail_(1%length), numItems_(0)
    {
      data_ = new T[length_];

      //initialize trigger/lock
      triggerCondFull_ = new pthread_cond_t();
      pthread_cond_init(triggerCondFull_, NULL);
      triggerCondEmpty_ = new pthread_cond_t();
      pthread_cond_init(triggerCondEmpty_, NULL);
      syncMutex_ = new pthread_mutex_t();
      pthread_mutex_init(syncMutex_, NULL);
    }

    /** Just a cleanup.
        !!! ATTENTION: WHEN USING POINTERS AS TEMPLATE TYPE ONLY THE !!!
        !!! BUFFER IS DELETED - NOT THE DATA 'BEHIND' THE POINTERS   !!!
     */
    ~FiFoQueue()
    {
      delete data_;

      pthread_cond_destroy(triggerCondFull_);
      delete triggerCondFull_;
      pthread_cond_destroy(triggerCondEmpty_);
      delete triggerCondEmpty_;
      pthread_mutex_destroy(syncMutex_);
      delete syncMutex_;
    }


    /** Return the number of items in queue
     */
    inline unsigned GetNumItems()
    { unsigned numItems;
      pthread_mutex_lock(syncMutex_);
      numItems = numItems_;
      pthread_mutex_unlock(syncMutex_);
      return numItems_; }


    /** Returns the size of underlying ringbuffer
        -> maximum number of items in queue
     */
    inline unsigned GetBufferSize()
    { return length_; }


    /** Append an datum as newest item. Syncron in the sens of blocking,
        that is, threads inserting an item to a full queue are delayed until
        a slot is available (item has been fetched)
     */
    inline int AppendSynced(T &item)
    { 
      //get exclusive access
      pthread_mutex_lock(syncMutex_);
      while (numItems_ == length_) {
        //ensure consisntency for full buffer
        BIASASSERT((head_+1)%length_ == tail_);
        //in case of full buffer wait for free slot
        //...while needed because more than 1 waiting process might
        //   be unblocked from cond_signal!
        pthread_cond_wait(triggerCondFull_,syncMutex_);
      }

      //signal new item inserted
      pthread_cond_signal(triggerCondEmpty_);

      //make head_ point to next slot
      head_ = (head_+1)%length_;
      //insert data at head_
      data_[head_] = item;
      //increase number of used slots
      numItems_++;
      //release exclusiv access
      pthread_mutex_unlock(syncMutex_);

      return 0;
    }

    /** Append an datum as newest item. Asyncron in the sens of non-blocking,
        that is, threads inserting an item to a full queue will overwrite the
        oldest item. This gets 'lost'.
        !!! ATTENTION: OVERWRITING A POINTER TYPE DOES NO FREE MEMORY !!!
     */
    inline int AppendUnsynced(T &item)
    { 
      //get exclusive access
      pthread_mutex_lock(syncMutex_);

      //make tail_ point to next slot
      head_ = (head_+1)%length_;
      //insert data at tail_
      data_[head_] = item;

      if (numItems_ == length_) {
        //if buffer was full before insert, oldest item is deleted
        // -> increase tail_
        tail_ = (tail_+1)%length_;
        //release exclusiv access
        pthread_mutex_unlock(syncMutex_);
        //return message that oldest item was 'lost'
        return 1;
      }

      //signal new item inserted
      pthread_cond_signal(triggerCondEmpty_);
 
      //increase number of used slots if buffer was not full before insert
      numItems_++;
      //release exclusiv access
      pthread_mutex_unlock(syncMutex_);

      return 0;
    }

    /** Fetch the oldest item from the queue. Again synced is blocking,
        thread will wait for an item if buffer is empty.
     */
    inline int FetchSynced(T &item)
    {
      //get exclusive access
      pthread_mutex_lock(syncMutex_);
      while (numItems_ == 0) {
        //ensure consisntency for empty buffer
        BIASASSERT((head_+1)%length_ == tail_);
        //in case of buffer is empty wait for item
        //...while needed because more than 1 waiting process might
        //   be unblocked from pthread_cond_signal!
        pthread_cond_wait(triggerCondEmpty_,syncMutex_);
      }

      //signal new slot is available
      pthread_cond_signal(triggerCondFull_);
      
      //fetch item at tail_
      item = data_[tail_];
      //make tail_ point to next slot
      tail_ = (tail_+1)%length_;
      //decrease number of used slots if buffer was not empty before insert
      numItems_--;
      //release exclusiv access
      pthread_mutex_unlock(syncMutex_);

      return 0;
    }

    /** Fetch the oldest item from the queue. Again unsynced is non-blocking,
        thread won't wait for an item if buffer is empty. In this case the input-/
        output-param is left unchanged and 1 is returned.
     */
    inline int FetchUnsynced(T &item)
    {
      //get exclusive access
      pthread_mutex_lock(syncMutex_);

      if (numItems_ == 0) {
        //if buffer was empty before fetch, no item is returned...
        // release exclusiv access (only possible if not synced)
        pthread_mutex_unlock(syncMutex_);
        //signal that no item was available
        return 1;
      } 
        
      //signal new slot is available
      pthread_cond_signal(triggerCondFull_);

      //fetch data at head_
      item = data_[tail_];
      //make head_ point to next slot
      tail_ = (tail_+1)%length_;
      //decrease item counter
      numItems_--;
      //release exclusiv access
      pthread_mutex_unlock(syncMutex_);

      return 0;
    }

  protected:

    //prohibit copy -> no consistent copy feasible
    FiFoQueue(FiFoQueue &data){};

    ///array holding cycled buffer
    T* data_;

    /// pthread condition variable for waiting for non full buf
    pthread_cond_t *triggerCondFull_;
    /// pthread condition variable for waiting for non empty buf
    pthread_cond_t *triggerCondEmpty_;
    /// mutex for sync. locking (conditions and access)
    pthread_mutex_t *syncMutex_;

    ///length of buffer (number of slots)
    const unsigned length_;

    ///first (oldest) element in queue
    unsigned head_;
    ///last (newest) element in queue
    unsigned tail_;
    ///number of items in buffer
    unsigned numItems_;
  };
}

#endif // __FiFoQueue_hh