steam-dispatcher.cpp

Go to the documentation of this file.

/*
  SteamDispatcher  -  Steam-Layer command dispatch and execution

  Copyright (C)         Lumiera.org
    2008,               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.

* *****************************************************/


#include "lib/error.hpp"
#include "include/logging.h"
#include "steam/control/steam-dispatcher.hpp"
#include "steam/control/command-dispatch.hpp"
#include "steam/control/command-queue.hpp"
#include "steam/control/looper.hpp"
#include "steam/control/session-command-service.hpp"
#include "steam/mobject/session.hpp"
#include "lib/depend-inject.hpp"
#include "lib/sync-barrier.hpp"
#include "lib/thread.hpp"
#include "lib/util.hpp"                

#include <utility>
#include <memory>
  
using lib::Sync;
using lib::SyncBarrier;
using lib::ThreadHookable;
using lib::RecursiveLock_Waitable;
using std::make_unique;
using std::move;

namespace steam {
namespace control {
  
  namespace error = lumiera::error;
  
  
  /********************************************************************/
  class DispatcherLoop
    : public CommandDispatch
    , public Sync<RecursiveLock_Waitable>
    {
      using ServiceHandle = lib::DependInject<SessionCommandService>::ServiceInstance<>;
      using Launch = lib::ThreadHookable::Launch;
      
      ServiceHandle commandService_;

      SyncBarrier      init_;
      CommandQueue    queue_;
      string          error_;
      Looper         looper_;
      ThreadHookable thread_;
      
    public:
      template<typename FUN>
      DispatcherLoop (FUN&& atExit)
        : commandService_{ServiceHandle::NOT_YET_STARTED}
        , queue_{}
        , error_{}
        , looper_([&]() -> bool
                    {
                      return not queue_.empty();
                    })
        , thread_{ //----the-Session-Thread---------------
                  Launch{&DispatcherLoop::runSessionThread, this}
                        .threadID("Session")
                        .atExit([this, signalEndOfThread = move(atExit)]
                                (lib::thread::ThreadWrapper& handle)
                                  {
                                    handle.detach_thread_from_wrapper();
                                    signalEndOfThread (&error_);
                                  })}
        {
          init_.sync();   // done with setup; loop may run now....
          INFO (session, "Steam-Dispatcher running...");
            {
              Lock sync{this}; // open public session interface:
              commandService_.createInstance(*this);
            }
        }
      
     ~DispatcherLoop()
        {
          try {
              commandService_.shutdown();  // redundant call, to ensure session interface is closed reliably
              INFO (session, "Steam-Dispatcher stopped.");
              if (thread_)
                ALERT (session, "Dispatcher destroyed while Session thread is running. The rest is silence.");
            }
          ERROR_LOG_AND_IGNORE(session, "Stopping the Steam-Dispatcher");
        }
      
      void
      activateCommandProecssing()
        {
          Lock sync{this};
          looper_.enableProcessing(true);
          INFO (command, "Session command processing activated.");
          sync.notify_all();
        }
      
      void
      deactivateCommandProecssing()
        {
          Lock sync{this};
          looper_.enableProcessing(false);
          INFO (command, "Session command interface closed.");
          sync.notify_all();
        }
      
      void
      requestStop()  noexcept
        {
          Lock sync{this};
          commandService_.shutdown(); // closes Session interface
          looper_.triggerShutdown();
          sync.notify_all();
        }
      
      void
      awaitStateProcessed()  const
        {
          Lock{this, [&]{ return isStateSynched(); }};
           //  wake-up typically by updateState()
        }
      
      size_t
      size()  const
        {
          Lock sync{this};
          return queue_.size();
        }
      
      
      /* === CommandDispatch interface === */
      
      void
      enqueue (Command&& cmd)  override
        {
          Lock sync{this};
          queue_.feed (move (cmd));
          sync.notify_all();
        }
      
      void
      clear()  override
        {
          Lock sync{this};
          queue_.clear();
          sync.notify_all();
        }
      
    private:
      void
      runSessionThread ()
        {
          init_.sync();
          try
            {
              while (looper_.shallLoop())
                {
                  awaitAction();
                  if (looper_.isDying())
                    break;
                  if (looper_.runBuild())
                    startBuilder();
                  else
                    if (looper_.isWorking())
                      processCommands();
                  updateState();
                }
            }
          catch (std::exception& problem)
            {  // could also be lumiera::Error
              error_ = problem.what();
              lumiera_error();        // clear error flag
            }
          catch (...)
            {
              error_ = string{lumiera_error()};
            }
           // Session thread terminates...
        } //  atExit-λ will invoke ~DispatcherLoop()
      
      void
      awaitAction()   
        {
          Lock{this}.wait_for (looper_.getTimeout()
                              ,[&]{ return looper_.requireAction(); });
        }
      
      void
      updateState()   
        {
          looper_.markStateProcessed();
          if (looper_.isDisabled())     // otherwise wake-up would not be safe
            getMonitor(this).notify_all();
        }
      
      bool
      isStateSynched()  const
        {
          if (thread_.invokedWithinThread())
            throw error::Fatal("Possible Deadlock. "
                               "Attempt to synchronise to a command processing check point "
                               "from within the (single) session thread."
                              , error::LUMIERA_ERROR_LIFECYCLE);
          return not looper_.hasPendingChanges();
        }
      
      void
      processCommands()
        {
          Command cmd;
            {
              Lock sync{this};
              if (not queue_.empty())
                cmd = queue_.pop();
            }
          if (cmd)
            {
              INFO (command, "+++ dispatch %s", cStr(cmd));          
              
              if (util::startsWith (string(cmd.getID()), "test"))
                {
                  INFO (command, "+++ -------->>> bang!");
                  auto resultState = cmd();
                  resultState.maybeThrow();
                }
            }
        }
      
      void
      startBuilder()
        {
          INFO (builder, "+++ start the Steam-Builder...");
        }
    };
  
  
  
  
  
  
  lib::Depend<SteamDispatcher> SteamDispatcher::instance;
  
  
  /* ======== SteamDispatcher implementation ======== */
  
