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

Add Buffer for consumer channel (#9578) 

Signed-off-by: xiaofan-luan <xiaofan.luan@zilliz.com>
pull/10000/head
Xiaofan 2021-10-15 20:31:16 +08:00 committed by GitHub
parent da5195aab8
commit d845153de4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 6 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
internal/datanode/flow_graph_dmstream_input_node.go
View File

@ -19,6 +19,7 @@ package datanode
import ( import (
"context" "context"
"fmt" "fmt"
"time"
"github.com/milvus-io/milvus/internal/log" "github.com/milvus-io/milvus/internal/log"
"github.com/milvus-io/milvus/internal/proto/internalpb" "github.com/milvus-io/milvus/internal/proto/internalpb"
@ -47,11 +48,13 @@ func newDmInputNode(ctx context.Context, seekPos *internalpb.MsgPosition, dmNode
if seekPos != nil { if seekPos != nil {
seekPos.ChannelName = pchannelName seekPos.ChannelName = pchannelName
log.Debug("datanode Seek", zap.String("channelName", seekPos.GetChannelName())) start := time.Now()
log.Debug("datanode begin to seek: " + seekPos.GetChannelName())
err = insertStream.Seek([]*internalpb.MsgPosition{seekPos}) err = insertStream.Seek([]*internalpb.MsgPosition{seekPos})
if err != nil { if err != nil {
return nil, err return nil, err
} }
log.Debug("datanode Seek successfully: "+seekPos.GetChannelName(), zap.Int64("elapse ", time.Since(start).Milliseconds()))
} }
node := flowgraph.NewInputNode(insertStream, "dmInputNode", dmNodeConfig.maxQueueLength, dmNodeConfig.maxParallelism) node := flowgraph.NewInputNode(insertStream, "dmInputNode", dmNodeConfig.maxQueueLength, dmNodeConfig.maxParallelism)

5
internal/msgstream/mq_msgstream.go
View File

@ -874,11 +874,6 @@ func (ms *MqTtMsgStream) Seek(msgPositions []*internalpb.MsgPosition) error {
} }
ms.addConsumer(consumer, mp.ChannelName) ms.addConsumer(consumer, mp.ChannelName)
//TODO: May cause problem
//if len(consumer.Chan()) == 0 {
// return nil
//}
runLoop := true runLoop := true
for runLoop { for runLoop {
select { select {

2
internal/util/mqclient/pulsar_consumer.go
View File

@ -34,7 +34,7 @@ func (pc *pulsarConsumer) Subscription() string {
func (pc *pulsarConsumer) Chan() <-chan ConsumerMessage { func (pc *pulsarConsumer) Chan() <-chan ConsumerMessage {
if pc.msgChannel == nil { if pc.msgChannel == nil {
pc.once.Do(func() { pc.once.Do(func() {
pc.msgChannel = make(chan ConsumerMessage) pc.msgChannel = make(chan ConsumerMessage, 256)
// this part handles msgstream expectation when the consumer is not seeked // this part handles msgstream expectation when the consumer is not seeked
// pulsar's default behavior is setting postition to the earliest pointer when client of the same subscription pointer is not acked // pulsar's default behavior is setting postition to the earliest pointer when client of the same subscription pointer is not acked
// yet, our message stream is to setting to the very start point of the topic // yet, our message stream is to setting to the very start point of the topic

2
internal/util/mqclient/rmq_consumer.go
View File

@ -34,7 +34,7 @@ func (rc *RmqConsumer) Subscription() string {
func (rc *RmqConsumer) Chan() <-chan ConsumerMessage { func (rc *RmqConsumer) Chan() <-chan ConsumerMessage {
if rc.msgChannel == nil { if rc.msgChannel == nil {
rc.once.Do(func() { rc.once.Do(func() {
rc.msgChannel = make(chan ConsumerMessage) rc.msgChannel = make(chan ConsumerMessage, 256)
go func() { go func() {
for { //nolint:gosimple for { //nolint:gosimple
select { select {