gsh-digital-services/seaweedFS
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactoring

Browse Source
This commit is contained in:
Chris Lu
2020-07-20 03:34:06 -07:00
parent 1d724ab237
commit d02c0fe0c0
1 changed files with 12 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
weed/filer2/filechunk_manifest.go
View File

@@ -13,6 +13,10 @@ import (
"github.com/chrislusf/seaweedfs/weed/util" "github.com/chrislusf/seaweedfs/weed/util"
) )
const (
ManifestBatch = 10000
)
func HasChunkManifest(chunks []*filer_pb.FileChunk) bool { func HasChunkManifest(chunks []*filer_pb.FileChunk) bool {
for _, chunk := range chunks { for _, chunk := range chunks {
if chunk.IsChunkManifest { if chunk.IsChunkManifest {
@@ -51,10 +55,11 @@ func ResolveChunkManifest(lookupFileIdFn LookupFileIdFunctionType, chunks []*fil
return return
} }
// TODO fetch from cache for weed mount?
func fetchChunk(lookupFileIdFn LookupFileIdFunctionType, fileId string, cipherKey []byte, isGzipped bool) ([]byte, error) { func fetchChunk(lookupFileIdFn LookupFileIdFunctionType, fileId string, cipherKey []byte, isGzipped bool) ([]byte, error) {
urlString, err := lookupFileIdFn(fileId) urlString, err := lookupFileIdFn(fileId)
if err != nil { if err != nil {
glog.V(1).Infof("operation LookupFileId %s failed, err: %v", fileId, err) glog.Errorf("operation LookupFileId %s failed, err: %v", fileId, err)
return nil, err return nil, err
} }
var buffer bytes.Buffer var buffer bytes.Buffer
@@ -69,8 +74,8 @@ func fetchChunk(lookupFileIdFn LookupFileIdFunctionType, fileId string, cipherKe
return buffer.Bytes(), nil return buffer.Bytes(), nil
} }
func MaybeManifestize(saveFunc SaveDataAsChunkFunctionType, dataChunks []*filer_pb.FileChunk) (chunks []*filer_pb.FileChunk, err error) { func MaybeManifestize(saveFunc SaveDataAsChunkFunctionType, inputChunks []*filer_pb.FileChunk) (chunks []*filer_pb.FileChunk, err error) {
return doMaybeManifestize(saveFunc, dataChunks, 10000, mergeIntoManifest) 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) { 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) {
@@ -84,15 +89,14 @@ func doMaybeManifestize(saveFunc SaveDataAsChunkFunctionType, inputChunks []*fil
} }
} }
manifestBatch := mergeFactor
remaining := len(dataChunks) remaining := len(dataChunks)
for i := 0; i+manifestBatch <= len(dataChunks); i += manifestBatch { for i := 0; i+mergeFactor <= len(dataChunks); i += mergeFactor {
chunk, err := mergefn(saveFunc, dataChunks[i:i+manifestBatch]) chunk, err := mergefn(saveFunc, dataChunks[i:i+mergeFactor])
if err != nil { if err != nil {
return dataChunks, err return dataChunks, err
} }
chunks = append(chunks, chunk) chunks = append(chunks, chunk)
remaining -= manifestBatch remaining -= mergeFactor
} }
// remaining // remaining
for i := len(dataChunks) - remaining; i < len(dataChunks); i++ { for i := len(dataChunks) - remaining; i < len(dataChunks); i++ {
@@ -112,8 +116,7 @@ func mergeIntoManifest(saveFunc SaveDataAsChunkFunctionType, dataChunks []*filer
} }
minOffset, maxOffset := int64(math.MaxInt64), int64(math.MinInt64) minOffset, maxOffset := int64(math.MaxInt64), int64(math.MinInt64)
for k := 0; k < len(dataChunks); k++ { for _, chunk := range dataChunks {
chunk := dataChunks[k]
if minOffset > int64(chunk.Offset) { if minOffset > int64(chunk.Offset) {
minOffset = chunk.Offset minOffset = chunk.Offset
} }