Commit a2db868f authored by Sebastian Woelke's avatar Sebastian Woelke

Implement soft actor pinning

parent 59c6e837
...@@ -259,6 +259,10 @@ public: ...@@ -259,6 +259,10 @@ public:
size_t work_stealing_relaxed_steal_interval; size_t work_stealing_relaxed_steal_interval;
size_t work_stealing_relaxed_sleep_duration_us; size_t work_stealing_relaxed_sleep_duration_us;
// -- config parameters for numa aware work-stealing -------------------------
size_t numa_aware_work_stealing_neighborhood_level;
// -- config parameters for the logger --------------------------------------- // -- config parameters for the logger ---------------------------------------
std::string logger_filename; std::string logger_filename;
......
...@@ -40,7 +40,10 @@ public: ...@@ -40,7 +40,10 @@ public:
/// Enqueues `ptr` to the job list of the execution unit. /// Enqueues `ptr` to the job list of the execution unit.
/// @warning Must only be called from a {@link resumable} currently /// @warning Must only be called from a {@link resumable} currently
/// executed by this execution unit. /// executed by this execution unit.
virtual void exec_later(resumable* ptr) = 0; virtual void exec_later(resumable* ptr, bool high_prio = true) = 0;
/// Checks if `ptr` has a high memory locality to this execution_unit
virtual bool is_neighbor(execution_unit* ptr) const = 0;
/// Returns the enclosing actor system. /// Returns the enclosing actor system.
/// @warning Must be set before the execution unit calls `resume` on an actor. /// @warning Must be set before the execution unit calls `resume` on an actor.
......
...@@ -102,11 +102,14 @@ public: ...@@ -102,11 +102,14 @@ public:
template <class Worker> template <class Worker>
struct worker_data { struct worker_data {
using worker_matrix_t = std::vector<std::vector<Worker*>>; using neighbors_t = std::vector<Worker*>;
using worker_matrix_t = std::vector<neighbors_t>;
explicit worker_data(scheduler::abstract_coordinator* p) explicit worker_data(scheduler::abstract_coordinator* p)
: rengine(std::random_device{}()) : rengine(std::random_device{}())
, strategies(get_poll_strategies(p)) { , strategies(get_poll_strategies(p))
, neighborhood_level(
p->system().config().numa_aware_work_stealing_neighborhood_level) {
// nop // nop
} }
...@@ -145,7 +148,8 @@ public: ...@@ -145,7 +148,8 @@ public:
* static_cast<unsigned int>(current_node_id)]; * static_cast<unsigned int>(current_node_id)];
// iterate over all NUMA nodes and classify them in distance levels // iterate over all NUMA nodes and classify them in distance levels
// regarding to the current NUMA node // regarding to the current NUMA node
for (node_id_t x = 0; static_cast<unsigned int>(x) < num_of_dist_objs; ++x) { for (node_id_t x = 0; static_cast<unsigned int>(x) < num_of_dist_objs;
++x) {
node_set_t tmp_node_set = hwloc_bitmap_make_wrapper(); node_set_t tmp_node_set = hwloc_bitmap_make_wrapper();
hwloc_bitmap_set(tmp_node_set.get(), static_cast<unsigned int>(x)); hwloc_bitmap_set(tmp_node_set.get(), static_cast<unsigned int>(x));
auto tmp_pu_set = hwloc_bitmap_make_wrapper(); auto tmp_pu_set = hwloc_bitmap_make_wrapper();
...@@ -190,12 +194,14 @@ public: ...@@ -190,12 +194,14 @@ public:
for (auto current_lvl_it = result_matrix.begin(); for (auto current_lvl_it = result_matrix.begin();
current_lvl_it != result_matrix.end(); ++current_lvl_it) { current_lvl_it != result_matrix.end(); ++current_lvl_it) {
if (current_lvl_it != result_matrix.begin()) { if (current_lvl_it != result_matrix.begin()) {
std::copy(last_lvl_it->begin(), last_lvl_it->end(), std::back_inserter(*current_lvl_it)) ; std::copy(last_lvl_it->begin(), last_lvl_it->end(),
std::back_inserter(*current_lvl_it));
++last_lvl_it; ++last_lvl_it;
} }
} }
return result_matrix; return result_matrix;
} }
// This queue is exposed to other workers that may attempt to steal jobs // This queue is exposed to other workers that may attempt to steal jobs
// from it and the central scheduling unit can push new jobs to the queue. // from it and the central scheduling unit can push new jobs to the queue.
queue_type queue; queue_type queue;
...@@ -203,7 +209,7 @@ public: ...@@ -203,7 +209,7 @@ public:
std::default_random_engine rengine; std::default_random_engine rengine;
std::uniform_int_distribution<size_t> uniform; std::uniform_int_distribution<size_t> uniform;
std::vector<poll_strategy> strategies; std::vector<poll_strategy> strategies;
size_t neighborhood_level;
}; };
/// Create x workers. /// Create x workers.
...@@ -249,6 +255,17 @@ public: ...@@ -249,6 +255,17 @@ public:
HWLOC_CPUBIND_THREAD | HWLOC_CPUBIND_NOMEMBIND); HWLOC_CPUBIND_THREAD | HWLOC_CPUBIND_NOMEMBIND);
CALL_CAF_CRITICAL(res == -1, "hwloc_set_cpubind() failed"); CALL_CAF_CRITICAL(res == -1, "hwloc_set_cpubind() failed");
wdata.worker_matrix = wdata.init_worker_matrix(self, pu_set); wdata.worker_matrix = wdata.init_worker_matrix(self, pu_set);
auto wm_max_idx = wdata.worker_matrix.size() - 1;
if (wdata.neighborhood_level == 0) {
self->set_all_workers_are_neighbors(true);
} else if (wdata.neighborhood_level <= wm_max_idx) {
self->set_neighbors(
wdata.worker_matrix[wm_max_idx - wdata.neighborhood_level]);
self->set_all_workers_are_neighbors(false);
} else { //neighborhood_level > wm_max_idx
self->set_all_workers_are_neighbors(false);
}
} }
template <class Worker> template <class Worker>
...@@ -313,7 +330,8 @@ public: ...@@ -313,7 +330,8 @@ public:
} }
private: private:
// -- debug stuff -- // -- debug stuff --
friend std::ostream& operator <<(std::ostream& s, const hwloc_bitmap_wrapper& w); friend std::ostream& operator<<(std::ostream& s,
const hwloc_bitmap_wrapper& w);
}; };
......
...@@ -116,8 +116,8 @@ public: ...@@ -116,8 +116,8 @@ public:
/// Initalize worker thread. /// Initalize worker thread.
template <class Worker> template <class Worker>
void init_worker_thread(Worker*) { void init_worker_thread(Worker* self) {
// nop self->set_all_workers_are_neighbors(true);
} }
// Goes on a raid in quest for a shiny new job. // Goes on a raid in quest for a shiny new job.
......
...@@ -693,6 +693,9 @@ protected: ...@@ -693,6 +693,9 @@ protected:
exception_handler exception_handler_; exception_handler exception_handler_;
# endif // CAF_NO_EXCEPTIONS # endif // CAF_NO_EXCEPTIONS
/// Stores the home execution unit.
execution_unit* home_eu_;
/// @endcond /// @endcond
}; };
......
...@@ -43,11 +43,12 @@ public: ...@@ -43,11 +43,12 @@ public:
using policy_data = typename Policy::template worker_data<worker<Policy>>; using policy_data = typename Policy::template worker_data<worker<Policy>>;
worker(size_t worker_id, coordinator_ptr worker_parent, size_t throughput) worker(size_t worker_id, coordinator_ptr worker_parent, size_t throughput)
: execution_unit(&worker_parent->system()), : execution_unit(&worker_parent->system())
max_throughput_(throughput), , max_throughput_(throughput)
id_(worker_id), , id_(worker_id)
parent_(worker_parent), , parent_(worker_parent)
data_(worker_parent) { , all_workers_are_neighbors_(true)
, data_(worker_parent) {
// nop // nop
} }
...@@ -72,9 +73,13 @@ public: ...@@ -72,9 +73,13 @@ public:
/// Enqueues a new job to the worker's queue from an internal /// Enqueues a new job to the worker's queue from an internal
/// source, i.e., a job that is currently executed by this worker. /// source, i.e., a job that is currently executed by this worker.
/// @warning Must not be called from other threads. /// @warning Must not be called from other threads.
void exec_later(job_ptr job) override { void exec_later(job_ptr job, bool high_prio = true) override {
CAF_ASSERT(job != nullptr); CAF_ASSERT(job != nullptr);
policy_.internal_enqueue(this, job); CAF_LOG_TRACE(CAF_ARG(id()) << CAF_ARG(id_of(job)));
if (high_prio)
policy_.internal_enqueue(this, job);
else
policy_.external_enqueue(this, job);
} }
coordinator_ptr parent() { coordinator_ptr parent() {
...@@ -103,6 +108,25 @@ public: ...@@ -103,6 +108,25 @@ public:
return max_throughput_; return max_throughput_;
} }
bool is_neighbor(execution_unit* ptr) const {
if (all_workers_are_neighbors_)
return true;
for (auto e: neighbors_) {
if (ptr == e) {
return true;
}
}
return false;
}
void set_neighbors(const std::vector<worker*>& n) {
neighbors_= n;
}
void set_all_workers_are_neighbors(bool x) {
all_workers_are_neighbors_ = x;
}
private: private:
void run() { void run() {
CAF_SET_LOGGER_SYS(&system()); CAF_SET_LOGGER_SYS(&system());
...@@ -148,6 +172,9 @@ private: ...@@ -148,6 +172,9 @@ private:
size_t id_; size_t id_;
// pointer to central coordinator // pointer to central coordinator
coordinator_ptr parent_; coordinator_ptr parent_;
// other workers with a high memory locality
std::vector<worker*> neighbors_;
bool all_workers_are_neighbors_;
// policy-specific data // policy-specific data
policy_data data_; policy_data data_;
// instance of our policy object // instance of our policy object
......
...@@ -32,7 +32,10 @@ public: ...@@ -32,7 +32,10 @@ public:
explicit scoped_execution_unit(actor_system* sys = nullptr); explicit scoped_execution_unit(actor_system* sys = nullptr);
/// Delegates the resumable to the scheduler of `system()`. /// Delegates the resumable to the scheduler of `system()`.
void exec_later(resumable* ptr) override; void exec_later(resumable* ptr, bool high_prio = true) override;
/// It is assumed that `this` is never in the neighborhood of `ptr`.
bool is_neighbor(execution_unit* ptr) const override;
}; };
} // namespace caf } // namespace caf
......
...@@ -362,9 +362,12 @@ void abstract_coordinator::cleanup_and_release(resumable* ptr) { ...@@ -362,9 +362,12 @@ void abstract_coordinator::cleanup_and_release(resumable* ptr) {
dummy_unit(local_actor* job) : execution_unit(&job->home_system()) { dummy_unit(local_actor* job) : execution_unit(&job->home_system()) {
// nop // nop
} }
void exec_later(resumable* job) override { void exec_later(resumable* job, bool) override {
resumables.push_back(job); resumables.push_back(job);
} }
bool is_neighbor(execution_unit*) const override {
return false;
}
std::vector<resumable*> resumables; std::vector<resumable*> resumables;
}; };
switch (ptr->subtype()) { switch (ptr->subtype()) {
......
...@@ -192,10 +192,16 @@ public: ...@@ -192,10 +192,16 @@ public:
// nop // nop
} }
void exec_later(resumable*) override { void exec_later(resumable*, bool) override {
// should not happen in the first place // should not happen in the first place
CAF_LOG_ERROR("actor registry actor called exec_later during shutdown"); CAF_LOG_ERROR("actor registry actor called exec_later during shutdown");
} }
bool is_neighbor(execution_unit*) const override {
// should not happen in the first place
CAF_LOG_ERROR("actor registry actor called is_neighbor during shutdown");
return false;
}
}; };
} // namespace <anonymous> } // namespace <anonymous>
......
...@@ -114,6 +114,7 @@ actor_system_config::actor_system_config() ...@@ -114,6 +114,7 @@ actor_system_config::actor_system_config()
work_stealing_moderate_sleep_duration_us = 50; work_stealing_moderate_sleep_duration_us = 50;
work_stealing_relaxed_steal_interval = 1; work_stealing_relaxed_steal_interval = 1;
work_stealing_relaxed_sleep_duration_us = 10000; work_stealing_relaxed_sleep_duration_us = 10000;
numa_aware_work_stealing_neighborhood_level = 1;
logger_filename = "actor_log_[PID]_[TIMESTAMP]_[NODE].log"; logger_filename = "actor_log_[PID]_[TIMESTAMP]_[NODE].log";
logger_console = atom("NONE"); logger_console = atom("NONE");
middleman_network_backend = atom("default"); middleman_network_backend = atom("default");
...@@ -149,6 +150,9 @@ actor_system_config::actor_system_config() ...@@ -149,6 +150,9 @@ actor_system_config::actor_system_config()
"sets the frequency of steal attempts during relaxed polling") "sets the frequency of steal attempts during relaxed polling")
.add(work_stealing_relaxed_sleep_duration_us, "relaxed-sleep-duration", .add(work_stealing_relaxed_sleep_duration_us, "relaxed-sleep-duration",
"sets the sleep interval between poll attempts during relaxed polling"); "sets the sleep interval between poll attempts during relaxed polling");
opt_group{options_, "numa"}
.add(numa_aware_work_stealing_neighborhood_level, "neighborhood-level",
"defines the neighborhood radius (0=all, 1=next smaller group, 2=...)");
opt_group{options_, "logger"} opt_group{options_, "logger"}
.add(logger_filename, "filename", .add(logger_filename, "filename",
"sets the filesystem path of the log file") "sets the filesystem path of the log file")
......
...@@ -26,7 +26,9 @@ numa_aware_work_stealing::~numa_aware_work_stealing() { ...@@ -26,7 +26,9 @@ numa_aware_work_stealing::~numa_aware_work_stealing() {
// nop // nop
} }
std::ostream& operator<<(std::ostream& s, const numa_aware_work_stealing::hwloc_bitmap_wrapper& w) { std::ostream&
operator<<(std::ostream& s,
const numa_aware_work_stealing::hwloc_bitmap_wrapper& w) {
char* tmp = nullptr; char* tmp = nullptr;
hwloc_bitmap_asprintf(&tmp, w.get()); hwloc_bitmap_asprintf(&tmp, w.get());
s << std::string(tmp); s << std::string(tmp);
......
...@@ -96,16 +96,17 @@ error scheduled_actor::default_exception_handler(pointer ptr, ...@@ -96,16 +96,17 @@ error scheduled_actor::default_exception_handler(pointer ptr,
// -- constructors and destructors --------------------------------------------- // -- constructors and destructors ---------------------------------------------
scheduled_actor::scheduled_actor(actor_config& cfg) scheduled_actor::scheduled_actor(actor_config& cfg)
: local_actor(cfg), : local_actor(cfg)
timeout_id_(0), , timeout_id_(0)
default_handler_(print_and_drop), , default_handler_(print_and_drop)
error_handler_(default_error_handler), , error_handler_(default_error_handler)
down_handler_(default_down_handler), , down_handler_(default_down_handler)
exit_handler_(default_exit_handler), , exit_handler_(default_exit_handler)
private_thread_(nullptr) , private_thread_(nullptr)
# ifndef CAF_NO_EXCEPTIONS # ifndef CAF_NO_EXCEPTIONS
, exception_handler_(default_exception_handler) , exception_handler_(default_exception_handler)
# endif // CAF_NO_EXCEPTIONS # endif // CAF_NO_EXCEPTIONS
, home_eu_(cfg.host)
{ {
// nop // nop
} }
...@@ -135,10 +136,18 @@ void scheduled_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) { ...@@ -135,10 +136,18 @@ void scheduled_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) {
CAF_ASSERT(private_thread_ != nullptr); CAF_ASSERT(private_thread_ != nullptr);
private_thread_->resume(); private_thread_->resume();
} else { } else {
if (eu != nullptr) if (eu) {
eu->exec_later(this); // msg is received from an other scheduled actor
else if (eu == home_eu_ || eu->is_neighbor(home_eu_)) {
home_system().scheduler().enqueue(this); eu->exec_later(this, true); // internal enqueue
} else {
// `eu` has a high memory distance to this actor
home_eu_->exec_later(this, false); // external enqueued
}
} else {
// msg is received from non-actor or context or from a detached actor
home_eu_->exec_later(this, false); // external enqueue
}
} }
break; break;
} }
...@@ -564,6 +573,10 @@ bool scheduled_actor::activate(execution_unit* ctx) { ...@@ -564,6 +573,10 @@ bool scheduled_actor::activate(execution_unit* ctx) {
if (finalize()) { if (finalize()) {
CAF_LOG_DEBUG("actor_done() returned true right after make_behavior()"); CAF_LOG_DEBUG("actor_done() returned true right after make_behavior()");
return false; return false;
} else {
CAF_LOG_DEBUG("initialized actor:" << CAF_ARG(name()) << CAF_ARG(ctx));
if (home_eu_ != ctx)
home_eu_ = ctx;
} }
CAF_LOG_DEBUG("initialized actor:" << CAF_ARG(name())); CAF_LOG_DEBUG("initialized actor:" << CAF_ARG(name()));
} }
......
...@@ -29,8 +29,12 @@ scoped_execution_unit::scoped_execution_unit(actor_system* sys) ...@@ -29,8 +29,12 @@ scoped_execution_unit::scoped_execution_unit(actor_system* sys)
// nop // nop
} }
void scoped_execution_unit::exec_later(resumable* ptr) { void scoped_execution_unit::exec_later(resumable* ptr, bool) {
system().scheduler().enqueue(ptr); system().scheduler().enqueue(ptr);
} }
bool scoped_execution_unit::is_neighbor(execution_unit*) const {
return false;
}
} // namespace caf } // namespace caf
...@@ -37,10 +37,14 @@ public: ...@@ -37,10 +37,14 @@ public:
// nop // nop
} }
void exec_later(resumable* ptr) override { void exec_later(resumable* ptr, bool) override {
parent_->jobs.push_back(ptr); parent_->jobs.push_back(ptr);
} }
bool is_neighbor(execution_unit*) const override {
return false;
}
private: private:
test_coordinator* parent_; test_coordinator* parent_;
}; };
......
...@@ -292,7 +292,9 @@ public: ...@@ -292,7 +292,9 @@ public:
expected<doorman_ptr> new_tcp_doorman(uint16_t port, const char* in, expected<doorman_ptr> new_tcp_doorman(uint16_t port, const char* in,
bool reuse_addr) override; bool reuse_addr) override;
void exec_later(resumable* ptr) override; void exec_later(resumable* ptr, bool high_prio = true) override;
bool is_neighbor(execution_unit*) const override;
explicit default_multiplexer(actor_system* sys); explicit default_multiplexer(actor_system* sys);
......
...@@ -155,7 +155,8 @@ public: ...@@ -155,7 +155,8 @@ public:
void flush_runnables(); void flush_runnables();
protected: protected:
void exec_later(resumable* ptr) override; void exec_later(resumable* ptr, bool high_prio = true) override;
bool is_neighbor(execution_unit*) const override;
private: private:
using resumable_ptr = intrusive_ptr<resumable>; using resumable_ptr = intrusive_ptr<resumable>;
......
...@@ -799,7 +799,7 @@ default_multiplexer::~default_multiplexer() { ...@@ -799,7 +799,7 @@ default_multiplexer::~default_multiplexer() {
# endif # endif
} }
void default_multiplexer::exec_later(resumable* ptr) { void default_multiplexer::exec_later(resumable* ptr, bool) {
CAF_ASSERT(ptr); CAF_ASSERT(ptr);
switch (ptr->subtype()) { switch (ptr->subtype()) {
case resumable::io_actor: case resumable::io_actor:
...@@ -811,6 +811,10 @@ void default_multiplexer::exec_later(resumable* ptr) { ...@@ -811,6 +811,10 @@ void default_multiplexer::exec_later(resumable* ptr) {
} }
} }
bool default_multiplexer::is_neighbor(execution_unit*) const {
return false;
}
scribe_ptr default_multiplexer::new_scribe(native_socket fd) { scribe_ptr default_multiplexer::new_scribe(native_socket fd) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
class impl : public scribe { class impl : public scribe {
......
...@@ -521,7 +521,7 @@ void test_multiplexer::flush_runnables() { ...@@ -521,7 +521,7 @@ void test_multiplexer::flush_runnables() {
} while (!runnables.empty()); } while (!runnables.empty());
} }
void test_multiplexer::exec_later(resumable* ptr) { void test_multiplexer::exec_later(resumable* ptr, bool) {
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
switch (ptr->subtype()) { switch (ptr->subtype()) {
...@@ -539,6 +539,10 @@ void test_multiplexer::exec_later(resumable* ptr) { ...@@ -539,6 +539,10 @@ void test_multiplexer::exec_later(resumable* ptr) {
} }
} }
bool test_multiplexer::is_neighbor(execution_unit*) const {
return false;
}
void test_multiplexer::exec(resumable_ptr& ptr) { void test_multiplexer::exec(resumable_ptr& ptr) {
CAF_ASSERT(std::this_thread::get_id() == tid_); CAF_ASSERT(std::this_thread::get_id() == tid_);
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != 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