DwmThreadQueue.hh

Go to the documentation of this file.

//===========================================================================
// @(#) $DwmPath$
//===========================================================================
//  Copyright (c) Daniel W. McRobb 2000-2007, 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 _DWMTHREADQUEUE_HH_
#define _DWMTHREADQUEUE_HH_
 
#include <atomic>
#include <deque>
#include <iostream>
#include <mutex>
#include <condition_variable>
#include <type_traits>
 
namespace Dwm {
 
  namespace Thread {
 
    //------------------------------------------------------------------------
    //------------------------------------------------------------------------
    template <typename _ValueType> 
    class Queue
    {
    public:
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      Queue()
          : _maxLength(0), _queue(), _mutex(), _signalled(false),
            _lock(_mutex), _cv()
      {
        _lock.unlock();
      }
    
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      ~Queue()
      {
        std::unique_lock<std::mutex>  lk(_mutex);
        _queue.clear();
      }
     
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      uint32_t MaxLength() const
      {
        return(_maxLength);
      }
     
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      uint32_t MaxLength(uint32_t maxLength)
      {
        _maxLength = maxLength;
        return(_maxLength);
      }
 
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      typename std::deque<_ValueType>::size_type Length() const
      {
        std::lock_guard<std::mutex>  lk(_mutex);
        return _queue.size();
      }
      
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      bool PushBack(const _ValueType & value)
      {
        bool  rc = false;
        std::lock_guard<std::mutex>  lk(_mutex);
        if ((! _maxLength) ||
            (_queue.size() < _maxLength)) {
          _queue.push_back(value);
          _signalled = true;
          _cv.notify_all();
          rc = true;
        }
        return(rc);
      }
 
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      bool PushBack(_ValueType && value)
      {
        bool  rc = false;
        std::lock_guard<std::mutex>  lk(_mutex);
        if ((! _maxLength) ||
            (_queue.size() < _maxLength)) {
          _queue.push_back(value);
          _signalled = true;
          _cv.notify_all();
          rc = true;
        }
        return(rc);
      }
      
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      template <typename InputIterator>
      uint32_t PushBack(InputIterator firstIter, InputIterator lastIter)
      {
        uint32_t  rc = 0;
        if (firstIter != lastIter) {
          std::lock_guard<std::mutex>  lk(_mutex);
          if ((! _maxLength) ||
              (_queue.size() < _maxLength)) {
            uint32_t  oldSize = _queue.size();
            _queue.insert(_queue.end(), firstIter, lastIter);
            rc = _queue.size() - oldSize;
            if (rc) {
              _signalled = true;
              _cv.notify_all();
            }
          }
        }
        return(rc);
      }
 
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      bool PushFront(const _ValueType & value)
      {
        bool  rc = false;
        std::lock_guard<std::mutex>  lk(_mutex);
        if ((! _maxLength) ||
            (_queue.size() < _maxLength)) {
          _queue.push_front(value);
          _signalled = true;
          _cv.notify_all();
          rc = true;
        }
        return(rc);
      }
 
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      template <typename InputIterator>
      uint32_t PushFront(InputIterator firstIter, InputIterator lastIter)
      {
        uint32_t  rc = 0;
        if (firstIter != lastIter) {
          std::lock_guard<std::mutex>  lk(_mutex);
          if ((! _maxLength) ||
              (_queue.size() < _maxLength)) {
            uint32_t  oldSize = _queue.size();
            _queue.insert(_queue.begin(), firstIter, lastIter);
            rc = _queue.size() - oldSize;
            if (rc) {
              _signalled = true;
              _cv.notify_all();
            }
          }
        }
        return rc;
      }
      
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      void ConditionSignal()
      {
        _signalled = true;
        _cv.notify_one();
      }
     
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      bool ConditionWait()
      {
        Lock();
        _cv.wait(_lock, [&] { return _signalled.load(); });
        _signalled = false;
        Unlock();
        return true;
      }
     
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      template<class Rep, class Period>
      bool ConditionTimedWait(const std::chrono::duration<Rep,Period> & timeToWait)
      {
        bool rc = false;
        Lock();
        if (_cv.wait_for(_lock, timeToWait,
                         [&] { return _signalled.load(); })) {
          rc = true;
        }
        _signalled = false;
        Unlock();
        return(rc);
      }
     
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      bool PopFront(_ValueType & value)
      {
        bool  rc = false;
        std::lock_guard<std::mutex>  lk(_mutex);
        if (! _queue.empty()) {
          if (std::is_move_assignable<_ValueType>::value) {
            value = std::move(_queue.front());
          }
          else {
            value = _queue.front();
          }
          _queue.pop_front();
          rc = true;
        }
        return(rc);
      }
 
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      bool PopBack(_ValueType & value)
      {
        bool  rc = false;
        std::lock_guard<std::mutex>  lk(_mutex);
        if (! _queue.empty()) {
          if (std::is_move_assignable<_ValueType>::value) {
            value = std::move(_queue.back());
          }
          else {
            value = _queue.back();
          }
          _queue.pop_back();
          rc = true;
        }
        return(rc);
      }
     
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      bool WaitForNotEmpty()
      {
        bool  rc = false;
        Lock();
        _cv.wait(_lock, [&] { return (! _queue.empty()); });
        rc = true;
        Unlock();
        return(rc);
      }
     
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      template <class Rep, class Period>
      bool TimedWaitForNotEmpty(const std::chrono::duration<Rep, Period> & timeToWait)
      {
        bool rc = false;
        Lock();
        if (! _queue.empty()) {
          rc = true;
        }
        else {
          if (_cv.wait_for(_lock, timeToWait,
                           [&] { return _signalled.load(); })) {
            rc = (! _queue.empty());
          }
        }
        _signalled = false;
        Unlock();
        return(rc);
      }
     
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      bool Empty()
      {
        return(_queue.empty());
      }
 
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      void RandomShuffle()
      {
        std::lock_guard<std::mutex>  lk(_mutex);
        random_shuffle(_queue.begin(), _queue.end());
        return;
      }
     
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      uint32_t Copy(std::deque<_ValueType> & c)
      {
        uint32_t  rc = 0;
        if (! c.empty())
          c.clear();
       
        std::lock_guard<std::mutex>  lk(_mutex);
        typename std::deque<_ValueType>::iterator  iter = _queue.begin();
        for ( ; iter != _queue.end(); ++iter) {
          c.push_back(*iter);
          ++rc;
        }
        return(rc);
      }
 
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      uint32_t Swap(std::deque<_ValueType> & c)
      {
        std::lock_guard<std::mutex>  lk(_mutex);
        if (! _queue.empty()) {
          _queue.swap(c);
          _queue.clear();
        }
        return(c.size());
      }
      
    protected:
      uint32_t                      _maxLength;
      std::deque<_ValueType>        _queue;
      mutable std::mutex            _mutex;
      std::atomic<bool>             _signalled;
      std::unique_lock<std::mutex>  _lock;
      std::condition_variable       _cv;
     
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      void Lock()
      {
        _lock.lock();
        return;
      }
     
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      void Unlock()
      {
        _lock.unlock();
        return;
      }
     
    };
 
 
  }  // namespace Thread
  
}  // namespace Dwm
 
#endif  // _DWMTHREADQUEUE_HH_
 
//---------------------------- emacs settings -----------------------------
//  Local Variables:
//  mode: C++
//  tab-width: 2
//  indent-tabs-mode: nil
//  c-basic-offset: 2
//  End:
//-------------------------------------------------------------------------