c ++两个线程之间更快的变量交换一个std :: vector <float *>和array

Posted 2023-02-22

【中文标题】c ++两个线程之间更快的变量交换一个std :: vector <float *>和array

【英文标题】：c ++ faster variable exchange between two threads an std :: vector <float *> and array

【发布时间】：2020-05-17 11:17:13

【问题描述】：

我有一个在无限循环中计算数据的线程。 它产生三个结果

一个std::vector（每个向量元素后面都有一个浮点数 堆上 3 个元素的数组） 一个 int 数组* 一个int，表示int[]的大小。

另外两个线程现在应该处理这些数据（也是无穷无尽的）。由于所有三个线程的处理时间不同，因此有时会“跳过”数据。

std::vector<float*> a_vecor;
int* a_Array;
int a_Array_size;
std::mutex a_mutex;


void thread_A() 

    while (true) 
     
         calculate(); 
         a_mutex.lock();
         a_vector = getvector();
         a_Array = getArray();
         a_Array_size = getArraysize();
         a_mutex.unlock();
    


void thread_B() 

    while (true) 
     
         a_mutex.lock();
         std::vector<float*> b_vector = a_vector;
         int* b_Array = a_Array;
         int b_array_Size = a_array_Size; 
         a_mutex.unlock();
         calculate_b();
    


void thread_C() 

    while (true) 
     
         a_mutex.lock();
         std::vector<float*> c_vector = a_vector;
         int* c_Array = a_Array;
         int c_array_Size = a_array_Size; 
         a_mutex.unlock();
         calculate_c();
    

我的问题是如何将这些数据从一个线程传递到另一个线程？ 我实际上会在以下线程中复制三个参数，但我可以：

std::vector <float*> b = calculate();
std::vector <float*> a = b;

这是副本还是参考？仅作为参考的矢量元素怎么样？ 复制向量和数组的最快方法是什么？循环不是也在内部运行吗？ 信息流只有一个方向，一个进程生成数据，下一个进程只读取它。我可以利用这个吗？ 有没有互斥锁的替代品？

【问题讨论】：

std::vector <float*> a = b; 是一个副本，但是当生产者重新计算 b 时，a 的元素可能会变得无效。 @Metatron

向量的大小有几千个元素，但每个元素由3个元素组成的数组（新float [3]）组成

也许你应该先解决这个问题。分配是一项非常昂贵的操作。

【参考方案1】：

所以您希望一个线程写入数据，另一个线程读取数据？ - 使用互斥体执行此操作的标准方法是。您确实需要一个互斥锁，因为即使您尝试访问它，您正在读取的值也可能会发生变化（例如向量的大小） - 导致未定义的行为（可能是段错误！）。

同样对于vector，一旦你向它添加更多元素，它就不能保证在同一个位置——它可能需要更多的内存并且它被移动到其他地方！ - 如果您正在阅读，这很糟糕。

所以我真的看不出你有什么选择——除非你复制一份——但这更浪费。如果你的向量是固定大小的，你也许可以使用std::atomic<int/float> 的向量，但是如果你的向量没有改变大小，那么使用 std::array。

您可以使用双端队列（双端队列 - 实际上是一个链表），其中每个元素确实保留在内存中的同一位置，并且假设您的编写器没有从双端队列中删除项目，应该可以安全地读取！

所以我会从一个互斥锁开始——它非常易于使用，还有一些共享数据，例如：

// NOTE: uses the gate class from the link in the comments for convenience
// pseudo code only

std::vector<int> shared_data;
std::atomic<int> workers_finished = 0;
gate writer_gate;
gate worker_gate;


// writer thread
// ... write to shared_data ...
// Now reset workers finished count and signal workers to start
workers_finished = 0;
worker_gate.open();
// Wait for workers to finish
while (workers_finished < num_workers)

    writer_gate.wait_at_gate();

// All workers finish - Loop back to start or whatever...


// Worker thread:
// Note - all workers can look like this:
// Wait to be signalled by the writer
worker_gate.wait_at_gate();
// Writer signalled - do work here
// Finished work, increment workers finished count
workers_finished++;
// Signal the writer thread
writer_gate.open();
// loop back to start or whatever...

不过，我取决于你想怎么做。如果你想等待作者写完，或者你想和作者同时阅读？

另一种方法是：

作家确实工作 编写器完成后 - 生成多个线程来完成工作 等待线程完成（join_thread - 或使用 std::async / std::future）。 循环

我会说这是一种更简单的方法..

【讨论】：

单生产者多消费者，无锁队列在这里也很高效。 @code_fodder

@v78 我认为这真的取决于 OP 想要做什么。一些操作对原子行为是可以的，而另一些则不是。例如，如果您有一个线程向其中添加元素，我认为您不会想要使用没有任何互斥锁的 std::vector （因为整个向量可以在内存中重新分配，并且 OP 建议他想要 std::vector （ AFAIK）

如果您看到我的编辑，我目前拥有它。但是“复制过程”几乎与实际处理时间一样长。也许我弄错了，但如果我只传递一个 ref，线程 A 可以在 B / C 访问它时更改数据。如果我在整个访问中设置互斥锁，我也可以连续工作......

@Metatron 因为您使用的是互斥锁 - 无需复制数据。只需使用 a_vector 全局变量直接访问它 - 无需使用 getvector() 所做的任何事情（在哪里定义？）。我会尝试添加一个简单的例如

@Metatron 事实上我无法添加一个合理的例子，因为我不知道你在用数据做什么！ - calculate_a() 是做什么的？如果你想优化速度，你可以预先分配你的向量（但使用 std::array），所以你不需要向它添加/删除元素并使用 std::atomic （如前所述）然后你会' t 需要互斥体......但没有看到所有代码的作用 - 或者你试图实现的目标很难写出一个好的例子:)

【参考方案2】：

std::vector <float*> a = b; 复制了b，但因为它是一个指针向量，所以只有指针被复制，而不是它们指向的数据。

存储计算数据而不是按值存储以便于复制：

struct Data 
    std::vector<std::array<float, 3>> dataA;
    std::vector<int> dataB;
;

现在Data a = b; 将复制数据。

---

补充说明：

您可能希望添加condition_variable 以在生成新数据时发出信号，以避免消费者再次处理相同的数据：

std::condition_variable data_produced_cond;

可以复制消费者中的数据，但更有效的解决方案是使用ring-buffer，以便重新使用已分配的内存。

【讨论】：

您好，谢谢，我知道环形缓冲区，但我认为这并不能解决我的问题。我希望线程 A 始终评估传感器数据，线程 B 和 C 始终处理最近创建的数据记录（所有较旧的数据记录都不相关）。问题是所有线程都花费不同的时间长度，因此并不总是处理相同的数据。使用环形缓冲区，我仍然会遇到线程 A 可能会在线程 B 或 C 级读取中将数据写入站点的问题。

一开始只能想到B和C为自己复制数据（下载数据集）。只有复制似乎需要很长时间，因为我目前正在使用迭代器循环遍历元素的向量元素，并且实际上已经可以在这里运行评估......

我从问题中不明白这一点。无论如何，通过始终处理缓冲区中的最后一个元素并丢弃它之前的所有排队数据，很容易为此调整环形缓冲区解决方案。