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
38a661d8
Commit
38a661d8
authored
Jun 27, 2018
by
Joseph Noir
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Remove ASIO-based multiplexer, close #709
parent
43e2d1e0
Changes
9
Show whitespace changes
Inline
Side-by-side
Showing
9 changed files
with
2 additions
and
894 deletions
+2
-894
CMakeLists.txt
CMakeLists.txt
+0
-22
Jenkinsfile
Jenkinsfile
+0
-1
cmake/build_config.hpp.in
cmake/build_config.hpp.in
+0
-4
configure
configure
+0
-6
examples/caf-application.ini
examples/caf-application.ini
+0
-2
libcaf_io/caf/io/network/asio_multiplexer.hpp
libcaf_io/caf/io/network/asio_multiplexer.hpp
+0
-506
libcaf_io/caf/io/network/asio_multiplexer_impl.hpp
libcaf_io/caf/io/network/asio_multiplexer_impl.hpp
+0
-342
libcaf_io/src/middleman.cpp
libcaf_io/src/middleman.cpp
+0
-9
libcaf_openssl/caf/openssl/manager.hpp
libcaf_openssl/caf/openssl/manager.hpp
+2
-2
No files found.
CMakeLists.txt
View file @
38a661d8
...
...
@@ -389,23 +389,6 @@ to_int_value(CAF_NO_EXCEPTIONS)
to_int_value
(
CAF_NO_MEM_MANAGEMENT
)
to_int_value
(
CAF_ENABLE_RUNTIME_CHECKS
)
# find boost asio if the asio multiplexer should be used for testing
if
(
CAF_USE_ASIO
)
set
(
Boost_USE_MULTITHREADED ON
)
find_package
(
Boost COMPONENTS system
)
if
(
Boost_FOUND
)
include_directories
(
${
Boost_INCLUDE_DIRS
}
)
set
(
LD_DIRS
${
LD_DIRS
}
${
Boost_LIBRARIES
}
)
list
(
APPEND LDFLAGS
${
Boost_SYSTEM_LIBRARY
}
)
set
(
CAF_USE_ASIO_INT 1
)
else
()
set
(
CAF_USE_ASIO no
)
set
(
CAF_USE_ASIO_INT -1
)
endif
()
else
()
set
(
CAF_USE_ASIO_INT -1
)
endif
()
configure_file
(
"
${
CMAKE_CURRENT_SOURCE_DIR
}
/cmake/build_config.hpp.in"
"
${
CMAKE_CURRENT_SOURCE_DIR
}
/libcaf_core/caf/detail/build_config.hpp"
IMMEDIATE @ONLY
)
...
...
@@ -639,11 +622,6 @@ if(NOT CAF_NO_UNIT_TESTS)
string
(
REPLACE
" "
"_"
test_name
${
suite
}
)
set
(
caf_test
${
EXECUTABLE_OUTPUT_PATH
}
/caf-test
)
add_test
(
${
test_name
}
${
caf_test
}
-n -v 5 -s
"
${
suite
}
"
${
ARGN
}
)
# add some extra unit testing options when testing ASIO as well
if
(
CAF_USE_ASIO AND
"
${
suite
}
"
MATCHES
"^io_.+$"
AND NOT
"
${
suite
}
"
MATCHES
".+_udp$"
)
add_test
(
${
test_name
}
_asio
${
caf_test
}
-n -v 5 -s
"
${
suite
}
"
${
ARGN
}
--
"--caf#middleman.network-backend=asio"
)
endif
()
endmacro
()
list
(
LENGTH suites num_suites
)
message
(
STATUS
"Found
${
num_suites
}
test suites"
)
...
...
Jenkinsfile
View file @
38a661d8
...
...
@@ -69,7 +69,6 @@ unixOpts = "-DCAF_NO_PROTOBUF_EXAMPLES:BOOL=yes " +
"-DCAF_MORE_WARNINGS:BOOL=yes "
+
"-DCAF_ENABLE_ADDRESS_SANITIZER:BOOL=yes "
+
"-DCAF_ENABLE_RUNTIME_CHECKS:BOOL=yes "
+
"-DCAF_USE_ASIO:BOOL=yes "
+
"-DCAF_NO_BENCHMARKS:BOOL=yes "
+
"-DOPENSSL_ROOT_DIR=/usr/local/opt/openssl "
+
"-DOPENSSL_INCLUDE_DIR=/usr/local/opt/openssl/include"
...
...
cmake/build_config.hpp.in
View file @
38a661d8
...
...
@@ -33,10 +33,6 @@
#define CAF_ENABLE_RUNTIME_CHECKS
#endif
#if @CAF_USE_ASIO_INT@ != -1
#define CAF_USE_ASIO
#endif
#if @CAF_NO_EXCEPTIONS_INT@ != -1
#define CAF_NO_EXCEPTIONS
#endif
...
...
configure
View file @
38a661d8
...
...
@@ -76,9 +76,6 @@ Usage: $0 [OPTION]... [VAR=VALUE]...
--core-only sets no-examples, no-opencl, no-openssl,
no-tools, no-python, and no-io
Testing:
--with-asio use ASIO multiplexer in unit tests
Debugging:
--with-runtime-checks build with requirement checks at runtime
--with-log-level=LVL build with debugging output, possible values:
...
...
@@ -287,9 +284,6 @@ while [ $# -ne 0 ]; do
append_cache_entry CMAKE_OSX_ARCHITECTURES STRING
"
\$
(ARCHS_STANDARD_32_64_BIT)"
append_cache_entry CAF_IOS_DEPLOYMENT_TARGET STRING
"
$optarg
"
;;
--with-asio
)
append_cache_entry CAF_USE_ASIO BOOL
yes
;;
--with-log-level
=
*
)
level
=
`
echo
"
$optarg
"
|
tr
'[:lower:]'
'[:upper:]'
`
case
$level
in
...
...
examples/caf-application.ini
View file @
38a661d8
...
...
@@ -37,8 +37,6 @@ relaxed-sleep-duration=10ms
[middleman]
; configures whether MMs try to span a full mesh
enable-automatic-connections
=
false
; accepted alternative: 'asio' (only when compiling CAF with ASIO)
network-backend
=
'default'
; application identifier of this node, prevents connection to other CAF
; instances with different identifier
app-identifier
=
""
...
...
libcaf_io/caf/io/network/asio_multiplexer.hpp
deleted
100644 → 0
View file @
43e2d1e0
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* Raphael Hiesgen <raphael.hiesgen (at) haw-hamburg.de> *
* *
* 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/config.hpp"
CAF_PUSH_WARNINGS
#include "boost/asio.hpp"
CAF_POP_WARNINGS
#include "caf/logger.hpp"
#include "caf/io/receive_policy.hpp"
#include "caf/io/network/multiplexer.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/io/network/stream_manager.hpp"
#include "caf/io/network/acceptor_manager.hpp"
namespace
caf
{
namespace
io
{
namespace
network
{
/// Low-level error code.
using
error_code
=
boost
::
system
::
error_code
;
/// Low-level backend for IO multiplexing.
using
io_service
=
boost
::
asio
::
io_service
;
/// Low-level socket type used as default.
using
asio_tcp_socket
=
boost
::
asio
::
ip
::
tcp
::
socket
;
/// Low-level socket type used as default.
using
asio_tcp_socket_acceptor
=
boost
::
asio
::
ip
::
tcp
::
acceptor
;
class
multiplexer_backend
:
public
io_service
{
// nop
};
/// A wrapper for the boost::asio multiplexer
class
asio_multiplexer
:
public
multiplexer
{
public:
friend
class
io
::
middleman
;
friend
class
supervisor
;
scribe_ptr
new_scribe
(
asio_tcp_socket
&&
sock
);
scribe_ptr
new_scribe
(
native_socket
fd
)
override
;
expected
<
scribe_ptr
>
new_tcp_scribe
(
const
std
::
string
&
host
,
uint16_t
port
)
override
;
doorman_ptr
new_doorman
(
asio_tcp_socket_acceptor
&&
sock
);
doorman_ptr
new_doorman
(
native_socket
fd
)
override
;
expected
<
doorman_ptr
>
new_tcp_doorman
(
uint16_t
port
,
const
char
*
in
,
bool
reuse_addr
)
override
;
datagram_servant_ptr
new_datagram_servant
(
native_socket
fd
)
override
;
datagram_servant_ptr
new_datagram_servant_for_endpoint
(
native_socket
fd
,
const
ip_endpoint
&
ep
)
override
;
expected
<
datagram_servant_ptr
>
new_remote_udp_endpoint
(
const
std
::
string
&
host
,
uint16_t
port
)
override
;
expected
<
datagram_servant_ptr
>
new_local_udp_endpoint
(
uint16_t
port
,
const
char
*
in
=
nullptr
,
bool
reuse_addr
=
false
)
override
;
void
exec_later
(
resumable
*
ptr
)
override
;
asio_multiplexer
(
actor_system
*
sys
);
~
asio_multiplexer
();
supervisor_ptr
make_supervisor
()
override
;
bool
try_run_once
()
override
;
void
run_once
()
override
;
void
run
()
override
;
multiplexer_backend
*
pimpl
()
override
;
inline
boost
::
asio
::
io_service
&
service
()
{
return
service_
;
}
private:
multiplexer_backend
service_
;
};
template
<
class
T
>
connection_handle
conn_hdl_from_socket
(
T
&
sock
)
{
return
connection_handle
::
from_int
(
int64_from_native_socket
(
sock
.
native_handle
()));
}
template
<
class
T
>
accept_handle
accept_hdl_from_socket
(
T
&
sock
)
{
return
accept_handle
::
from_int
(
int64_from_native_socket
(
sock
.
native_handle
()));
}
/// @relates manager
using
manager_ptr
=
intrusive_ptr
<
manager
>
;
/// A stream capable of both reading and writing. The stream's input
/// data is forwarded to its {@link stream_manager manager}.
template
<
class
Socket
>
class
asio_stream
{
public:
/// A smart pointer to a stream manager.
using
manager_ptr
=
intrusive_ptr
<
stream_manager
>
;
/// A buffer class providing a compatible interface to `std::vector`.
using
buffer_type
=
std
::
vector
<
char
>
;
asio_stream
(
asio_multiplexer
&
ref
)
:
reading_
(
false
),
writing_
(
false
),
ack_writes_
(
false
),
fd_
(
ref
.
service
()),
backend_
(
ref
),
rd_buf_ready_
(
false
),
async_read_pending_
(
false
)
{
configure_read
(
receive_policy
::
at_most
(
1024
));
}
/// Returns the IO socket.
Socket
&
socket_handle
()
{
return
fd_
;
}
/// Returns the IO socket.
const
Socket
&
socket_handle
()
const
{
return
fd_
;
}
/// Initializes this stream, setting the socket handle to `fd`.
void
init
(
Socket
fd
)
{
fd_
=
std
::
move
(
fd
);
}
/// Starts reading data from the socket, forwarding incoming data to `mgr`.
void
start
(
stream_manager
*
mgr
)
{
CAF_ASSERT
(
mgr
!=
nullptr
);
activate
(
mgr
);
}
/// Configures how much data will be provided for the next `consume` callback.
/// @warning Must not be called outside the IO multiplexers event loop
/// once the stream has been started.
void
configure_read
(
receive_policy
::
config
config
)
{
rd_flag_
=
config
.
first
;
rd_size_
=
config
.
second
;
}
void
ack_writes
(
bool
enable
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
enable
));
ack_writes_
=
enable
;
}
/// Copies data to the write buffer.
/// @note Not thread safe.
void
write
(
const
void
*
buf
,
size_t
num_bytes
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
num_bytes
));
auto
first
=
reinterpret_cast
<
const
char
*>
(
buf
);
auto
last
=
first
+
num_bytes
;
wr_offline_buf_
.
insert
(
wr_offline_buf_
.
end
(),
first
,
last
);
}
/// Returns the write buffer of this stream.
/// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started.
buffer_type
&
wr_buf
()
{
return
wr_offline_buf_
;
}
buffer_type
&
rd_buf
()
{
return
rd_buf_
;
}
/// Sends the content of the write buffer, calling the `io_failure`
/// member function of `mgr` in case of an error.
/// @warning Must not be called outside the IO multiplexers event loop
/// once the stream has been started.
void
flush
(
const
manager_ptr
&
mgr
)
{
CAF_ASSERT
(
mgr
!=
nullptr
);
if
(
!
wr_offline_buf_
.
empty
()
&&
!
writing_
)
{
writing_
=
true
;
write_loop
(
mgr
);
}
}
/// Closes the network connection, thus stopping this stream.
void
stop
()
{
CAF_LOG_TRACE
(
""
);
fd_
.
close
();
}
void
stop_reading
()
{
CAF_LOG_TRACE
(
""
);
error_code
ec
;
// ignored
fd_
.
shutdown
(
boost
::
asio
::
ip
::
tcp
::
socket
::
shutdown_receive
,
ec
);
}
asio_multiplexer
&
backend
()
{
return
backend_
;
}
/// Activates the stream.
void
activate
(
stream_manager
*
mgr
)
{
reading_
=
true
;
read_loop
(
mgr
);
}
/// Stops activity of the stream.
void
passivate
()
{
reading_
=
false
;
}
private:
bool
read_one
(
stream_manager
*
ptr
,
size_t
num_bytes
)
{
if
(
!
reading_
)
{
// broker was passivated while async read was on its way
rd_buf_ready_
=
true
;
// make sure buf size matches read_bytes in case of async_read
if
(
rd_buf_
.
size
()
!=
num_bytes
)
rd_buf_
.
resize
(
num_bytes
);
return
false
;
}
if
(
ptr
->
consume
(
&
backend
(),
rd_buf_
.
data
(),
num_bytes
))
return
reading_
;
return
false
;
}
void
read_loop
(
manager_ptr
mgr
)
{
if
(
async_read_pending_
)
return
;
if
(
rd_buf_ready_
)
{
rd_buf_ready_
=
false
;
if
(
read_one
(
mgr
.
get
(),
rd_buf_
.
size
()))
read_loop
(
std
::
move
(
mgr
));
return
;
}
auto
cb
=
[
=
](
const
error_code
&
ec
,
size_t
read_bytes
)
mutable
{
async_read_pending_
=
false
;
CAF_LOG_TRACE
(
""
);
if
(
!
ec
)
{
// bail out early in case broker passivated stream in the meantime
if
(
read_one
(
mgr
.
get
(),
read_bytes
))
read_loop
(
std
::
move
(
mgr
));
}
else
{
mgr
->
io_failure
(
&
backend
(),
operation
::
read
);
}
};
switch
(
rd_flag_
)
{
case
receive_policy_flag
:
:
exactly
:
if
(
rd_buf_
.
size
()
<
rd_size_
)
rd_buf_
.
resize
(
rd_size_
);
async_read_pending_
=
true
;
boost
::
asio
::
async_read
(
fd_
,
boost
::
asio
::
buffer
(
rd_buf_
,
rd_size_
),
cb
);
break
;
case
receive_policy_flag
:
:
at_most
:
if
(
rd_buf_
.
size
()
<
rd_size_
)
rd_buf_
.
resize
(
rd_size_
);
async_read_pending_
=
true
;
fd_
.
async_read_some
(
boost
::
asio
::
buffer
(
rd_buf_
,
rd_size_
),
cb
);
break
;
case
receive_policy_flag
:
:
at_least
:
{
// read up to 10% more, but at least allow 100 bytes more
auto
min_size
=
rd_size_
+
std
::
max
<
size_t
>
(
100
,
rd_size_
/
10
);
if
(
rd_buf_
.
size
()
<
min_size
)
{
rd_buf_
.
resize
(
min_size
);
}
collect_data
(
mgr
,
0
);
break
;
}
}
}
void
write_loop
(
const
manager_ptr
&
mgr
)
{
if
(
wr_offline_buf_
.
empty
())
{
writing_
=
false
;
return
;
}
wr_buf_
.
clear
();
wr_buf_
.
swap
(
wr_offline_buf_
);
boost
::
asio
::
async_write
(
fd_
,
boost
::
asio
::
buffer
(
wr_buf_
),
[
=
](
const
error_code
&
ec
,
size_t
nb
)
{
CAF_LOG_TRACE
(
""
);
if
(
ec
)
{
CAF_LOG_DEBUG
(
CAF_ARG
(
ec
.
message
()));
mgr
->
io_failure
(
&
backend
(),
operation
::
read
);
writing_
=
false
;
return
;
}
CAF_LOG_DEBUG
(
CAF_ARG
(
nb
));
if
(
ack_writes_
)
mgr
->
data_transferred
(
&
backend
(),
nb
,
wr_offline_buf_
.
size
());
write_loop
(
mgr
);
});
}
void
collect_data
(
manager_ptr
mgr
,
size_t
collected_bytes
)
{
async_read_pending_
=
true
;
fd_
.
async_read_some
(
boost
::
asio
::
buffer
(
rd_buf_
.
data
()
+
collected_bytes
,
rd_buf_
.
size
()
-
collected_bytes
),
[
=
](
const
error_code
&
ec
,
size_t
nb
)
mutable
{
async_read_pending_
=
false
;
CAF_LOG_TRACE
(
CAF_ARG
(
nb
));
if
(
!
ec
)
{
auto
sum
=
collected_bytes
+
nb
;
if
(
sum
>=
rd_size_
)
{
if
(
read_one
(
mgr
.
get
(),
sum
))
read_loop
(
std
::
move
(
mgr
));
}
else
{
collect_data
(
std
::
move
(
mgr
),
sum
);
}
}
else
{
mgr
->
io_failure
(
&
backend
(),
operation
::
write
);
}
});
}
/// Set if read loop was started by user and unset if passivate is called.
bool
reading_
;
/// Set on flush, also indicates that an async_write is pending.
bool
writing_
;
/// Stores whether user requested ACK messages for async writes.
bool
ack_writes_
;
/// TCP socket for this connection.
Socket
fd_
;
/// Configures how chunk sizes are calculated.
receive_policy_flag
rd_flag_
;
/// Minimum, maximum, or exact size of a chunk, depending on `rd_flag_`.
size_t
rd_size_
;
/// Input buffer.
buffer_type
rd_buf_
;
/// Output buffer for ASIO.
buffer_type
wr_buf_
;
/// Swapped with `wr_buf_` before next write. Users write into this buffer
/// as long as `wr_buf_` is processed by ASIO.
buffer_type
wr_offline_buf_
;
/// Reference to our I/O backend.
asio_multiplexer
&
backend_
;
/// Signalizes that a scribe was passivated while an async read was pending.
bool
rd_buf_ready_
;
/// Makes sure no more than one async_read is pending at any given time
bool
async_read_pending_
;
};
/// An acceptor is responsible for accepting incoming connections.
template
<
class
SocketAcceptor
>
class
asio_acceptor
{
using
protocol_type
=
typename
SocketAcceptor
::
protocol_type
;
using
socket_type
=
boost
::
asio
::
basic_stream_socket
<
protocol_type
>
;
public:
/// A manager providing the `accept` member function.
using
manager_type
=
acceptor_manager
;
/// A smart pointer to an acceptor manager.
using
manager_ptr
=
intrusive_ptr
<
manager_type
>
;
asio_acceptor
(
asio_multiplexer
&
am
,
io_service
&
io
)
:
accepting_
(
false
),
backend_
(
am
),
accept_fd_
(
io
),
fd_valid_
(
false
),
fd_
(
io
),
async_accept_pending_
(
false
)
{
// nop
}
/// Returns the `multiplexer` this acceptor belongs to.
asio_multiplexer
&
backend
()
{
return
backend_
;
}
/// Returns the IO socket.
SocketAcceptor
&
socket_handle
()
{
return
accept_fd_
;
}
/// Returns the IO socket.
const
SocketAcceptor
&
socket_handle
()
const
{
return
accept_fd_
;
}
/// Returns the accepted socket. This member function should
/// be called only from the `new_connection` callback.
inline
socket_type
&
accepted_socket
()
{
return
fd_
;
}
/// Initializes this acceptor, setting the socket handle to `fd`.
void
init
(
SocketAcceptor
fd
)
{
accept_fd_
=
std
::
move
(
fd
);
}
/// Starts this acceptor, forwarding all incoming connections to
/// `manager`. The intrusive pointer will be released after the
/// acceptor has been closed or an IO error occured.
void
start
(
manager_type
*
mgr
)
{
activate
(
mgr
);
}
/// Starts the accept loop.
void
activate
(
manager_type
*
mgr
)
{
accepting_
=
true
;
accept_loop
(
mgr
);
}
/// Starts the accept loop.
void
passivate
()
{
accepting_
=
false
;
}
/// Closes the network connection, thus stopping this acceptor.
void
stop
()
{
accept_fd_
.
close
();
}
private:
bool
accept_one
(
manager_type
*
mgr
)
{
auto
res
=
mgr
->
new_connection
();
// moves fd_
// reset fd_ for next accept operation
fd_
=
socket_type
{
accept_fd_
.
get_io_service
()};
return
res
&&
accepting_
;
}
void
accept_loop
(
manager_ptr
mgr
)
{
if
(
async_accept_pending_
)
return
;
// accept "cached" connection first
if
(
fd_valid_
)
{
fd_valid_
=
false
;
if
(
accept_one
(
mgr
.
get
()))
accept_loop
(
std
::
move
(
mgr
));
return
;
}
async_accept_pending_
=
true
;
accept_fd_
.
async_accept
(
fd_
,
[
=
](
const
error_code
&
ec
)
mutable
{
CAF_LOG_TRACE
(
""
);
async_accept_pending_
=
false
;
if
(
!
ec
)
{
// if broker has passivated this in the meantime, cache fd_ for later
if
(
!
accepting_
)
{
fd_valid_
=
true
;
return
;
}
if
(
accept_one
(
mgr
.
get
()))
accept_loop
(
std
::
move
(
mgr
));
}
else
{
mgr
->
io_failure
(
&
backend
(),
operation
::
read
);
}
});
}
bool
accepting_
;
asio_multiplexer
&
backend_
;
SocketAcceptor
accept_fd_
;
bool
fd_valid_
;
socket_type
fd_
;
/// Makes sure no more than one async_accept is pending at any given time
bool
async_accept_pending_
;
};
}
// namesapce network
}
// namespace io
}
// namespace caf
libcaf_io/caf/io/network/asio_multiplexer_impl.hpp
deleted
100644 → 0
View file @
43e2d1e0
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* Raphael Hiesgen <raphael.hiesgen (at) haw-hamburg.de> *
* *
* 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/io/broker.hpp"
#include "caf/io/middleman.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/io/network/asio_multiplexer.hpp"
namespace
caf
{
namespace
io
{
namespace
network
{
namespace
{
/// A wrapper for the supervisor backend provided by boost::asio.
struct
asio_supervisor
:
public
multiplexer
::
supervisor
{
explicit
asio_supervisor
(
io_service
&
iob
)
:
work
(
iob
)
{
// nop
}
private:
boost
::
asio
::
io_service
::
work
work
;
};
}
// namespace anonymous
expected
<
asio_tcp_socket
>
new_tcp_connection
(
io_service
&
ios
,
const
std
::
string
&
host
,
uint16_t
port
)
{
asio_tcp_socket
fd
{
ios
};
using
boost
::
asio
::
ip
::
tcp
;
tcp
::
resolver
r
(
fd
.
get_io_service
());
tcp
::
resolver
::
query
q
(
host
,
std
::
to_string
(
port
));
boost
::
system
::
error_code
ec
;
auto
i
=
r
.
resolve
(
q
,
ec
);
if
(
ec
)
return
make_error
(
sec
::
cannot_connect_to_node
);
boost
::
asio
::
connect
(
fd
,
i
,
ec
);
if
(
ec
)
return
make_error
(
sec
::
cannot_connect_to_node
);
return
asio_tcp_socket
{
std
::
move
(
fd
)};
}
error
ip_bind
(
asio_tcp_socket_acceptor
&
fd
,
uint16_t
port
,
const
char
*
addr
,
bool
reuse_addr
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
port
)
<<
CAF_ARG
(
reuse_addr
));
using
boost
::
asio
::
ip
::
tcp
;
auto
bind_and_listen
=
[
&
](
tcp
::
endpoint
&
ep
)
->
error
{
CAF_LOG_DEBUG
(
"created IP endpoint:"
<<
CAF_ARG
(
ep
.
address
().
to_string
())
<<
CAF_ARG
(
ep
.
port
()));
boost
::
system
::
error_code
ec
;
fd
.
open
(
ep
.
protocol
());
if
(
reuse_addr
)
{
fd
.
set_option
(
tcp
::
acceptor
::
reuse_address
(
reuse_addr
),
ec
);
if
(
ec
)
return
sec
::
cannot_open_port
;
}
fd
.
bind
(
ep
,
ec
);
if
(
ec
)
return
sec
::
cannot_open_port
;
fd
.
listen
(
asio_tcp_socket_acceptor
::
max_connections
,
ec
);
if
(
ec
)
return
sec
::
cannot_open_port
;
return
none
;
};
if
(
addr
)
{
CAF_LOG_DEBUG
(
CAF_ARG
(
addr
));
tcp
::
endpoint
ep
(
boost
::
asio
::
ip
::
address
::
from_string
(
addr
),
port
);
CAF_LOG_DEBUG
(
"got 'em"
);
return
bind_and_listen
(
ep
);
}
else
{
CAF_LOG_DEBUG
(
"addr = nullptr"
);
tcp
::
endpoint
ep
(
tcp
::
v6
(),
port
);
return
bind_and_listen
(
ep
);
}
}
scribe_ptr
asio_multiplexer
::
new_scribe
(
asio_tcp_socket
&&
sock
)
{
CAF_LOG_TRACE
(
""
);
class
impl
:
public
scribe
{
public:
impl
(
asio_multiplexer
&
am
,
asio_tcp_socket
&&
s
)
:
scribe
(
network
::
conn_hdl_from_socket
(
s
)),
launched_
(
false
),
stream_
(
am
)
{
stream_
.
init
(
std
::
move
(
s
));
}
void
configure_read
(
receive_policy
::
config
config
)
override
{
CAF_LOG_TRACE
(
""
);
stream_
.
configure_read
(
config
);
if
(
!
launched_
)
{
launch
();
}
}
void
ack_writes
(
bool
enable
)
override
{
CAF_LOG_TRACE
(
CAF_ARG
(
enable
));
stream_
.
ack_writes
(
enable
);
}
std
::
vector
<
char
>&
wr_buf
()
override
{
return
stream_
.
wr_buf
();
}
std
::
vector
<
char
>&
rd_buf
()
override
{
return
stream_
.
rd_buf
();
}
void
stop_reading
()
override
{
CAF_LOG_TRACE
(
""
);
stream_
.
stop_reading
();
detach
(
&
stream_
.
backend
(),
false
);
}
void
flush
()
override
{
CAF_LOG_TRACE
(
""
);
stream_
.
flush
(
this
);
}
std
::
string
addr
()
const
override
{
return
stream_
.
socket_handle
().
remote_endpoint
().
address
().
to_string
();
}
uint16_t
port
()
const
override
{
return
stream_
.
socket_handle
().
remote_endpoint
().
port
();
}
void
launch
()
{
CAF_LOG_TRACE
(
""
);
CAF_ASSERT
(
!
launched_
);
launched_
=
true
;
stream_
.
start
(
this
);
}
void
add_to_loop
()
override
{
stream_
.
activate
(
this
);
}
void
remove_from_loop
()
override
{
stream_
.
passivate
();
}
private:
bool
launched_
;
asio_stream
<
asio_tcp_socket
>
stream_
;
};
return
make_counted
<
impl
>
(
*
this
,
std
::
move
(
sock
));
}
scribe_ptr
asio_multiplexer
::
new_scribe
(
native_socket
fd
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
fd
));
boost
::
system
::
error_code
ec
;
asio_tcp_socket
sock
{
service
()};
sock
.
assign
(
boost
::
asio
::
ip
::
tcp
::
v6
(),
fd
,
ec
);
if
(
ec
)
sock
.
assign
(
boost
::
asio
::
ip
::
tcp
::
v4
(),
fd
,
ec
);
if
(
ec
)
CAF_RAISE_ERROR
(
ec
.
message
());
return
new_scribe
(
std
::
move
(
sock
));
}
expected
<
scribe_ptr
>
asio_multiplexer
::
new_tcp_scribe
(
const
std
::
string
&
host
,
uint16_t
port
)
{
auto
sck
=
new_tcp_connection
(
service
(),
host
,
port
);
if
(
!
sck
)
return
std
::
move
(
sck
.
error
());
return
new_scribe
(
std
::
move
(
*
sck
));
}
doorman_ptr
asio_multiplexer
::
new_doorman
(
asio_tcp_socket_acceptor
&&
sock
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
sock
.
native_handle
()));
CAF_ASSERT
(
sock
.
native_handle
()
!=
network
::
invalid_native_socket
);
class
impl
:
public
doorman
{
public:
impl
(
asio_tcp_socket_acceptor
&&
s
,
network
::
asio_multiplexer
&
am
)
:
doorman
(
network
::
accept_hdl_from_socket
(
s
)),
acceptor_
(
am
,
s
.
get_io_service
())
{
acceptor_
.
init
(
std
::
move
(
s
));
}
bool
new_connection
()
override
{
CAF_LOG_TRACE
(
""
);
if
(
detached
())
// we are already disconnected from the broker while the multiplexer
// did not yet remove the socket, this can happen if an I/O event causes
// the broker to call close_all() while the pollset contained
// further activities for the broker
return
false
;
auto
&
am
=
acceptor_
.
backend
();
auto
sptr
=
am
.
new_scribe
(
std
::
move
(
acceptor_
.
accepted_socket
()));
auto
shdl
=
sptr
->
hdl
();
parent
()
->
add_scribe
(
std
::
move
(
sptr
));
return
doorman
::
new_connection
(
&
am
,
shdl
);
}
void
stop_reading
()
override
{
CAF_LOG_TRACE
(
""
);
acceptor_
.
stop
();
detach
(
&
acceptor_
.
backend
(),
false
);
}
void
launch
()
override
{
CAF_LOG_TRACE
(
""
);
acceptor_
.
start
(
this
);
}
std
::
string
addr
()
const
override
{
return
acceptor_
.
socket_handle
().
local_endpoint
().
address
().
to_string
();
}
uint16_t
port
()
const
override
{
return
acceptor_
.
socket_handle
().
local_endpoint
().
port
();
}
void
add_to_loop
()
override
{
acceptor_
.
activate
(
this
);
}
void
remove_from_loop
()
override
{
acceptor_
.
passivate
();
}
private:
network
::
asio_acceptor
<
asio_tcp_socket_acceptor
>
acceptor_
;
};
return
make_counted
<
impl
>
(
std
::
move
(
sock
),
*
this
);
}
doorman_ptr
asio_multiplexer
::
new_doorman
(
native_socket
fd
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
fd
));
asio_tcp_socket_acceptor
sock
{
service
()};
boost
::
system
::
error_code
ec
;
sock
.
assign
(
boost
::
asio
::
ip
::
tcp
::
v6
(),
fd
,
ec
);
if
(
ec
)
sock
.
assign
(
boost
::
asio
::
ip
::
tcp
::
v4
(),
fd
,
ec
);
if
(
ec
)
CAF_RAISE_ERROR
(
ec
.
message
());
return
new_doorman
(
std
::
move
(
sock
));
}
expected
<
doorman_ptr
>
asio_multiplexer
::
new_tcp_doorman
(
uint16_t
port
,
const
char
*
in
,
bool
rflag
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
port
)
<<
", addr = "
<<
(
in
?
in
:
"nullptr"
));
asio_tcp_socket_acceptor
fd
{
service
()};
auto
err
=
ip_bind
(
fd
,
port
,
in
,
rflag
);
if
(
err
)
return
err
;
return
new_doorman
(
std
::
move
(
fd
));
}
datagram_servant_ptr
asio_multiplexer
::
new_datagram_servant
(
native_socket
)
{
CAF_RAISE_ERROR
(
"UDP is not implemented for asio"
);
return
nullptr
;
}
datagram_servant_ptr
asio_multiplexer
::
new_datagram_servant_for_endpoint
(
native_socket
,
const
ip_endpoint
&
)
{
CAF_RAISE_ERROR
(
"UDP is not implemented for asio"
);
return
nullptr
;
}
expected
<
datagram_servant_ptr
>
asio_multiplexer
::
new_remote_udp_endpoint
(
const
std
::
string
&
,
uint16_t
)
{
CAF_RAISE_ERROR
(
"UDP is not implemented for asio"
);
return
sec
::
bad_function_call
;
}
expected
<
datagram_servant_ptr
>
asio_multiplexer
::
new_local_udp_endpoint
(
uint16_t
,
const
char
*
,
bool
)
{
CAF_RAISE_ERROR
(
"UDP is not implemented for asio"
);
return
sec
::
bad_function_call
;
}
void
asio_multiplexer
::
exec_later
(
resumable
*
rptr
)
{
auto
mt
=
system
().
config
().
scheduler_max_throughput
;
switch
(
rptr
->
subtype
())
{
case
resumable
:
:
io_actor
:
case
resumable
:
:
function_object
:
{
intrusive_ptr
<
resumable
>
ptr
{
rptr
,
false
};
service
().
post
([
=
]()
mutable
{
switch
(
ptr
->
resume
(
this
,
mt
))
{
case
resumable
:
:
resume_later
:
exec_later
(
ptr
.
release
());
break
;
case
resumable
:
:
done
:
case
resumable
:
:
awaiting_message
:
break
;
default:
;
// ignored
}
});
break
;
}
default:
system
().
scheduler
().
enqueue
(
rptr
);
}
}
asio_multiplexer
::
asio_multiplexer
(
actor_system
*
sys
)
:
multiplexer
(
sys
)
{
// nop
}
asio_multiplexer
::~
asio_multiplexer
()
{
//nop
}
multiplexer
::
supervisor_ptr
asio_multiplexer
::
make_supervisor
()
{
return
std
::
unique_ptr
<
asio_supervisor
>
(
new
asio_supervisor
(
service
()));
}
bool
asio_multiplexer
::
try_run_once
()
{
boost
::
system
::
error_code
ec
;
auto
num
=
service
().
poll
(
ec
);
if
(
ec
)
CAF_RAISE_ERROR
(
ec
.
message
());
return
num
>
0
;
}
void
asio_multiplexer
::
run_once
()
{
boost
::
system
::
error_code
ec
;
service
().
run_one
(
ec
);
if
(
ec
)
CAF_RAISE_ERROR
(
ec
.
message
());
}
void
asio_multiplexer
::
run
()
{
CAF_LOG_TRACE
(
"asio-based multiplexer"
);
boost
::
system
::
error_code
ec
;
service
().
run
(
ec
);
if
(
ec
)
CAF_RAISE_ERROR
(
ec
.
message
());
}
multiplexer_backend
*
asio_multiplexer
::
pimpl
()
{
return
&
service_
;
}
}
// namesapce network
}
// namespace io
}
// namespace caf
libcaf_io/src/middleman.cpp
View file @
38a661d8
...
...
@@ -57,11 +57,6 @@
#include "caf/detail/get_root_uuid.hpp"
#include "caf/detail/get_mac_addresses.hpp"
#ifdef CAF_USE_ASIO
#include "caf/io/network/asio_multiplexer.hpp"
#include "caf/io/network/asio_multiplexer_impl.hpp"
#endif // CAF_USE_ASIO
#ifdef CAF_WINDOWS
#include <io.h>
#include <fcntl.h>
...
...
@@ -93,10 +88,6 @@ actor_system::module* middleman::make(actor_system& sys, detail::type_list<>) {
auto
atm
=
get_or
(
sys
.
config
(),
"middleman.network-backend"
,
defaults
::
middleman
::
network_backend
);
switch
(
atom_uint
(
atm
))
{
# ifdef CAF_USE_ASIO
case
atom_uint
(
atom
(
"asio"
)):
return
new
mm_impl
<
network
::
asio_multiplexer
>
(
sys
);
# endif // CAF_USE_ASIO
case
atom_uint
(
atom
(
"testing"
)):
return
new
mm_impl
<
network
::
test_multiplexer
>
(
sys
);
default:
...
...
libcaf_openssl/caf/openssl/manager.hpp
View file @
38a661d8
...
...
@@ -63,8 +63,8 @@ public:
bool
authentication_enabled
();
/// Returns an OpenSSL manager using the default network backend.
/// @warning Creating an OpenSSL manager will fail when using
the ASIO
//
/ network backend or any other
custom implementation.
/// @warning Creating an OpenSSL manager will fail when using
//
a
custom implementation.
/// @throws `logic_error` if the middleman is not loaded or is not using the
/// default network backend.
static
actor_system
::
module
*
make
(
actor_system
&
,
detail
::
type_list
<>
);
...
...
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