seaweedfs/weed/filer/filechunk_manifest.go

273 lines
7.7 KiB
Go
Raw Normal View History

2020-09-01 07:21:19 +00:00
package filer
import (
"bytes"
"fmt"
"io"
"math"
2021-09-08 02:29:42 +00:00
"net/url"
"strings"
2022-03-02 21:50:46 +00:00
"sync"
"time"
"github.com/seaweedfs/seaweedfs/weed/wdclient"
2022-08-17 19:05:07 +00:00
"google.golang.org/protobuf/proto"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/util"
)
2020-07-20 10:34:06 +00:00
const (
2022-03-07 08:07:53 +00:00
ManifestBatch = 10000
2020-07-20 10:34:06 +00:00
)
2022-03-02 21:50:46 +00:00
var bytesBufferPool = sync.Pool{
New: func() interface{} {
return new(bytes.Buffer)
},
}
func HasChunkManifest(chunks []*filer_pb.FileChunk) bool {
for _, chunk := range chunks {
if chunk.IsChunkManifest {
return true
}
}
return false
}
2020-08-24 01:30:11 +00:00
func SeparateManifestChunks(chunks []*filer_pb.FileChunk) (manifestChunks, nonManifestChunks []*filer_pb.FileChunk) {
for _, c := range chunks {
if c.IsChunkManifest {
2020-08-24 01:30:11 +00:00
manifestChunks = append(manifestChunks, c)
} else {
nonManifestChunks = append(nonManifestChunks, c)
}
}
return
}
func ResolveChunkManifest(lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk, startOffset, stopOffset int64) (dataChunks, manifestChunks []*filer_pb.FileChunk, manifestResolveErr error) {
// TODO maybe parallel this
for _, chunk := range chunks {
if max(chunk.Offset, startOffset) >= min(chunk.Offset+int64(chunk.Size), stopOffset) {
continue
}
if !chunk.IsChunkManifest {
dataChunks = append(dataChunks, chunk)
continue
}
resolvedChunks, err := ResolveOneChunkManifest(lookupFileIdFn, chunk)
if err != nil {
return dataChunks, nil, err
}
manifestChunks = append(manifestChunks, chunk)
// recursive
subDataChunks, subManifestChunks, subErr := ResolveChunkManifest(lookupFileIdFn, resolvedChunks, startOffset, stopOffset)
if subErr != nil {
return dataChunks, nil, subErr
}
dataChunks = append(dataChunks, subDataChunks...)
manifestChunks = append(manifestChunks, subManifestChunks...)
}
return
}
func ResolveOneChunkManifest(lookupFileIdFn wdclient.LookupFileIdFunctionType, chunk *filer_pb.FileChunk) (dataChunks []*filer_pb.FileChunk, manifestResolveErr error) {
if !chunk.IsChunkManifest {
return
}
// IsChunkManifest
2022-03-02 21:50:46 +00:00
bytesBuffer := bytesBufferPool.Get().(*bytes.Buffer)
2022-03-06 13:06:04 +00:00
bytesBuffer.Reset()
2022-03-02 21:50:46 +00:00
defer bytesBufferPool.Put(bytesBuffer)
err := fetchWholeChunk(bytesBuffer, lookupFileIdFn, chunk.GetFileIdString(), chunk.CipherKey, chunk.IsCompressed)
if err != nil {
return nil, fmt.Errorf("fail to read manifest %s: %v", chunk.GetFileIdString(), err)
}
m := &filer_pb.FileChunkManifest{}
2022-03-02 21:50:46 +00:00
if err := proto.Unmarshal(bytesBuffer.Bytes(), m); err != nil {
return nil, fmt.Errorf("fail to unmarshal manifest %s: %v", chunk.GetFileIdString(), err)
}
// recursive
filer_pb.AfterEntryDeserialization(m.Chunks)
return m.Chunks, nil
}
2020-07-20 10:34:06 +00:00
// TODO fetch from cache for weed mount?
2022-03-02 21:50:46 +00:00
func fetchWholeChunk(bytesBuffer *bytes.Buffer, lookupFileIdFn wdclient.LookupFileIdFunctionType, fileId string, cipherKey []byte, isGzipped bool) error {
urlStrings, err := lookupFileIdFn(fileId)
if err != nil {
2020-07-20 10:34:06 +00:00
glog.Errorf("operation LookupFileId %s failed, err: %v", fileId, err)
2022-03-02 21:50:46 +00:00
return err
}
err = retriedStreamFetchChunkData(bytesBuffer, urlStrings, cipherKey, isGzipped, true, 0, 0)
if err != nil {
return err
}
2022-03-02 21:50:46 +00:00
return nil
2020-10-09 06:19:42 +00:00
}
func fetchChunkRange(buffer []byte, lookupFileIdFn wdclient.LookupFileIdFunctionType, fileId string, cipherKey []byte, isGzipped bool, offset int64) (int, error) {
urlStrings, err := lookupFileIdFn(fileId)
if err != nil {
glog.Errorf("operation LookupFileId %s failed, err: %v", fileId, err)
return 0, err
}
return retriedFetchChunkData(buffer, urlStrings, cipherKey, isGzipped, false, offset)
}
2022-02-26 10:16:47 +00:00
func retriedFetchChunkData(buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64) (n int, err error) {
2020-10-09 06:19:42 +00:00
2020-10-13 07:29:46 +00:00
var shouldRetry bool
2020-11-01 10:36:43 +00:00
for waitTime := time.Second; waitTime < util.RetryWaitTime; waitTime += waitTime / 2 {
for _, urlString := range urlStrings {
2022-02-26 10:16:47 +00:00
n = 0
2021-09-08 02:29:42 +00:00
if strings.Contains(urlString, "%") {
urlString = url.PathEscape(urlString)
}
2022-02-26 10:16:47 +00:00
shouldRetry, err = util.ReadUrlAsStream(urlString+"?readDeleted=true", cipherKey, isGzipped, isFullChunk, offset, len(buffer), func(data []byte) {
2022-02-26 11:23:06 +00:00
if n < len(buffer) {
x := copy(buffer[n:], data)
n += x
}
})
2020-10-13 07:29:46 +00:00
if !shouldRetry {
break
}
if err != nil {
glog.V(0).Infof("read %s failed, err: %v", urlString, err)
} else {
break
}
}
2020-10-14 02:50:46 +00:00
if err != nil && shouldRetry {
2020-10-14 02:50:22 +00:00
glog.V(0).Infof("retry reading in %v", waitTime)
2020-10-09 07:01:47 +00:00
time.Sleep(waitTime)
} else {
break
}
}
2022-02-26 10:16:47 +00:00
return n, err
}
func retriedStreamFetchChunkData(writer io.Writer, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, size int) (err error) {
var shouldRetry bool
var totalWritten int
for waitTime := time.Second; waitTime < util.RetryWaitTime; waitTime += waitTime / 2 {
for _, urlString := range urlStrings {
var localProcessed int
shouldRetry, err = util.ReadUrlAsStream(urlString+"?readDeleted=true", cipherKey, isGzipped, isFullChunk, offset, size, func(data []byte) {
if totalWritten > localProcessed {
toBeSkipped := totalWritten - localProcessed
if len(data) <= toBeSkipped {
localProcessed += len(data)
return // skip if already processed
}
2021-08-13 18:31:43 +00:00
data = data[toBeSkipped:]
localProcessed += toBeSkipped
}
writer.Write(data)
localProcessed += len(data)
totalWritten += len(data)
})
if !shouldRetry {
break
}
if err != nil {
glog.V(0).Infof("read %s failed, err: %v", urlString, err)
} else {
break
}
}
2021-08-13 18:13:30 +00:00
if err != nil && shouldRetry {
glog.V(0).Infof("retry reading in %v", waitTime)
time.Sleep(waitTime)
} else {
break
}
}
return err
}
2020-07-20 10:34:06 +00:00
func MaybeManifestize(saveFunc SaveDataAsChunkFunctionType, inputChunks []*filer_pb.FileChunk) (chunks []*filer_pb.FileChunk, err error) {
return doMaybeManifestize(saveFunc, inputChunks, ManifestBatch, mergeIntoManifest)
}
func doMaybeManifestize(saveFunc SaveDataAsChunkFunctionType, inputChunks []*filer_pb.FileChunk, mergeFactor int, mergefn func(saveFunc SaveDataAsChunkFunctionType, dataChunks []*filer_pb.FileChunk) (manifestChunk *filer_pb.FileChunk, err error)) (chunks []*filer_pb.FileChunk, err error) {
var dataChunks []*filer_pb.FileChunk
for _, chunk := range inputChunks {
if !chunk.IsChunkManifest {
dataChunks = append(dataChunks, chunk)
} else {
chunks = append(chunks, chunk)
}
}
remaining := len(dataChunks)
2020-07-20 10:34:06 +00:00
for i := 0; i+mergeFactor <= len(dataChunks); i += mergeFactor {
chunk, err := mergefn(saveFunc, dataChunks[i:i+mergeFactor])
if err != nil {
return dataChunks, err
}
chunks = append(chunks, chunk)
2020-07-20 10:34:06 +00:00
remaining -= mergeFactor
}
// remaining
for i := len(dataChunks) - remaining; i < len(dataChunks); i++ {
chunks = append(chunks, dataChunks[i])
}
return
}
func mergeIntoManifest(saveFunc SaveDataAsChunkFunctionType, dataChunks []*filer_pb.FileChunk) (manifestChunk *filer_pb.FileChunk, err error) {
2020-07-21 05:01:39 +00:00
filer_pb.BeforeEntrySerialization(dataChunks)
// create and serialize the manifest
data, serErr := proto.Marshal(&filer_pb.FileChunkManifest{
Chunks: dataChunks,
})
if serErr != nil {
return nil, fmt.Errorf("serializing manifest: %v", serErr)
}
minOffset, maxOffset := int64(math.MaxInt64), int64(math.MinInt64)
2020-07-20 10:34:06 +00:00
for _, chunk := range dataChunks {
if minOffset > int64(chunk.Offset) {
minOffset = chunk.Offset
}
if maxOffset < int64(chunk.Size)+chunk.Offset {
maxOffset = int64(chunk.Size) + chunk.Offset
}
}
manifestChunk, err = saveFunc(bytes.NewReader(data), "", 0)
if err != nil {
return nil, err
}
manifestChunk.IsChunkManifest = true
manifestChunk.Offset = minOffset
manifestChunk.Size = uint64(maxOffset - minOffset)
return
}
type SaveDataAsChunkFunctionType func(reader io.Reader, name string, offset int64) (chunk *filer_pb.FileChunk, err error)