MPMC Channel：Go channel 与 crossbeam-channel 的实现对比

// runtime/chan.go (Go 1.22+)
type hchan struct {
    qcount   uint           // 当前 buffer 中的元素数量
    dataqsiz uint           // 环形缓冲区容量（make 时指定）
    buf      unsafe.Pointer // 指向环形缓冲区的指针
    elemsize uint16         // 单个元素的大小
    closed   uint32         // 是否已关闭
    timer    *timer         // 用于 select 超时
    elemtype *_type         // 元素类型（用于 GC）
    sendx    uint           // 发送索引（下一个写入位置）
    recvx    uint           // 接收索引（下一个读取位置）
    recvq    waitq          // 等待接收的 goroutine 队列
    sendq    waitq          // 等待发送的 goroutine 队列
    lock     mutex          // 保护整个 hchan 的互斥锁
}

type waitq struct {
    first *sudog
    last  *sudog
}

type sudog struct {
    g       *g             // 被阻塞的 goroutine
    next    *sudog         // 链表下一个节点
    prev    *sudog         // 链表上一个节点
    elem    unsafe.Pointer // 发送/接收的数据指针
    isSelect bool          // 是否在 select 中使用
    success  bool          // 通信是否成功
    // ... 省略其他字段
}

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
    // 路径一：直接发送给等待中的接收者
    lock(&c.lock)
    if sg := c.recvq.dequeue(); sg != nil {
        // 有 goroutine 在等待接收，直接拷贝数据到它的栈上
        send(c, sg, ep, func() { unlock(&c.lock) }, 3)
        return true
    }

    // 路径二：buffer 未满，写入环形缓冲区
    if c.qcount < c.dataqsiz {
        qp := chanbuf(c, c.sendx)
        typedmemmove(c.elemtype, qp, ep)
        c.sendx++
        if c.sendx == c.dataqsiz {
            c.sendx = 0
        }
        c.qcount++
        unlock(&c.lock)
        return true
    }

    // 路径三：buffer 已满，阻塞当前 goroutine
    if !block {
        unlock(&c.lock)
        return false
    }
    gp := getg()
    mysg := acquireSudog()
    mysg.elem = ep
    mysg.g = gp
    c.sendq.enqueue(mysg)
    gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceBlockChanSend, 2)
    // ... 唤醒后的清理逻辑
    return true
}

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
    lock(&c.lock)

    // 路径一：有发送者在等待
    if sg := c.sendq.dequeue(); sg != nil {
        recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
        return true, true
    }

    // 路径二：buffer 非空
    if c.qcount > 0 {
        qp := chanbuf(c, c.recvx)
        if ep != nil {
            typedmemmove(c.elemtype, ep, qp)
        }
        typedmemclr(c.elemtype, qp)
        c.recvx++
        if c.recvx == c.dataqsiz {
            c.recvx = 0
        }
        c.qcount--
        unlock(&c.lock)
        return true, true
    }

    // 路径三：buffer 为空，阻塞
    if !block {
        unlock(&c.lock)
        return false, false
    }
    gp := getg()
    mysg := acquireSudog()
    mysg.elem = ep
    mysg.g = gp
    c.recvq.enqueue(mysg)
    gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanRecv, traceBlockChanRecv, 2)
    // ... 唤醒后的清理逻辑
    return true, true
}

func selectgo(cas0 *scase, order0 *uint16, pc0 *uintptr,
              nsends, nrecvs int, block bool) (int, bool) {
    // 第一步：将所有 case 随机打乱（pollorder）
    // 第二步：按 channel 地址排序（lockorder），避免死锁
    // 第三步：遍历所有 case，尝试非阻塞操作
    // 第四步：如果都不成功，将当前 goroutine 挂到所有 channel 的等待队列上
    // 第五步：gopark 阻塞
    // 第六步：被唤醒后，从所有其他 channel 的等待队列中摘除自己
}

// crossbeam-channel/src/flavors/mod.rs
pub(crate) enum Flavor<T> {
    Array(channel::Channel<T>),   // 有界：环形数组
    List(channel::Channel<T>),    // 无界：链表块
    Zero(channel::Channel<T>),    // 零容量：直接交接
}

// crossbeam-channel/src/flavors/array.rs
pub(crate) struct Channel<T> {
    head: CachePadded<AtomicUsize>,   // 消费者游标
    tail: CachePadded<AtomicUsize>,   // 生产者游标
    buffer: Box<[Slot<T>]>,           // 槽位数组
    cap: usize,                        // 容量
    one_lap: usize,                    // 一圈的步长（用于区分满/空）
    // ...
}

struct Slot<T> {
    stamp: AtomicUsize,               // 槽位标记（编码了 lap 信息）
    msg: UnsafeCell<MaybeUninit<T>>,  // 消息存储
}

// crossbeam-utils/src/cache_padded.rs
#[repr(align(128))]
pub struct CachePadded<T> {
    value: T,
}

pub(crate) fn send(&self, msg: T, deadline: Option<Instant>)
    -> Result<(), SendTimeoutError<T>>
{
    let backoff = Backoff::new();
    let mut tail = self.tail.load(Ordering::Relaxed);

    loop {
        // 从 tail 中解码出 index 和 lap
        let index = tail & (self.one_lap - 1);
        let lap = tail & !(self.one_lap - 1);

        // 读取目标槽位的 stamp
        let slot = unsafe { self.buffer.get_unchecked(index) };
        let stamp = slot.stamp.load(Ordering::Acquire);

        // 如果 stamp == tail，说明槽位空闲，可以写入
        if stamp == tail {
            // CAS 推进 tail
            let new_tail = if index + 1 < self.cap {
                tail + 1
            } else {
                lap.wrapping_add(self.one_lap)
            };

            match self.tail.compare_exchange_weak(
                tail, new_tail,
                Ordering::SeqCst, Ordering::Relaxed,
            ) {
                Ok(_) => {
                    // 写入消息
                    unsafe { slot.msg.get().write(MaybeUninit::new(msg)); }
                    // 更新 stamp，通知消费者
                    slot.stamp.store(tail + 1, Ordering::Release);
                    return Ok(());
                }
                Err(t) => {
                    tail = t;
                    backoff.spin();
                }
            }
        } else if stamp.wrapping_add(self.one_lap) == tail + 1 {
            // 槽位仍被占用，channel 已满
            // ... 处理阻塞或返回错误
        } else {
            // 被其他线程抢先，重新加载 tail
            backoff.spin();
            tail = self.tail.load(Ordering::Relaxed);
        }
    }
}

// crossbeam-channel/src/flavors/list.rs
struct Block<T> {
    next: AtomicPtr<Block<T>>,        // 指向下一个 block
    slots: [Slot<T>; BLOCK_CAP],      // 每个 block 32 个槽位
}

pub(crate) struct Channel<T> {
    head: CachePadded<Position<T>>,   // 消费端位置
    tail: CachePadded<Position<T>>,   // 生产端位置
}

struct Position<T> {
    index: AtomicUsize,               // 全局索引
    block: AtomicPtr<Block<T>>,       // 当前所在 block
}

pub(crate) struct Channel<T> {
    inner: Mutex<Inner>,
    // ...
}

struct Inner {
    senders: Waker,
    receivers: Waker,
    is_disconnected: bool,
}

// com.lmax.disruptor.RingBuffer
public final class RingBuffer<E> extends RingBufferFields<E> {
    // 左侧填充：7 个 long 字段 = 56 字节 + 对象头 8 字节 = 64 字节
    // 确保 cursor 独占一个 cache line
    public static final long INITIAL_CURSOR_VALUE = Sequence.INITIAL_VALUE;

    // RingBufferFields 中的关键字段
    private final long indexMask;        // capacity - 1，用于位运算取模
    private final Object[] entries;      // 预分配的事件数组
    protected final int bufferSize;
    protected final Sequencer sequencer; // 核心协调器
}

// lib/ring/rte_ring_core.h
struct rte_ring_headtail {
    volatile uint32_t head;  // 头指针
    volatile uint32_t tail;  // 尾指针
    RTE_STD_C11
    union {
        uint32_t single;     // 是否单生产者/消费者模式
    };
};

struct rte_ring {
    char name[RTE_RING_NAMESIZE];
    int flags;
    const struct rte_memzone *memzone;

    uint32_t size;           // 环大小（2 的幂）
    uint32_t mask;           // size - 1，用于位运算取模
    uint32_t capacity;       // 实际可用容量 = size - 1

    // 生产端和消费端分开放置，消除 false sharing
    char pad0 __rte_cache_aligned;
    struct rte_ring_headtail prod __rte_cache_aligned;
    char pad1 __rte_cache_aligned;
    struct rte_ring_headtail cons __rte_cache_aligned;
};

// 简化后的入队逻辑
static inline unsigned int
__rte_ring_mp_do_enqueue(struct rte_ring *r, void * const *obj_table,
                         unsigned int n, unsigned int *free_space)
{
    uint32_t prod_head, prod_next;
    uint32_t free_entries;

    // 第一阶段：CAS 抢占位置
    do {
        prod_head = r->prod.head;
        const uint32_t cons_tail = r->cons.tail;
        free_entries = r->capacity - (prod_head - cons_tail);
        if (n > free_entries) {
            // 空间不足
            return 0;
        }
        prod_next = prod_head + n;
        // CAS 推进 prod.head
    } while (!__atomic_compare_exchange_n(&r->prod.head,
              &prod_head, prod_next, 0,
              __ATOMIC_ACQUIRE, __ATOMIC_RELAXED));

    // 第二阶段：写入数据
    ENQUEUE_PTRS(r, &r[1], prod_head, obj_table, n, void *);
    rte_smp_wmb();  // 写屏障

    // 第三阶段：等待前面的生产者完成，然后推进 tail
    while (r->prod.tail != prod_head)
        rte_pause();
    r->prod.tail = prod_next;

    return n;
}

// 一次入队 32 个包
unsigned int nb_enq = rte_ring_mp_enqueue_burst(ring, (void **)pkts, 32, NULL);

// 一次出队 32 个包
unsigned int nb_deq = rte_ring_mc_dequeue_burst(ring, (void **)pkts, 32, NULL);

/* mpmc_queue.h - Lock-free bounded MPMC queue using per-slot stamps */
#ifndef MPMC_QUEUE_H
#define MPMC_QUEUE_H

#include <stdatomic.h>
#include <stdbool.h>
#include <stddef.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <sched.h>

/* 确保 cache line 对齐，消除 false sharing */
#define CACHE_LINE_SIZE 64
#define CACHE_ALIGNED __attribute__((aligned(CACHE_LINE_SIZE)))

/* 编译期断言：容量必须是 2 的幂 */
#define IS_POWER_OF_TWO(x) (((x) != 0) && (((x) & ((x) - 1)) == 0))

/* 单个槽位结构：stamp + 数据 */
typedef struct mpmc_slot {
    atomic_size_t stamp;       /* 槽位标记，编码当前轮次 */
    void         *data;        /* 用户数据指针 */
} mpmc_slot_t;

/* 队列主体结构 */
typedef struct mpmc_queue {
    /* 生产端游标，独占 cache line */
    CACHE_ALIGNED atomic_size_t tail;

    /* 消费端游标，独占 cache line */
    CACHE_ALIGNED atomic_size_t head;

    /* 不变量：初始化后只读 */
    CACHE_ALIGNED size_t        capacity;
    size_t        mask;         /* capacity - 1 */
    mpmc_slot_t  *slots;        /* 槽位数组 */
} mpmc_queue_t;

/* 自旋等待辅助函数 */
static inline void cpu_pause(void)
{
#if defined(__x86_64__) || defined(__i386__)
    __asm__ volatile("pause" ::: "memory");
#elif defined(__aarch64__)
    __asm__ volatile("yield" ::: "memory");
#else
    /* 通用回退 */
    atomic_thread_fence(memory_order_seq_cst);
#endif
}

/*
 * 初始化队列
 * capacity 必须是 2 的幂
 * 返回 0 表示成功，-1 表示失败
 */
static inline int mpmc_queue_init(mpmc_queue_t *q, size_t capacity)
{
    if (!q || !IS_POWER_OF_TWO(capacity)) {
        return -1;
    }

    q->capacity = capacity;
    q->mask     = capacity - 1;

    /* 分配对齐内存 */
    q->slots = (mpmc_slot_t *)aligned_alloc(
        CACHE_LINE_SIZE,
        sizeof(mpmc_slot_t) * capacity
    );
    if (!q->slots) {
        return -1;
    }

    /* 初始化每个槽位的 stamp 为其索引值 */
    for (size_t i = 0; i < capacity; i++) {
        atomic_store_explicit(&q->slots[i].stamp, i, memory_order_relaxed);
        q->slots[i].data = NULL;
    }

    atomic_store_explicit(&q->head, 0, memory_order_relaxed);
    atomic_store_explicit(&q->tail, 0, memory_order_relaxed);

    return 0;
}

/* 释放队列资源 */
static inline void mpmc_queue_destroy(mpmc_queue_t *q)
{
    if (q && q->slots) {
        free(q->slots);
        q->slots = NULL;
    }
}

/*
 * 非阻塞入队
 * 成功返回 true，队列满返回 false
 */
static inline bool mpmc_queue_push(mpmc_queue_t *q, void *data)
{
    size_t tail = atomic_load_explicit(&q->tail, memory_order_relaxed);

    for (;;) {
        mpmc_slot_t *slot = &q->slots[tail & q->mask];
        size_t stamp = atomic_load_explicit(&slot->stamp, memory_order_acquire);

        intptr_t diff = (intptr_t)stamp - (intptr_t)tail;

        if (diff == 0) {
            /* 槽位空闲，尝试 CAS 推进 tail */
            if (atomic_compare_exchange_weak_explicit(
                    &q->tail, &tail, tail + 1,
                    memory_order_acq_rel,
                    memory_order_relaxed))
            {
                /* CAS 成功，写入数据并更新 stamp */
                slot->data = data;
                atomic_store_explicit(
                    &slot->stamp, tail + 1, memory_order_release
                );
                return true;
            }
            /* CAS 失败，tail 已被更新，重试 */
        } else if (diff < 0) {
            /* 槽位仍被占用（尚未被消费），队列满 */
            return false;
        } else {
            /* 被其他生产者抢先推进了 tail，重新加载 */
            tail = atomic_load_explicit(&q->tail, memory_order_relaxed);
        }
    }
}

/*
 * 非阻塞出队
 * 成功返回 true 并通过 out 返回数据，队列空返回 false
 */
static inline bool mpmc_queue_pop(mpmc_queue_t *q, void **out)
{
    size_t head = atomic_load_explicit(&q->head, memory_order_relaxed);

    for (;;) {
        mpmc_slot_t *slot = &q->slots[head & q->mask];
        size_t stamp = atomic_load_explicit(&slot->stamp, memory_order_acquire);

        intptr_t diff = (intptr_t)stamp - (intptr_t)(head + 1);

        if (diff == 0) {
            /* 槽位有数据，尝试 CAS 推进 head */
            if (atomic_compare_exchange_weak_explicit(
                    &q->head, &head, head + 1,
                    memory_order_acq_rel,
                    memory_order_relaxed))
            {
                /* CAS 成功，读取数据并更新 stamp */
                *out = slot->data;
                atomic_store_explicit(
                    &slot->stamp, head + q->capacity,
                    memory_order_release
                );
                return true;
            }
            /* CAS 失败，head 已被更新，重试 */
        } else if (diff < 0) {
            /* 槽位尚未被写入，队列空 */
            return false;
        } else {
            /* 被其他消费者抢先推进了 head，重新加载 */
            head = atomic_load_explicit(&q->head, memory_order_relaxed);
        }
    }
}

/*
 * 阻塞入队（自旋等待）
 * 在实际使用中可以替换为更复杂的退避策略
 */
static inline void mpmc_queue_push_wait(mpmc_queue_t *q, void *data)
{
    unsigned int spin = 0;
    while (!mpmc_queue_push(q, data)) {
        if (spin < 16) {
            cpu_pause();  /* 短自旋 */
            spin++;
        } else {
            sched_yield();  /* 让出 CPU */
            spin = 0;
        }
    }
}

/*
 * 阻塞出队（自旋等待）
 */
static inline void *mpmc_queue_pop_wait(mpmc_queue_t *q)
{
    void *data = NULL;
    unsigned int spin = 0;
    while (!mpmc_queue_pop(q, &data)) {
        if (spin < 16) {
            cpu_pause();
            spin++;
        } else {
            sched_yield();
            spin = 0;
        }
    }
    return data;
}

/*
 * 查询当前队列中的元素数量（近似值）
 * 由于 head 和 tail 的读取不是原子的，
 * 在高并发场景下返回值可能不精确
 */
static inline size_t mpmc_queue_size(const mpmc_queue_t *q)
{
    size_t tail = atomic_load_explicit(
        (atomic_size_t *)&q->tail, memory_order_acquire
    );
    size_t head = atomic_load_explicit(
        (atomic_size_t *)&q->head, memory_order_acquire
    );
    return tail - head;
}

/* 查询队列是否为空 */
static inline bool mpmc_queue_empty(const mpmc_queue_t *q)
{
    return mpmc_queue_size(q) == 0;
}

/* 查询队列是否已满 */
static inline bool mpmc_queue_full(const mpmc_queue_t *q)
{
    return mpmc_queue_size(q) >= q->capacity;
}

#endif /* MPMC_QUEUE_H */

# 编译：mpmc_test.c 中 include "mpmc_queue.h"，使用 checksum 验证正确性
# 4 生产者各发送 100 万条消息，4 消费者接收并累加校验和
gcc -O2 -std=c11 -pthread -o mpmc_test mpmc_test.c
./mpmc_test
# 输出: Checksum in == Checksum out => PASS

package main

import (
    "fmt"
    "runtime"
    "sync"
    "time"
)

const (
    numProducers = 8
    numConsumers = 8
    itemsPerProd = 10_000_000
    chanSize     = 65536
)

func main() {
    runtime.GOMAXPROCS(runtime.NumCPU())
    ch := make(chan int64, chanSize)
    var wgProd, wgCons sync.WaitGroup

    start := time.Now()

    // 启动消费者
    wgCons.Add(numConsumers)
    for i := 0; i < numConsumers; i++ {
        go func() {
            defer wgCons.Done()
            for range ch {
                // 消费
            }
        }()
    }

    // 启动生产者
    wgProd.Add(numProducers)
    for i := 0; i < numProducers; i++ {
        go func(id int) {
            defer wgProd.Done()
            base := int64(id) * itemsPerProd
            for j := int64(0); j < itemsPerProd; j++ {
                ch <- base + j
            }
        }(i)
    }

    wgProd.Wait()
    close(ch)
    wgCons.Wait()

    elapsed := time.Since(start)
    total := int64(numProducers) * itemsPerProd
    throughput := float64(total) / elapsed.Seconds()
    fmt.Printf("Total: %d msgs in %v\n", total, elapsed)
    fmt.Printf("Throughput: %.2f M msg/s\n", throughput/1e6)
}

// 阻塞模式（默认）
ch <- msg  // 如果 ch 已满，goroutine 进入 sendq

// 非阻塞模式（select + default）
select {
case ch <- msg:
    // 发送成功
default:
    // channel 满，执行降级逻辑
}

// 手动实现 DropOld 策略
func sendDropOld(ch chan int, msg int) {
    select {
    case ch <- msg:
        return
    default:
        // channel 满，丢弃最旧消息
        select {
        case <-ch:
            // 读出一个旧消息丢弃
        default:
        }
        // 再次尝试写入
        select {
        case ch <- msg:
        default:
        }
    }
}

use crossbeam_channel::{bounded, TrySendError, SendTimeoutError};
use std::time::Duration;

let (tx, rx) = bounded(1024);

// 阻塞发送
tx.send(msg).unwrap();

// 非阻塞发送
match tx.try_send(msg) {
    Ok(()) => { /* 成功 */ }
    Err(TrySendError::Full(msg)) => { /* 队列满，msg 被返回 */ }
    Err(TrySendError::Disconnected(msg)) => { /* 接收端已断开 */ }
}

// 带超时的发送
match tx.send_timeout(msg, Duration::from_millis(100)) {
    Ok(()) => { /* 成功 */ }
    Err(SendTimeoutError::Timeout(msg)) => { /* 超时 */ }
    Err(SendTimeoutError::Disconnected(msg)) => { /* 断开 */ }
}

func safeWorker(ctx context.Context, ch <-chan int) {
    for {
        select {
        case <-ctx.Done():
            return
        case val, ok := <-ch:
            if !ok {
                return
            }
            _ = val
        }
    }
}

// 读取 stamp：需要 acquire 语义
// 确保看到 stamp 对应的数据写入
size_t stamp = atomic_load_explicit(&slot->stamp, memory_order_acquire);

// CAS tail：需要 acq_rel 语义
// acquire：看到最新的 tail 值
// release：让其他线程看到我们之前的操作
atomic_compare_exchange_weak_explicit(
    &q->tail, &tail, tail + 1,
    memory_order_acq_rel, memory_order_relaxed
);

// 更新 stamp：需要 release 语义
// 确保消费者在看到新 stamp 时也能看到写入的数据
atomic_store_explicit(&slot->stamp, tail + 1, memory_order_release);