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 2022/08/11 06:48:41 UTC
[skywalking-banyandb] branch main updated: Fix tsdb's leaked goroutines (#154)
This is an automated email from the ASF dual-hosted git repository.
hanahmily 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 42c81ee Fix tsdb's leaked goroutines (#154)
42c81ee is described below
commit 42c81eec7c07616d9a86123b62bb0f97eb5c30f5
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu Aug 11 14:48:35 2022 +0800
Fix tsdb's leaked goroutines (#154)
* Fix tsdb's leaked goroutines
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/tsdb/block.go | 31 ++++++++++++++++-----
banyand/tsdb/bucket/bucket.go | 5 +++-
banyand/tsdb/bucket/queue_test.go | 7 +++++
banyand/tsdb/bucket/strategy.go | 52 +++++++++++++++++++++++-------------
banyand/tsdb/bucket/strategy_test.go | 9 ++++++-
banyand/tsdb/segment.go | 7 +++--
banyand/tsdb/shard_test.go | 4 +++
dist/LICENSE | 2 +-
go.mod | 2 +-
go.sum | 4 +--
10 files changed, 90 insertions(+), 33 deletions(-)
diff --git a/banyand/tsdb/block.go b/banyand/tsdb/block.go
index 18393c0..30744a2 100644
--- a/banyand/tsdb/block.go
+++ b/banyand/tsdb/block.go
@@ -70,6 +70,7 @@ type block struct {
blockID uint16
encodingMethod EncodingMethod
flushCh chan struct{}
+ stopCh chan struct{}
}
type blockOpts struct {
@@ -96,7 +97,7 @@ func newBlock(ctx context.Context, opts blockOpts) (b *block, err error) {
l: logger.Fetch(ctx, "block"),
TimeRange: timeRange,
Reporter: bucket.NewTimeBasedReporter(timeRange, clock),
- flushCh: make(chan struct{}),
+ flushCh: make(chan struct{}, 1),
ref: &atomic.Int32{},
closed: &atomic.Bool{},
queue: opts.queue,
@@ -106,11 +107,7 @@ func newBlock(ctx context.Context, opts blockOpts) (b *block, err error) {
if position != nil {
b.position = position.(common.Position)
}
- go func() {
- for range b.flushCh {
- b.flush()
- }
- }()
+
return b, b.open()
}
@@ -167,6 +164,17 @@ func (b *block) open() (err error) {
}
b.closableLst = append(b.closableLst, b.invertedIndex, b.lsmIndex)
b.ref.Store(0)
+ b.stopCh = make(chan struct{})
+ go func() {
+ for {
+ select {
+ case <-b.flushCh:
+ b.flush()
+ case <-b.stopCh:
+ return
+ }
+ }
+ }()
b.closed.Store(false)
return nil
}
@@ -245,11 +253,22 @@ func (b *block) flush() {
func (b *block) close() {
b.lock.Lock()
defer b.lock.Unlock()
+ if b.closed.Load() {
+ return
+ }
b.closed.Store(true)
b.waitDone()
for _, closer := range b.closableLst {
_ = closer.Close()
}
+ close(b.stopCh)
+}
+
+func (b *block) stopThenClose() {
+ if b.Reporter != nil {
+ b.Stop()
+ }
+ b.close()
}
func (b *block) Closed() bool {
diff --git a/banyand/tsdb/bucket/bucket.go b/banyand/tsdb/bucket/bucket.go
index 615293b..ad5b051 100644
--- a/banyand/tsdb/bucket/bucket.go
+++ b/banyand/tsdb/bucket/bucket.go
@@ -49,6 +49,9 @@ type timeBasedReporter struct {
}
func NewTimeBasedReporter(timeRange timestamp.TimeRange, clock timestamp.Clock) Reporter {
+ if timeRange.End.Before(clock.Now()) {
+ return nil
+ }
return &timeBasedReporter{
TimeRange: timeRange,
reporterStopCh: make(chan struct{}),
@@ -57,7 +60,7 @@ func NewTimeBasedReporter(timeRange timestamp.TimeRange, clock timestamp.Clock)
}
func (tr *timeBasedReporter) Report() Channel {
- ch := make(Channel)
+ ch := make(Channel, 1)
interval := tr.Duration() >> 4
if interval < 100*time.Millisecond {
interval = 100 * time.Millisecond
diff --git a/banyand/tsdb/bucket/queue_test.go b/banyand/tsdb/bucket/queue_test.go
index fb90169..0eb6fe0 100644
--- a/banyand/tsdb/bucket/queue_test.go
+++ b/banyand/tsdb/bucket/queue_test.go
@@ -19,6 +19,7 @@ package bucket_test
import (
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
+ "github.com/onsi/gomega/gleak"
"github.com/apache/skywalking-banyandb/banyand/tsdb/bucket"
)
@@ -36,6 +37,12 @@ func entryID(id uint16) queueEntryID {
}
var _ = Describe("Queue", func() {
+ BeforeEach(func() {
+ goods := gleak.Goroutines()
+ DeferCleanup(func() {
+ Eventually(gleak.Goroutines).ShouldNot(gleak.HaveLeaked(goods))
+ })
+ })
It("pushes data", func() {
evictLst := make([]queueEntryID, 0)
l, err := bucket.NewQueue(128, func(id interface{}) {
diff --git a/banyand/tsdb/bucket/strategy.go b/banyand/tsdb/bucket/strategy.go
index cf928d6..de8011c 100644
--- a/banyand/tsdb/bucket/strategy.go
+++ b/banyand/tsdb/bucket/strategy.go
@@ -42,6 +42,7 @@ type Strategy struct {
current atomic.Value
currentRatio uint64
logger *logger.Logger
+ stopCh chan struct{}
}
type StrategyOptions func(*Strategy)
@@ -68,8 +69,9 @@ func NewStrategy(ctrl Controller, options ...StrategyOptions) (*Strategy, error)
return nil, errors.Wrap(ErrInvalidParameter, "controller is absent")
}
strategy := &Strategy{
- ctrl: ctrl,
- ratio: 0.8,
+ ctrl: ctrl,
+ ratio: 0.8,
+ stopCh: make(chan struct{}),
}
for _, opt := range options {
opt(strategy)
@@ -89,8 +91,11 @@ func (s *Strategy) Run() {
}
go func(s *Strategy) {
for {
+
c := s.current.Load().(Reporter).Report()
- s.observe(c)
+ if !s.observe(c) {
+ return
+ }
}
}(s)
}
@@ -104,30 +109,39 @@ func (s *Strategy) String() string {
math.Float64frombits(atomic.LoadUint64(&s.currentRatio)))
}
-func (s *Strategy) observe(c Channel) {
+func (s *Strategy) observe(c Channel) bool {
var err error
var next Reporter
moreBucket := true
- for status := range c {
- ratio := Ratio(status.Volume) / Ratio(status.Capacity)
- atomic.StoreUint64(&s.currentRatio, math.Float64bits(float64(ratio)))
- if ratio >= s.ratio && next == nil && moreBucket {
- next, err = s.ctrl.Next()
- if errors.Is(err, ErrNoMoreBucket) {
- moreBucket = false
- } else if err != nil {
- s.logger.Err(err).Msg("failed to create the next bucket")
+ for {
+ select {
+ case status, more := <-c:
+ if !more {
+ return true
}
- }
- if ratio >= 1.0 {
- s.ctrl.OnMove(s.current.Load().(Reporter), next)
- if next != nil {
- s.current.Store(next)
+ ratio := Ratio(status.Volume) / Ratio(status.Capacity)
+ atomic.StoreUint64(&s.currentRatio, math.Float64bits(float64(ratio)))
+ if ratio >= s.ratio && next == nil && moreBucket {
+ next, err = s.ctrl.Next()
+ if errors.Is(err, ErrNoMoreBucket) {
+ moreBucket = false
+ } else if err != nil {
+ s.logger.Err(err).Msg("failed to create the next bucket")
+ }
}
- return
+ if ratio >= 1.0 {
+ s.ctrl.OnMove(s.current.Load().(Reporter), next)
+ if next != nil {
+ s.current.Store(next)
+ }
+ return true
+ }
+ case <-s.stopCh:
+ return false
}
}
}
func (s *Strategy) Close() {
+ close(s.stopCh)
}
diff --git a/banyand/tsdb/bucket/strategy_test.go b/banyand/tsdb/bucket/strategy_test.go
index b73c044..aca1022 100644
--- a/banyand/tsdb/bucket/strategy_test.go
+++ b/banyand/tsdb/bucket/strategy_test.go
@@ -21,11 +21,18 @@ import (
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
+ "github.com/onsi/gomega/gleak"
"github.com/apache/skywalking-banyandb/banyand/tsdb/bucket"
)
var _ = Describe("Strategy", func() {
+ BeforeEach(func() {
+ goods := gleak.Goroutines()
+ DeferCleanup(func() {
+ Eventually(gleak.Goroutines).ShouldNot(gleak.HaveLeaked(goods))
+ })
+ })
Context("Applying the strategy", func() {
var strategy *bucket.Strategy
It("uses the golden settings", func() {
@@ -127,7 +134,7 @@ type reporter struct {
}
func (r *reporter) Report() bucket.Channel {
- ch := make(bucket.Channel)
+ ch := make(bucket.Channel, r.capacity)
go func() {
var volume int
for i := 0; i < r.capacity; i++ {
diff --git a/banyand/tsdb/segment.go b/banyand/tsdb/segment.go
index 453548e..be2cb7a 100644
--- a/banyand/tsdb/segment.go
+++ b/banyand/tsdb/segment.go
@@ -110,7 +110,10 @@ func (s *segment) close() {
if s.globalIndex != nil {
s.globalIndex.Close()
}
- s.Stop()
+ s.blockManageStrategy.Close()
+ if s.Reporter != nil {
+ s.Stop()
+ }
}
func (s *segment) closeBlock(id uint16) {
@@ -373,6 +376,6 @@ func (bc *blockController) load(suffix, path string) (b *block, err error) {
func (bc *blockController) close() {
for _, s := range bc.lst {
- s.close()
+ s.stopThenClose()
}
}
diff --git a/banyand/tsdb/shard_test.go b/banyand/tsdb/shard_test.go
index 6571d53..2c928c6 100644
--- a/banyand/tsdb/shard_test.go
+++ b/banyand/tsdb/shard_test.go
@@ -22,6 +22,7 @@ import (
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
+ "github.com/onsi/gomega/gleak"
"github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/banyand/tsdb"
@@ -35,8 +36,10 @@ var _ = Describe("Shard", func() {
var deferFn func()
var shard tsdb.Shard
var clock timestamp.MockClock
+ var goods []gleak.Goroutine
BeforeEach(func() {
+ goods = gleak.Goroutines()
var err error
tmp, deferFn, err = test.NewSpace()
Expect(err).NotTo(HaveOccurred())
@@ -48,6 +51,7 @@ var _ = Describe("Shard", func() {
GinkgoWriter.Printf("shard state:%+v \n", shard.State())
shard.Close()
deferFn()
+ Eventually(gleak.Goroutines).ShouldNot(gleak.HaveLeaked(goods))
})
It("generates several segments and blocks", func() {
By("open 4 blocks")
diff --git a/dist/LICENSE b/dist/LICENSE
index 08d2763..35af8dd 100644
--- a/dist/LICENSE
+++ b/dist/LICENSE
@@ -292,7 +292,7 @@ MIT licenses
github.com/json-iterator/go v1.1.12 MIT
github.com/mitchellh/mapstructure v1.5.0 MIT
github.com/onsi/ginkgo/v2 v2.1.4 MIT
- github.com/onsi/gomega v1.19.0 MIT
+ github.com/onsi/gomega v1.20.0 MIT
github.com/pelletier/go-toml/v2 v2.0.1 MIT
github.com/rs/zerolog v1.26.1 MIT
github.com/sirupsen/logrus v1.7.0 MIT
diff --git a/go.mod b/go.mod
index 1cecc88..aa1fb74 100644
--- a/go.mod
+++ b/go.mod
@@ -19,7 +19,7 @@ require (
github.com/klauspost/compress v1.15.6
github.com/oklog/run v1.1.0
github.com/onsi/ginkgo/v2 v2.1.4
- github.com/onsi/gomega v1.19.0
+ github.com/onsi/gomega v1.20.0
github.com/pkg/errors v0.9.1
github.com/prometheus/client_golang v1.12.2
github.com/rs/zerolog v1.26.1
diff --git a/go.sum b/go.sum
index b1b19e4..c617f2b 100644
--- a/go.sum
+++ b/go.sum
@@ -346,8 +346,8 @@ github.com/oklog/run v1.1.0/go.mod h1:sVPdnTZT1zYwAJeCMu2Th4T21pA3FPOQRfWjQlk7DV
github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U=
github.com/onsi/ginkgo/v2 v2.1.4 h1:GNapqRSid3zijZ9H77KrgVG4/8KqiyRsxcSxe+7ApXY=
github.com/onsi/ginkgo/v2 v2.1.4/go.mod h1:um6tUpWM/cxCK3/FK8BXqEiUMUwRgSM4JXG47RKZmLU=
-github.com/onsi/gomega v1.19.0 h1:4ieX6qQjPP/BfC3mpsAtIGGlxTWPeA3Inl/7DtXw1tw=
-github.com/onsi/gomega v1.19.0/go.mod h1:LY+I3pBVzYsTBU1AnDwOSxaYi9WoWiqgwooUqq9yPro=
+github.com/onsi/gomega v1.20.0 h1:8W0cWlwFkflGPLltQvLRB7ZVD5HuP6ng320w2IS245Q=
+github.com/onsi/gomega v1.20.0/go.mod h1:DtrZpjmvpn2mPm4YWQa0/ALMDj9v4YxLgojwPeREyVo=
github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=