Repository: pingcap/tiflow
Pull request #4664 (2022-02-23 10:02:03)

config(ticdc): make table actor batch size configurable
<!--
Thank you for contributing to TiDB-CDC! Please read MD's [CONTRIBUTING](https://github.com/pingcap/tidb-cdc/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve?
<!--
Please create an issue first to describe the problem.

There MUST be one line starting with "Issue Number:  " and 
linking the relevant issues via the "close" or "ref".

For more info, check https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue.
 -->

Issue Number: ref https://github.com/pingcap/tiflow/issues/3881

this is a part of table pipeline actor model refactor https://github.com/pingcap/tiflow/issues/3881

### What is changed and how it works?


### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->
add a changefeed level configuration to set the table actor batch size

 - Unit test
 - Integration test

Code changes

 - Has exported function/method change
 - Has exported variable/fields change
 - Has interface methods change
 - Has persistent data change


### Release note <!-- bugfixes or new feature need a release note -->

```release-note
`None`.
```


ti-chi-bot: [REVIEW NOTIFICATION]

This pull request has not been approved.


To complete the [pull request process](https://book.prow.tidb.io/#/en/workflows/pr), please ask the reviewers in the [list](https://prow.tidb.io/tichi/repos/pingcap/tiflow/pulls/4664/owners) to review by filling `/cc @reviewer` in the comment.
After your PR has acquired the required number of LGTMs, you can assign this pull request to the committer in the [list](https://prow.tidb.io/tichi/repos/pingcap/tiflow/pulls/4664/owners) by filling  `/assign @committer` in the comment to help you merge this pull request.

The full list of commands accepted by this bot can be found [here](https://prow.tidb.io/command-help?repo=pingcap%2Ftiflow).

<details>

Reviewer can indicate their review by submitting an approval review.
Reviewer can cancel approval by submitting a request changes review.
</details>

<!--Review Notification Identifier-->
hi-rustin: Please refine your PR title.
codecov-commenter: # [Codecov](https://codecov.io/gh/pingcap/tiflow/pull/4664?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) Report
> Merging [#4664](https://codecov.io/gh/pingcap/tiflow/pull/4664?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) (dca89f1) into [master](https://codecov.io/gh/pingcap/tiflow/commit/96075543f5692d4a75eec041c82bac672eb3594b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) (9607554) will **decrease** coverage by `0.6209%`.
> The diff coverage is `40.3028%`.


| Flag | Coverage Δ | |
|---|---|---|
| cdc | `59.5143% <43.3663%> (-0.4080%)` | :arrow_down: |
| dm | `51.2513% <37.3191%> (-0.7776%)` | :arrow_down: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap#carryforward-flags-in-the-pull-request-comment) to find out more.

```diff
@@               Coverage Diff                @@
##             master      #4664        +/-   ##
================================================
- Coverage   55.6402%   55.0192%   -0.6210%     
================================================
  Files           494        513        +19     
  Lines         61283      63754      +2471     
================================================
+ Hits          34098      35077       +979     
- Misses        23750      25186      +1436     
- Partials       3435       3491        +56     
```

hi-rustin: /merge
ti-chi-bot: This pull request has been accepted and is ready to merge. <details>Commit hash: 1a81b3bd033463a68669b06cc67e52f3a272875e</details>

Patch:
diff --git a/cdc/processor/pipeline/actor_node_context.go b/cdc/processor/pipeline/actor_node_context.go
index 4be3600bb..a33cc4e3c 100644
--- a/cdc/processor/pipeline/actor_node_context.go
+++ b/cdc/processor/pipeline/actor_node_context.go
@@ -20,13 +20,14 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/pkg/actor"
 	"github.com/pingcap/tiflow/pkg/actor/message"
+	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/pingcap/tiflow/pkg/context"
 	"github.com/pingcap/tiflow/pkg/pipeline"
 	"go.uber.org/zap"
 )
 
-// send a tick message to actor if we get 32 pipeline messages
-const messagesPerTick = 32
+// defaultEventBatchSize specifies that if we get 32 pipeline events, a tick message is sent to the actor.
+const defaultEventBatchSize = uint32(32)
 
 // actorNodeContext implements the NodeContext interface, with this we do not need
 // to change too much logic to implement the table actor.
@@ -34,16 +35,16 @@ const messagesPerTick = 32
 // the Throw function handle error and stop the actor
 type actorNodeContext struct {
 	sdtContext.Context
-	outputCh             chan pipeline.Message
-	tableActorRouter     *actor.Router
-	tableActorID         actor.ID
-	changefeedVars       *context.ChangefeedVars
-	globalVars           *context.GlobalVars
-	tickMessageThreshold int32
-	// noTickMessageCount is the count of pipeline message that no tick message is sent to actor
-	noTickMessageCount int32
-	tableName          string
-	throw              func(error)
+	outputCh         chan pipeline.Message
+	tableActorRouter *actor.Router
+	tableActorID     actor.ID
+	changefeedVars   *context.ChangefeedVars
+	globalVars       *context.GlobalVars
+	eventBatchSize   uint32
+	// eventCount is the count of pipeline event that no tick message is sent to actor
+	eventCount uint32
+	tableName  string
+	throw      func(error)
 }
 
 func newContext(stdCtx sdtContext.Context,
@@ -53,22 +54,26 @@ func newContext(stdCtx sdtContext.Context,
 	changefeedVars *context.ChangefeedVars,
 	globalVars *context.GlobalVars,
 	throw func(error)) *actorNodeContext {
+	batchSize := defaultEventBatchSize
+	if config.GetGlobalServerConfig().Debug.TableActor != nil {
+		batchSize = config.GetGlobalServerConfig().Debug.TableActor.EventBatchSize
+	}
 	return &actorNodeContext{
-		Context:              stdCtx,
-		outputCh:             make(chan pipeline.Message, defaultOutputChannelSize),
-		tableActorRouter:     tableActorRouter,
-		tableActorID:         tableActorID,
-		changefeedVars:       changefeedVars,
-		globalVars:           globalVars,
-		tickMessageThreshold: messagesPerTick,
-		noTickMessageCount:   0,
-		tableName:            tableName,
-		throw:                throw,
+		Context:          stdCtx,
+		outputCh:         make(chan pipeline.Message, defaultOutputChannelSize),
+		tableActorRouter: tableActorRouter,
+		tableActorID:     tableActorID,
+		changefeedVars:   changefeedVars,
+		globalVars:       globalVars,
+		eventBatchSize:   batchSize,
+		eventCount:       0,
+		tableName:        tableName,
+		throw:            throw,
 	}
 }
 
-func (c *actorNodeContext) setTickMessageThreshold(threshold int32) {
-	atomic.StoreInt32(&c.tickMessageThreshold, threshold)
+func (c *actorNodeContext) setEventBatchSize(eventBatchSize uint32) {
+	atomic.StoreUint32(&c.eventBatchSize, eventBatchSize)
 }
 
 func (c *actorNodeContext) GlobalVars() *context.GlobalVars {
@@ -129,12 +134,12 @@ func (c *actorNodeContext) tryGetProcessedMessage() *pipeline.Message {
 }
 
 func (c *actorNodeContext) trySendTickMessage() {
-	threshold := atomic.LoadInt32(&c.tickMessageThreshold)
-	atomic.AddInt32(&c.noTickMessageCount, 1)
-	count := atomic.LoadInt32(&c.noTickMessageCount)
-	// resolvedTs message will be sent by puller periodically
+	threshold := atomic.LoadUint32(&c.eventBatchSize)
+	atomic.AddUint32(&c.eventCount, 1)
+	count := atomic.LoadUint32(&c.eventCount)
+	// resolvedTs event will be sent by puller periodically
 	if count >= threshold {
 		_ = c.tableActorRouter.Send(c.tableActorID, message.TickMessage())
-		atomic.StoreInt32(&c.noTickMessageCount, 0)
+		atomic.StoreUint32(&c.eventCount, 0)
 	}
 }
diff --git a/pkg/config/debug.go b/pkg/config/debug.go
index 73ccca44c..bd4af8a3c 100644
--- a/pkg/config/debug.go
+++ b/pkg/config/debug.go
@@ -19,7 +19,8 @@ import "github.com/pingcap/errors"
 type DebugConfig struct {
 	// identify if the table actor is enabled for table pipeline
 	// TODO: turn on after GA.
-	EnableTableActor bool `toml:"enable-table-actor" json:"enable-table-actor"`
+	EnableTableActor bool              `toml:"enable-table-actor" json:"enable-table-actor"`
+	TableActor       *TableActorConfig `toml:"table-actor" json:"table-actor"`
 
 	// EnableDBSorter enables db sorter.
 	//
diff --git a/pkg/config/server_config.go b/pkg/config/server_config.go
index 4873f7409..bb930dd6f 100644
--- a/pkg/config/server_config.go
+++ b/pkg/config/server_config.go
@@ -98,7 +98,10 @@ var defaultServerConfig = &ServerConfig{
 		RegionScanLimit:  40,
 	},
 	Debug: &DebugConfig{
-		EnableTableActor:   false,
+		EnableTableActor: false,
+		TableActor: &TableActorConfig{
+			EventBatchSize: 32,
+		},
 		EnableNewScheduler: false,
 		// Default leveldb sorter config
 		EnableDBSorter: false,
diff --git a/pkg/config/table_actor.go b/pkg/config/table_actor.go
new file mode 100644
index 000000000..04ab1546f
--- /dev/null
+++ b/pkg/config/table_actor.go
@@ -0,0 +1,20 @@
+// Copyright 2022 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package config
+
+// TableActorConfig represents config used for table actor
+type TableActorConfig struct {
+	// EventBatchSize represents the batch size of events that table actor processed per Poll
+	EventBatchSize uint32 `toml:"event-batch-size" json:"event-batch-size"`
+}


Test patch:
diff --git a/cdc/processor/pipeline/actor_node_context_test.go b/cdc/processor/pipeline/actor_node_context_test.go
index f477a03b0..0f4e2b66e 100644
--- a/cdc/processor/pipeline/actor_node_context_test.go
+++ b/cdc/processor/pipeline/actor_node_context_test.go
@@ -19,6 +19,7 @@ import (
 	"time"
 
 	"github.com/pingcap/errors"
+	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/processor/pipeline/system"
 	"github.com/pingcap/tiflow/pkg/actor"
 	"github.com/pingcap/tiflow/pkg/actor/message"
@@ -29,12 +30,12 @@ import (
 
 func TestContext(t *testing.T) {
 	t.Parallel()
-	ctx := newContext(sdtContext.TODO(), t.Name(), nil, 1, &context.ChangefeedVars{ID: "zzz"}, &context.GlobalVars{}, throwDoNothing)
+	ctx := newContext(sdtContext.TODO(), t.Name(), nil, 1, &context.ChangefeedVars{ID: "zzz", Info: &model.ChangeFeedInfo{}}, &context.GlobalVars{}, throwDoNothing)
 	require.NotNil(t, ctx.GlobalVars())
 	require.Equal(t, "zzz", ctx.ChangefeedVars().ID)
 	require.Equal(t, actor.ID(1), ctx.tableActorID)
 	ctx.SendToNextNode(pipeline.BarrierMessage(1))
-	require.Equal(t, int32(1), ctx.noTickMessageCount)
+	require.Equal(t, uint32(1), ctx.eventCount)
 	wait(t, 500*time.Millisecond, func() {
 		msg := ctx.Message()
 		require.Equal(t, pipeline.MessageTypeBarrier, msg.Tp)
@@ -90,7 +91,7 @@ func TestSendToNextNodeNoTickMessage(t *testing.T) {
 	fa := &forwardActor{ch: ch}
 	require.Nil(t, sys.System().Spawn(mb, fa))
 	actorContext := newContext(ctx, t.Name(), sys.Router(), actorID, &context.ChangefeedVars{ID: "abc"}, &context.GlobalVars{}, throwDoNothing)
-	actorContext.setTickMessageThreshold(2)
+	actorContext.setEventBatchSize(2)
 	actorContext.SendToNextNode(pipeline.BarrierMessage(1))
 	time.Sleep(100 * time.Millisecond)
 	require.Equal(t, 0, len(ch))
diff --git a/pkg/cmd/server/server_test.go b/pkg/cmd/server/server_test.go
index 7921137f9..bbf131fe4 100644
--- a/pkg/cmd/server/server_test.go
+++ b/pkg/cmd/server/server_test.go
@@ -176,7 +176,10 @@ func TestParseCfg(t *testing.T) {
 		},
 		Debug: &config.DebugConfig{
 			EnableTableActor: false,
-			EnableDBSorter:   false,
+			TableActor: &config.TableActorConfig{
+				EventBatchSize: 32,
+			},
+			EnableDBSorter: false,
 			DB: &config.DBConfig{
 				Count:                       8,
 				Concurrency:                 128,
@@ -315,7 +318,10 @@ server-worker-pool-size = 16
 		},
 		Debug: &config.DebugConfig{
 			EnableTableActor: false,
-			EnableDBSorter:   false,
+			TableActor: &config.TableActorConfig{
+				EventBatchSize: 32,
+			},
+			EnableDBSorter: false,
 			DB: &config.DBConfig{
 				Count:                       5,
 				Concurrency:                 6,
@@ -452,7 +458,10 @@ cert-allowed-cn = ["dd","ee"]
 		},
 		Debug: &config.DebugConfig{
 			EnableTableActor: false,
-			EnableDBSorter:   false,
+			TableActor: &config.TableActorConfig{
+				EventBatchSize: 32,
+			},
+			EnableDBSorter: false,
 			DB: &config.DBConfig{
 				Count:                       8,
 				Concurrency:                 128,
@@ -508,7 +517,10 @@ unknown3 = 3
 	require.Nil(t, err)
 	require.Equal(t, &config.DebugConfig{
 		EnableTableActor: false,
-		EnableDBSorter:   false,
+		TableActor: &config.TableActorConfig{
+			EventBatchSize: 32,
+		},
+		EnableDBSorter: false,
 		DB: &config.DBConfig{
 			Count:                       8,
 			Concurrency:                 128,
diff --git a/pkg/config/config_test_data.go b/pkg/config/config_test_data.go
index 7c8d254fb..fca23e8dd 100644
--- a/pkg/config/config_test_data.go
+++ b/pkg/config/config_test_data.go
@@ -109,6 +109,9 @@ const (
   },
   "debug": {
     "enable-table-actor": false,
+    "table-actor": {
+      "event-batch-size": 32
+    },
     "enable-db-sorter": false,
     "db": {
       "count": 8,