Commit 96218a3b authored by Joseph Noir's avatar Joseph Noir

This seems to work for testing

parent fbcedf80
......@@ -375,8 +375,9 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
template<class Rep = int, class Period = std::ratio<1>>
void set_timeout(std::chrono::duration<Rep, Period> timeout,
atom_value atm, uint32_t id) {
auto n = actor_clock::clock_type::now();
scheduled_actor::clock().set_ordinary_timeout(n + timeout, this, atm, id);
//auto n = actor_clock::clock_type::now();
//scheduled_actor::clock().set_ordinary_timeout(n + timeout, this, atm, id);
this->delayed_send(this, timeout, atm, id);
}
/// Returns the `multiplexer` running this broker.
......@@ -541,9 +542,8 @@ struct newb_acceptor : public newb_base, public caf::ref_counted {
writing_(false),
args_(std::forward<Ts>(xs)...) {
// nop
if (sockfd == io::network::invalid_native_socket) {
std::cerr << "Creating newb with invalid socket" << std::endl;
}
if (sockfd == io::network::invalid_native_socket)
CAF_LOG_ERROR("Creating newb with invalid socket");
}
newb_acceptor(const newb_acceptor& other) = delete;
......
......@@ -21,6 +21,7 @@
#include <cstdint>
#include <unordered_map>
#include <chrono>
#include <random>
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
......@@ -31,7 +32,7 @@ namespace caf {
namespace policy {
using id_type = uint16_t;
using reliability_atom = atom_constant<atom("ordering")>;
using reliability_atom = atom_constant<atom("reliabilit")>;
struct reliability_header {
id_type id;
......@@ -56,10 +57,17 @@ struct reliability {
using result_type = typename Next::result_type;
id_type id_write = 0;
// TODO: Make this configurable.
std::chrono::milliseconds retransmit_to = std::chrono::milliseconds(100);
std::chrono::milliseconds retransmit_to = std::chrono::milliseconds(40);
io::newb<message_type>* parent;
Next next;
std::unordered_map<id_type, io::byte_buffer> unacked;
std::random_device rd;
std::mt19937 mt;
std::uniform_int_distribution<int> dist;
reliability() : mt(rd()), dist(0, 9) {
// nop
}
void init(io::newb<message_type>* n) {
parent = n;
......@@ -72,10 +80,17 @@ struct reliability {
reliability_header hdr;
binary_deserializer bd(&parent->backend(), bytes, count);
bd(hdr);
auto r = dist(mt);
if (r == 0) {
std::cerr << "not this time: " << hdr.id << std::endl;
return none;
}
if (hdr.is_ack) {
// TODO: Cancel timeout.
unacked.erase(hdr.id);
std::cerr << "got ack for " << hdr.id << std::endl;
} else {
std::cerr << "got header: " << hdr.id << std::endl;
// Send ack.
auto& buf = parent->wr_buf();
binary_serializer bs(&parent->backend(), buf);
......@@ -96,7 +111,7 @@ struct reliability {
// Retransmit the packet.
auto& packet = unacked[retransmit_id];
auto& buf = parent->wr_buf();
buf.insert(buf.begin(), packet.begin(), packet.end());
buf.insert(buf.end(), packet.begin(), packet.end());
parent->flush();
parent->set_timeout(retransmit_to, reliability_atom::value, retransmit_id);
}
......@@ -121,6 +136,7 @@ struct reliability {
// Add to unacked.
unacked.emplace(id_write,
io::byte_buffer(buf.begin() + hstart, buf.end()));
std::cerr << "awaiting ack for " << id_write << std::endl;
id_write += 1;
}
};
......
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