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

fix:rm put SSmlHandleInfo to taos

This commit is contained in:
wangmm0220 2022-12-08 09:36:23 +08:00
parent 5711572b71
commit ef6e9f481f
3 changed files with 35 additions and 105 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/client/inc/clientInt.h
View File

@ -149,7 +149,6 @@ typedef struct STscObj {
int32_t numOfReqs; // number of sqlObj bound to this connection
SAppInstInfo* pAppInfo;
SHashObj* pRequests;
void* smlHandle;
} STscObj;
typedef struct SResultColumn {
@ -323,7 +322,6 @@ void destroyTscObj(void* pObj);
STscObj* acquireTscObj(int64_t rid);
int32_t releaseTscObj(int64_t rid);
void destroyAppInst(SAppInstInfo* pAppInfo);
void smlDestroyInfo(void *data);
uint64_t generateRequestId();

20
source/client/src/clientEnv.c
View File

@ -76,13 +76,20 @@ static void deregisterRequest(SRequestObj *pRequest) {
"current:%d, app current:%d",
pRequest->self, pTscObj->id, pRequest->requestId, duration / 1000.0, num, currentInst);
tscPerf("insert duration %" PRId64 "us: syntax:%" PRId64 "us, ctg:%" PRId64 "us, semantic:%" PRId64
"us, exec:%" PRId64 "us, stmtType:%d",
duration, pRequest->metric.syntaxEnd - pRequest->metric.syntaxStart,
pRequest->metric.ctgEnd - pRequest->metric.ctgStart, pRequest->metric.semanticEnd -
pRequest->metric.ctgEnd, pRequest->metric.execEnd - pRequest->metric.semanticEnd,
pRequest->stmtType);
if (QUERY_NODE_VNODE_MODIF_STMT == pRequest->stmtType) {
tscPerf("insert duration %" PRId64 "us: syntax:%" PRId64 "us, ctg:%" PRId64 "us, semantic:%" PRId64
"us, exec:%" PRId64 "us",
duration, pRequest->metric.syntaxEnd - pRequest->metric.syntaxStart,
pRequest->metric.ctgEnd - pRequest->metric.ctgStart, pRequest->metric.semanticEnd -
pRequest->metric.ctgEnd, pRequest->metric.execEnd - pRequest->metric.semanticEnd);
atomic_add_fetch_64((int64_t *)&pActivity->insertElapsedTime, duration);
// tscPerf("insert duration %" PRId64 "us: syntax:%" PRId64 "us, ctg:%" PRId64 "us, semantic:%" PRId64
// "us, exec:%" PRId64 "us",
// duration, pRequest->metric.syntaxEnd - pRequest->metric.syntaxStart,
// pRequest->metric.ctgEnd - pRequest->metric.ctgStart, pRequest->metric.semanticEnd -
// pRequest->metric.ctgEnd, pRequest->metric.execEnd - pRequest->metric.semanticEnd);
// atomic_add_fetch_64((int64_t *)&pActivity->insertElapsedTime, duration);
} else if (QUERY_NODE_SELECT_STMT == pRequest->stmtType) {
// tscPerf("select duration %" PRId64 "us: syntax:%" PRId64 "us, ctg:%" PRId64 "us, semantic:%" PRId64
// "us, planner:%" PRId64 "us, exec:%" PRId64 "us, reqId:0x%" PRIx64,
@ -237,7 +244,6 @@ void destroyTscObj(void *pObj) {
}
taosThreadMutexDestroy(&pTscObj->mutex);
taosMemoryFree(pTscObj);
smlDestroyInfo(pTscObj->smlHandle);
tscTrace("end to destroy tscObj %" PRIx64 " p:%p", tscId, pTscObj);
}

118
source/client/src/clientSml.c
View File

@ -1852,9 +1852,8 @@ static int32_t smlPushCols(SArray *colsArray, SArray *cols) {
return TSDB_CODE_SUCCESS;
}
void smlDestroyInfo(void *data) {
if (!data) return;
SSmlHandle *info = (SSmlHandle *)data;
void smlDestroyInfo(SSmlHandle *info) {
if (!info) return;
qDestroyQuery(info->pQuery);
// destroy info->childTables
@ -1881,7 +1880,6 @@ void smlDestroyInfo(void *data) {
// destroy info->pVgHash
taosHashCleanup(info->pVgHash);
destroyRequest(info->pRequest);
tmp = info->superTableTagKeyStr;
while (tmp) {
@ -1910,78 +1908,15 @@ void smlDestroyInfo(void *data) {
taosArrayDestroy(info->preLineTagKV);
taosArrayDestroy(info->preLineColKV);
for(int i = 0; i < info->lineNum; i++){
taosArrayDestroy(info->lines[i].colArray);
}
taosMemoryFree(info->lines);
cJSON_Delete(info->root);
taosMemoryFreeClear(info);
}
int32_t smlInitInfo(void *data, SRequestObj *request, bool isRawLine, int32_t ttl,
SMLProtocolType protocol, int8_t precision, int32_t lineNum) {
if (!data) return TSDB_CODE_SML_INVALID_DATA;
SSmlHandle *info = (SSmlHandle *)data;
info->id = smlGenId();
info->pRequest = request;
info->isRawLine = isRawLine;
info->ttl = ttl;
info->precision = precision;
info->protocol = protocol;
info->msgBuf.buf = info->pRequest->msgBuf;
info->msgBuf.len = ERROR_MSG_BUF_DEFAULT_SIZE;
info->pQuery = smlInitHandle();
if(lineNum > info->lineNum && !info->dataFormat){
void *tmp = taosMemoryRealloc(info->lines, lineNum * sizeof(SSmlLineInfo));
if(tmp == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
info->lines = tmp;
}
info->lineNum = lineNum;
return TSDB_CODE_SUCCESS;
}
void smlClearInfo(void *data) {
if (!data) return;
SSmlHandle *info = (SSmlHandle *)data;
// clear info->childTables
NodeList *tmp = info->childTables;
while (tmp) {
if(tmp->data.used){
smlDestroyTableInfo((SSmlTableInfo *)(tmp->data.value));
tmp->data.used = false;
}
tmp = tmp->next;
}
// tmp = info->superTableTagKeyStr;
// while (tmp) {
// taosMemoryFree(tmp->data.value);
// tmp->data.used = false;
// tmp = tmp->next;
// }
//
// tmp = info->superTableColKeyStr;
// while (tmp) {
// taosMemoryFree(tmp->data.value);
// tmp->data.used = false;
// tmp = tmp->next;
// }
if(!info->dataFormat){
for(int i = 0; i < info->lineNum; i++){
taosArrayDestroy(info->lines[i].colArray);
}
memset(info->lines, 0, info->lineNum * sizeof(SSmlLineInfo));
taosMemoryFree(info->lines);
}
cJSON_Delete(info->root);
qDestroyQuery(info->pQuery);
info->pQuery = NULL;
taosMemoryFreeClear(info);
}
static SSmlHandle *smlBuildSmlInfo(TAOS *taos) {
@ -1996,8 +1931,10 @@ static SSmlHandle *smlBuildSmlInfo(TAOS *taos) {
uError("SML:0x%" PRIx64 " get catalog error %d", info->id, code);
goto cleanup;
}
info->dataFormat = true;
info->pVgHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
info->id = smlGenId();
info->pQuery = smlInitHandle();
info->dataFormat = true;
if (NULL == info->pVgHash) {
uError("create SSmlHandle failed");
@ -2856,40 +2793,28 @@ TAOS_RES *taos_schemaless_insert_inner(TAOS *taos, char *lines[], char *rawLine,
return NULL;
}
STscObj *pTscObj = acquireTscObj(*(int64_t *)taos);
if (pTscObj == NULL) {
terrno = TSDB_CODE_TSC_DISCONNECTED;
return NULL;
}
SRequestObj *request = (SRequestObj *)createRequest(*(int64_t *)taos, TSDB_SQL_INSERT, reqid);
if (request == NULL) {
uError("SML:taos_schemaless_insert error request is null");
return NULL;
}
SSmlHandle *info = NULL;
if(pTscObj->smlHandle == NULL){
info = smlBuildSmlInfo(taos);
if (info == NULL) {
request->code = TSDB_CODE_OUT_OF_MEMORY;
uError("SML:taos_schemaless_insert error SSmlHandle is null");
goto end;
}
pTscObj->smlHandle = info;
}else{
info = (SSmlHandle *)(pTscObj->smlHandle);
smlClearInfo(info);
}
SSmlMsgBuf msg = {ERROR_MSG_BUF_DEFAULT_SIZE, request->msgBuf};
int ret = smlInitInfo(info, request, rawLine != NULL,
ttl, protocol, precision, numLines);
if(ret != TSDB_CODE_SUCCESS){
request->code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
smlBuildInvalidDataMsg(&msg, "smlInitInfo error", NULL);
SSmlHandle *info = smlBuildSmlInfo(taos);
if (info == NULL) {
request->code = TSDB_CODE_OUT_OF_MEMORY;
uError("SML:taos_schemaless_insert error SSmlHandle is null");
goto end;
}
info->pRequest = request;
info->isRawLine = rawLine != NULL;
info->ttl = ttl;
info->precision = precision;
info->protocol = protocol;
info->msgBuf.buf = info->pRequest->msgBuf;
info->msgBuf.len = ERROR_MSG_BUF_DEFAULT_SIZE;
info->lineNum = numLines;
SSmlMsgBuf msg = {ERROR_MSG_BUF_DEFAULT_SIZE, request->msgBuf};
if (request->pDb == NULL) {
request->code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
smlBuildInvalidDataMsg(&msg, "Database not specified", NULL);
@ -2925,6 +2850,7 @@ TAOS_RES *taos_schemaless_insert_inner(TAOS *taos, char *lines[], char *rawLine,
end:
uDebug("resultend:%s", request->msgBuf);
smlDestroyInfo(info);
return (TAOS_RES *)request;
}