block-flow.hpp

Go to the documentation of this file.

/*
  BLOCK-FLOW.hpp  -  specialised custom allocator to manage scheduler data

  Copyright (C)         Lumiera.org
    2023,               Hermann Vosseler <Ichthyostega@web.de>

  This program 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.

  This program is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  GNU General Public License for more details.

  You should have received a copy of the GNU General Public License
  along with this program; if not, write to the Free Software
  Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.

*/


#ifndef SRC_VAULT_GEAR_BLOCK_FLOW_H_
#define SRC_VAULT_GEAR_BLOCK_FLOW_H_


#include "vault/common.hpp"
#include "vault/gear/activity.hpp"
#include "vault/mem/extent-family.hpp"
#include "lib/time/timevalue.hpp"
#include "lib/iter-explorer.hpp"
#include "lib/format-util.hpp"
#include "lib/nocopy.hpp"
#include "lib/util.hpp"

#include <utility>


namespace vault{
namespace gear {
  
  using util::isnil;
  using lib::time::Time;
  using lib::time::FSecs;
  using lib::time::TimeVar;
  using lib::time::Duration;
  using lib::time::FrameRate;
  
  namespace err = lumiera::error;
  
  
  namespace blockFlow {
    
    const size_t BLOCK_EXPAND_SAFETY_LIMIT = 3000;
    
    
    struct DefaultConfig
      {
        /* === characteristic parameters === */
        const static size_t EPOCH_SIZ = 100;    
        const Duration DUTY_CYCLE{FSecs(1)};    
        const size_t INITIAL_STREAMS = 2;       
        
        /* === algorithm tuning settings === */
        const double TARGET_FILL = 0.90;        
        const double BOOST_FACTOR = 0.85;       
        const double DAMP_THRESHOLD = 0.08;     
        
        /* === contextual assumptions === */
        const size_t ACTIVITIES_PER_FRAME = 10; 
        const size_t REFERENCE_FPS  =  25;      
        const size_t OVERLOAD_LIMIT =  60;      
      };
    
    struct RenderConfig
      : DefaultConfig
      {
        const static size_t EPOCH_SIZ = 500;
        const size_t INITIAL_STREAMS = 5;
      };
    
    template<class CONF>
    struct Strategy
      {
        CONF const&
        config()  const
          {           // Meyers Singleton
            static const CONF configInstance;
            return configInstance;
          }
        
        size_t
        framesPerEpoch()  const
          {
            return config().EPOCH_SIZ / config().ACTIVITIES_PER_FRAME;
          }
        
        size_t
        initialFrameRate()  const
          {
            return config().INITIAL_STREAMS * config().REFERENCE_FPS;
          }
        
        Duration
        initialEpochStep()  const
          {
            return Duration{TimeValue(framesPerEpoch() * TimeValue::SCALE / initialFrameRate())};
          }
        
        size_t
        initialEpochCnt()  const       
          {
            return util::max(2*_raw(config().DUTY_CYCLE) / _raw(initialEpochStep()), 2u);
          }
        
        size_t
        averageEpochs()  const
          {
            return util::max (initialEpochCnt(), 6u);
          }
        
        double
        boostFactor()  const
          {
            return config().BOOST_FACTOR;
          }
        
        double
        boostFactorOverflow()  const   
          {
            return pow(config().BOOST_FACTOR, 5.0/config().EPOCH_SIZ);
          }
        
