oracleobjectstorage: Use rclone's rate limiter in mutipart transfers
This commit is contained in:
Manoj Ghosh
Nick Craig-Wood
parent
de185de215
commit
27f5297e8d
10 changed files with 1152 additions and 159 deletions
334
backend/oracleobjectstorage/multipart.go
Normal file
334
backend/oracleobjectstorage/multipart.go
Normal file
|
|
@ -0,0 +1,334 @@
|
|||
//go:build !plan9 && !solaris && !js
|
||||
// +build !plan9,!solaris,!js
|
||||
|
||||
package oracleobjectstorage
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/md5"
|
||||
"encoding/base64"
|
||||
"fmt"
|
||||
"io"
|
||||
"sort"
|
||||
"strconv"
|
||||
"sync"
|
||||
|
||||
"github.com/oracle/oci-go-sdk/v65/common"
|
||||
"github.com/oracle/oci-go-sdk/v65/objectstorage"
|
||||
"github.com/rclone/rclone/fs"
|
||||
"github.com/rclone/rclone/fs/chunksize"
|
||||
"github.com/rclone/rclone/lib/atexit"
|
||||
"github.com/rclone/rclone/lib/pacer"
|
||||
"github.com/rclone/rclone/lib/readers"
|
||||
"golang.org/x/sync/errgroup"
|
||||
)
|
||||
|
||||
var warnStreamUpload sync.Once
|
||||
|
||||
func (o *Object) uploadMultipart(
|
||||
ctx context.Context,
|
||||
putReq *objectstorage.PutObjectRequest,
|
||||
in io.Reader,
|
||||
src fs.ObjectInfo) (err error) {
|
||||
uploadID, uploadedParts, err := o.createMultipartUpload(ctx, putReq)
|
||||
if err != nil {
|
||||
fs.Errorf(o, "failed to create multipart upload-id err: %v", err)
|
||||
return err
|
||||
}
|
||||
return o.uploadParts(ctx, putReq, in, src, uploadID, uploadedParts)
|
||||
}
|
||||
|
||||
func (o *Object) createMultipartUpload(ctx context.Context, putReq *objectstorage.PutObjectRequest) (
|
||||
uploadID string, uploadedParts map[int]objectstorage.MultipartUploadPartSummary, err error) {
|
||||
bucketName, bucketPath := o.split()
|
||||
f := o.fs
|
||||
if f.opt.AttemptResumeUpload {
|
||||
fs.Debugf(o, "attempting to resume upload for %v (if any)", o.remote)
|
||||
resumeUploads, err := o.fs.findLatestMultipartUpload(ctx, bucketName, bucketPath)
|
||||
if err == nil && len(resumeUploads) > 0 {
|
||||
uploadID = *resumeUploads[0].UploadId
|
||||
uploadedParts, err = f.listMultipartUploadParts(ctx, bucketName, bucketPath, uploadID)
|
||||
if err == nil {
|
||||
fs.Debugf(o, "resuming with existing upload id: %v", uploadID)
|
||||
return uploadID, uploadedParts, err
|
||||
}
|
||||
}
|
||||
}
|
||||
req := objectstorage.CreateMultipartUploadRequest{
|
||||
NamespaceName: common.String(o.fs.opt.Namespace),
|
||||
BucketName: common.String(bucketName),
|
||||
}
|
||||
req.Object = common.String(bucketPath)
|
||||
if o.fs.opt.StorageTier != "" {
|
||||
storageTier, ok := objectstorage.GetMappingStorageTierEnum(o.fs.opt.StorageTier)
|
||||
if !ok {
|
||||
return "", nil, fmt.Errorf("not a valid storage tier: %v", o.fs.opt.StorageTier)
|
||||
}
|
||||
req.StorageTier = storageTier
|
||||
}
|
||||
o.applyMultipartUploadOptions(putReq, &req)
|
||||
|
||||
var resp objectstorage.CreateMultipartUploadResponse
|
||||
err = o.fs.pacer.Call(func() (bool, error) {
|
||||
resp, err = o.fs.srv.CreateMultipartUpload(ctx, req)
|
||||
return shouldRetry(ctx, resp.HTTPResponse(), err)
|
||||
})
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
uploadID = *resp.UploadId
|
||||
fs.Debugf(o, "created new upload id: %v", uploadID)
|
||||
return uploadID, nil, err
|
||||
}
|
||||
|
||||
func (o *Object) uploadParts(
|
||||
ctx context.Context,
|
||||
putReq *objectstorage.PutObjectRequest,
|
||||
in io.Reader,
|
||||
src fs.ObjectInfo,
|
||||
uploadID string,
|
||||
uploadedParts map[int]objectstorage.MultipartUploadPartSummary) (err error) {
|
||||
bucketName, bucketPath := o.split()
|
||||
f := o.fs
|
||||
|
||||
// make concurrency machinery
|
||||
concurrency := f.opt.UploadConcurrency
|
||||
if concurrency < 1 {
|
||||
concurrency = 1
|
||||
}
|
||||
|
||||
uploadParts := f.opt.MaxUploadParts
|
||||
if uploadParts < 1 {
|
||||
uploadParts = 1
|
||||
} else if uploadParts > maxUploadParts {
|
||||
uploadParts = maxUploadParts
|
||||
}
|
||||
|
||||
// calculate size of parts
|
||||
partSize := f.opt.ChunkSize
|
||||
fileSize := src.Size()
|
||||
|
||||
// size can be -1 here meaning we don't know the size of the incoming file. We use ChunkSize
|
||||
// buffers here (default 5 MiB). With a maximum number of parts (10,000) this will be a file of
|
||||
// 48 GiB which seems like a not too unreasonable limit.
|
||||
if fileSize == -1 {
|
||||
warnStreamUpload.Do(func() {
|
||||
fs.Logf(f, "Streaming uploads using chunk size %v will have maximum file size of %v",
|
||||
f.opt.ChunkSize, fs.SizeSuffix(int64(partSize)*int64(uploadParts)))
|
||||
})
|
||||
} else {
|
||||
partSize = chunksize.Calculator(o, fileSize, uploadParts, f.opt.ChunkSize)
|
||||
}
|
||||
|
||||
uploadCtx, cancel := context.WithCancel(ctx)
|
||||
defer atexit.OnError(&err, func() {
|
||||
cancel()
|
||||
if o.fs.opt.LeavePartsOnError {
|
||||
return
|
||||
}
|
||||
fs.Debugf(o, "Cancelling multipart upload")
|
||||
errCancel := o.fs.abortMultiPartUpload(
|
||||
context.Background(),
|
||||
bucketName,
|
||||
bucketPath,
|
||||
uploadID)
|
||||
if errCancel != nil {
|
||||
fs.Debugf(o, "Failed to cancel multipart upload: %v", errCancel)
|
||||
} else {
|
||||
fs.Debugf(o, "canceled and aborted multipart upload: %v", uploadID)
|
||||
}
|
||||
})()
|
||||
|
||||
var (
|
||||
g, gCtx = errgroup.WithContext(uploadCtx)
|
||||
finished = false
|
||||
partsMu sync.Mutex // to protect parts
|
||||
parts []*objectstorage.CommitMultipartUploadPartDetails
|
||||
off int64
|
||||
md5sMu sync.Mutex
|
||||
md5s []byte
|
||||
tokens = pacer.NewTokenDispenser(concurrency)
|
||||
memPool = o.fs.getMemoryPool(int64(partSize))
|
||||
)
|
||||
|
||||
addMd5 := func(md5binary *[md5.Size]byte, partNum int64) {
|
||||
md5sMu.Lock()
|
||||
defer md5sMu.Unlock()
|
||||
start := partNum * md5.Size
|
||||
end := start + md5.Size
|
||||
if extend := end - int64(len(md5s)); extend > 0 {
|
||||
md5s = append(md5s, make([]byte, extend)...)
|
||||
}
|
||||
copy(md5s[start:end], (*md5binary)[:])
|
||||
}
|
||||
|
||||
for partNum := int64(1); !finished; partNum++ {
|
||||
// Get a block of memory from the pool and token which limits concurrency.
|
||||
tokens.Get()
|
||||
buf := memPool.Get()
|
||||
|
||||
free := func() {
|
||||
// return the memory and token
|
||||
memPool.Put(buf)
|
||||
tokens.Put()
|
||||
}
|
||||
|
||||
// Fail fast, in case an errgroup managed function returns an error
|
||||
// gCtx is cancelled. There is no point in uploading all the other parts.
|
||||
if gCtx.Err() != nil {
|
||||
free()
|
||||
break
|
||||
}
|
||||
|
||||
// Read the chunk
|
||||
var n int
|
||||
n, err = readers.ReadFill(in, buf) // this can never return 0, nil
|
||||
if err == io.EOF {
|
||||
if n == 0 && partNum != 1 { // end if no data and if not first chunk
|
||||
free()
|
||||
break
|
||||
}
|
||||
finished = true
|
||||
} else if err != nil {
|
||||
free()
|
||||
return fmt.Errorf("multipart upload failed to read source: %w", err)
|
||||
}
|
||||
buf = buf[:n]
|
||||
|
||||
partNum := partNum
|
||||
fs.Debugf(o, "multipart upload starting chunk %d size %v offset %v/%v", partNum, fs.SizeSuffix(n), fs.SizeSuffix(off), fs.SizeSuffix(fileSize))
|
||||
off += int64(n)
|
||||
g.Go(func() (err error) {
|
||||
defer free()
|
||||
partLength := int64(len(buf))
|
||||
|
||||
// create checksum of buffer for integrity checking
|
||||
md5sumBinary := md5.Sum(buf)
|
||||
addMd5(&md5sumBinary, partNum-1)
|
||||
md5sum := base64.StdEncoding.EncodeToString(md5sumBinary[:])
|
||||
if uploadedPart, ok := uploadedParts[int(partNum)]; ok {
|
||||
if md5sum == *uploadedPart.Md5 {
|
||||
fs.Debugf(o, "matched uploaded part found, part num %d, skipping part, md5=%v", partNum, md5sum)
|
||||
partsMu.Lock()
|
||||
parts = append(parts, &objectstorage.CommitMultipartUploadPartDetails{
|
||||
PartNum: uploadedPart.PartNumber,
|
||||
Etag: uploadedPart.Etag,
|
||||
})
|
||||
partsMu.Unlock()
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
req := objectstorage.UploadPartRequest{
|
||||
NamespaceName: common.String(o.fs.opt.Namespace),
|
||||
BucketName: common.String(bucketName),
|
||||
ObjectName: common.String(bucketPath),
|
||||
UploadId: common.String(uploadID),
|
||||
UploadPartNum: common.Int(int(partNum)),
|
||||
ContentLength: common.Int64(partLength),
|
||||
ContentMD5: common.String(md5sum),
|
||||
}
|
||||
o.applyPartUploadOptions(putReq, &req)
|
||||
var resp objectstorage.UploadPartResponse
|
||||
err = f.pacer.Call(func() (bool, error) {
|
||||
req.UploadPartBody = io.NopCloser(bytes.NewReader(buf))
|
||||
resp, err = f.srv.UploadPart(gCtx, req)
|
||||
if err != nil {
|
||||
if partNum <= int64(concurrency) {
|
||||
return shouldRetry(gCtx, resp.HTTPResponse(), err)
|
||||
}
|
||||
// retry all chunks once have done the first batch
|
||||
return true, err
|
||||
}
|
||||
partsMu.Lock()
|
||||
parts = append(parts, &objectstorage.CommitMultipartUploadPartDetails{
|
||||
PartNum: common.Int(int(partNum)),
|
||||
Etag: resp.ETag,
|
||||
})
|
||||
partsMu.Unlock()
|
||||
return false, nil
|
||||
})
|
||||
if err != nil {
|
||||
fs.Errorf(o, "multipart upload failed to upload part:%d err: %v", partNum, err)
|
||||
return fmt.Errorf("multipart upload failed to upload part: %w", err)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
err = g.Wait()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// sort the completed parts by part number
|
||||
sort.Slice(parts, func(i, j int) bool {
|
||||
return *parts[i].PartNum < *parts[j].PartNum
|
||||
})
|
||||
|
||||
var resp objectstorage.CommitMultipartUploadResponse
|
||||
resp, err = o.commitMultiPart(ctx, uploadID, parts)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
fs.Debugf(o, "multipart upload %v committed.", uploadID)
|
||||
hashOfHashes := md5.Sum(md5s)
|
||||
wantMultipartMd5 := base64.StdEncoding.EncodeToString(hashOfHashes[:]) + "-" + strconv.Itoa(len(parts))
|
||||
gotMultipartMd5 := *resp.OpcMultipartMd5
|
||||
if wantMultipartMd5 != gotMultipartMd5 {
|
||||
fs.Errorf(o, "multipart upload corrupted: multipart md5 differ: expecting %s but got %s", wantMultipartMd5, gotMultipartMd5)
|
||||
return fmt.Errorf("multipart upload corrupted: md5 differ: expecting %s but got %s", wantMultipartMd5, gotMultipartMd5)
|
||||
}
|
||||
fs.Debugf(o, "multipart upload %v md5 matched: expecting %s and got %s", uploadID, wantMultipartMd5, gotMultipartMd5)
|
||||
return nil
|
||||
}
|
||||
|
||||
// commits the multipart upload
|
||||
func (o *Object) commitMultiPart(ctx context.Context, uploadID string, parts []*objectstorage.CommitMultipartUploadPartDetails) (resp objectstorage.CommitMultipartUploadResponse, err error) {
|
||||
bucketName, bucketPath := o.split()
|
||||
req := objectstorage.CommitMultipartUploadRequest{
|
||||
NamespaceName: common.String(o.fs.opt.Namespace),
|
||||
BucketName: common.String(bucketName),
|
||||
ObjectName: common.String(bucketPath),
|
||||
UploadId: common.String(uploadID),
|
||||
}
|
||||
var partsToCommit []objectstorage.CommitMultipartUploadPartDetails
|
||||
for _, part := range parts {
|
||||
partsToCommit = append(partsToCommit, *part)
|
||||
}
|
||||
req.PartsToCommit = partsToCommit
|
||||
err = o.fs.pacer.Call(func() (bool, error) {
|
||||
resp, err = o.fs.srv.CommitMultipartUpload(ctx, req)
|
||||
// if multipart is corrupted, we will abort the uploadId
|
||||
if o.isMultiPartUploadCorrupted(err) {
|
||||
fs.Debugf(o, "multipart uploadId %v is corrupted, aborting...", uploadID)
|
||||
errCancel := o.fs.abortMultiPartUpload(
|
||||
context.Background(),
|
||||
bucketName,
|
||||
bucketPath,
|
||||
uploadID)
|
||||
if errCancel != nil {
|
||||
fs.Debugf(o, "Failed to abort multipart upload: %v, ignoring.", errCancel)
|
||||
} else {
|
||||
fs.Debugf(o, "aborted multipart upload: %v", uploadID)
|
||||
}
|
||||
return false, err
|
||||
}
|
||||
return shouldRetry(ctx, resp.HTTPResponse(), err)
|
||||
})
|
||||
return resp, err
|
||||
}
|
||||
|
||||
func (o *Object) isMultiPartUploadCorrupted(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
// Check if this ocierr object, and if it is multipart commit error
|
||||
if ociError, ok := err.(common.ServiceError); ok {
|
||||
// If it is a timeout then we want to retry that
|
||||
if ociError.GetCode() == "InvalidUploadPart" {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue