Commit b56f485f authored by Jakob Otto's avatar Jakob Otto

Use basp::worker for serializing

parent 58cfcbd2
...@@ -28,11 +28,14 @@ ...@@ -28,11 +28,14 @@
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/callback.hpp" #include "caf/callback.hpp"
#include "caf/detail/worker_hub.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/net/basp/connection_state.hpp" #include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp" #include "caf/net/basp/constants.hpp"
#include "caf/net/basp/header.hpp" #include "caf/net/basp/header.hpp"
#include "caf/net/basp/message_queue.hpp"
#include "caf/net/basp/message_type.hpp" #include "caf/net/basp/message_type.hpp"
#include "caf/net/basp/worker.hpp"
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
#include "caf/net/packet_writer.hpp" #include "caf/net/packet_writer.hpp"
#include "caf/net/receive_policy.hpp" #include "caf/net/receive_policy.hpp"
...@@ -57,6 +60,8 @@ public: ...@@ -57,6 +60,8 @@ public:
using byte_span = span<const byte>; using byte_span = span<const byte>;
using hub_type = detail::worker_hub<worker>;
struct test_tag {}; struct test_tag {};
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
...@@ -193,6 +198,10 @@ private: ...@@ -193,6 +198,10 @@ private:
/// Provides pointers to the actor system as well as the registry, /// Provides pointers to the actor system as well as the registry,
/// serializers and deserializer. /// serializers and deserializer.
scoped_execution_unit executor_; scoped_execution_unit executor_;
std::unique_ptr<message_queue> queue_;
std::unique_ptr<hub_type> hub_;
}; };
} // namespace basp } // namespace basp
......
...@@ -77,5 +77,5 @@ public: ...@@ -77,5 +77,5 @@ public:
}; };
} // namespace basp } // namespace basp
} // namespace io } // namespace net
} // namespace caf } // namespace caf
...@@ -50,7 +50,7 @@ public: ...@@ -50,7 +50,7 @@ public:
using scheduler_type = scheduler::abstract_coordinator; using scheduler_type = scheduler::abstract_coordinator;
using buffer_type = std::vector<char>; using buffer_type = std::vector<byte>;
using hub_type = detail::worker_hub<worker>; using hub_type = detail::worker_hub<worker>;
......
...@@ -44,7 +44,10 @@ namespace caf { ...@@ -44,7 +44,10 @@ namespace caf {
namespace net { namespace net {
namespace basp { namespace basp {
application::application(proxy_registry& proxies) : proxies_(proxies) { application::application(proxy_registry& proxies)
: proxies_(proxies),
queue_{std::unique_ptr<message_queue>{new message_queue}},
hub_{std::unique_ptr<hub_type>{new hub_type}} {
// nop // nop
} }
...@@ -248,34 +251,39 @@ error application::handle_handshake(packet_writer&, header hdr, ...@@ -248,34 +251,39 @@ error application::handle_handshake(packet_writer&, header hdr,
error application::handle_actor_message(packet_writer&, header hdr, error application::handle_actor_message(packet_writer&, header hdr,
byte_span payload) { byte_span payload) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size())); auto worker = hub_->pop();
// Deserialize payload. buffer_type buf(payload.begin(), payload.end());
actor_id src_id = 0; if (worker != nullptr) {
node_id src_node; CAF_LOG_DEBUG("launch BASP worker for deserializing an actor_message");
actor_id dst_id = 0; worker->launch(node_id{}, hdr, buf);
std::vector<strong_actor_ptr> fwd_stack; } else {
message content; CAF_LOG_DEBUG(
binary_deserializer source{&executor_, payload}; "out of BASP workers, continue deserializing an actor_message");
if (auto err = source(src_node, src_id, dst_id, fwd_stack, content)) // If no worker is available then we have no other choice than to take
return err; // the performance hit and deserialize in this thread.
// Sanity checks. struct handler : remote_message_handler<handler> {
if (dst_id == 0) handler(message_queue* queue, proxy_registry* proxies,
return ec::invalid_payload; actor_system* system, node_id last_hop, basp::header& hdr,
// Try to fetch the receiver. buffer_type& payload)
auto dst_hdl = system().registry().get(dst_id); : queue_(queue),
if (dst_hdl == nullptr) { proxies_(proxies),
CAF_LOG_DEBUG("no actor found for given ID, drop message"); system_(system),
return caf::none; 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_.get(), &proxies_, system_, node_id{}, hdr, buf};
f.handle_remote_message(&executor_);
} }
// Try to fetch the sender.
strong_actor_ptr src_hdl;
if (src_node != none && src_id != 0)
src_hdl = proxies_.get_or_put(src_node, src_id);
// Ship the message.
auto ptr = make_mailbox_element(std::move(src_hdl),
make_message_id(hdr.operation_data),
std::move(fwd_stack), std::move(content));
dst_hdl->get()->enqueue(std::move(ptr), nullptr);
return none; return none;
} }
......
...@@ -75,9 +75,9 @@ test::peer_entry& test::emplace(const node_id& peer_id, stream_socket first, ...@@ -75,9 +75,9 @@ test::peer_entry& test::emplace(const node_id& peer_id, stream_socket first,
using transport_type = stream_transport<basp::application>; using transport_type = stream_transport<basp::application>;
nonblocking(second, true); nonblocking(second, true);
auto mpx = mm_.mpx(); auto mpx = mm_.mpx();
basp::application app{proxies_};
auto mgr = make_endpoint_manager(mpx, mm_.system(), auto mgr = make_endpoint_manager(mpx, mm_.system(),
transport_type{second, transport_type{second, std::move(app)});
basp::application{proxies_}});
if (auto err = mgr->init()) { if (auto err = mgr->init()) {
CAF_LOG_ERROR("mgr->init() failed: " << mm_.system().render(err)); CAF_LOG_ERROR("mgr->init() failed: " << mm_.system().render(err));
CAF_RAISE_ERROR("mgr->init() failed"); CAF_RAISE_ERROR("mgr->init() failed");
......
...@@ -19,6 +19,7 @@ ...@@ -19,6 +19,7 @@
#include "caf/net/basp/worker.hpp" #include "caf/net/basp/worker.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/byte.hpp"
#include "caf/net/basp/message_queue.hpp" #include "caf/net/basp/message_queue.hpp"
#include "caf/proxy_registry.hpp" #include "caf/proxy_registry.hpp"
#include "caf/scheduler/abstract_coordinator.hpp" #include "caf/scheduler/abstract_coordinator.hpp"
...@@ -60,5 +61,5 @@ resumable::resume_result worker::resume(execution_unit* ctx, size_t) { ...@@ -60,5 +61,5 @@ resumable::resume_result worker::resume(execution_unit* ctx, size_t) {
} }
} // namespace basp } // namespace basp
} // namespace io } // namespace net
} // namespace caf } // namespace caf
...@@ -25,10 +25,10 @@ ...@@ -25,10 +25,10 @@
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/make_actor.hpp" #include "caf/make_actor.hpp"
#include "caf/net/basp/message_queue.hpp" #include "caf/net/basp/message_queue.hpp"
#include "caf/proxy_registry.hpp" #include "caf/proxy_registry.hpp"
#include "caf/serializer_impl.hpp"
using namespace caf; using namespace caf;
...@@ -105,9 +105,9 @@ CAF_TEST(deliver serialized message) { ...@@ -105,9 +105,9 @@ CAF_TEST(deliver serialized message) {
CAF_REQUIRE_NOT_EQUAL(hub.peek(), nullptr); CAF_REQUIRE_NOT_EQUAL(hub.peek(), nullptr);
auto w = hub.pop(); auto w = hub.pop();
CAF_MESSAGE("create a fake message + BASP header"); CAF_MESSAGE("create a fake message + BASP header");
std::vector<char> payload; std::vector<byte> payload;
std::vector<strong_actor_ptr> stages; std::vector<strong_actor_ptr> stages;
binary_serializer sink{sys, payload}; serializer_impl<std::vector<byte>> sink{sys, payload};
if (auto err = sink(node_id{}, self->id(), testee.id(), stages, if (auto err = sink(node_id{}, self->id(), testee.id(), stages,
make_message(ok_atom::value))) make_message(ok_atom::value)))
CAF_FAIL("unable to serialize message: " << sys.render(err)); CAF_FAIL("unable to serialize message: " << sys.render(err));
......
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