具有多个生产者的高效消费者线程

Question

我试图通过跳过昂贵的事件操作来提高生产者/消费者线程的情况,如果有必要,可以使用以下方法:

//cas(variable, compare, set) is atomic compare and swap
//queue is already lock free

running = false


// dd item to queue – producer thread(s)

if(cas(running, false, true))
{
  // We effectively obtained a lock on signalling the event
  add_to_queue()
  signal_event()
}
else
{
  // Most of the time if things are busy we should not be signalling the event
  add_to_queue()

  if(cas(running, false, true))
    signal_event()
}

...

// Process queue, single consumer thread

reset_event()

while(1)
{
  wait_for_auto_reset_event() // Preferably IOCP

  for(int i = 0; i &lt SpinCount; ++i)
    process_queue()

  cas(running, true, false)

  if(queue_not_empty())
    if(cas(running, false, true))
      signal_event()
}

显然试图让这些东西正确是有点棘手的(!)所以上面的伪代码是正确的吗？一个解决方案可以向事件发出超过完全需要的信号,但不是每个项目都能这样做的解决方案.

Answer 1

查了很多案例，没看出问题。但这有点复杂。我想也许你会对queue_not_empty / add_to_queue racing 有问题。但看起来两条路径中的后主宰 CAS 都涵盖了这种情况。

\n\n

CAS 很昂贵（不像信号那么昂贵）。如果您预计跳过信号是常见的情况，我将按如下方式对 CAS 进行编码：

\n\n

bool cas(variable, old_val, new_val) {\n   if (variable != old_val) return false\n   asm cmpxchg\n}\n

\n\n

像这样的无锁结构是Jinx（我从事的产品）非常擅长测试的东西。因此，您可能需要使用评估许可证来测试无锁队列和信号优化逻辑。

\n\n\n\n

编辑：也许你可以简化这个逻辑。

\n\n

running = false \n\n// add item to queue \xe2\x80\x93 producer thread(s) \nadd_to_queue()\nif (cas(running, false, true)) {\n   signal_event()\n}\n\n// Process queue, single consumer thread \n\nreset_event() \n\nwhile(1) \n{ \n    wait_for_auto_reset_event() // Preferably IOCP \n\n   for(int i = 0; i &lt SpinCount; ++i) \n       process_queue() \n\n   cas(running, true, false)  // this could just be a memory barriered store of false\n\n   if(queue_not_empty()) \n      if(cas(running, false, true)) \n         signal_event() \n} \n

\n\n

现在 cas/signal 总是彼此相邻，因此可以将它们移至子例程中。

\n