PacQueue.h

Go to the documentation of this file.

/***************************************************************************
 *   Copyright (C) 2008-2010 by Matthias Nagl                              *
 *   mnagl@uni-goettingen.de                                               *
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU General Public License as published by  *
 *   the Free Software Foundation; either version 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 *   This program 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 General Public License for more details.                          *
 *                                                                         *
 *   You should have received a copy of the GNU General Public License     *
 *   along with this program; if not, write to the                         *
 *   Free Software Foundation, Inc.,                                       *
 *   59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.             *
 ***************************************************************************/

#ifndef PACQUEUE_H
#define PACQUEUE_H

#include <QSemaphore>
#include <QtDebug>
#include "pacDataTypes.h"

template <class T>
class PacQueue
{
  public:
    PacQueue(int buffersize=0);
    ~PacQueue();
    void setMaxLength(const int maxlength);
    int maxLength() const { return buffersize; };

    inline Pac::QueueStatus enqueue(T *w)
    {
      if (freefields.tryAcquire(1)) {
        qint32 cfield = writepos.fetchAndAddOrdered(1);
        buffer[cfield % buffersize] = w;
        usedfields.release(1);
        // reset counter carefully if necessary
        if (cfield >= buffersize)
          writepos.testAndSetOrdered(cfield, cfield % buffersize);
        // update counter
        enqueuedchunks++;
        return Pac::EnqueueSuccess;
      }
      droppedchunks++;
      return Pac::QueueIsFull;
    }

    inline T * dequeue()
    {
      usedfields.acquire(1);
      qint32 cfield = readpos.fetchAndAddOrdered(1);
      T *w = buffer[cfield % buffersize];
      freefields.release(1);
      // reset counter carefully if necessary
      if (cfield >= buffersize)
        readpos.testAndSetOrdered(cfield, cfield % buffersize);
      return w;
    }

    inline bool nonblockingDequeue(T ** dq)
    {
      if (!dq)
        return false;
      if (usedfields.tryAcquire(1)) {
        qint32 cfield = readpos.fetchAndAddOrdered(1);
        *dq = buffer[cfield % buffersize];
        freefields.release(1);
        // reset counter carefully if necessary
        if (cfield >= buffersize)
          readpos.testAndSetOrdered(cfield, cfield % buffersize);
        return true;
      }
      *dq = 0;
      return false;
    }

    inline T ** head() const
    {
      if (usedfields.available() > 0) {
        return &buffer[readpos % buffersize];
      }
      return 0;
    }

    inline qlonglong getHeadSeqNum() const
    {
      T **h = head();
      if (h == 0)
        return -1;
      else if (*h == 0)
        return 0;
      else
        return **h;
    }

    void getStats(int *enqueuedchunks, int *droppedchunks);
    void clear();

  private:
    T **buffer;
    mutable QSemaphore usedfields, freefields;
    qint32 buffersize;
    QAtomicInt readpos, writepos; 
    int enqueuedchunks; 
    int droppedchunks; 
};


template <class T> PacQueue<T>::PacQueue(int buffersize)
    : buffer(0), usedfields(0), freefields(buffersize), buffersize(buffersize), readpos(0), writepos(0), enqueuedchunks(0), droppedchunks(0)
{
  buffer = new T*[buffersize];
}


template <class T> PacQueue<T>::~PacQueue()
{
  delete [] buffer;
}

template <class T> void PacQueue<T>::setMaxLength(const int maxlength)
{
  if (usedfields.available() > 0)
    qDebug() << "WARNING: Queues should be empty before resizing!";
  clear();
  buffersize = maxlength;
  delete [] buffer;
  buffer = new T*[buffersize];
  freefields.acquire(freefields.available());
  freefields.release(buffersize);
  readpos = 0;
  writepos = 0;
}

template <class T> void PacQueue<T>::getStats(int *enqueuedchunks, int *droppedchunks)
{
  *enqueuedchunks = this->enqueuedchunks;
  this->enqueuedchunks=0;
  *droppedchunks = this->droppedchunks;
  this->droppedchunks=0;
}

template <class T> void PacQueue<T>::clear()
{
  T *tmp;
  while (nonblockingDequeue(&tmp))
    delete tmp;
}

#endif