        Duration
        timeStep_cutOff()  const       
          {
            return Duration{TimeValue(_raw(initialEpochStep()) / config().OVERLOAD_LIMIT)};
          }
      };
    
    
    
    
    template<class ALO>
    class Epoch
      : public ALO::Extent
      {
        using RawIter = typename ALO::iterator;
        using SIZ     = typename ALO::Extent::SIZ;
        
        Epoch()    = delete;
        
      public:
        struct EpochGate
          : Activity
          {
            EpochGate()
              : Activity{int(0), Time::ANYTIME}
              {
                // initialise allocation usage marker: start at last usable slot
                next = this + (Epoch::SIZ() - 1);
                ENSURE (next != this);
              }
            // default copyable
            
            activity::Instant&
            deadline()
              {
                return data_.condition.dead;
              }
            
            bool
            isAlive (Time deadline)
              {
                return data_.condition.isHold() // NOTE: expected callback keeps alive
                    or not data_.condition.isDead (deadline);
              }
            
            size_t
            filledSlots()  const
              {
                const Activity* firstAllocPoint{this + (Epoch::SIZ()-1)};
                return firstAllocPoint - next;
              }
            
            bool
            hasFreeSlot()  const
              { // see C++ § 5.9 : comparison of pointers within same array
                return next > this;
              }
            
            Activity*
            claimNextSlot()
              {
                REQUIRE (hasFreeSlot());
                return next--;
              }
          };
        
        
        EpochGate& gate() { return static_cast<EpochGate&> ((*this)[0]); }
        Time   deadline() { return Time{gate().deadline()};              }
        
        double
        getFillFactor()
          {
            return double(gate().filledSlots()) / (SIZ()-1);
          }
        
        
        static Epoch&
        implantInto (RawIter storageSlot)
          {
            Epoch& target = static_cast<Epoch&> (*storageSlot);
            new(&target[0]) EpochGate{};
            return target;
          }
        
        static Epoch&
        setup (RawIter storageSlot, Time deadline)
          {
            Epoch& newEpoch{implantInto (storageSlot)};
            newEpoch.gate().deadline() = deadline;
            return newEpoch;
          }
      };
  
  
  }//(End)namespace blockFlow
  
