diff --git a/qa/L0_backend_python/model_readiness/test.sh b/qa/L0_backend_python/model_readiness/test.sh index cc87aeacd8..b25cfab6ec 100755 --- a/qa/L0_backend_python/model_readiness/test.sh +++ b/qa/L0_backend_python/model_readiness/test.sh @@ -1,5 +1,5 @@ #!/bin/bash -# Copyright 2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# Copyright 2025-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 @@ -47,7 +47,7 @@ cp ../../python_models/$MODEL_NAME/config.pbtxt ./models/$MODEL_NAME/config.pbtx for SIGNAL in 11 9; do echo -e "\n***\n*** Testing model_readiness with Signal $SIGNAL\n***" SERVER_LOG="./model_readiness_signal_${SIGNAL}_server.log" - CLIENT_LOG="./model_readiness_${SIGNAL}_client.log" + CLIENT_LOG="./model_readiness_signal_${SIGNAL}_client.log" run_server if [ "$SERVER_PID" == "0" ]; then @@ -107,10 +107,87 @@ for SIGNAL in 11 9; do kill_server done +# +# Test User-Defined Model Readiness Function +# +echo -e "\n***\n*** Testing User-Defined is_ready() Function\n***" + +# Helper function to set up test models with different readiness behaviors based on config parameters +setup_readiness_test_model() { + local model_name=$1 + local return_value=$2 + local delay_secs=$3 + + mkdir -p ./models/$model_name/1/ + if [ "$model_name" == "is_ready_fn_coroutine_returns_true" ]; then + cp ./test_models/readiness_coroutine_model.py ./models/$model_name/1/model.py + else + cp ./test_models/readiness_model.py ./models/$model_name/1/model.py + fi + cp ./models/identity_fp32/config.pbtxt ./models/$model_name/config.pbtxt + sed -i "s/^name:.*/name: \"$model_name\"/" ./models/$model_name/config.pbtxt + cat >> ./models/$model_name/config.pbtxt << EOF +parameters: { + key: "READINESS_FN_RETURN_VALUE" + value: { string_value: "$return_value" } +} +parameters: { + key: "READINESS_FN_DELAY_SECS" + value: { string_value: "$delay_secs" } +} +EOF +} + +# Create readiness test models using shared model.py + config parameters +setup_readiness_test_model "is_ready_fn_returns_true" "true" "0.1" +setup_readiness_test_model "is_ready_fn_returns_false" "false" "0.1" +setup_readiness_test_model "is_ready_fn_raises_error" "exception" "0.1" +setup_readiness_test_model "is_ready_fn_returns_non_boolean" "non_boolean" "0.1" +setup_readiness_test_model "is_ready_fn_timeout" "true" "8" +setup_readiness_test_model "is_ready_fn_coroutine_returns_true" "coroutine" "0.1" + +# Decoupled model has a unique execute() and its own config +mkdir -p ./models/is_ready_fn_returns_true_decoupled/1/ +cp ./test_models/is_ready_fn_returns_true_decoupled/model.py \ + ./models/is_ready_fn_returns_true_decoupled/1/model.py +cp ./test_models/is_ready_fn_returns_true_decoupled/config.pbtxt \ + ./models/is_ready_fn_returns_true_decoupled/config.pbtxt + +# Start server with all models +SERVER_ARGS="--model-repository=$(pwd)/models --backend-directory=${BACKEND_DIR} --log-verbose=1" +SERVER_LOG="./test_user_defined_model_readiness_function_server.log" +CLIENT_LOG="./test_user_defined_model_readiness_function_client.log" + +run_server +if [ "$SERVER_PID" == "0" ]; then + cat $SERVER_LOG + echo -e "\n***\n*** Failed to start $SERVER\n***" + exit 1 +fi + +set +e + +echo "Running TestUserDefinedModelReadinessFunction..." +python3 -m unittest test_model_readiness.TestUserDefinedModelReadinessFunction -v >> ${CLIENT_LOG} 2>&1 +TEST_EXIT_CODE=$? + +if [ $TEST_EXIT_CODE -ne 0 ]; then + echo -e "\n***\n*** TestUserDefinedModelReadinessFunction FAILED\n***" + cat ${CLIENT_LOG} + RET=1 +else + echo -e "\n***\n*** TestUserDefinedModelReadinessFunction PASSED\n***" +fi + +set -e +kill_server + + +# Final result if [ $RET -eq 0 ]; then - echo -e "\n***\n*** Test Passed\n***" + echo -e "\n***\n*** All Model Readiness Tests Passed\n***" else - echo -e "\n***\n*** Test FAILED\n***" + echo -e "\n***\n*** Model Readiness Tests FAILED\n***" fi exit $RET diff --git a/qa/L0_backend_python/model_readiness/test_model_readiness.py b/qa/L0_backend_python/model_readiness/test_model_readiness.py index 65d1c81d8a..5330eb1c83 100644 --- a/qa/L0_backend_python/model_readiness/test_model_readiness.py +++ b/qa/L0_backend_python/model_readiness/test_model_readiness.py @@ -1,4 +1,4 @@ -# Copyright 2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# Copyright 2025-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 @@ -24,29 +24,102 @@ # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +import queue +import threading +import time import unittest +from functools import partial +import numpy as np import tritonclient.grpc as grpcclient import tritonclient.http as httpclient +from tritonclient.utils import InferenceServerException + +URL_HTTP = "localhost:8000" +URL_GRPC = "localhost:8001" +DEFAULT_RESPONSE_TIMEOUT = 60 + + +class UserData: + def __init__(self): + self._response_queue = queue.Queue() + + +def callback(user_data, result, error): + if error: + user_data._response_queue.put(error) + else: + user_data._response_queue.put(result) + + +def prepare_infer_args(input_value): + """Create InferInput and InferRequestedOutput lists for decoupled inference.""" + input_data = np.array([[input_value]], dtype=np.int32) + infer_input = [grpcclient.InferInput("IN", input_data.shape, "INT32")] + infer_input[0].set_data_from_numpy(input_data) + outputs = [grpcclient.InferRequestedOutput("OUT")] + return infer_input, outputs + + +def collect_responses(user_data, expected_responses_count): + """ + Collect up to `expected_responses_count` responses from user_data. + """ + errors = [] + responses = [] + recv_count = 0 + while recv_count < expected_responses_count: + try: + result = user_data._response_queue.get(timeout=DEFAULT_RESPONSE_TIMEOUT) + except queue.Empty: + raise Exception( + f"No response received within {DEFAULT_RESPONSE_TIMEOUT} seconds." + ) + if type(result) == InferenceServerException: + errors.append(result) + break + else: + responses.append(result.as_numpy("OUT")[0]) + recv_count = recv_count + 1 + + return errors, responses + + +def call_inference_identity_model(model_name, protocol, client): + """Send an inference request and verify the output matches the input.""" + shape = (1, 8) + input_data = np.ones(shape, dtype=np.float32) + + if protocol == "http": + inputs = [httpclient.InferInput("INPUT0", input_data.shape, "FP32")] + else: + inputs = [grpcclient.InferInput("INPUT0", input_data.shape, "FP32")] + + inputs[0].set_data_from_numpy(input_data) + result = client.infer(model_name, inputs) + output_data = result.as_numpy("OUTPUT0") + + np.testing.assert_array_almost_equal( + input_data, + output_data, + err_msg=f"Inference output mismatch for {model_name}", + ) class TestModelReadiness(unittest.TestCase): def setUp(self): self.model_name = "identity_fp32" - self.url_http = "localhost:8000" - self.url_grpc = "localhost:8001" - self.client_http = httpclient.InferenceServerClient(url=self.url_http) - self.client_grpc = grpcclient.InferenceServerClient(url=self.url_grpc) + self.client_http = httpclient.InferenceServerClient(url=URL_HTTP) + self.client_grpc = grpcclient.InferenceServerClient(url=URL_GRPC) def test_model_ready(self): - print(f"\nTesting if model '{self.model_name}' is READY ...") - # Check HTTP try: is_ready = self.client_http.is_model_ready(self.model_name) self.assertTrue( is_ready, f"[HTTP] Model {self.model_name} should be READY but is NOT" ) + call_inference_identity_model(self.model_name, "http", self.client_http) except Exception as e: self.fail(f"[HTTP] Unexpected error: {str(e)}") @@ -56,12 +129,11 @@ def test_model_ready(self): self.assertTrue( is_ready, f"[gRPC] Model {self.model_name} should be READY but is NOT" ) + call_inference_identity_model(self.model_name, "grpc", self.client_grpc) except Exception as e: self.fail(f"[gRPC] Unexpected error: {str(e)}") def test_model_not_ready(self): - print(f"\nTesting if model '{self.model_name}' is NOT READY ...") - # Check HTTP try: is_ready = self.client_http.is_model_ready(self.model_name) @@ -83,5 +155,364 @@ def test_model_not_ready(self): self.fail(f"[gRPC] Unexpected error: {str(e)}") +class TestUserDefinedModelReadinessFunction(unittest.TestCase): + """ + Test user-defined is_ready() function + """ + + def setUp(self): + self.client_http = httpclient.InferenceServerClient(url=URL_HTTP) + self.client_grpc = grpcclient.InferenceServerClient(url=URL_GRPC) + + def _run_inference_decoupled(self, index, model_name, expected_responses_count): + """Send a decoupled streaming inference request and verify responses.""" + user_data = UserData() + with grpcclient.InferenceServerClient(URL_GRPC) as triton_client: + try: + inputs, outputs = prepare_infer_args(expected_responses_count) + triton_client.start_stream(callback=partial(callback, user_data)) + triton_client.async_stream_infer( + model_name=model_name, inputs=inputs, outputs=outputs + ) + + # Collect and verify responses + errors, responses = collect_responses( + user_data, expected_responses_count + ) + self.assertEqual( + len(responses), + expected_responses_count, + f"Index: {index} - Expected {expected_responses_count} responses, got {len(responses)}", + ) + self.assertEqual( + len(errors), + 0, + f"Index: {index} - Expected 0 errors, got {len(errors)}", + ) + + # Verify correctness of successful responses + for idx, output in enumerate(responses): + self.assertEqual( + output, + expected_responses_count, + msg=f"Response {idx} has incorrect value - {output}", + ) + finally: + triton_client.stop_stream() + + def test_multiple_concurrent_ready_and_infer_requests_decoupled(self): + model_name = "is_ready_fn_returns_true_decoupled" + num_requests = 16 + response_count = 8 + readiness_errors = [] + infer_errors = [] + + def readiness_wrapper(index, model_name): + try: + with grpcclient.InferenceServerClient(url=URL_GRPC) as triton_client: + is_ready = triton_client.is_model_ready(model_name) + if not is_ready: + raise AssertionError( + f"Index: {index} - GRPC client - Model {model_name} should be READY" + ) + except Exception as e: + readiness_errors.append((index, str(e))) + + def inference_wrapper(index, model_name): + try: + self._run_inference_decoupled(index, model_name, response_count) + except Exception as e: + infer_errors.append((index, str(e))) + + # Launch concurrent threads + threads = [] + for i in range(num_requests): + # Start threads with slight delay + time.sleep(0.1) + t1 = threading.Thread( + target=inference_wrapper, args=(i, model_name), name=f"infer-{i}" + ) + t2 = threading.Thread( + target=readiness_wrapper, args=(i, model_name), name=f"ready-{i}" + ) + threads.extend([t1, t2]) + t1.start() + t2.start() + + # Wait for all requests to complete + for t in threads: + t.join(timeout=120) + + for t in threads: + self.assertFalse(t.is_alive(), f"Threads are not completed: {t.name}") + + self.assertEqual( + len(readiness_errors), 0, f"Readiness errors: {readiness_errors}" + ) + self.assertEqual(len(infer_errors), 0, f"Inference errors: {infer_errors}") + + def test_is_ready_coroutine_returns_true(self): + model_name = "is_ready_fn_coroutine_returns_true" + for _ in range(5): + self.assertTrue( + self.client_http.is_model_ready(model_name), + f"HTTP - Model {model_name} (coroutine) should be READY", + ) + self.assertTrue( + self.client_grpc.is_model_ready(model_name), + f"gRPC - Model {model_name} (coroutine) should be READY", + ) + call_inference_identity_model(model_name, "http", self.client_http) + call_inference_identity_model(model_name, "grpc", self.client_grpc) + + def test_is_ready_returns_true(self): + model_name = "is_ready_fn_returns_true" + num_requests = 10 + + # Send multiple requests in sequence to ensure consistent behavior + for i in range(num_requests): + self.assertTrue( + self.client_http.is_model_ready(model_name), + f"iteration {i} - HTTP client - Model {model_name} should be READY", + ) + self.assertTrue( + self.client_grpc.is_model_ready(model_name), + f"iteration {i} - GRPC client - Model {model_name} should be READY", + ) + + # Verify inference is unaffected by readiness checks. + call_inference_identity_model(model_name, "http", self.client_http) + call_inference_identity_model(model_name, "grpc", self.client_grpc) + + def test_is_ready_returns_false(self): + model_name = "is_ready_fn_returns_false" + num_requests = 10 + + # Send multiple requests in sequence to ensure consistent behavior + for i in range(num_requests): + self.assertFalse( + self.client_http.is_model_ready(model_name), + f"iteration {i} - HTTP client - Model {model_name} should be NOT READY", + ) + self.assertFalse( + self.client_grpc.is_model_ready(model_name), + f"iteration {i} - GRPC client - Model {model_name} should be NOT READY", + ) + + # Verify inference is unaffected by readiness checks. + call_inference_identity_model(model_name, "http", self.client_http) + call_inference_identity_model(model_name, "grpc", self.client_grpc) + + def test_is_ready_raises_exception(self): + model_name = "is_ready_fn_raises_error" + num_requests = 10 + + # Send multiple requests in sequence to ensure consistent behavior + for i in range(num_requests): + self.assertFalse( + self.client_http.is_model_ready(model_name), + f"iteration {i} - HTTP client - Model {model_name} should be NOT READY (exception)", + ) + self.assertFalse( + self.client_grpc.is_model_ready(model_name), + f"iteration {i} - GRPC client - Model {model_name} should be NOT READY (exception)", + ) + + # Verify inference is unaffected by readiness checks. + call_inference_identity_model(model_name, "http", self.client_http) + call_inference_identity_model(model_name, "grpc", self.client_grpc) + + # Verify a healthy model is still ready to confirm server stability. + model_name = "is_ready_fn_returns_true" + for i in range(num_requests): + self.assertTrue( + self.client_http.is_model_ready(model_name), + f"iteration {i} - HTTP client - Model {model_name} should be READY", + ) + self.assertTrue( + self.client_grpc.is_model_ready(model_name), + f"iteration {i} - GRPC client - Model {model_name} should be READY", + ) + + # Verify inference is unaffected by readiness checks. + call_inference_identity_model(model_name, "http", self.client_http) + call_inference_identity_model(model_name, "grpc", self.client_grpc) + + def test_is_ready_returns_non_boolean(self): + model_name = "is_ready_fn_returns_non_boolean" + num_requests = 10 + + # Send multiple requests in sequence to ensure consistent behavior + for i in range(num_requests): + self.assertFalse( + self.client_http.is_model_ready(model_name), + f"iteration {i} - HTTP client - Model {model_name} should be NOT READY (wrong return type)", + ) + self.assertFalse( + self.client_grpc.is_model_ready(model_name), + f"iteration {i} - GRPC client - Model {model_name} should be NOT READY (wrong return type)", + ) + + # Verify inference is unaffected by readiness checks. + call_inference_identity_model(model_name, "http", self.client_http) + call_inference_identity_model(model_name, "grpc", self.client_grpc) + + # Verify a healthy model is still ready to confirm server stability. + model_name = "is_ready_fn_returns_true" + for i in range(num_requests): + self.assertTrue( + self.client_http.is_model_ready(model_name), + f"iteration {i} - HTTP client - Model {model_name} should be READY", + ) + self.assertTrue( + self.client_grpc.is_model_ready(model_name), + f"iteration {i} - GRPC client - Model {model_name} should be READY", + ) + + # Verify inference is unaffected by readiness checks. + call_inference_identity_model(model_name, "http", self.client_http) + call_inference_identity_model(model_name, "grpc", self.client_grpc) + + def test_is_ready_takes_long_time(self): + model_name = "is_ready_fn_timeout" + num_requests = 10 + + # Send multiple requests in sequence to ensure consistent behavior + for i in range(num_requests): + # This call should time out and return NOT_READY. + # Note: the stub will continue running is_ready() + # in the background (similar to the inference flow) + # even after the backend readiness timeout expires. + is_ready = self.client_http.is_model_ready(model_name) + self.assertFalse( + is_ready, + f"iteration {i} - HTTP client - Model {model_name} should timeout and return NOT READY", + ) + + call_inference_identity_model(model_name, "http", self.client_http) + + # This call should not create another internal IPC message. + # It must wait for the in-flight readiness check + # and return READY once that check completes. + is_ready = self.client_grpc.is_model_ready(model_name) + self.assertTrue( + is_ready, + f"iteration {i} - GRPC client - Model {model_name} should be READY", + ) + + call_inference_identity_model(model_name, "grpc", self.client_grpc) + + def test_multiple_concurrent_ready_and_infer_requests(self): + model_name = "is_ready_fn_returns_true" + ready_results = {"http": [], "grpc": []} + ready_errors = {"http": [], "grpc": []} + infer_results = {"http": [], "grpc": []} + infer_errors = {"http": [], "grpc": []} + num_requests = 16 + + def check_model_readiness(protocol, index): + try: + if protocol == "http": + with httpclient.InferenceServerClient(url=URL_HTTP) as client_http: + is_ready = client_http.is_model_ready(model_name) + ready_results["http"].append((index, is_ready)) + else: + with grpcclient.InferenceServerClient(url=URL_GRPC) as client_grpc: + is_ready = client_grpc.is_model_ready(model_name) + ready_results["grpc"].append((index, is_ready)) + except Exception as e: + ready_errors[protocol].append((index, str(e))) + + def do_inference(protocol, index): + try: + if protocol == "http": + with httpclient.InferenceServerClient(url=URL_HTTP) as client_http: + start = time.time() + call_inference_identity_model(model_name, protocol, client_http) + elapsed = time.time() - start + infer_results["http"].append((index, True, elapsed)) + else: + with grpcclient.InferenceServerClient(url=URL_GRPC) as client_grpc: + start = time.time() + call_inference_identity_model(model_name, protocol, client_grpc) + elapsed = time.time() - start + infer_results["grpc"].append((index, True, elapsed)) + except Exception as e: + infer_errors[protocol].append((index, str(e))) + + # Launch concurrent readiness and inference requests. + http_threads = [] + for i in range(num_requests): + t1 = threading.Thread(target=check_model_readiness, args=("http", i)) + t2 = threading.Thread(target=do_inference, args=("http", i)) + http_threads.extend([t1, t2]) + t1.start() + t2.start() + + # Wait for all requests to complete + for t in http_threads: + t.join(timeout=60) + + for t in http_threads: + self.assertFalse(t.is_alive(), f"HTTP threads are not completed") + + grpc_threads = [] + for i in range(num_requests): + t1 = threading.Thread(target=check_model_readiness, args=("grpc", i)) + t2 = threading.Thread(target=do_inference, args=("grpc", i)) + grpc_threads.extend([t1, t2]) + t1.start() + t2.start() + + # Wait for all requests to complete + for t in grpc_threads: + t.join(timeout=60) + + for t in grpc_threads: + self.assertFalse(t.is_alive(), f"gRPC threads are not completed") + + # Verify no errors in readiness checks + self.assertEqual( + len(ready_errors["http"]), 0, f"HTTP errors: {ready_errors['http']}" + ) + self.assertEqual( + len(ready_errors["grpc"]), 0, f"gRPC errors: {ready_errors['grpc']}" + ) + self.assertEqual( + len(ready_results["http"]), + num_requests, + f"Expected {num_requests} HTTP results", + ) + self.assertEqual( + len(ready_results["grpc"]), + num_requests, + f"Expected {num_requests} gRPC results", + ) + + # All should be True + for idx, ready in ready_results["http"]: + self.assertTrue(ready, f"HTTP check {idx} should be ready") + for idx, ready in ready_results["grpc"]: + self.assertTrue(ready, f"gRPC check {idx} should be ready") + + # Verify no errors in inference + self.assertEqual( + len(infer_errors["http"]), 0, f"Errors occurred: {infer_errors['http']}" + ) + self.assertEqual( + len(infer_errors["grpc"]), 0, f"Errors occurred: {infer_errors['grpc']}" + ) + self.assertEqual( + len(infer_results["http"]), + num_requests, + f"Expected {num_requests} HTTP inference results", + ) + self.assertEqual( + len(infer_results["grpc"]), + num_requests, + f"Expected {num_requests} gRPC inference results", + ) + + if __name__ == "__main__": unittest.main() diff --git a/qa/L0_backend_python/model_readiness/test_models/is_ready_fn_returns_true_decoupled/config.pbtxt b/qa/L0_backend_python/model_readiness/test_models/is_ready_fn_returns_true_decoupled/config.pbtxt new file mode 100644 index 0000000000..1537b804ab --- /dev/null +++ b/qa/L0_backend_python/model_readiness/test_models/is_ready_fn_returns_true_decoupled/config.pbtxt @@ -0,0 +1,57 @@ +# Copyright 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 +# are met: +# * Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# * Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# * Neither the name of NVIDIA CORPORATION nor the names of its +# contributors may be used to endorse or promote products derived +# from this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS ``AS IS'' AND ANY +# EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR +# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR +# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, +# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR +# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY +# OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + + +backend: "python" +max_batch_size: 1 + +input [ + { + name: "IN" + data_type: TYPE_INT32 + dims: [ 1 ] + } +] + +output [ + { + name: "OUT" + data_type: TYPE_INT32 + dims: [ 1 ] + } +] + +instance_group [ + { + count: 1 + kind: KIND_CPU + } +] + +model_transaction_policy { + decoupled: true +} + diff --git a/qa/L0_backend_python/model_readiness/test_models/is_ready_fn_returns_true_decoupled/model.py b/qa/L0_backend_python/model_readiness/test_models/is_ready_fn_returns_true_decoupled/model.py new file mode 100644 index 0000000000..e2a4dfcb0d --- /dev/null +++ b/qa/L0_backend_python/model_readiness/test_models/is_ready_fn_returns_true_decoupled/model.py @@ -0,0 +1,63 @@ +# Copyright 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 +# are met: +# * Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# * Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# * Neither the name of NVIDIA CORPORATION nor the names of its +# contributors may be used to endorse or promote products derived +# from this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS ``AS IS'' AND ANY +# EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR +# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR +# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, +# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR +# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY +# OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + + +import time + +import numpy as np +import triton_python_backend_utils as pb_utils + + +class TritonPythonModel: + """ + Decoupled model that produces N responses based on input value. + """ + + def execute(self, requests): + for request in requests: + # Get input - number of responses to produce + in_tensor = pb_utils.get_input_tensor_by_name(request, "IN") + count = in_tensor.as_numpy().item() + + response_sender = request.get_response_sender() + out_tensor = pb_utils.Tensor("OUT", np.array([count], dtype=np.int32)) + + # Produce 'count' responses, each with 'count' as the output value + for i in range(count): + # Simulate some processing delay + time.sleep(0.1) + response = pb_utils.InferenceResponse(output_tensors=[out_tensor]) + response_sender.send(response) + + # Send final flag + response_sender.send(flags=pb_utils.TRITONSERVER_RESPONSE_COMPLETE_FINAL) + + return None + + def is_ready(self) -> bool: + # Simulate some processing delay + time.sleep(0.2) + return True diff --git a/qa/L0_backend_python/model_readiness/test_models/readiness_coroutine_model.py b/qa/L0_backend_python/model_readiness/test_models/readiness_coroutine_model.py new file mode 100644 index 0000000000..ce33afe700 --- /dev/null +++ b/qa/L0_backend_python/model_readiness/test_models/readiness_coroutine_model.py @@ -0,0 +1,58 @@ +# Copyright 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 +# are met: +# * Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# * Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# * Neither the name of NVIDIA CORPORATION nor the names of its +# contributors may be used to endorse or promote products derived +# from this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS ``AS IS'' AND ANY +# EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR +# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR +# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, +# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR +# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY +# OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +import asyncio +import json + +import triton_python_backend_utils as pb_utils + + +class TritonPythonModel: + """ + Parameterized test model for async is_ready() testing. + + Behavior is controlled via config.pbtxt parameters: + READINESS_FN_DELAY_SECS - seconds to await before returning (e.g. "0.1") + """ + + def initialize(self, args): + model_config = json.loads(args["model_config"]) + params = model_config.get("parameters", {}) + self.readiness_delay_secs = float( + params.get("READINESS_FN_DELAY_SECS", {}).get("string_value", "0.1") + ) + + def execute(self, requests): + responses = [] + for request in requests: + input_tensor = pb_utils.get_input_tensor_by_name(request, "INPUT0") + out_tensor = pb_utils.Tensor("OUTPUT0", input_tensor.as_numpy()) + responses.append(pb_utils.InferenceResponse([out_tensor])) + return responses + + async def is_ready(self): + await asyncio.sleep(self.readiness_delay_secs) + return True diff --git a/qa/L0_backend_python/model_readiness/test_models/readiness_model.py b/qa/L0_backend_python/model_readiness/test_models/readiness_model.py new file mode 100644 index 0000000000..444bd9e983 --- /dev/null +++ b/qa/L0_backend_python/model_readiness/test_models/readiness_model.py @@ -0,0 +1,72 @@ +# Copyright 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 +# are met: +# * Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# * Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# * Neither the name of NVIDIA CORPORATION nor the names of its +# contributors may be used to endorse or promote products derived +# from this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS ``AS IS'' AND ANY +# EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR +# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR +# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, +# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR +# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY +# OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +import json +import time + +import triton_python_backend_utils as pb_utils + + +class TritonPythonModel: + """ + Parameterized test model for user-defined is_ready() testing. + + Behavior is controlled via config.pbtxt parameters: + READINESS_FN_RETURN_VALUE - "true", "false", "exception", or "non_boolean" + READINESS_FN_DELAY_SECS - seconds to sleep before returning (e.g. "0.1") + """ + + def initialize(self, args): + model_config = json.loads(args["model_config"]) + params = model_config.get("parameters", {}) + self.readiness_return_value = params.get("READINESS_FN_RETURN_VALUE", {}).get( + "string_value", "true" + ) + self.readiness_delay_secs = float( + params.get("READINESS_FN_DELAY_SECS", {}).get("string_value", "0") + ) + + def execute(self, requests): + responses = [] + for request in requests: + input_tensor = pb_utils.get_input_tensor_by_name(request, "INPUT0") + out_tensor = pb_utils.Tensor("OUTPUT0", input_tensor.as_numpy()) + responses.append(pb_utils.InferenceResponse([out_tensor])) + return responses + + def is_ready(self): + if self.readiness_delay_secs > 0: + time.sleep(self.readiness_delay_secs) + + if self.readiness_return_value == "true": + return True + elif self.readiness_return_value == "false": + return False + elif self.readiness_return_value == "exception": + raise RuntimeError("Internal check failed – model is not ready") + elif self.readiness_return_value == "non_boolean": + return "ready" + return True