518 lines
15 KiB
C++
518 lines
15 KiB
C++
/*
|
||
WorkForce(Test) - verify worker thread service
|
||
|
||
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.
|
||
|
||
* *****************************************************************/
|
||
|
||
/** @file work-force-test.cpp
|
||
** unit test \ref WorkForce_test
|
||
*/
|
||
|
||
|
||
#include "lib/test/run.hpp"
|
||
#include "vault/gear/work-force.hpp"
|
||
#include "lib/thread.hpp"
|
||
#include "lib/sync.hpp"
|
||
|
||
#include <functional>
|
||
#include <thread>
|
||
#include <chrono>
|
||
#include <set>
|
||
|
||
using test::Test;
|
||
|
||
|
||
namespace vault{
|
||
namespace gear {
|
||
namespace test {
|
||
|
||
using std::this_thread::sleep_for;
|
||
using namespace std::chrono_literals;
|
||
using std::chrono::milliseconds;
|
||
using lib::Thread;
|
||
|
||
|
||
namespace {
|
||
using WorkFun = std::function<work::SIG_WorkFun>;
|
||
using FinalFun = std::function<work::SIG_FinalHook>;
|
||
|
||
/**
|
||
* Helper: setup a Worker-Pool configuration for the test.
|
||
* Derived from the default configuration, it allows to bind
|
||
* a lambda as work-functor and to tweak other parameters.
|
||
*/
|
||
template<class FUN>
|
||
auto
|
||
setup (FUN&& workFun)
|
||
{
|
||
struct Setup
|
||
: work::Config
|
||
{
|
||
WorkFun doWork;
|
||
FinalFun finalHook = [](bool){ /*NOP*/ };
|
||
|
||
milliseconds IDLE_WAIT = work::Config::IDLE_WAIT;
|
||
size_t DISMISS_CYCLES = work::Config::DISMISS_CYCLES;
|
||
|
||
Setup (FUN&& workFun)
|
||
: doWork{std::forward<FUN> (workFun)}
|
||
{ }
|
||
|
||
Setup&&
|
||
withFinalHook (FinalFun finalFun)
|
||
{
|
||
finalHook = move (finalFun);
|
||
return move(*this);
|
||
}
|
||
|
||
Setup&&
|
||
withSleepPeriod (std::chrono::milliseconds millis)
|
||
{
|
||
IDLE_WAIT = millis;
|
||
return move(*this);
|
||
}
|
||
|
||
Setup&&
|
||
dismissAfter (size_t cycles)
|
||
{
|
||
DISMISS_CYCLES = cycles;
|
||
return move(*this);
|
||
}
|
||
|
||
};
|
||
|
||
return Setup{std::forward<FUN> (workFun)};
|
||
}
|
||
}
|
||
|
||
|
||
|
||
|
||
/*************************************************************************//**
|
||
* @test WorkForce-Service: maintain a pool of active worker threads.
|
||
* @warning this test relies on empirical timings and can be brittle.
|
||
* @see SchedulerUsage_test
|
||
*/
|
||
class WorkForce_test : public Test
|
||
{
|
||
|
||
virtual void
|
||
run (Arg)
|
||
{
|
||
simpleUsage();
|
||
|
||
verify_pullWork();
|
||
verify_workerHalt();
|
||
verify_workerSleep();
|
||
verify_workerRetard();
|
||
verify_workerDismiss();
|
||
verify_finalHook();
|
||
verify_detectError();
|
||
verify_defaultPool();
|
||
verify_scalePool();
|
||
verify_countActive();
|
||
verify_dtor_blocks();
|
||
}
|
||
|
||
|
||
/** @test demonstrate simple worker pool usage
|
||
*/
|
||
void
|
||
simpleUsage()
|
||
{
|
||
atomic<uint> check{0};
|
||
WorkForce wof{setup ([&]{ ++check; return activity::PASS; })};
|
||
// ^^^ this is the doWork-λ
|
||
CHECK (0 == check);
|
||
|
||
wof.activate();
|
||
sleep_for(20ms);
|
||
|
||
CHECK (0 < check); // λ invoked in the worker threads
|
||
}
|
||
|
||
|
||
|
||
/** @test the given work-functor is invoked repeatedly, once activated.
|
||
*/
|
||
void
|
||
verify_pullWork()
|
||
{
|
||
atomic<uint> check{0};
|
||
WorkForce wof{setup ([&]{ ++check; return activity::PASS; })};
|
||
|
||
CHECK (0 == check);
|
||
|
||
wof.incScale();
|
||
sleep_for(20ms);
|
||
|
||
uint invocations = check;
|
||
CHECK (0 < invocations);
|
||
|
||
sleep_for(2ms);
|
||
CHECK (invocations < check);
|
||
|
||
invocations = check;
|
||
sleep_for(2ms);
|
||
CHECK (invocations < check);
|
||
|
||
wof.awaitShutdown();
|
||
|
||
invocations = check;
|
||
sleep_for(2ms);
|
||
CHECK (invocations == check);
|
||
}
|
||
|
||
|
||
|
||
/** @test can cause a worker to terminate by return-value from the work-functor
|
||
*/
|
||
void
|
||
verify_workerHalt()
|
||
{
|
||
atomic<uint> check{0};
|
||
atomic<activity::Proc> control{activity::PASS};
|
||
WorkForce wof{setup ([&]{ ++check; return activity::Proc(control); })};
|
||
|
||
wof.incScale();
|
||
sleep_for(1ms);
|
||
|
||
uint invocations = check;
|
||
CHECK (0 < invocations);
|
||
|
||
control = activity::HALT;
|
||
sleep_for(1ms);
|
||
|
||
invocations = check;
|
||
sleep_for(10ms);
|
||
CHECK (invocations == check);
|
||
}
|
||
|
||
|
||
|
||
/** @test a worker can be sent to sleep, throttling the poll frequency.
|
||
*/
|
||
void
|
||
verify_workerSleep()
|
||
{
|
||
atomic<uint> check{0};
|
||
WorkForce wof{setup ([&]{ ++check; return activity::WAIT; })
|
||
.withSleepPeriod (10ms)};
|
||
|
||
wof.incScale();
|
||
sleep_for(1ms);
|
||
|
||
CHECK (1 == check);
|
||
|
||
sleep_for(10us);
|
||
CHECK (1 == check);
|
||
|
||
sleep_for(12ms); // after waiting one sleep-period...
|
||
CHECK (2 == check); // ...functor invoked again
|
||
}
|
||
|
||
|
||
|
||
/** @test a worker can be retarded and throttled in case of contention.
|
||
*/
|
||
void
|
||
verify_workerRetard()
|
||
{
|
||
atomic<uint> check{0};
|
||
{ // ▽▽▽▽ regular work-cycles without delay
|
||
WorkForce wof{setup ([&]{ ++check; return activity::PASS; })};
|
||
wof.incScale();
|
||
sleep_for(5ms);
|
||
}
|
||
uint cyclesPASS{check};
|
||
check = 0;
|
||
{ // ▽▽▽▽ signals »contention«
|
||
WorkForce wof{setup ([&]{ ++check; return activity::KICK; })};
|
||
wof.incScale();
|
||
sleep_for(5ms);
|
||
}
|
||
uint cyclesKICK{check};
|
||
CHECK (cyclesKICK < cyclesPASS);
|
||
CHECK (cyclesKICK < 50);
|
||
}
|
||
|
||
|
||
|
||
/** @test when a worker is sent into sleep-cycles for an extended time,
|
||
* the worker terminates itself.
|
||
*/
|
||
void
|
||
verify_workerDismiss()
|
||
{
|
||
atomic<uint> check{0};
|
||
WorkForce wof{setup ([&]{ ++check; return activity::WAIT; })
|
||
.withSleepPeriod (10ms)
|
||
.dismissAfter(5)};
|
||
|
||
wof.incScale();
|
||
sleep_for(1ms);
|
||
|
||
CHECK (1 == check);
|
||
|
||
sleep_for(12ms);
|
||
CHECK (2 == check); // after one wait cycle, one further invocation
|
||
|
||
sleep_for(100ms);
|
||
CHECK (5 == check); // only 5 invocations total...
|
||
CHECK (0 == wof.size()); // ...after that, the worker terminated
|
||
}
|
||
|
||
|
||
|
||
/** @test verify invocation of a thread-termination callback
|
||
*/
|
||
void
|
||
verify_finalHook()
|
||
{
|
||
atomic<uint> exited{0};
|
||
atomic<activity::Proc> control{activity::PASS};
|
||
WorkForce wof{setup([&]{ return activity::Proc(control); })
|
||
.withFinalHook([&](bool){ ++exited; })};
|
||
|
||
CHECK (0 == exited);
|
||
|
||
wof.activate();
|
||
sleep_for(10ms);
|
||
CHECK (wof.size() == work::Config::COMPUTATION_CAPACITY);
|
||
CHECK (0 == exited);
|
||
|
||
control = activity::HALT;
|
||
sleep_for(10ms);
|
||
CHECK (0 == wof.size());
|
||
CHECK (exited == work::Config::COMPUTATION_CAPACITY);
|
||
}
|
||
|
||
|
||
|
||
/** @test exceptions emanating from within the worker are catched
|
||
* and reported by setting the isFailure argument flag of
|
||
* the `finalHook` functor invoked at worker termination.
|
||
*/
|
||
void
|
||
verify_detectError()
|
||
{
|
||
atomic<uint> check{0};
|
||
atomic<uint> errors{0};
|
||
WorkForce wof{setup ([&]{
|
||
if (++check == 555)
|
||
throw error::State("evil");
|
||
return activity::PASS;
|
||
})
|
||
.withFinalHook([&](bool isFailure)
|
||
{
|
||
if (isFailure)
|
||
++errors;
|
||
})};
|
||
CHECK (0 == check);
|
||
CHECK (0 == errors);
|
||
|
||
wof.incScale();
|
||
wof.incScale();
|
||
wof.incScale();
|
||
|
||
sleep_for(10us);
|
||
CHECK (3 == wof.size());
|
||
CHECK (0 < check);
|
||
CHECK (0 == errors);
|
||
|
||
sleep_for(200ms); // wait for the programmed disaster
|
||
CHECK (2 == wof.size());
|
||
CHECK (1 == errors);
|
||
}
|
||
|
||
|
||
|
||
/** @test by default, the WorkForce is initially inactive;
|
||
* once activated, it scales up to the number of cores
|
||
* reported by the runtime system.
|
||
*/
|
||
void
|
||
verify_defaultPool()
|
||
{
|
||
atomic<uint> check{0};
|
||
WorkForce wof{setup ([&]{ ++check; return activity::PASS; })};
|
||
|
||
// after construction, the WorkForce is inactive
|
||
CHECK (0 == wof.size());
|
||
CHECK (0 == check);
|
||
|
||
wof.activate();
|
||
sleep_for(20ms);
|
||
|
||
CHECK (0 < check);
|
||
CHECK (wof.size() == work::Config::COMPUTATION_CAPACITY);
|
||
CHECK (work::Config::COMPUTATION_CAPACITY == std::thread::hardware_concurrency());
|
||
}
|
||
|
||
|
||
|
||
/** @test the number of (separate) workers can be scaled up,
|
||
* both stepwise and as fraction of full hardware concurrency
|
||
*/
|
||
void
|
||
verify_scalePool()
|
||
{
|
||
/** helper to count distinct thread-IDs */
|
||
class UniqueCnt
|
||
: public std::set<std::thread::id>
|
||
, public lib::Sync<>
|
||
{
|
||
public:
|
||
void
|
||
mark (std::thread::id const& tID)
|
||
{
|
||
Lock guard{this};
|
||
this->insert(tID);
|
||
}
|
||
|
||
operator size_t() const
|
||
{
|
||
Lock guard{this};
|
||
return this->size();
|
||
}
|
||
}
|
||
uniqueCnt;
|
||
|
||
WorkForce wof{setup ([&]{
|
||
uniqueCnt.mark(std::this_thread::get_id());
|
||
return activity::PASS;
|
||
})};
|
||
|
||
CHECK (0 == uniqueCnt);
|
||
CHECK (0 == wof.size());
|
||
|
||
wof.incScale();
|
||
sleep_for(1ms);
|
||
CHECK (1 == uniqueCnt);
|
||
CHECK (1 == wof.size());
|
||
|
||
wof.incScale();
|
||
sleep_for(1ms);
|
||
CHECK (2 == uniqueCnt);
|
||
CHECK (2 == wof.size());
|
||
|
||
|
||
auto fullCnt = work::Config::COMPUTATION_CAPACITY;
|
||
|
||
wof.activate (1.0);
|
||
sleep_for(5ms);
|
||
CHECK (fullCnt == uniqueCnt);
|
||
CHECK (fullCnt == wof.size());
|
||
|
||
wof.activate (2.0);
|
||
sleep_for(10ms);
|
||
CHECK (2*fullCnt == uniqueCnt);
|
||
CHECK (2*fullCnt == wof.size());
|
||
|
||
wof.awaitShutdown();
|
||
CHECK (0 == wof.size());
|
||
|
||
uniqueCnt.clear();
|
||
sleep_for(5ms);
|
||
CHECK (0 == uniqueCnt);
|
||
|
||
wof.activate (0.5);
|
||
sleep_for(5ms);
|
||
CHECK (fullCnt/2 == uniqueCnt);
|
||
CHECK (fullCnt/2 == wof.size());
|
||
}
|
||
|
||
|
||
|
||
/** @test dynamically determine count of currently active workers.
|
||
*/
|
||
void
|
||
verify_countActive()
|
||
{
|
||
atomic<uint> check{0};
|
||
WorkForce wof{setup ([&]{
|
||
++check;
|
||
if (check == 5'000 or check == 5'110)
|
||
return activity::HALT;
|
||
else
|
||
return activity::PASS;
|
||
})};
|
||
|
||
CHECK (0 == wof.size());
|
||
|
||
wof.incScale();
|
||
wof.incScale();
|
||
wof.incScale();
|
||
sleep_for(15us); // this may be fragile; must be sufficiently short
|
||
|
||
CHECK (3 == wof.size());
|
||
|
||
while (check < 6'000)
|
||
sleep_for(15ms); // .....sufficiently long to count way beyond 10'000
|
||
CHECK (check > 6'000);
|
||
CHECK (1 == wof.size());
|
||
}
|
||
|
||
|
||
|
||
/** @test verify that the WorkForce dtor waits for all active threads to disappear
|
||
* - use a work-functor which keeps all workers blocked
|
||
* - start the WorkForce within a separate thread
|
||
* - in this separate thread, cause the WorkForce destructor to be called
|
||
* - in the test main thread release the work-functor blocking
|
||
* - at this point, all workers return, detect shutdown and terminate
|
||
*/
|
||
void
|
||
verify_dtor_blocks()
|
||
{
|
||
atomic<bool> trapped{true};
|
||
auto blockingWork = [&]{
|
||
while (trapped)
|
||
/* spin */;
|
||
return activity::PASS;
|
||
};
|
||
|
||
atomic<bool> pool_scaled_up{false};
|
||
atomic<bool> shutdown_done{false};
|
||
|
||
Thread operate{"controller"
|
||
,[&] {
|
||
{// nested scope...
|
||
WorkForce wof{setup (blockingWork)};
|
||
|
||
wof.activate();
|
||
sleep_for (10ms);
|
||
CHECK (wof.size() == work::Config::COMPUTATION_CAPACITY);
|
||
pool_scaled_up = true;
|
||
} // WorkForce goes out of scope => dtor called
|
||
|
||
// when reaching this point, dtor has terminated
|
||
shutdown_done = true;
|
||
}};
|
||
|
||
CHECK (operate); // operate-thread is in running state
|
||
sleep_for(100ms);
|
||
|
||
CHECK (pool_scaled_up);
|
||
CHECK (not shutdown_done); // all workers are trapped in the work-functor
|
||
// thus the destructor can't dismantle the pool
|
||
trapped = false;
|
||
sleep_for(40ms);
|
||
CHECK (shutdown_done);
|
||
CHECK (not operate); // operate-thread has detached and terminated
|
||
}
|
||
};
|
||
|
||
|
||
/** Register this test class... */
|
||
LAUNCHER (WorkForce_test, "unit engine");
|
||
|
||
|
||
}}} // namespace vault::gear::test
|