Commit 63a997d7 authored by Joseph Noir's avatar Joseph Noir

WIP passing datagrams through the stack

parent 645c15ac
......@@ -110,31 +110,31 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
optional<response_promise> callback;
};
// stores meta information for dagram endpoints
// stores meta information for dgram endpoints
struct endpoint_context {
// denotes what message we expect from the remote node next
basp::connection_state cstate;
// out current processed BSAP header
// stores information about remote datagram enpoints
struct remote_endpoint {
// local sink
datagram_sink_handle sink;
// ordering information
uint32_t sequence_number_send = 0;
uint32_t sequence_number_receive = 0;
// TODO: local buffer for ordering
uint16_t remote_port;
}
// our current processed BSAP header
basp::header hdr;
// local sink
datagram_sink_handle sink;
// local source
datagram_source_handle source;
// network-agnositc node identifier
node_id id;
// ports
uint16_t local_port;
uint16_t remote_port;
// ordering information
uint32_t sequence_number_send = 0;
uint32_t sequence_number_receive = 0;
// TODO: local buffer for ordering
// pending operations to be performed after handshake completed
optional<response_promise> callback;
// TODO: reliability things
};
void set_context(connection_handle hdl);
void set_context(datagram_source_handle hdl);
// pointer to ourselves
broker* self;
......@@ -144,7 +144,7 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// keeps context information for all open connections
std::unordered_map<connection_handle, connection_context> tcp_ctx;
std::unordered_map<datagram_sink_handle, endpoint_context> udp_ctx;
std::unordered_map<datagram_source_handle, endpoint_context> udp_ctx;
// points to the current context for callbacks such as `make_proxy`
connection_context* this_context = nullptr;
......
......@@ -44,7 +44,7 @@ public:
~datagram_source();
/// Implicitly starts the read loop on first call.
/// Configure buffer size for next accepted datagram.
virtual void configure_datagram_size(size_t buf_size) = 0;
/// Returns the current input buffer.
......@@ -54,6 +54,9 @@ public:
void io_failure(execution_unit* ctx, network::operation op) override;
// needs to be launched explicitly
virtual void launch() = 0;
protected:
message detach_message() override;
};
......
......@@ -152,6 +152,7 @@ public:
/// a remote actor or an `error`.
template <class ActorHandle = actor>
expected<ActorHandle> remote_actor(uri u) {
CAF_LOG_TRACE(CAF_ARG(u));
detail::type_list<ActorHandle> tk;
auto x = remote_actor(system().message_types(tk), u);
if (!x)
......
......@@ -284,6 +284,10 @@ private:
intrusive_ptr<uri_private> d_;
};
inline std::string to_string(const str_bounds& bounds) {
return std::string(bounds.first, bounds.second);
}
} // namespace io
} // namespace caf
......@@ -293,6 +297,10 @@ inline ostream& operator<<(ostream& ostr, const caf::io::uri& what) {
return (ostr << what.str());
}
inline ostream& operator<<(ostream& ostr, const caf::io::str_bounds& bounds) {
return (ostr << string(bounds.first, bounds.second));
}
} // namespace std
#endif // CAF_IO_URI_HPP
......@@ -220,6 +220,7 @@ abstract_broker::add_datagram_sink(network::native_socket fd) {
void
abstract_broker::add_datagram_source(const intrusive_ptr<datagram_source>& ptr) {
datagram_sources_.emplace(ptr->hdl(), ptr);
ptr->launch();
// TODO: some launching of things?
}
......
......@@ -465,6 +465,30 @@ void basp_broker_state::set_context(connection_handle hdl) {
this_context = &i->second;
}
void basp_broker_state::set_context(datagram_source_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
/*
auto i = udp_ctx.find(hdl);
if (i == udp_ctx.end()) {
i = udp_ctx.emplace(
hdl,
endpoint_context{
basp::header{
basp::message_type::client_handshake, 0, 0, 0, none, none,
invalid_actor_id, invalid_actor_id
},
none, hdl,
none,
0, 0,
0, 0,
none
}
).first;
}
// TODO: set some context?
*/
}
/******************************************************************************
* basp_broker *
******************************************************************************/
......@@ -521,6 +545,13 @@ behavior basp_broker::make_behavior() {
ctx.cstate = next;
}
},
// received from underlying broker implementation
[=](new_datagram_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
CAF_LOG_DEBUG("Received new_datagram_msg: " << CAF_ARG(msg));
auto& hdl = msg.handle;
state.instance.handle(context(), msg,
},
// received from proxy instances
[=](forward_atom, strong_actor_ptr& src,
const std::vector<strong_actor_ptr>& fwd_stack,
......@@ -661,11 +692,9 @@ behavior basp_broker::make_behavior() {
auto& ctx = state.udp_ctx[hdl];
ctx.sink = hdl;
ctx.remote_port = port;
ctx.cstate = basp::await_header;
ctx.callback = rp;
// await server handshake
// TODO: Is there special processing required?
// configure_read(hdl, receive_policy::exactly(basp::header_size));
// TODO: Start handshake with server as there is no way for
// the server to initiate this.
} else {
CAF_LOG_DEBUG("failed to assign datagram sink from handle"
<< CAF_ARG(res));
......@@ -687,6 +716,7 @@ behavior basp_broker::make_behavior() {
return unit;
},
[=](close_atom, uint16_t port) -> result<void> {
// TODO: Should this accept a uri to close only related ports?
if (port == 0)
return sec::cannot_close_invalid_port;
// it is well-defined behavior to not have an actor published here,
......
......@@ -87,7 +87,7 @@ bool cc_valid_socket(caf::io::network::native_socket fd) {
// calls a C functions and returns an error if `predicate(var)` returns false
#define CALL_CFUN(var, predicate, fun_name, expr) \
auto var = expr; \
if (!predicate(var)) \
if (!predicate(var)) \
return make_error(sec::network_syscall_failed, \
fun_name, last_socket_error_as_string())
......@@ -95,7 +95,7 @@ bool cc_valid_socket(caf::io::network::native_socket fd) {
#ifdef CAF_WINDOWS
#define CALL_CRITICAL_CFUN(var, predicate, funname, expr) \
auto var = expr; \
if (!predicate(var)) { \
if (!predicate(var)) { \
fprintf(stderr, "[FATAL] %s:%u: syscall failed: %s returned %s\n", \
__FILE__, __LINE__, funname, last_socket_error_as_string().c_str());\
abort(); \
......@@ -949,7 +949,7 @@ default_multiplexer::add_datagram_source(abstract_broker* self,
return 0;
return *x;
}
void launch() {
void launch() override {
CAF_LOG_TRACE("");
CAF_ASSERT(!launched_);
launched_ = true;
......@@ -1493,7 +1493,7 @@ void datagram_sender::flush(const manager_ptr& mgr) {
}
void datagram_sender::start(manager_type* mgr) {
CAF_ASSERT(mgr != nullptr);
CAF_ASSERT(mgr != nullptr);
activate(mgr);
}
......@@ -1601,8 +1601,7 @@ void datagram_receiver::handle_event(operation op) {
CAF_LOG_TRACE(CAF_ARG(op));
switch (op) {
case operation::read: {
// loop until an error occurs or we have nothing more to read
// or until we have handled 50 reads
// Read next datagram
size_t rb;
if (!receive_datagram(rb, fd(), rd_buf_.data(), rd_buf_.size(),
last_sender, sender_len)) {
......@@ -1860,7 +1859,7 @@ expected<native_socket> new_datagram_sink_impl(const std::string& host,
}
sguard.close();
// IPv4 fallback
return new_tcp_connection(host, port, ipv4);
return new_datagram_sink_impl(host, port, ipv4);
}
if (!ip_connect<AF_INET>(fd, res->first, port)) {
CAF_LOG_INFO("could not connect to:" << CAF_ARG(host) << CAF_ARG(port));
......
......@@ -157,7 +157,7 @@ expected<node_id> middleman::connect(uri u) {
expected<uint16_t> middleman::publish(const strong_actor_ptr& whom,
std::set<std::string> sigs,
uri u, bool ru) {
CAF_LOG_TRACE(CAF_ARG(whom) << CAF_ARG(sigs) << CAF_ARG(port));
CAF_LOG_TRACE(CAF_ARG(whom) << CAF_ARG(sigs) << CAF_ARG(u));
if (!whom)
return sec::cannot_publish_invalid_actor;
auto f = make_function_view(actor_handle());
......@@ -187,7 +187,7 @@ expected<uint16_t> middleman::publish_local_groups(uint16_t port,
}
expected<void> middleman::unpublish(const actor_addr& whom, uri u) {
CAF_LOG_TRACE(CAF_ARG(whom) << CAF_ARG(port));
CAF_LOG_TRACE(CAF_ARG(whom) << CAF_ARG(u));
auto f = make_function_view(actor_handle());
return f(unpublish_atom::value, whom, u);
}
......
......@@ -94,6 +94,7 @@ public:
using endpoint = io::uri;
behavior_type make_behavior() override {
CAF_SET_LOGGER_SYS(&this->system()); // TODO: remove this?
CAF_LOG_TRACE("");
return {
[=](publish_atom, strong_actor_ptr& whom,
......@@ -128,7 +129,7 @@ public:
}
if (std::distance(u.scheme().first, u.scheme().second) >= 3 &&
equal(std::begin(tcp), std::end(tcp), u.scheme().first)) {
// connect to endpoint and initiate handhsake etc.
// connect to endpoint and initiate handshake etc. (TCP)
auto y = system().middleman().backend().new_tcp_scribe(host, port);
if (!y) {
rp.deliver(std::move(y.error()));
......@@ -163,7 +164,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.
// connect to endpoint and initiate handhsake etc. (UDP)
auto y = system().middleman().backend().new_datagram_sink(host, port);
if (!y) {
rp.deliver(std::move(y.error()));
......@@ -196,8 +197,10 @@ public:
pending_.erase(i);
}
);
} else {
// sec::unsupported_protocol;
}
return {}; // sec::unsupported_protocol;
return {};
},
[=](unpublish_atom atm, actor_addr addr, uri& u) -> del_res {
CAF_LOG_TRACE("");
......@@ -230,8 +233,8 @@ public:
private:
put_res put(const uri& u, strong_actor_ptr& whom,
mpi_set& sigs, bool reuse_addr = false) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(whom) << CAF_ARG(sigs)
<< CAF_ARG(in) << CAF_ARG(reuse_addr));
CAF_LOG_TRACE(CAF_ARG(u) << CAF_ARG(whom) << CAF_ARG(sigs)
<< CAF_ARG(reuse_addr));
uint16_t actual_port;
// treat empty strings or 0.0.0.0 as nullptr
auto addr = std::string(u.host().first, u.host().second);
......@@ -240,6 +243,7 @@ private:
in = addr.c_str();
if (std::distance(u.scheme().first, u.scheme().second) >= 3 &&
equal(u.scheme().first, u.scheme().second, std::begin(tcp))) {
// TCP
auto res = system().middleman().backend().new_tcp_doorman(u.port_as_int(),
in, reuse_addr);
if (!res)
......@@ -250,6 +254,7 @@ private:
std::move(whom), std::move(sigs));
} else if (std::distance(u.scheme().first, u.scheme().second) >= 3 &&
equal(u.scheme().first, u.scheme().second, std::begin(udp))) {
// UDP
auto res
= system().middleman().backend().new_datagram_source(u.port_as_int(),
in, reuse_addr);
......
......@@ -324,7 +324,7 @@ expected<void> test_multiplexer::assign_datagram_source(abstract_broker* ptr,
uint16_t port() const override {
return mpx_->port(hdl());
}
void launch() {
void launch() override {
// nop
}
void add_to_loop() override {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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/config.hpp"
#define CAF_SUITE io_datagram
#include "caf/test/unit_test.hpp"
#include "caf/all.hpp"
#include "caf/detail/build_config.hpp"
#include "caf/io/all.hpp"
using namespace std;
using namespace caf;
using namespace caf::io;
namespace {
constexpr auto host = "127.0.0.1";
constexpr uint16_t port = 1234;
constexpr auto uri_no_port = "udp://127.0.0.1";
// constexpr auto uri_with_port = "udp://127.0.0.1:1234";
class config : public actor_system_config {
public:
config() {
load<io::middleman>();
add_message_type<vector<int>>("vector<int>");
actor_system_config::parse(test::engine::argc(),
test::engine::argv());
}
};
struct fixture {
//fixture() {
// CAF_SET_LOGGER_SYS(&server_side);
// CAF_SET_LOGGER_SYS(&client_side);
//}
config server_side_config;
actor_system server_side{server_side_config};
config client_side_config;
actor_system client_side{client_side_config};
io::middleman& server_side_mm = server_side.middleman();
io::middleman& client_side_mm = client_side.middleman();
};
behavior make_pong_behavior() {
return {
[](int val) -> int {
CAF_MESSAGE("pong with " << ++val);
return val;
}
};
}
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(datagrams, fixture)
CAF_TEST(test_datagram_sinks) {
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_datagram_sink");
auto res1 = mp.new_datagram_sink(host, port + 0);
CAF_REQUIRE(res1);
CAF_MESSAGE("Calling assign_datagram_sink");
auto res2 = mp.assign_datagram_sink(basp, *res1);
CAF_REQUIRE(res2);
CAF_MESSAGE("Calling add_datagram_sink");
auto res3 = mp.add_datagram_sink(basp, host, port + 1);
CAF_REQUIRE(res3);
}
CAF_TEST(test_datagram_sources) {
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_datagram_source");
auto res1 = mp.new_datagram_source(port + 0);
CAF_REQUIRE(res1);
CAF_MESSAGE("Calling assign_datagram_source");
auto res2 = mp.assign_datagram_source(basp, res1->first);
CAF_REQUIRE(res2);
CAF_MESSAGE("Calling add_datagram_source");
auto res3 = mp.add_datagram_source(basp, port + 1, nullptr);
CAF_REQUIRE(res3);
}
CAF_TEST(test_datagram_publish) {
auto pong = client_side.spawn(make_pong_behavior);
auto res1 = io::uri::make(uri_no_port);
CAF_REQUIRE(res1);
auto res2 = client_side_mm.publish(pong, *res1);
CAF_REQUIRE(res2);
CAF_MESSAGE("Published pong on port: " + to_string(*res2));
anon_send_exit(pong, exit_reason::user_shutdown);
}
CAF_TEST(test_datagram_remote_actor) {
auto pong = server_side.spawn(make_pong_behavior);
auto res1 = io::uri::make(uri_no_port);
CAF_REQUIRE(res1);
auto res2 = server_side_mm.publish(pong, *res1);
CAF_CHECK(res2);
auto res3 = io::uri::make(string(uri_no_port) + ":" + to_string(*res2));
CAF_CHECK(res3);
CAF_MESSAGE("Published pong on: " + to_string(*res3) + ".");
CAF_SET_LOGGER_SYS(&server_side);
CAF_MESSAGE("Local call to remote actor should acquire the actor.");
auto res4 = server_side_mm.remote_actor(*res3);
CAF_CHECK(res4);
CAF_MESSAGE("Checking from different actor system next.");
auto res5 = client_side_mm.remote_actor(*res3);
CAF_CHECK(res5);
anon_send_exit(pong, exit_reason::user_shutdown);
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -35,7 +35,8 @@ using namespace std;
namespace {
constexpr char host_uri[] = "tcp://127.0.0.1";
constexpr char uri_tcp[] = "tcp://127.0.0.1";
constexpr char uri_udp[] = "udp://127.0.0.1";
class config : public actor_system_config {
public:
......@@ -142,14 +143,14 @@ CAF_TEST_FIXTURE_SCOPE(dynamic_remote_actor_tests, fixture)
CAF_TEST(identity_semantics) {
// server side
auto server = server_side.spawn(make_pong_behavior);
auto uri_no_port = io::uri::make(host_uri);
auto uri_no_port = io::uri::make(uri_tcp);
CAF_REQUIRE(uri_no_port);
CAF_EXP_THROW(port1, server_side_mm.publish(server, *uri_no_port));
CAF_EXP_THROW(port2, server_side_mm.publish(server, *uri_no_port));
CAF_REQUIRE_NOT_EQUAL(port1, port2);
auto uri_port1 = io::uri::make(string(host_uri) + ":" + to_string(port1));
auto uri_port1 = io::uri::make(string(uri_tcp) + ":" + to_string(port1));
CAF_REQUIRE(uri_port1);
auto uri_port2 = io::uri::make(string(host_uri) + ":" + to_string(port2));
auto uri_port2 = io::uri::make(string(uri_tcp) + ":" + to_string(port2));
CAF_REQUIRE(uri_port2);
CAF_EXP_THROW(same_server, server_side_mm.remote_actor(*uri_port2));
CAF_REQUIRE_EQUAL(same_server, server);
......@@ -163,13 +164,13 @@ CAF_TEST(identity_semantics) {
CAF_TEST(ping_pong) {
// server side
auto server_uri = io::uri::make(host_uri);
auto server_uri = io::uri::make(uri_tcp);
CAF_REQUIRE(server_uri);
CAF_EXP_THROW(port,
server_side_mm.publish(server_side.spawn(make_pong_behavior),
*server_uri));
// client side
auto uri_with_port = io::uri::make(string(host_uri) + ":" + to_string(port));
auto uri_with_port = io::uri::make(string(uri_tcp) + ":" + to_string(port));
CAF_REQUIRE(uri_with_port);
CAF_EXP_THROW(pong, client_side_mm.remote_actor(*uri_with_port));
client_side.spawn(make_ping_behavior, pong);
......@@ -177,12 +178,12 @@ CAF_TEST(ping_pong) {
CAF_TEST(custom_message_type) {
// server side
auto server_uri = io::uri::make(host_uri);
auto server_uri = io::uri::make(uri_tcp);
CAF_REQUIRE(server_uri);
CAF_EXP_THROW(port, server_side_mm.publish(server_side.spawn(make_sort_behavior),
*server_uri));
// client side
auto uri_with_port = io::uri::make(string(host_uri) + ":" + to_string(port));
auto uri_with_port = io::uri::make(string(uri_tcp) + ":" + to_string(port));
CAF_REQUIRE(uri_with_port);
CAF_EXP_THROW(sorter, client_side_mm.remote_actor(*uri_with_port));
client_side.spawn(make_sort_requester_behavior, sorter);
......@@ -190,12 +191,89 @@ CAF_TEST(custom_message_type) {
CAF_TEST(remote_link) {
// server side
auto server_uri = io::uri::make(host_uri);
auto server_uri = io::uri::make(uri_tcp);
CAF_REQUIRE(server_uri);
CAF_EXP_THROW(port, server_side_mm.publish(server_side.spawn(fragile_mirror),
*server_uri));
// client side
auto uri_with_port = io::uri::make(string(host_uri) + ":" + to_string(port));
auto uri_with_port = io::uri::make(string(uri_tcp) + ":" + to_string(port));
CAF_REQUIRE(uri_with_port);
CAF_EXP_THROW(mirror, client_side_mm.remote_actor(*uri_with_port));
auto linker = client_side.spawn(linking_actor, mirror);
scoped_actor self{client_side};
self->wait_for(linker);
CAF_MESSAGE("linker exited");
self->wait_for(mirror);
CAF_MESSAGE("mirror exited");
}
// same test using UDP instead of TCP
CAF_TEST(identity_semantics_udp) {
// server side
auto server = server_side.spawn(make_pong_behavior);
auto uri_no_port = io::uri::make(uri_udp);
CAF_REQUIRE(uri_no_port);
CAF_EXP_THROW(port1, server_side_mm.publish(server, *uri_no_port));
CAF_EXP_THROW(port2, server_side_mm.publish(server, *uri_no_port));
CAF_REQUIRE_NOT_EQUAL(port1, port2);
auto uri_port1 = io::uri::make(string(uri_udp) + ":" + to_string(port1));
CAF_REQUIRE(uri_port1);
auto uri_port2 = io::uri::make(string(uri_udp) + ":" + to_string(port2));
CAF_MESSAGE("Created URIs to acquire proxies.");
CAF_REQUIRE(uri_port2);
CAF_EXP_THROW(same_server, server_side_mm.remote_actor(*uri_port2));
CAF_MESSAGE("Acquired local actor through remote_actor call.");
CAF_REQUIRE_EQUAL(same_server, server);
CAF_CHECK_EQUAL(same_server->node(), server_side.node());
CAF_MESSAGE("Acquiring proxies through separate actor systems.");
CAF_EXP_THROW(server1, client_side_mm.remote_actor(*uri_port1));
CAF_EXP_THROW(server2, client_side_mm.remote_actor(*uri_port2));
CAF_MESSAGE("Multiple proxies of the same actor should be the same.");
CAF_CHECK_EQUAL(server1, client_side_mm.remote_actor(*uri_port1));
CAF_CHECK_EQUAL(server2, client_side_mm.remote_actor(*uri_port2));
CAF_MESSAGE("Shutting down server.");
anon_send_exit(server, exit_reason::user_shutdown);
}
CAF_TEST(ping_pong_udp) {
// server side
auto server_uri = io::uri::make(uri_udp);
CAF_REQUIRE(server_uri);
CAF_EXP_THROW(port,
server_side_mm.publish(server_side.spawn(make_pong_behavior),
*server_uri));
CAF_MESSAGE("Created server.");
// client side
auto uri_with_port = io::uri::make(string(uri_udp) + ":" + to_string(port));
CAF_REQUIRE(uri_with_port);
CAF_EXP_THROW(pong, client_side_mm.remote_actor(*uri_with_port));
CAF_MESSAGE("Acquired actor proxy.");
client_side.spawn(make_ping_behavior, pong);
CAF_MESSAGE("Started ping-pong.");
}
CAF_TEST(custom_message_type_udp) {
// server side
auto server_uri = io::uri::make(uri_udp);
CAF_REQUIRE(server_uri);
CAF_EXP_THROW(port, server_side_mm.publish(server_side.spawn(make_sort_behavior),
*server_uri));
// client side
auto uri_with_port = io::uri::make(string(uri_udp) + ":" + to_string(port));
CAF_REQUIRE(uri_with_port);
CAF_EXP_THROW(sorter, client_side_mm.remote_actor(*uri_with_port));
client_side.spawn(make_sort_requester_behavior, sorter);
}
CAF_TEST(remote_link_udp) {
// server side
auto server_uri = io::uri::make(uri_udp);
CAF_REQUIRE(server_uri);
CAF_EXP_THROW(port, server_side_mm.publish(server_side.spawn(fragile_mirror),
*server_uri));
// client side
auto uri_with_port = io::uri::make(string(uri_udp) + ":" + to_string(port));
CAF_REQUIRE(uri_with_port);
CAF_EXP_THROW(mirror, client_side_mm.remote_actor(*uri_with_port));
auto linker = client_side.spawn(linking_actor, mirror);
......
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