无锁队列：Michael-Scott 算法与 ABA 问题

// memory_order_relaxed: 无序，仅保证原子性
atomic_store_explicit(&x, 1, memory_order_relaxed);

// memory_order_acquire: 读屏障——本次 load 之后的所有读写不会被重排到本次之前
val = atomic_load_explicit(&x, memory_order_acquire);

// memory_order_release: 写屏障——本次 store 之前的所有读写不会被重排到本次之后
atomic_store_explicit(&x, 1, memory_order_release);

// memory_order_seq_cst: 全序——所有线程看到的操作顺序一致
// 最强保证，也是 atomic 操作的默认内存序
atomic_store(&x, 1); // 等价于 memory_order_seq_cst

// 伪代码：以下三步在硬件层面是原子的
bool CAS(ptr, expected, desired) {
    if (*ptr == expected) {
        *ptr = desired;
        return true;
    } else {
        expected = *ptr; // 更新 expected 为当前值
        return false;
    }
}

_Atomic int x = 0;
int expected = 0;
int desired = 1;
// weak 版本，可能虚假失败，但在循环中更高效
while (!atomic_compare_exchange_weak(&x, &expected, desired)) {
    // expected 已被自动更新为 x 的当前值
    desired = expected + 1;
}

typedef struct node {
    void          *value;
    _Atomic(struct node *) next;
} node_t;

typedef struct queue {
    _Atomic(node_t *) head;
    _Atomic(node_t *) tail;
} queue_t;

void enqueue(queue_t *q, void *value) {
    node_t *node = malloc(sizeof(node_t));
    node->value = value;
    atomic_store_explicit(&node->next, NULL, memory_order_relaxed);

    node_t *tail, *next;
    for (;;) {
        tail = atomic_load_explicit(&q->tail, memory_order_acquire);
        next = atomic_load_explicit(&tail->next, memory_order_acquire);

        // 检查 tail 是否仍然一致
        if (tail == atomic_load_explicit(&q->tail, memory_order_relaxed)) {
            if (next == NULL) {
                // tail 确实是最后一个节点，尝试挂上新节点
                if (atomic_compare_exchange_weak_explicit(
                        &tail->next, &next, node,
                        memory_order_release, memory_order_relaxed)) {
                    // 成功，尝试推进 tail（失败也无所谓，别的线程会帮忙）
                    atomic_compare_exchange_strong_explicit(
                        &q->tail, &tail, node,
                        memory_order_release, memory_order_relaxed);
                    return;
                }
            } else {
                // tail 落后了，帮它推进
                atomic_compare_exchange_strong_explicit(
                    &q->tail, &tail, next,
                    memory_order_release, memory_order_relaxed);
            }
        }
    }
}

void *dequeue(queue_t *q) {
    node_t *head, *tail, *next;
    void *value;

    for (;;) {
        head = atomic_load_explicit(&q->head, memory_order_acquire);
        tail = atomic_load_explicit(&q->tail, memory_order_acquire);
        next = atomic_load_explicit(&head->next, memory_order_acquire);

        if (head == atomic_load_explicit(&q->head, memory_order_relaxed)) {
            if (head == tail) {
                if (next == NULL) {
                    // 队列为空
                    return NULL;
                }
                // tail 落后了，帮忙推进
                atomic_compare_exchange_strong_explicit(
                    &q->tail, &tail, next,
                    memory_order_release, memory_order_relaxed);
            } else {
                // 读取值必须在 CAS 之前
                value = next->value;
                if (atomic_compare_exchange_weak_explicit(
                        &q->head, &head, next,
                        memory_order_release, memory_order_relaxed)) {
                    // 成功：head 前移，old sentinel 可以释放
                    free(head);  // 注意：这里有 ABA 隐患！
                    return value;
                }
            }
        }
    }
}

// lock-free stack, push/pop
// 栈状态: top -> A -> B -> C

// 线程 1:
//   pop(): 读到 top=A, next=B
//   准备 CAS(top, A, B)
//   被挂起

// 线程 2:
//   pop() 成功，top=B, 释放 A
//   pop() 成功，top=C, 释放 B
//   push(A') -- malloc 返回地址恰好 == A，且 A'->next = C
//   栈状态: top -> A'(地址==A) -> C

