Commit 8e88d559 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'issue/1146'

Close #1146.
parents 5dd53c56 f5874b0b
......@@ -19,10 +19,12 @@ is based on [Keep a Changelog](https://keepachangelog.com).
### Fixed
- Setting an invalid credit policy no longer results in a segfault (#1140).
- RC 1 of version 0.18 introduced a regression that prevented CAF from writing
- Version 0.18.0-rc.1 introduced a regression that prevented CAF from writing
parameters parsed from configuration files back to variables. The original
behavior has been restored, i.e., variables synchronize with user input from
configuration files and CLI arguments (#1145).
- Restore correct functionality of `middleman::remote_lookup` (#1146). This
fixes a regression introduced in version 0.18.0-rc.1
## [0.18.0-rc.1] - 2020-09-09
......
......@@ -366,6 +366,7 @@ CAF_BEGIN_TYPE_ID_BLOCK(core_module, 0)
CAF_ADD_ATOM(core_module, caf, put_atom)
CAF_ADD_ATOM(core_module, caf, receive_atom)
CAF_ADD_ATOM(core_module, caf, redirect_atom)
CAF_ADD_ATOM(core_module, caf, registry_lookup_atom)
CAF_ADD_ATOM(core_module, caf, reset_atom)
CAF_ADD_ATOM(core_module, caf, resolve_atom)
CAF_ADD_ATOM(core_module, caf, spawn_atom)
......
......@@ -131,8 +131,8 @@ behavior config_serv_impl(stateful_actor<kvstate>* self) {
self->state.subscribers[subscriber].erase(key);
self->state.data[key].second.erase(subscriber);
},
// get a 'named' actor from local registry
[=](get_atom, const std::string& name) {
// get a 'named' actor from the local registry
[=](registry_lookup_atom, const std::string& name) {
return self->home_system().registry().get(name);
},
};
......
......@@ -111,3 +111,7 @@ caf_add_test_suites(caf-io-test
io.unpublish
io.worker
)
if(NOT WIN32)
caf_add_test_suites(caf-io-test io.middleman)
endif()
......@@ -286,6 +286,13 @@ public:
return new impl(sys);
}
/// @private
actor get_named_broker(const std::string& name) {
if (auto i = named_brokers_.find(name); i != named_brokers_.end())
return i->second;
return {};
}
protected:
middleman(actor_system& sys);
......
......@@ -50,6 +50,7 @@
#include "caf/logger.hpp"
#include "caf/make_counted.hpp"
#include "caf/node_id.hpp"
#include "caf/others.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/sec.hpp"
......@@ -281,12 +282,16 @@ strong_actor_ptr middleman::remote_lookup(std::string name,
scoped_actor self{system(), true};
auto id = basp::header::config_server_id;
self->send(basp, forward_atom_v, nid, id,
make_message(get_atom_v, std::move(name)));
self->receive([&](strong_actor_ptr& addr) { result = std::move(addr); },
after(std::chrono::minutes(5)) >>
[] {
// nop
});
make_message(registry_lookup_atom_v, std::move(name)));
self->receive(
[&](strong_actor_ptr& addr) { result = std::move(addr); },
others >> [](message& msg) -> skippable_result {
CAF_LOG_ERROR(
"middleman received unexpected remote_lookup result:" << msg);
return message{};
},
after(std::chrono::minutes(5)) >>
[&] { CAF_LOG_WARNING("remote_lookup for" << name << "timed out"); });
return result;
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
// Note: this suite is disabled via CMake on Windows, because it lacks the
// socketpair() function.
#define CAF_SUITE io.middleman
#include "caf/io/middleman.hpp"
#include "caf/test/io_dsl.hpp"
#include <sys/socket.h>
#include <sys/types.h>
#include "caf/actor.hpp"
#include "caf/actor_system.hpp"
#include "caf/behavior.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/io/network/scribe_impl.hpp"
#include "caf/scoped_actor.hpp"
using namespace caf;
namespace {
// Unlike our usual fixtures, this test suite does *not* use the test
// coordinator.
struct node_fixture {
struct config : actor_system_config {
config() {
load<io::middleman>();
set("caf.scheduler.policy", "sharing");
set("caf.scheduler.max-threads", 1);
set("caf.middleman.workers", 0);
}
};
node_fixture() : sys(cfg), mm(sys.middleman()), mpx(mm.backend()), self(sys) {
basp_broker = mm.get_named_broker("BASP");
}
config cfg;
actor_system sys;
io::middleman& mm;
io::network::multiplexer& mpx;
scoped_actor self;
actor basp_broker;
};
struct fixture {
node_fixture earth;
node_fixture mars;
fixture() {
// Connect the two BASP brokers via connected socket pair.
int sockets[2];
if (socketpair(AF_UNIX, SOCK_STREAM, 0, sockets) != 0)
CAF_FAIL("socketpair failed");
io::network::nonblocking(sockets[0], true);
io::network::nonblocking(sockets[1], true);
auto server_scribe = earth.mpx.new_scribe(sockets[0]);
auto client_scribe = mars.mpx.new_scribe(sockets[1]);
anon_send(earth.basp_broker, publish_atom_v, std::move(server_scribe),
uint16_t(4242), strong_actor_ptr{}, std::set<std::string>{});
mars.self
->request(mars.basp_broker, std::chrono::minutes(1), connect_atom_v,
std::move(client_scribe), uint16_t(4242))
.receive(
[&](node_id& nid, strong_actor_ptr&, std::set<std::string>&) {
if (nid != earth.sys.node())
CAF_FAIL("mars failed to connect to earth: unexpected node ID");
},
[&](const caf::error& err) {
CAF_FAIL("mars failed to connect to earth: " << err);
});
}
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(middleman_tests, fixture)
CAF_TEST(remote_lookup allows registry lookups on other nodes) {
auto testee_impl = []() -> behavior {
return {
[](int32_t x, int32_t y) { return x + y; },
};
};
auto testee = earth.sys.spawn(testee_impl);
earth.sys.registry().put("testee", testee);
auto testee_proxy_ptr = mars.mm.remote_lookup("testee", earth.sys.node());
auto testee_proxy = actor_cast<actor>(testee_proxy_ptr);
CAF_CHECK_EQUAL(testee->node(), testee_proxy.node());
CAF_CHECK_EQUAL(testee->id(), testee_proxy.id());
mars.self
->request(testee_proxy, std::chrono::minutes(1), int32_t{7}, int32_t{8})
.receive([](int32_t result) { CAF_CHECK_EQUAL(result, 15); },
[](caf::error& err) { CAF_FAIL("request failed: " << err); });
anon_send_exit(testee, exit_reason::user_shutdown);
}
CAF_TEST_FIXTURE_SCOPE_END()
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