Commit 9563d041 authored by Dominik Charousset's avatar Dominik Charousset

fixed initialization order in broker

this patch fixes a bug that could cause brokers to halt
parent 63b6051c
...@@ -158,8 +158,6 @@ class broker : public extend<local_actor>:: ...@@ -158,8 +158,6 @@ class broker : public extend<local_actor>::
void enqueue(msg_hdr_cref, any_tuple, execution_unit*) override; void enqueue(msg_hdr_cref, any_tuple, execution_unit*) override;
bool initialized() const;
template<typename F> template<typename F>
static broker_ptr from(F fun) { static broker_ptr from(F fun) {
// transform to STD function here, because GCC is unable // transform to STD function here, because GCC is unable
...@@ -186,10 +184,12 @@ class broker : public extend<local_actor>:: ...@@ -186,10 +184,12 @@ class broker : public extend<local_actor>::
typedef std::unique_ptr<broker::doorman> doorman_pointer; typedef std::unique_ptr<broker::doorman> doorman_pointer;
virtual behavior make_behavior() = 0; bool initialized() const;
/** @endcond */ /** @endcond */
virtual behavior make_behavior() = 0;
private: private:
actor fork_impl(std::function<void (broker*)> fun, actor fork_impl(std::function<void (broker*)> fun,
...@@ -210,14 +210,14 @@ class broker : public extend<local_actor>:: ...@@ -210,14 +210,14 @@ class broker : public extend<local_actor>::
void erase_acceptor(int id); void erase_acceptor(int id);
void init_broker();
std::map<accept_handle, doorman_pointer> m_accept; std::map<accept_handle, doorman_pointer> m_accept;
std::map<connection_handle, scribe_pointer> m_io; std::map<connection_handle, scribe_pointer> m_io;
policy::not_prioritizing m_priority_policy; policy::not_prioritizing m_priority_policy;
policy::sequential_invoke m_invoke_policy; policy::sequential_invoke m_invoke_policy;
bool m_initialized;
}; };
} // namespace io } // namespace io
......
...@@ -58,29 +58,16 @@ constexpr size_t default_max_buffer_size = 65535; ...@@ -58,29 +58,16 @@ constexpr size_t default_max_buffer_size = 65535;
} // namespace <anonymous> } // namespace <anonymous>
class default_broker : public broker { class default_broker : public broker {
public: public:
typedef std::function<behavior (broker*)> function_type; typedef std::function<behavior (broker*)> function_type;
default_broker(function_type f) : m_fun(std::move(f)) { } default_broker(function_type f) : m_fun(std::move(f)) { }
behavior make_behavior() override { behavior make_behavior() override {
CPPA_PUSH_AID(id()); return m_fun(this);
CPPA_LOG_TRACE("");
enqueue({invalid_actor_addr, channel{this}},
make_any_tuple(atom("INITMSG")),
nullptr);
return (
on(atom("INITMSG")) >> [=] {
unbecome();
auto bhvr = m_fun(this);
if (bhvr) become(std::move(bhvr));
}
);
} }
private: private:
...@@ -399,20 +386,16 @@ void broker::enqueue(msg_hdr_cref hdr, any_tuple msg, execution_unit*) { ...@@ -399,20 +386,16 @@ void broker::enqueue(msg_hdr_cref hdr, any_tuple msg, execution_unit*) {
} }
bool broker::initialized() const { bool broker::initialized() const {
return true; return m_initialized;
} }
void broker::init_broker() { broker::broker() : m_initialized(false) {
// acquire implicit reference count held by the middleman // acquire implicit reference count held by the middleman
ref(); ref();
// actor is running now // actor is running now
get_actor_registry()->inc_running(); get_actor_registry()->inc_running();
} }
broker::broker() {
init_broker();
}
void broker::cleanup(std::uint32_t reason) { void broker::cleanup(std::uint32_t reason) {
super::cleanup(reason); super::cleanup(reason);
get_actor_registry()->dec_running(); get_actor_registry()->dec_running();
...@@ -435,10 +418,32 @@ void broker::write(const connection_handle& hdl, util::buffer&& buf) { ...@@ -435,10 +418,32 @@ void broker::write(const connection_handle& hdl, util::buffer&& buf) {
broker_ptr init_and_launch(broker_ptr ptr) { broker_ptr init_and_launch(broker_ptr ptr) {
CPPA_PUSH_AID(ptr->id()); CPPA_PUSH_AID(ptr->id());
CPPA_LOGF_TRACE("init and launch broker with id " << ptr->id()); CPPA_LOGF_TRACE("init and launch broker with id " << ptr->id());
// exec initialization code // we want to make sure initialization is executed in MM context
auto bhvr = ptr->make_behavior(); auto self = ptr.get();
if (bhvr) ptr->become(std::move(bhvr)); ptr->become(
CPPA_LOGF_WARNING_IF(!ptr->has_behavior(), "broker w/o behavior spawned"); on(atom("INITMSG")) >> [self] {
CPPA_LOGF_TRACE(CPPA_ARG(self));
self->unbecome();
auto mm = get_middleman();
// launch backend now, because user-defined initialization
// might call functions like add_connection
for (auto& kvp : self->m_io) {
CPPA_LOGF_DEBUG("launch scribe " << kvp.second.get());
mm->continue_reader(kvp.second.get());
}
for (auto& kvp : self->m_accept) {
CPPA_LOGF_DEBUG("launch doorman " << kvp.second.get());
mm->continue_reader(kvp.second.get());
}
self->m_initialized = true;
// run user-defined initialization code
auto bhvr = self->make_behavior();
if (bhvr) self->become(std::move(bhvr));
}
);
ptr->enqueue({invalid_actor_addr, ptr},
make_any_tuple(atom("INITMSG")),
nullptr);
return ptr; return ptr;
} }
...@@ -466,26 +471,29 @@ connection_handle broker::add_connection(input_stream_ptr in, output_stream_ptr ...@@ -466,26 +471,29 @@ connection_handle broker::add_connection(input_stream_ptr in, output_stream_ptr
auto id = connection_handle::from_int(in->read_handle()); auto id = connection_handle::from_int(in->read_handle());
auto ires = m_io.insert(make_pair(id, create_unique<scribe>(this, move(in), move(out)))); auto ires = m_io.insert(make_pair(id, create_unique<scribe>(this, move(in), move(out))));
CPPA_REQUIRE(ires.second == true); CPPA_REQUIRE(ires.second == true);
auto sptr = ires.first->second.get(); // 'launch' backend only if broker is already initialized
auto mm = get_middleman(); if (initialized()) {
mm->run_later([=] { CPPA_LOG_DEBUG("launch scribe " << ires.first->second.get());
// 'launch' backends auto sptr = ires.first->second.get();
auto mm = get_middleman();
mm->continue_reader(sptr); mm->continue_reader(sptr);
}); }
return id; return id;
} }
accept_handle broker::add_acceptor(acceptor_uptr ptr) { accept_handle broker::add_acceptor(acceptor_uptr ptr) {
CPPA_LOG_TRACE(CPPA_MARG(ptr, get));
using namespace std; using namespace std;
auto id = accept_handle::from_int(ptr->file_handle()); auto id = accept_handle::from_int(ptr->file_handle());
auto ires = m_accept.insert(make_pair(id, create_unique<doorman>(this, move(ptr)))); auto ires = m_accept.insert(make_pair(id, create_unique<doorman>(this, move(ptr))));
CPPA_REQUIRE(ires.second == true); CPPA_REQUIRE(ires.second == true);
auto aptr = ires.first->second.get(); // 'launch' backend only if broker is already initialized
auto mm = get_middleman(); if (initialized()) {
mm->run_later([=] { CPPA_LOG_DEBUG("launch doorman " << ires.first->second.get());
// 'launch' backends auto aptr = ires.first->second.get();
auto mm = get_middleman();
mm->continue_reader(aptr); mm->continue_reader(aptr);
}); }
return id; return id;
} }
......
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