From 458ff7d10677a957f29bb7e6af91d46f219f15e5 Mon Sep 17 00:00:00 2001
From: HaHaJeff <jeffzhouhhh@gmail.com>
Date: Fri, 9 Feb 2024 07:43:51 +0000
Subject: [PATCH] fixed memory leak in errsim.

---
 src/logservice/palf/log_io_worker.cpp         | 9 +++++++++
 src/logservice/palf/log_io_worker_wrapper.cpp | 8 +++++++-
 2 files changed, 16 insertions(+), 1 deletion(-)

diff --git a/src/logservice/palf/log_io_worker.cpp b/src/logservice/palf/log_io_worker.cpp
index d1489f9a2..69cc7728c 100644
--- a/src/logservice/palf/log_io_worker.cpp
+++ b/src/logservice/palf/log_io_worker.cpp
@@ -361,6 +361,7 @@ int LogIOWorker::BatchLogIOFlushLogTaskMgr::init(int64_t batch_width,
     PALF_LOG(ERROR, "batch_io_task_array_ init failed", K(ret));
   } else {
     for (int i = 0; i < batch_width  && OB_SUCC(ret); i++) {
+      bool last_io_task_push_success = false;
       char *ptr = reinterpret_cast<char*>(mtl_malloc(sizeof(BatchLogIOFlushLogTask), "LogIOTask"));
       BatchLogIOFlushLogTask *io_task = NULL;
       if (NULL == ptr) {
@@ -369,12 +370,19 @@ int LogIOWorker::BatchLogIOFlushLogTaskMgr::init(int64_t batch_width,
       } else if (FALSE_IT(io_task = new(ptr)(BatchLogIOFlushLogTask))) {
       } else if (OB_FAIL(io_task->init(batch_depth, allocator))) {
         PALF_LOG(ERROR, "BatchLogIOFlushLogTask init failed", K(ret));
+        // NB: push batch will not failed becaue batch_io_task_array_ has reserved.
       } else if (OB_FAIL(batch_io_task_array_.push_back(io_task))) {
         PALF_LOG(ERROR, "batch_io_task_array_ push_back failed", K(ret), KP(io_task));
       } else {
+        last_io_task_push_success = true;
         PALF_LOG(INFO, "BatchLogIOFlushLogTask init success", K(ret), K(i),
                  KP(io_task));
       }
+      if (!last_io_task_push_success && NULL != io_task) {
+        io_task->destroy();
+        mtl_free(io_task);
+        io_task = NULL;
+      }
     }
     batch_width_ = usable_count_ = batch_width;
     wait_cost_stat_ = wait_cost_stat;
@@ -397,6 +405,7 @@ void LogIOWorker::BatchLogIOFlushLogTaskMgr::destroy()
     }
   }
   wait_cost_stat_ = NULL;
+  batch_io_task_array_.destroy();
 }
 
 int LogIOWorker::BatchLogIOFlushLogTaskMgr::insert(LogIOFlushLogTask *io_task)
diff --git a/src/logservice/palf/log_io_worker_wrapper.cpp b/src/logservice/palf/log_io_worker_wrapper.cpp
index dbbc35682..901061698 100644
--- a/src/logservice/palf/log_io_worker_wrapper.cpp
+++ b/src/logservice/palf/log_io_worker_wrapper.cpp
@@ -72,7 +72,7 @@ int LogIOWorkerWrapper::init(const LogIOWorkerConfig &config,
     LOG_INFO("success to init LogIOWorkerWrapper", K(config), K(tenant_id), KPC(this));
   }
   if (OB_FAIL(ret) && OB_INIT_TWICE != ret) {
-    destory_and_free_log_io_workers_();
+    destroy();
   }
   return ret;
 }
@@ -144,6 +144,7 @@ int LogIOWorkerWrapper::create_and_init_log_io_workers_(const LogIOWorkerConfig
                                                         IPalfEnvImpl *palf_env_impl)
 {
   int ret = OB_SUCCESS;
+  log_writer_parallelism_ = 0;
   const int64_t log_writer_parallelism = config.io_worker_num_;
   log_io_workers_ = reinterpret_cast<LogIOWorker *>(share::mtl_malloc(
     (log_writer_parallelism) * sizeof(LogIOWorker), "LogIOWS"));
@@ -161,11 +162,16 @@ int LogIOWorkerWrapper::create_and_init_log_io_workers_(const LogIOWorkerConfig
       PALF_LOG(WARN, "init LogIOWorker failed", K(i), K(config), K(tenant_id),
                K(cb_thread_pool_tg_id), KP(allocator), KP(palf_env_impl));
     } else {
+      log_writer_parallelism_++;
       PALF_LOG(INFO, "init LogIOWorker success", K(i), K(config), K(tenant_id),
                K(cb_thread_pool_tg_id), KP(allocator), KP(palf_env_impl), KP(iow),
                KP(log_io_workers_));
     }
   }
+  if (OB_FAIL(ret)) {
+    destory_and_free_log_io_workers_();
+    log_writer_parallelism_ = -1;
+  }
   return ret;
 }