DwmLoadBalancer.hh

Go to the documentation of this file.

//===========================================================================
// @(#) $DwmPath$
//===========================================================================
//  Copyright (c) Daniel W. McRobb 2008, 2009, 2016, 2018
//  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 _DWMLOADBALANCER_HH_
#define _DWMLOADBALANCER_HH_
 
#include <algorithm>
#include <atomic>
#include <deque>
#include <chrono>
#include <thread>
#include <vector>
 
#include "DwmThreadQueue.hh"
 
namespace Dwm {
 
  //--------------------------------------------------------------------------
  //--------------------------------------------------------------------------
  template <typename ItemT>
  class LoadBalancer
  {
  public:
    //------------------------------------------------------------------------
    //------------------------------------------------------------------------
    class Worker
    {
    public:
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      Worker(std::chrono::microseconds waitNotEmptyMicroseconds = std::chrono::microseconds(1000))
          : _workQueue(), _keepRunning(false), _thread(),
            _waitUsecs(waitNotEmptyMicroseconds)
      {}
 
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      virtual ~Worker()
      {
        Stop();
      }
      
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      bool AddWork(const ItemT & item)
      {
        return _workQueue.PushBack(item);
      }
      
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      template <typename InputIterator>
      bool AddWork(InputIterator firstIter, InputIterator lastIter)
      {
        return (_workQueue.PushBack(firstIter, lastIter) > 0);
      }
      
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      uint32_t QueueLength() const
      {
        return _workQueue.Length();
      }
      
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      bool ReadyForWork() const
      {
        return (_keepRunning
                && ((! _workQueue.MaxLength())
                    || (_workQueue.Length() < _workQueue.MaxLength())));
      }
 
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      bool ReadyForWork(size_t numEntries) const
      {
        return (_keepRunning
                && ((! _workQueue.MaxLength())
                    || ((_workQueue.Length() + numEntries)
                        <= _workQueue.MaxLength())));
      }
 
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      void MaxWork(uint32_t maxItems)
      {
        _workQueue.MaxLength(maxItems);
      }
      
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      bool Start()
      {
        bool  rc = false;
        _keepRunning = true;
        _thread = std::thread(&Worker::Run, this);
        rc = true;
        return rc;
      }
      
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      void Stop()
      {
        _keepRunning = false;
        if (_thread.joinable()) {
          _thread.join();
        }
        return;
      }
 
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      bool IsRunning()
      {
        return (_thread.joinable());
      }
      
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      void Run()
      {
        while (_keepRunning || (_workQueue.Length())) {
          if (_workQueue.TimedWaitForNotEmpty(_waitUsecs.load())) {
            std::deque<ItemT>  myCopy;
            _workQueue.Swap(myCopy);
            //  try processing in bulk
            if (! ProcessWork(myCopy)) {
              //  else fall back to one at a time.
              for (auto i : myCopy) {
                ProcessWork(i);
              }
            }
          }
        }
        return;
      }
      
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      virtual void ProcessWork(ItemT & item) = 0;
 
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      virtual bool ProcessWork(std::deque<ItemT> & items)
      {
        return false;
      }
 
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      std::chrono::microseconds WaitUsecs() const
      {
        return _waitUsecs;
      }
      
    protected:
      Thread::Queue<ItemT>                    _workQueue;
      std::atomic<bool>                       _keepRunning;
      std::thread                             _thread;
      std::atomic<std::chrono::microseconds>  _waitUsecs;
    };
 
    //------------------------------------------------------------------------
    //------------------------------------------------------------------------
    void AddWorker(Worker *worker)
    {
      std::lock_guard<std::mutex>  lock(_mtx);
      _workers.push_back(std::unique_ptr<Worker>(worker));
      return;
    }
 
