refactor(cppbridge): SetFinalizer to AddCleanup

pp/go/cppbridge/data_storage.go

@@ -4,27 +4,50 @@ import (
 	"runtime"
 	"sync/atomic"
 	"unsafe"
+
+	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/prometheus/prometheus/pp/go/util"
+)
+
+var (
+	dsCreate = util.NewUnconflictRegisterer(prometheus.DefaultRegisterer).NewCounter(
+		prometheus.CounterOpts{
+			Name:        "prompp_cppbridge_cpp_objects_create_count",
+			Help:        "Current number of created C++ objects.",
+			ConstLabels: prometheus.Labels{"object": "data_storage"},
+		},
+	)
+
+	dsFinalize = util.NewUnconflictRegisterer(prometheus.DefaultRegisterer).NewCounter(
+		prometheus.CounterOpts{
+			Name:        "prompp_cppbridge_cpp_objects_finalize_count",
+			Help:        "Current number of finalized C++ objects.",
+			ConstLabels: prometheus.Labels{"object": "data_storage"},
+		},
+	)
 )
 
 // DataStorage is Go wrapper around series_data::Data_storage.
 type DataStorage struct {
-	dataStorage       uintptr
-	gcDestroyDetector *uint64
-	timeInterval      atomic.Pointer[TimeInterval]
+	dataStorage  uintptr
+	timeInterval atomic.Pointer[TimeInterval]
 }
 
 // NewDataStorage - constructor.
 func NewDataStorage(collectMetrics, useArenas bool) *DataStorage {
 	ds := &DataStorage{
-		dataStorage:       seriesDataDataStorageCtor(collectMetrics, useArenas),
-		gcDestroyDetector: &gcDestroyDetector,
-		timeInterval:      atomic.Pointer[TimeInterval]{},
+		dataStorage:  seriesDataDataStorageCtor(collectMetrics, useArenas),
+		timeInterval: atomic.Pointer[TimeInterval]{},
 	}
 	ds.timeInterval.Store(newInvalidTimeIntervalPtr())
 
-	runtime.SetFinalizer(ds, func(ds *DataStorage) {
-		seriesDataDataStorageDtor(ds.dataStorage)
-	})
+	runtime.AddCleanup(ds, func(pointer uintptr) {
+		seriesDataDataStorageDtor(pointer)
+		dsFinalize.Inc()
+	}, ds.dataStorage)
+
+	dsCreate.Inc()
 
 	return ds
 }
@@ -101,10 +124,8 @@ func (ds *DataStorage) CreateUnusedSeriesDataUnloader() *UnusedSeriesDataUnloade
 		unloader: seriesDataUnusedSeriesDataUnloaderCtor(ds.dataStorage),
 		ds:       ds,
 	}
-
-	runtime.SetFinalizer(unloader, func(u *UnusedSeriesDataUnloader) {
-		seriesDataUnusedSeriesDataUnloaderDtor(u.unloader)
-	})
+	runtime.KeepAlive(ds)
+	runtime.AddCleanup(unloader, seriesDataUnusedSeriesDataUnloaderDtor, unloader.unloader)
 
 	return unloader
 }
