this repo has no description
run N threads of compactor
authored by
Brian Olson
and committed by
Brian Olson
ac6e9502
5f33e2cc
+34
-41
+34
-41
bgs/compactor.go
+34
-41
bgs/compactor.go
···
105
105
stats *carstore.CompactionStats
106
106
}
107
107
108
+
func (cstate *CompactorState) set(uid models.Uid, did, status string, stats *carstore.CompactionStats) {
109
+
cstate.latestUID = uid
110
+
cstate.latestDID = did
111
+
cstate.status = status
112
+
cstate.stats = stats
113
+
}
114
+
108
115
// Compactor is a compactor daemon that compacts repos in the background
109
116
type Compactor struct {
110
117
q *uniQueue
111
-
state *CompactorState
112
118
stateLk sync.RWMutex
113
119
exit chan struct{}
114
-
exited chan struct{}
115
120
requeueInterval time.Duration
116
121
requeueLimit int
117
122
requeueShardCount int
118
123
requeueFast bool
124
+
125
+
numWorkers int
126
+
wg sync.WaitGroup
119
127
}
120
128
121
129
type CompactorOptions struct {
···
123
131
RequeueLimit int
124
132
RequeueShardCount int
125
133
RequeueFast bool
134
+
NumWorkers int
126
135
}
127
136
128
137
func DefaultCompactorOptions() *CompactorOptions {
···
131
140
RequeueLimit: 0,
132
141
RequeueShardCount: 50,
133
142
RequeueFast: true,
143
+
NumWorkers: 2,
134
144
}
135
145
}
136
146
···
143
153
q: &uniQueue{
144
154
members: make(map[models.Uid]struct{}),
145
155
},
146
-
state: &CompactorState{},
147
156
exit: make(chan struct{}),
148
-
exited: make(chan struct{}),
149
157
requeueInterval: opts.RequeueInterval,
150
158
requeueLimit: opts.RequeueLimit,
151
159
requeueFast: opts.RequeueFast,
152
160
requeueShardCount: opts.RequeueShardCount,
161
+
numWorkers: opts.NumWorkers,
153
162
}
154
163
}
155
164
···
158
167
Targets []carstore.CompactionTarget
159
168
}
160
169
161
-
func (c *Compactor) SetState(uid models.Uid, did, status string, stats *carstore.CompactionStats) {
162
-
c.stateLk.Lock()
163
-
defer c.stateLk.Unlock()
164
-
165
-
c.state.latestUID = uid
166
-
c.state.latestDID = did
167
-
c.state.status = status
168
-
c.state.stats = stats
169
-
}
170
-
171
-
func (c *Compactor) GetState() *CompactorState {
172
-
c.stateLk.RLock()
173
-
defer c.stateLk.RUnlock()
174
-
175
-
return &CompactorState{
176
-
latestUID: c.state.latestUID,
177
-
latestDID: c.state.latestDID,
178
-
status: c.state.status,
179
-
stats: c.state.stats,
180
-
}
181
-
}
182
-
183
170
var errNoReposToCompact = fmt.Errorf("no repos to compact")
184
171
185
172
// Start starts the compactor
186
173
func (c *Compactor) Start(bgs *BGS) {
187
174
log.Info("starting compactor")
188
-
go c.doWork(bgs)
175
+
c.wg.Add(c.numWorkers)
176
+
for _ = range c.numWorkers {
177
+
go c.doWork(bgs, &c.wg)
178
+
}
189
179
if c.requeueInterval > 0 {
190
180
go func() {
191
181
log.Infow("starting compactor requeue routine",
···
217
207
func (c *Compactor) Shutdown() {
218
208
log.Info("stopping compactor")
219
209
close(c.exit)
220
-
<-c.exited
210
+
c.wg.Wait()
221
211
log.Info("compactor stopped")
222
212
}
223
213
224
-
func (c *Compactor) doWork(bgs *BGS) {
214
+
func (c *Compactor) doWork(bgs *BGS, wg *sync.WaitGroup) {
215
+
defer wg.Done()
225
216
for {
226
217
select {
227
218
case <-c.exit:
228
219
log.Info("compactor worker exiting, no more active compactions running")
229
-
close(c.exited)
230
220
return
231
221
default:
232
222
}
···
240
230
time.Sleep(time.Second * 5)
241
231
continue
242
232
}
243
-
state = c.GetState()
244
233
log.Errorw("failed to compact repo",
245
234
"err", err,
246
235
"uid", state.latestUID,
···
263
252
}
264
253
}
265
254
266
-
func (c *Compactor) compactNext(ctx context.Context, bgs *BGS) (*CompactorState, error) {
255
+
func (c *Compactor) compactNext(ctx context.Context, bgs *BGS) (state CompactorState, err error) {
267
256
ctx, span := otel.Tracer("compactor").Start(ctx, "CompactNext")
268
257
defer span.End()
269
258
270
259
item, ok := c.q.Pop()
271
260
if !ok || item == nil {
272
-
return nil, errNoReposToCompact
261
+
err = errNoReposToCompact
262
+
return
273
263
}
274
264
275
-
c.SetState(item.uid, "unknown", "getting_user", nil)
265
+
state.set(item.uid, "unknown", "getting_user", nil)
276
266
277
267
user, err := bgs.lookupUserByUID(ctx, item.uid)
278
268
if err != nil {
279
269
span.RecordError(err)
280
-
c.SetState(item.uid, "unknown", "failed_getting_user", nil)
281
-
return nil, fmt.Errorf("failed to get user %d: %w", item.uid, err)
270
+
state.set(item.uid, "unknown", "failed_getting_user", nil)
271
+
err = fmt.Errorf("failed to get user %d: %w", item.uid, err)
272
+
return
282
273
}
283
274
284
275
span.SetAttributes(attribute.String("repo", user.Did), attribute.Int("uid", int(item.uid)))
285
276
286
-
c.SetState(item.uid, user.Did, "compacting", nil)
277
+
state.set(item.uid, user.Did, "compacting", nil)
287
278
288
279
start := time.Now()
289
280
st, err := bgs.repoman.CarStore().CompactUserShards(ctx, item.uid, item.fast)
290
281
if err != nil {
291
282
span.RecordError(err)
292
-
c.SetState(item.uid, user.Did, "failed_compacting", nil)
293
-
return nil, fmt.Errorf("failed to compact shards for user %d: %w", item.uid, err)
283
+
state.set(item.uid, user.Did, "failed_compacting", nil)
284
+
err = fmt.Errorf("failed to compact shards for user %d: %w", item.uid, err)
285
+
return
294
286
}
295
287
compactionDuration.Observe(time.Since(start).Seconds())
296
288
···
302
294
attribute.Int("refs", st.TotalRefs),
303
295
)
304
296
305
-
c.SetState(item.uid, user.Did, "done", st)
297
+
state.set(item.uid, user.Did, "done", st)
306
298
307
-
return c.GetState(), nil
299
+
err = nil
300
+
return
308
301
}
309
302
310
303
func (c *Compactor) EnqueueRepo(ctx context.Context, user User, fast bool) {