admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2026-01-07 19:31:51 +08:00
Code Issues Packages Projects Releases Wiki Activity

enhance: Fast close msgdispatcher target (#34803)

Browse Source 
/kind improvement

issue: https://github.com/milvus-io/milvus/issues/34075

Signed-off-by: bigsheeper <yihao.dai@zilliz.com>
This commit is contained in:
yihao.dai 2024-07-29 00:25:49 +08:00 committed by GitHub
parent f77f5364b2
commit 4a3d98d88c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 15 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
pkg/mq/msgdispatcher/dispatcher_test.go
View File

@ -28,6 +28,7 @@ import (
"github.com/milvus-io/milvus/pkg/mq/common"
"github.com/milvus-io/milvus/pkg/mq/msgstream"
"github.com/milvus-io/milvus/pkg/util/lifetime"
)
func TestDispatcher(t *testing.T) {
@ -77,11 +78,13 @@ func TestDispatcher(t *testing.T) {
vchannel: "mock_vchannel_0",
pos: nil,
ch: output,
cancelCh: lifetime.NewSafeChan(),
})
d.AddTarget(&target{
vchannel: "mock_vchannel_1",
pos: nil,
ch: nil,
cancelCh: lifetime.NewSafeChan(),
})
num := d.TargetNum()
assert.Equal(t, 2, num)
@ -110,6 +113,7 @@ func TestDispatcher(t *testing.T) {
vchannel: "mock_vchannel_0",
pos: nil,
ch: output,
cancelCh: lifetime.NewSafeChan(),
}
assert.Equal(t, cap(output), cap(target.ch))
wg := &sync.WaitGroup{}

11
pkg/mq/msgdispatcher/target.go
View File

@ -21,6 +21,10 @@ import (
"sync"
"time"
"go.uber.org/zap"
"github.com/milvus-io/milvus/pkg/log"
"github.com/milvus-io/milvus/pkg/util/lifetime"
"github.com/milvus-io/milvus/pkg/util/paramtable"
)
@ -32,6 +36,8 @@ type target struct {
closeMu sync.Mutex
closeOnce sync.Once
closed bool
cancelCh lifetime.SafeChan
}
func newTarget(vchannel string, pos *Pos) *target {
@ -39,12 +45,14 @@ func newTarget(vchannel string, pos *Pos) *target {
vchannel: vchannel,
ch: make(chan *MsgPack, paramtable.Get().MQCfg.TargetBufSize.GetAsInt()),
pos: pos,
cancelCh: lifetime.NewSafeChan(),
}
t.closed = false
return t
}
func (t *target) close() {
t.cancelCh.Close()
t.closeMu.Lock()
defer t.closeMu.Unlock()
t.closeOnce.Do(func() {
@ -61,6 +69,9 @@ func (t *target) send(pack *MsgPack) error {
}
maxTolerantLag := paramtable.Get().MQCfg.MaxTolerantLag.GetAsDuration(time.Second)
select {
case <-t.cancelCh.CloseCh():
log.Info("target closed", zap.String("vchannel", t.vchannel))
return nil
case <-time.After(maxTolerantLag):
return fmt.Errorf("send target timeout, vchannel=%s, timeout=%s", t.vchannel, maxTolerantLag)
case t.ch <- pack: