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

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

This commit is contained in:
dapan1121 2022-03-25 15:43:53 +08:00
parent 64f0fe2f82 5cc8da591a
commit e561496466
2 changed files with 21 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -154,20 +154,20 @@ static void destroyThrdObj(SCliThrdObj* pThrd);
} \ } \
} while (0) } while (0)
#define CONN_GET_MSGCTX_BY_AHANDLE(conn, ahandle) \ #define CONN_GET_MSGCTX_BY_AHANDLE(conn, ahandle) \
do { \ do { \
int i = 0, sz = transQueueSize(&conn->cliMsgs); \ int i = 0, sz = transQueueSize(&conn->cliMsgs); \
for (; i < sz; i++) { \ for (; i < sz; i++) { \
pMsg = transQueueGet(&conn->cliMsgs, i); \ pMsg = transQueueGet(&conn->cliMsgs, i); \
if (pMsg != NULL && (uint64_t)pMsg->ctx->ahandle == ahandle) { \ if (pMsg != NULL && pMsg->ctx != NULL && (uint64_t)pMsg->ctx->ahandle == ahandle) { \
break; \ break; \
} \ } \
} \ } \
if (i == sz) { \ if (i == sz) { \
pMsg = NULL; \ pMsg = NULL; \
} else { \ } else { \
pMsg = transQueueRm(&conn->cliMsgs, i); \ pMsg = transQueueRm(&conn->cliMsgs, i); \
} \ } \
} while (0) } while (0)
#define CONN_GET_NEXT_SENDMSG(conn) \ #define CONN_GET_NEXT_SENDMSG(conn) \
do { \ do { \

12
source/libs/transport/src/transSrv.c
View File

@ -644,6 +644,7 @@ static void uvDestroyConn(uv_handle_t* handle) {
// free(conn); // free(conn);
if (thrd->quit && QUEUE_IS_EMPTY(&thrd->conn)) { if (thrd->quit && QUEUE_IS_EMPTY(&thrd->conn)) {
tTrace("work thread quit");
uv_loop_close(thrd->loop); uv_loop_close(thrd->loop);
uv_stop(thrd->loop); uv_stop(thrd->loop);
} }
@ -705,12 +706,12 @@ End:
return NULL; return NULL;
} }
void uvHandleQuit(SSrvMsg* msg, SWorkThrdObj* thrd) { void uvHandleQuit(SSrvMsg* msg, SWorkThrdObj* thrd) {
thrd->quit = true;
if (QUEUE_IS_EMPTY(&thrd->conn)) { if (QUEUE_IS_EMPTY(&thrd->conn)) {
uv_loop_close(thrd->loop); uv_loop_close(thrd->loop);
uv_stop(thrd->loop); uv_stop(thrd->loop);
} else { } else {
destroyAllConn(thrd); destroyAllConn(thrd);
thrd->quit = true;
} }
free(msg); free(msg);
} }
@ -773,15 +774,16 @@ void sendQuitToWorkThrd(SWorkThrdObj* pThrd) {
void transCloseServer(void* arg) { void transCloseServer(void* arg) {
// impl later // impl later
SServerObj* srv = arg; SServerObj* srv = arg;
for (int i = 0; i < srv->numOfThreads; i++) {
sendQuitToWorkThrd(srv->pThreadObj[i]);
destroyWorkThrd(srv->pThreadObj[i]);
}
tDebug("send quit msg to accept thread"); tDebug("send quit msg to accept thread");
uv_async_send(srv->pAcceptAsync); uv_async_send(srv->pAcceptAsync);
taosThreadJoin(srv->thread, NULL); taosThreadJoin(srv->thread, NULL);
for (int i = 0; i < srv->numOfThreads; i++) {
sendQuitToWorkThrd(srv->pThreadObj[i]);
destroyWorkThrd(srv->pThreadObj[i]);
}
free(srv->pThreadObj); free(srv->pThreadObj);
free(srv->pAcceptAsync); free(srv->pAcceptAsync);
free(srv->loop); free(srv->loop);