RONDB-969: Change RDRS to prevent excessive malloc ops. Simple peak-tracking i.e. keep the max buffers allocated for peak load

Author: smkniazi

storage/ndb/rest-server/rest-api-server/internal/dal/heap/heap.go

@@ -1,7 +1,7 @@
 /*
 
  * This file is part of the RonDB REST API Server
- * Copyright (c) 2023 Hopsworks AB
+ * Copyright (c) 2025 Hopsworks AB
  *
  * This program is free software: you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
@@ -37,7 +37,6 @@ type Heap struct {
 	buffers             []*NativeBuffer
 	buffersStats        MemoryStats
 	mutex               *sync.Mutex
-	minAllocatedBuffers int64
 	allocatedBuffers    int64
 	maxAllocatedBuffers int64
 }
@@ -68,7 +67,6 @@ func New() (heap *Heap, releaseBuffers func(), err error) {
 	heap = &Heap{
 		buffers:             []*NativeBuffer{},
 		mutex:               &sync.Mutex{},
-		minAllocatedBuffers: preAllocatedBuffers,
 		allocatedBuffers:    preAllocatedBuffers,
 		maxAllocatedBuffers: maxAllocatedBuffers,
 		buffersStats: MemoryStats{
@@ -101,9 +99,10 @@ func (heap *Heap) releaseAllBuffers() {
 	heap.mutex.Lock()
 	defer heap.mutex.Unlock()
 
-	if heap.buffersStats.BuffersCount != int64(len(heap.buffers)) {
+	stats := heap.getNativeBuffersStatsInt()
+	if stats.BuffersCount != int64(len(heap.buffers)) {
 		log.Warnf("Shutting down heap. Number of free buffers do not match. Expecting: %d, Got: %d.",
-			heap.buffersStats.BuffersCount, int64(len(heap.buffers)))
+			stats.BuffersCount, int64(len(heap.buffers)))
 	}
 
 	for _, buffer := range heap.buffers {
@@ -136,21 +135,19 @@ func (heap *Heap) GetBuffer() (buff *NativeBuffer, returnBuff func(), err error)
 func (heap *Heap) returnBuffer(buffer *NativeBuffer) {
 	heap.mutex.Lock()
 	defer heap.mutex.Unlock()
-
-	if int64(len(heap.buffers)) < heap.minAllocatedBuffers {
-		heap.buffers = append(heap.buffers, buffer)
-	} else {
-		C.free(buffer.Buffer)
-		heap.allocatedBuffers--
-		heap.buffersStats.DeallocationsCount++
-	}
+	heap.buffers = append(heap.buffers, buffer)
 }
 
 func (heap *Heap) GetNativeBuffersStats() MemoryStats {
 	heap.mutex.Lock()
 	defer heap.mutex.Unlock()
+	return heap.getNativeBuffersStatsInt()
+}
 
-	// Only (De)AllocationsCount are updated continuously. Update the others now.
+/*
+*	with out locks
+ */
+func (heap *Heap) getNativeBuffersStatsInt() MemoryStats {
 	heap.buffersStats.BuffersCount = heap.allocatedBuffers
 	heap.buffersStats.FreeBuffers = int64(len(heap.buffers))
 	return heap.buffersStats

storage/ndb/rest-server/rest-api-server/internal/dal/heap/heap_test.go

@@ -84,7 +84,7 @@ func TestHeap(t *testing.T) {
 	// Note that we now have more FreeBuffers than initially
 	// 	TODO: Figure out whether we want this to happen
 	stats = heap.GetNativeBuffersStats()
-	if stats.FreeBuffers != int64(conf.Internal.PreAllocatedBuffers) {
+	if stats.FreeBuffers != allocations {
 		t.Fatalf("Number of free buffers does not match. Expecting: %d, Got: %d",
 			allocations, stats.FreeBuffers)
 	}