Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
A
actor-incubator
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
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-incubator
Commits
1516ac5f
Unverified
Commit
1516ac5f
authored
Jul 24, 2019
by
Joseph Noir
Committed by
GitHub
Jul 24, 2019
Browse files
Options
Browse Files
Download
Plain Diff
Merge pull request #4
Add scaffold for new multiplexer API
parents
84e9b300
2d2f9ec9
Changes
22
Hide whitespace changes
Inline
Side-by-side
Showing
22 changed files
with
1461 additions
and
240 deletions
+1461
-240
CMakeLists.txt
CMakeLists.txt
+1
-1
libcaf_net/CMakeLists.txt
libcaf_net/CMakeLists.txt
+4
-0
libcaf_net/caf/detail/socket_sys_aliases.hpp
libcaf_net/caf/detail/socket_sys_aliases.hpp
+45
-0
libcaf_net/caf/net/fwd.hpp
libcaf_net/caf/net/fwd.hpp
+43
-0
libcaf_net/caf/net/multiplexer.hpp
libcaf_net/caf/net/multiplexer.hpp
+127
-0
libcaf_net/caf/net/network_socket.hpp
libcaf_net/caf/net/network_socket.hpp
+4
-32
libcaf_net/caf/net/operation.hpp
libcaf_net/caf/net/operation.hpp
+49
-0
libcaf_net/caf/net/pipe_socket.hpp
libcaf_net/caf/net/pipe_socket.hpp
+9
-3
libcaf_net/caf/net/pollset_updater.hpp
libcaf_net/caf/net/pollset_updater.hpp
+63
-0
libcaf_net/caf/net/socket.hpp
libcaf_net/caf/net/socket.hpp
+8
-14
libcaf_net/caf/net/socket_manager.hpp
libcaf_net/caf/net/socket_manager.hpp
+96
-0
libcaf_net/caf/net/stream_socket.hpp
libcaf_net/caf/net/stream_socket.hpp
+81
-0
libcaf_net/src/multiplexer.cpp
libcaf_net/src/multiplexer.cpp
+251
-0
libcaf_net/src/network_socket.cpp
libcaf_net/src/network_socket.cpp
+8
-68
libcaf_net/src/pipe_socket.cpp
libcaf_net/src/pipe_socket.cpp
+25
-96
libcaf_net/src/pollset_updater.cpp
libcaf_net/src/pollset_updater.cpp
+66
-0
libcaf_net/src/socket.cpp
libcaf_net/src/socket.cpp
+77
-22
libcaf_net/src/socket_manager.cpp
libcaf_net/src/socket_manager.cpp
+66
-0
libcaf_net/src/stream_socket.cpp
libcaf_net/src/stream_socket.cpp
+191
-0
libcaf_net/test/multiplexer.cpp
libcaf_net/test/multiplexer.cpp
+175
-0
libcaf_net/test/network_socket.cpp
libcaf_net/test/network_socket.cpp
+1
-4
libcaf_net/test/stream_socket.cpp
libcaf_net/test/stream_socket.cpp
+71
-0
No files found.
CMakeLists.txt
View file @
1516ac5f
...
@@ -390,7 +390,7 @@ if(NOT CAF_NO_UNIT_TESTS)
...
@@ -390,7 +390,7 @@ if(NOT CAF_NO_UNIT_TESTS)
macro
(
make_test suite
)
macro
(
make_test suite
)
string
(
REPLACE
" "
"_"
test_name
${
suite
}
)
string
(
REPLACE
" "
"_"
test_name
${
suite
}
)
set
(
caf_test
${
EXECUTABLE_OUTPUT_PATH
}
/incubator-test
)
set
(
caf_test
${
EXECUTABLE_OUTPUT_PATH
}
/incubator-test
)
add_test
(
${
test_name
}
${
caf_test
}
-r 300 -n -v 5 -s
"
${
suite
}
"
${
ARGN
}
)
add_test
(
${
test_name
}
${
caf_test
}
-r 300 -n -v 5 -s
"
^
${
suite
}
$
"
${
ARGN
}
)
endmacro
()
endmacro
()
list
(
LENGTH suites num_suites
)
list
(
LENGTH suites num_suites
)
message
(
STATUS
"Found
${
num_suites
}
test suites"
)
message
(
STATUS
"Found
${
num_suites
}
test suites"
)
...
...
libcaf_net/CMakeLists.txt
View file @
1516ac5f
...
@@ -8,9 +8,13 @@ file(GLOB_RECURSE LIBCAF_NET_HDRS "caf/*.hpp")
...
@@ -8,9 +8,13 @@ file(GLOB_RECURSE LIBCAF_NET_HDRS "caf/*.hpp")
# list cpp files excluding platform-dependent files
# list cpp files excluding platform-dependent files
set
(
LIBCAF_NET_SRCS
set
(
LIBCAF_NET_SRCS
src/host.cpp
src/host.cpp
src/multiplexer.cpp
src/network_socket.cpp
src/network_socket.cpp
src/pipe_socket.cpp
src/pipe_socket.cpp
src/pollset_updater.cpp
src/socket.cpp
src/socket.cpp
src/socket_manager.cpp
src/stream_socket.cpp
)
)
add_custom_target
(
libcaf_net
)
add_custom_target
(
libcaf_net
)
...
...
libcaf_net/caf/detail/socket_sys_aliases.hpp
0 → 100644
View file @
1516ac5f
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/config.hpp"
namespace
caf
{
namespace
net
{
#ifdef CAF_WINDOWS
using
setsockopt_ptr
=
const
char
*
;
using
getsockopt_ptr
=
char
*
;
using
socket_send_ptr
=
const
char
*
;
using
socket_recv_ptr
=
char
*
;
using
socket_size_type
=
int
;
#else // CAF_WINDOWS
using
setsockopt_ptr
=
const
void
*
;
using
getsockopt_ptr
=
void
*
;
using
socket_send_ptr
=
const
void
*
;
using
socket_recv_ptr
=
void
*
;
using
socket_size_type
=
unsigned
;
#endif // CAF_WINDOWS
}
// namespace net
}
// namespace caf
libcaf_net/caf/net/fwd.hpp
0 → 100644
View file @
1516ac5f
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <memory>
#include "caf/intrusive_ptr.hpp"
namespace
caf
{
namespace
net
{
class
multiplexer
;
class
socket_manager
;
struct
network_socket
;
struct
pipe_socket
;
struct
socket
;
struct
stream_socket
;
using
socket_manager_ptr
=
intrusive_ptr
<
socket_manager
>
;
using
multiplexer_ptr
=
std
::
shared_ptr
<
multiplexer
>
;
using
weak_multiplexer_ptr
=
std
::
weak_ptr
<
multiplexer
>
;
}
// namespace net
}
// namespace caf
libcaf_net/caf/net/multiplexer.hpp
0 → 100644
View file @
1516ac5f
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <memory>
#include <mutex>
#include <thread>
#include "caf/net/fwd.hpp"
#include "caf/net/operation.hpp"
#include "caf/net/pipe_socket.hpp"
#include "caf/net/socket.hpp"
#include "caf/ref_counted.hpp"
extern
"C"
{
struct
pollfd
;
}
// extern "C"
namespace
caf
{
namespace
net
{
/// Multiplexes any number of ::socket_manager objects with a ::socket.
class
multiplexer
:
public
std
::
enable_shared_from_this
<
multiplexer
>
{
public:
// -- member types -----------------------------------------------------------
using
pollfd_list
=
std
::
vector
<
pollfd
>
;
using
manager_list
=
std
::
vector
<
socket_manager_ptr
>
;
// -- constructors, destructors, and assignment operators --------------------
multiplexer
();
~
multiplexer
();
error
init
();
// -- properties -------------------------------------------------------------
/// Returns the number of currently active socket managers.
size_t
num_socket_managers
()
const
noexcept
;
/// Returns the index of `mgr` in the pollset or `-1`.
ptrdiff_t
index_of
(
const
socket_manager_ptr
&
mgr
);
// -- thread-safe signaling --------------------------------------------------
/// Causes the multiplexer to update its event bitmask for `mgr`.
/// @thread-safe
void
update
(
const
socket_manager_ptr
&
mgr
);
/// Closes the pipe for signaling updates to the multiplexer. After closing
/// the pipe, calls to `update` no longer have any effect.
/// @thread-safe
void
close_pipe
();
// -- control flow -----------------------------------------------------------
/// Polls I/O activity once and runs all socket event handlers that become
/// ready as a result.
bool
poll_once
(
bool
blocking
);
/// Polls until no socket event handler remains.
void
run
();
/// Processes all updates on the socket managers.
void
handle_updates
();
protected:
// -- utility functions ------------------------------------------------------
/// Handles an I/O event on given manager.
void
handle
(
const
socket_manager_ptr
&
mgr
,
int
mask
);
/// Adds a new socket manager to the pollset.
void
add
(
socket_manager_ptr
mgr
);
// -- member variables -------------------------------------------------------
/// Bookkeeping data for managed sockets.
pollfd_list
pollset_
;
/// Maps sockets to their owning managers by storing the managers in the same
/// order as their sockets appear in `pollset_`.
manager_list
managers_
;
/// Managers that updated their event mask and need updating.
manager_list
dirty_managers_
;
/// Stores the ID of the thread this multiplexer is running in. Set when
/// calling `init()`.
std
::
thread
::
id
tid_
;
/// Used for pushing updates to the multiplexer's thread.
pipe_socket
write_handle_
;
/// Guards `write_handle_`.
std
::
mutex
write_lock_
;
};
/// @relates multiplexer
using
multiplexer_ptr
=
std
::
shared_ptr
<
multiplexer
>
;
/// @relates multiplexer
using
weak_multiplexer_ptr
=
std
::
weak_ptr
<
multiplexer
>
;
}
// namespace net
}
// namespace caf
libcaf_net/caf/net/network_socket.hpp
View file @
1516ac5f
...
@@ -32,6 +32,7 @@
...
@@ -32,6 +32,7 @@
namespace
caf
{
namespace
caf
{
namespace
net
{
namespace
net
{
/// A bidirectional network communication endpoint.
struct
network_socket
:
abstract_socket
<
network_socket
>
{
struct
network_socket
:
abstract_socket
<
network_socket
>
{
using
super
=
abstract_socket
<
network_socket
>
;
using
super
=
abstract_socket
<
network_socket
>
;
...
@@ -42,18 +43,6 @@ struct network_socket : abstract_socket<network_socket> {
...
@@ -42,18 +43,6 @@ struct network_socket : abstract_socket<network_socket> {
}
}
};
};
/// Identifies the invalid socket.
/// @relates network_socket
constexpr
auto
invalid_network_socket
=
network_socket
{
invalid_socket_id
};
/// Enables or disables keepalive on `x`.
/// @relates network_socket
error
keepalive
(
network_socket
x
,
bool
new_value
);
/// Enables or disables Nagle's algorithm on `x`.
/// @relates network_socket
error
tcp_nodelay
(
network_socket
x
,
bool
new_value
);
/// Enables or disables `SIGPIPE` events from `x`.
/// Enables or disables `SIGPIPE` events from `x`.
/// @relates network_socket
/// @relates network_socket
error
allow_sigpipe
(
network_socket
x
,
bool
new_value
);
error
allow_sigpipe
(
network_socket
x
,
bool
new_value
);
...
@@ -63,13 +52,13 @@ error allow_sigpipe(network_socket x, bool new_value);
...
@@ -63,13 +52,13 @@ error allow_sigpipe(network_socket x, bool new_value);
error
allow_udp_connreset
(
network_socket
x
,
bool
new_value
);
error
allow_udp_connreset
(
network_socket
x
,
bool
new_value
);
/// Get the socket buffer size for `x`.
/// Get the socket buffer size for `x`.
/// @pre `x != invalid_
network_
socket`
/// @pre `x != invalid_socket`
/// @relates network_socket
/// @relates network_socket
expected
<
in
t
>
send_buffer_size
(
network_socket
x
);
expected
<
size_
t
>
send_buffer_size
(
network_socket
x
);
/// Set the socket buffer size for `x`.
/// Set the socket buffer size for `x`.
/// @relates network_socket
/// @relates network_socket
error
send_buffer_size
(
network_socket
x
,
int
new_value
);
error
send_buffer_size
(
network_socket
x
,
size_t
capacity
);
/// Returns the locally assigned port of `x`.
/// Returns the locally assigned port of `x`.
/// @relates network_socket
/// @relates network_socket
...
@@ -99,22 +88,5 @@ void shutdown_write(network_socket x);
...
@@ -99,22 +88,5 @@ void shutdown_write(network_socket x);
/// @relates network_socket
/// @relates network_socket
void
shutdown
(
network_socket
x
);
void
shutdown
(
network_socket
x
);
/// Transmits data from `x` to its peer.
/// @param x Connected endpoint.
/// @param buf Points to the message to send.
/// @param buf_size Specifies the size of the buffer in bytes.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates pipe_socket
variant
<
size_t
,
std
::
errc
>
write
(
network_socket
x
,
const
void
*
buf
,
size_t
buf_size
);
/// Receives data from `x`.
/// @param x Connected endpoint.
/// @param buf Points to destination buffer.
/// @param buf_size Specifies the maximum size of the buffer in bytes.
/// @returns The number of received bytes on success, otherwise an error code.
/// @relates pipe_socket
variant
<
size_t
,
std
::
errc
>
read
(
network_socket
x
,
void
*
buf
,
size_t
buf_size
);
}
// namespace net
}
// namespace net
}
// namespace caf
}
// namespace caf
libcaf_net/caf/net/operation.hpp
0 → 100644
View file @
1516ac5f
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
namespace
caf
{
namespace
net
{
/// Values for representing bitmask of I/O operations.
enum
class
operation
{
none
=
0x00
,
read
=
0x01
,
write
=
0x02
,
read_write
=
0x03
,
};
constexpr
operation
operator
|
(
operation
x
,
operation
y
)
{
return
static_cast
<
operation
>
(
static_cast
<
int
>
(
x
)
|
static_cast
<
int
>
(
y
));
}
constexpr
operation
operator
&
(
operation
x
,
operation
y
)
{
return
static_cast
<
operation
>
(
static_cast
<
int
>
(
x
)
&
static_cast
<
int
>
(
y
));
}
constexpr
operation
operator
^
(
operation
x
,
operation
y
)
{
return
static_cast
<
operation
>
(
static_cast
<
int
>
(
x
)
^
static_cast
<
int
>
(
y
));
}
constexpr
operation
operator
~
(
operation
x
)
{
return
static_cast
<
operation
>
(
~
static_cast
<
int
>
(
x
));
}
}
// namespace net
}
// namespace caf
libcaf_net/caf/net/pipe_socket.hpp
View file @
1516ac5f
...
@@ -30,6 +30,7 @@
...
@@ -30,6 +30,7 @@
namespace
caf
{
namespace
caf
{
namespace
net
{
namespace
net
{
/// A unidirectional communication endpoint for inter-process communication.
struct
pipe_socket
:
abstract_socket
<
pipe_socket
>
{
struct
pipe_socket
:
abstract_socket
<
pipe_socket
>
{
using
super
=
abstract_socket
<
pipe_socket
>
;
using
super
=
abstract_socket
<
pipe_socket
>
;
...
@@ -51,8 +52,7 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe();
...
@@ -51,8 +52,7 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe();
/// @param buf_size Specifies the size of the buffer in bytes.
/// @param buf_size Specifies the size of the buffer in bytes.
/// @returns The number of written bytes on success, otherwise an error code.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates pipe_socket
/// @relates pipe_socket
variant
<
size_t
,
std
::
errc
>
write
(
pipe_socket
x
,
const
void
*
buf
,
variant
<
size_t
,
sec
>
write
(
pipe_socket
x
,
const
void
*
buf
,
size_t
buf_size
);
size_t
buf_size
);
/// Receives data from `x`.
/// Receives data from `x`.
/// @param x Connected endpoint.
/// @param x Connected endpoint.
...
@@ -60,7 +60,13 @@ variant<size_t, std::errc> write(pipe_socket x, const void* buf,
...
@@ -60,7 +60,13 @@ variant<size_t, std::errc> write(pipe_socket x, const void* buf,
/// @param buf_size Specifies the maximum size of the buffer in bytes.
/// @param buf_size Specifies the maximum size of the buffer in bytes.
/// @returns The number of received bytes on success, otherwise an error code.
/// @returns The number of received bytes on success, otherwise an error code.
/// @relates pipe_socket
/// @relates pipe_socket
variant
<
size_t
,
std
::
errc
>
read
(
pipe_socket
x
,
void
*
buf
,
size_t
buf_size
);
variant
<
size_t
,
sec
>
read
(
pipe_socket
x
,
void
*
buf
,
size_t
buf_size
);
/// Converts the result from I/O operation on a ::pipe_socket to either an
/// error code or a non-zero positive integer.
/// @relates pipe_socket
variant
<
size_t
,
sec
>
check_pipe_socket_io_res
(
std
::
make_signed
<
size_t
>::
type
res
);
}
// namespace net
}
// namespace net
}
// namespace caf
}
// namespace caf
libcaf_net/caf/net/pollset_updater.hpp
0 → 100644
View file @
1516ac5f
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <array>
#include <cstdint>
#include "caf/net/pipe_socket.hpp"
#include "caf/net/socket_manager.hpp"
namespace
caf
{
namespace
net
{
class
pollset_updater
:
public
socket_manager
{
public:
// -- member types -----------------------------------------------------------
using
super
=
socket_manager
;
// -- constructors, destructors, and assignment operators --------------------
pollset_updater
(
pipe_socket
read_handle
,
multiplexer_ptr
parent
);
~
pollset_updater
()
override
;
// -- properties -------------------------------------------------------------
/// Returns the managed socket.
pipe_socket
handle
()
const
noexcept
{
return
socket_cast
<
pipe_socket
>
(
handle_
);
}
// -- interface functions ----------------------------------------------------
bool
handle_read_event
()
override
;
bool
handle_write_event
()
override
;
void
handle_error
(
sec
code
)
override
;
private:
std
::
array
<
char
,
sizeof
(
intptr_t
)
>
buf_
;
size_t
buf_size_
;
};
}
// namespace net
}
// namespace caf
libcaf_net/caf/net/socket.hpp
View file @
1516ac5f
...
@@ -19,6 +19,7 @@
...
@@ -19,6 +19,7 @@
#pragma once
#pragma once
#include <string>
#include <string>
#include <system_error>
#include <type_traits>
#include <type_traits>
#include "caf/config.hpp"
#include "caf/config.hpp"
...
@@ -40,27 +41,24 @@ struct socket : abstract_socket<socket> {
...
@@ -40,27 +41,24 @@ struct socket : abstract_socket<socket> {
/// Denotes the invalid socket.
/// Denotes the invalid socket.
constexpr
auto
invalid_socket
=
socket
{
invalid_socket_id
};
constexpr
auto
invalid_socket
=
socket
{
invalid_socket_id
};
/// Converts between different socket types.
template
<
class
To
,
class
From
>
To
socket_cast
(
From
x
)
{
return
To
{
x
.
id
};
}
/// Close socket `x`.
/// Close socket `x`.
/// @relates socket
/// @relates socket
void
close
(
socket
x
);
void
close
(
socket
x
);
/// Returns the last socket error in this thread as an integer.
/// Returns the last socket error in this thread as an integer.
/// @relates socket
/// @relates socket
int
last_socket_error
();
std
::
errc
last_socket_error
();
/// Returns the last socket error as human-readable string.
/// Returns the last socket error as human-readable string.
/// @relates socket
/// @relates socket
std
::
string
last_socket_error_as_string
();
std
::
string
last_socket_error_as_string
();
/// Returns a human-readable string for a given socket error.
/// @relates socket
std
::
string
socket_error_as_string
(
int
errcode
);
/// Returns whether `errcode` indicates that an operation would block or return
/// nothing at the moment and can be tried again at a later point.
/// @relates socket
bool
would_block_or_temporarily_unavailable
(
int
errcode
);
/// Sets x to be inherited by child processes if `new_value == true`
/// Sets x to be inherited by child processes if `new_value == true`
/// or not if `new_value == false`. Not implemented on Windows.
/// or not if `new_value == false`. Not implemented on Windows.
/// @relates socket
/// @relates socket
...
@@ -70,9 +68,5 @@ error child_process_inherit(socket x, bool new_value);
...
@@ -70,9 +68,5 @@ error child_process_inherit(socket x, bool new_value);
/// @relates socket
/// @relates socket
error
nonblocking
(
socket
x
,
bool
new_value
);
error
nonblocking
(
socket
x
,
bool
new_value
);
/// Convenience functions for checking the result of `recv` or `send`.
/// @relates socket
bool
is_error
(
std
::
make_signed
<
size_t
>::
type
res
,
bool
is_nonblock
);
}
// namespace net
}
// namespace net
}
// namespace caf
}
// namespace caf
libcaf_net/caf/net/socket_manager.hpp
0 → 100644
View file @
1516ac5f
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <atomic>
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/operation.hpp"
#include "caf/net/socket.hpp"
#include "caf/ref_counted.hpp"
namespace
caf
{
namespace
net
{
/// Manages the lifetime of a single socket and handles any I/O events on it.
class
socket_manager
:
public
ref_counted
{
public:
// -- constructors, destructors, and assignment operators --------------------
/// @pre `parent != nullptr`
/// @pre `handle != invalid_socket`
socket_manager
(
socket
handle
,
const
multiplexer_ptr
&
parent
);
virtual
~
socket_manager
();
socket_manager
(
const
socket_manager
&
)
=
delete
;
socket_manager
&
operator
=
(
const
socket_manager
&
)
=
delete
;
// -- properties -------------------------------------------------------------
/// Returns the managed socket.
socket
handle
()
const
noexcept
{
return
handle_
;
}
/// Returns registered operations (read, write, or both).
operation
mask
()
const
noexcept
;
/// Adds given flag(s) to the event mask and updates the parent on success.
/// @returns `false` if `mask() | flag == mask()`, `true` otherwise.
/// @pre `has_parent()`
/// @pre `flag != operation::none`
bool
mask_add
(
operation
flag
)
noexcept
;
/// Deletes given flag(s) from the event mask and updates the parent on
/// success.
/// @returns `false` if `mask() & ~flag == mask()`, `true` otherwise.
/// @pre `has_parent()`
/// @pre `flag != operation::none`
bool
mask_del
(
operation
flag
)
noexcept
;
// -- pure virtual member functions ------------------------------------------
/// Called whenever the socket received new data.
virtual
bool
handle_read_event
()
=
0
;
/// Called whenever the socket is allowed to send data.
virtual
bool
handle_write_event
()
=
0
;
/// Called when the remote side becomes unreachable due to an error.
/// @param reason The error code as reported by the operating system.
virtual
void
handle_error
(
sec
code
)
=
0
;
protected:
// -- member variables -------------------------------------------------------
socket
handle_
;
std
::
atomic
<
operation
>
mask_
;
weak_multiplexer_ptr
parent_
;
};
using
socket_manager_ptr
=
intrusive_ptr
<
socket_manager
>
;
}
// namespace net
}
// namespace caf
libcaf_net/caf/net/stream_socket.hpp
0 → 100644
View file @
1516ac5f
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/fwd.hpp"
#include "caf/net/network_socket.hpp"
namespace
caf
{
namespace
net
{
/// A connection-oriented network communication endpoint for bidirectional byte
/// streams.
struct
stream_socket
:
abstract_socket
<
stream_socket
>
{
using
super
=
abstract_socket
<
stream_socket
>
;
using
super
::
super
;
constexpr
operator
socket
()
const
noexcept
{
return
socket
{
id
};
}
constexpr
operator
network_socket
()
const
noexcept
{
return
network_socket
{
id
};
}
};
/// Creates two connected sockets to mimic network communication (usually for
/// testing purposes).
/// @relates stream_socket
expected
<
std
::
pair
<
stream_socket
,
stream_socket
>>
make_stream_socket_pair
();
/// Enables or disables keepalive on `x`.
/// @relates network_socket
error
keepalive
(
stream_socket
x
,
bool
new_value
);
/// Enables or disables Nagle's algorithm on `x`.
/// @relates stream_socket
error
nodelay
(
stream_socket
x
,
bool
new_value
);
/// Receives data from `x`.
/// @param x Connected endpoint.
/// @param buf Points to destination buffer.
/// @param buf_size Specifies the maximum size of the buffer in bytes.
/// @returns The number of received bytes on success, an error code otherwise.
/// @relates pipe_socket
/// @post either the result is a `sec` or a positive (non-zero) integer
variant
<
size_t
,
sec
>
read
(
stream_socket
x
,
void
*
buf
,
size_t
buf_size
);
/// Transmits data from `x` to its peer.
/// @param x Connected endpoint.
/// @param buf Points to the message to send.
/// @param buf_size Specifies the size of the buffer in bytes.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates pipe_socket
/// @post either the result is a `sec` or a positive (non-zero) integer
variant
<
size_t
,
sec
>
write
(
stream_socket
x
,
const
void
*
buf
,
size_t
buf_size
);
/// Converts the result from I/O operation on a ::stream_socket to either an
/// error code or a non-zero positive integer.
/// @relates stream_socket
variant
<
size_t
,
sec
>
check_stream_socket_io_res
(
std
::
make_signed
<
size_t
>::
type
res
);
}
// namespace net
}
// namespace caf
libcaf_net/src/multiplexer.cpp
0 → 100644
View file @
1516ac5f
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#include "caf/net/multiplexer.hpp"
#include <algorithm>
#include "caf/config.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/logger.hpp"
#include "caf/net/operation.hpp"
#include "caf/net/pollset_updater.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/sec.hpp"
#include "caf/variant.hpp"
#ifndef CAF_WINDOWS
# include <poll.h>
#else
# include "caf/detail/socket_sys_includes.hpp"
#endif // CAF_WINDOWS
namespace
caf
{
namespace
net
{
#ifndef POLLRDHUP
# define POLLRDHUP POLLHUP
#endif
#ifndef POLLPRI
# define POLLPRI POLLIN
#endif
namespace
{
#ifdef CAF_WINDOWS
// From the MSDN: If the POLLPRI flag is set on a socket for the Microsoft
// Winsock provider, the WSAPoll function will fail.
const
short
input_mask
=
POLLIN
;
#else
const
short
input_mask
=
POLLIN
|
POLLPRI
;
#endif
const
short
error_mask
=
POLLRDHUP
|
POLLERR
|
POLLHUP
|
POLLNVAL
;
const
short
output_mask
=
POLLOUT
;
short
to_bitmask
(
operation
op
)
{
switch
(
op
)
{
case
operation
:
:
read
:
return
input_mask
;
case
operation
:
:
write
:
return
output_mask
;
case
operation
:
:
read_write
:
return
input_mask
|
output_mask
;
default:
return
0
;
}
}
}
// namespace
multiplexer
::
multiplexer
()
{
// nop
}
multiplexer
::~
multiplexer
()
{
// nop
}
error
multiplexer
::
init
()
{
auto
pipe_handles
=
make_pipe
();
if
(
!
pipe_handles
)
return
std
::
move
(
pipe_handles
.
error
());
tid_
=
std
::
this_thread
::
get_id
();
add
(
make_counted
<
pollset_updater
>
(
pipe_handles
->
first
,
shared_from_this
()));
write_handle_
=
pipe_handles
->
second
;
return
none
;
}
size_t
multiplexer
::
num_socket_managers
()
const
noexcept
{
return
managers_
.
size
();
}
ptrdiff_t
multiplexer
::
index_of
(
const
socket_manager_ptr
&
mgr
)
{
auto
first
=
managers_
.
begin
();
auto
last
=
managers_
.
end
();
auto
i
=
std
::
find
(
first
,
last
,
mgr
);
return
i
==
last
?
-
1
:
std
::
distance
(
first
,
i
);
}
void
multiplexer
::
update
(
const
socket_manager_ptr
&
mgr
)
{
if
(
std
::
this_thread
::
get_id
()
==
tid_
)
{
auto
i
=
std
::
find
(
dirty_managers_
.
begin
(),
dirty_managers_
.
end
(),
mgr
);
if
(
i
==
dirty_managers_
.
end
())
dirty_managers_
.
emplace_back
(
mgr
);
}
else
{
mgr
->
ref
();
auto
value
=
reinterpret_cast
<
intptr_t
>
(
mgr
.
get
());
variant
<
size_t
,
sec
>
res
;
{
// Lifetime scope of guard.
std
::
lock_guard
<
std
::
mutex
>
guard
{
write_lock_
};
if
(
write_handle_
!=
invalid_socket
)
res
=
write
(
write_handle_
,
&
value
,
sizeof
(
intptr_t
));
else
res
=
sec
::
socket_invalid
;
}
if
(
holds_alternative
<
sec
>
(
res
))
mgr
->
deref
();
}
}
void
multiplexer
::
close_pipe
()
{
std
::
lock_guard
<
std
::
mutex
>
guard
{
write_lock_
};
if
(
write_handle_
!=
invalid_socket
)
{
close
(
write_handle_
);
write_handle_
=
pipe_socket
{};
}
}
bool
multiplexer
::
poll_once
(
bool
blocking
)
{
if
(
pollset_
.
empty
())
return
false
;
// We'll call poll() until poll() succeeds or fails.
for
(;;)
{
int
presult
;
#ifdef CAF_WINDOWS
presult
=
::
WSAPoll
(
pollset_
.
data
(),
static_cast
<
ULONG
>
(
pollset_
.
size
()),
blocking
?
-
1
:
0
);
#else
presult
=
::
poll
(
pollset_
.
data
(),
static_cast
<
nfds_t
>
(
pollset_
.
size
()),
blocking
?
-
1
:
0
);
#endif
if
(
presult
<
0
)
{
auto
code
=
last_socket_error
();
switch
(
code
)
{
case
std
:
:
errc
::
interrupted
:
{
// A signal was caught. Simply try again.
CAF_LOG_DEBUG
(
"received errc::interrupted, try again"
);
break
;
}
case
std
:
:
errc
::
not_enough_memory
:
{
CAF_LOG_ERROR
(
"poll() failed due to insufficient memory"
);
// There's not much we can do other than try again in hope someone
// else releases memory.
break
;
}
default:
{
// Must not happen.
auto
int_code
=
static_cast
<
int
>
(
code
);
auto
msg
=
std
::
generic_category
().
message
(
int_code
);
string_view
prefix
=
"poll() failed: "
;
msg
.
insert
(
msg
.
begin
(),
prefix
.
begin
(),
prefix
.
end
());
CAF_CRITICAL
(
msg
.
c_str
());
}
}
// Rinse and repeat.
continue
;
}
CAF_LOG_DEBUG
(
"poll() on"
<<
pollset_
.
size
()
<<
"sockets reported"
<<
presult
<<
"event(s)"
);
// No activity.
if
(
presult
==
0
)
return
false
;
// Scan pollset for events.
CAF_LOG_DEBUG
(
"scan pollset for socket events"
);
for
(
size_t
i
=
0
;
i
<
pollset_
.
size
()
&&
presult
>
0
;
++
i
)
{
auto
&
x
=
pollset_
[
i
];
if
(
x
.
revents
!=
0
)
{
handle
(
managers_
[
i
],
x
.
revents
);
--
presult
;
}
}
handle_updates
();
return
true
;
}
}
void
multiplexer
::
run
()
{
while
(
!
pollset_
.
empty
())
poll_once
(
true
);
}
void
multiplexer
::
handle_updates
()
{
for
(
auto
mgr
:
dirty_managers_
)
{
auto
index
=
index_of
(
mgr
.
get
());
if
(
index
==
-
1
)
{
add
(
std
::
move
(
mgr
));
}
else
{
// Update or remove an existing manager in the pollset.
if
(
mgr
->
mask
()
==
operation
::
none
)
{
pollset_
.
erase
(
pollset_
.
begin
()
+
index
);
managers_
.
erase
(
managers_
.
begin
()
+
index
);
}
else
{
pollset_
[
index
].
events
=
to_bitmask
(
mgr
->
mask
());
}
}
}
dirty_managers_
.
clear
();
}
void
multiplexer
::
handle
(
const
socket_manager_ptr
&
mgr
,
int
mask
)
{
CAF_LOG_TRACE
(
CAF_ARG2
(
"socket"
,
mgr
->
handle
())
<<
CAF_ARG
(
mask
));
CAF_ASSERT
(
mgr
!=
nullptr
);
bool
checkerror
=
true
;
if
((
mask
&
input_mask
)
!=
0
)
{
checkerror
=
false
;
if
(
!
mgr
->
handle_read_event
())
mgr
->
mask_del
(
operation
::
read
);
}
if
((
mask
&
output_mask
)
!=
0
)
{
checkerror
=
false
;
if
(
!
mgr
->
handle_write_event
())
mgr
->
mask_del
(
operation
::
write
);
}
if
(
checkerror
&&
((
mask
&
error_mask
)
!=
0
))
{
if
(
mask
&
POLLNVAL
)
mgr
->
handle_error
(
sec
::
socket_invalid
);
else
if
(
mask
&
POLLHUP
)
mgr
->
handle_error
(
sec
::
socket_disconnected
);
else
mgr
->
handle_error
(
sec
::
socket_operation_failed
);
mgr
->
mask_del
(
operation
::
read_write
);
}
}
void
multiplexer
::
add
(
socket_manager_ptr
mgr
)
{
pollfd
new_entry
{
socket_cast
<
socket_id
>
(
mgr
->
handle
()),
to_bitmask
(
mgr
->
mask
()),
0
};
pollset_
.
emplace_back
(
new_entry
);
managers_
.
emplace_back
(
std
::
move
(
mgr
));
}
}
// namespace net
}
// namespace caf
libcaf_net/src/network_socket.cpp
View file @
1516ac5f
...
@@ -22,6 +22,7 @@
...
@@ -22,6 +22,7 @@
#include "caf/config.hpp"
#include "caf/config.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_aliases.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/error.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/expected.hpp"
...
@@ -59,33 +60,12 @@ namespace net {
...
@@ -59,33 +60,12 @@ namespace net {
#if defined(CAF_MACOS) || defined(CAF_IOS) || defined(CAF_BSD)
#if defined(CAF_MACOS) || defined(CAF_IOS) || defined(CAF_BSD)
# define CAF_HAS_NOSIGPIPE_SOCKET_FLAG
# define CAF_HAS_NOSIGPIPE_SOCKET_FLAG
const
int
no_sigpipe_io_flag
=
0
;
#elif defined(CAF_WINDOWS)
const
int
no_sigpipe_io_flag
=
0
;
#else
// Use flags to recv/send on Linux/Android but no socket option.
const
int
no_sigpipe_io_flag
=
MSG_NOSIGNAL
;
#endif
#endif
#ifdef CAF_WINDOWS
#ifdef CAF_WINDOWS
using
setsockopt_ptr
=
const
char
*
;
using
getsockopt_ptr
=
char
*
;
using
socket_send_ptr
=
const
char
*
;
using
socket_recv_ptr
=
char
*
;
using
socket_size_type
=
int
;
error
keepalive
(
network_socket
x
,
bool
new_value
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
x
)
<<
CAF_ARG
(
new_value
));
char
value
=
new_value
?
1
:
0
;
CAF_NET_SYSCALL
(
"setsockopt"
,
res
,
!=
,
0
,
setsockopt
(
x
.
id
,
SOL_SOCKET
,
SO_KEEPALIVE
,
&
value
,
static_cast
<
int
>
(
sizeof
(
value
))));
return
none
;
}
error
allow_sigpipe
(
network_socket
x
,
bool
)
{
error
allow_sigpipe
(
network_socket
x
,
bool
)
{
if
(
x
==
invalid_
network_
socket
)
if
(
x
==
invalid_socket
)
return
make_error
(
sec
::
network_syscall_failed
,
"setsockopt"
,
return
make_error
(
sec
::
network_syscall_failed
,
"setsockopt"
,
"invalid socket"
);
"invalid socket"
);
return
none
;
return
none
;
...
@@ -101,20 +81,6 @@ error allow_udp_connreset(network_socket x, bool new_value) {
...
@@ -101,20 +81,6 @@ error allow_udp_connreset(network_socket x, bool new_value) {
}
}
#else // CAF_WINDOWS
#else // CAF_WINDOWS
using
setsockopt_ptr
=
const
void
*
;
using
getsockopt_ptr
=
void
*
;
using
socket_send_ptr
=
const
void
*
;
using
socket_recv_ptr
=
void
*
;
using
socket_size_type
=
unsigned
;
error
keepalive
(
network_socket
x
,
bool
new_value
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
x
)
<<
CAF_ARG
(
new_value
));
int
value
=
new_value
?
1
:
0
;
CAF_NET_SYSCALL
(
"setsockopt"
,
res
,
!=
,
0
,
setsockopt
(
x
.
id
,
SOL_SOCKET
,
SO_KEEPALIVE
,
&
value
,
static_cast
<
unsigned
>
(
sizeof
(
value
))));
return
none
;
}
error
allow_sigpipe
(
network_socket
x
,
bool
new_value
)
{
error
allow_sigpipe
(
network_socket
x
,
bool
new_value
)
{
# ifdef CAF_HAS_NOSIGPIPE_SOCKET_FLAG
# ifdef CAF_HAS_NOSIGPIPE_SOCKET_FLAG
...
@@ -123,7 +89,7 @@ error allow_sigpipe(network_socket x, bool new_value) {
...
@@ -123,7 +89,7 @@ error allow_sigpipe(network_socket x, bool new_value) {
setsockopt
(
x
.
id
,
SOL_SOCKET
,
SO_NOSIGPIPE
,
&
value
,
setsockopt
(
x
.
id
,
SOL_SOCKET
,
SO_NOSIGPIPE
,
&
value
,
static_cast
<
unsigned
>
(
sizeof
(
value
))));
static_cast
<
unsigned
>
(
sizeof
(
value
))));
# else // CAF_HAS_NO_SIGPIPE_SOCKET_FLAG
# else // CAF_HAS_NO_SIGPIPE_SOCKET_FLAG
if
(
x
==
invalid_
network_
socket
)
if
(
x
==
invalid_socket
)
return
make_error
(
sec
::
network_syscall_failed
,
"setsockopt"
,
return
make_error
(
sec
::
network_syscall_failed
,
"setsockopt"
,
"invalid socket"
);
"invalid socket"
);
# endif // CAF_HAS_NO_SIGPIPE_SOCKET_FLAG
# endif // CAF_HAS_NO_SIGPIPE_SOCKET_FLAG
...
@@ -132,7 +98,7 @@ error allow_sigpipe(network_socket x, bool new_value) {
...
@@ -132,7 +98,7 @@ error allow_sigpipe(network_socket x, bool new_value) {
error
allow_udp_connreset
(
network_socket
x
,
bool
)
{
error
allow_udp_connreset
(
network_socket
x
,
bool
)
{
// SIO_UDP_CONNRESET only exists on Windows
// SIO_UDP_CONNRESET only exists on Windows
if
(
x
==
invalid_
network_
socket
)
if
(
x
==
invalid_socket
)
return
make_error
(
sec
::
network_syscall_failed
,
"WSAIoctl"
,
return
make_error
(
sec
::
network_syscall_failed
,
"WSAIoctl"
,
"invalid socket"
);
"invalid socket"
);
return
none
;
return
none
;
...
@@ -140,17 +106,18 @@ error allow_udp_connreset(network_socket x, bool) {
...
@@ -140,17 +106,18 @@ error allow_udp_connreset(network_socket x, bool) {
#endif // CAF_WINDOWS
#endif // CAF_WINDOWS
expected
<
in
t
>
send_buffer_size
(
network_socket
x
)
{
expected
<
size_
t
>
send_buffer_size
(
network_socket
x
)
{
int
size
=
0
;
int
size
=
0
;
socket_size_type
ret_size
=
sizeof
(
size
);
socket_size_type
ret_size
=
sizeof
(
size
);
CAF_NET_SYSCALL
(
"getsockopt"
,
res
,
!=
,
0
,
CAF_NET_SYSCALL
(
"getsockopt"
,
res
,
!=
,
0
,
getsockopt
(
x
.
id
,
SOL_SOCKET
,
SO_SNDBUF
,
getsockopt
(
x
.
id
,
SOL_SOCKET
,
SO_SNDBUF
,
reinterpret_cast
<
getsockopt_ptr
>
(
&
size
),
reinterpret_cast
<
getsockopt_ptr
>
(
&
size
),
&
ret_size
));
&
ret_size
));
return
s
ize
;
return
s
tatic_cast
<
size_t
>
(
size
)
;
}
}
error
send_buffer_size
(
network_socket
x
,
int
new_value
)
{
error
send_buffer_size
(
network_socket
x
,
size_t
capacity
)
{
auto
new_value
=
static_cast
<
int
>
(
capacity
);
CAF_NET_SYSCALL
(
"setsockopt"
,
res
,
!=
,
0
,
CAF_NET_SYSCALL
(
"setsockopt"
,
res
,
!=
,
0
,
setsockopt
(
x
.
id
,
SOL_SOCKET
,
SO_SNDBUF
,
setsockopt
(
x
.
id
,
SOL_SOCKET
,
SO_SNDBUF
,
reinterpret_cast
<
setsockopt_ptr
>
(
&
new_value
),
reinterpret_cast
<
setsockopt_ptr
>
(
&
new_value
),
...
@@ -158,16 +125,6 @@ error send_buffer_size(network_socket x, int new_value) {
...
@@ -158,16 +125,6 @@ error send_buffer_size(network_socket x, int new_value) {
return
none
;
return
none
;
}
}
error
tcp_nodelay
(
network_socket
x
,
bool
new_value
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
x
)
<<
CAF_ARG
(
new_value
));
int
flag
=
new_value
?
1
:
0
;
CAF_NET_SYSCALL
(
"setsockopt"
,
res
,
!=
,
0
,
setsockopt
(
x
.
id
,
IPPROTO_TCP
,
TCP_NODELAY
,
reinterpret_cast
<
setsockopt_ptr
>
(
&
flag
),
static_cast
<
socket_size_type
>
(
sizeof
(
flag
))));
return
none
;
}
expected
<
std
::
string
>
local_addr
(
network_socket
x
)
{
expected
<
std
::
string
>
local_addr
(
network_socket
x
)
{
sockaddr_storage
st
;
sockaddr_storage
st
;
socket_size_type
st_len
=
sizeof
(
st
);
socket_size_type
st_len
=
sizeof
(
st
);
...
@@ -236,22 +193,5 @@ void shutdown(network_socket x) {
...
@@ -236,22 +193,5 @@ void shutdown(network_socket x) {
::
shutdown
(
x
.
id
,
2
);
::
shutdown
(
x
.
id
,
2
);
}
}
variant
<
size_t
,
std
::
errc
>
write
(
network_socket
x
,
const
void
*
buf
,
size_t
buf_size
)
{
auto
res
=
::
send
(
x
.
id
,
reinterpret_cast
<
socket_send_ptr
>
(
buf
),
buf_size
,
no_sigpipe_io_flag
);
if
(
res
<
0
)
return
static_cast
<
std
::
errc
>
(
errno
);
return
static_cast
<
size_t
>
(
res
);
}
variant
<
size_t
,
std
::
errc
>
read
(
network_socket
x
,
void
*
buf
,
size_t
buf_size
)
{
auto
res
=
::
recv
(
x
.
id
,
reinterpret_cast
<
socket_recv_ptr
>
(
buf
),
buf_size
,
no_sigpipe_io_flag
);
if
(
res
<
0
)
return
static_cast
<
std
::
errc
>
(
errno
);
return
static_cast
<
size_t
>
(
res
);
}
}
// namespace net
}
// namespace net
}
// namespace caf
}
// namespace caf
libcaf_net/src/pipe_socket.cpp
View file @
1516ac5f
...
@@ -20,13 +20,14 @@
...
@@ -20,13 +20,14 @@
#include <cstdio>
#include <cstdio>
#include <cstdlib>
#include <cstdlib>
#include <utility>
#include "caf/config.hpp"
#include "caf/config.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/error.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/expected.hpp"
#include "caf/net/
network
_socket.hpp"
#include "caf/net/
stream
_socket.hpp"
#include "caf/sec.hpp"
#include "caf/sec.hpp"
#include "caf/variant.hpp"
#include "caf/variant.hpp"
...
@@ -35,99 +36,27 @@ namespace net {
...
@@ -35,99 +36,27 @@ namespace net {
#ifdef CAF_WINDOWS
#ifdef CAF_WINDOWS
/**************************************************************************\
* Based on work of others; *
* original header: *
* *
* Copyright 2007, 2010 by Nathan C. Myers <ncm@cantrip.org> *
* Redistribution and use in source and binary forms, with or without *
* modification, are permitted provided that the following conditions *
* are met: *
* *
* Redistributions of source code must retain the above copyright notice, *
* this list of conditions and the following disclaimer. *
* *
* Redistributions in binary form must reproduce the above copyright *
* notice, this list of conditions and the following disclaimer in the *
* documentation and/or other materials provided with the distribution. *
* *
* The name of the author must not be used to endorse or promote products *
* derived from this software without specific prior written permission. *
* *
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS *
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT *
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR *
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT *
* HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, *
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT *
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, *
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY *
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT *
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE *
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. *
\**************************************************************************/
expected
<
std
::
pair
<
pipe_socket
,
pipe_socket
>>
make_pipe
()
{
expected
<
std
::
pair
<
pipe_socket
,
pipe_socket
>>
make_pipe
()
{
auto
addrlen
=
static_cast
<
int
>
(
sizeof
(
sockaddr_in
));
// Windows has no support for unidirectional pipes. Emulate pipes by using a
socket_id
socks
[
2
]
=
{
invalid_socket_id
,
invalid_socket_id
};
// pair of regular TCP sockets with read/write channels closed appropriately.
CAF_NET_SYSCALL
(
"socket"
,
listener
,
==
,
invalid_socket_id
,
if
(
auto
result
=
make_stream_socket_pair
())
{
::
socket
(
AF_INET
,
SOCK_STREAM
,
IPPROTO_TCP
));
shutdown_write
(
result
->
first
);
union
{
shutdown_read
(
result
->
second
);
sockaddr_in
inaddr
;
return
std
::
make_pair
(
socket_cast
<
pipe_socket
>
(
result
->
first
),
sockaddr
addr
;
socket_cast
<
pipe_socket
>
(
result
->
second
));
}
a
;
}
else
{
memset
(
&
a
,
0
,
sizeof
(
a
));
return
std
::
move
(
result
.
error
());
a
.
inaddr
.
sin_family
=
AF_INET
;
}
a
.
inaddr
.
sin_addr
.
s_addr
=
htonl
(
INADDR_LOOPBACK
);
a
.
inaddr
.
sin_port
=
0
;
// makes sure all sockets are closed in case of an error
auto
guard
=
detail
::
make_scope_guard
([
&
]
{
auto
e
=
WSAGetLastError
();
close
(
socket
{
listener
});
close
(
socket
{
socks
[
0
]});
close
(
socket
{
socks
[
1
]});
WSASetLastError
(
e
);
});
// bind listener to a local port
int
reuse
=
1
;
CAF_NET_SYSCALL
(
"setsockopt"
,
tmp1
,
!=
,
0
,
setsockopt
(
listener
,
SOL_SOCKET
,
SO_REUSEADDR
,
reinterpret_cast
<
char
*>
(
&
reuse
),
static_cast
<
int
>
(
sizeof
(
reuse
))));
CAF_NET_SYSCALL
(
"bind"
,
tmp2
,
!=
,
0
,
bind
(
listener
,
&
a
.
addr
,
static_cast
<
int
>
(
sizeof
(
a
.
inaddr
))));
// Read the port in use: win32 getsockname may only set the port number
// (http://msdn.microsoft.com/library/ms738543.aspx).
memset
(
&
a
,
0
,
sizeof
(
a
));
CAF_NET_SYSCALL
(
"getsockname"
,
tmp3
,
!=
,
0
,
getsockname
(
listener
,
&
a
.
addr
,
&
addrlen
));
a
.
inaddr
.
sin_addr
.
s_addr
=
htonl
(
INADDR_LOOPBACK
);
a
.
inaddr
.
sin_family
=
AF_INET
;
// set listener to listen mode
CAF_NET_SYSCALL
(
"listen"
,
tmp5
,
!=
,
0
,
listen
(
listener
,
1
));
// create read-only end of the pipe
DWORD
flags
=
0
;
CAF_NET_SYSCALL
(
"WSASocketW"
,
read_fd
,
==
,
invalid_socket_id
,
WSASocketW
(
AF_INET
,
SOCK_STREAM
,
0
,
nullptr
,
0
,
flags
));
CAF_NET_SYSCALL
(
"connect"
,
tmp6
,
!=
,
0
,
connect
(
read_fd
,
&
a
.
addr
,
static_cast
<
int
>
(
sizeof
(
a
.
inaddr
))));
// get write-only end of the pipe
CAF_NET_SYSCALL
(
"accept"
,
write_fd
,
==
,
invalid_socket_id
,
accept
(
listener
,
nullptr
,
nullptr
));
close
(
socket
{
listener
});
guard
.
disable
();
return
std
::
make_pair
(
read_fd
,
write_fd
);
}
}
variant
<
size_t
,
std
::
errc
>
write
(
pipe_socket
x
,
const
void
*
buf
,
variant
<
size_t
,
sec
>
write
(
pipe_socket
x
,
const
void
*
buf
,
size_t
buf_size
)
{
size_t
buf_size
)
{
// On Windows, a pipe consists of two stream sockets.
// On Windows, a pipe consists of two stream sockets.
return
write
(
network_socket
{
x
.
id
}
,
buf
,
buf_size
);
return
write
(
socket_cast
<
stream_socket
>
(
x
)
,
buf
,
buf_size
);
}
}
variant
<
size_t
,
s
td
::
err
c
>
read
(
pipe_socket
x
,
void
*
buf
,
size_t
buf_size
)
{
variant
<
size_t
,
s
e
c
>
read
(
pipe_socket
x
,
void
*
buf
,
size_t
buf_size
)
{
// On Windows, a pipe consists of two stream sockets.
// On Windows, a pipe consists of two stream sockets.
return
read
(
network_socket
{
x
.
id
}
,
buf
,
buf_size
);
return
read
(
socket_cast
<
stream_socket
>
(
x
)
,
buf
,
buf_size
);
}
}
#else // CAF_WINDOWS
#else // CAF_WINDOWS
...
@@ -151,22 +80,22 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() {
...
@@ -151,22 +80,22 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() {
return
std
::
make_pair
(
pipe_socket
{
pipefds
[
0
]},
pipe_socket
{
pipefds
[
1
]});
return
std
::
make_pair
(
pipe_socket
{
pipefds
[
0
]},
pipe_socket
{
pipefds
[
1
]});
}
}
variant
<
size_t
,
std
::
errc
>
write
(
pipe_socket
x
,
const
void
*
buf
,
variant
<
size_t
,
sec
>
write
(
pipe_socket
x
,
const
void
*
buf
,
size_t
buf_size
)
{
size_t
buf_size
)
{
auto
res
=
::
write
(
x
.
id
,
buf
,
buf_size
);
auto
res
=
::
write
(
x
.
id
,
buf
,
buf_size
);
if
(
res
<
0
)
return
check_pipe_socket_io_res
(
res
);
return
static_cast
<
std
::
errc
>
(
errno
);
return
static_cast
<
size_t
>
(
res
);
}
}
variant
<
size_t
,
s
td
::
err
c
>
read
(
pipe_socket
x
,
void
*
buf
,
size_t
buf_size
)
{
variant
<
size_t
,
s
e
c
>
read
(
pipe_socket
x
,
void
*
buf
,
size_t
buf_size
)
{
auto
res
=
::
read
(
x
.
id
,
buf
,
buf_size
);
auto
res
=
::
read
(
x
.
id
,
buf
,
buf_size
);
if
(
res
<
0
)
return
check_pipe_socket_io_res
(
res
);
return
static_cast
<
std
::
errc
>
(
errno
);
return
static_cast
<
size_t
>
(
res
);
}
}
#endif // CAF_WINDOWS
#endif // CAF_WINDOWS
variant
<
size_t
,
sec
>
check_pipe_socket_io_res
(
std
::
make_signed
<
size_t
>::
type
res
)
{
return
check_stream_socket_io_res
(
res
);
}
}
// namespace net
}
// namespace net
}
// namespace caf
}
// namespace caf
libcaf_net/src/pollset_updater.cpp
0 → 100644
View file @
1516ac5f
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#include "caf/net/pollset_updater.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/sec.hpp"
#include "caf/variant.hpp"
namespace
caf
{
namespace
net
{
pollset_updater
::
pollset_updater
(
pipe_socket
read_handle
,
multiplexer_ptr
parent
)
:
super
(
read_handle
,
std
::
move
(
parent
)),
buf_size_
(
0
)
{
mask_
=
operation
::
read
;
nonblocking
(
read_handle
,
true
);
}
pollset_updater
::~
pollset_updater
()
{
// nop
}
bool
pollset_updater
::
handle_read_event
()
{
for
(;;)
{
auto
res
=
read
(
handle
(),
buf_
.
data
(),
buf_
.
size
()
-
buf_size_
);
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
res
))
{
buf_size_
+=
*
num_bytes
;
if
(
buf_
.
size
()
==
buf_size_
)
{
buf_size_
=
0
;
auto
value
=
*
reinterpret_cast
<
intptr_t
*>
(
buf_
.
data
());
socket_manager_ptr
mgr
{
reinterpret_cast
<
socket_manager
*>
(
value
),
false
};
if
(
auto
ptr
=
parent_
.
lock
())
ptr
->
update
(
mgr
);
}
}
else
{
return
get
<
sec
>
(
res
)
==
sec
::
unavailable_or_would_block
;
}
}
}
bool
pollset_updater
::
handle_write_event
()
{
return
false
;
}
void
pollset_updater
::
handle_error
(
sec
)
{
// nop
}
}
// namespace net
}
// namespace caf
libcaf_net/src/socket.cpp
View file @
1516ac5f
...
@@ -18,10 +18,14 @@
...
@@ -18,10 +18,14 @@
#include "caf/net/socket.hpp"
#include "caf/net/socket.hpp"
#include <system_error>
#include "caf/config.hpp"
#include "caf/config.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/logger.hpp"
#include "caf/logger.hpp"
#include "caf/sec.hpp"
#include "caf/variant.hpp"
namespace
caf
{
namespace
caf
{
namespace
net
{
namespace
net
{
...
@@ -32,15 +36,80 @@ void close(socket fd) {
...
@@ -32,15 +36,80 @@ void close(socket fd) {
closesocket
(
fd
.
id
);
closesocket
(
fd
.
id
);
}
}
int
last_socket_error
()
{
# define ERRC_CASE(wsa_code, stl_code) \
return
WSAGetLastError
();
case wsa_code: \
return errc::stl_code
std
::
errc
last_socket_error
()
{
// Unfortunately, MS does not define errc consistent with the WSA error
// codes. Hence, we cannot simply use static_cast<> but have to perform a
// switch-case.
using
std
::
errc
;
int
wsa_code
=
WSAGetLastError
();
switch
(
wsa_code
)
{
ERRC_CASE
(
WSA_INVALID_HANDLE
,
invalid_argument
);
ERRC_CASE
(
WSA_NOT_ENOUGH_MEMORY
,
not_enough_memory
);
ERRC_CASE
(
WSA_INVALID_PARAMETER
,
invalid_argument
);
ERRC_CASE
(
WSAEINTR
,
interrupted
);
ERRC_CASE
(
WSAEBADF
,
bad_file_descriptor
);
ERRC_CASE
(
WSAEACCES
,
permission_denied
);
ERRC_CASE
(
WSAEFAULT
,
bad_address
);
ERRC_CASE
(
WSAEINVAL
,
invalid_argument
);
ERRC_CASE
(
WSAEMFILE
,
too_many_files_open
);
ERRC_CASE
(
WSAEWOULDBLOCK
,
operation_would_block
);
ERRC_CASE
(
WSAEINPROGRESS
,
operation_in_progress
);
ERRC_CASE
(
WSAEALREADY
,
connection_already_in_progress
);
ERRC_CASE
(
WSAENOTSOCK
,
not_a_socket
);
ERRC_CASE
(
WSAEDESTADDRREQ
,
destination_address_required
);
ERRC_CASE
(
WSAEMSGSIZE
,
message_size
);
ERRC_CASE
(
WSAEPROTOTYPE
,
wrong_protocol_type
);
ERRC_CASE
(
WSAENOPROTOOPT
,
no_protocol_option
);
ERRC_CASE
(
WSAEPROTONOSUPPORT
,
protocol_not_supported
);
// Windows returns this error code if the *type* argument to socket() is
// invalid. POSIX returns EINVAL.
ERRC_CASE
(
WSAESOCKTNOSUPPORT
,
invalid_argument
);
ERRC_CASE
(
WSAEOPNOTSUPP
,
operation_not_supported
);
// Windows returns this error code if the *protocol* argument to socket() is
// invalid. POSIX returns EINVAL.
ERRC_CASE
(
WSAEPFNOSUPPORT
,
invalid_argument
);
ERRC_CASE
(
WSAEAFNOSUPPORT
,
address_family_not_supported
);
ERRC_CASE
(
WSAEADDRINUSE
,
address_in_use
);
ERRC_CASE
(
WSAEADDRNOTAVAIL
,
address_not_available
);
ERRC_CASE
(
WSAENETDOWN
,
network_down
);
ERRC_CASE
(
WSAENETUNREACH
,
network_unreachable
);
ERRC_CASE
(
WSAENETRESET
,
network_reset
);
ERRC_CASE
(
WSAECONNABORTED
,
connection_aborted
);
ERRC_CASE
(
WSAECONNRESET
,
connection_reset
);
ERRC_CASE
(
WSAENOBUFS
,
no_buffer_space
);
ERRC_CASE
(
WSAEISCONN
,
already_connected
);
ERRC_CASE
(
WSAENOTCONN
,
not_connected
);
// Windows returns this error code when writing to a socket with closed
// output channel. POSIX returns EPIPE.
ERRC_CASE
(
WSAESHUTDOWN
,
broken_pipe
);
ERRC_CASE
(
WSAETIMEDOUT
,
timed_out
);
ERRC_CASE
(
WSAECONNREFUSED
,
connection_refused
);
ERRC_CASE
(
WSAELOOP
,
too_many_symbolic_link_levels
);
ERRC_CASE
(
WSAENAMETOOLONG
,
filename_too_long
);
ERRC_CASE
(
WSAEHOSTUNREACH
,
host_unreachable
);
ERRC_CASE
(
WSAENOTEMPTY
,
directory_not_empty
);
ERRC_CASE
(
WSANOTINITIALISED
,
network_down
);
ERRC_CASE
(
WSAEDISCON
,
already_connected
);
ERRC_CASE
(
WSAENOMORE
,
not_connected
);
ERRC_CASE
(
WSAECANCELLED
,
operation_canceled
);
ERRC_CASE
(
WSATRY_AGAIN
,
resource_unavailable_try_again
);
ERRC_CASE
(
WSANO_RECOVERY
,
state_not_recoverable
);
}
fprintf
(
stderr
,
"[FATAL] %s:%u: unrecognized WSA error code (%d)
\n
"
,
__FILE__
,
__LINE__
,
wsa_code
);
abort
();
}
}
std
::
string
socket_error_as_string
(
int
errcode
)
{
std
::
string
last_socket_error_as_string
()
{
int
wsa_code
=
WSAGetLastError
();
LPTSTR
errorText
=
NULL
;
LPTSTR
errorText
=
NULL
;
FormatMessage
(
FORMAT_MESSAGE_FROM_SYSTEM
|
FORMAT_MESSAGE_ALLOCATE_BUFFER
FormatMessage
(
FORMAT_MESSAGE_FROM_SYSTEM
|
FORMAT_MESSAGE_ALLOCATE_BUFFER
|
FORMAT_MESSAGE_IGNORE_INSERTS
,
|
FORMAT_MESSAGE_IGNORE_INSERTS
,
nullptr
,
err
code
,
MAKELANGID
(
LANG_NEUTRAL
,
SUBLANG_DEFAULT
),
nullptr
,
wsa_
code
,
MAKELANGID
(
LANG_NEUTRAL
,
SUBLANG_DEFAULT
),
(
LPTSTR
)
&
errorText
,
0
,
nullptr
);
(
LPTSTR
)
&
errorText
,
0
,
nullptr
);
std
::
string
result
;
std
::
string
result
;
if
(
errorText
!=
nullptr
)
{
if
(
errorText
!=
nullptr
)
{
...
@@ -51,10 +120,6 @@ std::string socket_error_as_string(int errcode) {
...
@@ -51,10 +120,6 @@ std::string socket_error_as_string(int errcode) {
return
result
;
return
result
;
}
}
std
::
string
last_socket_error_as_string
()
{
return
socket_error_as_string
(
last_socket_error
());
}
bool
would_block_or_temporarily_unavailable
(
int
errcode
)
{
bool
would_block_or_temporarily_unavailable
(
int
errcode
)
{
return
errcode
==
WSAEWOULDBLOCK
||
errcode
==
WSATRY_AGAIN
;
return
errcode
==
WSAEWOULDBLOCK
||
errcode
==
WSATRY_AGAIN
;
}
}
...
@@ -79,12 +144,10 @@ void close(socket fd) {
...
@@ -79,12 +144,10 @@ void close(socket fd) {
::
close
(
fd
.
id
);
::
close
(
fd
.
id
);
}
}
int
last_socket_error
()
{
std
::
errc
last_socket_error
()
{
return
errno
;
// TODO: Linux and macOS both have some non-POSIX error codes that should get
}
// mapped accordingly.
return
static_cast
<
std
::
errc
>
(
errno
);
std
::
string
socket_error_as_string
(
int
error_code
)
{
return
strerror
(
error_code
);
}
}
std
::
string
last_socket_error_as_string
()
{
std
::
string
last_socket_error_as_string
()
{
...
@@ -117,13 +180,5 @@ error nonblocking(socket x, bool new_value) {
...
@@ -117,13 +180,5 @@ error nonblocking(socket x, bool new_value) {
#endif // CAF_WINDOWS
#endif // CAF_WINDOWS
bool
is_error
(
std
::
make_signed
<
size_t
>::
type
res
,
bool
is_nonblock
)
{
if
(
res
<
0
)
{
auto
err
=
last_socket_error
();
return
!
is_nonblock
||
!
would_block_or_temporarily_unavailable
(
err
);
}
return
false
;
}
}
// namespace net
}
// namespace net
}
// namespace caf
}
// namespace caf
libcaf_net/src/socket_manager.cpp
0 → 100644
View file @
1516ac5f
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#include "caf/net/socket_manager.hpp"
#include "caf/config.hpp"
#include "caf/net/multiplexer.hpp"
namespace
caf
{
namespace
net
{
socket_manager
::
socket_manager
(
socket
handle
,
const
multiplexer_ptr
&
parent
)
:
handle_
(
handle
),
mask_
(
operation
::
none
),
parent_
(
parent
)
{
CAF_ASSERT
(
parent
!=
nullptr
);
CAF_ASSERT
(
handle_
!=
invalid_socket
);
}
socket_manager
::~
socket_manager
()
{
close
(
handle_
);
}
operation
socket_manager
::
mask
()
const
noexcept
{
return
mask_
.
load
();
}
bool
socket_manager
::
mask_add
(
operation
flag
)
noexcept
{
CAF_ASSERT
(
flag
!=
operation
::
none
);
auto
x
=
mask
();
while
((
x
&
flag
)
!=
flag
)
if
(
mask_
.
compare_exchange_strong
(
x
,
x
|
flag
))
{
if
(
auto
ptr
=
parent_
.
lock
())
ptr
->
update
(
this
);
return
true
;
}
return
false
;
}
bool
socket_manager
::
mask_del
(
operation
flag
)
noexcept
{
CAF_ASSERT
(
flag
!=
operation
::
none
);
auto
x
=
mask
();
while
((
x
&
flag
)
!=
operation
::
none
)
if
(
mask_
.
compare_exchange_strong
(
x
,
x
&
~
flag
))
{
if
(
auto
ptr
=
parent_
.
lock
())
ptr
->
update
(
this
);
return
true
;
}
return
false
;
}
}
// namespace net
}
// namespace caf
libcaf_net/src/stream_socket.cpp
0 → 100644
View file @
1516ac5f
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#include "caf/net/stream_socket.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_aliases.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/expected.hpp"
#include "caf/logger.hpp"
#include "caf/variant.hpp"
namespace
caf
{
namespace
net
{
#ifdef CAF_WINDOWS
constexpr
int
no_sigpipe_io_flag
=
0
;
/**************************************************************************\
* Based on work of others; *
* original header: *
* *
* Copyright 2007, 2010 by Nathan C. Myers <ncm@cantrip.org> *
* Redistribution and use in source and binary forms, with or without *
* modification, are permitted provided that the following conditions *
* are met: *
* *
* Redistributions of source code must retain the above copyright notice, *
* this list of conditions and the following disclaimer. *
* *
* Redistributions in binary form must reproduce the above copyright *
* notice, this list of conditions and the following disclaimer in the *
* documentation and/or other materials provided with the distribution. *
* *
* The name of the author must not be used to endorse or promote products *
* derived from this software without specific prior written permission. *
* *
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS *
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT *
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR *
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT *
* HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, *
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT *
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, *
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY *
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT *
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE *
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. *
\**************************************************************************/
expected
<
std
::
pair
<
stream_socket
,
stream_socket
>>
make_stream_socket_pair
()
{
auto
addrlen
=
static_cast
<
int
>
(
sizeof
(
sockaddr_in
));
socket_id
socks
[
2
]
=
{
invalid_socket_id
,
invalid_socket_id
};
CAF_NET_SYSCALL
(
"socket"
,
listener
,
==
,
invalid_socket_id
,
::
socket
(
AF_INET
,
SOCK_STREAM
,
IPPROTO_TCP
));
union
{
sockaddr_in
inaddr
;
sockaddr
addr
;
}
a
;
memset
(
&
a
,
0
,
sizeof
(
a
));
a
.
inaddr
.
sin_family
=
AF_INET
;
a
.
inaddr
.
sin_addr
.
s_addr
=
htonl
(
INADDR_LOOPBACK
);
a
.
inaddr
.
sin_port
=
0
;
// makes sure all sockets are closed in case of an error
auto
guard
=
detail
::
make_scope_guard
([
&
]
{
auto
e
=
WSAGetLastError
();
close
(
socket
{
listener
});
close
(
socket
{
socks
[
0
]});
close
(
socket
{
socks
[
1
]});
WSASetLastError
(
e
);
});
// bind listener to a local port
int
reuse
=
1
;
CAF_NET_SYSCALL
(
"setsockopt"
,
tmp1
,
!=
,
0
,
setsockopt
(
listener
,
SOL_SOCKET
,
SO_REUSEADDR
,
reinterpret_cast
<
char
*>
(
&
reuse
),
static_cast
<
int
>
(
sizeof
(
reuse
))));
CAF_NET_SYSCALL
(
"bind"
,
tmp2
,
!=
,
0
,
bind
(
listener
,
&
a
.
addr
,
static_cast
<
int
>
(
sizeof
(
a
.
inaddr
))));
// Read the port in use: win32 getsockname may only set the port number
// (http://msdn.microsoft.com/library/ms738543.aspx).
memset
(
&
a
,
0
,
sizeof
(
a
));
CAF_NET_SYSCALL
(
"getsockname"
,
tmp3
,
!=
,
0
,
getsockname
(
listener
,
&
a
.
addr
,
&
addrlen
));
a
.
inaddr
.
sin_addr
.
s_addr
=
htonl
(
INADDR_LOOPBACK
);
a
.
inaddr
.
sin_family
=
AF_INET
;
// set listener to listen mode
CAF_NET_SYSCALL
(
"listen"
,
tmp5
,
!=
,
0
,
listen
(
listener
,
1
));
// create read-only end of the pipe
DWORD
flags
=
0
;
CAF_NET_SYSCALL
(
"WSASocketW"
,
read_fd
,
==
,
invalid_socket_id
,
WSASocketW
(
AF_INET
,
SOCK_STREAM
,
0
,
nullptr
,
0
,
flags
));
CAF_NET_SYSCALL
(
"connect"
,
tmp6
,
!=
,
0
,
connect
(
read_fd
,
&
a
.
addr
,
static_cast
<
int
>
(
sizeof
(
a
.
inaddr
))));
// get write-only end of the pipe
CAF_NET_SYSCALL
(
"accept"
,
write_fd
,
==
,
invalid_socket_id
,
accept
(
listener
,
nullptr
,
nullptr
));
close
(
socket
{
listener
});
guard
.
disable
();
return
std
::
make_pair
(
read_fd
,
write_fd
);
}
error
keepalive
(
stream_socket
x
,
bool
new_value
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
x
)
<<
CAF_ARG
(
new_value
));
char
value
=
new_value
?
1
:
0
;
CAF_NET_SYSCALL
(
"setsockopt"
,
res
,
!=
,
0
,
setsockopt
(
x
.
id
,
SOL_SOCKET
,
SO_KEEPALIVE
,
&
value
,
static_cast
<
int
>
(
sizeof
(
value
))));
return
none
;
}
#else // CAF_WINDOWS
# if defined(CAF_MACOS) || defined(CAF_IOS) || defined(CAF_BSD)
constexpr
int
no_sigpipe_io_flag
=
0
;
# else
constexpr
int
no_sigpipe_io_flag
=
MSG_NOSIGNAL
;
# endif
expected
<
std
::
pair
<
stream_socket
,
stream_socket
>>
make_stream_socket_pair
()
{
int
sockets
[
2
];
CAF_NET_SYSCALL
(
"socketpair"
,
spair_res
,
!=
,
0
,
socketpair
(
AF_UNIX
,
SOCK_STREAM
,
0
,
sockets
));
return
std
::
make_pair
(
stream_socket
{
sockets
[
0
]},
stream_socket
{
sockets
[
1
]});
}
error
keepalive
(
stream_socket
x
,
bool
new_value
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
x
)
<<
CAF_ARG
(
new_value
));
int
value
=
new_value
?
1
:
0
;
CAF_NET_SYSCALL
(
"setsockopt"
,
res
,
!=
,
0
,
setsockopt
(
x
.
id
,
SOL_SOCKET
,
SO_KEEPALIVE
,
&
value
,
static_cast
<
unsigned
>
(
sizeof
(
value
))));
return
none
;
}
#endif // CAF_WINDOWS
error
nodelay
(
stream_socket
x
,
bool
new_value
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
x
)
<<
CAF_ARG
(
new_value
));
int
flag
=
new_value
?
1
:
0
;
CAF_NET_SYSCALL
(
"setsockopt"
,
res
,
!=
,
0
,
setsockopt
(
x
.
id
,
IPPROTO_TCP
,
TCP_NODELAY
,
reinterpret_cast
<
setsockopt_ptr
>
(
&
flag
),
static_cast
<
socket_size_type
>
(
sizeof
(
flag
))));
return
none
;
}
variant
<
size_t
,
sec
>
read
(
stream_socket
x
,
void
*
buf
,
size_t
buf_size
)
{
auto
res
=
::
recv
(
x
.
id
,
reinterpret_cast
<
socket_recv_ptr
>
(
buf
),
buf_size
,
no_sigpipe_io_flag
);
return
check_stream_socket_io_res
(
res
);
}
variant
<
size_t
,
sec
>
write
(
stream_socket
x
,
const
void
*
buf
,
size_t
buf_size
)
{
auto
res
=
::
send
(
x
.
id
,
reinterpret_cast
<
socket_send_ptr
>
(
buf
),
buf_size
,
no_sigpipe_io_flag
);
return
check_stream_socket_io_res
(
res
);
}
variant
<
size_t
,
sec
>
check_stream_socket_io_res
(
std
::
make_signed
<
size_t
>::
type
res
)
{
if
(
res
==
0
)
return
sec
::
socket_disconnected
;
if
(
res
<
0
)
{
auto
code
=
last_socket_error
();
if
(
code
==
std
::
errc
::
operation_would_block
||
code
==
std
::
errc
::
resource_unavailable_try_again
)
return
sec
::
unavailable_or_would_block
;
return
sec
::
socket_operation_failed
;
}
return
static_cast
<
size_t
>
(
res
);
}
}
// namespace net
}
// namespace caf
libcaf_net/test/multiplexer.cpp
0 → 100644
View file @
1516ac5f
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#define CAF_SUITE multiplexer
#include "caf/net/multiplexer.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include <new>
#include <tuple>
#include <vector>
#include "caf/net/socket_manager.hpp"
#include "caf/net/stream_socket.hpp"
using
namespace
caf
;
using
namespace
caf
::
net
;
namespace
{
class
dummy_manager
:
public
socket_manager
{
public:
dummy_manager
(
size_t
&
manager_count
,
stream_socket
handle
,
multiplexer_ptr
parent
)
:
socket_manager
(
handle
,
std
::
move
(
parent
)),
count_
(
manager_count
),
rd_buf_pos_
(
0
)
{
++
count_
;
rd_buf_
.
resize
(
1024
);
}
~
dummy_manager
()
{
--
count_
;
}
stream_socket
handle
()
const
noexcept
{
return
socket_cast
<
stream_socket
>
(
handle_
);
}
bool
handle_read_event
()
override
{
if
(
read_capacity
()
<
1024
)
rd_buf_
.
resize
(
rd_buf_
.
size
()
+
2048
);
auto
res
=
read
(
handle
(),
read_position_begin
(),
read_capacity
());
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
res
))
{
CAF_ASSERT
(
*
num_bytes
>
0
);
rd_buf_pos_
+=
*
num_bytes
;
return
true
;
}
return
get
<
sec
>
(
res
)
==
sec
::
unavailable_or_would_block
;
}
bool
handle_write_event
()
override
{
if
(
wr_buf_
.
size
()
==
0
)
return
false
;
auto
res
=
write
(
handle
(),
wr_buf_
.
data
(),
wr_buf_
.
size
());
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
res
))
{
CAF_ASSERT
(
*
num_bytes
>
0
);
wr_buf_
.
erase
(
wr_buf_
.
begin
(),
wr_buf_
.
begin
()
+
*
num_bytes
);
return
wr_buf_
.
size
()
>
0
;
}
return
get
<
sec
>
(
res
)
==
sec
::
unavailable_or_would_block
;
}
void
handle_error
(
sec
code
)
override
{
CAF_FAIL
(
"handle_error called with code "
<<
code
);
}
void
send
(
string_view
x
)
{
wr_buf_
.
insert
(
wr_buf_
.
end
(),
x
.
begin
(),
x
.
end
());
}
std
::
string
receive
()
{
std
::
string
result
(
rd_buf_
.
data
(),
read_position_begin
());
rd_buf_pos_
=
0
;
return
result
;
}
private:
char
*
read_position_begin
()
{
return
rd_buf_
.
data
()
+
rd_buf_pos_
;
}
char
*
read_position_end
()
{
return
rd_buf_
.
data
()
+
rd_buf_
.
size
();
}
size_t
read_capacity
()
const
{
return
rd_buf_
.
size
()
-
rd_buf_pos_
;
}
size_t
&
count_
;
size_t
rd_buf_pos_
;
std
::
vector
<
char
>
wr_buf_
;
std
::
vector
<
char
>
rd_buf_
;
};
using
dummy_manager_ptr
=
intrusive_ptr
<
dummy_manager
>
;
struct
fixture
:
host_fixture
{
fixture
()
:
manager_count
(
0
),
mpx
(
std
::
make_shared
<
multiplexer
>
())
{
// nop
}
~
fixture
()
{
mpx
.
reset
();
CAF_REQUIRE_EQUAL
(
manager_count
,
0u
);
}
void
exhaust
()
{
while
(
mpx
->
poll_once
(
false
))
;
// Repeat.
}
size_t
manager_count
;
multiplexer_ptr
mpx
;
};
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
multiplexer_tests
,
fixture
)
CAF_TEST
(
default
construction
)
{
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
0u
);
}
CAF_TEST
(
init
)
{
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
0u
);
CAF_REQUIRE_EQUAL
(
mpx
->
init
(),
none
);
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
1u
);
mpx
->
close_pipe
();
exhaust
();
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
0u
);
// Calling run must have no effect now.
mpx
->
run
();
}
CAF_TEST
(
send
and
receive
)
{
CAF_REQUIRE_EQUAL
(
mpx
->
init
(),
none
);
auto
sockets
=
unbox
(
make_stream_socket_pair
());
auto
alice
=
make_counted
<
dummy_manager
>
(
manager_count
,
sockets
.
first
,
mpx
);
auto
bob
=
make_counted
<
dummy_manager
>
(
manager_count
,
sockets
.
second
,
mpx
);
alice
->
mask_add
(
operation
::
read
);
bob
->
mask_add
(
operation
::
read
);
mpx
->
handle_updates
();
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
3u
);
alice
->
send
(
"hello bob"
);
alice
->
mask_add
(
operation
::
write
);
mpx
->
handle_updates
();
exhaust
();
CAF_CHECK_EQUAL
(
bob
->
receive
(),
"hello bob"
);
}
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_net/test/network_socket.cpp
View file @
1516ac5f
...
@@ -30,10 +30,7 @@ using namespace caf::net;
...
@@ -30,10 +30,7 @@ using namespace caf::net;
CAF_TEST_FIXTURE_SCOPE
(
network_socket_tests
,
host_fixture
)
CAF_TEST_FIXTURE_SCOPE
(
network_socket_tests
,
host_fixture
)
CAF_TEST
(
invalid
socket
)
{
CAF_TEST
(
invalid
socket
)
{
auto
x
=
invalid_network_socket
;
network_socket
x
;
CAF_CHECK_EQUAL
(
keepalive
(
x
,
true
),
sec
::
network_syscall_failed
);
CAF_CHECK_EQUAL
(
tcp_nodelay
(
x
,
true
),
sec
::
network_syscall_failed
);
CAF_CHECK_EQUAL
(
allow_sigpipe
(
x
,
true
),
sec
::
network_syscall_failed
);
CAF_CHECK_EQUAL
(
allow_udp_connreset
(
x
,
true
),
sec
::
network_syscall_failed
);
CAF_CHECK_EQUAL
(
allow_udp_connreset
(
x
,
true
),
sec
::
network_syscall_failed
);
CAF_CHECK_EQUAL
(
send_buffer_size
(
x
),
sec
::
network_syscall_failed
);
CAF_CHECK_EQUAL
(
send_buffer_size
(
x
),
sec
::
network_syscall_failed
);
CAF_CHECK_EQUAL
(
local_port
(
x
),
sec
::
network_syscall_failed
);
CAF_CHECK_EQUAL
(
local_port
(
x
),
sec
::
network_syscall_failed
);
...
...
libcaf_net/test/stream_socket.cpp
0 → 100644
View file @
1516ac5f
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#define CAF_SUITE stream_socket
#include "caf/net/stream_socket.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
using
namespace
caf
;
using
namespace
caf
::
net
;
CAF_TEST_FIXTURE_SCOPE
(
network_socket_tests
,
host_fixture
)
CAF_TEST
(
invalid
socket
)
{
stream_socket
x
;
CAF_CHECK_EQUAL
(
keepalive
(
x
,
true
),
sec
::
network_syscall_failed
);
CAF_CHECK_EQUAL
(
nodelay
(
x
,
true
),
sec
::
network_syscall_failed
);
CAF_CHECK_EQUAL
(
allow_sigpipe
(
x
,
true
),
sec
::
network_syscall_failed
);
}
CAF_TEST
(
connected
socket
pair
)
{
std
::
vector
<
char
>
wr_buf
{
1
,
2
,
4
,
8
,
16
,
32
,
64
};
std
::
vector
<
char
>
rd_buf
(
124
);
CAF_MESSAGE
(
"create sockets and configure nonblocking I/O"
);
auto
x
=
unbox
(
make_stream_socket_pair
());
CAF_CHECK_EQUAL
(
nonblocking
(
x
.
first
,
true
),
caf
::
none
);
CAF_CHECK_EQUAL
(
nonblocking
(
x
.
second
,
true
),
caf
::
none
);
CAF_CHECK_NOT_EQUAL
(
unbox
(
send_buffer_size
(
x
.
first
)),
0u
);
CAF_CHECK_NOT_EQUAL
(
unbox
(
send_buffer_size
(
x
.
second
)),
0u
);
CAF_MESSAGE
(
"verify nonblocking communication"
);
CAF_CHECK_EQUAL
(
read
(
x
.
first
,
rd_buf
.
data
(),
rd_buf
.
size
()),
sec
::
unavailable_or_would_block
);
CAF_CHECK_EQUAL
(
read
(
x
.
second
,
rd_buf
.
data
(),
rd_buf
.
size
()),
sec
::
unavailable_or_would_block
);
CAF_MESSAGE
(
"transfer data from first to second socket"
);
CAF_CHECK_EQUAL
(
write
(
x
.
first
,
wr_buf
.
data
(),
wr_buf
.
size
()),
wr_buf
.
size
());
CAF_CHECK_EQUAL
(
read
(
x
.
second
,
rd_buf
.
data
(),
rd_buf
.
size
()),
wr_buf
.
size
());
CAF_CHECK
(
std
::
equal
(
wr_buf
.
begin
(),
wr_buf
.
end
(),
rd_buf
.
begin
()));
rd_buf
.
assign
(
rd_buf
.
size
(),
0
);
CAF_MESSAGE
(
"transfer data from second to first socket"
);
CAF_CHECK_EQUAL
(
write
(
x
.
second
,
wr_buf
.
data
(),
wr_buf
.
size
()),
wr_buf
.
size
());
CAF_CHECK_EQUAL
(
read
(
x
.
first
,
rd_buf
.
data
(),
rd_buf
.
size
()),
wr_buf
.
size
());
CAF_CHECK
(
std
::
equal
(
wr_buf
.
begin
(),
wr_buf
.
end
(),
rd_buf
.
begin
()));
rd_buf
.
assign
(
rd_buf
.
size
(),
0
);
CAF_MESSAGE
(
"shut down first socket and observe shutdown on the second one"
);
close
(
x
.
first
);
CAF_CHECK_EQUAL
(
read
(
x
.
second
,
rd_buf
.
data
(),
rd_buf
.
size
()),
sec
::
socket_disconnected
);
CAF_MESSAGE
(
"done (cleanup)"
);
close
(
x
.
second
);
}
CAF_TEST_FIXTURE_SCOPE_END
()
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