scheduler-commutator.hpp

Go to the documentation of this file.

/*
  SCHEDULER-COMMUTATOR.hpp  -  coordination layer of the render engine scheduler

   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_COMMUTATOR_H_
#define SRC_VAULT_GEAR_SCHEDULER_COMMUTATOR_H_


#include "vault/common.hpp"
#include "vault/gear/activity.hpp"
#include "vault/gear/scheduler-invocation.hpp"
#include "vault/gear/load-controller.hpp"
#include "vault/gear/activity-lang.hpp"
#include "lib/time/timevalue.hpp"
#include "lib/format-string.hpp"
#include "lib/nocopy.hpp"

#include <thread>
#include <atomic>


namespace vault{
namespace gear {
  
  using lib::time::Offset;
  using lib::time::FSecs;
  using lib::time::Time;
  using std::atomic;
  using std::memory_order::memory_order_relaxed;
  using std::memory_order::memory_order_acquire;
  using std::memory_order::memory_order_release;
  using std::chrono_literals::operator ""us;
  using std::chrono::microseconds;
  
  namespace { // Configuration / Scheduling limit
    
    microseconds GROOMING_WAIT_CYCLE{70us};   
    
    auto inline thisThread() { return std::this_thread::get_id(); }
  }
  
  
  
  /*************************************************************/
  class SchedulerCommutator
    : util::NonCopyable
    {
      using ThreadID = std::thread::id;
      atomic<ThreadID> groomingToken_{};
      
      
    public:
      SchedulerCommutator()  = default;
      
      bool
      acquireGoomingToken()  noexcept
        {
          ThreadID expect_noThread;                   // expect no one else to be in...
          return groomingToken_.compare_exchange_strong (expect_noThread, thisThread()
                                                        ,memory_order_acquire // success also constitutes an acquire barrier
                                                        ,memory_order_relaxed // failure has no synchronisation ramifications
                                                        );
        }
      
      void
      dropGroomingToken()  noexcept
        {          // expect that this thread actually holds the Grooming-Token
          REQUIRE (groomingToken_.load(memory_order_relaxed) == thisThread());
          const ThreadID noThreadHoldsIt;
          groomingToken_.store (noThreadHoldsIt, memory_order_release);
        }
      
      bool
      holdsGroomingToken (ThreadID id)  noexcept
        {
          return id == groomingToken_.load (memory_order_relaxed);
        }
      
      
      class ScopedGroomingGuard;
      ScopedGroomingGuard requireGroomingTokenHere();
      
      
      
      void
      maybeFeed (SchedulerInvocation& layer1)
        {
          if (layer1.hasPendingInput()
              and (holdsGroomingToken(thisThread())
                   or acquireGoomingToken()))
            layer1.feedPrioritisation();
        }
      
      bool
      maintainQueueHead (SchedulerInvocation& layer1, Time now)
        {
          ENSURE (holdsGroomingToken (thisThread()));
          layer1.feedPrioritisation();
          while (layer1.isOutdated (now) and not layer1.isOutOfTime(now))
            layer1.pullHead();
          return not layer1.isOutOfTime(now);
        }
      
      ActivationEvent
      findWork (SchedulerInvocation& layer1, Time now)
        {
          if (holdsGroomingToken (thisThread())
              or acquireGoomingToken())
            {
              layer1.feedPrioritisation();
              while (layer1.isOutdated (now) and not layer1.isOutOfTime(now))
                layer1.pullHead();
              if (not maintainQueueHead (layer1,now))
                ALERT (engine, "MISSED compulsory job -- should raise Scheduler-Emergency");   
              else
              if (layer1.isDue (now))
                return layer1.pullHead();
            }
          return ActivationEvent();
        }
      
      
      
      /***********************************************************/
      activity::Proc
      postChain (ActivationEvent event, SchedulerInvocation& layer1)
        {
          if (holdsGroomingToken (thisThread()))
            layer1.feedPrioritisation (move (event));
          else
            layer1.instruct (move (event));
          return activity::PASS;
        }

      
      template<class DISPATCH, class CLOCK>
      activity::Proc
      dispatchCapacity (SchedulerInvocation&, LoadController&, DISPATCH, CLOCK);
      
      
      
    private:
      activity::Proc
      scatteredDelay (Time now, Time head
                     ,LoadController& loadController
                     ,LoadController::Capacity capacity);
      
      void
      ensureDroppedGroomingToken()
        {
          if (holdsGroomingToken (thisThread()))
            dropGroomingToken();
        }
      
      struct WorkerInstruction
        {
          activity::Proc lastResult = activity::PASS;
          
          operator activity::Proc()
            {
              return activity::SKIP == lastResult? activity::PASS
                                                 : lastResult;
            }
          
          template<class FUN>
          WorkerInstruction
          performStep (FUN step)
            {
              if (activity::PASS == lastResult)
                lastResult = step();
              return move(*this);
            }
        };
    };
  
  
  
  
  
  
  template<class DISPATCH, class CLOCK>
  inline activity::Proc
  SchedulerCommutator::dispatchCapacity (SchedulerInvocation& layer1
                                        ,LoadController& loadController
                                        ,DISPATCH executeActivity
                                        ,CLOCK getSchedTime
                                        )
  {
    try {
        auto res = WorkerInstruction{}
                      .performStep([&]{
                                        maybeFeed(layer1);
                                        Time now = getSchedTime();
                                        Time head = layer1.headTime();
                                        return scatteredDelay(now, head, loadController,
                                                  loadController.markIncomingCapacity (head,now));
                                      })
                      .performStep([&]{
                                        Time now = getSchedTime();
                                        auto toDispatch = findWork (layer1,now);
                                        if (not toDispatch) return activity::KICK; // contention
                                        return executeActivity (toDispatch);
                                      })
                      .performStep([&]{
                                        maybeFeed(layer1);
                                        Time now = getSchedTime();
                                        Time head = layer1.headTime();
                                        return scatteredDelay(now, head, loadController,
                                                  loadController.markOutgoingCapacity (head,now));
                                      });
        
        // ensure lock clean-up
        if (res != activity::PASS)
          ensureDroppedGroomingToken();
        return res;
      }
    catch(...)
      {
        ensureDroppedGroomingToken();
        throw;
      }
  }
  
  
  inline activity::Proc
  SchedulerCommutator::scatteredDelay (Time now, Time head
                                      ,LoadController& loadController
                                      ,LoadController::Capacity capacity)
  {
    auto doTargetedSleep = [&]
          { // ensure not to block the Scheduler after management work
            ensureDroppedGroomingToken();
            // relocate this thread(capacity) to a time where its more useful
            Offset targetedDelay = loadController.scatteredDelayTime (now, capacity);
            std::this_thread::sleep_for (std::chrono::microseconds (_raw(targetedDelay)));
          };
    auto doTendNextHead = [&]
          {
            if (not loadController.tendedNext(head)
                and (holdsGroomingToken(thisThread())
                    or acquireGoomingToken()))
              loadController.tendNext(head);
          };
    
    switch (capacity) {
      case LoadController::DISPATCH:
        return activity::PASS;
      case LoadController::SPINTIME:
        std::this_thread::yield();
        return activity::SKIP;     //  prompts to abort chain but call again immediately
      case LoadController::IDLEWAIT:
        return activity::WAIT;     //  prompts to switch this thread into sleep mode
      case LoadController::TENDNEXT:
        doTendNextHead();
        doTargetedSleep();         //  let this thread wait until next head time is due
        return activity::SKIP;
      default:
        doTargetedSleep();
        return activity::SKIP;     //  prompts to abort this processing-chain for good
      }
  }
  
  
  
  
  
  class SchedulerCommutator::ScopedGroomingGuard
    : util::MoveOnly
    {
      SchedulerCommutator& commutator_;
      bool handledActively_;
      
      bool
      ensureHoldsToken()
        {
          if (commutator_.holdsGroomingToken(thisThread()))
            return false;
          while (not commutator_.acquireGoomingToken())
            std::this_thread::sleep_for (GROOMING_WAIT_CYCLE);
          return true;
        }
      
    public:
      ScopedGroomingGuard (SchedulerCommutator& layer2)
        : commutator_(layer2)
        , handledActively_{ensureHoldsToken()}
        { }
      
     ~ScopedGroomingGuard()
        {
          if (handledActively_ and
              commutator_.holdsGroomingToken(thisThread()))
            commutator_.dropGroomingToken();
        }
    };
  
  
  inline SchedulerCommutator::ScopedGroomingGuard
  SchedulerCommutator::requireGroomingTokenHere()
    {
      return ScopedGroomingGuard(*this);
    }

  
  
}} // namespace vault::gear
#endif /*SRC_VAULT_GEAR_SCHEDULER_COMMUTATOR_H_*/