c++ - 具有任意作业参数的无锁作业队列

Question

目前我正在使用 Microsoft 的 Concurrency::concurrent_queue 类跨核心分配工作。工作线程从这个队列中提取工作。我有多个工作制作人(但如果真的有必要，我可以只和一个制作人一起生活)。

作业对象目前是这样实现的:

struct Job
{
    void(*function_pointer)(void*); // Job function (used to execute the task)
    void* arguments;    // Job arguments (gets passed to the job function)
};

如您所见，该系统要求作业生产者为每个作业的参数分配内存并保持分配状态直到作业完成。这正是我想要摆脱的，因为它很烦人而且容易出错。当然，我可以用 int argument; 之类的东西替换 void* arguments;。但是我的工作有任意数量和类型的参数。

一个简单的解决方案是将 void* arguments; 替换为如下内容:

char* arguments;    // Points to byte buffer that contains the arguments.

然后通过调用 new 为每个作业的参数分配内存(作业完成时会自动删除)。虽然这可行，但每个作业的分配对性能不利。
我考虑过使用循环缓冲区(用于存储所有作业的回调和参数)，但我找不到任何可以以无锁方式处理任意大小元素的实现。

简而言之，我希望能够做这样的事情:

struct Job_arguments
{
    int argument1;
    char argument2;
    float argument3;
};
Job_arguments arguments(1, 2, 3.0f);
job_system->add_job(&callback, arguments); // Arguments get copied.

我希望我的问题的解释很清楚。如果没有，我可以详细说明。

编辑:
对不起，如果我不清楚。我试图摆脱 Concurrency::concurrent_queue 以将其替换为不会导致分配的内容(如循环缓冲区)。作业还应以不会导致分配发生的方式存储。

Answer 1

你见过 std::function<void()> 吗？

它类型删除任何可调用的对象。要捆绑参数，只需创建一个 lambda [=]{ some_function( some_arguments ); }并将其传递给 std::function<void()> .

为了效率起见，搬进去，只完成一次分配。希望try_pop写得合理，搬进了目的地。

使用 [=]捕获，如[&]捕获是危险的。如果你需要一个缓冲区，你可以很容易地 [=]复制指针，然后手动删除(危险且容易出错)，使用 C++14 样式捕获列表和拥有缓冲区([buff=std::move(buff),=])，或者使用自己的 operator() 编写您自己的自定义对象.

[=]可能有点危险，因为它会默默地捕获所有提到的内容，因此您可以列出要按值捕获的变量 [var1, var2, var2] .

Concurrency::concurrent_queue< std::function< void() > > safe_queue;

// add to queue a call to `hello( some_string, some_int )`

void hello_to_queue( std::string some_string, int some_int ) {
  safe_queue.push( [=]{ hello(some_string, some_int); } );
}

此拷贝 some_string和 some_int进入 lambda，将 lambda 移入 std::function<void()> ，然后存储 std::function<void()>在队列中。

// dequeuer, runs elements until it cannot find any more:
void try_run() {
  std::function<void()> f;
  while(safe_queue.try_pop(f)) {
    f();
    f = nullptr;
  }
}

这会反复尝试提取 std::function<void()> s 来自队列。每次成功时，它都会运行提取的函数。然后在尝试填充 std::function 之前回收内存再次，因为为什么不早点做，而不是在并发代码中。

自然地，一个真实的用例涉及信号等。

A std::function<?>存储如何调用存储的对象，以及对象的任何存储。

lambda 可以捕获您想要的任何值，并公开一些要调用的代码。

它们一起让您可以存储“稍后在没有更多参数的情况下运行它”，将其存储在您的并发队列中，然后在另一个线程中运行它。

std::function实现可以将 SBO 用于捕获的小数据案例，以消除任何分配。 lambda 捕获中的移动表达式可以防止复制任何拥有的缓冲区(如 std::string 或 std::vector 或 std::unique_ptr )。在 std 函数的良好质量实现中，捕获的数据与类型删除 pimpl 一起存储，因此它将与您的手动实现的效率相匹配。由于它使用 RAII 来拥有数据，因此没有泄漏的风险。