milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix possible deadlock (#7428) 

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>
pull/7440/head
congqixia 2021-09-02 14:56:10 +08:00 committed by GitHub
parent b718bcb1c7
commit d3027c0d28
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 7 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
internal/msgstream/mq_msgstream.go
View File

@ -96,9 +96,9 @@ func (ms *mqMsgStream) AsProducer(channels []string) {
}
ms.producerLock.Lock()
defer ms.producerLock.Unlock()
ms.producers[channel] = pp
ms.producerChannels = append(ms.producerChannels, channel)
ms.producerLock.Unlock()
return nil
}
err := retry.Do(context.TODO(), fn, retry.Attempts(20), retry.Sleep(time.Millisecond*200))
@ -131,9 +131,9 @@ func (ms *mqMsgStream) AsConsumer(channels []string, subName string) {
}
ms.consumerLock.Lock()
defer ms.consumerLock.Unlock()
ms.consumers[channel] = pc
ms.consumerChannels = append(ms.consumerChannels, channel)
ms.consumerLock.Unlock()
return nil
}
err := retry.Do(context.TODO(), fn, retry.Attempts(20), retry.Sleep(time.Millisecond*200))
@ -243,15 +243,18 @@ func (ms *mqMsgStream) Produce(msgPack *MsgPack) error {
trace.InjectContextToPulsarMsgProperties(sp.Context(), msg.Properties)
ms.producerLock.Lock()
if err := ms.producers[channel].Send(
spanCtx,
msg,
); err != nil {
ms.producerLock.Unlock()
trace.LogError(sp, err)
sp.Finish()
return err
}
sp.Finish()
ms.producerLock.Unlock()
}
}
return nil
@ -285,6 +288,7 @@ func (ms *mqMsgStream) Broadcast(msgPack *MsgPack) error {
spanCtx,
msg,
); err != nil {
ms.producerLock.Unlock()
trace.LogError(sp, err)
sp.Finish()
return err
@ -494,8 +498,8 @@ func (ms *MqTtMsgStream) AsConsumer(channels []string, subName string) {
}
ms.consumerLock.Lock()
defer ms.consumerLock.Unlock()
ms.addConsumer(pc, channel)
ms.consumerLock.Unlock()
return nil
}
err := retry.Do(context.TODO(), fn, retry.Attempts(20), retry.Sleep(time.Millisecond*200))