admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2025-12-07 09:38:39 +08:00
Code Issues Packages Projects Releases Wiki Activity

enhance: Close component in topological order when unsub channel (#40796)

Browse Source 
Related to #40795

---------

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>
This commit is contained in:
congqixia 2025-03-21 10:22:11 +08:00 committed by GitHub
parent d7df78a6c9
commit 9824615efb
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 11 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
internal/querynodev2/services.go
View File

@ -357,10 +357,11 @@ func (node *QueryNode) UnsubDmChannel(ctx context.Context, req *querypb.UnsubDmC
defer node.unsubscribingChannels.Remove(req.GetChannelName()) defer node.unsubscribingChannels.Remove(req.GetChannelName())
delegator, ok := node.delegators.GetAndRemove(req.GetChannelName()) delegator, ok := node.delegators.GetAndRemove(req.GetChannelName())
if ok { if ok {
node.pipelineManager.Remove(req.GetChannelName())
// close the delegator first to block all coming query/search requests // close the delegator first to block all coming query/search requests
delegator.Close() delegator.Close()
node.pipelineManager.Remove(req.GetChannelName())
node.manager.Segment.RemoveBy(ctx, segments.WithChannel(req.GetChannelName()), segments.WithType(segments.SegmentTypeGrowing)) node.manager.Segment.RemoveBy(ctx, segments.WithChannel(req.GetChannelName()), segments.WithType(segments.SegmentTypeGrowing))
node.manager.Collection.Unref(req.GetCollectionID(), 1) node.manager.Collection.Unref(req.GetCollectionID(), 1)
} }

11
internal/util/pipeline/stream_pipeline.go
View File

@ -67,7 +67,11 @@ func (p *streamPipeline) work() {
case <-p.closeCh: case <-p.closeCh:
log.Ctx(context.TODO()).Debug("stream pipeline input closed") log.Ctx(context.TODO()).Debug("stream pipeline input closed")
return return
case msg := <-p.input: case msg, ok := <-p.input:
if !ok {
log.Ctx(context.TODO()).Debug("stream pipeline input closed")
return
}
p.lastAccessTime.Store(time.Now()) p.lastAccessTime.Store(time.Now())
log.Ctx(context.TODO()).RatedDebug(10, "stream pipeline fetch msg", zap.Int("sum", len(msg.Msgs))) log.Ctx(context.TODO()).RatedDebug(10, "stream pipeline fetch msg", zap.Int("sum", len(msg.Msgs)))
p.pipeline.inputChannel <- msg p.pipeline.inputChannel <- msg
@ -158,12 +162,15 @@ func (p *streamPipeline) Start() error {
func (p *streamPipeline) Close() { func (p *streamPipeline) Close() {
p.closeOnce.Do(func() { p.closeOnce.Do(func() {
// close datasource first
p.dispatcher.Deregister(p.vChannel)
// close stream input
close(p.closeCh) close(p.closeCh)
p.closeWg.Wait() p.closeWg.Wait()
if p.scanner != nil { if p.scanner != nil {
p.scanner.Close() p.scanner.Close()
} }
p.dispatcher.Deregister(p.vChannel) // close the underline pipeline
p.pipeline.Close() p.pipeline.Close()
}) })
} }