Cannot remove future from background; KeyError: <Future at ... state=finished returned NoneType>

[zjq1996518]

Description
When deploying a model using the vllm backend of Triton Server version 25.10, a large number of error logs with the message "Cannot remove future from background; KeyError: <Future at ... state=finished returned NoneType>" are continuously output after sending inference requests to the model. However, the request results are returned normally. Currently, there are no additional exception logs and no effective debugging methods available, making it impossible to locate the root cause and handle the issue.

E1124 02:31:22.019371 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x78543047f020 state=finished returned NoneType>"
E1124 02:31:22.030277 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x7857a2b699a0 state=finished returned NoneType>"
E1124 02:31:22.038870 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x7857a2b6be30 state=finished returned NoneType>"
E1124 02:31:22.043864 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x7857a2b69250 state=finished returned NoneType>"
E1124 02:31:22.076672 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x7857941dffe0 state=finished returned NoneType>"
E1124 02:31:22.091349 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x7857941dcbf0 state=finished returned NoneType>"
E1124 02:31:22.097797 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x7857941dd0a0 state=finished returned NoneType>"
E1124 02:31:22.515902 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x78579419ba10 state=finished returned NoneType>"
E1124 02:31:22.529051 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x78579419bf20 state=finished returned NoneType>"
E1124 02:31:22.550514 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x785430438b60 state=finished returned NoneType>"
E1124 02:31:22.554332 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x785430438ef0 state=finished returned NoneType>"
E1124 02:31:22.639060 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x78543045f020 state=finished returned NoneType>"
E1124 02:31:22.642232 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x7857941dc770 state=finished returned NoneType>"
E1124 02:31:22.661225 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x78576fa9ffe0 state=finished returned NoneType>"
E1124 02:31:22.675548 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x7857941de540 state=finished returned NoneType>"
E1124 02:31:22.700992 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x7857a1363f50 state=finished returned NoneType>"
E1124 02:31:22.712640 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x7853e0bcc290 state=finished returned NoneType>"
E1124 02:31:22.718041 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x7853e0bcf0b0 state=finished returned NoneType>"
E1124 02:31:22.730276 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x7857941df710 state=finished returned NoneType>"
E1124 02:31:22.737700 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x7853e0bcc800 state=finished returned NoneType>"
E1124 02:31:22.739400 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x7853e0bccdd0 state=finished returned NoneType>"
E1124 02:31:22.742066 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x7853e0bccbf0 state=finished returned NoneType>"
E1124 02:31:23.252599 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x7857941dd8e0 state=finished returned NoneType>"
E1124 02:31:23.266090 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x7857a1f1e450 state=finished returned NoneType>"
E1124 02:31:23.345243 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x78543045c590 state=finished returned NoneType>"
E1124 02:31:23.347122 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x7857a1360dd0 state=finished returned NoneType>"
E1124 02:31:23.355554 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x7857a1361fd0 state=finished returned NoneType>"
E1124 02:31:23.363940 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x78576fa9d580 state=finished returned NoneType>"
E1124 02:31:23.368886 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x78576fa9dcd0 state=finished returned NoneType>"
E1124 02:31:23.428142 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x7857941c27b0 state=finished returned NoneType>"
E1124 02:31:23.843190 568 pb_stub.cc:953] "Cannot remove future from background; KeyError: <Future at 0x7857a1361a30 state=finished returned NoneType>"

Triton Information
Triton version: 25.10
Usage method: Official Triton container (vllm backend)

Are you using the Triton container or did you build it yourself?
I need both the ONNX backend and VLLM backend, so I built a custom Triton image with the following Dockerfile commands:

FROM repo.yxuer.com:30100/nvidia/tritonserver:25.10-py3 AS base  
FROM repo.yxuer.com:30100/nvidia/tritonserver:25.10-vllm-python-py3  

# Copy ONNX Runtime and TensorRT backends from the base image  
COPY --from=base /opt/tritonserver/backends/onnxruntime /opt/tritonserver/backends/onnxruntime  
COPY --from=base /opt/tritonserver/backends/tensorrt /opt/tritonserver/backends/tensorrt  

