diff --git a/weed/filer/filer_notify.go b/weed/filer/filer_notify.go index 311874875..c66235afc 100644 --- a/weed/filer/filer_notify.go +++ b/weed/filer/filer_notify.go @@ -200,7 +200,27 @@ func isChunkNotFoundError(err error) bool { httpNotFoundPattern.MatchString(errMsg) } -func (f *Filer) ReadPersistedLogBuffer(startPosition log_buffer.MessagePosition, stopTsNs int64, eachLogEntryFn log_buffer.EachLogEntryFuncType) (lastTsNs int64, isDone bool, err error) { +// persistedLogReplayLimit caps concurrent legacy replays; each holds a chunk +// reader per source filer, so a reconnect storm of pre-offload clients would +// otherwise pin many GB. Metadata-chunks clients take sendLogFileRefs and never +// reach this path. +const persistedLogReplayLimit = 16 + +var persistedLogReplaySem = make(chan struct{}, persistedLogReplayLimit) + +func (f *Filer) ReadPersistedLogBuffer(ctx context.Context, startPosition log_buffer.MessagePosition, stopTsNs int64, eachLogEntryFn log_buffer.EachLogEntryFuncType) (lastTsNs int64, isDone bool, err error) { + + // Cap concurrent replays; bail if the stream is already gone so cancelled + // clients do not park on the semaphore. + if err := ctx.Err(); err != nil { + return 0, false, err + } + select { + case persistedLogReplaySem <- struct{}{}: + defer func() { <-persistedLogReplaySem }() + case <-ctx.Done(): + return 0, false, ctx.Err() + } visitor, visitErr := f.collectPersistedLogBuffer(startPosition, stopTsNs) if visitErr != nil { diff --git a/weed/filer/stream.go b/weed/filer/stream.go index de53134d1..4acf93862 100644 --- a/weed/filer/stream.go +++ b/weed/filer/stream.go @@ -484,6 +484,8 @@ func (c *ChunkStreamReader) fetchChunkToBuffer(chunkView *ChunkView) error { return err } var buffer bytes.Buffer + // pre-size to the known chunk size; avoids bytes.Buffer's doubling regrowth + buffer.Grow(int(chunkView.ViewSize)) var shouldRetry bool jwt := JwtForVolumeServer(chunkView.FileId) for _, urlString := range urlStrings { diff --git a/weed/mount/meta_cache/meta_cache_subscribe.go b/weed/mount/meta_cache/meta_cache_subscribe.go index d2f472468..18b35446c 100644 --- a/weed/mount/meta_cache/meta_cache_subscribe.go +++ b/weed/mount/meta_cache/meta_cache_subscribe.go @@ -2,8 +2,10 @@ package meta_cache import ( "context" + "io" "strings" + "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -66,6 +68,9 @@ func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.Fil prefix = prefix + "/" } + // Read persisted log chunks directly from volume servers, keeping the replay + // cost off the filer's heap (see LogFileReaderFn below). + lookupFn := filer.LookupFn(client) metadataFollowOption := &pb.MetadataFollowOption{ ClientName: "mount", ClientId: selfSignature, @@ -77,6 +82,9 @@ func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.Fil StartTsNs: lastTsNs, StopTsNs: 0, EventErrorType: pb.FatalOnError, + LogFileReaderFn: func(chunks []*filer_pb.FileChunk) (io.ReadCloser, error) { + return filer.NewChunkStreamReaderFromLookup(context.Background(), lookupFn, chunks), nil + }, } util.RetryUntil("followMetaUpdates", func() error { metadataFollowOption.ClientEpoch++ diff --git a/weed/pb/filer_pb_direct_read.go b/weed/pb/filer_pb_direct_read.go index 294b78c7a..2a1a78c32 100644 --- a/weed/pb/filer_pb_direct_read.go +++ b/weed/pb/filer_pb_direct_read.go @@ -144,6 +144,23 @@ func readMultiFilersMerged( streams := make([]filerStream, len(filerOrder)) var wg sync.WaitGroup + // A genuine (non chunk-not-found) read error must fail the whole replay: the + // caller advances its cursor only on success, so a swallowed error leaves a + // permanent gap. stop aborts the other readers; fatalErr is read on exit. + stop := make(chan struct{}) + var stopOnce sync.Once + closeStop := func() { stopOnce.Do(func() { close(stop) }) } + var fatalMu sync.Mutex + var fatalErr error + setFatal := func(e error) { + fatalMu.Lock() + if fatalErr == nil { + fatalErr = e + } + fatalMu.Unlock() + closeStop() + } + for i, filerId := range filerOrder { entryCh := make(chan *filer_pb.LogEntry, 512) streams[i] = filerStream{filerId: filerId, entryCh: entryCh} @@ -152,10 +169,20 @@ func readMultiFilersMerged( go func(refs []*filer_pb.LogFileChunkRef, ch chan *filer_pb.LogEntry) { defer wg.Done() defer close(ch) - readFilerFilesToChannel(refs, newReader, startTsNs, stopTsNs, ch) + readFilerFilesToChannel(refs, newReader, startTsNs, stopTsNs, ch, stop, setFatal) }(perFiler[filerId], entryCh) } + // Stop readers, drain channels so none block on a send, then wait for exit. + drainAndWait := func() { + closeStop() + for i := range streams { + for range streams[i].entryCh { + } + } + wg.Wait() + } + // Seed the min-heap with the first entry from each filer pq := &logEntryHeap{} heap.Init(pq) @@ -167,15 +194,23 @@ func readMultiFilersMerged( // Merge loop for pq.Len() > 0 { + // stop is closed only by setFatal here, so a closed stop means a reader + // aborted; lock-free bail on the hot path. + select { + case <-stop: + drainAndWait() + fatalMu.Lock() + fe := fatalErr + fatalMu.Unlock() + return lastTsNs, fe + default: + } + item := heap.Pop(pq).(*logEntryHeapItem) lastTsNs, err = processOneLogEntry(item.entry, filter, processEventFn) if err != nil { - for i := range streams { - for range streams[i].entryCh { - } - } - wg.Wait() + drainAndWait() return } @@ -184,7 +219,13 @@ func readMultiFilersMerged( } } - wg.Wait() + drainAndWait() + fatalMu.Lock() + fe := fatalErr + fatalMu.Unlock() + if fe != nil { + return lastTsNs, fe + } return } @@ -193,6 +234,8 @@ func readFilerFilesToChannel( newReader LogFileReaderFn, startTsNs, stopTsNs int64, ch chan *filer_pb.LogEntry, + stop <-chan struct{}, + setFatal func(error), ) { type prefetchResult struct { entries []*filer_pb.LogEntry @@ -214,7 +257,12 @@ func readFilerFilesToChannel( } for i, ref := range refs { - result := <-pendingCh + var result prefetchResult + select { + case result = <-pendingCh: + case <-stop: + return + } if i+1 < len(refs) { pendingCh = startPrefetch(refs[i+1]) @@ -223,14 +271,19 @@ func readFilerFilesToChannel( if result.err != nil { if isChunkNotFound(result.err) { glog.V(0).Infof("skip log file filer=%s ts=%d: %v", ref.FilerId, ref.FileTsNs, result.err) - } else { - glog.Errorf("read log file filer=%s ts=%d: %v", ref.FilerId, ref.FileTsNs, result.err) + continue } - continue + glog.Errorf("read log file filer=%s ts=%d: %v", ref.FilerId, ref.FileTsNs, result.err) + setFatal(fmt.Errorf("read log file filer=%s ts=%d: %w", ref.FilerId, ref.FileTsNs, result.err)) + return } for _, entry := range result.entries { - ch <- entry + select { + case ch <- entry: + case <-stop: + return + } } } } diff --git a/weed/pb/filer_pb_direct_read_test.go b/weed/pb/filer_pb_direct_read_test.go index 68f71a306..19a8e20d8 100644 --- a/weed/pb/filer_pb_direct_read_test.go +++ b/weed/pb/filer_pb_direct_read_test.go @@ -278,3 +278,54 @@ func TestDirectReadVsServerSideThroughput(t *testing.T) { t.Logf("Speedup: %.1fx (parallel + prefetch + no gRPC vs server-side sequential)", directRate/serverRate) } } + +// failingReaderFn returns err for the given file key, delegating otherwise. +func failingReaderFn(base LogFileReaderFn, failKey string, err error) LogFileReaderFn { + return func(chunks []*filer_pb.FileChunk) (io.ReadCloser, error) { + if len(chunks) > 0 && chunks[0].FileId == failKey { + return nil, err + } + return base(chunks) + } +} + +// A real (non not-found) read error must fail the whole replay, not silently +// drop the file and advance the cursor. +func TestReadLogFileRefsMultiFilerGenuineErrorAborts(t *testing.T) { + files := newTestLogFiles(3, 2, 10, 0) + failKey := files.refs[2].Chunks[0].FileId // filer01's first file + readerFn := failingReaderFn(files.readerFn(), failKey, fmt.Errorf("failed to locate %s", failKey)) + + var count int64 + _, err := ReadLogFileRefs(files.refs, readerFn, 0, 0, + PathFilter{PathPrefix: "/"}, + func(resp *filer_pb.SubscribeMetadataResponse) error { + atomic.AddInt64(&count, 1) + return nil + }) + if err == nil { + t.Fatalf("expected error from genuine read failure, got nil (delivered=%d)", count) + } +} + +// A chunk-not-found error skips only that file (volume gone), not the replay. +func TestReadLogFileRefsMultiFilerNotFoundSkips(t *testing.T) { + files := newTestLogFiles(3, 2, 10, 0) + skipKey := files.refs[2].Chunks[0].FileId // filer01's first file + readerFn := failingReaderFn(files.readerFn(), skipKey, fmt.Errorf("volume not found: %s", skipKey)) + + var count int64 + _, err := ReadLogFileRefs(files.refs, readerFn, 0, 0, + PathFilter{PathPrefix: "/"}, + func(resp *filer_pb.SubscribeMetadataResponse) error { + atomic.AddInt64(&count, 1) + return nil + }) + if err != nil { + t.Fatalf("chunk-not-found should be skipped, got error: %v", err) + } + expected := int64(files.totalEvents() - 10) // one skipped file's events + if count != expected { + t.Fatalf("expected %d events after skipping one file, got %d", expected, count) + } +} diff --git a/weed/server/filer_grpc_server_sub_meta.go b/weed/server/filer_grpc_server_sub_meta.go index 6bec4e2c7..0b5a1c3f8 100644 --- a/weed/server/filer_grpc_server_sub_meta.go +++ b/weed/server/filer_grpc_server_sub_meta.go @@ -215,7 +215,7 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, if req.ClientSupportsMetadataChunks { processedTsNs, isDone, readPersistedLogErr = fs.sendLogFileRefs(ctx, stream, lastReadTime, req.UntilNs) } else { - processedTsNs, isDone, readPersistedLogErr = fs.filer.ReadPersistedLogBuffer(lastReadTime, req.UntilNs, eachLogEntryFn) + processedTsNs, isDone, readPersistedLogErr = fs.filer.ReadPersistedLogBuffer(ctx, lastReadTime, req.UntilNs, eachLogEntryFn) } if readPersistedLogErr != nil { return fmt.Errorf("reading from persisted logs: %w", readPersistedLogErr) @@ -367,7 +367,7 @@ func (fs *FilerServer) SubscribeLocalMetadata(req *filer_pb.SubscribeMetadataReq if req.ClientSupportsMetadataChunks { processedTsNs, isDone, readPersistedLogErr = fs.sendLogFileRefs(ctx, stream, lastReadTime, req.UntilNs) } else { - processedTsNs, isDone, readPersistedLogErr = fs.filer.ReadPersistedLogBuffer(lastReadTime, req.UntilNs, eachLogEntryFn) + processedTsNs, isDone, readPersistedLogErr = fs.filer.ReadPersistedLogBuffer(ctx, lastReadTime, req.UntilNs, eachLogEntryFn) } if readPersistedLogErr != nil { glog.V(0).Infof("read on disk %v local subscribe %s from %+v: %v", clientName, req.PathPrefix, lastReadTime, readPersistedLogErr)