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

fix read buf

This commit is contained in:
yihaoDeng 2022-07-27 14:52:44 +08:00
parent ea2c6f6805
commit 5ed23e7912
3 changed files with 18 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -60,6 +60,7 @@ typedef struct SCliThrd {
int64_t pid; // pid int64_t pid; // pid
uv_loop_t* loop; uv_loop_t* loop;
SAsyncPool* asyncPool; SAsyncPool* asyncPool;
uv_idle_t* idle;
uv_timer_t timer; uv_timer_t timer;
void* pool; // conn pool void* pool; // conn pool
@ -116,6 +117,7 @@ static void cliSendCb(uv_write_t* req, int status);
// callback after conn to server // callback after conn to server
static void cliConnCb(uv_connect_t* req, int status); static void cliConnCb(uv_connect_t* req, int status);
static void cliAsyncCb(uv_async_t* handle); static void cliAsyncCb(uv_async_t* handle);
static void cliIdleCb(uv_idle_t* handle);
static int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg); static int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg);
@ -962,6 +964,10 @@ static void cliAsyncCb(uv_async_t* handle) {
} }
if (pThrd->stopMsg != NULL) cliHandleQuit(pThrd->stopMsg, pThrd); if (pThrd->stopMsg != NULL) cliHandleQuit(pThrd->stopMsg, pThrd);
} }
static void cliIdleCb(uv_idle_t* handle) {
SCliThrd* thrd = handle->data;
tTrace("do idle work");
}
static void* cliWorkThread(void* arg) { static void* cliWorkThread(void* arg) {
SCliThrd* pThrd = (SCliThrd*)arg; SCliThrd* pThrd = (SCliThrd*)arg;
@ -1024,6 +1030,11 @@ static SCliThrd* createThrdObj() {
uv_timer_init(pThrd->loop, &pThrd->timer); uv_timer_init(pThrd->loop, &pThrd->timer);
pThrd->timer.data = pThrd; pThrd->timer.data = pThrd;
// pThrd->idle = taosMemoryCalloc(1, sizeof(uv_idle_t));
// uv_idle_init(pThrd->loop, pThrd->idle);
// pThrd->idle->data = pThrd;
// uv_idle_start(pThrd->idle, cliIdleCb);
pThrd->pool = createConnPool(4); pThrd->pool = createConnPool(4);
transDQCreate(pThrd->loop, &pThrd->delayQueue); transDQCreate(pThrd->loop, &pThrd->delayQueue);
@ -1045,6 +1056,8 @@ static void destroyThrdObj(SCliThrd* pThrd) {
transDQDestroy(pThrd->delayQueue, destroyCmsg); transDQDestroy(pThrd->delayQueue, destroyCmsg);
transDQDestroy(pThrd->timeoutQueue, NULL); transDQDestroy(pThrd->timeoutQueue, NULL);
taosMemoryFree(pThrd->idle);
taosMemoryFree(pThrd->loop); taosMemoryFree(pThrd->loop);
taosMemoryFree(pThrd); taosMemoryFree(pThrd);
} }

6
source/libs/transport/src/transComm.c
View File

@ -152,16 +152,18 @@ int transDumpFromBuffer(SConnBuffer* connBuf, char** buf) {
int transResetBuffer(SConnBuffer* connBuf) { int transResetBuffer(SConnBuffer* connBuf) {
SConnBuffer* p = connBuf; SConnBuffer* p = connBuf;
if (p->total <= p->len) { if (p->total < p->len) {
int left = p->len - p->total; int left = p->len - p->total;
memmove(p->buf, p->buf + p->total, left); memmove(p->buf, p->buf + p->total, left);
p->left = -1; p->left = -1;
p->total = 0; p->total = 0;
p->len = left; p->len = left;
} else { } else if (p->total == p->len) {
p->left = -1; p->left = -1;
p->total = 0; p->total = 0;
p->len = 0; p->len = 0;
} else {
assert(0);
} }
return 0; return 0;
} }

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

@ -238,7 +238,7 @@ static void uvHandleReq(SSvrConn* pConn) {
transMsg.msgType = pHead->msgType; transMsg.msgType = pHead->msgType;
transMsg.code = pHead->code; transMsg.code = pHead->code;
transClearBuffer(&pConn->readBuf); // transClearBuffer(&pConn->readBuf);
pConn->inType = pHead->msgType; pConn->inType = pHead->msgType;
if (pConn->status == ConnNormal) { if (pConn->status == ConnNormal) {