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

Rename timetickSync APIs and variables for better readability (#15264)

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

8
internal/rootcoord/root_coord.go
View File

@ -1036,18 +1036,18 @@ func (c *Core) Init() error {
chanMap := c.MetaTable.ListCollectionPhysicalChannels() chanMap := c.MetaTable.ListCollectionPhysicalChannels()
c.chanTimeTick = newTimeTickSync(c.ctx, c.session.ServerID, c.msFactory, chanMap) c.chanTimeTick = newTimeTickSync(c.ctx, c.session.ServerID, c.msFactory, chanMap)
c.chanTimeTick.addProxy(c.session) c.chanTimeTick.addSession(c.session)
c.proxyClientManager = newProxyClientManager(c) c.proxyClientManager = newProxyClientManager(c)
log.Debug("RootCoord, set proxy manager") log.Debug("RootCoord, set proxy manager")
c.proxyManager = newProxyManager( c.proxyManager = newProxyManager(
c.ctx, c.ctx,
c.etcdCli, c.etcdCli,
c.chanTimeTick.clearProxy, c.chanTimeTick.clearSessions,
c.proxyClientManager.GetProxyClients, c.proxyClientManager.GetProxyClients,
) )
c.proxyManager.AddSession(c.chanTimeTick.addProxy, c.proxyClientManager.AddProxyClient) c.proxyManager.AddSession(c.chanTimeTick.addSession, c.proxyClientManager.AddProxyClient)
c.proxyManager.DelSession(c.chanTimeTick.delProxy, c.proxyClientManager.DelProxyClient) c.proxyManager.DelSession(c.chanTimeTick.delSession, c.proxyClientManager.DelProxyClient)
c.metricsCacheManager = metricsinfo.NewMetricsCacheManager() c.metricsCacheManager = metricsinfo.NewMetricsCacheManager()

12
internal/rootcoord/root_coord_test.go
View File

@ -756,11 +756,11 @@ func TestRootCoord(t *testing.T) {
//assert.True(t, ok) //assert.True(t, ok)
//assert.Greater(t, ddm.Base.Timestamp, uint64(0)) //assert.Greater(t, ddm.Base.Timestamp, uint64(0))
core.chanTimeTick.lock.Lock() core.chanTimeTick.lock.Lock()
assert.Equal(t, len(core.chanTimeTick.proxyTimeTick), 2) assert.Equal(t, len(core.chanTimeTick.sess2ChanTsMap), 2)
pt, ok := core.chanTimeTick.proxyTimeTick[core.session.ServerID] pt, ok := core.chanTimeTick.sess2ChanTsMap[core.session.ServerID]
assert.True(t, ok) assert.True(t, ok)
assert.Equal(t, shardsNum, int32(len(pt.chanTs))) assert.Equal(t, shardsNum, int32(len(pt.chanTsMap)))
for chanName, ts := range pt.chanTs { for chanName, ts := range pt.chanTsMap {
assert.Contains(t, createMeta.PhysicalChannelNames, chanName) assert.Contains(t, createMeta.PhysicalChannelNames, chanName)
assert.Equal(t, pt.defaultTs, ts) assert.Equal(t, pt.defaultTs, ts)
} }
@ -1838,7 +1838,7 @@ func TestRootCoord(t *testing.T) {
s, _ := core.UpdateChannelTimeTick(ctx, msg0) s, _ := core.UpdateChannelTimeTick(ctx, msg0)
assert.Equal(t, commonpb.ErrorCode_Success, s.ErrorCode) assert.Equal(t, commonpb.ErrorCode_Success, s.ErrorCode)
time.Sleep(100 * time.Millisecond) time.Sleep(100 * time.Millisecond)
//t.Log(core.chanTimeTick.proxyTimeTick) //t.Log(core.chanTimeTick.sess2ChanTsMap)
msg1 := &internalpb.ChannelTimeTickMsg{ msg1 := &internalpb.ChannelTimeTickMsg{
Base: &commonpb.MsgBase{ Base: &commonpb.MsgBase{
@ -1865,7 +1865,7 @@ func TestRootCoord(t *testing.T) {
time.Sleep(100 * time.Millisecond) time.Sleep(100 * time.Millisecond)
// 2 proxy, 1 rootcoord // 2 proxy, 1 rootcoord
assert.Equal(t, 3, core.chanTimeTick.getProxyNum()) assert.Equal(t, 3, core.chanTimeTick.getSessionNum())
// add 3 proxy channels // add 3 proxy channels
assert.Equal(t, 3, core.chanTimeTick.getDmlChannelNum()-numChan) assert.Equal(t, 3, core.chanTimeTick.getDmlChannelNum()-numChan)

84
internal/rootcoord/timeticksync.go
View File

@ -45,13 +45,13 @@ var (
type timetickSync struct { type timetickSync struct {
ctx context.Context ctx context.Context
sourceID int64 sourceID typeutil.UniqueID
dmlChannels *dmlChannels // used for insert dmlChannels *dmlChannels // used for insert
deltaChannels *dmlChannels // used for delete deltaChannels *dmlChannels // used for delete
lock sync.Mutex lock sync.Mutex
proxyTimeTick map[typeutil.UniqueID]*chanTsMsg sess2ChanTsMap map[typeutil.UniqueID]*chanTsMsg
sendChan chan map[typeutil.UniqueID]*chanTsMsg sendChan chan map[typeutil.UniqueID]*chanTsMsg
// record ddl timetick info // record ddl timetick info
@ -61,25 +61,25 @@ type timetickSync struct {
} }
type chanTsMsg struct { type chanTsMsg struct {
chanTs map[string]typeutil.Timestamp chanTsMap map[string]typeutil.Timestamp
defaultTs typeutil.Timestamp defaultTs typeutil.Timestamp
cnt int64 cnt int64
} }
func newChanTsMsg(in *internalpb.ChannelTimeTickMsg, cnt int64) *chanTsMsg { func newChanTsMsg(in *internalpb.ChannelTimeTickMsg, cnt int64) *chanTsMsg {
msg := &chanTsMsg{ msg := &chanTsMsg{
chanTs: make(map[string]typeutil.Timestamp), chanTsMap: make(map[string]typeutil.Timestamp),
defaultTs: in.DefaultTimestamp, defaultTs: in.DefaultTimestamp,
cnt: cnt, cnt: cnt,
} }
for idx := range in.ChannelNames { for idx := range in.ChannelNames {
msg.chanTs[in.ChannelNames[idx]] = in.Timestamps[idx] msg.chanTsMap[in.ChannelNames[idx]] = in.Timestamps[idx]
} }
return msg return msg
} }
func (c *chanTsMsg) getTimetick(channelName string) typeutil.Timestamp { func (c *chanTsMsg) getTimetick(channelName string) typeutil.Timestamp {
if ts, ok := c.chanTs[channelName]; ok { if ts, ok := c.chanTsMap[channelName]; ok {
return ts return ts
} }
return c.defaultTs return c.defaultTs
@ -117,7 +117,7 @@ func newTimeTickSync(ctx context.Context, sourceID int64, factory msgstream.Fact
deltaChannels: deltaChannels, deltaChannels: deltaChannels,
lock: sync.Mutex{}, lock: sync.Mutex{},
proxyTimeTick: make(map[typeutil.UniqueID]*chanTsMsg), sess2ChanTsMap: make(map[typeutil.UniqueID]*chanTsMsg),
sendChan: make(chan map[typeutil.UniqueID]*chanTsMsg, 16), sendChan: make(chan map[typeutil.UniqueID]*chanTsMsg, 16),
ddlLock: sync.RWMutex{}, ddlLock: sync.RWMutex{},
@ -129,16 +129,16 @@ func newTimeTickSync(ctx context.Context, sourceID int64, factory msgstream.Fact
// sendToChannel send all channels' timetick to sendChan // sendToChannel send all channels' timetick to sendChan
// lock is needed by the invoker // lock is needed by the invoker
func (t *timetickSync) sendToChannel() { func (t *timetickSync) sendToChannel() {
if len(t.proxyTimeTick) == 0 { if len(t.sess2ChanTsMap) == 0 {
return return
} }
// detect whether rootcoord receives ttMsg from all proxy nodes // detect whether rootcoord receives ttMsg from all source sessions
maxCnt := int64(0) maxCnt := int64(0)
idleProxyList := make([]typeutil.UniqueID, 0, len(t.proxyTimeTick)) idleSessionList := make([]typeutil.UniqueID, 0, len(t.sess2ChanTsMap))
for id, v := range t.proxyTimeTick { for id, v := range t.sess2ChanTsMap {
if v == nil { if v == nil {
idleProxyList = append(idleProxyList, id) idleSessionList = append(idleSessionList, id)
} else { } else {
if maxCnt < v.cnt { if maxCnt < v.cnt {
maxCnt = v.cnt maxCnt = v.cnt
@ -146,20 +146,20 @@ func (t *timetickSync) sendToChannel() {
} }
} }
if len(idleProxyList) > 0 { if len(idleSessionList) > 0 {
// give warning every 2 second if not get ttMsg from proxy nodes // give warning every 2 second if not get ttMsg from source sessions
if maxCnt%10 == 0 { if maxCnt%10 == 0 {
log.Warn("proxy idle for long time", zap.Any("proxy list", idleProxyList), log.Warn("session idle for long time", zap.Any("idle session list", idleSessionList),
zap.Any("idle time", Params.ProxyCfg.TimeTickInterval.Milliseconds()*maxCnt)) zap.Any("idle time", Params.ProxyCfg.TimeTickInterval.Milliseconds()*maxCnt))
} }
return return
} }
// clear proxyTimeTick and send a clone // clear sess2ChanTsMap and send a clone
ptt := make(map[typeutil.UniqueID]*chanTsMsg) ptt := make(map[typeutil.UniqueID]*chanTsMsg)
for k, v := range t.proxyTimeTick { for k, v := range t.sess2ChanTsMap {
ptt[k] = v ptt[k] = v
t.proxyTimeTick[k] = nil t.sess2ChanTsMap[k] = nil
} }
t.sendChan <- ptt t.sendChan <- ptt
} }
@ -221,9 +221,9 @@ func (t *timetickSync) updateTimeTick(in *internalpb.ChannelTimeTickMsg, reason
return fmt.Errorf("invalid TimeTickMsg") return fmt.Errorf("invalid TimeTickMsg")
} }
prev, ok := t.proxyTimeTick[in.Base.SourceID] prev, ok := t.sess2ChanTsMap[in.Base.SourceID]
if !ok { if !ok {
return fmt.Errorf("skip ChannelTimeTickMsg from un-recognized proxy node %d", in.Base.SourceID) return fmt.Errorf("skip ChannelTimeTickMsg from un-recognized session %d", in.Base.SourceID)
} }
// if ddl operation not finished, skip current ts update // if ddl operation not finished, skip current ts update
@ -247,41 +247,41 @@ func (t *timetickSync) updateTimeTick(in *internalpb.ChannelTimeTickMsg, reason
} }
if prev == nil { if prev == nil {
t.proxyTimeTick[in.Base.SourceID] = newChanTsMsg(in, 1) t.sess2ChanTsMap[in.Base.SourceID] = newChanTsMsg(in, 1)
} else { } else {
t.proxyTimeTick[in.Base.SourceID] = newChanTsMsg(in, prev.cnt+1) t.sess2ChanTsMap[in.Base.SourceID] = newChanTsMsg(in, prev.cnt+1)
} }
t.sendToChannel() t.sendToChannel()
return nil return nil
} }
func (t *timetickSync) addProxy(sess *sessionutil.Session) { func (t *timetickSync) addSession(sess *sessionutil.Session) {
t.lock.Lock() t.lock.Lock()
defer t.lock.Unlock() defer t.lock.Unlock()
t.proxyTimeTick[sess.ServerID] = nil t.sess2ChanTsMap[sess.ServerID] = nil
log.Debug("Add proxy for timeticksync", zap.Int64("serverID", sess.ServerID)) log.Debug("Add session for timeticksync", zap.Int64("serverID", sess.ServerID))
} }
func (t *timetickSync) delProxy(sess *sessionutil.Session) { func (t *timetickSync) delSession(sess *sessionutil.Session) {
t.lock.Lock() t.lock.Lock()
defer t.lock.Unlock() defer t.lock.Unlock()
if _, ok := t.proxyTimeTick[sess.ServerID]; ok { if _, ok := t.sess2ChanTsMap[sess.ServerID]; ok {
delete(t.proxyTimeTick, sess.ServerID) delete(t.sess2ChanTsMap, sess.ServerID)
log.Debug("Remove proxy from timeticksync", zap.Int64("serverID", sess.ServerID)) log.Debug("Remove session from timeticksync", zap.Int64("serverID", sess.ServerID))
t.sendToChannel() t.sendToChannel()
} }
} }
func (t *timetickSync) clearProxy(sess []*sessionutil.Session) { func (t *timetickSync) clearSessions(sess []*sessionutil.Session) {
t.lock.Lock() t.lock.Lock()
defer t.lock.Unlock() defer t.lock.Unlock()
for _, s := range sess { for _, s := range sess {
t.proxyTimeTick[s.ServerID] = nil t.sess2ChanTsMap[s.ServerID] = nil
} }
} }
// StartWatch watch proxy node change and process all channels' timetick msg // StartWatch watch session change and process all channels' timetick msg
func (t *timetickSync) startWatch(wg *sync.WaitGroup) { func (t *timetickSync) startWatch(wg *sync.WaitGroup) {
defer wg.Done() defer wg.Done()
@ -297,15 +297,15 @@ func (t *timetickSync) startWatch(wg *sync.WaitGroup) {
case <-t.ctx.Done(): case <-t.ctx.Done():
log.Debug("rootcoord context done", zap.Error(t.ctx.Err())) log.Debug("rootcoord context done", zap.Error(t.ctx.Err()))
return return
case proxyTimetick, ok := <-t.sendChan: case sessTimetick, ok := <-t.sendChan:
if !ok { if !ok {
log.Debug("timetickSync sendChan closed") log.Debug("timetickSync sendChan closed")
return return
} }
// reduce each channel to get min timestamp // reduce each channel to get min timestamp
local := proxyTimetick[t.sourceID] local := sessTimetick[t.sourceID]
if len(local.chanTs) == 0 { if len(local.chanTsMap) == 0 {
continue continue
} }
@ -313,14 +313,14 @@ func (t *timetickSync) startWatch(wg *sync.WaitGroup) {
checker.Check() checker.Check()
} }
hdr := fmt.Sprintf("send ts to %d channels", len(local.chanTs)) hdr := fmt.Sprintf("send ts to %d channels", len(local.chanTsMap))
tr := timerecord.NewTimeRecorder(hdr) tr := timerecord.NewTimeRecorder(hdr)
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
for chanName, ts := range local.chanTs { for chanName, ts := range local.chanTsMap {
wg.Add(1) wg.Add(1)
go func(chanName string, ts typeutil.Timestamp) { go func(chanName string, ts typeutil.Timestamp) {
mints := ts mints := ts
for _, tt := range proxyTimetick { for _, tt := range sessTimetick {
currTs := tt.getTimetick(chanName) currTs := tt.getTimetick(chanName)
if currTs < mints { if currTs < mints {
mints = currTs mints = currTs
@ -337,7 +337,7 @@ func (t *timetickSync) startWatch(wg *sync.WaitGroup) {
// rootcoord send tt msg to all channels every 200ms by default // rootcoord send tt msg to all channels every 200ms by default
if span > Params.ProxyCfg.TimeTickInterval { if span > Params.ProxyCfg.TimeTickInterval {
log.Warn("rootcoord send tt to all channels too slowly", log.Warn("rootcoord send tt to all channels too slowly",
zap.Int("chanNum", len(local.chanTs)), zap.Int64("span", span.Milliseconds())) zap.Int("chanNum", len(local.chanTsMap)), zap.Int64("span", span.Milliseconds()))
} }
} }
} }
@ -375,11 +375,11 @@ func (t *timetickSync) sendTimeTickToChannel(chanNames []string, ts typeutil.Tim
return nil return nil
} }
// GetProxyNum return the num of detected proxy node // GetSessionNum return the num of detected sessions
func (t *timetickSync) getProxyNum() int { func (t *timetickSync) getSessionNum() int {
t.lock.Lock() t.lock.Lock()
defer t.lock.Unlock() defer t.lock.Unlock()
return len(t.proxyTimeTick) return len(t.sess2ChanTsMap)
} }
/////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////

6
internal/rootcoord/timeticksync_test.go
View File

@ -55,7 +55,7 @@ func TestTimetickSync(t *testing.T) {
defer wg.Done() defer wg.Done()
ttSync.sendToChannel() ttSync.sendToChannel()
ttSync.proxyTimeTick[1] = nil ttSync.sess2ChanTsMap[1] = nil
ttSync.sendToChannel() ttSync.sendToChannel()
msg := &internalpb.ChannelTimeTickMsg{ msg := &internalpb.ChannelTimeTickMsg{
@ -63,7 +63,7 @@ func TestTimetickSync(t *testing.T) {
MsgType: commonpb.MsgType_TimeTick, MsgType: commonpb.MsgType_TimeTick,
}, },
} }
ttSync.proxyTimeTick[1] = newChanTsMsg(msg, 1) ttSync.sess2ChanTsMap[1] = newChanTsMsg(msg, 1)
ttSync.sendToChannel() ttSync.sendToChannel()
}) })
@ -97,7 +97,7 @@ func TestTimetickSync(t *testing.T) {
msg.Timestamps = append(msg.Timestamps, uint64(2)) msg.Timestamps = append(msg.Timestamps, uint64(2))
msg.DefaultTimestamp = uint64(200) msg.DefaultTimestamp = uint64(200)
cttMsg := newChanTsMsg(msg, 1) cttMsg := newChanTsMsg(msg, 1)
ttSync.proxyTimeTick[msg.Base.SourceID] = cttMsg ttSync.sess2ChanTsMap[msg.Base.SourceID] = cttMsg
ttSync.ddlMinTs = uint64(100) ttSync.ddlMinTs = uint64(100)
err = ttSync.updateTimeTick(msg, "1") err = ttSync.updateTimeTick(msg, "1")