Merge branch 'main' into jwyman/upper-bound

Author: whoisj

README.md

@@ -213,6 +213,7 @@ under-development version).
 
 ```bash
 $ git checkout main
+$ mkdir -p build && cd build
 ```
 
 If building the Java client you must first install Maven and a JDK
@@ -223,60 +224,23 @@ the `default-jdk` package:
 $ apt-get install default-jdk maven
 ```
 
-Building on Windows vs. non-Windows requires different invocations
-because Triton on Windows does not yet support all the build options.
-
-#### Non-Windows
-
 Use *cmake* to configure the build. You should adjust the flags depending on
 the components of Triton Client you are working and would like to build.
 
-If you are building on a release branch (or on a development branch
-that is based off of a release branch), then you must also use
-additional cmake arguments to point to that release branch for repos
-that the client build depends on. For example, if you are building the
-r21.10 client branch then you need to use the following additional
-cmake flags:
-
-```
--DTRITON_COMMON_REPO_TAG=r21.10
--DTRITON_THIRD_PARTY_REPO_TAG=r21.10
--DTRITON_CORE_REPO_TAG=r21.10
-```
-
-Then use *make* to build the clients and examples.
-
-```
-$ make cc-clients python-clients java-clients
-```
-
-When the build completes the libraries and examples can be found in
-the install directory.
-
-#### Windows
-
-To build the clients you must install an appropriate C++ compiler and
-other dependencies required for the build. The easiest way to do this
-is to create the [Windows min Docker
-image](https://github.com/triton-inference-server/server/blob/main/docs/customization_guide/build.md#windows-10-min-image)
-and the perform the build within a container launched from that image.
-
-```
-> docker run  -it --rm win10-py3-min powershell
-```
-
-It is not necessary to use Docker or the win10-py3-min container for
-the build, but if you do not you must install the appropriate
-dependencies onto your host system.
-
-Next use *cmake* to configure the build. If you are not building
-within the win10-py3-min container then you will likely need to adjust
-the CMAKE_TOOLCHAIN_FILE location in the following command.
-
 ```
-$ mkdir build
-$ cd build
-$ cmake -DVCPKG_TARGET_TRIPLET=x64-windows -DCMAKE_TOOLCHAIN_FILE='/vcpkg/scripts/buildsystems/vcpkg.cmake' -DCMAKE_INSTALL_PREFIX=install -DTRITON_ENABLE_CC_GRPC=ON -DTRITON_ENABLE_PYTHON_GRPC=ON -DTRITON_ENABLE_GPU=OFF -DTRITON_ENABLE_EXAMPLES=ON -DTRITON_ENABLE_TESTS=ON ..
+cmake \
+  -DCMAKE_BUILD_TYPE=Release \
+  -DCMAKE_INSTALL_PREFIX=$(pwd)/install \
+  -DTRITON_ENABLE_CC_HTTP=ON \
+  -DTRITON_ENABLE_CC_GRPC=ON \
+  -DTRITON_ENABLE_PYTHON_HTTP=ON \
+  -DTRITON_ENABLE_PYTHON_GRPC=ON \
+  -DTRITON_ENABLE_JAVA_HTTP=ON \
+  -DTRITON_ENABLE_EXAMPLES=ON \
+  -DTRITON_ENABLE_TESTS=ON \
+  -DTRITON_ENABLE_GPU=ON \
+  -DTRITON_ENABLE_ZLIB=ON \
+  ..
 ```
 
 If you are building on a release branch (or on a development branch
@@ -292,11 +256,10 @@ cmake flags:
 -DTRITON_CORE_REPO_TAG=r21.10
 ```
 
-Then use msbuild.exe to build.
+Then use *make* to build the clients and examples.
 
 ```
-$ msbuild.exe cc-clients.vcxproj -p:Configuration=Release -clp:ErrorsOnly
-$ msbuild.exe python-clients.vcxproj -p:Configuration=Release -clp:ErrorsOnly
+$ make -j$(nproc)/ cc-clients python-clients java-clients
 ```
 
 When the build completes the libraries and examples can be found in
