关键词：Triton

前言

在前文《AI模型部署：一文搞定Triton Inference Server的常用基础配置和功能特性》中介绍了Triton Inference Server的基础配置，包括输入输出、模型和版本管理、前后预处理等，本篇介绍在推理阶段常用的配置，包括多实例并发、动态批处理、模型预热，这些是Triton的核心特性。本篇以Python作为Triton的后端，和其他后端的设置有特殊和不同，公开资料较少，建议收藏。

内容摘要

• 执行实例设置
• 并发请求测试
• 模型预热
• 请求合并动态批处理

执行实例设置和并发

Triton通过config.pbtxt中的instance_group来设置模型执行的实例，包括实例数量，CPU/GPU设备资源。如果在config.pbtxt中不指定instance_group，默认情况下Triton会给当前环境下所有可得的每个GPU设置一个执行实例。
在docker run启动命名中指定--gpus参数，将gpu设备添加到容器中，all代表将所有gpu设备都添加进去

docker run --gpus=all \
--rm -p18999:8000 -p18998:8001 -p18997:8002 \
-v /home/model_repository/:/models \
nvcr.io/nvidia/tritonserver:21.02-py3 \
tritonserver \
--model-repository=/models

观察Triton的启动日志，一共2个模型string和string_batch，在3个gpu（0，1，2）上分别分配了一个执行实例，相当于每个模型有3个gpu执行实例，对应后台Triton会启动3个子进程

...
I0328 06:42:26.406186 1 python.cc:615] TRITONBACKEND_ModelInstanceInitialize: string_batch (GPU device 0)
I0328 06:42:26.504449 1 python.cc:615] TRITONBACKEND_ModelInstanceInitialize: string (GPU device 0)
I0328 06:42:34.868080 1 python.cc:615] TRITONBACKEND_ModelInstanceInitialize: string (GPU device 1)
I0328 06:42:34.874191 1 python.cc:615] TRITONBACKEND_ModelInstanceInitialize: string_batch (GPU device 1)
I0328 06:42:40.886786 1 python.cc:615] TRITONBACKEND_ModelInstanceInitialize: string (GPU device 2)
I0328 06:42:40.887770 1 python.cc:615] TRITONBACKEND_ModelInstanceInitialize: string_batch (GPU device 2)
...

如果不添加gpu设备到容器，在docker run中删除--gpus参数，此时Triton只能使用cpu作为计算设备

docker run  \
--rm -p18999:8000 -p18998:8001 -p18997:8002 \
-v /home/model_repository/:/models \
nvcr.io/nvidia/tritonserver:21.02-py3 \
tritonserver \
--model-repository=/models

Triton启动日志如下，每个模型在cpu下启动了一个实例

I0328 06:51:17.795794 1 python.cc:615] TRITONBACKEND_ModelInstanceInitialize: string_batch (CPU device 0)
I0328 06:51:17.897220 1 python.cc:615] TRITONBACKEND_ModelInstanceInitialize: string (CPU device 0)

默认的每个gpu分配一个执行实例的效果等同于在config.txtpb中设置如下

  instance_group [
    {
      count: 1
      kind: KIND_GPU
      gpus: [ 0 ]
    },
    {
      count: 1
      kind: KIND_GPU
      gpus: [ 1 ]
    },
    {
      count: 1
      kind: KIND_GPU
      gpus: [ 2 ]
    }
  ]

或者

  instance_group [
    {
      count: 1
      kind: KIND_GPU
      gpus: [ 0, 1, 2 ]
    }
  ]

其中count表示每个设备下执行实例数量，kind代表计算设备，KIND_GPU代表gpu设备，gpus指定gpu编号，同样如果设置计算设备为cpu，即使容器中有gpu设备也会按照cpu类运行，设置cpu两个执行实例如下

 instance_group [
    {
      count: 2
      kind: KIND_CPU
    }
  ]

对于以Python为后端的情况，尽管在Triton服务端已经申明了GPU设备，还是需要在model.py脚本层再显式申明一次，将模型和数据加载到指定GPU设备上，否则Python后端会自动将所有实例加载在GPU:0上。具体操作方法是在model.py的初始化阶段通过model_instance_kind，model_instance_device_id参数拿到config.pbtxt中指定的设备，在model.py中获取设备信息代码样例如下

