diff --git a/qa/L0_simple_ensemble/ensemble_backpressure_test.py b/qa/L0_simple_ensemble/ensemble_backpressure_test.py index fdbe00d028..e57b5b8cc0 100755 --- a/qa/L0_simple_ensemble/ensemble_backpressure_test.py +++ b/qa/L0_simple_ensemble/ensemble_backpressure_test.py @@ -30,6 +30,7 @@ sys.path.append("../common") +import os import queue import threading import time @@ -45,6 +46,8 @@ SERVER_URL = "localhost:8001" DEFAULT_RESPONSE_TIMEOUT = 60 EXPECTED_INFER_OUTPUT = 0.5 +MODEL_ENSEMBLE_PARALLEL_FAILED_ENQUEUE = "ensemble_parallel_step_failed_enqueue" +EXPECTED_PARALLEL_FAILED_ENQUEUE_OUTPUT = 4.0 NUM_REQUESTS = 16 NUM_RESPONSES_PER_REQUEST = 8 @@ -76,7 +79,7 @@ def prepare_infer_args(input_value, enable_batching=False): return infer_input, outputs -def collect_responses(user_data): +def collect_responses(user_data, timeout=DEFAULT_RESPONSE_TIMEOUT): """ Collect responses from user_data until the final response flag is seen. """ @@ -84,11 +87,9 @@ def collect_responses(user_data): responses = [] while True: try: - result = user_data._response_queue.get(timeout=DEFAULT_RESPONSE_TIMEOUT) + result = user_data._response_queue.get(timeout=timeout) except queue.Empty: - raise Exception( - f"No response received within {DEFAULT_RESPONSE_TIMEOUT} seconds." - ) + raise Exception(f"No response received within {timeout} seconds.") if isinstance(result, InferenceServerException): errors.append(result) @@ -486,5 +487,68 @@ def test_step2_max_queue_size(self): self._run_inference(model_name=model_name, expected_responses_count=32) +class EnsembleParallelFailedEnqueueTest(tu.TestResultCollector): + def _run_inference(self, expected_responses_count=32): + """ + Exercise a fan-out ensemble where one parallel branch hits queue-full + first. Successful responses emitted before the failure should still be + correct, and the stream should terminate with exactly one queue-full + error. + """ + user_data = UserData() + with grpcclient.InferenceServerClient(SERVER_URL) 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_ENSEMBLE_PARALLEL_FAILED_ENQUEUE, + inputs=inputs, + outputs=outputs, + ) + + errors, responses = collect_responses(user_data, timeout=15) + self.assertLess( + len(responses), + expected_responses_count, + "Expected the parallel slow branch to queue-fill before all " + "responses completed.", + ) + self.assertEqual( + len(errors), + 1, + "Expected exactly one queue-full error from the parallel " + "failed-enqueue path.", + ) + self.assertIn( + "Exceeds maximum queue size", + str(errors[0]), + f"Expected queue size error, got: {str(errors[0])}", + ) + + for idx, resp in enumerate(responses): + output = resp.as_numpy("OUT") + self.assertAlmostEqual( + float(np.squeeze(output)), + EXPECTED_PARALLEL_FAILED_ENQUEUE_OUTPUT, + places=5, + msg=f"Response {idx} has incorrect value - {output}", + ) + finally: + triton_client.stop_stream() + + def test_parallel_step_failed_enqueue(self): + """ + Repeat the same request according to PARALLEL_FAILED_ENQUEUE_LOOPS. + """ + loop_count = int(os.environ.get("PARALLEL_FAILED_ENQUEUE_LOOPS", "1")) + self.assertGreaterEqual( + loop_count, 1, "PARALLEL_FAILED_ENQUEUE_LOOPS must be >= 1" + ) + + for iteration in range(loop_count): + with self.subTest(iteration=iteration): + self._run_inference() + + if __name__ == "__main__": unittest.main() diff --git a/qa/L0_simple_ensemble/test.sh b/qa/L0_simple_ensemble/test.sh index 95a0efa4a2..1e62c91e7b 100755 --- a/qa/L0_simple_ensemble/test.sh +++ b/qa/L0_simple_ensemble/test.sh @@ -241,6 +241,186 @@ kill $SERVER_PID wait $SERVER_PID +######## Test parallel-step failed enqueue path in ensemble scheduler ######## +PARALLEL_FAILED_ENQUEUE_MODEL_DIR="`pwd`/parallel_failed_enqueue_test_models" +rm -rf ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR} + +mkdir -p ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/ensemble_parallel_step_failed_enqueue/1 +mkdir -p ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/decoupled_producer_parallel_queue/1 +mkdir -p ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/slow_consumer_queue_limited/1 +mkdir -p ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/fast_consumer/1 +mkdir -p ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/join_add_sub/1 + +# Producer emits repeated responses with a larger payload value so the +# queue-limited branch fills first. +cp ${BACKPRESSURE_TEST_MODEL_DIR}/decoupled_producer/1/model.py \ + ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/decoupled_producer_parallel_queue/1 +cp ${BACKPRESSURE_TEST_MODEL_DIR}/decoupled_producer/config.pbtxt \ + ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/decoupled_producer_parallel_queue/ +sed -i 's/name: "decoupled_producer"/name: "decoupled_producer_parallel_queue"/g' \ + ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/decoupled_producer_parallel_queue/config.pbtxt +sed -i 's/0.5/2.0/g' \ + ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/decoupled_producer_parallel_queue/1/model.py + +# Queue-limited branch used to trigger a failed enqueue. +cp ../python_models/ground_truth/model.py \ + ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/slow_consumer_queue_limited/1 +cp ../python_models/ground_truth/config.pbtxt \ + ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/slow_consumer_queue_limited/ +sed -i 's/name: "ground_truth"/name: "slow_consumer_queue_limited"/g' \ + ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/slow_consumer_queue_limited/config.pbtxt +sed -i 's/max_batch_size: 64/max_batch_size: 1/g' \ + ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/slow_consumer_queue_limited/config.pbtxt +cat >> ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/slow_consumer_queue_limited/config.pbtxt << 'EOF' + +dynamic_batching { + preferred_batch_size: [ 1 ] + default_queue_policy { + max_queue_size: 1 + } +} +EOF + +# Parallel branch with the same interface and no added delay. +cp ../python_models/ground_truth/model.py ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/fast_consumer/1 +cp ../python_models/ground_truth/config.pbtxt ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/fast_consumer/ +sed -i 's/name: "ground_truth"/name: "fast_consumer"/g' \ + ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/fast_consumer/config.pbtxt +sed -i 's/max_batch_size: 64/max_batch_size: 1/g' \ + ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/fast_consumer/config.pbtxt +sed -i 's/time.sleep(delay)/time.sleep(0)/g' \ + ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/fast_consumer/1/model.py + +# Join both parallel branches into the ensemble output. +cp ../python_models/join_add_sub/model.py ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/join_add_sub/1 +cp ../python_models/join_add_sub/config.pbtxt ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/join_add_sub/ + +cat > ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/ensemble_parallel_step_failed_enqueue/config.pbtxt << 'EOF' +name: "ensemble_parallel_step_failed_enqueue" +platform: "ensemble" +max_batch_size: 0 + +input [ + { + name: "IN" + data_type: TYPE_INT32 + dims: [ 1 ] + } +] + +output [ + { + name: "OUT" + data_type: TYPE_FP32 + dims: [ 1 ] + } +] + +ensemble_scheduling { + step [ + { + model_name: "decoupled_producer_parallel_queue" + model_version: -1 + input_map { + key: "IN" + value: "IN" + } + output_map { + key: "OUT" + value: "intermediate" + } + }, + { + model_name: "slow_consumer_queue_limited" + model_version: -1 + input_map { + key: "INPUT0" + value: "intermediate" + } + output_map { + key: "OUTPUT0" + value: "slow_out" + } + }, + { + model_name: "fast_consumer" + model_version: -1 + input_map { + key: "INPUT0" + value: "intermediate" + } + output_map { + key: "OUTPUT0" + value: "fast_out" + } + }, + { + model_name: "join_add_sub" + model_version: -1 + input_map { + key: "INPUT0" + value: "slow_out" + } + input_map { + key: "INPUT1" + value: "fast_out" + } + output_map { + key: "OUTPUT0" + value: "OUT" + } + } + ] +} +EOF + +BACKPRESSURE_TEST_PY=./ensemble_backpressure_test.py +TEST_NAME="EnsembleParallelFailedEnqueueTest.test_parallel_step_failed_enqueue" +SERVER_LOG="./ensemble_parallel_failed_enqueue_test_server.log" +CLIENT_LOG="./ensemble_parallel_failed_enqueue_test_client.log" +rm -f $SERVER_LOG $CLIENT_LOG + +SERVER_ARGS="--model-repository=${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}" +run_server +if [ "$SERVER_PID" == "0" ]; then + echo -e "\n***\n*** Failed to start $SERVER\n***" + cat $SERVER_LOG + exit 1 +fi + +set +e +PARALLEL_FAILED_ENQUEUE_LOOPS=${PARALLEL_FAILED_ENQUEUE_LOOPS:-1} \ +python $BACKPRESSURE_TEST_PY $TEST_NAME -v >> $CLIENT_LOG 2>&1 +if [ $? -ne 0 ]; then + RET=1 + cat $CLIENT_LOG +else + check_test_results $TEST_RESULT_FILE 1 + if [ $? -ne 0 ]; then + cat $CLIENT_LOG + echo -e "\n***\n*** Test Result Verification Failed\n***" + RET=1 + fi +fi + +if ! kill -0 $SERVER_PID > /dev/null 2>&1; then + cat $SERVER_LOG + echo -e "\n***\n*** Server exited during parallel failed enqueue test\n***" + RET=1 +else + wait_for_server_live $SERVER_PID 5 + if [ "$WAIT_RET" != "0" ]; then + cat $SERVER_LOG + echo -e "\n***\n*** Server did not remain live after parallel failed enqueue test\n***" + RET=1 + fi +fi +set -e + +kill $SERVER_PID > /dev/null 2>&1 || true +wait $SERVER_PID > /dev/null 2>&1 || true + + ######## Test backpressure feature - 'max_inflight_requests' config option ######## ENSEMBLE_BACKPRESSURE_TEST_MODEL_DIR="`pwd`/ensemble_backpressure_test_models" rm -rf ${ENSEMBLE_BACKPRESSURE_TEST_MODEL_DIR} diff --git a/qa/python_models/join_add_sub/config.pbtxt b/qa/python_models/join_add_sub/config.pbtxt new file mode 100644 index 0000000000..f1abd51ec3 --- /dev/null +++ b/qa/python_models/join_add_sub/config.pbtxt @@ -0,0 +1,60 @@ +# 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. + +name: "join_add_sub" +backend: "python" +max_batch_size: 1 + +input [ + { + name: "INPUT0" + data_type: TYPE_FP32 + dims: [ 1 ] + } +] +input [ + { + name: "INPUT1" + data_type: TYPE_FP32 + dims: [ 1 ] + } +] +output [ + { + name: "OUTPUT0" + data_type: TYPE_FP32 + dims: [ 1 ] + } +] +output [ + { + name: "OUTPUT1" + data_type: TYPE_FP32 + dims: [ 1 ] + } +] + +instance_group [{ kind: KIND_CPU }] diff --git a/qa/python_models/join_add_sub/model.py b/qa/python_models/join_add_sub/model.py new file mode 100644 index 0000000000..79371cf38f --- /dev/null +++ b/qa/python_models/join_add_sub/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 numpy as np +import triton_python_backend_utils as pb_utils + + +class TritonPythonModel: + def initialize(self, args): + self.model_config = model_config = json.loads(args["model_config"]) + + output0_config = pb_utils.get_output_config_by_name(model_config, "OUTPUT0") + output1_config = pb_utils.get_output_config_by_name(model_config, "OUTPUT1") + + self.output0_dtype = pb_utils.triton_string_to_numpy( + output0_config["data_type"] + ) + self.output1_dtype = pb_utils.triton_string_to_numpy( + output1_config["data_type"] + ) + + def execute(self, requests): + output0_dtype = self.output0_dtype + output1_dtype = self.output1_dtype + + responses = [] + for request in requests: + in_0 = pb_utils.get_input_tensor_by_name(request, "INPUT0") + in_1 = pb_utils.get_input_tensor_by_name(request, "INPUT1") + if ( + in_0.as_numpy().dtype.type is np.bytes_ + or in_0.as_numpy().dtype == np.object_ + ): + out_0, out_1 = ( + in_0.as_numpy().astype(np.int32) + in_1.as_numpy().astype(np.int32), + in_0.as_numpy().astype(np.int32) - in_1.as_numpy().astype(np.int32), + ) + else: + out_0, out_1 = ( + in_0.as_numpy() + in_1.as_numpy(), + in_0.as_numpy() - in_1.as_numpy(), + ) + + out_tensor_0 = pb_utils.Tensor("OUTPUT0", out_0.astype(output0_dtype)) + out_tensor_1 = pb_utils.Tensor("OUTPUT1", out_1.astype(output1_dtype)) + responses.append(pb_utils.InferenceResponse([out_tensor_0, out_tensor_1])) + return responses