redo(ticdc): limit worker number in redo reader (#8296) (#8308)

close #8085
pingcap · Feb 21, 2023 · 7831ce2 · 7831ce2
1 parent d791b17
commit 7831ce2
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 7 deletions.
diff --git a/cdc/redo/reader/file.go b/cdc/redo/reader/file.go
@@ -81,21 +81,21 @@ func newReader(ctx context.Context, cfg *readerConfig) ([]fileReader, error) {
 	if cfg == nil {
 		return nil, cerror.WrapError(cerror.ErrRedoConfigInvalid, errors.New("readerConfig can not be nil"))
 	}
+	if cfg.workerNums == 0 {
+		cfg.workerNums = defaultWorkerNum
+	}
 
 	if cfg.useExternalStorage {
 		extStorage, err := redo.InitExternalStorage(ctx, cfg.uri)
 		if err != nil {
 			return nil, err
 		}
 
-		err = downLoadToLocal(ctx, cfg.dir, extStorage, cfg.fileType)
+		err = downLoadToLocal(ctx, cfg.dir, extStorage, cfg.fileType, cfg.workerNums)
 		if err != nil {
 			return nil, cerror.WrapError(cerror.ErrRedoDownloadFailed, err)
 		}
 	}
-	if cfg.workerNums == 0 {
-		cfg.workerNums = defaultWorkerNum
-	}
 
 	rr, err := openSelectedFiles(ctx, cfg.dir, cfg.fileType, cfg.startTs, cfg.workerNums)
 	if err != nil {
@@ -141,17 +141,26 @@ func selectDownLoadFile(
 }
 
 func downLoadToLocal(
-	ctx context.Context, dir string, extStorage storage.ExternalStorage, fixedType string,
+	ctx context.Context, dir string,
+	extStorage storage.ExternalStorage,
+	fixedType string, workerNum int,
 ) error {
 	files, err := selectDownLoadFile(ctx, extStorage, fixedType)
 	if err != nil {
 		return err
 	}
 
+	limit := make(chan struct{}, workerNum)
 	eg, eCtx := errgroup.WithContext(ctx)
 	for _, file := range files {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case limit <- struct{}{}:
+		}
 		f := file
 		eg.Go(func() error {
+			defer func() { <-limit }()
 			data, err := extStorage.ReadFile(eCtx, f)
 			if err != nil {
 				return cerror.WrapError(cerror.ErrExternalStorageAPI, err)

diff --git a/cdc/redo/reader/reader.go b/cdc/redo/reader/reader.go
@@ -106,6 +106,9 @@ func newLogReader(ctx context.Context, cfg *LogReaderConfig) (*LogReader, error)
 	if cfg == nil {
 		return nil, cerror.WrapError(cerror.ErrRedoConfigInvalid, errors.New("LogReaderConfig can not be nil"))
 	}
+	if cfg.WorkerNums == 0 {
+		cfg.WorkerNums = defaultWorkerNum
+	}
 
 	logReader := &LogReader{
 		cfg: cfg,
@@ -120,7 +123,7 @@ func newLogReader(ctx context.Context, cfg *LogReaderConfig) (*LogReader, error)
 		if err != nil {
 			return nil, cerror.WrapError(cerror.ErrRedoFileOp, err)
 		}
-		err = downLoadToLocal(ctx, cfg.Dir, extStorage, redo.RedoMetaFileType)
+		err = downLoadToLocal(ctx, cfg.Dir, extStorage, redo.RedoMetaFileType, cfg.WorkerNums)
 		if err != nil {
 			return nil, cerror.WrapError(cerror.ErrRedoDownloadFailed, err)
 		}

diff --git a/pkg/cmd/redo/meta.go b/pkg/cmd/redo/meta.go
@@ -29,7 +29,7 @@ func newMetaOptions() *metaOptions {
 	return &metaOptions{}
 }
 
-// run runs the `redo apply` command.
+// run runs the `redo meta` command.
 func (o *metaOptions) run(cmd *cobra.Command) error {
 	ctx := cmdcontext.GetDefaultContext()