Merge remote-tracking branch 'origin/fix/hb' into fix/tsim

This commit is contained in:
Shengliang Guan 2022-04-24 17:59:19 +08:00
commit fcd0d2722f
22 changed files with 1558 additions and 1211 deletions

View File

@ -38,11 +38,12 @@ typedef struct SRpcConnInfo {
typedef struct SRpcMsg { typedef struct SRpcMsg {
tmsg_t msgType; tmsg_t msgType;
void *pCont; void * pCont;
int contLen; int contLen;
int32_t code; int32_t code;
void *handle; // rpc handle returned to app void * handle; // rpc handle returned to app
void *ahandle; // app handle set by client void * ahandle; // app handle set by client
int64_t refId; // refid, used by server
int noResp; // has response or not(default 0, 0: resp, 1: no resp); int noResp; // has response or not(default 0, 0: resp, 1: no resp);
int persistHandle; // persist handle or not int persistHandle; // persist handle or not
@ -54,8 +55,8 @@ typedef struct {
uint16_t clientPort; uint16_t clientPort;
SRpcMsg rpcMsg; SRpcMsg rpcMsg;
int32_t rspLen; int32_t rspLen;
void *pRsp; void * pRsp;
void *pNode; void * pNode;
} SNodeMsg; } SNodeMsg;
typedef void (*RpcCfp)(void *parent, SRpcMsg *, SEpSet *); typedef void (*RpcCfp)(void *parent, SRpcMsg *, SEpSet *);
@ -64,7 +65,7 @@ typedef int (*RpcRfp)(void *parent, SRpcMsg *, SEpSet *);
typedef struct SRpcInit { typedef struct SRpcInit {
uint16_t localPort; // local port uint16_t localPort; // local port
char *label; // for debug purpose char * label; // for debug purpose
int numOfThreads; // number of threads to handle connections int numOfThreads; // number of threads to handle connections
int sessions; // number of sessions allowed int sessions; // number of sessions allowed
int8_t connType; // TAOS_CONN_UDP, TAOS_CONN_TCPC, TAOS_CONN_TCPS int8_t connType; // TAOS_CONN_UDP, TAOS_CONN_TCPC, TAOS_CONN_TCPS
@ -97,23 +98,23 @@ typedef struct {
typedef struct { typedef struct {
int32_t msgType; int32_t msgType;
void *val; void * val;
int32_t (*clone)(void *src, void **dst); int32_t (*clone)(void *src, void **dst);
void (*freeFunc)(const void *arg); void (*freeFunc)(const void *arg);
} SRpcBrokenlinkVal; } SRpcBrokenlinkVal;
typedef struct { typedef struct {
SHashObj *args; SHashObj * args;
SRpcBrokenlinkVal brokenVal; SRpcBrokenlinkVal brokenVal;
} SRpcCtx; } SRpcCtx;
int32_t rpcInit(); int32_t rpcInit();
void rpcCleanup(); void rpcCleanup();
void *rpcOpen(const SRpcInit *pRpc); void * rpcOpen(const SRpcInit *pRpc);
void rpcClose(void *); void rpcClose(void *);
void *rpcMallocCont(int contLen); void * rpcMallocCont(int contLen);
void rpcFreeCont(void *pCont); void rpcFreeCont(void *pCont);
void *rpcReallocCont(void *ptr, int contLen); void * rpcReallocCont(void *ptr, int contLen);
// Because taosd supports multi-process mode // Because taosd supports multi-process mode
// These functions should not be used on the server side // These functions should not be used on the server side

View File

@ -59,6 +59,10 @@ static inline int32_t dmBuildMsg(SNodeMsg *pMsg, SRpcMsg *pRpc) {
pMsg->clientIp = connInfo.clientIp; pMsg->clientIp = connInfo.clientIp;
pMsg->clientPort = connInfo.clientPort; pMsg->clientPort = connInfo.clientPort;
memcpy(&pMsg->rpcMsg, pRpc, sizeof(SRpcMsg)); memcpy(&pMsg->rpcMsg, pRpc, sizeof(SRpcMsg));
if ((pRpc->msgType & 1u)) {
assert(pRpc->refId != 0);
}
// assert(pRpc->handle != NULL && pRpc->refId != 0 && pMsg->rpcMsg.refId != 0);
return 0; return 0;
} }
@ -107,7 +111,7 @@ _OVER:
} }
} }
SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = code}; SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = code, .refId = pRpc->refId};
tmsgSendRsp(&rsp); tmsgSendRsp(&rsp);
} }
dTrace("msg:%p, is freed", pMsg); dTrace("msg:%p, is freed", pMsg);
@ -134,7 +138,8 @@ static void dmProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
if (pDnode->status != DND_STAT_RUNNING) { if (pDnode->status != DND_STAT_RUNNING) {
dError("msg:%s ignored since dnode not running, handle:%p app:%p", TMSG_INFO(msgType), pMsg->handle, pMsg->ahandle); dError("msg:%s ignored since dnode not running, handle:%p app:%p", TMSG_INFO(msgType), pMsg->handle, pMsg->ahandle);
if (isReq) { if (isReq) {
SRpcMsg rspMsg = {.handle = pMsg->handle, .code = TSDB_CODE_APP_NOT_READY, .ahandle = pMsg->ahandle}; SRpcMsg rspMsg = {
.handle = pMsg->handle, .code = TSDB_CODE_APP_NOT_READY, .ahandle = pMsg->ahandle, .refId = pMsg->refId};
rpcSendResponse(&rspMsg); rpcSendResponse(&rspMsg);
} }
rpcFreeCont(pMsg->pCont); rpcFreeCont(pMsg->pCont);
@ -143,7 +148,8 @@ static void dmProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
if (isReq && pMsg->pCont == NULL) { if (isReq && pMsg->pCont == NULL) {
dError("req:%s not processed since its empty, handle:%p app:%p", TMSG_INFO(msgType), pMsg->handle, pMsg->ahandle); dError("req:%s not processed since its empty, handle:%p app:%p", TMSG_INFO(msgType), pMsg->handle, pMsg->ahandle);
SRpcMsg rspMsg = {.handle = pMsg->handle, .code = TSDB_CODE_INVALID_MSG_LEN, .ahandle = pMsg->ahandle}; SRpcMsg rspMsg = {
.handle = pMsg->handle, .code = TSDB_CODE_INVALID_MSG_LEN, .ahandle = pMsg->ahandle, .refId = pMsg->refId};
rpcSendResponse(&rspMsg); rpcSendResponse(&rspMsg);
return; return;
} }
@ -151,7 +157,8 @@ static void dmProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
if (pWrapper == NULL) { if (pWrapper == NULL) {
dError("msg:%s not processed since no handle, handle:%p app:%p", TMSG_INFO(msgType), pMsg->handle, pMsg->ahandle); dError("msg:%s not processed since no handle, handle:%p app:%p", TMSG_INFO(msgType), pMsg->handle, pMsg->ahandle);
if (isReq) { if (isReq) {
SRpcMsg rspMsg = {.handle = pMsg->handle, .code = TSDB_CODE_MSG_NOT_PROCESSED, .ahandle = pMsg->ahandle}; SRpcMsg rspMsg = {
.handle = pMsg->handle, .code = TSDB_CODE_MSG_NOT_PROCESSED, .ahandle = pMsg->ahandle, .refId = pMsg->refId};
rpcSendResponse(&rspMsg); rpcSendResponse(&rspMsg);
} }
rpcFreeCont(pMsg->pCont); rpcFreeCont(pMsg->pCont);
@ -170,6 +177,9 @@ static void dmProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
} }
dTrace("msg:%s will be processed by %s, app:%p", TMSG_INFO(msgType), pWrapper->name, pMsg->ahandle); dTrace("msg:%s will be processed by %s, app:%p", TMSG_INFO(msgType), pWrapper->name, pMsg->ahandle);
if (isReq) {
assert(pMsg->refId != 0);
}
dmProcessRpcMsg(pWrapper, pMsg, pEpSet); dmProcessRpcMsg(pWrapper, pMsg, pEpSet);
} }
@ -317,7 +327,7 @@ static void dmConsumeChildQueue(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int16_t
if (code != 0) { if (code != 0) {
dError("msg:%p, failed to process since code:0x%04x:%s", pMsg, code & 0XFFFF, tstrerror(code)); dError("msg:%p, failed to process since code:0x%04x:%s", pMsg, code & 0XFFFF, tstrerror(code));
if (pRpc->msgType & 1U) { if (pRpc->msgType & 1U) {
SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = terrno}; SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = terrno, .refId = pRpc->refId};
dmSendRsp(pWrapper, &rsp); dmSendRsp(pWrapper, &rsp);
} }

View File

@ -105,7 +105,7 @@ void dmStopMonitorThread(SDnode *pDnode) {
} }
static void dmProcessMgmtQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) { static void dmProcessMgmtQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
SDnode *pDnode = pInfo->ahandle; SDnode * pDnode = pInfo->ahandle;
SRpcMsg *pRpc = &pMsg->rpcMsg; SRpcMsg *pRpc = &pMsg->rpcMsg;
int32_t code = -1; int32_t code = -1;
dTrace("msg:%p, will be processed in dnode-mgmt queue", pMsg); dTrace("msg:%p, will be processed in dnode-mgmt queue", pMsg);
@ -150,7 +150,7 @@ static void dmProcessMgmtQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
if (pRpc->msgType & 1u) { if (pRpc->msgType & 1u) {
if (code != 0) code = terrno; if (code != 0) code = terrno;
SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = code}; SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = code, .refId = pRpc->refId};
rpcSendResponse(&rsp); rpcSendResponse(&rsp);
} }

View File

@ -177,7 +177,7 @@ void dmProcessServerStatusReq(SDnode *pDnode, SRpcMsg *pReq) {
SServerStatusRsp statusRsp = {0}; SServerStatusRsp statusRsp = {0};
dmGetServerStatus(pDnode, &statusRsp); dmGetServerStatus(pDnode, &statusRsp);
SRpcMsg rspMsg = {.handle = pReq->handle, .ahandle = pReq->ahandle}; SRpcMsg rspMsg = {.handle = pReq->handle, .ahandle = pReq->ahandle, .refId = pReq->refId};
int32_t rspLen = tSerializeSServerStatusRsp(NULL, 0, &statusRsp); int32_t rspLen = tSerializeSServerStatusRsp(NULL, 0, &statusRsp);
if (rspLen < 0) { if (rspLen < 0) {
rspMsg.code = TSDB_CODE_OUT_OF_MEMORY; rspMsg.code = TSDB_CODE_OUT_OF_MEMORY;

View File

@ -17,7 +17,8 @@
#include "bmInt.h" #include "bmInt.h"
static void bmSendErrorRsp(SNodeMsg *pMsg, int32_t code) { static void bmSendErrorRsp(SNodeMsg *pMsg, int32_t code) {
SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .ahandle = pMsg->rpcMsg.ahandle, .code = code}; SRpcMsg rpcRsp = {
.handle = pMsg->rpcMsg.handle, .ahandle = pMsg->rpcMsg.ahandle, .code = code, .refId = pMsg->rpcMsg.refId};
tmsgSendRsp(&rpcRsp); tmsgSendRsp(&rpcRsp);
dTrace("msg:%p, is freed", pMsg); dTrace("msg:%p, is freed", pMsg);
@ -38,6 +39,7 @@ static void bmSendErrorRsps(STaosQall *qall, int32_t numOfMsgs, int32_t code) {
static inline void bmSendRsp(SNodeMsg *pMsg, int32_t code) { static inline void bmSendRsp(SNodeMsg *pMsg, int32_t code) {
SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle, SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle,
.ahandle = pMsg->rpcMsg.ahandle, .ahandle = pMsg->rpcMsg.ahandle,
.refId = pMsg->rpcMsg.refId,
.code = code, .code = code,
.pCont = pMsg->pRsp, .pCont = pMsg->pRsp,
.contLen = pMsg->rspLen}; .contLen = pMsg->rspLen};
@ -101,7 +103,7 @@ static void bmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
} }
int32_t bmProcessWriteMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { int32_t bmProcessWriteMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SBnodeMgmt *pMgmt = pWrapper->pMgmt; SBnodeMgmt * pMgmt = pWrapper->pMgmt;
SMultiWorker *pWorker = &pMgmt->writeWorker; SMultiWorker *pWorker = &pMgmt->writeWorker;
dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name); dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);
@ -110,7 +112,7 @@ int32_t bmProcessWriteMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
} }
int32_t bmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { int32_t bmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SBnodeMgmt *pMgmt = pWrapper->pMgmt; SBnodeMgmt * pMgmt = pWrapper->pMgmt;
SSingleWorker *pWorker = &pMgmt->monitorWorker; SSingleWorker *pWorker = &pMgmt->monitorWorker;
dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name); dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);

View File

@ -19,6 +19,7 @@
static inline void mmSendRsp(SNodeMsg *pMsg, int32_t code) { static inline void mmSendRsp(SNodeMsg *pMsg, int32_t code) {
SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle, SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle,
.ahandle = pMsg->rpcMsg.ahandle, .ahandle = pMsg->rpcMsg.ahandle,
.refId = pMsg->rpcMsg.refId,
.code = code, .code = code,
.pCont = pMsg->pRsp, .pCont = pMsg->pRsp,
.contLen = pMsg->rspLen}; .contLen = pMsg->rspLen};

View File

@ -19,6 +19,7 @@
static inline void qmSendRsp(SNodeMsg *pMsg, int32_t code) { static inline void qmSendRsp(SNodeMsg *pMsg, int32_t code) {
SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle, SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle,
.ahandle = pMsg->rpcMsg.ahandle, .ahandle = pMsg->rpcMsg.ahandle,
.refId = pMsg->rpcMsg.refId,
.code = code, .code = code,
.pCont = pMsg->pRsp, .pCont = pMsg->pRsp,
.contLen = pMsg->rspLen}; .contLen = pMsg->rspLen};

View File

@ -19,6 +19,7 @@
static inline void smSendRsp(SNodeMsg *pMsg, int32_t code) { static inline void smSendRsp(SNodeMsg *pMsg, int32_t code) {
SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle, SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle,
.ahandle = pMsg->rpcMsg.ahandle, .ahandle = pMsg->rpcMsg.ahandle,
.refId = pMsg->rpcMsg.refId,
.code = code, .code = code,
.pCont = pMsg->pRsp, .pCont = pMsg->pRsp,
.contLen = pMsg->rspLen}; .contLen = pMsg->rspLen};
@ -149,7 +150,7 @@ static FORCE_INLINE int32_t smGetSWTypeFromMsg(SRpcMsg *pMsg) {
} }
int32_t smProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { int32_t smProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SSnodeMgmt *pMgmt = pWrapper->pMgmt; SSnodeMgmt * pMgmt = pWrapper->pMgmt;
SMultiWorker *pWorker = taosArrayGetP(pMgmt->uniqueWorkers, 0); SMultiWorker *pWorker = taosArrayGetP(pMgmt->uniqueWorkers, 0);
if (pWorker == NULL) { if (pWorker == NULL) {
terrno = TSDB_CODE_INVALID_MSG; terrno = TSDB_CODE_INVALID_MSG;
@ -162,7 +163,7 @@ int32_t smProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
} }
int32_t smProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { int32_t smProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SSnodeMgmt *pMgmt = pWrapper->pMgmt; SSnodeMgmt * pMgmt = pWrapper->pMgmt;
SSingleWorker *pWorker = &pMgmt->monitorWorker; SSingleWorker *pWorker = &pMgmt->monitorWorker;
dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name); dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);
@ -171,7 +172,7 @@ int32_t smProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
} }
int32_t smProcessUniqueMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { int32_t smProcessUniqueMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SSnodeMgmt *pMgmt = pWrapper->pMgmt; SSnodeMgmt * pMgmt = pWrapper->pMgmt;
int32_t index = smGetSWIdFromMsg(&pMsg->rpcMsg); int32_t index = smGetSWIdFromMsg(&pMsg->rpcMsg);
SMultiWorker *pWorker = taosArrayGetP(pMgmt->uniqueWorkers, index); SMultiWorker *pWorker = taosArrayGetP(pMgmt->uniqueWorkers, index);
if (pWorker == NULL) { if (pWorker == NULL) {
@ -185,7 +186,7 @@ int32_t smProcessUniqueMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
} }
int32_t smProcessSharedMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { int32_t smProcessSharedMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SSnodeMgmt *pMgmt = pWrapper->pMgmt; SSnodeMgmt * pMgmt = pWrapper->pMgmt;
SSingleWorker *pWorker = &pMgmt->sharedWorker; SSingleWorker *pWorker = &pMgmt->sharedWorker;
dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name); dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);

View File

