Repository: matrixorigin/matrixone
Pull request #10105 (2023-06-16 09:50:06)

fix issue 8599: move task schedule to a separate goroutine
## What type of PR is this?

- [ ] API-change
- [x] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #8599  #9794 

## What this PR does / why we need it:

moving the task schedule from the ticker normal routine to a separate goroutine can avoid the hakeeper's health check and tick update operations being blocked by the task schedule, or the tick will be skipped and can not correctly estimate the time passing. and added some unit tests.



matrix-meow: @gouhongshen Thanks for your contributions!

Title:
The title is clear and concise. It indicates that the pull request is meant to fix issue #8599 by moving the task schedule to a separate goroutine.

Body:
The body of the pull request is also clear and concise. It indicates that moving the task schedule to a separate goroutine can avoid the hakeeper's health check and tick update operations being blocked by the task schedule, or the tick will be skipped and can not correctly estimate the time passing. The pull request also includes some unit tests.

Changes:
The changes made in the pull request are well documented. The changes include adding a new function `tickerForTaskSchedule` to move the task schedule to a separate goroutine, modifying the `ticker` function to call the `tickerForTaskSchedule` function, and modifying the `hakeeperCheck` function to call the `getCheckerStateFromLeader` function. The changes also include adding a new test function `TestTickerForTaskSchedule` to test the `tickerForTaskSchedule` function.

Overall, the pull request seems to be well thought out and well documented. However, there are a few issues that need to be addressed.

Issues:
1. The `getCheckerStateFromLeader` function does not handle errors properly. If an error occurs, the function should return `nil` for the `state` variable and the `term` variable should be returned as is. This will ensure that the `hakeeperCheck` function does not panic when an

Patch:
diff --git a/pkg/logservice/store.go b/pkg/logservice/store.go
index fa225b445..2654ca64a 100644
--- a/pkg/logservice/store.go
+++ b/pkg/logservice/store.go
@@ -667,6 +667,34 @@ func (l *store) queryLog(ctx context.Context, shardID uint64,
 	}
 }
 
