mirror of
https://github.com/pgsty/minio.git
synced 2026-07-21 21:20:23 +03:00
Use 'minio/pkg/sync/errgroup' and 'minio/pkg/workers' (#17069)
This commit is contained in:
Praveen raj Mani
@@ -1,133 +0,0 @@
|
||||
// Copyright (c) 2015-2021 MinIO, Inc.
|
||||
//
|
||||
// This file is part of MinIO Object Storage stack
|
||||
//
|
||||
// This program is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU Affero General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// This program is distributed in the hope that it will be useful
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU Affero General Public License for more details.
|
||||
//
|
||||
// You should have received a copy of the GNU Affero General Public License
|
||||
// along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
package errgroup
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
)
|
||||
|
||||
// A Group is a collection of goroutines working on subtasks that are part of
|
||||
// the same overall task.
|
||||
//
|
||||
// A zero Group can be used if errors should not be tracked.
|
||||
type Group struct {
|
||||
firstErr int64 // ref: https://golang.org/pkg/sync/atomic/#pkg-note-BUG
|
||||
wg sync.WaitGroup
|
||||
bucket chan struct{}
|
||||
errs []error
|
||||
cancel context.CancelFunc
|
||||
ctxCancel <-chan struct{} // nil if no context.
|
||||
ctxErr func() error
|
||||
}
|
||||
|
||||
// WithNErrs returns a new Group with length of errs slice upto nerrs,
|
||||
// upon Wait() errors are returned collected from all tasks.
|
||||
func WithNErrs(nerrs int) *Group {
|
||||
return &Group{errs: make([]error, nerrs), firstErr: -1}
|
||||
}
|
||||
|
||||
// Wait blocks until all function calls from the Go method have returned, then
|
||||
// returns the slice of errors from all function calls.
|
||||
func (g *Group) Wait() []error {
|
||||
g.wg.Wait()
|
||||
if g.cancel != nil {
|
||||
g.cancel()
|
||||
}
|
||||
return g.errs
|
||||
}
|
||||
|
||||
// WaitErr blocks until all function calls from the Go method have returned, then
|
||||
// returns the first error returned.
|
||||
func (g *Group) WaitErr() error {
|
||||
g.wg.Wait()
|
||||
if g.cancel != nil {
|
||||
g.cancel()
|
||||
}
|
||||
if g.firstErr >= 0 && len(g.errs) > int(g.firstErr) {
|
||||
// len(g.errs) > int(g.firstErr) is for then used uninitialized.
|
||||
return g.errs[g.firstErr]
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// WithConcurrency allows to limit the concurrency of the group.
|
||||
// This must be called before starting any async processes.
|
||||
// There is no order to which functions are allowed to run.
|
||||
// If n <= 0 no concurrency limits are enforced.
|
||||
// g is modified and returned as well.
|
||||
func (g *Group) WithConcurrency(n int) *Group {
|
||||
if n <= 0 {
|
||||
g.bucket = nil
|
||||
return g
|
||||
}
|
||||
|
||||
// Fill bucket with tokens
|
||||
g.bucket = make(chan struct{}, n)
|
||||
for i := 0; i < n; i++ {
|
||||
g.bucket <- struct{}{}
|
||||
}
|
||||
return g
|
||||
}
|
||||
|
||||
// WithCancelOnError will return a context that is canceled
|
||||
// as soon as an error occurs.
|
||||
// The returned CancelFunc must always be called similar to context.WithCancel.
|
||||
// If the supplied context is canceled any goroutines waiting for execution are also canceled.
|
||||
func (g *Group) WithCancelOnError(ctx context.Context) (context.Context, context.CancelFunc) {
|
||||
ctx, g.cancel = context.WithCancel(ctx)
|
||||
g.ctxCancel = ctx.Done()
|
||||
g.ctxErr = ctx.Err
|
||||
return ctx, g.cancel
|
||||
}
|
||||
|
||||
// Go calls the given function in a new goroutine.
|
||||
//
|
||||
// The errors will be collected in errs slice and returned by Wait().
|
||||
func (g *Group) Go(f func() error, index int) {
|
||||
g.wg.Add(1)
|
||||
go func() {
|
||||
defer g.wg.Done()
|
||||
if g.bucket != nil {
|
||||
// Wait for token
|
||||
select {
|
||||
case <-g.bucket:
|
||||
defer func() {
|
||||
// Put back token..
|
||||
g.bucket <- struct{}{}
|
||||
}()
|
||||
case <-g.ctxCancel:
|
||||
if len(g.errs) > index {
|
||||
atomic.CompareAndSwapInt64(&g.firstErr, -1, int64(index))
|
||||
g.errs[index] = g.ctxErr()
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
if err := f(); err != nil {
|
||||
if len(g.errs) > index {
|
||||
atomic.CompareAndSwapInt64(&g.firstErr, -1, int64(index))
|
||||
g.errs[index] = err
|
||||
}
|
||||
if g.cancel != nil {
|
||||
g.cancel()
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
@@ -1,53 +0,0 @@
|
||||
// Copyright (c) 2015-2021 MinIO, Inc.
|
||||
//
|
||||
// This file is part of MinIO Object Storage stack
|
||||
//
|
||||
// This program is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU Affero General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// This program is distributed in the hope that it will be useful
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU Affero General Public License for more details.
|
||||
//
|
||||
// You should have received a copy of the GNU Affero General Public License
|
||||
// along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
package errgroup
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"reflect"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestGroupWithNErrs(t *testing.T) {
|
||||
err1 := fmt.Errorf("errgroup_test: 1")
|
||||
err2 := fmt.Errorf("errgroup_test: 2")
|
||||
|
||||
cases := []struct {
|
||||
errs []error
|
||||
}{
|
||||
{errs: []error{nil}},
|
||||
{errs: []error{err1}},
|
||||
{errs: []error{err1, nil}},
|
||||
{errs: []error{err1, nil, err2}},
|
||||
}
|
||||
|
||||
for j, tc := range cases {
|
||||
t.Run(fmt.Sprintf("Test%d", j+1), func(t *testing.T) {
|
||||
g := WithNErrs(len(tc.errs))
|
||||
for i, err := range tc.errs {
|
||||
err := err
|
||||
g.Go(func() error { return err }, i)
|
||||
}
|
||||
|
||||
gotErrs := g.Wait()
|
||||
if !reflect.DeepEqual(gotErrs, tc.errs) {
|
||||
t.Errorf("Expected %#v, got %#v", tc.errs, gotErrs)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -1,63 +0,0 @@
|
||||
// Copyright (c) 2022-2023 MinIO, Inc.
|
||||
//
|
||||
// This file is part of MinIO Object Storage stack
|
||||
//
|
||||
// This program is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU Affero General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// This program is distributed in the hope that it will be useful
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU Affero General Public License for more details.
|
||||
//
|
||||
// You should have received a copy of the GNU Affero General Public License
|
||||
// along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
package workers
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// Workers provides a bounded semaphore with the ability to wait until all
|
||||
// concurrent jobs finish.
|
||||
type Workers struct {
|
||||
wg sync.WaitGroup
|
||||
queue chan struct{}
|
||||
}
|
||||
|
||||
// New creates a Workers object which allows up to n jobs to proceed
|
||||
// concurrently. n must be > 0.
|
||||
func New(n int) (*Workers, error) {
|
||||
if n <= 0 {
|
||||
return nil, errors.New("n must be > 0")
|
||||
}
|
||||
|
||||
queue := make(chan struct{}, n)
|
||||
for i := 0; i < n; i++ {
|
||||
queue <- struct{}{}
|
||||
}
|
||||
return &Workers{
|
||||
queue: queue,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Take is how a job (goroutine) can Take its turn.
|
||||
func (jt *Workers) Take() {
|
||||
jt.wg.Add(1)
|
||||
<-jt.queue
|
||||
}
|
||||
|
||||
// Give is how a job (goroutine) can give back its turn once done.
|
||||
func (jt *Workers) Give() {
|
||||
jt.queue <- struct{}{}
|
||||
jt.wg.Done()
|
||||
}
|
||||
|
||||
// Wait waits for all ongoing concurrent jobs to complete
|
||||
func (jt *Workers) Wait() {
|
||||
jt.wg.Wait()
|
||||
}
|
||||
@@ -1,148 +0,0 @@
|
||||
// Copyright (c) 2022-2023 MinIO, Inc.
|
||||
//
|
||||
// This file is part of MinIO Object Storage stack
|
||||
//
|
||||
// This program is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU Affero General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// This program is distributed in the hope that it will be useful
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU Affero General Public License for more details.
|
||||
//
|
||||
// You should have received a copy of the GNU Affero General Public License
|
||||
// along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
package workers
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestWorkers(t *testing.T) {
|
||||
tests := []struct {
|
||||
n int
|
||||
jobs int
|
||||
mustFail bool
|
||||
}{
|
||||
{
|
||||
n: 0,
|
||||
jobs: 5,
|
||||
mustFail: true,
|
||||
},
|
||||
{
|
||||
n: -1,
|
||||
jobs: 5,
|
||||
mustFail: true,
|
||||
},
|
||||
{
|
||||
n: 1,
|
||||
jobs: 5,
|
||||
},
|
||||
{
|
||||
n: 2,
|
||||
jobs: 5,
|
||||
},
|
||||
{
|
||||
n: 5,
|
||||
jobs: 10,
|
||||
},
|
||||
{
|
||||
n: 10,
|
||||
jobs: 5,
|
||||
},
|
||||
}
|
||||
testFn := func(n, jobs int, mustFail bool) {
|
||||
var mu sync.Mutex
|
||||
var jobsDone int
|
||||
// Create workers for n concurrent workers
|
||||
jt, err := New(n)
|
||||
if err == nil && mustFail {
|
||||
t.Fatal("Expected test to return error")
|
||||
}
|
||||
if err != nil && mustFail {
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatalf("Unexpected error: %v", err)
|
||||
}
|
||||
|
||||
for i := 0; i < jobs; i++ {
|
||||
jt.Take()
|
||||
go func() { // Launch a worker after acquiring a token
|
||||
defer jt.Give() // Give token back once done
|
||||
mu.Lock()
|
||||
jobsDone++
|
||||
mu.Unlock()
|
||||
}()
|
||||
}
|
||||
jt.Wait() // Wait for all workers to complete
|
||||
if jobsDone != jobs {
|
||||
t.Fatalf("Expected %d jobs to be done but only %d were done", jobs, jobsDone)
|
||||
}
|
||||
}
|
||||
|
||||
for i, test := range tests {
|
||||
t.Run(fmt.Sprintf("test-%d", i), func(t *testing.T) {
|
||||
testFn(test.n, test.jobs, test.mustFail)
|
||||
})
|
||||
}
|
||||
|
||||
// Verify that workers can be reused after full drain
|
||||
t.Run("test-workers-reuse", func(t *testing.T) {
|
||||
var mu sync.Mutex
|
||||
jt, _ := New(5)
|
||||
for reuse := 0; reuse < 3; reuse++ {
|
||||
var jobsDone int
|
||||
for i := 0; i < 10; i++ {
|
||||
jt.Take()
|
||||
go func() {
|
||||
defer jt.Give()
|
||||
mu.Lock()
|
||||
jobsDone++
|
||||
mu.Unlock()
|
||||
}()
|
||||
}
|
||||
jt.Wait()
|
||||
if jobsDone != 10 {
|
||||
t.Fatalf("Expected %d jobs to be complete but only %d were", 10, jobsDone)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func benchmarkWorkers(b *testing.B, n, jobs int) {
|
||||
b.ReportAllocs()
|
||||
b.RunParallel(func(pb *testing.PB) {
|
||||
for pb.Next() {
|
||||
var mu sync.Mutex
|
||||
var jobsDone int
|
||||
jt, _ := New(n)
|
||||
for i := 0; i < jobs; i++ {
|
||||
jt.Take()
|
||||
go func() {
|
||||
defer jt.Give()
|
||||
mu.Lock()
|
||||
jobsDone++
|
||||
mu.Unlock()
|
||||
}()
|
||||
}
|
||||
jt.Wait()
|
||||
if jobsDone != jobs {
|
||||
b.Fail()
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func BenchmarkWorkers_N5_J10(b *testing.B) {
|
||||
benchmarkWorkers(b, 5, 10)
|
||||
}
|
||||
|
||||
func BenchmarkWorkers_N5_J100(b *testing.B) {
|
||||
benchmarkWorkers(b, 5, 100)
|
||||
}
|
||||
Reference in New Issue
Block a user