session-command-function-test.cpp

Go to the documentation of this file.

/*
  SessionCommandFunction(Test)  -  function test of command dispatch via SessionCommand facade

   Copyright (C)
     2017,            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.

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


#include "lib/test/run.hpp"
#include "lib/test/test-helper.hpp"
extern "C" {
#include "common/interfaceregistry.h"
}

#include "steam/control/steam-dispatcher.hpp"
#include "steam/control/command-def.hpp"
#include "include/session-command-facade.h"
#include "lib/typed-counter.hpp"
#include "lib/format-string.hpp"
#include "lib/sync-barrier.hpp"
#include "lib/thread.hpp"
#include "lib/symbol.hpp"
#include "lib/util.hpp"

#include <boost/lexical_cast.hpp>
#include <chrono>
#include <string>
#include <vector>
#include <deque>


namespace steam {
namespace control {
namespace test    {
  
  
  using boost::lexical_cast;
  using std::this_thread::sleep_for;
  using std::chrono::microseconds;
  using namespace std::chrono_literals;
  using steam::control::SessionCommand;
  using lib::test::randTime;
  using lib::diff::GenNode;
  using lib::diff::Rec;
  using lib::time::Time;
  using lib::time::TimeVar;
  using lib::time::Duration;
  using lib::time::Offset;
  using lib::time::FSecs;
  using lib::FamilyMember;
  using lib::SyncBarrier;
  using lib::Symbol;
  using util::_Fmt;
  using util::isnil;
  using std::string;
  using std::vector;
  using std::deque;
  using std::rand;
  
  
  namespace { // test fixture...
    
    /* === parameters for multi-threaded stress test === */
    
    uint NUM_THREADS_DEFAULT = 50;   
    uint NUM_INVOC_PER_THRED = 10;
    uint MAX_RAND_DELAY_us   = 50;   
    
    void
    maybeOverride (uint& configSetting, Arg cmdline, uint paramNr)
    {
      if (paramNr < cmdline.size())
        configSetting = lexical_cast<uint>(cmdline[paramNr]);
    }
    
    
    /* === mock operation to be dispatched as command === */
    
    const Symbol COMMAND_ID{"test.dispatch.function.command"};
    const Symbol COMMAND_I1{"test.dispatch.function.command.instance-1"};
    const Symbol COMMAND_I2{"test.dispatch.function.command.instance-2"};
    
    TimeVar testCommandState = randTime();
    
    void
    operate (Duration dur, Offset offset, int factor)
    {
      testCommandState += Offset(dur) + offset*factor;
    }
    
    Time
    capture (Duration, Offset, int)
    {
      return testCommandState;
    }
    
    void
    undoIt (Duration, Offset, int, Time oldState)
    {
      testCommandState = oldState;
    }
    
    
  }//(End) test fixture
  
  
