diff --git a/Common/Cpp/Concurrency/Backends/ThreadPool_Default.h b/Common/Cpp/Concurrency/Backends/ThreadPool_Default.h index b2944c630f..16bce27f6e 100644 --- a/Common/Cpp/Concurrency/Backends/ThreadPool_Default.h +++ b/Common/Cpp/Concurrency/Backends/ThreadPool_Default.h @@ -47,6 +47,15 @@ class ThreadPool_Default final : public ThreadPool{ } virtual WallDuration cpu_time() const override; + virtual size_t queued_tasks() const override{ + std::lock_guard lg(m_lock); + return m_queue.size(); + } + virtual size_t busy_threads() const override{ + std::lock_guard lg(m_lock); + return m_busy_count; + } + public: [[nodiscard]] virtual AsyncTask dispatch(std::function&& func) override; diff --git a/Common/Cpp/Concurrency/Backends/Thread_Qt.tpp b/Common/Cpp/Concurrency/Backends/Thread_Qt.tpp index 860c4505c7..f75e36fce7 100644 --- a/Common/Cpp/Concurrency/Backends/Thread_Qt.tpp +++ b/Common/Cpp/Concurrency/Backends/Thread_Qt.tpp @@ -10,6 +10,7 @@ #include "Common/Cpp/Containers/Pimpl.tpp" #include "Common/Cpp/Concurrency/Qt6.9ThreadBugWorkaround.h" #include "Common/Cpp/Concurrency/Thread.h" +#include "Common/Cpp/TestRunners/HangWatchdog.h" //using std::cout; //using std::endl; @@ -64,6 +65,8 @@ struct Thread::Data : public QThread{ virtual void run() override{ // cout << "starting: " << this << endl; + // Register this thread for the hang watchdog's stack dump. + HangWatchdog::register_thread(); m_function(); // m_finished.store(true, std::memory_order_acquire); // cout << "exiting: " << this << endl; diff --git a/Common/Cpp/Concurrency/Backends/Thread_StdThread.tpp b/Common/Cpp/Concurrency/Backends/Thread_StdThread.tpp index b3e42146f8..d301cfdad8 100644 --- a/Common/Cpp/Concurrency/Backends/Thread_StdThread.tpp +++ b/Common/Cpp/Concurrency/Backends/Thread_StdThread.tpp @@ -9,6 +9,7 @@ #include #include "Common/Cpp/Containers/Pimpl.tpp" #include "Common/Cpp/Concurrency/Thread.h" +#include "Common/Cpp/TestRunners/HangWatchdog.h" namespace PokemonAutomation{ @@ -46,7 +47,11 @@ Thread::Thread() = default; // - Uses CONSTRUCT_TOKEN to construct Pimpl with Data constructor // - Thread starts running immediately (same as std::thread behavior) Thread::Thread(std::function&& function) - : m_data(CONSTRUCT_TOKEN, std::move(function)) + : m_data(CONSTRUCT_TOKEN, [function = std::move(function)](){ + // Register the new thread for the hang watchdog's stack dump. + HangWatchdog::register_thread(); + function(); + }) {} void Thread::join(){ diff --git a/Common/Cpp/Concurrency/ThreadPool.h b/Common/Cpp/Concurrency/ThreadPool.h index 29ea2cf01c..61ee953465 100644 --- a/Common/Cpp/Concurrency/ThreadPool.h +++ b/Common/Cpp/Concurrency/ThreadPool.h @@ -34,6 +34,15 @@ class ThreadPool{ virtual size_t max_threads() const = 0; virtual WallDuration cpu_time() const = 0; + // Diagnostics: how many tasks are queued up waiting for a thread, and + // how many threads are currently executing a task. + virtual size_t queued_tasks() const{ + return 0; + } + virtual size_t busy_threads() const{ + return 0; + } + public: // As of this writing, tasks dispatched earlier are not allowed to block // on tasks that are dispatched later as it may cause a deadlock. diff --git a/Common/Cpp/Logging/FileLogger.cpp b/Common/Cpp/Logging/FileLogger.cpp index 76252a1f78..9a5ca6e3e1 100644 --- a/Common/Cpp/Logging/FileLogger.cpp +++ b/Common/Cpp/Logging/FileLogger.cpp @@ -4,7 +4,9 @@ * */ +#include #include +#include #include "Common/Cpp/PrettyPrint.h" #include "Common/Cpp/Filesystem/FilePath.h" #include "Common/Cpp/Filesystem/Filesystem.h" @@ -37,6 +39,23 @@ FileLogger::FileLogger(ThreadPool& thread_pool, FileLoggerConfig config) FileLogger::~FileLogger(){ stop(); } +size_t FileLogger::pending_count() const{ + // std::mutex has no try_lock_for, so poll try_lock() for up to ~500ms. + // The flusher only ever holds the lock briefly (file I/O happens outside + // of it), so this essentially never times out; the timeout is there so the + // diagnostic itself can never deadlock behind a stuck flusher. + std::unique_lock lg(m_lock, std::defer_lock); + for (int i = 0; i < 50; i++){ + if (lg.try_lock()){ + return m_queue.size(); + } + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + return (size_t)-1; +} + + + void FileLogger::stop() noexcept{ if (!m_thread){ return; diff --git a/Common/Cpp/Logging/FileLogger.h b/Common/Cpp/Logging/FileLogger.h index 3d3b9ccc71..31338d8543 100644 --- a/Common/Cpp/Logging/FileLogger.h +++ b/Common/Cpp/Logging/FileLogger.h @@ -48,6 +48,13 @@ class FileLogger : public Logger{ ~FileLogger(); void stop() noexcept; + // Diagnostics: # of log entries currently queued for writing, or + // (size_t)-1 if the lock could not be taken in time. + size_t pending_count() const; + size_t max_queue_size() const{ + return m_config.max_queue_size; + } + // Logger interface implementation virtual void log(const std::string& msg, Color color = Color()) override; virtual void log(std::string&& msg, Color color = Color()) override; diff --git a/Common/Cpp/TestRunners/HangWatchdog.cpp b/Common/Cpp/TestRunners/HangWatchdog.cpp new file mode 100644 index 0000000000..e03af1b337 --- /dev/null +++ b/Common/Cpp/TestRunners/HangWatchdog.cpp @@ -0,0 +1,720 @@ +/* Hang Watchdog + * + * From: https://github.com/PokemonAutomation/ + * + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "Common/Cpp/Concurrency/Mutex.h" +#include "Common/Cpp/Concurrency/Thread.h" +#include "HangWatchdog.h" + + +#if defined(_WIN32) + +#include +#include +#pragma comment(lib, "dbghelp.lib") + +#else + +#include +#include +#include +#include + +#if defined(__linux__) +#include +#include +#endif + +#endif + + + +namespace PokemonAutomation{ +namespace HangWatchdog{ + + + +// --------------------------------------------------------------------------- +// Raw stderr output. +// +// The dump goes to the stderr file descriptor/handle directly, not stdio: +// when output is piped (as in CI) the stdio buffers are lost when the +// process is terminated, and the app may have redirected std::cerr into its +// logger. +// --------------------------------------------------------------------------- + +static void raw_stderr_write(const char* data, size_t n){ + if (n == 0){ + return; + } + #if defined(_WIN32) + HANDLE handle = GetStdHandle(STD_ERROR_HANDLE); + if (handle != nullptr && handle != INVALID_HANDLE_VALUE){ + DWORD written = 0; + WriteFile(handle, data, (DWORD)n, &written, nullptr); + } + #else + if (::write(2, data, n) < 0){ + // Nothing more we can do. + } + #endif +} + +static void raw_stderr_line(const std::string& line){ + raw_stderr_write(line.data(), line.size()); +} + +// The id of the calling thread, for labels and for skipping ourselves. +// On Linux use the kernel tid directly: some glibc builds use an opaque +// (pointer) pthread_t, and pthread_self() is only the tid on standard +// builds. +#if !defined(_WIN32) +static uintptr_t my_thread_id(){ + #if defined(__linux__) + return (uintptr_t)syscall(SYS_gettid); + #else + return (uintptr_t)pthread_self(); + #endif +} +#endif + + + +// --------------------------------------------------------------------------- +// Thread registry - declared here, defined in the stack-dump section +// below. register_thread() calls the platform one. +// --------------------------------------------------------------------------- + +#if defined(_WIN32) +void register_thread_windows(); +#else +void register_thread_posix(); +#endif + + +// --------------------------------------------------------------------------- +// Shared state +// --------------------------------------------------------------------------- + +namespace{ + + +struct State{ + Mutex lock; + bool running = false; + WallDuration timeout = WallDuration::zero(); + WallClock start_time{}; + WallClock last_progress{}; + std::atomic completed_count{0}; + std::function state_provider; + Thread thread; +}; + +State& state(){ + static State s; + return s; +} + +// PA_HANG_WATCHDOG: integer seconds; 0 (or negative) disables the watchdog. +WallDuration configured_timeout(WallDuration default_timeout){ + const char* env = ::getenv("PA_HANG_WATCHDOG"); + if (env != nullptr){ + char* end = nullptr; + long seconds = std::strtol(env, &end, 10); + if (end != env){ + if (seconds <= 0){ + return WallDuration::zero(); + } + return std::chrono::seconds(seconds); + } + } + return default_timeout; +} + + + +} // namespace + + +void register_thread(){ + #if defined(_WIN32) + register_thread_windows(); + #else + register_thread_posix(); + #endif +} + + + +// --------------------------------------------------------------------------- +// All-thread stack dump: Windows +// +// Suspend every registered thread, walk its stack with dbghelp's +// StackWalk64, symbolize with SymFromAddr, resume it, move on. Frames come +// out as "symbol + module_offset (absolute_address)"; without PDBs the +// offset is still enough to locate the function in the binary. +// +// Threads are enumerated from the program's own registry +// (register_thread()) rather than with the toolhelp snapshot API: the +// THREADLIST32-based enumeration is not available in every SDK/MSVC +// configuration. Library-created threads (Qt, ONNX...) that the program +// never registered are not captured - noted at the end, as on macOS. +// --------------------------------------------------------------------------- + +#if defined(_WIN32) + +// Registry of thread ids created by the program (see register_thread()). +// On Windows this is the source of the dump's thread list. +static Mutex g_registry_lock_windows; +static std::vector g_registry_windows; + +namespace{ + +constexpr int kMaxFrames = 48; + +DWORD native_machine_type(){ + #if defined(_M_X64) + return IMAGE_FILE_MACHINE_AMD64; + #elif defined(_M_ARM64) + return IMAGE_FILE_MACHINE_ARM64; + #else + return IMAGE_FILE_MACHINE_I386; + #endif +} + +void dump_all_thread_stacks_windows(){ + HANDLE h_process = OpenProcess( + PROCESS_QUERY_INFORMATION | PROCESS_VM_READ, + FALSE, + GetCurrentProcessId() + ); + if (h_process == nullptr){ + raw_stderr_line("Stack dump: OpenProcess failed (error " + std::to_string((int)::GetLastError()) + ")\n"); + return; + } + SymSetOptions(SymGetOptions() | SYMOPT_UNDNAME); + SymInitialize(h_process, nullptr, TRUE); + + std::vector tids; + { + std::lock_guard lg(g_registry_lock_windows); + tids = g_registry_windows; + } + + DWORD self_tid = GetCurrentThreadId(); + DWORD machine = native_machine_type(); + int captured = 0; + + for (DWORD tid : tids){ + if (tid == self_tid){ + continue; // Don't suspend the thread that is doing the dump. + } + + HANDLE h_thread = OpenThread(THREAD_SUSPEND_RESUME | THREAD_GET_CONTEXT, FALSE, tid); + if (h_thread == nullptr){ + continue; + } + + DWORD suspend_count = SuspendThread(h_thread); + if (suspend_count == 0xFFFFFFFF){ + CloseHandle(h_thread); + continue; + } + + CONTEXT ctx; + std::memset(&ctx, 0, sizeof(ctx)); + ctx.ContextFlags = CONTEXT_FULL; + if (GetThreadContext(h_thread, &ctx)){ + STACKFRAME64 sf; + std::memset(&sf, 0, sizeof(sf)); + #if defined(_M_ARM64) || defined(_M_ARM) + // The "Machine" field only exists in the ARM STACKFRAME64. + sf.Machine = machine; + #endif + #if defined(_M_X64) || defined(_M_ARM64) + sf.AddrPC.Offset = ctx.Rip; + sf.AddrFrame.Offset = ctx.Rbp; + sf.AddrStack.Offset = ctx.Rsp; + #else + sf.AddrPC.Offset = ctx.Eip; + sf.AddrFrame.Offset = ctx.Ebx; + sf.AddrStack.Offset = ctx.Esp; + #endif + + char header[80]; + int len = std::snprintf( + header, + sizeof(header), + "\n===== Thread 0x%lx =====\n", + (unsigned long)tid + ); + if (len > 0){ + raw_stderr_write(header, (size_t)len); + } + + int depth = 0; + while (depth < kMaxFrames && + StackWalk64( + machine, + h_process, + h_thread, + &sf, + &ctx, + nullptr, + nullptr, + SymGetModuleBase64, + nullptr + ) + ){ + if (sf.AddrPC.Offset == 0){ + break; + } + + SYMBOL_INFO sym; + std::memset(&sym, 0, sizeof(sym)); + sym.SizeOfStruct = sizeof(sym); + sym.MaxNameLen = sizeof(sym.Name) - 1; + DWORD64 displacement = 0; + const char* name = "??"; + if (SymFromAddr(h_process, sf.AddrPC.Offset, &displacement, &sym)){ + name = sym.Name; + } + + DWORD64 module_base = SymGetModuleBase64(h_process, sf.AddrPC.Offset); + char line[600]; + int line_len = std::snprintf( + line, + sizeof(line), + " #%02d %s + 0x%016llX (0x%016llX)\n", + depth, + name, + (unsigned long long)(sf.AddrPC.Offset - module_base), + (unsigned long long)sf.AddrPC.Offset + ); + if (line_len > 0){ + raw_stderr_write(line, (size_t)line_len); + } + depth++; + } + if (depth == 0){ + raw_stderr_line(" (no frames captured)\n"); + } + captured++; + } + + ResumeThread(h_thread); + CloseHandle(h_thread); + } + + if (captured == 0){ + raw_stderr_line("Stack dump: no threads captured\n"); + }else{ + char note[160]; + int len = std::snprintf( + note, + sizeof(note), + "(note: %d of %d registered threads captured; library-created threads are not registered)\n", + captured, + (int)tids.size() + ); + if (len > 0){ + raw_stderr_write(note, (size_t)len); + } + } + + SymCleanup(h_process); + CloseHandle(h_process); +} + +} // namespace + +void register_thread_windows(){ + std::lock_guard lg(g_registry_lock_windows); + DWORD self = GetCurrentThreadId(); + for (DWORD tid : g_registry_windows){ + if (tid == self){ + return; + } + } + g_registry_windows.push_back(self); +} + +#else // !defined(_WIN32) + + +// --------------------------------------------------------------------------- +// All-thread stack dump: Linux/macOS +// +// Install a handler for SIGUSR2 that captures the *receiving* thread's +// backtrace() into a fixed slot, then send SIGUSR2 to every other thread +// in the process. Threads blocked in a syscall (futex wait, mutex lock, +// file I/O...) are interrupted and run the handler, so this captures +// exactly where each thread is parked. The frames are symbolized later, +// on the watchdog thread, with backtrace_symbols_fd(). +// --------------------------------------------------------------------------- + +namespace{ + +constexpr int kSignal = SIGUSR2; +constexpr int kMaxThreads = 512; +constexpr int kMaxFrames = 48; +constexpr int kWaitIterations = 200; // 200 x 10ms = up to 2s + +struct Slot{ + std::atomic done{false}; + uintptr_t tid = 0; + int n = 0; + void* frames[kMaxFrames]; +}; + +// Registry of threads created by the program: the real pthread handle plus +// the label id recorded at creation. On macOS (and on Linux builds where +// pthread_t is an opaque value, not the kernel tid) the dump signals these +// handles, because a /proc tid cannot be cast to pthread_t there. +struct RegistryEntry{ + pthread_t handle; + uintptr_t label; +}; +static Mutex g_registry_lock; +static std::vector g_registry; + +void registry_push(){ + std::lock_guard lg(g_registry_lock); + pthread_t self = pthread_self(); + uintptr_t label = my_thread_id(); + for (const auto& entry : g_registry){ + if (entry.label == label){ + return; + } + } + g_registry.emplace_back(self, label); +} + +static Slot g_slots[kMaxThreads]; +static std::atomic g_slot_count{0}; +// Per-thread re-entrancy guard. While a thread is inside the handler, +// SIGUSR2 is masked for that thread, so a second delivery to it cannot +// normally happen; this guards the pathological case. A process-wide +// guard would be wrong: it would drop captures from other threads that +// happen to run their handler at the same time. +static thread_local bool g_in_handler = false; + +static void hang_signal_handler(int){ + if (g_in_handler){ + return; + } + g_in_handler = true; + int slot = g_slot_count.fetch_add(1, std::memory_order_acq_rel); + if (slot < kMaxThreads){ + Slot& s = g_slots[slot]; + s.tid = my_thread_id(); + s.n = backtrace(s.frames, kMaxFrames); + s.done.store(true, std::memory_order_release); + } + g_in_handler = false; +} + +std::vector enumerate_thread_ids(){ + std::vector ret; + #if defined(__linux__) + // /proc/self/task lists every thread id in the process. + DIR* dir = ::opendir("/proc/self/task"); + if (dir != nullptr){ + dirent* ent = nullptr; + while ((ent = ::readdir(dir)) != nullptr){ + if (ent->d_name[0] >= '1' && ent->d_name[0] <= '9'){ + ret.push_back((uintptr_t)std::strtoul(ent->d_name, nullptr, 10)); + } + } + ::closedir(dir); + } + #else + // macOS: use the registry filled by register_thread(). + static Mutex lock; + static std::vector threads; + std::lock_guard lg(lock); + for (pthread_t t : threads){ + ret.push_back((uintptr_t)t); + } + #endif + return ret; +} + +void dump_all_thread_stacks_posix(){ + struct sigaction sa; + std::memset(&sa, 0, sizeof(sa)); + sa.sa_handler = hang_signal_handler; + sigemptyset(&sa.sa_mask); + // No SA_NODEFER: while a thread is in the handler the signal is masked + // for it, so a process-directed signal (the fallback below) is delivered + // to a *different* thread each time - which is what makes the fallback + // reach every thread in turn. + sa.sa_flags = SA_RESTART; + ::sigaction(kSignal, &sa, nullptr); + + g_slot_count.store(0, std::memory_order_relaxed); + for (auto& slot : g_slots){ + slot.done.store(false, std::memory_order_relaxed); + } + + uintptr_t self = my_thread_id(); + + // How to signal the other threads: + // + // - Standard Linux: pthread_t IS the kernel tid, so signal every thread + // listed in /proc/self/task directly - including threads created by + // libraries (Qt) that the program never registered. + // - macOS, and non-standard Linux builds where pthread_t is an opaque + // value: a /proc tid cannot be cast to pthread_t, so signal the + // threads registered via register_thread() instead. Library threads + // that did not register will not be captured (noted at the end). + bool tid_is_pthread_t = false; + #if defined(__linux__) + tid_is_pthread_t = (my_thread_id() == (uintptr_t)pthread_self()); + #endif + + int signaled = 0; + if (tid_is_pthread_t){ + for (uintptr_t tid : enumerate_thread_ids()){ + if (tid == self){ + continue; + } + if (::pthread_kill((pthread_t)tid, kSignal) == 0){ + signaled++; + } + } + }else{ + std::lock_guard lg(g_registry_lock); + for (const auto& entry : g_registry){ + if (entry.label == self){ + continue; + } + if (::pthread_kill(entry.handle, kSignal) == 0){ + signaled++; + } + } + } + int expected = signaled; + + // Wait for the handlers to run. A few threads may have exited between + // enumeration and delivery. + for (int i = 0; i < kWaitIterations && g_slot_count.load(std::memory_order_acquire) < expected; i++){ + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + + int n = g_slot_count.load(std::memory_order_acquire); + for (int c = 0; c < n; c++){ + Slot& s = g_slots[c]; + if (!s.done.load(std::memory_order_acquire)){ + continue; + } + char header[80]; + int len = std::snprintf( + header, + sizeof(header), + "\n===== Thread 0x%lx =====\n", + (unsigned long)s.tid + ); + if (len > 0){ + raw_stderr_write(header, (size_t)len); + } + // Writes one symbolized frame per line, straight to fd 2. + backtrace_symbols_fd(s.frames, s.n, 2); + } + + if (n < expected){ + char note[128]; + int len = std::snprintf( + note, + sizeof(note), + "(note: captured %d of %d threads; the rest may have exited or not received the signal)\n", + n, + expected + ); + if (len > 0){ + raw_stderr_write(note, (size_t)len); + } + } + + ::signal(kSignal, SIG_DFL); +} + +} // namespace + +void register_thread_posix(){ + registry_push(); +} + +#endif // !defined(_WIN32) + + + +// --------------------------------------------------------------------------- +// The dump + terminate +// --------------------------------------------------------------------------- + +namespace{ + + +void dump_hang(WallDuration elapsed_since_progress, WallDuration timeout){ + using std::chrono::duration; + using std::chrono::duration_cast; + + State& s = state(); + uint64_t completed = s.completed_count.load(std::memory_order_acquire); + WallClock now = current_time(); + WallClock start_time = WallClock{}; + { + std::lock_guard lg(s.lock); + start_time = s.start_time; + } + + std::ostringstream os; + os << "===================================================================\n"; + os << "PA-HANG-WATCHDOG: zero progress - no test completed for " + << duration_cast>(elapsed_since_progress).count() + << "s (timeout: " + << duration_cast>(timeout).count() + << "s)\n"; + os << "Watchdog started " + << duration_cast>(now - start_time).count() + << "s ago. Tests completed so far: " << completed << "\n"; + os << "===================================================================\n"; + raw_stderr_line(os.str()); + + // 1. Registered state: runner counts, pool states, logger queues. + std::function provider; + { + std::lock_guard lg(s.lock); + provider = s.state_provider; + } + if (provider){ + try{ + raw_stderr_line(provider()); + }catch (...){ + raw_stderr_line("(state provider threw)\n"); + } + } + + // 2. Where every thread in the process is parked. + raw_stderr_line("\n---- All-thread stack traces ----\n"); + try{ + #if defined(_WIN32) + dump_all_thread_stacks_windows(); + #else + dump_all_thread_stacks_posix(); + #endif + }catch (...){ + raw_stderr_line("(stack dump threw)\n"); + } + + raw_stderr_line( + "PA-HANG-WATCHDOG: terminating process (exit code " + std::to_string(EXIT_CODE) + ")\n" + ); + std::_Exit(EXIT_CODE); +} + +void thread_body(){ + register_thread(); // Include the watchdog thread itself in the registry. + while (true){ + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + + State& s = state(); + bool running = false; + WallDuration elapsed_since_progress = WallDuration::zero(); + WallDuration timeout = WallDuration::zero(); + { + std::lock_guard lg(s.lock); + running = s.running; + if (running){ + elapsed_since_progress = current_time() - s.last_progress; + timeout = s.timeout; + } + } + if (!running){ + return; + } + if (elapsed_since_progress >= timeout){ + dump_hang(elapsed_since_progress, timeout); + // Not reached: dump_hang() terminates the process. + } + } +} + + +} // namespace + + + +// --------------------------------------------------------------------------- +// Public API +// --------------------------------------------------------------------------- + + +void start(WallDuration no_progress_timeout){ + State& s = state(); + WallDuration effective = configured_timeout(no_progress_timeout); + { + std::lock_guard lg(s.lock); + if (s.running || effective <= WallDuration::zero()){ + return; + } + s.timeout = effective; + s.start_time = current_time(); + s.last_progress = s.start_time; + s.completed_count.store(0, std::memory_order_relaxed); + s.running = true; + } + register_thread(); // Include the calling (main) thread in the registry. + s.thread = Thread([]{ + thread_body(); + }); +} + +void stop(){ + State& s = state(); + bool was_running = false; + { + std::lock_guard lg(s.lock); + was_running = s.running; + s.running = false; + s.state_provider = nullptr; + } + if (was_running){ + // The thread exits within one 500ms tick of s.running going false. + s.thread.join(); + } +} + +void note_test_completed(){ + State& s = state(); + s.completed_count.fetch_add(1, std::memory_order_acq_rel); + { + std::lock_guard lg(s.lock); + s.last_progress = current_time(); + } +} + +void set_state_provider(std::function&& provider){ + State& s = state(); + std::lock_guard lg(s.lock); + s.state_provider = std::move(provider); +} + + + +} +} diff --git a/Common/Cpp/TestRunners/HangWatchdog.h b/Common/Cpp/TestRunners/HangWatchdog.h new file mode 100644 index 0000000000..24c1c7c7d3 --- /dev/null +++ b/Common/Cpp/TestRunners/HangWatchdog.h @@ -0,0 +1,65 @@ +/* Hang Watchdog + * + * From: https://github.com/PokemonAutomation/ + * + * A watchdog for long-running test phases (e.g., the command-line unit + * tests). If no test has COMPLETED for a period of time, it writes + * diagnostics (registered state providers + all-thread stack traces) + * straight to the stderr file descriptor and terminates the process, + * instead of leaving a hung CI runner stuck for hours. + * + * "Zero progress" is defined by the caller's use of + * note_test_completed(): it is called each time a test finishes, so the + * watchdog fires when the test suite stops completing tests - even if + * threads are still spinning or blocked. + * + * The no-progress timeout can be overridden with the environment + * variable PA_HANG_WATCHDOG (integer seconds; 0 disables the watchdog). + * + */ + +#ifndef PokemonAutomation_HangWatchdog_H +#define PokemonAutomation_HangWatchdog_H + +#include +#include +#include "Common/Cpp/Time.h" + +namespace PokemonAutomation{ +namespace HangWatchdog{ + + +// The exit code used when the watchdog terminates the process. Chosen to +// match the conventional "timed out" code so CI logs are self-explanatory. +constexpr int EXIT_CODE = 124; + + +// Start the watchdog with the given no-progress timeout. +// +// The timeout is clamped/overridden by the PA_HANG_WATCHDOG environment +// variable if it is set. Idempotent: a second call while running does +// nothing. +void start(WallDuration no_progress_timeout); + +// Stop the watchdog and wait for its thread to finish. Safe to call if it +// was never started or was already stopped. +void stop(); + +// Call each time a test completes. Resets the no-progress timer. +// Cheap: one atomic increment and one brief lock. +void note_test_completed(); + +// Register a callback that returns text to include in the dump (e.g., the +// runner's pending/running/completed counts, thread pool states, logger +// queue depth). Must not block. Cleared by stop(). +void set_state_provider(std::function&& provider); + +// Register the calling thread so it can be included in the stack dump on +// platforms without system-wide thread enumeration (macOS). +// No-op elsewhere (Linux uses /proc/self/task, Windows a toolhelp snapshot). +void register_thread(); + +} +} + +#endif diff --git a/Common/Cpp/TestRunners/ParallelUnitTestRunner.cpp b/Common/Cpp/TestRunners/ParallelUnitTestRunner.cpp index 50c8024bee..9681b784ea 100644 --- a/Common/Cpp/TestRunners/ParallelUnitTestRunner.cpp +++ b/Common/Cpp/TestRunners/ParallelUnitTestRunner.cpp @@ -9,6 +9,7 @@ #include "Common/Cpp/PrettyPrint.h" #include "Common/Cpp/MemoryUtilization/MemoryUtilization.h" #include "Common/Cpp/Concurrency/ReverseLockGuard.h" +#include "Common/Cpp/TestRunners/HangWatchdog.h" #include "ParallelUnitTestRunner.h" //#include @@ -180,6 +181,7 @@ void UnitTestRunner::dispatch_test(const std::string& name){ const UnitTest& test = *iter_current->second.test; m_current_memory += test.memory(); m_current_threads += test.threads(); + node.dispatched_at = current_time(); { ReverseLockGuard lg(m_lock); @@ -229,6 +231,9 @@ void UnitTestRunner::dispatch_test(const std::string& name){ m_current_threads -= test.threads(); } m_cv.notify_all(); + + // A completed test is forward progress for the hang watchdog. + HangWatchdog::note_test_completed(); }); } @@ -239,6 +244,28 @@ void UnitTestRunner::dispatch_test(const std::string& name){ } +std::string UnitTestRunner::diagnostics_state(){ + using std::chrono::duration; + using std::chrono::duration_cast; + + std::lock_guard lg(m_lock); + + std::ostringstream os; + os << "Runner: pending=" << m_test_by_name.size() + << " running=" << m_currently_running.size() + << " completed=" << m_completed.size() + << " current_threads=" << m_current_threads << "/" << m_max_threads + << " current_memory=" << m_current_memory << "/" << m_max_memory << " bytes" + << " cancelled=" << (cancelled() ? "yes" : "no") + << "\n"; + + WallClock now = current_time(); + for (const auto& kv : m_currently_running){ + double seconds = duration_cast>(now - kv.second.dispatched_at).count(); + os << " Running for " << seconds << "s: " << kv.first << "\n"; + } + return os.str(); +} } diff --git a/Common/Cpp/TestRunners/ParallelUnitTestRunner.h b/Common/Cpp/TestRunners/ParallelUnitTestRunner.h index 6be19a603e..66f0accb6a 100644 --- a/Common/Cpp/TestRunners/ParallelUnitTestRunner.h +++ b/Common/Cpp/TestRunners/ParallelUnitTestRunner.h @@ -15,6 +15,7 @@ #include "Common/Cpp/Concurrency/ConditionVariable.h" #include "Common/Cpp/Concurrency/AsyncTask.h" #include "Common/Cpp/Concurrency/ThreadPool.h" +#include "Common/Cpp/Time.h" #include "UnitTest.h" namespace PokemonAutomation{ @@ -77,6 +78,11 @@ class UnitTestRunner final : public CancellableScope{ virtual bool cancel(std::exception_ptr reason) noexcept override; + // A short, human-readable snapshot of the runner's state, for + // diagnostics (e.g., the hang watchdog's dump). Never logs. + std::string diagnostics_state(); + + private: void dispatch_test(const std::string& name); @@ -102,6 +108,7 @@ class UnitTestRunner final : public CancellableScope{ struct DispatchedEntry{ std::shared_ptr test; AsyncTask task; + WallClock dispatched_at; }; std::map m_completed; diff --git a/SerialPrograms/Source/ComputerPrograms/ComputerPrograms_UnitTestRunner.cpp b/SerialPrograms/Source/ComputerPrograms/ComputerPrograms_UnitTestRunner.cpp index 8448f23aad..11676b3b71 100644 --- a/SerialPrograms/Source/ComputerPrograms/ComputerPrograms_UnitTestRunner.cpp +++ b/SerialPrograms/Source/ComputerPrograms/ComputerPrograms_UnitTestRunner.cpp @@ -4,11 +4,15 @@ * */ +#include +#include + #include "Common/Cpp/ListenerSet.h" #include "Common/Cpp/MoveOnlyFunction_Tests.h" #include "Common/Cpp/ScopeExit.h" #include "Common/Cpp/PrettyPrint.h" #include "Common/Cpp/Sockets/Sockets_Tests.h" +#include "Common/Cpp/TestRunners/HangWatchdog.h" #include "Common/Cpp/TestRunners/TestImageDistortionManager.h" #include "Common/Cpp/TestRunners/UnitTestDatabase.h" #include "Common/Qt/QtThreadPool_Tests.h" @@ -18,6 +22,7 @@ #include "CommonFramework/AudioPipeline/SinkPipeline_Tests.h" #include "CommonFramework/AudioPipeline/IO/AudioFileLoader_Tests.h" #include "CommonFramework/AudioPipeline/Spectrum/FFTStreamer.h" +#include "Common/Cpp/Logging/FileLogger.h" #include "CommonFramework/GlobalAutoPaths.h" #include "CommonFramework/ProgramStats/StatsTracking.h" #include "CommonFramework/Tools/GlobalThreadPools.h" @@ -40,6 +45,10 @@ #include "ComputerPrograms_UnitTestRunner.h" namespace PokemonAutomation{ + +// Defined in Main.cpp; no header declares it. +FileLogger& global_file_logger(); + namespace ComputerPrograms{ @@ -384,6 +393,41 @@ bool CommandLineUnitTestRunner::run(bool augment_color_distortions){ for (const auto& test : UNIT_TESTS_ALL()){ runner.add_test(test.second); } + + // Hang watchdog: if no test COMPLETES for 60s, dump the runner, pool and + // logger state plus the stack trace of every thread in the process, + // straight to the stderr file descriptor, then terminate (exit 124). + // This turns a hung CI runner - which would otherwise sit until the 8h + // job timeout - into a failed job with a full diagnostic dump. + // PA_HANG_WATCHDOG overrides the timeout in seconds (0 disables). + HangWatchdog::set_state_provider([&runner]{ + std::ostringstream os; + os << runner.diagnostics_state(); + auto pool_line = [](const char* label, ThreadPool& pool){ + std::ostringstream line; + line << "Pool [" << label << "]: threads=" << pool.current_threads() + << " max=" << pool.max_threads() + << " queued=" << pool.queued_tasks() + << " busy=" << pool.busy_threads(); + return line.str(); + }; + os << pool_line("unlimited_normal", GlobalThreadPools::unlimited_normal()) << "\n"; + os << pool_line("computation_normal", GlobalThreadPools::computation_normal()) << "\n"; + size_t pending = global_file_logger().pending_count(); + os << "FileLogger: pending="; + if (pending == (size_t)-1){ + os << "(lock busy)"; + }else{ + os << pending; + } + os << " max_queue=" << global_file_logger().max_queue_size() << "\n"; + return os.str(); + }); + HangWatchdog::start(std::chrono::seconds(60)); + ScopeExit stop_watchdog_on_exit([]{ + HangWatchdog::stop(); + }); + runner.run(); m_logger.log( diff --git a/SerialPrograms/cmake/SourceFiles.cmake b/SerialPrograms/cmake/SourceFiles.cmake index eef6a10969..365acf2306 100644 --- a/SerialPrograms/cmake/SourceFiles.cmake +++ b/SerialPrograms/cmake/SourceFiles.cmake @@ -178,6 +178,8 @@ file(GLOB CORE_LIBRARY_SOURCES ../Common/Cpp/Strings/StringTools.h ../Common/Cpp/Strings/Unicode.cpp ../Common/Cpp/Strings/Unicode.h + ../Common/Cpp/TestRunners/HangWatchdog.cpp + ../Common/Cpp/TestRunners/HangWatchdog.h ../Common/Cpp/Time.cpp ../Common/Cpp/Time.h ../Common/Cpp/UiWrapper.h