MyFavMirrors/minio
1
0
Fork 0
mirror of https://github.com/minio/minio.git synced 2025-02-28 05:49:16 -05:00
Code Issues Packages Projects Releases Wiki Activity

use the new errgroup API whereever applicable (#11466)

Browse Source 
start using the new errgroup concurrency control
API introduced in #11457
This commit is contained in:
Harshavardhana 2021-02-09 12:08:25 -08:00 committed by GitHub
parent 84d400487f
commit fe3c39b583
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 22 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
cmd/bucket-handlers.go
View File

@ -126,7 +126,10 @@ func initFederatorBackend(buckets []BucketInfo, objLayer ObjectLayer) {
// Add/update buckets that are not registered with the DNS // Add/update buckets that are not registered with the DNS
bucketsToBeUpdatedSlice := bucketsToBeUpdated.ToSlice() bucketsToBeUpdatedSlice := bucketsToBeUpdated.ToSlice()
g := errgroup.WithNErrs(len(bucketsToBeUpdatedSlice)) g := errgroup.WithNErrs(len(bucketsToBeUpdatedSlice)).WithConcurrency(50)
ctx, cancel := g.WithCancelOnError(GlobalContext)
defer cancel()
for index := range bucketsToBeUpdatedSlice { for index := range bucketsToBeUpdatedSlice {
index := index index := index
g.Go(func() error { g.Go(func() error {
@ -134,14 +137,13 @@ func initFederatorBackend(buckets []BucketInfo, objLayer ObjectLayer) {
}, index) }, index)
} }
for _, err := range g.Wait() { if err := g.WaitErr(); err != nil {
if err != nil { logger.LogIf(ctx, err)
logger.LogIf(GlobalContext, err) return
}
} }
for _, bucket := range bucketsInConflict.ToSlice() { for _, bucket := range bucketsInConflict.ToSlice() {
logger.LogIf(GlobalContext, fmt.Errorf("Unable to add bucket DNS entry for bucket %s, an entry exists for the same bucket by a different tenant. This local bucket will be ignored. Bucket names are globally unique in federated deployments. Use path style requests on following addresses '%v' to access this bucket.", bucket, globalDomainIPs.ToSlice())) logger.LogIf(ctx, fmt.Errorf("Unable to add bucket DNS entry for bucket %s, an entry exists for the same bucket by a different tenant. This local bucket will be ignored. Bucket names are globally unique in federated deployments. Use path style requests on following addresses '%v' to access this bucket", bucket, globalDomainIPs.ToSlice()))
} }
var wg sync.WaitGroup var wg sync.WaitGroup

22
cmd/bucket-listobjects-handlers.go
View File

@ -31,27 +31,21 @@ import (
) )
func concurrentDecryptETag(ctx context.Context, objects []ObjectInfo) { func concurrentDecryptETag(ctx context.Context, objects []ObjectInfo) {
inParallel := func(objects []ObjectInfo) { g := errgroup.WithNErrs(len(objects)).WithConcurrency(500)
g := errgroup.WithNErrs(len(objects)) _, cancel := g.WithCancelOnError(ctx)
defer cancel()
for index := range objects { for index := range objects {
index := index index := index
g.Go(func() error { g.Go(func() error {
size, err := objects[index].GetActualSize()
if err == nil {
objects[index].Size = size
}
objects[index].ETag = objects[index].GetActualETag(nil) objects[index].ETag = objects[index].GetActualETag(nil)
objects[index].Size, _ = objects[index].GetActualSize()
return nil return nil
}, index) }, index)
} }
g.Wait() g.WaitErr()
}
const maxConcurrent = 500
for {
if len(objects) < maxConcurrent {
inParallel(objects)
return
}
inParallel(objects[:maxConcurrent])
objects = objects[maxConcurrent:]
}
} }
// Validate all the ListObjects query arguments, returns an APIErrorCode // Validate all the ListObjects query arguments, returns an APIErrorCode