  template<class CONF>
  class FlowDiagnostic;
  
  
  
  
  /******************************************************/
  template<class CONF = blockFlow::DefaultConfig>
  class BlockFlow
    : public blockFlow::Strategy<CONF>
    , util::NonCopyable
    {
      constexpr static size_t EPOCH_SIZ = CONF::EPOCH_SIZ;
      
    public:
      using Allocator = mem::ExtentFamily<Activity, EPOCH_SIZ>;
      using Strategy  = blockFlow::Strategy<CONF>;
      using RawIter   = typename Allocator::iterator;
      using Extent    = typename Allocator::Extent;
      using Epoch     = blockFlow::Epoch<Allocator>;
      
      using Strategy::config;
      
    private:
      Allocator alloc_;
      TimeVar epochStep_;
      
      
      static Epoch&
      asEpoch (Extent& extent)
        {
          return static_cast<Epoch&> (extent);
        }
      
      class StorageAdaptor
        : public RawIter
        {
          Epoch* curr_{nullptr};
          
          Epoch*
          accessEpoch()
            {
              return RawIter::checkPoint()? & asEpoch (RawIter::yield())
                                          : nullptr;
            }
          
        public:
          StorageAdaptor()  = default;
          StorageAdaptor(RawIter it)
            : RawIter{it}
            , curr_{accessEpoch()}
            { }
          
          bool
          checkPoint()  const
            {
              return bool(curr_);
            }
          
          Epoch&
          yield()  const
            {
              return *curr_;
            }
          
          void
          iterNext()
            {
              RawIter::validatePos(curr_);
              RawIter::iterNext();
              curr_ = accessEpoch();
            }
          
          void
          expandAlloc (size_t cnt =1)
            {
              RawIter::expandAlloc(cnt);
              curr_ = accessEpoch();
            }
        };
      
      
      
    public:
      BlockFlow()
        : alloc_{Strategy::initialEpochCnt()}
        , epochStep_{Strategy::initialEpochStep()}
        { }
      
      Duration
      getEpochStep()  const
        {
          return Duration{epochStep_};
        }
      
      void
      adjustEpochStep (double factor)
        {
          double stretched = _raw(epochStep_) * factor;
          gavl_time_t microTicks(floor (stretched));
          epochStep_ = TimeValue{microTicks};
        }
      
      
      
      using EpochIter = lib::IterableDecorator<Epoch, StorageAdaptor>;
      
      
      class AllocatorHandle
        {
          EpochIter epoch_;
          BlockFlow* flow_;
          
        public:
          AllocatorHandle(RawIter slot, BlockFlow* parent)
            : epoch_{slot}
            , flow_{parent}
          { }
          
          /*************************************************/
          template<typename...ARGS>
          Activity&
          create (ARGS&& ...args)
            {
              return *new(claimSlot()) Activity {std::forward<ARGS> (args)...};
            }
          
          Time currDeadline() const { return epoch_->deadline(); }
          bool hasFreeSlot()  const { return epoch_->gate().hasFreeSlot(); }
          
          
        private:
          void*
          claimSlot() 
            {
              while (not (epoch_ and
                          epoch_->gate().hasFreeSlot()))
                  // Epoch overflow...
                {//  shift to following Epoch; possibly allocate
                  if (not epoch_)
                    {
                      auto lastDeadline = flow_->lastEpoch().deadline();
                      epoch_.expandAlloc(); // may throw out-of-memory..
                      ENSURE (epoch_);
                      Epoch::setup (epoch_, lastDeadline + flow_->getEpochStep());
                    }
                  else
                    {
                      flow_->markEpochOverflow();
                      ++epoch_;
                    }
                }
              return epoch_->gate().claimNextSlot();
            }
        };
      
      
      /* ===== public BlockFlow API ===== */
      
      AllocatorHandle
      until (Time deadline)
        {
          if (isnil (alloc_))
            {//just create new Epoch one epochStep ahead
              alloc_.openNew();
              Epoch::setup (alloc_.begin(), deadline + Time{epochStep_});
              return AllocatorHandle{alloc_.begin(), this};
            }
          else
            {//find out how the given time relates to existing Epochs
              if (firstEpoch().deadline() >= deadline)
                // way into the past ... put it in the first available Epoch
                return AllocatorHandle{alloc_.begin(), this};
              else
              if (lastEpoch().deadline() < deadline)
                {  // a deadline beyond the established Epochs...
                  //  create a grid of new epochs up to the requested point
                  TimeVar lastDeadline = lastEpoch().deadline();
                  auto distance = _raw(deadline) - _raw(lastDeadline);
                  EpochIter nextEpoch{alloc_.end()};
                  ENSURE (not nextEpoch);      // not valid yet, but we will allocate starting there...
                  auto requiredNew = distance / _raw(epochStep_);
                  ___sanityCheckAlloc(requiredNew);
                  if (distance % _raw(epochStep_) > 0)
                    ++requiredNew;  // fractional:  requested deadline lies within last epoch
                  nextEpoch.expandAlloc (requiredNew);
                  // nextEpoch now points to the first new Epoch
                  for ( ; 0 < requiredNew; --requiredNew)
                    {
                      REQUIRE (nextEpoch);
                      lastDeadline += epochStep_;
                      Epoch::setup (nextEpoch, lastDeadline);
                      if (deadline <= lastDeadline)
                        {
                          ENSURE (requiredNew == 1);
                          return AllocatorHandle{nextEpoch, this};
                        }     // break out and return handle to allocate into the matching Epoch
                      ++nextEpoch;
                    }
                  NOTREACHED ("Logic of counting new Epochs");
                }
              else
                for (EpochIter epochIt{alloc_.begin()}; epochIt; ++epochIt)
                  if (epochIt->deadline() >= deadline)
                    return AllocatorHandle{epochIt, this};
              
              NOTREACHED ("Inconsistency in BlockFlow Epoch deadline organisation");
            }
        }
      
      void
      discardBefore (Time deadline)
        {
          if (isnil (alloc_)
              or firstEpoch().deadline() > deadline)
            return;
          
          size_t toDiscard{0};
          for (Epoch& epoch : allEpochs())
            {
              if (epoch.gate().isAlive (deadline))
                break;
              ++toDiscard;
              auto currDeadline = epoch.deadline();
              auto epochDuration = currDeadline - updatePastDeadline(currDeadline);
              markEpochUnderflow (epochDuration, epoch.getFillFactor());
            }
          // ask to discard the enumerated Extents
          alloc_.dropOld (toDiscard);
        }
      
      
      void
      markEpochOverflow()
        {
          if (epochStep_ > _cache_timeStep_cutOff)
            adjustEpochStep (_cache_boostFactorOverflow);
        }
      // caching access to the config saves 15-30% per allocation
      Duration _cache_timeStep_cutOff = Strategy::timeStep_cutOff();
      double _cache_boostFactorOverflow = Strategy::boostFactorOverflow();
      
      void
      markEpochUnderflow (TimeVar actualLen, double fillFactor)
        {
          auto interpolate = [&](auto f, auto v1, auto v2) { return f*v2 + (1-f)*v1; };
          
          // use actual fill as signal, set desired fill-level as goal
          fillFactor /= config().TARGET_FILL;
          auto THRESH = config().DAMP_THRESHOLD;
          double adjust =
            fillFactor > THRESH? fillFactor   //  limit signal for almost empty Epochs to avoid overshooting
                               : interpolate (1 - fillFactor/THRESH, fillFactor, Strategy::boostFactor());
          
          // damped adjustment towards ideal size
          double contribution = double(_raw(actualLen)) / _raw(epochStep_) / adjust;
          
          // Exponential MA: mean ≔ mean · (N-1)/N  + newVal/N
          auto N = Strategy::averageEpochs();
          double avgFactor = (contribution + N-1) / N;    // contribution = newVal / mean  => can extract factor
          adjustEpochStep (avgFactor);
        }
      
      
      void
      announceAdditionalFlow (FrameRate additionalFps)
        {
          FrameRate currFps{Strategy::framesPerEpoch(), Duration{epochStep_}};
          currFps += additionalFps;
          TimeVar adaptedSpacing = Strategy::framesPerEpoch() / currFps;
          epochStep_ = util::max (adaptedSpacing, _cache_timeStep_cutOff);
        }
      
      
    private:
      Epoch&
      firstEpoch()
        {
          REQUIRE (not isnil (alloc_));
          return asEpoch(*alloc_.begin());
        }
      Epoch&
      lastEpoch()
        {
          REQUIRE (not isnil (alloc_));
          return asEpoch(*alloc_.last());
        }
      
      EpochIter
      allEpochs()
        {
          return alloc_.begin();
        }
      
      Time
      updatePastDeadline (TimeVar newDeadline)
        {
          if (pastDeadline_ == Time::ANYTIME)
            pastDeadline_ = newDeadline - epochStep_;
          TimeVar previous = pastDeadline_;
          pastDeadline_ = newDeadline;
          return previous;
        }
      TimeVar pastDeadline_{Time::ANYTIME};
      
      void
      ___sanityCheckAlloc (size_t newBlockCnt)
        {
          if (newBlockCnt > blockFlow::BLOCK_EXPAND_SAFETY_LIMIT)
            throw err::Fatal{"Deadline expansion causes allocation of "
                              +util::showSize(newBlockCnt) +"blocks > "
                              +util::showSize(blockFlow::BLOCK_EXPAND_SAFETY_LIMIT)
                            , err::LUMIERA_ERROR_CAPACITY};
        }
      
      
      friend class FlowDiagnostic<CONF>;
    };
  
  
  
  
  
  
  
  
  /* ===== Test / Diagnostic ===== */
  
