pkg/p2p(ticdc): fix default p2p config

cdc/puller/multiplexing_puller.go

@@ -72,6 +72,7 @@ type tableProgress struct {
 	}
 
 	scheduled atomic.Bool
+	start     time.Time
 }
 
 func (p *tableProgress) handleResolvedSpans(ctx context.Context, e *model.ResolvedSpans) (err error) {
@@ -95,7 +96,10 @@ func (p *tableProgress) handleResolvedSpans(ctx context.Context, e *model.Resolv
 			zap.String("namespace", p.changefeed.Namespace),
 			zap.String("changefeed", p.changefeed.ID),
 			zap.String("tableName", p.tableName),
-			zap.Uint64("resolvedTs", resolvedTs))
+			zap.Any("tableID", p.spans),
+			zap.Uint64("resolvedTs", resolvedTs),
+			zap.Duration("duration", time.Since(p.start)),
+		)
 	}
 	if resolvedTs > p.resolvedTs.Load() {
 		p.resolvedTs.Store(resolvedTs)
@@ -230,6 +234,7 @@ func (p *MultiplexingPuller) subscribe(
 
 		resolvedEventsCache: make(chan kv.MultiplexingEvent, tableResolvedTsBufferSize),
 		tsTracker:           frontier.NewFrontier(0, spans...),
+		start:               time.Now(),
 	}
 
 	progress.consume.f = func(

cdc/scheduler/internal/v3/agent/table.go

@@ -272,7 +272,7 @@ func (t *tableSpan) injectDispatchTableTask(task *dispatchTableTask) {
 		t.task = task
 		return
 	}
-	log.Debug("schedulerv3: table inject dispatch table task ignored,"+
+	log.Warn("schedulerv3: table inject dispatch table task ignored,"+
 		"since there is one not finished yet",
 		zap.String("namespace", t.changefeedID.Namespace),
 		zap.String("changefeed", t.changefeedID.ID),

cdc/scheduler/internal/v3/transport/transport.go

@@ -65,6 +65,10 @@
 		// FIXME it's an unbounded buffer, and may cause OOM!
 		msgBuf []*schedulepb.Message
 	}
+	lastPrintTime time.Time
+	ignoreCount   int64
+	totalMsg      int64
+	role          Role
 }
 
 // Role of the transport user.
@@ -89,6 +93,7 @@
 		peerTopic:     peerTopic,
 		messageServer: server,
 		messageRouter: router,
+		role:          role,
 	}
 	var err error
 	trans.errCh, err = trans.messageServer.SyncAddHandler(
@@ -112,6 +117,7 @@
 func (t *p2pTransport) Send(
 	ctx context.Context, msgs []*schedulepb.Message,
 ) error {
+	t.totalMsg += int64(len(msgs))
 	for i := range msgs {
 		value := msgs[i]
 		to := value.To
@@ -127,6 +133,18 @@
 		_, err := client.TrySendMessage(ctx, t.peerTopic, value)
 		if err != nil {
 			if cerror.ErrPeerMessageSendTryAgain.Equal(err) {
+				t.ignoreCount++
+				if time.Since(t.lastPrintTime) > 30*time.Second {
+					log.Warn("schedulerv3: message send failed since ignored, retry later",
+						zap.String("namespace", t.changefeed.Namespace),
+						zap.String("changefeed", t.changefeed.ID),
+						zap.String("to", to),
+						zap.Int64("ignoreCount", t.ignoreCount),
+						zap.Int64("totalMsg", t.totalMsg),
+						zap.Float64("ignoreRate", float64(t.ignoreCount)/float64(t.totalMsg)),
+						zap.String("role", string(t.role)),
+					)
+				}
 				return nil
 			}
 			if cerror.ErrPeerMessageClientClosed.Equal(err) {

pkg/cmd/server/server_test.go

@@ -183,12 +183,12 @@ func TestParseCfg(t *testing.T) {
 			// We expect the default configuration here.
 			Messages: &config.MessagesConfig{
 				ClientMaxBatchInterval:       config.TomlDuration(time.Millisecond * 10),
-				ClientMaxBatchSize:           8 * 1024 * 1024,
-				ClientMaxBatchCount:          128,
+				ClientMaxBatchSize:           64 * 1024 * 1024,
+				ClientMaxBatchCount:          1024,
 				ClientRetryRateLimit:         1.0,
 				ServerMaxPendingMessageCount: 102400,
 				ServerAckInterval:            config.TomlDuration(time.Millisecond * 100),
-				ServerWorkerPoolSize:         4,
+				ServerWorkerPoolSize:         8,
 				MaxRecvMsgSize:               256 * 1024 * 1024,
 				KeepAliveTimeout:             config.TomlDuration(time.Second * 10),
 				KeepAliveTime:                config.TomlDuration(time.Second * 30),
@@ -469,12 +469,12 @@ cert-allowed-cn = ["dd","ee"]
 			// We expect the default configuration here.
 			Messages: &config.MessagesConfig{
 				ClientMaxBatchInterval:       config.TomlDuration(time.Millisecond * 10),
-				ClientMaxBatchSize:           8 * 1024 * 1024,
-				ClientMaxBatchCount:          128,
+				ClientMaxBatchSize:           64 * 1024 * 1024,
+				ClientMaxBatchCount:          1024,
 				ClientRetryRateLimit:         1.0,
 				ServerMaxPendingMessageCount: 102400,
 				ServerAckInterval:            config.TomlDuration(time.Millisecond * 100),
-				ServerWorkerPoolSize:         4,
+				ServerWorkerPoolSize:         8,
 				MaxRecvMsgSize:               256 * 1024 * 1024,
 				KeepAliveTimeout:             config.TomlDuration(time.Second * 10),
 				KeepAliveTime:                config.TomlDuration(time.Second * 30),
@@ -534,12 +534,12 @@ unknown3 = 3
 		// We expect the default configuration here.
 		Messages: &config.MessagesConfig{
 			ClientMaxBatchInterval:       config.TomlDuration(time.Millisecond * 10),
-			ClientMaxBatchSize:           8 * 1024 * 1024,
-			ClientMaxBatchCount:          128,
+			ClientMaxBatchSize:           64 * 1024 * 1024,
+			ClientMaxBatchCount:          1024,
 			ClientRetryRateLimit:         1.0,
 			ServerMaxPendingMessageCount: 102400,
 			ServerAckInterval:            config.TomlDuration(time.Millisecond * 100),
-			ServerWorkerPoolSize:         4,
+			ServerWorkerPoolSize:         8,
 			MaxRecvMsgSize:               256 * 1024 * 1024,
 			KeepAliveTimeout:             config.TomlDuration(time.Second * 10),
 			KeepAliveTime:                config.TomlDuration(time.Second * 30),

pkg/config/config_test_data.go

@@ -166,12 +166,12 @@ const (
     },
     "messages": {
       "client-max-batch-interval": 10000000,
-      "client-max-batch-size": 8388608,
-      "client-max-batch-count": 128,
+      "client-max-batch-size": 67108864,
+      "client-max-batch-count": 1024,
       "client-retry-rate-limit": 1,
       "server-max-pending-message-count": 102400,
       "server-ack-interval": 100000000,
-      "server-worker-pool-size": 4,
+      "server-worker-pool-size": 8,
       "max-recv-msg-size": 268435456,
       "keep-alive-time": 30000000000,
       "keep-alive-timeout": 10000000000

pkg/config/messages.go

@@ -49,12 +49,12 @@ type MessagesConfig struct {
 var defaultMessageConfig = &MessagesConfig{
 	// Note that ClientMaxBatchInterval may increase the checkpoint latency.
 	ClientMaxBatchInterval:       TomlDuration(time.Millisecond * 10),
-	ClientMaxBatchSize:           8 * 1024 * 1024, // 8MB
-	ClientMaxBatchCount:          128,
+	ClientMaxBatchSize:           64 * 1024 * 1024, // 64MB
+	ClientMaxBatchCount:          1024,
 	ClientRetryRateLimit:         1.0, // Once per second
 	ServerMaxPendingMessageCount: 102400,
 	ServerAckInterval:            TomlDuration(time.Millisecond * 100),
-	ServerWorkerPoolSize:         4,
+	ServerWorkerPoolSize:         8,
 	MaxRecvMsgSize:               defaultMaxRecvMsgSize,
 	KeepAliveTime:                TomlDuration(time.Second * 30),
 	KeepAliveTimeout:             TomlDuration(time.Second * 10),
@@ -66,7 +66,7 @@ const (
 
 	// clientSendChannelSize represents the size of an internal channel used to buffer
 	// unsent messages.
-	clientSendChannelSize = 128
+	clientSendChannelSize = 1024
 
 	// clientDialTimeout represents the timeout given to gRPC to dial. 5 seconds seems reasonable
 	// because it is unlikely that the latency between TiCDC nodes is larger than 5 seconds.