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/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 // ============================================================================= 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..323eceba73 100644 --- a/python/cuopt/cuopt/grpc/linear_programming/grpc_client.pyx +++ b/python/cuopt/cuopt/grpc/linear_programming/grpc_client.pyx @@ -191,7 +191,6 @@ def _forward_incumbent_to_settings(settings, index, objective, assignment, job_c 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 +220,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 +238,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 +257,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 +287,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 +468,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 +476,20 @@ 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` + Pass ``settings`` with :meth:`SolverSettings.set_mip_callback` registered :class:`GetSolutionCallback` instances (same as local solve). 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 callback is None and settings is None: - raise GrpcError("callback or settings is required") + if settings is None: + raise GrpcError("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 + return _forward_incumbent_to_settings( + settings, index, objective, assignment, job_complete + ) incumbent_client = self._spawn_client() thread = threading.Thread(