Commit 307cb3eb authored by Dominik Charousset's avatar Dominik Charousset

Perform multiple reads per I/O event, relates #432

parent 8839cd9c
......@@ -121,6 +121,7 @@ public:
// Config parameters of middleman.
atom_value middleman_network_backend;
bool middleman_enable_automatic_connections;
size_t middleman_max_consecutive_reads;
// System parameters that are set while initializing modules.
node_id network_id;
......
......@@ -141,6 +141,7 @@ actor_system_config::actor_system_config() {
scheduler_profiling_ms_resolution = 100;
middleman_network_backend = atom("default");
middleman_enable_automatic_connections = false;
middleman_max_consecutive_reads = 50;
nexus_port = 0;
}
......@@ -169,6 +170,8 @@ actor_system_config::actor_system_config(int argc, char** argv)
.bind("middleman.network-backend", middleman_network_backend)
.bind("middleman.enable-automatic-connections",
middleman_enable_automatic_connections)
.bind("middleman.max_consecutive_reads",
middleman_max_consecutive_reads)
.bind("probe.nexus-host", nexus_host)
.bind("probe.nexus-port", nexus_port);
detail::parse_ini(ini, consumer, std::cerr);
......@@ -200,6 +203,9 @@ actor_system_config::actor_system_config(int argc, char** argv)
{"caf#middleman.enable-automatic-connections",
"enables or disables automatic connection management (off per default)",
middleman_enable_automatic_connections},
{"caf#middleman.max_consecutive_reads",
"sets the maximum number of allowed I/O reads before scheduling others",
middleman_max_consecutive_reads},
{"caf#probe.nexus-host",
"sets the hostname or IP address for connecting to the Nexus",
nexus_host},
......
......@@ -425,9 +425,9 @@ public:
void handle_event(operation op) override;
private:
void read_loop();
void prepare_next_read();
void write_loop();
void prepare_next_write();
// state for reading
manager_ptr reader_;
......
......@@ -174,12 +174,21 @@ public:
max_throughput_ = x;
}
inline size_t max_consecutive_reads() const {
return max_consecutive_reads_;
}
inline void max_consecutive_reads(size_t x) {
max_consecutive_reads_ = x;
}
protected:
/// Identifies the thread this multiplexer
/// is running in. Must be set by the subclass.
std::thread::id tid_;
size_t max_throughput_;
size_t max_consecutive_reads_;
};
using multiplexer_ptr = std::unique_ptr<multiplexer>;
......
......@@ -1016,7 +1016,7 @@ void stream::start(const manager_ptr& mgr) {
CAF_ASSERT(mgr != nullptr);
reader_ = mgr;
backend().add(operation::read, fd(), this);
read_loop();
prepare_next_read();
}
void stream::configure_read(receive_policy::config config) {
......@@ -1042,7 +1042,7 @@ void stream::flush(const manager_ptr& mgr) {
backend().add(operation::write, fd(), this);
writer_ = mgr;
writing_ = true;
write_loop();
prepare_next_write();
}
}
......@@ -1064,17 +1064,23 @@ void stream::handle_event(operation op) {
CAF_LOG_TRACE(CAF_ARG(op));
switch (op) {
case operation::read: {
size_t rb; // read bytes
if (! read_some(rb, fd(),
rd_buf_.data() + collected_,
rd_buf_.size() - collected_)) {
reader_->io_failure(&backend(), operation::read);
backend().del(operation::read, fd(), this);
} else if (rb > 0) {
// loop until an error occurs or we have nothing more to read
// or until we have handled 50 reads
size_t rb;
for (size_t i = 0; i < backend().max_consecutive_reads(); ++i) {
if (! read_some(rb, fd(),
rd_buf_.data() + collected_,
rd_buf_.size() - collected_)) {
reader_->io_failure(&backend(), operation::read);
backend().del(operation::read, fd(), this);
return;
}
if (rb == 0)
return;
collected_ += rb;
if (collected_ >= read_threshold_) {
reader_->consume(&backend(), rd_buf_.data(), collected_);
read_loop();
prepare_next_read();
}
}
break;
......@@ -1095,7 +1101,7 @@ void stream::handle_event(operation op) {
remaining + wr_offline_buf_.size());
// prepare next send (or stop sending)
if (remaining == 0)
write_loop();
prepare_next_write();
}
break;
}
......@@ -1110,7 +1116,7 @@ void stream::handle_event(operation op) {
}
}
void stream::read_loop() {
void stream::prepare_next_read() {
collected_ = 0;
switch (rd_flag_) {
case receive_policy_flag::exactly:
......@@ -1134,7 +1140,7 @@ void stream::read_loop() {
}
}
void stream::write_loop() {
void stream::prepare_next_write() {
CAF_LOG_TRACE(CAF_ARG(wr_buf_.size()) << CAF_ARG(wr_offline_buf_.size()));
written_ = 0;
wr_buf_.clear();
......
......@@ -342,6 +342,7 @@ void middleman::init(actor_system_config& cfg) {
cfg.network_id.swap(this_node);
// set scheduling parameters for multiplexer
backend().max_throughput(cfg.scheduler_max_throughput);
backend().max_consecutive_reads(cfg.middleman_max_consecutive_reads);
}
actor_system::module::id_t middleman::id() const {
......
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