[yinggeh]

@zjq1996518 Can you provide a simple reproducer so that our engineers can investigate?

[zjq1996518]

my base model is https://github.com/triton-inference-server/vllm_backend/blob/main/src/model.py
weights: https://huggingface.co/opendatalab/MinerU2.5-2509-1.2B

model.py (mineru)

import os
import re
import sys
from dataclasses import dataclass
from time import time

import numpy as np
from PIL import Image


import json
from transformers import AutoTokenizer
# https://github.com/triton-inference-server/vllm_backend/blob/main/src/model.py
# rename TritonPythonModel to VllmModel
from packages.entity.vllm_model import VllmModel 
import triton_python_backend_utils as pb_utils


@dataclass
class MinerUSamplingParams:
    temperature: float | None = 0.0
    top_p: float | None = 0.01
    top_k: int | None = 1
    presence_penalty: float | None = 0.0
    frequency_penalty: float | None = 0.0
    repetition_penalty: float | None = 1.0
    no_repeat_ngram_size: int | None = 100
    max_new_tokens: int | None = None


class BlockType:
    TEXT = "text"  # 文本
    TITLE = "title"  # 段落标题
    TABLE = "table"  # 表格
    IMAGE = "image"  # 图像
    CODE = "code"  # 代码
    ALGORITHM = "algorithm"  # 算法/伪代码
    HEADER = "header"  # 页眉
    FOOTER = "footer"  # 页脚
    PAGE_NUMBER = "page_number"  # 页码
    PAGE_FOOTNOTE = "page_footnote"  # 脚注
    ASIDE_TEXT = "aside_text"  # 侧栏文本(装订线等)
    EQUATION = "equation"  # 公式(独立公式)
    EQUATION_BLOCK = "equation_block"  # 公式块(多行公式)
    REF_TEXT = "ref_text"  # 参考文献(一条)
    LIST = "list"  # 列表块(无序/有序列表)
    PHONETIC = "phonetic"  # 注音符号

    # captions
    TABLE_CAPTION = "table_caption"  # 表格标题
    IMAGE_CAPTION = "image_caption"  # 图像标题
    CODE_CAPTION = "code_caption"  # 代码标题
    TABLE_FOOTNOTE = "table_footnote"  # 表格脚注
    IMAGE_FOOTNOTE = "image_footnote"  # 图像脚注

    UNKNOWN = "unknown"  # 未知块


BLOCK_TYPES = {
    BlockType.TEXT,
    BlockType.TITLE,
    BlockType.TABLE,
    BlockType.IMAGE,
    BlockType.CODE,
    BlockType.HEADER,
    BlockType.FOOTER,
    BlockType.PAGE_NUMBER,
    BlockType.PAGE_FOOTNOTE,
    BlockType.ASIDE_TEXT,
    BlockType.EQUATION,
    BlockType.EQUATION_BLOCK,
    BlockType.REF_TEXT,
    BlockType.TABLE_CAPTION,
    BlockType.IMAGE_CAPTION,
    BlockType.TABLE_FOOTNOTE,
    BlockType.IMAGE_FOOTNOTE,
    BlockType.ALGORITHM,
    BlockType.CODE_CAPTION,
    BlockType.LIST,
    BlockType.PHONETIC,
    BlockType.UNKNOWN
}

ANGLE_OPTIONS = {None, 0, 90, 180, 270}


class TritonPythonModel(VllmModel):

    def __init__(self):
        super().__init__()
        self.image_size = (1036, 1036)
        self.model_max_length = 16384

        self.prompts = {
            "table": "\nTable Recognition:",
            "equation": "\nFormula Recognition:",
            "[default]": "\nText Recognition:",
            "[layout]": "\nLayout Detection:",
        }
        self.sampling_params = {
            "table": MinerUSamplingParams(presence_penalty=1.0, frequency_penalty=0.005),
            "equation": MinerUSamplingParams(presence_penalty=1.0, frequency_penalty=0.05),
            "[default]": MinerUSamplingParams(presence_penalty=1.0, frequency_penalty=0.05),
            "[layout]": MinerUSamplingParams(),
        }

        self.angle_mapping = {
            "<|rotate_up|>": 0,
            "<|rotate_right|>": 90,
            "<|rotate_down|>": 180,
            "<|rotate_left|>": 270,
        }

        self._layout_re = r"^<\|box_start\|>(\d+)\s+(\d+)\s+(\d+)\s+(\d+)<\|box_end\|><\|ref_start\|>(\w+?)<\|ref_end\|>(.*)$"

    def initialize(self, args):
        super().initialize(args)
        model_path = self.vllm_engine_config['model']
        self.tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
        self.logger.info("MinerU2.5 model and tokenizer initialized.")

    async def get_img_params(self, request):
            origin_img = pb_utils.get_input_tensor_by_name(request, 'img')
            if origin_img is not None:
                origin_img = origin_img.as_numpy()
        if origin_img is None:
            raise RuntimeError()
        if origin_img.size == 0:
            raise RuntimeError()
        return origin_img

    def build_messages(self, prompt: str) -> list[dict]:
        messages = [{"role": "system", "content": 'You are a helpful assistant.'}]
        user_messages = [
            {"type": "image"},
            {"type": "text", "text": prompt},
        ]
        messages.append({"role": "user", "content": user_messages})
        return messages

    def extra_sampling_parameters(self, param_dict, request=None):
        task = pb_utils.get_input_tensor_by_name(request, "task").as_numpy()[0].decode("utf-8")
        match task:
            case 'ocr':
                sampling_param = self.sampling_params.get('[default]')
            case 'layout':
                sampling_param = self.sampling_params.get('[layout]')
            case 'formula_ocr':
                sampling_param = self.sampling_params.get('equation')
            case 'table_ocr':
                sampling_param = self.sampling_params.get('table')
            case _:
                sampling_param = self.sampling_params.get('[default]')

        sampling_parameters = {
            "temperature": sampling_param.temperature,
            "top_p": sampling_param.top_p,
            "top_k": sampling_param.top_k,
            "presence_penalty": sampling_param.presence_penalty,
            "frequency_penalty": sampling_param.frequency_penalty,
            "repetition_penalty": sampling_param.repetition_penalty,
            # max_tokens should smaller than model max length
            "max_tokens": 16384,
            "skip_special_tokens": False,
            "extra_args": {
                "no_repeat_ngram_size": sampling_param.no_repeat_ngram_size
            }
        }

        return sampling_parameters

    def get_prompt_by_task(self, task, img):
        match task:
            case 'ocr':
                prompt = self.prompts.get('[default]')
            case 'layout':
                img = Image.fromarray(img).resize(self.image_size, Image.Resampling.BICUBIC)
                prompt = self.prompts.get('[layout]')
            case 'formula_ocr':
                prompt = self.prompts.get('equation')
            case 'table_ocr':
                prompt = self.prompts.get('table')
            case _:
                prompt = self.prompts.get('[default]')

        prompt = self.tokenizer.apply_chat_template(
            self.build_messages(prompt),
            tokenize=False,
            add_generation_prompt=True)
        vllm_prompts = {"prompt": prompt, "multi_modal_data": {"image": img}}

        return vllm_prompts


    async def get_prompt(self, request, **kwargs):
        task = pb_utils.get_input_tensor_by_name(request, "task").as_numpy()[0].decode("utf-8")
        img = await self.get_img_params(request)
        vllm_prompts = self.get_prompt_by_task(task, img)
        return vllm_prompts

    async def execute(self, requests, **kwargs):
        return super().execute(requests)

model.json

{
    "model": "./weights",
    "trust_remote_code": true,
    "gpu_memory_utilization": 0.75,
    "max_model_len": 8192,
    "enable_prefix_caching": false,
    "max_num_batched_tokens": 16384,
    "enable_chunked_prefill": true
}

config.pbtxt

name: "minerU"
backend: "python"
default_model_filename: "model.py"
max_batch_size: 0

input [
  {
    name: "task"
    data_type: TYPE_STRING
    dims: [1]
  },
  {
    name: "img"
    data_type: TYPE_UINT8
    dims: [-1, -1, -1]
    optional: true
  },
  {
    name: "stream"
    data_type: TYPE_BOOL
    dims: [ 1 ]
    optional: true
  },
  {
    name: "sampling_parameters"
    data_type: TYPE_STRING
    dims: [ 1 ]
    optional: true
  },
  {
    name: "exclude_input_in_output"
    data_type: TYPE_BOOL
    dims: [ 1 ]
    optional: true
  }
]

instance_group [
  {
    count: 1
    kind: KIND_GPU
  }
]

model_transaction_policy {
  decoupled: True
}

[yinggeh]

@zjq1996518 I could not reproduce in the latest 25.12 container. Can you give it a try and let me know the result?
The prompt I used
curl -X POST localhost:8000/v2/models/minerU/generate -d '{"task": "What is Triton Inference Server?", "parameters": {"stream": false, "temperature": 0}}'

[zjq1996518]

@zjq1996518 I could not reproduce in the latest 25.12 container. Can you give it a try and let me know the result? The prompt I used curl -X POST localhost:8000/v2/models/minerU/generate -d '{"task": "What is Triton Inference Server?", "parameters": {"stream": false, "temperature": 0}}'

This is an OCR model. You need to pass an image with the following parameters:
{"task": "ocr", "img": "xxx"}
where img is a three-channel image with a shape of [h, w, 3].

[yinggeh]

@zjq1996518 Can you also provide a sample client for me to reproduce the error your reported?

[zjq1996518]

@yinggeh
Okay, I am using triton_client 2.62.0 and referenced your early client code. I just reviewed it again, and the current code seems to be different from what I referenced at that time.

Since our project uses etcd to implement client-side load balancing on its own, providing the complete client code would be quite complex. I can only provide some core parts of the code.

I am using the async_stream_infer method of InferenceServerClient in triton_client.grpc.
Because this method does not natively support async capabilities, I made some modifications: when there are results in the response stream, they are placed into an asyncio queue to ensure the caller can retrieve results using await.

My apologies for forgetting to provide the _types definition. I've made a few minor modifications to the package_inputs method.

_types = {
    'str': (object, 'BYTES'),
    'uint8': (np.uint8, 'UINT8'),
    'fp16': (np.float16, 'FP16'),
    'fp32': (np.float32, 'FP32'),
    'fp64': (np.float64, 'FP64'),
    'int32': (np.int32, 'INT32'),
    'int64': (np.int64, 'INT64'),
    'uint16': (np.uint16, 'UINT16'),
    'bool': (bool, 'BOOL')
}

def package_inputs(inputs, mode='grpc'):
    packages = []
    numpy_data = []
    for name, param, _type in inputs:
        if isinstance(param, np.ndarray):
            data = param
        elif isinstance(param, list) or isinstance(param, tuple):
            data = np.array(param, dtype=_types[_type][0])
        else:
            data = np.array([param], dtype=_types[_type][0])
        numpy_data.append(data)
    for i, (name, _, _type) in enumerate(inputs):
        param = numpy_data[i]
        InferInput = httpclient.InferInput if mode == 'http' else grpcclient.InferInput
        _input = InferInput(name, list(param.shape), _types[_type][1])
        _input.set_data_from_numpy(param)
        packages.append(_input)
    return packages

def package_outputs(outputs, mode='grpc'):
    return [httpclient.InferRequestedOutput(output) if mode == 'http' else grpcclient.InferRequestedOutput(output)
            for output in outputs]

