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
98b77d32
Commit
98b77d32
authored
Oct 21, 2019
by
Jakob Otto
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Update branch to new API
parent
a3fc9442
Changes
6
Hide whitespace changes
Inline
Side-by-side
Showing
6 changed files
with
204 additions
and
77 deletions
+204
-77
libcaf_net/caf/net/datagram_transport.hpp
libcaf_net/caf/net/datagram_transport.hpp
+109
-40
libcaf_net/caf/net/fwd.hpp
libcaf_net/caf/net/fwd.hpp
+1
-1
libcaf_net/caf/net/transport_worker_dispatcher.hpp
libcaf_net/caf/net/transport_worker_dispatcher.hpp
+53
-22
libcaf_net/caf/net/udp_datagram_socket.hpp
libcaf_net/caf/net/udp_datagram_socket.hpp
+15
-0
libcaf_net/test/datagram_transport.cpp
libcaf_net/test/datagram_transport.cpp
+24
-12
libcaf_net/test/transport_worker_dispatcher.cpp
libcaf_net/test/transport_worker_dispatcher.cpp
+2
-2
No files found.
libcaf_net/caf/net/datagram_transport.hpp
View file @
98b77d32
...
...
@@ -28,6 +28,7 @@
#include "caf/fwd.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/logger.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp"
...
...
@@ -46,14 +47,18 @@ class datagram_transport {
public:
// -- member types -----------------------------------------------------------
using
buffer_type
=
std
::
vector
<
byte
>
;
using
buffer_cache_type
=
std
::
vector
<
buffer_type
>
;
using
factory_type
=
Factory
;
using
transport_type
=
datagram_transport
;
using
application_type
=
typename
Factory
::
application_type
;
using
dispatcher_type
=
transport_worker_dispatcher
<
factory_type
,
ip_endpoint
>
;
using
dispatcher_type
=
transport_worker_dispatcher
<
datagram_transport
,
factory_type
,
ip_endpoint
>
;
// -- constructors, destructors, and assignment operators --------------------
...
...
@@ -62,16 +67,46 @@ public:
handle_
(
handle
),
max_consecutive_reads_
(
0
),
read_threshold_
(
1024
),
collected_
(
0
),
max_
(
1024
),
rd_flag_
(
receive_policy_flag
::
exactly
)
{
rd_flag_
(
receive_policy_flag
::
exactly
),
manager_
(
nullptr
)
{
// nop
}
// -- properties -------------------------------------------------------------
udp_datagram_socket
handle
()
const
noexcept
{
return
handle_
;
}
actor_system
&
system
()
{
return
manager
().
system
();
}
application_type
&
application
()
{
// TODO: This wont work. We need information on which application is wanted
return
dispatcher_
.
application
();
}
transport_type
&
transport
()
{
return
*
this
;
}
endpoint_manager
&
manager
()
{
return
*
manager_
;
}
// -- public member functions ------------------------------------------------
template
<
class
Parent
>
error
init
(
Parent
&
parent
)
{
auto
&
cfg
=
system
().
config
();
auto
max_header_bufs
=
get_or
(
cfg
,
"middleman.max-header-buffers"
,
defaults
::
middleman
::
max_header_buffers
);
header_bufs_
.
reserve
(
max_header_bufs
);
auto
max_payload_bufs
=
get_or
(
cfg
,
"middleman.max-payload-buffers"
,
defaults
::
middleman
::
max_payload_buffers
);
payload_bufs_
.
reserve
(
max_payload_bufs
);
if
(
auto
err
=
dispatcher_
.
init
(
parent
))
return
err
;
parent
.
mask_add
(
operation
::
read
);
...
...
@@ -85,7 +120,7 @@ public:
if
(
auto
res
=
get_if
<
std
::
pair
<
size_t
,
ip_endpoint
>>
(
&
ret
))
{
auto
num_bytes
=
res
->
first
;
auto
ep
=
res
->
second
;
collected_
+=
(
num_bytes
>
0
)
?
static_cast
<
size_t
>
(
num_bytes
)
:
0
;
read_buf_
.
resize
(
num_bytes
)
;
dispatcher_
.
handle_data
(
parent
,
make_span
(
read_buf_
),
std
::
move
(
ep
));
prepare_next_read
();
}
else
{
...
...
@@ -113,18 +148,27 @@ public:
}
template
<
class
Parent
>
void
resolve
(
Parent
&
parent
,
const
std
::
string
&
path
,
actor
listener
)
{
dispatcher_
.
resolve
(
parent
,
path
,
listener
);
void
resolve
(
Parent
&
,
const
uri
&
locator
,
const
actor
&
listener
)
{
dispatcher_
.
resolve
(
*
this
,
locator
,
listener
);
}
template
<
class
Parent
>
void
new_proxy
(
Parent
&
,
const
node_id
&
peer
,
actor_id
id
)
{
dispatcher_
.
new_proxy
(
*
this
,
peer
,
id
);
}
template
<
class
Parent
>
void
timeout
(
Parent
&
parent
,
atom_value
value
,
uint64_t
id
)
{
auto
decorator
=
make_write_packet_decorator
(
*
this
,
parent
);
dispatcher_
.
timeout
(
decorator
,
value
,
id
);
void
local_actor_down
(
Parent
&
,
const
node_id
&
peer
,
actor_id
id
,
error
reason
)
{
dispatcher_
.
local_actor_down
(
*
this
,
peer
,
id
,
std
::
move
(
reason
)
);
}
template
<
class
Parent
>
uint64_t
set_timeout
(
uint64_t
timeout_id
,
ip_endpoint
ep
)
{
void
timeout
(
Parent
&
,
atom_value
value
,
uint64_t
id
)
{
dispatcher_
.
timeout
(
*
this
,
value
,
id
);
}
void
set_timeout
(
uint64_t
timeout_id
,
ip_endpoint
ep
)
{
dispatcher_
.
set_timeout
(
timeout_id
,
ep
);
}
...
...
@@ -132,13 +176,8 @@ public:
dispatcher_
.
handle_error
(
code
);
}
udp_datagram_socket
handle
()
const
noexcept
{
return
handle_
;
}
void
prepare_next_read
()
{
read_buf_
.
clear
();
collected_
=
0
;
// This cast does nothing, but prevents a weird compiler error on GCC
// <= 4.9.
// TODO: remove cast when dropping support for GCC 4.9.
...
...
@@ -170,54 +209,84 @@ public:
prepare_next_read
();
}
void
write_packet
(
span
<
const
byte
>
header
,
span
<
const
byte
>
payload
,
typename
dispatcher_type
::
id_type
id
)
{
std
::
vector
<
byte
>
buf
;
buf
.
reserve
(
header
.
size
()
+
payload
.
size
());
buf
.
insert
(
buf
.
end
(),
header
.
begin
(),
header
.
end
());
buf
.
insert
(
buf
.
end
(),
payload
.
begin
(),
payload
.
end
());
packet_queue_
.
emplace_back
(
id
,
std
::
move
(
buf
));
void
write_packet
(
ip_endpoint
ep
,
span
<
buffer_type
*>
buffers
)
{
CAF_ASSERT
(
!
buffers
.
empty
());
if
(
packet_queue_
.
empty
())
manager
().
register_writing
();
// By convention, the first buffer is a header buffer. Every other buffer is
// a payload buffer.
packet_queue_
.
emplace_back
(
ep
,
buffers
);
}
// -- buffer management ------------------------------------------------------
buffer_type
next_header_buffer
()
{
return
next_buffer_impl
(
header_bufs_
);
}
buffer_type
next_payload_buffer
()
{
return
next_buffer_impl
(
payload_bufs_
);
}
/// Helper struct for managing outgoing packets
struct
packet
{
ip_endpoint
destination
;
std
::
vector
<
byte
>
bytes
;
size_t
payload_buf_num
;
buffer_cache_type
bytes
;
packet
(
ip_endpoint
destination
,
std
::
vector
<
byte
>
bytes
)
:
destination
(
destination
),
bytes
(
std
::
move
(
bytes
))
{
// nop
packet
(
ip_endpoint
destination
,
span
<
buffer_type
*>
bufs
)
:
destination
(
destination
)
{
payload_buf_num
=
bufs
.
size
()
-
1
;
for
(
auto
buf
:
bufs
)
bytes
.
emplace_back
(
false
,
std
::
move
(
*
buf
));
}
};
private:
// -- utility functions ------------------------------------------------------
static
buffer_type
next_buffer_impl
(
buffer_cache_type
cache
)
{
if
(
cache
.
empty
())
{
return
{};
}
auto
buf
=
std
::
move
(
cache
.
back
());
cache
.
pop_back
();
return
buf
;
}
bool
write_some
()
{
if
(
packet_queue_
.
empty
())
return
false
;
auto
&
next_packet
=
packet_queue_
.
front
();
auto
send_res
=
write
(
handle_
,
make_span
(
next_packet
.
bytes
),
next_packet
.
destination
);
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
send_res
))
{
CAF_LOG_DEBUG
(
CAF_ARG
(
handle_
.
id
)
<<
CAF_ARG
(
*
num_bytes
));
packet_queue_
.
pop_front
();
return
true
;
while
(
!
packet_queue_
.
empty
())
{
auto
&
next_packet
=
packet_queue_
.
front
();
auto
send_res
=
write
(
handle_
,
next_packet
.
bytes
,
next_packet
.
destination
);
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
send_res
))
{
CAF_LOG_DEBUG
(
CAF_ARG
(
handle_
.
id
)
<<
CAF_ARG
(
*
num_bytes
));
packet_queue_
.
pop_front
();
return
true
;
}
auto
err
=
get
<
sec
>
(
send_res
);
CAF_LOG_DEBUG
(
"send failed"
<<
CAF_ARG
(
err
));
dispatcher_
.
handle_error
(
err
);
}
auto
err
=
get
<
sec
>
(
send_res
);
CAF_LOG_DEBUG
(
"send failed"
<<
CAF_ARG
(
err
));
dispatcher_
.
handle_error
(
err
);
return
false
;
}
dispatcher_type
dispatcher_
;
udp_datagram_socket
handle_
;
buffer_cache_type
header_bufs_
;
buffer_cache_type
payload_bufs_
;
std
::
vector
<
byte
>
read_buf_
;
std
::
deque
<
packet
>
packet_queue_
;
size_t
max_consecutive_reads_
;
size_t
read_threshold_
;
size_t
collected_
;
//
size_t collected_;
size_t
max_
;
receive_policy_flag
rd_flag_
;
endpoint_manager
*
manager_
;
};
}
// namespace net
...
...
libcaf_net/caf/net/fwd.hpp
View file @
98b77d32
...
...
@@ -30,7 +30,7 @@ namespace net {
template
<
class
Application
,
class
IdType
=
unit_t
>
class
transport_worker
;
template
<
class
Application
,
class
IdType
=
unit_t
>
template
<
class
Transport
,
class
Application
,
class
IdType
=
unit_t
>
class
transport_worker_dispatcher
;
// -- classes ------------------------------------------------------------------
...
...
libcaf_net/caf/net/transport_worker_dispatcher.hpp
View file @
98b77d32
...
...
@@ -18,6 +18,7 @@
#pragma once
#include <caf/logger.hpp>
#include <unordered_map>
#include "caf/byte.hpp"
...
...
@@ -33,7 +34,7 @@ namespace caf {
namespace
net
{
/// implements a dispatcher that dispatches between transport and workers.
template
<
class
ApplicationFactory
,
class
IdType
>
template
<
class
Transport
,
class
ApplicationFactory
,
class
IdType
>
class
transport_worker_dispatcher
{
public:
// -- member types -----------------------------------------------------------
...
...
@@ -42,6 +43,8 @@ public:
using
factory_type
=
ApplicationFactory
;
using
transport_type
=
Transport
;
using
application_type
=
typename
ApplicationFactory
::
application_type
;
using
worker_type
=
transport_worker
<
application_type
,
id_type
>
;
...
...
@@ -50,7 +53,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
transport_worker_dispatcher
(
factory_type
factory
)
explicit
transport_worker_dispatcher
(
factory_type
factory
)
:
factory_
(
std
::
move
(
factory
))
{
// nop
}
...
...
@@ -65,42 +68,48 @@ public:
template
<
class
Parent
>
error
handle_data
(
Parent
&
parent
,
span
<
byte
>
data
,
id_type
id
)
{
auto
it
=
workers_by_id_
.
fin
d
(
id
);
if
(
it
==
workers_by_id_
.
end
()
)
{
// TODO: where to get
node_id
from here?
auto
worker
=
find_by_i
d
(
id
);
if
(
!
worker
)
{
// TODO: where to get
id_type
from here?
add_new_worker
(
parent
,
node_id
{},
id
);
it
=
workers_by_id_
.
fin
d
(
id
);
worker
=
find_by_i
d
(
id
);
}
auto
worker
=
it
->
second
;
return
worker
->
handle_data
(
parent
,
data
);
}
template
<
class
Parent
>
void
write_message
(
Parent
&
parent
,
std
::
unique_ptr
<
endpoint_manager_queue
::
message
>
msg
)
{
auto
sender
=
msg
->
msg
->
send
er
;
if
(
!
send
er
)
auto
receiver
=
msg
->
receiv
er
;
if
(
!
receiv
er
)
return
;
auto
nid
=
send
er
->
node
();
auto
it
=
workers_by_node_
.
find
(
nid
);
if
(
it
==
workers_by_node_
.
end
()
)
{
auto
nid
=
receiv
er
->
node
();
auto
worker
=
find_by_node
(
nid
);
if
(
!
worker
)
{
// TODO: where to get id_type from here?
add_new_worker
(
parent
,
nid
,
id_type
{});
it
=
workers_by_node_
.
find
(
nid
);
worker
=
find_by_node
(
nid
);
}
auto
worker
=
it
->
second
;
worker
->
write_message
(
parent
,
std
::
move
(
msg
));
}
template
<
class
Parent
>
void
resolve
(
Parent
&
parent
,
const
std
::
string
&
path
,
actor
listener
)
{
// TODO path should be uri to lookup the corresponding worker
// if enpoint is known -> resolve actor through worker
// if not connect to endpoint?!
if
(
workers_by_id_
.
empty
())
return
;
auto
worker
=
workers_by_id_
.
begin
()
->
second
;
worker
->
resolve
(
parent
,
path
,
listener
);
void
resolve
(
Parent
&
parent
,
const
uri
&
locator
,
const
actor
&
listener
)
{
if
(
auto
worker
=
find_by_node
(
make_node_id
(
locator
)))
worker
->
resolve
(
parent
,
locator
.
path
(),
listener
);
}
template
<
class
Parent
>
void
new_proxy
(
Parent
&
,
const
node_id
&
nid
,
actor_id
id
)
{
if
(
auto
worker
=
find_by_node
(
nid
))
worker
->
new_proxy
(
*
this
,
nid
,
id
);
}
template
<
class
Parent
>
void
local_actor_down
(
Parent
&
,
const
node_id
&
nid
,
actor_id
id
,
error
reason
)
{
if
(
auto
worker
=
find_by_node
(
nid
))
worker
->
local_actor_down
(
*
this
,
nid
,
id
,
std
::
move
(
reason
));
}
template
<
class
...
Ts
>
...
...
@@ -134,6 +143,28 @@ public:
}
private:
worker_ptr
find_by_node
(
const
node_id
&
nid
)
{
if
(
workers_by_node_
.
empty
())
return
nullptr
;
auto
it
=
workers_by_node_
.
find
(
nid
);
if
(
it
==
workers_by_node_
.
end
())
{
CAF_LOG_ERROR
(
"could not find worker by node: "
<<
CAF_ARG
(
nid
));
return
nullptr
;
}
return
it
->
second
;
}
worker_ptr
find_by_id
(
const
IdType
&
id
)
{
if
(
workers_by_id_
.
empty
())
return
nullptr
;
auto
it
=
workers_by_id_
.
find
(
id
);
if
(
it
==
workers_by_id_
.
end
())
{
CAF_LOG_ERROR
(
"could not find worker by node: "
<<
CAF_ARG
(
id
));
return
nullptr
;
}
return
it
->
second
;
}
// -- worker lookups ---------------------------------------------------------
std
::
unordered_map
<
id_type
,
worker_ptr
>
workers_by_id_
;
...
...
libcaf_net/caf/net/udp_datagram_socket.hpp
View file @
98b77d32
...
...
@@ -56,9 +56,24 @@ error allow_connreset(udp_datagram_socket x, bool new_value);
variant
<
std
::
pair
<
size_t
,
ip_endpoint
>
,
sec
>
read
(
udp_datagram_socket
x
,
span
<
byte
>
buf
);
/// Sends the content of `bufs` as a datagram to the endpoint `ep` on socket
/// `x`.
/// @param x x The UDP socket for sending datagrams.
/// @param bufs Points to the datagram to send, scattered across up to 10
/// buffers.
/// @param ep The enpoint to send the datagram to.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates udp_datagram_socket
/// @post either the result is a `sec` or a positive (non-zero) integer
/// @pre `bufs.size() < 10`
variant
<
size_t
,
sec
>
write
(
udp_datagram_socket
x
,
std
::
initializer_list
<
span
<
const
byte
>>
bufs
,
ip_endpoint
ep
);
/// Sends the content of `buf` as a datagram to the endpoint `ep` on socket `x`.
/// @param x The UDP socket for sending datagrams.
/// @param buf The buffer to send.
/// @param ep The enpoint to send the datagram to.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates udp_datagram_socket
variant
<
size_t
,
sec
>
write
(
udp_datagram_socket
x
,
span
<
const
byte
>
buf
,
...
...
libcaf_net/test/datagram_transport.cpp
View file @
98b77d32
...
...
@@ -20,10 +20,9 @@
#include "caf/net/datagram_transport.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/byte.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
...
...
@@ -42,6 +41,8 @@ namespace {
constexpr
string_view
hello_manager
=
"hello manager!"
;
class
dummy_application_factory
;
struct
fixture
:
test_coordinator_fixture
<>
,
host_fixture
{
fixture
()
{
mpx
=
std
::
make_shared
<
multiplexer
>
();
...
...
@@ -50,7 +51,6 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
}
bool
handle_io_event
()
override
{
mpx
->
handle_updates
();
return
mpx
->
poll_once
(
false
);
}
...
...
@@ -59,6 +59,8 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
class
dummy_application
{
public:
using
transport_type
=
datagram_transport
<
dummy_application_factory
>
;
dummy_application
(
std
::
shared_ptr
<
std
::
vector
<
byte
>>
rec_buf
)
:
rec_buf_
(
std
::
move
(
rec_buf
)){
// nop
...
...
@@ -73,18 +75,19 @@ public:
template
<
class
Transport
>
void
write_message
(
Transport
&
transport
,
std
::
unique_ptr
<
endpoint_manager
::
message
>
msg
)
{
std
::
unique_ptr
<
endpoint_manager
_queue
::
message
>
msg
)
{
transport
.
write_packet
(
span
<
byte
>
{},
msg
->
payload
);
}
template
<
class
Parent
>
void
handle_data
(
Parent
&
,
span
<
const
byte
>
data
)
{
error
handle_data
(
Parent
&
,
span
<
const
byte
>
data
)
{
rec_buf_
->
clear
();
rec_buf_
->
insert
(
rec_buf_
->
begin
(),
data
.
begin
(),
data
.
end
());
return
none
;
}
template
<
class
Parent
>
void
resolve
(
Parent
&
parent
,
const
std
::
string
&
path
,
actor
listener
)
{
void
resolve
(
Parent
&
parent
,
string_view
path
,
const
actor
&
listener
)
{
actor_id
aid
=
42
;
auto
hid
=
"0011223344556677889900112233445566778899"
;
auto
nid
=
unbox
(
make_node_id
(
42
,
hid
));
...
...
@@ -94,7 +97,17 @@ public:
&
parent
.
system
(),
cfg
,
std
::
move
(
ptr
));
anon_send
(
listener
,
resolve_atom
::
value
,
std
::
move
(
path
),
p
);
anon_send
(
listener
,
resolve_atom
::
value
,
path
,
p
);
}
template
<
class
Parent
>
void
new_proxy
(
Parent
&
,
actor_id
)
{
// nop
}
template
<
class
Parent
>
void
local_actor_down
(
Parent
&
,
actor_id
,
error
)
{
// nop
}
template
<
class
Transport
>
...
...
@@ -119,11 +132,12 @@ private:
std
::
shared_ptr
<
std
::
vector
<
byte
>>
rec_buf_
;
};
struct
dummy_application_factory
{
class
dummy_application_factory
{
public:
using
application_type
=
dummy_application
;
dummy_application_factory
(
std
::
shared_ptr
<
std
::
vector
<
byte
>>
buf
)
:
buf_
(
buf
)
{
explicit
dummy_application_factory
(
std
::
shared_ptr
<
std
::
vector
<
byte
>>
buf
)
:
buf_
(
std
::
move
(
buf
)
)
{
// nop
}
...
...
@@ -159,7 +173,6 @@ CAF_TEST(receive) {
transport
.
configure_read
(
net
::
receive_policy
::
exactly
(
hello_manager
.
size
()));
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
transport
);
CAF_CHECK_EQUAL
(
mgr
->
init
(),
none
);
mpx
->
handle_updates
();
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
2u
);
CAF_CHECK_EQUAL
(
write
(
sender
,
as_bytes
(
make_span
(
hello_manager
)),
ep
),
hello_manager
.
size
());
...
...
@@ -172,7 +185,6 @@ CAF_TEST(receive) {
// TODO: test is disabled until resolve in transport_worker_dispatcher is
// implemented correctly.
// Idea is to use caf::uri instead of std::string.
/*
CAF_TEST(resolve and proxy communication) {
using transport_type = datagram_transport<dummy_application_factory>;
...
...
libcaf_net/test/transport_worker_dispatcher.cpp
View file @
98b77d32
...
...
@@ -180,8 +180,8 @@ uri operator"" _u(const char* cstr, size_t cstr_len) {
}
struct
fixture
:
host_fixture
{
using
dispatcher_type
=
transport_worker_dispatcher
<
dummy_application_factory
,
ip_endpoint
>
;
using
dispatcher_type
=
transport_worker_dispatcher
<
dummy_transport
,
dummy_application_factory
,
ip_endpoint
>
;
fixture
()
:
buf
{
std
::
make_shared
<
buffer_type
>
()},
...
...
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