Commit 1b25ff29 authored by Dominik Charousset's avatar Dominik Charousset

Add send ACKs for brokers

parent 8181610d
...@@ -99,6 +99,9 @@ public: ...@@ -99,6 +99,9 @@ public:
/// @param config Contains the new receive policy. /// @param config Contains the new receive policy.
void configure_read(connection_handle hdl, receive_policy::config config); void configure_read(connection_handle hdl, receive_policy::config config);
/// Enables or disables write notifications for given connection.
void ack_writes(connection_handle hdl, bool enable);
/// Returns the write buffer for given connection. /// Returns the write buffer for given connection.
std::vector<char>& wr_buf(connection_handle hdl); std::vector<char>& wr_buf(connection_handle hdl);
......
...@@ -25,6 +25,7 @@ ...@@ -25,6 +25,7 @@
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/stateful_actor.hpp"
#include "caf/abstract_event_based_actor.hpp" #include "caf/abstract_event_based_actor.hpp"
#include "caf/io/scribe.hpp" #include "caf/io/scribe.hpp"
...@@ -70,6 +71,10 @@ protected: ...@@ -70,6 +71,10 @@ protected:
using broker_ptr = intrusive_ptr<broker>; using broker_ptr = intrusive_ptr<broker>;
/// Convenience template alias for declaring state-based brokers.
template <class State>
using stateful_broker = stateful_actor<State, broker>;
} // namespace io } // namespace io
} // namespace caf } // namespace caf
......
...@@ -133,6 +133,7 @@ public: ...@@ -133,6 +133,7 @@ public:
asio_stream(asio_multiplexer& ref) asio_stream(asio_multiplexer& ref)
: writing_(false), : writing_(false),
ack_writes_(false),
fd_(ref.service()), fd_(ref.service()),
backend_(ref) { backend_(ref) {
configure_read(receive_policy::at_most(1024)); configure_read(receive_policy::at_most(1024));
...@@ -167,6 +168,11 @@ public: ...@@ -167,6 +168,11 @@ public:
rd_size_ = config.second; rd_size_ = config.second;
} }
void ack_writes(bool enable) {
CAF_LOG_TRACE(CAF_ARG(enable));
ack_writes_ = enable;
}
/// Copies data to the write buffer. /// Copies data to the write buffer.
/// @note Not thread safe. /// @note Not thread safe.
void write(const void* buf, size_t num_bytes) { void write(const void* buf, size_t num_bytes) {
...@@ -263,15 +269,16 @@ private: ...@@ -263,15 +269,16 @@ private:
fd_, boost::asio::buffer(wr_buf_), fd_, boost::asio::buffer(wr_buf_),
[=](const boost::system::error_code& ec, size_t nb) { [=](const boost::system::error_code& ec, size_t nb) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
static_cast<void>(nb); // silence compiler warning if (ec) {
if (! ec) {
CAF_LOG_DEBUG(CAF_ARG(nb));
write_loop(mgr);
} else {
CAF_LOG_DEBUG(CAF_ARG(ec.message())); CAF_LOG_DEBUG(CAF_ARG(ec.message()));
mgr->io_failure(&backend(), operation::read); mgr->io_failure(&backend(), operation::read);
writing_ = false; writing_ = false;
return;
} }
CAF_LOG_DEBUG(CAF_ARG(nb));
if (ack_writes_)
mgr->data_transferred(&backend(), nb, wr_offline_buf_.size());
write_loop(mgr);
}); });
} }
...@@ -295,6 +302,7 @@ private: ...@@ -295,6 +302,7 @@ private:
} }
bool writing_; bool writing_;
bool ack_writes_;
Socket fd_; Socket fd_;
receive_policy_flag rd_flag_; receive_policy_flag rd_flag_;
size_t rd_size_; size_t rd_size_;
......
...@@ -134,6 +134,10 @@ connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self, ...@@ -134,6 +134,10 @@ connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self,
launch(); launch();
} }
} }
void ack_writes(bool enable) override {
CAF_LOG_TRACE(CAF_ARG(enable));
stream_.ack_writes(enable);
}
std::vector<char>& wr_buf() override { std::vector<char>& wr_buf() override {
return stream_.wr_buf(); return stream_.wr_buf();
} }
......
...@@ -406,8 +406,9 @@ public: ...@@ -406,8 +406,9 @@ public:
stream(default_multiplexer& backend_ref) stream(default_multiplexer& backend_ref)
: event_handler(backend_ref), : event_handler(backend_ref),
sock_(backend_ref), sock_(backend_ref),
threshold_(1), read_threshold_(1),
collected_(0), collected_(0),
ack_writes_(false),
writing_(false), writing_(false),
written_(0) { written_(0) {
configure_read(receive_policy::at_most(1024)); configure_read(receive_policy::at_most(1024));
...@@ -447,6 +448,10 @@ public: ...@@ -447,6 +448,10 @@ public:
max_ = config.second; max_ = config.second;
} }
void ack_writes(bool x) {
ack_writes_ = x;
}
/// Copies data to the write buffer. /// Copies data to the write buffer.
/// @warning Not thread safe. /// @warning Not thread safe.
void write(const void* buf, size_t num_bytes) { void write(const void* buf, size_t num_bytes) {
...@@ -500,7 +505,7 @@ public: ...@@ -500,7 +505,7 @@ public:
backend().del(operation::read, sock_.fd(), this); backend().del(operation::read, sock_.fd(), this);
} else if (rb > 0) { } else if (rb > 0) {
collected_ += rb; collected_ += rb;
if (collected_ >= threshold_) { if (collected_ >= read_threshold_) {
reader_->consume(&backend(), rd_buf_.data(), collected_); reader_->consume(&backend(), rd_buf_.data(), collected_);
read_loop(); read_loop();
} }
...@@ -514,23 +519,24 @@ public: ...@@ -514,23 +519,24 @@ public:
wr_buf_.size() - written_)) { wr_buf_.size() - written_)) {
writer_->io_failure(&backend(), operation::write); writer_->io_failure(&backend(), operation::write);
backend().del(operation::write, sock_.fd(), this); backend().del(operation::write, sock_.fd(), this);
} } else if (wb > 0) {
else if (wb > 0) {
written_ += wb; written_ += wb;
if (written_ >= wr_buf_.size()) { 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) // prepare next send (or stop sending)
if (remaining == 0)
write_loop(); write_loop();
} }
}
break; break;
} }
case operation::propagate_error: case operation::propagate_error:
if (reader_) { if (reader_)
reader_->io_failure(&backend(), operation::read); reader_->io_failure(&backend(), operation::read);
} if (writer_)
if (writer_) {
writer_->io_failure(&backend(), operation::write); writer_->io_failure(&backend(), operation::write);
}
// backend will delete this handler anyway, // backend will delete this handler anyway,
// no need to call backend().del() here // no need to call backend().del() here
break; break;
...@@ -549,13 +555,13 @@ private: ...@@ -549,13 +555,13 @@ private:
if (rd_buf_.size() != max_) { if (rd_buf_.size() != max_) {
rd_buf_.resize(max_); rd_buf_.resize(max_);
} }
threshold_ = max_; read_threshold_ = max_;
break; break;
case receive_policy_flag::at_most: case receive_policy_flag::at_most:
if (rd_buf_.size() != max_) { if (rd_buf_.size() != max_) {
rd_buf_.resize(max_); rd_buf_.resize(max_);
} }
threshold_ = 1; read_threshold_ = 1;
break; break;
case receive_policy_flag::at_least: { case 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
...@@ -564,7 +570,7 @@ private: ...@@ -564,7 +570,7 @@ private:
if (rd_buf_.size() != max_size) { if (rd_buf_.size() != max_size) {
rd_buf_.resize(max_size); rd_buf_.resize(max_size);
} }
threshold_ = max_; read_threshold_ = max_;
break; break;
} }
} }
...@@ -586,13 +592,14 @@ private: ...@@ -586,13 +592,14 @@ private:
Socket sock_; Socket sock_;
// reading // reading
manager_ptr reader_; manager_ptr reader_;
size_t threshold_; size_t read_threshold_;
size_t collected_; size_t collected_;
size_t max_; size_t max_;
receive_policy_flag rd_flag_; receive_policy_flag rd_flag_;
buffer_type rd_buf_; buffer_type rd_buf_;
// writing // writing
manager_ptr writer_; manager_ptr writer_;
bool ack_writes_;
bool writing_; bool writing_;
size_t written_; size_t written_;
buffer_type wr_buf_; buffer_type wr_buf_;
......
...@@ -36,8 +36,12 @@ public: ...@@ -36,8 +36,12 @@ public:
~stream_manager(); ~stream_manager();
/// Called by the underlying IO device whenever it received data. /// Called by the underlying I/O device whenever it received data.
virtual void consume(execution_unit* ctx, const void* buf, size_t bsize) = 0; virtual void consume(execution_unit* ctx, const void* buf, size_t bsize) = 0;
/// Called by the underlying I/O device whenever it sent data.
virtual void data_transferred(execution_unit* ctx, size_t num_bytes,
size_t remaining_bytes) = 0;
}; };
} // namespace network } // namespace network
......
...@@ -78,8 +78,12 @@ public: ...@@ -78,8 +78,12 @@ public:
/// Returns the input buffer of the scribe identified by `hdl`. /// Returns the input buffer of the scribe identified by `hdl`.
buffer_type& input_buffer(connection_handle hdl); buffer_type& input_buffer(connection_handle hdl);
/// Returns the configured read policy of the scribe identified by `hdl`.
receive_policy::config& read_config(connection_handle hdl); receive_policy::config& read_config(connection_handle hdl);
/// Returns whether the scribe identified by `hdl` receives write ACKs.
bool& ack_writes(connection_handle hdl);
/// Returns `true` if this handle has been closed /// Returns `true` if this handle has been closed
/// for reading, `false` otherwise. /// for reading, `false` otherwise.
bool& stopped_reading(connection_handle hdl); bool& stopped_reading(connection_handle hdl);
...@@ -144,6 +148,7 @@ private: ...@@ -144,6 +148,7 @@ private:
receive_policy::config recv_conf; receive_policy::config recv_conf;
bool stopped_reading = false; bool stopped_reading = false;
intrusive_ptr<scribe> ptr; intrusive_ptr<scribe> ptr;
bool ack_writes = false;
}; };
struct doorman_data { struct doorman_data {
......
...@@ -46,6 +46,9 @@ public: ...@@ -46,6 +46,9 @@ public:
/// Implicitly starts the read loop on first call. /// Implicitly starts the read loop on first call.
virtual void configure_read(receive_policy::config config) = 0; virtual void configure_read(receive_policy::config config) = 0;
/// Enables or disables write notifications.
virtual void ack_writes(bool enable) = 0;
/// Returns the current output buffer. /// Returns the current output buffer.
virtual std::vector<char>& wr_buf() = 0; virtual std::vector<char>& wr_buf() = 0;
...@@ -58,7 +61,9 @@ public: ...@@ -58,7 +61,9 @@ public:
void io_failure(execution_unit* ctx, network::operation op) override; void io_failure(execution_unit* ctx, network::operation op) override;
void consume(execution_unit* ctx, const void* buf, size_t buf_size) override; void consume(execution_unit*, const void*, size_t) override;
void data_transferred(execution_unit*, size_t, size_t) override;
protected: protected:
message detach_message() override; message detach_message() override;
......
...@@ -73,6 +73,7 @@ struct new_data_msg { ...@@ -73,6 +73,7 @@ struct new_data_msg {
std::vector<char> buf; std::vector<char> buf;
}; };
/// @relates new_data_msg
inline std::string to_string(const new_data_msg& x) { inline std::string to_string(const new_data_msg& x) {
std::ostringstream os; std::ostringstream os;
os << std::setfill('0') << std::hex << std::right; os << std::setfill('0') << std::hex << std::right;
...@@ -98,6 +99,45 @@ void serialize(T& in_out, new_data_msg& x, const unsigned int) { ...@@ -98,6 +99,45 @@ void serialize(T& in_out, new_data_msg& x, const unsigned int) {
in_out & x.buf; in_out & x.buf;
} }
/// Signalizes that a certain amount of bytes has been written.
struct data_transferred_msg {
/// Handle to the related connection.
connection_handle handle;
/// Number of transferred bytes.
uint64_t written;
/// Number of remaining bytes in all send buffers.
uint64_t remaining;
};
/// @relates data_transferred_msg
inline std::string to_string(const data_transferred_msg& x) {
return "data_transferred_msg"
+ deep_to_string(std::forward_as_tuple(x.handle, x.written,
x.remaining));
}
/// @relates data_transferred_msg
inline bool operator==(const data_transferred_msg& x,
const data_transferred_msg& y) {
return x.handle == y.handle
&& x.written == y.written
&& x.remaining == y.remaining;
}
/// @relates data_transferred_msg
inline bool operator!=(const data_transferred_msg& x,
const data_transferred_msg& y) {
return ! (x == y);
}
/// @relates new_data_msg
template <class T>
void serialize(T& in_out, data_transferred_msg& x, const unsigned int) {
in_out & x.handle;
in_out & x.written;
in_out & x.remaining;
}
/// Signalizes that a {@link broker} connection has been closed. /// Signalizes that a {@link broker} connection has been closed.
struct connection_closed_msg { struct connection_closed_msg {
/// Handle to the closed connection. /// Handle to the closed connection.
......
...@@ -99,6 +99,11 @@ void abstract_broker::configure_read(connection_handle hdl, ...@@ -99,6 +99,11 @@ void abstract_broker::configure_read(connection_handle hdl,
by_id(hdl).configure_read(cfg); by_id(hdl).configure_read(cfg);
} }
void abstract_broker::ack_writes(connection_handle hdl, bool enable) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(enable));
by_id(hdl).ack_writes(enable);
}
std::vector<char>& abstract_broker::wr_buf(connection_handle hdl) { std::vector<char>& abstract_broker::wr_buf(connection_handle hdl) {
return by_id(hdl).wr_buf(); return by_id(hdl).wr_buf();
} }
......
...@@ -779,6 +779,10 @@ connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self, ...@@ -779,6 +779,10 @@ connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self,
stream_.configure_read(config); stream_.configure_read(config);
if (! launched_) launch(); if (! launched_) launch();
} }
void ack_writes(bool enable) override {
CAF_LOG_TRACE(CAF_ARG(enable));
stream_.ack_writes(enable);
}
std::vector<char>& wr_buf() override { std::vector<char>& wr_buf() override {
return stream_.wr_buf(); return stream_.wr_buf();
} }
......
...@@ -62,6 +62,18 @@ void scribe::consume(execution_unit* ctx, const void*, size_t num_bytes) { ...@@ -62,6 +62,18 @@ void scribe::consume(execution_unit* ctx, const void*, size_t num_bytes) {
} }
} }
void scribe::data_transferred(execution_unit* ctx, size_t written,
size_t remaining) {
CAF_LOG_TRACE(CAF_ARG(written) << CAF_ARG(remaining));
if (detached())
return;
data_transferred_msg tmp{hdl(), written, remaining};
auto ptr = mailbox_element::make_joint(invalid_actor_addr,
invalid_message_id,
{}, tmp);
parent()->exec_single_event(ctx, ptr);
}
void scribe::io_failure(execution_unit* ctx, network::operation op) { void scribe::io_failure(execution_unit* ctx, network::operation op) {
CAF_LOG_TRACE(CAF_ARG(hdl()) << CAF_ARG(op)); CAF_LOG_TRACE(CAF_ARG(hdl()) << CAF_ARG(op));
// keep compiler happy when compiling w/o logging // keep compiler happy when compiling w/o logging
......
...@@ -61,6 +61,10 @@ void test_multiplexer::assign_tcp_scribe(abstract_broker* ptr, ...@@ -61,6 +61,10 @@ void test_multiplexer::assign_tcp_scribe(abstract_broker* ptr,
mpx_->read_config(hdl()) = config; mpx_->read_config(hdl()) = config;
} }
void ack_writes(bool enable) override {
mpx_->ack_writes(hdl()) = enable;
}
std::vector<char>& wr_buf() override { std::vector<char>& wr_buf() override {
return mpx_->output_buffer(hdl()); return mpx_->output_buffer(hdl());
} }
...@@ -223,6 +227,10 @@ receive_policy::config& test_multiplexer::read_config(connection_handle hdl) { ...@@ -223,6 +227,10 @@ receive_policy::config& test_multiplexer::read_config(connection_handle hdl) {
return scribe_data_[hdl].recv_conf; return scribe_data_[hdl].recv_conf;
} }
bool& test_multiplexer::ack_writes(connection_handle hdl) {
return scribe_data_[hdl].ack_writes;
}
bool& test_multiplexer::stopped_reading(connection_handle hdl) { bool& test_multiplexer::stopped_reading(connection_handle hdl) {
return scribe_data_[hdl].stopped_reading; return scribe_data_[hdl].stopped_reading;
} }
......
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