-
Notifications
You must be signed in to change notification settings - Fork 172
/
copy-multipart.go
102 lines (93 loc) · 3.29 KB
/
copy-multipart.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
package s3
import (
"context"
"io"
"io/ioutil"
"math"
"sort"
"sync"
context2 "github.com/pydio/cells/common/utils/context"
"github.com/pydio/minio-go"
"go.uber.org/zap"
"github.com/pydio/cells/common/log"
)
// minPartSize - minimum part size 64MiB per object after which
// putObject behaves internally as multipart.
const minPartSize = 1024 * 1024 * 64
// maxPartsCount - maximum number of parts for a single multipart session.
const maxPartsCount = 10000
// Max CopyObjectSize is 5GB => if greater, we have to switch to multipart
const MaxCopyObjectSize = 1024 * 1024 * 1024 * 5
func optimalPartInfo(objectSize int64) (totalPartsCount int, partSize int64, lastPartSize int64) {
// Use floats for part size for all calculations to avoid
// overflows during float64 to int64 conversions.
partSizeFlt := math.Ceil(float64(objectSize / maxPartsCount))
partSizeFlt = math.Ceil(partSizeFlt/minPartSize) * minPartSize
// Total parts count.
totalPartsCount = int(math.Ceil(float64(objectSize) / partSizeFlt))
// Part size.
partSize = int64(partSizeFlt)
// Last part size.
lastPartSize = objectSize - int64(totalPartsCount-1)*partSize
return totalPartsCount, partSize, lastPartSize
}
func CopyObjectMultipart(ctx context.Context, client *minio.Core, srcObject minio.ObjectInfo, srcBucket, srcPath, destBucket, destPath string, metadata map[string]string, progress io.Reader) error {
log.Logger(ctx).Debug("Entering MultipartUpload for COPY")
if metadata != nil {
ctx = context2.WithAdditionalMetadata(ctx, metadata)
}
// We have to use multipart copy
uploadID, err := client.NewMultipartUploadWithContext(ctx, destBucket, destPath, minio.PutObjectOptions{UserMetadata: metadata})
if err != nil {
log.Logger(ctx).Error("New Multipart Error", zap.Error(err))
return err
}
partSize := srcObject.Size / 10
totalPartsCount, partSize, lastPartSize := optimalPartInfo(srcObject.Size)
var parts []minio.CompletePart
queue := make(chan struct{}, 15)
wg := &sync.WaitGroup{}
wg.Add(totalPartsCount)
var copyErr error
for i := 0; i < totalPartsCount; i++ {
queue <- struct{}{}
go func(index int) {
defer func() {
wg.Done()
<-queue
}()
if copyErr != nil {
return
}
length := partSize
if index == totalPartsCount-1 {
length = lastPartSize
}
startOffset := int64(index) * partSize
log.Logger(ctx).Debug("COPY PART", zap.Int("part", index), zap.Int64("offset", startOffset), zap.Int64("length", length))
p, er := client.CopyObjectPartWithContext(ctx, srcBucket, srcPath, destBucket, destPath, uploadID, index+1, startOffset, length, nil)
if er != nil {
log.Logger(ctx).Error("CopyObjectPart Error - other parts will be ignored", zap.Error(err))
copyErr = er
} else if progress != nil {
io.CopyN(ioutil.Discard, progress, length)
}
parts = append(parts, p)
}(i)
}
wg.Wait()
if copyErr != nil {
client.AbortMultipartUploadWithContext(ctx, destBucket, destPath, uploadID)
return copyErr
}
// Resort parts in correct order
sort.Slice(parts, func(i, j int) bool {
return parts[i].PartNumber < parts[j].PartNumber
})
log.Logger(ctx).Debug("Finishing Multipart")
_, er := client.CompleteMultipartUploadWithContext(ctx, destBucket, destPath, uploadID, parts)
if er != nil {
log.Logger(ctx).Error("CompleteMultipart Error", zap.Error(err))
}
return er
}