def initialize(self, args):
  device = "cuda" if args["model_instance_kind"] == "GPU" else "cpu"
  device_id = args["model_instance_device_id"]
  self.device = f"{device}:{device_id}"
  self.model = BertForSequenceClassification.from_pretrained(model_path).to(self.device).eval()


def execute(self, requests):
  encoding = self.tokenizer.batch_encode_plus(
                text,
                max_length=512,
                add_special_tokens=True,
                return_token_type_ids=False,
                padding=True,
                return_attention_mask=True,
                return_tensors='pt',
                truncation=True
            ).to(self.device)

并发请求测试

Triton启动的多个模型实例可以并行的处理请求，Trion会自动分配请求到空闲的执行实例，从而加快推理服务的处理速度，提高GPU的利用率。

多个实例并行执行示意图

根据上一节所交代的设置，分别在config.pbtxt中设置对应的kind，gpus，count参数，通过不同的设备和执行实力数来测试Triton服务对一个Bert-Base微调的情感分类模型的推理性能，设置如下，最多三块GPU设备，每个设备最多3个实例

在客户端使用Python线程池设置20个并发500个请求任务，每个任务的batch_size为64，即64个句子的情感推理，请求代码如下

import re
import time
import json
import requests
from concurrent.futures import ThreadPoolExecutor

import torch.nn


def handle(sid):
    print("---------start:{}".format(sid))
    text = ["句子1...", "句子2...", "句子3...", "句子4..."] * 16
    url = "http://0.0.0.0:18999/v2/models/sentiment/infer"
    raw_data = {
        "inputs": [
            {
                "name": "text",
                "datatype": "BYTES",
                "shape": [64, 1],
                "data": text
            }
        ],
        "outputs": [
            {
                "name": "prob",
                "shape": [64, -1],
            }
        ]
    }
    res = requests.post(url, json.dumps(raw_data, ensure_ascii=True), headers={"Content_Type": "application/json"},
                        timeout=2000)

    print(str(sid) + ",".join(json.loads(res.text)["outputs"][0]["data"]))


if __name__ == '__main__':
    n = 500
    sid_list = list(range(n))
    t1 = time.time()
    with ThreadPoolExecutor(20) as executor:
        for i in sid_list:
            executor.submit(handle, i)
    t2 = time.time()
    print("耗时:", (t2 - t1) / n)

计算平均响应耗时如下图所示

情感分类Bert模型的推理平均响应时间

其中GPU设备推理性能自然远强于CPU，在本例中至少是50倍的差距。GPU设备数量和推理性能呈现出倍数的关系，部署的GPU越多，性能越强。在实例数量相同的时候，多GPU单实例部署比单GPU多实例部署性能更高，考虑到是单GPU负载过高导致性能下降，同样的在GPU数相同的情况下，每块GPU 3个实例性能反而还略低于2个实例。

模型预热

一般的，服务后端需要先对模型进行加载和初始化，在某些情况下会存在延迟初始化，直到后端接受到第一条或者少量的推理请求，这导致服务端在推理第一批的请求时耗时异常高。Triton在config.txtpb配置中设置了模型预热参数model_warmup，使得在正式提供推理服务之前模型能够完全初始化。
以上一节的自然语言情感分类为例，我们给模型的每一个实例设置预热如下

input [ 
    {
        name: "text"
        dims: [ -1 ]
        data_type: TYPE_STRING
    }
]
output [...]
instance_group [
  {
  count: 1
  kind: KIND_GPU
  gpus: [ 0, 1 ]
  }
]
model_warmup  [
  {
    name: "random_input"
    batch_size: 1
    inputs: {
      key: "text"
      value: {
        data_type: TYPE_STRING
        dims: [ 1 ]
        input_data_file: "raw_data"
          }
       }
   }
]

预热的本质是提前给到一组数据，让模型在加载初始化之后对这组数据进行推理，从而完成完整的模型初始化步骤，在model_warmup的inputs指定了预设数据的信息，其中key要和input的name对应，data_type和input的data_type对应，dims必须是确定的维度，不能为-1，input_data_file约定了预设的数据在一个路径文件中，Triton会去容器中的/models/model_name/warmup/input_data_file下拿到这个文件的数据，映射到宿主机上该位置在和config.pbtxt同一级目录下，input_data_file的位置如下

