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

Merge branch 'feature/scheduler' of https://github.com/taosdata/TDengine into scheduler

This commit is contained in:
yihaoDeng 2022-03-24 09:26:56 +08:00
parent 3cbce27a9f b340ec5e49
commit 4ec585addc
1 changed files with 19 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -169,6 +169,19 @@ static void destroyThrdObj(SCliThrdObj* pThrd);
pMsg = transQueueRm(&conn->cliMsgs, i); \ pMsg = transQueueRm(&conn->cliMsgs, i); \
} \ } \
} while (0) } while (0)
#define CONN_GET_NEXT_SENDMSG(conn) \
do { \
int i = 0; \
do { \
pCliMsg = transQueueGet(&conn->cliMsgs, i++); \
if (pCliMsg && 0 == pCliMsg->sent) { \
break; \
} \
} while (pCliMsg != NULL); \
if (pCliMsg == NULL) { \
goto _RETURN; \
} \
} while (0)
#define CONN_HANDLE_THREAD_QUIT(thrd) \ #define CONN_HANDLE_THREAD_QUIT(thrd) \
do { \ do { \
@ -203,19 +216,12 @@ static void* cliWorkThread(void* arg);
bool cliMaySendCachedMsg(SCliConn* conn) { bool cliMaySendCachedMsg(SCliConn* conn) {
if (!transQueueEmpty(&conn->cliMsgs)) { if (!transQueueEmpty(&conn->cliMsgs)) {
SCliMsg* pCliMsg = NULL; SCliMsg* pCliMsg = NULL;
int i = 0; CONN_GET_NEXT_SENDMSG(conn);
do {
pCliMsg = transQueueGet(&conn->cliMsgs, i++);
if (pCliMsg && 0 == pCliMsg->sent) {
break;
}
} while (pCliMsg != NULL);
if (pCliMsg == NULL) {
return false;
}
cliSend(conn); cliSend(conn);
} }
return false; return false;
_RETURN:
return false;
} }
void cliHandleResp(SCliConn* conn) { void cliHandleResp(SCliConn* conn) {
SCliThrdObj* pThrd = conn->hostThrd; SCliThrdObj* pThrd = conn->hostThrd;
@ -570,17 +576,7 @@ void cliSend(SCliConn* pConn) {
assert(!transQueueEmpty(&pConn->cliMsgs)); assert(!transQueueEmpty(&pConn->cliMsgs));
SCliMsg* pCliMsg = NULL; SCliMsg* pCliMsg = NULL;
int i = 0; CONN_GET_NEXT_SENDMSG(pConn);
do {
pCliMsg = transQueueGet(&pConn->cliMsgs, i++);
if (pCliMsg && 0 == pCliMsg->sent) {
break;
}
} while (pCliMsg != NULL);
if (pCliMsg == NULL) {
return;
}
pCliMsg->sent = 1; pCliMsg->sent = 1;
STransConnCtx* pCtx = pCliMsg->ctx; STransConnCtx* pCtx = pCliMsg->ctx;
@ -635,6 +631,8 @@ void cliSend(SCliConn* pConn) {
pConn->writeReq.data = pConn; pConn->writeReq.data = pConn;
uv_write(&pConn->writeReq, (uv_stream_t*)pConn->stream, &wb, 1, cliSendCb); uv_write(&pConn->writeReq, (uv_stream_t*)pConn->stream, &wb, 1, cliSendCb);
return;
_RETURN:
return; return;
} }