Skip to content
Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 11 additions & 5 deletions cpp/docs/grpc-server-architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -291,11 +291,17 @@ 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. 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.

### Job Cancellation

Expand Down
61 changes: 55 additions & 6 deletions cpp/src/grpc/server/grpc_server_main.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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();

Expand Down Expand Up @@ -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;
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Outdated
}

ensure_log_dir_exists();

Expand Down Expand Up @@ -288,6 +300,11 @@ 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();
// 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();
Expand Down Expand Up @@ -324,11 +341,43 @@ 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 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 %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(1);
server->Shutdown(deadline);
}
if (server) { server->Shutdown(); }
});

server->Wait();
Expand Down
12 changes: 5 additions & 7 deletions cpp/src/grpc/server/grpc_server_types.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -350,7 +345,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();
Expand Down
6 changes: 6 additions & 0 deletions cpp/src/grpc/server/grpc_worker.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
91 changes: 88 additions & 3 deletions cpp/src/grpc/server/grpc_worker_infra.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -208,12 +208,97 @@ void spawn_workers()
}
}

void kill_all_workers()
{
for (pid_t pid : worker_pids) {
if (pid > 0) { kill(pid, SIGKILL); }
}
}
Comment thread
tmckayus marked this conversation as resolved.

void close_all_server_worker_pipes()
{
std::lock_guard<std::mutex> 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<std::mutex> 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<std::mutex> wlock(waiters_mutex);
for (auto& [job_id, waiter] : waiting_threads) {
(void)job_id;
{
std::lock_guard<std::mutex> 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;
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<int>(pid));
}
}
worker_pids.clear();
}
Expand Down
37 changes: 37 additions & 0 deletions cpp/tests/linear_programming/grpc/grpc_integration_test.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<int32_t, double> 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::seconds>(std::chrono::steady_clock::now() - start);
EXPECT_LT(elapsed.count(), 15);
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.

TEST_F(ErrorRecoveryTests, ClientTimeoutConfiguration)
{
ASSERT_TRUE(start_server());
Expand Down
1 change: 1 addition & 0 deletions docs/cuopt/source/cuopt-grpc/grpc-server-architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
Loading