.
├── 1
│   ├── model.py
│   ├── sentiment
│   │   ├── config.json
│   │   ├── pytorch_model.bin
│   │   ├── special_tokens_map.json
│   │   ├── tokenizer_config.json
│   │   └── vocab.txt
├── config.pbtxt
└── warmup
    └── raw_data

input_data_file内容为自定义构造的预设数据，对于字符串输入使用tritonclient客户端进行构造，将字符串转化为输入需要的字节形式，例如将“我爱你美丽的中国”改造为预设数据输入

# pip install tritonclient
import numpy as np
from tritonclient.utils import serialize_byte_tensor

serialized = serialize_byte_tensor(
    np.array(["我爱你美丽的中国".encode("utf-8")], dtype=object)
)
with open("raw_data", "wb") as fh:
    fh.write(serialized.item())

config.pbtxt和预设的样例数据文件./warmup/raw_data设置完毕后，启动Triton服务，日志如下

2024-04-02 06:13:53,199 - model.py[line:107] - INFO: {'text': ['我爱你美丽的中国']}
2024-04-02 06:13:54,310 - model.py[line:129] - INFO: {'prob': array([b'\xe6\xad\xa3\xe5\x90\x91'], dtype='|S6')}
2024-04-02 06:13:54,312 - model.py[line:107] - INFO: {'text': ['我爱你美丽的中国']}
2024-04-02 06:13:55,319 - model.py[line:129] - INFO: {'prob': array([b'\xe6\xad\xa3\xe5\x90\x91'], dtype='|S6')}

日志显示在每个执行实例上都将样例数据输送给模型推理了一次，同样的我们测试在加入模型预热之后模型的推理性能，和上一节的测试结果做对比

在本例中添加预热的模式平均响应耗时比不添加预热下降了约1～2毫秒，降低了模型初次推理导致的性能损失。
本例以input_data_file引入外部数据文件的形式构造预设数据，除此之外更简单的一种是在model_warmup直接指定random_data，按照指定的维度和数据类型生成一组随机数，model_warmup设置如下

model_warmup  [
  {
    name: "random_input"
    batch_size: 1
    inputs: {
      key: "x"
      value: {
        data_type: TYPE_FP32
        dims: [ 3 ]
        random_data: true
          }
       }
   }
]

该例子表示随机生成一个[1, 3]的浮点数向量，输入给模型进行预热。

请求合并动态批处理

批量推理可以提高推理吞吐量和GPU的利用率，动态批处理指的是Triton服务端会组合推理请求，从而动态创建批处理来高吞吐量，这块内容由Triton的调度策略决定。
默认的调度策略Default Scheduler不会主动合并请求，而是仅将所有请求分发到各个执行实例上，动态批处理策略Dynamic Batcher它可以在服务端将多个batch_size较小的请求组合在一起形成一个batch_size较大的任务，从而提高吞吐量和GPU利用率，Dynamic Batcher在config.pbtxt中进行指定，一个例子如下

max_batch_size: 16
dynamic_batching {
    preferred_batch_size: [ 4, 8 ]
    max_queue_delay_microseconds: 100
}

• preferred_batch_size：期望达到的batch_size，可以指定一个数值，也可以是一个包含多个值的数组，本例代表期望组合成大小为4或者8的batch_size，尽可能的将batch_size组为指定的值，batch_size不能超过max_batch_size
• max_queue_delay_microseconds：组合batch的最大时间限制，单位为微秒，本例代表组合batch最长时间为100微秒，超过这个时间则停止组合batch，会把已经打进batch的请求进行推理。这个时间限制越大，延迟越大，但是越容易组合到大的batch_size，这个时间限制越小。延迟越小，但是吞吐量降低，因此该参数是一个延迟和吞吐之间的trade off

