Commit a7e3a939 authored by Joseph Noir's avatar Joseph Noir

Add message for io failures, some related cleanup

parent 5fe3b5f6
...@@ -49,6 +49,7 @@ set(LIBCAF_IO_SRCS ...@@ -49,6 +49,7 @@ set(LIBCAF_IO_SRCS
src/socket_guard.cpp src/socket_guard.cpp
src/newb_tcp.cpp src/newb_tcp.cpp
src/newb_udp.cpp src/newb_udp.cpp
src/newb.cpp
) )
add_custom_target(libcaf_io) add_custom_target(libcaf_io)
......
...@@ -36,6 +36,7 @@ ...@@ -36,6 +36,7 @@
#include "caf/io/network/event_handler.hpp" #include "caf/io/network/event_handler.hpp"
#include "caf/io/network/interfaces.hpp" #include "caf/io/network/interfaces.hpp"
#include "caf/io/network/native_socket.hpp" #include "caf/io/network/native_socket.hpp"
#include "caf/io/network/rw_state.hpp"
#include "caf/mixin/behavior_changer.hpp" #include "caf/mixin/behavior_changer.hpp"
#include "caf/mixin/requester.hpp" #include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp" #include "caf/mixin/sender.hpp"
...@@ -75,83 +76,75 @@ using header_writer = caf::callback<byte_buffer&>; ...@@ -75,83 +76,75 @@ using header_writer = caf::callback<byte_buffer&>;
// -- newb base ---------------------------------------------------------------- // -- newb base ----------------------------------------------------------------
struct newb_base : public network::event_handler { struct newb_base : public network::event_handler {
newb_base(default_multiplexer& dm, native_socket sockfd) newb_base(default_multiplexer& dm, native_socket sockfd);
: event_handler(dm, sockfd) {
// nop
}
virtual void start() = 0; virtual void start() = 0;
virtual void stop() = 0; virtual void stop() = 0;
virtual void flush() = 0;
virtual void io_error(operation op, error err) = 0;
}; };
// -- transport policy --------------------------------------------------------- // -- transport policy ---------------------------------------------------------
struct transport_policy { struct transport_policy {
transport_policy() : received_bytes{0}, max_consecutive_reads{50}, count{0} { transport_policy();
// nop
}
virtual ~transport_policy() { virtual ~transport_policy();
// nop
}
virtual error write_some(network::event_handler*) { virtual io::network::rw_state write_some(newb_base*);
return none;
}
virtual error read_some(network::event_handler*) { virtual io::network::rw_state read_some(newb_base*);
return none;
}
virtual bool should_deliver() { virtual bool should_deliver();
return true;
}
virtual void prepare_next_read(network::event_handler*) { virtual bool must_read_more(newb_base*);
// nop
}
virtual void prepare_next_write(network::event_handler*) { virtual void prepare_next_read(newb_base*);
// nop
}
virtual void configure_read(receive_policy::config) { virtual void prepare_next_write(newb_base*);
// nop
}
virtual void flush(network::event_handler*) { virtual void configure_read(receive_policy::config);
// nop
}
virtual byte_buffer& wr_buf() { virtual void flush(newb_base*);
return offline_buffer;
} virtual byte_buffer& wr_buf();
template <class T> template <class T>
error read_some(network::event_handler* parent, protocol_policy<T>& policy) { error read_some(newb_base* parent, protocol_policy<T>& policy) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
auto mcr = max_consecutive_reads; size_t reads = 0;
for (size_t i = 0; i < mcr; ++i) { while (reads < max_consecutive_reads || must_read_more(parent)) {
auto res = read_some(parent); auto read_result = read_some(parent);
// The return statements seems weird, needs cleanup. switch (read_result) {
if (res) case rw_state::success:
return res; // std::cout << "read success" << std::endl;
if (received_bytes == 0)
return none;
if (should_deliver()) { if (should_deliver()) {
res = policy.read(receive_buffer.data(), received_bytes); auto res = policy.read(receive_buffer.data(), received_bytes);
prepare_next_read(parent); prepare_next_read(parent);
if (!res) if (res)
return res; return res;
} }
break;
case rw_state::indeterminate:
std::cout << "read indeterminate" << std::endl;
// No error, but don't continue reading.
return none;
case rw_state::failure:
std::cout << "read failure" << std::endl;
// Reading failed.
return sec::runtime_error;
}
++reads;
} }
return none; return none;
} }
virtual expected<native_socket> virtual expected<native_socket>
connect(const std::string&, uint16_t, connect(const std::string&, uint16_t,
optional<io::network::protocol::network> = none) { optional<io::network::protocol::network> = none);
return sec::bad_function_call;
}
size_t received_bytes; size_t received_bytes;
size_t max_consecutive_reads; size_t max_consecutive_reads;
...@@ -159,7 +152,6 @@ struct transport_policy { ...@@ -159,7 +152,6 @@ struct transport_policy {
byte_buffer offline_buffer; byte_buffer offline_buffer;
byte_buffer receive_buffer; byte_buffer receive_buffer;
byte_buffer send_buffer; byte_buffer send_buffer;
size_t count;
}; };
using transport_policy_ptr = std::unique_ptr<transport_policy>; using transport_policy_ptr = std::unique_ptr<transport_policy>;
...@@ -167,19 +159,16 @@ using transport_policy_ptr = std::unique_ptr<transport_policy>; ...@@ -167,19 +159,16 @@ using transport_policy_ptr = std::unique_ptr<transport_policy>;
// -- accept policy ------------------------------------------------------------ // -- accept policy ------------------------------------------------------------
struct accept_policy { struct accept_policy {
virtual ~accept_policy() { virtual ~accept_policy();
// nop
}
virtual expected<native_socket> create_socket(uint16_t port, const char* host, virtual expected<native_socket> create_socket(uint16_t port, const char* host,
bool reuse = false) = 0; bool reuse = false) = 0;
virtual std::pair<native_socket, transport_policy_ptr> virtual std::pair<native_socket, transport_policy_ptr> accept(newb_base*) = 0;
accept(network::event_handler*) = 0;
virtual error multiplex_write() { virtual error multiplex_write() {
return none; return none;
}; }
virtual void init(newb_base&) = 0; virtual void init(newb_base&) = 0;
}; };
...@@ -187,9 +176,7 @@ struct accept_policy { ...@@ -187,9 +176,7 @@ struct accept_policy {
// -- protocol policy ---------------------------------------------------------- // -- protocol policy ----------------------------------------------------------
struct protocol_policy_base { struct protocol_policy_base {
virtual ~protocol_policy_base() { virtual ~protocol_policy_base();
// nop
}
virtual error read(char* bytes, size_t count) = 0; virtual error read(char* bytes, size_t count) = 0;
...@@ -380,6 +367,34 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -380,6 +367,34 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
passivate(); passivate();
} }
void io_error(operation op, error err) override {
auto mptr = make_mailbox_element(nullptr, invalid_message_id, {},
io_error_msg{op, std::move(err)});
switch (scheduled_actor::consume(*mptr)) {
case im_success:
this->finalize();
break;
case im_skipped:
this->push_to_cache(std::move(mptr));
break;
case im_dropped:
CAF_LOG_INFO("broker dropped read error message");
break;
}
switch (op) {
case operation::read:
passivate();
break;
case operation::write:
event_handler::backend().del(operation::write, fd(), this);
break;
case operation::propagate_error:
// TODO: What should happen here?
break;
}
}
/// Get a write buffer to write data to be sent by this broker.
write_handle<Message> wr_buf(header_writer* hw) { write_handle<Message> wr_buf(header_writer* hw) {
auto& buf = transport->wr_buf(); auto& buf = transport->wr_buf();
auto hstart = buf.size(); auto hstart = buf.size();
...@@ -392,23 +407,26 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -392,23 +407,26 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
return transport->wr_buf(); return transport->wr_buf();
} }
void flush() override { void flush() {
transport->flush(this); transport->flush(this);
} }
error read_event() { void read_event() {
return transport->read_some(this, *protocol); auto err = transport->read_some(this, *protocol);
if (err)
io_error(operation::read, std::move(err));
} }
void write_event() { void write_event() {
transport->write_some(this); if (transport->write_some(this) == rw_state::failure)
io_error(operation::write, sec::runtime_error);
} }
void handle_error() { void handle_error() {
CAF_CRITICAL("got error to handle: not implemented"); CAF_CRITICAL("got error to handle: not implemented");
} }
// Protocol policies can set timeouts using a custom message. /// Set a timeout for a protocol policy layer.
template<class Rep = int, class Period = std::ratio<1>> template<class Rep = int, class Period = std::ratio<1>>
void set_timeout(std::chrono::duration<Rep, Period> timeout, void set_timeout(std::chrono::duration<Rep, Period> timeout,
atom_value atm, uint32_t id) { atom_value atm, uint32_t id) {
...@@ -416,24 +434,22 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -416,24 +434,22 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
scheduled_actor::clock().set_ordinary_timeout(n + timeout, this, atm, id); scheduled_actor::clock().set_ordinary_timeout(n + timeout, this, atm, id);
} }
// Allow protocol policies to enqueue a data for sending. Probably required for
// reliability to send ACKs. The problem is that only the headers of policies
// lower, well closer to the transport, should write their headers. So we're
// facing a similiar porblem to slicing here.
// void (char* bytes, size_t count);
/// Returns the `multiplexer` running this broker. /// Returns the `multiplexer` running this broker.
network::multiplexer& backend() { network::multiplexer& backend() {
return event_handler::backend(); return event_handler::backend();
} }
/// Pass a message from a protocol policy layer to the broker for processing.
void handle(Message& m) { void handle(Message& m) {
std::swap(msg(), m); std::swap(msg(), m);
scheduled_actor::activate(scheduled_actor::context(), value_); scheduled_actor::activate(scheduled_actor::context(), value_);
} }
/// Override this to set the behavior of the broker.
virtual behavior make_behavior() = 0; virtual behavior make_behavior() = 0;
/// Configure the number of bytes read for the next packet. (Can be ignored by
/// the transport policy if its protocol does not support this functionality.)
void configure_read(receive_policy::config config) { void configure_read(receive_policy::config config) {
transport->configure_read(config); transport->configure_read(config);
} }
...@@ -448,13 +464,13 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -448,13 +464,13 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
ctx); ctx);
} }
/// @endcond
// -- policies --------------------------------------------------------------- // -- policies ---------------------------------------------------------------
std::unique_ptr<transport_policy> transport; std::unique_ptr<transport_policy> transport;
std::unique_ptr<protocol_policy<Message>> protocol; std::unique_ptr<protocol_policy<Message>> protocol;
/// @endcond
private: private:
// -- message element -------------------------------------------------------- // -- message element --------------------------------------------------------
...@@ -468,12 +484,12 @@ private: ...@@ -468,12 +484,12 @@ private:
// -- new broker acceptor ------------------------------------------------------ // -- new broker acceptor ------------------------------------------------------
template <class Message> template <class Message>
struct newb_acceptor : public network::event_handler { struct newb_acceptor : public newb_base {
// -- constructors and destructors ------------------------------------------- // -- constructors and destructors -------------------------------------------
newb_acceptor(default_multiplexer& dm, native_socket sockfd) newb_acceptor(default_multiplexer& dm, native_socket sockfd)
: event_handler(dm, sockfd) { : newb_base(dm, sockfd) {
// nop // nop
} }
...@@ -486,7 +502,8 @@ struct newb_acceptor : public network::event_handler { ...@@ -486,7 +502,8 @@ struct newb_acceptor : public network::event_handler {
read_event(); read_event();
break; break;
case network::operation::write: case network::operation::write:
// nop // Required to multiplex over a single socket.
write_event();
break; break;
case network::operation::propagate_error: case network::operation::propagate_error:
CAF_LOG_DEBUG("acceptor got error operation"); CAF_LOG_DEBUG("acceptor got error operation");
...@@ -505,34 +522,40 @@ struct newb_acceptor : public network::event_handler { ...@@ -505,34 +522,40 @@ struct newb_acceptor : public network::event_handler {
// -- members ---------------------------------------------------------------- // -- members ----------------------------------------------------------------
error read_event() { void read_event() {
native_socket sock; native_socket sock;
transport_policy_ptr transport; transport_policy_ptr transport;
std::tie(sock, transport) = acceptor->accept(this); std::tie(sock, transport) = acceptor->accept(this);
auto en = create_newb(sock, std::move(transport)); auto en = create_newb(sock, std::move(transport));
if (!en) if (!en) {
return std::move(en.error()); io_error(operation::read, std::move(en.error()));
return;
}
auto ptr = caf::actor_cast<caf::abstract_actor*>(*en); auto ptr = caf::actor_cast<caf::abstract_actor*>(*en);
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
auto& ref = dynamic_cast<newb<Message>&>(*ptr); auto& ref = dynamic_cast<newb<Message>&>(*ptr);
acceptor->init(ref); acceptor->init(ref);
return none;
} }
error write_event() { void write_event() {
return acceptor->multiplex_write(); acceptor->multiplex_write();
} }
void start() { void start() override {
event_handler::activate(); event_handler::activate();
} }
void stop() { void stop() override {
CAF_LOG_TRACE(CAF_ARG2("fd", fd())); CAF_LOG_TRACE(CAF_ARG2("fd", fd()));
close_read_channel(); close_read_channel();
passivate(); passivate();
} }
void io_error(operation op, error err) override {
std::cerr << "operation " << to_string(op) << " failed: "
<< backend().system().render(err) << std::endl;
}
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;
......
...@@ -30,8 +30,11 @@ ...@@ -30,8 +30,11 @@
#include "caf/io/accept_handle.hpp" #include "caf/io/accept_handle.hpp"
#include "caf/io/datagram_handle.hpp" #include "caf/io/datagram_handle.hpp"
#include "caf/io/connection_handle.hpp" #include "caf/io/connection_handle.hpp"
#include "caf/io/network/operation.hpp"
#include "caf/io/network/receive_buffer.hpp" #include "caf/io/network/receive_buffer.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
...@@ -182,6 +185,17 @@ inspect(Inspector& f, datagram_servant_closed_msg& x) { ...@@ -182,6 +185,17 @@ inspect(Inspector& f, datagram_servant_closed_msg& x) {
return f(meta::type_name("datagram_servant_closed_msg"), x.handles); return f(meta::type_name("datagram_servant_closed_msg"), x.handles);
} }
// Signals a read error to a newb.
struct io_error_msg {
caf::io::network::operation op;
error err;
};
template <class Inspector>
typename Inspector::result_type
inspect(Inspector& f, io_error_msg& x) {
return f(meta::type_name("io_error_msg"), x.op, x.err);
}
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -140,8 +140,6 @@ struct ordering { ...@@ -140,8 +140,6 @@ struct ordering {
} }
error timeout(atom_value atm, uint32_t id) { error timeout(atom_value atm, uint32_t id) {
std::cout << "got timeout message" << std::endl;
std::abort();
if (atm == ordering_atom::value) { if (atm == ordering_atom::value) {
error err = none; error err = none;
sequence_type seq = static_cast<sequence_type>(id); sequence_type seq = static_cast<sequence_type>(id);
......
...@@ -27,19 +27,19 @@ namespace policy { ...@@ -27,19 +27,19 @@ namespace policy {
struct tcp_transport : public io::network::transport_policy { struct tcp_transport : public io::network::transport_policy {
tcp_transport(); tcp_transport();
error read_some(io::network::event_handler* parent) override; io::network::rw_state read_some(io::network::newb_base* parent) override;
bool should_deliver() override; bool should_deliver() override;
void prepare_next_read(io::network::event_handler*) override; void prepare_next_read(io::network::newb_base*) override;
void configure_read(io::receive_policy::config config) override; void configure_read(io::receive_policy::config config) override;
error write_some(io::network::event_handler* parent) override; io::network::rw_state write_some(io::network::newb_base* parent) override;
void prepare_next_write(io::network::event_handler* parent) override; void prepare_next_write(io::network::newb_base* parent) override;
void flush(io::network::event_handler* parent) override; void flush(io::network::newb_base* parent) override;
expected<io::network::native_socket> expected<io::network::native_socket>
connect(const std::string& host, uint16_t port, connect(const std::string& host, uint16_t port,
...@@ -61,7 +61,7 @@ struct accept_tcp : public io::network::accept_policy { ...@@ -61,7 +61,7 @@ struct accept_tcp : public io::network::accept_policy {
create_socket(uint16_t port,const char* host,bool reuse = false) override; create_socket(uint16_t port,const char* host,bool reuse = false) override;
std::pair<io::network::native_socket, io::network::transport_policy_ptr> std::pair<io::network::native_socket, io::network::transport_policy_ptr>
accept(io::network::event_handler* parent) override; accept(io::network::newb_base* parent) override;
void init(io::network::newb_base& n) override; void init(io::network::newb_base& n) override;
}; };
......
...@@ -27,26 +27,26 @@ namespace policy { ...@@ -27,26 +27,26 @@ namespace policy {
struct udp_transport : public io::network::transport_policy { struct udp_transport : public io::network::transport_policy {
udp_transport(); udp_transport();
error read_some(io::network::event_handler* parent) override; io::network::rw_state read_some(io::network::newb_base* parent) override;
inline bool should_deliver() override { inline bool should_deliver() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
return received_bytes != 0 && sender == endpoint; return received_bytes != 0 && sender == endpoint;
} }
void prepare_next_read(io::network::event_handler*) override; void prepare_next_read(io::network::newb_base*) override;
inline void configure_read(io::receive_policy::config) override { inline void configure_read(io::receive_policy::config) override {
// nop // nop
} }
error write_some(io::network::event_handler* parent) override; io::network::rw_state write_some(io::network::newb_base* parent) override;
void prepare_next_write(io::network::event_handler* parent) override; void prepare_next_write(io::network::newb_base* parent) override;
io::network::byte_buffer& wr_buf() override; io::network::byte_buffer& wr_buf() override;
void flush(io::network::event_handler* parent) override; void flush(io::network::newb_base* parent) override;
expected<io::network::native_socket> expected<io::network::native_socket>
connect(const std::string& host, uint16_t port, connect(const std::string& host, uint16_t port,
...@@ -73,7 +73,7 @@ struct accept_udp : public io::network::accept_policy { ...@@ -73,7 +73,7 @@ struct accept_udp : public io::network::accept_policy {
create_socket(uint16_t port, const char* host, bool reuse = false) override; create_socket(uint16_t port, const char* host, bool reuse = false) override;
std::pair<io::network::native_socket, io::network::transport_policy_ptr> std::pair<io::network::native_socket, io::network::transport_policy_ptr>
accept(io::network::event_handler*) override; accept(io::network::newb_base*) override;
void init(io::network::newb_base& n) override; void init(io::network::newb_base& n) override;
}; };
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/newb.hpp"
namespace caf {
namespace io {
namespace network {
// -- newb_base ----------------------------------------------------------------
newb_base::newb_base(default_multiplexer& dm, native_socket sockfd)
: event_handler(dm, sockfd) {
// nop
}
// -- transport_policy ---------------------------------------------------------
transport_policy::transport_policy()
: received_bytes{0},
max_consecutive_reads{50} {
// nop
}
transport_policy::~transport_policy() {
// nop
}
io::network::rw_state transport_policy::write_some(newb_base*) {
return io::network::rw_state::indeterminate;
}
io::network::rw_state transport_policy::read_some(newb_base*) {
return io::network::rw_state::indeterminate;
}
bool transport_policy::should_deliver() {
return true;
}
bool transport_policy::must_read_more(newb_base*) {
return false;
}
void transport_policy::prepare_next_read(newb_base*) {
// nop
}
void transport_policy::prepare_next_write(newb_base*) {
// nop
}
void transport_policy::configure_read(receive_policy::config) {
// nop
}
void transport_policy::flush(newb_base*) {
// nop
}
byte_buffer& transport_policy::wr_buf() {
return offline_buffer;
}
expected<native_socket>
transport_policy::connect(const std::string&, uint16_t,
optional<io::network::protocol::network>) {
return sec::bad_function_call;
}
// -- accept_policy ------------------------------------------------------------
accept_policy::~accept_policy() {
// nop
}
// -- protocol_policy_base -----------------------------------------------------
protocol_policy_base::~protocol_policy_base() {
// nop
}
} // namespace network
} // namespace io
} // namespace caf
...@@ -71,7 +71,7 @@ tcp_transport::tcp_transport() ...@@ -71,7 +71,7 @@ tcp_transport::tcp_transport()
// nop // nop
} }
error tcp_transport::read_some(io::network::event_handler* parent) { io::network::rw_state tcp_transport::read_some(io::network::newb_base* parent) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
size_t len = receive_buffer.size() - collected; size_t len = receive_buffer.size() - collected;
void* buf = receive_buffer.data() + collected; void* buf = receive_buffer.data() + collected;
...@@ -79,14 +79,14 @@ error tcp_transport::read_some(io::network::event_handler* parent) { ...@@ -79,14 +79,14 @@ error tcp_transport::read_some(io::network::event_handler* parent) {
reinterpret_cast<io::network::socket_recv_ptr>(buf), reinterpret_cast<io::network::socket_recv_ptr>(buf),
len, io::network::no_sigpipe_io_flag); len, io::network::no_sigpipe_io_flag);
if (io::network::is_error(sres, true) || sres == 0) { 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.
// recv returns 0 when the peer has performed an orderly shutdown CAF_LOG_DEBUG("recv failed" << CAF_ARG(sres));
return sec::runtime_error; return io::network::rw_state::failure;
} }
size_t result = (sres > 0) ? static_cast<size_t>(sres) : 0; size_t result = (sres > 0) ? static_cast<size_t>(sres) : 0;
collected += result; collected += result;
received_bytes = collected; received_bytes = collected;
return none; return io::network::rw_state::success;
} }
bool tcp_transport::should_deliver() { bool tcp_transport::should_deliver() {
...@@ -94,7 +94,7 @@ bool tcp_transport::should_deliver() { ...@@ -94,7 +94,7 @@ bool tcp_transport::should_deliver() {
return collected >= read_threshold; return collected >= read_threshold;
} }
void tcp_transport::prepare_next_read(io::network::event_handler*) { void tcp_transport::prepare_next_read(io::network::newb_base*) {
collected = 0; collected = 0;
received_bytes = 0; received_bytes = 0;
switch (rd_flag) { switch (rd_flag) {
...@@ -109,7 +109,7 @@ void tcp_transport::prepare_next_read(io::network::event_handler*) { ...@@ -109,7 +109,7 @@ void tcp_transport::prepare_next_read(io::network::event_handler*) {
read_threshold = 1; read_threshold = 1;
break; break;
case io::receive_policy_flag::at_least: { case io::receive_policy_flag::at_least: {
// read up to 10% more, but at least allow 100 bytes more // Read up to 10% more, but at least allow 100 bytes more.
auto maximumsize = maximum + std::max<size_t>(100, maximum / 10); auto maximumsize = maximum + std::max<size_t>(100, maximum / 10);
if (receive_buffer.size() != maximumsize) if (receive_buffer.size() != maximumsize)
receive_buffer.resize(maximumsize); receive_buffer.resize(maximumsize);
...@@ -124,25 +124,27 @@ void tcp_transport::configure_read(io::receive_policy::config config) { ...@@ -124,25 +124,27 @@ void tcp_transport::configure_read(io::receive_policy::config config) {
maximum = config.second; maximum = config.second;
} }
error tcp_transport::write_some(io::network::event_handler* parent) { io::network::rw_state tcp_transport::write_some(io::network::newb_base* parent) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
const void* buf = send_buffer.data() + written; const void* buf = send_buffer.data() + written;
auto len = send_buffer.size() - written; auto len = send_buffer.size() - written;
auto sres = ::send(parent->fd(), auto sres = ::send(parent->fd(),
reinterpret_cast<io::network::socket_send_ptr>(buf), reinterpret_cast<io::network::socket_send_ptr>(buf),
len, io::network::no_sigpipe_io_flag); len, io::network::no_sigpipe_io_flag);
if (io::network::is_error(sres, true)) if (io::network::is_error(sres, true)) {
return sec::runtime_error; CAF_LOG_ERROR("send failed"
<< CAF_ARG(io::network::last_socket_error_as_string()));
return io::network::rw_state::failure;
}
size_t result = (sres > 0) ? static_cast<size_t>(sres) : 0; size_t result = (sres > 0) ? static_cast<size_t>(sres) : 0;
written += result; written += result;
count += 1;
auto remaining = send_buffer.size() - written; auto remaining = send_buffer.size() - written;
if (remaining == 0) if (remaining == 0)
prepare_next_write(parent); prepare_next_write(parent);
return none; return io::network::rw_state::success;
} }
void tcp_transport::prepare_next_write(io::network::event_handler* parent) { void tcp_transport::prepare_next_write(io::network::newb_base* parent) {
written = 0; written = 0;
send_buffer.clear(); send_buffer.clear();
if (offline_buffer.empty()) { if (offline_buffer.empty()) {
...@@ -154,7 +156,7 @@ void tcp_transport::prepare_next_write(io::network::event_handler* parent) { ...@@ -154,7 +156,7 @@ void tcp_transport::prepare_next_write(io::network::event_handler* parent) {
} }
} }
void tcp_transport::flush(io::network::event_handler* parent) { void tcp_transport::flush(io::network::newb_base* parent) {
CAF_ASSERT(parent != nullptr); CAF_ASSERT(parent != nullptr);
CAF_LOG_TRACE(CAF_ARG(offline_buffer.size())); CAF_LOG_TRACE(CAF_ARG(offline_buffer.size()));
if (!offline_buffer.empty() && !writing) { if (!offline_buffer.empty() && !writing) {
...@@ -180,7 +182,7 @@ accept_tcp::create_socket(uint16_t port, const char* host, bool reuse) { ...@@ -180,7 +182,7 @@ accept_tcp::create_socket(uint16_t port, const char* host, bool reuse) {
} }
std::pair<io::network::native_socket, io::network::transport_policy_ptr> std::pair<io::network::native_socket, io::network::transport_policy_ptr>
accept_tcp::accept(io::network::event_handler* parent) { accept_tcp::accept(io::network::newb_base* parent) {
using namespace io::network; using namespace io::network;
sockaddr_storage addr; sockaddr_storage addr;
std::memset(&addr, 0, sizeof(addr)); std::memset(&addr, 0, sizeof(addr));
......
...@@ -71,7 +71,7 @@ udp_transport::udp_transport() ...@@ -71,7 +71,7 @@ udp_transport::udp_transport()
// nop // nop
} }
error udp_transport::read_some(io::network::event_handler* parent) { io::network::rw_state udp_transport::read_some(io::network::newb_base* parent) {
CAF_LOG_TRACE(CAF_ARG(parent->fd())); CAF_LOG_TRACE(CAF_ARG(parent->fd()));
memset(sender.address(), 0, sizeof(sockaddr_storage)); memset(sender.address(), 0, sizeof(sockaddr_storage));
io::network::socket_size_type len = sizeof(sockaddr_storage); io::network::socket_size_type len = sizeof(sockaddr_storage);
...@@ -80,11 +80,8 @@ error udp_transport::read_some(io::network::event_handler* parent) { ...@@ -80,11 +80,8 @@ error udp_transport::read_some(io::network::event_handler* parent) {
auto sres = ::recvfrom(parent->fd(), buf_ptr, buf_len, auto sres = ::recvfrom(parent->fd(), buf_ptr, buf_len,
0, sender.address(), &len); 0, sender.address(), &len);
if (io::network::is_error(sres, true)) { if (io::network::is_error(sres, true)) {
CAF_LOG_ERROR("recvfrom returned" << CAF_ARG(sres)); CAF_LOG_ERROR("recvfrom failed" << CAF_ARG(sres));
return sec::runtime_error; return io::network::rw_state::failure;
} else if (io::network::would_block_or_temporarily_unavailable(
io::network::last_socket_error())) {
return sec::end_of_stream;
} }
if (sres == 0) if (sres == 0)
CAF_LOG_INFO("Received empty datagram"); CAF_LOG_INFO("Received empty datagram");
...@@ -97,15 +94,15 @@ error udp_transport::read_some(io::network::event_handler* parent) { ...@@ -97,15 +94,15 @@ error udp_transport::read_some(io::network::event_handler* parent) {
endpoint = sender; endpoint = sender;
first_message = false; first_message = false;
} }
return none; return io::network::rw_state::success;
} }
void udp_transport::prepare_next_read(io::network::event_handler*) { void udp_transport::prepare_next_read(io::network::newb_base*) {
received_bytes = 0; received_bytes = 0;
receive_buffer.resize(maximum); receive_buffer.resize(maximum);
} }
error udp_transport::write_some(io::network::event_handler* parent) { io::network::rw_state udp_transport::write_some(io::network::newb_base* parent) {
using namespace caf::io::network; using namespace caf::io::network;
CAF_LOG_TRACE(CAF_ARG(parent->fd()) << CAF_ARG(send_buffer.size())); CAF_LOG_TRACE(CAF_ARG(parent->fd()) << CAF_ARG(send_buffer.size()));
socket_size_type len = static_cast<socket_size_type>(*endpoint.clength()); socket_size_type len = static_cast<socket_size_type>(*endpoint.clength());
...@@ -114,22 +111,24 @@ error udp_transport::write_some(io::network::event_handler* parent) { ...@@ -114,22 +111,24 @@ error udp_transport::write_some(io::network::event_handler* parent) {
auto sres = ::sendto(parent->fd(), buf_ptr, buf_len, auto sres = ::sendto(parent->fd(), buf_ptr, buf_len,
0, endpoint.caddress(), len); 0, endpoint.caddress(), len);
if (is_error(sres, true)) { if (is_error(sres, true)) {
std::cerr << "sento failed: " << last_socket_error_as_string() << std::endl; CAF_LOG_ERROR("sendto failed" << CAF_ARG(sres));
std::abort(); return io::network::rw_state::failure;
CAF_LOG_ERROR("sendto returned" << CAF_ARG(sres));
return sec::runtime_error;
} }
size_t sent_bytes = (sres > 0) ? static_cast<size_t>(sres) : 0;
// TODO: This only works if we always write send_sizes.front() // TODO: This only works if we always write send_sizes.front()
if (sent_bytes != buf_len) {
CAF_LOG_ERROR("failed to send complete datagram");
return io::network::rw_state::failure;
}
send_sizes.pop_front(); send_sizes.pop_front();
written += (sres > 0) ? static_cast<size_t>(sres) : 0; written += sent_bytes;
auto remaining = send_buffer.size() - written; auto remaining = send_buffer.size() - written;
count += 1;
if (remaining == 0) if (remaining == 0)
prepare_next_write(parent); prepare_next_write(parent);
return none; return io::network::rw_state::success;
} }
void udp_transport::prepare_next_write(io::network::event_handler* parent) { void udp_transport::prepare_next_write(io::network::newb_base* parent) {
written = 0; written = 0;
send_buffer.clear(); send_buffer.clear();
send_sizes.clear(); send_sizes.clear();
...@@ -157,7 +156,7 @@ io::network::byte_buffer& udp_transport::wr_buf() { ...@@ -157,7 +156,7 @@ io::network::byte_buffer& udp_transport::wr_buf() {
return offline_buffer; return offline_buffer;
} }
void udp_transport::flush(io::network::event_handler* parent) { void udp_transport::flush(io::network::newb_base* parent) {
CAF_ASSERT(parent != nullptr); CAF_ASSERT(parent != nullptr);
CAF_LOG_TRACE(CAF_ARG(offline_buffer.size())); CAF_LOG_TRACE(CAF_ARG(offline_buffer.size()));
if (!offline_buffer.empty() && !writing) { if (!offline_buffer.empty() && !writing) {
...@@ -187,7 +186,7 @@ accept_udp::create_socket(uint16_t port, const char* host, bool reuse) { ...@@ -187,7 +186,7 @@ accept_udp::create_socket(uint16_t port, const char* host, bool reuse) {
} }
std::pair<io::network::native_socket, io::network::transport_policy_ptr> std::pair<io::network::native_socket, io::network::transport_policy_ptr>
accept_udp::accept(io::network::event_handler*) { accept_udp::accept(io::network::newb_base*) {
auto res = io::network::new_local_udp_endpoint_impl(0, nullptr); auto res = io::network::new_local_udp_endpoint_impl(0, nullptr);
if (!res) { if (!res) {
CAF_LOG_DEBUG("failed to create local endpoint"); CAF_LOG_DEBUG("failed to create local endpoint");
......
...@@ -300,7 +300,7 @@ struct accept_policy_impl : public network::accept_policy { ...@@ -300,7 +300,7 @@ struct accept_policy_impl : public network::accept_policy {
} }
std::pair<native_socket, network::transport_policy_ptr> std::pair<native_socket, network::transport_policy_ptr>
accept(network::event_handler*) override { accept(network::newb_base*) override {
// TODO: For UDP read the message into a buffer. Create a new socket. // TODO: For UDP read the message into a buffer. Create a new socket.
// Move the buffer into the transport policy as the new receive buffer. // Move the buffer into the transport policy as the new receive buffer.
network::transport_policy_ptr ptr{new network::transport_policy}; network::transport_policy_ptr ptr{new network::transport_policy};
...@@ -427,8 +427,7 @@ CAF_TEST(read event) { ...@@ -427,8 +427,7 @@ CAF_TEST(read event) {
bs(payload); bs(payload);
dummy.transport->received_bytes = buf.size(); dummy.transport->received_bytes = buf.size();
CAF_MESSAGE("trigger a read event"); CAF_MESSAGE("trigger a read event");
auto err = dummy.read_event(); dummy.read_event();
CAF_REQUIRE(!err);
CAF_MESSAGE("check the basp header and payload"); CAF_MESSAGE("check the basp header and payload");
CAF_REQUIRE(!dummy.messages.empty()); CAF_REQUIRE(!dummy.messages.empty());
auto& msg = dummy.messages.front().first; auto& msg = dummy.messages.front().first;
...@@ -475,8 +474,7 @@ CAF_TEST(timeouts) { ...@@ -475,8 +474,7 @@ CAF_TEST(timeouts) {
CAF_MESSAGE("trigger read event"); CAF_MESSAGE("trigger read event");
auto& dummy = deref<dummy_basp_newb>(self); auto& dummy = deref<dummy_basp_newb>(self);
dummy.transport->max_consecutive_reads = 1; dummy.transport->max_consecutive_reads = 1;
auto err = dummy.read_event(); dummy.read_event();
CAF_REQUIRE(!err);
CAF_CHECK(!dummy.expected.empty()); CAF_CHECK(!dummy.expected.empty());
CAF_MESSAGE("trigger waiting timeouts"); CAF_MESSAGE("trigger waiting timeouts");
// Trigger timeout. // Trigger timeout.
......
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