Commit 8532a64a authored by Joseph Noir's avatar Joseph Noir

Move TCP/UDP transport policies to separate files

parent 85ef2191
...@@ -6,45 +6,7 @@ ...@@ -6,45 +6,7 @@
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp" #include "caf/binary_serializer.hpp"
#include "caf/detail/call_cfun.hpp" #include "caf/detail/call_cfun.hpp"
#include "caf/policy/newb_tcp.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 <io.h>
# include <windows.h>
# include <winsock2.h>
# include <ws2ipdef.h>
# include <ws2tcpip.h>
#else
# include <unistd.h>
# include <arpa/inet.h>
# include <cerrno>
# include <fcntl.h>
# include <netdb.h>
# include <netinet/in.h>
# include <netinet/ip.h>
# include <netinet/tcp.h>
# include <sys/socket.h>
# include <sys/types.h>
# ifdef CAF_POLL_MULTIPLEXER
# include <poll.h>
# elif defined(CAF_EPOLL_MULTIPLEXER)
# include <sys/epoll.h>
# else
# error "neither CAF_POLL_MULTIPLEXER nor CAF_EPOLL_MULTIPLEXER defined"
# endif
#endif
using namespace caf; using namespace caf;
...@@ -158,135 +120,6 @@ struct basp { ...@@ -158,135 +120,6 @@ struct basp {
} }
}; };
struct tcp_transport : public io::network::transport_policy {
tcp_transport()
: read_threshold{0},
collected{0},
maximum{0},
writing{false},
written{0} {
// nop
}
error read_some(io::network::event_handler* parent) override {
CAF_LOG_TRACE("");
std::cerr << "read some called" << std::endl;
size_t len = receive_buffer.size() - collected;
receive_buffer.resize(len);
void* buf = receive_buffer.data() + collected;
auto sres = ::recv(parent->fd(),
reinterpret_cast<io::network::socket_recv_ptr>(buf),
len, io::network::no_sigpipe_io_flag);
if (io::network::is_error(sres, true) || sres == 0) {
std::cerr << "read some error" << std::endl;
// recv returns 0 when the peer has performed an orderly shutdown
return sec::runtime_error;
}
size_t result = (sres > 0) ? static_cast<size_t>(sres) : 0;
collected += result;
received_bytes = collected;
return none;
}
bool should_deliver() override {
CAF_LOG_DEBUG(CAF_ARG(collected) << CAF_ARG(read_threshold));
return collected >= read_threshold;
}
void prepare_next_read(io::network::event_handler*) override {
collected = 0;
received_bytes = 0;
switch (rd_flag) {
case io::receive_policy_flag::exactly:
if (receive_buffer.size() != maximum)
receive_buffer.resize(maximum);
read_threshold = maximum;
break;
case io::receive_policy_flag::at_most:
if (receive_buffer.size() != maximum)
receive_buffer.resize(maximum);
read_threshold = 1;
break;
case io::receive_policy_flag::at_least: {
// read up to 10% more, but at least allow 100 bytes more
auto maximumsize = maximum + std::max<size_t>(100, maximum / 10);
if (receive_buffer.size() != maximumsize)
receive_buffer.resize(maximumsize);
read_threshold = maximum;
break;
}
}
}
void configure_read(io::receive_policy::config config) override {
rd_flag = config.first;
maximum = config.second;
}
error write_some(io::network::event_handler* parent) override {
CAF_LOG_TRACE("");
const void* buf = send_buffer.data() + written;
auto len = send_buffer.size() - written;
auto sres = ::send(parent->fd(),
reinterpret_cast<io::network::socket_send_ptr>(buf),
len, io::network::no_sigpipe_io_flag);
if (io::network::is_error(sres, true))
return sec::runtime_error;
size_t result = (sres > 0) ? static_cast<size_t>(sres) : 0;
written += result;
auto remaining = send_buffer.size() - written;
if (remaining == 0)
prepare_next_write(parent);
return none;
}
void prepare_next_write(io::network::event_handler* parent) override {
written = 0;
send_buffer.clear();
if (offline_buffer.empty()) {
parent->backend().del(io::network::operation::write,
parent->fd(), parent);
writing = false;
} else {
send_buffer.swap(offline_buffer);
}
}
io::network::byte_buffer& wr_buf() {
return offline_buffer;
}
void flush(io::network::event_handler* parent) override {
CAF_ASSERT(parent != nullptr);
CAF_LOG_TRACE(CAF_ARG(offline_buffer.size()));
if (!offline_buffer.empty() && !writing) {
parent->backend().add(io::network::operation::write,
parent->fd(), parent);
writing = true;
prepare_next_write(parent);
}
}
expected<native_socket>
connect(const std::string& host, uint16_t port,
optional<io::network::protocol::network> preferred = none) override {
auto res = io::network::new_tcp_connection(host, port, preferred);
if (!res)
std::cerr << "failed to create new TCP connection" << std::endl;
return res;
}
// State for reading.
size_t read_threshold;
size_t collected;
size_t maximum;
io::receive_policy_flag rd_flag;
// State for writing.
bool writing;
size_t written;
};
template <class T> template <class T>
struct tcp_protocol struct tcp_protocol
: public io::network::protocol_policy<typename T::message_type> { : public io::network::protocol_policy<typename T::message_type> {
...@@ -377,37 +210,6 @@ struct basp_newb : public io::network::newb<new_basp_message> { ...@@ -377,37 +210,6 @@ struct basp_newb : public io::network::newb<new_basp_message> {
actor responder; actor responder;
}; };
struct accept_tcp : public io::network::accept_policy<new_basp_message> {
expected<native_socket> create_socket(uint16_t port, const char* host,
bool reuse = false) override {
return io::network::new_tcp_acceptor_impl(port, host, reuse);
}
std::pair<native_socket, io::network::transport_policy_ptr>
accept(io::network::event_handler* parent) override {
using namespace io::network;
sockaddr_storage addr;
std::memset(&addr, 0, sizeof(addr));
socket_size_type addrlen = sizeof(addr);
auto result = ::accept(parent->fd(), reinterpret_cast<sockaddr*>(&addr),
&addrlen);
if (result == invalid_native_socket) {
auto err = last_socket_error();
if (!would_block_or_temporarily_unavailable(err)) {
return {invalid_native_socket, nullptr};
}
}
std::cerr << "accepted connection" << std::endl;
transport_policy_ptr ptr{new tcp_transport};
return {result, std::move(ptr)};
}
void init(io::network::newb<new_basp_message>& n) override {
n.start();
}
};
template <class ProtocolPolicy> template <class ProtocolPolicy>
struct tcp_acceptor struct tcp_acceptor
: public io::network::newb_acceptor<typename ProtocolPolicy::message_type> { : public io::network::newb_acceptor<typename ProtocolPolicy::message_type> {
...@@ -454,13 +256,16 @@ struct tcp_test_broker_state { ...@@ -454,13 +256,16 @@ struct tcp_test_broker_state {
void caf_main(actor_system& sys, const actor_system_config&) { void caf_main(actor_system& sys, const actor_system_config&) {
using acceptor_t = tcp_acceptor<tcp_protocol<basp>>; using acceptor_t = tcp_acceptor<tcp_protocol<basp>>;
using io::network::make_server_newb; using caf::io::network::make_server_newb;
using io::network::make_client_newb; using caf::io::network::make_client_newb;
using caf::policy::accept_tcp;
using caf::policy::tcp_transport;
const char* host = "localhost"; const char* host = "localhost";
const uint16_t port = 12345; const uint16_t port = 12345;
scoped_actor self{sys}; scoped_actor self{sys};
auto running = [=](event_based_actor* self, std::string name, actor , actor b) -> behavior { auto running = [=](event_based_actor* self, std::string name,
actor, actor b) -> behavior {
return { return {
[=](std::string str) { [=](std::string str) {
aout(self) << "[" << name << "] received '" << str << "'" << std::endl; aout(self) << "[" << name << "] received '" << str << "'" << std::endl;
...@@ -471,7 +276,8 @@ void caf_main(actor_system& sys, const actor_system_config&) { ...@@ -471,7 +276,8 @@ void caf_main(actor_system& sys, const actor_system_config&) {
}, },
}; };
}; };
auto init = [=](event_based_actor* self, std::string name, actor m) -> behavior { auto init = [=](event_based_actor* self, std::string name,
actor m) -> behavior {
self->set_default_handler(skip); self->set_default_handler(skip);
return { return {
[=](actor b) { [=](actor b) {
...@@ -486,11 +292,13 @@ void caf_main(actor_system& sys, const actor_system_config&) { ...@@ -486,11 +292,13 @@ void caf_main(actor_system& sys, const actor_system_config&) {
auto client_helper = sys.spawn(init, "c", self); auto client_helper = sys.spawn(init, "c", self);
aout(self) << "creating new server" << std::endl; aout(self) << "creating new server" << std::endl;
auto server_ptr = make_server_newb<acceptor_t, accept_tcp>(sys, port, nullptr, true); auto server_ptr = make_server_newb<acceptor_t, accept_tcp>(sys, port, nullptr,
true);
server_ptr->responder = server_helper; server_ptr->responder = server_helper;
aout(self) << "creating new client" << std::endl; aout(self) << "creating new client" << std::endl;
auto client = make_client_newb<basp_newb, tcp_transport, tcp_protocol<basp>>(sys, host, port); auto client = make_client_newb<basp_newb, tcp_transport,
tcp_protocol<basp>>(sys, host, port);
self->send(client, responder_atom::value, client_helper); self->send(client, responder_atom::value, client_helper);
self->send(client_helper, send_atom::value, "hallo"); self->send(client_helper, send_atom::value, "hallo");
......
...@@ -6,45 +6,7 @@ ...@@ -6,45 +6,7 @@
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp" #include "caf/binary_serializer.hpp"
#include "caf/detail/call_cfun.hpp" #include "caf/detail/call_cfun.hpp"
#include "caf/policy/newb_udp.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 <io.h>
# include <windows.h>
# include <winsock2.h>
# include <ws2ipdef.h>
# include <ws2tcpip.h>
#else
# include <unistd.h>
# include <arpa/inet.h>
# include <cerrno>
# include <fcntl.h>
# include <netdb.h>
# include <netinet/in.h>
# include <netinet/ip.h>
# include <netinet/tcp.h>
# include <sys/socket.h>
# include <sys/types.h>
# ifdef CAF_POLL_MULTIPLEXER
# include <poll.h>
# elif defined(CAF_EPOLL_MULTIPLEXER)
# include <sys/epoll.h>
# else
# error "neither CAF_POLL_MULTIPLEXER nor CAF_EPOLL_MULTIPLEXER defined"
# endif
#endif
using namespace caf; using namespace caf;
...@@ -249,150 +211,6 @@ struct ordering { ...@@ -249,150 +211,6 @@ struct ordering {
} }
}; };
struct udp_transport : public io::network::transport_policy {
udp_transport()
: maximum{std::numeric_limits<uint16_t>::max()},
first_message{true},
writing{false},
written{0},
offline_sum{0} {
// nop
}
error read_some(io::network::event_handler* parent) override {
CAF_LOG_TRACE(CAF_ARG(parent->fd()));
memset(sender.address(), 0, sizeof(sockaddr_storage));
io::network::socket_size_type len = sizeof(sockaddr_storage);
auto buf_ptr = static_cast<io::network::socket_recv_ptr>(receive_buffer.data());
auto buf_len = receive_buffer.size();
auto sres = ::recvfrom(parent->fd(), buf_ptr, buf_len,
0, sender.address(), &len);
if (io::network::is_error(sres, true)) {
CAF_LOG_ERROR("recvfrom returned" << CAF_ARG(sres));
return sec::runtime_error;
} else if (io::network::would_block_or_temporarily_unavailable(
io::network::last_socket_error())) {
return sec::end_of_stream;
}
if (sres == 0)
CAF_LOG_INFO("Received empty datagram");
else if (sres > static_cast<io::network::signed_size_type>(buf_len))
CAF_LOG_WARNING("recvfrom cut of message, only received "
<< CAF_ARG(buf_len) << " of " << CAF_ARG(sres) << " bytes");
received_bytes = (sres > 0) ? static_cast<size_t>(sres) : 0;
*sender.length() = static_cast<size_t>(len);
if (first_message) {
endpoint = sender;
first_message = false;
}
return none;
}
bool should_deliver() override {
CAF_LOG_TRACE("");
return received_bytes != 0 && sender == endpoint;
}
void prepare_next_read(io::network::event_handler*) override {
received_bytes = 0;
receive_buffer.resize(maximum);
}
void configure_read(io::receive_policy::config) override {
// nop
}
error write_some(io::network::event_handler* parent) override {
std::cerr << "sending on socket: " << parent->fd() << std::endl;
using namespace caf::io::network;
CAF_LOG_TRACE(CAF_ARG(parent->fd()) << CAF_ARG(send_buffer.size()));
socket_size_type len = static_cast<socket_size_type>(*endpoint.clength());
auto buf_ptr = reinterpret_cast<socket_send_ptr>(send_buffer.data() + written);
auto buf_len = send_sizes.front();
auto sres = ::sendto(parent->fd(), buf_ptr, buf_len,
0, endpoint.caddress(), len);
std::cerr << "sent " << sres << " bytes to " << to_string(endpoint) << std::endl;
if (is_error(sres, true)) {
std::cerr << "sento failed: " << last_socket_error_as_string() << std::endl;
std::abort();
CAF_LOG_ERROR("sendto returned" << CAF_ARG(sres));
return sec::runtime_error;
}
send_sizes.pop_front();
written += (sres > 0) ? static_cast<size_t>(sres) : 0;
auto remaining = send_buffer.size() - written;
std::cerr << "wrote '" << written << "' got '" << remaining << "' bytes left to write" << std::endl;
if (remaining == 0)
prepare_next_write(parent);
return none;
}
void prepare_next_write(io::network::event_handler* parent) override {
written = 0;
send_buffer.clear();
send_sizes.clear();
if (offline_buffer.empty()) {
writing = false;
parent->backend().del(io::network::operation::write,
parent->fd(), parent);
} else {
// Add size of last chunk.
offline_sizes.push_back(offline_buffer.size() - offline_sum);
// Switch buffers.
send_buffer.swap(offline_buffer);
send_sizes.swap(offline_sizes);
// Reset sum.
offline_sum = 0;
}
}
io::network::byte_buffer& wr_buf() {
if (!offline_buffer.empty()) {
auto chunk_size = offline_buffer.size() - offline_sum;
offline_sizes.push_back(chunk_size);
offline_sum += chunk_size;
std::cerr << "adding chunk '" << chunk_size << "' up to total of '" << offline_sum << std::endl;
}
return offline_buffer;
}
void flush(io::network::event_handler* parent) override {
CAF_ASSERT(parent != nullptr);
CAF_LOG_TRACE(CAF_ARG(offline_buffer.size()));
if (!offline_buffer.empty() && !writing) {
parent->backend().add(io::network::operation::write,
parent->fd(), parent);
writing = true;
prepare_next_write(parent);
}
}
expected<native_socket>
connect(const std::string& host, uint16_t port,
optional<io::network::protocol::network> preferred = none) override {
auto res = io::network::new_remote_udp_endpoint_impl(host, port, preferred);
if (!res)
return std::move(res.error());
endpoint = res->second;
return res->first;
}
// State for reading.
size_t maximum;
bool first_message;
// State for writing.
bool writing;
size_t written;
size_t offline_sum;
std::deque<size_t> send_sizes;
std::deque<size_t> offline_sizes;
// UDP endpoints.
io::network::ip_endpoint endpoint;
io::network::ip_endpoint sender;
};
template <class T> template <class T>
struct udp_protocol struct udp_protocol
: public io::network::protocol_policy<typename T::message_type> { : public io::network::protocol_policy<typename T::message_type> {
...@@ -476,34 +294,6 @@ struct basp_newb : public io::network::newb<new_basp_message> { ...@@ -476,34 +294,6 @@ struct basp_newb : public io::network::newb<new_basp_message> {
actor responder; actor responder;
}; };
struct accept_udp
: public io::network::accept_policy<new_basp_message> {
expected<native_socket> create_socket(uint16_t port, const char* host,
bool reuse = false) override {
auto res = io::network::new_local_udp_endpoint_impl(port, host, reuse);
if (!res)
return std::move(res.error());
return (*res).first;
}
std::pair<native_socket, io::network::transport_policy_ptr>
accept(io::network::event_handler*) override {
auto res = io::network::new_local_udp_endpoint_impl(0, nullptr);
if (!res) {
CAF_LOG_DEBUG("failed to create local endpoint");
return {invalid_native_socket, nullptr};
}
auto sock = std::move(res->first);
io::network::transport_policy_ptr ptr{new udp_transport};
return {sock, std::move(ptr)};
}
void init(io::network::newb<new_basp_message>& n) override {
n.start();
}
};
template <class ProtocolPolicy> template <class ProtocolPolicy>
struct udp_acceptor struct udp_acceptor
: public io::network::newb_acceptor<typename ProtocolPolicy::message_type> { : public io::network::newb_acceptor<typename ProtocolPolicy::message_type> {
...@@ -543,8 +333,10 @@ struct udp_test_broker_state { ...@@ -543,8 +333,10 @@ struct udp_test_broker_state {
void caf_main(actor_system& sys, const actor_system_config&) { void caf_main(actor_system& sys, const actor_system_config&) {
using acceptor_t = udp_acceptor<udp_protocol<ordering<basp>>>; using acceptor_t = udp_acceptor<udp_protocol<ordering<basp>>>;
using io::network::make_server_newb; using caf::io::network::make_server_newb;
using io::network::make_client_newb; using caf::io::network::make_client_newb;
using caf::policy::udp_transport;
using caf::policy::accept_udp;
const char* host = "localhost"; const char* host = "localhost";
const uint16_t port = 12345; const uint16_t port = 12345;
scoped_actor self{sys}; scoped_actor self{sys};
......
...@@ -47,6 +47,8 @@ set(LIBCAF_IO_SRCS ...@@ -47,6 +47,8 @@ set(LIBCAF_IO_SRCS
src/udp.cpp src/udp.cpp
src/native_socket.cpp src/native_socket.cpp
src/socket_guard.cpp src/socket_guard.cpp
src/newb_tcp.cpp
src/newb_udp.cpp
) )
add_custom_target(libcaf_io) add_custom_target(libcaf_io)
......
...@@ -74,6 +74,19 @@ namespace network { ...@@ -74,6 +74,19 @@ namespace network {
using byte_buffer = std::vector<char>; using byte_buffer = std::vector<char>;
using header_writer = caf::callback<byte_buffer&>; using header_writer = caf::callback<byte_buffer&>;
// -- newb base ----------------------------------------------------------------
struct newb_base : public network::event_handler {
newb_base(default_multiplexer& dm, native_socket sockfd)
: event_handler(dm, sockfd) {
// nop
}
virtual void start() = 0;
virtual void stop() = 0;
virtual void flush() = 0;
};
// -- transport policy --------------------------------------------------------- // -- transport policy ---------------------------------------------------------
struct transport_policy { struct transport_policy {
...@@ -154,7 +167,6 @@ using transport_policy_ptr = std::unique_ptr<transport_policy>; ...@@ -154,7 +167,6 @@ using transport_policy_ptr = std::unique_ptr<transport_policy>;
// -- accept policy ------------------------------------------------------------ // -- accept policy ------------------------------------------------------------
template <class Message>
struct accept_policy { struct accept_policy {
virtual ~accept_policy() { virtual ~accept_policy() {
// nop // nop
...@@ -166,7 +178,7 @@ struct accept_policy { ...@@ -166,7 +178,7 @@ struct accept_policy {
virtual std::pair<native_socket, transport_policy_ptr> virtual std::pair<native_socket, transport_policy_ptr>
accept(network::event_handler*) = 0; accept(network::event_handler*) = 0;
virtual void init(newb<Message>&) = 0; virtual void init(newb_base&) = 0;
}; };
// -- protocol policy ---------------------------------------------------------- // -- protocol policy ----------------------------------------------------------
...@@ -221,7 +233,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -221,7 +233,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
with<mixin::sender, mixin::requester, with<mixin::sender, mixin::requester,
mixin::behavior_changer>, mixin::behavior_changer>,
public dynamically_typed_actor_base, public dynamically_typed_actor_base,
public network::event_handler { public newb_base {
using super = typename extend<scheduled_actor, newb<Message>>:: using super = typename extend<scheduled_actor, newb<Message>>::
template with<mixin::sender, mixin::requester, mixin::behavior_changer>; template with<mixin::sender, mixin::requester, mixin::behavior_changer>;
...@@ -231,7 +243,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -231,7 +243,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
newb(actor_config& cfg, default_multiplexer& dm, native_socket sockfd) newb(actor_config& cfg, default_multiplexer& dm, native_socket sockfd)
: super(cfg), : super(cfg),
event_handler(dm, sockfd) { newb_base(dm, sockfd) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
} }
...@@ -341,7 +353,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -341,7 +353,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
super::setf(super::is_initialized_flag); super::setf(super::is_initialized_flag);
} }
void start() { void start() override {
CAF_PUSH_AID_FROM_PTR(this); CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
intrusive_ptr_add_ref(super::ctrl()); intrusive_ptr_add_ref(super::ctrl());
...@@ -351,7 +363,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -351,7 +363,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
transport->prepare_next_read(this); transport->prepare_next_read(this);
} }
void stop() { void stop() override {
CAF_PUSH_AID_FROM_PTR(this); CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
intrusive_ptr_release(super::ctrl()); intrusive_ptr_release(super::ctrl());
...@@ -369,7 +381,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -369,7 +381,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
return {this, protocol.get(), &buf, hstart, hlen}; return {this, protocol.get(), &buf, hstart, hlen};
} }
void flush() { void flush() override {
transport->flush(this); transport->flush(this);
} }
...@@ -505,7 +517,7 @@ struct newb_acceptor : public network::event_handler { ...@@ -505,7 +517,7 @@ struct newb_acceptor : public network::event_handler {
virtual expected<actor> create_newb(native_socket sock, virtual expected<actor> create_newb(native_socket sock,
transport_policy_ptr pol) = 0; transport_policy_ptr pol) = 0;
std::unique_ptr<accept_policy<Message>> acceptor; std::unique_ptr<accept_policy> acceptor;
}; };
// -- factories ---------------------------------------------------------------- // -- factories ----------------------------------------------------------------
...@@ -553,7 +565,6 @@ std::unique_ptr<NewbAcceptor> make_server_newb(actor_system& sys, ...@@ -553,7 +565,6 @@ std::unique_ptr<NewbAcceptor> make_server_newb(actor_system& sys,
return ptr; return ptr;
} }
} // 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/network/newb.hpp"
#include "caf/io/network/native_socket.hpp"
namespace caf {
namespace policy {
struct tcp_transport : public io::network::transport_policy {
tcp_transport();
error read_some(io::network::event_handler* parent) override;
bool should_deliver() override;
void prepare_next_read(io::network::event_handler*) override;
void configure_read(io::receive_policy::config config) override;
error write_some(io::network::event_handler* parent) override;
void prepare_next_write(io::network::event_handler* parent) override;
inline io::network::byte_buffer& wr_buf() {
return offline_buffer;
}
void flush(io::network::event_handler* parent) override;
expected<io::network::native_socket>
connect(const std::string& host, uint16_t port,
optional<io::network::protocol::network> preferred = none) override;
// State for reading.
size_t read_threshold;
size_t collected;
size_t maximum;
io::receive_policy_flag rd_flag;
// State for writing.
bool writing;
size_t written;
};
struct accept_tcp : public io::network::accept_policy {
expected<io::network::native_socket>
create_socket(uint16_t port,const char* host,bool reuse = false) override;
std::pair<io::network::native_socket, io::network::transport_policy_ptr>
accept(io::network::event_handler* parent) override;
void init(io::network::newb_base& n) override;
};
} // 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/newb.hpp"
#include "caf/io/network/native_socket.hpp"
namespace caf {
namespace policy {
struct udp_transport : public io::network::transport_policy {
udp_transport();
error read_some(io::network::event_handler* parent) override;
inline bool should_deliver() override {
CAF_LOG_TRACE("");
return received_bytes != 0 && sender == endpoint;
}
void prepare_next_read(io::network::event_handler*) override;
inline void configure_read(io::receive_policy::config) override {
// nop
}
error write_some(io::network::event_handler* parent) override;
void prepare_next_write(io::network::event_handler* parent) override;
io::network::byte_buffer& wr_buf();
void flush(io::network::event_handler* parent) override;
expected<io::network::native_socket>
connect(const std::string& host, uint16_t port,
optional<io::network::protocol::network> preferred = none) override;
// State for reading.
size_t maximum;
bool first_message;
// State for writing.
bool writing;
size_t written;
size_t offline_sum;
std::deque<size_t> send_sizes;
std::deque<size_t> offline_sizes;
// UDP endpoints.
io::network::ip_endpoint endpoint;
io::network::ip_endpoint sender;
};
struct accept_udp : public io::network::accept_policy {
expected<io::network::native_socket>
create_socket(uint16_t port, const char* host, bool reuse = false) override;
std::pair<io::network::native_socket, io::network::transport_policy_ptr>
accept(io::network::event_handler*) override;
void init(io::network::newb_base& n) override;
};
} // 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/policy/newb_tcp.hpp"
#include "caf/config.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 <io.h>
# include <windows.h>
# include <winsock2.h>
# include <ws2ipdef.h>
# include <ws2tcpip.h>
#else
# include <unistd.h>
# include <arpa/inet.h>
# include <cerrno>
# include <fcntl.h>
# include <netdb.h>
# include <netinet/in.h>
# include <netinet/ip.h>
# include <netinet/tcp.h>
# include <sys/socket.h>
# include <sys/types.h>
# ifdef CAF_POLL_MULTIPLEXER
# include <poll.h>
# elif defined(CAF_EPOLL_MULTIPLEXER)
# include <sys/epoll.h>
# else
# error "neither CAF_POLL_MULTIPLEXER nor CAF_EPOLL_MULTIPLEXER defined"
# endif
#endif
namespace caf {
namespace policy {
tcp_transport::tcp_transport()
: read_threshold{0},
collected{0},
maximum{0},
writing{false},
written{0} {
// nop
}
error tcp_transport::read_some(io::network::event_handler* parent) {
CAF_LOG_TRACE("");
std::cerr << "read some called" << std::endl;
size_t len = receive_buffer.size() - collected;
receive_buffer.resize(len);
void* buf = receive_buffer.data() + collected;
auto sres = ::recv(parent->fd(),
reinterpret_cast<io::network::socket_recv_ptr>(buf),
len, io::network::no_sigpipe_io_flag);
if (io::network::is_error(sres, true) || sres == 0) {
std::cerr << "read some error" << std::endl;
// recv returns 0 when the peer has performed an orderly shutdown
return sec::runtime_error;
}
size_t result = (sres > 0) ? static_cast<size_t>(sres) : 0;
collected += result;
received_bytes = collected;
return none;
}
bool tcp_transport::should_deliver() {
CAF_LOG_DEBUG(CAF_ARG(collected) << CAF_ARG(read_threshold));
return collected >= read_threshold;
}
void tcp_transport::prepare_next_read(io::network::event_handler*) {
collected = 0;
received_bytes = 0;
switch (rd_flag) {
case io::receive_policy_flag::exactly:
if (receive_buffer.size() != maximum)
receive_buffer.resize(maximum);
read_threshold = maximum;
break;
case io::receive_policy_flag::at_most:
if (receive_buffer.size() != maximum)
receive_buffer.resize(maximum);
read_threshold = 1;
break;
case io::receive_policy_flag::at_least: {
// read up to 10% more, but at least allow 100 bytes more
auto maximumsize = maximum + std::max<size_t>(100, maximum / 10);
if (receive_buffer.size() != maximumsize)
receive_buffer.resize(maximumsize);
read_threshold = maximum;
break;
}
}
}
void tcp_transport::configure_read(io::receive_policy::config config) {
rd_flag = config.first;
maximum = config.second;
}
error tcp_transport::write_some(io::network::event_handler* parent) {
CAF_LOG_TRACE("");
const void* buf = send_buffer.data() + written;
auto len = send_buffer.size() - written;
auto sres = ::send(parent->fd(),
reinterpret_cast<io::network::socket_send_ptr>(buf),
len, io::network::no_sigpipe_io_flag);
if (io::network::is_error(sres, true))
return sec::runtime_error;
size_t result = (sres > 0) ? static_cast<size_t>(sres) : 0;
written += result;
auto remaining = send_buffer.size() - written;
if (remaining == 0)
prepare_next_write(parent);
return none;
}
void tcp_transport::prepare_next_write(io::network::event_handler* parent) {
written = 0;
send_buffer.clear();
if (offline_buffer.empty()) {
parent->backend().del(io::network::operation::write,
parent->fd(), parent);
writing = false;
} else {
send_buffer.swap(offline_buffer);
}
}
void tcp_transport::flush(io::network::event_handler* parent) {
CAF_ASSERT(parent != nullptr);
CAF_LOG_TRACE(CAF_ARG(offline_buffer.size()));
if (!offline_buffer.empty() && !writing) {
parent->backend().add(io::network::operation::write,
parent->fd(), parent);
writing = true;
prepare_next_write(parent);
}
}
expected<io::network::native_socket>
tcp_transport::connect(const std::string& host, uint16_t port,
optional<io::network::protocol::network> preferred) {
auto res = io::network::new_tcp_connection(host, port, preferred);
if (!res)
std::cerr << "failed to create new TCP connection" << std::endl;
return res;
}
expected<io::network::native_socket>
accept_tcp::create_socket(uint16_t port,const char* host,bool reuse) {
return io::network::new_tcp_acceptor_impl(port, host, reuse);
}
std::pair<io::network::native_socket, io::network::transport_policy_ptr>
accept_tcp::accept(io::network::event_handler* parent) {
using namespace io::network;
sockaddr_storage addr;
std::memset(&addr, 0, sizeof(addr));
socket_size_type addrlen = sizeof(addr);
auto result = ::accept(parent->fd(), reinterpret_cast<sockaddr*>(&addr),
&addrlen);
if (result == invalid_native_socket) {
auto err = last_socket_error();
if (!would_block_or_temporarily_unavailable(err)) {
return {invalid_native_socket, nullptr};
}
}
std::cerr << "accepted connection" << std::endl;
transport_policy_ptr ptr{new tcp_transport};
return {result, std::move(ptr)};
}
void accept_tcp::init(io::network::newb_base& n) {
n.start();
}
} // 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/policy/newb_udp.hpp"
#include "caf/config.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 <io.h>
# include <windows.h>
# include <winsock2.h>
# include <ws2ipdef.h>
# include <ws2tcpip.h>
#else
# include <unistd.h>
# include <arpa/inet.h>
# include <cerrno>
# include <fcntl.h>
# include <netdb.h>
# include <netinet/in.h>
# include <netinet/ip.h>
# include <netinet/tcp.h>
# include <sys/socket.h>
# include <sys/types.h>
# ifdef CAF_POLL_MULTIPLEXER
# include <poll.h>
# elif defined(CAF_EPOLL_MULTIPLEXER)
# include <sys/epoll.h>
# else
# error "neither CAF_POLL_MULTIPLEXER nor CAF_EPOLL_MULTIPLEXER defined"
# endif
#endif
namespace caf {
namespace policy {
udp_transport::udp_transport()
: maximum{std::numeric_limits<uint16_t>::max()},
first_message{true},
writing{false},
written{0},
offline_sum{0} {
// nop
}
error udp_transport::read_some(io::network::event_handler* parent) {
CAF_LOG_TRACE(CAF_ARG(parent->fd()));
memset(sender.address(), 0, sizeof(sockaddr_storage));
io::network::socket_size_type len = sizeof(sockaddr_storage);
auto buf_ptr = static_cast<io::network::socket_recv_ptr>(receive_buffer.data());
auto buf_len = receive_buffer.size();
auto sres = ::recvfrom(parent->fd(), buf_ptr, buf_len,
0, sender.address(), &len);
if (io::network::is_error(sres, true)) {
CAF_LOG_ERROR("recvfrom returned" << CAF_ARG(sres));
return sec::runtime_error;
} else if (io::network::would_block_or_temporarily_unavailable(
io::network::last_socket_error())) {
return sec::end_of_stream;
}
if (sres == 0)
CAF_LOG_INFO("Received empty datagram");
else if (sres > static_cast<io::network::signed_size_type>(buf_len))
CAF_LOG_WARNING("recvfrom cut of message, only received "
<< CAF_ARG(buf_len) << " of " << CAF_ARG(sres) << " bytes");
received_bytes = (sres > 0) ? static_cast<size_t>(sres) : 0;
*sender.length() = static_cast<size_t>(len);
if (first_message) {
endpoint = sender;
first_message = false;
}
return none;
}
void udp_transport::prepare_next_read(io::network::event_handler*) {
received_bytes = 0;
receive_buffer.resize(maximum);
}
error udp_transport::write_some(io::network::event_handler* parent) {
std::cerr << "sending on socket: " << parent->fd() << std::endl;
using namespace caf::io::network;
CAF_LOG_TRACE(CAF_ARG(parent->fd()) << CAF_ARG(send_buffer.size()));
socket_size_type len = static_cast<socket_size_type>(*endpoint.clength());
auto buf_ptr = reinterpret_cast<socket_send_ptr>(send_buffer.data() + written);
auto buf_len = send_sizes.front();
auto sres = ::sendto(parent->fd(), buf_ptr, buf_len,
0, endpoint.caddress(), len);
std::cerr << "sent " << sres << " bytes to " << to_string(endpoint) << std::endl;
if (is_error(sres, true)) {
std::cerr << "sento failed: " << last_socket_error_as_string() << std::endl;
std::abort();
CAF_LOG_ERROR("sendto returned" << CAF_ARG(sres));
return sec::runtime_error;
}
send_sizes.pop_front();
written += (sres > 0) ? static_cast<size_t>(sres) : 0;
auto remaining = send_buffer.size() - written;
std::cerr << "wrote '" << written << "' got '" << remaining << "' bytes left to write" << std::endl;
if (remaining == 0)
prepare_next_write(parent);
return none;
}
void udp_transport::prepare_next_write(io::network::event_handler* parent) {
written = 0;
send_buffer.clear();
send_sizes.clear();
if (offline_buffer.empty()) {
writing = false;
parent->backend().del(io::network::operation::write,
parent->fd(), parent);
} else {
// Add size of last chunk.
offline_sizes.push_back(offline_buffer.size() - offline_sum);
// Switch buffers.
send_buffer.swap(offline_buffer);
send_sizes.swap(offline_sizes);
// Reset sum.
offline_sum = 0;
}
}
io::network::byte_buffer& udp_transport::wr_buf() {
if (!offline_buffer.empty()) {
auto chunk_size = offline_buffer.size() - offline_sum;
offline_sizes.push_back(chunk_size);
offline_sum += chunk_size;
std::cerr << "adding chunk '" << chunk_size << "' up to total of '" << offline_sum << std::endl;
}
return offline_buffer;
}
void udp_transport::flush(io::network::event_handler* parent) {
CAF_ASSERT(parent != nullptr);
CAF_LOG_TRACE(CAF_ARG(offline_buffer.size()));
if (!offline_buffer.empty() && !writing) {
parent->backend().add(io::network::operation::write,
parent->fd(), parent);
writing = true;
prepare_next_write(parent);
}
}
expected<io::network::native_socket>
udp_transport::connect(const std::string& host, uint16_t port,
optional<io::network::protocol::network> preferred) {
auto res = io::network::new_remote_udp_endpoint_impl(host, port, preferred);
if (!res)
return std::move(res.error());
endpoint = res->second;
return res->first;
}
expected<io::network::native_socket>
accept_udp::create_socket(uint16_t port, const char* host, bool reuse) {
auto res = io::network::new_local_udp_endpoint_impl(port, host, reuse);
if (!res)
return std::move(res.error());
return (*res).first;
}
std::pair<io::network::native_socket, io::network::transport_policy_ptr>
accept_udp::accept(io::network::event_handler*) {
auto res = io::network::new_local_udp_endpoint_impl(0, nullptr);
if (!res) {
CAF_LOG_DEBUG("failed to create local endpoint");
return {io::network::invalid_native_socket, nullptr};
}
auto sock = std::move(res->first);
io::network::transport_policy_ptr ptr{new udp_transport};
return {sock, std::move(ptr)};
}
void accept_udp::init(io::network::newb_base& n) {
n.start();
}
} // namespace policy
} // namespace caf
...@@ -302,7 +302,7 @@ struct dummy_basp_newb : network::newb<new_basp_message> { ...@@ -302,7 +302,7 @@ struct dummy_basp_newb : network::newb<new_basp_message> {
} }
}; };
struct accept_policy_impl : public network::accept_policy<new_basp_message> { struct accept_policy_impl : public network::accept_policy {
expected<native_socket> create_socket(uint16_t, const char*, bool) override { expected<native_socket> create_socket(uint16_t, const char*, bool) override {
return sec::bad_function_call; return sec::bad_function_call;
} }
...@@ -315,7 +315,7 @@ struct accept_policy_impl : public network::accept_policy<new_basp_message> { ...@@ -315,7 +315,7 @@ struct accept_policy_impl : public network::accept_policy<new_basp_message> {
return {invalid_native_socket, std::move(ptr)}; return {invalid_native_socket, std::move(ptr)};
} }
void init(network::newb<new_basp_message>& n) override { void init(network::newb_base& n) override {
n.handle_event(network::operation::read); n.handle_event(network::operation::read);
} }
}; };
......
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