class AsyncDecoupledClient:

    def __init__(self):
        self.queue = None
        self.event_loop = None
        self.decoupled_sort_reqs = {}
        self.req_count = 0

    async def __aenter__(self):
        self.queue = asyncio.Queue()
        self.event_loop = asyncio.get_running_loop()
        # This method uses etcd for load balancing + connection pool, so it is not provided here.
        self._triton_clients = await self.get_triton_clients()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # This method uses etcd for load balancing + connection pool, so it is not provided here.
        self.release_triton_clients()

    def decoupled_callback(self, host, result, error):
        asyncio.run_coroutine_threadsafe(self._decoupled_callback(host, result, error), self.event_loop)

    async def _decoupled_callback(self, host, result, error):
        if error is not None:
            await self.queue.put((host, error))
            return
        await self.queue.put((host, result))

    async def infer(
            self,
            model_name,
            inputs,
            outputs,
            mode='grpc',
            model_version='',
            vllm_request_num=1,
            compression_algorithm=None,
            vllm_stream=False,
            *args,
            **kwargs
    ):
        # This method uses etcd for load balancing + connection pool, so it is not provided here.
        # You can directly initialize with InferenceServerClient. 
        # The 'host' parameter is only used to record which node the request was sent to,
        # so any string value is acceptable.
        triton_client, host = await self.get_triton_client(model_name, mode, *args, **kwargs)
        # Create receive stream
        if triton_client._stream is None:
            triton_client.start_stream(callback=partial(self.decoupled_callback, host),
                                            compression_algorithm=compression_algorithm)

        request_id = str(uuid.uuid1())
        # Record the sequence number of the current request for subsequent sorting.
        # It doesn't matter if vllm_request_num is not 1, it won't affect sorting.
        self.decoupled_sort_reqs[request_id] = self.req_count
        self.req_count += vllm_request_num

        triton_client.async_stream_infer(model_name,
                                              package_inputs(inputs, mode=mode),
                                              model_version=model_version,
                                              request_id=request_id,
                                              outputs=package_outputs(outputs, mode=mode),
                                              enable_empty_final_response=True)
        return partial(self.decoupled_wait, outputs)

    @staticmethod
    def decoupled_process(data_item, vllm_stream, result_dict, outputs):
        response = data_item.get_response()
        if response.outputs:
            if result_dict.get(response.id) is None:
                result_dict[response.id] = []
            result_dict[response.id].append(result2dict(data_item, outputs))

        # For streaming mode: consider the request complete only when the final response is received.
        # For non-streaming mode: consider the request complete as long as one response is returned.
        return (response.parameters.get("triton_final_response").bool_param and vllm_stream) or not vllm_stream

    def decoupled_last_process(self, result_dict):
        # Sort the results
        sorted_result = []
        for key, sort_index in self.decoupled_sort_reqs.items():
            sorted_result.append(result_dict[key])
        self.req_count = 0
        self.decoupled_sort_reqs = {}
        return sorted_result

    async def decoupled_wait(self, outputs, timeout=600, vllm_stream=False):
        result_dict = {}
        completed_requests = 0
        while completed_requests < self.req_count:
            host, data_item = await asyncio.wait_for(self.queue.get(), timeout=timeout)
            if type(data_item) == InferenceServerException:
                raise data_item
            if data_item is None:
                raise RuntimeError('Server communication timeout')
            if self.decoupled_process(data_item, vllm_stream, result_dict, outputs):
                completed_requests += 1
        return self.decoupled_last_process(result_dict)

The above is the core code. After sending the request, to retrieve results:

def minerU(task, img, exclude_input_in_output=False, additional_outputs=None,
           decoupled_client: AsyncDecoupledClient=None, *args, **kwargs):
    """
    @param decoupled_client: Instance of AsyncDecoupledClient
    @param task: (str) Task type: ocr, layout, formula_ocr, table_ocr
    @param additional_outputs: Additional output fields to return
    @return: A callable wait function that blocks and returns the final result when invoked.
    """
    inputs = [
        ('task', task, 'str'),
        ('img', img, 'uint8'),
        ('stream', False, 'bool'),
        ('exclude_input_in_output', exclude_input_in_output, 'bool'),
    ]
    outputs = ['text_output']

    return decoupled_client.infer(
        'minerU',
        inputs=inputs,
        outputs=outputs,
        # Whether to enable streaming output
        vllm_stream=False,
        *args,
        **kwargs)

