MyFavMirrors/minio
1
0
Fork 0
mirror of https://github.com/minio/minio.git synced 2025-04-22 03:24:38 -04:00
Code Issues Packages Projects Releases Wiki Activity

Do not use parallel deletes to avoid random I/O (#6178)

Browse Source 
The current code for deleting 1000 objects simultaneously
causes significant random I/O, which on slower drives
leads to servers disconnecting in a distributed setup.

Simplify this by serially deleting and reducing the
chattiness of this operation.
This commit is contained in:
Harshavardhana 2018-07-20 21:21:01 -07:00 committed by kannappanr
parent 76ddf4d32f
commit f1be356cc6
1 changed files with 17 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
cmd/bucket-handlers.go
View File

@ -28,7 +28,6 @@ import (
"path" "path"
"path/filepath" "path/filepath"
"strings" "strings"
"sync"
"github.com/gorilla/mux" "github.com/gorilla/mux"
@ -311,34 +310,24 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
return return
} }
var wg = &sync.WaitGroup{} // Allocate a new wait group. deleteObject := objectAPI.DeleteObject
var dErrs = make([]error, len(deleteObjects.Objects)) if api.CacheAPI() != nil {
deleteObject = api.CacheAPI().DeleteObject
// Delete all requested objects in parallel. }
for index, object := range deleteObjects.Objects {
wg.Add(1) var dErrs = make([]error, len(deleteObjects.Objects))
go func(i int, obj ObjectIdentifier) { for index, object := range deleteObjects.Objects {
defer wg.Done() // If the request is denied access, each item
// If the request is denied access, each item // should be marked as 'AccessDenied'
// should be marked as 'AccessDenied' if s3Error == ErrAccessDenied {
if s3Error == ErrAccessDenied { dErrs[index] = PrefixAccessDenied{
dErrs[i] = PrefixAccessDenied{ Bucket: bucket,
Bucket: bucket, Object: object.ObjectName,
Object: obj.ObjectName, }
} continue
return }
} dErrs[index] = deleteObject(ctx, bucket, object.ObjectName)
deleteObject := objectAPI.DeleteObject
if api.CacheAPI() != nil {
deleteObject = api.CacheAPI().DeleteObject
}
dErr := deleteObject(ctx, bucket, obj.ObjectName)
if dErr != nil {
dErrs[i] = dErr
}
}(index, object)
} }
wg.Wait()
// Collect deleted objects and errors if any. // Collect deleted objects and errors if any.
var deletedObjects []ObjectIdentifier var deletedObjects []ObjectIdentifier