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

Remove session from timetickSync (#15255)

Browse Source 
Signed-off-by: yudong.cai <yudong.cai@zilliz.com>
This commit is contained in:
Cai Yudong 2022-01-18 00:07:37 +08:00 committed by GitHub
parent ac3e30dbab
commit 1b1d4e502b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 14 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
internal/rootcoord/root_coord.go
View File

@ -1032,7 +1032,7 @@ func (c *Core) Init() error {
}
chanMap := c.MetaTable.ListCollectionPhysicalChannels()
c.chanTimeTick = newTimeTickSync(c.ctx, c.session, c.msFactory, chanMap)
c.chanTimeTick = newTimeTickSync(c.ctx, c.session.ServerID, c.msFactory, chanMap)
c.chanTimeTick.addProxy(c.session)
c.proxyClientManager = newProxyClientManager(c)
@ -1040,7 +1040,7 @@ func (c *Core) Init() error {
c.proxyManager = newProxyManager(
c.ctx,
c.etcdCli,
c.chanTimeTick.getProxy,
c.chanTimeTick.clearProxy,
c.proxyClientManager.GetProxyClients,
)
c.proxyManager.AddSession(c.chanTimeTick.addProxy, c.proxyClientManager.AddProxyClient)

18
internal/rootcoord/timeticksync.go
View File

@ -44,8 +44,8 @@ var (
)
type timetickSync struct {
ctx context.Context
session *sessionutil.Session
ctx context.Context
sourceID int64
dmlChannels *dmlChannels // used for insert
deltaChannels *dmlChannels // used for delete
@ -85,7 +85,7 @@ func (c *chanTsMsg) getTimetick(channelName string) typeutil.Timestamp {
return c.defaultTs
}
func newTimeTickSync(ctx context.Context, session *sessionutil.Session, factory msgstream.Factory, chanMap map[typeutil.UniqueID][]string) *timetickSync {
func newTimeTickSync(ctx context.Context, sourceID int64, factory msgstream.Factory, chanMap map[typeutil.UniqueID][]string) *timetickSync {
// initialize dml channels used for insert
dmlChannels := newDmlChannels(ctx, factory, Params.RootCoordCfg.DmlChannelName, Params.RootCoordCfg.DmlChannelNum)
// initialize delta channels used for delete, share Params.DmlChannelNum with dmlChannels
@ -110,8 +110,8 @@ func newTimeTickSync(ctx context.Context, session *sessionutil.Session, factory
}
return &timetickSync{
ctx: ctx,
session: session,
ctx: ctx,
sourceID: sourceID,
dmlChannels: dmlChannels,
deltaChannels: deltaChannels,
@ -236,7 +236,7 @@ func (t *timetickSync) updateTimeTick(in *internalpb.ChannelTimeTickMsg, reason
return nil
}
if in.Base.SourceID == t.session.ServerID {
if in.Base.SourceID == t.sourceID {
if prev != nil && in.DefaultTimestamp <= prev.defaultTs {
log.Debug("timestamp go back", zap.Int64("source id", in.Base.SourceID),
zap.Uint64("curr ts", in.DefaultTimestamp),
@ -273,7 +273,7 @@ func (t *timetickSync) delProxy(sess *sessionutil.Session) {
}
}
func (t *timetickSync) getProxy(sess []*sessionutil.Session) {
func (t *timetickSync) clearProxy(sess []*sessionutil.Session) {
t.lock.Lock()
defer t.lock.Unlock()
for _, s := range sess {
@ -304,7 +304,7 @@ func (t *timetickSync) startWatch(wg *sync.WaitGroup) {
}
// reduce each channel to get min timestamp
local := proxyTimetick[t.session.ServerID]
local := proxyTimetick[t.sourceID]
if len(local.chanTs) == 0 {
continue
}
@ -356,7 +356,7 @@ func (t *timetickSync) sendTimeTickToChannel(chanNames []string, ts typeutil.Tim
MsgType: commonpb.MsgType_TimeTick,
MsgID: 0,
Timestamp: ts,
SourceID: t.session.ServerID,
SourceID: t.sourceID,
},
}
timeTickMsg := &msgstream.TimeTickMsg{

10
internal/rootcoord/timeticksync_test.go
View File

@ -26,15 +26,11 @@ import (
"github.com/milvus-io/milvus/internal/msgstream"
"github.com/milvus-io/milvus/internal/proto/commonpb"
"github.com/milvus-io/milvus/internal/proto/internalpb"
"github.com/milvus-io/milvus/internal/util/sessionutil"
)
func TestTimetickSync(t *testing.T) {
ctx := context.Background()
session := &sessionutil.Session{
ServerID: 100,
}
sourceID := int64(100)
factory := msgstream.NewPmsFactory()
m := map[string]interface{}{
@ -51,7 +47,7 @@ func TestTimetickSync(t *testing.T) {
Params.RootCoordCfg.DmlChannelNum = 2
Params.RootCoordCfg.DmlChannelName = "rootcoord-dml"
Params.RootCoordCfg.DeltaChannelName = "rootcoord-delta"
ttSync := newTimeTickSync(ctx, session, factory, nil)
ttSync := newTimeTickSync(ctx, sourceID, factory, nil)
var wg sync.WaitGroup
wg.Add(1)
@ -108,7 +104,7 @@ func TestTimetickSync(t *testing.T) {
assert.Nil(t, err)
ttSync.ddlMinTs = uint64(300)
ttSync.session.ServerID = int64(1)
ttSync.sourceID = int64(1)
err = ttSync.updateTimeTick(msg, "1")
assert.Nil(t, err)
})