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

add http fail fast

This commit is contained in:
yihaoDeng 2023-12-06 11:44:41 +08:00
parent 0ffc707f1b
commit 15f690dd6d
1 changed files with 23 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
source/libs/transport/src/thttp.c
View File

@ -34,6 +34,7 @@ typedef struct SHttpModule {
SAsyncPool* asyncPool; SAsyncPool* asyncPool;
TdThread thread; TdThread thread;
SHashObj* connStatusTable; SHashObj* connStatusTable;
int8_t quit;
} SHttpModule; } SHttpModule;
typedef struct SHttpMsg { typedef struct SHttpMsg {
@ -190,19 +191,37 @@ static void httpDestroyMsg(SHttpMsg* msg) {
taosMemoryFree(msg->cont); taosMemoryFree(msg->cont);
taosMemoryFree(msg); taosMemoryFree(msg);
} }
static void httpMayDiscardMsg(SHttpModule* http, SAsyncItem* item) {
SHttpMsg *msg = NULL, *quitMsg = NULL;
int8_t quit = atomic_load_8(&http->quit);
if (quit == 1) {
while (!QUEUE_IS_EMPTY(&item->qmsg)) {
queue* h = QUEUE_HEAD(&item->qmsg);
QUEUE_REMOVE(h);
msg = QUEUE_DATA(h, SHttpMsg, q);
if (!msg->quit) {
httpDestroyMsg(msg);
} else {
quitMsg = msg;
}
}
QUEUE_PUSH(&item->qmsg, &quitMsg->q);
}
}
static void httpAsyncCb(uv_async_t* handle) { static void httpAsyncCb(uv_async_t* handle) {
SAsyncItem* item = handle->data; SAsyncItem* item = handle->data;
SHttpModule* http = item->pThrd; SHttpModule* http = item->pThrd;
SHttpMsg *msg = NULL, *quitMsg = NULL; SHttpMsg *msg = NULL, *quitMsg = NULL;
queue wq;
queue wq;
QUEUE_INIT(&wq); QUEUE_INIT(&wq);
static int32_t BATCH_SIZE = 5; static int32_t BATCH_SIZE = 5;
int32_t count = 0; int32_t count = 0;
taosThreadMutexLock(&item->mtx); taosThreadMutexLock(&item->mtx);
httpMayDiscardMsg(http, item);
while (!QUEUE_IS_EMPTY(&item->qmsg) && count++ < BATCH_SIZE) { while (!QUEUE_IS_EMPTY(&item->qmsg) && count++ < BATCH_SIZE) {
queue* h = QUEUE_HEAD(&item->qmsg); queue* h = QUEUE_HEAD(&item->qmsg);
@ -526,6 +545,8 @@ void transHttpEnvDestroy() {
return; return;
} }
SHttpModule* load = taosAcquireRef(httpRefMgt, httpRef); SHttpModule* load = taosAcquireRef(httpRefMgt, httpRef);
atomic_store_8(&load->quit, 1);
httpSendQuit(); httpSendQuit();
taosThreadJoin(load->thread, NULL); taosThreadJoin(load->thread, NULL);