Commit 823ad2cc authored by Dominik Charousset's avatar Dominik Charousset

Add caf-run tool, revert publish API

Return an `uint16_t` instead of `maybe<uint16_t>` from all publish variants
again. Add the new caf-run tool under tools/.
parent 9728b50c
...@@ -25,6 +25,9 @@ endif() ...@@ -25,6 +25,9 @@ endif()
if(NOT CAF_NO_OPENCL) if(NOT CAF_NO_OPENCL)
set(CAF_NO_OPENCL no) set(CAF_NO_OPENCL no)
endif() endif()
if(NOT CAF_NO_TOOLS)
set(CAF_NO_TOOLS no)
endif()
if(NOT CAF_NO_SUMMARY) if(NOT CAF_NO_SUMMARY)
set(CAF_NO_SUMMARY no) set(CAF_NO_SUMMARY no)
endif() endif()
...@@ -443,6 +446,9 @@ add_optional_caf_lib(riac) ...@@ -443,6 +446,9 @@ add_optional_caf_lib(riac)
# build examples if not being told otherwise # build examples if not being told otherwise
add_optional_caf_binaries(examples) add_optional_caf_binaries(examples)
# build tools if not being told otherwise
add_optional_caf_binaries(tools)
# build nexus if not being told otherwise # build nexus if not being told otherwise
add_optional_caf_binaries(nexus CAF_NO_RIAC) add_optional_caf_binaries(nexus CAF_NO_RIAC)
...@@ -565,6 +571,7 @@ macro(invertYesNo in out) ...@@ -565,6 +571,7 @@ macro(invertYesNo in out)
endmacro() endmacro()
# invert CAF_NO_* variables for nicer output # invert CAF_NO_* variables for nicer output
invertYesNo(CAF_NO_EXAMPLES CAF_BUILD_EXAMPLES) invertYesNo(CAF_NO_EXAMPLES CAF_BUILD_EXAMPLES)
invertYesNo(CAF_NO_TOOLS CAF_BUILD_TOOLS)
invertYesNo(CAF_NO_UNIT_TESTS CAF_BUILD_UNIT_TESTS) invertYesNo(CAF_NO_UNIT_TESTS CAF_BUILD_UNIT_TESTS)
invertYesNo(CAF_NO_RIAC CAF_BUILD_RIAC) invertYesNo(CAF_NO_RIAC CAF_BUILD_RIAC)
invertYesNo(CAF_NO_NEXUS CAF_BUILD_NEXUS) invertYesNo(CAF_NO_NEXUS CAF_BUILD_NEXUS)
...@@ -589,6 +596,7 @@ if(NOT CAF_NO_SUMMARY) ...@@ -589,6 +596,7 @@ if(NOT CAF_NO_SUMMARY)
"\nLog level: ${LOG_LEVEL_STR}" "\nLog level: ${LOG_LEVEL_STR}"
"\nWith mem. mgmt.: ${CAF_BUILD_MEM_MANAGEMENT}" "\nWith mem. mgmt.: ${CAF_BUILD_MEM_MANAGEMENT}"
"\n" "\n"
"\nBuild tools: ${CAF_BUILD_TOOLS}"
"\nBuild examples: ${CAF_BUILD_EXAMPLES}" "\nBuild examples: ${CAF_BUILD_EXAMPLES}"
"\nBuild unit tests: ${CAF_BUILD_UNIT_TESTS}" "\nBuild unit tests: ${CAF_BUILD_UNIT_TESTS}"
"\nBuild riac: ${CAF_BUILD_RIAC}" "\nBuild riac: ${CAF_BUILD_RIAC}"
......
...@@ -65,6 +65,7 @@ Usage: $0 [OPTION]... [VAR=VALUE]... ...@@ -65,6 +65,7 @@ Usage: $0 [OPTION]... [VAR=VALUE]...
--no-cash build without cash --no-cash build without cash
--no-benchmarks build without benchmarks --no-benchmarks build without benchmarks
--no-riac build without riac --no-riac build without riac
--no-tools build without CAF tools such as caf-run
--no-summary do not print configuration before building --no-summary do not print configuration before building
Testing: Testing:
...@@ -350,6 +351,9 @@ while [ $# -ne 0 ]; do ...@@ -350,6 +351,9 @@ while [ $# -ne 0 ]; do
--no-riac) --no-riac)
append_cache_entry CAF_NO_RIAC BOOL yes append_cache_entry CAF_NO_RIAC BOOL yes
;; ;;
--no-tools)
append_cache_entry CAF_NO_TOOLS BOOL yes
;;
--no-summary) --no-summary)
append_cache_entry CAF_NO_SUMMARY BOOL yes append_cache_entry CAF_NO_SUMMARY BOOL yes
;; ;;
......
...@@ -183,9 +183,13 @@ maybe<uint16_t> as_u16(const std::string& str) { ...@@ -183,9 +183,13 @@ maybe<uint16_t> as_u16(const std::string& str) {
int main(int argc, char** argv) { int main(int argc, char** argv) {
using std::string; using std::string;
actor_system_config cfg{argc, argv};
if (cfg.cli_helptext_printed)
return 0;
cfg.load<io::middleman>();
uint16_t port = 0; uint16_t port = 0;
string host = "localhost"; string host = "localhost";
auto res = message_builder(argv + 1, argv + argc).extract_opts({ auto res = cfg.args_remainder.extract_opts({
{"server,s", "run in server mode"}, {"server,s", "run in server mode"},
{"client,c", "run in client mode"}, {"client,c", "run in client mode"},
{"port,p", "set port", port}, {"port,p", "set port", port},
...@@ -200,8 +204,6 @@ int main(int argc, char** argv) { ...@@ -200,8 +204,6 @@ int main(int argc, char** argv) {
return cerr << "*** too many arguments" << endl << res.helptext << endl, 1; return cerr << "*** too many arguments" << endl << res.helptext << endl, 1;
if (res.opts.count("port") == 0) if (res.opts.count("port") == 0)
return cerr << "*** no port given" << endl << res.helptext << endl, 1; return cerr << "*** no port given" << endl << res.helptext << endl, 1;
actor_system_config cfg;
cfg.load<io::middleman>();
actor_system system{cfg}; actor_system system{cfg};
if (res.opts.count("server") > 0) { if (res.opts.count("server") > 0) {
cout << "run in server mode" << endl; cout << "run in server mode" << endl;
......
...@@ -284,9 +284,7 @@ int main(int argc, char** argv) { ...@@ -284,9 +284,7 @@ int main(int argc, char** argv) {
// try to publish math actor at given port // try to publish math actor at given port
cout << "*** try publish at port " << port << endl; cout << "*** try publish at port " << port << endl;
auto p = system.middleman().publish(calc, port); auto p = system.middleman().publish(calc, port);
if (! p) cout << "*** server successfully published at port " << p << endl;
return cerr << "*** error: " << system.render(p.error()) << endl, 1;
cout << "*** server successfully published at port " << *p << endl;
cout << "*** press [enter] to quit" << endl; cout << "*** press [enter] to quit" << endl;
string dummy; string dummy;
std::getline(std::cin, dummy); std::getline(std::cin, dummy);
......
...@@ -32,9 +32,7 @@ int main(int argc, char** argv) { ...@@ -32,9 +32,7 @@ int main(int argc, char** argv) {
if (res.opts.count("port") == 0 || port <= 1024) if (res.opts.count("port") == 0 || port <= 1024)
return cerr << "*** no valid port given" << endl << res.helptext << endl, 1; return cerr << "*** no valid port given" << endl << res.helptext << endl, 1;
actor_system system{actor_system_config{}.load<io::middleman>()}; actor_system system{actor_system_config{}.load<io::middleman>()};
auto pres = system.middleman().publish_local_groups(port); system.middleman().publish_local_groups(port);
if (! pres)
return cerr << "*** error: " << system.render(pres.error()) << endl, 1;
cout << "type 'quit' to shutdown the server" << endl; cout << "type 'quit' to shutdown the server" << endl;
string line; string line;
while (getline(cin, line)) while (getline(cin, line))
......
...@@ -168,6 +168,22 @@ public: ...@@ -168,6 +168,22 @@ public:
return *this; return *this;
} }
/// Stores whether the help text for this config object was
/// printed. If set to `true`, the application should not use
/// this config object to initialize an `actor_system` and
/// return from `main` immediately.
bool cli_helptext_printed;
/// Stores whether this node was started in slave mode.
bool slave_mode;
/// Stores the name of this node when started in slave mode.
std::string slave_name;
/// Stores credentials for connecting to the bootstrap node
/// when using the caf-run launcher.
std::string bootstrap_node;
/// Stores CLI arguments that were not consumed by CAF. /// Stores CLI arguments that were not consumed by CAF.
message args_remainder; message args_remainder;
...@@ -188,14 +204,17 @@ public: ...@@ -188,14 +204,17 @@ public:
size_t middleman_max_consecutive_reads; size_t middleman_max_consecutive_reads;
size_t middleman_heartbeat_interval; size_t middleman_heartbeat_interval;
// System parameters that are set while initializing modules.
node_id network_id;
proxy_registry* network_proxies;
// Config parameters of RIAC probes. // Config parameters of RIAC probes.
std::string nexus_host; std::string nexus_host;
uint16_t nexus_port; uint16_t nexus_port;
// Config parameters of the OpenCL module.
std::string opencl_device_ids;
// System parameters that are set while initializing modules.
node_id network_id;
proxy_registry* network_proxies;
private: private:
value_factories_by_name value_factories_by_name_; value_factories_by_name value_factories_by_name_;
value_factories_by_rtti value_factories_by_rtti_; value_factories_by_rtti value_factories_by_rtti_;
......
...@@ -129,6 +129,11 @@ void replace_all(std::string& str, ...@@ -129,6 +129,11 @@ void replace_all(std::string& str,
} }
} }
template<size_t S>
bool starts_with(const std::string& str, const char (&prefix)[S]) {
return str.compare(0, S, prefix);
}
template <class T> template <class T>
typename std::enable_if< typename std::enable_if<
std::is_arithmetic<T>::value, std::is_arithmetic<T>::value,
......
...@@ -197,7 +197,9 @@ actor_system_config::option::~option() { ...@@ -197,7 +197,9 @@ actor_system_config::option::~option() {
// in this config class, we have (1) hard-coded defaults that are overridden // in this config class, we have (1) hard-coded defaults that are overridden
// by (2) INI-file contents that are in turn overridden by (3) CLI arguments // by (2) INI-file contents that are in turn overridden by (3) CLI arguments
actor_system_config::actor_system_config() { actor_system_config::actor_system_config()
: cli_helptext_printed(false),
slave_mode(false) {
// (1) hard-coded defaults // (1) hard-coded defaults
scheduler_policy = atom("stealing"); scheduler_policy = atom("stealing");
scheduler_max_threads = std::max(std::thread::hardware_concurrency(), scheduler_max_threads = std::max(std::thread::hardware_concurrency(),
...@@ -238,17 +240,20 @@ actor_system_config::actor_system_config() { ...@@ -238,17 +240,20 @@ actor_system_config::actor_system_config() {
"sets the hostname or IP address for connecting to the Nexus") "sets the hostname or IP address for connecting to the Nexus")
.add(nexus_port, "probe.nexus-port", .add(nexus_port, "probe.nexus-port",
"sets the port for connecting to the Nexus"); "sets the port for connecting to the Nexus");
opt_group(options_, "opencl")
.add(opencl_device_ids, "device-ids",
"restricts which OpenCL devices are accessed by CAF");
} }
actor_system_config::actor_system_config(int argc, char** argv) actor_system_config::actor_system_config(int argc, char** argv)
: actor_system_config() { : actor_system_config() {
if (argc < 2) if (argc < 1)
return; return;
auto args = message_builder(argv, argv + argc).move_to_message(); auto args = message_builder(argv + 1, argv + argc).move_to_message();
// extract config file name first, since INI files are overruled by CLI args // extract config file name first, since INI files are overruled by CLI args
std::string config_file_name; std::string config_file_name;
args.extract_opts({ args.extract_opts({
{"caf-config-file", "", config_file_name} {"caf#config-file", "", config_file_name}
}); });
// (2) content of the INI file overrides hard-coded defaults // (2) content of the INI file overrides hard-coded defaults
if (! config_file_name.empty()) { if (! config_file_name.empty()) {
...@@ -262,15 +267,27 @@ actor_system_config::actor_system_config(int argc, char** argv) ...@@ -262,15 +267,27 @@ actor_system_config::actor_system_config(int argc, char** argv)
std::vector<message::cli_arg> cargs; std::vector<message::cli_arg> cargs;
for (auto& x : options_) for (auto& x : options_)
cargs.emplace_back(x->to_cli_arg()); cargs.emplace_back(x->to_cli_arg());
cargs.emplace_back("caf-dump-config", "print config in INI format to stdout"); cargs.emplace_back("caf#dump-config", "print config in INI format to stdout");
cargs.emplace_back("caf-help", "print this text"); cargs.emplace_back("caf#help", "print this text");
cargs.emplace_back("caf-config-file", "parse INI file", config_file_name); cargs.emplace_back("caf#config-file", "parse INI file", config_file_name);
cargs.emplace_back("caf#slave-mode", "run in slave mode");
cargs.emplace_back("caf#slave-name", "set name for this slave", slave_name);
cargs.emplace_back("caf#bootstrap-node", "set bootstrapping", bootstrap_node);
auto res = args.extract_opts(std::move(cargs), nullptr, true); auto res = args.extract_opts(std::move(cargs), nullptr, true);
using std::cerr; using std::cerr;
using std::cout; using std::cout;
using std::endl; using std::endl;
if (res.opts.count("caf-help")) if (res.opts.count("caf#help")) {
cli_helptext_printed = true;
cout << res.helptext << endl; cout << res.helptext << endl;
}
if (res.opts.count("caf#slave-mode")) {
slave_mode = true;
if (slave_name.empty())
std::cerr << "running in slave mode but no name was configured" << endl;
if (bootstrap_node.empty())
std::cerr << "running in slave mode without bootstrap node" << endl;
}
auto verify_atom_opt = [](std::initializer_list<atom_value> xs, atom_value& x, auto verify_atom_opt = [](std::initializer_list<atom_value> xs, atom_value& x,
const char* xname) { const char* xname) {
if (std::find(xs.begin(), xs.end(), x) == xs.end()) { if (std::find(xs.begin(), xs.end(), x) == xs.end()) {
...@@ -287,7 +304,8 @@ actor_system_config::actor_system_config(int argc, char** argv) ...@@ -287,7 +304,8 @@ actor_system_config::actor_system_config(int argc, char** argv)
}, middleman_network_backend, "middleman.network-backend"); }, middleman_network_backend, "middleman.network-backend");
verify_atom_opt({atom("stealing"), atom("sharing")}, verify_atom_opt({atom("stealing"), atom("sharing")},
scheduler_policy, "scheduler.policy "); scheduler_policy, "scheduler.policy ");
if (res.opts.count("caf-dump-config")) { if (res.opts.count("caf#dump-config")) {
cli_helptext_printed = true;
cout << std::boolalpha cout << std::boolalpha
<< "[scheduler]" << endl << "[scheduler]" << endl
<< "policy=" << deep_to_string(scheduler_policy) << endl << "policy=" << deep_to_string(scheduler_policy) << endl
......
...@@ -53,9 +53,8 @@ public: ...@@ -53,9 +53,8 @@ public:
/// @returns The actual port the OS uses after `bind()`. If `port == 0` the OS /// @returns The actual port the OS uses after `bind()`. If `port == 0` the OS
/// chooses a random high-level port. /// chooses a random high-level port.
template <class Handle> template <class Handle>
maybe<uint16_t> publish(const Handle& whom, uint16_t port, uint16_t publish(const Handle& whom, uint16_t port,
const char* in = nullptr, const char* in = nullptr, bool reuse_addr = false) {
bool reuse_addr = false) {
return publish(whom.address(), system().message_types(whom), return publish(whom.address(), system().message_types(whom),
port, in, reuse_addr); port, in, reuse_addr);
} }
...@@ -64,8 +63,7 @@ public: ...@@ -64,8 +63,7 @@ public:
/// on address `addr` and `port`. /// on address `addr` and `port`.
/// @returns The actual port the OS uses after `bind()`. If `port == 0` /// @returns The actual port the OS uses after `bind()`. If `port == 0`
/// the OS chooses a random high-level port. /// the OS chooses a random high-level port.
maybe<uint16_t> publish_local_groups(uint16_t port, uint16_t publish_local_groups(uint16_t port, const char* addr = nullptr);
const char* addr = nullptr);
/// Unpublishes `whom` by closing `port` or all assigned ports if `port == 0`. /// Unpublishes `whom` by closing `port` or all assigned ports if `port == 0`.
/// @param whom Actor that should be unpublished at `port`. /// @param whom Actor that should be unpublished at `port`.
......
...@@ -140,7 +140,7 @@ uint16_t middleman::publish(const actor_addr& whom, std::set<std::string> sigs, ...@@ -140,7 +140,7 @@ uint16_t middleman::publish(const actor_addr& whom, std::set<std::string> sigs,
return result; return result;
} }
maybe<uint16_t> middleman::publish_local_groups(uint16_t port, const char* in) { uint16_t middleman::publish_local_groups(uint16_t port, const char* in) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(in)); CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(in));
auto group_nameserver = [](event_based_actor* self) -> behavior { auto group_nameserver = [](event_based_actor* self) -> behavior {
return { return {
...@@ -150,19 +150,17 @@ maybe<uint16_t> middleman::publish_local_groups(uint16_t port, const char* in) { ...@@ -150,19 +150,17 @@ maybe<uint16_t> middleman::publish_local_groups(uint16_t port, const char* in) {
}; };
}; };
auto gn = system().spawn<hidden>(group_nameserver); auto gn = system().spawn<hidden>(group_nameserver);
maybe<uint16_t> result;
try { try {
result = publish(gn, port, in); auto result = publish(gn, port, in);
add_shutdown_cb([gn] {
anon_send_exit(gn, exit_reason::user_shutdown);
});
return result;
} }
catch (std::exception&) { catch (std::exception&) {
anon_send_exit(gn, exit_reason::user_shutdown); anon_send_exit(gn, exit_reason::user_shutdown);
throw; throw;
} }
if (result)
add_shutdown_cb([gn] {
anon_send_exit(gn, exit_reason::user_shutdown);
});
return result;
} }
void middleman::unpublish(const actor_addr& whom, uint16_t port) { void middleman::unpublish(const actor_addr& whom, uint16_t port) {
......
...@@ -547,8 +547,7 @@ CAF_TEST(publish_and_connect) { ...@@ -547,8 +547,7 @@ CAF_TEST(publish_and_connect) {
auto ax = accept_handle::from_int(4242); auto ax = accept_handle::from_int(4242);
mpx()->provide_acceptor(4242, ax); mpx()->provide_acceptor(4242, ax);
auto res = system.middleman().publish(self(), 4242); auto res = system.middleman().publish(self(), 4242);
CAF_REQUIRE(res.valid()); CAF_REQUIRE(res == 4242);
CAF_CHECK(*res == 4242);
mpx()->flush_runnables(); // process publish message in basp_broker mpx()->flush_runnables(); // process publish message in basp_broker
connect_node(0, ax, self()->id()); connect_node(0, ax, self()->id());
} }
......
...@@ -123,18 +123,16 @@ CAF_TEST(identity_semantics) { ...@@ -123,18 +123,16 @@ CAF_TEST(identity_semantics) {
auto server = server_side.spawn(make_pong_behavior); auto server = server_side.spawn(make_pong_behavior);
auto port1 = server_side_mm.publish(server, 0, local_host); auto port1 = server_side_mm.publish(server, 0, local_host);
auto port2 = server_side_mm.publish(server, 0, local_host); auto port2 = server_side_mm.publish(server, 0, local_host);
CAF_REQUIRE(port1 && *port1); CAF_REQUIRE(port1 != port2);
CAF_REQUIRE(port2 && *port2); auto same_server = server_side_mm.remote_actor(local_host, port2);
CAF_CHECK_NOT_EQUAL(*port1, *port2);
auto same_server = server_side_mm.remote_actor(local_host, *port2);
CAF_REQUIRE_EQUAL(same_server, server); CAF_REQUIRE_EQUAL(same_server, server);
CAF_CHECK_EQUAL(same_server->node(), server_side.node()); CAF_CHECK_EQUAL(same_server->node(), server_side.node());
// client side // client side
auto server1 = client_side_mm.remote_actor(local_host, *port1); auto server1 = client_side_mm.remote_actor(local_host, port1);
auto server2 = client_side_mm.remote_actor(local_host, *port2); auto server2 = client_side_mm.remote_actor(local_host, port2);
CAF_CHECK(server1 && server2); CAF_CHECK(server1 && server2);
CAF_CHECK_EQUAL(server1, client_side_mm.remote_actor(local_host, *port1)); CAF_CHECK_EQUAL(server1, client_side_mm.remote_actor(local_host, port1));
CAF_CHECK_EQUAL(server2, client_side_mm.remote_actor(local_host, *port2)); CAF_CHECK_EQUAL(server2, client_side_mm.remote_actor(local_host, port2));
// cleanup // cleanup
caf::anon_send_exit(server, caf::exit_reason::user_shutdown); caf::anon_send_exit(server, caf::exit_reason::user_shutdown);
} }
...@@ -143,9 +141,8 @@ CAF_TEST(ping_pong) { ...@@ -143,9 +141,8 @@ CAF_TEST(ping_pong) {
// server side // server side
auto port = server_side_mm.publish(server_side.spawn(make_pong_behavior), auto port = server_side_mm.publish(server_side.spawn(make_pong_behavior),
0, local_host); 0, local_host);
CAF_REQUIRE(port && *port);
// client side // client side
auto pong = client_side_mm.remote_actor(local_host, *port); auto pong = client_side_mm.remote_actor(local_host, port);
CAF_REQUIRE(pong); CAF_REQUIRE(pong);
client_side.spawn(make_ping_behavior, pong); client_side.spawn(make_ping_behavior, pong);
} }
...@@ -154,9 +151,8 @@ CAF_TEST(custom_message_type) { ...@@ -154,9 +151,8 @@ CAF_TEST(custom_message_type) {
// server side // server side
auto port = server_side_mm.publish(server_side.spawn(make_sort_behavior), auto port = server_side_mm.publish(server_side.spawn(make_sort_behavior),
0, local_host); 0, local_host);
CAF_REQUIRE(port && *port);
// client side // client side
auto sorter = client_side_mm.remote_actor(local_host, *port); auto sorter = client_side_mm.remote_actor(local_host, port);
CAF_REQUIRE(sorter); CAF_REQUIRE(sorter);
client_side.spawn(make_sort_requester_behavior, sorter); client_side.spawn(make_sort_requester_behavior, sorter);
} }
......
...@@ -125,18 +125,18 @@ CAF_TEST_FIXTURE_SCOPE(dynamic_remote_group_tests, fixture) ...@@ -125,18 +125,18 @@ CAF_TEST_FIXTURE_SCOPE(dynamic_remote_group_tests, fixture)
CAF_TEST(remote_group_conn) { CAF_TEST(remote_group_conn) {
// server side // server side
auto port = server_side_mm.publish_local_groups(0); auto port = server_side_mm.publish_local_groups(0);
CAF_REQUIRE(port && *port); CAF_REQUIRE(port != 0);
// client side // client side
CAF_CHECK(client_side_mm.remote_group("whatever", local_host, *port)); CAF_CHECK(client_side_mm.remote_group("whatever", local_host, port));
} }
CAF_TEST(server_side_group_comm) { CAF_TEST(server_side_group_comm) {
// server side // server side
auto port = server_side_mm.publish(server_side.spawn(make_server_behavior), auto port = server_side_mm.publish(server_side.spawn(make_server_behavior),
0, local_host); 0, local_host);
CAF_REQUIRE(port && *port); CAF_REQUIRE(port != 0);
// client side // client side
auto server = client_side_mm.remote_actor(local_host, *port); auto server = client_side_mm.remote_actor(local_host, port);
CAF_REQUIRE(server); CAF_REQUIRE(server);
scoped_actor group_resolver(client_side, true); scoped_actor group_resolver(client_side, true);
group grp; group grp;
...@@ -152,9 +152,9 @@ CAF_TEST(client_side_group_comm) { ...@@ -152,9 +152,9 @@ CAF_TEST(client_side_group_comm) {
// server side // server side
auto port = server_side_mm.publish(server_side.spawn(make_server_behavior), auto port = server_side_mm.publish(server_side.spawn(make_server_behavior),
0, local_host); 0, local_host);
CAF_REQUIRE(port && *port); CAF_REQUIRE(port != 0);
// client side // client side
auto server = client_side_mm.remote_actor(local_host, *port); auto server = client_side_mm.remote_actor(local_host, port);
CAF_REQUIRE(server); CAF_REQUIRE(server);
client_side.spawn(make_client_behavior, server, client_side.spawn(make_client_behavior, server,
client_side.groups().get("local", "foobar")); client_side.groups().get("local", "foobar"));
......
...@@ -108,9 +108,8 @@ void run_client(int argc, char** argv, uint16_t port) { ...@@ -108,9 +108,8 @@ void run_client(int argc, char** argv, uint16_t port) {
void run_server(int argc, char** argv) { void run_server(int argc, char** argv) {
actor_system system{actor_system_config{argc, argv}.load<io::middleman>()}; actor_system system{actor_system_config{argc, argv}.load<io::middleman>()};
auto serv = system.spawn(server); auto serv = system.spawn(server);
auto mport = system.middleman().publish(serv, 0); auto port = system.middleman().publish(serv, 0);
CAF_REQUIRE(mport); CAF_REQUIRE(port != 0);
auto port = *mport;
CAF_MESSAGE("published server at port " << port); CAF_MESSAGE("published server at port " << port);
std::thread child([=] { run_client(argc, argv, port); }); std::thread child([=] { run_client(argc, argv, port); });
child.join(); child.join();
......
...@@ -110,9 +110,8 @@ void run_server(int argc, char** argv) { ...@@ -110,9 +110,8 @@ void run_server(int argc, char** argv) {
.add_message_type<ping>("ping") .add_message_type<ping>("ping")
.add_message_type<pong>("pong"); .add_message_type<pong>("pong");
actor_system system{cfg}; actor_system system{cfg};
auto mport = system.middleman().publish(system.spawn(server), 0, "127.0.0.1"); auto port = system.middleman().publish(system.spawn(server), 0, "127.0.0.1");
CAF_REQUIRE(mport); CAF_REQUIRE(port != 0);
auto port = *mport;
CAF_MESSAGE("running on port " << port << ", start client"); CAF_MESSAGE("running on port " << port << ", start client");
std::thread child{[=] { run_client(argc, argv, port); }}; std::thread child{[=] { run_client(argc, argv, port); }};
child.join(); child.join();
......
...@@ -95,25 +95,25 @@ CAF_TEST_FIXTURE_SCOPE(unpublish_tests, fixture) ...@@ -95,25 +95,25 @@ CAF_TEST_FIXTURE_SCOPE(unpublish_tests, fixture)
CAF_TEST(unpublishing) { CAF_TEST(unpublishing) {
auto port = system.middleman().publish(testee, 0); auto port = system.middleman().publish(testee, 0);
CAF_REQUIRE(port); CAF_REQUIRE(port != 0);
CAF_MESSAGE("published actor on port " << *port); CAF_MESSAGE("published actor on port " << port);
CAF_MESSAGE("test invalid unpublish"); CAF_MESSAGE("test invalid unpublish");
auto testee2 = system.spawn<dummy>(); auto testee2 = system.spawn<dummy>();
system.middleman().unpublish(testee2, *port); system.middleman().unpublish(testee2, port);
auto x0 = remote_actor("127.0.0.1", *port); auto x0 = remote_actor("127.0.0.1", port);
CAF_CHECK(x0 != testee2); CAF_CHECK(x0 != testee2);
CAF_CHECK(x0 == testee); CAF_CHECK(x0 == testee);
anon_send_exit(testee2, exit_reason::kill); anon_send_exit(testee2, exit_reason::kill);
CAF_MESSAGE("unpublish testee"); CAF_MESSAGE("unpublish testee");
system.middleman().unpublish(testee, *port); system.middleman().unpublish(testee, port);
CAF_MESSAGE("check whether testee is still available via cache"); CAF_MESSAGE("check whether testee is still available via cache");
auto x1 = remote_actor("127.0.0.1", *port); auto x1 = remote_actor("127.0.0.1", port);
CAF_CHECK(x1 && *x1 == testee); CAF_CHECK(x1 && *x1 == testee);
CAF_MESSAGE("fake dead of testee and check if testee becomes unavailable"); CAF_MESSAGE("fake dead of testee and check if testee becomes unavailable");
anon_send(system.middleman().actor_handle(), down_msg{testee.address(), anon_send(system.middleman().actor_handle(), down_msg{testee.address(),
exit_reason::normal}); exit_reason::normal});
// must fail now // must fail now
auto x2 = remote_actor("127.0.0.1", *port); auto x2 = remote_actor("127.0.0.1", port);
CAF_CHECK(! x2); CAF_CHECK(! x2);
} }
......
cmake_minimum_required(VERSION 2.8)
project(caf_tools CXX)
add_custom_target(all_tools)
include_directories(${LIBCAF_INCLUDE_DIRS})
if(${CMAKE_SYSTEM_NAME} MATCHES "Window")
set(WSLIB -lws2_32)
else ()
set(WSLIB)
endif()
macro(add name)
add_executable(${name} ${name}.cpp ${ARGN})
target_link_libraries(${name}
${LD_FLAGS}
${CAF_LIBRARIES}
${PTHREAD_LIBRARIES}
${WSLIB})
install(FILES ${name}.cpp DESTINATION share/caf/tools/${folder})
add_dependencies(${name} all_tools)
endmacro()
add(caf-run)
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <unistd.h>
#include <cstdio>
#include <string>
#include <vector>
#include <sstream>
#include <fstream>
#include <iostream>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
using namespace caf;
using std::cout;
using std::cerr;
using std::endl;
using std::string;
using std::vector;
static constexpr const char base64_tbl[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
"abcdefghijklmnopqrstuvwxyz"
"0123456789+/";
std::string encode_base64(const string& str) {
std::string result;
// consumes three characters from input
auto consume = [&](const char* i) {
int buf[] {
(i[0] & 0xfc) >> 2,
((i[0] & 0x03) << 4) + ((i[1] & 0xf0) >> 4),
((i[1] & 0x0f) << 2) + ((i[2] & 0xc0) >> 6),
i[2] & 0x3f
};
for (auto x : buf)
result += base64_tbl[x];
};
// iterate string in chunks of three characters
auto i = str.begin();
for ( ; std::distance(i, str.end()) >= 3; i += 3)
consume(&(*i));
if (i != str.end()) {
// "fill" string with 0s
char cbuf[] = {0, 0, 0};
std::copy(i, str.end(), cbuf);
consume(cbuf);
// override filled characters (garbage) with '='
for (auto j = result.end() - (3 - (str.size() % 3)); j != result.end(); ++j)
*j = '=';
}
return result;
}
class host_desc {
public:
std::string host;
int cpu_slots;
string opencl_device_ids;
host_desc(std::string host_addr, int slots, string cldevices)
: host(std::move(host_addr)),
cpu_slots(slots),
opencl_device_ids(std::move(cldevices)) {
// nop
}
host_desc(host_desc&&) = default;
host_desc(const host_desc&) = default;
host_desc& operator=(host_desc&&) = default;
host_desc& operator=(const host_desc&) = default;
static void append(vector<host_desc>& xs, const string& line, size_t num) {
vector<string> fields;
split(fields, line, is_any_of(" "), token_compress_on);
if (fields.empty())
return;
host_desc hd;
hd.host = std::move(fields.front());
hd.cpu_slots = 0;
hd.opencl_device_ids = "";
for (auto i = fields.begin() + 1; i != fields.end(); ++i) {
if (starts_with(*i, "device_ids=")) {
hd.opencl_device_ids.assign(std::find(i->begin(), i->end(), '=') + 1,
i->end());
} else if (sscanf(i->c_str(), "slots=%d", &hd.cpu_slots) != 0) {
cerr << "invalid option at line " << num << ": " << *i << endl;
}
}
xs.emplace_back(std::move(hd));
}
private:
host_desc() = default;
};
std::vector<host_desc> read_hostfile(const string& fname) {
std::vector<host_desc> result;
std::ifstream in{fname};
std::string line;
size_t line_num = 0;
while (std::getline(in, line))
host_desc::append(result, line, ++line_num);
return result;
}
bool run_ssh(actor_system& system, const string& wdir,
const string& cmd, const string& host) {
std::cout << "runssh, wdir: " << wdir << " cmd: " << cmd
<< " host: " << host << std::endl;
// pack command before sending it to avoid any issue with shell escaping
string full_cmd = "cd ";
full_cmd += wdir;
full_cmd += '\n';
full_cmd += cmd;
auto packed = encode_base64(full_cmd);
std::ostringstream oss;
oss << "ssh -Y -o ServerAliveInterval=60 " << host
<< " \"echo " << packed << " | base64 --decode | /bin/sh\"";
//return system(oss.str().c_str());
string line;
std::cout << "popen: " << oss.str() << std::endl;
auto fp = popen(oss.str().c_str(), "r");
if (! fp)
return false;
char buf[512];
auto eob = buf + sizeof(buf); // end-of-buf
auto pred = [](char c) { return c == 0 || c == '\n'; };
scoped_actor self{system};
while (fgets(buf, sizeof(buf), fp)) {
auto i = buf;
auto e = std::find_if(i, eob, pred);
line.insert(line.end(), i, e);
while (e != eob && *e != 0) {
aout(self) << line << std::endl;
line.clear();
i = e + 1;
e = std::find_if(i, eob, pred);
line.insert(line.end(), i, e);
}
}
pclose(fp);
std::cout << "host down: " << host << std::endl;
if (! line.empty())
aout(self) << line << std::endl;
return true;
}
void bootstrap(actor_system& system,
const string& wdir,
const host_desc& master,
vector<host_desc> slaves,
const string& cmd,
vector<string> args) {
using io::network::interfaces;
if (! args.empty())
args.erase(args.begin());
scoped_actor self{system};
// open a random port and generate a list of all
// possible addresses slaves can use to connect to us
auto port = system.middleman().publish(self, 0);
// run a slave process at master host if user defined slots > 1 for it
if (master.cpu_slots > 1)
slaves.emplace_back(master.host, master.cpu_slots - 1,
master.opencl_device_ids);
for (auto& slave : slaves) {
using namespace caf::io::network;
// build SSH command and pack it to avoid any issue with shell escaping
std::thread{[=, &system](actor bootstrapper) {
std::ostringstream oss;
oss << cmd;
if (slave.cpu_slots > 0)
oss << " --caf#scheduler.max-threads=" << slave.cpu_slots;
if (! slave.opencl_device_ids.empty())
oss << " --caf#opencl-devices=" << slave.opencl_device_ids;
oss << " --caf#slave-mode"
<< " --caf#slave-name=" << slave.host
<< " --caf#bootstrap-node=";
bool is_first = true;
interfaces::traverse({protocol::ipv4, protocol::ipv6},
[&](const char*, protocol, bool lo, const char* x) {
if (lo)
return;
if (! is_first)
oss << ",";
else
is_first = false;
oss << x << "/" << port;
});
for (auto& arg : args)
oss << " " << arg;
if (! run_ssh(system, wdir, oss.str(), slave.host))
anon_send(bootstrapper, slave.host);
}, actor{self}}.detach();
}
std::string slaveslist;
for (size_t i = 0; i < slaves.size(); ++i) {
self->receive(
[&](ok_atom, const string& host, uint16_t port) {
if (! slaveslist.empty())
slaveslist += ',';
slaveslist += host;
slaveslist += '/';
slaveslist += std::to_string(port);
},
[](const string& node) {
cerr << "unable to launch process via SSH at node " << node << endl;
}
);
}
// run (and wait for) master
std::ostringstream oss;
oss << cmd << " --caf#slave-nodes=" << slaveslist << " " << join(args, " ");
run_ssh(system, wdir, oss.str(), master.host);
}
int main(int argc, char** argv) {
actor_system_config cfg{argc, argv};
if (cfg.cli_helptext_printed)
return 0;
if (cfg.slave_mode)
return cerr << "cannot use slave mode in caf-run tool" << endl, 1;
string hostfile;
std::unique_ptr<char, void (*)(void*)> pwd{getcwd(nullptr, 0), ::free};
string wdir;
auto res = cfg.args_remainder.extract_opts({
{"hostfile", "path to the hostfile", hostfile},
{"wdir", wdir}
});
if (hostfile.empty())
return cerr << "no hostfile specified or hostfile is empty" << endl, 1;
auto& remainder = res.remainder;
if (remainder.empty())
return cerr << "empty command line" << endl, 1;
auto cmd = std::move(remainder.get_as_mutable<std::string>(0));
vector<string> xs;
remainder.drop(1).extract([&](string& x) { xs.emplace_back(std::move(x)); });
auto hosts = read_hostfile(hostfile);
if (hosts.empty())
return cerr << "no valid entry in hostfile" << endl, 1;
actor_system system{cfg};
auto master = hosts.front();
hosts.erase(hosts.begin());
bootstrap(system, (wdir.empty()) ? pwd.get() : wdir.c_str(), master,
std::move(hosts), cmd, xs);
}
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