diff --git a/weed/util/log_buffer/log_buffer.go b/weed/util/log_buffer/log_buffer.go index 471d3e140..e8ee7e5ed 100644 --- a/weed/util/log_buffer/log_buffer.go +++ b/weed/util/log_buffer/log_buffer.go @@ -81,6 +81,7 @@ type LogBuffer struct { subscribersMu sync.RWMutex subscribers map[string]chan struct{} // subscriberID -> notification channel isStopping *atomic.Bool + shutdownCh chan struct{} // closed by ShutdownLogBuffer to wake blocked subscribers isAllFlushed bool flushChan chan *dataToFlush // Offset range tracking for Kafka integration @@ -104,6 +105,7 @@ func NewLogBuffer(name string, flushInterval time.Duration, flushFn LogFlushFunc subscribers: make(map[string]chan struct{}), flushChan: make(chan *dataToFlush, 256), isStopping: new(atomic.Bool), + shutdownCh: make(chan struct{}), offset: 0, // Will be initialized from existing data if available diskChunkCache: &DiskChunkCache{ chunks: make(map[int64]*CachedDiskChunk), @@ -492,6 +494,10 @@ func (logBuffer *LogBuffer) ShutdownLogBuffer() { if isAlreadyStopped { return } + // Wake any subscribers blocked in awaitNotificationOrTimeout so they can + // notice IsStopping() and exit promptly, even on an idle buffer where no + // flush notification would otherwise fire. + close(logBuffer.shutdownCh) toFlush := logBuffer.copyToFlush() logBuffer.flushChan <- toFlush close(logBuffer.flushChan) @@ -515,6 +521,12 @@ func (logBuffer *LogBuffer) loopFlush() { logBuffer.lastFlushTsNs.Store(d.stopTime.UnixNano()) } + // Wake readers that may be waiting to retry disk reads after the flush lands. + if logBuffer.notifyFn != nil { + logBuffer.notifyFn() + } + logBuffer.notifySubscribers() + // Signal completion if there's a callback channel if d.done != nil { close(d.done) diff --git a/weed/util/log_buffer/log_read.go b/weed/util/log_buffer/log_read.go index 11cda011f..5ad843905 100644 --- a/weed/util/log_buffer/log_read.go +++ b/weed/util/log_buffer/log_read.go @@ -18,6 +18,15 @@ var ( ResumeFromDiskError = fmt.Errorf("resumeFromDisk") ) +// notificationHealthCheckInterval bounds how long an idle subscriber blocks +// on the notification channel before re-checking state (client disconnect via +// waitForDataFn, LogBuffer shutdown, timestamp advancement). notifyChan is the +// primary wakeup path when new data arrives or a flush lands; this timeout is +// the safety net for any missed notification and also caps the latency to +// notice that the subscriber should exit. Balances idle CPU and log noise +// against client-disconnect detection latency. +const notificationHealthCheckInterval = 250 * time.Millisecond + type MessagePosition struct { Time time.Time // timestamp of the message Offset int64 // Kafka offset for offset-based positioning, or batch index for timestamp-based @@ -49,6 +58,26 @@ func (mp MessagePosition) GetOffset() int64 { return mp.Offset // Offset is stored directly } +// awaitNotificationOrTimeout blocks until one of: +// - a new-data / flush notification arrives on notifyChan (returns true) +// - the LogBuffer is shut down via ShutdownLogBuffer (returns true; callers +// re-check IsStopping() and exit) +// - notificationHealthCheckInterval elapses (returns false; caller +// re-checks client-disconnect and other state) +func (logBuffer *LogBuffer) awaitNotificationOrTimeout(notifyChan <-chan struct{}) bool { + timer := time.NewTimer(notificationHealthCheckInterval) + defer timer.Stop() + + select { + case <-notifyChan: + return true + case <-logBuffer.shutdownCh: + return true + case <-timer.C: + return false + } +} + func (logBuffer *LogBuffer) LoopProcessLogData(readerName string, startPosition MessagePosition, stopTsNs int64, waitForDataFn func() bool, eachLogDataFn EachLogEntryFuncType) (lastReadPosition MessagePosition, isDone bool, err error) { @@ -100,13 +129,19 @@ func (logBuffer *LogBuffer) LoopProcessLogData(readerName string, startPosition } // Wait for notification or timeout (instant wake-up when data arrives) - select { - case <-notifyChan: - // New data available, retry immediately + if logBuffer.awaitNotificationOrTimeout(notifyChan) { glog.V(3).Infof("%s: Woke up from notification after ResumeFromDiskError", readerName) - case <-time.After(10 * time.Millisecond): - // Timeout, retry anyway (fallback for edge cases) - glog.V(4).Infof("%s: Notification timeout after ResumeFromDiskError, polling", readerName) + } else { + glog.V(4).Infof("%s: Notification timeout after ResumeFromDiskError, rechecking state", readerName) + } + + // If the LogBuffer is shutting down, exit cleanly instead of looping + // on ResumeFromDiskError. awaitNotificationOrTimeout returns true on + // shutdown (shutdownCh closed), which would otherwise spin here + // because ReadFromBuffer keeps returning ResumeFromDiskError. + if logBuffer.IsStopping() { + isDone = true + return } // Continue to next iteration (don't return ResumeFromDiskError) @@ -143,17 +178,18 @@ func (logBuffer *LogBuffer) LoopProcessLogData(readerName string, startPosition return } // Wait for notification or timeout (instant wake-up when data arrives) - select { - case <-notifyChan: - // New data available, break and retry read + if logBuffer.awaitNotificationOrTimeout(notifyChan) { glog.V(3).Infof("%s: Woke up from notification (LoopProcessLogData)", readerName) + } else if lastTsNs != logBuffer.LastTsNs.Load() { break - case <-time.After(10 * time.Millisecond): - // Timeout, check if timestamp changed - if lastTsNs != logBuffer.LastTsNs.Load() { - break - } - glog.V(4).Infof("%s: Notification timeout (LoopProcessLogData), polling", readerName) + } else { + glog.V(4).Infof("%s: Notification timeout (LoopProcessLogData), rechecking state", readerName) + } + // Exit the wait loop on shutdown so we don't spin against a + // closed shutdownCh. + if logBuffer.IsStopping() { + isDone = true + return } } if logBuffer.IsStopping() { @@ -294,13 +330,15 @@ func (logBuffer *LogBuffer) LoopProcessLogDataWithOffset(readerName string, star } // Wait for notification or timeout (instant wake-up when data arrives) - select { - case <-notifyChan: - // New data available, retry immediately + if logBuffer.awaitNotificationOrTimeout(notifyChan) { glog.V(3).Infof("%s: Woke up from notification after disk read", readerName) - case <-time.After(10 * time.Millisecond): - // Timeout, retry anyway (fallback for edge cases) - glog.V(4).Infof("%s: Notification timeout, polling", readerName) + } else { + glog.V(4).Infof("%s: Notification timeout, rechecking state", readerName) + } + + // Exit cleanly on shutdown so we don't loop on ResumeFromDiskError. + if logBuffer.IsStopping() { + return lastReadPosition, true, nil } // Continue to next iteration (don't return ResumeFromDiskError) @@ -338,13 +376,14 @@ func (logBuffer *LogBuffer) LoopProcessLogDataWithOffset(readerName string, star return lastReadPosition, true, nil } // Wait for notification or timeout (instant wake-up when data arrives) - select { - case <-notifyChan: - // New data available, retry immediately + if logBuffer.awaitNotificationOrTimeout(notifyChan) { glog.V(3).Infof("%s: Woke up from notification for offset-based read", readerName) - case <-time.After(10 * time.Millisecond): - // Timeout, retry anyway (fallback for edge cases) - glog.V(4).Infof("%s: Notification timeout for offset-based, polling", readerName) + } else { + glog.V(4).Infof("%s: Notification timeout for offset-based, rechecking state", readerName) + } + // On shutdown, exit cleanly instead of returning ResumeFromDiskError. + if logBuffer.IsStopping() { + return lastReadPosition, true, nil } return lastReadPosition, isDone, ResumeFromDiskError } @@ -357,17 +396,18 @@ func (logBuffer *LogBuffer) LoopProcessLogDataWithOffset(readerName string, star return lastReadPosition, true, nil } // Wait for notification or timeout (instant wake-up when data arrives) - select { - case <-notifyChan: - // New data available, break and retry read + if logBuffer.awaitNotificationOrTimeout(notifyChan) { glog.V(3).Infof("%s: Woke up from notification (main loop)", readerName) + } else if lastTsNs != logBuffer.LastTsNs.Load() { break - case <-time.After(10 * time.Millisecond): - // Timeout, check if timestamp changed - if lastTsNs != logBuffer.LastTsNs.Load() { - break - } - glog.V(4).Infof("%s: Notification timeout (main loop), polling", readerName) + } else { + glog.V(4).Infof("%s: Notification timeout (main loop), rechecking state", readerName) + } + // Exit the wait loop on shutdown so we don't spin against a + // closed shutdownCh. + if logBuffer.IsStopping() { + glog.V(4).Infof("%s: LogBuffer is stopping", readerName) + return lastReadPosition, true, nil } } if logBuffer.IsStopping() { @@ -388,8 +428,15 @@ func (logBuffer *LogBuffer) LoopProcessLogDataWithOffset(readerName string, star glog.V(4).Infof("%s: Client disconnected on empty buffer", readerName) return lastReadPosition, true, nil } - // Sleep to avoid busy-wait on empty buffer - time.Sleep(10 * time.Millisecond) + if logBuffer.awaitNotificationOrTimeout(notifyChan) { + glog.V(3).Infof("%s: Woke up from notification on empty buffer", readerName) + } else { + glog.V(4).Infof("%s: Empty buffer timeout, rechecking state", readerName) + } + // Exit cleanly on shutdown to avoid an idle spin on the empty buffer. + if logBuffer.IsStopping() { + return lastReadPosition, true, nil + } continue } diff --git a/weed/util/log_buffer/log_read_test.go b/weed/util/log_buffer/log_read_test.go index c46d31287..5acbaf80e 100644 --- a/weed/util/log_buffer/log_read_test.go +++ b/weed/util/log_buffer/log_read_test.go @@ -61,7 +61,7 @@ func TestLoopProcessLogDataWithOffset_EmptyBuffer(t *testing.T) { defer logBuffer.ShutdownLogBuffer() callCount := 0 - maxCalls := 10 + maxCalls := 4 mu := sync.Mutex{} waitForDataFn := func() bool { @@ -87,14 +87,12 @@ func TestLoopProcessLogDataWithOffset_EmptyBuffer(t *testing.T) { t.Errorf("Expected isDone=true when waitForDataFn returns false, got false") } - // With 10ms sleep per iteration, 10 iterations should take ~100ms minimum - minExpectedTime := time.Duration(maxCalls-1) * 10 * time.Millisecond + minExpectedTime := time.Duration(maxCalls-1) * notificationHealthCheckInterval if elapsed < minExpectedTime { t.Errorf("Loop exited too quickly (%v), expected at least %v (suggests busy-waiting)", elapsed, minExpectedTime) } - // But shouldn't take more than 2x expected (allows for some overhead) - maxExpectedTime := time.Duration(maxCalls) * 30 * time.Millisecond + maxExpectedTime := time.Duration(maxCalls+1) * notificationHealthCheckInterval if elapsed > maxExpectedTime { t.Errorf("Loop took too long: %v (expected < %v)", elapsed, maxExpectedTime) } @@ -122,7 +120,7 @@ func TestLoopProcessLogDataWithOffset_NoDataResumeFromDisk(t *testing.T) { defer logBuffer.ShutdownLogBuffer() callCount := 0 - maxCalls := 5 + maxCalls := 3 mu := sync.Mutex{} waitForDataFn := func() bool { @@ -148,15 +146,198 @@ func TestLoopProcessLogDataWithOffset_NoDataResumeFromDisk(t *testing.T) { t.Errorf("Expected isDone=true when waitForDataFn returns false, got false") } - // Should take at least (maxCalls-1) * 10ms due to sleep in ResumeFromDiskError path - minExpectedTime := time.Duration(maxCalls-1) * 10 * time.Millisecond + minExpectedTime := time.Duration(maxCalls-1) * notificationHealthCheckInterval if elapsed < minExpectedTime { - t.Errorf("Loop exited too quickly (%v), expected at least %v (suggests missing sleep)", elapsed, minExpectedTime) + t.Errorf("Loop exited too quickly (%v), expected at least %v (suggests missing wait)", elapsed, minExpectedTime) } t.Logf("Loop exited cleanly in %v after %d iterations (proper sleep detected)", elapsed, callCount) } +// TestLoopFlush_NotifiesSubscribersAfterFlush is a regression test for the +// issue #9007 fix: loopFlush must call notifySubscribers() after processing a +// flush so that readers parked on notifyChan wake up when a flush lands. The +// classic bug scenario is a reader that got ResumeFromDiskError, did a disk +// read that raced the flush and found nothing, and is now blocked on +// notifyChan waiting for the data that just hit disk. +// +// We drain the AddToBuffer notification first, then ForceFlush, and assert a +// new notification is delivered on notifyChan well before the fallback +// timeout. If the loopFlush notification is removed, this test fails by +// hitting the fallback. +func TestLoopFlush_NotifiesSubscribersAfterFlush(t *testing.T) { + flushFn := func(logBuffer *LogBuffer, startTime, stopTime time.Time, buf []byte, minOffset, maxOffset int64) { + } + logBuffer := NewLogBuffer("test", 1*time.Minute, flushFn, nil, nil) + defer logBuffer.ShutdownLogBuffer() + + notifyChan := logBuffer.RegisterSubscriber("flush-notify-test") + defer logBuffer.UnregisterSubscriber("flush-notify-test") + + if err := logBuffer.AddToBuffer(&mq_pb.DataMessage{ + Key: []byte("k"), + Value: []byte("v"), + TsNs: time.Now().UnixNano(), + }); err != nil { + t.Fatalf("AddToBuffer: %v", err) + } + + // Consume the AddToBuffer notification so the channel starts empty. + select { + case <-notifyChan: + case <-time.After(100 * time.Millisecond): + t.Fatal("expected a notification from AddToBuffer") + } + + // ForceFlush waits for loopFlush to process the flush. After it returns, + // loopFlush must have called notifySubscribers() again. + start := time.Now() + logBuffer.ForceFlush() + + select { + case <-notifyChan: + elapsed := time.Since(start) + // The fallback timeout is notificationHealthCheckInterval; the flush + // notification should arrive well before that. + if elapsed >= notificationHealthCheckInterval { + t.Errorf("flush notification too slow: %v (>= fallback %v)", elapsed, notificationHealthCheckInterval) + } + t.Logf("flush notification delivered in %v", elapsed) + case <-time.After(notificationHealthCheckInterval): + t.Fatalf("loopFlush did not notify subscribers within %v", notificationHealthCheckInterval) + } +} + +// TestLoopProcessLogDataWithOffset_WakesOnDataArrival drives a real +// LoopProcessLogDataWithOffset reader from an empty buffer (readFromDiskFn +// returns nothing, forcing the reader to park on notifyChan after the +// ResumeFromDiskError branch), then adds data from another goroutine and +// asserts the reader completes well before the fallback timeout would fire. +// This protects the end-to-end wake-up path; the loopFlush-specific +// notification is covered by TestLoopFlush_NotifiesSubscribersAfterFlush. +func TestLoopProcessLogDataWithOffset_WakesOnDataArrival(t *testing.T) { + readFromDiskFn := func(startPosition MessagePosition, stopTsNs int64, eachLogEntryFn EachLogEntryFuncType) (MessagePosition, bool, error) { + // No data on disk; return unchanged so the reader parks on notifyChan. + return startPosition, false, nil + } + flushFn := func(logBuffer *LogBuffer, startTime, stopTime time.Time, buf []byte, minOffset, maxOffset int64) { + } + logBuffer := NewLogBuffer("test", 1*time.Minute, flushFn, readFromDiskFn, nil) + defer logBuffer.ShutdownLogBuffer() + + received := make(chan struct{}) + eachLogEntryFn := func(logEntry *filer_pb.LogEntry, offset int64) (bool, error) { + close(received) + return true, nil // isDone + } + + waitForDataFn := func() bool { return true } + + startPosition := NewMessagePositionFromOffset(0) + + readerDone := make(chan struct{}) + go func() { + _, _, _ = logBuffer.LoopProcessLogDataWithOffset( + "wake-test", startPosition, 0, waitForDataFn, eachLogEntryFn) + close(readerDone) + }() + + // Give the reader time to reach awaitNotificationOrTimeout. Both wake + // paths under test (notifyChan via AddToBuffer and shutdownCh via + // ShutdownLogBuffer) are race-free even if the reader hasn't parked yet + // — the notification stays buffered / shutdownCh stays closed — but a + // generous head start makes it likelier we exercise the actual park-then- + // wake path rather than the already-pending fast path. 50ms is well below + // notificationHealthCheckInterval (250ms) and tolerates slow CI. + time.Sleep(50 * time.Millisecond) + + start := time.Now() + if err := logBuffer.AddToBuffer(&mq_pb.DataMessage{ + Key: []byte("k"), + Value: []byte("v"), + TsNs: time.Now().UnixNano(), + }); err != nil { + t.Fatalf("AddToBuffer: %v", err) + } + + select { + case <-received: + case <-time.After(notificationHealthCheckInterval): + t.Fatalf("reader did not process the entry within %v (fallback timeout)", notificationHealthCheckInterval) + } + <-readerDone + elapsed := time.Since(start) + + if elapsed >= notificationHealthCheckInterval { + t.Errorf("reader wake too slow: %v (>= fallback %v)", elapsed, notificationHealthCheckInterval) + } + t.Logf("reader processed the entry in %v after AddToBuffer", elapsed) +} + +// TestLoopProcessLogDataWithOffset_WakesOnShutdown verifies that a reader +// parked inside awaitNotificationOrTimeout via the ResumeFromDiskError branch +// exits promptly when ShutdownLogBuffer is called, without waiting for the +// 250ms health-check fallback. Regression guard for the IsStopping() shutdown +// path: if awaitNotificationOrTimeout returns true via shutdownCh and the +// caller does not check IsStopping(), the reader either spins against the +// closed shutdownCh or returns ResumeFromDiskError instead of exiting. +func TestLoopProcessLogDataWithOffset_WakesOnShutdown(t *testing.T) { + readFromDiskFn := func(startPosition MessagePosition, stopTsNs int64, eachLogEntryFn EachLogEntryFuncType) (MessagePosition, bool, error) { + // No data on disk; return unchanged so the reader parks on notifyChan. + return startPosition, false, nil + } + flushFn := func(logBuffer *LogBuffer, startTime, stopTime time.Time, buf []byte, minOffset, maxOffset int64) { + } + logBuffer := NewLogBuffer("test", 1*time.Minute, flushFn, readFromDiskFn, nil) + // Note: not deferring ShutdownLogBuffer; we trigger it explicitly below. + + eachLogEntryFn := func(logEntry *filer_pb.LogEntry, offset int64) (bool, error) { + return false, nil + } + waitForDataFn := func() bool { return true } + startPosition := NewMessagePositionFromOffset(0) + + type result struct { + isDone bool + err error + } + resultCh := make(chan result, 1) + go func() { + _, isDone, err := logBuffer.LoopProcessLogDataWithOffset( + "shutdown-test", startPosition, 0, waitForDataFn, eachLogEntryFn) + resultCh <- result{isDone: isDone, err: err} + }() + + // Give the reader time to reach awaitNotificationOrTimeout. Both wake + // paths under test (notifyChan via AddToBuffer and shutdownCh via + // ShutdownLogBuffer) are race-free even if the reader hasn't parked yet + // — the notification stays buffered / shutdownCh stays closed — but a + // generous head start makes it likelier we exercise the actual park-then- + // wake path rather than the already-pending fast path. 50ms is well below + // notificationHealthCheckInterval (250ms) and tolerates slow CI. + time.Sleep(50 * time.Millisecond) + + start := time.Now() + logBuffer.ShutdownLogBuffer() + + select { + case r := <-resultCh: + elapsed := time.Since(start) + if elapsed >= notificationHealthCheckInterval { + t.Errorf("reader did not wake on shutdown: %v (>= fallback %v)", elapsed, notificationHealthCheckInterval) + } + if !r.isDone { + t.Errorf("expected isDone=true on shutdown, got false") + } + if r.err != nil { + t.Errorf("expected err=nil on shutdown, got %v", r.err) + } + t.Logf("reader exited in %v after ShutdownLogBuffer", elapsed) + case <-time.After(2 * notificationHealthCheckInterval): + t.Fatalf("reader did not exit within %v after ShutdownLogBuffer", 2*notificationHealthCheckInterval) + } +} + // TestLoopProcessLogDataWithOffset_WithData tests normal operation with data func TestLoopProcessLogDataWithOffset_WithData(t *testing.T) { flushFn := func(logBuffer *LogBuffer, startTime, stopTime time.Time, buf []byte, minOffset, maxOffset int64) {}