在流水线执行中采用并行性
Employing parallelism in pipelined execution
我正在尝试开发一个管道,其中首先读取和处理数据,操作一次,以不同的方式操作,然后显示。我有一个设计,其中数据 IO 馈送到由第一个操纵器读取的缓冲区中。随后,第一操纵器写入另一个缓冲区,如果可能的话,第二个操纵器会读取该缓冲区。最后,第二个操纵器的输出被写入显示缓冲区,该缓冲区由可视化工具读取并使用 OpenGL 显示。
在我看来,这是一个相当简单的并行问题,其中每个任务都有自己的线程,它们通过数据缓冲区进行通信。但是,我遇到的所有线程程序教程似乎都表明,多线程是留给一些中间件(如OpenMP)来决定如何分配工作负载的东西。
我是开发多线程应用程序的新手,所以这可能是一个愚蠢的问题,但我所描述的可行吗,可以使用 OpenMP 等中间件完成吗?我意识到显而易见的答案是"尝试一下",而且我想这样做,但是教程没有阐明*如何*尝试它。
OpenMP 更适合轻松跨越多核 (SIMD) 的算法。其他情况是可能的,但在您的情况下,我认为直接使用线程会更好,并且更容易编码和维护。
我将答案分为两部分:没有OpenMP的通用解决方案,以及使用OpenMP的一些特定更改。
正如评论中提到的,您面临着生产者/消费者的问题,但有两次:一个线程正在填充缓冲区(生成一个项目),然后必须由第二个线程读取(和修改)该缓冲区(已消费)。问题的特殊性在于,第二个线程也是一个生产者(要绘制的图像),第三个线程是负责使用它的线程(可视化工具)。
如您所知,P/C 问题是使用缓冲区(可能是循环缓冲区或生成项队列)解决的,其中缓冲区的每个元素都标记为已生产或已使用,并且线程在添加或获取项目时具有独占访问权限。
让我们在以下示例程序中使用队列方法来解决您的问题。
- 生成的物料将存储在队列中。队列的前面包含最旧的元素,即必须首先使用的元素。
- 有两个队列:一个用于第一个操纵器生成的数据(并由第二个操纵器使用),另一个用于第二个操纵器生成的数据(将由另一个线程可视化)。
- 生产阶段很简单:获得对相应队列的独占访问权限,并在末尾插入元素。
- 消耗类似,但必须等待队列至少有一个元素(不为空)。
- 我添加了一些睡眠来模拟其他操作。
- 停止条件仅用于说明目的。
注意:为了简单起见,我假设您可以访问 C++11 编译器。使用其他 API 的实现相对相似。
#include <iostream>
#include <thread>
#include <mutex>
#include <atomic>
#include <chrono>
#include <list>
using namespace std::chrono_literals;
std::mutex g_data_produced_by_m1_mutex;
std::list<int> g_data_produced_by_m1;
std::mutex g_data_produced_by_m2_mutex;
std::list<int> g_data_produced_by_m2;
std::atomic<bool> stop = false;
void manipulator1_kernel()
{
while (!stop) {
// Producer 1: generate data
{
std::lock_guard<std::mutex> lock(g_data_produced_by_m1_mutex);
g_data_produced_by_m1.push_back(rand());
}
std::this_thread::sleep_for(100ms);
}
}
void manipulator2_kernel()
{
int data;
while (!stop) {
// Consumer 1
while (!stop) { // wait until there is an item to be consumed
{
std::lock_guard<std::mutex> lock(g_data_produced_by_m1_mutex);
if (!g_data_produced_by_m1.empty()) { // is there data to be consumed?
data = g_data_produced_by_m1.front(); // consume
g_data_produced_by_m1.pop_front();
break;
}
}
std::this_thread::sleep_for(100ms);
}
// Producer 2: modify and send to the visualizer
{
std::lock_guard<std::mutex> lock(g_data_produced_by_m2_mutex);
g_data_produced_by_m2.push_back(5 * data);
}
std::this_thread::sleep_for(100ms);
}
}
void visualizer_kernel()
{
int data;
while (!stop) {
// Consumer 2
while (!stop) { // wait until there is an item to be visualized
{
std::lock_guard<std::mutex> lock(g_data_produced_by_m2_mutex);
if (!g_data_produced_by_m2.empty()) {
data = g_data_produced_by_m2.front();
g_data_produced_by_m2.pop_front();
break;
}
}
std::this_thread::sleep_for(100ms);
}
std::cout << data << std::endl; // render to display
std::this_thread::sleep_for(100ms);
if (data % 8 == 0) stop = true; // some stop condition for the example
}
}
int main()
{
std::thread manipulator1(manipulator1_kernel);
std::thread manipulator2(manipulator2_kernel);
std::thread visualizer(visualizer_kernel);
visualizer.join();
manipulator2.join();
manipulator1.join();
return 0;
}
如果您仍然想使用 OpenMP,您可能最接近的是任务(我认为是从 OpenMP 3.0 开始)。我不太怎么使用它们,但是上面的程序可以重写为:
int main()
{
#pragma omp parallel
{
#pragma omp task
manipulator1_kernel();
#pragma omp task
manipulator2_kernel();
#pragma omp task
visualizer_kernel();
#pragma omp taskwait
}
return 0;
}
其余代码也可以更改为使用 OpenMP 功能,但我认为这回答了您的问题。
这种方法的主要问题是您必须为任务创建一个代码块才能存在于 OpenMPparallel
中,这很容易使应用程序的其余逻辑和结构复杂化。
为了解决这个特殊问题,英特尔®线程构建模块库包括特殊的结构。英特尔® TBB 是跨平台库,有助于多线程编程。 我们可以将应用程序中涉及的实体视为四个不同的任务提供程序。一种类型的任务是输入任务 - 提供输入数据的任务,另一种类型的任务由第一个操作例程提供,依此类推。
因此,用户唯一需要做的就是为这些任务提供主体。库中有几个 API,用于指定要处理的主体以及如何并行处理。其他所有内容(这里我的意思是线程创建、任务执行之间的同步、工作平衡等)都由库完成。
我想到的解决方案的最简单变体是使用parallel_pipeline函数。这是原型:
#include "tbb/pipeline.h"
using namespace tbb;
int main() {
parallel_pipeline(/*specify max number of bodies executed in parallel, e.g.*/16,
make_filter<void, input_data_type>(
filter::serial_in_order, // read data sequentially
[](flow_control& fc) -> input_data_type {
if ( /*check some stop condition: EOF, etc.*/ ) {
fc.stop();
return input_data_type(); // return dummy value
}
auto input_data = read_data();
return input_data;
}
) &
make_filter<input_data_type, manipulator1_output_type>(
filter::parallel, // process data in parallel by the first manipulator
[](input_data_type elem) -> manipulator1_output_type {
auto processed_elem = manipulator1::process(elem);
return processed_elem;
}
) &
make_filter<manipulator1_output_type, manipulator2_output_type>(
filter::parallel, // process data in parallel by the second manipulator
[](manipulator1_output_type elem) -> manipulator2_output_type {
auto processed_elem = manipulator2::process(elem);
return processed_elem;
}
) &
make_filter<manipulator2_output_type, void>(
filter::serial_in_order, // visualize frame by frame
[](manipulator2_output_type elem) {
visualize(elem);
}
)
);
return 0;
}
前提是实现了必要的功能(read_data、可视化)。这里input_data_type
、manipulator1_output_type
等是在流水线阶段之间传递的类型,操纵器的process
函数对传递的参数进行必要的计算。
顺便说一句,为了避免使用锁和其他同步原语,您可以使用库中的concurrent_bounded_queue并将输入数据放入此队列中,通过可能不同的线程(例如专用于 IO 操作),就像concurrent_bounded_queue_instance.push(elem)
一样简单,然后通过input_data_type elem; concurrent_bounded_queue_instance.pop(elem)
读取它。请注意,弹出项目是此处的阻止操作。concurrent_queue
提供了非阻塞try_pop
替代方案。
另一种可能性是使用tbb::flow_graph
及其节点来组织相同的流水线方案。看一下描述依赖项和数据流图的两个示例。您可能需要使用 sequencer_node 来正确排序项目执行(如有必要)。
值得阅读由 tbb 标签标记的 SO 问题,以了解其他人如何使用这个库。
你实现过单线程版本吗?
它们是关键步骤,如果没有它们,您可以获得高度并行设计的最佳实现,只是为了意识到瓶颈是缓冲区的 I/O 和/或线程同步和/或错误共享和/或缓存未命中或类似问题。
我首先尝试一个简单的线程池,其中包含按顺序执行所有步骤的任务。然后在分析了它的工作原理之后,什么是 CPU 消耗等。我会尝试更复杂的工具,总是将它们的性能与第一个简单版本进行比较
- OpenMP for 循环并行性问题
- 如何在cython中使用并行性
- 如何在 Skylake 架构上最大化 sqrt-heavy loop 的指令级并行性?
- 在流水线执行中采用并行性
- C++17 在并行性方面究竟发生了哪些变化?(权威文档在哪里?
- OpenMP 按需嵌套并行性
- 通用 C++11 函数包装器,用于基于任务的并行性
- 将执行从一个线程移动到另一个线程,以实现任务并行性并在将来调用
- 互斥任务集的线程级并行性
- 使用TBB的并行性-我们的检查列表中应该包含哪些内容
- Xeon phi卸载模式如何利用线程并行性和矢量化
- 并行性与线程 - 性能
- 从基于线程的流水线转向基于任务的并行性?(C++)
- 编译包含动态并行性的代码失败
- std::async 使用相同的线程,我的代码没有实现并行性
- 内部和外部并行性
- 某些子网格未使用CUDA动态并行性执行
- 用STL算法替换for循环,同时允许并行性
- 使用boost::Thread的线程管理和并行性
- CUDA动态并行性:使用纹理内存时无效的全局写入