  template<class CONF>
  class FlowDiagnostic
    {
      using Epoch = typename BlockFlow<CONF>::Epoch;
      
      BlockFlow<CONF>& flow_;
      
    public:
      FlowDiagnostic(BlockFlow<CONF>& theFlow)
        : flow_{theFlow}
      { }
      
      Time   first()     { return flow_.firstEpoch().deadline();}
      Time   last()      { return flow_.lastEpoch().deadline(); }
      size_t cntEpochs() { return watch(flow_.alloc_).active(); }
      size_t poolSize()  { return watch(flow_.alloc_).size();   }
      
      TimeValue
      find (Activity& someActivity)
        {
          for (Epoch& epoch : flow_.allEpochs())
            for (Activity& act : epoch)
              if (util::isSameObject (act, someActivity))
                return epoch.deadline();
          return Time::NEVER;
        }
      
      std::string
      allEpochs()
        {
          if (isnil (flow_.alloc_)) return "";
          auto deadlines = lib::explore (flow_.allEpochs())
                               .transform([](Epoch& a){ return TimeValue{a.deadline()}; });
          return util::join(deadlines, "|");
        }
      
      size_t
      cntElm()
        {
          size_t cnt{0};
          for (auto& epoch : flow_.allEpochs())
            cnt += epoch.gate().filledSlots();
          return cnt;
        }
    };
  
  template<class CONF>
  inline FlowDiagnostic<CONF>
  watch (BlockFlow<CONF>& theFlow)
  {
    return FlowDiagnostic{theFlow};
  }
  
  
  
}} // namespace vault::gear
#endif /*SRC_VAULT_GEAR_BLOCK_FLOW_H_*/