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

Merge pull request #14511 from taosdata/enh/killQuery 

enh: enhance kill query processing
This commit is contained in:
Yihao Deng 2022-07-04 19:39:32 +08:00 committed by GitHub
parent cce9f262aa f213212faf
commit 7ae8787cef
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 537 additions and 212 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/libs/scheduler/scheduler.h
View File

@ -130,7 +130,7 @@ void schedulerStopQueryHb(void *pTrans);
* Free the query job * Free the query job
* @param pJob * @param pJob
*/ */
void schedulerFreeJob(int64_t job, int32_t errCode); void schedulerFreeJob(int64_t* job, int32_t errCode);
void schedulerDestroy(void); void schedulerDestroy(void);

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

@ -110,9 +110,12 @@ typedef struct {
} 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 rpcCloseImpl(void *);
void *rpcMallocCont(int32_t contLen); void *rpcMallocCont(int32_t contLen);
void rpcFreeCont(void *pCont); void rpcFreeCont(void *pCont);
void *rpcReallocCont(void *ptr, int32_t contLen); void *rpcReallocCont(void *ptr, int32_t contLen);

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

@ -284,6 +284,7 @@ static FORCE_INLINE SReqResultInfo* tscGetCurResInfo(TAOS_RES* res) {
extern SAppInfo appInfo; extern SAppInfo appInfo;
extern int32_t clientReqRefPool; extern int32_t clientReqRefPool;
extern int32_t clientConnRefPool; extern int32_t clientConnRefPool;
extern void* tscQhandle;
__async_send_cb_fn_t getMsgRspHandle(int32_t msgType); __async_send_cb_fn_t getMsgRspHandle(int32_t msgType);
@ -337,7 +338,8 @@ int hbHandleRsp(SClientHbBatchRsp* hbRsp);
SAppHbMgr* appHbMgrInit(SAppInstInfo* pAppInstInfo, char* key); SAppHbMgr* appHbMgrInit(SAppInstInfo* pAppInstInfo, char* key);
void appHbMgrCleanup(void); void appHbMgrCleanup(void);
void hbRemoveAppHbMrg(SAppHbMgr** pAppHbMgr); void hbRemoveAppHbMrg(SAppHbMgr** pAppHbMgr);
void closeAllRequests(SHashObj *pRequests); void destroyAllRequests(SHashObj* pRequests);
void stopAllRequests(SHashObj* pRequests);
// conn level // conn level
int hbRegisterConn(SAppHbMgr* pAppHbMgr, int64_t tscRefId, int64_t clusterId, int8_t connType); int hbRegisterConn(SAppHbMgr* pAppHbMgr, int64_t tscRefId, int64_t clusterId, int8_t connType);
@ -356,6 +358,9 @@ int32_t removeMeta(STscObj* pTscObj, SArray* tbList); // todo move to clie
int32_t handleAlterTbExecRes(void* res, struct SCatalog* pCatalog); // todo move to xxx int32_t handleAlterTbExecRes(void* res, struct SCatalog* pCatalog); // todo move to xxx
bool qnodeRequired(SRequestObj* pRequest); bool qnodeRequired(SRequestObj* pRequest);
void initTscQhandle();
void cleanupTscQhandle();
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

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

@ -25,6 +25,7 @@
#include "tmsg.h" #include "tmsg.h"
#include "tref.h" #include "tref.h"
#include "trpc.h" #include "trpc.h"
#include "tsched.h"
#include "ttime.h" #include "ttime.h"
#define TSC_VAR_NOT_RELEASE 1 #define TSC_VAR_NOT_RELEASE 1
@ -34,9 +35,20 @@ SAppInfo appInfo;
int32_t clientReqRefPool = -1; int32_t clientReqRefPool = -1;
int32_t clientConnRefPool = -1; int32_t clientConnRefPool = -1;
void *tscQhandle = NULL;
static TdThreadOnce tscinit = PTHREAD_ONCE_INIT; static TdThreadOnce tscinit = PTHREAD_ONCE_INIT;
volatile int32_t tscInitRes = 0; volatile int32_t tscInitRes = 0;
void initTscQhandle() {
// init handle
tscQhandle = taosInitScheduler(4096, 5, "tsc");
}
void cleanupTscQhandle() {
// destroy handle
taosCleanUpScheduler(tscQhandle);
}
static int32_t registerRequest(SRequestObj *pRequest) { static int32_t registerRequest(SRequestObj *pRequest) {
STscObj *pTscObj = acquireTscObj(pRequest->pTscObj->id); STscObj *pTscObj = acquireTscObj(pRequest->pTscObj->id);
if (NULL == pTscObj) { if (NULL == pTscObj) {
@ -121,12 +133,31 @@ void *openTransporter(const char *user, const char *auth, int32_t numOfThread) {
return pDnodeConn; return pDnodeConn;
} }
void closeAllRequests(SHashObj *pRequests) { void destroyAllRequests(SHashObj *pRequests) {
void *pIter = taosHashIterate(pRequests, NULL); void *pIter = taosHashIterate(pRequests, NULL);
while (pIter != NULL) { while (pIter != NULL) {
int64_t *rid = pIter; int64_t *rid = pIter;
removeRequest(*rid); SRequestObj *pRequest = acquireRequest(*rid);
if (pRequest) {
destroyRequest(pRequest);
releaseRequest(*rid);
}
pIter = taosHashIterate(pRequests, pIter);
}
}
void stopAllRequests(SHashObj *pRequests) {
void *pIter = taosHashIterate(pRequests, NULL);
while (pIter != NULL) {
int64_t *rid = pIter;
SRequestObj *pRequest = acquireRequest(*rid);
if (pRequest) {
taos_stop_query(pRequest);
releaseRequest(*rid);
}
pIter = taosHashIterate(pRequests, pIter); pIter = taosHashIterate(pRequests, pIter);
} }
@ -153,12 +184,18 @@ void destroyAppInst(SAppInstInfo *pAppInfo) {
} }
void destroyTscObj(void *pObj) { void destroyTscObj(void *pObj) {
if (NULL == pObj) {
return;
}
STscObj *pTscObj = pObj; STscObj *pTscObj = pObj;
int64_t tscId = pTscObj->id;
tscTrace("begin to destroy tscObj %" PRIx64 " p:%p", tscId, pTscObj);
SClientHbKey connKey = {.tscRid = pTscObj->id, .connType = pTscObj->connType}; SClientHbKey connKey = {.tscRid = pTscObj->id, .connType = pTscObj->connType};
hbDeregisterConn(pTscObj->pAppInfo->pAppHbMgr, connKey); hbDeregisterConn(pTscObj->pAppInfo->pAppHbMgr, connKey);
int64_t connNum = atomic_sub_fetch_64(&pTscObj->pAppInfo->numOfConns, 1); int64_t connNum = atomic_sub_fetch_64(&pTscObj->pAppInfo->numOfConns, 1);
closeAllRequests(pTscObj->pRequests); destroyAllRequests(pTscObj->pRequests);
schedulerStopQueryHb(pTscObj->pAppInfo->pTransporter); schedulerStopQueryHb(pTscObj->pAppInfo->pTransporter);
tscDebug("connObj 0x%" PRIx64 " p:%p destroyed, remain inst totalConn:%" PRId64, pTscObj->id, pTscObj, tscDebug("connObj 0x%" PRIx64 " p:%p destroyed, remain inst totalConn:%" PRId64, pTscObj->id, pTscObj,
pTscObj->pAppInfo->numOfConns); pTscObj->pAppInfo->numOfConns);
@ -167,7 +204,9 @@ void destroyTscObj(void *pObj) {
destroyAppInst(pTscObj->pAppInfo); destroyAppInst(pTscObj->pAppInfo);
} }
taosThreadMutexDestroy(&pTscObj->mutex); taosThreadMutexDestroy(&pTscObj->mutex);
taosMemoryFreeClear(pTscObj); taosMemoryFree(pTscObj);
tscTrace("end to destroy tscObj %" PRIx64 " p:%p", tscId, pTscObj);
} }
void *createTscObj(const char *user, const char *auth, const char *db, int32_t connType, SAppInstInfo *pAppInfo) { void *createTscObj(const char *user, const char *auth, const char *db, int32_t connType, SAppInstInfo *pAppInfo) {
@ -261,14 +300,18 @@ int32_t releaseRequest(int64_t rid) { return taosReleaseRef(clientReqRefPool, ri
int32_t removeRequest(int64_t rid) { return taosRemoveRef(clientReqRefPool, rid); } int32_t removeRequest(int64_t rid) { return taosRemoveRef(clientReqRefPool, rid); }
void doDestroyRequest(void *p) { void doDestroyRequest(void *p) {
assert(p != NULL); if (NULL == p) {
return;
}
SRequestObj *pRequest = (SRequestObj *)p; SRequestObj *pRequest = (SRequestObj *)p;
int64_t reqId = pRequest->self;
tscTrace("begin to destroy request %" PRIx64 " p:%p", reqId, pRequest);
taosHashRemove(pRequest->pTscObj->pRequests, &pRequest->self, sizeof(pRequest->self)); taosHashRemove(pRequest->pTscObj->pRequests, &pRequest->self, sizeof(pRequest->self));
if (pRequest->body.queryJob != 0) { schedulerFreeJob(&pRequest->body.queryJob, 0);
schedulerFreeJob(pRequest->body.queryJob, 0);
}
taosMemoryFreeClear(pRequest->msgBuf); taosMemoryFreeClear(pRequest->msgBuf);
taosMemoryFreeClear(pRequest->sqlstr); taosMemoryFreeClear(pRequest->sqlstr);
@ -284,7 +327,9 @@ void doDestroyRequest(void *p) {
if (pRequest->self) { if (pRequest->self) {
deregisterRequest(pRequest); deregisterRequest(pRequest);
} }
taosMemoryFreeClear(pRequest); taosMemoryFree(pRequest);
tscTrace("end to destroy request %" PRIx64 " p:%p", reqId, pRequest);
} }
void destroyRequest(SRequestObj *pRequest) { void destroyRequest(SRequestObj *pRequest) {
@ -292,6 +337,8 @@ void destroyRequest(SRequestObj *pRequest) {
return; return;
} }
taos_stop_query(pRequest);
removeRequest(pRequest->self); removeRequest(pRequest->self);
} }
@ -299,7 +346,7 @@ void taos_init_imp(void) {
// In the APIs of other program language, taos_cleanup is not available yet. // In the APIs of other program language, taos_cleanup is not available yet.
// So, to make sure taos_cleanup will be invoked to clean up the allocated resource to suppress the valgrind warning. // So, to make sure taos_cleanup will be invoked to clean up the allocated resource to suppress the valgrind warning.
atexit(taos_cleanup); atexit(taos_cleanup);
initTscQhandle();
errno = TSDB_CODE_SUCCESS; errno = TSDB_CODE_SUCCESS;
taosSeedRand(taosGetTimestampSec()); taosSeedRand(taosGetTimestampSec());

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

@ -25,6 +25,7 @@
#include "tmsgtype.h" #include "tmsgtype.h"
#include "tpagedbuf.h" #include "tpagedbuf.h"
#include "tref.h" #include "tref.h"
#include "tsched.h"
static int32_t initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSet); static int32_t initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSet);
static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest); static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest);
@ -645,9 +646,7 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
pRequest->body.resInfo.execRes = res.res; pRequest->body.resInfo.execRes = res.res;
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
if (pRequest->body.queryJob != 0) { schedulerFreeJob(&pRequest->body.queryJob, 0);
schedulerFreeJob(pRequest->body.queryJob, 0);
}
pRequest->code = code; pRequest->code = code;
terrno = code; terrno = code;
@ -658,9 +657,7 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
TDMT_VND_CREATE_TABLE == pRequest->type) { TDMT_VND_CREATE_TABLE == pRequest->type) {
pRequest->body.resInfo.numOfRows = res.numOfRows; pRequest->body.resInfo.numOfRows = res.numOfRows;
if (pRequest->body.queryJob != 0) { schedulerFreeJob(&pRequest->body.queryJob, 0);
schedulerFreeJob(pRequest->body.queryJob, 0);
}
} }
pRequest->code = res.code; pRequest->code = res.code;
@ -792,10 +789,7 @@ void schedulerExecCb(SQueryResult* pResult, void* param, int32_t code) {
TDMT_VND_CREATE_TABLE == pRequest->type) { TDMT_VND_CREATE_TABLE == pRequest->type) {
pRequest->body.resInfo.numOfRows = pResult->numOfRows; pRequest->body.resInfo.numOfRows = pResult->numOfRows;
if (pRequest->body.queryJob != 0) { schedulerFreeJob(&pRequest->body.queryJob, 0);
schedulerFreeJob(pRequest->body.queryJob, 0);
pRequest->body.queryJob = 0;
}
} }
taosMemoryFree(pResult); taosMemoryFree(pResult);
@ -1239,7 +1233,16 @@ void updateTargetEpSet(SMsgSendInfo* pSendInfo, STscObj* pTscObj, SRpcMsg* pMsg,
} }
} }
void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) { typedef struct SchedArg {
SRpcMsg msg;
SEpSet* pEpset;
} SchedArg;
void doProcessMsgFromServer(SSchedMsg* schedMsg) {
SchedArg* arg = (SchedArg*)schedMsg->ahandle;
SRpcMsg* pMsg = &arg->msg;
SEpSet* pEpSet = arg->pEpset;
SMsgSendInfo* pSendInfo = (SMsgSendInfo*)pMsg->info.ahandle; SMsgSendInfo* pSendInfo = (SMsgSendInfo*)pMsg->info.ahandle;
assert(pMsg->info.ahandle != NULL); assert(pMsg->info.ahandle != NULL);
STscObj* pTscObj = NULL; STscObj* pTscObj = NULL;
@ -1272,7 +1275,8 @@ void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {
updateTargetEpSet(pSendInfo, pTscObj, pMsg, pEpSet); updateTargetEpSet(pSendInfo, pTscObj, pMsg, pEpSet);
SDataBuf buf = {.msgType = pMsg->msgType, .len = pMsg->contLen, .pData = NULL, .handle = pMsg->info.handle, .pEpSet = pEpSet}; SDataBuf buf = {
.msgType = pMsg->msgType, .len = pMsg->contLen, .pData = NULL, .handle = pMsg->info.handle, .pEpSet = pEpSet};
if (pMsg->contLen > 0) { if (pMsg->contLen > 0) {
buf.pData = taosMemoryCalloc(1, pMsg->contLen); buf.pData = taosMemoryCalloc(1, pMsg->contLen);
@ -1287,6 +1291,25 @@ void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {
pSendInfo->fp(pSendInfo->param, &buf, pMsg->code); pSendInfo->fp(pSendInfo->param, &buf, pMsg->code);
rpcFreeCont(pMsg->pCont); rpcFreeCont(pMsg->pCont);
destroySendMsgInfo(pSendInfo); destroySendMsgInfo(pSendInfo);
taosMemoryFree(arg);
}
void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {
SSchedMsg schedMsg = {0};
SEpSet* tEpSet = pEpSet != NULL ? taosMemoryCalloc(1, sizeof(SEpSet)) : NULL;
if (tEpSet != NULL) {
*tEpSet = *pEpSet;
}
SchedArg* arg = taosMemoryCalloc(1, sizeof(SchedArg));
arg->msg = *pMsg;
arg->pEpset = tEpSet;
schedMsg.fp = doProcessMsgFromServer;
schedMsg.ahandle = arg;
taosScheduleTask(tscQhandle, &schedMsg);
} }
TAOS* taos_connect_auth(const char* ip, const char* user, const char* auth, const char* db, uint16_t port) { TAOS* taos_connect_auth(const char* ip, const char* user, const char* auth, const char* db, uint16_t port) {

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

@ -47,11 +47,9 @@ int taos_options(TSDB_OPTION option, const void *arg, ...) {
atomic_store_32(&lock, 0); atomic_store_32(&lock, 0);
return ret; return ret;
} }
// this function may be called by user or system, or by both simultaneously. // this function may be called by user or system, or by both simultaneously.
void taos_cleanup(void) { void taos_cleanup(void) {
tscInfo("start to cleanup client environment"); tscInfo("start to cleanup client environment");
if (atomic_val_compare_exchange_32(&sentinel, TSC_VAR_NOT_RELEASE, TSC_VAR_RELEASED) != TSC_VAR_NOT_RELEASE) { if (atomic_val_compare_exchange_32(&sentinel, TSC_VAR_NOT_RELEASE, TSC_VAR_RELEASED) != TSC_VAR_NOT_RELEASE) {
return; return;
} }
@ -74,8 +72,8 @@ void taos_cleanup(void) {
catalogDestroy(); catalogDestroy();
schedulerDestroy(); schedulerDestroy();
cleanupTscQhandle();
rpcCleanup(); rpcCleanup();
tscInfo("all local resources released"); tscInfo("all local resources released");
taosCleanupCfg(); taosCleanupCfg();
taosCloseLog(); taosCloseLog();
@ -196,10 +194,10 @@ void taos_kill_query(TAOS *taos) {
if (NULL == taos) { if (NULL == taos) {
return; return;
} }
int64_t rid = *(int64_t*)taos;
int64_t rid = *(int64_t *)taos;
STscObj *pTscObj = acquireTscObj(rid); STscObj *pTscObj = acquireTscObj(rid);
closeAllRequests(pTscObj->pRequests); stopAllRequests(pTscObj->pRequests);
releaseTscObj(rid); releaseTscObj(rid);
} }
@ -480,9 +478,7 @@ void taos_stop_query(TAOS_RES *res) {
return; return;
} }
if (pRequest->body.queryJob) { schedulerFreeJob(&pRequest->body.queryJob, TSDB_CODE_TSC_QUERY_KILLED);
schedulerFreeJob(pRequest->body.queryJob, TSDB_CODE_TSC_QUERY_KILLED);
}
tscDebug("request %" PRIx64 " killed", pRequest->requestId); tscDebug("request %" PRIx64 " killed", pRequest->requestId);
} }
@ -722,8 +718,7 @@ int32_t createParseContext(const SRequestObj *pRequest, SParseContext **pCxt) {
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
} }
**pCxt = (SParseContext){ **pCxt = (SParseContext){.requestId = pRequest->requestId,
.requestId = pRequest->requestId,
.requestRid = pRequest->self, .requestRid = pRequest->self,
.acctId = pTscObj->acctId, .acctId = pTscObj->acctId,
.db = pRequest->pDb, .db = pRequest->pDb,
@ -739,8 +734,7 @@ int32_t createParseContext(const SRequestObj *pRequest, SParseContext **pCxt) {
.isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER)), .isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER)),
.async = true, .async = true,
.svrVer = pTscObj->sVer, .svrVer = pTscObj->sVer,
.nodeOffline = (pTscObj->pAppInfo->onlineDnodes < pTscObj->pAppInfo->totalDnodes) .nodeOffline = (pTscObj->pAppInfo->onlineDnodes < pTscObj->pAppInfo->totalDnodes)};
};
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }

23
source/libs/scheduler/src/schJob.c
View File

@ -184,9 +184,7 @@ FORCE_INLINE bool schJobNeedToStop(SSchJob *pJob, int8_t *pStatus) {
} }
if ((*pJob->chkKillFp)(pJob->chkKillParam)) { if ((*pJob->chkKillFp)(pJob->chkKillParam)) {
schUpdateJobStatus(pJob, JOB_TASK_STATUS_DROPPING);
schUpdateJobErrCode(pJob, TSDB_CODE_TSC_QUERY_KILLED); schUpdateJobErrCode(pJob, TSDB_CODE_TSC_QUERY_KILLED);
return true; return true;
} }
@ -811,14 +809,6 @@ int32_t schMoveTaskToExecList(SSchJob *pJob, SSchTask *pTask, bool *moved) {
*/ */
int32_t schTaskCheckSetRetry(SSchJob *pJob, SSchTask *pTask, int32_t errCode, bool *needRetry) { int32_t schTaskCheckSetRetry(SSchJob *pJob, SSchTask *pTask, int32_t errCode, bool *needRetry) {
int8_t status = 0;
if (schJobNeedToStop(pJob, &status)) {
*needRetry = false;
SCH_TASK_DLOG("task no more retry cause of job status, job status:%s", jobTaskStatusStr(status));
return TSDB_CODE_SUCCESS;
}
if (TSDB_CODE_SCH_TIMEOUT_ERROR == errCode) { if (TSDB_CODE_SCH_TIMEOUT_ERROR == errCode) {
pTask->maxExecTimes++; pTask->maxExecTimes++;
if (pTask->timeoutUsec < SCH_MAX_TASK_TIMEOUT_USEC) { if (pTask->timeoutUsec < SCH_MAX_TASK_TIMEOUT_USEC) {
@ -1277,7 +1267,7 @@ int32_t schProcessOnTaskStatusRsp(SQueryNodeEpId* pEpId, SArray* pStatusList) {
for (int32_t i = 0; i < taskNum; ++i) { for (int32_t i = 0; i < taskNum; ++i) {
STaskStatus *taskStatus = taosArrayGet(pStatusList, i); STaskStatus *taskStatus = taosArrayGet(pStatusList, i);
qDebug("QID:%" PRIx64 ",TID:0x%" PRIx64 ",EID:%d task status in server: %s", qDebug("QID:0x%" PRIx64 ",TID:0x%" PRIx64 ",EID:%d task status in server: %s",
taskStatus->queryId, taskStatus->taskId, taskStatus->execId, jobTaskStatusStr(taskStatus->status)); taskStatus->queryId, taskStatus->taskId, taskStatus->execId, jobTaskStatusStr(taskStatus->status));
SSchJob *pJob = schAcquireJob(taskStatus->refId); SSchJob *pJob = schAcquireJob(taskStatus->refId);
@ -1495,6 +1485,8 @@ void schFreeJobImpl(void *job) {
uint64_t queryId = pJob->queryId; uint64_t queryId = pJob->queryId;
int64_t refId = pJob->refId; int64_t refId = pJob->refId;
qDebug("QID:0x%" PRIx64 " begin to free sch job, refId:0x%" PRIx64 ", pointer:%p", queryId, refId, pJob);
if (pJob->status == JOB_TASK_STATUS_EXECUTING) { if (pJob->status == JOB_TASK_STATUS_EXECUTING) {
schCancelJob(pJob); schCancelJob(pJob);
} }
@ -1535,12 +1527,12 @@ void schFreeJobImpl(void *job) {
taosMemoryFreeClear(pJob->resData); taosMemoryFreeClear(pJob->resData);
taosMemoryFree(pJob); taosMemoryFree(pJob);
qDebug("QID:0x%" PRIx64 " sch job freed, refId:0x%" PRIx64 ", pointer:%p", queryId, refId, pJob);
int32_t jobNum = atomic_sub_fetch_32(&schMgmt.jobNum, 1); int32_t jobNum = atomic_sub_fetch_32(&schMgmt.jobNum, 1);
if (jobNum == 0) { if (jobNum == 0) {
schCloseJobRef(); schCloseJobRef();
} }
qDebug("QID:0x%" PRIx64 " sch job freed, refId:0x%" PRIx64 ", pointer:%p", queryId, refId, pJob);
} }
int32_t schLaunchStaticExplainJob(SSchedulerReq *pReq, SSchJob *pJob, bool sync) { int32_t schLaunchStaticExplainJob(SSchedulerReq *pReq, SSchJob *pJob, bool sync) {
@ -1687,11 +1679,6 @@ _return:
int32_t schDoTaskRedirect(SSchJob *pJob, SSchTask *pTask, SDataBuf* pData, int32_t rspCode) { int32_t schDoTaskRedirect(SSchJob *pJob, SSchTask *pTask, SDataBuf* pData, int32_t rspCode) {
int32_t code = 0; int32_t code = 0;
int8_t status = 0;
if (schJobNeedToStop(pJob, &status)) {
SCH_TASK_ELOG("redirect will no continue cause of job status %s", jobTaskStatusStr(status));
SCH_RET(atomic_load_32(&pJob->errCode));
}
if ((pTask->execId + 1) >= pTask->maxExecTimes) { if ((pTask->execId + 1) >= pTask->maxExecTimes) {
SCH_TASK_DLOG("task no more retry since reach max try times, execId:%d", pTask->execId); SCH_TASK_DLOG("task no more retry since reach max try times, execId:%d", pTask->execId);

6
source/libs/scheduler/src/schRemote.c
View File

@ -402,12 +402,16 @@ int32_t schHandleCallback(void *param, SDataBuf *pMsg, int32_t rspCode) {
goto _return; goto _return;
} }
code = schHandleResponseMsg(pJob, pTask, msgType, pMsg->pData, pMsg->len, rspCode); schHandleResponseMsg(pJob, pTask, msgType, pMsg->pData, pMsg->len, rspCode);
pMsg->pData = NULL; pMsg->pData = NULL;
_return: _return:
if (pTask) { if (pTask) {
if (code) {
schProcessOnTaskFailure(pJob, pTask, code);
}
SCH_UNLOCK_TASK(pTask); SCH_UNLOCK_TASK(pTask);
} }

23
source/libs/scheduler/src/scheduler.c
View File

@ -225,26 +225,33 @@ void schedulerStopQueryHb(void *pTrans) {
schCleanClusterHb(pTrans); schCleanClusterHb(pTrans);
} }
void schedulerFreeJob(int64_t job, int32_t errCode) { void schedulerFreeJob(int64_t* job, int32_t errCode) {
SSchJob *pJob = schAcquireJob(job); if (0 == *job) {
return;
}
SSchJob *pJob = schAcquireJob(*job);
if (NULL == pJob) { if (NULL == pJob) {
qError("acquire job from jobRef list failed, may be dropped, jobId:0x%" PRIx64, job); qError("acquire sch job failed, may be dropped, jobId:0x%" PRIx64, *job);
*job = 0;
return; return;
} }
int32_t code = schProcessOnJobDropped(pJob, errCode); int32_t code = schProcessOnJobDropped(pJob, errCode);
if (TSDB_CODE_SCH_JOB_IS_DROPPING == code) { if (TSDB_CODE_SCH_JOB_IS_DROPPING == code) {
SCH_JOB_DLOG("sch job is already dropping, refId:0x%" PRIx64, job); SCH_JOB_DLOG("sch job is already dropping, refId:0x%" PRIx64, *job);
*job = 0;
return; return;
} }
SCH_JOB_DLOG("start to remove job from jobRef list, refId:0x%" PRIx64, job); SCH_JOB_DLOG("start to remove job from jobRef list, refId:0x%" PRIx64, *job);
if (taosRemoveRef(schMgmt.jobRef, job)) { if (taosRemoveRef(schMgmt.jobRef, *job)) {
SCH_JOB_ELOG("remove job from job list failed, refId:0x%" PRIx64, job); SCH_JOB_ELOG("remove job from job list failed, refId:0x%" PRIx64, *job);
} }
schReleaseJob(job); schReleaseJob(*job);
*job = 0;
} }
void schedulerDestroy(void) { void schedulerDestroy(void) {

10
source/libs/scheduler/test/schedulerTests.cpp
View File

@ -457,7 +457,7 @@ void schtFreeQueryJob(int32_t freeThread) {
int64_t job = queryJobRefId; int64_t job = queryJobRefId;
if (job && atomic_val_compare_exchange_64(&queryJobRefId, job, 0)) { if (job && atomic_val_compare_exchange_64(&queryJobRefId, job, 0)) {
schedulerFreeJob(job, 0); schedulerFreeJob(&job, 0);
if (freeThread) { if (freeThread) {
if (++freeNum % schtTestPrintNum == 0) { if (++freeNum % schtTestPrintNum == 0) {
printf("FreeNum:%d\n", freeNum); printf("FreeNum:%d\n", freeNum);
@ -724,7 +724,7 @@ TEST(queryTest, normalCase) {
schReleaseJob(job); schReleaseJob(job);
schedulerFreeJob(job, 0); schedulerFreeJob(&job, 0);
schtFreeQueryDag(&dag); schtFreeQueryDag(&dag);
@ -828,7 +828,7 @@ TEST(queryTest, readyFirstCase) {
schReleaseJob(job); schReleaseJob(job);
schedulerFreeJob(job, 0); schedulerFreeJob(&job, 0);
schtFreeQueryDag(&dag); schtFreeQueryDag(&dag);
@ -940,7 +940,7 @@ TEST(queryTest, flowCtrlCase) {
schReleaseJob(job); schReleaseJob(job);
schedulerFreeJob(job, 0); schedulerFreeJob(&job, 0);
schtFreeQueryDag(&dag); schtFreeQueryDag(&dag);
@ -994,7 +994,7 @@ TEST(insertTest, normalCase) {
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
ASSERT_EQ(res.numOfRows, 20); ASSERT_EQ(res.numOfRows, 20);
schedulerFreeJob(insertJobRefId, 0); schedulerFreeJob(&insertJobRefId, 0);
schedulerDestroy(); schedulerDestroy();
} }

19
source/libs/transport/inc/transComm.h
View File

@ -253,7 +253,7 @@ int transAsyncSend(SAsyncPool* pool, queue* mq);
do { \ do { \
if (id > 0) { \ if (id > 0) { \
tTrace("handle step1"); \ tTrace("handle step1"); \
SExHandle* exh2 = transAcquireExHandle(id); \ SExHandle* exh2 = transAcquireExHandle(transGetRefMgt(), id); \
if (exh2 == NULL || id != exh2->refId) { \ if (exh2 == NULL || id != exh2->refId) { \
tTrace("handle %p except, may already freed, ignore msg, ref1: %" PRIu64 ", ref2 : %" PRIu64 "", exh1, \ tTrace("handle %p except, may already freed, ignore msg, ref1: %" PRIu64 ", ref2 : %" PRIu64 "", exh1, \
exh2 ? exh2->refId : 0, id); \ exh2 ? exh2->refId : 0, id); \
@ -261,7 +261,7 @@ int transAsyncSend(SAsyncPool* pool, queue* mq);
} \ } \
} else if (id == 0) { \ } else if (id == 0) { \
tTrace("handle step2"); \ tTrace("handle step2"); \
SExHandle* exh2 = transAcquireExHandle(id); \ SExHandle* exh2 = transAcquireExHandle(transGetRefMgt(), id); \
if (exh2 == NULL || id == exh2->refId) { \ if (exh2 == NULL || id == exh2->refId) { \
tTrace("handle %p except, may already freed, ignore msg, ref1: %" PRIu64 ", ref2 : %" PRIu64 "", exh1, id, \ tTrace("handle %p except, may already freed, ignore msg, ref1: %" PRIu64 ", ref2 : %" PRIu64 "", exh1, id, \
exh2 ? exh2->refId : 0); \ exh2 ? exh2->refId : 0); \
@ -391,14 +391,17 @@ void transThreadOnce();
void transInit(); void transInit();
void transCleanup(); void transCleanup();
int32_t transOpenExHandleMgt(int size); int32_t transOpenRefMgt(int size, void (*func)(void*));
void transCloseExHandleMgt(); void transCloseRefMgt(int32_t refMgt);
int64_t transAddExHandle(void* p); int64_t transAddExHandle(int32_t refMgt, void* p);
int32_t transRemoveExHandle(int64_t refId); int32_t transRemoveExHandle(int32_t refMgt, int64_t refId);
SExHandle* transAcquireExHandle(int64_t refId); void* transAcquireExHandle(int32_t refMgt, int64_t refId);
int32_t transReleaseExHandle(int64_t refId); int32_t transReleaseExHandle(int32_t refMgt, int64_t refId);
void transDestoryExHandle(void* handle); void transDestoryExHandle(void* handle);
int32_t transGetRefMgt();
int32_t transGetInstMgt();
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

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

@ -57,7 +57,7 @@ typedef struct {
void* parent; void* parent;
void* tcphandle; // returned handle from TCP initialization void* tcphandle; // returned handle from TCP initialization
int32_t refMgt; int64_t refId;
TdThreadMutex mutex; TdThreadMutex mutex;
} SRpcInfo; } SRpcInfo;

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

@ -76,16 +76,23 @@ void* rpcOpen(const SRpcInit* pInit) {
if (pInit->user) { if (pInit->user) {
memcpy(pRpc->user, pInit->user, strlen(pInit->user)); memcpy(pRpc->user, pInit->user, strlen(pInit->user));
} }
return pRpc;
int64_t refId = transAddExHandle(transGetInstMgt(), pRpc);
transAcquireExHandle(transGetInstMgt(), refId);
pRpc->refId = refId;
return (void*)refId;
} }
void rpcClose(void* arg) { void rpcClose(void* arg) {
tInfo("start to close rpc"); tInfo("start to close rpc");
transRemoveExHandle(transGetInstMgt(), (int64_t)arg);
transReleaseExHandle(transGetInstMgt(), (int64_t)arg);
tInfo("finish to close rpc");
return;
}
void rpcCloseImpl(void* arg) {
SRpcInfo* pRpc = (SRpcInfo*)arg; SRpcInfo* pRpc = (SRpcInfo*)arg;
(*taosCloseHandle[pRpc->connType])(pRpc->tcphandle); (*taosCloseHandle[pRpc->connType])(pRpc->tcphandle);
taosMemoryFree(pRpc); taosMemoryFree(pRpc);
tInfo("finish to close rpc");
return;
} }
void* rpcMallocCont(int32_t contLen) { void* rpcMallocCont(int32_t contLen) {
@ -144,7 +151,6 @@ void rpcSendResponse(const SRpcMsg* pMsg) { transSendResponse(pMsg); }
int32_t rpcGetConnInfo(void* thandle, SRpcConnInfo* pInfo) { return 0; } int32_t rpcGetConnInfo(void* thandle, SRpcConnInfo* pInfo) { return 0; }
void rpcRefHandle(void* handle, int8_t type) { void rpcRefHandle(void* handle, int8_t type) {
assert(type == TAOS_CONN_SERVER || type == TAOS_CONN_CLIENT); assert(type == TAOS_CONN_SERVER || type == TAOS_CONN_CLIENT);
(*taosRefHandle[type])(handle); (*taosRefHandle[type])(handle);

47
source/libs/transport/src/transCli.c
View File

@ -47,6 +47,7 @@ typedef struct SCliMsg {
queue q; queue q;
STransMsgType type; STransMsgType type;
int64_t refId;
uint64_t st; uint64_t st;
int sent; //(0: no send, 1: alread sent) int sent; //(0: no send, 1: alread sent)
} SCliMsg; } SCliMsg;
@ -505,13 +506,13 @@ static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port) {
} }
static void allocConnRef(SCliConn* conn, bool update) { static void allocConnRef(SCliConn* conn, bool update) {
if (update) { if (update) {
transRemoveExHandle(conn->refId); transRemoveExHandle(transGetRefMgt(), conn->refId);
conn->refId = -1; conn->refId = -1;
} }
SExHandle* exh = taosMemoryCalloc(1, sizeof(SExHandle)); SExHandle* exh = taosMemoryCalloc(1, sizeof(SExHandle));
exh->handle = conn; exh->handle = conn;
exh->pThrd = conn->hostThrd; exh->pThrd = conn->hostThrd;
exh->refId = transAddExHandle(exh); exh->refId = transAddExHandle(transGetRefMgt(), exh);
conn->refId = exh->refId; conn->refId = exh->refId;
} }
static void addConnToPool(void* pool, SCliConn* conn) { static void addConnToPool(void* pool, SCliConn* conn) {
@ -604,16 +605,14 @@ static void cliDestroyConn(SCliConn* conn, bool clear) {
tTrace("%s conn %p remove from conn pool", CONN_GET_INST_LABEL(conn), conn); tTrace("%s conn %p remove from conn pool", CONN_GET_INST_LABEL(conn), conn);
QUEUE_REMOVE(&conn->conn); QUEUE_REMOVE(&conn->conn);
QUEUE_INIT(&conn->conn); QUEUE_INIT(&conn->conn);
transRemoveExHandle(conn->refId); transRemoveExHandle(transGetRefMgt(), conn->refId);
conn->refId = -1; conn->refId = -1;
if (clear) { if (clear) {
if (!uv_is_closing((uv_handle_t*)conn->stream)) { if (!uv_is_closing((uv_handle_t*)conn->stream)) {
uv_read_stop(conn->stream);
uv_close((uv_handle_t*)conn->stream, cliDestroy); uv_close((uv_handle_t*)conn->stream, cliDestroy);
} }
//} else {
// cliDestroy((uv_handle_t*)conn->stream);
//}
} }
} }
static void cliDestroy(uv_handle_t* handle) { static void cliDestroy(uv_handle_t* handle) {
@ -622,7 +621,7 @@ static void cliDestroy(uv_handle_t* handle) {
} }
SCliConn* conn = handle->data; SCliConn* conn = handle->data;
transRemoveExHandle(conn->refId); transRemoveExHandle(transGetRefMgt(), conn->refId);
taosMemoryFree(conn->ip); taosMemoryFree(conn->ip);
conn->stream->data = NULL; conn->stream->data = NULL;
taosMemoryFree(conn->stream); taosMemoryFree(conn->stream);
@ -638,7 +637,6 @@ static bool cliHandleNoResp(SCliConn* conn) {
SCliMsg* pMsg = transQueueGet(&conn->cliMsgs, 0); SCliMsg* pMsg = transQueueGet(&conn->cliMsgs, 0);
if (REQUEST_NO_RESP(&pMsg->msg)) { if (REQUEST_NO_RESP(&pMsg->msg)) {
transQueuePop(&conn->cliMsgs); transQueuePop(&conn->cliMsgs);
// taosArrayRemove(msgs, 0);
destroyCmsg(pMsg); destroyCmsg(pMsg);
res = true; res = true;
} }
@ -749,7 +747,7 @@ static void cliHandleQuit(SCliMsg* pMsg, SCliThrd* pThrd) {
} }
static void cliHandleRelease(SCliMsg* pMsg, SCliThrd* pThrd) { static void cliHandleRelease(SCliMsg* pMsg, SCliThrd* pThrd) {
int64_t refId = (int64_t)(pMsg->msg.info.handle); int64_t refId = (int64_t)(pMsg->msg.info.handle);
SExHandle* exh = transAcquireExHandle(refId); SExHandle* exh = transAcquireExHandle(transGetRefMgt(), refId);
if (exh == NULL) { if (exh == NULL) {
tDebug("%" PRId64 " already release", refId); tDebug("%" PRId64 " already release", refId);
} }
@ -775,14 +773,14 @@ SCliConn* cliGetConn(SCliMsg* pMsg, SCliThrd* pThrd, bool* ignore) {
SCliConn* conn = NULL; SCliConn* conn = NULL;
int64_t refId = (int64_t)(pMsg->msg.info.handle); int64_t refId = (int64_t)(pMsg->msg.info.handle);
if (refId != 0) { if (refId != 0) {
SExHandle* exh = transAcquireExHandle(refId); SExHandle* exh = transAcquireExHandle(transGetRefMgt(), refId);
if (exh == NULL) { if (exh == NULL) {
*ignore = true; *ignore = true;
destroyCmsg(pMsg); destroyCmsg(pMsg);
return NULL; return NULL;
} else { } else {
conn = exh->handle; conn = exh->handle;
transReleaseExHandle(refId); transReleaseExHandle(transGetRefMgt(), refId);
} }
return conn; return conn;
}; };
@ -984,6 +982,7 @@ void cliSendQuit(SCliThrd* thrd) {
} }
void cliWalkCb(uv_handle_t* handle, void* arg) { void cliWalkCb(uv_handle_t* handle, void* arg) {
if (!uv_is_closing(handle)) { if (!uv_is_closing(handle)) {
uv_read_stop((uv_stream_t*)handle);
uv_close(handle, cliDestroy); uv_close(handle, cliDestroy);
} }
} }
@ -1073,7 +1072,7 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
if (retry) { if (retry) {
pMsg->sent = 0; pMsg->sent = 0;
pCtx->retryCnt += 1; pCtx->retryCnt += 1;
if (code == TSDB_CODE_RPC_NETWORK_UNAVAIL) { if (code == TSDB_CODE_RPC_NETWORK_UNAVAIL || code == TSDB_CODE_RPC_BROKEN_LINK) {
cliCompareAndSwap(&pCtx->retryLimit, TRANS_RETRY_COUNT_LIMIT, EPSET_GET_SIZE(&pCtx->epSet) * 3); cliCompareAndSwap(&pCtx->retryLimit, TRANS_RETRY_COUNT_LIMIT, EPSET_GET_SIZE(&pCtx->epSet) * 3);
if (pCtx->retryCnt < pCtx->retryLimit) { if (pCtx->retryCnt < pCtx->retryLimit) {
transUnrefCliHandle(pConn); transUnrefCliHandle(pConn);
@ -1161,12 +1160,12 @@ void transUnrefCliHandle(void* handle) {
} }
SCliThrd* transGetWorkThrdFromHandle(int64_t handle) { SCliThrd* transGetWorkThrdFromHandle(int64_t handle) {
SCliThrd* pThrd = NULL; SCliThrd* pThrd = NULL;
SExHandle* exh = transAcquireExHandle(handle); SExHandle* exh = transAcquireExHandle(transGetRefMgt(), handle);
if (exh == NULL) { if (exh == NULL) {
return NULL; return NULL;
} }
pThrd = exh->pThrd; pThrd = exh->pThrd;
transReleaseExHandle(handle); transReleaseExHandle(transGetRefMgt(), handle);
return pThrd; return pThrd;
} }
SCliThrd* transGetWorkThrd(STrans* trans, int64_t handle) { SCliThrd* transGetWorkThrd(STrans* trans, int64_t handle) {
@ -1193,10 +1192,13 @@ void transReleaseCliHandle(void* handle) {
} }
void transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransCtx* ctx) { void transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransCtx* ctx) {
STrans* pTransInst = (STrans*)shandle; STrans* pTransInst = (STrans*)transAcquireExHandle(transGetInstMgt(), (int64_t)shandle);
if (pTransInst == NULL) return;
SCliThrd* pThrd = transGetWorkThrd(pTransInst, (int64_t)pReq->info.handle); SCliThrd* pThrd = transGetWorkThrd(pTransInst, (int64_t)pReq->info.handle);
if (pThrd == NULL) { if (pThrd == NULL) {
transFreeMsg(pReq->pCont); transFreeMsg(pReq->pCont);
transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
return; return;
} }
@ -1217,19 +1219,24 @@ void transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STra
cliMsg->msg = *pReq; cliMsg->msg = *pReq;
cliMsg->st = taosGetTimestampUs(); cliMsg->st = taosGetTimestampUs();
cliMsg->type = Normal; cliMsg->type = Normal;
cliMsg->refId = (int64_t)shandle;
STraceId* trace = &pReq->info.traceId; STraceId* trace = &pReq->info.traceId;
tGTrace("%s send request at thread:%08" PRId64 ", dst: %s:%d, app:%p", transLabel(pTransInst), pThrd->pid, tGTrace("%s send request at thread:%08" PRId64 ", dst: %s:%d, app:%p", transLabel(pTransInst), pThrd->pid,
EPSET_GET_INUSE_IP(&pCtx->epSet), EPSET_GET_INUSE_PORT(&pCtx->epSet), pReq->info.ahandle); EPSET_GET_INUSE_IP(&pCtx->epSet), EPSET_GET_INUSE_PORT(&pCtx->epSet), pReq->info.ahandle);
ASSERT(transAsyncSend(pThrd->asyncPool, &(cliMsg->q)) == 0); ASSERT(transAsyncSend(pThrd->asyncPool, &(cliMsg->q)) == 0);
transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
return; return;
} }
void transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransMsg* pRsp) { void transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransMsg* pRsp) {
STrans* pTransInst = (STrans*)shandle; STrans* pTransInst = (STrans*)transAcquireExHandle(transGetInstMgt(), (int64_t)shandle);
if (pTransInst == NULL) return;
SCliThrd* pThrd = transGetWorkThrd(pTransInst, (int64_t)pReq->info.handle); SCliThrd* pThrd = transGetWorkThrd(pTransInst, (int64_t)pReq->info.handle);
if (pThrd == NULL) { if (pThrd == NULL) {
transFreeMsg(pReq->pCont); transFreeMsg(pReq->pCont);
transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
return; return;
} }
tsem_t* sem = taosMemoryCalloc(1, sizeof(tsem_t)); tsem_t* sem = taosMemoryCalloc(1, sizeof(tsem_t));
@ -1250,6 +1257,7 @@ void transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransM
cliMsg->msg = *pReq; cliMsg->msg = *pReq;
cliMsg->st = taosGetTimestampUs(); cliMsg->st = taosGetTimestampUs();
cliMsg->type = Normal; cliMsg->type = Normal;
cliMsg->refId = (int64_t)shandle;
STraceId* trace = &pReq->info.traceId; STraceId* trace = &pReq->info.traceId;
tGTrace("%s send request at thread:%08" PRId64 ", dst: %s:%d, app:%p", transLabel(pTransInst), pThrd->pid, tGTrace("%s send request at thread:%08" PRId64 ", dst: %s:%d, app:%p", transLabel(pTransInst), pThrd->pid,
@ -1259,13 +1267,16 @@ void transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransM
tsem_wait(sem); tsem_wait(sem);
tsem_destroy(sem); tsem_destroy(sem);
taosMemoryFree(sem); taosMemoryFree(sem);
transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
return; return;
} }
/* /*
* *
**/ **/
void transSetDefaultAddr(void* shandle, const char* ip, const char* fqdn) { void transSetDefaultAddr(void* shandle, const char* ip, const char* fqdn) {
STrans* pTransInst = shandle; STrans* pTransInst = (STrans*)transAcquireExHandle(transGetInstMgt(), (int64_t)shandle);
if (pTransInst == NULL) return;
SCvtAddr cvtAddr = {0}; SCvtAddr cvtAddr = {0};
if (ip != NULL && fqdn != NULL) { if (ip != NULL && fqdn != NULL) {
@ -1280,11 +1291,13 @@ void transSetDefaultAddr(void* shandle, const char* ip, const char* fqdn) {
SCliMsg* cliMsg = taosMemoryCalloc(1, sizeof(SCliMsg)); SCliMsg* cliMsg = taosMemoryCalloc(1, sizeof(SCliMsg));
cliMsg->ctx = pCtx; cliMsg->ctx = pCtx;
cliMsg->type = Update; cliMsg->type = Update;
cliMsg->refId = (int64_t)shandle;
SCliThrd* thrd = ((SCliObj*)pTransInst->tcphandle)->pThreadObj[i]; SCliThrd* thrd = ((SCliObj*)pTransInst->tcphandle)->pThreadObj[i];
tDebug("%s update epset at thread:%08" PRId64 "", pTransInst->label, thrd->pid); tDebug("%s update epset at thread:%08" PRId64 "", pTransInst->label, thrd->pid);
transAsyncSend(thrd->asyncPool, &(cliMsg->q)); transAsyncSend(thrd->asyncPool, &(cliMsg->q));
} }
transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
} }
#endif #endif

40
source/libs/transport/src/transComm.c
View File

@ -19,6 +19,7 @@
static TdThreadOnce transModuleInit = PTHREAD_ONCE_INIT; static TdThreadOnce transModuleInit = PTHREAD_ONCE_INIT;
static int32_t refMgt; static int32_t refMgt;
static int32_t instMgt;
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;
@ -479,53 +480,54 @@ bool transEpSetIsEqual(SEpSet* a, SEpSet* b) {
} }
return true; return true;
} }
static int32_t transGetRefMgt() {
//
return refMgt;
}
static void transInitEnv() { static void transInitEnv() {
refMgt = transOpenExHandleMgt(50000); refMgt = transOpenRefMgt(50000, transDestoryExHandle);
instMgt = taosOpenRef(50, rpcCloseImpl);
uv_os_setenv("UV_TCP_SINGLE_ACCEPT", "1"); uv_os_setenv("UV_TCP_SINGLE_ACCEPT", "1");
} }
static void transDestroyEnv() { static void transDestroyEnv() {
// close ref transCloseRefMgt(refMgt);
transCloseExHandleMgt(); transCloseRefMgt(instMgt);
} }
void transInit() { void transInit() {
// init env // init env
taosThreadOnce(&transModuleInit, transInitEnv); taosThreadOnce(&transModuleInit, transInitEnv);
} }
int32_t transGetRefMgt() { return refMgt; }
int32_t transGetInstMgt() { return instMgt; }
void transCleanup() { void transCleanup() {
// clean env // clean env
transDestroyEnv(); transDestroyEnv();
} }
int32_t transOpenExHandleMgt(int size) { int32_t transOpenRefMgt(int size, void (*func)(void*)) {
// added into once later // added into once later
return taosOpenRef(size, transDestoryExHandle); return taosOpenRef(size, func);
} }
void transCloseExHandleMgt() { void transCloseRefMgt(int32_t mgt) {
// close ref // close ref
taosCloseRef(transGetRefMgt()); taosCloseRef(mgt);
} }
int64_t transAddExHandle(void* p) { int64_t transAddExHandle(int32_t refMgt, void* p) {
// acquire extern handle // acquire extern handle
return taosAddRef(transGetRefMgt(), p); return taosAddRef(refMgt, p);
} }
int32_t transRemoveExHandle(int64_t refId) { int32_t transRemoveExHandle(int32_t refMgt, int64_t refId) {
// acquire extern handle // acquire extern handle
return taosRemoveRef(transGetRefMgt(), refId); return taosRemoveRef(refMgt, refId);
} }
SExHandle* transAcquireExHandle(int64_t refId) { void* transAcquireExHandle(int32_t refMgt, int64_t refId) {
// acquire extern handle // acquire extern handle
return (SExHandle*)taosAcquireRef(transGetRefMgt(), refId); return (void*)taosAcquireRef(refMgt, refId);
} }
int32_t transReleaseExHandle(int64_t refId) { int32_t transReleaseExHandle(int32_t refMgt, int64_t refId) {
// release extern handle // release extern handle
return taosReleaseRef(transGetRefMgt(), refId); return taosReleaseRef(refMgt, refId);
} }
void transDestoryExHandle(void* handle) { void transDestoryExHandle(void* handle) {
if (handle == NULL) { if (handle == NULL) {

38
source/libs/transport/src/transSvr.c
View File

@ -261,7 +261,7 @@ static void uvHandleReq(SSvrConn* pConn) {
// 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.info.ahandle = (void*)pHead->ahandle; transMsg.info.ahandle = (void*)pHead->ahandle;
transMsg.info.handle = (void*)transAcquireExHandle(pConn->refId); transMsg.info.handle = (void*)transAcquireExHandle(transGetRefMgt(), pConn->refId);
transMsg.info.refId = pConn->refId; transMsg.info.refId = pConn->refId;
transMsg.info.traceId = pHead->traceId; transMsg.info.traceId = pHead->traceId;
@ -279,7 +279,7 @@ static void uvHandleReq(SSvrConn* pConn) {
pConnInfo->clientPort = ntohs(pConn->addr.sin_port); pConnInfo->clientPort = ntohs(pConn->addr.sin_port);
tstrncpy(pConnInfo->user, pConn->user, sizeof(pConnInfo->user)); tstrncpy(pConnInfo->user, pConn->user, sizeof(pConnInfo->user));
transReleaseExHandle(pConn->refId); transReleaseExHandle(transGetRefMgt(), pConn->refId);
STrans* pTransInst = pConn->pTransInst; STrans* pTransInst = pConn->pTransInst;
(*pTransInst->cfp)(pTransInst->parent, &transMsg, NULL); (*pTransInst->cfp)(pTransInst->parent, &transMsg, NULL);
@ -507,15 +507,15 @@ void uvWorkerAsyncCb(uv_async_t* handle) {
SExHandle* exh1 = transMsg.info.handle; SExHandle* exh1 = transMsg.info.handle;
int64_t refId = transMsg.info.refId; int64_t refId = transMsg.info.refId;
SExHandle* exh2 = transAcquireExHandle(refId); SExHandle* exh2 = transAcquireExHandle(transGetRefMgt(), refId);
if (exh2 == NULL || exh1 != exh2) { if (exh2 == NULL || exh1 != exh2) {
tTrace("handle except msg %p, ignore it", exh1); tTrace("handle except msg %p, ignore it", exh1);
transReleaseExHandle(refId); transReleaseExHandle(transGetRefMgt(), refId);
destroySmsg(msg); destroySmsg(msg);
continue; continue;
} }
msg->pConn = exh1->handle; msg->pConn = exh1->handle;
transReleaseExHandle(refId); transReleaseExHandle(transGetRefMgt(), refId);
(*transAsyncHandle[msg->type])(msg, pThrd); (*transAsyncHandle[msg->type])(msg, pThrd);
} }
} }
@ -757,8 +757,8 @@ static SSvrConn* createConn(void* hThrd) {
SExHandle* exh = taosMemoryMalloc(sizeof(SExHandle)); SExHandle* exh = taosMemoryMalloc(sizeof(SExHandle));
exh->handle = pConn; exh->handle = pConn;
exh->pThrd = pThrd; exh->pThrd = pThrd;
exh->refId = transAddExHandle(exh); exh->refId = transAddExHandle(transGetRefMgt(), exh);
transAcquireExHandle(exh->refId); transAcquireExHandle(transGetRefMgt(), exh->refId);
pConn->refId = exh->refId; pConn->refId = exh->refId;
transRefSrvHandle(pConn); transRefSrvHandle(pConn);
@ -789,14 +789,14 @@ static void destroyConnRegArg(SSvrConn* conn) {
} }
} }
static int reallocConnRef(SSvrConn* conn) { static int reallocConnRef(SSvrConn* conn) {
transReleaseExHandle(conn->refId); transReleaseExHandle(transGetRefMgt(), conn->refId);
transRemoveExHandle(conn->refId); transRemoveExHandle(transGetRefMgt(), conn->refId);
// avoid app continue to send msg on invalid handle // avoid app continue to send msg on invalid handle
SExHandle* exh = taosMemoryMalloc(sizeof(SExHandle)); SExHandle* exh = taosMemoryMalloc(sizeof(SExHandle));
exh->handle = conn; exh->handle = conn;
exh->pThrd = conn->hostThrd; exh->pThrd = conn->hostThrd;
exh->refId = transAddExHandle(exh); exh->refId = transAddExHandle(transGetRefMgt(), exh);
transAcquireExHandle(exh->refId); transAcquireExHandle(transGetRefMgt(), exh->refId);
conn->refId = exh->refId; conn->refId = exh->refId;
return 0; return 0;
@ -808,8 +808,8 @@ static void uvDestroyConn(uv_handle_t* handle) {
} }
SWorkThrd* thrd = conn->hostThrd; SWorkThrd* thrd = conn->hostThrd;
transReleaseExHandle(conn->refId); transReleaseExHandle(transGetRefMgt(), conn->refId);
transRemoveExHandle(conn->refId); transRemoveExHandle(transGetRefMgt(), conn->refId);
tDebug("%s conn %p destroy", transLabel(thrd->pTransInst), conn); tDebug("%s conn %p destroy", transLabel(thrd->pTransInst), conn);
transQueueDestroy(&conn->srvMsgs); transQueueDestroy(&conn->srvMsgs);
@ -1046,11 +1046,11 @@ void transReleaseSrvHandle(void* handle) {
tTrace("%s conn %p start to release", transLabel(pThrd->pTransInst), exh->handle); tTrace("%s conn %p start to release", transLabel(pThrd->pTransInst), exh->handle);
transAsyncSend(pThrd->asyncPool, &m->q); transAsyncSend(pThrd->asyncPool, &m->q);
transReleaseExHandle(refId); transReleaseExHandle(transGetRefMgt(), refId);
return; return;
_return1: _return1:
tTrace("handle %p failed to send to release handle", exh); tTrace("handle %p failed to send to release handle", exh);
transReleaseExHandle(refId); transReleaseExHandle(transGetRefMgt(), refId);
return; return;
_return2: _return2:
tTrace("handle %p failed to send to release handle", exh); tTrace("handle %p failed to send to release handle", exh);
@ -1075,12 +1075,12 @@ void transSendResponse(const STransMsg* msg) {
STraceId* trace = (STraceId*)&msg->info.traceId; STraceId* trace = (STraceId*)&msg->info.traceId;
tGTrace("conn %p start to send resp (1/2)", exh->handle); tGTrace("conn %p start to send resp (1/2)", exh->handle);
transAsyncSend(pThrd->asyncPool, &m->q); transAsyncSend(pThrd->asyncPool, &m->q);
transReleaseExHandle(refId); transReleaseExHandle(transGetRefMgt(), refId);
return; return;
_return1: _return1:
tTrace("handle %p failed to send resp", exh); tTrace("handle %p failed to send resp", exh);
rpcFreeCont(msg->pCont); rpcFreeCont(msg->pCont);
transReleaseExHandle(refId); transReleaseExHandle(transGetRefMgt(), refId);
return; return;
_return2: _return2:
tTrace("handle %p failed to send resp", exh); tTrace("handle %p failed to send resp", exh);
@ -1104,13 +1104,13 @@ void transRegisterMsg(const STransMsg* msg) {
tTrace("%s conn %p start to register brokenlink callback", transLabel(pThrd->pTransInst), exh->handle); tTrace("%s conn %p start to register brokenlink callback", transLabel(pThrd->pTransInst), exh->handle);
transAsyncSend(pThrd->asyncPool, &m->q); transAsyncSend(pThrd->asyncPool, &m->q);
transReleaseExHandle(refId); transReleaseExHandle(transGetRefMgt(), refId);
return; return;
_return1: _return1:
tTrace("handle %p failed to register brokenlink", exh); tTrace("handle %p failed to register brokenlink", exh);
rpcFreeCont(msg->pCont); rpcFreeCont(msg->pCont);
transReleaseExHandle(refId); transReleaseExHandle(transGetRefMgt(), refId);
return; return;
_return2: _return2:
tTrace("handle %p failed to register brokenlink", exh); tTrace("handle %p failed to register brokenlink", exh);

273
tests/script/api/stopquery.c
View File

@ -52,6 +52,7 @@ typedef struct {
typedef struct SSP_CB_PARAM { typedef struct SSP_CB_PARAM {
TAOS *taos; TAOS *taos;
bool fetch; bool fetch;
bool free;
int32_t *end; int32_t *end;
} SSP_CB_PARAM; } SSP_CB_PARAM;
@ -156,7 +157,58 @@ void sqCloseQueryCb(void *param, TAOS_RES *pRes, int code) {
} }
} }
int sqSyncStopQuery(bool fetch) { void sqKillFetchCb(void *param, TAOS_RES *pRes, int numOfRows) {
SSP_CB_PARAM *qParam = (SSP_CB_PARAM *)param;
taos_kill_query(qParam->taos);
*qParam->end = 1;
}
void sqKillQueryCb(void *param, TAOS_RES *pRes, int code) {
SSP_CB_PARAM *qParam = (SSP_CB_PARAM *)param;
if (code == 0 && pRes) {
if (qParam->fetch) {
taos_fetch_rows_a(pRes, sqKillFetchCb, param);
} else {
taos_kill_query(qParam->taos);
*qParam->end = 1;
}
} else {
sqExit("select", taos_errstr(pRes));
}
}
void sqAsyncFetchCb(void *param, TAOS_RES *pRes, int numOfRows) {
SSP_CB_PARAM *qParam = (SSP_CB_PARAM *)param;
if (numOfRows > 0) {
taos_fetch_rows_a(pRes, sqAsyncFetchCb, param);
} else {
*qParam->end = 1;
if (qParam->free) {
taos_free_result(pRes);
}
}
}
void sqAsyncQueryCb(void *param, TAOS_RES *pRes, int code) {
SSP_CB_PARAM *qParam = (SSP_CB_PARAM *)param;
if (code == 0 && pRes) {
if (qParam->fetch) {
taos_fetch_rows_a(pRes, sqAsyncFetchCb, param);
} else {
if (qParam->free) {
taos_free_result(pRes);
}
*qParam->end = 1;
}
} else {
sqExit("select", taos_errstr(pRes));
}
}
int sqStopSyncQuery(bool fetch) {
CASE_ENTER(); CASE_ENTER();
for (int32_t i = 0; i < runTimes; ++i) { for (int32_t i = 0; i < runTimes; ++i) {
char sql[1024] = {0}; char sql[1024] = {0};
@ -189,7 +241,7 @@ int sqSyncStopQuery(bool fetch) {
CASE_LEAVE(); CASE_LEAVE();
} }
int sqAsyncStopQuery(bool fetch) { int sqStopAsyncQuery(bool fetch) {
CASE_ENTER(); CASE_ENTER();
for (int32_t i = 0; i < runTimes; ++i) { for (int32_t i = 0; i < runTimes; ++i) {
char sql[1024] = {0}; char sql[1024] = {0};
@ -219,7 +271,7 @@ int sqAsyncStopQuery(bool fetch) {
CASE_LEAVE(); CASE_LEAVE();
} }
int sqSyncFreeQuery(bool fetch) { int sqFreeSyncQuery(bool fetch) {
CASE_ENTER(); CASE_ENTER();
for (int32_t i = 0; i < runTimes; ++i) { for (int32_t i = 0; i < runTimes; ++i) {
char sql[1024] = {0}; char sql[1024] = {0};
@ -250,7 +302,7 @@ int sqSyncFreeQuery(bool fetch) {
CASE_LEAVE(); CASE_LEAVE();
} }
int sqAsyncFreeQuery(bool fetch) { int sqFreeAsyncQuery(bool fetch) {
CASE_ENTER(); CASE_ENTER();
for (int32_t i = 0; i < runTimes; ++i) { for (int32_t i = 0; i < runTimes; ++i) {
char sql[1024] = {0}; char sql[1024] = {0};
@ -280,7 +332,7 @@ int sqAsyncFreeQuery(bool fetch) {
CASE_LEAVE(); CASE_LEAVE();
} }
int sqSyncCloseQuery(bool fetch) { int sqCloseSyncQuery(bool fetch) {
CASE_ENTER(); CASE_ENTER();
for (int32_t i = 0; i < runTimes; ++i) { for (int32_t i = 0; i < runTimes; ++i) {
char sql[1024] = {0}; char sql[1024] = {0};
@ -310,7 +362,7 @@ int sqSyncCloseQuery(bool fetch) {
CASE_LEAVE(); CASE_LEAVE();
} }
int sqAsyncCloseQuery(bool fetch) { int sqCloseAsyncQuery(bool fetch) {
CASE_ENTER(); CASE_ENTER();
for (int32_t i = 0; i < runTimes; ++i) { for (int32_t i = 0; i < runTimes; ++i) {
char sql[1024] = {0}; char sql[1024] = {0};
@ -360,8 +412,38 @@ void *syncQueryThreadFp(void *arg) {
taos_fetch_row(pRes); taos_fetch_row(pRes);
} }
if (qParam->free) {
taos_free_result(pRes); taos_free_result(pRes);
} }
}
void *asyncQueryThreadFp(void *arg) {
SSP_CB_PARAM* qParam = (SSP_CB_PARAM*)arg;
char sql[1024] = {0};
int32_t code = 0;
TAOS *taos = taos_connect(hostName, "root", "taosdata", NULL, 0);
if (taos == NULL) sqExit("taos_connect", taos_errstr(NULL));
qParam->taos = taos;
sprintf(sql, "reset query cache");
sqExecSQLE(taos, sql);
sprintf(sql, "use %s", dbName);
sqExecSQLE(taos, sql);
sprintf(sql, "select * from %s", tbName);
int32_t qEnd = 0;
SSP_CB_PARAM param = {0};
param.fetch = qParam->fetch;
param.end = &qEnd;
taos_query_a(taos, sql, sqAsyncQueryCb, &param);
while (0 == qEnd) {
usleep(5000);
}
}
void *closeThreadFp(void *arg) { void *closeThreadFp(void *arg) {
SSP_CB_PARAM* qParam = (SSP_CB_PARAM*)arg; SSP_CB_PARAM* qParam = (SSP_CB_PARAM*)arg;
@ -376,7 +458,22 @@ void *closeThreadFp(void *arg) {
} }
int sqConSyncCloseQuery(bool fetch) {
void *killThreadFp(void *arg) {
SSP_CB_PARAM* qParam = (SSP_CB_PARAM*)arg;
while (true) {
if (qParam->taos) {
usleep(rand() % 10000);
taos_kill_query(qParam->taos);
break;
}
usleep(1);
}
}
int sqConCloseSyncQuery(bool fetch) {
CASE_ENTER(); CASE_ENTER();
pthread_t qid, cid; pthread_t qid, cid;
for (int32_t i = 0; i < runTimes; ++i) { for (int32_t i = 0; i < runTimes; ++i) {
@ -391,26 +488,160 @@ int sqConSyncCloseQuery(bool fetch) {
CASE_LEAVE(); CASE_LEAVE();
} }
int sqConCloseAsyncQuery(bool fetch) {
CASE_ENTER();
pthread_t qid, cid;
for (int32_t i = 0; i < runTimes; ++i) {
SSP_CB_PARAM param = {0};
param.fetch = fetch;
pthread_create(&qid, NULL, asyncQueryThreadFp, (void*)&param);
pthread_create(&cid, NULL, closeThreadFp, (void*)&param);
pthread_join(qid, NULL);
pthread_join(cid, NULL);
}
CASE_LEAVE();
}
int sqKillSyncQuery(bool fetch) {
CASE_ENTER();
for (int32_t i = 0; i < runTimes; ++i) {
char sql[1024] = {0};
int32_t code = 0;
TAOS *taos = taos_connect(hostName, "root", "taosdata", NULL, 0);
if (taos == NULL) sqExit("taos_connect", taos_errstr(NULL));
sprintf(sql, "reset query cache");
sqExecSQL(taos, sql);
sprintf(sql, "use %s", dbName);
sqExecSQL(taos, sql);
sprintf(sql, "select * from %s", tbName);
TAOS_RES* pRes = taos_query(taos, sql);
code = taos_errno(pRes);
if (code) {
sqExit("taos_query", taos_errstr(pRes));
}
if (fetch) {
taos_fetch_row(pRes);
}
taos_kill_query(taos);
taos_close(taos);
}
CASE_LEAVE();
}
int sqKillAsyncQuery(bool fetch) {
CASE_ENTER();
for (int32_t i = 0; i < runTimes; ++i) {
char sql[1024] = {0};
int32_t code = 0;
TAOS *taos = taos_connect(hostName, "root", "taosdata", NULL, 0);
if (taos == NULL) sqExit("taos_connect", taos_errstr(NULL));
sprintf(sql, "reset query cache");
sqExecSQL(taos, sql);
sprintf(sql, "use %s", dbName);
sqExecSQL(taos, sql);
sprintf(sql, "select * from %s", tbName);
int32_t qEnd = 0;
SSP_CB_PARAM param = {0};
param.fetch = fetch;
param.end = &qEnd;
param.taos = taos;
taos_query_a(taos, sql, sqKillQueryCb, &param);
while (0 == qEnd) {
usleep(5000);
}
taos_close(taos);
}
CASE_LEAVE();
}
int sqConKillSyncQuery(bool fetch) {
CASE_ENTER();
pthread_t qid, cid;
for (int32_t i = 0; i < runTimes; ++i) {
SSP_CB_PARAM param = {0};
param.fetch = fetch;
pthread_create(&qid, NULL, syncQueryThreadFp, (void*)&param);
pthread_create(&cid, NULL, killThreadFp, (void*)&param);
pthread_join(qid, NULL);
pthread_join(cid, NULL);
}
CASE_LEAVE();
}
int sqConKillAsyncQuery(bool fetch) {
CASE_ENTER();
pthread_t qid, cid;
for (int32_t i = 0; i < runTimes; ++i) {
SSP_CB_PARAM param = {0};
param.fetch = fetch;
pthread_create(&qid, NULL, asyncQueryThreadFp, (void*)&param);
pthread_create(&cid, NULL, killThreadFp, (void*)&param);
pthread_join(qid, NULL);
pthread_join(cid, NULL);
}
CASE_LEAVE();
}
void sqRunAllCase(void) { void sqRunAllCase(void) {
/* /*
sqSyncStopQuery(false); sqStopSyncQuery(false);
sqSyncStopQuery(true); sqStopSyncQuery(true);
sqAsyncStopQuery(false); sqStopAsyncQuery(false);
sqAsyncStopQuery(true); sqStopAsyncQuery(true);
sqSyncFreeQuery(false); sqFreeSyncQuery(false);
sqSyncFreeQuery(true); sqFreeSyncQuery(true);
sqAsyncFreeQuery(false); sqFreeAsyncQuery(false);
sqAsyncFreeQuery(true); sqFreeAsyncQuery(true);
sqSyncCloseQuery(false); sqCloseSyncQuery(false);
sqSyncCloseQuery(true); sqCloseSyncQuery(true);
sqAsyncCloseQuery(false); sqCloseAsyncQuery(false);
sqAsyncCloseQuery(true); sqCloseAsyncQuery(true);
sqConCloseSyncQuery(false);
sqConCloseSyncQuery(true);
sqConCloseAsyncQuery(false);
sqConCloseAsyncQuery(true);
*/ */
sqConSyncCloseQuery(false);
sqConSyncCloseQuery(true);
#if 0
sqKillSyncQuery(false);
sqKillSyncQuery(true);
sqKillAsyncQuery(false);
sqKillAsyncQuery(true);
#endif
//sqConKillSyncQuery(false);
sqConKillSyncQuery(true);
#if 0
sqConKillAsyncQuery(false);
sqConKillAsyncQuery(true);
#endif
int32_t l = 5;
while (l) {
printf("%d\n", l--);
sleep(1);
}
} }