TBB Parallel Pipeline 详细用法指南
tbb::parallel_pipeline 是 TBB 库中一个强大的流水线并行处理工具，专为序列化数据处理而设计。它可以将一系列处理步骤组织成流水线，每个阶段可以并行或串行执行。
基本概念
- 流水线（Pipeline）：多个处理阶段的组合
- 过滤器（Filter）：单个处理阶段
- 令牌（Token）：控制并发度的单位
- 过滤模式：
filter::serial_in_order：串行执行且保持输入顺序filter::serial_out_of_order：串行执行但不保证顺序filter::parallel：并行执行多个项目
基本语法
#include <tbb/pipeline.h>
tbb::parallel_pipeline(
size_t max_tokens, // 并发度（流水线中同时处理的项目数）
filter1 & filter2 & filter3 & ... & filterN // 流水线过滤器链接
);
每个过滤器可以使用 tbb::make_filter 创建：
tbb::make_filter<InputType, OutputType>(
tbb::filter::mode, // 过滤器模式（串行或并行）
[](InputType input) -> OutputType { // 过滤器函数
// 处理逻辑
return output;
}
);
详细示例
以下是一个完整的示例，展示了使用 parallel_pipeline 处理人体检索的流程：
#include <tbb/pipeline.h>
#include <tbb/task_group.h>
#include <vector>
#include <string>
#include <iostream>
struct SearchRequest {
std::string query;
int searchId;
};
struct DatabaseRecord {
std::vector<float> features;
std::string recordId;
};
struct ScoredResult {
std::string recordId;
float score;
};
struct SearchResult {
std::vector<ScoredResult> results;
int searchId;
};
// 人体检索流水线示例
void personSearchPipeline(const std::vector<SearchRequest>& requests) {
// 结果存储
std::vector<SearchResult> finalResults;
std::mutex resultsMutex;
// 1. 设置并行度（令牌数）- 可根据CPU核心数调整
size_t maxTokens = 16;
tbb::parallel_pipeline(
// 最大并发处理的项目数
maxTokens,
// 第一阶段：输入阶段 - 通常是串行的，提供输入数据
tbb::make_filter<void, SearchRequest*>(
tbb::filter::serial_in_order,
[&requests, counter = size_t{0}](tbb::flow_control& fc) mutable -> SearchRequest* {
if (counter >= requests.size()) {
// 结束流水线
fc.stop();
return nullptr;
}
// 返回下一个请求指针
return const_cast<SearchRequest*>(&requests[counter++]);
}
) &
// 第二阶段：特征提取 - 可并行处理多个请求
tbb::make_filter<SearchRequest*, std::pair<SearchRequest*, std::vector<float>>>(
tbb::filter::parallel,
[](SearchRequest* request) -> std::pair<SearchRequest*, std::vector<float>> {
// 模拟从请求中提取特征
std::vector<float> features(128, 0.5f); // 假设的特征向量
std::cout << "提取特征: 请求ID " << request->searchId << std::endl;
return {request, features};
}
) &
// 第三阶段：数据库查询 - I/O操作，可并行执行以重叠等待时间
tbb::make_filter<std::pair<SearchRequest*, std::vector<float>>,
std::pair<SearchRequest*, std::vector<DatabaseRecord>>>(
tbb::filter::parallel,
[](std::pair<SearchRequest*, std::vector<float>> input) ->
std::pair<SearchRequest*, std::vector<DatabaseRecord>> {
// 模拟数据库查询，获取候选记录
std::cout << "数据库查询: 请求ID " << input.first->searchId << std::endl;
// 模拟从数据库检索记录
std::vector<DatabaseRecord> records;
for (int i = 0; i < 10; i++) {
records.push_back({
std::vector<float>(128, 0.3f), // 假设的特征
"record_" + std::to_string(i)
});
}
return {input.first, records};
}
) &
// 第四阶段：特征比对 - 计算密集型，适合并行
tbb::make_filter<std::pair<SearchRequest*, std::vector<DatabaseRecord>>,
std::pair<SearchRequest*, std::vector<ScoredResult>>>(
tbb::filter::parallel,
[](std::pair<SearchRequest*, std::vector<DatabaseRecord>> input) ->
std::pair<SearchRequest*, std::vector<ScoredResult>> {
auto request = input.first;
auto& records = input.second;
std::cout << "特征比对: 请求ID " << request->searchId << std::endl;
// 模拟特征比对和评分
std::vector<ScoredResult> scoredResults;
for (const auto& record : records) {
// 简单模拟评分计算
float score = 0.8f + (static_cast<float>(rand()) / RAND_MAX) * 0.2f;
scoredResults.push_back({record.recordId, score});
}
return {request, scoredResults};
}
) &
// 第五阶段：结果排序 - 按照顺序处理每个结果集更好
tbb::make_filter<std::pair<SearchRequest*, std::vector<ScoredResult>>, SearchResult*>(
tbb::filter::serial_in_order,
[](std::pair<SearchRequest*, std::vector<ScoredResult>> input) -> SearchResult* {
auto request = input.first;
auto& scoredResults = input.second;
std::cout << "结果排序: 请求ID " << request->searchId << std::endl;
// 根据分数排序
std::sort(scoredResults.begin(), scoredResults.end(),
[](const ScoredResult& a, const ScoredResult& b) {
return a.score > b.score;
});
// 创建结果对象
auto* result = new SearchResult{scoredResults, request->searchId};
return result;
}
) &
// 第六阶段：结果收集 - 通常是串行的，将结果保存
tbb::make_filter<SearchResult*, void>(
tbb::filter::serial_in_order,
[&finalResults, &resultsMutex](SearchResult* result) {
std::cout << "收集结果: 请求ID " << result->searchId
<< ", 找到 " << result->results.size() << " 条匹配" << std::endl;
// 保存结果
{
std::lock_guard<std::mutex> lock(resultsMutex);
finalResults.push_back(*result);
}
// 释放内存
delete result;
}
)
);
// 流水线执行完成后，finalResults 包含所有处理结果
std::cout << "处理完成，共 " << finalResults.size() << " 条结果" << std::endl;
}
在人体检索系统中的应用
针对人体检索系统，parallel_pipeline 可以用于以下场景：
flowchart LR
A["输入请求\n(串行)"] --> B["特征提取\n(并行)"]
B --> C["数据库查询\n(并行)"]
C --> D["特征比对\n(并行)"]
D --> E["结果聚合\n(串行)"]
E --> F["后处理和输出\n(串行)"]
实际应用示例
// 人体检索系统中的具体应用
void personTrackPipeline(const std::vector<TrackRequest>& requests) {
tbb::parallel_pipeline(
// 设置合理的并发度
std::min(static_cast<size_t>(16), requests.size()),
// 输入阶段
tbb::make_filter<void, TrackRequest*>(
tbb::filter::serial_in_order,
[&requests, i = 0](tbb::flow_control& fc) mutable -> TrackRequest* {
if (i >= requests.size()) {
fc.stop();
return nullptr;
}
return const_cast<TrackRequest*>(&requests[i++]);
}
) &
// 并行检索实名有抓拍档案
tbb::make_filter<TrackRequest*, std::pair<TrackRequest*, ArchiveResults>>(
tbb::filter::parallel,
[](TrackRequest* req) -> std::pair<TrackRequest*, ArchiveResults> {
ArchiveResults results = searchRealNameWithCapture(req->parameters);
return {req, results};
}
) &
// 并行检索实名无抓拍档案和非实名档案
tbb::make_filter<std::pair<TrackRequest*, ArchiveResults>,
std::tuple<TrackRequest*, ArchiveResults, ArchiveResults, ArchiveResults>>(
tbb::filter::parallel,
[](std::pair<TrackRequest*, ArchiveResults> in) ->
std::tuple<TrackRequest*, ArchiveResults, ArchiveResults, ArchiveResults> {
auto req = in.first;
auto realWithCapture = in.second;
// 并行执行这两个检索（使用内部TBB任务组）
ArchiveResults realNoCapture;
ArchiveResults unrealArchives;
tbb::task_group g;
g.run([&]{ realNoCapture = searchRealNameNoCapture(req->parameters); });
g.run([&]{ unrealArchives = searchNoName(req->parameters); });
g.wait();
return {req, realWithCapture, realNoCapture, unrealArchives};
}
) &
// 并行检索实时记录
tbb::make_filter<std::tuple<TrackRequest*, ArchiveResults, ArchiveResults, ArchiveResults>,
std::tuple<TrackRequest*, ArchiveResults, ArchiveResults, ArchiveResults, RecordResults>>(
tbb::filter::parallel,
[](std::tuple<TrackRequest*, ArchiveResults, ArchiveResults, ArchiveResults> in) ->
std::tuple<TrackRequest*, ArchiveResults, ArchiveResults, ArchiveResults, RecordResults> {
auto req = std::get<0>(in);
RecordResults rtRecords = searchRealtime(req->parameters);
return {req, std::get<1>(in), std::get<2>(in), std::get<3>(in), rtRecords};
}
) &
// 结果合并与排序
tbb::make_filter<std::tuple<TrackRequest*, ArchiveResults, ArchiveResults, ArchiveResults, RecordResults>,
SearchResponse*>(
tbb::filter::serial_in_order,
[](std::tuple<TrackRequest*, ArchiveResults, ArchiveResults, ArchiveResults, RecordResults> in) ->
SearchResponse* {
auto req = std::get<0>(in);
// 合并所有结果
SearchResponse* response = new SearchResponse();
response->requestId = req->requestId;
// 合并不同来源的结果
mergeResults(response->results, std::get<1>(in)); // 实名有抓拍
mergeResults(response->results, std::get<2>(in)); // 实名无抓拍
mergeResults(response->results, std::get<3>(in)); // 非实名
mergeResults(response->results, std::get<4>(in)); // 实时记录
// 排序
sortResults(response->results);
return response;
}
) &
// 最终处理与存储
tbb::make_filter<SearchResponse*, void>(
tbb::filter::serial_in_order,
[](SearchResponse* response) {
// 存储结果到缓存
storeResultToCache(response);
// 输出或发送结果
sendResultToClient(response);
delete response;
}
)
);
}
性能优化技巧
- 调整令牌数量：根据工作负载和CPU核数调整
max_tokens - 选择适当的过滤模式：
- 对I/O密集型操作使用
parallel或serial_out_of_order - 对必须保持顺序的操作使用
serial_in_order
- 对I/O密集型操作使用
- 合理分配工作量：避免某一阶段成为瓶颈
- 减少数据复制：尽可能使用指针或引用传递数据
- 合并短小阶段：流水线开销也需要考虑，过短的阶段可能得不偿失
通过合理设计并行流水线，可以充分利用多核CPU的处理能力，显著提升人体检索系统的性能。
No comments yet.