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

fix: handle except

This commit is contained in:
yihaoDeng 2022-06-23 20:06:22 +08:00
parent d5f5c33c9c
commit 0d904d5aa3
3 changed files with 171 additions and 176 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -13,11 +13,11 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#include "os.h"
#include "catalog.h" #include "catalog.h"
#include "functionMgt.h"
#include "clientInt.h" #include "clientInt.h"
#include "clientLog.h" #include "clientLog.h"
#include "functionMgt.h"
#include "os.h"
#include "query.h" #include "query.h"
#include "scheduler.h" #include "scheduler.h"
#include "tcache.h" #include "tcache.h"
@ -38,7 +38,7 @@ static TdThreadOnce tscinit = PTHREAD_ONCE_INIT;
volatile int32_t tscInitRes = 0; volatile int32_t tscInitRes = 0;
static void registerRequest(SRequestObj *pRequest) { static void registerRequest(SRequestObj *pRequest) {
STscObj *pTscObj = acquireTscObj(*(int64_t*)pRequest->pTscObj->id); STscObj *pTscObj = acquireTscObj(*(int64_t *)pRequest->pTscObj->id);
assert(pTscObj != NULL); assert(pTscObj != NULL);
@ -54,14 +54,14 @@ static void registerRequest(SRequestObj *pRequest) {
int32_t currentInst = atomic_add_fetch_64((int64_t *)&pSummary->currentRequests, 1); int32_t currentInst = atomic_add_fetch_64((int64_t *)&pSummary->currentRequests, 1);
tscDebug("0x%" PRIx64 " new Request from connObj:0x%" PRIx64 tscDebug("0x%" PRIx64 " new Request from connObj:0x%" PRIx64
", current:%d, app current:%d, total:%d, reqId:0x%" PRIx64, ", current:%d, app current:%d, total:%d, reqId:0x%" PRIx64,
pRequest->self, *(int64_t*)pRequest->pTscObj->id, num, currentInst, total, pRequest->requestId); pRequest->self, *(int64_t *)pRequest->pTscObj->id, num, currentInst, total, pRequest->requestId);
} }
} }
static void deregisterRequest(SRequestObj *pRequest) { static void deregisterRequest(SRequestObj *pRequest) {
assert(pRequest != NULL); assert(pRequest != NULL);
STscObj *pTscObj = pRequest->pTscObj; STscObj * pTscObj = pRequest->pTscObj;
SAppClusterSummary *pActivity = &pTscObj->pAppInfo->summary; SAppClusterSummary *pActivity = &pTscObj->pAppInfo->summary;
int32_t currentInst = atomic_sub_fetch_64((int64_t *)&pActivity->currentRequests, 1); int32_t currentInst = atomic_sub_fetch_64((int64_t *)&pActivity->currentRequests, 1);
@ -70,8 +70,8 @@ static void deregisterRequest(SRequestObj *pRequest) {
int64_t duration = taosGetTimestampUs() - pRequest->metric.start; int64_t duration = taosGetTimestampUs() - pRequest->metric.start;
tscDebug("0x%" PRIx64 " free Request from connObj: 0x%" PRIx64 ", reqId:0x%" PRIx64 " elapsed:%" PRIu64 tscDebug("0x%" PRIx64 " free Request from connObj: 0x%" PRIx64 ", reqId:0x%" PRIx64 " elapsed:%" PRIu64
" ms, current:%d, app current:%d", " ms, current:%d, app current:%d",
pRequest->self, *(int64_t*)pTscObj->id, pRequest->requestId, duration / 1000, num, currentInst); pRequest->self, *(int64_t *)pTscObj->id, pRequest->requestId, duration / 1000, num, currentInst);
releaseTscObj(*(int64_t*)pTscObj->id); releaseTscObj(*(int64_t *)pTscObj->id);
} }
// todo close the transporter properly // todo close the transporter properly
@ -80,7 +80,7 @@ void closeTransporter(STscObj *pTscObj) {
return; return;
} }
tscDebug("free transporter:%p in connObj: 0x%" PRIx64, pTscObj->pAppInfo->pTransporter, *(int64_t*)pTscObj->id); tscDebug("free transporter:%p in connObj: 0x%" PRIx64, pTscObj->pAppInfo->pTransporter, *(int64_t *)pTscObj->id);
rpcClose(pTscObj->pAppInfo->pTransporter); rpcClose(pTscObj->pAppInfo->pTransporter);
} }
@ -128,16 +128,17 @@ void closeAllRequests(SHashObj *pRequests) {
void destroyTscObj(void *pObj) { void destroyTscObj(void *pObj) {
STscObj *pTscObj = pObj; STscObj *pTscObj = pObj;
SClientHbKey connKey = {.tscRid = *(int64_t*)pTscObj->id, .connType = pTscObj->connType}; SClientHbKey connKey = {.tscRid = *(int64_t *)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); closeAllRequests(pTscObj->pRequests);
schedulerStopQueryHb(pTscObj->pAppInfo->pTransporter); schedulerStopQueryHb(pTscObj->pAppInfo->pTransporter);
if (0 == connNum) { if (0 == connNum) {
// TODO // TODO
//closeTransporter(pTscObj); closeTransporter(pTscObj);
} }
tscDebug("connObj 0x%" PRIx64 " destroyed, totalConn:%" PRId64, *(int64_t*)pTscObj->id, pTscObj->pAppInfo->numOfConns); tscDebug("connObj 0x%" PRIx64 " destroyed, totalConn:%" PRId64, *(int64_t *)pTscObj->id,
pTscObj->pAppInfo->numOfConns);
taosThreadMutexDestroy(&pTscObj->mutex); taosThreadMutexDestroy(&pTscObj->mutex);
taosMemoryFreeClear(pTscObj); taosMemoryFreeClear(pTscObj);
} }
@ -167,10 +168,10 @@ void *createTscObj(const char *user, const char *auth, const char *db, int32_t c
taosThreadMutexInit(&pObj->mutex, NULL); taosThreadMutexInit(&pObj->mutex, NULL);
pObj->id = taosMemoryMalloc(sizeof(int64_t)); pObj->id = taosMemoryMalloc(sizeof(int64_t));
*(int64_t*)pObj->id = taosAddRef(clientConnRefPool, pObj); *(int64_t *)pObj->id = taosAddRef(clientConnRefPool, pObj);
pObj->schemalessType = 1; pObj->schemalessType = 1;
tscDebug("connObj created, 0x%" PRIx64, *(int64_t*)pObj->id); tscDebug("connObj created, 0x%" PRIx64, *(int64_t *)pObj->id);
return pObj; return pObj;
} }
@ -325,7 +326,7 @@ int taos_options_imp(TSDB_OPTION option, const char *str) {
return 0; return 0;
} }
SConfig *pCfg = taosGetCfg(); SConfig * pCfg = taosGetCfg();
SConfigItem *pItem = NULL; SConfigItem *pItem = NULL;
switch (option) { switch (option) {

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

@ -1,4 +1,4 @@
/* * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com> /** Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
* *
* This program is free software: you can use, redistribute, and/or modify * This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3 * it under the terms of the GNU Affero General Public License, version 3
@ -54,7 +54,7 @@ typedef struct SCliMsg {
int sent; //(0: no send, 1: alread sent) int sent; //(0: no send, 1: alread sent)
} SCliMsg; } SCliMsg;
typedef struct SCliThrdObj { typedef struct SCliThrd {
TdThread thread; // tid TdThread thread; // tid
int64_t pid; // pid int64_t pid; // pid
uv_loop_t* loop; uv_loop_t* loop;
@ -72,13 +72,13 @@ typedef struct SCliThrdObj {
SCvtAddr cvtAddr; SCvtAddr cvtAddr;
bool quit; bool quit;
} SCliThrdObj; } SCliThrd;
typedef struct SCliObj { typedef struct SCliObj {
char label[TSDB_LABEL_LEN]; char label[TSDB_LABEL_LEN];
int32_t index; int32_t index;
int numOfThreads; int numOfThreads;
SCliThrdObj** pThreadObj; SCliThrd** pThreadObj;
} SCliObj; } SCliObj;
typedef struct SConnList { typedef struct SConnList {
@ -106,7 +106,7 @@ static void cliAsyncCb(uv_async_t* handle);
static int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg); static int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg);
static SCliConn* cliCreateConn(SCliThrdObj* thrd); static SCliConn* cliCreateConn(SCliThrd* thrd);
static void cliDestroyConn(SCliConn* pConn, bool clear /*clear tcp handle or not*/); static void cliDestroyConn(SCliConn* pConn, bool clear /*clear tcp handle or not*/);
static void cliDestroy(uv_handle_t* handle); static void cliDestroy(uv_handle_t* handle);
static void cliSend(SCliConn* pConn); static void cliSend(SCliConn* pConn);
@ -122,14 +122,14 @@ static void cliHandleResp(SCliConn* conn);
static void cliHandleExcept(SCliConn* conn); static void cliHandleExcept(SCliConn* conn);
// handle req from app // handle req from app
static void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd); static void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd);
static void cliHandleQuit(SCliMsg* pMsg, SCliThrdObj* pThrd); static void cliHandleQuit(SCliMsg* pMsg, SCliThrd* pThrd);
static void cliHandleRelease(SCliMsg* pMsg, SCliThrdObj* pThrd); static void cliHandleRelease(SCliMsg* pMsg, SCliThrd* pThrd);
static void cliHandleUpdate(SCliMsg* pMsg, SCliThrdObj* pThrd); static void cliHandleUpdate(SCliMsg* pMsg, SCliThrd* pThrd);
static void (*cliAsyncHandle[])(SCliMsg* pMsg, SCliThrdObj* pThrd) = {cliHandleReq, cliHandleQuit, cliHandleRelease, static void (*cliAsyncHandle[])(SCliMsg* pMsg, SCliThrd* pThrd) = {cliHandleReq, cliHandleQuit, cliHandleRelease, NULL,
NULL, cliHandleUpdate}; cliHandleUpdate};
static void cliSendQuit(SCliThrdObj* thrd); static void cliSendQuit(SCliThrd* thrd);
static void destroyUserdata(STransMsg* userdata); static void destroyUserdata(STransMsg* userdata);
static int cliRBChoseIdx(STrans* pTransInst); static int cliRBChoseIdx(STrans* pTransInst);
@ -137,8 +137,8 @@ static int cliRBChoseIdx(STrans* pTransInst);
static void destroyCmsg(SCliMsg* cmsg); static void destroyCmsg(SCliMsg* cmsg);
static void transDestroyConnCtx(STransConnCtx* ctx); static void transDestroyConnCtx(STransConnCtx* ctx);
// thread obj // thread obj
static SCliThrdObj* createThrdObj(); static SCliThrd* createThrdObj();
static void destroyThrdObj(SCliThrdObj* pThrd); static void destroyThrdObj(SCliThrd* pThrd);
static void cliWalkCb(uv_handle_t* handle, void* arg); static void cliWalkCb(uv_handle_t* handle, void* arg);
@ -174,12 +174,12 @@ static void cliReleaseUnfinishedMsg(SCliConn* conn) {
idx = -1; \ idx = -1; \
} else { \ } else { \
ASYNC_CHECK_HANDLE((exh), refId); \ ASYNC_CHECK_HANDLE((exh), refId); \
pThrd = (SCliThrdObj*)(exh)->pThrd; \ pThrd = (SCliThrd*)(exh)->pThrd; \
} \ } \
} while (0) } while (0)
#define CONN_PERSIST_TIME(para) (para * 1000 * 10) #define CONN_PERSIST_TIME(para) (para * 1000 * 10)
#define CONN_GET_HOST_THREAD(conn) (conn ? ((SCliConn*)conn)->hostThrd : NULL) #define CONN_GET_HOST_THREAD(conn) (conn ? ((SCliConn*)conn)->hostThrd : NULL)
#define CONN_GET_INST_LABEL(conn) (((STrans*)(((SCliThrdObj*)(conn)->hostThrd)->pTransInst))->label) #define CONN_GET_INST_LABEL(conn) (((STrans*)(((SCliThrd*)(conn)->hostThrd)->pTransInst))->label)
#define CONN_SHOULD_RELEASE(conn, head) \ #define CONN_SHOULD_RELEASE(conn, head) \
do { \ do { \
if ((head)->release == 1 && (head->msgLen) == sizeof(*head)) { \ if ((head)->release == 1 && (head->msgLen) == sizeof(*head)) { \
@ -195,7 +195,7 @@ static void cliReleaseUnfinishedMsg(SCliConn* conn) {
destroyCmsg(pMsg); \ destroyCmsg(pMsg); \
cliReleaseUnfinishedMsg(conn); \ cliReleaseUnfinishedMsg(conn); \
if (status != ConnInPool) { \ if (status != ConnInPool) { \
addConnToPool(((SCliThrdObj*)conn->hostThrd)->pool, conn); \ addConnToPool(((SCliThrd*)conn->hostThrd)->pool, conn); \
} \ } \
transRemoveExHandle(refMgt, conn->refId); \ transRemoveExHandle(refMgt, conn->refId); \
return; \ return; \
@ -279,8 +279,8 @@ _RETURN:
return false; return false;
} }
void cliHandleResp(SCliConn* conn) { void cliHandleResp(SCliConn* conn) {
SCliThrdObj* pThrd = conn->hostThrd; SCliThrd* pThrd = conn->hostThrd;
STrans* pTransInst = pThrd->pTransInst; STrans* pTransInst = pThrd->pTransInst;
STransMsgHead* pHead = (STransMsgHead*)(conn->readBuf.buf); STransMsgHead* pHead = (STransMsgHead*)(conn->readBuf.buf);
pHead->code = htonl(pHead->code); pHead->code = htonl(pHead->code);
@ -379,9 +379,9 @@ void cliHandleExcept(SCliConn* pConn) {
return; return;
} }
} }
SCliThrdObj* pThrd = pConn->hostThrd; SCliThrd* pThrd = pConn->hostThrd;
STrans* pTransInst = pThrd->pTransInst; STrans* pTransInst = pThrd->pTransInst;
bool once = false; bool once = false;
do { do {
SCliMsg* pMsg = transQueuePop(&pConn->cliMsgs); SCliMsg* pMsg = transQueuePop(&pConn->cliMsgs);
if (pMsg == NULL && once) { if (pMsg == NULL && once) {
@ -424,9 +424,9 @@ void cliHandleExcept(SCliConn* pConn) {
} }
void cliTimeoutCb(uv_timer_t* handle) { void cliTimeoutCb(uv_timer_t* handle) {
SCliThrdObj* pThrd = handle->data; SCliThrd* pThrd = handle->data;
STrans* pTransInst = pThrd->pTransInst; STrans* pTransInst = pThrd->pTransInst;
int64_t currentTime = pThrd->nextTimeout; int64_t currentTime = pThrd->nextTimeout;
tTrace("%s conn timeout, try to remove expire conn from conn pool", pTransInst->label); tTrace("%s conn timeout, try to remove expire conn from conn pool", pTransInst->label);
SConnList* p = taosHashIterate((SHashObj*)pThrd->pool, NULL); SConnList* p = taosHashIterate((SHashObj*)pThrd->pool, NULL);
@ -501,7 +501,7 @@ static void allocConnRef(SCliConn* conn, bool update) {
conn->refId = exh->refId; conn->refId = exh->refId;
} }
static void addConnToPool(void* pool, SCliConn* conn) { static void addConnToPool(void* pool, SCliConn* conn) {
SCliThrdObj* thrd = conn->hostThrd; SCliThrd* thrd = conn->hostThrd;
CONN_HANDLE_THREAD_QUIT(thrd); CONN_HANDLE_THREAD_QUIT(thrd);
allocConnRef(conn, true); allocConnRef(conn, true);
@ -562,7 +562,7 @@ static void cliRecvCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) {
} }
} }
static SCliConn* cliCreateConn(SCliThrdObj* pThrd) { static SCliConn* cliCreateConn(SCliThrd* pThrd) {
SCliConn* conn = taosMemoryCalloc(1, sizeof(SCliConn)); SCliConn* conn = taosMemoryCalloc(1, sizeof(SCliConn));
// read/write stream handle // read/write stream handle
conn->stream = (uv_stream_t*)taosMemoryMalloc(sizeof(uv_tcp_t)); conn->stream = (uv_stream_t*)taosMemoryMalloc(sizeof(uv_tcp_t));
@ -615,7 +615,7 @@ static bool cliHandleNoResp(SCliConn* conn) {
} }
if (res == true) { if (res == true) {
if (cliMaySendCachedMsg(conn) == false) { if (cliMaySendCachedMsg(conn) == false) {
SCliThrdObj* thrd = conn->hostThrd; SCliThrd* thrd = conn->hostThrd;
addConnToPool(thrd->pool, conn); addConnToPool(thrd->pool, conn);
} }
} }
@ -651,8 +651,8 @@ void cliSend(SCliConn* pConn) {
STransConnCtx* pCtx = pCliMsg->ctx; STransConnCtx* pCtx = pCliMsg->ctx;
SCliThrdObj* pThrd = pConn->hostThrd; SCliThrd* pThrd = pConn->hostThrd;
STrans* pTransInst = pThrd->pTransInst; STrans* pTransInst = pThrd->pTransInst;
STransMsg* pMsg = (STransMsg*)(&pCliMsg->msg); STransMsg* pMsg = (STransMsg*)(&pCliMsg->msg);
if (pMsg->pCont == 0) { if (pMsg->pCont == 0) {
@ -709,7 +709,7 @@ void cliConnCb(uv_connect_t* req, int status) {
cliSend(pConn); cliSend(pConn);
} }
static void cliHandleQuit(SCliMsg* pMsg, SCliThrdObj* pThrd) { static void cliHandleQuit(SCliMsg* pMsg, SCliThrd* pThrd) {
tDebug("cli work thread %p start to quit", pThrd); tDebug("cli work thread %p start to quit", pThrd);
destroyCmsg(pMsg); destroyCmsg(pMsg);
destroyConnPool(pThrd->pool); destroyConnPool(pThrd->pool);
@ -720,7 +720,7 @@ static void cliHandleQuit(SCliMsg* pMsg, SCliThrdObj* pThrd) {
// uv_stop(pThrd->loop); // uv_stop(pThrd->loop);
} }
static void cliHandleRelease(SCliMsg* pMsg, SCliThrdObj* pThrd) { static void cliHandleRelease(SCliMsg* pMsg, SCliThrd* pThrd) {
SCliConn* conn = pMsg->msg.info.handle; SCliConn* conn = pMsg->msg.info.handle;
tDebug("%s conn %p start to release to inst", CONN_GET_INST_LABEL(conn), conn); tDebug("%s conn %p start to release to inst", CONN_GET_INST_LABEL(conn), conn);
@ -735,39 +735,30 @@ static void cliHandleRelease(SCliMsg* pMsg, SCliThrdObj* pThrd) {
transUnrefCliHandle(conn); transUnrefCliHandle(conn);
} }
} }
static void cliHandleUpdate(SCliMsg* pMsg, SCliThrdObj* pThrd) { static void cliHandleUpdate(SCliMsg* pMsg, SCliThrd* pThrd) {
STransConnCtx* pCtx = pMsg->ctx; STransConnCtx* pCtx = pMsg->ctx;
pThrd->cvtAddr = pCtx->cvtAddr; pThrd->cvtAddr = pCtx->cvtAddr;
destroyCmsg(pMsg); destroyCmsg(pMsg);
} }
SCliConn* cliGetConn(SCliMsg* pMsg, SCliThrdObj* pThrd) { SCliConn* cliGetConn(SCliMsg* pMsg, SCliThrd* pThrd) {
SCliConn* conn = NULL; SCliConn* conn = NULL;
// SExHandleWrap* exWrap = &pMsg->msg.info.handle; int64_t refId = (int64_t)(pMsg->msg.info.handle);
// if (exWrap != NULL) { if (refId != 0) {
//} SExHandle* exh = transAcquireExHandle(refMgt, refId);
if (exh == NULL) {
// SExHandle* exh = transAcquireExHandle(refMgt, exWrap->refId); assert(0);
// if (exh == NULL) { } else {
// if (pInfo->refId != 0) { conn = exh->handle;
// tTrace("%s conn %p ignore msg", CONN_GET_INST_LABEL(conn), conn); transReleaseExHandle(refMgt, refId);
// assert(0); }
// return NULL; return conn;
// } };
//} else {
// transReleaseExHandle(refMgt, pInfo->refId);
// return exh->handle;
//}
STransConnCtx* pCtx = pMsg->ctx; STransConnCtx* pCtx = pMsg->ctx;
conn = getConnFromPool(pThrd->pool, EPSET_GET_INUSE_IP(&pCtx->epSet), EPSET_GET_INUSE_PORT(&pCtx->epSet)); conn = getConnFromPool(pThrd->pool, EPSET_GET_INUSE_IP(&pCtx->epSet), EPSET_GET_INUSE_PORT(&pCtx->epSet));
if (conn != NULL) { if (conn != NULL) {
SExHandle* exh = taosMemoryCalloc(1, sizeof(SExHandle));
exh->handle = conn;
exh->pThrd = pThrd;
exh->refId = transAddExHandle(refMgt, exh);
conn->refId = exh->refId;
tTrace("%s conn %p get from conn pool", CONN_GET_INST_LABEL(conn), conn); tTrace("%s conn %p get from conn pool", CONN_GET_INST_LABEL(conn), conn);
} else { } else {
tTrace("%s not found conn in conn pool %p", ((STrans*)pThrd->pTransInst)->label, pThrd->pool); tTrace("%s not found conn in conn pool %p", ((STrans*)pThrd->pTransInst)->label, pThrd->pool);
@ -785,7 +776,7 @@ void cliMayCvtFqdnToIp(SEpSet* pEpSet, SCvtAddr* pCvtAddr) {
} }
} }
} }
void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) { void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) {
STransConnCtx* pCtx = pMsg->ctx; STransConnCtx* pCtx = pMsg->ctx;
STrans* pTransInst = pThrd->pTransInst; STrans* pTransInst = pThrd->pTransInst;
@ -833,9 +824,9 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
} }
} }
static void cliAsyncCb(uv_async_t* handle) { static void cliAsyncCb(uv_async_t* handle) {
SAsyncItem* item = handle->data; SAsyncItem* item = handle->data;
SCliThrdObj* pThrd = item->pThrd; SCliThrd* pThrd = item->pThrd;
SCliMsg* pMsg = NULL; SCliMsg* pMsg = NULL;
// batch process to avoid to lock/unlock frequently // batch process to avoid to lock/unlock frequently
queue wq; queue wq;
@ -861,7 +852,7 @@ static void cliAsyncCb(uv_async_t* handle) {
} }
static void* cliWorkThread(void* arg) { static void* cliWorkThread(void* arg) {
SCliThrdObj* pThrd = (SCliThrdObj*)arg; SCliThrd* pThrd = (SCliThrd*)arg;
pThrd->pid = taosGetSelfPthreadId(); pThrd->pid = taosGetSelfPthreadId();
setThreadName("trans-cli-work"); setThreadName("trans-cli-work");
uv_run(pThrd->loop, UV_RUN_DEFAULT); uv_run(pThrd->loop, UV_RUN_DEFAULT);
@ -874,10 +865,10 @@ void* transInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads,
STrans* pTransInst = shandle; STrans* pTransInst = shandle;
memcpy(cli->label, label, strlen(label)); memcpy(cli->label, label, strlen(label));
cli->numOfThreads = numOfThreads; cli->numOfThreads = numOfThreads;
cli->pThreadObj = (SCliThrdObj**)taosMemoryCalloc(cli->numOfThreads, sizeof(SCliThrdObj*)); cli->pThreadObj = (SCliThrd**)taosMemoryCalloc(cli->numOfThreads, sizeof(SCliThrd*));
for (int i = 0; i < cli->numOfThreads; i++) { for (int i = 0; i < cli->numOfThreads; i++) {
SCliThrdObj* pThrd = createThrdObj(); SCliThrd* pThrd = createThrdObj();
pThrd->nextTimeout = taosGetTimestampMs() + CONN_PERSIST_TIME(pTransInst->idleTime); pThrd->nextTimeout = taosGetTimestampMs() + CONN_PERSIST_TIME(pTransInst->idleTime);
pThrd->pTransInst = shandle; pThrd->pTransInst = shandle;
@ -911,8 +902,8 @@ static void destroyCmsg(SCliMsg* pMsg) {
taosMemoryFree(pMsg); taosMemoryFree(pMsg);
} }
static SCliThrdObj* createThrdObj() { static SCliThrd* createThrdObj() {
SCliThrdObj* pThrd = (SCliThrdObj*)taosMemoryCalloc(1, sizeof(SCliThrdObj)); SCliThrd* pThrd = (SCliThrd*)taosMemoryCalloc(1, sizeof(SCliThrd));
QUEUE_INIT(&pThrd->msg); QUEUE_INIT(&pThrd->msg);
taosThreadMutexInit(&pThrd->msgMtx, NULL); taosThreadMutexInit(&pThrd->msgMtx, NULL);
@ -930,7 +921,7 @@ static SCliThrdObj* createThrdObj() {
pThrd->quit = false; pThrd->quit = false;
return pThrd; return pThrd;
} }
static void destroyThrdObj(SCliThrdObj* pThrd) { static void destroyThrdObj(SCliThrd* pThrd) {
if (pThrd == NULL) { if (pThrd == NULL) {
return; return;
} }
@ -951,7 +942,7 @@ static void transDestroyConnCtx(STransConnCtx* ctx) {
taosMemoryFree(ctx); taosMemoryFree(ctx);
} }
void cliSendQuit(SCliThrdObj* thrd) { void cliSendQuit(SCliThrd* thrd) {
// cli can stop gracefully // cli can stop gracefully
SCliMsg* msg = taosMemoryCalloc(1, sizeof(SCliMsg)); SCliMsg* msg = taosMemoryCalloc(1, sizeof(SCliMsg));
msg->type = Quit; msg->type = Quit;
@ -973,15 +964,16 @@ int cliRBChoseIdx(STrans* pTransInst) {
static void doDelayTask(void* param) { static void doDelayTask(void* param) {
STaskArg* arg = param; STaskArg* arg = param;
SCliMsg* pMsg = arg->param1; SCliMsg* pMsg = arg->param1;
SCliThrdObj* pThrd = arg->param2; SCliThrd* pThrd = arg->param2;
cliHandleReq(pMsg, pThrd); cliHandleReq(pMsg, pThrd);
taosMemoryFree(arg); taosMemoryFree(arg);
} }
int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) { int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
SCliThrdObj* pThrd = pConn->hostThrd; SCliThrd* pThrd = pConn->hostThrd;
STrans* pTransInst = pThrd->pTransInst; STrans* pTransInst = pThrd->pTransInst;
if (pMsg == NULL || pMsg->ctx == NULL) { if (pMsg == NULL || pMsg->ctx == NULL) {
tTrace("%s conn %p handle resp", pTransInst->label, pConn); tTrace("%s conn %p handle resp", pTransInst->label, pConn);
@ -995,57 +987,60 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
if (pCtx->retryCount == 0) { if (pCtx->retryCount == 0) {
pCtx->origEpSet = pCtx->epSet; pCtx->origEpSet = pCtx->epSet;
} }
/* /*
* upper layer handle retry if code equal TSDB_CODE_RPC_NETWORK_UNAVAIL * upper layer handle retry if code equal TSDB_CODE_RPC_NETWORK_UNAVAIL
*/ */
tmsg_t msgType = pCtx->msgType; if (CONN_NO_PERSIST_BY_APP(pConn)) {
if ((pTransInst->retry != NULL && pEpSet->numOfEps > 1 && (pTransInst->retry(pResp->code))) || tmsg_t msgType = pCtx->msgType;
(pResp->code == TSDB_CODE_RPC_NETWORK_UNAVAIL || pResp->code == TSDB_CODE_APP_NOT_READY || if ((pTransInst->retry != NULL && pEpSet->numOfEps > 1 && (pTransInst->retry(pResp->code))) ||
pResp->code == TSDB_CODE_NODE_NOT_DEPLOYED || pResp->code == TSDB_CODE_SYN_NOT_LEADER)) { (pResp->code == TSDB_CODE_RPC_NETWORK_UNAVAIL || pResp->code == TSDB_CODE_APP_NOT_READY ||
pMsg->sent = 0; pResp->code == TSDB_CODE_NODE_NOT_DEPLOYED || pResp->code == TSDB_CODE_SYN_NOT_LEADER)) {
tTrace("try to send req to next node"); pMsg->sent = 0;
pMsg->st = taosGetTimestampUs(); tTrace("try to send req to next node");
pMsg->st = taosGetTimestampUs();
pCtx->retryCount += 1; pCtx->retryCount += 1;
if (pResp->code == TSDB_CODE_RPC_NETWORK_UNAVAIL) { if (pResp->code == TSDB_CODE_RPC_NETWORK_UNAVAIL) {
if (pCtx->retryCount < pEpSet->numOfEps * 3) { if (pCtx->retryCount < pEpSet->numOfEps * 3) {
pEpSet->inUse = (++pEpSet->inUse) % pEpSet->numOfEps; pEpSet->inUse = (++pEpSet->inUse) % pEpSet->numOfEps;
STaskArg* arg = taosMemoryMalloc(sizeof(STaskArg));
arg->param1 = pMsg;
arg->param2 = pThrd;
transDQSched(pThrd->delayQueue, doDelayTask, arg, TRANS_RETRY_INTERVAL);
transPrintEpSet(pEpSet);
tTrace("%s use local epset, inUse: %d, retry count:%d, limit: %d", pTransInst->label, pEpSet->inUse,
pCtx->retryCount + 1, pEpSet->numOfEps * 3);
transUnrefCliHandle(pConn);
return -1;
}
} else if (pCtx->retryCount < TRANS_RETRY_COUNT_LIMIT) {
if (pResp->contLen == 0) {
pEpSet->inUse = (++pEpSet->inUse) % pEpSet->numOfEps;
transPrintEpSet(&pCtx->epSet);
tTrace("%s use local epset, inUse: %d, retry count:%d, limit: %d", pTransInst->label, pEpSet->inUse,
pCtx->retryCount + 1, TRANS_RETRY_COUNT_LIMIT);
} else {
SEpSet epSet = {0};
tDeserializeSEpSet(pResp->pCont, pResp->contLen, &epSet);
pCtx->epSet = epSet;
transPrintEpSet(&pCtx->epSet);
tTrace("%s use remote epset, inUse: %d, retry count:%d, limit: %d", pTransInst->label, pEpSet->inUse,
pCtx->retryCount + 1, TRANS_RETRY_COUNT_LIMIT);
}
if (pConn->status != ConnInPool) {
addConnToPool(pThrd->pool, pConn);
}
STaskArg* arg = taosMemoryMalloc(sizeof(STaskArg)); STaskArg* arg = taosMemoryMalloc(sizeof(STaskArg));
arg->param1 = pMsg; arg->param1 = pMsg;
arg->param2 = pThrd; arg->param2 = pThrd;
transDQSched(pThrd->delayQueue, doDelayTask, arg, TRANS_RETRY_INTERVAL); transDQSched(pThrd->delayQueue, doDelayTask, arg, TRANS_RETRY_INTERVAL);
transPrintEpSet(pEpSet);
tTrace("%s use local epset, inUse: %d, retry count:%d, limit: %d", pTransInst->label, pEpSet->inUse,
pCtx->retryCount + 1, pEpSet->numOfEps * 3);
transUnrefCliHandle(pConn);
return -1; return -1;
} }
} else if (pCtx->retryCount < TRANS_RETRY_COUNT_LIMIT) {
if (pResp->contLen == 0) {
pEpSet->inUse = (++pEpSet->inUse) % pEpSet->numOfEps;
transPrintEpSet(&pCtx->epSet);
tTrace("%s use local epset, inUse: %d, retry count:%d, limit: %d", pTransInst->label, pEpSet->inUse,
pCtx->retryCount + 1, TRANS_RETRY_COUNT_LIMIT);
} else {
SEpSet epSet = {0};
tDeserializeSEpSet(pResp->pCont, pResp->contLen, &epSet);
pCtx->epSet = epSet;
transPrintEpSet(&pCtx->epSet);
tTrace("%s use remote epset, inUse: %d, retry count:%d, limit: %d", pTransInst->label, pEpSet->inUse,
pCtx->retryCount + 1, TRANS_RETRY_COUNT_LIMIT);
}
if (pConn->status != ConnInPool) {
addConnToPool(pThrd->pool, pConn);
}
STaskArg* arg = taosMemoryMalloc(sizeof(STaskArg));
arg->param1 = pMsg;
arg->param2 = pThrd;
transDQSched(pThrd->delayQueue, doDelayTask, arg, TRANS_RETRY_INTERVAL);
return -1;
} }
} }
@ -1101,9 +1096,9 @@ void transUnrefCliHandle(void* handle) {
cliDestroyConn((SCliConn*)handle, true); cliDestroyConn((SCliConn*)handle, true);
} }
} }
SCliThrdObj* transGetWorkThrdFromHandle(int64_t handle) { SCliThrd* transGetWorkThrdFromHandle(int64_t handle) {
SCliThrdObj* pThrd = NULL; SCliThrd* pThrd = NULL;
SExHandle* exh = transAcquireExHandle(refMgt, handle); SExHandle* exh = transAcquireExHandle(refMgt, handle);
if (exh == NULL) { if (exh == NULL) {
return NULL; return NULL;
} }
@ -1111,17 +1106,16 @@ SCliThrdObj* transGetWorkThrdFromHandle(int64_t handle) {
transReleaseExHandle(refMgt, handle); transReleaseExHandle(refMgt, handle);
return pThrd; return pThrd;
} }
SCliThrdObj* transGetWorkThrd(STrans* trans, int64_t handle) { SCliThrd* transGetWorkThrd(STrans* trans, int64_t handle) {
int idx = -1;
if (handle == 0) { if (handle == 0) {
idx = cliRBChoseIdx(trans); int idx = cliRBChoseIdx(trans);
return ((SCliObj*)trans->tcphandle)->pThreadObj[idx]; return ((SCliObj*)trans->tcphandle)->pThreadObj[idx];
} }
return transGetWorkThrdFromHandle(handle); return transGetWorkThrdFromHandle(handle);
} }
void transReleaseCliHandle(void* handle) { void transReleaseCliHandle(void* handle) {
int idx = -1; int idx = -1;
SCliThrdObj* pThrd = transGetWorkThrdFromHandle((int64_t)handle); SCliThrd* pThrd = transGetWorkThrdFromHandle((int64_t)handle);
if (pThrd == NULL) { if (pThrd == NULL) {
return; return;
} }
@ -1136,8 +1130,8 @@ 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*)shandle;
SCliThrdObj* 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);
return; return;
@ -1169,8 +1163,8 @@ void transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STra
} }
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*)shandle;
SCliThrdObj* 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);
return; return;
@ -1224,7 +1218,7 @@ void transSetDefaultAddr(void* shandle, const char* ip, const char* fqdn) {
cliMsg->ctx = pCtx; cliMsg->ctx = pCtx;
cliMsg->type = Update; cliMsg->type = Update;
SCliThrdObj* 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);
transSendAsync(thrd->asyncPool, &(cliMsg->q)); transSendAsync(thrd->asyncPool, &(cliMsg->q));

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

@ -65,7 +65,7 @@ typedef struct SSvrMsg {
STransMsgType type; STransMsgType type;
} SSvrMsg; } SSvrMsg;
typedef struct SWorkThrdObj { typedef struct SWorkThrd {
TdThread thread; TdThread thread;
uv_connect_t connect_req; uv_connect_t connect_req;
uv_pipe_t* pipe; uv_pipe_t* pipe;
@ -78,7 +78,7 @@ typedef struct SWorkThrdObj {
queue conn; queue conn;
void* pTransInst; void* pTransInst;
bool quit; bool quit;
} SWorkThrdObj; } SWorkThrd;
typedef struct SServerObj { typedef struct SServerObj {
TdThread thread; TdThread thread;
@ -86,10 +86,10 @@ typedef struct SServerObj {
uv_loop_t* loop; uv_loop_t* loop;
// work thread info // work thread info
int workerIdx; int workerIdx;
int numOfThreads; int numOfThreads;
int numOfWorkerReady; int numOfWorkerReady;
SWorkThrdObj** pThreadObj; SWorkThrd** pThreadObj;
uv_pipe_t pipeListen; uv_pipe_t pipeListen;
uv_pipe_t** pipe; uv_pipe_t** pipe;
@ -135,12 +135,12 @@ static void destroyConnRegArg(SSvrConn* conn);
static int reallocConnRef(SSvrConn* conn); static int reallocConnRef(SSvrConn* conn);
static void uvHandleQuit(SSvrMsg* msg, SWorkThrdObj* thrd); static void uvHandleQuit(SSvrMsg* msg, SWorkThrd* thrd);
static void uvHandleRelease(SSvrMsg* msg, SWorkThrdObj* thrd); static void uvHandleRelease(SSvrMsg* msg, SWorkThrd* thrd);
static void uvHandleResp(SSvrMsg* msg, SWorkThrdObj* thrd); static void uvHandleResp(SSvrMsg* msg, SWorkThrd* thrd);
static void uvHandleRegister(SSvrMsg* msg, SWorkThrdObj* thrd); static void uvHandleRegister(SSvrMsg* msg, SWorkThrd* thrd);
static void (*transAsyncHandle[])(SSvrMsg* msg, SWorkThrdObj* thrd) = {uvHandleResp, uvHandleQuit, uvHandleRelease, static void (*transAsyncHandle[])(SSvrMsg* msg, SWorkThrd* thrd) = {uvHandleResp, uvHandleQuit, uvHandleRelease,
uvHandleRegister, NULL}; uvHandleRegister, NULL};
static int32_t exHandlesMgt; static int32_t exHandlesMgt;
@ -160,7 +160,7 @@ static void* transWorkerThread(void* arg);
static void* transAcceptThread(void* arg); static void* transAcceptThread(void* arg);
// add handle loop // add handle loop
static bool addHandleToWorkloop(SWorkThrdObj* pThrd, char* pipeName); static bool addHandleToWorkloop(SWorkThrd* pThrd, char* pipeName);
static bool addHandleToAcceptloop(void* arg); static bool addHandleToAcceptloop(void* arg);
#define CONN_SHOULD_RELEASE(conn, head) \ #define CONN_SHOULD_RELEASE(conn, head) \
@ -233,7 +233,7 @@ static void uvHandleReq(SSvrConn* pConn) {
// wreq->data = pConn; // wreq->data = pConn;
// uv_read_stop((uv_stream_t*)pConn->pTcp); // uv_read_stop((uv_stream_t*)pConn->pTcp);
// transRefSrvHandle(pConn); // transRefSrvHandle(pConn);
// uv_queue_work(((SWorkThrdObj*)pConn->hostThrd)->loop, wreq, uvWorkDoTask, uvWorkAfterTask); // uv_queue_work(((SWorkThrd*)pConn->hostThrd)->loop, wreq, uvWorkDoTask, uvWorkAfterTask);
CONN_SHOULD_RELEASE(pConn, pHead); CONN_SHOULD_RELEASE(pConn, pHead);
@ -478,7 +478,7 @@ static void destroySmsg(SSvrMsg* smsg) {
transFreeMsg(smsg->msg.pCont); transFreeMsg(smsg->msg.pCont);
taosMemoryFree(smsg); taosMemoryFree(smsg);
} }
static void destroyAllConn(SWorkThrdObj* pThrd) { static void destroyAllConn(SWorkThrd* pThrd) {
tTrace("thread %p destroy all conn ", 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);
@ -493,10 +493,10 @@ static void destroyAllConn(SWorkThrdObj* pThrd) {
} }
} }
void uvWorkerAsyncCb(uv_async_t* handle) { void uvWorkerAsyncCb(uv_async_t* handle) {
SAsyncItem* item = handle->data; SAsyncItem* item = handle->data;
SWorkThrdObj* pThrd = item->pThrd; SWorkThrd* pThrd = item->pThrd;
SSvrConn* conn = NULL; SSvrConn* conn = NULL;
queue wq; queue wq;
// batch process to avoid to lock/unlock frequently // batch process to avoid to lock/unlock frequently
taosThreadMutexLock(&item->mtx); taosThreadMutexLock(&item->mtx);
@ -624,7 +624,7 @@ void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) {
assert(buf->base[0] == notify[0]); assert(buf->base[0] == notify[0]);
taosMemoryFree(buf->base); taosMemoryFree(buf->base);
SWorkThrdObj* pThrd = q->data; SWorkThrd* pThrd = q->data;
uv_pipe_t* pipe = (uv_pipe_t*)q; uv_pipe_t* pipe = (uv_pipe_t*)q;
if (!uv_pipe_pending_count(pipe)) { if (!uv_pipe_pending_count(pipe)) {
@ -692,10 +692,10 @@ void uvOnPipeConnectionCb(uv_connect_t* connect, int status) {
if (status != 0) { if (status != 0) {
return; return;
} }
SWorkThrdObj* pThrd = container_of(connect, SWorkThrdObj, connect_req); SWorkThrd* pThrd = container_of(connect, SWorkThrd, connect_req);
uv_read_start((uv_stream_t*)pThrd->pipe, uvAllocConnBufferCb, uvOnConnectionCb); uv_read_start((uv_stream_t*)pThrd->pipe, uvAllocConnBufferCb, uvOnConnectionCb);
} }
static bool addHandleToWorkloop(SWorkThrdObj* pThrd, char* pipeName) { static bool addHandleToWorkloop(SWorkThrd* pThrd, char* pipeName) {
pThrd->loop = (uv_loop_t*)taosMemoryMalloc(sizeof(uv_loop_t)); pThrd->loop = (uv_loop_t*)taosMemoryMalloc(sizeof(uv_loop_t));
if (0 != uv_loop_init(pThrd->loop)) { if (0 != uv_loop_init(pThrd->loop)) {
return false; return false;
@ -748,14 +748,14 @@ static bool addHandleToAcceptloop(void* arg) {
} }
void* transWorkerThread(void* arg) { void* transWorkerThread(void* arg) {
setThreadName("trans-worker"); setThreadName("trans-worker");
SWorkThrdObj* pThrd = (SWorkThrdObj*)arg; SWorkThrd* pThrd = (SWorkThrd*)arg;
uv_run(pThrd->loop, UV_RUN_DEFAULT); uv_run(pThrd->loop, UV_RUN_DEFAULT);
return NULL; return NULL;
} }
static SSvrConn* createConn(void* hThrd) { static SSvrConn* createConn(void* hThrd) {
SWorkThrdObj* pThrd = hThrd; SWorkThrd* pThrd = hThrd;
SSvrConn* pConn = (SSvrConn*)taosMemoryCalloc(1, sizeof(SSvrConn)); SSvrConn* pConn = (SSvrConn*)taosMemoryCalloc(1, sizeof(SSvrConn));
QUEUE_INIT(&pConn->queue); QUEUE_INIT(&pConn->queue);
@ -818,7 +818,7 @@ static void uvDestroyConn(uv_handle_t* handle) {
if (conn == NULL) { if (conn == NULL) {
return; return;
} }
SWorkThrdObj* thrd = conn->hostThrd; SWorkThrd* thrd = conn->hostThrd;
transReleaseExHandle(refMgt, conn->refId); transReleaseExHandle(refMgt, conn->refId);
transRemoveExHandle(refMgt, conn->refId); transRemoveExHandle(refMgt, conn->refId);
@ -863,7 +863,7 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
srv->numOfThreads = numOfThreads; srv->numOfThreads = numOfThreads;
srv->workerIdx = 0; srv->workerIdx = 0;
srv->numOfWorkerReady = 0; srv->numOfWorkerReady = 0;
srv->pThreadObj = (SWorkThrdObj**)taosMemoryCalloc(srv->numOfThreads, sizeof(SWorkThrdObj*)); srv->pThreadObj = (SWorkThrd**)taosMemoryCalloc(srv->numOfThreads, sizeof(SWorkThrd*));
srv->pipe = (uv_pipe_t**)taosMemoryCalloc(srv->numOfThreads, sizeof(uv_pipe_t*)); srv->pipe = (uv_pipe_t**)taosMemoryCalloc(srv->numOfThreads, sizeof(uv_pipe_t*));
srv->ip = ip; srv->ip = ip;
srv->port = port; srv->port = port;
@ -888,7 +888,7 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
assert(0 == uv_listen((uv_stream_t*)&srv->pipeListen, SOMAXCONN, uvPipeListenCb)); assert(0 == uv_listen((uv_stream_t*)&srv->pipeListen, SOMAXCONN, uvPipeListenCb));
for (int i = 0; i < srv->numOfThreads; i++) { for (int i = 0; i < srv->numOfThreads; i++) {
SWorkThrdObj* thrd = (SWorkThrdObj*)taosMemoryCalloc(1, sizeof(SWorkThrdObj)); SWorkThrd* thrd = (SWorkThrd*)taosMemoryCalloc(1, sizeof(SWorkThrd));
thrd->pTransInst = shandle; thrd->pTransInst = shandle;
thrd->quit = false; thrd->quit = false;
srv->pThreadObj[i] = thrd; srv->pThreadObj[i] = thrd;
@ -933,7 +933,7 @@ End:
return NULL; return NULL;
} }
void uvHandleQuit(SSvrMsg* msg, SWorkThrdObj* thrd) { void uvHandleQuit(SSvrMsg* msg, SWorkThrd* thrd) {
thrd->quit = true; thrd->quit = true;
if (QUEUE_IS_EMPTY(&thrd->conn)) { if (QUEUE_IS_EMPTY(&thrd->conn)) {
uv_walk(thrd->loop, uvWalkCb, NULL); uv_walk(thrd->loop, uvWalkCb, NULL);
@ -942,7 +942,7 @@ void uvHandleQuit(SSvrMsg* msg, SWorkThrdObj* thrd) {
} }
taosMemoryFree(msg); taosMemoryFree(msg);
} }
void uvHandleRelease(SSvrMsg* msg, SWorkThrdObj* thrd) { void uvHandleRelease(SSvrMsg* msg, SWorkThrd* thrd) {
SSvrConn* conn = msg->pConn; SSvrConn* conn = msg->pConn;
if (conn->status == ConnAcquire) { if (conn->status == ConnAcquire) {
reallocConnRef(conn); reallocConnRef(conn);
@ -956,12 +956,12 @@ void uvHandleRelease(SSvrMsg* msg, SWorkThrdObj* thrd) {
} }
destroySmsg(msg); destroySmsg(msg);
} }
void uvHandleResp(SSvrMsg* msg, SWorkThrdObj* thrd) { void uvHandleResp(SSvrMsg* msg, SWorkThrd* thrd) {
// send msg to client // send msg to client
tDebug("%s conn %p start to send resp (2/2)", transLabel(thrd->pTransInst), msg->pConn); tDebug("%s conn %p start to send resp (2/2)", transLabel(thrd->pTransInst), msg->pConn);
uvStartSendResp(msg); uvStartSendResp(msg);
} }
void uvHandleRegister(SSvrMsg* msg, SWorkThrdObj* thrd) { void uvHandleRegister(SSvrMsg* msg, SWorkThrd* thrd) {
SSvrConn* conn = msg->pConn; SSvrConn* conn = msg->pConn;
tDebug("%s conn %p register brokenlink callback", transLabel(thrd->pTransInst), conn); tDebug("%s conn %p register brokenlink callback", transLabel(thrd->pTransInst), conn);
if (conn->status == ConnAcquire) { if (conn->status == ConnAcquire) {
@ -982,7 +982,7 @@ void uvHandleRegister(SSvrMsg* msg, SWorkThrdObj* thrd) {
taosMemoryFree(msg); taosMemoryFree(msg);
} }
} }
void destroyWorkThrd(SWorkThrdObj* pThrd) { void destroyWorkThrd(SWorkThrd* pThrd) {
if (pThrd == NULL) { if (pThrd == NULL) {
return; return;
} }
@ -993,7 +993,7 @@ void destroyWorkThrd(SWorkThrdObj* pThrd) {
taosMemoryFree(pThrd->loop); taosMemoryFree(pThrd->loop);
taosMemoryFree(pThrd); taosMemoryFree(pThrd);
} }
void sendQuitToWorkThrd(SWorkThrdObj* pThrd) { void sendQuitToWorkThrd(SWorkThrd* pThrd) {
SSvrMsg* msg = taosMemoryCalloc(1, sizeof(SSvrMsg)); SSvrMsg* msg = taosMemoryCalloc(1, sizeof(SSvrMsg));
msg->type = Quit; msg->type = Quit;
tDebug("server send quit msg to work thread"); tDebug("server send quit msg to work thread");
@ -1060,7 +1060,7 @@ void transReleaseSrvHandle(void* handle) {
ASYNC_CHECK_HANDLE(exh, refId); ASYNC_CHECK_HANDLE(exh, refId);
SWorkThrdObj* pThrd = exh->pThrd; SWorkThrd* pThrd = exh->pThrd;
ASYNC_ERR_JRET(pThrd); ASYNC_ERR_JRET(pThrd);
STransMsg tmsg = {.code = 0, .info.handle = exh, .info.ahandle = NULL, .info.refId = refId}; STransMsg tmsg = {.code = 0, .info.handle = exh, .info.ahandle = NULL, .info.refId = refId};
@ -1090,7 +1090,7 @@ void transSendResponse(const STransMsg* msg) {
STransMsg tmsg = *msg; STransMsg tmsg = *msg;
tmsg.info.refId = refId; tmsg.info.refId = refId;
SWorkThrdObj* pThrd = exh->pThrd; SWorkThrd* pThrd = exh->pThrd;
ASYNC_ERR_JRET(pThrd); ASYNC_ERR_JRET(pThrd);
SSvrMsg* m = taosMemoryCalloc(1, sizeof(SSvrMsg)); SSvrMsg* m = taosMemoryCalloc(1, sizeof(SSvrMsg));
@ -1120,7 +1120,7 @@ void transRegisterMsg(const STransMsg* msg) {
STransMsg tmsg = *msg; STransMsg tmsg = *msg;
tmsg.info.refId = refId; tmsg.info.refId = refId;
SWorkThrdObj* pThrd = exh->pThrd; SWorkThrd* pThrd = exh->pThrd;
ASYNC_ERR_JRET(pThrd); ASYNC_ERR_JRET(pThrd);
SSvrMsg* m = taosMemoryCalloc(1, sizeof(SSvrMsg)); SSvrMsg* m = taosMemoryCalloc(1, sizeof(SSvrMsg));