mount: improve read throughput with parallel chunk fetching (#7569)
* mount: improve read throughput with parallel chunk fetching This addresses issue #7504 where a single weed mount FUSE instance does not fully utilize node network bandwidth when reading large files. Changes: - Add -concurrentReaders mount option (default: 16) to control the maximum number of parallel chunk fetches during read operations - Implement parallel section reading in ChunkGroup.ReadDataAt() using errgroup for better throughput when reading across multiple sections - Enhance ReaderCache with MaybeCacheMany() to prefetch multiple chunks ahead in parallel during sequential reads (now prefetches 4 chunks) - Increase ReaderCache limit dynamically based on concurrentReaders to support higher read parallelism The bottleneck was that chunks were being read sequentially even when they reside on different volume servers. By introducing parallel chunk fetching, a single mount instance can now better saturate available network bandwidth. Fixes: #7504 * fmt * Address review comments: make prefetch configurable, improve error handling Changes: 1. Add DefaultPrefetchCount constant (4) to reader_at.go 2. Add GetPrefetchCount() method to ChunkGroup that derives prefetch count from concurrentReaders (1/4 ratio, min 1, max 8) 3. Pass prefetch count through NewChunkReaderAtFromClient 4. Fix error handling in readDataAtParallel to prioritize errgroup error 5. Update all callers to use DefaultPrefetchCount constant For mount operations, prefetch scales with -concurrentReaders: - concurrentReaders=16 (default) -> prefetch=4 - concurrentReaders=32 -> prefetch=8 (capped) - concurrentReaders=4 -> prefetch=1 For non-mount paths (WebDAV, query engine, MQ), uses DefaultPrefetchCount. * fmt * Refactor: use variadic parameter instead of new function name Use NewChunkGroup with optional concurrentReaders parameter instead of creating a separate NewChunkGroupWithConcurrency function. This maintains backward compatibility - existing callers without the parameter get the default of 16 concurrent readers. * Use explicit concurrentReaders parameter instead of variadic * Refactor: use MaybeCache with count parameter instead of new MaybeCacheMany function * Address nitpick review comments - Add upper bound (128) on concurrentReaders to prevent excessive goroutine fan-out - Cap readerCacheLimit at 256 accordingly - Fix SetChunks: use Lock() instead of RLock() since we are writing to group.sections
This commit is contained in:
@@ -13,6 +13,12 @@ import (
|
||||
"github.com/seaweedfs/seaweedfs/weed/wdclient"
|
||||
)
|
||||
|
||||
// DefaultPrefetchCount is the default number of chunks to prefetch ahead during
|
||||
// sequential reads. This value is used when prefetch count is not explicitly
|
||||
// configured (e.g., WebDAV, query engine, message queue). For mount operations,
|
||||
// the prefetch count is derived from the -concurrentReaders option.
|
||||
const DefaultPrefetchCount = 4
|
||||
|
||||
type ChunkReadAt struct {
|
||||
masterClient *wdclient.MasterClient
|
||||
chunkViews *IntervalList[*ChunkView]
|
||||
@@ -20,6 +26,7 @@ type ChunkReadAt struct {
|
||||
readerCache *ReaderCache
|
||||
readerPattern *ReaderPattern
|
||||
lastChunkFid string
|
||||
prefetchCount int // Number of chunks to prefetch ahead during sequential reads
|
||||
ctx context.Context // Context used for cancellation during chunk read operations
|
||||
}
|
||||
|
||||
@@ -35,8 +42,9 @@ var _ = io.Closer(&ChunkReadAt{})
|
||||
// - No high availability (single filer address, no automatic failover)
|
||||
//
|
||||
// For NEW code, especially mount operations, use wdclient.FilerClient instead:
|
||||
// filerClient := wdclient.NewFilerClient(filerAddresses, grpcDialOption, dataCenter, opts)
|
||||
// lookupFn := filerClient.GetLookupFileIdFunction()
|
||||
//
|
||||
// filerClient := wdclient.NewFilerClient(filerAddresses, grpcDialOption, dataCenter, opts)
|
||||
// lookupFn := filerClient.GetLookupFileIdFunction()
|
||||
//
|
||||
// This provides:
|
||||
// - Bounded cache with configurable size
|
||||
@@ -56,7 +64,7 @@ func LookupFn(filerClient filer_pb.FilerClient) wdclient.LookupFileIdFunctionTyp
|
||||
var vidCacheLock sync.RWMutex
|
||||
cacheSize := 0
|
||||
const maxCacheSize = 10000 // Simple bound to prevent unbounded growth
|
||||
|
||||
|
||||
return func(ctx context.Context, fileId string) (targetUrls []string, err error) {
|
||||
vid := VolumeId(fileId)
|
||||
vidCacheLock.RLock()
|
||||
@@ -123,13 +131,14 @@ func LookupFn(filerClient filer_pb.FilerClient) wdclient.LookupFileIdFunctionTyp
|
||||
}
|
||||
}
|
||||
|
||||
func NewChunkReaderAtFromClient(ctx context.Context, readerCache *ReaderCache, chunkViews *IntervalList[*ChunkView], fileSize int64) *ChunkReadAt {
|
||||
func NewChunkReaderAtFromClient(ctx context.Context, readerCache *ReaderCache, chunkViews *IntervalList[*ChunkView], fileSize int64, prefetchCount int) *ChunkReadAt {
|
||||
|
||||
return &ChunkReadAt{
|
||||
chunkViews: chunkViews,
|
||||
fileSize: fileSize,
|
||||
readerCache: readerCache,
|
||||
readerPattern: NewReaderPattern(),
|
||||
prefetchCount: prefetchCount,
|
||||
ctx: ctx,
|
||||
}
|
||||
}
|
||||
@@ -246,8 +255,10 @@ func (c *ChunkReadAt) readChunkSliceAt(ctx context.Context, buffer []byte, chunk
|
||||
if c.lastChunkFid != "" {
|
||||
c.readerCache.UnCache(c.lastChunkFid)
|
||||
}
|
||||
if nextChunkViews != nil {
|
||||
c.readerCache.MaybeCache(nextChunkViews) // just read the next chunk if at the very beginning
|
||||
if nextChunkViews != nil && c.prefetchCount > 0 {
|
||||
// Prefetch multiple chunks ahead for better sequential read throughput
|
||||
// This keeps the network pipeline full with parallel chunk fetches
|
||||
c.readerCache.MaybeCache(nextChunkViews, c.prefetchCount)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user