From 8b8e723fe09feb31000e9bfb5c67e7edaf809432 Mon Sep 17 00:00:00 2001 From: Boris Batkin Date: Sat, 5 Sep 2026 19:41:04 -0700 Subject: [PATCH 01/14] fio: a long-lived child process - spawn_process, process_drain/poll/wait/terminate/kill/pid/alive, close_process - the primitive a supervisor drives on its own clock popen_argv is block-scoped and blocks to EOF; a supervisor needs a handle it can poll and drain on a 250 ms tick, signal, and reap. Nine externs in fio_core next to popen_argv, an opaque `SubProcess?` handle, and a `process_running` sentinel that process_poll / process_wait answer while the child is alive. - spawn_process(argv, cwd, env): no shell, forward-slash argv[0] on every platform; `cwd` empty inherits; `env` is KEY=VALUE overrides on the inherited environment. stdout+stderr merge into one non-blocking read end. Windows: kill-on-close job object, exe resolved from the parent's directory as CreateProcess does. POSIX: the child leads its own process group so killpg reaches the tree; a relative argv[0] naming a path is made absolute before the child chdir's, matching Windows. - process_drain(p, blk): fires blk once per complete line ready this tick (partial lines buffer across calls, a final unterminated line flushes at EOF); returns false once stdout is closed. Never blocks the tick. - process_terminate / process_kill: TerminateJobObject 15 / 9 on Windows, killpg SIGTERM / SIGKILL on POSIX. - close_process: frees the handle; a still-running child dies with it (job close on Windows, SIGKILL + reap on POSIX), so nothing leaks or zombies. - daslib/fio: `with_process(argv[, cwd, env]) <| $(var p) {}` RAII over spawn/close; `process` typedef. tests/fio/test_process.das drives a daslang child through the whole lifecycle - env and cwd observed in its output, lines drained one per call, running while parked, the real exit code from wait, alive before and dead after - and terminates a parked one. The fixture returns its code from main: a das `exit()` unwinds as an abnormal termination and the CLI reports 1. Co-Authored-By: Claude Fable 5.1 --- daslib/fio.das | 17 + include/daScript/simulate/aot_builtin_fio.h | 12 + src/builtin/module_builtin_fio.cpp | 358 ++++++++++++++++++++ tests/fio/_fixture_process_child.das | 32 ++ tests/fio/test_process.das | 106 ++++++ 5 files changed, 525 insertions(+) create mode 100644 tests/fio/_fixture_process_child.das create mode 100644 tests/fio/test_process.das diff --git a/daslib/fio.das b/daslib/fio.das index cde51d9df6..b8a1222c2c 100644 --- a/daslib/fio.das +++ b/daslib/fio.das @@ -690,6 +690,23 @@ def rmdir_rec_result(path : string) : fs_result_bool { return fs_result_bool(value = res) } +typedef process = SubProcess? + +def with_process(argv : array; cwd : string; env : array; blk : block<(var p : process) : void>) { + //! Spawn `argv` as a long-lived child (no shell; a forward-slash ``argv[0]`` spawns everywhere), run + //! `blk` with the live handle, then close it on scope exit - closing kills a child still running (the + //! Windows job tree, the POSIX group). `cwd` empty inherits the parent's; `env` is ``KEY=VALUE`` overrides. + var p = unsafe(spawn_process(argv, cwd, env)) + invoke(blk, p) + unsafe(close_process(p)) +} + +def with_process(argv : array; blk : block<(var p : process) : void>) { + //! `with_process` inheriting the parent's directory and environment. + let noenv : array + with_process(argv, "", noenv, blk) +} + def run_and_capture(args : array; var output : string&; timeout_sec : float = 0.0) : int { //! Run an external command and capture its stdout+stderr (merged into one pipe by the underlying ``popen_argv``). Returns the process exit code; //! -1 means the spawn itself failed. No shell is involved, and a forward-slash ``args[0]`` spawns on every platform (``popen_argv`` hands Windows the backslash spelling). diff --git a/include/daScript/simulate/aot_builtin_fio.h b/include/daScript/simulate/aot_builtin_fio.h index d13f719cc4..bea3d7dc06 100644 --- a/include/daScript/simulate/aot_builtin_fio.h +++ b/include/daScript/simulate/aot_builtin_fio.h @@ -115,6 +115,18 @@ namespace das { DAS_API bool builtin_spawn_argv ( const Array & args_arr, Context * context, LineInfoArg * at ); DAS_API int builtin_popen_argv ( const Array & args_arr, float timeout_sec, const TBlock & blk, Context * context, LineInfoArg * at ); DAS_API int builtin_popen_argv_pipe ( const Array & args_arr, const TBlock & blk, Context * context, LineInfoArg * at ); + // A long-lived child process: spawned once, polled and drained across many ticks, unlike the + // block-scoped popen_argv. The handle (das `SubProcess?`) is opaque; free it with close_process. + struct DasSubProcess; + DAS_API DasSubProcess * builtin_spawn_process ( const Array & argv, const char * cwd, const Array & env, Context * context, LineInfoArg * at ); + DAS_API bool builtin_process_drain ( DasSubProcess * p, const TBlock & blk, Context * context, LineInfoArg * at ); + DAS_API int builtin_process_poll ( DasSubProcess * p, Context * context, LineInfoArg * at ); + DAS_API int builtin_process_wait ( DasSubProcess * p, float timeout_sec, Context * context, LineInfoArg * at ); + DAS_API void builtin_process_terminate ( DasSubProcess * p, Context * context, LineInfoArg * at ); + DAS_API void builtin_process_kill ( DasSubProcess * p, Context * context, LineInfoArg * at ); + DAS_API int builtin_process_pid ( DasSubProcess * p, Context * context, LineInfoArg * at ); + DAS_API bool builtin_process_alive ( int32_t pid, Context * context, LineInfoArg * at ); + DAS_API void builtin_close_process ( DasSubProcess * p, Context * context, LineInfoArg * at ); DAS_API char * get_full_file_name ( const char * path, Context * context, LineInfoArg * ); DAS_API char * builtin_resolve_this_module_dir ( const char * baked_path, bool standalone, Context * context ); DAS_API bool builtin_remove_file ( const char * path ); diff --git a/src/builtin/module_builtin_fio.cpp b/src/builtin/module_builtin_fio.cpp index 80dfc48d13..94729d24a4 100644 --- a/src/builtin/module_builtin_fio.cpp +++ b/src/builtin/module_builtin_fio.cpp @@ -20,6 +20,9 @@ #include #define DAS_POPEN_TIMEOUT 0x7FFFFF01 +// process_poll / process_wait return this while the child is still running (INT32_MIN, so it +// never collides with a real exit code or signal number). +#define DAS_PROCESS_RUNNING (-2147483647-1) MAKE_TYPE_FACTORY(clock, das::Time)// use MAKE_TYPE_FACTORY out of namespace. Some compilers not happy otherwise @@ -235,6 +238,15 @@ namespace das { char * builtin_fs_create_temp_directory ( const char * prefix, char * & error, Context * context, LineInfoArg * at ) GENERATE_IO_STUB int builtin_popen_argv ( const Array & args_arr, float timeout_sec, const TBlock & blk, Context * context, LineInfoArg * at ) GENERATE_IO_STUB int builtin_popen_argv_pipe ( const Array & args_arr, const TBlock & blk, Context * context, LineInfoArg * at ) GENERATE_IO_STUB + DasSubProcess * builtin_spawn_process ( const Array & argv, const char * cwd, const Array & env, Context * context, LineInfoArg * at ) GENERATE_IO_STUB + bool builtin_process_drain ( DasSubProcess * p, const TBlock & blk, Context * context, LineInfoArg * at ) GENERATE_IO_STUB + int builtin_process_poll ( DasSubProcess * p, Context * context, LineInfoArg * at ) GENERATE_IO_STUB + int builtin_process_wait ( DasSubProcess * p, float timeout_sec, Context * context, LineInfoArg * at ) GENERATE_IO_STUB + void builtin_process_terminate ( DasSubProcess * p, Context * context, LineInfoArg * at ) GENERATE_IO_STUB + void builtin_process_kill ( DasSubProcess * p, Context * context, LineInfoArg * at ) GENERATE_IO_STUB + int builtin_process_pid ( DasSubProcess * p, Context * context, LineInfoArg * at ) GENERATE_IO_STUB + bool builtin_process_alive ( int32_t pid, Context * context, LineInfoArg * at ) GENERATE_IO_STUB + void builtin_close_process ( DasSubProcess * p, Context * context, LineInfoArg * at ) GENERATE_IO_STUB void * register_dynamic_module_silent ( const char * path, const char * mod_name, Context * context, LineInfoArg * at ) GENERATE_IO_STUB void for_each_registered_native_path ( const TBlock & block, Context * context, LineInfoArg * at ) GENERATE_IO_STUB void for_each_registered_dynamic_module ( const TBlock & block, Context * context, LineInfoArg * at ) GENERATE_IO_STUB @@ -317,6 +329,7 @@ namespace das { #include #include #include +#include // errno for non-blocking process_drain namespace das { void builtin_sleep ( uint32_t msec ) { @@ -1707,6 +1720,320 @@ namespace das { #endif } + // A long-lived child: spawn once, then poll / drain / signal across many ticks. popen_argv + // is block-scoped and blocks to EOF; this hands back an opaque handle a supervisor drives on + // its own clock. The read end (stdout+stderr merged) is non-blocking so drain never stalls + // the tick; on Windows the child sits in a kill-on-close job object and its group is signalled + // as a tree, on POSIX the child leads its own process group and killpg reaches the tree. + // process_poll / process_wait answer DAS_PROCESS_RUNNING while the child is still alive. + struct DasSubProcess { +#ifdef _WIN32 + HANDLE hProcess = nullptr; + HANDLE hJob = nullptr; + HANDLE hRead = INVALID_HANDLE_VALUE; + DWORD pid = 0; +#else + pid_t pid = -1; + int fd = -1; +#endif + std::string buf; // partial-line accumulator across drains + bool stdoutOpen = true; + bool reaped = false; + int exitCode = 0; + }; + +#ifdef _WIN32 + static string winBuildEnvBlock ( const Array & env ) { + vector entries; + LPCH base = GetEnvironmentStringsA(); + if ( base ) { + for ( LPCH e = base; *e; e += strlen(e) + 1 ) entries.emplace_back(e); + FreeEnvironmentStringsA(base); + } + char ** ov = (char **) env.data; + for ( uint64_t i = 0; i < env.size; ++i ) { + if ( !ov[i] ) continue; + string entry = ov[i]; + size_t eq = entry.find('='); + string key = eq == string::npos ? entry : entry.substr(0, eq); + for ( auto & e : entries ) { // replace an existing key (case-insensitive) + size_t k = e.find('='); + string ek = k == string::npos ? e : e.substr(0, k); + if ( ek.size() == key.size() && _stricmp(ek.c_str(), key.c_str()) == 0 ) { e.clear(); break; } + } + entries.push_back(entry); + } + string block; + for ( auto & e : entries ) { if ( e.empty() ) continue; block.append(e); block.push_back('\0'); } + block.push_back('\0'); // the block ends in a second NUL + return block; + } +#endif + + DasSubProcess * builtin_spawn_process ( const Array & argv_arr, const char * cwd, const Array & env, + Context * context, LineInfoArg * at ) { + if ( argv_arr.size == 0 ) { + context->throw_error_at(at, "spawn_process with empty argv"); + return nullptr; + } + char ** argv = (char **) argv_arr.data; + if ( !argv[0] ) { + context->throw_error_at(at, "spawn_process with null exe"); + return nullptr; + } + bool hasCwd = cwd && *cwd; +#ifdef _WIN32 + SECURITY_ATTRIBUTES sa; + sa.nLength = sizeof(SECURITY_ATTRIBUTES); + sa.bInheritHandle = TRUE; + sa.lpSecurityDescriptor = NULL; + HANDLE hRead = NULL, hWrite = NULL; + if ( !CreatePipe(&hRead, &hWrite, &sa, 0) ) { + context->throw_error_at(at, "spawn_process: CreatePipe failed"); + return nullptr; + } + SetHandleInformation(hRead, HANDLE_FLAG_INHERIT, 0); // parent's read end stays in-process + HANDLE hNull = CreateFileA("NUL", GENERIC_READ, FILE_SHARE_READ | FILE_SHARE_WRITE, &sa, + OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL, NULL); + HANDLE hJob = CreateJobObjectA(NULL, NULL); + if ( hJob ) { + JOBOBJECT_EXTENDED_LIMIT_INFORMATION jeli; + memset(&jeli, 0, sizeof(jeli)); + jeli.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE; + SetInformationJobObject(hJob, JobObjectExtendedLimitInformation, &jeli, sizeof(jeli)); + } + STARTUPINFOA si; + memset(&si, 0, sizeof(si)); + si.cb = sizeof(si); + si.dwFlags = STARTF_USESTDHANDLES; + si.hStdInput = (hNull == INVALID_HANDLE_VALUE) ? NULL : hNull; + si.hStdOutput = hWrite; + si.hStdError = hWrite; + string cmdLine = winBuildCommandLine(argv, argv_arr.size); + string envBlock; + LPVOID lpEnv = NULL; + if ( env.size ) { envBlock = winBuildEnvBlock(env); lpEnv = (LPVOID)&envBlock[0]; } + PROCESS_INFORMATION pi; + memset(&pi, 0, sizeof(pi)); + BOOL ok = CreateProcessA(NULL, (LPSTR)cmdLine.c_str(), NULL, NULL, TRUE, + CREATE_NO_WINDOW | CREATE_SUSPENDED, lpEnv, hasCwd ? cwd : NULL, &si, &pi); + CloseHandle(hWrite); + if ( hNull != INVALID_HANDLE_VALUE ) CloseHandle(hNull); + if ( !ok ) { + CloseHandle(hRead); + if ( hJob ) CloseHandle(hJob); + context->throw_error_at(at, "spawn_process: CreateProcess failed"); + return nullptr; + } + if ( hJob ) AssignProcessToJobObject(hJob, pi.hProcess); + ResumeThread(pi.hThread); + CloseHandle(pi.hThread); + DasSubProcess * p = new DasSubProcess(); + p->hProcess = pi.hProcess; + p->hJob = hJob; + p->hRead = hRead; + p->pid = pi.dwProcessId; + return p; +#else + vector cargv; + cargv.reserve(argv_arr.size + 1); + for ( uint64_t i = 0; i < argv_arr.size; ++i ) cargv.push_back(argv[i] ? argv[i] : (char *)""); + cargv.push_back(nullptr); + // A relative argv[0] that names a path (has a '/') resolves against the caller's directory, + // not the child's cwd - so make it absolute before the child chdir's, matching Windows, where + // CreateProcess already searches the exe from the parent's directory rather than lpCurrentDirectory. + // A bare name (no '/') is a PATH lookup, which chdir does not affect - leave it alone. + string absExe; + if ( hasCwd && cargv[0][0] && cargv[0][0] != '/' && strchr(cargv[0], '/') ) { + char cwdbuf[4096]; + if ( getcwd(cwdbuf, sizeof(cwdbuf)) ) { + absExe = string(cwdbuf) + "/" + cargv[0]; + cargv[0] = (char *)absExe.c_str(); + } + } + int pipefd[2]; + if ( pipe(pipefd) == -1 ) { + context->throw_error_at(at, "spawn_process: pipe failed"); + return nullptr; + } + char ** ov = (char **) env.data; + pid_t pid = fork(); + if ( pid == -1 ) { + close(pipefd[0]); + close(pipefd[1]); + context->throw_error_at(at, "spawn_process: fork failed"); + return nullptr; + } + if ( pid == 0 ) { + close(pipefd[0]); + int devnull = open("/dev/null", O_RDONLY); + if ( devnull >= 0 ) { dup2(devnull, STDIN_FILENO); close(devnull); } + dup2(pipefd[1], STDOUT_FILENO); + dup2(pipefd[1], STDERR_FILENO); + close(pipefd[1]); + setpgid(0, 0); // lead a group so killpg reaches the tree + if ( hasCwd && chdir(cwd) != 0 ) _exit(127); + for ( uint64_t i = 0; i < env.size; ++i ) if ( ov[i] ) putenv(strdup(ov[i])); + execvp(cargv[0], cargv.data()); + _exit(127); + } + close(pipefd[1]); + fcntl(pipefd[0], F_SETFL, O_NONBLOCK); // drain never blocks the tick + DasSubProcess * p = new DasSubProcess(); + p->pid = pid; + p->fd = pipefd[0]; + return p; +#endif + } + + bool builtin_process_drain ( DasSubProcess * p, const TBlock & blk, + Context * context, LineInfoArg * at ) { + if ( !p ) { context->throw_error_at(at, "process_drain on null process"); return false; } + if ( p->stdoutOpen ) { + char tmp[4096]; +#ifdef _WIN32 + for ( ;; ) { + DWORD avail = 0; + if ( !PeekNamedPipe(p->hRead, NULL, 0, NULL, &avail, NULL) ) { p->stdoutOpen = false; break; } + if ( avail == 0 ) break; + DWORD toRead = avail > sizeof(tmp) ? (DWORD)sizeof(tmp) : avail; + DWORD got = 0; + if ( !ReadFile(p->hRead, tmp, toRead, &got, NULL) || got == 0 ) { p->stdoutOpen = false; break; } + p->buf.append(tmp, got); + } +#else + for ( ;; ) { + ssize_t n = read(p->fd, tmp, sizeof(tmp)); + if ( n > 0 ) { p->buf.append(tmp, (size_t)n); continue; } + if ( n == 0 ) { p->stdoutOpen = false; break; } // EOF: child closed stdout + if ( errno == EAGAIN || errno == EWOULDBLOCK ) break; // nothing ready this tick + p->stdoutOpen = false; break; // real read error + } +#endif + } + size_t start = 0, nl; + while ( (nl = p->buf.find('\n', start)) != string::npos ) { + string line = p->buf.substr(start, nl - start); + if ( !line.empty() && line.back() == '\r' ) line.pop_back(); + char * s = context->allocateString(line.data(), (uint32_t)line.size(), at); + vec4f cargs[1]; cargs[0] = cast::from(s); + context->invoke(blk, cargs, nullptr, at); + start = nl + 1; + } + p->buf.erase(0, start); + if ( !p->stdoutOpen && !p->buf.empty() ) { // a last line with no newline + string line = p->buf; + if ( !line.empty() && line.back() == '\r' ) line.pop_back(); + char * s = context->allocateString(line.data(), (uint32_t)line.size(), at); + vec4f cargs[1]; cargs[0] = cast::from(s); + context->invoke(blk, cargs, nullptr, at); + p->buf.clear(); + } + return p->stdoutOpen; + } + + int builtin_process_poll ( DasSubProcess * p, Context * context, LineInfoArg * at ) { + if ( !p ) { context->throw_error_at(at, "process_poll on null process"); return DAS_PROCESS_RUNNING; } + if ( p->reaped ) return p->exitCode; +#ifdef _WIN32 + if ( WaitForSingleObject(p->hProcess, 0) == WAIT_TIMEOUT ) return DAS_PROCESS_RUNNING; + DWORD code = 0; GetExitCodeProcess(p->hProcess, &code); + p->reaped = true; p->exitCode = (int)code; return p->exitCode; +#else + int status = 0; + pid_t r = waitpid(p->pid, &status, WNOHANG); + if ( r == 0 ) return DAS_PROCESS_RUNNING; + p->reaped = true; + p->exitCode = r < 0 ? -1 + : WIFEXITED(status) ? WEXITSTATUS(status) : WIFSIGNALED(status) ? WTERMSIG(status) : status; + return p->exitCode; +#endif + } + + int builtin_process_wait ( DasSubProcess * p, float timeout_sec, Context * context, LineInfoArg * at ) { + if ( !p ) { context->throw_error_at(at, "process_wait on null process"); return DAS_PROCESS_RUNNING; } + if ( p->reaped ) return p->exitCode; +#ifdef _WIN32 + DWORD ms = timeout_sec <= 0.0f ? INFINITE : (DWORD)(timeout_sec * 1000.0f); + if ( WaitForSingleObject(p->hProcess, ms) == WAIT_TIMEOUT ) return DAS_PROCESS_RUNNING; + DWORD code = 0; GetExitCodeProcess(p->hProcess, &code); + p->reaped = true; p->exitCode = (int)code; return p->exitCode; +#else + auto deadline = std::chrono::steady_clock::now() + + std::chrono::milliseconds((int)(timeout_sec * 1000.0f)); + for ( ;; ) { + int status = 0; + pid_t r = waitpid(p->pid, &status, WNOHANG); + if ( r > 0 ) { + p->reaped = true; + p->exitCode = WIFEXITED(status) ? WEXITSTATUS(status) + : WIFSIGNALED(status) ? WTERMSIG(status) : status; + return p->exitCode; + } + if ( r < 0 ) { p->reaped = true; p->exitCode = -1; return -1; } + if ( timeout_sec > 0.0f && std::chrono::steady_clock::now() >= deadline ) + return DAS_PROCESS_RUNNING; + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } +#endif + } + + void builtin_process_terminate ( DasSubProcess * p, Context * context, LineInfoArg * at ) { + if ( !p ) { context->throw_error_at(at, "process_terminate on null process"); return; } +#ifdef _WIN32 + if ( p->hJob ) TerminateJobObject(p->hJob, 15); + else if ( p->hProcess ) TerminateProcess(p->hProcess, 15); +#else + killpg(p->pid, SIGTERM); +#endif + } + + void builtin_process_kill ( DasSubProcess * p, Context * context, LineInfoArg * at ) { + if ( !p ) { context->throw_error_at(at, "process_kill on null process"); return; } +#ifdef _WIN32 + if ( p->hJob ) TerminateJobObject(p->hJob, 9); + else if ( p->hProcess ) TerminateProcess(p->hProcess, 9); +#else + killpg(p->pid, SIGKILL); +#endif + } + + int builtin_process_pid ( DasSubProcess * p, Context * context, LineInfoArg * at ) { + if ( !p ) { context->throw_error_at(at, "process_pid on null process"); return 0; } + return (int)p->pid; + } + + bool builtin_process_alive ( int32_t pid, Context *, LineInfoArg * ) { + if ( pid <= 0 ) return false; +#ifdef _WIN32 + HANDLE h = OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, FALSE, (DWORD)pid); + if ( !h ) return false; + DWORD code = 0; BOOL ok = GetExitCodeProcess(h, &code); + CloseHandle(h); + return ok && code == STILL_ACTIVE; +#else + // signal 0 probes without delivering; EPERM means it exists but is not ours to signal. + if ( ::kill((pid_t)pid, 0) == 0 ) return true; + return errno == EPERM; +#endif + } + + void builtin_close_process ( DasSubProcess * p, Context *, LineInfoArg * ) { + if ( !p ) return; +#ifdef _WIN32 + if ( p->hRead && p->hRead != INVALID_HANDLE_VALUE ) CloseHandle(p->hRead); + if ( p->hProcess ) CloseHandle(p->hProcess); + if ( p->hJob ) CloseHandle(p->hJob); // kill-on-close reaps a still-running tree +#else + if ( p->fd >= 0 ) close(p->fd); + if ( !p->reaped ) { // never leave a zombie + int status = 0; + if ( waitpid(p->pid, &status, WNOHANG) == 0 ) { killpg(p->pid, SIGKILL); waitpid(p->pid, &status, 0); } + } +#endif + delete p; + } + int builtin_system ( const char * cmd, Context * context, LineInfoArg * at ) { if ( !cmd ) { context->throw_error_at(at, "system of null"); @@ -2456,6 +2783,7 @@ namespace das { MAKE_TYPE_FACTORY(FStat, das::FStat) MAKE_TYPE_FACTORY(FILE,FILE) +MAKE_TYPE_FACTORY(SubProcess, das::DasSubProcess) MAKE_TYPE_FACTORY(DiskSpaceInfo, das::DiskSpaceInfo) namespace das { @@ -2497,6 +2825,7 @@ namespace das { addBuiltinDependency(lib, Module::require("strings")); // type addAnnotation(new DummyTypeAnnotation("FILE", "FILE", 16, 16)); + addAnnotation(new DummyTypeAnnotation("SubProcess", "das::DasSubProcess", sizeof(void *), alignof(void *))); addAnnotation(new FStatAnnotation(lib)); // seek constants addConstant(*this, "seek_set", SEEK_SET); @@ -2674,6 +3003,35 @@ namespace das { SideEffects::modifyExternal, "builtin_popen_argv_pipe") ->args({"args","scope","context","at"})->unsafeOperation = true; addConstant(*this, "popen_timed_out", DAS_POPEN_TIMEOUT); + // long-lived child process (spawn once, poll/drain/signal across ticks) + addExtern(*this, lib, "spawn_process", + SideEffects::modifyExternal, "builtin_spawn_process") + ->args({"argv","cwd","env","context","at"})->unsafeOperation = true; + addExtern(*this, lib, "process_drain", + SideEffects::modifyExternal, "builtin_process_drain") + ->args({"process","block","context","at"})->unsafeOperation = true; + addExtern(*this, lib, "process_poll", + SideEffects::modifyExternal, "builtin_process_poll") + ->args({"process","context","at"})->unsafeOperation = true; + addExtern(*this, lib, "process_wait", + SideEffects::modifyExternal, "builtin_process_wait") + ->args({"process","timeout","context","at"})->unsafeOperation = true; + addExtern(*this, lib, "process_terminate", + SideEffects::modifyExternal, "builtin_process_terminate") + ->args({"process","context","at"})->unsafeOperation = true; + addExtern(*this, lib, "process_kill", + SideEffects::modifyExternal, "builtin_process_kill") + ->args({"process","context","at"})->unsafeOperation = true; + addExtern(*this, lib, "process_pid", + SideEffects::accessExternal, "builtin_process_pid") + ->args({"process","context","at"})->unsafeOperation = true; + addExtern(*this, lib, "process_alive", + SideEffects::accessExternal, "builtin_process_alive") + ->args({"pid","context","at"})->unsafeOperation = true; + addExtern(*this, lib, "close_process", + SideEffects::modifyExternal, "builtin_close_process") + ->args({"process","context","at"})->unsafeOperation = true; + addConstant(*this, "process_running", DAS_PROCESS_RUNNING); addExtern(*this, lib, "system", SideEffects::modifyExternal, "builtin_system") ->args({"command","context","at"})->unsafeOperation = true; diff --git a/tests/fio/_fixture_process_child.das b/tests/fio/_fixture_process_child.das new file mode 100644 index 0000000000..77ffdde1c9 --- /dev/null +++ b/tests/fio/_fixture_process_child.das @@ -0,0 +1,32 @@ +options gen2 +options no_aot + +require daslib/clargs +require daslib/fio + +// Child for test_process.das: prints a few known lines (env override, working directory, two +// plain lines), signals ready, waits for the parent's release event, then exits with code 7. +// Codes come back from main - a das `exit()` unwinds as an abnormal termination and the CLI +// reports 1, so it cannot carry a code. The elapsed-time check is only a deadlock guard. +[export] +def main() : int { + let args <- get_cli_arguments() + if (length(args) != 2) return 2 + let ready = args[0] + let release = args[1] + var env_val = "" + if (has_env_variable("DAS_PROC_TEST")) { + env_val = get_env_variable("DAS_PROC_TEST") + } + print("env={env_val}\n") + print("cwd={base_name(getcwd())}\n") + print("hello one\n") + print("hello two\n") + if (!fwrite(ready, "ready")) return 3 + let started = ref_time_ticks() + while (!fexist(release)) { + if (get_time_usec(started) > 30000000) return 4 + sleep(5u) + } + return 7 +} diff --git a/tests/fio/test_process.das b/tests/fio/test_process.das new file mode 100644 index 0000000000..ec6354552b --- /dev/null +++ b/tests/fio/test_process.das @@ -0,0 +1,106 @@ +options gen2 +options no_aot + +require dastest/testing_boost public +require daslib/fio + +// argv[0] is the running interpreter (dastest is launched as `daslang dastest/dastest.das ...`), +// so it spawns the daslang we want the child to run. Same trick as popen_argv.das. +def das_exe() : string { + let args <- get_command_line_arguments() + return empty(args) ? "" : args[0] +} + +def has_line(lines : array; s : string) : bool { + for (l in lines) { + if (l == s) return true + } + return false +} + +def child_argv(fixture, ready, release : string) : array { + return [das_exe(), "-dasroot", get_das_root(), fixture, "--", ready, release] +} + +[test] +def test_process_lifecycle(t : T?) { + t |> run("spawn, drain lines, poll running, wait exit code, apply env + cwd") @(t : T?) { + let tmp = create_temp_directory_result("das process test ") + if (!(tmp is value)) { + t |> failure("could not create temp directory: {tmp as error}") + return + } + let root_dir = tmp as value + let ready = path_join(root_dir, "ready") + let release = path_join(root_dir, "release") + let fixture = path_join(get_das_root(), "tests/fio/_fixture_process_child.das") + var p = unsafe(spawn_process(child_argv(fixture, ready, release), root_dir, ["DAS_PROC_TEST=marker42"])) + t |> success(p != null, "spawn_process returned a handle") + // Drain the child's stdout until it signals ready; the file event drives completion, + // the elapsed check is only a guard. + var lines : array + let started = ref_time_ticks() + while (!fexist(ready)) { + unsafe(process_drain(p) $(line) { + lines |> push(clone_string(line)) + }) + if (get_time_usec(started) > 15000000) { + t |> failure("child never signaled ready") + break + } + sleep(5u) + } + unsafe(process_drain(p) $(line) { + lines |> push(clone_string(line)) + }) + t |> success(has_line(lines, "env=marker42"), "env override reached the child: {lines}") + t |> success(has_line(lines, "cwd={base_name(root_dir)}"), "cwd applied to the child: {lines}") + t |> success(has_line(lines, "hello one") && has_line(lines, "hello two"), + "both stdout lines drained one per line: {lines}") + t |> equal(unsafe(process_poll(p)), process_running, "child is still running while parked on release") + let pid = unsafe(process_pid(p)) + t |> success(pid > 0, "pid is positive: {pid}") + t |> success(unsafe(process_alive(pid)), "process_alive is true while running") + t |> success(fwrite(release, "go"), "wrote the release event") + t |> equal(unsafe(process_wait(p, 5.0)), 7, "child exited with its own code") + t |> success(!unsafe(process_alive(pid)), "process_alive is false after exit") + unsafe(close_process(p)) + let cleanup = rmdir_rec_result(root_dir) + t |> success(cleanup is value && cleanup as value, "temp dir removed") + } +} + +[test] +def test_process_terminate(t : T?) { + t |> run("terminate stops a child parked on a release that never comes") @(t : T?) { + let tmp = create_temp_directory_result("das process kill ") + if (!(tmp is value)) { + t |> failure("could not create temp directory: {tmp as error}") + return + } + let root_dir = tmp as value + let ready = path_join(root_dir, "ready") + let release = path_join(root_dir, "release") // deliberately never written + let fixture = path_join(get_das_root(), "tests/fio/_fixture_process_child.das") + let noenv : array + var p = unsafe(spawn_process(child_argv(fixture, ready, release), "", noenv)) + t |> success(p != null, "spawn_process returned a handle") + let started = ref_time_ticks() + while (!fexist(ready)) { + unsafe(process_drain(p) $(line) {}) + if (get_time_usec(started) > 15000000) { + t |> failure("child never signaled ready") + break + } + sleep(5u) + } + let pid = unsafe(process_pid(p)) + unsafe(process_terminate(p)) + let rc = unsafe(process_wait(p, 5.0)) + t |> success(rc != process_running, "child stopped after terminate (rc={rc})") + t |> success(!unsafe(process_alive(pid)), "process_alive is false after terminate") + unsafe(close_process(p)) + let cleanup = rmdir_rec_result(root_dir) + t |> success(cleanup is value && cleanup as value, "temp dir removed") + } +} From f4bc9e8f8bb4b5d3e9a27655127ac2224f4f4464 Mon Sep 17 00:00:00 2001 From: Boris Batkin Date: Sat, 5 Sep 2026 20:13:58 -0700 Subject: [PATCH 02/14] watchdog: the supervisor in daslang - utils/watchdog/watchdog.das, a state machine the host ticks, with the Python's config, discovery, stages, crash bundles and exit-code contract The library half of the watchdog.py replacement. One module, no Python: - WatchdogConfig is a [CommandLineArgs] struct. watchdog.json beside the program sets each flag's default (a key becomes a `--flag value` the CLI did not supply), an unknown key is fatal by name, and everything after `--` goes to the child verbatim - clargs drops flag-shaped positionals, so the split is explicit. Discovery: main.das beside bin/Release/daslang is script mode, one *.exe is program mode, anything else is an error. - Emitter writes {"ts","event",...} as one compact JSON line per record to a file that rotates at 20 MB x 5, and to stdout. - StageDetector ranks the child's startup lines and reports only forward moves; the tune-restart marker rewinds the rank. @tune events log at the kernel boundaries. Health and shutdown go over dasHV's client. - Crash bundles: crash.json, the log, the dump, symbols, the tune sidecar, the JIT artifacts the child named. WER policy read/install via reg.exe. - Supervisor.tick() is one 250 ms step: drain, health, poll, classify an exit - 0 ends, 3 with the marker relaunches at once, 3 without backs off, 4 relaunches, anything else is a crash with bounded backoff. The host owns the loop and calls request_stop() on its signal; run() is the loop for a host that owns nothing else. watchdog_start() does everything before the loop: config, the dump policy, the single-instance pid file. Cut on purpose: the exchange UI (consent, sidecar fold, tray rails, the DAS_TUNE_CONTROL stop file) and the tray; notifications stay, spawned through the platform's own helper, no binding needed. tests/watchdog drives the library under dastest with a daslang child: config layering, discovery, stage ranks, the log envelope and rotation, crash-bundle-restart, config restart, the script-mode tune bootstrap, and the stop-file handshake from a host stop request. Co-Authored-By: Claude Fable 5.1 --- tests/.das_test | 5 + tests/aot/CMakeLists.txt | 2 +- tests/watchdog/_fixture_watchdog_child.das | 62 + tests/watchdog/test_watchdog.das | 349 ++++++ utils/watchdog/watchdog.das | 1210 ++++++++++++++++++++ 5 files changed, 1627 insertions(+), 1 deletion(-) create mode 100644 tests/watchdog/_fixture_watchdog_child.das create mode 100644 tests/watchdog/test_watchdog.das create mode 100644 utils/watchdog/watchdog.das diff --git a/tests/.das_test b/tests/.das_test index 2f5edab403..26b6869e09 100644 --- a/tests/.das_test +++ b/tests/.das_test @@ -13,6 +13,11 @@ def can_visit_folder(folder_name : string; var result : bool?) { *result = has_module("dashv") return } + // the supervisor polls health over dasHV's client + if (folder_name == "watchdog") { + *result = has_module("dashv") + return + } if (folder_name == "dasPUGIXML") { *result = has_module("pugixml") return diff --git a/tests/aot/CMakeLists.txt b/tests/aot/CMakeLists.txt index 3e1a08d586..2af8115100 100644 --- a/tests/aot/CMakeLists.txt +++ b/tests/aot/CMakeLists.txt @@ -391,7 +391,7 @@ set(DAS_AOT_SUITES jobque json jsonrpc language linq lint long_array_table loops lpipe lsp macro_boost macro_call match math mcp md_boost module_cache module_tests option promote quote reader_macro regex rtti safe_addr soa spoof strings stbimage table_packed template - tests type_lattice type_traits typemacro uri with_boost delegate) + tests type_lattice type_traits typemacro uri watchdog with_boost delegate) foreach(_s IN LISTS DAS_AOT_SUITES) string(TOUPPER ${_s} _u) # suites with an irregular dir / a filter / a curated list define AOT__FILES above; diff --git a/tests/watchdog/_fixture_watchdog_child.das b/tests/watchdog/_fixture_watchdog_child.das new file mode 100644 index 0000000000..05e982307f --- /dev/null +++ b/tests/watchdog/_fixture_watchdog_child.das @@ -0,0 +1,62 @@ +options gen2 +options no_aot + +require daslib/clargs +require daslib/fio + +// The supervised child for test_watchdog.das. `state_dir mode`: a run counter in state_dir +// tells the child which launch it is, and `mode` picks the story it acts out: +// crash-then-ok run 1 prints the ready line and exits 9; run 2 prints it and exits 0 +// tune-then-ok run 1 prints the tune marker and exits 3; run 2 exits 0 +// config-restart run 1 exits 4; run 2 exits 0 +// park prints a line, waits for the stop file named by CADMUS_STOP_FILE, exits 0 +// Codes return from main: a das exit() is an abnormal termination and reports 1. +def run_number(state_dir : string) : int { + let path = path_join(state_dir, "runs") + var runs = 0 + if (fexist(path)) { + fopen(path, "rb") $(f) { + if (f != null) { + runs = to_int(fread(f)) + } + } + } + runs++ + fwrite(path, "{runs}") + return runs +} + +[export] +def main() : int { + let args <- get_cli_arguments() + if (length(args) != 2) return 2 + let run = run_number(args[0]) + let mode = args[1] + if (mode == "crash-then-ok") { + print("listening on http://127.0.0.1:1/\n") + return run == 1 ? 9 : 0 + } + if (mode == "tune-then-ok") { + if (run == 1) { + print("llvm_tune: tuning scope 'fixture'\n") + print("llvm_tune: restart to apply the winners\n") + return 3 + } + return 0 + } + if (mode == "config-restart") { + return run == 1 ? 4 : 0 + } + if (mode == "park") { + print("parked\n") + let stop_file = has_env_variable("CADMUS_STOP_FILE") ? get_env_variable("CADMUS_STOP_FILE") : "" + if (empty(stop_file)) return 5 + let started = ref_time_ticks() + while (!fexist(stop_file)) { + if (get_time_usec(started) > 30000000) return 6 + sleep(5u) + } + return 0 + } + return 7 +} diff --git a/tests/watchdog/test_watchdog.das b/tests/watchdog/test_watchdog.das new file mode 100644 index 0000000000..4d5c1e2133 --- /dev/null +++ b/tests/watchdog/test_watchdog.das @@ -0,0 +1,349 @@ +options gen2 +options no_aot + +require dastest/testing_boost public +require daslib/fio +require daslib/json_boost +require daslib/strings_boost +require ../../utils/watchdog/watchdog.das + +// argv[0] is the daslang running dastest; the supervised "program" is that binary running the +// fixture, so program mode and script mode both spawn a real daslang child. Absolute, because a +// relative --program resolves against --cwd, which is a temp dir here. +def das_exe() : string { + let args <- get_command_line_arguments() + return empty(args) ? "" : get_full_file_name(args[0]) +} + +def fixture_path() : string { + return path_join(get_das_root(), "tests/watchdog/_fixture_watchdog_child.das") +} + +def make_temp(t : T?; prefix : string) : string { + let tmp = create_temp_directory_result(prefix) + if (!(tmp is value)) { + t |> failure("could not create temp directory: {tmp as error}") + return "" + } + return tmp as value +} + +def read_events(log_path : string) : array { + var events : array + if (!fexist(log_path)) return <- events + var text : string + fopen(log_path, "rb") $(f) { + if (f != null) { + text = fread(f) + } + } + for (line in split(text, "\n")) { + if (empty(strip(line))) continue + var error : string + var js = read_json(line, error) + if (js != null) { + events |> push(js) + } + } + return <- events +} + +def event_names(events : array) : array { + return <- [for (ev in events); ev?["event"] ?? ""] +} + +def count_event(events : array; name : string) : int { + var n = 0 + for (ev in events) { + if ((ev?["event"] ?? "") == name) { + n++ + } + } + return n +} + +def event_int(events : array; name, key : string; nth : int = 0) : int { + var seen = 0 + for (ev in events) { + if ((ev?["event"] ?? "") != name) continue + if (seen == nth) return int(ev?[key] ?? -999999l) + seen++ + } + return -999999 +} + +def child_args(state_dir, mode : string) : array { + return ["-dasroot", get_das_root(), fixture_path(), "--", state_dir, mode] +} + +def program_mode_args(root, state_dir, mode : string) : array { + var args <- ["--program", das_exe(), "--name", "wdtest", "--cwd", root, "--no-health", + "--stable-seconds", "0.1", "--max-restart-delay", "0.5", "--"] + args |> push_from(child_args(state_dir, mode)) + return <- args +} + +[test] +def test_config_layering(t : T?) { + t |> run("watchdog.json sets flag defaults, a flag on the command line wins, -- goes to the child") @(t : T?) { + let root = make_temp(t, "wd config ") + if (empty(root)) return + fwrite(path_join(root, "watchdog.json"), + "\{\"name\": \"fromjson\", \"health_interval\": 7, \"no_health\": true, \"server_args\": [\"a\", \"b\"]\}") + let resolved <- resolve_config(["--name", "fromcli", "--program", das_exe()], root) + t |> equal(resolved.error, "") + t |> equal(resolved.cfg.name, "fromcli", "the flag beats the config") + t |> equal(resolved.cfg.health_interval, 7.0, "the config sets a default") + t |> success(resolved.cfg.no_health, "a true bool in the config sets the flag") + t |> equal(length(resolved.cfg.server_args), 2, "server_args default from the config: {resolved.cfg.server_args}") + let with_tail <- resolve_config(["--program", das_exe(), "--", "-x", "child arg"], root) + t |> equal(with_tail.error, "") + t |> equal(length(with_tail.cfg.server_args), 2) + if (length(with_tail.cfg.server_args) == 2) { + t |> equal(with_tail.cfg.server_args[0], "-x", "flag-shaped child args survive") + t |> equal(with_tail.cfg.server_args[1], "child arg") + } + t |> equal(with_tail.cfg.name, "fromjson", "the config's name applies when no flag names one") + t |> success(ends_with(with_tail.cfg.log, "fromjson-watchdog.log"), "log path keyed on the name: {with_tail.cfg.log}") + fwrite(path_join(root, "watchdog.json"), "\{\"helth_url\": \"x\"\}") + let typo <- resolve_config(["--program", das_exe()], root) + t |> success(find(typo.error, "helth_url") >= 0, "an unknown key is fatal and named: {typo.error}") + let cleanup = rmdir_rec_result(root) + t |> success(cleanup is value && cleanup as value, "temp dir removed") + } +} + +[test] +def test_discovery(t : T?) { + t |> run("main.das beside bin/Release/daslang means script mode; two exes are ambiguous; nothing is an error") @(t : T?) { + let root = make_temp(t, "wd discover ") + if (empty(root)) return + mkdir(path_join(root, "bin")) + mkdir(path_join(root, "bin/Release")) + fwrite(path_join(root, "main.das"), "") + fwrite(path_join(root, "bin/Release/daslang"), "") + let script_mode <- resolve_config(["--no-health"], root) + t |> equal(script_mode.error, "") + t |> success(ends_with(script_mode.cfg.script, "main.das"), "script discovered: {script_mode.cfg.script}") + t |> success(ends_with(script_mode.cfg.daslang, "daslang"), "daslang discovered beside it: {script_mode.cfg.daslang}") + t |> equal(script_mode.cfg.name, base_name(root), "a script names itself after its directory") + let command <- child_command(script_mode.cfg) + t |> success(length(command) >= 4 && command[1] == "-jit" && command[3] == "--", "script command shape: {command}") + remove(path_join(root, "main.das")) + fwrite(path_join(root, "one.exe"), "") + fwrite(path_join(root, "two.exe"), "") + let ambiguous <- resolve_config([], root) + t |> success(find(ambiguous.error, "2 executables") >= 0, "two exes are refused: {ambiguous.error}") + remove(path_join(root, "two.exe")) + let program_mode <- resolve_config([], root) + t |> equal(program_mode.error, "") + t |> equal(program_mode.cfg.name, "one", "an executable names itself") + remove(path_join(root, "one.exe")) + let nothing <- resolve_config([], root) + t |> success(find(nothing.error, "nothing to supervise") >= 0, "an empty directory is an error: {nothing.error}") + let cleanup = rmdir_rec_result(root) + t |> success(cleanup is value && cleanup as value, "temp dir removed") + } +} + +[test] +def test_stages_are_ranked_and_monotonic(t : T?) { + t |> run("only a forward move reports; the tune restart rewinds the rank") @(t : T?) { + var detector = new StageDetector() + var stage = Stage(name = "", rank = 0, since = 0.0) + var moved = detector->advance(stage, "LLVM JIT: DLL cache miss, codegen for foo.dll", 1.0) + t |> success(moved != null && (moved?["stage"] ?? "") == "jit_codegen", "first move: jit_codegen") + moved = detector->advance(stage, "Library foo linked - ok", 2.5) + t |> success(moved != null && (moved?["stage"] ?? "") == "jit_linked", "forward move: jit_linked") + t |> equal(moved?["after"] ?? "", "jit_codegen", "the previous stage is named") + t |> equal(float(moved?["elapsed_s"] ?? 0.0lf), 1.5, "and how long it took") + moved = detector->advance(stage, "LLVM JIT: DLL cache miss, codegen for bar.dll", 3.0) + t |> success(moved == null, "a lower-ranked line mid-tune does not flap the stage back") + moved = detector->advance(stage, "llvm_tune: restart to apply the winners", 4.0) + t |> success(moved != null && (moved?["stage"] ?? "") == "tune_restart", "the restart marker is a move") + t |> equal(stage.rank, 0, "and it resets the rank so the sequence can replay") + moved = detector->advance(stage, "LLVM JIT: DLL cache hit foo.dll", 5.0) + t |> success(moved != null && (moved?["stage"] ?? "") == "jit_cached", "the replay advances again") + t |> success(detector->advance(stage, "nothing to see here", 6.0) == null, "an unmatched line is not a stage") + unsafe { + delete detector + } + } +} + +[test] +def test_parse_child_event(t : T?) { + t |> run("@tune k=v ... splits into a kind and fields; other lines do not") @(t : T?) { + let ev <- parse_child_event("@tune plan scope=metal total=12", "@tune ") + t |> equal(ev.kind, "plan") + t |> equal(ev.fields?["scope"] ?? "", "metal") + t |> equal(ev.fields?["total"] ?? "", "12") + let none <- parse_child_event("plain child output", "@tune ") + t |> equal(none.kind, "") + } +} + +[test] +def test_emitter_envelope_and_rotation(t : T?) { + t |> run("one JSON object per line with ts and event, fields beside them; the file rotates at the size cap") @(t : T?) { + let root = make_temp(t, "wd log ") + if (empty(root)) return + let log_path = path_join(root, "logs/x-watchdog.log") + var log = new Emitter(log_path) + log.tee = false + log->emit("hello", JV((pid = 42, note = "n"))) + let events <- read_events(log_path) + t |> equal(length(events), 1) + if (length(events) == 1) { + t |> equal(events[0]?["event"] ?? "", "hello") + t |> equal(int(events[0]?["pid"] ?? 0l), 42) + t |> equal(events[0]?["note"] ?? "", "n") + t |> success(length(events[0]?["ts"] ?? "") >= 19, "an ISO timestamp: {events[0]?["ts"] ?? ""}") + } + log.max_bytes = 1l + log->emit("second") + t |> success(fexist("{log_path}.1"), "the full file rotated to .1") + let after <- read_events(log_path) + t |> equal(length(after), 1, "the live file holds only the record written after rotation") + unsafe { + delete log + } + let cleanup = rmdir_rec_result(root) + t |> success(cleanup is value && cleanup as value, "temp dir removed") + } +} + +[test] +def test_supervise_crash_then_clean_exit(t : T?) { + t |> run("a crash is reported, bundled and restarted with backoff; exit 0 ends supervision") @(t : T?) { + notifications_enabled = false + let root = make_temp(t, "wd crash ") + if (empty(root)) return + let state_dir = path_join(root, "state") + mkdir(state_dir) + var code = 0 + var sup = watchdog_start(program_mode_args(root, state_dir, "crash-then-ok"), root, 0, code) + t |> success(sup != null, "supervision started (code {code})") + if (sup == null) return + t |> equal(sup->run(), 0, "supervision ends with 0 after the child's clean exit") + let events <- read_events(sup.cfg.log) + t |> equal(count_event(events, "child_started"), 2, "the child ran twice: {event_names(events)}") + t |> equal(event_int(events, "crash", "code"), 9, "the first exit was reported as a crash") + t |> equal(count_event(events, "crash_bundle"), 1, "one bundle collected") + t |> equal(event_int(events, "child_exited", "code", 1), 0, "the second exit was clean") + t |> equal(count_event(events, "intentional_shutdown"), 1) + t |> equal(count_event(events, "watchdog_stopped"), 1) + t |> equal(count_event(events, "stage"), 2, "the ready line moved the stage on both runs") + var bundles = 0 + dir(sup.cfg.crash_dir) $(fname) { + if (starts_with(fname, "wdtest-")) { + bundles++ + t |> success(fexist(path_join(path_join(sup.cfg.crash_dir, fname), "crash.json")), "crash.json in the bundle") + } + } + t |> equal(bundles, 1) + unsafe { + delete sup + } + let cleanup = rmdir_rec_result(root) + t |> success(cleanup is value && cleanup as value, "temp dir removed") + } +} + +[test] +def test_supervise_config_restart(t : T?) { + t |> run("exit 4 relaunches at once with no crash report") @(t : T?) { + notifications_enabled = false + let root = make_temp(t, "wd config restart ") + if (empty(root)) return + let state_dir = path_join(root, "state") + mkdir(state_dir) + var code = 0 + var sup = watchdog_start(program_mode_args(root, state_dir, "config-restart"), root, 0, code) + t |> success(sup != null, "supervision started (code {code})") + if (sup == null) return + t |> equal(sup->run(), 0) + let events <- read_events(sup.cfg.log) + t |> equal(count_event(events, "config_restart_relaunch"), 1, "{event_names(events)}") + t |> equal(count_event(events, "crash"), 0, "no crash was reported") + t |> equal(count_event(events, "child_started"), 2) + unsafe { + delete sup + } + let cleanup = rmdir_rec_result(root) + t |> success(cleanup is value && cleanup as value, "temp dir removed") + } +} + +[test] +def test_supervise_tune_bootstrap(t : T?) { + t |> run("script mode: exit 3 after the restart marker relaunches at once") @(t : T?) { + notifications_enabled = false + let root = make_temp(t, "wd tune ") + if (empty(root)) return + let state_dir = path_join(root, "state") + mkdir(state_dir) + let args <- ["--daslang", das_exe(), "--script", fixture_path(), "--name", "wdtune", "--cwd", root, + "--no-health", "--stable-seconds", "0.1", "--max-restart-delay", "0.5", "--", state_dir, "tune-then-ok"] + var code = 0 + var sup = watchdog_start(args, root, 0, code) + t |> success(sup != null, "supervision started (code {code})") + if (sup == null) return + t |> equal(sup->run(), 0) + let events <- read_events(sup.cfg.log) + t |> equal(count_event(events, "tune_bootstrap_complete"), 1, "{event_names(events)}") + t |> equal(count_event(events, "crash"), 0) + t |> equal(count_event(events, "child_started"), 2) + unsafe { + delete sup + } + let cleanup = rmdir_rec_result(root) + t |> success(cleanup is value && cleanup as value, "temp dir removed") + } +} + +[test] +def test_request_stop_uses_the_stop_file(t : T?) { + t |> run("the host's stop request writes the stop file; the parked child exits 0 and supervision ends") @(t : T?) { + notifications_enabled = false + let root = make_temp(t, "wd stop ") + if (empty(root)) return + let state_dir = path_join(root, "state") + mkdir(state_dir) + var args <- ["--program", das_exe(), "--name", "wdpark", "--cwd", root, "--no-health", + "--stop-file", path_join(root, "stop"), "--"] + args |> push_from(child_args(state_dir, "park")) + var code = 0 + var sup = watchdog_start(args, root, 0, code) + t |> success(sup != null, "supervision started (code {code})") + if (sup == null) return + var parked = false + let started = ref_time_ticks() + while (!parked && get_time_usec(started) < 15000000) { + sup->tick() + for (ev in read_events(sup.cfg.log)) { + if ((ev?["event"] ?? "") == "child" && (ev?["message"] ?? "") == "parked") { + parked = true + } + } + } + t |> success(parked, "the child reported it was parked") + sup->request_stop() + var done = false + let stop_started = ref_time_ticks() + while (!done && get_time_usec(stop_started) < 15000000) { + done = sup->tick() + } + t |> success(done, "supervision ended after the stop request") + t |> equal(sup.result, 0) + let events <- read_events(sup.cfg.log) + t |> equal(count_event(events, "stop_file_requested"), 1, "{event_names(events)}") + t |> equal(event_int(events, "child_exited", "code"), 0, "the child left through its stop-file handshake") + t |> success(!fexist(path_join(root, "stop")), "the stop file is cleared at the end") + unsafe { + delete sup + } + let cleanup = rmdir_rec_result(root) + t |> success(cleanup is value && cleanup as value, "temp dir removed") + } +} diff --git a/utils/watchdog/watchdog.das b/utils/watchdog/watchdog.das new file mode 100644 index 0000000000..1d64302280 --- /dev/null +++ b/utils/watchdog/watchdog.das @@ -0,0 +1,1210 @@ +options gen2 +options indenting = 4 + +module watchdog shared + +require strings +require math +require daslib/strings_boost +require daslib/fio +require daslib/clargs +require daslib/result +require daslib/json_boost +require daslib/regex +require daslib/regex_boost +require dashv/dashv_boost + +//! Supervise a daslang program (`daslang -jit main.das`) or a standalone executable: restart it with +//! bounded backoff, capture crashes into bundles, report startup stages, poll health. Exit 0 is an +//! intentional shutdown, 3 a tune bootstrap (relaunched at once when the child printed the restart +//! marker), 4 a config restart, anything else a crash. The host drives `Supervisor.tick()` on its +//! own clock and calls `request_stop()` on a signal; `watchdog_main` runs that loop for a host that +//! owns nothing else. + +let public TUNE_RESTART_EXIT = 3 +let public CONFIG_RESTART_EXIT = 4 +let public CONFIG_NAME = "watchdog.json" +let private HEALTH_HEARTBEAT_SECONDS = 300.0 +let private TICK_MS = 250u +let private LOG_MAX_BYTES = 20l * 1024l * 1024l +let private LOG_BACKUPS = 5 +let private TUNE_EVENT_PREFIX = "@tune " + +// --------------------------------------------------------------------------------------------- +// configuration: flags, watchdog.json beside the program, layout discovery, identity + +[CommandLineArgs] +struct public WatchdogConfig { + @clarg_doc = "Identity the log, pid file and notifications key on; derived from the program" + name : string + @clarg_doc = "Run this standalone program instead of daslang -jit