changefeedccl: avoid heap allocation in timers

stevendanna · stevendanna · commit c6fb9cf8f6fd · 2025-09-30T13:56:15.000+01:00
Unfortunately, the closure from Start() seemed to always be heap
allocated. While minor, it was enough to spot when reading
rangefeed-related memory profiles.

```
BenchmarkTimerStart/closure-based                8992018               128.1 ns/op            24 B/op          1 allocs/op
BenchmarkTimerStart/handle-based                12124416                98.81 ns/op            0 B/op          0 allocs/op
```

Epic: none
Release note: None
diff --git a/pkg/BUILD.bazel b/pkg/BUILD.bazel
@@ -39,6 +39,7 @@ ALL_TESTS = [
     "//pkg/ccl/changefeedccl/resolvedspan:resolvedspan_test",
     "//pkg/ccl/changefeedccl/schemafeed:schemafeed_test",
     "//pkg/ccl/changefeedccl/tableset:tableset_test",
+    "//pkg/ccl/changefeedccl/timers:timers_test",
     "//pkg/ccl/changefeedccl:changefeedccl_test",
     "//pkg/ccl/cliccl:cliccl_test",
     "//pkg/ccl/cloudccl/amazon:amazon_test",
@@ -958,6 +959,7 @@ GO_TARGETS = [
     "//pkg/ccl/changefeedccl/tableset:tableset",
     "//pkg/ccl/changefeedccl/tableset:tableset_test",
     "//pkg/ccl/changefeedccl/timers:timers",
+    "//pkg/ccl/changefeedccl/timers:timers_test",
     "//pkg/ccl/changefeedccl:changefeedccl",
     "//pkg/ccl/changefeedccl:changefeedccl_test",
     "//pkg/ccl/cliccl:cliccl",
diff --git a/pkg/ccl/changefeedccl/batching_sink.go b/pkg/ccl/changefeedccl/batching_sink.go
@@ -359,7 +359,7 @@ func (s *batchingSink) runBatchingWorker(ctx context.Context) {
 		batch, _ := req.(*sinkBatch)
 		defer s.metrics.recordSinkIOInflightChange(int64(-batch.numMessages))
 		s.metrics.recordSinkIOInflightChange(int64(batch.numMessages))
-		defer s.metrics.timers().DownstreamClientSend.Start()()
+		defer s.metrics.timers().DownstreamClientSend.Start().End()
 
 		return s.client.Flush(ctx, batch.payload)
 	}
diff --git a/pkg/ccl/changefeedccl/changefeed_processors.go b/pkg/ccl/changefeedccl/changefeed_processors.go
@@ -1843,7 +1843,7 @@ func (cf *changeFrontier) checkpointJobProgress(
 ) error {
 	ctx, sp := tracing.ChildSpan(ctx, "changefeed.frontier.checkpoint_job_progress")
 	defer sp.Finish()
-	defer cf.sliMetrics.Timers.CheckpointJobProgress.Start()()
+	defer cf.sliMetrics.Timers.CheckpointJobProgress.Start().End()
 
 	if cf.knobs.RaiseRetryableError != nil {
 		if err := cf.knobs.RaiseRetryableError(); err != nil {
@@ -1924,7 +1924,7 @@ func (cf *changeFrontier) maybePersistFrontier(ctx context.Context) error {
 	}); err != nil {
 		return err
 	}
-	persistDuration := timer()
+	persistDuration := timer.End()
 	cf.frontierPersistenceLimiter.doneSave(persistDuration)
 	return nil
 }
@@ -1950,11 +1950,11 @@ func (cf *changeFrontier) manageProtectedTimestamps(
 	recordPTSMetricsErrorTime := cf.sliMetrics.Timers.PTSManageError.Start()
 	defer func() {
 		if err != nil {
-			recordPTSMetricsErrorTime()
+			recordPTSMetricsErrorTime.End()
 			return
 		}
 		if updated {
-			recordPTSMetricsTime()
+			recordPTSMetricsTime.End()
 		}
 	}()
 
diff --git a/pkg/ccl/changefeedccl/changefeed_stmt.go b/pkg/ccl/changefeedccl/changefeed_stmt.go
@@ -446,7 +446,7 @@ func changefeedPlanHook(
 				}
 				return err
 			}
-			recordPTSMetricsTime()
+			recordPTSMetricsTime.End()
 		}
 
 		// Start the job.
diff --git a/pkg/ccl/changefeedccl/event_processing.go b/pkg/ccl/changefeedccl/event_processing.go
@@ -442,7 +442,7 @@ func (c *kvEventToRowConsumer) encodeAndEmit(
 		}
 	}
 
-	stop := c.metrics.Timers.Encode.Start()
+	timer := c.metrics.Timers.Encode.Start()
 	if c.encodingOpts.Format == changefeedbase.OptFormatParquet {
 		return c.encodeForParquet(
 			ctx, updatedRow, prevRow, topic, schemaTS, updatedRow.MvccTimestamp,
@@ -466,7 +466,7 @@ func (c *kvEventToRowConsumer) encodeAndEmit(
 	// Since we're done processing/converting this event, and will not use much more
 	// than len(key)+len(bytes) worth of resources, adjust allocation to match.
 	alloc.AdjustBytesToTarget(ctx, int64(len(keyCopy)+len(valueCopy)))
-	stop()
+	timer.End()
 
 	headers, err := c.makeRowHeaders(ctx, updatedRow)
 	if err != nil {
diff --git a/pkg/ccl/changefeedccl/kvfeed/kv_feed.go b/pkg/ccl/changefeedccl/kvfeed/kv_feed.go
@@ -732,7 +732,7 @@ func copyFromSourceToDestUntilTableEvent(
 		// from rangefeed) and checks if a table event was encountered at or before
 		// said timestamp. If so, it replaces the copy boundary with the table event.
 		checkForTableEvent = func(ts hlc.Timestamp) error {
-			defer st.KVFeedWaitForTableEvent.Start()()
+			defer st.KVFeedWaitForTableEvent.Start().End()
 			// There's no need to check for table events again if we already found one
 			// since that should already be the earliest one.
 			if _, ok := boundary.(*errTableEventReached); ok {
@@ -829,7 +829,7 @@ func copyFromSourceToDestUntilTableEvent(
 
 		// writeToDest writes an event to the dest.
 		writeToDest = func(e kvevent.Event) error {
-			defer st.KVFeedBuffer.Start()()
+			defer st.KVFeedBuffer.Start().End()
 
 			switch e.Type() {
 			case kvevent.TypeKV, kvevent.TypeFlush:
diff --git a/pkg/ccl/changefeedccl/kvfeed/physical_kv_feed.go b/pkg/ccl/changefeedccl/kvfeed/physical_kv_feed.go
@@ -144,13 +144,13 @@ func (p *rangefeed) handleRangefeedEvent(ctx context.Context, e *kvpb.RangeFeedE
 				return err
 			}
 		}
-		stop := p.st.RangefeedBufferValue.Start()
+		timer := p.st.RangefeedBufferValue.Start()
 		if err := p.memBuf.Add(
 			ctx, kvevent.MakeKVEvent(e),
 		); err != nil {
 			return err
 		}
-		stop()
+		timer.End()
 	case *kvpb.RangeFeedCheckpoint:
 		ev := e.ShallowCopy()
 		ev.Checkpoint.ResolvedTS = quantizeTS(ev.Checkpoint.ResolvedTS, p.cfg.WithFrontierQuantize)
@@ -163,13 +163,13 @@ func (p *rangefeed) handleRangefeedEvent(ctx context.Context, e *kvpb.RangeFeedE
 		if p.knobs.ShouldSkipCheckpoint != nil && p.knobs.ShouldSkipCheckpoint(t) {
 			return nil
 		}
-		stop := p.st.RangefeedBufferCheckpoint.Start()
+		timer := p.st.RangefeedBufferCheckpoint.Start()
 		if err := p.memBuf.Add(
 			ctx, kvevent.MakeResolvedEvent(ev, jobspb.ResolvedSpan_NONE),
 		); err != nil {
 			return err
 		}
-		stop()
+		timer.End()
 	case *kvpb.RangeFeedSSTable:
 		// For now, we just error on SST ingestion, since we currently don't
 		// expect SST ingestion into spans with active changefeeds.
diff --git a/pkg/ccl/changefeedccl/sink_cloudstorage.go b/pkg/ccl/changefeedccl/sink_cloudstorage.go
@@ -892,7 +892,7 @@ func (f *cloudStorageSinkFile) flushToStorage(
 	ctx context.Context, es cloud.ExternalStorage, dest string, m metricsRecorder,
 ) error {
 	defer f.releaseAlloc(ctx)
-	defer m.timers().DownstreamClientSend.Start()()
+	defer m.timers().DownstreamClientSend.Start().End()
 
 	if f.rawSize == 0 {
 		// This method shouldn't be called with an empty file, but be defensive
diff --git a/pkg/ccl/changefeedccl/sink_kafka.go b/pkg/ccl/changefeedccl/sink_kafka.go
@@ -382,10 +382,10 @@ func (s *kafkaSink) EmitRow(
 	// inside sarama, set the DownstreamClientSend timer to the same value as
 	// BatchHistNanos.
 	recordOneMessageCb := s.metrics.recordOneMessage()
-	downstreamClientSendCb := s.metrics.timers().DownstreamClientSend.Start()
+	downstreamClientSend := s.metrics.timers().DownstreamClientSend.Start()
 	updateMetrics := func(mvcc hlc.Timestamp, bytes int, compressedBytes int) {
 		recordOneMessageCb(mvcc, bytes, compressedBytes)
-		downstreamClientSendCb()
+		downstreamClientSend.End()
 	}
 
 	recordHeaders := make([]sarama.RecordHeader, 0, len(headers))
diff --git a/pkg/ccl/changefeedccl/sink_pulsar.go b/pkg/ccl/changefeedccl/sink_pulsar.go
@@ -262,7 +262,7 @@ func (p *pulsarSink) msgCallback(
 	oneMsgCb := p.metrics.recordOneMessage()
 
 	return func(id pulsar.MessageID, message *pulsar.ProducerMessage, err error) {
-		sendCb()
+		sendCb.End()
 		if err == nil {
 			oneMsgCb(mvcc, len(message.Payload), len(message.Payload))
 		} else {
diff --git a/pkg/ccl/changefeedccl/timers/BUILD.bazel b/pkg/ccl/changefeedccl/timers/BUILD.bazel
@@ -1,4 +1,4 @@
-load("@io_bazel_rules_go//go:def.bzl", "go_library")
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "timers",
@@ -12,3 +12,14 @@ go_library(
         "@com_github_prometheus_client_golang//prometheus",
     ],
 )
+
+go_test(
+    name = "timers_test",
+    srcs = ["timers_bench_test.go"],
+    embed = [":timers"],
+    deps = [
+        "//pkg/util/metric",
+        "//pkg/util/metric/aggmetric",
+        "@com_github_prometheus_client_golang//prometheus",
+    ],
+)
diff --git a/pkg/ccl/changefeedccl/timers/timers.go b/pkg/ccl/changefeedccl/timers/timers.go
@@ -103,20 +103,29 @@ type timer struct {
 	hist *aggmetric.Histogram
 }
 
-func (t *timer) Start() (end func() time.Duration) {
+func (t *timer) Start() TimerHandle {
 	if t == nil {
-		return func() time.Duration { return 0 }
+		return TimerHandle{}
 	}
+	return TimerHandle{start: crtime.NowMono(), hist: t.hist}
+}
+
+type TimerHandle struct {
+	start crtime.Mono
+	hist  *aggmetric.Histogram
+}
 
-	start := crtime.NowMono()
-	return func() time.Duration {
-		elapsed := start.Elapsed()
-		t.hist.RecordValue(elapsed.Nanoseconds())
-		return elapsed
+// End records the elapsed time and returns the duration.
+func (th TimerHandle) End() time.Duration {
+	if th.hist == nil {
+		return 0
 	}
+	elapsed := th.start.Elapsed()
+	th.hist.RecordValue(elapsed.Nanoseconds())
+	return elapsed
 }
 
 func (t *timer) Time(cb func()) {
-	defer t.Start()()
+	defer t.Start().End()
 	cb()
 }
diff --git a/pkg/ccl/changefeedccl/timers/timers_bench_test.go b/pkg/ccl/changefeedccl/timers/timers_bench_test.go
@@ -0,0 +1,40 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package timers
+
+import (
+	"testing"
+	"time"
+
+	"github.com/cockroachdb/cockroach/pkg/util/metric"
+	"github.com/cockroachdb/cockroach/pkg/util/metric/aggmetric"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+func BenchmarkTimerHandle(b *testing.B) {
+	histOpts := metric.HistogramOptions{
+		Metadata: metric.Metadata{
+			Name:        "test.timer",
+			Help:        "Test timer",
+			Unit:        metric.Unit_NANOSECONDS,
+			Measurement: "Latency",
+		},
+		Duration: time.Hour,
+		Buckets:  prometheus.ExponentialBucketsRange(float64(1*time.Microsecond), float64(1*time.Hour), 60),
+		Mode:     metric.HistogramModePrometheus,
+	}
+	builder := aggmetric.MakeBuilder("scope")
+	hist := builder.Histogram(histOpts).AddChild("test")
+	timer := &timer{hist: hist}
+
+	b.ReportAllocs()
+	b.ResetTimer()
+	for range b.N {
+		handle := timer.Start()
+		_ = handle.End()
+	}
+
+}

Original file line number	Diff line number	Diff line change
`@@ -359,7 +359,7 @@ func (s *batchingSink) runBatchingWorker(ctx context.Context) {`
`359`	`359`	`batch, _ := req.(*sinkBatch)`
`360`	`360`	`defer s.metrics.recordSinkIOInflightChange(int64(-batch.numMessages))`
`361`	`361`	`s.metrics.recordSinkIOInflightChange(int64(batch.numMessages))`
`362`		`- defer s.metrics.timers().DownstreamClientSend.Start()()`
	`362`	`+ defer s.metrics.timers().DownstreamClientSend.Start().End()`
`363`	`363`
`364`	`364`	`return s.client.Flush(ctx, batch.payload)`
`365`	`365`	`}`
Original file line number	Diff line number	Diff line change
`@@ -446,7 +446,7 @@ func changefeedPlanHook(`
`446`	`446`	`}`
`447`	`447`	`return err`
`448`	`448`	`}`
`449`		`- recordPTSMetricsTime()`
	`449`	`+ recordPTSMetricsTime.End()`
`450`	`450`	`}`
`451`	`451`
`452`	`452`	`// Start the job.`