Commit f00b4a53 authored by Dominik Charousset's avatar Dominik Charousset

Add remote spawning capabilities to middleman

parent c0e117cc
......@@ -120,6 +120,9 @@ using open_atom = atom_constant<atom("OPEN")>;
/// Generic 'CLOSE' atom, e.g., for closing a port or file.
using close_atom = atom_constant<atom("CLOSE")>;
/// Generic 'SPAWN' atom, e.g., for spawning remote actors.
using spawn_atom = atom_constant<atom("SPAWN")>;
} // namespace caf
namespace std {
......
......@@ -463,7 +463,9 @@ public:
return (mid.is_request()) ? mid.response_id() : message_id();
}
void forward_message(const actor& dest, message_priority mp);
void forward_current_message(const actor& dest);
void forward_current_message(const actor& dest, message_priority mp);
template <class... Ts>
void delegate(message_priority mp, const actor& dest, Ts&&... xs) {
......
......@@ -49,9 +49,13 @@ public:
}
/// Sends `response_message` and invalidates this handle afterwards.
void deliver(message response_message) const;
template <class... Ts>
void deliver(Ts&&... xs) const {
deliver_impl(make_message(std::forward<Ts>(xs)...));
}
private:
void deliver_impl(message response_message) const;
actor_addr from_;
actor_addr to_;
message_id id_;
......
......@@ -45,12 +45,18 @@ behavior announce_actor_type_server(stateful_actor<spawner_state>* self) {
[=](add_atom, std::string& name, spawn_fun& f) {
self->state.funs_.emplace(std::move(name), std::move(f));
},
[=](get_atom, const std::string& name, message& args) -> spawn_result {
[=](get_atom, const std::string& name, message& args)
-> either<ok_atom, actor_addr, std::set<std::string>>
::or_else<error_atom, std::string> {
auto i = self->state.funs_.find(name);
if (i == self->state.funs_.end())
return std::make_pair(invalid_actor_addr, std::set<std::string>{});
return {error_atom::value, "no actor type found named " + name};
auto f = i->second;
return f(args);
auto res = f(args);
if (res.first == invalid_actor_addr)
return {error_atom::value, "cannot initialize an actor type " + name
+ " using the provided arguments"};
return {ok_atom::value, res.first, res.second};
},
others >> [=] {
CAF_LOGF_WARNING("Unexpected message: "
......
......@@ -30,7 +30,7 @@ event_based_actor::~event_based_actor() {
void event_based_actor::forward_to(const actor& whom,
message_priority prio) {
forward_message(whom, prio);
forward_current_message(whom, prio);
}
void event_based_actor::initialize() {
......
......@@ -108,10 +108,16 @@ std::vector<group> local_actor::joined_groups() const {
return result;
}
void local_actor::forward_message(const actor& dest, message_priority prio) {
if (! dest) {
void local_actor::forward_current_message(const actor& dest) {
if (! dest)
return;
dest->enqueue(std::move(current_element_), host());
}
void local_actor::forward_current_message(const actor& dest,
message_priority prio) {
if (! dest)
return;
}
auto mid = current_element_->mid;
current_element_->mid = prio == message_priority::high
? mid.with_high_priority()
......@@ -348,7 +354,6 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
} else {
auto res = post_process_invoke_res(this, mid,
fun(current_mailbox_element()->msg));
ptr.swap(current_mailbox_element());
if (! res) {
CAF_LOG_WARNING("sync failure occured in actor "
<< "with ID " << id());
......
......@@ -30,10 +30,9 @@ response_promise::response_promise(const actor_addr& from, const actor_addr& to,
CAF_ASSERT(id.is_response() || ! id.valid());
}
void response_promise::deliver(message msg) const {
if (! to_) {
void response_promise::deliver_impl(message msg) const {
if (! to_)
return;
}
auto to = actor_cast<abstract_actor_ptr>(to_);
auto from = actor_cast<abstract_actor_ptr>(from_);
to->enqueue(from_, id_, std::move(msg), from->host());
......
......@@ -47,22 +47,26 @@ struct fixture {
void set_aut(message args, bool expect_fail = false) {
CAF_MESSAGE("set aut");
scoped_actor self;
self->sync_send(spawner, get_atom::value, "test_actor", std::move(args)).await(
[&](spawn_result& res) {
self->on_sync_failure([&] {
CAF_TEST_ERROR("received unexpeced sync. response: "
<< to_string(self->current_message()));
});
if (expect_fail) {
CAF_REQUIRE(res.first == invalid_actor_addr);
return;
self->sync_send(spawner, get_atom::value, "test_actor", std::move(args)).await(
[&](error_atom, const std::string&) {
CAF_TEST_VERBOSE("received error_atom (expected)");
}
CAF_REQUIRE(res.first != invalid_actor_addr);
CAF_CHECK(res.second.empty());
aut = actor_cast<actor>(res.first);
},
others >> [&] {
CAF_TEST_ERROR("unexpected message: "
<< to_string(self->current_message()));
);
} else {
self->sync_send(spawner, get_atom::value, "test_actor", std::move(args)).await(
[&](ok_atom, actor_addr res, const std::set<std::string>& ifs) {
CAF_REQUIRE(res != invalid_actor_addr);
aut = actor_cast<actor>(res);
CAF_CHECK(ifs.empty());
}
);
}
}
~fixture() {
{ // lifetime scope of scoped_actor
......
......@@ -32,6 +32,7 @@
#include "caf/io/remote_group.hpp"
#include "caf/io/set_middleman.hpp"
#include "caf/io/receive_policy.hpp"
#include "caf/io/middleman_actor.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/publish_local_groups.hpp"
......
......@@ -80,6 +80,17 @@ namespace io {
/// -> either (ok_atom)
/// or (error_atom, string error_string)
///
/// // Spawns an actor on a remote node, initializing it
/// // using the arguments stored in `msg`.
/// // @param nid ID of the remote node that should spawn the actor.
/// // @param name Announced type name of the actor.
/// // @param args Initialization arguments for the actor.
/// // @returns The address of the spawned actor and its interface
/// // description on success; an error string otherwise.
/// (spawn_atom, node_id nid, string name, message args)
/// -> either (ok_atom, actor_addr, set<string>
/// or (error_atom, string error_string)
///
/// }
/// ~~~
using middleman_actor =
......@@ -103,6 +114,10 @@ using middleman_actor =
replies_to<close_atom, uint16_t>
::with_either<ok_atom>
::or_else<error_atom, std::string>,
replies_to<spawn_atom, node_id, std::string, message>
::with_either<ok_atom, actor_addr, std::set<std::string>>
::or_else<error_atom, std::string>>;
/// Returns a handle for asynchronous networking operations.
......
......@@ -401,6 +401,26 @@ behavior basp_broker::make_behavior() {
}
return make_message(error_atom::value, "no doorman for given port found");
},
[=](spawn_atom, const node_id& nid, std::string& type, message& args)
-> delegated<either<ok_atom, actor_addr, std::set<std::string>>
::or_else<error_atom, std::string>> {
auto err = [=](std::string errmsg) {
auto rp = make_response_promise();
rp.deliver(error_atom::value, std::move(errmsg));
};
auto na = state.instance.named_actors(nid);
if (! na) {
err("no connection to requested node");
return {};
}
auto i = na->find(atom("spawner"));
if (i == na->end()) {
err("no spawn server on requested node");
return {};
}
delegate(i->second, get_atom::value, std::move(type), std::move(args));
return {};
},
// catch-all error handler
others >>
[=] {
......
......@@ -186,12 +186,18 @@ public:
}
return {};
},
[=](unpublish_atom, actor_addr& whom, uint16_t port) -> del_res {
delegate(broker_, unpublish_atom::value, std::move(whom), port);
[=](unpublish_atom, const actor_addr&, uint16_t) -> del_res {
forward_current_message(broker_);
return {};
},
[=](close_atom, uint16_t port) -> del_res {
delegate(broker_, close_atom::value, port);
[=](close_atom, uint16_t) -> del_res {
forward_current_message(broker_);
return {};
},
[=](spawn_atom, const node_id&, const std::string&, const message&)
-> delegated<either<ok_atom, actor_addr, std::set<std::string>>
::or_else<error_atom, std::string>> {
forward_current_message(broker_);
return {};
}
};
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#include "caf/config.hpp"
#define CAF_SUITE io_remote_spawn
#include "caf/test/unit_test.hpp"
#include <thread>
#include <string>
#include <cstring>
#include <sstream>
#include <iostream>
#include <functional>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
#include "caf/detail/run_program.hpp"
#include "caf/experimental/announce_actor_type.hpp"
#ifdef CAF_USE_ASIO
#include "caf/io/network/asio_multiplexer.hpp"
#endif // CAF_USE_ASIO
using namespace caf;
using namespace caf::experimental;
namespace {
behavior mirror(event_based_actor* self) {
return {
others >> [=] {
return self->current_message();
}
};
}
behavior client(event_based_actor* self, actor serv) {
self->send(serv, ok_atom::value);
return {
others >> [=] {
CAF_TEST_ERROR("unexpected message: "
<< to_string(self->current_message()));
}
};
}
struct server_state {
actor client; // the spawn we connect to the server in our main
actor aut; // our mirror
};
behavior server(stateful_actor<server_state>* self) {
self->on_sync_failure([=] {
CAF_TEST_ERROR("unexpected sync response: "
<< to_string(self->current_message()));
});
return {
[=](ok_atom) {
auto s = self->current_sender();
CAF_REQUIRE(s != invalid_actor_addr);
CAF_REQUIRE(s.is_remote());
self->state.client = actor_cast<actor>(s);
auto mm = io::get_middleman_actor();
self->sync_send(mm, spawn_atom::value,
s.node(), "mirror", make_message()).then(
[=](ok_atom, const actor_addr& addr, const std::set<std::string>& ifs) {
CAF_REQUIRE(addr != invalid_actor_addr);
CAF_CHECK(ifs.empty());
self->state.aut = actor_cast<actor>(addr);
self->send(self->state.aut, "hello mirror");
self->become(
[=](const std::string& str) {
CAF_CHECK(self->current_sender() == self->state.aut);
CAF_CHECK(str == "hello mirror");
self->send_exit(self->state.aut, exit_reason::kill);
self->send_exit(self->state.client, exit_reason::kill);
self->quit();
}
);
},
[=](error_atom, const std::string& errmsg) {
CAF_TEST_ERROR("could not spawn mirror: " << errmsg);
}
);
}
};
}
} // namespace <anonymous>
CAF_TEST(remote_spawn) {
announce_actor_type("mirror", mirror);
auto argv = caf::test::engine::argv();
auto argc = caf::test::engine::argc();
uint16_t port;
auto r = message_builder(argv, argv + argc).extract_opts({
{"server,s", "run as server (don't run client"},
{"client,c", "add client port (two needed)", port},
{"use-asio", "use ASIO network backend (if available)"}
});
if (! r.error.empty() || r.opts.count("help") > 0 || ! r.remainder.empty()) {
std::cout << r.error << std::endl << std::endl << r.helptext << std::endl;
return;
}
auto use_asio = r.opts.count("use-asio") > 0;
if (use_asio) {
# ifdef CAF_USE_ASIO
CAF_MESSAGE("enable ASIO backend");
io::set_middleman<io::network::asio_multiplexer>();
# endif // CAF_USE_ASIO
}
if (r.opts.count("client") > 0) {
auto serv = io::remote_actor("localhost", port);
spawn(client, serv);
await_all_actors_done();
return;
}
auto serv = spawn(server);
port = io::publish(serv, 0);
CAF_TEST_INFO("published server at port " << port);
if (r.opts.count("server") == 0) {
auto child = detail::run_program(invalid_actor, argv[0], "-n", "-s",
CAF_XSTR(CAF_SUITE), "--", "-c",
port, (use_asio ? "--use-asio" : ""));
child.join();
}
await_all_actors_done();
}
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