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
|
package util
import (
"errors"
"sync"
"time"
"github.com/chrislusf/seaweedfs/weed/glog"
)
var (
TimeoutErr = errors.New("timeout")
)
// A bufferedChan implemented by a buffered channel
type ResourcePool struct {
sync.Mutex
bufferedChan chan interface{}
poolSizeLimit int
inuse int
newFn func() (interface{}, error)
}
func NewResourcePool(poolSizeLimit int, newFn func() (interface{}, error)) *ResourcePool {
p := &ResourcePool{
poolSizeLimit: poolSizeLimit,
newFn: newFn,
bufferedChan: make(chan interface{}, poolSizeLimit),
}
return p
}
func (p *ResourcePool) Size() int {
p.Lock()
defer p.Unlock()
return len(p.bufferedChan) + p.inuse
}
func (p *ResourcePool) Free() int {
p.Lock()
defer p.Unlock()
return p.poolSizeLimit - p.inuse
}
func (p *ResourcePool) Get(timeout time.Duration) (interface{}, error) {
d, err := p.get(timeout)
if err != nil {
return nil, err
}
if d == nil && p.newFn != nil {
var err error
d, err = p.newFn()
if err != nil {
return nil, err
}
}
p.Lock()
defer p.Unlock()
p.inuse++
return d, nil
}
func (p *ResourcePool) Release(v interface{}) {
p.Lock()
defer p.Unlock()
if p.inuse == 0 {
glog.V(0).Infof("released too many times?")
return
}
p.bufferedChan <- v
p.inuse--
}
func (p *ResourcePool) get(timeout time.Duration) (interface{}, error) {
select {
case v := <-p.bufferedChan:
return v, nil
default:
}
if p.Free() > 0 {
d, err := p.newFn()
if err != nil {
return nil, err
}
return d, nil
}
// wait for an freed item
select {
case v := <-p.bufferedChan:
return v, nil
case <-time.After(timeout):
}
return nil, TimeoutErr
}
|
