Unverified Commit b5a8c4bc authored by Joseph Noir's avatar Joseph Noir Committed by GitHub

Merge pull request #863

Make node_id opaque and add an URI-based alternative
parents c58b6a08 9d4ff082
...@@ -54,6 +54,7 @@ set(LIBCAF_CORE_SRCS ...@@ -54,6 +54,7 @@ set(LIBCAF_CORE_SRCS
src/event_based_actor.cpp src/event_based_actor.cpp
src/execution_unit.cpp src/execution_unit.cpp
src/exit_reason.cpp src/exit_reason.cpp
src/fnv_hash.cpp
src/forwarding_actor_proxy.cpp src/forwarding_actor_proxy.cpp
src/get_mac_addresses.cpp src/get_mac_addresses.cpp
src/get_process_id.cpp src/get_process_id.cpp
......
...@@ -20,16 +20,16 @@ ...@@ -20,16 +20,16 @@
#include <atomic> #include <atomic>
#include "caf/fwd.hpp" #include "caf/config.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/node_id.hpp" #include "caf/fwd.hpp"
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/weak_intrusive_ptr.hpp"
#include "caf/meta/type_name.hpp"
#include "caf/meta/save_callback.hpp"
#include "caf/meta/load_callback.hpp" #include "caf/meta/load_callback.hpp"
#include "caf/meta/omittable_if_none.hpp" #include "caf/meta/omittable_if_none.hpp"
#include "caf/meta/save_callback.hpp"
#include "caf/meta/type_name.hpp"
#include "caf/node_id.hpp"
#include "caf/weak_intrusive_ptr.hpp"
namespace caf { namespace caf {
......
...@@ -40,8 +40,12 @@ std::string to_string(const atom_value& what); ...@@ -40,8 +40,12 @@ std::string to_string(const atom_value& what);
/// @relates atom_value /// @relates atom_value
atom_value to_lowercase(atom_value x); atom_value to_lowercase(atom_value x);
/// @relates atom_value
atom_value atom_from_string(string_view x); atom_value atom_from_string(string_view x);
/// @relates atom_value
int compare(atom_value x, atom_value y);
/// Creates an atom from given string literal. /// Creates an atom from given string literal.
template <size_t Size> template <size_t Size>
constexpr atom_value atom(char const (&str)[Size]) { constexpr atom_value atom(char const (&str)[Size]) {
...@@ -219,4 +223,3 @@ struct hash<caf::atom_value> { ...@@ -219,4 +223,3 @@ struct hash<caf::atom_value> {
}; };
} // namespace std } // namespace std
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#pragma once
#include <cstdint>
#include <type_traits>
#include "caf/detail/type_traits.hpp"
namespace caf {
namespace detail {
/// Non-cryptographic hash function named after Glenn Fowler, Landon Curt Noll,
/// and Kiem-Phong Vo.
///
/// See:
/// - https://en.wikipedia.org/wiki/Fowler%E2%80%93Noll%E2%80%93Vo_hash_function
/// - http://www.isthe.com/chongo/tech/comp/fnv/index.html
size_t fnv_hash(const unsigned char* first, const unsigned char* last);
size_t fnv_hash_append(size_t intermediate, const unsigned char* first,
const unsigned char* last);
template <class T>
enable_if_t<has_data_member<T>::value, size_t> fnv_hash(const T& x) {
auto ptr = x.data();
auto first = reinterpret_cast<const uint8_t*>(ptr);
auto last = first + (sizeof(decay_t<decltype(*ptr)>) * x.size());
return fnv_hash(first, last);
}
template <class T>
enable_if_t<has_data_member<T>::value, size_t>
fnv_hash_append(size_t intermediate, const T& x) {
auto ptr = x.data();
auto first = reinterpret_cast<const uint8_t*>(ptr);
auto last = first + (sizeof(decay_t<decltype(*ptr)>) * x.size());
return fnv_hash_append(intermediate, first, last);
}
template <class T>
enable_if_t<std::is_integral<T>::value, size_t> fnv_hash(const T& x) {
auto first = reinterpret_cast<const uint8_t*>(&x);
return fnv_hash(first, first + sizeof(T));
}
template <class T>
enable_if_t<std::is_integral<T>::value, size_t> fnv_hash_append(size_t interim,
const T& x) {
auto first = reinterpret_cast<const uint8_t*>(&x);
return fnv_hash_append(interim, first, first + sizeof(T));
}
} // namespace detail
} // namespace caf
...@@ -23,135 +23,198 @@ ...@@ -23,135 +23,198 @@
#include <cstdint> #include <cstdint>
#include <functional> #include <functional>
#include "caf/atom.hpp"
#include "caf/detail/comparable.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/none.hpp" #include "caf/none.hpp"
#include "caf/error.hpp"
#include "caf/config.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/make_counted.hpp" #include "caf/uri.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/meta/type_name.hpp"
#include "caf/meta/hex_formatted.hpp"
#include "caf/detail/comparable.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/detail/type_traits.hpp"
namespace caf { namespace caf {
/// A node ID consists of a host ID and process ID. The host ID identifies /// A node ID is an opaque value for representing CAF instances in the network.
/// the physical machine in the network, whereas the process ID identifies
/// the running system-level process on that machine.
class node_id { class node_id {
public: public:
~node_id(); // -- member types -----------------------------------------------------------
node_id() = default; // A reference counted, implementation-specific implementation of a node ID.
class data : public ref_counted {
public:
~data() override;
node_id(const node_id&) = default; virtual bool valid() const noexcept = 0;
node_id(const none_t&); virtual size_t hash_code() const noexcept = 0;
node_id& operator=(const node_id&) = default; virtual atom_value implementation_id() const noexcept = 0;
node_id& operator=(const none_t&); virtual int compare(const data& other) const noexcept = 0;
virtual void print(std::string& dst) const = 0;
virtual error serialize(serializer& sink) const = 0;
virtual error deserialize(deserializer& source) = 0;
};
// A technology-agnostic node identifier with process ID and hash value.
class default_data final : public data {
public:
// -- constants ------------------------------------------------------------
/// A 160 bit hash (20 bytes). /// A 160 bit hash (20 bytes).
static constexpr size_t host_id_size = 20; static constexpr size_t host_id_size = 20;
/// The size of a `node_id` in serialized form. /// Identifies this data implementation type.
static constexpr size_t serialized_size = host_id_size + sizeof(uint32_t); static constexpr atom_value class_id = atom("default");
// -- member types ---------------------------------------------------------
/// Represents a 160 bit hash. /// Represents a 160 bit hash.
using host_id_type = std::array<uint8_t, host_id_size>; using host_id_type = std::array<uint8_t, host_id_size>;
/// Creates a node ID from `process_id` and `hash`. // -- constructors, destructors, and assignment operators ------------------
/// @param procid System-wide unique process identifier.
/// @param hash Unique node id as hexadecimal string representation.
node_id(uint32_t procid, const std::string& hash);
/// Creates a node ID from `process_id` and `hash`. default_data();
/// @param procid System-wide unique process identifier.
/// @param hid Unique node id.
node_id(uint32_t procid, const host_id_type& hid);
/// Identifies the running process. default_data(uint32_t pid, const host_id_type& host);
/// @returns A system-wide unique process identifier.
uint32_t process_id() const;
/// Identifies the host system. // -- factory functions ----------------------------------------------------
/// @returns A hash build from the MAC address of the first network device
/// and the UUID of the root partition (mounted in "/" or "C:").
const host_id_type& host_id() const;
/// Queries whether this node is not default-constructed. /// Returns an ID for this node.
explicit operator bool() const; static node_id local(const actor_system_config& cfg);
void swap(node_id&); // -- properties -----------------------------------------------------------
/// @cond PRIVATE uint32_t process_id() const noexcept {
return pid_;
}
// A reference counted container for host ID and process ID. const host_id_type host_id() const noexcept {
class data : public ref_counted { return host_;
public: }
static intrusive_ptr<data> create_singleton();
int compare(const node_id& other) const; // -- utility functions ----------------------------------------------------
~data() override; static bool valid(const host_id_type& x) noexcept;
// -- interface implementation ---------------------------------------------
bool valid() const noexcept override;
data(); size_t hash_code() const noexcept override;
data(uint32_t procid, host_id_type hid); atom_value implementation_id() const noexcept override;
data(uint32_t procid, const std::string& hash); int compare(const data& other) const noexcept override;
data(const data&) = default; void print(std::string& dst) const override;
data& operator=(const data&) = default; error serialize(serializer& sink) const override;
bool valid() const; error deserialize(deserializer& source) override;
private:
// -- member variables -----------------------------------------------------
uint32_t pid_; uint32_t pid_;
host_id_type host_; host_id_type host_;
}; };
// "inherited" from comparable<node_id> // A technology-agnostic node identifier using an URI.
int compare(const node_id& other) const; class uri_data final : public data {
public:
// -- constants ------------------------------------------------------------
/// Identifies this data implementation type.
static constexpr atom_value class_id = atom("uri");
// -- constructors, destructors, and assignment operators ------------------
explicit uri_data(uri value);
// -- properties -----------------------------------------------------------
const uri& value() const noexcept {
return value_;
}
// -- interface implementation ---------------------------------------------
bool valid() const noexcept override;
size_t hash_code() const noexcept override;
atom_value implementation_id() const noexcept override;
int compare(const data& other) const noexcept override;
void print(std::string& dst) const override;
error serialize(serializer& sink) const override;
error deserialize(deserializer& source) override;
private:
// -- member variables -----------------------------------------------------
uri value_;
};
// -- constructors, destructors, and assignment operators --------------------
// "inherited" from comparable<node_id, invalid_node_id_t> constexpr node_id() noexcept {
int compare(const none_t&) const; // nop
}
explicit node_id(intrusive_ptr<data> dataptr); explicit node_id(intrusive_ptr<data> dataptr);
template <class Inspector> node_id& operator=(const none_t&);
friend detail::enable_if_t<Inspector::reads_state,
typename Inspector::result_type> node_id(node_id&&) = default;
inspect(Inspector& f, node_id& x) {
data tmp; node_id(const node_id&) = default;
data* ptr = x ? x.data_.get() : &tmp;
return f(meta::type_name("node_id"), ptr->pid_, node_id& operator=(node_id&&) = default;
meta::hex_formatted(), ptr->host_);
node_id& operator=(const node_id&) = default;
~node_id();
// -- properties -------------------------------------------------------------
/// Queries whether this node is not default-constructed.
explicit operator bool() const;
/// Compares this instance to `other`.
/// @returns -1 if `*this < other`, 0 if `*this == other`, and 1 otherwise.
int compare(const node_id& other) const noexcept;
/// Exchanges the value of this object with `other`.
void swap(node_id& other);
/// @cond PRIVATE
error serialize(serializer& sink) const;
error deserialize(deserializer& source);
data* operator->() noexcept {
return data_.get();
} }
template <class Inspector> const data* operator->() const noexcept {
friend detail::enable_if_t<Inspector::writes_state, return data_.get();
typename Inspector::result_type> }
inspect(Inspector& f, node_id& x) {
data tmp; data& operator*() noexcept {
// write changes to tmp back to x at scope exit return *data_;
auto sg = detail::make_scope_guard([&] { }
if (!tmp.valid())
x.data_.reset(); const data& operator*() const noexcept {
else if (!x || !x.data_->unique()) return *data_;
x.data_ = make_counted<data>(tmp);
else
*x.data_ = tmp;
});
return f(meta::type_name("node_id"), tmp.pid_,
meta::hex_formatted(), tmp.host_);
} }
/// @endcond /// @endcond
...@@ -161,60 +224,96 @@ private: ...@@ -161,60 +224,96 @@ private:
}; };
/// @relates node_id /// @relates node_id
inline bool operator==(const node_id& x, none_t) { inline bool operator==(const node_id& x, const node_id& y) noexcept {
return x.compare(y) == 0;
}
/// @relates node_id
inline bool operator!=(const node_id& x, const node_id& y) noexcept {
return x.compare(y) != 0;
}
/// @relates node_id
inline bool operator<(const node_id& x, const node_id& y) noexcept {
return x.compare(y) < 0;
}
/// @relates node_id
inline bool operator<=(const node_id& x, const node_id& y) noexcept {
return x.compare(y) <= 0;
}
/// @relates node_id
inline bool operator>(const node_id& x, const node_id& y) noexcept {
return x.compare(y) > 0;
}
/// @relates node_id
inline bool operator>=(const node_id& x, const node_id& y) noexcept {
return x.compare(y) >= 0;
}
/// @relates node_id
inline bool operator==(const node_id& x, const none_t&) noexcept {
return !x; return !x;
} }
/// @relates node_id /// @relates node_id
inline bool operator==(none_t, const node_id& x) { inline bool operator==(const none_t&, const node_id& x) noexcept {
return !x; return !x;
} }
/// @relates node_id /// @relates node_id
inline bool operator!=(const node_id& x, none_t) { inline bool operator!=(const node_id& x, const none_t&) noexcept {
return static_cast<bool>(x); return static_cast<bool>(x);
} }
/// @relates node_id /// @relates node_id
inline bool operator!=(none_t, const node_id& x) { inline bool operator!=(const none_t&, const node_id& x) noexcept {
return static_cast<bool>(x); return static_cast<bool>(x);
} }
inline bool operator==(const node_id& lhs, const node_id& rhs) { /// @relates node_id
return lhs.compare(rhs) == 0; error inspect(serializer& sink, const node_id& x);
}
inline bool operator!=(const node_id& lhs, const node_id& rhs) { /// @relates node_id
return !(lhs == rhs); error inspect(deserializer& source, node_id& x);
}
inline bool operator<(const node_id& lhs, const node_id& rhs) { /// Appends `x` in human-readable string representation to `str`.
return lhs.compare(rhs) < 0; /// @relates node_id
} void append_to_string(std::string& str, const node_id& x);
/// Converts `x` into a human-readable string representation. /// Converts `x` into a human-readable string representation.
/// @relates node_id /// @relates node_id
std::string to_string(const node_id& x); std::string to_string(const node_id& x);
/// Appends `y` in human-readable string representation to `x`. /// Creates a node ID from the URI `from`.
/// @relates node_id
node_id make_node_id(uri from);
/// Creates a node ID from `process_id` and `host_id`.
/// @param process_id System-wide unique process identifier.
/// @param host_id Unique hash value representing a single CAF node.
/// @relates node_id /// @relates node_id
void append_to_string(std::string& x, const node_id& y); node_id make_node_id(uint32_t process_id,
const node_id::default_data::host_id_type& host_id);
/// Creates a node ID from `process_id` and `host_hash`.
/// @param process_id System-wide unique process identifier.
/// @param host_hash Unique node ID as hexadecimal string representation.
/// @relates node_id
optional<node_id> make_node_id(uint32_t process_id,
const std::string& host_hash);
} // namespace caf } // namespace caf
namespace std{ namespace std {
template<> template<>
struct hash<caf::node_id> { struct hash<caf::node_id> {
size_t operator()(const caf::node_id& nid) const { size_t operator()(const caf::node_id& x) const noexcept {
if (nid == caf::none) return x ? x->hash_code() : 0;
return 0;
// xor the first few bytes from the node ID and the process ID
auto x = static_cast<size_t>(nid.process_id());
auto y = *(reinterpret_cast<const size_t*>(&nid.host_id()));
return x ^ y;
} }
}; };
} // namespace std } // namespace std
...@@ -105,6 +105,9 @@ public: ...@@ -105,6 +105,9 @@ public:
/// Returns the fragment component. /// Returns the fragment component.
string_view fragment() const noexcept; string_view fragment() const noexcept;
/// Returns a hash code over all components.
size_t hash_code() const noexcept;
// -- comparison ------------------------------------------------------------- // -- comparison -------------------------------------------------------------
int compare(const uri& other) const noexcept; int compare(const uri& other) const noexcept;
...@@ -135,3 +138,14 @@ std::string to_string(const uri& x); ...@@ -135,3 +138,14 @@ std::string to_string(const uri& x);
error parse(string_view str, uri& dest); error parse(string_view str, uri& dest);
} // namespace caf } // namespace caf
namespace std {
template <>
struct hash<caf::uri> {
size_t operator()(const caf::uri& x) const noexcept {
return x.hash_code();
}
};
} // namespace std
...@@ -71,4 +71,8 @@ std::string to_string(const atom_value& x) { ...@@ -71,4 +71,8 @@ std::string to_string(const atom_value& x) {
return std::string(str.begin(), str.begin() + len); return std::string(str.begin(), str.begin() + len);
} }
int compare(atom_value x, atom_value y) {
return memcmp(&x, &y, sizeof(atom_value));
}
} // namespace caf } // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#include "caf/detail/fnv_hash.hpp"
namespace caf {
namespace detail {
namespace {
template <size_t IntegerSize>
struct hash_conf_helper;
template <>
struct hash_conf_helper<4> {
static constexpr size_t basis = 2166136261u;
static constexpr size_t prime = 16777619u;
};
template <>
struct hash_conf_helper<8> {
static constexpr size_t basis = 14695981039346656037u;
static constexpr size_t prime = 1099511628211u;
};
struct hash_conf : hash_conf_helper<sizeof(size_t)> {};
} // namespace
size_t fnv_hash(const unsigned char* first, const unsigned char* last) {
return fnv_hash_append(hash_conf::basis, first, last);
}
size_t fnv_hash_append(size_t intermediate, const unsigned char* first,
const unsigned char* last) {
auto result = intermediate;
for (; first != last; ++first) {
result *= hash_conf::prime;
result ^= *first;
}
return result;
}
} // namespace detail
} // namespace caf
...@@ -16,182 +16,257 @@ ...@@ -16,182 +16,257 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/node_id.hpp"
#include <cstdio> #include <cstdio>
#include <cstring> #include <cstring>
#include <sstream>
#include <iterator> #include <iterator>
#include <sstream>
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/node_id.hpp"
#include "caf/serializer.hpp"
#include "caf/deserializer.hpp" #include "caf/deserializer.hpp"
#include "caf/detail/get_mac_addresses.hpp"
#include "caf/detail/get_process_id.hpp"
#include "caf/detail/get_root_uuid.hpp"
#include "caf/detail/parser/ascii_to_int.hpp"
#include "caf/detail/ripemd_160.hpp"
#include "caf/logger.hpp"
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
#include "caf/sec.hpp"
#include "caf/serializer.hpp"
#include "caf/string_algorithms.hpp" #include "caf/string_algorithms.hpp"
#include "caf/primitive_variant.hpp"
#include "caf/logger.hpp"
#include "caf/detail/ripemd_160.hpp"
#include "caf/detail/get_root_uuid.hpp"
#include "caf/detail/get_process_id.hpp"
#include "caf/detail/get_mac_addresses.hpp"
namespace caf { namespace caf {
node_id::data::~data() {
// nop
}
node_id::default_data::default_data() : pid_(0) {
memset(host_.data(), 0, host_.size());
}
node_id::default_data::default_data(uint32_t pid, const host_id_type& host)
: pid_(pid), host_(host) {
// nop
}
namespace { namespace {
uint32_t invalid_process_id = 0; std::atomic<uint8_t> system_id;
node_id::host_id_type invalid_host_id; } // namespace
} // namespace <anonymous> node_id node_id::default_data::local(const actor_system_config&) {
CAF_LOG_TRACE("");
auto ifs = detail::get_mac_addresses();
std::vector<std::string> macs;
macs.reserve(ifs.size());
for (auto& i : ifs)
macs.emplace_back(std::move(i.second));
auto hd_serial_and_mac_addr = join(macs, "") + detail::get_root_uuid();
host_id_type hid;
detail::ripemd_160(hid, hd_serial_and_mac_addr);
// This hack enables multiple actor systems in a single process by overriding
// the last byte in the node ID with the actor system "ID".
hid.back() = system_id.fetch_add(1);
return make_node_id(detail::get_process_id(), hid);
}
node_id::~node_id() { bool node_id::default_data::valid(const host_id_type& x) noexcept {
// nop auto is_zero = [](uint8_t x) { return x == 0; };
return !std::all_of(x.begin(), x.end(), is_zero);
} }
node_id::node_id(const none_t&) { bool node_id::default_data::valid() const noexcept {
// nop return pid_ != 0 && valid(host_);
} }
node_id& node_id::operator=(const none_t&) { size_t node_id::default_data::hash_code() const noexcept {
data_.reset(); // XOR the first few bytes from the node ID and the process ID.
return *this; auto x = static_cast<size_t>(pid_);
auto y = *reinterpret_cast<const size_t*>(host_.data());
return x ^ y;
} }
node_id::node_id(intrusive_ptr<data> dataptr) : data_(std::move(dataptr)) { atom_value node_id::default_data::implementation_id() const noexcept {
// nop return class_id;
} }
node_id::node_id(uint32_t procid, const std::string& hash) int node_id::default_data::compare(const data& other) const noexcept {
: data_(make_counted<data>(procid, hash)) { if (this == &other)
// nop return 0;
auto other_id = other.implementation_id();
if (class_id != other_id)
return caf::compare(class_id, other_id);
auto& x = static_cast<const default_data&>(other);
if (pid_ != x.pid_)
return pid_ < x.pid_ ? -1 : 1;
return memcmp(host_.data(), x.host_.data(), host_.size());
} }
node_id::node_id(uint32_t procid, const host_id_type& hid) void node_id::default_data::print(std::string& dst) const {
: data_(make_counted<data>(procid, hid)) { if (!valid()) {
// nop dst += "invalid-node";
return;
}
detail::append_hex(dst, host_);
dst += '#';
dst += std::to_string(pid_);
} }
int node_id::compare(const none_t&) const { error node_id::default_data::serialize(serializer& sink) const {
return data_ ? 1 : 0; // invalid instances are always smaller return sink(pid_, host_);
} }
int node_id::compare(const node_id& other) const { error node_id::default_data::deserialize(deserializer& source) {
if (this == &other || data_ == other.data_) return source(pid_, host_);
return 0; // shortcut for comparing to self or identical instances
if (!data_ != !other.data_)
return data_ ? 1 : -1; // invalid instances are always smaller
// use mismatch instead of strncmp because the
// latter bails out on the first 0-byte
auto last = host_id().end();
auto ipair = std::mismatch(host_id().begin(), last, other.host_id().begin());
if (ipair.first == last)
return static_cast<int>(process_id())-static_cast<int>(other.process_id());
if (*ipair.first < *ipair.second)
return -1;
return 1;
}
node_id::data::data() : pid_(0) {
memset(host_.data(), 0, host_.size());
} }
node_id::data::data(uint32_t procid, host_id_type hid) node_id::uri_data::uri_data(uri value) : value_(std::move(value)) {
: pid_(procid),
host_(hid) {
// nop // nop
} }
node_id::data::data(uint32_t procid, const std::string& hash) : pid_(procid) { bool node_id::uri_data::valid() const noexcept {
if (hash.size() != (host_id_size * 2)) { return !value_.empty();
host_ = invalid_host_id;
return;
}
auto hex_value = [](char c) -> uint8_t {
if (isalpha(c) != 0) {
if (c >= 'a' && c <= 'f')
return static_cast<uint8_t>((c - 'a') + 10);
if (c >= 'A' && c <= 'F')
return static_cast<uint8_t>((c - 'A') + 10);
}
return isdigit(c) != 0 ? static_cast<uint8_t>(c - '0') : 0;
};
auto j = hash.c_str();
for (size_t i = 0; i < host_id_size; ++i) {
// read two characters, each representing 4 bytes
host_[i] = static_cast<uint8_t>(hex_value(j[0]) << 4) | hex_value(j[1]);
j += 2;
}
} }
node_id::data::~data() { size_t node_id::uri_data::hash_code() const noexcept {
// nop std::hash<uri> f;
return f(value_);
} }
bool node_id::data::valid() const { atom_value node_id::uri_data::implementation_id() const noexcept {
auto is_zero = [](uint8_t x) { return x == 0; }; return class_id;
return pid_ != 0 && !std::all_of(host_.begin(), host_.end(), is_zero);
} }
namespace { int node_id::uri_data::compare(const data& other) const noexcept {
if (this == &other)
return 0;
auto other_id = other.implementation_id();
if (class_id != other_id)
return caf::compare(class_id, other_id);
return value_.compare(static_cast<const uri_data&>(other).value_);
}
std::atomic<uint8_t> system_id; void node_id::uri_data::print(std::string& dst) const {
if (!valid()) {
dst += "invalid-node";
return;
}
dst += to_string(value_);
}
} // <anonymous> error node_id::uri_data::serialize(serializer& sink) const {
return sink(value_);
}
// initializes singleton error node_id::uri_data::deserialize(deserializer& source) {
intrusive_ptr<node_id::data> node_id::data::create_singleton() { return source(value_);
CAF_LOG_TRACE("");
auto ifs = detail::get_mac_addresses();
std::vector<std::string> macs;
macs.reserve(ifs.size());
for (auto& i : ifs) {
macs.emplace_back(std::move(i.second));
}
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
intrusive_ptr<data> result;
result.reset(new node_id::data(detail::get_process_id(), nid), false);
return result;
} }
uint32_t node_id::process_id() const { node_id& node_id::operator=(const none_t&) {
return data_ ? data_->pid_ : invalid_process_id; data_.reset();
return *this;
} }
const node_id::host_id_type& node_id::host_id() const { node_id::node_id(intrusive_ptr<data> data) : data_(std::move(data)) {
return data_ ? data_->host_ : invalid_host_id; // nop
}
node_id::~node_id() {
// nop
} }
node_id::operator bool() const { node_id::operator bool() const {
return static_cast<bool>(data_); return static_cast<bool>(data_);
} }
int node_id::compare(const node_id& other) const noexcept {
if (this == &other || data_ == other.data_)
return 0;
if (data_ == nullptr)
return other.data_ == nullptr ? 0 : -1;
return other.data_ == nullptr ? 1 : data_->compare(*other.data_);
}
void node_id::swap(node_id& x) { void node_id::swap(node_id& x) {
data_.swap(x.data_); data_.swap(x.data_);
} }
error node_id::serialize(serializer& sink) const {
if (data_ && data_->valid()) {
if (auto err = sink(data_->implementation_id()))
return err;
return data_->serialize(sink);
}
return sink(atom(""));
}
error node_id::deserialize(deserializer& source) {
atom_value impl;
if (auto err = source(impl))
return err;
if (impl == atom("")) {
data_.reset();
return none;
}
if (impl == atom("default")) {
if (data_ == nullptr || data_->implementation_id() != atom("default"))
data_.reset(new default_data);
return data_->deserialize(source);
}
return sec::unknown_type;
}
error inspect(serializer& sink, const node_id& x) {
return x.serialize(sink);
}
error inspect(deserializer& source, node_id& x) {
return x.deserialize(source);
}
void append_to_string(std::string& str, const node_id& x) {
if (x != none)
x->print(str);
else
str += "invalid-node";
}
std::string to_string(const node_id& x) { std::string to_string(const node_id& x) {
std::string result; std::string result;
append_to_string(result, x); append_to_string(result, x);
return result; return result;
} }
void append_to_string(std::string& x, const node_id& y) { node_id make_node_id(uri from) {
if (!y) { auto ptr = make_counted<node_id::uri_data>(std::move(from));
x += "invalid-node"; return node_id{std::move(ptr)};
return; }
node_id make_node_id(uint32_t process_id,
const node_id::default_data::host_id_type& host_id) {
auto ptr = make_counted<node_id::default_data>(process_id, host_id);
return node_id{std::move(ptr)};
}
optional<node_id> make_node_id(uint32_t process_id,
const std::string& host_hash) {
using node_data = node_id::default_data;
if (host_hash.size() != node_data::host_id_size * 2)
return none;
detail::parser::ascii_to_int<16, uint8_t> xvalue;
node_data::host_id_type host_id;
for (size_t i = 0; i < node_data::host_id_size; i += 2) {
// Read two characters, each representing 4 bytes.
if (!isxdigit(host_hash[i]) || !isxdigit(host_hash[i + 1]))
return none;
host_id[i / 2] = (xvalue(host_hash[i]) << 4) | xvalue(host_hash[i + 1]);
} }
detail::append_hex(x, reinterpret_cast<const uint8_t*>(y.host_id().data()), if (!node_data::valid(host_id))
y.host_id().size()); return none;
x += '#'; return make_node_id(process_id, host_id);
x += std::to_string(y.process_id());
} }
} // namespace caf } // namespace caf
...@@ -19,6 +19,7 @@ ...@@ -19,6 +19,7 @@
#include "caf/uri.hpp" #include "caf/uri.hpp"
#include "caf/deserializer.hpp" #include "caf/deserializer.hpp"
#include "caf/detail/fnv_hash.hpp"
#include "caf/detail/parser/read_uri.hpp" #include "caf/detail/parser/read_uri.hpp"
#include "caf/detail/uri_impl.hpp" #include "caf/detail/uri_impl.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
...@@ -64,6 +65,10 @@ string_view uri::fragment() const noexcept { ...@@ -64,6 +65,10 @@ string_view uri::fragment() const noexcept {
return impl_->fragment; return impl_->fragment;
} }
size_t uri::hash_code() const noexcept {
return detail::fnv_hash(str());
}
// -- comparison --------------------------------------------------------------- // -- comparison ---------------------------------------------------------------
int uri::compare(const uri& other) const noexcept { int uri::compare(const uri& other) const noexcept {
......
...@@ -541,7 +541,7 @@ void basp_broker::set_context(connection_handle hdl) { ...@@ -541,7 +541,7 @@ void basp_broker::set_context(connection_handle hdl) {
invalid_actor_id}; invalid_actor_id};
i = ctx i = ctx
.emplace(hdl, basp::endpoint_context{basp::await_header, hdr, hdl, .emplace(hdl, basp::endpoint_context{basp::await_header, hdr, hdl,
none, 0, 0, none}) node_id{}, 0, 0, none})
.first; .first;
} }
this_context = &i->second; this_context = &i->second;
......
...@@ -18,10 +18,10 @@ ...@@ -18,10 +18,10 @@
#include "caf/io/network/ip_endpoint.hpp" #include "caf/io/network/ip_endpoint.hpp"
#include "caf/sec.hpp" #include "caf/detail/fnv_hash.hpp"
#include "caf/logger.hpp"
#include "caf/io/network/native_socket.hpp" #include "caf/io/network/native_socket.hpp"
#include "caf/logger.hpp"
#include "caf/sec.hpp"
#ifdef CAF_WINDOWS #ifdef CAF_WINDOWS
# include <winsock2.h> # include <winsock2.h>
...@@ -41,41 +41,8 @@ ...@@ -41,41 +41,8 @@
using sa_family_t = short; using sa_family_t = short;
#endif #endif
namespace { using caf::detail::fnv_hash;
using caf::detail::fnv_hash_append;
template <class SizeType = size_t>
struct hash_conf {
template <class T = SizeType>
static constexpr caf::detail::enable_if_t<(sizeof(T) == 4), size_t> basis() {
return 2166136261u;
}
template <class T = SizeType>
static constexpr caf::detail::enable_if_t<(sizeof(T) == 4), size_t> prime() {
return 16777619u;
}
template <class T = SizeType>
static constexpr caf::detail::enable_if_t<(sizeof(T) == 8), size_t> basis() {
return 14695981039346656037u;
}
template <class T = SizeType>
static constexpr caf::detail::enable_if_t<(sizeof(T) == 8), size_t> prime() {
return 1099511628211u;
}
};
constexpr uint8_t static_bytes[] = {
0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0xFF, 0xFF
};
constexpr size_t prehash(int i = 11) {
return (i > 0)
? (prehash(i - 1) * hash_conf<>::prime()) ^ static_bytes[i]
: (hash_conf<>::basis() * hash_conf<>::prime()) ^ static_bytes[i];
}
} // namespace <anonymous>
namespace caf { namespace caf {
namespace io { namespace io {
...@@ -145,32 +112,16 @@ size_t ep_hash::operator()(const sockaddr& sa) const noexcept { ...@@ -145,32 +112,16 @@ size_t ep_hash::operator()(const sockaddr& sa) const noexcept {
} }
size_t ep_hash::hash(const sockaddr_in* sa) const noexcept { size_t ep_hash::hash(const sockaddr_in* sa) const noexcept {
auto& addr = sa->sin_addr; auto result = fnv_hash(sa->sin_addr.s_addr);
size_t res = prehash(); result = fnv_hash_append(result, sa->sin_port);
// the first loop was replaces with `constexpr size_t prehash()` return result;
for (int i = 0; i < 4; ++i) {
res = res * hash_conf<>::prime();
res = res ^ ((addr.s_addr >> i) & 0xFF);
}
res = res * hash_conf<>::prime();
res = res ^ (sa->sin_port >> 1);
res = res * hash_conf<>::prime();
res = res ^ (sa->sin_port & 0xFF);
return res;
} }
size_t ep_hash::hash(const sockaddr_in6* sa) const noexcept { size_t ep_hash::hash(const sockaddr_in6* sa) const noexcept {
auto& addr = sa->sin6_addr; auto& addr = sa->sin6_addr.s6_addr;
size_t res = hash_conf<>::basis(); auto result = fnv_hash(addr, addr + 16);
for (int i = 0; i < 16; ++i) { result = fnv_hash_append(result, sa->sin6_port);
res = res * hash_conf<>::prime(); return result;
res = res ^ addr.s6_addr[i];
}
res = res * hash_conf<>::prime();
res = res ^ (sa->sin6_port >> 1);
res = res * hash_conf<>::prime();
res = res ^ (sa->sin6_port & 0xFF);
return res;
} }
bool operator==(const ip_endpoint& lhs, const ip_endpoint& rhs) { bool operator==(const ip_endpoint& lhs, const ip_endpoint& rhs) {
......
...@@ -384,8 +384,8 @@ void middleman::init(actor_system_config& cfg) { ...@@ -384,8 +384,8 @@ void middleman::init(actor_system_config& cfg) {
.add_message_type<connection_handle>("@connection_handle") .add_message_type<connection_handle>("@connection_handle")
.add_message_type<connection_passivated_msg>("@connection_passivated_msg") .add_message_type<connection_passivated_msg>("@connection_passivated_msg")
.add_message_type<acceptor_passivated_msg>("@acceptor_passivated_msg"); .add_message_type<acceptor_passivated_msg>("@acceptor_passivated_msg");
// compute and set ID for this network node // Compute and set ID for this network node.
node_id this_node{node_id::data::create_singleton()}; auto this_node = node_id::default_data::local(cfg);
system().node_.swap(this_node); system().node_.swap(this_node);
// give config access to slave mode implementation // give config access to slave mode implementation
cfg.slave_mode_fun = &middleman::exec_slave_mode; cfg.slave_mode_fun = &middleman::exec_slave_mode;
......
...@@ -62,7 +62,7 @@ node_id routing_table::lookup_direct(const connection_handle& hdl) const { ...@@ -62,7 +62,7 @@ node_id routing_table::lookup_direct(const connection_handle& hdl) const {
auto i = direct_by_hdl_.find(hdl); auto i = direct_by_hdl_.find(hdl);
if (i != direct_by_hdl_.end()) if (i != direct_by_hdl_.end())
return i->second; return i->second;
return none; return {};
} }
optional<connection_handle> optional<connection_handle>
...@@ -71,24 +71,24 @@ routing_table::lookup_direct(const node_id& nid) const { ...@@ -71,24 +71,24 @@ routing_table::lookup_direct(const node_id& nid) const {
auto i = direct_by_nid_.find(nid); auto i = direct_by_nid_.find(nid);
if (i != direct_by_nid_.end()) if (i != direct_by_nid_.end())
return i->second; return i->second;
return none; return {};
} }
node_id routing_table::lookup_indirect(const node_id& nid) const { node_id routing_table::lookup_indirect(const node_id& nid) const {
std::unique_lock<std::mutex> guard{mtx_}; std::unique_lock<std::mutex> guard{mtx_};
auto i = indirect_.find(nid); auto i = indirect_.find(nid);
if (i == indirect_.end()) if (i == indirect_.end())
return none; return {};
if (!i->second.empty()) if (!i->second.empty())
return *i->second.begin(); return *i->second.begin();
return none; return {};
} }
node_id routing_table::erase_direct(const connection_handle& hdl) { node_id routing_table::erase_direct(const connection_handle& hdl) {
std::unique_lock<std::mutex> guard{mtx_}; std::unique_lock<std::mutex> guard{mtx_};
auto i = direct_by_hdl_.find(hdl); auto i = direct_by_hdl_.find(hdl);
if (i == direct_by_hdl_.end()) if (i == direct_by_hdl_.end())
return none; return {};
direct_by_nid_.erase(i->second); direct_by_nid_.erase(i->second);
node_id result = std::move(i->second); node_id result = std::move(i->second);
direct_by_hdl_.erase(i->first); direct_by_hdl_.erase(i->first);
......
...@@ -129,12 +129,13 @@ public: ...@@ -129,12 +129,13 @@ public:
registry_ = &sys.registry(); registry_ = &sys.registry();
registry_->put((*self_)->id(), actor_cast<strong_actor_ptr>(*self_)); registry_->put((*self_)->id(), actor_cast<strong_actor_ptr>(*self_));
// first remote node is everything of this_node + 1, then +2, etc. // first remote node is everything of this_node + 1, then +2, etc.
auto pid = static_cast<node_id::default_data&>(*this_node_).process_id();
auto hid = static_cast<node_id::default_data&>(*this_node_).host_id();
for (uint32_t i = 0; i < num_remote_nodes; ++i) { for (uint32_t i = 0; i < num_remote_nodes; ++i) {
auto& n = nodes_[i]; auto& n = nodes_[i];
node_id::host_id_type tmp = this_node_.host_id(); for (auto& c : hid)
for (auto& c : tmp) ++c;
c = static_cast<uint8_t>(c + i + 1); n.id = make_node_id(++pid, hid);
n.id = node_id{this_node_.process_id() + i + 1, tmp};
n.connection = connection_handle::from_int(i + 1); n.connection = connection_handle::from_int(i + 1);
new (&n.dummy_actor) scoped_actor(sys); new (&n.dummy_actor) scoped_actor(sys);
// register all pseudo remote actors in the registry // register all pseudo remote actors in the registry
......
...@@ -82,10 +82,9 @@ struct fixture : test_coordinator_fixture<> { ...@@ -82,10 +82,9 @@ struct fixture : test_coordinator_fixture<> {
node_id last_hop; node_id last_hop;
actor testee; actor testee;
fixture() fixture() : proxies_backend(sys), proxies(sys, proxies_backend) {
: proxies_backend(sys), auto tmp = make_node_id(123, "0011223344556677889900112233445566778899");
proxies(sys, proxies_backend), last_hop = unbox(std::move(tmp));
last_hop(123, "0011223344556677889900112233445566778899") {
testee = sys.spawn<lazy_init>(testee_impl); testee = sys.spawn<lazy_init>(testee_impl);
sys.registry().put(testee.id(), testee); sys.registry().put(testee.id(), testee);
} }
......
...@@ -150,7 +150,10 @@ std::istream& operator>>(std::istream& in, node_id& x) { ...@@ -150,7 +150,10 @@ std::istream& operator>>(std::istream& in, node_id& x) {
string node_hex_id; string node_hex_id;
uint32_t pid; uint32_t pid;
if (in >> rd_line(node_hex_id, '#') >> pid) { if (in >> rd_line(node_hex_id, '#') >> pid) {
x = node_id{pid, node_hex_id}; if (auto nid = make_node_id(pid, node_hex_id))
x = std::move(*nid);
else
in.setstate(std::ios::failbit);
} }
return in; return in;
} }
......
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