mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-01 18:00:46 +01:00
Anton Rozanov
e7f5c6873a
* ec.balance: spread concurrent shard moves over target nodes The planner gives every freshly encoded volume the same target order, so running each volume's moves in plan order sends all concurrent copies to the same node while the other nodes idle. Pick, among the volumes not already moving, the pending move whose target receives the fewest moves right now. One volume still never has two moves at once, and moves of one shard keep their plan order. * ec.balance: let a move into a node wait for the move of its volume leaving it The planner counts the slot an outgoing shard frees as room for a later incoming shard of the same volume, and the receiver refuses a shard it has no slot for. Make the spread tests deterministic: hold the first moves until all have started, and check that errors of several volumes are all returned. * ec.balance: pin that the spread moves run concurrently and in plan order without parallelism
94 lines
2.4 KiB
Go
94 lines
2.4 KiB
Go
package util
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"sync"
|
|
)
|
|
|
|
// ErrorWaitGroup implements a goroutine wait group which aggregates errors, if any.
|
|
type ErrorWaitGroup struct {
|
|
maxConcurrency int
|
|
wg *sync.WaitGroup
|
|
wgSem chan bool
|
|
errors []error
|
|
errorsMu sync.Mutex
|
|
}
|
|
|
|
type ErrorWaitGroupTask func() error
|
|
|
|
// ExecuteParallelTaskGroups runs tasks from each group sequentially and runs
|
|
// independent groups with up to maxParallelization concurrency. Volume
|
|
// balancing uses one group per already-reserved volume move.
|
|
func ExecuteParallelTaskGroups(maxParallelization int, taskGroups [][]ErrorWaitGroupTask) error {
|
|
ewg := NewErrorWaitGroup(maxParallelization)
|
|
for _, taskGroup := range taskGroups {
|
|
taskGroup := taskGroup
|
|
ewg.Add(func() error {
|
|
for _, task := range taskGroup {
|
|
if err := task(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
return ewg.Wait()
|
|
}
|
|
|
|
func NewErrorWaitGroup(maxConcurrency int) *ErrorWaitGroup {
|
|
if maxConcurrency <= 0 {
|
|
// no concurrency = one task at the time
|
|
maxConcurrency = 1
|
|
}
|
|
return &ErrorWaitGroup{
|
|
maxConcurrency: maxConcurrency,
|
|
wg: &sync.WaitGroup{},
|
|
wgSem: make(chan bool, maxConcurrency),
|
|
}
|
|
}
|
|
|
|
// Reset restarts an ErrorWaitGroup, keeping original settings. Errors and pending goroutines, if any, are flushed.
|
|
func (ewg *ErrorWaitGroup) Reset() {
|
|
close(ewg.wgSem)
|
|
|
|
ewg.wg = &sync.WaitGroup{}
|
|
ewg.wgSem = make(chan bool, ewg.maxConcurrency)
|
|
ewg.errors = nil
|
|
}
|
|
|
|
// Add queues an ErrorWaitGroupTask to be executed as a goroutine.
|
|
func (ewg *ErrorWaitGroup) Add(f ErrorWaitGroupTask) {
|
|
if ewg.maxConcurrency <= 1 {
|
|
// keep run order deterministic when parallelization is off
|
|
ewg.errors = append(ewg.errors, f())
|
|
return
|
|
}
|
|
|
|
ewg.wg.Add(1)
|
|
go func() {
|
|
ewg.wgSem <- true
|
|
|
|
err := f()
|
|
ewg.errorsMu.Lock()
|
|
ewg.errors = append(ewg.errors, err)
|
|
ewg.errorsMu.Unlock()
|
|
|
|
<-ewg.wgSem
|
|
ewg.wg.Done()
|
|
}()
|
|
}
|
|
|
|
// AddErrorf adds an error to an ErrorWaitGroupTask result, without queueing any goroutines.
|
|
func (ewg *ErrorWaitGroup) AddErrorf(format string, a ...interface{}) {
|
|
ewg.errorsMu.Lock()
|
|
ewg.errors = append(ewg.errors, fmt.Errorf(format, a...))
|
|
ewg.errorsMu.Unlock()
|
|
}
|
|
|
|
// Wait sleeps until all ErrorWaitGroupTasks are completed, then returns errors for them.
|
|
func (ewg *ErrorWaitGroup) Wait() error {
|
|
ewg.wg.Wait()
|
|
return errors.Join(ewg.errors...)
|
|
}
|