配置文件中的dynamic_batching只是将请求进行聚合，在model.py的自定义后端中实际的作用是requests变成了多个request组成的集合，而不开启dynamic_batching则requests里面只有一个request，这一点在execute方法的注释中有说明 **“Depending on the batching configuration (e.g. Dynamic Batching) used, requests may contain multiple requests” **

    def execute(self, requests):
        """Depending on the batching configuration (e.g. Dynamic
        Batching) used, `requests` may contain multiple requests. 
        """
        for request in requests:
            pass

在以Python作为后端的情况下，光是在配置中设置dynamic_batching是不够的，它仅能够聚合请求，而要实现真正的模型批量推理，需要对model.py进行改造，简单而言就是将requests下多个request的请求数据沿着第一个维度拼接起来形成一个更大的batch，对完成的batch做一次推理，推理完成后再根据没有request自身的batch大小，按照顺序拆分成每个response，注意返回的response数量和request数量必须相等，我们对情感分类的后端model.py进行改造如下

    def execute(self, requests):
        responses = []
        # TODO 记录下每个请求的数据和数据batch大小
        batch_text, batch_len = [], []
        for request in requests:
            text = pb_utils.get_input_tensor_by_name(request, "text").as_numpy()
            text = np.char.decode(text, "utf-8").squeeze(1).tolist()
            batch_text.extend(text)
            batch_len.append(len(text))
        # 日志输出传入信息
        in_log_info = {
            "text": batch_text,
        }
        logging.info(in_log_info)

        encoding = self.tokenizer.batch_encode_plus(
            batch_text,
            max_length=512,
            add_special_tokens=True,
            return_token_type_ids=False,
            padding=True,
            return_attention_mask=True,
            return_tensors='pt',
            truncation=True
        ).to(self.device)
        with torch.no_grad():
            outputs = self.model(**encoding)
            prob = torch.nn.functional.softmax(outputs.logits, dim=1).argmax(dim=1).detach().cpu().numpy().tolist()
            prob = np.array([self.label_map[x].encode("utf8") for x in prob])

            # 日志输出处理后的信息
            out_log_info = {
                "prob": prob
            }
            logging.info(out_log_info)

        # TODO 响应数要和请求数一致
        start = 0
        for i in range(len(requests)):
            end = start + batch_len[i]
            out_tensor = pb_utils.Tensor("prob", np.array(prob[start:end]).astype(self.output_response_dtype))
            start += batch_len[i]
            final_inference_response = pb_utils.InferenceResponse(output_tensors=[out_tensor])
            responses.append(final_inference_response)
            
        return responses

简单而言该段代码将for循环中所有request数据进行拼接，最终只推理一次，返回时又根据各request大小做拆分。
接下来我们重新设置config.pbtxt，设置max_queue_delay_microseconds为200000，即0.2秒，preferred_batch_size暂不设置，另外还设置了最大批次大小max_batch_size为10,根据前文所述，批量聚合应该最大不超过max_batch_size

max_batch_size: 10
dynamic_batching {
    max_queue_delay_microseconds: 200000
}

额外的我们在model.py中打印出requests的数量，观察动态批处理是否生效，以及是否在requests上生效

def execute(self, requests):
        print("---------本次获取的请求数", len(requests))
        responses = []
        # TODO 记录下每个请求的数据和数据batch大小
        batch_text, batch_len = [], []
        for request in requests:

我们只使用一个实例启动服务，然后在客户端我们对每个请求只发送一条句子也就是batch_size=1，采用20个并发请求该服务，一共请求100个句子

samples = []
with open("./ChnSentiCorp.txt", encoding="utf8") as f:
    for line in f.readlines():
        samples.append(",".join(line.strip().split(",")[1:]))

samples = samples[:100]

def handle(sid):
    print("---------start:{}".format(sid))
    text = [samples[sid]]
    url = "http://0.0.0.0:18999/v2/models/sentiment/infer"
    raw_data = {
        "inputs": [
            {
                "name": "text",
                "datatype": "BYTES",
                "shape": [1, 1],
                "data": text
            }
        ],
        "outputs": [
            {
                "name": "prob",
                "shape": [1, -1],
            }
        ]
    }
    res = requests.post(url, json.dumps(raw_data, ensure_ascii=True), headers={"Content_Type": "application/json"},
                        timeout=2000)

    print("{},{},{}".format(str(sid), text, str(json.loads(res.text)["outputs"])))


if __name__ == '__main__':
    n = 100
    sid_list = list(range(n))
    t1 = time.time()
    with ThreadPoolExecutor(20) as executor:
        for i in sid_list:
            executor.submit(handle, i)
    t2 = time.time()
    print("耗时:", (t2 - t1) / n)

运行客户端，我们观察服务端日志，除第一次请求外，每次请求长度都是10，和max_batch_size一致，第一次请求长度1原因是模型预热导致。

---------本次获取的请求数 1
---------本次获取的请求数 10
---------本次获取的请求数 10
---------本次获取的请求数 10
---------本次获取的请求数 10
---------本次获取的请求数 10
---------本次获取的请求数 10
---------本次获取的请求数 10
---------本次获取的请求数 10
---------本次获取的请求数 10
---------本次获取的请求数 10

我们看原生的返回日志，发现推理的结果也是10个一批，说明不仅请求层面，模型推理层面也是实现了10个一次批处理

2024-04-03 02:59:42,001 - model.py[line:133] - INFO: {'prob': array([b'\xe6\xad\xa3\xe5\x90\x91', b'\xe6\xad\xa3\xe5\x90\x91',
       b'\xe6\xad\xa3\xe5\x90\x91', b'\xe6\xad\xa3\xe5\x90\x91',
       b'\xe8\xb4\x9f\xe9\x9d\xa2', b'\xe6\xad\xa3\xe5\x90\x91',
       b'\xe6\xad\xa3\xe5\x90\x91', b'\xe6\xad\xa3\xe5\x90\x91',
       b'\xe6\xad\xa3\xe5\x90\x91', b'\xe6\xad\xa3\xe5\x90\x91'],
      dtype='|S6')}

再观察推理的结果，虽然在后端将部分请求合并和推理，但是返回层由于加入了拆分逻辑，最后还是一个请求对应一条返回结果，推理的结果和不使用dynamic_batching的结果是完全一致的，笔者已经做过比对。

...
93,['"入住的是度假区的豪华海景房,前台给了5楼(最高6楼),然后差不多100%的海景,虽然是挂牌5星的,但是本人觉得是4星的标准,和我后来入住的5星喜来登差了蛮多的,不过整体来说还是符合他家的价钱的."'],[{'name': 'prob', 'datatype': 'BYTES', 'shape': [1], 'data': ['正向']}]
97,['酒店的基本设施一般，但服务态度确实很不错，房间8楼以下就是新装修的，8楼的房间就比较成旧，洗澡有单独的整体浴室，水比较大空调的风也很足，这个宾馆好像是属于海军的南海舰队，地理位置也很好，靠近省委火车站，离黄兴步行街也就三站地，值得入住'],[{'name': 'prob', 'datatype': 'BYTES', 'shape': [1], 'data': ['正向']}]
90,['"其他都可以,尽管不够5星标准,但还是很干净宽敞,最不能忍受的在度假区吃的自助海鲜晚餐,整个被苍蝇包围了,眼看着食物上落满了苍蝇,花了不少钱,落了一肚火....过道的海滩很美,很静,"'],[{'name': 'prob', 'datatype': 'BYTES', 'shape': [1], 'data': ['负面']}]
...

在初步跑通了Python后端的dynamic_batching之后，我们调整部分参数的设置，看看会有什么变化，我们先将请求数从100调整为102，这样注定有2个请求无法合并为最大批次10，我们看看Triton会如何处理，服务端日志如下

---------本次获取的请求数 1
---------本次获取的请求数 10
---------本次获取的请求数 10
---------本次获取的请求数 10
---------本次获取的请求数 10
---------本次获取的请求数 10
---------本次获取的请求数 10
---------本次获取的请求数 10
---------本次获取的请求数 10
---------本次获取的请求数 10
---------本次获取的请求数 10
---------本次获取的请求数 2

可见最后一批次只合并了2个请求，而此时max_queue_delay_microseconds生效，超过了0.2秒也是会打包这个批次送给模型推理。我们调整max_queue_delay_microseconds，使其变为20秒，重启服务

