Commit 0c22a39a authored by Joseph Noir's avatar Joseph Noir Committed by GitHub

Merge pull request #4

Add scaffold for new multiplexer API
parents 6ae9f8d0 5b127358
...@@ -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)
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
...@@ -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<int> 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
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
...@@ -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
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
...@@ -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
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
...@@ -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<int> 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 size; return static_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
...@@ -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, 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) {
// 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, 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) {
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
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
...@@ -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, errcode, 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
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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()
...@@ -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);
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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()
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment