load-controller.hpp

Go to the documentation of this file.

/*
  LOAD-CONTROLLER.hpp  -  coordinator for scheduler resource usage
 
   Copyright (C)
     2023,            Hermann Vosseler <Ichthyostega@web.de>
 
 **Lumiera** 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. See the file COPYING for further details.
 
*/
 
 
#ifndef SRC_VAULT_GEAR_LOAD_CONTROLLER_H_
#define SRC_VAULT_GEAR_LOAD_CONTROLLER_H_
 
 
#include "lib/error.hpp"
//#include "vault/gear/block-flow.hpp"
//#include "vault/gear/activity-lang.hpp"
//#include "lib/symbol.hpp"
#include "lib/time/timevalue.hpp"
#include "lib/nocopy.hpp"
#include "lib/util.hpp"
#include "lib/format-cout.hpp"
 
//#include <string>
#include <cmath>
#include <atomic>
#include <chrono>
#include <utility>
#include <functional>
 
 
namespace vault{
namespace gear {
  
//  using util::isnil;
//  using std::string;
  
  using util::max;
  using util::limited;
  using lib::time::Time;
  using lib::time::FSecs;
  using lib::time::TimeVar;
  using lib::time::TimeValue;
  using lib::time::Duration;
  using lib::time::Offset;
  using std::chrono_literals::operator ""ms;
  using std::chrono_literals::operator ""us;
  using std::function;
  using std::atomic_int64_t;
  using std::memory_order_relaxed;
  
  namespace { // Scheduler default config
    
    inline TimeValue
    _uTicks (std::chrono::microseconds us)
    {
      return TimeValue{us.count()};
    }
    
    
    Duration SLEEP_HORIZON{_uTicks (20ms)}; 
    Duration WORK_HORIZON {_uTicks ( 5ms)}; 
    Duration NEAR_HORIZON {_uTicks (50us)}; 
    Duration STANDARD_LAG {_uTicks(200us)}; 
    
    const double LAG_SAMPLE_DAMPING = 2;    
  }
  
  
  
  class LoadController
    : util::NonCopyable
    {
    public:
      struct Wiring
        {
          function<size_t()> maxCapacity      {[]{ return 1; }};
          function<size_t()> currWorkForceSize{[]{ return 0; }};
          function<void(uint)> stepUpWorkForce{[](uint){/*NOP*/}};
        };
      
      explicit
      LoadController (Wiring&& wiring)
        : wiring_{std::move (wiring)}
        { }
      
      LoadController()
        : LoadController{Wiring{}}
        { }
      
    private:
      const Wiring wiring_;
      
      TimeVar tendedHead_{Time::ANYTIME};
      
      
      atomic_int64_t sampledLag_{0};
      
      void
      markLagSample (Time head, Time now)
        {                                        // negative when free capacity
          double lag = _raw(std::clamp<TimeVar> (now - (head.isRegular()? head:now)
                                                , -SLEEP_HORIZON
                                                , WORK_HORIZON));
          const double alpha = LAG_SAMPLE_DAMPING / (1 + wiring_.maxCapacity());
          int64_t average = sampledLag_.load (memory_order_relaxed);
          int64_t newAverage;
          do newAverage = std::floor (lag*alpha + (1-alpha)*average);
          while (not sampledLag_.compare_exchange_weak (average, newAverage, memory_order_relaxed));
        }
      
    public:
      int64_t
      averageLag()  const
        {
          return sampledLag_.load (memory_order_relaxed);
        }
      
      int64_t
      setCurrentAverageLag (int64_t lag)
        {
          return sampledLag_.exchange(lag, memory_order_relaxed);
        }
      
      double
      effectiveLoad()  const
        {
          double lag = averageLag();
          lag -= _raw(STANDARD_LAG);
          lag /= _raw(WORK_HORIZON);
          lag *= 10;
          double lagFactor = lag<0? 1/(1-lag): 1+lag;
          double loadFactor = wiring_.currWorkForceSize() / double(wiring_.maxCapacity());
          return loadFactor * lagFactor;
        }
      
      void
      updateState (Time)
        {
          //
          auto lag = averageLag();
          if (lag > _raw(WORK_HORIZON))
              wiring_.stepUpWorkForce(+4);
          else
          if (averageLag() > 2*_raw(STANDARD_LAG))
              wiring_.stepUpWorkForce(+1);
        }
      
      void
      markWorkerExit()
        {
        }
      
      void
      ensureCapacity (Time startHorizon)
        {
          if (startHorizon > 2* SLEEP_HORIZON)
            return;
          if (averageLag() > 2*_raw(STANDARD_LAG))
            wiring_.stepUpWorkForce(+1);
        }
      
      bool
      tendedNext (Time nextHead)  const
        {
          return not nextHead.isRegular()   // note: empty queue reports Time::NEVER
              or nextHead == tendedHead_;
        }
      
      void
      tendNext (Time nextHead)
        {
          tendedHead_ = nextHead;
        }
      
      
      
      enum Capacity {DISPATCH   
                    ,TENDNEXT   
                    ,SPINTIME   
                    ,NEARTIME   
                    ,WORKTIME   
                    ,IDLEWAIT   
                    };
      
      static Capacity
      classifyTimeHorizon (Offset off)
        {
          if (off > SLEEP_HORIZON) return IDLEWAIT;
          if (off > WORK_HORIZON)  return WORKTIME;
          if (off > NEAR_HORIZON)  return NEARTIME;
          if (off > Time::ZERO)    return SPINTIME;
          else                     return DISPATCH;
        }
      
      
      Capacity
      markOutgoingCapacity (Time head, Time now)
        {
          auto horizon = classifyTimeHorizon (Offset{head - now});
          return horizon > SPINTIME
             and not tendedNext(head)? TENDNEXT
                                     : horizon==IDLEWAIT ? WORKTIME    // re-randomise sleeper cycles
                                                         : horizon;
        }
      
      Capacity
      markIncomingCapacity (Time head, Time now)
        {
          markLagSample (head,now);
          return classifyTimeHorizon (Offset{head - now})
               > NEARTIME ? IDLEWAIT
                          : markOutgoingCapacity(head,now);
        }
      
      
      
      Offset
      scatteredDelayTime (Time now, Capacity capacity)
        {
          auto scatter = [&](Duration horizon)
                            {
                              lib::time::raw_time_64 wrap = hash_value(now) % _raw(horizon);
                              ENSURE (0 <= wrap and wrap < _raw(horizon));
                              return TimeValue{wrap};
                            };
          
          TimeVar headDistance = max (tendedHead_-now, Time::ZERO);
          
          switch (capacity) {
            case DISPATCH:
              return Offset::ZERO;
            case SPINTIME:
              return Offset::ZERO;
            case TENDNEXT:
              return Offset{headDistance};
            case NEARTIME:
              return Offset{headDistance  + scatter(Offset{limited (NEAR_HORIZON,headDistance,WORK_HORIZON)})};
            case WORKTIME:
            case IDLEWAIT:
              return Offset{headDistance  + scatter(SLEEP_HORIZON)};
            default:
              NOTREACHED ("uncovered work capacity classification.");
            }
        }
    };
  
  
  
}}// namespace vault::gear
#endif /*SRC_VAULT_GEAR_LOAD_CONTROLLER_H_*/