【发布时间】：2018-07-18 07:45:56
【问题描述】：
为了提高我的 C 技能，我实现了一个线程安全且无锁的队列。该算法来自 Maurice Herlihy 和 Nir Shavit 所著的《多处理器编程的艺术》一书的第 10.5 章，顺便说一句，这是一本很棒的书。
到目前为止，一切正常，但我需要帮助解决以下问题：
问题
free(first) 行在 lfq_deq() 方法中被注释掉，因为如果队列被多个 dequeuer 使用，它可能会导致段错误。如果线程 T1 和 T2 正在出队并且 T1 释放节点而 T2 仍在使用它，则 T2 将产生段错误。
释放此内存的优雅方法是什么？由于我不重用节点，我应该没有 ABA 问题，对吧？还是您认为重用节点并为 ABA 问题实施已知解决方案更容易？
lfq.h
标题提供了一个简单的主要测试方法。
#pragma once
#include <stdlib.h>
typedef struct Node {
void* data;
struct Node* next;
} lfq_node_t;
typedef struct Queue {
lfq_node_t* head;
lfq_node_t* tail;
} lfq_t;
lfq_t* lfq_new();
void lfq_free(lfq_t* q);
void lfq_enq(lfq_t* q, void* data);
void* lfq_deq(lfq_t* q);
lfq.c
#include "lfq.h"
#include <pthread.h>
#include <stdio.h>
#define CAS(a, b, c) __sync_bool_compare_and_swap(a, b, c)
lfq_t* lfq_new() {
lfq_t* q = malloc(sizeof(*q));
lfq_node_t* sentinel = malloc(sizeof(*sentinel));
sentinel->data = sentinel->next = NULL;
q->head = q->tail = sentinel;
return q;
}
void lfq_free(lfq_t* q) {
lfq_node_t *next, *node = q->head;
while (node != NULL) {
next = node->next;
free(node);
node = next;
}
free(q);
}
void lfq_enq(lfq_t* q, void* data) {
lfq_node_t *node, *last, *next;
node = malloc(sizeof(*node));
node->data = data;
node->next = NULL;
while (1) {
last = q->tail;
next = last->next;
if (last == q->tail) {
if (next == NULL) {
if (CAS(&(last->next), next, node)) {
CAS(&(q->tail), last, node);
return;
}
} else {
CAS(&(q->tail), last, next);
}
}
}
}
void* lfq_deq(lfq_t* q) {
lfq_node_t *first, *last, *next;
while (1) {
first = q->head;
last = q->tail;
next = first->next;
if (first == q->head) {
if (first == last) {
if (next == NULL) return NULL;
CAS(&(q->tail), last, next);
} else {
void* data = first->next->data;
if (CAS(&(q->head), first, next)) {
// free(first);
return data;
}
}
}
}
}
main.c
一个简单的测试队列的main方法：
#include "lfq.h"
#include <stdio.h>
int main() {
int values[] = {1, 2, 3, 4, 5};
lfq_t* q = lfq_new();
for (int i = 0; i < 5; ++i) {
printf("ENQ %i\n", values[i]);
lfq_enq(q, &values[i]);
}
for (int i = 0; i < 5; ++i) printf("DEQ %i\n", *(int*)lfq_deq(q));
lfq_free(q);
return 0;
}
【问题讨论】：
-
循环缓冲区（固定大小的数组）是队列的常用解决方案。 Garbage-collection 通过解决释放问题，实际上使基于节点的链表队列更容易。
-
顺便说一句，您可能应该使用 C11
stdatomic而不是 GNU C 的 old (and not recommended anymore)__sync_bool_compare_and_swapbuiltin 的非原子变量。first = q->head;... stuff ...if (first == q->head)可以 CSE 两次读取q->head因为它不是原子的（或volatile，但认真地只是在你的结构中使用_Atomic成员。标准原子获取负载可以解决这个问题. C11 CAS给你内存位置的当前值以及一个bool结果，所以你不必在循环中重新读取它 -
谢谢！我会重写CAS指令等等。我还实现了一个基于数组的解决方案，但出于好奇，我对这个问题的答案很感兴趣。
-
Related: Lock Free stack implementation idea - currently broken 是关于基于节点的 stack （非常类似于队列），但 BeeOnRope 关于内存回收问题的回答可能也与您有关。
标签： c multithreading memory-management memory-leaks lock-free