You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2026/04/22 05:29:21 UTC
(skywalking-banyandb) branch feat/merge created (now 1c5114938)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch feat/merge
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
at 1c5114938 fix(trace): harden merge dispatcher lifecycle and metric semantics
This branch includes the following new commits:
new 6c546f86f feat(trace): introduce fair fast/slow lane scheduling for part merges
new 1c5114938 fix(trace): harden merge dispatcher lifecycle and metric semantics
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(skywalking-banyandb) 01/02: feat(trace): introduce fair fast/slow lane scheduling for part merges
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch feat/merge
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 6c546f86ff64c8336829c9d3714e1f1c6fadc17b
Author: Hongtao Gao <ha...@gmail.com>
AuthorDate: Tue Apr 21 04:21:02 2026 +0000
feat(trace): introduce fair fast/slow lane scheduling for part merges
Previously a single merge worker processed all file-part merges serially,
so a long-running large merge could stall short merges indefinitely. Split
the merge pipeline into a dispatcher goroutine plus two lane workers
(fast lane sized CPUs/2, slow lane sized 1) and classify each merge by
total compressed size against a memory-derived threshold (memLimit/16,
clamped to at least 32MB). In-flight tracking prevents the same part
being scheduled twice across lanes. Expose queue wait time as
total_merge_queue_latency for observability.
Fixes apache/skywalking#13776
---
CHANGES.md | 1 +
banyand/trace/merger.go | 252 +++++++++++++++++++++++++++++++++++--------
banyand/trace/merger_test.go | 249 ++++++++++++++++++++++++++++++++++++++++++
banyand/trace/metrics.go | 27 +++--
banyand/trace/tstable.go | 2 +
5 files changed, 483 insertions(+), 48 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 0dcd8a960..ff8ad5adc 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -12,6 +12,7 @@ Release Notes.
- Sync lifecycle e2e test from SkyWalking stages test.
- 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`.
### Bug Fixes
diff --git a/banyand/trace/merger.go b/banyand/trace/merger.go
index 875f98fb8..934a201b6 100644
--- a/banyand/trace/merger.go
+++ b/banyand/trace/merger.go
@@ -21,6 +21,7 @@ import (
"errors"
"fmt"
"path/filepath"
+ "sync"
"sync/atomic"
"time"
@@ -40,72 +41,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
+ parts []*partWrapper
+ toBeMerged map[uint64]struct{}
+ typ string
+}
+
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 {
- 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
- }
- epoch = curSnapshot.epoch
+ curSnapshot := tst.currentSnapshot()
+ if curSnapshot == nil {
+ return
+ }
+ if curSnapshot.epoch > lastProcessedEpoch {
+ tst.incTotalMergeLoopStarted(1)
+ select {
+ case triggerCh <- struct{}{}:
+ default:
}
- ew = flusherNotifier.Add(epoch, tst.loopCloser.CloseNotify())
- return ew == nil
- }() {
+ lastProcessedEpoch = curSnapshot.epoch
+ tst.incTotalMergeLoopFinished(1)
+ }
+ 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
+ }
+
+ typ := fmt.Sprintf("%s_%s", mergeTypeFile, lane)
+ req := &mergeDispatchRequest{
+ parts: dst,
+ toBeMerged: toBeMerged,
+ typ: typ,
+ 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.inFlightMu.Lock()
+ for _, pw := range dst {
+ delete(tst.inFlight, pw.ID())
+ }
+ tst.inFlightMu.Unlock()
+ for _, pw := range dst {
+ pw.decRef()
+ }
+ 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)
+ }
+ select {
+ case mergeMaxConcurrencyCh <- struct{}{}:
+ case <-tst.loopCloser.CloseNotify():
+ tst.inFlightMu.Lock()
+ for _, pw := range req.parts {
+ delete(tst.inFlight, pw.ID())
+ }
+ tst.inFlightMu.Unlock()
+ for _, pw := range req.parts {
+ pw.decRef()
+ }
+ return
+ }
+
+ _, mergeErr := tst.mergePartsThenSendIntroduction(
+ snapshotCreatorMerger, req.parts, req.toBeMerged, merges,
+ tst.loopCloser.CloseNotify(), req.typ,
+ )
+ <-mergeMaxConcurrencyCh
+
+ tst.inFlightMu.Lock()
+ for _, pw := range req.parts {
+ delete(tst.inFlight, pw.ID())
+ }
+ tst.inFlightMu.Unlock()
+ for _, pw := range req.parts {
+ pw.decRef()
+ }
+
+ if mergeErr != nil {
+ if !errors.Is(mergeErr, errClosed) {
+ tst.l.Logger.Warn().Err(mergeErr).Str("typ", req.typ).Msg("merge lane worker error")
+ tst.incTotalMergeLoopErr(1)
+ }
+ }
}
- return dst, nil
}
func (tst *tsTable) mergePartsThenSendIntroduction(creator snapshotCreator, parts []*partWrapper, merged map[uint64]struct{}, merges chan *mergerIntroduction,
@@ -249,12 +412,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 f162b400e..131c1bf81 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"
@@ -1456,3 +1457,251 @@ func Test_flush_cleansUpOnSidxFlushError(t *testing.T) {
sidxPartPath := filepath.Join(tmpPath, sidxDirName, "idx1", fmt.Sprintf("%016x", partID))
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 range merges {
+ }
+ }()
+
+ // 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, fmt.Sprintf("%s_%s", mergeTypeFile, mergeLaneSlow), req.typ)
+ 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: fmt.Sprintf("%s_%s", mergeTypeFile, 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()
+}
diff --git a/banyand/trace/metrics.go b/banyand/trace/metrics.go
index 95479e817..9066fca3b 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
}
@@ -250,6 +251,13 @@ func (tst *tsTable) incTotalMerged(delta int, typ string) {
tst.metrics.totalMerged.Inc(float64(delta), typ)
}
+func (tst *tsTable) incTotalMergeQueueLatency(delta float64, typ string) {
+ if tst == nil || tst.metrics == nil {
+ return
+ }
+ tst.metrics.totalMergeQueueLatency.Inc(delta, typ)
+}
+
func (tst *tsTable) addPendingDataCount(delta int64) {
tst.pendingDataCount.Add(delta)
if tst.metrics == nil {
@@ -301,9 +309,14 @@ func (m *metrics) DeleteAll() {
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("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 {
@@ -335,6 +348,7 @@ func (s *supplier) newMetrics(p common.Position) storage.Metrics {
totalMergedParts: factory.NewCounter("total_merged_parts", "type"),
totalMergeLatency: factory.NewCounter("total_merge_latency", "type"),
totalMerged: factory.NewCounter("total_merged", "type"),
+ totalMergeQueueLatency: factory.NewCounter("total_merge_queue_latency", "type"),
tbMetrics: tbMetrics{
totalMemParts: factory.NewGauge("total_mem_part", common.ShardLabelNames()...),
totalMemElements: factory.NewGauge("total_mem_elements", common.ShardLabelNames()...),
@@ -381,6 +395,7 @@ func (qs *queueSupplier) newMetrics(p common.Position) (storage.Metrics, observa
totalMergedParts: factory.NewCounter("total_merged_parts", "type"),
totalMergeLatency: factory.NewCounter("total_merge_latency", "type"),
totalMerged: factory.NewCounter("total_merged", "type"),
+ totalMergeQueueLatency: factory.NewCounter("total_merge_queue_latency", "type"),
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 dcb68f9b9..59e797725 100644
--- a/banyand/trace/tstable.go
+++ b/banyand/trace/tstable.go
@@ -65,6 +65,8 @@ type tsTable struct {
option option
curPartID uint64
pendingDataCount atomic.Int64
+ inFlight map[uint64]struct{}
+ inFlightMu sync.RWMutex
sync.RWMutex
shardID common.ShardID
}
(skywalking-banyandb) 02/02: fix(trace): harden merge dispatcher lifecycle and metric semantics
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch feat/merge
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 1c511493848736869eb869deac7be0560ca5eb4d
Author: Hongtao Gao <ha...@gmail.com>
AuthorDate: Wed Apr 22 05:27:10 2026 +0000
fix(trace): harden merge dispatcher lifecycle and metric semantics
- Keep mergeLoop running when currentSnapshot returns nil so a transient
nil during startup or shutdown no longer terminates the merger.
- Drain remaining buffered dispatch requests when a lane worker exits on
CloseNotify, releasing their inFlight entries and part references.
- Move totalMergeLoopStarted/Finished to bracket the actual merge in the
worker so the counters track execution instead of trigger sends.
---
banyand/trace/merger.go | 65 +++++++++++++++++++++++--------------------------
1 file changed, 30 insertions(+), 35 deletions(-)
diff --git a/banyand/trace/merger.go b/banyand/trace/merger.go
index 934a201b6..ab95ab6ed 100644
--- a/banyand/trace/merger.go
+++ b/banyand/trace/merger.go
@@ -120,20 +120,16 @@ func (tst *tsTable) mergeLoop(merges chan *mergerIntroduction, flusherNotifier w
case <-tst.loopCloser.CloseNotify():
return
case <-ew.Watch():
- curSnapshot := tst.currentSnapshot()
- if curSnapshot == nil {
- return
- }
- if curSnapshot.epoch > lastProcessedEpoch {
- tst.incTotalMergeLoopStarted(1)
- select {
- case triggerCh <- struct{}{}:
- default:
+ if curSnapshot := tst.currentSnapshot(); curSnapshot != nil {
+ if curSnapshot.epoch > lastProcessedEpoch {
+ select {
+ case triggerCh <- struct{}{}:
+ default:
+ }
+ lastProcessedEpoch = curSnapshot.epoch
}
- lastProcessedEpoch = curSnapshot.epoch
- tst.incTotalMergeLoopFinished(1)
+ curSnapshot.decRef()
}
- curSnapshot.decRef()
ew = flusherNotifier.Add(lastProcessedEpoch, tst.loopCloser.CloseNotify())
if ew == nil {
return
@@ -215,14 +211,7 @@ func (tst *tsTable) dispatchAllMerges(threshold uint64, fastCh, slowCh chan *mer
select {
case targetCh <- req:
case <-tst.loopCloser.CloseNotify():
- tst.inFlightMu.Lock()
- for _, pw := range dst {
- delete(tst.inFlight, pw.ID())
- }
- tst.inFlightMu.Unlock()
- for _, pw := range dst {
- pw.decRef()
- }
+ tst.releaseDispatchRequest(req)
return true
}
}
@@ -236,31 +225,26 @@ func (tst *tsTable) mergeLaneWorker(ch chan *mergeDispatchRequest, merges chan *
select {
case mergeMaxConcurrencyCh <- struct{}{}:
case <-tst.loopCloser.CloseNotify():
- tst.inFlightMu.Lock()
- for _, pw := range req.parts {
- delete(tst.inFlight, pw.ID())
- }
- tst.inFlightMu.Unlock()
- for _, pw := range req.parts {
- pw.decRef()
+ 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,
)
+ tst.incTotalMergeLoopFinished(1)
<-mergeMaxConcurrencyCh
- tst.inFlightMu.Lock()
- for _, pw := range req.parts {
- delete(tst.inFlight, pw.ID())
- }
- tst.inFlightMu.Unlock()
- for _, pw := range req.parts {
- pw.decRef()
- }
+ tst.releaseDispatchRequest(req)
if mergeErr != nil {
if !errors.Is(mergeErr, errClosed) {
@@ -271,6 +255,17 @@ func (tst *tsTable) mergeLaneWorker(ch chan *mergeDispatchRequest, merges chan *
}
}
+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()
+ }
+}
+
func (tst *tsTable) mergePartsThenSendIntroduction(creator snapshotCreator, parts []*partWrapper, merged map[uint64]struct{}, merges chan *mergerIntroduction,
closeCh <-chan struct{}, typ string,
) (*partWrapper, error) {