MyFavMirrors/minio
1
0
Fork 0
mirror of https://github.com/minio/minio.git synced 2025-07-14 19:31:52 -04:00
Code Issues Packages Projects Releases Wiki Activity

optimize multipart upload

Browse Source 
cherry-pick 33cee9f38a3e662ad68973ab48f595f0435d423f from master
branch for improving multipart upload and lock handling
This commit is contained in:
Harshavardhana 2021-07-14 23:46:33 -07:00
parent 5258a68b45
commit 38709c84b7
1 changed files with 18 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
cmd/erasure-multipart.go
View File

@ -358,14 +358,19 @@ func (er erasureObjects) CopyObjectPart(ctx context.Context, srcBucket, srcObjec
//
// Implements S3 compatible Upload Part API.
func (er erasureObjects) PutObjectPart(ctx context.Context, bucket, object, uploadID string, partID int, r *PutObjReader, opts ObjectOptions) (pi PartInfo, err error) {
uploadIDLock := er.NewNSLock(bucket, pathJoin(object, uploadID))
if err = uploadIDLock.GetRLock(ctx, globalOperationTimeout); err != nil {
partIDLock := er.NewNSLock(bucket, pathJoin(object, uploadID, strconv.Itoa(partID)))
if err = partIDLock.GetLock(ctx, globalOperationTimeout); err != nil {
return PartInfo{}, err
}
defer partIDLock.Unlock()
uploadIDRLock := er.NewNSLock(bucket, pathJoin(object, uploadID))
if err = uploadIDRLock.GetRLock(ctx, globalOperationTimeout); err != nil {
return PartInfo{}, err
}
readLocked := true
defer func() {
if readLocked {
uploadIDLock.RUnlock()
if uploadIDRLock != nil {
uploadIDRLock.RUnlock()
}
}()
@ -389,6 +394,10 @@ func (er erasureObjects) PutObjectPart(ctx context.Context, bucket, object, uplo
partsMetadata, errs = readAllFileInfo(ctx, er.getDisks(), minioMetaMultipartBucket,
uploadIDPath, "")
// Unlock upload id locks before, so others can get it.
uploadIDRLock.RUnlock()
uploadIDRLock = nil
// get Quorum for this object
_, writeQuorum, err := objectQuorumFromMeta(ctx, er, partsMetadata, errs)
if err != nil {
@ -468,14 +477,12 @@ func (er erasureObjects) PutObjectPart(ctx context.Context, bucket, object, uplo
}
}
// Unlock here before acquiring write locks all concurrent
// PutObjectParts would serialize here updating `xl.meta`
uploadIDLock.RUnlock()
readLocked = false
if err = uploadIDLock.GetLock(ctx, globalOperationTimeout); err != nil {
// Acquire write lock to update metadata.
uploadIDWLock := er.NewNSLock(bucket, pathJoin(object, uploadID))
if err = uploadIDWLock.GetLock(ctx, globalOperationTimeout); err != nil {
return PartInfo{}, err
}
defer uploadIDLock.Unlock()
defer uploadIDWLock.Unlock()
// Validates if upload ID exists.
if err = er.checkUploadIDExists(ctx, bucket, object, uploadID); err != nil {