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

Make creating collection/partition follow tt mechanism (#19628) 

Signed-off-by: longjiquan <jiquan.long@zilliz.com>

Signed-off-by: longjiquan <jiquan.long@zilliz.com>
pull/19632/head
Jiquan Long 2022-10-09 12:32:57 +08:00 committed by GitHub
parent 246693acb7
commit bdd6171231
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 273 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
internal/rootcoord/root_coord.go
View File

@ -193,9 +193,23 @@ func (c *Core) sendTimeTick(t Timestamp, reason string) error {
}
func (c *Core) sendMinDdlTsAsTt() {
minDdlTs := c.ddlTsLockManager.GetMinDdlTs()
err := c.sendTimeTick(minDdlTs, "timetick loop")
if err != nil {
minBgDdlTs := c.ddlTsLockManager.GetMinDdlTs()
minNormalDdlTs := c.scheduler.GetMinDdlTs()
minDdlTs := funcutil.Min(minBgDdlTs, minNormalDdlTs)
// zero -> ddlTsLockManager and scheduler not started.
if minDdlTs == typeutil.ZeroTimestamp {
log.Warn("zero ts was met, this should be only occurred in starting state", zap.Uint64("minBgDdlTs", minBgDdlTs), zap.Uint64("minNormalDdlTs", minNormalDdlTs))
return
}
// max -> abnormal case, impossible.
if minDdlTs == typeutil.MaxTimestamp {
log.Warn("ddl ts is abnormal, max ts was met", zap.Uint64("minBgDdlTs", minBgDdlTs), zap.Uint64("minNormalDdlTs", minNormalDdlTs))
return
}
if err := c.sendTimeTick(minDdlTs, "timetick loop"); err != nil {
log.Warn("failed to send timetick", zap.Error(err))
}
}

25
internal/rootcoord/root_coord_test.go
View File

@ -1285,12 +1285,28 @@ func TestCore_sendMinDdlTsAsTt(t *testing.T) {
ddlManager.GetMinDdlTsFunc = func() Timestamp {
return 100
}
sched := newMockScheduler()
sched.GetMinDdlTsFunc = func() Timestamp {
return 100
}
c := newTestCore(
withTtSynchronizer(ticker),
withDdlTsLockManager(ddlManager))
withDdlTsLockManager(ddlManager),
withScheduler(sched))
c.sendMinDdlTsAsTt() // no session.
ticker.addSession(&sessionutil.Session{ServerID: TestRootCoordID})
c.sendMinDdlTsAsTt()
sched.GetMinDdlTsFunc = func() Timestamp {
return typeutil.ZeroTimestamp
}
c.sendMinDdlTsAsTt() // zero ts
sched.GetMinDdlTsFunc = func() Timestamp {
return typeutil.MaxTimestamp
}
ddlManager.GetMinDdlTsFunc = func() Timestamp {
return typeutil.MaxTimestamp
}
c.sendMinDdlTsAsTt()
}
func TestCore_startTimeTickLoop(t *testing.T) {
@ -1300,9 +1316,14 @@ func TestCore_startTimeTickLoop(t *testing.T) {
ddlManager.GetMinDdlTsFunc = func() Timestamp {
return 100
}
sched := newMockScheduler()
sched.GetMinDdlTsFunc = func() Timestamp {
return 100
}
c := newTestCore(
withTtSynchronizer(ticker),
withDdlTsLockManager(ddlManager))
withDdlTsLockManager(ddlManager),
withScheduler(sched))
ctx, cancel := context.WithCancel(context.Background())
c.ctx = ctx
Params.ProxyCfg.TimeTickInterval = time.Millisecond

36
internal/rootcoord/scheduler.go
View File

@ -3,6 +3,12 @@ package rootcoord
import (
"context"
"sync"
"time"
"github.com/milvus-io/milvus/internal/log"
"go.uber.org/atomic"
"go.uber.org/zap"
"github.com/milvus-io/milvus/internal/tso"
@ -13,6 +19,7 @@ type IScheduler interface {
Start()
Stop()
AddTask(t task) error
GetMinDdlTs() Timestamp
}
type scheduler struct {
@ -26,6 +33,8 @@ type scheduler struct {
taskChan chan task
lock sync.Mutex
minDdlTs atomic.Uint64
}
func newScheduler(ctx context.Context, idAllocator allocator.Interface, tsoAllocator tso.Allocator) *scheduler {
@ -38,6 +47,7 @@ func newScheduler(ctx context.Context, idAllocator allocator.Interface, tsoAlloc
idAllocator: idAllocator,
tsoAllocator: tsoAllocator,
taskChan: make(chan task, n),
minDdlTs: *atomic.NewUint64(0),
}
}
@ -52,6 +62,7 @@ func (s *scheduler) Stop() {
}
func (s *scheduler) execute(task task) {
defer s.setMinDdlTs(task.GetTs()) // we should update ts, whatever task succeeds or not.
if err := task.Prepare(task.GetCtx()); err != nil {
task.NotifyDone(err)
return
@ -62,16 +73,33 @@ func (s *scheduler) execute(task task) {
func (s *scheduler) taskLoop() {
defer s.wg.Done()
ticker := time.NewTicker(Params.ProxyCfg.TimeTickInterval)
defer ticker.Stop()
for {
select {
case <-s.ctx.Done():
return
case <-ticker.C:
s.updateLatestTsoAsMinDdlTs()
case task := <-s.taskChan:
s.execute(task)
}
}
}
func (s *scheduler) updateLatestTsoAsMinDdlTs() {
if len(s.taskChan) > 0 {
return
}
ts, err := s.tsoAllocator.GenerateTSO(1)
if err != nil {
log.Warn("failed to generate tso, ignore to update min ddl ts", zap.Error(err))
} else {
s.setMinDdlTs(ts)
}
}
func (s *scheduler) setID(task task) error {
id, err := s.idAllocator.AllocOne()
if err != nil {
@ -108,3 +136,11 @@ func (s *scheduler) AddTask(task task) error {
s.enqueue(task)
return nil
}
func (s *scheduler) GetMinDdlTs() Timestamp {
return s.minDdlTs.Load()
}
func (s *scheduler) setMinDdlTs(ts Timestamp) {
s.minDdlTs.Store(ts)
}

3
internal/rootcoord/scheduler_test.go
View File

@ -191,6 +191,8 @@ func Test_scheduler_updateDdlMinTsLoop(t *testing.T) {
time.Sleep(time.Millisecond * 4)
assert.Greater(t, s.GetMinDdlTs(), Timestamp(100))
// add task to queue.
n := 10
for i := 0; i < n; i++ {
@ -219,6 +221,7 @@ func Test_scheduler_updateDdlMinTsLoop(t *testing.T) {
s.Start()
time.Sleep(time.Millisecond * 4)
assert.Zero(t, s.GetMinDdlTs())
s.Stop()
})
}

56
internal/util/funcutil/aggregation.go Normal file
View File

@ -0,0 +1,56 @@
package funcutil
/*
type aggFunc[T constraints.Ordered] func(t1, t2 T) T
func agg[T constraints.Ordered](op aggFunc[T], s ...T) T {
l := len(s)
if l <= 0 {
var zero T
return zero
}
m := s[0]
for i := 1; i < l; i++ {
m = op(s[i], m)
}
return m
}
func getMin[T constraints.Ordered](t1, t2 T) T {
if t1 < t2 {
return t1
}
return t2
}
func getMax[T constraints.Ordered](t1, t2 T) T {
if t1 < t2 {
return t2
}
return t1
}
func getSum[T constraints.Ordered](t1, t2 T) T {
return t1 + t2
}
func Min[T constraints.Ordered](s ...T) T {
return agg[T](getMin[T], s...)
}
func Max[T constraints.Ordered](s ...T) T {
return agg[T](getMax[T], s...)
}
func Sum[T constraints.Ordered](s ...T) T {
return agg[T](getSum[T], s...)
}
*/
func Min(t1, t2 uint64) uint64 {
if t1 < t2 {
return t1
}
return t2
}

138
internal/util/funcutil/aggregation_test.go Normal file
View File

@ -0,0 +1,138 @@
package funcutil
import (
"testing"
"github.com/stretchr/testify/assert"
)
/*
func TestMin(t *testing.T) {
assert.Equal(t, uint(0), Min[uint]())
assert.Equal(t, uint(1), Min[uint](100, 1))
assert.Equal(t, uint(1), Min[uint](100, 1, 1000))
assert.Equal(t, uint16(0), Min[uint16]())
assert.Equal(t, uint16(1), Min[uint16](100, 1))
assert.Equal(t, uint16(1), Min[uint16](100, 1, 1000))
assert.Equal(t, uint32(0), Min[uint32]())
assert.Equal(t, uint32(1), Min[uint32](100, 1))
assert.Equal(t, uint32(1), Min[uint32](100, 1, 1000))
assert.Equal(t, uint64(0), Min[uint64]())
assert.Equal(t, uint64(1), Min[uint64](100, 1))
assert.Equal(t, uint64(1), Min[uint64](100, 1, 1000))
assert.Equal(t, 0, Min[int]())
assert.Equal(t, 1, Min[int](100, 1))
assert.Equal(t, 1, Min[int](100, 1, 1000))
assert.Equal(t, int16(0), Min[int16]())
assert.Equal(t, int16(1), Min[int16](100, 1))
assert.Equal(t, int16(1), Min[int16](100, 1, 1000))
assert.Equal(t, int32(0), Min[int32]())
assert.Equal(t, int32(1), Min[int32](100, 1))
assert.Equal(t, int32(1), Min[int32](100, 1, 1000))
assert.Equal(t, int64(0), Min[int64]())
assert.Equal(t, int64(1), Min[int64](100, 1))
assert.Equal(t, int64(1), Min[int64](100, 1, 1000))
}
func TestMax(t *testing.T) {
assert.Equal(t, uint(0), Max[uint]())
assert.Equal(t, uint(100), Max[uint](100, 1))
assert.Equal(t, uint(1000), Max[uint](100, 1, 1000))
assert.Equal(t, uint16(0), Max[uint16]())
assert.Equal(t, uint16(100), Max[uint16](100, 1))
assert.Equal(t, uint16(1000), Max[uint16](100, 1, 1000))
assert.Equal(t, uint32(0), Max[uint32]())
assert.Equal(t, uint32(100), Max[uint32](100, 1))
assert.Equal(t, uint32(1000), Max[uint32](100, 1, 1000))
assert.Equal(t, uint64(0), Max[uint64]())
assert.Equal(t, uint64(100), Max[uint64](100, 1))
assert.Equal(t, uint64(1000), Max[uint64](100, 1, 1000))
assert.Equal(t, 0, Max[int]())
assert.Equal(t, 100, Max[int](100, 1))
assert.Equal(t, 1000, Max[int](100, 1, 1000))
assert.Equal(t, int16(0), Max[int16]())
assert.Equal(t, int16(100), Max[int16](100, 1))
assert.Equal(t, int16(1000), Max[int16](100, 1, 1000))
assert.Equal(t, int32(0), Max[int32]())
assert.Equal(t, int32(100), Max[int32](100, 1))
assert.Equal(t, int32(1000), Max[int32](100, 1, 1000))
assert.Equal(t, int64(0), Max[int64]())
assert.Equal(t, int64(100), Max[int64](100, 1))
assert.Equal(t, int64(1000), Max[int64](100, 1, 1000))
}
func TestSum(t *testing.T) {
assert.Equal(t, uint(0), Sum[uint]())
assert.Equal(t, uint(101), Sum[uint](100, 1))
assert.Equal(t, uint(1101), Sum[uint](100, 1, 1000))
assert.Equal(t, uint16(0), Sum[uint16]())
assert.Equal(t, uint16(101), Sum[uint16](100, 1))
assert.Equal(t, uint16(1101), Sum[uint16](100, 1, 1000))
assert.Equal(t, uint32(0), Sum[uint32]())
assert.Equal(t, uint32(101), Sum[uint32](100, 1))
assert.Equal(t, uint32(1101), Sum[uint32](100, 1, 1000))
assert.Equal(t, uint64(0), Sum[uint64]())
assert.Equal(t, uint64(101), Sum[uint64](100, 1))
assert.Equal(t, uint64(1101), Sum[uint64](100, 1, 1000))
assert.Equal(t, 0, Sum[int]())
assert.Equal(t, 101, Sum[int](100, 1))
assert.Equal(t, 1101, Sum[int](100, 1, 1000))
assert.Equal(t, int16(0), Sum[int16]())
assert.Equal(t, int16(101), Sum[int16](100, 1))
assert.Equal(t, int16(1101), Sum[int16](100, 1, 1000))
assert.Equal(t, int32(0), Sum[int32]())
assert.Equal(t, int32(101), Sum[int32](100, 1))
assert.Equal(t, int32(1101), Sum[int32](100, 1, 1000))
assert.Equal(t, int64(0), Sum[int64]())
assert.Equal(t, int64(101), Sum[int64](100, 1))
assert.Equal(t, int64(1101), Sum[int64](100, 1, 1000))
}
*/
func TestMin(t *testing.T) {
type args struct {
t1 uint64
t2 uint64
}
tests := []struct {
name string
args args
want uint64
}{
{
args: args{t1: 100, t2: 1},
want: 1,
},
{
args: args{t1: 1, t2: 100},
want: 1,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
assert.Equalf(t, tt.want, Min(tt.args.t1, tt.args.t2), "Min(%v, %v)", tt.args.t1, tt.args.t2)
})
}
}