Copilot commented on code in PR #13898:
URL: https://github.com/apache/trafficserver/pull/13898#discussion_r4459118867
##########
src/iocore/eventsystem/Watchdog.cc:
##########
@@ -37,73 +42,448 @@
namespace Watchdog
{
+using ts::Metrics;
+
DbgCtl dbg_ctl_watchdog("watchdog");
-Monitor::Monitor(EThread *threads[], size_t n_threads,
std::chrono::milliseconds timeout_ms)
- : _threads(threads, threads + n_threads), _timeout{timeout_ms}
+namespace
+{
+ // this ensures `long long` duration for older compilers to match format
specifiers
+ using printable_micro = std::chrono::duration<long long, std::micro>;
+ using printable_milli = std::chrono::duration<long long, std::milli>;
+
+ /** Current value of accept_refuse(). Written only by the monitor thread and
the config callback.
+
+ Read from the accept paths on every accept loop.
+ */
+ std::atomic<bool> accept_refuse_state{false};
+
+ /** Current value of proxy.config.exec_thread.watchdog.mode.
+
+ File scope rather than a Monitor member on purpose:
RecRegisterConfigUpdateCb has no
+ unregister counterpart, so a callback that captured the Monitor would
outlive it and fire
+ against freed memory at shutdown. Keeping the mode here means the
callback touches only static
+ storage and remains valid for the life of the process.
+ */
+ std::atomic<Mode> current_mode{Mode::DISABLED};
+
+ /** proxy.config.exec_thread.watchdog.timeout_refuse_ms - threshold for
accept_refuse().
+
+ Zero disables it entirely. Independent of the shrink timeout that drives
the scan period: the
+ accept paths stopping and the accept queues shrinking are separate
decisions on the same peak,
+ so this can be reloaded freely without disturbing the scan rate.
+ */
+ std::atomic<int64_t> refuse_timeout_ms{0};
+
+ /// proxy.config.exec_thread.watchdog.backlog_min_permille - floor for
backlog_scale_permille().
+ std::atomic<int32_t> backlog_min_permille{50};
+
+ /// Published listen backlog scale, in permille of the configured value.
1000 == unthrottled.
+ std::atomic<int32_t> backlog_scale{1000};
+
+ /** proxy.config.exec_thread.watchdog.backlog_decrease_permille -
multiplicative decrease.
+
+ Expressed as the permille of the *current* scale retained on each
saturated scan, rather than
+ as a divisor, so the whole range is reachable rather than just 1/2, 1/3,
1/4: 500 is the
+ classic halving, 667 is a divide by 1.5, 900 is a slow squeeze, 975 is a
crawl.
+
+ Doubles as the release deadband - see release_threshold(). One number
for how much margin is
+ wanted against the overload, applied to the queue depth and to the
signal alike.
+ */
+ std::atomic<int32_t> backlog_decrease_permille{500};
+
+ /** proxy.config.exec_thread.watchdog.backlog_increase_permille -
proportional increase.
+
+ Permille of the scale *currently in force* offered as the step on each
recovered scan, before
+ the headroom attenuation in recovery_step_permille(). Equivalently,
permille points of the
+ configured backlog at full depth, which is where the two readings
coincide and why one number
+ serves as both the rate and the ceiling on the step - the scale never
exceeds 1000, so the
+ proportional value can never exceed the configured points.
+
+ Proportional to the current depth rather than a flat number of points
because what re-saturates
+ the threads is the count of connections admitted, not the fraction of
the baseline they
+ represent, and a flat step is a fixed count whatever depth is in force.
At the bottom of a hard
+ shed that count is a multiple of the whole queue: at 100 points of a
65,535 baseline, the first
+ recovered scan after a shed to 20 permille would readmit 6,553
connections onto a queue holding
+ 1,310. The same 100 here readmits 131, and the step grows with the queue
as it climbs out.
+ */
+ std::atomic<int32_t> backlog_increase_permille{100};
+
+ int
+ update_watchdog_mode(const char * /* name */, RecDataT /* type */, RecData
data, void * /* cookie */)
+ {
+ auto mode = static_cast<Mode>(data.rec_int);
+ if (mode != Mode::DISABLED && mode != Mode::BACKPRESSURE) {
+ Warning("Watchdog: ignoring out of range
proxy.config.exec_thread.watchdog.mode %" PRId64, data.rec_int);
+ return REC_ERR_OKAY;
+ }
+
+ Mode const prev = current_mode.exchange(mode, std::memory_order_relaxed);
+ if (prev == mode) {
+ return REC_ERR_OKAY;
+ }
+
+ // Drop any backpressure the outgoing mode had engaged. The monitor
re-derives everything from
+ // the new mode on its next scan, so the worst case is up to half a
timeout of not shedding -
+ // strictly the safe direction. Leaving it set would be the unsafe one:
switching to mode 0
+ // would otherwise strand the accept queues shrunk forever.
+ //
+ // The reporting gauges are left to the monitor rather than cleared here.
It owns them, and
+ // sync_state_gauges() corrects them on its next scan; writing them from
this thread would race
+ // its stores.
+ if (mode == Mode::DISABLED) {
+ accept_refuse_state.store(false, std::memory_order_relaxed);
+ backlog_scale.store(1000, std::memory_order_relaxed);
+ }
+
+ Note("Watchdog: mode changed from %d to %d", static_cast<int>(prev),
static_cast<int>(mode));
+ return REC_ERR_OKAY;
+ }
+
+ int
+ update_refuse_timeout(const char * /* name */, RecDataT /* type */, RecData
data, void * /* cookie */)
+ {
+ int64_t const want = std::max<int64_t>(0, data.rec_int);
+ refuse_timeout_ms.store(want, std::memory_order_relaxed);
+
+ // Turning it off must resume accepting here rather than at the next scan
- the whole point of
+ // setting it to zero is to stop refusing immediately. Turning it on is
left to the monitor, which
+ // has an actual peak to compare against.
+ if (want == 0) {
+ accept_refuse_state.store(false, std::memory_order_relaxed);
+ }
+ return REC_ERR_OKAY;
+ }
+
+ int
+ update_backlog_min_permille(const char * /* name */, RecDataT /* type */,
RecData data, void * /* cookie */)
+ {
+
backlog_min_permille.store(static_cast<int32_t>(std::clamp<int64_t>(data.rec_int,
1, 1000)), std::memory_order_relaxed);
+ return REC_ERR_OKAY;
+ }
+
+ int
+ update_backlog_decrease_permille(const char * /* name */, RecDataT /* type
*/, RecData data, void * /* cookie */)
+ {
+
backlog_decrease_permille.store(static_cast<int32_t>(std::clamp<int64_t>(data.rec_int,
1, 999)), std::memory_order_relaxed);
+ return REC_ERR_OKAY;
+ }
+
+ int
+ update_backlog_increase_permille(const char * /* name */, RecDataT /* type
*/, RecData data, void * /* cookie */)
+ {
+
backlog_increase_permille.store(static_cast<int32_t>(std::clamp<int64_t>(data.rec_int,
1, 1000)), std::memory_order_relaxed);
+ return REC_ERR_OKAY;
+ }
+
+} // namespace
+
+void
+register_config_callbacks()
+{
+ RecInt mode =
RecGetRecordInt("proxy.config.exec_thread.watchdog.mode").value_or(0);
+ current_mode.store(static_cast<Mode>(mode), std::memory_order_relaxed);
+ RecRegisterConfigUpdateCb("proxy.config.exec_thread.watchdog.mode",
update_watchdog_mode, nullptr);
+
+ RecInt refuse_timeout =
RecGetRecordInt("proxy.config.exec_thread.watchdog.timeout_refuse_ms").value_or(0);
+ refuse_timeout_ms.store(std::max<int64_t>(0, refuse_timeout),
std::memory_order_relaxed);
+
RecRegisterConfigUpdateCb("proxy.config.exec_thread.watchdog.timeout_refuse_ms",
update_refuse_timeout, nullptr);
+
+ RecInt min_permille =
RecGetRecordInt("proxy.config.exec_thread.watchdog.backlog_min_permille").value_or(50);
+
backlog_min_permille.store(static_cast<int32_t>(std::clamp<int64_t>(min_permille,
1, 1000)), std::memory_order_relaxed);
+
RecRegisterConfigUpdateCb("proxy.config.exec_thread.watchdog.backlog_min_permille",
update_backlog_min_permille, nullptr);
+
+ RecInt dec_permille =
RecGetRecordInt("proxy.config.exec_thread.watchdog.backlog_decrease_permille").value_or(500);
+
backlog_decrease_permille.store(static_cast<int32_t>(std::clamp<int64_t>(dec_permille,
1, 999)), std::memory_order_relaxed);
+
RecRegisterConfigUpdateCb("proxy.config.exec_thread.watchdog.backlog_decrease_permille",
update_backlog_decrease_permille,
+ nullptr);
+
+ RecInt inc_permille =
RecGetRecordInt("proxy.config.exec_thread.watchdog.backlog_increase_permille").value_or(100);
+
backlog_increase_permille.store(static_cast<int32_t>(std::clamp<int64_t>(inc_permille,
1, 1000)), std::memory_order_relaxed);
+
RecRegisterConfigUpdateCb("proxy.config.exec_thread.watchdog.backlog_increase_permille",
update_backlog_increase_permille,
+ nullptr);
+}
+
+void
+record_awake_span(Heartbeat &hb, std::chrono::steady_clock::duration span)
+{
+ auto ns = static_cast<uint64_t>(std::max<int64_t>(0,
std::chrono::duration_cast<std::chrono::nanoseconds>(span).count()));
+ // Only the owning EThread raises this; the monitor's exchange to zero is
the only other writer, so a
+ // plain load/store pair could lose a reset. Use CAS to keep the max
semantics under that race.
+ uint64_t prev = hb.peak_awake.load(std::memory_order_relaxed);
+ while (prev < ns && !hb.peak_awake.compare_exchange_weak(prev, ns,
std::memory_order_relaxed, std::memory_order_relaxed)) {
+ ;
+ }
+}
+
+bool
+accept_refuse()
+{
+ // Gated on the live configuration so a scan that raced a config callback
and published a stale
+ // result cannot re-enable an action that has since been turned off.
+ return accept_refuse_state.load(std::memory_order_relaxed) &&
+ current_mode.load(std::memory_order_relaxed) == Mode::BACKPRESSURE &&
+ refuse_timeout_ms.load(std::memory_order_relaxed) > 0;
+}
+
+int
+backlog_scale_permille()
+{
+ if (current_mode.load(std::memory_order_relaxed) != Mode::BACKPRESSURE) {
+ return 1000;
+ }
+ return backlog_scale.load(std::memory_order_relaxed);
+}
+
+SleepScope::SleepScope(Heartbeat &hb) : _hb(hb)
+{
+ auto now = std::chrono::steady_clock::now();
+
+ // Publish the span that just ended, so a burst finishing between two
monitor scans is not lost.
+ auto last_wake = _hb.last_wake.load(std::memory_order_relaxed);
+ if (last_wake != std::chrono::steady_clock::time_point::min()) {
+ record_awake_span(_hb, now - last_wake);
+ }
+
+ // Enter the odd (sleeping) phase.
+ _hb.seq.fetch_add(1, std::memory_order_relaxed);
+}
+
+SleepScope::~SleepScope()
+{
+ _hb.last_wake.store(std::chrono::steady_clock::now(),
std::memory_order_relaxed);
+ // Back to even (awake). Release so the monitor cannot observe the even
sequence before the
+ // timestamp store above.
+ _hb.seq.fetch_add(1, std::memory_order_release);
+}
+
+Monitor::Monitor(EThread *threads[], size_t n_threads,
std::chrono::milliseconds shrink_timeout_ms, bool legacy)
+ : _threads(threads, threads + n_threads),
_shrink_timeout{shrink_timeout_ms}, _legacy{legacy}
{
- // Precondition: timeout_ms must be > 0. A timeout of 0 indicates the
watchdog is disabled
+ // Precondition: shrink_timeout_ms must be > 0. A timeout of 0 indicates the
watchdog is disabled
// and the caller should not instantiate the Monitor (see traffic_server.cc).
- ink_release_assert(timeout_ms.count() > 0);
+ ink_release_assert(shrink_timeout_ms.count() > 0);
+ if (!_legacy) {
+ create_metrics();
+ }
_watchdog_thread = std::thread(std::bind_front(&Monitor::monitor_loop,
this));
}
+void
+Monitor::create_metrics()
+{
+ _max_awake_ms =
Metrics::Gauge::createPtr("proxy.process.watchdog.max_awake_ms");
+ _max_awake_ms_hwm =
Metrics::Gauge::createPtr("proxy.process.watchdog.max_awake_ms_hwm");
+ _shrink_exceeded_count =
Metrics::Counter::createPtr("proxy.process.watchdog.timeout_shrink_exceeded_count");
+ _refuse_exceeded_count =
Metrics::Counter::createPtr("proxy.process.watchdog.timeout_refuse_exceeded_count");
+ _shrink_gauge =
Metrics::Gauge::createPtr("proxy.process.watchdog.accept_shrink");
+ _shrink_count =
Metrics::Counter::createPtr("proxy.process.watchdog.accept_shrink_engaged_count");
+ _refuse_gauge =
Metrics::Gauge::createPtr("proxy.process.watchdog.accept_refuse");
+ _refuse_count =
Metrics::Counter::createPtr("proxy.process.watchdog.accept_refuse_engaged_count");
+ _backlog_permille =
Metrics::Gauge::createPtr("proxy.process.watchdog.listen_backlog_permille");
+ Metrics::Gauge::store(_backlog_permille, 1000);
+}
+
Monitor::~Monitor()
{
_shutdown.store(true, std::memory_order_release);
_watchdog_thread.join();
+ // Never leave the accept paths throttled behind a monitor that is no longer
scanning.
+ accept_refuse_state.store(false, std::memory_order_relaxed);
+ backlog_scale.store(1000, std::memory_order_relaxed);
+}
+
+std::chrono::steady_clock::duration
+Monitor::scan(std::chrono::steady_clock::time_point now)
+{
+ std::chrono::steady_clock::duration peak =
std::chrono::steady_clock::duration::zero();
+
+ for (size_t i = 0; i < _threads.size(); ++i) {
+ EThread *t = _threads[i];
+
+ // Completed spans the thread finished since the previous scan. Draining
to zero here is what makes
+ // this a per-scan peak rather than a lifetime high water mark.
+ auto recorded =
std::chrono::nanoseconds{t->heartbeat_state.peak_awake.exchange(0,
std::memory_order_relaxed)};
+ peak = std::max<std::chrono::steady_clock::duration>(peak,
recorded);
+
+ // Seqlock read of the sleep/wake pair. An odd sequence means the thread
is inside the blocking
+ // poll, and a changed sequence means it transitioned while we were
reading - either way the pair
+ // is not a coherent snapshot and the live reading is skipped. peak_awake
above already accounts
+ // for that thread's completed work, so skipping loses nothing.
+ uint64_t const seq_before =
t->heartbeat_state.seq.load(std::memory_order_acquire);
+ if (seq_before & 1) {
+ continue; // sleeping
+ }
+
+ auto last_wake =
t->heartbeat_state.last_wake.load(std::memory_order_relaxed);
+
+ std::atomic_thread_fence(std::memory_order_acquire);
+ if (t->heartbeat_state.seq.load(std::memory_order_relaxed) != seq_before) {
+ continue; // torn - the thread slept or woke mid-read
+ }
+
+ if (last_wake == std::chrono::steady_clock::time_point::min()) {
+ continue; // event loop hasn't completed its first poll
+ }
+
+ // Live snapshot of the span in progress. A thread stuck in a single long
span never reaches the
+ // top of its next loop to record itself in peak_awake, so it only shows
up here.
+ auto awake_duration = now - last_wake;
+ if (awake_duration < std::chrono::steady_clock::duration::zero()) {
+ // @a now is sampled once at the top of the scan, before this thread's
seqlock read, so a thread
+ // that woke in between can report a last_wake fractionally newer than
it. Harmless and always
+ // sub-millisecond, but it must not be fed to std::max as a duration.
+ continue;
+ }
+ peak = std::max(peak, awake_duration);
+
+ if (awake_duration > _shrink_timeout) {
+ // Monitor thread is the sole reader (and warned_seq writer), so relaxed
accesses are race-free.
+ uint64_t warned_seq =
t->heartbeat_state.warned_seq.load(std::memory_order_relaxed);
+ if (warned_seq < seq_before) {
+ if (_legacy) {
+ // The deprecated timeout_ms warned here, and a config that still
sets it gets what it asked for.
+ Warning("Watchdog: [ET_NET %zu] has been awake for %lld ms", i,
+
std::chrono::duration_cast<printable_milli>(awake_duration).count());
+ } else {
+ // Debug rather than Warning: on a busy proxy the timeout is crossed
routinely, and a log line
+ // per event loop that runs long is noise an operator cannot act on.
The condition is already
+ // reported continuously and cheaply as
proxy.process.watchdog.max_awake_ms, and counted per
+ // scan as proxy.process.watchdog.timeout_shrink_exceeded_count,
which is the right thing to
+ // alert on.
+ // This is here for the case where you need to know *which* thread.
+ //
+ // Reported once per loop iteration, not once per scan.
+ Dbg(dbg_ctl_watchdog, "[ET_NET %zu] has been awake for %lld ms
(shrink timeout %lld ms)", i,
+
std::chrono::duration_cast<printable_milli>(awake_duration).count(),
+
std::chrono::duration_cast<printable_milli>(_shrink_timeout).count());
+ }
+ t->heartbeat_state.warned_seq.store(seq_before,
std::memory_order_relaxed);
+ }
+ }
+ }
+
+ return peak;
}
void
-Monitor::monitor_loop() const
+Monitor::sync_state_gauges()
{
- // this ensures `long long` duration for older compilers to match format
specifiers
- using printable_micro = std::chrono::duration<long long, std::micro>;
- using printable_milli = std::chrono::duration<long long, std::milli>;
+ bool const refusing = accept_refuse_state.load(std::memory_order_relaxed);
+ if (refusing != _refuse_reported) {
+ _refuse_reported = refusing;
+ Metrics::Gauge::store(_refuse_gauge, refusing ? 1 : 0);
+ }
+}
+void
+Monitor::monitor_loop()
+{
// Divide by a floating point 2 to avoid truncation to zero.
- auto sleep_time = _timeout / 2.0;
+ auto sleep_time = _shrink_timeout / 2.0;
ink_release_assert(sleep_time.count() > 0);
- Dbg(dbg_ctl_watchdog, "Starting watchdog with timeout %lld ms on %zu
threads. sleep_time = %lld us",
- std::chrono::duration_cast<printable_milli>(_timeout).count(),
_threads.size(),
+ Dbg(dbg_ctl_watchdog, "Starting watchdog with shrink timeout %lld ms, mode
%d, on %zu threads. sleep_time = %lld us",
+ std::chrono::duration_cast<printable_milli>(_shrink_timeout).count(),
+ static_cast<int>(current_mode.load(std::memory_order_relaxed)),
_threads.size(),
std::chrono::duration_cast<printable_micro>(sleep_time).count());
ink_set_thread_name("[WATCHDOG]");
while (!_shutdown.load(std::memory_order_acquire)) {
- std::chrono::time_point<std::chrono::steady_clock> now =
std::chrono::steady_clock::now();
- for (size_t i = 0; i < _threads.size(); ++i) {
- EThread *t = _threads[i];
- // Relaxed load: each heartbeat field has a single writer (its EThread)
so per-object coherence suffices.
- std::chrono::time_point<std::chrono::steady_clock> last_sleep =
t->heartbeat_state.last_sleep.load(std::memory_order_relaxed);
- if (last_sleep == std::chrono::steady_clock::time_point::min()) {
- // initial value sentinel - event loop hasn't started
- continue;
- }
- // Same reasoning for relaxed load on wake timestamp.
- std::chrono::time_point<std::chrono::steady_clock> last_wake =
t->heartbeat_state.last_wake.load(std::memory_order_relaxed);
+ auto peak = scan(std::chrono::steady_clock::now());
+
+ if (_legacy) {
+ std::this_thread::sleep_for(sleep_time);
+ continue;
+ }
+
+ auto peak_ms = std::chrono::duration_cast<printable_milli>(peak).count();
+ Metrics::Gauge::store(_max_awake_ms, peak_ms);
+
+ // Monotonic high water mark alongside the per-scan gauge above.
max_awake_ms covers only the
+ // interval since the previous scan and is overwritten by the next one, so
a spike that happens to
+ // land between two scrapes of the metric is never seen at all - and the
scan period is half the
+ // shrink timeout, which is far shorter than any realistic scrape
interval. This one only ever
+ // rises, so the worst span the process has served is still readable after
the fact.
+ if (peak_ms > _hwm_ms) {
+ _hwm_ms = peak_ms;
+ Metrics::Gauge::store(_max_awake_ms_hwm, peak_ms);
+ }
+
+ // Re-read the mode every scan - it is reloadable, so it can change under
us at any point.
+ Mode const mode = current_mode.load(std::memory_order_relaxed);
- if (last_wake == std::chrono::steady_clock::time_point::min() ||
last_wake < last_sleep) {
- // not yet woken from last sleep
- continue;
+ // The arithmetic lives in advance() so it can be exercised without a
running monitor. This loop gathers
+ // its inputs, applies the result to the published atomics and metrics,
and reports the edges. The refuse
+ // state is read back from the published atomic rather than mirrored in a
member, because
+ // update_refuse_timeout() and update_watchdog_mode() also clear it - a
local mirror would go stale against
+ // those and suppress the store that re-engages.
+ PolicyConfig config;
+ config.backpressure = mode == Mode::BACKPRESSURE;
+ config.shrink_timeout = _shrink_timeout;
+ config.refuse_timeout =
std::chrono::milliseconds{refuse_timeout_ms.load(std::memory_order_relaxed)};
+ config.decrease_permille =
backlog_decrease_permille.load(std::memory_order_relaxed);
+ config.increase_permille =
backlog_increase_permille.load(std::memory_order_relaxed);
+ config.min_permille =
backlog_min_permille.load(std::memory_order_relaxed);
+
+ PolicyState const before{_shrink_engaged,
accept_refuse_state.load(std::memory_order_relaxed), _backlog_scale,
_trouble_scale};
+ auto const result = advance(config, before, peak);
+ PolicyState const &after = result.state;
+
+ // Counted per saturated scan and deliberately independent of @c mode, so
mode 0 - report only -
+ // still gets a monotonic record of how often the threads crossed the
shrink threshold. The engaged
+ // counters below only move on a state edge, which says nothing about how
long or how often, and
+ // the max_awake_ms gauge can miss the crossing entirely between scrapes.
Dividing this by the scan
+ // rate gives the fraction of time spent over the threshold. The refuse
threshold is counted the same way.
+ if (result.saturated) {
+ Metrics::Counter::increment(_shrink_exceeded_count);
+ }
+ if (result.refuse_exceeded) {
+ Metrics::Counter::increment(_refuse_exceeded_count);
+ }
+
+ _shrink_engaged = after.shrink_engaged;
+ _trouble_scale = after.trouble_scale;
+
+ if (after.shrink_engaged != before.shrink_engaged) {
+ Metrics::Gauge::store(_shrink_gauge, after.shrink_engaged ? 1 : 0);
+ if (after.shrink_engaged) {
+ Metrics::Counter::increment(_shrink_count);
}
+ Dbg(dbg_ctl_watchdog, "backlog shrink -> %s - peak awake %lld ms, shrink
timeout %lld ms",
+ after.shrink_engaged ? "on" : "off", peak_ms,
std::chrono::duration_cast<printable_milli>(_shrink_timeout).count());
+ }
- auto awake_duration = now - last_wake;
- if (awake_duration > _timeout) {
- // Monitor thread is the sole reader (and warned_seq writer), so
relaxed accesses are race-free.
- uint64_t seq =
t->heartbeat_state.seq.load(std::memory_order_relaxed);
- uint64_t warned_seq =
t->heartbeat_state.warned_seq.load(std::memory_order_relaxed);
- if (warned_seq < seq) {
- // Warn once per loop iteration
- Warning("Watchdog: [ET_NET %zu] has been awake for %lld ms", i,
-
std::chrono::duration_cast<printable_milli>(awake_duration).count());
- t->heartbeat_state.warned_seq.store(seq, std::memory_order_relaxed);
- }
+ // Also compare against the published value: a config callback resets it
without touching
+ // _backlog_scale, and the two must be reconciled even when the monitor's
own scale is unchanged.
+ if (after.scale != before.scale || after.scale !=
backlog_scale.load(std::memory_order_relaxed)) {
+ _backlog_scale = after.scale;
+ backlog_scale.store(after.scale, std::memory_order_relaxed);
+ Metrics::Gauge::store(_backlog_permille, after.scale);
+ Dbg(dbg_ctl_watchdog, "listen backlog scale -> %d permille - peak awake
%lld ms", after.scale, peak_ms);
+ }
+
+ if (after.refuse != before.refuse) {
+ accept_refuse_state.store(after.refuse, std::memory_order_relaxed);
+ if (after.refuse) {
+ Metrics::Counter::increment(_refuse_count);
}
+ Dbg(dbg_ctl_watchdog, "accept refuse -> %s - peak awake %lld ms, refuse
timeout %lld ms", after.refuse ? "on" : "off",
+ peak_ms, static_cast<long long>(config.refuse_timeout.count()));
}
+ sync_state_gauges();
+
std::this_thread::sleep_for(sleep_time);
}
+
+ if (accept_refuse_state.load(std::memory_order_relaxed) || _backlog_scale !=
1000) {
+ accept_refuse_state.store(false, std::memory_order_relaxed);
+ backlog_scale.store(1000, std::memory_order_relaxed);
+ Metrics::Gauge::store(_shrink_gauge, 0);
+ Metrics::Gauge::store(_backlog_permille, 1000);
+ }
+ sync_state_gauges();
Review Comment:
Legacy-mode monitors (`_legacy == true`) never call `create_metrics()`,
leaving metric pointers (e.g. `_shrink_gauge`, `_backlog_permille`,
`_refuse_gauge`) null. When the loop exits, the unconditional cleanup block and
`sync_state_gauges()` will dereference those null pointers and can crash on
shutdown. Fix by guarding all metric stores/`sync_state_gauges()` behind `if
(!_legacy)` (or early-return before the cleanup section when `_legacy` is
true), so legacy warn-only mode never touches metric pointers.
##########
src/iocore/net/Server.cc:
##########
@@ -158,6 +160,13 @@ Server::listen(bool non_blocking, const
NetProcessor::AcceptOptions &opt)
close();
}
+ if (opt.f_non_fatal_listen) {
+ // A supplementary listener. Losing it degrades behavior but must not take
the proxy down.
+ Warning("Could not bind or listen to port %d, mptcp enabled: %d (error:
%d) %s %d", ats_ip_port_host_order(&addr),
+ prot == IPPROTO_MPTCP, errno, strerror(errno), res);
+ return res;
+ }
Review Comment:
This warning logs `errno`/`strerror(errno)` at `Lerror`, but `errno` may no
longer correspond to the failing syscall after cleanup work (e.g., `close()` or
other calls that can mutate `errno`). Capture the failing `errno` into a local
(e.g., `saved_errno`) at the point of failure and log that instead. Also, the
message says 'bind or listen' even for failures from `socket()`, `chmod()`, or
`chown()` paths shown in this diff; consider broadening phrasing (e.g., 'Could
not set up listener') to keep the warning accurate.
##########
src/iocore/eventsystem/Watchdog.cc:
##########
@@ -37,73 +42,448 @@
namespace Watchdog
{
+using ts::Metrics;
+
DbgCtl dbg_ctl_watchdog("watchdog");
-Monitor::Monitor(EThread *threads[], size_t n_threads,
std::chrono::milliseconds timeout_ms)
- : _threads(threads, threads + n_threads), _timeout{timeout_ms}
+namespace
+{
+ // this ensures `long long` duration for older compilers to match format
specifiers
+ using printable_micro = std::chrono::duration<long long, std::micro>;
+ using printable_milli = std::chrono::duration<long long, std::milli>;
+
+ /** Current value of accept_refuse(). Written only by the monitor thread and
the config callback.
+
+ Read from the accept paths on every accept loop.
+ */
+ std::atomic<bool> accept_refuse_state{false};
+
+ /** Current value of proxy.config.exec_thread.watchdog.mode.
+
+ File scope rather than a Monitor member on purpose:
RecRegisterConfigUpdateCb has no
+ unregister counterpart, so a callback that captured the Monitor would
outlive it and fire
+ against freed memory at shutdown. Keeping the mode here means the
callback touches only static
+ storage and remains valid for the life of the process.
+ */
+ std::atomic<Mode> current_mode{Mode::DISABLED};
+
+ /** proxy.config.exec_thread.watchdog.timeout_refuse_ms - threshold for
accept_refuse().
+
+ Zero disables it entirely. Independent of the shrink timeout that drives
the scan period: the
+ accept paths stopping and the accept queues shrinking are separate
decisions on the same peak,
+ so this can be reloaded freely without disturbing the scan rate.
+ */
+ std::atomic<int64_t> refuse_timeout_ms{0};
+
+ /// proxy.config.exec_thread.watchdog.backlog_min_permille - floor for
backlog_scale_permille().
+ std::atomic<int32_t> backlog_min_permille{50};
+
+ /// Published listen backlog scale, in permille of the configured value.
1000 == unthrottled.
+ std::atomic<int32_t> backlog_scale{1000};
+
+ /** proxy.config.exec_thread.watchdog.backlog_decrease_permille -
multiplicative decrease.
+
+ Expressed as the permille of the *current* scale retained on each
saturated scan, rather than
+ as a divisor, so the whole range is reachable rather than just 1/2, 1/3,
1/4: 500 is the
+ classic halving, 667 is a divide by 1.5, 900 is a slow squeeze, 975 is a
crawl.
+
+ Doubles as the release deadband - see release_threshold(). One number
for how much margin is
+ wanted against the overload, applied to the queue depth and to the
signal alike.
+ */
+ std::atomic<int32_t> backlog_decrease_permille{500};
+
+ /** proxy.config.exec_thread.watchdog.backlog_increase_permille -
proportional increase.
+
+ Permille of the scale *currently in force* offered as the step on each
recovered scan, before
+ the headroom attenuation in recovery_step_permille(). Equivalently,
permille points of the
+ configured backlog at full depth, which is where the two readings
coincide and why one number
+ serves as both the rate and the ceiling on the step - the scale never
exceeds 1000, so the
+ proportional value can never exceed the configured points.
+
+ Proportional to the current depth rather than a flat number of points
because what re-saturates
+ the threads is the count of connections admitted, not the fraction of
the baseline they
+ represent, and a flat step is a fixed count whatever depth is in force.
At the bottom of a hard
+ shed that count is a multiple of the whole queue: at 100 points of a
65,535 baseline, the first
+ recovered scan after a shed to 20 permille would readmit 6,553
connections onto a queue holding
+ 1,310. The same 100 here readmits 131, and the step grows with the queue
as it climbs out.
+ */
+ std::atomic<int32_t> backlog_increase_permille{100};
+
+ int
+ update_watchdog_mode(const char * /* name */, RecDataT /* type */, RecData
data, void * /* cookie */)
+ {
+ auto mode = static_cast<Mode>(data.rec_int);
+ if (mode != Mode::DISABLED && mode != Mode::BACKPRESSURE) {
+ Warning("Watchdog: ignoring out of range
proxy.config.exec_thread.watchdog.mode %" PRId64, data.rec_int);
+ return REC_ERR_OKAY;
+ }
+
+ Mode const prev = current_mode.exchange(mode, std::memory_order_relaxed);
+ if (prev == mode) {
+ return REC_ERR_OKAY;
+ }
+
+ // Drop any backpressure the outgoing mode had engaged. The monitor
re-derives everything from
+ // the new mode on its next scan, so the worst case is up to half a
timeout of not shedding -
+ // strictly the safe direction. Leaving it set would be the unsafe one:
switching to mode 0
+ // would otherwise strand the accept queues shrunk forever.
+ //
+ // The reporting gauges are left to the monitor rather than cleared here.
It owns them, and
+ // sync_state_gauges() corrects them on its next scan; writing them from
this thread would race
+ // its stores.
+ if (mode == Mode::DISABLED) {
+ accept_refuse_state.store(false, std::memory_order_relaxed);
+ backlog_scale.store(1000, std::memory_order_relaxed);
+ }
+
+ Note("Watchdog: mode changed from %d to %d", static_cast<int>(prev),
static_cast<int>(mode));
+ return REC_ERR_OKAY;
+ }
+
+ int
+ update_refuse_timeout(const char * /* name */, RecDataT /* type */, RecData
data, void * /* cookie */)
+ {
+ int64_t const want = std::max<int64_t>(0, data.rec_int);
+ refuse_timeout_ms.store(want, std::memory_order_relaxed);
+
+ // Turning it off must resume accepting here rather than at the next scan
- the whole point of
+ // setting it to zero is to stop refusing immediately. Turning it on is
left to the monitor, which
+ // has an actual peak to compare against.
+ if (want == 0) {
+ accept_refuse_state.store(false, std::memory_order_relaxed);
+ }
+ return REC_ERR_OKAY;
+ }
+
+ int
+ update_backlog_min_permille(const char * /* name */, RecDataT /* type */,
RecData data, void * /* cookie */)
+ {
+
backlog_min_permille.store(static_cast<int32_t>(std::clamp<int64_t>(data.rec_int,
1, 1000)), std::memory_order_relaxed);
+ return REC_ERR_OKAY;
+ }
+
+ int
+ update_backlog_decrease_permille(const char * /* name */, RecDataT /* type
*/, RecData data, void * /* cookie */)
+ {
+
backlog_decrease_permille.store(static_cast<int32_t>(std::clamp<int64_t>(data.rec_int,
1, 999)), std::memory_order_relaxed);
+ return REC_ERR_OKAY;
+ }
+
+ int
+ update_backlog_increase_permille(const char * /* name */, RecDataT /* type
*/, RecData data, void * /* cookie */)
+ {
+
backlog_increase_permille.store(static_cast<int32_t>(std::clamp<int64_t>(data.rec_int,
1, 1000)), std::memory_order_relaxed);
+ return REC_ERR_OKAY;
+ }
+
+} // namespace
+
+void
+register_config_callbacks()
+{
+ RecInt mode =
RecGetRecordInt("proxy.config.exec_thread.watchdog.mode").value_or(0);
+ current_mode.store(static_cast<Mode>(mode), std::memory_order_relaxed);
+ RecRegisterConfigUpdateCb("proxy.config.exec_thread.watchdog.mode",
update_watchdog_mode, nullptr);
+
+ RecInt refuse_timeout =
RecGetRecordInt("proxy.config.exec_thread.watchdog.timeout_refuse_ms").value_or(0);
+ refuse_timeout_ms.store(std::max<int64_t>(0, refuse_timeout),
std::memory_order_relaxed);
+
RecRegisterConfigUpdateCb("proxy.config.exec_thread.watchdog.timeout_refuse_ms",
update_refuse_timeout, nullptr);
+
+ RecInt min_permille =
RecGetRecordInt("proxy.config.exec_thread.watchdog.backlog_min_permille").value_or(50);
+
backlog_min_permille.store(static_cast<int32_t>(std::clamp<int64_t>(min_permille,
1, 1000)), std::memory_order_relaxed);
+
RecRegisterConfigUpdateCb("proxy.config.exec_thread.watchdog.backlog_min_permille",
update_backlog_min_permille, nullptr);
+
+ RecInt dec_permille =
RecGetRecordInt("proxy.config.exec_thread.watchdog.backlog_decrease_permille").value_or(500);
+
backlog_decrease_permille.store(static_cast<int32_t>(std::clamp<int64_t>(dec_permille,
1, 999)), std::memory_order_relaxed);
+
RecRegisterConfigUpdateCb("proxy.config.exec_thread.watchdog.backlog_decrease_permille",
update_backlog_decrease_permille,
+ nullptr);
+
+ RecInt inc_permille =
RecGetRecordInt("proxy.config.exec_thread.watchdog.backlog_increase_permille").value_or(100);
+
backlog_increase_permille.store(static_cast<int32_t>(std::clamp<int64_t>(inc_permille,
1, 1000)), std::memory_order_relaxed);
+
RecRegisterConfigUpdateCb("proxy.config.exec_thread.watchdog.backlog_increase_permille",
update_backlog_increase_permille,
+ nullptr);
+}
+
+void
+record_awake_span(Heartbeat &hb, std::chrono::steady_clock::duration span)
+{
+ auto ns = static_cast<uint64_t>(std::max<int64_t>(0,
std::chrono::duration_cast<std::chrono::nanoseconds>(span).count()));
+ // Only the owning EThread raises this; the monitor's exchange to zero is
the only other writer, so a
+ // plain load/store pair could lose a reset. Use CAS to keep the max
semantics under that race.
+ uint64_t prev = hb.peak_awake.load(std::memory_order_relaxed);
+ while (prev < ns && !hb.peak_awake.compare_exchange_weak(prev, ns,
std::memory_order_relaxed, std::memory_order_relaxed)) {
+ ;
+ }
+}
+
+bool
+accept_refuse()
+{
+ // Gated on the live configuration so a scan that raced a config callback
and published a stale
+ // result cannot re-enable an action that has since been turned off.
+ return accept_refuse_state.load(std::memory_order_relaxed) &&
+ current_mode.load(std::memory_order_relaxed) == Mode::BACKPRESSURE &&
+ refuse_timeout_ms.load(std::memory_order_relaxed) > 0;
+}
+
+int
+backlog_scale_permille()
+{
+ if (current_mode.load(std::memory_order_relaxed) != Mode::BACKPRESSURE) {
+ return 1000;
+ }
+ return backlog_scale.load(std::memory_order_relaxed);
+}
+
+SleepScope::SleepScope(Heartbeat &hb) : _hb(hb)
+{
+ auto now = std::chrono::steady_clock::now();
+
+ // Publish the span that just ended, so a burst finishing between two
monitor scans is not lost.
+ auto last_wake = _hb.last_wake.load(std::memory_order_relaxed);
+ if (last_wake != std::chrono::steady_clock::time_point::min()) {
+ record_awake_span(_hb, now - last_wake);
+ }
+
+ // Enter the odd (sleeping) phase.
+ _hb.seq.fetch_add(1, std::memory_order_relaxed);
+}
+
+SleepScope::~SleepScope()
+{
+ _hb.last_wake.store(std::chrono::steady_clock::now(),
std::memory_order_relaxed);
+ // Back to even (awake). Release so the monitor cannot observe the even
sequence before the
+ // timestamp store above.
+ _hb.seq.fetch_add(1, std::memory_order_release);
+}
+
+Monitor::Monitor(EThread *threads[], size_t n_threads,
std::chrono::milliseconds shrink_timeout_ms, bool legacy)
+ : _threads(threads, threads + n_threads),
_shrink_timeout{shrink_timeout_ms}, _legacy{legacy}
{
- // Precondition: timeout_ms must be > 0. A timeout of 0 indicates the
watchdog is disabled
+ // Precondition: shrink_timeout_ms must be > 0. A timeout of 0 indicates the
watchdog is disabled
// and the caller should not instantiate the Monitor (see traffic_server.cc).
- ink_release_assert(timeout_ms.count() > 0);
+ ink_release_assert(shrink_timeout_ms.count() > 0);
+ if (!_legacy) {
+ create_metrics();
+ }
_watchdog_thread = std::thread(std::bind_front(&Monitor::monitor_loop,
this));
}
+void
+Monitor::create_metrics()
+{
+ _max_awake_ms =
Metrics::Gauge::createPtr("proxy.process.watchdog.max_awake_ms");
+ _max_awake_ms_hwm =
Metrics::Gauge::createPtr("proxy.process.watchdog.max_awake_ms_hwm");
+ _shrink_exceeded_count =
Metrics::Counter::createPtr("proxy.process.watchdog.timeout_shrink_exceeded_count");
+ _refuse_exceeded_count =
Metrics::Counter::createPtr("proxy.process.watchdog.timeout_refuse_exceeded_count");
+ _shrink_gauge =
Metrics::Gauge::createPtr("proxy.process.watchdog.accept_shrink");
+ _shrink_count =
Metrics::Counter::createPtr("proxy.process.watchdog.accept_shrink_engaged_count");
+ _refuse_gauge =
Metrics::Gauge::createPtr("proxy.process.watchdog.accept_refuse");
+ _refuse_count =
Metrics::Counter::createPtr("proxy.process.watchdog.accept_refuse_engaged_count");
+ _backlog_permille =
Metrics::Gauge::createPtr("proxy.process.watchdog.listen_backlog_permille");
+ Metrics::Gauge::store(_backlog_permille, 1000);
+}
+
Monitor::~Monitor()
{
_shutdown.store(true, std::memory_order_release);
_watchdog_thread.join();
+ // Never leave the accept paths throttled behind a monitor that is no longer
scanning.
+ accept_refuse_state.store(false, std::memory_order_relaxed);
+ backlog_scale.store(1000, std::memory_order_relaxed);
+}
+
+std::chrono::steady_clock::duration
+Monitor::scan(std::chrono::steady_clock::time_point now)
+{
+ std::chrono::steady_clock::duration peak =
std::chrono::steady_clock::duration::zero();
+
+ for (size_t i = 0; i < _threads.size(); ++i) {
+ EThread *t = _threads[i];
+
+ // Completed spans the thread finished since the previous scan. Draining
to zero here is what makes
+ // this a per-scan peak rather than a lifetime high water mark.
+ auto recorded =
std::chrono::nanoseconds{t->heartbeat_state.peak_awake.exchange(0,
std::memory_order_relaxed)};
+ peak = std::max<std::chrono::steady_clock::duration>(peak,
recorded);
+
+ // Seqlock read of the sleep/wake pair. An odd sequence means the thread
is inside the blocking
+ // poll, and a changed sequence means it transitioned while we were
reading - either way the pair
+ // is not a coherent snapshot and the live reading is skipped. peak_awake
above already accounts
+ // for that thread's completed work, so skipping loses nothing.
+ uint64_t const seq_before =
t->heartbeat_state.seq.load(std::memory_order_acquire);
+ if (seq_before & 1) {
+ continue; // sleeping
+ }
+
+ auto last_wake =
t->heartbeat_state.last_wake.load(std::memory_order_relaxed);
+
+ std::atomic_thread_fence(std::memory_order_acquire);
+ if (t->heartbeat_state.seq.load(std::memory_order_relaxed) != seq_before) {
+ continue; // torn - the thread slept or woke mid-read
+ }
+
+ if (last_wake == std::chrono::steady_clock::time_point::min()) {
+ continue; // event loop hasn't completed its first poll
+ }
+
+ // Live snapshot of the span in progress. A thread stuck in a single long
span never reaches the
+ // top of its next loop to record itself in peak_awake, so it only shows
up here.
+ auto awake_duration = now - last_wake;
+ if (awake_duration < std::chrono::steady_clock::duration::zero()) {
+ // @a now is sampled once at the top of the scan, before this thread's
seqlock read, so a thread
+ // that woke in between can report a last_wake fractionally newer than
it. Harmless and always
+ // sub-millisecond, but it must not be fed to std::max as a duration.
+ continue;
+ }
+ peak = std::max(peak, awake_duration);
+
+ if (awake_duration > _shrink_timeout) {
+ // Monitor thread is the sole reader (and warned_seq writer), so relaxed
accesses are race-free.
+ uint64_t warned_seq =
t->heartbeat_state.warned_seq.load(std::memory_order_relaxed);
+ if (warned_seq < seq_before) {
+ if (_legacy) {
+ // The deprecated timeout_ms warned here, and a config that still
sets it gets what it asked for.
+ Warning("Watchdog: [ET_NET %zu] has been awake for %lld ms", i,
+
std::chrono::duration_cast<printable_milli>(awake_duration).count());
+ } else {
+ // Debug rather than Warning: on a busy proxy the timeout is crossed
routinely, and a log line
+ // per event loop that runs long is noise an operator cannot act on.
The condition is already
+ // reported continuously and cheaply as
proxy.process.watchdog.max_awake_ms, and counted per
+ // scan as proxy.process.watchdog.timeout_shrink_exceeded_count,
which is the right thing to
+ // alert on.
+ // This is here for the case where you need to know *which* thread.
+ //
+ // Reported once per loop iteration, not once per scan.
+ Dbg(dbg_ctl_watchdog, "[ET_NET %zu] has been awake for %lld ms
(shrink timeout %lld ms)", i,
+
std::chrono::duration_cast<printable_milli>(awake_duration).count(),
+
std::chrono::duration_cast<printable_milli>(_shrink_timeout).count());
+ }
+ t->heartbeat_state.warned_seq.store(seq_before,
std::memory_order_relaxed);
+ }
+ }
+ }
+
+ return peak;
}
void
-Monitor::monitor_loop() const
+Monitor::sync_state_gauges()
{
- // this ensures `long long` duration for older compilers to match format
specifiers
- using printable_micro = std::chrono::duration<long long, std::micro>;
- using printable_milli = std::chrono::duration<long long, std::milli>;
+ bool const refusing = accept_refuse_state.load(std::memory_order_relaxed);
+ if (refusing != _refuse_reported) {
+ _refuse_reported = refusing;
+ Metrics::Gauge::store(_refuse_gauge, refusing ? 1 : 0);
+ }
+}
+void
+Monitor::monitor_loop()
+{
// Divide by a floating point 2 to avoid truncation to zero.
- auto sleep_time = _timeout / 2.0;
+ auto sleep_time = _shrink_timeout / 2.0;
ink_release_assert(sleep_time.count() > 0);
- Dbg(dbg_ctl_watchdog, "Starting watchdog with timeout %lld ms on %zu
threads. sleep_time = %lld us",
- std::chrono::duration_cast<printable_milli>(_timeout).count(),
_threads.size(),
+ Dbg(dbg_ctl_watchdog, "Starting watchdog with shrink timeout %lld ms, mode
%d, on %zu threads. sleep_time = %lld us",
+ std::chrono::duration_cast<printable_milli>(_shrink_timeout).count(),
+ static_cast<int>(current_mode.load(std::memory_order_relaxed)),
_threads.size(),
std::chrono::duration_cast<printable_micro>(sleep_time).count());
ink_set_thread_name("[WATCHDOG]");
while (!_shutdown.load(std::memory_order_acquire)) {
- std::chrono::time_point<std::chrono::steady_clock> now =
std::chrono::steady_clock::now();
- for (size_t i = 0; i < _threads.size(); ++i) {
- EThread *t = _threads[i];
- // Relaxed load: each heartbeat field has a single writer (its EThread)
so per-object coherence suffices.
- std::chrono::time_point<std::chrono::steady_clock> last_sleep =
t->heartbeat_state.last_sleep.load(std::memory_order_relaxed);
- if (last_sleep == std::chrono::steady_clock::time_point::min()) {
- // initial value sentinel - event loop hasn't started
- continue;
- }
- // Same reasoning for relaxed load on wake timestamp.
- std::chrono::time_point<std::chrono::steady_clock> last_wake =
t->heartbeat_state.last_wake.load(std::memory_order_relaxed);
+ auto peak = scan(std::chrono::steady_clock::now());
+
+ if (_legacy) {
+ std::this_thread::sleep_for(sleep_time);
+ continue;
+ }
+
+ auto peak_ms = std::chrono::duration_cast<printable_milli>(peak).count();
+ Metrics::Gauge::store(_max_awake_ms, peak_ms);
+
+ // Monotonic high water mark alongside the per-scan gauge above.
max_awake_ms covers only the
+ // interval since the previous scan and is overwritten by the next one, so
a spike that happens to
+ // land between two scrapes of the metric is never seen at all - and the
scan period is half the
+ // shrink timeout, which is far shorter than any realistic scrape
interval. This one only ever
+ // rises, so the worst span the process has served is still readable after
the fact.
+ if (peak_ms > _hwm_ms) {
+ _hwm_ms = peak_ms;
+ Metrics::Gauge::store(_max_awake_ms_hwm, peak_ms);
+ }
Review Comment:
Legacy-mode monitors (`_legacy == true`) never call `create_metrics()`,
leaving metric pointers (e.g. `_shrink_gauge`, `_backlog_permille`,
`_refuse_gauge`) null. When the loop exits, the unconditional cleanup block and
`sync_state_gauges()` will dereference those null pointers and can crash on
shutdown. Fix by guarding all metric stores/`sync_state_gauges()` behind `if
(!_legacy)` (or early-return before the cleanup section when `_legacy` is
true), so legacy warn-only mode never touches metric pointers.
##########
doc/admin-guide/files/records.yaml.en.rst:
##########
@@ -434,18 +434,451 @@ Thread Variables
This option only has an affect when |TS| has been compiled with
``--enable-hwloc``.
-.. ts:cv:: CONFIG proxy.config.exec_thread.watchdog.timeout_ms INT 0
+.. ts:cv:: CONFIG proxy.config.exec_thread.watchdog.timeout_shrink_ms INT 0
:units: milliseconds
- Set the timeout for the exec thread watchdog in milliseconds. If an exec
thread
- does not heartbeat within this time period, the watchdog will log a warning
message.
- If this value is zero, the watchdog is disabled.
-
- The default of this watchdot timeout is set to 0 (disabled) for ATS 10.2 for
+ Peak awake span at which the exec thread watchdog considers the event
threads saturated,
+ in milliseconds. If this value is zero, the watchdog is disabled entirely
and
+ ``proxy.process.watchdog.max_awake_ms`` is not reported.
+
+ Under :ts:cv:`proxy.config.exec_thread.watchdog.mode` ``1`` this is also
the threshold at
+ which the accept queues start to shrink - hence the name. See that setting
for what the
+ shrinking does and
:ts:cv:`proxy.config.exec_thread.watchdog.timeout_refuse_ms` for the
+ other, independent action available.
+
+ An event thread awake longer than this is reported under the ``watchdog``
debug tag,
+ once per loop iteration. It is deliberately not a warning - on a busy proxy
the
+ threshold is crossed routinely, and a log line per long event loop is noise
that
+ cannot be acted on. Alert on
``proxy.process.watchdog.timeout_shrink_exceeded_count``
+ instead - a counter of the scans that crossed this threshold, maintained in
every
+ mode - and enable the debug tag when you need to know which thread is
responsible.
+ ``proxy.process.watchdog.max_awake_ms`` shows the current peak, but it is a
gauge
+ republished every half of this value, so a spike between two scrapes never
reaches your
+ metrics pipeline; ``proxy.process.watchdog.max_awake_ms_hwm`` retains the
worst one.
+
+ This value fixes the watchdog's scan period at half of it, so it is read
once at startup
+ and requires a restart to change. Every other watchdog setting, including
the other two
+ thresholds, is reloadable.
+
+ The default of this watchdog timeout is set to 0 (disabled) for ATS 10.2 for
compatibility. We recommend that administrators set a reasonable
value, such as 1000, for production configurations, in order to
catch hung plugins, or server overload scenarios.
+ See :ts:cv:`proxy.config.exec_thread.watchdog.mode` for making the watchdog
act on
+ this threshold rather than only reporting it.
+
+.. ts:cv:: CONFIG proxy.config.exec_thread.watchdog.timeout_ms INT 0
+ :units: milliseconds
+ :deprecated:
+
+ This setting is deprecated in favor of
:ts:cv:`proxy.config.exec_thread.watchdog.timeout_shrink_ms`.
+ It keeps its previous behavior: when it is non-zero and
+ :ts:cv:`proxy.config.exec_thread.watchdog.timeout_shrink_ms` is zero, the
watchdog runs and logs a
+ warning for each event loop iteration that stays awake longer than this
value, and a deprecation
+ warning naming the replacement is logged at startup. In that case the
watchdog does nothing else -
+ it publishes no ``proxy.process.watchdog`` metrics and ignores
+ :ts:cv:`proxy.config.exec_thread.watchdog.mode` and every other watchdog
setting. If
+ :ts:cv:`proxy.config.exec_thread.watchdog.timeout_shrink_ms` is non-zero,
this setting is ignored
+ and a warning is logged.
+
+.. ts:cv:: CONFIG proxy.config.exec_thread.watchdog.mode INT 0
+ :reloadable:
+
+ Select whether the watchdog acts on the peak awake span reported by
+ ``proxy.process.watchdog.max_awake_ms`` or merely reports it. This setting
has no effect
+ when :ts:cv:`proxy.config.exec_thread.watchdog.timeout_shrink_ms` is zero,
since the
+ watchdog is then not running at all.
+
+ =====
=======================================================================
+ Value Description
+ =====
=======================================================================
+ ``0`` Report only. Warnings and metrics, no change in behavior.
+ ``1`` Apply inbound backpressure while the event threads are saturated.
+ =====
=======================================================================
+
+ A change takes effect on the watchdog's next scan, at most half of
+ :ts:cv:`proxy.config.exec_thread.watchdog.timeout_shrink_ms` later. Setting
the mode to ``0``
+ releases any backpressure immediately and restores the accept queues, so it
is always
+ safe to disable under load without waiting for a scan.
+
+ Mode ``1`` offers two *independent* actions, each with its own threshold on
the same peak
+ awake span. Neither escalates into the other and neither requires the
other, so they can be
+ used together or one at a time. Requests already in flight are never
touched by either.
+
+ **Shrink the accept queues** - at
+ :ts:cv:`proxy.config.exec_thread.watchdog.timeout_shrink_ms`. |TS|
re-listens on each
+ non-exempt listen socket at a progressively smaller fraction of
+ :ts:cv:`proxy.config.net.listen_backlog` and lets the *kernel* shed the
excess, decreasing
+ multiplicatively for as long as the threads stay saturated and recovering
proportionally once
+ they do not, floored at
:ts:cv:`proxy.config.exec_thread.watchdog.backlog_min_permille`. The
+ fraction in force is published as
``proxy.process.watchdog.listen_backlog_permille`` and the
+ absolute depth as ``proxy.process.net.watchdog_listen_backlog``.
+
+ Accepting continues throughout. A bounded accept queue only drains as fast
as the acceptors
+ that read it call ``accept()``, so the kernel ends up admitting new
connections at approximately the
+ rate |TS| can actually serve - a closed loop that needs no per-connection
decision and costs
+ nothing per connection. With :ts:cv:`proxy.config.accept_threads` ``0``
every event thread accepts, so
+ the queue drains at the combined rate of the threads that are keeping up
and a single stalled
+ thread does not hold it back. It also keeps working when no event thread is
scheduled at all, which
+ a userspace refusal cannot. The connections shed this way are not refused,
though - see
+ *What shed clients see* below.
+
+ Scaling the queue dynamically also generalizes a statically tuned backlog.
A fixed low
+ :ts:cv:`proxy.config.net.listen_backlog` protects against overload but
cannot absorb a
+ legitimate burst such as a restart thundering-herd; scaling keeps a
generous queue when
+ healthy and tightens it only while saturated.
+
+ **Stop reading accepts** - at
+ :ts:cv:`proxy.config.exec_thread.watchdog.timeout_refuse_ms`, ``0`` and
therefore off by
+ default. |TS| stops taking connections off its listen sockets altogether.
Pending
+ connections stay in the kernel accept queue, and once that queue fills the
kernel drops new
+ connection attempts on its own, exactly as it does for a shrunk queue. This
costs |TS| nothing at all - no ``accept()``, no
+ descriptor, no work charged to the event threads - and a client whose
connection is picked up
+ once the threads recover sees no error whatsoever. That is the advantage
over
+ :ts:cv:`proxy.config.net.connections_throttle`, where a shed connection is
still accepted
+ and closed.
+
+ The decision to stop reading accepts is **global**, not per thread. It is
made from the peak awake
+ span across all event threads, and every acceptor consults that one value.
With
+ :ts:cv:`proxy.config.accept_threads` ``0``, where each event thread accepts
for itself, one stalled
+ thread therefore makes *every* thread stop accepting, including those that
are idle and could have
+ served the connections. Only the loopback companion listener, which is
exempt, keeps accepting.
+
+ Choose the action to match the accept configuration:
+
+ - With dedicated accept threads (:ts:cv:`proxy.config.accept_threads`
greater than ``0``), either
+ action is suitable. The accept threads are not the ones being saturated,
so pausing them keeps new
+ work off the event threads, and shrinking the queue lets them keep
draining at a bounded rate.
+ - With :ts:cv:`proxy.config.accept_threads` ``0``, prefer shrinking the
accept queues. Threads that are
+ keeping up continue to accept, so a stalled thread does not stall
admission. Use stopping accepts
+ only if halting every thread on one thread's stall is what you want.
+
+ Set the refuse threshold *above* the shrink threshold if you use both: the
cheap,
+ self-limiting control then acts first, and the blunt one only at a depth of
saturation where
+ handing the queue to the kernel has not been enough. They compose - with
both engaged the
+ queue is bounded and nothing is draining it.
+
+ **Protecting local connections.** Backpressure applies to a whole listen
socket, so a
+ local connection arriving on the same port as other traffic would be
affected along with
+ it. To avoid that, whenever the watchdog is enabled
+ |TS| additionally binds a **loopback companion listener** for each
wildcard-bound TCP
+ proxy port, and exempts only that listener from backpressure:
+
+ .. code-block:: yaml
+
+ # configured:
+ proxy.config.http.server_ports: >-
+ 80
+ 443:ssl
+
+ # additionally bound when watchdog.timeout_shrink_ms is non-zero:
+ # 127.0.0.1:80 (exempt from backpressure)
+ # 127.0.0.1:443 (exempt from backpressure)
+
+ The kernel delivers a connection to the most specific matching bind, so a
connection to
+ ``localhost`` lands on the exempt listener while everything arriving from
off box lands
+ on the original one and is subject to backpressure as normal. No change to
local
+ clients is required, and because the exemption is a property of the socket
+ rather than of an individual connection it can be honored *before*
``accept()`` - which
+ is what makes not reading accepts free. Exempt listeners are never resized.
+
+ .. important::
+
+ Two listening sockets overlapping on one port requires ``SO_REUSEPORT``
on **both**,
+ so |TS| sets it on the port's wildcard socket as well as the loopback
one whenever a
+ companion listener is opened. (``SO_REUSEADDR`` is not sufficient on
Linux.) A side
+ effect is that any process running with the same effective uid can then
bind that port
+ and receive a share of its connections. If that is unacceptable in your
environment,
+ leave :ts:cv:`proxy.config.exec_thread.watchdog.timeout_shrink_ms` at
``0`` to disable the
+ watchdog, or bind your proxy ports to explicit addresses with ``ip-in=``
so no
+ companion is opened.
+
+ Companion listeners are opened only for ports that are plain TCP and bound
to a wildcard
+ address. A port already pinned to a specific address with ``ip-in=`` was
deliberately
+ scoped, and adding a loopback bind would be exposure that was not asked
for. They are
+ opened when the watchdog is *enabled*, not when mode ``1`` is selected,
since the mode is
+ reloadable and binding sockets in response to a config change would be risk
for no
+ benefit; an exempt listener behaves identically to any other when no
backpressure is
+ engaged. A companion that fails to bind logs a warning and is skipped - it
never prevents
+ |TS| from starting.
+
+ .. note::
+
+ The companion listener ignores :ts:cv:`proxy.config.exec_thread.listen`
and always uses a
+ single socket shared by every event thread, even when the proxy port it
shadows uses one
+ listen socket per thread.
+
+ Per-thread listeners exist to remove contention on a single accept
queue, which matters at
+ high connection rates but not for the low volume of local connections.
More importantly,
+ the ``SO_REUSEPORT`` group they form selects a socket by hashing the
connection's four-tuple,
+ which is blind to what the owning thread is doing - a connection could
be steered to a thread that
+ is mid-stall while every other thread sits idle. One shared socket
behaves the opposite way:
+ ``EPOLLEXCLUSIVE`` wakes a thread that is currently blocked polling, and
a thread that is
+ behind on work is not, so the connection lands on one that is free.
+
+ This also means the companion adds exactly one socket per port rather
than one per thread.
+
+ **What shed clients see.** On Linux a full accept queue does not refuse a
new connection - it
+ *silently drops its SYN*. No SYN-ACK and no RST is sent. The client's
``connect()`` does not
+ fail; it retransmits the SYN with exponential backoff (1s, 2s, 4s, 8s, ...
about 127 seconds in
+ total with the default ``net.ipv4.tcp_syn_retries``) and either fails when
its own connect
+ timeout fires, or gets in once the queue has room and sees the wait as a
latency step of a second
+ or more. Both actions shed this way. Whether that is acceptable depends on
the clients: anything
+ with a short connect timeout reports a connection error rather than failing
over quickly, and
+ nothing gets a fast, explicit refusal.
+
+ Shrinking the backlog does not reset connections already queued beyond the
new depth. They stay
+ queued and are accepted normally; only new SYNs are dropped, until the
queue falls below the new
+ depth.
+
+ .. important::
+
+ ``net.ipv4.tcp_abort_on_overflow`` does **not** turn this shedding into
resets. It applies only
+ to a narrower race: the queue had room when the SYN arrived but filled
before the client's
+ final ACK. With ``0`` the kernel drops that ACK and retransmits the
SYN-ACK, while the client
+ already believes it is connected and may have sent its request or TLS
ClientHello; with ``1``
+ it sends an RST instead. Under sustained overload the queue is already
full when the SYN
+ arrives, so almost every shed connection takes the silent-drop path
whatever this is set to.
+
+ .. important::
+
+ Kernel-side drops are invisible to |TS|. Watch the listen-queue overflow
counters
+ (``ListenOverflows`` and ``ListenDrops``, reported by ``netstat -s`` and
``nstat``) alongside
+ ``proxy.process.net.watchdog_listen_backlog``.
+
+ .. note::
+
+ On Linux the SYN (half-open) queue limit is derived from the same
backlog, so shrinking toward
+ :ts:cv:`proxy.config.exec_thread.watchdog.backlog_min_permille` can make
the kernel start
+ sending SYN cookies (with the default ``net.ipv4.tcp_syncookies=1``)
without any real SYN
+ flood. The kernel then logs ``Possible SYN flooding on port X. Sending
cookies.``, and a
+ connection established through a cookie can lose TCP options - window
scaling and SACK when
+ TCP timestamps are not in use - and has its MSS rounded down to one of a
few fixed values. The
+ details vary by kernel version, so check against yours before relying on
a small floor.
+
+ .. note::
+
+ Resizing a listening socket's backlog by calling ``listen()`` again is
not specified by
+ POSIX. It works on Linux. Verify on your platform before relying on mode
``1`` - if the
+ resize fails, |TS| warns and the listener keeps its previous depth.
+
+ .. note::
+
+ Mode ``1`` only affects *new connections*. Requests arriving on existing
keepalive
+ connections and new HTTP/2 streams on existing sessions pass through
untouched. On a
+ workload where most requests are served over reused connections it will
engage and have
+ almost no effect on load. It is the right choice for connection-churn
overload - TLS
+ handshake storms, thundering-herd reconnects after an upstream failover
- where the cost
+ is in accepting connections rather than in serving requests.
+
+ .. note::
+
+ Neither action helps when a thread is stalled rather than busy - a
plugin blocking on a
+ synchronous DNS lookup or disk read is not going to finish sooner
because |TS| stopped
+ taking work. Compare ``proxy.process.watchdog.max_awake_ms`` against
+ :ts:stat:`proxy.process.eventloop.events.10s` before enabling mode
``1``: a high awake
+ span with a *low* event count means a stall, not load.
+
+ Mode ``1`` does not replace :ts:cv:`proxy.config.net.connections_throttle`,
which remains
+ in effect underneath. That limit protects file descriptors, a hard
resource; this is a
+ latency heuristic layered above it.
+
+.. ts:cv:: CONFIG proxy.config.exec_thread.watchdog.timeout_refuse_ms INT 0
+ :units: milliseconds
+ :reloadable:
+
+ Peak awake span at which :ts:cv:`proxy.config.exec_thread.watchdog.mode`
``1`` stops reading
+ accepts altogether - new inbound connections are left in the kernel accept
queue instead of
+ being taken off it. ``0``, the default, disables this action; the accept
queue shrink at
+ :ts:cv:`proxy.config.exec_thread.watchdog.timeout_shrink_ms` then runs on
its own. See
+ :ts:cv:`proxy.config.exec_thread.watchdog.mode` for what each action does.
+
+ This is a threshold on saturation *depth*, not a duration, and it is
independent of the shrink
+ threshold in both directions: it neither requires it to be crossed first
nor stops it acting.
+ The watchdog's scan period does not depend on this value, which is why it
is reloadable while
+ :ts:cv:`proxy.config.exec_thread.watchdog.timeout_shrink_ms` is not.
Setting it back to ``0``
+ resumes accepting immediately rather than at the next scan.
+
+ Normally set this *above*
+ :ts:cv:`proxy.config.exec_thread.watchdog.timeout_shrink_ms`. Shrinking the
queue is
+ self-limiting - |TS| keeps accepting, so inflow settles near what it can
serve - whereas not
+ reading accepts is blunt, and is worth reaching for only at a depth of
saturation where the
+ first control has visibly not been enough. Setting it below the shrink
threshold means the
+ accepts stop before the queue is ever bounded.
+
+ .. important::
+
+ Not reading accepts is only free while the queue can hold what arrives.
|TS|'s drain rate on
+ a listener in this state is exactly zero, so the accept queue can only
fill - it absorbs
+ roughly the arrival rate multiplied by the time spent above this
threshold, and past that
+ the kernel drops new SYNs at the queue's own depth, which the shrink
threshold may already
+ have reduced - see :ts:cv:`proxy.config.exec_thread.watchdog.mode` for
what those clients
+ see. On a small :ts:cv:`proxy.config.net.listen_backlog` that point
arrives quickly, and
+ while it lasts |TS| is serving nothing at all off that socket - strictly
worse than the
+ shrink action, which keeps serving at the rate it can. Size this
threshold so that reaching
+ it really is worse than not accepting.
+
+ Between engaging and releasing there is a deadband, as for every watchdog
action: accepts
+ resume only once the peak has fallen to
+ :ts:cv:`proxy.config.exec_thread.watchdog.backlog_decrease_permille` of
this value, floored at
+ half. The state is published as ``proxy.process.watchdog.accept_refuse``,
each pass that declined to
+ drain a listen socket is counted in
``proxy.process.net.watchdog_accepts_refused``, and every scan
+ over the threshold - in any mode - is counted in
+ ``proxy.process.watchdog.timeout_refuse_exceeded_count``.
+
+.. ts:cv:: CONFIG proxy.config.exec_thread.watchdog.backlog_min_permille INT 50
+ :reloadable:
+
+ Floor for the dynamic accept queue depth, in permille of
+ :ts:cv:`proxy.config.net.listen_backlog`, under
+ :ts:cv:`proxy.config.exec_thread.watchdog.mode` ``1``. The default of
``50`` is 5%.
+
+ The floor exists because a backlog of zero does not stop new connections
outright - the
+ kernel still admits at least one - while an extremely small queue drops so
many SYNs that
+ recovery is slow and bursty: every dropped client backs off for a second or
more before it
+ retries, and a small backlog also makes SYN cookies more likely (see
+ :ts:cv:`proxy.config.exec_thread.watchdog.mode`). It also guarantees the
node never becomes
+ completely unreachable through this mechanism.
+
+ Permille rather than percent throughout, because the interesting range is
compressed near the
+ bottom: on a large backlog the step from 1% to 2% is thousands of
connections, and percent
+ cannot express anything between them.
+
+ The resulting depth depends on the baseline. With
:ts:cv:`proxy.config.net.listen_backlog` left
+ at ``-1`` the baseline is the resolved system ``somaxconn``, so the same
permille yields very
+ different depths on different hosts. Set
:ts:cv:`proxy.config.net.listen_backlog` explicitly if
+ you want a predictable floor.
+
+.. ts:cv:: CONFIG proxy.config.exec_thread.watchdog.backlog_decrease_permille
INT 500
+ :reloadable:
+
+ How much of the current accept queue depth is *retained* on each saturated
scan under
+ :ts:cv:`proxy.config.exec_thread.watchdog.mode` ``1``. Lower values shed
capacity faster.
+
+ Expressed as permille retained rather than as a divisor so the whole range
is reachable, not
+ just 1/2, 1/3, 1/4. Starting from ``1000`` with a floor of ``250``:
+
+ ========= ===============================================================
+ Value Trajectory (one step per scan)
+ ========= ===============================================================
+ ``500`` 1000 → 500 → 250
+ ``667`` 1000 → 667 → 444 → 296 → 250
+ ``800`` 1000 → 800 → 640 → 512 → 409 → 327 → 261 → 250
+ ``900`` 1000 → 900 → 810 → 729 → 656 → 590 → 531 → 477 → …
+ ``975`` 1000 → 975 → 950 → 926 → 902 → 879 → 857 → 835 → …
+ ========= ===============================================================
+
+ A scan always gives up at least one point while refusing, so integer
truncation cannot stall
+ the decrease at small scales.
Review Comment:
This section is documenting
`proxy.config.exec_thread.watchdog.backlog_decrease_permille` (the
shrink/backlog decrease). The sentence 'gives up at least one point while
refusing' appears to be a copy/paste mismatch and should refer to
shrinking/decreasing the backlog scale (not refusing accepts). Updating wording
here avoids confusing shrink vs refuse behaviors.
##########
src/iocore/eventsystem/Watchdog.cc:
##########
@@ -37,73 +42,448 @@
namespace Watchdog
{
+using ts::Metrics;
+
DbgCtl dbg_ctl_watchdog("watchdog");
-Monitor::Monitor(EThread *threads[], size_t n_threads,
std::chrono::milliseconds timeout_ms)
- : _threads(threads, threads + n_threads), _timeout{timeout_ms}
+namespace
+{
+ // this ensures `long long` duration for older compilers to match format
specifiers
+ using printable_micro = std::chrono::duration<long long, std::micro>;
+ using printable_milli = std::chrono::duration<long long, std::milli>;
+
+ /** Current value of accept_refuse(). Written only by the monitor thread and
the config callback.
+
+ Read from the accept paths on every accept loop.
+ */
+ std::atomic<bool> accept_refuse_state{false};
+
+ /** Current value of proxy.config.exec_thread.watchdog.mode.
+
+ File scope rather than a Monitor member on purpose:
RecRegisterConfigUpdateCb has no
+ unregister counterpart, so a callback that captured the Monitor would
outlive it and fire
+ against freed memory at shutdown. Keeping the mode here means the
callback touches only static
+ storage and remains valid for the life of the process.
+ */
+ std::atomic<Mode> current_mode{Mode::DISABLED};
+
+ /** proxy.config.exec_thread.watchdog.timeout_refuse_ms - threshold for
accept_refuse().
+
+ Zero disables it entirely. Independent of the shrink timeout that drives
the scan period: the
+ accept paths stopping and the accept queues shrinking are separate
decisions on the same peak,
+ so this can be reloaded freely without disturbing the scan rate.
+ */
+ std::atomic<int64_t> refuse_timeout_ms{0};
+
+ /// proxy.config.exec_thread.watchdog.backlog_min_permille - floor for
backlog_scale_permille().
+ std::atomic<int32_t> backlog_min_permille{50};
+
+ /// Published listen backlog scale, in permille of the configured value.
1000 == unthrottled.
+ std::atomic<int32_t> backlog_scale{1000};
+
+ /** proxy.config.exec_thread.watchdog.backlog_decrease_permille -
multiplicative decrease.
+
+ Expressed as the permille of the *current* scale retained on each
saturated scan, rather than
+ as a divisor, so the whole range is reachable rather than just 1/2, 1/3,
1/4: 500 is the
+ classic halving, 667 is a divide by 1.5, 900 is a slow squeeze, 975 is a
crawl.
+
+ Doubles as the release deadband - see release_threshold(). One number
for how much margin is
+ wanted against the overload, applied to the queue depth and to the
signal alike.
+ */
+ std::atomic<int32_t> backlog_decrease_permille{500};
+
+ /** proxy.config.exec_thread.watchdog.backlog_increase_permille -
proportional increase.
+
+ Permille of the scale *currently in force* offered as the step on each
recovered scan, before
+ the headroom attenuation in recovery_step_permille(). Equivalently,
permille points of the
+ configured backlog at full depth, which is where the two readings
coincide and why one number
+ serves as both the rate and the ceiling on the step - the scale never
exceeds 1000, so the
+ proportional value can never exceed the configured points.
+
+ Proportional to the current depth rather than a flat number of points
because what re-saturates
+ the threads is the count of connections admitted, not the fraction of
the baseline they
+ represent, and a flat step is a fixed count whatever depth is in force.
At the bottom of a hard
+ shed that count is a multiple of the whole queue: at 100 points of a
65,535 baseline, the first
+ recovered scan after a shed to 20 permille would readmit 6,553
connections onto a queue holding
+ 1,310. The same 100 here readmits 131, and the step grows with the queue
as it climbs out.
+ */
+ std::atomic<int32_t> backlog_increase_permille{100};
+
+ int
+ update_watchdog_mode(const char * /* name */, RecDataT /* type */, RecData
data, void * /* cookie */)
+ {
+ auto mode = static_cast<Mode>(data.rec_int);
+ if (mode != Mode::DISABLED && mode != Mode::BACKPRESSURE) {
+ Warning("Watchdog: ignoring out of range
proxy.config.exec_thread.watchdog.mode %" PRId64, data.rec_int);
+ return REC_ERR_OKAY;
+ }
+
+ Mode const prev = current_mode.exchange(mode, std::memory_order_relaxed);
+ if (prev == mode) {
+ return REC_ERR_OKAY;
+ }
+
+ // Drop any backpressure the outgoing mode had engaged. The monitor
re-derives everything from
+ // the new mode on its next scan, so the worst case is up to half a
timeout of not shedding -
+ // strictly the safe direction. Leaving it set would be the unsafe one:
switching to mode 0
+ // would otherwise strand the accept queues shrunk forever.
+ //
+ // The reporting gauges are left to the monitor rather than cleared here.
It owns them, and
+ // sync_state_gauges() corrects them on its next scan; writing them from
this thread would race
+ // its stores.
+ if (mode == Mode::DISABLED) {
+ accept_refuse_state.store(false, std::memory_order_relaxed);
+ backlog_scale.store(1000, std::memory_order_relaxed);
+ }
+
+ Note("Watchdog: mode changed from %d to %d", static_cast<int>(prev),
static_cast<int>(mode));
+ return REC_ERR_OKAY;
+ }
+
+ int
+ update_refuse_timeout(const char * /* name */, RecDataT /* type */, RecData
data, void * /* cookie */)
+ {
+ int64_t const want = std::max<int64_t>(0, data.rec_int);
+ refuse_timeout_ms.store(want, std::memory_order_relaxed);
+
+ // Turning it off must resume accepting here rather than at the next scan
- the whole point of
+ // setting it to zero is to stop refusing immediately. Turning it on is
left to the monitor, which
+ // has an actual peak to compare against.
+ if (want == 0) {
+ accept_refuse_state.store(false, std::memory_order_relaxed);
+ }
+ return REC_ERR_OKAY;
+ }
+
+ int
+ update_backlog_min_permille(const char * /* name */, RecDataT /* type */,
RecData data, void * /* cookie */)
+ {
+
backlog_min_permille.store(static_cast<int32_t>(std::clamp<int64_t>(data.rec_int,
1, 1000)), std::memory_order_relaxed);
+ return REC_ERR_OKAY;
+ }
+
+ int
+ update_backlog_decrease_permille(const char * /* name */, RecDataT /* type
*/, RecData data, void * /* cookie */)
+ {
+
backlog_decrease_permille.store(static_cast<int32_t>(std::clamp<int64_t>(data.rec_int,
1, 999)), std::memory_order_relaxed);
+ return REC_ERR_OKAY;
+ }
+
+ int
+ update_backlog_increase_permille(const char * /* name */, RecDataT /* type
*/, RecData data, void * /* cookie */)
+ {
+
backlog_increase_permille.store(static_cast<int32_t>(std::clamp<int64_t>(data.rec_int,
1, 1000)), std::memory_order_relaxed);
+ return REC_ERR_OKAY;
+ }
+
+} // namespace
+
+void
+register_config_callbacks()
+{
+ RecInt mode =
RecGetRecordInt("proxy.config.exec_thread.watchdog.mode").value_or(0);
+ current_mode.store(static_cast<Mode>(mode), std::memory_order_relaxed);
Review Comment:
`register_config_callbacks()` stores the startup
`proxy.config.exec_thread.watchdog.mode` directly into `current_mode` without
validating it, while the dynamic callback validates and clamps to known values.
If an out-of-range value is present at startup, `current_mode` can be set to an
invalid enum and the watchdog gating logic can behave unpredictably until the
next config reload. Validate/clamp `mode` on startup the same way
`update_watchdog_mode()` does (and emit the same warning) before storing to
`current_mode`.
--
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]
