DwmConcurrentQueue.hh

Go to the documentation of this file.

//===========================================================================
// @(#) $DwmPath$
//===========================================================================
//  Copyright (c) Daniel W. McRobb 2016
//  All rights reserved.
//
//  Redistribution and use in source and binary forms, with or without
//  modification, are permitted provided that the following conditions
//  are met:
//
//  1. Redistributions of source code must retain the above copyright
//     notice, this list of conditions and the following disclaimer.
//  2. Redistributions in binary form must reproduce the above copyright
//     notice, this list of conditions and the following disclaimer in the
//     documentation and/or other materials provided with the distribution.
//  3. The names of the authors and copyright holders may not be used to
//     endorse or promote products derived from this software without
//     specific prior written permission.
//
//  IN NO EVENT SHALL DANIEL W. MCROBB BE LIABLE TO ANY PARTY FOR
//  DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES,
//  INCLUDING LOST PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE,
//  EVEN IF DANIEL W. MCROBB HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH
//  DAMAGE.
//
//  THE SOFTWARE PROVIDED HEREIN IS ON AN "AS IS" BASIS, AND
//  DANIEL W. MCROBB HAS NO OBLIGATION TO PROVIDE MAINTENANCE, SUPPORT,
//  UPDATES, ENHANCEMENTS, OR MODIFICATIONS. DANIEL W. MCROBB MAKES NO
//  REPRESENTATIONS AND EXTENDS NO WARRANTIES OF ANY KIND, EITHER
//  IMPLIED OR EXPRESS, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
//  WARRANTIES OF MERCHANTABILITY OR FITNESS FOR A PARTICULAR PURPOSE,
//  OR THAT THE USE OF THIS SOFTWARE WILL NOT INFRINGE ANY PATENT,
//  TRADEMARK OR OTHER RIGHTS.
//===========================================================================
 
//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
 
#ifndef _DWMCONCURRENTQUEUE_HH_
#define _DWMCONCURRENTQUEUE_HH_
 
extern "C" {
  #include <unistd.h>
}
 
#include <atomic>
#include <memory>
#include <thread>
#include <vector>
 
namespace Dwm {
 
  namespace Thread {
 
    //------------------------------------------------------------------------
    //------------------------------------------------------------------------
    template <typename T>
    class ConcurrentQueue
    {
    public:
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      ConcurrentQueue(uint64_t maxLength = 10000,
                      useconds_t pushSleepUsecs = 1000)
          :  _length(0), _maxLength(maxLength),
             _pushSleepUsecs(pushSleepUsecs)
      {
        _front = _back = new Entry(T());
        _pushingLocked = _poppingLocked = false;
      }
      
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      ~ConcurrentQueue()
      {
        while (_front != nullptr) {
          Entry *tmp = _front;
          _front = tmp->next;
          delete tmp;
        }
      }
      
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      bool PopFront(T & result)
      {
        while (_poppingLocked.exchange(true));
        if (_front->next != nullptr) {
          //  queue is not empty.
          Entry  *oldFirst = _front;
          _front = _front->next;
          result = _front->value;
          _poppingLocked = false;
          //  delete the old front Entry.
          delete oldFirst;
          //  decrement the queue length.
          --_length;
          return true;
        }
        _poppingLocked = false;
        return false;
      }
 
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      bool PopFront(std::vector<T> & result)
      {
        typename std::vector<T>::size_type  numEntries = 0;
        //  lcok against others popping from the queue
        while (_poppingLocked.exchange(true));
        if (_front->next != nullptr) {
          //  queue is not empty.  Save the existing front.
          Entry  *oldFront = _front;
          //  Move front to the end of the queue since we're going to take
          //  all entries from the queue.  Keep track of how many entries
          //  we'll take, and adjust _length at end.
          while (_front->next != nullptr) {
            _front = _front->next;
            ++numEntries;
          }
          _length -= numEntries;
          //  Save the new location of _front.  This is where we'll stop
          //  popping entries.  Once we unlock popping, another thread
          //  could come in and move _front.
          Entry  *newFront = _front;
          //  Unlock popping so other threads can pop entries.
          _poppingLocked = false;
          //  Resize the output vector to hold the popped entries.
          result.resize(numEntries);
          //  Populate the output vector and delete the old entries.
          typename std::vector<T>::size_type  i = 0;
          Entry  *entry = oldFront->next;
          while (entry != newFront) {
            Entry  *delEntry = entry;
            result[i] = entry->value;
            entry = entry->next;
            ++i;
            delete delEntry;
          }
          //  Delete the old front entry.
          delete oldFront;
          return true;
        }
        _poppingLocked = false;
        return false;
      }
 
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      bool PushBack(const T & t)
      {
        while (_length > _maxLength) { usleep(_pushSleepUsecs); };
        Entry *tmp = new Entry(t);
        while (_pushingLocked.exchange(true));
        _back->next = tmp;
        _back = tmp;
        _pushingLocked = false;
        ++_length;
        return true;
      }
 
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      bool PushBack(const std::vector<T> & t)
      {
        while (_length > _maxLength) { usleep(_pushSleepUsecs); }
        std::vector<Entry *> newEntries(t.size());
        typename std::vector<T>::size_type i = 0;
        newEntries[i] = new Entry(t[i]);
        ++i;
        for (; i < t.size(); ++i) {
          newEntries[i] = new Entry(t[i]);
          newEntries[i-1]->next = newEntries[i];
        }
        while (_pushingLocked.exchange(true));
        _back->next = newEntries[0];
        _back = newEntries[t.size() - 1];
        _pushingLocked = false;
        _length += t.size();
        return true;
      }
 
    private:
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      class Entry
      {
      public:
        Entry(const T & val) 
            : value(val), next(nullptr) 
        {}
        T                      value;
        std::atomic<Entry *>   next;
      };
 
      Entry                  *_front;
      std::atomic<bool>       _poppingLocked;
      Entry                  *_back;
      std::atomic<bool>       _pushingLocked;
      std::atomic<uint64_t>   _length;
      uint64_t                _maxLength;
      useconds_t              _pushSleepUsecs;
    };
    
 
  }  // namespace Thread
  
}  // namespace Dwm
 
#endif  // _DWMCONCURRENTQUEUE_HH_
 
//---------------------------- emacs settings -----------------------------
//  Local Variables:
//  mode: C++
//  tab-width: 2
//  indent-tabs-mode: nil
//  c-basic-offset: 2
//  End:
//-------------------------------------------------------------------------