From 1655e56c21823e9604aed4a736f05208266a528c Mon Sep 17 00:00:00 2001 From: Matthew Wittwer Date: Mon, 6 Apr 2026 18:15:43 +0000 Subject: [PATCH 1/3] Add test cases --- .../ensemble_backpressure_test.py | 76 +++++- qa/L0_simple_ensemble/test.sh | 242 ++++++++++++++++++ 2 files changed, 315 insertions(+), 3 deletions(-) diff --git a/qa/L0_simple_ensemble/ensemble_backpressure_test.py b/qa/L0_simple_ensemble/ensemble_backpressure_test.py index fdbe00d028..d693a2a0bf 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,9 +46,17 @@ SERVER_URL = "localhost:8001" DEFAULT_RESPONSE_TIMEOUT = 60 EXPECTED_INFER_OUTPUT = 0.5 +<<<<<<< HEAD NUM_REQUESTS = 16 NUM_RESPONSES_PER_REQUEST = 8 +======= +MODEL_ENSEMBLE_DISABLED = "ensemble_disabled_max_inflight_requests" +MODEL_ENSEMBLE_LIMIT_4 = "ensemble_max_inflight_requests_limit_4" +MODEL_ENSEMBLE_LIMIT_1 = "ensemble_max_inflight_requests_limit_1" +MODEL_ENSEMBLE_PARALLEL_FAILED_ENQUEUE = "ensemble_parallel_step_failed_enqueue" +EXPECTED_PARALLEL_FAILED_ENQUEUE_OUTPUT = 4.0 +>>>>>>> 322782df (Add test cases) class UserData: @@ -76,7 +85,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,10 +93,10 @@ 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." + f"No response received within {timeout} seconds." ) if isinstance(result, InferenceServerException): @@ -486,5 +495,66 @@ 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): + """ + Verify that a parallel ensemble step returns a queue-full error and + preserves valid responses that completed before the failure. + """ + 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..75c199dc8c 100755 --- a/qa/L0_simple_ensemble/test.sh +++ b/qa/L0_simple_ensemble/test.sh @@ -241,9 +241,251 @@ kill $SERVER_PID wait $SERVER_PID +<<<<<<< HEAD ######## 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} +======= +######## Test parallel-step failed enqueue path in ensemble scheduler ######## +MODEL_DIR="`pwd`/parallel_failed_enqueue_test_models" +rm -rf ${MODEL_DIR} + +mkdir -p ${MODEL_DIR}/ensemble_parallel_step_failed_enqueue/1 +mkdir -p ${MODEL_DIR}/decoupled_producer_parallel_queue/1 +mkdir -p ${MODEL_DIR}/slow_consumer_queue_limited/1 +mkdir -p ${MODEL_DIR}/fast_consumer/1 +mkdir -p ${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_models/decoupled_producer/1/model.py \ + ${MODEL_DIR}/decoupled_producer_parallel_queue/1 +cp ./backpressure_test_models/decoupled_producer/config.pbtxt \ + ${MODEL_DIR}/decoupled_producer_parallel_queue/ +sed -i 's/name: "decoupled_producer"/name: "decoupled_producer_parallel_queue"/g' \ + ${MODEL_DIR}/decoupled_producer_parallel_queue/config.pbtxt +sed -i 's/0.5/2.0/g' \ + ${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 \ + ${MODEL_DIR}/slow_consumer_queue_limited/1 +cp ../python_models/ground_truth/config.pbtxt \ + ${MODEL_DIR}/slow_consumer_queue_limited/ +sed -i 's/name: "ground_truth"/name: "slow_consumer_queue_limited"/g' \ + ${MODEL_DIR}/slow_consumer_queue_limited/config.pbtxt +sed -i 's/max_batch_size: 64/max_batch_size: 1/g' \ + ${MODEL_DIR}/slow_consumer_queue_limited/config.pbtxt +cat >> ${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 ${MODEL_DIR}/fast_consumer/1 +cp ../python_models/ground_truth/config.pbtxt ${MODEL_DIR}/fast_consumer/ +sed -i 's/name: "ground_truth"/name: "fast_consumer"/g' \ + ${MODEL_DIR}/fast_consumer/config.pbtxt +sed -i 's/max_batch_size: 64/max_batch_size: 1/g' \ + ${MODEL_DIR}/fast_consumer/config.pbtxt +sed -i 's/time.sleep(delay)/time.sleep(0)/g' \ + ${MODEL_DIR}/fast_consumer/1/model.py + +# Join both parallel branches into the ensemble output. +cp ../python_models/add_sub/model.py ${MODEL_DIR}/join_add_sub/1 +cat > ${MODEL_DIR}/join_add_sub/config.pbtxt << 'EOF' +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 }] +EOF + +cat > ${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=${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 +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 ensemble backpressure feature (max_inflight_requests parameter) ######## +MODEL_DIR="`pwd`/backpressure_test_models" +mkdir -p ${MODEL_DIR}/ensemble_disabled_max_inflight_requests/1 + +rm -rf ${MODEL_DIR}/slow_consumer +mkdir -p ${MODEL_DIR}/slow_consumer/1 +cp ../python_models/ground_truth/model.py ${MODEL_DIR}/slow_consumer/1 +cp ../python_models/ground_truth/config.pbtxt ${MODEL_DIR}/slow_consumer/ +sed -i 's/name: "ground_truth"/name: "slow_consumer"/g' ${MODEL_DIR}/slow_consumer/config.pbtxt + +# Create ensemble with "max_inflight_requests = 4" +rm -rf ${MODEL_DIR}/ensemble_max_inflight_requests_limit_4 +mkdir -p ${MODEL_DIR}/ensemble_max_inflight_requests_limit_4/1 +cp ${MODEL_DIR}/ensemble_disabled_max_inflight_requests/config.pbtxt ${MODEL_DIR}/ensemble_max_inflight_requests_limit_4/ +sed -i 's/ensemble_scheduling {/ensemble_scheduling {\n max_inflight_requests: 4/g' \ + ${MODEL_DIR}/ensemble_max_inflight_requests_limit_4/config.pbtxt + +# Create ensemble with "max_inflight_requests = 1" +rm -rf ${MODEL_DIR}/ensemble_max_inflight_requests_limit_1 +mkdir -p ${MODEL_DIR}/ensemble_max_inflight_requests_limit_1/1 +cp ${MODEL_DIR}/ensemble_disabled_max_inflight_requests/config.pbtxt ${MODEL_DIR}/ensemble_max_inflight_requests_limit_1/ +sed -i 's/platform: "ensemble"/name: "ensemble_max_inflight_requests_limit_1"\nplatform: "ensemble"/g' \ + ${MODEL_DIR}/ensemble_max_inflight_requests_limit_1/config.pbtxt +sed -i 's/ensemble_scheduling {/ensemble_scheduling {\n max_inflight_requests: 1/g' \ + ${MODEL_DIR}/ensemble_max_inflight_requests_limit_1/config.pbtxt +>>>>>>> 322782df (Add test cases) TEST_NAME="EnsembleBackpressureTest" SERVER_LOG="./ensemble_backpressure_test_server.log" From dfeafda5e6a9ec9399f99076b1e3f0048f1d1717 Mon Sep 17 00:00:00 2001 From: Matthew Wittwer Date: Mon, 6 Apr 2026 21:19:14 +0000 Subject: [PATCH 2/3] update testing layout --- qa/L0_simple_ensemble/test.sh | 39 +----------- qa/python_models/join_add_sub/config.pbtxt | 60 ++++++++++++++++++ qa/python_models/join_add_sub/model.py | 72 ++++++++++++++++++++++ 3 files changed, 134 insertions(+), 37 deletions(-) create mode 100644 qa/python_models/join_add_sub/config.pbtxt create mode 100644 qa/python_models/join_add_sub/model.py diff --git a/qa/L0_simple_ensemble/test.sh b/qa/L0_simple_ensemble/test.sh index 75c199dc8c..e980b38cc1 100755 --- a/qa/L0_simple_ensemble/test.sh +++ b/qa/L0_simple_ensemble/test.sh @@ -297,43 +297,8 @@ sed -i 's/time.sleep(delay)/time.sleep(0)/g' \ ${MODEL_DIR}/fast_consumer/1/model.py # Join both parallel branches into the ensemble output. -cp ../python_models/add_sub/model.py ${MODEL_DIR}/join_add_sub/1 -cat > ${MODEL_DIR}/join_add_sub/config.pbtxt << 'EOF' -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 }] -EOF +cp ../python_models/join_add_sub/model.py ${MODEL_DIR}/join_add_sub/1 +cp ../python_models/join_add_sub/config.pbtxt ${MODEL_DIR}/join_add_sub/ cat > ${MODEL_DIR}/ensemble_parallel_step_failed_enqueue/config.pbtxt << 'EOF' name: "ensemble_parallel_step_failed_enqueue" 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 From 6ef2d231dc5e5e911f86556bcd5b6b465da149b4 Mon Sep 17 00:00:00 2001 From: Matthew Wittwer Date: Mon, 6 Apr 2026 22:07:22 +0000 Subject: [PATCH 3/3] Actually resolve merge conflicts --- .../ensemble_backpressure_test.py | 20 ++--- qa/L0_simple_ensemble/test.sh | 89 +++++++------------ 2 files changed, 38 insertions(+), 71 deletions(-) diff --git a/qa/L0_simple_ensemble/ensemble_backpressure_test.py b/qa/L0_simple_ensemble/ensemble_backpressure_test.py index d693a2a0bf..e57b5b8cc0 100755 --- a/qa/L0_simple_ensemble/ensemble_backpressure_test.py +++ b/qa/L0_simple_ensemble/ensemble_backpressure_test.py @@ -46,17 +46,11 @@ SERVER_URL = "localhost:8001" DEFAULT_RESPONSE_TIMEOUT = 60 EXPECTED_INFER_OUTPUT = 0.5 -<<<<<<< HEAD +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 -======= -MODEL_ENSEMBLE_DISABLED = "ensemble_disabled_max_inflight_requests" -MODEL_ENSEMBLE_LIMIT_4 = "ensemble_max_inflight_requests_limit_4" -MODEL_ENSEMBLE_LIMIT_1 = "ensemble_max_inflight_requests_limit_1" -MODEL_ENSEMBLE_PARALLEL_FAILED_ENQUEUE = "ensemble_parallel_step_failed_enqueue" -EXPECTED_PARALLEL_FAILED_ENQUEUE_OUTPUT = 4.0 ->>>>>>> 322782df (Add test cases) class UserData: @@ -95,9 +89,7 @@ def collect_responses(user_data, timeout=DEFAULT_RESPONSE_TIMEOUT): try: result = user_data._response_queue.get(timeout=timeout) except queue.Empty: - raise Exception( - f"No response received within {timeout} seconds." - ) + raise Exception(f"No response received within {timeout} seconds.") if isinstance(result, InferenceServerException): errors.append(result) @@ -498,8 +490,10 @@ def test_step2_max_queue_size(self): class EnsembleParallelFailedEnqueueTest(tu.TestResultCollector): def _run_inference(self, expected_responses_count=32): """ - Verify that a parallel ensemble step returns a queue-full error and - preserves valid responses that completed before the failure. + 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: diff --git a/qa/L0_simple_ensemble/test.sh b/qa/L0_simple_ensemble/test.sh index e980b38cc1..1e62c91e7b 100755 --- a/qa/L0_simple_ensemble/test.sh +++ b/qa/L0_simple_ensemble/test.sh @@ -241,42 +241,37 @@ kill $SERVER_PID wait $SERVER_PID -<<<<<<< HEAD -######## 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} -======= ######## Test parallel-step failed enqueue path in ensemble scheduler ######## -MODEL_DIR="`pwd`/parallel_failed_enqueue_test_models" -rm -rf ${MODEL_DIR} +PARALLEL_FAILED_ENQUEUE_MODEL_DIR="`pwd`/parallel_failed_enqueue_test_models" +rm -rf ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR} -mkdir -p ${MODEL_DIR}/ensemble_parallel_step_failed_enqueue/1 -mkdir -p ${MODEL_DIR}/decoupled_producer_parallel_queue/1 -mkdir -p ${MODEL_DIR}/slow_consumer_queue_limited/1 -mkdir -p ${MODEL_DIR}/fast_consumer/1 -mkdir -p ${MODEL_DIR}/join_add_sub/1 +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_models/decoupled_producer/1/model.py \ - ${MODEL_DIR}/decoupled_producer_parallel_queue/1 -cp ./backpressure_test_models/decoupled_producer/config.pbtxt \ - ${MODEL_DIR}/decoupled_producer_parallel_queue/ +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' \ - ${MODEL_DIR}/decoupled_producer_parallel_queue/config.pbtxt + ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/decoupled_producer_parallel_queue/config.pbtxt sed -i 's/0.5/2.0/g' \ - ${MODEL_DIR}/decoupled_producer_parallel_queue/1/model.py + ${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 \ - ${MODEL_DIR}/slow_consumer_queue_limited/1 + ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/slow_consumer_queue_limited/1 cp ../python_models/ground_truth/config.pbtxt \ - ${MODEL_DIR}/slow_consumer_queue_limited/ + ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/slow_consumer_queue_limited/ sed -i 's/name: "ground_truth"/name: "slow_consumer_queue_limited"/g' \ - ${MODEL_DIR}/slow_consumer_queue_limited/config.pbtxt + ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/slow_consumer_queue_limited/config.pbtxt sed -i 's/max_batch_size: 64/max_batch_size: 1/g' \ - ${MODEL_DIR}/slow_consumer_queue_limited/config.pbtxt -cat >> ${MODEL_DIR}/slow_consumer_queue_limited/config.pbtxt << 'EOF' + ${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 ] @@ -287,20 +282,20 @@ dynamic_batching { EOF # Parallel branch with the same interface and no added delay. -cp ../python_models/ground_truth/model.py ${MODEL_DIR}/fast_consumer/1 -cp ../python_models/ground_truth/config.pbtxt ${MODEL_DIR}/fast_consumer/ +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' \ - ${MODEL_DIR}/fast_consumer/config.pbtxt + ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/fast_consumer/config.pbtxt sed -i 's/max_batch_size: 64/max_batch_size: 1/g' \ - ${MODEL_DIR}/fast_consumer/config.pbtxt + ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/fast_consumer/config.pbtxt sed -i 's/time.sleep(delay)/time.sleep(0)/g' \ - ${MODEL_DIR}/fast_consumer/1/model.py + ${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 ${MODEL_DIR}/join_add_sub/1 -cp ../python_models/join_add_sub/config.pbtxt ${MODEL_DIR}/join_add_sub/ +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 > ${MODEL_DIR}/ensemble_parallel_step_failed_enqueue/config.pbtxt << 'EOF' +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 @@ -385,7 +380,7 @@ 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=${MODEL_DIR}" +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***" @@ -398,6 +393,7 @@ 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 @@ -425,32 +421,9 @@ kill $SERVER_PID > /dev/null 2>&1 || true wait $SERVER_PID > /dev/null 2>&1 || true -######## Test ensemble backpressure feature (max_inflight_requests parameter) ######## -MODEL_DIR="`pwd`/backpressure_test_models" -mkdir -p ${MODEL_DIR}/ensemble_disabled_max_inflight_requests/1 - -rm -rf ${MODEL_DIR}/slow_consumer -mkdir -p ${MODEL_DIR}/slow_consumer/1 -cp ../python_models/ground_truth/model.py ${MODEL_DIR}/slow_consumer/1 -cp ../python_models/ground_truth/config.pbtxt ${MODEL_DIR}/slow_consumer/ -sed -i 's/name: "ground_truth"/name: "slow_consumer"/g' ${MODEL_DIR}/slow_consumer/config.pbtxt - -# Create ensemble with "max_inflight_requests = 4" -rm -rf ${MODEL_DIR}/ensemble_max_inflight_requests_limit_4 -mkdir -p ${MODEL_DIR}/ensemble_max_inflight_requests_limit_4/1 -cp ${MODEL_DIR}/ensemble_disabled_max_inflight_requests/config.pbtxt ${MODEL_DIR}/ensemble_max_inflight_requests_limit_4/ -sed -i 's/ensemble_scheduling {/ensemble_scheduling {\n max_inflight_requests: 4/g' \ - ${MODEL_DIR}/ensemble_max_inflight_requests_limit_4/config.pbtxt - -# Create ensemble with "max_inflight_requests = 1" -rm -rf ${MODEL_DIR}/ensemble_max_inflight_requests_limit_1 -mkdir -p ${MODEL_DIR}/ensemble_max_inflight_requests_limit_1/1 -cp ${MODEL_DIR}/ensemble_disabled_max_inflight_requests/config.pbtxt ${MODEL_DIR}/ensemble_max_inflight_requests_limit_1/ -sed -i 's/platform: "ensemble"/name: "ensemble_max_inflight_requests_limit_1"\nplatform: "ensemble"/g' \ - ${MODEL_DIR}/ensemble_max_inflight_requests_limit_1/config.pbtxt -sed -i 's/ensemble_scheduling {/ensemble_scheduling {\n max_inflight_requests: 1/g' \ - ${MODEL_DIR}/ensemble_max_inflight_requests_limit_1/config.pbtxt ->>>>>>> 322782df (Add test cases) +######## 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} TEST_NAME="EnsembleBackpressureTest" SERVER_LOG="./ensemble_backpressure_test_server.log"