diff --git a/weed/util/log_buffer/log_buffer.go b/weed/util/log_buffer/log_buffer.go index 01d6100c9..78235a5cc 100644 --- a/weed/util/log_buffer/log_buffer.go +++ b/weed/util/log_buffer/log_buffer.go @@ -189,9 +189,9 @@ type LogBuffer struct { notifyFn func() // Per-subscriber notification channels for instant wake-up subscribersMu sync.RWMutex - subscribers map[string]chan struct{} // subscriberID -> notification channel + subscribers map[string]*subscription // subscriberID -> shared notification channel // Notified only when a flush lands, for readers that cannot act on an append - flushSubscribers map[string]chan struct{} + flushSubscribers map[string]*subscription isStopping *atomic.Bool shutdownCh chan struct{} // closed by ShutdownLogBuffer to wake blocked subscribers loopsDone sync.WaitGroup // loopFlush and loopInterval signal exit @@ -223,8 +223,8 @@ func NewLogBuffer(name string, flushInterval time.Duration, flushFn LogFlushFunc flushFn: flushFn, ReadFromDiskFn: readFromDiskFn, notifyFn: notifyFn, - subscribers: make(map[string]chan struct{}), - flushSubscribers: make(map[string]chan struct{}), + subscribers: make(map[string]*subscription), + flushSubscribers: make(map[string]*subscription), flushChan: make(chan *dataToFlush, flushQueueDepth), isStopping: new(atomic.Bool), shutdownCh: make(chan struct{}), @@ -242,32 +242,59 @@ func NewLogBuffer(name string, flushInterval time.Duration, flushFn LogFlushFunc return lb } +// subscription is one notification channel and the number of readers holding +// it. Registrations that share a subscriberID share the channel - two streams +// of the same client, or an old one that has not noticed its replacement yet - +// so it may only be closed once the last of them unregisters. Closing it under +// a reader parked in awaitNotificationOrTimeoutFor makes every receive there +// succeed instantly, spinning that reader on a full core for the rest of its +// life. +type subscription struct { + notifyChan chan struct{} + refCount int +} + +func registerSubscription(subscriptions map[string]*subscription, subscriberID string) chan struct{} { + if existing, exists := subscriptions[subscriberID]; exists { + existing.refCount++ + return existing.notifyChan + } + + // Create buffered channel (size 1) so notifications never block + sub := &subscription{notifyChan: make(chan struct{}, 1), refCount: 1} + subscriptions[subscriberID] = sub + return sub.notifyChan +} + +func unregisterSubscription(subscriptions map[string]*subscription, subscriberID string) { + sub, exists := subscriptions[subscriberID] + if !exists { + return + } + sub.refCount-- + if sub.refCount > 0 { + return + } + close(sub.notifyChan) + delete(subscriptions, subscriberID) +} + // RegisterSubscriber registers a subscriber for instant notifications when data is written // Returns a channel that will receive notifications (<1ms latency) func (logBuffer *LogBuffer) RegisterSubscriber(subscriberID string) chan struct{} { logBuffer.subscribersMu.Lock() defer logBuffer.subscribersMu.Unlock() - // Check if already registered - if existingChan, exists := logBuffer.subscribers[subscriberID]; exists { - return existingChan - } - - // Create buffered channel (size 1) so notifications never block - notifyChan := make(chan struct{}, 1) - logBuffer.subscribers[subscriberID] = notifyChan - return notifyChan + return registerSubscription(logBuffer.subscribers, subscriberID) } // UnregisterSubscriber removes a subscriber and closes its notification channel +// once no other registration of the same subscriberID is left holding it. func (logBuffer *LogBuffer) UnregisterSubscriber(subscriberID string) { logBuffer.subscribersMu.Lock() defer logBuffer.subscribersMu.Unlock() - if ch, exists := logBuffer.subscribers[subscriberID]; exists { - close(ch) - delete(logBuffer.subscribers, subscriberID) - } + unregisterSubscription(logBuffer.subscribers, subscriberID) } // RegisterFlushSubscriber registers a subscriber woken only when a flush lands. @@ -279,23 +306,16 @@ func (logBuffer *LogBuffer) RegisterFlushSubscriber(subscriberID string) chan st logBuffer.subscribersMu.Lock() defer logBuffer.subscribersMu.Unlock() - if existingChan, exists := logBuffer.flushSubscribers[subscriberID]; exists { - return existingChan - } - notifyChan := make(chan struct{}, 1) - logBuffer.flushSubscribers[subscriberID] = notifyChan - return notifyChan + return registerSubscription(logBuffer.flushSubscribers, subscriberID) } // UnregisterFlushSubscriber removes a flush subscriber and closes its channel +// once no other registration of the same subscriberID is left holding it. func (logBuffer *LogBuffer) UnregisterFlushSubscriber(subscriberID string) { logBuffer.subscribersMu.Lock() defer logBuffer.subscribersMu.Unlock() - if ch, exists := logBuffer.flushSubscribers[subscriberID]; exists { - close(ch) - delete(logBuffer.flushSubscribers, subscriberID) - } + unregisterSubscription(logBuffer.flushSubscribers, subscriberID) } // IsOffsetInMemory checks if the given offset is available in the in-memory buffer @@ -357,9 +377,9 @@ func (logBuffer *LogBuffer) notifySubscribers() { return // No subscribers, skip notification } - for _, notifyChan := range logBuffer.subscribers { + for _, sub := range logBuffer.subscribers { select { - case notifyChan <- struct{}{}: + case sub.notifyChan <- struct{}{}: // Notification sent successfully default: // Channel full - subscriber hasn't consumed previous notification yet @@ -373,9 +393,9 @@ func (logBuffer *LogBuffer) notifyFlushSubscribers() { logBuffer.subscribersMu.RLock() defer logBuffer.subscribersMu.RUnlock() - for _, notifyChan := range logBuffer.flushSubscribers { + for _, sub := range logBuffer.flushSubscribers { select { - case notifyChan <- struct{}{}: + case sub.notifyChan <- struct{}{}: default: } } diff --git a/weed/util/log_buffer/log_read_test.go b/weed/util/log_buffer/log_read_test.go index f1c3eb38f..980d326f7 100644 --- a/weed/util/log_buffer/log_read_test.go +++ b/weed/util/log_buffer/log_read_test.go @@ -3,6 +3,7 @@ package log_buffer import ( "context" "sync" + "sync/atomic" "testing" "time" @@ -574,6 +575,71 @@ func TestLoopProcessLogData_SlowConsumerFallsBehind(t *testing.T) { } } +// TestLoopProcessLogData_DuplicateReaderLeaving is a regression test for issue +// #10810: an idle metadata subscriber pinning a full core inside +// awaitNotificationOrTimeoutFor. Two readers registered under the same name +// share one notification channel, which happens whenever a client reconnects +// before its previous stream noticed the disconnect. When the departing one +// closed that channel, the surviving reader's select won on it instantly on +// every pass, so the loop ran flat out - allocating a timer per iteration - +// until the client went away. +func TestLoopProcessLogData_DuplicateReaderLeaving(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() + + const readerName = "localMeta:s3@" + startPosition := NewMessagePosition(time.Now().UnixNano(), -2) + eachLogEntryFn := func(logEntry *filer_pb.LogEntry) (bool, error) { return false, nil } + + // startReader runs a reader and reports how many times it went round the + // loop; the loop calls waitForDataFn exactly once per pass. + startReader := func(iterations *atomic.Int64, stop *atomic.Bool) chan struct{} { + done := make(chan struct{}) + go func() { + defer close(done) + logBuffer.LoopProcessLogData(readerName, startPosition, 0, func() bool { + iterations.Add(1) + return !stop.Load() + }, eachLogEntryFn) + }() + for iterations.Load() == 0 { + time.Sleep(time.Millisecond) + } + return done + } + + var leavingIterations, survivorIterations atomic.Int64 + var stopLeaving, stopSurvivor atomic.Bool + leavingDone := startReader(&leavingIterations, &stopLeaving) + survivorDone := startReader(&survivorIterations, &stopSurvivor) + + stopLeaving.Store(true) + <-leavingDone + + const observeFor = 500 * time.Millisecond + before := survivorIterations.Load() + time.Sleep(observeFor) + spun := survivorIterations.Load() - before + + // A low iteration count only means "not spinning" if the reader was still + // there to spin. Had LoopProcessLogData returned when its duplicate left, + // spun would be 0 and the count below would pass while proving nothing. + select { + case <-survivorDone: + t.Fatal("surviving reader returned while its duplicate unregistered; it must keep reading") + default: + } + + stopSurvivor.Store(true) + <-survivorDone + + maxIterations := int64(observeFor/notificationHealthCheckInterval) + 1 + if spun > maxIterations { + t.Errorf("surviving reader looped %d times in %v, expected at most %d (suggests busy-waiting)", spun, observeFor, maxIterations) + } +} + // BenchmarkLoopProcessLogDataWithOffset_EmptyBuffer benchmarks the performance // of the loop with an empty buffer to ensure no busy-waiting func BenchmarkLoopProcessLogDataWithOffset_EmptyBuffer(b *testing.B) {