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

Merge branch '3.0' of https://github.com/taosdata/TDengine into enh/TD-30987-4

This commit is contained in:
Hongze Cheng 2024-07-17 21:19:03 +08:00
parent cb055eca80 a3c0acb78a
commit c1b5a80e20
30 changed files with 1721 additions and 1067 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
include/libs/parser/parser.h
View File

@ -163,8 +163,8 @@ int rawBlockBindData(SQuery *query, STableMeta* pTableMeta, void* data, SVCr
int numFields, bool needChangeLength, char* errstr, int32_t errstrLen);
int32_t rewriteToVnodeModifyOpStmt(SQuery* pQuery, SArray* pBufArray);
SArray* serializeVgroupsCreateTableBatch(SHashObj* pVgroupHashmap);
SArray* serializeVgroupsDropTableBatch(SHashObj* pVgroupHashmap);
int32_t serializeVgroupsCreateTableBatch(SHashObj* pVgroupHashmap, SArray** pOut);
int32_t serializeVgroupsDropTableBatch(SHashObj* pVgroupHashmap, SArray** pOut);
void destoryCatalogReq(SCatalogReq *pCatalogReq);
bool isPrimaryKeyImpl(SNode* pExpr);
int32_t insAppendStmtTableDataCxt(SHashObj* pAllVgHash, STableColsData* pTbData, STableDataCxt* pTbCtx, SStbInterlaceInfo* pBuildInfo);

8
include/util/tqueue.h
View File

@ -73,10 +73,10 @@ struct STaosQnode {
char item[];
};
STaosQueue *taosOpenQueue();
int32_t taosOpenQueue(STaosQueue **queue);
void taosCloseQueue(STaosQueue *queue);
void taosSetQueueFp(STaosQueue *queue, FItem itemFp, FItems itemsFp);
void *taosAllocateQitem(int32_t size, EQItype itype, int64_t dataSize);
int32_t taosAllocateQitem(int32_t size, EQItype itype, int64_t dataSize, void **item);
void taosFreeQitem(void *pItem);
int32_t taosWriteQitem(STaosQueue *queue, void *pItem);
int32_t taosReadQitem(STaosQueue *queue, void **ppItem);
@ -87,7 +87,7 @@ int64_t taosQueueMemorySize(STaosQueue *queue);
void taosSetQueueCapacity(STaosQueue *queue, int64_t size);
void taosSetQueueMemoryCapacity(STaosQueue *queue, int64_t mem);
STaosQall *taosAllocateQall();
int32_t taosAllocateQall(STaosQall **qall);
void taosFreeQall(STaosQall *qall);
int32_t taosReadAllQitems(STaosQueue *queue, STaosQall *qall);
int32_t taosGetQitem(STaosQall *qall, void **ppItem);
@ -97,7 +97,7 @@ int64_t taosQallMemSize(STaosQall *qll);
int64_t taosQallUnAccessedItemSize(STaosQall *qall);
int64_t taosQallUnAccessedMemSize(STaosQall *qall);
STaosQset *taosOpenQset();
int32_t taosOpenQset(STaosQset **qset);
void taosCloseQset(STaosQset *qset);
void taosQsetThreadResume(STaosQset *qset);
int32_t taosAddIntoQset(STaosQset *qset, STaosQueue *queue, void *ahandle);

82
source/client/src/clientMonitor.c
View File

