grpc implementation additional coderabbit review fixes

Author: tmckayus

GRPC_QUICK_START.md

@@ -194,9 +194,9 @@ With solver options:
 cuopt_cli model.mps --time-limit 30 --relaxation
 ```
 
-### C API
+### C++ API
 
-```c
+```cpp
 #include <cuopt/linear_programming/solve.hpp>
 #include <cuopt/linear_programming/cpu_optimization_problem.hpp>
 
@@ -212,11 +212,10 @@ gRPC server when they are set.
 
 | Symptom | Check |
 |---------|-------|
-| `CUOPT_REMOTE_HOST and/or CUOPT_REMOTE_PORT not set` | Both env vars must be set for remote execution. |
 | Connection refused | Verify the server is running and the host/port are correct. |
 | TLS handshake failure | Ensure `CUOPT_TLS_ENABLED=1` is set and certificate paths are correct. |
-| Timeout on large problems | Increase `CUOPT_CHUNK_SIZE` or `CUOPT_MAX_MESSAGE_BYTES`. |
 | `Cannot open TLS file: ...` | The path in the TLS env var does not exist or is not readable. |
+| Timeout on large problems | Increase the solver `time_limit` or the client `timeout_seconds`. |
 
 ## Further Reading
 

ci/utils/install_protobuf_grpc.sh

@@ -69,6 +69,15 @@ while [[ $# -gt 0 ]]; do
     esac
 done
 
+if [[ -z "$PREFIX" || "$PREFIX" == "/" ]]; then
+    echo "ERROR: Invalid PREFIX: '$PREFIX'" >&2
+    exit 1
+fi
+if [[ -z "$BUILD_DIR" || "$BUILD_DIR" == "/" ]]; then
+    echo "ERROR: Invalid BUILD_DIR: '$BUILD_DIR'" >&2
+    exit 1
+fi
+
 echo "=============================================="
 echo "Installing gRPC ${GRPC_VERSION} from source"
 echo "  Prefix: ${PREFIX}"

cpp/src/grpc/client/grpc_client.cpp

@@ -217,28 +217,20 @@ void grpc_client_t::start_log_streaming(const std::string& job_id)
 
 void grpc_client_t::stop_log_streaming()
 {
-  constexpr auto kLogJoinTimeout = std::chrono::seconds(5);
-
   stop_logs_.store(true);
-  // Cancel the in-flight streaming RPC from this thread.  reader->Read()
-  // blocks until the server sends a message, so the stop_logs_ flag alone
-  // is not enough — TryCancel makes Read() return false immediately.
+  // Cancel the in-flight streaming RPC so reader->Read() returns false
+  // immediately instead of blocking until the server sends a message.
   {
     std::lock_guard<std::mutex> lk(log_context_mutex_);
     if (active_log_context_) {
       static_cast<grpc::ClientContext*>(active_log_context_)->TryCancel();
     }
   }
-  if (log_thread_ && log_thread_->joinable()) {
-    auto future = std::async(std::launch::async, [this]() { log_thread_->join(); });
-    if (future.wait_for(kLogJoinTimeout) == std::future_status::timeout) {
-      GRPC_CLIENT_DEBUG_LOG(config_,
-                            "[grpc_client] WARNING: log streaming thread did not exit within "
-                              << kLogJoinTimeout.count() << "s; detaching");
-      log_thread_->detach();
-    }
-  }
-  log_thread_.reset();
+  // Move to local so we can join without racing against other callers.
+  // TryCancel above guarantees the thread will unblock promptly.
+  std::unique_ptr<std::thread> t;
+  std::swap(t, log_thread_);
+  if (t && t->joinable()) { t->join(); }
 }
 
 // =============================================================================
@@ -1082,6 +1074,11 @@ bool grpc_client_t::download_chunked_result(const std::string& job_id,
       int64_t elems_received = chunk_resp.elements_in_chunk();
       const auto& data       = chunk_resp.data();
 
+      if (elems_received < 0 || elems_received > elems_wanted ||
+          elems_received > total_elems - elem_offset) {
+        last_error_ = "GetResultChunk: invalid element count";
+        return false;
+      }
       if (static_cast<int64_t>(data.size()) != elems_received * elem_size) {
         last_error_ = "GetResultChunk: data size mismatch";
         return false;

cpp/src/grpc/grpc_problem_mapper.cpp

@@ -447,6 +447,7 @@ void map_chunked_arrays_to_problem(const cuopt::remote::ChunkedProblemHeader& he
       const char* nul = static_cast<const char*>(std::memchr(s, '\0', s_end - s));
       if (!nul) nul = s_end;
       names.emplace_back(s, nul);
+      if (nul == s_end) break;
       s = nul + 1;
     }
     return names;

cpp/src/grpc/server/grpc_job_management.cpp

@@ -49,8 +49,10 @@ bool send_incumbent_pipe(int fd, const std::vector<uint8_t>& data)
 
 bool recv_incumbent_pipe(int fd, std::vector<uint8_t>& data)
 {
+  static constexpr uint64_t kMaxIncumbentBytes = 256ULL * 1024 * 1024;
   uint64_t size;
   if (!read_from_pipe(fd, &size, sizeof(size))) return false;
+  if (size > kMaxIncumbentBytes) return false;
   data.resize(size);
   if (size > 0 && !read_from_pipe(fd, data.data(), size)) return false;
   return true;
@@ -88,6 +90,7 @@ std::pair<bool, std::string> submit_job_async(std::vector<uint8_t>&& request_dat
   job_queue[slot].worker_index.store(-1);
   job_queue[slot].data_sent.store(false);
   job_queue[slot].is_chunked = false;
+  job_queue[slot].worker_pid = 0;
 
   {
     std::lock_guard<std::mutex> lock(pending_data_mutex);
@@ -135,6 +138,7 @@ std::pair<bool, std::string> submit_chunked_job_async(PendingChunkedUpload&& chu
   job_queue[slot].worker_index.store(-1);
   job_queue[slot].data_sent.store(false);
   job_queue[slot].is_chunked = true;
+  job_queue[slot].worker_pid = 0;
 
   {
     std::lock_guard<std::mutex> lock(pending_data_mutex);

cpp/src/grpc/server/grpc_pipe_serialization.hpp

@@ -22,6 +22,9 @@
 // may silently cap this to /proc/sys/fs/pipe-max-size.
 static constexpr int kPipeBufferSize = 1024 * 1024;
 
+static constexpr uint64_t kMaxPipeArrayBytes  = 4ULL * 1024 * 1024 * 1024;
+static constexpr uint32_t kMaxPipeArrayFields = 10000;
+
 // Pipe I/O primitives defined in grpc_job_management.cpp.
 bool write_to_pipe(int fd, const void* data, size_t size);
 bool read_from_pipe(int fd, void* data, size_t size, int timeout_ms = 120000);
@@ -135,13 +138,15 @@ inline bool read_chunked_request_from_pipe(int fd,
 
   uint32_t num_arrays;
   if (!read_from_pipe(fd, &num_arrays, sizeof(num_arrays))) return false;
+  if (num_arrays > kMaxPipeArrayFields) return false;
 
   // Read each field's raw bytes directly into the output map, keyed by field_id.
   for (uint32_t i = 0; i < num_arrays; ++i) {
     int32_t field_id;
     uint64_t total_bytes;
     if (!read_from_pipe(fd, &field_id, sizeof(field_id))) return false;
     if (!read_from_pipe(fd, &total_bytes, sizeof(total_bytes))) return false;
+    if (total_bytes > kMaxPipeArrayBytes) return false;
     auto& dest = arrays_out[field_id];
     dest.resize(static_cast<size_t>(total_bytes));
     if (total_bytes > 0 && !read_from_pipe(fd, dest.data(), static_cast<size_t>(total_bytes)))
@@ -188,12 +193,14 @@ inline bool read_result_from_pipe(int fd,
 
   uint32_t num_arrays;
   if (!read_from_pipe(fd, &num_arrays, sizeof(num_arrays))) return false;
+  if (num_arrays > kMaxPipeArrayFields) return false;
 
   for (uint32_t i = 0; i < num_arrays; ++i) {
     int32_t field_id;
     uint64_t total_bytes;
     if (!read_from_pipe(fd, &field_id, sizeof(field_id))) return false;
     if (!read_from_pipe(fd, &total_bytes, sizeof(total_bytes))) return false;
+    if (total_bytes > kMaxPipeArrayBytes) return false;
     auto& dest = arrays_out[field_id];
     dest.resize(static_cast<size_t>(total_bytes));
     if (total_bytes > 0 && !read_from_pipe(fd, dest.data(), static_cast<size_t>(total_bytes)))

cpp/src/grpc/server/grpc_server_main.cpp

@@ -156,6 +156,7 @@ int main(int argc, char** argv)
     std::cerr << "[Server] Failed to mmap control: " << strerror(errno) << "\n";
     return 1;
   }
+  new (shm_ctrl) SharedMemoryControl{};
 
   for (size_t i = 0; i < MAX_JOBS; ++i) {
     new (&job_queue[i]) JobQueueEntry{};
@@ -222,8 +223,15 @@ int main(int argc, char** argv)
     creds = grpc::InsecureServerCredentials();
   }
 
+  signal(SIGPIPE, SIG_IGN);
   spawn_workers();
 
+  if (worker_pids.empty()) {
+    std::cerr << "[Server] No workers started; exiting\n";
+    cleanup_shared_memory();
+    return 1;
+  }
+
   std::thread result_thread(result_retrieval_thread);
   std::thread incumbent_thread(incumbent_retrieval_thread);
   std::thread monitor_thread(worker_monitor_thread);

cpp/src/grpc/server/grpc_worker_infra.cpp

@@ -147,6 +147,7 @@ void spawn_workers()
 void wait_for_workers()
 {
   for (pid_t pid : worker_pids) {
+    if (pid <= 0) continue;
     int status;
     while (waitpid(pid, &status, 0) < 0 && errno == EINTR) {}
   }

cpp/tests/linear_programming/grpc/grpc_test_log_capture.hpp

@@ -370,7 +370,7 @@ class GrpcTestLogCapture {
   std::vector<LogEntry> client_logs_;
   std::string server_log_path_;
   std::streampos server_log_start_pos_ = 0;  // Position in server log file when test started
-  bool test_start_marked_              = false;
+  std::atomic<bool> test_start_marked_{false};
 };
 
 }  // namespace cuopt::linear_programming::testing

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

@@ -77,6 +77,8 @@ def _wait_for_port(port, timeout=15):
 def _cpu_only_env(port):
     """Return an 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)
     env["CUDA_VISIBLE_DEVICES"] = ""
     env["CUOPT_REMOTE_HOST"] = "localhost"
     env["CUOPT_REMOTE_PORT"] = str(port)
@@ -110,14 +112,18 @@ def _run(cmd):
     server_key = os.path.join(cert_dir, "server.key")
     server_csr = os.path.join(cert_dir, "server.csr")
     server_crt = os.path.join(cert_dir, "server.crt")
+    server_ext = os.path.join(cert_dir, "server.ext")
     if not _run(
         f"openssl req -newkey rsa:2048 -keyout {server_key} -out {server_csr} "
         f"-nodes -subj '/CN=localhost' 2>/dev/null"
     ):
         return False
+    with open(server_ext, "w") as f:
+        f.write("subjectAltName=DNS:localhost,IP:127.0.0.1\n")
     if not _run(
         f"openssl x509 -req -in {server_csr} -CA {ca_crt} -CAkey {ca_key} "
-        f"-CAcreateserial -out {server_crt} -days 1 2>/dev/null"
+        f"-CAcreateserial -out {server_crt} -days 1 "
+        f"-extfile {server_ext} 2>/dev/null"
     ):
         return False