Tom/fix memory leak

rust_api/libmr/mod.rs

@@ -9,7 +9,8 @@
 //! This module provides with the Map-Reduce operations supported.
 
 use crate::libmr_c_raw::bindings::{
-    MRRecordType, MR_CalculateSlot, MR_ClusterIsInClusterMode, MR_Init, MR_IsMySlot, RedisModuleCtx,
+    MRRecordType, MR_CalculateSlot, MR_ClusterIsInClusterMode, MR_Fini as MR_Fini_C, MR_Init,
+    MR_IsMySlot, RedisModuleCtx,
 };
 use redis_module::Context;
 
@@ -59,6 +60,10 @@ pub fn mr_init(ctx: &Context, num_threads: usize, password: Option<&str>) {
     }
 }
 
+pub fn mr_fini() {
+    unsafe { MR_Fini_C() };
+}
+
 pub fn calc_slot(s: &[u8]) -> usize {
     unsafe { MR_CalculateSlot(s.as_ptr() as *const c_char, s.len()) }
 }

src/cluster.c

@@ -532,7 +532,7 @@ static int checkTLS(char** client_key, char** client_cert, char** ca_cert, char*
             MR_FREE(*client_cert);
         }
         if (*ca_cert) {
-            MR_FREE(*client_cert);
+            MR_FREE(*ca_cert);
         }
     }
 
@@ -684,8 +684,8 @@ static void MR_ConnectToShard(Node* n){
         return;
     }
     if (c->err) {
-        /* Let *c leak for now... */
         RedisModule_Log(mr_staticCtx, "warning", "Error: %s\n", c->errstr);
+        redisAsyncFree(c);
         return;
     }
     c->data = n;
@@ -747,7 +747,10 @@ static void MR_NodeFree(Node* n){
     MR_NodeFreeInternals(n);
 }
 
-static void MR_ClusterFree(){
+static void MR_ClusterFree(void){
+    if(!clusterCtx.CurrCluster){
+        return;
+    }
     if(clusterCtx.CurrCluster->myId){
         MR_FREE(clusterCtx.CurrCluster->myId);
     }
@@ -779,6 +782,10 @@ static void MR_ClusterFree(){
     memset(clusterCtx.myId, '0', REDISMODULE_NODE_ID_LEN);
 }
 
+void MR_ClusterFini(void){
+    MR_ClusterFree();
+}
+
 static Node* MR_GetNode(const char* id){
     mr_dictEntry *entry = mr_dictFind(clusterCtx.CurrCluster->nodes, id);
     Node* n = NULL;

src/cluster.h

@@ -37,6 +37,8 @@ size_t MR_ClusterGetSize();
 
 int MR_ClusterInit(RedisModuleCtx* rctx, char *password);
 
+void MR_ClusterFini(void);
+
 size_t MR_ClusterGetSlotByKey(const char* key, size_t len);
 
 int MR_ClusterIsMySlot(size_t slot);

src/mr.c

@@ -1557,6 +1557,10 @@ int MR_Init(RedisModuleCtx* ctx, size_t numThreads, char *password) {
     return REDISMODULE_OK;
 }
 
+LIBMR_API void MR_Fini(void) {
+    MR_ClusterFini();
+}
+
 int MR_RegisterObject(MRObjectType* t) {
     mrCtx.objectTypesDict = array_append(mrCtx.objectTypesDict, t);
     t->id = array_len(mrCtx.objectTypesDict) - 1;

src/mr.h

@@ -181,6 +181,9 @@ LIBMR_API void MR_FreeExecution(Execution* e);
 /* Initialize mr library */
 LIBMR_API int MR_Init(struct RedisModuleCtx* ctx, size_t numThreads, char *password);
 
+/* Teardown mr library (frees cluster state) */
+LIBMR_API void MR_Fini(void);
+
 /* Register a new object type */
 LIBMR_API int MR_RegisterObject(MRObjectType* t);
 

tests/mr_test_module/src/lib.rs

@@ -23,7 +23,7 @@ use redis_module::{
 
 use mr::libmr::{
     accumulator::AccumulateStep, base_object::BaseObject, calc_slot,
-    execution_builder::create_builder, filter::FilterStep, mapper::MapStep, mr_init,
+    execution_builder::create_builder, filter::FilterStep, mapper::MapStep, mr_fini, mr_init,
     reader::Reader, record::Record, remote_task::run_on_all_shards, remote_task::run_on_key,
     remote_task::RemoteTask, RustMRError,
 };
@@ -867,13 +867,19 @@ fn init_func(ctx: &Context, args: &[RedisString]) -> Status {
     Status::Ok
 }
 
+fn deinit_func(_ctx: &Context) -> Status {
+    mr_fini();
+    Status::Ok
+}
+
 #[cfg(not(test))]
 redis_module! {
     name: "lmrtest",
     version: 99_99_99,
     allocator: (RedisAlloc, RedisAlloc),
     data_types: [],
     init: init_func,
+    deinit: deinit_func,
     commands: [
         ["lmrtest.dbsize", lmr_dbsize, "readonly", 0,0,0, AclCategory::None],
         ["lmrtest.get", lmr_get, "readonly", 0,0,0, AclCategory::None],