Unverified Commit 7911ddc1 authored by Joseph Noir's avatar Joseph Noir Committed by GitHub

Merge pull request #716

Split up default multiplexer file, close #710 
parents 8f0085f2 52b95149
......@@ -35,6 +35,18 @@ set(LIBCAF_IO_SRCS
src/scribe.cpp
src/stream_manager.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/native_socket.cpp
src/socket_guard.cpp
)
add_custom_target(libcaf_io)
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <cstdio>
#include <cstdlib>
#include "caf/error.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/sec.hpp"
namespace caf {
namespace detail {
/// Predicate for `ccall` meaning "expected result of f is 0".
inline bool cc_zero(int value) {
return value == 0;
}
/// Predicate for `ccall` meaning "expected result of f is 1".
inline bool cc_one(int value) {
return value == 1;
}
/// Predicate for `ccall` meaning "expected result of f is not -1".
inline bool cc_not_minus1(int value) {
return value != -1;
}
/// Predicate for `ccall` meaning "expected result of f is a valid socket".
inline bool cc_valid_socket(caf::io::network::native_socket fd) {
return fd != caf::io::network::invalid_native_socket;
}
/// 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())
/// 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)
} // namespace detail
} // namespace caf
......@@ -5,7 +5,7 @@
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* 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 *
......@@ -18,18 +18,24 @@
#pragma once
namespace caf {
#include "caf/io/network/native_socket.hpp"
namespace caf {
namespace detail {
} // namespace detail
class socket_guard {
public:
explicit socket_guard(io::network::native_socket fd);
namespace opencl {
namespace detail {
~socket_guard();
using namespace caf::detail;
io::network::native_socket release();
void close();
private:
io::network::native_socket fd_;
};
} // namespace detail
} // namespace opencl
} // namespace caf
......@@ -53,6 +53,7 @@ using datagram_servant_ptr = intrusive_ptr<datagram_servant>;
namespace network {
class multiplexer;
class default_multiplexer;
} // 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/native_socket.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/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
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/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
......@@ -18,19 +18,44 @@
#pragma once
#include <cstddef>
#include <string>
#include <cstdint>
#include "caf/config.hpp"
#include "caf/expected.hpp"
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 socket_size_type = int;
#else
using setsockopt_ptr = const void*;
using getsockopt_ptr = void*;
using socket_send_ptr = const void*;
using socket_recv_ptr = void*;
using socket_size_type = unsigned;
#endif
using signed_size_type = std::make_signed<size_t>::type;
// More bootstrapping.
extern const int ec_out_of_memory;
extern const int ec_interrupted_syscall;
extern const int no_sigpipe_socket_flag;
extern const int no_sigpipe_io_flag;
#ifdef CAF_WINDOWS
using native_socket = size_t;
constexpr native_socket invalid_native_socket = static_cast<native_socket>(-1);
inline int64_t int64_from_native_socket(native_socket sock) {
return sock == invalid_native_socket ? -1 : static_cast<uint64_t>(sock);
return sock == invalid_native_socket ? -1 : static_cast<int64_t>(sock);
}
#else
using native_socket = int;
......@@ -40,7 +65,60 @@ namespace network {
}
#endif
/// Returns the last socket error as an integer.
int last_socket_error();
/// Close socket `fd`.
void close_socket(native_socket fd);
/// Returns true if `errcode` indicates that an operation would
/// block or return nothing at the moment and can be tried again
/// at a later point.
bool would_block_or_temporarily_unavailable(int errcode);
/// 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(signed_size_type 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);
} // 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
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/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 <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. *
******************************************************************************/
#pragma once
#include "caf/io/network/rw_state.hpp"
#include "caf/io/network/native_socket.hpp"
namespace caf {
namespace policy {
/// Policy object for wrapping default TCP operations.
struct tcp {
/// 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).
static io::network::rw_state read_some(size_t& result,
io::network::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).
static io::network::rw_state write_some(size_t& result,
io::network::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
static bool try_accept(io::network::native_socket& result,
io::network::native_socket fd);
};
} // namespace policy
} // 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/ip_endpoint.hpp"
#include "caf/io/network/native_socket.hpp"
namespace caf {
namespace policy {
/// Policy object for wrapping default UDP operations.
struct udp {
/// 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`.
static bool read_datagram(size_t& result, io::network::native_socket fd,
void* buf, size_t buf_len,
io::network::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).
static bool write_datagram(size_t& result, io::network::native_socket fd,
void* buf, size_t buf_len,
const io::network::ip_endpoint& ep);
};
} // namespace policy
} // 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/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: ; // nop
};
}
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/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
This diff is collapsed.
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/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/default_multiplexer.hpp"
#ifdef CAF_WINDOWS
# include <winsock2.h>
#else
# include <sys/socket.h>
#endif
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_));
close_socket(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
......@@ -21,6 +21,8 @@
#include "caf/sec.hpp"
#include "caf/logger.hpp"
#include "caf/io/network/native_socket.hpp"
#ifdef CAF_WINDOWS
# include <winsock2.h>
# include <windows.h>
......@@ -209,12 +211,12 @@ std::string host(const ip_endpoint& ep) {
case AF_INET:
inet_ntop(AF_INET,
&const_cast<sockaddr_in*>(reinterpret_cast<const sockaddr_in*>(ep.caddress()))->sin_addr,
addr, static_cast<socklen_t>(*ep.clength()));
addr, static_cast<socket_size_type>(*ep.clength()));
break;
case AF_INET6:
inet_ntop(AF_INET6,
&const_cast<sockaddr_in6*>(reinterpret_cast<const sockaddr_in6*>(ep.caddress()))->sin6_addr,
addr, static_cast<socklen_t>(*ep.clength()));
addr, static_cast<socket_size_type>(*ep.clength()));
break;
default:
addr[0] = '\0';
......
This diff is collapsed.
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <cstdint>
#include "caf/io/network/pipe_reader.hpp"
#include "caf/io/network/default_multiplexer.hpp"
#ifdef CAF_WINDOWS
# include <winsock2.h>
#else
# include <unistd.h>
# include <sys/socket.h>
#endif
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() {
std::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/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/detail/socket_guard.hpp"
#ifdef CAF_WINDOWS
# include <winsock2.h>
#else
# include <unistd.h>
#endif
#include "caf/logger.hpp"
namespace caf {
namespace detail {
socket_guard::socket_guard(io::network::native_socket fd) : fd_(fd) {
// nop
}
socket_guard::~socket_guard() {
close();
}
io::network::native_socket socket_guard::release() {
auto fd = fd_;
fd_ = io::network::invalid_native_socket;
return fd;
}
void socket_guard::close() {
if (fd_ != io::network::invalid_native_socket) {
CAF_LOG_DEBUG("close socket" << CAF_ARG(fd_));
io::network::close_socket(fd_);
fd_ = io::network::invalid_native_socket;
}
}
} // namespace detail
} // namespace detail
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/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: ; // nop
}
}
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
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
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