test/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2024-11-30 02:48:45 +08:00
Code Issues Actions 11 Packages Projects Releases Wiki Activity

Fix RootCoord double updates TSO (#22715)

Browse Source 
Signed-off-by: yah01 <yang.cen@zilliz.com>
This commit is contained in:
yah01 2023-03-13 18:01:53 +08:00 committed by GitHub
parent a4031da634
commit fa2c1c1f40
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 7 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
internal/proxy/timestamp.go
View File

@ -47,7 +47,7 @@ func newTimestampAllocator(tso timestampAllocatorInterface, peerID UniqueID) (*t
func (ta *timestampAllocator) alloc(ctx context.Context, count uint32) ([]Timestamp, error) {
tr := timerecord.NewTimeRecorder("applyTimestamp")
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
req := &rootcoordpb.AllocTimestampRequest{
Base: commonpbutil.NewMsgBase(

7
internal/rootcoord/root_coord.go
View File

@ -238,15 +238,12 @@ func (c *Core) tsLoop() {
select {
case <-tsoTicker.C:
if err := c.tsoAllocator.UpdateTSO(); err != nil {
log.Warn("failed to update timestamp: ", zap.Error(err))
log.Warn("failed to update tso", zap.Error(err))
continue
}
ts := c.tsoAllocator.GetLastSavedTime()
metrics.RootCoordTimestampSaved.Set(float64(ts.Unix()))
if err := c.tsoAllocator.UpdateTSO(); err != nil {
log.Warn("failed to update id: ", zap.Error(err))
continue
}
case <-ctx.Done():
return
}

8
internal/tso/global_allocator.go
View File

@ -30,13 +30,13 @@
package tso
import (
"log"
"sync/atomic"
"time"
"github.com/cockroachdb/errors"
"github.com/milvus-io/milvus/internal/kv"
"github.com/milvus-io/milvus/internal/log"
"github.com/milvus-io/milvus/internal/util/tsoutil"
"github.com/milvus-io/milvus/internal/util/typeutil"
"go.uber.org/zap"
@ -119,15 +119,15 @@ func (gta *GlobalTSOAllocator) GenerateTSO(count uint32) (uint64, error) {
current := (*atomicObject)(atomic.LoadPointer(&gta.tso.TSO))
if current == nil || current.physical.Equal(typeutil.ZeroTime) {
// If it's leader, maybe SyncTimestamp hasn't completed yet
log.Println("sync hasn't completed yet, wait for a while")
log.Info("sync hasn't completed yet, wait for a while")
time.Sleep(200 * time.Millisecond)
continue
}
physical = current.physical.UnixNano() / int64(time.Millisecond)
physical = current.physical.UnixMilli()
logical = atomic.AddInt64(&current.logical, int64(count))
if logical >= maxLogical && gta.LimitMaxLogic {
log.Println("logical part outside of max logical interval, please check ntp time",
log.Info("logical part outside of max logical interval, please check ntp time",
zap.Int("retry-count", i))
time.Sleep(UpdateTimestampStep)
continue