C++(11) "High Performance"并发单个写入器

Question

我有一个对象 B，由 10 个 double 值组成。 double 值由一个传感器产生，该传感器进行通信 通过以太网，每 1-2 毫秒发送一次更新。 B 的实例必须在程序中的其他 2 个地方使用，以 做一些计算和可视化。

在另一个线程中，使用另一个传感器每 4-10 毫秒更新一个对象 A，其中包含约 1000 个 double 值。

两个对象都有一个时间签名(在传感器更新到达时使用 boost::chrono::high_resolution_clock 获得)

现在我想使用两个对象 A 和 B 几乎已经更新了 同时计算对象 C 的一些值。 这应该在两个线程都运行时完成，并且输出 用于做一些可视化，计算平均值等。 整个过程运行1-2小时， 之后仅使用计算的 C 实例， 不再需要 A 和 B 的实例。

在线程之间实现这种通信和数据共享的推荐方法(或设计模式)是什么？

目前，整个结构实现得很糟糕，而且 构成A的线程直接与构成B的线程通信 不使用互斥锁等同步方法。

• 我应该使用无锁双端队列来存储A 和B吗 并在编写 C 的线程中读取它？
• 我是否应该使用像 boost::signals2 这样的(线程安全的)观察者模式来“发送”A 和 B 的实例？
• 或者不同的东西？

Answer 1

我假设您在没有实时保证的情况下在 PC 和操作系统上进行所有计算。因此，只要您处于毫秒范围内，您就不会受到严格的实时限制。因此，使用无锁数据结构并不是绝对必要的。但是你可以。他们也有很好的表现。如果你愿意，你可以为此使用 boosts lockfree 数据结构。但你可能想要一个 pop()阻塞直到队列不为空的函数，因为否则你需要忙等待或者使用某种信号量或条件变量来等待队列不为空的状态。

您的竞争很低，只有几个线程。在低争用情况下锁定互斥锁通常需要 25ns 的数量级。所以这不是问题。现在内存分配通常也非常快，通常平均不到 100ns。所以这是我的看法:

• 使用std::queue<std::packaged_task<void()>>作为任务队列。您也可以决定排队 std::function<void()>而不是 std::packaged_task<void()>()如果你不需要std::future s 为计算结果。使用 std::mutex 保护对其的访问.有std::condition_variable那就是等待队列非空。我曾经实现过 a generic blocking concurrent queue on github这比那更有效，但仅使用标准库方法。您只需将模板参数指定为 T=std::packaged_task<void()>您将拥有所需的数据结构。

• 创建话题 std::thread它除了执行传入的任务外什么都不做。要退出线程，请设置一些 std::atomic<bool>在线程的最后一个任务中标记，因此线程停止执行，您可以 join()它优雅地。同样，您可以将该功能放入类中。类的析构函数应该告诉任务执行线程停止并执行 join() .我实现了这样一个 class in the same repository on github .你可以自由使用它。

• 现在您可以将任务从从硬件接收数据的两个线程传递给任务执行器。只需将适当的 lambda 传递给执行需要完成的工作的任务执行器。如果您决定使用我的实现，您可以编写如下内容

  void receiveData( const Data & data, cu::ParallelExecutor & executor )
  {
      executor.addTask( [=]{ /* Do something with `data` */ } );
  }
  

这样一切都是线程安全的，具有低开销和高性能。