Add support for caching multipart in writethrough mode (#13507)

cmd/disk-cache.go

@@ -59,6 +59,13 @@ const (
 	CommitFailed cacheCommitStatus = "failed"
 )
 
+const (
+	// CommitWriteBack allows staging and write back of cached content for single object uploads
+	CommitWriteBack string = "writeback"
+	// CommitWriteThrough allows caching multipart uploads to disk synchronously
+	CommitWriteThrough string = "writethrough"
+)
+
 // String returns string representation of status
 func (s cacheCommitStatus) String() string {
 	return string(s)
@@ -80,6 +87,13 @@ type CacheObjectLayer interface {
 	DeleteObjects(ctx context.Context, bucket string, objects []ObjectToDelete, opts ObjectOptions) ([]DeletedObject, []error)
 	PutObject(ctx context.Context, bucket, object string, data *PutObjReader, opts ObjectOptions) (objInfo ObjectInfo, err error)
 	CopyObject(ctx context.Context, srcBucket, srcObject, destBucket, destObject string, srcInfo ObjectInfo, srcOpts, dstOpts ObjectOptions) (objInfo ObjectInfo, err error)
+	// Multipart operations.
+	NewMultipartUpload(ctx context.Context, bucket, object string, opts ObjectOptions) (uploadID string, err error)
+	PutObjectPart(ctx context.Context, bucket, object, uploadID string, partID int, data *PutObjReader, opts ObjectOptions) (info PartInfo, err error)
+	AbortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) error
+	CompleteMultipartUpload(ctx context.Context, bucket, object, uploadID string, uploadedParts []CompletePart, opts ObjectOptions) (objInfo ObjectInfo, err error)
+	CopyObjectPart(ctx context.Context, srcBucket, srcObject, dstBucket, dstObject, uploadID string, partID int, startOffset int64, length int64, srcInfo ObjectInfo, srcOpts, dstOpts ObjectOptions) (pi PartInfo, e error)
+
 	// Storage operations.
 	StorageInfo(ctx context.Context) CacheStorageInfo
 	CacheStats() *CacheStats
@@ -94,21 +108,26 @@ type cacheObjects struct {
 	// number of accesses after which to cache an object
 	after int
 	// commit objects in async manner
-	commitWriteback bool
+	commitWriteback    bool
+	commitWritethrough bool
+
 	// if true migration is in progress from v1 to v2
 	migrating bool
-	// mutex to protect migration bool
-	migMutex sync.Mutex
 	// retry queue for writeback cache mode to reattempt upload to backend
 	wbRetryCh chan ObjectInfo
 	// Cache stats
 	cacheStats *CacheStats
 
-	InnerGetObjectNInfoFn func(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, h http.Header, lockType LockType, opts ObjectOptions) (gr *GetObjectReader, err error)
-	InnerGetObjectInfoFn  func(ctx context.Context, bucket, object string, opts ObjectOptions) (objInfo ObjectInfo, err error)
-	InnerDeleteObjectFn   func(ctx context.Context, bucket, object string, opts ObjectOptions) (objInfo ObjectInfo, err error)
-	InnerPutObjectFn      func(ctx context.Context, bucket, object string, data *PutObjReader, opts ObjectOptions) (objInfo ObjectInfo, err error)
-	InnerCopyObjectFn     func(ctx context.Context, srcBucket, srcObject, destBucket, destObject string, srcInfo ObjectInfo, srcOpts, dstOpts ObjectOptions) (objInfo ObjectInfo, err error)
+	InnerGetObjectNInfoFn          func(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, h http.Header, lockType LockType, opts ObjectOptions) (gr *GetObjectReader, err error)
+	InnerGetObjectInfoFn           func(ctx context.Context, bucket, object string, opts ObjectOptions) (objInfo ObjectInfo, err error)
+	InnerDeleteObjectFn            func(ctx context.Context, bucket, object string, opts ObjectOptions) (objInfo ObjectInfo, err error)
+	InnerPutObjectFn               func(ctx context.Context, bucket, object string, data *PutObjReader, opts ObjectOptions) (objInfo ObjectInfo, err error)
+	InnerCopyObjectFn              func(ctx context.Context, srcBucket, srcObject, destBucket, destObject string, srcInfo ObjectInfo, srcOpts, dstOpts ObjectOptions) (objInfo ObjectInfo, err error)
+	InnerNewMultipartUploadFn      func(ctx context.Context, bucket, object string, opts ObjectOptions) (uploadID string, err error)
+	InnerPutObjectPartFn           func(ctx context.Context, bucket, object, uploadID string, partID int, data *PutObjReader, opts ObjectOptions) (info PartInfo, err error)
+	InnerAbortMultipartUploadFn    func(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) error
+	InnerCompleteMultipartUploadFn func(ctx context.Context, bucket, object, uploadID string, uploadedParts []CompletePart, opts ObjectOptions) (objInfo ObjectInfo, err error)
+	InnerCopyObjectPartFn          func(ctx context.Context, srcBucket, srcObject, dstBucket, dstObject, uploadID string, partID int, startOffset int64, length int64, srcInfo ObjectInfo, srcOpts, dstOpts ObjectOptions) (pi PartInfo, e error)
 }
 
 func (c *cacheObjects) incHitsToMeta(ctx context.Context, dcache *diskCache, bucket, object string, size int64, eTag string, rs *HTTPRangeSpec) error {
@@ -488,8 +507,6 @@ func (c *cacheObjects) CacheStats() (cs *CacheStats) {
 
 // skipCache() returns true if cache migration is in progress
 func (c *cacheObjects) skipCache() bool {
-	c.migMutex.Lock()
-	defer c.migMutex.Unlock()
 	return c.migrating
 }
 
@@ -619,8 +636,6 @@ func (c *cacheObjects) migrateCacheFromV1toV2(ctx context.Context) {
 	}
 
 	// update migration status
-	c.migMutex.Lock()
-	defer c.migMutex.Unlock()
 	c.migrating = false
 	logStartupMessage(color.Blue("Cache migration completed successfully."))
 }
@@ -759,13 +774,14 @@ func newServerCacheObjects(ctx context.Context, config cache.Config) (CacheObjec
 		return nil, err
 	}
 	c := &cacheObjects{
-		cache:           cache,
-		exclude:         config.Exclude,
-		after:           config.After,
-		migrating:       migrateSw,
-		migMutex:        sync.Mutex{},
-		commitWriteback: config.CommitWriteback,
-		cacheStats:      newCacheStats(),
+		cache:              cache,
+		exclude:            config.Exclude,
+		after:              config.After,
+		migrating:          migrateSw,
+		commitWriteback:    config.CacheCommitMode == CommitWriteBack,
+		commitWritethrough: config.CacheCommitMode == CommitWriteThrough,
+
+		cacheStats: newCacheStats(),
 		InnerGetObjectInfoFn: func(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) {
 			return newObjectLayerFn().GetObjectInfo(ctx, bucket, object, opts)
 		},
@@ -781,6 +797,21 @@ func newServerCacheObjects(ctx context.Context, config cache.Config) (CacheObjec
 		InnerCopyObjectFn: func(ctx context.Context, srcBucket, srcObject, destBucket, destObject string, srcInfo ObjectInfo, srcOpts, dstOpts ObjectOptions) (objInfo ObjectInfo, err error) {
 			return newObjectLayerFn().CopyObject(ctx, srcBucket, srcObject, destBucket, destObject, srcInfo, srcOpts, dstOpts)
 		},
+		InnerNewMultipartUploadFn: func(ctx context.Context, bucket, object string, opts ObjectOptions) (uploadID string, err error) {
+			return newObjectLayerFn().NewMultipartUpload(ctx, bucket, object, opts)
+		},
+		InnerPutObjectPartFn: func(ctx context.Context, bucket, object, uploadID string, partID int, data *PutObjReader, opts ObjectOptions) (info PartInfo, err error) {
+			return newObjectLayerFn().PutObjectPart(ctx, bucket, object, uploadID, partID, data, opts)
+		},
+		InnerAbortMultipartUploadFn: func(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) error {
+			return newObjectLayerFn().AbortMultipartUpload(ctx, bucket, object, uploadID, opts)
+		},
+		InnerCompleteMultipartUploadFn: func(ctx context.Context, bucket, object, uploadID string, uploadedParts []CompletePart, opts ObjectOptions) (objInfo ObjectInfo, err error) {
+			return newObjectLayerFn().CompleteMultipartUpload(ctx, bucket, object, uploadID, uploadedParts, opts)
+		},
+		InnerCopyObjectPartFn: func(ctx context.Context, srcBucket, srcObject, dstBucket, dstObject, uploadID string, partID int, startOffset int64, length int64, srcInfo ObjectInfo, srcOpts, dstOpts ObjectOptions) (pi PartInfo, e error) {
+			return newObjectLayerFn().CopyObjectPart(ctx, srcBucket, srcObject, dstBucket, dstObject, uploadID, partID, startOffset, length, srcInfo, srcOpts, dstOpts)
+		},
 	}
 	c.cacheStats.GetDiskStats = func() []CacheDiskStats {
 		cacheDiskStats := make([]CacheDiskStats, len(c.cache))
@@ -859,3 +890,247 @@ func (c *cacheObjects) queuePendingWriteback(ctx context.Context) {
 		}
 	}
 }
+
+// NewMultipartUpload - Starts a new multipart upload operation to backend - if writethrough mode is enabled, starts caching the multipart.
+func (c *cacheObjects) NewMultipartUpload(ctx context.Context, bucket, object string, opts ObjectOptions) (uploadID string, err error) {
+	newMultipartUploadFn := c.InnerNewMultipartUploadFn
+	dcache, err := c.getCacheToLoc(ctx, bucket, object)
+	if err != nil {
+		// disk cache could not be located,execute backend call.
+		return newMultipartUploadFn(ctx, bucket, object, opts)
+	}
+	if c.skipCache() {
+		return newMultipartUploadFn(ctx, bucket, object, opts)
+	}
+
+	if opts.ServerSideEncryption != nil { // avoid caching encrypted objects
+		dcache.Delete(ctx, bucket, object)
+		return newMultipartUploadFn(ctx, bucket, object, opts)
+	}
+
+	// skip cache for objects with locks
+	objRetention := objectlock.GetObjectRetentionMeta(opts.UserDefined)
+	legalHold := objectlock.GetObjectLegalHoldMeta(opts.UserDefined)
+	if objRetention.Mode.Valid() || legalHold.Status.Valid() {
+		dcache.Delete(ctx, bucket, object)
+		return newMultipartUploadFn(ctx, bucket, object, opts)
+	}
+
+	// fetch from backend if cache exclude pattern or cache-control
+	// directive set to exclude
+	if c.isCacheExclude(bucket, object) {
+		dcache.Delete(ctx, bucket, object)
+		return newMultipartUploadFn(ctx, bucket, object, opts)
+	}
+	if !c.commitWritethrough {
+		return newMultipartUploadFn(ctx, bucket, object, opts)
+	}
+
+	// perform multipart upload on backend and cache simultaneously
+	uploadID, err = newMultipartUploadFn(ctx, bucket, object, opts)
+	dcache.NewMultipartUpload(GlobalContext, bucket, object, uploadID, opts)
+	return uploadID, err
+}
+
+// PutObjectPart streams part to cache concurrently if writethrough mode is enabled. Otherwise redirects the call to remote
+func (c *cacheObjects) PutObjectPart(ctx context.Context, bucket, object, uploadID string, partID int, data *PutObjReader, opts ObjectOptions) (info PartInfo, err error) {
+	putObjectPartFn := c.InnerPutObjectPartFn
+	dcache, err := c.getCacheToLoc(ctx, bucket, object)
+	if err != nil {
+		// disk cache could not be located,execute backend call.
+		return putObjectPartFn(ctx, bucket, object, uploadID, partID, data, opts)
+	}
+
+	if !c.commitWritethrough {
+		return putObjectPartFn(ctx, bucket, object, uploadID, partID, data, opts)
+	}
+	if c.skipCache() {
+		return putObjectPartFn(ctx, bucket, object, uploadID, partID, data, opts)
+	}
+	size := data.Size()
+
+	// avoid caching part if space unavailable
+	if !dcache.diskSpaceAvailable(size) {
+		return putObjectPartFn(ctx, bucket, object, uploadID, partID, data, opts)
+	}
+
+	if opts.ServerSideEncryption != nil {
+		dcache.Delete(ctx, bucket, object)
+		return putObjectPartFn(ctx, bucket, object, uploadID, partID, data, opts)
+	}
+
+	// skip cache for objects with locks
+	objRetention := objectlock.GetObjectRetentionMeta(opts.UserDefined)
+	legalHold := objectlock.GetObjectLegalHoldMeta(opts.UserDefined)
+	if objRetention.Mode.Valid() || legalHold.Status.Valid() {
+		dcache.Delete(ctx, bucket, object)
+		return putObjectPartFn(ctx, bucket, object, uploadID, partID, data, opts)
+	}
+
+	// fetch from backend if cache exclude pattern or cache-control
+	// directive set to exclude
+	if c.isCacheExclude(bucket, object) {
+		dcache.Delete(ctx, bucket, object)
+		return putObjectPartFn(ctx, bucket, object, uploadID, partID, data, opts)
+	}
+
+	info = PartInfo{}
+	// Initialize pipe to stream data to backend
+	pipeReader, pipeWriter := io.Pipe()
+	hashReader, err := hash.NewReader(pipeReader, size, "", "", data.ActualSize())
+	if err != nil {
+		return
+	}
+	// Initialize pipe to stream data to cache
+	rPipe, wPipe := io.Pipe()
+	pinfoCh := make(chan PartInfo)
+	cinfoCh := make(chan PartInfo)
+
+	errorCh := make(chan error)
+	go func() {
+		info, err = putObjectPartFn(ctx, bucket, object, uploadID, partID, NewPutObjReader(hashReader), opts)
+		if err != nil {
+			close(pinfoCh)
+			pipeReader.CloseWithError(err)
+			rPipe.CloseWithError(err)
+			errorCh <- err
+			return
+		}
+		close(errorCh)
+		pinfoCh <- info
+	}()
+	go func() {
+		pinfo, perr := dcache.PutObjectPart(GlobalContext, bucket, object, uploadID, partID, rPipe, data.Size(), opts)
+		if perr != nil {
+			rPipe.CloseWithError(perr)
+			close(cinfoCh)
+			// clean up upload
+			dcache.AbortUpload(bucket, object, uploadID)
+			return
+		}
+		cinfoCh <- pinfo
+	}()
+
+	mwriter := cacheMultiWriter(pipeWriter, wPipe)
+	_, err = io.Copy(mwriter, data)
+	pipeWriter.Close()
+	wPipe.Close()
+
+	if err != nil {
+		err = <-errorCh
+		return PartInfo{}, err
+	}
+	info = <-pinfoCh
+	cachedInfo := <-cinfoCh
+	if info.PartNumber == cachedInfo.PartNumber {
+		cachedInfo.ETag = info.ETag
+		cachedInfo.LastModified = info.LastModified
+		dcache.SavePartMetadata(GlobalContext, bucket, object, uploadID, partID, cachedInfo)
+	}
+	return info, err
+}
+
+// CopyObjectPart behaves similar to PutObjectPart - caches part to upload dir if writethrough mode is enabled.
+func (c *cacheObjects) CopyObjectPart(ctx context.Context, srcBucket, srcObject, dstBucket, dstObject, uploadID string, partID int, startOffset int64, length int64, srcInfo ObjectInfo, srcOpts, dstOpts ObjectOptions) (pi PartInfo, e error) {
+	copyObjectPartFn := c.InnerCopyObjectPartFn
+	dcache, err := c.getCacheToLoc(ctx, dstBucket, dstObject)
+	if err != nil {
+		// disk cache could not be located,execute backend call.
+		return copyObjectPartFn(ctx, srcBucket, srcObject, dstBucket, dstObject, uploadID, partID, startOffset, length, srcInfo, srcOpts, dstOpts)
+	}
+
+	if !c.commitWritethrough {
+		return copyObjectPartFn(ctx, srcBucket, srcObject, dstBucket, dstObject, uploadID, partID, startOffset, length, srcInfo, srcOpts, dstOpts)
+	}
+	if err := dcache.uploadIDExists(dstBucket, dstObject, uploadID); err != nil {
+		return copyObjectPartFn(ctx, srcBucket, srcObject, dstBucket, dstObject, uploadID, partID, startOffset, length, srcInfo, srcOpts, dstOpts)
+	}
+	partInfo, err := copyObjectPartFn(ctx, srcBucket, srcObject, dstBucket, dstObject, uploadID, partID, startOffset, length, srcInfo, srcOpts, dstOpts)
+	if err != nil {
+		return pi, toObjectErr(err, dstBucket, dstObject)
+	}
+	go func() {
+		isSuffixLength := false
+		if startOffset < 0 {
+			isSuffixLength = true
+		}
+
+		rs := &HTTPRangeSpec{
+			IsSuffixLength: isSuffixLength,
+			Start:          startOffset,
+			End:            startOffset + length,
+		}
+		// fill cache in the background
+		bReader, bErr := c.InnerGetObjectNInfoFn(GlobalContext, srcBucket, srcObject, rs, http.Header{}, readLock, ObjectOptions{})
+		if bErr != nil {
+			return
+		}
+		defer bReader.Close()
+		// avoid cache overwrite if another background routine filled cache
+		dcache.PutObjectPart(GlobalContext, dstBucket, dstObject, uploadID, partID, bReader, length, ObjectOptions{UserDefined: getMetadata(bReader.ObjInfo)})
+	}()
+	// Success.
+	return partInfo, nil
+}
+
+// CompleteMultipartUpload - completes multipart upload operation on the backend. If writethrough mode is enabled, this also
+// finalizes the upload saved in cache multipart dir.
+func (c *cacheObjects) CompleteMultipartUpload(ctx context.Context, bucket, object, uploadID string, uploadedParts []CompletePart, opts ObjectOptions) (oi ObjectInfo, err error) {
+	completeMultipartUploadFn := c.InnerCompleteMultipartUploadFn
+	if !c.commitWritethrough {
+		return completeMultipartUploadFn(ctx, bucket, object, uploadID, uploadedParts, opts)
+	}
+	dcache, err := c.getCacheToLoc(ctx, bucket, object)
+	if err != nil {
+		// disk cache could not be located,execute backend call.
+		return completeMultipartUploadFn(ctx, bucket, object, uploadID, uploadedParts, opts)
+	}
+
+	// perform multipart upload on backend and cache simultaneously
+	oi, err = completeMultipartUploadFn(ctx, bucket, object, uploadID, uploadedParts, opts)
+	if err == nil {
+		// fill cache in the background
+		go func() {
+			_, err := dcache.CompleteMultipartUpload(ctx, bucket, object, uploadID, uploadedParts, oi, opts)
+			if err != nil {
+				// fill cache in the background
+				bReader, bErr := c.InnerGetObjectNInfoFn(GlobalContext, bucket, object, nil, http.Header{}, readLock, ObjectOptions{})
+				if bErr != nil {
+					return
+				}
+				defer bReader.Close()
+				oi, _, err := dcache.Stat(GlobalContext, bucket, object)
+				// avoid cache overwrite if another background routine filled cache
+				if err != nil || oi.ETag != bReader.ObjInfo.ETag {
+					dcache.Put(GlobalContext, bucket, object, bReader, bReader.ObjInfo.Size, nil, ObjectOptions{UserDefined: getMetadata(bReader.ObjInfo)}, false)
+				}
+			}
+		}()
+	}
+	return
+}
+
+// AbortMultipartUpload - aborts multipart upload on backend and cache.
+func (c *cacheObjects) AbortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) error {
+	abortMultipartUploadFn := c.InnerAbortMultipartUploadFn
+	if !c.commitWritethrough {
+		return abortMultipartUploadFn(ctx, bucket, object, uploadID, opts)
+	}
+	dcache, err := c.getCacheToLoc(ctx, bucket, object)
+	if err != nil {
+		// disk cache could not be located,execute backend call.
+		return abortMultipartUploadFn(ctx, bucket, object, uploadID, opts)
+	}
+	if err = dcache.uploadIDExists(bucket, object, uploadID); err != nil {
+		return toObjectErr(err, bucket, object, uploadID)
+	}
+
+	// execute backend operation
+	err = abortMultipartUploadFn(ctx, bucket, object, uploadID, opts)
+	if err != nil {
+		return err
+	}
+	// abort multipart upload on cache
+	go dcache.AbortUpload(bucket, object, uploadID)
+	return nil
+}