Commit 8f7812e6 authored by Dominik Charousset's avatar Dominik Charousset

Avoid putting two stream stages to the same worker

parent c42dcaf0
......@@ -64,7 +64,7 @@ public:
}
template <class Coordinator>
void central_enqueue(Coordinator* self, resumable* job) {
void central_enqueue(Coordinator* self, resumable* job, execution_unit*) {
enqueue(self, job);
}
......
......@@ -102,8 +102,11 @@ public:
}
template <class Coordinator>
void central_enqueue(Coordinator* self, resumable* job) {
void central_enqueue(Coordinator* self, resumable* job,
execution_unit* avoid) {
auto w = self->worker_by_id(d(self).next_worker++ % self->num_workers());
while (w == avoid)
w = self->worker_by_id(d(self).next_worker++ % self->num_workers());
w->external_enqueue(job);
}
......
......@@ -58,8 +58,8 @@ public:
return utility_actors_.size();
}
/// Puts `what` into the queue of a randomly chosen worker.
virtual void enqueue(resumable* what) = 0;
/// Puts `what` into the queue of a worker other than `avoid`.
virtual void enqueue(resumable* what, execution_unit* avoid = nullptr) = 0;
inline actor_system& system() {
return system_;
......
......@@ -145,8 +145,8 @@ protected:
timer_.join();
}
void enqueue(resumable* ptr) override {
policy_.central_enqueue(this, ptr);
void enqueue(resumable* ptr, execution_unit* avoid) override {
policy_.central_enqueue(this, ptr, avoid);
}
detail::thread_safe_actor_clock& clock() noexcept override {
......
......@@ -147,7 +147,7 @@ protected:
void stop() override;
void enqueue(resumable* ptr) override;
void enqueue(resumable* ptr, execution_unit* avoid) override;
private:
void inline_all_enqueues_helper();
......
......@@ -166,10 +166,12 @@ void scheduled_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) {
CAF_ASSERT(private_thread_ != nullptr);
private_thread_->resume();
} else {
if (eu != nullptr)
// We pull actors towards us for ordinary messages but push them away
// for stream processing.
if (eu != nullptr && !mid.is_stream_message())
eu->exec_later(this);
else
home_system().scheduler().enqueue(this);
home_system().scheduler().enqueue(this, eu);
}
break;
}
......
......@@ -91,7 +91,7 @@ void test_coordinator::stop() {
trigger_timeouts();
}
void test_coordinator::enqueue(resumable* ptr) {
void test_coordinator::enqueue(resumable* ptr, execution_unit*) {
CAF_LOG_TRACE("");
jobs.push_back(ptr);
if (after_next_enqueue_ != nullptr) {
......
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