  SteamDispatcher::SteamDispatcher()  { }
  SteamDispatcher::~SteamDispatcher() { }
  
  bool
  SteamDispatcher::start (Subsys::SigTerm termNotification)
  {
    Lock sync{this};
    if (runningLoop_) return false;
    
    runningLoop_ =
         make_unique<DispatcherLoop>(
            [=](string* problemIndicator)
                { // when the Session thread ends....
                  SteamDispatcher::endRunningLoopState();
                  termNotification (problemIndicator);
                });
    
    if (active_)
      runningLoop_->activateCommandProecssing();
    return true;
  }
  
  void
  SteamDispatcher::endRunningLoopState()
  {
    Lock sync{this};
    if (runningLoop_)
      runningLoop_.reset();  // delete DispatcherLoop object
    else
      WARN (command, "clean-up of DispatcherLoop invoked, "
                     "while SteamDispatcher is not marked as 'running'. "
                     "Likely an error in lifecycle logic, as the only one "
                     "intended to delete this object is the loop thread itself.");
  }
  
  
  bool
  SteamDispatcher::isRunning()
  {
    Lock sync{this};
    return bool(runningLoop_);
  }
  
  
  void
  SteamDispatcher::requestStop()  noexcept
  {
    try {
      Lock sync{this};
      if (runningLoop_)
        runningLoop_->requestStop();
    }
    ERROR_LOG_AND_IGNORE (command, "Request for Session Loop Thread to terminate");
  }
  
  
  
  void
  SteamDispatcher::activate()
  {
    Lock sync{this};
    active_ = true;
    if (runningLoop_)
      runningLoop_->activateCommandProecssing();
  }
  
  
  void
  SteamDispatcher::deactivate()
  {
    Lock sync{this};
    active_ = false;
    if (runningLoop_)
      runningLoop_->deactivateCommandProecssing();
  }
  
  
  void
  SteamDispatcher::awaitDeactivation()
  {
    Lock sync{this};
    if (runningLoop_)
      runningLoop_->awaitStateProcessed();
  }
  
  
  void
  SteamDispatcher::clear()
  {
    Lock sync{this};
    if (not empty())
      {
        WARN (command, "DISCARDING pending Session commands.");
        REQUIRE (runningLoop_);
        runningLoop_->clear();
      }
  }
  
  
  bool
  SteamDispatcher::empty()  const
  {
    Lock sync{this};
    return not runningLoop_
        or 0 == runningLoop_->size();
  }
  
  
  
  
}} // namespace steam::control