Commit b57ead03 authored by Dominik Charousset's avatar Dominik Charousset

Make sure brokers stick to one backend

parent 065ad856
......@@ -365,7 +365,9 @@ public:
std::vector<connection_handle> connections() const;
/// Returns the `multiplexer` running this broker.
network::multiplexer& backend();
network::multiplexer& backend() {
return *backend_;
}
protected:
void init_broker();
......@@ -436,6 +438,7 @@ private:
get_map(hdl).emplace(hdl, std::move(ptr));
}
network::multiplexer* backend_ = nullptr;
scribe_map scribes_;
doorman_map doormen_;
datagram_servant_map datagram_servants_;
......
......@@ -18,39 +18,36 @@
#include "caf/actor_system.hpp"
#include "caf/config.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/network/multiplexer.hpp"
#include "caf/logger.hpp"
#include "caf/make_counted.hpp"
#include "caf/none.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/span.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/event_based_actor.hpp"
namespace caf::io {
void abstract_broker::enqueue(strong_actor_ptr src, message_id mid, message msg,
execution_unit*) {
enqueue(make_mailbox_element(std::move(src), mid, {}, std::move(msg)),
&backend());
backend_);
}
void abstract_broker::enqueue(mailbox_element_ptr ptr, execution_unit*) {
CAF_PUSH_AID(id());
scheduled_actor::enqueue(std::move(ptr), &backend());
scheduled_actor::enqueue(std::move(ptr), backend_);
}
void abstract_broker::launch(execution_unit* eu, bool lazy, bool hide) {
CAF_PUSH_AID_FROM_PTR(this);
CAF_ASSERT(eu != nullptr);
CAF_ASSERT(eu == &backend());
CAF_ASSERT(dynamic_cast<network::multiplexer*>(eu) != nullptr);
backend_ = static_cast<network::multiplexer*>(eu);
CAF_LOG_TRACE(CAF_ARG(lazy) << CAF_ARG(hide));
// add implicit reference count held by middleman/multiplexer
if (!hide)
register_at_system();
if (lazy && mailbox().try_block())
......@@ -359,7 +356,7 @@ resumable::subtype_t abstract_broker::subtype() const {
resumable::resume_result
abstract_broker::resume(execution_unit* ctx, size_t mt) {
CAF_ASSERT(ctx != nullptr);
CAF_ASSERT(ctx == &backend());
CAF_ASSERT(ctx == backend_);
return scheduled_actor::resume(ctx, mt);
}
......@@ -380,10 +377,6 @@ abstract_broker::abstract_broker(actor_config& cfg) : scheduled_actor(cfg) {
// nop
}
network::multiplexer& abstract_broker::backend() {
return system().middleman().backend();
}
void abstract_broker::launch_servant(doorman_ptr& ptr) {
// A doorman needs to be launched in addition to being initialized. This
// allows CAF to assign doorman to uninitialized brokers.
......
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