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

Merge pull request #14068 from taosdata/fix/mnode 

fix: release/redirect problem
This commit is contained in:
Yihao Deng 2022-06-22 12:02:05 +08:00 committed by GitHub
parent f9dc73879f 5b9786bfc3
commit 8adfc5eeca
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 9 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -175,6 +175,7 @@ static void cliReleaseUnfinishedMsg(SCliConn* conn) {
#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)) { \
int connStatus = conn->status; \
uint64_t ahandle = head->ahandle; \ uint64_t ahandle = head->ahandle; \
CONN_GET_MSGCTX_BY_AHANDLE(conn, ahandle); \ CONN_GET_MSGCTX_BY_AHANDLE(conn, ahandle); \
conn->status = ConnRelease; \ conn->status = ConnRelease; \
@ -186,7 +187,9 @@ static void cliReleaseUnfinishedMsg(SCliConn* conn) {
} \ } \
destroyCmsg(pMsg); \ destroyCmsg(pMsg); \
cliReleaseUnfinishedMsg(conn); \ cliReleaseUnfinishedMsg(conn); \
addConnToPool(((SCliThrdObj*)conn->hostThrd)->pool, conn); \ if (connStatus != ConnInPool) { \
addConnToPool(((SCliThrdObj*)conn->hostThrd)->pool, conn); \
} \
return; \ return; \
} \ } \
} while (0) } while (0)
@ -450,7 +453,7 @@ void* destroyConnPool(void* pool) {
while (connList != NULL) { while (connList != NULL) {
while (!QUEUE_IS_EMPTY(&connList->conn)) { while (!QUEUE_IS_EMPTY(&connList->conn)) {
queue* h = QUEUE_HEAD(&connList->conn); queue* h = QUEUE_HEAD(&connList->conn);
QUEUE_REMOVE(h); // QUEUE_REMOVE(h);
SCliConn* c = QUEUE_DATA(h, SCliConn, conn); SCliConn* c = QUEUE_DATA(h, SCliConn, conn);
cliDestroyConn(c, true); cliDestroyConn(c, true);
} }
@ -481,6 +484,7 @@ static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port) {
conn->status = ConnNormal; conn->status = ConnNormal;
QUEUE_REMOVE(&conn->conn); QUEUE_REMOVE(&conn->conn);
QUEUE_INIT(&conn->conn); QUEUE_INIT(&conn->conn);
assert(h == &conn->conn);
return conn; return conn;
} }
static void addConnToPool(void* pool, SCliConn* conn) { static void addConnToPool(void* pool, SCliConn* conn) {
@ -562,9 +566,7 @@ static SCliConn* cliCreateConn(SCliThrdObj* pThrd) {
} }
static void cliDestroyConn(SCliConn* conn, bool clear) { 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);
if (clear) { if (clear) {
uv_close((uv_handle_t*)conn->stream, cliDestroy); uv_close((uv_handle_t*)conn->stream, cliDestroy);
} }
@ -1008,7 +1010,9 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
tTrace("%s use remote epset, inUse: %d, retry count:%d, limit: %d", pTransInst->label, pEpSet->inUse, tTrace("%s use remote epset, inUse: %d, retry count:%d, limit: %d", pTransInst->label, pEpSet->inUse,
pCtx->retryCount + 1, TRANS_RETRY_COUNT_LIMIT); pCtx->retryCount + 1, TRANS_RETRY_COUNT_LIMIT);
} }
addConnToPool(pThrd->pool, pConn); if (pConn->status != ConnInPool) {
addConnToPool(pThrd->pool, pConn);
}
STaskArg* arg = taosMemoryMalloc(sizeof(STaskArg)); STaskArg* arg = taosMemoryMalloc(sizeof(STaskArg));
arg->param1 = pMsg; arg->param1 = pMsg;