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
9e81e4f5
Commit
9e81e4f5
authored
Nov 13, 2019
by
Jakob Otto
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Add transport_base to datagram_transport
parent
af9be3ef
Changes
6
Hide whitespace changes
Inline
Side-by-side
Showing
6 changed files
with
160 additions
and
243 deletions
+160
-243
libcaf_net/caf/net/datagram_transport.hpp
libcaf_net/caf/net/datagram_transport.hpp
+46
-136
libcaf_net/caf/net/fwd.hpp
libcaf_net/caf/net/fwd.hpp
+8
-3
libcaf_net/caf/net/stream_transport.hpp
libcaf_net/caf/net/stream_transport.hpp
+80
-38
libcaf_net/caf/net/transport_base.hpp
libcaf_net/caf/net/transport_base.hpp
+17
-52
libcaf_net/caf/net/transport_worker_dispatcher.hpp
libcaf_net/caf/net/transport_worker_dispatcher.hpp
+6
-11
libcaf_net/test/transport_worker_dispatcher.cpp
libcaf_net/test/transport_worker_dispatcher.cpp
+3
-3
No files found.
libcaf_net/caf/net/datagram_transport.hpp
View file @
9e81e4f5
...
@@ -24,24 +24,29 @@
...
@@ -24,24 +24,29 @@
#include "caf/byte.hpp"
#include "caf/byte.hpp"
#include "caf/error.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/fwd.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/logger.hpp"
#include "caf/logger.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/transport_base.hpp"
#include "caf/net/transport_worker_dispatcher.hpp"
#include "caf/net/transport_worker_dispatcher.hpp"
#include "caf/net/udp_datagram_socket.hpp"
#include "caf/net/udp_datagram_socket.hpp"
#include "caf/sec.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp"
#include "caf/variant.hpp"
namespace
caf
{
namespace
caf
::
net
{
namespace
net
{
template
<
class
Factory
>
using
datagram_transport_base
=
transport_base
<
datagram_transport
<
Factory
>
,
transport_worker_dispatcher
<
Factory
,
ip_endpoint
>
,
udp_datagram_socket
,
Factory
,
ip_endpoint
>
;
/// Implements a udp_transport policy that manages a datagram socket.
/// Implements a udp_transport policy that manages a datagram socket.
template
<
class
Factory
>
template
<
class
Factory
>
class
datagram_transport
{
class
datagram_transport
:
public
datagram_transport_base
<
Factory
>
{
public:
public:
// Maximal UDP-packet size
// Maximal UDP-packet size
static
constexpr
size_t
max_datagram_size
=
std
::
numeric_limits
<
static
constexpr
size_t
max_datagram_size
=
std
::
numeric_limits
<
...
@@ -49,163 +54,83 @@ public:
...
@@ -49,163 +54,83 @@ public:
// -- member types -----------------------------------------------------------
// -- member types -----------------------------------------------------------
using
id_type
=
ip_endpoint
;
using
factory_type
=
Factory
;
using
buffer_type
=
std
::
vector
<
byte
>
;
using
id_type
=
ip_endpoint
;
using
buffer_cache_type
=
std
::
vector
<
buffer_type
>
;
using
application_type
=
typename
factory_type
::
application_type
;
using
factory_type
=
Factory
;
using
super
=
datagram_transport_base
<
factory_type
>
;
using
transport_type
=
datagram_transport
<
Factory
>
;
using
buffer_type
=
typename
super
::
buffer_type
;
using
dispatcher_type
=
transport_worker_dispatcher
<
transport_type
,
id_type
>
;
using
buffer_cache_type
=
typename
super
::
buffer_cache_type
;
using
application_type
=
typename
Factory
::
application_type
;
// -- constructors, destructors, and assignment operators --------------------
// -- constructors, destructors, and assignment operators --------------------
datagram_transport
(
udp_datagram_socket
handle
,
factory_type
factory
)
datagram_transport
(
udp_datagram_socket
handle
,
factory_type
factory
)
:
dispatcher_
(
*
this
,
std
::
move
(
factory
)),
:
super
(
handle
,
std
::
move
(
factory
))
{
handle_
(
handle
),
read_buf_
(
max_datagram_size
),
manager_
(
nullptr
)
{
// nop
// nop
}
}
// -- properties -------------------------------------------------------------
udp_datagram_socket
handle
()
const
noexcept
{
return
handle_
;
}
actor_system
&
system
()
{
return
manager
().
system
();
}
transport_type
&
transport
()
{
return
*
this
;
}
endpoint_manager
&
manager
()
{
return
*
manager_
;
}
// -- public member functions ------------------------------------------------
// -- public member functions ------------------------------------------------
template
<
class
Parent
>
error
init
(
endpoint_manager
&
manager
)
override
{
error
init
(
Parent
&
parent
)
{
if
(
auto
err
=
super
::
init
(
manager
))
manager_
=
&
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
(
*
this
))
return
err
;
return
err
;
prepare_next_read
();
return
none
;
return
none
;
}
}
template
<
class
Parent
>
bool
handle_read_event
(
endpoint_manager
&
)
override
{
bool
handle_read_event
(
Parent
&
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
this
->
handle_
.
id
));
CAF_LOG_TRACE
(
CAF_ARG
(
handle_
.
id
));
auto
ret
=
read
(
this
->
handle_
,
make_span
(
this
->
read_buf_
));
auto
ret
=
read
(
handle_
,
make_span
(
read_buf_
));
if
(
auto
res
=
get_if
<
std
::
pair
<
size_t
,
ip_endpoint
>>
(
&
ret
))
{
if
(
auto
res
=
get_if
<
std
::
pair
<
size_t
,
ip_endpoint
>>
(
&
ret
))
{
auto
num_bytes
=
res
->
first
;
auto
num_bytes
=
res
->
first
;
auto
ep
=
res
->
second
;
auto
ep
=
res
->
second
;
read_buf_
.
resize
(
num_bytes
);
this
->
read_buf_
.
resize
(
num_bytes
);
dispatcher_
.
handle_data
(
*
this
,
make_span
(
read_buf_
),
std
::
move
(
ep
));
this
->
next_layer_
.
handle_data
(
*
this
,
make_span
(
this
->
read_buf_
),
std
::
move
(
ep
));
prepare_next_read
();
prepare_next_read
();
}
else
{
}
else
{
auto
err
=
get
<
sec
>
(
ret
);
auto
err
=
get
<
sec
>
(
ret
);
CAF_LOG_DEBUG
(
"send failed"
<<
CAF_ARG
(
err
));
CAF_LOG_DEBUG
(
"send failed"
<<
CAF_ARG
(
err
));
dispatch
er_
.
handle_error
(
err
);
this
->
next_lay
er_
.
handle_error
(
err
);
return
false
;
return
false
;
}
}
return
true
;
return
true
;
}
}
template
<
class
Parent
>
bool
handle_write_event
(
endpoint_manager
&
manager
)
override
{
bool
handle_write_event
(
Parent
&
parent
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
this
->
handle_
.
id
)
CAF_LOG_TRACE
(
CAF_ARG
(
handle_
.
id
)
<<
CAF_ARG2
(
"queue-size"
,
packet_queue_
.
size
()));
<<
CAF_ARG2
(
"queue-size"
,
packet_queue_
.
size
()));
// Try to write leftover data.
// Try to write leftover data.
write_some
();
write_some
();
// Get new data from parent.
// Get new data from parent.
for
(
auto
msg
=
parent
.
next_message
();
msg
!=
nullptr
;
for
(
auto
msg
=
manager
.
next_message
();
msg
!=
nullptr
;
msg
=
parent
.
next_message
())
{
msg
=
manager
.
next_message
())
{
dispatch
er_
.
write_message
(
*
this
,
std
::
move
(
msg
));
this
->
next_lay
er_
.
write_message
(
*
this
,
std
::
move
(
msg
));
}
}
// Write prepared data.
// Write prepared data.
return
write_some
();
return
write_some
();
}
}
template
<
class
Parent
>
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
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
>
void
timeout
(
Parent
&
,
atom_value
value
,
uint64_t
id
)
{
dispatcher_
.
timeout
(
*
this
,
value
,
id
);
}
void
set_timeout
(
uint64_t
timeout_id
,
id_type
id
)
{
dispatcher_
.
set_timeout
(
timeout_id
,
id
);
}
void
handle_error
(
sec
code
)
{
dispatcher_
.
handle_error
(
code
);
}
error
add_new_worker
(
node_id
node
,
id_type
id
)
{
error
add_new_worker
(
node_id
node
,
id_type
id
)
{
auto
worker
=
dispatch
er_
.
add_new_worker
(
*
this
,
node
,
id
);
auto
worker
=
this
->
next_lay
er_
.
add_new_worker
(
*
this
,
node
,
id
);
if
(
!
worker
)
if
(
!
worker
)
return
worker
.
error
();
return
worker
.
error
();
return
none
;
return
none
;
}
}
void
prepare_next_read
()
{
void
write_packet
(
id_type
id
,
span
<
buffer_type
*>
buffers
)
override
{
read_buf_
.
clear
();
read_buf_
.
resize
(
max_datagram_size
);
}
void
configure_read
(
receive_policy
::
config
)
{
// nop
}
void
write_packet
(
id_type
id
,
span
<
buffer_type
*>
buffers
)
{
CAF_ASSERT
(
!
buffers
.
empty
());
CAF_ASSERT
(
!
buffers
.
empty
());
if
(
packet_queue_
.
empty
())
if
(
packet_queue_
.
empty
())
manager
().
register_writing
();
this
->
manager
().
register_writing
();
// By convention, the first buffer is a header buffer. Every other buffer is
// By convention, the first buffer is a header buffer. Every other buffer is
// a payload buffer.
// a payload buffer.
packet_queue_
.
emplace_back
(
id
,
buffers
);
packet_queue_
.
emplace_back
(
id
,
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
/// Helper struct for managing outgoing packets
struct
packet
{
struct
packet
{
id_type
id
;
id_type
id
;
...
@@ -224,31 +149,26 @@ public:
...
@@ -224,31 +149,26 @@ public:
private:
private:
// -- utility functions ------------------------------------------------------
// -- utility functions ------------------------------------------------------
static
buffer_type
next_buffer_impl
(
buffer_cache_type
cache
)
{
void
prepare_next_read
()
{
if
(
cache
.
empty
())
{
this
->
read_buf_
.
resize
(
max_datagram_size
);
return
{};
}
auto
buf
=
std
::
move
(
cache
.
back
());
cache
.
pop_back
();
return
buf
;
}
}
bool
write_some
()
{
bool
write_some
()
{
CAF_LOG_TRACE
(
CAF_ARG
(
handle_
.
id
));
CAF_LOG_TRACE
(
CAF_ARG
(
this
->
handle_
.
id
));
// Helper function to sort empty buffers back into the right caches.
// Helper function to sort empty buffers back into the right caches.
auto
recycle
=
[
&
]()
{
auto
recycle
=
[
&
]()
{
auto
&
front
=
packet_queue_
.
front
();
auto
&
front
=
packet_queue_
.
front
();
auto
&
bufs
=
front
.
bytes
;
auto
&
bufs
=
front
.
bytes
;
auto
it
=
bufs
.
begin
();
auto
it
=
bufs
.
begin
();
if
(
header_bufs_
.
size
()
<
header_bufs_
.
capacity
())
{
if
(
this
->
header_bufs_
.
size
()
<
this
->
header_bufs_
.
capacity
())
{
it
->
clear
();
it
->
clear
();
header_bufs_
.
emplace_back
(
std
::
move
(
*
it
++
));
this
->
header_bufs_
.
emplace_back
(
std
::
move
(
*
it
++
));
}
}
for
(;
for
(;
it
!=
bufs
.
end
()
it
!=
bufs
.
end
()
&&
payload_bufs_
.
size
()
<
payload_bufs_
.
capacity
();
&&
this
->
payload_bufs_
.
size
()
<
this
->
payload_bufs_
.
capacity
();
++
it
)
{
++
it
)
{
it
->
clear
();
it
->
clear
();
payload_bufs_
.
emplace_back
(
std
::
move
(
*
it
));
this
->
payload_bufs_
.
emplace_back
(
std
::
move
(
*
it
));
}
}
packet_queue_
.
pop_front
();
packet_queue_
.
pop_front
();
};
};
...
@@ -258,9 +178,9 @@ private:
...
@@ -258,9 +178,9 @@ private:
std
::
vector
<
std
::
vector
<
byte
>*>
ptrs
;
std
::
vector
<
std
::
vector
<
byte
>*>
ptrs
;
for
(
auto
&
buf
:
packet
.
bytes
)
for
(
auto
&
buf
:
packet
.
bytes
)
ptrs
.
emplace_back
(
&
buf
);
ptrs
.
emplace_back
(
&
buf
);
auto
write_ret
=
write
(
handle_
,
make_span
(
ptrs
),
packet
.
id
);
auto
write_ret
=
write
(
this
->
handle_
,
make_span
(
ptrs
),
packet
.
id
);
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
write_ret
))
{
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
write_ret
))
{
CAF_LOG_DEBUG
(
CAF_ARG
(
handle_
.
id
)
<<
CAF_ARG
(
*
num_bytes
));
CAF_LOG_DEBUG
(
CAF_ARG
(
this
->
handle_
.
id
)
<<
CAF_ARG
(
*
num_bytes
));
CAF_LOG_WARNING_IF
(
*
num_bytes
<
packet
.
size
,
CAF_LOG_WARNING_IF
(
*
num_bytes
<
packet
.
size
,
"packet was not sent completely"
);
"packet was not sent completely"
);
recycle
();
recycle
();
...
@@ -268,7 +188,7 @@ private:
...
@@ -268,7 +188,7 @@ private:
auto
err
=
get
<
sec
>
(
write_ret
);
auto
err
=
get
<
sec
>
(
write_ret
);
if
(
err
!=
sec
::
unavailable_or_would_block
)
{
if
(
err
!=
sec
::
unavailable_or_would_block
)
{
CAF_LOG_DEBUG
(
"send failed"
<<
CAF_ARG
(
err
));
CAF_LOG_DEBUG
(
"send failed"
<<
CAF_ARG
(
err
));
dispatch
er_
.
handle_error
(
err
);
this
->
next_lay
er_
.
handle_error
(
err
);
return
false
;
return
false
;
}
}
return
true
;
return
true
;
...
@@ -277,17 +197,7 @@ private:
...
@@ -277,17 +197,7 @@ private:
return
false
;
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_
;
std
::
deque
<
packet
>
packet_queue_
;
endpoint_manager
*
manager_
;
};
};
}
// namespace net
}
// namespace caf::net
}
// namespace caf
libcaf_net/caf/net/fwd.hpp
View file @
9e81e4f5
...
@@ -26,15 +26,18 @@ namespace caf::net {
...
@@ -26,15 +26,18 @@ namespace caf::net {
// -- templates ----------------------------------------------------------------
// -- templates ----------------------------------------------------------------
template
<
class
Application
>
class
stream_transport
;
template
<
class
Factory
>
class
datagram_transport
;
template
<
class
Application
,
class
IdType
=
unit_t
>
template
<
class
Application
,
class
IdType
=
unit_t
>
class
transport_worker
;
class
transport_worker
;
template
<
class
Transport
,
class
IdType
=
unit_t
>
template
<
class
Transport
,
class
IdType
=
unit_t
>
class
transport_worker_dispatcher
;
class
transport_worker_dispatcher
;
template
<
class
Application
>
class
stream_transport
;
// -- classes ------------------------------------------------------------------
// -- classes ------------------------------------------------------------------
class
endpoint_manager
;
class
endpoint_manager
;
...
@@ -51,6 +54,8 @@ struct socket;
...
@@ -51,6 +54,8 @@ struct socket;
struct
stream_socket
;
struct
stream_socket
;
struct
tcp_accept_socket
;
struct
tcp_accept_socket
;
struct
tcp_stream_socket
;
struct
tcp_stream_socket
;
struct
datagram_socket
;
struct
udp_datagram_socket
;
// -- smart pointers -----------------------------------------------------------
// -- smart pointers -----------------------------------------------------------
...
...
libcaf_net/caf/net/stream_transport.hpp
View file @
9e81e4f5
...
@@ -38,13 +38,13 @@
...
@@ -38,13 +38,13 @@
namespace
caf
::
net
{
namespace
caf
::
net
{
template
<
class
Application
>
template
<
class
Application
>
using
transport_base_type
=
transport_base
<
stream_transport
<
Application
>
,
using
stream_transport_base
=
transport_base
<
transport_worker
<
Application
>
,
stream_transport
<
Application
>
,
transport_worker
<
Application
>
,
stream_socket
,
stream_socket
,
Application
,
unit_t
>
;
Application
,
unit_t
>
;
/// Implements a stream_transport that manages a stream socket.
/// Implements a stream_transport that manages a stream socket.
template
<
class
Application
>
template
<
class
Application
>
class
stream_transport
:
public
transport_base_typ
e
<
Application
>
{
class
stream_transport
:
public
stream_transport_bas
e
<
Application
>
{
public:
public:
// -- member types -----------------------------------------------------------
// -- member types -----------------------------------------------------------
...
@@ -58,12 +58,19 @@ public:
...
@@ -58,12 +58,19 @@ public:
using
buffer_cache_type
=
std
::
vector
<
buffer_type
>
;
using
buffer_cache_type
=
std
::
vector
<
buffer_type
>
;
using
super
=
transport_base_type
<
application_type
>
;
using
id_type
=
unit_t
;
using
super
=
stream_transport_base
<
application_type
>
;
// -- constructors, destructors, and assignment operators --------------------
// -- constructors, destructors, and assignment operators --------------------
stream_transport
(
stream_socket
handle
,
application_type
application
)
stream_transport
(
stream_socket
handle
,
application_type
application
)
:
super
(
handle
,
std
::
move
(
application
)),
written_
(
0
)
{
:
super
(
handle
,
std
::
move
(
application
)),
written_
(
0
),
read_threshold_
(
1024
),
collected_
(
0
),
max_
(
1024
),
rd_flag_
(
net
::
receive_policy_flag
::
exactly
)
{
// nop
// nop
}
}
...
@@ -75,30 +82,27 @@ public:
...
@@ -75,30 +82,27 @@ public:
}
}
bool
handle_read_event
(
endpoint_manager
&
)
override
{
bool
handle_read_event
(
endpoint_manager
&
)
override
{
auto
buf
=
super
::
read_buf_
.
data
()
+
super
::
collected_
;
auto
buf
=
this
->
read_buf_
.
data
()
+
this
->
collected_
;
size_t
len
=
super
::
read_threshold_
-
super
::
collected_
;
size_t
len
=
this
->
read_threshold_
-
this
->
collected_
;
CAF_LOG_TRACE
(
CAF_ARG
(
super
::
handle
().
id
)
<<
CAF_ARG
(
len
));
CAF_LOG_TRACE
(
CAF_ARG
(
this
->
handle
().
id
)
<<
CAF_ARG
(
len
));
auto
ret
=
read
(
super
::
handle_
,
make_span
(
buf
,
len
));
auto
ret
=
read
(
this
->
handle_
,
make_span
(
buf
,
len
));
// Update state.
// Update state.
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
ret
))
{
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
ret
))
{
CAF_LOG_DEBUG
(
CAF_ARG
(
len
)
CAF_LOG_DEBUG
(
CAF_ARG
(
len
)
<<
CAF_ARG
(
super
::
handle_
.
id
)
<<
CAF_ARG
(
*
num_bytes
));
<<
CAF_ARG
(
this
->
handle_
.
id
)
<<
CAF_ARG
(
*
num_bytes
));
super
::
collected_
+=
*
num_bytes
;
this
->
collected_
+=
*
num_bytes
;
if
(
super
::
collected_
>=
super
::
read_threshold_
)
{
if
(
this
->
collected_
>=
this
->
read_threshold_
)
{
if
(
auto
err
=
super
::
next_layer_
.
handle_data
(
*
this
,
if
(
auto
err
=
this
->
next_layer_
.
handle_data
(
*
this
,
this
->
read_buf_
))
{
super
::
read_buf_
))
{
CAF_LOG_WARNING
(
"handle_data failed:"
<<
CAF_ARG
(
err
));
CAF_LOG_WARNING
(
"handle_data failed:"
<<
CAF_ARG
(
err
));
return
false
;
return
false
;
}
}
// TODO: Is this the proper way or would
this
->
prepare_next_read
();
// `transport_base_type<Application>::prepare_next_read()` be better?
super
::
prepare_next_read
();
}
}
}
else
{
}
else
{
auto
err
=
get
<
sec
>
(
ret
);
auto
err
=
get
<
sec
>
(
ret
);
if
(
err
!=
sec
::
unavailable_or_would_block
)
{
if
(
err
!=
sec
::
unavailable_or_would_block
)
{
CAF_LOG_DEBUG
(
"receive failed"
<<
CAF_ARG
(
err
));
CAF_LOG_DEBUG
(
"receive failed"
<<
CAF_ARG
(
err
));
super
::
next_layer_
.
handle_error
(
err
);
this
->
next_layer_
.
handle_error
(
err
);
return
false
;
return
false
;
}
}
}
}
...
@@ -112,54 +116,86 @@ public:
...
@@ -112,54 +116,86 @@ public:
// TODO: dont read all messages at once - get one by one.
// TODO: dont read all messages at once - get one by one.
for
(
auto
msg
=
parent
.
next_message
();
msg
!=
nullptr
;
for
(
auto
msg
=
parent
.
next_message
();
msg
!=
nullptr
;
msg
=
parent
.
next_message
())
{
msg
=
parent
.
next_message
())
{
super
::
next_layer_
.
write_message
(
*
this
,
std
::
move
(
msg
));
this
->
next_layer_
.
write_message
(
*
this
,
std
::
move
(
msg
));
}
}
// Write prepared data.
// Write prepared data.
return
write_some
();
return
write_some
();
}
}
void
write_packet
(
unit_t
,
span
<
buffer_type
*>
buffers
)
override
{
void
write_packet
(
id_type
,
span
<
buffer_type
*>
buffers
)
override
{
CAF_ASSERT
(
!
buffers
.
empty
());
CAF_ASSERT
(
!
buffers
.
empty
());
if
(
super
::
write_queue_
.
empty
())
if
(
this
->
write_queue_
.
empty
())
super
::
manager
().
register_writing
();
this
->
manager
().
register_writing
();
// By convention, the first buffer is a header buffer. Every other buffer is
// By convention, the first buffer is a header buffer. Every other buffer is
// a payload buffer.
// a payload buffer.
auto
i
=
buffers
.
begin
();
auto
i
=
buffers
.
begin
();
super
::
write_queue_
.
emplace_back
(
true
,
std
::
move
(
*
(
*
i
++
)));
this
->
write_queue_
.
emplace_back
(
true
,
std
::
move
(
*
(
*
i
++
)));
while
(
i
!=
buffers
.
end
())
while
(
i
!=
buffers
.
end
())
super
::
write_queue_
.
emplace_back
(
false
,
std
::
move
(
*
(
*
i
++
)));
this
->
write_queue_
.
emplace_back
(
false
,
std
::
move
(
*
(
*
i
++
)));
}
void
configure_read
(
receive_policy
::
config
cfg
)
override
{
rd_flag_
=
cfg
.
first
;
max_
=
cfg
.
second
;
prepare_next_read
();
}
}
private:
private:
// -- utility functions ------------------------------------------------------
// -- utility functions ------------------------------------------------------
void
prepare_next_read
()
{
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.
switch
(
static_cast
<
net
::
receive_policy_flag
>
(
rd_flag_
))
{
case
net
:
:
receive_policy_flag
::
exactly
:
if
(
this
->
read_buf_
.
size
()
!=
max_
)
this
->
read_buf_
.
resize
(
max_
);
read_threshold_
=
max_
;
break
;
case
net
:
:
receive_policy_flag
::
at_most
:
if
(
this
->
read_buf_
.
size
()
!=
max_
)
this
->
read_buf_
.
resize
(
max_
);
read_threshold_
=
1
;
break
;
case
net
:
:
receive_policy_flag
::
at_least
:
{
// read up to 10% more, but at least allow 100 bytes more
auto
max_size
=
max_
+
std
::
max
<
size_t
>
(
100
,
max_
/
10
);
if
(
this
->
read_buf_
.
size
()
!=
max_size
)
this
->
read_buf_
.
resize
(
max_size
);
read_threshold_
=
max_
;
break
;
}
}
}
bool
write_some
()
{
bool
write_some
()
{
CAF_LOG_TRACE
(
CAF_ARG
(
super
::
handle_
.
id
));
CAF_LOG_TRACE
(
CAF_ARG
(
this
->
handle_
.
id
));
// Helper function to sort empty buffers back into the right caches.
// Helper function to sort empty buffers back into the right caches.
auto
recycle
=
[
&
]()
{
auto
recycle
=
[
&
]()
{
auto
&
front
=
super
::
write_queue_
.
front
();
auto
&
front
=
this
->
write_queue_
.
front
();
auto
&
is_header
=
front
.
first
;
auto
&
is_header
=
front
.
first
;
auto
&
buf
=
front
.
second
;
auto
&
buf
=
front
.
second
;
written_
=
0
;
written_
=
0
;
buf
.
clear
();
buf
.
clear
();
if
(
is_header
)
{
if
(
is_header
)
{
if
(
super
::
header_bufs_
.
size
()
<
super
::
header_bufs_
.
capacity
())
if
(
this
->
header_bufs_
.
size
()
<
this
->
header_bufs_
.
capacity
())
super
::
header_bufs_
.
emplace_back
(
std
::
move
(
buf
));
this
->
header_bufs_
.
emplace_back
(
std
::
move
(
buf
));
}
else
if
(
super
::
payload_bufs_
.
size
()
}
else
if
(
this
->
payload_bufs_
.
size
()
<
this
->
payload_bufs_
.
capacity
())
{
<
super
::
payload_bufs_
.
capacity
())
{
this
->
payload_bufs_
.
emplace_back
(
std
::
move
(
buf
));
super
::
payload_bufs_
.
emplace_back
(
std
::
move
(
buf
));
}
}
super
::
write_queue_
.
pop_front
();
write_queue_
.
pop_front
();
};
};
// Write buffers from the write_queue_ for as long as possible.
// Write buffers from the write_queue_ for as long as possible.
while
(
!
super
::
write_queue_
.
empty
())
{
while
(
!
write_queue_
.
empty
())
{
auto
&
buf
=
super
::
write_queue_
.
front
().
second
;
auto
&
buf
=
write_queue_
.
front
().
second
;
CAF_ASSERT
(
!
buf
.
empty
());
CAF_ASSERT
(
!
buf
.
empty
());
auto
data
=
buf
.
data
()
+
written_
;
auto
data
=
buf
.
data
()
+
written_
;
auto
len
=
buf
.
size
()
-
written_
;
auto
len
=
buf
.
size
()
-
written_
;
auto
write_ret
=
write
(
super
::
handle
(),
make_span
(
data
,
len
));
auto
write_ret
=
write
(
this
->
handle
(),
make_span
(
data
,
len
));
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
write_ret
))
{
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
write_ret
))
{
CAF_LOG_DEBUG
(
CAF_ARG
(
super
::
handle_
.
id
)
<<
CAF_ARG
(
*
num_bytes
));
CAF_LOG_DEBUG
(
CAF_ARG
(
this
->
handle_
.
id
)
<<
CAF_ARG
(
*
num_bytes
));
if
(
*
num_bytes
+
written_
>=
buf
.
size
())
{
if
(
*
num_bytes
+
written_
>=
buf
.
size
())
{
recycle
();
recycle
();
written_
=
0
;
written_
=
0
;
...
@@ -171,7 +207,7 @@ private:
...
@@ -171,7 +207,7 @@ private:
auto
err
=
get
<
sec
>
(
write_ret
);
auto
err
=
get
<
sec
>
(
write_ret
);
if
(
err
!=
sec
::
unavailable_or_would_block
)
{
if
(
err
!=
sec
::
unavailable_or_would_block
)
{
CAF_LOG_DEBUG
(
"send failed"
<<
CAF_ARG
(
err
));
CAF_LOG_DEBUG
(
"send failed"
<<
CAF_ARG
(
err
));
super
::
next_layer_
.
handle_error
(
err
);
this
->
next_layer_
.
handle_error
(
err
);
return
false
;
return
false
;
}
}
return
true
;
return
true
;
...
@@ -180,7 +216,13 @@ private:
...
@@ -180,7 +216,13 @@ private:
return
false
;
return
false
;
}
}
std
::
deque
<
std
::
pair
<
bool
,
buffer_type
>>
write_queue_
;
size_t
written_
;
size_t
written_
;
size_t
read_threshold_
;
size_t
collected_
;
size_t
max_
;
receive_policy_flag
rd_flag_
;
};
};
}
// namespace caf::net
}
// namespace caf::net
libcaf_net/caf/net/transport_base.hpp
View file @
9e81e4f5
...
@@ -20,6 +20,7 @@
...
@@ -20,6 +20,7 @@
#include "caf/actor_system_config.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/byte.hpp"
#include "caf/byte.hpp"
#include "caf/detail/overload.hpp"
#include "caf/error.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/expected.hpp"
#include "caf/fwd.hpp"
#include "caf/fwd.hpp"
...
@@ -57,15 +58,10 @@ public:
...
@@ -57,15 +58,10 @@ public:
// -- constructors, destructors, and assignment operators --------------------
// -- constructors, destructors, and assignment operators --------------------
transport_base
(
Handl
e
handle
,
application_type
application
)
transport_base
(
handle_typ
e
handle
,
application_type
application
)
:
next_layer_
(
std
::
move
(
application
)),
:
next_layer_
(
std
::
move
(
application
)),
handle_
(
handle
),
handle_
(
handle
),
// max_consecutive_reads_(0),
// max_consecutive_reads_(0),
read_threshold_
(
1024
),
collected_
(
0
),
max_
(
1024
),
rd_flag_
(
net
::
receive_policy_flag
::
exactly
),
written_
(
0
),
manager_
(
nullptr
)
{
manager_
(
nullptr
)
{
// nop
// nop
}
}
...
@@ -112,8 +108,16 @@ public:
...
@@ -112,8 +108,16 @@ public:
virtual
bool
handle_write_event
(
endpoint_manager
&
parent
)
=
0
;
virtual
bool
handle_write_event
(
endpoint_manager
&
parent
)
=
0
;
void
resolve
(
endpoint_manager
&
,
const
uri
&
locator
,
const
actor
&
listener
)
{
auto
resolve
(
endpoint_manager
&
,
const
uri
&
locator
,
const
actor
&
listener
)
{
next_layer_
.
resolve
(
*
this
,
locator
.
path
(),
listener
);
auto
f
=
detail
::
make_overload
(
[
&
](
auto
&
layer
)
->
decltype
(
layer
.
resolve
(
*
this
,
locator
,
listener
))
{
return
layer
.
resolve
(
*
this
,
locator
,
listener
);
},
[
&
](
auto
&
layer
)
->
decltype
(
layer
.
resolve
(
*
this
,
locator
.
path
(),
listener
))
{
return
layer
.
resolve
(
*
this
,
locator
.
path
(),
listener
);
});
f
(
next_layer_
);
}
}
void
new_proxy
(
endpoint_manager
&
,
const
node_id
&
peer
,
actor_id
id
)
{
void
new_proxy
(
endpoint_manager
&
,
const
node_id
&
peer
,
actor_id
id
)
{
...
@@ -129,49 +133,18 @@ public:
...
@@ -129,49 +133,18 @@ public:
next_layer_
.
timeout
(
*
this
,
value
,
id
);
next_layer_
.
timeout
(
*
this
,
value
,
id
);
}
}
template
<
class
...
Ts
>
void
set_timeout
(
uint64_t
timeout_id
,
id_type
id
)
{
void
set_timeout
(
uint64_t
,
Ts
&&
...)
{
next_layer_
.
set_timeout
(
timeout_id
,
id
);
// TODO: implement me!
}
}
void
handle_error
(
sec
code
)
{
void
handle_error
(
sec
code
)
{
next_layer_
.
handle_error
(
code
);
next_layer_
.
handle_error
(
code
);
}
}
void
prepare_next_read
()
{
virtual
void
configure_read
(
receive_policy
::
config
){
collected_
=
0
;
// nop
// This cast does nothing, but prevents a weird compiler error on GCC
};
// <= 4.9.
// TODO: remove cast when dropping support for GCC 4.9.
switch
(
static_cast
<
net
::
receive_policy_flag
>
(
rd_flag_
))
{
case
net
:
:
receive_policy_flag
::
exactly
:
if
(
read_buf_
.
size
()
!=
max_
)
read_buf_
.
resize
(
max_
);
read_threshold_
=
max_
;
break
;
case
net
:
:
receive_policy_flag
::
at_most
:
if
(
read_buf_
.
size
()
!=
max_
)
read_buf_
.
resize
(
max_
);
read_threshold_
=
1
;
break
;
case
net
:
:
receive_policy_flag
::
at_least
:
{
// read up to 10% more, but at least allow 100 bytes more
auto
max_size
=
max_
+
std
::
max
<
size_t
>
(
100
,
max_
/
10
);
if
(
read_buf_
.
size
()
!=
max_size
)
read_buf_
.
resize
(
max_size
);
read_threshold_
=
max_
;
break
;
}
}
}
void
configure_read
(
receive_policy
::
config
cfg
)
{
rd_flag_
=
cfg
.
first
;
max_
=
cfg
.
second
;
prepare_next_read
();
}
// TODO: make this work for both datagram and stream oriented transports.
virtual
void
write_packet
(
id_type
id
,
span
<
buffer_type
*>
buffers
)
=
0
;
virtual
void
write_packet
(
id_type
id
,
span
<
buffer_type
*>
buffers
)
=
0
;
// -- buffer management ------------------------------------------------------
// -- buffer management ------------------------------------------------------
...
@@ -204,16 +177,8 @@ protected:
...
@@ -204,16 +177,8 @@ protected:
buffer_cache_type
payload_bufs_
;
buffer_cache_type
payload_bufs_
;
buffer_type
read_buf_
;
buffer_type
read_buf_
;
std
::
deque
<
std
::
pair
<
bool
,
buffer_type
>>
write_queue_
;
// TODO implement retries using this member!
// TODO implement retries using this member!
// size_t max_consecutive_reads_;
// size_t max_consecutive_reads_;
size_t
read_threshold_
;
size_t
collected_
;
size_t
max_
;
receive_policy_flag
rd_flag_
;
size_t
written_
;
endpoint_manager
*
manager_
;
endpoint_manager
*
manager_
;
};
};
...
...
libcaf_net/caf/net/transport_worker_dispatcher.hpp
View file @
9e81e4f5
...
@@ -32,20 +32,17 @@
...
@@ -32,20 +32,17 @@
#include "caf/span.hpp"
#include "caf/span.hpp"
#include "caf/unit.hpp"
#include "caf/unit.hpp"
namespace
caf
{
namespace
caf
::
net
{
namespace
net
{
/// Implements a dispatcher that dispatches between transport and workers.
/// Implements a dispatcher that dispatches between transport and workers.
template
<
class
Transport
,
class
IdType
>
template
<
class
Factory
,
class
IdType
>
class
transport_worker_dispatcher
{
class
transport_worker_dispatcher
{
public:
public:
// -- member types -----------------------------------------------------------
// -- member types -----------------------------------------------------------
using
id_type
=
IdType
;
using
id_type
=
IdType
;
using
transport_type
=
Transport
;
using
factory_type
=
Factory
;
using
factory_type
=
typename
transport_type
::
factory_type
;
using
application_type
=
typename
factory_type
::
application_type
;
using
application_type
=
typename
factory_type
::
application_type
;
...
@@ -55,8 +52,8 @@ public:
...
@@ -55,8 +52,8 @@ public:
// -- constructors, destructors, and assignment operators --------------------
// -- constructors, destructors, and assignment operators --------------------
transport_worker_dispatcher
(
transport_type
&
transport
,
factory_type
factory
)
transport_worker_dispatcher
(
factory_type
factory
)
:
factory_
(
std
::
move
(
factory
))
,
transport_
(
&
transport
)
{
:
factory_
(
std
::
move
(
factory
))
{
// nop
// nop
}
}
...
@@ -176,8 +173,6 @@ private:
...
@@ -176,8 +173,6 @@ private:
std
::
unordered_map
<
uint64_t
,
worker_ptr
>
workers_by_timeout_id_
;
std
::
unordered_map
<
uint64_t
,
worker_ptr
>
workers_by_timeout_id_
;
factory_type
factory_
;
factory_type
factory_
;
transport_type
*
transport_
;
};
};
}
// namespace net
}
// namespace caf::net
}
// namespace caf
libcaf_net/test/transport_worker_dispatcher.cpp
View file @
9e81e4f5
...
@@ -182,12 +182,12 @@ uri operator"" _u(const char* cstr, size_t cstr_len) {
...
@@ -182,12 +182,12 @@ uri operator"" _u(const char* cstr, size_t cstr_len) {
}
}
struct
fixture
:
host_fixture
{
struct
fixture
:
host_fixture
{
using
dispatcher_type
=
transport_worker_dispatcher
<
dummy_
transport
,
using
dispatcher_type
=
transport_worker_dispatcher
<
dummy_
application_factory
,
ip_endpoint
>
;
ip_endpoint
>
;
fixture
()
fixture
()
:
buf
{
std
::
make_shared
<
buffer_type
>
()},
:
buf
{
std
::
make_shared
<
buffer_type
>
()},
dispatcher
{
dummy
,
dummy
_application_factory
{
buf
}},
dispatcher
{
dummy_application_factory
{
buf
}},
dummy
{
buf
}
{
dummy
{
buf
}
{
add_new_workers
();
add_new_workers
();
}
}
...
@@ -272,7 +272,7 @@ struct fixture : host_fixture {
...
@@ -272,7 +272,7 @@ struct fixture : host_fixture {
CAF_TEST_FIXTURE_SCOPE
(
transport_worker_dispatcher_test
,
fixture
)
CAF_TEST_FIXTURE_SCOPE
(
transport_worker_dispatcher_test
,
fixture
)
CAF_TEST
(
init
)
{
CAF_TEST
(
init
)
{
dispatcher_type
dispatcher
{
dummy
,
dummy
_application_factory
{
buf
}};
dispatcher_type
dispatcher
{
dummy_application_factory
{
buf
}};
CAF_CHECK_EQUAL
(
dispatcher
.
init
(
dummy
),
none
);
CAF_CHECK_EQUAL
(
dispatcher
.
init
(
dummy
),
none
);
}
}
...
...
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