async with AsyncDecoupledClient() as decoupled_client: 
    # Use any uint8-type numpy array for the image
    for cropped in crops:
        wait = await minerU('ocr', img=cropped, decoupled_client=decoupled_client,
                            compression_algorithm='deflate')
        s = time()
        all_time += time() - s
    avg_time = all_time / len(crops)
    rst = await wait()

[yinggeh]

@zjq1996518 I am still unable to reproduce the issue letting the client run for minutes on Triton 25.10 version. Below is my client implementation

import asyncio
import numpy as np
import uuid
from functools import partial
from time import time
import tritonclient.grpc as grpcclient
import tritonclient.http as httpclient
from tritonclient.utils import InferenceServerException

_types = {
    'str': (object, 'BYTES'),
    'uint8': (np.uint8, 'UINT8'),
    'fp16': (np.float16, 'FP16'),
    'fp32': (np.float32, 'FP32'),
    'fp64': (np.float64, 'FP64'),
    'int32': (np.int32, 'INT32'),
    'int64': (np.int64, 'INT64'),
    'uint16': (np.uint16, 'UINT16'),
    'bool': (bool, 'BOOL')
}

def package_inputs(inputs, mode='grpc'):
    packages = []
    numpy_data = []
    for name, param, _type in inputs:
        if isinstance(param, np.ndarray):
            data = param
        elif isinstance(param, list) or isinstance(param, tuple):
            data = np.array(param, dtype=_types[_type][0])
        else:
            data = np.array([param], dtype=_types[_type][0])
        numpy_data.append(data)
    for i, (name, _, _type) in enumerate(inputs):
        param = numpy_data[i]
        InferInput = httpclient.InferInput if mode == 'http' else grpcclient.InferInput
        _input = InferInput(name, list(param.shape), _types[_type][1])
        _input.set_data_from_numpy(param)
        packages.append(_input)
    return packages

def package_outputs(outputs, mode='grpc'):
    return [httpclient.InferRequestedOutput(output) if mode == 'http' else grpcclient.InferRequestedOutput(output)
            for output in outputs]

def result2dict(data_item, outputs):
    """
    Convert Triton InferResult to dictionary format.
    
    Args:
        data_item: InferResult object from Triton client
        outputs: List of output names
    
    Returns:
        Dictionary mapping output names to numpy arrays
    """
    result_dict = {}
    for output_name in outputs:
        try:
            output_data = data_item.as_numpy(output_name)
            result_dict[output_name] = output_data
        except Exception as e:
            # Handle cases where output might not be present
            result_dict[output_name] = None
    return result_dict

