diff --git a/examples/bls_error_code/README.md b/examples/bls_error_code/README.md new file mode 100644 index 00000000..824c090a --- /dev/null +++ b/examples/bls_error_code/README.md @@ -0,0 +1,41 @@ +# BLS Error Code Propagation Test + +Verifies that error codes are preserved when Python models are chained via +BLS (Business Logic Scripting). + +## The Problem + +Prior to the fix, when Model A returned an `InferenceResponse` with a specific +error code (e.g. `NOT_FOUND`, `UNSUPPORTED`), Model B calling it via BLS would +always see `INTERNAL` regardless of the original code. + +See [triton-inference-server/server#7804](https://github.com/triton-inference-server/server/issues/7804). + +## Models + +- **error_source** — accepts an `ERROR_CODE` int and returns an + `InferenceResponse` with that specific `TritonError` code. +- **bls_error_caller** — calls `error_source` via BLS, reads the error code + from the response, and returns it as `RECEIVED_CODE`. + +## Running + +```bash +# Start Triton with the test models as the model repository +tritonserver --model-repository=examples/bls_error_code + +# In another terminal, run the test (requires tritonclient[http]) +pip install tritonclient[http] +python examples/bls_error_code/test_error_code_propagation.py +``` + +## Expected Output + +``` + Sent: NOT_FOUND (2) -> Got: NOT_FOUND (2) [OK] + Sent: INVALID_ARG (3) -> Got: INVALID_ARG (3) [OK] + Sent: UNAVAILABLE (4) -> Got: UNAVAILABLE (4) [OK] + Sent: UNSUPPORTED (5) -> Got: UNSUPPORTED (5) [OK] + Sent: CANCELLED (7) -> Got: CANCELLED (7) [OK] +ALL 5 TESTS PASSED -- error codes preserved through BLS +``` diff --git a/examples/bls_error_code/bls_error_caller/1/model.py b/examples/bls_error_code/bls_error_caller/1/model.py new file mode 100644 index 00000000..53942cd0 --- /dev/null +++ b/examples/bls_error_code/bls_error_caller/1/model.py @@ -0,0 +1,47 @@ +import numpy as np +import triton_python_backend_utils as pb_utils + + +class TritonPythonModel: + """Calls error_source via BLS and reports back the received error code. + + If error codes propagate correctly, RECEIVED_CODE should match the + ERROR_CODE sent to error_source. + """ + + def initialize(self, args): + pass + + def execute(self, requests): + responses = [] + for request in requests: + error_code_tensor = pb_utils.get_input_tensor_by_name( + request, "ERROR_CODE" + ) + + bls_request = pb_utils.InferenceRequest( + model_name="error_source", + requested_output_names=["OUTPUT"], + inputs=[pb_utils.Tensor("ERROR_CODE", error_code_tensor.as_numpy())], + ) + + bls_response = bls_request.exec() + + if bls_response.has_error(): + error = bls_response.error() + received_code = int(error.code()) + error_msg = str(error.message()) + else: + received_code = -1 + error_msg = "no error" + + out_code = pb_utils.Tensor( + "RECEIVED_CODE", np.array([received_code], dtype=np.int32) + ) + out_msg = pb_utils.Tensor( + "ERROR_MESSAGE", np.array([error_msg], dtype=object) + ) + responses.append( + pb_utils.InferenceResponse(output_tensors=[out_code, out_msg]) + ) + return responses diff --git a/examples/bls_error_code/bls_error_caller/config.pbtxt b/examples/bls_error_code/bls_error_caller/config.pbtxt new file mode 100644 index 00000000..061acb36 --- /dev/null +++ b/examples/bls_error_code/bls_error_caller/config.pbtxt @@ -0,0 +1,23 @@ +name: "bls_error_caller" +backend: "python" +max_batch_size: 0 +input [ + { + name: "ERROR_CODE" + data_type: TYPE_INT32 + dims: [ 1 ] + } +] +output [ + { + name: "RECEIVED_CODE" + data_type: TYPE_INT32 + dims: [ 1 ] + }, + { + name: "ERROR_MESSAGE" + data_type: TYPE_STRING + dims: [ 1 ] + } +] +instance_group [{ kind: KIND_CPU }] diff --git a/examples/bls_error_code/error_source/1/model.py b/examples/bls_error_code/error_source/1/model.py new file mode 100644 index 00000000..cb7bd4a8 --- /dev/null +++ b/examples/bls_error_code/error_source/1/model.py @@ -0,0 +1,37 @@ +import triton_python_backend_utils as pb_utils + + +class TritonPythonModel: + """Returns an InferenceResponse with a specific TritonError code. + + Used to verify that BLS callers receive the original error code + rather than a hardcoded INTERNAL. + """ + + def initialize(self, args): + pass + + def execute(self, requests): + responses = [] + for request in requests: + error_code = pb_utils.get_input_tensor_by_name( + request, "ERROR_CODE" + ).as_numpy()[0] + + code_map = { + 0: pb_utils.TritonError.UNKNOWN, + 1: pb_utils.TritonError.INTERNAL, + 2: pb_utils.TritonError.NOT_FOUND, + 3: pb_utils.TritonError.INVALID_ARG, + 4: pb_utils.TritonError.UNAVAILABLE, + 5: pb_utils.TritonError.UNSUPPORTED, + 6: pb_utils.TritonError.ALREADY_EXISTS, + 7: pb_utils.TritonError.CANCELLED, + } + + triton_code = code_map.get(error_code, pb_utils.TritonError.INTERNAL) + error = pb_utils.TritonError( + f"Test error with code {error_code}", triton_code + ) + responses.append(pb_utils.InferenceResponse(error=error)) + return responses diff --git a/examples/bls_error_code/error_source/config.pbtxt b/examples/bls_error_code/error_source/config.pbtxt new file mode 100644 index 00000000..a6b3748c --- /dev/null +++ b/examples/bls_error_code/error_source/config.pbtxt @@ -0,0 +1,18 @@ +name: "error_source" +backend: "python" +max_batch_size: 0 +input [ + { + name: "ERROR_CODE" + data_type: TYPE_INT32 + dims: [ 1 ] + } +] +output [ + { + name: "OUTPUT" + data_type: TYPE_FP32 + dims: [ 1 ] + } +] +instance_group [{ kind: KIND_CPU }] diff --git a/examples/bls_error_code/test_error_code_propagation.py b/examples/bls_error_code/test_error_code_propagation.py new file mode 100644 index 00000000..c7a274bd --- /dev/null +++ b/examples/bls_error_code/test_error_code_propagation.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python3 +"""Integration test: BLS error code propagation. + +Verifies that when a Python model returns an InferenceResponse with a +specific TritonError code, a BLS caller model receives the same code +(not a hardcoded INTERNAL). + +Usage: + # Start Triton with the test models: + tritonserver --model-repository=examples/bls_error_code + + # Run this test: + python examples/bls_error_code/test_error_code_propagation.py +""" + +import sys +import time + +import numpy as np +import tritonclient.http as httpclient + +CODE_NAMES = { + 0: "UNKNOWN", + 1: "INTERNAL", + 2: "NOT_FOUND", + 3: "INVALID_ARG", + 4: "UNAVAILABLE", + 5: "UNSUPPORTED", + 6: "ALREADY_EXISTS", + 7: "CANCELLED", +} + + +def wait_for_server(url="localhost:8000", timeout=60): + client = httpclient.InferenceServerClient(url) + start = time.time() + while time.time() - start < timeout: + try: + if client.is_server_ready(): + return client + except Exception: + pass + time.sleep(1) + raise RuntimeError(f"Server not ready after {timeout}s") + + +def test_error_code(client, send_code): + inputs = [httpclient.InferInput("ERROR_CODE", [1], "INT32")] + inputs[0].set_data_from_numpy(np.array([send_code], dtype=np.int32)) + outputs = [ + httpclient.InferRequestedOutput("RECEIVED_CODE"), + httpclient.InferRequestedOutput("ERROR_MESSAGE"), + ] + result = client.infer("bls_error_caller", inputs, outputs=outputs) + received = result.as_numpy("RECEIVED_CODE")[0] + msg = result.as_numpy("ERROR_MESSAGE")[0] + if isinstance(msg, bytes): + msg = msg.decode() + return received, msg + + +def main(): + print("Waiting for Triton server...") + client = wait_for_server() + print("Server ready!\n") + + print("=" * 70) + print("BLS Error Code Propagation - Integration Test") + print("=" * 70) + + # Test codes that are NOT the default INTERNAL (1), to catch regressions. + test_codes = [2, 3, 4, 5, 7] # NOT_FOUND, INVALID_ARG, UNAVAILABLE, UNSUPPORTED, CANCELLED + failures = 0 + + for code in test_codes: + received, msg = test_error_code(client, code) + sent_name = CODE_NAMES.get(code, f"?{code}") + recv_name = CODE_NAMES.get(received, f"?{received}") + status = "OK" if received == code else "FAIL" + if status == "FAIL": + failures += 1 + print( + f" Sent: {sent_name:15s} ({code}) -> " + f"Got: {recv_name:15s} ({received}) [{status}]" + ) + + print() + print("=" * 70) + if failures == 0: + print(f"ALL {len(test_codes)} TESTS PASSED -- error codes preserved through BLS") + else: + print(f"{failures}/{len(test_codes)} TESTS FAILED -- error codes lost") + print("=" * 70) + return 1 if failures else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/infer_request.cc b/src/infer_request.cc index ce6004db..c5c12725 100644 --- a/src/infer_request.cc +++ b/src/infer_request.cc @@ -557,14 +557,16 @@ InferRequest::Exec(const bool is_decoupled) PbString::LoadFromSharedMemory(shm_pool, response_batch->error); auto error_response = std::make_unique( std::vector>{}, - std::make_shared(pb_string->String())); + std::make_shared( + pb_string->String(), response_batch->error_code)); return error_response; } else { auto error_response = std::make_unique( std::vector>{}, std::make_shared( - "An error occurred while performing BLS request.")); + "An error occurred while performing BLS request.", + response_batch->error_code)); return error_response; } @@ -573,7 +575,12 @@ InferRequest::Exec(const bool is_decoupled) catch (const PythonBackendException& pb_exception) { auto error_response = std::make_unique( std::vector>{}, - std::make_shared(pb_exception.what())); + std::make_shared( + pb_exception.what(), + pb_exception.HasErrorCode() + ? static_cast( + pb_exception.ErrorCode()) + : TRITONSERVER_ERROR_INTERNAL)); return error_response; } diff --git a/src/pb_exception.h b/src/pb_exception.h index 6f96d02a..d50a929a 100644 --- a/src/pb_exception.h +++ b/src/pb_exception.h @@ -38,9 +38,21 @@ namespace triton { namespace backend { namespace python { struct PythonBackendException : std::exception { PythonBackendException(const std::string& message) : message_(message) {} + // Constructor that preserves an error code (stored as int to avoid + // depending on tritonserver.h in this lightweight header). + PythonBackendException(const std::string& message, int error_code) + : message_(message), error_code_(error_code), has_error_code_(true) + { + } + const char* what() const throw() { return message_.c_str(); } + int ErrorCode() const { return error_code_; } + bool HasErrorCode() const { return has_error_code_; } + std::string message_; + int error_code_ = 0; + bool has_error_code_ = false; }; }}} // namespace triton::backend::python diff --git a/src/pb_stub.cc b/src/pb_stub.cc index 92b5bbd7..8ce5cabd 100644 --- a/src/pb_stub.cc +++ b/src/pb_stub.cc @@ -684,6 +684,7 @@ Stub::ProcessRequests(RequestBatch* request_batch_shm_ptr) std::optional> response_batch; bool has_exception = false; + int exception_error_code = TRITONSERVER_ERROR_INTERNAL; std::string error_string; std::unique_ptr error_string_shm; std::string err_message; @@ -727,6 +728,9 @@ Stub::ProcessRequests(RequestBatch* request_batch_shm_ptr) catch (const PythonBackendException& pb_exception) { has_exception = true; error_string = pb_exception.what(); + if (pb_exception.HasErrorCode()) { + exception_error_code = pb_exception.ErrorCode(); + } } catch (const py::error_already_set& error) { has_exception = true; @@ -763,6 +767,8 @@ Stub::ProcessRequests(RequestBatch* request_batch_shm_ptr) error_string_shm = PbString::Create(shm_pool_, err_message); response_batch_shm_ptr->error = error_string_shm->ShmHandle(); response_batch_shm_ptr->is_error_set = true; + response_batch_shm_ptr->error_code = + static_cast(exception_error_code); response_batch_shm_ptr->batch_size = 0; // Once the error is sent to the backend, the backend is supposed to close // all response factories if not already closed, so closing all response @@ -1548,12 +1554,14 @@ Stub::ProcessBLSResponseDecoupled(std::unique_ptr& ipc_message) PbString::LoadFromSharedMemory(shm_pool_, response_batch->error); infer_response = std::make_unique( std::vector>{}, - std::make_shared(pb_string->String())); + std::make_shared( + pb_string->String(), response_batch->error_code)); } else { infer_response = std::make_unique( std::vector>{}, std::make_shared( - "An error occurred while performing BLS request.")); + "An error occurred while performing BLS request.", + response_batch->error_code)); } } diff --git a/src/pb_utils.h b/src/pb_utils.h index 18591608..a24f4a3d 100644 --- a/src/pb_utils.h +++ b/src/pb_utils.h @@ -85,9 +85,11 @@ constexpr uint64_t kUserModelReadinessTimeoutMs = 5000; do { \ TRITONSERVER_Error* tie_err__ = (X); \ if (tie_err__ != nullptr) { \ + auto error_code__ = TRITONSERVER_ErrorCode(tie_err__); \ auto error_message = std::string(TRITONSERVER_ErrorMessage(tie_err__)); \ TRITONSERVER_ErrorDelete(tie_err__); \ - throw PythonBackendException(error_message); \ + throw PythonBackendException( \ + error_message, static_cast(error_code__)); \ } \ } while (false) @@ -177,6 +179,10 @@ struct ResponseBatch : SendMessageBase { // Indicates whether the response factory has been deleted or not. bool is_response_factory_deleted = false; + + // The error code for batch-level errors. Without this field, batch errors + // always become TRITONSERVER_ERROR_INTERNAL regardless of the actual code. + TRITONSERVER_Error_Code error_code = TRITONSERVER_ERROR_INTERNAL; }; enum LogLevel { kInfo = 0, kWarning, kError, kVerbose }; diff --git a/src/python_be.cc b/src/python_be.cc index 4c9ca49e..976b3bb0 100644 --- a/src/python_be.cc +++ b/src/python_be.cc @@ -679,6 +679,10 @@ ModelInstanceState::ExecuteBLSRequest( catch (const PythonBackendException& pb_exception) { if (is_response_batch_set) { response_batch->has_error = true; + if (pb_exception.HasErrorCode()) { + response_batch->error_code = + static_cast(pb_exception.ErrorCode()); + } LOG_IF_EXCEPTION( pb_error_message = PbString::Create(Stub()->ShmPool(), pb_exception.what())); @@ -1665,11 +1669,11 @@ ModelInstanceState::ProcessRequests( auto error = PbString::LoadFromSharedMemory( Stub()->ShmPool(), response_batch_shm_ptr->error); return TRITONSERVER_ErrorNew( - TRITONSERVER_ERROR_INTERNAL, error->String().c_str()); + response_batch_shm_ptr->error_code, error->String().c_str()); } return TRITONSERVER_ErrorNew( - TRITONSERVER_ERROR_INTERNAL, "Failed to process the requests."); + response_batch_shm_ptr->error_code, "Failed to process the requests."); } if (response_batch_shm_ptr->batch_size > 0) { @@ -1908,6 +1912,7 @@ ModelInstanceState::PrepareResponseBatch( (*response_batch)->is_error_set = false; (*response_batch)->cleanup = false; (*response_batch)->response_size = 1; + (*response_batch)->error_code = TRITONSERVER_ERROR_INTERNAL; } void @@ -1978,6 +1983,10 @@ ModelInstanceState::SendBLSDecoupledResponse( catch (const PythonBackendException& pb_exception) { if (is_response_batch_set) { response_batch->has_error = true; + if (pb_exception.HasErrorCode()) { + response_batch->error_code = + static_cast(pb_exception.ErrorCode()); + } LOG_IF_EXCEPTION( pb_error_message = PbString::Create(Stub()->ShmPool(), pb_exception.what())); diff --git a/src/request_executor.cc b/src/request_executor.cc index 716d3c56..7fe5478b 100644 --- a/src/request_executor.cc +++ b/src/request_executor.cc @@ -40,7 +40,10 @@ TRITONSERVER_Error* CreateTritonErrorFromException(const PythonBackendException& pb_exception) { return TRITONSERVER_ErrorNew( - TRITONSERVER_ERROR_INTERNAL, pb_exception.what()); + pb_exception.HasErrorCode() + ? static_cast(pb_exception.ErrorCode()) + : TRITONSERVER_ERROR_INTERNAL, + pb_exception.what()); } TRITONSERVER_Error* @@ -522,9 +525,13 @@ RequestExecutor::Infer( TRITONSERVER_InferenceRequestDelete(irequest), "Failed to delete inference request."); - throw PythonBackendException( - std::string("Model ") + model_name + - " - Error when running inference: " + pb_exception.what()); + std::string msg = std::string("Model ") + model_name + + " - Error when running inference: " + pb_exception.what(); + if (pb_exception.HasErrorCode()) { + throw PythonBackendException(msg, pb_exception.ErrorCode()); + } else { + throw PythonBackendException(msg); + } } return response_future;