聊聊zerolog的diode.Writer

聊聊zerolog的diode.Writer

本文主要研究一下zerolog的diode.Writer

github.com/rs/zerolog@v1.20.0/diode/diode.go

// Writer is a io.Writer wrapper that uses a diode to make Write lock-free,
// non-blocking and thread safe.
type Writer struct {
    w    io.Writer
    d    diodeFetcher
    c    context.CancelFunc
    done chan struct{}
}

func NewWriter(w io.Writer, size int, pollInterval time.Duration, f Alerter) Writer {
    ctx, cancel := context.WithCancel(context.Background())
    dw := Writer{
        w:    w,
        c:    cancel,
        done: make(chan struct{}),
    }
    if f == nil {
        f = func(int) {}
    }
    d := diodes.NewManyToOne(size, diodes.AlertFunc(f))
    if pollInterval  0 {
        dw.d = diodes.NewPoller(d,
            diodes.WithPollingInterval(pollInterval),
            diodes.WithPollingContext(ctx))
    } else {
        dw.d = diodes.NewWaiter(d,
            diodes.WithWaiterContext(ctx))
    }
    go dw.poll()
    return dw
}

diode.Writer是一个lock-free,non-blocking及thread safe的Writer；它借助了diodes来实现；NewWriter会创建diode.Writer，并启动dw.poll()

github.com/rs/zerolog@v1.20.0/diode/diode.go

func (dw Writer) poll() {
    defer close(dw.done)
    for {
        d := dw.d.Next()
        if d == nil {
            return
        }
        p := *(*[]byte)(d)
        dw.w.Write(p)

        // Proper usage of a sync.Pool requires each entry to have approximately
        // the same memory cost. To obtain this property when the stored type
        // contains a variably-sized buffer, we add a hard limit on the maximum buffer
        // to place back in the pool.
        //
        // See https://golang.org/issue/23199
        const maxSize = 1  16 // 64KiB
        if cap(p) = maxSize {
            bufPool.Put(p[:0])
        }
    }
}

poll方法使用for循环执行dw.d.Next()及dw.w.Write(p)

github.com/rs/zerolog@v1.20.0/diode/diode.go

type diodeFetcher interface {
    diodes.Diode
    Next() diodes.GenericDataType
}

// Diode is any implementation of a diode.
type Diode interface {
    Set(GenericDataType)
    TryNext() (GenericDataType, bool)
}

diodeFetcher接口内嵌了Diode接口，定义了Next方法

github.com/rs/zerolog@v1.20.0/diode/internal/diodes/poller.go

// 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
    }
}

Poller实现了diodeFetcher接口的Next方法，它使用for循环，不断通过p.Diode.TryNext()来获取data

github.com/rs/zerolog@v1.20.0/diode/internal/diodes/many_to_one.go

// ManyToOne diode is optimal for many writers (go-routines B-n) and a single
// reader (go-routine A). It is not thread safe for multiple readers.
type ManyToOne struct {
    writeIndex uint64
    readIndex  uint64
    buffer     []unsafe.Pointer
    alerter    Alerter
}

// Set sets the data in the next slot of the ring buffer.
func (d *ManyToOne) Set(data GenericDataType) {
    for {
        writeIndex := atomic.AddUint64(d.writeIndex, 1)
        idx := writeIndex % uint64(len(d.buffer))
        old := atomic.LoadPointer(d.buffer[idx])

        if old != nil 
            (*bucket)(old) != nil 
            (*bucket)(old).seq  writeIndex-uint64(len(d.buffer)) {
            log.Println(Diode set collision: consider using a larger diode)
            continue
        }

        newBucket := bucket{
            data: data,
            seq:  writeIndex,
        }

        if !atomic.CompareAndSwapPointer(d.buffer[idx], old, unsafe.Pointer(newBucket)) {
            log.Println(Diode set collision: consider using a larger diode)
            continue
        }

        return
    }
}

// TryNext will attempt to read from the next slot of the ring buffer.
// If there is not data available, it will return (nil, false).
func (d *ManyToOne) 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
}

ManyToOne实现了Diode接口的Set和TryNext方法

func diodeDemo() {
    wr := diode.NewWriter(os.Stdout, 1000, 10*time.Millisecond, func(missed int) {
        fmt.Printf(Logger Dropped %d messages, missed)
    })
    log := zerolog.New(wr)
    log.Print(test)

    time.Sleep(1 * time.Second)
}

