fix: memory leak of geos

source/client/src/clientMain.c

@@ -23,6 +23,7 @@
 #include "query.h"
 #include "scheduler.h"
 #include "tdatablock.h"
+#include "tgeosctx.h"
 #include "tglobal.h"
 #include "tmsg.h"
 #include "tref.h"
@@ -94,6 +95,7 @@ void taos_cleanup(void) {
   tmqMgmtClose();
 
   DestroyRegexCache();
+  destroyThreadLocalGeosCtx();
 
   tscInfo("all local resources released");
   taosCleanupCfg();

source/dnode/mgmt/node_mgmt/src/dmMgmt.c

@@ -19,6 +19,7 @@
 #include "index.h"
 #include "qworker.h"
 #include "tcompression.h"
+#include "tgeosctx.h"
 #include "tglobal.h"
 #include "tgrant.h"
 #include "tstream.h"
@@ -121,6 +122,7 @@ void dmCleanupDnode(SDnode *pDnode) {
   streamMetaCleanup();
   indexCleanup();
   taosConvDestroy();
+  destroyThreadLocalGeosCtx();
 
   // compress destroy
   tsCompressExit();

source/util/src/tgeosctx.c

@@ -15,38 +15,82 @@
 
 #include "tgeosctx.h"
 #include "tdef.h"
+#include "tlockfree.h"
+#include "tlog.h"
 
-static threadlocal SGeosContext tlGeosCtx = {0};
+typedef struct {
+  SGeosContext *pool;
+  int32_t       capacity;
+  int32_t       size;
+  SRWLatch      lock;
+} SGeosContextPool;
 
-SGeosContext* getThreadLocalGeosCtx() { return &tlGeosCtx; }
+static SGeosContextPool sGeosPool = {0};
 
-void destroyThreadLocalGeosCtx() {
+static threadlocal SGeosContext *tlGeosCtx = NULL;
-  if (tlGeosCtx.WKTReader) {
+
-    GEOSWKTReader_destroy_r(tlGeosCtx.handle, tlGeosCtx.WKTReader);
+SGeosContext *getThreadLocalGeosCtx() {
-    tlGeosCtx.WKTReader = NULL;
+  if (tlGeosCtx) return tlGeosCtx;
+
+  taosWLockLatch(&sGeosPool.lock);
+  if (sGeosPool.size >= sGeosPool.capacity) {
+    sGeosPool.capacity += 64;
+    void *tmp = taosMemoryRealloc(sGeosPool.pool, sGeosPool.capacity * sizeof(SGeosContext));
+    if (!tmp) {
+      taosWUnLockLatch(&sGeosPool.lock);
+      terrno = TSDB_CODE_OUT_OF_MEMORY;
+      return NULL;
+    }
+    sGeosPool.pool = tmp;
+    TAOS_MEMSET(sGeosPool.pool + sGeosPool.size, 0, (sGeosPool.capacity - sGeosPool.size) * sizeof(SGeosContext));
+  }
+  tlGeosCtx = sGeosPool.pool + sGeosPool.size;
+  ++sGeosPool.size;
+  taosWUnLockLatch(&sGeosPool.lock);
+
+  return tlGeosCtx;
+}
+
+static void destroyGeosCtx(SGeosContext *pCtx) {
+  if (pCtx) {
+    if (pCtx->WKTReader) {
+      GEOSWKTReader_destroy_r(pCtx->handle, pCtx->WKTReader);
+      pCtx->WKTReader = NULL;
     }
 
-  if (tlGeosCtx.WKTWriter) {
+    if (pCtx->WKTWriter) {
-    GEOSWKTWriter_destroy_r(tlGeosCtx.handle, tlGeosCtx.WKTWriter);
+      GEOSWKTWriter_destroy_r(pCtx->handle, pCtx->WKTWriter);
-    tlGeosCtx.WKTWriter = NULL;
+      pCtx->WKTWriter = NULL;
     }
 
-  if (tlGeosCtx.WKBReader) {
+    if (pCtx->WKBReader) {
-    GEOSWKBReader_destroy_r(tlGeosCtx.handle, tlGeosCtx.WKBReader);
+      GEOSWKBReader_destroy_r(pCtx->handle, pCtx->WKBReader);
-    tlGeosCtx.WKBReader = NULL;
+      pCtx->WKBReader = NULL;
     }
 
-  if (tlGeosCtx.WKBWriter) {
+    if (pCtx->WKBWriter) {
-    GEOSWKBWriter_destroy_r(tlGeosCtx.handle, tlGeosCtx.WKBWriter);
+      GEOSWKBWriter_destroy_r(pCtx->handle, pCtx->WKBWriter);
-    tlGeosCtx.WKBWriter = NULL;
+      pCtx->WKBWriter = NULL;
     }
 
-  if (tlGeosCtx.WKTRegex) {
+    if (pCtx->WKTRegex) {
-    destroyRegexes(tlGeosCtx.WKTRegex, tlGeosCtx.WKTMatchData);
+      destroyRegexes(pCtx->WKTRegex, pCtx->WKTMatchData);
+      pCtx->WKTRegex = NULL;
+      pCtx->WKTMatchData = NULL;
     }
 
-  if (tlGeosCtx.handle) {
+    if (pCtx->handle) {
-    GEOS_finish_r(tlGeosCtx.handle);
+      GEOS_finish_r(pCtx->handle);
-    tlGeosCtx.handle = NULL;
+      pCtx->handle = NULL;
+    }
   }
 }
+
+void destroyThreadLocalGeosCtx() {
+  uInfo("geos ctx is cleaned up");
+  if (!sGeosPool.pool) return;
+  for (int32_t i = 0; i < sGeosPool.size; ++i) {
+    destroyGeosCtx(sGeosPool.pool + i);
+  }
+  taosMemoryFreeClear(sGeosPool.pool);
+}

source/util/src/tsched.c

@@ -178,7 +178,7 @@ void *taosProcessSchedQueue(void *scheduler) {
       (*(msg.tfp))(msg.ahandle, msg.thandle);
   }
 
-  destroyThreadLocalGeosCtx();
+  // destroyThreadLocalGeosCtx();
 
   return NULL;
 }

source/util/src/tworker.c

@@ -105,7 +105,7 @@ static void *tQWorkerThreadFp(SQueueWorker *worker) {
     taosUpdateItemSize(qinfo.queue, 1);
   }
 
-  destroyThreadLocalGeosCtx();
+  // destroyThreadLocalGeosCtx();
   DestoryThreadLocalRegComp();
 
   return NULL;
@@ -665,7 +665,7 @@ static void *tQueryAutoQWorkerThreadFp(SQueryAutoQWorker *worker) {
     }
   }
 
-  destroyThreadLocalGeosCtx();
+  // destroyThreadLocalGeosCtx();
   DestoryThreadLocalRegComp();
 
   return NULL;