watchdog: alert on ET_NET thread stalls beyond threshold

doc/admin-guide/files/records.yaml.en.rst

@@ -419,6 +419,18 @@ 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
+   :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
+   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.
+
 .. ts:cv:: CONFIG proxy.config.system.file_max_pct FLOAT 0.9
 
    Set the maximum number of file handles for the traffic_server process as a percentage of the fs.file-max proc value in Linux. The default is 90%.

include/iocore/eventsystem/EThread.h

@@ -33,6 +33,7 @@
 #include "iocore/eventsystem/PriorityEventQueue.h"
 #include "iocore/eventsystem/ProtectedQueue.h"
 #include "tsutil/Histogram.h"
+#include "iocore/eventsystem/Watchdog.h"
 
 #if TS_USE_HWLOC
 struct hwloc_obj;
@@ -584,6 +585,8 @@ class EThread : public Thread
 
   Metrics metrics;
 
+  Watchdog::Heartbeat heartbeat_state;
+
 private:
   void cons_common();
 };

include/iocore/eventsystem/Watchdog.h

@@ -0,0 +1,70 @@
+/** @file
+
+  A watchdog for event loops
+
+  Each event thread advertises its current state through a lightweight
+  "heartbeat" struct: the thread publishes the timestamps for the most recent
+  sleep/wake pair along with a monotonically increasing sequence number.
+  `Watchdog::Monitor`, started from `traffic_server.cc`, runs in its own
+  `std::thread` and periodically scans those heartbeats; if a thread has been
+  awake longer than the configured timeout it emits a warning (timeout values
+  come from `proxy.config.exec_thread.watchdog.timeout_ms`, where 0 disables
+  the monitor).  The monitor never touches event-system locks, keeping the
+  runtime overhead in the hot loop confined to a handful of atomic updates.
+
+  @section license License
+
+  Licensed to the Apache Software Foundation (ASF) under one
+  or more contributor license agreements.  See the NOTICE file
+  distributed with this work for additional information
+  regarding copyright ownership.  The ASF licenses this file
+  to you under the Apache License, Version 2.0 (the
+  "License"); you may not use this file except in compliance
+  with the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+
+ */
+
+#pragma once
+
+#include <atomic>
+#include <chrono>
+#include <vector>
+#include <thread>
+
+class EThread;
+
+namespace Watchdog
+{
+struct Heartbeat {
+  std::atomic<std::chrono::time_point<std::chrono::steady_clock>> last_sleep{
+    std::chrono::steady_clock::time_point::min()}; // set right before sleeping (e.g. before epoll_wait)
+  std::atomic<std::chrono::time_point<std::chrono::steady_clock>> last_wake{
+    std::chrono::steady_clock::time_point::min()}; // set right after waking from sleep (e.g. epoll_wait returns)
+  std::atomic<uint64_t> seq{0};                    // increment on each loop - used to deduplicate warnings
+  std::atomic<uint64_t> warned_seq{0};             // last seq we logged a warning about
+};
+
+class Monitor
+{
+public:
+  explicit Monitor(EThread *threads[], size_t n_threads, std::chrono::milliseconds timeout_ms);
+  ~Monitor();
+  Monitor() = delete;
+
+private:
+  const std::vector<EThread *>    _threads;
+  std::thread                     _watchdog_thread;
+  const std::chrono::milliseconds _timeout;
+  std::atomic<bool>               _shutdown = false;
+  void                            monitor_loop() const;
+};
+
+} // namespace Watchdog

src/iocore/eventsystem/CMakeLists.txt

@@ -35,6 +35,7 @@ add_library(
   ConfigProcessor.cc
   RecRawStatsImpl.cc
   RecProcess.cc
+  Watchdog.cc
 )
 add_library(ts::inkevent ALIAS inkevent)
 

src/iocore/eventsystem/UnixEThread.cc

@@ -304,8 +304,17 @@ EThread::execute_regular()
     ink_hrtime post_drain  = ink_get_hrtime();
     ink_hrtime drain_queue = post_drain - loop_start_time;
 
+    // watchdog kick - pre-sleep
+    // Relaxed store because this EThread is the only writer and the watchdog only needs a coherent timestamp.
+    this->heartbeat_state.last_sleep.store(std::chrono::steady_clock::now(), std::memory_order_relaxed);
+
     tail_cb->waitForActivity(sleep_time);
 
+    // watchdog kick - post-wake
+    // Relaxed store/fetch because the monitor thread is the single reader and per-field coherence is sufficient.
+    this->heartbeat_state.last_wake.store(std::chrono::steady_clock::now(), std::memory_order_relaxed);
+    this->heartbeat_state.seq.fetch_add(1, std::memory_order_relaxed);
+
     // loop cleanup
     loop_finish_time = ink_get_hrtime();
     // @a delta can be negative due to time of day adjustments (which apparently happen quite frequently). I

src/iocore/eventsystem/Watchdog.cc

