diff --git a/cpp/docs/grpc-server-architecture.md b/cpp/docs/grpc-server-architecture.md index cfbf9ab51..c79b18375 100644 --- a/cpp/docs/grpc-server-architecture.md +++ b/cpp/docs/grpc-server-architecture.md @@ -291,11 +291,23 @@ 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 +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. 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 + 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. ### Job Cancellation diff --git a/cpp/src/grpc/server/grpc_pipe_io.cpp b/cpp/src/grpc/server/grpc_pipe_io.cpp index 3bcf44531..2f507e8de 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 ae6cdb136..d307f5ede 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,19 @@ 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); + 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; + } ensure_log_dir_exists(); @@ -263,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)); @@ -288,13 +311,19 @@ int main(int argc, char** argv) auto shutdown_all = [&]() { keep_running = false; shm_ctrl->shutdown_requested = true; + cancel_all_active_jobs_for_shutdown(); + kill_all_workers(); 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(); }; @@ -324,11 +353,56 @@ int main(int argc, char** argv) server_max_message_bytes() / kMiB); SERVER_LOG_INFO("[gRPC Server] Press Ctrl+C to shutdown"); - 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 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); + + // 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) { + SERVER_LOG_ERROR("[Server] sigwait failed: %s", strerror(rc)); + sig = SIGTERM; + } + + 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; } + + 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(); + kill_all_workers(); + // 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); } - if (server) { server->Shutdown(); } }); server->Wait(); @@ -337,6 +411,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; } diff --git a/cpp/src/grpc/server/grpc_server_threads.cpp b/cpp/src/grpc/server/grpc_server_threads.cpp index 75fdc39b2..4f8a2d44d 100644 --- a/cpp/src/grpc/server/grpc_server_threads.cpp +++ b/cpp/src/grpc/server/grpc_server_threads.cpp @@ -14,44 +14,64 @@ 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); + } else { + SERVER_LOG_ERROR("[Server] Failed to restart worker %zu", dw.index); + } } 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 cc96972cc..2a0b231fe 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; @@ -330,13 +331,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 @@ -350,7 +346,10 @@ 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 close_all_server_worker_pipes(); 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 373652f66..3d0d00f83 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 7097badf1..139d86b5f 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; } @@ -201,21 +211,119 @@ 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 wait_for_workers() +void kill_all_workers() { + std::lock_guard lock(worker_pids_mutex); 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); } + } +} + +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) { + 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. + // 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; + { + 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; + } + } + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + } + + { + 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 4d2ab1e48..6a3b0cf7d 100644 --- a/cpp/tests/linear_programming/grpc/grpc_integration_test.cpp +++ b/cpp/tests/linear_programming/grpc/grpc_integration_test.cpp @@ -190,12 +190,44 @@ class ServerProcess { int port() const { return port_; } + pid_t pid() const { return pid_; } + bool is_running() const { if (pid_ <= 0) return false; 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 ""; @@ -1816,6 +1848,53 @@ 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::string job_id = submit_result.job_id; + + // 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. 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); + 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 364341046..450947de6 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. - **`DeleteResult`** also cancels a queued or running job (same kill/skip behavior as ``CancelJob``), then removes all server-side state for that ``job_id``. ## Further reading