Commit d7101bdc authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'master' into issue/814

parents fd2fd0b1 b6a2aefd
...@@ -32,7 +32,6 @@ set(LIBCAF_CORE_SRCS ...@@ -32,7 +32,6 @@ set(LIBCAF_CORE_SRCS
src/behavior_impl.cpp src/behavior_impl.cpp
src/behavior_stack.cpp src/behavior_stack.cpp
src/binary_deserializer.cpp src/binary_deserializer.cpp
src/binary_serializer.cpp
src/blocking_actor.cpp src/blocking_actor.cpp
src/blocking_behavior.cpp src/blocking_behavior.cpp
src/chars.cpp src/chars.cpp
......
...@@ -56,15 +56,12 @@ namespace caf { ...@@ -56,15 +56,12 @@ namespace caf {
template <class T> template <class T>
struct mpi_field_access { struct mpi_field_access {
std::string operator()(const uniform_type_info_map& types) { std::string operator()(const uniform_type_info_map& types) {
auto nr = type_nr<T>::value; auto result = types.portable_name(type_nr<T>::value, &typeid(T));
if (nr != 0) if (result == types.default_type_name()) {
return *types.portable_name(nr, nullptr); result = "<invalid-type[typeid ";
auto ptr = types.portable_name(0, &typeid(T)); result += typeid(T).name();
if (ptr != nullptr) result += "]>";
return *ptr; }
std::string result = "<invalid-type[typeid ";
result += typeid(T).name();
result += "]>";
return result; return result;
} }
}; };
......
...@@ -106,6 +106,10 @@ public: ...@@ -106,6 +106,10 @@ public:
/// @private /// @private
settings content; settings content;
/// Extracts all parameters from the config, including entries with default
/// values.
virtual settings dump_content() const;
/// Sets a config by using its INI name `config_name` to `config_value`. /// Sets a config by using its INI name `config_name` to `config_value`.
template <class T> template <class T>
actor_system_config& set(string_view name, T&& value) { actor_system_config& set(string_view name, T&& value) {
...@@ -350,7 +354,7 @@ private: ...@@ -350,7 +354,7 @@ private:
error adjust_content(); error adjust_content();
}; };
/// @private /// Returns all user-provided configuration parameters.
const settings& content(const actor_system_config& cfg); const settings& content(const actor_system_config& cfg);
/// Tries to retrieve the value associated to `name` from `cfg`. /// Tries to retrieve the value associated to `name` from `cfg`.
......
...@@ -22,7 +22,9 @@ ...@@ -22,7 +22,9 @@
#include <cstdint> #include <cstdint>
#include <vector> #include <vector>
#include "caf/byte.hpp"
#include "caf/deserializer.hpp" #include "caf/deserializer.hpp"
#include "caf/span.hpp"
namespace caf { namespace caf {
...@@ -33,17 +35,27 @@ public: ...@@ -33,17 +35,27 @@ public:
using super = deserializer; using super = deserializer;
using buffer = std::vector<char>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
binary_deserializer(actor_system& sys, const char* buf, size_t buf_size); binary_deserializer(actor_system& sys, span<const byte> bytes);
binary_deserializer(execution_unit* ctx, const char* buf, size_t buf_size); binary_deserializer(execution_unit* ctx, span<const byte> bytes);
template <class T>
binary_deserializer(actor_system& sys, const std::vector<T>& buf)
: binary_deserializer(sys, as_bytes(make_span(buf))) {
// nop
}
template <class T>
binary_deserializer(execution_unit* ctx, const std::vector<T>& buf)
: binary_deserializer(ctx, as_bytes(make_span(buf))) {
// nop
}
binary_deserializer(actor_system& sys, const buffer& buf); binary_deserializer(actor_system& sys, const char* buf, size_t buf_size);
binary_deserializer(execution_unit* ctx, const buffer& buf); binary_deserializer(execution_unit* ctx, const char* buf, size_t buf_size);
// -- overridden member functions -------------------------------------------- // -- overridden member functions --------------------------------------------
...@@ -60,24 +72,24 @@ public: ...@@ -60,24 +72,24 @@ public:
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
/// Returns the current read position. /// Returns the current read position.
const char* current() const { const char* current() const CAF_DEPRECATED_MSG("use remaining() instead");
return current_;
}
/// Returns the past-the-end iterator. /// Returns the past-the-end iterator.
const char* end() const { const char* end() const CAF_DEPRECATED_MSG("use remaining() instead");
return end_;
}
/// Returns how many bytes are still available to read. /// Returns how many bytes are still available to read.
size_t remaining() const; size_t remaining() const noexcept {
return static_cast<size_t>(end_ - current_);
}
/// Jumps `num_bytes` forward. /// Returns the remaining bytes.
/// @pre `num_bytes <= remaining()` span<const byte> remainder() const noexcept {
void skip(size_t num_bytes) { return make_span(current_, end_);
current_ += num_bytes;
} }
/// Jumps `num_bytes` forward.
/// @pre `num_bytes <= remaining()`
void skip(size_t num_bytes);
protected: protected:
error apply_impl(int8_t&) override; error apply_impl(int8_t&) override;
...@@ -113,8 +125,8 @@ private: ...@@ -113,8 +125,8 @@ private:
return current_ + read_size <= end_; return current_ + read_size <= end_;
} }
const char* current_; const byte* current_;
const char* end_; const byte* end_;
}; };
} // namespace caf } // namespace caf
...@@ -18,97 +18,12 @@ ...@@ -18,97 +18,12 @@
#pragma once #pragma once
#include <cstddef>
#include <cstdint>
#include <vector> #include <vector>
#include "caf/serializer.hpp" #include "caf/serializer_impl.hpp"
namespace caf { namespace caf {
/// Implements the serializer interface with a binary serialization protocol. using binary_serializer = serializer_impl<std::vector<char>>;
class binary_serializer final : public serializer {
public:
// -- member types -----------------------------------------------------------
using super = serializer;
using buffer = std::vector<char>;
// -- constructors, destructors, and assignment operators --------------------
binary_serializer(actor_system& sys, buffer& buf);
binary_serializer(execution_unit* ctx, buffer& buf);
// -- position management ----------------------------------------------------
/// Sets the write position to given offset.
/// @pre `offset <= buf.size()`
void seek(size_t offset);
/// Jumps `num_bytes` forward. Resizes the buffer (filling it with zeros)
/// when skipping past the end.
void skip(size_t num_bytes);
// -- overridden member functions --------------------------------------------
error begin_object(uint16_t& typenr, std::string& name) override;
error end_object() override;
error begin_sequence(size_t& list_size) override;
error end_sequence() override;
error apply_raw(size_t num_bytes, void* data) override;
// -- properties -------------------------------------------------------------
buffer& buf() {
return buf_;
}
const buffer& buf() const {
return buf_;
}
size_t write_pos() const noexcept {
return write_pos_;
}
protected:
error apply_impl(int8_t&) override;
error apply_impl(uint8_t&) override;
error apply_impl(int16_t&) override;
error apply_impl(uint16_t&) override;
error apply_impl(int32_t&) override;
error apply_impl(uint32_t&) override;
error apply_impl(int64_t&) override;
error apply_impl(uint64_t&) override;
error apply_impl(float&) override;
error apply_impl(double&) override;
error apply_impl(long double&) override;
error apply_impl(std::string&) override;
error apply_impl(std::u16string&) override;
error apply_impl(std::u32string&) override;
private:
buffer& buf_;
size_t write_pos_;
};
} // namespace caf } // namespace caf
...@@ -166,6 +166,10 @@ private: ...@@ -166,6 +166,10 @@ private:
data_ = std::string{x}; data_ = std::string{x};
} }
void set(string_view x) {
data_ = std::string{x.begin(), x.end()};
}
template <class T> template <class T>
detail::enable_if_t<detail::is_one_of<T, real, atom, timespan, uri, string, detail::enable_if_t<detail::is_one_of<T, real, atom, timespan, uri, string,
list, dictionary>::value> list, dictionary>::value>
......
...@@ -201,8 +201,9 @@ template <class Iterator, class Sentinel, class Consumer> ...@@ -201,8 +201,9 @@ template <class Iterator, class Sentinel, class Consumer>
void read_ini_section(state<Iterator, Sentinel>& ps, Consumer&& consumer) { void read_ini_section(state<Iterator, Sentinel>& ps, Consumer&& consumer) {
using std::swap; using std::swap;
std::string tmp; std::string tmp;
auto alnum = [](char x) { return isalnum(x) || x == '_'; };
auto alnum_or_dash = [](char x) { auto alnum_or_dash = [](char x) {
return isalnum(x) || x == '-' || x == '_'; return isalnum(x) || x == '_' || x == '-';
}; };
auto emit_key = [&] { auto emit_key = [&] {
std::string key; std::string key;
...@@ -211,14 +212,15 @@ void read_ini_section(state<Iterator, Sentinel>& ps, Consumer&& consumer) { ...@@ -211,14 +212,15 @@ void read_ini_section(state<Iterator, Sentinel>& ps, Consumer&& consumer) {
}; };
auto g = make_scope_guard([&] { auto g = make_scope_guard([&] {
if (ps.code <= pec::trailing_character) if (ps.code <= pec::trailing_character)
consumer.end_map(); consumer.end_map();
}); });
// clang-format off
start(); start();
// Dispatches to read sections, comments, or key/value pairs. // Dispatches to read sections, comments, or key/value pairs.
term_state(init) { term_state(init) {
transition(init, " \t\n") transition(init, " \t\n")
fsm_epsilon(read_ini_comment(ps, consumer), init, ';') fsm_epsilon(read_ini_comment(ps, consumer), init, ';')
transition(read_key_name, alphanumeric_chars, tmp = ch) transition(read_key_name, alnum, tmp = ch)
} }
// Reads a key of a "key=value" line. // Reads a key of a "key=value" line.
state(read_key_name) { state(read_key_name) {
...@@ -244,6 +246,44 @@ void read_ini_section(state<Iterator, Sentinel>& ps, Consumer&& consumer) { ...@@ -244,6 +246,44 @@ void read_ini_section(state<Iterator, Sentinel>& ps, Consumer&& consumer) {
transition(init, '\n') transition(init, '\n')
} }
fin(); fin();
// clang-format on
}
/// Reads a nested group, e.g., "[foo.bar]" would consume "[foo.]" in read_ini
/// and then delegate to this function for parsing "bar]".
template <class Iterator, class Sentinel, class Consumer>
void read_nested_group(state<Iterator, Sentinel>& ps, Consumer&& consumer) {
using std::swap;
std::string key;
auto alnum = [](char x) { return isalnum(x) || x == '_'; };
auto alnum_or_dash = [](char x) {
return isalnum(x) || x == '_' || x == '-';
};
auto begin_section = [&]() -> decltype(consumer.begin_map()) {
consumer.key(std::move(key));
return consumer.begin_map();
};
auto g = make_scope_guard([&] {
if (ps.code <= pec::trailing_character)
consumer.end_map();
});
// clang-format off
start();
// Scanning for first section.
state(init) {
epsilon(read_sub_section, alnum)
}
// Read the sub section key after reading an '[xxx.'.
state(read_sub_section) {
transition(read_sub_section, alnum_or_dash, key += ch)
fsm_transition(read_nested_group(ps, begin_section()), done, '.')
fsm_transition(read_ini_section(ps, begin_section()), done, ']')
}
term_state(done) {
// nop
}
fin();
// clang-format on
} }
/// Reads an INI formatted input. /// Reads an INI formatted input.
...@@ -251,8 +291,9 @@ template <class Iterator, class Sentinel, class Consumer> ...@@ -251,8 +291,9 @@ template <class Iterator, class Sentinel, class Consumer>
void read_ini(state<Iterator, Sentinel>& ps, Consumer&& consumer) { void read_ini(state<Iterator, Sentinel>& ps, Consumer&& consumer) {
using std::swap; using std::swap;
std::string tmp{"global"}; std::string tmp{"global"};
auto alnum = [](char x) { return isalnum(x) || x == '_'; };
auto alnum_or_dash = [](char x) { auto alnum_or_dash = [](char x) {
return isalnum(x) || x == '-' || x == '_'; return isalnum(x) || x == '_' || x == '-';
}; };
auto begin_section = [&]() -> decltype(consumer.begin_map()) { auto begin_section = [&]() -> decltype(consumer.begin_map()) {
std::string key; std::string key;
...@@ -260,6 +301,7 @@ void read_ini(state<Iterator, Sentinel>& ps, Consumer&& consumer) { ...@@ -260,6 +301,7 @@ void read_ini(state<Iterator, Sentinel>& ps, Consumer&& consumer) {
consumer.key(std::move(key)); consumer.key(std::move(key));
return consumer.begin_map(); return consumer.begin_map();
}; };
// clang-format off
start(); start();
// Scanning for first section. // Scanning for first section.
term_state(init) { term_state(init) {
...@@ -271,11 +313,12 @@ void read_ini(state<Iterator, Sentinel>& ps, Consumer&& consumer) { ...@@ -271,11 +313,12 @@ void read_ini(state<Iterator, Sentinel>& ps, Consumer&& consumer) {
// Read the section key after reading an '['. // Read the section key after reading an '['.
state(start_section) { state(start_section) {
transition(start_section, " \t") transition(start_section, " \t")
transition(read_section_name, alphabetic_chars, tmp = ch) transition(read_section_name, alnum, tmp = ch)
} }
// Reads a section name such as "[foo]". // Reads a section name such as "[foo]".
state(read_section_name) { state(read_section_name) {
transition(read_section_name, alnum_or_dash, tmp += ch) transition(read_section_name, alnum_or_dash, tmp += ch)
fsm_transition(read_nested_group(ps, begin_section()), return_to_global, '.')
epsilon(close_section) epsilon(close_section)
} }
// Wait for the closing ']', preceded by any number of whitespaces. // Wait for the closing ']', preceded by any number of whitespaces.
...@@ -287,6 +330,7 @@ void read_ini(state<Iterator, Sentinel>& ps, Consumer&& consumer) { ...@@ -287,6 +330,7 @@ void read_ini(state<Iterator, Sentinel>& ps, Consumer&& consumer) {
epsilon(init, any_char, tmp = "global") epsilon(init, any_char, tmp = "global")
} }
fin(); fin();
// clang-format on
} }
} // namespace parser } // namespace parser
......
...@@ -33,30 +33,58 @@ ...@@ -33,30 +33,58 @@
#include "caf/detail/type_list.hpp" #include "caf/detail/type_list.hpp"
#define CAF_HAS_MEMBER_TRAIT(name) \ #define CAF_HAS_MEMBER_TRAIT(name) \
template <class T> \ template <class T> \
struct has_##name##_member { \ class has_##name##_member { \
template <class U> \ private: \
static auto sfinae(U* x) -> decltype(x->name(), std::true_type()); \ template <class U> \
static auto sfinae(U* x) -> decltype(x->name(), std::true_type()); \
\ \
template <class U> \ template <class U> \
static auto sfinae(...) -> std::false_type; \ static auto sfinae(...) -> std::false_type; \
\ \
using type = decltype(sfinae<T>(nullptr)); \ using sfinae_type = decltype(sfinae<T>(nullptr)); \
static constexpr bool value = type::value; \ \
} public: \
static constexpr bool value = sfinae_type::value; \
}
#define CAF_HAS_ALIAS_TRAIT(name) \
template <class T> \
class has_##name##_alias { \
private: \
template <class C> \
static std::true_type sfinae(C* ptr, typename C::name* arg = nullptr); \
\
static std::false_type sfinae(void* ptr); \
\
using sfinae_type = decltype(sfinae(static_cast<T*>(nullptr))); \
\
public: \
static constexpr bool value = sfinae_type::value; \
}
namespace caf { namespace caf {
namespace detail { namespace detail {
// -- backport of C++14 additions ----------------------------------------------
template <class T> template <class T>
using decay_t = typename std::decay<T>::type; using decay_t = typename std::decay<T>::type;
template <bool B, class T, class F>
using conditional_t = typename std::conditional<B, T, F>::type;
template <bool V, class T = void> template <bool V, class T = void>
using enable_if_t = typename std::enable_if<V, T>::type; using enable_if_t = typename std::enable_if<V, T>::type;
// -- custom traits ------------------------------------------------------------
template <class Trait, class T = void> template <class Trait, class T = void>
using enable_if_tt = typename std::enable_if<Trait::value, T>::type; using enable_if_tt = typename std::enable_if<Trait::value, T>::type;
template <class T>
using remove_reference_t = typename std::remove_reference<T>::type;
/// Checks whether `T` is inspectable by `Inspector`. /// Checks whether `T` is inspectable by `Inspector`.
template <class Inspector, class T> template <class Inspector, class T>
class is_inspectable { class is_inspectable {
...@@ -730,5 +758,35 @@ struct is_pair : std::false_type {}; ...@@ -730,5 +758,35 @@ struct is_pair : std::false_type {};
template <class First, class Second> template <class First, class Second>
struct is_pair<std::pair<First, Second>> : std::true_type {}; struct is_pair<std::pair<First, Second>> : std::true_type {};
// -- traits to check for STL-style type aliases -------------------------------
CAF_HAS_ALIAS_TRAIT(value_type);
CAF_HAS_ALIAS_TRAIT(key_type);
CAF_HAS_ALIAS_TRAIT(mapped_type);
// -- constexpr functions for use in enable_if & friends -----------------------
/// Checks whether T behaves like a `std::map` or a `std::unordered_map`.
template <class T>
struct is_map_like {
static constexpr bool value = is_iterable<T>::value
&& has_key_type_alias<T>::value
&& has_mapped_type_alias<T>::value;
};
/// Checks whether T behaves like a `std::vector` or a `std::list`.
template <class T>
struct is_list_like {
static constexpr bool value = is_iterable<T>::value
&& has_value_type_alias<T>::value
&& !has_key_type_alias<T>::value
&& !has_mapped_type_alias<T>::value;
};
} // namespace detail } // namespace detail
} // namespace caf } // namespace caf
#undef CAF_HAS_MEMBER_TRAIT
#undef CAF_HAS_ALIAS_TRAIT
...@@ -22,6 +22,7 @@ ...@@ -22,6 +22,7 @@
#include <map> #include <map>
#include <memory> #include <memory>
#include <tuple> #include <tuple>
#include <vector>
#include "caf/detail/is_one_of.hpp" #include "caf/detail/is_one_of.hpp"
#include "caf/detail/is_primitive_config_value.hpp" #include "caf/detail/is_primitive_config_value.hpp"
...@@ -29,6 +30,8 @@ ...@@ -29,6 +30,8 @@
namespace caf { namespace caf {
// clang-format off
// -- 1 param templates -------------------------------------------------------- // -- 1 param templates --------------------------------------------------------
template <class> class behavior_type_of; template <class> class behavior_type_of;
...@@ -39,6 +42,8 @@ template <class> class intrusive_cow_ptr; ...@@ -39,6 +42,8 @@ template <class> class intrusive_cow_ptr;
template <class> class intrusive_ptr; template <class> class intrusive_ptr;
template <class> class optional; template <class> class optional;
template <class> class param; template <class> class param;
template <class> class serializer_impl;
template <class> class span;
template <class> class stream; template <class> class stream;
template <class> class stream_sink; template <class> class stream_sink;
template <class> class stream_source; template <class> class stream_source;
...@@ -59,19 +64,21 @@ template <class, class, class> class broadcast_downstream_manager; ...@@ -59,19 +64,21 @@ template <class, class, class> class broadcast_downstream_manager;
// -- variadic templates ------------------------------------------------------- // -- variadic templates -------------------------------------------------------
template <class...> class result; template <class...> class cow_tuple;
template <class...> class variant;
template <class...> class delegated; template <class...> class delegated;
template <class...> class result; template <class...> class result;
template <class...> class typed_actor; template <class...> class typed_actor;
template <class...> class typed_actor_pointer; template <class...> class typed_actor_pointer;
template <class...> class typed_event_based_actor; template <class...> class typed_event_based_actor;
template <class...> class typed_response_promise; template <class...> class typed_response_promise;
template <class...> class variant;
// -- variadic templates with fixed arguments ---------------------------------- // -- variadic templates with fixed arguments ----------------------------------
// //
template <class, class...> class output_stream; template <class, class...> class output_stream;
// clang-format on
// -- classes ------------------------------------------------------------------ // -- classes ------------------------------------------------------------------
class abstract_actor; class abstract_actor;
...@@ -89,7 +96,6 @@ class actor_system; ...@@ -89,7 +96,6 @@ class actor_system;
class actor_system_config; class actor_system_config;
class behavior; class behavior;
class binary_deserializer; class binary_deserializer;
class binary_serializer;
class blocking_actor; class blocking_actor;
class config_option; class config_option;
class config_option_adder; class config_option_adder;
...@@ -167,17 +173,18 @@ config_option make_config_option(T& storage, string_view category, ...@@ -167,17 +173,18 @@ config_option make_config_option(T& storage, string_view category,
// -- enums -------------------------------------------------------------------- // -- enums --------------------------------------------------------------------
enum class atom_value : uint64_t; enum class atom_value : uint64_t;
enum class byte : uint8_t;
enum class sec : uint8_t; enum class sec : uint8_t;
enum class stream_priority; enum class stream_priority;
// -- aliases ------------------------------------------------------------------ // -- aliases ------------------------------------------------------------------
using actor_id = uint64_t; using actor_id = uint64_t;
using binary_serializer = serializer_impl<std::vector<char>>;
using ip_address = ipv6_address; using ip_address = ipv6_address;
using ip_subnet = ipv6_subnet; using ip_subnet = ipv6_subnet;
using stream_slot = uint16_t;
using settings = dictionary<config_value>; using settings = dictionary<config_value>;
using stream_slot = uint16_t;
// -- functions ---------------------------------------------------------------- // -- functions ----------------------------------------------------------------
......
...@@ -82,15 +82,15 @@ public: ...@@ -82,15 +82,15 @@ public:
using type = detail::decay_t<decltype(*i)>; using type = detail::decay_t<decltype(*i)>;
// Ship full batches. // Ship full batches.
while (std::distance(i, e) >= desired_batch_size) { while (std::distance(i, e) >= desired_batch_size) {
std::vector<type> tmp{std::make_move_iterator(i), std::vector<type> tmp(std::make_move_iterator(i),
std::make_move_iterator(i + desired_batch_size)}; std::make_move_iterator(i + desired_batch_size));
emit_batch(self, desired_batch_size, make_message(std::move(tmp))); emit_batch(self, desired_batch_size, make_message(std::move(tmp)));
i += desired_batch_size; i += desired_batch_size;
} }
// Ship underful batch only if `force_underful` is set. // Ship underful batch only if `force_underful` is set.
if (i != e && force_underfull) { if (i != e && force_underfull) {
std::vector<type> tmp{std::make_move_iterator(i), std::vector<type> tmp(std::make_move_iterator(i),
std::make_move_iterator(e)}; std::make_move_iterator(e));
auto tmp_size = static_cast<int32_t>(tmp.size()); auto tmp_size = static_cast<int32_t>(tmp.size());
emit_batch(self, tmp_size, make_message(std::move(tmp))); emit_batch(self, tmp_size, make_message(std::move(tmp)));
return e; return e;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <cstddef>
#include <cstdint>
#include <cstring>
#include <iomanip>
#include <sstream>
#include <vector>
#include "caf/detail/ieee_754.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/serializer.hpp"
namespace caf {
/// Implements the serializer interface with a binary serialization protocol.
template <class Container>
class serializer_impl final : public serializer {
public:
// -- member types -----------------------------------------------------------
using super = serializer;
using container_type = Container;
using value_type = typename container_type::value_type;
// -- invariants -------------------------------------------------------------
static_assert(sizeof(value_type) == 1,
"container must have a byte-sized value type");
// -- constructors, destructors, and assignment operators --------------------
serializer_impl(actor_system& sys, container_type& buf)
: super(sys), buf_(buf), write_pos_(buf_.size()) {
// nop
}
serializer_impl(execution_unit* ctx, container_type& buf)
: super(ctx), buf_(buf), write_pos_(buf_.size()) {
// nop
}
// -- position management ----------------------------------------------------
/// Sets the write position to given offset.
/// @pre `offset <= buf.size()`
void seek(size_t offset) {
write_pos_ = offset;
}
/// Jumps `num_bytes` forward. Resizes the buffer (filling it with zeros)
/// when skipping past the end.
void skip(size_t num_bytes) {
auto remaining = buf_.size() - write_pos_;
if (remaining < num_bytes)
buf_.insert(buf_.end(), num_bytes - remaining, 0);
write_pos_ += num_bytes;
}
// -- overridden member functions --------------------------------------------
error begin_object(uint16_t& nr, std::string& name) override {
if (nr != 0)
return apply(nr);
if (auto err = apply(nr))
return err;
return apply(name);
}
error end_object() override {
return none;
}
error begin_sequence(size_t& list_size) override {
// Use varbyte encoding to compress sequence size on the wire.
// For 64-bit values, the encoded representation cannot get larger than 10
// bytes. A scratch space of 16 bytes suffices as upper bound.
uint8_t buf[16];
auto i = buf;
auto x = static_cast<uint32_t>(list_size);
while (x > 0x7f) {
*i++ = (static_cast<uint8_t>(x) & 0x7f) | 0x80;
x >>= 7;
}
*i++ = static_cast<uint8_t>(x) & 0x7f;
apply_raw(static_cast<size_t>(i - buf), buf);
return none;
}
error end_sequence() override {
return none;
}
error apply_raw(size_t num_bytes, void* data) override {
CAF_ASSERT(write_pos_ <= buf_.size());
static_assert((sizeof(value_type) == 1), "sizeof(value_type) > 1");
auto ptr = reinterpret_cast<value_type*>(data);
auto buf_size = buf_.size();
if (write_pos_ == buf_size) {
buf_.insert(buf_.end(), ptr, ptr + num_bytes);
} else if (write_pos_ + num_bytes <= buf_size) {
memcpy(buf_.data() + write_pos_, ptr, num_bytes);
} else {
auto remaining = buf_size - write_pos_;
CAF_ASSERT(remaining < num_bytes);
memcpy(buf_.data() + write_pos_, ptr, remaining);
buf_.insert(buf_.end(), ptr + remaining, ptr + num_bytes);
}
write_pos_ += num_bytes;
CAF_ASSERT(write_pos_ <= buf_.size());
return none;
}
// -- properties -------------------------------------------------------------
container_type& buf() {
return buf_;
}
const container_type& buf() const {
return buf_;
}
size_t write_pos() const noexcept {
return write_pos_;
}
protected:
error apply_impl(int8_t& x) override {
return apply_raw(sizeof(int8_t), &x);
}
error apply_impl(uint8_t& x) override {
return apply_raw(sizeof(uint8_t), &x);
}
error apply_impl(int16_t& x) override {
return apply_int(*this, static_cast<uint16_t>(x));
}
error apply_impl(uint16_t& x) override {
return apply_int(*this, x);
}
error apply_impl(int32_t& x) override {
return apply_int(*this, static_cast<uint32_t>(x));
}
error apply_impl(uint32_t& x) override {
return apply_int(*this, x);
}
error apply_impl(int64_t& x) override {
return apply_int(*this, static_cast<uint64_t>(x));
}
error apply_impl(uint64_t& x) override {
return apply_int(*this, x);
}
error apply_impl(float& x) override {
return apply_int(*this, detail::pack754(x));
}
error apply_impl(double& x) override {
return apply_int(*this, detail::pack754(x));
}
error apply_impl(long double& x) override {
// The IEEE-754 conversion does not work for long double
// => fall back to string serialization (event though it sucks).
std::ostringstream oss;
oss << std::setprecision(std::numeric_limits<long double>::digits) << x;
auto tmp = oss.str();
return apply_impl(tmp);
}
error apply_impl(std::string& x) override {
auto str_size = x.size();
if (str_size == 0)
return error::eval([&] { return begin_sequence(str_size); },
[&] { return end_sequence(); });
auto data = const_cast<char*>(x.c_str());
return error::eval([&] { return begin_sequence(str_size); },
[&] { return apply_raw(str_size, data); },
[&] { return end_sequence(); });
}
error apply_impl(std::u16string& x) override {
auto str_size = x.size();
if (auto err = begin_sequence(str_size))
return err;
for (auto c : x) {
// The standard does not guarantee that char16_t is exactly 16 bits.
if (auto err = apply_int(*this, static_cast<uint16_t>(c)))
return err;
}
return none;
}
error apply_impl(std::u32string& x) override {
auto str_size = x.size();
if (auto err = begin_sequence(str_size))
return err;
for (auto c : x) {
// The standard does not guarantee that char32_t is exactly 32 bits.
if (auto err = apply_int(*this, static_cast<uint32_t>(c)))
return err;
}
return none;
}
private:
template <class T>
error apply_int(serializer_impl<Container>& bs, T x) {
auto y = detail::to_network_order(x);
return bs.apply_raw(sizeof(T), &y);
}
container_type& buf_;
size_t write_pos_;
};
} // namespace caf
...@@ -30,29 +30,18 @@ namespace caf { ...@@ -30,29 +30,18 @@ namespace caf {
/// @relates config_value /// @relates config_value
using settings = dictionary<config_value>; using settings = dictionary<config_value>;
/// Tries to retrieve the value associated to `name` from `xs`.
/// @relates config_value
const config_value* get_if(const settings* xs, string_view name);
/// Tries to retrieve the value associated to `name` from `xs`. /// Tries to retrieve the value associated to `name` from `xs`.
/// @relates config_value /// @relates config_value
template <class T> template <class T>
optional<T> get_if(const settings* xs, string_view name) { optional<T> get_if(const settings* xs, string_view name) {
// Access the key directly unless the user specified a dot-separated path. if (auto value = get_if(xs, name))
auto pos = name.find('.'); if (auto ptr = get_if<T>(value))
if (pos == std::string::npos) { return *ptr;
auto i = xs->find(name); return none;
if (i == xs->end())
return none;
// We can't simply return the result here, because it might be a pointer.
auto result = get_if<T>(&i->second);
if (result)
return *result;
return none;
}
// We're dealing with a `<category>.<key>`-formatted string, extract the
// sub-settings by category and recurse.
auto i = xs->find(name.substr(0, pos));
if (i == xs->end() || !holds_alternative<config_value::dictionary>(i->second))
return none;
return get_if<T>(&get<config_value::dictionary>(i->second),
name.substr(pos + 1));
} }
template <class T> template <class T>
...@@ -92,6 +81,19 @@ config_value& put(settings& dict, string_view key, T&& value) { ...@@ -92,6 +81,19 @@ config_value& put(settings& dict, string_view key, T&& value) {
return put_impl(dict, key, tmp); return put_impl(dict, key, tmp);
} }
/// Converts `value` to a `config_value` and assigns it to `key` if `value` is
/// currently missing in `xs`.
/// @param xs Dictionary of key-value pairs.
/// @param key Human-readable nested keys in the form `category.key`.
/// @param value New value for given `key`.
template <class T>
void put_missing(settings& xs, string_view key, T&& value) {
if (get_if(&xs, key) != nullptr)
return;
config_value tmp{std::forward<T>(value)};
put_impl(xs, key, tmp);
}
/// Inserts a new list named `name` into the dictionary `xs` and returns /// Inserts a new list named `name` into the dictionary `xs` and returns
/// a reference to it. Overrides existing entries with the same name. /// a reference to it. Overrides existing entries with the same name.
config_value::list& put_list(settings& xs, std::string name); config_value::list& put_list(settings& xs, std::string name);
......
...@@ -211,7 +211,7 @@ span<byte> as_writable_bytes(span<T> xs) { ...@@ -211,7 +211,7 @@ span<byte> as_writable_bytes(span<T> xs) {
/// Convenience function to make using `caf::span` more convenient without the /// Convenience function to make using `caf::span` more convenient without the
/// deduction guides. /// deduction guides.
template <class T> template <class T>
auto make_span(T& xs) -> span<detail::decay_t<decltype(xs[0])>> { auto make_span(T& xs) -> span<detail::remove_reference_t<decltype(xs[0])>> {
return {xs.data(), xs.size()}; return {xs.data(), xs.size()};
} }
......
...@@ -75,32 +75,45 @@ public: ...@@ -75,32 +75,45 @@ public:
/// Returns the portable name for given type information or `nullptr` /// Returns the portable name for given type information or `nullptr`
/// if no mapping was found. /// if no mapping was found.
const std::string* portable_name(uint16_t nr, const std::type_info* ti) const; const std::string& portable_name(uint16_t nr, const std::type_info* ti) const;
/// Returns the portable name for given type information or `nullptr` /// Returns the portable name for given type information or `nullptr`
/// if no mapping was found. /// if no mapping was found.
inline const std::string* const std::string&
portable_name(const std::pair<uint16_t, const std::type_info*>& x) const { portable_name(const std::pair<uint16_t, const std::type_info*>& x) const {
return portable_name(x.first, x.second); return portable_name(x.first, x.second);
} }
/// Returns the enclosing actor system. /// Returns the enclosing actor system.
inline actor_system& system() const { actor_system& system() const {
return system_; return system_;
} }
/// Returns the default type name for unknown types.
const std::string& default_type_name() const {
return default_type_name_;
}
private: private:
uniform_type_info_map(actor_system& sys); uniform_type_info_map(actor_system& sys);
/// Reference to the parent system.
actor_system& system_; actor_system& system_;
// message types /// Value factories for builtin types.
std::array<value_factory_kvp, type_nrs - 1> builtin_; std::array<value_factory_kvp, type_nrs - 1> builtin_;
/// Values factories for user-defined types.
value_factories_by_name ad_hoc_; value_factories_by_name ad_hoc_;
/// Lock for accessing `ad_hoc_`.`
mutable detail::shared_spinlock ad_hoc_mtx_; mutable detail::shared_spinlock ad_hoc_mtx_;
// message type names /// Names of builtin types.
std::array<std::string, type_nrs - 1> builtin_names_; std::array<std::string, type_nrs - 1> builtin_names_;
/// Displayed name for unknown types.
std::string default_type_name_;
}; };
} // namespace caf } // namespace caf
......
...@@ -129,12 +129,6 @@ actor_system_config::actor_system_config() ...@@ -129,12 +129,6 @@ actor_system_config::actor_system_config()
"schedule utility actors instead of dedicating threads") "schedule utility actors instead of dedicating threads")
.add<bool>("manual-multiplexing", .add<bool>("manual-multiplexing",
"disables background activity of the multiplexer") "disables background activity of the multiplexer")
.add<size_t>("cached-udp-buffers",
"maximum for cached UDP send buffers (default: 10)")
.add<size_t>("max-pending-messages",
"maximum for reordering of UDP receive buffers (default: 10)")
.add<bool>("disable-tcp", "disables communication via TCP")
.add<bool>("enable-udp", "enable communication via UDP")
.add<size_t>("workers", "number of deserialization workers"); .add<size_t>("workers", "number of deserialization workers");
opt_group(custom_options_, "opencl") opt_group(custom_options_, "opencl")
.add<std::vector<size_t>>("device-ids", "whitelist for OpenCL devices"); .add<std::vector<size_t>>("device-ids", "whitelist for OpenCL devices");
...@@ -154,6 +148,73 @@ actor_system_config::actor_system_config() ...@@ -154,6 +148,73 @@ actor_system_config::actor_system_config()
error_renderers.emplace(atom("exit"), render_exit_reason); error_renderers.emplace(atom("exit"), render_exit_reason);
} }
settings actor_system_config::dump_content() const {
settings result = content;
// -- streaming parameters
auto& stream_group = result["stream"].as_dictionary();
put_missing(stream_group, "desired-batch-complexity",
defaults::stream::desired_batch_complexity);
put_missing(stream_group, "max-batch-delay",
defaults::stream::max_batch_delay);
put_missing(stream_group, "credit-round-interval",
defaults::stream::credit_round_interval);
// -- scheduler parameters
auto& scheduler_group = result["scheduler"].as_dictionary();
put_missing(scheduler_group, "policy", defaults::scheduler::policy);
put_missing(scheduler_group, "max-threads", defaults::scheduler::max_threads);
put_missing(scheduler_group, "max-throughput",
defaults::scheduler::max_throughput);
put_missing(scheduler_group, "enable-profiling", false);
put_missing(scheduler_group, "profiling-resolution",
defaults::scheduler::profiling_resolution);
put_missing(scheduler_group, "profiling-output-file", std::string{});
// -- work-stealing parameters
auto& work_stealing_group = result["work-stealing"].as_dictionary();
put_missing(work_stealing_group, "aggressive-poll-attempts",
defaults::work_stealing::aggressive_poll_attempts);
put_missing(work_stealing_group, "aggressive-steal-interval",
defaults::work_stealing::aggressive_steal_interval);
put_missing(work_stealing_group, "moderate-poll-attempts",
defaults::work_stealing::moderate_poll_attempts);
put_missing(work_stealing_group, "moderate-steal-interval",
defaults::work_stealing::moderate_steal_interval);
put_missing(work_stealing_group, "moderate-sleep-duration",
defaults::work_stealing::moderate_sleep_duration);
put_missing(work_stealing_group, "relaxed-steal-interval",
defaults::work_stealing::relaxed_steal_interval);
put_missing(work_stealing_group, "relaxed-sleep-duration",
defaults::work_stealing::relaxed_sleep_duration);
// -- logger parameters
auto& logger_group = result["logger"].as_dictionary();
put_missing(logger_group, "file-name", defaults::logger::file_name);
put_missing(logger_group, "file-format", defaults::logger::file_format);
put_missing(logger_group, "file-verbosity", defaults::logger::file_verbosity);
put_missing(logger_group, "console", defaults::logger::console);
put_missing(logger_group, "console-format", defaults::logger::console_format);
put_missing(logger_group, "console-verbosity",
defaults::logger::console_verbosity);
put_missing(logger_group, "component-blacklist", std::vector<atom_value>{});
put_missing(logger_group, "inline-output", false);
// -- middleman parameters
auto& middleman_group = result["middleman"].as_dictionary();
put_missing(middleman_group, "app-identifiers",
defaults::middleman::app_identifiers);
put_missing(middleman_group, "enable-automatic-connections", false);
put_missing(middleman_group, "max-consecutive-reads",
defaults::middleman::max_consecutive_reads);
put_missing(middleman_group, "heartbeat-interval",
defaults::middleman::heartbeat_interval);
put_missing(middleman_group, "workers", defaults::middleman::workers);
// -- opencl parameters
auto& openssl_group = result["openssl"].as_dictionary();
put_missing(openssl_group, "certificate", std::string{});
put_missing(openssl_group, "key", std::string{});
put_missing(openssl_group, "passphrase", std::string{});
put_missing(openssl_group, "capath", std::string{});
put_missing(openssl_group, "cafile", std::string{});
return result;
}
std::string std::string
actor_system_config::make_help_text(const std::vector<message::cli_arg>& xs) { actor_system_config::make_help_text(const std::vector<message::cli_arg>& xs) {
auto is_no_caf_option = [](const message::cli_arg& arg) { auto is_no_caf_option = [](const message::cli_arg& arg) {
...@@ -238,6 +299,45 @@ bool operator!=(ini_iter iter, ini_sentinel) { ...@@ -238,6 +299,45 @@ bool operator!=(ini_iter iter, ini_sentinel) {
return !iter.ini->fail(); return !iter.ini->fail();
} }
struct indentation {
size_t size;
};
indentation operator+(indentation x, size_t y) noexcept {
return {x.size + y};
}
std::ostream& operator<<(std::ostream& out, indentation indent) {
for (size_t i = 0; i < indent.size; ++i)
out.put(' ');
return out;
}
void print(const config_value::dictionary& xs, indentation indent) {
using std::cout;
for (const auto& kvp : xs) {
if (kvp.first == "dump-config")
continue;
if (auto submap = get_if<config_value::dictionary>(&kvp.second)) {
cout << indent << kvp.first << " {\n";
print(*submap, indent + 2);
cout << indent << "}\n";
} else if (auto lst = get_if<config_value::list>(&kvp.second)) {
if (lst->empty()) {
cout << indent << kvp.first << " = []\n";
} else {
cout << indent << kvp.first << " = [\n";
auto list_indent = indent + 2;
for (auto& x : *lst)
cout << list_indent << to_string(x) << ",\n";
cout << indent << "]\n";
}
} else {
cout << indent << kvp.first << " = " << to_string(kvp.second) << '\n';
}
}
};
} // namespace <anonymous> } // namespace <anonymous>
error actor_system_config::parse(string_list args, std::istream& ini) { error actor_system_config::parse(string_list args, std::istream& ini) {
...@@ -254,12 +354,8 @@ error actor_system_config::parse(string_list args, std::istream& ini) { ...@@ -254,12 +354,8 @@ error actor_system_config::parse(string_list args, std::istream& ini) {
using std::make_move_iterator; using std::make_move_iterator;
auto res = custom_options_.parse(content, args); auto res = custom_options_.parse(content, args);
if (res.second != args.end()) { if (res.second != args.end()) {
if (res.first != pec::success) { if (res.first != pec::success && starts_with(*res.second, "-"))
return make_error(res.first, *res.second); return make_error(res.first, *res.second);
std::cerr << "error: at argument \"" << *res.second
<< "\": " << to_string(res.first) << std::endl;
cli_helptext_printed = true;
}
auto first = args.begin(); auto first = args.begin();
first += std::distance(args.cbegin(), res.second); first += std::distance(args.cbegin(), res.second);
remainder.insert(remainder.end(), make_move_iterator(first), remainder.insert(remainder.end(), make_move_iterator(first),
...@@ -273,16 +369,9 @@ error actor_system_config::parse(string_list args, std::istream& ini) { ...@@ -273,16 +369,9 @@ error actor_system_config::parse(string_list args, std::istream& ini) {
bool long_help = get_or(content, "long-help", false); bool long_help = get_or(content, "long-help", false);
std::cout << custom_options_.help_text(!long_help) << std::endl; std::cout << custom_options_.help_text(!long_help) << std::endl;
} }
// Generate INI dump if needed. // Generate config dump if needed.
if (!cli_helptext_printed && get_or(content, "dump-config", false)) { if (!cli_helptext_printed && get_or(content, "dump-config", false)) {
for (auto& category : content) { print(dump_content(), indentation{0});
if (auto dict = get_if<config_value::dictionary>(&category.second)) {
std::cout << '[' << category.first << "]\n";
for (auto& kvp : *dict)
if (kvp.first != "dump-config")
std::cout << kvp.first << '=' << to_string(kvp.second) << '\n';
}
}
std::cout << std::flush; std::cout << std::flush;
cli_helptext_printed = true; cli_helptext_printed = true;
} }
......
...@@ -26,7 +26,6 @@ ...@@ -26,7 +26,6 @@
#include "caf/detail/network_order.hpp" #include "caf/detail/network_order.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
namespace caf { namespace caf {
namespace { namespace {
...@@ -49,27 +48,31 @@ error apply_float(binary_deserializer& bs, T& x) { ...@@ -49,27 +48,31 @@ error apply_float(binary_deserializer& bs, T& x) {
return none; return none;
} }
} // namespace <anonmyous> } // namespace
binary_deserializer::binary_deserializer(actor_system& sys, const char* buf, binary_deserializer::binary_deserializer(actor_system& sys,
size_t buf_size) span<const byte> bytes)
: super(sys), current_(buf), end_(buf + buf_size) { : super(sys), current_(bytes.data()), end_(bytes.data() + bytes.size()) {
// nop // nop
} }
binary_deserializer::binary_deserializer(execution_unit* ctx, const char* buf, binary_deserializer::binary_deserializer(execution_unit* ctx,
size_t buf_size) span<const byte> bytes)
: super(ctx), current_(buf), end_(buf + buf_size) { : super(ctx), current_(bytes.data()), end_(bytes.data() + bytes.size()) {
// nop // nop
} }
binary_deserializer::binary_deserializer(actor_system& sys, const buffer& buf) binary_deserializer::binary_deserializer(actor_system& sys, const char* buf,
: binary_deserializer(sys, buf.data(), buf.size()) { size_t buf_size)
: binary_deserializer(sys, make_span(reinterpret_cast<const byte*>(buf),
buf_size)) {
// nop // nop
} }
binary_deserializer::binary_deserializer(execution_unit* ctx, const buffer& buf) binary_deserializer::binary_deserializer(execution_unit* ctx, const char* buf,
: binary_deserializer(ctx, buf.data(), buf.size()) { size_t buf_size)
: binary_deserializer(ctx, make_span(reinterpret_cast<const byte*>(buf),
buf_size)) {
// nop // nop
} }
...@@ -112,8 +115,17 @@ error binary_deserializer::apply_raw(size_t num_bytes, void* storage) { ...@@ -112,8 +115,17 @@ error binary_deserializer::apply_raw(size_t num_bytes, void* storage) {
return none; return none;
} }
size_t binary_deserializer::remaining() const { const char* binary_deserializer::current() const {
return static_cast<size_t>(end_ - current_); return reinterpret_cast<const char*>(current_);
}
const char* binary_deserializer::end() const {
return reinterpret_cast<const char*>(end_);
}
void binary_deserializer::skip(size_t num_bytes) {
CAF_ASSERT(num_bytes <= remaining());
current_ += num_bytes;
} }
error binary_deserializer::apply_impl(int8_t& x) { error binary_deserializer::apply_impl(int8_t& x) {
...@@ -172,7 +184,7 @@ error binary_deserializer::apply_impl(std::string& x) { ...@@ -172,7 +184,7 @@ error binary_deserializer::apply_impl(std::string& x) {
return err; return err;
if (!range_check(str_size)) if (!range_check(str_size))
return sec::end_of_stream; return sec::end_of_stream;
x.assign(current_, current_ + str_size); x.assign(reinterpret_cast<const char*>(current_), str_size);
current_ += str_size; current_ += str_size;
return end_sequence(); return end_sequence();
} }
......
...@@ -79,7 +79,7 @@ inbound_path::inbound_path(stream_manager_ptr mgr_ptr, stream_slots id, ...@@ -79,7 +79,7 @@ inbound_path::inbound_path(stream_manager_ptr mgr_ptr, stream_slots id,
mgr->register_input_path(this); mgr->register_input_path(this);
CAF_STREAM_LOG_DEBUG(mgr->self()->name() CAF_STREAM_LOG_DEBUG(mgr->self()->name()
<< "opens input stream with element type" << "opens input stream with element type"
<< *mgr->self()->system().types().portable_name(in_type) << mgr->self()->system().types().portable_name(in_type)
<< "at slot" << id.receiver << "from" << hdl); << "at slot" << id.receiver << "from" << hdl);
} }
......
...@@ -169,8 +169,8 @@ error message::save(serializer& sink, const type_erased_tuple& x) { ...@@ -169,8 +169,8 @@ error message::save(serializer& sink, const type_erased_tuple& x) {
auto n = x.size(); auto n = x.size();
for (size_t i = 0; i < n; ++i) { for (size_t i = 0; i < n; ++i) {
auto rtti = x.type(i); auto rtti = x.type(i);
auto ptr = types.portable_name(rtti); const auto& portable_name = types.portable_name(rtti);
if (ptr == nullptr) { if (portable_name == types.default_type_name()) {
std::cerr << "[ERROR]: cannot serialize message because a type was " std::cerr << "[ERROR]: cannot serialize message because a type was "
"not added to the types list, typeid name: " "not added to the types list, typeid name: "
<< (rtti.second != nullptr ? rtti.second->name() << (rtti.second != nullptr ? rtti.second->name()
...@@ -181,7 +181,7 @@ error message::save(serializer& sink, const type_erased_tuple& x) { ...@@ -181,7 +181,7 @@ error message::save(serializer& sink, const type_erased_tuple& x) {
: "-not-available-"); : "-not-available-");
} }
tname += '+'; tname += '+';
tname += *ptr; tname += portable_name;
} }
auto save_loop = [&]() -> error { auto save_loop = [&]() -> error {
for (size_t i = 0; i < n; ++i) { for (size_t i = 0; i < n; ++i) {
......
...@@ -88,34 +88,57 @@ bool proxy_registry::empty() const { ...@@ -88,34 +88,57 @@ bool proxy_registry::empty() const {
void proxy_registry::erase(const node_id& nid) { void proxy_registry::erase(const node_id& nid) {
CAF_LOG_TRACE(CAF_ARG(nid)); CAF_LOG_TRACE(CAF_ARG(nid));
std::unique_lock<std::mutex> guard{mtx_}; // Move submap for `nid` to a local variable.
auto i = proxies_.find(nid); proxy_map tmp;
if (i == proxies_.end()) {
return; using std::swap;
for (auto& kvp : i->second) std::unique_lock<std::mutex> guard{mtx_};
auto i = proxies_.find(nid);
if (i == proxies_.end())
return;
swap(i->second, tmp);
proxies_.erase(i);
}
// Call kill_proxy outside the critical section.
for (auto& kvp : tmp)
kill_proxy(kvp.second, exit_reason::remote_link_unreachable); kill_proxy(kvp.second, exit_reason::remote_link_unreachable);
proxies_.erase(i);
} }
void proxy_registry::erase(const node_id& nid, actor_id aid, error rsn) { void proxy_registry::erase(const node_id& nid, actor_id aid, error rsn) {
CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid)); CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid));
std::unique_lock<std::mutex> guard{mtx_}; // Try to find the actor handle in question.
auto i = proxies_.find(nid); strong_actor_ptr erased_proxy;
if (i != proxies_.end()) { {
auto& submap = i->second; using std::swap;
auto j = submap.find(aid); std::unique_lock<std::mutex> guard{mtx_};
if (j == submap.end()) auto i = proxies_.find(nid);
return; if (i != proxies_.end()) {
kill_proxy(j->second, std::move(rsn)); auto& submap = i->second;
submap.erase(j); auto j = submap.find(aid);
if (submap.empty()) if (j == submap.end())
proxies_.erase(i); return;
swap(j->second, erased_proxy);
submap.erase(j);
if (submap.empty())
proxies_.erase(i);
}
} }
// Call kill_proxy outside the critical section.
if (erased_proxy != nullptr)
kill_proxy(erased_proxy, std::move(rsn));
} }
void proxy_registry::clear() { void proxy_registry::clear() {
std::unique_lock<std::mutex> guard{mtx_}; CAF_LOG_TRACE("");
for (auto& kvp : proxies_) // Move the content of proxies_ to a local variable.
std::unordered_map<node_id, proxy_map> tmp;
{
using std::swap;
std::unique_lock<std::mutex> guard{mtx_};
swap(proxies_, tmp);
}
// Call kill_proxy outside the critical section.
for (auto& kvp : tmp)
for (auto& sub_kvp : kvp.second) for (auto& sub_kvp : kvp.second)
kill_proxy(sub_kvp.second, exit_reason::remote_link_unreachable); kill_proxy(sub_kvp.second, exit_reason::remote_link_unreachable);
proxies_.clear(); proxies_.clear();
......
...@@ -22,6 +22,25 @@ ...@@ -22,6 +22,25 @@
namespace caf { namespace caf {
const config_value* get_if(const settings* xs, string_view name) {
// Access the key directly unless the user specified a dot-separated path.
auto pos = name.find('.');
if (pos == std::string::npos) {
auto i = xs->find(name);
if (i == xs->end())
return nullptr;
// We can't simply return the result here, because it might be a pointer.
return &i->second;
}
// We're dealing with a `<category>.<key>`-formatted string, extract the
// sub-settings by category and recurse.
auto i = xs->find(name.substr(0, pos));
if (i == xs->end() || !holds_alternative<config_value::dictionary>(i->second))
return nullptr;
return get_if(&get<config_value::dictionary>(i->second),
name.substr(pos + 1));
}
std::string get_or(const settings& xs, string_view name, std::string get_or(const settings& xs, string_view name,
string_view default_value) { string_view default_value) {
auto result = get_if<std::string>(&xs, name); auto result = get_if<std::string>(&xs, name);
......
...@@ -148,21 +148,22 @@ uniform_type_info_map::make_value(const std::type_info& x) const { ...@@ -148,21 +148,22 @@ uniform_type_info_map::make_value(const std::type_info& x) const {
return nullptr; return nullptr;
} }
const std::string* const std::string&
uniform_type_info_map::portable_name(uint16_t nr, uniform_type_info_map::portable_name(uint16_t nr,
const std::type_info* ti) const { const std::type_info* ti) const {
if (nr != 0) if (nr != 0)
return &builtin_names_[nr - 1]; return builtin_names_[nr - 1];
if (ti == nullptr) if (ti == nullptr)
return nullptr; return default_type_name_;
auto& custom_names = system().config().type_names_by_rtti; auto& custom_names = system().config().type_names_by_rtti;
auto i = custom_names.find(std::type_index(*ti)); auto i = custom_names.find(std::type_index(*ti));
if (i != custom_names.end()) if (i != custom_names.end())
return &(i->second); return i->second;
return nullptr; return default_type_name_;
} }
uniform_type_info_map::uniform_type_info_map(actor_system& sys) : system_(sys) { uniform_type_info_map::uniform_type_info_map(actor_system& sys)
: system_(sys), default_type_name_("???") {
sorted_builtin_types list; sorted_builtin_types list;
fill_builtins(builtin_, list, 0); fill_builtins(builtin_, list, 0);
for (size_t i = 0; i < builtin_names_.size(); ++i) for (size_t i = 0; i < builtin_names_.size(); ++i)
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE actor_system_config
#include "caf/actor_system_config.hpp"
#include "caf/test/dsl.hpp"
#include <sstream>
#include <string>
#include <vector>
using namespace caf;
namespace {
using string_list = std::vector<std::string>;
struct config : actor_system_config {
config() {
opt_group{custom_options_, "?foo"}
.add<std::string>("bar,b", "some string parameter");
}
};
struct fixture {
fixture() {
ini << "[foo]\nbar=\"hello\"";
}
error parse(string_list args) {
opts.clear();
remainder.clear();
config cfg;
auto result = cfg.parse(std::move(args), ini);
opts = content(cfg);
remainder = cfg.remainder;
return result;
}
std::stringstream ini;
settings opts;
std::vector<std::string> remainder;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(actor_system_config_tests, fixture)
CAF_TEST(parsing - without CLI arguments) {
CAF_CHECK_EQUAL(parse({}), none);
CAF_CHECK(remainder.empty());
CAF_CHECK_EQUAL(get_or(opts, "foo.bar", ""), "hello");
}
CAF_TEST(parsing - without CLI remainder) {
CAF_MESSAGE("CLI long name");
CAF_CHECK_EQUAL(parse({"--foo.bar=test"}), none);
CAF_CHECK(remainder.empty());
CAF_CHECK_EQUAL(get_or(opts, "foo.bar", ""), "test");
CAF_MESSAGE("CLI abbreviated long name");
CAF_CHECK_EQUAL(parse({"--bar=test"}), none);
CAF_CHECK(remainder.empty());
CAF_CHECK_EQUAL(get_or(opts, "foo.bar", ""), "test");
CAF_MESSAGE("CLI short name");
CAF_CHECK_EQUAL(parse({"-b", "test"}), none);
CAF_CHECK(remainder.empty());
CAF_CHECK_EQUAL(get_or(opts, "foo.bar", ""), "test");
CAF_MESSAGE("CLI short name without whitespace");
CAF_CHECK_EQUAL(parse({"-btest"}), none);
CAF_CHECK(remainder.empty());
CAF_CHECK_EQUAL(get_or(opts, "foo.bar", ""), "test");
}
CAF_TEST(parsing - with CLI remainder) {
CAF_MESSAGE("valid remainder");
CAF_CHECK_EQUAL(parse({"-b", "test", "hello", "world"}), none);
CAF_CHECK_EQUAL(get_or(opts, "foo.bar", ""), "test");
CAF_CHECK_EQUAL(remainder, string_list({"hello", "world"}));
CAF_MESSAGE("invalid remainder");
CAF_CHECK_NOT_EQUAL(parse({"-b", "test", "-abc"}), none);
}
CAF_TEST_FIXTURE_SCOPE_END()
...@@ -105,7 +105,12 @@ log_type make_log(Ts&&... xs) { ...@@ -105,7 +105,12 @@ log_type make_log(Ts&&... xs) {
return log_type{std::forward<Ts>(xs)...}; return log_type{std::forward<Ts>(xs)...};
} }
// Tests basic functionality.
const char* ini0 = R"( const char* ini0 = R"(
[1group]
1value=321
[_foo]
_bar=11
[logger] [logger]
padding= 10 padding= 10
file-name = "foobar.ini" ; our file name file-name = "foobar.ini" ; our file name
...@@ -138,7 +143,18 @@ tcp://localhost:8080 ...@@ -138,7 +143,18 @@ tcp://localhost:8080
>,<udp://remotehost?trust=false>] >,<udp://remotehost?trust=false>]
)"; )";
// clang-format off
const auto ini0_log = make_log( const auto ini0_log = make_log(
"key: 1group",
"{",
"key: 1value",
"value (integer): 321",
"}",
"key: _foo",
"{",
"key: _bar",
"value (integer): 11",
"}",
"key: logger", "key: logger",
"{", "{",
"key: padding", "key: padding",
...@@ -186,6 +202,45 @@ const auto ini0_log = make_log( ...@@ -186,6 +202,45 @@ const auto ini0_log = make_log(
"]", "]",
"}" "}"
); );
// clang-format on
// Tests nested parameters.
const char* ini1 = R"(
foo {
bar = {
value1 = 1
}
value2 = 2
}
[bar.foo]
value3 = 3
)";
// clang-format off
const auto ini1_log = make_log(
"key: global",
"{",
"key: foo",
"{",
"key: bar",
"{",
"key: value1",
"value (integer): 1",
"}",
"key: value2",
"value (integer): 2",
"}",
"}",
"key: bar",
"{",
"key: foo",
"{",
"key: value3",
"value (integer): 3",
"}",
"}"
);
// clang-format on
} // namespace <anonymous> } // namespace <anonymous>
...@@ -206,6 +261,7 @@ CAF_TEST(section with valid key-value pairs) { ...@@ -206,6 +261,7 @@ CAF_TEST(section with valid key-value pairs) {
CAF_CHECK_EQUAL(parse(" [ foo ] "), make_log("key: foo", "{", "}")); CAF_CHECK_EQUAL(parse(" [ foo ] "), make_log("key: foo", "{", "}"));
CAF_CHECK_EQUAL(parse("\n[a-b];foo\n;bar"), make_log("key: a-b", "{", "}")); CAF_CHECK_EQUAL(parse("\n[a-b];foo\n;bar"), make_log("key: a-b", "{", "}"));
CAF_CHECK_EQUAL(parse(ini0), ini0_log); CAF_CHECK_EQUAL(parse(ini0), ini0_log);
CAF_CHECK_EQUAL(parse(ini1), ini1_log);
} }
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE serializer_impl
#include "caf/serializer_impl.hpp"
#include "caf/test/dsl.hpp"
#include <cstring>
#include <vector>
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp"
#include "caf/duration.hpp"
#include "caf/timestamp.hpp"
using namespace caf;
namespace {
enum class test_enum {
a,
b,
c,
};
struct test_data {
test_data(int32_t i32, int64_t i64, float f32, double f64, caf::duration dur,
caf::timestamp ts, test_enum te, const std::string& str)
: i32_(i32),
i64_(i64),
f32_(f32),
f64_(f64),
dur_(dur),
ts_(ts),
te_(te),
str_(str) {
// nop
}
test_data()
: test_data(-345, -1234567890123456789ll, 3.45, 54.3,
caf::duration(caf::time_unit::seconds, 123),
caf::timestamp{
caf::timestamp::duration{1478715821 * 1000000000ll}},
test_enum::b, "Lorem ipsum dolor sit amet.") {
// nop
}
int32_t i32_;
int64_t i64_;
float f32_;
double f64_;
caf::duration dur_;
caf::timestamp ts_;
test_enum te_;
std::string str_;
friend bool operator==(const test_data& data, const test_data& other) {
return (data.f64_ == other.f64_ && data.i32_ == other.i32_
&& data.i64_ == other.i64_ && data.str_ == other.str_
&& data.te_ == other.te_ && data.ts_ == other.ts_);
}
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, test_data& x) {
return f(caf::meta::type_name("test_data"), x.i32_, x.i64_, x.f32_, x.f64_,
x.dur_, x.ts_, x.te_, x.str_);
}
struct serialization_fixture {
caf::actor_system_config cfg;
caf::actor_system sys{cfg};
test_data data_to_serialize;
test_data deserialized_data{0,
0,
0,
0,
caf::duration(caf::time_unit::seconds, 0),
caf::timestamp{caf::timestamp::duration{0}},
test_enum::a,
""};
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(serializer_impl_tests, serialization_fixture)
CAF_TEST(serialize and deserialize with std::vector<char>) {
using container_type = std::vector<char>;
std::vector<char> binary_serializer_buffer;
container_type serializer_impl_buffer;
binary_serializer sink1{sys, binary_serializer_buffer};
serializer_impl<container_type> sink2{sys, serializer_impl_buffer};
if (auto err = sink1(data_to_serialize))
CAF_FAIL("serialization failed: " << sys.render(err));
if (auto err = sink2(data_to_serialize))
CAF_FAIL("serialization failed: " << sys.render(err));
CAF_CHECK_EQUAL(memcmp(binary_serializer_buffer.data(),
serializer_impl_buffer.data(),
binary_serializer_buffer.size()),
0);
binary_deserializer source(sys, serializer_impl_buffer);
if (auto err = source(deserialized_data))
CAF_FAIL("deserialization failed: " << sys.render(err));
CAF_CHECK_EQUAL(data_to_serialize, deserialized_data);
}
CAF_TEST(serialize and deserialize with std::vector<byte>) {
using container_type = std::vector<byte>;
std::vector<char> binary_serializer_buffer;
container_type serializer_impl_buffer;
binary_serializer sink1{sys, binary_serializer_buffer};
serializer_impl<container_type> sink2{sys, serializer_impl_buffer};
if (auto err = sink1(data_to_serialize))
CAF_FAIL("serialization failed: " << sys.render(err));
if (auto err = sink2(data_to_serialize))
CAF_FAIL("serialization failed: " << sys.render(err));
CAF_CHECK_EQUAL(memcmp(binary_serializer_buffer.data(),
serializer_impl_buffer.data(),
binary_serializer_buffer.size()),
0);
binary_deserializer source(sys, serializer_impl_buffer);
if (auto err = source(deserialized_data))
CAF_FAIL("deserialization failed: " << sys.render(err));
CAF_CHECK_EQUAL(data_to_serialize, deserialized_data);
}
CAF_TEST(serialize and deserialize with std::vector<uint8_t>) {
using container_type = std::vector<uint8_t>;
std::vector<char> binary_serializer_buffer;
container_type serializer_impl_buffer;
binary_serializer sink1{sys, binary_serializer_buffer};
serializer_impl<container_type> sink2{sys, serializer_impl_buffer};
if (auto err = sink1(data_to_serialize))
CAF_FAIL("serialization failed: " << sys.render(err));
if (auto err = sink2(data_to_serialize))
CAF_FAIL("serialization failed: " << sys.render(err));
CAF_CHECK_EQUAL(memcmp(binary_serializer_buffer.data(),
serializer_impl_buffer.data(),
binary_serializer_buffer.size()),
0);
binary_deserializer source(sys, serializer_impl_buffer);
if (auto err = source(deserialized_data))
CAF_FAIL("deserialization failed: " << sys.render(err));
CAF_CHECK_EQUAL(data_to_serialize, deserialized_data);
}
CAF_TEST_FIXTURE_SCOPE_END();
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment