watchdog: alert on ET_NET thread stalls beyond threshold

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,60 @@
+/** @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.
+
+ */
+
+#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,15 @@ EThread::execute_regular()
     ink_hrtime post_drain  = ink_get_hrtime();
     ink_hrtime drain_queue = post_drain - loop_start_time;
 
+    // watchdog kick - pre-sleep
+    this->heartbeat_state.last_sleep.store(std::chrono::steady_clock::now(), std::memory_order_relaxed);
+
     tail_cb->waitForActivity(sleep_time);
 
+    // watchdog kick - post-wake
+    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,95 @@
+/** @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 "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), _watchdog_thread{std::bind_front(&Monitor::monitor_loop, this)}, _timeout{timeout_ms}
+{
+  ink_assert(timeout_ms.count() > 0);
+}
+
+Monitor::~Monitor()
+{
+  _shutdown.store(true);
+  _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());
+
+  while (!_shutdown.load()) {
+    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];
+      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;
+      }
+      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) {
+        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
+          Error("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.thread_watchdog.timeout_ms", RECD_INT, "1000", RECU_RESTART_TS, RR_NULL, RECC_INT, "[1-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,12 @@ 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.thread_watchdog.timeout_ms").value_or(1000);
+  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) {