class AsyncDecoupledClient:

    def __init__(self):
        self.queue = None
        self.event_loop = None
        self.decoupled_sort_reqs = {}
        self.req_count = 0

    async def __aenter__(self):
        self.queue = asyncio.Queue()
        self.event_loop = asyncio.get_running_loop()
        # This method uses etcd for load balancing + connection pool, so it is not provided here.
        self._triton_clients = await self.get_triton_clients()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # This method uses etcd for load balancing + connection pool, so it is not provided here.
        self.release_triton_clients()

    async def get_triton_clients(self):
        """
        Get Triton clients for connection pooling/load balancing.
        This is a simplified version - replace with your etcd-based implementation.
        
        Returns:
            List of InferenceServerClient instances
        """
        # Simple implementation: return a single client
        # For production, implement etcd-based service discovery and load balancing here
        # Example: query etcd for available Triton server endpoints, create clients for each
        
        # Default to localhost:8001 for GRPC, can be extended to support multiple endpoints
        url = "localhost:8001"
        mode = "grpc"  # Default mode, can be made configurable
        
        if mode == 'grpc':
            client = grpcclient.InferenceServerClient(url=url, verbose=False)
        else:
            client = httpclient.InferenceServerClient(url=url, verbose=False)
        
        return [client]

    def release_triton_clients(self):
        """Release Triton clients (cleanup)."""
        if hasattr(self, '_triton_clients') and self._triton_clients:
            for client in self._triton_clients:
                if hasattr(client, 'close'):
                    client.close()
            self._triton_clients = []

    async def get_triton_client(self, model_name, mode='grpc', *args, **kwargs):
        """
        Get a Triton client for a specific model.
        This is a simplified version - replace with your etcd-based implementation.
        
        Args:
            model_name: Name of the model (not used in simple implementation)
            mode: 'grpc' or 'http'
            *args, **kwargs: Additional arguments
        
        Returns:
            Tuple of (client, host_string)
        """
        # This method uses etcd for load balancing + connection pool, so it is not provided here.
        # You can directly initialize with InferenceServerClient. 
        # The 'host' parameter is only used to record which node the request was sent to,
        # so any string value is acceptable.
        
        # Simple implementation: use the first client from the pool
        # For production, implement load balancing logic here (round-robin, least connections, etc.)
        if hasattr(self, '_triton_clients') and self._triton_clients:
            client = self._triton_clients[0]
            host = "localhost:8001"  # Default host identifier
        else:
            # Fallback: create a new client if pool is empty
            url = "localhost:8001"
            if mode == 'grpc':
                client = grpcclient.InferenceServerClient(url=url, verbose=False)
            else:
                client = httpclient.InferenceServerClient(url=url, verbose=False)
            host = url
        
        return client, host

    def decoupled_callback(self, host, result, error):
        asyncio.run_coroutine_threadsafe(self._decoupled_callback(host, result, error), self.event_loop)

    async def _decoupled_callback(self, host, result, error):
        if error is not None:
            await self.queue.put((host, error))
            return
        await self.queue.put((host, result))

    async def infer(
            self,
            model_name,
            inputs,
            outputs,
            mode='grpc',
            model_version='',
            vllm_request_num=1,
            compression_algorithm=None,
            vllm_stream=False,
            *args,
            **kwargs
    ):
        # This method uses etcd for load balancing + connection pool, so it is not provided here.
        # You can directly initialize with InferenceServerClient. 
        # The 'host' parameter is only used to record which node the request was sent to,
        # so any string value is acceptable.
        triton_client, host = await self.get_triton_client(model_name, mode, *args, **kwargs)
        # Create receive stream
        if triton_client._stream is None:
            triton_client.start_stream(callback=partial(self.decoupled_callback, host),
                                            compression_algorithm=compression_algorithm)

        request_id = str(uuid.uuid1())
        # Record the sequence number of the current request for subsequent sorting.
        # It doesn't matter if vllm_request_num is not 1, it won't affect sorting.
        self.decoupled_sort_reqs[request_id] = self.req_count
        self.req_count += vllm_request_num

        triton_client.async_stream_infer(model_name,
                                              package_inputs(inputs, mode=mode),
                                              model_version=model_version,
                                              request_id=request_id,
                                              outputs=package_outputs(outputs, mode=mode),
                                              enable_empty_final_response=True)
        return partial(self.decoupled_wait, outputs)

    @staticmethod
    def decoupled_process(data_item, vllm_stream, result_dict, outputs):
        response = data_item.get_response()
        if response.outputs:
            if result_dict.get(response.id) is None:
                result_dict[response.id] = []
            result_dict[response.id].append(result2dict(data_item, outputs))

        # For streaming mode: consider the request complete only when the final response is received.
        # For non-streaming mode: consider the request complete as long as one response is returned.
        return (response.parameters.get("triton_final_response").bool_param and vllm_stream) or not vllm_stream

    def decoupled_last_process(self, result_dict):
        # Sort the results
        sorted_result = []
        for key, sort_index in self.decoupled_sort_reqs.items():
            sorted_result.append(result_dict[key])
        self.req_count = 0
        self.decoupled_sort_reqs = {}
        return sorted_result

    async def decoupled_wait(self, outputs, timeout=600, vllm_stream=False):
        result_dict = {}
        completed_requests = 0
        while completed_requests < self.req_count:
            host, data_item = await asyncio.wait_for(self.queue.get(), timeout=timeout)
            if type(data_item) == InferenceServerException:
                raise data_item
            if data_item is None:
                raise RuntimeError('Server communication timeout')
            if self.decoupled_process(data_item, vllm_stream, result_dict, outputs):
                completed_requests += 1
        return self.decoupled_last_process(result_dict)