    //------------------------------------------------------------------------
    //------------------------------------------------------------------------
    bool AddWork(ItemT item)
    {
      bool  workAdded = false;
      {
        std::lock_guard<std::mutex>  lock(_mtx);
        if (_workers.empty()) {
          //  No workers to perform work!!!
          return false;
        }
      }
      while (! workAdded) {
        {
          //  in separate scope so we don't hold lock for too long
          std::lock_guard<std::mutex>  lock(_mtx);
          auto  w = std::min_element(_workers.begin(), _workers.end(),
                                     [&] (const std::unique_ptr<Worker> & a,
                                          const std::unique_ptr<Worker> & b)
                                     { return (a->QueueLength()
                                               < b->QueueLength()); });
          if (w != _workers.end()) {
            workAdded = (*w)->AddWork(item);
          }
        }
        if (! workAdded) {
          std::this_thread::sleep_for(MinimumWaitForWorkerReady());
        }
      }
      return workAdded;
    }
 
    //------------------------------------------------------------------------
    //------------------------------------------------------------------------
    template <typename InputIterator>
    bool AddWork(InputIterator firstIter, InputIterator lastIter)
    {
      bool  workAdded = false;
      {
        std::lock_guard<std::mutex>  lock(_mtx);
        if (_workers.empty()) {
          //  No workers to perform work!!!
          return false;
        }
      }
      while (! WorkerReady(lastIter - firstIter)) {
        std::this_thread::sleep_for(std::chrono::microseconds(MinimumWaitForWorkerReady()));
      }
 
      std::lock_guard<std::mutex>  lock(_mtx);
      auto  w = std::min_element(_workers.begin(), _workers.end(),
                                 [&] (const std::unique_ptr<Worker> & a,
                                      const std::unique_ptr<Worker> & b)
                                 { return (a->QueueLength() < b->QueueLength()); });
      return (*w)->AddWork(firstIter, lastIter);
    }
 
    //------------------------------------------------------------------------
    //------------------------------------------------------------------------
    void Stop()
    {
      std::lock_guard<std::mutex>  lock(_mtx);
      for (auto & w : _workers) {
        w->Stop();
      }
      return;
    }
 
    //------------------------------------------------------------------------
    //------------------------------------------------------------------------
    const std::vector<std::unique_ptr<Worker>> & Workers() const
    {
      return _workers;
    }
    
  private:
    mutable std::mutex                    _mtx;
    std::vector<std::unique_ptr<Worker>>  _workers;
    
    //------------------------------------------------------------------------
    //------------------------------------------------------------------------
    bool WorkerReady()
    {
      bool  rc = false;
      std::lock_guard<std::mutex>  lock(_mtx);
      for (auto & w : _workers) {
        if (w->ReadyForWork()) {
          rc = true;
          break;
        }
      }
      return rc;
    }
 
    //------------------------------------------------------------------------
    //------------------------------------------------------------------------
    bool WorkerReady(size_t numEntries) const
    {
      bool  rc = false;
      std::lock_guard<std::mutex>  lock(_mtx);
      for (auto & w : _workers) {
        if (w->ReadyForWork(numEntries)) {
          rc = true;
          break;
        }
      }
      return rc;
    }
    
    //------------------------------------------------------------------------
    //------------------------------------------------------------------------
    std::chrono::microseconds MinimumWaitForWorkerReady() const
    {
      std::chrono::microseconds  rc(1000);
      std::lock_guard<std::mutex>  lock(_mtx);
      auto  w = std::min_element(_workers.begin(), _workers.end(),
                                 [&] (const std::unique_ptr<Worker> & a,
                                      const std::unique_ptr<Worker> & b)
                                 { return (a->WaitUsecs() < b->WaitUsecs())\
; });
      if (w != _workers.end()) {
        rc = std::chrono::microseconds((*w)->WaitUsecs());
      }
      return rc;
    }
    
  };
  
}  // namespace Dwm
 
 
#endif  // _DWMLOADBALANCER_HH_
 
 
//---------------------------- emacs settings -----------------------------
//  Local Variables:
//  mode: C++
//  tab-width: 2
//  indent-tabs-mode: nil
//  c-basic-offset: 2
//  End:
//-------------------------------------------------------------------------