Merge pull request #689 from svenssonaxel/RONDB-896

mronstro · web-flow · commit 0a8df2f417ec · 2025-05-08T23:50:14.000+03:00
[RONDB-896] rdrs2: Fix memory leak in getNextReqRS_Buffer
diff --git a/storage/ndb/rest-server2/server/src/batch_feature_store_ctrl.cpp b/storage/ndb/rest-server2/server/src/batch_feature_store_ctrl.cpp
@@ -38,6 +38,7 @@
 #include <vector>
 #include <EventLogger.hpp>
 #include <ArenaMalloc.hpp>
+#include <util/require.h>
 
 extern EventLogger *g_eventLogger;
 
@@ -221,16 +222,20 @@ void BatchFeatureStoreCtrl::batch_featureStore(
     Uint32 request_buffer_size = globalConfigs.internal.reqBufferSize * 2;
     Uint32 request_buffer_limit = request_buffer_size / 2;
     Uint32 current_head = 0;
-    RS_Buffer current_request_buffer = rsBufferArrayManager.get_req_buffer();
+    reqBuffs[0] = rsBufferArrayManager.get_req_buffer();
     respBuffs[0] = rsBufferArrayManager.get_resp_buffer();
+    Uint32 current_request_buffer_idx = 0;
     for (Uint32 i = 0; i < noOps; i++) {
-      RS_Buffer reqBuff = getNextReqRS_Buffer(current_head,
-                                              request_buffer_limit,
-                                              current_request_buffer,
-                                              i);
-      reqBuffs[i]  = reqBuff;
+      if (i > 0)
+        reqBuffs[i] = getNextReqRS_Buffer(
+          current_head,
+          request_buffer_limit,
+          reqBuffs[current_request_buffer_idx],
+          current_request_buffer_idx,
+          i
+        );
       status = create_native_request(readParams[i],
-                                     (Uint32*)reqBuff.buffer,
+                                     (Uint32*)reqBuffs[i].buffer,
                                      current_head);
       if (static_cast<drogon::HttpStatusCode>(status.http_code) !=
             drogon::HttpStatusCode::k200OK) {
@@ -241,10 +246,12 @@ void BatchFeatureStoreCtrl::batch_featureStore(
         return;
       }
       UintPtr length_ptr =
-        reinterpret_cast<UintPtr>(reqBuff.buffer) +
+        reinterpret_cast<UintPtr>(reqBuffs[i].buffer) +
           static_cast<UintPtr>(PK_REQ_LENGTH_IDX) * ADDRESS_SIZE;
       Uint32 *length_ptr_casted = reinterpret_cast<Uint32*>(length_ptr);
       reqBuffs[i].size = *length_ptr_casted;
+      require(reqBuffs[i].buffer + reqBuffs[i].size <=
+              reqBuffs[current_request_buffer_idx].buffer + current_head);
     }
     metricsUpdater.set_key_requests(noOps);
     // pk_batch_read
diff --git a/storage/ndb/rest-server2/server/src/batch_pk_read_ctrl.cpp b/storage/ndb/rest-server2/server/src/batch_pk_read_ctrl.cpp
@@ -33,6 +33,7 @@
 #include <simdjson.h>
 #include <EventLogger.hpp>
 #include <ArenaMalloc.hpp>
+#include <util/require.h>
 
 extern EventLogger *g_eventLogger;
 
@@ -201,18 +202,22 @@ void BatchPKReadCtrl::batchPKRead(
     Uint32 request_buffer_size = globalConfigs.internal.reqBufferSize * 2;
     Uint32 request_buffer_limit = request_buffer_size / 2;
     Uint32 current_head = 0;
-    RS_Buffer current_request_buffer = rsBufferArrayManager.get_req_buffer();
+    reqBuffs[0] = rsBufferArrayManager.get_req_buffer();
     respBuffs[0] = rsBufferArrayManager.get_resp_buffer();
+    Uint32 current_request_buffer_idx = 0;
     for (Uint32 i = 0; i < noOps; i++) {
-      RS_Buffer reqBuff = getNextReqRS_Buffer(current_head,
-                                              request_buffer_limit,
-                                              current_request_buffer,
-                                              i);
-      reqBuffs[i]  = reqBuff;
+      if (i > 0)
+        reqBuffs[i] = getNextReqRS_Buffer(
+          current_head,
+          request_buffer_limit,
+          reqBuffs[current_request_buffer_idx],
+          current_request_buffer_idx,
+          i
+        );
       DEB_BPK_CTRL("Buffer: %p, current_head: %u",
         reqBuff.buffer, current_head);
       status = create_native_request(reqStructs[i],
-                                     (Uint32*)reqBuff.buffer,
+                                     (Uint32*)reqBuffs[i].buffer,
                                      current_head);
       if (unlikely(static_cast<drogon::HttpStatusCode>(status.http_code) !=
           drogon::HttpStatusCode::k200OK)) {
@@ -222,10 +227,12 @@ void BatchPKReadCtrl::batchPKRead(
         release_array_buffers(reqBuffs.data(), respBuffs.data(), i);
         return;
       }
-      UintPtr length_ptr = reinterpret_cast<UintPtr>(reqBuff.buffer) +
+      UintPtr length_ptr = reinterpret_cast<UintPtr>(reqBuffs[i].buffer) +
         static_cast<UintPtr>(PK_REQ_LENGTH_IDX) * ADDRESS_SIZE;
       Uint32 *length_ptr_casted = reinterpret_cast<Uint32*>(length_ptr);
       reqBuffs[i].size = *length_ptr_casted;
+      require(reqBuffs[i].buffer + reqBuffs[i].size <=
+              reqBuffs[current_request_buffer_idx].buffer + current_head);
     }
 
     metricsUpdater.set_key_requests(noOps);
diff --git a/storage/ndb/rest-server2/server/src/buffer_manager.hpp b/storage/ndb/rest-server2/server/src/buffer_manager.hpp
@@ -133,12 +133,14 @@ class RS_BufferArrayManager {
 
   void return_req_buffer(RS_Buffer buffer) {
     NdbMutex_Lock(reqBufferMutex);
+    buffer.next_allocated_buffer = 0;
     reqBufferArray.push_back(buffer);
     NdbMutex_Unlock(reqBufferMutex);
   }
 
   void return_resp_buffer(RS_Buffer buffer) {
     NdbMutex_Lock(respBufferMutex);
+    buffer.next_allocated_buffer = 0;
     respBufferArray.push_back(buffer);
     NdbMutex_Unlock(respBufferMutex);
   }
diff --git a/storage/ndb/rest-server2/server/src/db_operations/pk/pkr_operation.cpp b/storage/ndb/rest-server2/server/src/db_operations/pk/pkr_operation.cpp
@@ -509,13 +509,17 @@ RS_Status BatchKeyOperations::create_response(RS_Buffer *respBuffs) {
   Uint32 response_buffer_limit = response_buffer_size / 2;
   Uint32 current_head = 0;
   Uint32 response_length = 0;
-  RS_Buffer current_response_buffer = respBuffs[0];;
+  Uint32 current_response_buffer_idx = 0;
   for (size_t i = 0; i < m_numOperations; i++) {
     current_head += response_length;
-    respBuffs[i] = getNextRespRS_Buffer(current_head,
-                                        response_buffer_limit,
-                                        current_response_buffer,
-                                        i);
+    if (i > 0)
+      respBuffs[i] = getNextRespRS_Buffer(
+        current_head,
+        response_buffer_limit,
+        respBuffs[current_response_buffer_idx],
+        current_response_buffer_idx,
+        i
+      );
     KeyOperation *key_op = &m_key_ops[i];
     PKRResponse *resp =
       new (&key_op->m_resp) PKRResponse(&respBuffs[i]);
diff --git a/storage/ndb/rest-server2/server/src/encoding.cpp b/storage/ndb/rest-server2/server/src/encoding.cpp
@@ -35,8 +35,8 @@
 extern EventLogger *g_eventLogger;
 
 #if (defined(VM_TRACE) || defined(ERROR_INSERT))
-#define DEBUG_ENC 1
-#define DEBUG_ENC_RESP 1
+//#define DEBUG_ENC 1
+//#define DEBUG_ENC_RESP 1
 #endif
 
 #ifdef DEBUG_ENC
@@ -225,48 +225,48 @@ RS_Status create_native_request(PKReadParams &pkReadParams,
 RS_Buffer getNextReqRS_Buffer(Uint32 &current_head,
                               Uint32 request_buffer_limit,
                               RS_Buffer &current_request_buffer,
+                              Uint32 &current_request_buffer_idx,
                               Uint32 index) {
-  RS_Buffer reqBuff;
-  if (index == 0) { // First buffer already allocated
-    return current_request_buffer;
-  } else if (current_head >= request_buffer_limit) {
+  require(index > 0);
+  if (current_head >= request_buffer_limit) {
     current_request_buffer.next_allocated_buffer = index;
-    current_request_buffer = rsBufferArrayManager.get_req_buffer();
+    current_request_buffer_idx = index;
     current_head = 0;
+    RS_Buffer reqBuff = rsBufferArrayManager.get_req_buffer();
     DEB_ENC("Allocating a new request buffer index = %u, ptr: %p",
-      index, current_request_buffer.buffer);
-    return current_request_buffer;
-  } else {
-    reqBuff.next_allocated_buffer = 0xFFFFFFFF; // Garbage
-    reqBuff.buffer = current_request_buffer.buffer + current_head;
-    reqBuff.size = (globalConfigs.internal.reqBufferSize * 2) - current_head;
-    DEB_ENC("Reuse request buffer index = %u at pos: %u, ptr: %p",
-      index, current_head, reqBuff.buffer);
+      index, reqBuff.buffer);
+    return reqBuff;
   }
+  RS_Buffer reqBuff;
+  reqBuff.next_allocated_buffer = 0xFFFFFFFF; // Garbage
+  reqBuff.buffer = current_request_buffer.buffer + current_head;
+  reqBuff.size = (globalConfigs.internal.reqBufferSize * 2) - current_head;
+  DEB_ENC("Reuse request buffer index = %u at pos: %u, ptr: %p",
+    index, current_head, reqBuff.buffer);
   return reqBuff;
 }
 
 RS_Buffer getNextRespRS_Buffer(Uint32 &current_head,
                                Uint32 response_buffer_limit,
                                RS_Buffer &current_response_buffer,
+                               Uint32 &current_response_buffer_idx,
                                Uint32 index) {
-  RS_Buffer respBuff;
-  if (index == 0) { // First buffer already allocated
-    return current_response_buffer;
-  } else if (current_head >= response_buffer_limit) {
+  require(index > 0);
+  if (current_head >= response_buffer_limit) {
     current_response_buffer.next_allocated_buffer = index;
-    current_response_buffer = rsBufferArrayManager.get_resp_buffer();
+    current_response_buffer_idx = index;
     current_head = 0;
+    RS_Buffer respBuff = rsBufferArrayManager.get_resp_buffer();
     DEB_ENC("Allocating a new response buffer index = %u, ptr: %p",
-      index, current_response_buffer.buffer);
-    return current_response_buffer;
-  } else {
-    respBuff.next_allocated_buffer = 0xFFFFFFFF; // Garbage
-    respBuff.buffer = current_response_buffer.buffer + current_head;
-    respBuff.size = (globalConfigs.internal.respBufferSize * 2) - current_head;
-    DEB_ENC("Reuse response buffer index = %u at pos: %u, ptr: %p",
-      index, current_head, respBuff.buffer);
+      index, respBuff.buffer);
+    return respBuff;
   }
+  RS_Buffer respBuff;
+  respBuff.next_allocated_buffer = 0xFFFFFFFF; // Garbage
+  respBuff.buffer = current_response_buffer.buffer + current_head;
+  respBuff.size = (globalConfigs.internal.respBufferSize * 2) - current_head;
+  DEB_ENC("Reuse response buffer index = %u at pos: %u, ptr: %p",
+    index, current_head, respBuff.buffer);
   return respBuff;
 }
 
diff --git a/storage/ndb/rest-server2/server/src/encoding.hpp b/storage/ndb/rest-server2/server/src/encoding.hpp
@@ -38,10 +38,12 @@ RS_Status process_pkread_response(ArenaMalloc*,
 RS_Buffer getNextReqRS_Buffer(Uint32 &head,
                               Uint32 request_buffer_limit,
                               RS_Buffer &current_request_buffer,
+                              Uint32 &current_request_buffer_idx,
                               Uint32 index);
 RS_Buffer getNextRespRS_Buffer(Uint32 &head,
                                Uint32 response_buffer_limit,
                                RS_Buffer &current_response_buffer,
+                               Uint32 &current_response_buffer_idx,
                                Uint32 index);
 void release_array_buffers(RS_Buffer *req_buffers,
                            RS_Buffer *resp_buffers,
diff --git a/storage/ndb/rest-server2/server/src/feature_store_ctrl.cpp b/storage/ndb/rest-server2/server/src/feature_store_ctrl.cpp
@@ -44,6 +44,7 @@
 #include <vector>
 #include <EventLogger.hpp>
 #include <ArenaMalloc.hpp>
+#include <util/require.h>
 
 extern EventLogger *g_eventLogger;
 
@@ -712,17 +713,21 @@ void FeatureStoreCtrl::featureStore(
     Uint32 request_buffer_size = globalConfigs.internal.reqBufferSize * 2;
     Uint32 request_buffer_limit = request_buffer_size / 2;
     Uint32 current_head = 0;
-    RS_Buffer current_request_buffer = rsBufferArrayManager.get_req_buffer();
+    reqBuffs[0] = rsBufferArrayManager.get_req_buffer();
     respBuffs[0] = rsBufferArrayManager.get_resp_buffer();
+    Uint32 current_request_buffer_idx = 0;
     for (Uint32 i = 0; i < noOps; i++) {
-      RS_Buffer reqBuff = getNextReqRS_Buffer(current_head,
-                                              request_buffer_limit,
-                                              current_request_buffer,
-                                              i);
-      reqBuffs[i]  = reqBuff;
+      if (i > 0)
+        reqBuffs[i]  = getNextReqRS_Buffer(
+          current_head,
+          request_buffer_limit,
+          reqBuffs[current_request_buffer_idx],
+          current_request_buffer_idx,
+          i
+        );
       RS_Status status =
         create_native_request(readParams[i],
-                              (Uint32*)reqBuff.buffer,
+                              (Uint32*)reqBuffs[i].buffer,
                               current_head);
       if (unlikely(static_cast<drogon::HttpStatusCode>(status.http_code) !=
                      drogon::HttpStatusCode::k200OK)) {
@@ -733,10 +738,12 @@ void FeatureStoreCtrl::featureStore(
         return;
       }
       UintPtr length_ptr =
-        reinterpret_cast<UintPtr>(reqBuff.buffer) +
+        reinterpret_cast<UintPtr>(reqBuffs[i].buffer) +
         static_cast<UintPtr>(PK_REQ_LENGTH_IDX) * ADDRESS_SIZE;
       Uint32 *length_ptr_casted = reinterpret_cast<Uint32*>(length_ptr);
       reqBuffs[i].size = *length_ptr_casted;
+      require(reqBuffs[i].buffer + reqBuffs[i].size <=
+              reqBuffs[current_request_buffer_idx].buffer + current_head);
     }
     // pk_batch_read
     {

Original file line number	Diff line number	Diff line change
`@@ -133,12 +133,14 @@ class RS_BufferArrayManager {`
`133`	`133`
`134`	`134`	`void return_req_buffer(RS_Buffer buffer) {`
`135`	`135`	`NdbMutex_Lock(reqBufferMutex);`
	`136`	`+ buffer.next_allocated_buffer = 0;`
`136`	`137`	`reqBufferArray.push_back(buffer);`
`137`	`138`	`NdbMutex_Unlock(reqBufferMutex);`
`138`	`139`	`}`
`139`	`140`
`140`	`141`	`void return_resp_buffer(RS_Buffer buffer) {`
`141`	`142`	`NdbMutex_Lock(respBufferMutex);`
	`143`	`+ buffer.next_allocated_buffer = 0;`
`142`	`144`	`respBufferArray.push_back(buffer);`
`143`	`145`	`NdbMutex_Unlock(respBufferMutex);`
`144`	`146`	`}`