diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 398e46e6a4..ef5a854d20 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -1533,9 +1533,7 @@ typedef struct SMqSetCVgReq { char* sql; char* logicalPlan; char* physicalPlan; - uint32_t qmsgLen; - void* qmsg; - //SSubQueryMsg msg; + char* qmsg; } SMqSetCVgReq; static FORCE_INLINE int32_t tEncodeSSubQueryMsg(void** buf, const SSubQueryMsg* pMsg) { @@ -1567,7 +1565,6 @@ static FORCE_INLINE int32_t tEncodeSMqSetCVgReq(void** buf, const SMqSetCVgReq* tlen += taosEncodeString(buf, pReq->sql); tlen += taosEncodeString(buf, pReq->logicalPlan); tlen += taosEncodeString(buf, pReq->physicalPlan); - tlen += taosEncodeFixedU32(buf, pReq->qmsgLen); tlen += taosEncodeString(buf, (char*)pReq->qmsg); //tlen += tEncodeSSubQueryMsg(buf, &pReq->msg); return tlen; @@ -1582,7 +1579,6 @@ static FORCE_INLINE void* tDecodeSMqSetCVgReq(void* buf, SMqSetCVgReq* pReq) { buf = taosDecodeString(buf, &pReq->sql); buf = taosDecodeString(buf, &pReq->logicalPlan); buf = taosDecodeString(buf, &pReq->physicalPlan); - buf = taosDecodeFixedU32(buf, &pReq->qmsgLen); buf = taosDecodeString(buf, (char**)&pReq->qmsg); //buf = tDecodeSSubQueryMsg(buf, &pReq->msg); return buf; @@ -1600,37 +1596,41 @@ typedef struct SMqColData { int16_t colId; int16_t type; int16_t bytes; - char data[]; -} SMqColData; +} SMqColMeta; typedef struct SMqTbData { int64_t uid; - int32_t numOfCols; int32_t numOfRows; - SMqColData colData[]; + char colData[]; } SMqTbData; typedef struct SMqTopicBlk { - char topicName[TSDB_TOPIC_FNAME_LEN]; - int64_t committedOffset; - int64_t reqOffset; - int64_t rspOffset; - int32_t skipLogNum; - int32_t bodyLen; - int32_t numOfTb; - SMqTbData tbData[]; + char topicName[TSDB_TOPIC_FNAME_LEN]; + int64_t committedOffset; + int64_t reqOffset; + int64_t rspOffset; + int32_t skipLogNum; + int32_t bodyLen; + int32_t numOfTb; + SMqTbData* tbData; } SMqTopicData; typedef struct SMqConsumeRsp { - int64_t reqId; - int64_t consumerId; - int32_t bodyLen; - int32_t numOfTopics; - SMqTopicData data[]; + int64_t consumerId; + int32_t numOfCols; + SMqColMeta* meta; + int32_t numOfTopics; + SMqTopicData* data; } SMqConsumeRsp; +static FORCE_INLINE int32_t tEncodeSMqConsumeRsp(void** buf, const SMqConsumeRsp* pRsp) { + int32_t tlen = 0; + return tlen; +} + // one req for one vg+topic typedef struct SMqConsumeReq { + SMsgHead head; //0: commit only, current offset //1: consume only, poll next offset //2: commit current and consume next offset @@ -1663,7 +1663,7 @@ typedef struct SMqCMGetSubEpRsp { static FORCE_INLINE int32_t tEncodeSMqSubVgEp(void** buf, const SMqSubVgEp* pVgEp) { int32_t tlen = 0; - tlen += taosEncodeFixedI16(buf, pVgEp->vgId); + tlen += taosEncodeFixedI32(buf, pVgEp->vgId); tlen += taosEncodeSEpSet(buf, &pVgEp->epSet); return tlen; } diff --git a/include/libs/executor/executor.h b/include/libs/executor/executor.h index d2e602a5d6..26733696d2 100644 --- a/include/libs/executor/executor.h +++ b/include/libs/executor/executor.h @@ -24,6 +24,7 @@ extern "C" { typedef void* qTaskInfo_t; typedef void* DataSinkHandle; +struct SRpcMsg; struct SSubplan; /** @@ -208,6 +209,8 @@ void** qReleaseTask(void* pMgmt, void* pQInfo, bool freeHandle); */ void** qDeregisterQInfo(void* pMgmt, void* pQInfo); +void qProcessFetchRsp(void* parent, struct SRpcMsg* pMsg, struct SEpSet* pEpSet); + #ifdef __cplusplus } #endif diff --git a/include/libs/qworker/qworker.h b/include/libs/qworker/qworker.h index 5d815d15e0..5e3320ffdb 100644 --- a/include/libs/qworker/qworker.h +++ b/include/libs/qworker/qworker.h @@ -49,9 +49,10 @@ typedef struct { } SQWorkerStat; typedef int32_t (*putReqToQueryQFp)(void *, struct SRpcMsg *); +typedef int32_t (*sendReqToDnodeFp)(void *, struct SEpSet *, struct SRpcMsg *); - -int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qWorkerMgmt, void *nodeObj, putReqToQueryQFp fp); +int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qWorkerMgmt, void *nodeObj, + putReqToQueryQFp fp1, sendReqToDnodeFp fp2); int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg); @@ -65,6 +66,8 @@ int32_t qWorkerProcessStatusMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg); int32_t qWorkerProcessFetchMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg); +int32_t qWorkerProcessFetchRsp(void *node, void *qWorkerMgmt, SRpcMsg *pMsg); + int32_t qWorkerProcessCancelMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg); int32_t qWorkerProcessDropMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg); diff --git a/include/libs/wal/wal.h b/include/libs/wal/wal.h index 641b485f4c..45f1d88c30 100644 --- a/include/libs/wal/wal.h +++ b/include/libs/wal/wal.h @@ -19,6 +19,7 @@ #include "tarray.h" #include "tdef.h" #include "tlog.h" +#include "tmsg.h" #ifdef __cplusplus extern "C" { #endif @@ -159,7 +160,7 @@ int32_t walAlter(SWal *, SWalCfg *pCfg); void walClose(SWal *); // write -int64_t walWrite(SWal *, int64_t index, uint8_t msgType, const void *body, int32_t bodyLen); +int64_t walWrite(SWal *, int64_t index, tmsg_t msgType, const void *body, int32_t bodyLen); void walFsync(SWal *, bool force); // apis for lifecycle management diff --git a/include/util/taoserror.h b/include/util/taoserror.h index 69bf085491..8f49fce558 100644 --- a/include/util/taoserror.h +++ b/include/util/taoserror.h @@ -253,6 +253,7 @@ int32_t* taosGetErrno(); #define TSDB_CODE_MND_INVALID_TOPIC_OPTION TAOS_DEF_ERROR_CODE(0, 0x03E4) #define TSDB_CODE_MND_TOPIC_OPTION_UNCHNAGED TAOS_DEF_ERROR_CODE(0, 0x03E5) #define TSDB_CODE_MND_NAME_CONFLICT_WITH_STB TAOS_DEF_ERROR_CODE(0, 0x03E6) +#define TSDB_CODE_MND_CONSUMER_NOT_EXIST TAOS_DEF_ERROR_CODE(0, 0x03E7) // dnode #define TSDB_CODE_DND_ACTION_IN_PROGRESS TAOS_DEF_ERROR_CODE(0, 0x0400) diff --git a/source/client/src/clientImpl.c b/source/client/src/clientImpl.c index 28a4e0c87d..df086463db 100644 --- a/source/client/src/clientImpl.c +++ b/source/client/src/clientImpl.c @@ -328,6 +328,7 @@ struct tmq_t { char clientId[256]; int64_t consumerId; int64_t status; + tsem_t rspSem; STscObj* pTscObj; tmq_commit_cb* commit_cb; int32_t nextTopicIdx; @@ -344,6 +345,7 @@ tmq_t* taos_consumer_new(void* conn, tmq_conf_t* conf, char* errstr, int32_t err strcpy(pTmq->clientId, conf->clientId); strcpy(pTmq->groupId, conf->groupId); pTmq->commit_cb = conf->commit_cb; + tsem_init(&pTmq->rspSem, 0, 0); pTmq->consumerId = generateRequestId() & ((uint64_t)-1 >> 1); pTmq->clientTopics = taosArrayInit(0, sizeof(SMqClientTopic)); return pTmq; @@ -372,11 +374,27 @@ int32_t tmq_list_append(tmq_list_t* ptr, char* src) { } +int32_t tmq_null_cb(void* param, const SDataBuf* pMsg, int32_t code) { + if (code == 0) { + // + } + // + return 0; +} + TAOS_RES* tmq_subscribe(tmq_t* tmq, tmq_list_t* topic_list) { SRequestObj *pRequest = NULL; - tmq->status = 1; int32_t sz = topic_list->cnt; - tmq->clientTopics = taosArrayInit(sz, sizeof(void*)); + //destroy ex + taosArrayDestroy(tmq->clientTopics); + tmq->clientTopics = taosArrayInit(sz, sizeof(SMqClientTopic)); + + SCMSubscribeReq req; + req.topicNum = sz; + req.consumerId = tmq->consumerId; + req.consumerGroup = strdup(tmq->groupId); + req.topicNames = taosArrayInit(sz, sizeof(void*)); + for (int i = 0; i < sz; i++) { char* topicName = topic_list->elems[i]; @@ -391,16 +409,21 @@ TAOS_RES* tmq_subscribe(tmq_t* tmq, tmq_list_t* topic_list) { } tNameExtractFullName(&name, topicFname); tscDebug("subscribe topic: %s", topicFname); - taosArrayPush(tmq->clientTopics, &topicFname); + SMqClientTopic topic = { + .nextVgIdx = 0, + .sql = NULL, + .sqlLen = 0, + .topicId = 0, + .topicName = topicFname, + .vgs = NULL + }; + topic.vgs = taosArrayInit(0, sizeof(SMqClientVg)); + taosArrayPush(tmq->clientTopics, &topic); /*SMqClientTopic topic = {*/ /*.*/ /*};*/ + taosArrayPush(req.topicNames, &topicFname); } - SCMSubscribeReq req; - req.topicNum = taosArrayGetSize(tmq->clientTopics); - req.consumerId = tmq->consumerId; - req.consumerGroup = strdup(tmq->groupId); - req.topicNames = tmq->clientTopics; int tlen = tSerializeSCMSubscribeReq(NULL, &req); void* buf = malloc(tlen); @@ -419,18 +442,19 @@ TAOS_RES* tmq_subscribe(tmq_t* tmq, tmq_list_t* topic_list) { pRequest->body.requestMsg = (SDataBuf){ .pData = buf, .len = tlen }; - SMsgSendInfo* body = buildMsgInfoImpl(pRequest); + SMsgSendInfo* sendInfo = buildMsgInfoImpl(pRequest); + /*sendInfo->fp*/ SEpSet epSet = getEpSet_s(&tmq->pTscObj->pAppInfo->mgmtEp); int64_t transporterId = 0; - asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, body); + asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, sendInfo); tsem_wait(&pRequest->body.rspSem); _return: - if (body != NULL) { - destroySendMsgInfo(body); - } + /*if (sendInfo != NULL) {*/ + /*destroySendMsgInfo(sendInfo);*/ + /*}*/ if (pRequest != NULL && terrno != TSDB_CODE_SUCCESS) { pRequest->code = terrno; @@ -569,19 +593,19 @@ TAOS_RES *taos_create_topic(TAOS* taos, const char* topicName, const char* sql, pRequest->body.requestMsg = (SDataBuf){ .pData = buf, .len = tlen }; pRequest->type = TDMT_MND_CREATE_TOPIC; - SMsgSendInfo* body = buildMsgInfoImpl(pRequest); + SMsgSendInfo* sendInfo = buildMsgInfoImpl(pRequest); SEpSet epSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp); int64_t transporterId = 0; - asyncSendMsgToServer(pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, body); + asyncSendMsgToServer(pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, sendInfo); tsem_wait(&pRequest->body.rspSem); _return: qDestroyQuery(pQueryNode); - if (body != NULL) { - destroySendMsgInfo(body); - } + /*if (sendInfo != NULL) {*/ + /*destroySendMsgInfo(sendInfo);*/ + /*}*/ if (pRequest != NULL && terrno != TSDB_CODE_SUCCESS) { pRequest->code = terrno; @@ -602,6 +626,12 @@ int32_t tmq_poll_cb_inner(void* param, const SDataBuf* pMsg, int32_t code) { int32_t tmq_ask_ep_cb(void* param, const SDataBuf* pMsg, int32_t code) { tmq_t* tmq = (tmq_t*)param; + if (code != 0) { + tsem_post(&tmq->rspSem); + return 0; + } + tscDebug("tmq ask ep cb called"); + bool set = false; SMqCMGetSubEpRsp rsp; tDecodeSMqCMGetSubEpRsp(pMsg->pData, &rsp); int32_t sz = taosArrayGetSize(rsp.topics); @@ -616,38 +646,35 @@ int32_t tmq_ask_ep_cb(void* param, const SDataBuf* pMsg, int32_t code) { for (int32_t j = 0; j < vgSz; j++) { SMqSubVgEp* pVgEp = taosArrayGet(pTopicEp->vgs, j); SMqClientVg clientVg = { + .pollCnt = 0, + .committedOffset = -1, + .currentOffset = -1, .vgId = pVgEp->vgId, .epSet = pVgEp->epSet }; taosArrayPush(topic.vgs, &clientVg); + set = true; } taosArrayPush(tmq->clientTopics, &topic); } + if(set) tmq->status = 1; // unlock + tsem_post(&tmq->rspSem); return 0; } tmq_message_t* tmq_consume_poll(tmq_t* tmq, int64_t blocking_time) { - if (tmq->clientTopics == NULL || taosArrayGetSize(tmq->clientTopics) == 0) { - return NULL; - } - SRequestObj *pRequest = NULL; - SMqConsumeReq req = {0}; - req.reqType = 1; - req.blockingTime = blocking_time; - req.consumerId = tmq->consumerId; - tmq_message_t* tmq_message = NULL; - strcpy(req.cgroup, tmq->groupId); - if (taosArrayGetSize(tmq->clientTopics) == 0) { + if (taosArrayGetSize(tmq->clientTopics) == 0 || tmq->status == 0) { int32_t tlen = sizeof(SMqCMGetSubEpReq); SMqCMGetSubEpReq* buf = malloc(tlen); if (buf == NULL) { tscError("failed to malloc get subscribe ep buf"); } - buf->consumerId = htobe64(buf->consumerId); + buf->consumerId = htobe64(tmq->consumerId); + strcpy(buf->cgroup, tmq->groupId); - pRequest = createRequest(tmq->pTscObj, NULL, NULL, TDMT_MND_GET_SUB_EP); + SRequestObj *pRequest = createRequest(tmq->pTscObj, NULL, NULL, TDMT_MND_GET_SUB_EP); if (pRequest == NULL) { tscError("failed to malloc subscribe ep request"); } @@ -664,24 +691,39 @@ tmq_message_t* tmq_consume_poll(tmq_t* tmq, int64_t blocking_time) { int64_t transporterId = 0; asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, sendInfo); - tsem_wait(&pRequest->body.rspSem); + tsem_wait(&tmq->rspSem); } - SMqClientTopic* pTopic = taosArrayGetP(tmq->clientTopics, tmq->nextTopicIdx); + if (taosArrayGetSize(tmq->clientTopics) == 0) { + tscDebug("consumer:%ld poll but not assigned", tmq->consumerId); + return NULL; + } + + SMqConsumeReq* pReq = malloc(sizeof(SMqConsumeReq)); + pReq->reqType = 1; + pReq->blockingTime = blocking_time; + pReq->consumerId = tmq->consumerId; + tmq_message_t* tmq_message = NULL; + strcpy(pReq->cgroup, tmq->groupId); + + SMqClientTopic* pTopic = taosArrayGet(tmq->clientTopics, tmq->nextTopicIdx); tmq->nextTopicIdx = (tmq->nextTopicIdx + 1) % taosArrayGetSize(tmq->clientTopics); - strcpy(req.topic, pTopic->topicName); + strcpy(pReq->topic, pTopic->topicName); int32_t nextVgIdx = pTopic->nextVgIdx; pTopic->nextVgIdx = (nextVgIdx + 1) % taosArrayGetSize(pTopic->vgs); SMqClientVg* pVg = taosArrayGet(pTopic->vgs, nextVgIdx); - req.offset = pVg->currentOffset; + pReq->offset = pVg->currentOffset; - pRequest->body.requestMsg = (SDataBuf){ .pData = &req, .len = sizeof(SMqConsumeReq) }; - pRequest->type = TDMT_VND_CONSUME; + pReq->head.vgId = htonl(pVg->vgId); + pReq->head.contLen = htonl(sizeof(SMqConsumeReq)); + + SRequestObj* pRequest = createRequest(tmq->pTscObj, NULL, NULL, TDMT_VND_CONSUME); + pRequest->body.requestMsg = (SDataBuf){ .pData = pReq, .len = sizeof(SMqConsumeReq) }; SMsgSendInfo* sendInfo = buildMsgInfoImpl(pRequest); - sendInfo->requestObjRefId = 0; - sendInfo->param = &tmq_message; - sendInfo->fp = tmq_poll_cb_inner; + /*sendInfo->requestObjRefId = 0;*/ + /*sendInfo->param = &tmq_message;*/ + /*sendInfo->fp = tmq_poll_cb_inner;*/ int64_t transporterId = 0; asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &pVg->epSet, &transporterId, sendInfo); diff --git a/source/client/test/clientTests.cpp b/source/client/test/clientTests.cpp index a1adf58f6a..b73079741c 100644 --- a/source/client/test/clientTests.cpp +++ b/source/client/test/clientTests.cpp @@ -536,7 +536,7 @@ TEST(testCase, create_topic_Test) { if (taos_errno(pRes) != 0) { printf("error in use db, reason:%s\n", taos_errstr(pRes)); } - taos_free_result(pRes); + //taos_free_result(pRes); TAOS_FIELD* pFields = taos_fetch_fields(pRes); ASSERT_TRUE(pFields == nullptr); @@ -570,30 +570,32 @@ TEST(testCase, create_topic_Test) { //taos_close(pConn); //} -//TEST(testCase, tmq_subscribe_Test) { - //TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0); - //assert(pConn != NULL); +#if 0 +TEST(testCase, tmq_subscribe_Test) { + TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0); + assert(pConn != NULL); - //TAOS_RES* pRes = taos_query(pConn, "use abc1"); - //if (taos_errno(pRes) != 0) { - //printf("error in use db, reason:%s\n", taos_errstr(pRes)); - //} - //taos_free_result(pRes); + TAOS_RES* pRes = taos_query(pConn, "use abc1"); + if (taos_errno(pRes) != 0) { + printf("error in use db, reason:%s\n", taos_errstr(pRes)); + } + taos_free_result(pRes); - //tmq_conf_t* conf = tmq_conf_new(); - //tmq_conf_set(conf, "group.id", "tg1"); - //tmq_t* tmq = taos_consumer_new(pConn, conf, NULL, 0); + tmq_conf_t* conf = tmq_conf_new(); + tmq_conf_set(conf, "group.id", "tg1"); + tmq_t* tmq = taos_consumer_new(pConn, conf, NULL, 0); - //tmq_list_t* topic_list = tmq_list_new(); - //tmq_list_append(topic_list, "test_topic_1"); - //tmq_subscribe(tmq, topic_list); + tmq_list_t* topic_list = tmq_list_new(); + tmq_list_append(topic_list, "test_topic_1"); + tmq_subscribe(tmq, topic_list); - //while (1) { - //tmq_message_t* msg = tmq_consume_poll(tmq, 0); - //printf("get msg\n"); + while (1) { + tmq_message_t* msg = tmq_consume_poll(tmq, 0); + printf("get msg\n"); //if (msg == NULL) break; - //} -//} + } +} +#endif TEST(testCase, tmq_consume_Test) { } diff --git a/source/dnode/mgmt/impl/src/dndEnv.c b/source/dnode/mgmt/impl/src/dndEnv.c index 74fb5f1437..02dced53c2 100644 --- a/source/dnode/mgmt/impl/src/dndEnv.c +++ b/source/dnode/mgmt/impl/src/dndEnv.c @@ -289,6 +289,7 @@ int32_t dndInit(const SDnodeEnvCfg *pCfg) { .charset = pCfg->charset, .nthreads = pCfg->numOfCommitThreads, .putReqToVQueryQFp = dndPutReqToVQueryQ, + .sendReqToDnodeFp = dndSendReqToDnode }; if (vnodeInit(&vnodeOpt) != 0) { diff --git a/source/dnode/mgmt/impl/src/dndMgmt.c b/source/dnode/mgmt/impl/src/dndMgmt.c index b127fb1d64..dce5c01895 100644 --- a/source/dnode/mgmt/impl/src/dndMgmt.c +++ b/source/dnode/mgmt/impl/src/dndMgmt.c @@ -529,7 +529,7 @@ int32_t dndInitMgmt(SDnode *pDnode) { } if (pMgmt->dropped) { - dError("dnode will not start for its already dropped"); + dError("dnode not start since its already dropped"); return -1; } diff --git a/source/dnode/mgmt/impl/src/dndTransport.c b/source/dnode/mgmt/impl/src/dndTransport.c index 275b09fc0f..257c72bff5 100644 --- a/source/dnode/mgmt/impl/src/dndTransport.c +++ b/source/dnode/mgmt/impl/src/dndTransport.c @@ -122,6 +122,7 @@ static void dndInitMsgFp(STransMgmt *pMgmt) { pMgmt->msgFp[TMSG_INDEX(TDMT_VND_QUERY)] = dndProcessVnodeQueryMsg; pMgmt->msgFp[TMSG_INDEX(TDMT_VND_QUERY_CONTINUE)] = dndProcessVnodeQueryMsg; pMgmt->msgFp[TMSG_INDEX(TDMT_VND_FETCH)] = dndProcessVnodeFetchMsg; + pMgmt->msgFp[TMSG_INDEX(TDMT_VND_FETCH_RSP)] = dndProcessVnodeFetchMsg; pMgmt->msgFp[TMSG_INDEX(TDMT_VND_ALTER_TABLE)] = dndProcessVnodeWriteMsg; pMgmt->msgFp[TMSG_INDEX(TDMT_VND_UPDATE_TAG_VAL)] = dndProcessVnodeWriteMsg; pMgmt->msgFp[TMSG_INDEX(TDMT_VND_TABLE_META)] = dndProcessVnodeFetchMsg; @@ -148,6 +149,7 @@ static void dndInitMsgFp(STransMgmt *pMgmt) { pMgmt->msgFp[TMSG_INDEX(TDMT_VND_SHOW_TABLES_FETCH)] = dndProcessVnodeFetchMsg; pMgmt->msgFp[TMSG_INDEX(TDMT_VND_MQ_SET_CONN)] = dndProcessVnodeWriteMsg; pMgmt->msgFp[TMSG_INDEX(TDMT_VND_MQ_SET_CUR)] = dndProcessVnodeFetchMsg; + pMgmt->msgFp[TMSG_INDEX(TDMT_VND_CONSUME)] = dndProcessVnodeFetchMsg; } static void dndProcessResponse(void *parent, SRpcMsg *pRsp, SEpSet *pEpSet) { diff --git a/source/dnode/mgmt/impl/src/dndVnodes.c b/source/dnode/mgmt/impl/src/dndVnodes.c index dcb73d13c7..c27f3a151e 100644 --- a/source/dnode/mgmt/impl/src/dndVnodes.c +++ b/source/dnode/mgmt/impl/src/dndVnodes.c @@ -527,7 +527,6 @@ static void dndGenerateVnodeCfg(SCreateVnodeReq *pCreate, SVnodeCfg *pCfg) { pCfg->vgId = pCreate->vgId; pCfg->wsize = pCreate->cacheBlockSize; pCfg->ssize = pCreate->cacheBlockSize; - pCfg->wsize = pCreate->cacheBlockSize; pCfg->lsize = pCreate->cacheBlockSize; pCfg->isHeapAllocator = true; pCfg->ttl = 4; diff --git a/source/dnode/mnode/impl/inc/mndDef.h b/source/dnode/mnode/impl/inc/mndDef.h index 02ce3a1591..5ec9173fc8 100644 --- a/source/dnode/mnode/impl/inc/mndDef.h +++ b/source/dnode/mnode/impl/inc/mndDef.h @@ -363,9 +363,7 @@ typedef struct SMqConsumerEp { int64_t consumerId; // -1 for unassigned int64_t lastConsumerHbTs; int64_t lastVgHbTs; - uint32_t qmsgLen; char* qmsg; - //SSubQueryMsg qExec; } SMqConsumerEp; static FORCE_INLINE int32_t tEncodeSMqConsumerEp(void** buf, SMqConsumerEp* pConsumerEp) { @@ -374,9 +372,10 @@ static FORCE_INLINE int32_t tEncodeSMqConsumerEp(void** buf, SMqConsumerEp* pCon tlen += taosEncodeFixedI32(buf, pConsumerEp->status); tlen += taosEncodeSEpSet(buf, &pConsumerEp->epSet); tlen += taosEncodeFixedI64(buf, pConsumerEp->consumerId); + tlen += taosEncodeFixedI64(buf, pConsumerEp->lastConsumerHbTs); + tlen += taosEncodeFixedI64(buf, pConsumerEp->lastVgHbTs); //tlen += tEncodeSSubQueryMsg(buf, &pConsumerEp->qExec); - tlen += taosEncodeFixedU32(buf, pConsumerEp->qmsgLen); - tlen += taosEncodeBinary(buf, pConsumerEp->qmsg, pConsumerEp->qmsgLen); + tlen += taosEncodeString(buf, pConsumerEp->qmsg); return tlen; } @@ -385,9 +384,10 @@ static FORCE_INLINE void* tDecodeSMqConsumerEp(void** buf, SMqConsumerEp* pConsu buf = taosDecodeFixedI32(buf, &pConsumerEp->status); buf = taosDecodeSEpSet(buf, &pConsumerEp->epSet); buf = taosDecodeFixedI64(buf, &pConsumerEp->consumerId); + buf = taosDecodeFixedI64(buf, &pConsumerEp->lastConsumerHbTs); + buf = taosDecodeFixedI64(buf, &pConsumerEp->lastVgHbTs); //buf = tDecodeSSubQueryMsg(buf, &pConsumerEp->qExec); - buf = taosDecodeFixedU32(buf, &pConsumerEp->qmsgLen); - buf = taosDecodeBinary(buf, (void**)&pConsumerEp->qmsg, pConsumerEp->qmsgLen); + buf = taosDecodeString(buf, &pConsumerEp->qmsg); return buf; } @@ -423,18 +423,27 @@ static FORCE_INLINE SMqSubscribeObj* tNewSubscribeObj() { free(pSub); return NULL; } - pSub->idleConsumer = taosArrayInit(0, sizeof(SMqConsumerEp)); - if (pSub->assigned == NULL) { + pSub->lostConsumer = taosArrayInit(0, sizeof(SMqConsumerEp)); + if (pSub->lostConsumer == NULL) { taosArrayDestroy(pSub->availConsumer); - taosArrayDestroy(pSub->idleConsumer); + taosArrayDestroy(pSub->assigned); + free(pSub); + return NULL; + } + pSub->idleConsumer = taosArrayInit(0, sizeof(SMqConsumerEp)); + if (pSub->idleConsumer == NULL) { + taosArrayDestroy(pSub->availConsumer); + taosArrayDestroy(pSub->assigned); + taosArrayDestroy(pSub->lostConsumer); free(pSub); return NULL; } pSub->unassignedVg = taosArrayInit(0, sizeof(SMqConsumerEp)); - if (pSub->assigned == NULL) { + if (pSub->unassignedVg == NULL) { taosArrayDestroy(pSub->availConsumer); + taosArrayDestroy(pSub->assigned); + taosArrayDestroy(pSub->lostConsumer); taosArrayDestroy(pSub->idleConsumer); - taosArrayDestroy(pSub->unassignedVg); free(pSub); return NULL; } @@ -461,6 +470,13 @@ static FORCE_INLINE int32_t tEncodeSubscribeObj(void** buf, const SMqSubscribeOb tlen += tEncodeSMqConsumerEp(buf, pCEp); } + sz = taosArrayGetSize(pSub->lostConsumer); + tlen += taosEncodeFixedI32(buf, sz); + for (int32_t i = 0; i < sz; i++) { + SMqConsumerEp* pCEp = taosArrayGet(pSub->lostConsumer, i); + tlen += tEncodeSMqConsumerEp(buf, pCEp); + } + sz = taosArrayGetSize(pSub->idleConsumer); tlen += taosEncodeFixedI32(buf, sz); for (int32_t i = 0; i < sz; i++) { @@ -485,20 +501,47 @@ static FORCE_INLINE void* tDecodeSubscribeObj(void* buf, SMqSubscribeObj* pSub) int32_t sz; buf = taosDecodeFixedI32(buf, &sz); - pSub->assigned = taosArrayInit(sz, sizeof(int64_t)); - if (pSub->assigned == NULL) { + pSub->availConsumer = taosArrayInit(sz, sizeof(int64_t)); + if (pSub->availConsumer == NULL) { return NULL; } for (int32_t i = 0; i < sz; i++) { int64_t consumerId; buf = taosDecodeFixedI64(buf, &consumerId); - taosArrayPush(pSub->assigned, &consumerId); + taosArrayPush(pSub->availConsumer, &consumerId); + } + + buf = taosDecodeFixedI32(buf, &sz); + pSub->assigned = taosArrayInit(sz, sizeof(SMqConsumerEp)); + if (pSub->assigned == NULL) { + taosArrayDestroy(pSub->availConsumer); + return NULL; + } + for (int32_t i = 0; i < sz; i++) { + SMqConsumerEp cEp; + buf = tDecodeSMqConsumerEp(buf, &cEp); + taosArrayPush(pSub->assigned, &cEp); + } + + buf = taosDecodeFixedI32(buf, &sz); + pSub->lostConsumer = taosArrayInit(sz, sizeof(SMqConsumerEp)); + if (pSub->lostConsumer == NULL) { + taosArrayDestroy(pSub->availConsumer); + taosArrayDestroy(pSub->assigned); + return NULL; + } + for (int32_t i = 0; i < sz; i++) { + SMqConsumerEp cEp; + buf = tDecodeSMqConsumerEp(buf, &cEp); + taosArrayPush(pSub->lostConsumer, &cEp); } buf = taosDecodeFixedI32(buf, &sz); pSub->idleConsumer = taosArrayInit(sz, sizeof(SMqConsumerEp)); if (pSub->idleConsumer == NULL) { + taosArrayDestroy(pSub->availConsumer); taosArrayDestroy(pSub->assigned); + taosArrayDestroy(pSub->lostConsumer); return NULL; } for (int32_t i = 0; i < sz; i++) { @@ -507,10 +550,13 @@ static FORCE_INLINE void* tDecodeSubscribeObj(void* buf, SMqSubscribeObj* pSub) taosArrayPush(pSub->idleConsumer, &cEp); } + buf = taosDecodeFixedI32(buf, &sz); pSub->unassignedVg = taosArrayInit(sz, sizeof(SMqConsumerEp)); if (pSub->unassignedVg == NULL) { + taosArrayDestroy(pSub->availConsumer); taosArrayDestroy(pSub->assigned); + taosArrayDestroy(pSub->lostConsumer); taosArrayDestroy(pSub->idleConsumer); return NULL; } @@ -580,7 +626,10 @@ static FORCE_INLINE int32_t tEncodeSMqConsumerTopic(void** buf, SMqConsumerTopic int32_t tlen = 0; tlen += taosEncodeString(buf, pConsumerTopic->name); tlen += taosEncodeFixedI32(buf, pConsumerTopic->epoch); - int32_t sz = taosArrayGetSize(pConsumerTopic->pVgInfo); + int32_t sz = 0; + if (pConsumerTopic->pVgInfo != NULL) { + sz = taosArrayGetSize(pConsumerTopic->pVgInfo); + } tlen += taosEncodeFixedI32(buf, sz); for (int32_t i = 0; i < sz; i++) { int32_t* pVgInfo = taosArrayGet(pConsumerTopic->pVgInfo, i); diff --git a/source/dnode/mnode/impl/src/mndDb.c b/source/dnode/mnode/impl/src/mndDb.c index 39be41a4e5..32088d84a9 100644 --- a/source/dnode/mnode/impl/src/mndDb.c +++ b/source/dnode/mnode/impl/src/mndDb.c @@ -226,10 +226,10 @@ static int32_t mndCheckDbCfg(SMnode *pMnode, SDbCfg *pCfg) { if (pCfg->cacheBlockSize < TSDB_MIN_CACHE_BLOCK_SIZE || pCfg->cacheBlockSize > TSDB_MAX_CACHE_BLOCK_SIZE) return -1; if (pCfg->totalBlocks < TSDB_MIN_TOTAL_BLOCKS || pCfg->totalBlocks > TSDB_MAX_TOTAL_BLOCKS) return -1; if (pCfg->daysPerFile < TSDB_MIN_DAYS_PER_FILE || pCfg->daysPerFile > TSDB_MAX_DAYS_PER_FILE) return -1; - if (pCfg->daysToKeep0 < pCfg->daysPerFile) return -1; if (pCfg->daysToKeep0 < TSDB_MIN_KEEP || pCfg->daysToKeep0 > TSDB_MAX_KEEP) return -1; if (pCfg->daysToKeep1 < TSDB_MIN_KEEP || pCfg->daysToKeep1 > TSDB_MAX_KEEP) return -1; if (pCfg->daysToKeep2 < TSDB_MIN_KEEP || pCfg->daysToKeep2 > TSDB_MAX_KEEP) return -1; + if (pCfg->daysToKeep0 < pCfg->daysPerFile) return -1; if (pCfg->daysToKeep0 > pCfg->daysToKeep1) return -1; if (pCfg->daysToKeep1 > pCfg->daysToKeep2) return -1; if (pCfg->minRows < TSDB_MIN_MIN_ROW_FBLOCK || pCfg->minRows > TSDB_MAX_MIN_ROW_FBLOCK) return -1; @@ -498,7 +498,7 @@ static int32_t mndProcessCreateDbReq(SMnodeMsg *pReq) { return TSDB_CODE_MND_ACTION_IN_PROGRESS; } -static int32_t mndSetDbCfgFromAlterDbMsg(SDbObj *pDb, SAlterDbReq *pAlter) { +static int32_t mndSetDbCfgFromAlterDbReq(SDbObj *pDb, SAlterDbReq *pAlter) { terrno = TSDB_CODE_MND_DB_OPTION_UNCHANGED; if (pAlter->totalBlocks >= 0 && pAlter->totalBlocks != pDb->cfg.totalBlocks) { @@ -649,7 +649,7 @@ static int32_t mndProcessAlterDbReq(SMnodeMsg *pReq) { SDbObj dbObj = {0}; memcpy(&dbObj, pDb, sizeof(SDbObj)); - int32_t code = mndSetDbCfgFromAlterDbMsg(&dbObj, pAlter); + int32_t code = mndSetDbCfgFromAlterDbReq(&dbObj, pAlter); if (code != 0) { mndReleaseDb(pMnode, pDb); mError("db:%s, failed to alter since %s", pAlter->db, tstrerror(code)); @@ -1143,7 +1143,7 @@ static int32_t mndRetrieveDbs(SMnodeMsg *pReq, SShowObj *pShow, char *data, int3 prec = TSDB_TIME_PRECISION_NANO_STR; break; default: - assert(false); + prec = "none"; break; } STR_WITH_SIZE_TO_VARSTR(pWrite, prec, 2); diff --git a/source/dnode/mnode/impl/src/mndFunc.c b/source/dnode/mnode/impl/src/mndFunc.c index f9dde7bc75..0e086fdb22 100644 --- a/source/dnode/mnode/impl/src/mndFunc.c +++ b/source/dnode/mnode/impl/src/mndFunc.c @@ -20,6 +20,7 @@ #include "mndTrans.h" #define SDB_FUNC_VER 1 +#define SDB_FUNC_RESERVE_SIZE 64 static SSdbRaw *mndFuncActionEncode(SFuncObj *pFunc); static SSdbRow *mndFuncActionDecode(SSdbRaw *pRaw); @@ -60,7 +61,7 @@ void mndCleanupFunc(SMnode *pMnode) {} static SSdbRaw *mndFuncActionEncode(SFuncObj *pFunc) { terrno = TSDB_CODE_OUT_OF_MEMORY; - int32_t size = pFunc->commentSize + pFunc->codeSize + sizeof(SFuncObj); + int32_t size = pFunc->commentSize + pFunc->codeSize + sizeof(SFuncObj) + SDB_FUNC_RESERVE_SIZE; SSdbRaw *pRaw = sdbAllocRaw(SDB_FUNC, SDB_FUNC_VER, size); if (pRaw == NULL) goto FUNC_ENCODE_OVER; @@ -78,6 +79,7 @@ static SSdbRaw *mndFuncActionEncode(SFuncObj *pFunc) { SDB_SET_INT32(pRaw, dataPos, pFunc->codeSize, FUNC_ENCODE_OVER) SDB_SET_BINARY(pRaw, dataPos, pFunc->pComment, pFunc->commentSize, FUNC_ENCODE_OVER) SDB_SET_BINARY(pRaw, dataPos, pFunc->pCode, pFunc->codeSize, FUNC_ENCODE_OVER) + SDB_SET_RESERVE(pRaw, dataPos, SDB_FUNC_RESERVE_SIZE, FUNC_ENCODE_OVER) SDB_SET_DATALEN(pRaw, dataPos, FUNC_ENCODE_OVER); terrno = 0; @@ -131,6 +133,7 @@ static SSdbRow *mndFuncActionDecode(SSdbRaw *pRaw) { SDB_GET_BINARY(pRaw, dataPos, pFunc->pComment, pFunc->commentSize, FUNC_DECODE_OVER) SDB_GET_BINARY(pRaw, dataPos, pFunc->pCode, pFunc->codeSize, FUNC_DECODE_OVER) + SDB_GET_RESERVE(pRaw, dataPos, SDB_FUNC_RESERVE_SIZE, FUNC_DECODE_OVER) terrno = 0; diff --git a/source/dnode/mnode/impl/src/mndSubscribe.c b/source/dnode/mnode/impl/src/mndSubscribe.c index df6a4a82f3..87f66297a4 100644 --- a/source/dnode/mnode/impl/src/mndSubscribe.c +++ b/source/dnode/mnode/impl/src/mndSubscribe.c @@ -66,13 +66,13 @@ int32_t mndInitSubscribe(SMnode *pMnode) { static int32_t mndProcessGetSubEpReq(SMnodeMsg *pMsg) { SMnode *pMnode = pMsg->pMnode; - SMqCMGetSubEpReq *pReq = (SMqCMGetSubEpReq *)pMsg->pCont; + SMqCMGetSubEpReq *pReq = (SMqCMGetSubEpReq *)pMsg->rpcMsg.pCont; SMqCMGetSubEpRsp rsp; int64_t consumerId = be64toh(pReq->consumerId); SMqConsumerObj *pConsumer = mndAcquireConsumer(pMsg->pMnode, consumerId); if (pConsumer == NULL) { - /*terrno = */ + terrno = TSDB_CODE_MND_CONSUMER_NOT_EXIST; return -1; } ASSERT(strcmp(pReq->cgroup, pConsumer->cgroup) == 0); @@ -91,9 +91,13 @@ static int32_t mndProcessGetSubEpReq(SMnodeMsg *pMsg) { int32_t assignedSz = taosArrayGetSize(pSub->assigned); topicEp.vgs = taosArrayInit(assignedSz, sizeof(SMqSubVgEp)); for (int32_t j = 0; j < assignedSz; j++) { - SMqConsumerEp *pCEp = taosArrayGet(pSub->assigned, i); + SMqConsumerEp *pCEp = taosArrayGet(pSub->assigned, j); if (pCEp->consumerId == consumerId) { - taosArrayPush(pSub->assigned, pCEp); + SMqSubVgEp vgEp = { + .epSet = pCEp->epSet, + .vgId = pCEp->vgId + }; + taosArrayPush(topicEp.vgs, &vgEp); } } if (taosArrayGetSize(topicEp.vgs) != 0) { @@ -101,7 +105,7 @@ static int32_t mndProcessGetSubEpReq(SMnodeMsg *pMsg) { } } int32_t tlen = tEncodeSMqCMGetSubEpRsp(NULL, &rsp); - void *buf = malloc(tlen); + void *buf = rpcMallocCont(tlen); if (buf == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; return -1; @@ -161,8 +165,6 @@ static int32_t mndProcessMqTimerMsg(SMnodeMsg *pMsg) { pReq->sql = strdup(pTopic->sql); pReq->logicalPlan = strdup(pTopic->logicalPlan); pReq->physicalPlan = strdup(pTopic->physicalPlan); - pReq->qmsgLen = pCEp->qmsgLen; - /*memcpy(pReq->qmsg, pCEp->qmsg, pCEp->qmsgLen);*/ pReq->qmsg = strdup(pCEp->qmsg); int32_t tlen = tEncodeSMqSetCVgReq(NULL, pReq); void *reqStr = malloc(tlen); @@ -192,7 +194,7 @@ static int32_t mndProcessMqTimerMsg(SMnodeMsg *pMsg) { if (mndTransPrepare(pMnode, pTrans) != 0) { mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr()); } - mndReleaseTopic(pMnode, pTopic); + /*mndReleaseTopic(pMnode, pTopic);*/ mndTransDrop(pTrans); } pIter = sdbFetch(pSdb, SDB_SUBSCRIBE, NULL, (void **)&pSub); @@ -220,6 +222,7 @@ static int mndInitUnassignedVg(SMnode *pMnode, SMqTopicObj *pTopic, SArray *unas for (int32_t i = 0; i < sz; i++) { SMqConsumerEp CEp; CEp.status = 0; + CEp.consumerId = -1; CEp.lastConsumerHbTs = CEp.lastVgHbTs = -1; STaskInfo *pTaskInfo = taosArrayGet(pArray, i); tConvertQueryAddrToEpSet(&CEp.epSet, &pTaskInfo->addr); @@ -227,13 +230,6 @@ static int mndInitUnassignedVg(SMnode *pMnode, SMqTopicObj *pTopic, SArray *unas * CEp.epSet.fqdn[2], CEp.epSet.fqdn[3], CEp.epSet.fqdn[4]);*/ CEp.vgId = pTaskInfo->addr.nodeId; CEp.qmsg = strdup(pTaskInfo->msg->msg); - CEp.qmsgLen = strlen(CEp.qmsg) + 1; - printf("abc:\n%s\n", CEp.qmsg); - /*CEp.qmsg = malloc(CEp.qmsgLen);*/ - /*if (CEp.qmsg == NULL) {*/ - /*return -1;*/ - /*}*/ - /*memcpy(CEp.qmsg, pTaskInfo->msg->msg, pTaskInfo->msg->contentLen);*/ taosArrayPush(unassignedVg, &CEp); } @@ -257,8 +253,7 @@ static int mndBuildMqSetConsumerVgReq(SMnode *pMnode, STrans *pTrans, SMqConsume req.sql = pTopic->sql; req.logicalPlan = pTopic->logicalPlan; req.physicalPlan = pTopic->physicalPlan; - req.qmsg = strdup(pCEp->qmsg); - req.qmsgLen = strlen(req.qmsg); + req.qmsg = pCEp->qmsg; int32_t tlen = tEncodeSMqSetCVgReq(NULL, &req); void *buf = malloc(sizeof(SMsgHead) + tlen); if (buf == NULL) { @@ -631,14 +626,14 @@ static int32_t mndProcessSubscribeReq(SMnodeMsg *pMsg) { mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr()); if (newSub) taosArrayDestroy(newSub); mndTransDrop(pTrans); - mndReleaseConsumer(pMnode, pConsumer); + /*mndReleaseConsumer(pMnode, pConsumer);*/ return -1; } if (newSub) taosArrayDestroy(newSub); mndTransDrop(pTrans); - mndReleaseConsumer(pMnode, pConsumer); - return 0; + /*mndReleaseConsumer(pMnode, pConsumer);*/ + return TSDB_CODE_MND_ACTION_IN_PROGRESS; } static int32_t mndProcessSubscribeInternalRsp(SMnodeMsg *pRsp) { diff --git a/source/dnode/mnode/impl/src/mndTopic.c b/source/dnode/mnode/impl/src/mndTopic.c index fa043cf7a0..6b4cb4ba59 100644 --- a/source/dnode/mnode/impl/src/mndTopic.c +++ b/source/dnode/mnode/impl/src/mndTopic.c @@ -237,7 +237,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SMnodeMsg *pMsg, SCMCreateTopicReq tstrncpy(topicObj.db, pDb->name, TSDB_DB_FNAME_LEN); topicObj.createTime = taosGetTimestampMs(); topicObj.updateTime = topicObj.createTime; - topicObj.uid = mndGenerateUid(pCreate->name, TSDB_TABLE_FNAME_LEN); + topicObj.uid = mndGenerateUid(pCreate->name, strlen(pCreate->name)); topicObj.dbUid = pDb->uid; topicObj.version = 1; topicObj.sql = strdup(pCreate->sql); diff --git a/source/dnode/mnode/impl/src/mndTrans.c b/source/dnode/mnode/impl/src/mndTrans.c index 03226c7400..808e6dcbe5 100644 --- a/source/dnode/mnode/impl/src/mndTrans.c +++ b/source/dnode/mnode/impl/src/mndTrans.c @@ -921,7 +921,7 @@ static int32_t mndProcessTransMsg(SMnodeMsg *pMsg) { void mndTransPullup(SMnode *pMnode) { STrans *pTrans = NULL; - void *pIter = NULL; + void * pIter = NULL; while (1) { pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans); @@ -930,4 +930,6 @@ void mndTransPullup(SMnode *pMnode) { mndTransExecute(pMnode, pTrans); sdbRelease(pMnode->pSdb, pTrans); } + + sdbWriteFile(pMnode->pSdb); } diff --git a/source/dnode/mnode/impl/src/mndVgroup.c b/source/dnode/mnode/impl/src/mndVgroup.c index 04fedbb3ce..73c37560fc 100644 --- a/source/dnode/mnode/impl/src/mndVgroup.c +++ b/source/dnode/mnode/impl/src/mndVgroup.c @@ -178,7 +178,7 @@ static int32_t mndVgroupActionUpdate(SSdb *pSdb, SVgObj *pOld, SVgObj *pNew) { SVgObj *mndAcquireVgroup(SMnode *pMnode, int32_t vgId) { SSdb *pSdb = pMnode->pSdb; SVgObj *pVgroup = sdbAcquire(pSdb, SDB_VGROUP, &vgId); - if (pVgroup == NULL) { + if (pVgroup == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) { terrno = TSDB_CODE_MND_VGROUP_NOT_EXIST; } return pVgroup; diff --git a/source/dnode/mnode/sdb/src/sdb.c b/source/dnode/mnode/sdb/src/sdb.c index 39b5bb4d5b..26809ea059 100644 --- a/source/dnode/mnode/sdb/src/sdb.c +++ b/source/dnode/mnode/sdb/src/sdb.c @@ -64,11 +64,7 @@ SSdb *sdbInit(SSdbOpt *pOption) { void sdbCleanup(SSdb *pSdb) { mDebug("start to cleanup sdb"); - if (pSdb->curVer > pSdb->lastCommitVer) { - mDebug("write sdb file for current ver:%" PRId64 " larger than last commit ver:%" PRId64, pSdb->curVer, - pSdb->lastCommitVer); - sdbWriteFile(pSdb); - } + sdbWriteFile(pSdb); if (pSdb->currDir != NULL) { tfree(pSdb->currDir); diff --git a/source/dnode/mnode/sdb/src/sdbFile.c b/source/dnode/mnode/sdb/src/sdbFile.c index 1f6d6cbda8..6d17423324 100644 --- a/source/dnode/mnode/sdb/src/sdbFile.c +++ b/source/dnode/mnode/sdb/src/sdbFile.c @@ -221,7 +221,7 @@ PARSE_SDB_DATA_ERROR: return code; } -int32_t sdbWriteFile(SSdb *pSdb) { +static int32_t sdbWriteFileImp(SSdb *pSdb) { int32_t code = 0; char tmpfile[PATH_MAX] = {0}; @@ -229,7 +229,8 @@ int32_t sdbWriteFile(SSdb *pSdb) { char curfile[PATH_MAX] = {0}; snprintf(curfile, sizeof(curfile), "%s%ssdb.data", pSdb->currDir, TD_DIRSEP); - mDebug("start to write file:%s", curfile); + mDebug("start to write file:%s, current ver:%" PRId64 ", commit ver:%" PRId64, curfile, pSdb->curVer, + pSdb->lastCommitVer); FileFd fd = taosOpenFileCreateWriteTrunc(tmpfile); if (fd <= 0) { @@ -323,12 +324,20 @@ int32_t sdbWriteFile(SSdb *pSdb) { return code; } +int32_t sdbWriteFile(SSdb *pSdb) { + if (pSdb->curVer == pSdb->lastCommitVer) { + return 0; + } + + return sdbWriteFileImp(pSdb); +} + int32_t sdbDeploy(SSdb *pSdb) { if (sdbRunDeployFp(pSdb) != 0) { return -1; } - if (sdbWriteFile(pSdb) != 0) { + if (sdbWriteFileImp(pSdb) != 0) { return -1; } diff --git a/source/dnode/vnode/inc/tsdb.h b/source/dnode/vnode/inc/tsdb.h index 6a9eb43ca8..5f5acc1b05 100644 --- a/source/dnode/vnode/inc/tsdb.h +++ b/source/dnode/vnode/inc/tsdb.h @@ -75,7 +75,6 @@ typedef struct STsdbQueryCond { } STsdbQueryCond; typedef struct { - void *pTable; TSKEY lastKey; uint64_t uid; } STableKeyInfo; @@ -141,7 +140,7 @@ bool isTsdbCacheLastRow(tsdbReadHandleT* pTsdbReadHandle); */ int32_t tsdbQuerySTableByTagCond(STsdb* tsdb, uint64_t uid, TSKEY skey, const char* pTagCond, size_t len, int16_t tagNameRelType, const char* tbnameCond, STableGroupInfo* pGroupInfo, - SColIndex* pColIndex, int32_t numOfCols, uint64_t reqId); + SColIndex* pColIndex, int32_t numOfCols, uint64_t reqId, uint64_t taskId); /** * get num of rows in mem table * diff --git a/source/dnode/vnode/inc/vnode.h b/source/dnode/vnode/inc/vnode.h index 2accfd6279..b56c5b30fa 100644 --- a/source/dnode/vnode/inc/vnode.h +++ b/source/dnode/vnode/inc/vnode.h @@ -33,6 +33,7 @@ extern "C" { typedef struct SVnode SVnode; typedef struct SDnode SDnode; typedef int32_t (*PutReqToVQueryQFp)(SDnode *pDnode, struct SRpcMsg *pReq); +typedef int32_t (*SendReqToDnodeFp)(SDnode *pDnode, struct SEpSet *epSet, struct SRpcMsg *rpcMsg); typedef struct STqCfg { // TODO @@ -64,6 +65,7 @@ typedef struct { const char *charset; uint16_t nthreads; // number of commit threads. 0 for no threads and a schedule queue should be given (TODO) PutReqToVQueryQFp putReqToVQueryQFp; + SendReqToDnodeFp sendReqToDnodeFp; } SVnodeOpt; typedef struct STqReadHandle { diff --git a/source/dnode/vnode/src/inc/vnd.h b/source/dnode/vnode/src/inc/vnd.h index 1fa65b2a73..f442697fb0 100644 --- a/source/dnode/vnode/src/inc/vnd.h +++ b/source/dnode/vnode/src/inc/vnd.h @@ -55,6 +55,7 @@ typedef struct SVnodeMgr { // For vnode Mgmt SDnode* pDnode; PutReqToVQueryQFp putReqToVQueryQFp; + SendReqToDnodeFp sendReqToDnodeFp; } SVnodeMgr; extern SVnodeMgr vnodeMgr; @@ -85,6 +86,7 @@ struct SVnode { int vnodeScheduleTask(SVnodeTask* task); int32_t vnodePutReqToVQueryQ(SVnode* pVnode, struct SRpcMsg* pReq); +void vnodeSendReqToDnode(SVnode* pVnode, struct SEpSet* epSet, struct SRpcMsg* pReq); // For Log extern int32_t vDebugFlag; diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index c164487aa2..e953d59527 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -679,6 +679,7 @@ int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) { int rspLen = 0; STqConsumerHandle* pConsumer = tqHandleGet(pTq->tqMeta, consumerId); + ASSERT(pConsumer); int sz = taosArrayGetSize(pConsumer->topics); for (int i = 0; i < sz; i++) { @@ -735,23 +736,6 @@ int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) { break; } if (pDataBlock != NULL) { - SMqTbData tbData = { - .uid = pDataBlock->info.uid, - .numOfCols = pDataBlock->info.numOfCols, - .numOfRows = pDataBlock->info.rows, - }; - for (int i = 0; i < pDataBlock->info.numOfCols; i++) { - SColumnInfoData* pColData = taosArrayGet(pDataBlock->pDataBlock, i); - int32_t sz = pColData->info.bytes * pDataBlock->info.rows; - SMqColData colData = { - .bytes = pColData->info.bytes, - .colId = pColData->info.colId, - .type = pColData->info.type, - }; - memcpy(colData.data, pColData->pData, colData.bytes * pDataBlock->info.rows); - memcpy(&tbData.colData[i], &colData, sz); - } - /*pDataBlock->info.*/ taosArrayPush(pRes, pDataBlock); } else { break; @@ -773,12 +757,12 @@ int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) { if (pTopic->buffer.lastOffset == -1 || pReq->offset > pTopic->buffer.lastOffset) { pTopic->buffer.lastOffset = pReq->offset; } - // put output into rsp - SMqConsumeRsp rsp = { - .consumerId = consumerId, - .numOfTopics = 1 - }; } + // put output into rsp + SMqConsumeRsp rsp = { + .consumerId = consumerId, + .numOfTopics = 1 + }; return 0; } @@ -792,6 +776,8 @@ int32_t tqProcessSetConnReq(STQ* pTq, char* msg) { } strcpy(pConsumer->cgroup, req.cgroup); pConsumer->topics = taosArrayInit(0, sizeof(STqTopicHandle)); + pConsumer->consumerId = req.newConsumerId; + pConsumer->epoch = 0; STqTopicHandle* pTopic = calloc(sizeof(STqTopicHandle), 1); if (pTopic == NULL) { @@ -802,6 +788,8 @@ int32_t tqProcessSetConnReq(STQ* pTq, char* msg) { pTopic->sql = strdup(req.sql); pTopic->logicalPlan = strdup(req.logicalPlan); pTopic->physicalPlan = strdup(req.physicalPlan); + pTopic->committedOffset = -1; + pTopic->currentOffset = -1; pTopic->buffer.firstOffset = -1; pTopic->buffer.lastOffset = -1; @@ -814,6 +802,8 @@ int32_t tqProcessSetConnReq(STQ* pTq, char* msg) { pTopic->buffer.output[i].task = qCreateStreamExecTaskInfo(req.qmsg, pReadHandle); } taosArrayPush(pConsumer->topics, pTopic); + tqHandleMovePut(pTq->tqMeta, req.newConsumerId, pConsumer); + tqHandleCommit(pTq->tqMeta, req.newConsumerId); terrno = TSDB_CODE_SUCCESS; return 0; } @@ -832,14 +822,29 @@ STqReadHandle* tqInitSubmitMsgScanner(SMeta* pMeta) { void tqReadHandleSetMsg(STqReadHandle* pReadHandle, SSubmitMsg* pMsg, int64_t ver) { pReadHandle->pMsg = pMsg; + pMsg->length = htonl(pMsg->length); + pMsg->numOfBlocks = htonl(pMsg->numOfBlocks); tInitSubmitMsgIter(pMsg, &pReadHandle->msgIter); pReadHandle->ver = ver; memset(&pReadHandle->blkIter, 0, sizeof(SSubmitBlkIter)); } bool tqNextDataBlock(STqReadHandle* pHandle) { - while (tGetSubmitMsgNext(&pHandle->msgIter, &pHandle->pBlock) >= 0) { - if (pHandle->tbUid == pHandle->pBlock->uid) return true; + while (1) { + if (tGetSubmitMsgNext(&pHandle->msgIter, &pHandle->pBlock) < 0) { + return false; + } + if (pHandle->pBlock == NULL) return false; + + pHandle->pBlock->uid = htobe64(pHandle->pBlock->uid); + if (pHandle->tbUid == pHandle->pBlock->uid){ + pHandle->pBlock->tid = htonl(pHandle->pBlock->tid); + pHandle->pBlock->sversion = htonl(pHandle->pBlock->sversion); + pHandle->pBlock->dataLen = htonl(pHandle->pBlock->dataLen); + pHandle->pBlock->schemaLen = htonl(pHandle->pBlock->schemaLen); + pHandle->pBlock->numOfRows = htons(pHandle->pBlock->numOfRows); + return true; + } } return false; } @@ -855,8 +860,18 @@ int tqRetrieveDataBlockInfo(STqReadHandle* pHandle, SDataBlockInfo* pBlockInfo) SArray* tqRetrieveDataBlock(STqReadHandle* pHandle) { int32_t sversion = pHandle->pBlock->sversion; - SSchemaWrapper* pSchemaWrapper = metaGetTableSchema(pHandle->pMeta, pHandle->pBlock->uid, sversion, true); - STSchema* pTschema = metaGetTbTSchema(pHandle->pMeta, pHandle->pBlock->uid, sversion); + //TODO : change sversion + STSchema* pTschema = metaGetTbTSchema(pHandle->pMeta, pHandle->pBlock->uid, 0); + + tb_uid_t quid; + STbCfg* pTbCfg = metaGetTbInfoByUid(pHandle->pMeta, pHandle->pBlock->uid); + if (pTbCfg->type == META_CHILD_TABLE) { + quid = pTbCfg->ctbCfg.suid; + } else { + quid = pHandle->pBlock->uid; + } + + SSchemaWrapper* pSchemaWrapper = metaGetTableSchema(pHandle->pMeta, quid, 0, true); SArray* pArray = taosArrayInit(pSchemaWrapper->nCols, sizeof(SColumnInfoData)); if (pArray == NULL) { return NULL; diff --git a/source/dnode/vnode/src/tsdb/tsdbRead.c b/source/dnode/vnode/src/tsdb/tsdbRead.c index 372a5820b7..1bca7aea10 100644 --- a/source/dnode/vnode/src/tsdb/tsdbRead.c +++ b/source/dnode/vnode/src/tsdb/tsdbRead.c @@ -81,7 +81,6 @@ enum { CHECKINFO_CHOSEN_BOTH = 2 //for update=2(merge case) }; - typedef struct STableCheckInfo { uint64_t tableId; TSKEY lastKey; @@ -209,31 +208,6 @@ static SArray* getDefaultLoadColumns(STsdbReadHandle* pTsdbReadHandle, bool load return pLocalIdList; } -static void tsdbMayTakeMemSnapshot(STsdbReadHandle* pTsdbReadHandle, SArray* psTable) { -// assert(pTsdbReadHandle != NULL && pTsdbReadHandle->pMemRef != NULL); -// -// STsdbMemTable* pMemRef = pTsdbReadHandle->pMemRef; -// if (pTsdbReadHandle->pMemRef->ref++ == 0) { -// tsdbTakeMemSnapshot(pTsdbReadHandle->pTsdb, &(pMemRef->snapshot), psTable); -// } -// -// taosArrayDestroy(psTable); -} - -static void tsdbMayUnTakeMemSnapshot(STsdbReadHandle* pTsdbReadHandle) { -// assert(pTsdbReadHandle != NULL); -// STsdbMemTable* pMemRef = pTsdbReadHandle->pMemRef; -// if (pMemRef == NULL) { // it has been freed -// return; -// } -// -// if (--pMemRef->ref == 0) { -// tsdbUnTakeMemSnapShot(pTsdbReadHandle->pTsdb, &(pMemRef->snapshot)); -// } -// -// pTsdbReadHandle->pMemRef = NULL; -} - //int64_t tsdbGetNumOfRowsInMemTable(tsdbReadHandleT* pHandle) { // STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*) pHandle; // @@ -263,9 +237,9 @@ static void tsdbMayUnTakeMemSnapshot(STsdbReadHandle* pTsdbReadHandle) { // return rows; //} -static SArray* createCheckInfoFromTableGroup(STsdbReadHandle* pTsdbReadHandle, STableGroupInfo* pGroupList, SArray** psTable) { - size_t sizeOfGroup = taosArrayGetSize(pGroupList->pGroupList); - assert(sizeOfGroup >= 1); +static SArray* createCheckInfoFromTableGroup(STsdbReadHandle* pTsdbReadHandle, STableGroupInfo* pGroupList) { + size_t numOfGroup = taosArrayGetSize(pGroupList->pGroupList); + assert(numOfGroup >= 1); // allocate buffer in order to load data blocks from file SArray* pTableCheckInfo = taosArrayInit(pGroupList->numOfTables, sizeof(STableCheckInfo)); @@ -273,14 +247,8 @@ static SArray* createCheckInfoFromTableGroup(STsdbReadHandle* pTsdbReadHandle, S return NULL; } - SArray* pTable = taosArrayInit(4, sizeof(STable*)); - if (pTable == NULL) { - taosArrayDestroy(pTableCheckInfo); - return NULL; - } - // todo apply the lastkey of table check to avoid to load header file - for (int32_t i = 0; i < sizeOfGroup; ++i) { + for (int32_t i = 0; i < numOfGroup; ++i) { SArray* group = *(SArray**) taosArrayGet(pGroupList->pGroupList, i); size_t gsize = taosArrayGetSize(group); @@ -289,12 +257,7 @@ static SArray* createCheckInfoFromTableGroup(STsdbReadHandle* pTsdbReadHandle, S for (int32_t j = 0; j < gsize; ++j) { STableKeyInfo* pKeyInfo = (STableKeyInfo*) taosArrayGet(group, j); - STableCheckInfo info = { .lastKey = pKeyInfo->lastKey}; -// assert(info.pTableObj != NULL && (info.pTableObj->type == TSDB_NORMAL_TABLE || -// info.pTableObj->type == TSDB_CHILD_TABLE || info.pTableObj->type == TSDB_STREAM_TABLE)); - - info.tableId = pKeyInfo->uid; - + STableCheckInfo info = { .lastKey = pKeyInfo->lastKey, .tableId = pKeyInfo->uid}; if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) { if (info.lastKey == INT64_MIN || info.lastKey < pTsdbReadHandle->window.skey) { info.lastKey = pTsdbReadHandle->window.skey; @@ -310,13 +273,8 @@ static SArray* createCheckInfoFromTableGroup(STsdbReadHandle* pTsdbReadHandle, S } } -// taosArraySort(pTableCheckInfo, tsdbCheckInfoCompar); - size_t gsize = taosArrayGetSize(pTableCheckInfo); -// for (int32_t i = 0; i < gsize; ++i) { -// STableCheckInfo* pInfo = (STableCheckInfo*) taosArrayGet(pTableCheckInfo, i); -// } - - *psTable = pTable; + // TODO group table according to the tag value. + taosArraySort(pTableCheckInfo, tsdbCheckInfoCompar); return pTableCheckInfo; } @@ -484,18 +442,17 @@ tsdbReadHandleT* tsdbQueryTables(STsdb* tsdb, STsdbQueryCond* pCond, STableGroup return (tsdbReadHandleT*) pTsdbReadHandle; } - SArray* psTable = NULL; - // todo apply the lastkey of table check to avoid to load header file - pTsdbReadHandle->pTableCheckInfo = createCheckInfoFromTableGroup(pTsdbReadHandle, groupList, &psTable); + pTsdbReadHandle->pTableCheckInfo = createCheckInfoFromTableGroup(pTsdbReadHandle, groupList); if (pTsdbReadHandle->pTableCheckInfo == NULL) { // tsdbCleanupQueryHandle(pTsdbReadHandle); - taosArrayDestroy(psTable); terrno = TSDB_CODE_TDB_OUT_OF_MEMORY; return NULL; } - tsdbDebug("%p total numOfTable:%" PRIzu " in query, %s", pTsdbReadHandle, taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo), pTsdbReadHandle->idStr); + tsdbDebug("%p total numOfTable:%" PRIzu " in this query, group %"PRIzu" %s", pTsdbReadHandle, taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo), + taosArrayGetSize(groupList->pGroupList), pTsdbReadHandle->idStr); + return (tsdbReadHandleT) pTsdbReadHandle; } @@ -2640,7 +2597,7 @@ static int32_t getAllTableList(SMeta* pMeta, uint64_t uid, SArray* list) { break; } - STableKeyInfo info = {.pTable = NULL, .lastKey = TSKEY_INITIAL_VAL, uid = id}; + STableKeyInfo info = {.lastKey = TSKEY_INITIAL_VAL, uid = id}; taosArrayPush(list, &info); } @@ -3231,7 +3188,7 @@ STimeWindow updateLastrowForEachGroup(STableGroupInfo *groupList) { if (key < lastKey) { key = lastKey; - keyInfo.pTable = pInfo->pTable; +// keyInfo.pTable = pInfo->pTable; keyInfo.lastKey = key; pInfo->lastKey = key; @@ -3245,29 +3202,19 @@ STimeWindow updateLastrowForEachGroup(STableGroupInfo *groupList) { } } - // clear current group, unref unused table - for (int32_t i = 0; i < numOfTables; ++i) { - STableKeyInfo* pInfo = (STableKeyInfo*)taosArrayGet(pGroup, i); - - // keyInfo.pTable may be NULL here. - if (pInfo->pTable != keyInfo.pTable) { -// tsdbUnRefTable(pInfo->pTable); - } - } - // more than one table in each group, only one table left for each group - if (keyInfo.pTable != NULL) { - totalNumOfTable++; - if (taosArrayGetSize(pGroup) == 1) { - // do nothing - } else { - taosArrayClear(pGroup); - taosArrayPush(pGroup, &keyInfo); - } - } else { // mark all the empty groups, and remove it later - taosArrayDestroy(pGroup); - taosArrayPush(emptyGroup, &j); - } +// if (keyInfo.pTable != NULL) { +// totalNumOfTable++; +// if (taosArrayGetSize(pGroup) == 1) { +// // do nothing +// } else { +// taosArrayClear(pGroup); +// taosArrayPush(pGroup, &keyInfo); +// } +// } else { // mark all the empty groups, and remove it later +// taosArrayDestroy(pGroup); +// taosArrayPush(emptyGroup, &j); +// } } // window does not being updated, so set the original @@ -3457,11 +3404,13 @@ void filterPrepare(void* expr, void* param) { } } +#endif static int32_t tableGroupComparFn(const void *p1, const void *p2, const void *param) { +#if 0 STableGroupSupporter* pTableGroupSupp = (STableGroupSupporter*) param; - STable* pTable1 = ((STableKeyInfo*) p1)->pTable; - STable* pTable2 = ((STableKeyInfo*) p2)->pTable; + STable* pTable1 = ((STableKeyInfo*) p1)->uid; + STable* pTable2 = ((STableKeyInfo*) p2)->uid; for (int32_t i = 0; i < pTableGroupSupp->numOfCols; ++i) { SColIndex* pColIndex = &pTableGroupSupp->pCols[i]; @@ -3509,10 +3458,9 @@ static int32_t tableGroupComparFn(const void *p1, const void *p2, const void *pa return ret; } } - +#endif return 0; } -#endif static int tsdbCheckInfoCompar(const void* key1, const void* key2) { if (((STableCheckInfo*)key1)->tableId < ((STableCheckInfo*)key2)->tableId) { @@ -3528,10 +3476,9 @@ static int tsdbCheckInfoCompar(const void* key1, const void* key2) { void createTableGroupImpl(SArray* pGroups, SArray* pTableList, size_t numOfTables, TSKEY skey, STableGroupSupporter* pSupp, __ext_compar_fn_t compareFn) { STable* pTable = taosArrayGetP(pTableList, 0); - SArray* g = taosArrayInit(16, sizeof(STableKeyInfo)); - STableKeyInfo info = {.pTable = pTable, .lastKey = skey}; + STableKeyInfo info = {.lastKey = skey}; taosArrayPush(g, &info); for (int32_t i = 1; i < numOfTables; ++i) { @@ -3542,13 +3489,13 @@ void createTableGroupImpl(SArray* pGroups, SArray* pTableList, size_t numOfTable assert(ret == 0 || ret == -1); if (ret == 0) { - STableKeyInfo info1 = {.pTable = *p, .lastKey = skey}; + STableKeyInfo info1 = {.lastKey = skey}; taosArrayPush(g, &info1); } else { taosArrayPush(pGroups, &g); // current group is ended, start a new group g = taosArrayInit(16, sizeof(STableKeyInfo)); - STableKeyInfo info1 = {.pTable = *p, .lastKey = skey}; + STableKeyInfo info1 = {.lastKey = skey}; taosArrayPush(g, &info1); } } @@ -3581,8 +3528,8 @@ SArray* createTableGroup(SArray* pTableList, SSchemaWrapper* pTagSchema, SColInd sup.pTagSchema = pTagSchema->pSchema; sup.pCols = pCols; -// taosqsort(pTableList->pData, size, sizeof(STableKeyInfo), &sup, tableGroupComparFn); -// createTableGroupImpl(pTableGroup, pTableList, size, skey, &sup, tableGroupComparFn); + taosqsort(pTableList->pData, size, sizeof(STableKeyInfo), &sup, tableGroupComparFn); + createTableGroupImpl(pTableGroup, pTableList, size, skey, &sup, tableGroupComparFn); } return pTableGroup; @@ -3690,16 +3637,16 @@ SArray* createTableGroup(SArray* pTableList, SSchemaWrapper* pTagSchema, SColInd int32_t tsdbQuerySTableByTagCond(STsdb* tsdb, uint64_t uid, TSKEY skey, const char* pTagCond, size_t len, int16_t tagNameRelType, const char* tbnameCond, STableGroupInfo* pGroupInfo, - SColIndex* pColIndex, int32_t numOfCols, uint64_t reqId) { + SColIndex* pColIndex, int32_t numOfCols, uint64_t reqId, uint64_t taskId) { STbCfg* pTbCfg = metaGetTbInfoByUid(tsdb->pMeta, uid); if (pTbCfg == NULL) { - tsdbError("%p failed to get stable, uid:%"PRIu64", reqId:0x%"PRIx64, tsdb, uid, reqId); + tsdbError("%p failed to get stable, uid:%"PRIu64", TID:0x%"PRIx64" QID:0x%"PRIx64, tsdb, uid, taskId, reqId); terrno = TSDB_CODE_TDB_INVALID_TABLE_ID; goto _error; } if (pTbCfg->type != META_SUPER_TABLE) { - tsdbError("%p query normal tag not allowed, uid:%" PRIu64 ", reId:0x%"PRIx64, tsdb, uid, reqId); + tsdbError("%p query normal tag not allowed, uid:%" PRIu64 ", TID:0x%"PRIx64" QID:0x%"PRIx64, tsdb, uid, taskId, reqId); terrno = TSDB_CODE_OPS_NOT_SUPPORT; //basically, this error is caused by invalid sql issued by client goto _error; } @@ -3718,8 +3665,8 @@ int32_t tsdbQuerySTableByTagCond(STsdb* tsdb, uint64_t uid, TSKEY skey, const ch pGroupInfo->numOfTables = (uint32_t) taosArrayGetSize(res); pGroupInfo->pGroupList = createTableGroup(res, pTagSchema, pColIndex, numOfCols, skey); - tsdbDebug("%p no table name/tag condition, all tables qualified, numOfTables:%u, group:%zu", tsdb, - pGroupInfo->numOfTables, taosArrayGetSize(pGroupInfo->pGroupList)); + tsdbDebug("%p no table name/tag condition, all tables qualified, numOfTables:%u, group:%zu, TID:0x%"PRIx64" QID:0x%"PRIx64, tsdb, + pGroupInfo->numOfTables, taosArrayGetSize(pGroupInfo->pGroupList), taskId, reqId); taosArrayDestroy(res); return ret; @@ -3892,7 +3839,7 @@ void tsdbCleanupQueryHandle(tsdbReadHandleT queryHandle) { tfree(pTsdbReadHandle->statis); if (!emptyQueryTimewindow(pTsdbReadHandle)) { - tsdbMayUnTakeMemSnapshot(pTsdbReadHandle); +// tsdbMayUnTakeMemSnapshot(pTsdbReadHandle); } else { assert(pTsdbReadHandle->pTableCheckInfo == NULL); } diff --git a/source/dnode/vnode/src/vnd/vnodeMgr.c b/source/dnode/vnode/src/vnd/vnodeMgr.c index d762844120..477deed8c8 100644 --- a/source/dnode/vnode/src/vnd/vnodeMgr.c +++ b/source/dnode/vnode/src/vnd/vnodeMgr.c @@ -26,6 +26,7 @@ int vnodeInit(const SVnodeOpt *pOption) { vnodeMgr.stop = false; vnodeMgr.putReqToVQueryQFp = pOption->putReqToVQueryQFp; + vnodeMgr.sendReqToDnodeFp = pOption->sendReqToDnodeFp; // Start commit handers if (pOption->nthreads > 0) { @@ -96,6 +97,10 @@ int32_t vnodePutReqToVQueryQ(SVnode* pVnode, struct SRpcMsg* pReq) { return (*vnodeMgr.putReqToVQueryQFp)(pVnode->pDnode, pReq); } +void vnodeSendReqToDnode(SVnode* pVnode, struct SEpSet* epSet, struct SRpcMsg* pReq) { + (*vnodeMgr.sendReqToDnodeFp)(pVnode->pDnode, epSet, pReq); +} + /* ------------------------ STATIC METHODS ------------------------ */ static void* loop(void* arg) { setThreadName("vnode-commit"); diff --git a/source/dnode/vnode/src/vnd/vnodeQuery.c b/source/dnode/vnode/src/vnd/vnodeQuery.c index b211894281..6692264c54 100644 --- a/source/dnode/vnode/src/vnd/vnodeQuery.c +++ b/source/dnode/vnode/src/vnd/vnodeQuery.c @@ -21,7 +21,7 @@ static int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg); int vnodeQueryOpen(SVnode *pVnode) { return qWorkerInit(NODE_TYPE_VNODE, pVnode->vgId, NULL, (void **)&pVnode->pQuery, pVnode, - (putReqToQueryQFp)vnodePutReqToVQueryQ); + (putReqToQueryQFp)vnodePutReqToVQueryQ, (sendReqToDnodeFp)vnodeSendReqToDnode); } int vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) { @@ -43,6 +43,8 @@ int vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg) { switch (pMsg->msgType) { case TDMT_VND_FETCH: return qWorkerProcessFetchMsg(pVnode, pVnode->pQuery, pMsg); + case TDMT_VND_FETCH_RSP: + return qWorkerProcessFetchRsp(pVnode, pVnode->pQuery, pMsg); case TDMT_VND_RES_READY: return qWorkerProcessReadyMsg(pVnode, pVnode->pQuery, pMsg); case TDMT_VND_TASKS_STATUS: diff --git a/source/libs/executor/inc/executil.h b/source/libs/executor/inc/executil.h index 702581edb9..130e46fc4c 100644 --- a/source/libs/executor/inc/executil.h +++ b/source/libs/executor/inc/executil.h @@ -38,7 +38,7 @@ #define GET_RES_WINDOW_KEY_LEN(_l) ((_l) + sizeof(uint64_t)) #define GET_RES_EXT_WINDOW_KEY_LEN(_l) ((_l) + sizeof(uint64_t) + POINTER_BYTES) -#define GET_TASKID(_t) (((SExecTaskInfo*)(_t))->id.idstr) +#define GET_TASKID(_t) (((SExecTaskInfo*)(_t))->id.str) #define curTimeWindowIndex(_winres) ((_winres)->curIndex) diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index f36071f46c..7af2e9f14f 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -239,8 +239,7 @@ typedef struct STaskIdInfo { uint64_t queryId; // this is also a request id uint64_t subplanId; uint64_t templateId; - uint64_t taskId; // this is a subplan id - char *idstr; + char *str; } STaskIdInfo; typedef struct SExecTaskInfo { @@ -639,7 +638,6 @@ int32_t buildArithmeticExprFromMsg(SExprInfo *pArithExprInfo, void *pQueryMsg); bool isTaskKilled(SExecTaskInfo *pTaskInfo); int32_t checkForQueryBuf(size_t numOfTables); bool checkNeedToCompressQueryCol(SQInfo *pQInfo); -bool doBuildResCheck(SQInfo* pQInfo); void setQueryStatus(STaskRuntimeEnv *pRuntimeEnv, int8_t status); bool onlyQueryTags(STaskAttr* pQueryAttr); diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c index a2eb5dc339..a32268bf47 100644 --- a/source/libs/executor/src/executor.c +++ b/source/libs/executor/src/executor.c @@ -18,20 +18,20 @@ #include "executorimpl.h" #include "planner.h" -static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, uint64_t reqId) { +static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, char* id) { ASSERT(pOperator != NULL); if (pOperator->operatorType != OP_StreamScan) { if (pOperator->numOfDownstream == 0) { - qError("failed to find stream scan operator to set the input data block, reqId:0x%" PRIx64, reqId); + qError("failed to find stream scan operator to set the input data block, %s" PRIx64, id); return TSDB_CODE_QRY_APP_ERROR; } if (pOperator->numOfDownstream > 1) { // not handle this in join query - qError("join not supported for stream block scan, reqId:0x%" PRIx64, reqId); + qError("join not supported for stream block scan, %s" PRIx64, id); return TSDB_CODE_QRY_APP_ERROR; } - return doSetStreamBlock(pOperator->pDownstream[0], input, reqId); + return doSetStreamBlock(pOperator->pDownstream[0], input, id); } else { SStreamBlockScanInfo* pInfo = pOperator->info; tqReadHandleSetMsg(pInfo->readerHandle, input, 0); @@ -50,7 +50,7 @@ int32_t qSetStreamInput(qTaskInfo_t tinfo, void* input) { SExecTaskInfo* pTaskInfo = (SExecTaskInfo*) tinfo; - int32_t code = doSetStreamBlock(pTaskInfo->pRoot, input, pTaskInfo->id.queryId); + int32_t code = doSetStreamBlock(pTaskInfo->pRoot, input, GET_TASKID(pTaskInfo)); if (code != TSDB_CODE_SUCCESS) { qError("%s failed to set the stream block data", GET_TASKID(pTaskInfo)); } else { diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 16487fe747..00ce6f0a74 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -5115,7 +5115,7 @@ static void destroySendMsgInfo(SMsgSendInfo* pMsgBody) { tfree(pMsgBody); } -void processRspMsg(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) { +void qProcessFetchRsp(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) { SMsgSendInfo *pSendInfo = (SMsgSendInfo *) pMsg->ahandle; assert(pMsg->ahandle != NULL); @@ -5146,7 +5146,7 @@ static SSDataBlock* doLoadRemoteData(void* param, bool* newgroup) { size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources); if (pExchangeInfo->current >= totalSources) { - qDebug("%s all %"PRIzu" source(s) are exhausted, total rows:%"PRIu64" bytes:%"PRIu64", elapsed:%.2f ms", pTaskInfo->id.idstr, totalSources, + qDebug("%s all %"PRIzu" source(s) are exhausted, total rows:%"PRIu64" bytes:%"PRIu64", elapsed:%.2f ms", GET_TASKID(pTaskInfo), totalSources, pExchangeInfo->totalRows, pExchangeInfo->totalSize, pExchangeInfo->totalElapsed/1000.0); return NULL; } @@ -5208,7 +5208,7 @@ static SSDataBlock* doLoadRemoteData(void* param, bool* newgroup) { int64_t el = taosGetTimestampUs() - startTs; pExchangeInfo->totalElapsed += el; - qDebug("%s all %"PRIzu" sources are exhausted, total rows: %"PRIu64" bytes:%"PRIu64", elapsed:%.2f ms", pTaskInfo->id.idstr, totalSources, + qDebug("%s all %"PRIzu" sources are exhausted, total rows: %"PRIu64" bytes:%"PRIu64", elapsed:%.2f ms", GET_TASKID(pTaskInfo), totalSources, pExchangeInfo->totalRows, pExchangeInfo->totalSize, pExchangeInfo->totalElapsed/1000.0); return NULL; } else { @@ -5296,13 +5296,14 @@ SOperatorInfo* createExchangeOperatorInfo(const SArray* pSources, const SArray* pOperator->exec = doLoadRemoteData; pOperator->pTaskInfo = pTaskInfo; +#if 1 { // todo refactor SRpcInit rpcInit; memset(&rpcInit, 0, sizeof(rpcInit)); rpcInit.localPort = 0; rpcInit.label = "EX"; rpcInit.numOfThreads = 1; - rpcInit.cfp = processRspMsg; + rpcInit.cfp = qProcessFetchRsp; rpcInit.sessions = tsMaxConnections; rpcInit.connType = TAOS_CONN_CLIENT; rpcInit.user = (char *)"root"; @@ -5316,7 +5317,7 @@ SOperatorInfo* createExchangeOperatorInfo(const SArray* pSources, const SArray* return NULL; // todo } } - +#endif return pOperator; } @@ -5467,6 +5468,9 @@ SOperatorInfo* createStreamScanOperatorInfo(void *streamReadHandle, SArray* pExp return NULL; } + // todo dynamic set the value of 4096 + pInfo->pRes = createOutputBuf_rv(pExprInfo, 4096); + int32_t numOfOutput = (int32_t) taosArrayGetSize(pExprInfo); SArray* pColList = taosArrayInit(numOfOutput, sizeof(int32_t)); for(int32_t i = 0; i < numOfOutput; ++i) { @@ -7431,7 +7435,7 @@ SOperatorInfo* createTagScanOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SExprInfo SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo)); pOperator->name = "SeqTableTagScan"; -// pOperator->operatorType = OP_TagScan; + pOperator->operatorType = OP_TagScan; pOperator->blockingOptr = false; pOperator->status = OP_IN_EXECUTING; pOperator->info = pInfo; @@ -7741,11 +7745,10 @@ static SExecTaskInfo* createExecTaskInfo(uint64_t queryId, uint64_t taskId) { pTaskInfo->cost.created = taosGetTimestampMs(); pTaskInfo->id.queryId = queryId; - pTaskInfo->id.taskId = taskId; char* p = calloc(1, 128); snprintf(p, 128, "TID:0x%"PRIx64" QID:0x%"PRIx64, taskId, queryId); - pTaskInfo->id.idstr = strdup(p); + pTaskInfo->id.str = strdup(p); return pTaskInfo; } @@ -7822,7 +7825,7 @@ static tsdbReadHandleT doCreateDataReadHandle(STableScanPhyNode* pTableScanNode, if (tableType == TSDB_SUPER_TABLE) { code = - tsdbQuerySTableByTagCond(readerHandle, uid, window.skey, NULL, 0, 0, NULL, &groupInfo, NULL, 0, queryId); + tsdbQuerySTableByTagCond(readerHandle, uid, window.skey, NULL, 0, 0, NULL, &groupInfo, NULL, 0, queryId, taskId); if (code != TSDB_CODE_SUCCESS) { goto _error; } @@ -7832,7 +7835,7 @@ static tsdbReadHandleT doCreateDataReadHandle(STableScanPhyNode* pTableScanNode, SArray* pa = taosArrayInit(1, sizeof(STableKeyInfo)); - STableKeyInfo info = {.pTable = NULL, .lastKey = 0, .uid = uid}; + STableKeyInfo info = {.lastKey = 0, .uid = uid}; taosArrayPush(pa, &info); taosArrayPush(groupInfo.pGroupList, &pa); } @@ -8827,35 +8830,17 @@ void* freeColumnInfo(SColumnInfo* pColumnInfo, int32_t numOfCols) { } void doDestroyTask(SExecTaskInfo *pTaskInfo) { - qDebug("%s start to free execTask", GET_TASKID(pTaskInfo)); doDestroyTableQueryInfo(&pTaskInfo->tableqinfoGroupInfo); // taosArrayDestroy(pTaskInfo->summary.queryProfEvents); // taosHashCleanup(pTaskInfo->summary.operatorProfResults); + tfree(pTaskInfo->sql); + tfree(pTaskInfo->id.str); qDebug("%s execTask is freed", GET_TASKID(pTaskInfo)); + tfree(pTaskInfo); } -bool doBuildResCheck(SQInfo* pQInfo) { - bool buildRes = false; - - pthread_mutex_lock(&pQInfo->lock); - - pQInfo->dataReady = QUERY_RESULT_READY; - buildRes = needBuildResAfterQueryComplete(pQInfo); - - // clear qhandle owner, it must be in the secure area. other thread may run ahead before current, after it is - // put into task to be executed. - assert(pQInfo->owner == taosGetSelfPthreadId()); - pQInfo->owner = 0; - - pthread_mutex_unlock(&pQInfo->lock); - - // used in retrieve blocking model. - tsem_post(&pQInfo->ready); - return buildRes; -} - static void doSetTagValueToResultBuf(char* output, const char* val, int16_t type, int16_t bytes) { if (val == NULL) { setNull(output, type, bytes); diff --git a/source/libs/qworker/inc/qworkerInt.h b/source/libs/qworker/inc/qworkerInt.h index d07e20d5c1..a19a02069b 100644 --- a/source/libs/qworker/inc/qworkerInt.h +++ b/source/libs/qworker/inc/qworkerInt.h @@ -137,6 +137,7 @@ typedef struct SQWorkerMgmt { SHashObj *ctxHash; //key: queryId+taskId, value: SQWTaskCtx void *nodeObj; putReqToQueryQFp putToQueueFp; + sendReqToDnodeFp sendReqFp; } SQWorkerMgmt; #define QW_FPARAMS_DEF SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId diff --git a/source/libs/qworker/src/qworker.c b/source/libs/qworker/src/qworker.c index c28e30333c..8fc8a783c4 100644 --- a/source/libs/qworker/src/qworker.c +++ b/source/libs/qworker/src/qworker.c @@ -1313,12 +1313,13 @@ _return: QW_RET(code); } -int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qWorkerMgmt, void *nodeObj, putReqToQueryQFp fp) { - if (NULL == qWorkerMgmt || NULL == nodeObj || NULL == fp) { +int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qWorkerMgmt, void *nodeObj, + putReqToQueryQFp fp1, sendReqToDnodeFp fp2) { + if (NULL == qWorkerMgmt || NULL == nodeObj || NULL == fp1 || NULL == fp2) { qError("invalid param to init qworker"); QW_RET(TSDB_CODE_QRY_INVALID_INPUT); } - + SQWorkerMgmt *mgmt = calloc(1, sizeof(SQWorkerMgmt)); if (NULL == mgmt) { qError("calloc %d failed", (int32_t)sizeof(SQWorkerMgmt)); @@ -1361,7 +1362,8 @@ int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qW mgmt->nodeType = nodeType; mgmt->nodeId = nodeId; mgmt->nodeObj = nodeObj; - mgmt->putToQueueFp = fp; + mgmt->putToQueueFp = fp1; + mgmt->sendReqFp = fp2; *qWorkerMgmt = mgmt; diff --git a/source/libs/qworker/src/qworkerMsg.c b/source/libs/qworker/src/qworkerMsg.c index 56c882f404..11783335b3 100644 --- a/source/libs/qworker/src/qworkerMsg.c +++ b/source/libs/qworker/src/qworkerMsg.c @@ -421,6 +421,11 @@ int32_t qWorkerProcessFetchMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) { return TSDB_CODE_SUCCESS; } +int32_t qWorkerProcessFetchRsp(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) { + qProcessFetchRsp(NULL, pMsg, NULL); + return TSDB_CODE_SUCCESS; +} + int32_t qWorkerProcessCancelMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) { if (NULL == node || NULL == qWorkerMgmt || NULL == pMsg) { return TSDB_CODE_QRY_INVALID_INPUT; diff --git a/source/libs/qworker/test/qworkerTests.cpp b/source/libs/qworker/test/qworkerTests.cpp index b9d84b25aa..4c81123bf2 100644 --- a/source/libs/qworker/test/qworkerTests.cpp +++ b/source/libs/qworker/test/qworkerTests.cpp @@ -1081,7 +1081,7 @@ TEST(rcTest, shortExecshortDelay) { qwtTestStop = false; qwtTestQuitThreadNum = 0; - code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue); + code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue, NULL); ASSERT_EQ(code, 0); qwtTestMaxExecTaskUsec = 0; @@ -1162,7 +1162,7 @@ TEST(rcTest, longExecshortDelay) { qwtTestStop = false; qwtTestQuitThreadNum = 0; - code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue); + code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue, NULL); ASSERT_EQ(code, 0); qwtTestMaxExecTaskUsec = 1000000; @@ -1245,7 +1245,7 @@ TEST(rcTest, shortExeclongDelay) { qwtTestStop = false; qwtTestQuitThreadNum = 0; - code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue); + code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue, NULL); ASSERT_EQ(code, 0); qwtTestMaxExecTaskUsec = 0; diff --git a/source/libs/transport/inc/transComm.h b/source/libs/transport/inc/transComm.h index c7a23a2482..3f2aa1170e 100644 --- a/source/libs/transport/inc/transComm.h +++ b/source/libs/transport/inc/transComm.h @@ -202,6 +202,8 @@ bool transDecompressMsg(char* msg, int32_t len, int32_t* flen); void transConnCtxDestroy(STransConnCtx* ctx); void transFreeMsg(void* msg); + +// typedef struct SConnBuffer { char* buf; int len; @@ -209,4 +211,9 @@ typedef struct SConnBuffer { int left; } SConnBuffer; +int transInitBuffer(SConnBuffer* buf); +int transClearBuffer(SConnBuffer* buf); +int transDestroyBuffer(SConnBuffer* buf); +int transAllocBuffer(SConnBuffer* connBuf, uv_buf_t* uvBuf); + #endif diff --git a/source/libs/transport/src/trans.c b/source/libs/transport/src/trans.c index bc9a9de318..91f9a8ead2 100644 --- a/source/libs/transport/src/trans.c +++ b/source/libs/transport/src/trans.c @@ -30,7 +30,8 @@ void* rpcOpen(const SRpcInit* pInit) { tstrncpy(pRpc->label, pInit->label, strlen(pInit->label)); } pRpc->cfp = pInit->cfp; - pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads; + // pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads; + pRpc->numOfThreads = pInit->numOfThreads; pRpc->connType = pInit->connType; pRpc->idleTime = pInit->idleTime; pRpc->tcphandle = (*taosInitHandle[pRpc->connType])(0, pInit->localPort, pRpc->label, pRpc->numOfThreads, NULL, pRpc); @@ -55,7 +56,13 @@ void* rpcMallocCont(int contLen) { } return start + sizeof(STransMsgHead); } -void rpcFreeCont(void* cont) { return; } +void rpcFreeCont(void* cont) { + // impl + if (cont == NULL) { + return; + } + free((char*)cont - TRANS_MSG_OVERHEAD); +} void* rpcReallocCont(void* ptr, int contLen) { return NULL; } void rpcSendRedirectRsp(void* pConn, const SEpSet* pEpSet) {} diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index f265acf8c1..d64df9b0f3 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -31,6 +31,7 @@ typedef struct SCliConn { char secured; uint64_t expireTime; int8_t notifyCount; // timers already notify to client + int32_t ref; } SCliConn; typedef struct SCliMsg { @@ -94,8 +95,10 @@ static void clientHandleExcept(SCliConn* conn); // handle req from app static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd); -static void clientMsgDestroy(SCliMsg* pMsg); -static void destroyTransConnCtx(STransConnCtx* ctx); +static void destroyUserdata(SRpcMsg* userdata); + +static void destroyCmsg(SCliMsg* cmsg); +static void transDestroyConnCtx(STransConnCtx* ctx); // thread obj static SCliThrdObj* createThrdObj(); static void destroyThrdObj(SCliThrdObj* pThrd); @@ -103,7 +106,8 @@ static void destroyThrdObj(SCliThrdObj* pThrd); static void* clientThread(void* arg); static void clientHandleResp(SCliConn* conn) { - STransConnCtx* pCtx = ((SCliMsg*)conn->data)->ctx; + SCliMsg* pMsg = conn->data; + STransConnCtx* pCtx = pMsg->ctx; SRpcInfo* pRpc = pCtx->pTransInst; STransMsgHead* pHead = (STransMsgHead*)(conn->readBuf.buf); @@ -112,41 +116,53 @@ static void clientHandleResp(SCliConn* conn) { SRpcMsg rpcMsg; rpcMsg.contLen = transContLenFromMsg(pHead->msgLen); - rpcMsg.pCont = transContFromHead(pHead); + rpcMsg.pCont = transContFromHead((char*)pHead); rpcMsg.code = pHead->code; rpcMsg.msgType = pHead->msgType; rpcMsg.ahandle = pCtx->ahandle; + tDebug("conn %p handle resp", conn); (pRpc->cfp)(NULL, &rpcMsg, NULL); conn->notifyCount += 1; + // buf's mem alread translated to rpcMsg.pCont + transClearBuffer(&conn->readBuf); + + uv_read_start((uv_stream_t*)conn->stream, clientAllocBufferCb, clientReadCb); + SCliThrdObj* pThrd = conn->hostThrd; - tfree(conn->data); addConnToPool(pThrd->pool, pCtx->ip, pCtx->port, conn); + destroyCmsg(pMsg); + conn->data = NULL; // start thread's timer of conn pool if not active if (!uv_is_active((uv_handle_t*)pThrd->pTimer) && pRpc->idleTime > 0) { uv_timer_start((uv_timer_t*)pThrd->pTimer, clientTimeoutCb, CONN_PERSIST_TIME(pRpc->idleTime) / 2, 0); } - destroyTransConnCtx(pCtx); } static void clientHandleExcept(SCliConn* pConn) { + if (pConn->data == NULL) { + // handle conn except in conn pool + clientConnDestroy(pConn, true); + return; + } + tDebug("conn %p start to destroy", pConn); SCliMsg* pMsg = pConn->data; - STransConnCtx* pCtx = pMsg->ctx; - SRpcInfo* pRpc = pCtx->pTransInst; + destroyUserdata(&pMsg->msg); - transFreeMsg((pMsg->msg.pCont)); - pMsg->msg.pCont = NULL; + STransConnCtx* pCtx = pMsg->ctx; SRpcMsg rpcMsg = {0}; rpcMsg.ahandle = pCtx->ahandle; rpcMsg.code = -1; // SRpcInfo* pRpc = pMsg->ctx->pRpc; - (pRpc->cfp)(NULL, &rpcMsg, NULL); - tfree(pConn->data); + (pCtx->pTransInst->cfp)(NULL, &rpcMsg, NULL); pConn->notifyCount += 1; - destroyTransConnCtx(pCtx); + + destroyCmsg(pMsg); + pConn->data = NULL; + // transDestroyConnCtx(pCtx); clientConnDestroy(pConn, true); } @@ -213,14 +229,20 @@ static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port) { } queue* h = QUEUE_HEAD(&plist->conn); QUEUE_REMOVE(h); - return QUEUE_DATA(h, SCliConn, conn); + + SCliConn* conn = QUEUE_DATA(h, SCliConn, conn); + QUEUE_INIT(&conn->conn); + return conn; } static void addConnToPool(void* pool, char* ip, uint32_t port, SCliConn* conn) { char key[128] = {0}; + tstrncpy(key, ip, strlen(ip)); tstrncpy(key + strlen(key), (char*)(&port), sizeof(port)); + tDebug("conn %p added to conn pool, read buf cap: %d", conn, conn->readBuf.cap); SRpcInfo* pRpc = ((SCliThrdObj*)conn->hostThrd)->pTransInst; + conn->expireTime = taosGetTimestampMs() + CONN_PERSIST_TIME(pRpc->idleTime); SConnList* plist = taosHashGet((SHashObj*)pool, key, strlen(key)); conn->notifyCount = 0; @@ -237,40 +259,18 @@ static bool clientReadComplete(SConnBuffer* data) { if (msgLen > data->len) { data->left = msgLen - data->len; return false; - } else { + } else if (msgLen == data->len) { + data->left = 0; return true; } } else { return false; } } -static void clientAllocReadBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) { - // impl later - static const int CAPACITY = 512; - +static void clientAllocBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) { SCliConn* conn = handle->data; SConnBuffer* pBuf = &conn->readBuf; - if (pBuf->cap == 0) { - pBuf->buf = (char*)calloc(1, CAPACITY * sizeof(char)); - pBuf->len = 0; - pBuf->cap = CAPACITY; - pBuf->left = -1; - - buf->base = pBuf->buf; - buf->len = CAPACITY; - } else { - if (pBuf->len >= pBuf->cap) { - if (pBuf->left == -1) { - pBuf->cap *= 2; - pBuf->buf = realloc(pBuf->buf, pBuf->cap); - } else if (pBuf->len + pBuf->left > pBuf->cap) { - pBuf->cap = pBuf->len + pBuf->left; - pBuf->buf = realloc(pBuf->buf, pBuf->cap); - } - } - buf->base = pBuf->buf + pBuf->len; - buf->len = pBuf->cap - pBuf->len; - } + transAllocBuffer(pBuf, buf); } static void clientReadCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) { // impl later @@ -279,6 +279,7 @@ static void clientReadCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf if (nread > 0) { pBuf->len += nread; if (clientReadComplete(pBuf)) { + uv_read_stop((uv_stream_t*)conn->stream); tDebug("conn %p read complete", conn); clientHandleResp(conn); } else { @@ -288,10 +289,12 @@ static void clientReadCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf } assert(nread <= 0); if (nread == 0) { - tError("conn %p closed", conn); + // ref http://docs.libuv.org/en/v1.x/stream.html?highlight=uv_read_start#c.uv_read_cb + // nread might be 0, which does not indicate an error or EOF. This is equivalent to EAGAIN or EWOULDBLOCK under + // read(2). return; } - if (nread < 0) { + if (nread < 0 || nread == UV_EOF) { tError("conn %p read error: %s", conn, uv_err_name(nread)); clientHandleExcept(conn); } @@ -300,43 +303,47 @@ static void clientReadCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf } static void clientConnDestroy(SCliConn* conn, bool clear) { - tDebug("conn %p destroy", conn); - if (clear) { - uv_close((uv_handle_t*)conn->stream, NULL); + // + conn->ref--; + if (conn->ref == 0) { + tDebug("conn %p remove from conn pool", conn); + QUEUE_REMOVE(&conn->conn); + tDebug("conn %p remove from conn pool successfully", conn); + if (clear) { + uv_close((uv_handle_t*)conn->stream, clientDestroy); + } } - free(conn->stream); - free(conn->readBuf.buf); - free(conn->writeReq); - free(conn); } static void clientDestroy(uv_handle_t* handle) { SCliConn* conn = handle->data; - // QUEUE_REMOVE(&conn->conn); - clientConnDestroy(conn, false); + // transDestroyBuffer(&conn->readBuf); + + free(conn->stream); + free(conn->writeReq); + tDebug("conn %p destroy successfully", conn); + free(conn); + + // clientConnDestroy(conn, false); } static void clientWriteCb(uv_write_t* req, int status) { SCliConn* pConn = req->data; - - SCliMsg* pMsg = pConn->data; - transFreeMsg((pMsg->msg.pCont)); - pMsg->msg.pCont = NULL; - if (status == 0) { tDebug("conn %p data already was written out", pConn); + SCliMsg* pMsg = pConn->data; + if (pMsg == NULL) { + destroy + // handle + return; + } + destroyUserdata(&pMsg->msg); } else { tError("conn %p failed to write: %s", pConn, uv_err_name(status)); clientHandleExcept(pConn); return; } SCliThrdObj* pThrd = pConn->hostThrd; - // if (pConn->stream == NULL) { - // pConn->stream = (uv_stream_t*)malloc(sizeof(uv_tcp_t)); - // uv_tcp_init(pThrd->loop, (uv_tcp_t*)pConn->stream); - // pConn->stream->data = pConn; - //} - uv_read_start((uv_stream_t*)pConn->stream, clientAllocReadBufferCb, clientReadCb); - // impl later + uv_read_start((uv_stream_t*)pConn->stream, clientAllocBufferCb, clientReadCb); } static void clientWrite(SCliConn* pConn) { @@ -381,14 +388,11 @@ static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) { tDebug("conn %p get from conn pool", conn); conn->data = pMsg; conn->writeReq->data = conn; - - conn->readBuf.len = 0; - memset(conn->readBuf.buf, 0, conn->readBuf.cap); - conn->readBuf.left = -1; + transDestroyBuffer(&conn->readBuf); clientWrite(conn); } else { SCliConn* conn = calloc(1, sizeof(SCliConn)); - + conn->ref++; // read/write stream handle conn->stream = (uv_stream_t*)malloc(sizeof(uv_tcp_t)); uv_tcp_init(pThrd->loop, (uv_tcp_t*)(conn->stream)); @@ -397,6 +401,7 @@ static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) { // write req handle conn->writeReq = malloc(sizeof(uv_write_t)); conn->writeReq->data = conn; + QUEUE_INIT(&conn->conn); conn->connReq.data = conn; @@ -459,8 +464,20 @@ void* taosInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads, } return cli; } -static void clientMsgDestroy(SCliMsg* pMsg) { - // impl later + +static void destroyUserdata(SRpcMsg* userdata) { + if (userdata->pCont == NULL) { + return; + } + transFreeMsg(userdata->pCont); + userdata->pCont = NULL; +} +static void destroyCmsg(SCliMsg* pMsg) { + if (pMsg == NULL) { + return; + } + transDestroyConnCtx(pMsg->ctx); + destroyUserdata(&pMsg->msg); free(pMsg); } static SCliThrdObj* createThrdObj() { @@ -493,7 +510,7 @@ static void destroyThrdObj(SCliThrdObj* pThrd) { free(pThrd); } -static void destroyTransConnCtx(STransConnCtx* ctx) { +static void transDestroyConnCtx(STransConnCtx* ctx) { if (ctx != NULL) { free(ctx->ip); } diff --git a/source/libs/transport/src/transComm.c b/source/libs/transport/src/transComm.c index 5bece11bec..ca39f85eb3 100644 --- a/source/libs/transport/src/transComm.c +++ b/source/libs/transport/src/transComm.c @@ -198,4 +198,51 @@ void transFreeMsg(void* msg) { } free((char*)msg - sizeof(STransMsgHead)); } + +int transInitBuffer(SConnBuffer* buf) { + transClearBuffer(buf); + return 0; +} +int transClearBuffer(SConnBuffer* buf) { + memset(buf, 0, sizeof(*buf)); + return 0; +} +int transAllocBuffer(SConnBuffer* connBuf, uv_buf_t* uvBuf) { + /* + * formate of data buffer: + * |<--------------------------data from socket------------------------------->| + * |<------STransMsgHead------->|<-------------------other data--------------->| + */ + static const int CAPACITY = 1024; + + SConnBuffer* p = connBuf; + if (p->cap == 0) { + p->buf = (char*)calloc(CAPACITY, sizeof(char)); + p->len = 0; + p->cap = CAPACITY; + p->left = -1; + + uvBuf->base = p->buf; + uvBuf->len = CAPACITY; + } else { + if (p->len >= p->cap) { + if (p->left == -1) { + p->cap *= 2; + p->buf = realloc(p->buf, p->cap); + } else if (p->len + p->left > p->cap) { + p->cap = p->len + p->left; + p->buf = realloc(p->buf, p->len + p->left); + } + } + uvBuf->base = p->buf + p->len; + uvBuf->len = p->cap - p->len; + } + return 0; +} +int transDestroyBuffer(SConnBuffer* buf) { + if (buf->cap > 0) { + tfree(buf->buf); + } + transClearBuffer(buf); +} #endif diff --git a/source/libs/transport/src/transSrv.c b/source/libs/transport/src/transSrv.c index c70b1a5b28..f36d9bd493 100644 --- a/source/libs/transport/src/transSrv.c +++ b/source/libs/transport/src/transSrv.c @@ -17,7 +17,7 @@ #include "transComm.h" -typedef struct SConn { +typedef struct SSrvConn { uv_tcp_t* pTcp; uv_write_t* pWriter; uv_timer_t* pTimer; @@ -26,13 +26,14 @@ typedef struct SConn { queue queue; int ref; int persist; // persist connection or not - SConnBuffer connBuf; // read buf, + SConnBuffer readBuf; // read buf, int inType; void* pTransInst; // rpc init void* ahandle; // void* hostThrd; + void* pSrvMsg; - SRpcMsg sendMsg; + // SRpcMsg sendMsg; // del later char secured; int spi; @@ -40,7 +41,13 @@ typedef struct SConn { char user[TSDB_UNI_LEN]; // user ID for the link char secret[TSDB_PASSWORD_LEN]; char ckey[TSDB_PASSWORD_LEN]; // ciphering key -} SConn; +} SSrvConn; + +typedef struct SSrvMsg { + SSrvConn* pConn; + SRpcMsg msg; + queue q; +} SSrvMsg; typedef struct SWorkThrdObj { pthread_t thread; @@ -48,8 +55,8 @@ typedef struct SWorkThrdObj { int fd; uv_loop_t* loop; uv_async_t* workerAsync; // - queue conn; - pthread_mutex_t connMtx; + queue msg; + pthread_mutex_t msgMtx; void* pTransInst; } SWorkThrdObj; @@ -68,9 +75,9 @@ typedef struct SServerObj { static const char* notify = "a"; // refactor later -static int transAddAuthPart(SConn* pConn, char* msg, int msgLen); +static int transAddAuthPart(SSrvConn* pConn, char* msg, int msgLen); -static int uvAuthMsg(SConn* pConn, char* msg, int msgLen); +static int uvAuthMsg(SSrvConn* pConn, char* msg, int msgLen); static void uvAllocConnBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf); static void uvAllocReadBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf); @@ -82,12 +89,13 @@ static void uvOnAcceptCb(uv_stream_t* stream, int status); static void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf); static void uvWorkerAsyncCb(uv_async_t* handle); -static void uvPrepareSendData(SConn* conn, uv_buf_t* wb); - +static void uvPrepareSendData(SSrvMsg* msg, uv_buf_t* wb); +static void uvStartSendResp(SSrvMsg* msg); +static void destroySmsg(SSrvMsg* smsg); // check whether already read complete packet -static bool readComplete(SConnBuffer* buf); -static SConn* createConn(); -static void destroyConn(SConn* conn, bool clear /*clear handle or not*/); +static bool readComplete(SConnBuffer* buf); +static SSrvConn* createConn(); +static void destroyConn(SSrvConn* conn, bool clear /*clear handle or not*/); static void uvDestroyConn(uv_handle_t* handle); @@ -105,31 +113,9 @@ void uvAllocReadBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* b * |<--------------------------data from socket------------------------------->| * |<------STransMsgHead------->|<-------------------other data--------------->| */ - static const int CAPACITY = 1024; - - SConn* conn = handle->data; - SConnBuffer* pBuf = &conn->connBuf; - if (pBuf->cap == 0) { - pBuf->buf = (char*)calloc(CAPACITY, sizeof(char)); - pBuf->len = 0; - pBuf->cap = CAPACITY; - pBuf->left = -1; - - buf->base = pBuf->buf; - buf->len = CAPACITY; - } else { - if (pBuf->len >= pBuf->cap) { - if (pBuf->left == -1) { - pBuf->cap *= 2; - pBuf->buf = realloc(pBuf->buf, pBuf->cap); - } else if (pBuf->len + pBuf->left > pBuf->cap) { - pBuf->cap = pBuf->len + pBuf->left; - pBuf->buf = realloc(pBuf->buf, pBuf->len + pBuf->left); - } - } - buf->base = pBuf->buf + pBuf->len; - buf->len = pBuf->cap - pBuf->len; - } + SSrvConn* conn = handle->data; + SConnBuffer* pBuf = &conn->readBuf; + transAllocBuffer(pBuf, buf); } // check data read from socket completely or not @@ -159,7 +145,7 @@ static bool readComplete(SConnBuffer* data) { // // impl later // STransMsgHead* pHead = (STransMsgHead*)pRecv->msg; // SRpcInfo* pRpc = (SRpcInfo*)pRecv->shandle; -// SConn* pConn = pRecv->thandle; +// SSrvConn* pConn = pRecv->thandle; // tDump(pRecv->msg, pRecv->msgLen); // terrno = 0; // // SRpcReqContext* pContest; @@ -167,7 +153,7 @@ static bool readComplete(SConnBuffer* data) { // // do auth and check //} -static int uvAuthMsg(SConn* pConn, char* msg, int len) { +static int uvAuthMsg(SSrvConn* pConn, char* msg, int len) { STransMsgHead* pHead = (STransMsgHead*)msg; int code = 0; @@ -222,14 +208,14 @@ static int uvAuthMsg(SConn* pConn, char* msg, int len) { // refers specifically to query or insert timeout static void uvHandleActivityTimeout(uv_timer_t* handle) { - SConn* conn = handle->data; + SSrvConn* conn = handle->data; tDebug("%p timeout since no activity", conn); } -static void uvHandleReq(SConn* pConn) { +static void uvHandleReq(SSrvConn* pConn) { SRecvInfo info; SRecvInfo* p = &info; - SConnBuffer* pBuf = &pConn->connBuf; + SConnBuffer* pBuf = &pConn->readBuf; p->msg = pBuf->buf; p->msgLen = pBuf->len; p->ip = 0; @@ -255,7 +241,6 @@ static void uvHandleReq(SConn* pConn) { pHead->code = htonl(pHead->code); int32_t dlen = 0; - SRpcMsg rpcMsg; if (transDecompressMsg(NULL, 0, NULL)) { // add compress later // pHead = rpcDecompressRpcMsg(pHead); @@ -264,6 +249,8 @@ static void uvHandleReq(SConn* pConn) { // impl later // } + + SRpcMsg rpcMsg; rpcMsg.contLen = transContLenFromMsg(pHead->msgLen); rpcMsg.pCont = pHead->content; rpcMsg.msgType = pHead->msgType; @@ -271,6 +258,7 @@ static void uvHandleReq(SConn* pConn) { rpcMsg.ahandle = NULL; rpcMsg.handle = pConn; + transClearBuffer(&pConn->readBuf); pConn->ref++; (*(pRpc->cfp))(pRpc->parent, &rpcMsg, NULL); // uv_timer_start(pConn->pTimer, uvHandleActivityTimeout, pRpc->idleTime * 10000, 0); @@ -280,8 +268,8 @@ static void uvHandleReq(SConn* pConn) { void uvOnReadCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf) { // opt - SConn* conn = cli->data; - SConnBuffer* pBuf = &conn->connBuf; + SSrvConn* conn = cli->data; + SConnBuffer* pBuf = &conn->readBuf; if (nread > 0) { pBuf->len += nread; tDebug("conn %p read summroy, total read: %d, current read: %d", conn, pBuf->len, (int)nread); @@ -294,11 +282,12 @@ void uvOnReadCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf) { return; } if (nread == 0) { - tDebug("conn %p except read", conn); - // destroyConn(conn, true); return; } - if (nread != UV_EOF) { + if (nread < 0 || nread != UV_EOF) { + if (conn->ref > 1) { + conn->ref++; // ref > 1 signed that write is in progress + } tDebug("conn %p read error: %s", conn, uv_err_name(nread)); destroyConn(conn, true); } @@ -310,25 +299,23 @@ void uvAllocConnBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* b void uvOnTimeoutCb(uv_timer_t* handle) { // opt - SConn* pConn = handle->data; + SSrvConn* pConn = handle->data; tDebug("conn %p time out", pConn); } void uvOnWriteCb(uv_write_t* req, int status) { - SConn* conn = req->data; + SSrvConn* conn = req->data; - SConnBuffer* buf = &conn->connBuf; - buf->len = 0; - memset(buf->buf, 0, buf->cap); - buf->left = -1; - - SRpcMsg* pMsg = &conn->sendMsg; - transFreeMsg(pMsg->pCont); + SSrvMsg* smsg = conn->pSrvMsg; + destroySmsg(smsg); + conn->pSrvMsg = NULL; + transClearBuffer(&conn->readBuf); if (status == 0) { tDebug("conn %p data already was written on stream", conn); } else { tDebug("conn %p failed to write data, %s", conn, uv_err_name(status)); + // destroyConn(conn, true); } // opt @@ -341,16 +328,16 @@ static void uvOnPipeWriteCb(uv_write_t* req, int status) { } } -static void uvPrepareSendData(SConn* conn, uv_buf_t* wb) { +static void uvPrepareSendData(SSrvMsg* smsg, uv_buf_t* wb) { // impl later; - tDebug("conn %p prepare to send resp", conn); - SRpcMsg* pMsg = &conn->sendMsg; + tDebug("conn %p prepare to send resp", smsg->pConn); + SRpcMsg* pMsg = &smsg->msg; if (pMsg->pCont == 0) { pMsg->pCont = (void*)rpcMallocCont(0); pMsg->contLen = 0; } STransMsgHead* pHead = transHeadFromCont(pMsg->pCont); - pHead->msgType = conn->inType + 1; + pHead->msgType = smsg->pConn->inType + 1; // add more info char* msg = (char*)pHead; int32_t len = transMsgLenFromCont(pMsg->contLen); @@ -361,28 +348,53 @@ static void uvPrepareSendData(SConn* conn, uv_buf_t* wb) { wb->base = msg; wb->len = len; } +static void uvStartSendResp(SSrvMsg* smsg) { + // impl + uv_buf_t wb; + uvPrepareSendData(smsg, &wb); + + SSrvConn* pConn = smsg->pConn; + uv_timer_stop(pConn->pTimer); + + pConn->pSrvMsg = smsg; + // conn->pWriter->data = smsg; + uv_write(pConn->pWriter, (uv_stream_t*)pConn->pTcp, &wb, 1, uvOnWriteCb); + + // SRpcMsg* rpcMsg = smsg->msg; + + return; +} +static void destroySmsg(SSrvMsg* smsg) { + if (smsg == NULL) { + return; + } + transFreeMsg(smsg->msg.pCont); + free(smsg); +} void uvWorkerAsyncCb(uv_async_t* handle) { SWorkThrdObj* pThrd = handle->data; - SConn* conn = NULL; + SSrvConn* conn = NULL; queue wq; // batch process to avoid to lock/unlock frequently - pthread_mutex_lock(&pThrd->connMtx); - QUEUE_MOVE(&pThrd->conn, &wq); - pthread_mutex_unlock(&pThrd->connMtx); + pthread_mutex_lock(&pThrd->msgMtx); + QUEUE_MOVE(&pThrd->msg, &wq); + pthread_mutex_unlock(&pThrd->msgMtx); while (!QUEUE_IS_EMPTY(&wq)) { queue* head = QUEUE_HEAD(&wq); QUEUE_REMOVE(head); - SConn* conn = QUEUE_DATA(head, SConn, queue); - if (conn == NULL) { - tError("except occurred, do nothing"); - return; - } - uv_buf_t wb; - uvPrepareSendData(conn, &wb); - uv_timer_stop(conn->pTimer); - uv_write(conn->pWriter, (uv_stream_t*)conn->pTcp, &wb, 1, uvOnWriteCb); + SSrvMsg* msg = QUEUE_DATA(head, SSrvMsg, q); + if (msg == NULL) { + tError("except occurred, continue"); + continue; + } + uvStartSendResp(msg); + // uv_buf_t wb; + // uvPrepareSendData(msg, &wb); + // uv_timer_stop(conn->pTimer); + + // uv_write(conn->pWriter, (uv_stream_t*)conn->pTcp, &wb, 1, uvOnWriteCb); } } @@ -435,7 +447,7 @@ void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) { uv_handle_type pending = uv_pipe_pending_type(pipe); assert(pending == UV_TCP); - SConn* pConn = createConn(); + SSrvConn* pConn = createConn(); pConn->pTransInst = pThrd->pTransInst; /* init conn timer*/ @@ -484,8 +496,8 @@ static bool addHandleToWorkloop(void* arg) { pThrd->pipe->data = pThrd; - QUEUE_INIT(&pThrd->conn); - pthread_mutex_init(&pThrd->connMtx, NULL); + QUEUE_INIT(&pThrd->msg); + pthread_mutex_init(&pThrd->msgMtx, NULL); pThrd->workerAsync = malloc(sizeof(uv_async_t)); uv_async_init(pThrd->loop, pThrd->workerAsync, uvWorkerAsyncCb); @@ -523,34 +535,43 @@ void* workerThread(void* arg) { uv_run(pThrd->loop, UV_RUN_DEFAULT); } -static SConn* createConn() { - SConn* pConn = (SConn*)calloc(1, sizeof(SConn)); +static SSrvConn* createConn() { + SSrvConn* pConn = (SSrvConn*)calloc(1, sizeof(SSrvConn)); + tDebug("conn %p created", pConn); ++pConn->ref; return pConn; } -static void destroyConn(SConn* conn, bool clear) { +static void destroyConn(SSrvConn* conn, bool clear) { if (conn == NULL) { return; } - if (--conn->ref == 0) { + // SRpcMsg* pMsg = &conn->sendMsg; + // transFreeMsg(pMsg->pCont); + // pMsg->pCont = NULL; + + tDebug("conn %p try to destroy", conn); + if (--conn->ref > 0) { return; } + transDestroyBuffer(&conn->readBuf); + destroySmsg(conn->pSrvMsg); + conn->pSrvMsg = NULL; + if (clear) { - uv_close((uv_handle_t*)conn->pTcp, NULL); + uv_close((uv_handle_t*)conn->pTcp, uvDestroyConn); } +} +static void uvDestroyConn(uv_handle_t* handle) { + SSrvConn* conn = handle->data; + tDebug("conn %p destroy", conn); uv_timer_stop(conn->pTimer); free(conn->pTimer); - free(conn->pTcp); - free(conn->connBuf.buf); + // free(conn->pTcp); free(conn->pWriter); free(conn); } -static void uvDestroyConn(uv_handle_t* handle) { - SConn* conn = handle->data; - destroyConn(conn, false); -} -static int transAddAuthPart(SConn* pConn, char* msg, int msgLen) { +static int transAddAuthPart(SSrvConn* pConn, char* msg, int msgLen) { STransMsgHead* pHead = (STransMsgHead*)msg; if (pConn->spi && pConn->secured == 0) { @@ -632,6 +653,7 @@ void destroyWorkThrd(SWorkThrdObj* pThrd) { pthread_join(pThrd->thread, NULL); // free(srv->pipe[i]); free(pThrd->loop); + pthread_mutex_destroy(&pThrd->msgMtx); free(pThrd); } void taosCloseServer(void* arg) { @@ -648,17 +670,20 @@ void taosCloseServer(void* arg) { } void rpcSendResponse(const SRpcMsg* pMsg) { - SConn* pConn = pMsg->handle; + SSrvConn* pConn = pMsg->handle; SWorkThrdObj* pThrd = pConn->hostThrd; - // opt later - pConn->sendMsg = *pMsg; - pthread_mutex_lock(&pThrd->connMtx); - QUEUE_PUSH(&pThrd->conn, &pConn->queue); - pthread_mutex_unlock(&pThrd->connMtx); + SSrvMsg* srvMsg = calloc(1, sizeof(SSrvMsg)); + srvMsg->pConn = pConn; + srvMsg->msg = *pMsg; + + pthread_mutex_lock(&pThrd->msgMtx); + QUEUE_PUSH(&pThrd->msg, &srvMsg->q); + pthread_mutex_unlock(&pThrd->msgMtx); + tDebug("conn %p start to send resp", pConn); - uv_async_send(pConn->pWorkerAsync); + uv_async_send(pThrd->workerAsync); } #endif diff --git a/source/libs/transport/test/rclient.c b/source/libs/transport/test/rclient.c index 84814f39fc..f9ad20c065 100644 --- a/source/libs/transport/test/rclient.c +++ b/source/libs/transport/test/rclient.c @@ -64,6 +64,7 @@ static void *sendRequest(void *param) { // tsem_wait(&pInfo->rspSem); tsem_wait(&pInfo->rspSem); tDebug("recv response succefully"); + // usleep(100000000); } diff --git a/source/libs/transport/test/rserver.c b/source/libs/transport/test/rserver.c index 12d8a01819..d1a587f4e5 100644 --- a/source/libs/transport/test/rserver.c +++ b/source/libs/transport/test/rserver.c @@ -165,6 +165,7 @@ int main(int argc, char *argv[]) { tError("failed to start RPC server"); return -1; } + // sleep(5); tInfo("RPC server is running, ctrl-c to exit"); @@ -172,7 +173,6 @@ int main(int argc, char *argv[]) { dataFd = open(dataName, O_APPEND | O_CREAT | O_WRONLY, S_IRWXU | S_IRWXG | S_IRWXO); if (dataFd < 0) tInfo("failed to open data file, reason:%s", strerror(errno)); } - qhandle = taosOpenQueue(); qset = taosOpenQset(); taosAddIntoQset(qset, qhandle, NULL); diff --git a/source/libs/wal/CMakeLists.txt b/source/libs/wal/CMakeLists.txt index 4d2dd97c87..bcf759e04f 100644 --- a/source/libs/wal/CMakeLists.txt +++ b/source/libs/wal/CMakeLists.txt @@ -11,6 +11,7 @@ target_link_libraries( PUBLIC cjson PUBLIC os PUBLIC util + PUBLIC common ) if(${BUILD_TEST}) diff --git a/source/libs/wal/src/walWrite.c b/source/libs/wal/src/walWrite.c index 2bc328b4e2..a4b34dee37 100644 --- a/source/libs/wal/src/walWrite.c +++ b/source/libs/wal/src/walWrite.c @@ -257,7 +257,7 @@ static int walWriteIndex(SWal *pWal, int64_t ver, int64_t offset) { return 0; } -int64_t walWrite(SWal *pWal, int64_t index, uint8_t msgType, const void *body, int32_t bodyLen) { +int64_t walWrite(SWal *pWal, int64_t index, tmsg_t msgType, const void *body, int32_t bodyLen) { if (pWal == NULL) return -1; int code = 0; diff --git a/tests/script/tsim/dnode/basic1.sim b/tests/script/tsim/dnode/basic1.sim index e689bb3261..6061b6ece1 100644 --- a/tests/script/tsim/dnode/basic1.sim +++ b/tests/script/tsim/dnode/basic1.sim @@ -178,9 +178,9 @@ if $rows != 3 then endi sql select * from st -#if $rows != 15 then -# return -1 -#endi +if $rows != 15 then + return -1 +endi print =============== drop dnode sql drop dnode 2; diff --git a/tests/script/tsim/table/basic1.sim b/tests/script/tsim/table/basic1.sim index f47630b737..776b09b813 100644 --- a/tests/script/tsim/table/basic1.sim +++ b/tests/script/tsim/table/basic1.sim @@ -139,9 +139,9 @@ endi print =============== query data frpm st sql select * from st -#if $rows != 21 then -# return -1 -#endi +if $rows != 21 then + return -1 +endi system sh/exec.sh -n dnode1 -s stop -x SIGINT system sh/exec.sh -n dnode1 -s start @@ -200,8 +200,8 @@ endi print =============== query data frpm st sql select * from st -#if $rows != 21 then -# return -1 -#endi +if $rows != 21 then + return -1 +endi system sh/exec.sh -n dnode1 -s stop -x SIGINT \ No newline at end of file