Skip to content

Commit 7a1ff16

Browse files
mattwittwermattwittwer
authored
test: Add test case for RequestTracker counter mismatch (#8722)
1 parent 399a795 commit 7a1ff16

4 files changed

Lines changed: 381 additions & 5 deletions

File tree

‎qa/L0_simple_ensemble/ensemble_backpressure_test.py‎

Lines changed: 69 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030

3131
sys.path.append("../common")
3232

33+
import os
3334
import queue
3435
import threading
3536
import time
@@ -45,6 +46,8 @@
4546
SERVER_URL = "localhost:8001"
4647
DEFAULT_RESPONSE_TIMEOUT = 60
4748
EXPECTED_INFER_OUTPUT = 0.5
49+
MODEL_ENSEMBLE_PARALLEL_FAILED_ENQUEUE = "ensemble_parallel_step_failed_enqueue"
50+
EXPECTED_PARALLEL_FAILED_ENQUEUE_OUTPUT = 4.0
4851

4952
NUM_REQUESTS = 16
5053
NUM_RESPONSES_PER_REQUEST = 8
@@ -76,19 +79,17 @@ def prepare_infer_args(input_value, enable_batching=False):
7679
return infer_input, outputs
7780

7881

79-
def collect_responses(user_data):
82+
def collect_responses(user_data, timeout=DEFAULT_RESPONSE_TIMEOUT):
8083
"""
8184
Collect responses from user_data until the final response flag is seen.
8285
"""
8386
errors = []
8487
responses = []
8588
while True:
8689
try:
87-
result = user_data._response_queue.get(timeout=DEFAULT_RESPONSE_TIMEOUT)
90+
result = user_data._response_queue.get(timeout=timeout)
8891
except queue.Empty:
89-
raise Exception(
90-
f"No response received within {DEFAULT_RESPONSE_TIMEOUT} seconds."
91-
)
92+
raise Exception(f"No response received within {timeout} seconds.")
9293

9394
if isinstance(result, InferenceServerException):
9495
errors.append(result)
@@ -486,5 +487,68 @@ def test_step2_max_queue_size(self):
486487
self._run_inference(model_name=model_name, expected_responses_count=32)
487488

488489

490+
class EnsembleParallelFailedEnqueueTest(tu.TestResultCollector):
491+
def _run_inference(self, expected_responses_count=32):
492+
"""
493+
Exercise a fan-out ensemble where one parallel branch hits queue-full
494+
first. Successful responses emitted before the failure should still be
495+
correct, and the stream should terminate with exactly one queue-full
496+
error.
497+
"""
498+
user_data = UserData()
499+
with grpcclient.InferenceServerClient(SERVER_URL) as triton_client:
500+
try:
501+
inputs, outputs = prepare_infer_args(expected_responses_count)
502+
triton_client.start_stream(callback=partial(callback, user_data))
503+
triton_client.async_stream_infer(
504+
model_name=MODEL_ENSEMBLE_PARALLEL_FAILED_ENQUEUE,
505+
inputs=inputs,
506+
outputs=outputs,
507+
)
508+
509+
errors, responses = collect_responses(user_data, timeout=15)
510+
self.assertLess(
511+
len(responses),
512+
expected_responses_count,
513+
"Expected the parallel slow branch to queue-fill before all "
514+
"responses completed.",
515+
)
516+
self.assertEqual(
517+
len(errors),
518+
1,
519+
"Expected exactly one queue-full error from the parallel "
520+
"failed-enqueue path.",
521+
)
522+
self.assertIn(
523+
"Exceeds maximum queue size",
524+
str(errors[0]),
525+
f"Expected queue size error, got: {str(errors[0])}",
526+
)
527+
528+
for idx, resp in enumerate(responses):
529+
output = resp.as_numpy("OUT")
530+
self.assertAlmostEqual(
531+
float(np.squeeze(output)),
532+
EXPECTED_PARALLEL_FAILED_ENQUEUE_OUTPUT,
533+
places=5,
534+
msg=f"Response {idx} has incorrect value - {output}",
535+
)
536+
finally:
537+
triton_client.stop_stream()
538+
539+
def test_parallel_step_failed_enqueue(self):
540+
"""
541+
Repeat the same request according to PARALLEL_FAILED_ENQUEUE_LOOPS.
542+
"""
543+
loop_count = int(os.environ.get("PARALLEL_FAILED_ENQUEUE_LOOPS", "1"))
544+
self.assertGreaterEqual(
545+
loop_count, 1, "PARALLEL_FAILED_ENQUEUE_LOOPS must be >= 1"
546+
)
547+
548+
for iteration in range(loop_count):
549+
with self.subTest(iteration=iteration):
550+
self._run_inference()
551+
552+
489553
if __name__ == "__main__":
490554
unittest.main()

‎qa/L0_simple_ensemble/test.sh‎

Lines changed: 180 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -241,6 +241,186 @@ kill $SERVER_PID
241241
wait $SERVER_PID
242242

243243

244+
######## Test parallel-step failed enqueue path in ensemble scheduler ########
245+
PARALLEL_FAILED_ENQUEUE_MODEL_DIR="`pwd`/parallel_failed_enqueue_test_models"
246+
rm -rf ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}
247+
248+
mkdir -p ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/ensemble_parallel_step_failed_enqueue/1
249+
mkdir -p ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/decoupled_producer_parallel_queue/1
250+
mkdir -p ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/slow_consumer_queue_limited/1
251+
mkdir -p ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/fast_consumer/1
252+
mkdir -p ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/join_add_sub/1
253+
254+
# Producer emits repeated responses with a larger payload value so the
255+
# queue-limited branch fills first.
256+
cp ${BACKPRESSURE_TEST_MODEL_DIR}/decoupled_producer/1/model.py \
257+
${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/decoupled_producer_parallel_queue/1
258+
cp ${BACKPRESSURE_TEST_MODEL_DIR}/decoupled_producer/config.pbtxt \
259+
${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/decoupled_producer_parallel_queue/
260+
sed -i 's/name: "decoupled_producer"/name: "decoupled_producer_parallel_queue"/g' \
261+
${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/decoupled_producer_parallel_queue/config.pbtxt
262+
sed -i 's/0.5/2.0/g' \
263+
${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/decoupled_producer_parallel_queue/1/model.py
264+
265+
# Queue-limited branch used to trigger a failed enqueue.
266+
cp ../python_models/ground_truth/model.py \
267+
${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/slow_consumer_queue_limited/1
268+
cp ../python_models/ground_truth/config.pbtxt \
269+
${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/slow_consumer_queue_limited/
270+
sed -i 's/name: "ground_truth"/name: "slow_consumer_queue_limited"/g' \
271+
${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/slow_consumer_queue_limited/config.pbtxt
272+
sed -i 's/max_batch_size: 64/max_batch_size: 1/g' \
273+
${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/slow_consumer_queue_limited/config.pbtxt
274+
cat >> ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/slow_consumer_queue_limited/config.pbtxt << 'EOF'
275+
276+
dynamic_batching {
277+
preferred_batch_size: [ 1 ]
278+
default_queue_policy {
279+
max_queue_size: 1
280+
}
281+
}
282+
EOF
283+
284+
# Parallel branch with the same interface and no added delay.
285+
cp ../python_models/ground_truth/model.py ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/fast_consumer/1
286+
cp ../python_models/ground_truth/config.pbtxt ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/fast_consumer/
287+
sed -i 's/name: "ground_truth"/name: "fast_consumer"/g' \
288+
${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/fast_consumer/config.pbtxt
289+
sed -i 's/max_batch_size: 64/max_batch_size: 1/g' \
290+
${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/fast_consumer/config.pbtxt
291+
sed -i 's/time.sleep(delay)/time.sleep(0)/g' \
292+
${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/fast_consumer/1/model.py
293+
294+
# Join both parallel branches into the ensemble output.
295+
cp ../python_models/join_add_sub/model.py ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/join_add_sub/1
296+
cp ../python_models/join_add_sub/config.pbtxt ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/join_add_sub/
297+
298+
cat > ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/ensemble_parallel_step_failed_enqueue/config.pbtxt << 'EOF'
299+
name: "ensemble_parallel_step_failed_enqueue"
300+
platform: "ensemble"
301+
max_batch_size: 0
302+
303+
input [
304+
{
305+
name: "IN"
306+
data_type: TYPE_INT32
307+
dims: [ 1 ]
308+
}
309+
]
310+
311+
output [
312+
{
313+
name: "OUT"
314+
data_type: TYPE_FP32
315+
dims: [ 1 ]
316+
}
317+
]
318+
319+
ensemble_scheduling {
320+
step [
321+
{
322+
model_name: "decoupled_producer_parallel_queue"
323+
model_version: -1
324+
input_map {
325+
key: "IN"
326+
value: "IN"
327+
}
328+
output_map {
329+
key: "OUT"
330+
value: "intermediate"
331+
}
332+
},
333+
{
334+
model_name: "slow_consumer_queue_limited"
335+
model_version: -1
336+
input_map {
337+
key: "INPUT0"
338+
value: "intermediate"
339+
}
340+
output_map {
341+
key: "OUTPUT0"
342+
value: "slow_out"
343+
}
344+
},
345+
{
346+
model_name: "fast_consumer"
347+
model_version: -1
348+
input_map {
349+
key: "INPUT0"
350+
value: "intermediate"
351+
}
352+
output_map {
353+
key: "OUTPUT0"
354+
value: "fast_out"
355+
}
356+
},
357+
{
358+
model_name: "join_add_sub"
359+
model_version: -1
360+
input_map {
361+
key: "INPUT0"
362+
value: "slow_out"
363+
}
364+
input_map {
365+
key: "INPUT1"
366+
value: "fast_out"
367+
}
368+
output_map {
369+
key: "OUTPUT0"
370+
value: "OUT"
371+
}
372+
}
373+
]
374+
}
375+
EOF
376+
377+
BACKPRESSURE_TEST_PY=./ensemble_backpressure_test.py
378+
TEST_NAME="EnsembleParallelFailedEnqueueTest.test_parallel_step_failed_enqueue"
379+
SERVER_LOG="./ensemble_parallel_failed_enqueue_test_server.log"
380+
CLIENT_LOG="./ensemble_parallel_failed_enqueue_test_client.log"
381+
rm -f $SERVER_LOG $CLIENT_LOG
382+
383+
SERVER_ARGS="--model-repository=${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}"
384+
run_server
385+
if [ "$SERVER_PID" == "0" ]; then
386+
echo -e "\n***\n*** Failed to start $SERVER\n***"
387+
cat $SERVER_LOG
388+
exit 1
389+
fi
390+
391+
set +e
392+
PARALLEL_FAILED_ENQUEUE_LOOPS=${PARALLEL_FAILED_ENQUEUE_LOOPS:-1} \
393+
python $BACKPRESSURE_TEST_PY $TEST_NAME -v >> $CLIENT_LOG 2>&1
394+
if [ $? -ne 0 ]; then
395+
RET=1
396+
cat $CLIENT_LOG
397+
else
398+
check_test_results $TEST_RESULT_FILE 1
399+
if [ $? -ne 0 ]; then
400+
cat $CLIENT_LOG
401+
echo -e "\n***\n*** Test Result Verification Failed\n***"
402+
RET=1
403+
fi
404+
fi
405+
406+
if ! kill -0 $SERVER_PID > /dev/null 2>&1; then
407+
cat $SERVER_LOG
408+
echo -e "\n***\n*** Server exited during parallel failed enqueue test\n***"
409+
RET=1
410+
else
411+
wait_for_server_live $SERVER_PID 5
412+
if [ "$WAIT_RET" != "0" ]; then
413+
cat $SERVER_LOG
414+
echo -e "\n***\n*** Server did not remain live after parallel failed enqueue test\n***"
415+
RET=1
416+
fi
417+
fi
418+
set -e
419+
420+
kill $SERVER_PID > /dev/null 2>&1 || true
421+
wait $SERVER_PID > /dev/null 2>&1 || true
422+
423+
244424
######## Test backpressure feature - 'max_inflight_requests' config option ########
245425
ENSEMBLE_BACKPRESSURE_TEST_MODEL_DIR="`pwd`/ensemble_backpressure_test_models"
246426
rm -rf ${ENSEMBLE_BACKPRESSURE_TEST_MODEL_DIR}

‎qa/python_models/join_add_sub/config.pbtxt‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,60 @@
1+
# Copyright 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2+
#
3+
# Redistribution and use in source and binary forms, with or without
4+
# modification, are permitted provided that the following conditions
5+
# are met:
6+
# * Redistributions of source code must retain the above copyright
7+
# notice, this list of conditions and the following disclaimer.
8+
# * Redistributions in binary form must reproduce the above copyright
9+
# notice, this list of conditions and the following disclaimer in the
10+
# documentation and/or other materials provided with the distribution.
11+
# * Neither the name of NVIDIA CORPORATION nor the names of its
12+
# contributors may be used to endorse or promote products derived
13+
# from this software without specific prior written permission.
14+
#
15+
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS ``AS IS'' AND ANY
16+
# EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
17+
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
18+
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
19+
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
20+
# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
21+
# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
22+
# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY
23+
# OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
24+
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
25+
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
26+
27+
name: "join_add_sub"
28+
backend: "python"
29+
max_batch_size: 1
30+
31+
input [
32+
{
33+
name: "INPUT0"
34+
data_type: TYPE_FP32
35+
dims: [ 1 ]
36+
}
37+
]
38+
input [
39+
{
40+
name: "INPUT1"
41+
data_type: TYPE_FP32
42+
dims: [ 1 ]
43+
}
44+
]
45+
output [
46+
{
47+
name: "OUTPUT0"
48+
data_type: TYPE_FP32
49+
dims: [ 1 ]
50+
}
51+
]
52+
output [
53+
{
54+
name: "OUTPUT1"
55+
data_type: TYPE_FP32
56+
dims: [ 1 ]
57+
}
58+
]
59+
60+
instance_group [{ kind: KIND_CPU }]

0 commit comments

Comments
 (0)