输出

{level:debug,message:test}

zerolog借助diodes提供了一个lock-free,non-blocking及thread safe的diode.Writer

• zerolog

聊聊zerolog的diode.Writer 相关文章

1. 聊聊zerolog的Formatter
   

   序 本文主要研究一下zerolog的Formatter Formatter github.com/rs/zerolog@v1.20.0/console.go // Formatter transforms the input into a formatted string.type Formatter func(interface{}) string Formatter接口定义了一个func用于将interface{}转换为st

2. 最全总结 | 聊聊 Python 办公自动化之 Word（中）
   

   1. 前言 上一篇文章，对 Word 写入数据的一些常见操作进行了总结 最全总结 | 聊聊 Python 办公自动化之 Word（上） 相比写入数据，读取数据同样很实用！ 本篇文章，将谈谈如何全面读取一个 Word 文档中的数据，并会指出一些要注意的点 2. 基本信息 我们同样

3. 最全总结 | 聊聊 Python 办公自动化之 Excel（中）
   

   最全总结 | 聊聊 Python 办公自动化之 Excel（中） 聊聊 Python 数据处理全家桶（Memca 篇） 点击上方“AirPython”，选择“加为星标” 第一时间关注 Python 技术干货！ 上一篇文章中，我们聊到使用xlrd、xlwt、xlutils 这一组合操作 Excel 的方法 最全总结 |

4. 中秋前夕，聊聊canvas
   

   中秋前夕，聊聊canvas 首先上图： 今天，我们前端群问了一个这样的问题，然后就开始了激烈的讨论。 那么下面咱们一起来看看这个问题，这个问题问了两个小问题： 1.如何在 canvas 上绘制多边形? 2.鼠标怎么选中绘制的某一个图形？ 那么咱们就来分为两个问题解

5. 面试官：聊聊对Vue.js框架的理解
   

   面试官：聊聊对Vue.js框架的理解 开发者（KaiFaX） 面向全栈工程师的开发者 专注于前端、Java/Python/Go/PHP的技术社区 作者 |yacan8 来源 | https://github.com/yacan8/blog/issues/26 本文为一次前端技术分享的演讲稿，所以尽力不贴 Vue.js 的源码，因为贴

6. 聊聊 Python 的双向队列
   

   Python教程 今天介绍双向队列。 虽然可以使用 Python 列表的 .append 和 .pop 方法模拟栈或者队列，但删除列表的第一个元素或者在第一个元素之前添加一个新元素，都非常耗时。因为需要把列表中的所有元素向后移动。 Python 的双向队列使用 collections.deque

7. 【Web技术】756- 聊聊如何设计组件
   

   【Web技术】756- 聊聊如何设计组件 作者：我想写文章啊 来源：https://juejin.im/post/6878099828497186823 现今的web开发通过前后端分离的技术拆分为了web后端开发与web前端开发，值得指出的是，web前端开发早已不是传统意义上的开发模式了，转而变成了 web

8. 聊聊监控
   

   聊聊监控 之前说要聊聊监控，这篇来填坑了。 指标 《踩坑记：Goroutine泄漏》开篇那张截图，展示了单个服务进程启动的 Goroutine 数量；除此之外，我们的服务进程在后台还采集了很多其他指标，例如： 当前存活在堆上的对象所占空间 这些数据是哪儿来的呢？run

9. 聊聊TCP/IP和Http/Https协议中的一些高频面试知识点(一)
   

   聊聊TCP/IP和Http/Https协议中的一些高频面试知识点(一) TCP/IP和Http协议已经是老生常谈的话题了，也是每个程序猿必备的网络通信基础，今天我们聊一聊面试过程中经常会出现的有关TCP/IP和Http协议的一些知识。本篇文章呢只是带大家了解一些关键的知识点，不

10. 1024,聊聊HTML
    

    1024,聊聊HTML HTML（超文本标记语言）是一种用于创建网页的标准标记语言。 HTML 不需要编译，可以直接由浏览器执行，它的解析依赖于浏览器的内核。 它不是一种编程语言，而是一种标记语言。 1111111 000000000 222222222222222 444444444 1::::::1 00:::::::