Commit 39db309e authored by Dominik Charousset's avatar Dominik Charousset

Allow the BASP instance to offload deserialization

parent f8688545
...@@ -56,6 +56,8 @@ manual-multiplexing=false ...@@ -56,6 +56,8 @@ manual-multiplexing=false
disable-tcp=false disable-tcp=false
; enable communication via UDP ; enable communication via UDP
enable-udp=false enable-udp=false
; configures how many background workers are spawned for deserialization
max-workers=<number of cores / 4>
; when compiling with logging enabled ; when compiling with logging enabled
[logger] [logger]
......
...@@ -82,6 +82,7 @@ extern const size_t max_consecutive_reads; ...@@ -82,6 +82,7 @@ extern const size_t max_consecutive_reads;
extern const size_t heartbeat_interval; extern const size_t heartbeat_interval;
extern const size_t cached_udp_buffers; extern const size_t cached_udp_buffers;
extern const size_t max_pending_msgs; extern const size_t max_pending_msgs;
extern const size_t workers;
} // namespace middleman } // namespace middleman
......
...@@ -126,7 +126,8 @@ actor_system_config::actor_system_config() ...@@ -126,7 +126,8 @@ actor_system_config::actor_system_config()
.add<size_t>("max-pending-messages", .add<size_t>("max-pending-messages",
"maximum for reordering of UDP receive buffers (default: 10)") "maximum for reordering of UDP receive buffers (default: 10)")
.add<bool>("disable-tcp", "disables communication via TCP") .add<bool>("disable-tcp", "disables communication via TCP")
.add<bool>("enable-udp", "enable communication via UDP"); .add<bool>("enable-udp", "enable communication via UDP")
.add<size_t>("workers", "number of deserialization workers");
opt_group(custom_options_, "opencl") opt_group(custom_options_, "opencl")
.add<std::vector<size_t>>("device-ids", "whitelist for OpenCL devices"); .add<std::vector<size_t>>("device-ids", "whitelist for OpenCL devices");
opt_group(custom_options_, "openssl") opt_group(custom_options_, "openssl")
......
...@@ -92,6 +92,7 @@ const size_t max_consecutive_reads = 50; ...@@ -92,6 +92,7 @@ const size_t max_consecutive_reads = 50;
const size_t heartbeat_interval = 0; const size_t heartbeat_interval = 0;
const size_t cached_udp_buffers = 10; const size_t cached_udp_buffers = 10;
const size_t max_pending_msgs = 10; const size_t max_pending_msgs = 10;
const size_t workers = std::thread::hardware_concurrency() / 4;
} // namespace middleman } // namespace middleman
......
...@@ -27,8 +27,10 @@ ...@@ -27,8 +27,10 @@
#include "caf/io/basp/buffer_type.hpp" #include "caf/io/basp/buffer_type.hpp"
#include "caf/io/basp/connection_state.hpp" #include "caf/io/basp/connection_state.hpp"
#include "caf/io/basp/header.hpp" #include "caf/io/basp/header.hpp"
#include "caf/io/basp/message_queue.hpp"
#include "caf/io/basp/message_type.hpp" #include "caf/io/basp/message_type.hpp"
#include "caf/io/basp/routing_table.hpp" #include "caf/io/basp/routing_table.hpp"
#include "caf/io/basp/worker_hub.hpp"
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
#include "caf/variant.hpp" #include "caf/variant.hpp"
...@@ -43,13 +45,19 @@ class instance { ...@@ -43,13 +45,19 @@ class instance {
public: public:
/// Provides a callback-based interface for certain BASP events. /// Provides a callback-based interface for certain BASP events.
class callee { class callee {
protected:
using buffer_type = std::vector<char>;
public: public:
// -- member types ---------------------------------------------------------
using buffer_type = std::vector<char>;
// -- constructors, destructors, and assignment operators ------------------
explicit callee(actor_system& sys, proxy_registry::backend& backend); explicit callee(actor_system& sys, proxy_registry::backend& backend);
virtual ~callee(); virtual ~callee();
// -- pure virtual functions -----------------------------------------------
/// Called if a server handshake was received and /// Called if a server handshake was received and
/// the connection to `nid` is established. /// the connection to `nid` is established.
virtual void finalize_handshake(const node_id& nid, actor_id aid, virtual void finalize_handshake(const node_id& nid, actor_id aid,
...@@ -86,16 +94,6 @@ public: ...@@ -86,16 +94,6 @@ public:
return namespace_; return namespace_;
} }
/// Returns the hosting actor system.
actor_system& system() {
return namespace_.system();
}
/// Returns the system-wide configuration.
const actor_system_config& config() const {
return namespace_.system().config();
}
/// Returns a reference to the sent buffer. /// Returns a reference to the sent buffer.
virtual buffer_type& get_buffer(connection_handle hdl) = 0; virtual buffer_type& get_buffer(connection_handle hdl) = 0;
...@@ -208,7 +206,11 @@ public: ...@@ -208,7 +206,11 @@ public:
} }
actor_system& system() { actor_system& system() {
return callee_.system(); return callee_.proxies().system();
}
const actor_system_config& config() {
return system().config();
} }
bool handle(execution_unit* ctx, connection_handle hdl, header& hdr, bool handle(execution_unit* ctx, connection_handle hdl, header& hdr,
...@@ -222,6 +224,8 @@ private: ...@@ -222,6 +224,8 @@ private:
published_actor_map published_actors_; published_actor_map published_actors_;
node_id this_node_; node_id this_node_;
callee& callee_; callee& callee_;
message_queue queue_;
worker_hub hub_;
}; };
/// @} /// @}
......
...@@ -24,6 +24,7 @@ ...@@ -24,6 +24,7 @@
#include "caf/actor_proxy.hpp" #include "caf/actor_proxy.hpp"
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/detail/sync_request_bouncer.hpp" #include "caf/detail/sync_request_bouncer.hpp"
#include "caf/execution_unit.hpp" #include "caf/execution_unit.hpp"
#include "caf/io/basp/header.hpp" #include "caf/io/basp/header.hpp"
...@@ -50,6 +51,9 @@ public: ...@@ -50,6 +51,9 @@ public:
message msg; message msg;
auto mid = make_message_id(dref.hdr_.operation_data); auto mid = make_message_id(dref.hdr_.operation_data);
binary_deserializer source{ctx, dref.payload_}; binary_deserializer source{ctx, dref.payload_};
// Make sure to drop the message in case we return abnormally.
auto guard = detail::make_scope_guard(
[&] { dref.queue_->drop(ctx, dref.msg_id_); });
// Registry setup. // Registry setup.
dref.proxies_->set_last_hop(&dref.last_hop_); dref.proxies_->set_last_hop(&dref.last_hop_);
// Get the local receiver. // Get the local receiver.
...@@ -117,9 +121,10 @@ public: ...@@ -117,9 +121,10 @@ public:
} }
} }
// Ship the message. // Ship the message.
dst->enqueue(make_mailbox_element(std::move(src), mid, std::move(stages), guard.disable();
std::move(msg)), dref.queue_->push(ctx, dref.msg_id_, std::move(dst),
ctx); make_mailbox_element(std::move(src), mid,
std::move(stages), std::move(msg)));
} }
}; };
......
...@@ -24,6 +24,8 @@ ...@@ -24,6 +24,8 @@
#include "caf/defaults.hpp" #include "caf/defaults.hpp"
#include "caf/io/basp/remote_message_handler.hpp" #include "caf/io/basp/remote_message_handler.hpp"
#include "caf/io/basp/version.hpp" #include "caf/io/basp/version.hpp"
#include "caf/io/basp/worker.hpp"
#include "caf/settings.hpp"
#include "caf/streambuf.hpp" #include "caf/streambuf.hpp"
namespace caf { namespace caf {
...@@ -44,6 +46,10 @@ instance::instance(abstract_broker* parent, callee& lstnr) ...@@ -44,6 +46,10 @@ instance::instance(abstract_broker* parent, callee& lstnr)
this_node_(parent->system().node()), this_node_(parent->system().node()),
callee_(lstnr) { callee_(lstnr) {
CAF_ASSERT(this_node_ != none); CAF_ASSERT(this_node_ != none);
auto workers = get_or(config(), "middleman.workers",
defaults::middleman::workers);
for (size_t i = 0; i < workers; ++i)
hub_.push_new_worker(queue_, proxies());
} }
connection_state instance::handle(execution_unit* ctx, connection_state instance::handle(execution_unit* ctx,
...@@ -219,7 +225,7 @@ void instance::write_server_handshake(execution_unit* ctx, buffer_type& out_buf, ...@@ -219,7 +225,7 @@ void instance::write_server_handshake(execution_unit* ctx, buffer_type& out_buf,
} }
CAF_LOG_DEBUG_IF(!pa && port, "no actor published"); CAF_LOG_DEBUG_IF(!pa && port, "no actor published");
auto writer = make_callback([&](serializer& sink) -> error { auto writer = make_callback([&](serializer& sink) -> error {
auto app_ids = get_or(callee_.config(), "middleman.app-identifiers", auto app_ids = get_or(config(), "middleman.app-identifiers",
defaults::middleman::app_identifiers); defaults::middleman::app_identifiers);
auto aid = invalid_actor_id; auto aid = invalid_actor_id;
auto iface = std::set<std::string>{}; auto iface = std::set<std::string>{};
...@@ -299,7 +305,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -299,7 +305,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
return false; return false;
} }
// Check the application ID. // Check the application ID.
auto whitelist = get_or(callee_.config(), "middleman.app-identifiers", auto whitelist = get_or(config(), "middleman.app-identifiers",
defaults::middleman::app_identifiers); defaults::middleman::app_identifiers);
auto i = std::find_first_of(app_ids.begin(), app_ids.end(), auto i = std::find_first_of(app_ids.begin(), app_ids.end(),
whitelist.begin(), whitelist.end()); whitelist.begin(), whitelist.end());
...@@ -382,24 +388,40 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -382,24 +388,40 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
} }
// fall through // fall through
case message_type::direct_message: { case message_type::direct_message: {
struct handler : remote_message_handler<handler> { auto worker = hub_.pop();
handler(proxy_registry* proxies, actor_system* system, node_id last_hop, auto last_hop = tbl_.lookup_direct(hdl);
basp::header& hdr, buffer_type& payload) if (worker != nullptr) {
: proxies_(proxies), CAF_LOG_DEBUG("launch BASP worker for deserializing a"
system_(system), << hdr.operation);
last_hop_(std::move(last_hop)), worker->launch(last_hop, hdr, *payload);
hdr_(hdr), } else {
payload_(payload) { CAF_LOG_DEBUG("out of BASP workers, continue deserializing a"
// nop << hdr.operation);
} // If no worker is available then we have no other choice than to take
proxy_registry* proxies_; // the performance hit and deserialize in this thread.
actor_system* system_; struct handler : remote_message_handler<handler> {
node_id last_hop_; handler(message_queue* queue, proxy_registry* proxies,
basp::header& hdr_; actor_system* system, node_id last_hop, basp::header& hdr,
buffer_type& payload_; buffer_type& payload)
}; : queue_(queue),
handler f{&proxies(), &system(), tbl_.lookup_direct(hdl), hdr, *payload}; proxies_(proxies),
f.handle_remote_message(callee_.current_execution_unit()); system_(system),
last_hop_(std::move(last_hop)),
hdr_(hdr),
payload_(payload) {
msg_id_ = queue_->new_id();
}
message_queue* queue_;
proxy_registry* proxies_;
actor_system* system_;
node_id last_hop_;
basp::header& hdr_;
buffer_type& payload_;
uint64_t msg_id_;
};
handler f{&queue_, &proxies(), &system(), last_hop, hdr, *payload};
f.handle_remote_message(callee_.current_execution_unit());
}
break; break;
} }
case message_type::monitor_message: { case message_type::monitor_message: {
......
...@@ -111,11 +111,12 @@ struct node { ...@@ -111,11 +111,12 @@ struct node {
class fixture { class fixture {
public: public:
fixture(bool autoconn = false) fixture(bool autoconn = false)
: sys(cfg.load<io::middleman, network::test_multiplexer>() : sys(cfg.load<io::middleman, network::test_multiplexer>()
.set("middleman.enable-automatic-connections", autoconn) .set("middleman.enable-automatic-connections", autoconn)
.set("scheduler.policy", autoconn ? caf::atom("testing") .set("middleman.workers", size_t{0})
: caf::atom("stealing")) .set("scheduler.policy",
.set("middleman.attach-utility-actors", autoconn)) { autoconn ? caf::atom("testing") : caf::atom("stealing"))
.set("middleman.attach-utility-actors", autoconn)) {
auto& mm = sys.middleman(); auto& mm = sys.middleman();
mpx_ = dynamic_cast<network::test_multiplexer*>(&mm.backend()); mpx_ = dynamic_cast<network::test_multiplexer*>(&mm.backend());
CAF_REQUIRE(mpx_ != nullptr); CAF_REQUIRE(mpx_ != nullptr);
......
...@@ -563,6 +563,7 @@ public: ...@@ -563,6 +563,7 @@ public:
cfg.set("scheduler.policy", caf::atom("testing")); cfg.set("scheduler.policy", caf::atom("testing"));
cfg.set("logger.inline-output", true); cfg.set("logger.inline-output", true);
cfg.set("middleman.network-backend", caf::atom("testing")); cfg.set("middleman.network-backend", caf::atom("testing"));
cfg.set("middleman.workers", size_t{0});
return cfg; return cfg;
} }
......
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