From 58dd2aa31eaa30996f51bad293ed0d8ef6334353 Mon Sep 17 00:00:00 2001 From: Trevor McKay Date: Wed, 22 Jul 2026 18:54:24 -0400 Subject: [PATCH 1/7] fix(grpc): terminate workers on server shutdown Signed-off-by: Trevor McKay --- cpp/docs/grpc-server-architecture.md | 11 +++-- cpp/src/grpc/server/grpc_server_main.cpp | 13 ++++- cpp/src/grpc/server/grpc_server_types.hpp | 2 + cpp/src/grpc/server/grpc_worker.cpp | 6 +++ cpp/src/grpc/server/grpc_worker_infra.cpp | 49 +++++++++++++++++++ .../grpc/grpc_integration_test.cpp | 37 ++++++++++++++ .../cuopt-grpc/grpc-server-architecture.md | 1 + 7 files changed, 114 insertions(+), 5 deletions(-) diff --git a/cpp/docs/grpc-server-architecture.md b/cpp/docs/grpc-server-architecture.md index 425047cc23..06bad9b4f2 100644 --- a/cpp/docs/grpc-server-architecture.md +++ b/cpp/docs/grpc-server-architecture.md @@ -292,10 +292,13 @@ If a worker process crashes: ### Graceful Shutdown On SIGINT/SIGTERM: -1. Set `shm_ctrl->shutdown_requested = true` -2. Workers finish current job and exit -3. Main process waits for workers -4. Cleanup shared memory segments +1. Set `keep_running = false` and `shm_ctrl->shutdown_requested = true` +2. Mark all queued/running jobs `CANCELLED` and wake any `WaitForCompletion` waiters +3. `SIGKILL` all worker processes (mid-solve workers do not poll the shutdown flag) +4. Shut down the gRPC server with a short deadline so lingering RPCs cannot block exit +5. Join background threads, `waitpid` workers, and clean up shared memory + +Workers ignore SIGINT/SIGTERM so only the parent process owns shutdown; the parent always force-kills workers rather than waiting for the current solve to finish. ### Job Cancellation diff --git a/cpp/src/grpc/server/grpc_server_main.cpp b/cpp/src/grpc/server/grpc_server_main.cpp index ae6cdb1360..bda7f0c893 100644 --- a/cpp/src/grpc/server/grpc_server_main.cpp +++ b/cpp/src/grpc/server/grpc_server_main.cpp @@ -288,6 +288,7 @@ int main(int argc, char** argv) auto shutdown_all = [&]() { keep_running = false; shm_ctrl->shutdown_requested = true; + cancel_all_active_jobs_for_shutdown(); result_cv.notify_all(); if (result_thread.joinable()) result_thread.join(); @@ -324,11 +325,21 @@ int main(int argc, char** argv) server_max_message_bytes() / kMiB); SERVER_LOG_INFO("[gRPC Server] Press Ctrl+C to shutdown"); + // On signal: cancel active jobs and SIGKILL workers before gRPC Shutdown so + // WaitForCompletion / StreamLogs RPCs unblock and we never hang on mid-solve + // workers. Use a deadline in case an RPC is stuck for another reason. std::thread shutdown_thread([&server]() { while (keep_running.load()) { std::this_thread::sleep_for(std::chrono::milliseconds(100)); } - if (server) { server->Shutdown(); } + SERVER_LOG_INFO("[Server] Shutdown signal received; cancelling jobs and killing workers"); + if (shm_ctrl) { shm_ctrl->shutdown_requested = true; } + cancel_all_active_jobs_for_shutdown(); + kill_all_workers(); + if (server) { + auto deadline = std::chrono::system_clock::now() + std::chrono::seconds(5); + server->Shutdown(deadline); + } }); server->Wait(); diff --git a/cpp/src/grpc/server/grpc_server_types.hpp b/cpp/src/grpc/server/grpc_server_types.hpp index 044936fc64..062073bae5 100644 --- a/cpp/src/grpc/server/grpc_server_types.hpp +++ b/cpp/src/grpc/server/grpc_server_types.hpp @@ -350,7 +350,9 @@ void cleanup_shared_memory(); void log_worker_gpu_layout(); bool init_worker_cuda_environment(int worker_id); void spawn_workers(); +void kill_all_workers(); void wait_for_workers(); +void cancel_all_active_jobs_for_shutdown(); void worker_monitor_thread(); void result_retrieval_thread(); void incumbent_retrieval_thread(); diff --git a/cpp/src/grpc/server/grpc_worker.cpp b/cpp/src/grpc/server/grpc_worker.cpp index 373652f669..3d0d00f839 100644 --- a/cpp/src/grpc/server/grpc_worker.cpp +++ b/cpp/src/grpc/server/grpc_worker.cpp @@ -602,6 +602,12 @@ void worker_process(int worker_id) { SERVER_LOG_INFO("[Worker %d] Started (PID: %d)", worker_id, getpid()); + // Parent owns SIGINT/SIGTERM shutdown. Ignoring here prevents the inherited + // soft handler from leaving mid-solve workers alive after Ctrl-C while the + // parent waits on them. + signal(SIGINT, SIG_IGN); + signal(SIGTERM, SIG_IGN); + if (!init_worker_cuda_environment(worker_id)) { SERVER_LOG_ERROR("[Worker %d] CUDA environment initialization failed; exiting", worker_id); _exit(1); diff --git a/cpp/src/grpc/server/grpc_worker_infra.cpp b/cpp/src/grpc/server/grpc_worker_infra.cpp index 7097badf14..cdf310c076 100644 --- a/cpp/src/grpc/server/grpc_worker_infra.cpp +++ b/cpp/src/grpc/server/grpc_worker_infra.cpp @@ -208,8 +208,57 @@ void spawn_workers() } } +void kill_all_workers() +{ + for (pid_t pid : worker_pids) { + if (pid > 0) { kill(pid, SIGKILL); } + } +} + +void cancel_all_active_jobs_for_shutdown() +{ + if (job_queue != nullptr) { + for (size_t i = 0; i < MAX_JOBS; ++i) { + if (job_queue[i].ready.load(std::memory_order_acquire)) { + job_queue[i].cancelled.store(true, std::memory_order_release); + } + } + } + + { + std::lock_guard lock(tracker_mutex); + for (auto& [job_id, info] : job_tracker) { + (void)job_id; + if (info.status == JobStatus::QUEUED || info.status == JobStatus::PROCESSING) { + info.status = JobStatus::CANCELLED; + info.error_message = "Server shutting down"; + } + } + } + + { + std::lock_guard wlock(waiters_mutex); + for (auto& [job_id, waiter] : waiting_threads) { + (void)job_id; + { + std::lock_guard waiter_lock(waiter->mutex); + waiter->error_message = "Server shutting down"; + waiter->success = false; + waiter->ready = true; + } + waiter->cv.notify_all(); + } + waiting_threads.clear(); + } + + result_cv.notify_all(); +} + void wait_for_workers() { + // Mid-solve workers only check shutdown_requested between jobs, so force-kill + // before waitpid or Ctrl-C / SIGTERM can hang until the current solve ends. + kill_all_workers(); for (pid_t pid : worker_pids) { if (pid <= 0) continue; int status; diff --git a/cpp/tests/linear_programming/grpc/grpc_integration_test.cpp b/cpp/tests/linear_programming/grpc/grpc_integration_test.cpp index a3aafc3936..87641fef09 100644 --- a/cpp/tests/linear_programming/grpc/grpc_integration_test.cpp +++ b/cpp/tests/linear_programming/grpc/grpc_integration_test.cpp @@ -190,6 +190,8 @@ class ServerProcess { int port() const { return port_; } + pid_t pid() const { return pid_; } + bool is_running() const { if (pid_ <= 0) return false; @@ -1695,6 +1697,41 @@ TEST_F(ErrorRecoveryTests, ClientHandlesServerCrashDuringSolve) EXPECT_FALSE(status_result.error_message.empty()); } +TEST_F(ErrorRecoveryTests, SigintDuringRunningJobShutsDownPromptly) +{ + ASSERT_TRUE(start_server()); + auto client = create_client(); + ASSERT_NE(client, nullptr); + + std::string mps_path = get_test_mip_path("neos5-free-bound.mps"); + auto problem = load_problem_from_file(mps_path); + + mip_solver_settings_t settings; + settings.time_limit = 120.0; + + auto submit_result = client->submit_mip(problem, settings); + ASSERT_TRUE(submit_result.success); + + std::this_thread::sleep_for(std::chrono::seconds(2)); + ASSERT_TRUE(server_.is_running()); + + auto start = std::chrono::steady_clock::now(); + ASSERT_EQ(kill(server_.pid(), SIGINT), 0); + + // Server must exit well before the solve time_limit; previously Ctrl-C could + // hang until the mid-solve worker finished. + while (server_.is_running()) { + auto elapsed = std::chrono::steady_clock::now() - start; + ASSERT_LT(elapsed, std::chrono::seconds(15)) + << "Server did not shut down promptly after SIGINT during a running job"; + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + } + + auto elapsed = + std::chrono::duration_cast(std::chrono::steady_clock::now() - start); + EXPECT_LT(elapsed.count(), 15); +} + TEST_F(ErrorRecoveryTests, ClientTimeoutConfiguration) { ASSERT_TRUE(start_server()); diff --git a/docs/cuopt/source/cuopt-grpc/grpc-server-architecture.md b/docs/cuopt/source/cuopt-grpc/grpc-server-architecture.md index 4054ee3abd..a5aa12296c 100644 --- a/docs/cuopt/source/cuopt-grpc/grpc-server-architecture.md +++ b/docs/cuopt/source/cuopt-grpc/grpc-server-architecture.md @@ -28,6 +28,7 @@ Implementation details (IPC layout, C++ source map, chunked transfer internals) - If a **worker process crashes**, jobs it was running are marked **FAILED**; the server can spawn replacement workers (see contributor doc for details). - **`CancelJob`** cancels **queued** jobs immediately (the worker skips them). If the solver has already started, the **worker process is killed** and the job is marked **CANCELLED**; a replacement worker is spawned automatically. +- **Ctrl-C / SIGTERM** cancels active jobs, kills worker processes, and shuts the server down without waiting for an in-flight solve to finish. ## Further reading From d2ccf8116bf0af42087f1d81484c9dcb2093fcfa Mon Sep 17 00:00:00 2001 From: Trevor McKay Date: Thu, 23 Jul 2026 11:56:51 -0400 Subject: [PATCH 2/7] strengthen grpc worker shutdown semantics in the server --- cpp/docs/grpc-server-architecture.md | 7 ++-- cpp/src/grpc/server/grpc_server_main.cpp | 9 +++-- cpp/src/grpc/server/grpc_server_types.hpp | 1 + cpp/src/grpc/server/grpc_worker_infra.cpp | 42 +++++++++++++++++++++-- 4 files changed, 51 insertions(+), 8 deletions(-) diff --git a/cpp/docs/grpc-server-architecture.md b/cpp/docs/grpc-server-architecture.md index 06bad9b4f2..f8bc372146 100644 --- a/cpp/docs/grpc-server-architecture.md +++ b/cpp/docs/grpc-server-architecture.md @@ -295,10 +295,11 @@ On SIGINT/SIGTERM: 1. Set `keep_running = false` and `shm_ctrl->shutdown_requested = true` 2. Mark all queued/running jobs `CANCELLED` and wake any `WaitForCompletion` waiters 3. `SIGKILL` all worker processes (mid-solve workers do not poll the shutdown flag) -4. Shut down the gRPC server with a short deadline so lingering RPCs cannot block exit -5. Join background threads, `waitpid` workers, and clean up shared memory +4. Close server-side worker pipes so background threads blocked on pipe I/O unblock +5. Shut down the gRPC server with a short deadline so lingering RPCs cannot block exit +6. Join background threads, wait briefly for workers (`waitpid` with a grace period), and clean up shared memory -Workers ignore SIGINT/SIGTERM so only the parent process owns shutdown; the parent always force-kills workers rather than waiting for the current solve to finish. +Workers ignore SIGINT/SIGTERM so only the parent process owns shutdown; the parent always force-kills workers rather than waiting for the current solve to finish. Mid-CUDA workers can sit in uninterruptible D-state after `SIGKILL`, so `waitpid` is bounded — stragglers are abandoned for init/the test harness to reap. ### Job Cancellation diff --git a/cpp/src/grpc/server/grpc_server_main.cpp b/cpp/src/grpc/server/grpc_server_main.cpp index bda7f0c893..6d923a9520 100644 --- a/cpp/src/grpc/server/grpc_server_main.cpp +++ b/cpp/src/grpc/server/grpc_server_main.cpp @@ -289,6 +289,10 @@ int main(int argc, char** argv) keep_running = false; shm_ctrl->shutdown_requested = true; cancel_all_active_jobs_for_shutdown(); + kill_all_workers(); + // Close our pipe ends so any background thread blocked in write/read + // returns instead of delaying join past Ctrl-C. + close_all_server_worker_pipes(); result_cv.notify_all(); if (result_thread.joinable()) result_thread.join(); @@ -327,7 +331,7 @@ int main(int argc, char** argv) // On signal: cancel active jobs and SIGKILL workers before gRPC Shutdown so // WaitForCompletion / StreamLogs RPCs unblock and we never hang on mid-solve - // workers. Use a deadline in case an RPC is stuck for another reason. + // workers. Use a short deadline in case an RPC is stuck for another reason. std::thread shutdown_thread([&server]() { while (keep_running.load()) { std::this_thread::sleep_for(std::chrono::milliseconds(100)); @@ -336,8 +340,9 @@ int main(int argc, char** argv) if (shm_ctrl) { shm_ctrl->shutdown_requested = true; } cancel_all_active_jobs_for_shutdown(); kill_all_workers(); + close_all_server_worker_pipes(); if (server) { - auto deadline = std::chrono::system_clock::now() + std::chrono::seconds(5); + auto deadline = std::chrono::system_clock::now() + std::chrono::seconds(2); server->Shutdown(deadline); } }); diff --git a/cpp/src/grpc/server/grpc_server_types.hpp b/cpp/src/grpc/server/grpc_server_types.hpp index 062073bae5..9961207a44 100644 --- a/cpp/src/grpc/server/grpc_server_types.hpp +++ b/cpp/src/grpc/server/grpc_server_types.hpp @@ -351,6 +351,7 @@ void log_worker_gpu_layout(); bool init_worker_cuda_environment(int worker_id); void spawn_workers(); void kill_all_workers(); +void close_all_server_worker_pipes(); void wait_for_workers(); void cancel_all_active_jobs_for_shutdown(); void worker_monitor_thread(); diff --git a/cpp/src/grpc/server/grpc_worker_infra.cpp b/cpp/src/grpc/server/grpc_worker_infra.cpp index cdf310c076..c87c06bf16 100644 --- a/cpp/src/grpc/server/grpc_worker_infra.cpp +++ b/cpp/src/grpc/server/grpc_worker_infra.cpp @@ -215,6 +215,14 @@ void kill_all_workers() } } +void close_all_server_worker_pipes() +{ + std::lock_guard lock(worker_pipes_mutex); + for (auto& wp : worker_pipes) { + close_all_worker_pipes(wp); + } +} + void cancel_all_active_jobs_for_shutdown() { if (job_queue != nullptr) { @@ -258,11 +266,39 @@ void wait_for_workers() { // Mid-solve workers only check shutdown_requested between jobs, so force-kill // before waitpid or Ctrl-C / SIGTERM can hang until the current solve ends. + // A worker killed mid-CUDA can sit in uninterruptible D-state while the GPU + // driver tears down; a blocking waitpid would then hang the whole server + // (and ignore SIGTERM because we retry on EINTR). Bound the wait and abandon + // stragglers — the test harness / init reaps them. kill_all_workers(); + + constexpr auto kShutdownWait = std::chrono::seconds(2); + auto deadline = std::chrono::steady_clock::now() + kShutdownWait; + while (std::chrono::steady_clock::now() < deadline) { + bool any_alive = false; + for (pid_t& pid : worker_pids) { + if (pid <= 0) continue; + int status = 0; + pid_t reaped = waitpid(pid, &status, WNOHANG); + if (reaped == pid || (reaped < 0 && errno == ECHILD)) { + pid = 0; + } else if (reaped < 0 && errno == EINTR) { + any_alive = true; + } else { + any_alive = true; + } + } + if (!any_alive) break; + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + } + for (pid_t pid : worker_pids) { - if (pid <= 0) continue; - int status; - while (waitpid(pid, &status, 0) < 0 && errno == EINTR) {} + if (pid > 0) { + kill(pid, SIGKILL); + SERVER_LOG_WARN( + "[Server] Worker pid %d did not exit within shutdown grace period; abandoning", + static_cast(pid)); + } } worker_pids.clear(); } From 38ca744fefd3e5260ddf2682890c6c7e1c468cef Mon Sep 17 00:00:00 2001 From: Trevor McKay Date: Thu, 23 Jul 2026 15:01:40 -0400 Subject: [PATCH 3/7] fix(grpc): handle Ctrl-C via sigwait, not signal() The SIGINT integration test still timed out at 15s after bounded waitpid, which points at shutdown never starting. In a multithreaded gRPC/CUDA process, async signal handlers are often never invoked because library threads mask signals. Block SIGINT/SIGTERM before spawning threads, consume them with sigwait on a dedicated thread, and add a 3s _exit watchdog if cleanup wedges. Signed-off-by: Trevor McKay --- cpp/docs/grpc-server-architecture.md | 4 +- cpp/src/grpc/server/grpc_server_main.cpp | 53 ++++++++++++++++++----- cpp/src/grpc/server/grpc_server_types.hpp | 9 +--- 3 files changed, 48 insertions(+), 18 deletions(-) diff --git a/cpp/docs/grpc-server-architecture.md b/cpp/docs/grpc-server-architecture.md index f8bc372146..3361367069 100644 --- a/cpp/docs/grpc-server-architecture.md +++ b/cpp/docs/grpc-server-architecture.md @@ -291,13 +291,15 @@ If a worker process crashes: ### Graceful Shutdown -On SIGINT/SIGTERM: +On SIGINT/SIGTERM (delivered to a dedicated `sigwait` thread — async `signal()` +handlers are unreliable once gRPC/CUDA threads mask signals): 1. Set `keep_running = false` and `shm_ctrl->shutdown_requested = true` 2. Mark all queued/running jobs `CANCELLED` and wake any `WaitForCompletion` waiters 3. `SIGKILL` all worker processes (mid-solve workers do not poll the shutdown flag) 4. Close server-side worker pipes so background threads blocked on pipe I/O unblock 5. Shut down the gRPC server with a short deadline so lingering RPCs cannot block exit 6. Join background threads, wait briefly for workers (`waitpid` with a grace period), and clean up shared memory +7. A 3s watchdog calls `_exit(0)` if the clean path wedges (e.g. GPU driver D-state) Workers ignore SIGINT/SIGTERM so only the parent process owns shutdown; the parent always force-kills workers rather than waiting for the current solve to finish. Mid-CUDA workers can sit in uninterruptible D-state after `SIGKILL`, so `waitpid` is bounded — stragglers are abandoned for init/the test harness to reap. diff --git a/cpp/src/grpc/server/grpc_server_main.cpp b/cpp/src/grpc/server/grpc_server_main.cpp index 6d923a9520..dc2b1c006c 100644 --- a/cpp/src/grpc/server/grpc_server_main.cpp +++ b/cpp/src/grpc/server/grpc_server_main.cpp @@ -22,6 +22,8 @@ #include #include +#include + // Defined in grpc_service_impl.cpp std::unique_ptr create_cuopt_grpc_service(); @@ -184,8 +186,18 @@ int main(int argc, char** argv) SERVER_LOG_INFO("Port: %d", config.port); SERVER_LOG_INFO("Workers: %d", config.num_workers); - signal(SIGINT, signal_handler); - signal(SIGTERM, signal_handler); + // Block shutdown signals in this thread before spawning workers / gRPC / + // background threads so they inherit the mask. A dedicated sigwait thread + // (started later) is then the only place SIGINT/SIGTERM are consumed. + // Async signal() handlers are unreliable once libraries mask signals. + sigset_t shutdown_sigset; + sigemptyset(&shutdown_sigset); + sigaddset(&shutdown_sigset, SIGINT); + sigaddset(&shutdown_sigset, SIGTERM); + if (pthread_sigmask(SIG_BLOCK, &shutdown_sigset, nullptr) != 0) { + SERVER_LOG_ERROR("[Server] Failed to block shutdown signals: %s", strerror(errno)); + return 1; + } ensure_log_dir_exists(); @@ -329,20 +341,41 @@ int main(int argc, char** argv) server_max_message_bytes() / kMiB); SERVER_LOG_INFO("[gRPC Server] Press Ctrl+C to shutdown"); - // On signal: cancel active jobs and SIGKILL workers before gRPC Shutdown so - // WaitForCompletion / StreamLogs RPCs unblock and we never hang on mid-solve - // workers. Use a short deadline in case an RPC is stuck for another reason. - std::thread shutdown_thread([&server]() { - while (keep_running.load()) { - std::this_thread::sleep_for(std::chrono::milliseconds(100)); + // Dedicated signal thread: sigwait is reliable in multithreaded processes + // where gRPC/CUDA may mask signals on their threads. On SIGINT/SIGTERM we + // cancel jobs, kill workers, close pipes, and shut down gRPC. A watchdog + // forces process exit if cleanup hangs (e.g. GPU driver D-state). + sigset_t shutdown_sigset; + sigemptyset(&shutdown_sigset); + sigaddset(&shutdown_sigset, SIGINT); + sigaddset(&shutdown_sigset, SIGTERM); + + std::thread shutdown_thread([&server, shutdown_sigset]() mutable { + int sig = 0; + int rc = sigwait(&shutdown_sigset, &sig); + if (rc != 0) { + SERVER_LOG_ERROR("[Server] sigwait failed: %s", strerror(rc)); + sig = SIGTERM; } - SERVER_LOG_INFO("[Server] Shutdown signal received; cancelling jobs and killing workers"); + + SERVER_LOG_INFO("[Server] Shutdown signal %d received; cancelling jobs and killing workers", + sig); + keep_running = false; if (shm_ctrl) { shm_ctrl->shutdown_requested = true; } + + // If joins / waitpid / gRPC Shutdown wedge, still die promptly so Ctrl-C + // and the integration test cannot hang indefinitely. + std::thread([]() { + std::this_thread::sleep_for(std::chrono::seconds(3)); + SERVER_LOG_ERROR("[Server] Shutdown watchdog expired; forcing exit"); + _exit(0); + }).detach(); + cancel_all_active_jobs_for_shutdown(); kill_all_workers(); close_all_server_worker_pipes(); if (server) { - auto deadline = std::chrono::system_clock::now() + std::chrono::seconds(2); + auto deadline = std::chrono::system_clock::now() + std::chrono::seconds(1); server->Shutdown(deadline); } }); diff --git a/cpp/src/grpc/server/grpc_server_types.hpp b/cpp/src/grpc/server/grpc_server_types.hpp index 9961207a44..6e07f5e037 100644 --- a/cpp/src/grpc/server/grpc_server_types.hpp +++ b/cpp/src/grpc/server/grpc_server_types.hpp @@ -330,13 +330,8 @@ inline std::string read_file_to_string(const std::string& path) // Signal handling // ============================================================================= -inline void signal_handler(int signal) -{ - if (signal == SIGINT || signal == SIGTERM) { - keep_running = false; - if (shm_ctrl) { shm_ctrl->shutdown_requested = true; } - } -} +// SIGINT/SIGTERM are handled via sigwait on a dedicated thread (see main). +// Using signal() handlers is unreliable once gRPC/CUDA threads mask signals. // ============================================================================= // Forward declarations From bcddd9d7c12815316073736f1d6b8cd5df68c7c5 Mon Sep 17 00:00:00 2001 From: Trevor McKay Date: Thu, 23 Jul 2026 15:21:41 -0400 Subject: [PATCH 4/7] fix(grpc): address CodeRabbit shutdown review notes Serialize worker_pids behind a mutex so the monitor and shutdown kill path cannot race. Have the SIGINT test reap via waitpid so a zombie parent is not treated as still running. Log pthread_sigmask failures with its return code rather than errno. Signed-off-by: Trevor McKay --- cpp/src/grpc/server/grpc_server_main.cpp | 17 ++++- cpp/src/grpc/server/grpc_server_threads.cpp | 66 ++++++++++++------- cpp/src/grpc/server/grpc_server_types.hpp | 1 + cpp/src/grpc/server/grpc_worker_infra.cpp | 53 +++++++++------ .../grpc/grpc_integration_test.cpp | 58 +++++++++++++--- 5 files changed, 140 insertions(+), 55 deletions(-) diff --git a/cpp/src/grpc/server/grpc_server_main.cpp b/cpp/src/grpc/server/grpc_server_main.cpp index dc2b1c006c..5e07454a60 100644 --- a/cpp/src/grpc/server/grpc_server_main.cpp +++ b/cpp/src/grpc/server/grpc_server_main.cpp @@ -194,8 +194,9 @@ int main(int argc, char** argv) sigemptyset(&shutdown_sigset); sigaddset(&shutdown_sigset, SIGINT); sigaddset(&shutdown_sigset, SIGTERM); - if (pthread_sigmask(SIG_BLOCK, &shutdown_sigset, nullptr) != 0) { - SERVER_LOG_ERROR("[Server] Failed to block shutdown signals: %s", strerror(errno)); + int mask_rc = pthread_sigmask(SIG_BLOCK, &shutdown_sigset, nullptr); + if (mask_rc != 0) { + SERVER_LOG_ERROR("[Server] Failed to block shutdown signals: %s", strerror(mask_rc)); return 1; } @@ -275,7 +276,17 @@ int main(int argc, char** argv) log_worker_gpu_layout(); spawn_workers(); - cuopt_expects(!worker_pids.empty(), error_type_t::RuntimeError, "No workers started"); + { + std::lock_guard lock(worker_pids_mutex); + bool any_worker = false; + for (pid_t pid : worker_pids) { + if (pid > 0) { + any_worker = true; + break; + } + } + cuopt_expects(any_worker, error_type_t::RuntimeError, "No workers started"); + } } catch (const cuopt::logic_error& e) { SERVER_LOG_ERROR("[Server] %s", format_cuopt_error(e)); diff --git a/cpp/src/grpc/server/grpc_server_threads.cpp b/cpp/src/grpc/server/grpc_server_threads.cpp index 75fdc39b26..3e634e673c 100644 --- a/cpp/src/grpc/server/grpc_server_threads.cpp +++ b/cpp/src/grpc/server/grpc_server_threads.cpp @@ -14,44 +14,62 @@ void worker_monitor_thread() SERVER_LOG_INFO("[Server] Worker monitor thread started"); while (keep_running) { - for (size_t i = 0; i < worker_pids.size(); ++i) { - pid_t pid = worker_pids[i]; - if (pid <= 0) continue; + // Snapshot which slots need attention under the pid-list lock, then do + // mark/respawn work without holding it across fork(). + struct DeadWorker { + size_t index; + pid_t pid; + bool was_clean_shutdown_exit; + }; + std::vector dead; - int status; - pid_t result = waitpid(pid, &status, WNOHANG); + { + std::lock_guard lock(worker_pids_mutex); + for (size_t i = 0; i < worker_pids.size(); ++i) { + pid_t pid = worker_pids[i]; + if (pid <= 0) continue; + + int status = 0; + pid_t wait_ret = waitpid(pid, &status, WNOHANG); + if (wait_ret != pid) continue; - if (result == pid) { - int exit_code = WIFEXITED(status) ? WEXITSTATUS(status) : -1; - bool signaled = WIFSIGNALED(status); - int signal_num = signaled ? WTERMSIG(status) : 0; + int exit_code = WIFEXITED(status) ? WEXITSTATUS(status) : -1; + bool signaled = WIFSIGNALED(status); + int signal_num = signaled ? WTERMSIG(status) : 0; + bool was_clean_shutdown_exit = false; if (signaled) { SERVER_LOG_ERROR("[Server] Worker %d killed by signal %d", pid, signal_num); } else if (exit_code != 0) { SERVER_LOG_ERROR("[Server] Worker %d exited with code %d", pid, exit_code); + } else if (shm_ctrl && shm_ctrl->shutdown_requested) { + was_clean_shutdown_exit = true; } else { - if (shm_ctrl && shm_ctrl->shutdown_requested) { - worker_pids[i] = 0; - continue; - } SERVER_LOG_ERROR("[Server] Worker %d exited unexpectedly", pid); } - mark_worker_jobs_failed(pid); + worker_pids[i] = 0; + dead.push_back({i, pid, was_clean_shutdown_exit}); + } + } - if (keep_running && shm_ctrl && !shm_ctrl->shutdown_requested) { - pid_t new_pid = spawn_single_worker(static_cast(i)); - if (new_pid > 0) { - worker_pids[i] = new_pid; - SERVER_LOG_INFO("[Server] Restarted worker %zu with PID %d", i, new_pid); - } else { - worker_pids[i] = 0; - } - } else { - worker_pids[i] = 0; + for (const auto& dw : dead) { + if (dw.was_clean_shutdown_exit) continue; + + mark_worker_jobs_failed(dw.pid); + + if (!(keep_running && shm_ctrl && !shm_ctrl->shutdown_requested)) { continue; } + + pid_t new_pid = spawn_single_worker(static_cast(dw.index)); + { + std::lock_guard lock(worker_pids_mutex); + if (dw.index < worker_pids.size() && worker_pids[dw.index] == 0) { + worker_pids[dw.index] = (new_pid > 0) ? new_pid : 0; } } + if (new_pid > 0) { + SERVER_LOG_INFO("[Server] Restarted worker %zu with PID %d", dw.index, new_pid); + } } std::this_thread::sleep_for(std::chrono::milliseconds(100)); diff --git a/cpp/src/grpc/server/grpc_server_types.hpp b/cpp/src/grpc/server/grpc_server_types.hpp index 6e07f5e037..dd1766ffaf 100644 --- a/cpp/src/grpc/server/grpc_server_types.hpp +++ b/cpp/src/grpc/server/grpc_server_types.hpp @@ -247,6 +247,7 @@ inline ResultQueueEntry* result_queue = nullptr; inline SharedMemoryControl* shm_ctrl = nullptr; inline std::vector worker_pids; +inline std::mutex worker_pids_mutex; inline ServerConfig config; diff --git a/cpp/src/grpc/server/grpc_worker_infra.cpp b/cpp/src/grpc/server/grpc_worker_infra.cpp index c87c06bf16..e4164c3ad3 100644 --- a/cpp/src/grpc/server/grpc_worker_infra.cpp +++ b/cpp/src/grpc/server/grpc_worker_infra.cpp @@ -201,15 +201,19 @@ pid_t spawn_worker(int worker_id, bool is_replacement) void spawn_workers() { + std::lock_guard lock(worker_pids_mutex); + // Index i is worker_id: keep failed startups as 0 so monitor/respawn and + // pipe tables stay aligned even when some initial forks fail. + worker_pids.assign(static_cast(config.num_workers), 0); for (int i = 0; i < config.num_workers; ++i) { pid_t pid = spawn_worker(i, false); - if (pid < 0) { continue; } - worker_pids.push_back(pid); + if (pid > 0) { worker_pids[static_cast(i)] = pid; } } } void kill_all_workers() { + std::lock_guard lock(worker_pids_mutex); for (pid_t pid : worker_pids) { if (pid > 0) { kill(pid, SIGKILL); } } @@ -276,31 +280,40 @@ void wait_for_workers() auto deadline = std::chrono::steady_clock::now() + kShutdownWait; while (std::chrono::steady_clock::now() < deadline) { bool any_alive = false; - for (pid_t& pid : worker_pids) { - if (pid <= 0) continue; - int status = 0; - pid_t reaped = waitpid(pid, &status, WNOHANG); - if (reaped == pid || (reaped < 0 && errno == ECHILD)) { - pid = 0; - } else if (reaped < 0 && errno == EINTR) { - any_alive = true; - } else { - any_alive = true; + { + std::lock_guard lock(worker_pids_mutex); + for (pid_t& pid : worker_pids) { + if (pid <= 0) continue; + int status = 0; + pid_t reaped = waitpid(pid, &status, WNOHANG); + if (reaped == pid || (reaped < 0 && errno == ECHILD)) { + pid = 0; + } else if (reaped < 0 && errno == EINTR) { + any_alive = true; + } else { + any_alive = true; + } + } + if (!any_alive) { + worker_pids.clear(); + return; } } - if (!any_alive) break; std::this_thread::sleep_for(std::chrono::milliseconds(50)); } - for (pid_t pid : worker_pids) { - if (pid > 0) { - kill(pid, SIGKILL); - SERVER_LOG_WARN( - "[Server] Worker pid %d did not exit within shutdown grace period; abandoning", - static_cast(pid)); + { + std::lock_guard lock(worker_pids_mutex); + for (pid_t pid : worker_pids) { + if (pid > 0) { + kill(pid, SIGKILL); + SERVER_LOG_WARN( + "[Server] Worker pid %d did not exit within shutdown grace period; abandoning", + static_cast(pid)); + } } + worker_pids.clear(); } - worker_pids.clear(); } pid_t spawn_single_worker(int worker_id) { return spawn_worker(worker_id, true); } diff --git a/cpp/tests/linear_programming/grpc/grpc_integration_test.cpp b/cpp/tests/linear_programming/grpc/grpc_integration_test.cpp index 87641fef09..6bfdfda383 100644 --- a/cpp/tests/linear_programming/grpc/grpc_integration_test.cpp +++ b/cpp/tests/linear_programming/grpc/grpc_integration_test.cpp @@ -198,6 +198,36 @@ class ServerProcess { return kill(pid_, 0) == 0; } + // Wait until the server parent has exited and been reaped. Unlike is_running(), + // this treats a zombie as exited (kill(pid,0) is true for zombies). + // Returns true once the parent has exited. If leftover workers are still + // alive (e.g. GPU D-state), lifecycle state is preserved so TearDown's + // stop() can finish draining the process group. + bool wait_exited(std::chrono::milliseconds timeout) + { + if (pid_ <= 0) return true; + + auto deadline = std::chrono::steady_clock::now() + timeout; + while (true) { + int status = 0; + pid_t ret = waitpid(pid_, &status, WNOHANG); + if (ret == pid_ || (ret < 0 && errno == ECHILD)) { + auto remaining = std::chrono::duration_cast( + deadline - std::chrono::steady_clock::now()); + if (remaining > std::chrono::milliseconds(2000)) { + remaining = std::chrono::milliseconds(2000); + } + if (remaining.count() < 0) { remaining = std::chrono::milliseconds(0); } + if (reap_group(remaining)) { clear_lifecycle_state(); } + // Parent is gone either way; leave pid_/pgid_ intact if the group is + // not fully drained so TearDown can still signal it. + return true; + } + if (std::chrono::steady_clock::now() >= deadline) { return false; } + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + } + } + std::string log_path() const { if (port_ <= 0) return ""; @@ -1711,21 +1741,33 @@ TEST_F(ErrorRecoveryTests, SigintDuringRunningJobShutsDownPromptly) auto submit_result = client->submit_mip(problem, settings); ASSERT_TRUE(submit_result.success); + std::string job_id = submit_result.job_id; - std::this_thread::sleep_for(std::chrono::seconds(2)); + // Wait until a worker has claimed the job so SIGINT interrupts an active solve. + bool processing = false; + for (int i = 0; i < 40; ++i) { + auto status = client->check_status(job_id); + if (status.status == job_status_t::PROCESSING) { + processing = true; + break; + } + if (status.status == job_status_t::COMPLETED || status.status == job_status_t::FAILED || + status.status == job_status_t::CANCELLED) { + break; + } + std::this_thread::sleep_for(std::chrono::milliseconds(250)); + } + ASSERT_TRUE(processing) << "Job never reached PROCESSING before SIGINT"; ASSERT_TRUE(server_.is_running()); auto start = std::chrono::steady_clock::now(); ASSERT_EQ(kill(server_.pid(), SIGINT), 0); // Server must exit well before the solve time_limit; previously Ctrl-C could - // hang until the mid-solve worker finished. - while (server_.is_running()) { - auto elapsed = std::chrono::steady_clock::now() - start; - ASSERT_LT(elapsed, std::chrono::seconds(15)) - << "Server did not shut down promptly after SIGINT during a running job"; - std::this_thread::sleep_for(std::chrono::milliseconds(100)); - } + // hang until the mid-solve worker finished. Use waitpid-based polling so a + // zombie parent is not mistaken for a still-running process. + ASSERT_TRUE(server_.wait_exited(std::chrono::seconds(15))) + << "Server did not shut down promptly after SIGINT during a running job"; auto elapsed = std::chrono::duration_cast(std::chrono::steady_clock::now() - start); From cf60d12b2d7b08ea239c8840e65be403afd2acab Mon Sep 17 00:00:00 2001 From: Trevor McKay Date: Fri, 24 Jul 2026 10:27:15 -0400 Subject: [PATCH 5/7] grpc server log failure of worker restart Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> --- cpp/src/grpc/server/grpc_server_threads.cpp | 2 ++ 1 file changed, 2 insertions(+) diff --git a/cpp/src/grpc/server/grpc_server_threads.cpp b/cpp/src/grpc/server/grpc_server_threads.cpp index 3e634e673c..4f8a2d44d4 100644 --- a/cpp/src/grpc/server/grpc_server_threads.cpp +++ b/cpp/src/grpc/server/grpc_server_threads.cpp @@ -69,6 +69,8 @@ void worker_monitor_thread() } if (new_pid > 0) { SERVER_LOG_INFO("[Server] Restarted worker %zu with PID %d", dw.index, new_pid); + } else { + SERVER_LOG_ERROR("[Server] Failed to restart worker %zu", dw.index); } } From e02c544236d36ac180f851e9b8972af29ee7ebb4 Mon Sep 17 00:00:00 2001 From: Trevor McKay Date: Fri, 24 Jul 2026 14:37:37 -0400 Subject: [PATCH 6/7] fix(grpc): make shutdown watchdog cancelable with margin The previous 3s _exit(0) watchdog could fire during a healthy shutdown whose bounded work alone is ~3s (gRPC deadline + waitpid grace) before thread joins. Disarm the watchdog after cleanup completes, give it a 10s margin, and use a nonzero forced-exit status. Signed-off-by: Trevor McKay --- cpp/docs/grpc-server-architecture.md | 5 +++- cpp/src/grpc/server/grpc_server_main.cpp | 32 +++++++++++++++++------- 2 files changed, 27 insertions(+), 10 deletions(-) diff --git a/cpp/docs/grpc-server-architecture.md b/cpp/docs/grpc-server-architecture.md index a869293d25..1af10da629 100644 --- a/cpp/docs/grpc-server-architecture.md +++ b/cpp/docs/grpc-server-architecture.md @@ -299,7 +299,10 @@ handlers are unreliable once gRPC/CUDA threads mask signals): 4. Close server-side worker pipes so background threads blocked on pipe I/O unblock 5. Shut down the gRPC server with a short deadline so lingering RPCs cannot block exit 6. Join background threads, wait briefly for workers (`waitpid` with a grace period), and clean up shared memory -7. A 3s watchdog calls `_exit(0)` if the clean path wedges (e.g. GPU driver D-state) +7. A cancelable ~10s watchdog calls `_exit(1)` if the clean path wedges (e.g. GPU + driver D-state). The margin is above the bounded clean-path work (gRPC + Shutdown deadline + worker wait grace + thread joins). On a healthy + shutdown the watchdog is disarmed before process exit. Workers ignore SIGINT/SIGTERM so only the parent process owns shutdown; the parent always force-kills workers rather than waiting for the current solve to finish. Mid-CUDA workers can sit in uninterruptible D-state after `SIGKILL`, so `waitpid` is bounded — stragglers are abandoned for init/the test harness to reap. diff --git a/cpp/src/grpc/server/grpc_server_main.cpp b/cpp/src/grpc/server/grpc_server_main.cpp index 5e07454a60..050fd7559c 100644 --- a/cpp/src/grpc/server/grpc_server_main.cpp +++ b/cpp/src/grpc/server/grpc_server_main.cpp @@ -354,14 +354,21 @@ int main(int argc, char** argv) // Dedicated signal thread: sigwait is reliable in multithreaded processes // where gRPC/CUDA may mask signals on their threads. On SIGINT/SIGTERM we - // cancel jobs, kill workers, close pipes, and shut down gRPC. A watchdog - // forces process exit if cleanup hangs (e.g. GPU driver D-state). + // cancel jobs, kill workers, close pipes, and shut down gRPC. A cancelable + // watchdog forces process exit only if cleanup hangs (e.g. GPU D-state). sigset_t shutdown_sigset; sigemptyset(&shutdown_sigset); sigaddset(&shutdown_sigset, SIGINT); sigaddset(&shutdown_sigset, SIGTERM); - std::thread shutdown_thread([&server, shutdown_sigset]() mutable { + // Shared so the detached watchdog can observe cancellation after a healthy + // shutdown_all() completes. Margin is well above the bounded pieces of the + // clean path (gRPC Shutdown deadline 1s + wait_for_workers grace 2s + joins). + // static constexpr: nested lambdas can use it without capturing. + static constexpr auto kShutdownWatchdog = std::chrono::seconds(10); + auto shutdown_watchdog_cancelled = std::make_shared>(false); + + std::thread shutdown_thread([&server, shutdown_sigset, shutdown_watchdog_cancelled]() mutable { int sig = 0; int rc = sigwait(&shutdown_sigset, &sig); if (rc != 0) { @@ -374,12 +381,16 @@ int main(int argc, char** argv) keep_running = false; if (shm_ctrl) { shm_ctrl->shutdown_requested = true; } - // If joins / waitpid / gRPC Shutdown wedge, still die promptly so Ctrl-C - // and the integration test cannot hang indefinitely. - std::thread([]() { - std::this_thread::sleep_for(std::chrono::seconds(3)); - SERVER_LOG_ERROR("[Server] Shutdown watchdog expired; forcing exit"); - _exit(0); + std::thread([shutdown_watchdog_cancelled]() { + auto deadline = std::chrono::steady_clock::now() + kShutdownWatchdog; + while (!shutdown_watchdog_cancelled->load(std::memory_order_acquire)) { + if (std::chrono::steady_clock::now() >= deadline) { + SERVER_LOG_ERROR("[Server] Shutdown watchdog expired after %llds; forcing unclean exit", + static_cast(kShutdownWatchdog.count())); + _exit(1); + } + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + } }).detach(); cancel_all_active_jobs_for_shutdown(); @@ -397,6 +408,9 @@ int main(int argc, char** argv) SERVER_LOG_INFO("[Server] Shutting down..."); shutdown_all(); + // Disarm the watchdog so a healthy cleanup cannot race into _exit(1). + shutdown_watchdog_cancelled->store(true, std::memory_order_release); + SERVER_LOG_INFO("[Server] Shutdown complete"); return 0; } From eefa9b380df579ce87653e44e12af567c355f1c3 Mon Sep 17 00:00:00 2001 From: Trevor McKay Date: Mon, 27 Jul 2026 10:45:44 -0400 Subject: [PATCH 7/7] fix(grpc): make read and write pipe loops poll and check shutdown --- cpp/docs/grpc-server-architecture.md | 9 +- cpp/src/grpc/server/grpc_pipe_io.cpp | 107 ++++++++++++++++------ cpp/src/grpc/server/grpc_server_main.cpp | 11 ++- cpp/src/grpc/server/grpc_worker_infra.cpp | 10 ++ 4 files changed, 103 insertions(+), 34 deletions(-) diff --git a/cpp/docs/grpc-server-architecture.md b/cpp/docs/grpc-server-architecture.md index 1af10da629..c79b183754 100644 --- a/cpp/docs/grpc-server-architecture.md +++ b/cpp/docs/grpc-server-architecture.md @@ -296,9 +296,12 @@ handlers are unreliable once gRPC/CUDA threads mask signals): 1. Set `keep_running = false` and `shm_ctrl->shutdown_requested = true` 2. Mark all queued/running jobs `CANCELLED` and wake any `WaitForCompletion` waiters 3. `SIGKILL` all worker processes (mid-solve workers do not poll the shutdown flag) -4. Close server-side worker pipes so background threads blocked on pipe I/O unblock -5. Shut down the gRPC server with a short deadline so lingering RPCs cannot block exit -6. Join background threads, wait briefly for workers (`waitpid` with a grace period), and clean up shared memory +4. Shut down the gRPC server with a short deadline so lingering RPCs cannot block exit +5. Join background threads (pipe I/O is interruptible via `shutdown_requested` + + `O_NONBLOCK`/`poll`; do **not** close pipe FDs while writers may still be in + `write()` — closing from another thread does not unblock an in-flight write) +6. Close server-side worker pipes, wait briefly for workers (`waitpid` with a grace + period), and clean up shared memory 7. A cancelable ~10s watchdog calls `_exit(1)` if the clean path wedges (e.g. GPU driver D-state). The margin is above the bounded clean-path work (gRPC Shutdown deadline + worker wait grace + thread joins). On a healthy diff --git a/cpp/src/grpc/server/grpc_pipe_io.cpp b/cpp/src/grpc/server/grpc_pipe_io.cpp index 3bcf445311..2f507e8de3 100644 --- a/cpp/src/grpc/server/grpc_pipe_io.cpp +++ b/cpp/src/grpc/server/grpc_pipe_io.cpp @@ -6,26 +6,61 @@ #ifdef CUOPT_ENABLE_GRPC #include "grpc_server_logger.hpp" +#include "grpc_server_types.hpp" +#include #include +#include #include #include #include #include +namespace { + +bool shutdown_requested() +{ + return shm_ctrl != nullptr && shm_ctrl->shutdown_requested.load(std::memory_order_acquire); +} + +} // namespace + bool write_to_pipe(int fd, const void* data, size_t size) { const uint8_t* ptr = static_cast(data); size_t remaining = size; + + // Interruptible write: poll with a short timeout and abort when shutdown is + // requested. Callers must use O_NONBLOCK write ends so write() returns + // EAGAIN instead of blocking after poll reports POLLOUT for a partial window. while (remaining > 0) { - ssize_t written = ::write(fd, ptr, remaining); - if (written <= 0) { - if (errno == EINTR) continue; + if (shutdown_requested()) { return false; } + + struct pollfd pfd = {fd, POLLOUT, 0}; + int pr; + do { + pr = poll(&pfd, 1, 100); + } while (pr < 0 && errno == EINTR); + if (pr < 0) { + SERVER_LOG_ERROR("[Server] poll() failed on pipe write: %s", strerror(errno)); + return false; + } + if (pr == 0) { continue; } + if (pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) { + SERVER_LOG_ERROR("[Server] Pipe write error/hangup detected"); return false; } - ptr += written; - remaining -= written; + + ssize_t written = ::write(fd, ptr, remaining); + if (written > 0) { + ptr += written; + remaining -= written; + continue; + } + if (written < 0 && (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK)) { continue; } + SERVER_LOG_ERROR("[Server] Pipe write error: %s", strerror(errno)); + return false; } return true; } @@ -35,31 +70,49 @@ bool read_from_pipe(int fd, void* data, size_t size, int timeout_ms) uint8_t* ptr = static_cast(data); size_t remaining = size; - // Poll once to enforce timeout before the first read. After data starts - // flowing, blocking read() is sufficient — if the writer dies the pipe - // closes and read() returns 0 (EOF). Avoids ~10k extra poll() syscalls - // per bulk transfer. - struct pollfd pfd = {fd, POLLIN, 0}; - int pr; - do { - pr = poll(&pfd, 1, timeout_ms); - } while (pr < 0 && errno == EINTR); - if (pr < 0) { - SERVER_LOG_ERROR("[Server] poll() failed on pipe: %s", strerror(errno)); - return false; - } - if (pr == 0) { - SERVER_LOG_ERROR("[Server] Timeout waiting for pipe data (waited %dms)", timeout_ms); - return false; - } - if (pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) { - SERVER_LOG_ERROR("[Server] Pipe error/hangup detected"); - return false; - } + // timeout_ms only bounds waiting for the *first* readable byte. + // Once data starts flowing, the transfer is open-ended aside + // from shutdown checks and EOF/errors. Poll in short slices so shutdown can + // abort while waiting. + auto first_deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(timeout_ms); + bool saw_data = false; while (remaining > 0) { + if (shutdown_requested()) { return false; } + + int wait_ms = 100; + if (!saw_data) { + auto now = std::chrono::steady_clock::now(); + if (now >= first_deadline) { + SERVER_LOG_ERROR("[Server] Timeout waiting for pipe data (waited %dms)", timeout_ms); + return false; + } + auto remaining_ms = + std::chrono::duration_cast(first_deadline - now).count(); + wait_ms = static_cast(std::min(remaining_ms, 100)); + } + + struct pollfd pfd = {fd, POLLIN, 0}; + int pr; + do { + pr = poll(&pfd, 1, wait_ms); + } while (pr < 0 && errno == EINTR); + if (pr < 0) { + SERVER_LOG_ERROR("[Server] poll() failed on pipe: %s", strerror(errno)); + return false; + } + if (pr == 0) { continue; } + if (pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) { + // POLLHUP with POLLIN can still have readable bytes; only fail if no POLLIN. + if (!(pfd.revents & POLLIN)) { + SERVER_LOG_ERROR("[Server] Pipe error/hangup detected"); + return false; + } + } + ssize_t nread = ::read(fd, ptr, remaining); if (nread > 0) { + saw_data = true; ptr += nread; remaining -= nread; continue; @@ -68,7 +121,7 @@ bool read_from_pipe(int fd, void* data, size_t size, int timeout_ms) SERVER_LOG_ERROR("[Server] Pipe EOF (writer closed)"); return false; } - if (errno == EINTR) continue; + if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) { continue; } SERVER_LOG_ERROR("[Server] Pipe read error: %s", strerror(errno)); return false; } diff --git a/cpp/src/grpc/server/grpc_server_main.cpp b/cpp/src/grpc/server/grpc_server_main.cpp index 050fd7559c..d307f5ede0 100644 --- a/cpp/src/grpc/server/grpc_server_main.cpp +++ b/cpp/src/grpc/server/grpc_server_main.cpp @@ -313,16 +313,17 @@ int main(int argc, char** argv) shm_ctrl->shutdown_requested = true; cancel_all_active_jobs_for_shutdown(); kill_all_workers(); - // Close our pipe ends so any background thread blocked in write/read - // returns instead of delaying join past Ctrl-C. - close_all_server_worker_pipes(); result_cv.notify_all(); + // Join writers/readers before closing FDs. Closing a pipe FD from another + // thread does not unblock an in-flight write on Linux and can race with + // FD-number reuse. Pipe I/O aborts via shutdown_requested + O_NONBLOCK. if (result_thread.joinable()) result_thread.join(); if (incumbent_thread.joinable()) incumbent_thread.join(); if (monitor_thread.joinable()) monitor_thread.join(); if (reaper_thread.joinable()) reaper_thread.join(); + close_all_server_worker_pipes(); wait_for_workers(); cleanup_shared_memory(); }; @@ -395,7 +396,9 @@ int main(int argc, char** argv) cancel_all_active_jobs_for_shutdown(); kill_all_workers(); - close_all_server_worker_pipes(); + // Do not close server-side pipes here: result/incumbent threads may still + // be in write_to_pipe/read_from_pipe. They abort via shutdown_requested; + // shutdown_all() closes FDs after those threads join. if (server) { auto deadline = std::chrono::system_clock::now() + std::chrono::seconds(1); server->Shutdown(deadline); diff --git a/cpp/src/grpc/server/grpc_worker_infra.cpp b/cpp/src/grpc/server/grpc_worker_infra.cpp index e4164c3ad3..139d86b5ff 100644 --- a/cpp/src/grpc/server/grpc_worker_infra.cpp +++ b/cpp/src/grpc/server/grpc_worker_infra.cpp @@ -121,6 +121,8 @@ bool create_worker_pipes(int worker_id) wp.worker_read_fd = fds[0]; wp.to_worker_fd = fds[1]; fcntl(wp.to_worker_fd, F_SETPIPE_SZ, kPipeBufferSize); + // Nonblocking write end: write_to_pipe polls + retries so shutdown can abort. + fcntl(wp.to_worker_fd, F_SETFL, fcntl(wp.to_worker_fd, F_GETFL) | O_NONBLOCK); if (pipe(fds) < 0) { SERVER_LOG_ERROR("[Server] Failed to create output pipe for worker %d", worker_id); @@ -130,6 +132,9 @@ bool create_worker_pipes(int worker_id) wp.from_worker_fd = fds[0]; wp.worker_write_fd = fds[1]; fcntl(wp.worker_write_fd, F_SETPIPE_SZ, kPipeBufferSize); + fcntl(wp.worker_write_fd, F_SETFL, fcntl(wp.worker_write_fd, F_GETFL) | O_NONBLOCK); + // Parent read end is also nonblocking; read_from_pipe polls before read. + fcntl(wp.from_worker_fd, F_SETFL, fcntl(wp.from_worker_fd, F_GETFL) | O_NONBLOCK); if (pipe(fds) < 0) { SERVER_LOG_ERROR("[Server] Failed to create incumbent pipe for worker %d", worker_id); @@ -138,6 +143,11 @@ bool create_worker_pipes(int worker_id) } wp.incumbent_from_worker_fd = fds[0]; wp.worker_incumbent_write_fd = fds[1]; + fcntl(wp.worker_incumbent_write_fd, + F_SETFL, + fcntl(wp.worker_incumbent_write_fd, F_GETFL) | O_NONBLOCK); + fcntl( + wp.incumbent_from_worker_fd, F_SETFL, fcntl(wp.incumbent_from_worker_fd, F_GETFL) | O_NONBLOCK); return true; }