fix(grpc): terminate workers on server shutdown

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
 

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 <algorithm>
 #include <cerrno>
+#include <chrono>
 #include <cstdint>
 #include <cstring>
 
 #include <poll.h>
 #include <unistd.h>
 
+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<const uint8_t*>(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<uint8_t*>(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<std::chrono::milliseconds>(first_deadline - now).count();
+      wait_ms = static_cast<int>(std::min<long>(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;
   }

cpp/src/grpc/server/grpc_server_main.cpp

@@ -22,6 +22,8 @@
 #include <argparse/argparse.hpp>
 #include <cuopt/version_config.hpp>
 
+#include <pthread.h>
+
 // Defined in grpc_service_impl.cpp
 std::unique_ptr<grpc::Service> 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<std::mutex> 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<std::atomic<bool>>(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<long long>(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;
 }