JakeHillion/storj
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

remove ctx from repairer struct (#535)

Browse Source
This commit is contained in:
Cameron 2018-10-25 14:59:36 -04:00 committed by GitHub
parent e2c0dd437a
commit f7828e73ea
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 13 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
pkg/datarepair/repairer/config.go
View File

@ -28,6 +28,6 @@ func (c Config) Run(ctx context.Context, server *provider.Provider) (err error)
queue := q.NewQueue(client) queue := q.NewQueue(client)
repairer := newRepairer(ctx, queue, c.Interval, c.MaxRepair) repairer := newRepairer(queue, c.Interval, c.MaxRepair)
return repairer.Run() return repairer.Run(ctx)
} }

24
pkg/datarepair/repairer/repairer.go
View File

@ -16,21 +16,19 @@ import (
// Repairer is the interface for the data repair queue // Repairer is the interface for the data repair queue
type Repairer interface { type Repairer interface {
Repair(seg *pb.InjuredSegment) error Repair(ctx context.Context, seg *pb.InjuredSegment) error
Run() error Run(ctx context.Context) error
} }
// repairer holds important values for data repair // repairer holds important values for data repair
type repairer struct { type repairer struct {
ctx context.Context
queue q.RepairQueue queue q.RepairQueue
limiter *sync2.Limiter limiter *sync2.Limiter
ticker *time.Ticker ticker *time.Ticker
} }
func newRepairer(ctx context.Context, queue q.RepairQueue, interval time.Duration, concurrency int) *repairer { func newRepairer(queue q.RepairQueue, interval time.Duration, concurrency int) *repairer {
return &repairer{ return &repairer{
ctx: ctx,
queue: queue, queue: queue,
limiter: sync2.NewLimiter(concurrency), limiter: sync2.NewLimiter(concurrency),
ticker: time.NewTicker(interval), ticker: time.NewTicker(interval),
@ -38,8 +36,8 @@ func newRepairer(ctx context.Context, queue q.RepairQueue, interval time.Duratio
} }
// Run the repairer loop // Run the repairer loop
func (r *repairer) Run() (err error) { func (r *repairer) Run(ctx context.Context) (err error) {
defer mon.Task()(&r.ctx)(&err) defer mon.Task()(&ctx)(&err)
// wait for all repairs to complete // wait for all repairs to complete
defer r.limiter.Wait() defer r.limiter.Wait()
@ -47,8 +45,8 @@ func (r *repairer) Run() (err error) {
for { for {
select { select {
case <-r.ticker.C: // wait for the next interval to happen case <-r.ticker.C: // wait for the next interval to happen
case <-r.ctx.Done(): // or the repairer is canceled via context case <-ctx.Done(): // or the repairer is canceled via context
return r.ctx.Err() return ctx.Err()
} }
seg, err := r.queue.Dequeue() seg, err := r.queue.Dequeue()
@ -58,8 +56,8 @@ func (r *repairer) Run() (err error) {
continue continue
} }
r.limiter.Go(r.ctx, func() { r.limiter.Go(ctx, func() {
err := r.Repair(&seg) err := r.Repair(ctx, &seg)
if err != nil { if err != nil {
zap.L().Error("Repair failed", zap.Error(err)) zap.L().Error("Repair failed", zap.Error(err))
} }
@ -68,8 +66,8 @@ func (r *repairer) Run() (err error) {
} }
// Repair starts repair of the segment // Repair starts repair of the segment
func (r *repairer) Repair(seg *pb.InjuredSegment) (err error) { func (r *repairer) Repair(ctx context.Context, seg *pb.InjuredSegment) (err error) {
defer mon.Task()(&r.ctx)(&err) defer mon.Task()(&ctx)(&err)
// TODO: // TODO:
zap.L().Debug("Repairing", zap.Any("segment", seg)) zap.L().Debug("Repairing", zap.Any("segment", seg))
return err return err