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
c4b3efe6
Commit
c4b3efe6
authored
Nov 15, 2019
by
Joseph Noir
Committed by
GitHub
Nov 15, 2019
Browse files
Options
Browse Files
Download
Plain Diff
Merge pull request #41 from actor-framework/topic/add-transport-base
Add transport_base class
parents
ecda6528
5cf246d8
Changes
8
Hide whitespace changes
Inline
Side-by-side
Showing
8 changed files
with
349 additions
and
355 deletions
+349
-355
libcaf_net/caf/net/datagram_transport.hpp
libcaf_net/caf/net/datagram_transport.hpp
+61
-145
libcaf_net/caf/net/fwd.hpp
libcaf_net/caf/net/fwd.hpp
+10
-4
libcaf_net/caf/net/stream_transport.hpp
libcaf_net/caf/net/stream_transport.hpp
+76
-175
libcaf_net/caf/net/transport_base.hpp
libcaf_net/caf/net/transport_base.hpp
+182
-0
libcaf_net/caf/net/transport_worker.hpp
libcaf_net/caf/net/transport_worker.hpp
+5
-9
libcaf_net/caf/net/transport_worker_dispatcher.hpp
libcaf_net/caf/net/transport_worker_dispatcher.hpp
+10
-18
libcaf_net/test/transport_worker.cpp
libcaf_net/test/transport_worker.cpp
+1
-1
libcaf_net/test/transport_worker_dispatcher.cpp
libcaf_net/test/transport_worker_dispatcher.cpp
+4
-3
No files found.
libcaf_net/caf/net/datagram_transport.hpp
View file @
c4b3efe6
...
...
@@ -19,29 +19,30 @@
#pragma once
#include <deque>
#include <
unordered_map
>
#include <
vector
>
#include "caf/byte.hpp"
#include "caf/error.hpp"
#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"
#include "caf/net/
transport_base
.hpp"
#include "caf/net/transport_worker_dispatcher.hpp"
#include "caf/net/udp_datagram_socket.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp"
namespace
caf
{
namespace
net
{
namespace
caf
::
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.
template
<
class
Factory
>
class
datagram_transport
{
class
datagram_transport
:
public
datagram_transport_base
<
Factory
>
{
public:
// Maximal UDP-packet size
static
constexpr
size_t
max_datagram_size
=
std
::
numeric_limits
<
...
...
@@ -49,163 +50,87 @@ public:
// -- member types -----------------------------------------------------------
using
id_type
=
ip_endpoint
;
using
buffer_type
=
std
::
vector
<
byte
>
;
using
factory_type
=
Factory
;
using
buffer_cache_type
=
std
::
vector
<
buffer_type
>
;
using
id_type
=
ip_endpoint
;
using
factory_type
=
Factory
;
using
application_type
=
typename
factory_type
::
application_type
;
using
transport_type
=
datagram_transport
<
Factory
>
;
using
super
=
datagram_transport_base
<
factory_type
>
;
using
dispatcher_type
=
transport_worker_dispatcher
<
transport_type
,
id_type
>
;
using
buffer_type
=
typename
super
::
buffer_type
;
using
application_type
=
typename
Factory
::
application
_type
;
using
buffer_cache_type
=
typename
super
::
buffer_cache
_type
;
// -- constructors, destructors, and assignment operators --------------------
datagram_transport
(
udp_datagram_socket
handle
,
factory_type
factory
)
:
dispatcher_
(
*
this
,
std
::
move
(
factory
)),
handle_
(
handle
),
read_buf_
(
max_datagram_size
),
manager_
(
nullptr
)
{
:
super
(
handle
,
std
::
move
(
factory
))
{
// 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 ------------------------------------------------
template
<
class
Parent
>
error
init
(
Parent
&
parent
)
{
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
))
error
init
(
endpoint_manager
&
manager
)
override
{
CAF_LOG_TRACE
(
""
);
if
(
auto
err
=
super
::
init
(
manager
))
return
err
;
prepare_next_read
();
return
none
;
}
template
<
class
Parent
>
bool
handle_read_event
(
Parent
&
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
handle_
.
id
));
auto
ret
=
read
(
handle_
,
make_span
(
read_buf_
));
bool
handle_read_event
(
endpoint_manager
&
)
override
{
CAF_LOG_TRACE
(
CAF_ARG
(
this
->
handle_
.
id
));
auto
ret
=
read
(
this
->
handle_
,
make_span
(
this
->
read_buf_
));
if
(
auto
res
=
get_if
<
std
::
pair
<
size_t
,
ip_endpoint
>>
(
&
ret
))
{
auto
num_bytes
=
res
->
first
;
CAF_LOG_DEBUG
(
"received "
<<
num_bytes
<<
" bytes"
);
auto
ep
=
res
->
second
;
read_buf_
.
resize
(
num_bytes
);
dispatcher_
.
handle_data
(
*
this
,
make_span
(
read_buf_
),
std
::
move
(
ep
));
this
->
read_buf_
.
resize
(
num_bytes
);
this
->
next_layer_
.
handle_data
(
*
this
,
make_span
(
this
->
read_buf_
),
std
::
move
(
ep
));
prepare_next_read
();
}
else
{
auto
err
=
get
<
sec
>
(
ret
);
CAF_LOG_DEBUG
(
"send failed"
<<
CAF_ARG
(
err
));
dispatch
er_
.
handle_error
(
err
);
this
->
next_lay
er_
.
handle_error
(
err
);
return
false
;
}
return
true
;
}
template
<
class
Parent
>
bool
handle_write_event
(
Parent
&
parent
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
handle_
.
id
)
bool
handle_write_event
(
endpoint_manager
&
manager
)
override
{
CAF_LOG_TRACE
(
CAF_ARG
(
this
->
handle_
.
id
)
<<
CAF_ARG2
(
"queue-size"
,
packet_queue_
.
size
()));
// Try to write leftover data.
write_some
();
// Get new data from parent.
for
(
auto
msg
=
parent
.
next_message
();
msg
!=
nullptr
;
msg
=
parent
.
next_message
())
{
dispatch
er_
.
write_message
(
*
this
,
std
::
move
(
msg
));
for
(
auto
msg
=
manager
.
next_message
();
msg
!=
nullptr
;
msg
=
manager
.
next_message
())
{
this
->
next_lay
er_
.
write_message
(
*
this
,
std
::
move
(
msg
));
}
// Write prepared data.
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
);
}
// TODO: remove this function. `resolve` should add workers when needed.
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
)
return
worker
.
error
();
return
none
;
}
void
prepare_next_read
()
{
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
)
{
void
write_packet
(
id_type
id
,
span
<
buffer_type
*>
buffers
)
override
{
CAF_LOG_TRACE
(
""
);
CAF_ASSERT
(
!
buffers
.
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
// a payload buffer.
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
struct
packet
{
id_type
id
;
...
...
@@ -219,75 +144,66 @@ public:
bytes
.
emplace_back
(
std
::
move
(
*
buf
));
}
}
std
::
vector
<
std
::
vector
<
byte
>*>
get_buffer_ptrs
()
{
std
::
vector
<
std
::
vector
<
byte
>*>
ptrs
;
for
(
auto
&
buf
:
bytes
)
ptrs
.
emplace_back
(
&
buf
);
return
ptrs
;
}
};
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
;
void
prepare_next_read
()
{
this
->
read_buf_
.
resize
(
max_datagram_size
);
}
bool
write_some
()
{
CAF_LOG_TRACE
(
CAF_ARG
(
handle_
.
id
));
CAF_LOG_TRACE
(
CAF_ARG
2
(
"handle"
,
this
->
handle_
.
id
));
// Helper function to sort empty buffers back into the right caches.
auto
recycle
=
[
&
]()
{
auto
&
front
=
packet_queue_
.
front
();
auto
&
bufs
=
front
.
bytes
;
auto
it
=
bufs
.
begin
();
if
(
header_bufs_
.
size
()
<
header_bufs_
.
capacity
())
{
if
(
this
->
header_bufs_
.
size
()
<
this
->
header_bufs_
.
capacity
())
{
it
->
clear
();
header_bufs_
.
emplace_back
(
std
::
move
(
*
it
++
));
this
->
header_bufs_
.
emplace_back
(
std
::
move
(
*
it
++
));
}
for
(;
it
!=
bufs
.
end
()
&&
payload_bufs_
.
size
()
<
payload_bufs_
.
capacity
();
for
(;
it
!=
bufs
.
end
()
&&
this
->
payload_bufs_
.
size
()
<
this
->
payload_bufs_
.
capacity
();
++
it
)
{
it
->
clear
();
payload_bufs_
.
emplace_back
(
std
::
move
(
*
it
));
this
->
payload_bufs_
.
emplace_back
(
std
::
move
(
*
it
));
}
packet_queue_
.
pop_front
();
};
// Write as many bytes as possible.
while
(
!
packet_queue_
.
empty
())
{
auto
&
packet
=
packet_queue_
.
front
();
std
::
vector
<
std
::
vector
<
byte
>*>
ptrs
;
for
(
auto
&
buf
:
packet
.
bytes
)
ptrs
.
emplace_back
(
&
buf
);
auto
write_ret
=
write
(
handle_
,
make_span
(
ptrs
),
packet
.
id
);
auto
ptrs
=
packet
.
get_buffer_ptrs
();
auto
write_ret
=
write
(
this
->
handle_
,
make_span
(
ptrs
),
packet
.
id
);
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
,
"packet was not sent completely"
);
recycle
();
}
else
{
auto
err
=
get
<
sec
>
(
write_ret
);
if
(
err
!=
sec
::
unavailable_or_would_block
)
{
CAF_LOG_
DEBUG
(
"send
failed"
<<
CAF_ARG
(
err
));
dispatch
er_
.
handle_error
(
err
);
CAF_LOG_
ERROR
(
"write
failed"
<<
CAF_ARG
(
err
));
this
->
next_lay
er_
.
handle_error
(
err
);
return
false
;
}
CAF_LOG_DEBUG
(
"write returned `unavailable_or_would_block`"
);
return
true
;
}
}
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_
;
endpoint_manager
*
manager_
;
};
}
// namespace net
}
// namespace caf
}
// namespace caf::net
libcaf_net/caf/net/fwd.hpp
View file @
c4b3efe6
...
...
@@ -22,11 +22,16 @@
#include "caf/intrusive_ptr.hpp"
namespace
caf
{
namespace
net
{
namespace
caf
::
net
{
// -- templates ----------------------------------------------------------------
template
<
class
Application
>
class
stream_transport
;
template
<
class
Factory
>
class
datagram_transport
;
template
<
class
Application
,
class
IdType
=
unit_t
>
class
transport_worker
;
...
...
@@ -49,6 +54,8 @@ struct socket;
struct
stream_socket
;
struct
tcp_accept_socket
;
struct
tcp_stream_socket
;
struct
datagram_socket
;
struct
udp_datagram_socket
;
// -- smart pointers -----------------------------------------------------------
...
...
@@ -58,5 +65,4 @@ using multiplexer_ptr = std::shared_ptr<multiplexer>;
using
socket_manager_ptr
=
intrusive_ptr
<
socket_manager
>
;
using
weak_multiplexer_ptr
=
std
::
weak_ptr
<
multiplexer
>
;
}
// namespace net
}
// namespace caf
}
// namespace caf::net
libcaf_net/caf/net/stream_transport.hpp
View file @
c4b3efe6
...
...
@@ -5,7 +5,7 @@
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-201
8
Dominik Charousset *
* Copyright 2011-201
9
Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
...
...
@@ -18,248 +18,163 @@
#pragma once
#include "caf/actor_system_config.hpp"
#include "caf/byte.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include <deque>
#include <vector>
#include "caf/fwd.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"
#include "caf/net/stream_socket.hpp"
#include "caf/net/transport_base.hpp"
#include "caf/net/transport_worker.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp"
namespace
caf
{
namespace
net
{
namespace
caf
::
net
{
template
<
class
Application
>
using
stream_transport_base
=
transport_base
<
stream_transport
<
Application
>
,
transport_worker
<
Application
>
,
stream_socket
,
Application
,
unit_t
>
;
/// Implements a stream_transport that manages a stream socket.
template
<
class
Application
>
class
stream_transport
{
class
stream_transport
:
public
stream_transport_base
<
Application
>
{
public:
// -- member types -----------------------------------------------------------
using
application_type
=
Application
;
using
transport_type
=
stream_transport
;
using
worker_type
=
transport_worker
<
application_type
>
;
using
buffer_type
=
std
::
vector
<
byte
>
;
using
id_type
=
unit_t
;
using
super
=
stream_transport_base
<
application_type
>
;
using
buffer_cache_type
=
std
::
vector
<
buffer_type
>
;
using
buffer_type
=
typename
super
::
buffer_type
;
using
write_queue_type
=
std
::
deque
<
std
::
pair
<
bool
,
buffer_type
>>
;
// -- constructors, destructors, and assignment operators --------------------
stream_transport
(
stream_socket
handle
,
application_type
application
)
:
worker_
(
std
::
move
(
application
)),
handle_
(
handle
),
// max_consecutive_reads_(0),
:
super
(
handle
,
std
::
move
(
application
)),
written_
(
0
),
read_threshold_
(
1024
),
collected_
(
0
),
max_
(
1024
),
rd_flag_
(
net
::
receive_policy_flag
::
exactly
),
written_
(
0
),
manager_
(
nullptr
)
{
rd_flag_
(
net
::
receive_policy_flag
::
exactly
)
{
// nop
}
// -- properties -------------------------------------------------------------
stream_socket
handle
()
const
noexcept
{
return
handle_
;
}
actor_system
&
system
()
{
return
manager
().
system
();
}
application_type
&
application
()
{
return
worker_
.
application
();
}
transport_type
&
transport
()
{
return
*
this
;
}
endpoint_manager
&
manager
()
{
return
*
manager_
;
}
// -- member functions -------------------------------------------------------
template
<
class
Parent
>
error
init
(
Parent
&
parent
)
{
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
=
worker_
.
init
(
*
this
))
return
err
;
return
none
;
}
template
<
class
Parent
>
bool
handle_read_event
(
Parent
&
)
{
auto
buf
=
read_buf_
.
data
()
+
collected_
;
size_t
len
=
read_threshold_
-
collected_
;
CAF_LOG_TRACE
(
CAF_ARG
(
handle_
.
id
)
<<
CAF_ARG
(
len
));
auto
ret
=
read
(
handle_
,
make_span
(
buf
,
len
));
bool
handle_read_event
(
endpoint_manager
&
)
override
{
auto
buf
=
this
->
read_buf_
.
data
()
+
this
->
collected_
;
size_t
len
=
this
->
read_threshold_
-
this
->
collected_
;
CAF_LOG_TRACE
(
CAF_ARG2
(
"handle"
,
this
->
handle
().
id
)
<<
CAF_ARG2
(
"missing"
,
len
));
auto
ret
=
read
(
this
->
handle_
,
make_span
(
buf
,
len
));
// Update state.
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
ret
))
{
CAF_LOG_DEBUG
(
CAF_ARG
(
len
)
<<
CAF_ARG
(
handle_
.
id
)
<<
CAF_ARG
(
*
num_bytes
));
collected_
+=
*
num_bytes
;
if
(
collected_
>=
read_threshold_
)
{
if
(
auto
err
=
worker_
.
handle_data
(
*
this
,
read_buf_
))
{
CAF_LOG_WARNING
(
"handle_data failed:"
<<
CAF_ARG
(
err
));
CAF_LOG_DEBUG
(
CAF_ARG
(
len
)
<<
CAF_ARG
(
this
->
handle_
.
id
)
<<
CAF_ARG
(
*
num_bytes
));
this
->
collected_
+=
*
num_bytes
;
if
(
this
->
collected_
>=
this
->
read_threshold_
)
{
if
(
auto
err
=
this
->
next_layer_
.
handle_data
(
*
this
,
this
->
read_buf_
))
{
CAF_LOG_ERROR
(
"handle_data failed: "
<<
CAF_ARG
(
err
));
return
false
;
}
prepare_next_read
();
this
->
prepare_next_read
();
}
}
else
{
auto
err
=
get
<
sec
>
(
ret
);
if
(
err
!=
sec
::
unavailable_or_would_block
)
{
CAF_LOG_DEBUG
(
"re
ceive
failed"
<<
CAF_ARG
(
err
));
work
er_
.
handle_error
(
err
);
CAF_LOG_DEBUG
(
"re
ad
failed"
<<
CAF_ARG
(
err
));
this
->
next_lay
er_
.
handle_error
(
err
);
return
false
;
}
}
return
true
;
}
template
<
class
Parent
>
bool
handle_write_event
(
Parent
&
parent
)
{
bool
handle_write_event
(
endpoint_manager
&
parent
)
override
{
CAF_LOG_TRACE
(
CAF_ARG2
(
"handle"
,
this
->
handle
().
id
));
// Try to write leftover data.
write_some
();
// Get new data from parent.
// TODO: dont read all messages at once - get one by one.
for
(
auto
msg
=
parent
.
next_message
();
msg
!=
nullptr
;
msg
=
parent
.
next_message
())
{
work
er_
.
write_message
(
*
this
,
std
::
move
(
msg
));
this
->
next_lay
er_
.
write_message
(
*
this
,
std
::
move
(
msg
));
}
// Write prepared data.
return
write_some
();
}
template
<
class
Parent
>
void
resolve
(
Parent
&
,
const
uri
&
locator
,
const
actor
&
listener
)
{
worker_
.
resolve
(
*
this
,
locator
.
path
(),
listener
);
}
template
<
class
Parent
>
void
new_proxy
(
Parent
&
,
const
node_id
&
peer
,
actor_id
id
)
{
worker_
.
new_proxy
(
*
this
,
peer
,
id
);
}
template
<
class
Parent
>
void
local_actor_down
(
Parent
&
,
const
node_id
&
peer
,
actor_id
id
,
error
reason
)
{
worker_
.
local_actor_down
(
*
this
,
peer
,
id
,
std
::
move
(
reason
));
}
template
<
class
Parent
>
void
timeout
(
Parent
&
,
atom_value
value
,
uint64_t
id
)
{
worker_
.
timeout
(
*
this
,
value
,
id
);
void
write_packet
(
id_type
,
span
<
buffer_type
*>
buffers
)
override
{
CAF_LOG_TRACE
(
""
);
CAF_ASSERT
(
!
buffers
.
empty
());
if
(
this
->
write_queue_
.
empty
())
this
->
manager
().
register_writing
();
// By convention, the first buffer is a header buffer. Every other buffer is
// a payload buffer.
auto
i
=
buffers
.
begin
();
this
->
write_queue_
.
emplace_back
(
true
,
std
::
move
(
*
(
*
i
++
)));
while
(
i
!=
buffers
.
end
())
this
->
write_queue_
.
emplace_back
(
false
,
std
::
move
(
*
(
*
i
++
)));
}
template
<
class
...
Ts
>
void
set_timeout
(
uint64_t
,
Ts
&&
...)
{
// nop
void
configure_read
(
receive_policy
::
config
cfg
)
override
{
rd_flag_
=
cfg
.
first
;
max_
=
cfg
.
second
;
prepare_next_read
();
}
void
handle_error
(
sec
code
)
{
worker_
.
handle_error
(
code
);
}
private:
// -- utility functions ------------------------------------------------------
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.
switch
(
static_cast
<
net
::
receive_policy_flag
>
(
rd_flag_
))
{
switch
(
rd_flag_
)
{
case
net
:
:
receive_policy_flag
::
exactly
:
if
(
read_buf_
.
size
()
!=
max_
)
read_buf_
.
resize
(
max_
);
if
(
this
->
read_buf_
.
size
()
!=
max_
)
this
->
read_buf_
.
resize
(
max_
);
read_threshold_
=
max_
;
break
;
case
net
:
:
receive_policy_flag
::
at_most
:
if
(
read_buf_
.
size
()
!=
max_
)
read_buf_
.
resize
(
max_
);
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
(
read_buf_
.
size
()
!=
max_size
)
read_buf_
.
resize
(
max_size
);
if
(
this
->
read_buf_
.
size
()
!=
max_size
)
this
->
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
();
}
void
write_packet
(
unit_t
,
span
<
buffer_type
*>
buffers
)
{
CAF_ASSERT
(
!
buffers
.
empty
());
if
(
write_queue_
.
empty
())
manager
().
register_writing
();
// By convention, the first buffer is a header buffer. Every other buffer is
// a payload buffer.
auto
i
=
buffers
.
begin
();
write_queue_
.
emplace_back
(
true
,
std
::
move
(
*
(
*
i
++
)));
while
(
i
!=
buffers
.
end
())
write_queue_
.
emplace_back
(
false
,
std
::
move
(
*
(
*
i
++
)));
}
// -- buffer management ------------------------------------------------------
buffer_type
next_header_buffer
()
{
return
next_buffer_impl
(
header_bufs_
);
}
buffer_type
next_payload_buffer
()
{
return
next_buffer_impl
(
payload_bufs_
);
}
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
()
{
CAF_LOG_TRACE
(
CAF_ARG
(
handle_
.
id
));
CAF_LOG_TRACE
(
CAF_ARG
2
(
"handle"
,
this
->
handle_
.
id
));
// Helper function to sort empty buffers back into the right caches.
auto
recycle
=
[
&
]()
{
auto
&
front
=
write_queue_
.
front
();
auto
&
front
=
this
->
write_queue_
.
front
();
auto
&
is_header
=
front
.
first
;
auto
&
buf
=
front
.
second
;
written_
=
0
;
buf
.
clear
();
if
(
is_header
)
{
if
(
header_bufs_
.
size
()
<
header_bufs_
.
capacity
())
header_bufs_
.
emplace_back
(
std
::
move
(
buf
));
}
else
if
(
payload_bufs_
.
size
()
<
payload_bufs_
.
capacity
())
{
payload_bufs_
.
emplace_back
(
std
::
move
(
buf
));
if
(
this
->
header_bufs_
.
size
()
<
this
->
header_bufs_
.
capacity
())
this
->
header_bufs_
.
emplace_back
(
std
::
move
(
buf
));
}
else
if
(
this
->
payload_bufs_
.
size
()
<
this
->
payload_bufs_
.
capacity
())
{
this
->
payload_bufs_
.
emplace_back
(
std
::
move
(
buf
));
}
write_queue_
.
pop_front
();
};
...
...
@@ -269,9 +184,9 @@ private:
CAF_ASSERT
(
!
buf
.
empty
());
auto
data
=
buf
.
data
()
+
written_
;
auto
len
=
buf
.
size
()
-
written_
;
auto
write_ret
=
write
(
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
))
{
CAF_LOG_DEBUG
(
CAF_ARG
(
handle_
.
id
)
<<
CAF_ARG
(
*
num_bytes
));
CAF_LOG_DEBUG
(
CAF_ARG
(
this
->
handle_
.
id
)
<<
CAF_ARG
(
*
num_bytes
));
if
(
*
num_bytes
+
written_
>=
buf
.
size
())
{
recycle
();
written_
=
0
;
...
...
@@ -283,7 +198,7 @@ private:
auto
err
=
get
<
sec
>
(
write_ret
);
if
(
err
!=
sec
::
unavailable_or_would_block
)
{
CAF_LOG_DEBUG
(
"send failed"
<<
CAF_ARG
(
err
));
work
er_
.
handle_error
(
err
);
this
->
next_lay
er_
.
handle_error
(
err
);
return
false
;
}
return
true
;
...
...
@@ -292,26 +207,12 @@ private:
return
false
;
}
worker_type
worker_
;
stream_socket
handle_
;
buffer_cache_type
header_bufs_
;
buffer_cache_type
payload_bufs_
;
buffer_type
read_buf_
;
std
::
deque
<
std
::
pair
<
bool
,
buffer_type
>>
write_queue_
;
// TODO implement retries using this member!
// size_t max_consecutive_reads_;
write_queue_type
write_queue_
;
size_t
written_
;
size_t
read_threshold_
;
size_t
collected_
;
size_t
max_
;
receive_policy_flag
rd_flag_
;
size_t
written_
;
endpoint_manager
*
manager_
;
};
}
// namespace net
}
// namespace caf
}
// namespace caf::net
libcaf_net/caf/net/transport_base.hpp
0 → 100644
View file @
c4b3efe6
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/detail/overload.hpp"
#include "caf/fwd.hpp"
#include "caf/logger.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/receive_policy.hpp"
namespace
caf
::
net
{
/// Implements base class for transports.
template
<
class
Transport
,
class
NextLayer
,
class
Handle
,
class
Application
,
class
IdType
>
class
transport_base
{
public:
// -- member types -----------------------------------------------------------
using
next_layer_type
=
NextLayer
;
using
handle_type
=
Handle
;
using
transport_type
=
Transport
;
using
application_type
=
Application
;
using
id_type
=
IdType
;
using
buffer_type
=
std
::
vector
<
byte
>
;
using
buffer_cache_type
=
std
::
vector
<
buffer_type
>
;
// -- constructors, destructors, and assignment operators --------------------
transport_base
(
handle_type
handle
,
application_type
application
)
:
next_layer_
(
std
::
move
(
application
)),
handle_
(
handle
),
// max_consecutive_reads_(0),
manager_
(
nullptr
)
{
// nop
}
// -- properties -------------------------------------------------------------
handle_type
handle
()
const
noexcept
{
return
handle_
;
}
actor_system
&
system
()
{
return
manager
().
system
();
}
application_type
&
application
()
{
return
next_layer_
.
application
();
}
transport_type
&
transport
()
{
return
*
reinterpret_cast
<
transport_type
*>
(
this
);
}
endpoint_manager
&
manager
()
{
return
*
manager_
;
}
// -- transport member functions ---------------------------------------------
virtual
error
init
(
endpoint_manager
&
parent
)
{
CAF_LOG_TRACE
(
""
);
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
=
next_layer_
.
init
(
*
this
))
return
err
;
return
none
;
}
auto
resolve
(
endpoint_manager
&
,
const
uri
&
locator
,
const
actor
&
listener
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
locator
)
<<
CAF_ARG
(
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
)
{
next_layer_
.
new_proxy
(
*
this
,
peer
,
id
);
}
void
local_actor_down
(
endpoint_manager
&
,
const
node_id
&
peer
,
actor_id
id
,
error
reason
)
{
next_layer_
.
local_actor_down
(
*
this
,
peer
,
id
,
std
::
move
(
reason
));
}
void
timeout
(
endpoint_manager
&
,
atom_value
value
,
uint64_t
id
)
{
next_layer_
.
timeout
(
*
this
,
value
,
id
);
}
void
set_timeout
(
uint64_t
timeout_id
,
id_type
id
)
{
next_layer_
.
set_timeout
(
timeout_id
,
id
);
}
void
handle_error
(
sec
code
)
{
next_layer_
.
handle_error
(
code
);
}
// -- (pure) virtual functions -----------------------------------------------
virtual
void
configure_read
(
receive_policy
::
config
){
// nop
};
virtual
bool
handle_read_event
(
endpoint_manager
&
)
=
0
;
virtual
bool
handle_write_event
(
endpoint_manager
&
parent
)
=
0
;
virtual
void
write_packet
(
id_type
id
,
span
<
buffer_type
*>
buffers
)
=
0
;
// -- buffer management ------------------------------------------------------
buffer_type
next_header_buffer
()
{
return
next_buffer_impl
(
header_bufs_
);
}
buffer_type
next_payload_buffer
()
{
return
next_buffer_impl
(
payload_bufs_
);
}
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
;
}
protected:
next_layer_type
next_layer_
;
handle_type
handle_
;
buffer_cache_type
header_bufs_
;
buffer_cache_type
payload_bufs_
;
buffer_type
read_buf_
;
// TODO implement retries using this member! Should this go into stream_trans?
// size_t max_consecutive_reads_;
endpoint_manager
*
manager_
;
};
}
// namespace caf::net
libcaf_net/caf/net/transport_worker.hpp
View file @
c4b3efe6
...
...
@@ -18,16 +18,12 @@
#pragma once
#include "caf/byte.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/endpoint_manager_queue.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/packet_writer_decorator.hpp"
#include "caf/span.hpp"
#include "caf/unit.hpp"
namespace
caf
{
namespace
net
{
namespace
caf
::
net
{
/// Implements a worker for transport protocols.
template
<
class
Application
,
class
IdType
>
...
...
@@ -41,7 +37,8 @@ public:
// -- constructors, destructors, and assignment operators --------------------
transport_worker
(
application_type
application
,
id_type
id
=
id_type
{})
explicit
transport_worker
(
application_type
application
,
id_type
id
=
id_type
{})
:
application_
(
std
::
move
(
application
)),
id_
(
std
::
move
(
id
))
{
// nop
}
...
...
@@ -119,5 +116,4 @@ template <class Application, class IdType = unit_t>
using
transport_worker_ptr
=
std
::
shared_ptr
<
transport_worker
<
Application
,
IdType
>>
;
}
// namespace net
}
// namespace caf
}
// namespace caf::net
libcaf_net/caf/net/transport_worker_dispatcher.hpp
View file @
c4b3efe6
...
...
@@ -18,34 +18,27 @@
#pragma once
#include <caf/logger.hpp>
#include <caf/sec.hpp>
#include <unordered_map>
#include "caf/byte.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/logger.hpp"
#include "caf/net/endpoint_manager_queue.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/packet_writer_decorator.hpp"
#include "caf/net/transport_worker.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"
#include "caf/span.hpp"
#include "caf/unit.hpp"
namespace
caf
{
namespace
net
{
namespace
caf
::
net
{
/// Implements a dispatcher that dispatches between transport and workers.
template
<
class
Transport
,
class
IdType
>
template
<
class
Factory
,
class
IdType
>
class
transport_worker_dispatcher
{
public:
// -- member types -----------------------------------------------------------
using
id_type
=
IdType
;
using
transport_type
=
Transport
;
using
factory_type
=
typename
transport_type
::
factory_type
;
using
factory_type
=
Factory
;
using
application_type
=
typename
factory_type
::
application_type
;
...
...
@@ -55,8 +48,8 @@ public:
// -- constructors, destructors, and assignment operators --------------------
transport_worker_dispatcher
(
transport_type
&
transport
,
factory_type
factory
)
:
factory_
(
std
::
move
(
factory
))
,
transport_
(
&
transport
)
{
explicit
transport_worker_dispatcher
(
factory_type
factory
)
:
factory_
(
std
::
move
(
factory
))
{
// nop
}
...
...
@@ -141,6 +134,7 @@ public:
template
<
class
Parent
>
expected
<
worker_ptr
>
add_new_worker
(
Parent
&
parent
,
node_id
node
,
id_type
id
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
node
)
<<
CAF_ARG
(
id
));
auto
application
=
factory_
.
make
();
auto
worker
=
std
::
make_shared
<
worker_type
>
(
std
::
move
(
application
),
id
);
if
(
auto
err
=
worker
->
init
(
parent
))
...
...
@@ -176,8 +170,6 @@ private:
std
::
unordered_map
<
uint64_t
,
worker_ptr
>
workers_by_timeout_id_
;
factory_type
factory_
;
transport_type
*
transport_
;
};
}
// namespace net
}
// namespace caf
}
// namespace caf::net
libcaf_net/test/transport_worker.cpp
View file @
c4b3efe6
...
...
@@ -25,9 +25,9 @@
#include "caf/byte.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/span.hpp"
...
...
libcaf_net/test/transport_worker_dispatcher.cpp
View file @
c4b3efe6
...
...
@@ -23,6 +23,7 @@
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/make_actor.hpp"
#include "caf/monitorable_actor.hpp"
#include "caf/node_id.hpp"
...
...
@@ -182,12 +183,12 @@ uri operator"" _u(const char* cstr, size_t cstr_len) {
}
struct
fixture
:
host_fixture
{
using
dispatcher_type
=
transport_worker_dispatcher
<
dummy_
transport
,
using
dispatcher_type
=
transport_worker_dispatcher
<
dummy_
application_factory
,
ip_endpoint
>
;
fixture
()
:
buf
{
std
::
make_shared
<
buffer_type
>
()},
dispatcher
{
dummy
,
dummy
_application_factory
{
buf
}},
dispatcher
{
dummy_application_factory
{
buf
}},
dummy
{
buf
}
{
add_new_workers
();
}
...
...
@@ -272,7 +273,7 @@ struct fixture : host_fixture {
CAF_TEST_FIXTURE_SCOPE
(
transport_worker_dispatcher_test
,
fixture
)
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
);
}
...
...
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