disttask: Change rand choose TiDB to round selection TiDB to be an executor (#43440)

ddl/scheduler.go

@@ -132,7 +132,11 @@ func (b *backfillSchedulerHandle) SplitSubtask(_ context.Context, subtask []byte
  pid := sm.PhysicalTableID
  parTbl := b.ptbl.(table.PartitionedTable)
 
- startKey, endKey, err := getTableRange(b.jc, d.ddlCtx, parTbl.GetPartition(pid), b.job.SnapshotVer, b.job.Priority)
+ currentVer, err1 := getValidCurrentVersion(d.store)
+ if err1 != nil {
+ return nil, errors.Trace(err1)
+ }
+ startKey, endKey, err := getTableRange(b.jc, d.ddlCtx, parTbl.GetPartition(pid), currentVer.Ver, b.job.Priority)
  if err != nil {
  logutil.BgLogger().Error("[ddl] get table range error", zap.Error(err))
  return nil, err

disttask/framework/dispatcher/dispatcher.go

@@ -16,7 +16,7 @@ package dispatcher
 
 import (
  "context"
- "math/rand"
+ "fmt"
  "time"
 
  "github.com/pingcap/errors"
@@ -401,9 +401,14 @@ func (d *dispatcher) processNormalFlow(gTask *proto.Task) (err error) {
  return nil
  }
 
+ // Generate all available TiDB nodes for this global tasks.
+ serverNodes, err1 := GenerateSchedulerNodes(d.ctx)
+ if err1 != nil {
+ return err1
+ }
  subTasks := make([]*proto.Subtask, 0, len(metas))
- for _, meta := range metas {
- instanceID, err := GetEligibleInstance(d.ctx)
+ for i, meta := range metas {
+ instanceID, err := GetEligibleInstance(serverNodes, i)
  if err != nil {
  logutil.BgLogger().Warn("get a eligible instance failed", zap.Int64("gTask ID", gTask.ID), zap.Error(err))
  return err
@@ -414,24 +419,33 @@ func (d *dispatcher) processNormalFlow(gTask *proto.Task) (err error) {
 }
 
 // GetEligibleInstance gets an eligible instance.
-func GetEligibleInstance(ctx context.Context) (string, error) {
+func GetEligibleInstance(serverNodes []*infosync.ServerInfo, pos int) (string, error) {
+ if pos >= len(serverNodes) && pos < 0 {
+ errMsg := fmt.Sprintf("available TiDB nodes range is 0 to %d, but request position: %d", len(serverNodes)-1, pos)
+ return "", errors.New(errMsg)
+ }
+ if len(serverNodes) == 0 {
+ return "", errors.New("no available TiDB node")
+ }
+ pos = pos % len(serverNodes)
+ return serverNodes[pos].ID, nil
+}
+
+// GenerateSchedulerNodes generate a eligible TiDB nodes.
+func GenerateSchedulerNodes(ctx context.Context) ([]*infosync.ServerInfo, error) {
  serverInfos, err := infosync.GetAllServerInfo(ctx)
  if err != nil {
- return "", err
+ return nil, err
  }
  if len(serverInfos) == 0 {
- return "", errors.New("not found instance")
+ return nil, errors.New("not found instance")
  }
 
- // TODO: Consider valid instances, and then consider scheduling strategies.
- num := rand.Intn(len(serverInfos))
- for _, info := range serverInfos {
- if num == 0 {
- return info.ID, nil
- }
- num--
+ serverNodes := make([]*infosync.ServerInfo, 0, len(serverInfos))
+ for _, serverInfo := range serverInfos {
+ serverNodes = append(serverNodes, serverInfo)
  }
- return "", errors.New("not found instance")
+ return serverNodes, nil
 }
 
 // GetAllSchedulerIDs gets all the scheduler IDs.

disttask/framework/dispatcher/dispatcher_test.go

@@ -69,7 +69,8 @@ func TestGetInstance(t *testing.T) {
  // test no server
  mockedAllServerInfos := map[string]*infosync.ServerInfo{}
  require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/domain/infosync/mockGetAllServerInfo", makeFailpointRes(mockedAllServerInfos)))
- instanceID, err := dispatcher.GetEligibleInstance(ctx)
+ serverNodes, err := dispatcher.GenerateSchedulerNodes(ctx)
+ instanceID, _ := dispatcher.GetEligibleInstance(serverNodes, 0)
  require.Lenf(t, instanceID, 0, "instanceID:%d", instanceID)
  require.EqualError(t, err, "not found instance")
  instanceIDs, err := dsp.GetAllSchedulerIDs(ctx, 1)
@@ -89,7 +90,9 @@ func TestGetInstance(t *testing.T) {
  },
  }
  require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/domain/infosync/mockGetAllServerInfo", makeFailpointRes(mockedAllServerInfos)))
- instanceID, err = dispatcher.GetEligibleInstance(ctx)
+ serverNodes, err = dispatcher.GenerateSchedulerNodes(ctx)
+ require.NoError(t, err)
+ instanceID, err = dispatcher.GetEligibleInstance(serverNodes, 0)
  require.NoError(t, err)
  if instanceID != uuids[0] && instanceID != uuids[1] {
  require.FailNowf(t, "expected uuids:%d,%d, actual uuid:%d", uuids[0], uuids[1], instanceID)