golang disruptor lockfree 高性能无锁队列的设计实现原理

rfyiamcool

腾讯的老哥在社区中开源了 golang lockfree 的库，本人看到该库的设计很有意思，就参与该项目的开发改进，已经是该项目的 contributor 了. 该库使用 golang 开发，相比社区中其他 golang lockfree 来说，api 更友好，性能更好，其提供了各种的休眠阻塞策略，避免过度的 spin 引发 cpu 开销升高。

项目地址: https://github.com/bruceshao/lockfree

lockfree queue 的基本原理还是好理解的，设计上借鉴参考了无锁队列的标杆 java disruptor，另外在代码中很多性能的优化点也是参照了 java disruptor。

lockfree queue 无锁的设计大概流程是这样，首先需要一个原子递增的发号器。生产者写数据时，先拿到一个 seq 序号，通过位运算找到 ringbuffer 的位置，如何 ringbuffer 还有空余空间，只需写到 ringbuffer 对应位置即可，如果空间已满，则需要等待。而读取数据只需判断对应结构的 seq 跟 consumer seq 是否一致即可。

什么场景下需要使用 lockfree 无锁队列?

就 go lockfree 和 disruptor 的设计来说，首先对性能很敏感，另外 consumer 为计算密集型或非阻塞操作，比如你使用 disruptor 队列做通知队列，consumer 收到事件后，进行 http 请求操作，这显然是无法体现 lockfree 队列的性能。反而因为 consumer 消费过慢，引发 producer 和 consumer 的 spin 自旋开销。

我的理解，性能敏感且 consumer 处理的足够快的场景可以使用 lockfree 无锁队列。

lockfree vs channel 的性能表现

如下所述.

  1. 在goroutine数量比较小时，lockfree 和 channel 性能差别不明显；
  2. 当goroutine打到一定数量（大于1000）后，lockfree无论从时间还是 QR 都远远超过chan；

一句话，大多数场景不需要 lockfree 无锁队列，除非追求机制的性能体验。平时用 golang channel 足矣了，如果 golang channel 出现并发的性能瓶颈，其实也可以变通下，切分多个 channel 来分担 mutex 锁竞争冲突，以提高 channel 的读写并行吞吐。

lockfree 如何使用 ?

该库的使用很简单。先实例化 lockfree 队列对象，然后就可并发写，读则是注册一个回调方法，lockfree consumer 消费数后会回调注册的方法。

package main

import (
    "fmt"
    "sync"
    "sync/atomic"
    "time"

    "github.com/bruceshao/lockfree"
)

var (
    goSize    = 10000
    sizePerGo = 10000

    total = goSize * sizePerGo
)

func main() {
    // lockfree计时
    now := time.Now()

    // 创建事件处理器
    handler := &eventHandler[uint64]{
        signal: make(chan struct{}, 0),
        now:    now,
    }

    // 创建消费端串行处理的Lockfree
    lf := lockfree.NewLockfree[uint64](
        1024*1024,
        handler,
        lockfree.NewSleepBlockStrategy(time.Millisecond),
    )

    // 启动Lockfree
    if err := lf.Start(); err != nil {
        panic(err)
    }

    // 获取生产者对象
    producer := lf.Producer()

    // 并发写入
    var wg sync.WaitGroup
    wg.Add(goSize)
    for i := 0; i < goSize; i++ {
        go func(start int) {
            for j := 0; j < sizePerGo; j++ {
                err := producer.Write(uint64(start*sizePerGo + j + 1))
                if err != nil {
                    panic(err)
                }
            }
            wg.Done()
        }(i)
    }

    // wait for producer
    wg.Wait()

    fmt.Printf("producer has been writed, write count: %v, time cost: %v \n", total, time.Since(now).String())

    // wait for consumer
    handler.wait()

    // 关闭Lockfree
    lf.Close()
}

type eventHandler[T uint64] struct {
    signal   chan struct{}
    gcounter uint64
    now      time.Time
}

func (h *eventHandler[T]) OnEvent(v uint64) {
    cur := atomic.AddUint64(&h.gcounter, 1)
    if cur == uint64(total) {
        fmt.Printf("eventHandler has been consumed already, read count: %v, time cose: %v\n", total, time.Since(h.now))
        close(h.signal)
        return
    }

    if cur%10000000 == 0 {
        fmt.Printf("eventHandler consume %v\n", cur)
    }
}

func (h *eventHandler[T]) wait() {
    <-h.signal
}

go lockfree 源码分析

实例化 lockfree 对象

type Lockfree[T any] struct {
    writer   *Producer[T]
    consumer *consumer[T]
    status   int32
}

func NewLockfree[T any](capacity int, handler EventHandler[T], blocks blockStrategy) *Lockfree[T] {
    // 重新计算正确的容量
    capacity = minSuitableCap(capacity)

    // 发号器
    seqer := newSequencer(capacity)

    // 实例化 ringbuffer，队列是环形数组
    rbuf := newRingBuffer[T](capacity)

    // 实例化定义消费者
    cmer := newConsumer[T](rbuf, handler, seqer, blocks)

    // 实例化生产者
    writer := newProducer[T](seqer, rbuf, blocks)
    return &Lockfree[T]{
        writer:   writer,
        consumer: cmer,
        status:   READY,
    }
}

启动 lockfree

Start() 用来启动 lockfree，对于 consumer 来说会启动一个消费协程消费 ringbuffer 队列中的数据，而对于 producer 来说，只是把 status 只为 running 即可。

func (d *Lockfree[T]) Start() error {
    // 保证并发安全
    if atomic.CompareAndSwapInt32(&d.status, READY, RUNNING) {
        // 启动消费者
        if err := d.consumer.start(); err != nil {
            // 启动失败，重置 status
            atomic.CompareAndSwapInt32(&d.status, RUNNING, READY)
            return err
        }

        // 启动生产者
        if err := d.writer.start(); err != nil {
            // 恢复现场
            atomic.CompareAndSwapInt32(&d.status, RUNNING, READY)
            return err
        }
        return nil
    }
    return fmt.Errorf(StartErrorFormat, "Disruptor")
}

发号器

lockfree queue 通常都需要一个原子发号器的，其实就是递增的 uint64。lockfree 库里为了避免并发下带来的伪共享问题，使用填充的方式让 uint64 独占缓存行 cache line。

type sequencer struct {
    wc       *cursor
    rc       uint64 // 读取游标，因为该值仅会被一个g修改，所以不需要使用cursor
    capacity uint64
}

type cursor struct {
    // cache line 填充，这里适配了 64 和 128 字节的缓存行。
    p1, p2, p3, p4, p5, p6, p7       uint64
    v                                uint64
    p9, p10, p11, p12, p13, p14, p15 uint64
}

func newSequencer(capacity int) *sequencer {
    return &sequencer{
        wc:       newCursor(),
        rc:       1,
        capacity: uint64(capacity),
    }
}

// nextRead 获取下个要读取的位置
// 使用原子操作解决data race问题
func (s *sequencer) nextRead() uint64 {
    return atomic.LoadUint64(&s.rc)
}

func (s *sequencer) readIncrement() uint64 {
    return atomic.AddUint64(&s.rc, 1)
}

生产者

初始化 producer 对象

type Producer[T any] struct {
    seqer    *sequencer
    rbuf     *ringBuffer[T]
    blocks   blockStrategy
    capacity uint64
    status   int32
}

func newProducer[T any](seqer *sequencer, rbuf *ringBuffer[T], blocks blockStrategy) *Producer[T] {
    return &Producer[T]{
        seqer:    seqer,
        rbuf:     rbuf,
        blocks:   blocks,
        capacity: rbuf.cap(),
        status:   READY,
    }
}

func (q *Producer[T]) start() error {
    if atomic.CompareAndSwapInt32(&q.status, READY, RUNNING) {
        return nil
    }
    return fmt.Errorf(StartErrorFormat, "Producer")
}

写数据

先获取写的 seq 序号，判断 ringbuffer 是否可写，如可写，直接写入到对应的 ringbuffer 位置即可，不可写则需要自旋判断，直到可写为止。

func (q *Producer[T]) Write(v T) error {
    if q.closed() {
        return ClosedError
    }

    // 获取写的位置
    next := q.seqer.wc.increment()

    for {
        r := atomic.LoadUint64(&q.seqer.rc) - 1

        // 判断 ringbuffer 是否已满，consumer 的 seq + ringbuffer 长度为判断条件.
        if next <= r+q.capacity {
            // 可以写入数据，将数据写入到指定位置
            q.rbuf.write(next-1, v)
            // 释放，防止消费端阻塞
            q.blocks.release()
            return nil
        }

        // 队列已满，则暂时让出 cpu 资源，等待 runtime 下次调度，如果 consumer 消费不及时，那么这里会引发自旋问题。
        runtime.Gosched()

        // 再次判断是否已关闭
        if q.closed() {
            return ClosedError
        }
    }
}

如果消费者处理速度慢一些，那么队列会无空闲位置，也就是已满。那么生产者无法 write，内部会触发 runtime.Gosched 退让资源和切出调度，如果一段时间内未消费，生产者必然会因为这个 spin 操作引发 cpu 开销飙高。

所以，上面一直有说 lockfree / disruptor 的使用场景，消费者要足够的快。

runtime.Gosched 是暂停当前的 G，然后把该 G 放到队列尾部，重新切 g0 选择新的 G 进行调度。这个设计很好，社区中一些开源库有使用 atomic cas 实现的锁，当无法 cas 时，会执行 gosched 切换调度。

但就 lockfree 场景来说，如果 consumer 消费的不够快，那么 producer write 失败后会一直频繁的 gosched。这时候 runq 只有 producer，runtime goready 调度起来后又无法写，这个操作一直循环下来会使 cpu 飙高满载。

所以我这边又做了进一步的调整，写入时可以选择是否 sleep 休眠一下。经过压测得出 sleep 开销如下。

  • 10us 的空闲时间，cpu开销在 3% 左右;
  • 5us 的空间实现，cpu 开销在 10% 左右。
  • < 5us 的空间实现，cpu 开销接近 100%。
// WriteSleep 当 ringbuffer 已满时，先 cpu pause 则 sched 让出资源，最后进行 sleep 休眠，以此避免频繁 spin 引发的 cpu 开销。
func (q *Producer[T]) WriteSleep(v T, dur time.Duration) error {
    return q.write(v, dur)
}

func (q *Producer[T]) write(v T, dur time.Duration) error {
    if q.closed() {
        return ClosedError
    }

    next := q.seqer.wc.increment()
    var i = 0
    for {
        // 判断是否可以写入
        r := atomic.LoadUint64(&q.seqer.rc) - 1
        if next <= r+q.capacity {
            // 可以写入数据，将数据写入到指定位置
            q.rbuf.write(next-1, v)
            // 释放，防止消费端阻塞
            q.blocks.release()
            return nil
        }

        if i < spin {
            procyield(30)
        } else if i < spin+passiveSpin {
            runtime.Gosched()
        } else {
            time.Sleep(dur)
            i = 0
        }
        i++

        // 再次判断是否已关闭
        if q.closed() {
            return ClosedError
        }
    }
}

😁 关于 lockfree write 写失败 spin 引发的 cpu 过高问题，已经给作者提价 pull request (pr)，但作者还没给合并，说要考虑用户的选择再考虑下。我表示不理解。

https://github.com/bruceshao/lockfree/pull/15

consumer 消费者

初始化 consumer

type consumer[T any] struct {
    status int32 // 运行状态
    rbuf   *ringBuffer[T]
    seqer  *sequencer
    blocks blockStrategy
    hdl    EventHandler[T]
}

func newConsumer[T any](rbuf *ringBuffer[T], hdl EventHandler[T], sequer *sequencer, blocks blockStrategy) *consumer[T] {
    return &consumer[T]{
        rbuf:   rbuf,
        seqer:  sequer,
        hdl:    hdl,
        blocks: blocks,
        status: READY,
    }
}

func (c *consumer[T]) start() error {
    // 保证单例
    if atomic.CompareAndSwapInt32(&c.status, READY, RUNNING) {
        go c.handle()
        return nil
    }
    return fmt.Errorf(StartErrorFormat, "Consumer")
}

consumer 消费数据

func (c *consumer[T]) handle() {
    // 获取 read 的 位置
    rc := c.seqer.nextRead()
    for {
        if c.closed() {
            return
        }
        var i = 0
        for {
            if c.closed() {
                return
            }

            // 位置的对象是否可读，可读获取对象的数据.
            if v, exist := c.rbuf.contains(rc - 1); exist {
                // 既然可读了，则递增 read 位置，下次轮询新位置.
                rc = c.seqer.readIncrement()

                // 调用注册的回调方法，把数据传递过去.
                c.hdl.OnEvent(v)
                i = 0
                break
            }

            // 当无数据可读时, 可按照策略进行阻塞等待.
            // 当轮询小于 4 次时, 每次执行 30 次 cpu pause 
            if i < spin {
                procyield(30)
            } else if i < spin+passiveSpin {
                // 空闲阻塞策略，当轮询在 4-6 次时, 每次执行 runtime sched 让出调度资源
                runtime.Gosched()
            } else {
                // 自定义阻塞策略
                c.blocks.block()

                // 重置为 0
                i = 0
            }
            i++
        }
    }
}

判断 ringbuffer 某个 seq 是否可读 ?

lockfree 开始的设计使用 bitmap 判断 ringbuffer 各个位置的数据情况，后来改成在每个 entry 对象里存放了 producer 写入的 seq。这样判断位置是否可读只需判断 entry.seq 是否一致即可。

func (r *ringBuffer[T]) contains(c uint64) (T, bool) {
    x := &r.buf[c&r.capMask]
    if atomic.LoadUint64(&x.c) == c+1 {
        v := x.val
        return v, true
    }
    return r.tDefault, false
}

总结

lockfree queue 的设计还是很简单，首先需要一个原子递增的发号器，生产者并发写下，先拿到一个 seq 序号，然后 ringbuffer 不满下，只需写到 ringbuffer 对应位置即可。而读取数据只需判断对应结构的 seq 跟 consumer seq 是否一致即可。

至于 lockfree 的性能表现，如下所述.

  1. 在goroutine数量比较小时，lockfree和chan性能差别不明显；
  2. 当goroutine打到一定数量（大于1000）后，lockfree无论从时间还是QR都远远超过chan；

一句话，大多数场景不需要 lockfree 无锁队列，除非追求机制的性能体验。平时用 golang channel 足矣了，如果 golang channel 出现并发的性能瓶颈，其实也可以变通下，切分多个 channel 来分担 mutex 锁竞争冲突，以提高 channel 的读写并行吞吐。

大家觉得文章对你有些作用！ 如果想赏钱，可以用微信扫描下面的二维码，感谢!
另外再次标注博客原地址  xiaorui.cc