Commit da984663 authored by Joseph Noir's avatar Joseph Noir

Split up default multiplexer file

parent 8f0085f2
...@@ -35,6 +35,17 @@ set(LIBCAF_IO_SRCS ...@@ -35,6 +35,17 @@ set(LIBCAF_IO_SRCS
src/scribe.cpp src/scribe.cpp
src/stream_manager.cpp src/stream_manager.cpp
src/test_multiplexer.cpp src/test_multiplexer.cpp
src/acceptor.cpp
src/datagram_handler.cpp
src/datagram_servant_impl.cpp
src/doorman_impl.cpp
src/event_handler.cpp
src/pipe_reader.cpp
src/scribe_impl.cpp
src/stream.cpp
src/tcp.cpp
src/udp.cpp
src/socket_utils.cpp
) )
add_custom_target(libcaf_io) add_custom_target(libcaf_io)
......
...@@ -53,6 +53,7 @@ using datagram_servant_ptr = intrusive_ptr<datagram_servant>; ...@@ -53,6 +53,7 @@ using datagram_servant_ptr = intrusive_ptr<datagram_servant>;
namespace network { namespace network {
class multiplexer; class multiplexer;
class default_multiplexer;
} // namespace network } // namespace network
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/logger.hpp"
#include "caf/ref_counted.hpp"
#include "caf/io/fwd.hpp"
#include "caf/io/network/operation.hpp"
#include "caf/io/network/event_handler.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/io/network/acceptor_manager.hpp"
namespace caf {
namespace io {
namespace network {
/// An acceptor is responsible for accepting incoming connections.
class acceptor : public event_handler {
public:
/// A manager providing the `accept` member function.
using manager_type = acceptor_manager;
/// A smart pointer to an acceptor manager.
using manager_ptr = intrusive_ptr<manager_type>;
acceptor(default_multiplexer& backend_ref, native_socket sockfd);
/// Returns the accepted socket. This member function should
/// be called only from the `new_connection` callback.
inline native_socket& accepted_socket() {
return sock_;
}
/// Starts this acceptor, forwarding all incoming connections to
/// `manager`. The intrusive pointer will be released after the
/// acceptor has been closed or an IO error occured.
void start(acceptor_manager* mgr);
/// Activates the acceptor.
void activate(acceptor_manager* mgr);
/// Closes the network connection and removes this handler from its parent.
void stop_reading();
void removed_from_loop(operation op) override;
protected:
template <class Policy>
void handle_event_impl(io::network::operation op, Policy& policy) {
CAF_LOG_TRACE(CAF_ARG(fd()) << CAF_ARG(op));
if (mgr_ && op == operation::read) {
native_socket sockfd = invalid_native_socket;
if (policy.try_accept(sockfd, fd())) {
if (sockfd != invalid_native_socket) {
sock_ = sockfd;
mgr_->new_connection();
}
}
}
}
private:
manager_ptr mgr_;
native_socket sock_;
};
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/io/fwd.hpp"
#include "caf/io/network/acceptor.hpp"
#include "caf/io/network/operation.hpp"
#include "caf/io/network/native_socket.hpp"
namespace caf {
namespace io {
namespace network {
/// A concrete acceptor with a technology-dependent policy.
template <class ProtocolPolicy>
class acceptor_impl : public acceptor {
public:
template <class... Ts>
acceptor_impl(default_multiplexer& mpx, native_socket sockfd, Ts&&... xs)
: acceptor(mpx, sockfd),
policy_(std::forward<Ts>(xs)...) {
// nop
}
void handle_event(io::network::operation op) override {
this->handle_event_impl(op, policy_);
}
private:
ProtocolPolicy policy_;
};
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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 <vector>
#include <unordered_map>
#include "caf/logger.hpp"
#include "caf/ref_counted.hpp"
#include "caf/io/fwd.hpp"
#include "caf/io/receive_policy.hpp"
#include "caf/io/network/ip_endpoint.hpp"
#include "caf/io/network/event_handler.hpp"
#include "caf/io/network/socket_utils.hpp"
#include "caf/io/network/datagram_manager.hpp"
namespace caf {
namespace io {
namespace network {
class datagram_handler : public event_handler {
public:
/// A smart pointer to a datagram manager.
using manager_ptr = intrusive_ptr<datagram_manager>;
/// A buffer class providing a compatible interface to `std::vector`.
using write_buffer_type = std::vector<char>;
using read_buffer_type = network::receive_buffer;
/// A job for sending a datagram consisting of the sender and a buffer.
using job_type = std::pair<datagram_handle, write_buffer_type>;
datagram_handler(default_multiplexer& backend_ref, native_socket sockfd);
/// Starts reading data from the socket, forwarding incoming data to `mgr`.
void start(datagram_manager* mgr);
/// Activates the datagram handler.
void activate(datagram_manager* mgr);
void ack_writes(bool x);
/// Copies data to the write buffer.
/// @warning Not thread safe.
void write(datagram_handle hdl, const void* buf, size_t num_bytes);
/// Returns the write buffer of this enpoint.
/// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started.
inline write_buffer_type& wr_buf(datagram_handle hdl) {
wr_offline_buf_.emplace_back();
wr_offline_buf_.back().first = hdl;
return wr_offline_buf_.back().second;
}
/// Enqueues a buffer to be sent as a datagram.
/// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started.
inline void enqueue_datagram(datagram_handle hdl, std::vector<char> buf) {
wr_offline_buf_.emplace_back(hdl, move(buf));
}
/// Returns the read buffer of this stream.
/// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started.
inline read_buffer_type& rd_buf() {
return rd_buf_;
}
/// Sends the content of the write buffer, calling the `io_failure`
/// member function of `mgr` in case of an error.
/// @warning Must not be called outside the IO multiplexers event loop
/// once the stream has been started.
void flush(const manager_ptr& mgr);
/// Closes the read channel of the underlying socket and removes
/// this handler from its parent.
void stop_reading();
void removed_from_loop(operation op) override;
void add_endpoint(datagram_handle hdl, const ip_endpoint& ep,
const manager_ptr mgr);
std::unordered_map<datagram_handle, ip_endpoint>& endpoints();
const std::unordered_map<datagram_handle, ip_endpoint>& endpoints() const;
void remove_endpoint(datagram_handle hdl);
inline ip_endpoint& sending_endpoint() {
return sender_;
}
protected:
template <class Policy>
void handle_event_impl(io::network::operation op, Policy& policy) {
CAF_LOG_TRACE(CAF_ARG(op));
auto mcr = max_consecutive_reads_;
switch (op) {
case io::network::operation::read: {
// Loop until an error occurs or we have nothing more to read
// or until we have handled `mcr` reads.
for (size_t i = 0; i < mcr; ++i) {
auto res = policy.read_datagram(num_bytes_, fd(), rd_buf_.data(),
rd_buf_.size(), sender_);
if (!handle_read_result(res))
return;
}
break;
}
case io::network::operation::write: {
size_t wb; // written bytes
auto itr = ep_by_hdl_.find(wr_buf_.first);
// maybe this could be an assert?
if (itr == ep_by_hdl_.end())
CAF_RAISE_ERROR("got write event for undefined endpoint");
auto& id = itr->first;
auto& ep = itr->second;
std::vector<char> buf;
std::swap(buf, wr_buf_.second);
auto size_as_int = static_cast<int>(buf.size());
if (size_as_int > send_buffer_size_) {
send_buffer_size_ = size_as_int;
send_buffer_size(fd(), size_as_int);
}
auto res = policy.write_datagram(wb, fd(), buf.data(), buf.size(), ep);
handle_write_result(res, id, buf, wb);
break;
}
case operation::propagate_error:
handle_error();
break;
}
}
private:
size_t max_consecutive_reads_;
void prepare_next_read();
void prepare_next_write();
bool handle_read_result(bool read_result);
void handle_write_result(bool write_result, datagram_handle id,
std::vector<char>& buf, size_t wb);
void handle_error();
// known endpoints and broker servants
std::unordered_map<ip_endpoint, datagram_handle> hdl_by_ep_;
std::unordered_map<datagram_handle, ip_endpoint> ep_by_hdl_;
// state for reading
const size_t max_datagram_size_;
size_t num_bytes_;
read_buffer_type rd_buf_;
manager_ptr reader_;
ip_endpoint sender_;
// state for writing
int send_buffer_size_;
bool ack_writes_;
bool writing_;
std::deque<job_type> wr_offline_buf_;
job_type wr_buf_;
manager_ptr writer_;
};
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/io/fwd.hpp"
#include "caf/io/network/operation.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/io/network/datagram_handler.hpp"
namespace caf {
namespace io {
namespace network {
/// A concrete datagram_handler with a technology-dependent policy.
template <class ProtocolPolicy>
class datagram_handler_impl : public datagram_handler {
public:
template <class... Ts>
datagram_handler_impl(default_multiplexer& mpx, native_socket sockfd,
Ts&&... xs)
: datagram_handler(mpx, sockfd),
policy_(std::forward<Ts>(xs)...) {
// nop
}
void handle_event(io::network::operation op) override {
this->handle_event_impl(op, policy_);
}
private:
ProtocolPolicy policy_;
};
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/io/fwd.hpp"
#include "caf/io/datagram_servant.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/io/network/datagram_handler_impl.hpp"
#include "caf/io/network/policy/udp.hpp"
namespace caf {
namespace io {
namespace network {
/// Default datagram servant implementation.
class datagram_servant_impl : public datagram_servant {
using id_type = int64_t;
public:
datagram_servant_impl(default_multiplexer& mx, native_socket sockfd,
int64_t id);
bool new_endpoint(network::receive_buffer& buf) override;
void ack_writes(bool enable) override;
std::vector<char>& wr_buf(datagram_handle hdl) override;
void enqueue_datagram(datagram_handle hdl, std::vector<char> buf) override;
network::receive_buffer& rd_buf() override;
void stop_reading() override;
void flush() override;
std::string addr() const override;
uint16_t port(datagram_handle hdl) const override;
uint16_t local_port() const override;
std::vector<datagram_handle> hdls() const override;
void add_endpoint(const ip_endpoint& ep, datagram_handle hdl) override;
void remove_endpoint(datagram_handle hdl) override;
void launch() override;
void add_to_loop() override;
void remove_from_loop() override;
void detach_handles() override;
private:
bool launched_;
datagram_handler_impl<policy::udp> handler_;
};
} // namespace network
} // namespace io
} // namespace caf
...@@ -36,9 +36,14 @@ ...@@ -36,9 +36,14 @@
#include "caf/io/datagram_handle.hpp" #include "caf/io/datagram_handle.hpp"
#include "caf/io/datagram_servant.hpp" #include "caf/io/datagram_servant.hpp"
#include "caf/io/connection_handle.hpp" #include "caf/io/connection_handle.hpp"
#include "caf/io/network/rw_state.hpp"
#include "caf/io/network/operation.hpp" #include "caf/io/network/operation.hpp"
#include "caf/io/network/ip_endpoint.hpp" #include "caf/io/network/ip_endpoint.hpp"
#include "caf/io/network/multiplexer.hpp" #include "caf/io/network/multiplexer.hpp"
#include "caf/io/network/pipe_reader.hpp"
#include "caf/io/network/socket_utils.hpp"
#include "caf/io/network/event_handler.hpp"
#include "caf/io/network/receive_buffer.hpp" #include "caf/io/network/receive_buffer.hpp"
#include "caf/io/network/stream_manager.hpp" #include "caf/io/network/stream_manager.hpp"
#include "caf/io/network/acceptor_manager.hpp" #include "caf/io/network/acceptor_manager.hpp"
...@@ -48,32 +53,6 @@ ...@@ -48,32 +53,6 @@
#include "caf/logger.hpp" #include "caf/logger.hpp"
#ifdef CAF_WINDOWS
# ifndef WIN32_LEAN_AND_MEAN
# define WIN32_LEAN_AND_MEAN
# endif // WIN32_LEAN_AND_MEAN
# ifndef NOMINMAX
# define NOMINMAX
# endif
# ifdef CAF_MINGW
# undef _WIN32_WINNT
# undef WINVER
# define _WIN32_WINNT WindowsVista
# define WINVER WindowsVista
# include <w32api.h>
# endif
# include <winsock2.h>
# include <windows.h>
# include <ws2tcpip.h>
# include <ws2ipdef.h>
#else
# include <unistd.h>
# include <cerrno>
# include <sys/socket.h>
# include <netinet/in.h>
# include <netinet/ip.h>
#endif
// poll xs epoll backend // poll xs epoll backend
#if !defined(CAF_LINUX) || defined(CAF_POLL_IMPL) // poll() multiplexer #if !defined(CAF_LINUX) || defined(CAF_POLL_IMPL) // poll() multiplexer
# define CAF_POLL_MULTIPLEXER # define CAF_POLL_MULTIPLEXER
...@@ -95,49 +74,6 @@ namespace caf { ...@@ -95,49 +74,6 @@ namespace caf {
namespace io { namespace io {
namespace network { namespace network {
// annoying platform-dependent bootstrapping
#ifdef CAF_WINDOWS
using setsockopt_ptr = const char*;
using getsockopt_ptr = char*;
using socket_send_ptr = const char*;
using socket_recv_ptr = char*;
using socklen_t = int;
using ssize_t = std::make_signed<size_t>::type;
inline int last_socket_error() { return WSAGetLastError(); }
inline bool would_block_or_temporarily_unavailable(int errcode) {
return errcode == WSAEWOULDBLOCK || errcode == WSATRY_AGAIN;
}
constexpr int ec_out_of_memory = WSAENOBUFS;
constexpr int ec_interrupted_syscall = WSAEINTR;
#else
using setsockopt_ptr = const void*;
using getsockopt_ptr = void*;
using socket_send_ptr = const void*;
using socket_recv_ptr = void*;
inline void closesocket(int fd) { close(fd); }
inline int last_socket_error() { return errno; }
inline bool would_block_or_temporarily_unavailable(int errcode) {
return errcode == EAGAIN || errcode == EWOULDBLOCK;
}
constexpr int ec_out_of_memory = ENOMEM;
constexpr int ec_interrupted_syscall = EINTR;
#endif
// platform-dependent SIGPIPE setup
#if defined(CAF_MACOS) || defined(CAF_IOS) || defined(CAF_BSD)
// Use the socket option but no flags to recv/send on macOS/iOS/BSD.
constexpr int no_sigpipe_socket_flag = SO_NOSIGPIPE;
constexpr int no_sigpipe_io_flag = 0;
#elif defined(CAF_WINDOWS)
// Do nothing on Windows (SIGPIPE does not exist).
constexpr int no_sigpipe_socket_flag = 0;
constexpr int no_sigpipe_io_flag = 0;
#else
// Use flags to recv/send on Linux/Android but no socket option.
constexpr int no_sigpipe_socket_flag = 0;
constexpr int no_sigpipe_io_flag = MSG_NOSIGNAL;
#endif
// poll vs epoll backend // poll vs epoll backend
#if !defined(CAF_LINUX) || defined(CAF_POLL_IMPL) // poll() multiplexer #if !defined(CAF_LINUX) || defined(CAF_POLL_IMPL) // poll() multiplexer
# ifdef CAF_WINDOWS # ifdef CAF_WINDOWS
...@@ -164,190 +100,6 @@ namespace network { ...@@ -164,190 +100,6 @@ namespace network {
/// Platform-specific native acceptor socket type. /// Platform-specific native acceptor socket type.
using native_socket_acceptor = native_socket; using native_socket_acceptor = native_socket;
/// Returns the last socket error as human-readable string.
std::string last_socket_error_as_string();
/// Creates two connected sockets. The former is the read handle
/// and the latter is the write handle.
std::pair<native_socket, native_socket> create_pipe();
/// Sets fd to nonblocking if `set_nonblocking == true`
/// or to blocking if `set_nonblocking == false`
/// throws `network_error` on error
expected<void> nonblocking(native_socket fd, bool new_value);
/// Enables or disables Nagle's algorithm on `fd`.
/// @throws network_error
expected<void> tcp_nodelay(native_socket fd, bool new_value);
/// Enables or disables `SIGPIPE` events from `fd`.
expected<void> allow_sigpipe(native_socket fd, bool new_value);
/// Enables or disables `SIO_UDP_CONNRESET`error on `fd`.
expected<void> allow_udp_connreset(native_socket fd, bool new_value);
/// Get the socket buffer size for `fd`.
expected<int> send_buffer_size(native_socket fd);
/// Set the socket buffer size for `fd`.
expected<void> send_buffer_size(native_socket fd, int new_value);
/// Denotes the returned state of read and write operations on sockets.
enum class rw_state {
/// Reports that bytes could be read or written.
success,
/// Reports that the socket is closed or faulty.
failure,
/// Reports that an empty buffer is in use and no operation was performed.
indeterminate
};
/// Convenience functions for checking the result of `recv` or `send`.
bool is_error(ssize_t res, bool is_nonblock);
/// Reads up to `len` bytes from `fd,` writing the received data
/// to `buf`. Returns `true` as long as `fd` is readable and `false`
/// if the socket has been closed or an IO error occured. The number
/// of read bytes is stored in `result` (can be 0).
rw_state read_some(size_t& result, native_socket fd, void* buf, size_t len);
/// Writes up to `len` bytes from `buf` to `fd`.
/// Returns `true` as long as `fd` is readable and `false`
/// if the socket has been closed or an IO error occured. The number
/// of written bytes is stored in `result` (can be 0).
rw_state write_some(size_t& result, native_socket fd, const void* buf,
size_t len);
/// Tries to accept a new connection from `fd`. On success,
/// the new connection is stored in `result`. Returns true
/// as long as
bool try_accept(native_socket& result, native_socket fd);
/// Function signature of `read_some`.
using read_some_fun = decltype(read_some)*;
/// Function signature of `wite_some`.
using write_some_fun = decltype(write_some)*;
/// Function signature of `try_accept`.
using try_accept_fun = decltype(try_accept)*;
/// Policy object for wrapping default TCP operations.
struct tcp_policy {
static read_some_fun read_some;
static write_some_fun write_some;
static try_accept_fun try_accept;
};
/// Write a datagram containing `buf_len` bytes to `fd` addressed
/// at the endpoint in `sa` with size `sa_len`. Returns true as long
/// as no IO error occurs. The number of written bytes is stored in
/// `result` and the sender is stored in `ep`.
bool read_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len,
ip_endpoint& ep);
/// Reveice a datagram of up to `len` bytes. Larger datagrams are truncated.
/// Up to `sender_len` bytes of the receiver address is written into
/// `sender_addr`. Returns `true` if no IO error occurred. The number of
/// received bytes is stored in `result` (can be 0).
bool write_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len,
const ip_endpoint& ep);
/// Function signature of read_datagram
using read_datagram_fun = decltype(read_datagram)*;
/// Function signature of write_datagram
using write_datagram_fun = decltype(write_datagram)*;
/// Policy object for wrapping default UDP operations
struct udp_policy {
static read_datagram_fun read_datagram;
static write_datagram_fun write_datagram;
};
/// Returns the locally assigned port of `fd`.
expected<uint16_t> local_port_of_fd(native_socket fd);
/// Returns the locally assigned address of `fd`.
expected<std::string> local_addr_of_fd(native_socket fd);
/// Returns the port used by the remote host of `fd`.
expected<uint16_t> remote_port_of_fd(native_socket fd);
/// Returns the remote host address of `fd`.
expected<std::string> remote_addr_of_fd(native_socket fd);
class default_multiplexer;
/// A socket I/O event handler.
class event_handler {
public:
event_handler(default_multiplexer& dm, native_socket sockfd);
virtual ~event_handler();
/// Returns true once the requested operation is done, i.e.,
/// to signalize the multiplexer to remove this handler.
/// The handler remains in the event loop as long as it returns false.
virtual void handle_event(operation op) = 0;
/// Callback to signalize that this handler has been removed
/// from the event loop for operations of type `op`.
virtual void removed_from_loop(operation op) = 0;
/// Returns the native socket handle for this handler.
inline native_socket fd() const {
return fd_;
}
/// Returns the `multiplexer` this acceptor belongs to.
inline default_multiplexer& backend() {
return backend_;
}
/// Returns the bit field storing the subscribed events.
inline int eventbf() const {
return eventbf_;
}
/// Sets the bit field storing the subscribed events.
inline void eventbf(int value) {
eventbf_ = value;
}
/// Checks whether `close_read` has been called.
inline bool read_channel_closed() const {
return read_channel_closed_;
}
/// Closes the read channel of the underlying socket.
void close_read_channel();
/// Removes the file descriptor from the event loop of the parent.
void passivate();
protected:
/// Adds the file descriptor to the event loop of the parent.
void activate();
void set_fd_flags();
int eventbf_;
native_socket fd_;
bool read_channel_closed_;
default_multiplexer& backend_;
};
/// An event handler for the internal event pipe.
class pipe_reader : public event_handler {
public:
pipe_reader(default_multiplexer& dm);
void removed_from_loop(operation op) override;
void handle_event(operation op) override;
void init(native_socket sock_fd);
resumable* try_read_next();
};
class default_multiplexer : public multiplexer { class default_multiplexer : public multiplexer {
public: public:
friend class io::middleman; // disambiguate reference friend class io::middleman; // disambiguate reference
...@@ -509,6 +261,38 @@ private: ...@@ -509,6 +261,38 @@ private:
size_t max_throughput_; size_t max_throughput_;
}; };
/// Reads up to `len` bytes from `fd,` writing the received data
/// to `buf`. Returns `true` as long as `fd` is readable and `false`
/// if the socket has been closed or an IO error occured. The number
/// of read bytes is stored in `result` (can be 0).
rw_state read_some(size_t& result, native_socket fd, void* buf, size_t len);
/// Writes up to `len` bytes from `buf` to `fd`.
/// Returns `true` as long as `fd` is readable and `false`
/// if the socket has been closed or an IO error occured. The number
/// of written bytes is stored in `result` (can be 0).
rw_state write_some(size_t& result, native_socket fd, const void* buf,
size_t len);
/// Tries to accept a new connection from `fd`. On success,
/// the new connection is stored in `result`. Returns true
/// as long as
bool try_accept(native_socket& result, native_socket fd);
/// Write a datagram containing `buf_len` bytes to `fd` addressed
/// at the endpoint in `sa` with size `sa_len`. Returns true as long
/// as no IO error occurs. The number of written bytes is stored in
/// `result` and the sender is stored in `ep`.
bool read_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len,
ip_endpoint& ep);
/// Reveice a datagram of up to `len` bytes. Larger datagrams are truncated.
/// Up to `sender_len` bytes of the receiver address is written into
/// `sender_addr`. Returns `true` if no IO error occurred. The number of
/// received bytes is stored in `result` (can be 0).
bool write_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len,
const ip_endpoint& ep);
inline connection_handle conn_hdl_from_socket(native_socket fd) { inline connection_handle conn_hdl_from_socket(native_socket fd) {
return connection_handle::from_int(int64_from_native_socket(fd)); return connection_handle::from_int(int64_from_native_socket(fd));
} }
...@@ -517,449 +301,6 @@ inline accept_handle accept_hdl_from_socket(native_socket fd) { ...@@ -517,449 +301,6 @@ inline accept_handle accept_hdl_from_socket(native_socket fd) {
return accept_handle::from_int(int64_from_native_socket(fd)); return accept_handle::from_int(int64_from_native_socket(fd));
} }
/// A stream capable of both reading and writing. The stream's input
/// data is forwarded to its {@link stream_manager manager}.
class stream : public event_handler {
public:
/// A smart pointer to a stream manager.
using manager_ptr = intrusive_ptr<stream_manager>;
/// A buffer class providing a compatible
/// interface to `std::vector`.
using buffer_type = std::vector<char>;
stream(default_multiplexer& backend_ref, native_socket sockfd);
/// Starts reading data from the socket, forwarding incoming data to `mgr`.
void start(stream_manager* mgr);
/// Activates the stream.
void activate(stream_manager* mgr);
/// Configures how much data will be provided for the next `consume` callback.
/// @warning Must not be called outside the IO multiplexers event loop
/// once the stream has been started.
void configure_read(receive_policy::config config);
void ack_writes(bool x);
/// Copies data to the write buffer.
/// @warning Not thread safe.
void write(const void* buf, size_t num_bytes);
/// Returns the write buffer of this stream.
/// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started.
inline buffer_type& wr_buf() {
return wr_offline_buf_;
}
/// Returns the read buffer of this stream.
/// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started.
inline buffer_type& rd_buf() {
return rd_buf_;
}
/// Sends the content of the write buffer, calling the `io_failure`
/// member function of `mgr` in case of an error.
/// @warning Must not be called outside the IO multiplexers event loop
/// once the stream has been started.
void flush(const manager_ptr& mgr);
/// Closes the read channel of the underlying socket and removes
/// this handler from its parent.
void stop_reading();
void removed_from_loop(operation op) override;
/// Forces this stream to subscribe to write events if no data is in the
/// write buffer.
void force_empty_write(const manager_ptr& mgr) {
if (!writing_) {
backend().add(operation::write, fd(), this);
writer_ = mgr;
writing_ = true;
}
}
protected:
template <class Policy>
void handle_event_impl(io::network::operation op, Policy& policy) {
CAF_LOG_TRACE(CAF_ARG(op));
auto mcr = max_consecutive_reads_;
switch (op) {
case io::network::operation::read: {
// Loop until an error occurs or we have nothing more to read
// or until we have handled `mcr` reads.
size_t rb;
for (size_t i = 0; i < mcr; ++i) {
switch (policy.read_some(rb, fd(), rd_buf_.data() + collected_,
rd_buf_.size() - collected_)) {
case rw_state::failure:
reader_->io_failure(&backend(), operation::read);
passivate();
return;
case rw_state::indeterminate:
return;
case rw_state::success:
if (rb == 0)
return;
collected_ += rb;
if (collected_ >= read_threshold_) {
auto res = reader_->consume(&backend(), rd_buf_.data(),
collected_);
prepare_next_read();
if (!res) {
passivate();
return;
}
}
}
}
break;
}
case io::network::operation::write: {
size_t wb; // written bytes
switch (policy.write_some(wb, fd(), wr_buf_.data() + written_,
wr_buf_.size() - written_)) {
case rw_state::failure:
writer_->io_failure(&backend(), operation::write);
backend().del(operation::write, fd(), this);
break;
case rw_state::indeterminate:
prepare_next_write();
break;
case rw_state::success:
written_ += wb;
CAF_ASSERT(written_ <= wr_buf_.size());
auto remaining = wr_buf_.size() - written_;
if (ack_writes_)
writer_->data_transferred(&backend(), wb,
remaining + wr_offline_buf_.size());
// prepare next send (or stop sending)
if (remaining == 0)
prepare_next_write();
}
break;
}
case operation::propagate_error:
if (reader_)
reader_->io_failure(&backend(), operation::read);
if (writer_)
writer_->io_failure(&backend(), operation::write);
// backend will delete this handler anyway,
// no need to call backend().del() here
}
}
private:
void prepare_next_read();
void prepare_next_write();
size_t max_consecutive_reads_;
// state for reading
manager_ptr reader_;
size_t read_threshold_;
size_t collected_;
size_t max_;
receive_policy_flag rd_flag_;
buffer_type rd_buf_;
// state for writing
manager_ptr writer_;
bool ack_writes_;
bool writing_;
size_t written_;
buffer_type wr_buf_;
buffer_type wr_offline_buf_;
};
/// A concrete stream with a technology-dependent policy for sending and
/// receiving data from a socket.
template <class ProtocolPolicy>
class stream_impl : public stream {
public:
template <class... Ts>
stream_impl(default_multiplexer& mpx, native_socket sockfd, Ts&&... xs)
: stream(mpx, sockfd),
policy_(std::forward<Ts>(xs)...) {
// nop
}
void handle_event(io::network::operation op) override {
this->handle_event_impl(op, policy_);
}
private:
ProtocolPolicy policy_;
};
/// An acceptor is responsible for accepting incoming connections.
class acceptor : public event_handler {
public:
/// A manager providing the `accept` member function.
using manager_type = acceptor_manager;
/// A smart pointer to an acceptor manager.
using manager_ptr = intrusive_ptr<manager_type>;
acceptor(default_multiplexer& backend_ref, native_socket sockfd);
/// Returns the accepted socket. This member function should
/// be called only from the `new_connection` callback.
inline native_socket& accepted_socket() {
return sock_;
}
/// Starts this acceptor, forwarding all incoming connections to
/// `manager`. The intrusive pointer will be released after the
/// acceptor has been closed or an IO error occured.
void start(acceptor_manager* mgr);
/// Activates the acceptor.
void activate(acceptor_manager* mgr);
/// Closes the network connection and removes this handler from its parent.
void stop_reading();
void removed_from_loop(operation op) override;
protected:
template <class Policy>
void handle_event_impl(io::network::operation op, Policy& policy) {
CAF_LOG_TRACE(CAF_ARG(fd()) << CAF_ARG(op));
if (mgr_ && op == operation::read) {
native_socket sockfd = invalid_native_socket;
if (policy.try_accept(sockfd, fd())) {
if (sockfd != invalid_native_socket) {
sock_ = sockfd;
mgr_->new_connection();
}
}
}
}
private:
manager_ptr mgr_;
native_socket sock_;
};
/// A concrete acceptor with a technology-dependent policy.
template <class ProtocolPolicy>
class acceptor_impl : public acceptor {
public:
template <class... Ts>
acceptor_impl(default_multiplexer& mpx, native_socket sockfd, Ts&&... xs)
: acceptor(mpx, sockfd),
policy_(std::forward<Ts>(xs)...) {
// nop
}
void handle_event(io::network::operation op) override {
this->handle_event_impl(op, policy_);
}
private:
ProtocolPolicy policy_;
};
class datagram_handler : public event_handler {
public:
/// A smart pointer to a datagram manager.
using manager_ptr = intrusive_ptr<datagram_manager>;
/// A buffer class providing a compatible interface to `std::vector`.
using write_buffer_type = std::vector<char>;
using read_buffer_type = network::receive_buffer;
/// A job for sending a datagram consisting of the sender and a buffer.
using job_type = std::pair<datagram_handle, write_buffer_type>;
datagram_handler(default_multiplexer& backend_ref, native_socket sockfd);
/// Starts reading data from the socket, forwarding incoming data to `mgr`.
void start(datagram_manager* mgr);
/// Activates the datagram handler.
void activate(datagram_manager* mgr);
void ack_writes(bool x);
/// Copies data to the write buffer.
/// @warning Not thread safe.
void write(datagram_handle hdl, const void* buf, size_t num_bytes);
/// Returns the write buffer of this enpoint.
/// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started.
inline write_buffer_type& wr_buf(datagram_handle hdl) {
wr_offline_buf_.emplace_back();
wr_offline_buf_.back().first = hdl;
return wr_offline_buf_.back().second;
}
/// Enqueues a buffer to be sent as a datagram.
/// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started.
inline void enqueue_datagram(datagram_handle hdl, std::vector<char> buf) {
wr_offline_buf_.emplace_back(hdl, move(buf));
}
/// Returns the read buffer of this stream.
/// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started.
inline read_buffer_type& rd_buf() {
return rd_buf_;
}
/// Sends the content of the write buffer, calling the `io_failure`
/// member function of `mgr` in case of an error.
/// @warning Must not be called outside the IO multiplexers event loop
/// once the stream has been started.
void flush(const manager_ptr& mgr);
/// Closes the read channel of the underlying socket and removes
/// this handler from its parent.
void stop_reading();
void removed_from_loop(operation op) override;
void add_endpoint(datagram_handle hdl, const ip_endpoint& ep,
const manager_ptr mgr);
std::unordered_map<datagram_handle, ip_endpoint>& endpoints();
const std::unordered_map<datagram_handle, ip_endpoint>& endpoints() const;
void remove_endpoint(datagram_handle hdl);
inline ip_endpoint& sending_endpoint() {
return sender_;
}
protected:
template <class Policy>
void handle_event_impl(io::network::operation op, Policy& policy) {
CAF_LOG_TRACE(CAF_ARG(op));
auto mcr = max_consecutive_reads_;
switch (op) {
case io::network::operation::read: {
// Loop until an error occurs or we have nothing more to read
// or until we have handled `mcr` reads.
for (size_t i = 0; i < mcr; ++i) {
if (!policy.read_datagram(num_bytes_, fd(), rd_buf_.data(),
rd_buf_.size(), sender_)) {
reader_->io_failure(&backend(), operation::read);
passivate();
return;
}
if (num_bytes_ > 0) {
rd_buf_.resize(num_bytes_);
auto itr = hdl_by_ep_.find(sender_);
bool consumed = false;
if (itr == hdl_by_ep_.end())
consumed = reader_->new_endpoint(rd_buf_);
else
consumed = reader_->consume(&backend(), itr->second, rd_buf_);
prepare_next_read();
if (!consumed) {
passivate();
return;
}
}
}
break;
}
case io::network::operation::write: {
size_t wb; // written bytes
auto itr = ep_by_hdl_.find(wr_buf_.first);
// maybe this could be an assert?
if (itr == ep_by_hdl_.end())
CAF_RAISE_ERROR("got write event for undefined endpoint");
auto& id = itr->first;
auto& ep = itr->second;
std::vector<char> buf;
std::swap(buf, wr_buf_.second);
auto size_as_int = static_cast<int>(buf.size());
if (size_as_int > send_buffer_size_) {
send_buffer_size_ = size_as_int;
send_buffer_size(fd(), size_as_int);
}
if (!policy.write_datagram(wb, fd(), buf.data(),
buf.size(), ep)) {
writer_->io_failure(&backend(), operation::write);
backend().del(operation::write, fd(), this);
} else if (wb > 0) {
CAF_ASSERT(wb == buf.size());
if (ack_writes_)
writer_->datagram_sent(&backend(), id, wb, std::move(buf));
prepare_next_write();
} else {
if (writer_)
writer_->io_failure(&backend(), operation::write);
}
break;
}
case operation::propagate_error:
if (reader_)
reader_->io_failure(&backend(), operation::read);
if (writer_)
writer_->io_failure(&backend(), operation::write);
// backend will delete this handler anyway,
// no need to call backend().del() here
}
}
private:
size_t max_consecutive_reads_;
void prepare_next_read();
void prepare_next_write();
// known endpoints and broker servants
std::unordered_map<ip_endpoint, datagram_handle> hdl_by_ep_;
std::unordered_map<datagram_handle, ip_endpoint> ep_by_hdl_;
// state for reading
const size_t max_datagram_size_;
size_t num_bytes_;
read_buffer_type rd_buf_;
manager_ptr reader_;
ip_endpoint sender_;
// state for writing
int send_buffer_size_;
bool ack_writes_;
bool writing_;
std::deque<job_type> wr_offline_buf_;
job_type wr_buf_;
manager_ptr writer_;
};
/// A concrete datagram_handler with a technology-dependent policy.
template <class ProtocolPolicy>
class datagram_handler_impl : public datagram_handler {
public:
template <class... Ts>
datagram_handler_impl(default_multiplexer& mpx, native_socket sockfd,
Ts&&... xs)
: datagram_handler(mpx, sockfd),
policy_(std::forward<Ts>(xs)...) {
// nop
}
void handle_event(io::network::operation op) override {
this->handle_event_impl(op, policy_);
}
private:
ProtocolPolicy policy_;
};
expected<native_socket> expected<native_socket>
new_tcp_connection(const std::string& host, uint16_t port, new_tcp_connection(const std::string& host, uint16_t port,
optional<protocol::network> preferred = none); optional<protocol::network> preferred = none);
...@@ -967,108 +308,6 @@ new_tcp_connection(const std::string& host, uint16_t port, ...@@ -967,108 +308,6 @@ new_tcp_connection(const std::string& host, uint16_t port,
expected<native_socket> expected<native_socket>
new_tcp_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr); new_tcp_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr);
/// Default doorman implementation.
class doorman_impl : public doorman {
public:
doorman_impl(default_multiplexer& mx, native_socket sockfd);
bool new_connection() override;
void stop_reading() override;
void launch() override;
std::string addr() const override;
uint16_t port() const override;
void add_to_loop() override;
void remove_from_loop() override;
protected:
acceptor_impl<tcp_policy> acceptor_;
};
/// Default scribe implementation.
class scribe_impl : public scribe {
public:
scribe_impl(default_multiplexer& mx, native_socket sockfd);
void configure_read(receive_policy::config config) override;
void ack_writes(bool enable) override;
std::vector<char>& wr_buf() override;
std::vector<char>& rd_buf() override;
void stop_reading() override;
void flush() override;
std::string addr() const override;
uint16_t port() const override;
void launch();
void add_to_loop() override;
void remove_from_loop() override;
protected:
bool launched_;
stream_impl<tcp_policy> stream_;
};
/// Default datagram servant implementation
class datagram_servant_impl : public datagram_servant {
using id_type = int64_t;
public:
datagram_servant_impl(default_multiplexer& mx, native_socket sockfd,
int64_t id);
bool new_endpoint(network::receive_buffer& buf) override;
void ack_writes(bool enable) override;
std::vector<char>& wr_buf(datagram_handle hdl) override;
void enqueue_datagram(datagram_handle hdl, std::vector<char> buf) override;
network::receive_buffer& rd_buf() override;
void stop_reading() override;
void flush() override;
std::string addr() const override;
uint16_t port(datagram_handle hdl) const override;
uint16_t local_port() const override;
std::vector<datagram_handle> hdls() const override;
void add_endpoint(const ip_endpoint& ep, datagram_handle hdl) override;
void remove_endpoint(datagram_handle hdl) override;
void launch() override;
void add_to_loop() override;
void remove_from_loop() override;
void detach_handles() override;
private:
bool launched_;
datagram_handler_impl<udp_policy> handler_;
};
expected<std::pair<native_socket, ip_endpoint>> expected<std::pair<native_socket, ip_endpoint>>
new_remote_udp_endpoint_impl(const std::string& host, uint16_t port, new_remote_udp_endpoint_impl(const std::string& host, uint16_t port,
optional<protocol::network> preferred = none); optional<protocol::network> preferred = none);
...@@ -1081,4 +320,3 @@ new_local_udp_endpoint_impl(uint16_t port, const char* addr, ...@@ -1081,4 +320,3 @@ new_local_udp_endpoint_impl(uint16_t port, const char* addr,
} // namespace network } // namespace network
} // namespace io } // namespace io
} // namespace caf } // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/io/fwd.hpp"
#include "caf/io/doorman.hpp"
#include "caf/io/network/acceptor_impl.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/io/network/policy/tcp.hpp"
namespace caf {
namespace io {
namespace network {
/// Default doorman implementation.
class doorman_impl : public doorman {
public:
doorman_impl(default_multiplexer& mx, native_socket sockfd);
bool new_connection() override;
void stop_reading() override;
void launch() override;
std::string addr() const override;
uint16_t port() const override;
void add_to_loop() override;
void remove_from_loop() override;
protected:
acceptor_impl<policy::tcp> acceptor_;
};
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/io/fwd.hpp"
#include "caf/io/network/operation.hpp"
#include "caf/io/network/native_socket.hpp"
namespace caf {
namespace io {
namespace network {
/// A socket I/O event handler.
class event_handler {
public:
event_handler(default_multiplexer& dm, native_socket sockfd);
virtual ~event_handler();
/// Returns true once the requested operation is done, i.e.,
/// to signalize the multiplexer to remove this handler.
/// The handler remains in the event loop as long as it returns false.
virtual void handle_event(operation op) = 0;
/// Callback to signalize that this handler has been removed
/// from the event loop for operations of type `op`.
virtual void removed_from_loop(operation op) = 0;
/// Returns the native socket handle for this handler.
inline native_socket fd() const {
return fd_;
}
/// Returns the `multiplexer` this acceptor belongs to.
inline default_multiplexer& backend() {
return backend_;
}
/// Returns the bit field storing the subscribed events.
inline int eventbf() const {
return eventbf_;
}
/// Sets the bit field storing the subscribed events.
inline void eventbf(int value) {
eventbf_ = value;
}
/// Checks whether `close_read` has been called.
inline bool read_channel_closed() const {
return read_channel_closed_;
}
/// Closes the read channel of the underlying socket.
void close_read_channel();
/// Removes the file descriptor from the event loop of the parent.
void passivate();
protected:
/// Adds the file descriptor to the event loop of the parent.
void activate();
void set_fd_flags();
int eventbf_;
native_socket fd_;
bool read_channel_closed_;
default_multiplexer& backend_;
};
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/io/fwd.hpp"
#include "caf/io/network/operation.hpp"
#include "caf/io/network/event_handler.hpp"
#include "caf/io/network/native_socket.hpp"
namespace caf {
namespace io {
namespace network {
/// An event handler for the internal event pipe.
class pipe_reader : public event_handler {
public:
pipe_reader(default_multiplexer& dm);
void removed_from_loop(operation op) override;
void handle_event(operation op) override;
void init(native_socket sock_fd);
resumable* try_read_next();
};
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/io/network/default_multiplexer.hpp"
namespace caf {
namespace io {
namespace network {
namespace policy {
/// Function signature of `read_some`.
using read_some_fun = decltype(read_some)*;
/// Function signature of `wite_some`.
using write_some_fun = decltype(write_some)*;
/// Function signature of `try_accept`.
using try_accept_fun = decltype(try_accept)*;
/// Policy object for wrapping default TCP operations.
struct tcp {
static read_some_fun read_some;
static write_some_fun write_some;
static try_accept_fun try_accept;
};
} // namespace policy
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/io/network/default_multiplexer.hpp"
namespace caf {
namespace io {
namespace network {
namespace policy {
/// Function signature of read_datagram
using read_datagram_fun = decltype(read_datagram)*;
/// Function signature of write_datagram
using write_datagram_fun = decltype(write_datagram)*;
/// Policy object for wrapping default UDP operations
struct udp {
static read_datagram_fun read_datagram;
static write_datagram_fun write_datagram;
};
} // namespace policy
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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 io {
namespace network {
/// Denotes the returned state of read and write operations on sockets.
enum class rw_state {
/// Reports that bytes could be read or written.
success,
/// Reports that the socket is closed or faulty.
failure,
/// Reports that an empty buffer is in use and no operation was performed.
indeterminate
};
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/io/fwd.hpp"
#include "caf/io/scribe.hpp"
#include "caf/io/network/stream_impl.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/io/network/policy/tcp.hpp"
namespace caf {
namespace io {
namespace network {
/// Default scribe implementation.
class scribe_impl : public scribe {
public:
scribe_impl(default_multiplexer& mx, native_socket sockfd);
void configure_read(receive_policy::config config) override;
void ack_writes(bool enable) override;
std::vector<char>& wr_buf() override;
std::vector<char>& rd_buf() override;
void stop_reading() override;
void flush() override;
std::string addr() const override;
uint16_t port() const override;
void launch();
void add_to_loop() override;
void remove_from_loop() override;
protected:
bool launched_;
stream_impl<policy::tcp> stream_;
};
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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 <string>
#include "caf/expected.hpp"
#include "caf/io/network/native_socket.hpp"
#ifdef CAF_WINDOWS
# ifndef WIN32_LEAN_AND_MEAN
# define WIN32_LEAN_AND_MEAN
# endif // WIN32_LEAN_AND_MEAN
# ifndef NOMINMAX
# define NOMINMAX
# endif
# ifdef CAF_MINGW
# undef _WIN32_WINNT
# undef WINVER
# define _WIN32_WINNT WindowsVista
# define WINVER WindowsVista
# include <w32api.h>
# endif
# include <winsock2.h>
# include <windows.h>
# include <ws2tcpip.h>
# include <ws2ipdef.h>
#else
# include <unistd.h>
# include <cerrno>
# include <sys/socket.h>
# include <netinet/in.h>
# include <netinet/ip.h>
#endif
namespace caf {
namespace io {
namespace network {
// annoying platform-dependent bootstrapping
#ifdef CAF_WINDOWS
using setsockopt_ptr = const char*;
using getsockopt_ptr = char*;
using socket_send_ptr = const char*;
using socket_recv_ptr = char*;
using socklen_t = int;
using ssize_t = std::make_signed<size_t>::type;
inline int last_socket_error() { return WSAGetLastError(); }
inline bool would_block_or_temporarily_unavailable(int errcode) {
return errcode == WSAEWOULDBLOCK || errcode == WSATRY_AGAIN;
}
constexpr int ec_out_of_memory = WSAENOBUFS;
constexpr int ec_interrupted_syscall = WSAEINTR;
#else
using setsockopt_ptr = const void*;
using getsockopt_ptr = void*;
using socket_send_ptr = const void*;
using socket_recv_ptr = void*;
inline void closesocket(int fd) { close(fd); }
inline int last_socket_error() { return errno; }
inline bool would_block_or_temporarily_unavailable(int errcode) {
return errcode == EAGAIN || errcode == EWOULDBLOCK;
}
constexpr int ec_out_of_memory = ENOMEM;
constexpr int ec_interrupted_syscall = EINTR;
#endif
// platform-dependent SIGPIPE setup
#if defined(CAF_MACOS) || defined(CAF_IOS) || defined(CAF_BSD)
// Use the socket option but no flags to recv/send on macOS/iOS/BSD.
constexpr int no_sigpipe_socket_flag = SO_NOSIGPIPE;
constexpr int no_sigpipe_io_flag = 0;
#elif defined(CAF_WINDOWS)
// Do nothing on Windows (SIGPIPE does not exist).
constexpr int no_sigpipe_socket_flag = 0;
constexpr int no_sigpipe_io_flag = 0;
#else
// Use flags to recv/send on Linux/Android but no socket option.
constexpr int no_sigpipe_socket_flag = 0;
constexpr int no_sigpipe_io_flag = MSG_NOSIGNAL;
#endif
/// Returns the last socket error as human-readable string.
std::string last_socket_error_as_string();
/// Creates two connected sockets. The former is the read handle
/// and the latter is the write handle.
std::pair<native_socket, native_socket> create_pipe();
/// Sets fd to nonblocking if `set_nonblocking == true`
/// or to blocking if `set_nonblocking == false`
/// throws `network_error` on error
expected<void> nonblocking(native_socket fd, bool new_value);
/// Enables or disables Nagle's algorithm on `fd`.
/// @throws network_error
expected<void> tcp_nodelay(native_socket fd, bool new_value);
/// Enables or disables `SIGPIPE` events from `fd`.
expected<void> allow_sigpipe(native_socket fd, bool new_value);
/// Enables or disables `SIO_UDP_CONNRESET`error on `fd`.
expected<void> allow_udp_connreset(native_socket fd, bool new_value);
/// Get the socket buffer size for `fd`.
expected<int> send_buffer_size(native_socket fd);
/// Set the socket buffer size for `fd`.
expected<void> send_buffer_size(native_socket fd, int new_value);
/// Convenience functions for checking the result of `recv` or `send`.
bool is_error(ssize_t res, bool is_nonblock);
/// Returns the locally assigned port of `fd`.
expected<uint16_t> local_port_of_fd(native_socket fd);
/// Returns the locally assigned address of `fd`.
expected<std::string> local_addr_of_fd(native_socket fd);
/// Returns the port used by the remote host of `fd`.
expected<uint16_t> remote_port_of_fd(native_socket fd);
/// Returns the remote host address of `fd`.
expected<std::string> remote_addr_of_fd(native_socket fd);
/// @cond PRIVTE
// Utility functions to privde access to sockaddr fields.
auto addr_of(sockaddr_in& what) -> decltype(what.sin_addr)&;
auto family_of(sockaddr_in& what) -> decltype(what.sin_family)&;
auto port_of(sockaddr_in& what) -> decltype(what.sin_port)&;
auto addr_of(sockaddr_in6& what) -> decltype(what.sin6_addr)&;
auto family_of(sockaddr_in6& what) -> decltype(what.sin6_family)&;
auto port_of(sockaddr_in6& what) -> decltype(what.sin6_port)&;
auto port_of(sockaddr& what) -> decltype(port_of(std::declval<sockaddr_in&>()));
/// @endcond
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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 <vector>
#include "caf/logger.hpp"
#include "caf/ref_counted.hpp"
#include "caf/io/fwd.hpp"
#include "caf/io/receive_policy.hpp"
#include "caf/io/network/rw_state.hpp"
#include "caf/io/network/event_handler.hpp"
#include "caf/io/network/stream_manager.hpp"
#include "caf/io/network/event_handler.hpp"
namespace caf {
namespace io {
namespace network {
/// A stream capable of both reading and writing. The stream's input
/// data is forwarded to its {@link stream_manager manager}.
class stream : public event_handler {
public:
/// A smart pointer to a stream manager.
using manager_ptr = intrusive_ptr<stream_manager>;
/// A buffer class providing a compatible
/// interface to `std::vector`.
using buffer_type = std::vector<char>;
stream(default_multiplexer& backend_ref, native_socket sockfd);
/// Starts reading data from the socket, forwarding incoming data to `mgr`.
void start(stream_manager* mgr);
/// Activates the stream.
void activate(stream_manager* mgr);
/// Configures how much data will be provided for the next `consume` callback.
/// @warning Must not be called outside the IO multiplexers event loop
/// once the stream has been started.
void configure_read(receive_policy::config config);
void ack_writes(bool x);
/// Copies data to the write buffer.
/// @warning Not thread safe.
void write(const void* buf, size_t num_bytes);
/// Returns the write buffer of this stream.
/// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started.
inline buffer_type& wr_buf() {
return wr_offline_buf_;
}
/// Returns the read buffer of this stream.
/// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started.
inline buffer_type& rd_buf() {
return rd_buf_;
}
/// Sends the content of the write buffer, calling the `io_failure`
/// member function of `mgr` in case of an error.
/// @warning Must not be called outside the IO multiplexers event loop
/// once the stream has been started.
void flush(const manager_ptr& mgr);
/// Closes the read channel of the underlying socket and removes
/// this handler from its parent.
void stop_reading();
void removed_from_loop(operation op) override;
/// Forces this stream to subscribe to write events if no data is in the
/// write buffer.
void force_empty_write(const manager_ptr& mgr);
protected:
template <class Policy>
void handle_event_impl(io::network::operation op, Policy& policy) {
CAF_LOG_TRACE(CAF_ARG(op));
auto mcr = max_consecutive_reads_;
switch (op) {
case io::network::operation::read: {
// Loop until an error occurs or we have nothing more to read
// or until we have handled `mcr` reads.
size_t rb;
for (size_t i = 0; i < mcr; ++i) {
auto res = policy.read_some(rb, fd(), rd_buf_.data() + collected_,
rd_buf_.size() - collected_);
if (!handle_read_result(res, rb))
return;
}
break;
}
case io::network::operation::write: {
size_t wb; // Written bytes.
auto res = policy.write_some(wb, fd(), wr_buf_.data() + written_,
wr_buf_.size() - written_);
handle_write_result(res, wb);
break;
}
case operation::propagate_error:
handle_error_propagation();
break;
}
}
private:
void prepare_next_read();
void prepare_next_write();
bool handle_read_result(rw_state read_result, size_t rb);
void handle_write_result(rw_state write_result, size_t wb);
void handle_error_propagation();
size_t max_consecutive_reads_;
// State for reading.
manager_ptr reader_;
size_t read_threshold_;
size_t collected_;
size_t max_;
receive_policy_flag rd_flag_;
buffer_type rd_buf_;
// State for writing.
manager_ptr writer_;
bool ack_writes_;
bool writing_;
size_t written_;
buffer_type wr_buf_;
buffer_type wr_offline_buf_;
};
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/io/network/stream.hpp"
namespace caf {
namespace io {
namespace network {
/// A concrete stream with a technology-dependent policy for sending and
/// receiving data from a socket.
template <class ProtocolPolicy>
class stream_impl : public stream {
public:
template <class... Ts>
stream_impl(default_multiplexer& mpx, native_socket sockfd, Ts&&... xs)
: stream(mpx, sockfd),
policy_(std::forward<Ts>(xs)...) {
// nop
}
void handle_event(io::network::operation op) override {
this->handle_event_impl(op, policy_);
}
private:
ProtocolPolicy policy_;
};
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/io/network/acceptor.hpp"
#include "caf/logger.hpp"
namespace caf {
namespace io {
namespace network {
acceptor::acceptor(default_multiplexer& backend_ref, native_socket sockfd)
: event_handler(backend_ref, sockfd),
sock_(invalid_native_socket) {
// nop
}
void acceptor::start(acceptor_manager* mgr) {
CAF_LOG_TRACE(CAF_ARG2("fd", fd()));
CAF_ASSERT(mgr != nullptr);
activate(mgr);
}
void acceptor::activate(acceptor_manager* mgr) {
if (!mgr_) {
mgr_.reset(mgr);
event_handler::activate();
}
}
void acceptor::stop_reading() {
CAF_LOG_TRACE(CAF_ARG2("fd", fd()));
close_read_channel();
passivate();
}
void acceptor::removed_from_loop(operation op) {
CAF_LOG_TRACE(CAF_ARG2("fd", fd()) << CAF_ARG(op));
if (op == operation::read)
mgr_.reset();
}
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/io/network/datagram_handler.hpp"
#include <algorithm>
#include "caf/logger.hpp"
#include "caf/defaults.hpp"
#include "caf/config_value.hpp"
#include "caf/io/network/socket_utils.hpp"
#include "caf/io/network/default_multiplexer.hpp"
namespace {
constexpr size_t receive_buffer_size = std::numeric_limits<uint16_t>::max();
} // namespace anonymous
namespace caf {
namespace io {
namespace network {
datagram_handler::datagram_handler(default_multiplexer& backend_ref,
native_socket sockfd)
: event_handler(backend_ref, sockfd),
max_consecutive_reads_(
get_or(backend().system().config(), "middleman.max-consecutive-reads",
defaults::middleman::max_consecutive_reads)),
max_datagram_size_(receive_buffer_size),
rd_buf_(receive_buffer_size),
send_buffer_size_(0),
ack_writes_(false),
writing_(false) {
allow_udp_connreset(sockfd, false);
auto es = send_buffer_size(sockfd);
if (!es)
CAF_LOG_ERROR("cannot determine socket buffer size");
else
send_buffer_size_ = *es;
}
void datagram_handler::start(datagram_manager* mgr) {
CAF_LOG_TRACE(CAF_ARG2("fd", fd()));
CAF_ASSERT(mgr != nullptr);
activate(mgr);
}
void datagram_handler::activate(datagram_manager* mgr) {
if (!reader_) {
reader_.reset(mgr);
event_handler::activate();
prepare_next_read();
}
}
void datagram_handler::ack_writes(bool x) {
ack_writes_ = x;
}
void datagram_handler::write(datagram_handle hdl, const void* buf,
size_t num_bytes) {
wr_offline_buf_.emplace_back();
wr_offline_buf_.back().first = hdl;
auto cbuf = reinterpret_cast<const char*>(buf);
wr_offline_buf_.back().second.assign(cbuf,
cbuf + static_cast<ptrdiff_t>(num_bytes));
}
void datagram_handler::flush(const manager_ptr& mgr) {
CAF_ASSERT(mgr != nullptr);
CAF_LOG_TRACE(CAF_ARG(wr_offline_buf_.size()));
if (!wr_offline_buf_.empty() && !writing_) {
backend().add(operation::write, fd(), this);
writer_ = mgr;
writing_ = true;
prepare_next_write();
}
}
std::unordered_map<datagram_handle, ip_endpoint>& datagram_handler::endpoints() {
return ep_by_hdl_;
}
const std::unordered_map<datagram_handle, ip_endpoint>&
datagram_handler::endpoints() const {
return ep_by_hdl_;
}
void datagram_handler::add_endpoint(datagram_handle hdl, const ip_endpoint& ep,
const manager_ptr mgr) {
auto itr = hdl_by_ep_.find(ep);
if (itr == hdl_by_ep_.end()) {
hdl_by_ep_[ep] = hdl;
ep_by_hdl_[hdl] = ep;
writer_ = mgr;
} else if (!writer_) {
writer_ = mgr;
} else {
CAF_LOG_ERROR("cannot assign a second servant to the endpoint "
<< to_string(ep));
abort();
}
}
void datagram_handler::remove_endpoint(datagram_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
auto itr = ep_by_hdl_.find(hdl);
if (itr != ep_by_hdl_.end()) {
hdl_by_ep_.erase(itr->second);
ep_by_hdl_.erase(itr);
}
}
void datagram_handler::stop_reading() {
CAF_LOG_TRACE("");
close_read_channel();
passivate();
}
void datagram_handler::removed_from_loop(operation op) {
switch (op) {
case operation::read: reader_.reset(); break;
case operation::write: writer_.reset(); break;
case operation::propagate_error: break;
};
}
void datagram_handler::prepare_next_read() {
CAF_LOG_TRACE(CAF_ARG(wr_buf_.second.size())
<< CAF_ARG(wr_offline_buf_.size()));
rd_buf_.resize(max_datagram_size_);
}
void datagram_handler::prepare_next_write() {
CAF_LOG_TRACE(CAF_ARG(wr_offline_buf_.size()));
wr_buf_.second.clear();
if (wr_offline_buf_.empty()) {
writing_ = false;
backend().del(operation::write, fd(), this);
} else {
wr_buf_.swap(wr_offline_buf_.front());
wr_offline_buf_.pop_front();
}
}
bool datagram_handler::handle_read_result(bool read_result) {
if (!read_result) {
reader_->io_failure(&backend(), operation::read);
passivate();
return false;
}
if (num_bytes_ > 0) {
rd_buf_.resize(num_bytes_);
auto itr = hdl_by_ep_.find(sender_);
bool consumed = false;
if (itr == hdl_by_ep_.end())
consumed = reader_->new_endpoint(rd_buf_);
else
consumed = reader_->consume(&backend(), itr->second, rd_buf_);
prepare_next_read();
if (!consumed) {
passivate();
return false;
}
}
return true;
}
void datagram_handler::handle_write_result(bool write_result, datagram_handle id,
std::vector<char>& buf, size_t wb) {
if (!write_result) {
writer_->io_failure(&backend(), operation::write);
backend().del(operation::write, fd(), this);
} else if (wb > 0) {
CAF_ASSERT(wb == buf.size());
if (ack_writes_)
writer_->datagram_sent(&backend(), id, wb, std::move(buf));
prepare_next_write();
} else {
if (writer_)
writer_->io_failure(&backend(), operation::write);
}
}
void datagram_handler::handle_error() {
if (reader_)
reader_->io_failure(&backend(), operation::read);
if (writer_)
writer_->io_failure(&backend(), operation::write);
// backend will delete this handler anyway,
// no need to call backend().del() here
}
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/io/network/datagram_servant_impl.hpp"
#include <algorithm>
#include "caf/logger.hpp"
#include "caf/io/network/socket_utils.hpp"
#include "caf/io/network/default_multiplexer.hpp"
namespace caf {
namespace io {
namespace network {
datagram_servant_impl::datagram_servant_impl(default_multiplexer& mx,
native_socket sockfd, int64_t id)
: datagram_servant(datagram_handle::from_int(id)),
launched_(false),
handler_(mx, sockfd) {
// nop
}
bool datagram_servant_impl::new_endpoint(network::receive_buffer& buf) {
CAF_LOG_TRACE("");
if (detached())
// We are already disconnected from the broker while the multiplexer
// did not yet remove the socket, this can happen if an I/O event
// causes the broker to call close_all() while the pollset contained
// further activities for the broker.
return false;
// A datagram that has a source port of zero is valid and never requires a
// reply. In the case of CAF we can simply drop it as nothing but the
// handshake could be communicated which we could not reply to.
// Source: TCP/IP Illustrated, Chapter 10.2
if (network::port(handler_.sending_endpoint()) == 0)
return true;
auto& dm = handler_.backend();
auto hdl = datagram_handle::from_int(dm.next_endpoint_id());
add_endpoint(handler_.sending_endpoint(), hdl);
parent()->add_hdl_for_datagram_servant(this, hdl);
return consume(&dm, hdl, buf);
}
void datagram_servant_impl::ack_writes(bool enable) {
CAF_LOG_TRACE(CAF_ARG(enable));
handler_.ack_writes(enable);
}
std::vector<char>& datagram_servant_impl::wr_buf(datagram_handle hdl) {
return handler_.wr_buf(hdl);
}
void datagram_servant_impl::enqueue_datagram(datagram_handle hdl,
std::vector<char> buffer) {
handler_.enqueue_datagram(hdl, std::move(buffer));
}
network::receive_buffer& datagram_servant_impl::rd_buf() {
return handler_.rd_buf();
}
void datagram_servant_impl::stop_reading() {
CAF_LOG_TRACE("");
handler_.stop_reading();
detach_handles();
detach(&handler_.backend(), false);
}
void datagram_servant_impl::flush() {
CAF_LOG_TRACE("");
handler_.flush(this);
}
std::string datagram_servant_impl::addr() const {
auto x = remote_addr_of_fd(handler_.fd());
if (!x)
return "";
return *x;
}
uint16_t datagram_servant_impl::port(datagram_handle hdl) const {
auto& eps = handler_.endpoints();
auto itr = eps.find(hdl);
if (itr == eps.end())
return 0;
return network::port(itr->second);
}
uint16_t datagram_servant_impl::local_port() const {
auto x = local_port_of_fd(handler_.fd());
if (!x)
return 0;
return *x;
}
std::vector<datagram_handle> datagram_servant_impl::hdls() const {
std::vector<datagram_handle> result;
result.reserve(handler_.endpoints().size());
for (auto& p : handler_.endpoints())
result.push_back(p.first);
return result;
}
void datagram_servant_impl::add_endpoint(const ip_endpoint& ep,
datagram_handle hdl) {
handler_.add_endpoint(hdl, ep, this);
}
void datagram_servant_impl::remove_endpoint(datagram_handle hdl) {
handler_.remove_endpoint(hdl);
}
void datagram_servant_impl::launch() {
CAF_LOG_TRACE("");
CAF_ASSERT(!launched_);
launched_ = true;
handler_.start(this);
}
void datagram_servant_impl::add_to_loop() {
handler_.activate(this);
}
void datagram_servant_impl::remove_from_loop() {
handler_.passivate();
}
void datagram_servant_impl::detach_handles() {
for (auto& p : handler_.endpoints()) {
if (p.first != hdl())
parent()->erase(p.first);
}
}
} // namespace network
} // namespace io
} // namespace caf
...@@ -18,15 +18,26 @@ ...@@ -18,15 +18,26 @@
#include "caf/io/network/default_multiplexer.hpp" #include "caf/io/network/default_multiplexer.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/defaults.hpp" #include "caf/defaults.hpp"
#include "caf/optional.hpp"
#include "caf/make_counted.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/io/broker.hpp" #include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
#include "caf/io/network/interfaces.hpp" #include "caf/io/network/stream.hpp"
#include "caf/io/network/acceptor.hpp"
#include "caf/io/network/protocol.hpp" #include "caf/io/network/protocol.hpp"
#include "caf/make_counted.hpp" #include "caf/io/network/interfaces.hpp"
#include "caf/optional.hpp" #include "caf/io/network/pipe_reader.hpp"
#include "caf/io/network/scribe_impl.hpp"
#include "caf/io/network/doorman_impl.hpp"
#include "caf/io/network/datagram_handler.hpp"
#include "caf/io/network/datagram_servant_impl.hpp"
#include "caf/io/network/policy/tcp.hpp"
#include "caf/io/network/policy/udp.hpp"
#include "caf/scheduler/abstract_coordinator.hpp" #include "caf/scheduler/abstract_coordinator.hpp"
#ifdef CAF_WINDOWS #ifdef CAF_WINDOWS
...@@ -48,12 +59,8 @@ ...@@ -48,12 +59,8 @@
using std::string; using std::string;
// -- Utiliy functions for converting errno into CAF errors --------------------
namespace { namespace {
constexpr size_t receive_buffer_size = std::numeric_limits<uint16_t>::max();
// safe ourselves some typing // safe ourselves some typing
constexpr auto ipv4 = caf::io::network::protocol::ipv4; constexpr auto ipv4 = caf::io::network::protocol::ipv4;
constexpr auto ipv6 = caf::io::network::protocol::ipv6; constexpr auto ipv6 = caf::io::network::protocol::ipv6;
...@@ -68,11 +75,6 @@ bool cc_one(int value) { ...@@ -68,11 +75,6 @@ bool cc_one(int value) {
return value == 1; return value == 1;
} }
// predicate for `ccall` meaning "expected result of f is not -1"
bool cc_not_minus1(int value) {
return value != -1;
}
// predicate for `ccall` meaning "expected result of f is a valid socket" // predicate for `ccall` meaning "expected result of f is a valid socket"
bool cc_valid_socket(caf::io::network::native_socket fd) { bool cc_valid_socket(caf::io::network::native_socket fd) {
return fd != caf::io::network::invalid_native_socket; return fd != caf::io::network::invalid_native_socket;
...@@ -106,180 +108,6 @@ namespace caf { ...@@ -106,180 +108,6 @@ namespace caf {
namespace io { namespace io {
namespace network { namespace network {
// -- OS-specific functions for sockets and pipes ------------------------------
#ifndef CAF_WINDOWS
string last_socket_error_as_string() {
return strerror(errno);
}
expected<void> nonblocking(native_socket fd, bool new_value) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(new_value));
// read flags for fd
CALL_CFUN(rf, cc_not_minus1, "fcntl", fcntl(fd, F_GETFL, 0));
// calculate and set new flags
auto wf = new_value ? (rf | O_NONBLOCK) : (rf & (~(O_NONBLOCK)));
CALL_CFUN(set_res, cc_not_minus1, "fcntl", fcntl(fd, F_SETFL, wf));
return unit;
}
expected<void> allow_sigpipe(native_socket fd, bool new_value) {
if (no_sigpipe_socket_flag != 0) {
int value = new_value ? 0 : 1;
CALL_CFUN(res, cc_zero, "setsockopt",
setsockopt(fd, SOL_SOCKET, no_sigpipe_socket_flag, &value,
static_cast<unsigned>(sizeof(value))));
}
return unit;
}
expected<void> allow_udp_connreset(native_socket, bool) {
// nop; SIO_UDP_CONNRESET only exists on Windows
return unit;
}
std::pair<native_socket, native_socket> create_pipe() {
int pipefds[2];
if (pipe(pipefds) != 0) {
perror("pipe");
exit(EXIT_FAILURE);
}
return {pipefds[0], pipefds[1]};
}
#else // CAF_WINDOWS
string last_socket_error_as_string() {
LPTSTR errorText = NULL;
auto hresult = last_socket_error();
FormatMessage( // use system message tables to retrieve error text
FORMAT_MESSAGE_FROM_SYSTEM
// allocate buffer on local heap for error text
| FORMAT_MESSAGE_ALLOCATE_BUFFER
// Important! will fail otherwise, since we're not
// (and CANNOT) pass insertion parameters
| FORMAT_MESSAGE_IGNORE_INSERTS,
nullptr, // unused with FORMAT_MESSAGE_FROM_SYSTEM
hresult, MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
(LPTSTR) & errorText, // output
0, // minimum size for output buffer
nullptr); // arguments - see note
std::string result;
if (errorText != nullptr) {
result = errorText;
// release memory allocated by FormatMessage()
LocalFree(errorText);
}
return result;
}
expected<void> nonblocking(native_socket fd, bool new_value) {
u_long mode = new_value ? 1 : 0;
CALL_CFUN(res, cc_zero, "ioctlsocket", ioctlsocket(fd, FIONBIO, &mode));
return unit;
}
expected<void> allow_sigpipe(native_socket, bool) {
// nop; SIGPIPE does not exist on Windows
return unit;
}
expected<void> allow_udp_connreset(native_socket fd, bool new_value) {
DWORD bytes_returned = 0;
CALL_CFUN(res, cc_zero, "WSAIoctl",
WSAIoctl(fd, SIO_UDP_CONNRESET, &new_value, sizeof(new_value),
NULL, 0, &bytes_returned, NULL, NULL));
return unit;
}
/**************************************************************************\
* 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. *
\**************************************************************************/
std::pair<native_socket, native_socket> create_pipe() {
socklen_t addrlen = sizeof(sockaddr_in);
native_socket socks[2] = {invalid_native_socket, invalid_native_socket};
CALL_CRITICAL_CFUN(listener, cc_valid_socket, "socket",
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();
closesocket(listener);
closesocket(socks[0]);
closesocket(socks[1]);
WSASetLastError(e);
});
// bind listener to a local port
int reuse = 1;
CALL_CRITICAL_CFUN(tmp1, cc_zero, "setsockopt",
setsockopt(listener, SOL_SOCKET, SO_REUSEADDR,
reinterpret_cast<char*>(&reuse),
static_cast<int>(sizeof(reuse))));
CALL_CRITICAL_CFUN(tmp2, cc_zero, "bind",
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));
CALL_CRITICAL_CFUN(tmp3, cc_zero, "getsockname",
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
CALL_CRITICAL_CFUN(tmp5, cc_zero, "listen", listen(listener, 1));
// create read-only end of the pipe
DWORD flags = 0;
CALL_CRITICAL_CFUN(read_fd, cc_valid_socket, "WSASocketW",
WSASocketW(AF_INET, SOCK_STREAM, 0, nullptr, 0, flags));
CALL_CRITICAL_CFUN(tmp6, cc_zero, "connect",
connect(read_fd, &a.addr,
static_cast<int>(sizeof(a.inaddr))));
// get write-only end of the pipe
CALL_CRITICAL_CFUN(write_fd, cc_valid_socket, "accept",
accept(listener, nullptr, nullptr));
closesocket(listener);
guard.disable();
return std::make_pair(read_fd, write_fd);
}
#endif
// -- Platform-dependent abstraction over epoll() or poll() -------------------- // -- Platform-dependent abstraction over epoll() or poll() --------------------
#ifdef CAF_EPOLL_MULTIPLEXER #ifdef CAF_EPOLL_MULTIPLEXER
...@@ -597,60 +425,8 @@ namespace network { ...@@ -597,60 +425,8 @@ namespace network {
} }
#endif // CAF_EPOLL_MULTIPLEXER #endif // CAF_EPOLL_MULTIPLEXER
// -- Helper functions for defining bitmasks of event handlers ----------------- // -- Free functions for sending and receiving data
int add_flag(operation op, int bf) {
switch (op) {
case operation::read:
return bf | input_mask;
case operation::write:
return bf | output_mask;
case operation::propagate_error:
CAF_LOG_ERROR("unexpected operation");
break;
}
// weird stuff going on
return 0;
}
int del_flag(operation op, int bf) {
switch (op) {
case operation::read:
return bf & ~input_mask;
case operation::write:
return bf & ~output_mask;
case operation::propagate_error:
CAF_LOG_ERROR("unexpected operation");
break;
}
// weird stuff going on
return 0;
}
// -- Platform-independent free functions --------------------------------------
expected<void> tcp_nodelay(native_socket fd, bool new_value) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(new_value));
int flag = new_value ? 1 : 0;
CALL_CFUN(res, cc_zero, "setsockopt",
setsockopt(fd, IPPROTO_TCP, TCP_NODELAY,
reinterpret_cast<setsockopt_ptr>(&flag),
static_cast<socklen_t>(sizeof(flag))));
return unit;
}
bool is_error(ssize_t res, bool is_nonblock) {
if (res < 0) {
auto err = last_socket_error();
if (!is_nonblock || !would_block_or_temporarily_unavailable(err)) {
return true;
}
// don't report an error in case of
// spurious wakeup or something similar
}
return false;
}
rw_state read_some(size_t& result, native_socket fd, void* buf, size_t len) { rw_state read_some(size_t& result, native_socket fd, void* buf, size_t len) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(len)); CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(len));
...@@ -677,7 +453,7 @@ rw_state write_some(size_t& result, native_socket fd, const void* buf, ...@@ -677,7 +453,7 @@ rw_state write_some(size_t& result, native_socket fd, const void* buf,
return rw_state::success; return rw_state::success;
} }
bool try_accept(native_socket& result, native_socket fd) { bool try_accept(native_socket& result, native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd)); CAF_LOG_TRACE(CAF_ARG(fd));
sockaddr_storage addr; sockaddr_storage addr;
memset(&addr, 0, sizeof(addr)); memset(&addr, 0, sizeof(addr));
...@@ -728,20 +504,36 @@ bool write_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len, ...@@ -728,20 +504,36 @@ bool write_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len,
return true; return true;
} }
// -- Policy class for TCP wrapping above free functions ----------------------- // -- Helper functions for defining bitmasks of event handlers -----------------
read_some_fun tcp_policy::read_some = network::read_some;
write_some_fun tcp_policy::write_some = network::write_some;
try_accept_fun tcp_policy::try_accept = network::try_accept;
// -- Policy class for UDP wrappign above free functions -----------------------
read_datagram_fun udp_policy::read_datagram = network::read_datagram;
write_datagram_fun udp_policy::write_datagram = network::write_datagram; int add_flag(operation op, int bf) {
switch (op) {
case operation::read:
return bf | input_mask;
case operation::write:
return bf | output_mask;
case operation::propagate_error:
CAF_LOG_ERROR("unexpected operation");
break;
}
// weird stuff going on
return 0;
}
int del_flag(operation op, int bf) {
switch (op) {
case operation::read:
return bf & ~input_mask;
case operation::write:
return bf & ~output_mask;
case operation::propagate_error:
CAF_LOG_ERROR("unexpected operation");
break;
}
// weird stuff going on
return 0;
}
// -- Platform-independent parts of the default_multiplexer -------------------- // -- Platform-independent parts of the default_multiplexer --------------------
bool default_multiplexer::try_run_once() { bool default_multiplexer::try_run_once() {
...@@ -990,342 +782,6 @@ int64_t default_multiplexer::next_endpoint_id() { ...@@ -990,342 +782,6 @@ int64_t default_multiplexer::next_endpoint_id() {
return servant_ids_++; return servant_ids_++;
} }
event_handler::event_handler(default_multiplexer& dm, native_socket sockfd)
: eventbf_(0),
fd_(sockfd),
read_channel_closed_(false),
backend_(dm) {
set_fd_flags();
}
event_handler::~event_handler() {
if (fd_ != invalid_native_socket) {
CAF_LOG_DEBUG("close socket" << CAF_ARG(fd_));
closesocket(fd_);
}
}
void event_handler::close_read_channel() {
if (fd_ == invalid_native_socket || read_channel_closed_)
return;
::shutdown(fd_, 0); // 0 identifies the read channel on Win & UNIX
read_channel_closed_ = true;
}
void event_handler::passivate() {
backend().del(operation::read, fd(), this);
}
void event_handler::activate() {
backend().add(operation::read, fd(), this);
}
void event_handler::set_fd_flags() {
if (fd_ == invalid_native_socket)
return;
// enable nonblocking IO, disable Nagle's algorithm, and suppress SIGPIPE
nonblocking(fd_, true);
tcp_nodelay(fd_, true);
allow_sigpipe(fd_, false);
}
pipe_reader::pipe_reader(default_multiplexer& dm)
: event_handler(dm, invalid_native_socket) {
// nop
}
void pipe_reader::removed_from_loop(operation) {
// nop
}
resumable* pipe_reader::try_read_next() {
intptr_t ptrval;
// on windows, we actually have sockets, otherwise we have file handles
# ifdef CAF_WINDOWS
auto res = recv(fd(), reinterpret_cast<socket_recv_ptr>(&ptrval),
sizeof(ptrval), 0);
# else
auto res = read(fd(), &ptrval, sizeof(ptrval));
# endif
if (res != sizeof(ptrval))
return nullptr;
return reinterpret_cast<resumable*>(ptrval);
}
void pipe_reader::handle_event(operation op) {
CAF_LOG_TRACE(CAF_ARG(op));
if (op == operation::read) {
auto ptr = try_read_next();
if (ptr != nullptr)
backend().resume({ptr, false});
}
// else: ignore errors
}
void pipe_reader::init(native_socket sock_fd) {
fd_ = sock_fd;
}
stream::stream(default_multiplexer& backend_ref, native_socket sockfd)
: event_handler(backend_ref, sockfd),
max_consecutive_reads_(
get_or(backend().system().config(), "middleman.max-consecutive-reads",
defaults::middleman::max_consecutive_reads)),
read_threshold_(1),
collected_(0),
ack_writes_(false),
writing_(false),
written_(0) {
configure_read(receive_policy::at_most(1024));
}
void stream::start(stream_manager* mgr) {
CAF_ASSERT(mgr != nullptr);
activate(mgr);
}
void stream::activate(stream_manager* mgr) {
if (!reader_) {
reader_.reset(mgr);
event_handler::activate();
prepare_next_read();
}
}
void stream::configure_read(receive_policy::config config) {
rd_flag_ = config.first;
max_ = config.second;
}
void stream::ack_writes(bool x) {
ack_writes_ = x;
}
void stream::write(const void* buf, size_t num_bytes) {
CAF_LOG_TRACE(CAF_ARG(num_bytes));
auto first = reinterpret_cast<const char*>(buf);
auto last = first + num_bytes;
wr_offline_buf_.insert(wr_offline_buf_.end(), first, last);
}
void stream::flush(const manager_ptr& mgr) {
CAF_ASSERT(mgr != nullptr);
CAF_LOG_TRACE(CAF_ARG(wr_offline_buf_.size()));
if (!wr_offline_buf_.empty() && !writing_) {
backend().add(operation::write, fd(), this);
writer_ = mgr;
writing_ = true;
prepare_next_write();
}
}
void stream::stop_reading() {
CAF_LOG_TRACE("");
close_read_channel();
passivate();
}
void stream::removed_from_loop(operation op) {
CAF_LOG_TRACE(CAF_ARG(op));
switch (op) {
case operation::read: reader_.reset(); break;
case operation::write: writer_.reset(); break;
case operation::propagate_error: break;
}
}
void stream::prepare_next_read() {
collected_ = 0;
switch (rd_flag_) {
case receive_policy_flag::exactly:
if (rd_buf_.size() != max_)
rd_buf_.resize(max_);
read_threshold_ = max_;
break;
case receive_policy_flag::at_most:
if (rd_buf_.size() != max_)
rd_buf_.resize(max_);
read_threshold_ = 1;
break;
case receive_policy_flag::at_least: {
// read up to 10% more, but at least allow 100 bytes more
auto max_size = max_ + std::max<size_t>(100, max_ / 10);
if (rd_buf_.size() != max_size)
rd_buf_.resize(max_size);
read_threshold_ = max_;
break;
}
}
}
void stream::prepare_next_write() {
CAF_LOG_TRACE(CAF_ARG(wr_buf_.size()) << CAF_ARG(wr_offline_buf_.size()));
written_ = 0;
wr_buf_.clear();
if (wr_offline_buf_.empty()) {
writing_ = false;
backend().del(operation::write, fd(), this);
} else {
wr_buf_.swap(wr_offline_buf_);
}
}
acceptor::acceptor(default_multiplexer& backend_ref, native_socket sockfd)
: event_handler(backend_ref, sockfd),
sock_(invalid_native_socket) {
// nop
}
void acceptor::start(acceptor_manager* mgr) {
CAF_LOG_TRACE(CAF_ARG2("fd", fd()));
CAF_ASSERT(mgr != nullptr);
activate(mgr);
}
void acceptor::activate(acceptor_manager* mgr) {
if (!mgr_) {
mgr_.reset(mgr);
event_handler::activate();
}
}
void acceptor::stop_reading() {
CAF_LOG_TRACE(CAF_ARG2("fd", fd()));
close_read_channel();
passivate();
}
void acceptor::removed_from_loop(operation op) {
CAF_LOG_TRACE(CAF_ARG2("fd", fd()) << CAF_ARG(op));
if (op == operation::read)
mgr_.reset();
}
datagram_handler::datagram_handler(default_multiplexer& backend_ref,
native_socket sockfd)
: event_handler(backend_ref, sockfd),
max_consecutive_reads_(
get_or(backend().system().config(), "middleman.max-consecutive-reads",
defaults::middleman::max_consecutive_reads)),
max_datagram_size_(receive_buffer_size),
rd_buf_(receive_buffer_size),
send_buffer_size_(0),
ack_writes_(false),
writing_(false) {
allow_udp_connreset(sockfd, false);
auto es = send_buffer_size(sockfd);
if (!es)
CAF_LOG_ERROR("cannot determine socket buffer size");
else
send_buffer_size_ = *es;
}
void datagram_handler::start(datagram_manager* mgr) {
CAF_LOG_TRACE(CAF_ARG2("fd", fd()));
CAF_ASSERT(mgr != nullptr);
activate(mgr);
}
void datagram_handler::activate(datagram_manager* mgr) {
if (!reader_) {
reader_.reset(mgr);
event_handler::activate();
prepare_next_read();
}
}
void datagram_handler::ack_writes(bool x) {
ack_writes_ = x;
}
void datagram_handler::write(datagram_handle hdl, const void* buf,
size_t num_bytes) {
wr_offline_buf_.emplace_back();
wr_offline_buf_.back().first = hdl;
auto cbuf = reinterpret_cast<const char*>(buf);
wr_offline_buf_.back().second.assign(cbuf,
cbuf + static_cast<ptrdiff_t>(num_bytes));
}
void datagram_handler::flush(const manager_ptr& mgr) {
CAF_ASSERT(mgr != nullptr);
CAF_LOG_TRACE(CAF_ARG(wr_offline_buf_.size()));
if (!wr_offline_buf_.empty() && !writing_) {
backend().add(operation::write, fd(), this);
writer_ = mgr;
writing_ = true;
prepare_next_write();
}
}
std::unordered_map<datagram_handle, ip_endpoint>& datagram_handler::endpoints() {
return ep_by_hdl_;
}
const std::unordered_map<datagram_handle, ip_endpoint>&
datagram_handler::endpoints() const {
return ep_by_hdl_;
}
void datagram_handler::add_endpoint(datagram_handle hdl, const ip_endpoint& ep,
const manager_ptr mgr) {
auto itr = hdl_by_ep_.find(ep);
if (itr == hdl_by_ep_.end()) {
hdl_by_ep_[ep] = hdl;
ep_by_hdl_[hdl] = ep;
writer_ = mgr;
} else if (!writer_) {
writer_ = mgr;
} else {
CAF_LOG_ERROR("cannot assign a second servant to the endpoint "
<< to_string(ep));
abort();
}
}
void datagram_handler::remove_endpoint(datagram_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
auto itr = ep_by_hdl_.find(hdl);
if (itr != ep_by_hdl_.end()) {
hdl_by_ep_.erase(itr->second);
ep_by_hdl_.erase(itr);
}
}
void datagram_handler::stop_reading() {
CAF_LOG_TRACE("");
close_read_channel();
passivate();
}
void datagram_handler::removed_from_loop(operation op) {
switch (op) {
case operation::read: reader_.reset(); break;
case operation::write: writer_.reset(); break;
case operation::propagate_error: break;
};
}
void datagram_handler::prepare_next_read() {
CAF_LOG_TRACE(CAF_ARG(wr_buf_.second.size())
<< CAF_ARG(wr_offline_buf_.size()));
rd_buf_.resize(max_datagram_size_);
}
void datagram_handler::prepare_next_write() {
CAF_LOG_TRACE(CAF_ARG(wr_offline_buf_.size()));
wr_buf_.second.clear();
if (wr_offline_buf_.empty()) {
writing_ = false;
backend().del(operation::write, fd(), this);
} else {
wr_buf_.swap(wr_offline_buf_.front());
wr_offline_buf_.pop_front();
}
}
class socket_guard { class socket_guard {
public: public:
explicit socket_guard(native_socket fd) : fd_(fd) { explicit socket_guard(native_socket fd) : fd_(fd) {
...@@ -1354,42 +810,6 @@ private: ...@@ -1354,42 +810,6 @@ private:
native_socket fd_; native_socket fd_;
}; };
auto addr_of(sockaddr_in& what) -> decltype(what.sin_addr)& {
return what.sin_addr;
}
auto family_of(sockaddr_in& what) -> decltype(what.sin_family)& {
return what.sin_family;
}
auto port_of(sockaddr_in& what) -> decltype(what.sin_port)& {
return what.sin_port;
}
auto addr_of(sockaddr_in6& what) -> decltype(what.sin6_addr)& {
return what.sin6_addr;
}
auto family_of(sockaddr_in6& what) -> decltype(what.sin6_family)& {
return what.sin6_family;
}
auto port_of(sockaddr_in6& what) -> decltype(what.sin6_port)& {
return what.sin6_port;
}
auto port_of(sockaddr& what) -> decltype(port_of(std::declval<sockaddr_in&>())) {
switch (what.sa_family) {
case AF_INET:
return port_of(reinterpret_cast<sockaddr_in&>(what));
case AF_INET6:
return port_of(reinterpret_cast<sockaddr_in6&>(what));
default:
break;
}
CAF_CRITICAL("invalid protocol family");
}
template <int Family> template <int Family>
bool ip_connect(native_socket fd, const std::string& host, uint16_t port) { bool ip_connect(native_socket fd, const std::string& host, uint16_t port) {
CAF_LOG_TRACE("Family =" << (Family == AF_INET ? "AF_INET" : "AF_INET6") CAF_LOG_TRACE("Family =" << (Family == AF_INET ? "AF_INET" : "AF_INET6")
...@@ -1466,23 +886,6 @@ expected<void> set_inaddr_any(native_socket fd, sockaddr_in6& sa) { ...@@ -1466,23 +886,6 @@ expected<void> set_inaddr_any(native_socket fd, sockaddr_in6& sa) {
return unit; return unit;
} }
expected<int> send_buffer_size(native_socket fd) {
int size;
socklen_t ret_size = sizeof(size);
CALL_CFUN(res, cc_zero, "getsockopt",
getsockopt(fd, SOL_SOCKET, SO_SNDBUF,
reinterpret_cast<getsockopt_ptr>(&size), &ret_size));
return size;
}
expected<void> send_buffer_size(native_socket fd, int new_value) {
CALL_CFUN(res, cc_zero, "setsockopt",
setsockopt(fd, SOL_SOCKET, SO_SNDBUF,
reinterpret_cast<setsockopt_ptr>(&new_value),
static_cast<socklen_t>(sizeof(int))));
return unit;
}
template <int Family, int SockType = SOCK_STREAM> template <int Family, int SockType = SOCK_STREAM>
expected<native_socket> new_ip_acceptor_impl(uint16_t port, const char* addr, expected<native_socket> new_ip_acceptor_impl(uint16_t port, const char* addr,
bool reuse_addr, bool any) { bool reuse_addr, bool any) {
...@@ -1604,308 +1007,6 @@ new_local_udp_endpoint_impl(uint16_t port, const char* addr, bool reuse, ...@@ -1604,308 +1007,6 @@ new_local_udp_endpoint_impl(uint16_t port, const char* addr, bool reuse,
return std::make_pair(fd, proto); return std::make_pair(fd, proto);
} }
expected<std::string> local_addr_of_fd(native_socket fd) {
sockaddr_storage st;
socklen_t st_len = sizeof(st);
sockaddr* sa = reinterpret_cast<sockaddr*>(&st);
CALL_CFUN(tmp1, cc_zero, "getsockname", getsockname(fd, sa, &st_len));
char addr[INET6_ADDRSTRLEN] {0};
switch (sa->sa_family) {
case AF_INET:
return inet_ntop(AF_INET, &reinterpret_cast<sockaddr_in*>(sa)->sin_addr,
addr, sizeof(addr));
case AF_INET6:
return inet_ntop(AF_INET6,
&reinterpret_cast<sockaddr_in6*>(sa)->sin6_addr,
addr, sizeof(addr));
default:
break;
}
return make_error(sec::invalid_protocol_family,
"local_addr_of_fd", sa->sa_family);
}
expected<uint16_t> local_port_of_fd(native_socket fd) {
sockaddr_storage st;
socklen_t st_len = sizeof(st);
CALL_CFUN(tmp, cc_zero, "getsockname",
getsockname(fd, reinterpret_cast<sockaddr*>(&st), &st_len));
return ntohs(port_of(reinterpret_cast<sockaddr&>(st)));
}
expected<std::string> remote_addr_of_fd(native_socket fd) {
sockaddr_storage st;
socklen_t st_len = sizeof(st);
sockaddr* sa = reinterpret_cast<sockaddr*>(&st);
CALL_CFUN(tmp, cc_zero, "getpeername", getpeername(fd, sa, &st_len));
char addr[INET6_ADDRSTRLEN] {0};
switch (sa->sa_family) {
case AF_INET:
return inet_ntop(AF_INET, &reinterpret_cast<sockaddr_in*>(sa)->sin_addr,
addr, sizeof(addr));
case AF_INET6:
return inet_ntop(AF_INET6,
&reinterpret_cast<sockaddr_in6*>(sa)->sin6_addr,
addr, sizeof(addr));
default:
break;
}
return make_error(sec::invalid_protocol_family,
"remote_addr_of_fd", sa->sa_family);
}
expected<uint16_t> remote_port_of_fd(native_socket fd) {
sockaddr_storage st;
socklen_t st_len = sizeof(st);
CALL_CFUN(tmp, cc_zero, "getpeername",
getpeername(fd, reinterpret_cast<sockaddr*>(&st), &st_len));
return ntohs(port_of(reinterpret_cast<sockaddr&>(st)));
}
// -- default doorman and scribe implementations -------------------------------
doorman_impl::doorman_impl(default_multiplexer& mx, native_socket sockfd)
: doorman(network::accept_hdl_from_socket(sockfd)),
acceptor_(mx, sockfd) {
// nop
}
bool doorman_impl::new_connection() {
CAF_LOG_TRACE("");
if (detached())
// we are already disconnected from the broker while the multiplexer
// did not yet remove the socket, this can happen if an I/O event causes
// the broker to call close_all() while the pollset contained
// further activities for the broker
return false;
auto& dm = acceptor_.backend();
auto sptr = dm.new_scribe(acceptor_.accepted_socket());
auto hdl = sptr->hdl();
parent()->add_scribe(std::move(sptr));
return doorman::new_connection(&dm, hdl);
}
void doorman_impl::stop_reading() {
CAF_LOG_TRACE("");
acceptor_.stop_reading();
detach(&acceptor_.backend(), false);
}
void doorman_impl::launch() {
CAF_LOG_TRACE("");
acceptor_.start(this);
}
std::string doorman_impl::addr() const {
auto x = local_addr_of_fd(acceptor_.fd());
if (!x)
return "";
return std::move(*x);
}
uint16_t doorman_impl::port() const {
auto x = local_port_of_fd(acceptor_.fd());
if (!x)
return 0;
return *x;
}
void doorman_impl::add_to_loop() {
acceptor_.activate(this);
}
void doorman_impl::remove_from_loop() {
acceptor_.passivate();
}
scribe_impl::scribe_impl(default_multiplexer& mx, native_socket sockfd)
: scribe(network::conn_hdl_from_socket(sockfd)),
launched_(false),
stream_(mx, sockfd) {
// nop
}
void scribe_impl::configure_read(receive_policy::config config) {
CAF_LOG_TRACE("");
stream_.configure_read(config);
if (!launched_)
launch();
}
void scribe_impl::ack_writes(bool enable) {
CAF_LOG_TRACE(CAF_ARG(enable));
stream_.ack_writes(enable);
}
std::vector<char>& scribe_impl::wr_buf() {
return stream_.wr_buf();
}
std::vector<char>& scribe_impl::rd_buf() {
return stream_.rd_buf();
}
void scribe_impl::stop_reading() {
CAF_LOG_TRACE("");
stream_.stop_reading();
detach(&stream_.backend(), false);
}
void scribe_impl::flush() {
CAF_LOG_TRACE("");
stream_.flush(this);
}
std::string scribe_impl::addr() const {
auto x = remote_addr_of_fd(stream_.fd());
if (!x)
return "";
return *x;
}
uint16_t scribe_impl::port() const {
auto x = remote_port_of_fd(stream_.fd());
if (!x)
return 0;
return *x;
}
void scribe_impl::launch() {
CAF_LOG_TRACE("");
CAF_ASSERT(!launched_);
launched_ = true;
stream_.start(this);
}
void scribe_impl::add_to_loop() {
stream_.activate(this);
}
void scribe_impl::remove_from_loop() {
stream_.passivate();
}
datagram_servant_impl::datagram_servant_impl(default_multiplexer& mx,
native_socket sockfd, int64_t id)
: datagram_servant(datagram_handle::from_int(id)),
launched_(false),
handler_(mx, sockfd) {
// nop
}
bool datagram_servant_impl::new_endpoint(network::receive_buffer& buf) {
CAF_LOG_TRACE("");
if (detached())
// we are already disconnected from the broker while the multiplexer
// did not yet remove the socket, this can happen if an I/O event
// causes the broker to call close_all() while the pollset contained
// further activities for the broker
return false;
// A datagram that has a source port of zero is valid and never requires a
// reply. In the case of CAF we can simply drop it as nothing but the
// handshake could be communicated which we could not reply to.
// Source: TCP/IP Illustrated, Chapter 10.2
if (network::port(handler_.sending_endpoint()) == 0)
return true;
auto& dm = handler_.backend();
auto hdl = datagram_handle::from_int(dm.next_endpoint_id());
add_endpoint(handler_.sending_endpoint(), hdl);
parent()->add_hdl_for_datagram_servant(this, hdl);
return consume(&dm, hdl, buf);
}
void datagram_servant_impl::ack_writes(bool enable) {
CAF_LOG_TRACE(CAF_ARG(enable));
handler_.ack_writes(enable);
}
std::vector<char>& datagram_servant_impl::wr_buf(datagram_handle hdl) {
return handler_.wr_buf(hdl);
}
void datagram_servant_impl::enqueue_datagram(datagram_handle hdl,
std::vector<char> buffer) {
handler_.enqueue_datagram(hdl, std::move(buffer));
}
network::receive_buffer& datagram_servant_impl::rd_buf() {
return handler_.rd_buf();
}
void datagram_servant_impl::stop_reading() {
CAF_LOG_TRACE("");
handler_.stop_reading();
detach_handles();
detach(&handler_.backend(), false);
}
void datagram_servant_impl::flush() {
CAF_LOG_TRACE("");
handler_.flush(this);
}
std::string datagram_servant_impl::addr() const {
auto x = remote_addr_of_fd(handler_.fd());
if (!x)
return "";
return *x;
}
uint16_t datagram_servant_impl::port(datagram_handle hdl) const {
auto& eps = handler_.endpoints();
auto itr = eps.find(hdl);
if (itr == eps.end())
return 0;
return network::port(itr->second);
}
uint16_t datagram_servant_impl::local_port() const {
auto x = local_port_of_fd(handler_.fd());
if (!x)
return 0;
return *x;
}
std::vector<datagram_handle> datagram_servant_impl::hdls() const {
std::vector<datagram_handle> result;
result.reserve(handler_.endpoints().size());
for (auto& p : handler_.endpoints())
result.push_back(p.first);
return result;
}
void datagram_servant_impl::add_endpoint(const ip_endpoint& ep,
datagram_handle hdl) {
handler_.add_endpoint(hdl, ep, this);
}
void datagram_servant_impl::remove_endpoint(datagram_handle hdl) {
handler_.remove_endpoint(hdl);
}
void datagram_servant_impl::launch() {
CAF_LOG_TRACE("");
CAF_ASSERT(!launched_);
launched_ = true;
handler_.start(this);
}
void datagram_servant_impl::add_to_loop() {
handler_.activate(this);
}
void datagram_servant_impl::remove_from_loop() {
handler_.passivate();
}
void datagram_servant_impl::detach_handles() {
for (auto& p : handler_.endpoints()) {
if (p.first != hdl())
parent()->erase(p.first);
}
}
} // namespace network } // namespace network
} // namespace io } // namespace io
} // namespace caf } // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/io/network/doorman_impl.hpp"
#include <algorithm>
#include "caf/logger.hpp"
#include "caf/io/network/socket_utils.hpp"
#include "caf/io/network/default_multiplexer.hpp"
namespace caf {
namespace io {
namespace network {
doorman_impl::doorman_impl(default_multiplexer& mx, native_socket sockfd)
: doorman(network::accept_hdl_from_socket(sockfd)),
acceptor_(mx, sockfd) {
// nop
}
bool doorman_impl::new_connection() {
CAF_LOG_TRACE("");
if (detached())
// we are already disconnected from the broker while the multiplexer
// did not yet remove the socket, this can happen if an I/O event causes
// the broker to call close_all() while the pollset contained
// further activities for the broker
return false;
auto& dm = acceptor_.backend();
auto sptr = dm.new_scribe(acceptor_.accepted_socket());
auto hdl = sptr->hdl();
parent()->add_scribe(std::move(sptr));
return doorman::new_connection(&dm, hdl);
}
void doorman_impl::stop_reading() {
CAF_LOG_TRACE("");
acceptor_.stop_reading();
detach(&acceptor_.backend(), false);
}
void doorman_impl::launch() {
CAF_LOG_TRACE("");
acceptor_.start(this);
}
std::string doorman_impl::addr() const {
auto x = local_addr_of_fd(acceptor_.fd());
if (!x)
return "";
return std::move(*x);
}
uint16_t doorman_impl::port() const {
auto x = local_port_of_fd(acceptor_.fd());
if (!x)
return 0;
return *x;
}
void doorman_impl::add_to_loop() {
acceptor_.activate(this);
}
void doorman_impl::remove_from_loop() {
acceptor_.passivate();
}
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/io/network/event_handler.hpp"
#include "caf/logger.hpp"
#include "caf/io/network/socket_utils.hpp"
#include "caf/io/network/default_multiplexer.hpp"
namespace caf {
namespace io {
namespace network {
event_handler::event_handler(default_multiplexer& dm, native_socket sockfd)
: eventbf_(0),
fd_(sockfd),
read_channel_closed_(false),
backend_(dm) {
set_fd_flags();
}
event_handler::~event_handler() {
if (fd_ != invalid_native_socket) {
CAF_LOG_DEBUG("close socket" << CAF_ARG(fd_));
closesocket(fd_);
}
}
void event_handler::close_read_channel() {
if (fd_ == invalid_native_socket || read_channel_closed_)
return;
::shutdown(fd_, 0); // 0 identifies the read channel on Win & UNIX
read_channel_closed_ = true;
}
void event_handler::passivate() {
backend().del(operation::read, fd(), this);
}
void event_handler::activate() {
backend().add(operation::read, fd(), this);
}
void event_handler::set_fd_flags() {
if (fd_ == invalid_native_socket)
return;
// enable nonblocking IO, disable Nagle's algorithm, and suppress SIGPIPE
nonblocking(fd_, true);
tcp_nodelay(fd_, true);
allow_sigpipe(fd_, false);
}
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/logger.hpp"
#include "caf/io/network/pipe_reader.hpp"
#include "caf/io/network/socket_utils.hpp"
#include "caf/io/network/default_multiplexer.hpp"
namespace caf {
namespace io {
namespace network {
pipe_reader::pipe_reader(default_multiplexer& dm)
: event_handler(dm, invalid_native_socket) {
// nop
}
void pipe_reader::removed_from_loop(operation) {
// nop
}
resumable* pipe_reader::try_read_next() {
intptr_t ptrval;
// on windows, we actually have sockets, otherwise we have file handles
# ifdef CAF_WINDOWS
auto res = recv(fd(), reinterpret_cast<socket_recv_ptr>(&ptrval),
sizeof(ptrval), 0);
# else
auto res = read(fd(), &ptrval, sizeof(ptrval));
# endif
if (res != sizeof(ptrval))
return nullptr;
return reinterpret_cast<resumable*>(ptrval);
}
void pipe_reader::handle_event(operation op) {
CAF_LOG_TRACE(CAF_ARG(op));
if (op == operation::read) {
auto ptr = try_read_next();
if (ptr != nullptr)
backend().resume({ptr, false});
}
// else: ignore errors
}
void pipe_reader::init(native_socket sock_fd) {
fd_ = sock_fd;
}
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/io/network/scribe_impl.hpp"
#include <algorithm>
#include "caf/logger.hpp"
#include "caf/io/network/socket_utils.hpp"
#include "caf/io/network/default_multiplexer.hpp"
namespace caf {
namespace io {
namespace network {
scribe_impl::scribe_impl(default_multiplexer& mx, native_socket sockfd)
: scribe(network::conn_hdl_from_socket(sockfd)),
launched_(false),
stream_(mx, sockfd) {
// nop
}
void scribe_impl::configure_read(receive_policy::config config) {
CAF_LOG_TRACE("");
stream_.configure_read(config);
if (!launched_)
launch();
}
void scribe_impl::ack_writes(bool enable) {
CAF_LOG_TRACE(CAF_ARG(enable));
stream_.ack_writes(enable);
}
std::vector<char>& scribe_impl::wr_buf() {
return stream_.wr_buf();
}
std::vector<char>& scribe_impl::rd_buf() {
return stream_.rd_buf();
}
void scribe_impl::stop_reading() {
CAF_LOG_TRACE("");
stream_.stop_reading();
detach(&stream_.backend(), false);
}
void scribe_impl::flush() {
CAF_LOG_TRACE("");
stream_.flush(this);
}
std::string scribe_impl::addr() const {
auto x = remote_addr_of_fd(stream_.fd());
if (!x)
return "";
return *x;
}
uint16_t scribe_impl::port() const {
auto x = remote_port_of_fd(stream_.fd());
if (!x)
return 0;
return *x;
}
void scribe_impl::launch() {
CAF_LOG_TRACE("");
CAF_ASSERT(!launched_);
launched_ = true;
stream_.start(this);
}
void scribe_impl::add_to_loop() {
stream_.activate(this);
}
void scribe_impl::remove_from_loop() {
stream_.passivate();
}
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/sec.hpp"
#include "caf/logger.hpp"
#include "caf/io/network/protocol.hpp"
#include "caf/io/network/socket_utils.hpp"
#ifdef CAF_WINDOWS
# include <winsock2.h>
# include <ws2tcpip.h> // socklen_t, etc. (MSVC20xx)
# include <windows.h>
# include <io.h>
#else
# include <cerrno>
# include <netdb.h>
# include <fcntl.h>
# include <sys/types.h>
# include <arpa/inet.h>
# include <sys/socket.h>
# include <netinet/in.h>
# include <netinet/tcp.h>
# include <utility>
#endif
using std::string;
namespace {
// predicate for `ccall` meaning "expected result of f is 0"
bool cc_zero(int value) {
return value == 0;
}
// predicate for `ccall` meaning "expected result of f is not -1"
bool cc_not_minus1(int value) {
return value != -1;
}
// calls a C functions and returns an error if `predicate(var)` returns false
#define CALL_CFUN(var, predicate, fun_name, expr) \
auto var = expr; \
if (!predicate(var)) \
return make_error(sec::network_syscall_failed, \
fun_name, last_socket_error_as_string())
#ifdef CAF_WINDOWS
// calls a C functions and calls exit() if `predicate(var)` returns false
#define CALL_CRITICAL_CFUN(var, predicate, funname, expr) \
auto var = expr; \
if (!predicate(var)) { \
fprintf(stderr, "[FATAL] %s:%u: syscall failed: %s returned %s\n", \
__FILE__, __LINE__, funname, last_socket_error_as_string().c_str());\
abort(); \
} static_cast<void>(0)
#ifndef SIO_UDP_CONNRESET
#define SIO_UDP_CONNRESET _WSAIOW(IOC_VENDOR, 12)
#endif
#endif // CAF_WINDOWS
} // namespace <anonymous>
namespace caf {
namespace io {
namespace network {
#ifndef CAF_WINDOWS
string last_socket_error_as_string() {
return strerror(errno);
}
expected<void> nonblocking(native_socket fd, bool new_value) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(new_value));
// read flags for fd
CALL_CFUN(rf, cc_not_minus1, "fcntl", fcntl(fd, F_GETFL, 0));
// calculate and set new flags
auto wf = new_value ? (rf | O_NONBLOCK) : (rf & (~(O_NONBLOCK)));
CALL_CFUN(set_res, cc_not_minus1, "fcntl", fcntl(fd, F_SETFL, wf));
return unit;
}
expected<void> allow_sigpipe(native_socket fd, bool new_value) {
if (no_sigpipe_socket_flag != 0) {
int value = new_value ? 0 : 1;
CALL_CFUN(res, cc_zero, "setsockopt",
setsockopt(fd, SOL_SOCKET, no_sigpipe_socket_flag, &value,
static_cast<unsigned>(sizeof(value))));
}
return unit;
}
expected<void> allow_udp_connreset(native_socket, bool) {
// nop; SIO_UDP_CONNRESET only exists on Windows
return unit;
}
std::pair<native_socket, native_socket> create_pipe() {
int pipefds[2];
if (pipe(pipefds) != 0) {
perror("pipe");
exit(EXIT_FAILURE);
}
return {pipefds[0], pipefds[1]};
}
#else // CAF_WINDOWS
string last_socket_error_as_string() {
LPTSTR errorText = NULL;
auto hresult = last_socket_error();
FormatMessage( // use system message tables to retrieve error text
FORMAT_MESSAGE_FROM_SYSTEM
// allocate buffer on local heap for error text
| FORMAT_MESSAGE_ALLOCATE_BUFFER
// Important! will fail otherwise, since we're not
// (and CANNOT) pass insertion parameters
| FORMAT_MESSAGE_IGNORE_INSERTS,
nullptr, // unused with FORMAT_MESSAGE_FROM_SYSTEM
hresult, MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
(LPTSTR) & errorText, // output
0, // minimum size for output buffer
nullptr); // arguments - see note
std::string result;
if (errorText != nullptr) {
result = errorText;
// release memory allocated by FormatMessage()
LocalFree(errorText);
}
return result;
}
expected<void> nonblocking(native_socket fd, bool new_value) {
u_long mode = new_value ? 1 : 0;
CALL_CFUN(res, cc_zero, "ioctlsocket", ioctlsocket(fd, FIONBIO, &mode));
return unit;
}
expected<void> allow_sigpipe(native_socket, bool) {
// nop; SIGPIPE does not exist on Windows
return unit;
}
expected<void> allow_udp_connreset(native_socket fd, bool new_value) {
DWORD bytes_returned = 0;
CALL_CFUN(res, cc_zero, "WSAIoctl",
WSAIoctl(fd, SIO_UDP_CONNRESET, &new_value, sizeof(new_value),
NULL, 0, &bytes_returned, NULL, NULL));
return unit;
}
/**************************************************************************\
* 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. *
\**************************************************************************/
std::pair<native_socket, native_socket> create_pipe() {
socklen_t addrlen = sizeof(sockaddr_in);
native_socket socks[2] = {invalid_native_socket, invalid_native_socket};
CALL_CRITICAL_CFUN(listener, cc_valid_socket, "socket",
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();
closesocket(listener);
closesocket(socks[0]);
closesocket(socks[1]);
WSASetLastError(e);
});
// bind listener to a local port
int reuse = 1;
CALL_CRITICAL_CFUN(tmp1, cc_zero, "setsockopt",
setsockopt(listener, SOL_SOCKET, SO_REUSEADDR,
reinterpret_cast<char*>(&reuse),
static_cast<int>(sizeof(reuse))));
CALL_CRITICAL_CFUN(tmp2, cc_zero, "bind",
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));
CALL_CRITICAL_CFUN(tmp3, cc_zero, "getsockname",
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
CALL_CRITICAL_CFUN(tmp5, cc_zero, "listen", listen(listener, 1));
// create read-only end of the pipe
DWORD flags = 0;
CALL_CRITICAL_CFUN(read_fd, cc_valid_socket, "WSASocketW",
WSASocketW(AF_INET, SOCK_STREAM, 0, nullptr, 0, flags));
CALL_CRITICAL_CFUN(tmp6, cc_zero, "connect",
connect(read_fd, &a.addr,
static_cast<int>(sizeof(a.inaddr))));
// get write-only end of the pipe
CALL_CRITICAL_CFUN(write_fd, cc_valid_socket, "accept",
accept(listener, nullptr, nullptr));
closesocket(listener);
guard.disable();
return std::make_pair(read_fd, write_fd);
}
#endif
expected<int> send_buffer_size(native_socket fd) {
int size;
socklen_t ret_size = sizeof(size);
CALL_CFUN(res, cc_zero, "getsockopt",
getsockopt(fd, SOL_SOCKET, SO_SNDBUF,
reinterpret_cast<getsockopt_ptr>(&size), &ret_size));
return size;
}
expected<void> send_buffer_size(native_socket fd, int new_value) {
CALL_CFUN(res, cc_zero, "setsockopt",
setsockopt(fd, SOL_SOCKET, SO_SNDBUF,
reinterpret_cast<setsockopt_ptr>(&new_value),
static_cast<socklen_t>(sizeof(int))));
return unit;
}
expected<void> tcp_nodelay(native_socket fd, bool new_value) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(new_value));
int flag = new_value ? 1 : 0;
CALL_CFUN(res, cc_zero, "setsockopt",
setsockopt(fd, IPPROTO_TCP, TCP_NODELAY,
reinterpret_cast<setsockopt_ptr>(&flag),
static_cast<socklen_t>(sizeof(flag))));
return unit;
}
bool is_error(ssize_t res, bool is_nonblock) {
if (res < 0) {
auto err = last_socket_error();
if (!is_nonblock || !would_block_or_temporarily_unavailable(err)) {
return true;
}
// don't report an error in case of
// spurious wakeup or something similar
}
return false;
}
expected<std::string> local_addr_of_fd(native_socket fd) {
sockaddr_storage st;
socklen_t st_len = sizeof(st);
sockaddr* sa = reinterpret_cast<sockaddr*>(&st);
CALL_CFUN(tmp1, cc_zero, "getsockname", getsockname(fd, sa, &st_len));
char addr[INET6_ADDRSTRLEN] {0};
switch (sa->sa_family) {
case AF_INET:
return inet_ntop(AF_INET, &reinterpret_cast<sockaddr_in*>(sa)->sin_addr,
addr, sizeof(addr));
case AF_INET6:
return inet_ntop(AF_INET6,
&reinterpret_cast<sockaddr_in6*>(sa)->sin6_addr,
addr, sizeof(addr));
default:
break;
}
return make_error(sec::invalid_protocol_family,
"local_addr_of_fd", sa->sa_family);
}
expected<uint16_t> local_port_of_fd(native_socket fd) {
sockaddr_storage st;
socklen_t st_len = sizeof(st);
CALL_CFUN(tmp, cc_zero, "getsockname",
getsockname(fd, reinterpret_cast<sockaddr*>(&st), &st_len));
return ntohs(port_of(reinterpret_cast<sockaddr&>(st)));
}
expected<std::string> remote_addr_of_fd(native_socket fd) {
sockaddr_storage st;
socklen_t st_len = sizeof(st);
sockaddr* sa = reinterpret_cast<sockaddr*>(&st);
CALL_CFUN(tmp, cc_zero, "getpeername", getpeername(fd, sa, &st_len));
char addr[INET6_ADDRSTRLEN] {0};
switch (sa->sa_family) {
case AF_INET:
return inet_ntop(AF_INET, &reinterpret_cast<sockaddr_in*>(sa)->sin_addr,
addr, sizeof(addr));
case AF_INET6:
return inet_ntop(AF_INET6,
&reinterpret_cast<sockaddr_in6*>(sa)->sin6_addr,
addr, sizeof(addr));
default:
break;
}
return make_error(sec::invalid_protocol_family,
"remote_addr_of_fd", sa->sa_family);
}
expected<uint16_t> remote_port_of_fd(native_socket fd) {
sockaddr_storage st;
socklen_t st_len = sizeof(st);
CALL_CFUN(tmp, cc_zero, "getpeername",
getpeername(fd, reinterpret_cast<sockaddr*>(&st), &st_len));
return ntohs(port_of(reinterpret_cast<sockaddr&>(st)));
}
auto addr_of(sockaddr_in& what) -> decltype(what.sin_addr)& {
return what.sin_addr;
}
auto family_of(sockaddr_in& what) -> decltype(what.sin_family)& {
return what.sin_family;
}
auto port_of(sockaddr_in& what) -> decltype(what.sin_port)& {
return what.sin_port;
}
auto addr_of(sockaddr_in6& what) -> decltype(what.sin6_addr)& {
return what.sin6_addr;
}
auto family_of(sockaddr_in6& what) -> decltype(what.sin6_family)& {
return what.sin6_family;
}
auto port_of(sockaddr_in6& what) -> decltype(what.sin6_port)& {
return what.sin6_port;
}
auto port_of(sockaddr& what) -> decltype(port_of(std::declval<sockaddr_in&>())) {
switch (what.sa_family) {
case AF_INET:
return port_of(reinterpret_cast<sockaddr_in&>(what));
case AF_INET6:
return port_of(reinterpret_cast<sockaddr_in6&>(what));
default:
break;
}
CAF_CRITICAL("invalid protocol family");
}
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/io/network/stream.hpp"
#include <algorithm>
#include "caf/logger.hpp"
#include "caf/defaults.hpp"
#include "caf/config_value.hpp"
#include "caf/io/network/socket_utils.hpp"
#include "caf/io/network/default_multiplexer.hpp"
namespace caf {
namespace io {
namespace network {
stream::stream(default_multiplexer& backend_ref, native_socket sockfd)
: event_handler(backend_ref, sockfd),
max_consecutive_reads_(
get_or(backend().system().config(), "middleman.max-consecutive-reads",
defaults::middleman::max_consecutive_reads)),
read_threshold_(1),
collected_(0),
ack_writes_(false),
writing_(false),
written_(0) {
configure_read(receive_policy::at_most(1024));
}
void stream::start(stream_manager* mgr) {
CAF_ASSERT(mgr != nullptr);
activate(mgr);
}
void stream::activate(stream_manager* mgr) {
if (!reader_) {
reader_.reset(mgr);
event_handler::activate();
prepare_next_read();
}
}
void stream::configure_read(receive_policy::config config) {
rd_flag_ = config.first;
max_ = config.second;
}
void stream::ack_writes(bool x) {
ack_writes_ = x;
}
void stream::write(const void* buf, size_t num_bytes) {
CAF_LOG_TRACE(CAF_ARG(num_bytes));
auto first = reinterpret_cast<const char*>(buf);
auto last = first + num_bytes;
wr_offline_buf_.insert(wr_offline_buf_.end(), first, last);
}
void stream::flush(const manager_ptr& mgr) {
CAF_ASSERT(mgr != nullptr);
CAF_LOG_TRACE(CAF_ARG(wr_offline_buf_.size()));
if (!wr_offline_buf_.empty() && !writing_) {
backend().add(operation::write, fd(), this);
writer_ = mgr;
writing_ = true;
prepare_next_write();
}
}
void stream::stop_reading() {
CAF_LOG_TRACE("");
close_read_channel();
passivate();
}
void stream::removed_from_loop(operation op) {
CAF_LOG_TRACE(CAF_ARG(op));
switch (op) {
case operation::read: reader_.reset(); break;
case operation::write: writer_.reset(); break;
case operation::propagate_error: break;
}
}
void stream::force_empty_write(const manager_ptr& mgr) {
if (!writing_) {
backend().add(operation::write, fd(), this);
writer_ = mgr;
writing_ = true;
}
}
void stream::prepare_next_read() {
collected_ = 0;
switch (rd_flag_) {
case receive_policy_flag::exactly:
if (rd_buf_.size() != max_)
rd_buf_.resize(max_);
read_threshold_ = max_;
break;
case receive_policy_flag::at_most:
if (rd_buf_.size() != max_)
rd_buf_.resize(max_);
read_threshold_ = 1;
break;
case receive_policy_flag::at_least: {
// read up to 10% more, but at least allow 100 bytes more
auto max_size = max_ + std::max<size_t>(100, max_ / 10);
if (rd_buf_.size() != max_size)
rd_buf_.resize(max_size);
read_threshold_ = max_;
break;
}
}
}
void stream::prepare_next_write() {
CAF_LOG_TRACE(CAF_ARG(wr_buf_.size()) << CAF_ARG(wr_offline_buf_.size()));
written_ = 0;
wr_buf_.clear();
if (wr_offline_buf_.empty()) {
writing_ = false;
backend().del(operation::write, fd(), this);
} else {
wr_buf_.swap(wr_offline_buf_);
}
}
bool stream::handle_read_result(rw_state read_result, size_t rb) {
switch (read_result) {
case rw_state::failure:
reader_->io_failure(&backend(), operation::read);
passivate();
return false;
case rw_state::indeterminate:
return false;
case rw_state::success:
if (rb == 0)
return false;
collected_ += rb;
if (collected_ >= read_threshold_) {
auto res = reader_->consume(&backend(), rd_buf_.data(),
collected_);
prepare_next_read();
if (!res) {
passivate();
return false;
}
}
break;
}
return true;
}
void stream::handle_write_result(rw_state write_result, size_t wb) {
switch (write_result) {
case rw_state::failure:
writer_->io_failure(&backend(), operation::write);
backend().del(operation::write, fd(), this);
break;
case rw_state::indeterminate:
prepare_next_write();
break;
case rw_state::success:
written_ += wb;
CAF_ASSERT(written_ <= wr_buf_.size());
auto remaining = wr_buf_.size() - written_;
if (ack_writes_)
writer_->data_transferred(&backend(), wb,
remaining + wr_offline_buf_.size());
// prepare next send (or stop sending)
if (remaining == 0)
prepare_next_write();
break;
}
}
void stream::handle_error_propagation() {
if (reader_)
reader_->io_failure(&backend(), operation::read);
if (writer_)
writer_->io_failure(&backend(), operation::write);
// backend will delete this handler anyway,
// no need to call backend().del() here
}
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/io/network/policy/tcp.hpp"
namespace caf {
namespace io {
namespace network {
namespace policy {
read_some_fun tcp::read_some = caf::io::network::read_some;
write_some_fun tcp::write_some = caf::io::network::write_some;
try_accept_fun tcp::try_accept = caf::io::network::try_accept;
} // policy
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/io/network/policy/udp.hpp"
namespace caf {
namespace io {
namespace network {
namespace policy {
read_datagram_fun udp::read_datagram = caf::io::network::read_datagram;
write_datagram_fun udp::write_datagram = caf::io::network::write_datagram;
} // policy
} // namespace network
} // namespace io
} // namespace caf
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