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 fix/hzcheng_3.0

This commit is contained in:
Hongze Cheng 2022-05-20 02:24:27 +00:00
parent 9d0116ea0a 9ca54b9478
commit f7cc44d616
29 changed files with 1621 additions and 382 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
include/libs/stream/tstream.h
View File

@ -107,6 +107,7 @@ static FORCE_INLINE void streamDataSubmitRefDec(SStreamDataSubmit* pDataSubmit)
if (ref == 0) { if (ref == 0) {
taosMemoryFree(pDataSubmit->data); taosMemoryFree(pDataSubmit->data);
taosMemoryFree(pDataSubmit->dataRef); taosMemoryFree(pDataSubmit->dataRef);
taosFreeQitem(pDataSubmit);
} }
} }
@ -279,6 +280,12 @@ typedef struct {
SArray* res; // SArray<SSDataBlock> SArray* res; // SArray<SSDataBlock>
} SStreamSinkReq; } SStreamSinkReq;
typedef struct {
SMsgHead head;
int64_t streamId;
int32_t taskId;
} SStreamTaskRunReq;
int32_t streamEnqueueDataSubmit(SStreamTask* pTask, SStreamDataSubmit* input); int32_t streamEnqueueDataSubmit(SStreamTask* pTask, SStreamDataSubmit* input);
int32_t streamEnqueueDataBlk(SStreamTask* pTask, SStreamDataBlock* input); int32_t streamEnqueueDataBlk(SStreamTask* pTask, SStreamDataBlock* input);
int32_t streamDequeueOutput(SStreamTask* pTask, void** output); int32_t streamDequeueOutput(SStreamTask* pTask, void** output);

14
include/libs/transport/trpc.h
View File

