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

test

This commit is contained in:
yihaoDeng 2022-03-10 13:50:40 +08:00
parent aa6e480e58
commit fdb64e7c5d
2 changed files with 18 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -225,7 +225,7 @@ void setResSchemaInfo(SReqResultInfo* pResInfo, const SSchema* pSchema, int32_t
}
int32_t scheduleQuery(SRequestObj* pRequest, SQueryDag* pDag, SArray* pNodeList) {
void* pTransporter = pRequest->pTscObj->pAppInfo->pTransporter;
void* pTransporter = pRequest->pTscObj->pAppInfo->pTransporter;
SQueryResult res = {.code = 0, .numOfRows = 0, .msgSize = ERROR_MSG_BUF_DEFAULT_SIZE, .msg = pRequest->msgBuf};
int32_t code = schedulerExecJob(pTransporter, pNodeList, pDag, &pRequest->body.queryJob, pRequest->sqlstr, &res);
if (code != TSDB_CODE_SUCCESS) {
@ -239,12 +239,12 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryDag* pDag, SArray* pNodeList)
if (TSDB_SQL_INSERT == pRequest->type || TSDB_SQL_CREATE_TABLE == pRequest->type) {
pRequest->body.resInfo.numOfRows = res.numOfRows;
if (pRequest->body.queryJob != 0) {
schedulerFreeJob(pRequest->body.queryJob);
}
}
pRequest->code = res.code;
return pRequest->code;
}

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

@ -157,14 +157,12 @@ static void clientHandleResp(SCliConn* conn) {
conn->secured = pHead->secured;
if (pCtx->pSem == NULL) {
if (pCtx->pSem == NULL) {
tTrace("%s client conn %p handle resp", pTransInst->label, conn);
(pTransInst->cfp)(pTransInst->parent, &rpcMsg, NULL);
} else {
tTrace("%s client conn(sync) %p handle resp", pTransInst->label, conn);
memcpy((char*)pCtx->pRsp, (char*)&rpcMsg, sizeof(rpcMsg));
tsem_post(pCtx->pSem);
}
tTrace("%s client conn %p handle resp", pTransInst->label, conn);
(pTransInst->cfp)(pTransInst->parent, &rpcMsg, NULL);
} else {
tTrace("%s client conn(sync) %p handle resp", pTransInst->label, conn);
memcpy((char*)pCtx->pRsp, (char*)&rpcMsg, sizeof(rpcMsg));
tsem_post(pCtx->pSem);
}
conn->ctnRdCnt += 1;
@ -175,14 +173,14 @@ static void clientHandleResp(SCliConn* conn) {
if (pTransInst->noPool == true) {
destroyCmsg(conn->data);
clientConnDestroy(conn, true);
return;
} else {
addConnToPool(pThrd->pool, pCtx->ip, pCtx->port, conn);
destroyCmsg(conn->data);
conn->data = NULL;
}
} else {
// app decide to free or not
}
destroyCmsg(conn->data);
conn->data = NULL;
// start thread's timer of conn pool if not active
if (!uv_is_active((uv_handle_t*)pThrd->timer) && pTransInst->idleTime > 0) {
// uv_timer_start((uv_timer_t*)pThrd->timer, clientTimeoutCb, CONN_PERSIST_TIME(pRpc->idleTime) / 2, 0);
@ -206,11 +204,16 @@ static void clientHandleExcept(SCliConn* pConn) {
rpcMsg.msgType = pMsg->msg.msgType + 1;
if (pCtx->pSem == NULL) {
tTrace("%s client conn %p handle resp", pTransInst->label, pConn);
(pTransInst->cfp)(pTransInst->parent, &rpcMsg, NULL);
} else {
tTrace("%s client conn(sync) %p handle resp", pTransInst->label, pConn);
memcpy((char*)(pCtx->pRsp), (char*)(&rpcMsg), sizeof(rpcMsg));
tsem_post(pCtx->pSem);
}
destroyCmsg(pConn->data);
pConn->data = NULL;
tTrace("%s client conn %p start to destroy", CONN_GET_INST_LABEL(pConn), pConn);
clientConnDestroy(pConn, true);
}