-
Notifications
You must be signed in to change notification settings - Fork 0
/
buffer_syncer.go
106 lines (86 loc) · 2.52 KB
/
buffer_syncer.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
package xlog
import (
"bufio"
"context"
"sync"
"time"
"go.uber.org/zap/zapcore"
)
type bufferWriterSyncer struct {
sync.Mutex
bufferWriter *bufio.Writer
ticker *time.Ticker
}
const (
// defaultBufferSize sizes the buffer associated with each WriterSync.
defaultBufferSize = 256 * 1024
// defaultFlushInterval means the default flush interval
defaultFlushInterval = 30 * time.Second
)
// CloseFunc should be called when the caller exits to clean up buffers.
type CloseFunc func() error
// Buffer wraps a WriteSyncer in a buffer to improve performance,
// if bufferSize = 0, we set it to defaultBufferSize
// if flushInterval = 0, we set it to defaultFlushInterval
func Buffer(ws zapcore.WriteSyncer, bufferSize int, flushInterval time.Duration) (zapcore.WriteSyncer, CloseFunc) {
if _, ok := ws.(*bufferWriterSyncer); ok {
// no need to layer on another buffer
return ws, func() error { return nil }
}
ctx, cancel := context.WithCancel(context.Background())
if bufferSize == 0 {
bufferSize = defaultBufferSize
}
if flushInterval == 0 {
flushInterval = defaultFlushInterval
}
ticker := time.NewTicker(flushInterval)
ws = &bufferWriterSyncer{
bufferWriter: bufio.NewWriterSize(ws, bufferSize),
ticker: ticker,
}
// flush buffer every interval
// we do not need to exit this goroutine until closefunc called explicitly
go func() {
defer ticker.Stop()
for {
select {
case <-ticker.C:
// the background goroutine just keep syncing
// until the close func is called.
_ = ws.Sync()
case <-ctx.Done():
return
}
}
}()
closefunc := func() error {
cancel()
return ws.Sync()
}
return ws, closefunc
}
// Write ...
func (s *bufferWriterSyncer) Write(bs []byte) (int, error) {
// bufio is not goroutine safe, so add lock writer here
s.Lock()
defer s.Unlock()
// there are some logic internal for bufio.Writer here:
// 1. when the buffer is enough, data would not be flushed.
// 2. when the buffer is not enough, data would be flushed as soon as the buffer fills up.
// this would lead to xlog spliting, which is not acceptable for xlog collector
// so we need to flush bufferWriter before writing the data into bufferWriter
if len(bs) > s.bufferWriter.Available() && s.bufferWriter.Buffered() > 0 {
if err := s.bufferWriter.Flush(); err != nil {
return 0, err
}
}
return s.bufferWriter.Write(bs)
}
// Sync ...
func (s *bufferWriterSyncer) Sync() error {
// bufio is not goroutine safe, so add lock writer here
s.Lock()
defer s.Unlock()
return s.bufferWriter.Flush()
}