Commit a4ad9d7b authored by Joseph Noir's avatar Joseph Noir

Add packet queue to datagram communicator

parent 0daae04f
......@@ -285,7 +285,7 @@ public:
return false;
}
if (tcp_based)
write_client_handshake(ctx, path->wr_buf, hdr.source_node);
write_client_handshake(ctx, apply_visitor(wr_buf_, path->hdl), hdr.source_node);
callee_.learned_new_node_directly(hdr.source_node);
callee_.finalize_handshake(hdr.source_node, aid, sigs);
flush(*path);
......
......@@ -81,7 +81,6 @@ public:
/// Describes a routing path to a node.
struct endpoint {
buffer_type& wr_buf;
const node_id& next_hop;
endpoint_handle hdl;
};
......
......@@ -174,6 +174,7 @@ private:
// visitors
addr_visitor addr_;
port_visitor port_;
wr_buf_visitor wr_buf_;
};
} // namespace io
......
......@@ -603,7 +603,8 @@ public:
/// @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_;
wr_offline_buf_.emplace_back();
return wr_offline_buf_.back();
}
/// Returns the read buffer of this stream.
......@@ -653,8 +654,8 @@ private:
manager_ptr writer_;
bool ack_writes_;
bool writing_;
buffer_type wr_buf_;
buffer_type wr_offline_buf_;
std::vector<char> wr_buf_;
std::deque<std::vector<char>> wr_offline_buf_;
// general state
// signifies whether to adapt the sender of the next received messages
......
......@@ -173,8 +173,9 @@ void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) {
<< CAF_ARG(nid));
return;
}
instance.write_kill_proxy(self->context(), path->wr_buf,
nid, aid, rsn);
instance.write_kill_proxy(self->context(),
apply_visitor(wr_buf_vis, path->hdl),
nid, aid, rsn);
instance.tbl().flush(*path);
};
auto ptr = actor_cast<strong_actor_ptr>(entry);
......@@ -355,7 +356,8 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
0, 0, this_node(), nid, tmp.id(), invalid_actor_id};
// writing std::numeric_limits<actor_id>::max() is a hack to get
// this send-to-named-actor feature working with older CAF releases
instance.write(self->context(), path->wr_buf, hdr, &writer);
instance.write(self->context(), apply_visitor(wr_buf_vis, path->hdl), hdr,
&writer);
instance.flush(*path);
}
......@@ -521,7 +523,8 @@ basp_broker_state::purge_visitor::operator()(const dgram_scribe_handle& h) {
basp_broker::basp_broker(actor_config& cfg)
: stateful_actor<basp_broker_state, broker>(cfg),
addr_(this),
port_(this) {
port_(this),
wr_buf_(this) {
// nop
}
......@@ -638,7 +641,8 @@ behavior basp_broker::make_behavior() {
basp::header::named_receiver_flag,
0, cme->mid.integer_value(), state.this_node(),
dest_node, src->id(), invalid_actor_id};
state.instance.write(context(), path->wr_buf, hdr, &writer);
state.instance.write(context(), apply_visitor(wr_buf_, path->hdl), hdr,
&writer);
state.instance.flush(*path);
return delegated<message>();
},
......
......@@ -1641,7 +1641,7 @@ void dgram_communicator::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);
wr_offline_buf_.insert(wr_offline_buf_.end(), std::vector<char>{first, last});
}
void dgram_communicator::flush(const manager_ptr& mgr) {
......@@ -1674,7 +1674,6 @@ void dgram_communicator::handle_event(operation op) {
switch (op) {
case operation::read: {
// incoming message should signify a new remote enpoint
// --> create a new dgram_scribe
size_t rb;
if (!receive_datagram(rb, fd(), rd_buf_.data(), rd_buf_.size(),
sockaddr_, sockaddr_len_)) {
......@@ -1682,13 +1681,6 @@ void dgram_communicator::handle_event(operation op) {
passivate();
return;
}
// {
// std::string host;
// uint16_t port;
// std::tie(host, port) = sender_from_sockaddr(sockaddr_, sockaddr_len_);
// std::cerr << "[COM] " << fd() << " received " << rb << " bytes from "
// << host << ":" << port << std::endl;
// }
// currently handles the change from acceptor
// to communicator, TODO: Find a better solution
// Either keep sending to the same endpoint,
......@@ -1699,8 +1691,6 @@ void dgram_communicator::handle_event(operation op) {
remote_endpoint_addr_ = std::move(sockaddr_);
remote_endpoint_addr_len_ = sockaddr_len_;
sockaddr_len_ = 0;
// std::cerr << "[COM] Adapted new endpoint: " << host_ << ":" << port_
// << std::endl;
}
if (rb > 0) {
auto res = reader_->consume(&backend(), rd_buf_.data(), rb);
......@@ -1720,17 +1710,11 @@ void dgram_communicator::handle_event(operation op) {
writer_->io_failure(&backend(), operation::write);
backend().del(operation::write, fd(), this);
} else if (wb > 0) {
// std::cerr << "[COM] " << fd() << " sent " << wb << " bytes to "
// << host_ << ":" << port_ << std::endl;
CAF_ASSERT(wb == wr_buf_.size());
if (ack_writes_)
writer_->datagram_sent(&backend(), wb);
prepare_next_write();
} else {
// TODO: remove this if sure that datagrams are either written
// as a whole or not at all
// std::cerr << "[DC] Partial datagram wrtten: " << wb
// << " of " << wr_buf_.size() << std::endl;
if (writer_)
writer_->io_failure(&backend(), operation::write);
}
......@@ -1748,11 +1732,18 @@ void dgram_communicator::handle_event(operation op) {
void dgram_communicator::prepare_next_write() {
CAF_LOG_TRACE(CAF_ARG(wr_buf_.size()) << CAF_ARG(wr_offline_buf_.size()));
wr_buf_.clear();
/*
while (wr_offline_buf_.size() > 0 && wr_offline_buf_.front().size() == 0) {
std::cerr << "Removing empy element from buffer ... " << std::endl;
wr_offline_buf_.pop_front();
}
*/
if (wr_offline_buf_.empty()) {
writing_ = false;
backend().del(operation::write, fd(), this);
} else {
wr_buf_.swap(wr_offline_buf_);
wr_buf_.swap(wr_offline_buf_.front());
wr_offline_buf_.pop_front();
}
}
......
......@@ -58,7 +58,7 @@ bool dgram_scribe::consume(execution_unit* ctx, const void*, size_t num_bytes) {
auto result = invoke_mailbox_element(ctx);
// swap buffer back to stream and implicitly flush wr_buf()
msg_buf.swap(buf);
// flush(); // <-- TODO: here from scribe, not sure why?
flush(); // <-- TODO: here from scribe, not sure why?
return result;
}
......
......@@ -88,7 +88,7 @@ connection_state instance::handle(execution_unit* ctx,
CAF_LOG_DEBUG("forward message");
auto path = lookup(hdr.dest_node);
if (path) {
binary_serializer bs{ctx, path->wr_buf};
binary_serializer bs{ctx, apply_visitor(wr_buf_, path->hdl)};
auto e = bs(hdr);
if (e)
return err();
......@@ -242,7 +242,7 @@ void instance::write(execution_unit* ctx, const routing_table::endpoint& r,
header& hdr, payload_writer* writer) {
CAF_LOG_TRACE(CAF_ARG(hdr));
CAF_ASSERT(hdr.payload_len == 0 || writer != nullptr);
write(ctx, r.wr_buf, hdr, writer);
write(ctx, apply_visitor(wr_buf_, r.hdl), hdr, writer);
tbl_.flush(r);
}
......@@ -317,7 +317,7 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
header hdr{message_type::dispatch_message, 0, 0, mid.integer_value(),
sender ? sender->node() : this_node(), receiver->node(),
sender ? sender->id() : invalid_actor_id, receiver->id()};
write(ctx, path->wr_buf, hdr, &writer);
write(ctx, apply_visitor(wr_buf_, path->hdl), hdr, &writer);
flush(*path);
notify<hook::message_sent>(sender, path->next_hop, receiver, mid, msg);
return true;
......
......@@ -39,7 +39,7 @@ routing_table::~routing_table() {
optional<routing_table::endpoint> routing_table::lookup(const node_id& target) {
auto hdl = lookup_hdl(target);
if (hdl)
return endpoint{apply_visitor(wr_buf_, *hdl), target, *hdl};
return endpoint{/*apply_visitor(wr_buf_, *hdl),*/ target, *hdl};
// pick first available indirect route
/*
auto i = indirect_.find(target);
......
......@@ -41,7 +41,6 @@ behavior client(broker* self, connection_handle hdl) {
buf.resize(200);
std::iota(buf.begin(), buf.end(), 0);
self->write(hdl, buf.size(), buf.data());
CAF_REQUIRE_EQUAL(self->wr_buf(hdl).size(), 200u);
self->configure_read(hdl, receive_policy::at_least(1));
self->flush(hdl);
return {
......
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