@ -23,6 +23,7 @@
static inline void vmSendRsp(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int32_t code) { static inline void vmSendRsp(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int32_t code) {
SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle, SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle,
.ahandle = pMsg->rpcMsg.ahandle, .ahandle = pMsg->rpcMsg.ahandle,
.refId = pMsg->rpcMsg.refId,
.code = code, .code = code,
.pCont = pMsg->pRsp, .pCont = pMsg->pRsp,
.contLen = pMsg->rspLen}; .contLen = pMsg->rspLen};
@ -126,6 +127,7 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
rsp.code = 0; rsp.code = 0;
rsp.handle = pRpc->handle; rsp.handle = pRpc->handle;
rsp.ahandle = pRpc->ahandle; rsp.ahandle = pRpc->ahandle;
rsp.refId = pRpc->refId;
int32_t code = vnodeProcessWriteReq(pVnode->pImpl, pRpc, version++, &rsp); int32_t code = vnodeProcessWriteReq(pVnode->pImpl, pRpc, version++, &rsp);
tmsgSendRsp(&rsp); tmsgSendRsp(&rsp);
@ -134,13 +136,14 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
// sync integration response // sync integration response
for (int i = 0; i < taosArrayGetSize(pArray); i++) { for (int i = 0; i < taosArrayGetSize(pArray); i++) {
SNodeMsg *pMsg; SNodeMsg *pMsg;
SRpcMsg *pRpc; SRpcMsg * pRpc;
pMsg = *(SNodeMsg **)taosArrayGet(pArray, i); pMsg = *(SNodeMsg **)taosArrayGet(pArray, i);
pRpc = &pMsg->rpcMsg; pRpc = &pMsg->rpcMsg;
rsp.ahandle = pRpc->ahandle; rsp.ahandle = pRpc->ahandle;
rsp.handle = pRpc->handle; rsp.handle = pRpc->handle;
rsp.refId = pRpc->refId;
rsp.pCont = NULL; rsp.pCont = NULL;
rsp.contLen = 0; rsp.contLen = 0;
@ -172,11 +175,9 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) { static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
SVnodeObj *pVnode = pInfo->ahandle; SVnodeObj *pVnode = pInfo->ahandle;
SNodeMsg *pMsg = NULL; SNodeMsg * pMsg = NULL;
SRpcMsg rsp; SRpcMsg rsp;
// static int64_t version = 0;
for (int32_t i = 0; i < numOfMsgs; ++i) { for (int32_t i = 0; i < numOfMsgs; ++i) {
#if 1 #if 1
// sync integration // sync integration
@ -208,6 +209,7 @@ static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
if (pMsg->rpcMsg.handle != NULL && pMsg->rpcMsg.ahandle != NULL) { if (pMsg->rpcMsg.handle != NULL && pMsg->rpcMsg.ahandle != NULL) {
rsp.ahandle = pMsg->rpcMsg.ahandle; rsp.ahandle = pMsg->rpcMsg.ahandle;
rsp.handle = pMsg->rpcMsg.handle; rsp.handle = pMsg->rpcMsg.handle;
rsp.refId = pMsg->rpcMsg.refId;
tmsgSendRsp(&rsp); tmsgSendRsp(&rsp);
} }
#endif #endif
@ -216,7 +218,7 @@ static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) { static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
SVnodeObj *pVnode = pInfo->ahandle; SVnodeObj *pVnode = pInfo->ahandle;
SNodeMsg *pMsg = NULL; SNodeMsg * pMsg = NULL;
for (int32_t i = 0; i < numOfMsgs; ++i) { for (int32_t i = 0; i < numOfMsgs; ++i) {
taosGetQitem(qall, (void **)&pMsg); taosGetQitem(qall, (void **)&pMsg);
@ -229,7 +231,7 @@ static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOf
static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) { static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
SVnodeObj *pVnode = pInfo->ahandle; SVnodeObj *pVnode = pInfo->ahandle;
SNodeMsg *pMsg = NULL; SNodeMsg * pMsg = NULL;
for (int32_t i = 0; i < numOfMsgs; ++i) { for (int32_t i = 0; i < numOfMsgs; ++i) {
taosGetQitem(qall, (void **)&pMsg); taosGetQitem(qall, (void **)&pMsg);
@ -246,7 +248,7 @@ static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
} }
static int32_t vmPutNodeMsgToQueue(SVnodesMgmt *pMgmt, SNodeMsg *pMsg, EQueueType qtype) { static int32_t vmPutNodeMsgToQueue(SVnodesMgmt *pMgmt, SNodeMsg *pMsg, EQueueType qtype) {
SRpcMsg *pRpc = &pMsg->rpcMsg; SRpcMsg * pRpc = &pMsg->rpcMsg;
SMsgHead *pHead = pRpc->pCont; SMsgHead *pHead = pRpc->pCont;
pHead->contLen = ntohl(pHead->contLen); pHead->contLen = ntohl(pHead->contLen);
pHead->vgId = ntohl(pHead->vgId); pHead->vgId = ntohl(pHead->vgId);
@ -315,7 +317,7 @@ int32_t vmProcessMergeMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
} }
int32_t vmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { int32_t vmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SVnodesMgmt *pMgmt = pWrapper->pMgmt; SVnodesMgmt * pMgmt = pWrapper->pMgmt;
SSingleWorker *pWorker = &pMgmt->mgmtWorker; SSingleWorker *pWorker = &pMgmt->mgmtWorker;
dTrace("msg:%p, will be written to vnode-mgmt queue, worker:%s", pMsg, pWorker->name); dTrace("msg:%p, will be written to vnode-mgmt queue, worker:%s", pMsg, pWorker->name);
taosWriteQitem(pWorker->queue, pMsg); taosWriteQitem(pWorker->queue, pMsg);
@ -323,7 +325,7 @@ int32_t vmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
} }
int32_t vmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { int32_t vmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
SVnodesMgmt *pMgmt = pWrapper->pMgmt; SVnodesMgmt * pMgmt = pWrapper->pMgmt;
SSingleWorker *pWorker = &pMgmt->monitorWorker; SSingleWorker *pWorker = &pMgmt->monitorWorker;
dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name); dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);
@ -333,7 +335,7 @@ int32_t vmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
static int32_t vmPutRpcMsgToQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, EQueueType qtype) { static int32_t vmPutRpcMsgToQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, EQueueType qtype) {
SVnodesMgmt *pMgmt = pWrapper->pMgmt; SVnodesMgmt *pMgmt = pWrapper->pMgmt;
SMsgHead *pHead = pRpc->pCont; SMsgHead * pHead = pRpc->pCont;
SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId); SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
if (pVnode == NULL) return -1; if (pVnode == NULL) return -1;
@ -346,6 +348,7 @@ static int32_t vmPutRpcMsgToQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, EQueueT
} else { } else {
dTrace("msg:%p, is created, type:%s", pMsg, TMSG_INFO(pRpc->msgType)); dTrace("msg:%p, is created, type:%s", pMsg, TMSG_INFO(pRpc->msgType));
pMsg->rpcMsg = *pRpc; pMsg->rpcMsg = *pRpc;
// if (pMsg->rpcMsg.handle != NULL) assert(pMsg->rpcMsg.refId != 0);
switch (qtype) { switch (qtype) {
case QUERY_QUEUE: case QUERY_QUEUE:
dTrace("msg:%p, will be put into vnode-query queue", pMsg); dTrace("msg:%p, will be put into vnode-query queue", pMsg);

View File

@ -135,6 +135,7 @@ typedef struct {
int32_t failedTimes; int32_t failedTimes;
void* rpcHandle; void* rpcHandle;
void* rpcAHandle; void* rpcAHandle;
int64_t rpcRefId;
void* rpcRsp; void* rpcRsp;
int32_t rpcRspLen; int32_t rpcRspLen;
SArray* redoLogs; SArray* redoLogs;

View File

@ -193,9 +193,9 @@ TRANS_ENCODE_OVER:
static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) { static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
SSdbRow *pRow = NULL; SSdbRow * pRow = NULL;
STrans *pTrans = NULL; STrans * pTrans = NULL;
char *pData = NULL; char * pData = NULL;
int32_t dataLen = 0; int32_t dataLen = 0;
int8_t sver = 0; int8_t sver = 0;
int32_t redoLogNum = 0; int32_t redoLogNum = 0;
@ -456,7 +456,7 @@ static int32_t mndTransActionUpdate(SSdb *pSdb, STrans *pOld, STrans *pNew) {
} }
static STrans *mndAcquireTrans(SMnode *pMnode, int32_t transId) { static STrans *mndAcquireTrans(SMnode *pMnode, int32_t transId) {
SSdb *pSdb = pMnode->pSdb; SSdb * pSdb = pMnode->pSdb;
STrans *pTrans = sdbAcquire(pSdb, SDB_TRANS, &transId); STrans *pTrans = sdbAcquire(pSdb, SDB_TRANS, &transId);
if (pTrans == NULL) { if (pTrans == NULL) {
terrno = TSDB_CODE_MND_TRANS_NOT_EXIST; terrno = TSDB_CODE_MND_TRANS_NOT_EXIST;
@ -484,6 +484,7 @@ STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnType type, const S
pTrans->createdTime = taosGetTimestampMs(); pTrans->createdTime = taosGetTimestampMs();
pTrans->rpcHandle = pReq->handle; pTrans->rpcHandle = pReq->handle;
pTrans->rpcAHandle = pReq->ahandle; pTrans->rpcAHandle = pReq->ahandle;
pTrans->rpcRefId = pReq->refId;
pTrans->redoLogs = taosArrayInit(MND_TRANS_ARRAY_SIZE, sizeof(void *)); pTrans->redoLogs = taosArrayInit(MND_TRANS_ARRAY_SIZE, sizeof(void *));
pTrans->undoLogs = taosArrayInit(MND_TRANS_ARRAY_SIZE, sizeof(void *)); pTrans->undoLogs = taosArrayInit(MND_TRANS_ARRAY_SIZE, sizeof(void *));
pTrans->commitLogs = taosArrayInit(MND_TRANS_ARRAY_SIZE, sizeof(void *)); pTrans->commitLogs = taosArrayInit(MND_TRANS_ARRAY_SIZE, sizeof(void *));
@ -625,7 +626,7 @@ static int32_t mndCheckTransCanBeStartedInParallel(SMnode *pMnode, STrans *pNewT
if (mndIsBasicTrans(pNewTrans)) return 0; if (mndIsBasicTrans(pNewTrans)) return 0;
STrans *pTrans = NULL; STrans *pTrans = NULL;
void *pIter = NULL; void * pIter = NULL;
int32_t code = 0; int32_t code = 0;
while (1) { while (1) {
@ -703,6 +704,7 @@ int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans) {
pNew->rpcHandle = pTrans->rpcHandle; pNew->rpcHandle = pTrans->rpcHandle;
pNew->rpcAHandle = pTrans->rpcAHandle; pNew->rpcAHandle = pTrans->rpcAHandle;
pNew->rpcRefId = pTrans->rpcRefId;
pNew->rpcRsp = pTrans->rpcRsp; pNew->rpcRsp = pTrans->rpcRsp;
pNew->rpcRspLen = pTrans->rpcRspLen; pNew->rpcRspLen = pTrans->rpcRspLen;
pTrans->rpcRsp = NULL; pTrans->rpcRsp = NULL;
@ -767,6 +769,7 @@ static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
SRpcMsg rspMsg = {.handle = pTrans->rpcHandle, SRpcMsg rspMsg = {.handle = pTrans->rpcHandle,
.code = pTrans->code, .code = pTrans->code,
.ahandle = pTrans->rpcAHandle, .ahandle = pTrans->rpcAHandle,
.refId = pTrans->rpcRefId,
.pCont = rpcCont, .pCont = rpcCont,
.contLen = pTrans->rpcRspLen}; .contLen = pTrans->rpcRspLen};
tmsgSendRsp(&rspMsg); tmsgSendRsp(&rspMsg);
@ -827,7 +830,7 @@ HANDLE_ACTION_RSP_OVER:
} }
static int32_t mndTransExecuteLogs(SMnode *pMnode, SArray *pArray) { static int32_t mndTransExecuteLogs(SMnode *pMnode, SArray *pArray) {
SSdb *pSdb = pMnode->pSdb; SSdb * pSdb = pMnode->pSdb;
int32_t arraySize = taosArrayGetSize(pArray); int32_t arraySize = taosArrayGetSize(pArray);
if (arraySize == 0) return 0; if (arraySize == 0) return 0;
@ -1202,11 +1205,11 @@ static int32_t mndKillTrans(SMnode *pMnode, STrans *pTrans) {
} }
static int32_t mndProcessKillTransReq(SNodeMsg *pReq) { static int32_t mndProcessKillTransReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode; SMnode * pMnode = pReq->pNode;
SKillTransReq killReq = {0}; SKillTransReq killReq = {0};
int32_t code = -1; int32_t code = -1;
SUserObj *pUser = NULL; SUserObj * pUser = NULL;
STrans *pTrans = NULL; STrans * pTrans = NULL;
if (tDeserializeSKillTransReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &killReq) != 0) { if (tDeserializeSKillTransReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &killReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG; terrno = TSDB_CODE_INVALID_MSG;
@ -1246,7 +1249,7 @@ KILL_OVER:
void mndTransPullup(SMnode *pMnode) { void mndTransPullup(SMnode *pMnode) {
STrans *pTrans = NULL; STrans *pTrans = NULL;
void *pIter = NULL; void * pIter = NULL;
while (1) { while (1) {
pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans); pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
@ -1261,11 +1264,11 @@ void mndTransPullup(SMnode *pMnode) {
static int32_t mndRetrieveTrans(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) { static int32_t mndRetrieveTrans(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
SMnode *pMnode = pReq->pNode; SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb; SSdb * pSdb = pMnode->pSdb;
int32_t numOfRows = 0; int32_t numOfRows = 0;
STrans *pTrans = NULL; STrans *pTrans = NULL;
int32_t cols = 0; int32_t cols = 0;
char *pWrite; char * pWrite;
while (numOfRows < rows) { while (numOfRows < rows) {
pShow->pIter = sdbFetch(pSdb, SDB_TRANS, pShow->pIter, (void **)&pTrans); pShow->pIter = sdbFetch(pSdb, SDB_TRANS, pShow->pIter, (void **)&pTrans);

View File

@ -22,21 +22,21 @@ int vnodeQueryOpen(SVnode *pVnode) {
void vnodeQueryClose(SVnode *pVnode) { qWorkerDestroy((void **)&pVnode->pQuery); } void vnodeQueryClose(SVnode *pVnode) { qWorkerDestroy((void **)&pVnode->pQuery); }
int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg) { int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg) {
STbCfg *pTbCfg = NULL; STbCfg * pTbCfg = NULL;
STbCfg *pStbCfg = NULL; STbCfg * pStbCfg = NULL;
tb_uid_t uid; tb_uid_t uid;
int32_t nCols; int32_t nCols;
int32_t nTagCols; int32_t nTagCols;
SSchemaWrapper *pSW = NULL; SSchemaWrapper *pSW = NULL;
STableMetaRsp *pTbMetaMsg = NULL; STableMetaRsp * pTbMetaMsg = NULL;
STableMetaRsp metaRsp = {0}; STableMetaRsp metaRsp = {0};
SSchema *pTagSchema; SSchema * pTagSchema;
SRpcMsg rpcMsg; SRpcMsg rpcMsg;
int msgLen = 0; int msgLen = 0;
int32_t code = 0; int32_t code = 0;
char tableFName[TSDB_TABLE_FNAME_LEN]; char tableFName[TSDB_TABLE_FNAME_LEN];
int32_t rspLen = 0; int32_t rspLen = 0;
void *pRsp = NULL; void * pRsp = NULL;
STableInfoReq infoReq = {0}; STableInfoReq infoReq = {0};
if (tDeserializeSTableInfoReq(pMsg->pCont, pMsg->contLen, &infoReq) != 0) { if (tDeserializeSTableInfoReq(pMsg->pCont, pMsg->contLen, &infoReq) != 0) {
@ -142,6 +142,7 @@ _exit:
rpcMsg.handle = pMsg->handle; rpcMsg.handle = pMsg->handle;
rpcMsg.ahandle = pMsg->ahandle; rpcMsg.ahandle = pMsg->ahandle;
rpcMsg.refId = pMsg->refId;
rpcMsg.pCont = pRsp; rpcMsg.pCont = pRsp;
rpcMsg.contLen = rspLen; rpcMsg.contLen = rspLen;
rpcMsg.code = code; rpcMsg.code = code;

View File

@ -113,7 +113,7 @@ void vnodeSyncCommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cb
pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state), beginIndex); pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state), beginIndex);
syncRpcMsgLog2(logBuf, (SRpcMsg *)pMsg); syncRpcMsgLog2(logBuf, (SRpcMsg *)pMsg);
SVnode *pVnode = (SVnode *)(pFsm->data); SVnode * pVnode = (SVnode *)(pFsm->data);
SyncApplyMsg *pSyncApplyMsg = syncApplyMsgBuild2(pMsg, pVnode->config.vgId, &cbMeta); SyncApplyMsg *pSyncApplyMsg = syncApplyMsgBuild2(pMsg, pVnode->config.vgId, &cbMeta);
SRpcMsg applyMsg; SRpcMsg applyMsg;
syncApplyMsg2RpcMsg(pSyncApplyMsg, &applyMsg); syncApplyMsg2RpcMsg(pSyncApplyMsg, &applyMsg);
@ -133,6 +133,7 @@ void vnodeSyncCommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cb
if (ret == 1 && cbMeta.state == TAOS_SYNC_STATE_LEADER) { if (ret == 1 && cbMeta.state == TAOS_SYNC_STATE_LEADER) {
applyMsg.handle = saveRpcMsg.handle; applyMsg.handle = saveRpcMsg.handle;
applyMsg.ahandle = saveRpcMsg.ahandle; applyMsg.ahandle = saveRpcMsg.ahandle;
applyMsg.refId = saveRpcMsg.refId;
} else { } else {
applyMsg.handle = NULL; applyMsg.handle = NULL;
applyMsg.ahandle = NULL; applyMsg.ahandle = NULL;

File diff suppressed because it is too large Load Diff

View File

@ -60,7 +60,6 @@ enum {
QW_WRITE, QW_WRITE,
}; };
enum { enum {
QW_NOT_EXIST_RET_ERR = 1, QW_NOT_EXIST_RET_ERR = 1,
QW_NOT_EXIST_ADD, QW_NOT_EXIST_ADD,
@ -73,14 +72,15 @@ typedef struct SQWDebug {
} SQWDebug; } SQWDebug;
typedef struct SQWConnInfo { typedef struct SQWConnInfo {
void *handle; void * handle;
void *ahandle; void * ahandle;
int64_t refId;
} SQWConnInfo; } SQWConnInfo;
typedef struct SQWMsg { typedef struct SQWMsg {
void *node; void * node;
int32_t code; int32_t code;
char *msg; char * msg;
int32_t msgLen; int32_t msgLen;
SQWConnInfo connInfo; SQWConnInfo connInfo;
} SQWMsg; } SQWMsg;
@ -100,7 +100,6 @@ typedef struct SQWPhaseOutput {
#endif #endif
} SQWPhaseOutput; } SQWPhaseOutput;
typedef struct SQWTaskStatus { typedef struct SQWTaskStatus {
int64_t refId; // job's refId int64_t refId; // job's refId
int32_t code; int32_t code;
@ -124,8 +123,8 @@ typedef struct SQWTaskCtx {
int8_t events[QW_EVENT_MAX]; int8_t events[QW_EVENT_MAX];
qTaskInfo_t taskHandle; void *taskHandle;
DataSinkHandle sinkHandle; void *sinkHandle;
} SQWTaskCtx; } SQWTaskCtx;
typedef struct SQWSchStatus { typedef struct SQWSchStatus {
@ -134,7 +133,7 @@ typedef struct SQWSchStatus {
SQWConnInfo hbConnInfo; SQWConnInfo hbConnInfo;
SQueryNodeEpId hbEpId; SQueryNodeEpId hbEpId;
SRWLatch tasksLock; SRWLatch tasksLock;
SHashObj *tasksHash; // key:queryId+taskId, value: SQWTaskStatus SHashObj * tasksHash; // key:queryId+taskId, value: SQWTaskStatus
} SQWSchStatus; } SQWSchStatus;
// Qnode/Vnode level task management // Qnode/Vnode level task management
@ -142,7 +141,7 @@ typedef struct SQWorkerMgmt {
SQWorkerCfg cfg; SQWorkerCfg cfg;
int8_t nodeType; int8_t nodeType;
int32_t nodeId; int32_t nodeId;
void *timer; void * timer;
tmr_h hbTimer; tmr_h hbTimer;
SRWLatch schLock; SRWLatch schLock;
// SRWLatch ctxLock; // SRWLatch ctxLock;
@ -171,41 +170,86 @@ typedef struct SQWorkerMgmt {
#define QW_TASK_NOT_EXIST(code) (TSDB_CODE_QRY_SCH_NOT_EXIST == (code) || TSDB_CODE_QRY_TASK_NOT_EXIST == (code)) #define QW_TASK_NOT_EXIST(code) (TSDB_CODE_QRY_SCH_NOT_EXIST == (code) || TSDB_CODE_QRY_TASK_NOT_EXIST == (code))
#define QW_TASK_ALREADY_EXIST(code) (TSDB_CODE_QRY_TASK_ALREADY_EXIST == (code)) #define QW_TASK_ALREADY_EXIST(code) (TSDB_CODE_QRY_TASK_ALREADY_EXIST == (code))
#define QW_TASK_READY(status) (status == JOB_TASK_STATUS_SUCCEED || status == JOB_TASK_STATUS_FAILED || status == JOB_TASK_STATUS_CANCELLED || status == JOB_TASK_STATUS_PARTIAL_SUCCEED) #define QW_TASK_READY(status) \
#define QW_SET_QTID(id, qId, tId) do { *(uint64_t *)(id) = (qId); *(uint64_t *)((char *)(id) + sizeof(qId)) = (tId); } while (0) (status == JOB_TASK_STATUS_SUCCEED || status == JOB_TASK_STATUS_FAILED || status == JOB_TASK_STATUS_CANCELLED || \
#define QW_GET_QTID(id, qId, tId) do { (qId) = *(uint64_t *)(id); (tId) = *(uint64_t *)((char *)(id) + sizeof(qId)); } while (0) status == JOB_TASK_STATUS_PARTIAL_SUCCEED)
#define QW_SET_QTID(id, qId, tId) \
do { \
*(uint64_t *)(id) = (qId); \
*(uint64_t *)((char *)(id) + sizeof(qId)) = (tId); \
} while (0)
#define QW_GET_QTID(id, qId, tId) \
do { \
(qId) = *(uint64_t *)(id); \
(tId) = *(uint64_t *)((char *)(id) + sizeof(qId)); \
} while (0)
#define QW_ERR_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; return _code; } } while (0) #define QW_ERR_RET(c) \
#define QW_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; } return _code; } while (0) do { \
#define QW_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { terrno = code; goto _return; } } while (0) int32_t _code = c; \
if (_code != TSDB_CODE_SUCCESS) { \
terrno = _code; \
return _code; \
} \
} while (0)
#define QW_RET(c) \
do { \
int32_t _code = c; \
if (_code != TSDB_CODE_SUCCESS) { \
terrno = _code; \
} \
return _code; \
} while (0)
#define QW_ERR_JRET(c) \
do { \
code = c; \
if (code != TSDB_CODE_SUCCESS) { \
terrno = code; \
goto _return; \
} \
} while (0)
#define QW_ELOG(param, ...) qError("QW:%p " param, mgmt, __VA_ARGS__) #define QW_ELOG(param, ...) qError("QW:%p " param, mgmt, __VA_ARGS__)
#define QW_DLOG(param, ...) qDebug("QW:%p " param, mgmt, __VA_ARGS__) #define QW_DLOG(param, ...) qDebug("QW:%p " param, mgmt, __VA_ARGS__)
#define QW_DUMP(param, ...) do { if (gQWDebug.dumpEnable) { qDebug("QW:%p " param, mgmt, __VA_ARGS__); } } while (0) #define QW_DUMP(param, ...) \
do { \
if (gQWDebug.dumpEnable) { \
qDebug("QW:%p " param, mgmt, __VA_ARGS__); \
} \
} while (0)
#define QW_SCH_ELOG(param, ...) qError("QW:%p SID:%" PRIx64 " " param, mgmt, sId, __VA_ARGS__)
#define QW_SCH_DLOG(param, ...) qDebug("QW:%p SID:%" PRIx64 " " param, mgmt, sId, __VA_ARGS__)
#define QW_SCH_ELOG(param, ...) qError("QW:%p SID:%"PRIx64" " param, mgmt, sId, __VA_ARGS__) #define QW_TASK_ELOG(param, ...) qError("QW:%p QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_SCH_DLOG(param, ...) qDebug("QW:%p SID:%"PRIx64" " param, mgmt, sId, __VA_ARGS__) #define QW_TASK_WLOG(param, ...) qWarn("QW:%p QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_DLOG(param, ...) qDebug("QW:%p QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_DLOGL(param, ...) \
qDebugL("QW:%p QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_ELOG(param, ...) qError("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__) #define QW_TASK_ELOG_E(param) qError("QW:%p QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, qId, tId)
#define QW_TASK_WLOG(param, ...) qWarn("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__) #define QW_TASK_WLOG_E(param) qWarn("QW:%p QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, qId, tId)
#define QW_TASK_DLOG(param, ...) qDebug("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__) #define QW_TASK_DLOG_E(param) qDebug("QW:%p QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, qId, tId)
#define QW_TASK_DLOGL(param, ...) qDebugL("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_ELOG_E(param) qError("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId) #define QW_SCH_TASK_ELOG(param, ...) \
#define QW_TASK_WLOG_E(param) qWarn("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId) qError("QW:%p SID:0x%" PRIx64 ",QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, sId, qId, tId, __VA_ARGS__)
#define QW_TASK_DLOG_E(param) qDebug("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId) #define QW_SCH_TASK_WLOG(param, ...) \
qWarn("QW:%p SID:0x%" PRIx64 ",QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, sId, qId, tId, __VA_ARGS__)
#define QW_SCH_TASK_DLOG(param, ...) \
qDebug("QW:%p SID:0x%" PRIx64 ",QID:0x%" PRIx64 ",TID:0x%" PRIx64 " " param, mgmt, sId, qId, tId, __VA_ARGS__)
#define QW_SCH_TASK_ELOG(param, ...) qError("QW:%p SID:0x%"PRIx64",QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__) #define QW_LOCK_DEBUG(...) \
#define QW_SCH_TASK_WLOG(param, ...) qWarn("QW:%p SID:0x%"PRIx64",QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__) do { \
#define QW_SCH_TASK_DLOG(param, ...) qDebug("QW:%p SID:0x%"PRIx64",QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__) if (gQWDebug.lockEnable) { \
qDebug(__VA_ARGS__); \
#define QW_LOCK_DEBUG(...) do { if (gQWDebug.lockEnable) { qDebug(__VA_ARGS__); } } while (0) } \
} while (0)
#define TD_RWLATCH_WRITE_FLAG_COPY 0x40000000 #define TD_RWLATCH_WRITE_FLAG_COPY 0x40000000
#define QW_LOCK(type, _lock) do { \ #define QW_LOCK(type, _lock) \
do { \
if (QW_READ == (type)) { \ if (QW_READ == (type)) { \
assert(atomic_load_32((_lock)) >= 0); \ assert(atomic_load_32((_lock)) >= 0); \
QW_LOCK_DEBUG("QW RLOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ QW_LOCK_DEBUG("QW RLOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
@ -219,9 +263,10 @@ typedef struct SQWorkerMgmt {
QW_LOCK_DEBUG("QW WLOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ QW_LOCK_DEBUG("QW WLOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
assert(atomic_load_32((_lock)) == TD_RWLATCH_WRITE_FLAG_COPY); \ assert(atomic_load_32((_lock)) == TD_RWLATCH_WRITE_FLAG_COPY); \
} \ } \
} while (0) } while (0)
#define QW_UNLOCK(type, _lock) do { \ #define QW_UNLOCK(type, _lock) \
do { \
if (QW_READ == (type)) { \ if (QW_READ == (type)) { \
assert(atomic_load_32((_lock)) > 0); \ assert(atomic_load_32((_lock)) > 0); \
QW_LOCK_DEBUG("QW RULOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ QW_LOCK_DEBUG("QW RULOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
@ -235,7 +280,7 @@ typedef struct SQWorkerMgmt {
QW_LOCK_DEBUG("QW WULOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ QW_LOCK_DEBUG("QW WULOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
assert(atomic_load_32((_lock)) >= 0); \ assert(atomic_load_32((_lock)) >= 0); \
} \ } \
} while (0) } while (0)
#ifdef __cplusplus #ifdef __cplusplus
} }

View File

@ -1,13 +1,13 @@
#include "qworker.h" #include "qworker.h"
#include "tcommon.h" #include "dataSinkMgt.h"
#include "executor.h" #include "executor.h"
#include "planner.h" #include "planner.h"
#include "query.h" #include "query.h"
#include "qworkerInt.h" #include "qworkerInt.h"
#include "qworkerMsg.h" #include "qworkerMsg.h"
#include "tcommon.h"
#include "tmsg.h" #include "tmsg.h"
#include "tname.h" #include "tname.h"
#include "dataSinkMgt.h"
SQWDebug gQWDebug = {.statusEnable = true, .dumpEnable = true}; SQWDebug gQWDebug = {.statusEnable = true, .dumpEnable = true};
@ -29,9 +29,8 @@ int32_t qwDbgValidateStatus(QW_FPARAMS_DEF, int8_t oriStatus, int8_t newStatus,
switch (oriStatus) { switch (oriStatus) {
case JOB_TASK_STATUS_NULL: case JOB_TASK_STATUS_NULL:
if (newStatus != JOB_TASK_STATUS_EXECUTING if (newStatus != JOB_TASK_STATUS_EXECUTING && newStatus != JOB_TASK_STATUS_FAILED &&
&& newStatus != JOB_TASK_STATUS_FAILED newStatus != JOB_TASK_STATUS_NOT_START) {
&& newStatus != JOB_TASK_STATUS_NOT_START) {
QW_ERR_JRET(TSDB_CODE_QRY_APP_ERROR); QW_ERR_JRET(TSDB_CODE_QRY_APP_ERROR);
} }
@ -43,30 +42,24 @@ int32_t qwDbgValidateStatus(QW_FPARAMS_DEF, int8_t oriStatus, int8_t newStatus,
break; break;
case JOB_TASK_STATUS_EXECUTING: case JOB_TASK_STATUS_EXECUTING:
if (newStatus != JOB_TASK_STATUS_PARTIAL_SUCCEED if (newStatus != JOB_TASK_STATUS_PARTIAL_SUCCEED && newStatus != JOB_TASK_STATUS_SUCCEED &&
&& newStatus != JOB_TASK_STATUS_SUCCEED newStatus != JOB_TASK_STATUS_FAILED && newStatus != JOB_TASK_STATUS_CANCELLING &&
&& newStatus != JOB_TASK_STATUS_FAILED newStatus != JOB_TASK_STATUS_CANCELLED && newStatus != JOB_TASK_STATUS_DROPPING) {
&& newStatus != JOB_TASK_STATUS_CANCELLING
&& newStatus != JOB_TASK_STATUS_CANCELLED
&& newStatus != JOB_TASK_STATUS_DROPPING) {
QW_ERR_JRET(TSDB_CODE_QRY_APP_ERROR); QW_ERR_JRET(TSDB_CODE_QRY_APP_ERROR);
} }
break; break;
case JOB_TASK_STATUS_PARTIAL_SUCCEED: case JOB_TASK_STATUS_PARTIAL_SUCCEED:
if (newStatus != JOB_TASK_STATUS_EXECUTING if (newStatus != JOB_TASK_STATUS_EXECUTING && newStatus != JOB_TASK_STATUS_SUCCEED &&
&& newStatus != JOB_TASK_STATUS_SUCCEED newStatus != JOB_TASK_STATUS_CANCELLED && newStatus != JOB_TASK_STATUS_FAILED &&
&& newStatus != JOB_TASK_STATUS_CANCELLED newStatus != JOB_TASK_STATUS_DROPPING) {
&& newStatus != JOB_TASK_STATUS_FAILED
&& newStatus != JOB_TASK_STATUS_DROPPING) {
QW_ERR_JRET(TSDB_CODE_QRY_APP_ERROR); QW_ERR_JRET(TSDB_CODE_QRY_APP_ERROR);
} }
break; break;
case JOB_TASK_STATUS_SUCCEED: case JOB_TASK_STATUS_SUCCEED:
if (newStatus != JOB_TASK_STATUS_CANCELLED if (newStatus != JOB_TASK_STATUS_CANCELLED && newStatus != JOB_TASK_STATUS_DROPPING &&
&& newStatus != JOB_TASK_STATUS_DROPPING newStatus != JOB_TASK_STATUS_FAILED) {
&& newStatus != JOB_TASK_STATUS_FAILED) {
QW_ERR_JRET(TSDB_CODE_QRY_APP_ERROR); QW_ERR_JRET(TSDB_CODE_QRY_APP_ERROR);
} }
@ -103,9 +96,7 @@ _return:
QW_RET(code); QW_RET(code);
} }
void qwDbgDumpSchInfo(SQWSchStatus *sch, int32_t i) { void qwDbgDumpSchInfo(SQWSchStatus *sch, int32_t i) {}
}
void qwDbgDumpMgmtInfo(SQWorkerMgmt *mgmt) { void qwDbgDumpMgmtInfo(SQWorkerMgmt *mgmt) {
if (!gQWDebug.dumpEnable) { if (!gQWDebug.dumpEnable) {
@ -116,7 +107,7 @@ void qwDbgDumpMgmtInfo(SQWorkerMgmt *mgmt) {
QW_DUMP("total remain schduler num:%d", taosHashGetSize(mgmt->schHash)); QW_DUMP("total remain schduler num:%d", taosHashGetSize(mgmt->schHash));
void *key = NULL; void * key = NULL;
size_t keyLen = 0; size_t keyLen = 0;
int32_t i = 0; int32_t i = 0;
SQWSchStatus *sch = NULL; SQWSchStatus *sch = NULL;
@ -195,10 +186,10 @@ int32_t qwSetTaskStatus(QW_FPARAMS_DEF, SQWTaskStatus *task, int8_t status) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t qwAddSchedulerImpl(SQWorkerMgmt *mgmt, uint64_t sId, int32_t rwType) { int32_t qwAddSchedulerImpl(SQWorkerMgmt *mgmt, uint64_t sId, int32_t rwType) {
SQWSchStatus newSch = {0}; SQWSchStatus newSch = {0};
newSch.tasksHash = taosHashInit(mgmt->cfg.maxSchTaskNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK); newSch.tasksHash =
taosHashInit(mgmt->cfg.maxSchTaskNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
if (NULL == newSch.tasksHash) { if (NULL == newSch.tasksHash) {
QW_SCH_ELOG("taosHashInit %d failed", mgmt->cfg.maxSchTaskNum); QW_SCH_ELOG("taosHashInit %d failed", mgmt->cfg.maxSchTaskNum);
QW_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY); QW_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
@ -257,10 +248,7 @@ int32_t qwAcquireScheduler(SQWorkerMgmt *mgmt, uint64_t sId, int32_t rwType, SQW
return qwAcquireSchedulerImpl(mgmt, sId, rwType, sch, QW_NOT_EXIST_RET_ERR); return qwAcquireSchedulerImpl(mgmt, sId, rwType, sch, QW_NOT_EXIST_RET_ERR);
} }
void qwReleaseScheduler(int32_t rwType, SQWorkerMgmt *mgmt) { void qwReleaseScheduler(int32_t rwType, SQWorkerMgmt *mgmt) { QW_UNLOCK(rwType, &mgmt->schLock); }
QW_UNLOCK(rwType, &mgmt->schLock);
}
int32_t qwAcquireTaskStatus(QW_FPARAMS_DEF, int32_t rwType, SQWSchStatus *sch, SQWTaskStatus **task) { int32_t qwAcquireTaskStatus(QW_FPARAMS_DEF, int32_t rwType, SQWSchStatus *sch, SQWTaskStatus **task) {
char id[sizeof(qId) + sizeof(tId)] = {0}; char id[sizeof(qId) + sizeof(tId)] = {0};
@ -276,8 +264,6 @@ int32_t qwAcquireTaskStatus(QW_FPARAMS_DEF, int32_t rwType, SQWSchStatus *sch, S
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t qwAddTaskStatusImpl(QW_FPARAMS_DEF, SQWSchStatus *sch, int32_t rwType, int32_t status, SQWTaskStatus **task) { int32_t qwAddTaskStatusImpl(QW_FPARAMS_DEF, SQWSchStatus *sch, int32_t rwType, int32_t status, SQWTaskStatus **task) {
int32_t code = 0; int32_t code = 0;
@ -329,16 +315,12 @@ _return:
QW_RET(code); QW_RET(code);
} }
int32_t qwAddAcquireTaskStatus(QW_FPARAMS_DEF, int32_t rwType, SQWSchStatus *sch, int32_t status,
int32_t qwAddAcquireTaskStatus(QW_FPARAMS_DEF, int32_t rwType, SQWSchStatus *sch, int32_t status, SQWTaskStatus **task) { SQWTaskStatus **task) {
return qwAddTaskStatusImpl(QW_FPARAMS(), sch, rwType, status, task); return qwAddTaskStatusImpl(QW_FPARAMS(), sch, rwType, status, task);
} }
void qwReleaseTaskStatus(int32_t rwType, SQWSchStatus *sch) { QW_UNLOCK(rwType, &sch->tasksLock); }
void qwReleaseTaskStatus(int32_t rwType, SQWSchStatus *sch) {
QW_UNLOCK(rwType, &sch->tasksLock);
}
int32_t qwAcquireTaskCtx(QW_FPARAMS_DEF, SQWTaskCtx **ctx) { int32_t qwAcquireTaskCtx(QW_FPARAMS_DEF, SQWTaskCtx **ctx) {
char id[sizeof(qId) + sizeof(tId)] = {0}; char id[sizeof(qId) + sizeof(tId)] = {0};
@ -398,17 +380,11 @@ int32_t qwAddTaskCtxImpl(QW_FPARAMS_DEF, bool acquire, SQWTaskCtx **ctx) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t qwAddTaskCtx(QW_FPARAMS_DEF) { int32_t qwAddTaskCtx(QW_FPARAMS_DEF) { QW_RET(qwAddTaskCtxImpl(QW_FPARAMS(), false, NULL)); }
QW_RET(qwAddTaskCtxImpl(QW_FPARAMS(), false, NULL));
}
int32_t qwAddAcquireTaskCtx(QW_FPARAMS_DEF, SQWTaskCtx **ctx) { int32_t qwAddAcquireTaskCtx(QW_FPARAMS_DEF, SQWTaskCtx **ctx) { return qwAddTaskCtxImpl(QW_FPARAMS(), true, ctx); }
return qwAddTaskCtxImpl(QW_FPARAMS(), true, ctx);
}
void qwReleaseTaskCtx(SQWorkerMgmt *mgmt, void *ctx) { void qwReleaseTaskCtx(SQWorkerMgmt *mgmt, void *ctx) { taosHashRelease(mgmt->ctxHash, ctx); }
taosHashRelease(mgmt->ctxHash, ctx);
}
void qwFreeTaskHandle(QW_FPARAMS_DEF, qTaskInfo_t *taskHandle) { void qwFreeTaskHandle(QW_FPARAMS_DEF, qTaskInfo_t *taskHandle) {
// Note: free/kill may in RC // Note: free/kill may in RC
@ -430,10 +406,10 @@ int32_t qwKillTaskHandle(QW_FPARAMS_DEF, SQWTaskCtx *ctx) {
QW_RET(code); QW_RET(code);
} }
void qwFreeTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx) { void qwFreeTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx) {
tmsgReleaseHandle(ctx->ctrlConnInfo.handle, TAOS_CONN_SERVER); tmsgReleaseHandle(ctx->ctrlConnInfo.handle, TAOS_CONN_SERVER);
ctx->ctrlConnInfo.handle = NULL; ctx->ctrlConnInfo.handle = NULL;
ctx->ctrlConnInfo.refId = -1;
// NO need to release dataConnInfo // NO need to release dataConnInfo
@ -445,7 +421,6 @@ void qwFreeTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx) {
} }
} }
int32_t qwDropTaskCtx(QW_FPARAMS_DEF) { int32_t qwDropTaskCtx(QW_FPARAMS_DEF) {
char id[sizeof(qId) + sizeof(tId)] = {0}; char id[sizeof(qId) + sizeof(tId)] = {0};
QW_SET_QTID(id, qId, tId); QW_SET_QTID(id, qId, tId);
@ -476,7 +451,7 @@ int32_t qwDropTaskCtx(QW_FPARAMS_DEF) {
} }
int32_t qwDropTaskStatus(QW_FPARAMS_DEF) { int32_t qwDropTaskStatus(QW_FPARAMS_DEF) {
SQWSchStatus *sch = NULL; SQWSchStatus * sch = NULL;
SQWTaskStatus *task = NULL; SQWTaskStatus *task = NULL;
int32_t code = 0; int32_t code = 0;
@ -513,7 +488,7 @@ _return:
} }
int32_t qwUpdateTaskStatus(QW_FPARAMS_DEF, int8_t status) { int32_t qwUpdateTaskStatus(QW_FPARAMS_DEF, int8_t status) {
SQWSchStatus *sch = NULL; SQWSchStatus * sch = NULL;
SQWTaskStatus *task = NULL; SQWTaskStatus *task = NULL;
int32_t code = 0; int32_t code = 0;
@ -541,7 +516,6 @@ int32_t qwDropTask(QW_FPARAMS_DEF) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t qwHandleTaskComplete(QW_FPARAMS_DEF, SQWTaskCtx *ctx) { int32_t qwHandleTaskComplete(QW_FPARAMS_DEF, SQWTaskCtx *ctx) {
qTaskInfo_t *taskHandle = &ctx->taskHandle; qTaskInfo_t *taskHandle = &ctx->taskHandle;
@ -553,6 +527,7 @@ int32_t qwHandleTaskComplete(QW_FPARAMS_DEF, SQWTaskCtx *ctx) {
SQWConnInfo connInfo = {0}; SQWConnInfo connInfo = {0};
connInfo.handle = ctx->ctrlConnInfo.handle; connInfo.handle = ctx->ctrlConnInfo.handle;
connInfo.refId = ctx->ctrlConnInfo.refId;
QW_ERR_RET(qwBuildAndSendExplainRsp(&connInfo, execInfo, resNum)); QW_ERR_RET(qwBuildAndSendExplainRsp(&connInfo, execInfo, resNum));
} }
@ -563,15 +538,14 @@ int32_t qwHandleTaskComplete(QW_FPARAMS_DEF, SQWTaskCtx *ctx) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryEnd) { int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryEnd) {
int32_t code = 0; int32_t code = 0;
bool qcontinue = true; bool qcontinue = true;
SSDataBlock* pRes = NULL; SSDataBlock * pRes = NULL;
uint64_t useconds = 0; uint64_t useconds = 0;
int32_t i = 0; int32_t i = 0;
int32_t execNum = 0; int32_t execNum = 0;
qTaskInfo_t *taskHandle = &ctx->taskHandle; qTaskInfo_t * taskHandle = &ctx->taskHandle;
DataSinkHandle sinkHandle = ctx->sinkHandle; DataSinkHandle sinkHandle = ctx->sinkHandle;
while (true) { while (true) {
@ -586,7 +560,7 @@ int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryEnd) {
++execNum; ++execNum;
if (NULL == pRes) { if (NULL == pRes) {
QW_TASK_DLOG("qExecTask end with empty res, useconds:%"PRIu64, useconds); QW_TASK_DLOG("qExecTask end with empty res, useconds:%" PRIu64, useconds);
dsEndPut(sinkHandle, useconds); dsEndPut(sinkHandle, useconds);
@ -649,7 +623,7 @@ int32_t qwGenerateSchHbRsp(SQWorkerMgmt *mgmt, SQWSchStatus *sch, SQWHbInfo *hbI
return TSDB_CODE_QRY_OUT_OF_MEMORY; return TSDB_CODE_QRY_OUT_OF_MEMORY;
} }
void *key = NULL; void * key = NULL;
size_t keyLen = 0; size_t keyLen = 0;
int32_t i = 0; int32_t i = 0;
STaskStatus status = {0}; STaskStatus status = {0};
@ -659,7 +633,7 @@ int32_t qwGenerateSchHbRsp(SQWorkerMgmt *mgmt, SQWSchStatus *sch, SQWHbInfo *hbI
SQWTaskStatus *taskStatus = (SQWTaskStatus *)pIter; SQWTaskStatus *taskStatus = (SQWTaskStatus *)pIter;
key = taosHashGetKey(pIter, &keyLen); key = taosHashGetKey(pIter, &keyLen);
//TODO GET EXECUTOR API TO GET MORE INFO // TODO GET EXECUTOR API TO GET MORE INFO
QW_GET_QTID(key, status.queryId, status.taskId); QW_GET_QTID(key, status.queryId, status.taskId);
status.status = taskStatus->status; status.status = taskStatus->status;
@ -676,7 +650,6 @@ int32_t qwGenerateSchHbRsp(SQWorkerMgmt *mgmt, SQWSchStatus *sch, SQWHbInfo *hbI
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t qwGetResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen, void **rspMsg, SOutputData *pOutput) { int32_t qwGetResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen, void **rspMsg, SOutputData *pOutput) {
int32_t len = 0; int32_t len = 0;
SRetrieveTableRsp *rsp = NULL; SRetrieveTableRsp *rsp = NULL;
@ -738,7 +711,7 @@ int32_t qwGetResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen, void
int32_t qwHandlePrePhaseEvents(QW_FPARAMS_DEF, int8_t phase, SQWPhaseInput *input, SQWPhaseOutput *output) { int32_t qwHandlePrePhaseEvents(QW_FPARAMS_DEF, int8_t phase, SQWPhaseInput *input, SQWPhaseOutput *output) {
int32_t code = 0; int32_t code = 0;
SQWTaskCtx *ctx = NULL; SQWTaskCtx * ctx = NULL;
SQWConnInfo *dropConnection = NULL; SQWConnInfo *dropConnection = NULL;
SQWConnInfo *cancelConnection = NULL; SQWConnInfo *cancelConnection = NULL;
@ -753,12 +726,12 @@ int32_t qwHandlePrePhaseEvents(QW_FPARAMS_DEF, int8_t phase, SQWPhaseInput *inpu
QW_LOCK(QW_WRITE, &ctx->lock); QW_LOCK(QW_WRITE, &ctx->lock);
if (QW_PHASE_PRE_FETCH == phase) { if (QW_PHASE_PRE_FETCH == phase) {
atomic_store_8((int8_t*)&ctx->queryFetched, true); atomic_store_8((int8_t *)&ctx->queryFetched, true);
} else { } else {
atomic_store_8(&ctx->phase, phase); atomic_store_8(&ctx->phase, phase);
} }
if (atomic_load_8((int8_t*)&ctx->queryEnd)) { if (atomic_load_8((int8_t *)&ctx->queryEnd)) {
QW_TASK_ELOG_E("query already end"); QW_TASK_ELOG_E("query already end");
QW_ERR_JRET(TSDB_CODE_QW_MSG_ERROR); QW_ERR_JRET(TSDB_CODE_QW_MSG_ERROR);
} }
@ -828,7 +801,8 @@ int32_t qwHandlePrePhaseEvents(QW_FPARAMS_DEF, int8_t phase, SQWPhaseInput *inpu
} }
if (ctx->rspCode) { if (ctx->rspCode) {
QW_TASK_ELOG("task already failed at phase %s, error:%x - %s", qwPhaseStr(phase), ctx->rspCode, tstrerror(ctx->rspCode)); QW_TASK_ELOG("task already failed at phase %s, error:%x - %s", qwPhaseStr(phase), ctx->rspCode,
tstrerror(ctx->rspCode));
QW_ERR_JRET(ctx->rspCode); QW_ERR_JRET(ctx->rspCode);
} }
@ -856,10 +830,9 @@ _return:
QW_RET(code); QW_RET(code);
} }
int32_t qwHandlePostPhaseEvents(QW_FPARAMS_DEF, int8_t phase, SQWPhaseInput *input, SQWPhaseOutput *output) { int32_t qwHandlePostPhaseEvents(QW_FPARAMS_DEF, int8_t phase, SQWPhaseInput *input, SQWPhaseOutput *output) {
int32_t code = 0; int32_t code = 0;
SQWTaskCtx *ctx = NULL; SQWTaskCtx * ctx = NULL;
SQWConnInfo connInfo = {0}; SQWConnInfo connInfo = {0};
SQWConnInfo *readyConnection = NULL; SQWConnInfo *readyConnection = NULL;
@ -882,6 +855,7 @@ int32_t qwHandlePostPhaseEvents(QW_FPARAMS_DEF, int8_t phase, SQWPhaseInput *inp
} }
#else #else
connInfo.handle = ctx->ctrlConnInfo.handle; connInfo.handle = ctx->ctrlConnInfo.handle;
connInfo.refId = ctx->ctrlConnInfo.refId;
readyConnection = &connInfo; readyConnection = &connInfo;
QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_READY); QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_READY);
@ -902,7 +876,8 @@ int32_t qwHandlePostPhaseEvents(QW_FPARAMS_DEF, int8_t phase, SQWPhaseInput *inp
} }
if (ctx->rspCode) { if (ctx->rspCode) {
QW_TASK_ELOG("task already failed, phase %s, error:%x - %s", qwPhaseStr(phase), ctx->rspCode, tstrerror(ctx->rspCode)); QW_TASK_ELOG("task already failed, phase %s, error:%x - %s", qwPhaseStr(phase), ctx->rspCode,
tstrerror(ctx->rspCode));
QW_ERR_JRET(ctx->rspCode); QW_ERR_JRET(ctx->rspCode);
} }
@ -946,7 +921,7 @@ int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, int8_t taskType, int8_t ex
SQWPhaseInput input = {0}; SQWPhaseInput input = {0};
qTaskInfo_t pTaskInfo = NULL; qTaskInfo_t pTaskInfo = NULL;
DataSinkHandle sinkHandle = NULL; DataSinkHandle sinkHandle = NULL;
SQWTaskCtx *ctx = NULL; SQWTaskCtx * ctx = NULL;
QW_ERR_JRET(qwRegisterQueryBrokenLinkArg(QW_FPARAMS(), &qwMsg->connInfo)); QW_ERR_JRET(qwRegisterQueryBrokenLinkArg(QW_FPARAMS(), &qwMsg->connInfo));
@ -959,6 +934,7 @@ int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, int8_t taskType, int8_t ex
atomic_store_ptr(&ctx->ctrlConnInfo.handle, qwMsg->connInfo.handle); atomic_store_ptr(&ctx->ctrlConnInfo.handle, qwMsg->connInfo.handle);
atomic_store_ptr(&ctx->ctrlConnInfo.ahandle, qwMsg->connInfo.ahandle); atomic_store_ptr(&ctx->ctrlConnInfo.ahandle, qwMsg->connInfo.ahandle);
atomic_store_64(&ctx->ctrlConnInfo.refId, qwMsg->connInfo.refId);
QW_TASK_DLOGL("subplan json string, len:%d, %s", qwMsg->msgLen, qwMsg->msg); QW_TASK_DLOGL("subplan json string, len:%d, %s", qwMsg->msgLen, qwMsg->msg);
@ -979,10 +955,10 @@ int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, int8_t taskType, int8_t ex
QW_ERR_JRET(TSDB_CODE_QRY_APP_ERROR); QW_ERR_JRET(TSDB_CODE_QRY_APP_ERROR);
} }
//QW_ERR_JRET(qwBuildAndSendQueryRsp(&qwMsg->connInfo, code)); // QW_ERR_JRET(qwBuildAndSendQueryRsp(&qwMsg->connInfo, code));
//QW_TASK_DLOG("query msg rsped, handle:%p, code:%x - %s", qwMsg->connInfo.handle, code, tstrerror(code)); // QW_TASK_DLOG("query msg rsped, handle:%p, code:%x - %s", qwMsg->connInfo.handle, code, tstrerror(code));
//queryRsped = true; // queryRsped = true;
atomic_store_ptr(&ctx->taskHandle, pTaskInfo); atomic_store_ptr(&ctx->taskHandle, pTaskInfo);
atomic_store_ptr(&ctx->sinkHandle, sinkHandle); atomic_store_ptr(&ctx->sinkHandle, sinkHandle);
@ -996,7 +972,7 @@ _return:
input.code = code; input.code = code;
code = qwHandlePostPhaseEvents(QW_FPARAMS(), QW_PHASE_POST_QUERY, &input, NULL); code = qwHandlePostPhaseEvents(QW_FPARAMS(), QW_PHASE_POST_QUERY, &input, NULL);
//if (!queryRsped) { // if (!queryRsped) {
// qwBuildAndSendQueryRsp(&qwMsg->connInfo, code); // qwBuildAndSendQueryRsp(&qwMsg->connInfo, code);
// QW_TASK_DLOG("query msg rsped, handle:%p, code:%x - %s", qwMsg->connInfo.handle, code, tstrerror(code)); // QW_TASK_DLOG("query msg rsped, handle:%p, code:%x - %s", qwMsg->connInfo.handle, code, tstrerror(code));
//} //}
@ -1030,8 +1006,9 @@ int32_t qwProcessReady(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_READY); QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_READY);
if (atomic_load_8((int8_t*)&ctx->queryEnd) || atomic_load_8((int8_t*)&ctx->queryFetched)) { if (atomic_load_8((int8_t *)&ctx->queryEnd) || atomic_load_8((int8_t *)&ctx->queryFetched)) {
QW_TASK_ELOG("got ready msg at wrong status, queryEnd:%d, queryFetched:%d", atomic_load_8((int8_t*)&ctx->queryEnd), atomic_load_8((int8_t*)&ctx->queryFetched)); QW_TASK_ELOG("got ready msg at wrong status, queryEnd:%d, queryFetched:%d", atomic_load_8((int8_t *)&ctx->queryEnd),
atomic_load_8((int8_t *)&ctx->queryFetched));
QW_ERR_JRET(TSDB_CODE_QW_MSG_ERROR); QW_ERR_JRET(TSDB_CODE_QW_MSG_ERROR);
} }
@ -1067,12 +1044,11 @@ _return:
QW_RET(TSDB_CODE_SUCCESS); QW_RET(TSDB_CODE_SUCCESS);
} }
int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg) { int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
SQWTaskCtx *ctx = NULL; SQWTaskCtx * ctx = NULL;
int32_t code = 0; int32_t code = 0;
SQWPhaseInput input = {0}; SQWPhaseInput input = {0};
void *rsp = NULL; void * rsp = NULL;
int32_t dataLen = 0; int32_t dataLen = 0;
bool queryEnd = false; bool queryEnd = false;
@ -1081,8 +1057,8 @@ int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx)); QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx));
atomic_store_8((int8_t*)&ctx->queryInQueue, 0); atomic_store_8((int8_t *)&ctx->queryInQueue, 0);
atomic_store_8((int8_t*)&ctx->queryContinue, 0); atomic_store_8((int8_t *)&ctx->queryContinue, 0);
QW_ERR_JRET(qwExecTask(QW_FPARAMS(), ctx, &queryEnd)); QW_ERR_JRET(qwExecTask(QW_FPARAMS(), ctx, &queryEnd));
@ -1093,7 +1069,7 @@ int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
if ((!sOutput.queryEnd) && (DS_BUF_LOW == sOutput.bufStatus || DS_BUF_EMPTY == sOutput.bufStatus)) { if ((!sOutput.queryEnd) && (DS_BUF_LOW == sOutput.bufStatus || DS_BUF_EMPTY == sOutput.bufStatus)) {
QW_TASK_DLOG("task not end and buf is %s, need to continue query", qwBufStatusStr(sOutput.bufStatus)); QW_TASK_DLOG("task not end and buf is %s, need to continue query", qwBufStatusStr(sOutput.bufStatus));
atomic_store_8((int8_t*)&ctx->queryContinue, 1); atomic_store_8((int8_t *)&ctx->queryContinue, 1);
} }
if (rsp) { if (rsp) {
@ -1101,20 +1077,21 @@ int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
qwBuildFetchRsp(rsp, &sOutput, dataLen, qComplete); qwBuildFetchRsp(rsp, &sOutput, dataLen, qComplete);
if (qComplete) { if (qComplete) {
atomic_store_8((int8_t*)&ctx->queryEnd, true); atomic_store_8((int8_t *)&ctx->queryEnd, true);
} }
qwMsg->connInfo = ctx->dataConnInfo; qwMsg->connInfo = ctx->dataConnInfo;
QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_FETCH); QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_FETCH);
qwBuildAndSendFetchRsp(&qwMsg->connInfo, rsp, dataLen, code); qwBuildAndSendFetchRsp(&qwMsg->connInfo, rsp, dataLen, code);
QW_TASK_DLOG("fetch rsp send, handle:%p, code:%x - %s, dataLen:%d", qwMsg->connInfo.handle, code, tstrerror(code), dataLen); QW_TASK_DLOG("fetch rsp send, handle:%p, code:%x - %s, dataLen:%d", qwMsg->connInfo.handle, code,
tstrerror(code), dataLen);
} else { } else {
atomic_store_8((int8_t*)&ctx->queryContinue, 1); atomic_store_8((int8_t *)&ctx->queryContinue, 1);
} }
} }
_return: _return:
if (NULL == ctx) { if (NULL == ctx) {
break; break;
@ -1127,17 +1104,18 @@ _return:
qwMsg->connInfo = ctx->dataConnInfo; qwMsg->connInfo = ctx->dataConnInfo;
qwBuildAndSendFetchRsp(&qwMsg->connInfo, rsp, 0, code); qwBuildAndSendFetchRsp(&qwMsg->connInfo, rsp, 0, code);
QW_TASK_DLOG("fetch rsp send, handle:%p, code:%x - %s, dataLen:%d", qwMsg->connInfo.handle, code, tstrerror(code), 0); QW_TASK_DLOG("fetch rsp send, handle:%p, code:%x - %s, dataLen:%d", qwMsg->connInfo.handle, code, tstrerror(code),
0);
} }
QW_LOCK(QW_WRITE, &ctx->lock); QW_LOCK(QW_WRITE, &ctx->lock);
if (queryEnd || code || 0 == atomic_load_8((int8_t*)&ctx->queryContinue)) { if (queryEnd || code || 0 == atomic_load_8((int8_t *)&ctx->queryContinue)) {
// Note: if necessary, fetch need to put cquery to queue again // Note: if necessary, fetch need to put cquery to queue again
atomic_store_8(&ctx->phase, 0); atomic_store_8(&ctx->phase, 0);
QW_UNLOCK(QW_WRITE,&ctx->lock); QW_UNLOCK(QW_WRITE, &ctx->lock);
break; break;
} }
QW_UNLOCK(QW_WRITE,&ctx->lock); QW_UNLOCK(QW_WRITE, &ctx->lock);
} while (true); } while (true);
input.code = code; input.code = code;
@ -1146,13 +1124,12 @@ _return:
QW_RET(TSDB_CODE_SUCCESS); QW_RET(TSDB_CODE_SUCCESS);
} }
int32_t qwProcessFetch(QW_FPARAMS_DEF, SQWMsg *qwMsg) { int32_t qwProcessFetch(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
int32_t code = 0; int32_t code = 0;
int32_t dataLen = 0; int32_t dataLen = 0;
bool locked = false; bool locked = false;
SQWTaskCtx *ctx = NULL; SQWTaskCtx * ctx = NULL;
void *rsp = NULL; void * rsp = NULL;
SQWPhaseInput input = {0}; SQWPhaseInput input = {0};
QW_ERR_JRET(qwHandlePrePhaseEvents(QW_FPARAMS(), QW_PHASE_PRE_FETCH, &input, NULL)); QW_ERR_JRET(qwHandlePrePhaseEvents(QW_FPARAMS(), QW_PHASE_PRE_FETCH, &input, NULL));
@ -1172,7 +1149,7 @@ int32_t qwProcessFetch(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
qwBuildFetchRsp(rsp, &sOutput, dataLen, qComplete); qwBuildFetchRsp(rsp, &sOutput, dataLen, qComplete);
if (qComplete) { if (qComplete) {
atomic_store_8((int8_t*)&ctx->queryEnd, true); atomic_store_8((int8_t *)&ctx->queryEnd, true);
} }
} }
@ -1184,11 +1161,11 @@ int32_t qwProcessFetch(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
// RC WARNING // RC WARNING
if (QW_IS_QUERY_RUNNING(ctx)) { if (QW_IS_QUERY_RUNNING(ctx)) {
atomic_store_8((int8_t*)&ctx->queryContinue, 1); atomic_store_8((int8_t *)&ctx->queryContinue, 1);
} else if (0 == atomic_load_8((int8_t*)&ctx->queryInQueue)) { } else if (0 == atomic_load_8((int8_t *)&ctx->queryInQueue)) {
qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_EXECUTING); qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_EXECUTING);
atomic_store_8((int8_t*)&ctx->queryInQueue, 1); atomic_store_8((int8_t *)&ctx->queryInQueue, 1);
QW_ERR_JRET(qwBuildAndSendCQueryMsg(QW_FPARAMS(), &qwMsg->connInfo)); QW_ERR_JRET(qwBuildAndSendCQueryMsg(QW_FPARAMS(), &qwMsg->connInfo));
} }
@ -1211,13 +1188,13 @@ _return:
if (code || rsp) { if (code || rsp) {
qwBuildAndSendFetchRsp(&qwMsg->connInfo, rsp, dataLen, code); qwBuildAndSendFetchRsp(&qwMsg->connInfo, rsp, dataLen, code);
QW_TASK_DLOG("fetch rsp send, handle:%p, code:%x - %s, dataLen:%d", qwMsg->connInfo.handle, code, tstrerror(code), dataLen); QW_TASK_DLOG("fetch rsp send, handle:%p, code:%x - %s, dataLen:%d", qwMsg->connInfo.handle, code, tstrerror(code),
dataLen);
} }
QW_RET(TSDB_CODE_SUCCESS); QW_RET(TSDB_CODE_SUCCESS);
} }
int32_t qwProcessDrop(QW_FPARAMS_DEF, SQWMsg *qwMsg) { int32_t qwProcessDrop(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
int32_t code = 0; int32_t code = 0;
bool rsped = false; bool rsped = false;
@ -1288,7 +1265,7 @@ _return:
int32_t qwProcessHbLinkBroken(SQWorkerMgmt *mgmt, SQWMsg *qwMsg, SSchedulerHbReq *req) { int32_t qwProcessHbLinkBroken(SQWorkerMgmt *mgmt, SQWMsg *qwMsg, SSchedulerHbReq *req) {
int32_t code = 0; int32_t code = 0;
SSchedulerHbRsp rsp = {0}; SSchedulerHbRsp rsp = {0};
SQWSchStatus *sch = NULL; SQWSchStatus * sch = NULL;
QW_ERR_RET(qwAcquireAddScheduler(mgmt, req->sId, QW_READ, &sch)); QW_ERR_RET(qwAcquireAddScheduler(mgmt, req->sId, QW_READ, &sch));
@ -1314,7 +1291,7 @@ int32_t qwProcessHbLinkBroken(SQWorkerMgmt *mgmt, SQWMsg *qwMsg, SSchedulerHbReq
int32_t qwProcessHb(SQWorkerMgmt *mgmt, SQWMsg *qwMsg, SSchedulerHbReq *req) { int32_t qwProcessHb(SQWorkerMgmt *mgmt, SQWMsg *qwMsg, SSchedulerHbReq *req) {
int32_t code = 0; int32_t code = 0;
SSchedulerHbRsp rsp = {0}; SSchedulerHbRsp rsp = {0};
SQWSchStatus *sch = NULL; SQWSchStatus * sch = NULL;
if (qwMsg->code) { if (qwMsg->code) {
QW_RET(qwProcessHbLinkBroken(mgmt, qwMsg, req)); QW_RET(qwProcessHbLinkBroken(mgmt, qwMsg, req));
@ -1335,8 +1312,8 @@ int32_t qwProcessHb(SQWorkerMgmt *mgmt, SQWMsg *qwMsg, SSchedulerHbReq *req) {
QW_UNLOCK(QW_WRITE, &sch->hbConnLock); QW_UNLOCK(QW_WRITE, &sch->hbConnLock);
QW_DLOG("hb connection updated, sId:%" PRIx64 ", nodeId:%d, fqdn:%s, port:%d, handle:%p, ahandle:%p", QW_DLOG("hb connection updated, sId:%" PRIx64 ", nodeId:%d, fqdn:%s, port:%d, handle:%p, ahandle:%p", req->sId,
req->sId, req->epId.nodeId, req->epId.ep.fqdn, req->epId.ep.port, qwMsg->connInfo.handle, qwMsg->connInfo.ahandle); req->epId.nodeId, req->epId.ep.fqdn, req->epId.ep.port, qwMsg->connInfo.handle, qwMsg->connInfo.ahandle);
qwReleaseScheduler(QW_READ, mgmt); qwReleaseScheduler(QW_READ, mgmt);
@ -1355,12 +1332,11 @@ _return:
QW_RET(TSDB_CODE_SUCCESS); QW_RET(TSDB_CODE_SUCCESS);
} }
void qwProcessHbTimerEvent(void *param, void *tmrId) { void qwProcessHbTimerEvent(void *param, void *tmrId) {
SQWorkerMgmt *mgmt = (SQWorkerMgmt *)param; SQWorkerMgmt *mgmt = (SQWorkerMgmt *)param;
SQWSchStatus *sch = NULL; SQWSchStatus *sch = NULL;
int32_t taskNum = 0; int32_t taskNum = 0;
SQWHbInfo *rspList = NULL; SQWHbInfo * rspList = NULL;
int32_t code = 0; int32_t code = 0;
qwDbgDumpMgmtInfo(mgmt); qwDbgDumpMgmtInfo(mgmt);
@ -1382,7 +1358,7 @@ void qwProcessHbTimerEvent(void *param, void *tmrId) {
return; return;
} }
void *key = NULL; void * key = NULL;
size_t keyLen = 0; size_t keyLen = 0;
int32_t i = 0; int32_t i = 0;
@ -1452,14 +1428,16 @@ int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qW
mgmt->cfg.maxSchTaskNum = QW_DEFAULT_SCH_TASK_NUMBER; mgmt->cfg.maxSchTaskNum = QW_DEFAULT_SCH_TASK_NUMBER;
} }
mgmt->schHash = taosHashInit(mgmt->cfg.maxSchedulerNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_ENTRY_LOCK); mgmt->schHash = taosHashInit(mgmt->cfg.maxSchedulerNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false,
HASH_ENTRY_LOCK);
if (NULL == mgmt->schHash) { if (NULL == mgmt->schHash) {
taosMemoryFreeClear(mgmt); taosMemoryFreeClear(mgmt);
qError("init %d scheduler hash failed", mgmt->cfg.maxSchedulerNum); qError("init %d scheduler hash failed", mgmt->cfg.maxSchedulerNum);
QW_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY); QW_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
} }
mgmt->ctxHash = taosHashInit(mgmt->cfg.maxTaskNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK); mgmt->ctxHash =
taosHashInit(mgmt->cfg.maxTaskNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
if (NULL == mgmt->ctxHash) { if (NULL == mgmt->ctxHash) {
qError("init %d task ctx hash failed", mgmt->cfg.maxTaskNum); qError("init %d task ctx hash failed", mgmt->cfg.maxTaskNum);
QW_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY); QW_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
@ -1509,9 +1487,9 @@ void qWorkerDestroy(void **qWorkerMgmt) {
taosTmrStopA(&mgmt->hbTimer); taosTmrStopA(&mgmt->hbTimer);
taosTmrCleanUp(mgmt->timer); taosTmrCleanUp(mgmt->timer);
//TODO STOP ALL QUERY // TODO STOP ALL QUERY
//TODO FREE ALL // TODO FREE ALL
taosHashCleanup(mgmt->ctxHash); taosHashCleanup(mgmt->ctxHash);
taosHashCleanup(mgmt->schHash); taosHashCleanup(mgmt->schHash);
@ -1520,7 +1498,7 @@ void qWorkerDestroy(void **qWorkerMgmt) {
} }
int32_t qwGetSchTasksStatus(SQWorkerMgmt *mgmt, uint64_t sId, SSchedulerStatusRsp **rsp) { int32_t qwGetSchTasksStatus(SQWorkerMgmt *mgmt, uint64_t sId, SSchedulerStatusRsp **rsp) {
/* /*
SQWSchStatus *sch = NULL; SQWSchStatus *sch = NULL;
int32_t taskNum = 0; int32_t taskNum = 0;
@ -1562,32 +1540,29 @@ int32_t qwGetSchTasksStatus(SQWorkerMgmt *mgmt, uint64_t sId, SSchedulerStatusRs
qwReleaseScheduler(QW_READ, mgmt); qwReleaseScheduler(QW_READ, mgmt);
(*rsp)->num = taskNum; (*rsp)->num = taskNum;
*/ */
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t qwUpdateSchLastAccess(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId) { int32_t qwUpdateSchLastAccess(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId) {
SQWSchStatus *sch = NULL; SQWSchStatus *sch = NULL;
/* /*
QW_ERR_RET(qwAcquireScheduler(QW_READ, mgmt, sId, &sch)); QW_ERR_RET(qwAcquireScheduler(QW_READ, mgmt, sId, &sch));
sch->lastAccessTs = taosGetTimestampSec(); sch->lastAccessTs = taosGetTimestampSec();
qwReleaseScheduler(QW_READ, mgmt); qwReleaseScheduler(QW_READ, mgmt);
*/ */
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t qwGetTaskStatus(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, int8_t *taskStatus) { int32_t qwGetTaskStatus(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, int8_t *taskStatus) {
SQWSchStatus *sch = NULL; SQWSchStatus * sch = NULL;
SQWTaskStatus *task = NULL; SQWTaskStatus *task = NULL;
int32_t code = 0; int32_t code = 0;
/* /*
if (qwAcquireScheduler(QW_READ, mgmt, sId, &sch)) { if (qwAcquireScheduler(QW_READ, mgmt, sId, &sch)) {
*taskStatus = JOB_TASK_STATUS_NULL; *taskStatus = JOB_TASK_STATUS_NULL;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -1604,18 +1579,17 @@ int32_t qwGetTaskStatus(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t
qwReleaseTask(QW_READ, sch); qwReleaseTask(QW_READ, sch);
qwReleaseScheduler(QW_READ, mgmt); qwReleaseScheduler(QW_READ, mgmt);
*/ */
QW_RET(code); QW_RET(code);
} }
int32_t qwCancelTask(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId) { int32_t qwCancelTask(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId) {
SQWSchStatus *sch = NULL; SQWSchStatus * sch = NULL;
SQWTaskStatus *task = NULL; SQWTaskStatus *task = NULL;
int32_t code = 0; int32_t code = 0;
/* /*
QW_ERR_RET(qwAcquireAddScheduler(QW_READ, mgmt, sId, &sch)); QW_ERR_RET(qwAcquireAddScheduler(QW_READ, mgmt, sId, &sch));
QW_ERR_JRET(qwAcquireAddTask(mgmt, QW_READ, sch, qId, tId, JOB_TASK_STATUS_NOT_START, &task)); QW_ERR_JRET(qwAcquireAddTask(mgmt, QW_READ, sch, qId, tId, JOB_TASK_STATUS_NOT_START, &task));
@ -1628,16 +1602,15 @@ int32_t qwCancelTask(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tI
int8_t oriStatus = task->status; int8_t oriStatus = task->status;
int8_t newStatus = 0; int8_t newStatus = 0;
if (task->status == JOB_TASK_STATUS_CANCELLED || task->status == JOB_TASK_STATUS_NOT_START || task->status == JOB_TASK_STATUS_CANCELLING || task->status == JOB_TASK_STATUS_DROPPING) { if (task->status == JOB_TASK_STATUS_CANCELLED || task->status == JOB_TASK_STATUS_NOT_START || task->status ==
QW_UNLOCK(QW_WRITE, &task->lock); JOB_TASK_STATUS_CANCELLING || task->status == JOB_TASK_STATUS_DROPPING) { QW_UNLOCK(QW_WRITE, &task->lock);
qwReleaseTask(QW_READ, sch); qwReleaseTask(QW_READ, sch);
qwReleaseScheduler(QW_READ, mgmt); qwReleaseScheduler(QW_READ, mgmt);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} else if (task->status == JOB_TASK_STATUS_FAILED || task->status == JOB_TASK_STATUS_SUCCEED || task->status == JOB_TASK_STATUS_PARTIAL_SUCCEED) { } else if (task->status == JOB_TASK_STATUS_FAILED || task->status == JOB_TASK_STATUS_SUCCEED || task->status ==
QW_ERR_JRET(qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_CANCELLED)); JOB_TASK_STATUS_PARTIAL_SUCCEED) { QW_ERR_JRET(qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_CANCELLED)); } else {
} else {
QW_ERR_JRET(qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_CANCELLING)); QW_ERR_JRET(qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_CANCELLING));
} }
@ -1652,7 +1625,7 @@ int32_t qwCancelTask(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tI
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
_return: _return:
if (task) { if (task) {
QW_UNLOCK(QW_WRITE, &task->lock); QW_UNLOCK(QW_WRITE, &task->lock);
@ -1663,9 +1636,7 @@ _return:
if (sch) { if (sch) {
qwReleaseScheduler(QW_READ, mgmt); qwReleaseScheduler(QW_READ, mgmt);
} }
*/ */
QW_RET(code); QW_RET(code);
} }

View File

@ -1,14 +1,13 @@
#include "qworker.h" #include "qworkerMsg.h"
#include "tcommon.h" #include "dataSinkMgt.h"
#include "executor.h" #include "executor.h"
#include "planner.h" #include "planner.h"
#include "query.h" #include "query.h"
#include "qworker.h"
#include "qworkerInt.h" #include "qworkerInt.h"
#include "qworkerMsg.h" #include "tcommon.h"
#include "tmsg.h" #include "tmsg.h"
#include "tname.h" #include "tname.h"
#include "dataSinkMgt.h"
int32_t qwMallocFetchRsp(int32_t length, SRetrieveTableRsp **rsp) { int32_t qwMallocFetchRsp(int32_t length, SRetrieveTableRsp **rsp) {
int32_t msgSize = sizeof(SRetrieveTableRsp) + length; int32_t msgSize = sizeof(SRetrieveTableRsp) + length;
@ -26,8 +25,6 @@ int32_t qwMallocFetchRsp(int32_t length, SRetrieveTableRsp **rsp) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
void qwBuildFetchRsp(void *msg, SOutputData *input, int32_t len, bool qComplete) { void qwBuildFetchRsp(void *msg, SOutputData *input, int32_t len, bool qComplete) {
SRetrieveTableRsp *rsp = (SRetrieveTableRsp *)msg; SRetrieveTableRsp *rsp = (SRetrieveTableRsp *)msg;
@ -39,7 +36,6 @@ void qwBuildFetchRsp(void *msg, SOutputData *input, int32_t len, bool qComplete)
rsp->numOfRows = htonl(input->numOfRows); rsp->numOfRows = htonl(input->numOfRows);
} }
void qwFreeFetchRsp(void *msg) { void qwFreeFetchRsp(void *msg) {
if (msg) { if (msg) {
rpcFreeCont(msg); rpcFreeCont(msg);
@ -50,13 +46,14 @@ int32_t qwBuildAndSendQueryRsp(SQWConnInfo *pConn, int32_t code) {
SQueryTableRsp rsp = {.code = code}; SQueryTableRsp rsp = {.code = code};
int32_t contLen = tSerializeSQueryTableRsp(NULL, 0, &rsp); int32_t contLen = tSerializeSQueryTableRsp(NULL, 0, &rsp);
void *msg = rpcMallocCont(contLen); void * msg = rpcMallocCont(contLen);
tSerializeSQueryTableRsp(msg, contLen, &rsp); tSerializeSQueryTableRsp(msg, contLen, &rsp);
SRpcMsg rpcRsp = { SRpcMsg rpcRsp = {
.msgType = TDMT_VND_QUERY_RSP, .msgType = TDMT_VND_QUERY_RSP,
.handle = pConn->handle, .handle = pConn->handle,
.ahandle = pConn->ahandle, .ahandle = pConn->ahandle,
.refId = pConn->refId,
.pCont = msg, .pCont = msg,
.contLen = contLen, .contLen = contLen,
.code = code, .code = code,
@ -74,6 +71,7 @@ int32_t qwBuildAndSendReadyRsp(SQWConnInfo *pConn, int32_t code) {
SRpcMsg rpcRsp = { SRpcMsg rpcRsp = {
.msgType = TDMT_VND_RES_READY_RSP, .msgType = TDMT_VND_RES_READY_RSP,
.handle = pConn->handle, .handle = pConn->handle,
.refId = pConn->refId,
.ahandle = NULL, .ahandle = NULL,
.pCont = pRsp, .pCont = pRsp,
.contLen = sizeof(*pRsp), .contLen = sizeof(*pRsp),
@ -89,13 +87,14 @@ int32_t qwBuildAndSendExplainRsp(SQWConnInfo *pConn, SExplainExecInfo *execInfo,
SExplainRsp rsp = {.numOfPlans = num, .subplanInfo = execInfo}; SExplainRsp rsp = {.numOfPlans = num, .subplanInfo = execInfo};
int32_t contLen = tSerializeSExplainRsp(NULL, 0, &rsp); int32_t contLen = tSerializeSExplainRsp(NULL, 0, &rsp);
void *pRsp = rpcMallocCont(contLen); void * pRsp = rpcMallocCont(contLen);
tSerializeSExplainRsp(pRsp, contLen, &rsp); tSerializeSExplainRsp(pRsp, contLen, &rsp);
SRpcMsg rpcRsp = { SRpcMsg rpcRsp = {
.msgType = TDMT_VND_EXPLAIN_RSP, .msgType = TDMT_VND_EXPLAIN_RSP,
.handle = pConn->handle, .handle = pConn->handle,
.ahandle = pConn->ahandle, .ahandle = pConn->ahandle,
.refId = pConn->refId,
.pCont = pRsp, .pCont = pRsp,
.contLen = contLen, .contLen = contLen,
.code = 0, .code = 0,
@ -108,13 +107,14 @@ int32_t qwBuildAndSendExplainRsp(SQWConnInfo *pConn, SExplainExecInfo *execInfo,
int32_t qwBuildAndSendHbRsp(SQWConnInfo *pConn, SSchedulerHbRsp *pStatus, int32_t code) { int32_t qwBuildAndSendHbRsp(SQWConnInfo *pConn, SSchedulerHbRsp *pStatus, int32_t code) {
int32_t contLen = tSerializeSSchedulerHbRsp(NULL, 0, pStatus); int32_t contLen = tSerializeSSchedulerHbRsp(NULL, 0, pStatus);
void *pRsp = rpcMallocCont(contLen); void * pRsp = rpcMallocCont(contLen);
tSerializeSSchedulerHbRsp(pRsp, contLen, pStatus); tSerializeSSchedulerHbRsp(pRsp, contLen, pStatus);
SRpcMsg rpcRsp = { SRpcMsg rpcRsp = {
.msgType = TDMT_VND_QUERY_HEARTBEAT_RSP, .msgType = TDMT_VND_QUERY_HEARTBEAT_RSP,
.handle = pConn->handle, .handle = pConn->handle,
.ahandle = pConn->ahandle, .ahandle = pConn->ahandle,
.refId = pConn->refId,
.pCont = pRsp, .pCont = pRsp,
.contLen = contLen, .contLen = contLen,
.code = code, .code = code,
@ -136,6 +136,7 @@ int32_t qwBuildAndSendFetchRsp(SQWConnInfo *pConn, SRetrieveTableRsp *pRsp, int3
.msgType = TDMT_VND_FETCH_RSP, .msgType = TDMT_VND_FETCH_RSP,
.handle = pConn->handle, .handle = pConn->handle,
.ahandle = pConn->ahandle, .ahandle = pConn->ahandle,
.refId = pConn->refId,
.pCont = pRsp, .pCont = pRsp,
.contLen = sizeof(*pRsp) + dataLength, .contLen = sizeof(*pRsp) + dataLength,
.code = code, .code = code,
@ -154,6 +155,7 @@ int32_t qwBuildAndSendCancelRsp(SQWConnInfo *pConn, int32_t code) {
.msgType = TDMT_VND_CANCEL_TASK_RSP, .msgType = TDMT_VND_CANCEL_TASK_RSP,
.handle = pConn->handle, .handle = pConn->handle,
.ahandle = pConn->ahandle, .ahandle = pConn->ahandle,
.refId = pConn->refId,
.pCont = pRsp, .pCont = pRsp,
.contLen = sizeof(*pRsp), .contLen = sizeof(*pRsp),
.code = code, .code = code,
@ -171,6 +173,7 @@ int32_t qwBuildAndSendDropRsp(SQWConnInfo *pConn, int32_t code) {
.msgType = TDMT_VND_DROP_TASK_RSP, .msgType = TDMT_VND_DROP_TASK_RSP,
.handle = pConn->handle, .handle = pConn->handle,
.ahandle = pConn->ahandle, .ahandle = pConn->ahandle,
.refId = pConn->refId,
.pCont = pRsp, .pCont = pRsp,
.contLen = sizeof(*pRsp), .contLen = sizeof(*pRsp),
.code = code, .code = code,
@ -220,12 +223,13 @@ int32_t qwBuildAndSendShowRsp(SRpcMsg *pMsg, int32_t code) {
showRsp.tableMeta.numOfColumns = cols; showRsp.tableMeta.numOfColumns = cols;
int32_t bufLen = tSerializeSShowRsp(NULL, 0, &showRsp); int32_t bufLen = tSerializeSShowRsp(NULL, 0, &showRsp);
void *pBuf = rpcMallocCont(bufLen); void * pBuf = rpcMallocCont(bufLen);
tSerializeSShowRsp(pBuf, bufLen, &showRsp); tSerializeSShowRsp(pBuf, bufLen, &showRsp);
SRpcMsg rpcMsg = { SRpcMsg rpcMsg = {
.handle = pMsg->handle, .handle = pMsg->handle,
.ahandle = pMsg->ahandle, .ahandle = pMsg->ahandle,
.refId = pMsg->refId,
.pCont = pBuf, .pCont = pBuf,
.contLen = bufLen, .contLen = bufLen,
.code = code, .code = code,
@ -235,7 +239,7 @@ int32_t qwBuildAndSendShowRsp(SRpcMsg *pMsg, int32_t code) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t qwBuildAndSendShowFetchRsp(SRpcMsg *pMsg, SVShowTablesFetchReq* pFetchReq) { int32_t qwBuildAndSendShowFetchRsp(SRpcMsg *pMsg, SVShowTablesFetchReq *pFetchReq) {
SVShowTablesFetchRsp *pRsp = (SVShowTablesFetchRsp *)rpcMallocCont(sizeof(SVShowTablesFetchRsp)); SVShowTablesFetchRsp *pRsp = (SVShowTablesFetchRsp *)rpcMallocCont(sizeof(SVShowTablesFetchRsp));
int32_t handle = htonl(pFetchReq->id); int32_t handle = htonl(pFetchReq->id);
@ -243,6 +247,7 @@ int32_t qwBuildAndSendShowFetchRsp(SRpcMsg *pMsg, SVShowTablesFetchReq* pFetchRe
SRpcMsg rpcMsg = { SRpcMsg rpcMsg = {
.handle = pMsg->handle, .handle = pMsg->handle,
.ahandle = pMsg->ahandle, .ahandle = pMsg->ahandle,
.refId = pMsg->refId,
.pCont = pRsp, .pCont = pRsp,
.contLen = sizeof(*pRsp), .contLen = sizeof(*pRsp),
.code = 0, .code = 0,
@ -253,7 +258,7 @@ int32_t qwBuildAndSendShowFetchRsp(SRpcMsg *pMsg, SVShowTablesFetchReq* pFetchRe
} }
int32_t qwBuildAndSendCQueryMsg(QW_FPARAMS_DEF, SQWConnInfo *pConn) { int32_t qwBuildAndSendCQueryMsg(QW_FPARAMS_DEF, SQWConnInfo *pConn) {
SQueryContinueReq * req = (SQueryContinueReq *)rpcMallocCont(sizeof(SQueryContinueReq)); SQueryContinueReq *req = (SQueryContinueReq *)rpcMallocCont(sizeof(SQueryContinueReq));
if (NULL == req) { if (NULL == req) {
QW_SCH_TASK_ELOG("rpcMallocCont %d failed", (int32_t)sizeof(SQueryContinueReq)); QW_SCH_TASK_ELOG("rpcMallocCont %d failed", (int32_t)sizeof(SQueryContinueReq));
QW_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY); QW_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
@ -268,6 +273,7 @@ int32_t qwBuildAndSendCQueryMsg(QW_FPARAMS_DEF, SQWConnInfo *pConn) {
.handle = pConn->handle, .handle = pConn->handle,
.ahandle = pConn->ahandle, .ahandle = pConn->ahandle,
.msgType = TDMT_VND_QUERY_CONTINUE, .msgType = TDMT_VND_QUERY_CONTINUE,
.refId = pConn->refId,
.pCont = req, .pCont = req,
.contLen = sizeof(SQueryContinueReq), .contLen = sizeof(SQueryContinueReq),
.code = 0, .code = 0,
@ -285,9 +291,8 @@ int32_t qwBuildAndSendCQueryMsg(QW_FPARAMS_DEF, SQWConnInfo *pConn) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t qwRegisterQueryBrokenLinkArg(QW_FPARAMS_DEF, SQWConnInfo *pConn) { int32_t qwRegisterQueryBrokenLinkArg(QW_FPARAMS_DEF, SQWConnInfo *pConn) {
STaskDropReq * req = (STaskDropReq *)rpcMallocCont(sizeof(STaskDropReq)); STaskDropReq *req = (STaskDropReq *)rpcMallocCont(sizeof(STaskDropReq));
if (NULL == req) { if (NULL == req) {
QW_SCH_TASK_ELOG("rpcMallocCont %d failed", (int32_t)sizeof(STaskDropReq)); QW_SCH_TASK_ELOG("rpcMallocCont %d failed", (int32_t)sizeof(STaskDropReq));
QW_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY); QW_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
@ -302,6 +307,7 @@ int32_t qwRegisterQueryBrokenLinkArg(QW_FPARAMS_DEF, SQWConnInfo *pConn) {
SRpcMsg pMsg = { SRpcMsg pMsg = {
.handle = pConn->handle, .handle = pConn->handle,
.ahandle = pConn->ahandle, .ahandle = pConn->ahandle,
.refId = pConn->refId,
.msgType = TDMT_VND_DROP_TASK, .msgType = TDMT_VND_DROP_TASK,
.pCont = req, .pCont = req,
.contLen = sizeof(STaskDropReq), .contLen = sizeof(STaskDropReq),
@ -337,6 +343,7 @@ int32_t qwRegisterHbBrokenLinkArg(SQWorkerMgmt *mgmt, uint64_t sId, SQWConnInfo
SRpcMsg pMsg = { SRpcMsg pMsg = {
.handle = pConn->handle, .handle = pConn->handle,
.ahandle = pConn->ahandle, .ahandle = pConn->ahandle,
.refId = pConn->refId,
.msgType = TDMT_VND_QUERY_HEARTBEAT, .msgType = TDMT_VND_QUERY_HEARTBEAT,
.pCont = msg, .pCont = msg,
.contLen = sizeof(SSchedulerHbReq), .contLen = sizeof(SSchedulerHbReq),
@ -348,8 +355,6 @@ int32_t qwRegisterHbBrokenLinkArg(SQWorkerMgmt *mgmt, uint64_t sId, SQWConnInfo
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) { int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
if (NULL == node || NULL == qWorkerMgmt || NULL == pMsg) { if (NULL == node || NULL == qWorkerMgmt || NULL == pMsg) {
QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT); QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
@ -379,8 +384,9 @@ int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
SQWMsg qwMsg = {.node = node, .msg = msg->msg + msg->sqlLen, .msgLen = msg->phyLen}; SQWMsg qwMsg = {.node = node, .msg = msg->msg + msg->sqlLen, .msgLen = msg->phyLen};
qwMsg.connInfo.handle = pMsg->handle; qwMsg.connInfo.handle = pMsg->handle;
qwMsg.connInfo.ahandle = pMsg->ahandle; qwMsg.connInfo.ahandle = pMsg->ahandle;
qwMsg.connInfo.refId = pMsg->refId;
char* sql = strndup(msg->msg, msg->sqlLen); char *sql = strndup(msg->msg, msg->sqlLen);
QW_SCH_TASK_DLOG("processQuery start, node:%p, handle:%p, sql:%s", node, pMsg->handle, sql); QW_SCH_TASK_DLOG("processQuery start, node:%p, handle:%p, sql:%s", node, pMsg->handle, sql);
taosMemoryFreeClear(sql); taosMemoryFreeClear(sql);
@ -397,8 +403,8 @@ int32_t qWorkerProcessCQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
bool queryDone = false; bool queryDone = false;
SQueryContinueReq *msg = (SQueryContinueReq *)pMsg->pCont; SQueryContinueReq *msg = (SQueryContinueReq *)pMsg->pCont;
bool needStop = false; bool needStop = false;
SQWTaskCtx *handles = NULL; SQWTaskCtx * handles = NULL;
SQWorkerMgmt *mgmt = (SQWorkerMgmt *)qWorkerMgmt; SQWorkerMgmt * mgmt = (SQWorkerMgmt *)qWorkerMgmt;
if (NULL == msg || pMsg->contLen < sizeof(*msg)) { if (NULL == msg || pMsg->contLen < sizeof(*msg)) {
QW_ELOG("invalid cquery msg, msg:%p, msgLen:%d", msg, pMsg->contLen); QW_ELOG("invalid cquery msg, msg:%p, msgLen:%d", msg, pMsg->contLen);
@ -413,6 +419,7 @@ int32_t qWorkerProcessCQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
SQWMsg qwMsg = {.node = node, .msg = NULL, .msgLen = 0}; SQWMsg qwMsg = {.node = node, .msg = NULL, .msgLen = 0};
qwMsg.connInfo.handle = pMsg->handle; qwMsg.connInfo.handle = pMsg->handle;
qwMsg.connInfo.ahandle = pMsg->ahandle; qwMsg.connInfo.ahandle = pMsg->ahandle;
qwMsg.connInfo.refId = pMsg->refId;
QW_SCH_TASK_DLOG("processCQuery start, node:%p, handle:%p", node, pMsg->handle); QW_SCH_TASK_DLOG("processCQuery start, node:%p, handle:%p", node, pMsg->handle);
@ -423,7 +430,7 @@ int32_t qWorkerProcessCQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t qWorkerProcessReadyMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg){ int32_t qWorkerProcessReadyMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
if (NULL == node || NULL == qWorkerMgmt || NULL == pMsg) { if (NULL == node || NULL == qWorkerMgmt || NULL == pMsg) {
return TSDB_CODE_QRY_INVALID_INPUT; return TSDB_CODE_QRY_INVALID_INPUT;
} }
@ -447,6 +454,7 @@ int32_t qWorkerProcessReadyMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg){
SQWMsg qwMsg = {.node = node, .msg = NULL, .msgLen = 0}; SQWMsg qwMsg = {.node = node, .msg = NULL, .msgLen = 0};
qwMsg.connInfo.handle = pMsg->handle; qwMsg.connInfo.handle = pMsg->handle;
qwMsg.connInfo.ahandle = pMsg->ahandle; qwMsg.connInfo.ahandle = pMsg->ahandle;
qwMsg.connInfo.refId = pMsg->refId;
QW_SCH_TASK_DLOG("processReady start, node:%p, handle:%p", node, pMsg->handle); QW_SCH_TASK_DLOG("processReady start, node:%p, handle:%p", node, pMsg->handle);
@ -475,11 +483,11 @@ int32_t qWorkerProcessStatusMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
SSchedulerStatusRsp *sStatus = NULL; SSchedulerStatusRsp *sStatus = NULL;
//QW_ERR_JRET(qwGetSchTasksStatus(qWorkerMgmt, msg->sId, &sStatus)); // QW_ERR_JRET(qwGetSchTasksStatus(qWorkerMgmt, msg->sId, &sStatus));
_return: _return:
//QW_ERR_RET(qwBuildAndSendStatusRsp(pMsg, sStatus)); // QW_ERR_RET(qwBuildAndSendStatusRsp(pMsg, sStatus));
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -509,6 +517,7 @@ int32_t qWorkerProcessFetchMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
SQWMsg qwMsg = {.node = node, .msg = NULL, .msgLen = 0}; SQWMsg qwMsg = {.node = node, .msg = NULL, .msgLen = 0};
qwMsg.connInfo.handle = pMsg->handle; qwMsg.connInfo.handle = pMsg->handle;
qwMsg.connInfo.ahandle = pMsg->ahandle; qwMsg.connInfo.ahandle = pMsg->ahandle;
qwMsg.connInfo.refId = pMsg->refId;
QW_SCH_TASK_DLOG("processFetch start, node:%p, handle:%p", node, pMsg->handle); QW_SCH_TASK_DLOG("processFetch start, node:%p, handle:%p", node, pMsg->handle);
@ -529,7 +538,7 @@ int32_t qWorkerProcessCancelMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
return TSDB_CODE_QRY_INVALID_INPUT; return TSDB_CODE_QRY_INVALID_INPUT;
} }
SQWorkerMgmt *mgmt = (SQWorkerMgmt *)qWorkerMgmt; SQWorkerMgmt * mgmt = (SQWorkerMgmt *)qWorkerMgmt;
int32_t code = 0; int32_t code = 0;
STaskCancelReq *msg = pMsg->pCont; STaskCancelReq *msg = pMsg->pCont;
if (NULL == msg || pMsg->contLen < sizeof(*msg)) { if (NULL == msg || pMsg->contLen < sizeof(*msg)) {
@ -550,8 +559,9 @@ int32_t qWorkerProcessCancelMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
SQWMsg qwMsg = {.node = node, .msg = NULL, .msgLen = 0}; SQWMsg qwMsg = {.node = node, .msg = NULL, .msgLen = 0};
qwMsg.connInfo.handle = pMsg->handle; qwMsg.connInfo.handle = pMsg->handle;
qwMsg.connInfo.ahandle = pMsg->ahandle; qwMsg.connInfo.ahandle = pMsg->ahandle;
qwMsg.connInfo.refId = pMsg->refId;
//QW_ERR_JRET(qwCancelTask(qWorkerMgmt, msg->sId, msg->queryId, msg->taskId)); // QW_ERR_JRET(qwCancelTask(qWorkerMgmt, msg->sId, msg->queryId, msg->taskId));
_return: _return:
@ -588,6 +598,7 @@ int32_t qWorkerProcessDropMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
SQWMsg qwMsg = {.node = node, .msg = NULL, .msgLen = 0, .code = pMsg->code}; SQWMsg qwMsg = {.node = node, .msg = NULL, .msgLen = 0, .code = pMsg->code};
qwMsg.connInfo.handle = pMsg->handle; qwMsg.connInfo.handle = pMsg->handle;
qwMsg.connInfo.ahandle = pMsg->ahandle; qwMsg.connInfo.ahandle = pMsg->ahandle;
qwMsg.connInfo.refId = pMsg->refId;
if (TSDB_CODE_RPC_NETWORK_UNAVAIL == pMsg->code) { if (TSDB_CODE_RPC_NETWORK_UNAVAIL == pMsg->code) {
QW_SCH_TASK_DLOG("receive drop task due to network broken, error:%s", tstrerror(pMsg->code)); QW_SCH_TASK_DLOG("receive drop task due to network broken, error:%s", tstrerror(pMsg->code));
@ -609,7 +620,7 @@ int32_t qWorkerProcessHbMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
int32_t code = 0; int32_t code = 0;
SSchedulerHbReq req = {0}; SSchedulerHbReq req = {0};
SQWorkerMgmt *mgmt = (SQWorkerMgmt *)qWorkerMgmt; SQWorkerMgmt * mgmt = (SQWorkerMgmt *)qWorkerMgmt;
if (NULL == pMsg->pCont) { if (NULL == pMsg->pCont) {
QW_ELOG("invalid hb msg, msg:%p, msgLen:%d", pMsg->pCont, pMsg->contLen); QW_ELOG("invalid hb msg, msg:%p, msgLen:%d", pMsg->pCont, pMsg->contLen);
@ -626,6 +637,7 @@ int32_t qWorkerProcessHbMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
SQWMsg qwMsg = {.node = node, .msg = NULL, .msgLen = 0, .code = pMsg->code}; SQWMsg qwMsg = {.node = node, .msg = NULL, .msgLen = 0, .code = pMsg->code};
qwMsg.connInfo.handle = pMsg->handle; qwMsg.connInfo.handle = pMsg->handle;
qwMsg.connInfo.ahandle = pMsg->ahandle; qwMsg.connInfo.ahandle = pMsg->ahandle;
qwMsg.connInfo.refId = pMsg->refId;
if (TSDB_CODE_RPC_NETWORK_UNAVAIL == pMsg->code) { if (TSDB_CODE_RPC_NETWORK_UNAVAIL == pMsg->code) {
QW_SCH_DLOG("receive Hb msg due to network broken, error:%s", tstrerror(pMsg->code)); QW_SCH_DLOG("receive Hb msg due to network broken, error:%s", tstrerror(pMsg->code));

View File

@ -369,7 +369,7 @@ int32_t schBuildTaskRalation(SSchJob *pJob, SHashObj *planToTask) {
} }
for (int32_t n = 0; n < childNum; ++n) { for (int32_t n = 0; n < childNum; ++n) {
SSubplan *child = (SSubplan *)nodesListGetNode(pPlan->pChildren, n); SSubplan * child = (SSubplan *)nodesListGetNode(pPlan->pChildren, n);
SSchTask **childTask = taosHashGet(planToTask, &child, POINTER_BYTES); SSchTask **childTask = taosHashGet(planToTask, &child, POINTER_BYTES);
if (NULL == childTask || NULL == *childTask) { if (NULL == childTask || NULL == *childTask) {
SCH_TASK_ELOG("subplan children relationship error, level:%d, taskIdx:%d, childIdx:%d", i, m, n); SCH_TASK_ELOG("subplan children relationship error, level:%d, taskIdx:%d, childIdx:%d", i, m, n);
@ -401,7 +401,7 @@ int32_t schBuildTaskRalation(SSchJob *pJob, SHashObj *planToTask) {
} }
for (int32_t n = 0; n < parentNum; ++n) { for (int32_t n = 0; n < parentNum; ++n) {
SSubplan *parent = (SSubplan *)nodesListGetNode(pPlan->pParents, n); SSubplan * parent = (SSubplan *)nodesListGetNode(pPlan->pParents, n);
SSchTask **parentTask = taosHashGet(planToTask, &parent, POINTER_BYTES); SSchTask **parentTask = taosHashGet(planToTask, &parent, POINTER_BYTES);
if (NULL == parentTask || NULL == *parentTask) { if (NULL == parentTask || NULL == *parentTask) {
SCH_TASK_ELOG("subplan parent relationship error, level:%d, taskIdx:%d, childIdx:%d", i, m, n); SCH_TASK_ELOG("subplan parent relationship error, level:%d, taskIdx:%d, childIdx:%d", i, m, n);
@ -491,7 +491,7 @@ int32_t schValidateAndBuildJob(SQueryPlan *pDag, SSchJob *pJob) {
SSchLevel level = {0}; SSchLevel level = {0};
SNodeListNode *plans = NULL; SNodeListNode *plans = NULL;
int32_t taskNum = 0; int32_t taskNum = 0;
SSchLevel *pLevel = NULL; SSchLevel * pLevel = NULL;
level.status = JOB_TASK_STATUS_NOT_START; level.status = JOB_TASK_STATUS_NOT_START;
@ -1267,7 +1267,7 @@ int32_t schUpdateTaskExecNodeHandle(SSchTask *pTask, void *handle, int32_t rspCo
int32_t schHandleCallback(void *param, const SDataBuf *pMsg, int32_t msgType, int32_t rspCode) { int32_t schHandleCallback(void *param, const SDataBuf *pMsg, int32_t msgType, int32_t rspCode) {
int32_t code = 0; int32_t code = 0;
SSchTaskCallbackParam *pParam = (SSchTaskCallbackParam *)param; SSchTaskCallbackParam *pParam = (SSchTaskCallbackParam *)param;
SSchTask *pTask = NULL; SSchTask * pTask = NULL;
SSchJob *pJob = schAcquireJob(pParam->refId); SSchJob *pJob = schAcquireJob(pParam->refId);
if (NULL == pJob) { if (NULL == pJob) {
@ -1617,8 +1617,8 @@ _return:
int32_t schMakeHbRpcCtx(SSchJob *pJob, SSchTask *pTask, SRpcCtx *pCtx) { int32_t schMakeHbRpcCtx(SSchJob *pJob, SSchTask *pTask, SRpcCtx *pCtx) {
int32_t code = 0; int32_t code = 0;
SSchHbCallbackParam *param = NULL; SSchHbCallbackParam *param = NULL;
SMsgSendInfo *pMsgSendInfo = NULL; SMsgSendInfo * pMsgSendInfo = NULL;
SQueryNodeAddr *addr = taosArrayGet(pTask->candidateAddrs, pTask->candidateIdx); SQueryNodeAddr * addr = taosArrayGet(pTask->candidateAddrs, pTask->candidateIdx);
SQueryNodeEpId epId = {0}; SQueryNodeEpId epId = {0};
epId.nodeId = addr->nodeId; epId.nodeId = addr->nodeId;
@ -1759,10 +1759,10 @@ int32_t schCloneHbRpcCtx(SRpcCtx *pSrc, SRpcCtx *pDst) {
} }
SRpcCtxVal dst = {0}; SRpcCtxVal dst = {0};
void *pIter = taosHashIterate(pSrc->args, NULL); void * pIter = taosHashIterate(pSrc->args, NULL);
while (pIter) { while (pIter) {
SRpcCtxVal *pVal = (SRpcCtxVal *)pIter; SRpcCtxVal *pVal = (SRpcCtxVal *)pIter;
int32_t *msgType = taosHashGetKey(pIter, NULL); int32_t * msgType = taosHashGetKey(pIter, NULL);
dst = *pVal; dst = *pVal;
dst.val = NULL; dst.val = NULL;
@ -1916,7 +1916,7 @@ _return:
int32_t schBuildAndSendMsg(SSchJob *pJob, SSchTask *pTask, SQueryNodeAddr *addr, int32_t msgType) { int32_t schBuildAndSendMsg(SSchJob *pJob, SSchTask *pTask, SQueryNodeAddr *addr, int32_t msgType) {
uint32_t msgSize = 0; uint32_t msgSize = 0;
void *msg = NULL; void * msg = NULL;
int32_t code = 0; int32_t code = 0;
bool isCandidateAddr = false; bool isCandidateAddr = false;
bool persistHandle = false; bool persistHandle = false;
@ -2673,7 +2673,7 @@ int32_t schedulerGetTasksStatus(int64_t job, SArray *pSub) {
SSchLevel *pLevel = taosArrayGet(pJob->levels, i); SSchLevel *pLevel = taosArrayGet(pJob->levels, i);
for (int32_t m = 0; m < pLevel->taskNum; ++m) { for (int32_t m = 0; m < pLevel->taskNum; ++m) {
SSchTask *pTask = taosArrayGet(pLevel->subTasks, m); SSchTask * pTask = taosArrayGet(pLevel->subTasks, m);
SQuerySubDesc subDesc = {.tid = pTask->taskId, .status = pTask->status}; SQuerySubDesc subDesc = {.tid = pTask->taskId, .status = pTask->status};
taosArrayPush(pSub, &subDesc); taosArrayPush(pSub, &subDesc);

View File

@ -327,6 +327,10 @@ void transQueueClear(STransQueue* queue);
*/ */
void transQueueDestroy(STransQueue* queue); void transQueueDestroy(STransQueue* queue);
/*
* init global func
*/
void transThreadOnce();
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

View File

@ -16,6 +16,8 @@
#include "transComm.h" #include "transComm.h"
// static TdThreadOnce transModuleInit = PTHREAD_ONCE_INIT;
int transAuthenticateMsg(void* pMsg, int msgLen, void* pAuth, void* pKey) { int transAuthenticateMsg(void* pMsg, int msgLen, void* pAuth, void* pKey) {
T_MD5_CTX context; T_MD5_CTX context;
int ret = -1; int ret = -1;
@ -361,5 +363,10 @@ void transQueueDestroy(STransQueue* queue) {
transQueueClear(queue); transQueueClear(queue);
taosArrayDestroy(queue->q); taosArrayDestroy(queue->q);
} }
// int32_t transGetExHandle() {
// static
//}
// void transThreadOnce() {
// taosThreadOnce(&transModuleInit, );
//}
#endif #endif

View File

@ -17,6 +17,11 @@
#include "transComm.h" #include "transComm.h"
static TdThreadOnce transModuleInit = PTHREAD_ONCE_INIT;
static char* notify = "a";
static int transSrvInst = 0;
typedef struct { typedef struct {
int notifyCount; // int notifyCount; //
int init; // init or not int init; // init or not
@ -45,6 +50,7 @@ typedef struct SSrvConn {
struct sockaddr_in addr; struct sockaddr_in addr;
struct sockaddr_in locaddr; struct sockaddr_in locaddr;
int64_t refId;
int spi; int spi;
char info[64]; char info[64];
char user[TSDB_UNI_LEN]; // user ID for the link char user[TSDB_UNI_LEN]; // user ID for the link
@ -89,7 +95,12 @@ typedef struct SServerObj {
uv_async_t* pAcceptAsync; // just to quit from from accept thread uv_async_t* pAcceptAsync; // just to quit from from accept thread
} SServerObj; } SServerObj;
static const char* notify = "a"; // handle
typedef struct SExHandle {
void* handle;
int64_t refId;
SWorkThrdObj* pThrd;
} SExHandle;
static void uvAllocConnBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf); static void uvAllocConnBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf);
static void uvAllocRecvBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf); static void uvAllocRecvBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf);
@ -130,6 +141,17 @@ static void uvHandleRegister(SSrvMsg* msg, SWorkThrdObj* thrd);
static void (*transAsyncHandle[])(SSrvMsg* msg, SWorkThrdObj* thrd) = {uvHandleResp, uvHandleQuit, uvHandleRelease, static void (*transAsyncHandle[])(SSrvMsg* msg, SWorkThrdObj* thrd) = {uvHandleResp, uvHandleQuit, uvHandleRelease,
uvHandleRegister}; uvHandleRegister};
static int32_t exHandlesMgt;
void uvInitExHandleMgt();
void uvOpenExHandleMgt(int size);
void uvCloseExHandleMgt();
int64_t uvAddExHandle(void* p);
int32_t uvRemoveExHandle(int64_t refId);
int32_t uvReleaseExHandle(int64_t refId);
void uvDestoryExHandle(void* handle);
SExHandle* uvAcquireExHandle(int64_t refId);
static void uvDestroyConn(uv_handle_t* handle); static void uvDestroyConn(uv_handle_t* handle);
// server and worker thread // server and worker thread
@ -168,6 +190,40 @@ static bool addHandleToAcceptloop(void* arg);
uv_loop_close(loop); \ uv_loop_close(loop); \
} while (0); } while (0);
#define ASYNC_ERR_JRET(thrd) \
do { \
if (thrd->quit) { \
tTrace("worker thread already quit, ignore msg"); \
goto _return1; \
} \
} while (0)
#define ASYNC_CHECK_HANDLE(exh1, refId) \
do { \
if (refId > 0) { \
tTrace("server handle step1"); \
SExHandle* exh2 = uvAcquireExHandle(refId); \
if (exh2 == NULL || refId != exh2->refId) { \
tTrace("server handle %p except, may already freed, ignore msg, ref1: %" PRIu64 ", ref2 : %" PRIu64 "", exh1, \
exh1->refId, refId); \
goto _return1; \
} \
} else if (refId == 0) { \
tTrace("server handle step2"); \
SExHandle* exh2 = uvAcquireExHandle(refId); \
if (exh2 == NULL || refId != exh2->refId) { \
tTrace("server handle %p except, may already freed, ignore msg, ref1: %" PRIu64 ", ref2 : %" PRIu64 "", exh1, \
refId, exh2 ? exh2->refId : 0); \
goto _return1; \
} else { \
refId = exh1->refId; \
} \
} else if (refId == -1) { \
tTrace("server handle step3"); \
goto _return2; \
} \
} while (0)
void uvAllocRecvBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) { void uvAllocRecvBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
SSrvConn* conn = handle->data; SSrvConn* conn = handle->data;
SConnBuffer* pBuf = &conn->readBuf; SConnBuffer* pBuf = &conn->readBuf;
@ -233,7 +289,15 @@ static void uvHandleReq(SSrvConn* pConn) {
// 1. server application should not send resp on handle // 1. server application should not send resp on handle
// 2. once send out data, cli conn released to conn pool immediately // 2. once send out data, cli conn released to conn pool immediately
// 3. not mixed with persist // 3. not mixed with persist
transMsg.handle = pConn;
transMsg.handle = (void*)uvAcquireExHandle(pConn->refId);
tTrace("server handle %p conn: %p translated to app, refId: %" PRIu64 "", transMsg.handle, pConn, pConn->refId);
transMsg.refId = pConn->refId;
assert(transMsg.handle != NULL);
if (pHead->noResp == 1) {
transMsg.refId = -1;
}
uvReleaseExHandle(pConn->refId);
STrans* pTransInst = pConn->pTransInst; STrans* pTransInst = pConn->pTransInst;
(*pTransInst->cfp)(pTransInst->parent, &transMsg, NULL); (*pTransInst->cfp)(pTransInst->parent, &transMsg, NULL);
@ -404,6 +468,7 @@ static void destroySmsg(SSrvMsg* smsg) {
taosMemoryFree(smsg); taosMemoryFree(smsg);
} }
static void destroyAllConn(SWorkThrdObj* pThrd) { static void destroyAllConn(SWorkThrdObj* pThrd) {
tTrace("thread %p destroy all conn ", pThrd);
while (!QUEUE_IS_EMPTY(&pThrd->conn)) { while (!QUEUE_IS_EMPTY(&pThrd->conn)) {
queue* h = QUEUE_HEAD(&pThrd->conn); queue* h = QUEUE_HEAD(&pThrd->conn);
QUEUE_REMOVE(h); QUEUE_REMOVE(h);
@ -436,7 +501,26 @@ void uvWorkerAsyncCb(uv_async_t* handle) {
tError("unexcept occurred, continue"); tError("unexcept occurred, continue");
continue; continue;
} }
// release handle to rpc init
if (msg->type == Quit) {
(*transAsyncHandle[msg->type])(msg, pThrd); (*transAsyncHandle[msg->type])(msg, pThrd);
continue;
} else {
STransMsg transMsg = msg->msg;
SExHandle* exh1 = transMsg.handle;
int64_t refId = transMsg.refId;
SExHandle* exh2 = uvAcquireExHandle(refId);
if (exh2 == NULL || exh1 != exh2) {
tTrace("server handle %p except msg, ignore it", exh1);
uvReleaseExHandle(refId);
destroySmsg(msg);
continue;
}
msg->pConn = exh1->handle;
uvReleaseExHandle(refId);
(*transAsyncHandle[msg->type])(msg, pThrd);
}
} }
} }
static void uvWalkCb(uv_handle_t* handle, void* arg) { static void uvWalkCb(uv_handle_t* handle, void* arg) {
@ -658,8 +742,15 @@ static SSrvConn* createConn(void* hThrd) {
pConn->broken = false; pConn->broken = false;
pConn->status = ConnNormal; pConn->status = ConnNormal;
SExHandle* exh = taosMemoryMalloc(sizeof(SExHandle));
exh->handle = pConn;
exh->pThrd = pThrd;
exh->refId = uvAddExHandle(exh);
uvAcquireExHandle(exh->refId);
pConn->refId = exh->refId;
transRefSrvHandle(pConn); transRefSrvHandle(pConn);
tTrace("server conn %p created", pConn); tTrace("server handle %p, conn %p created, refId: %" PRId64 "", exh, pConn, pConn->refId);
return pConn; return pConn;
} }
@ -667,6 +758,7 @@ static void destroyConn(SSrvConn* conn, bool clear) {
if (conn == NULL) { if (conn == NULL) {
return; return;
} }
transDestroyBuffer(&conn->readBuf); transDestroyBuffer(&conn->readBuf);
if (clear) { if (clear) {
tTrace("server conn %p to be destroyed", conn); tTrace("server conn %p to be destroyed", conn);
@ -681,6 +773,9 @@ static void uvDestroyConn(uv_handle_t* handle) {
} }
SWorkThrdObj* thrd = conn->hostThrd; SWorkThrdObj* thrd = conn->hostThrd;
uvReleaseExHandle(conn->refId);
uvRemoveExHandle(conn->refId);
tDebug("server conn %p destroy", conn); tDebug("server conn %p destroy", conn);
// uv_timer_stop(&conn->pTimer); // uv_timer_stop(&conn->pTimer);
transQueueDestroy(&conn->srvMsgs); transQueueDestroy(&conn->srvMsgs);
@ -705,6 +800,10 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
srv->port = port; srv->port = port;
uv_loop_init(srv->loop); uv_loop_init(srv->loop);
taosThreadOnce(&transModuleInit, uvInitExHandleMgt);
transSrvInst++;
// uvOpenExHandleMgt(10000);
for (int i = 0; i < srv->numOfThreads; i++) { for (int i = 0; i < srv->numOfThreads; i++) {
SWorkThrdObj* thrd = (SWorkThrdObj*)taosMemoryCalloc(1, sizeof(SWorkThrdObj)); SWorkThrdObj* thrd = (SWorkThrdObj*)taosMemoryCalloc(1, sizeof(SWorkThrdObj));
thrd->quit = false; thrd->quit = false;
@ -749,6 +848,44 @@ End:
transCloseServer(srv); transCloseServer(srv);
return NULL; return NULL;
} }
void uvInitExHandleMgt() {
// init exhandle mgt
uvOpenExHandleMgt(10000);
}
void uvOpenExHandleMgt(int size) {
// added into once later
exHandlesMgt = taosOpenRef(size, uvDestoryExHandle);
}
void uvCloseExHandleMgt() {
// close ref
taosCloseRef(exHandlesMgt);
}
int64_t uvAddExHandle(void* p) {
// acquire extern handle
return taosAddRef(exHandlesMgt, p);
}
int32_t uvRemoveExHandle(int64_t refId) {
// acquire extern handle
return taosRemoveRef(exHandlesMgt, refId);
}
SExHandle* uvAcquireExHandle(int64_t refId) {
// acquire extern handle
return (SExHandle*)taosAcquireRef(exHandlesMgt, refId);
}
int32_t uvReleaseExHandle(int64_t refId) {
// release extern handle
return taosReleaseRef(exHandlesMgt, refId);
}
void uvDestoryExHandle(void* handle) {
if (handle == NULL) {
return;
}
taosMemoryFree(handle);
}
void uvHandleQuit(SSrvMsg* msg, SWorkThrdObj* thrd) { void uvHandleQuit(SSrvMsg* msg, SWorkThrdObj* thrd) {
thrd->quit = true; thrd->quit = true;
if (QUEUE_IS_EMPTY(&thrd->conn)) { if (QUEUE_IS_EMPTY(&thrd->conn)) {
@ -759,7 +896,6 @@ void uvHandleQuit(SSrvMsg* msg, SWorkThrdObj* thrd) {
taosMemoryFree(msg); taosMemoryFree(msg);
} }
void uvHandleRelease(SSrvMsg* msg, SWorkThrdObj* thrd) { void uvHandleRelease(SSrvMsg* msg, SWorkThrdObj* thrd) {
// release handle to rpc init
SSrvConn* conn = msg->pConn; SSrvConn* conn = msg->pConn;
if (conn->status == ConnAcquire) { if (conn->status == ConnAcquire) {
if (!transQueuePush(&conn->srvMsgs, msg)) { if (!transQueuePush(&conn->srvMsgs, msg)) {
@ -840,6 +976,11 @@ void transCloseServer(void* arg) {
taosMemoryFree(srv->pipe); taosMemoryFree(srv->pipe);
taosMemoryFree(srv); taosMemoryFree(srv);
transSrvInst--;
if (transSrvInst == 0) {
uvCloseExHandleMgt();
}
} }
void transRefSrvHandle(void* handle) { void transRefSrvHandle(void* handle) {
@ -862,57 +1003,98 @@ void transUnrefSrvHandle(void* handle) {
} }
void transReleaseSrvHandle(void* handle) { void transReleaseSrvHandle(void* handle) {
if (handle == NULL) { SExHandle* exh = handle;
return; int64_t refId = exh->refId;
}
SSrvConn* pConn = handle;
SWorkThrdObj* pThrd = pConn->hostThrd;
STransMsg tmsg = {.code = 0, .handle = handle, .ahandle = NULL}; ASYNC_CHECK_HANDLE(exh, refId);
SWorkThrdObj* pThrd = exh->pThrd;
ASYNC_ERR_JRET(pThrd);
STransMsg tmsg = {.code = 0, .handle = exh, .ahandle = NULL, .refId = refId};
SSrvMsg* srvMsg = taosMemoryCalloc(1, sizeof(SSrvMsg)); SSrvMsg* srvMsg = taosMemoryCalloc(1, sizeof(SSrvMsg));
srvMsg->msg = tmsg; srvMsg->msg = tmsg;
srvMsg->type = Release; srvMsg->type = Release;
srvMsg->pConn = pConn;
tTrace("server conn %p start to release", pConn); tTrace("server conn %p start to release", exh->handle);
transSendAsync(pThrd->asyncPool, &srvMsg->q); transSendAsync(pThrd->asyncPool, &srvMsg->q);
uvReleaseExHandle(refId);
return;
_return1:
tTrace("server handle %p failed to send to release handle", exh);
uvReleaseExHandle(refId);
return;
_return2:
tTrace("server handle %p failed to send to release handle", exh);
return;
} }
void transSendResponse(const STransMsg* pMsg) { void transSendResponse(const STransMsg* msg) {
if (pMsg->handle == NULL) { SExHandle* exh = msg->handle;
return; int64_t refId = msg->refId;
} ASYNC_CHECK_HANDLE(exh, refId);
SSrvConn* pConn = pMsg->handle; assert(refId != 0);
SWorkThrdObj* pThrd = pConn->hostThrd;
if (pThrd->quit) { STransMsg tmsg = *msg;
return; tmsg.refId = refId;
}
SWorkThrdObj* pThrd = exh->pThrd;
ASYNC_ERR_JRET(pThrd);
SSrvMsg* srvMsg = taosMemoryCalloc(1, sizeof(SSrvMsg)); SSrvMsg* srvMsg = taosMemoryCalloc(1, sizeof(SSrvMsg));
srvMsg->pConn = pConn; srvMsg->msg = tmsg;
srvMsg->msg = *pMsg;
srvMsg->type = Normal; srvMsg->type = Normal;
tTrace("server conn %p start to send resp (1/2)", pConn); tTrace("server conn %p start to send resp (1/2)", exh->handle);
transSendAsync(pThrd->asyncPool, &srvMsg->q); transSendAsync(pThrd->asyncPool, &srvMsg->q);
uvReleaseExHandle(refId);
return;
_return1:
tTrace("server handle %p failed to send resp", exh);
rpcFreeCont(msg->pCont);
uvReleaseExHandle(refId);
return;
_return2:
tTrace("server handle %p failed to send resp", exh);
rpcFreeCont(msg->pCont);
return;
} }
void transRegisterMsg(const STransMsg* msg) { void transRegisterMsg(const STransMsg* msg) {
if (msg->handle == NULL) { SExHandle* exh = msg->handle;
return; int64_t refId = msg->refId;
} ASYNC_CHECK_HANDLE(exh, refId);
SSrvConn* pConn = msg->handle;
SWorkThrdObj* pThrd = pConn->hostThrd; STransMsg tmsg = *msg;
tmsg.refId = refId;
SWorkThrdObj* pThrd = exh->pThrd;
ASYNC_ERR_JRET(pThrd);
SSrvMsg* srvMsg = taosMemoryCalloc(1, sizeof(SSrvMsg)); SSrvMsg* srvMsg = taosMemoryCalloc(1, sizeof(SSrvMsg));
srvMsg->pConn = pConn; srvMsg->msg = tmsg;
srvMsg->msg = *msg;
srvMsg->type = Register; srvMsg->type = Register;
tTrace("server conn %p start to register brokenlink callback", pConn); tTrace("server conn %p start to register brokenlink callback", exh->handle);
transSendAsync(pThrd->asyncPool, &srvMsg->q); transSendAsync(pThrd->asyncPool, &srvMsg->q);
uvReleaseExHandle(refId);
return;
_return1:
tTrace("server handle %p failed to send to register brokenlink", exh);
rpcFreeCont(msg->pCont);
uvReleaseExHandle(refId);
return;
_return2:
tTrace("server handle %p failed to send to register brokenlink", exh);
rpcFreeCont(msg->pCont);
} }
int transGetConnInfo(void* thandle, STransHandleInfo* pInfo) {
SSrvConn* pConn = thandle;
struct sockaddr_in addr = pConn->addr;
int transGetConnInfo(void* thandle, STransHandleInfo* pInfo) {
if (thandle == NULL) {
tTrace("invalid handle %p, failed to Get Conn info", thandle);
return -1;
}
SExHandle* ex = thandle;
SSrvConn* pConn = ex->handle;
struct sockaddr_in addr = pConn->addr;
pInfo->clientIp = (uint32_t)(addr.sin_addr.s_addr); pInfo->clientIp = (uint32_t)(addr.sin_addr.s_addr);
pInfo->clientPort = ntohs(addr.sin_port); pInfo->clientPort = ntohs(addr.sin_port);
tstrncpy(pInfo->user, pConn->user, sizeof(pInfo->user)); tstrncpy(pInfo->user, pConn->user, sizeof(pInfo->user));