From 240e95c444cb85c8ea0ce1859f9ab77a423040af Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Sun, 23 Aug 2026 16:57:55 +0700 Subject: [PATCH 1/8] db/etl: split the sortable buffer into pooled 1MB chunks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit sortableBuffer held every key/value byte in one slice, so a Collect that outgrew it re-allocated and copied everything collected so far - for a 256MB buffer that is ~256MB of memmove and ~512MB live at the peak. Store the bytes in 1MB chunks drawn from a shared sync.Pool: Put never copies what is already there, Reset hands the chunks back, and an idle pooled buffer no longer pins its busiest run's RAM. entryLoc.offset packs the chunk index with the offset inside the chunk; an entry never straddles a chunk, so Get still returns direct references. An entry larger than one chunk gets a private chunk of its own. BenchmarkCollect (no Prealloc, the path collectors actually take): 10k_largebuf 518.8µs -> 116.8µs -77% 8.7MB/op -> 0.7MB/op 100k_largebuf 3.286ms -> 1.137ms -65% 87MB/op -> 8.5MB/op Cost of the extra indirection, both on prealloc'd buffers that never grow: Put +11%, Sort +8..24%. Claude-Session: https://claude.ai/code/session_01WFkAYPPhqPe1NXg41Nph78 --- db/etl/buffers.go | 159 +++++++++++++++++++++++++++++++++++--------- db/etl/collector.go | 4 +- db/etl/etl_test.go | 86 +++++++++++++++++++++++- 3 files changed, 210 insertions(+), 39 deletions(-) diff --git a/db/etl/buffers.go b/db/etl/buffers.go index ef37fc0b7f7..2559818ab13 100644 --- a/db/etl/buffers.go +++ b/db/etl/buffers.go @@ -26,6 +26,7 @@ import ( "sort" "strconv" "sync" + "unsafe" "github.com/c2h5oh/datasize" @@ -79,7 +80,11 @@ func writeSortedEntries(w io.Writer, entries []sortableBufferEntry) error { var BufferOptimalSize = dbg.EnvDataSize("ETL_OPTIMAL", 256*datasize.MB) /* var because we want to sometimes change it from tests or command-line flags */ -// 3_domains * 2 + 3_history * 1 + 4_indices * 2 = 17 etl collectors, 17*(256Mb/8) = 512Mb - for all collectros +// etlSmallBufRAM (BufferOptimalSize/8) bounds the flush threshold so a full +// set of domain/history/index flush collectors (~17 per batch writer) stays +// around 512 MB when all run full. Pooled buffers start empty and take chunks +// as they fill; Reset returns those chunks to the shared dataChunks pool, so an +// idle buffer doesn't pin the RAM its busiest run needed. var etlSmallBufRAM = dbg.EnvDataSize("ETL_SMALL", BufferOptimalSize/8) var SmallSortableBuffers = NewAllocator(&sync.Pool{ New: func() any { @@ -93,6 +98,31 @@ var LargeSortableBuffers = NewAllocator(&sync.Pool{ }, }) +const ( + // sortableBuffer stores key/value bytes in dataChunkSize blocks. entryLoc.offset + // packs the chunk index and the offset inside the chunk, so the index range is + // what limits one buffer to maxDataChunks. + dataChunkBits = 20 + dataChunkSize = 1 << dataChunkBits + maxDataChunks = math.MaxInt32>>dataChunkBits + 1 +) + +// dataChunks are shared by all sortableBuffer instances: a buffer takes chunks as +// it fills and gives them back on Reset, instead of pinning its peak size forever. +var dataChunks = sync.Pool{New: func() any { + c := make([]byte, dataChunkSize) + return &c +}} + +func getDataChunk() []byte { return *dataChunks.Get().(*[]byte) } + +func putDataChunk(c []byte) { + if len(c) != dataChunkSize { // private chunk of an oversized entry + return + } + dataChunks.Put(&c) +} + type Buffer interface { // Put does copy `k` and `v` Put(k, v []byte) @@ -119,9 +149,10 @@ var ( _ Buffer = &oldestEntrySortableBuffer{} ) -// entryLoc stores the location of a key/value pair within sortableBuffer.data. -// Key occupies data[offset : offset+keyLen], value follows at data[offset+max(0,keyLen) : ...+valLen]. -// keyLen/valLen of -1 indicates nil. +// entryLoc stores the location of a key/value pair inside sortableBuffer. +// offset packs the chunk index and the offset inside that chunk: +// idx<= maxDataChunks { + panic(fmt.Sprintf("etl: sortableBuffer exceeded %d chunks", maxDataChunks)) + } + if n > dataChunkSize { + b.cur = make([]byte, n) + } else { + b.cur = getDataChunk() + } + b.chunks = append(b.chunks, b.cur) + b.bases = append(b.bases, unsafe.Pointer(&b.cur[0])) + b.curBase = int32(len(b.chunks)-1) << dataChunkBits //nolint:gosec + b.curOff = 0 + b.chunkBytes += len(b.cur) +} + +// entryData points at e's first byte: the key, immediately followed by the value. +// bases[i] is chunks[i]'s first byte - one load per lookup instead of a slice +// header, which the sort comparator does twice per comparison. +func (b *sortableBuffer) entryData(e *entryLoc) unsafe.Pointer { + return unsafe.Add(b.bases[e.offset>>dataChunkBits], e.offset&(dataChunkSize-1)) +} + // Put adds key and value to the buffer. These slices will not be accessed later, // so no copying is necessary func (b *sortableBuffer) Put(k, v []byte) { e := entryLoc{ - offset: int32(len(b.data)), //nolint:gosec keyLen: int32(len(k)), //nolint:gosec valLen: int32(len(v)), //nolint:gosec insertionOrder: int32(len(b.entries)), //nolint:gosec @@ -159,11 +223,26 @@ func (b *sortableBuffer) Put(k, v []byte) { if v == nil { e.valLen = -1 } + if n := len(k) + len(v); n > 0 { + off := b.curOff + if int(off)+n > len(b.cur) { + b.nextChunk(n) + off = 0 + } + data := b.cur[off:] + copy(data, k) + copy(data[len(k):], v) + e.offset = b.curBase | off + b.curOff = off + int32(n) //nolint:gosec + } b.entries = append(b.entries, e) - b.data = append(append(b.data, k...), v...) } -func (b *sortableBuffer) Size() int { return len(b.data) + len(b.entries)*entryLocSize } +// Size counts the bytes of every chunk taken so far, minus the unused tail of +// the chunk being filled - so it tracks RAM held, not just bytes stored. +func (b *sortableBuffer) Size() int { + return b.chunkBytes - (len(b.cur) - int(b.curOff)) + len(b.entries)*entryLocSize +} func (b *sortableBuffer) Len() int { return len(b.entries) @@ -172,22 +251,24 @@ func (b *sortableBuffer) Len() int { func (b *sortableBuffer) Get(i int) ([]byte, []byte) { e := &b.entries[i] kLen, vLen := int(e.keyLen), int(e.valLen) - keyOffset := int(e.offset) - valOffset := keyOffset - if kLen > 0 { - valOffset += kLen - } var key, val []byte - if kLen > 0 { - key = b.data[keyOffset : keyOffset+kLen] - } else if kLen == 0 { + if kLen == 0 { key = []byte{} } - if vLen > 0 { - val = b.data[valOffset : valOffset+vLen] - } else if vLen == 0 { + if vLen == 0 { val = []byte{} } + if kLen <= 0 && vLen <= 0 { + return key, val + } + p := b.entryData(e) + if kLen > 0 { + key = unsafe.Slice((*byte)(p), kLen) + p = unsafe.Add(p, kLen) + } + if vLen > 0 { + val = unsafe.Slice((*byte)(p), vLen) + } return key, val } @@ -195,23 +276,35 @@ func (b *sortableBuffer) Prealloc(predictKeysAmount, predictDataSize int) Buffer if cap(b.entries) < predictKeysAmount { b.entries = make([]entryLoc, 0, predictKeysAmount) } - if cap(b.data) < predictDataSize { - b.data = make([]byte, 0, predictDataSize) + if n := predictDataSize/dataChunkSize + 1; cap(b.chunks) < n { + b.chunks = slices.Grow(b.chunks, n) + b.bases = slices.Grow(b.bases, n) } return b } func (b *sortableBuffer) Reset() { b.entries = b.entries[:0] - b.data = b.data[:0] + for i, c := range b.chunks { + putDataChunk(c) + b.chunks[i], b.bases[i] = nil, nil + } + b.chunks, b.bases = b.chunks[:0], b.bases[:0] + b.cur, b.curBase, b.curOff = nil, 0, 0 + b.chunkBytes = 0 } func (b *sortableBuffer) SizeLimit() int { return b.optimalSize } func (b *sortableBuffer) Sort() { - data := b.data + bases := b.bases + key := func(e entryLoc) []byte { + if e.keyLen <= 0 { + return nil + } + p := unsafe.Add(bases[e.offset>>dataChunkBits], e.offset&(dataChunkSize-1)) + return unsafe.Slice((*byte)(p), e.keyLen) + } cmp := func(a, b entryLoc) int { - aKey := data[a.offset : a.offset+max(a.keyLen, 0)] - bKey := data[b.offset : b.offset+max(b.keyLen, 0)] - if c := bytes.Compare(aKey, bKey); c != 0 { + if c := bytes.Compare(key(a), key(b)); c != 0 { return c } return int(a.insertionOrder - b.insertionOrder) // StableSort: preserve insertion order for duplicate keys @@ -231,10 +324,9 @@ func (b *sortableBuffer) Write(w io.Writer) error { for i := range b.entries { e := &b.entries[i] kLen, vLen := int(e.keyLen), int(e.valLen) - keyOffset := int(e.offset) - valOffset := keyOffset - if kLen > 0 { - valOffset += kLen + var p unsafe.Pointer + if kLen > 0 || vLen > 0 { + p = b.entryData(e) } // write key n := binary.PutVarint(numBuf[:], int64(e.keyLen)) @@ -242,9 +334,10 @@ func (b *sortableBuffer) Write(w io.Writer) error { return err } if kLen > 0 { - if _, err := w.Write(b.data[keyOffset : keyOffset+kLen]); err != nil { + if _, err := w.Write(unsafe.Slice((*byte)(p), kLen)); err != nil { return err } + p = unsafe.Add(p, kLen) } // write value n = binary.PutVarint(numBuf[:], int64(e.valLen)) @@ -252,7 +345,7 @@ func (b *sortableBuffer) Write(w io.Writer) error { return err } if vLen > 0 { - if _, err := w.Write(b.data[valOffset : valOffset+vLen]); err != nil { + if _, err := w.Write(unsafe.Slice((*byte)(p), vLen)); err != nil { return err } } diff --git a/db/etl/collector.go b/db/etl/collector.go index da677a69edf..59b28d50eeb 100644 --- a/db/etl/collector.go +++ b/db/etl/collector.go @@ -45,9 +45,7 @@ func (a *Allocator) Put(b Buffer) { if b == nil { return } - //if cast, ok := b.(*sortableBuffer); ok { - // log.Warn("[dbg] return buf", "cap(cast.data)", cap(cast.data), "cap(cast.lens)", cap(cast.lens)) - //} + b.Reset() // release the data chunks now: an idle pooled buffer must not pin them a.p.Put(b) } func (a *Allocator) Get() Buffer { diff --git a/db/etl/etl_test.go b/db/etl/etl_test.go index 8b7faac5a4e..393c859ee5b 100644 --- a/db/etl/etl_test.go +++ b/db/etl/etl_test.go @@ -488,10 +488,10 @@ func TestReuseCollectorAfterLoad(t *testing.T) { require.Equal(t, 1, see) c.Close() - // buffers are not lost - require.Empty(t, buf.data) + // buffers are not lost: entries keep their cap, data chunks went back to the pool + require.Empty(t, buf.chunks) require.Empty(t, buf.entries) - require.NotZero(t, cap(buf.data)) + require.Zero(t, buf.Size()) require.NotZero(t, cap(buf.entries)) // teset that no data visible @@ -1532,3 +1532,83 @@ func TestVmtouchMmap(t *testing.T) { } vmtouch("AFTER full scan") } + +// TestSortableBufferChunks pins the chunked layout: key/value bytes live in +// fixed-size chunks, so a growing buffer never re-allocates and copies the +// bytes it already holds. +func TestSortableBufferChunks(t *testing.T) { + buf := NewSortableBuffer(256 * datasize.MB) + + const entries = 512 + val := bytes.Repeat([]byte{0xAB}, 16*1024) // 512*16KB = 8MB of values + key := make([]byte, 8) + for i := range entries { + binary.BigEndian.PutUint64(key, uint64(i)) + buf.Put(key, val) + } + + require.Equal(t, entries, buf.Len()) + require.Greater(t, len(buf.chunks), 1, "data must be split into chunks") + for i, c := range buf.chunks { + require.Equal(t, dataChunkSize, cap(c), "chunk %d", i) + } + + for i := range entries { + binary.BigEndian.PutUint64(key, uint64(i)) + k, v := buf.Get(i) + require.Equal(t, key, k, "entry %d", i) + require.Equal(t, val, v, "entry %d", i) + } +} + +// TestSortableBufferOversizedEntry: an entry bigger than one chunk gets a chunk +// of its own - Get must still return one contiguous slice per key and value. +func TestSortableBufferOversizedEntry(t *testing.T) { + buf := NewSortableBuffer(256 * datasize.MB) + + big := bytes.Repeat([]byte{0xCD}, dataChunkSize+7) + buf.Put([]byte{0x01}, []byte("small")) + buf.Put([]byte{0x02}, big) + buf.Put([]byte{0x03}, []byte("after")) + + k, v := buf.Get(1) + require.Equal(t, []byte{0x02}, k) + require.Equal(t, big, v) + k, v = buf.Get(2) + require.Equal(t, []byte{0x03}, k) + require.Equal(t, []byte("after"), v) + + w := bytes.NewBuffer(nil) + require.NoError(t, buf.Write(w)) + m := &mmapBytesReader{data: w.Bytes()} + for i := range buf.Len() { + wantK, wantV := buf.Get(i) + gotK, err := readField(m) + require.NoError(t, err) + gotV, err := readField(m) + require.NoError(t, err) + require.Equal(t, wantK, gotK) + require.Equal(t, wantV, gotV) + } +} + +// TestSortableBufferResetReleasesChunks: Reset hands the chunks back to the +// shared pool, so an idle pooled buffer doesn't pin the RAM it once needed. +func TestSortableBufferResetReleasesChunks(t *testing.T) { + buf := NewSortableBuffer(256 * datasize.MB) + val := bytes.Repeat([]byte{0xEF}, 16*1024) + for i := range 512 { + buf.Put(binary.BigEndian.AppendUint64(nil, uint64(i)), val) + } + require.NotEmpty(t, buf.chunks) + + buf.Reset() + require.Empty(t, buf.chunks) + require.Zero(t, buf.Size()) + require.Zero(t, buf.Len()) + + buf.Put([]byte{0x01}, []byte("reused")) + k, v := buf.Get(0) + require.Equal(t, []byte{0x01}, k) + require.Equal(t, []byte("reused"), v) +} From 3aa83cdc2f3bdb96f7d61bc2550a62ae3384c52c Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Sun, 23 Aug 2026 17:03:11 +0700 Subject: [PATCH 2/8] db/etl: say what Size counts Claude-Session: https://claude.ai/code/session_01WFkAYPPhqPe1NXg41Nph78 --- db/etl/buffers.go | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/db/etl/buffers.go b/db/etl/buffers.go index 2559818ab13..a6057e1dacb 100644 --- a/db/etl/buffers.go +++ b/db/etl/buffers.go @@ -238,8 +238,7 @@ func (b *sortableBuffer) Put(k, v []byte) { b.entries = append(b.entries, e) } -// Size counts the bytes of every chunk taken so far, minus the unused tail of -// the chunk being filled - so it tracks RAM held, not just bytes stored. +// Size counts the stored bytes plus the tails wasted by the chunks already filled. func (b *sortableBuffer) Size() int { return b.chunkBytes - (len(b.cur) - int(b.curOff)) + len(b.entries)*entryLocSize } From 82d7403ce5941050b85aa76287fca0e3283423e8 Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Sun, 23 Aug 2026 17:20:01 +0700 Subject: [PATCH 3/8] db/etl: dispose providers before recycling the buffer Close released the buffer first, and a KeepInRAM provider reads straight from it. That was harmless while Reset only truncated the buffer's own slice; now Reset hands the chunks to a pool other collectors draw from, so any read after Close would race a different goroutine's writes. Claude-Session: https://claude.ai/code/session_01WFkAYPPhqPe1NXg41Nph78 --- db/etl/collector.go | 14 ++++++++------ db/etl/etl_test.go | 28 ++++++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 6 deletions(-) diff --git a/db/etl/collector.go b/db/etl/collector.go index 59b28d50eeb..bdb18e0cbc2 100644 --- a/db/etl/collector.go +++ b/db/etl/collector.go @@ -259,6 +259,14 @@ func (c *Collector) Load(db kv.RwTx, toBucket string, loadFunc LoadFunc, args Tr } func (c *Collector) Close() { + // Providers first: a KeepInRAM one reads straight from `buf`, whose chunks + // Reset hands to a pool that other collectors draw from. + if c.dataProviders != nil { //idempotency + for _, p := range c.dataProviders { + p.Dispose() + } + c.dataProviders = nil + } if c.buf != nil { //idempotency if c.allocator != nil { c.allocator.Put(c.buf) @@ -267,12 +275,6 @@ func (c *Collector) Close() { c.buf.Reset() } } - if c.dataProviders != nil { //idempotency - for _, p := range c.dataProviders { - p.Dispose() - } - c.dataProviders = nil - } c.allFlushed = false } diff --git a/db/etl/etl_test.go b/db/etl/etl_test.go index 393c859ee5b..c691acd38cd 100644 --- a/db/etl/etl_test.go +++ b/db/etl/etl_test.go @@ -1612,3 +1612,31 @@ func TestSortableBufferResetReleasesChunks(t *testing.T) { require.Equal(t, []byte{0x01}, k) require.Equal(t, []byte("reused"), v) } + +// disposeProbe records whether the collector still owned its data chunks when +// the provider was disposed. +type disposeProbe struct { + buf *sortableBuffer + sawOwnChunks bool +} + +func (p *disposeProbe) Next() ([]byte, []byte, error) { return nil, nil, io.EOF } +func (p *disposeProbe) Wait() error { return nil } +func (p *disposeProbe) String() string { return "disposeProbe" } +func (p *disposeProbe) Dispose() { p.sawOwnChunks = len(p.buf.chunks) > 0 } + +// TestCloseDisposesProvidersBeforeBuffer: KeepInRAM hands out a provider backed +// by the collector's own buffer, and Reset gives that buffer's chunks to a pool +// other collectors draw from. So Close must be done with every provider before +// it recycles the buffer. +func TestCloseDisposesProvidersBeforeBuffer(t *testing.T) { + allocator := NewAllocator(&sync.Pool{New: func() any { return NewSortableBuffer(BufferOptimalSize) }}) + c := NewCollectorWithAllocator(t.Name(), t.TempDir(), allocator, log.New()) + require.NoError(t, c.Collect([]byte{1}, []byte{2})) + + probe := &disposeProbe{buf: c.buf.(*sortableBuffer)} + c.dataProviders = append(c.dataProviders, probe) + c.Close() + + require.True(t, probe.sawOwnChunks, "buffer was recycled before its providers were disposed") +} From 82f5199a32c3c157b8ea41c82c9abf1f6abca9c6 Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Sun, 23 Aug 2026 17:29:07 +0700 Subject: [PATCH 4/8] db/etl: drop the unsafe base-pointer table Look up chunk bytes through the chunks slice itself. The parallel []unsafe.Pointer bought ~12 points of Sort geomean by saving a slice-header load per key, but it duplicated ownership of every chunk to buy back part of a cost the chunking itself creates. Claude-Session: https://claude.ai/code/session_01WFkAYPPhqPe1NXg41Nph78 --- db/etl/buffers.go | 40 +++++++++++++++++----------------------- 1 file changed, 17 insertions(+), 23 deletions(-) diff --git a/db/etl/buffers.go b/db/etl/buffers.go index a6057e1dacb..d73b3bdcff6 100644 --- a/db/etl/buffers.go +++ b/db/etl/buffers.go @@ -26,7 +26,6 @@ import ( "sort" "strconv" "sync" - "unsafe" "github.com/c2h5oh/datasize" @@ -176,7 +175,6 @@ type sortableBuffer struct { // far. All chunks are dataChunkSize, except the private chunk an entry // larger than that gets. cur is the chunk being filled. chunks [][]byte - bases []unsafe.Pointer cur []byte curBase int32 // packed location of cur's first byte: curIdx<>dataChunkBits], e.offset&(dataChunkSize-1)) +// entryData returns e's bytes: the key, immediately followed by the value. +func (b *sortableBuffer) entryData(e *entryLoc) []byte { + return b.chunks[e.offset>>dataChunkBits][e.offset&(dataChunkSize-1):] } // Put adds key and value to the buffer. These slices will not be accessed later, @@ -260,13 +255,13 @@ func (b *sortableBuffer) Get(i int) ([]byte, []byte) { if kLen <= 0 && vLen <= 0 { return key, val } - p := b.entryData(e) + data := b.entryData(e) if kLen > 0 { - key = unsafe.Slice((*byte)(p), kLen) - p = unsafe.Add(p, kLen) + key = data[:kLen:kLen] + data = data[kLen:] } if vLen > 0 { - val = unsafe.Slice((*byte)(p), vLen) + val = data[:vLen:vLen] } return key, val } @@ -277,7 +272,6 @@ func (b *sortableBuffer) Prealloc(predictKeysAmount, predictDataSize int) Buffer } if n := predictDataSize/dataChunkSize + 1; cap(b.chunks) < n { b.chunks = slices.Grow(b.chunks, n) - b.bases = slices.Grow(b.bases, n) } return b } @@ -286,21 +280,21 @@ func (b *sortableBuffer) Reset() { b.entries = b.entries[:0] for i, c := range b.chunks { putDataChunk(c) - b.chunks[i], b.bases[i] = nil, nil + b.chunks[i] = nil } - b.chunks, b.bases = b.chunks[:0], b.bases[:0] + b.chunks = b.chunks[:0] b.cur, b.curBase, b.curOff = nil, 0, 0 b.chunkBytes = 0 } func (b *sortableBuffer) SizeLimit() int { return b.optimalSize } func (b *sortableBuffer) Sort() { - bases := b.bases + chunks := b.chunks key := func(e entryLoc) []byte { if e.keyLen <= 0 { return nil } - p := unsafe.Add(bases[e.offset>>dataChunkBits], e.offset&(dataChunkSize-1)) - return unsafe.Slice((*byte)(p), e.keyLen) + off := e.offset & (dataChunkSize - 1) + return chunks[e.offset>>dataChunkBits][off : off+e.keyLen] } cmp := func(a, b entryLoc) int { if c := bytes.Compare(key(a), key(b)); c != 0 { @@ -323,9 +317,9 @@ func (b *sortableBuffer) Write(w io.Writer) error { for i := range b.entries { e := &b.entries[i] kLen, vLen := int(e.keyLen), int(e.valLen) - var p unsafe.Pointer + var data []byte if kLen > 0 || vLen > 0 { - p = b.entryData(e) + data = b.entryData(e) } // write key n := binary.PutVarint(numBuf[:], int64(e.keyLen)) @@ -333,10 +327,10 @@ func (b *sortableBuffer) Write(w io.Writer) error { return err } if kLen > 0 { - if _, err := w.Write(unsafe.Slice((*byte)(p), kLen)); err != nil { + if _, err := w.Write(data[:kLen]); err != nil { return err } - p = unsafe.Add(p, kLen) + data = data[kLen:] } // write value n = binary.PutVarint(numBuf[:], int64(e.valLen)) @@ -344,7 +338,7 @@ func (b *sortableBuffer) Write(w io.Writer) error { return err } if vLen > 0 { - if _, err := w.Write(unsafe.Slice((*byte)(p), vLen)); err != nil { + if _, err := w.Write(data[:vLen]); err != nil { return err } } From 33f173688eea3c21a533937a86479e68bbda142a Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Sun, 23 Aug 2026 17:36:37 +0700 Subject: [PATCH 5/8] db/etl: say what the chunk constants mean 20 and 1<<20 read as magic: the power of two is what lets entryLoc.offset pack the chunk index with the in-chunk offset, and the index range is what caps a buffer at 2GB. Claude-Session: https://claude.ai/code/session_01WFkAYPPhqPe1NXg41Nph78 --- db/etl/buffers.go | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/db/etl/buffers.go b/db/etl/buffers.go index d73b3bdcff6..bc6f1aaea0b 100644 --- a/db/etl/buffers.go +++ b/db/etl/buffers.go @@ -98,11 +98,15 @@ var LargeSortableBuffers = NewAllocator(&sync.Pool{ }) const ( - // sortableBuffer stores key/value bytes in dataChunkSize blocks. entryLoc.offset - // packs the chunk index and the offset inside the chunk, so the index range is - // what limits one buffer to maxDataChunks. + // sortableBuffer stores key/value bytes in chunks of a power-of-two size, so + // entryLoc.offset can pack the chunk index with the offset inside the chunk + // and splitting the two is a shift and a mask. 1MB is also the least a + // collector can hold once it takes a chunk at all. dataChunkBits = 20 - dataChunkSize = 1 << dataChunkBits + dataChunkSize = 1 << dataChunkBits // 1MB + + // The chunk index takes what is left of a positive int32, so one buffer + // addresses 2GB - the ceiling NewSortableBuffer already puts on optimalSize. maxDataChunks = math.MaxInt32>>dataChunkBits + 1 ) From 0d693ab5ca2c038924ecbb3bbc3d8e4725472d23 Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Sun, 23 Aug 2026 17:58:37 +0700 Subject: [PATCH 6/8] save --- db/etl/buffers.go | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/db/etl/buffers.go b/db/etl/buffers.go index bc6f1aaea0b..40d139f1728 100644 --- a/db/etl/buffers.go +++ b/db/etl/buffers.go @@ -33,15 +33,15 @@ import ( ) const ( - //SliceBuffer - just simple slice w + // SliceBuffer - just simple slice w SortableSliceBuffer = iota - //SortableAppendBuffer - map[k] [v1 v2 v3] + // SortableAppendBuffer - map[k] [v1 v2 v3] SortableAppendBuffer // SortableOldestAppearedBuffer - buffer that keeps only the oldest entries. // if first v1 was added under key K, then v2; only v1 will stay SortableOldestAppearedBuffer - //BufIOSize - 128 pages | default is 1 page | increasing over `64 * 4096` doesn't show speedup on SSD/NVMe, but show speedup in cloud drives + // BufIOSize - 128 pages | default is 1 page | increasing over `64 * 4096` doesn't show speedup on SSD/NVMe, but show speedup in cloud drives BufIOSize = 128 * 4096 entryLocSize = 16 // sizeof(entryLoc): insertionOrder(4) + offset(4) + keyLen(4) + valLen(4) @@ -87,7 +87,7 @@ var BufferOptimalSize = dbg.EnvDataSize("ETL_OPTIMAL", 256*datasize.MB) /* var var etlSmallBufRAM = dbg.EnvDataSize("ETL_SMALL", BufferOptimalSize/8) var SmallSortableBuffers = NewAllocator(&sync.Pool{ New: func() any { - return NewSortableBuffer(etlSmallBufRAM).Prealloc(1_024, int(etlSmallBufRAM/32)) + return NewSortableBuffer(etlSmallBufRAM).Prealloc(int(etlSmallBufRAM/512), int(etlSmallBufRAM)) // SortableBuffer does Prealloc only metadata slices - not buffers itself }, }) var etlLargeBufRAM = BufferOptimalSize @@ -380,6 +380,7 @@ func (b *appendSortableBuffer) SizeLimit() int { return b.optimalSize } func (b *appendSortableBuffer) Len() int { return len(b.entries) } + func (b *appendSortableBuffer) Sort() { b.sortedBuf = b.sortedBuf[:0] if cap(b.sortedBuf) < len(b.entries) { @@ -402,11 +403,13 @@ func (b *appendSortableBuffer) Swap(i, j int) { func (b *appendSortableBuffer) Get(i int) ([]byte, []byte) { return b.sortedBuf[i].key, b.sortedBuf[i].value } + func (b *appendSortableBuffer) Reset() { b.sortedBuf = nil b.entries = make(map[string][]byte) b.size = 0 } + func (b *appendSortableBuffer) Prealloc(predictKeysAmount, predictDataSize int) Buffer { b.entries = make(map[string][]byte, predictKeysAmount) // maps have no cap(), always recreate if cap(b.sortedBuf) < predictKeysAmount { @@ -478,11 +481,13 @@ func (b *oldestEntrySortableBuffer) Swap(i, j int) { func (b *oldestEntrySortableBuffer) Get(i int) ([]byte, []byte) { return b.sortedBuf[i].key, b.sortedBuf[i].value } + func (b *oldestEntrySortableBuffer) Reset() { b.sortedBuf = nil b.entries = make(map[string][]byte) b.size = 0 } + func (b *oldestEntrySortableBuffer) Prealloc(predictKeysAmount, predictDataSize int) Buffer { b.entries = make(map[string][]byte, predictKeysAmount) // maps have no cap(), always recreate if cap(b.sortedBuf) < predictKeysAmount { @@ -494,6 +499,7 @@ func (b *oldestEntrySortableBuffer) Prealloc(predictKeysAmount, predictDataSize func (b *oldestEntrySortableBuffer) Write(w io.Writer) error { return writeSortedEntries(w, b.sortedBuf) } + func (b *oldestEntrySortableBuffer) CheckFlushSize() bool { return b.size >= b.optimalSize } From 3fa807519c88218baff97968ccaf3231990441a1 Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Sun, 23 Aug 2026 18:05:10 +0700 Subject: [PATCH 7/8] db/etl: import sync in the etl test file release/3.6 does not carry the lazy-draw test that pulled the import in on main, and TestCloseDisposesProvidersBeforeBuffer builds its own pool. Claude-Session: https://claude.ai/code/session_01WFkAYPPhqPe1NXg41Nph78 --- db/etl/etl_test.go | 1 + 1 file changed, 1 insertion(+) diff --git a/db/etl/etl_test.go b/db/etl/etl_test.go index c691acd38cd..d72cfdb80f1 100644 --- a/db/etl/etl_test.go +++ b/db/etl/etl_test.go @@ -29,6 +29,7 @@ import ( "path/filepath" "slices" "strings" + "sync" "testing" "github.com/c2h5oh/datasize" From aa7cb3696497dae226f140a4bae0137d7b38c92e Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Mon, 24 Aug 2026 09:23:54 +0700 Subject: [PATCH 8/8] save --- db/etl/buffers.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/db/etl/buffers.go b/db/etl/buffers.go index 40d139f1728..972b7c778cc 100644 --- a/db/etl/buffers.go +++ b/db/etl/buffers.go @@ -87,7 +87,7 @@ var BufferOptimalSize = dbg.EnvDataSize("ETL_OPTIMAL", 256*datasize.MB) /* var var etlSmallBufRAM = dbg.EnvDataSize("ETL_SMALL", BufferOptimalSize/8) var SmallSortableBuffers = NewAllocator(&sync.Pool{ New: func() any { - return NewSortableBuffer(etlSmallBufRAM).Prealloc(int(etlSmallBufRAM/512), int(etlSmallBufRAM)) // SortableBuffer does Prealloc only metadata slices - not buffers itself + return NewSortableBuffer(etlSmallBufRAM) }, }) var etlLargeBufRAM = BufferOptimalSize