From e740a1013656c035d4762b492fa84440f3f266b0 Mon Sep 17 00:00:00 2001 From: wlwilliamx Date: Mon, 10 Aug 2026 16:06:55 +0800 Subject: [PATCH 1/2] redo: make row flush batch size configurable Let file size and the flush interval control persistence by default. Add consistent.flush-batch-size and wire it through both writer backends. close pingcap/ticdc#5936 --- api/v2/changefeed_toml_test.go | 2 + api/v2/model.go | 14 ++- api/v2/model_test.go | 8 +- downstreamadapter/sink/redo/sink.go | 6 +- downstreamadapter/sink/redo/sink_test.go | 9 +- pkg/config/consistent.go | 10 ++ pkg/config/replica_config.go | 1 + pkg/config/replica_config_test.go | 38 ++++++ pkg/redo/config.go | 4 +- pkg/redo/testutil/config.go | 2 + pkg/redo/writer/config.go | 8 ++ pkg/redo/writer/file/file.go | 12 +- pkg/redo/writer/file/file_test.go | 63 +++++++++- pkg/redo/writer/memory/file_worker.go | 6 +- pkg/redo/writer/memory/file_worker_test.go | 131 +++++++++++++++++++++ pkg/redo/writer/writer_test.go | 5 + 16 files changed, 302 insertions(+), 17 deletions(-) create mode 100644 pkg/redo/writer/memory/file_worker_test.go diff --git a/api/v2/changefeed_toml_test.go b/api/v2/changefeed_toml_test.go index 459d26fc42..7b2014e003 100644 --- a/api/v2/changefeed_toml_test.go +++ b/api/v2/changefeed_toml_test.go @@ -102,6 +102,7 @@ func TestChangeFeedInfoTOMLRoundTripToInternal(t *testing.T) { Level: util.AddressOf("eventual"), MaxLogSize: util.AddressOf(int64(128)), FlushIntervalInMs: util.AddressOf(int64(2000)), + FlushBatchSize: util.AddressOf(2048), Storage: util.AddressOf("s3://test"), }, }, @@ -131,6 +132,7 @@ func TestChangeFeedInfoTOMLRoundTripToInternal(t *testing.T) { require.Equal(t, 10*time.Minute, *wrapper.Config.SyncPointInterval) require.Equal(t, "correctness", util.GetOrZero(wrapper.Config.Integrity.IntegrityCheckLevel)) require.Equal(t, "eventual", util.GetOrZero(wrapper.Config.Consistent.Level)) + require.Equal(t, 2048, util.GetOrZero(wrapper.Config.Consistent.FlushBatchSize)) } // TestDefaultConfigTOMLRoundTripToInternal encodes the full default replica diff --git a/api/v2/model.go b/api/v2/model.go index b96788e467..f8fa4d2a2d 100644 --- a/api/v2/model.go +++ b/api/v2/model.go @@ -328,6 +328,9 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig( if c.Consistent.FlushIntervalInMs != nil { res.Consistent.FlushIntervalInMs = c.Consistent.FlushIntervalInMs } + if c.Consistent.FlushBatchSize != nil { + res.Consistent.FlushBatchSize = c.Consistent.FlushBatchSize + } if c.Consistent.MetaFlushIntervalInMs != nil { res.Consistent.MetaFlushIntervalInMs = c.Consistent.MetaFlushIntervalInMs } @@ -981,6 +984,9 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig { if cloned.Consistent.FlushIntervalInMs != nil { res.Consistent.FlushIntervalInMs = cloned.Consistent.FlushIntervalInMs } + if cloned.Consistent.FlushBatchSize != nil { + res.Consistent.FlushBatchSize = cloned.Consistent.FlushBatchSize + } if cloned.Consistent.MetaFlushIntervalInMs != nil { res.Consistent.MetaFlushIntervalInMs = cloned.Consistent.MetaFlushIntervalInMs } @@ -1266,9 +1272,11 @@ type ColumnSelector struct { // ConsistentConfig represents replication consistency config for a changefeed // This is a duplicate of config.ConsistentConfig type ConsistentConfig struct { - Level *string `json:"level,omitempty" toml:"level,omitempty"` - MaxLogSize *int64 `json:"max_log_size,omitempty" toml:"max-log-size,omitempty"` - FlushIntervalInMs *int64 `json:"flush_interval,omitempty" toml:"flush-interval,omitempty"` + Level *string `json:"level,omitempty" toml:"level,omitempty"` + MaxLogSize *int64 `json:"max_log_size,omitempty" toml:"max-log-size,omitempty"` + FlushIntervalInMs *int64 `json:"flush_interval,omitempty" toml:"flush-interval,omitempty"` + // FlushBatchSize is the row-count flush threshold. Zero disables it. + FlushBatchSize *int `json:"flush_batch_size,omitempty" toml:"flush-batch-size,omitempty"` MetaFlushIntervalInMs *int64 `json:"meta_flush_interval,omitempty" toml:"meta-flush-interval,omitempty"` EncodingWorkerNum *int `json:"encoding_worker_num,omitempty" toml:"encoding-worker-num,omitempty"` FlushWorkerNum *int `json:"flush_worker_num,omitempty" toml:"flush-worker-num,omitempty"` diff --git a/api/v2/model_test.go b/api/v2/model_test.go index 3a194a3f9f..cee72282f0 100644 --- a/api/v2/model_test.go +++ b/api/v2/model_test.go @@ -192,23 +192,29 @@ func TestReplicaConfigConversionBatchFields(t *testing.T) { require.Nil(t, apiNoBatch.EventCollectorBatchBytes) } -func TestReplicaConfigConversionRedoBatchField(t *testing.T) { +// TestReplicaConfigConversionRedoBatchFields verifies redo-specific batch +// settings survive API-to-internal and internal-to-API conversion unchanged. +func TestReplicaConfigConversionRedoBatchFields(t *testing.T) { t.Parallel() apiCfg := &ReplicaConfig{ Consistent: &ConsistentConfig{ EventCollectorBatchCount: util.AddressOf(4096), + FlushBatchSize: util.AddressOf(2048), }, } internalCfg := apiCfg.ToInternalReplicaConfig() require.NotNil(t, internalCfg.Consistent) require.Equal(t, 4096, util.GetOrZero(internalCfg.Consistent.EventCollectorBatchCount)) + require.Equal(t, 2048, util.GetOrZero(internalCfg.Consistent.FlushBatchSize)) apiCfgBack := ToAPIReplicaConfig(internalCfg) require.NotNil(t, apiCfgBack.Consistent) require.NotNil(t, apiCfgBack.Consistent.EventCollectorBatchCount) require.Equal(t, 4096, *apiCfgBack.Consistent.EventCollectorBatchCount) + require.NotNil(t, apiCfgBack.Consistent.FlushBatchSize) + require.Equal(t, 2048, *apiCfgBack.Consistent.FlushBatchSize) } func TestReplicaConfigConversionMySQLAsyncDDLTimeout(t *testing.T) { diff --git a/downstreamadapter/sink/redo/sink.go b/downstreamadapter/sink/redo/sink.go index d19fbf9c1e..52a524f076 100644 --- a/downstreamadapter/sink/redo/sink.go +++ b/downstreamadapter/sink/redo/sink.go @@ -33,6 +33,10 @@ import ( "golang.org/x/sync/errgroup" ) +// dmlWriterInputBatchSize bounds each transfer from the unlimited sink buffer +// into the writer. It is independent of the writer's persistent flush policy. +const dmlWriterInputBatchSize = 1024 + // Sink manages redo log writer, buffers un-persistent redo logs, calculates // redo log resolved ts. It implements Sink interface. type Sink struct { @@ -237,7 +241,7 @@ func (s *Sink) Close() { } func (s *Sink) sendMessages(ctx context.Context) error { - buffer := make([]*commonEvent.RedoRowEvent, 0, redo.DefaultFlushBatchSize) + buffer := make([]*commonEvent.RedoRowEvent, 0, dmlWriterInputBatchSize) for { select { case <-ctx.Done(): diff --git a/downstreamadapter/sink/redo/sink_test.go b/downstreamadapter/sink/redo/sink_test.go index 1207572f4d..388e7be1f5 100644 --- a/downstreamadapter/sink/redo/sink_test.go +++ b/downstreamadapter/sink/redo/sink_test.go @@ -340,6 +340,9 @@ func runBenchTest(b *testing.B, storage string, useFileBackend bool) { require.ErrorIs(b, eg.Wait(), context.Canceled) } +// TestRedoSinkSendMessagesInBatch fills the unlimited buffer, drains it through +// the bounded writer-input batches, and verifies this transport batch remains +// independent of the configurable persistent flush threshold. func TestRedoSinkSendMessagesInBatch(t *testing.T) { t.Parallel() @@ -365,8 +368,8 @@ func TestRedoSinkSendMessagesInBatch(t *testing.T) { } gomock.InOrder( - expectWriteBatch(redo.DefaultFlushBatchSize), - expectWriteBatch(redo.DefaultFlushBatchSize), + expectWriteBatch(dmlWriterInputBatchSize), + expectWriteBatch(dmlWriterInputBatchSize), expectWriteBatch(17), ) @@ -380,7 +383,7 @@ func TestRedoSinkSendMessagesInBatch(t *testing.T) { doneCh <- s.sendMessages(ctx) }() - totalEvents := redo.DefaultFlushBatchSize*2 + 17 + totalEvents := dmlWriterInputBatchSize*2 + 17 events := make([]*commonEvent.RedoRowEvent, 0, totalEvents) for range totalEvents { events = append(events, &commonEvent.RedoRowEvent{}) diff --git a/pkg/config/consistent.go b/pkg/config/consistent.go index eb151d7598..d8cb195761 100644 --- a/pkg/config/consistent.go +++ b/pkg/config/consistent.go @@ -37,6 +37,9 @@ type ConsistentConfig struct { // FlushIntervalInMs is the flush interval(ms) of redo log to flush log to storage. // Default is 2000ms. FlushIntervalInMs *int64 `toml:"flush-interval" json:"flush-interval,omitempty"` + // FlushBatchSize is the number of row events that triggers a redo log flush. + // A value of 0 disables count-based flushing. Default is 0. + FlushBatchSize *int `toml:"flush-batch-size" json:"flush-batch-size,omitempty"` // MetaFlushIntervalInMs is the flush interval(ms) of redo log to // flush meta(resolvedTs and checkpointTs) to storage. // Default is 200ms. @@ -101,6 +104,13 @@ func (c *ConsistentConfig) validateAndAdjust(enableIOCheck bool) error { fmt.Sprintf("The consistent.flush-interval:%d must be equal or greater than %d", util.GetOrZero(c.FlushIntervalInMs), redo.MinFlushIntervalInMs)) } + if c.FlushBatchSize == nil { + c.FlushBatchSize = util.AddressOf(redo.DefaultFlushBatchSize) + } + if *c.FlushBatchSize < 0 { + return errors.ErrInvalidReplicaConfig.FastGenByArgs( + "consistent.flush-batch-size must be set not smaller than 0") + } if util.GetOrZero(c.MetaFlushIntervalInMs) == 0 { c.MetaFlushIntervalInMs = util.AddressOf(int64(redo.DefaultMetaFlushIntervalInMs)) diff --git a/pkg/config/replica_config.go b/pkg/config/replica_config.go index d5298a08fd..319b1c6e7e 100644 --- a/pkg/config/replica_config.go +++ b/pkg/config/replica_config.go @@ -92,6 +92,7 @@ var defaultReplicaConfig = &ReplicaConfig{ Level: util.AddressOf("none"), MaxLogSize: util.AddressOf(redo.DefaultMaxLogSize), FlushIntervalInMs: util.AddressOf(int64(redo.DefaultFlushIntervalInMs)), + FlushBatchSize: util.AddressOf(redo.DefaultFlushBatchSize), MetaFlushIntervalInMs: util.AddressOf(int64(redo.DefaultMetaFlushIntervalInMs)), EncodingWorkerNum: util.AddressOf(redo.DefaultEncodingWorkerNum), FlushWorkerNum: util.AddressOf(redo.DefaultFlushWorkerNum), diff --git a/pkg/config/replica_config_test.go b/pkg/config/replica_config_test.go index 3a5aba8b58..1f70aece8d 100644 --- a/pkg/config/replica_config_test.go +++ b/pkg/config/replica_config_test.go @@ -242,6 +242,44 @@ func TestReplicaConfigValidateBatchConfig(t *testing.T) { assertBatchConfig(nil, util.AddressOf(-1), "event-collector-batch-bytes") } +// TestConsistentFlushBatchSizeValidation verifies that adjustment supplies the +// disabled-by-default value, preserves non-negative overrides, and rejects a +// negative row-count threshold before the redo writer is created. +func TestConsistentFlushBatchSizeValidation(t *testing.T) { + require.Equal(t, 0, util.GetOrZero(GetDefaultReplicaConfig().Consistent.FlushBatchSize)) + + tests := []struct { + name string + value *int + wantValue int + wantErr bool + }{ + {name: "unset uses disabled default", wantValue: 0}, + {name: "zero disables count based flush", value: util.AddressOf(0), wantValue: 0}, + {name: "positive value enables count based flush", value: util.AddressOf(2048), wantValue: 2048}, + {name: "negative value is rejected", value: util.AddressOf(-1), wantErr: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := &ConsistentConfig{ + Level: util.AddressOf("eventual"), + FlushBatchSize: tt.value, + Storage: util.AddressOf("blackhole://"), + } + + err := cfg.validateAndAdjust(false) + if tt.wantErr { + require.ErrorContains(t, err, "consistent.flush-batch-size") + return + } + require.NoError(t, err) + require.NotNil(t, cfg.FlushBatchSize) + require.Equal(t, tt.wantValue, *cfg.FlushBatchSize) + }) + } +} + func TestReplicaConfig_EnableRedoIOCheck_DefaultValue(t *testing.T) { config := GetDefaultReplicaConfig() require.True(t, util.GetOrZero(config.EnableRedoIOCheck)) diff --git a/pkg/redo/config.go b/pkg/redo/config.go index 6cce071a54..9238573895 100644 --- a/pkg/redo/config.go +++ b/pkg/redo/config.go @@ -49,8 +49,8 @@ const ( DefaultMetaFlushIntervalInMs = 200 // MinFlushIntervalInMs is the minimum flush interval for redo log. MinFlushIntervalInMs = 50 - // DefaultFlushBatchSize is the default flush batch size for redo log. - DefaultFlushBatchSize = 1024 + // DefaultFlushBatchSize disables count-based redo log flushing by default. + DefaultFlushBatchSize = 0 // DefaultEncodingWorkerNum is the default number of encoding workers. DefaultEncodingWorkerNum = 16 diff --git a/pkg/redo/testutil/config.go b/pkg/redo/testutil/config.go index ef1cfe35b1..47126caae3 100644 --- a/pkg/redo/testutil/config.go +++ b/pkg/redo/testutil/config.go @@ -24,6 +24,7 @@ func NewConsistentConfig(storage string) *config.ConsistentConfig { level := string(redo.ConsistentLevelEventual) maxLogSize := int64(redo.DefaultMaxLogSize) flushIntervalInMs := int64(redo.DefaultFlushIntervalInMs) + flushBatchSize := redo.DefaultFlushBatchSize metaFlushIntervalInMs := int64(redo.MinFlushIntervalInMs) encodingWorkerNum := redo.DefaultEncodingWorkerNum flushWorkerNum := redo.DefaultFlushWorkerNum @@ -35,6 +36,7 @@ func NewConsistentConfig(storage string) *config.ConsistentConfig { MaxLogSize: util.AddressOf(maxLogSize), Storage: util.AddressOf(storage), FlushIntervalInMs: util.AddressOf(flushIntervalInMs), + FlushBatchSize: util.AddressOf(flushBatchSize), MetaFlushIntervalInMs: util.AddressOf(metaFlushIntervalInMs), EncodingWorkerNum: util.AddressOf(encodingWorkerNum), FlushWorkerNum: util.AddressOf(flushWorkerNum), diff --git a/pkg/redo/writer/config.go b/pkg/redo/writer/config.go index b9781f7e07..7147f9d2af 100644 --- a/pkg/redo/writer/config.go +++ b/pkg/redo/writer/config.go @@ -43,6 +43,8 @@ type Config struct { // Shared by file and memory backends as the flush ticker interval. flushIntervalInMs int64 + // Shared by file and memory backends as the optional row-count flush threshold. + flushBatchSize int // Used only by the memory backend for encoding workers. encodingWorkerNum int @@ -77,6 +79,7 @@ func NewConfig(changefeedID common.ChangeFeedID, consistentCfg *config.Consisten maxLogSizeInBytes: util.GetOrZero(consistentCfg.MaxLogSize) * redo.Megabyte, useFileBackend: util.GetOrZero(consistentCfg.UseFileBackend), flushIntervalInMs: util.GetOrZero(consistentCfg.FlushIntervalInMs), + flushBatchSize: util.GetOrZero(consistentCfg.FlushBatchSize), encodingWorkerNum: util.GetOrZero(consistentCfg.EncodingWorkerNum), flushWorkerNum: util.GetOrZero(consistentCfg.FlushWorkerNum), compression: util.GetOrZero(consistentCfg.Compression), @@ -153,6 +156,11 @@ func (cfg *Config) FlushIntervalInMs() int64 { return cfg.flushIntervalInMs } +// FlushBatchSize returns the row-count flush threshold. Zero disables it. +func (cfg *Config) FlushBatchSize() int { + return cfg.flushBatchSize +} + func (cfg *Config) EncodingWorkerNum() int { return cfg.encodingWorkerNum } diff --git a/pkg/redo/writer/file/file.go b/pkg/redo/writer/file/file.go index 759d429337..de2f80ebaa 100644 --- a/pkg/redo/writer/file/file.go +++ b/pkg/redo/writer/file/file.go @@ -58,6 +58,7 @@ type fileWriterConfig interface { Dir() string MaxLogSizeInBytes() int64 FlushIntervalInMs() int64 + FlushBatchSize() int FlushWorkerNum() int UseExternalStorage() bool } @@ -66,6 +67,7 @@ type localFileConfig struct { dir string maxLogSizeInBytes int64 flushIntervalInMs int64 + flushBatchSize int flushWorkerNum int } @@ -89,6 +91,10 @@ func (cfg *localFileConfig) FlushIntervalInMs() int64 { return cfg.flushIntervalInMs } +func (cfg *localFileConfig) FlushBatchSize() int { + return cfg.flushBatchSize +} + func (cfg *localFileConfig) FlushWorkerNum() int { return cfg.flushWorkerNum } @@ -356,7 +362,8 @@ func (w *Writer) encode(ctx context.Context) error { ticker := time.NewTicker(d) defer ticker.Stop() num := 0 - cacheEventPostFlush := make([]func(), 0, redo.DefaultFlushBatchSize) + flushBatchSize := w.cfg.FlushBatchSize() + cacheEventPostFlush := make([]func(), 0) flush := func() error { err := w.Flush() if err != nil { @@ -384,7 +391,8 @@ func (w *Writer) encode(ctx context.Context) error { return err } num++ - if num >= redo.DefaultFlushBatchSize { + // Zero leaves file size and the periodic ticker as the only flush triggers. + if flushBatchSize > 0 && num >= flushBatchSize { err := flush() if err != nil { return errors.Trace(err) diff --git a/pkg/redo/writer/file/file_test.go b/pkg/redo/writer/file/file_test.go index 02f890ec47..d3ede0c5ee 100644 --- a/pkg/redo/writer/file/file_test.go +++ b/pkg/redo/writer/file/file_test.go @@ -436,17 +436,22 @@ func TestRotateFileWithoutFileAllocator(t *testing.T) { w.Close() } +// TestRunFlushesOnBatchBoundaryAndExecutesPostFlush configures a small row +// threshold, writes up to that boundary, and verifies callbacks run only after +// the configured count causes the file backend to flush. func TestRunFlushesOnBatchBoundaryAndExecutesPostFlush(t *testing.T) { t.Parallel() dir := t.TempDir() flushIntervalInMs := int64(60 * 1000) + flushBatchSize := 4 flushWorkerNum := 9 batchWriterCfg := newTestWriterConfig( t, common.NewChangeFeedIDWithName("test-run-batch", common.DefaultKeyspaceName), &config.ConsistentConfig{ FlushIntervalInMs: &flushIntervalInMs, + FlushBatchSize: &flushBatchSize, FlushWorkerNum: &flushWorkerNum, Storage: util.AddressOf("file://" + dir), }, @@ -461,7 +466,7 @@ func TestRunFlushesOnBatchBoundaryAndExecutesPostFlush(t *testing.T) { }() postFlushCnt := atomic.NewInt64(0) - for i := 0; i < redo.DefaultFlushBatchSize-1; i++ { + for i := 0; i < flushBatchSize-1; i++ { ts := uint64(i + 1) w.GetInputCh() <- &pevent.RedoRowEvent{ StartTs: ts, @@ -480,7 +485,7 @@ func TestRunFlushesOnBatchBoundaryAndExecutesPostFlush(t *testing.T) { default: } - ts := uint64(redo.DefaultFlushBatchSize) + ts := uint64(flushBatchSize) w.GetInputCh() <- &pevent.RedoRowEvent{ StartTs: ts, CommitTs: ts, @@ -490,10 +495,62 @@ func TestRunFlushesOnBatchBoundaryAndExecutesPostFlush(t *testing.T) { } require.Eventually(t, func() bool { - return postFlushCnt.Load() == int64(redo.DefaultFlushBatchSize) + return postFlushCnt.Load() == int64(flushBatchSize) }, 10*time.Second, 20*time.Millisecond) cancel() require.ErrorIs(t, <-runErrCh, context.Canceled) require.NoError(t, w.Close()) } + +// TestRunDisablesCountBasedFlushWithZero writes more rows than the old fixed +// boundary while using a long ticker interval, then verifies the file backend +// processes them without executing callbacks through a row-count flush. +func TestRunDisablesCountBasedFlushWithZero(t *testing.T) { + t.Parallel() + + const legacyFlushBatchSize = 1024 + dir := t.TempDir() + flushIntervalInMs := int64(60 * 1000) + flushBatchSize := 0 + flushWorkerNum := 9 + writerCfg := newTestWriterConfig( + t, + common.NewChangeFeedIDWithName("test-run-disabled-batch", common.DefaultKeyspaceName), + &config.ConsistentConfig{ + FlushIntervalInMs: &flushIntervalInMs, + FlushBatchSize: &flushBatchSize, + FlushWorkerNum: &flushWorkerNum, + Storage: util.AddressOf("file://" + dir), + }, + ) + w, err := NewFileWriter(context.Background(), writerCfg, redo.RedoRowLogFileType) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + runErrCh := make(chan error, 1) + go func() { + runErrCh <- w.Run(ctx) + }() + + postFlushCnt := atomic.NewInt64(0) + lastCommitTs := uint64(legacyFlushBatchSize + 2) + for ts := uint64(1); ts <= lastCommitTs; ts++ { + w.GetInputCh() <- &pevent.RedoRowEvent{ + StartTs: ts, + CommitTs: ts, + Callback: func() { + postFlushCnt.Inc() + }, + } + } + + require.Eventually(t, func() bool { + return w.eventCommitTS.Load() == lastCommitTs + }, 10*time.Second, 20*time.Millisecond) + require.Zero(t, postFlushCnt.Load()) + + cancel() + require.ErrorIs(t, <-runErrCh, context.Canceled) + require.NoError(t, w.Close()) +} diff --git a/pkg/redo/writer/memory/file_worker.go b/pkg/redo/writer/memory/file_worker.go index ba67651544..9689345a1e 100644 --- a/pkg/redo/writer/memory/file_worker.go +++ b/pkg/redo/writer/memory/file_worker.go @@ -213,7 +213,8 @@ func (f *fileWorkerGroup) bgWriteLogs( ticker := time.NewTicker(d) defer ticker.Stop() num := 0 - cacheEventPostFlush := make([]func(), 0, redo.DefaultFlushBatchSize) + flushBatchSize := f.cfg.FlushBatchSize() + cacheEventPostFlush := make([]func(), 0) flush := func() error { err := f.flushAll(egCtx) if err != nil { @@ -245,7 +246,8 @@ func (f *fileWorkerGroup) bgWriteLogs( return errors.Trace(err) } num++ - if num > redo.DefaultFlushBatchSize { + // Zero leaves file size and the periodic ticker as the only flush triggers. + if flushBatchSize > 0 && num >= flushBatchSize { err := flush() if err != nil { return errors.Trace(err) diff --git a/pkg/redo/writer/memory/file_worker_test.go b/pkg/redo/writer/memory/file_worker_test.go new file mode 100644 index 0000000000..eded46bd9a --- /dev/null +++ b/pkg/redo/writer/memory/file_worker_test.go @@ -0,0 +1,131 @@ +// Copyright 2026 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 memory + +import ( + "context" + "testing" + "time" + + "github.com/pingcap/ticdc/pkg/common" + "github.com/pingcap/ticdc/pkg/redo/testutil" + "github.com/pingcap/ticdc/pkg/redo/writer" + "github.com/pingcap/ticdc/pkg/util" + "github.com/stretchr/testify/require" + "go.uber.org/atomic" +) + +func newTestFileWorkerGroup( + t *testing.T, inputCh chan *polymorphicRedoEvent, flushBatchSize int, +) *fileWorkerGroup { + consistentCfg := testutil.NewConsistentConfig("blackhole://") + consistentCfg.MaxLogSize = util.AddressOf(int64(1)) + consistentCfg.FlushIntervalInMs = util.AddressOf(int64(time.Hour / time.Millisecond)) + consistentCfg.FlushBatchSize = util.AddressOf(flushBatchSize) + consistentCfg.FlushWorkerNum = util.AddressOf(1) + cfg, err := writer.NewConfig( + common.NewChangeFeedIDWithName(t.Name(), common.DefaultKeyspaceName), + consistentCfg, + ) + require.NoError(t, err) + return newFileWorkerGroup(cfg, inputCh, nil) +} + +// TestFileWorkerFlushesAtConfiguredBatchSize configures a three-row threshold, +// feeds exactly three encoded events, acknowledges the sealed file, and verifies +// that all callbacks run only after the count-based flush completes. +func TestFileWorkerFlushesAtConfiguredBatchSize(t *testing.T) { + inputCh := make(chan *polymorphicRedoEvent) + fileWorkers := newTestFileWorkerGroup(t, inputCh, 3) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + runErrCh := make(chan error, 1) + go func() { + runErrCh <- fileWorkers.bgWriteLogs(ctx, inputCh) + }() + + fileFlushedCh := make(chan struct{}) + go func() { + file := <-fileWorkers.flushCh + file.markFlushed() + close(fileFlushedCh) + }() + + postFlushCount := atomic.NewInt64(0) + for i := 1; i <= 3; i++ { + inputCh <- &polymorphicRedoEvent{ + commitTs: uint64(i), + data: []byte{byte(i)}, + callback: func() { + postFlushCount.Inc() + }, + } + } + + require.Eventually(t, func() bool { + return postFlushCount.Load() == 3 + }, time.Second, 10*time.Millisecond) + <-fileFlushedCh + + cancel() + require.ErrorIs(t, <-runErrCh, context.Canceled) +} + +// TestFileWorkerDisablesCountBasedFlushWithZero feeds more rows than the old +// hard-coded limit while the ticker and file-size limit cannot fire. Processing +// must continue without waiting for a count-triggered file flush or callbacks. +func TestFileWorkerDisablesCountBasedFlushWithZero(t *testing.T) { + const legacyFlushBatchSize = 1024 + + inputCh := make(chan *polymorphicRedoEvent) + fileWorkers := newTestFileWorkerGroup(t, inputCh, 0) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + runErrCh := make(chan error, 1) + go func() { + runErrCh <- fileWorkers.bgWriteLogs(ctx, inputCh) + }() + + postFlushCount := atomic.NewInt64(0) + sendDoneCh := make(chan struct{}) + go func() { + defer close(sendDoneCh) + for i := 1; i <= legacyFlushBatchSize+2; i++ { + select { + case <-ctx.Done(): + return + case inputCh <- &polymorphicRedoEvent{ + commitTs: uint64(i), + data: []byte{byte(i)}, + callback: func() { + postFlushCount.Inc() + }, + }: + } + } + }() + + select { + case <-sendDoneCh: + case <-time.After(5 * time.Second): + cancel() + require.FailNow(t, "file worker blocked on a disabled count-based flush") + } + require.Zero(t, postFlushCount.Load()) + + cancel() + require.ErrorIs(t, <-runErrCh, context.Canceled) +} diff --git a/pkg/redo/writer/writer_test.go b/pkg/redo/writer/writer_test.go index c49d022951..214d32bc73 100644 --- a/pkg/redo/writer/writer_test.go +++ b/pkg/redo/writer/writer_test.go @@ -28,9 +28,12 @@ import ( func TestNewConfigUsesConsistentConfigValues(t *testing.T) { t.Parallel() + // Configure every writer-owned option, build the runtime config, and verify + // that each value, including the optional row flush threshold, is preserved. changefeedID := common.NewChangeFeedIDWithName("test-cf", common.DefaultKeyspaceName) maxLogSize := int64(128) flushIntervalInMs := int64(1234) + flushBatchSize := 2048 encodingWorkerNum := 5 flushWorkerNum := 6 compressionType := "lz4" @@ -38,6 +41,7 @@ func TestNewConfigUsesConsistentConfigValues(t *testing.T) { consistentCfg := testutil.NewConsistentConfig("nfs:///tmp/redo") consistentCfg.MaxLogSize = util.AddressOf(maxLogSize) consistentCfg.FlushIntervalInMs = util.AddressOf(flushIntervalInMs) + consistentCfg.FlushBatchSize = util.AddressOf(flushBatchSize) consistentCfg.EncodingWorkerNum = util.AddressOf(encodingWorkerNum) consistentCfg.FlushWorkerNum = util.AddressOf(flushWorkerNum) consistentCfg.Compression = util.AddressOf(compressionType) @@ -53,6 +57,7 @@ func TestNewConfigUsesConsistentConfigValues(t *testing.T) { require.True(t, cfg.UseExternalStorage()) require.Equal(t, maxLogSize*redo.Megabyte, cfg.MaxLogSizeInBytes()) require.Equal(t, flushIntervalInMs, cfg.FlushIntervalInMs()) + require.Equal(t, flushBatchSize, cfg.FlushBatchSize()) require.Equal(t, encodingWorkerNum, cfg.EncodingWorkerNum()) require.Equal(t, flushWorkerNum, cfg.FlushWorkerNum()) require.Equal(t, flushConcurrency, cfg.FlushConcurrency()) From 41f3158b83e3055dcc39aba56504189ed41736d3 Mon Sep 17 00:00:00 2001 From: wlwilliamx Date: Mon, 10 Aug 2026 17:58:48 +0800 Subject: [PATCH 2/2] redo: release callbacks after size rotation Release post-flush callbacks as soon as size-rotated redo files are durable. Clear invoked callback slots to avoid retaining receivers through slice capacity, while preserving callback order across concurrent memory-backend uploads. --- pkg/redo/writer/file/file.go | 56 ++++++++++----- pkg/redo/writer/file/file_test.go | 66 +++++++++++++++++ pkg/redo/writer/memory/file_worker.go | 83 +++++++++++++++++----- pkg/redo/writer/memory/file_worker_test.go | 64 +++++++++++++++++ 4 files changed, 231 insertions(+), 38 deletions(-) diff --git a/pkg/redo/writer/file/file.go b/pkg/redo/writer/file/file.go index de2f80ebaa..b42a0f87d7 100644 --- a/pkg/redo/writer/file/file.go +++ b/pkg/redo/writer/file/file.go @@ -228,24 +228,33 @@ func (w *Writer) Run(ctx context.Context) error { // Write implement write interface // TODO: more general api with fileName generated by caller func (w *Writer) Write(rawData []byte) (int, error) { + n, _, err := w.writeRawData(rawData) + return n, err +} + +// writeRawData reports whether writing rawData first made the previous file +// durable through a size-triggered rotation. +func (w *Writer) writeRawData(rawData []byte) (int, bool, error) { w.Lock() defer w.Unlock() writeLen := int64(len(rawData)) if writeLen > w.cfg.MaxLogSizeInBytes() { - return 0, errors.ErrRedoFileSizeExceed.GenWithStackByArgs(writeLen, w.cfg.MaxLogSizeInBytes()) + return 0, false, errors.ErrRedoFileSizeExceed.GenWithStackByArgs(writeLen, w.cfg.MaxLogSizeInBytes()) } if w.file == nil { if err := w.openNew(); err != nil { - return 0, err + return 0, false, err } } + rotated := false if w.size+writeLen > w.cfg.MaxLogSizeInBytes() { if err := w.rotate(); err != nil { - return 0, err + return 0, false, err } + rotated = true } if w.maxCommitTS.Load() < w.eventCommitTS.Load() { @@ -254,7 +263,7 @@ func (w *Writer) Write(rawData []byte) (int, error) { // ref: https://github.com/etcd-io/etcd/pull/5250 lenField, padBytes := writer.EncodeFrameSize(len(rawData)) if err := w.writeUint64(lenField, w.uint64buf); err != nil { - return 0, err + return 0, rotated, err } if padBytes != 0 { @@ -263,12 +272,12 @@ func (w *Writer) Write(rawData []byte) (int, error) { n, err := w.bw.Write(rawData) if err != nil { - return 0, err + return 0, rotated, err } w.metricWriteBytes.Add(float64(n)) w.size += int64(n) - return n, err + return n, rotated, nil } // AdvanceTs implement Advance interface @@ -327,25 +336,22 @@ func (w *Writer) GetInputCh() chan writer.RedoEvent { return w.inputCh } -func (w *Writer) write(event writer.RedoEvent) error { +func (w *Writer) write(event writer.RedoEvent) (bool, error) { rl := event.ToRedoLog() if rl.Type == commonEvent.RedoLogTypeDDL { rl.RedoDDL.SetTableSchemaStore(w.tableSchemaStore) } data, err := codec.MarshalRedoLog(rl, nil) if err != nil { - return errors.WrapError(errors.ErrMarshalFailed, err) + return false, errors.WrapError(errors.ErrMarshalFailed, err) } w.AdvanceTs(rl.GetCommitTs()) - _, err = w.Write(data) - if err != nil { - return err - } - return nil + _, rotated, err := w.writeRawData(data) + return rotated, err } func (w *Writer) SyncWrite(event writer.RedoEvent) error { - err := w.write(event) + _, err := w.write(event) if err != nil { return err } @@ -364,16 +370,22 @@ func (w *Writer) encode(ctx context.Context) error { num := 0 flushBatchSize := w.cfg.FlushBatchSize() cacheEventPostFlush := make([]func(), 0) + runCachedPostFlush := func() { + for i, fn := range cacheEventPostFlush { + // Clear the slot before invocation so retained slice capacity does not + // keep the callback receiver alive after it has been acknowledged. + cacheEventPostFlush[i] = nil + fn() + } + cacheEventPostFlush = cacheEventPostFlush[:0] + } flush := func() error { err := w.Flush() if err != nil { return err } - for _, fn := range cacheEventPostFlush { - fn() - } + runCachedPostFlush() num = 0 - cacheEventPostFlush = cacheEventPostFlush[:0] return nil } for { @@ -386,7 +398,13 @@ func (w *Writer) encode(ctx context.Context) error { return errors.Trace(err) } case e := <-w.inputCh: - err := w.write(e) + rotated, err := w.write(e) + if rotated { + // The previous file is durable before the current event is written, + // so only callbacks accumulated before this event can run here. + runCachedPostFlush() + num = 0 + } if err != nil { return err } diff --git a/pkg/redo/writer/file/file_test.go b/pkg/redo/writer/file/file_test.go index d3ede0c5ee..bff13c90bf 100644 --- a/pkg/redo/writer/file/file_test.go +++ b/pkg/redo/writer/file/file_test.go @@ -28,6 +28,7 @@ import ( "github.com/pingcap/ticdc/pkg/fsutil" "github.com/pingcap/ticdc/pkg/metrics" "github.com/pingcap/ticdc/pkg/redo" + "github.com/pingcap/ticdc/pkg/redo/codec" "github.com/pingcap/ticdc/pkg/redo/writer" "github.com/pingcap/ticdc/pkg/util" "github.com/pingcap/ticdc/pkg/uuid" @@ -503,6 +504,71 @@ func TestRunFlushesOnBatchBoundaryAndExecutesPostFlush(t *testing.T) { require.NoError(t, w.Close()) } +// TestRunReleasesCallbacksAfterSizeRotation writes one event that exactly fills +// a local redo file, confirms its callback remains pending, then writes a second +// event to rotate the first file. The first callback must run after that durable +// rotation while the second event remains unacknowledged in the current file. +func TestRunReleasesCallbacksAfterSizeRotation(t *testing.T) { + dir := t.TempDir() + firstCallbackDone := make(chan struct{}) + firstEvent := &pevent.RedoRowEvent{ + StartTs: 1, + CommitTs: 1, + Callback: func() { + close(firstCallbackDone) + }, + } + encodedFirstEvent, err := codec.MarshalRedoLog(firstEvent.ToRedoLog(), nil) + require.NoError(t, err) + _, padBytes := writer.EncodeFrameSize(len(encodedFirstEvent)) + maxLogSizeInBytes := int64(8 + len(encodedFirstEvent) + padBytes) + + w, err := newWriter(&localFileConfig{ + dir: dir, + maxLogSizeInBytes: maxLogSizeInBytes, + flushIntervalInMs: int64(time.Hour / time.Millisecond), + flushBatchSize: 0, + flushWorkerNum: 1, + }, redo.RedoRowLogFileType, nil) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + runErrCh := make(chan error, 1) + go func() { + runErrCh <- w.Run(ctx) + }() + + w.GetInputCh() <- firstEvent + require.Eventually(t, func() bool { + return w.eventCommitTS.Load() == firstEvent.CommitTs + }, 10*time.Second, 10*time.Millisecond) + select { + case <-firstCallbackDone: + require.FailNow(t, "callback ran before the file became durable") + default: + } + + secondCallbackCount := atomic.NewInt64(0) + secondEvent := &pevent.RedoRowEvent{ + StartTs: 2, + CommitTs: 2, + Callback: func() { + secondCallbackCount.Inc() + }, + } + w.GetInputCh() <- secondEvent + select { + case <-firstCallbackDone: + case <-time.After(10 * time.Second): + require.FailNow(t, "callback was not released after size rotation") + } + require.Zero(t, secondCallbackCount.Load()) + + cancel() + require.ErrorIs(t, <-runErrCh, context.Canceled) + require.NoError(t, w.Close()) +} + // TestRunDisablesCountBasedFlushWithZero writes more rows than the old fixed // boundary while using a long ticker interval, then verifies the file backend // processes them without executing callbacks through a row-count flush. diff --git a/pkg/redo/writer/memory/file_worker.go b/pkg/redo/writer/memory/file_worker.go index 9689345a1e..381df37126 100644 --- a/pkg/redo/writer/memory/file_worker.go +++ b/pkg/redo/writer/memory/file_worker.go @@ -47,6 +47,8 @@ type fileCache struct { filename string flushed chan struct{} writer *dataWriter + + postFlushCallbacks []func() } type dataWriter struct { @@ -83,6 +85,22 @@ func (f *fileCache) markFlushed() { } } +func (f *fileCache) addPostFlushCallback(callback func()) { + if callback != nil { + f.postFlushCallbacks = append(f.postFlushCallbacks, callback) + } +} + +// runPostFlushCallbacks clears each slot before invocation so the retained +// slice capacity cannot keep callback receivers alive after the file is durable. +func (f *fileCache) runPostFlushCallbacks() { + for i, callback := range f.postFlushCallbacks { + f.postFlushCallbacks[i] = nil + callback() + } + f.postFlushCallbacks = nil +} + type fileWorkerGroup struct { cfg *writer.Config op *writer.LogWriterOptions @@ -214,23 +232,27 @@ func (f *fileWorkerGroup) bgWriteLogs( defer ticker.Stop() num := 0 flushBatchSize := f.cfg.FlushBatchSize() - cacheEventPostFlush := make([]func(), 0) flush := func() error { err := f.flushAll(egCtx) if err != nil { return err } - for _, fn := range cacheEventPostFlush { - fn() - } num = 0 - cacheEventPostFlush = cacheEventPostFlush[:0] return nil } for { + // A size-rotated file can finish independently of the current file. + // Release only the durable prefix to preserve input callback order. + f.releaseFlushedFiles() + var firstRotatedFileFlushed <-chan struct{} + if len(f.files) > 1 { + firstRotatedFileFlushed = f.files[0].flushed + } select { case <-egCtx.Done(): return errors.Trace(egCtx.Err()) + case <-firstRotatedFileFlushed: + continue case <-ticker.C: err := flush() if err != nil { @@ -241,10 +263,13 @@ func (f *fileWorkerGroup) bgWriteLogs( log.Error("inputCh of redo file worker is closed unexpectedly") return errors.ErrUnexpected.FastGenByArgs("inputCh of redo file worker is closed unexpectedly") } - err := f.writeToCache(egCtx, event) + rotated, err := f.writeToCache(egCtx, event) if err != nil { return errors.Trace(err) } + if rotated { + num = 0 + } num++ // Zero leaves file size and the periodic ticker as the only flush triggers. if flushBatchSize > 0 && num >= flushBatchSize { @@ -252,9 +277,6 @@ func (f *fileWorkerGroup) bgWriteLogs( if err != nil { return errors.Trace(err) } - event.PostFlush() - } else { - cacheEventPostFlush = append(cacheEventPostFlush, event.PostFlush) } } } @@ -322,46 +344,48 @@ func (f *fileWorkerGroup) newFileCache(data []byte, commitTs common.Ts) *fileCac func (f *fileWorkerGroup) writeToCache( egCtx context.Context, event *polymorphicRedoEvent, -) (err error) { +) (rotated bool, err error) { commitTs := event.commitTs data := event.data if len(data) == 0 { - return errors.ErrUnexpected.FastGenByArgs("encoded redo event data is empty") + return false, errors.ErrUnexpected.FastGenByArgs("encoded redo event data is empty") } writeLen := int64(len(data)) if writeLen > f.cfg.MaxLogSizeInBytes() { // TODO: maybe we need to deal with the oversized commonEvent. - return errors.ErrRedoFileSizeExceed.GenWithStackByArgs(writeLen, f.cfg.MaxLogSizeInBytes()) + return false, errors.ErrRedoFileSizeExceed.GenWithStackByArgs(writeLen, f.cfg.MaxLogSizeInBytes()) } defer f.metricWriteBytes.Add(float64(writeLen)) if len(f.files) == 0 { file := f.newFileCache(data, commitTs) if file == nil { - return errors.ErrRedoWriterStopped.FastGenByArgs("failed to create file cache") + return false, errors.ErrRedoWriterStopped.FastGenByArgs("failed to create file cache") } + file.addPostFlushCallback(event.callback) f.files = append(f.files, file) - return nil + return false, nil } file := f.files[len(f.files)-1] if file.fileSize+writeLen > f.cfg.MaxLogSizeInBytes() { select { case <-egCtx.Done(): - return errors.Trace(egCtx.Err()) + return false, errors.Trace(egCtx.Err()) case f.flushCh <- file: } file := f.newFileCache(data, commitTs) if file == nil { - return errors.ErrRedoWriterStopped.FastGenByArgs("failed to create file cache") + return false, errors.ErrRedoWriterStopped.FastGenByArgs("failed to create file cache") } + file.addPostFlushCallback(event.callback) f.files = append(f.files, file) - return nil + return true, nil } _, err = file.writer.Write(data) if err != nil { - return err + return false, err } file.fileSize += writeLen @@ -371,7 +395,24 @@ func (f *fileWorkerGroup) writeToCache( if commitTs < file.minCommitTs { file.minCommitTs = commitTs } - return nil + file.addPostFlushCallback(event.callback) + return false, nil +} + +// releaseFlushedFiles invokes callbacks for the durable prefix of rotated +// files. The last file is still writable and must remain pending until a flush. +func (f *fileWorkerGroup) releaseFlushedFiles() { + for len(f.files) > 1 { + file := f.files[0] + select { + case <-file.flushed: + file.runPostFlushCallbacks() + f.files[0] = nil + f.files = f.files[1:] + default: + return + } + } } func (f *fileWorkerGroup) flushAll(egCtx context.Context) error { @@ -393,6 +434,10 @@ func (f *fileWorkerGroup) flushAll(egCtx context.Context) error { return errors.Trace(err) } } + for _, file := range f.files { + file.runPostFlushCallbacks() + } + clear(f.files) f.files = f.files[:0] return nil } diff --git a/pkg/redo/writer/memory/file_worker_test.go b/pkg/redo/writer/memory/file_worker_test.go index eded46bd9a..3c15cb3a39 100644 --- a/pkg/redo/writer/memory/file_worker_test.go +++ b/pkg/redo/writer/memory/file_worker_test.go @@ -129,3 +129,67 @@ func TestFileWorkerDisablesCountBasedFlushWithZero(t *testing.T) { cancel() require.ErrorIs(t, <-runErrCh, context.Canceled) } + +// TestFileWorkerReleasesSizeRotatedCallbacksInOrder creates three events that +// each force the previous file to rotate, completes the second upload before +// the first, and verifies callbacks run in input order as the durable prefix +// advances. It also checks that invoked callback slots no longer retain their +// function values while the current unflushed file remains unacknowledged. +func TestFileWorkerReleasesSizeRotatedCallbacksInOrder(t *testing.T) { + const eventSize = 600 * 1024 + + inputCh := make(chan *polymorphicRedoEvent) + fileWorkers := newTestFileWorkerGroup(t, inputCh, 0) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + runErrCh := make(chan error, 1) + go func() { + runErrCh <- fileWorkers.bgWriteLogs(ctx, inputCh) + }() + + callbackOrder := make(chan int, 3) + for i := 1; i <= 3; i++ { + index := i + inputCh <- &polymorphicRedoEvent{ + commitTs: uint64(i), + data: make([]byte, eventSize), + callback: func() { + callbackOrder <- index + }, + } + } + + firstFile := <-fileWorkers.flushCh + secondFile := <-fileWorkers.flushCh + firstCallbackSlots := firstFile.postFlushCallbacks + secondCallbackSlots := secondFile.postFlushCallbacks + require.Len(t, firstCallbackSlots, 1) + require.Len(t, secondCallbackSlots, 1) + + secondFile.markFlushed() + require.Never(t, func() bool { + return len(callbackOrder) != 0 + }, 100*time.Millisecond, 10*time.Millisecond) + + firstFile.markFlushed() + select { + case index := <-callbackOrder: + require.Equal(t, 1, index) + case <-time.After(time.Second): + require.FailNow(t, "first rotated file callback was not released") + } + select { + case index := <-callbackOrder: + require.Equal(t, 2, index) + case <-time.After(time.Second): + require.FailNow(t, "second rotated file callback was not released") + } + + require.Nil(t, firstCallbackSlots[0]) + require.Nil(t, secondCallbackSlots[0]) + require.Empty(t, callbackOrder) + + cancel() + require.ErrorIs(t, <-runErrCh, context.Canceled) +}