From 38709c84b7434efa160a751c714a618d3e7fde39 Mon Sep 17 00:00:00 2001
From: Harshavardhana <harsha@minio.io>
Date: Wed, 14 Jul 2021 23:46:33 -0700
Subject: [PATCH] optimize multipart upload

cherry-pick 33cee9f38a3e662ad68973ab48f595f0435d423f from master
branch for improving multipart upload and lock handling
---
 cmd/erasure-multipart.go | 29 ++++++++++++++++++-----------
 1 file changed, 18 insertions(+), 11 deletions(-)

diff --git a/cmd/erasure-multipart.go b/cmd/erasure-multipart.go
index abedbe4fa..c92a051d3 100644
--- a/cmd/erasure-multipart.go
+++ b/cmd/erasure-multipart.go
@@ -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 {