scheduler.hpp

Go to the documentation of this file.

/*
  SCHEDULER.hpp  -  coordination of render activities under timing and dependency constraints

   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_SCHEDULER_H_
#define SRC_VAULT_GEAR_SCHEDULER_H_


#include "lib/error.hpp"
#include "vault/gear/block-flow.hpp"
#include "vault/gear/work-force.hpp"
#include "vault/gear/activity-lang.hpp"
#include "vault/gear/scheduler-commutator.hpp"
#include "vault/gear/scheduler-invocation.hpp"
#include "vault/gear/load-controller.hpp"
#include "vault/gear/engine-observer.hpp"
#include "vault/real-clock.hpp"
#include  "lib/nocopy.hpp"

#include <optional>
#include <utility>


namespace vault{
namespace gear {
  
  using std::move;
  using lib::time::Time;
  using lib::time::FSecs;
  using lib::time::Offset;
  
  namespace test {  // declared friend for test access
    class SchedulerService_test;
  }
  
  namespace { // Scheduler default config
    
    const auto   IDLE_WAIT = 20ms;            
    const size_t DISMISS_CYCLES = 100;        
    Offset DUTY_CYCLE_PERIOD{FSecs(1,20)};    
    Offset DUTY_CYCLE_TOLERANCE{FSecs(2,10)}; 
    Offset FUTURE_PLANNING_LIMIT{FSecs{20}};  
  }
  
  
  class Scheduler;
  
  class ScheduleSpec
    : util::MoveAssign
    {
      Job job_;
      TimeVar start_{Time::ANYTIME};
      TimeVar death_{Time::NEVER};
      ManifestationID manID_{};
      bool isCompulsory_{false};
      
      Scheduler* theScheduler_;
      std::optional<activity::Term> term_;
      
    public:
      ScheduleSpec (Scheduler& sched, Job job)
        : job_{job}
        , theScheduler_{&sched}
        , term_{std::nullopt}
        { }
      
      ScheduleSpec
      startOffset (microseconds afterNow)
        {
          start_ = RealClock::now() + _uTicks(afterNow);
          return move(*this);
        }
      
      ScheduleSpec
      startTime (Time fixedTime)
        {
          start_ = fixedTime;
          return move(*this);
        }
      
      ScheduleSpec
      lifeWindow (microseconds afterStart)
        {
          death_ = start_ + _uTicks(afterStart);
          return move(*this);
        }
      
      ScheduleSpec
      manifestation (ManifestationID manID)
        {
          manID_ = manID;
          return move(*this);
        }
      
      ScheduleSpec
      compulsory (bool indeed =true)
        {
          isCompulsory_ = indeed;
          return move(*this);
        }
      
      
      ScheduleSpec post();
      
      ScheduleSpec linkToSuccessor (ScheduleSpec&,  bool unlimitedTime =false);
      ScheduleSpec linkToPredecessor(ScheduleSpec&, bool unlimitedTime =false);
    private:
      void maybeBuildTerm();
    };
  
  
  
  /******************************************************/
  class Scheduler
    : util::NonCopyable
    {
      struct Setup : work::Config
        {
          Scheduler& scheduler;
          activity::Proc doWork() { return scheduler.doWork();           }
          void finalHook (bool _) { scheduler.handleWorkerTermination(_);}
        };
      
      
      SchedulerInvocation layer1_;
      SchedulerCommutator layer2_;
      WorkForce<Setup> workForce_;
      
      ActivityLang activityLang_;
      LoadController loadControl_;
      EngineObserver& engineObserver_;
      
      
    public:
      Scheduler (BlockFlowAlloc& activityAllocator
                ,EngineObserver& engineObserver)
        : layer1_{}
        , layer2_{}
        , workForce_{Setup{IDLE_WAIT, DISMISS_CYCLES, *this}}
        , activityLang_{activityAllocator}
        , loadControl_{connectMonitoring()}
        , engineObserver_{engineObserver}
        { }
      
      
      bool
      empty()  const
        {
          return layer1_.empty();
        }
      
      void
      ignite()
        {
          TRACE (engine, "Ignite Scheduler Dispatch.");
          bool force_continued_run{true};
          handleDutyCycle (RealClock::now(), force_continued_run);
          if (not empty())
            workForce_.activate();
        }
      
      
      void
      terminateProcessing()
        {
          TRACE (engine, "Forcibly terminate Scheduler Dispatch.");
          workForce_.awaitShutdown();
          layer1_.discardSchedule();
        }
      
      
      double
      getLoadIndicator()
        {
          return loadControl_.effectiveLoad();
        }
      
      
      void
      seedCalcStream (Job planningJob
                     ,ManifestationID manID = ManifestationID()
                     ,FrameRate expectedAdditionalLoad = FrameRate(25))
        {
          auto guard = layer2_.requireGroomingTokenHere();  // allow mutation
          layer1_.activate (manID);
          activityLang_.announceLoad (expectedAdditionalLoad);
          continueMetaJob (RealClock::now(), planningJob, manID);
        }
      
      
      void
      continueMetaJob (Time nextStart
                      ,Job planningJob
                      ,ManifestationID manID = ManifestationID())
        {
          bool isCompulsory = true;
          Time deadline = nextStart + DUTY_CYCLE_TOLERANCE;
          // place the meta-Job into the timeline...
          postChain ({activityLang_.buildMetaJob(planningJob, nextStart, deadline)
                                   .post()
                     , nextStart
                     , deadline
                     , manID
                     , isCompulsory});
        }
      
      
      ScheduleSpec
      defineSchedule (Job job)
        {
          return ScheduleSpec{*this, job};
        }
      
      
      
      activity::Proc doWork();
      
      
    private:
      void postChain (ActivationEvent);
      void sanityCheck (ActivationEvent const&);
      void handleDutyCycle (Time now, bool =false);
      void handleWorkerTermination (bool isFailure);
      void maybeScaleWorkForce (Time startHorizon);
      
      void triggerEmergency();
      
      
      LoadController::Wiring
      connectMonitoring()
        {
          LoadController::Wiring setup;
          setup.maxCapacity = []{ return work::Config::COMPUTATION_CAPACITY; };
          setup.currWorkForceSize = [this]{ return workForce_.size(); };
          setup.stepUpWorkForce   = [this](uint steps){ workForce_.incScale(steps); };
          return setup;
        }
      
      Time
      getSchedTime()
        {
          return RealClock::now();
        }
      
      class ExecutionCtx;
      friend class ExecutionCtx;
      
      friend class ScheduleSpec;
      
      friend class test::SchedulerService_test;
    };
  
  
  
  
  
  
  class WorkTiming
    : public EngineEvent
    {
      using Payload = EngineEvent::Payload<Time>;
      using EngineEvent::EngineEvent;
      
      static Symbol WORKSTART;
      static Symbol WORKSTOP;
      
    public:
      static WorkTiming start (Time now) { return WorkTiming{WORKSTART, Payload{now}}; }
      static WorkTiming stop  (Time now) { return WorkTiming{WORKSTOP,  Payload{now}}; }
    };
  
  class Scheduler::ExecutionCtx
    : util::NonCopyable
    {
      Scheduler& scheduler_;
    public:
      
      ActivationEvent rootEvent;
      
      ExecutionCtx(Scheduler& self, ActivationEvent toDispatch)
        : scheduler_{self}
        , rootEvent{toDispatch}
      { }
      
      
      /* ==== Implementation of the Concept ExecutionCtx ==== */
      
      activity::Proc
      post (Time when, Time dead, Activity* chain, ExecutionCtx& ctx)
        {
          REQUIRE (chain);
          ActivationEvent chainEvent = ctx.rootEvent;
          chainEvent.refineTo (chain, when, dead);
          scheduler_.sanityCheck (chainEvent);
          return scheduler_.layer2_.postChain (chainEvent, scheduler_.layer1_);
        }
      
      void
      work (Time now, size_t qualifier)
        {
          scheduler_.layer2_.dropGroomingToken();
          scheduler_.engineObserver_.dispatchEvent(qualifier, WorkTiming::start(now));
        }
      
      void
      done (Time now, size_t qualifier)
        {
          scheduler_.engineObserver_.dispatchEvent(qualifier, WorkTiming::stop(now));
        }
      
      activity::Proc
      tick (Time now)
        {
          scheduler_.handleDutyCycle (now);
          return activity::PASS;
        }
      
      Time
      getSchedTime()
        {
          return scheduler_.getSchedTime();
        }
    };
  
  
  
  
  /***********************************************************************/
  inline activity::Proc
  Scheduler::doWork()
  {
    return layer2_.dispatchCapacity (layer1_
                                    ,loadControl_
                                    ,[this](ActivationEvent toDispatch)
                                            {
                                              ExecutionCtx ctx{*this, toDispatch};
                                              return ActivityLang::dispatchChain (toDispatch, ctx);
                                            }
                                    ,[this] { return getSchedTime(); }
                                    );
  }
  
  
  
  
  
  inline ScheduleSpec
  ScheduleSpec::post()
  {  // execute term-builder on-demand...
    maybeBuildTerm();
    
     // set up new schedule by retrieving the Activity-chain...
    theScheduler_->postChain ({term_->post(), start_
                                            , death_
                                            , manID_
                                            , isCompulsory_});
    return move(*this);
  }
  
  inline void
  ScheduleSpec::maybeBuildTerm()
  {
    if (term_) return;
    term_ = move(
      theScheduler_->activityLang_
          .buildCalculationJob (job_, start_,death_));
  }
  
  inline ScheduleSpec
  ScheduleSpec::linkToSuccessor (ScheduleSpec& succSpec, bool unlimitedTime)
    {
      this->maybeBuildTerm();
      succSpec.maybeBuildTerm();
      term_->appendNotificationTo (*succSpec.term_, unlimitedTime);
      return move(*this);
    }
  
  inline ScheduleSpec
  ScheduleSpec::linkToPredecessor (ScheduleSpec& predSpec, bool unlimitedTime)
    {
      predSpec.maybeBuildTerm();
      this->maybeBuildTerm();
      predSpec.term_->appendNotificationTo (*term_, unlimitedTime);
      return move(*this);
    }
  
  
  
  inline void
  Scheduler::sanityCheck (ActivationEvent const& event)
    {
      if (not event)
        throw error::Logic ("Empty event passed into Scheduler entrance");
      if (event.startTime() == Time::ANYTIME)
        throw error::Fatal ("Attempt to schedule an Activity without valid start time");
      if (event.deathTime() == Time::NEVER)
        throw error::Fatal ("Attempt to schedule an Activity without valid deadline");
      Time now{getSchedTime()};
      Offset toDeadline{now, event.deathTime()};
      if (toDeadline > FUTURE_PLANNING_LIMIT)
        throw error::Fatal (util::_Fmt{"Attempt to schedule Activity %s "
                                       "with a deadline by %s into the future"}
                                      % *event.activity
                                      % toDeadline);
    }
  
  
  
  
  /*****************************************************************/
  inline void
  Scheduler::postChain (ActivationEvent actEvent)
  {
    sanityCheck (actEvent);
    maybeScaleWorkForce (actEvent.startTime());
    layer2_.postChain (actEvent, layer1_);
  }
  
  

  
  inline void
  Scheduler::handleDutyCycle (Time now, bool forceContinuation)
  {
    auto guard = layer2_.requireGroomingTokenHere();
    
    //  consolidate queue and discard outdated tasks
    if (not layer2_.maintainQueueHead(layer1_,now))
      {          // missed deadline of compulsory task
        triggerEmergency();
        return;//   leave everything as-is
      }
    
    // clean-up of obsolete Activities
    activityLang_.discardBefore (now);
    
    loadControl_.updateState (now);
    
    if (not empty() or forceContinuation)
      {// prepare next duty cycle »tick«
        Time nextTick = now + (forceContinuation? WORK_HORIZON : DUTY_CYCLE_PERIOD);
        Time deadline = nextTick + DUTY_CYCLE_TOLERANCE;
        Activity& tickActivity = activityLang_.createTick (deadline);
        ActivationEvent tickEvent{tickActivity, nextTick, deadline, ManifestationID(), true};
        layer2_.postChain (tickEvent, layer1_);
      } // *deliberately* use low-level entrance
  }    //  to avoid ignite() cycles and derailed load-regulation
  
  inline void
  Scheduler::handleWorkerTermination (bool isFailure)
  {
    if (isFailure)
      triggerEmergency();
    else
      loadControl_.markWorkerExit();
  }
  
  inline void
  Scheduler::maybeScaleWorkForce (Time startHorizon)
  {
    if (empty())
      ignite();
    else
      loadControl_.ensureCapacity (startHorizon);
  }
  
  inline void
  Scheduler::triggerEmergency()
  {
    UNIMPLEMENTED ("scheduler overrun -- trigger Emergency");
  }
  
  
  
}} // namespace vault::gear
#endif /*SRC_VAULT_GEAR_SCHEDULER_H_*/