empiredan commented on code in PR #1074:
URL: https://github.com/apache/incubator-pegasus/pull/1074#discussion_r931105450
##########
src/rdsn/src/utils/metrics.cpp:
##########
@@ -132,27 +185,78 @@ uint64_t
percentile_timer::generate_initial_delay_ms(uint64_t interval_ms)
return (rand::next_u64() % interval_seconds + 1) * 1000 + rand::next_u64()
% 1000;
}
-percentile_timer::percentile_timer(uint64_t interval_ms, exec_fn exec)
+percentile_timer::percentile_timer(uint64_t interval_ms, on_exec_fn on_exec,
on_close_fn on_close)
: _initial_delay_ms(generate_initial_delay_ms(interval_ms)),
_interval_ms(interval_ms),
- _exec(exec),
+ _on_exec(on_exec),
+ _on_close(on_close),
+ _state(state::kRunning),
+ _completed(),
_timer(new
boost::asio::deadline_timer(tools::shared_io_service::instance().ios))
{
_timer->expires_from_now(boost::posix_time::milliseconds(_initial_delay_ms));
_timer->async_wait(std::bind(&percentile_timer::on_timer, this,
std::placeholders::_1));
}
+void percentile_timer::close()
+{
+ // If the timer has already expired when cancel() is called, then the
handlers for asynchronous
+ // wait operations will:
+ // * have already been invoked; or
+ // * have been queued for invocation in the near future.
+ //
+ // These handlers can no longer be cancelled, and therefore are passed an
error code that
+ // indicates the successful completion of the wait operation. Thus set the
state of timer to
+ // kClosing to tell on_timer() that the timer should be closed even if it
is not called with
+ // operation_canceled.
+ auto expected_state = state::kRunning;
+ if (_state.compare_exchange_strong(expected_state, state::kClosing)) {
+ _timer->cancel();
+ }
+}
+
+void percentile_timer::wait() { _completed.wait(); }
+
+void percentile_timer::on_close()
+{
+ _on_close();
+ _completed.notify();
+}
+
void percentile_timer::on_timer(const boost::system::error_code &ec)
{
+// This macro is defined for the case that handlers for asynchronous wait
operations are no
+// longer cancelled. It just checks the internal state atomically (since
close() can also be
+// called simultaneously) for kClosing; once it's matched, it will not execute
future callbacks
+// to stop the timer.
+#define TRY_PROCESS_TIMER_CLOSING()
\
+ do {
\
+ auto expected_state = state::kClosing;
\
+ if (_state.compare_exchange_strong(expected_state, state::kClosed)) {
\
+ on_close();
\
+ return;
\
+ }
\
+ } while (0)
+
if (dsn_unlikely(!!ec)) {
dassert_f(ec == boost::system::errc::operation_canceled,
"failed to exec on_timer with an error that cannot be
handled: {}",
ec.message());
+
+ // Cancel can only be launched by close().
+ auto expected_state = state::kClosing;
+ dassert_f(_state.compare_exchange_strong(expected_state,
state::kClosed),
+ "wrong state for percentile_timer: {}, while expecting
closing state",
+ static_cast<int>(expected_state));
+ on_close();
+
return;
}
- _exec();
+ TRY_PROCESS_TIMER_CLOSING();
+ _on_exec();
+ TRY_PROCESS_TIMER_CLOSING();
Review Comment:
Sure.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