@@ -0,0 +1,104 @@
+/** @file
+
+  A watchdog for event loops
+
+  @section license License
+
+  Licensed to the Apache Software Foundation (ASF) under one
+  or more contributor license agreements.  See the NOTICE file
+  distributed with this work for additional information
+  regarding copyright ownership.  The ASF licenses this file
+  to you under the Apache License, Version 2.0 (the
+  "License"); you may not use this file except in compliance
+  with the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+
+ */
+
+#include "iocore/eventsystem/Watchdog.h"
+#include "iocore/eventsystem/EThread.h"
+#include "tscore/Diags.h"
+#include "tscore/ink_assert.h"
+#include "tscore/ink_thread.h"
+#include "tsutil/DbgCtl.h"
+
+#include <atomic>
+#include <chrono>
+#include <thread>
+#include <functional>
+
+namespace Watchdog
+{
+
+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}
+{
+  // Precondition: 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);
+  _watchdog_thread = std::thread(std::bind_front(&Monitor::monitor_loop, this));
+}
+
+Monitor::~Monitor()
+{
+  _shutdown.store(true, std::memory_order_release);
+  _watchdog_thread.join();
+}
+
+void
+Monitor::monitor_loop() const
+{
+  // Divide by a floating point 2 to avoid truncation to zero.
+  auto sleep_time = _timeout / 2.0;
+  ink_release_assert(sleep_time.count() > 0);
+  Dbg(dbg_ctl_watchdog, "Starting watchdog with timeout %" PRIu64 " ms on %zu threads.  sleep_time = %" PRIu64 " us",
+      _timeout.count(), _threads.size(), std::chrono::duration_cast<std::chrono::microseconds>(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);
+
+      if (last_wake == std::chrono::steady_clock::time_point::min() || last_wake < last_sleep) {
+        // not yet woken from last sleep
+        continue;
+      }
+
+      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 %" PRIu64 " ms", i,
+                  std::chrono::duration_cast<std::chrono::milliseconds>(awake_duration).count());
+          t->heartbeat_state.warned_seq.store(seq, std::memory_order_relaxed);
+        }
+      }
+    }
+
+    std::this_thread::sleep_for(sleep_time);
+  }
+  Dbg(dbg_ctl_watchdog, "Stopping watchdog");
+}
+} // namespace Watchdog

src/records/RecordsConfig.cc

@@ -1529,7 +1529,12 @@ static constexpr RecordElement RecordsConfig[] =
   {RECT_CONFIG, "proxy.config.io_uring.wq_workers_unbounded", RECD_INT, "0", RECU_NULL, RR_NULL, RECC_NULL, nullptr, RECA_NULL},
   {RECT_CONFIG, "proxy.config.aio.mode", RECD_STRING, "auto", RECU_DYNAMIC, RR_NULL, RECC_STR, "(auto|io_uring|thread)", RECA_NULL},
 #endif
-
+  //###########
+  //#
+  //# Thread watchdog
+  //#
+  //###########
+  {RECT_CONFIG, "proxy.config.exec_thread.watchdog.timeout_ms", RECD_INT, "0", RECU_RESTART_TS, RR_NULL, RECC_INT, "[0-10000]", RECA_NULL}
 };
 // clang-format on
 

src/traffic_server/traffic_server.cc

@@ -32,6 +32,7 @@
 
 #include "iocore/aio/AIO.h"
 #include "iocore/cache/Store.h"
+#include "iocore/eventsystem/Watchdog.h"
 #include "tscore/TSSystemState.h"
 #include "tscore/Version.h"
 #include "tscore/ink_platform.h"
@@ -213,6 +214,8 @@ int cmd_block = 0;
 // -1: cache is already initialized, don't delay.
 int delay_listen_for_cache = 0;
 
+std::unique_ptr<Watchdog::Monitor> watchdog = nullptr;
+
 ArgumentDescription argument_descriptions[] = {
   {"net_threads",       'n', "Number of Net Threads",                                                                 "I",     &num_of_net_threads,             "PROXY_NET_THREADS",       nullptr                    },
   {"udp_threads",       'U', "Number of UDP Threads",                                                                 "I",     &num_of_udp_threads,             "PROXY_UDP_THREADS",       nullptr                    },
@@ -266,6 +269,9 @@ struct AutoStopCont : public Continuation {
   int
   mainEvent(int /* event */, Event * /* e */)
   {
+    // Stop the watchdog before shutting threads down
+    watchdog.reset();
+
     TSSystemState::stop_ssl_handshaking();
 
     APIHook *hook = g_lifecycle_hooks->get(TS_LIFECYCLE_SHUTDOWN_HOOK);
@@ -2131,6 +2137,14 @@ main(int /* argc ATS_UNUSED */, const char **argv)
   RecRegisterConfigUpdateCb("proxy.config.dump_mem_info_frequency", init_memory_tracker, nullptr);
   init_memory_tracker(nullptr, RECD_NULL, RecData(), nullptr);
 
+  // Start the watchdog
+  int watchdog_timeout_ms = RecGetRecordInt("proxy.config.exec_thread.watchdog.timeout_ms").value_or(0);
+  if (watchdog_timeout_ms > 0) {
+    watchdog = std::make_unique<Watchdog::Monitor>(eventProcessor.thread_group[ET_NET]._thread,
+                                                   static_cast<size_t>(eventProcessor.thread_group[ET_NET]._count),
+                                                   std::chrono::milliseconds{watchdog_timeout_ms});
+  }
+
   {
     auto s{RecGetRecordStringAlloc("proxy.config.diags.debug.client_ip")};
     if (auto p{ats_as_c_str(s)}; p) {