Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
A
Actor Framework
Project overview
Project overview
Details
Activity
Releases
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Issues
0
Issues
0
List
Boards
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Operations
Operations
Metrics
Environments
Analytics
Analytics
CI / CD
Repository
Value Stream
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
cpp-libs
Actor Framework
Commits
52409b7e
Commit
52409b7e
authored
Jul 31, 2019
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Reimplement node_id as opaque identifier
parent
c58b6a08
Changes
10
Show whitespace changes
Inline
Side-by-side
Showing
10 changed files
with
313 additions
and
256 deletions
+313
-256
libcaf_core/caf/actor_control_block.hpp
libcaf_core/caf/actor_control_block.hpp
+6
-6
libcaf_core/caf/atom.hpp
libcaf_core/caf/atom.hpp
+4
-1
libcaf_core/caf/node_id.hpp
libcaf_core/caf/node_id.hpp
+136
-121
libcaf_core/src/atom.cpp
libcaf_core/src/atom.cpp
+4
-0
libcaf_core/src/node_id.cpp
libcaf_core/src/node_id.cpp
+147
-114
libcaf_io/src/basp_broker.cpp
libcaf_io/src/basp_broker.cpp
+1
-1
libcaf_io/src/middleman.cpp
libcaf_io/src/middleman.cpp
+2
-2
libcaf_io/src/routing_table.cpp
libcaf_io/src/routing_table.cpp
+5
-5
libcaf_io/test/basp.cpp
libcaf_io/test/basp.cpp
+5
-4
libcaf_io/test/worker.cpp
libcaf_io/test/worker.cpp
+3
-2
No files found.
libcaf_core/caf/actor_control_block.hpp
View file @
52409b7e
...
@@ -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_i
d.hpp"
#include "caf/
fw
d.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
{
...
...
libcaf_core/caf/atom.hpp
View file @
52409b7e
...
@@ -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
libcaf_core/caf/node_id.hpp
View file @
52409b7e
...
@@ -23,135 +23,165 @@
...
@@ -23,135 +23,165 @@
#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/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
class
node_id
:
detail
::
comparable
<
node_id
>
,
/// the running system-level process on that machine.
detail
::
comparable
<
node_id
,
none_t
>
{
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:
// static intrusive_ptr<data> create_singleton();
node_id
(
const
node_id
&
)
=
default
;
~
data
()
override
;
node_id
(
const
none_t
&
)
;
virtual
bool
valid
()
const
noexcept
=
0
;
node_id
&
operator
=
(
const
node_id
&
)
=
default
;
virtual
size_t
hash_code
()
const
noexcept
=
0
;
node_id
&
operator
=
(
const
none_t
&
);
virtual
atom_value
implementation_id
()
const
noexcept
=
0
;
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 ---------------------------------------------
data
()
;
bool
valid
()
const
noexcept
override
;
data
(
uint32_t
procid
,
host_id_type
hid
)
;
size_t
hash_code
()
const
noexcept
override
;
data
(
uint32_t
procid
,
const
std
::
string
&
hash
)
;
atom_value
implementation_id
()
const
noexcept
override
;
data
(
const
data
&
)
=
default
;
int
compare
(
const
data
&
other
)
const
noexcept
override
;
data
&
operator
=
(
const
data
&
)
=
default
;
void
print
(
std
::
string
&
dst
)
const
override
;
bool
valid
()
const
;
error
serialize
(
serializer
&
sink
)
const
override
;
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>
// -- constructors, destructors, and assignment operators --------------------
int
compare
(
const
node_id
&
other
)
const
;
// "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
;
/// Compares this instance as if comparing to a default-constructed
/// `node_id`.
/// @returns `compare(node_id{})`.
int
compare
(
const
none_t
&
)
const
;
/// 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
();
}
const
data
*
operator
->
()
const
noexcept
{
return
data_
.
get
();
}
data
&
operator
*
()
noexcept
{
return
*
data_
;
}
}
template
<
class
Inspector
>
const
data
&
operator
*
()
const
noexcept
{
friend
detail
::
enable_if_t
<
Inspector
::
writes_state
,
return
*
data_
;
typename
Inspector
::
result_type
>
inspect
(
Inspector
&
f
,
node_id
&
x
)
{
data
tmp
;
// write changes to tmp back to x at scope exit
auto
sg
=
detail
::
make_scope_guard
([
&
]
{
if
(
!
tmp
.
valid
())
x
.
data_
.
reset
();
else
if
(
!
x
||
!
x
.
data_
->
unique
())
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 +191,45 @@ private:
...
@@ -161,60 +191,45 @@ private:
};
};
/// @relates node_id
/// @relates node_id
inline
bool
operator
==
(
const
node_id
&
x
,
none_t
)
{
error
inspect
(
serializer
&
sink
,
const
node_id
&
x
);
return
!
x
;
}
/// @relates node_id
/// @relates node_id
inline
bool
operator
==
(
none_t
,
const
node_id
&
x
)
{
error
inspect
(
deserializer
&
source
,
node_id
&
x
);
return
!
x
;
}
/// @relates node_id
/// @relates node_id
inline
bool
operator
!=
(
const
node_id
&
x
,
none_t
)
{
std
::
string
to_string
(
const
node_id
&
x
);
return
static_cast
<
bool
>
(
x
);
}
/// Appends `x` in human-readable string representation to `str`.
/// @relates node_id
/// @relates node_id
inline
bool
operator
!=
(
none_t
,
const
node_id
&
x
)
{
void
append_to_string
(
std
::
string
&
str
,
const
node_id
&
x
);
return
static_cast
<
bool
>
(
x
);
}
inline
bool
operator
==
(
const
node_id
&
lhs
,
const
node_id
&
rhs
)
{
return
lhs
.
compare
(
rhs
)
==
0
;
}
inline
bool
operator
!=
(
const
node_id
&
lhs
,
const
node_id
&
rhs
)
{
return
!
(
lhs
==
rhs
);
}
inline
bool
operator
<
(
const
node_id
&
lhs
,
const
node_id
&
rhs
)
{
return
lhs
.
compare
(
rhs
)
<
0
;
}
/// 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 `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
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
/// @relates node_id
void
append_to_string
(
std
::
string
&
x
,
const
node_id
&
y
);
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
libcaf_core/src/atom.cpp
View file @
52409b7e
...
@@ -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
libcaf_core/src/node_id.cpp
View file @
52409b7e
...
@@ -16,165 +16,185 @@
...
@@ -16,165 +16,185 @@
* 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
;
}
// <anonymous>
}
// 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
&
node_id
::
operator
=
(
const
none_t
&
)
{
:
pid_
(
procid
),
data_
.
reset
();
host_
(
hid
)
{
return
*
this
;
// nop
}
}
node_id
::
data
::
data
(
uint32_t
procid
,
const
std
::
string
&
hash
)
:
pid_
(
procid
)
{
node_id
::
node_id
(
intrusive_ptr
<
data
>
data
)
:
data_
(
std
::
move
(
data
))
{
if
(
hash
.
size
()
!=
(
host_id_size
*
2
))
{
// nop
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
()
{
node_id
::
~
node_id
()
{
// nop
// nop
}
}
bool
node_id
::
data
::
valid
()
const
{
node_id
::
operator
bool
()
const
{
auto
is_zero
=
[](
uint8_t
x
)
{
return
x
==
0
;
};
return
static_cast
<
bool
>
(
data_
);
return
pid_
!=
0
&&
!
std
::
all_of
(
host_
.
begin
(),
host_
.
end
(),
is_zero
);
}
}
namespace
{
int
node_id
::
compare
(
const
none_t
&
)
const
{
// Invalid instances are always smaller.
return
data_
?
1
:
0
;
}
std
::
atomic
<
uint8_t
>
system_id
;
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_
);
}
}
// <anonymous>
void
node_id
::
swap
(
node_id
&
x
)
{
data_
.
swap
(
x
.
data_
);
}
// initializes singleton
error
node_id
::
serialize
(
serializer
&
sink
)
const
{
intrusive_ptr
<
node_id
::
data
>
node_id
::
data
::
create_singleton
()
{
if
(
data_
&&
data_
->
valid
())
{
CAF_LOG_TRACE
(
""
);
if
(
auto
err
=
sink
(
data_
->
implementation_id
()))
auto
ifs
=
detail
::
get_mac_addresses
();
return
err
;
std
::
vector
<
std
::
string
>
macs
;
return
data_
->
serialize
(
sink
);
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
();
return
sink
(
atom
(
""
));
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
{
error
node_id
::
deserialize
(
deserializer
&
source
)
{
return
data_
?
data_
->
pid_
:
invalid_process_id
;
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
;
}
}
const
node_id
::
host_id_type
&
node_id
::
host_id
()
const
{
error
inspect
(
serializer
&
sink
,
const
node_id
&
x
)
{
return
data_
?
data_
->
host_
:
invalid_host_id
;
return
x
.
serialize
(
sink
)
;
}
}
node_id
::
operator
bool
()
const
{
error
inspect
(
deserializer
&
source
,
node_id
&
x
)
{
return
static_cast
<
bool
>
(
data_
);
return
x
.
deserialize
(
source
);
}
}
void
node_id
::
swap
(
node_id
&
x
)
{
void
append_to_string
(
std
::
string
&
str
,
const
node_id
&
x
)
{
data_
.
swap
(
x
.
data_
);
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
)
{
...
@@ -183,15 +203,28 @@ std::string to_string(const node_id& x) {
...
@@ -183,15 +203,28 @@ std::string to_string(const node_id& x) {
return
result
;
return
result
;
}
}
void
append_to_string
(
std
::
string
&
x
,
const
node_id
&
y
)
{
node_id
make_node_id
(
uint32_t
process_id
,
if
(
!
y
)
{
const
node_id
::
default_data
::
host_id_type
&
host_id
)
{
x
+=
"invalid-node"
;
auto
ptr
=
make_counted
<
node_id
::
default_data
>
(
process_id
,
host_id
);
return
;
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
libcaf_io/src/basp_broker.cpp
View file @
52409b7e
...
@@ -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
,
no
ne
,
0
,
0
,
none
})
no
de_id
{}
,
0
,
0
,
none
})
.
first
;
.
first
;
}
}
this_context
=
&
i
->
second
;
this_context
=
&
i
->
second
;
...
...
libcaf_io/src/middleman.cpp
View file @
52409b7e
...
@@ -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
;
...
...
libcaf_io/src/routing_table.cpp
View file @
52409b7e
...
@@ -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
);
...
...
libcaf_io/test/basp.cpp
View file @
52409b7e
...
@@ -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
...
...
libcaf_io/test/worker.cpp
View file @
52409b7e
...
@@ -84,8 +84,9 @@ struct fixture : test_coordinator_fixture<> {
...
@@ -84,8 +84,9 @@ struct fixture : test_coordinator_fixture<> {
fixture
()
fixture
()
:
proxies_backend
(
sys
),
:
proxies_backend
(
sys
),
proxies
(
sys
,
proxies_backend
),
proxies
(
sys
,
proxies_backend
)
{
last_hop
(
123
,
"0011223344556677889900112233445566778899"
)
{
auto
tmp
=
make_node_id
(
123
,
"0011223344556677889900112233445566778899"
);
last_hop
=
unbox
(
std
::
move
(
tmp
));
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
);
}
}
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment