storj/pkg/datarepair/repairer/repairer.go

75 lines
1.6 KiB
Go
Raw Normal View History

// Copyright (C) 2018 Storj Labs, Inc.
// See LICENSE for copying information.
package repairer
import (
"context"
"time"
"go.uber.org/zap"
"storj.io/storj/internal/sync2"
q "storj.io/storj/pkg/datarepair/queue"
"storj.io/storj/pkg/pb"
)
// Repairer is the interface for the data repair queue
type Repairer interface {
2018-10-25 19:59:36 +01:00
Repair(ctx context.Context, seg *pb.InjuredSegment) error
Run(ctx context.Context) error
}
// repairer holds important values for data repair
type repairer struct {
queue q.RepairQueue
limiter *sync2.Limiter
ticker *time.Ticker
}
2018-10-25 19:59:36 +01:00
func newRepairer(queue q.RepairQueue, interval time.Duration, concurrency int) *repairer {
return &repairer{
queue: queue,
limiter: sync2.NewLimiter(concurrency),
ticker: time.NewTicker(interval),
}
}
// Run the repairer loop
2018-10-25 19:59:36 +01:00
func (r *repairer) Run(ctx context.Context) (err error) {
defer mon.Task()(&ctx)(&err)
// wait for all repairs to complete
defer r.limiter.Wait()
for {
select {
case <-r.ticker.C: // wait for the next interval to happen
2018-10-25 19:59:36 +01:00
case <-ctx.Done(): // or the repairer is canceled via context
return ctx.Err()
}
seg, err := r.queue.Dequeue()
if err != nil {
// TODO: only log when err != ErrQueueEmpty
zap.L().Error("dequeue", zap.Error(err))
continue
}
2018-10-25 19:59:36 +01:00
r.limiter.Go(ctx, func() {
err := r.Repair(ctx, &seg)
if err != nil {
zap.L().Error("Repair failed", zap.Error(err))
}
})
}
}
// Repair starts repair of the segment
2018-10-25 19:59:36 +01:00
func (r *repairer) Repair(ctx context.Context, seg *pb.InjuredSegment) (err error) {
defer mon.Task()(&ctx)(&err)
// TODO:
zap.L().Debug("Repairing", zap.Any("segment", seg))
return err
}