From 1cbb6b7371added9f1ed401534aa68b5b85c5b0d Mon Sep 17 00:00:00 2001
From: Ichthyostega
Date: Sat, 16 Dec 2023 23:47:50 +0100
Subject: [PATCH] Scheduler-test: rework handling of notifications in the
Activity-Language
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
While the recent refactoring...
206c67cc
...was a step into the right direction, it pushed too hard,
overlooking the requirement to protect the scheduler contents
and thus all of the Activity-chains against concurrent modification.
Moreover, the recent solution still seems not quite orthogonal.
Thus the handling of notifications was thoroughly reworked:
- the explicit "double-dispatch" was removed, since actual usage
of the language indicates that we only need notifications to
Gate (and Hook), but not to any other conceivable Activity.
- thus it seems unnecessary to turn "notification" into some kind
of secondary work mode. Rather, it is folded as special case
into the regular dispatch.
This leads to new processing rules:
- a POST goes into λ-post (obviously... that's its meaning)
- a NOTIFY now passes its *target* into λ-post
- λ-post invokes ''dispatch''
- and **dispatching a Gate now implies to notify the Gate**
This greatly simplifies the »state machine« in the Activity-Language,
but also incurs some limitations (which seems adequate, since it is
now clear that we do not ''schedule'' or ''dispatch'' arbitrary
Activities — rather we'll do this only with POST and NOTIFY,
and all further processing happens by passing activation
along the chain, without involving the Scheduler)
---
src/vault/gear/activity-lang.cpp | 4 +-
src/vault/gear/activity-term.hpp | 2 +-
src/vault/gear/activity.hpp | 122 +++----
src/vault/gear/scheduler.hpp | 7 -
tests/vault/gear/activity-detector-test.cpp | 23 +-
tests/vault/gear/activity-detector.hpp | 102 +++---
tests/vault/gear/scheduler-activity-test.cpp | 148 ++++----
wiki/thinkPad.ichthyo.mm | 335 ++++++++++++++++++-
8 files changed, 536 insertions(+), 207 deletions(-)
diff --git a/src/vault/gear/activity-lang.cpp b/src/vault/gear/activity-lang.cpp
index 6924ee4e8..96f966a7d 100644
--- a/src/vault/gear/activity-lang.cpp
+++ b/src/vault/gear/activity-lang.cpp
@@ -107,8 +107,8 @@ namespace gear {
+ util::toString(data_.timing.quality);
case NOTIFY:
return util::showAddr(data_.notification.target)
- + ", report="
- + util::toString(data_.notification.report);
+ + ", timing:"
+ + util::toString(Time{data_.notification.timing});
case GATE:
return "<"
+ util::toString(data_.condition.rest)
diff --git a/src/vault/gear/activity-term.hpp b/src/vault/gear/activity-term.hpp
index fdfc50f6f..e472bd7a4 100644
--- a/src/vault/gear/activity-term.hpp
+++ b/src/vault/gear/activity-term.hpp
@@ -176,7 +176,7 @@ namespace gear {
REQUIRE (notificationSrc.is (Activity::NOTIFY));
setupGate();
gate_->incDependencies();
- notificationSrc.setNotificationTarget (gate_);
+ notificationSrc.setNotificationTarget (gate_, Time{post_->data_.timeWindow.life});
return *this;
}
diff --git a/src/vault/gear/activity.hpp b/src/vault/gear/activity.hpp
index 813284221..169f65b77 100644
--- a/src/vault/gear/activity.hpp
+++ b/src/vault/gear/activity.hpp
@@ -186,6 +186,8 @@ namespace gear {
, Time now
, void* executionCtx) =0;
+ virtual Time getDeadline() const =0;
+
virtual std::string
diagnostic() const
{
@@ -204,8 +206,6 @@ namespace gear {
* The Execution Context need to be passed to any Activity _activation;_
* it provides the _bindings_ for functionality defined only on a conceptual
* level, and provided by an opaque implementation (actually the Scheduler)
- * @remark `getWaitDelay` was once used for Gate, but is now an obscure
- * fall-back for _other notifications_ (retained for future use)
*/
template
constexpr void
@@ -216,7 +216,6 @@ namespace gear {
ASSERT_MEMBER_FUNCTOR (&EXE::done, void(Time, size_t));
ASSERT_MEMBER_FUNCTOR (&EXE::tick, Proc(Time));
- ASSERT_MEMBER_FUNCTOR (&EXE::getWaitDelay, Offset());
ASSERT_MEMBER_FUNCTOR (&EXE::getSchedTime, Time());
}
@@ -291,6 +290,7 @@ namespace gear {
bool isDead (Time now) const { return dead <= now;}
bool isHold () const { return rest > 0; }
bool isFree (Time now) const { return not (isHold() or isDead(now)); }
+ Time getDeadline() const { return Time{dead}; }
void incDependencies() { ++rest; }
Time
@@ -320,7 +320,7 @@ namespace gear {
struct Notification
{
Activity* target;
- size_t report;
+ Instant timing;
};
@@ -367,10 +367,11 @@ namespace gear {
}
explicit
- Activity (Activity* target) noexcept
+ Activity (Activity* target, Time limitWhen =Time::ANYTIME) noexcept
: Activity{NOTIFY}
{
data_.notification.target = target;
+ data_.notification.timing = limitWhen;
}
explicit
@@ -424,13 +425,10 @@ namespace gear {
template
activity::Proc dispatch (Time now, EXE& executionCtx);
- template
- activity::Proc notify (Time now, EXE& executionCtx);
-
/* === special case access and manipulation === */
bool
- is (Activity::Verb expectedVerb)
+ is (Activity::Verb expectedVerb) const
{
return expectedVerb == this->verb_;
}
@@ -443,10 +441,11 @@ namespace gear {
}
void
- setNotificationTarget (Activity* target)
+ setNotificationTarget (Activity* target, Time limitStart =Time::ANYTIME)
{
REQUIRE (is (NOTIFY));
data_.notification.target = target;
+ data_.notification.timing = limitStart;
}
Time
@@ -532,15 +531,6 @@ namespace gear {
return activity::PASS;
}
- template
- activity::Proc
- notifyTarget (Time now, EXE& executionCtx)
- {
- REQUIRE (NOTIFY == verb_);
- REQUIRE (data_.notification.target);
- return data_.notification.target->notify (now, executionCtx);
- }
-
template
activity::Proc
checkGate (Time now, EXE&)
@@ -554,9 +544,8 @@ namespace gear {
return activity::PASS;
}
- template
activity::Proc
- receiveGateNotification (Time now, EXE& executionCtx)
+ receiveGateNotification (Time now)
{
REQUIRE (GATE == verb_);
if (data_.condition.rest > 0)
@@ -565,33 +554,39 @@ namespace gear {
// maybe the Gate has been opened by this notification?
if (data_.condition.isFree(now))
{//yes => activate gated chain but lock redundant invocations
- Time dead = data_.condition.lockPermanently();
- return postChain (now,dead, executionCtx);
+ data_.condition.lockPermanently();
+ return activity::PASS;
} }
- return activity::PASS;
- }
-
- template
- activity::Proc
- dispatchSelf (Time when, Time dead, EXE& executionCtx)
- {
- return executionCtx.post (when, dead, this, executionCtx);
- }
-
- template
- activity::Proc
- dispatchSelfDelayed (Time now, Time dead, EXE& executionCtx)
- {
- dispatchSelf (now + executionCtx.getWaitDelay(), dead, executionCtx);
return activity::SKIP;
}
template
activity::Proc
- postChain (Time when, Time dead, EXE& executionCtx)
+ postSelf (Time now, EXE& executionCtx)
{
REQUIRE (next);
- return executionCtx.post (when, dead, next, executionCtx);
+ if (is(POST))
+ return executionCtx.post (Time{data_.timeWindow.life},Time{data_.timeWindow.dead}, this, executionCtx);
+ else
+ return executionCtx.post (now,Time::NEVER, this, executionCtx);
+ }
+
+ template
+ activity::Proc
+ postNotify (Time now, EXE& executionCtx)
+ {
+ REQUIRE (is(NOTIFY));
+ Activity* target = data_.notification.target;
+ REQUIRE (target);
+ REQUIRE (not target->is(HOOK) or target->data_.callback.hook);
+ Time startHint = target->is(GATE) or
+ target->is(HOOK)? Time{data_.notification.timing}
+ : now;
+ Time deadline = target->is(GATE)? target->data_.condition.getDeadline()
+ : target->is(HOOK)? target->data_.callback.hook->getDeadline()
+ : Time::NEVER;
+ // indirectly forward to Activity::dispatch()
+ return executionCtx.post (startHint,deadline, target, executionCtx);
}
template
@@ -650,11 +645,11 @@ namespace gear {
case WORKSTOP:
return signalStop (now, executionCtx);
case NOTIFY:
- return dispatch (now, executionCtx); //▷ special processing for the Notification
+ return postNotify (now, executionCtx);
+ case POST:
+ return postSelf (now, executionCtx);
case GATE:
return checkGate (now, executionCtx);
- case POST:
- return dispatchSelf (Time{data_.timeWindow.life},Time{data_.timeWindow.dead}, executionCtx);
case FEED:
return activity::PASS;
case HOOK:
@@ -673,6 +668,11 @@ namespace gear {
* by invoking the `post`-λ on the \a executionCtx, or by _activating_
* a `POST`-Activity. Control flow passing here has acquired the `GroomingToken`
* and can thus assume single threaded execution until `WORKSTART`.
+ *
+ * Notably this entrance is used to implement _gating_ to wait for prerequisites;
+ * when a notification is passed to a `GATE`-Activity, the embedded counter is
+ * decremented; after all prerequisites are „checked off“ this way, the
+ * Activity-chain behind the Gate is activated.
* @param now the scheduler-time; assuming that a call through the `post`-λ will
* only be actually de-queued when scheduler-time equals the start time
* defined in the POST activity and passed through the `post`-λ as parameter.
@@ -688,46 +688,18 @@ namespace gear {
activity::_verify_usable_as_ExecutionContext();
switch (verb_) {
- case NOTIFY:
- return notifyTarget (now, executionCtx);
case POST:
case FEED: // signal just to proceed with next...
return activity::PASS;
+ case GATE:
+ return receiveGateNotification (now);
+ case HOOK:
+ return notifyHook (now, executionCtx);
default:
return activate (now, executionCtx);
}
}
- /**
- * Special operation to receive a message or trigger from some other Activity.
- * Notably this is used to implement _gating_ to wait for prerequisites; when
- * a notification is passed to a `GATE`-Activity, the embedded counter is
- * decremented; after all prerequisites are „checked off“ this way, the
- * Activity-chain behind the Gate is activated.
- * @note this function is invoked from the context of the source, and
- * thus any follow-up actions beyond that scope are re-POSTed,
- * after possibly performing the GATE-check.
- */
- template
- activity::Proc
- Activity::notify (Time now, EXE& executionCtx)
- {
- activity::_verify_usable_as_ExecutionContext();
-
- switch (verb_) {
- case GATE:
- return receiveGateNotification (now, executionCtx);
- case HOOK:
- return notifyHook (now, executionCtx);
- case POST:
- case FEED: // ▽▽▽▽▽ implies to use a deadline from the context
- return postChain (now,Time::NEVER, executionCtx);
- default:
- return dispatchSelfDelayed (now,Time::NEVER, executionCtx);
- } // Fallback: self-re-dispatch for async execution (-> getWaitDelay())
- }
-
-
}} // namespace vault::gear
#endif /*SRC_VAULT_GEAR_ACTIVITY_H_*/
diff --git a/src/vault/gear/scheduler.hpp b/src/vault/gear/scheduler.hpp
index f10198005..6d304a254 100644
--- a/src/vault/gear/scheduler.hpp
+++ b/src/vault/gear/scheduler.hpp
@@ -140,7 +140,6 @@ namespace gear {
const auto IDLE_WAIT = 20ms; ///< sleep-recheck cycle for workers deemed _idle_
const size_t DISMISS_CYCLES = 100; ///< number of wait cycles before an idle worker terminates completely
- Offset POLL_WAIT_DELAY{FSecs(1,1000)}; ///< delay until re-evaluating after notification (obscure feature, retained for future use)
Offset DUTY_CYCLE_PERIOD{FSecs(1,20)}; ///< period of the regular scheduler »tick« for state maintenance.
Offset DUTY_CYCLE_TOLERANCE{FSecs(1,10)}; ///< maximum slip tolerated on duty-cycle start before triggering Scheduler-emergency
}
@@ -559,12 +558,6 @@ namespace gear {
return activity::PASS;
}
- Offset
- getWaitDelay()
- {
- return POLL_WAIT_DELAY;
- }
-
/** access high-resolution-clock, rounded to µ-Ticks */
Time
getSchedTime()
diff --git a/tests/vault/gear/activity-detector-test.cpp b/tests/vault/gear/activity-detector-test.cpp
index 4dcd10996..e0588baf0 100644
--- a/tests/vault/gear/activity-detector-test.cpp
+++ b/tests/vault/gear/activity-detector-test.cpp
@@ -240,8 +240,8 @@ namespace test {
CHECK (detector.verifyInvocation(CTX_TICK).arg(t));
detector.incrementSeq();
- ctx.tick.returning(activity::KILL);
- CHECK (activity::KILL == ctx.tick(t));
+ ctx.tick.returning(activity::KICK);
+ CHECK (activity::KICK == ctx.tick(t));
CHECK (detector.verifyInvocation(CTX_TICK).timeArg(t));
CHECK (detector.verifyInvocation(CTX_WORK).timeArg(t)
@@ -360,21 +360,28 @@ namespace test {
void
watch_notification()
{
+ Time tt{11,11}; // start time of the NOTIFY
+ Time ts{22,22}; // start time of the target-chain
+ Time td{33,33}; // deadline for the target-chain
ActivityDetector detector;
Activity chain;
- Activity gate{1, Time{22,22}};
+ Activity gate{1, td};
gate.next = &chain;
- Activity notification{&gate};
+ Activity notification{&gate, ts}; // note: follow-up start time `ts` injected here
CHECK (gate.data_.condition.rest == 1);
detector.insertActivationTap (notification.data_.notification.target);
- Time tt{11,11};
- notification.dispatch (tt, detector.executionCtx);
+ notification.activate (tt, detector.executionCtx); // dispatch time `tt` (is actually irrelevant here)
+ // activating the NOTIFY causes it to POST its target, thereby setting the deadline from the GATE
+ CHECK (detector.verifyInvocation("CTX-post").arg("22.022","33.033", "tap-GATE", "≺test::CTX≻"));
- CHECK (detector.verifyInvocation("tap-GATE").arg("11.011 --notify-↯> Act(GATE")
- .beforeInvocation("CTX-post").arg("11.011","22.022", "Act(TICK", "≺test::CTX≻"));
+ detector.incrementSeq();
+ // to see the effect of the instrumentation, we need to mimic the behaviour of λ-post,
+ // which is to call Activity::dispatch() on the given target
+ notification.data_.notification.target->dispatch (ts, detector.executionCtx); // note: using `ts` for the follow-up chain
+ CHECK (detector.verifyInvocation("tap-GATE").seq(1).arg("22.022 --notify-↯> Act(GATE"));
CHECK (gate.data_.condition.rest == 0);
}
diff --git a/tests/vault/gear/activity-detector.hpp b/tests/vault/gear/activity-detector.hpp
index 2b3dc59c5..cb896dd5f 100644
--- a/tests/vault/gear/activity-detector.hpp
+++ b/tests/vault/gear/activity-detector.hpp
@@ -102,6 +102,7 @@ namespace test {
using lib::time::FSecs;
using lib::time::Offset;
using lib::meta::RebindVariadic;
+ using util::unConst;
using util::isnil;
using std::forward;
using std::move;
@@ -119,7 +120,6 @@ namespace test {
const string CTX_DONE{"CTX-done"};
const string CTX_TICK{"CTX-tick"};
- Offset POLL_WAIT_DELAY{FSecs(1)};
Time SCHED_TIME_MARKER{555,5}; ///< marker value for "current scheduler time" used in tests
}
@@ -355,61 +355,80 @@ namespace test {
{
Logger log_;
TimeVar invoked_{Time::ANYTIME};
-
+
+ Activity*
+ target()
+ {
+ return reinterpret_cast (data_.callback.arg);
+ }
+
+ Activity const*
+ target() const
+ {
+ return unConst(this)->target();
+ }
+
activity::Proc
activation ( Activity& thisHook
, Time now
, void* executionCtx) override
- {
- REQUIRE (thisHook.is (Activity::HOOK));
- invoked_ = now;
- if (data_.callback.arg == 0)
- {// no adapted target; just record this activation
- log_(util::toString(now) + " ⧐ ");
- return activity::PASS;
- }
- else
- {// forward activation to the adapted target Activity
- Activity& target = *reinterpret_cast (data_.callback.arg);
- auto ctx = *static_cast (executionCtx);
- log_(util::toString(now) + " ⧐ " + util::toString (target));
- return target.activate (now, ctx);
- }
- }
+ {
+ REQUIRE (thisHook.is (Activity::HOOK));
+ invoked_ = now;
+ if (not target())
+ {// no adapted target; just record this activation
+ log_(util::toString(now) + " ⧐ ");
+ return activity::PASS;
+ }
+ else
+ {// forward activation to the adapted target Activity
+ auto ctx = *static_cast (executionCtx);
+ log_(util::toString(now) + " ⧐ " + util::toString (*target()));
+ return target()->activate (now, ctx);
+ }
+ }
activity::Proc
notify ( Activity& thisHook
, Time now
, void* executionCtx) override
- {
- REQUIRE (thisHook.is (Activity::HOOK));
- invoked_ = now;
- if (data_.callback.arg == 0)
- {// no adapted target; just record this notification
- log_(util::toString(now) + " --notify-↯• ");
- return activity::PASS;
- }
- else
- {// forward notification-dispatch to the adapted target Activity
- Activity& target = *reinterpret_cast (data_.callback.arg);
- auto ctx = *static_cast (executionCtx);
- log_(util::toString(now) + " --notify-↯> " + util::toString (target));
- return target.notify (now, ctx);
- }
- }
+ {
+ REQUIRE (thisHook.is (Activity::HOOK));
+ invoked_ = now;
+ if (not target())
+ {// no adapted target; just record this notification
+ log_(util::toString(now) + " --notify-↯• ");
+ return activity::PASS;
+ }
+ else
+ {// forward notification-dispatch to the adapted target Activity
+ auto ctx = *static_cast (executionCtx);
+ log_(util::toString(now) + " --notify-↯> " + util::toString (*target()));
+ return target()->dispatch (now, ctx);
+ }
+ }
- std::string
- diagnostic() const override
- {
- return "Probe("+string{log_}+")";
- }
+ Time
+ getDeadline() const override
+ {
+ if (target() and target()->is(Activity::GATE))
+ return target()->data_.condition.getDeadline();
+ else
+ return Time::NEVER;
+ }
+
+ std::string
+ diagnostic() const override
+ {
+ return "Probe("+string{log_}+")";
+ }
public:
ActivityProbe (string id, EventLog& masterLog, uint const& invocationSeqNr)
: Activity{*this, 0}
, log_{id, masterLog, invocationSeqNr}
{ }
-
+
ActivityProbe (Activity const& subject, string id, EventLog& masterLog, uint const& invocationSeqNr)
: Activity{*this, reinterpret_cast (&subject)}
, log_{id, masterLog, invocationSeqNr}
@@ -428,7 +447,7 @@ namespace test {
{
if (act and act->verb_ == HOOK)
{
- ActivityProbe* probe = dynamic_cast (act->data_.callback.hook);
+ ActivityProbe* probe = dynamic_cast (act->data_.callback.hook);
if (probe)
return probe->invoked_;
}
@@ -591,7 +610,6 @@ namespace test {
_DiagnosticFun::Type done;
_DiagnosticFun::Type tick;
- function getWaitDelay = [] { return POLL_WAIT_DELAY; };
function
-
-
+
@@ -103455,10 +103589,193 @@ Date: Thu Apr 20 18:53:17 2023 +0200
wäre ja so zu erwarten gemäß Topologie...