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
dd74bb94
Commit
dd74bb94
authored
Apr 09, 2018
by
Joseph Noir
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Simplify the connectivity status
parent
736ec449
Changes
7
Show whitespace changes
Inline
Side-by-side
Showing
7 changed files
with
62 additions
and
113 deletions
+62
-113
libcaf_io/caf/io/basp/instance.hpp
libcaf_io/caf/io/basp/instance.hpp
+2
-4
libcaf_io/caf/io/basp/routing_table.hpp
libcaf_io/caf/io/basp/routing_table.hpp
+7
-22
libcaf_io/src/basp_broker.cpp
libcaf_io/src/basp_broker.cpp
+34
-38
libcaf_io/src/instance.cpp
libcaf_io/src/instance.cpp
+7
-10
libcaf_io/src/routing_table.cpp
libcaf_io/src/routing_table.cpp
+6
-31
libcaf_io/test/basp.cpp
libcaf_io/test/basp.cpp
+3
-4
libcaf_io/test/basp_udp.cpp
libcaf_io/test/basp_udp.cpp
+3
-4
No files found.
libcaf_io/caf/io/basp/instance.hpp
View file @
dd74bb94
...
...
@@ -335,7 +335,7 @@ public:
// Close this connection if we already established communication.
// FIXME: Should we allow multiple "connections" over different
// transport protocols?
if
(
tbl_
.
lookup
(
hdr
.
source_node
))
{
if
(
tbl_
.
lookup
(
hdr
.
source_node
)
.
hdl
)
{
CAF_LOG_INFO
(
"close connection since we already have a "
"connection: "
<<
CAF_ARG
(
hdr
.
source_node
));
callee_
.
finalize_handshake
(
hdr
.
source_node
,
aid
,
sigs
);
...
...
@@ -344,7 +344,6 @@ public:
// Add this node to our contacts.
CAF_LOG_INFO
(
"new endpoint:"
<<
CAF_ARG
(
hdr
.
source_node
));
tbl_
.
add
(
hdl
,
hdr
.
source_node
);
tbl_
.
status
(
hdr
.
source_node
,
routing_table
::
connectivity
::
established
);
// TODO: Add addresses to share with other nodes?
// Write handshake as client in response.
if
(
tcp_based
)
...
...
@@ -372,7 +371,7 @@ public:
}
}
auto
new_node
=
(
this_node
()
!=
hdr
.
source_node
&&
!
tbl_
.
lookup
(
hdr
.
source_node
));
&&
!
tbl_
.
lookup
(
hdr
.
source_node
)
.
known
);
if
(
!
new_node
)
{
if
(
tcp_based
)
{
CAF_LOG_INFO
(
"received second client handshake:"
...
...
@@ -383,7 +382,6 @@ public:
// Add this node to our contacts.
CAF_LOG_INFO
(
"new endpoint:"
<<
CAF_ARG
(
hdr
.
source_node
));
tbl_
.
add
(
hdl
,
hdr
.
source_node
);
tbl_
.
status
(
hdr
.
source_node
,
routing_table
::
connectivity
::
established
);
// TODO: Add addresses for future sharing of contact info.
}
// Since udp is unreliable we answer, maybe our message was lost.
...
...
libcaf_io/caf/io/basp/routing_table.hpp
View file @
dd74bb94
...
...
@@ -60,20 +60,10 @@ public:
virtual
~
routing_table
();
/// Describes the communication status for a remote endpoint.
enum
connectivity
{
/// There is currently an ongoing communication attempt.
pending
,
/// Communication with the node is established.
established
,
/// Communication failure detected.
failed
};
/// Result for lookups that includes the state and a potential handle.
struct
contact
{
/// Tracks the state to determine if we can sent messages or have to buffer.
connectivity
conn
;
/// Result for a lookup of a node.
struct
lookup_result
{
/// Tracks whether the node is already known.
bool
known
;
/// Servant handle to communicate with the node -- if already created.
optional
<
endpoint_handle
>
hdl
;
};
...
...
@@ -88,7 +78,7 @@ public:
/// Returns the state for communication with `nid` along with a handle
/// if communication is established or `none` if `nid` is unknown.
optional
<
contact
>
lookup
(
const
node_id
&
nid
)
const
;
lookup_result
lookup
(
const
node_id
&
nid
)
const
;
/// Adds a new endpoint to the table.
/// @pre `hdl != invalid_connection_handle && nid != none`
...
...
@@ -111,12 +101,6 @@ public:
return
parent_
;
}
/// Set the communication state of node with `nid`.
bool
status
(
const
node_id
&
nid
,
connectivity
new_status
);
/// Get the communication state of node with `nid`.
optional
<
connectivity
>
status
(
const
node_id
&
nid
);
/// Set the forwarding node that first mentioned `hdl`.
bool
forwarder
(
const
node_id
&
nid
,
endpoint_handle
hdl
);
...
...
@@ -133,7 +117,8 @@ public:
public:
/// Entry to bundle information for a remote endpoint.
struct
node_info
{
contact
details
;
/// Handle for the node if communication is established.
optional
<
endpoint_handle
>
hdl
;
/// Interfaces of the nodes for sharing with neighbors.
network
::
address_listing
addrs
;
/// The endpoint who told us about the node.
...
...
libcaf_io/src/basp_broker.cpp
View file @
dd74bb94
...
...
@@ -96,7 +96,9 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
// This member function is being called whenever we deserialize a
// payload received from a remote node; if a remote node A sends
// us a handle to a third node B, then we assume that A offers a route to B.
if
(
nid
!=
this_context
->
id
&&
!
instance
.
tbl
().
lookup
(
nid
))
auto
lr
=
instance
.
tbl
().
lookup
(
nid
);
// TODO: make sure the lr is addressed as intended, seems wrong atm.
if
(
nid
!=
this_context
->
id
&&
!
lr
.
known
)
// TODO: Try to establish communication with the new node.
CAF_CRITICAL
(
"Not implemented."
);
// TODO: Everything below has to happen once we establish communication?
...
...
@@ -104,9 +106,8 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
// send trigger an error message if we cannot contact the remote node.
// We need to tell remote side we are watching this actor now;
// use a direct route if possible, i.e., when talking to a third node.
auto
ec
=
instance
.
tbl
().
lookup
(
nid
);
// TODO: Should this communication already be established?
if
(
!
ec
)
{
if
(
lr
.
known
&&
!
lr
.
hdl
)
{
// This happens if and only if we don't have a path to `nid`
// and current_context_->hdl has been blacklisted.
CAF_LOG_INFO
(
"cannot create a proxy instance for an actor "
...
...
@@ -139,7 +140,7 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
get_buffer
(
this_context
->
hdl
),
nid
,
aid
,
ctx
.
requires_ordering
?
ctx
.
seq_outgoing
++
:
0
);
instance
.
flush
(
*
ec
->
hdl
);
instance
.
flush
(
*
lr
.
hdl
);
mm
->
notify
<
hook
::
new_remote_actor
>
(
res
);
return
res
;
}
...
...
@@ -186,22 +187,20 @@ void basp_broker_state::send_kill_proxy_instance(const node_id& nid,
CAF_LOG_TRACE
(
CAF_ARG
(
nid
)
<<
CAF_ARG
(
aid
)
<<
CAF_ARG
(
rsn
));
if
(
rsn
==
none
)
rsn
=
exit_reason
::
unknown
;
auto
ec
=
instance
.
tbl
().
lookup
(
nid
);
/// TODO: Let's assume that the handle is valid if the status is established.
if
(
!
ec
||
ec
->
conn
==
basp
::
routing_table
::
connectivity
::
failed
)
{
auto
res
=
instance
.
tbl
().
lookup
(
nid
);
if
(
!
res
.
known
)
{
CAF_LOG_INFO
(
"cannot send exit message for proxy, host unreachable"
<<
CAF_ARG
(
nid
));
return
;
}
auto
c
=
std
::
move
(
*
ec
);
if
(
c
.
conn
==
basp
::
routing_table
::
connectivity
::
established
)
{
if
(
res
.
hdl
)
{
auto
hdl
=
std
::
move
(
*
res
.
hdl
);
instance
.
write_kill_proxy
(
self
->
context
(),
get_buffer
(
*
c
.
hdl
),
get_buffer
(
hdl
),
nid
,
aid
,
rsn
,
visit
(
seq_num_visitor
{
this
},
*
c
.
hdl
));
instance
.
flush
(
*
c
.
hdl
);
visit
(
seq_num_visitor
{
this
},
hdl
));
instance
.
flush
(
hdl
);
}
else
{
// TODO: Buffer message until communication is enstablished.
buffer_type
buf
;
instance
.
write_kill_proxy
(
self
->
context
(),
buf
,
nid
,
aid
,
rsn
,
0
);
...
...
@@ -379,26 +378,24 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
auto
msg
=
make_message
(
sys_atom
::
value
,
get_atom
::
value
,
"info"
);
return
sink
(
name_atm
,
stages
,
msg
);
});
auto
ec
=
instance
.
tbl
().
lookup
(
nid
);
if
(
!
ec
||
ec
->
conn
==
basp
::
routing_table
::
connectivity
::
failed
)
{
auto
res
=
instance
.
tbl
().
lookup
(
nid
);
if
(
!
res
.
known
)
{
CAF_LOG_ERROR
(
"learned_new_node called, but no route to nid"
);
return
;
}
auto
c
=
std
::
move
(
*
ec
);
// send message to SpawnServ of remote node
basp
::
header
hdr
{
basp
::
message_type
::
dispatch_message
,
basp
::
header
::
named_receiver_flag
,
0
,
0
,
this_node
(),
nid
,
tmp
.
id
(),
invalid_actor_id
,
0
};
// sequence number only available with connectivity
if
(
c
.
conn
==
basp
::
routing_table
::
connectivity
::
established
)
{
hdr
.
sequence_number
=
visit
(
seq_num_visitor
{
this
},
*
c
.
hdl
);
if
(
res
.
hdl
)
{
auto
hdl
=
std
::
move
(
*
res
.
hdl
);
hdr
.
sequence_number
=
visit
(
seq_num_visitor
{
this
},
hdl
);
// writing std::numeric_limits<actor_id>::max() is a hack to get
// this send-to-named-actor feature working with older CAF releases
instance
.
write
(
self
->
context
(),
get_buffer
(
*
c
.
hdl
),
hdr
,
&
writer
);
instance
.
flush
(
*
c
.
hdl
);
instance
.
write
(
self
->
context
(),
get_buffer
(
hdl
),
hdr
,
&
writer
);
instance
.
flush
(
hdl
);
}
else
{
// TODO: Implement this, can it happen?
buffer_type
buf
;
instance
.
write
(
self
->
context
(),
buf
,
hdr
,
&
writer
);
pending_connectivity
[
nid
].
emplace_back
(
std
::
move
(
buf
));
...
...
@@ -568,7 +565,7 @@ void basp_broker_state::send_buffered_messages(execution_unit* ctx,
datagram_handle
hdl
)
{
if
(
pending_connectivity
.
count
(
nid
)
>
0
)
{
for
(
auto
&
msg
:
pending_connectivity
[
nid
])
{
// TODO:
add sequence number
// TODO:
Validate that this actually works.
auto
seq_num
=
next_sequence_number
(
hdl
);
auto
seq_size
=
sizeof
(
basp
::
sequence_type
);
auto
offset
=
basp
::
header_size
-
seq_size
;
...
...
@@ -605,9 +602,9 @@ basp_broker_state::get_buffer(connection_handle hdl) {
basp_broker_state
::
buffer_type
&
basp_broker_state
::
get_buffer
(
node_id
nid
)
{
auto
ec
=
instance
.
tbl
().
lookup
(
nid
);
if
(
ec
&&
ec
->
conn
==
basp
::
routing_table
::
connectivity
::
established
&&
ec
->
hdl
)
{
return
get_buffer
(
*
(
ec
->
hdl
)
);
auto
res
=
instance
.
tbl
().
lookup
(
nid
);
if
(
res
.
known
&&
res
.
hdl
)
{
return
get_buffer
(
*
res
.
hdl
);
}
auto
msgs
=
pending_connectivity
[
nid
];
msgs
.
emplace_back
();
...
...
@@ -778,12 +775,11 @@ behavior basp_broker::make_behavior() {
<<
", "
<<
CAF_ARG
(
msg
));
if
(
!
src
)
return
sec
::
invalid_argument
;
auto
ec
=
this
->
state
.
instance
.
tbl
().
lookup
(
dest_node
);
if
(
!
ec
||
ec
->
conn
==
basp
::
routing_table
::
connectivity
::
failed
)
{
auto
res
=
this
->
state
.
instance
.
tbl
().
lookup
(
dest_node
);
if
(
!
res
.
known
)
{
CAF_LOG_ERROR
(
"host unknown or unreachable"
);
return
sec
::
no_route_to_receiving_node
;
}
auto
c
=
std
::
move
(
*
ec
);
if
(
system
().
node
()
==
src
->
node
())
system
().
registry
().
put
(
src
->
id
(),
src
);
auto
writer
=
make_callback
([
&
](
serializer
&
sink
)
->
error
{
...
...
@@ -794,13 +790,13 @@ behavior basp_broker::make_behavior() {
0
,
cme
->
mid
.
integer_value
(),
state
.
this_node
(),
dest_node
,
src
->
id
(),
invalid_actor_id
,
0
};
if
(
c
.
conn
==
basp
::
routing_table
::
connectivity
::
established
)
{
hdr
.
sequence_number
=
visit
(
seq_num_visitor
{
&
state
},
*
c
.
hdl
);
state
.
instance
.
write
(
context
(),
state
.
get_buffer
(
*
c
.
hdl
),
if
(
res
.
hdl
)
{
auto
hdl
=
std
::
move
(
*
res
.
hdl
);
hdr
.
sequence_number
=
visit
(
seq_num_visitor
{
&
state
},
hdl
);
state
.
instance
.
write
(
context
(),
state
.
get_buffer
(
hdl
),
hdr
,
&
writer
);
state
.
instance
.
flush
(
*
c
.
hdl
);
state
.
instance
.
flush
(
hdl
);
}
else
{
// TODO: Buffer the message in the basp broker.
std
::
vector
<
char
>
buf
;
state
.
instance
.
write
(
context
(),
buf
,
hdr
,
&
writer
);
state
.
pending_connectivity
[
dest_node
].
emplace_back
(
buf
);
...
...
@@ -935,10 +931,10 @@ behavior basp_broker::make_behavior() {
->
std
::
tuple
<
node_id
,
std
::
string
,
uint16_t
>
{
std
::
string
addr
;
uint16_t
port
=
0
;
auto
ec
=
state
.
instance
.
tbl
().
lookup
(
x
);
if
(
ec
&&
ec
->
hdl
)
{
addr
=
visit
(
addr_visitor
{
this
},
*
ec
->
hdl
);
port
=
visit
(
port_visitor
{
this
},
*
ec
->
hdl
);
auto
res
=
state
.
instance
.
tbl
().
lookup
(
x
);
if
(
res
.
known
&&
res
.
hdl
)
{
addr
=
visit
(
addr_visitor
{
this
},
*
res
.
hdl
);
port
=
visit
(
port_visitor
{
this
},
*
res
.
hdl
);
}
return
std
::
make_tuple
(
x
,
std
::
move
(
addr
),
port
);
},
...
...
libcaf_io/src/instance.cpp
View file @
dd74bb94
...
...
@@ -252,13 +252,11 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
CAF_LOG_TRACE
(
CAF_ARG
(
sender
)
<<
CAF_ARG
(
receiver
)
<<
CAF_ARG
(
mid
)
<<
CAF_ARG
(
msg
));
CAF_ASSERT
(
receiver
&&
system
().
node
()
!=
receiver
->
node
());
auto
ec
=
tbl_
.
lookup
(
receiver
->
node
());
/// TODO: Let's assume that the handle is valid if the status is established.
if
(
!
ec
||
ec
->
conn
==
routing_table
::
connectivity
::
failed
)
{
auto
res
=
tbl_
.
lookup
(
receiver
->
node
());
if
(
!
res
.
known
)
{
notify
<
hook
::
message_sending_failed
>
(
sender
,
receiver
,
mid
,
msg
);
return
false
;
}
auto
c
=
std
::
move
(
*
ec
);
auto
writer
=
make_callback
([
&
](
serializer
&
sink
)
->
error
{
return
sink
(
const_cast
<
std
::
vector
<
strong_actor_ptr
>&>
(
forwarding_stack
),
const_cast
<
message
&>
(
msg
));
...
...
@@ -267,15 +265,14 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
sender
?
sender
->
node
()
:
this_node
(),
receiver
->
node
(),
sender
?
sender
->
id
()
:
invalid_actor_id
,
receiver
->
id
(),
0
};
if
(
c
.
conn
==
routing_table
::
connectivity
::
established
)
{
hdr
.
sequence_number
=
visit
(
seq_num_visitor
{
callee_
},
*
c
.
hdl
);
write
(
ctx
,
callee_
.
get_buffer
(
*
c
.
hdl
),
hdr
,
&
writer
);
flush
(
*
c
.
hdl
);
if
(
res
.
hdl
)
{
auto
hdl
=
std
::
move
(
*
res
.
hdl
);
hdr
.
sequence_number
=
visit
(
seq_num_visitor
{
callee_
},
hdl
);
write
(
ctx
,
callee_
.
get_buffer
(
hdl
),
hdr
,
&
writer
);
flush
(
hdl
);
notify
<
hook
::
message_sent
>
(
sender
,
receiver
->
node
(),
receiver
,
mid
,
msg
);
return
true
;
}
else
{
// lr.cs == routing_table::communication::pending
// TODO: Buffer the message in the basp broker.
write
(
ctx
,
callee_
.
get_buffer
(
receiver
->
node
()),
hdr
,
&
writer
);
// TODO: should the hook really be called here, or should we delay this
// until communication is established?
...
...
libcaf_io/src/routing_table.cpp
View file @
dd74bb94
...
...
@@ -37,17 +37,11 @@ node_id routing_table::lookup(const endpoint_handle& hdl) const {
return
get_opt
(
nid_by_hdl_
,
hdl
,
none
);
}
optional
<
routing_table
::
contact
>
routing_table
::
lookup
(
const
node_id
&
nid
)
const
{
routing_table
::
lookup_result
routing_table
::
lookup
(
const
node_id
&
nid
)
const
{
auto
i
=
node_information_base_
.
find
(
nid
);
if
(
i
!=
node_information_base_
.
end
())
return
i
->
second
.
details
;
/*
auto i = hdl_by_nid_.find(nid);
if (i != hdl_by_nid_.end())
return i->second;
*/
return
none
;
return
{
true
,
i
->
second
.
hdl
};
return
{
false
,
none
};
}
void
routing_table
::
erase
(
const
endpoint_handle
&
hdl
,
erase_callback
&
cb
)
{
...
...
@@ -68,16 +62,14 @@ void routing_table::add(const endpoint_handle& hdl, const node_id& nid) {
CAF_ASSERT
(
node_information_base_
.
count
(
nid
)
==
0
);
nid_by_hdl_
.
emplace
(
hdl
,
nid
);
//hdl_by_nid_.emplace(nid, hdl);
node_information_base_
[
nid
]
=
node_info
{{
connectivity
::
established
,
hdl
},
{},
none
};
node_information_base_
[
nid
]
=
node_info
{
hdl
,
{},
none
};
parent_
->
parent
().
notify
<
hook
::
new_connection_established
>
(
nid
);
}
void
routing_table
::
add
(
const
node_id
&
nid
)
{
//CAF_ASSERT(hdl_by_nid_.count(nid) == 0);
CAF_ASSERT
(
node_information_base_
.
count
(
nid
)
==
0
);
node_information_base_
[
nid
]
=
node_info
{{
connectivity
::
pending
,
none
},
{},
none
};
node_information_base_
[
nid
]
=
node_info
{
none
,
{},
none
};
// TODO: Some new related hook?
//parent_->parent().notify<hook::new_connection_established>(nid);
}
...
...
@@ -85,25 +77,8 @@ void routing_table::add(const node_id& nid) {
bool
routing_table
::
reachable
(
const
node_id
&
dest
)
{
auto
i
=
node_information_base_
.
find
(
dest
);
if
(
i
!=
node_information_base_
.
end
())
return
i
->
second
.
details
.
conn
==
connectivity
::
established
;
return
false
;
}
bool
routing_table
::
status
(
const
node_id
&
nid
,
routing_table
::
connectivity
new_status
)
{
auto
i
=
node_information_base_
.
find
(
nid
);
if
(
i
==
node_information_base_
.
end
())
return
i
->
second
.
hdl
!=
none
;
return
false
;
i
->
second
.
details
.
conn
=
new_status
;
return
true
;
}
optional
<
routing_table
::
connectivity
>
routing_table
::
status
(
const
node_id
&
nid
)
{
auto
i
=
node_information_base_
.
find
(
nid
);
if
(
i
==
node_information_base_
.
end
())
return
none
;
return
i
->
second
.
details
.
conn
;
}
bool
routing_table
::
forwarder
(
const
node_id
&
nid
,
...
...
libcaf_io/test/basp.cpp
View file @
dd74bb94
...
...
@@ -310,10 +310,9 @@ public:
make_message
(
sys_atom
::
value
,
get_atom
::
value
,
"info"
));
// test whether basp instance correctly updates the
// routing table upon receiving client handshakes
auto
ec
=
tbl
().
lookup
(
n
.
id
);
CAF_REQUIRE
(
ec
);
CAF_REQUIRE
(
ec
->
hdl
);
CAF_CHECK_EQUAL
(
*
ec
->
hdl
,
n
.
connection
);
auto
res
=
tbl
().
lookup
(
n
.
id
);
CAF_REQUIRE
(
res
.
hdl
);
CAF_CHECK_EQUAL
(
*
res
.
hdl
,
n
.
connection
);
}
std
::
pair
<
basp
::
header
,
buffer
>
read_from_out_buf
(
connection_handle
hdl
)
{
...
...
libcaf_io/test/basp_udp.cpp
View file @
dd74bb94
...
...
@@ -309,10 +309,9 @@ public:
make_message
(
sys_atom
::
value
,
get_atom
::
value
,
"info"
));
// test whether basp instance correctly updates the
// routing table upon receiving client handshakes
auto
ec
=
tbl
().
lookup
(
n
.
id
);
CAF_REQUIRE
(
ec
);
CAF_REQUIRE
(
ec
->
hdl
);
CAF_CHECK_EQUAL
(
*
ec
->
hdl
,
n
.
endpoint
);
auto
res
=
tbl
().
lookup
(
n
.
id
);
CAF_REQUIRE
(
res
.
hdl
);
CAF_CHECK_EQUAL
(
*
res
.
hdl
,
n
.
endpoint
);
}
std
::
pair
<
basp
::
header
,
buffer
>
read_from_out_buf
(
datagram_handle
hdl
)
{
...
...
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