this repo has no description
half of compactor workers pop randomly
authored by
Brian Olson
and committed by
Brian Olson
49556e41
ac6e9502
+70
-11
+70
-11
bgs/compactor.go
+70
-11
bgs/compactor.go
···
2
2
3
3
import (
4
4
"context"
5
+
crypto_rand "crypto/rand"
5
6
"fmt"
7
+
"math/rand/v2"
6
8
"sync"
7
9
"time"
8
10
···
22
24
q []queueItem
23
25
members map[models.Uid]struct{}
24
26
lk sync.Mutex
27
+
rnd *rand.Rand
25
28
}
26
29
27
30
// Append appends a uid to the end of the queue if it doesn't already exist
···
82
85
}
83
86
84
87
// Pop pops the first item off the front of the queue
85
-
func (q *uniQueue) Pop() (*queueItem, bool) {
88
+
func (q *uniQueue) Pop() (item queueItem, ok bool) {
86
89
q.lk.Lock()
87
90
defer q.lk.Unlock()
88
91
89
92
if len(q.q) == 0 {
90
-
return nil, false
93
+
ok = false
94
+
return
91
95
}
92
96
93
-
item := q.q[0]
97
+
item = q.q[0]
94
98
q.q = q.q[1:]
95
99
delete(q.members, item.uid)
96
100
97
101
compactionQueueDepth.Dec()
98
-
return &item, true
102
+
ok = true
103
+
return
104
+
}
105
+
106
+
// PopRandom pops a random item off the of the queue
107
+
// Note: this disrupts the sorted order of the queue and in-order is no longer quite in-order. The randomly popped element is replaced with the last element.
108
+
func (q *uniQueue) PopRandom() (item queueItem, ok bool) {
109
+
q.lk.Lock()
110
+
defer q.lk.Unlock()
111
+
112
+
if len(q.q) == 0 {
113
+
ok = false
114
+
return
115
+
}
116
+
117
+
if len(q.q) == 1 {
118
+
item = q.q[0]
119
+
q.q = nil
120
+
} else {
121
+
pos := q.rnd.IntN(len(q.q))
122
+
item = q.q[pos]
123
+
last := len(q.q) - 1
124
+
q.q[pos] = q.q[last]
125
+
q.q = q.q[:last]
126
+
}
127
+
delete(q.members, item.uid)
128
+
129
+
compactionQueueDepth.Dec()
130
+
ok = true
131
+
return
99
132
}
100
133
101
134
type CompactorState struct {
···
144
177
}
145
178
}
146
179
180
+
func newRandFromRoot() *rand.Rand {
181
+
var seed [32]byte
182
+
crypto_rand.Read(seed[:])
183
+
chacha := rand.NewChaCha8(seed)
184
+
return rand.New(chacha)
185
+
}
186
+
147
187
func NewCompactor(opts *CompactorOptions) *Compactor {
148
188
if opts == nil {
149
189
opts = DefaultCompactorOptions()
···
152
192
return &Compactor{
153
193
q: &uniQueue{
154
194
members: make(map[models.Uid]struct{}),
195
+
rnd: newRandFromRoot(),
155
196
},
156
197
exit: make(chan struct{}),
157
198
requeueInterval: opts.RequeueInterval,
···
173
214
func (c *Compactor) Start(bgs *BGS) {
174
215
log.Info("starting compactor")
175
216
c.wg.Add(c.numWorkers)
176
-
for _ = range c.numWorkers {
177
-
go c.doWork(bgs, &c.wg)
217
+
for i := range c.numWorkers {
218
+
strategy := NextInOrder
219
+
if i%2 != 0 {
220
+
strategy = NextRandom
221
+
}
222
+
go c.doWork(bgs, &c.wg, strategy)
178
223
}
179
224
if c.requeueInterval > 0 {
180
225
go func() {
···
211
256
log.Info("compactor stopped")
212
257
}
213
258
214
-
func (c *Compactor) doWork(bgs *BGS, wg *sync.WaitGroup) {
259
+
func (c *Compactor) doWork(bgs *BGS, wg *sync.WaitGroup, strategy NextStrategy) {
215
260
defer wg.Done()
216
261
for {
217
262
select {
···
223
268
224
269
ctx := context.Background()
225
270
start := time.Now()
226
-
state, err := c.compactNext(ctx, bgs)
271
+
state, err := c.compactNext(ctx, bgs, strategy)
227
272
if err != nil {
228
273
if err == errNoReposToCompact {
229
274
log.Debug("no repos to compact, waiting and retrying")
···
252
297
}
253
298
}
254
299
255
-
func (c *Compactor) compactNext(ctx context.Context, bgs *BGS) (state CompactorState, err error) {
300
+
type NextStrategy int
301
+
302
+
const (
303
+
NextInOrder NextStrategy = iota
304
+
NextRandom
305
+
)
306
+
307
+
func (c *Compactor) compactNext(ctx context.Context, bgs *BGS, strategy NextStrategy) (state CompactorState, err error) {
256
308
ctx, span := otel.Tracer("compactor").Start(ctx, "CompactNext")
257
309
defer span.End()
258
310
259
-
item, ok := c.q.Pop()
260
-
if !ok || item == nil {
311
+
var item queueItem
312
+
var ok bool
313
+
switch strategy {
314
+
case NextRandom:
315
+
item, ok = c.q.PopRandom()
316
+
default:
317
+
item, ok = c.q.Pop()
318
+
}
319
+
if !ok {
261
320
err = errNoReposToCompact
262
321
return
263
322
}