我正在为密集型网络应用程序实现一个无锁的单生产者单消费者队列。我有一堆工作线程在它们自己的单独队列中接收工作，然后它们取消排队并处理这些队列。
从这些队列中删除锁大大提高了高负载下的性能，但是当队列为空时，它们不再阻塞，这反过来又会导致 CPU 使用率飙升。
如何有效地使线程阻塞，直到它可以成功取消排队或被终止/中断？
如果你使用的是Linux，可以考虑使用Futex。 它通过使用原子操作而不是像互斥锁那样的内核调用来提供非锁定实现的性能，但是如果您由于某些条件不为真（即锁争用）而需要将进程设置为空闲，它将进行适当的内核调用以使进程进入睡眠状态并在将来的事件中将其唤醒。 它基本上就像一个非常快的信号量。
在 Linux 上，futex 可以用来阻塞线程。但请注意，Futexes很棘手！
更新：条件变量比 futexes 更安全，并且更具可移植性。但是，条件变量与互斥锁结合使用，因此严格来说，结果将不再是无锁的。但是，如果您的主要目标是性能（而不是全局进度的保证），并且锁定部分（即线程唤醒后检查的条件）很小，则可能会获得令人满意的结果，而无需深入了解将 futexes 集成到算法中的微妙之处。
如果你使用的是Windows，你将无法使用futexes，但Windows Vista有一个类似的机制，称为键控事件。不幸的是，这不是已发布 API 的一部分（它是一个 NTDLL 本机 API），但只要您接受它可能会在未来版本的 Windows 中更改的警告（并且您不需要在 Vista 之前的内核上运行），您就可以使用它。请务必阅读我上面链接的文章。以下是它如何工作的未经测试的草图：
/* Interlocked SList queue using keyed event signaling */
struct queue {
SLIST_HEADER slist;
// Note: Multiple queues can (and should) share a keyed event handle
HANDLE keyed_event;
// Initial value: 0
// Prior to blocking, the queue_pop function increments this to 1, then
// rechecks the queue. If it finds an item, it attempts to compxchg back to
// 0; if this fails, then it's racing with a push, and has to block
LONG block_flag;
};
void init_queue(queue *qPtr) {
NtCreateKeyedEvent(&qPtr->keyed_event, -1, NULL, 0);
InitializeSListHead(&qPtr->slist);
qPtr->blocking = 0;
}
void queue_push(queue *qPtr, SLIST_ENTRY *entry) {
InterlockedPushEntrySList(&qPtr->slist, entry);
// Transition block flag 1 -> 0. If this succeeds (block flag was 1), we
// have committed to a keyed-event handshake
LONG oldv = InterlockedCompareExchange(&qPtr->block_flag, 0, 1);
if (oldv) {
NtReleaseKeyedEvent(qPtr->keyed_event, (PVOID)qPtr, FALSE, NULL);
}
}
SLIST_ENTRY *queue_pop(queue *qPtr) {
SLIST_ENTRY *entry = InterlockedPopEntrySList(&qPtr->slist);
if (entry)
return entry; // fast path
// Transition block flag 0 -> 1. We must recheck the queue after this point
// in case we race with queue_push; however since ReleaseKeyedEvent
// blocks until it is matched up with a wait, we must perform the wait if
// queue_push sees us
LONG oldv = InterlockedCompareExchange(&qPtr->block_flag, 1, 0);
assert(oldv == 0);
entry = InterlockedPopEntrySList(&qPtr->slist);
if (entry) {
// Try to abort
oldv = InterlockedCompareExchange(&qPtr->block_flag, 0, 1);
if (oldv == 1)
return entry; // nobody saw us, we can just exit with the value
}
// Either we don't have an entry, or we are forced to wait because
// queue_push saw our block flag. So do the wait
NtWaitForKeyedEvent(qPtr->keyed_event, (PVOID)qPtr, FALSE, NULL);
// block_flag has been reset by queue_push
if (!entry)
entry = InterlockedPopEntrySList(&qPtr->slist);
assert(entry);
return entry;
}
您还可以使用类似的协议，使用细长读写锁和条件变量，以及无锁快速路径。这些是键控事件的包装器，因此与直接使用键控事件相比，它们可能会产生更多的开销。
你试过有条件等待吗？当队列变为空时，只需开始等待新作业即可。将作业放入队列的线程应触发信号。这样，您仅在队列为空时才使用锁。
https://computing.llnl.gov/tutorials/pthreads/#ConditionVariables
您可以使用 sigwait（） 函数使线程休眠。 您可以使用pthread_kill唤醒线程。 这比条件变量快得多。
您可以在等待时添加睡眠。只需选择您愿意等待的最大时间，然后执行类似操作（伪代码，因为我不记得 pthread 语法）：
WAIT_TIME = 100; // Set this to whatever you're happy with
while(loop_condition) {
thing = get_from_queue()
if(thing == null) {
sleep(WAIT_TIME);
} else {
handle(thing);
}
}
即使是像 100 毫秒睡眠这样短暂的事情也应该显着降低 CPU 使用率。我不确定上下文切换在什么时候会比忙于等待更糟糕。