/
resumable.go
280 lines (232 loc) · 9 KB
/
resumable.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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
package kodocli
import (
"context"
"errors"
"io"
"os"
"sync"
"github.com/service-sdk/go-sdk-qn/v2/operation/internal/x/xlog.v7"
)
// ----------------------------------------------------------
var (
ErrInvalidPutProgress = errors.New("invalid put progress")
ErrPutFailed = errors.New("resumable put failed")
ErrUnmatchedChecksum = errors.New("unmatched checksum")
)
const (
InvalidCtx = 701 // UP: 无效的上下文(bput),可能情况:Ctx非法或者已经被淘汰(太久未使用)
)
const (
defaultWorkers = 4
defaultChunkSize = 256 * 1024 // 256k
defaultTryTimes = 3
)
type Settings struct {
TaskQsize int // 可选。任务队列大小。为 0 表示取 Workers * 4。
Workers int // 并行 Goroutine 数目。
ChunkSize int // 默认的Chunk大小,不设定则为256k
TryTimes int // 默认的尝试次数,不设定则为3
}
var settings = Settings{
TaskQsize: defaultWorkers * 4,
Workers: defaultWorkers,
ChunkSize: defaultChunkSize,
TryTimes: defaultTryTimes,
}
func SetSettings(v *Settings) {
settings = *v
if settings.Workers == 0 {
settings.Workers = defaultWorkers
}
if settings.TaskQsize == 0 {
settings.TaskQsize = settings.Workers * 4
}
if settings.ChunkSize == 0 {
settings.ChunkSize = defaultChunkSize
}
if settings.TryTimes == 0 {
settings.TryTimes = defaultTryTimes
}
}
// ----------------------------------------------------------
var tasks chan func()
func worker(tasks chan func()) {
for {
task := <-tasks
task()
}
}
func initWorkers() {
tasks = make(chan func(), settings.TaskQsize)
for i := 0; i < settings.Workers; i++ {
go worker(tasks)
}
}
func notifyNil(blkIdx int, blkSize int, ret *BlkputRet) {}
func notifyErrNil(blkIdx int, blkSize int, err error) {}
// ----------------------------------------------------------
const (
blockBits = 22
blockMask = (1 << blockBits) - 1
)
func BlockCount(fsize int64) int {
return int((fsize + blockMask) >> blockBits)
}
// ----------------------------------------------------------
type BlkputRet struct {
Ctx string `json:"ctx"`
Checksum string `json:"checksum"`
Crc32 uint32 `json:"crc32"`
Offset uint32 `json:"offset"`
Host string `json:"host"`
}
type RputExtra struct {
Params map[string]string // 可选。用户自定义参数,以"x:"开头 否则忽略
XMeta map[string]string // 可选。
MimeType string // 可选。
ChunkSize int // 可选。每次上传的Chunk大小
TryTimes int // 可选。尝试次数
Progresses []BlkputRet // 可选。上传进度
Notify func(blkIdx int, blkSize int, ret *BlkputRet) // 可选。进度提示(注意多个block是并行传输的)
NotifyErr func(blkIdx int, blkSize int, err error)
}
var once sync.Once
// ----------------------------------------------------------
// 上传一个文件,支持断点续传和分块上传。
//
// ctx 是请求的上下文。
// ret 是上传成功后返回的数据。如果 uptoken 中没有设置 CallbackUrl 或 ReturnBody,那么返回的数据结构是 PutRet 结构。
// uptoken 是由业务服务器颁发的上传凭证。
// key 是要上传的文件访问路径。比如:"foo/bar.jpg"。注意我们建议 key 不要以 '/' 开头。另外,key 为空字符串是合法的。
// f 是文件内容的访问接口。考虑到需要支持分块上传和断点续传,要的是 io.ReaderAt 接口,而不是 io.Reader。
// fsize 是要上传的文件大小。
// extra 是上传的一些可选项。详细见 RputExtra 结构的描述。
func (p Uploader) Rput(
ctx context.Context, ret interface{}, uptoken string,
key string, f io.ReaderAt, fsize int64, extra *RputExtra) error {
return p.rput(ctx, ret, uptoken, key, true, f, fsize, extra)
}
// 上传一个文件,支持断点续传和分块上传。文件的访问路径(key)自动生成。
// 如果 uptoken 中设置了 SaveKey,那么按 SaveKey 要求的规则生成 key,否则自动以文件的 hash 做 key。
//
// ctx 是请求的上下文。
// ret 是上传成功后返回的数据。如果 uptoken 中没有设置 CallbackUrl 或 ReturnBody,那么返回的数据结构是 PutRet 结构。
// uptoken 是由业务服务器颁发的上传凭证。
// f 是文件内容的访问接口。考虑到需要支持分块上传和断点续传,要的是 io.ReaderAt 接口,而不是 io.Reader。
// fsize 是要上传的文件大小。
// extra 是上传的一些可选项。详细见 RputExtra 结构的描述。
func (p Uploader) RputWithoutKey(
ctx context.Context, ret interface{}, uptoken string, f io.ReaderAt, fsize int64, extra *RputExtra) error {
return p.rput(ctx, ret, uptoken, "", false, f, fsize, extra)
}
// 上传一个文件,支持断点续传和分块上传。
// 和 Rput 不同的只是一个通过提供文件路径来访问文件内容,一个通过 io.ReaderAt 来访问。
//
// ctx 是请求的上下文。
// ret 是上传成功后返回的数据。如果 uptoken 中没有设置 CallbackUrl 或 ReturnBody,那么返回的数据结构是 PutRet 结构。
// uptoken 是由业务服务器颁发的上传凭证。
// key 是要上传的文件访问路径。比如:"foo/bar.jpg"。注意我们建议 key 不要以 '/' 开头。另外,key 为空字符串是合法的。
// localFile 是要上传的文件的本地路径。
// extra 是上传的一些可选项。详细见 RputExtra 结构的描述。
func (p Uploader) RputFile(
ctx context.Context, ret interface{}, uptoken, key, localFile string, extra *RputExtra) (err error) {
return p.rputFile(ctx, ret, uptoken, key, true, localFile, extra)
}
// 上传一个文件,支持断点续传和分块上传。文件的访问路径(key)自动生成。
// 如果 uptoken 中设置了 SaveKey,那么按 SaveKey 要求的规则生成 key,否则自动以文件的 hash 做 key。
// 和 RputWithoutKey 不同的只是一个通过提供文件路径来访问文件内容,一个通过 io.ReaderAt 来访问。
//
// ctx 是请求的上下文。
// ret 是上传成功后返回的数据。如果 uptoken 中没有设置 CallbackUrl 或 ReturnBody,那么返回的数据结构是 PutRet 结构。
// uptoken 是由业务服务器颁发的上传凭证。
// localFile 是要上传的文件的本地路径。
// extra 是上传的一些可选项。详细见 RputExtra 结构的描述。
func (p Uploader) RputFileWithoutKey(
ctx context.Context, ret interface{}, uptoken, localFile string, extra *RputExtra) (err error) {
return p.rputFile(ctx, ret, uptoken, "", false, localFile, extra)
}
// ----------------------------------------------------------
func (p Uploader) rput(
ctx context.Context, ret interface{}, uptoken string,
key string, hasKey bool, f io.ReaderAt, fsize int64, extra *RputExtra) error {
once.Do(initWorkers)
xl := xlog.NewWith(ctx)
blockCnt := BlockCount(fsize)
if extra == nil {
extra = new(RputExtra)
}
if extra.Progresses == nil {
extra.Progresses = make([]BlkputRet, blockCnt)
} else if len(extra.Progresses) != blockCnt {
return ErrInvalidPutProgress
}
if extra.ChunkSize == 0 {
extra.ChunkSize = settings.ChunkSize
}
if extra.TryTimes == 0 {
extra.TryTimes = settings.TryTimes
}
if extra.Notify == nil {
extra.Notify = notifyNil
}
if extra.NotifyErr == nil {
extra.NotifyErr = notifyErrNil
}
var wg sync.WaitGroup
wg.Add(blockCnt)
last := blockCnt - 1
blkSize := 1 << blockBits
nfails := 0
p.Conn.Client = newUptokenClient(uptoken, p.Conn.Transport)
failedUpHosts := make(map[string]struct{})
for i := 0; i < blockCnt; i++ {
blkIdx := i
blkSize1 := blkSize
if i == last {
offbase := int64(blkIdx) << blockBits
blkSize1 = int(fsize - offbase)
}
task := func() {
defer wg.Done()
tryTimes := extra.TryTimes
lzRetry:
upHost := p.chooseUpHost(failedUpHosts)
err := p.resumableBput(ctx, upHost, &extra.Progresses[blkIdx], f, blkIdx, blkSize1, extra)
if err != nil {
failedUpHosts[upHost] = struct{}{}
failHostName(upHost)
if tryTimes > 1 {
tryTimes--
elog.Infof("[%s] resumable.Put retrying ...", xl.ReqId)
goto lzRetry
}
elog.Warnf("[%s] resumable.Put failed: blkIdx=%d err=%s", xl.ReqId, blkIdx, err)
extra.NotifyErr(blkIdx, blkSize1, err)
nfails++
} else {
succeedHostName(upHost)
}
}
tasks <- task
}
wg.Wait()
if nfails != 0 {
return ErrPutFailed
}
return p.mkfile(ctx, p.chooseUpHost(failedUpHosts), ret, key, hasKey, fsize, extra)
}
func (p Uploader) rputFile(
ctx context.Context, ret interface{}, uptoken string,
key string, hasKey bool, localFile string, extra *RputExtra) (err error) {
f, err := os.Open(localFile)
if err != nil {
return
}
defer f.Close()
fi, err := f.Stat()
if err != nil {
return
}
return p.rput(ctx, ret, uptoken, key, hasKey, f, fi.Size(), extra)
}
// ----------------------------------------------------------