aboutsummaryrefslogtreecommitdiff
path: root/weed/filer/reader_at.go
diff options
context:
space:
mode:
Diffstat (limited to 'weed/filer/reader_at.go')
-rw-r--r--weed/filer/reader_at.go138
1 files changed, 100 insertions, 38 deletions
diff --git a/weed/filer/reader_at.go b/weed/filer/reader_at.go
index 9f338782e..fa51df687 100644
--- a/weed/filer/reader_at.go
+++ b/weed/filer/reader_at.go
@@ -3,52 +3,72 @@ package filer
import (
"context"
"fmt"
- "io"
- "sync"
-
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/util/chunk_cache"
"github.com/chrislusf/seaweedfs/weed/wdclient"
+ "github.com/golang/groupcache/singleflight"
+ "io"
+ "math/rand"
+ "sync"
)
type ChunkReadAt struct {
masterClient *wdclient.MasterClient
chunkViews []*ChunkView
- lookupFileId func(fileId string) (targetUrl string, err error)
+ lookupFileId LookupFileIdFunctionType
readerLock sync.Mutex
fileSize int64
- 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 {
- return func(fileId string) (targetUrl string, err error) {
- err = filerClient.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
- vid := VolumeId(fileId)
- resp, err := client.LookupVolume(context.Background(), &filer_pb.LookupVolumeRequest{
- VolumeIds: []string{vid},
- })
- if err != nil {
- return err
- }
- locations := resp.LocationsMap[vid]
- if locations == nil || len(locations.Locations) == 0 {
- glog.V(0).Infof("failed to locate %s", fileId)
- return fmt.Errorf("failed to locate %s", fileId)
- }
+ vidCache := make(map[string]*filer_pb.Locations)
+ return func(fileId string) (targetUrls []string, err error) {
+ vid := VolumeId(fileId)
+ locations, found := vidCache[vid]
+
+ if !found {
+ // println("looking up volume", vid)
+ err = filerClient.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+ resp, err := client.LookupVolume(context.Background(), &filer_pb.LookupVolumeRequest{
+ VolumeIds: []string{vid},
+ })
+ if err != nil {
+ return err
+ }
+
+ locations = resp.LocationsMap[vid]
+ if locations == nil || len(locations.Locations) == 0 {
+ glog.V(0).Infof("failed to locate %s", fileId)
+ return fmt.Errorf("failed to locate %s", fileId)
+ }
+ vidCache[vid] = locations
+
+ return nil
+ })
+ }
- volumeServerAddress := filerClient.AdjustedUrl(locations.Locations[0].Url)
+ for _, loc := range locations.Locations {
+ volumeServerAddress := filerClient.AdjustedUrl(loc.Url)
+ targetUrl := fmt.Sprintf("http://%s/%s", volumeServerAddress, fileId)
+ targetUrls = append(targetUrls, targetUrl)
+ }
- targetUrl = fmt.Sprintf("http://%s/%s", volumeServerAddress, fileId)
+ for i := len(targetUrls) - 1; i > 0; i-- {
+ j := rand.Intn(i + 1)
+ targetUrls[i], targetUrls[j] = targetUrls[j], targetUrls[i]
+ }
- return nil
- })
return
}
}
@@ -76,10 +96,16 @@ func (c *ChunkReadAt) doReadAt(p []byte, offset int64) (n int, err error) {
var buffer []byte
startOffset, remaining := offset, int64(len(p))
+ var nextChunk *ChunkView
for i, chunk := range c.chunkViews {
if remaining <= 0 {
break
}
+ if i+1 < len(c.chunkViews) {
+ nextChunk = c.chunkViews[i+1]
+ } else {
+ nextChunk = nil
+ }
if startOffset < chunk.LogicOffset {
gap := int(chunk.LogicOffset - startOffset)
glog.V(4).Infof("zero [%d,%d)", startOffset, startOffset+int64(gap))
@@ -95,7 +121,7 @@ func (c *ChunkReadAt) doReadAt(p []byte, offset int64) (n int, err error) {
continue
}
glog.V(4).Infof("read [%d,%d), %d/%d chunk %s [%d,%d)", chunkStart, chunkStop, i, len(c.chunkViews), chunk.FileId, chunk.LogicOffset-chunk.Offset, chunk.LogicOffset-chunk.Offset+int64(chunk.Size))
- buffer, err = c.readFromWholeChunkData(chunk)
+ buffer, err = c.readFromWholeChunkData(chunk, nextChunk)
if err != nil {
glog.Errorf("fetching chunk %+v: %v\n", chunk, err)
return
@@ -123,27 +149,63 @@ func (c *ChunkReadAt) doReadAt(p []byte, offset int64) (n int, err error) {
}
-func (c *ChunkReadAt) readFromWholeChunkData(chunkView *ChunkView) (chunkData []byte, err error) {
+func (c *ChunkReadAt) readFromWholeChunkData(chunkView *ChunkView, nextChunkViews ...*ChunkView) (chunkData []byte, err error) {
- glog.V(4).Infof("readFromWholeChunkData %s offset %d [%d,%d) size at least %d", chunkView.FileId, chunkView.Offset, chunkView.LogicOffset, chunkView.LogicOffset+int64(chunkView.Size), chunkView.ChunkSize)
+ if c.lastChunkFileId == chunkView.FileId {
+ return c.lastChunkData, nil
+ }
- chunkData = c.chunkCache.GetChunk(chunkView.FileId, chunkView.ChunkSize)
- if chunkData != nil {
- glog.V(4).Infof("cache hit %s [%d,%d)", chunkView.FileId, chunkView.LogicOffset-chunkView.Offset, chunkView.LogicOffset-chunkView.Offset+int64(len(chunkData)))
- } else {
- glog.V(4).Infof("doFetchFullChunkData %s", chunkView.FileId)
- chunkData, err = c.doFetchFullChunkData(chunkView.FileId, chunkView.CipherKey, chunkView.IsGzipped)
- if err != nil {
- return
+ v, doErr := c.readOneWholeChunk(chunkView)
+
+ if doErr != nil {
+ return nil, doErr
+ }
+
+ chunkData = v.([]byte)
+
+ c.lastChunkData = chunkData
+ c.lastChunkFileId = chunkView.FileId
+
+ for _, nextChunkView := range nextChunkViews {
+ if c.chunkCache != nil && nextChunkView != nil {
+ go c.readOneWholeChunk(nextChunkView)
}
- c.chunkCache.SetChunk(chunkView.FileId, chunkData)
}
return
}
-func (c *ChunkReadAt) doFetchFullChunkData(fileId string, cipherKey []byte, isGzipped bool) ([]byte, error) {
+func (c *ChunkReadAt) readOneWholeChunk(chunkView *ChunkView) (interface{}, error) {
+
+ var err error
+
+ return c.fetchGroup.Do(chunkView.FileId, func() (interface{}, error) {
+
+ glog.V(4).Infof("readFromWholeChunkData %s offset %d [%d,%d) size at least %d", chunkView.FileId, chunkView.Offset, chunkView.LogicOffset, chunkView.LogicOffset+int64(chunkView.Size), chunkView.ChunkSize)
+
+ data := c.chunkCache.GetChunk(chunkView.FileId, chunkView.ChunkSize)
+ if data != nil {
+ glog.V(4).Infof("cache hit %s [%d,%d)", chunkView.FileId, chunkView.LogicOffset-chunkView.Offset, chunkView.LogicOffset-chunkView.Offset+int64(len(data)))
+ } else {
+ var err error
+ data, err = c.doFetchFullChunkData(chunkView)
+ if err != nil {
+ return data, err
+ }
+ c.chunkCache.SetChunk(chunkView.FileId, data)
+ }
+ return data, err
+ })
+}
+
+func (c *ChunkReadAt) doFetchFullChunkData(chunkView *ChunkView) ([]byte, error) {
+
+ glog.V(4).Infof("+ doFetchFullChunkData %s", chunkView.FileId)
+
+ data, err := fetchChunk(c.lookupFileId, chunkView.FileId, chunkView.CipherKey, chunkView.IsGzipped)
+
+ glog.V(4).Infof("- doFetchFullChunkData %s", chunkView.FileId)
- return fetchChunk(c.lookupFileId, fileId, cipherKey, isGzipped)
+ return data, err
}
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-24 20:06:27 +0000