diff --git a/weed/util/log_buffer/log_buffer.go b/weed/util/log_buffer/log_buffer.go index 33f5ac317..86318072f 100644 --- a/weed/util/log_buffer/log_buffer.go +++ b/weed/util/log_buffer/log_buffer.go @@ -94,6 +94,12 @@ type LogBuffer struct { hasOffsets bool // Disk chunk cache for historical data reads diskChunkCache *DiskChunkCache + // curSnap is a GC-owned copy of the current window's append-only prefix + // buf[:len(curSnap)], shared by all readers so each byte is copied once per + // window instead of once per reader. Extended lazily under curSnapMu; reset + // at seal. Existing holders keep their prefix slices, which never mutate. + curSnapMu sync.Mutex + curSnap []byte sync.RWMutex } @@ -607,6 +613,13 @@ func (logBuffer *LogBuffer) copyToFlushInternal(withCallback bool) *dataToFlush // CRITICAL: logBuffer.offset is the "next offset to assign", so last offset in buffer is offset-1 lastOffsetInBuffer := logBuffer.offset - 1 logBuffer.buf = logBuffer.prevBuffers.SealBuffer(logBuffer.startTime, logBuffer.stopTime, logBuffer.buf, logBuffer.pos, logBuffer.bufferStartOffset, lastOffsetInBuffer) + // Hand a fully extended prefix snapshot to the sealed slot so sealed + // readers reuse it instead of re-copying the window; reset for the next + // window either way (holders keep their immutable prefix slices). + if len(logBuffer.curSnap) == logBuffer.pos { + logBuffer.prevBuffers.buffers[len(logBuffer.prevBuffers.buffers)-1].snapshot = logBuffer.curSnap[:logBuffer.pos:logBuffer.pos] + } + logBuffer.curSnap = nil // Use zero time (time.Time{}) not epoch time (time.Unix(0,0)) // Epoch time (1970) breaks time-based reads after flush logBuffer.startTime = time.Time{} @@ -702,7 +715,11 @@ func (d *dataToFlush) releaseMemory() { bufferPool.Put(d.data) } -func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bufferCopy *bytes.Buffer, batchIndex int64, err error) { +// ReadFromBuffer returns the in-memory log data at lastReadPosition. isPooled +// reports whether the returned buffer is a private pooled copy the caller must +// return via ReleaseMemory; when false the buffer wraps a snapshot shared with +// other readers and must not be released (or written to). +func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bufferCopy *bytes.Buffer, batchIndex int64, isPooled bool, err error) { logBuffer.RLock() defer logBuffer.RUnlock() @@ -729,19 +746,19 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu // Case 3: try disk read (historical data might exist) if requestedOffset < logBuffer.offset { // Data was in the buffer range but buffer is now empty = flushed to disk - return nil, -2, ResumeFromDiskError + return nil, -2, false, ResumeFromDiskError } // requestedOffset == logBuffer.offset: Current position // CRITICAL: For subscribers starting from offset 0, try disk read first // (historical data might exist from previous runs) if requestedOffset == 0 && logBuffer.bufferStartOffset == 0 && logBuffer.offset == 0 { // Initial state: try disk read before waiting for new data - return nil, -2, ResumeFromDiskError + return nil, -2, false, ResumeFromDiskError } // Otherwise, wait for new data to arrive - return nil, logBuffer.offset, nil + return nil, logBuffer.offset, false, nil } - return copiedBytes(logBuffer.buf[:logBuffer.pos]), logBuffer.offset, nil + return logBuffer.currentSnapshotView(0, logBuffer.pos), logBuffer.offset, false, nil } // Check previous buffers for the requested offset @@ -751,9 +768,9 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu // (prevBuffers are created when buffer is flushed) if buf.size == 0 { // Empty prevBuffer covering this offset means data was flushed - return nil, -2, ResumeFromDiskError + return nil, -2, false, ResumeFromDiskError } - return copiedBytes(buf.buf[:buf.size]), buf.offset, nil + return sharedBufferView(buf, 0), buf.offset, false, nil } } @@ -761,16 +778,16 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu if requestedOffset < logBuffer.bufferStartOffset { // Data not in current buffers - must be on disk (flushed or never existed) // Return ResumeFromDiskError to trigger disk read - return nil, -2, ResumeFromDiskError + return nil, -2, false, ResumeFromDiskError } if requestedOffset > logBuffer.offset { // Future data, not available yet - return nil, logBuffer.offset, nil + return nil, logBuffer.offset, false, nil } // Offset not found - return nil - return nil, logBuffer.offset, nil + return nil, logBuffer.offset, false, nil } // TIMESTAMP-BASED READ (original logic) @@ -799,7 +816,7 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu // Buffer is empty - return ResumeFromDiskError so caller can read from disk // This fixes issue #4977 where SubscribeMetadata stalls because // MetaAggregator.MetaLogBuffer is empty in single-filer setups - return nil, -2, ResumeFromDiskError + return nil, -2, false, ResumeFromDiskError } else if lastReadPosition.Time.Before(tsMemory) { // case 2.3 // For time-based reads, only check timestamp for disk reads // Don't use offset comparisons as they're not meaningful for time-based subscriptions @@ -813,7 +830,7 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu // Treat first read with sentinel/zero offset as inclusive of earliest in-memory data } else { // Data not in memory buffers - read from disk - return nil, -2, ResumeFromDiskError + return nil, -2, false, ResumeFromDiskError } } @@ -822,17 +839,17 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu if lastReadPosition.Time.Equal(logBuffer.stopTime) && !logBuffer.stopTime.IsZero() { // For first-read sentinel/zero offset, allow inclusive read at the boundary if lastReadPosition.Offset > 0 { - return nil, logBuffer.offset, nil + return nil, logBuffer.offset, false, nil } } if lastReadPosition.Time.After(logBuffer.stopTime) && !logBuffer.stopTime.IsZero() { - return nil, logBuffer.offset, nil + return nil, logBuffer.offset, false, nil } // Also check prevBuffers when current buffer is empty (startTime is zero) if lastReadPosition.Time.Before(logBuffer.startTime) || logBuffer.startTime.IsZero() { for _, buf := range logBuffer.prevBuffers.buffers { if buf.startTime.After(lastReadPosition.Time) { - return copiedBytes(buf.buf[:buf.size]), buf.offset, nil + return sharedBufferView(buf, 0), buf.offset, false, nil } if !buf.startTime.After(lastReadPosition.Time) && buf.stopTime.After(lastReadPosition.Time) { searchTime := lastReadPosition.Time @@ -843,19 +860,19 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu if err != nil { // Buffer corruption detected - return error wrapped with ErrBufferCorrupted glog.Errorf("ReadFromBuffer: buffer corruption in prevBuffer: %v", err) - return nil, -1, fmt.Errorf("%w: %v", ErrBufferCorrupted, err) + return nil, -1, false, fmt.Errorf("%w: %v", ErrBufferCorrupted, err) } if pos < buf.size { - return copiedBytes(buf.buf[pos:buf.size]), buf.offset, nil + return sharedBufferView(buf, pos), buf.offset, false, nil } } } // If current buffer is not empty, return it if logBuffer.pos > 0 { - return copiedBytes(logBuffer.buf[:logBuffer.pos]), logBuffer.offset, nil + return logBuffer.currentSnapshotView(0, logBuffer.pos), logBuffer.offset, false, nil } // Buffer is empty and no data in prevBuffers - wait for new data - return nil, logBuffer.offset, nil + return nil, logBuffer.offset, false, nil } lastTs := lastReadPosition.Time.UnixNano() @@ -887,7 +904,7 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu if err != nil { // Buffer corruption detected in binary search glog.Errorf("ReadFromBuffer: buffer corruption at idx[%d] pos %d: %v", mid, pos, err) - return nil, -1, fmt.Errorf("%w: %v", ErrBufferCorrupted, err) + return nil, -1, false, fmt.Errorf("%w: %v", ErrBufferCorrupted, err) } if t <= searchTs { l = mid + 1 @@ -898,11 +915,11 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu if err != nil { // Buffer corruption detected in binary search (previous entry) glog.Errorf("ReadFromBuffer: buffer corruption at idx[%d] pos %d: %v", mid-1, logBuffer.idx[mid-1], err) - return nil, -1, fmt.Errorf("%w: %v", ErrBufferCorrupted, err) + return nil, -1, false, fmt.Errorf("%w: %v", ErrBufferCorrupted, err) } } if prevT <= searchTs { - return copiedBytes(logBuffer.buf[pos:logBuffer.pos]), logBuffer.offset, nil + return logBuffer.currentSnapshotView(pos, logBuffer.pos), logBuffer.offset, false, nil } h = mid } @@ -910,9 +927,13 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu // Binary search didn't find the timestamp - data may have been flushed to disk already // Returning -2 signals to caller that data is not available in memory - return nil, -2, nil + return nil, -2, false, nil } + +// ReleaseMemory returns a pooled buffer for reuse. Only call it for buffers +// ReadFromBuffer reported as pooled: recycling a shared sealed-window view +// would let the next copiedBytes overwrite bytes other readers still hold. func (logBuffer *LogBuffer) ReleaseMemory(b *bytes.Buffer) { bufferPool.Put(b) } @@ -944,6 +965,37 @@ func copiedBytes(buf []byte) (copied *bytes.Buffer) { return } +// sharedBufferView wraps the sealed window's shared snapshot from pos onward. +// The returned buffer aliases memory shared with other readers: it must be +// treated as read-only and never passed to ReleaseMemory (the full-slice cap +// makes an accidental append reallocate instead of scribbling on the snapshot). +func sharedBufferView(mb *MemBuffer, pos int) *bytes.Buffer { + snap := mb.sharedSnapshot() + return bytes.NewBuffer(snap[pos:len(snap):len(snap)]) +} + +// currentSnapshotView returns buf[from:to] of the current window as a view of +// the shared prefix snapshot, extending the snapshot to cover [0:to) first. +// buf[:pos] is append-only until the window seals, so extension only ever +// appends stable bytes; earlier holders' slices are unaffected. Callers must +// hold the read lock (keeps buf and pos stable during extension). +func (logBuffer *LogBuffer) currentSnapshotView(from, to int) *bytes.Buffer { + logBuffer.curSnapMu.Lock() + if cap(logBuffer.curSnap) < to { + // First use in this window (or the rare window whose array outgrew the + // previous one): size to the window array so extensions never reallocate. + grown := make([]byte, len(logBuffer.curSnap), len(logBuffer.buf)) + copy(grown, logBuffer.curSnap) + logBuffer.curSnap = grown + } + if len(logBuffer.curSnap) < to { + logBuffer.curSnap = append(logBuffer.curSnap, logBuffer.buf[len(logBuffer.curSnap):to]...) + } + snap := logBuffer.curSnap[:to] + logBuffer.curSnapMu.Unlock() + return bytes.NewBuffer(snap[from:to:to]) +} + func readTs(buf []byte, pos int) (size int, ts int64, err error) { // Bounds check for size field (overflow-safe) if pos < 0 || pos > len(buf)-4 { diff --git a/weed/util/log_buffer/log_buffer_corruption_test.go b/weed/util/log_buffer/log_buffer_corruption_test.go index 2f7a029e6..1d2a3fb67 100644 --- a/weed/util/log_buffer/log_buffer_corruption_test.go +++ b/weed/util/log_buffer/log_buffer_corruption_test.go @@ -118,7 +118,7 @@ func TestReadFromBufferCorruption(t *testing.T) { // Try to read - should detect corruption startPos := MessagePosition{Time: lb.startTime} - buf, offset, err := lb.ReadFromBuffer(startPos) + buf, offset, _, err := lb.ReadFromBuffer(startPos) // Should return corruption error if err == nil { diff --git a/weed/util/log_buffer/log_buffer_flush_gap_test.go b/weed/util/log_buffer/log_buffer_flush_gap_test.go index dc010f1b8..9a4536890 100644 --- a/weed/util/log_buffer/log_buffer_flush_gap_test.go +++ b/weed/util/log_buffer/log_buffer_flush_gap_test.go @@ -148,7 +148,7 @@ func TestFlushOffsetGap_ReproduceDataLoss(t *testing.T) { for testOffset := int64(0); testOffset < currentOffset; testOffset += 10 { // Try to read from buffer requestPosition := NewMessagePositionFromOffset(testOffset) - buf, _, err := logBuffer.ReadFromBuffer(requestPosition) + buf, _, _, err := logBuffer.ReadFromBuffer(requestPosition) isReadable := (buf != nil && len(buf.Bytes()) > 0) || err == ResumeFromDiskError status := "OK" diff --git a/weed/util/log_buffer/log_buffer_queryability_test.go b/weed/util/log_buffer/log_buffer_queryability_test.go index 4774f25d8..ebe4b4510 100644 --- a/weed/util/log_buffer/log_buffer_queryability_test.go +++ b/weed/util/log_buffer/log_buffer_queryability_test.go @@ -50,7 +50,7 @@ func TestBufferQueryability(t *testing.T) { // Test immediate queryability - read from buffer starting from beginning startPosition := NewMessagePosition(0, 0) // Start from beginning - bufferCopy, batchIndex, err := logBuffer.ReadFromBuffer(startPosition) + bufferCopy, batchIndex, _, err := logBuffer.ReadFromBuffer(startPosition) if err != nil { t.Fatalf("ReadFromBuffer failed: %v", err) @@ -131,7 +131,7 @@ func TestMultipleEntriesQueryability(t *testing.T) { // Read all entries startPosition := NewMessagePosition(0, 0) - bufferCopy, batchIndex, err := logBuffer.ReadFromBuffer(startPosition) + bufferCopy, batchIndex, _, err := logBuffer.ReadFromBuffer(startPosition) if err != nil { t.Fatalf("ReadFromBuffer failed: %v", err) @@ -207,7 +207,7 @@ func TestSchemaRegistryScenario(t *testing.T) { // Simulate the SQL query scenario - read from offset 0 startPosition := NewMessagePosition(0, 0) - bufferCopy, _, err := logBuffer.ReadFromBuffer(startPosition) + bufferCopy, _, _, err := logBuffer.ReadFromBuffer(startPosition) if err != nil { t.Fatalf("Schema registry scenario failed: %v", err) @@ -268,7 +268,7 @@ func TestTimeBasedFirstReadBeforeEarliest(t *testing.T) { // Start read 1ns before earliest memory, with offset sentinel (-2) startPos := NewMessagePosition(baseTs.Add(-time.Nanosecond).UnixNano(), -2) - buf, _, err := logBuffer.ReadFromBuffer(startPos) + buf, _, _, err := logBuffer.ReadFromBuffer(startPos) if err != nil { t.Fatalf("ReadFromBuffer returned err: %v", err) } @@ -293,7 +293,7 @@ func TestEarliestTimeExactRead(t *testing.T) { } startPos := NewMessagePosition(ts.UnixNano(), -2) - buf, _, err := logBuffer.ReadFromBuffer(startPos) + buf, _, _, err := logBuffer.ReadFromBuffer(startPos) if err != nil { t.Fatalf("ReadFromBuffer err: %v", err) } diff --git a/weed/util/log_buffer/log_buffer_test.go b/weed/util/log_buffer/log_buffer_test.go index 1d4243f54..64798819c 100644 --- a/weed/util/log_buffer/log_buffer_test.go +++ b/weed/util/log_buffer/log_buffer_test.go @@ -101,7 +101,7 @@ func TestReadFromBufferTimestampBased_AfterFlushReturnsNewerData(t *testing.T) { } } - buf, _, err := lb.ReadFromBuffer(NewMessagePosition(sealed.stopTime.UnixNano(), sealed.offset)) + buf, _, _, err := lb.ReadFromBuffer(NewMessagePosition(sealed.stopTime.UnixNano(), sealed.offset)) if err != nil { t.Fatalf("ReadFromBuffer returned error: %v", err) } @@ -200,7 +200,7 @@ func TestReadFromBuffer_OldOffsetReturnsResumeFromDiskError(t *testing.T) { requestPosition := NewMessagePositionFromOffset(tt.requestedOffset) // Try to read from the buffer - buf, batchIdx, err := lb.ReadFromBuffer(requestPosition) + buf, batchIdx, _, err := lb.ReadFromBuffer(requestPosition) // Verify the error matches expectations if tt.expectError != nil { @@ -257,7 +257,7 @@ func TestReadFromBuffer_OldOffsetWithNoPrevBuffers(t *testing.T) { // Before the fix, this would return (nil, offset, nil) causing an infinite wait // After the fix, this should return ResumeFromDiskError - buf, batchIdx, err := lb.ReadFromBuffer(requestPosition) + buf, batchIdx, _, err := lb.ReadFromBuffer(requestPosition) t.Logf("DEBUG: ReadFromBuffer returned: buf=%v, batchIdx=%d, err=%v", buf != nil, batchIdx, err) t.Logf("DEBUG: Buffer state: bufferStartOffset=%d, offset=%d, pos=%d", @@ -295,7 +295,7 @@ func TestReadFromBuffer_EmptyBufferAtCurrentOffset(t *testing.T) { // BUG: Without fix, this returns empty buffer instead of checking disk // FIX: Should return ResumeFromDiskError because buffer is empty (pos=0) despite valid range - buf, batchIdx, err := lb.ReadFromBuffer(requestPosition) + buf, batchIdx, _, err := lb.ReadFromBuffer(requestPosition) t.Logf("DEBUG: ReadFromBuffer returned: buf=%v, batchIdx=%d, err=%v", buf != nil, batchIdx, err) t.Logf("DEBUG: Buffer state: bufferStartOffset=%d, offset=%d, pos=%d", @@ -362,7 +362,7 @@ func TestReadFromBuffer_OffsetRanges(t *testing.T) { for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { requestPosition := NewMessagePositionFromOffset(tc.requestedOffset) - _, _, err := lb.ReadFromBuffer(requestPosition) + _, _, _, err := lb.ReadFromBuffer(requestPosition) if tc.expectedError != nil { if err != tc.expectedError { @@ -424,7 +424,7 @@ func TestReadFromBuffer_InitializedFromDisk(t *testing.T) { // Schema Registry tries to read offset 0 (should be on disk) requestPosition := NewMessagePositionFromOffset(0) - buf, batchIdx, err := lb.ReadFromBuffer(requestPosition) + buf, batchIdx, _, err := lb.ReadFromBuffer(requestPosition) t.Logf("After writing new message:") t.Logf(" bufferStartOffset=%d, offset=%d, pos=%d", lb.bufferStartOffset, lb.offset, lb.pos) diff --git a/weed/util/log_buffer/log_read.go b/weed/util/log_buffer/log_read.go index 7557005ae..977787cf5 100644 --- a/weed/util/log_buffer/log_read.go +++ b/weed/util/log_buffer/log_read.go @@ -101,6 +101,7 @@ func (logBuffer *LogBuffer) LoopProcessLogData(readerName string, startPosition // loop through all messages var bytesBuf *bytes.Buffer + var bytesBufPooled bool var batchIndex int64 lastReadPosition = startPosition var entryCounter int64 @@ -113,7 +114,7 @@ func (logBuffer *LogBuffer) LoopProcessLogData(readerName string, startPosition // responsive 250ms cadence resumes for active readers. caughtUpToDiskHead := false defer func() { - if bytesBuf != nil { + if bytesBuf != nil && bytesBufPooled { logBuffer.ReleaseMemory(bytesBuf) } // println("LoopProcessLogData", readerName, "sent messages total", entryCounter) @@ -121,10 +122,11 @@ func (logBuffer *LogBuffer) LoopProcessLogData(readerName string, startPosition for { - if bytesBuf != nil { + if bytesBuf != nil && bytesBufPooled { logBuffer.ReleaseMemory(bytesBuf) + bytesBuf = nil // keep the deferred release from double-freeing if ReadFromBuffer panics } - bytesBuf, batchIndex, err = logBuffer.ReadFromBuffer(lastReadPosition) + bytesBuf, batchIndex, bytesBufPooled, err = logBuffer.ReadFromBuffer(lastReadPosition) if err == ResumeFromDiskError { // Try to read from disk if readFromDiskFn is available if logBuffer.ReadFromDiskFn != nil { @@ -148,11 +150,11 @@ func (logBuffer *LogBuffer) LoopProcessLogData(readerName string, startPosition // HasData() and ReadFromBuffer lock separately, so a racing write can make // HasData() see data the empty-buffer read missed. Re-read; only bail if the // position is genuinely behind the in-memory window (flushed to disk). - reBuf, _, reErr := logBuffer.ReadFromBuffer(lastReadPosition) + reBuf, _, rePooled, reErr := logBuffer.ReadFromBuffer(lastReadPosition) if reErr == ResumeFromDiskError { return lastReadPosition, isDone, ResumeFromDiskError } - if reBuf != nil { + if reBuf != nil && rePooled { logBuffer.ReleaseMemory(reBuf) } continue @@ -315,13 +317,14 @@ func (logBuffer *LogBuffer) LoopProcessLogDataWithOffset(readerName string, star // loop through all messages var bytesBuf *bytes.Buffer + var bytesBufPooled bool var offset int64 lastReadPosition = startPosition var entryCounter int64 // See LoopProcessLogData for the caughtUpToDiskHead invariant. caughtUpToDiskHead := false defer func() { - if bytesBuf != nil { + if bytesBuf != nil && bytesBufPooled { logBuffer.ReleaseMemory(bytesBuf) } // println("LoopProcessLogDataWithOffset", readerName, "sent messages total", entryCounter) @@ -335,10 +338,11 @@ func (logBuffer *LogBuffer) LoopProcessLogDataWithOffset(readerName string, star return } - if bytesBuf != nil { + if bytesBuf != nil && bytesBufPooled { logBuffer.ReleaseMemory(bytesBuf) + bytesBuf = nil // keep the deferred release from double-freeing if ReadFromBuffer panics } - bytesBuf, offset, err = logBuffer.ReadFromBuffer(lastReadPosition) + bytesBuf, offset, bytesBufPooled, err = logBuffer.ReadFromBuffer(lastReadPosition) glog.V(4).Infof("ReadFromBuffer for %s returned bytesBuf=%v, offset=%d, err=%v", readerName, bytesBuf != nil, offset, err) // Check for buffer corruption error before other error handling @@ -374,11 +378,11 @@ func (logBuffer *LogBuffer) LoopProcessLogDataWithOffset(readerName string, star // HasData() and ReadFromBuffer lock separately, so a racing write can make // HasData() see data the empty-buffer read missed. Re-read; only bail if the // position is genuinely behind the in-memory window (flushed to disk). - reBuf, _, reErr := logBuffer.ReadFromBuffer(lastReadPosition) + reBuf, _, rePooled, reErr := logBuffer.ReadFromBuffer(lastReadPosition) if reErr == ResumeFromDiskError { return lastReadPosition, isDone, ResumeFromDiskError } - if reBuf != nil { + if reBuf != nil && rePooled { logBuffer.ReleaseMemory(reBuf) } continue diff --git a/weed/util/log_buffer/sealed_buffer.go b/weed/util/log_buffer/sealed_buffer.go index e00148500..a637b23f8 100644 --- a/weed/util/log_buffer/sealed_buffer.go +++ b/weed/util/log_buffer/sealed_buffer.go @@ -2,6 +2,7 @@ package log_buffer import ( "fmt" + "sync" "time" ) @@ -12,6 +13,25 @@ type MemBuffer struct { stopTime time.Time startOffset int64 // First offset in this buffer offset int64 // Last offset in this buffer (endOffset) + + // snapshot is a GC-owned copy of buf[:size] shared by all readers of this + // sealed window, so N subscribers reading the same window cost one copy + // instead of N pooled copies. Created lazily by the first reader; travels + // with the window when SealBuffer shifts slots. Immutable once created. + snapMu sync.Mutex + snapshot []byte +} + +// sharedSnapshot returns the shared read-only copy of this sealed window, +// creating it on first use. Callers must hold the LogBuffer read lock, which +// keeps buf stable (SealBuffer mutates slots only under the write lock). +func (mb *MemBuffer) sharedSnapshot() []byte { + mb.snapMu.Lock() + defer mb.snapMu.Unlock() + if mb.snapshot == nil { + mb.snapshot = append([]byte(nil), mb.buf[:mb.size]...) + } + return mb.snapshot } type SealedBuffers struct { @@ -41,6 +61,7 @@ func (sbs *SealedBuffers) SealBuffer(startTime, stopTime time.Time, buf []byte, sbs.buffers[i].stopTime = sbs.buffers[i+1].stopTime sbs.buffers[i].startOffset = sbs.buffers[i+1].startOffset sbs.buffers[i].offset = sbs.buffers[i+1].offset + sbs.buffers[i].snapshot = sbs.buffers[i+1].snapshot // snapshot follows its window } sbs.buffers[size-1].buf = buf sbs.buffers[size-1].size = pos @@ -48,6 +69,7 @@ func (sbs *SealedBuffers) SealBuffer(startTime, stopTime time.Time, buf []byte, sbs.buffers[size-1].stopTime = stopTime sbs.buffers[size-1].startOffset = startOffset sbs.buffers[size-1].offset = endOffset + sbs.buffers[size-1].snapshot = nil return oldBuf } diff --git a/weed/util/log_buffer/shared_snapshot_test.go b/weed/util/log_buffer/shared_snapshot_test.go new file mode 100644 index 000000000..370f9a82c --- /dev/null +++ b/weed/util/log_buffer/shared_snapshot_test.go @@ -0,0 +1,268 @@ +package log_buffer + +import ( + "fmt" + "sync" + "testing" + "time" + + "google.golang.org/protobuf/proto" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" +) + +func addEntries(tb testing.TB, lb *LogBuffer, n int, payloadSize int, tag string) { + payload := make([]byte, payloadSize) + for i := 0; i < n; i++ { + key := []byte(fmt.Sprintf("/%s/%d", tag, i)) + if err := lb.AddDataToBuffer(key, payload, time.Now().UnixNano()); err != nil { + tb.Fatal(err) + } + } +} + +// Sealed-window reads must hand every reader the same shared snapshot rather +// than a fresh copy each. +func TestSealedReadsShareOneSnapshot(t *testing.T) { + lb := NewLogBuffer("test", time.Hour, func(*LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, nil, nil) + defer lb.ShutdownLogBuffer() + + addEntries(t, lb, 10, 1024, "a") + lb.ForceFlush() // seal the window + addEntries(t, lb, 1, 1024, "b") + + pos := NewMessagePosition(1, -2) // far in the past => sealed window hit + buf1, _, pooled1, err1 := lb.ReadFromBuffer(pos) + buf2, _, pooled2, err2 := lb.ReadFromBuffer(pos) + if err1 != nil || err2 != nil { + t.Fatalf("read errors: %v %v", err1, err2) + } + if buf1 == nil || buf2 == nil { + t.Fatal("expected sealed data in memory") + } + if pooled1 || pooled2 { + t.Fatalf("sealed reads should not be pooled copies: %v %v", pooled1, pooled2) + } + b1, b2 := buf1.Bytes(), buf2.Bytes() + if len(b1) == 0 || len(b1) != len(b2) { + t.Fatalf("unexpected lengths %d %d", len(b1), len(b2)) + } + if &b1[0] != &b2[0] { + t.Fatal("sealed reads made independent copies; expected one shared snapshot") + } +} + +// A shared snapshot must stay intact after its window rotates out and the +// backing array is recycled and overwritten by new writes. +func TestSharedSnapshotSurvivesRecycle(t *testing.T) { + lb := NewLogBuffer("test", time.Hour, func(*LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, nil, nil) + defer lb.ShutdownLogBuffer() + + addEntries(t, lb, 5, 2048, "keep") + lb.ForceFlush() + + pos := NewMessagePosition(1, -2) + buf, _, pooled, err := lb.ReadFromBuffer(pos) + if err != nil || buf == nil || pooled { + t.Fatalf("expected shared sealed read, got buf=%v pooled=%v err=%v", buf != nil, pooled, err) + } + before := append([]byte(nil), buf.Bytes()...) + + // Rotate the sealed window all the way out so its array is recycled and + // overwritten with different content. + for i := 0; i < PreviousBufferCount+1; i++ { + addEntries(t, lb, 5, 2048, "overwrite") + lb.ForceFlush() + } + addEntries(t, lb, 5, 2048, "overwrite") + + if string(before) != string(buf.Bytes()) { + t.Fatal("shared snapshot content changed after its window was recycled") + } +} + +// Race test: concurrent lagging readers against a writer that keeps sealing +// and recycling windows. Every delivered entry must unmarshal cleanly with the +// expected payload — corruption here means a reader saw recycled bytes. +func TestSharedSnapshotConcurrentIntegrity(t *testing.T) { + lb := NewLogBuffer("test", time.Hour, func(*LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, nil, nil) + defer lb.ShutdownLogBuffer() + + const payloadByte = 0x5A + payload := make([]byte, 8*1024) + for i := range payload { + payload[i] = payloadByte + } + + var stop sync.WaitGroup + done := make(chan struct{}) + stop.Add(1) + go func() { // writer: fill and seal aggressively so arrays recycle under the readers + defer stop.Done() + for i := 0; ; i++ { + select { + case <-done: + return + default: + } + if err := lb.AddDataToBuffer([]byte(fmt.Sprintf("/k/%d", i)), payload, time.Now().UnixNano()); err != nil { + t.Error(err) + return + } + if i%200 == 0 { + lb.ForceFlush() + } + } + }() + + var readers sync.WaitGroup + errCh := make(chan error, 16) + for r := 0; r < 8; r++ { + readers.Add(1) + go func(r int) { + defer readers.Done() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + start := NewMessagePosition(time.Now().Add(-500*time.Millisecond).UnixNano(), -2) + lb.LoopProcessLogData(fmt.Sprintf("r%d", r), start, 0, + func() bool { return time.Now().Before(deadline) }, + func(le *filer_pb.LogEntry) (bool, error) { + if len(le.Data) != len(payload) { + err := fmt.Errorf("payload length %d, want %d", len(le.Data), len(payload)) + select { + case errCh <- err: + default: + } + return true, err + } + for _, b := range le.Data { + if b != payloadByte { + err := fmt.Errorf("corrupted payload byte %x", b) + select { + case errCh <- err: + default: + } + return true, err + } + } + return false, nil + }) + } + }(r) + } + readers.Wait() + close(done) + stop.Wait() + + select { + case err := <-errCh: + t.Fatal(err) + default: + } +} + +// Current-window reads must share the prefix snapshot: two behind-readers get +// views of the same backing array, and the view content must stay intact while +// the writer keeps appending and eventually seals and recycles the window. +func TestCurrentWindowPrefixSharing(t *testing.T) { + lb := NewLogBuffer("test", time.Hour, func(*LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, nil, nil) + defer lb.ShutdownLogBuffer() + + addEntries(t, lb, 8, 1024, "cur") + + pos := NewMessagePosition(1, -2) // behind => full current-window read + buf1, _, pooled1, err1 := lb.ReadFromBuffer(pos) + buf2, _, pooled2, err2 := lb.ReadFromBuffer(pos) + if err1 != nil || err2 != nil || buf1 == nil || buf2 == nil { + t.Fatalf("reads: %v %v %v %v", buf1 != nil, err1, buf2 != nil, err2) + } + if pooled1 || pooled2 { + t.Fatalf("current-window behind-reads should be shared, got pooled %v %v", pooled1, pooled2) + } + b1, b2 := buf1.Bytes(), buf2.Bytes() + if len(b1) == 0 || len(b1) != len(b2) || &b1[0] != &b2[0] { + t.Fatalf("expected one shared prefix snapshot, lens %d %d", len(b1), len(b2)) + } + before := append([]byte(nil), b1...) + + // Keep appending: the prefix must be extended, not reallocated or mutated. + addEntries(t, lb, 8, 1024, "more") + buf3, _, _, err3 := lb.ReadFromBuffer(pos) + if err3 != nil || buf3 == nil { + t.Fatalf("read3: %v %v", buf3 != nil, err3) + } + if len(buf3.Bytes()) <= len(before) { + t.Fatalf("extended read %d should exceed first read %d", len(buf3.Bytes()), len(before)) + } + if string(buf3.Bytes()[:len(before)]) != string(before) { + t.Fatal("prefix changed when the snapshot was extended") + } + + // Seal + rotate the window fully out; earlier views must stay intact. + for i := 0; i < PreviousBufferCount+1; i++ { + lb.ForceFlush() + addEntries(t, lb, 4, 1024, "rotate") + } + if string(b1) != string(before) { + t.Fatal("shared current-window view changed after seal and recycle") + } +} + +// A sealed window whose prefix snapshot was fully extended must reuse it +// rather than re-copying on the first sealed read. +func TestSealHandsOffCompleteSnapshot(t *testing.T) { + lb := NewLogBuffer("test", time.Hour, func(*LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, nil, nil) + defer lb.ShutdownLogBuffer() + + addEntries(t, lb, 6, 512, "h") + pos := NewMessagePosition(1, -2) + buf1, _, _, err := lb.ReadFromBuffer(pos) // extends prefix snapshot to full window + if err != nil || buf1 == nil { + t.Fatalf("read1: %v %v", buf1 != nil, err) + } + lb.ForceFlush() + addEntries(t, lb, 1, 512, "next") + + buf2, _, pooled, err := lb.ReadFromBuffer(pos) // sealed read of the same window + if err != nil || buf2 == nil || pooled { + t.Fatalf("read2: buf=%v pooled=%v err=%v", buf2 != nil, pooled, err) + } + b1, b2 := buf1.Bytes(), buf2.Bytes() + if len(b1) == 0 || len(b1) > len(b2) || &b1[0] != &b2[0] { + t.Fatalf("sealed read should reuse the handed-off snapshot (lens %d %d)", len(b1), len(b2)) + } +} + +// Sanity: sealed-read content must byte-match what a proto round-trip expects. +func TestSharedSnapshotContentMatches(t *testing.T) { + lb := NewLogBuffer("test", time.Hour, func(*LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, nil, nil) + defer lb.ShutdownLogBuffer() + + addEntries(t, lb, 3, 512, "x") + lb.ForceFlush() + + buf, _, _, err := lb.ReadFromBuffer(NewMessagePosition(1, -2)) + if err != nil || buf == nil { + t.Fatalf("read: buf=%v err=%v", buf != nil, err) + } + data := buf.Bytes() + count := 0 + for pos := 0; pos+4 < len(data); { + size, _, err := readTs(data, pos) + if err != nil { + t.Fatalf("entry %d: %v", count, err) + } + var le filer_pb.LogEntry + if err := proto.Unmarshal(data[pos+4:pos+4+size], &le); err != nil { + t.Fatalf("entry %d unmarshal: %v", count, err) + } + if want := fmt.Sprintf("/x/%d", count); string(le.Key) != want { + t.Fatalf("entry %d key %q, want %q", count, le.Key, want) + } + pos += 4 + size + count++ + } + if count != 3 { + t.Fatalf("read %d entries, want 3", count) + } +}