2016-11-21 08:42:53 +01:00
|
|
|
/*
|
|
|
|
* Minio Cloud Storage, (C) 2016 Minio, Inc.
|
|
|
|
*
|
|
|
|
* Licensed under the Apache License, Version 2.0 (the "License");
|
|
|
|
* you may not use this file except in compliance with the License.
|
|
|
|
* You may obtain a copy of the License at
|
|
|
|
*
|
|
|
|
* http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
*
|
|
|
|
* Unless required by applicable law or agreed to in writing, software
|
|
|
|
* distributed under the License is distributed on an "AS IS" BASIS,
|
|
|
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
|
|
* See the License for the specific language governing permissions and
|
|
|
|
* limitations under the License.
|
|
|
|
*/
|
|
|
|
|
|
|
|
package cmd
|
|
|
|
|
|
|
|
import (
|
|
|
|
"errors"
|
2017-01-17 02:05:00 +01:00
|
|
|
"io"
|
|
|
|
"os"
|
2016-11-21 08:42:53 +01:00
|
|
|
"reflect"
|
|
|
|
"sync"
|
|
|
|
"time"
|
|
|
|
)
|
|
|
|
|
|
|
|
// Error sent by appendParts go-routine when there are holes in parts.
|
|
|
|
// For ex. let's say client uploads part-2 before part-1 in which case we
|
|
|
|
// can not append and have to wait till part-1 is uploaded. Hence we return
|
|
|
|
// this error. Currently this error is not used in the caller.
|
|
|
|
var errPartsMissing = errors.New("required parts missing")
|
|
|
|
|
|
|
|
// Error sent when appendParts go-routine has waited long enough and timedout.
|
2017-01-17 02:05:00 +01:00
|
|
|
var errAppendPartsTimeout = errors.New("appendParts go-routine timeout")
|
2016-11-21 08:42:53 +01:00
|
|
|
|
|
|
|
// Timeout value for the appendParts go-routine.
|
2017-01-17 02:05:00 +01:00
|
|
|
var appendPartsTimeout = 24 * 60 * 60 * time.Second // 24 Hours.
|
2016-11-21 08:42:53 +01:00
|
|
|
|
|
|
|
// Holds a map of uploadID->appendParts go-routine
|
|
|
|
type backgroundAppend struct {
|
|
|
|
sync.Mutex
|
2017-01-17 02:05:00 +01:00
|
|
|
infoMap map[string]bgAppendPartsInfo
|
|
|
|
appendFile io.WriteCloser
|
2016-11-21 08:42:53 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
// Input to the appendParts go-routine
|
|
|
|
type bgAppendPartsInput struct {
|
|
|
|
meta fsMetaV1 // list of parts that need to be appended
|
|
|
|
errCh chan error // error sent by appendParts go-routine
|
|
|
|
}
|
|
|
|
|
|
|
|
// Identifies an appendParts go-routine.
|
|
|
|
type bgAppendPartsInfo struct {
|
2016-11-23 12:04:04 +01:00
|
|
|
inputCh chan bgAppendPartsInput
|
|
|
|
timeoutCh chan struct{} // closed by appendParts go-routine when it timesout
|
|
|
|
abortCh chan struct{} // closed after abort of upload to end the appendParts go-routine
|
|
|
|
completeCh chan struct{} // closed after complete of upload to end the appendParts go-routine
|
2016-11-21 08:42:53 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
// Called after a part is uploaded so that it can be appended in the background.
|
2017-01-17 02:05:00 +01:00
|
|
|
func (fs fsObjects) append(bucket, object, uploadID string, meta fsMetaV1) chan error {
|
|
|
|
fs.bgAppend.Lock()
|
|
|
|
info, ok := fs.bgAppend.infoMap[uploadID]
|
2016-11-21 08:42:53 +01:00
|
|
|
if !ok {
|
|
|
|
// Corresponding appendParts go-routine was not found, create a new one. Would happen when the first
|
|
|
|
// part of a multipart upload is uploaded.
|
|
|
|
inputCh := make(chan bgAppendPartsInput)
|
|
|
|
timeoutCh := make(chan struct{})
|
2016-11-23 12:04:04 +01:00
|
|
|
abortCh := make(chan struct{})
|
|
|
|
completeCh := make(chan struct{})
|
2016-11-21 08:42:53 +01:00
|
|
|
|
2016-11-23 12:04:04 +01:00
|
|
|
info = bgAppendPartsInfo{inputCh, timeoutCh, abortCh, completeCh}
|
2017-01-17 02:05:00 +01:00
|
|
|
fs.bgAppend.infoMap[uploadID] = info
|
2016-11-21 08:42:53 +01:00
|
|
|
|
2017-01-17 02:05:00 +01:00
|
|
|
go fs.appendParts(bucket, object, uploadID, info)
|
2016-11-21 08:42:53 +01:00
|
|
|
}
|
2017-01-17 02:05:00 +01:00
|
|
|
fs.bgAppend.Unlock()
|
|
|
|
|
2016-11-23 21:50:09 +01:00
|
|
|
errCh := make(chan error)
|
2016-11-21 08:42:53 +01:00
|
|
|
go func() {
|
|
|
|
// send input in a goroutine as send on the inputCh can block if appendParts go-routine
|
|
|
|
// is busy appending a part.
|
|
|
|
select {
|
|
|
|
case <-info.timeoutCh:
|
|
|
|
// This is to handle a rare race condition where we found info in b.infoMap
|
|
|
|
// but soon after that appendParts go-routine timed out.
|
2016-11-23 21:50:09 +01:00
|
|
|
errCh <- errAppendPartsTimeout
|
2016-11-21 08:42:53 +01:00
|
|
|
case info.inputCh <- bgAppendPartsInput{meta, errCh}:
|
|
|
|
}
|
|
|
|
}()
|
2017-01-17 02:05:00 +01:00
|
|
|
|
2016-11-23 21:50:09 +01:00
|
|
|
return errCh
|
2016-11-21 08:42:53 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
// Called on complete-multipart-upload. Returns nil if the required parts have been appended.
|
2017-01-17 02:05:00 +01:00
|
|
|
func (fs *fsObjects) complete(bucket, object, uploadID string, meta fsMetaV1) error {
|
|
|
|
fs.bgAppend.Lock()
|
|
|
|
defer fs.bgAppend.Unlock()
|
|
|
|
|
|
|
|
info, ok := fs.bgAppend.infoMap[uploadID]
|
|
|
|
delete(fs.bgAppend.infoMap, uploadID)
|
2016-11-21 08:42:53 +01:00
|
|
|
if !ok {
|
|
|
|
return errPartsMissing
|
|
|
|
}
|
2017-01-17 02:05:00 +01:00
|
|
|
|
2016-11-21 08:42:53 +01:00
|
|
|
errCh := make(chan error)
|
2017-01-17 02:05:00 +01:00
|
|
|
|
2016-11-21 08:42:53 +01:00
|
|
|
select {
|
|
|
|
case <-info.timeoutCh:
|
|
|
|
// This is to handle a rare race condition where we found info in b.infoMap
|
|
|
|
// but soon after that appendParts go-routine timedouted out.
|
|
|
|
return errAppendPartsTimeout
|
|
|
|
case info.inputCh <- bgAppendPartsInput{meta, errCh}:
|
|
|
|
}
|
2017-01-17 02:05:00 +01:00
|
|
|
|
2016-11-21 08:42:53 +01:00
|
|
|
err := <-errCh
|
|
|
|
|
2016-11-23 12:04:04 +01:00
|
|
|
close(info.completeCh)
|
2016-11-21 08:42:53 +01:00
|
|
|
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
// Called after complete-multipart-upload or abort-multipart-upload so that the appendParts go-routine is not left dangling.
|
2017-01-17 02:05:00 +01:00
|
|
|
func (fs fsObjects) abort(uploadID string) {
|
|
|
|
fs.bgAppend.Lock()
|
|
|
|
defer fs.bgAppend.Unlock()
|
|
|
|
|
|
|
|
info, ok := fs.bgAppend.infoMap[uploadID]
|
2016-11-21 08:42:53 +01:00
|
|
|
if !ok {
|
|
|
|
return
|
|
|
|
}
|
2017-01-17 02:05:00 +01:00
|
|
|
|
|
|
|
delete(fs.bgAppend.infoMap, uploadID)
|
|
|
|
|
2016-12-03 08:33:06 +01:00
|
|
|
info.abortCh <- struct{}{}
|
2016-11-21 08:42:53 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
// This is run as a go-routine that appends the parts in the background.
|
2017-01-17 02:05:00 +01:00
|
|
|
func (fs fsObjects) appendParts(bucket, object, uploadID string, info bgAppendPartsInfo) {
|
|
|
|
appendPath := pathJoin(fs.fsPath, minioMetaTmpBucket, fs.fsUUID, uploadID)
|
2016-11-21 08:42:53 +01:00
|
|
|
// Holds the list of parts that is already appended to the "append" file.
|
|
|
|
appendMeta := fsMetaV1{}
|
2017-01-17 02:05:00 +01:00
|
|
|
|
2016-12-09 05:35:07 +01:00
|
|
|
// Allocate staging read buffer.
|
|
|
|
buf := make([]byte, readSizeV1)
|
2016-11-21 08:42:53 +01:00
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case input := <-info.inputCh:
|
|
|
|
// We receive on this channel when new part gets uploaded or when complete-multipart sends
|
|
|
|
// a value on this channel to confirm if all the required parts are appended.
|
|
|
|
meta := input.meta
|
2017-01-17 02:05:00 +01:00
|
|
|
|
2016-11-21 08:42:53 +01:00
|
|
|
for {
|
|
|
|
// Append should be done such a way that if part-3 and part-2 is uploaded before part-1, we
|
|
|
|
// wait till part-1 is uploaded after which we append part-2 and part-3 as well in this for-loop.
|
|
|
|
part, appendNeeded := partToAppend(meta, appendMeta)
|
|
|
|
if !appendNeeded {
|
|
|
|
if reflect.DeepEqual(meta.Parts, appendMeta.Parts) {
|
|
|
|
// Sending nil is useful so that the complete-multipart-upload knows that
|
|
|
|
// all the required parts have been appended.
|
|
|
|
input.errCh <- nil
|
|
|
|
} else {
|
|
|
|
// Sending error is useful so that complete-multipart-upload can fall-back to
|
|
|
|
// its own append process.
|
|
|
|
input.errCh <- errPartsMissing
|
|
|
|
}
|
|
|
|
break
|
|
|
|
}
|
2017-01-17 02:05:00 +01:00
|
|
|
|
|
|
|
if err := fs.appendPart(bucket, object, uploadID, part, buf); err != nil {
|
|
|
|
fsRemoveFile(appendPath)
|
2016-11-23 21:50:09 +01:00
|
|
|
appendMeta.Parts = nil
|
2016-11-21 08:42:53 +01:00
|
|
|
input.errCh <- err
|
|
|
|
break
|
|
|
|
}
|
2017-01-17 02:05:00 +01:00
|
|
|
|
2016-11-21 08:42:53 +01:00
|
|
|
appendMeta.AddObjectPart(part.Number, part.Name, part.ETag, part.Size)
|
|
|
|
}
|
2016-11-23 12:04:04 +01:00
|
|
|
case <-info.abortCh:
|
|
|
|
// abort-multipart-upload closed abortCh to end the appendParts go-routine.
|
2017-01-17 02:05:00 +01:00
|
|
|
fsRemoveFile(appendPath)
|
|
|
|
|
|
|
|
// So that any racing PutObjectPart does not leave a dangling go-routine.
|
|
|
|
close(info.timeoutCh)
|
|
|
|
|
2016-11-23 12:04:04 +01:00
|
|
|
return
|
|
|
|
case <-info.completeCh:
|
|
|
|
// complete-multipart-upload closed completeCh to end the appendParts go-routine.
|
2016-12-10 01:10:18 +01:00
|
|
|
close(info.timeoutCh) // So that any racing PutObjectPart does not leave a dangling go-routine.
|
2016-11-21 08:42:53 +01:00
|
|
|
return
|
|
|
|
case <-time.After(appendPartsTimeout):
|
|
|
|
// Timeout the goroutine to garbage collect its resources. This would happen if the client initiates
|
|
|
|
// a multipart upload and does not complete/abort it.
|
2017-01-17 02:05:00 +01:00
|
|
|
fs.bgAppend.Lock()
|
|
|
|
delete(fs.bgAppend.infoMap, uploadID)
|
|
|
|
fs.bgAppend.Unlock()
|
|
|
|
|
2016-11-21 08:42:53 +01:00
|
|
|
// Delete the temporary append file as well.
|
2017-01-17 02:05:00 +01:00
|
|
|
fsRemoveFile(appendPath)
|
2016-11-21 08:42:53 +01:00
|
|
|
|
|
|
|
close(info.timeoutCh)
|
2016-11-23 21:50:09 +01:00
|
|
|
return
|
2016-11-21 08:42:53 +01:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Appends the "part" to the append-file inside "tmp/" that finally gets moved to the actual location
|
|
|
|
// upon complete-multipart-upload.
|
2017-01-17 02:05:00 +01:00
|
|
|
func (fs fsObjects) appendPart(bucket, object, uploadID string, part objectPartInfo, buf []byte) error {
|
|
|
|
partPath := pathJoin(fs.fsPath, minioMetaMultipartBucket, bucket, object, uploadID, part.Name)
|
2016-11-21 08:42:53 +01:00
|
|
|
|
2017-02-24 18:20:40 +01:00
|
|
|
var offset int64
|
2017-01-17 02:05:00 +01:00
|
|
|
// Read each file part to start writing to the temporary concatenated object.
|
|
|
|
file, size, err := fsOpenFile(partPath, offset)
|
|
|
|
if err != nil {
|
|
|
|
if err == errFileNotFound {
|
|
|
|
return errPartsMissing
|
2016-11-21 08:42:53 +01:00
|
|
|
}
|
2017-01-17 02:05:00 +01:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
defer file.Close()
|
|
|
|
|
|
|
|
tmpObjPath := pathJoin(fs.fsPath, minioMetaTmpBucket, fs.fsUUID, uploadID)
|
|
|
|
// No need to hold a lock, this is a unique file and will be only written
|
|
|
|
// to one one process per uploadID per minio process.
|
|
|
|
wfile, err := os.OpenFile(preparePath(tmpObjPath), os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0666)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
defer wfile.Close()
|
|
|
|
|
|
|
|
// Fallocate more space as we concatenate.
|
|
|
|
if err = fsFAllocate(int(wfile.Fd()), 0, size); err != nil {
|
|
|
|
return err
|
2016-11-21 08:42:53 +01:00
|
|
|
}
|
2017-01-17 02:05:00 +01:00
|
|
|
|
|
|
|
_, err = io.CopyBuffer(wfile, file, buf)
|
|
|
|
return err
|
2016-11-21 08:42:53 +01:00
|
|
|
}
|