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

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;
}

这是我上面代码中的注释所对应的问题:

在create_queue()中，返回之前我需要某种障碍吗？我想知道我是否从运行在CPU0上的线程0调用此函数，然后使用恰好在CPU1上运行的线程1中返回的指针吗，线程1是否可能看到未完全初始化的queue_t结构？

是否需要在enqueue()中设置屏障，以确保在初始化新节点的所有字段之前，不会将新节点链接到队列中？

我需要dequeue()中的障碍物吗？我觉得没有人是正确的，但是如果我想确保看到任何已完成的入队，则可能需要一个人。

更新:我试图通过代码中的注释使之清楚，但是此队列的HEAD始终指向虚拟节点。这是一种通用技术，它使生产者仅需访问TAIL，而消费者仅需访问HEAD。空队列将包含一个虚拟节点，并且 dequeue()始终返回HEAD之后的节点(如果有)。随着节点出队，虚拟节点前进，并且先前的“虚拟”被释放。

Answer 1

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

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

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

3.)
这将没有任何效果，因为在障碍之前没有任何东西，但是从理论上讲，您可能需要在old_tail->next = the_queue->tail中的enqueue()之后添加障碍。编译器不会在函数外部重新排序，但是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;