ci(test): fix grpc server GPU-memory leak and run cuopt tests in parallel (#1512)

## Summary

- Reap `cuopt_grpc_server` reliably — start it in its own process group and kill the whole group on teardown, so the server and its `--workers` child are cleaned up instead of leaking their GPU context across tests/runs.
- Group grpc-server-backed test classes onto a single xdist worker (`--dist loadgroup`) so at most one server runs at a time.
- Run the cuopt Python tests in parallel at `-n 4`, with `--max-worker-restart=0` so a crashed worker fails once instead of respawning into a crash loop.
- Skip the incumbent-callback test when the problem is solved at the root node (no branch-and-bound, so no incumbent callbacks fire), and fix a latent always-true termination assertion.

Authors:
  - Ramakrishna Prabhu (https://github.com/ramakrishnap-nv)

Approvers:
  - Trevor McKay (https://github.com/tmckayus)

URL: #1512

Author: ramakrishnap-nv

ci/run_cuopt_pytests.sh

@@ -36,7 +36,9 @@ rc=0
 if [ "${IS_NIGHTLY}" = "nightly" ]; then
     pytest -s --cache-clear --reruns 2 --reruns-delay 5 -p cuopt_rerun_xml "$@" tests || rc=$?
 else
-    pytest -s --cache-clear -n 1 "$@" tests || rc=$?
+    # loadgroup keeps xdist_group (grpc server) tests on one worker;
+    # max-worker-restart=0 stops a crashed worker from respawning.
+    pytest -s --cache-clear -n 4 --dist loadgroup --max-worker-restart=0 "$@" tests || rc=$?
 fi
 
 # If not a crash, exit normally

python/cuopt/cuopt/tests/fixtures/grpc_server_fixtures.py

@@ -116,6 +116,54 @@ def server_env():
     return env
 
 
+def _set_pdeathsig():
+    """SIGKILL this child if the spawning worker dies (Linux; best effort)."""
+    try:
+        import ctypes
+
+        PR_SET_PDEATHSIG = 1
+        ctypes.CDLL("libc.so.6", use_errno=True).prctl(
+            PR_SET_PDEATHSIG, signal.SIGKILL
+        )
+    except Exception:
+        pass
+
+
+def spawn_server(cmd, env=None):
+    """Start ``cuopt_grpc_server`` in its own process group, dying with the
+    spawning worker, so it and its ``--workers`` child can be reaped together
+    (see ``kill_server``) and don't leak GPU memory.
+    """
+    return subprocess.Popen(
+        cmd,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+        env=env,
+        start_new_session=True,
+        preexec_fn=_set_pdeathsig,
+    )
+
+
+def kill_server(proc):
+    """Terminate the server's whole process group (parent + ``--workers`` child)."""
+    if proc is None:
+        return
+    try:
+        pgid = os.getpgid(proc.pid)
+    except (ProcessLookupError, OSError):
+        return
+    for sig in (signal.SIGTERM, signal.SIGKILL):
+        try:
+            os.killpg(pgid, sig)
+        except (ProcessLookupError, OSError):
+            return
+        try:
+            proc.wait(timeout=5)
+            return
+        except subprocess.TimeoutExpired:
+            continue
+
+
 # Backward-compatible alias used by tests that yield client env dicts.
 cpu_only_env = client_remote_env
 
@@ -128,7 +176,7 @@ def start_grpc_server(port_offset):
 
     port = int(os.environ.get("CUOPT_TEST_PORT_BASE", "18000")) + port_offset
     client_env = client_remote_env(port)
-    proc = subprocess.Popen(
+    proc = spawn_server(
         [
             server_bin,
             "--port",
@@ -137,8 +185,6 @@ def start_grpc_server(port_offset):
             "1",
             "--log-to-console",
         ],
-        stdout=subprocess.DEVNULL,
-        stderr=subprocess.DEVNULL,
         env=server_env(),
     )
     time.sleep(0.5)
@@ -148,8 +194,7 @@ def start_grpc_server(port_offset):
             "binary may be unable to load shared libraries in this environment"
         )
     if not wait_for_grpc_client(port, timeout=30):
-        proc.kill()
-        proc.wait()
+        kill_server(proc)
         pytest.fail(
             "cuopt_grpc_server TCP port opened but gRPC client could not connect "
             "within 30s"
@@ -159,17 +204,8 @@ def start_grpc_server(port_offset):
 
 
 def stop_grpc_server(proc):
-    """Gracefully shut down a server process."""
-    if proc.poll() is not None:
-        proc.wait()
-        return
-
-    proc.send_signal(signal.SIGTERM)
-    try:
-        proc.wait(timeout=5)
-    except subprocess.TimeoutExpired:
-        proc.kill()
-        proc.wait()
+    """Gracefully shut down a server process and its worker child."""
+    kill_server(proc)
 
 
 @pytest.fixture(scope="class")

python/cuopt/cuopt/tests/linear_programming/test_cpu_only_execution.py

@@ -17,7 +17,6 @@
 import os
 import re
 import shutil
-import signal
 import socket
 import subprocess
 import sys
@@ -28,6 +27,14 @@
 from cuopt import linear_programming
 from cuopt.linear_programming.solver.solver_parameters import CUOPT_TIME_LIMIT
 
+# Also re-executed as a standalone script (_run_in_subprocess), where conftest
+# hasn't added fixtures/ to sys.path; add it so this import works either way.
+sys.path.insert(
+    0,
+    os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "fixtures"),
+)
+from grpc_server_fixtures import kill_server, spawn_server  # noqa: E402
+
 logger = logging.getLogger(__name__)
 
 RAPIDS_DATASET_ROOT_DIR = os.environ.get(
@@ -79,7 +86,7 @@ def _wait_for_port(port, timeout=15):
 
 
 def _cpu_only_env(port):
-    """Return an env dict that hides all GPUs and enables remote mode."""
+    """Return a *client* env dict that hides all GPUs and enables remote mode."""
     env = os.environ.copy()
     for key in [k for k in env if k.startswith("CUOPT_TLS_")]:
         env.pop(key)
@@ -435,10 +442,8 @@ def _start_grpc_server_fixture(port_offset):
         + port_offset
         + worker_id
     )
-    proc = subprocess.Popen(
+    proc = spawn_server(
         [server_bin, "--port", str(port), "--workers", "1"],
-        stdout=subprocess.DEVNULL,
-        stderr=subprocess.DEVNULL,
     )
     time.sleep(0.5)
     if proc.poll() is not None:
@@ -447,28 +452,23 @@ def _start_grpc_server_fixture(port_offset):
             "binary may be unable to load shared libraries in this environment"
         )
     if not _wait_for_port(port, timeout=15):
-        proc.kill()
-        proc.wait()
+        kill_server(proc)
         pytest.fail("cuopt_grpc_server failed to start within 15s")
 
     return proc, _cpu_only_env(port)
 
 
 def _stop_grpc_server(proc):
-    """Gracefully shut down a server process."""
-    proc.send_signal(signal.SIGTERM)
-    try:
-        proc.wait(timeout=5)
-    except subprocess.TimeoutExpired:
-        proc.kill()
-        proc.wait()
+    """Gracefully shut down a server process and its worker child."""
+    kill_server(proc)
 
 
 # ---------------------------------------------------------------------------
 # CPU-only Python tests  (subprocess required)
 # ---------------------------------------------------------------------------
 
 
+@pytest.mark.xdist_group(name="grpc_server")
 class TestCPUOnlyExecution:
     """Tests that run with CUDA_VISIBLE_DEVICES='' to simulate CPU-only hosts.
 
@@ -515,6 +515,7 @@ def test_warmstart_cpu_only(self, cpu_only_env_with_server):
 # ---------------------------------------------------------------------------
 
 
+@pytest.mark.xdist_group(name="grpc_server")
 class TestCuoptCliCPUOnly:
     """Test that cuopt_cli runs without CUDA in remote-execution mode.
 
@@ -711,6 +712,7 @@ def test_mip_solution_values(self):
 # ---------------------------------------------------------------------------
 
 
+@pytest.mark.xdist_group(name="grpc_server")
 class TestTLSExecution:
     """Test remote execution over a TLS-encrypted channel.
 
@@ -729,7 +731,7 @@ def tls_env_with_server(self, tmp_path_factory):
             pytest.skip("cuopt_grpc_server not found")
 
         port = int(os.environ.get("CUOPT_TEST_PORT_BASE", "18000")) + 850
-        proc = subprocess.Popen(
+        proc = spawn_server(
             [
                 server_bin,
                 "--port",
@@ -742,12 +744,9 @@ def tls_env_with_server(self, tmp_path_factory):
                 "--tls-key",
                 os.path.join(cert_dir, "server.key"),
             ],
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
         )
         if not _wait_for_port(port, timeout=15):
-            proc.kill()
-            proc.wait()
+            kill_server(proc)
             pytest.fail("TLS cuopt_grpc_server failed to start within 15s")
 
         env = _tls_env(port, cert_dir, mtls=False)
@@ -768,6 +767,7 @@ def test_lp_solve_tls(self, tls_env_with_server):
 # ---------------------------------------------------------------------------
 
 
+@pytest.mark.xdist_group(name="grpc_server")
 class TestMTLSExecution:
     """Test remote execution over an mTLS-encrypted channel.
 
@@ -787,7 +787,7 @@ def mtls_server_info(self, tmp_path_factory):
             pytest.skip("cuopt_grpc_server not found")
 
         port = int(os.environ.get("CUOPT_TEST_PORT_BASE", "18000")) + 900
-        proc = subprocess.Popen(
+        proc = spawn_server(
             [
                 server_bin,
                 "--port",
@@ -803,22 +803,14 @@ def mtls_server_info(self, tmp_path_factory):
                 os.path.join(cert_dir, "ca.crt"),
                 "--require-client-cert",
             ],
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
         )
         if not _wait_for_port(port, timeout=15):
-            proc.kill()
-            proc.wait()
+            kill_server(proc)
             pytest.fail("mTLS cuopt_grpc_server failed to start within 15s")
 
         yield {"port": port, "cert_dir": cert_dir}
 
-        proc.send_signal(signal.SIGTERM)
-        try:
-            proc.wait(timeout=5)
-        except subprocess.TimeoutExpired:
-            proc.kill()
-            proc.wait()
+        kill_server(proc)
 
     def test_lp_solve_mtls(self, mtls_server_info):
         """LP solve succeeds over an mTLS channel with valid client cert."""

python/cuopt/cuopt/tests/linear_programming/test_grpc_client.py

@@ -63,6 +63,7 @@ def _infeasible_lp_problem():
     return problem
 
 
+@pytest.mark.xdist_group(name="grpc_server")
 @pytest.mark.filterwarnings("ignore::DeprecationWarning")
 class TestGrpcClient:
     grpc_port_offset = GRPC_PORT_OFFSET_CLIENT

python/cuopt/cuopt/tests/linear_programming/test_incumbent_callbacks.py

@@ -86,14 +86,23 @@ def set_solution(
         settings.set_mip_callback(set_callback, user_data)
     solution = solver.Solve(data_model_obj, settings)
 
-    assert get_callback.n_callbacks > 0
+    termination = solution.get_termination_status()
+    assert termination in (
+        MILPTerminationStatus.FeasibleFound,
+        MILPTerminationStatus.Optimal,
+    )
+
+    # Incumbent callbacks only fire during branch-and-bound. If the problem is
+    # solved at the root node (presolve or integral LP relaxation), no
+    # callbacks are triggered — skip rather than fail in that case.
+    if get_callback.n_callbacks == 0:
+        pytest.skip(
+            f"No incumbent callbacks fired (termination={termination}); "
+            "problem was likely solved at the root node without branching"
+        )
+
     if include_set_callback:
         assert set_callback.n_callbacks > 0
-    assert (
-        solution.get_termination_status()
-        == MILPTerminationStatus.FeasibleFound
-        or MILPTerminationStatus.Optimal
-    )
 
     for sol in get_callback.solutions:
         utils.check_solution(