...
1
2
3
4
5
6
7 package singleflight
8
9 import "sync"
10
11
12 type call struct {
13 wg sync.WaitGroup
14
15
16
17 val any
18 err error
19
20
21
22
23 dups int
24 chans []chan<- Result
25 }
26
27
28
29 type Group struct {
30 mu sync.Mutex
31 m map[string]*call
32 }
33
34
35
36 type Result struct {
37 Val any
38 Err error
39 Shared bool
40 }
41
42
43
44
45
46
47 func (g *Group) Do(key string, fn func() (any, error)) (v any, err error, shared bool) {
48 g.mu.Lock()
49 if g.m == nil {
50 g.m = make(map[string]*call)
51 }
52 if c, ok := g.m[key]; ok {
53 c.dups++
54 g.mu.Unlock()
55 c.wg.Wait()
56 return c.val, c.err, true
57 }
58 c := new(call)
59 c.wg.Add(1)
60 g.m[key] = c
61 g.mu.Unlock()
62
63 g.doCall(c, key, fn)
64 return c.val, c.err, c.dups > 0
65 }
66
67
68
69 func (g *Group) DoChan(key string, fn func() (any, error)) <-chan Result {
70 ch := make(chan Result, 1)
71 g.mu.Lock()
72 if g.m == nil {
73 g.m = make(map[string]*call)
74 }
75 if c, ok := g.m[key]; ok {
76 c.dups++
77 c.chans = append(c.chans, ch)
78 g.mu.Unlock()
79 return ch
80 }
81 c := &call{chans: []chan<- Result{ch}}
82 c.wg.Add(1)
83 g.m[key] = c
84 g.mu.Unlock()
85
86 go g.doCall(c, key, fn)
87
88 return ch
89 }
90
91
92 func (g *Group) doCall(c *call, key string, fn func() (any, error)) {
93 c.val, c.err = fn()
94
95 g.mu.Lock()
96 c.wg.Done()
97 if g.m[key] == c {
98 delete(g.m, key)
99 }
100 for _, ch := range c.chans {
101 ch <- Result{c.val, c.err, c.dups > 0}
102 }
103 g.mu.Unlock()
104 }
105
106
107
108
109
110
111 func (g *Group) ForgetUnshared(key string) bool {
112 g.mu.Lock()
113 defer g.mu.Unlock()
114 c, ok := g.m[key]
115 if !ok {
116 return true
117 }
118 if c.dups == 0 {
119 delete(g.m, key)
120 return true
121 }
122 return false
123 }
124
View as plain text