aboutsummaryrefslogtreecommitdiff
path: root/weed/filer
diff options
context:
space:
mode:
Diffstat (limited to 'weed/filer')
-rw-r--r--weed/filer/filechunk_manifest.go21
-rw-r--r--weed/filer/reader_at.go24
-rw-r--r--weed/filer/stream.go64
3 files changed, 66 insertions, 43 deletions
diff --git a/weed/filer/filechunk_manifest.go b/weed/filer/filechunk_manifest.go
index 37b172357..6fcd7abc2 100644
--- a/weed/filer/filechunk_manifest.go
+++ b/weed/filer/filechunk_manifest.go
@@ -84,21 +84,26 @@ func ResolveOneChunkManifest(lookupFileIdFn LookupFileIdFunctionType, chunk *fil
// TODO fetch from cache for weed mount?
func fetchChunk(lookupFileIdFn LookupFileIdFunctionType, fileId string, cipherKey []byte, isGzipped bool) ([]byte, error) {
- urlString, err := lookupFileIdFn(fileId)
+ urlStrings, err := lookupFileIdFn(fileId)
if err != nil {
glog.Errorf("operation LookupFileId %s failed, err: %v", fileId, err)
return nil, err
}
var buffer bytes.Buffer
- err = util.ReadUrlAsStream(urlString, cipherKey, isGzipped, true, 0, 0, func(data []byte) {
- buffer.Write(data)
- })
- if err != nil {
- glog.V(0).Infof("read %s failed, err: %v", fileId, err)
- return nil, err
+
+ for _, urlString := range urlStrings {
+ err = util.ReadUrlAsStream(urlString, cipherKey, isGzipped, true, 0, 0, func(data []byte) {
+ buffer.Write(data)
+ })
+ if err != nil {
+ glog.V(0).Infof("read %s failed, err: %v", fileId, err)
+ buffer.Reset()
+ } else {
+ break
+ }
}
- return buffer.Bytes(), nil
+ return buffer.Bytes(), err
}
func MaybeManifestize(saveFunc SaveDataAsChunkFunctionType, inputChunks []*filer_pb.FileChunk) (chunks []*filer_pb.FileChunk, err error) {
diff --git a/weed/filer/reader_at.go b/weed/filer/reader_at.go
index 3aeb9a94d..74f037557 100644
--- a/weed/filer/reader_at.go
+++ b/weed/filer/reader_at.go
@@ -5,7 +5,6 @@ import (
"fmt"
"github.com/golang/groupcache/singleflight"
"io"
- "math/rand"
"sync"
"github.com/chrislusf/seaweedfs/weed/glog"
@@ -17,24 +16,24 @@ import (
type ChunkReadAt struct {
masterClient *wdclient.MasterClient
chunkViews []*ChunkView
- lookupFileId func(fileId string) (targetUrl string, err error)
+ lookupFileId LookupFileIdFunctionType
readerLock sync.Mutex
fileSize int64
- fetchGroup singleflight.Group
- lastChunkFileId string
- lastChunkData []byte
- chunkCache chunk_cache.ChunkCache
+ fetchGroup singleflight.Group
+ lastChunkFileId string
+ lastChunkData []byte
+ chunkCache chunk_cache.ChunkCache
}
// var _ = io.ReaderAt(&ChunkReadAt{})
-type LookupFileIdFunctionType func(fileId string) (targetUrl string, err error)
+type LookupFileIdFunctionType func(fileId string) (targetUrls []string, err error)
func LookupFn(filerClient filer_pb.FilerClient) LookupFileIdFunctionType {
vidCache := make(map[string]*filer_pb.Locations)
- return func(fileId string) (targetUrl string, err error) {
+ return func(fileId string) (targetUrls []string, err error) {
vid := VolumeId(fileId)
locations, found := vidCache[vid]
@@ -59,8 +58,11 @@ func LookupFn(filerClient filer_pb.FilerClient) LookupFileIdFunctionType {
})
}
- volumeServerAddress := filerClient.AdjustedUrl(locations.Locations[rand.Intn(len(locations.Locations))].Url)
- targetUrl = fmt.Sprintf("http://%s/%s", volumeServerAddress, fileId)
+ for _, loc := range locations.Locations {
+ volumeServerAddress := filerClient.AdjustedUrl(loc.Url)
+ targetUrl := fmt.Sprintf("http://%s/%s", volumeServerAddress, fileId)
+ targetUrls = append(targetUrls, targetUrl)
+ }
return
}
@@ -142,7 +144,7 @@ func (c *ChunkReadAt) doReadAt(p []byte, offset int64) (n int, err error) {
}
-func (c *ChunkReadAt) readFromWholeChunkData(chunkView *ChunkView, nextChunkViews... *ChunkView) (chunkData []byte, err error) {
+func (c *ChunkReadAt) readFromWholeChunkData(chunkView *ChunkView, nextChunkViews ...*ChunkView) (chunkData []byte, err error) {
if c.lastChunkFileId == chunkView.FileId {
return c.lastChunkData, nil
diff --git a/weed/filer/stream.go b/weed/filer/stream.go
index dc6e414ca..d98229379 100644
--- a/weed/filer/stream.go
+++ b/weed/filer/stream.go
@@ -17,27 +17,32 @@ func StreamContent(masterClient *wdclient.MasterClient, w io.Writer, chunks []*f
// fmt.Printf("start to stream content for chunks: %+v\n", chunks)
chunkViews := ViewFromChunks(masterClient.LookupFileId, chunks, offset, size)
- fileId2Url := make(map[string]string)
+ fileId2Url := make(map[string][]string)
for _, chunkView := range chunkViews {
- urlString, err := masterClient.LookupFileId(chunkView.FileId)
+ urlStrings, err := masterClient.LookupFileId(chunkView.FileId)
if err != nil {
glog.V(1).Infof("operation LookupFileId %s failed, err: %v", chunkView.FileId, err)
return err
}
- fileId2Url[chunkView.FileId] = urlString
+ fileId2Url[chunkView.FileId] = urlStrings
}
for _, chunkView := range chunkViews {
- urlString := fileId2Url[chunkView.FileId]
- err := util.ReadUrlAsStream(urlString, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.Offset, int(chunkView.Size), func(data []byte) {
- w.Write(data)
- })
- if err != nil {
- glog.V(1).Infof("read %s failed, err: %v", chunkView.FileId, err)
- return err
+ urlStrings := fileId2Url[chunkView.FileId]
+ for _, urlString := range urlStrings {
+ err := util.ReadUrlAsStream(urlString, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.Offset, int(chunkView.Size), func(data []byte) {
+ w.Write(data)
+ })
+ if err != nil {
+ // data already written to w would be wrong
+ // but usually there are nothing written if fails to read
+ glog.V(1).Infof("read %s failed, err: %v", chunkView.FileId, err)
+ } else {
+ break
+ }
}
}
@@ -51,24 +56,28 @@ func ReadAll(masterClient *wdclient.MasterClient, chunks []*filer_pb.FileChunk)
buffer := bytes.Buffer{}
- lookupFileIdFn := func(fileId string) (targetUrl string, err error) {
+ lookupFileIdFn := func(fileId string) (targetUrls []string, err error) {
return masterClient.LookupFileId(fileId)
}
chunkViews := ViewFromChunks(lookupFileIdFn, chunks, 0, math.MaxInt64)
for _, chunkView := range chunkViews {
- urlString, err := lookupFileIdFn(chunkView.FileId)
+ urlStrings, err := lookupFileIdFn(chunkView.FileId)
if err != nil {
glog.V(1).Infof("operation LookupFileId %s failed, err: %v", chunkView.FileId, err)
return nil, err
}
- err = util.ReadUrlAsStream(urlString, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.Offset, int(chunkView.Size), func(data []byte) {
- buffer.Write(data)
- })
- if err != nil {
- glog.V(1).Infof("read %s failed, err: %v", chunkView.FileId, err)
- return nil, err
+ for _, urlString := range urlStrings {
+ err = util.ReadUrlAsStream(urlString, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.Offset, int(chunkView.Size), func(data []byte) {
+ buffer.Write(data)
+ })
+ if err != nil {
+ glog.V(1).Infof("read %s failed, err: %v", chunkView.FileId, err)
+ buffer.Reset()
+ } else {
+ break
+ }
}
}
return buffer.Bytes(), nil
@@ -89,7 +98,7 @@ var _ = io.ReadSeeker(&ChunkStreamReader{})
func NewChunkStreamReaderFromFiler(masterClient *wdclient.MasterClient, chunks []*filer_pb.FileChunk) *ChunkStreamReader {
- lookupFileIdFn := func(fileId string) (targetUrl string, err error) {
+ lookupFileIdFn := func(fileId string) (targetUrl []string, err error) {
return masterClient.LookupFileId(fileId)
}
@@ -169,17 +178,24 @@ func (c *ChunkStreamReader) Seek(offset int64, whence int) (int64, error) {
}
func (c *ChunkStreamReader) fetchChunkToBuffer(chunkView *ChunkView) error {
- urlString, err := c.lookupFileId(chunkView.FileId)
+ urlStrings, err := c.lookupFileId(chunkView.FileId)
if err != nil {
glog.V(1).Infof("operation LookupFileId %s failed, err: %v", chunkView.FileId, err)
return err
}
var buffer bytes.Buffer
- err = util.ReadUrlAsStream(urlString, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.Offset, int(chunkView.Size), func(data []byte) {
- buffer.Write(data)
- })
+ for _, urlString := range urlStrings {
+ err = util.ReadUrlAsStream(urlString, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.Offset, int(chunkView.Size), func(data []byte) {
+ buffer.Write(data)
+ })
+ if err != nil {
+ glog.V(1).Infof("read %s failed, err: %v", chunkView.FileId, err)
+ buffer.Reset()
+ } else {
+ break
+ }
+ }
if err != nil {
- glog.V(1).Infof("read %s failed, err: %v", chunkView.FileId, err)
return err
}
c.buffer = buffer.Bytes()
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-24 14:41:02 +0000