vendor/github.com/syndtr/goleveldb/leveldb/util/buffer_pool.go

   1 // Copyright (c) 2014, Suryandaru Triandana <syndtr@gmail.com>
   2 // All rights reserved.
   3 //
   4 // Use of this source code is governed by a BSD-style license that can be
   5 // found in the LICENSE file.
   6
   7 package util
   8
   9 import (
  10         "fmt"
  11         "sync"
  12         "sync/atomic"
  13         "time"
  14 )
  15
  16 type buffer struct {
  17         b    []byte
  18         miss int
  19 }
  20
  21 // BufferPool is a 'buffer pool'.
  22 type BufferPool struct {
  23         pool      [6]chan []byte
  24         size      [5]uint32
  25         sizeMiss  [5]uint32
  26         sizeHalf  [5]uint32
  27         baseline  [4]int
  28         baseline0 int
  29
  30         mu     sync.RWMutex
  31         closed bool
  32         closeC chan struct{}
  33
  34         get     uint32
  35         put     uint32
  36         half    uint32
  37         less    uint32
  38         equal   uint32
  39         greater uint32
  40         miss    uint32
  41 }
  42
  43 func (p *BufferPool) poolNum(n int) int {
  44         if n <= p.baseline0 && n > p.baseline0/2 {
  45                 return 0
  46         }
  47         for i, x := range p.baseline {
  48                 if n <= x {
  49                         return i + 1
  50                 }
  51         }
  52         return len(p.baseline) + 1
  53 }
  54
  55 // Get returns buffer with length of n.
  56 func (p *BufferPool) Get(n int) []byte {
  57         if p == nil {
  58                 return make([]byte, n)
  59         }
  60
  61         p.mu.RLock()
  62         defer p.mu.RUnlock()
  63
  64         if p.closed {
  65                 return make([]byte, n)
  66         }
  67
  68         atomic.AddUint32(&p.get, 1)
  69
  70         poolNum := p.poolNum(n)
  71         pool := p.pool[poolNum]
  72         if poolNum == 0 {
  73                 // Fast path.
  74                 select {
  75                 case b := <-pool:
  76                         switch {
  77                         case cap(b) > n:
  78                                 if cap(b)-n >= n {
  79                                         atomic.AddUint32(&p.half, 1)
  80                                         select {
  81                                         case pool <- b:
  82                                         default:
  83                                         }
  84                                         return make([]byte, n)
  85                                 } else {
  86                                         atomic.AddUint32(&p.less, 1)
  87                                         return b[:n]
  88                                 }
  89                         case cap(b) == n:
  90                                 atomic.AddUint32(&p.equal, 1)
  91                                 return b[:n]
  92                         default:
  93                                 atomic.AddUint32(&p.greater, 1)
  94                         }
  95                 default:
  96                         atomic.AddUint32(&p.miss, 1)
  97                 }
  98
  99                 return make([]byte, n, p.baseline0)
 100         } else {
 101                 sizePtr := &p.size[poolNum-1]
 102
 103                 select {
 104                 case b := <-pool:
 105                         switch {
 106                         case cap(b) > n:
 107                                 if cap(b)-n >= n {
 108                                         atomic.AddUint32(&p.half, 1)
 109                                         sizeHalfPtr := &p.sizeHalf[poolNum-1]
 110                                         if atomic.AddUint32(sizeHalfPtr, 1) == 20 {
 111                                                 atomic.StoreUint32(sizePtr, uint32(cap(b)/2))
 112                                                 atomic.StoreUint32(sizeHalfPtr, 0)
 113                                         } else {
 114                                                 select {
 115                                                 case pool <- b:
 116                                                 default:
 117                                                 }
 118                                         }
 119                                         return make([]byte, n)
 120                                 } else {
 121                                         atomic.AddUint32(&p.less, 1)
 122                                         return b[:n]
 123                                 }
 124                         case cap(b) == n:
 125                                 atomic.AddUint32(&p.equal, 1)
 126                                 return b[:n]
 127                         default:
 128                                 atomic.AddUint32(&p.greater, 1)
 129                                 if uint32(cap(b)) >= atomic.LoadUint32(sizePtr) {
 130                                         select {
 131                                         case pool <- b:
 132                                         default:
 133                                         }
 134                                 }
 135                         }
 136                 default:
 137                         atomic.AddUint32(&p.miss, 1)
 138                 }
 139
 140                 if size := atomic.LoadUint32(sizePtr); uint32(n) > size {
 141                         if size == 0 {
 142                                 atomic.CompareAndSwapUint32(sizePtr, 0, uint32(n))
 143                         } else {
 144                                 sizeMissPtr := &p.sizeMiss[poolNum-1]
 145                                 if atomic.AddUint32(sizeMissPtr, 1) == 20 {
 146                                         atomic.StoreUint32(sizePtr, uint32(n))
 147                                         atomic.StoreUint32(sizeMissPtr, 0)
 148                                 }
 149                         }
 150                         return make([]byte, n)
 151                 } else {
 152                         return make([]byte, n, size)
 153                 }
 154         }
 155 }
 156
 157 // Put adds given buffer to the pool.
 158 func (p *BufferPool) Put(b []byte) {
 159         if p == nil {
 160                 return
 161         }
 162
 163         p.mu.RLock()
 164         defer p.mu.RUnlock()
 165
 166         if p.closed {
 167                 return
 168         }
 169
 170         atomic.AddUint32(&p.put, 1)
 171
 172         pool := p.pool[p.poolNum(cap(b))]
 173         select {
 174         case pool <- b:
 175         default:
 176         }
 177
 178 }
 179
 180 func (p *BufferPool) Close() {
 181         if p == nil {
 182                 return
 183         }
 184
 185         p.mu.Lock()
 186         if !p.closed {
 187                 p.closed = true
 188                 p.closeC <- struct{}{}
 189         }
 190         p.mu.Unlock()
 191 }
 192
 193 func (p *BufferPool) String() string {
 194         if p == nil {
 195                 return "<nil>"
 196         }
 197
 198         return fmt.Sprintf("BufferPool{B·%d Z·%v Zm·%v Zh·%v G·%d P·%d H·%d <·%d =·%d >·%d M·%d}",
 199                 p.baseline0, p.size, p.sizeMiss, p.sizeHalf, p.get, p.put, p.half, p.less, p.equal, p.greater, p.miss)
 200 }
 201
 202 func (p *BufferPool) drain() {
 203         ticker := time.NewTicker(2 * time.Second)
 204         defer ticker.Stop()
 205         for {
 206                 select {
 207                 case <-ticker.C:
 208                         for _, ch := range p.pool {
 209                                 select {
 210                                 case <-ch:
 211                                 default:
 212                                 }
 213                         }
 214                 case <-p.closeC:
 215                         close(p.closeC)
 216                         for _, ch := range p.pool {
 217                                 close(ch)
 218                         }
 219                         return
 220                 }
 221         }
 222 }
 223
 224 // NewBufferPool creates a new initialized 'buffer pool'.
 225 func NewBufferPool(baseline int) *BufferPool {
 226         if baseline <= 0 {
 227                 panic("baseline can't be <= 0")
 228         }
 229         p := &BufferPool{
 230                 baseline0: baseline,
 231                 baseline:  [...]int{baseline / 4, baseline / 2, baseline * 2, baseline * 4},
 232                 closeC:    make(chan struct{}, 1),
 233         }
 234         for i, cap := range []int{2, 2, 4, 4, 2, 1} {
 235                 p.pool[i] = make(chan []byte, cap)
 236         }
 237         go p.drain()
 238         return p
 239 }