You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2026/04/22 13:30:30 UTC
(skywalking-banyandb) branch main updated: feat(trace): introduce fair fast/slow lane scheduling for part merges (#1086)
This is an automated email from the ASF dual-hosted git repository.
wu-sheng pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new b5e54b1ae feat(trace): introduce fair fast/slow lane scheduling for part merges (#1086)
b5e54b1ae is described below
commit b5e54b1aeb56afeccc10d442bb1605305dc8281f
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Wed Apr 22 21:30:24 2026 +0800
feat(trace): introduce fair fast/slow lane scheduling for part merges (#1086)
---
CHANGES.md | 1 +
banyand/trace/flusher.go | 2 +-
banyand/trace/merger.go | 259 +++++++++++++++++++++++++++++++++++--------
banyand/trace/merger_test.go | 254 +++++++++++++++++++++++++++++++++++++++++-
banyand/trace/metrics.go | 57 ++++++----
banyand/trace/tstable.go | 6 +-
6 files changed, 507 insertions(+), 72 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 21a3ba955..4ed286561 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -13,6 +13,7 @@ Release Notes.
- Add `noDuplicates` verification to all e2e expected files to detect duplicate data in query results.
- Add periodic health check for property schema connection.
- Persist segment end time in per-segment metadata so boundaries don't shift across restarts or config changes.
+- Introduce fair fast/slow lane scheduling for trace part merges to prevent short merges from being blocked by long-running merges; expose queue wait time as `total_merge_queue_latency`.
- [Breaking Change] Remove etcd components. The property-based schema registry is now the only supported mode.
- All `--etcd-*` CLI flags have been removed.
- The `--namespace` CLI flag has been removed (it previously configured the etcd key prefix).
diff --git a/banyand/trace/flusher.go b/banyand/trace/flusher.go
index 8ebe7baf0..e0cf41ba5 100644
--- a/banyand/trace/flusher.go
+++ b/banyand/trace/flusher.go
@@ -149,7 +149,7 @@ func (tst *tsTable) mergeMemParts(snp *snapshot, mergeCh chan *mergerIntroductio
// merge memory must not be closed by the tsTable.close
closeCh := make(chan struct{})
newPart, err := tst.mergePartsThenSendIntroduction(snapshotCreatorMergedFlusher, memParts,
- currentMergedIDs, mergeCh, closeCh, mergeTypeMem)
+ currentMergedIDs, mergeCh, closeCh, mergeTypeMem, "")
close(closeCh)
if err != nil {
if errors.Is(err, errClosed) {
diff --git a/banyand/trace/merger.go b/banyand/trace/merger.go
index 8e041f572..7e1bba3e7 100644
--- a/banyand/trace/merger.go
+++ b/banyand/trace/merger.go
@@ -21,6 +21,7 @@ import (
"errors"
"fmt"
"path/filepath"
+ "sync"
"sync/atomic"
"time"
@@ -41,76 +42,234 @@ var (
mergeTypeFile = "file"
)
+var (
+ mergeLaneFast = "fast"
+ mergeLaneSlow = "slow"
+)
+
+const defaultSmallMergeThreshold = 32 << 20 // 32MB fallback
+
+func computeSmallMergeThreshold() uint64 {
+ memLimit, err := cgroups.MemoryLimit()
+ if err != nil || memLimit <= 0 {
+ return defaultSmallMergeThreshold
+ }
+ threshold := uint64(memLimit) / 16
+ if threshold < defaultSmallMergeThreshold {
+ return defaultSmallMergeThreshold
+ }
+ return threshold
+}
+
+type mergeDispatchRequest struct {
+ enqueuedAt time.Time
+ toBeMerged map[uint64]struct{}
+ typ string
+ lane string
+ parts []*partWrapper
+}
+
func (tst *tsTable) mergeLoop(merges chan *mergerIntroduction, flusherNotifier watcher.Channel) {
defer tst.loopCloser.Done()
- var epoch uint64
+ var lastProcessedEpoch uint64
ew := flusherNotifier.Add(0, tst.loopCloser.CloseNotify())
if ew == nil {
return
}
- var pwsChunk []*partWrapper
+ threshold := computeSmallMergeThreshold()
+ fastWorkers := max(1, cgroups.CPUs()/2)
+ fastCh := make(chan *mergeDispatchRequest, fastWorkers)
+ slowCh := make(chan *mergeDispatchRequest, 1)
+ triggerCh := make(chan struct{}, 1)
+
+ var workersWg, dispatcherWg sync.WaitGroup
+
+ for i := 0; i < fastWorkers; i++ {
+ workersWg.Add(1)
+ go func() {
+ defer workersWg.Done()
+ tst.mergeLaneWorker(fastCh, merges)
+ }()
+ }
+ workersWg.Add(1)
+ go func() {
+ defer workersWg.Done()
+ tst.mergeLaneWorker(slowCh, merges)
+ }()
+
+ dispatcherWg.Add(1)
+ go func() {
+ defer dispatcherWg.Done()
+ tst.dispatcherLoop(triggerCh, threshold, fastCh, slowCh)
+ }()
+
+ // Shutdown order: stop dispatcher first so no new work enters the lane
+ // channels, then close the lane channels so idle workers exit their range
+ // loops, then wait for workers to drain any in-flight merges.
+ defer func() {
+ close(triggerCh)
+ dispatcherWg.Wait()
+ close(fastCh)
+ close(slowCh)
+ workersWg.Wait()
+ }()
for {
select {
case <-tst.loopCloser.CloseNotify():
return
case <-ew.Watch():
- if func() bool {
- curSnapshot := tst.currentSnapshot()
- if curSnapshot == nil {
- return false
- }
- defer curSnapshot.decRef()
- if curSnapshot.epoch != epoch {
+ if curSnapshot := tst.currentSnapshot(); curSnapshot != nil {
+ if curSnapshot.epoch > lastProcessedEpoch {
select {
- case mergeMaxConcurrencyCh <- struct{}{}:
- defer func() {
- <-mergeMaxConcurrencyCh
- }()
- case <-tst.loopCloser.CloseNotify():
- return true
- }
- tst.incTotalMergeLoopStarted(1)
- defer tst.incTotalMergeLoopFinished(1)
- var err error
- if pwsChunk, err = tst.mergeSnapshot(curSnapshot, merges, pwsChunk[:0]); err != nil {
- if errors.Is(err, errClosed) {
- return true
- }
- tst.l.Logger.Warn().Err(err).Msgf("cannot merge snapshot: %d", curSnapshot.epoch)
- tst.incTotalMergeLoopErr(1)
- return false
+ case triggerCh <- struct{}{}:
+ default:
}
- epoch = curSnapshot.epoch
+ lastProcessedEpoch = curSnapshot.epoch
}
- ew = flusherNotifier.Add(epoch, tst.loopCloser.CloseNotify())
- return ew == nil
- }() {
+ curSnapshot.decRef()
+ }
+ ew = flusherNotifier.Add(lastProcessedEpoch, tst.loopCloser.CloseNotify())
+ if ew == nil {
return
}
}
}
}
-func (tst *tsTable) mergeSnapshot(curSnapshot *snapshot, merges chan *mergerIntroduction, dst []*partWrapper) ([]*partWrapper, error) {
- freeDiskSize := tst.freeDiskSpace(tst.root)
- var toBeMerged map[uint64]struct{}
- dst, toBeMerged = tst.getPartsToMerge(curSnapshot, freeDiskSize, dst)
- if len(dst) < 2 {
- return nil, nil
+func (tst *tsTable) dispatcherLoop(triggerCh chan struct{}, threshold uint64, fastCh, slowCh chan *mergeDispatchRequest) {
+ for {
+ select {
+ case <-tst.loopCloser.CloseNotify():
+ return
+ case _, ok := <-triggerCh:
+ if !ok {
+ return
+ }
+ if tst.dispatchAllMerges(threshold, fastCh, slowCh) {
+ return
+ }
+ }
}
- if _, err := tst.mergePartsThenSendIntroduction(snapshotCreatorMerger, dst,
- toBeMerged, merges, tst.loopCloser.CloseNotify(), mergeTypeFile); err != nil {
- return dst, err
+}
+
+func (tst *tsTable) dispatchAllMerges(threshold uint64, fastCh, slowCh chan *mergeDispatchRequest) bool {
+ for {
+ curSnapshot := tst.currentSnapshot()
+ if curSnapshot == nil {
+ return false
+ }
+ freeDiskSize := tst.freeDiskSpace(tst.root)
+ var dst []*partWrapper
+ dst, toBeMerged := tst.getPartsToMerge(curSnapshot, freeDiskSize, dst)
+ if len(dst) < 2 {
+ curSnapshot.decRef()
+ return false
+ }
+ for _, pw := range dst {
+ pw.incRef()
+ }
+ curSnapshot.decRef()
+
+ tst.inFlightMu.Lock()
+ if tst.inFlight == nil {
+ tst.inFlight = make(map[uint64]struct{})
+ }
+ for _, pw := range dst {
+ tst.inFlight[pw.ID()] = struct{}{}
+ }
+ tst.inFlightMu.Unlock()
+
+ var totalSize uint64
+ for _, pw := range dst {
+ totalSize += pw.p.partMetadata.CompressedSizeBytes
+ }
+
+ lane := mergeLaneSlow
+ targetCh := slowCh
+ if totalSize < threshold {
+ lane = mergeLaneFast
+ targetCh = fastCh
+ }
+
+ req := &mergeDispatchRequest{
+ parts: dst,
+ toBeMerged: toBeMerged,
+ typ: mergeTypeFile,
+ lane: lane,
+ enqueuedAt: time.Now(),
+ }
+
+ tst.l.Info().
+ Str("lane", lane).
+ Uint64("totalSize", totalSize).
+ Uint64("threshold", threshold).
+ Int("partCount", len(dst)).
+ Msg("dispatching merge")
+
+ select {
+ case targetCh <- req:
+ case <-tst.loopCloser.CloseNotify():
+ tst.releaseDispatchRequest(req)
+ return true
+ }
+ }
+}
+
+func (tst *tsTable) mergeLaneWorker(ch chan *mergeDispatchRequest, merges chan *mergerIntroduction) {
+ for req := range ch {
+ if !req.enqueuedAt.IsZero() {
+ tst.incTotalMergeQueueLatency(time.Since(req.enqueuedAt).Seconds(), req.typ, req.lane)
+ }
+ select {
+ case mergeMaxConcurrencyCh <- struct{}{}:
+ case <-tst.loopCloser.CloseNotify():
+ tst.releaseDispatchRequest(req)
+ // Drain remaining buffered requests so their inFlight entries and
+ // part references are released. The lane channel is closed by the
+ // mergeLoop shutdown defer after the dispatcher exits, which lets
+ // this range loop terminate.
+ for pending := range ch {
+ tst.releaseDispatchRequest(pending)
+ }
+ return
+ }
+
+ tst.incTotalMergeLoopStarted(1)
+ _, mergeErr := tst.mergePartsThenSendIntroduction(
+ snapshotCreatorMerger, req.parts, req.toBeMerged, merges,
+ tst.loopCloser.CloseNotify(), req.typ, req.lane,
+ )
+ tst.incTotalMergeLoopFinished(1)
+ <-mergeMaxConcurrencyCh
+
+ tst.releaseDispatchRequest(req)
+
+ if mergeErr != nil {
+ if !errors.Is(mergeErr, errClosed) {
+ tst.l.Logger.Warn().Err(mergeErr).Str("typ", req.typ).Str("lane", req.lane).Msg("merge lane worker error")
+ tst.incTotalMergeLoopErr(1)
+ }
+ }
+ }
+}
+
+func (tst *tsTable) releaseDispatchRequest(req *mergeDispatchRequest) {
+ tst.inFlightMu.Lock()
+ for _, pw := range req.parts {
+ delete(tst.inFlight, pw.ID())
+ }
+ tst.inFlightMu.Unlock()
+ for _, pw := range req.parts {
+ pw.decRef()
}
- return dst, nil
}
func (tst *tsTable) mergePartsThenSendIntroduction(creator snapshotCreator, parts []*partWrapper, merged map[uint64]struct{}, merges chan *mergerIntroduction,
- closeCh <-chan struct{}, typ string,
+ closeCh <-chan struct{}, typ string, lane string,
) (*partWrapper, error) {
reservedSpace := tst.reserveSpace(parts)
defer releaseDiskSpace(reservedSpace)
@@ -121,9 +280,9 @@ func (tst *tsTable) mergePartsThenSendIntroduction(creator snapshotCreator, part
return nil, err
}
elapsed := time.Since(start)
- tst.incTotalMergeLatency(elapsed.Seconds(), typ)
- tst.incTotalMerged(1, typ)
- tst.incTotalMergedParts(len(parts), typ)
+ tst.incTotalMergeLatency(elapsed.Seconds(), typ, lane)
+ tst.incTotalMerged(1, typ, lane)
+ tst.incTotalMergedParts(len(parts), typ, lane)
if elapsed > 30*time.Second {
var totalCount uint64
for _, pw := range parts {
@@ -183,9 +342,10 @@ func (tst *tsTable) mergePartsThenSendIntroduction(creator snapshotCreator, part
}
mergerIntroductionMap[sidxName] = mergerIntroduction
elapsed = time.Since(start)
- tst.incTotalMergeLatency(elapsed.Seconds(), fmt.Sprintf("%s_%s", typ, sidxName))
- tst.incTotalMerged(1, fmt.Sprintf("%s_%s", typ, sidxName))
- tst.incTotalMergedParts(len(parts), fmt.Sprintf("%s_%s", typ, sidxName))
+ sidxTyp := fmt.Sprintf("%s_%s", typ, sidxName)
+ tst.incTotalMergeLatency(elapsed.Seconds(), sidxTyp, lane)
+ tst.incTotalMerged(1, sidxTyp, lane)
+ tst.incTotalMergedParts(len(parts), sidxTyp, lane)
if elapsed > 30*time.Second {
tst.l.Warn().Int("mergedPartsCount", len(parts)).Str("sidxName", sidxName).Dur("elapsed", elapsed).Msg("sidx merge parts took too long")
}
@@ -250,12 +410,17 @@ var reservedDiskSpace uint64
func (tst *tsTable) getPartsToMerge(snapshot *snapshot, freeDiskSize uint64, dst []*partWrapper) ([]*partWrapper, map[uint64]struct{}) {
var parts []*partWrapper
+ tst.inFlightMu.RLock()
for _, pw := range snapshot.parts {
if pw.mp != nil || pw.p.partMetadata.TotalCount < 1 {
continue
}
+ if _, inFlight := tst.inFlight[pw.ID()]; inFlight {
+ continue
+ }
parts = append(parts, pw)
}
+ tst.inFlightMu.RUnlock()
dst = tst.option.mergePolicy.getPartsToMerge(dst, parts, freeDiskSize)
if len(dst) == 0 {
diff --git a/banyand/trace/merger_test.go b/banyand/trace/merger_test.go
index e2c5c4e9f..422aa72c4 100644
--- a/banyand/trace/merger_test.go
+++ b/banyand/trace/merger_test.go
@@ -23,6 +23,7 @@ import (
"path/filepath"
"reflect"
"testing"
+ "time"
"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
@@ -1391,7 +1392,7 @@ func Test_mergePartsThenSendIntroduction_cleansUpOnSidxMergeError(t *testing.T)
merges := make(chan *mergerIntroduction, 1)
closeCh := make(chan struct{})
- _, err := tst.mergePartsThenSendIntroduction(snapshotCreatorMerger, parts, merged, merges, closeCh, mergeTypeFile)
+ _, err := tst.mergePartsThenSendIntroduction(snapshotCreatorMerger, parts, merged, merges, closeCh, mergeTypeFile, mergeLaneFast)
require.Error(t, err)
require.Contains(t, err.Error(), "sidx merge failed")
@@ -1457,6 +1458,257 @@ func Test_flush_cleansUpOnSidxFlushError(t *testing.T) {
require.False(t, fileSystem.IsExist(sidxPartPath), "sidx part directory should be removed on sidx flush failure")
}
+func Test_computeSmallMergeThreshold(t *testing.T) {
+ threshold := computeSmallMergeThreshold()
+ require.GreaterOrEqual(t, threshold, uint64(defaultSmallMergeThreshold),
+ "threshold should be at least the default (%d bytes)", defaultSmallMergeThreshold)
+}
+
+func Test_getPartsToMerge_skipsInFlightParts(t *testing.T) {
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
+
+ fileSystem := fs.NewLocalFileSystem()
+
+ // Create 4 file parts
+ var allParts []*partWrapper
+ for i := range 4 {
+ partID := uint64(100 + i)
+ mp := generateMemPart()
+ mp.mustInitFromTraces(tsTS1)
+ mp.mustFlush(fileSystem, partPath(tmpPath, partID))
+ p := mustOpenFilePart(partID, tmpPath, fileSystem)
+ p.partMetadata.ID = partID
+ allParts = append(allParts, newPartWrapper(nil, p))
+ releaseMemPart(mp)
+ }
+
+ closer := run.NewCloser(1)
+ defer closer.Done()
+
+ tst := &tsTable{
+ pm: protector.Nop{},
+ fileSystem: fileSystem,
+ root: tmpPath,
+ loopCloser: closer,
+ l: logger.GetLogger("test"),
+ curPartID: 200,
+ option: option{
+ mergePolicy: newDefaultMergePolicyForTesting(),
+ },
+ }
+
+ snp := &snapshot{
+ parts: allParts,
+ epoch: 1,
+ }
+ snp.incRef()
+ defer snp.decRef()
+
+ // Without inFlight: all parts should be eligible
+ dst, _ := tst.getPartsToMerge(snp, uint64(1<<30), nil)
+ require.GreaterOrEqual(t, len(dst), 2, "should find parts to merge when none are in-flight")
+
+ // Mark part 100 and 101 as in-flight
+ tst.inFlight = make(map[uint64]struct{})
+ tst.inFlight[100] = struct{}{}
+ tst.inFlight[101] = struct{}{}
+
+ // With inFlight: parts 100 and 101 should be skipped
+ dst2, _ := tst.getPartsToMerge(snp, uint64(1<<30), nil)
+ if len(dst2) > 0 {
+ for _, pw := range dst2 {
+ require.NotEqual(t, uint64(100), pw.ID(), "part 100 should be skipped (in-flight)")
+ require.NotEqual(t, uint64(101), pw.ID(), "part 101 should be skipped (in-flight)")
+ }
+ }
+
+ // Mark all parts as in-flight: no parts should be eligible
+ tst.inFlight[102] = struct{}{}
+ tst.inFlight[103] = struct{}{}
+ dst3, _ := tst.getPartsToMerge(snp, uint64(1<<30), nil)
+ require.Nil(t, dst3, "no parts should be eligible when all are in-flight")
+}
+
+func Test_dispatchAllMerges_laneClassification(t *testing.T) {
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
+
+ fileSystem := fs.NewLocalFileSystem()
+
+ closer := run.NewCloser(1)
+
+ tst := &tsTable{
+ pm: protector.Nop{},
+ fileSystem: fileSystem,
+ root: tmpPath,
+ loopCloser: closer,
+ l: logger.GetLogger("test"),
+ curPartID: 200,
+ option: option{
+ mergePolicy: newDefaultMergePolicyForTesting(),
+ },
+ }
+
+ // Use a very low threshold to force all merges into slow lane
+ smallThreshold := uint64(1) // 1 byte - everything is "large"
+ slowCh := make(chan *mergeDispatchRequest, 10)
+ fastCh := make(chan *mergeDispatchRequest, 10)
+
+ // Create 3 file parts and build a snapshot
+ var allParts []*partWrapper
+ for i := range 3 {
+ partID := uint64(200 + i)
+ mp := generateMemPart()
+ mp.mustInitFromTraces(tsTS1)
+ mp.mustFlush(fileSystem, partPath(tmpPath, partID))
+ p := mustOpenFilePart(partID, tmpPath, fileSystem)
+ p.partMetadata.ID = partID
+ allParts = append(allParts, newPartWrapper(nil, p))
+ releaseMemPart(mp)
+ }
+
+ snp := &snapshot{
+ parts: allParts,
+ epoch: 1,
+ }
+ snp.incRef()
+ tst.snapshot = snp
+ defer snp.decRef()
+
+ // Drain introductions
+ merges := make(chan *mergerIntroduction, 10)
+ go func() {
+ for m := range merges {
+ _ = m
+ }
+ }()
+
+ // Run dispatchAllMerges - it should dispatch to slowCh since threshold is 1 byte
+ dispatchDone := make(chan struct{})
+ go func() {
+ defer close(dispatchDone)
+ tst.dispatchAllMerges(smallThreshold, fastCh, slowCh)
+ }()
+
+ // Should receive on slowCh
+ select {
+ case req := <-slowCh:
+ require.Equal(t, mergeTypeFile, req.typ)
+ require.Equal(t, mergeLaneSlow, req.lane)
+ require.GreaterOrEqual(t, len(req.parts), 2, "should merge at least 2 parts")
+ // Verify inFlight was set
+ tst.inFlightMu.RLock()
+ for _, pw := range req.parts {
+ _, exists := tst.inFlight[pw.ID()]
+ require.True(t, exists, "part %d should be in inFlight map", pw.ID())
+ }
+ tst.inFlightMu.RUnlock()
+ // Don't clear inFlight — dispatchAllMerges loop will find no eligible parts and exit naturally
+ case <-fastCh:
+ t.Fatal("should not dispatch to fast lane with threshold=1")
+ case <-time.After(5 * time.Second):
+ t.Fatal("timed out waiting for dispatch")
+ }
+
+ // Wait for dispatchAllMerges to finish (it exits when all parts are in inFlight)
+ select {
+ case <-dispatchDone:
+ case <-time.After(5 * time.Second):
+ t.Fatal("dispatchAllMerges did not finish")
+ }
+
+ // Close the closer after dispatchAllMerges has exited
+ closer.Done()
+ closer.CloseThenWait()
+ close(fastCh)
+ close(slowCh)
+ close(merges)
+}
+
+func Test_mergeLaneWorker_cleansUpInFlight(t *testing.T) {
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
+
+ fileSystem := fs.NewLocalFileSystem()
+
+ closer := run.NewCloser(1)
+
+ tst := &tsTable{
+ pm: protector.Nop{},
+ fileSystem: fileSystem,
+ root: tmpPath,
+ loopCloser: closer,
+ l: logger.GetLogger("test"),
+ curPartID: 0,
+ option: option{
+ mergePolicy: newDefaultMergePolicyForTesting(),
+ },
+ inFlight: make(map[uint64]struct{}),
+ }
+
+ // Create 2 file parts to merge
+ var parts []*partWrapper
+ for i := range 2 {
+ partID := uint64(300 + i)
+ mp := generateMemPart()
+ mp.mustInitFromTraces(tsTS1)
+ mp.mustFlush(fileSystem, partPath(tmpPath, partID))
+ p := mustOpenFilePart(partID, tmpPath, fileSystem)
+ p.partMetadata.ID = partID
+ parts = append(parts, newPartWrapper(nil, p))
+ releaseMemPart(mp)
+ }
+
+ // Mark parts as inFlight
+ toBeMerged := make(map[uint64]struct{})
+ for _, pw := range parts {
+ tst.inFlight[pw.ID()] = struct{}{}
+ toBeMerged[pw.ID()] = struct{}{}
+ }
+
+ require.Len(t, tst.inFlight, 2)
+
+ // Create channels
+ ch := make(chan *mergeDispatchRequest, 1)
+ merges := make(chan *mergerIntroduction, 1)
+
+ // Start worker in background
+ workerDone := make(chan struct{})
+ go func() {
+ defer close(workerDone)
+ tst.mergeLaneWorker(ch, merges)
+ }()
+
+ // Send a dispatch request
+ ch <- &mergeDispatchRequest{
+ parts: parts,
+ toBeMerged: toBeMerged,
+ typ: mergeTypeFile,
+ lane: mergeLaneFast,
+ }
+ close(ch)
+
+ // The worker will try to merge and send introduction.
+ // Since there's no introducerLoop, mergePartsThenSendIntroduction
+ // will block on mi.applied. Close the closer to unblock it.
+ time.Sleep(100 * time.Millisecond)
+ closer.Done()
+ closer.CloseThenWait()
+
+ // Wait for worker to finish
+ select {
+ case <-workerDone:
+ case <-time.After(5 * time.Second):
+ t.Fatal("worker did not finish")
+ }
+
+ // Verify inFlight was cleaned up
+ tst.inFlightMu.RLock()
+ require.Empty(t, tst.inFlight, "inFlight map should be empty after worker completes")
+ tst.inFlightMu.RUnlock()
+}
+
func Test_mergePartsWithConflictingTagTypes(t *testing.T) {
tests := []struct {
name string
diff --git a/banyand/trace/metrics.go b/banyand/trace/metrics.go
index 95479e817..2d84dccee 100644
--- a/banyand/trace/metrics.go
+++ b/banyand/trace/metrics.go
@@ -61,9 +61,10 @@ type metrics struct {
totalFlushIntroLatency meter.Counter
totalFlushLatency meter.Counter
- totalMergedParts meter.Counter
- totalMergeLatency meter.Counter
- totalMerged meter.Counter
+ totalMergedParts meter.Counter
+ totalMergeLatency meter.Counter
+ totalMerged meter.Counter
+ totalMergeQueueLatency meter.Counter
tbMetrics
}
@@ -229,25 +230,32 @@ func (tst *tsTable) incTotalFlushLatency(delta float64) {
tst.metrics.totalFlushLatency.Inc(delta)
}
-func (tst *tsTable) incTotalMergedParts(delta int, typ string) {
+func (tst *tsTable) incTotalMergedParts(delta int, typ, lane string) {
if tst == nil || tst.metrics == nil {
return
}
- tst.metrics.totalMergedParts.Inc(float64(delta), typ)
+ tst.metrics.totalMergedParts.Inc(float64(delta), typ, lane)
}
-func (tst *tsTable) incTotalMergeLatency(delta float64, typ string) {
+func (tst *tsTable) incTotalMergeLatency(delta float64, typ, lane string) {
if tst == nil || tst.metrics == nil {
return
}
- tst.metrics.totalMergeLatency.Inc(delta, typ)
+ tst.metrics.totalMergeLatency.Inc(delta, typ, lane)
}
-func (tst *tsTable) incTotalMerged(delta int, typ string) {
+func (tst *tsTable) incTotalMerged(delta int, typ, lane string) {
if tst == nil || tst.metrics == nil {
return
}
- tst.metrics.totalMerged.Inc(float64(delta), typ)
+ tst.metrics.totalMerged.Inc(float64(delta), typ, lane)
+}
+
+func (tst *tsTable) incTotalMergeQueueLatency(delta float64, typ, lane string) {
+ if tst == nil || tst.metrics == nil {
+ return
+ }
+ tst.metrics.totalMergeQueueLatency.Inc(delta, typ, lane)
}
func (tst *tsTable) addPendingDataCount(delta int64) {
@@ -298,12 +306,17 @@ func (m *metrics) DeleteAll() {
m.totalFlushPauseBreak.Delete()
m.totalFlushLatency.Delete()
- m.totalMergedParts.Delete("mem")
- m.totalMergeLatency.Delete("mem")
- m.totalMerged.Delete("mem")
- m.totalMergedParts.Delete("file")
- m.totalMergeLatency.Delete("file")
- m.totalMerged.Delete("file")
+ m.totalMergedParts.Delete("mem", "")
+ m.totalMergeLatency.Delete("mem", "")
+ m.totalMerged.Delete("mem", "")
+ m.totalMergedParts.Delete("file", "fast")
+ m.totalMergeLatency.Delete("file", "fast")
+ m.totalMerged.Delete("file", "fast")
+ m.totalMergedParts.Delete("file", "slow")
+ m.totalMergeLatency.Delete("file", "slow")
+ m.totalMerged.Delete("file", "slow")
+ m.totalMergeQueueLatency.Delete("file", "fast")
+ m.totalMergeQueueLatency.Delete("file", "slow")
}
func (s *supplier) newMetrics(p common.Position) storage.Metrics {
@@ -332,9 +345,10 @@ func (s *supplier) newMetrics(p common.Position) storage.Metrics {
totalFlushPauseBreak: factory.NewCounter("total_flush_pause_break"),
totalFlushIntroLatency: factory.NewCounter("total_flush_intro_latency"),
totalFlushLatency: factory.NewCounter("total_flush_latency"),
- totalMergedParts: factory.NewCounter("total_merged_parts", "type"),
- totalMergeLatency: factory.NewCounter("total_merge_latency", "type"),
- totalMerged: factory.NewCounter("total_merged", "type"),
+ totalMergedParts: factory.NewCounter("total_merged_parts", "type", "lane"),
+ totalMergeLatency: factory.NewCounter("total_merge_latency", "type", "lane"),
+ totalMerged: factory.NewCounter("total_merged", "type", "lane"),
+ totalMergeQueueLatency: factory.NewCounter("total_merge_queue_latency", "type", "lane"),
tbMetrics: tbMetrics{
totalMemParts: factory.NewGauge("total_mem_part", common.ShardLabelNames()...),
totalMemElements: factory.NewGauge("total_mem_elements", common.ShardLabelNames()...),
@@ -378,9 +392,10 @@ func (qs *queueSupplier) newMetrics(p common.Position) (storage.Metrics, observa
totalFlushPauseBreak: factory.NewCounter("total_flush_pause_break"),
totalFlushIntroLatency: factory.NewCounter("total_flush_intro_latency"),
totalFlushLatency: factory.NewCounter("total_flush_latency"),
- totalMergedParts: factory.NewCounter("total_merged_parts", "type"),
- totalMergeLatency: factory.NewCounter("total_merge_latency", "type"),
- totalMerged: factory.NewCounter("total_merged", "type"),
+ totalMergedParts: factory.NewCounter("total_merged_parts", "type", "lane"),
+ totalMergeLatency: factory.NewCounter("total_merge_latency", "type", "lane"),
+ totalMerged: factory.NewCounter("total_merged", "type", "lane"),
+ totalMergeQueueLatency: factory.NewCounter("total_merge_queue_latency", "type", "lane"),
tbMetrics: tbMetrics{
totalMemParts: factory.NewGauge("total_mem_part", common.ShardLabelNames()...),
totalMemElements: factory.NewGauge("total_mem_elements", common.ShardLabelNames()...),
diff --git a/banyand/trace/tstable.go b/banyand/trace/tstable.go
index 05d5e11d1..952e8987b 100644
--- a/banyand/trace/tstable.go
+++ b/banyand/trace/tstable.go
@@ -48,8 +48,9 @@ const (
)
type tsTable struct {
- pm protector.Memory
fileSystem fs.FileSystem
+ pm protector.Memory
+ inFlight map[uint64]struct{}
handoffCtrl *handoffController
metrics *metrics
snapshot *snapshot
@@ -59,12 +60,13 @@ type tsTable struct {
sidxMap map[string]sidx.SIDX
introductions chan *introduction
p common.Position
- root string
group string
+ root string
gc garbageCleaner
option option
curPartID uint64
pendingDataCount atomic.Int64
+ inFlightMu sync.RWMutex
sync.RWMutex
shardID common.ShardID
}