+func (l *store) tickerForTaskSchedule(ctx context.Context, duration time.Duration) {
+	ticker := time.NewTicker(duration)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			state, _ := l.getCheckerStateFromLeader()
+			if state != nil && state.State == pb.HAKeeperRunning {
+				l.taskSchedule(state)
+			}
+
+		case <-ctx.Done():
+			return
+		}
+
+		// l.taskSchedule could be blocking a long time, this extra select
+		// can give a chance immediately to check the ctx status when it resumes.
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			// nothing to do
+		}
+	}
+
+}
+
 func (l *store) ticker(ctx context.Context) {
 	if l.cfg.HAKeeperTickInterval.Duration == 0 {
 		panic("invalid HAKeeperTickInterval")
@@ -685,6 +713,12 @@ func (l *store) ticker(ctx context.Context) {
 	haTicker := time.NewTicker(l.cfg.HAKeeperCheckInterval.Duration)
 	defer haTicker.Stop()
 
+	// moving task schedule from the ticker normal routine to a
+	// separate goroutine can avoid the hakeeper's health check and tick update
+	// operations being blocked by task schedule, or the tick will be skipped and
+	// can not correctly estimate the time passing.
+	go l.tickerForTaskSchedule(ctx, l.cfg.HAKeeperCheckInterval.Duration)
+
 	for {
 		select {
 		case <-ticker.C:
diff --git a/pkg/logservice/store_hakeeper_check.go b/pkg/logservice/store_hakeeper_check.go
index 85a156f2a..df4d7162c 100644
--- a/pkg/logservice/store_hakeeper_check.go
+++ b/pkg/logservice/store_hakeeper_check.go
@@ -112,25 +112,35 @@ func (l *store) updateIDAlloc(count uint64) error {
 	return nil
 }
 
-var debugPrintHAKeeperState atomic.Bool
-
-func (l *store) hakeeperCheck() {
+func (l *store) getCheckerStateFromLeader() (*pb.CheckerState, uint64) {
 	isLeader, term, err := l.isLeaderHAKeeper()
 	if err != nil {
 		l.runtime.Logger().Error("failed to get HAKeeper Leader ID", zap.Error(err))
-		return
+		return nil, term
 	}
 
 	if !isLeader {
 		l.taskScheduler.StopScheduleCronTask()
-		return
+		return nil, term
 	}
 	state, err := l.getCheckerState()
 	if err != nil {
 		// TODO: check whether this is temp error
 		l.runtime.Logger().Error("failed to get checker state", zap.Error(err))
+		return nil, term
+	}
+
+	return state, term
+}
+
+var debugPrintHAKeeperState atomic.Bool
+
+func (l *store) hakeeperCheck() {
+	state, term := l.getCheckerStateFromLeader()
+	if state == nil {
 		return
 	}
+
 	switch state.State {
 	case pb.HAKeeperCreated:
 		l.runtime.Logger().Warn("waiting for initial cluster info to be set, check skipped")
@@ -147,7 +157,6 @@ func (l *store) hakeeperCheck() {
 				zap.Uint64("next id", state.NextId))
 		}
 		l.healthCheck(term, state)
-		l.taskSchedule(state)
 	default:
 		panic("unknown HAKeeper state")
 	}


Test patch:
diff --git a/pkg/logservice/store_hakeeper_check_test.go b/pkg/logservice/store_hakeeper_check_test.go
index 30f9feffa..227727860 100644
--- a/pkg/logservice/store_hakeeper_check_test.go
+++ b/pkg/logservice/store_hakeeper_check_test.go
@@ -494,6 +494,37 @@ func TestHAKeeperCanBootstrapAndRepairShards(t *testing.T) {
 	runHAKeeperClusterTest(t, fn)
 }
 
+func TestGetCheckerStateFromLeader(t *testing.T) {
+	fn := func(t *testing.T, store *store) {
+		ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Second*10))
+		defer cancel()
+
+		for {
+			select {
+			case <-ctx.Done():
+				t.Error("test deadline reached")
+				return
+
+			default:
+				isLeader, termA, err := store.isLeaderHAKeeper()
+				state, termB := store.getCheckerStateFromLeader()
+				require.NoError(t, err)
+				assert.Equal(t, termB, termA)
+
+				if !isLeader {
+					assert.Equal(t, (*pb.CheckerState)(nil), state)
+				} else {
+					assert.NotEqual(t, (*pb.CheckerState)(nil), state)
+					return
+				}
+				time.Sleep(time.Second)
+			}
+		}
+	}
+
+	runHAKeeperStoreTest(t, false, fn)
+}
+
 func TestGetCheckerState(t *testing.T) {
 	fn := func(t *testing.T, store *store) {
 		state, err := store.getCheckerState()
diff --git a/pkg/logservice/store_test.go b/pkg/logservice/store_test.go
index 49d44ef55..543b6fe1e 100644
--- a/pkg/logservice/store_test.go
+++ b/pkg/logservice/store_test.go
@@ -17,6 +17,7 @@ package logservice
 import (
 	"context"
 	"fmt"
+	"github.com/matrixorigin/matrixone/pkg/pb/task"
 	"math"
 	"sync/atomic"
 	"testing"
@@ -298,6 +299,98 @@ func TestQueryLog(t *testing.T) {
 	runStoreTest(t, fn)
 }
 
+func proceedHAKeeperToRunning(t *testing.T, store *store) {
+	state, err := store.getCheckerState()
+	assert.NoError(t, err)
+	assert.Equal(t, pb.HAKeeperCreated, state.State)
+
+	err = store.setInitialClusterInfo(1, 1, 1)
+	assert.NoError(t, err)
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	hb := store.getHeartbeatMessage()
+	_, err = store.addLogStoreHeartbeat(ctx, hb)
+	assert.NoError(t, err)
+
+	state, err = store.getCheckerState()
+	assert.NoError(t, err)
+	assert.Equal(t, pb.HAKeeperBootstrapping, state.State)
+
+	_, term, err := store.isLeaderHAKeeper()
+	assert.NoError(t, err)
+
+	store.bootstrap(term, state)
+	state, err = store.getCheckerState()
+
+	assert.NoError(t, err)
+	assert.Equal(t, pb.HAKeeperBootstrapCommandsReceived, state.State)
+
+	cmd, err := store.getCommandBatch(ctx, store.id())
+	require.NoError(t, err)
+	require.Equal(t, 1, len(cmd.Commands))
+	assert.True(t, cmd.Commands[0].Bootstrapping)
+
+	// handle startReplica to make sure logHeartbeat msg contain shards info,
+	// which used in store.checkBootstrap to determine if all log shards ready
+	service := &Service{store: store}
+	service.handleStartReplica(cmd.Commands[0])
+
+	for state.State != pb.HAKeeperRunning && store.bootstrapCheckCycles > 0 {
+		func() {
+			ctx, cancel = context.WithTimeout(context.Background(), time.Second)
+			defer cancel()
+
+			_, err = store.addLogStoreHeartbeat(ctx, store.getHeartbeatMessage())
+			assert.NoError(t, err)
+
+			store.checkBootstrap(state)
+			state, err = store.getCheckerState()
+			assert.NoError(t, err)
+
+			time.Sleep(time.Millisecond * 100)
+		}()
+	}
+
+	assert.Equal(t, pb.HAKeeperRunning, state.State)
+}
+
+// test if the tickerForTaskSchedule can push forward these routine
+func TestTickerForTaskSchedule(t *testing.T) {
+	fn := func(t *testing.T, store *store, taskService taskservice.TaskService) {
+
+		tickerCxt, tickerCancel := context.WithCancel(context.Background())
+		defer tickerCancel()
+
+		//do task schedule background
+		go store.tickerForTaskSchedule(tickerCxt, time.Millisecond*10)
+
+		// making hakeeper state proceeds to running before test task schedule
+		proceedHAKeeperToRunning(t, store)
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+
+		err := taskService.Create(ctx, task.TaskMetadata{ID: "1234"})
+		assert.NoError(t, err)
+
+		cnUUID := uuid.New().String()
+		cmd := pb.CNStoreHeartbeat{UUID: cnUUID}
+		_, err = store.addCNStoreHeartbeat(ctx, cmd)
+		assert.NoError(t, err)
+
+		// waiting the background taskSchedule operation
+		// schedule task we created to CN node
+		time.Sleep(time.Millisecond * 200)
+
+		tasks, err := taskService.QueryTask(ctx, taskservice.WithTaskRunnerCond(taskservice.EQ, cnUUID))
+		assert.NoError(t, err)
+		assert.Equal(t, 1, len(tasks))
+	}
+
+	runHakeeperTaskServiceTest(t, fn)
+}
+
 func TestHAKeeperTick(t *testing.T) {
 	fn := func(t *testing.T, store *store) {
 		peers := make(map[uint64]dragonboat.Target)