Commit 7a89b0ff authored by Matthias Vallentin's avatar Matthias Vallentin Committed by Dominik Charousset

Bracket worker job resumption with hooks.

The function after_resume allowed for hooking the point directly after a
worker returned from resuming. This commit adds the before_resume hook,
thereby fully bracketing the job execution and making hooking complete.
parent c37aed83
...@@ -80,13 +80,25 @@ class scheduler_policy { ...@@ -80,13 +80,25 @@ class scheduler_policy {
template <class Worker> template <class Worker>
void before_shutdown(Worker* self); void before_shutdown(Worker* self);
/**
* Called immediately before resuming an actor.
*/
template <class Worker>
void before_resume(Worker* self, resumable* job);
/** /**
* Called whenever an actor has been resumed. This function can * Called whenever an actor has been resumed. This function can
* prepare some fields before the next resume operation takes place * prepare some fields before the next resume operation takes place
* or perform cleanup actions between to actor runs. * or perform cleanup actions between to actor runs.
*/ */
template <class Worker> template <class Worker>
void after_resume(Worker* self); void after_resume(Worker* self, resumable* job);
/**
* Called whenever an actor has completed a job.
*/
template <class Worker>
void after_completion(Worker* self, resumable* job);
/** /**
* Applies given functor to all resumables attached to a worker. * Applies given functor to all resumables attached to a worker.
......
...@@ -117,8 +117,6 @@ class work_stealing { ...@@ -117,8 +117,6 @@ class work_stealing {
template <class Worker> template <class Worker>
void internal_enqueue(Worker* self, resumable* job) { void internal_enqueue(Worker* self, resumable* job) {
d(self).queue.prepend(job); d(self).queue.prepend(job);
// give others the opportunity to steal from us
after_resume(self);
} }
template <class Worker> template <class Worker>
...@@ -179,7 +177,17 @@ class work_stealing { ...@@ -179,7 +177,17 @@ class work_stealing {
} }
template <class Worker> template <class Worker>
void after_resume(Worker*) { void before_resume(Worker*, resumable*) {
// nop
}
template <class Worker>
void after_resume(Worker*, resumable*) {
// nop
}
template <class Worker>
void after_completion(Worker*, resumable*) {
// nop // nop
} }
......
...@@ -126,25 +126,31 @@ class worker : public execution_unit { ...@@ -126,25 +126,31 @@ class worker : public execution_unit {
CAF_REQUIRE(job != nullptr); CAF_REQUIRE(job != nullptr);
CAF_LOG_DEBUG("resume actor " << id_of(job)); CAF_LOG_DEBUG("resume actor " << id_of(job));
CAF_PUSH_AID_FROM_PTR(dynamic_cast<abstract_actor*>(job)); CAF_PUSH_AID_FROM_PTR(dynamic_cast<abstract_actor*>(job));
m_policy.before_resume(this, job);
switch (job->resume(this, m_max_throughput)) { switch (job->resume(this, m_max_throughput)) {
case resumable::resume_later: { case resumable::resume_later: {
m_policy.resume_job_later(this, job); m_policy.resume_job_later(this, job);
m_policy.after_resume(this, job);
break; break;
} }
case resumable::done: { case resumable::done: {
m_policy.after_resume(this, job);
m_policy.after_completion(this, job);
job->detach_from_scheduler(); job->detach_from_scheduler();
break; break;
} }
case resumable::awaiting_message: { case resumable::awaiting_message: {
// resumable will be enqueued again later // resumable will be enqueued again later
m_policy.after_resume(this, job);
break; break;
} }
case resumable::shutdown_execution_unit: { case resumable::shutdown_execution_unit: {
m_policy.after_resume(this, job);
m_policy.after_completion(this, job);
m_policy.before_shutdown(this); m_policy.before_shutdown(this);
return; return;
} }
} }
m_policy.after_resume(this);
} }
} }
// number of messages each actor is allowed to consume per resume // number of messages each actor is allowed to consume per resume
......
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