Commit 1b542de9 authored by Joseph Noir's avatar Joseph Noir

Remove measurements

parent e40b6668
......@@ -56,13 +56,6 @@ add(remoting basp_tcp_newb)
add(remoting raw_tcp_newb)
add(remoting raw_udp_newb)
# measurements not meant to stay in CAF
add(measurements one_raw_tcp)
add(measurements one_raw_udp)
add(measurements one_basp_tcp)
add(measurements one_basp_udp)
add(measurements four_udp)
# basic I/O with brokers
add(broker simple_broker)
add(broker simple_http_broker)
......
#include "caf/io/network/newb.hpp"
#include "caf/logger.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/detail/call_cfun.hpp"
#include "caf/policy/newb_basp.hpp"
#include "caf/policy/newb_ordering.hpp"
#include "caf/policy/newb_udp.hpp"
using namespace caf;
using caf::io::network::default_multiplexer;
using caf::io::network::invalid_native_socket;
using caf::io::network::make_client_newb;
using caf::io::network::make_server_newb;
using caf::io::network::native_socket;
using caf::policy::accept_udp;
using caf::policy::udp_protocol;
using caf::policy::udp_transport;
namespace {
using ordering_atom = atom_constant<atom("ordering")>;
constexpr size_t chunk_size = 1024; //8192; //128; //1024;
struct dummy_transport : public io::network::transport_policy {
dummy_transport() {
// nop
}
inline error read_some(io::network::event_handler*) override {
return none;
}
inline bool should_deliver() override {
return true;
}
void prepare_next_read(io::network::event_handler*) override {
received_bytes = 0;
receive_buffer.resize(maximum);
}
inline void configure_read(io::receive_policy::config) override {
// nop
}
inline error write_some(io::network::event_handler* parent) override {
written += send_sizes.front();
send_sizes.pop_front();
auto remaining = send_buffer.size() - written;
count += 1;
if (remaining == 0)
prepare_next_write(parent);
return none;
}
void prepare_next_write(io::network::event_handler*) override {
written = 0;
send_buffer.clear();
send_sizes.clear();
if (offline_buffer.empty()) {
writing = false;
} else {
offline_sizes.push_back(offline_buffer.size() - offline_sum);
// Switch buffers.
send_buffer.swap(offline_buffer);
send_sizes.swap(offline_sizes);
// Reset sum.
offline_sum = 0;
}
}
io::network::byte_buffer& wr_buf() override {
if (!offline_buffer.empty()) {
auto chunk_size = offline_buffer.size() - offline_sum;
offline_sizes.push_back(chunk_size);
offline_sum += chunk_size;
}
return offline_buffer;
}
void flush(io::network::event_handler* parent) override {
if (!offline_buffer.empty() && !writing) {
writing = true;
prepare_next_write(parent);
}
}
expected<io::network::native_socket>
connect(const std::string&, uint16_t,
optional<io::network::protocol::network> = none) override {
return invalid_native_socket;
}
// State for reading.
size_t maximum;
bool first_message;
// State for writing.
bool writing;
size_t written;
size_t offline_sum;
std::deque<size_t> send_sizes;
std::deque<size_t> offline_sizes;
};
struct raw_newb : public io::network::newb<caf::policy::new_basp_message> {
using message_type = caf::policy::new_basp_message;
raw_newb(caf::actor_config& cfg, default_multiplexer& dm,
native_socket sockfd)
: newb<message_type>(cfg, dm, sockfd) {
// nop
CAF_LOG_TRACE("");
}
void handle(message_type&) override {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE("");
}
behavior make_behavior() override {
set_default_handler(print_and_drop);
return {
// Must be implemented at the moment, will be cought by the broker in a
// later implementation.
[=](atom_value atm, uint32_t id) {
protocol->timeout(atm, id);
}
};
}
};
class config : public actor_system_config {
public:
int iterations = 10;
config() {
opt_group{custom_options_, "global"}
.add(iterations, "iterations,i", "set iterations");
}
};
void caf_main(actor_system& sys, const config& cfg) {
using clock = std::chrono::system_clock;
using resolution = std::chrono::milliseconds;
auto n = io::network::make_newb<raw_newb>(sys, invalid_native_socket);
auto ptr = caf::actor_cast<caf::abstract_actor*>(n);
auto& ref = dynamic_cast<raw_newb&>(*ptr);
ref.transport.reset(new dummy_transport);
ref.protocol.reset(new udp_protocol<policy::ordering<caf::policy::datagram_basp>>(&ref));
auto start = clock::now();
for (int i = 0; i < cfg.iterations; ++i) {
auto hw = caf::make_callback([&](io::network::byte_buffer& buf) -> error {
binary_serializer bs(sys, buf);
bs(policy::basp_header{0, actor_id{}, actor_id{}});
return none;
});
auto whdl = ref.wr_buf(&hw);
CAF_ASSERT(whdl.buf != nullptr);
CAF_ASSERT(whdl.protocol != nullptr);
binary_serializer bs(sys, *whdl.buf);
auto start = whdl.buf->size();
whdl.buf->resize(start + chunk_size);
std::fill(whdl.buf->begin() + start, whdl.buf->end(), 'a');
}
auto end = clock::now();
auto ticks = std::chrono::duration_cast<resolution>(end - start).count();
std::cout << cfg.iterations << ", " << ticks << std::endl;
}
} // namespace anonymous
CAF_MAIN(io::middleman);
#include "caf/io/network/newb.hpp"
#include "caf/logger.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/detail/call_cfun.hpp"
#include "caf/policy/newb_basp.hpp"
#include "caf/policy/newb_tcp.hpp"
using namespace caf;
using caf::io::network::default_multiplexer;
using caf::io::network::invalid_native_socket;
using caf::io::network::make_client_newb;
using caf::io::network::make_server_newb;
using caf::io::network::native_socket;
using caf::policy::accept_tcp;
using caf::policy::tcp_protocol;
using caf::policy::tcp_transport;
namespace {
using interval_atom = atom_constant<atom("interval")>;
using ordering_atom = atom_constant<atom("ordering")>;
using send_atom = atom_constant<atom("send")>;
using quit_atom = atom_constant<atom("quit")>;
using responder_atom = atom_constant<atom("responder")>;
constexpr size_t chunk_size = 1024; //128; //8192; //1024;
struct basp_newb : public io::network::newb<policy::new_basp_message> {
using message_type = policy::new_basp_message;
basp_newb(caf::actor_config& cfg, default_multiplexer& dm,
native_socket sockfd)
: newb<message_type>(cfg, dm, sockfd),
running(true),
interval_counter(0),
received_messages(0),
interval(5000) {
// nop
CAF_LOG_TRACE("");
}
void handle(message_type& msg) override {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE("");
if (msg.payload_len == 1) {
// nop
} else {
received_messages += 1;
if (received_messages % 1000 == 0)
std::cout << "received " << received_messages << " messages" << std::endl;
// nop
}
}
behavior make_behavior() override {
set_default_handler(print_and_drop);
return {
// Must be implemented at the moment, will be cought by the broker in a
// later implementation.
[=](atom_value atm, uint32_t id) {
protocol->timeout(atm, id);
},
[=](send_atom, char c) {
if (running) {
delayed_send(this, interval, send_atom::value, char((c + 1) % 256));
auto hw = caf::make_callback([&](io::network::byte_buffer& buf) -> error {
binary_serializer bs(&backend(), buf);
bs(policy::basp_header{0, id(), actor_id{}});
return none;
});
auto whdl = wr_buf(&hw);
CAF_ASSERT(whdl.buf != nullptr);
CAF_ASSERT(whdl.protocol != nullptr);
binary_serializer bs(&backend(), *whdl.buf);
auto start = whdl.buf->size();
whdl.buf->resize(start + chunk_size);
std::fill(whdl.buf->begin() + start, whdl.buf->end(), c);
}
},
[=](responder_atom, actor r) {
std::cout << "got responder assigned" << std::endl;
responder = r;
send(r, this);
},
[=](interval_atom) {
if (running) {
delayed_send(this, std::chrono::seconds(1), interval_atom::value);
interval_counter += 1;
data.emplace_back(interval,
transport->count,
transport->offline_buffer.size());
if (interval_counter % 10 == 0) {
auto cnt = interval.count();
auto dec = cnt > 1000 ? 1000 : (cnt > 100 ? 100 : 10);
interval -= std::chrono::microseconds(dec);
}
transport->count = 0;
if (interval.count() <= 0)
running = false;
} else {
std::map<size_t, std::vector<size_t>> aggregate;
for (auto& t : data) {
auto expected = (1000000 / get<0>(t).count());
aggregate[expected].push_back(get<1>(t));
std::cerr << expected << ", " << get<1>(t) << ", " << get<2>(t) << std::endl;
}
for (auto& p : aggregate) {
std::cerr << p.first;
for (auto v : p.second)
std::cerr << ", " << v;
std::cerr << std::endl;
}
send(this, quit_atom::value);
}
},
[=](quit_atom) {
std::cout << "got quit message" << std::endl;
// Remove from multiplexer loop.
stop();
// Quit actor.
quit();
send(responder, quit_atom::value);
}
};
}
bool running;
actor responder;
uint32_t interval_counter;
uint32_t received_messages;
std::chrono::microseconds interval;
// values: measurement point, current interval, messages sent in interval, offline buffer size
std::vector<std::tuple<std::chrono::microseconds, size_t, size_t>> data;
};
template <class ProtocolPolicy>
struct tcp_acceptor
: public io::network::newb_acceptor<typename ProtocolPolicy::message_type> {
using super = io::network::newb_acceptor<typename ProtocolPolicy::message_type>;
tcp_acceptor(default_multiplexer& dm, native_socket sockfd)
: super(dm, sockfd) {
// nop
}
expected<actor> create_newb(native_socket sockfd,
io::network::transport_policy_ptr pol) override {
CAF_LOG_TRACE(CAF_ARG(sockfd));
std::cout << "creating newb" << std::endl;
auto n = io::network::make_newb<basp_newb>(this->backend().system(), sockfd);
auto ptr = caf::actor_cast<caf::abstract_actor*>(n);
if (ptr == nullptr)
return sec::runtime_error;
auto& ref = dynamic_cast<basp_newb&>(*ptr);
ref.transport = std::move(pol);
ref.protocol.reset(new ProtocolPolicy(&ref));
ref.responder = responder;
ref.configure_read(io::receive_policy::exactly(policy::basp_header_len));
anon_send(responder, n);
return n;
}
actor responder;
};
class config : public actor_system_config {
public:
uint16_t port = 12345;
std::string host = "127.0.0.1";
bool is_server = false;
config() {
opt_group{custom_options_, "global"}
.add(port, "port,P", "set port")
.add(host, "host,H", "set host")
.add(is_server, "server,s", "set server");
}
};
void caf_main(actor_system& sys, const config& cfg) {
using acceptor_t = tcp_acceptor<tcp_protocol<policy::stream_basp>>;
const char* host = cfg.host.c_str();
const uint16_t port = cfg.port;
scoped_actor self{sys};
auto running = [=](event_based_actor* self, std::string,
actor m, actor) -> behavior {
return {
[=](quit_atom) {
self->send(m, quit_atom::value);
}
};
};
auto init = [=](event_based_actor* self, std::string name,
actor m) -> behavior {
self->set_default_handler(skip);
return {
[=](actor b) {
std::cout << "[" << name << "] got broker, let's do this" << std::endl;
self->become(running(self, name, m, b));
self->set_default_handler(print_and_drop);
}
};
};
auto dummy_broker = [](io::broker*) -> behavior {
return {
[](io::new_connection_msg&) {
std::cout << "got new connection" << std::endl;
}
};
};
auto name = cfg.is_server ? "server" : "client";
auto helper = sys.spawn(init, name, self);
actor nb;
auto await_done = [&]() {
self->receive(
[&](quit_atom) {
std::cout << "done" << std::endl;
}
);
};
if (cfg.is_server) {
std::cout << "creating new server" << std::endl;
auto server_ptr = make_server_newb<acceptor_t, accept_tcp>(sys, port, nullptr,
true);
//server_ptr->responder = helper;
//std::cout << "creating new client" << std::endl;
//auto client = make_client_newb<basp_newb, tcp_transport,
//tcp_protocol<raw_tcp>>(sys, host, port);
// If I don't do this, our newb acceptor will never get events ...
auto b = sys.middleman().spawn_server(dummy_broker, port + 1);
await_done();
} else {
std::cout << "creating new client" << std::endl;
auto client = make_client_newb<basp_newb, tcp_transport,
tcp_protocol<policy::stream_basp>>(sys, host, port);
self->send(client, responder_atom::value, helper);
self->send(client, send_atom::value, char(0));
self->send(client, interval_atom::value);
await_done();
}
}
} // namespace anonymous
CAF_MAIN(io::middleman);
#include "caf/io/network/newb.hpp"
#include "caf/logger.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/detail/call_cfun.hpp"
#include "caf/policy/newb_udp.hpp"
#include "caf/policy/newb_basp.hpp"
#include "caf/policy/newb_ordering.hpp"
using namespace caf;
using namespace caf::policy;
using caf::io::network::default_multiplexer;
using caf::io::network::invalid_native_socket;
using caf::io::network::make_client_newb;
using caf::io::network::make_server_newb;
using caf::io::network::native_socket;
namespace {
using interval_atom = atom_constant<atom("interval")>;
using ordering_atom = atom_constant<atom("ordering")>;
using send_atom = atom_constant<atom("send")>;
using quit_atom = atom_constant<atom("quit")>;
using responder_atom = atom_constant<atom("responder")>;
using start_atom = atom_constant<atom("start")>;
using handshake_atom = atom_constant<atom("handshake")>;
constexpr size_t chunk_size = 8192; //8192; //128; //1024;
struct raw_newb : public io::network::newb<new_basp_message> {
using message_type = new_basp_message;
raw_newb(caf::actor_config& cfg, default_multiplexer& dm,
native_socket sockfd)
: newb<message_type>(cfg, dm, sockfd),
running(true),
is_client(true),
interval_counter(0),
received_messages(0),
interval(5000) {
// nop
CAF_LOG_TRACE("");
}
void handle(message_type& msg) override {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE("");
if (is_client) {
send(responder, handshake_atom::value);
} else if (msg.payload_len == 1) {
auto byte = *msg.payload;
if (byte == 'h')
std::cout << "I'll consider this the handshake" << std::endl;
else if (byte == 'q')
send(this, quit_atom::value);
send(this, handshake_atom::value);
} else {
if (msg.payload_len != chunk_size)
std::cout << "Hmmm, payload is " << msg.payload_len << " and not "
<< chunk_size << std::endl;
received_messages += 1;
if (received_messages % 1000 == 0)
std::cout << "received " << received_messages << " messages" << std::endl;
//std::cout << "received message" << std::endl;
// nop
}
}
behavior make_behavior() override {
set_default_handler(print_and_drop);
return {
// Must be implemented at the moment, will be cought by the broker in a
// later implementation.
[=](atom_value atm, uint32_t id) {
protocol->timeout(atm, id);
},
[=](handshake_atom) {
auto hw = caf::make_callback([&](io::network::byte_buffer& buf) -> error {
binary_serializer bs(&backend(), buf);
bs(basp_header{0, id(), actor_id{}});
return none;
});
auto whdl = wr_buf(&hw);
CAF_ASSERT(whdl.buf != nullptr);
CAF_ASSERT(whdl.protocol != nullptr);
binary_serializer bs(&backend(), *whdl.buf);
whdl.buf->push_back('h');
},
[=](send_atom, char c) {
if (running) {
delayed_send(this, interval, send_atom::value, char((c + 1) % 256));
auto hw = caf::make_callback([&](io::network::byte_buffer& buf) -> error {
binary_serializer bs(&backend(), buf);
bs(basp_header{0, id(), actor_id{}});
return none;
});
auto whdl = wr_buf(&hw);
CAF_ASSERT(whdl.buf != nullptr);
CAF_ASSERT(whdl.protocol != nullptr);
binary_serializer bs(&backend(), *whdl.buf);
auto start = whdl.buf->size();
whdl.buf->resize(start + chunk_size);
std::fill(whdl.buf->begin() + start, whdl.buf->end(), c);
}
},
[=](responder_atom, actor r) {
std::cout << "got responder assigned" << std::endl;
responder = r;
send(r, this);
},
[=](interval_atom) {
if (running) {
delayed_send(this, std::chrono::seconds(1), interval_atom::value);
data.emplace_back(interval,
transport->count,
transport->offline_buffer.size());
interval_counter += 1;
if (interval_counter % 10 == 0) {
auto cnt = interval.count();
auto dec = cnt > 1000 ? 1000 : (cnt > 100 ? 100 : 10);
interval -= std::chrono::microseconds(dec);
}
transport->count = 0;
if (interval.count() <= 0)
running = false;
} else {
std::map<size_t, std::vector<size_t>> aggregate;
for (auto& t : data) {
auto expected = (1000000 / get<0>(t).count());
aggregate[expected].push_back(get<1>(t));
}
for (auto& p : aggregate) {
std::cerr << p.first;
for (auto v : p.second)
std::cerr << ", " << v;
std::cerr << std::endl;
}
/*
for (auto& t : data)
std::cerr << (1000000 / get<0>(t).count()) << ", "
<< get<1>(t) << ", " << get<2>(t) << std::endl;
*/
send(this, quit_atom::value);
}
},
[=](quit_atom) {
std::cout << "got quit message" << std::endl;
// Remove from multiplexer loop.
stop();
// Quit actor.
quit();
send(responder, quit_atom::value);
}
};
}
bool running;
bool is_client;
actor responder;
uint32_t interval_counter;
uint32_t received_messages;
std::chrono::microseconds interval;
// values: measurement point, current interval, messages sent in interval, offline buffer size
std::vector<std::tuple<std::chrono::microseconds, size_t, size_t>> data;
};
template <class ProtocolPolicy>
struct udp_acceptor
: public io::network::newb_acceptor<typename ProtocolPolicy::message_type> {
using super = io::network::newb_acceptor<typename ProtocolPolicy::message_type>;
udp_acceptor(default_multiplexer& dm, native_socket sockfd)
: super(dm, sockfd) {
// nop
}
~udp_acceptor() {
std::cout << "terminating udp acceptor" << std::endl;
}
expected<actor> create_newb(native_socket sockfd,
io::network::transport_policy_ptr pol) override {
CAF_LOG_TRACE(CAF_ARG(sockfd));
std::cout << "creating newb" << std::endl;
auto n = io::network::make_newb<raw_newb>(this->backend().system(), sockfd);
auto ptr = caf::actor_cast<caf::abstract_actor*>(n);
if (ptr == nullptr)
return sec::runtime_error;
auto& ref = dynamic_cast<raw_newb&>(*ptr);
ref.transport = std::move(pol);
ref.protocol.reset(new ProtocolPolicy(&ref));
ref.responder = responder;
// Read first message from this socket
ref.is_client = false;
ref.transport->prepare_next_read(this);
ref.transport->read_some(this, *ref.protocol.get());
// TODO: Just a workaround.
anon_send(responder, n);
return n;
}
actor responder;
};
class config : public actor_system_config {
public:
uint16_t port = 12345;
std::string host = "127.0.0.1";
bool is_server = false;
config() {
opt_group{custom_options_, "global"}
.add(port, "port,P", "set port")
.add(host, "host,H", "set host")
.add(is_server, "server,s", "set server");
}
};
void caf_main(actor_system& sys, const config& cfg) {
using policy_t = udp_protocol<ordering<datagram_basp>>;
using acceptor_t = udp_acceptor<policy_t>;
const char* host = cfg.host.c_str();
const uint16_t port = cfg.port;
scoped_actor self{sys};
auto running = [=](event_based_actor* self, std::string name,
actor m, actor) -> behavior {
return {
[=](handshake_atom) {
std::cout << "[" << name << "] got server" << std::endl;
self->send(m, quit_atom::value);
},
[=](quit_atom) {
self->send(m, quit_atom::value);
}
};
};
auto init = [=](event_based_actor* self, std::string name,
actor m) -> behavior {
self->set_default_handler(skip);
return {
[=](actor b) {
std::cout << "[" << name << "] got broker, let's do this" << std::endl;
self->become(running(self, name, m, b));
self->set_default_handler(print_and_drop);
}
};
};
auto dummy_broker = [](io::broker*) -> behavior {
return {
[](io::new_connection_msg&) {
std::cout << "got new connection" << std::endl;
}
};
};
auto name = cfg.is_server ? "server" : "client";
auto helper = sys.spawn(init, name, self);
actor nb;
auto await_done = [&](std::string msg) {
self->receive(
[&](quit_atom) {
std::cout << msg << std::endl;
}
);
};
if (cfg.is_server) {
std::cout << "creating new server" << std::endl;
auto server_ptr = make_server_newb<acceptor_t, accept_udp>(sys, port, nullptr,
true);
// If I don't do this, our newb acceptor will never get events ...
auto b = sys.middleman().spawn_server(dummy_broker, port + 1);
await_done("done");
} else {
std::cout << "creating new client" << std::endl;
auto client = make_client_newb<raw_newb, udp_transport, policy_t>(sys, host,
port);
self->send(client, responder_atom::value, helper);
self->send(client, handshake_atom::value);
await_done("let's start");
self->send(client, send_atom::value, char(0));
self->send(client, interval_atom::value);
await_done("done");
std::abort();
}
}
} // namespace anonymous
CAF_MAIN(io::middleman);
#include "caf/io/network/newb.hpp"
#include "caf/logger.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/detail/call_cfun.hpp"
#include "caf/policy/newb_raw.hpp"
#include "caf/policy/newb_tcp.hpp"
using namespace caf;
using caf::io::network::default_multiplexer;
using caf::io::network::invalid_native_socket;
using caf::io::network::make_client_newb;
using caf::io::network::make_server_newb;
using caf::io::network::native_socket;
using caf::policy::accept_tcp;
using caf::policy::tcp_protocol;
using caf::policy::tcp_transport;
namespace {
using interval_atom = atom_constant<atom("interval")>;
using ordering_atom = atom_constant<atom("ordering")>;
using send_atom = atom_constant<atom("send")>;
using quit_atom = atom_constant<atom("quit")>;
using responder_atom = atom_constant<atom("responder")>;
constexpr size_t chunk_size = 8192; //128; //8192; //1024;
struct raw_newb : public io::network::newb<policy::raw_data_message> {
using message_type = policy::raw_data_message;
raw_newb(caf::actor_config& cfg, default_multiplexer& dm,
native_socket sockfd)
: newb<message_type>(cfg, dm, sockfd),
running(true),
interval_counter(0),
received_messages(0),
interval(5000) {
// nop
CAF_LOG_TRACE("");
}
void handle(message_type& msg) override {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE("");
if (msg.payload_len == 1) {
// nop
} else {
received_messages += 1;
if (received_messages % 1000 == 0)
std::cout << "received " << received_messages << " messages" << std::endl;
//std::cout << "received message" << std::endl;
// nop
}
}
behavior make_behavior() override {
set_default_handler(print_and_drop);
return {
// Must be implemented at the moment, will be cought by the broker in a
// later implementation.
[=](atom_value atm, uint32_t id) {
protocol->timeout(atm, id);
},
[=](send_atom, char c) {
if (running) {
delayed_send(this, interval, send_atom::value, char((c + 1) % 256));
auto whdl = wr_buf(nullptr);
CAF_ASSERT(whdl.buf != nullptr);
CAF_ASSERT(whdl.protocol != nullptr);
binary_serializer bs(&backend(), *whdl.buf);
whdl.buf->resize(chunk_size);
std::fill(whdl.buf->begin(), whdl.buf->end(), c);
}
},
[=](responder_atom, actor r) {
std::cout << "got responder assigned" << std::endl;
responder = r;
send(r, this);
},
[=](interval_atom) {
if (running) {
delayed_send(this, std::chrono::seconds(1), interval_atom::value);
data.emplace_back(interval,
transport->count,
transport->offline_buffer.size());
interval_counter += 1;
if (interval_counter % 10 == 0) {
auto cnt = interval.count();
auto dec = cnt > 1000 ? 1000 : (cnt > 100 ? 100 : 10);
interval -= std::chrono::microseconds(dec);
}
transport->count = 0;
if (interval.count() <= 0)
running = false;
} else {
std::map<size_t, std::vector<size_t>> aggregate;
for (auto& t : data) {
auto expected = (1000000 / get<0>(t).count());
aggregate[expected].push_back(get<1>(t));
}
for (auto& p : aggregate) {
std::cerr << p.first;
for (auto v : p.second)
std::cerr << ", " << v;
std::cerr << std::endl;
}
send(this, quit_atom::value);
}
},
[=](quit_atom) {
std::cout << "got quit message" << std::endl;
// Remove from multiplexer loop.
stop();
// Quit actor.
quit();
send(responder, quit_atom::value);
}
};
}
bool running;
actor responder;
uint32_t interval_counter;
uint32_t received_messages;
std::chrono::microseconds interval;
// values: measurement point, current interval, messages sent in interval, offline buffer size
std::vector<std::tuple<std::chrono::microseconds, size_t, size_t>> data;
};
template <class ProtocolPolicy>
struct tcp_acceptor
: public io::network::newb_acceptor<typename ProtocolPolicy::message_type> {
using super = io::network::newb_acceptor<typename ProtocolPolicy::message_type>;
tcp_acceptor(default_multiplexer& dm, native_socket sockfd)
: super(dm, sockfd) {
// nop
}
expected<actor> create_newb(native_socket sockfd,
io::network::transport_policy_ptr pol) override {
CAF_LOG_TRACE(CAF_ARG(sockfd));
std::cout << "tcp_acceptor::creating newb" << std::endl;
auto n = io::network::make_newb<raw_newb>(this->backend().system(), sockfd);
auto ptr = caf::actor_cast<caf::abstract_actor*>(n);
if (ptr == nullptr)
return sec::runtime_error;
auto& ref = dynamic_cast<raw_newb&>(*ptr);
ref.transport = std::move(pol);
ref.protocol.reset(new ProtocolPolicy(&ref));
ref.responder = responder;
ref.configure_read(io::receive_policy::exactly(chunk_size));
anon_send(responder, n);
return n;
}
actor responder;
};
class config : public actor_system_config {
public:
uint16_t port = 12345;
std::string host = "127.0.0.1";
bool is_server = false;
config() {
opt_group{custom_options_, "global"}
.add(port, "port,P", "set port")
.add(host, "host,H", "set host")
.add(is_server, "server,s", "set server");
}
};
struct state {
size_t count = 0;
};
void caf_main(actor_system& sys, const config& cfg) {
using acceptor_t = tcp_acceptor<tcp_protocol<policy::raw>>;
const char* host = cfg.host.c_str();
const uint16_t port = cfg.port;
scoped_actor self{sys};
auto running = [=](event_based_actor* self, std::string,
actor m, actor) -> behavior {
return {
[=](quit_atom) {
self->send(m, quit_atom::value);
}
};
};
auto init = [=](event_based_actor* self, std::string name,
actor m) -> behavior {
self->set_default_handler(skip);
return {
[=](actor b) {
std::cout << "[" << name << "] got broker, let's do this" << std::endl;
self->become(running(self, name, m, b));
self->set_default_handler(print_and_drop);
}
};
};
auto dummy_broker = [](io::stateful_broker<state>* self) -> behavior {
return {
[=](io::new_connection_msg& msg) {
std::cout << "got new connection" << std::endl;
self->configure_read(msg.handle, io::receive_policy::exactly(chunk_size));
},
[=](io::new_data_msg&) {
self->state.count += 1;
if (self->state.count % 1000 == 0)
std::cout << "received " << self->state.count << " messages" << std::endl;
}
};
};
auto name = cfg.is_server ? "server" : "client";
auto helper = sys.spawn(init, name, self);
actor nb;
auto await_done = [&]() {
self->receive(
[&](quit_atom) {
std::cout << "done" << std::endl;
}
);
};
if (cfg.is_server) {
std::cout << "creating new server" << std::endl;
auto server_ptr = make_server_newb<acceptor_t, accept_tcp>(sys, port, nullptr,
true);
// If I don't do this, our newb acceptor will never get events ...
auto b = sys.middleman().spawn_server(dummy_broker, port + 1);
await_done();
} else {
std::cout << "creating new client" << std::endl;
auto client = make_client_newb<raw_newb, tcp_transport,
tcp_protocol<policy::raw>>(sys, host, port);
self->send(client, responder_atom::value, helper);
self->send(client, send_atom::value, char(0));
self->send(client, interval_atom::value);
await_done();
}
}
} // namespace anonymous
CAF_MAIN(io::middleman);
#include "caf/io/network/newb.hpp"
#include "caf/logger.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/detail/call_cfun.hpp"
#include "caf/policy/newb_raw.hpp"
#include "caf/policy/newb_udp.hpp"
using namespace caf;
using caf::io::network::default_multiplexer;
using caf::io::network::invalid_native_socket;
using caf::io::network::make_client_newb;
using caf::io::network::make_server_newb;
using caf::io::network::native_socket;
using caf::policy::accept_udp;
using caf::policy::udp_protocol;
using caf::policy::udp_transport;
namespace {
using interval_atom = atom_constant<atom("interval")>;
using ordering_atom = atom_constant<atom("ordering")>;
using send_atom = atom_constant<atom("send")>;
using quit_atom = atom_constant<atom("quit")>;
using responder_atom = atom_constant<atom("responder")>;
using start_atom = atom_constant<atom("start")>;
using handshake_atom = atom_constant<atom("handshake")>;
constexpr size_t chunk_size = 8192; //128; //1024;
struct raw_newb : public io::network::newb<policy::raw_data_message> {
using message_type = policy::raw_data_message;
raw_newb(caf::actor_config& cfg, default_multiplexer& dm,
native_socket sockfd)
: newb<message_type>(cfg, dm, sockfd),
running(true),
is_client(true),
interval_counter(0),
received_messages(0),
interval(5000) {
// nop
CAF_LOG_TRACE("");
}
void handle(message_type& msg) override {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE("");
if (is_client) {
send(responder, handshake_atom::value);
} else if (msg.payload_len == 1) {
auto byte = *msg.payload;
if (byte == 'h')
std::cout << "I'll consider this the handshake" << std::endl;
else if (byte == 'q')
send(this, quit_atom::value);
send(this, handshake_atom::value);
} else {
received_messages += 1;
if (received_messages % 1000 == 0)
std::cout << "received " << received_messages << " messages" << std::endl;
//std::cout << "received message" << std::endl;
// nop
}
}
behavior make_behavior() override {
set_default_handler(print_and_drop);
return {
// Must be implemented at the moment, will be cought by the broker in a
// later implementation.
[=](atom_value atm, uint32_t id) {
protocol->timeout(atm, id);
},
[=](handshake_atom) {
auto whdl = wr_buf(nullptr);
CAF_ASSERT(whdl.buf != nullptr);
CAF_ASSERT(whdl.protocol != nullptr);
binary_serializer bs(&backend(), *whdl.buf);
whdl.buf->push_back('h');
},
[=](send_atom, char c) {
if (running) {
delayed_send(this, interval, send_atom::value, char((c + 1) % 256));
auto whdl = wr_buf(nullptr);
CAF_ASSERT(whdl.buf != nullptr);
CAF_ASSERT(whdl.protocol != nullptr);
binary_serializer bs(&backend(), *whdl.buf);
whdl.buf->resize(chunk_size);
std::fill(whdl.buf->begin(), whdl.buf->end(), c);
}
},
[=](responder_atom, actor r) {
std::cout << "got responder assigned" << std::endl;
responder = r;
send(r, this);
},
[=](interval_atom) {
if (running) {
delayed_send(this, std::chrono::seconds(1), interval_atom::value);
data.emplace_back(interval,
transport->count,
transport->offline_buffer.size());
interval_counter += 1;
if (interval_counter % 10 == 0) {
auto cnt = interval.count();
auto dec = cnt > 1000 ? 1000 : (cnt > 100 ? 100 : 10);
interval -= std::chrono::microseconds(dec);
}
transport->count = 0;
if (interval.count() <= 0)
running = false;
} else {
std::map<size_t, std::vector<size_t>> aggregate;
for (auto& t : data) {
auto expected = (1000000 / get<0>(t).count());
aggregate[expected].push_back(get<1>(t));
}
for (auto& p : aggregate) {
std::cerr << p.first;
for (auto v : p.second)
std::cerr << ", " << v;
std::cerr << std::endl;
}
/*
for (auto& t : data)
std::cerr << (1000000 / get<0>(t).count()) << ", "
<< get<1>(t) << ", " << get<2>(t) << std::endl;
*/
send(this, quit_atom::value);
}
},
[=](quit_atom) {
std::cout << "got quit message" << std::endl;
// Remove from multiplexer loop.
stop();
// Quit actor.
quit();
send(responder, quit_atom::value);
}
};
}
bool running;
bool is_client;
actor responder;
uint32_t interval_counter;
uint32_t received_messages;
std::chrono::microseconds interval;
// values: measurement point, current interval, messages sent in interval, offline buffer size
std::vector<std::tuple<std::chrono::microseconds, size_t, size_t>> data;
};
template <class ProtocolPolicy>
struct udp_acceptor
: public io::network::newb_acceptor<typename ProtocolPolicy::message_type> {
using super = io::network::newb_acceptor<typename ProtocolPolicy::message_type>;
udp_acceptor(default_multiplexer& dm, native_socket sockfd)
: super(dm, sockfd) {
// nop
}
~udp_acceptor() {
std::cout << "terminating udp acceptor" << std::endl;
}
expected<actor> create_newb(native_socket sockfd,
io::network::transport_policy_ptr pol) override {
CAF_LOG_TRACE(CAF_ARG(sockfd));
std::cout << "creating newb" << std::endl;
auto n = io::network::make_newb<raw_newb>(this->backend().system(), sockfd);
auto ptr = caf::actor_cast<caf::abstract_actor*>(n);
if (ptr == nullptr)
return sec::runtime_error;
auto& ref = dynamic_cast<raw_newb&>(*ptr);
ref.transport = std::move(pol);
ref.protocol.reset(new ProtocolPolicy(&ref));
ref.responder = responder;
// Read first message from this socket
ref.is_client = false;
ref.transport->prepare_next_read(this);
ref.transport->read_some(this, *ref.protocol.get());
// TODO: Just a workaround.
anon_send(responder, n);
return n;
}
actor responder;
};
class config : public actor_system_config {
public:
uint16_t port = 12345;
std::string host = "127.0.0.1";
bool is_server = false;
config() {
opt_group{custom_options_, "global"}
.add(port, "port,P", "set port")
.add(host, "host,H", "set host")
.add(is_server, "server,s", "set server");
}
};
void caf_main(actor_system& sys, const config& cfg) {
using acceptor_t = udp_acceptor<udp_protocol<policy::raw>>;
const char* host = cfg.host.c_str();
const uint16_t port = cfg.port;
scoped_actor self{sys};
auto running = [=](event_based_actor* self, std::string name,
actor m, actor) -> behavior {
return {
[=](handshake_atom) {
std::cout << "[" << name << "] got server" << std::endl;
self->send(m, quit_atom::value);
},
[=](quit_atom) {
self->send(m, quit_atom::value);
}
};
};
auto init = [=](event_based_actor* self, std::string name,
actor m) -> behavior {
self->set_default_handler(skip);
return {
[=](actor b) {
std::cout << "[" << name << "] got broker, let's do this" << std::endl;
self->become(running(self, name, m, b));
self->set_default_handler(print_and_drop);
}
};
};
auto dummy_broker = [](io::broker*) -> behavior {
return {
[](io::new_connection_msg&) {
std::cout << "got new connection" << std::endl;
}
};
};
auto name = cfg.is_server ? "server" : "client";
auto helper = sys.spawn(init, name, self);
actor nb;
auto await_done = [&](std::string msg) {
self->receive(
[&](quit_atom) {
std::cout << msg << std::endl;
}
);
};
if (cfg.is_server) {
std::cout << "creating new server" << std::endl;
auto server_ptr = make_server_newb<acceptor_t, accept_udp>(sys, port, nullptr,
true);
// If I don't do this, our newb acceptor will never get events ...
auto b = sys.middleman().spawn_server(dummy_broker, port + 1);
await_done("done");
} else {
std::cout << "creating new client" << std::endl;
auto client = make_client_newb<raw_newb, udp_transport,
udp_protocol<policy::raw>>(sys, host, port);
self->send(client, responder_atom::value, helper);
self->send(client, handshake_atom::value);
await_done("let's start");
self->send(client, send_atom::value, char(0));
self->send(client, interval_atom::value);
await_done("done");
std::abort();
}
}
} // namespace anonymous
CAF_MAIN(io::middleman);
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