xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix: memory leak of geos

This commit is contained in:
kailixu 2024-08-19 18:36:39 +08:00
parent 800e7c4e7a
commit 0531a4f4bd
11 changed files with 71 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
include/util/tgeosctx.h
View File

@ -38,8 +38,10 @@ typedef struct SGeosContext {
char errMsg[512];
} SGeosContext;
SGeosContext *acquireThreadLocalGeosCtx();
SGeosContext *getThreadLocalGeosCtx();
void destroyThreadLocalGeosCtx();
const char *getGeosErrMsg(int32_t code);
void taosGeosDestroy();
#ifdef __cplusplus
}

2
source/client/src/clientMain.c
View File

@ -91,11 +91,11 @@ void taos_cleanup(void) {
tscDebug("rpc cleanup");
taosConvDestroy();
taosGeosDestroy();
tmqMgmtClose();
DestroyRegexCache();
destroyThreadLocalGeosCtx();
tscInfo("all local resources released");
taosCleanupCfg();

2
source/dnode/mgmt/node_mgmt/src/dmMgmt.c
View File

@ -122,7 +122,7 @@ void dmCleanupDnode(SDnode *pDnode) {
streamMetaCleanup();
indexCleanup();
taosConvDestroy();
destroyThreadLocalGeosCtx();
taosGeosDestroy();
// compress destroy
tsCompressExit();

2
source/libs/executor/src/sysscanoperator.c
View File

@ -979,7 +979,7 @@ static int32_t sysTableGetGeomText(char* iGeom, int32_t nGeom, char** output, in
if (TSDB_CODE_SUCCESS != (code = initCtxAsText()) ||
TSDB_CODE_SUCCESS != (code = doAsText(iGeom, nGeom, &outputWKT))) {
qError("geo text for systable failed:%s", getThreadLocalGeosCtx()->errMsg);
qError("geo text for systable failed:%s", getGeosErrMsg(code));
*output = NULL;
*nOutput = 0;
return code;

45
source/libs/geometry/src/geosWrapper.c
View File

@ -23,7 +23,8 @@ typedef char (*_geosPreparedRelationFunc_t)(GEOSContextHandle_t handle, const GE
void geosFreeBuffer(void *buffer) {
if (buffer) {
GEOSFree_r(getThreadLocalGeosCtx()->handle, buffer);
SGeosContext *pCtx = acquireThreadLocalGeosCtx();
if (pCtx) GEOSFree_r(pCtx->handle, buffer);
}
}
@ -36,6 +37,11 @@ int32_t initCtxMakePoint() {
int32_t code = TSDB_CODE_FAILED;
SGeosContext *geosCtx = getThreadLocalGeosCtx();
if (!geosCtx) {
code = TSDB_CODE_OUT_OF_MEMORY;
return code;
}
if (geosCtx->handle == NULL) {
geosCtx->handle = GEOS_init_r();
if (geosCtx->handle == NULL) {
@ -61,6 +67,11 @@ int32_t doMakePoint(double x, double y, unsigned char **outputGeom, size_t *size
int32_t code = TSDB_CODE_FAILED;
SGeosContext *geosCtx = getThreadLocalGeosCtx();
if (!geosCtx) {
code = TSDB_CODE_OUT_OF_MEMORY;
return code;
}
GEOSGeometry *geom = NULL;
unsigned char *wkb = NULL;
@ -166,6 +177,11 @@ int32_t initCtxGeomFromText() {
int32_t code = TSDB_CODE_FAILED;
SGeosContext *geosCtx = getThreadLocalGeosCtx();
if (!geosCtx) {
code = TSDB_CODE_OUT_OF_MEMORY;
return code;
}
if (geosCtx->handle == NULL) {
geosCtx->handle = GEOS_init_r();
if (geosCtx->handle == NULL) {
@ -202,6 +218,11 @@ int32_t doGeomFromText(const char *inputWKT, unsigned char **outputGeom, size_t
int32_t code = TSDB_CODE_FAILED;
SGeosContext *geosCtx = getThreadLocalGeosCtx();
if (!geosCtx) {
code = TSDB_CODE_OUT_OF_MEMORY;
return code;
}
GEOSGeometry *geom = NULL;
unsigned char *wkb = NULL;
@ -237,6 +258,11 @@ int32_t initCtxAsText() {
int32_t code = TSDB_CODE_FAILED;
SGeosContext *geosCtx = getThreadLocalGeosCtx();
if (!geosCtx) {
code = TSDB_CODE_OUT_OF_MEMORY;
return code;
}
if (geosCtx->handle == NULL) {
geosCtx->handle = GEOS_init_r();
if (geosCtx->handle == NULL) {
@ -273,6 +299,11 @@ int32_t doAsText(const unsigned char *inputGeom, size_t size, char **outputWKT)
int32_t code = TSDB_CODE_FAILED;
SGeosContext *geosCtx = getThreadLocalGeosCtx();
if (!geosCtx) {
code = TSDB_CODE_OUT_OF_MEMORY;
return code;
}
GEOSGeometry *geom = NULL;
char *wkt = NULL;
@ -304,6 +335,11 @@ int32_t initCtxRelationFunc() {
int32_t code = TSDB_CODE_FAILED;
SGeosContext *geosCtx = getThreadLocalGeosCtx();
if (!geosCtx) {
code = TSDB_CODE_OUT_OF_MEMORY;
return code;
}
if (geosCtx->handle == NULL) {
geosCtx->handle = GEOS_init_r();
if (geosCtx->handle == NULL) {
@ -329,6 +365,10 @@ int32_t doGeosRelation(const GEOSGeometry *geom1, const GEOSPreparedGeometry *pr
_geosPreparedRelationFunc_t swappedPreparedRelationFn) {
SGeosContext *geosCtx = getThreadLocalGeosCtx();
if (!geosCtx) {
return TSDB_CODE_OUT_OF_MEMORY;
}
if (!preparedGeom1) {
if (!swapped) {
ASSERT(relationFn);
@ -390,6 +430,9 @@ int32_t doContainsProperly(const GEOSGeometry *geom1, const GEOSPreparedGeometry
int32_t readGeometry(const unsigned char *input, GEOSGeometry **outputGeom,
const GEOSPreparedGeometry **outputPreparedGeom) {
SGeosContext *geosCtx = getThreadLocalGeosCtx();
if (!geosCtx) {
return TSDB_CODE_OUT_OF_MEMORY;
}
ASSERT(outputGeom); // it is not allowed if outputGeom is NULL
*outputGeom = NULL;

4
source/libs/parser/src/parInsertSql.c
View File

@ -655,7 +655,7 @@ static int32_t parseTagToken(const char** end, SToken* pToken, SSchema* pSchema,
code = parseGeometry(pToken, &output, &size);
if (code != TSDB_CODE_SUCCESS) {
code = buildSyntaxErrMsg(pMsgBuf, getThreadLocalGeosCtx()->errMsg, pToken->z);
code = buildSyntaxErrMsg(pMsgBuf, getGeosErrMsg(code), pToken->z);
} else if (size + VARSTR_HEADER_SIZE > pSchema->bytes) {
// Too long values will raise the invalid sql error message
code = generateSyntaxErrMsg(pMsgBuf, TSDB_CODE_PAR_VALUE_TOO_LONG, pSchema->name);
@ -1646,7 +1646,7 @@ static int32_t parseValueTokenImpl(SInsertParseContext* pCxt, const char** pSql,
code = parseGeometry(pToken, &output, &size);
if (code != TSDB_CODE_SUCCESS) {
code = buildSyntaxErrMsg(&pCxt->msg, getThreadLocalGeosCtx()->errMsg, pToken->z);
code = buildSyntaxErrMsg(&pCxt->msg, getGeosErrMsg(code), pToken->z);
}
// Too long values will raise the invalid sql error message
else if (size + VARSTR_HEADER_SIZE > pSchema->bytes) {

8
source/libs/scalar/src/sclvector.c
View File

@ -441,12 +441,12 @@ static FORCE_INLINE int32_t varToGeometry(char *buf, SScalarParam *pOut, int32_t
unsigned char *t = NULL;
char *output = NULL;
if (initCtxGeomFromText()) {
sclError("failed to init geometry ctx, %s", getThreadLocalGeosCtx()->errMsg);
if ((code = initCtxGeomFromText()) != 0) {
sclError("failed to init geometry ctx, %s", getGeosErrMsg(code));
SCL_ERR_JRET(TSDB_CODE_APP_ERROR);
}
if (doGeomFromText(buf, &t, &len)) {
sclInfo("failed to convert text to geometry, %s", getThreadLocalGeosCtx()->errMsg);
if ((code = doGeomFromText(buf, &t, &len))) {
sclInfo("failed to convert text to geometry, %s", getGeosErrMsg(code));
SCL_ERR_JRET(TSDB_CODE_SCALAR_CONVERT_ERROR);
}

15
source/util/src/tgeosctx.c
View File

@ -26,15 +26,18 @@ typedef struct {
} SGeosContextPool;
static SGeosContextPool sGeosPool = {0};
static threadlocal SGeosContext *tlGeosCtx = NULL;
SGeosContext *acquireThreadLocalGeosCtx() { return tlGeosCtx; }
SGeosContext *getThreadLocalGeosCtx() {
if (tlGeosCtx) return tlGeosCtx;
if (tlGeosCtx) {
return tlGeosCtx;
}
taosWLockLatch(&sGeosPool.lock);
if (sGeosPool.size >= sGeosPool.capacity) {
sGeosPool.capacity += 64;
sGeosPool.capacity += 128;
void *tmp = taosMemoryRealloc(sGeosPool.pool, sGeosPool.capacity * sizeof(SGeosContext));
if (!tmp) {
taosWUnLockLatch(&sGeosPool.lock);
@ -51,6 +54,8 @@ SGeosContext *getThreadLocalGeosCtx() {
return tlGeosCtx;
}
const char *getGeosErrMsg(int32_t code) { return tlGeosCtx ? tlGeosCtx->errMsg : (code != 0 ? tstrerror(code) : ""); }
static void destroyGeosCtx(SGeosContext *pCtx) {
if (pCtx) {
if (pCtx->WKTReader) {
@ -86,8 +91,8 @@ static void destroyGeosCtx(SGeosContext *pCtx) {
}
}
void destroyThreadLocalGeosCtx() {
uInfo("geos ctx is cleaned up");
void taosGeosDestroy() {
uInfo("geos is cleaned up");
if (!sGeosPool.pool) return;
for (int32_t i = 0; i < sGeosPool.size; ++i) {
destroyGeosCtx(sGeosPool.pool + i);

2
source/util/src/tsched.c
View File

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

2
source/util/src/tworker.c
View File

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

5
tools/shell/src/shellEngine.c
View File

@ -611,14 +611,14 @@ void shellPrintGeometry(const unsigned char *val, int32_t length, int32_t width)
code = initCtxAsText();
if (code != TSDB_CODE_SUCCESS) {
shellPrintString(getThreadLocalGeosCtx()->errMsg, width);
shellPrintString(getGeosErrMsg(code), width);
return;
}
char *outputWKT = NULL;
code = doAsText(val, length, &outputWKT);
if (code != TSDB_CODE_SUCCESS) {
shellPrintString(getThreadLocalGeosCtx()->errMsg, width); // should NOT happen
shellPrintString(getGeosErrMsg(code), width); // should NOT happen
return;
}
@ -1282,7 +1282,6 @@ void *shellThreadLoop(void *arg) {
taosResetTerminalMode();
} while (shellRunCommand(command, true) == 0);
destroyThreadLocalGeosCtx();
taosMemoryFreeClear(command);
shellWriteHistory();
shellExit();