单生产者单消费者队列中的内存屏障

Question

过去几周我阅读了大量关于内存模型、编译器重新排序、CPU 重新排序、内存屏障和无锁编程的知识，我想我现在让自己陷入了困惑。我写了一个单一的生产者单一的消费者队列，并试图找出我需要内存屏障的地方，以及是否需要一些操作是原子的。我的单生产者单消费者队列如下：

typedef struct queue_node_t {
    int data;
    struct queue_node_t *next;
} queue_node_t;

// Empty Queue looks like this:
// HEAD TAIL
//   |    |
// dummy_node

// Queue: insert at TAIL, remove from HEAD
//    HEAD           TAIL
//     |              |
// dummy_node -> 1 -> 2 -> NULL

typedef struct queue_t {
    queue_node_t *head; // consumer consumes from head
    queue_node_t *tail; // producer adds at tail
} queue_t;

queue_node_t *alloc_node(int data) {
    queue_node_t *new_node = (queue_node_t *)malloc(sizeof(queue_node_t));
    new_node->data = data;
    new_node->next = NULL;
    return new_node;
}

queue_t *create_queue() {
    queue_t *new_queue = (queue_t *)malloc(sizeof(queue_t));
    queue_node_t *dummy_node = alloc_node(0);
    dummy_node->next = NULL;
    new_queue->head = dummy_node;
    new_queue->tail = dummy_node;
    // 1. Do we need any kind of barrier to make sure that if the
    // thread that didn't call this performs a queue operation
    // and happens to run on a different CPU that queue structure
    // is fully observed by it? i.e. the head and tail are properly
    // initialized
    return new_queue;
}

// Enqueue modifies tail
void enqueue(queue_t *the_queue, int data) {
    queue_node_t *new_node = alloc_node(data);
    // insert at tail
    new_node->next = NULL;

    // Let us save off the existing tail
    queue_node_t *old_tail = the_queue->tail;

    // Make the new node the new tail
    the_queue->tail = new_node;

    // 2. Store/Store barrier needed here?

    // Link in the new node last so that a concurrent dequeue doesn't see
    // the node until we're done with it
    // I don't know that this needs to be atomic but it does need to have
    // release semantics so that this isn't visible until prior writes are done
    old_tail->next = the_queue->tail;
    return;
}

// Dequeue modifies head
bool dequeue(queue_t *the_queue, int *item) {
    // 3. Do I need any barrier here to make sure if an enqueue already happened
    // I can observe it? i.e., if an enqueue was called on 
    // an empty queue by thread 0 on CPU0 and dequeue is called
    // by thread 1 on CPU1
    // dequeue the oldest item (FIFO) which will be at the head
    if (the_queue->head->next == NULL) {
        return false;
    }
    *item = the_queue->head->next->data;
    queue_node_t *old_head = the_queue->head;
    the_queue->head = the_queue->head->next;
    free(old_head);
    return true;
}

这是我上面代码中与 cmets 对应的问题：

1. 在create_queue() 中，我返回之前是否需要某种屏障？我想知道我是否从在 CPU0 上运行的线程 0 调用此函数，然后使用恰好在 CPU1 上运行的线程 1 中返回的指针，线程 1 是否可能看到未完全初始化的 queue_t 结构？李>
2. 我是否需要在 enqueue() 中设置屏障以确保在所有新节点的字段都初始化之前，新节点不会链接到队列中？
3. 我需要在dequeue() 中设置屏障吗？我觉得没有一个是正确的，但如果我想确保看到任何已完成的队列，我可能需要一个。

更新：我试图用代码中的 cmets 说明清楚，但这个队列的 HEAD 总是指向一个虚拟节点。这是一种常见的技术，它使得生产者只需要访问 TAIL，而消费者只需要访问 HEAD。一个空队列将包含一个虚拟节点，dequeue() 总是返回 HEAD 之后的节点，如果有的话。当节点出队时，虚拟节点前进并且之前的“虚拟”被释放。

Answer 1

首先，这取决于您的特定硬件架构、操作系统、语言等。

1.) 不。因为无论如何你都需要一个额外的屏障来将指针传递给另一个线程

2.) 是的，old_tail->next = the_queue->tail需要在the_queue->tail = new_node之后执行

3.) 它不会有任何效果，因为障碍之前没有任何东西，但理论上你可能需要在enqueue() 中的old_tail->next = the_queue->tail 之后设置障碍。编译器不会在函数之外重新排序，但 CPU 可能会做类似的事情。 （非常不可能，但不是 100% 肯定）

OT：由于您已经在进行一些微优化，您可以为缓存添加一些填充

typedef struct queue_t {
    queue_node_t *head; // consumer consumes from head
    char cache_pad[64]; // head and tail shouldnt be in the same cache-line(->64 Byte)
    queue_node_t *tail; // producer adds at tail
} queue_t;

如果你真的有足够的内存可以浪费，你可以做这样的事情

typedef struct queue_node_t {
    int data;
    struct queue_node_t *next;
    char cache_pad[56]; // sizeof(queue_node_t) == 64; only for 32Bit
} queue_node_t;