@ -63,11 +63,6 @@ typedef struct SRpcMsg {
} SRpcMsg; } SRpcMsg;
typedef void (*RpcCfp)(void *parent, SRpcMsg *, SEpSet *rf); typedef void (*RpcCfp)(void *parent, SRpcMsg *, SEpSet *rf);
typedef int (*RpcAfp)(void *parent, char *tableId, char *spi, char *encrypt, char *secret, char *ckey);
///
// // SRpcMsg code
// REDIERE,
// NOT READY, EpSet
typedef bool (*RpcRfp)(int32_t code); typedef bool (*RpcRfp)(int32_t code);
typedef struct SRpcInit { typedef struct SRpcInit {
@ -80,18 +75,11 @@ typedef struct SRpcInit {
int idleTime; // milliseconds, 0 means idle timer is disabled int idleTime; // milliseconds, 0 means idle timer is disabled
// the following is for client app ecurity only // the following is for client app ecurity only
char *user; // user name char *user; // user name
char spi; // security parameter index
char encrypt; // encrypt algorithm
char *secret; // key for authentication
char *ckey; // ciphering key
// call back to process incoming msg, code shall be ignored by server app // call back to process incoming msg, code shall be ignored by server app
RpcCfp cfp; RpcCfp cfp;
// call back to retrieve the client auth info, for server app only
RpcAfp afp;
// user defined retry func // user defined retry func
RpcRfp rfp; RpcRfp rfp;

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

@ -60,7 +60,7 @@ static void registerRequest(SRequestObj *pRequest) {
static void deregisterRequest(SRequestObj *pRequest) { static void deregisterRequest(SRequestObj *pRequest) {
assert(pRequest != NULL); assert(pRequest != NULL);
STscObj *pTscObj = pRequest->pTscObj; STscObj * pTscObj = pRequest->pTscObj;
SInstanceSummary *pActivity = &pTscObj->pAppInfo->summary; SInstanceSummary *pActivity = &pTscObj->pAppInfo->summary;
int32_t currentInst = atomic_sub_fetch_64((int64_t *)&pActivity->currentRequests, 1); int32_t currentInst = atomic_sub_fetch_64((int64_t *)&pActivity->currentRequests, 1);
@ -91,7 +91,6 @@ static bool clientRpcRfp(int32_t code) {
} }
} }
// TODO refactor // TODO refactor
void *openTransporter(const char *user, const char *auth, int32_t numOfThread) { void *openTransporter(const char *user, const char *auth, int32_t numOfThread) {
SRpcInit rpcInit; SRpcInit rpcInit;
@ -105,10 +104,6 @@ void *openTransporter(const char *user, const char *auth, int32_t numOfThread) {
rpcInit.connType = TAOS_CONN_CLIENT; rpcInit.connType = TAOS_CONN_CLIENT;
rpcInit.user = (char *)user; rpcInit.user = (char *)user;
rpcInit.idleTime = tsShellActivityTimer * 1000; rpcInit.idleTime = tsShellActivityTimer * 1000;
rpcInit.ckey = "key";
rpcInit.spi = 1;
rpcInit.secret = (char *)auth;
void *pDnodeConn = rpcOpen(&rpcInit); void *pDnodeConn = rpcOpen(&rpcInit);
if (pDnodeConn == NULL) { if (pDnodeConn == NULL) {
tscError("failed to init connection to server"); tscError("failed to init connection to server");
@ -318,7 +313,7 @@ int taos_options_imp(TSDB_OPTION option, const char *str) {
return 0; return 0;
} }
SConfig *pCfg = taosGetCfg(); SConfig * pCfg = taosGetCfg();
SConfigItem *pItem = NULL; SConfigItem *pItem = NULL;
switch (option) { switch (option) {

18
source/client/src/clientImpl.c
View File

@ -291,7 +291,7 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
SQueryResult res = {.code = 0, .numOfRows = 0, .msgSize = ERROR_MSG_BUF_DEFAULT_SIZE, .msg = pRequest->msgBuf}; SQueryResult res = {.code = 0, .numOfRows = 0, .msgSize = ERROR_MSG_BUF_DEFAULT_SIZE, .msg = pRequest->msgBuf};
int32_t code = schedulerExecJob(pTransporter, pNodeList, pDag, &pRequest->body.queryJob, pRequest->sqlstr, int32_t code = schedulerExecJob(pTransporter, pNodeList, pDag, &pRequest->body.queryJob, pRequest->sqlstr,
pRequest->metric.start, &res); pRequest->metric.start, &res);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
if (pRequest->body.queryJob != 0) { if (pRequest->body.queryJob != 0) {
schedulerFreeJob(pRequest->body.queryJob); schedulerFreeJob(pRequest->body.queryJob);
@ -325,7 +325,7 @@ int32_t getQueryPlan(SRequestObj* pRequest, SQuery* pQuery, SArray** pNodeList)
int32_t validateSversion(SRequestObj* pRequest, void* res) { int32_t validateSversion(SRequestObj* pRequest, void* res) {
SArray* pArray = NULL; SArray* pArray = NULL;
int32_t code = 0; int32_t code = 0;
if (TDMT_VND_SUBMIT == pRequest->type) { if (TDMT_VND_SUBMIT == pRequest->type) {
SSubmitRsp* pRsp = (SSubmitRsp*)res; SSubmitRsp* pRsp = (SSubmitRsp*)res;
if (pRsp->nBlocks <= 0) { if (pRsp->nBlocks <= 0) {
@ -337,14 +337,13 @@ int32_t validateSversion(SRequestObj* pRequest, void* res) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
} }
for (int32_t i = 0; i < pRsp->nBlocks; ++i) { for (int32_t i = 0; i < pRsp->nBlocks; ++i) {
SSubmitBlkRsp *blk = pRsp->pBlocks + i; SSubmitBlkRsp* blk = pRsp->pBlocks + i;
STbSVersion tbSver = {.tbFName = blk->tblFName, .sver = blk->sver}; STbSVersion tbSver = {.tbFName = blk->tblFName, .sver = blk->sver};
taosArrayPush(pArray, &tbSver); taosArrayPush(pArray, &tbSver);
} }
} else if (TDMT_VND_QUERY == pRequest->type) { } else if (TDMT_VND_QUERY == pRequest->type) {
} }
SCatalog* pCatalog = NULL; SCatalog* pCatalog = NULL;
@ -365,11 +364,10 @@ void freeRequestRes(SRequestObj* pRequest, void* res) {
if (NULL == res) { if (NULL == res) {
return; return;
} }
if (TDMT_VND_SUBMIT == pRequest->type) { if (TDMT_VND_SUBMIT == pRequest->type) {
tFreeSSubmitRsp((SSubmitRsp*)res); tFreeSSubmitRsp((SSubmitRsp*)res);
} else if (TDMT_VND_QUERY == pRequest->type) { } else if (TDMT_VND_QUERY == pRequest->type) {
} }
} }
@ -1022,7 +1020,6 @@ TSDB_SERVER_STATUS taos_check_server_status(const char* fqdn, int port, char* de
SRpcInit rpcInit = {0}; SRpcInit rpcInit = {0};
char pass[TSDB_PASSWORD_LEN + 1] = {0}; char pass[TSDB_PASSWORD_LEN + 1] = {0};
taosEncryptPass_c((uint8_t*)("_pwd"), strlen("_pwd"), pass);
rpcInit.label = "CHK"; rpcInit.label = "CHK";
rpcInit.numOfThreads = 1; rpcInit.numOfThreads = 1;
rpcInit.cfp = NULL; rpcInit.cfp = NULL;
@ -1030,9 +1027,6 @@ TSDB_SERVER_STATUS taos_check_server_status(const char* fqdn, int port, char* de
rpcInit.connType = TAOS_CONN_CLIENT; rpcInit.connType = TAOS_CONN_CLIENT;
rpcInit.idleTime = tsShellActivityTimer * 1000; rpcInit.idleTime = tsShellActivityTimer * 1000;
rpcInit.user = "_dnd"; rpcInit.user = "_dnd";
rpcInit.ckey = "_key";
rpcInit.spi = 1;
rpcInit.secret = pass;
clientRpc = rpcOpen(&rpcInit); clientRpc = rpcOpen(&rpcInit);
if (clientRpc == NULL) { if (clientRpc == NULL) {

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

@ -49,9 +49,9 @@ int32_t dmProcessNodeMsg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg) {
} }
static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) { static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) {
SDnodeTrans *pTrans = &pDnode->trans; SDnodeTrans * pTrans = &pDnode->trans;
int32_t code = -1; int32_t code = -1;
SRpcMsg *pMsg = NULL; SRpcMsg * pMsg = NULL;
bool needRelease = false; bool needRelease = false;
SDnodeHandle *pHandle = &pTrans->msgHandles[TMSG_INDEX(pRpc->msgType)]; SDnodeHandle *pHandle = &pTrans->msgHandles[TMSG_INDEX(pRpc->msgType)];
SMgmtWrapper *pWrapper = NULL; SMgmtWrapper *pWrapper = NULL;
@ -179,11 +179,11 @@ int32_t dmInitMsgHandle(SDnode *pDnode) {
for (EDndNodeType ntype = DNODE; ntype < NODE_END; ++ntype) { for (EDndNodeType ntype = DNODE; ntype < NODE_END; ++ntype) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype]; SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];
SArray *pArray = (*pWrapper->func.getHandlesFp)(); SArray * pArray = (*pWrapper->func.getHandlesFp)();
if (pArray == NULL) return -1; if (pArray == NULL) return -1;
for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) { for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
SMgmtHandle *pMgmt = taosArrayGet(pArray, i); SMgmtHandle * pMgmt = taosArrayGet(pArray, i);
SDnodeHandle *pHandle = &pTrans->msgHandles[TMSG_INDEX(pMgmt->msgType)]; SDnodeHandle *pHandle = &pTrans->msgHandles[TMSG_INDEX(pMgmt->msgType)];
if (pMgmt->needCheckVgId) { if (pMgmt->needCheckVgId) {
pHandle->needCheckVgId = pMgmt->needCheckVgId; pHandle->needCheckVgId = pMgmt->needCheckVgId;
@ -276,15 +276,9 @@ int32_t dmInitClient(SDnode *pDnode) {
rpcInit.connType = TAOS_CONN_CLIENT; rpcInit.connType = TAOS_CONN_CLIENT;
rpcInit.idleTime = tsShellActivityTimer * 1000; rpcInit.idleTime = tsShellActivityTimer * 1000;
rpcInit.user = INTERNAL_USER; rpcInit.user = INTERNAL_USER;
rpcInit.ckey = INTERNAL_CKEY;
rpcInit.spi = 1;
rpcInit.parent = pDnode; rpcInit.parent = pDnode;
rpcInit.rfp = rpcRfp; rpcInit.rfp = rpcRfp;
char pass[TSDB_PASSWORD_LEN + 1] = {0};
taosEncryptPass_c((uint8_t *)(INTERNAL_SECRET), strlen(INTERNAL_SECRET), pass);
rpcInit.secret = pass;
pTrans->clientRpc = rpcOpen(&rpcInit); pTrans->clientRpc = rpcOpen(&rpcInit);
if (pTrans->clientRpc == NULL) { if (pTrans->clientRpc == NULL) {
dError("failed to init dnode rpc client"); dError("failed to init dnode rpc client");

6
source/dnode/mgmt/test/sut/src/client.cpp
View File

@ -48,10 +48,10 @@ void TestClient::DoInit() {
rpcInit.connType = TAOS_CONN_CLIENT; rpcInit.connType = TAOS_CONN_CLIENT;
rpcInit.idleTime = 30 * 1000; rpcInit.idleTime = 30 * 1000;
rpcInit.user = (char*)this->user; rpcInit.user = (char*)this->user;
rpcInit.ckey = (char*)"key"; // rpcInit.ckey = (char*)"key";
rpcInit.parent = this; rpcInit.parent = this;
rpcInit.secret = (char*)secretEncrypt; // rpcInit.secret = (char*)secretEncrypt;
rpcInit.spi = 1; // rpcInit.spi = 1;
clientRpc = rpcOpen(&rpcInit); clientRpc = rpcOpen(&rpcInit);
ASSERT(clientRpc); ASSERT(clientRpc);

64
source/dnode/vnode/src/tq/tq.c
View File

@ -112,19 +112,18 @@ int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList, bool isAdd) {
if (pIter == NULL) break; if (pIter == NULL) break;
pExec = (STqExec*)pIter; pExec = (STqExec*)pIter;
if (pExec->subType == TOPIC_SUB_TYPE__DB) { if (pExec->subType == TOPIC_SUB_TYPE__DB) {
if (isAdd) { if (!isAdd) {
continue;
} else {
int32_t sz = taosArrayGetSize(tbUidList); int32_t sz = taosArrayGetSize(tbUidList);
for (int32_t i = 0; i < sz; i++) { for (int32_t i = 0; i < sz; i++) {
int64_t tbUid = *(int64_t*)taosArrayGet(tbUidList, i); int64_t tbUid = *(int64_t*)taosArrayGet(tbUidList, i);
taosHashPut(pExec->pDropTbUid, &tbUid, sizeof(int64_t), NULL, 0); taosHashPut(pExec->pDropTbUid, &tbUid, sizeof(int64_t), NULL, 0);
} }
} }
} } else {
for (int32_t i = 0; i < 5; i++) { for (int32_t i = 0; i < 5; i++) {
int32_t code = qUpdateQualifiedTableId(pExec->task[i], tbUidList, isAdd); int32_t code = qUpdateQualifiedTableId(pExec->task[i], tbUidList, isAdd);
ASSERT(code == 0); ASSERT(code == 0);
}
} }
} }
return 0; return 0;
@ -1059,6 +1058,57 @@ int32_t tqProcessStreamTrigger(STQ* pTq, void* data, int32_t dataLen, int32_t wo
return 0; return 0;
} }
int32_t tqProcessStreamTriggerNew(STQ* pTq, SSubmitReq* data) {
SStreamDataSubmit* pSubmit = NULL;
// build data
pSubmit = taosAllocateQitem(sizeof(SStreamDataSubmit), DEF_QITEM);
if (pSubmit == NULL) return -1;
pSubmit->dataRef = taosMemoryMalloc(sizeof(int32_t));
if (pSubmit->dataRef == NULL) goto FAIL;
*pSubmit->dataRef = 1;
pSubmit->data = data;
pSubmit->type = STREAM_INPUT__DATA_BLOCK;
void* pIter = NULL;
while (1) {
pIter = taosHashIterate(pTq->pStreamTasks, pIter);
if (pIter == NULL) break;
SStreamTask* pTask = (SStreamTask*)pIter;
if (pTask->inputType == TASK_INPUT_TYPE__SUMBIT_BLOCK) {
streamEnqueueDataSubmit(pTask, pSubmit);
// TODO cal back pressure
}
// check run
int8_t execStatus = atomic_load_8(&pTask->status);
if (execStatus == TASK_STATUS__IDLE || execStatus == TASK_STATUS__CLOSING) {
SStreamTaskRunReq* pReq = taosMemoryMalloc(sizeof(SStreamTaskRunReq));
if (pReq == NULL) continue;
// TODO: do we need htonl?
pReq->head.vgId = pTq->pVnode->config.vgId;
pReq->streamId = pTask->streamId;
pReq->taskId = pTask->taskId;
SRpcMsg msg = {
.msgType = 0,
.pCont = pReq,
.contLen = sizeof(SStreamTaskRunReq),
};
tmsgPutToQueue(&pTq->pVnode->msgCb, FETCH_QUEUE, &msg);
}
}
streamDataSubmitRefDec(pSubmit);
return 0;
FAIL:
if (pSubmit) {
if (pSubmit->dataRef) {
taosMemoryFree(pSubmit->dataRef);
}
taosFreeQitem(pSubmit);
}
return -1;
}
int32_t tqProcessTaskExec(STQ* pTq, char* msg, int32_t msgLen, int32_t workerId) { int32_t tqProcessTaskExec(STQ* pTq, char* msg, int32_t msgLen, int32_t workerId) {
SStreamTaskExecReq req; SStreamTaskExecReq req;
tDecodeSStreamTaskExecReq(msg, &req); tDecodeSStreamTaskExecReq(msg, &req);

10
source/dnode/vnode/src/tq/tqRead.c
View File

@ -34,21 +34,11 @@ STqReadHandle* tqInitSubmitMsgScanner(SMeta* pMeta) {
int32_t tqReadHandleSetMsg(STqReadHandle* pReadHandle, SSubmitReq* pMsg, int64_t ver) { int32_t tqReadHandleSetMsg(STqReadHandle* pReadHandle, SSubmitReq* pMsg, int64_t ver) {
pReadHandle->pMsg = pMsg; pReadHandle->pMsg = pMsg;
// pMsg->length = htonl(pMsg->length);
// pMsg->numOfBlocks = htonl(pMsg->numOfBlocks);
// iterate and convert
if (tInitSubmitMsgIter(pMsg, &pReadHandle->msgIter) < 0) return -1; if (tInitSubmitMsgIter(pMsg, &pReadHandle->msgIter) < 0) return -1;
while (true) { while (true) {
if (tGetSubmitMsgNext(&pReadHandle->msgIter, &pReadHandle->pBlock) < 0) return -1; if (tGetSubmitMsgNext(&pReadHandle->msgIter, &pReadHandle->pBlock) < 0) return -1;
if (pReadHandle->pBlock == NULL) break; if (pReadHandle->pBlock == NULL) break;
// pReadHandle->pBlock->uid = htobe64(pReadHandle->pBlock->uid);
// pReadHandle->pBlock->suid = htobe64(pReadHandle->pBlock->suid);
// pReadHandle->pBlock->sversion = htonl(pReadHandle->pBlock->sversion);
// pReadHandle->pBlock->dataLen = htonl(pReadHandle->pBlock->dataLen);
// pReadHandle->pBlock->schemaLen = htonl(pReadHandle->pBlock->schemaLen);
// pReadHandle->pBlock->numOfRows = htons(pReadHandle->pBlock->numOfRows);
} }
if (tInitSubmitMsgIter(pMsg, &pReadHandle->msgIter) < 0) return -1; if (tInitSubmitMsgIter(pMsg, &pReadHandle->msgIter) < 0) return -1;

5
source/libs/function/inc/builtinsimpl.h
View File

@ -76,6 +76,11 @@ int32_t firstFunction(SqlFunctionCtx *pCtx);
int32_t lastFunction(SqlFunctionCtx *pCtx); int32_t lastFunction(SqlFunctionCtx *pCtx);
int32_t lastFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock); int32_t lastFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
bool getUniqueFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool uniqueFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
int32_t uniqueFunction(SqlFunctionCtx *pCtx);
int32_t uniqueFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
bool getTopBotFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv); bool getTopBotFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv);
int32_t topFunction(SqlFunctionCtx *pCtx); int32_t topFunction(SqlFunctionCtx *pCtx);
int32_t bottomFunction(SqlFunctionCtx *pCtx); int32_t bottomFunction(SqlFunctionCtx *pCtx);

41
source/libs/function/src/builtins.c
View File

@ -493,6 +493,21 @@ static int32_t translateFirstLast(SFunctionNode* pFunc, char* pErrBuf, int32_t l
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t translateUnique(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
if (1 != LIST_LENGTH(pFunc->pParameterList)) {
return TSDB_CODE_SUCCESS;
}
SNode* pPara = nodesListGetNode(pFunc->pParameterList, 0);
if (QUERY_NODE_COLUMN != nodeType(pPara)) {
return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR,
"The parameters of UNIQUE can only be columns");
}
pFunc->node.resType = ((SExprNode*)pPara)->resType;
return TSDB_CODE_SUCCESS;
}
static int32_t translateDiff(SFunctionNode* pFunc, char* pErrBuf, int32_t len) { static int32_t translateDiff(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
int32_t paraLen = LIST_LENGTH(pFunc->pParameterList); int32_t paraLen = LIST_LENGTH(pFunc->pParameterList);
if (paraLen == 0 || paraLen > 2) { if (paraLen == 0 || paraLen > 2) {
@ -878,14 +893,14 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
.finalizeFunc = lastFinalize .finalizeFunc = lastFinalize
}, },
{ {
.name = "diff", .name = "unique",
.type = FUNCTION_TYPE_DIFF, .type = FUNCTION_TYPE_UNIQUE,
.classification = FUNC_MGT_NONSTANDARD_SQL_FUNC | FUNC_MGT_TIMELINE_FUNC, .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_TIMELINE_FUNC,
.translateFunc = translateDiff, .translateFunc = translateUnique,
.getEnvFunc = getDiffFuncEnv, .getEnvFunc = getUniqueFuncEnv,
.initFunc = diffFunctionSetup, .initFunc = uniqueFunctionSetup,
.processFunc = diffFunction, .processFunc = uniqueFunction,
.finalizeFunc = functionFinalize .finalizeFunc = uniqueFinalize
}, },
{ {
.name = "histogram", .name = "histogram",
@ -907,6 +922,16 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
.processFunc = hllFunction, .processFunc = hllFunction,
.finalizeFunc = hllFinalize .finalizeFunc = hllFinalize
}, },
{
.name = "diff",
.type = FUNCTION_TYPE_DIFF,
.classification = FUNC_MGT_NONSTANDARD_SQL_FUNC | FUNC_MGT_TIMELINE_FUNC,
.translateFunc = translateDiff,
.getEnvFunc = getDiffFuncEnv,
.initFunc = diffFunctionSetup,
.processFunc = diffFunction,
.finalizeFunc = functionFinalize
},
{ {
.name = "state_count", .name = "state_count",
.type = FUNCTION_TYPE_STATE_COUNT, .type = FUNCTION_TYPE_STATE_COUNT,

168
source/libs/function/src/builtinsimpl.c
View File

@ -28,12 +28,15 @@
#define TAIL_MAX_POINTS_NUM 100 #define TAIL_MAX_POINTS_NUM 100
#define TAIL_MAX_OFFSET 100 #define TAIL_MAX_OFFSET 100
#define UNIQUE_MAX_RESULT_SIZE (1024*1024*10)
#define HLL_BUCKET_BITS 14 // The bits of the bucket #define HLL_BUCKET_BITS 14 // The bits of the bucket
#define HLL_DATA_BITS (64-HLL_BUCKET_BITS) #define HLL_DATA_BITS (64-HLL_BUCKET_BITS)
#define HLL_BUCKETS (1<<HLL_BUCKET_BITS) #define HLL_BUCKETS (1<<HLL_BUCKET_BITS)
#define HLL_BUCKET_MASK (HLL_BUCKETS-1) #define HLL_BUCKET_MASK (HLL_BUCKETS-1)
#define HLL_ALPHA_INF 0.721347520444481703680 // constant for 0.5/ln(2) #define HLL_ALPHA_INF 0.721347520444481703680 // constant for 0.5/ln(2)
typedef struct SSumRes { typedef struct SSumRes {
union { union {
int64_t isum; int64_t isum;
@ -197,6 +200,20 @@ typedef struct STailInfo {
STailItem **pItems; STailItem **pItems;
} STailInfo; } STailInfo;
typedef struct SUniqueItem {
int64_t timestamp;
bool isNull;
char data[];
} SUniqueItem;
typedef struct SUniqueInfo {
int32_t numOfPoints;
uint8_t colType;
int16_t colBytes;
SHashObj *pHash;
char pItems[];
} SUniqueInfo;
#define SET_VAL(_info, numOfElem, res) \ #define SET_VAL(_info, numOfElem, res) \
do { \ do { \
if ((numOfElem) <= 0) { \ if ((numOfElem) <= 0) { \
@ -216,6 +233,18 @@ typedef struct STailInfo {
} \ } \
} while (0); } while (0);
#define DO_UPDATE_SUBSID_RES(ctx, ts) \
do { \
for (int32_t _i = 0; _i < (ctx)->subsidiaries.num; ++_i) { \
SqlFunctionCtx* __ctx = (ctx)->subsidiaries.pCtx[_i]; \
if (__ctx->functionId == FUNCTION_TS_DUMMY) { \
__ctx->tag.i = (ts); \
__ctx->tag.nType = TSDB_DATA_TYPE_BIGINT; \
} \
__ctx->fpSet.process(__ctx); \
} \
} while (0)
#define UPDATE_DATA(ctx, left, right, num, sign, _ts) \ #define UPDATE_DATA(ctx, left, right, num, sign, _ts) \
do { \ do { \
if (((left) < (right)) ^ (sign)) { \ if (((left) < (right)) ^ (sign)) { \
@ -748,50 +777,6 @@ bool getMinmaxFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
return true; return true;
} }
#define GET_TS_LIST(x) ((TSKEY*)((x)->ptsList))
#define GET_TS_DATA(x, y) (GET_TS_LIST(x)[(y)])
#define DO_UPDATE_TAG_COLUMNS_WITHOUT_TS(ctx) \
do { \
for (int32_t _i = 0; _i < (ctx)->tagInfo.numOfTagCols; ++_i) { \
SqlFunctionCtx* __ctx = (ctx)->tagInfo.pTagCtxList[_i]; \
__ctx->fpSet.process(__ctx); \
} \
} while (0);
#define DO_UPDATE_SUBSID_RES(ctx, ts) \
do { \
for (int32_t _i = 0; _i < (ctx)->subsidiaries.num; ++_i) { \
SqlFunctionCtx* __ctx = (ctx)->subsidiaries.pCtx[_i]; \
if (__ctx->functionId == FUNCTION_TS_DUMMY) { \
__ctx->tag.i = (ts); \
__ctx->tag.nType = TSDB_DATA_TYPE_BIGINT; \
} \
__ctx->fpSet.process(__ctx); \
} \
} while (0)
#define UPDATE_DATA(ctx, left, right, num, sign, _ts) \
do { \
if (((left) < (right)) ^ (sign)) { \
(left) = (right); \
DO_UPDATE_SUBSID_RES(ctx, _ts); \
(num) += 1; \
} \
} while (0)
#define LOOPCHECK_N(val, _col, ctx, _t, _nrow, _start, sign, num) \
do { \
_t* d = (_t*)((_col)->pData); \
for (int32_t i = (_start); i < (_nrow) + (_start); ++i) { \
if (((_col)->hasNull) && colDataIsNull_f((_col)->nullbitmap, i)) { \
continue; \
} \
TSKEY ts = (ctx)->ptsList != NULL ? GET_TS_DATA(ctx, i) : 0; \
UPDATE_DATA(ctx, val, d[i], num, sign, ts); \
} \
} while (0)
static void saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STuplePos* pPos); static void saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STuplePos* pPos);
static void copyTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STuplePos* pPos); static void copyTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STuplePos* pPos);
@ -1994,6 +1979,99 @@ int32_t lastFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
return pResInfo->numOfRes; return pResInfo->numOfRes;
} }
bool getUniqueFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
pEnv->calcMemSize = sizeof(SUniqueInfo) + UNIQUE_MAX_RESULT_SIZE;
return true;
}
bool uniqueFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResInfo) {
if (!functionSetup(pCtx, pResInfo)) {
return false;
}
SUniqueInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo);
pInfo->numOfPoints = 0;
pInfo->colType = pCtx->resDataInfo.type;
pInfo->colBytes = pCtx->resDataInfo.bytes;
if (pInfo->pHash != NULL) {
taosHashClear(pInfo->pHash);
} else {
pInfo->pHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
}
return true;
}
static void doUniqueAdd(SUniqueInfo* pInfo, char *data, TSKEY ts, bool isNull) {
int32_t hashKeyBytes = IS_VAR_DATA_TYPE(pInfo->colType) ? varDataTLen(data) : pInfo->colBytes;
SUniqueItem *pHashItem = taosHashGet(pInfo->pHash, data, hashKeyBytes);
if (pHashItem == NULL) {
int32_t size = sizeof(SUniqueItem) + pInfo->colBytes;
SUniqueItem *pItem = (SUniqueItem *)(pInfo->pItems + pInfo->numOfPoints * size);
pItem->timestamp = ts;
memcpy(pItem->data, data, pInfo->colBytes);
taosHashPut(pInfo->pHash, data, hashKeyBytes, (char *)pItem, sizeof(SUniqueItem*));
pInfo->numOfPoints++;
} else if (pHashItem->timestamp > ts) {
pHashItem->timestamp = ts;
}
}
int32_t uniqueFunction(SqlFunctionCtx* pCtx) {
SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
SUniqueInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo);
SInputColumnInfoData* pInput = &pCtx->input;
TSKEY* tsList = (int64_t*)pInput->pPTS->pData;
SColumnInfoData* pInputCol = pInput->pData[0];
SColumnInfoData* pTsOutput = pCtx->pTsOutput;
SColumnInfoData* pOutput = (SColumnInfoData*)pCtx->pOutput;
int32_t startOffset = pCtx->offset;
for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; ++i) {
char* data = colDataGetData(pInputCol, i);
doUniqueAdd(pInfo, data, tsList[i], colDataIsNull_s(pInputCol, i));
if (sizeof(SUniqueInfo) + pInfo->numOfPoints * (sizeof(SUniqueItem) + pInfo->colBytes) >= UNIQUE_MAX_RESULT_SIZE) {
taosHashCleanup(pInfo->pHash);
return 0;
}
}
//taosqsort(pInfo->pItems, pInfo->numOfPoints, POINTER_BYTES, NULL, tailCompFn);
//for (int32_t i = 0; i < pInfo->numOfPoints; ++i) {
// int32_t pos = startOffset + i;
// STailItem *pItem = pInfo->pItems[i];
// if (pItem->isNull) {
// colDataAppendNULL(pOutput, pos);
// } else {
// colDataAppend(pOutput, pos, pItem->data, false);
// }
//}
pResInfo->numOfRes = pInfo->numOfPoints;
return TSDB_CODE_SUCCESS;
}
int32_t uniqueFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
SUniqueInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
int32_t slotId = pCtx->pExpr->base.resSchema.slotId;
SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
for (int32_t i = 0; i < pResInfo->numOfRes; ++i) {
SUniqueItem *pItem = (SUniqueItem *)(pInfo->pItems + i * (sizeof(SUniqueItem) + pInfo->colBytes));
colDataAppend(pCol, i, pItem->data, false);
//TODO: handle ts output
}
return pResInfo->numOfRes;
}
bool getDiffFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) { bool getDiffFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
pEnv->calcMemSize = sizeof(SDiffInfo); pEnv->calcMemSize = sizeof(SDiffInfo);
return true; return true;
@ -2106,7 +2184,7 @@ static void doHandleDiff(SDiffInfo* pDiffInfo, int32_t type, const char* pv, SCo
default: default:
ASSERT(0); ASSERT(0);
} }
} }
int32_t diffFunction(SqlFunctionCtx* pCtx) { int32_t diffFunction(SqlFunctionCtx* pCtx) {
SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx); SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);

222
source/libs/function/src/udfd.c
View File

@ -27,16 +27,16 @@
#include "trpc.h" #include "trpc.h"
typedef struct SUdfdContext { typedef struct SUdfdContext {
uv_loop_t *loop; uv_loop_t * loop;
uv_pipe_t ctrlPipe; uv_pipe_t ctrlPipe;
uv_signal_t intrSignal; uv_signal_t intrSignal;
char listenPipeName[PATH_MAX + UDF_LISTEN_PIPE_NAME_LEN + 2]; char listenPipeName[PATH_MAX + UDF_LISTEN_PIPE_NAME_LEN + 2];
uv_pipe_t listeningPipe; uv_pipe_t listeningPipe;
void *clientRpc; void * clientRpc;
SCorEpSet mgmtEp; SCorEpSet mgmtEp;
uv_mutex_t udfsMutex; uv_mutex_t udfsMutex;
SHashObj *udfsHash; SHashObj * udfsHash;
bool printVersion; bool printVersion;
} SUdfdContext; } SUdfdContext;
@ -45,7 +45,7 @@ SUdfdContext global;
typedef struct SUdfdUvConn { typedef struct SUdfdUvConn {
uv_stream_t *client; uv_stream_t *client;
char *inputBuf; char * inputBuf;
int32_t inputLen; int32_t inputLen;
int32_t inputCap; int32_t inputCap;
int32_t inputTotal; int32_t inputTotal;
@ -65,25 +65,25 @@ typedef struct SUdf {
uv_mutex_t lock; uv_mutex_t lock;
uv_cond_t condReady; uv_cond_t condReady;
char name[TSDB_FUNC_NAME_LEN]; char name[TSDB_FUNC_NAME_LEN];
int8_t funcType; int8_t funcType;
int8_t scriptType; int8_t scriptType;
int8_t outputType; int8_t outputType;
int32_t outputLen; int32_t outputLen;
int32_t bufSize; int32_t bufSize;
char path[PATH_MAX]; char path[PATH_MAX];
uv_lib_t lib; uv_lib_t lib;
TUdfScalarProcFunc scalarProcFunc; TUdfScalarProcFunc scalarProcFunc;
TUdfAggStartFunc aggStartFunc; TUdfAggStartFunc aggStartFunc;
TUdfAggProcessFunc aggProcFunc; TUdfAggProcessFunc aggProcFunc;
TUdfAggFinishFunc aggFinishFunc; TUdfAggFinishFunc aggFinishFunc;
TUdfInitFunc initFunc; TUdfInitFunc initFunc;
TUdfDestroyFunc destroyFunc; TUdfDestroyFunc destroyFunc;
} SUdf; } SUdf;
// TODO: add private udf structure. // TODO: add private udf structure.
@ -98,9 +98,9 @@ typedef enum EUdfdRpcReqRspType {
typedef struct SUdfdRpcSendRecvInfo { typedef struct SUdfdRpcSendRecvInfo {
EUdfdRpcReqRspType rpcType; EUdfdRpcReqRspType rpcType;
int32_t code; int32_t code;
void* param; void * param;
uv_sem_t resultSem; uv_sem_t resultSem;
} SUdfdRpcSendRecvInfo; } SUdfdRpcSendRecvInfo;
void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) { void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
@ -136,7 +136,7 @@ void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
tDeserializeSRetrieveFuncRsp(pMsg->pCont, pMsg->contLen, &retrieveRsp); tDeserializeSRetrieveFuncRsp(pMsg->pCont, pMsg->contLen, &retrieveRsp);
SFuncInfo *pFuncInfo = (SFuncInfo *)taosArrayGet(retrieveRsp.pFuncInfos, 0); SFuncInfo *pFuncInfo = (SFuncInfo *)taosArrayGet(retrieveRsp.pFuncInfos, 0);
SUdf* udf = msgInfo->param; SUdf * udf = msgInfo->param;
udf->funcType = pFuncInfo->funcType; udf->funcType = pFuncInfo->funcType;
udf->scriptType = pFuncInfo->scriptType; udf->scriptType = pFuncInfo->scriptType;
udf->outputType = pFuncInfo->outputType; udf->outputType = pFuncInfo->outputType;
@ -145,7 +145,8 @@ void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
char path[PATH_MAX] = {0}; char path[PATH_MAX] = {0};
snprintf(path, sizeof(path), "%s/lib%s.so", "/tmp", pFuncInfo->name); snprintf(path, sizeof(path), "%s/lib%s.so", "/tmp", pFuncInfo->name);
TdFilePtr file = taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC | TD_FILE_AUTO_DEL); TdFilePtr file =
taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC | TD_FILE_AUTO_DEL);
// TODO check for failure of flush to disk // TODO check for failure of flush to disk
taosWriteFile(file, pFuncInfo->pCode, pFuncInfo->codeSize); taosWriteFile(file, pFuncInfo->pCode, pFuncInfo->codeSize);
taosCloseFile(&file); taosCloseFile(&file);
@ -168,11 +169,11 @@ int32_t udfdFillUdfInfoFromMNode(void *clientRpc, char *udfName, SUdf *udf) {
taosArrayPush(retrieveReq.pFuncNames, udfName); taosArrayPush(retrieveReq.pFuncNames, udfName);
int32_t contLen = tSerializeSRetrieveFuncReq(NULL, 0, &retrieveReq); int32_t contLen = tSerializeSRetrieveFuncReq(NULL, 0, &retrieveReq);
void *pReq = rpcMallocCont(contLen); void * pReq = rpcMallocCont(contLen);
tSerializeSRetrieveFuncReq(pReq, contLen, &retrieveReq); tSerializeSRetrieveFuncReq(pReq, contLen, &retrieveReq);
taosArrayDestroy(retrieveReq.pFuncNames); taosArrayDestroy(retrieveReq.pFuncNames);
SUdfdRpcSendRecvInfo* msgInfo = taosMemoryCalloc(1, sizeof(SUdfdRpcSendRecvInfo)); SUdfdRpcSendRecvInfo *msgInfo = taosMemoryCalloc(1, sizeof(SUdfdRpcSendRecvInfo));
msgInfo->rpcType = UDFD_RPC_RETRIVE_FUNC; msgInfo->rpcType = UDFD_RPC_RETRIVE_FUNC;
msgInfo->param = udf; msgInfo->param = udf;
uv_sem_init(&msgInfo->resultSem, 0); uv_sem_init(&msgInfo->resultSem, 0);
@ -194,7 +195,7 @@ int32_t udfdFillUdfInfoFromMNode(void *clientRpc, char *udfName, SUdf *udf) {
int32_t udfdConnectToMnode() { int32_t udfdConnectToMnode() {
SConnectReq connReq = {0}; SConnectReq connReq = {0};
connReq.connType = CONN_TYPE__UDFD; connReq.connType = CONN_TYPE__UDFD;
tstrncpy(connReq.app, "udfd",sizeof(connReq.app)); tstrncpy(connReq.app, "udfd", sizeof(connReq.app));
tstrncpy(connReq.user, TSDB_DEFAULT_USER, sizeof(connReq.user)); tstrncpy(connReq.user, TSDB_DEFAULT_USER, sizeof(connReq.user));
char pass[TSDB_PASSWORD_LEN + 1] = {0}; char pass[TSDB_PASSWORD_LEN + 1] = {0};
taosEncryptPass_c((uint8_t *)(TSDB_DEFAULT_PASS), strlen(TSDB_DEFAULT_PASS), pass); taosEncryptPass_c((uint8_t *)(TSDB_DEFAULT_PASS), strlen(TSDB_DEFAULT_PASS), pass);
@ -203,7 +204,7 @@ int32_t udfdConnectToMnode() {
connReq.startTime = htobe64(taosGetTimestampMs()); connReq.startTime = htobe64(taosGetTimestampMs());
int32_t contLen = tSerializeSConnectReq(NULL, 0, &connReq); int32_t contLen = tSerializeSConnectReq(NULL, 0, &connReq);
void* pReq = rpcMallocCont(contLen); void * pReq = rpcMallocCont(contLen);
tSerializeSConnectReq(pReq, contLen, &connReq); tSerializeSConnectReq(pReq, contLen, &connReq);
SUdfdRpcSendRecvInfo *msgInfo = taosMemoryCalloc(1, sizeof(SUdfdRpcSendRecvInfo)); SUdfdRpcSendRecvInfo *msgInfo = taosMemoryCalloc(1, sizeof(SUdfdRpcSendRecvInfo));
@ -240,17 +241,17 @@ int32_t udfdLoadUdf(char *udfName, SUdf *udf) {
return TSDB_CODE_UDF_LOAD_UDF_FAILURE; return TSDB_CODE_UDF_LOAD_UDF_FAILURE;
} }
char initFuncName[TSDB_FUNC_NAME_LEN+5] = {0}; char initFuncName[TSDB_FUNC_NAME_LEN + 5] = {0};
char *initSuffix = "_init"; char *initSuffix = "_init";
strcpy(initFuncName, udfName); strcpy(initFuncName, udfName);
strncat(initFuncName, initSuffix, strlen(initSuffix)); strncat(initFuncName, initSuffix, strlen(initSuffix));
uv_dlsym(&udf->lib, initFuncName, (void**)(&udf->initFunc)); uv_dlsym(&udf->lib, initFuncName, (void **)(&udf->initFunc));
char destroyFuncName[TSDB_FUNC_NAME_LEN+5] = {0}; char destroyFuncName[TSDB_FUNC_NAME_LEN + 5] = {0};
char *destroySuffix = "_destroy"; char *destroySuffix = "_destroy";
strcpy(destroyFuncName, udfName); strcpy(destroyFuncName, udfName);
strncat(destroyFuncName, destroySuffix, strlen(destroySuffix)); strncat(destroyFuncName, destroySuffix, strlen(destroySuffix));
uv_dlsym(&udf->lib, destroyFuncName, (void**)(&udf->destroyFunc)); uv_dlsym(&udf->lib, destroyFuncName, (void **)(&udf->destroyFunc));
if (udf->funcType == TSDB_FUNC_TYPE_SCALAR) { if (udf->funcType == TSDB_FUNC_TYPE_SCALAR) {
char processFuncName[TSDB_FUNC_NAME_LEN] = {0}; char processFuncName[TSDB_FUNC_NAME_LEN] = {0};
@ -270,87 +271,86 @@ int32_t udfdLoadUdf(char *udfName, SUdf *udf) {
strncpy(finishFuncName, processFuncName, strlen(processFuncName)); strncpy(finishFuncName, processFuncName, strlen(processFuncName));
strncat(finishFuncName, finishSuffix, strlen(finishSuffix)); strncat(finishFuncName, finishSuffix, strlen(finishSuffix));
uv_dlsym(&udf->lib, finishFuncName, (void **)(&udf->aggFinishFunc)); uv_dlsym(&udf->lib, finishFuncName, (void **)(&udf->aggFinishFunc));
//TODO: merge // TODO: merge
} }
return 0; return 0;
} }
void udfdProcessSetupRequest(SUvUdfWork* uvUdf, SUdfRequest* request) { void udfdProcessSetupRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
// TODO: tracable id from client. connect, setup, call, teardown // TODO: tracable id from client. connect, setup, call, teardown
fnInfo( "setup request. seq num: %" PRId64 ", udf name: %s", request->seqNum, request->setup.udfName); fnInfo("setup request. seq num: %" PRId64 ", udf name: %s", request->seqNum, request->setup.udfName);
SUdfSetupRequest *setup = &request->setup; SUdfSetupRequest *setup = &request->setup;
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
SUdf *udf = NULL; SUdf * udf = NULL;
uv_mutex_lock(&global.udfsMutex); uv_mutex_lock(&global.udfsMutex);
SUdf **udfInHash = taosHashGet(global.udfsHash, request->setup.udfName, strlen(request->setup.udfName)); SUdf **udfInHash = taosHashGet(global.udfsHash, request->setup.udfName, strlen(request->setup.udfName));
if (udfInHash) { if (udfInHash) {
++(*udfInHash)->refCount; ++(*udfInHash)->refCount;
udf = *udfInHash; udf = *udfInHash;
uv_mutex_unlock(&global.udfsMutex); uv_mutex_unlock(&global.udfsMutex);
} else { } else {
SUdf *udfNew = taosMemoryCalloc(1, sizeof(SUdf)); SUdf *udfNew = taosMemoryCalloc(1, sizeof(SUdf));
udfNew->refCount = 1; udfNew->refCount = 1;
udfNew->state = UDF_STATE_INIT; udfNew->state = UDF_STATE_INIT;
uv_mutex_init(&udfNew->lock); uv_mutex_init(&udfNew->lock);
uv_cond_init(&udfNew->condReady); uv_cond_init(&udfNew->condReady);
udf = udfNew; udf = udfNew;
taosHashPut(global.udfsHash, request->setup.udfName, strlen(request->setup.udfName), &udfNew, sizeof(&udfNew)); taosHashPut(global.udfsHash, request->setup.udfName, strlen(request->setup.udfName), &udfNew, sizeof(&udfNew));
uv_mutex_unlock(&global.udfsMutex); uv_mutex_unlock(&global.udfsMutex);
}
uv_mutex_lock(&udf->lock);
if (udf->state == UDF_STATE_INIT) {
udf->state = UDF_STATE_LOADING;
code = udfdLoadUdf(setup->udfName, udf);
if (udf->initFunc) {
udf->initFunc();
} }
udf->state = UDF_STATE_READY;
uv_mutex_lock(&udf->lock); uv_cond_broadcast(&udf->condReady);
if (udf->state == UDF_STATE_INIT) { uv_mutex_unlock(&udf->lock);
udf->state = UDF_STATE_LOADING; } else {
code = udfdLoadUdf(setup->udfName, udf); while (udf->state != UDF_STATE_READY) {
if (udf->initFunc) { uv_cond_wait(&udf->condReady, &udf->lock);
udf->initFunc();
}
udf->state = UDF_STATE_READY;
uv_cond_broadcast(&udf->condReady);
uv_mutex_unlock(&udf->lock);
} else {
while (udf->state != UDF_STATE_READY) {
uv_cond_wait(&udf->condReady, &udf->lock);
}
uv_mutex_unlock(&udf->lock);
} }
SUdfcFuncHandle *handle = taosMemoryMalloc(sizeof(SUdfcFuncHandle)); uv_mutex_unlock(&udf->lock);
handle->udf = udf; }
SUdfcFuncHandle *handle = taosMemoryMalloc(sizeof(SUdfcFuncHandle));
handle->udf = udf;
SUdfResponse rsp; SUdfResponse rsp;
rsp.seqNum = request->seqNum; rsp.seqNum = request->seqNum;
rsp.type = request->type; rsp.type = request->type;
rsp.code = code; rsp.code = code;
rsp.setupRsp.udfHandle = (int64_t)(handle); rsp.setupRsp.udfHandle = (int64_t)(handle);
rsp.setupRsp.outputType = udf->outputType; rsp.setupRsp.outputType = udf->outputType;
rsp.setupRsp.outputLen = udf->outputLen; rsp.setupRsp.outputLen = udf->outputLen;
rsp.setupRsp.bufSize = udf->bufSize; rsp.setupRsp.bufSize = udf->bufSize;
int32_t len = encodeUdfResponse(NULL, &rsp); int32_t len = encodeUdfResponse(NULL, &rsp);
rsp.msgLen = len; rsp.msgLen = len;
void *bufBegin = taosMemoryMalloc(len); void *bufBegin = taosMemoryMalloc(len);
void *buf = bufBegin; void *buf = bufBegin;
encodeUdfResponse(&buf, &rsp); encodeUdfResponse(&buf, &rsp);
uvUdf->output = uv_buf_init(bufBegin, len); uvUdf->output = uv_buf_init(bufBegin, len);
taosMemoryFree(uvUdf->input.base); taosMemoryFree(uvUdf->input.base);
return; return;
} }
void udfdProcessCallRequest(SUvUdfWork *uvUdf, SUdfRequest *request) { void udfdProcessCallRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
SUdfCallRequest *call = &request->call; SUdfCallRequest *call = &request->call;
fnDebug("%" PRId64 "call request. call type %d, handle: %" PRIx64, request->seqNum, call->callType, fnDebug("%" PRId64 "call request. call type %d, handle: %" PRIx64, request->seqNum, call->callType, call->udfHandle);
call->udfHandle); SUdfcFuncHandle * handle = (SUdfcFuncHandle *)(call->udfHandle);
SUdfcFuncHandle *handle = (SUdfcFuncHandle *)(call->udfHandle); SUdf * udf = handle->udf;
SUdf *udf = handle->udf; SUdfResponse response = {0};
SUdfResponse response = {0}; SUdfResponse * rsp = &response;
SUdfResponse *rsp = &response;
SUdfCallResponse *subRsp = &rsp->callRsp; SUdfCallResponse *subRsp = &rsp->callRsp;
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
switch(call->callType) { switch (call->callType) {
case TSDB_UDF_CALL_SCALA_PROC: { case TSDB_UDF_CALL_SCALA_PROC: {
SUdfColumn output = {0}; SUdfColumn output = {0};
@ -363,9 +363,7 @@ void udfdProcessCallRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
break; break;
} }
case TSDB_UDF_CALL_AGG_INIT: { case TSDB_UDF_CALL_AGG_INIT: {
SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize), SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize), .bufLen = udf->bufSize, .numOfResult = 0};
.bufLen= udf->bufSize,
.numOfResult = 0};
udf->aggStartFunc(&outBuf); udf->aggStartFunc(&outBuf);
subRsp->resultBuf = outBuf; subRsp->resultBuf = outBuf;
break; break;
@ -373,9 +371,7 @@ void udfdProcessCallRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
case TSDB_UDF_CALL_AGG_PROC: { case TSDB_UDF_CALL_AGG_PROC: {
SUdfDataBlock input = {0}; SUdfDataBlock input = {0};
convertDataBlockToUdfDataBlock(&call->block, &input); convertDataBlockToUdfDataBlock(&call->block, &input);
SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize), SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize), .bufLen = udf->bufSize, .numOfResult = 0};
.bufLen= udf->bufSize,
.numOfResult = 0};
code = udf->aggProcFunc(&input, &call->interBuf, &outBuf); code = udf->aggProcFunc(&input, &call->interBuf, &outBuf);
freeUdfInterBuf(&call->interBuf); freeUdfInterBuf(&call->interBuf);
freeUdfDataDataBlock(&input); freeUdfDataDataBlock(&input);
@ -384,9 +380,7 @@ void udfdProcessCallRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
break; break;
} }
case TSDB_UDF_CALL_AGG_FIN: { case TSDB_UDF_CALL_AGG_FIN: {
SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize), SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize), .bufLen = udf->bufSize, .numOfResult = 0};
.bufLen= udf->bufSize,
.numOfResult = 0};
code = udf->aggFinishFunc(&call->interBuf, &outBuf); code = udf->aggFinishFunc(&call->interBuf, &outBuf);
freeUdfInterBuf(&call->interBuf); freeUdfInterBuf(&call->interBuf);
subRsp->resultBuf = outBuf; subRsp->resultBuf = outBuf;
@ -429,20 +423,19 @@ void udfdProcessCallRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
} }
default: default:
break; break;
} }
taosMemoryFree(uvUdf->input.base); taosMemoryFree(uvUdf->input.base);
return; return;
} }
void udfdProcessTeardownRequest(SUvUdfWork* uvUdf, SUdfRequest* request) { void udfdProcessTeardownRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
SUdfTeardownRequest *teardown = &request->teardown; SUdfTeardownRequest *teardown = &request->teardown;
fnInfo("teardown. seq number: %" PRId64 ", handle:%" PRIx64, request->seqNum, teardown->udfHandle); fnInfo("teardown. seq number: %" PRId64 ", handle:%" PRIx64, request->seqNum, teardown->udfHandle);
SUdfcFuncHandle *handle = (SUdfcFuncHandle *)(teardown->udfHandle); SUdfcFuncHandle *handle = (SUdfcFuncHandle *)(teardown->udfHandle);
SUdf *udf = handle->udf; SUdf * udf = handle->udf;
bool unloadUdf = false; bool unloadUdf = false;
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
uv_mutex_lock(&global.udfsMutex); uv_mutex_lock(&global.udfsMutex);
udf->refCount--; udf->refCount--;
@ -568,7 +561,7 @@ bool isUdfdUvMsgComplete(SUdfdUvConn *pipe) {
} }
void udfdHandleRequest(SUdfdUvConn *conn) { void udfdHandleRequest(SUdfdUvConn *conn) {
uv_work_t *work = taosMemoryMalloc(sizeof(uv_work_t)); uv_work_t * work = taosMemoryMalloc(sizeof(uv_work_t));
SUvUdfWork *udfWork = taosMemoryMalloc(sizeof(SUvUdfWork)); SUvUdfWork *udfWork = taosMemoryMalloc(sizeof(SUvUdfWork));
udfWork->client = conn->client; udfWork->client = conn->client;
udfWork->input = uv_buf_init(conn->inputBuf, conn->inputLen); udfWork->input = uv_buf_init(conn->inputBuf, conn->inputLen);
@ -653,11 +646,11 @@ static bool udfdRpcRfp(int32_t code) {
} }
} }
int initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSet) { int initEpSetFromCfg(const char *firstEp, const char *secondEp, SCorEpSet *pEpSet) {
pEpSet->version = 0; pEpSet->version = 0;
// init mnode ip set // init mnode ip set
SEpSet* mgmtEpSet = &(pEpSet->epSet); SEpSet *mgmtEpSet = &(pEpSet->epSet);
mgmtEpSet->numOfEps = 0; mgmtEpSet->numOfEps = 0;
mgmtEpSet->inUse = 0; mgmtEpSet->inUse = 0;
@ -694,7 +687,6 @@ int initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSe
return 0; return 0;
} }
int32_t udfdOpenClientRpc() { int32_t udfdOpenClientRpc() {
SRpcInit rpcInit = {0}; SRpcInit rpcInit = {0};
rpcInit.label = "UDFD"; rpcInit.label = "UDFD";
@ -704,15 +696,9 @@ int32_t udfdOpenClientRpc() {
rpcInit.connType = TAOS_CONN_CLIENT; rpcInit.connType = TAOS_CONN_CLIENT;
rpcInit.idleTime = tsShellActivityTimer * 1000; rpcInit.idleTime = tsShellActivityTimer * 1000;
rpcInit.user = TSDB_DEFAULT_USER; rpcInit.user = TSDB_DEFAULT_USER;
rpcInit.ckey = "key";
rpcInit.spi = 1;
rpcInit.parent = &global; rpcInit.parent = &global;
rpcInit.rfp = udfdRpcRfp; rpcInit.rfp = udfdRpcRfp;
char pass[TSDB_PASSWORD_LEN + 1] = {0};
taosEncryptPass_c((uint8_t *)(TSDB_DEFAULT_PASS), strlen(TSDB_DEFAULT_PASS), pass);
rpcInit.secret = pass;
global.clientRpc = rpcOpen(&rpcInit); global.clientRpc = rpcOpen(&rpcInit);
if (global.clientRpc == NULL) { if (global.clientRpc == NULL) {
fnError("failed to init dnode rpc client"); fnError("failed to init dnode rpc client");
@ -823,7 +809,7 @@ static int32_t udfdUvInit() {
return 0; return 0;
} }
static void udfdCloseWalkCb(uv_handle_t* handle, void* arg) { static void udfdCloseWalkCb(uv_handle_t *handle, void *arg) {
if (!uv_is_closing(handle)) { if (!uv_is_closing(handle)) {
uv_close(handle, NULL); uv_close(handle, NULL);
} }
@ -883,7 +869,7 @@ int main(int argc, char *argv[]) {
int32_t retryMnodeTimes = 0; int32_t retryMnodeTimes = 0;
int32_t code = 0; int32_t code = 0;
while (retryMnodeTimes++ < TSDB_MAX_REPLICA) { while (retryMnodeTimes++ < TSDB_MAX_REPLICA) {
uv_sleep(500 * ( 1 << retryMnodeTimes)); uv_sleep(500 * (1 << retryMnodeTimes));
code = udfdConnectToMnode(); code = udfdConnectToMnode();
if (code == 0) { if (code == 0) {
break; break;

26
source/libs/index/src/indexFst.c
View File

@ -99,7 +99,7 @@ void fstUnFinishedNodesAddSuffix(FstUnFinishedNodes* nodes, FstSlice bs, Output
if (fstSliceIsEmpty(s)) { if (fstSliceIsEmpty(s)) {
return; return;
} }
size_t sz = taosArrayGetSize(nodes->stack) - 1; int32_t sz = taosArrayGetSize(nodes->stack) - 1;
FstBuilderNodeUnfinished* un = taosArrayGet(nodes->stack, sz); FstBuilderNodeUnfinished* un = taosArrayGet(nodes->stack, sz);
assert(un->last == NULL); assert(un->last == NULL);
@ -130,11 +130,11 @@ void fstUnFinishedNodesAddSuffix(FstUnFinishedNodes* nodes, FstSlice bs, Output
uint64_t fstUnFinishedNodesFindCommPrefix(FstUnFinishedNodes* node, FstSlice bs) { uint64_t fstUnFinishedNodesFindCommPrefix(FstUnFinishedNodes* node, FstSlice bs) {
FstSlice* s = &bs; FstSlice* s = &bs;
size_t ssz = taosArrayGetSize(node->stack); // stack size int32_t ssz = taosArrayGetSize(node->stack); // stack size
uint64_t count = 0; uint64_t count = 0;
int32_t lsz; // data len int32_t lsz; // data len
uint8_t* data = fstSliceData(s, &lsz); uint8_t* data = fstSliceData(s, &lsz);
for (size_t i = 0; i < ssz && i < lsz; i++) { for (int32_t i = 0; i < ssz && i < lsz; i++) {
FstBuilderNodeUnfinished* un = taosArrayGet(node->stack, i); FstBuilderNodeUnfinished* un = taosArrayGet(node->stack, i);
if (un->last->inp == data[i]) { if (un->last->inp == data[i]) {
count++; count++;
@ -147,8 +147,8 @@ uint64_t fstUnFinishedNodesFindCommPrefix(FstUnFinishedNodes* node, FstSlice bs)
uint64_t fstUnFinishedNodesFindCommPrefixAndSetOutput(FstUnFinishedNodes* node, FstSlice bs, Output in, Output* out) { uint64_t fstUnFinishedNodesFindCommPrefixAndSetOutput(FstUnFinishedNodes* node, FstSlice bs, Output in, Output* out) {
FstSlice* s = &bs; FstSlice* s = &bs;
size_t lsz = (size_t)(s->end - s->start + 1); // data len int32_t lsz = (size_t)(s->end - s->start + 1); // data len
size_t ssz = taosArrayGetSize(node->stack); // stack size int32_t ssz = taosArrayGetSize(node->stack); // stack size
*out = in; *out = in;
uint64_t i = 0; uint64_t i = 0;
for (i = 0; i < lsz && i < ssz; i++) { for (i = 0; i < lsz && i < ssz; i++) {
@ -245,7 +245,7 @@ void fstStateCompileForOneTrans(FstCountingWriter* w, CompiledAddr addr, FstTran
return; return;
} }
void fstStateCompileForAnyTrans(FstCountingWriter* w, CompiledAddr addr, FstBuilderNode* node) { void fstStateCompileForAnyTrans(FstCountingWriter* w, CompiledAddr addr, FstBuilderNode* node) {
size_t sz = taosArrayGetSize(node->trans); int32_t sz = taosArrayGetSize(node->trans);
assert(sz <= 256); assert(sz <= 256);
uint8_t tSize = 0; uint8_t tSize = 0;
@ -253,7 +253,7 @@ void fstStateCompileForAnyTrans(FstCountingWriter* w, CompiledAddr addr, FstBuil
// finalOutput.is_zero() // finalOutput.is_zero()
bool anyOuts = (node->finalOutput != 0); bool anyOuts = (node->finalOutput != 0);
for (size_t i = 0; i < sz; i++) { for (int32_t i = 0; i < sz; i++) {
FstTransition* t = taosArrayGet(node->trans, i); FstTransition* t = taosArrayGet(node->trans, i);
tSize = TMAX(tSize, packDeltaSize(addr, t->addr)); tSize = TMAX(tSize, packDeltaSize(addr, t->addr));
oSize = TMAX(oSize, packSize(t->out)); oSize = TMAX(oSize, packSize(t->out));
@ -301,7 +301,7 @@ void fstStateCompileForAnyTrans(FstCountingWriter* w, CompiledAddr addr, FstBuil
/// for (uint8_t i = 0; i < 256; i++) { /// for (uint8_t i = 0; i < 256; i++) {
// index[i] = 255; // index[i] = 255;
///} ///}
for (size_t i = 0; i < sz; i++) { for (int32_t i = 0; i < sz; i++) {
FstTransition* t = taosArrayGet(node->trans, i); FstTransition* t = taosArrayGet(node->trans, i);
index[t->inp] = i; index[t->inp] = i;
// fstPackDeltaIn(w, addr, t->addr, tSize); // fstPackDeltaIn(w, addr, t->addr, tSize);
@ -731,7 +731,7 @@ bool fstNodeFindInput(FstNode* node, uint8_t b, uint64_t* res) {
} }
bool fstNodeCompile(FstNode* node, void* w, CompiledAddr lastAddr, CompiledAddr addr, FstBuilderNode* builderNode) { bool fstNodeCompile(FstNode* node, void* w, CompiledAddr lastAddr, CompiledAddr addr, FstBuilderNode* builderNode) {
size_t sz = taosArrayGetSize(builderNode->trans); int32_t sz = taosArrayGetSize(builderNode->trans);
assert(sz < 256); assert(sz < 256);
if (sz == 0 && builderNode->isFinal && builderNode->finalOutput == 0) { if (sz == 0 && builderNode->isFinal && builderNode->finalOutput == 0) {
return true; return true;
@ -959,8 +959,8 @@ void fstBuilderNodeUnfinishedAddOutputPrefix(FstBuilderNodeUnfinished* unNode, O
if (FST_BUILDER_NODE_IS_FINAL(unNode->node)) { if (FST_BUILDER_NODE_IS_FINAL(unNode->node)) {
unNode->node->finalOutput += out; unNode->node->finalOutput += out;
} }
size_t sz = taosArrayGetSize(unNode->node->trans); int32_t sz = taosArrayGetSize(unNode->node->trans);
for (size_t i = 0; i < sz; i++) { for (int32_t i = 0; i < sz; i++) {
FstTransition* trn = taosArrayGet(unNode->node->trans, i); FstTransition* trn = taosArrayGet(unNode->node->trans, i);
trn->out += out; trn->out += out;
} }
@ -1077,7 +1077,7 @@ bool fstGet(Fst* fst, FstSlice* b, Output* out) {
tOut = tOut + FST_NODE_FINAL_OUTPUT(root); tOut = tOut + FST_NODE_FINAL_OUTPUT(root);
} }
for (size_t i = 0; i < taosArrayGetSize(nodes); i++) { for (int32_t i = 0; i < taosArrayGetSize(nodes); i++) {
FstNode** node = (FstNode**)taosArrayGet(nodes, i); FstNode** node = (FstNode**)taosArrayGet(nodes, i);
fstNodeDestroy(*node); fstNodeDestroy(*node);
} }
@ -1352,7 +1352,7 @@ StreamWithStateResult* streamWithStateNextWith(StreamWithState* sws, StreamCallb
StreamState s2 = {.node = nextNode, .trans = 0, .out = {.null = false, .out = out}, .autState = nextState}; StreamState s2 = {.node = nextNode, .trans = 0, .out = {.null = false, .out = out}, .autState = nextState};
taosArrayPush(sws->stack, &s2); taosArrayPush(sws->stack, &s2);
size_t isz = taosArrayGetSize(sws->inp); int32_t isz = taosArrayGetSize(sws->inp);
uint8_t* buf = (uint8_t*)taosMemoryMalloc(isz * sizeof(uint8_t)); uint8_t* buf = (uint8_t*)taosMemoryMalloc(isz * sizeof(uint8_t));
for (uint32_t i = 0; i < isz; i++) { for (uint32_t i = 0; i < isz; i++) {
buf[i] = *(uint8_t*)taosArrayGet(sws->inp, i); buf[i] = *(uint8_t*)taosArrayGet(sws->inp, i);

6
source/libs/index/src/indexTfile.c
View File

@ -116,7 +116,7 @@ TFileCache* tfileCacheCreate(const char* path) {
continue; continue;
} }
TFileHeader* header = &reader->header; TFileHeader* header = &reader->header;
ICacheKey key = {.suid = header->suid, .colName = header->colName, .nColName = strlen(header->colName)}; ICacheKey key = {.suid = header->suid, .colName = header->colName, .nColName = (int32_t)strlen(header->colName)};
char buf[128] = {0}; char buf[128] = {0};
int32_t sz = indexSerialCacheKey(&key, buf); int32_t sz = indexSerialCacheKey(&key, buf);
@ -230,7 +230,7 @@ static int32_t tfSearchTerm(void* reader, SIndexTerm* tem, SIdxTempResult* tr) {
indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, found table info in tindex, time cost: %" PRIu64 "us", indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, found table info in tindex, time cost: %" PRIu64 "us",
tem->suid, tem->colName, tem->colVal, cost); tem->suid, tem->colName, tem->colVal, cost);
ret = tfileReaderLoadTableIds((TFileReader*)reader, offset, tr->total); ret = tfileReaderLoadTableIds((TFileReader*)reader, (int32_t)offset, tr->total);
cost = taosGetTimestampUs() - et; cost = taosGetTimestampUs() - et;
indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, load all table info, time cost: %" PRIu64 "us", tem->suid, indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, load all table info, time cost: %" PRIu64 "us", tem->suid,
tem->colName, tem->colVal, cost); tem->colName, tem->colVal, cost);
@ -890,7 +890,7 @@ static int tfileWriteFooter(TFileWriter* write) {
char buf[sizeof(tfileMagicNumber) + 1] = {0}; char buf[sizeof(tfileMagicNumber) + 1] = {0};
void* pBuf = (void*)buf; void* pBuf = (void*)buf;
taosEncodeFixedU64((void**)(void*)&pBuf, tfileMagicNumber); taosEncodeFixedU64((void**)(void*)&pBuf, tfileMagicNumber);
int nwrite = write->ctx->write(write->ctx, buf, strlen(buf)); int nwrite = write->ctx->write(write->ctx, buf, (int32_t)strlen(buf));
indexInfo("tfile write footer size: %d", write->ctx->size(write->ctx)); indexInfo("tfile write footer size: %d", write->ctx->size(write->ctx));
assert(nwrite == sizeof(tfileMagicNumber)); assert(nwrite == sizeof(tfileMagicNumber));

17
source/libs/index/src/indexUtil.c
View File

@ -37,14 +37,14 @@ static int iBinarySearch(SArray *arr, int s, int e, uint64_t k) {
} }
void iIntersection(SArray *inters, SArray *final) { void iIntersection(SArray *inters, SArray *final) {
int32_t sz = taosArrayGetSize(inters); int32_t sz = (int32_t)taosArrayGetSize(inters);
if (sz <= 0) { if (sz <= 0) {
return; return;
} }
MergeIndex *mi = taosMemoryCalloc(sz, sizeof(MergeIndex)); MergeIndex *mi = taosMemoryCalloc(sz, sizeof(MergeIndex));
for (int i = 0; i < sz; i++) { for (int i = 0; i < sz; i++) {
SArray *t = taosArrayGetP(inters, i); SArray *t = taosArrayGetP(inters, i);
mi[i].len = taosArrayGetSize(t); mi[i].len = (int32_t)taosArrayGetSize(t);
mi[i].idx = 0; mi[i].idx = 0;
} }
@ -70,7 +70,7 @@ void iIntersection(SArray *inters, SArray *final) {
taosMemoryFreeClear(mi); taosMemoryFreeClear(mi);
} }
void iUnion(SArray *inters, SArray *final) { void iUnion(SArray *inters, SArray *final) {
int32_t sz = taosArrayGetSize(inters); int32_t sz = (int32_t)taosArrayGetSize(inters);
if (sz <= 0) { if (sz <= 0) {
return; return;
} }
@ -82,7 +82,7 @@ void iUnion(SArray *inters, SArray *final) {
MergeIndex *mi = taosMemoryCalloc(sz, sizeof(MergeIndex)); MergeIndex *mi = taosMemoryCalloc(sz, sizeof(MergeIndex));
for (int i = 0; i < sz; i++) { for (int i = 0; i < sz; i++) {
SArray *t = taosArrayGetP(inters, i); SArray *t = taosArrayGetP(inters, i);
mi[i].len = taosArrayGetSize(t); mi[i].len = (int32_t)taosArrayGetSize(t);
mi[i].idx = 0; mi[i].idx = 0;
} }
while (1) { while (1) {
@ -117,8 +117,8 @@ void iUnion(SArray *inters, SArray *final) {
} }
void iExcept(SArray *total, SArray *except) { void iExcept(SArray *total, SArray *except) {
int32_t tsz = taosArrayGetSize(total); int32_t tsz = (int32_t)taosArrayGetSize(total);
int32_t esz = taosArrayGetSize(except); int32_t esz = (int32_t)taosArrayGetSize(except);
if (esz == 0 || tsz == 0) { if (esz == 0 || tsz == 0) {
return; return;
} }
@ -141,7 +141,10 @@ int uidCompare(const void *a, const void *b) {
// add more version compare // add more version compare
uint64_t u1 = *(uint64_t *)a; uint64_t u1 = *(uint64_t *)a;
uint64_t u2 = *(uint64_t *)b; uint64_t u2 = *(uint64_t *)b;
return u1 - u2; if (u1 == u2) {
return 0;
}
return u1 < u2 ? -1 : 1;
} }
int verdataCompare(const void *a, const void *b) { int verdataCompare(const void *a, const void *b) {
SIdxVerdata *va = (SIdxVerdata *)a; SIdxVerdata *va = (SIdxVerdata *)a;

4
source/libs/index/test/fstTest.cc
View File

@ -48,7 +48,7 @@ class FstWriter {
class FstReadMemory { class FstReadMemory {
public: public:
FstReadMemory(size_t size, const std::string& fileName = "/tmp/tindex.tindex") { FstReadMemory(int32_t size, const std::string& fileName = "/tmp/tindex.tindex") {
_wc = writerCtxCreate(TFile, fileName.c_str(), true, 64 * 1024); _wc = writerCtxCreate(TFile, fileName.c_str(), true, 64 * 1024);
_w = fstCountingWriterCreate(_wc); _w = fstCountingWriterCreate(_wc);
_size = size; _size = size;
@ -152,7 +152,7 @@ class FstReadMemory {
Fst* _fst; Fst* _fst;
FstSlice _s; FstSlice _s;
WriterCtx* _wc; WriterCtx* _wc;
size_t _size; int32_t _size;
}; };
#define L 100 #define L 100

144
source/libs/stream/src/tstream.c
View File

@ -247,6 +247,19 @@ int32_t streamTaskExec2(SStreamTask* pTask, SMsgCb* pMsgCb) {
void* data = NULL; void* data = NULL;
taosGetQitem(pTask->inputQAll, &data); taosGetQitem(pTask->inputQAll, &data);
if (data == NULL) break; if (data == NULL) break;
streamTaskExecImpl(pTask, data, pRes);
taosFreeQitem(data);
if (taosArrayGetSize(pRes) != 0) {
SStreamDataBlock* resQ = taosAllocateQitem(sizeof(void**), DEF_QITEM);
resQ->type = STREAM_INPUT__DATA_BLOCK;
resQ->blocks = pRes;
taosWriteQitem(pTask->outputQ, resQ);
pRes = taosArrayInit(0, sizeof(SSDataBlock));
if (pRes == NULL) goto FAIL;
}
} }
atomic_store_8(&pTask->status, TASK_STATUS__IDLE); atomic_store_8(&pTask->status, TASK_STATUS__IDLE);
@ -298,62 +311,66 @@ int32_t streamTaskSink(SStreamTask* pTask, SMsgCb* pMsgCb) {
} }
// dispatch // dispatch
if (pTask->dispatchType == TASK_DISPATCH__INPLACE) { // TODO dispatch guard
SRpcMsg dispatchMsg = {0}; int8_t outputStatus = atomic_load_8(&pTask->outputStatus);
if (streamBuildExecMsg(pTask, pRes, &dispatchMsg, NULL) < 0) { if (outputStatus == TASK_OUTPUT_STATUS__NORMAL) {
ASSERT(0); if (pTask->dispatchType == TASK_DISPATCH__INPLACE) {
return -1; SRpcMsg dispatchMsg = {0};
} if (streamBuildExecMsg(pTask, pRes, &dispatchMsg, NULL) < 0) {
ASSERT(0);
int32_t qType; return -1;
if (pTask->dispatchMsgType == TDMT_VND_TASK_PIPE_EXEC || pTask->dispatchMsgType == TDMT_SND_TASK_PIPE_EXEC) {
qType = FETCH_QUEUE;
} else if (pTask->dispatchMsgType == TDMT_VND_TASK_MERGE_EXEC ||
pTask->dispatchMsgType == TDMT_SND_TASK_MERGE_EXEC) {
qType = MERGE_QUEUE;
} else if (pTask->dispatchMsgType == TDMT_VND_TASK_WRITE_EXEC) {
qType = WRITE_QUEUE;
} else {
ASSERT(0);
}
tmsgPutToQueue(pMsgCb, qType, &dispatchMsg);
} else if (pTask->dispatchType == TASK_DISPATCH__FIXED) {
SRpcMsg dispatchMsg = {0};
SEpSet* pEpSet = NULL;
if (streamBuildExecMsg(pTask, pRes, &dispatchMsg, &pEpSet) < 0) {
ASSERT(0);
return -1;
}
tmsgSendReq(pEpSet, &dispatchMsg);
} else if (pTask->dispatchType == TASK_DISPATCH__SHUFFLE) {
SHashObj* pShuffleRes = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);
if (pShuffleRes == NULL) {
return -1;
}
int32_t sz = taosArrayGetSize(pRes);
for (int32_t i = 0; i < sz; i++) {
SSDataBlock* pDataBlock = taosArrayGet(pRes, i);
SArray* pArray = taosHashGet(pShuffleRes, &pDataBlock->info.groupId, sizeof(int64_t));
if (pArray == NULL) {
pArray = taosArrayInit(0, sizeof(SSDataBlock));
if (pArray == NULL) {
return -1;
}
taosHashPut(pShuffleRes, &pDataBlock->info.groupId, sizeof(int64_t), &pArray, sizeof(void*));
} }
taosArrayPush(pArray, pDataBlock);
}
if (streamShuffleDispatch(pTask, pMsgCb, pShuffleRes) < 0) { int32_t qType;
return -1; if (pTask->dispatchMsgType == TDMT_VND_TASK_PIPE_EXEC || pTask->dispatchMsgType == TDMT_SND_TASK_PIPE_EXEC) {
} qType = FETCH_QUEUE;
} else if (pTask->dispatchMsgType == TDMT_VND_TASK_MERGE_EXEC ||
pTask->dispatchMsgType == TDMT_SND_TASK_MERGE_EXEC) {
qType = MERGE_QUEUE;
} else if (pTask->dispatchMsgType == TDMT_VND_TASK_WRITE_EXEC) {
qType = WRITE_QUEUE;
} else {
ASSERT(0);
}
tmsgPutToQueue(pMsgCb, qType, &dispatchMsg);
} else { } else if (pTask->dispatchType == TASK_DISPATCH__FIXED) {
ASSERT(pTask->dispatchType == TASK_DISPATCH__NONE); SRpcMsg dispatchMsg = {0};
SEpSet* pEpSet = NULL;
if (streamBuildExecMsg(pTask, pRes, &dispatchMsg, &pEpSet) < 0) {
ASSERT(0);
return -1;
}
tmsgSendReq(pEpSet, &dispatchMsg);
} else if (pTask->dispatchType == TASK_DISPATCH__SHUFFLE) {
SHashObj* pShuffleRes = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);
if (pShuffleRes == NULL) {
return -1;
}
int32_t sz = taosArrayGetSize(pRes);
for (int32_t i = 0; i < sz; i++) {
SSDataBlock* pDataBlock = taosArrayGet(pRes, i);
SArray* pArray = taosHashGet(pShuffleRes, &pDataBlock->info.groupId, sizeof(int64_t));
if (pArray == NULL) {
pArray = taosArrayInit(0, sizeof(SSDataBlock));
if (pArray == NULL) {
return -1;
}
taosHashPut(pShuffleRes, &pDataBlock->info.groupId, sizeof(int64_t), &pArray, sizeof(void*));
}
taosArrayPush(pArray, pDataBlock);
}
if (streamShuffleDispatch(pTask, pMsgCb, pShuffleRes) < 0) {
return -1;
}
} else {
ASSERT(pTask->dispatchType == TASK_DISPATCH__NONE);
}
} }
} }
return 0; return 0;
@ -406,11 +423,32 @@ int32_t streamTaskProcessDispatchReq(SStreamTask* pTask, SMsgCb* pMsgCb, SStream
return 0; return 0;
} }
int32_t streamTaskProcessDispatchRsp(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDispatchRsp* pRsp) {
atomic_store_8(&pTask->inputStatus, pRsp->inputStatus);
if (pRsp->inputStatus == TASK_INPUT_STATUS__BLOCKED) {
// TODO: init recover timer
}
// continue dispatch
streamTaskSink(pTask, pMsgCb);
return 0;
}
int32_t streamTaskProcessRunReq(SStreamTask* pTask, SMsgCb* pMsgCb) {
streamTaskExec2(pTask, pMsgCb);
streamTaskSink(pTask, pMsgCb);
return 0;
}
int32_t streamTaskProcessRecoverReq(SStreamTask* pTask, char* msg) { int32_t streamTaskProcessRecoverReq(SStreamTask* pTask, char* msg) {
// //
return 0; return 0;
} }
int32_t streamTaskProcessRecoverRsp(SStreamTask* pTask, char* msg) {
//
return 0;
}
int32_t streamExecTask(SStreamTask* pTask, SMsgCb* pMsgCb, const void* input, int32_t inputType, int32_t workId) { int32_t streamExecTask(SStreamTask* pTask, SMsgCb* pMsgCb, const void* input, int32_t inputType, int32_t workId) {
SArray* pRes = NULL; SArray* pRes = NULL;
// source // source

4
source/libs/sync/src/syncIO.c
View File

@ -183,9 +183,6 @@ static int32_t syncIOStartInternal(SSyncIO *io) {
rpcInit.sessions = 100; rpcInit.sessions = 100;
rpcInit.idleTime = 100; rpcInit.idleTime = 100;
rpcInit.user = "sync-io"; rpcInit.user = "sync-io";
rpcInit.secret = "sync-io";
rpcInit.ckey = "key";
rpcInit.spi = 0;
rpcInit.connType = TAOS_CONN_CLIENT; rpcInit.connType = TAOS_CONN_CLIENT;
io->clientRpc = rpcOpen(&rpcInit); io->clientRpc = rpcOpen(&rpcInit);
@ -206,7 +203,6 @@ static int32_t syncIOStartInternal(SSyncIO *io) {
rpcInit.cfp = syncIOProcessRequest; rpcInit.cfp = syncIOProcessRequest;
rpcInit.sessions = 1000; rpcInit.sessions = 1000;
rpcInit.idleTime = 2 * 1500; rpcInit.idleTime = 2 * 1500;
rpcInit.afp = syncIOAuth;
rpcInit.parent = io; rpcInit.parent = io;
rpcInit.connType = TAOS_CONN_SERVER; rpcInit.connType = TAOS_CONN_SERVER;

12
source/libs/transport/inc/transportInt.h
View File

@ -52,23 +52,15 @@ typedef struct {
int idleTime; // milliseconds; int idleTime; // milliseconds;
uint16_t localPort; uint16_t localPort;
int8_t connType; int8_t connType;
int64_t index;
char label[TSDB_LABEL_LEN]; char label[TSDB_LABEL_LEN];
char user[TSDB_UNI_LEN]; // meter ID
char user[TSDB_UNI_LEN]; // meter ID
char spi; // security parameter index
char encrypt; // encrypt algorithm
char secret[TSDB_PASSWORD_LEN]; // secret for the link
char ckey[TSDB_PASSWORD_LEN]; // ciphering key
void (*cfp)(void* parent, SRpcMsg*, SEpSet*); void (*cfp)(void* parent, SRpcMsg*, SEpSet*);
bool (*retry)(int32_t code); bool (*retry)(int32_t code);
int index;
int32_t refCount; int32_t refCount;
void* parent; void* parent;
void* idPool; // handle to ID pool
void* tmrCtrl; // handle to timer
SHashObj* hash; // handle returned by hash utility
void* tcphandle; // returned handle from TCP initialization void* tcphandle; // returned handle from TCP initialization
TdThreadMutex mutex; TdThreadMutex mutex;
} SRpcInfo; } SRpcInfo;

3
source/libs/transport/src/trans.c
View File

@ -69,9 +69,6 @@ void* rpcOpen(const SRpcInit* pInit) {
if (pInit->user) { if (pInit->user) {
memcpy(pRpc->user, pInit->user, strlen(pInit->user)); memcpy(pRpc->user, pInit->user, strlen(pInit->user));
} }
if (pInit->secret) {
memcpy(pRpc->secret, pInit->secret, strlen(pInit->secret));
}
return pRpc; return pRpc;
} }
void rpcClose(void* arg) { void rpcClose(void* arg) {

1
source/libs/transport/test/pushServer.c
View File

@ -134,7 +134,6 @@ int main(int argc, char *argv[]) {
rpcInit.cfp = processRequestMsg; rpcInit.cfp = processRequestMsg;
rpcInit.sessions = 1000; rpcInit.sessions = 1000;
rpcInit.idleTime = 2 * 1500; rpcInit.idleTime = 2 * 1500;
rpcInit.afp = retrieveAuthInfo;
for (int i = 1; i < argc; ++i) { for (int i = 1; i < argc; ++i) {
if (strcmp(argv[i], "-p") == 0 && i < argc - 1) { if (strcmp(argv[i], "-p") == 0 && i < argc - 1) {

7
source/libs/transport/test/rclient.c
View File

@ -118,9 +118,6 @@ int main(int argc, char *argv[]) {
rpcInit.sessions = 100; rpcInit.sessions = 100;
rpcInit.idleTime = 100; rpcInit.idleTime = 100;
rpcInit.user = "michael"; rpcInit.user = "michael";
rpcInit.secret = secret;
rpcInit.ckey = "key";
rpcInit.spi = 1;
rpcInit.connType = TAOS_CONN_CLIENT; rpcInit.connType = TAOS_CONN_CLIENT;
rpcDebugFlag = 131; rpcDebugFlag = 131;
@ -144,9 +141,7 @@ int main(int argc, char *argv[]) {
} else if (strcmp(argv[i], "-u") == 0 && i < argc - 1) { } else if (strcmp(argv[i], "-u") == 0 && i < argc - 1) {
rpcInit.user = argv[++i]; rpcInit.user = argv[++i];
} else if (strcmp(argv[i], "-k") == 0 && i < argc - 1) { } else if (strcmp(argv[i], "-k") == 0 && i < argc - 1) {
rpcInit.secret = argv[++i];
} else if (strcmp(argv[i], "-spi") == 0 && i < argc - 1) { } else if (strcmp(argv[i], "-spi") == 0 && i < argc - 1) {
rpcInit.spi = atoi(argv[++i]);
} else if (strcmp(argv[i], "-d") == 0 && i < argc - 1) { } else if (strcmp(argv[i], "-d") == 0 && i < argc - 1) {
rpcDebugFlag = atoi(argv[++i]); rpcDebugFlag = atoi(argv[++i]);
} else { } else {
@ -160,8 +155,6 @@ int main(int argc, char *argv[]) {
printf(" [-n requests]: number of requests per thread, default is:%d\n", numOfReqs); printf(" [-n requests]: number of requests per thread, default is:%d\n", numOfReqs);
printf(" [-o compSize]: compression message size, default is:%d\n", tsCompressMsgSize); printf(" [-o compSize]: compression message size, default is:%d\n", tsCompressMsgSize);
printf(" [-u user]: user name for the connection, default is:%s\n", rpcInit.user); printf(" [-u user]: user name for the connection, default is:%s\n", rpcInit.user);
printf(" [-k secret]: password for the connection, default is:%s\n", rpcInit.secret);
printf(" [-spi SPI]: security parameter index, default is:%d\n", rpcInit.spi);
printf(" [-d debugFlag]: debug flag, default:%d\n", rpcDebugFlag); printf(" [-d debugFlag]: debug flag, default:%d\n", rpcDebugFlag);
printf(" [-h help]: print out this help\n\n"); printf(" [-h help]: print out this help\n\n");
exit(0); exit(0);

1
source/libs/transport/test/rserver.c
View File

@ -123,7 +123,6 @@ int main(int argc, char *argv[]) {
rpcInit.cfp = processRequestMsg; rpcInit.cfp = processRequestMsg;
rpcInit.sessions = 1000; rpcInit.sessions = 1000;
rpcInit.idleTime = 2 * 1500; rpcInit.idleTime = 2 * 1500;
rpcInit.afp = retrieveAuthInfo;
rpcDebugFlag = 131; rpcDebugFlag = 131;

25
source/libs/transport/test/syncClient.c
View File

@ -21,15 +21,15 @@
#include "tutil.h" #include "tutil.h"
typedef struct { typedef struct {
int index; int index;
SEpSet epSet; SEpSet epSet;
int num; int num;
int numOfReqs; int numOfReqs;
int msgSize; int msgSize;
tsem_t rspSem; tsem_t rspSem;
tsem_t * pOverSem; tsem_t * pOverSem;
TdThread thread; TdThread thread;
void * pRpc; void * pRpc;
} SInfo; } SInfo;
static void processResponse(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet) { static void processResponse(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet) {
SInfo *pInfo = (SInfo *)pMsg->info.ahandle; SInfo *pInfo = (SInfo *)pMsg->info.ahandle;
@ -103,7 +103,7 @@ int main(int argc, char *argv[]) {
char secret[20] = "mypassword"; char secret[20] = "mypassword";
struct timeval systemTime; struct timeval systemTime;
int64_t startTime, endTime; int64_t startTime, endTime;
TdThreadAttr thattr; TdThreadAttr thattr;
// server info // server info
epSet.inUse = 0; epSet.inUse = 0;
@ -119,9 +119,6 @@ int main(int argc, char *argv[]) {
rpcInit.sessions = 100; rpcInit.sessions = 100;
rpcInit.idleTime = 100; rpcInit.idleTime = 100;
rpcInit.user = "michael"; rpcInit.user = "michael";
rpcInit.secret = secret;
rpcInit.ckey = "key";
rpcInit.spi = 1;
rpcInit.connType = TAOS_CONN_CLIENT; rpcInit.connType = TAOS_CONN_CLIENT;
for (int i = 1; i < argc; ++i) { for (int i = 1; i < argc; ++i) {
@ -144,9 +141,7 @@ int main(int argc, char *argv[]) {
} else if (strcmp(argv[i], "-u") == 0 && i < argc - 1) { } else if (strcmp(argv[i], "-u") == 0 && i < argc - 1) {
rpcInit.user = argv[++i]; rpcInit.user = argv[++i];
} else if (strcmp(argv[i], "-k") == 0 && i < argc - 1) { } else if (strcmp(argv[i], "-k") == 0 && i < argc - 1) {
rpcInit.secret = argv[++i];
} else if (strcmp(argv[i], "-spi") == 0 && i < argc - 1) { } else if (strcmp(argv[i], "-spi") == 0 && i < argc - 1) {
rpcInit.spi = atoi(argv[++i]);
} else if (strcmp(argv[i], "-d") == 0 && i < argc - 1) { } else if (strcmp(argv[i], "-d") == 0 && i < argc - 1) {
rpcDebugFlag = atoi(argv[++i]); rpcDebugFlag = atoi(argv[++i]);
} else { } else {
@ -160,8 +155,6 @@ int main(int argc, char *argv[]) {
printf(" [-n requests]: number of requests per thread, default is:%d\n", numOfReqs); printf(" [-n requests]: number of requests per thread, default is:%d\n", numOfReqs);
printf(" [-o compSize]: compression message size, default is:%d\n", tsCompressMsgSize); printf(" [-o compSize]: compression message size, default is:%d\n", tsCompressMsgSize);
printf(" [-u user]: user name for the connection, default is:%s\n", rpcInit.user); printf(" [-u user]: user name for the connection, default is:%s\n", rpcInit.user);
printf(" [-k secret]: password for the connection, default is:%s\n", rpcInit.secret);
printf(" [-spi SPI]: security parameter index, default is:%d\n", rpcInit.spi);
printf(" [-d debugFlag]: debug flag, default:%d\n", rpcDebugFlag); printf(" [-d debugFlag]: debug flag, default:%d\n", rpcDebugFlag);
printf(" [-h help]: print out this help\n\n"); printf(" [-h help]: print out this help\n\n");
exit(0); exit(0);

6
source/libs/transport/test/transUT.cpp
View File

@ -50,9 +50,6 @@ class Client {
rpcInit_.numOfThreads = nThread; rpcInit_.numOfThreads = nThread;
rpcInit_.cfp = processResp; rpcInit_.cfp = processResp;
rpcInit_.user = (char *)user; rpcInit_.user = (char *)user;
rpcInit_.secret = (char *)secret;
rpcInit_.ckey = (char *)ckey;
rpcInit_.spi = 1;
rpcInit_.parent = this; rpcInit_.parent = this;
rpcInit_.connType = TAOS_CONN_CLIENT; rpcInit_.connType = TAOS_CONN_CLIENT;
this->transCli = rpcOpen(&rpcInit_); this->transCli = rpcOpen(&rpcInit_);
@ -117,9 +114,6 @@ class Server {
rpcInit_.numOfThreads = 5; rpcInit_.numOfThreads = 5;
rpcInit_.cfp = processReq; rpcInit_.cfp = processReq;
rpcInit_.user = (char *)user; rpcInit_.user = (char *)user;
rpcInit_.secret = (char *)secret;
rpcInit_.ckey = (char *)ckey;
rpcInit_.spi = 1;
rpcInit_.connType = TAOS_CONN_SERVER; rpcInit_.connType = TAOS_CONN_SERVER;
} }
void Start() { void Start() {

1112
tests/system-test/7-tmq/subscribeStb.py Normal file
View File

File diff suppressed because it is too large Load Diff

1
tests/system-test/fulltest.sh
View File

@ -64,3 +64,4 @@ python3 ./test.py -f 2-query/nestedQuery.py
python3 ./test.py -f 7-tmq/basic5.py python3 ./test.py -f 7-tmq/basic5.py
python3 ./test.py -f 7-tmq/subscribeDb.py python3 ./test.py -f 7-tmq/subscribeDb.py
python3 ./test.py -f 7-tmq/subscribeDb1.py python3 ./test.py -f 7-tmq/subscribeDb1.py
python3 ./test.py -f 7-tmq/subscribeStb.py

53
tests/test/c/tmqSim.c
View File

@ -98,16 +98,28 @@ static void printHelp() {
exit(EXIT_SUCCESS); exit(EXIT_SUCCESS);
} }
void initLogFile() { char* getCurrentTimeString(char* timeString) {
time_t now; time_t tTime = taosGetTimestampSec();
struct tm curTime; struct tm tm = *taosLocalTime(&tTime, NULL);
char filename[256]; sprintf(timeString, "%d-%02d-%02d %02d:%02d:%02d",
tm.tm_year + 1900,
tm.tm_mon + 1,
tm.tm_mday,
tm.tm_hour,
tm.tm_min,
tm.tm_sec);
now = taosTime(NULL); return timeString;
taosLocalTime(&now, &curTime); }
sprintf(filename, "%s/../log/tmqlog_%04d-%02d-%02d %02d-%02d-%02d.txt", configDir, curTime.tm_year + 1900,
curTime.tm_mon + 1, curTime.tm_mday, curTime.tm_hour, curTime.tm_min, curTime.tm_sec);
// sprintf(filename, "%s/../log/tmqlog.txt", configDir); void initLogFile() {
char filename[256];
char tmpString[128];
sprintf(filename,"%s/../log/tmqlog_%s.txt", configDir, getCurrentTimeString(tmpString));
//sprintf(filename, "%s/../log/tmqlog.txt", configDir);
TdFilePtr pFile = taosOpenFile(filename, TD_FILE_TEXT | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_STREAM); TdFilePtr pFile = taosOpenFile(filename, TD_FILE_TEXT | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_STREAM);
if (NULL == pFile) { if (NULL == pFile) {
fprintf(stderr, "Failed to open %s for save result\n", filename); fprintf(stderr, "Failed to open %s for save result\n", filename);
@ -117,9 +129,6 @@ void initLogFile() {
} }
void saveConfigToLogFile() { void saveConfigToLogFile() {
time_t tTime = taosGetTimestampSec();
struct tm tm = *taosLocalTime(&tTime, NULL);
taosFprintfFile(g_fp, "###################################################################\n"); taosFprintfFile(g_fp, "###################################################################\n");
taosFprintfFile(g_fp, "# configDir: %s\n", configDir); taosFprintfFile(g_fp, "# configDir: %s\n", configDir);
taosFprintfFile(g_fp, "# dbName: %s\n", g_stConfInfo.dbName); taosFprintfFile(g_fp, "# dbName: %s\n", g_stConfInfo.dbName);
@ -144,10 +153,11 @@ void saveConfigToLogFile() {
taosFprintfFile(g_fp, "%s:%s, ", g_stConfInfo.stThreads[i].key[k], g_stConfInfo.stThreads[i].value[k]); taosFprintfFile(g_fp, "%s:%s, ", g_stConfInfo.stThreads[i].key[k], g_stConfInfo.stThreads[i].value[k]);
} }
taosFprintfFile(g_fp, "\n"); taosFprintfFile(g_fp, "\n");
taosFprintfFile(g_fp, " expect rows: %d\n", g_stConfInfo.stThreads[i].expectMsgCnt);
} }
taosFprintfFile(g_fp, "# Test time: %d-%02d-%02d %02d:%02d:%02d\n", tm.tm_year + 1900, tm.tm_mon + 1, char tmpString[128];
tm.tm_mday, tm.tm_hour, tm.tm_min, tm.tm_sec); taosFprintfFile(g_fp, "# Test time: %s\n", getCurrentTimeString(tmpString));
taosFprintfFile(g_fp, "###################################################################\n"); taosFprintfFile(g_fp, "###################################################################\n");
} }
@ -316,10 +326,8 @@ int32_t saveConsumeResult(SThreadInfo* pInfo) {
sprintf(sqlStr, "insert into %s.consumeresult values (now, %d, %" PRId64 ", %" PRId64 ", %d)", g_stConfInfo.cdbName, sprintf(sqlStr, "insert into %s.consumeresult values (now, %d, %" PRId64 ", %" PRId64 ", %d)", g_stConfInfo.cdbName,
pInfo->consumerId, pInfo->consumeMsgCnt, pInfo->consumeRowCnt, pInfo->checkresult); pInfo->consumerId, pInfo->consumeMsgCnt, pInfo->consumeRowCnt, pInfo->checkresult);
time_t tTime = taosGetTimestampSec(); char tmpString[128];
struct tm tm = *taosLocalTime(&tTime, NULL); taosFprintfFile(g_fp, "%s, consume id %d result: %s\n", getCurrentTimeString(tmpString), pInfo->consumerId ,sqlStr);
taosFprintfFile(g_fp, "# save result: %d-%02d-%02d %02d:%02d:%02d, sql: %s\n", tm.tm_year + 1900, tm.tm_mon + 1,
tm.tm_mday, tm.tm_hour, tm.tm_min, tm.tm_sec, sqlStr);
TAOS_RES* pRes = taos_query(pConn, sqlStr); TAOS_RES* pRes = taos_query(pConn, sqlStr);
if (taos_errno(pRes) != 0) { if (taos_errno(pRes) != 0) {
@ -339,6 +347,9 @@ void loop_consume(SThreadInfo* pInfo) {
int64_t totalMsgs = 0; int64_t totalMsgs = 0;
int64_t totalRows = 0; int64_t totalRows = 0;
char tmpString[128];
taosFprintfFile(g_fp, "%s consumer id %d start to loop pull msg\n", getCurrentTimeString(tmpString), pInfo->consumerId);
while (running) { while (running) {
TAOS_RES* tmqMsg = tmq_consumer_poll(pInfo->tmq, g_stConfInfo.consumeDelay * 1000); TAOS_RES* tmqMsg = tmq_consumer_poll(pInfo->tmq, g_stConfInfo.consumeDelay * 1000);
if (tmqMsg) { if (tmqMsg) {
@ -351,11 +362,13 @@ void loop_consume(SThreadInfo* pInfo) {
totalMsgs++; totalMsgs++;
if (totalRows >= pInfo->expectMsgCnt) { if (totalRows >= pInfo->expectMsgCnt) {
taosFprintfFile(g_fp, "==== totalRows >= pInfo->expectMsgCnt, so break\n"); char tmpString[128];
taosFprintfFile(g_fp, "%s over than expect rows, so break consume\n", getCurrentTimeString(tmpString));
break; break;
} }
} else { } else {
taosFprintfFile(g_fp, "==== delay over time, so break\n"); char tmpString[128];
taosFprintfFile(g_fp, "%s no poll more msg when time over, break consume\n", getCurrentTimeString(tmpString));
break; break;
} }
} }

3
tools/shell/src/shellNettest.c
View File

@ -31,9 +31,6 @@ static void shellWorkAsClient() {
rpcInit.connType = TAOS_CONN_CLIENT; rpcInit.connType = TAOS_CONN_CLIENT;
rpcInit.idleTime = tsShellActivityTimer * 1000; rpcInit.idleTime = tsShellActivityTimer * 1000;
rpcInit.user = "_dnd"; rpcInit.user = "_dnd";
rpcInit.ckey = "_key";
rpcInit.spi = 1;
rpcInit.secret = pass;
clientRpc = rpcOpen(&rpcInit); clientRpc = rpcOpen(&rpcInit);
if (clientRpc == NULL) { if (clientRpc == NULL) {