@@ -515,6 +478,7 @@ examples demonstrate how to infer with AsyncIO.
 
 
 ### Request Cancellation
+#### Python gRPC client
 
 Starting from r23.10, triton python gRPC client can issue cancellation
 to inflight requests. This can be done by calling `cancel()` on the
@@ -560,6 +524,61 @@ asynchronous iterator returned by `stream_infer()` API.
 See more details about these APIs in
 [grpc/aio/\__init__.py](src/python/library/tritonclient/grpc/aio/__init__.py).
 
+#### C++ gRPC client
+
+Starting from r26.05, triton python C++ client can issue cancellation to
+inflight requests. This can be done by passing an optional
+`CallContext** ctx_out` to `AsyncInfer()`; the returned `CallContext` exposes
+`Cancel()`.
+
+```cpp
+  tc::CallContext* ctx = nullptr;
+  client->AsyncInfer(
+      callback, options, inputs, outputs,
+      /*headers=*/{}, GRPC_COMPRESS_NONE, &ctx);
+  ctx->Cancel();
+  delete ctx;
+```
+
+For batch fan-out via `AsyncInferMulti()`, pass an optional
+`std::vector<CallContext*>* ctxs_out`; on return it is sized to `inputs.size()`
+with one `CallContext` per leaf request (or `nullptr` for any request that
+failed locally before the RPC started). Cancellation is per-request — siblings
+continue to completion, and the multi callback still fires exactly once.
+
+```cpp
+  std::vector<tc::CallContext*> ctxs;
+  client->AsyncInferMulti(
+      multi_callback, options, inputs, outputs,
+      /*headers=*/{}, GRPC_COMPRESS_NONE, &ctxs);
+  for (auto* ctx : ctxs) {
+    if (ctx != nullptr) {
+      ctx->Cancel();
+    }
+  }
+  for (auto* ctx : ctxs) {
+    delete ctx;
+  }
+```
+
+For streaming requests, pass `true` to `StopStream()`. Cancelling after the
+RPC has completed is a safe no-op. The caller owns the returned `CallContext`
+and must `delete` it. Call sites that omit `ctx_out` keep the previous,
+non-cancellable behavior.
+
+```cpp
+  client->StartStream(callback, ...);
+  for (...) {
+    client->AsyncStreamInfer(options, inputs, outputs);
+  }
+  client->StopStream(/*cancel_requests=*/true);
+```
+
+See more details about these APIs in
+[grpc_client.h](src/c++/library/grpc_client.h).
+
+#### Server-side behavior
+
 See [request_cancellation](https://github.com/triton-inference-server/server/blob/main/docs/user_guide/request_cancellation.md)
 in the server user-guide to learn about how this is handled on the
 server side.

src/c++/library/grpc_client.cc

@@ -1,4 +1,4 @@
-// Copyright 2020-2023, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// Copyright 2020-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 //
 // Redistribution and use in source and binary forms, with or without
 // modification, are permitted provided that the following conditions
@@ -34,6 +34,7 @@
 #include <future>
 #include <iostream>
 #include <mutex>
+#include <new>
 #include <sstream>
 #include <string>
 
@@ -172,16 +173,19 @@ SetTimeout(const uint64_t& client_timeout_ms, grpc::ClientContext* context)
 class GrpcInferRequest : public InferRequest {
  public:
   GrpcInferRequest(InferenceServerClient::OnCompleteFn callback = nullptr)
-      : InferRequest(callback), grpc_status_(),
+      : InferRequest(callback),
+        grpc_context_(std::make_shared<grpc::ClientContext>()), grpc_status_(),
         grpc_response_(std::make_shared<inference::ModelInferResponse>())
   {
   }
 
   friend InferenceServerGrpcClient;
 
  private:
-  // Variables for GRPC call
-  grpc::ClientContext grpc_context_;
+  // Variables for GRPC call. The ClientContext is held by shared_ptr so that
+  // a CallContext returned to the caller can co-own it and safely call
+  // TryCancel() at any point during (or after) the RPC's lifetime.
+  std::shared_ptr<grpc::ClientContext> grpc_context_;
   grpc::Status grpc_status_;
   std::shared_ptr<inference::ModelInferResponse> grpc_response_;
 };
@@ -447,6 +451,17 @@ InferResultGrpc::InferResultGrpc(
 
 //==============================================================================
 
+Error
+CallContext::Cancel()
+{
+  if (grpc_context_) {
+    grpc_context_->TryCancel();
+  }
+  return Error::Success;
+}
+
+//==============================================================================
+
 // Advanced users can generically pass any channel arguments
 // through the channel_args parameter, including KeepAlive options.
 // Channel arguments provided by the user are at the user's own risk and
@@ -1154,7 +1169,8 @@ InferenceServerGrpcClient::AsyncInfer(
     OnCompleteFn callback, const InferOptions& options,
     const std::vector<InferInput*>& inputs,
     const std::vector<const InferRequestedOutput*>& outputs,
-    const Headers& headers, grpc_compression_algorithm compression_algorithm)
+    const Headers& headers, grpc_compression_algorithm compression_algorithm,
+    CallContext** ctx_out)
 {
   if (callback == nullptr) {
     return Error(
@@ -1170,13 +1186,14 @@ InferenceServerGrpcClient::AsyncInfer(
   async_request->Timer().CaptureTimestamp(RequestTimers::Kind::REQUEST_START);
   async_request->Timer().CaptureTimestamp(RequestTimers::Kind::SEND_START);
   for (const auto& it : headers) {
-    async_request->grpc_context_.AddMetadata(it.first, it.second);
+    async_request->grpc_context_->AddMetadata(it.first, it.second);
   }
 
   if (options.client_timeout_ != 0) {
-    SetTimeout(options.client_timeout_, &(async_request->grpc_context_));
+    SetTimeout(options.client_timeout_, async_request->grpc_context_.get());
   }
-  async_request->grpc_context_.set_compression_algorithm(compression_algorithm);
+  async_request->grpc_context_->set_compression_algorithm(
+      compression_algorithm);
 
   Error err = PreRunProcessing(options, inputs, outputs);
   if (!err.IsOk()) {
@@ -1189,7 +1206,7 @@ InferenceServerGrpcClient::AsyncInfer(
   std::unique_ptr<
       grpc::ClientAsyncResponseReader<inference::ModelInferResponse>>
       rpc(stub_->PrepareAsyncModelInfer(
-          &async_request->grpc_context_, infer_request_,
+          async_request->grpc_context_.get(), infer_request_,
           &async_request_completion_queue_));
 
   rpc->StartCall();
@@ -1198,6 +1215,13 @@ InferenceServerGrpcClient::AsyncInfer(
       async_request->grpc_response_.get(), &async_request->grpc_status_,
       (void*)async_request);
 
+  // Hand the caller a cancellation handle co-owning the same ClientContext.
+  // Placed after StartCall/Finish so *ctx_out is only set on success; the
+  // earlier validation failure paths leave *ctx_out unchanged (nullptr).
+  if (ctx_out != nullptr) {
+    *ctx_out = new CallContext(async_request->grpc_context_);
+  }
+
   if (verbose_) {
     std::cout << "Sent request";
     if (options.request_id_.size() != 0) {
@@ -1255,7 +1279,8 @@ InferenceServerGrpcClient::AsyncInferMulti(
     OnMultiCompleteFn callback, const std::vector<InferOptions>& options,
     const std::vector<std::vector<InferInput*>>& inputs,
     const std::vector<std::vector<const InferRequestedOutput*>>& outputs,
-    const Headers& headers, grpc_compression_algorithm compression_algorithm)
+    const Headers& headers, grpc_compression_algorithm compression_algorithm,
+    std::vector<CallContext*>* ctxs_out)
 {
   // Sanity check
   if ((inputs.size() != options.size()) && (options.size() != 1)) {
@@ -1284,6 +1309,13 @@ InferenceServerGrpcClient::AsyncInferMulti(
       new std::atomic<size_t>(inputs.size()));
   std::shared_ptr<std::vector<InferResult*>> responses(
       new std::vector<InferResult*>(inputs.size()));
+
+  // Pre-size so AsyncInfer() can write slot i directly; slots stay
+  // nullptr for requests that fail before the RPC starts.
+  if (ctxs_out != nullptr) {
+    ctxs_out->assign(inputs.size(), nullptr);
+  }
+
   for (int64_t i = 0; i < (int64_t)inputs.size(); ++i) {
     const auto& request_options = options[std::min(max_option_idx, i)];
     const auto& request_output = (max_output_idx == -1)
@@ -1301,9 +1333,10 @@ InferenceServerGrpcClient::AsyncInferMulti(
       }
     };
 
+    CallContext** ctx = (ctxs_out != nullptr) ? &(*ctxs_out)[i] : nullptr;
     auto err = AsyncInfer(
         cb, request_options, inputs[i], request_output, headers,
-        compression_algorithm);
+        compression_algorithm, ctx);
     if (!err.IsOk()) {
       // Create response with error as other requests may be sent and their
       // responses may not be accessed outside the callback.
@@ -1336,6 +1369,12 @@ InferenceServerGrpcClient::StartStream(
         "Callback function must be provided along with StartStream() call.");
   }
 
+  // Reset grpc_context_ in place: a previous StopStream(cancel_requests=
+  // true) leaves it cancelled, and ClientContext deletes operator= so we
+  // can't just reassign it.
+  grpc_context_.~ClientContext();
+  new (&grpc_context_) grpc::ClientContext();
+
   stream_callback_ = callback;
   enable_stream_stats_ = enable_stats;
 
@@ -1360,12 +1399,18 @@ InferenceServerGrpcClient::StartStream(
 }
 
 Error
-InferenceServerGrpcClient::StopStream()
+InferenceServerGrpcClient::StopStream(bool cancel_requests)
 {
   if (stream_worker_.joinable()) {
-    grpc_stream_->WritesDone();
-    // The reader thread will drain the stream properly
+    if (cancel_requests) {
+      stream_cancel_requested_.store(true, std::memory_order_release);
+      grpc_context_.TryCancel();
+    } else {
+      grpc_stream_->WritesDone();
+    }
+    // The reader thread drains the stream and completes the RPC (Finish).
     stream_worker_.join();
+    stream_cancel_requested_.store(false, std::memory_order_release);
     if (verbose_) {
       std::cout << "Stopped stream..." << std::endl;
     }
@@ -1603,7 +1648,14 @@ InferenceServerGrpcClient::AsyncTransfer()
       InferResult* async_result;
       Error err;
       if (!async_request->grpc_status_.ok()) {
-        err = Error(async_request->grpc_status_.error_message());
+        // Map gRPC CANCELLED to the same message the streaming-cancel path
+        // emits, matching tritonclient.grpc._utils.get_cancelled_error().
+        if (async_request->grpc_status_.error_code() ==
+            grpc::StatusCode::CANCELLED) {
+          err = Error("Locally cancelled by application!");
+        } else {
+          err = Error(async_request->grpc_status_.error_message());
+        }
       }
       async_request->Timer().CaptureTimestamp(RequestTimers::Kind::RECV_START);
       InferResultGrpc::Create(
@@ -1669,6 +1721,16 @@ InferenceServerGrpcClient::AsyncStreamTransfer()
     stream_callback_(stream_result);
     response = std::make_shared<inference::ModelStreamInferResponse>();
   }
+  // After StopStream(cancel_requests=true), Read() drains without a status,
+  // so synthesize one final callback carrying the Python-parity message.
+  if (stream_cancel_requested_.load(std::memory_order_acquire) && !exiting_) {
+    InferResult* cancel_result = nullptr;
+    auto cancel_response =
+        std::make_shared<inference::ModelStreamInferResponse>();
+    cancel_response->set_error_message("Locally cancelled by application!");
+    InferResultGrpc::Create(&cancel_result, cancel_response);
+    stream_callback_(cancel_result);
+  }
   grpc_stream_->Finish();
 }