def minerU(task, img, exclude_input_in_output=False, additional_outputs=None,
           decoupled_client: AsyncDecoupledClient=None, *args, **kwargs):
    """
    @param decoupled_client: Instance of AsyncDecoupledClient
    @param task: (str) Task type: ocr, layout, formula_ocr, table_ocr
    @param additional_outputs: Additional output fields to return
    @return: A callable wait function that blocks and returns the final result when invoked.
    """
    inputs = [
        ('task', task, 'str'),
        ('img', img, 'uint8'),
        ('stream', False, 'bool'),
        ('exclude_input_in_output', exclude_input_in_output, 'bool'),
    ]
    outputs = ['text_output']

    return decoupled_client.infer(
        'minerU',
        inputs=inputs,
        outputs=outputs,
        # Whether to enable streaming output
        vllm_stream=False,
        *args,
        **kwargs)

async def main(crops):
    """
    Example async function demonstrating how to use the AsyncDecoupledClient.
    
    Args:
        crops: List of cropped images (uint8 numpy arrays)
    
    Returns:
        Tuple of (results, average_time)
    """
    all_time = 0.0
    
    async with AsyncDecoupledClient() as decoupled_client: 
        # Use any uint8-type numpy array for the image
        for cropped in crops:
            wait = await minerU('ocr', img=cropped, decoupled_client=decoupled_client,
                                compression_algorithm='deflate')
            s = time()
            all_time += time() - s
        avg_time = all_time / len(crops)
        rst = await wait()


if __name__ == "__main__":
    # Create dummy image byte arrays (uint8 numpy arrays)
    # Simulating cropped images - you can replace these with actual image data
    dummy_img1 = np.random.randint(0, 255, size=(1036, 1036, 3), dtype=np.uint8)
    dummy_img2 = np.random.randint(0, 255, size=(1036, 1036, 3), dtype=np.uint8)
    dummy_img3 = np.random.randint(0, 255, size=(1036, 1036, 3), dtype=np.uint8)
    
    crops = [dummy_img1, dummy_img2, dummy_img3]
    
    # Run the async main function
    asyncio.run(main(crops))

[yinggeh]

Close due to inactivity. Please re-open the issue if there is further questions.

[zjq1996518]

@yinggeh I tried running your code and was not able to reproduce the error initially. However, when I modified the images to use random values generated with the following code snippet, the error I mentioned above occurred:

if __name__ == "__main__":
    # Create dummy image byte arrays (uint8 numpy arrays)
    # Simulating cropped images - you can replace these with actual image data
    import random

    crops = [np.random.randint(0, 255, size=(random.randint(20,256), random.randint(20,256), 3), dtype=np.uint8) for _ in  range(50)]

    # Run the async main function
    asyncio.run(main(crops))

It seems this error might be caused by the randomness of certain image dimensions? Unfortunately, I cannot debug this issue effectively because the error message is not descriptive enough.

[yinggeh]

@yinggeh I tried running your code and was not able to reproduce the error initially. However, when I modified the images to use random values generated with the following code snippet, the error I mentioned above occurred:

if name == "main":
# Create dummy image byte arrays (uint8 numpy arrays)
# Simulating cropped images - you can replace these with actual image data
import random

crops = [np.random.randint(0, 255, size=(random.randint(20,256), random.randint(20,256), 3), dtype=np.uint8) for _ in  range(50)]

# Run the async main function
asyncio.run(main(crops))

It seems this error might be caused by the randomness of certain image dimensions? Unfortunately, I cannot debug this issue effectively because the error message is not descriptive enough.

Thanks. I am now able to reproduce the error. I will investigate this backend behavior and meanwhile please stay patient.