Commit f435999b authored by Joseph Noir's avatar Joseph Noir

Remove indirect entries from routing table

parent 0951aaa3
......@@ -167,12 +167,20 @@ public:
/// @param buf_size Size of the receiver buffer for the next datagram.
void configure_datagram_size(dgram_doorman_handle hdl, size_t buf_size);
/// Modifies the buffer for received datagrams.
/// @param hdl Identifies the affected socket.
/// @param buf_size Size of the receiver buffer for the next datagram.
void configure_datagram_size(dgram_scribe_handle hdl, size_t buf_size);
/// Returns write buffer for given sink.
std::vector<char>& wr_buf(dgram_scribe_handle hdl);
/// Writes `data` into the buffer of a given sink.
void write(dgram_scribe_handle hdl, size_t data_size, const void* data);
/// Sends the content of the buffer for given connection.
void flush(dgram_scribe_handle hdl);
/// Returns the middleman instance this broker belongs to.
inline middleman& parent() {
return system().middleman();
......@@ -279,6 +287,10 @@ public:
/// or `0` if `hdl` is invalid.
uint16_t remote_port(dgram_scribe_handle hdl);
/// Returns the remote port associated to `hdl`
/// or `0` if `hdl` is invalid.
uint16_t local_port(dgram_scribe_handle hdl);
/// Returns the local port associated to `hdl` or `0` if `hdl` is invalid.
uint16_t local_port(dgram_doorman_handle hdl);
......
......@@ -118,6 +118,13 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
void set_context(dgram_scribe_handle hdl);
// visitors to handle variants
struct purge_visitor {
using result_type = void;
purge_visitor(basp_broker_state* ptr) : state{ptr} { }
result_type operator()(const connection_handle& hdl);
result_type operator()(const dgram_scribe_handle& hdl);
basp_broker_state* state;
};
purge_visitor purge_state_vis;
wr_buf_visitor wr_buf_vis;
......
......@@ -62,7 +62,7 @@ public:
return f(meta::type_name("connection_handle"), x.id_);
}
private:
private:
inline connection_handle(int64_t handle_id) : super(handle_id) {
// nop
}
......
......@@ -56,6 +56,9 @@ public:
bool new_endpoint(execution_unit* ctx, dgram_scribe_handle endpoint);
bool delegate_msg(execution_unit* ctx, dgram_scribe_handle endpoint,
const void* buf, size_t num_bytes);
// needs to be launched explicitly
virtual void launch() = 0;
......
......@@ -60,6 +60,8 @@ public:
/// content of the buffer via the network.
virtual void flush() = 0;
virtual uint16_t local_port() const = 0;
void io_failure(execution_unit* ctx, network::operation op) override;
bool consume(execution_unit*, const void*, size_t) override;
......
......@@ -570,10 +570,12 @@ public:
/// Activates the stream.
void activate(manager_type* mgr);
/*
/// Configures how much data will be provided for the next `consume` callback.
/// @warning Must not be called outside the IO multiplexers event loop
/// once the stream has been started.
void configure_read(receive_policy::config config);
*/
void ack_writes(bool x);
......
......@@ -34,7 +34,7 @@ public:
~dgram_acceptor_manager();
virtual bool new_endpoint() = 0;
virtual bool new_endpoint(const void* buf, size_t num_bytes) = 0;
};
......
......@@ -233,6 +233,23 @@ inspect(Inspector& f, dgram_doorman_passivated_msg& x) {
return f(meta::type_name("dgram_doorman_passivated_msg"), x.handle);
}
/// delegates a handshake to a dgram scribe
struct dgram_delegate_msg {
// Sender handle
dgram_doorman_handle source;
// Responsible scribe
dgram_scribe_handle handle;
// Buffer containing received data.
std::vector<char> buf;
};
/// @relates dgram_delegate_msg
template <class Inspector>
typename Inspector::result_type
inspect(Inspector& f, dgram_delegate_msg& x) {
return f(meta::type_name("dgram_delegate_msg"), x.handle);
}
} // namespace io
} // namespace caf
......
......@@ -20,8 +20,6 @@
#ifndef CAF_IO_VISITORS_HPP
#define CAF_IO_VISITORS_HPP
#include "caf/io/fwd.hpp"
#include "caf/io/abstract_broker.hpp"
namespace caf {
......@@ -29,15 +27,15 @@ namespace io {
struct wr_buf_visitor {
using result_type = std::vector<char>&;
wr_buf_visitor(abstract_broker* ptr) : ptr{ptr} { /* nop */ }
wr_buf_visitor(abstract_broker* ptr) : ptr{ptr} { }
template <typename Handle>
result_type operator()(const Handle& hdl) { return ptr->wr_buf(hdl); }
abstract_broker* ptr;
};
struct flush_visitor {
using result_type = std::vector<char>&;
flush_visitor(abstract_broker* ptr) : ptr{ptr} { /* nop */ }
using result_type = void;
flush_visitor(abstract_broker* ptr) : ptr{ptr} { }
template <typename Handle>
result_type operator()(const Handle& hdl) { return ptr->flush(hdl); }
abstract_broker* ptr;
......@@ -45,7 +43,7 @@ struct flush_visitor {
struct addr_visitor {
using result_type = std::string;
addr_visitor(abstract_broker* ptr) : ptr{ptr} { /* nop */ }
addr_visitor(abstract_broker* ptr) : ptr{ptr} { }
template <typename Handle>
result_type operator()(const Handle& hdl) { return ptr->remote_addr(hdl); }
abstract_broker* ptr;
......@@ -53,18 +51,16 @@ struct addr_visitor {
struct port_visitor {
using result_type = uint16_t;
port_visitor(abstract_broker* ptr) : ptr{ptr} { /* nop */ }
port_visitor(abstract_broker* ptr) : ptr{ptr} { }
template <typename Handle>
result_type operator()(const Handle& hdl) { return ptr->remote_port(hdl); }
abstract_broker* ptr;
};
struct purge_visitor {
using result_type = void;
purge_visitor(basp_broker_state* ptr) : state{ptr} { /* nop */ }
result_type operator()(const connection_handle& hdl);
result_type operator()(const dgram_scribe_handle& hdl);
basp_broker_state* state;
struct id_visitor {
using result_type = int64_t;
template <typename Handle>
result_type operator()(const Handle& hdl) { return hdl.id(); }
};
} // namespace io
......
......@@ -125,6 +125,14 @@ void abstract_broker::configure_datagram_size(dgram_doorman_handle hdl,
x->configure_datagram_size(buf_size);
}
void abstract_broker::configure_datagram_size(dgram_scribe_handle hdl,
size_t buf_size) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(buf_size));
auto x = by_id(hdl);
if (x)
x->configure_datagram_size(buf_size);
}
std::vector<char>& abstract_broker::wr_buf(dgram_scribe_handle hdl) {
auto x = by_id(hdl);
if (!x) {
......@@ -143,6 +151,12 @@ void abstract_broker::write(dgram_scribe_handle hdl, size_t bs,
out.insert(out.end(), first, last);
}
void abstract_broker::flush(dgram_scribe_handle hdl) {
auto x = by_id(hdl);
if (x)
x->flush();
}
std::vector<connection_handle> abstract_broker::connections() const {
std::vector<connection_handle> result;
result.reserve(scribes_.size());
......@@ -281,6 +295,11 @@ uint16_t abstract_broker::remote_port(dgram_scribe_handle hdl) {
return i != dgram_scribes_.end() ? i->second->port() : 0;
}
uint16_t abstract_broker::local_port(dgram_scribe_handle hdl) {
auto i = dgram_scribes_.find(hdl);
return i != dgram_scribes_.end() ? i->second->local_port() : 0;
}
uint16_t abstract_broker::local_port(dgram_doorman_handle hdl) {
auto i = dgram_doormans_.find(hdl);
return i != dgram_doormans_.end() ? i->second->port() : 0;
......
......@@ -497,6 +497,33 @@ void basp_broker_state::set_context(dgram_scribe_handle hdl) {
this_context = &i->second;
}
basp_broker_state::purge_visitor::result_type
basp_broker_state::purge_visitor::operator()(const connection_handle& h) {
auto i = state->tcp_ctx.find(h);
if (i != state->tcp_ctx.end()) {
auto& ref = i->second;
if (ref.callback) {
CAF_LOG_DEBUG("connection closed during handshake");
ref.callback->deliver(sec::disconnect_during_handshake);
}
state->tcp_ctx.erase(i);
}
}
basp_broker_state::purge_visitor::result_type
basp_broker_state::purge_visitor::operator()(const dgram_scribe_handle& h) {
auto i = state->udp_ctx.find(h);
if (i != state->udp_ctx.end()) {
auto& ref = i->second;
if (ref.callback) {
CAF_LOG_DEBUG("connection closed during handshake");
ref.callback->deliver(sec::disconnect_during_handshake);
}
state->udp_ctx.erase(i);
}
}
/******************************************************************************
* basp_broker *
******************************************************************************/
......@@ -657,23 +684,35 @@ behavior basp_broker::make_behavior() {
// received from underlying broker implementation
[=](const new_endpoint_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
auto& bi = state.instance;
bi.write_server_handshake(context(), wr_buf(msg.handle),
local_port(msg.source));
std::cerr << "new endpoint msg received" << std::endl;
//auto& bi = state.instance;
//bi.write_server_handshake(context(), wr_buf(msg.handle),
// local_port(msg.source));
static_cast<void>(msg);
// flush(msg.handle);
// TODO: is this right?
},
[=](const dgram_delegate_msg&) {
std::cerr << "Received new delegate message" << std::endl;
},
// received from underlying broker implementation
[=](const dgram_scribe_closed_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
// TODO: dgram_scribe_closed_msg
static_cast<void>(msg);
// TODO: currently we assume a node has gone offline once we lose
// a connection, we also could try to reach this node via other
// hops to be resilient to (rare) network failures or if a
// node is reachable via several interfaces and only one fails
auto nid = state.instance.tbl().lookup_node(msg.handle);
// tell BASP instance we've lost connection
state.instance.handle_node_shutdown(nid);
CAF_ASSERT(nid == none
|| !state.instance.tbl().reachable(nid));
},
// received from underlying broker implementation
[=](const dgram_acceptor_closed_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
// TODO: dgram_acceptor_closed_msg
static_cast<void>(msg);
auto port = local_port(msg.handle);
state.instance.remove_published_actor(port);
},
// received from middleman actor
[=](publish_atom, accept_handle hdl, uint16_t port,
......@@ -731,6 +770,7 @@ behavior basp_broker::make_behavior() {
[=](connect_atom, dgram_scribe_handle hdl,
const std::string& host, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(hdl.id()));
std::cerr << "connect for dgram to " << host << ":" << port << std::endl;
auto rp = make_response_promise();
auto res = assign_dgram_scribe(hdl, host, port);
if (res) {
......@@ -738,8 +778,11 @@ behavior basp_broker::make_behavior() {
ctx.hdl = hdl;
ctx.remote_port = port;
ctx.callback = rp;
// TODO: Start handshake with server as there is no way for
// the server to initiate this.
auto& bi = state.instance;
bi.write_server_handshake(context(), wr_buf(hdl), local_port(hdl));
//bi.write_client_handshake(context(), wr_buf(hdl), local_port(hdl));
flush(hdl);
configure_datagram_size(hdl, 1500);
} else {
CAF_LOG_DEBUG("failed to assign datagram sink from handle"
<< CAF_ARG(res));
......
......@@ -914,6 +914,9 @@ default_multiplexer::add_dgram_scribe(abstract_broker* self, native_socket fd,
void configure_datagram_size(size_t buf_size) override {
CAF_LOG_TRACE("");
communicator_.configure_datagram_size(buf_size);
// TODO: Do we need this?
if (!launched_)
launch();
}
void ack_writes(bool enable) override {
communicator_.ack_writes(enable);
......@@ -939,6 +942,12 @@ default_multiplexer::add_dgram_scribe(abstract_broker* self, native_socket fd,
return "";
return *x;
}
uint16_t local_port() const override {
auto x = local_port_of_fd(communicator_.fd());
if (!x)
return 0;
return *x;
}
uint16_t port() const override {
auto x = remote_port_of_fd(communicator_.fd());
if (!x)
......@@ -988,7 +997,7 @@ default_multiplexer::add_dgram_doorman(abstract_broker* self,
CAF_LOG_TRACE("");
acceptor_.configure_datagram_size(buf_size);
}
bool new_endpoint() override {
bool new_endpoint(const void* buf, size_t num_bytes) override {
CAF_LOG_TRACE("");
// TODO: this currently ignores the payload of the datagram
if (detached())
......@@ -1009,23 +1018,14 @@ default_multiplexer::add_dgram_doorman(abstract_broker* self,
auto hdl = dm.add_dgram_scribe(parent(), *fd,
endpoint_info.first, endpoint_info.second,
false);
return dgram_doorman::new_endpoint(&dm, hdl);
/*
auto hdl = dm.add_dgram_scribe(parent(), acceptor_.host(),
acceptor_.port());
if (hdl)
return dgram_doorman::new_endpoint(&dm, *hdl);
*/
/*
auto hdl = dm.add_tcp_scribe(parent(),
std::move(acceptor_.accepted_socket()));
return doorman::new_connection(&dm, hdl);
*/
/*
CAF_LOG_DEBUG("Creating an new dgram scribe failed for:"
<< CAF_ARG(acceptor_.host()) << CAF_ARG(acceptor_.port()));
return false;
*/
// TODO: needs a better design
// new_endpoint(...) registeres the new handle
// delegate_message assigns the new handle
// responsibility for the received msg.
auto res = dgram_doorman::new_endpoint(&dm, hdl);
if (!res)
return false;
return delegate_msg(&dm, hdl, buf, num_bytes);
}
void stop_reading() override {
CAF_LOG_TRACE("");
......@@ -1250,6 +1250,7 @@ bool try_accept(native_socket& result, native_socket fd) {
bool send_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len,
sockaddr_storage& sa, size_t sa_len) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(len));
std::cerr << "Sending datagram" << std::endl;
auto sres = ::sendto(fd, reinterpret_cast<socket_send_ptr>(buf), buf_len,
no_sigpipe_flag, reinterpret_cast<sockaddr*>(&sa),
sa_len);
......@@ -1265,6 +1266,7 @@ bool send_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len,
bool receive_datagram(size_t& result, native_socket fd, void* buf, size_t len,
sockaddr_storage& sender_addr, socklen_t& sender_len) {
std::cerr << "Receiving datagram" << std::endl;
sender_len = sizeof(sender_addr);
CAF_LOG_TRACE(CAF_ARG(fd));
auto sres = ::recvfrom(fd, buf, len, no_sigpipe_flag,
......@@ -1797,25 +1799,11 @@ void dgram_acceptor::handle_event(operation op) {
sender_from_sockaddr(sockaddr_, sockaddr_len_);
if (rb == 0)
return;
mgr_->new_endpoint();
/*
native_socket sockfd = invalid_native_socket;
if (try_accept(sockfd, fd())) {
if (sockfd != invalid_native_socket) {
sock_ = sockfd;
mgr_->new_endpoint();
}
}
*/
/*
auto res = reader_->consume(&backend(), rd_buf_.data(), rb);
packet_size_ = rb;
prepare_next_read();
if (!res) {
passivate();
return;
auto res = mgr_->new_endpoint(rd_buf_.data(), rb);
if (res) {
// What is the right way to propagate this?
CAF_LOG_DEBUG("Failure during creation of new udp endpoint");
}
*/
break;
}
case operation::write: {
......
......@@ -52,5 +52,26 @@ bool dgram_doorman::new_endpoint(execution_unit* ctx,
return invoke_mailbox_element(ctx);
}
bool dgram_doorman::delegate_msg(execution_unit* ctx,
dgram_scribe_handle endpoint,
const void*, size_t num_bytes) {
CAF_LOG_TRACE(CAF_ARG(endpoint) << CAF_ARG(num_bytes));
if (detached())
return false;
using delegate_t = dgram_delegate_msg;
using tmp_t = mailbox_element_vals<delegate_t>;
auto guard = parent_;
auto& buf = rd_buf();
CAF_ASSERT(buf.size() >= num_bytes);
buf.resize(num_bytes);
// std::vector<char> msg_buf;
// msg_buf.resize(num_bytes);
tmp_t tmp{strong_actor_ptr{}, message_id::make(),
mailbox_element::forwarding_stack{},
delegate_t{hdl(), endpoint, std::move(buf)}};
invoke_mailbox_element_impl(ctx,tmp);
return true;
}
} // namespace io
} // namespace caf
......@@ -261,6 +261,7 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
static_cast<void>(ctx);
static_cast<void>(dm);
static_cast<void>(hdr);
std::cerr << "New datagram received!" << std::endl;
return false;
};
......@@ -382,6 +383,7 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
void instance::write(execution_unit* ctx, buffer_type& buf,
header& hdr, payload_writer* pw) {
CAF_LOG_TRACE(CAF_ARG(hdr));
std::cerr << "In instance::write" << std::endl;
error err;
if (pw) {
auto pos = buf.size();
......@@ -399,8 +401,10 @@ void instance::write(execution_unit* ctx, buffer_type& buf,
binary_serializer bs{ctx, buf};
err = bs(hdr);
}
if (err)
if (err) {
CAF_LOG_ERROR(CAF_ARG(err));
std::cerr << "With error: " << to_string(err) << std::endl;
}
}
void instance::write_server_handshake(execution_unit* ctx,
......
......@@ -62,12 +62,20 @@ void routing_table::flush(const endpoint& r) {
}
node_id routing_table::lookup_node(const endpoint_handle& hdl) const {
return get_opt(direct_by_hdl_, hdl, none);
// return get_opt(direct_by_hdl_, hdl, none);
auto i = direct_by_hdl_.find(hdl);
if (i != direct_by_hdl_.end())
return i->second;
return none;
}
optional<routing_table::endpoint_handle>
routing_table::lookup_hdl(const node_id& nid) const {
return get_opt(direct_by_nid_, nid, none);
// return get_opt(direct_by_nid_, nid, none);
auto i = direct_by_nid_.find(nid);
if (i != direct_by_nid_.end())
return i->second;
return none;
}
/*
......@@ -159,7 +167,7 @@ size_t routing_table::erase(const node_id& dest, erase_callback& cb) {
*/
auto hdl = lookup_hdl(dest);
if (hdl) {
direct_by_hdl_.erase(hdl);
direct_by_hdl_.erase(*hdl);
direct_by_nid_.erase(dest);
parent_->parent().notify<hook::connection_lost>(dest);
++res;
......
......@@ -269,6 +269,14 @@ expected<void> test_multiplexer::assign_dgram_scribe(abstract_broker* ptr,
return "test";
}
uint16_t local_port() const override {
std::cerr << "test_multiplexer::assign_dgram_scribe::impl::local_port "
<< "not implementd." << std::endl;
abort();
return 0;
}
uint16_t port() const override {
return mpx_->port(hdl());
}
......@@ -345,7 +353,7 @@ expected<void> test_multiplexer::assign_dgram_doorman(abstract_broker* ptr,
// nop
}
bool new_endpoint() override {
bool new_endpoint(const void*, size_t) override {
// TODO: Implement me
/*
auto& mm = mpx_->pending_connects();
......@@ -376,7 +384,7 @@ expected<void> test_multiplexer::assign_dgram_doorman(abstract_broker* ptr,
std::string addr() const override {
return "test";
}
uint16_t port() const override {
return mpx_->port(hdl());
}
......
......@@ -22,30 +22,7 @@
namespace caf {
namespace io {
purge_visitor::result_type purge_visitor::operator()(const connection_handle& h) {
auto i = state->tcp_ctx.find(h);
if (i != state->tcp_ctx.end()) {
auto& ref = i->second;
if (ref.callback) {
CAF_LOG_DEBUG("connection closed during handshake");
ref.callback->deliver(sec::disconnect_during_handshake);
}
state->tcp_ctx.erase(i);
}
}
purge_visitor::result_type
purge_visitor::operator()(const dgram_scribe_handle& h) {
auto i = state->udp_ctx.find(h);
if (i != state->udp_ctx.end()) {
auto& ref = i->second;
if (ref.callback) {
CAF_LOG_DEBUG("connection closed during handshake");
ref.callback->deliver(sec::disconnect_during_handshake);
}
state->udp_ctx.erase(i);
}
}
} // namespace io
} // namespace caf
......
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