Commit 684109af authored by Dominik Charousset's avatar Dominik Charousset

Properly initialize max_throughput with > 0

The scheduler no longer accepts 0 as "magic number" for infinity. Instead,
numeric_limits<size_t>::max() is used as default and 0 is no longer accepted.
parent 9c06c504
...@@ -59,6 +59,7 @@ class event_based_resume { ...@@ -59,6 +59,7 @@ class event_based_resume {
resumable::resume_result resume(execution_unit* new_host, resumable::resume_result resume(execution_unit* new_host,
size_t max_throughput) override { size_t max_throughput) override {
CAF_REQUIRE(max_throughput > 0);
auto d = static_cast<Derived*>(this); auto d = static_cast<Derived*>(this);
d->host(new_host); d->host(new_host);
CAF_LOG_TRACE("id = " << d->id()); CAF_LOG_TRACE("id = " << d->id());
...@@ -105,11 +106,7 @@ class event_based_resume { ...@@ -105,11 +106,7 @@ class event_based_resume {
} }
} }
// max_throughput = 0 means infinite // max_throughput = 0 means infinite
size_t increment = max_throughput == 0 ? 0 : 1; for (size_t i = 0; i < max_throughput; ++i) {
if (max_throughput == 0) {
max_throughput = 1;
}
for (size_t i = 0; i < max_throughput; i += increment) {
auto ptr = d->next_message(); auto ptr = d->next_message();
if (ptr) { if (ptr) {
if (d->invoke_message(ptr)) { if (d->invoke_message(ptr)) {
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include <mutex> #include <mutex>
#include <thread> #include <thread>
#include <chrono> #include <chrono>
#include <limits>
#include <condition_variable> #include <condition_variable>
#include "caf/duration.hpp" #include "caf/duration.hpp"
...@@ -73,7 +74,8 @@ class no_scheduling { ...@@ -73,7 +74,8 @@ class no_scheduling {
std::thread([=] { std::thread([=] {
CAF_PUSH_AID(mself->id()); CAF_PUSH_AID(mself->id());
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
while (mself->resume(nullptr, 0) != resumable::done) { auto max_throughput = std::numeric_limits<size_t>::max();
while (mself->resume(nullptr, max_throughput) != resumable::done) {
// await new data before resuming actor // await new data before resuming actor
await_data(mself.get()); await_data(mself.get());
CAF_REQUIRE(self->mailbox().blocked() == false); CAF_REQUIRE(self->mailbox().blocked() == false);
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include <chrono> #include <chrono>
#include <memory> #include <memory>
#include <thread> #include <thread>
#include <limits>
#include <cstdint> #include <cstdint>
#include <functional> #include <functional>
#include <type_traits> #include <type_traits>
...@@ -280,7 +281,10 @@ class coordinator : public abstract_coordinator { ...@@ -280,7 +281,10 @@ class coordinator : public abstract_coordinator {
using policy_data = typename Policy::coordinator_data; using policy_data = typename Policy::coordinator_data;
coordinator(size_t nw = std::thread::hardware_concurrency()) : super(nw) { coordinator(size_t nw = std::thread::hardware_concurrency(),
size_t mt = std::numeric_limits<size_t>::max())
: super(nw),
m_max_throughput(mt) {
// nop // nop
} }
...@@ -408,7 +412,10 @@ void set_scheduler(scheduler::abstract_coordinator* ptr); ...@@ -408,7 +412,10 @@ void set_scheduler(scheduler::abstract_coordinator* ptr);
*/ */
template <class Policy = policy::work_stealing> template <class Policy = policy::work_stealing>
void set_scheduler(size_t nw = std::thread::hardware_concurrency(), void set_scheduler(size_t nw = std::thread::hardware_concurrency(),
size_t max_throughput = 0) { size_t max_throughput = std::numeric_limits<size_t>::max()) {
if (max_throughput == 0) {
throw std::invalid_argument("max_throughput must not be 0");
}
set_scheduler(new scheduler::coordinator<Policy>(nw, max_throughput)); set_scheduler(new scheduler::coordinator<Policy>(nw, max_throughput));
} }
......
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