@@ -118,6 +139,7 @@ type DataStorageQuery struct {
 func (ds *DataStorage) Query(query DataStorageQuery, downsamplingMs int64, selectHints unsafe.Pointer) DataStorageQueryResult {
 	sd := NewDataStorageSerializedData(ds)
 	querier, status := seriesDataDataStorageQueryV2(ds.dataStorage, query, sd, downsamplingMs, selectHints)
+	runtime.KeepAlive(ds)
 	runtime.KeepAlive(selectHints)
 	return DataStorageQueryResult{
 		Querier:        querier,
@@ -129,7 +151,9 @@ func (ds *DataStorage) Query(query DataStorageQuery, downsamplingMs int64, selec
 // InstantQuery .
 // Deprecated: InstantQuery .
 func (ds *DataStorage) InstantQuery(targetTimestamp int64, labelSetIDs []uint32, samples uintptr) DataStorageQueryResult {
-	return seriesDataDataStorageInstantQuery(ds.dataStorage, labelSetIDs, targetTimestamp, samples)
+	result := seriesDataDataStorageInstantQuery(ds.dataStorage, labelSetIDs, targetTimestamp, samples)
+	runtime.KeepAlive(ds)
+	return result
 }
 
 // QueryFirstTimestamps fills timestamps with the first sample timestamp (Prometheus ms) for each series in seriesIDs.

pp/go/cppbridge/entrypoint.go

@@ -16,12 +16,14 @@ package cppbridge
 // #cgo static LDFLAGS: -static -static-libgcc -static-libstdc++ -l:libstdc++.a -l:libm.a -l:libgcc_eh.a -l:libunwind.a -l:liblzma.a -l:libstdc++exp.a
 // #include "entrypoint.h"
 import "C" //nolint:gocritic // because otherwise it won't work
+
 import (
 	"runtime"
 	"time"
 	"unsafe" //nolint:gocritic // because otherwise it won't work
 
 	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/prometheus/prometheus/pp/go/cppbridge/fastcgo"
 	"github.com/prometheus/prometheus/pp/go/model"
 	"github.com/prometheus/prometheus/pp/go/util"
@@ -243,6 +245,22 @@ var (
 		},
 	)
 
+	// head_data_storage dtor
+	headDataStorageDtorSum = util.NewUnconflictRegisterer(prometheus.DefaultRegisterer).NewCounter(
+		prometheus.CounterOpts{
+			Name:        "prompp_cppbridge_unsafecall_nanoseconds_sum",
+			Help:        "The time duration cpp call.",
+			ConstLabels: prometheus.Labels{"object": "head_data_storage", "method": "dtor"},
+		},
+	)
+	headDataStorageDtorCount = util.NewUnconflictRegisterer(prometheus.DefaultRegisterer).NewCounter(
+		prometheus.CounterOpts{
+			Name:        "prompp_cppbridge_unsafecall_nanoseconds_count",
+			Help:        "The time duration cpp call.",
+			ConstLabels: prometheus.Labels{"object": "head_data_storage", "method": "dtor"},
+		},
+	)
+
 	// head_data_storage query
 	headDataStorageQuerySum = util.NewUnconflictRegisterer(prometheus.DefaultRegisterer).NewCounter(
 		prometheus.CounterOpts{
@@ -1070,7 +1088,7 @@ func walDecoderDtor(decoder uintptr) {
 	)
 }
 
-func walSegmentSamplesStorageListCtor(count uint64, storages *SegmentSamplesStorageList) {
+func walSegmentSamplesStorageListCtor(count uint64, storages *segmentSamplesStorageListCPP) {
 	args := struct {
 		count    uint64
 		storages uintptr
@@ -1115,7 +1133,7 @@ func walSegmentSamplesStorageClear(samplesStorage *CppSegmentSamplesStorage) {
 	)
 }
 
-func walSegmentSamplesStorageListDtor(s *SegmentSamplesStorageList) {
+func walSegmentSamplesStorageListDtor(s *segmentSamplesStorageListCPP) {
 	args := struct {
 		storages uintptr
 	}{uintptr(unsafe.Pointer(s))}
@@ -1128,7 +1146,7 @@ func walSegmentSamplesStorageListDtor(s *SegmentSamplesStorageList) {
 }
 
 func walSegmentSamplesStorageListSplitMessages(
-	s *SegmentSamplesStorageList,
+	s *segmentSamplesStorageListCPP,
 	messageSamplesThreshold uint32,
 ) []RWMessage {
 	args := struct {
@@ -2498,12 +2516,14 @@ func seriesDataDataStorageDtor(dataStorage uintptr) {
 	args := struct {
 		dataStorage uintptr
 	}{dataStorage}
-
+	start := time.Now()
 	testGC()
 	fastcgo.UnsafeCall1(
 		C.prompp_series_data_data_storage_dtor,
 		uintptr(unsafe.Pointer(&args)),
 	)
+	headDataStorageDtorSum.Add(float64(time.Since(start).Nanoseconds()))
+	headDataStorageDtorCount.Inc()
 }
 
 func seriesDataEncoderEncode(dataStorage uintptr, seriesID uint32, timestamp int64, value float64) {

pp/go/cppbridge/head.go

@@ -110,16 +110,14 @@ func initializeChunkRecoder(
 		dataStorage:    dataStorage,
 		serializedData: serializedData,
 	}
-
-	runtime.SetFinalizer(chunkRecoder, func(chunkRecoder *ChunkRecoder) {
-		seriesDataChunkRecoderDtor(chunkRecoder.recoder)
-	})
+	runtime.AddCleanup(chunkRecoder, seriesDataChunkRecoderDtor, chunkRecoder.recoder)
 
 	return chunkRecoder
 }
 
 func (recoder *ChunkRecoder) RecodeNextChunk() RecodedChunk {
 	seriesDataChunkRecoderRecodeNextChunk(recoder.recoder, &recoder.recodedChunk)
+	runtime.KeepAlive(recoder)
 	return recoder.recodedChunk
 }
 
@@ -234,6 +232,7 @@ type DataStorageSerializedDataSamplesIterator struct {
 func NewDataStorageSerializedDataSamplesIterator(serializedData *DataStorageSerializedData, chunkRef uint32) DataStorageSerializedDataSamplesIterator {
 	it := DataStorageSerializedDataSamplesIterator{}
 	seriesDataSerializedDataSamplesIteratorCtor(&it, serializedData.serializedData, chunkRef)
+	runtime.KeepAlive(serializedData)
 	return it
 }
 
@@ -247,6 +246,7 @@ func (it *DataStorageSerializedDataSamplesIterator) Seek(timestamp int64) {
 
 func (it *DataStorageSerializedDataSamplesIterator) Reset(serializedData *DataStorageSerializedData, chunkRef uint32) {
 	seriesDataSerializedDataSamplesIteratorReset(it, serializedData.serializedData, chunkRef)
+	runtime.KeepAlive(serializedData)
 }
 
 func (it *DataStorageSerializedDataSamplesIterator) HasData() bool {
@@ -286,6 +286,7 @@ type DataStorageSerializedDataAggregationIterator struct {
 func NewDataStorageSerializedDataAggregationIterator(serializedData *DataStorageSerializedData, chunkRef uint32) DataStorageSerializedDataAggregationIterator {
 	it := DataStorageSerializedDataAggregationIterator{}
 	seriesDataSerializedDataAggregationIteratorCtor(&it, serializedData.serializedData, chunkRef)
+	runtime.KeepAlive(serializedData)
 	return it
 }
 
@@ -295,6 +296,7 @@ func (it *DataStorageSerializedDataAggregationIterator) Next() {
 
 func (it *DataStorageSerializedDataAggregationIterator) Reset(serializedData *DataStorageSerializedData, chunkRef uint32) {
 	seriesDataSerializedDataAggregationIteratorReset(it, serializedData.serializedData, chunkRef)
+	runtime.KeepAlive(serializedData)
 }
 
 type DataStorageSerializedDataMultiSeriesIterator struct {
@@ -305,6 +307,7 @@ type DataStorageSerializedDataMultiSeriesIterator struct {
 func NewDataStorageSerializedDataMultiSeriesIterator(serializedData *DataStorageSerializedData, seriesIDs []uint32) DataStorageSerializedDataMultiSeriesIterator {
 	it := DataStorageSerializedDataMultiSeriesIterator{}
 	seriesDataSerializedDataMultiSeriesIteratorCtor(&it, serializedData.serializedData, seriesIDs)
+	runtime.KeepAlive(serializedData)
 	return it
 }
 
@@ -314,6 +317,7 @@ func (it *DataStorageSerializedDataMultiSeriesIterator) Next() {
 
 func (it *DataStorageSerializedDataMultiSeriesIterator) Reset(serializedData *DataStorageSerializedData, seriesIDs []uint32) {
 	seriesDataSerializedDataMultiSeriesIteratorReset(it, serializedData.serializedData, seriesIDs)
+	runtime.KeepAlive(serializedData)
 }
 
 func (it *DataStorageSerializedDataMultiSeriesIterator) Close() {
@@ -336,11 +340,10 @@ func (ds *DataStorage) CreateLoader(queriers []uintptr) *UnloadedDataLoader {
 		loader: seriesDataUnloadedDataLoaderCtor(ds.dataStorage, queriers),
 		ds:     ds,
 	}
+	runtime.KeepAlive(ds)
 	runtime.KeepAlive(queriers)
 
-	runtime.SetFinalizer(result, func(loader *UnloadedDataLoader) {
-		seriesDataUnloadedDataLoaderDtor(loader.loader)
-	})
+	runtime.AddCleanup(result, seriesDataUnloadedDataLoaderDtor, result.loader)
 
 	return result
 }
@@ -365,10 +368,9 @@ func (ds *DataStorage) CreateRevertableLoader(lss *LabelSetStorage, lsIdBatchSiz
 		},
 		lss: lss,
 	}
-
-	runtime.SetFinalizer(result, func(loader *UnloadedDataRevertableLoader) {
-		seriesDataUnloadedDataLoaderDtor(loader.loader)
-	})
+	runtime.KeepAlive(ds)
+	runtime.KeepAlive(lss)
+	runtime.AddCleanup(result, seriesDataUnloadedDataLoaderDtor, result.loader)
 
 	return result
 }

pp/go/cppbridge/head_wal.go

@@ -26,9 +26,7 @@ func NewHeadEncodedSegment(b []byte, samples uint32) *HeadEncodedSegment {
 		samples: samples,
 	}
 
-	runtime.SetFinalizer(s, func(s *HeadEncodedSegment) {
-		freeBytes(s.buf)
-	})
+	runtime.AddCleanup(s, freeBytes, s.buf)
 
 	return s
 }
@@ -80,10 +78,7 @@ func NewHeadWalEncoder(shardID uint16, logShards uint8, lss *LabelSetStorage) *H
 		lss:     lss,
 		encoder: headWalEncoderCtor(shardID, logShards, lss.Pointer()),
 	}
-
-	runtime.SetFinalizer(e, func(e *HeadWalEncoder) {
-		headWalEncoderDtor(e.encoder)
-	})
+	runtime.AddCleanup(e, headWalEncoderDtor, e.encoder)
 
 	return e
 }
@@ -130,10 +125,7 @@ func NewHeadWalDecoder(lss *LabelSetStorage, encoderVersion uint8) *HeadWalDecod
 		lss:     lss,
 		decoder: headWalDecoderCtor(lss.Pointer(), encoderVersion),
 	}
-
-	runtime.SetFinalizer(d, func(d *HeadWalDecoder) {
-		headWalDecoderDtor(d.decoder)
-	})
+	runtime.AddCleanup(d, headWalDecoderDtor, d.decoder)
 
 	return d
 }
@@ -171,10 +163,7 @@ func (d *HeadWalDecoder) CreateEncoder() (*HeadWalEncoder, error) {
 		lss:     d.lss,
 		encoder: encoder,
 	}
-
-	runtime.SetFinalizer(e, func(e *HeadWalEncoder) {
-		headWalEncoderDtor(e.encoder)
-	})
+	runtime.AddCleanup(e, headWalEncoderDtor, e.encoder)
 
 	return e, nil
 }

pp/go/cppbridge/index_writer.go

@@ -54,29 +54,31 @@ func NewIndexWriter(lss *LabelSetStorage) *IndexWriter {
 		output: newIndexWriterOutput(buffer, hasMorePostings),
 		lss:    lss,
 	}
-	runtime.SetFinalizer(writer, func(writer *IndexWriter) {
-		indexWriterDtor(writer.writer)
-	})
+	runtime.AddCleanup(writer, indexWriterDtor, writer.writer)
 	return writer
 }
 
 func (writer *IndexWriter) WriteHeader() []byte {
 	indexWriterWriteHeader(writer.writer)
+	runtime.KeepAlive(writer)
 	return writer.output.bytes()
 }
 
 func (writer *IndexWriter) WriteSymbols() []byte {
 	indexWriterWriteSymbols(writer.writer)
+	runtime.KeepAlive(writer)
 	return writer.output.bytes()
 }
 
 func (writer *IndexWriter) WriteSeries(ls_id uint32, chunks_meta []ChunkMetadata) []byte {
 	indexWriterWriteNextSeriesBatch(writer.writer, ls_id, chunks_meta)
+	runtime.KeepAlive(writer)
 	return writer.output.bytes()
 }
 
 func (writer *IndexWriter) WriteLabelIndices() []byte {
 	indexWriterWriteLabelIndices(writer.writer)
+	runtime.KeepAlive(writer)
 	return writer.output.bytes()
 }
 
@@ -85,20 +87,24 @@ func (writer *IndexWriter) WriteLabelIndices() []byte {
 // memory and is valid only until the next write_* call, so callers must consume it before looping.
 func (writer *IndexWriter) WriteNextPostingsBatch(maxBatchSize uint32) ([]byte, bool) {
 	indexWriterWritePostings(writer.writer, maxBatchSize)
+	runtime.KeepAlive(writer)
 	return writer.output.bytes(), writer.output.hasMore()
 }
 
 func (writer *IndexWriter) WriteLabelIndicesTable() []byte {
 	indexWriterWriteLabelIndicesTable(writer.writer)
+	runtime.KeepAlive(writer)
 	return writer.output.bytes()
 }
 
 func (writer *IndexWriter) WritePostingsTableOffsets() []byte {
 	indexWriterWritePostingsTableOffsets(writer.writer)
+	runtime.KeepAlive(writer)
 	return writer.output.bytes()
 }
 
 func (writer *IndexWriter) WriteTableOfContents() []byte {
 	indexWriterWriteTableOfContents(writer.writer)
+	runtime.KeepAlive(writer)
 	return writer.output.bytes()
 }