Commit d4df45ed authored by Jakob Otto's avatar Jakob Otto

Remove all opencl references

parent 25dc1ce3
...@@ -5,7 +5,6 @@ ...@@ -5,7 +5,6 @@
// Default CMake flags for release builds. // Default CMake flags for release builds.
defaultReleaseBuildFlags = [ defaultReleaseBuildFlags = [
'CAF_ENABLE_RUNTIME_CHECKS:BOOL=yes', 'CAF_ENABLE_RUNTIME_CHECKS:BOOL=yes',
'CAF_NO_OPENCL:BOOL=yes',
] ]
// Default CMake flags for debug builds. // Default CMake flags for debug builds.
...@@ -112,7 +111,6 @@ config = [ ...@@ -112,7 +111,6 @@ config = [
'libcaf_core/test', 'libcaf_core/test',
'libcaf_io/test', 'libcaf_io/test',
'libcaf_openssl/test', 'libcaf_openssl/test',
'libcaf_opencl',
], ],
], ],
] ]
......
...@@ -153,7 +153,6 @@ public: ...@@ -153,7 +153,6 @@ public:
enum id_t { enum id_t {
scheduler, scheduler,
middleman, middleman,
opencl_manager,
openssl_manager, openssl_manager,
network_manager, network_manager,
num_ids num_ids
...@@ -270,13 +269,6 @@ public: ...@@ -270,13 +269,6 @@ public:
/// @throws `std::logic_error` if module is not loaded. /// @throws `std::logic_error` if module is not loaded.
io::middleman& middleman(); io::middleman& middleman();
/// Returns `true` if the opencl module is available, `false` otherwise.
bool has_opencl_manager() const;
/// Returns the manager instance from the OpenCL module.
/// @throws `std::logic_error` if module is not loaded.
opencl::manager& opencl_manager() const;
/// Returns `true` if the openssl module is available, `false` otherwise. /// Returns `true` if the openssl module is available, `false` otherwise.
bool has_openssl_manager() const; bool has_openssl_manager() const;
...@@ -593,7 +585,7 @@ private: ...@@ -593,7 +585,7 @@ private:
void check_invariants() { void check_invariants() {
static_assert(!std::is_base_of<prohibit_top_level_spawn_marker, T>::value, static_assert(!std::is_base_of<prohibit_top_level_spawn_marker, T>::value,
"This actor type cannot be spawned through an actor system. " "This actor type cannot be spawned through an actor system. "
"Probably you have tried to spawn a broker or opencl actor."); "Probably you have tried to spawn a broker.");
} }
expected<strong_actor_ptr> expected<strong_actor_ptr>
......
...@@ -275,10 +275,6 @@ public: ...@@ -275,10 +275,6 @@ public:
/// @private /// @private
timespan stream_tick_duration() const noexcept; timespan stream_tick_duration() const noexcept;
// -- OpenCL parameters ------------------------------------------------------
std::string opencl_device_ids;
// -- OpenSSL parameters ----------------------------------------------------- // -- OpenSSL parameters -----------------------------------------------------
std::string openssl_certificate; std::string openssl_certificate;
......
...@@ -241,14 +241,6 @@ class middleman; ...@@ -241,14 +241,6 @@ class middleman;
} // namespace net } // namespace net
// -- OpenCL classes -----------------------------------------------------------
namespace opencl {
class manager;
} // namespace opencl
// -- scheduler classes -------------------------------------------------------- // -- scheduler classes --------------------------------------------------------
namespace scheduler { namespace scheduler {
......
...@@ -31,9 +31,9 @@ ...@@ -31,9 +31,9 @@
#include "caf/policy/work_sharing.hpp" #include "caf/policy/work_sharing.hpp"
#include "caf/policy/work_stealing.hpp" #include "caf/policy/work_stealing.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/scheduler/coordinator.hpp" #include "caf/scheduler/coordinator.hpp"
#include "caf/scheduler/test_coordinator.hpp" #include "caf/scheduler/test_coordinator.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
namespace caf { namespace caf {
...@@ -87,8 +87,8 @@ behavior config_serv_impl(stateful_actor<kvstate>* self) { ...@@ -87,8 +87,8 @@ behavior config_serv_impl(stateful_actor<kvstate>* self) {
// also iterate all subscribers for '*' // also iterate all subscribers for '*'
for (auto& subscriber : self->state.data[wildcard].second) for (auto& subscriber : self->state.data[wildcard].second)
if (subscriber != self->current_sender()) if (subscriber != self->current_sender())
self->send(actor_cast<actor>(subscriber), update_atom::value, self->send(actor_cast<actor>(subscriber), update_atom::value, key,
key, vp.first); vp.first);
}, },
// get a key/value pair // get a key/value pair
[=](get_atom, std::string& key) -> message { [=](get_atom, std::string& key) -> message {
...@@ -101,9 +101,9 @@ behavior config_serv_impl(stateful_actor<kvstate>* self) { ...@@ -101,9 +101,9 @@ behavior config_serv_impl(stateful_actor<kvstate>* self) {
return make_message(std::move(msgs)); return make_message(std::move(msgs));
} }
auto i = self->state.data.find(key); auto i = self->state.data.find(key);
return make_message(std::move(key), return make_message(std::move(key), i != self->state.data.end()
i != self->state.data.end() ? i->second.first ? i->second.first
: make_message()); : make_message());
}, },
// subscribe to a key // subscribe to a key
[=](subscribe_atom, const std::string& key) { [=](subscribe_atom, const std::string& key) {
...@@ -137,8 +137,7 @@ behavior config_serv_impl(stateful_actor<kvstate>* self) { ...@@ -137,8 +137,7 @@ behavior config_serv_impl(stateful_actor<kvstate>* self) {
// get a 'named' actor from local registry // get a 'named' actor from local registry
[=](get_atom, atom_value name) { [=](get_atom, atom_value name) {
return self->home_system().registry().get(name); return self->home_system().registry().get(name);
} }};
};
} }
// -- spawn server ------------------------------------------------------------- // -- spawn server -------------------------------------------------------------
...@@ -155,15 +154,12 @@ const char* spawn_serv_state::name = "spawn_server"; ...@@ -155,15 +154,12 @@ const char* spawn_serv_state::name = "spawn_server";
behavior spawn_serv_impl(stateful_actor<spawn_serv_state>* self) { behavior spawn_serv_impl(stateful_actor<spawn_serv_state>* self) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
return { return {[=](spawn_atom, const std::string& name, message& args,
[=](spawn_atom, const std::string& name, actor_system::mpi& xs) -> expected<strong_actor_ptr> {
message& args, actor_system::mpi& xs) CAF_LOG_TRACE(CAF_ARG(name) << CAF_ARG(args));
-> expected<strong_actor_ptr> { return self->system().spawn<strong_actor_ptr>(name, std::move(args),
CAF_LOG_TRACE(CAF_ARG(name) << CAF_ARG(args)); self->context(), true, &xs);
return self->system().spawn<strong_actor_ptr>(name, std::move(args), }};
self->context(), true, &xs);
}
};
} }
// -- stream server ------------------------------------------------------------ // -- stream server ------------------------------------------------------------
...@@ -204,8 +200,6 @@ const char* actor_system::module::name() const noexcept { ...@@ -204,8 +200,6 @@ const char* actor_system::module::name() const noexcept {
return "Scheduler"; return "Scheduler";
case middleman: case middleman:
return "Middleman"; return "Middleman";
case opencl_manager:
return "OpenCL Manager";
case openssl_manager: case openssl_manager:
return "OpenSSL Manager"; return "OpenSSL Manager";
case network_manager: case network_manager:
...@@ -244,9 +238,9 @@ actor_system::actor_system(actor_system_config& cfg) ...@@ -244,9 +238,9 @@ actor_system::actor_system(actor_system_config& cfg)
// set scheduler only if not explicitly loaded by user // set scheduler only if not explicitly loaded by user
if (!sched) { if (!sched) {
enum sched_conf { enum sched_conf {
stealing = 0x0001, stealing = 0x0001,
sharing = 0x0002, sharing = 0x0002,
testing = 0x0003, testing = 0x0003,
}; };
sched_conf sc = stealing; sched_conf sc = stealing;
namespace sr = defaults::scheduler; namespace sr = defaults::scheduler;
...@@ -377,17 +371,6 @@ io::middleman& actor_system::middleman() { ...@@ -377,17 +371,6 @@ io::middleman& actor_system::middleman() {
return *reinterpret_cast<io::middleman*>(clptr->subtype_ptr()); return *reinterpret_cast<io::middleman*>(clptr->subtype_ptr());
} }
bool actor_system::has_opencl_manager() const {
return modules_[module::opencl_manager] != nullptr;
}
opencl::manager& actor_system::opencl_manager() const {
auto& clptr = modules_[module::opencl_manager];
if (!clptr)
CAF_RAISE_ERROR("cannot access opencl manager: module not loaded");
return *reinterpret_cast<opencl::manager*>(clptr->subtype_ptr());
}
bool actor_system::has_openssl_manager() const { bool actor_system::has_openssl_manager() const {
return modules_[module::openssl_manager] != nullptr; return modules_[module::openssl_manager] != nullptr;
} }
...@@ -461,7 +444,7 @@ actor_system::dyn_spawn_impl(const std::string& name, message& args, ...@@ -461,7 +444,7 @@ actor_system::dyn_spawn_impl(const std::string& name, message& args,
execution_unit* ctx, bool check_interface, execution_unit* ctx, bool check_interface,
optional<const mpi&> expected_ifs) { optional<const mpi&> expected_ifs) {
CAF_LOG_TRACE(CAF_ARG(name) << CAF_ARG(args) << CAF_ARG(check_interface) CAF_LOG_TRACE(CAF_ARG(name) << CAF_ARG(args) << CAF_ARG(check_interface)
<< CAF_ARG(expected_ifs)); << CAF_ARG(expected_ifs));
if (name.empty()) if (name.empty())
return sec::invalid_argument; return sec::invalid_argument;
auto& fs = cfg_.actor_factories; auto& fs = cfg_.actor_factories;
......
...@@ -18,10 +18,10 @@ ...@@ -18,10 +18,10 @@
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include <limits>
#include <thread>
#include <fstream> #include <fstream>
#include <limits>
#include <sstream> #include <sstream>
#include <thread>
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/config_option.hpp" #include "caf/config_option.hpp"
...@@ -39,7 +39,7 @@ namespace { ...@@ -39,7 +39,7 @@ namespace {
constexpr const char* default_config_file = "caf-application.ini"; constexpr const char* default_config_file = "caf-application.ini";
} // namespace anonymous } // namespace
actor_system_config::~actor_system_config() { actor_system_config::~actor_system_config() {
// nop // nop
...@@ -49,10 +49,10 @@ actor_system_config::~actor_system_config() { ...@@ -49,10 +49,10 @@ actor_system_config::~actor_system_config() {
// 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), : cli_helptext_printed(false),
slave_mode(false), slave_mode(false),
config_file_path(default_config_file), config_file_path(default_config_file),
slave_mode_fun(nullptr) { slave_mode_fun(nullptr) {
// add `vector<T>` and `stream<T>` for each statically known type // add `vector<T>` and `stream<T>` for each statically known type
add_message_type_impl<stream<actor>>("stream<@actor>"); add_message_type_impl<stream<actor>>("stream<@actor>");
add_message_type_impl<stream<actor_addr>>("stream<@addr>"); add_message_type_impl<stream<actor_addr>>("stream<@addr>");
...@@ -75,7 +75,8 @@ actor_system_config::actor_system_config() ...@@ -75,7 +75,8 @@ actor_system_config::actor_system_config()
.add<bool>("help,h?", "print help text to STDERR and exit") .add<bool>("help,h?", "print help text to STDERR and exit")
.add<bool>("long-help", "print long help text to STDERR and exit") .add<bool>("long-help", "print long help text to STDERR and exit")
.add<bool>("dump-config", "print configuration to STDERR and exit") .add<bool>("dump-config", "print configuration to STDERR and exit")
.add<string>(config_file_path, "config-file", "set config file (default: caf-application.ini)"); .add<string>(config_file_path, "config-file",
"set config file (default: caf-application.ini)");
opt_group{custom_options_, "stream"} opt_group{custom_options_, "stream"}
.add<timespan>(stream_desired_batch_complexity, "desired-batch-complexity", .add<timespan>(stream_desired_batch_complexity, "desired-batch-complexity",
"processing time per batch") "processing time per batch")
...@@ -132,8 +133,6 @@ actor_system_config::actor_system_config() ...@@ -132,8 +133,6 @@ actor_system_config::actor_system_config()
.add<bool>("manual-multiplexing", .add<bool>("manual-multiplexing",
"disables background activity of the multiplexer") "disables background activity of the multiplexer")
.add<size_t>("workers", "number of deserialization workers"); .add<size_t>("workers", "number of deserialization workers");
opt_group(custom_options_, "opencl")
.add<std::vector<size_t>>("device-ids", "whitelist for OpenCL devices");
opt_group(custom_options_, "openssl") opt_group(custom_options_, "openssl")
.add<string>(openssl_certificate, "certificate", .add<string>(openssl_certificate, "certificate",
"path to the PEM-formatted certificate file") "path to the PEM-formatted certificate file")
...@@ -212,7 +211,7 @@ settings actor_system_config::dump_content() const { ...@@ -212,7 +211,7 @@ settings actor_system_config::dump_content() const {
put_missing(middleman_group, "heartbeat-interval", put_missing(middleman_group, "heartbeat-interval",
defaults::middleman::heartbeat_interval); defaults::middleman::heartbeat_interval);
put_missing(middleman_group, "workers", defaults::middleman::workers); put_missing(middleman_group, "workers", defaults::middleman::workers);
// -- opencl parameters // -- openssl parameters
auto& openssl_group = result["openssl"].as_dictionary(); auto& openssl_group = result["openssl"].as_dictionary();
put_missing(openssl_group, "certificate", std::string{}); put_missing(openssl_group, "certificate", std::string{});
put_missing(openssl_group, "key", std::string{}); put_missing(openssl_group, "key", std::string{});
...@@ -265,7 +264,7 @@ struct ini_iter { ...@@ -265,7 +264,7 @@ struct ini_iter {
} }
}; };
struct ini_sentinel { }; struct ini_sentinel {};
bool operator!=(ini_iter iter, ini_sentinel) { bool operator!=(ini_iter iter, ini_sentinel) {
return !iter.ini->fail(); return !iter.ini->fail();
...@@ -373,8 +372,8 @@ actor_system_config::add_error_category(atom_value x, error_renderer y) { ...@@ -373,8 +372,8 @@ actor_system_config::add_error_category(atom_value x, error_renderer y) {
return *this; return *this;
} }
actor_system_config& actor_system_config::set_impl(string_view name, actor_system_config&
config_value value) { actor_system_config::set_impl(string_view name, config_value value) {
if (name == "middleman.app-identifier") { if (name == "middleman.app-identifier") {
// TODO: Print a warning with 0.18 and remove this code with 0.19. // TODO: Print a warning with 0.18 and remove this code with 0.19.
value.convert_to_list(); value.convert_to_list();
...@@ -415,8 +414,8 @@ std::string actor_system_config::render(const error& err) { ...@@ -415,8 +414,8 @@ std::string actor_system_config::render(const error& err) {
return "unknown-error"; return "unknown-error";
} }
std::string actor_system_config::render_sec(uint8_t x, atom_value, std::string
const message& xs) { actor_system_config::render_sec(uint8_t x, atom_value, const message& xs) {
auto tmp = static_cast<sec>(x); auto tmp = static_cast<sec>(x);
return deep_to_string(meta::type_name("system_error"), tmp, return deep_to_string(meta::type_name("system_error"), tmp,
meta::omittable_if_empty(), xs); meta::omittable_if_empty(), xs);
...@@ -429,8 +428,8 @@ std::string actor_system_config::render_exit_reason(uint8_t x, atom_value, ...@@ -429,8 +428,8 @@ std::string actor_system_config::render_exit_reason(uint8_t x, atom_value,
meta::omittable_if_empty(), xs); meta::omittable_if_empty(), xs);
} }
std::string actor_system_config::render_pec(uint8_t x, atom_value, std::string
const message& xs) { actor_system_config::render_pec(uint8_t x, atom_value, const message& xs) {
auto tmp = static_cast<pec>(x); auto tmp = static_cast<pec>(x);
return deep_to_string(meta::type_name("parser_error"), tmp, return deep_to_string(meta::type_name("parser_error"), tmp,
meta::omittable_if_empty(), xs); meta::omittable_if_empty(), xs);
......
...@@ -19,19 +19,19 @@ ...@@ -19,19 +19,19 @@
#include <unistd.h> #include <unistd.h>
#include <cstdio> #include <cstdio>
#include <string>
#include <vector>
#include <sstream>
#include <fstream> #include <fstream>
#include <iostream> #include <iostream>
#include <sstream>
#include <string>
#include <vector>
#include "caf/all.hpp" #include "caf/all.hpp"
#include "caf/io/all.hpp" #include "caf/io/all.hpp"
using namespace caf; using namespace caf;
using std::cout;
using std::cerr; using std::cerr;
using std::cout;
using std::endl; using std::endl;
using std::string; using std::string;
using std::vector; using std::vector;
...@@ -44,18 +44,14 @@ std::string encode_base64(const string& str) { ...@@ -44,18 +44,14 @@ std::string encode_base64(const string& str) {
std::string result; std::string result;
// consumes three characters from input // consumes three characters from input
auto consume = [&](const char* i) { auto consume = [&](const char* i) {
int buf[] { int buf[]{(i[0] & 0xfc) >> 2, ((i[0] & 0x03) << 4) + ((i[1] & 0xf0) >> 4),
(i[0] & 0xfc) >> 2, ((i[1] & 0x0f) << 2) + ((i[2] & 0xc0) >> 6), i[2] & 0x3f};
((i[0] & 0x03) << 4) + ((i[1] & 0xf0) >> 4),
((i[1] & 0x0f) << 2) + ((i[2] & 0xc0) >> 6),
i[2] & 0x3f
};
for (auto x : buf) for (auto x : buf)
result += base64_tbl[x]; result += base64_tbl[x];
}; };
// iterate string in chunks of three characters // iterate string in chunks of three characters
auto i = str.begin(); auto i = str.begin();
for ( ; std::distance(i, str.end()) >= 3; i += 3) for (; std::distance(i, str.end()) >= 3; i += 3)
consume(&(*i)); consume(&(*i));
if (i != str.end()) { if (i != str.end()) {
// "fill" string with 0s // "fill" string with 0s
...@@ -73,12 +69,9 @@ class host_desc { ...@@ -73,12 +69,9 @@ class host_desc {
public: public:
std::string host; std::string host;
int cpu_slots; int cpu_slots;
string opencl_device_ids;
host_desc(std::string host_addr, int slots, string cldevices) host_desc(std::string host_addr, int slots, string cldevices)
: host(std::move(host_addr)), : host(std::move(host_addr)), cpu_slots(slots) {
cpu_slots(slots),
opencl_device_ids(std::move(cldevices)) {
// nop // nop
} }
...@@ -95,15 +88,6 @@ public: ...@@ -95,15 +88,6 @@ public:
host_desc hd; host_desc hd;
hd.host = std::move(fields.front()); hd.host = std::move(fields.front());
hd.cpu_slots = 0; 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)); xs.emplace_back(std::move(hd));
} }
...@@ -121,10 +105,10 @@ std::vector<host_desc> read_hostfile(const string& fname) { ...@@ -121,10 +105,10 @@ std::vector<host_desc> read_hostfile(const string& fname) {
return result; return result;
} }
bool run_ssh(actor_system& system, const string& wdir, bool run_ssh(actor_system& system, const string& wdir, const string& cmd,
const string& cmd, const string& host) { const string& host) {
std::cout << "runssh, wdir: " << wdir << " cmd: " << cmd std::cout << "runssh, wdir: " << wdir << " cmd: " << cmd << " host: " << host
<< " host: " << host << std::endl; << std::endl;
// pack command before sending it to avoid any issue with shell escaping // pack command before sending it to avoid any issue with shell escaping
string full_cmd = "cd "; string full_cmd = "cd ";
full_cmd += wdir; full_cmd += wdir;
...@@ -132,9 +116,9 @@ bool run_ssh(actor_system& system, const string& wdir, ...@@ -132,9 +116,9 @@ bool run_ssh(actor_system& system, const string& wdir,
full_cmd += cmd; full_cmd += cmd;
auto packed = encode_base64(full_cmd); auto packed = encode_base64(full_cmd);
std::ostringstream oss; std::ostringstream oss;
oss << "ssh -Y -o ServerAliveInterval=60 " << host oss << "ssh -Y -o ServerAliveInterval=60 " << host << R"( "echo )" << packed
<< R"( "echo )" << packed << R"( | base64 --decode | /bin/sh")"; << R"( | base64 --decode | /bin/sh")";
//return system(oss.str().c_str()); // return system(oss.str().c_str());
string line; string line;
std::cout << "popen: " << oss.str() << std::endl; std::cout << "popen: " << oss.str() << std::endl;
auto fp = popen(oss.str().c_str(), "r"); auto fp = popen(oss.str().c_str(), "r");
...@@ -163,12 +147,9 @@ bool run_ssh(actor_system& system, const string& wdir, ...@@ -163,12 +147,9 @@ bool run_ssh(actor_system& system, const string& wdir,
return true; return true;
} }
void bootstrap(actor_system& system, void bootstrap(actor_system& system, const string& wdir,
const string& wdir, const host_desc& master, vector<host_desc> slaves,
const host_desc& master, const string& cmd, vector<string> args) {
vector<host_desc> slaves,
const string& cmd,
vector<string> args) {
using io::network::interfaces; using io::network::interfaces;
if (!args.empty()) if (!args.empty())
args.erase(args.begin()); args.erase(args.begin());
...@@ -184,38 +165,37 @@ void bootstrap(actor_system& system, ...@@ -184,38 +165,37 @@ void bootstrap(actor_system& system,
auto port = *port_res; auto port = *port_res;
// run a slave process at master host if user defined slots > 1 for it // run a slave process at master host if user defined slots > 1 for it
if (master.cpu_slots > 1) if (master.cpu_slots > 1)
slaves.emplace_back(master.host, master.cpu_slots - 1, slaves.emplace_back(master.host, master.cpu_slots - 1);
master.opencl_device_ids);
for (auto& slave : slaves) { for (auto& slave : slaves) {
using namespace caf::io::network; using namespace caf::io::network;
// build SSH command and pack it to avoid any issue with shell escaping // build SSH command and pack it to avoid any issue with shell escaping
std::thread{[=, &system](actor bootstrapper) { std::thread{[=, &system](actor bootstrapper) {
std::ostringstream oss; std::ostringstream oss;
oss << cmd; oss << cmd;
if (slave.cpu_slots > 0) if (slave.cpu_slots > 0)
oss << " --caf#scheduler.max-threads=" << slave.cpu_slots; oss << " --caf#scheduler.max-threads=" << slave.cpu_slots;
if (!slave.opencl_device_ids.empty()) oss << " --caf#slave-mode"
oss << " --caf#opencl-devices=" << slave.opencl_device_ids; << " --caf#slave-name=" << slave.host
oss << " --caf#slave-mode" << " --caf#bootstrap-node=";
<< " --caf#slave-name=" << slave.host bool is_first = true;
<< " --caf#bootstrap-node="; interfaces::traverse({protocol::ipv4, protocol::ipv6},
bool is_first = true; [&](const char*, protocol::network,
interfaces::traverse( bool lo, const char* x) {
{protocol::ipv4, protocol::ipv6}, if (lo)
[&](const char*, protocol::network, bool lo, const char* x) { return;
if (lo) if (!is_first)
return; oss << ",";
if (!is_first) else
oss << ","; is_first = false;
else oss << x << "/" << port;
is_first = false; });
oss << x << "/" << port; for (auto& arg : args)
}); oss << " " << arg;
for (auto& arg : args) if (!run_ssh(system, wdir, oss.str(), slave.host))
oss << " " << arg; anon_send(bootstrapper, slave.host);
if (!run_ssh(system, wdir, oss.str(), slave.host)) },
anon_send(bootstrapper, slave.host); actor{self}}
}, actor{self}}.detach(); .detach();
} }
std::string slaveslist; std::string slaveslist;
for (size_t i = 0; i < slaves.size(); ++i) { for (size_t i = 0; i < slaves.size(); ++i) {
...@@ -229,8 +209,7 @@ void bootstrap(actor_system& system, ...@@ -229,8 +209,7 @@ void bootstrap(actor_system& system,
}, },
[](const string& node) { [](const string& node) {
cerr << "unable to launch process via SSH at node " << node << endl; cerr << "unable to launch process via SSH at node " << node << endl;
} });
);
} }
// run (and wait for) master // run (and wait for) master
std::ostringstream oss; std::ostringstream oss;
...@@ -249,8 +228,8 @@ namespace { ...@@ -249,8 +228,8 @@ namespace {
struct config : actor_system_config { struct config : actor_system_config {
config() { config() {
opt_group{custom_options_, "global"} opt_group{custom_options_, "global"}
.add(hostfile, "hostfile", "path to hostfile") .add(hostfile, "hostfile", "path to hostfile")
.add(wdir, "wdir", "working directory"); .add(wdir, "wdir", "working directory");
} }
string hostfile; string hostfile;
string wdir; string wdir;
...@@ -274,7 +253,7 @@ int main(int argc, char** argv) { ...@@ -274,7 +253,7 @@ int main(int argc, char** argv) {
auto cmd = std::move(remainder.front()); auto cmd = std::move(remainder.front());
vector<string> xs; vector<string> xs;
for (auto i = cfg.remainder.begin() + 1; i != cfg.remainder.end(); ++i) for (auto i = cfg.remainder.begin() + 1; i != cfg.remainder.end(); ++i)
xs.emplace_back(std::move(*i)); xs.emplace_back(std::move(*i));
auto hosts = read_hostfile(cfg.hostfile); auto hosts = read_hostfile(cfg.hostfile);
if (hosts.empty()) if (hosts.empty())
RETURN_WITH_ERROR("no valid entry in hostfile"); RETURN_WITH_ERROR("no valid entry in hostfile");
......
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