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/11/14 07:47:32 UTC
[skywalking-banyandb] branch main updated: Fix stream's leaked goroutines (#214)
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 2c996a5 Fix stream's leaked goroutines (#214)
2c996a5 is described below
commit 2c996a597d26269ef6faf911771045c4d8c31b41
Author: Mike <84...@qq.com>
AuthorDate: Mon Nov 14 15:47:27 2022 +0800
Fix stream's leaked goroutines (#214)
* Fix stream's leaked goroutines
Signed-off-by: mikechengwei <84...@qq.com>
Co-authored-by: Gao Hongtao <ha...@gmail.com>
---
banyand/metadata/schema/etcd.go | 4 +++-
banyand/queue/local.go | 19 +++++++++++++++++--
banyand/queue/queue.go | 6 ++++--
banyand/stream/metadata_test.go | 4 ++++
banyand/tsdb/index/writer.go | 15 ++++++++++++++-
pkg/bus/bus.go | 23 ++++++++++++++++++++++-
6 files changed, 64 insertions(+), 7 deletions(-)
diff --git a/banyand/metadata/schema/etcd.go b/banyand/metadata/schema/etcd.go
index b640e21..83ad69d 100644
--- a/banyand/metadata/schema/etcd.go
+++ b/banyand/metadata/schema/etcd.go
@@ -83,6 +83,7 @@ func (eh *eventHandler) InterestOf(kind Kind) bool {
type etcdSchemaRegistry struct {
server *embed.Etcd
+ client *clientv3.Client
kv clientv3.KV
handlers []*eventHandler
mux sync.RWMutex
@@ -144,7 +145,7 @@ func (e *etcdSchemaRegistry) StoppingNotify() <-chan struct{} {
func (e *etcdSchemaRegistry) Close() error {
e.server.Close()
- return nil
+ return e.client.Close()
}
func NewEtcdSchemaRegistry(options ...RegistryOption) (Registry, error) {
@@ -173,6 +174,7 @@ func NewEtcdSchemaRegistry(options ...RegistryOption) (Registry, error) {
reg := &etcdSchemaRegistry{
server: e,
kv: kvClient,
+ client: client,
}
return reg, nil
}
diff --git a/banyand/queue/local.go b/banyand/queue/local.go
index 5703acf..60dd747 100644
--- a/banyand/queue/local.go
+++ b/banyand/queue/local.go
@@ -20,6 +20,7 @@ package queue
import (
"github.com/apache/skywalking-banyandb/banyand/discovery"
"github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/run"
)
var (
@@ -28,8 +29,22 @@ var (
)
type local struct {
- local *bus.Bus
- repo discovery.ServiceRepo
+ local *bus.Bus
+ repo discovery.ServiceRepo
+ stopCh chan struct{}
+}
+
+// GracefulStop implements Queue
+func (l *local) GracefulStop() {
+ l.local.Close()
+ if l.stopCh != nil {
+ close(l.stopCh)
+ }
+}
+
+// Serve implements Queue
+func (l *local) Serve() run.StopNotify {
+ return l.stopCh
}
func (l *local) Subscribe(topic bus.Topic, listener bus.MessageListener) error {
diff --git a/banyand/queue/queue.go b/banyand/queue/queue.go
index 1dbe542..7941a9b 100644
--- a/banyand/queue/queue.go
+++ b/banyand/queue/queue.go
@@ -30,11 +30,13 @@ type Queue interface {
run.Unit
bus.Subscriber
bus.Publisher
+ run.Service
}
func NewQueue(_ context.Context, repo discovery.ServiceRepo) (Queue, error) {
return &local{
- repo: repo,
- local: bus.NewBus(),
+ repo: repo,
+ local: bus.NewBus(),
+ stopCh: make(chan struct{}),
}, nil
}
diff --git a/banyand/stream/metadata_test.go b/banyand/stream/metadata_test.go
index 7a70cf5..04b2260 100644
--- a/banyand/stream/metadata_test.go
+++ b/banyand/stream/metadata_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/event"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
@@ -33,13 +34,16 @@ import (
var _ = Describe("Metadata", func() {
var svcs *services
var deferFn func()
+ var goods []gleak.Goroutine
BeforeEach(func() {
+ goods = gleak.Goroutines()
svcs, deferFn = setUp()
})
AfterEach(func() {
deferFn()
+ Eventually(gleak.Goroutines).ShouldNot(gleak.HaveLeaked(goods))
})
Context("Manage group", func() {
diff --git a/banyand/tsdb/index/writer.go b/banyand/tsdb/index/writer.go
index 95dcf48..5e02ebc 100644
--- a/banyand/tsdb/index/writer.go
+++ b/banyand/tsdb/index/writer.go
@@ -33,6 +33,7 @@ import (
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/partition"
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/run"
)
type CallbackFn func()
@@ -72,6 +73,7 @@ type Writer struct {
enableGlobalIndex bool
ch chan Message
invertRuleIndex map[byte][]*partition.IndexRuleLocator
+ closer *run.Closer
}
func NewWriter(ctx context.Context, options WriterOptions) *Writer {
@@ -111,16 +113,27 @@ func NewWriter(ctx context.Context, options WriterOptions) *Writer {
}
w.ch = make(chan Message)
w.bootIndexGenerator()
+ w.closer = run.NewCloser(0)
return w
}
func (s *Writer) Write(value Message) {
go func(m Message) {
- s.ch <- m
+ if !s.closer.AddRunning() {
+ return
+ }
+ defer s.closer.Done()
+ select {
+ case <-s.closer.CloseNotify():
+ return
+ case s.ch <- m:
+ }
}(value)
}
func (s *Writer) Close() error {
+ s.closer.CloseThenWait()
+ close(s.ch)
return nil
}
diff --git a/pkg/bus/bus.go b/pkg/bus/bus.go
index ded140e..75d66c2 100644
--- a/pkg/bus/bus.go
+++ b/pkg/bus/bus.go
@@ -23,6 +23,8 @@ import (
"sync"
"go.uber.org/multierr"
+
+ "github.com/apache/skywalking-banyandb/pkg/run"
)
// Payload represents a simple data
@@ -92,11 +94,13 @@ func BiTopic(ID string) Topic {
type Bus struct {
topics map[Topic][]Channel
mutex sync.RWMutex
+ closer *run.Closer
}
func NewBus() *Bus {
b := new(Bus)
b.topics = make(map[Topic][]Channel)
+ b.closer = run.NewCloser(0)
return b
}
@@ -175,9 +179,17 @@ func (b *Bus) Publish(topic Topic, message ...Message) (Future, error) {
for _, each := range cc {
for _, m := range message {
go func(ch Channel, message Message) {
- ch <- Event{
+ if !b.closer.AddRunning() {
+ return
+ }
+ defer b.closer.Done()
+ select {
+ case <-b.closer.CloseNotify():
+ return
+ case ch <- Event{
m: message,
f: f,
+ }:
}
}(each, m)
}
@@ -223,3 +235,12 @@ func (b *Bus) Subscribe(topic Topic, listener MessageListener) error {
}(listener, ch)
return nil
}
+
+func (b *Bus) Close() {
+ b.closer.CloseThenWait()
+ for _, chs := range b.topics {
+ for _, ch := range chs {
+ close(ch)
+ }
+ }
+}