#define __DELAY__ sleep_for (20ms);
  
  
  
  
  
  /******************************************************************************************/
  class SessionCommandFunction_test : public Test
    {
      
      //------------------FIXTURE
    public:
      SessionCommandFunction_test()
        {
          CommandDef (COMMAND_ID)
               .operation (operate)
               .captureUndo (capture)
               .undoOperation (undoIt)
               ;
          Command(COMMAND_ID).storeDef(COMMAND_I1);
          Command(COMMAND_ID).storeDef(COMMAND_I2);
        }
     ~SessionCommandFunction_test()
        {
          Command::remove (COMMAND_ID);
          Command::remove (COMMAND_I1);
          Command::remove (COMMAND_I2);
        }
      //-------------(End)FIXTURE
      
      
      virtual void
      run (Arg args_for_stresstest)
        {
          lumiera_interfaceregistry_init();
          lumiera::throwOnError();
          
          startDispatcher();
          perform_simpleInvocation();
          perform_messageInvocation();
          perform_massivelyParallel(args_for_stresstest);
          stopDispatcher();
          
          lumiera_interfaceregistry_destroy();
        }
      
      
      void
      startDispatcher()
        {
          CHECK (not SteamDispatcher::instance().isRunning());
          
          SteamDispatcher::instance().start ([&] (string* problemMessage)
                                                 {
                                                   CHECK (isnil (*problemMessage));
                                                   thread_has_ended = true;
                                                 });
          
          CHECK (SteamDispatcher::instance().isRunning());
          CHECK (not thread_has_ended);
        }
      bool thread_has_ended{false};
      
      
      void
      stopDispatcher()
        {
          CHECK (SteamDispatcher::instance().isRunning());
          SteamDispatcher::instance().requestStop();
          
          __DELAY__
          CHECK (not SteamDispatcher::instance().isRunning());
          CHECK (thread_has_ended);
        }
      
      
      void
      perform_simpleInvocation()
        {
          string cmdID {COMMAND_I1};
          Rec arguments {Duration(15,10), Time(500,0), -1};
          
          CHECK (not Command(COMMAND_I1).canExec());
          SessionCommand::facade().bindArg (cmdID, arguments);
          CHECK (Command(COMMAND_I1).canExec());
          
          
          Time prevState = testCommandState;
          SessionCommand::facade().invoke(cmdID);
          
          __DELAY__
          CHECK (testCommandState - prevState == Time(0, 1));           // execution added 1500ms -1*500ms == 1sec
        }
      
      
      
      void
      perform_messageInvocation()
        {
          // this happens within some tangible UI element (widget / controller)
          GenNode commandMsg{string(COMMAND_I2), Rec{Duration(25,10), Time(500,0), -2}};
          CHECK (commandMsg.idi.getSym() == string{COMMAND_I2});
          CHECK (not Command::canExec(COMMAND_I2));
          Time prevState = testCommandState;
          
          // this happens, when CoreService receives command messages from UI-Bus
          SessionCommand::facade().trigger (commandMsg.idi.getSym(), commandMsg.data.get<Rec>());
          
          __DELAY__
          CHECK (testCommandState - prevState == Time(FSecs(3,2)));     // execution added 2500ms -2*500ms == 1.5sec
        }
      
      
      
      void
      perform_massivelyParallel (Arg args_for_stresstest)
        {
          seedRand();
          maybeOverride (NUM_THREADS_DEFAULT, args_for_stresstest, 1);
          maybeOverride (NUM_INVOC_PER_THRED, args_for_stresstest, 2);
          maybeOverride (MAX_RAND_DELAY_us,   args_for_stresstest, 3);
          
          
          // we'll run several instances of the following thread....
          class InvocationProducer
            : util::NonCopyable
            {
              SyncBarrier& barrier_;
              FamilyMember<InvocationProducer> id_;
              vector<string> cmdIDs_;
              lib::Random random_;
              
              lib::ThreadJoinable<void> thread_;
              
              Symbol
              cmdID (uint j)
                {
                  cmdIDs_.push_back (_Fmt("%s.thread-%02d.%d") % COMMAND_ID % id_ % j);
                  return cStr(cmdIDs_.back());
                }
              
              
            public:
              InvocationProducer (SyncBarrier& trigger)
                : barrier_{trigger}
                , random_{defaultGen}
                , thread_{"producer", [&]{ fabricateCommands(); }}
                { }
              
             ~InvocationProducer()
                {
                  thread_.join().maybeThrow();
                  for (auto& id : cmdIDs_)
                    Command::remove (cStr(id));
                }
             
            private:
              void
              fabricateCommands()
                {
                  barrier_.sync(); // barrier to unleash all threads together
                  
                  for (uint j=0; j<NUM_INVOC_PER_THRED; ++j)
                    {
                      auto cmd = Command(COMMAND_ID).storeDef(cmdID(j));
                      
                      __randomDelay();
                      sendCommandMessage (GenNode{string{cmd.getID()}, Rec{Duration(7*id_, 2), Time(500,0), -int(j)}});
                    }
                }
              
              static void
              sendCommandMessage (GenNode msg)
                {
                  SessionCommand::facade().trigger (msg.idi.getSym(), msg.data.get<Rec>());
                }
              
              void
              __randomDelay()
                {
                  if (not MAX_RAND_DELAY_us) return;
                  sleep_for (microseconds (1 + random_.i(MAX_RAND_DELAY_us)));   // random delay varying in steps of 1µs
                }
            };
          
          /* == controlling code in main thread == */
          Time prevState = testCommandState;
          
          FSecs expectedOffset{0};
          for (uint i=0; i<NUM_THREADS_DEFAULT; ++i)
            for (uint j=0; j<NUM_INVOC_PER_THRED; ++j)
              expectedOffset += FSecs(i*7,2) - FSecs(j,2);
          
          // fire up several threads to issue commands in parallel...
          SyncBarrier trigger{NUM_THREADS_DEFAULT + 1};
          deque<InvocationProducer> producerThreads;
          for (uint i=0; i<NUM_THREADS_DEFAULT; ++i)
            producerThreads.emplace_back (trigger);
          
          // start concurrent execution
          trigger.sync();
          
          // give the producer threads some head start...
          sleep_for (microseconds (MAX_RAND_DELAY_us * NUM_INVOC_PER_THRED / 2));
          __DELAY__
          
          // stop the dispatching to cause the queue to build up...
          SteamDispatcher::instance().deactivate();
          SteamDispatcher::instance().awaitDeactivation();
          
          __DELAY__
          SteamDispatcher::instance().activate();
          
          __DELAY__
          while (not SteamDispatcher::instance().empty());
          
          __DELAY__
          CHECK (testCommandState - prevState == Time(expectedOffset));
        }// Note: leaving this scope blocks for joining all producer threads
    };
  
  
  LAUNCHER (SessionCommandFunction_test, "function controller");
  
  
}}} // namespace steam::control::test