Commit bcdd0ba3 authored by Dominik Charousset's avatar Dominik Charousset

Streamline hook feature in test coordinator

parent ddb7092e
......@@ -101,12 +101,18 @@ public:
}
/// Tries to execute a single event.
/// Tries to execute a single event in FIFO order.
bool try_run_once();
/// Executes a single event or fails if no event is available.
/// Tries to execute a single event in LIFO order.
bool try_run_once_lifo();
/// Executes a single event in FIFO order or fails if no event is available.
void run_once();
/// Executes a single event in LIFO order or fails if no event is available.
void run_once_lifo();
/// Executes events until the job queue is empty and no pending timeouts are
/// left. Returns the number of processed events.
size_t run(size_t max_count = std::numeric_limits<size_t>::max());
......@@ -122,24 +128,18 @@ public:
/// events (first) and dispatched delayed messages (second).
std::pair<size_t, size_t> run_dispatch_loop();
/// Executes the next `num` enqueued jobs immediately.
inline void inline_next_enqueues(size_t num) {
inline_enqueues_ += num;
}
/// Executes the next enqueued job immediately.
inline void inline_next_enqueue() {
inline_next_enqueues(1);
template <class F>
void after_next_enqueue(F f) {
after_next_enqueue_ = f;
}
/// Executes the next `num` enqueued jobs immediately.
inline void set_inline_enqueues(size_t num) {
inline_enqueues_ = num;
}
/// Executes the next enqueued job immediately by using the
/// `after_next_enqueue` hook.
void inline_next_enqueue();
/// The function used for inlining enqueue operations. The default predicate
/// calls `run()`.
std::function<void()> inlining_hook;
/// Executes all enqueued jobs immediately by using the `after_next_enqueue`
/// hook.
void inline_all_enqueues();
protected:
void start() override;
......@@ -149,8 +149,10 @@ protected:
void enqueue(resumable* ptr) override;
private:
size_t inline_enqueues_;
bool inside_inlined_enqueue;
void inline_all_enqueues_helper();
/// User-provided callback for triggering custom code in `enqueue`.
std::function<void()> after_next_enqueue_;
};
} // namespace scheduler
......
......@@ -92,12 +92,8 @@ private:
} // namespace <anonymous>
test_coordinator::test_coordinator(actor_system& sys)
: super(sys),
inline_enqueues_(0) {
inlining_hook = [=] {
run();
};
test_coordinator::test_coordinator(actor_system& sys) : super(sys) {
// nop
}
void test_coordinator::start() {
dummy_worker worker{this};
......@@ -116,11 +112,10 @@ void test_coordinator::stop() {
void test_coordinator::enqueue(resumable* ptr) {
CAF_LOG_TRACE("");
jobs.push_back(ptr);
if (!inside_inlined_enqueue && inline_enqueues_ > 0) {
--inline_enqueues_;
inside_inlined_enqueue = true;
inlining_hook();
inside_inlined_enqueue = false;
if (after_next_enqueue_ != nullptr) {
std::function<void()> f;
f.swap(after_next_enqueue_);
f();
}
}
......@@ -144,12 +139,26 @@ bool test_coordinator::try_run_once() {
return true;
}
bool test_coordinator::try_run_once_lifo() {
if (jobs.empty())
return false;
if (jobs.size() >= 2)
std::rotate(jobs.rbegin(), jobs.rbegin() + 1, jobs.rend());
return try_run_once();
}
void test_coordinator::run_once() {
if (jobs.empty())
CAF_RAISE_ERROR("No job to run available.");
try_run_once();
}
void test_coordinator::run_once_lifo() {
if (jobs.empty())
CAF_RAISE_ERROR("No job to run available.");
try_run_once_lifo();
}
size_t test_coordinator::run(size_t max_count) {
size_t res = 0;
while (res < max_count && try_run_once())
......@@ -187,6 +196,19 @@ std::pair<size_t, size_t> test_coordinator::run_dispatch_loop() {
return res;
}
void test_coordinator::inline_next_enqueue() {
after_next_enqueue([=] { run_once_lifo(); });
}
void test_coordinator::inline_all_enqueues() {
after_next_enqueue([=] { inline_all_enqueues_helper(); });
}
void test_coordinator::inline_all_enqueues_helper() {
after_next_enqueue([=] { inline_all_enqueues_helper(); });
run_once_lifo();
}
} // namespace caf
} // namespace scheduler
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment