go-diodes源码阅读--单生产者，单消费者无锁队列(1)

起因

今天在扩展我们自己的日志库，突然发现一个很有意思的说法(我们日志库基于zerolog做的二次开发)

Thread-safe, lock-free, non-blocking writer
If your writer might be slow or not thread-safe and you need your log producers to never get slowed down by a slow writer, you can use a diode.Writer as follow:
You will need to install code.cloudfoundry.org/go-diodes to use this feature.

第一个印象就是使用了go-diodes这个库，就达到了Thread-safe, lock-free, non-blocking writer效果。沟起了好奇心，然后看了下go-diodes库的代码只有不到1k左右，决定花1-2小时通读下，以期达到减少zerolog达来的黑洞感，也能更好地使用它。

从第一个生产者－－消费者demo入手

这是一个生产者，消费者模型。我们用channel也可以很容易地实现这个模型，通过阅读下面的源代码解密这么做和channel比哪些不同点。

// 1.NewOneToOne
d := diodes.NewOneToOne(1024, diodes.AlertFunc(func(missed int) {
    log.Printf("Dropped %d messages", missed)
}))

// writer
go func() {
    for i := 0; i < 2048; i++ {
        // Warning: Do not use i. By taking the address,
        // you would not get each value
        j := i
        // 2.Set
        d.Set(diodes.GenericDataType(&j))
    }
}()

// reader
poller := diodes.NewPoller(d)
for {
    i := poller.Next()
    fmt.Println(*(*int)(i))
}

单生产者－－单消费者无锁队列 数据结构

• 很精简，其中```writeIndex uint64 ``标记生产都的位置
• readIndex uint64 标记消费都的位置
• alerter Alerter 丢数据调用的回调函数
• buffer []unsafe.Pointer 存放任意类型的指针slice，这点学习下

type OneToOne struct {
    buffer     []unsafe.Pointer
    writeIndex uint64
    readIndex  uint64
    alerter    Alerter
}

初始化函数: 查看NewOneToOne初始化做了什么？

构造OneToOne结构体，buffer里面是unsafe.Pointer类型的slice。alerter是出错回调函数

func NewOneToOne(size int, alerter Alerter) *OneToOne {
    if alerter == nil {
        alerter = AlertFunc(func(int) {})
    }

    return &OneToOne{
        buffer:  make([]unsafe.Pointer, size),
        alerter: alerter,
    }
}

Write: .Set做了什么？

• idx := d.writeIndex % uint64(len(d.buffer)) 使用取余符号把slice当成环使用
• newBucket := &bucket{...} 生成指针变量并赋值
• d.writeIndex++移动下写的位置
• atomic.StorePointer(&d.buffer[idx], unsafe.Pointer(newBucket)) 线程安全的指针赋值方式，相当于下面的写法。

lock()
d.buffer[idx] = newBucket
unlock()

• 总结:使用atomic包里面的原子变量函数，实行高效操作。

// Set sets the data in the next slot of the ring buffer.
func (d *OneToOne) Set(data GenericDataType) {
    idx := d.writeIndex % uint64(len(d.buffer))

    newBucket := &bucket{
        data: data,
        seq:  d.writeIndex,
    }
    d.writeIndex++

    atomic.StorePointer(&d.buffer[idx], unsafe.Pointer(newBucket))
}

Write: .Set中的疑问－－如果读得慢写得快，如何？

Set里面没有任何判断条件，从代码上来看，Set里面的逻辑是直接覆盖没读取的数据(你没看错，数据没被及时读取就gg)。

Read: NewPoller初始化做了什么？

这里面初始化代码用了uber最佳编程实践描述过的一个技巧，感兴趣可自行查看。

func NewPoller(d Diode, opts ...PollerConfigOption) *Poller {
    p := &Poller{
        Diode:    d,
        interval: 10 * time.Millisecond,
        ctx:      context.Background(),
    }

    for _, o := range opts {
        o(p)
    }

    return p
}

Read: Next函数－－读取迭代器

迭代器代码，用了sleep防止cpu一样空转。context如果取消直接退出。真正取数据的还是TryNext函数，等会儿看下。

// Next polls the diode until data is available or until the context is done.
// If the context is done, then nil will be returned.
func (p *Poller) Next() GenericDataType {
    for {
        data, ok := p.Diode.TryNext()
        if !ok {
            if p.isDone() {
                return nil
            }

            time.Sleep(p.interval)
            continue
        }
        return data
    }
}

Read: TryNext函数

TryNext主要作用是从buffer里面读取数据。

1. idx := d.readIndex % uint64(len(d.buffer)) 计算读下标
2. result := (*bucket)(atomic.SwapPointer(&d.buffer[idx], nil)) 原子交换变量
   当作于

lock()
result := d.buffer[idx]
d.buffer[idx] = nil
unlock()

3. result.seq < d.readIndex 判断
4. d.alerter.Alert(int(dropped)) 会善意地提醒你，它丢了多少条数据

func (d *OneToOne) TryNext() (data GenericDataType, ok bool) {
    // Read a value from the ring buffer based on the readIndex.
    idx := d.readIndex % uint64(len(d.buffer))
    result := (*bucket)(atomic.SwapPointer(&d.buffer[idx], nil))

    // When the result is nil that means the writer has not had the
    // opportunity to write a value into the diode. This value must be ignored
    // and the read head must not increment.
    if result == nil {
        return nil, false
    }

    // When the seq value is less than the current read index that means a
    // value was read from idx that was previously written but has since has
    // been dropped. This value must be ignored and the read head must not
    // increment.
    //
    // The simulation for this scenario assumes the fast forward occurred as
    // detailed below.
    //
    // 5. The reader reads again getting seq 5. It then reads again expecting
    //    seq 6 but gets seq 2. This is a read of a stale value that was
    //    effectively "dropped" so the read fails and the read head stays put.
    //    `| 4 | 5 | 2 | 3 |` r: 7, w: 6
    //
    if result.seq < d.readIndex {
        return nil, false
    }

    // When the seq value is greater than the current read index that means a
    // value was read from idx that overwrote the value that was expected to
    // be at this idx. This happens when the writer has lapped the reader. The
    // reader needs to catch up to the writer so it moves its write head to
    // the new seq, effectively dropping the messages that were not read in
    // between the two values.
    //
    // Here is a simulation of this scenario:
    //
    // 1. Both the read and write heads start at 0.
    //    `| nil | nil | nil | nil |` r: 0, w: 0
    // 2. The writer fills the buffer.
    //    `| 0 | 1 | 2 | 3 |` r: 0, w: 4
    // 3. The writer laps the read head.
    //    `| 4 | 5 | 2 | 3 |` r: 0, w: 6
    // 4. The reader reads the first value, expecting a seq of 0 but reads 4,
    //    this forces the reader to fast forward to 5.
    //    `| 4 | 5 | 2 | 3 |` r: 5, w: 6
    //
    if result.seq > d.readIndex {
        dropped := result.seq - d.readIndex
        d.readIndex = result.seq
        d.alerter.Alert(int(dropped))
    }

    // Only increment read index if a regular read occurred (where seq was
    // equal to readIndex) or a value was read that caused a fast forward
    // (where seq was greater than readIndex).
    d.readIndex++
    return result.data, true
}

注意事项

这个只能在两个go程间流转数据。一个write go程，一个read go程。

TODO

继续完善