From cff782fb94dce2113929c7c8311c7bc23484a40a Mon Sep 17 00:00:00 2001 From: Trevor McKay Date: Sat, 1 Aug 2026 20:15:52 -0400 Subject: [PATCH 1/3] feat(grpc): add get_result for Python client; require settings for incumbents Add grpc_client_t::get_result alongside the existing get_lp_result / get_mip_result APIs (internals unchanged for now). Wire the Python async client result() path through get_result so LP vs MIP comes from the server, and require SolverSettings mip callbacks for start_incumbent_stream. --- cpp/include/cuopt/grpc/cython_grpc_client.hpp | 5 +- cpp/src/grpc/client/cython_grpc_client.cpp | 27 ++---- cpp/src/grpc/client/grpc_client.cpp | 54 +++++++++++ cpp/src/grpc/client/grpc_client.hpp | 22 +++++ .../grpc/linear_programming/grpc_client.pxd | 2 +- .../grpc/linear_programming/grpc_client.pyx | 97 ++++++++----------- 6 files changed, 128 insertions(+), 79 deletions(-) diff --git a/cpp/include/cuopt/grpc/cython_grpc_client.hpp b/cpp/include/cuopt/grpc/cython_grpc_client.hpp index 593014cedc..5407383bab 100644 --- a/cpp/include/cuopt/grpc/cython_grpc_client.hpp +++ b/cpp/include/cuopt/grpc/cython_grpc_client.hpp @@ -132,9 +132,10 @@ class grpc_python_client_t { bool delete_job(const std::string& job_id, std::string& error_out); /** - * @param is_mip When true, fetch a MIP result; otherwise LP. + * Fetch the solution for a completed job. LP vs MIP is determined from the + * server response via grpc_client_t::get_result(). */ - grpc_result_outcome_t result(const std::string& job_id, bool is_mip); + grpc_result_outcome_t result(const std::string& job_id); /** * @brief Block until the job completes, collecting all solver log lines. diff --git a/cpp/src/grpc/client/cython_grpc_client.cpp b/cpp/src/grpc/client/cython_grpc_client.cpp index 9ba11459b5..53e650efb4 100644 --- a/cpp/src/grpc/client/cython_grpc_client.cpp +++ b/cpp/src/grpc/client/cython_grpc_client.cpp @@ -204,7 +204,7 @@ bool grpc_python_client_t::delete_job(const std::string& job_id, std::string& er return true; } -grpc_result_outcome_t grpc_python_client_t::result(const std::string& job_id, bool is_mip) +grpc_result_outcome_t grpc_python_client_t::result(const std::string& job_id) { grpc_result_outcome_t out; @@ -222,26 +222,19 @@ grpc_result_outcome_t grpc_python_client_t::result(const std::string& job_id, bo return out; } - out.solution = std::make_unique(); + auto remote = impl_->client.get_result(job_id); + if (!remote.success) { + out.error_message = remote.error_message; + return out; + } - if (is_mip) { - auto remote = impl_->client.get_mip_result(job_id); - if (!remote.success) { - out.error_message = remote.error_message; - out.solution.reset(); - return out; - } + out.solution = std::make_unique(); + if (remote.is_mip) { out.solution->problem_type = cuopt::mathematical_optimization::problem_category_t::MIP; - out.solution->mip_ret = remote.solution->to_cpu_mip_ret_t(); + out.solution->mip_ret = remote.mip_solution->to_cpu_mip_ret_t(); } else { - auto remote = impl_->client.get_lp_result(job_id); - if (!remote.success) { - out.error_message = remote.error_message; - out.solution.reset(); - return out; - } out.solution->problem_type = cuopt::mathematical_optimization::problem_category_t::LP; - out.solution->lp_ret = remote.solution->to_cpu_linear_programming_ret_t(); + out.solution->lp_ret = remote.lp_solution->to_cpu_linear_programming_ret_t(); } out.success = true; diff --git a/cpp/src/grpc/client/grpc_client.cpp b/cpp/src/grpc/client/grpc_client.cpp index ac28a2f767..920b1bae53 100644 --- a/cpp/src/grpc/client/grpc_client.cpp +++ b/cpp/src/grpc/client/grpc_client.cpp @@ -761,6 +761,58 @@ remote_mip_result_t grpc_client_t::get_mip_result(const std::string& j return result; } +template +remote_result_t grpc_client_t::get_result(const std::string& job_id) +{ + remote_result_t result; + + if (!is_connected()) { + result.error_message = "Not connected to server"; + return result; + } + + downloaded_result_t dl; + if (!get_result_or_download(job_id, dl)) { + result.error_message = last_error_; + return result; + } + + const bool is_mip = dl.was_chunked ? (dl.chunked_header->problem_category() == cuopt::remote::MIP) + : dl.response->has_mip_solution(); + + if (is_mip) { + if (dl.was_chunked) { + result.mip_solution = std::make_unique>( + chunked_result_to_mip_solution(*dl.chunked_header, dl.chunked_arrays)); + } else { + if (!dl.response->has_mip_solution()) { + result.error_message = "GetResult succeeded but no MIP solution in response"; + return result; + } + result.mip_solution = std::make_unique>( + map_proto_to_mip_solution(dl.response->mip_solution())); + } + result.is_mip = true; + result.success = true; + return result; + } + + if (dl.was_chunked) { + result.lp_solution = std::make_unique>( + chunked_result_to_lp_solution(*dl.chunked_header, dl.chunked_arrays)); + } else { + if (!dl.response->has_lp_solution()) { + result.error_message = "GetResult succeeded but no LP solution in response"; + return result; + } + result.lp_solution = std::make_unique>( + map_proto_to_lp_solution(dl.response->lp_solution())); + } + result.is_mip = false; + result.success = true; + return result; +} + // ============================================================================= // Polling helper // ============================================================================= @@ -1258,6 +1310,7 @@ template submit_result_t grpc_client_t::submit_mip( template remote_lp_result_t grpc_client_t::get_lp_result(const std::string& job_id); template remote_mip_result_t grpc_client_t::get_mip_result( const std::string& job_id); +template remote_result_t grpc_client_t::get_result(const std::string& job_id); template bool grpc_client_t::upload_chunked_arrays( const cpu_optimization_problem_t& problem, const cuopt::remote::ChunkedProblemHeader& header, @@ -1283,6 +1336,7 @@ template remote_lp_result_t grpc_client_t::get_lp_result( const std::string& job_id); template remote_mip_result_t grpc_client_t::get_mip_result( const std::string& job_id); +template remote_result_t grpc_client_t::get_result(const std::string& job_id); template bool grpc_client_t::upload_chunked_arrays( const cpu_optimization_problem_t& problem, const cuopt::remote::ChunkedProblemHeader& header, diff --git a/cpp/src/grpc/client/grpc_client.hpp b/cpp/src/grpc/client/grpc_client.hpp index 6920d61564..ebd5bd0dda 100644 --- a/cpp/src/grpc/client/grpc_client.hpp +++ b/cpp/src/grpc/client/grpc_client.hpp @@ -189,6 +189,18 @@ struct remote_mip_result_t { std::unique_ptr> solution; }; +/** + * @brief Result of get_result(): LP vs MIP is taken from the server response. + */ +template +struct remote_result_t { + bool success = false; + std::string error_message; + bool is_mip = false; + std::unique_ptr> lp_solution; + std::unique_ptr> mip_solution; +}; + /** * @brief gRPC client for remote cuOpt solving * @@ -341,6 +353,16 @@ class grpc_client_t { template remote_mip_result_t get_mip_result(const std::string& job_id); + /** + * @brief Get result for a completed job; LP vs MIP comes from the server. + * + * Used by the Python async gRPC client today. Existing internal call sites + * still use get_lp_result / get_mip_result; they can migrate to get_result + * in a later change. + */ + template + remote_result_t get_result(const std::string& job_id); + /** * @brief Cancel a running job * @param job_id The job ID to cancel diff --git a/python/cuopt/cuopt/grpc/linear_programming/grpc_client.pxd b/python/cuopt/cuopt/grpc/linear_programming/grpc_client.pxd index b474d9fb1f..8677266f13 100644 --- a/python/cuopt/cuopt/grpc/linear_programming/grpc_client.pxd +++ b/python/cuopt/cuopt/grpc/linear_programming/grpc_client.pxd @@ -89,7 +89,7 @@ cdef extern from "cuopt/grpc/cython_grpc_client.hpp" namespace "cuopt::cython": grpc_status_result_t wait(const string& job_id, int timeout_seconds) except + bint cancel(const string& job_id, string& error_out) except + bint delete_job(const string& job_id, string& error_out) except + - grpc_result_outcome_t result(const string& job_id, bint is_mip) except + + grpc_result_outcome_t result(const string& job_id) except + grpc_logs_result_t fetch_logs(const string& job_id, long long from_byte) except + bint stream_logs( const string& job_id, diff --git a/python/cuopt/cuopt/grpc/linear_programming/grpc_client.pyx b/python/cuopt/cuopt/grpc/linear_programming/grpc_client.pyx index f0d11585e3..acff991bd3 100644 --- a/python/cuopt/cuopt/grpc/linear_programming/grpc_client.pyx +++ b/python/cuopt/cuopt/grpc/linear_programming/grpc_client.pyx @@ -164,18 +164,11 @@ class _LogStreamHandler: raise -def _call_incumbent_callback(callback, index, objective, assignment, job_complete): - try: - return callback(index, objective, assignment, job_complete) - except TypeError: - return callback(index, objective, assignment) - - def _forward_incumbent_to_settings(settings, index, objective, assignment, job_complete): from cuopt.linear_programming.internals import GetSolutionCallback if job_complete: - return True + return for mip_callback in settings.get_mip_callbacks(): if mip_callback is None: continue @@ -186,12 +179,10 @@ def _forward_incumbent_to_settings(settings, index, objective, assignment, job_c mip_callback.get_solution( solution, cost, bound, mip_callback.user_data ) - return True cdef class Client: cdef unique_ptr[grpc_python_client_t] _client - cdef dict _job_is_mip cdef dict _log_threads cdef dict _log_thread_errors cdef dict _log_stream_state @@ -221,7 +212,6 @@ cdef class Client: options = _connect_options_from_tls(tls) self._client.reset(new grpc_python_client_t(host_cpp, port, options)) - self._job_is_mip = {} self._log_threads = {} self._log_thread_errors = {} self._log_stream_state = {} @@ -240,7 +230,6 @@ cdef class Client: def submit(self, problem, SolverSettings settings not None): cdef DataModel data_model cdef grpc_submit_result_t submit_result - cdef string job_id cdef bint mip data_model = self._as_data_model(problem) @@ -260,9 +249,7 @@ cdef class Client: ) if not submit_result.success: raise GrpcError(submit_result.error_message.decode("utf-8")) - job_id = submit_result.job_id - self._job_is_mip[job_id.decode("utf-8")] = bool(submit_result.is_mip) - return job_id.decode("utf-8") + return submit_result.job_id.decode("utf-8") def status(self, str job_id): cdef grpc_status_result_t status_result = self._client.get().status( @@ -292,19 +279,19 @@ cdef class Client: cdef string error_out if not self._client.get().delete_job(job_id.encode("utf-8"), error_out): raise GrpcError(error_out.decode("utf-8")) - self._job_is_mip.pop(job_id, None) - def result(self, str job_id, variable_names=None, is_mip=None): + def result(self, str job_id, variable_names=None): + """ + Fetch the solution for a completed job, or ``None`` if not ready. + + LP vs MIP is determined from the server response (via + ``grpc_client_t::get_result``). Pass ``variable_names`` (column order) + to key ``solution.get_vars()`` by name. + """ cdef grpc_result_outcome_t outcome - cdef bint fetch_mip cdef unique_ptr[solver_ret_t] sol_ret - if is_mip is not None: - fetch_mip = bool(is_mip) - else: - fetch_mip = self._job_is_mip.get(job_id, False) - - outcome = self._client.get().result(job_id.encode("utf-8"), fetch_mip) + outcome = self._client.get().result(job_id.encode("utf-8")) if outcome.not_ready: return None if not outcome.success: @@ -473,8 +460,7 @@ cdef class Client: def start_incumbent_stream( self, str job_id, - callback=None, - settings=None, + settings, from_index=0, poll_interval_ms=1000, ): @@ -482,31 +468,29 @@ cdef class Client: Poll for MIP incumbent solutions on a background thread until the job completes. - ``callback`` is invoked as ``callback(index, objective, assignment, - job_complete)``. Return ``False`` to cancel the job. ``assignment`` is - a list of variable values. - - Alternatively pass ``settings`` with :meth:`SolverSettings.set_mip_callback` - registered :class:`GetSolutionCallback` instances (same as local solve). + Pass the same ``settings`` used for :meth:`submit`, with at least one + ``GetSolutionCallback`` registered through + :meth:`~cuopt.linear_programming.solver_settings.SolverSettings.set_mip_callback` + (same as a local solve). The server collects and sends incumbents only + when ``submit`` sees at least one such callback on ``settings``. - Call :meth:`join_incumbent_stream` before :meth:`delete`. + Call :meth:`join_incumbent_stream` before :meth:`delete`. To cancel + early, call :meth:`cancel` from inside ``get_solution``. """ if job_id in self._incumbent_threads: raise GrpcError(f"incumbent stream already running for job {job_id}") - if callback is None and settings is None: - raise GrpcError("callback or settings is required") - - def combined(index, objective, assignment, job_complete): - if settings is not None: - if _forward_incumbent_to_settings( - settings, index, objective, assignment, job_complete - ) is False: - return False - if callback is not None: - return _call_incumbent_callback( - callback, index, objective, assignment, job_complete - ) - return True + if settings is None: + raise GrpcError("settings is required") + if not any(cb is not None for cb in settings.get_mip_callbacks()): + raise GrpcError( + "settings must have at least one mip callback " + "(SolverSettings.set_mip_callback)" + ) + + def deliver(index, objective, assignment, job_complete): + _forward_incumbent_to_settings( + settings, index, objective, assignment, job_complete + ) incumbent_client = self._spawn_client() thread = threading.Thread( @@ -514,7 +498,7 @@ cdef class Client: args=( incumbent_client, job_id, - combined, + deliver, from_index, poll_interval_ms, ), @@ -525,7 +509,7 @@ cdef class Client: return thread def join_incumbent_stream(self, str job_id, timeout=None): - """Wait for a background incumbent poll started by :meth:`start_incumbent_stream`.""" + """Wait for the background incumbent-stream thread started by :meth:`start_incumbent_stream`.""" thread = self._incumbent_threads.pop(job_id, None) if thread is not None: thread.join(timeout) @@ -537,24 +521,23 @@ cdef class Client: self, incumbent_client, str job_id, - callback, + deliver, from_index, poll_interval_ms, ): try: incumbent_client._poll_incumbents( - job_id, callback, from_index, poll_interval_ms + job_id, deliver, from_index, poll_interval_ms ) except Exception as exc: self._incumbent_thread_errors[job_id] = exc def _poll_incumbents( - self, str job_id, callback, from_index=0, poll_interval_ms=1000 + self, str job_id, deliver, from_index=0, poll_interval_ms=1000 ): cdef grpc_incumbents_result_t outcome cdef int64_t next_index = from_index cdef bint job_complete = False - cdef double objective cdef list assignment cdef size_t i poll_seconds = max(poll_interval_ms, 1) / 1000.0 @@ -573,16 +556,12 @@ cdef class Client: assignment = [] for i in range(entry.assignment.size()): assignment.append(entry.assignment[i]) - if _call_incumbent_callback( - callback, entry.index, entry.objective, assignment, False - ) is False: - self.cancel(job_id) - return + deliver(entry.index, entry.objective, assignment, False) next_index = outcome.next_index job_complete = outcome.job_complete if job_complete: - _call_incumbent_callback(callback, 0, 0.0, [], True) + deliver(0, 0.0, [], True) return time.sleep(poll_seconds) From dc593dc261aad68270c001cea9b67a181e34a4c0 Mon Sep 17 00:00:00 2001 From: Trevor McKay Date: Sat, 1 Aug 2026 20:35:49 -0400 Subject: [PATCH 2/3] test(grpc): cover get_result for LP/MIP unary and chunked paths Add mock client tests so the unified get_result API is exercised for both problem types and both download modes. --- .../grpc/grpc_client_test.cpp | 212 ++++++++++++++++++ 1 file changed, 212 insertions(+) diff --git a/cpp/tests/linear_programming/grpc/grpc_client_test.cpp b/cpp/tests/linear_programming/grpc/grpc_client_test.cpp index d97aac0971..b892e5d7ae 100644 --- a/cpp/tests/linear_programming/grpc/grpc_client_test.cpp +++ b/cpp/tests/linear_programming/grpc/grpc_client_test.cpp @@ -907,6 +907,218 @@ TEST_F(GrpcClientTest, ChunkedDownload_StartFails) EXPECT_TRUE(lp_result.error_message.find("StartChunkedDownload") != std::string::npos); } +// ============================================================================= +// get_result (unified LP/MIP) Tests (Mock) +// ============================================================================= + +TEST_F(GrpcClientTest, GetResultUnified_UnaryLP) +{ + EXPECT_CALL(*mock_stub_, CheckStatus(_, _, _)) + .WillOnce([](grpc::ClientContext*, + const cuopt::remote::StatusRequest&, + cuopt::remote::StatusResponse* resp) { + resp->set_job_status(cuopt::remote::COMPLETED); + resp->set_result_size_bytes(64); + resp->set_max_message_bytes(256 * 1024 * 1024); + return grpc::Status::OK; + }); + + EXPECT_CALL(*mock_stub_, GetResult(_, _, _)) + .WillOnce([](grpc::ClientContext*, + const cuopt::remote::GetResultRequest& req, + cuopt::remote::ResultResponse* resp) { + EXPECT_EQ(req.job_id(), "unified-lp-unary"); + cuopt::remote::LPSolution solution; + solution.add_primal_solution(1.5); + solution.add_primal_solution(2.5); + solution.set_primal_objective(-464.753); + solution.set_lp_termination_status(cuopt::remote::PDLP_OPTIMAL); + resp->mutable_lp_solution()->CopyFrom(solution); + resp->set_status(cuopt::remote::SUCCESS); + return grpc::Status::OK; + }); + + auto result = client_->get_result("unified-lp-unary"); + + EXPECT_TRUE(result.success) << result.error_message; + EXPECT_FALSE(result.is_mip); + ASSERT_NE(result.lp_solution, nullptr); + EXPECT_EQ(result.mip_solution, nullptr); + EXPECT_NEAR(result.lp_solution->get_objective_value(), -464.753, 0.01); +} + +TEST_F(GrpcClientTest, GetResultUnified_UnaryMIP) +{ + EXPECT_CALL(*mock_stub_, CheckStatus(_, _, _)) + .WillOnce([](grpc::ClientContext*, + const cuopt::remote::StatusRequest&, + cuopt::remote::StatusResponse* resp) { + resp->set_job_status(cuopt::remote::COMPLETED); + resp->set_result_size_bytes(64); + resp->set_max_message_bytes(256 * 1024 * 1024); + return grpc::Status::OK; + }); + + EXPECT_CALL(*mock_stub_, GetResult(_, _, _)) + .WillOnce([](grpc::ClientContext*, + const cuopt::remote::GetResultRequest& req, + cuopt::remote::ResultResponse* resp) { + EXPECT_EQ(req.job_id(), "unified-mip-unary"); + cuopt::remote::MIPSolution solution; + solution.add_mip_solution(1.0); + solution.add_mip_solution(0.0); + solution.set_mip_objective(42.0); + solution.set_mip_termination_status(cuopt::remote::MIP_OPTIMAL); + resp->mutable_mip_solution()->CopyFrom(solution); + resp->set_status(cuopt::remote::SUCCESS); + return grpc::Status::OK; + }); + + auto result = client_->get_result("unified-mip-unary"); + + EXPECT_TRUE(result.success) << result.error_message; + EXPECT_TRUE(result.is_mip); + ASSERT_NE(result.mip_solution, nullptr); + EXPECT_EQ(result.lp_solution, nullptr); + EXPECT_DOUBLE_EQ(result.mip_solution->get_objective_value(), 42.0); +} + +TEST_F(GrpcClientTest, GetResultUnified_ChunkedLP_FallbackOnResourceExhausted) +{ + EXPECT_CALL(*mock_stub_, CheckStatus(_, _, _)) + .WillOnce([](grpc::ClientContext*, + const cuopt::remote::StatusRequest&, + cuopt::remote::StatusResponse* resp) { + resp->set_job_status(cuopt::remote::COMPLETED); + resp->set_result_size_bytes(500); + resp->set_max_message_bytes(256 * 1024 * 1024); + return grpc::Status::OK; + }); + + EXPECT_CALL(*mock_stub_, GetResult(_, _, _)) + .WillOnce([](grpc::ClientContext*, + const cuopt::remote::GetResultRequest&, + cuopt::remote::ResultResponse*) { + return grpc::Status(grpc::StatusCode::RESOURCE_EXHAUSTED, "Too large"); + }); + + EXPECT_CALL(*mock_stub_, StartChunkedDownload(_, _, _)) + .WillOnce([](grpc::ClientContext*, + const cuopt::remote::StartChunkedDownloadRequest&, + cuopt::remote::StartChunkedDownloadResponse* resp) { + resp->set_download_id("dl-unified-lp"); + auto* h = resp->mutable_header(); + h->set_problem_category(cuopt::remote::LP); + h->set_lp_termination_status(cuopt::remote::PDLP_OPTIMAL); + h->set_primal_objective(-464.753); + auto* arr = h->add_arrays(); + arr->set_field_id(cuopt::remote::RESULT_PRIMAL_SOLUTION); + arr->set_total_elements(2); + arr->set_element_size_bytes(8); + resp->set_max_message_bytes(4 * 1024 * 1024); + return grpc::Status::OK; + }); + + EXPECT_CALL(*mock_stub_, GetResultChunk(_, _, _)) + .WillOnce([](grpc::ClientContext*, + const cuopt::remote::GetResultChunkRequest& req, + cuopt::remote::GetResultChunkResponse* resp) { + EXPECT_EQ(req.download_id(), "dl-unified-lp"); + EXPECT_EQ(req.field_id(), cuopt::remote::RESULT_PRIMAL_SOLUTION); + resp->set_download_id("dl-unified-lp"); + resp->set_field_id(req.field_id()); + resp->set_element_offset(0); + resp->set_elements_in_chunk(2); + double vals[2] = {1.5, 2.5}; + resp->set_data(reinterpret_cast(vals), sizeof(vals)); + return grpc::Status::OK; + }); + + EXPECT_CALL(*mock_stub_, FinishChunkedDownload(_, _, _)) + .WillOnce([](grpc::ClientContext*, + const cuopt::remote::FinishChunkedDownloadRequest& req, + cuopt::remote::FinishChunkedDownloadResponse* resp) { + resp->set_download_id(req.download_id()); + return grpc::Status::OK; + }); + + auto result = client_->get_result("unified-lp-chunked"); + + EXPECT_TRUE(result.success) << result.error_message; + EXPECT_FALSE(result.is_mip); + ASSERT_NE(result.lp_solution, nullptr); + EXPECT_EQ(result.mip_solution, nullptr); + EXPECT_NEAR(result.lp_solution->get_objective_value(), -464.753, 0.01); +} + +TEST_F(GrpcClientTest, GetResultUnified_ChunkedMIP_FallbackOnResourceExhausted) +{ + EXPECT_CALL(*mock_stub_, CheckStatus(_, _, _)) + .WillOnce([](grpc::ClientContext*, + const cuopt::remote::StatusRequest&, + cuopt::remote::StatusResponse* resp) { + resp->set_job_status(cuopt::remote::COMPLETED); + resp->set_result_size_bytes(500); + resp->set_max_message_bytes(256 * 1024 * 1024); + return grpc::Status::OK; + }); + + EXPECT_CALL(*mock_stub_, GetResult(_, _, _)) + .WillOnce([](grpc::ClientContext*, + const cuopt::remote::GetResultRequest&, + cuopt::remote::ResultResponse*) { + return grpc::Status(grpc::StatusCode::RESOURCE_EXHAUSTED, "Too large"); + }); + + EXPECT_CALL(*mock_stub_, StartChunkedDownload(_, _, _)) + .WillOnce([](grpc::ClientContext*, + const cuopt::remote::StartChunkedDownloadRequest&, + cuopt::remote::StartChunkedDownloadResponse* resp) { + resp->set_download_id("dl-unified-mip"); + auto* h = resp->mutable_header(); + h->set_problem_category(cuopt::remote::MIP); + h->set_mip_termination_status(cuopt::remote::MIP_OPTIMAL); + h->set_mip_objective(42.0); + auto* arr = h->add_arrays(); + arr->set_field_id(cuopt::remote::RESULT_MIP_SOLUTION); + arr->set_total_elements(2); + arr->set_element_size_bytes(8); + resp->set_max_message_bytes(4 * 1024 * 1024); + return grpc::Status::OK; + }); + + EXPECT_CALL(*mock_stub_, GetResultChunk(_, _, _)) + .WillOnce([](grpc::ClientContext*, + const cuopt::remote::GetResultChunkRequest& req, + cuopt::remote::GetResultChunkResponse* resp) { + EXPECT_EQ(req.download_id(), "dl-unified-mip"); + EXPECT_EQ(req.field_id(), cuopt::remote::RESULT_MIP_SOLUTION); + resp->set_download_id("dl-unified-mip"); + resp->set_field_id(req.field_id()); + resp->set_element_offset(0); + resp->set_elements_in_chunk(2); + double vals[2] = {1.0, 0.0}; + resp->set_data(reinterpret_cast(vals), sizeof(vals)); + return grpc::Status::OK; + }); + + EXPECT_CALL(*mock_stub_, FinishChunkedDownload(_, _, _)) + .WillOnce([](grpc::ClientContext*, + const cuopt::remote::FinishChunkedDownloadRequest& req, + cuopt::remote::FinishChunkedDownloadResponse* resp) { + resp->set_download_id(req.download_id()); + return grpc::Status::OK; + }); + + auto result = client_->get_result("unified-mip-chunked"); + + EXPECT_TRUE(result.success) << result.error_message; + EXPECT_TRUE(result.is_mip); + ASSERT_NE(result.mip_solution, nullptr); + EXPECT_EQ(result.lp_solution, nullptr); + EXPECT_DOUBLE_EQ(result.mip_solution->get_objective_value(), 42.0); +} + // ============================================================================= // Helper: Build minimal test problems // ============================================================================= From da6dd3e41f839d09557b2f389cf4feb986a1ada2 Mon Sep 17 00:00:00 2001 From: Trevor McKay Date: Sat, 1 Aug 2026 21:30:17 -0400 Subject: [PATCH 3/3] restore internal plubming for client callback= on incumbents This limits the changes. Internals will be cleaned up later --- .../grpc/linear_programming/grpc_client.pyx | 48 ++++++++++--------- 1 file changed, 26 insertions(+), 22 deletions(-) diff --git a/python/cuopt/cuopt/grpc/linear_programming/grpc_client.pyx b/python/cuopt/cuopt/grpc/linear_programming/grpc_client.pyx index acff991bd3..323eceba73 100644 --- a/python/cuopt/cuopt/grpc/linear_programming/grpc_client.pyx +++ b/python/cuopt/cuopt/grpc/linear_programming/grpc_client.pyx @@ -164,11 +164,18 @@ class _LogStreamHandler: raise +def _call_incumbent_callback(callback, index, objective, assignment, job_complete): + try: + return callback(index, objective, assignment, job_complete) + except TypeError: + return callback(index, objective, assignment) + + def _forward_incumbent_to_settings(settings, index, objective, assignment, job_complete): from cuopt.linear_programming.internals import GetSolutionCallback if job_complete: - return + return True for mip_callback in settings.get_mip_callbacks(): if mip_callback is None: continue @@ -179,6 +186,7 @@ def _forward_incumbent_to_settings(settings, index, objective, assignment, job_c mip_callback.get_solution( solution, cost, bound, mip_callback.user_data ) + return True cdef class Client: @@ -468,27 +476,18 @@ cdef class Client: Poll for MIP incumbent solutions on a background thread until the job completes. - Pass the same ``settings`` used for :meth:`submit`, with at least one - ``GetSolutionCallback`` registered through - :meth:`~cuopt.linear_programming.solver_settings.SolverSettings.set_mip_callback` - (same as a local solve). The server collects and sends incumbents only - when ``submit`` sees at least one such callback on ``settings``. + Pass ``settings`` with :meth:`SolverSettings.set_mip_callback` + registered :class:`GetSolutionCallback` instances (same as local solve). - Call :meth:`join_incumbent_stream` before :meth:`delete`. To cancel - early, call :meth:`cancel` from inside ``get_solution``. + Call :meth:`join_incumbent_stream` before :meth:`delete`. """ if job_id in self._incumbent_threads: raise GrpcError(f"incumbent stream already running for job {job_id}") if settings is None: raise GrpcError("settings is required") - if not any(cb is not None for cb in settings.get_mip_callbacks()): - raise GrpcError( - "settings must have at least one mip callback " - "(SolverSettings.set_mip_callback)" - ) - def deliver(index, objective, assignment, job_complete): - _forward_incumbent_to_settings( + def combined(index, objective, assignment, job_complete): + return _forward_incumbent_to_settings( settings, index, objective, assignment, job_complete ) @@ -498,7 +497,7 @@ cdef class Client: args=( incumbent_client, job_id, - deliver, + combined, from_index, poll_interval_ms, ), @@ -509,7 +508,7 @@ cdef class Client: return thread def join_incumbent_stream(self, str job_id, timeout=None): - """Wait for the background incumbent-stream thread started by :meth:`start_incumbent_stream`.""" + """Wait for a background incumbent poll started by :meth:`start_incumbent_stream`.""" thread = self._incumbent_threads.pop(job_id, None) if thread is not None: thread.join(timeout) @@ -521,23 +520,24 @@ cdef class Client: self, incumbent_client, str job_id, - deliver, + callback, from_index, poll_interval_ms, ): try: incumbent_client._poll_incumbents( - job_id, deliver, from_index, poll_interval_ms + job_id, callback, from_index, poll_interval_ms ) except Exception as exc: self._incumbent_thread_errors[job_id] = exc def _poll_incumbents( - self, str job_id, deliver, from_index=0, poll_interval_ms=1000 + self, str job_id, callback, from_index=0, poll_interval_ms=1000 ): cdef grpc_incumbents_result_t outcome cdef int64_t next_index = from_index cdef bint job_complete = False + cdef double objective cdef list assignment cdef size_t i poll_seconds = max(poll_interval_ms, 1) / 1000.0 @@ -556,12 +556,16 @@ cdef class Client: assignment = [] for i in range(entry.assignment.size()): assignment.append(entry.assignment[i]) - deliver(entry.index, entry.objective, assignment, False) + if _call_incumbent_callback( + callback, entry.index, entry.objective, assignment, False + ) is False: + self.cancel(job_id) + return next_index = outcome.next_index job_complete = outcome.job_complete if job_complete: - deliver(0, 0.0, [], True) + _call_incumbent_callback(callback, 0, 0.0, [], True) return time.sleep(poll_seconds)