move readLock on uploadId at EOF in PutObjectPart (minio#258)

we do not need to hold the read locks at the higher
layer instead before reading the body, instead hold
the read locks properly at the time of renamePart()
for protection from racy part overwrites to compete
with concurrent completeMultipart().

Author: harshavardhana

cmd/erasure-multipart.go

@@ -744,6 +744,15 @@ func (er erasureObjects) PutObjectPart(ctx context.Context, bucket, object, uplo
 	ctx = plkctx.Context()
 	defer partIDLock.Unlock(plkctx)
 
+	// Read lock for upload id, only held while reading the upload metadata.
+	uploadIDRLock := er.NewNSLock(bucket, pathJoin(object, uploadID))
+	rlkctx, err := uploadIDRLock.GetRLock(ctx, globalOperationTimeout)
+	if err != nil {
+		return PartInfo{}, err
+	}
+	ctx = rlkctx.Context()
+	defer uploadIDRLock.RUnlock(rlkctx)
+
 	onlineDisks, err = er.renamePart(ctx, onlineDisks, minioMetaTmpBucket, tmpPartPath, minioMetaMultipartBucket, partPath, partFI, writeQuorum)
 	if err != nil {
 		if errors.Is(err, errFileNotFound) {

cmd/erasure-server-pool.go

@@ -1858,16 +1858,6 @@ func (z *erasureServerPools) PutObjectPart(ctx context.Context, bucket, object,
 		return PartInfo{}, err
 	}
 
-	// Read lock for upload id.
-	// Only held while reading the upload metadata.
-	uploadIDRLock := z.NewNSLock(bucket, pathJoin(object, uploadID))
-	rlkctx, err := uploadIDRLock.GetRLock(ctx, globalOperationTimeout)
-	if err != nil {
-		return PartInfo{}, err
-	}
-	ctx = rlkctx.Context()
-	defer uploadIDRLock.RUnlock(rlkctx)
-
 	if z.SinglePool() {
 		return z.serverPools[0].PutObjectPart(ctx, bucket, object, uploadID, partID, data, opts)
 	}