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

Fix DML channel AsProducer after Close (#16352)

Browse Source 
Signed-off-by: xiaofan-luan <xiaofan.luan@zilliz.com>
This commit is contained in:
Xiaofan 2022-04-02 13:09:29 +08:00 committed by GitHub
parent 5494f3c318
commit 7f7379d55f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 3 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
internal/rootcoord/dml_channels.go
View File

@ -62,6 +62,7 @@ func newDmlChannels(ctx context.Context, factory msgstream.Factory, chanNamePref
log.Error("Failed to add msgstream", zap.String("name", name), zap.Error(err)) log.Error("Failed to add msgstream", zap.String("name", name), zap.Error(err))
panic("Failed to add msgstream") panic("Failed to add msgstream")
} }
ms.AsProducer([]string{name})
d.pool.Store(name, &dmlMsgStream{ d.pool.Store(name, &dmlMsgStream{
ms: ms, ms: ms,
mutex: sync.RWMutex{}, mutex: sync.RWMutex{},
@ -159,9 +160,6 @@ func (d *dmlChannels) addChannels(names ...string) {
dms := v.(*dmlMsgStream) dms := v.(*dmlMsgStream)
dms.mutex.Lock() dms.mutex.Lock()
if dms.refcnt == 0 {
dms.ms.AsProducer([]string{name})
}
dms.refcnt++ dms.refcnt++
dms.mutex.Unlock() dms.mutex.Unlock()
} }
@ -180,9 +178,8 @@ func (d *dmlChannels) removeChannels(names ...string) {
dms.mutex.Lock() dms.mutex.Lock()
if dms.refcnt > 0 { if dms.refcnt > 0 {
dms.refcnt-- dms.refcnt--
if dms.refcnt == 0 { } else {
dms.ms.Close() log.Warn("Try to remove channel with no ref count", zap.String("channel name", name))
}
} }
dms.mutex.Unlock() dms.mutex.Unlock()
} }