// 线程 1 恢复:
//   CAS(top, A, B) -- 成功！因为 top 的地址值确实 == A
//   但 B 已经被 free 了！top 指向一个已释放的内存
//   程序崩溃或数据损坏

// 方案一：利用高位
// x86-64 用户态地址高 16 位为 0（canonical form）
// 可以借用这 16 位做版本号，最多 65536 个版本
typedef uintptr_t tagged_ptr_t;

#define PTR_MASK  0x0000FFFFFFFFFFFFULL
#define TAG_SHIFT 48
#define TAG_MASK  0xFFFF000000000000ULL

static inline void *get_ptr(tagged_ptr_t tp) {
    return (void *)(tp & PTR_MASK);
}

static inline uint16_t get_tag(tagged_ptr_t tp) {
    return (uint16_t)(tp >> TAG_SHIFT);
}

static inline tagged_ptr_t make_tagged(void *ptr, uint16_t tag) {
    return ((uintptr_t)tag << TAG_SHIFT) | ((uintptr_t)ptr & PTR_MASK);
}

// 方案二：128 位 CAS（更安全，版本号空间更大）
typedef struct {
    void    *ptr;
    uint64_t tag;
} tagged_ptr_128_t;

// GCC/Clang 提供 __int128 和 __sync_bool_compare_and_swap 支持
// 或者使用平台特定的 cmpxchg16b 内联汇编

// 概念性伪代码
thread_local hazard_ptr_t hp[MAX_HP_PER_THREAD];

void *safe_read(atomic_ptr *src) {
    void *p;
    do {
        p = atomic_load(src);
        hp[0] = p;  // 声明 "我在用 p"
        // 再读一次确认没变
    } while (p != atomic_load(src));
    return p;
}

void safe_retire(void *p) {
    // 加入待回收列表
    retired_list_add(p);
    if (retired_list_size() > THRESHOLD) {
        scan_and_reclaim();
    }
}

void scan_and_reclaim() {
    // 收集所有线程的 hazard pointer
    set_t hazard_set = collect_all_hazard_pointers();
    for (each p in retired_list) {
        if (p not in hazard_set) {
            free(p);
            retired_list_remove(p);
        }
    }
}

// 全局 epoch（三值轮转：0, 1, 2）
atomic_uint global_epoch = 0;

// 每线程结构
typedef struct {
    atomic_uint local_epoch;
    atomic_bool active;
    retire_list_t retired[3]; // 每个 epoch 一个退休列表
} thread_data_t;

void critical_enter(thread_data_t *td) {
    atomic_store(&td->active, true);
    atomic_store(&td->local_epoch,
                 atomic_load(&global_epoch));
    atomic_thread_fence(memory_order_seq_cst);
}

void critical_exit(thread_data_t *td) {
    atomic_store(&td->active, false);
}

void try_advance_epoch(void) {
    uint32_t e = atomic_load(&global_epoch);
    // 检查所有活跃线程是否都在当前 epoch
    for (each thread td) {
        if (atomic_load(&td->active) &&
            atomic_load(&td->local_epoch) != e) {
            return; // 还有线程在旧 epoch
        }
    }
    // 所有线程都跟上了，推进 epoch
    if (atomic_compare_exchange_strong(&global_epoch, &e, (e + 1) % 3)) {
        // 回收两个 epoch 前的所有节点
        reclaim_epoch((e + 1) % 3);
    }
}

// CAS：争抢同一个位置
while (!CAS(&tail, expected, expected + 1)) {
    expected = tail;  // 失败，重试
}

// FAA：每个线程拿到不同的位置
uint64_t my_pos = atomic_fetch_add(&tail, 1);
// my_pos 是我独占的，不需要重试

typedef struct {
    uint64_t safe : 1;   // 是否安全（非转圈回绕）
    uint64_t idx  : 31;  // 期望的逻辑索引
    uint64_t val  : 32;  // 实际存储的值（或指针的低 32 位）
} crq_slot_t;

/* ms_queue_tagged.c
 * Michael-Scott lock-free queue with tagged pointers (ABA-safe).
 * Requires: x86-64, GCC/Clang, C11 atomics.
 */

#define _GNU_SOURCE
#include <stdatomic.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <assert.h>

/* ---------- tagged pointer helpers ---------- */

/*
 * 在 x86-64 上，用户态虚拟地址的高 16 位为 0。
 * 我们借用这 16 位存放单调递增的版本号。
 * 这给出 65536 个版本周期，对于大多数场景足够。
 */
typedef uintptr_t tagged_ptr_t;

#define TAG_BITS  16
#define TAG_SHIFT 48
#define PTR_MASK  ((UINT64_C(1) << TAG_SHIFT) - 1)
#define TAG_MASK  (~PTR_MASK)

static inline void *tp_ptr(tagged_ptr_t tp)
{
    return (void *)(tp & PTR_MASK);
}

static inline uint16_t tp_tag(tagged_ptr_t tp)
{
    return (uint16_t)(tp >> TAG_SHIFT);
}

static inline tagged_ptr_t tp_make(void *ptr, uint16_t tag)
{
    return ((uintptr_t)tag << TAG_SHIFT) | ((uintptr_t)ptr & PTR_MASK);
}

static inline tagged_ptr_t tp_inc_tag(tagged_ptr_t tp, void *new_ptr)
{
    uint16_t new_tag = (uint16_t)(tp_tag(tp) + 1);
    return tp_make(new_ptr, new_tag);
}

/* ---------- node ---------- */

typedef struct ms_node {
    void                    *value;
    _Atomic(tagged_ptr_t)    next;  /* tagged pointer to next ms_node */
} ms_node_t;

static ms_node_t *node_alloc(void *value)
{
    ms_node_t *n = aligned_alloc(64, sizeof(ms_node_t));
    if (!n) {
        perror("aligned_alloc");
        abort();
    }
    n->value = value;
    atomic_store_explicit(&n->next, tp_make(NULL, 0), memory_order_relaxed);
    return n;
}

static void node_free(ms_node_t *n)
{
    free(n);
}

/* ---------- queue ---------- */

typedef struct ms_queue {
    _Atomic(tagged_ptr_t) head;  /* tagged pointer to sentinel */
    _Atomic(tagged_ptr_t) tail;  /* tagged pointer to last node */
    char _pad[64 - 2 * sizeof(_Atomic(tagged_ptr_t))]; /* avoid false sharing */
} ms_queue_t;

void msq_init(ms_queue_t *q)
{
    ms_node_t *sentinel = node_alloc(NULL);
    tagged_ptr_t tp = tp_make(sentinel, 0);
    atomic_store_explicit(&q->head, tp, memory_order_relaxed);
    atomic_store_explicit(&q->tail, tp, memory_order_relaxed);
}

void msq_destroy(ms_queue_t *q)
{
    /* drain all remaining nodes */
    void *val;
    while ((val = NULL, 1)) {
        tagged_ptr_t head_tp = atomic_load_explicit(&q->head,
                                                     memory_order_acquire);
        ms_node_t *head = tp_ptr(head_tp);
        tagged_ptr_t next_tp = atomic_load_explicit(&head->next,
                                                     memory_order_acquire);
        ms_node_t *next = tp_ptr(next_tp);
        if (next == NULL) {
            /* only sentinel left */
            node_free(head);
            break;
        }
        atomic_store_explicit(&q->head,
                              tp_inc_tag(head_tp, next),
                              memory_order_relaxed);
        node_free(head);
    }
}

/* ---------- enqueue ---------- */

void msq_enqueue(ms_queue_t *q, void *value)
{
    ms_node_t *node = node_alloc(value);
    tagged_ptr_t tail_tp, next_tp;

    for (;;) {
        tail_tp = atomic_load_explicit(&q->tail, memory_order_acquire);
        ms_node_t *tail = tp_ptr(tail_tp);
        next_tp = atomic_load_explicit(&tail->next, memory_order_acquire);
        ms_node_t *next = tp_ptr(next_tp);

        /* 确认 tail 没有被其他线程改变 */
        if (tail_tp != atomic_load_explicit(&q->tail, memory_order_relaxed))
            continue;

        if (next == NULL) {
            /* tail 确实指向最后一个节点 */
            tagged_ptr_t new_next = tp_inc_tag(next_tp, node);
            if (atomic_compare_exchange_weak_explicit(
                    &tail->next, &next_tp, new_next,
                    memory_order_release, memory_order_relaxed)) {
                /* 链接成功，推进 tail（best-effort） */
                tagged_ptr_t new_tail = tp_inc_tag(tail_tp, node);
                atomic_compare_exchange_strong_explicit(
                    &q->tail, &tail_tp, new_tail,
                    memory_order_release, memory_order_relaxed);
                return;
            }
        } else {
            /* tail 落后了，帮忙推进 */
            tagged_ptr_t new_tail = tp_inc_tag(tail_tp, next);
            atomic_compare_exchange_strong_explicit(
                &q->tail, &tail_tp, new_tail,
                memory_order_release, memory_order_relaxed);
        }
    }
}

/* ---------- dequeue ---------- */

bool msq_dequeue(ms_queue_t *q, void **out_value)
{
    tagged_ptr_t head_tp, tail_tp, next_tp;

    for (;;) {
        head_tp = atomic_load_explicit(&q->head, memory_order_acquire);
        tail_tp = atomic_load_explicit(&q->tail, memory_order_acquire);
        ms_node_t *head = tp_ptr(head_tp);
        next_tp = atomic_load_explicit(&head->next, memory_order_acquire);
        ms_node_t *next = tp_ptr(next_tp);

        /* 确认 head 没有被其他线程改变 */
        if (head_tp != atomic_load_explicit(&q->head, memory_order_relaxed))
            continue;

        if (head == tp_ptr(tail_tp)) {
            if (next == NULL) {
                /* 队列为空 */
                return false;
            }
            /* tail 落后，帮忙推进 */
            tagged_ptr_t new_tail = tp_inc_tag(tail_tp, next);
            atomic_compare_exchange_strong_explicit(
                &q->tail, &tail_tp, new_tail,
                memory_order_release, memory_order_relaxed);
        } else {
            /* 读取值必须在 CAS 之前——CAS 成功后 next 可能被其他线程释放 */
            *out_value = next->value;

            tagged_ptr_t new_head = tp_inc_tag(head_tp, next);
            if (atomic_compare_exchange_weak_explicit(
                    &q->head, &head_tp, new_head,
                    memory_order_release, memory_order_relaxed)) {
                /*
                 * 成功。旧 sentinel（head）可以释放。
                 * 因为使用了 tagged pointer，即使地址被重用，
                 * 其他线程的 CAS 也会因版本号不同而失败。
                 */
                node_free(head);
                return true;
            }
        }
    }
}

/* ---------- utility ---------- */

bool msq_is_empty(ms_queue_t *q)
{
    tagged_ptr_t head_tp = atomic_load_explicit(&q->head, memory_order_acquire);
    ms_node_t *head = tp_ptr(head_tp);
    tagged_ptr_t next_tp = atomic_load_explicit(&head->next, memory_order_acquire);
    return tp_ptr(next_tp) == NULL;
}

/* ---------- simple test ---------- */

#ifdef MS_QUEUE_MAIN
#include <pthread.h>

#define NUM_THREADS   8
#define OPS_PER_THREAD 100000

static ms_queue_t g_queue;
static _Atomic uint64_t g_enqueue_sum = 0;
static _Atomic uint64_t g_dequeue_sum = 0;

static void *producer(void *arg)
{
    uint64_t tid = (uint64_t)arg;
    uint64_t local_sum = 0;
    for (uint64_t i = 0; i < OPS_PER_THREAD; i++) {
        uint64_t val = tid * OPS_PER_THREAD + i + 1;
        msq_enqueue(&g_queue, (void *)val);
        local_sum += val;
    }
    atomic_fetch_add_explicit(&g_enqueue_sum, local_sum, memory_order_relaxed);
    return NULL;
}

static void *consumer(void *arg)
{
    (void)arg;
    uint64_t local_sum = 0;
    uint64_t count = 0;
    while (count < OPS_PER_THREAD) {
        void *val;
        if (msq_dequeue(&g_queue, &val)) {
            local_sum += (uint64_t)val;
            count++;
        }
    }
    atomic_fetch_add_explicit(&g_dequeue_sum, local_sum, memory_order_relaxed);
    return NULL;
}

int main(void)
{
    msq_init(&g_queue);

    pthread_t producers[NUM_THREADS];
    pthread_t consumers[NUM_THREADS];

    for (uint64_t i = 0; i < NUM_THREADS; i++) {
        pthread_create(&producers[i], NULL, producer, (void *)i);
        pthread_create(&consumers[i], NULL, consumer, NULL);
    }
    for (int i = 0; i < NUM_THREADS; i++) {
        pthread_join(producers[i], NULL);
        pthread_join(consumers[i], NULL);
    }

    printf("enqueue sum = %lu\n", atomic_load(&g_enqueue_sum));
    printf("dequeue sum = %lu\n", atomic_load(&g_dequeue_sum));
    assert(atomic_load(&g_enqueue_sum) == atomic_load(&g_dequeue_sum));
    printf("PASS: sums match.\n");

    msq_destroy(&g_queue);
    return 0;
}
#endif /* MS_QUEUE_MAIN */

gcc -std=c11 -O2 -pthread -DMS_QUEUE_MAIN ms_queue_tagged.c -o ms_queue_test
./ms_queue_test

/* mutex_queue.c - baseline for comparison */

#include <pthread.h>
#include <stdlib.h>
#include <stdbool.h>

typedef struct mnode {
    void *value;
    struct mnode *next;
} mnode_t;

typedef struct {
    mnode_t        *head;
    mnode_t        *tail;
    pthread_mutex_t lock;
} mutex_queue_t;

void mq_init(mutex_queue_t *q)
{
    mnode_t *sentinel = calloc(1, sizeof(mnode_t));
    q->head = q->tail = sentinel;
    pthread_mutex_init(&q->lock, NULL);
}

void mq_enqueue(mutex_queue_t *q, void *value)
{
    mnode_t *node = malloc(sizeof(mnode_t));
    node->value = value;
    node->next  = NULL;

    pthread_mutex_lock(&q->lock);
    q->tail->next = node;
    q->tail = node;
    pthread_mutex_unlock(&q->lock);
}

bool mq_dequeue(mutex_queue_t *q, void **out)
{
    pthread_mutex_lock(&q->lock);
    mnode_t *sentinel = q->head;
    mnode_t *next = sentinel->next;
    if (next == NULL) {
        pthread_mutex_unlock(&q->lock);
        return false;
    }
    *out = next->value;
    q->head = next;
    pthread_mutex_unlock(&q->lock);
    free(sentinel);
    return true;
}

// OpenJDK ConcurrentLinkedQueue.offer()
// 简化版
public boolean offer(E e) {
    final Node<E> newNode = new Node<>(e);
    for (Node<E> t = tail, p = t;;) {
        Node<E> q = p.next;
        if (q == null) {
            if (NEXT.compareAndSet(p, null, newNode)) {
                // CAS 成功，但不一定更新 tail
                if (p != t)
                    TAIL.compareAndSet(this, t, newNode);
                return true;
            }
        } else if (p == q) {
            // 遇到 sentinel 被移除的情况，从 head 重新开始
            p = (t != (t = tail)) ? t : head;
        } else {
            // p 不是最后节点，前进
            p = (p != t && t != (t = tail)) ? t : q;
        }
    }
}

// DPDK rte_ring 核心结构（简化）
struct rte_ring {
    uint32_t size;      /* 必须是 2 的幂 */
    uint32_t mask;      /* size - 1 */
    volatile uint32_t prod_head;
    volatile uint32_t prod_tail;
    volatile uint32_t cons_head;
    volatile uint32_t cons_tail;
    void *ring[0];      /* 柔性数组 */
};

// DPDK enqueue 的两阶段协议（MPMC，简化）
static inline int
rte_ring_mp_enqueue(struct rte_ring *r, void *obj)
{
    uint32_t prod_head, prod_next, cons_tail;

    /* 阶段一：CAS 预留一个 slot */
    do {
        prod_head = r->prod_head;
        cons_tail = r->cons_tail;
        if (prod_head - cons_tail >= r->size)
            return -ENOBUFS;  /* 满了 */
        prod_next = prod_head + 1;
    } while (!CAS(&r->prod_head, prod_head, prod_next));

    /* 阶段二：写入数据 */
    r->ring[prod_head & r->mask] = obj;

    /* 阶段三：等之前的生产者都完成，然后更新 prod_tail */
    while (r->prod_tail != prod_head)
        _mm_pause();
    r->prod_tail = prod_next;

    return 0;
}