Commit 0696b6aa authored by Joseph Noir's avatar Joseph Noir

Add dgram_{scribe,doorman} impl in def multiplexer

parent 211c30f4
......@@ -54,7 +54,7 @@ public:
using dgram_doorman_base::new_endpoint;
bool new_endpoint(execution_unit* ctx, const void*, size_t num_bytes);
bool new_endpoint(execution_unit* ctx, dgram_scribe_handle endpoint);
// needs to be launched explicitly
virtual void launch() = 0;
......
......@@ -389,6 +389,9 @@ private:
void wr_dispatch_request(resumable* ptr);
// allows initialization of scribes with an exisiting sockaddr
dgram_scribe_handle add_dgram_scribe(abstract_broker* ptr, native_socket fd,
sockaddr_storage* addr, size_t len);
//resumable* rd_dispatch_request();
native_socket epollfd_; // unused in poll() implementation
......@@ -607,6 +610,8 @@ public:
void sender_from_sockaddr(sockaddr_storage& sa, size_t len);
void set_endpoint_addr(sockaddr_storage& sa, size_t len);
private:
void prepare_next_read();
......@@ -616,6 +621,7 @@ private:
manager_ptr reader_;
size_t dgram_size_;
buffer_type rd_buf_;
size_t bytes_read_;
// state for writing
manager_ptr writer_;
......@@ -626,7 +632,7 @@ private:
buffer_type wr_offline_buf_;
// general state
struct sockaddr_storage remote_endpoint_;
struct sockaddr_storage sockaddr_;
socklen_t sockaddr_len_;
std::string host_;
uint16_t port_;
......@@ -683,11 +689,13 @@ public:
void sender_from_sockaddr(sockaddr_storage& sa, size_t len);
std::pair<sockaddr_storage*,size_t> last_sender();
private:
void prepare_next_read();
// reader state
manager_ptr reader_;
manager_ptr mgr_;
size_t dgram_size_;
buffer_type rd_buf_;
......@@ -696,7 +704,7 @@ private:
socklen_t sockaddr_len_;
std::string host_;
uint16_t port_;
native_socket sock_; // TODO: Do I need this?
// native_socket sock_; // TODO: Do I need this?
};
expected<native_socket> new_tcp_connection(const std::string& host,
......
......@@ -111,6 +111,9 @@ public:
/// Returns the output buffer of the datagram source identified by `hdl`.
buffer_type& output_buffer(dgram_scribe_handle hdl);
/// Returns the input buffer of the scribe identified by `hdl`.
buffer_type& input_buffer(dgram_scribe_handle hdl);
/// Returns the input buffer of the scribe identified by `hdl`.
buffer_type& input_buffer(connection_handle hdl);
......
......@@ -156,28 +156,30 @@ inspect(Inspector& f, dgram_doorman_closed_msg& x) {
struct new_endpoint_msg {
/// Handle to the related datagram endpoint.
dgram_doorman_handle handle;
/// Buffer containing the received data.
std::vector<char> buf;
// Buffer containing the received data.
// std::vector<char> buf;
/// Handle to new dgram_scribe
dgram_scribe_handle endpoint;
};
/// @relates new_endpoint_msg
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, new_endpoint_msg& x) {
return f(meta::type_name("new_endpoint_msg"), x.handle, x.buf);
return f(meta::type_name("new_endpoint_msg"), x.handle, x.endpoint);
}
/// Signalizes that a datagram with a certain size has been sent.
struct new_datagram_msg {
// Handle to the endpoint used.
dgram_scribe_handle handle;
// number of bytes written.
uint64_t written;
// Buffer containing received data.
std::vector<char> buf;
};
/// @relates new_datagram_msg
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, new_datagram_msg& x) {
return f(meta::type_name("new_datagram_msg"), x.handle, x.written);
return f(meta::type_name("new_datagram_msg"), x.handle, x.buf);
}
/// Signalizes that a datagram with a certain size has been sent.
......
......@@ -860,6 +860,12 @@ accept_handle default_multiplexer::add_tcp_doorman(abstract_broker* self,
dgram_scribe_handle
default_multiplexer::add_dgram_scribe(abstract_broker* self,
native_socket fd) {
return add_dgram_scribe(self, fd, nullptr, 0);
}
dgram_scribe_handle
default_multiplexer::add_dgram_scribe(abstract_broker* self, native_socket fd,
sockaddr_storage* addr, size_t len) {
CAF_LOG_TRACE(CAF_ARG(fd));
CAF_ASSERT(fd != network::invalid_native_socket);
class impl : public dgram_scribe {
......@@ -915,11 +921,16 @@ default_multiplexer::add_dgram_scribe(abstract_broker* self,
void remove_from_loop() override {
communicator_.passivate();
}
void set_endpoint_addr(sockaddr_storage* addr, size_t len) {
communicator_.set_endpoint_addr(*addr, len);
}
private:
bool launched_;
network::dgram_communicator communicator_;
};
auto ptr = make_counted<impl>(self, *this, fd);
if (addr)
ptr->set_endpoint_addr(addr, len);
self->add_dgram_scribe(ptr);
return ptr->hdl();
}
......@@ -942,6 +953,7 @@ default_multiplexer::add_dgram_doorman(abstract_broker* self,
}
bool new_endpoint() override {
CAF_LOG_TRACE("");
// TODO: this currently ignores the payload of the datagram
if (detached())
// we are already disconnected from the broker while the multiplexer
// did not yet remove the socket, this can happen if an I/O event
......@@ -949,14 +961,34 @@ default_multiplexer::add_dgram_doorman(abstract_broker* self,
// further activities for the broker
return false;
auto& dm = acceptor_.backend();
static_cast<void>(dm);
auto fd = new_dgram_scribe_impl(acceptor_.host(),
acceptor_.port());
if (!fd) {
CAF_LOG_ERROR(CAF_ARG(fd.error()));
return false;
// return std::move(fd.error());
}
auto endpoint_info = acceptor_.last_sender();
auto hdl = dm.add_dgram_scribe(parent(), *fd,
endpoint_info.first,
endpoint_info.second);
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);
*/
// TODO: Implement me!
/*
CAF_LOG_DEBUG("Creating an new dgram scribe failed for:"
<< CAF_ARG(acceptor_.host()) << CAF_ARG(acceptor_.port()));
return false;
*/
}
void stop_reading() override {
CAF_LOG_TRACE("");
......@@ -1497,6 +1529,22 @@ dgram_communicator::dgram_communicator(default_multiplexer& backend_ref,
configure_datagram_size(1500);
}
void dgram_communicator::start(manager_type* mgr) {
CAF_ASSERT(mgr != nullptr);
activate(mgr);
}
void dgram_communicator::activate(manager_type* mgr) {
if (!writer_) {
writer_.reset(mgr);
event_handler::activate();
}
}
void dgram_communicator::configure_datagram_size(size_t size) {
dgram_size_ = size;
}
void dgram_communicator::ack_writes(bool x) {
ack_writes_ = x;
}
......@@ -1508,10 +1556,6 @@ void dgram_communicator::write(const void* buf, size_t num_bytes) {
wr_offline_buf_.insert(wr_offline_buf_.end(), first, last);
}
void dgram_communicator::configure_datagram_size(size_t size) {
dgram_size_ = size;
}
void dgram_communicator::flush(const manager_ptr& mgr) {
CAF_ASSERT(mgr != nullptr);
CAF_LOG_TRACE(CAF_ARG(wr_offline_buf_.size()));
......@@ -1522,18 +1566,6 @@ void dgram_communicator::flush(const manager_ptr& mgr) {
}
}
void dgram_communicator::start(manager_type* mgr) {
CAF_ASSERT(mgr != nullptr);
activate(mgr);
}
void dgram_communicator::activate(manager_type* mgr) {
if (!writer_) {
writer_.reset(mgr);
event_handler::activate();
}
}
void dgram_communicator::stop_reading() {
CAF_LOG_TRACE("");
close_read_channel();
......@@ -1541,16 +1573,39 @@ void dgram_communicator::stop_reading() {
}
void dgram_communicator::removed_from_loop(operation op) {
CAF_LOG_TRACE(CAF_ARG(fd()) << CAF_ARG(op));
if (op == operation::write)
writer_.reset();
switch (op) {
case operation::read: reader_.reset(); break;
case operation::write: writer_.reset(); break;
case operation::propagate_error: break;
};
}
void dgram_communicator::handle_event(operation op) {
CAF_LOG_TRACE(CAF_ARG(op));
switch (op) {
case operation::read: {
// This should not happen ...
// 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_)) {
reader_->io_failure(&backend(), operation::read);
passivate();
return;
}
if (host_.empty()) {
sender_from_sockaddr(sockaddr_, sockaddr_len_);
// TODO: connect to endpoint?
}
if (rb == 0)
return;
auto res = reader_->consume(&backend(), rd_buf_.data(), rb);
bytes_read_ = rb;
prepare_next_read();
if (!res) {
passivate();
return;
}
break;
}
case operation::write: {
......@@ -1592,6 +1647,11 @@ void dgram_communicator::prepare_next_write() {
}
}
void dgram_communicator::prepare_next_read() {
CAF_LOG_TRACE(CAF_ARG(wr_buf_.size()) << CAF_ARG(wr_offline_buf_.size()));
rd_buf_.resize(dgram_size_);
}
void dgram_communicator::sender_from_sockaddr(sockaddr_storage& sa, size_t) {
char addr[INET6_ADDRSTRLEN];
switch(sa.ss_family) {
......@@ -1617,6 +1677,11 @@ void dgram_communicator::sender_from_sockaddr(sockaddr_storage& sa, size_t) {
}
}
void dgram_communicator::set_endpoint_addr(sockaddr_storage& sa, size_t len) {
sockaddr_ = sa;
sockaddr_len_ = len;
}
dgram_acceptor::dgram_acceptor(default_multiplexer& backend_ref,
native_socket sockfd)
: event_handler(backend_ref, sockfd),
......@@ -1637,8 +1702,8 @@ void dgram_acceptor::start(manager_type* mgr) {
}
void dgram_acceptor::activate(manager_type* mgr) {
if (!reader_) {
reader_.reset(mgr);
if (!mgr_) {
mgr_.reset(mgr);
event_handler::activate();
}
}
......@@ -1651,24 +1716,35 @@ void dgram_acceptor::stop_reading() {
void dgram_acceptor::removed_from_loop(operation op) {
if (op == operation::read)
reader_.reset();
mgr_.reset();
}
void dgram_acceptor::handle_event(operation op) {
CAF_LOG_TRACE(CAF_ARG(op));
CAF_LOG_TRACE(CAF_ARG(fd()) << CAF_ARG(op));
switch (op) {
case operation::read: {
// Read next datagram
// 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_)) {
reader_->io_failure(&backend(), operation::read);
mgr_->io_failure(&backend(), operation::read);
passivate();
return;
}
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;
......@@ -1685,8 +1761,8 @@ void dgram_acceptor::handle_event(operation op) {
break;
}
case operation::propagate_error:
if (reader_)
reader_->io_failure(&backend(), operation::read);
if (mgr_)
mgr_->io_failure(&backend(), operation::read);
// backend will delete this handler anyway,
// no need to call backend().del() here
break;
......@@ -1718,8 +1794,12 @@ void dgram_acceptor::sender_from_sockaddr(sockaddr_storage& sa, size_t) {
}
}
std::pair<sockaddr_storage*,size_t> dgram_acceptor::last_sender() {
return std::make_pair(&sockaddr_, sockaddr_len_);
}
void dgram_acceptor::prepare_next_read() {
rd_buf_.resize(buf_size_);
rd_buf_.resize(dgram_size_);
}
class socket_guard {
......@@ -1925,7 +2005,7 @@ expected<native_socket> new_dgram_scribe_impl(const std::string& host,
uint16_t port,
optional<protocol> preferred) {
CAF_LOG_TRACE(CAF_ARG(host) << CAF_ARG(port) << CAF_ARG(preferred));
CAF_LOG_INFO("try to establish datagram sink to:" << CAF_ARG(host)
CAF_LOG_INFO("try to create dgram scribe for:" << CAF_ARG(host)
<< CAF_ARG(port));
auto res = interfaces::native_address(host, preferred);
if (!res) {
......@@ -1936,6 +2016,7 @@ expected<native_socket> new_dgram_scribe_impl(const std::string& host,
CAF_ASSERT(proto == ipv4 || proto == ipv6);
CALL_CFUN(fd, cc_valid_socket, "socket",
socket(proto == ipv4 ? AF_INET : AF_INET6, SOCK_DGRAM, 0));
/*
socket_guard sguard(fd);
if (proto == ipv6) {
if (ip_connect<AF_INET6>(fd, res->first, port)) {
......@@ -1953,6 +2034,8 @@ expected<native_socket> new_dgram_scribe_impl(const std::string& host,
}
CAF_LOG_INFO("successfully connected to host via IPv4");
return sguard.release();
*/
return fd;
}
expected<std::pair<native_socket, uint16_t>>
......@@ -1982,9 +2065,6 @@ new_dgram_doorman_impl(uint16_t port, const char* addr, bool reuse_addr) {
: new_ip_acceptor_impl<AF_INET6>(fd, port, addr);
if (!p)
return std::move(p.error());
// Not needed: CALL_CFUN(tmp2, cc_zero, "listen", listen(fd, SOMAXCONN));
// Should there be some handshake between nodes?
// TODO: Remove comments
// ok, no errors so far
CAF_LOG_DEBUG(CAF_ARG(fd) << CAF_ARG(p));
return std::make_pair(sguard.release(), *p);
......
......@@ -24,8 +24,7 @@
namespace caf {
namespace io {
dgram_doorman::dgram_doorman(abstract_broker* parent,
dgram_doorman_handle hdl)
dgram_doorman::dgram_doorman(abstract_broker* parent, dgram_doorman_handle hdl)
: dgram_doorman_base(parent, hdl) {
// nop
}
......@@ -45,10 +44,12 @@ void dgram_doorman::io_failure(execution_unit* ctx, network::operation op) {
detach(ctx, true);
}
bool dgram_doorman::new_endpoint(execution_unit* ctx, const void* buf,
size_t besize) {
// TODO: implement me!
return false;
// bool dgram_doorman::new_endpoint(execution_unit* ctx, const void* buf,
// size_t besize) {
bool dgram_doorman::new_endpoint(execution_unit* ctx,
dgram_scribe_handle endpoint) {
msg().endpoint = endpoint;
return invoke_mailbox_element(ctx);
}
} // namespace io
......
......@@ -37,8 +37,7 @@ message dgram_scribe::detach_message() {
return make_message(dgram_scribe_closed_msg{hdl()});
}
bool dgram_scribe::consume(execution_unit* ctx, const void*,
size_t num_bytes) {
bool dgram_scribe::consume(execution_unit* ctx, const void*, size_t num_bytes) {
CAF_ASSERT(ctx != nullptr);
CAF_LOG_TRACE(CAF_ARG(num_bytes));
if (detached())
......
......@@ -165,8 +165,7 @@ public:
} else if (std::distance(u.scheme().first, u.scheme().second) >= 3 &&
equal(std::begin(udp), std::end(udp), u.scheme().first)) {
// connect to endpoint and initiate handhsake etc. (UDP)
auto y
= system().middleman().backend().new_remote_endpoint(host, port);
auto y = system().middleman().backend().new_dgram_scribe(host, port);
if (!y) {
rp.deliver(std::move(y.error()));
return {};
......@@ -257,11 +256,11 @@ private:
equal(u.scheme().first, u.scheme().second, std::begin(udp))) {
// UDP
auto res
= system().middleman().backend().new_local_endpoint(u.port_as_int(),
= system().middleman().backend().new_dgram_doorman(u.port_as_int(),
in, reuse_addr);
if (!res)
return std::move(res.error());
endpoint_handle hdl = res->first;
dgram_doorman_handle hdl = res->first;
actual_port = res->second;
anon_send(broker_, publish_atom::value, hdl, actual_port,
std::move(whom), std::move(sigs));
......
......@@ -236,34 +236,52 @@ expected<void> test_multiplexer::assign_dgram_scribe(abstract_broker* ptr,
mpx_(mpx) {
// nop
}
void configure_datagram_size(size_t buf_size) override {
mpx_->buffer_size(hdl()) = buf_size;
}
void stop_reading() override {
mpx_->stopped_reading(hdl()) = true;
detach(mpx_, false);
}
void ack_writes(bool enable) override {
mpx_->ack_writes(hdl()) = enable;
}
std::vector<char>& rd_buf() override {
return mpx_->input_buffer(hdl());
}
std::vector<char>& wr_buf() override {
return mpx_->output_buffer(hdl());
}
std::string addr() const override {
return "test";
}
uint16_t port() const override {
return mpx_->port(hdl());
}
void launch() {
// nop
}
void flush() override {
// nop
}
void add_to_loop() override {
mpx_->passive_mode(hdl()) = false;
}
void remove_from_loop() override {
mpx_->passive_mode(hdl()) = true;
}
private:
test_multiplexer* mpx_;
};
......@@ -311,31 +329,55 @@ expected<void> test_multiplexer::assign_dgram_doorman(abstract_broker* ptr,
mpx_(mpx) {
// nop
}
bool new_endpoint() override {
// TODO: Implement me
/*
auto& mm = mpx_->pending_connects();
auto i = mm.find(hdl());
bool result = true;
if (i != mm.end()) {
result = doorman::new_connection(mpx_, i->second);
mm.erase(i);
}
return result;
*/
return false;
}
void configure_datagram_size(size_t buf_size) override {
mpx_->buffer_size(hdl()) = buf_size;
}
void stop_reading() override {
mpx_->stopped_reading(hdl()) = true;
detach(mpx_, false);
}
std::vector<char>& rd_buf() override {
return mpx_->input_buffer(hdl());
}
std::string addr() const override {
return "test";
}
uint16_t port() const override {
return mpx_->port(hdl());
}
void launch() override {
// nop
}
void add_to_loop() override {
mpx_->passive_mode(hdl()) = false;
}
void remove_from_loop() override {
mpx_->passive_mode(hdl()) = true;
}
private:
test_multiplexer* mpx_;
};
......@@ -437,6 +479,11 @@ uint16_t& test_multiplexer::port(accept_handle hdl) {
return doorman_data_[hdl].port;
}
test_multiplexer::buffer_type&
test_multiplexer::input_buffer(dgram_scribe_handle hdl) {
return dgram_scribe_data_[hdl].rd_buf;
}
test_multiplexer::buffer_type&
test_multiplexer::output_buffer(dgram_scribe_handle hdl) {
return dgram_scribe_data_[hdl].wr_buf;
......
......@@ -72,34 +72,34 @@ behavior make_pong_behavior() {
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(datagrams, fixture)
/*
CAF_TEST(test_remote_endpoint) {
CAF_TEST(test_dgram_scribe) {
auto& mp = client_side_mm.backend();
auto hdl = client_side_mm.named_broker<basp_broker>(atom("BASP"));
auto basp = static_cast<basp_broker*>(actor_cast<abstract_actor*>(hdl));
CAF_MESSAGE("Calling new_remote_endpoint");
auto res1 = mp.new_remote_endpoint(host, port + 0);
auto res1 = mp.new_dgram_scribe(host, port + 0);
CAF_REQUIRE(res1);
CAF_MESSAGE("Calling assign_endpoint (on \"remote\" endpoint).");
auto res2 = mp.assign_endpoint(basp, *res1);
auto res2 = mp.assign_dgram_scribe(basp, *res1);
CAF_REQUIRE(res2);
CAF_MESSAGE("Calling add_remote_endpoint.");
auto res3 = mp.add_remote_endpoint(basp, host, port + 1);
auto res3 = mp.add_dgram_scribe(basp, host, port + 1);
CAF_REQUIRE(res3);
}
CAF_TEST(test_local_endpoint) {
CAF_TEST(test_dgram_doorman) {
auto& mp = client_side_mm.backend();
auto hdl = client_side_mm.named_broker<basp_broker>(atom("BASP"));
auto basp = static_cast<basp_broker*>(actor_cast<abstract_actor*>(hdl));
CAF_MESSAGE("Calling new_local_endpoint.");
auto res1 = mp.new_local_endpoint(port + 0);
auto res1 = mp.new_dgram_doorman(port + 0);
CAF_REQUIRE(res1);
CAF_MESSAGE("Calling assign_endpoint (on \"local\" endpoint).");
auto res2 = mp.assign_endpoint(basp, res1->first);
auto res2 = mp.assign_dgram_doorman(basp, res1->first);
CAF_REQUIRE(res2);
CAF_MESSAGE("Calling add_local_endpoint.");
auto res3 = mp.add_local_endpoint(basp, port + 1, nullptr);
auto res3 = mp.add_dgram_doorman(basp, port + 1, nullptr);
CAF_REQUIRE(res3);
}
......@@ -134,5 +134,5 @@ CAF_TEST(test_datagram_remote_actor) {
anon_send_exit(pong, exit_reason::user_shutdown);
}
*/
CAF_TEST_FIXTURE_SCOPE_END()
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