DwmLoadBalancer.hh

Go to the documentation of this file.

//===========================================================================
// @(#) $DwmPath: dwm/libDwm/tags/libDwm-0.6.0/include/DwmLoadBalancer.hh 8401 $
// @(#) $Id: DwmLoadBalancer.hh 8401 2016-04-17 06:44:31Z dwm $
//===========================================================================
//  Copyright (c) Daniel W. McRobb 2008, 2009, 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 _DWMLOADBALANCER_HH_
#define _DWMLOADBALANCER_HH_

extern "C" {
  #include <unistd.h>
}

#include <algorithm>
#include <iostream>
#include <queue>
#include <string>
#include <thread>
#include <vector>

#include "DwmSysLogger.hh"
#include "DwmThreadQueue.hh"

namespace Dwm {

  //--------------------------------------------------------------------------
  //--------------------------------------------------------------------------
  template <typename ItemT>
  class LoadBalancer
  {
  public:
    //------------------------------------------------------------------------
    //------------------------------------------------------------------------
    class Worker
    {
    public:
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      Worker(uint32_t waitNotEmptyMicroseconds = 1000)
          : _workQueue(), _keepRunning(false), _thread(),
            _waitUsecs(waitNotEmptyMicroseconds)
      {}

      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      virtual ~Worker()
      {
        Stop();
      }
      
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      void AddWork(ItemT item)
      {
        _workQueue.PushBack(item);
        return;
      }
      
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      template <typename InputIterator>
      void AddWork(InputIterator firstIter, InputIterator lastIter)
      {
        _workQueue.PushBack(firstIter, lastIter);
        return;
      }
      
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      uint32_t QueueLength() const
      {
        return _workQueue.Length();
      }
      
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      bool ReadyForWork() const
      {
        return (_keepRunning
                && ((! _workQueue.MaxLength())
                    || (_workQueue.Length() < _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) {
          if (_workQueue.TimedWaitForNotEmpty(std::chrono::microseconds(_waitUsecs))) {
            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;
      }
      
    protected:
      Thread::Queue<ItemT>  _workQueue;
      bool                  _keepRunning;
      std::thread           _thread;
      uint32_t              _waitUsecs;
    };

    //------------------------------------------------------------------------
    //------------------------------------------------------------------------
    class CompareWorkers
    {
    public:
      //----------------------------------------------------------------------
      //----------------------------------------------------------------------
      bool operator () (const Worker * a,
                        const Worker * b) const
      {
        return (a->QueueLength() < b->QueueLength());
      }
    };

    //------------------------------------------------------------------------
    //------------------------------------------------------------------------
    void AddWorker(Worker *worker)
    {
      _workers.push_back(worker);
      return;
    }

    //------------------------------------------------------------------------
    //------------------------------------------------------------------------
    void AddWork(ItemT item)
    {
      auto  w = std::min_element(_workers.begin(), _workers.end(),
                                 CompareWorkers());
      if (w != _workers.end()) {
        (*w)->AddWork(item);
      }
      return;
    }

    //------------------------------------------------------------------------
    //------------------------------------------------------------------------
    template <typename InputIterator>
    void AddWork(InputIterator firstIter, InputIterator lastIter)
    {
      while (! WorkerReady()) {
        usleep(1000);
      }
      auto  w = std::min_element(_workers.begin(), _workers.end(),
                                 CompareWorkers());
      (*w)->AddWork(firstIter, lastIter);
      return;
    }

    //------------------------------------------------------------------------
    //------------------------------------------------------------------------
    void Stop()
    {
      for (auto w : _workers) {
        w->Stop();
      }
      return;
    }

    //------------------------------------------------------------------------
    //------------------------------------------------------------------------
    const std::vector<Worker *> & Workers() const
    {
      return _workers;
    }
    
  private:
    std::vector<Worker *>  _workers;

    //------------------------------------------------------------------------
    //------------------------------------------------------------------------
    bool WorkerReady()
    {
      bool  rc = false;
      for (auto w : _workers) {
        if (w->ReadyForWork()) {
          rc = true;
          break;
        }
      }
      return rc;
    }
    
  };
  
}  // namespace Dwm


#endif  // _DWMLOADBALANCER_HH_