Commit 97e107db authored by Dominik Charousset's avatar Dominik Charousset

Re-implement and integrate ASIO benchmark

parent 9c4790da
......@@ -485,7 +485,7 @@ if(NOT CAF_NO_UNIT_TESTS)
# add some extra unit testing options when testing ASIO as well
if(CAF_USE_ASIO AND "${suite}" MATCHES "^io_.+$")
add_test(${test_name}_asio ${caf_test} -n -v 5 -s
"${suite}" ${ARGN} -- --use-asio)
"${suite}" ${ARGN} -- "--caf#middleman.network-backend=asio")
endif()
endmacro ()
list(LENGTH suites num_suites)
......
......@@ -64,7 +64,6 @@ set (LIBCAF_CORE_SRCS
src/message_data.cpp
src/message_handler.cpp
src/node_id.cpp
src/parse_config.cpp
src/parse_ini.cpp
src/ref_counted.cpp
src/proxy_registry.cpp
......@@ -72,7 +71,6 @@ set (LIBCAF_CORE_SRCS
src/replies_to.cpp
src/resumable.cpp
src/ripemd_160.cpp
src/run_sub_unit_test.cpp
src/scoped_actor.cpp
src/scoped_execution_unit.cpp
src/serializer.cpp
......
......@@ -361,6 +361,12 @@ public:
return spawn_in_groups<T>({grp}, std::forward<Ts>(xs)...);
}
/// @cond PRIVATE
inline atom_value backend_name() const {
return backend_name_;
}
/// @endcond
private:
template <class T>
void check_invariants() {
......@@ -402,6 +408,7 @@ private:
module_array modules_;
io::middleman* middleman_;
scoped_execution_unit dummy_execution_unit_;
atom_value backend_name_;
};
} // namespace caf
......
......@@ -94,12 +94,10 @@ public:
}
/// Sets the parameter `name` to `val`.
using config_value = variant<std::string, double, int64_t, bool>;
actor_system_config& set(const std::string& name, config_value val);
using config_value = variant<std::string, double, int64_t, bool, atom_value>;
// Config parameters of scheduler.
std::string scheduler_policy;
atom_value scheduler_policy;
size_t scheduler_max_threads;
size_t scheduler_max_throughput;
bool scheduler_enable_profiling;
......@@ -107,6 +105,7 @@ public:
std::string scheduler_profiling_output_file;
// Config parameters of middleman.
atom_value middleman_network_backend;
bool middleman_enable_automatic_connections;
// System parameters that are set while initializing modules.
......
......@@ -30,6 +30,7 @@
#include "caf/group.hpp"
#include "caf/either.hpp"
#include "caf/extend.hpp"
#include "caf/logger.hpp"
#include "caf/channel.hpp"
#include "caf/message.hpp"
#include "caf/node_id.hpp"
......
......@@ -42,7 +42,7 @@ template <size_t Size>
constexpr atom_value atom(char const (&str)[Size]) {
// last character is the NULL terminator
static_assert(Size <= 11, "only 10 characters are allowed");
return static_cast<atom_value>(detail::atom_val(str, 0xF));
return static_cast<atom_value>(detail::atom_val(str));
}
/// Lifts an `atom_value` to a compile-time constant.
......
......@@ -59,6 +59,10 @@ public:
return result;
}
inline std::string operator()(char* cstr) const {
return (*this)(const_cast<const char*>(cstr));
}
inline std::string operator()(const std::string& str) const {
return (*this)(str.c_str());
}
......
......@@ -39,8 +39,8 @@ constexpr unsigned char encoding_table[] = {
// decodes 6bit characters to ASCII
constexpr char decoding_table[] = " 0123456789"
"ABCDEFGHIJKLMNOPQRSTUVWXYZ_"
"abcdefghijklmnopqrstuvwxyz";
"ABCDEFGHIJKLMNOPQRSTUVWXYZ_"
"abcdefghijklmnopqrstuvwxyz";
} // namespace <anonymous>
......@@ -48,7 +48,7 @@ constexpr uint64_t next_interim(uint64_t current, size_t char_code) {
return (current << 6) | encoding_table[(char_code <= 0x7F) ? char_code : 0];
}
constexpr uint64_t atom_val(const char* cstr, uint64_t interim = 0) {
constexpr uint64_t atom_val(const char* cstr, uint64_t interim = 0xF) {
return (*cstr == '\0') ?
interim :
atom_val(cstr + 1,
......
......@@ -20,22 +20,36 @@
#ifndef CAF_DETAIL_PARSE_INI_HPP
#define CAF_DETAIL_PARSE_INI_HPP
#include <string>
#include <istream>
#include <functional>
#include "caf/atom.hpp"
#include "caf/maybe.hpp"
#include "caf/parse_config.hpp"
#include "caf/variant.hpp"
namespace caf {
namespace detail {
/// Parse the given input stream as INI formatted data and calls the consumer
/// with every key-value pair.
/// @param raw_data Input stream of INI formatted text.
/// @param errors Output stream for parser errors.
/// @param consumer Callback consuming generated key-value pairs.
void parse_ini(std::istream& raw_data,
config_consumer consumer,
maybe<std::ostream&> errors = none);
struct parse_ini_t {
/// Denotes a configuration value.
using value = variant<std::string, double, int64_t, bool, atom_value>;
/// Denotes a callback for consuming configuration values.
using config_consumer = std::function<void (size_t, std::string, value&)>;
/// Parse the given input stream as INI formatted data and
/// calls the consumer with every key-value pair.
/// @param raw_data Input stream of INI formatted text.
/// @param errors Output stream for parser errors.
/// @param consumer Callback consuming generated key-value pairs.
void operator()(std::istream& raw_data,
config_consumer consumer,
maybe<std::ostream&> errors = none) const;
};
constexpr parse_ini_t parse_ini = parse_ini_t{};
} // namespace detail
} // namespace caf
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#ifndef CAF_DETAIL_RUN_PROGRAM_HPP
#define CAF_DETAIL_RUN_PROGRAM_HPP
#include <thread>
#include <vector>
#include <string>
#include <initializer_list>
#include "caf/send.hpp"
#include "caf/actor.hpp"
#include "caf/message.hpp"
namespace caf {
namespace detail {
std::thread run_sub_unit_test(caf::actor listener,
const char* path,
int max_runtime,
const char* suite_name,
bool set_asio_option,
std::initializer_list<std::string> args);
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_RUN_PROGRAM_HPP
......@@ -174,6 +174,9 @@ public:
/// Creates a CLI argument without data.
cli_arg(std::string name, std::string text);
/// Creates a CLI argument storing its matched argument in `dest`.
cli_arg(std::string name, std::string text, atom_value& dest);
/// Creates a CLI argument storing its matched argument in `dest`.
cli_arg(std::string name, std::string text, std::string& dest);
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#ifndef CAF_PARSER_CONFIG_HPP
#define CAF_PARSER_CONFIG_HPP
#include <string>
#include <istream>
#include <algorithm>
#include "caf/variant.hpp"
#include "caf/maybe.hpp"
namespace caf {
/// Denotes the format of a configuration file.
enum class config_format {
ini
};
/// Denotes a configuration value.
using config_value = variant<std::string, double, int64_t, bool>;
/// Denotes a callback for config parser implementations.
using config_consumer = std::function<void (std::string, config_value)>;
/// Read configuration from `input_stream` using given `format`.
/// @param input_stream ASCII-formatted configuration.
/// @param format Configuration format such as INI.
/// @param errors Output streams for error messages.
void parse_config(std::istream& input_stream, config_format format,
maybe<std::ostream&> errors = none);
/// Read configuration from `file_name` using given `format` or try to
/// deduce file format automatically if `cf == none`.
/// @param file_name Path to configuration file.
/// @param cf Forces the parser to use a specific file format unless `none`.
/// @param errors Output streams for error messages.
void parse_config(const std::string& file_name,
maybe<config_format> cf = none,
maybe<std::ostream&> errors = none);
} // namespace caf
#endif // CAF_PARSER_CONFIG_HPP
......@@ -58,6 +58,7 @@ actor_system::actor_system(actor_system_config&& cfg)
middleman_(nullptr),
dummy_execution_unit_(this) {
CAF_SET_LOGGER_SYS(this);
backend_name_ = cfg.middleman_network_backend;
for (auto& f : cfg.module_factories_) {
auto ptr = f(*this);
modules_[ptr->id()].reset(ptr);
......@@ -80,12 +81,12 @@ actor_system::actor_system(actor_system_config&& cfg)
profiled_sharing = 0x0102
};
sched_conf sc = stealing;
if (cfg.scheduler_policy == "work-sharing")
if (cfg.scheduler_policy == atom("sharing"))
sc = sharing;
else if (cfg.scheduler_policy != "work-stealing")
std::cerr << "[WARNING] \"" << cfg.scheduler_policy
<< "\" is an unrecognized scheduler pollicy, the "
"actor system will use work-stealing as fallback"
else if (cfg.scheduler_policy != atom("stealing"))
std::cerr << "[WARNING] " << deep_to_string(cfg.scheduler_policy)
<< " is an unrecognized scheduler pollicy, "
"falling back to 'stealing' (i.e. work-stealing)"
<< std::endl;
if (cfg.scheduler_enable_profiling)
sc = static_cast<sched_conf>(sc | profiled);
......
......@@ -21,23 +21,214 @@
#include <limits>
#include <thread>
#include <fstream>
#include "caf/message_builder.hpp"
#include "caf/detail/parse_ini.hpp"
namespace caf {
namespace {
struct type_name_visitor : static_visitor<const char*> {
const char* operator()(const std::string&) const {
return "a string";
}
const char* operator()(const double) const {
return "a double";
}
const char* operator()(const int64_t) const {
return "an integer";
}
const char* operator()(const size_t) const {
return "an unsigned integer";
}
const char* operator()(const bool) const {
return "a boolean";
}
const char* operator()(const atom_value) const {
return "an atom";
}
};
template <class T, class U>
bool assign_config_value(T& x, U& y) {
x = std::move(y);
return true;
}
// 32-bit platforms
template <class T>
inline typename std::enable_if<sizeof(T) == sizeof(uint32_t), bool>::type
unsigned_assign_in_range(T&, int64_t& x) {
return x <= std::numeric_limits<T>::max();
}
// 64-bit platforms
template <class T>
inline typename std::enable_if<sizeof(T) == sizeof(uint64_t), bool>::type
unsigned_assign_in_range(T&, int64_t&) {
return true;
}
bool assign_config_value(size_t& x, int64_t& y) {
if (y < 0 || ! unsigned_assign_in_range(x, y))
return false;
x = std::move(y);
return true;
}
class actor_system_config_reader {
public:
using config_value = actor_system_config::config_value;
using sink = std::function<void (size_t, config_value&)>;
template <class T>
actor_system_config_reader& bind(std::string arg_name, T& storage) {
auto fun = [&storage](size_t ln, config_value& x) {
// the INI parser accepts all integers as int64_t
using cfg_type =
typename std::conditional<
std::is_same<T, size_t>::value,
int64_t,
T
>::type;
if (get<cfg_type>(&x) && assign_config_value(storage, get<cfg_type>(x)))
return;
type_name_visitor tnv;
std::cerr << "error in line " << ln << ": expected "
<< tnv(storage) << " found "
<< apply_visitor(tnv, x) << std::endl;
};
sinks_.emplace(std::move(arg_name), fun);
return *this;
}
void operator()(size_t ln, std::string name, config_value& cv) {
auto i = sinks_.find(name);
if (i == sinks_.end())
std::cerr << "error in line " << ln << ": unrecognized parameter name \""
<< name << "\"";
else
(i->second)(ln, cv);
}
private:
std::map<std::string, sink> sinks_;
};
} // namespace <anonymous>
actor_system_config::actor_system_config() {
// hard coded defaults
scheduler_policy = "work-stealing";
// (1) hard-coded defaults
scheduler_policy = atom("stealing");
scheduler_max_threads = std::max(std::thread::hardware_concurrency(),
unsigned{4});
scheduler_max_throughput = std::numeric_limits<size_t>::max();
scheduler_enable_profiling = false;
scheduler_profiling_ms_resolution = 100;
middleman_network_backend = atom("default");
middleman_enable_automatic_connections = false;
}
actor_system_config::actor_system_config(int, char**)
actor_system_config::actor_system_config(int argc, char** argv)
: actor_system_config() {
// nop
if (argc < 2)
return;
auto args = message_builder(argv, argv + argc).move_to_message();
std::string config_file_name;
args.extract_opts({
{"caf-config-file", "", config_file_name}
});
// (2) content of the INI file overrides hard-coded defaults
if (! config_file_name.empty()) {
std::ifstream ini{config_file_name};
if (ini.good()) {
actor_system_config_reader consumer;
consumer.bind("scheduler.policy", scheduler_policy)
.bind("scheduler.scheduler-max-threads", scheduler_max_threads)
.bind("scheduler.max-throughput", scheduler_max_throughput)
.bind("scheduler.enable-profiling", scheduler_enable_profiling)
.bind("scheduler.profiling-ms-resolution",
scheduler_profiling_ms_resolution)
.bind("scheduler.profiling-output-file",
scheduler_profiling_output_file)
.bind("middleman.network-backend", middleman_network_backend)
.bind("middleman.enable-automatic-connections",
middleman_enable_automatic_connections);
detail::parse_ini(ini, consumer, std::cerr);
}
}
// (3) CLI options override the content of the INI file
auto res = args.extract_opts({
{"caf#scheduler.policy",
"sets the scheduling policy to either 'stealing' (default) or 'sharing'",
scheduler_policy},
{"caf#scheduler.scheduler-max-threads",
"sets a fixed number of worker threads for the scheduler",
scheduler_max_threads},
{"caf#scheduler.max-throughput",
"sets the maximum number of messages an actor consumes before yielding",
scheduler_max_throughput},
{"caf#scheduler.enable-profiling",
"enables or disables profiler output",
scheduler_enable_profiling},
{"caf#scheduler.profiling-ms-resolution",
"sets the rate in ms in which the profiler collects data",
scheduler_profiling_ms_resolution},
{"caf#scheduler.profiling-output-file",
"sets the output file for the profiler",
scheduler_profiling_output_file},
{"caf#middleman.network-backend",
"sets the network backend to either 'default' or 'asio' (if available)",
middleman_network_backend},
{"caf#middleman.enable-automatic-connections",
"enables or disables automatic connection management (off per default)",
middleman_enable_automatic_connections},
{"caf-dump-config", "print config in INI format to stdout"},
{"caf-help", "print this text"}
}, nullptr, true);
using std::cerr;
using std::cout;
using std::endl;
if (res.opts.count("caf-help"))
cout << res.helptext << endl;
auto verify_atom_opt = [](std::initializer_list<atom_value> xs, atom_value& x,
const char* xname) {
if (std::find(xs.begin(), xs.end(), x) == xs.end()) {
cerr << "[WARNING] invalid value for " << xname
<< " defined, falling back to "
<< deep_to_string(*xs.begin()) << endl;
x = *xs.begin();
}
};
verify_atom_opt({atom("default"),
# ifdef CAF_USE_ASIO
atom("asio")
# endif
}, middleman_network_backend, "middleman.network-backend");
verify_atom_opt({atom("stealing"), atom("sharing")},
scheduler_policy, "scheduler.policy ");
if (res.opts.count("caf-dump-config")) {
cout << std::boolalpha
<< "[scheduler]" << endl
<< "policy=" << deep_to_string(scheduler_policy) << endl
<< "max-threads=" << scheduler_max_threads << endl
<< "max-throughput=" << scheduler_max_throughput << endl
<< "enable-profiling=" << scheduler_enable_profiling << endl;
if (scheduler_enable_profiling)
cout << "profiling-ms-resolution="
<< scheduler_profiling_ms_resolution << endl
<< "profiling_output_file="
<< deep_to_string(scheduler_profiling_output_file) << endl;
cout << "[middleman]" << endl
<< "network-backend="
<< deep_to_string(middleman_network_backend) << endl
<< "enable-automatic-connections="
<< deep_to_string(middleman_enable_automatic_connections) << endl;
}
}
actor_system_config&
......@@ -46,12 +237,4 @@ actor_system_config::add_actor_factory(std::string name, actor_factory fun) {
return *this;
}
actor_system_config&
actor_system_config::set(const std::string& name, config_value val) {
// TODO: cover all parameters
if (name == "middleman.enable-automatic-connections" && get<bool>(&val))
middleman_enable_automatic_connections = get<bool>(val);
return *this;
}
} // namespace caf
......@@ -316,6 +316,19 @@ message::cli_arg::cli_arg(std::string nstr, std::string tstr, consumer f)
// nop
}
message::cli_arg::cli_arg(std::string nstr, std::string tstr, atom_value& arg)
: name(std::move(nstr)),
text(std::move(tstr)),
fun([&arg](const std::string& str) -> bool {
if (str.size() <= 10) {
arg = static_cast<atom_value>(detail::atom_val(str.c_str()));
return true;
}
return false;
}) {
// nop
}
message::cli_arg::cli_arg(std::string nstr, std::string tstr, std::string& arg)
: name(std::move(nstr)),
text(std::move(tstr)),
......
......@@ -131,6 +131,12 @@ void node_id::data::stop() {
// nop
}
namespace {
std::atomic<uint8_t> system_id;
} // <anonymous>
// initializes singleton
node_id::data* node_id::data::create_singleton() {
CAF_LOG_TRACE("");
......@@ -143,6 +149,10 @@ node_id::data* node_id::data::create_singleton() {
auto hd_serial_and_mac_addr = join(macs, "") + detail::get_root_uuid();
node_id::host_id_type nid;
detail::ripemd_160(nid, hd_serial_and_mac_addr);
// TODO: redesign network layer, make node_id an opaque type, etc.
// this hack enables multiple actor systems in a single process
// by overriding the last byte in the node ID with the actor system "ID"
nid.back() = system_id.fetch_add(1);
// note: pointer has a ref count of 1 -> implicitly held by detail::singletons
return new node_id::data(detail::get_process_id(), nid);
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/parse_config.hpp"
#include <vector>
#include <fstream>
#include <sstream>
#include <iostream>
#include "caf/atom.hpp"
#include "caf/send.hpp"
#include "caf/actor.hpp"
#include "caf/detail/parse_ini.hpp"
#include "caf/detail/optional_message_visitor.hpp"
namespace caf {
class message_visitor : public static_visitor<message> {
public:
template <class T>
message operator()(T& value) const {
return make_message(std::move(value));
}
};
void parse_config(std::istream& input, config_format format,
maybe<std::ostream&> errors) {
if (! input)
return;
// TODO: fixme
//auto cs = experimental::whereis(atom("ConfigServ"));
actor cs;
auto consume = [&](std::string key, config_value value) {
message_visitor mv;
anon_send(cs, put_atom::value, std::move(key), apply_visitor(mv, value));
};
switch (format) {
case config_format::ini:
detail::parse_ini(input, consume, errors);
}
}
void parse_config(const std::string& file_name,
maybe<config_format> format,
maybe<std::ostream&> errors) {
if (! format) {
// try to detect file format according to file extension
if (file_name.size() < 5)
std::cerr << "filename is to short" << std::endl;
else if (file_name.compare(file_name.size() - 4, 4, ".ini"))
parse_config(file_name, config_format::ini, errors);
else if (errors)
*errors << "error: unknown config file format" << std::endl;
return;
}
std::ifstream input{file_name};
if (! input) {
if (errors)
*errors << "error: unable to open " << file_name << std::endl;
return;
}
parse_config(input, *format, errors);
}
} // namespace caf
......@@ -25,22 +25,27 @@
#include <algorithm>
namespace caf {
namespace detail {
void detail::parse_ini(std::istream& input,
config_consumer consumer,
maybe<std::ostream&> errors) {
void parse_ini_t::operator()(std::istream& input, config_consumer consumer_fun,
maybe<std::ostream&> errors) const {
std::string group;
std::string line;
size_t ln = 0; // line number
// wraps a temporary into an (lvalue) config_value and calls `consumer_fun`
auto consumer = [&](size_t ln, std::string name, value x) {
consumer_fun(ln, std::move(name), x);
};
auto print = [&](const char* category, const char* str) {
if (errors)
*errors << category << " in line " << ln << ": " << str << std::endl;
*errors << category << " INI file line "
<< ln << ": " << str << std::endl;
};
auto print_error = [&](const char* str) {
print("error", str);
print("[ERROR]", str);
};
auto print_warning = [&](const char* str) {
print("warning", str);
print("[WARNING]", str);
};
while (std::getline(input, line)) {
++ln;
......@@ -92,9 +97,31 @@ void detail::parse_ini(std::istream& input,
return tolower(x) == tolower(y);
};
if (std::equal(bov, eol, true_str, icase_eq)) {
consumer(std::move(key), true);
consumer(ln, std::move(key), true);
} else if (std::equal(bov, eol, false_str, icase_eq)) {
consumer(std::move(key), false);
consumer(ln, std::move(key), false);
} else if (*bov == '\'') {
// end-of-atom iterator
auto eoa = eol - 1;
if (bov == eoa) {
print_error("stray '");
continue;
}
if (*eoa != '\'') {
print_error("atom not terminated by '");
continue;
}
++bov; // skip leading '
if (std::distance(bov, eoa) > 10) {
print_error("atoms are not allowed to have more than 10 characters");
continue;
}
// found an atom value, copy it into a null-terminated buffer
char buf[11];
std::copy(bov, eoa, buf);
buf[std::distance(bov, eoa)] = '\0';
atom_value result = atom(buf);
consumer(ln, std::move(key), result);
} else if (*bov == '"') {
// end-of-string iterator
auto eos = eol - 1;
......@@ -134,7 +161,7 @@ void detail::parse_ini(std::istream& input,
}
if (last_char_backslash)
print_warning("trailing quotation mark escaped");
consumer(std::move(key), std::move(result));
consumer(ln, std::move(key), std::move(result));
} else {
bool is_neg = *bov == '-';
if (is_neg && ++bov == eol) {
......@@ -147,7 +174,7 @@ void detail::parse_ini(std::istream& input,
if (e != &*eol)
print_error(err);
else
consumer(std::move(key), is_neg ? -res : res);
consumer(ln, std::move(key), is_neg ? -res : res);
};
auto set_dval = [&] {
char* e;
......@@ -155,7 +182,7 @@ void detail::parse_ini(std::istream& input,
if (e != &*eol)
print_error("invalid value");
else
consumer(std::move(key), is_neg ? -res : res);
consumer(ln, std::move(key), is_neg ? -res : res);
};
if (*bov == '0' && std::distance(bov, eol) > 1)
switch (*(bov + 1)) {
......@@ -181,4 +208,5 @@ void detail::parse_ini(std::istream& input,
}
}
} // namespace detail
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/detail/run_sub_unit_test.hpp"
#include <sstream>
#include <iostream>
#include "caf/config.hpp"
#include "caf/string_algorithms.hpp"
#ifdef CAF_MSVC
# include <windows.h>
#endif
namespace caf {
namespace detail {
#ifndef CAF_WINDOWS
std::thread run_sub_unit_test(actor rc,
const char* cpath,
int max_runtime,
const char* suite_name,
bool set_asio_option,
std::initializer_list<std::string> args) {
using namespace std;
string path = cpath;
replace_all(path, "'", "\\'");
ostringstream oss;
// set path and default options for sub unit tests
oss << "'" << path << "' "
<< "-n -s " << suite_name << " -r " << max_runtime << " --";
for (auto& arg : args)
oss << " " << arg;
if (set_asio_option)
oss << " --use-asio";
oss << " 2>&1";
string cmdstr = oss.str();
return std::thread{ [cmdstr, rc] {
// on FreeBSD, popen() hangs indefinitely in some cases
# ifdef CAF_BSD
system(cmdstr.c_str());
anon_send(rc, "");
# else
string output;
auto fp = popen(cmdstr.c_str(), "r");
if (! fp) {
cerr << "FATAL: command line failed: " << cmdstr
<< endl;
abort();
}
char buf[512];
while (fgets(buf, sizeof(buf), fp)) {
output += buf;
}
pclose(fp);
anon_send(rc, output);
# endif
}};
}
#else
std::thread run_sub_unit_test(actor rc,
const char* cpath,
int max_runtime,
const char* suite_name,
bool set_asio_option,
std::initializer_list<std::string> args) {
std::string path = cpath;
replace_all(path, "'", "\\'");
std::ostringstream oss;
// set path and default options for sub unit tests
oss << path
<< " -n -s " << suite_name << " -r " << max_runtime << " --";
for (auto& arg : args)
oss << " " << arg;
if (set_asio_option)
oss << " --use-asio";
return std::thread([rc](std::string cmdstr) {
STARTUPINFO si;
PROCESS_INFORMATION pi;
ZeroMemory(&si, sizeof(si));
si.cb = sizeof(si);
ZeroMemory(&pi, sizeof(pi));
CreateProcess(nullptr, // no module name
&cmdstr[0], // command line
nullptr, // process handle not inheritable
nullptr, // thread handle not inheritable
false, // no handle inheritance
0, // no creation flags
nullptr, // use parent's environment
nullptr, // use parent's directory
&si,
&pi);
// be a good parent and wait for our little child
WaitForSingleObject(pi.hProcess, INFINITE);
CloseHandle(pi.hProcess);
CloseHandle(pi.hThread);
anon_send(rc, "--- process output on windows not implemented yet ---");
}, oss.str());
}
#endif
} // namespace detail
} // namespace caf
......@@ -22,8 +22,8 @@
#define CAF_SUITE parse_ini
#include "caf/test/unit_test.hpp"
#include <iostream>
#include <sstream>
#include <iostream>
#include "caf/string_algorithms.hpp"
......@@ -88,10 +88,15 @@ public:
}
};
using config_value = detail::parse_ini_t::value;
struct fixture {
actor_system system;
fixture() : system(test::engine::argc(), test::engine::argv()) {
// nop
}
template <class F>
void load_impl(F consumer, const char* str) {
std::stringstream ss;
......@@ -112,7 +117,7 @@ struct fixture {
self->send(config_server, put_atom::value, kvp.first, message{});
}
);
auto consume = [&](std::string key, config_value value) {
auto consume = [&](size_t, std::string key, config_value& value) {
message_visitor mv;
anon_send(config_server, put_atom::value,
std::move(key), apply_visitor(mv, value));
......@@ -121,7 +126,7 @@ struct fixture {
}
void load(const char* str) {
auto consume = [&](std::string key, config_value value) {
auto consume = [&](size_t, std::string key, config_value& value) {
values.emplace(std::move(key), std::move(value));
};
load_impl(consume, str);
......@@ -216,19 +221,19 @@ struct fixture {
}
void check_case3() {
CAF_CHECK(has_error("error in line 2: missing ] at end of line"));
CAF_CHECK(has_error("error in line 3: value outside of a group"));
CAF_CHECK(has_error("error in line 6: no '=' found"));
CAF_CHECK(has_error("error in line 7: line starting with '='"));
CAF_CHECK(has_error("error in line 8: line ends with '='"));
CAF_CHECK(has_error("error in line 9: stray '\"'"));
CAF_CHECK(has_error("error in line 10: string not terminated by '\"'"));
CAF_CHECK(has_error("warning in line 12: trailing quotation mark escaped"));
CAF_CHECK(has_error("error in line 13: '-' is not a number"));
CAF_CHECK(has_error("error in line 14: invalid hex value"));
CAF_CHECK(has_error("error in line 15: invalid binary value"));
CAF_CHECK(has_error("error in line 16: invalid oct value"));
CAF_CHECK(has_error("error in line 17: invalid value"));
CAF_CHECK(has_error("[ERROR] INI file line 2: missing ] at end of line"));
CAF_CHECK(has_error("[ERROR] INI file line 3: value outside of a group"));
CAF_CHECK(has_error("[ERROR] INI file line 6: no '=' found"));
CAF_CHECK(has_error("[ERROR] INI file line 7: line starting with '='"));
CAF_CHECK(has_error("[ERROR] INI file line 8: line ends with '='"));
CAF_CHECK(has_error("[ERROR] INI file line 9: stray '\"'"));
CAF_CHECK(has_error("[ERROR] INI file line 10: string not terminated by '\"'"));
CAF_CHECK(has_error("[WARNING] INI file line 12: trailing quotation mark escaped"));
CAF_CHECK(has_error("[ERROR] INI file line 13: '-' is not a number"));
CAF_CHECK(has_error("[ERROR] INI file line 14: invalid hex value"));
CAF_CHECK(has_error("[ERROR] INI file line 15: invalid binary value"));
CAF_CHECK(has_error("[ERROR] INI file line 16: invalid oct value"));
CAF_CHECK(has_error("[ERROR] INI file line 17: invalid value"));
CAF_CHECK(num_values() == 2);
CAF_CHECK(value_is("test.some-int", 42));
CAF_CHECK(value_is("test.some-string", "hi there!"));
......
......@@ -253,8 +253,8 @@ bool valid(const header& hdr);
/// Size of a BASP header in serialized form
constexpr size_t header_size =
node_id::host_id_size * 2 + sizeof(uint32_t) * 2 +
sizeof(actor_id) * 2 + sizeof(uint32_t) * 2 + sizeof(uint64_t);
node_id::serialized_size * 2 + sizeof(actor_id) * 2
+ sizeof(uint32_t) * 2 + sizeof(uint64_t);
/// Describes an error during forwarding of BASP messages.
enum class error : uint64_t {
......
......@@ -242,7 +242,6 @@ public:
/// Returns a middleman using the default network backend.
static actor_system::module* make(actor_system&, detail::type_list<>);
/// Returns a middleman using `bg` as network backend.
template <class Backend>
static actor_system::module* make(actor_system& sys,
detail::type_list<Backend>) {
......
......@@ -32,6 +32,7 @@ CAF_POP_WARNINGS
#include "caf/io/receive_policy.hpp"
#include "caf/io/network/multiplexer.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/io/network/stream_manager.hpp"
#include "caf/io/network/acceptor_manager.hpp"
......@@ -43,13 +44,10 @@ namespace network {
using io_service = boost::asio::io_service;
/// Low-level socket type used as default.
using default_socket = boost::asio::ip::tcp::socket;
using asio_tcp_socket = boost::asio::ip::tcp::socket;
/// Low-level socket type used as default.
using default_socket_acceptor = boost::asio::ip::tcp::acceptor;
/// Platform-specific native socket type.
using native_socket = typename default_socket::native_handle_type;
using asio_tcp_socket_acceptor = boost::asio::ip::tcp::acceptor;
/// A wrapper for the boost::asio multiplexer
class asio_multiplexer : public multiplexer {
......@@ -74,12 +72,14 @@ public:
void assign_tcp_doorman(abstract_broker*, accept_handle hdl) override;
accept_handle add_tcp_doorman(abstract_broker*, default_socket_acceptor&& sock);
accept_handle add_tcp_doorman(abstract_broker*,
asio_tcp_socket_acceptor&& sock);
accept_handle add_tcp_doorman(abstract_broker*, native_socket fd) override;
std::pair<accept_handle, uint16_t>
add_tcp_doorman(abstract_broker*, uint16_t port, const char* in, bool rflag) override;
add_tcp_doorman(abstract_broker*, uint16_t port,
const char* in, bool rflag) override;
void exec_later(resumable* ptr) override;
......@@ -101,8 +101,8 @@ private:
io_service service_;
std::mutex mtx_sockets_;
std::mutex mtx_acceptors_;
std::map<int64_t, default_socket> unassigned_sockets_;
std::map<int64_t, default_socket_acceptor> unassigned_acceptors_;
std::map<int64_t, asio_tcp_socket> unassigned_sockets_;
std::map<int64_t, asio_tcp_socket_acceptor> unassigned_acceptors_;
};
template <class T>
......@@ -123,7 +123,7 @@ using manager_ptr = intrusive_ptr<manager>;
/// A stream capable of both reading and writing. The stream's input
/// data is forwarded to its {@link stream_manager manager}.
template <class Socket>
class stream {
class asio_stream {
public:
/// A smart pointer to a stream manager.
using manager_ptr = intrusive_ptr<stream_manager>;
......@@ -131,7 +131,7 @@ public:
/// A buffer class providing a compatible interface to `std::vector`.
using buffer_type = std::vector<char>;
stream(asio_multiplexer& ref)
asio_stream(asio_multiplexer& ref)
: writing_(false),
fd_(ref.service()),
backend_(ref) {
......@@ -170,7 +170,7 @@ public:
/// Copies data to the write buffer.
/// @note Not thread safe.
void write(const void* buf, size_t num_bytes) {
CAF_LOG_TRACE("num_bytes: " << num_bytes);
CAF_LOG_TRACE(CAF_ARG(num_bytes));
auto first = reinterpret_cast<const char*>(buf);
auto last = first + num_bytes;
wr_offline_buf_.insert(wr_offline_buf_.end(), first, last);
......@@ -306,7 +306,7 @@ private:
/// An acceptor is responsible for accepting incoming connections.
template <class SocketAcceptor>
class acceptor {
class asio_acceptor {
using protocol_type = typename SocketAcceptor::protocol_type;
using socket_type = boost::asio::basic_stream_socket<protocol_type>;
......@@ -317,7 +317,7 @@ public:
/// A smart pointer to an acceptor manager.
using manager_ptr = intrusive_ptr<manager_type>;
acceptor(asio_multiplexer& am, io_service& io)
asio_acceptor(asio_multiplexer& am, io_service& io)
: backend_(am),
accept_fd_(io),
fd_(io) {
......
......@@ -45,9 +45,9 @@ private:
} // namespace anonymous
default_socket new_tcp_connection(io_service& backend, const std::string& host,
asio_tcp_socket new_tcp_connection(io_service& backend, const std::string& host,
uint16_t port) {
default_socket fd{backend};
asio_tcp_socket fd{backend};
using boost::asio::ip::tcp;
tcp::resolver r(fd.get_io_service());
tcp::resolver::query q(host, std::to_string(port));
......@@ -64,27 +64,29 @@ default_socket new_tcp_connection(io_service& backend, const std::string& host,
return fd;
}
void ip_bind(default_socket_acceptor& fd, uint16_t port,
const char* addr = nullptr, bool reuse_addr = true) {
CAF_LOG_TRACE(CAF_ARG(port));
void ip_bind(asio_tcp_socket_acceptor& fd, uint16_t port,
const char* addr, bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(reuse_addr));
using boost::asio::ip::tcp;
try {
auto bind_and_listen = [&](tcp::endpoint& ep) {
CAF_LOG_DEBUG("created IP endpoint:" << CAF_ARG(ep.address().to_string())
<< CAF_ARG(ep.port()));
fd.open(ep.protocol());
fd.set_option(tcp::acceptor::reuse_address(reuse_addr));
if (reuse_addr)
fd.set_option(tcp::acceptor::reuse_address(reuse_addr));
fd.bind(ep);
fd.listen();
};
if (addr) {
CAF_LOG_DEBUG(CAF_ARG(addr));
tcp::endpoint ep(boost::asio::ip::address::from_string(addr), port);
CAF_LOG_DEBUG("got 'em");
bind_and_listen(ep);
CAF_LOG_DEBUG("created IPv6 endpoint: " << ep.address() << ":"
<< fd.local_endpoint().port());
} else {
CAF_LOG_DEBUG("addr = nullptr");
tcp::endpoint ep(tcp::v6(), port);
bind_and_listen(ep);
CAF_LOG_DEBUG("created IPv6 endpoint: " << ep.address() << ":"
<< fd.local_endpoint().port());
}
} catch (boost::system::system_error& se) {
if (se.code() == boost::system::errc::address_in_use) {
......@@ -96,7 +98,7 @@ void ip_bind(default_socket_acceptor& fd, uint16_t port,
connection_handle asio_multiplexer::new_tcp_scribe(const std::string& host,
uint16_t port) {
default_socket fd{new_tcp_connection(service(), host, port)};
asio_tcp_socket fd{new_tcp_connection(service(), host, port)};
auto id = int64_from_native_socket(fd.native_handle());
std::lock_guard<std::mutex> lock(mtx_sockets_);
unassigned_sockets_.insert(std::make_pair(id, std::move(fd)));
......@@ -162,7 +164,7 @@ connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self,
}
private:
bool launched_;
stream<Socket> stream_;
asio_stream<Socket> stream_;
};
auto ptr = make_counted<impl>(self, *this, std::move(sock));
self->add_scribe(ptr);
......@@ -173,7 +175,7 @@ connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self,
native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(self) << ", " << CAF_ARG(fd));
boost::system::error_code ec;
default_socket sock{service()};
asio_tcp_socket sock{service()};
sock.assign(boost::asio::ip::tcp::v6(), fd, ec);
if (ec) {
sock.assign(boost::asio::ip::tcp::v4(), fd, ec);
......@@ -194,7 +196,7 @@ connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self,
std::pair<accept_handle, uint16_t>
asio_multiplexer::new_tcp_doorman(uint16_t port, const char* in, bool rflag) {
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (in ? in : "nullptr"));
default_socket_acceptor fd{service()};
asio_tcp_socket_acceptor fd{service()};
ip_bind(fd, port, in, rflag);
auto id = int64_from_native_socket(fd.native_handle());
auto assigned_port = fd.local_endpoint().port();
......@@ -203,7 +205,8 @@ asio_multiplexer::new_tcp_doorman(uint16_t port, const char* in, bool rflag) {
return {accept_handle::from_int(id), assigned_port};
}
void asio_multiplexer::assign_tcp_doorman(abstract_broker* self, accept_handle hdl) {
void asio_multiplexer::assign_tcp_doorman(abstract_broker* self,
accept_handle hdl) {
CAF_LOG_TRACE("");
std::lock_guard<std::mutex> lock(mtx_acceptors_);
auto itr = unassigned_acceptors_.find(hdl.id());
......@@ -215,12 +218,12 @@ void asio_multiplexer::assign_tcp_doorman(abstract_broker* self, accept_handle h
accept_handle
asio_multiplexer::add_tcp_doorman(abstract_broker* self,
default_socket_acceptor&& sock) {
CAF_LOG_TRACE("sock.fd = " << sock.native_handle());
asio_tcp_socket_acceptor&& sock) {
CAF_LOG_TRACE(CAF_ARG(sock.native_handle()));
CAF_ASSERT(sock.native_handle() != network::invalid_native_socket);
class impl : public doorman {
public:
impl(abstract_broker* ptr, default_socket_acceptor&& s,
impl(abstract_broker* ptr, asio_tcp_socket_acceptor&& s,
network::asio_multiplexer& am)
: doorman(ptr, network::accept_hdl_from_socket(s)),
acceptor_(am, s.get_io_service()) {
......@@ -250,7 +253,7 @@ asio_multiplexer::add_tcp_doorman(abstract_broker* self,
return acceptor_.socket_handle().local_endpoint().port();
}
private:
network::acceptor<default_socket_acceptor> acceptor_;
network::asio_acceptor<asio_tcp_socket_acceptor> acceptor_;
};
auto ptr = make_counted<impl>(self, std::move(sock), *this);
self->add_doorman(ptr);
......@@ -259,7 +262,8 @@ asio_multiplexer::add_tcp_doorman(abstract_broker* self,
accept_handle asio_multiplexer::add_tcp_doorman(abstract_broker* self,
native_socket fd) {
default_socket_acceptor sock{service()};
CAF_LOG_TRACE(CAF_ARG(fd));
asio_tcp_socket_acceptor sock{service()};
boost::system::error_code ec;
sock.assign(boost::asio::ip::tcp::v6(), fd, ec);
if (ec) {
......@@ -272,9 +276,10 @@ accept_handle asio_multiplexer::add_tcp_doorman(abstract_broker* self,
}
std::pair<accept_handle, uint16_t>
asio_multiplexer::add_tcp_doorman(abstract_broker* self, uint16_t port, const char* in,
bool rflag) {
default_socket_acceptor fd{service()};
asio_multiplexer::add_tcp_doorman(abstract_broker* self, uint16_t port,
const char* in, bool rflag) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(in) << CAF_ARG(rflag));
asio_tcp_socket_acceptor fd{service()};
ip_bind(fd, port, in, rflag);
auto p = fd.local_endpoint().port();
return {add_tcp_doorman(self, std::move(fd)), p};
......
......@@ -52,10 +52,15 @@
#include "caf/actor_registry.hpp"
#include "caf/detail/get_mac_addresses.hpp"
#ifdef CAF_USE_ASIO
#include "caf/io/network/asio_multiplexer.hpp"
#include "caf/io/network/asio_multiplexer_impl.hpp"
#endif // CAF_USE_ASIO
#ifdef CAF_WINDOWS
# include <io.h>
# include <fcntl.h>
#endif
#include <io.h>
#include <fcntl.h>
#endif // CAF_WINDOWS
namespace caf {
namespace io {
......@@ -176,6 +181,23 @@ actor_system::module* middleman::make(actor_system& sys, detail::type_list<>) {
private:
network::default_multiplexer backend_;
};
# ifdef CAF_USE_ASIO
class asio_impl : public middleman {
public:
asio_impl(actor_system& ref) : middleman(ref), backend_(&ref) {
// nop
}
network::multiplexer& backend() override {
return backend_;
}
private:
network::asio_multiplexer backend_;
};
if (sys.backend_name() == atom("asio"))
return new asio_impl(sys);
# endif // CAF_USE_ASIO
return new impl(sys);
}
......@@ -267,11 +289,16 @@ actor_addr middleman::remote_actor(std::set<std::string> ifs,
[&](ok_atom, const node_id&, actor_addr res, std::set<std::string>& xs) {
CAF_LOG_TRACE(CAF_ARG(res) << CAF_ARG(xs));
if (!res)
throw network_error("no actor published at given port");
throw network_error("no actor published at port "
+ std::to_string(port));
if (! (xs.empty() && ifs.empty())
&& ! std::includes(xs.begin(), xs.end(), ifs.begin(), ifs.end()))
throw network_error("expected signature does not "
"comply to found signature");
&& ! std::includes(xs.begin(), xs.end(), ifs.begin(), ifs.end())) {
std::string what = "expected signature: ";
what += deep_to_string(xs);
what += ", found: ";
what += deep_to_string(ifs);
throw network_error(std::move(what));
}
result = std::move(res);
},
[&](error_atom, std::string& msg) {
......
......@@ -32,12 +32,6 @@
#include "caf/io/network/interfaces.hpp"
#include "caf/io/network/test_multiplexer.hpp"
#include "caf/detail/run_sub_unit_test.hpp"
#ifdef CAF_USE_ASIO
#include "caf/io/network/asio_multiplexer.hpp"
#endif // CAF_USE_ASIO
using namespace caf;
using namespace caf::io;
......
......@@ -30,12 +30,6 @@
#include "caf/string_algorithms.hpp"
#include "caf/detail/run_sub_unit_test.hpp"
#ifdef CAF_USE_ASIO
#include "caf/io/network/asio_multiplexer.hpp"
#endif // CAF_USE_ASIO
using namespace std;
using namespace caf;
using namespace caf::io;
......@@ -168,37 +162,37 @@ behavior peer_acceptor_fun(broker* self, const actor& buddy) {
};
}
void run_server(actor_system& system, bool spawn_client,
const char* bin_path, bool use_asio) {
void run_client(int argc, char** argv, uint16_t port) {
actor_system system{actor_system_config{argc, argv}.load<io::middleman>()};
auto p = system.spawn(ping, size_t{10});
CAF_MESSAGE("spawn_client...");
auto cl = system.middleman().spawn_client(peer_fun, "127.0.0.1", port, p);
CAF_REQUIRE(cl);
CAF_MESSAGE("spawn_client finished");
anon_send(p, kickoff_atom::value, *cl);
CAF_MESSAGE("`kickoff_atom` has been send");
system.await_all_actors_done();
}
void run_server(int argc, char** argv) {
actor_system system{actor_system_config{argc, argv}.load<io::middleman>()};
scoped_actor self{system};
CAF_MESSAGE("spawn peer acceptor");
auto serv = system.middleman().spawn_broker(peer_acceptor_fun,
system.spawn(pong));
std::thread child;
self->sync_send(serv, publish_atom::value).await(
[&](uint16_t port) {
CAF_MESSAGE("server is running on port " << port);
if (spawn_client) {
auto child = detail::run_sub_unit_test(self,
bin_path,
test::engine::max_runtime(),
CAF_XSTR(CAF_SUITE),
use_asio,
{"--client-port="
+ std::to_string(port)});
child.join();
self->receive(
[](const std::string& output) {
cout << endl << endl << "*** output of client program ***"
<< endl << output << endl;
}
);
}
child = std::thread([=] { run_client(argc, argv, port); });
},
others >> [&] {
CAF_TEST_ERROR("unexpected message: "
<< to_string(self->current_message()));
}
);
self->await_all_other_actors_done();
child.join();
}
} // namespace <anonymous>
......@@ -206,38 +200,5 @@ void run_server(actor_system& system, bool spawn_client,
CAF_TEST(test_broker) {
auto argc = test::engine::argc();
auto argv = test::engine::argv();
uint16_t port = 0;
auto r = message_builder(argv, argv + argc).extract_opts({
{"client-port,c", "set port for IO client", port},
{"server,s", "run in server mode"},
{"use-asio", "use ASIO network backend (if available)"}
});
if (! r.error.empty() || r.opts.count("help") > 0 || ! r.remainder.empty()) {
cout << r.error << endl << endl << r.helptext << endl;
return;
}
auto use_asio = r.opts.count("use-asio") > 0;
actor_system_config cfg;
# ifdef CAF_USE_ASIO
if (use_asio)
cfg.load<io::middleman, io::network::asio_multiplexer>();
else
# endif // CAF_USE_ASIO
cfg.load<io::middleman>();
actor_system system{cfg};
if (r.opts.count("client-port") > 0) {
auto p = system.spawn(ping, size_t{10});
CAF_MESSAGE("spawn_client...");
auto cl = system.middleman().spawn_client(peer_fun, "localhost", port, p);
CAF_REQUIRE(cl);
CAF_MESSAGE("spawn_client finished");
anon_send(p, kickoff_atom::value, *cl);
CAF_MESSAGE("`kickoff_atom` has been send");
} else if (r.opts.count("server") > 0) {
// run in server mode
run_server(system, false, "", use_asio);
} else {
run_server(system, true, test::engine::path(), use_asio);
}
system.await_all_actors_done();
run_server(argc, argv);
}
......@@ -181,7 +181,8 @@ class fixture {
public:
fixture() : system(actor_system_config{}
.load<io::middleman, network::test_multiplexer>()) {
mpx_ = static_cast<network::test_multiplexer*>(&system.middleman().backend());
mpx_ = dynamic_cast<network::test_multiplexer*>(&system.middleman().backend());
CAF_REQUIRE(mpx_ != nullptr);
// spawn the actor-under-test
aut_ = system.middleman().spawn_broker(server);
// assign the acceptor handle to the AUT
......
......@@ -32,13 +32,6 @@
#include "caf/all.hpp"
#include "caf/io/all.hpp"
#include "caf/logger.hpp"
#include "caf/detail/run_sub_unit_test.hpp"
#ifdef CAF_USE_ASIO
#include "caf/io/network/asio_multiplexer.hpp"
#endif // CAF_USE_ASIO
using namespace std;
using namespace caf;
......@@ -509,41 +502,60 @@ private:
bool run_in_loop_;
};
void test_remote_actor(actor_system& system, const char* path,
bool run_remote, bool use_asio) {
CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(run_remote) << CAF_ARG(use_asio));
void launch_remote_side(int argc, char** argv, uint16_t group_port,
uint16_t client_port1, uint16_t client_port2) {
actor_system_config cfg{argc, argv};
cfg.load<io::middleman>()
.add_message_type<actor_vector>("actor_vector");
CAF_LOG_TRACE(CAF_ARG(group_port) << CAF_ARG(client_port1)
<< CAF_ARG(client_port2));
CAF_MESSAGE("launch_remote_side(" << group_port
<< ", " << client_port1 << ", " << client_port2 << ")");
actor_system system{cfg};
scoped_actor self{system, true};
auto serv = system.middleman().remote_actor("127.0.0.1", client_port1);
auto serv2 = system.middleman().remote_actor("127.0.0.1", client_port2);
CAF_REQUIRE(serv);
// remote_actor is supposed to return the same server
// when connecting to the same host again
CAF_CHECK(serv == system.middleman().remote_actor("127.0.0.1", client_port1));
CAF_CHECK(serv2 == system.middleman().remote_actor("127.0.0.1", client_port2));
// connect to published groups
auto grp = system.middleman().remote_group("whatever", "127.0.0.1", group_port);
auto c = self->spawn<client, monitored>(serv);
self->receive(
[&](const down_msg& dm) {
CAF_CHECK(dm.source == c);
CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
}
);
system.await_all_actors_done();
}
void test_remote_actor(int argc, char** argv) {
actor_system_config cfg{argc, argv};
cfg.load<io::middleman>()
.add_message_type<actor_vector>("actor_vector");
actor_system system{cfg};
scoped_actor self{system, true};
auto serv = self->spawn<server, monitored>(! run_remote);
auto serv = self->spawn<server, monitored>();
// publish on two distinct ports and use the latter one afterwards
auto port1 = system.middleman().publish(serv, 0, "127.0.0.1");
CAF_CHECK(port1 && *port1 > 0);
CAF_REQUIRE(port1 && *port1 > 0);
CAF_MESSAGE("first publish succeeded on port " << *port1);
auto port2 = system.middleman().publish(serv, 0, "127.0.0.1");
CAF_CHECK(port2 && *port2 > 0);
CAF_REQUIRE(port2 && *port2 > 0);
CAF_MESSAGE("second publish succeeded on port " << *port2);
// publish local groups as well
auto gport = system.middleman().publish_local_groups(0);
CAF_CHECK(gport && *gport > 0);
CAF_REQUIRE(gport && *gport > 0);
CAF_MESSAGE("local groups published on port " << *gport);
// check whether accessing local actors via system.middleman().remote_actors
// works correctly, i.e., does not return a proxy instance
auto serv2 = system.middleman().remote_actor("127.0.0.1", *port2);
CAF_CHECK(serv2 && system.node() != serv2->node());
CAF_CHECK(serv == serv2);
thread child;
if (run_remote) {
CAF_MESSAGE("start child process");
child = detail::run_sub_unit_test(self,
path,
test::engine::max_runtime(),
CAF_XSTR(CAF_SUITE),
use_asio,
{"--client-port=" + std::to_string(*port2),
"--client-port=" + std::to_string(*port1),
"--group-port=" + std::to_string(*gport)});
} else {
CAF_MESSAGE("please run client with: "
<< "-c " << port2 << " -c " << port1 << " -g " << gport);
}
launch_remote_side(argc, argv, *gport, *port1, *port2);
self->receive(
[&](const down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm));
......@@ -553,16 +565,6 @@ void test_remote_actor(actor_system& system, const char* path,
);
CAF_MESSAGE("wait for other actors");
self->await_all_other_actors_done();
if (run_remote) {
CAF_MESSAGE("wait for child process");
child.join();
self->receive(
[](const std::string& output) {
cout << endl << endl << "*** output of client program ***"
<< endl << output << endl;
}
);
}
}
} // namespace <anonymous>
......@@ -570,65 +572,10 @@ void test_remote_actor(actor_system& system, const char* path,
CAF_TEST(remote_actors) {
auto argv = test::engine::argv();
auto argc = test::engine::argc();
std::vector<uint16_t> ports;
uint16_t gport = 0;
auto r = message_builder(argv, argv + argc).extract_opts({
{"server,s", "run in server mode"},
{"client-port,c", "add client port (two needed)", ports},
{"group-port,g", "set group port", gport},
{"use-asio", "use ASIO network backend (if available)"}
});
if (! r.error.empty() || r.opts.count("help") > 0 || ! r.remainder.empty()) {
cout << r.error << endl << endl << r.helptext << endl;
return;
}
actor_system_config cfg;
cfg.add_message_type<actor_vector>("actor_vector");
auto use_asio = r.opts.count("use-asio") > 0;
# ifdef CAF_USE_ASIO
if (use_asio)
cfg.load<io::middleman, io::network::asio_multiplexer>();
else
# endif // CAF_USE_ASIO
cfg.load<io::middleman>();
{
actor_system system{cfg};
if (r.opts.count("server") > 0) {
CAF_MESSAGE("don't run remote actor (server mode)");
test_remote_actor(system, argv[0], false, use_asio);
} else if (r.opts.count("client-port") > 0) {
if (ports.size() != 2 || r.opts.count("group-port") == 0) {
cerr << "*** expected exactly two ports and one group port" << endl
<< endl << r.helptext << endl;
return;
}
scoped_actor self{system, true};
auto serv = system.middleman().remote_actor("localhost", ports.front());
auto serv2 = system.middleman().remote_actor("localhost", ports.back());
CAF_REQUIRE(serv);
// remote_actor is supposed to return the same server
// when connecting to the same host again
{
CAF_CHECK(serv == system.middleman().remote_actor("localhost", ports.front()));
CAF_CHECK(serv2 == system.middleman().remote_actor("127.0.0.1", ports.back()));
}
// connect to published groups
auto grp = system.middleman().remote_group("whatever", "127.0.0.1", gport);
auto c = self->spawn<client, monitored>(serv);
self->receive(
[&](const down_msg& dm) {
CAF_CHECK(dm.source == c);
CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
}
);
} else {
test_remote_actor(system, test::engine::path(), true, use_asio);
}
system.await_all_actors_done();
}
test_remote_actor(argc, argv);
// check after dtor of actor system ran
// we either spawn a server or a client, in both cases
// there must have been exactly one dtor called
CAF_CHECK_EQUAL(s_destructors_called.load(), 1);
CAF_CHECK_EQUAL(s_on_exit_called.load(), 1);
// we spawn a server and a client, in both cases
// there must have been exactly one dtor called (two in total)
CAF_CHECK_EQUAL(s_destructors_called.load(), 2);
CAF_CHECK_EQUAL(s_on_exit_called.load(), 2);
}
......@@ -32,12 +32,6 @@
#include "caf/all.hpp"
#include "caf/io/all.hpp"
#include "caf/detail/run_sub_unit_test.hpp"
#ifdef CAF_USE_ASIO
#include "caf/io/network/asio_multiplexer.hpp"
#endif // CAF_USE_ASIO
using namespace caf;
namespace {
......@@ -100,54 +94,33 @@ behavior server(stateful_actor<server_state>* self) {
};
}
} // namespace <anonymous>
CAF_TEST(remote_spawn) {
auto argv = test::engine::argv();
auto argc = test::engine::argc();
uint16_t port = 0;
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},
{"port,p", "force a port in server mode", 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;
}
actor_system_config cfg;
cfg.add_actor_type("mirror", mirror);
auto use_asio = r.opts.count("use-asio") > 0;
# ifdef CAF_USE_ASIO
if (use_asio)
cfg.load<io::middleman, io::network::asio_multiplexer>();
else
# endif // CAF_USE_ASIO
cfg.load<io::middleman>();
void run_client(int argc, char** argv, uint16_t port) {
actor_system_config cfg{argc, argv};
cfg.load<io::middleman>()
.add_actor_type("mirror", mirror);
actor_system system{cfg};
if (r.opts.count("client") > 0) {
auto serv = system.middleman().remote_actor("localhost", port);
CAF_REQUIRE(serv);
system.spawn(client, serv);
system.await_all_actors_done();
return;
}
auto serv = system.middleman().remote_actor("localhost", port);
CAF_REQUIRE(serv);
system.spawn(client, serv);
system.await_all_actors_done();
}
void run_server(int argc, char** argv) {
actor_system system{actor_system_config{argc, argv}.load<io::middleman>()};
auto serv = system.spawn(server);
auto mport = system.middleman().publish(serv, port);
auto mport = system.middleman().publish(serv, 0);
CAF_REQUIRE(mport);
port = *mport;
auto port = *mport;
CAF_MESSAGE("published server at port " << port);
if (r.opts.count("server") == 0) {
CAF_MESSAGE("run client program");
auto child = detail::run_sub_unit_test(invalid_actor,
test::engine::path(),
test::engine::max_runtime(),
CAF_XSTR(CAF_SUITE),
use_asio,
{"--client="
+ std::to_string(port)});
child.join();
}
std::thread child([=] { run_client(argc, argv, port); });
system.await_all_actors_done();
child.join();
}
} // namespace <anonymous>
CAF_TEST(remote_spawn) {
auto argc = test::engine::argc();
auto argv = test::engine::argv();
run_server(argc, argv);
}
......@@ -32,12 +32,6 @@
#include "caf/string_algorithms.hpp"
#include "caf/detail/run_sub_unit_test.hpp"
#ifdef CAF_USE_ASIO
#include "caf/io/network/asio_multiplexer.hpp"
#endif // CAF_USE_ASIO
using namespace std;
using namespace caf;
using namespace caf::io;
......@@ -187,73 +181,40 @@ acceptor::behavior_type acceptor_fun(acceptor::broker_pointer self,
};
}
void run_server(actor_system& system, bool spawn_client,
const char* bin_path, bool use_asio) {
void run_client(int argc, char** argv, uint16_t port) {
actor_system system{actor_system_config{argc, argv}.load<io::middleman>()};
auto p = system.spawn(ping, size_t{10});
CAF_MESSAGE("spawn_client_typed...");
auto cl = system.middleman().spawn_client(peer_fun, "localhost", port, p);
CAF_REQUIRE(cl);
CAF_MESSAGE("spawn_client_typed finished");
anon_send(p, kickoff_atom::value, *cl);
CAF_MESSAGE("`kickoff_atom` has been send");
system.await_all_actors_done();
}
void run_server(int argc, char** argv) {
actor_system system{actor_system_config{argc, argv}.load<io::middleman>()};
scoped_actor self{system};
auto serv = system.middleman().spawn_broker(acceptor_fun, system.spawn(pong));
std::thread child;
self->sync_send(serv, publish_atom::value).await(
[&](uint16_t port) {
CAF_MESSAGE("server is running on port " << port);
if (spawn_client) {
auto child = detail::run_sub_unit_test(self,
bin_path,
test::engine::max_runtime(),
CAF_XSTR(CAF_SUITE),
use_asio,
{"--client-port="
+ std::to_string(port)});
CAF_MESSAGE("block till child process has finished");
child.join();
}
child = std::thread([=] {
run_client(argc, argv, port);
});
}
);
self->await_all_other_actors_done();
self->receive(
[](const std::string& output) {
cout << endl << endl << "*** output of client program ***"
<< endl << output << endl;
}
);
CAF_MESSAGE("wait for client system");
child.join();
}
} // namespace <anonymous>
CAF_TEST(test_typed_broker) {
auto argv = test::engine::argv();
auto argc = test::engine::argc();
uint16_t port = 0;
auto r = message_builder(argv, argv + argc).extract_opts({
{"client-port,c", "set port for IO client", port},
{"server,s", "run in server mode"},
{"use-asio", "use ASIO network backend (if available)"}
});
if (! r.error.empty() || r.opts.count("help") > 0 || ! r.remainder.empty()) {
cout << r.error << endl << endl << r.helptext << endl;
return;
}
actor_system_config cfg;
auto use_asio = r.opts.count("use-asio") > 0;
# ifdef CAF_USE_ASIO
if (use_asio)
cfg.load<io::middleman, io::network::asio_multiplexer>();
else
# endif // CAF_USE_ASIO
cfg.load<io::middleman>();
actor_system system{cfg};
if (r.opts.count("client-port") > 0) {
auto p = system.spawn(ping, size_t{10});
CAF_MESSAGE("spawn_client_typed...");
auto cl = system.middleman().spawn_client(peer_fun, "localhost", port, p);
CAF_REQUIRE(cl);
CAF_MESSAGE("spawn_client_typed finished");
anon_send(p, kickoff_atom::value, *cl);
CAF_MESSAGE("`kickoff_atom` has been send");
} else if (r.opts.count("server") > 0) {
// run in server mode
run_server(system, false, argv[0], use_asio);
} else {
run_server(system, true, test::engine::path(), use_asio);
}
CAF_MESSAGE("block on `await_all_actors_done`");
system.await_all_actors_done();
auto argv = test::engine::argv();
run_server(argc, argv);
}
......@@ -32,12 +32,6 @@
#include "caf/all.hpp"
#include "caf/io/all.hpp"
#include "caf/detail/run_sub_unit_test.hpp"
#ifdef CAF_USE_ASIO
#include "caf/io/network/asio_multiplexer.hpp"
#endif // CAF_USE_ASIO
using namespace std;
using namespace caf;
......@@ -80,18 +74,24 @@ server_type::behavior_type server() {
};
}
void run_client(actor_system& system, const char* host, uint16_t port) {
void run_client(int argc, char** argv, uint16_t port) {
actor_system_config cfg{argc, argv};
cfg.load<io::middleman>()
.add_message_type<ping>("ping")
.add_message_type<pong>("pong");
actor_system system{cfg};
// check whether invalid_argument is thrown
// when trying to connect to get an untyped
// handle to the server
try {
system.middleman().remote_actor(host, port);
system.middleman().remote_actor("127.0.0.1", port);
}
catch (network_error& e) {
CAF_MESSAGE(e.what());
}
CAF_MESSAGE("connect to typed_remote_actor");
auto serv = system.middleman().typed_remote_actor<server_type>(host, port);
auto serv = system.middleman().typed_remote_actor<server_type>("127.0.0.1",
port);
CAF_REQUIRE(serv);
scoped_actor self{system};
self->sync_send(serv, ping{42})
......@@ -104,64 +104,22 @@ void run_client(actor_system& system, const char* host, uint16_t port) {
});
}
uint16_t run_server(actor_system& system) {
auto port = system.middleman().publish(system.spawn(server), 0, "127.0.0.1");
CAF_REQUIRE(port);
CAF_MESSAGE("running on port " << *port);
return *port;
void run_server(int argc, char** argv) {
actor_system_config cfg{argc, argv};
cfg.load<io::middleman>()
.add_message_type<ping>("ping")
.add_message_type<pong>("pong");
actor_system system{cfg};
auto mport = system.middleman().publish(system.spawn(server), 0, "127.0.0.1");
CAF_REQUIRE(mport);
auto port = *mport;
CAF_MESSAGE("running on port " << port << ", start client");
std::thread child{[=] { run_client(argc, argv, port); }};
child.join();
}
CAF_TEST(test_typed_remote_actor) {
auto argv = test::engine::argv();
auto argc = test::engine::argc();
uint16_t port = 0;
auto r = message_builder(argv, argv + argc).extract_opts({
{"client-port,c", "set port for client", port},
{"server,s", "run in server mode"},
{"use-asio", "use ASIO network backend (if available)"}
});
if (! r.error.empty() || r.opts.count("help") > 0 || ! r.remainder.empty()) {
cout << r.error << endl << endl << r.helptext << endl;
return;
}
actor_system_config cfg;
cfg.add_message_type<ping>("ping")
.add_message_type<pong>("pong");
auto use_asio = r.opts.count("use-asio") > 0;
# ifdef CAF_USE_ASIO
if (use_asio)
cfg.load<io::middleman, io::network::asio_multiplexer>();
else
# endif // CAF_USE_ASIO
cfg.load<io::middleman>();
actor_system system{cfg};
if (r.opts.count("client-port") > 0) {
CAF_MESSAGE("run in client mode");
run_client(system, "localhost", port);
} else if (r.opts.count("server") > 0) {
CAF_MESSAGE("run in server mode");
run_server(system);
} else {
port = run_server(system);
// execute client_part() in a separate process,
// connected via localhost socket
scoped_actor self{system};
auto child = detail::run_sub_unit_test(self,
test::engine::path(),
test::engine::max_runtime(),
CAF_XSTR(CAF_SUITE),
use_asio,
{"--client-port="
+ std::to_string(port)});
CAF_MESSAGE("block till child process has finished");
child.join();
self->await_all_other_actors_done();
self->receive(
[](const std::string& output) {
cout << endl << endl << "*** output of client program ***"
<< endl << output << endl;
}
);
}
system.await_all_actors_done();
auto argv = test::engine::argv();
run_server(argc, argv);
}
......@@ -28,10 +28,6 @@
#include "caf/all.hpp"
#include "caf/io/all.hpp"
#ifdef CAF_USE_ASIO
#include "caf/io/network/asio_multiplexer.hpp"
#endif // CAF_USE_ASIO
using namespace caf;
namespace {
......@@ -69,17 +65,10 @@ void test_invalid_unpublish(actor_system& system, const actor& published,
}
CAF_TEST(unpublishing) {
actor_system_config cfg;
# ifdef CAF_USE_ASIO
auto argc = test::engine::argc();
auto argv = test::engine::argv();
if (argc == 1 && strcmp(argv[0], "--use-asio") == 0)
cfg.load<io::middleman, io::network::asio_multiplexer>();
else
# endif // CAF_USE_ASIO
cfg.load<io::middleman>();
{ // scope for local variables
actor_system system{cfg};
actor_system system{actor_system_config{argc, argv}.load<io::middleman>()};
auto d = system.spawn<dummy>();
auto port = system.middleman().publish(d, 0);
CAF_REQUIRE(port);
......
......@@ -551,10 +551,16 @@ int main(int argc, char** argv) {
return 1;
}
auto colorize = res.opts.count("no-colors") == 0;
if (divider < argc)
engine::args(argc - divider - 1, argv + divider + 1);
else
engine::args(0, argv);
std::vector<char*> args;
if (divider < argc) {
// make a new args vector that contains argv[0] and all remaining args
args.push_back(argv[0]);
for (int i = divider + 1; i < argc; ++i)
args.push_back(argv[i]);
engine::args(static_cast<int>(args.size()), args.data());
} else {
engine::args(1, argv);
}
if (res.opts.count("max-runtime") > 0)
engine::max_runtime(max_runtime);
auto result = engine::run(colorize, log_file, verbosity_console,
......
......@@ -18,7 +18,3 @@
******************************************************************************/
#include "caf/test/unit_test_impl.hpp"
#ifdef CAF_USE_ASIO
#include "caf/io/network/asio_multiplexer_impl.hpp"
#endif // CAF_USE_ASIO
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