Skip to content

Commit 78491dc

Browse files
authored
改变 pool.NewBuffer 的行为,响应固定容量的BufferPool (#20)
1 parent d07263d commit 78491dc

File tree

5 files changed

+20
-423
lines changed

5 files changed

+20
-423
lines changed

pool/benchmark/buffer_test.go

Lines changed: 0 additions & 103 deletions
This file was deleted.

pool/buffer.go

Lines changed: 11 additions & 86 deletions
Original file line numberDiff line numberDiff line change
@@ -5,27 +5,12 @@ package pool
55

66
import (
77
"bytes"
8-
"math/bits"
9-
"sort"
108
"sync"
11-
"sync/atomic"
129
"unsafe"
1310

1411
"github.com/thinkeridea/go-extend/exsync"
1512
)
1613

17-
const (
18-
bucketSize = 22
19-
minSizeBits = 6
20-
minSize = 1 << minSizeBits // 2^6=64
21-
maxSize = 1 << (minSizeBits + bucketSize - 1) // 2^28=256MB
22-
calibrateCallsThreshold = 10240
23-
)
24-
25-
var (
26-
buffBucket = [bucketSize]sync.Pool{}
27-
)
28-
2914
// BufferPool bytes.Buffer 的 sync.Pool 接口
3015
// 可以直接 Get *bytes.Buffer 并 Reset Buffer
3116
type BufferPool interface {
@@ -36,77 +21,6 @@ type BufferPool interface {
3621
Put(*bytes.Buffer)
3722
}
3823

39-
type buffer struct {
40-
index uint32
41-
calibrating uint32
42-
release uint32
43-
calls [bucketSize]uint32
44-
}
45-
46-
// NewBuffer 创建一个动态评估需求容量的 BufferPool,它们共享一个底层 bytes.Buffer 区间池
47-
// Deprecated: 这是错误测试产生的结果,它并没有相较 sync.Pool 有明显优势,详细参看: https://github.com/thinkeridea/go-extend/issues/17
48-
func NewBuffer(size int) BufferPool {
49-
b := &buffer{}
50-
b.index = uint32(buffBucketIndex(size))
51-
return b
52-
}
53-
54-
// Get 从 Pool 中获取一个 *bytes.Buffer 实例, 该实例已经被 Reset
55-
func (p *buffer) Get() *bytes.Buffer {
56-
idx := atomic.LoadUint32(&p.index)
57-
v := buffBucket[idx].Get()
58-
if v != nil {
59-
b := v.(*bytes.Buffer)
60-
b.Reset()
61-
return b
62-
}
63-
64-
return bytes.NewBuffer(make([]byte, 0, minSize<<idx))
65-
}
66-
67-
// Put 把 *bytes.Buffer 放回 Pool 中
68-
func (p *buffer) Put(b *bytes.Buffer) {
69-
if b.Cap() <= maxSize {
70-
buffBucket[buffBucketIndex(b.Cap())].Put(b)
71-
}
72-
73-
atomic.AddUint32(&p.calls[buffBucketIndex(bufferLen(b))], 1)
74-
if atomic.AddUint32(&p.release, 1) > calibrateCallsThreshold {
75-
p.calibrate()
76-
}
77-
}
78-
79-
func (p *buffer) calibrate() {
80-
if !atomic.CompareAndSwapUint32(&p.calibrating, 0, 1) {
81-
return
82-
}
83-
84-
var callSize [bucketSize]uint64
85-
for i := range callSize {
86-
callSize[i] = uint64(atomic.SwapUint32(&p.calls[i], 0))<<32 | minSize<<i
87-
}
88-
89-
sort.Slice(callSize[:], func(i, j int) bool {
90-
return callSize[i] > callSize[j]
91-
})
92-
93-
atomic.SwapUint32(&p.index, uint32(buffBucketIndex(int(callSize[0]<<32>>32))))
94-
atomic.StoreUint32(&p.release, 0)
95-
atomic.SwapUint32(&p.calibrating, 0)
96-
}
97-
98-
func buffBucketIndex(n int) int {
99-
if n == 0 {
100-
return 0
101-
}
102-
103-
idx := bits.Len32(uint32((n - 1) >> minSizeBits))
104-
if idx >= bucketSize {
105-
idx = bucketSize - 1
106-
}
107-
return idx
108-
}
109-
11024
var (
11125
buff64 exsync.OncePointer
11226
buff128 exsync.OncePointer
@@ -121,6 +35,17 @@ type bufferPool struct {
12135
sync.Pool
12236
}
12337

38+
// NewBuffer 创建一个指定容量的 BufferPool,这将是一个独立的非共享的,每次调用都会返回新的BufferPool
39+
func NewBuffer(size int) BufferPool {
40+
return &bufferPool{
41+
Pool: sync.Pool{
42+
New: func() interface{} {
43+
return bytes.NewBuffer(make([]byte, size))
44+
},
45+
},
46+
}
47+
}
48+
12449
func newBufferPool(size int) unsafe.Pointer {
12550
return unsafe.Pointer(&bufferPool{
12651
Pool: sync.Pool{

0 commit comments

Comments
 (0)