@ -1,13 +1,13 @@
#include "clientMonitor.h"
#include "clientLog.h"
#include "os.h"
#include "tmisce.h"
#include "ttime.h"
#include "ttimer.h"
#include "tglobal.h"
#include "tqueue.h"
#include "cJSON.h"
#include "clientInt.h"
#include "clientLog.h"
#include "os.h"
#include "tglobal.h"
#include "tmisce.h"
#include "tqueue.h"
#include "ttime.h"
#include "ttimer.h"
SRWLatch monitorLock;
void* monitorTimer;
@ -116,8 +116,12 @@ static int32_t monitorReportAsyncCB(void* param, SDataBuf* pMsg, int32_t code) {
if (code != 0) {
tscError("failed to send slow log:%s, clusterId:%" PRIx64, p->data, p->clusterId);
}
MonitorSlowLogData tmp = {.clusterId = p->clusterId, .type = p->type, .fileName = p->fileName,
.pFile= p->pFile, .offset = p->offset, .data = NULL};
MonitorSlowLogData tmp = {.clusterId = p->clusterId,
.type = p->type,
.fileName = p->fileName,
.pFile = p->pFile,
.offset = p->offset,
.data = NULL};
if (monitorPutData2MonitorQueue(tmp) == 0) {
p->fileName = NULL;
}
@ -247,7 +251,8 @@ void monitorCreateClient(int64_t clusterId) {
}
taos_collector_registry_register_collector(pMonitor->registry, pMonitor->colector);
pMonitor->counters = (SHashObj*)taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
pMonitor->counters =
(SHashObj*)taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
if (pMonitor->counters == NULL) {
tscError("failed to create monitor counters");
goto fail;
@ -265,7 +270,8 @@ void monitorCreateClient(int64_t clusterId) {
pMonitor = NULL;
goto fail;
}
pMonitor->timer = taosTmrStart(reportSendProcess, pInst->monitorParas.tsMonitorInterval * 1000, (void*)pMonitor, monitorTimer);
pMonitor->timer =
taosTmrStart(reportSendProcess, pInst->monitorParas.tsMonitorInterval * 1000, (void*)pMonitor, monitorTimer);
if (pMonitor->timer == NULL) {
tscError("failed to start timer");
goto fail;
@ -283,7 +289,8 @@ fail:
taosWUnLockLatch(&monitorLock);
}
void monitorCreateClientCounter(int64_t clusterId, const char* name, const char* help, size_t label_key_count, const char** label_keys) {
void monitorCreateClientCounter(int64_t clusterId, const char* name, const char* help, size_t label_key_count,
const char** label_keys) {
taosWLockLatch(&monitorLock);
MonitorClient** ppMonitor = (MonitorClient**)taosHashGet(monitorCounterHash, &clusterId, LONG_BYTES);
if (ppMonitor == NULL || *ppMonitor == NULL) {
@ -291,8 +298,7 @@ void monitorCreateClientCounter(int64_t clusterId, const char* name, const char*
goto end;
}
taos_counter_t* newCounter = taos_counter_new(name, help, label_key_count, label_keys);
if (newCounter == NULL)
return;
if (newCounter == NULL) return;
MonitorClient* pMonitor = *ppMonitor;
taos_collector_add_metric(pMonitor->colector, newCounter);
if (taosHashPut(pMonitor->counters, name, strlen(name), &newCounter, POINTER_BYTES) != 0) {
@ -300,7 +306,8 @@ void monitorCreateClientCounter(int64_t clusterId, const char* name, const char*
taos_counter_destroy(newCounter);
goto end;
}
tscInfo("[monitor] monitorCreateClientCounter %"PRIx64"(%p):%s : %p.", pMonitor->clusterId, pMonitor, name, newCounter);
tscInfo("[monitor] monitorCreateClientCounter %" PRIx64 "(%p):%s : %p.", pMonitor->clusterId, pMonitor, name,
newCounter);
end:
taosWUnLockLatch(&monitorLock);
@ -453,7 +460,8 @@ static int64_t getFileSize(char* path){
return fileSize;
}
static int32_t sendSlowLog(int64_t clusterId, char* data, TdFilePtr pFile, int64_t offset, SLOW_LOG_QUEUE_TYPE type, char* fileName, void* pTransporter, SEpSet *epSet){
static int32_t sendSlowLog(int64_t clusterId, char* data, TdFilePtr pFile, int64_t offset, SLOW_LOG_QUEUE_TYPE type,
char* fileName, void* pTransporter, SEpSet* epSet) {
if (data == NULL) {
taosMemoryFree(fileName);
return -1;
@ -473,7 +481,8 @@ static int32_t sendSlowLog(int64_t clusterId, char* data, TdFilePtr pFile, int64
return sendReport(pTransporter, epSet, data, MONITOR_TYPE_SLOW_LOG, pParam);
}
static int32_t monitorReadSend(int64_t clusterId, TdFilePtr pFile, int64_t* offset, int64_t size, SLOW_LOG_QUEUE_TYPE type, char* fileName){
static int32_t monitorReadSend(int64_t clusterId, TdFilePtr pFile, int64_t* offset, int64_t size,
SLOW_LOG_QUEUE_TYPE type, char* fileName) {
SAppInstInfo* pInst = getAppInstByClusterId(clusterId);
if (pInst == NULL) {
tscError("failed to get app instance by clusterId:%" PRId64, clusterId);
@ -481,7 +490,8 @@ static int32_t monitorReadSend(int64_t clusterId, TdFilePtr pFile, int64_t* offs
}
SEpSet ep = getEpSet_s(&pInst->mgmtEp);
char* data = readFile(pFile, offset, size);
return sendSlowLog(clusterId, data, (type == SLOW_LOG_READ_BEGINNIG ? pFile : NULL), *offset, type, fileName, pInst->pTransporter, &ep);
return sendSlowLog(clusterId, data, (type == SLOW_LOG_READ_BEGINNIG ? pFile : NULL), *offset, type, fileName,
pInst->pTransporter, &ep);
}
static void monitorSendSlowLogAtBeginning(int64_t clusterId, char** fileName, TdFilePtr pFile, int64_t offset) {
@ -567,7 +577,8 @@ static void processFileRemoved(SlowLogClient* pClient){
taosUnLockFile(pClient->pFile);
taosCloseFile(&(pClient->pFile));
TdFilePtr pFile = taosOpenFile(pClient->path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND | TD_FILE_READ | TD_FILE_TRUNC);
TdFilePtr pFile =
taosOpenFile(pClient->path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND | TD_FILE_READ | TD_FILE_TRUNC);
if (pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
tscError("failed to open file:%s since %s", pClient->path, terrstr());
@ -640,9 +651,7 @@ static void monitorSendAllSlowLogFromTempDir(int64_t clusterId){
}
char* name = taosGetDirEntryName(de);
if (strcmp(name, ".") == 0 ||
strcmp(name, "..") == 0 ||
strstr(name, namePrefix) == NULL) {
if (strcmp(name, ".") == 0 || strcmp(name, "..") == 0 || strstr(name, namePrefix) == NULL) {
tscInfo("skip file:%s, for cluster id:%" PRIx64, name, clusterId);
continue;
}
@ -702,7 +711,8 @@ static void* monitorThreadFunc(void *param){
if (slowLogData != NULL) {
if (slowLogData->type == SLOW_LOG_READ_BEGINNIG) {
if (slowLogData->pFile != NULL) {
monitorSendSlowLogAtBeginning(slowLogData->clusterId, &(slowLogData->fileName), slowLogData->pFile, slowLogData->offset);
monitorSendSlowLogAtBeginning(slowLogData->clusterId, &(slowLogData->fileName), slowLogData->pFile,
slowLogData->offset);
} else {
monitorSendAllSlowLogFromTempDir(slowLogData->clusterId);
}
@ -757,8 +767,11 @@ static void tscMonitorStop() {
}
int32_t monitorInit() {
int32_t code;
tscInfo("[monitor] tscMonitor init");
monitorCounterHash = (SHashObj*)taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
monitorCounterHash =
(SHashObj*)taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
if (monitorCounterHash == NULL) {
tscError("failed to create monitorCounterHash");
terrno = TSDB_CODE_OUT_OF_MEMORY;
@ -766,7 +779,8 @@ int32_t monitorInit() {
}
taosHashSetFreeFp(monitorCounterHash, destroyMonitorClient);
monitorSlowLogHash = (SHashObj*)taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
monitorSlowLogHash =
(SHashObj*)taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
if (monitorSlowLogHash == NULL) {
tscError("failed to create monitorSlowLogHash");
terrno = TSDB_CODE_OUT_OF_MEMORY;
@ -798,8 +812,9 @@ int32_t monitorInit() {
return -1;
}
monitorQueue = taosOpenQueue();
if(monitorQueue == NULL){
code = taosOpenQueue(&monitorQueue);
if (code) {
terrno = code;
tscError("open queue error since %s", terrstr());
return -1;
}
@ -829,17 +844,22 @@ void monitorClose() {
}
int32_t monitorPutData2MonitorQueue(MonitorSlowLogData data) {
int32_t code;
MonitorSlowLogData* slowLogData;
if (atomic_load_32(&slowLogFlag) == -2) {
tscError("[monitor] slow log thread is exiting");
return -1;
}
MonitorSlowLogData* slowLogData = taosAllocateQitem(sizeof(MonitorSlowLogData), DEF_QITEM, 0);
if (slowLogData == NULL) {
code = taosAllocateQitem(sizeof(MonitorSlowLogData), DEF_QITEM, 0, (void**)&slowLogData);
if (code) {
tscError("[monitor] failed to allocate slow log data");
return -1;
return terrno = code;
}
*slowLogData = data;
tscDebug("[monitor] write slow log to queue, clusterId:%"PRIx64 " type:%s, data:%s", slowLogData->clusterId, queueTypeStr[slowLogData->type], slowLogData->data);
tscDebug("[monitor] write slow log to queue, clusterId:%" PRIx64 " type:%s, data:%s", slowLogData->clusterId,
queueTypeStr[slowLogData->type], slowLogData->data);
if (taosWriteQitem(monitorQueue, slowLogData) == 0) {
tsem2_post(&monitorSem);
} else {

12
source/client/src/clientRawBlockWrite.c
View File

@ -1116,9 +1116,9 @@ static int32_t taosCreateTable(TAOS* taos, void* meta, int32_t metaLen) {
if (taosHashGetSize(pVgroupHashmap) == 0) {
goto end;
}
SArray* pBufArray = serializeVgroupsCreateTableBatch(pVgroupHashmap);
if (NULL == pBufArray) {
code = TSDB_CODE_OUT_OF_MEMORY;
SArray* pBufArray = NULL;
code = serializeVgroupsCreateTableBatch(pVgroupHashmap, &pBufArray);
if (TSDB_CODE_SUCCESS != code) {
goto end;
}
@ -1264,9 +1264,9 @@ static int32_t taosDropTable(TAOS* taos, void* meta, int32_t metaLen) {
if (taosHashGetSize(pVgroupHashmap) == 0) {
goto end;
}
SArray* pBufArray = serializeVgroupsDropTableBatch(pVgroupHashmap);
if (NULL == pBufArray) {
code = TSDB_CODE_OUT_OF_MEMORY;
SArray* pBufArray = NULL;
code = serializeVgroupsDropTableBatch(pVgroupHashmap, &pBufArray);
if (TSDB_CODE_SUCCESS != code) {
goto end;
}

66
source/client/src/clientTmq.c
View File

@ -733,11 +733,15 @@ end:
}
static void generateTimedTask(int64_t refId, int32_t type) {
tmq_t* tmq = taosAcquireRef(tmqMgmt.rsetId, refId);
tmq_t* tmq;
int8_t* pTaskType;
int32_t code;
tmq = taosAcquireRef(tmqMgmt.rsetId, refId);
if (tmq == NULL) return;
int8_t* pTaskType = taosAllocateQitem(sizeof(int8_t), DEF_QITEM, 0);
if (pTaskType != NULL){
code = taosAllocateQitem(sizeof(int8_t), DEF_QITEM, 0, (void**)&pTaskType);
if (code == TSDB_CODE_SUCCESS) {
*pTaskType = type;
if (taosWriteQitem(tmq->delayedTask, pTaskType) == 0) {
tsem2_post(&tmq->rspSem);
@ -890,7 +894,12 @@ static void defaultCommitCbFn(tmq_t* pTmq, int32_t code, void* param) {
}
int32_t tmqHandleAllDelayedTask(tmq_t* pTmq) {
STaosQall* qall = taosAllocateQall();
STaosQall* qall;
int32_t code;
code = taosAllocateQall(&qall);
if (code) return code;
taosReadAllQitems(pTmq->delayedTask, qall);
int32_t numOfItems = taosQallItemSize(qall);
@ -914,7 +923,8 @@ int32_t tmqHandleAllDelayedTask(tmq_t* pTmq) {
asyncCommitAllOffsets(pTmq, pCallbackFn, pTmq->commitCbUserParam);
tscDebug("consumer:0x%" PRIx64 " next commit to vnode(s) in %.2fs", pTmq->consumerId,
pTmq->autoCommitInterval / 1000.0);
taosTmrReset(tmqAssignDelayedCommitTask, pTmq->autoCommitInterval, (void*)(pTmq->refId), tmqMgmt.timer, &pTmq->commitTimer);
taosTmrReset(tmqAssignDelayedCommitTask, pTmq->autoCommitInterval, (void*)(pTmq->refId), tmqMgmt.timer,
&pTmq->commitTimer);
} else {
tscError("consumer:0x%" PRIx64 " invalid task type:%d", pTmq->consumerId, *pTaskType);
}
@ -1101,6 +1111,8 @@ void tmqMgmtClose(void) {
if (errstr != NULL) snprintf(errstr, errstrLen, MSG);
tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
int32_t code;
if (conf == NULL) {
SET_ERROR_MSG_TMQ("configure is null")
return NULL;
@ -1124,12 +1136,31 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
const char* pass = conf->pass == NULL ? TSDB_DEFAULT_PASS : conf->pass;
pTmq->clientTopics = taosArrayInit(0, sizeof(SMqClientTopic));
pTmq->mqueue = taosOpenQueue();
pTmq->delayedTask = taosOpenQueue();
pTmq->qall = taosAllocateQall();
code = taosOpenQueue(&pTmq->mqueue);
if (code) {
terrno = code;
tscError("consumer:0x%" PRIx64 " setup failed since %s, groupId:%s", pTmq->consumerId, terrstr(), pTmq->groupId);
SET_ERROR_MSG_TMQ("open queue failed")
goto _failed;
}
if (pTmq->clientTopics == NULL || pTmq->mqueue == NULL || pTmq->qall == NULL || pTmq->delayedTask == NULL ||
conf->groupId[0] == 0) {
code = taosOpenQueue(&pTmq->delayedTask);
if (code) {
terrno = code;
tscError("consumer:0x%" PRIx64 " setup failed since %s, groupId:%s", pTmq->consumerId, terrstr(), pTmq->groupId);
SET_ERROR_MSG_TMQ("open delayed task queue failed")
goto _failed;
}
code = taosAllocateQall(&pTmq->qall);
if (code) {
terrno = code;
tscError("consumer:0x%" PRIx64 " setup failed since %s, groupId:%s", pTmq->consumerId, terrstr(), pTmq->groupId);
SET_ERROR_MSG_TMQ("allocate qall failed")
goto _failed;
}
if (conf->groupId[0] == 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
tscError("consumer:0x%" PRIx64 " setup failed since %s, groupId:%s", pTmq->consumerId, terrstr(), pTmq->groupId);
SET_ERROR_MSG_TMQ("malloc tmq element failed or group is empty")
@ -1315,7 +1346,8 @@ int32_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) {
// init ep timer
tmq->epTimer = taosTmrStart(tmqAssignAskEpTask, DEFAULT_ASKEP_INTERVAL, (void*)(tmq->refId), tmqMgmt.timer);
// init auto commit timer
tmq->commitTimer = taosTmrStart(tmqAssignDelayedCommitTask, tmq->autoCommitInterval, (void*)(tmq->refId), tmqMgmt.timer);
tmq->commitTimer =
taosTmrStart(tmqAssignDelayedCommitTask, tmq->autoCommitInterval, (void*)(tmq->refId), tmqMgmt.timer);
FAIL:
taosArrayDestroyP(req.topicNames, taosMemoryFree);
@ -1381,11 +1413,11 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) {
goto FAIL;
}
SMqPollRspWrapper* pRspWrapper = taosAllocateQitem(sizeof(SMqPollRspWrapper), DEF_QITEM, 0);
if (pRspWrapper == NULL) {
SMqPollRspWrapper* pRspWrapper;
code = taosAllocateQitem(sizeof(SMqPollRspWrapper), DEF_QITEM, 0, (void**)&pRspWrapper);
if (code) {
tscWarn("consumer:0x%" PRIx64 " msg discard from vgId:%d, since out of memory", tmq->consumerId, vgId);
taosReleaseRef(tmqMgmt.rsetId, refId);
code = TSDB_CODE_OUT_OF_MEMORY;
goto FAIL;
}
@ -2600,9 +2632,9 @@ int32_t askEpCb(void* param, SDataBuf* pMsg, int32_t code) {
doUpdateLocalEp(tmq, head->epoch, &rsp);
tDeleteSMqAskEpRsp(&rsp);
} else {
SMqAskEpRspWrapper* pWrapper = taosAllocateQitem(sizeof(SMqAskEpRspWrapper), DEF_QITEM, 0);
if (pWrapper == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
SMqAskEpRspWrapper* pWrapper;
code = taosAllocateQitem(sizeof(SMqAskEpRspWrapper), DEF_QITEM, 0, (void**)&pWrapper);
if (code) {
goto END;
}

5
source/common/src/tname.c
View File

@ -98,7 +98,7 @@ SName* toName(int32_t acctId, const char* pDbName, const char* pTableName, SName
int32_t tNameExtractFullName(const SName* name, char* dst) {
// invalid full name format, abort
if (!tNameIsValid(name)) {
return -1;
return TSDB_CODE_INVALID_PARA;
}
int32_t len = snprintf(dst, TSDB_DB_FNAME_LEN, "%d.%s", name->acctId, name->dbname);
@ -154,8 +154,7 @@ int32_t tNameGetDbName(const SName* name, char* dst) {
const char* tNameGetDbNameP(const SName* name) { return &name->dbname[0]; }
int32_t tNameGetFullDbName(const SName* name, char* dst) {
snprintf(dst, TSDB_DB_FNAME_LEN, "%d.%s", name->acctId, name->dbname);
return 0;
return snprintf(dst, TSDB_DB_FNAME_LEN, "%d.%s", name->acctId, name->dbname);
}
bool tNameIsEmpty(const SName* name) { return name->type == 0 || name->acctId == 0; }

12
source/dnode/mgmt/mgmt_mnode/src/mmWorker.c
View File

@ -153,6 +153,8 @@ int32_t mmPutMsgToFetchQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
}
int32_t mmPutMsgToQueue(SMnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) {
int32_t code;
SSingleWorker *pWorker = NULL;
switch (qtype) {
case WRITE_QUEUE:
@ -181,13 +183,15 @@ int32_t mmPutMsgToQueue(SMnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) {
}
if (pWorker == NULL) return -1;
SRpcMsg *pMsg = taosAllocateQitem(sizeof(SRpcMsg), RPC_QITEM, pRpc->contLen);
if (pMsg == NULL) return -1;
SRpcMsg *pMsg;
code = taosAllocateQitem(sizeof(SRpcMsg), RPC_QITEM, pRpc->contLen, (void **)&pMsg);
if (code) return code;
memcpy(pMsg, pRpc, sizeof(SRpcMsg));
pRpc->pCont = NULL;
dTrace("msg:%p, is created and will put into %s queue, type:%s len:%d", pMsg, pWorker->name, TMSG_INFO(pRpc->msgType), pRpc->contLen);
int32_t code = mmPutMsgToWorker(pMgmt, pWorker, pMsg);
dTrace("msg:%p, is created and will put into %s queue, type:%s len:%d", pMsg, pWorker->name, TMSG_INFO(pRpc->msgType),
pRpc->contLen);
code = mmPutMsgToWorker(pMgmt, pWorker, pMsg);
if (code != 0) {
dTrace("msg:%p, is freed", pMsg);
rpcFreeCont(pMsg->pCont);

7
source/dnode/mgmt/mgmt_qnode/src/qmWorker.c
View File

@ -58,8 +58,11 @@ int32_t qmPutNodeMsgToFetchQueue(SQnodeMgmt *pMgmt, SRpcMsg *pMsg) {
}
int32_t qmPutRpcMsgToQueue(SQnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) {
SRpcMsg *pMsg = taosAllocateQitem(sizeof(SRpcMsg), RPC_QITEM, pRpc->contLen);
if (pMsg == NULL) return -1;
int32_t code;
SRpcMsg *pMsg;
code = taosAllocateQitem(sizeof(SRpcMsg), RPC_QITEM, pRpc->contLen, (void **)&pMsg);
if (code) return code;
memcpy(pMsg, pRpc, sizeof(SRpcMsg));
pRpc->pCont = NULL;

7
source/dnode/mgmt/mgmt_snode/src/smWorker.c
View File

@ -126,8 +126,11 @@ void smStopWorker(SSnodeMgmt *pMgmt) {
}
int32_t smPutMsgToQueue(SSnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) {
SRpcMsg *pMsg = taosAllocateQitem(sizeof(SRpcMsg), RPC_QITEM, pRpc->contLen);
if (pMsg == NULL) {
int32_t code;
SRpcMsg *pMsg;
code = taosAllocateQitem(sizeof(SRpcMsg), RPC_QITEM, pRpc->contLen, (void **)&pMsg);
if (code) {
rpcFreeCont(pRpc->pCont);
pRpc->pCont = NULL;
return -1;

8
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
View File

@ -311,6 +311,7 @@ int32_t vmPutMsgToMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
}
int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) {
int32_t code;
if (pRpc->contLen < sizeof(SMsgHead)) {
dError("invalid rpc msg with no msg head at pCont. pRpc:%p, type:%s, len:%d", pRpc, TMSG_INFO(pRpc->msgType),
pRpc->contLen);
@ -320,8 +321,9 @@ int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) {
}
EQItype itype = APPLY_QUEUE == qtype ? DEF_QITEM : RPC_QITEM;
SRpcMsg *pMsg = taosAllocateQitem(sizeof(SRpcMsg), itype, pRpc->contLen);
if (pMsg == NULL) {
SRpcMsg *pMsg;
code = taosAllocateQitem(sizeof(SRpcMsg), itype, pRpc->contLen, (void **)&pMsg);
if (code) {
rpcFreeCont(pRpc->pCont);
pRpc->pCont = NULL;
return -1;
@ -335,7 +337,7 @@ int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) {
memcpy(pMsg, pRpc, sizeof(SRpcMsg));
pRpc->pCont = NULL;
int32_t code = vmPutMsgToQueue(pMgmt, pMsg, qtype);
code = vmPutMsgToQueue(pMgmt, pMsg, qtype);
if (code != 0) {
dTrace("msg:%p, is freed", pMsg);
rpcFreeCont(pMsg->pCont);

4
source/dnode/mgmt/node_mgmt/src/dmTransport.c
View File

@ -210,8 +210,8 @@ static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) {
pRpc->info.wrapper = pWrapper;
EQItype itype = IsReq(pRpc) ? RPC_QITEM : DEF_QITEM; // rsp msg is not restricted by tsRpcQueueMemoryUsed
pMsg = taosAllocateQitem(sizeof(SRpcMsg), itype, pRpc->contLen);
if (pMsg == NULL) goto _OVER;
code = taosAllocateQitem(sizeof(SRpcMsg), itype, pRpc->contLen, (void **)&pMsg);
if (code) goto _OVER;
memcpy(pMsg, pRpc, sizeof(SRpcMsg));
dGTrace("msg:%p, is created, type:%s handle:%p len:%d", pMsg, TMSG_INFO(pRpc->msgType), pMsg->info.handle,

1
source/dnode/mnode/impl/src/mndStreamUtil.c
View File

@ -105,6 +105,7 @@ SArray *mndTakeVgroupSnapshot(SMnode *pMnode, bool *allReady) {
mInfo("vgId:%d replica:%d inconsistent with other vgroups replica:%d, not ready for stream operations",
pVgroup->vgId, pVgroup->replica, replica);
*allReady = false;
sdbRelease(pSdb, pVgroup);
break;
}
}

19
source/dnode/vnode/src/sma/smaRollup.c
View File

@ -362,6 +362,8 @@ static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat
* @return int32_t
*/
int32_t tdRSmaProcessCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, const char *tbName) {
int32_t code;
if ((param->qmsgLen[0] == 0) && (param->qmsgLen[1] == 0)) {
smaDebug("vgId:%d, no qmsg1/qmsg2 for rollup table %s %" PRIi64, SMA_VID(pSma), tbName, suid);
return TSDB_CODE_SUCCESS;
@ -401,8 +403,13 @@ int32_t tdRSmaProcessCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, con
pRSmaInfo->suid = suid;
T_REF_INIT_VAL(pRSmaInfo, 1);
if (!(pRSmaInfo->queue = taosOpenQueue()) || !(pRSmaInfo->qall = taosAllocateQall()) ||
tdSetRSmaInfoItemParams(pSma, param, pStat, pRSmaInfo, 0) < 0 ||
code = taosOpenQueue(&pRSmaInfo->queue);
if (code) goto _err;
code = taosAllocateQall(&pRSmaInfo->qall);
if (code) goto _err;
if (tdSetRSmaInfoItemParams(pSma, param, pStat, pRSmaInfo, 0) < 0 ||
tdSetRSmaInfoItemParams(pSma, param, pStat, pRSmaInfo, 1) < 0) {
goto _err;
}
@ -795,11 +802,13 @@ _exit:
*/
static int32_t tdExecuteRSmaImplAsync(SSma *pSma, int64_t version, const void *pMsg, int32_t len, int32_t inputType,
SRSmaInfo *pInfo, tb_uid_t suid) {
int32_t code;
int32_t size = RSMA_EXEC_MSG_HLEN + len; // header + payload
void * qItem = taosAllocateQitem(size, DEF_QITEM, 0);
void *qItem;
if (!qItem) {
return TSDB_CODE_FAILED;
code = taosAllocateQitem(size, DEF_QITEM, 0, (void **)&qItem);
if (code) {
return code;
}
void *pItem = qItem;

24
source/dnode/vnode/src/tq/tqUtil.c
View File

@ -164,7 +164,8 @@ static int32_t extractDataAndRspForNormalSubscribe(STQ* pTq, STqHandle* pHandle,
taosWUnLockLatch(&pTq->lock);
}
tOffsetCopy(&dataRsp.common.reqOffset, pOffset); // reqOffset represents the current date offset, may be changed if wal not exists
tOffsetCopy(&dataRsp.common.reqOffset,
pOffset); // reqOffset represents the current date offset, may be changed if wal not exists
code = tqSendDataRsp(pHandle, pMsg, pRequest, &dataRsp, TMQ_MSG_TYPE__POLL_DATA_RSP, vgId);
end : {
@ -185,7 +186,8 @@ end : {
int32_t len = pHead->bodyLen - sizeof(SMsgHead); \
tDecoderInit(&decoder, data, len); \
if (DECODE_FUNC(&decoder, &req) == 0 && (req.source & TD_REQ_FROM_TAOX) != 0) { \
tqDebug("tmq poll: consumer:0x%" PRIx64 " (epoch %d) iter log, jump meta for, vgId:%d offset %" PRId64 " msgType %d", \
tqDebug("tmq poll: consumer:0x%" PRIx64 " (epoch %d) iter log, jump meta for, vgId:%d offset %" PRId64 \
" msgType %d", \
pRequest->consumerId, pRequest->epoch, vgId, fetchVer, pHead->msgType); \
fetchVer++; \
DELETE_FUNC(&req); \
@ -195,8 +197,7 @@ end : {
DELETE_FUNC(&req); \
tDecoderClear(&decoder);
static void tDeleteCommon(void* parm) {
}
static void tDeleteCommon(void* parm) {}
static int32_t extractDataAndRspForDbStbSubscribe(STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest,
SRpcMsg* pMsg, STqOffsetVal* offset) {
@ -416,8 +417,8 @@ static void initMqRspHead(SMqRspHead* pMsgHead, int32_t type, int32_t epoch, int
pMsgHead->walever = ever;
}
int32_t tqSendBatchMetaPollRsp(STqHandle* pHandle, const SRpcMsg* pMsg, const SMqPollReq* pReq, const SMqBatchMetaRsp* pRsp,
int32_t vgId) {
int32_t tqSendBatchMetaPollRsp(STqHandle* pHandle, const SRpcMsg* pMsg, const SMqPollReq* pReq,
const SMqBatchMetaRsp* pRsp, int32_t vgId) {
int32_t len = 0;
int32_t code = 0;
tEncodeSize(tEncodeMqBatchMetaRsp, pRsp, len, code);
@ -444,8 +445,8 @@ int32_t tqSendBatchMetaPollRsp(STqHandle* pHandle, const SRpcMsg* pMsg, const SM
SRpcMsg resp = {.info = pMsg->info, .pCont = buf, .contLen = tlen, .code = 0};
tmsgSendRsp(&resp);
tqDebug("vgId:%d, from consumer:0x%" PRIx64 " (epoch %d) send rsp, res msg type: batch meta, size:%ld offset type:%d", vgId,
pReq->consumerId, pReq->epoch, taosArrayGetSize(pRsp->batchMetaReq), pRsp->rspOffset.type);
tqDebug("vgId:%d, from consumer:0x%" PRIx64 " (epoch %d) send rsp, res msg type: batch meta, size:%ld offset type:%d",
vgId, pReq->consumerId, pReq->epoch, taosArrayGetSize(pRsp->batchMetaReq), pRsp->rspOffset.type);
return 0;
}
@ -527,6 +528,7 @@ int32_t tqDoSendDataRsp(const SRpcHandleInfo* pRpcHandleInfo, const void* pRsp,
}
int32_t tqExtractDelDataBlock(const void* pData, int32_t len, int64_t ver, void** pRefBlock, int32_t type) {
int32_t code;
SDecoder* pCoder = &(SDecoder){0};
SDeleteRes* pRes = &(SDeleteRes){0};
@ -570,11 +572,11 @@ int32_t tqExtractDelDataBlock(const void* pData, int32_t len, int64_t ver, void*
taosArrayDestroy(pRes->uidList);
if (type == 0) {
*pRefBlock = taosAllocateQitem(sizeof(SStreamRefDataBlock), DEF_QITEM, 0);
if (*pRefBlock == NULL) {
code = taosAllocateQitem(sizeof(SStreamRefDataBlock), DEF_QITEM, 0, pRefBlock);
if (code) {
blockDataCleanup(pDelBlock);
taosMemoryFree(pDelBlock);
return TSDB_CODE_OUT_OF_MEMORY;
return code;
}
((SStreamRefDataBlock*)(*pRefBlock))->type = STREAM_INPUT__REF_DATA_BLOCK;

15
source/libs/executor/src/dataDeleter.c
View File

@ -130,9 +130,11 @@ static int32_t getStatus(SDataDeleterHandle* pDeleter) {
static int32_t putDataBlock(SDataSinkHandle* pHandle, const SInputData* pInput, bool* pContinue) {
SDataDeleterHandle* pDeleter = (SDataDeleterHandle*)pHandle;
SDataDeleterBuf* pBuf = taosAllocateQitem(sizeof(SDataDeleterBuf), DEF_QITEM, 0);
if (NULL == pBuf) {
return TSDB_CODE_OUT_OF_MEMORY;
SDataDeleterBuf* pBuf;
int32_t code = taosAllocateQitem(sizeof(SDataDeleterBuf), DEF_QITEM, 0, (void**)&pBuf);
if (code) {
return code;
}
if (!allocBuf(pDeleter, pInput, pBuf)) {
@ -270,12 +272,11 @@ int32_t createDataDeleter(SDataSinkManager* pManager, const SDataSinkNode* pData
deleter->pParam = pParam;
deleter->status = DS_BUF_EMPTY;
deleter->queryEnd = false;
deleter->pDataBlocks = taosOpenQueue();
taosThreadMutexInit(&deleter->mutex, NULL);
if (NULL == deleter->pDataBlocks) {
code = TSDB_CODE_OUT_OF_MEMORY;
code = taosOpenQueue(&deleter->pDataBlocks);
if (code) {
goto _end;
}
taosThreadMutexInit(&deleter->mutex, NULL);
*pHandle = deleter;
return code;

20
source/libs/executor/src/dataDispatcher.c
View File

@ -104,7 +104,8 @@ static void toDataCacheEntry(SDataDispatchHandle* pHandle, const SInputData* pIn
}
int32_t dataLen = blockEncode(pInput->pData, pHandle->pCompressBuf, numOfCols);
int32_t len = tsCompressString(pHandle->pCompressBuf, dataLen, 1, pEntry->data, pBuf->allocSize, ONE_STAGE_COMP, NULL, 0);
int32_t len =
tsCompressString(pHandle->pCompressBuf, dataLen, 1, pEntry->data, pBuf->allocSize, ONE_STAGE_COMP, NULL, 0);
if (len < dataLen) {
pEntry->compressed = 1;
pEntry->dataLen = len;
@ -168,9 +169,11 @@ static int32_t getStatus(SDataDispatchHandle* pDispatcher) {
static int32_t putDataBlock(SDataSinkHandle* pHandle, const SInputData* pInput, bool* pContinue) {
int32_t code = 0;
SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;
SDataDispatchBuf* pBuf = taosAllocateQitem(sizeof(SDataDispatchBuf), DEF_QITEM, 0);
if (NULL == pBuf) {
return TSDB_CODE_OUT_OF_MEMORY;
SDataDispatchBuf* pBuf;
code = taosAllocateQitem(sizeof(SDataDispatchBuf), DEF_QITEM, 0, (void**)&pBuf);
if (code) {
return code;
}
if (!allocBuf(pDispatcher, pInput, pBuf)) {
@ -228,7 +231,6 @@ static void getDataLength(SDataSinkHandle* pHandle, int64_t* pLen, int64_t* pRow
((SDataCacheEntry*)(pDispatcher->nextOutput.pData))->numOfRows);
}
static int32_t getDataBlock(SDataSinkHandle* pHandle, SOutputData* pOutput) {
SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;
if (NULL == pDispatcher->nextOutput.pData) {
@ -291,6 +293,8 @@ static int32_t getCacheSize(struct SDataSinkHandle* pHandle, uint64_t* size) {
}
int32_t createDataDispatcher(SDataSinkManager* pManager, const SDataSinkNode* pDataSink, DataSinkHandle* pHandle) {
int32_t code;
SDataDispatchHandle* dispatcher = taosMemoryCalloc(1, sizeof(SDataDispatchHandle));
if (NULL == dispatcher) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
@ -309,7 +313,11 @@ int32_t createDataDispatcher(SDataSinkManager* pManager, const SDataSinkNode* pD
dispatcher->pSchema = pDataSink->pInputDataBlockDesc;
dispatcher->status = DS_BUF_EMPTY;
dispatcher->queryEnd = false;
dispatcher->pDataBlocks = taosOpenQueue();
code = taosOpenQueue(&dispatcher->pDataBlocks);
if (code) {
terrno = code;
goto _return;
}
taosThreadMutexInit(&dispatcher->mutex, NULL);
if (NULL == dispatcher->pDataBlocks) {

2
source/libs/parser/inc/parTranslater.h
View File

@ -47,7 +47,7 @@ typedef struct STranslateContext {
SNode* pPostRoot;
} STranslateContext;
bool biRewriteToTbnameFunc(STranslateContext* pCxt, SNode** ppNode);
int32_t biRewriteToTbnameFunc(STranslateContext* pCxt, SNode** ppNode, bool* pRet);
int32_t biRewriteSelectStar(STranslateContext* pCxt, SSelectStmt* pSelect);
int32_t biCheckCreateTableTbnameCol(STranslateContext* pCxt, SCreateTableStmt* pStmt);
int32_t findTable(STranslateContext* pCxt, const char* pTableAlias, STableNode** pOutput);

2
source/libs/parser/inc/sql.y
View File

@ -509,7 +509,7 @@ cmd ::= SHOW USERS FULL.
cmd ::= SHOW USER PRIVILEGES. { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_USER_PRIVILEGES_STMT); }
cmd ::= SHOW db_kind_opt(A) DATABASES. {
pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_DATABASES_STMT);
setShowKind(pCxt, pCxt->pRootNode, A);
(void)setShowKind(pCxt, pCxt->pRootNode, A);
}
cmd ::= SHOW table_kind_db_name_cond_opt(A) TABLES like_pattern_opt(B). {
pCxt->pRootNode = createShowTablesStmt(pCxt, A, B, OP_TYPE_LIKE);

43
source/libs/parser/src/parAstCreater.c
View File

@ -101,7 +101,7 @@ static bool checkPassword(SAstCreateContext* pCxt, const SToken* pPasswordToken,
pCxt->errCode = generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_NAME_OR_PASSWD_TOO_LONG);
} else {
strncpy(pPassword, pPasswordToken->z, pPasswordToken->n);
strdequote(pPassword);
(void)strdequote(pPassword);
if (strtrim(pPassword) <= 0) {
pCxt->errCode = generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_PASSWD_EMPTY);
} else if (invalidPassword(pPassword)) {
@ -126,8 +126,8 @@ static int32_t parseEndpoint(SAstCreateContext* pCxt, const SToken* pEp, char* p
char ep[TSDB_FQDN_LEN + 1 + 5] = {0};
COPY_STRING_FORM_ID_TOKEN(ep, pEp);
strdequote(ep);
strtrim(ep);
(void)strdequote(ep);
(void)strtrim(ep);
if (NULL == pPort) {
strcpy(pFqdn, ep);
return TSDB_CODE_SUCCESS;
@ -371,7 +371,7 @@ SNode* createValueNode(SAstCreateContext* pCxt, int32_t dataType, const SToken*
val->literal = strndup(pLiteral->z, pLiteral->n);
if (TK_NK_ID != pLiteral->type && TK_TIMEZONE != pLiteral->type &&
(IS_VAR_DATA_TYPE(dataType) || TSDB_DATA_TYPE_TIMESTAMP == dataType)) {
trimString(pLiteral->z, pLiteral->n, val->literal, pLiteral->n);
(void)trimString(pLiteral->z, pLiteral->n, val->literal, pLiteral->n);
}
CHECK_OUT_OF_MEM(val->literal);
val->node.resType.type = dataType;
@ -784,7 +784,10 @@ SNode* createPlaceholderValueNode(SAstCreateContext* pCxt, const SToken* pLitera
return NULL;
}
}
taosArrayPush(pCxt->pPlaceholderValues, &val);
if (NULL == taosArrayPush(pCxt->pPlaceholderValues, &val)) {
pCxt->errCode = TSDB_CODE_OUT_OF_MEMORY;
}
CHECK_PARSER_STATUS(pCxt);
return (SNode*)val;
}
@ -905,7 +908,8 @@ SNode* createCastFunctionNode(SAstCreateContext* pCxt, SNode* pExpr, SDataType d
} else if (TSDB_DATA_TYPE_NCHAR == dt.type) {
func->node.resType.bytes = func->node.resType.bytes * TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE;
}
nodesListMakeAppend(&func->pParameterList, pExpr);
pCxt->errCode = nodesListMakeAppend(&func->pParameterList, pExpr);
CHECK_PARSER_STATUS(pCxt);
return (SNode*)func;
}
@ -923,8 +927,10 @@ SNode* createNodeListNodeEx(SAstCreateContext* pCxt, SNode* p1, SNode* p2) {
CHECK_OUT_OF_MEM(list);
list->pNodeList = nodesMakeList();
CHECK_OUT_OF_MEM(list->pNodeList);
nodesListAppend(list->pNodeList, p1);
nodesListAppend(list->pNodeList, p2);
pCxt->errCode = nodesListAppend(list->pNodeList, p1);
CHECK_PARSER_STATUS(pCxt);
pCxt->errCode = nodesListAppend(list->pNodeList, p2);
CHECK_PARSER_STATUS(pCxt);
return (SNode*)list;
}
@ -1121,7 +1127,8 @@ SNode* createGroupingSetNode(SAstCreateContext* pCxt, SNode* pNode) {
CHECK_OUT_OF_MEM(groupingSet);
groupingSet->groupingSetType = GP_TYPE_NORMAL;
groupingSet->pParameterList = nodesMakeList();
nodesListAppend(groupingSet->pParameterList, pNode);
pCxt->errCode = nodesListAppend(groupingSet->pParameterList, pNode);
CHECK_PARSER_STATUS(pCxt);
return (SNode*)groupingSet;
}
@ -1842,7 +1849,7 @@ SNode* createCreateSubTableFromFileClause(SAstCreateContext* pCxt, bool ignoreEx
pStmt->ignoreExists = ignoreExists;
pStmt->pSpecificTags = pSpecificTags;
if (TK_NK_STRING == pFilePath->type) {
trimString(pFilePath->z, pFilePath->n, pStmt->filePath, PATH_MAX);
(void)trimString(pFilePath->z, pFilePath->n, pStmt->filePath, PATH_MAX);
} else {
strncpy(pStmt->filePath, pFilePath->z, pFilePath->n);
}
@ -2087,7 +2094,7 @@ SNode* createShowTablesStmt(SAstCreateContext* pCxt, SShowTablesOption option, S
pDbName = createIdentifierValueNode(pCxt, &option.dbName);
}
SNode* pStmt = createShowStmtWithCond(pCxt, QUERY_NODE_SHOW_TABLES_STMT, pDbName, pTbName, tableCondType);
setShowKind(pCxt, pStmt, option.kind);
(void)setShowKind(pCxt, pStmt, option.kind);
return pStmt;
}
@ -2408,9 +2415,9 @@ SNode* createAlterDnodeStmt(SAstCreateContext* pCxt, const SToken* pDnode, const
} else {
pStmt->dnodeId = -1;
}
trimString(pConfig->z, pConfig->n, pStmt->config, sizeof(pStmt->config));
(void)trimString(pConfig->z, pConfig->n, pStmt->config, sizeof(pStmt->config));
if (NULL != pValue) {
trimString(pValue->z, pValue->n, pStmt->value, sizeof(pStmt->value));
(void)trimString(pValue->z, pValue->n, pStmt->value, sizeof(pStmt->value));
}
return (SNode*)pStmt;
}
@ -2591,9 +2598,9 @@ SNode* createAlterClusterStmt(SAstCreateContext* pCxt, const SToken* pConfig, co
CHECK_PARSER_STATUS(pCxt);
SAlterClusterStmt* pStmt = (SAlterClusterStmt*)nodesMakeNode(QUERY_NODE_ALTER_CLUSTER_STMT);
CHECK_OUT_OF_MEM(pStmt);
trimString(pConfig->z, pConfig->n, pStmt->config, sizeof(pStmt->config));
(void)trimString(pConfig->z, pConfig->n, pStmt->config, sizeof(pStmt->config));
if (NULL != pValue) {
trimString(pValue->z, pValue->n, pStmt->value, sizeof(pStmt->value));
(void)trimString(pValue->z, pValue->n, pStmt->value, sizeof(pStmt->value));
}
return (SNode*)pStmt;
}
@ -2602,9 +2609,9 @@ SNode* createAlterLocalStmt(SAstCreateContext* pCxt, const SToken* pConfig, cons
CHECK_PARSER_STATUS(pCxt);
SAlterLocalStmt* pStmt = (SAlterLocalStmt*)nodesMakeNode(QUERY_NODE_ALTER_LOCAL_STMT);
CHECK_OUT_OF_MEM(pStmt);
trimString(pConfig->z, pConfig->n, pStmt->config, sizeof(pStmt->config));
(void)trimString(pConfig->z, pConfig->n, pStmt->config, sizeof(pStmt->config));
if (NULL != pValue) {
trimString(pValue->z, pValue->n, pStmt->value, sizeof(pStmt->value));
(void)trimString(pValue->z, pValue->n, pStmt->value, sizeof(pStmt->value));
}
return (SNode*)pStmt;
}
@ -2869,7 +2876,7 @@ SNode* createKillQueryStmt(SAstCreateContext* pCxt, const SToken* pQueryId) {
CHECK_PARSER_STATUS(pCxt);
SKillQueryStmt* pStmt = (SKillQueryStmt*)nodesMakeNode(QUERY_NODE_KILL_QUERY_STMT);
CHECK_OUT_OF_MEM(pStmt);
trimString(pQueryId->z, pQueryId->n, pStmt->queryId, sizeof(pStmt->queryId) - 1);
(void)trimString(pQueryId->z, pQueryId->n, pStmt->queryId, sizeof(pStmt->queryId) - 1);
return (SNode*)pStmt;
}

1408
source/libs/parser/src/parTranslater.c
View File

File diff suppressed because it is too large Load Diff

4
source/libs/qcom/src/queryUtil.c
View File

@ -158,7 +158,9 @@ int32_t cleanupTaskQueue() {
}
int32_t taosAsyncExec(__async_exec_fn_t execFn, void* execParam, int32_t* code) {
SSchedMsg* pSchedMsg = taosAllocateQitem(sizeof(SSchedMsg), DEF_QITEM, 0);
SSchedMsg* pSchedMsg;
int32_t rc = taosAllocateQitem(sizeof(SSchedMsg), DEF_QITEM, 0, (void **)&pSchedMsg);
if (rc) return rc;
pSchedMsg->fp = NULL;
pSchedMsg->ahandle = execFn;
pSchedMsg->thandle = execParam;

8
source/libs/stream/src/streamCheckpoint.c
View File

@ -31,9 +31,11 @@ static void checkpointTriggerMonitorFn(void* param, void* tmrId);
SStreamDataBlock* createChkptTriggerBlock(SStreamTask* pTask, int32_t checkpointType, int64_t checkpointId,
int32_t transId, int32_t srcTaskId) {
SStreamDataBlock* pChkpoint = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, sizeof(SSDataBlock));
if (pChkpoint == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
SStreamDataBlock* pChkpoint;
int32_t code = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, sizeof(SSDataBlock), (void**)&pChkpoint);
if (code) {
terrno = code;
return NULL;
}

36
source/libs/stream/src/streamData.c
View File

@ -16,8 +16,11 @@
#include "streamInt.h"
SStreamDataBlock* createStreamBlockFromDispatchMsg(const SStreamDispatchReq* pReq, int32_t blockType, int32_t srcVg) {
SStreamDataBlock* pData = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, pReq->totalLen);
if (pData == NULL) {
SStreamDataBlock* pData;
int32_t code = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, pReq->totalLen, (void**)&pData);
if (code) {
terrno = code;
return NULL;
}
@ -69,10 +72,14 @@ SStreamDataBlock* createStreamBlockFromDispatchMsg(const SStreamDispatchReq* pRe
return pData;
}
SStreamDataBlock* createStreamBlockFromResults(SStreamQueueItem* pItem, SStreamTask* pTask, int64_t resultSize, SArray* pRes) {
SStreamDataBlock* pStreamBlocks = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, resultSize);
if (pStreamBlocks == NULL) {
SStreamDataBlock* createStreamBlockFromResults(SStreamQueueItem* pItem, SStreamTask* pTask, int64_t resultSize,
SArray* pRes) {
SStreamDataBlock* pStreamBlocks;
int32_t code = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, resultSize, (void**)&pStreamBlocks);
if (code) {
taosArrayClearEx(pRes, (FDelete)blockDataFreeRes);
terrno = code;
return NULL;
}
@ -132,8 +139,10 @@ int32_t streamRetrieveReqToData(const SStreamRetrieveReq* pReq, SStreamDataBlock
}
SStreamDataSubmit* streamDataSubmitNew(SPackedData* pData, int32_t type) {
SStreamDataSubmit* pDataSubmit = (SStreamDataSubmit*)taosAllocateQitem(sizeof(SStreamDataSubmit), DEF_QITEM, pData->msgLen);
if (pDataSubmit == NULL) {
SStreamDataSubmit* pDataSubmit;
int32_t code = taosAllocateQitem(sizeof(SStreamDataSubmit), DEF_QITEM, pData->msgLen, (void**)&pDataSubmit);
if (code) {
terrno = code;
return NULL;
}
@ -151,8 +160,11 @@ void streamDataSubmitDestroy(SStreamDataSubmit* pDataSubmit) {
}
SStreamMergedSubmit* streamMergedSubmitNew() {
SStreamMergedSubmit* pMerged = (SStreamMergedSubmit*)taosAllocateQitem(sizeof(SStreamMergedSubmit), DEF_QITEM, 0);
if (pMerged == NULL) {
SStreamMergedSubmit* pMerged;
int32_t code = taosAllocateQitem(sizeof(SStreamMergedSubmit), DEF_QITEM, 0, (void**)&pMerged);
if (code) {
terrno = code;
return NULL;
}
@ -212,7 +224,8 @@ SStreamQueueItem* streamQueueMergeQueueItem(SStreamQueueItem* dst, SStreamQueueI
taosFreeQitem(pElem);
return (SStreamQueueItem*)pMerged;
} else {
stDebug("block type:%s not merged with existed blocks list, type:%d", streamQueueItemGetTypeStr(pElem->type), dst->type);
stDebug("block type:%s not merged with existed blocks list, type:%d", streamQueueItemGetTypeStr(pElem->type),
dst->type);
return NULL;
}
}
@ -245,7 +258,8 @@ void streamFreeQitem(SStreamQueueItem* data) {
SStreamRefDataBlock* pRefBlock = (SStreamRefDataBlock*)data;
blockDataDestroy(pRefBlock->pBlock);
taosFreeQitem(pRefBlock);
} else if (type == STREAM_INPUT__CHECKPOINT || type == STREAM_INPUT__CHECKPOINT_TRIGGER || type == STREAM_INPUT__TRANS_STATE) {
} else if (type == STREAM_INPUT__CHECKPOINT || type == STREAM_INPUT__CHECKPOINT_TRIGGER ||
type == STREAM_INPUT__TRANS_STATE) {
SStreamDataBlock* pBlock = (SStreamDataBlock*)data;
taosArrayDestroyEx(pBlock->blocks, freeItems);
taosFreeQitem(pBlock);

38
source/libs/stream/src/streamQueue.c
View File

@ -42,18 +42,26 @@ static void streamQueueCleanup(SStreamQueue* pQueue) {
static void* streamQueueCurItem(SStreamQueue* queue) { return queue->qItem; }
SStreamQueue* streamQueueOpen(int64_t cap) {
int32_t code;
SStreamQueue* pQueue = taosMemoryCalloc(1, sizeof(SStreamQueue));
if (pQueue == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
pQueue->pQueue = taosOpenQueue();
pQueue->qall = taosAllocateQall();
code = taosOpenQueue(&pQueue->pQueue);
if (code) {
taosMemoryFreeClear(pQueue);
terrno = code;
return NULL;
}
if (pQueue->pQueue == NULL || pQueue->qall == NULL) {
if (pQueue->pQueue) taosCloseQueue(pQueue->pQueue);
if (pQueue->qall) taosFreeQall(pQueue->qall);
code = taosAllocateQall(&pQueue->qall);
if (code) {
taosCloseQueue(pQueue->pQueue);
taosMemoryFree(pQueue);
terrno = code;
return NULL;
}
@ -64,7 +72,8 @@ SStreamQueue* streamQueueOpen(int64_t cap) {
}
void streamQueueClose(SStreamQueue* pQueue, int32_t taskId) {
stDebug("s-task:0x%x free the queue:%p, items in queue:%d", taskId, pQueue->pQueue, taosQueueItemSize(pQueue->pQueue));
stDebug("s-task:0x%x free the queue:%p, items in queue:%d", taskId, pQueue->pQueue,
taosQueueItemSize(pQueue->pQueue));
streamQueueCleanup(pQueue);
taosFreeQall(pQueue->qall);
@ -174,7 +183,6 @@ EExtractDataCode streamTaskGetDataFromInputQ(SStreamTask* pTask, SStreamQueueIte
SStreamQueueItem* qItem = streamQueueNextItem(pTask->inputq.queue);
if (qItem == NULL) {
// restore the token to bucket
if (*numOfBlocks > 0) {
*blockSize = streamQueueItemGetSize(*pInput);
@ -332,16 +340,20 @@ int32_t streamTaskPutDataIntoInputQ(SStreamTask* pTask, SStreamQueueItem* pItem)
if (type != STREAM_INPUT__GET_RES && type != STREAM_INPUT__CHECKPOINT && type != STREAM_INPUT__CHECKPOINT_TRIGGER &&
(pTask->info.delaySchedParam != 0)) {
atomic_val_compare_exchange_8(&pTask->schedInfo.status, TASK_TRIGGER_STATUS__INACTIVE, TASK_TRIGGER_STATUS__ACTIVE);
stDebug("s-task:%s new data arrived, active the sched-trigger, triggerStatus:%d", pTask->id.idStr, pTask->schedInfo.status);
stDebug("s-task:%s new data arrived, active the sched-trigger, triggerStatus:%d", pTask->id.idStr,
pTask->schedInfo.status);
}
return 0;
}
int32_t streamTaskPutTranstateIntoInputQ(SStreamTask* pTask) {
SStreamDataBlock* pTranstate = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, sizeof(SSDataBlock));
if (pTranstate == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
int32_t code;
SStreamDataBlock* pTranstate;
code = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, sizeof(SSDataBlock), (void**)&pTranstate);
if (code) {
return code;
}
SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
@ -460,8 +472,6 @@ void streamTaskPutbackToken(STokenBucket* pBucket) {
}
// size in KB
void streamTaskConsumeQuota(STokenBucket* pBucket, int32_t bytes) {
pBucket->quotaRemain -= SIZE_IN_MiB(bytes);
}
void streamTaskConsumeQuota(STokenBucket* pBucket, int32_t bytes) { pBucket->quotaRemain -= SIZE_IN_MiB(bytes); }
void streamTaskInputFail(SStreamTask* pTask) { atomic_store_8(&pTask->inputq.status, TASK_INPUT_STATUS__FAILED); }

9
source/libs/stream/src/streamSched.c
View File

@ -134,11 +134,14 @@ void streamTaskSchedHelper(void* param, void* tmrId) {
stDebug("s-task:%s in checkpoint procedure, not retrieve result, next:%dms", id, nextTrigger);
} else {
if (status == TASK_TRIGGER_STATUS__ACTIVE) {
SStreamTrigger* pTrigger = taosAllocateQitem(sizeof(SStreamTrigger), DEF_QITEM, 0);
if (pTrigger == NULL) {
SStreamTrigger* pTrigger;
int32_t code = taosAllocateQitem(sizeof(SStreamTrigger), DEF_QITEM, 0, (void**)&pTrigger);
if (code) {
stError("s-task:%s failed to prepare retrieve data trigger, code:%s, try again in %dms", id, "out of memory",
nextTrigger);
taosTmrReset(streamTaskSchedHelper, nextTrigger, pTask, streamTimer, &pTask->schedInfo.pDelayTimer);
terrno = code;
return;
}
@ -156,7 +159,7 @@ void streamTaskSchedHelper(void* param, void* tmrId) {
atomic_store_8(&pTask->schedInfo.status, TASK_TRIGGER_STATUS__INACTIVE);
pTrigger->pBlock->info.type = STREAM_GET_ALL;
int32_t code = streamTaskPutDataIntoInputQ(pTask, (SStreamQueueItem*)pTrigger);
code = streamTaskPutDataIntoInputQ(pTask, (SStreamQueueItem*)pTrigger);
if (code != TSDB_CODE_SUCCESS) {
taosTmrReset(streamTaskSchedHelper, nextTrigger, pTask, streamTimer, &pTask->schedInfo.pDelayTimer);
return;

11
source/libs/stream/src/streamTask.c
View File

@ -980,10 +980,13 @@ char* createStreamTaskIdStr(int64_t streamId, int32_t taskId) {
}
static int32_t streamTaskEnqueueRetrieve(SStreamTask* pTask, SStreamRetrieveReq* pReq) {
SStreamDataBlock* pData = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, sizeof(SStreamDataBlock));
if (pData == NULL) {
int32_t code;
SStreamDataBlock* pData;
code = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, sizeof(SStreamDataBlock), (void**)&pData);
if (code) {
stError("s-task:%s failed to allocated retrieve-block", pTask->id.idStr);
return terrno;
return terrno = code;
}
// enqueue
@ -993,7 +996,7 @@ static int32_t streamTaskEnqueueRetrieve(SStreamTask* pTask, SStreamRetrieveReq*
pData->type = STREAM_INPUT__DATA_RETRIEVE;
pData->srcVgId = 0;
int32_t code = streamRetrieveReqToData(pReq, pData, pTask->id.idStr);
code = streamRetrieveReqToData(pReq, pData, pTask->id.idStr);
if (code != TSDB_CODE_SUCCESS) {
taosFreeQitem(pData);
return code;

8
source/libs/transport/test/svrBench.c
View File

@ -67,7 +67,7 @@ void *processShellMsg(void *arg) {
int type;
SQueueInfo qinfo = {0};
qall = taosAllocateQall();
taosAllocateQall(&qall);
while (1) {
int numOfMsgs = taosReadAllQitemsFromQset(multiQ->qset[idx], qall, &qinfo);
@ -129,7 +129,7 @@ void *processShellMsg(void *arg) {
void processRequestMsg(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet) {
SRpcMsg *pTemp;
pTemp = taosAllocateQitem(sizeof(SRpcMsg), DEF_QITEM, 0);
taosAllocateQitem(sizeof(SRpcMsg), DEF_QITEM, 0, (void **)&pTemp);
memcpy(pTemp, pMsg, sizeof(SRpcMsg));
int32_t idx = balance % multiQ->numOfThread;
@ -212,8 +212,8 @@ int main(int argc, char *argv[]) {
multiQ->qset = (STaosQset **)taosMemoryMalloc(sizeof(STaosQset *) * numOfAthread);
for (int i = 0; i < numOfAthread; i++) {
multiQ->qhandle[i] = taosOpenQueue();
multiQ->qset[i] = taosOpenQset();
taosOpenQueue(&multiQ->qhandle[i]);
taosOpenQset(&multiQ->qset[i]);
taosAddIntoQset(multiQ->qset[i], multiQ->qhandle[i], NULL);
}
TThread *threads = taosMemoryMalloc(sizeof(TThread) * numOfAthread);

4
source/util/src/thash.c
View File

@ -675,7 +675,11 @@ void taosHashTableResize(SHashObj *pHashObj) {
}
SHashNode *doCreateHashNode(const void *key, size_t keyLen, const void *pData, size_t dsize, uint32_t hashVal) {
#ifdef _TD_LOONGARCH_64
SHashNode *pNewNode = taosMemoryCalloc(1, sizeof(SHashNode) + keyLen + dsize + 1);
#else
SHashNode *pNewNode = taosMemoryMalloc(sizeof(SHashNode) + keyLen + dsize + 1);
#endif
if (pNewNode == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;

81
source/util/src/tqueue.c
View File

@ -17,6 +17,7 @@
#include "tqueue.h"
#include "taoserror.h"
#include "tlog.h"
#include "tutil.h"
int64_t tsRpcQueueMemoryAllowed = 0;
int64_t tsRpcQueueMemoryUsed = 0;
@ -58,20 +59,20 @@ struct STaosQall {
void taosSetQueueMemoryCapacity(STaosQueue *queue, int64_t cap) { queue->memLimit = cap; }
void taosSetQueueCapacity(STaosQueue *queue, int64_t size) { queue->itemLimit = size; }
STaosQueue *taosOpenQueue() {
STaosQueue *queue = taosMemoryCalloc(1, sizeof(STaosQueue));
if (queue == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
int32_t taosOpenQueue(STaosQueue **queue) {
*queue = taosMemoryCalloc(1, sizeof(STaosQueue));
if (*queue == NULL) {
return (terrno = TSDB_CODE_OUT_OF_MEMORY);
}
if (taosThreadMutexInit(&queue->mutex, NULL) != 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
int32_t code = taosThreadMutexInit(&(*queue)->mutex, NULL);
if (code) {
taosMemoryFreeClear(*queue);
return (terrno = TAOS_SYSTEM_ERROR(code));
}
uDebug("queue:%p is opened", queue);
return queue;
return 0;
}
void taosSetQueueFp(STaosQueue *queue, FItem itemFp, FItems itemsFp) {
@ -146,11 +147,12 @@ int64_t taosQueueMemorySize(STaosQueue *queue) {
return memOfItems;
}
void *taosAllocateQitem(int32_t size, EQItype itype, int64_t dataSize) {
int32_t taosAllocateQitem(int32_t size, EQItype itype, int64_t dataSize, void **item) {
*item = NULL;
STaosQnode *pNode = taosMemoryCalloc(1, sizeof(STaosQnode) + size);
if (pNode == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
return terrno = TSDB_CODE_OUT_OF_MEMORY;
}
pNode->dataSize = dataSize;
@ -165,15 +167,15 @@ void *taosAllocateQitem(int32_t size, EQItype itype, int64_t dataSize) {
tsRpcQueueMemoryAllowed);
atomic_sub_fetch_64(&tsRpcQueueMemoryUsed, size + dataSize);
taosMemoryFree(pNode);
terrno = TSDB_CODE_OUT_OF_RPC_MEMORY_QUEUE;
return NULL;
return (terrno = TSDB_CODE_OUT_OF_RPC_MEMORY_QUEUE);
}
uTrace("item:%p, node:%p is allocated, alloc:%" PRId64, pNode->item, pNode, alloced);
} else {
uTrace("item:%p, node:%p is allocated", pNode->item, pNode);
}
return pNode->item;
*item = pNode->item;
return 0;
}
void taosFreeQitem(void *pItem) {
@ -221,13 +223,17 @@ int32_t taosWriteQitem(STaosQueue *queue, void *pItem) {
}
queue->numOfItems++;
queue->memOfItems += (pNode->size + pNode->dataSize);
if (queue->qset) atomic_add_fetch_32(&queue->qset->numOfItems, 1);
if (queue->qset) {
atomic_add_fetch_32(&queue->qset->numOfItems, 1);
}
uTrace("item:%p is put into queue:%p, items:%d mem:%" PRId64, pItem, queue, queue->numOfItems, queue->memOfItems);
taosThreadMutexUnlock(&queue->mutex);
if (queue->qset) tsem_post(&queue->qset->sem);
if (queue->qset) {
tsem_post(&queue->qset->sem);
}
return code;
}
@ -241,10 +247,14 @@ int32_t taosReadQitem(STaosQueue *queue, void **ppItem) {
pNode = queue->head;
*ppItem = pNode->item;
queue->head = pNode->next;
if (queue->head == NULL) queue->tail = NULL;
if (queue->head == NULL) {
queue->tail = NULL;
}
queue->numOfItems--;
queue->memOfItems -= (pNode->size + pNode->dataSize);
if (queue->qset) atomic_sub_fetch_32(&queue->qset->numOfItems, 1);
if (queue->qset) {
atomic_sub_fetch_32(&queue->qset->numOfItems, 1);
}
code = 1;
uTrace("item:%p is read out from queue:%p, items:%d mem:%" PRId64, *ppItem, queue, queue->numOfItems,
queue->memOfItems);
@ -255,12 +265,12 @@ int32_t taosReadQitem(STaosQueue *queue, void **ppItem) {
return code;
}
STaosQall *taosAllocateQall() {
STaosQall *qall = taosMemoryCalloc(1, sizeof(STaosQall));
if (qall != NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
int32_t taosAllocateQall(STaosQall **qall) {
*qall = taosMemoryCalloc(1, sizeof(STaosQall));
if (*qall == NULL) {
return terrno = TSDB_CODE_OUT_OF_MEMORY;
}
return qall;
return 0;
}
void taosFreeQall(STaosQall *qall) { taosMemoryFree(qall); }
@ -290,7 +300,9 @@ int32_t taosReadAllQitems(STaosQueue *queue, STaosQall *qall) {
queue->memOfItems = 0;
uTrace("read %d items from queue:%p, items:%d mem:%" PRId64, numOfItems, queue, queue->numOfItems,
queue->memOfItems);
if (queue->qset) atomic_sub_fetch_32(&queue->qset->numOfItems, qall->numOfItems);
if (queue->qset) {
atomic_sub_fetch_32(&queue->qset->numOfItems, qall->numOfItems);
}
}
taosThreadMutexUnlock(&queue->mutex);
@ -326,18 +338,17 @@ int32_t taosGetQitem(STaosQall *qall, void **ppItem) {
return num;
}
STaosQset *taosOpenQset() {
STaosQset *qset = taosMemoryCalloc(sizeof(STaosQset), 1);
if (qset == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
int32_t taosOpenQset(STaosQset **qset) {
*qset = taosMemoryCalloc(sizeof(STaosQset), 1);
if (*qset == NULL) {
return terrno = TSDB_CODE_OUT_OF_MEMORY;
}
taosThreadMutexInit(&qset->mutex, NULL);
tsem_init(&qset->sem, 0, 0);
taosThreadMutexInit(&(*qset)->mutex, NULL);
tsem_init(&(*qset)->sem, 0, 0);
uDebug("qset:%p is opened", qset);
return qset;
return 0;
}
void taosCloseQset(STaosQset *qset) {
@ -533,9 +544,7 @@ int64_t taosQallUnAccessedMemSize(STaosQall *qall) { return qall->unAccessMemOfI
void taosResetQitems(STaosQall *qall) { qall->current = qall->start; }
int32_t taosGetQueueNumber(STaosQset *qset) { return qset->numOfQueues; }
void taosQueueSetThreadId(STaosQueue* pQueue, int64_t threadId) {
pQueue->threadId = threadId;
}
void taosQueueSetThreadId(STaosQueue *pQueue, int64_t threadId) { pQueue->threadId = threadId; }
int64_t taosQueueGetThreadId(STaosQueue *pQueue) { return pQueue->threadId; }

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

@ -16,20 +16,22 @@
#define _DEFAULT_SOURCE
#include "tworker.h"
#include "taoserror.h"
#include "tcompare.h"
#include "tgeosctx.h"
#include "tlog.h"
#include "tcompare.h"
#define QUEUE_THRESHOLD (1000 * 1000)
typedef void *(*ThreadFp)(void *param);
int32_t tQWorkerInit(SQWorkerPool *pool) {
pool->qset = taosOpenQset();
int32_t code = taosOpenQset(&pool->qset);
if (code) return code;
pool->workers = taosMemoryCalloc(pool->max, sizeof(SQueueWorker));
if (pool->workers == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
taosCloseQset(pool->qset);
return terrno = TSDB_CODE_OUT_OF_MEMORY;
}
(void)taosThreadMutexInit(&pool->mutex, NULL);
@ -109,8 +111,14 @@ static void *tQWorkerThreadFp(SQueueWorker *worker) {
}
STaosQueue *tQWorkerAllocQueue(SQWorkerPool *pool, void *ahandle, FItem fp) {
STaosQueue *queue = taosOpenQueue();
if (queue == NULL) return NULL;
int32_t code;
STaosQueue *queue;
code = taosOpenQueue(&queue);
if (code) {
terrno = code;
return NULL;
}
taosThreadMutexLock(&pool->mutex);
taosSetQueueFp(queue, fp, NULL);
@ -150,11 +158,17 @@ void tQWorkerFreeQueue(SQWorkerPool *pool, STaosQueue *queue) {
}
int32_t tAutoQWorkerInit(SAutoQWorkerPool *pool) {
pool->qset = taosOpenQset();
int32_t code;
code = taosOpenQset(&pool->qset);
if (code) {
return terrno = code;
}
pool->workers = taosArrayInit(2, sizeof(SQueueWorker *));
if (pool->workers == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
taosCloseQset(pool->qset);
return terrno = TSDB_CODE_OUT_OF_MEMORY;
}
(void)taosThreadMutexInit(&pool->mutex, NULL);
@ -228,8 +242,14 @@ static void *tAutoQWorkerThreadFp(SQueueWorker *worker) {
}
STaosQueue *tAutoQWorkerAllocQueue(SAutoQWorkerPool *pool, void *ahandle, FItem fp) {
STaosQueue *queue = taosOpenQueue();
if (queue == NULL) return NULL;
int32_t code;
STaosQueue *queue;
code = taosOpenQueue(&queue);
if (code) {
terrno = code;
return NULL;
}
taosThreadMutexLock(&pool->mutex);
taosSetQueueFp(queue, fp, NULL);
@ -377,18 +397,19 @@ STaosQueue *tWWorkerAllocQueue(SWWorkerPool *pool, void *ahandle, FItems fp) {
taosThreadMutexLock(&pool->mutex);
SWWorker *worker = pool->workers + pool->nextId;
int32_t code = -1;
STaosQueue *queue;
STaosQueue *queue = taosOpenQueue();
if (queue == NULL) goto _OVER;
code = taosOpenQueue(&queue);
if (code) goto _OVER;
taosSetQueueFp(queue, NULL, fp);
if (worker->qset == NULL) {
worker->qset = taosOpenQset();
if (worker->qset == NULL) goto _OVER;
code = taosOpenQset(&worker->qset);
if (code) goto _OVER;
taosAddIntoQset(worker->qset, queue, ahandle);
worker->qall = taosAllocateQall();
if (worker->qall == NULL) goto _OVER;
code = taosAllocateQall(&worker->qall);
if (code) goto _OVER;
TdThreadAttr thAttr;
taosThreadAttrInit(&thAttr);
@ -406,12 +427,10 @@ STaosQueue *tWWorkerAllocQueue(SWWorkerPool *pool, void *ahandle, FItems fp) {
pool->nextId = (pool->nextId + 1) % pool->max;
}
code = 0;
_OVER:
taosThreadMutexUnlock(&pool->mutex);
if (code == -1) {
if (code) {
if (queue != NULL) taosCloseQueue(queue);
if (worker->qset != NULL) taosCloseQset(worker->qset);
if (worker->qall != NULL) taosFreeQall(worker->qall);
@ -758,8 +777,9 @@ bool tQueryAutoQWorkerTryRecycleWorker(SQueryAutoQWorkerPool* pPool, SQueryAutoQ
}
int32_t tQueryAutoQWorkerInit(SQueryAutoQWorkerPool *pool) {
pool->qset = taosOpenQset();
if (!pool->qset) return terrno;
int32_t code;
code = taosOpenQset(&pool->qset);
if (code) return terrno = code;
pool->workers = tdListNew(sizeof(SQueryAutoQWorker));
if (!pool->workers) return TSDB_CODE_OUT_OF_MEMORY;
pool->backupWorkers = tdListNew(sizeof(SQueryAutoQWorker));
@ -867,8 +887,12 @@ void tQueryAutoQWorkerCleanup(SQueryAutoQWorkerPool *pPool) {
}
STaosQueue *tQueryAutoQWorkerAllocQueue(SQueryAutoQWorkerPool *pool, void *ahandle, FItem fp) {
STaosQueue *queue = taosOpenQueue();
if (queue == NULL) return NULL;
STaosQueue *queue;
int32_t code = taosOpenQueue(&queue);
if (code) {
terrno = code;
return NULL;
}
taosThreadMutexLock(&pool->poolLock);
taosSetQueueFp(queue, fp, NULL);
@ -915,9 +939,7 @@ STaosQueue *tQueryAutoQWorkerAllocQueue(SQueryAutoQWorkerPool *pool, void *ahand
return queue;
}
void tQueryAutoQWorkerFreeQueue(SQueryAutoQWorkerPool *pPool, STaosQueue *pQ) {
taosCloseQueue(pQ);
}
void tQueryAutoQWorkerFreeQueue(SQueryAutoQWorkerPool *pPool, STaosQueue *pQ) { taosCloseQueue(pQ); }
static int32_t tQueryAutoQWorkerAddWorker(SQueryAutoQWorkerPool *pool) {
// try backup pool