stress-test-rig.hpp

Go to the documentation of this file.

/*
  STRESS-TEST-RIG.hpp  -  setup for stress and performance investigation

  Copyright (C)         Lumiera.org
    2024,               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 VAULT_GEAR_TEST_STRESS_TEST_RIG_H
#define VAULT_GEAR_TEST_STRESS_TEST_RIG_H


#include "test-chain-load.hpp"
#include "lib/binary-search.hpp"
#include "lib/test/transiently.hpp"

#include "vault/gear/scheduler.hpp"
#include "lib/time/timevalue.hpp"
#include "lib/meta/function.hpp"
#include "lib/format-string.hpp"
#include "lib/format-cout.hpp"
#include "lib/gnuplot-gen.hpp"
#include "lib/stat/statistic.hpp"
#include "lib/stat/data.hpp"
#include "lib/util.hpp"

#include <algorithm>
#include <utility>
#include <vector>
#include <tuple>
#include <array>


namespace vault{
namespace gear {
namespace test {
  
  using std::make_tuple;
  using std::forward;
  
  
  template<size_t maxFan =DEFAULT_FAN>
  class StressTestRig
    : util::NonCopyable
    {
    public:
      using TestLoad  = TestChainLoad<maxFan>;
      using TestSetup = typename TestLoad::ScheduleCtx;
      
      
      /***********************************************************************/
      template<class CONF>
      static auto
      with()
        {
          return Launcher<CONF>{};
        }
      
      
      /* ======= default configuration (inherited) ======= */
      
      uint CONCURRENCY = work::Config::getDefaultComputationCapacity();
      bool INSTRUMENTATION = true;
      double EPSILON      = 0.01;          
      double UPPER_STRESS = 1.7;           
      double FAIL_LIMIT   = 2.0;           
      double TRIGGER_FAIL = 0.55;          
      double TRIGGER_SDEV = FAIL_LIMIT;    
      double TRIGGER_DELTA = 2*FAIL_LIMIT; 
      bool showRuns = false;    
      bool showStep = true;     
      bool showRes  = true;     
      bool showRef  = true;     
      
      static uint constexpr REPETITIONS{20};

      BlockFlowAlloc bFlow{};
      EngineObserver watch{};
      Scheduler scheduler{bFlow, watch};
      
      
      
    protected:
      auto
      testLoad(size_t nodes =64)
        {
          return TestLoad{nodes};
        }
      
      auto
      testSetup (TestLoad& testLoad)
        {
          return testLoad.setupSchedule(scheduler)
                         .withLevelDuration(200us)
                         .withJobDeadline(500ms)
                         .withUpfrontPlanning();
        }
      
      template<class CONF>
      struct Launcher : CONF
        {
          template<template<class> class TOOL, typename...ARGS>
          auto
          perform (ARGS&& ...args)
            {
              return TOOL<CONF>{}.perform (std::forward<ARGS> (args)...);
            }
        };
    };
  
  
  
  
  namespace bench { 
    
    using util::_Fmt;
    using util::min;
    using util::max;
    using std::vector;
    using std::declval;
    
    
    /**************************************************/
    template<class CONF>
    class BreakingPoint
      : public CONF
      {
        using TestLoad  = typename CONF::TestLoad;
        using TestSetup = typename TestLoad::ScheduleCtx;
        
        struct Res
          {
            double stressFac{0};
            double percentOff{0};
            double stdDev{0};
            double avgDelta{0};
            double avgTime{0};
            double expTime{0};
          };
        
        void
        configureTest (TestSetup& testSetup, double stressFac)
          {
            testSetup.withInstrumentation(CONF::INSTRUMENTATION)          // side-effect: clear existing statistics
                     .withAdaptedSchedule(stressFac, CONF::CONCURRENCY, adjustmentFac);
          }
        
        Res
        runProbes (TestSetup& testSetup, double stressFac)
          {
            auto sqr = [](auto n){ return n*n; };
            Res res;
            auto& [sf,pf,sdev,avgD,avgT,expT] = res;
            sf   = stressFac;
            std::array<double, CONF::REPETITIONS> runTime;
            for (uint i=0; i<CONF::REPETITIONS; ++i)
              {
                runTime[i] = testSetup.launch_and_wait() / 1000;
                avgT += runTime[i];
                maybeAdaptScaleEmpirically (testSetup, stressFac);
              }
            expT = testSetup.getExpectedEndTime() / 1000;
            avgT /= CONF::REPETITIONS;
            avgD = (avgT-expT); // can be < 0
            for (uint i=0; i<CONF::REPETITIONS; ++i)
              {
                sdev += sqr (runTime[i] - avgT);
                double delta = (runTime[i] - expT);
                bool fail = (delta > CONF::FAIL_LIMIT);
                if (fail)
                  ++ pf;
                showRun(i, delta, runTime[i], runTime[i] > avgT, fail);
              }
            pf /= CONF::REPETITIONS;
            sdev = sqrt (sdev/CONF::REPETITIONS);
            showStep(res);
            return res;
          }
        
        bool
        decideBreakPoint (Res& res)
          {
            return res.percentOff > 0.99
               or( res.percentOff > CONF::TRIGGER_FAIL
               and res.stdDev     > CONF::TRIGGER_SDEV
               and res.avgDelta   > CONF::TRIGGER_DELTA);
          }
        
        template<class FUN>
        Res
        conductBinarySearch (FUN&& runTestCase, vector<Res> const& results)
          {
            double breakPoint = lib::binarySearch_upper (forward<FUN> (runTestCase)
                                                        , 0.0, CONF::UPPER_STRESS
                                                        , CONF::EPSILON);
            uint s = results.size();
            ENSURE (s >= 2);
            Res res;
            auto& [sf,pf,sdev,avgD,avgT,expT] = res;
            // average data over the last three steps investigated for smoothing
            uint points = min (results.size(), 3u);
            for (uint i=results.size()-points; i<results.size(); ++i)
              {
                Res const& resx = results[i];
                pf   += resx.percentOff;
                sdev += resx.stdDev;
                avgD += resx.avgDelta;
                avgT += resx.avgTime;
                expT += resx.expTime;
              }
            pf   /= points;
            sdev /= points;
            avgD /= points;
            avgT /= points;
            expT /= points;
            sf = breakPoint;
            return res;
          }
        
        double adjustmentFac{1.0};
        size_t gaugeProbes = 3 * CONF::REPETITIONS;
        
        void
        maybeAdaptScaleEmpirically (TestSetup& testSetup, double stressFac)
          {
            if (not gaugeProbes) return;
            double gain = util::limited (0, pow(stressFac, 9), 1);
            if (gain < 0.2) return;
            //
            double formFac = testSetup.determineEmpiricFormFactor (CONF::CONCURRENCY);
            adjustmentFac = gain*formFac + (1-gain)*adjustmentFac;
            testSetup.withAdaptedSchedule(stressFac, CONF::CONCURRENCY, adjustmentFac);
            --gaugeProbes;
          }
        
        
        _Fmt fmtRun_ {"....·%-2d:  Δ=%4.1f        t=%4.1f  %s %s"};                          //      i % Δ  % t % t>avg?  % fail?
        _Fmt fmtStep_{ "%4.2f|  : ∅Δ=%4.1f±%-4.2f  ∅t=%4.1f  %s %%%-3.0f -- expect:%4.1fms"};// stress % ∅Δ % σ % ∅t % fail % pecentOff % t-expect
        _Fmt fmtResSDv_{"%9s= %5.2f ±%4.2f%s"};
        _Fmt fmtResVal_{"%9s: %5.2f%s"};
        
        void
        showRun(uint i, double delta, double t, bool over, bool fail)
          {
            if (CONF::showRuns)
              cout << fmtRun_ % i % delta % t % (over? "+":"-") % (fail? "●":"○")
                   << endl;
          }
        
        void
        showStep(Res& res)
          {
            if (CONF::showStep)
              cout << fmtStep_ % res.stressFac % res.avgDelta % res.stdDev % res.avgTime
                               % (decideBreakPoint(res)? "—◆—":"—◇—")
                               % (100*res.percentOff) % res.expTime
                   << endl;
          }
        
        void
        showRes(Res& res)
          {
            if (CONF::showRes)
              {
                cout << fmtResVal_ % "stresFac" % res.stressFac             % ""  <<endl;
                cout << fmtResVal_ %     "fail" %(res.percentOff * 100)     % '%' <<endl;
                cout << fmtResSDv_ %    "delta" % res.avgDelta % res.stdDev % "ms"<<endl;
                cout << fmtResVal_ %  "runTime" % res.avgTime               % "ms"<<endl;
                cout << fmtResVal_ % "expected" % res.expTime               % "ms"<<endl;
              }
          }
        
        void
        showRef(TestSetup& testSetup)
          {
            if (CONF::showRef)
              cout << fmtResVal_ % "refTime"
                                 % (testSetup.calcRuntimeReference() /1000)
                                 % "ms" << endl;
          }
        
        
      public:
        auto
        perform()
          {
            TRANSIENTLY(work::Config::COMPUTATION_CAPACITY) = CONF::CONCURRENCY;
            
            TestLoad testLoad = CONF::testLoad().buildTopology();
            TestSetup testSetup = CONF::testSetup (testLoad);
            
            vector<Res> observations;
            auto performEvaluation = [&](double stressFac)
                                        {
                                          configureTest (testSetup, stressFac);
                                          auto res = runProbes (testSetup, stressFac);
                                          observations.push_back (res);
                                          return decideBreakPoint(res);
                                        };
            
            Res res = conductBinarySearch (move(performEvaluation), observations);
            showRes (res);
            showRef (testSetup);
            return make_tuple (res.stressFac, res.avgDelta, res.avgTime);
          }
      };
    
    
    
    
    
    /**************************************************/
    template<class CONF>
    class ParameterRange
      : public CONF
      {
        using TestLoad  = typename CONF::TestLoad;
        using TestSetup = typename TestLoad::ScheduleCtx;
        
        // Type binding for data evaluation
        using Param = typename CONF::Param;
        using Table = typename CONF::Table;
        
        
        void
        runTest (Param param, Table& data)
          {
            TestLoad testLoad = CONF::testLoad(param).buildTopology();
            TestSetup testSetup = CONF::testSetup (testLoad)
                                       .withInstrumentation();    // Note: by default Schedule with CONF::LEVEL_STEP
            double millis = testSetup.launch_and_wait() / 1000;
            auto stat = testSetup.getInvocationStatistic();
            CONF::collectResult (data, param, millis, stat);
          }
        
      public:
        Table
        perform (Param lower, Param upper)
          {
            TRANSIENTLY(work::Config::COMPUTATION_CAPACITY) = CONF::CONCURRENCY;
            
            Param dist = upper - lower;
            uint cnt = CONF::REPETITIONS;
            vector<Param> points;
            points.reserve (cnt);
            Param minP{upper}, maxP{lower};
            for (uint i=0; i<cnt; ++i)
              {
                auto random = double(rand())/RAND_MAX;
                Param pos = lower + Param(floor (random*dist + 0.5));
                points.push_back(pos);
                minP = min (pos, minP);
                maxP = max (pos, maxP);
              }
            // ensure the bounds participate in test
            if (maxP < upper) points[cnt-2] = upper;
            if (minP > lower) points[cnt-1] = lower;
            
            Table results;
            for (Param point : points)
              runTest (point, results);
            return results;
          }
      };
    
    
    
    /* ====== Preconfigured ParamRange-Evaluations ====== */
    
    using lib::stat::Column;
    using lib::stat::DataTable;
    using lib::stat::DataSpan;
    using lib::stat::CSVData;
    using IncidenceStat = lib::IncidenceCount::Statistic;
    
    template<typename F, typename G>
    inline auto
    linearRegression (Column<F> const& x, Column<G> const& y)
    {
      lib::stat::RegressionData points;
      size_t cnt = min (x.data.size(), y.data.size());
      points.reserve (cnt);
      for (size_t i=0; i < cnt; ++i)
        points.emplace_back (x.data[i], y.data[i]);
      return lib::stat::computeLinearRegression (points);
    }
    
    struct LoadPeak_ParamRange_Evaluation
      {
        using Param = size_t;
        
        struct DataRow
          {
            Column<Param>  param   {"load size"};    // independent variable / control parameter
            Column<double> time    {"result time"};
            Column<double> conc    {"concurrency"};
            Column<double> jobtime {"avg jobtime"};
            Column<double> impeded {"avg impeded"};
            
            auto allColumns()
            { return std::tie(param
                             ,time
                             ,conc
                             ,jobtime
                             ,impeded
                             );
            }
          };
        
        using Table = DataTable<DataRow>;
        
        void
        collectResult(Table& data, Param param, double millis, bench::IncidenceStat const& stat)
          {
            (void)millis;
            data.newRow();
            data.param = param;
            data.time  = stat.coveredTime / 1000;
            data.conc  = stat.avgConcurrency;
            data.jobtime = stat.activeTime / stat.activationCnt;
            data.impeded = (stat.timeAtConc(1) + stat.timeAtConc(0))/stat.activationCnt;
          }
        
        
        static double
        avgConcurrency (Table const& results)
          {
            return lib::stat::average (DataSpan<double> (results.conc.data));
          }
        
        static string
        renderGnuplot (Table const& results)
          {
            using namespace lib::gnuplot_gen;
            string csv = results.renderCSV();
            Param maxParam = * std::max_element (results.param.data.begin(), results.param.data.end());
            Param xtics = maxParam > 500? 50
                        : maxParam > 200? 20
                        : maxParam > 100? 10
                        :                  5;
            return scatterRegression(
                    ParamRecord().set (KEY_CSVData,  csv)
                                 .set (KEY_TermSize, "600,600")
                                 .set (KEY_Xtics,    int64_t(xtics))
                                 .set (KEY_Xlabel,  "load size ⟶ number of jobs")
                                 .set (KEY_Ylabel,  "active time ⟶ ms")
                                 .set (KEY_Y2label, "concurrent threads ⟶")
                                 .set (KEY_Y3label, "avg job time ⟶ µs")
                                );
          }
      };
    //
  }// namespace bench
}}}// namespace vault::gear::test
#endif /*VAULT_GEAR_TEST_STRESS_TEST_RIG_H*/