Mirror/FairMQ
1
0
Fork 0
mirror of https://github.com/FairRootGroup/FairMQ.git synced 2025-10-13 08:41:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

Format

Browse Source
This commit is contained in:
Alexey Rybalchenko 2020-05-17 14:42:38 +02:00
parent b56e32eb11
commit 2916a491b9
1 changed files with 91 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
fairmq/tools/RateLimit.h
View File

@ -40,7 +40,7 @@ class RateLimiter
{ {
using clock = std::chrono::steady_clock; using clock = std::chrono::steady_clock;
public: public:
/** /**
* Constructs a rate limiter. * Constructs a rate limiter.
* *
@ -59,15 +59,15 @@ public:
} }
skip_check_count = std::max(1, int(std::chrono::milliseconds(5) / tw_req)); skip_check_count = std::max(1, int(std::chrono::milliseconds(5) / tw_req));
count = skip_check_count; count = skip_check_count;
//std::cerr << "skip_check_count: " << skip_check_count << '\n'; // std::cerr << "skip_check_count: " << skip_check_count << '\n';
} }
/** /**
* Call this function at the end of the iteration rate limited loop. * Call this function at the end of the iteration rate limited loop.
* *
* This function might use `std::this_thread::sleep_for` to limit the iteration rate. If no sleeps * This function might use `std::this_thread::sleep_for` to limit the iteration rate. If no
* are necessary, the function will back off checking for the time to further allow increased * sleeps are necessary, the function will back off checking for the time to further allow
* iteration rates (until the requested rate or 1s between rechecks is reached). * increased iteration rates (until the requested rate or 1s between rechecks is reached).
*/ */
void maybe_sleep() void maybe_sleep()
{ {
@ -79,20 +79,23 @@ public:
} else { } else {
tw = (1 * tw + 3 * (now - start_time) / skip_check_count) / 4; tw = (1 * tw + 3 * (now - start_time) / skip_check_count) / 4;
} }
//std::ostringstream s; s << "tw = " << std::setw(10) << duration_cast<nanoseconds>(tw).count() << "ns, req = " << duration_cast<nanoseconds>(tw_req).count() << "ns, "; // std::ostringstream s; s << "tw = " << std::setw(10) <<
// duration_cast<nanoseconds>(tw).count() << "ns, req = " <<
// duration_cast<nanoseconds>(tw_req).count() << "ns, ";
if (tw > tw_req * 65 / 64) { if (tw > tw_req * 65 / 64) {
// the time between maybe_sleep calls is more than 1% too long // the time between maybe_sleep calls is more than 1% too long
// fix it by reducing ts towards 0 and if ts = 0 doesn't suffice, increase // fix it by reducing ts towards 0 and if ts = 0 doesn't suffice, increase
// skip_check_count // skip_check_count
if (ts > clock::duration::zero()) { if (ts > clock::duration::zero()) {
ts = std::max(clock::duration::zero(), ts = std::max(clock::duration::zero(), ts - (tw - tw_req) * skip_check_count * 1 / 2);
ts - (tw - tw_req) * skip_check_count * 1 / 2); // std::cerr << s.str() << "maybe_sleep: going too slow; sleep less: " <<
//std::cerr << s.str() << "maybe_sleep: going too slow; sleep less: " << duration_cast<microseconds>(ts).count() << "µs\n"; // duration_cast<microseconds>(ts).count() << "µs\n";
} else { } else {
skip_check_count = skip_check_count =
std::min(int(seconds(1) / tw_req), // recheck at least every second std::min(int(seconds(1) / tw_req), // recheck at least every second
(skip_check_count * 5 + 3) / 4); (skip_check_count * 5 + 3) / 4);
//std::cerr << s.str() << "maybe_sleep: going too slow; work more: " << skip_check_count << "\n"; // std::cerr << s.str() << "maybe_sleep: going too slow; work more: " <<
// skip_check_count << "\n";
} }
} else if (tw < tw_req * 63 / 64) { } else if (tw < tw_req * 63 / 64) {
// the time between maybe_sleep calls is more than 1% too short // the time between maybe_sleep calls is more than 1% too short
@ -107,10 +110,12 @@ public:
if (skip_check_count > min_skip_count) { if (skip_check_count > min_skip_count) {
assert(ts == clock::duration::zero()); assert(ts == clock::duration::zero());
skip_check_count = std::max(min_skip_count, skip_check_count * 3 / 4); skip_check_count = std::max(min_skip_count, skip_check_count * 3 / 4);
//std::cerr << s.str() << "maybe_sleep: going too fast; work less: " << skip_check_count << "\n"; // std::cerr << s.str() << "maybe_sleep: going too fast; work less: " <<
// skip_check_count << "\n";
} else { } else {
ts += (tw_req - tw) * (skip_check_count * 7) / 8; ts += (tw_req - tw) * (skip_check_count * 7) / 8;
//std::cerr << s.str() << "maybe_sleep: going too fast; sleep more: " << duration_cast<microseconds>(ts).count() << "µs\n"; // std::cerr << s.str() << "maybe_sleep: going too fast; sleep more: " <<
// duration_cast<microseconds>(ts).count() << "µs\n";
} }
} }
@ -122,7 +127,7 @@ public:
} }
} }
private: private:
clock::duration tw{}, //! deduced duration between maybe_sleep calls clock::duration tw{}, //! deduced duration between maybe_sleep calls
ts{}, //! sleep duration ts{}, //! sleep duration
tw_req; //! requested duration between maybe_sleep calls tw_req; //! requested duration between maybe_sleep calls