Skip to content
Snippets Groups Projects
Commit a78cd3e2 authored by Bingyi Sun 【孙秉义】's avatar Bingyi Sun 【孙秉义】 Committed by yefu.chen
Browse files

Add proxy service timetick loop to expire allcoations


Signed-off-by: default avatarsunby <bingyi.sun@zilliz.com>
parent 35656ea9
No related branches found
No related tags found
No related merge requests found
......@@ -34,6 +34,7 @@ type ParamTable struct {
SegmentInfoChannelName string
DataServiceSubscriptionName string
K2SChannelNames []string
ProxyTimeTickChannelName string
SegmentFlushMetaPath string
Log log.Config
......@@ -73,6 +74,7 @@ func (p *ParamTable) Init() {
p.initK2SChannelNames()
p.initSegmentFlushMetaPath()
p.initLogCfg()
p.initProxyServiceTimeTickChannelName()
})
}
......@@ -243,3 +245,11 @@ func (p *ParamTable) initLogCfg() {
p.Log.File.Filename = ""
}
}
func (p *ParamTable) initProxyServiceTimeTickChannelName() {
ch, err := p.Load("msgChannel.chanNamePrefix.proxyServiceTimeTick")
if err != nil {
panic(err)
}
p.ProxyTimeTickChannelName = ch
}
......@@ -288,9 +288,10 @@ func (s *Server) checkMasterIsHealthy() error {
func (s *Server) startServerLoop() {
s.serverLoopCtx, s.serverLoopCancel = context.WithCancel(s.ctx)
s.serverLoopWg.Add(2)
s.serverLoopWg.Add(3)
go s.startStatsChannel(s.serverLoopCtx)
go s.startSegmentFlushChannel(s.serverLoopCtx)
go s.startProxyServiceTimeTickLoop(s.serverLoopCtx)
}
func (s *Server) startStatsChannel(ctx context.Context) {
......@@ -354,6 +355,36 @@ func (s *Server) startSegmentFlushChannel(ctx context.Context) {
}
}
func (s *Server) startProxyServiceTimeTickLoop(ctx context.Context) {
defer logutil.LogPanic()
defer s.serverLoopWg.Done()
flushStream, _ := s.msFactory.NewMsgStream(ctx)
flushStream.AsConsumer([]string{Params.ProxyTimeTickChannelName}, Params.DataServiceSubscriptionName)
flushStream.Start()
defer flushStream.Close()
for {
select {
case <-ctx.Done():
log.Debug("Proxy service timetick loop shut down")
default:
}
msgPack := flushStream.Consume()
s.allocMu.Lock()
for _, msg := range msgPack.Msgs {
if msg.Type() != commonpb.MsgType_TimeTick {
log.Warn("receive unknown msg from proxy service timetick", zap.Stringer("msgType", msg.Type()))
continue
}
tMsg := msg.(*msgstream.TimeTickMsg)
traceCtx := context.TODO()
if err := s.segAllocator.ExpireAllocations(traceCtx, tMsg.Base.Timestamp); err != nil {
log.Error("expire allocations error", zap.Error(err))
}
}
s.allocMu.Unlock()
}
}
func (s *Server) startDDChannel(ctx context.Context) {
defer s.serverLoopWg.Done()
ddStream, _ := s.msFactory.NewMsgStream(ctx)
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment