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 2024/01/09 02:35:47 UTC
(skywalking-banyandb) 02/04: Fix linter
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch measure-merger
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit e69360b21629ae99926e047882f3187f32745647
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Tue Jan 9 01:32:58 2024 +0000
Fix linter
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/internal/storage/tsdb.go | 2 +-
banyand/measure/block.go | 1 -
banyand/measure/block_reader.go | 4 ++--
banyand/measure/flusher.go | 38 ++++++++++----------------------------
banyand/measure/gc.go | 13 -------------
banyand/measure/introducer.go | 4 ++++
banyand/measure/merger.go | 28 ++++++++++++++++++++++++----
banyand/measure/part_iter.go | 4 ++--
pkg/bytes/buffer.go | 1 +
pkg/fs/local_file_system.go | 12 ++++++------
10 files changed, 50 insertions(+), 57 deletions(-)
diff --git a/banyand/internal/storage/tsdb.go b/banyand/internal/storage/tsdb.go
index 87bb0291..16b5fb09 100644
--- a/banyand/internal/storage/tsdb.go
+++ b/banyand/internal/storage/tsdb.go
@@ -63,10 +63,10 @@ func generateSegID(unit IntervalUnit, suffix int) segmentID {
type database[T TSTable, O any] struct {
logger *logger.Logger
index *seriesIndex
+ p common.Position
location string
sLst []*shard[T, O]
opts TSDBOpts[T, O]
- p common.Position
sync.RWMutex
sLen uint32
}
diff --git a/banyand/measure/block.go b/banyand/measure/block.go
index e1af6a5f..7da65663 100644
--- a/banyand/measure/block.go
+++ b/banyand/measure/block.go
@@ -225,7 +225,6 @@ func (b *block) unmarshalTagFamily(decoder *encoding.BytesBlockDecoder, tfIndex
}
}
}
- return
}
func (b *block) unmarshalTagFamilyFromSeqReaders(decoder *encoding.BytesBlockDecoder, tfIndex int, name string,
diff --git a/banyand/measure/block_reader.go b/banyand/measure/block_reader.go
index 5596682a..219d0a94 100644
--- a/banyand/measure/block_reader.go
+++ b/banyand/measure/block_reader.go
@@ -80,11 +80,11 @@ func releaseSeqReader(sr *seqReader) {
var seqReaderPool sync.Pool
type seqReaders struct {
+ tagFamilyMetadata map[string]*seqReader
+ tagFamilies map[string]*seqReader
primary seqReader
timestamps seqReader
fieldValues seqReader
- tagFamilyMetadata map[string]*seqReader
- tagFamilies map[string]*seqReader
}
func (sr *seqReaders) reset() {
diff --git a/banyand/measure/flusher.go b/banyand/measure/flusher.go
index 2ce4e787..16ae914d 100644
--- a/banyand/measure/flusher.go
+++ b/banyand/measure/flusher.go
@@ -53,20 +53,16 @@ func (tst *tsTable) flusherLoop(flushCh chan *flusherIntroduction, mergeCh chan
}
tst.RUnlock()
if curSnapshot != nil {
- newSnapshot, err := tst.mergeMemParts(curSnapshot, mergeCh)
+ merged, err := tst.mergeMemParts(curSnapshot, mergeCh)
if err != nil {
tst.l.Logger.Warn().Err(err).Msgf("cannot merge snapshot: %d", curSnapshot.epoch)
curSnapshot.decRef()
continue
}
- var toBePersistedSnapshot *snapshot
- if newSnapshot == nil {
+ if !merged {
tst.flush(curSnapshot, flushCh)
- toBePersistedSnapshot = curSnapshot
- } else {
- toBePersistedSnapshot = newSnapshot
}
- epoch = tst.persistSnapshot(toBePersistedSnapshot)
+ epoch = curSnapshot.epoch
// Notify merger to start a new round of merge.
// This round might have be triggered in pauseFlusherToPileupMemParts.
flusherWatchers.Notify(math.MaxUint64)
@@ -80,7 +76,6 @@ func (tst *tsTable) flusherLoop(flushCh chan *flusherIntroduction, mergeCh chan
if epochWatcher == nil {
return
}
- tst.gc.clean()
}
}
}
@@ -105,9 +100,8 @@ func (tst *tsTable) pauseFlusherToPileupMemParts(epoch uint64, flushWatcher watc
return flusherWatchers
}
-func (tst *tsTable) mergeMemParts(snp *snapshot, mergeCh chan *mergerIntroduction) (*snapshot, error) {
+func (tst *tsTable) mergeMemParts(snp *snapshot, mergeCh chan *mergerIntroduction) (bool, error) {
var memParts []*partWrapper
- var remainingParts []*partWrapper
mergedIDs := make(map[uint64]struct{})
for i := range snp.parts {
if snp.parts[i].mp != nil {
@@ -115,10 +109,9 @@ func (tst *tsTable) mergeMemParts(snp *snapshot, mergeCh chan *mergerIntroductio
mergedIDs[snp.parts[i].ID()] = struct{}{}
continue
}
- remainingParts = append(remainingParts, snp.parts[i])
}
if len(memParts) < 2 {
- return nil, nil
+ return false, nil
}
// merge memory must not be closed by the tsTable.close
closeCh := make(chan struct{})
@@ -126,24 +119,14 @@ func (tst *tsTable) mergeMemParts(snp *snapshot, mergeCh chan *mergerIntroductio
close(closeCh)
if err != nil {
if errors.Is(err, errClosed) {
- return &snapshot{
- epoch: snp.epoch,
- parts: append(remainingParts, newPart),
- creator: snapshotCreatorMergedFlusher,
- ref: 1,
- }, nil
+ return true, nil
}
- return nil, err
+ return false, err
}
if newPart == nil {
- return nil, nil
+ return false, nil
}
- return &snapshot{
- epoch: snp.epoch,
- parts: append(remainingParts, newPart),
- creator: snapshotCreatorMergedFlusher,
- ref: 1,
- }, nil
+ return true, nil
}
func (tst *tsTable) flush(snapshot *snapshot, flushCh chan *flusherIntroduction) {
@@ -174,12 +157,11 @@ func (tst *tsTable) flush(snapshot *snapshot, flushCh chan *flusherIntroduction)
}
}
-func (tst *tsTable) persistSnapshot(snapshot *snapshot) uint64 {
+func (tst *tsTable) persistSnapshot(snapshot *snapshot) {
var partNames []string
for i := range snapshot.parts {
partNames = append(partNames, partName(snapshot.parts[i].ID()))
}
tst.mustWriteSnapshot(snapshot.epoch, partNames)
tst.gc.registerSnapshot(snapshot)
- return snapshot.epoch
}
diff --git a/banyand/measure/gc.go b/banyand/measure/gc.go
index 7bd68644..039d5295 100644
--- a/banyand/measure/gc.go
+++ b/banyand/measure/gc.go
@@ -84,17 +84,4 @@ OUTER:
g.parent.fileSystem.MustRMAll(partPath(g.parent.root, partID))
delete(g.knownPartFiles, partID)
}
- ee := g.parent.fileSystem.ReadDir(g.parent.root)
- for i := range ee {
- if ee[i].IsDir() {
- p, err := parseEpoch(ee[i].Name())
- if err != nil {
- g.parent.l.Info().Err(err).Msg("cannot parse part file name. skip and delete it")
- continue
- }
- if _, ok := g.knownPartFiles[p]; !ok {
- g.parent.l.Error().Msgf("unknown part file: %d", p)
- }
- }
- }
}
diff --git a/banyand/measure/introducer.go b/banyand/measure/introducer.go
index ef394a70..e3bb2e01 100644
--- a/banyand/measure/introducer.go
+++ b/banyand/measure/introducer.go
@@ -120,9 +120,11 @@ func (tst *tsTable) introducerLoop(flushCh chan *flusherIntroduction, mergeCh ch
epoch++
case next := <-flushCh:
tst.introduceFlushed(next, epoch)
+ tst.gc.cleanSnapshots()
epoch++
case next := <-mergeCh:
tst.introduceMerged(next, epoch)
+ tst.gc.clean()
epoch++
case epochWatcher := <-watcherCh:
introducerWatchers.Add(epochWatcher)
@@ -159,6 +161,7 @@ func (tst *tsTable) introduceFlushed(nextIntroduction *flusherIntroduction, epoc
nextSnp := cur.merge(epoch, nextIntroduction.flushed)
nextSnp.creator = snapshotCreatorFlusher
tst.replaceSnapshot(&nextSnp)
+ tst.persistSnapshot(&nextSnp)
if nextIntroduction.applied != nil {
close(nextIntroduction.applied)
}
@@ -175,6 +178,7 @@ func (tst *tsTable) introduceMerged(nextIntroduction *mergerIntroduction, epoch
nextSnp.parts = append(nextSnp.parts, nextIntroduction.newPart)
nextSnp.creator = nextIntroduction.creator
tst.replaceSnapshot(&nextSnp)
+ tst.persistSnapshot(&nextSnp)
if nextIntroduction.applied != nil {
close(nextIntroduction.applied)
}
diff --git a/banyand/measure/merger.go b/banyand/measure/merger.go
index 1bfb1620..11d92788 100644
--- a/banyand/measure/merger.go
+++ b/banyand/measure/merger.go
@@ -91,17 +91,37 @@ func (tst *tsTable) mergePartsThenSendIntroduction(creator snapshotCreator, part
if err != nil {
return nil, err
}
- if snapshotCreatorMerger == creator && tst.l.Info().Enabled() {
+ elapsed := time.Since(start)
+ if elapsed > 30*time.Second {
var totalCount uint64
for _, pw := range parts {
totalCount += pw.p.partMetadata.TotalCount
}
- tst.l.Info().
+ tst.l.Warn().
Uint64("beforeTotalCount", totalCount).
Uint64("afterTotalCount", newPart.p.partMetadata.TotalCount).
Int("beforePartCount", len(parts)).
- TimeDiff("duration", time.Now(), start).
- Msg("background merger merges on-disk parts")
+ Dur("elapsed", elapsed).
+ Msg("background merger takes too long")
+ } else if snapshotCreatorMerger == creator && tst.l.Info().Enabled() {
+ var minCount, maxCount, totalCount uint64
+ for _, pw := range parts {
+ totalCount += pw.p.partMetadata.TotalCount
+ if minCount == 0 || minCount > pw.p.partMetadata.TotalCount {
+ minCount = pw.p.partMetadata.TotalCount
+ }
+ if maxCount < pw.p.partMetadata.TotalCount {
+ maxCount = pw.p.partMetadata.TotalCount
+ }
+ }
+ if minCount*uint64(len(parts)) < maxCount {
+ tst.l.Info().
+ Uint64("beforeTotalCount", totalCount).
+ Uint64("afterTotalCount", newPart.p.partMetadata.TotalCount).
+ Int("beforePartCount", len(parts)).
+ Dur("elapsed", elapsed).
+ Msg("background merger merges too big difference parts")
+ }
}
mi := generateMergerIntroduction()
defer releaseMergerIntroduction(mi)
diff --git a/banyand/measure/part_iter.go b/banyand/measure/part_iter.go
index b294d4e6..c44d114a 100644
--- a/banyand/measure/part_iter.go
+++ b/banyand/measure/part_iter.go
@@ -237,13 +237,13 @@ func (pi *partIter) findBlock() bool {
}
type partMergeIter struct {
- err error
seqReaders seqReaders
+ err error
primaryBlockMetadata []primaryBlockMetadata
- partID uint64
compressedPrimaryBuf []byte
primaryBuf []byte
block blockPointer
+ partID uint64
primaryMetadataIdx int
}
diff --git a/pkg/bytes/buffer.go b/pkg/bytes/buffer.go
index 6408f4c9..be4db36b 100644
--- a/pkg/bytes/buffer.go
+++ b/pkg/bytes/buffer.go
@@ -62,6 +62,7 @@ func (b *Buffer) Read(offset int64, buffer []byte) (int, error) {
return n, err
}
+// StreamRead implements fs.Reader.
func (b *Buffer) StreamRead() io.Reader {
return &reader{bb: b}
}
diff --git a/pkg/fs/local_file_system.go b/pkg/fs/local_file_system.go
index 1ac96860..b3438265 100644
--- a/pkg/fs/local_file_system.go
+++ b/pkg/fs/local_file_system.go
@@ -376,7 +376,7 @@ func (file *LocalFile) Readv(offset int64, iov *[][]byte) (int, error) {
// StreamRead is used to read the entire file using streaming read.
func (file *LocalFile) StreamRead() io.Reader {
reader := bufio.NewReader(file.file)
- return &Iter{reader: reader, fileName: file.file.Name()}
+ return &seqReader{reader: reader, fileName: file.file.Name()}
}
// Size is used to get the file written data's size and return an error if the file does not exist. The unit of file size is Byte.
@@ -425,15 +425,15 @@ func (file *LocalFile) Close() error {
return nil
}
-type Iter struct {
- fileName string
+type seqReader struct {
reader *bufio.Reader
+ fileName string
}
-func (iter *Iter) Read(p []byte) (int, error) {
- rsize, err := iter.reader.Read(p)
+func (i *seqReader) Read(p []byte) (int, error) {
+ rsize, err := i.reader.Read(p)
if err != nil {
- return readErrorHandle("ReadStream operation", err, iter.fileName, rsize)
+ return readErrorHandle("ReadStream operation", err, i.fileName, rsize)
}
return rsize, nil
}