Commit 500d672c authored by Joseph Noir's avatar Joseph Noir

Add test for UDP ordering

parent 8164dc6a
......@@ -120,6 +120,12 @@ public:
/// input buffer usually managed by the operating system.
buffer_type& virtual_network_buffer(dgram_scribe_handle hdl);
/// Models pending data on the network, i.e., the network
/// input buffer usually manged by the operating system.
/// Allows enqueueing messages to the front to allow
/// simulation of out of order delivery.
buffer_type& virtual_network_buffer_front(dgram_scribe_handle hdl);
/// Models pending data on the network, i.e., the network
/// input buffer usually managed by the operating system.
buffer_type& virtual_network_buffer(dgram_doorman_handle hdl);
......@@ -302,8 +308,6 @@ private:
uint16_t local_port;
std::deque<buffer_type> xbuf;
std::deque<buffer_type> wr_buf;
//buffer_type xbuf;
//buffer_type wr_buf;
buffer_type rd_buf;
size_t rcv_buffer_size = 1500;
bool stopped_reading = false;
......
......@@ -58,6 +58,9 @@ bool instance::deliver_pending(execution_unit* ctx, endpoint_context& ep) {
std::vector<char>* payload = nullptr;
auto itr = ep.pending.find(ep.seq_incoming);
while (itr != ep.pending.end()) {
//std::cerr << "[++] '" << to_string(itr->second.first.operation)
// << "' with seq '" << itr->second.first.sequence_number
// << "' (was pending)" << std::endl;
ep.hdr = std::move(itr->second.first);
payload = &itr->second.second;
if (!handle_msg(ctx, get<dgram_scribe_handle>(ep.hdl),
......@@ -160,7 +163,7 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm,
auto e = bd(ep.hdr);
if (e || !valid(ep.hdr)) {
CAF_LOG_WARNING("received invalid header:" << CAF_ARG(ep.hdr));
std::cerr << "[!!] invalid header!" << std::endl;
//std::cerr << "[!!] invalid header!" << std::endl;
return err();
}
CAF_LOG_DEBUG(CAF_ARG(ep.hdr));
......@@ -173,12 +176,12 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm,
}
}
// TODO: Ordering
std::cerr << "[<<] '" << to_string(ep.hdr.operation)
<< "' with seq '" << ep.hdr.sequence_number << "'" << std::endl;
//std::cerr << "[<<] '" << to_string(ep.hdr.operation)
// << "' with seq '" << ep.hdr.sequence_number << "'" << std::endl;
if (ep.hdr.sequence_number != ep.seq_incoming) {
std::cerr << "[!!] '" << to_string(ep.hdr.operation) << "' with seq '"
<< ep.hdr.sequence_number << "' (!= " << ep.seq_incoming
<< ")" << std::endl;
//std::cerr << "[!!] '" << to_string(ep.hdr.operation) << "' with seq '"
// << ep.hdr.sequence_number << "' (!= " << ep.seq_incoming
// << ")" << std::endl;
auto s = ep.hdr.sequence_number;
auto h = std::move(ep.hdr);
auto b = std::move(pl_buf);
......@@ -220,18 +223,18 @@ bool instance::handle(execution_unit* ctx, new_endpoint_msg& em,
// client handshake for UDP should be sequence number 0
if (e || !valid(ep.hdr)) {
CAF_LOG_WARNING("received invalid header:" << CAF_ARG(ep.hdr));
std::cerr << "[<<] invalid header!" << std::endl;
//std::cerr << "[<<] invalid header!" << std::endl;
return err();
}
if (ep.hdr.sequence_number != ep.seq_incoming) {
CAF_LOG_WARNING("Handshake with unexected sequence number: "
<< CAF_ARG(ep.hdr.sequence_number));
std::cerr << "[<<] Unexpected sequence number '" << ep.hdr.sequence_number
<< "'in client handshake" << std::endl;
//std::cerr << "[<<] Unexpected sequence number '" << ep.hdr.sequence_number
// << "'in client handshake" << std::endl;
ep.seq_incoming = ep.hdr.sequence_number + 1;
} else {
std::cerr << "[<<] '" << to_string(ep.hdr.operation) << "' with seq '"
<< ep.hdr.sequence_number << "'" << std::endl;
//std::cerr << "[<<] '" << to_string(ep.hdr.operation) << "' with seq '"
// << ep.hdr.sequence_number << "'" << std::endl;
ep.seq_incoming += 1;
}
CAF_LOG_DEBUG(CAF_ARG(ep.hdr));
......@@ -369,9 +372,9 @@ 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 << "[>>] '" << to_string(hdr.operation)
<< "' with seq '" << hdr.sequence_number << "'"
<< std::endl;
//std::cerr << "[>>] '" << to_string(hdr.operation)
// << "' with seq '" << hdr.sequence_number << "'"
// << std::endl;
error err;
if (pw) {
auto pos = buf.size();
......
......@@ -471,6 +471,13 @@ test_multiplexer::virtual_network_buffer(dgram_scribe_handle hdl) {
return buf.back();
}
test_multiplexer::buffer_type&
test_multiplexer::virtual_network_buffer_front(dgram_scribe_handle hdl) {
auto& buf = dgram_scribe_data_[hdl].xbuf;
buf.emplace_front();
return buf.front();
}
test_multiplexer::buffer_type&
test_multiplexer::virtual_network_buffer(dgram_doorman_handle hdl) {
auto& buf = dgram_doorman_data_[hdl].xbuf;
......
......@@ -540,6 +540,36 @@ public:
return *this;
}
template <class... Ts>
mock_t& enqueue_back(dgram_scribe_handle hdl, basp::header hdr,
const Ts&... xs) {
buffer buf;
this_->to_buf(buf, hdr, nullptr, xs...);
CAF_MESSAGE("adding msg " << to_string(hdr.operation)
<< " with " << (buf.size() - basp::header_size)
<< " bytes payload to back of queue");
this_->mpx()->virtual_network_buffer(hdl) = buf;
return *this;
}
template <class... Ts>
mock_t& enqueue_front(dgram_scribe_handle hdl, basp::header hdr,
const Ts&... xs) {
buffer buf;
this_->to_buf(buf, hdr, nullptr, xs...);
CAF_MESSAGE("adding msg " << to_string(hdr.operation)
<< " with " << (buf.size() - basp::header_size)
<< " bytes payload to front of queue");
this_->mpx()->virtual_network_buffer_front(hdl) = buf;
return *this;
}
mock_t& deliver(dgram_scribe_handle hdl, size_t num_messages = 1) {
for (size_t i = 0; i < num_messages; ++i)
this_->mpx()->read_datagram(hdl);
return *this;
}
private:
fixture* this_;
size_t num = 1;
......@@ -1180,6 +1210,120 @@ CAF_TEST(remote_actor_and_send_udp) {
);
}
CAF_TEST(out_of_order_delivery_udp) {
constexpr const char* prot = "udp";
constexpr const char* lo = "localhost";
constexpr uint16_t port = 4242;
auto u = uri::make(std::string(prot) + "://" + std::string(lo) + ":" +
std::to_string(port));
CAF_REQUIRE(u);
CAF_MESSAGE("self: " << to_string(self()->address()));
mpx()->provide_dgram_scribe(lo, 4242, jupiter().connection);
CAF_REQUIRE(mpx()->has_pending_dgram_scribe(lo, 4242));
auto mm1 = system.middleman().actor_handle();
actor result;
auto f = self()->request(mm1, infinite, connect_atom::value, *u);
// wait until BASP broker has received and processed the connect message
while (!aut()->valid(jupiter().connection))
mpx()->exec_runnable();
CAF_REQUIRE(!mpx()->has_pending_dgram_scribe(lo, 4242));
// build a fake server handshake containing the id of our first pseudo actor
CAF_MESSAGE("server handshake => client handshake + proxy announcement");
auto na = registry()->named_actors();
mock(jupiter().connection,
{basp::message_type::server_handshake, 0, 0, basp::version,
jupiter().id, none,
jupiter().dummy_actor->id(), invalid_actor_id},
std::string{},
jupiter().dummy_actor->id(),
uint32_t{0})
.expect(jupiter().connection,
basp::message_type::client_handshake, no_flags, 1u,
no_operation_data, this_node(), node_id{none},
invalid_actor_id, invalid_actor_id, std::string{})
.expect(jupiter().connection,
basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
no_operation_data, this_node(), jupiter().id,
any_vals, invalid_actor_id,
spawn_serv_atom,
std::vector<actor_id>{},
make_message(sys_atom::value, get_atom::value, "info"))
.expect(jupiter().connection,
basp::message_type::announce_proxy, no_flags, no_payload,
no_operation_data, this_node(), jupiter().id,
invalid_actor_id, jupiter().dummy_actor->id());
CAF_MESSAGE("BASP broker should've send the proxy");
f.receive(
[&](node_id nid, strong_actor_ptr res, std::set<std::string> ifs) {
CAF_REQUIRE(res);
auto aptr = actor_cast<abstract_actor*>(res);
CAF_REQUIRE(dynamic_cast<forwarding_actor_proxy*>(aptr) != nullptr);
CAF_CHECK_EQUAL(proxies().count_proxies(jupiter().id), 1u);
CAF_CHECK_EQUAL(nid, jupiter().id);
CAF_CHECK_EQUAL(res->node(), jupiter().id);
CAF_CHECK_EQUAL(res->id(), jupiter().dummy_actor->id());
CAF_CHECK(ifs.empty());
auto proxy = proxies().get(jupiter().id, jupiter().dummy_actor->id());
CAF_REQUIRE(proxy != nullptr);
CAF_REQUIRE(proxy == res);
result = actor_cast<actor>(res);
},
[&](error& err) {
CAF_FAIL("error: " << system.render(err));
}
);
CAF_MESSAGE("send message to proxy");
anon_send(actor_cast<actor>(result), 42);
mpx()->flush_runnables();
mock()
.expect(jupiter().connection,
basp::message_type::dispatch_message, no_flags, any_vals,
no_operation_data, this_node(), jupiter().id,
invalid_actor_id, jupiter().dummy_actor->id(),
std::vector<actor_id>{},
make_message(42));
// auto msg = make_message("hi there!");
uint16_t seq_number = 1;
auto next_hdr = [&]() -> basp::header {
return {basp::message_type::dispatch_message, 0, 0, 0,
jupiter().id, this_node(),
jupiter().dummy_actor->id(), self()->id(),
seq_number++};
};
mock()
.enqueue_back(jupiter().connection, next_hdr(), std::vector<actor_id>{},
make_message(0))
.enqueue_back(jupiter().connection, next_hdr(), std::vector<actor_id>{},
make_message(1))
.enqueue_front(jupiter().connection, next_hdr(), std::vector<actor_id>{},
make_message(2))
.enqueue_back(jupiter().connection, next_hdr(), std::vector<actor_id>{},
make_message(3))
.enqueue_back(jupiter().connection, next_hdr(), std::vector<actor_id>{},
make_message(4))
.enqueue_back(jupiter().connection, next_hdr(), std::vector<actor_id>{},
make_message(5))
.enqueue_front(jupiter().connection, next_hdr(), std::vector<actor_id>{},
make_message(6))
.enqueue_back(jupiter().connection, next_hdr(), std::vector<actor_id>{},
make_message(7))
.enqueue_back(jupiter().connection, next_hdr(), std::vector<actor_id>{},
make_message(8))
.enqueue_front(jupiter().connection, next_hdr(), std::vector<actor_id>{},
make_message(9))
.deliver(jupiter().connection, 10);
int expected_next = 0;
self()->receive_while([&] { return expected_next < 9; }) (
[&](int val) {
CAF_CHECK_EQUAL(to_string(self()->current_sender()), to_string(result));
CAF_CHECK_EQUAL(self()->current_sender(), result.address());
CAF_CHECK_EQUAL(expected_next, val);
++expected_next;
}
);
}
CAF_TEST(actor_serialize_and_deserialize_udp) {
auto testee_impl = [](event_based_actor* testee_self) -> behavior {
testee_self->set_default_handler(reflect_and_quit);
......
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