dynamic_batching {
    max_queue_delay_microseconds: 20000000
}

以同样的并发请求102条数据到服务端，日志如下

2024-04-03 03:14:38,322 - model.py[line:133] - INFO: {'prob': array([b'\xe6\xad\xa3\xe5\x90\x91', b'\xe8\xb4\x9f\xe9\x9d\xa2',
       b'\xe6\xad\xa3\xe5\x90\x91', b'\xe6\xad\xa3\xe5\x90\x91',
       b'\xe6\xad\xa3\xe5\x90\x91', b'\xe6\xad\xa3\xe5\x90\x91',
       b'\xe6\xad\xa3\xe5\x90\x91', b'\xe6\xad\xa3\xe5\x90\x91',
       b'\xe6\xad\xa3\xe5\x90\x91', b'\xe6\xad\xa3\xe5\x90\x91'],
      dtype='|S6')}
2024-04-03 03:14:58,255 - model.py[line:133] - INFO: {'prob': array([b'\xe8\xb4\x9f\xe9\x9d\xa2', b'\xe8\xb4\x9f\xe9\x9d\xa2'],
      dtype='|S6')}

服务端推理在最后一个批次卡住，卡住20秒，最后批次在2024-04-03 03:14:58推理了两条样本，而上一个批次在2024-04-03 03:14:38推理了10条样本，中间正好间隔20秒，可见此时max_queue_delay_microseconds的超时生效，那为什么之前没有感觉到明显的延迟？因为之前max_queue_delay_microseconds很小只有0.2秒。max_queue_delay_microseconds越大越有可能拼接到大的batch_size，而带来的后果是高延迟。
我们再对请求批次大小进行测试，刚才都是以batch_size=1进行请求，如果每次请求的批次量不定，比如为2，3，4..，dynamic_batching能否正常工作？修改客户端代码如下

def handle(sid):
    print("---------start:{}".format(sid))
    import random
    rand = random.randint(1, 6)
    text = [samples[sid:sid + rand]]
    url = "http://0.0.0.0:18999/v2/models/sentiment/infer"
    raw_data = {
        "inputs": [
            {
                "name": "text",
                "datatype": "BYTES",
                "shape": [rand, 1],
                "data": text
            }
        ],
        "outputs": [
            {
                "name": "prob",
                "shape": [rand, -1],
            }
        ]
    }
    res = requests.post(url, json.dumps(raw_data, ensure_ascii=True), headers={"Content_Type": "application/json"},
                        timeout=2000)

每次请求都随机从数据集中挑选1～6条数据作为一个批次请求，服务端日志如下

---------本次获取的请求数 1
---------本次获取的请求数 3
---------本次获取的请求数 2
---------本次获取的请求数 4
...
---------本次获取的请求数 2
---------本次获取的请求数 2

由于请求的批次不确定，因此每次合并的request也不定，有些requests合并4个才够10，而有些只合并2个就达到10了。
在官方文档中不推荐设置preferred_batch_size，大部分模型设置preferred_batch_size没有意义，除非模型对某个指定批次下有异于其他批次大小的突出性能能力，一般而言只需要设置max_batch_size和max_queue_delay_microseconds即可。
接下来我们测试使用动态批处理和不是用动态批处理的平均响应时间，我们控制推理设备，实例等其他外部条件不变，具体是使用1块GPU，1个实例，并发64，服务端最大推理批次64，服务端组合批次最大等待时间0.02秒，并发请求500次，在此条件下分别测试加入动态批处理和不加入动态批处理的性能，分别的我们调整客户端每个请求自身所带的batch_size大小，分别测试携带1，2，4，8，16条数据。

动态批处理对推理效率的影响

测试结论如上图所示，得到以下2点结论

• 1.开启动态批处理单条数据的响应时间低于不开启，推理效率明显更高，但是随着单个请求自身的batch_size增大，这个差距越来越小，就是说如果客户端逐渐进行批量发送，服务端动态批处理的效果越来越不明显
• 2.如果服务端开启了动态批处理，客户端已经没有必要刻意的批量发送数据了，从图上看黄线在客户端批次是1，2，4，8，16的时候，响应时间并没有明显的波动

全文完毕。