Commit f4dfa77f authored by Dominik Charousset's avatar Dominik Charousset

Adhere to coding conventions

parent 93f87bf4
...@@ -18,17 +18,17 @@ ...@@ -18,17 +18,17 @@
#pragma once #pragma once
#include <mutex>
#include <utility>
#include <functional> #include <functional>
#include <mutex>
#include <unordered_map> #include <unordered_map>
#include <utility>
#include "caf/fwd.hpp"
#include "caf/node_id.hpp"
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/actor_proxy.hpp" #include "caf/actor_proxy.hpp"
#include "caf/exit_reason.hpp" #include "caf/exit_reason.hpp"
#include "caf/fwd.hpp"
#include "caf/node_id.hpp"
namespace caf { namespace caf {
......
...@@ -16,18 +16,18 @@ ...@@ -16,18 +16,18 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include <utility>
#include <algorithm> #include <algorithm>
#include <utility>
#include "caf/node_id.hpp"
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
#include "caf/serializer.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/deserializer.hpp" #include "caf/deserializer.hpp"
#include "caf/node_id.hpp"
#include "caf/proxy_registry.hpp" #include "caf/proxy_registry.hpp"
#include "caf/serializer.hpp"
#include "caf/logger.hpp"
#include "caf/actor_registry.hpp" #include "caf/actor_registry.hpp"
#include "caf/logger.hpp"
namespace caf { namespace caf {
...@@ -36,8 +36,7 @@ proxy_registry::backend::~backend() { ...@@ -36,8 +36,7 @@ proxy_registry::backend::~backend() {
} }
proxy_registry::proxy_registry(actor_system& sys, backend& be) proxy_registry::proxy_registry(actor_system& sys, backend& be)
: system_(sys), : system_(sys), backend_(be) {
backend_(be) {
// nop // nop
} }
...@@ -46,7 +45,7 @@ proxy_registry::~proxy_registry() { ...@@ -46,7 +45,7 @@ proxy_registry::~proxy_registry() {
} }
size_t proxy_registry::count_proxies(const node_id& node) const { size_t proxy_registry::count_proxies(const node_id& node) const {
std::unique_lock<std::mutex> guard {mtx_}; std::unique_lock<std::mutex> guard{mtx_};
auto i = proxies_.find(node); auto i = proxies_.find(node);
return i != proxies_.end() ? i->second.size() : 0; return i != proxies_.end() ? i->second.size() : 0;
} }
......
...@@ -121,6 +121,7 @@ behavior basp_broker::make_behavior() { ...@@ -121,6 +121,7 @@ behavior basp_broker::make_behavior() {
send(this, tick_atom::value, heartbeat_interval); send(this, tick_atom::value, heartbeat_interval);
} }
return { return {
// received from underlying broker implementation // received from underlying broker implementation
[=](new_data_msg& msg) { [=](new_data_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle)); CAF_LOG_TRACE(CAF_ARG(msg.handle));
...@@ -144,11 +145,11 @@ behavior basp_broker::make_behavior() { ...@@ -144,11 +145,11 @@ behavior basp_broker::make_behavior() {
[=](forward_atom, strong_actor_ptr& src, [=](forward_atom, strong_actor_ptr& src,
const std::vector<strong_actor_ptr>& fwd_stack, strong_actor_ptr& dest, const std::vector<strong_actor_ptr>& fwd_stack, strong_actor_ptr& dest,
message_id mid, const message& msg) { message_id mid, const message& msg) {
CAF_LOG_TRACE(CAF_ARG(src) << CAF_ARG(dest) CAF_LOG_TRACE(CAF_ARG(src)
<< CAF_ARG(mid) << CAF_ARG(msg)); << CAF_ARG(dest) << CAF_ARG(mid) << CAF_ARG(msg));
if (!dest || system().node() == dest->node()) { if (!dest || system().node() == dest->node()) {
CAF_LOG_WARNING("cannot forward to invalid or local actor:" CAF_LOG_WARNING(
<< CAF_ARG(dest)); "cannot forward to invalid or local actor:" << CAF_ARG(dest));
return; return;
} }
if (src && system().node() == src->node()) if (src && system().node() == src->node())
...@@ -167,14 +168,13 @@ behavior basp_broker::make_behavior() { ...@@ -167,14 +168,13 @@ behavior basp_broker::make_behavior() {
if (cme == nullptr || cme->sender == nullptr) if (cme == nullptr || cme->sender == nullptr)
return sec::invalid_argument; return sec::invalid_argument;
CAF_LOG_TRACE(CAF_ARG2("sender", cme->sender) CAF_LOG_TRACE(CAF_ARG2("sender", cme->sender)
<< ", " << CAF_ARG(dest_node) << ", " << CAF_ARG(dest_node) << ", " << CAF_ARG(dest_name)
<< ", " << CAF_ARG(dest_name)
<< ", " << CAF_ARG(msg)); << ", " << CAF_ARG(msg));
auto& sender = cme->sender; auto& sender = cme->sender;
if (system().node() == sender->node()) if (system().node() == sender->node())
system().registry().put(sender->id(), sender); system().registry().put(sender->id(), sender);
if (!instance.dispatch(context(), sender, cme->stages, if (!instance.dispatch(context(), sender, cme->stages, dest_node,
dest_node, static_cast<uint64_t>(dest_name), static_cast<uint64_t>(dest_name),
basp::header::named_receiver_flag, cme->mid, basp::header::named_receiver_flag, cme->mid,
msg)) { msg)) {
detail::sync_request_bouncer srb{exit_reason::remote_link_unreachable}; detail::sync_request_bouncer srb{exit_reason::remote_link_unreachable};
...@@ -198,8 +198,8 @@ behavior basp_broker::make_behavior() { ...@@ -198,8 +198,8 @@ behavior basp_broker::make_behavior() {
CAF_LOG_DEBUG("write monitor_message:" << CAF_ARG(proxy)); CAF_LOG_DEBUG("write monitor_message:" << CAF_ARG(proxy));
// tell remote side we are monitoring this actor now // tell remote side we are monitoring this actor now
auto hdl = route->hdl; auto hdl = route->hdl;
instance.write_monitor_message(context(), get_buffer(hdl), instance.write_monitor_message(context(), get_buffer(hdl), proxy->node(),
proxy->node(), proxy->id()); proxy->id());
flush(hdl); flush(hdl);
}, },
// received from underlying broker implementation // received from underlying broker implementation
...@@ -214,11 +214,11 @@ behavior basp_broker::make_behavior() { ...@@ -214,11 +214,11 @@ behavior basp_broker::make_behavior() {
// received from underlying broker implementation // received from underlying broker implementation
[=](const connection_closed_msg& msg) { [=](const connection_closed_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle)); CAF_LOG_TRACE(CAF_ARG(msg.handle));
// We might still have pending messages from this connection. To make // We might still have pending messages from this connection. To
// sure there's no BASP worker deserializing a message, we are sending // make sure there's no BASP worker deserializing a message, we are
// us a message through the queue. This message gets delivered only // sending us a message through the queue. This message gets
// after all received messages up to this point were deserialized // delivered only after all received messages up to this point were
// and delivered. // deserialized and delivered.
auto& q = instance.queue(); auto& q = instance.queue();
auto msg_id = q.new_id(); auto msg_id = q.new_id();
q.push(context(), msg_id, ctrl(), q.push(context(), msg_id, ctrl(),
...@@ -245,8 +245,8 @@ behavior basp_broker::make_behavior() { ...@@ -245,8 +245,8 @@ behavior basp_broker::make_behavior() {
// received from middleman actor // received from middleman actor
[=](publish_atom, doorman_ptr& ptr, uint16_t port, [=](publish_atom, doorman_ptr& ptr, uint16_t port,
const strong_actor_ptr& whom, std::set<std::string>& sigs) { const strong_actor_ptr& whom, std::set<std::string>& sigs) {
CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(port) CAF_LOG_TRACE(CAF_ARG(ptr)
<< CAF_ARG(whom) << CAF_ARG(sigs)); << CAF_ARG(port) << CAF_ARG(whom) << CAF_ARG(sigs));
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
add_doorman(std::move(ptr)); add_doorman(std::move(ptr));
if (whom) if (whom)
...@@ -284,8 +284,7 @@ behavior basp_broker::make_behavior() { ...@@ -284,8 +284,7 @@ behavior basp_broker::make_behavior() {
[&](const strong_actor_ptr&, uint16_t x) -> error { [&](const strong_actor_ptr&, uint16_t x) -> error {
close(hdl_by_port(x)); close(hdl_by_port(x));
return none; return none;
} });
);
if (instance.remove_published_actor(whom, port, &cb) == 0) if (instance.remove_published_actor(whom, port, &cb) == 0)
return sec::no_actor_published_at_port; return sec::no_actor_published_at_port;
return unit; return unit;
...@@ -293,8 +292,8 @@ behavior basp_broker::make_behavior() { ...@@ -293,8 +292,8 @@ behavior basp_broker::make_behavior() {
[=](close_atom, uint16_t port) -> result<void> { [=](close_atom, uint16_t port) -> result<void> {
if (port == 0) if (port == 0)
return sec::cannot_close_invalid_port; return sec::cannot_close_invalid_port;
// It is well-defined behavior to not have an actor published here, hence // It is well-defined behavior to not have an actor published here,
// the result can be ignored safely. // hence the result can be ignored safely.
instance.remove_published_actor(port, nullptr); instance.remove_published_actor(port, nullptr);
auto res = close(hdl_by_port(port)); auto res = close(hdl_by_port(port));
if (res) if (res)
...@@ -314,8 +313,8 @@ behavior basp_broker::make_behavior() { ...@@ -314,8 +313,8 @@ behavior basp_broker::make_behavior() {
}, },
[=](tick_atom, size_t interval) { [=](tick_atom, size_t interval) {
instance.handle_heartbeat(context()); instance.handle_heartbeat(context());
delayed_send(this, std::chrono::milliseconds{interval}, delayed_send(this, std::chrono::milliseconds{interval}, tick_atom::value,
tick_atom::value, interval); interval);
}}; }};
} }
...@@ -325,9 +324,8 @@ proxy_registry* basp_broker::proxy_registry_ptr() { ...@@ -325,9 +324,8 @@ proxy_registry* basp_broker::proxy_registry_ptr() {
resumable::resume_result basp_broker::resume(execution_unit* ctx, size_t mt) { resumable::resume_result basp_broker::resume(execution_unit* ctx, size_t mt) {
ctx->proxy_registry_ptr(&instance.proxies()); ctx->proxy_registry_ptr(&instance.proxies());
auto guard = detail::make_scope_guard([=] { auto guard = detail::make_scope_guard(
ctx->proxy_registry_ptr(nullptr); [=] { ctx->proxy_registry_ptr(nullptr); });
});
return super::resume(ctx, mt); return super::resume(ctx, mt);
} }
...@@ -348,8 +346,9 @@ strong_actor_ptr basp_broker::make_proxy(node_id nid, actor_id aid) { ...@@ -348,8 +346,9 @@ strong_actor_ptr basp_broker::make_proxy(node_id nid, actor_id aid) {
// create proxy and add functor that will be called if we // create proxy and add functor that will be called if we
// receive a basp::down_message // receive a basp::down_message
actor_config cfg; actor_config cfg;
auto res = make_actor<forwarding_actor_proxy, strong_actor_ptr>( auto res = make_actor<forwarding_actor_proxy, strong_actor_ptr>(aid, nid,
aid, nid, &(system()), cfg, this); &(system()),
cfg, this);
strong_actor_ptr selfptr{ctrl()}; strong_actor_ptr selfptr{ctrl()};
res->get()->attach_functor([=](const error& rsn) { res->get()->attach_functor([=](const error& rsn) {
mm->backend().post([=] { mm->backend().post([=] {
...@@ -406,8 +405,8 @@ void basp_broker::send_basp_down_message(const node_id& nid, actor_id aid, ...@@ -406,8 +405,8 @@ void basp_broker::send_basp_down_message(const node_id& nid, actor_id aid,
CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid) << CAF_ARG(rsn)); CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid) << CAF_ARG(rsn));
auto path = instance.tbl().lookup(nid); auto path = instance.tbl().lookup(nid);
if (!path) { if (!path) {
CAF_LOG_INFO("cannot send exit message for proxy, no route to host:" CAF_LOG_INFO(
<< CAF_ARG(nid)); "cannot send exit message for proxy, no route to host:" << CAF_ARG(nid));
return; return;
} }
instance.write_down_message(context(), get_buffer(path->hdl), nid, aid, rsn); instance.write_down_message(context(), get_buffer(path->hdl), nid, aid, rsn);
...@@ -460,6 +459,7 @@ void basp_broker::learned_new_node(const node_id& nid) { ...@@ -460,6 +459,7 @@ void basp_broker::learned_new_node(const node_id& nid) {
// skip messages until we receive the initial ok_atom // skip messages until we receive the initial ok_atom
tself->set_default_handler(skip); tself->set_default_handler(skip);
return { return {
[=](ok_atom, const std::string& /* key == "info" */, [=](ok_atom, const std::string& /* key == "info" */,
const strong_actor_ptr& config_serv, const std::string& /* name */) { const strong_actor_ptr& config_serv, const std::string& /* name */) {
CAF_LOG_TRACE(CAF_ARG(config_serv)); CAF_LOG_TRACE(CAF_ARG(config_serv));
...@@ -468,22 +468,19 @@ void basp_broker::learned_new_node(const node_id& nid) { ...@@ -468,22 +468,19 @@ void basp_broker::learned_new_node(const node_id& nid) {
if (!config_serv) if (!config_serv)
return; return;
tself->monitor(config_serv); tself->monitor(config_serv);
tself->become( tself->become([=](spawn_atom, std::string& type, message& args)
[=](spawn_atom, std::string& type, message& args)
-> delegated<strong_actor_ptr, std::set<std::string>> { -> delegated<strong_actor_ptr, std::set<std::string>> {
CAF_LOG_TRACE(CAF_ARG(type) << CAF_ARG(args)); CAF_LOG_TRACE(CAF_ARG(type) << CAF_ARG(args));
tself->delegate(actor_cast<actor>(std::move(config_serv)), tself->delegate(actor_cast<actor>(std::move(config_serv)),
get_atom::value, std::move(type), get_atom::value, std::move(type), std::move(args));
std::move(args));
return {}; return {};
} });
);
}, },
after(std::chrono::minutes(5)) >> [=] { after(std::chrono::minutes(5)) >>
[=] {
CAF_LOG_INFO("no spawn server found:" << CAF_ARG(nid)); CAF_LOG_INFO("no spawn server found:" << CAF_ARG(nid));
tself->quit(); tself->quit();
} }};
};
}); });
spawn_servers.emplace(nid, tmp); spawn_servers.emplace(nid, tmp);
using namespace detail; using namespace detail;
...@@ -536,8 +533,12 @@ void basp_broker::set_context(connection_handle hdl) { ...@@ -536,8 +533,12 @@ void basp_broker::set_context(connection_handle hdl) {
auto i = ctx.find(hdl); auto i = ctx.find(hdl);
if (i == ctx.end()) { if (i == ctx.end()) {
CAF_LOG_DEBUG("create new BASP context:" << CAF_ARG(hdl)); CAF_LOG_DEBUG("create new BASP context:" << CAF_ARG(hdl));
basp::header hdr{basp::message_type::server_handshake, 0, 0, 0, basp::header hdr{basp::message_type::server_handshake,
invalid_actor_id, invalid_actor_id}; 0,
0,
0,
invalid_actor_id,
invalid_actor_id};
i = ctx i = ctx
.emplace(hdl, basp::endpoint_context{basp::await_header, hdr, hdl, .emplace(hdl, basp::endpoint_context{basp::await_header, hdr, hdl,
none, 0, 0, none}) none, 0, 0, none})
......
...@@ -454,9 +454,10 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -454,9 +454,10 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if (dest_node == this_node_) { if (dest_node == this_node_) {
// Delay this message to make sure we don't skip in-flight messages. // Delay this message to make sure we don't skip in-flight messages.
auto msg_id = queue_.new_id(); auto msg_id = queue_.new_id();
auto ptr = make_mailbox_element( auto ptr = make_mailbox_element(nullptr, make_message_id(), {},
nullptr, make_message_id(), {}, delete_atom::value, source_node, delete_atom::value, source_node,
hdr.source_actor, std::move(fail_state)); hdr.source_actor,
std::move(fail_state));
queue_.push(callee_.current_execution_unit(), msg_id, queue_.push(callee_.current_execution_unit(), msg_id,
callee_.this_actor(), std::move(ptr)); callee_.this_actor(), std::move(ptr));
} else { } else {
......
...@@ -58,9 +58,7 @@ void message_queue::push(execution_unit* ctx, uint64_t id, ...@@ -58,9 +58,7 @@ void message_queue::push(execution_unit* ctx, uint64_t id,
return; return;
} }
// Get the insertion point. // Get the insertion point.
auto pred = [&](const actor_msg& x) { auto pred = [&](const actor_msg& x) { return x.id >= id; };
return x.id >= id;
};
pending.emplace(std::find_if(first, last, pred), pending.emplace(std::find_if(first, last, pred),
actor_msg{id, std::move(receiver), std::move(content)}); actor_msg{id, std::move(receiver), std::move(content)});
} }
......
...@@ -31,11 +31,9 @@ using namespace caf; ...@@ -31,11 +31,9 @@ using namespace caf;
namespace { namespace {
behavior testee_impl() { behavior testee_impl() {
return { return {[](ok_atom, int) {
[](ok_atom, int) {
// nop // nop
} }};
};
} }
struct fixture : test_coordinator_fixture<> { struct fixture : test_coordinator_fixture<> {
...@@ -59,8 +57,7 @@ struct fixture : test_coordinator_fixture<> { ...@@ -59,8 +57,7 @@ struct fixture : test_coordinator_fixture<> {
} }
}; };
} // namespace
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(message_queue_tests, fixture) CAF_TEST_FIXTURE_SCOPE(message_queue_tests, fixture)
...@@ -77,7 +74,7 @@ CAF_TEST(ascending IDs) { ...@@ -77,7 +74,7 @@ CAF_TEST(ascending IDs) {
CAF_CHECK_EQUAL(queue.next_undelivered, 0u); CAF_CHECK_EQUAL(queue.next_undelivered, 0u);
} }
CAF_TEST(push order 0-1-2) { CAF_TEST(push order 0 - 1 - 2) {
acquire_ids(3); acquire_ids(3);
push(0); push(0);
expect((ok_atom, int), from(self).to(testee).with(_, 0)); expect((ok_atom, int), from(self).to(testee).with(_, 0));
...@@ -87,7 +84,7 @@ CAF_TEST(push order 0-1-2) { ...@@ -87,7 +84,7 @@ CAF_TEST(push order 0-1-2) {
expect((ok_atom, int), from(self).to(testee).with(_, 2)); expect((ok_atom, int), from(self).to(testee).with(_, 2));
} }
CAF_TEST(push order 0-2-1) { CAF_TEST(push order 0 - 2 - 1) {
acquire_ids(3); acquire_ids(3);
push(0); push(0);
expect((ok_atom, int), from(self).to(testee).with(_, 0)); expect((ok_atom, int), from(self).to(testee).with(_, 0));
...@@ -98,7 +95,7 @@ CAF_TEST(push order 0-2-1) { ...@@ -98,7 +95,7 @@ CAF_TEST(push order 0-2-1) {
expect((ok_atom, int), from(self).to(testee).with(_, 2)); expect((ok_atom, int), from(self).to(testee).with(_, 2));
} }
CAF_TEST(push order 1-0-2) { CAF_TEST(push order 1 - 0 - 2) {
acquire_ids(3); acquire_ids(3);
push(1); push(1);
disallow((ok_atom, int), from(self).to(testee)); disallow((ok_atom, int), from(self).to(testee));
...@@ -109,8 +106,7 @@ CAF_TEST(push order 1-0-2) { ...@@ -109,8 +106,7 @@ CAF_TEST(push order 1-0-2) {
expect((ok_atom, int), from(self).to(testee).with(_, 2)); expect((ok_atom, int), from(self).to(testee).with(_, 2));
} }
CAF_TEST(push order 1 - 2 - 0) {
CAF_TEST(push order 1-2-0) {
acquire_ids(3); acquire_ids(3);
push(1); push(1);
disallow((ok_atom, int), from(self).to(testee)); disallow((ok_atom, int), from(self).to(testee));
...@@ -122,7 +118,7 @@ CAF_TEST(push order 1-2-0) { ...@@ -122,7 +118,7 @@ CAF_TEST(push order 1-2-0) {
expect((ok_atom, int), from(self).to(testee).with(_, 2)); expect((ok_atom, int), from(self).to(testee).with(_, 2));
} }
CAF_TEST(push order 2-0-1) { CAF_TEST(push order 2 - 0 - 1) {
acquire_ids(3); acquire_ids(3);
push(2); push(2);
disallow((ok_atom, int), from(self).to(testee)); disallow((ok_atom, int), from(self).to(testee));
...@@ -133,7 +129,7 @@ CAF_TEST(push order 2-0-1) { ...@@ -133,7 +129,7 @@ CAF_TEST(push order 2-0-1) {
expect((ok_atom, int), from(self).to(testee).with(_, 2)); expect((ok_atom, int), from(self).to(testee).with(_, 2));
} }
CAF_TEST(push order 2-1-0) { CAF_TEST(push order 2 - 1 - 0) {
acquire_ids(3); acquire_ids(3);
push(2); push(2);
disallow((ok_atom, int), from(self).to(testee)); disallow((ok_atom, int), from(self).to(testee));
......
...@@ -36,11 +36,9 @@ using namespace caf; ...@@ -36,11 +36,9 @@ using namespace caf;
namespace { namespace {
behavior testee_impl() { behavior testee_impl() {
return { return {[](ok_atom) {
[](ok_atom) {
// nop // nop
} }};
};
} }
class mock_actor_proxy : public actor_proxy { class mock_actor_proxy : public actor_proxy {
...@@ -61,7 +59,7 @@ public: ...@@ -61,7 +59,7 @@ public:
class mock_proxy_registry_backend : public proxy_registry::backend { class mock_proxy_registry_backend : public proxy_registry::backend {
public: public:
mock_proxy_registry_backend(actor_system& sys) : sys_(sys) { mock_proxy_registry_backend(actor_system& sys) : sys_(sys) {
//nop // nop
} }
strong_actor_ptr make_proxy(node_id nid, actor_id aid) override { strong_actor_ptr make_proxy(node_id nid, actor_id aid) override {
...@@ -98,7 +96,7 @@ struct fixture : test_coordinator_fixture<> { ...@@ -98,7 +96,7 @@ struct fixture : test_coordinator_fixture<> {
} }
}; };
} // namespace <anonymous> } // namespace
CAF_TEST_FIXTURE_SCOPE(worker_tests, fixture) CAF_TEST_FIXTURE_SCOPE(worker_tests, fixture)
......
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