Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
T
TDengine
未验证 提交 94ae5544 编写于 作者: D dapan1121 提交者： GitHub
浏览文件
操作

Merge pull request #17757 from taosdata/opt/httpModule 

enh: opt http module
上级 fd92f4b6 d21d88be
隐藏空白更改
内联 并排
Showing with 246 addition and 104 deletion
source/libs/transport/inc/transComm.h
浏览文件 @ 94ae5544
......@@ -428,6 +428,7 @@ void transDestoryExHandle(void* handle);
int32_t transGetRefMgt();
int32_t transGetInstMgt();
void transHttpEnvDestroy();
#ifdef __cplusplus
}
#endif
......
source/libs/transport/src/thttp.c
浏览文件 @ 94ae5544
......@@ -20,21 +20,52 @@
#include "thttp.h"
#include "taoserror.h"
#include "tlog.h"
#include "transComm.h"
// clang-format on
#define HTTP_RECV_BUF_SIZE 1024
typedef struct SHttpModule {
uv_loop_t* loop;
SAsyncPool* asyncPool;
TdThread thread;
} SHttpModule;
typedef struct SHttpMsg {
queue q;
char* server;
int32_t port;
char* cont;
int32_t len;
EHttpCompFlag flag;
int8_t quit;
SHttpModule* http;
} SHttpMsg;
typedef struct SHttpClient {
uv_connect_t conn;
uv_tcp_t tcp;
uv_write_t req;
uv_buf_t* wbuf;
char* rbuf;
char* addr;
uint16_t port;
uv_connect_t conn;
uv_tcp_t tcp;
uv_write_t req;
uv_buf_t* wbuf;
char* rbuf;
char* addr;
uint16_t port;
struct sockaddr_in dest;
} SHttpClient;
static TdThreadOnce transHttpInit = PTHREAD_ONCE_INIT;
static SHttpModule* thttp = NULL;
static void transHttpEnvInit();
static void httpHandleReq(SHttpMsg* msg);
static void httpHandleQuit(SHttpMsg* msg);
static int32_t httpSendQuit();
static int32_t taosSendHttpReportImpl(const char* server, uint16_t port, char* pCont, int32_t contLen,
EHttpCompFlag flag);
static int32_t taosBuildHttpHeader(const char* server, int32_t contLen, char* pHead, int32_t headLen,
EHttpCompFlag flag) {
if (flag == HTTP_FLAT) {
......@@ -53,6 +84,7 @@ static int32_t taosBuildHttpHeader(const char* server, int32_t contLen, char* pH
"Content-Length: %d\n\n",
server, contLen);
} else {
terrno = TSDB_CODE_INVALID_CFG;
return -1;
}
}
......@@ -126,116 +158,217 @@ _OVER:
return code;
}
static FORCE_INLINE int32_t taosBuildDstAddr(const char* server, uint16_t port, struct sockaddr_in* dest) {
uint32_t ip = taosGetIpv4FromFqdn(server);
if (ip == 0xffffffff) {
tError("http-report failed to get http server:%s since %s", server, errno == 0 ? "invalid http server" : terrstr());
return -1;
}
char buf[128] = {0};
tinet_ntoa(buf, ip);
uv_ip4_addr(buf, port, dest);
return 0;
}
static void* httpThread(void* arg) {
SHttpModule* http = (SHttpModule*)arg;
setThreadName("http-cli-send-thread");
uv_run(http->loop, UV_RUN_DEFAULT);
return NULL;
}
static void httpDestroyMsg(SHttpMsg* msg) {
if (msg == NULL) return;
taosMemoryFree(msg->server);
taosMemoryFree(msg->cont);
taosMemoryFree(msg);
}
static void httpAsyncCb(uv_async_t* handle) {
SAsyncItem* item = handle->data;
SHttpModule* http = item->pThrd;
SHttpMsg *msg = NULL, *quitMsg = NULL;
queue wq;
taosThreadMutexLock(&item->mtx);
QUEUE_MOVE(&item->qmsg, &wq);
taosThreadMutexUnlock(&item->mtx);
int count = 0;
while (!QUEUE_IS_EMPTY(&wq)) {
queue* h = QUEUE_HEAD(&wq);
QUEUE_REMOVE(h);
msg = QUEUE_DATA(h, SHttpMsg, q);
if (msg->quit) {
quitMsg = msg;
} else {
httpHandleReq(msg);
}
}
if (quitMsg) httpHandleQuit(quitMsg);
}
static FORCE_INLINE void destroyHttpClient(SHttpClient* cli) {
taosMemoryFree(cli->wbuf[0].base);
taosMemoryFree(cli->wbuf[1].base);
taosMemoryFree(cli->wbuf);
taosMemoryFree(cli->rbuf);
taosMemoryFree(cli->addr);
taosMemoryFree(cli);
}
static FORCE_INLINE void clientCloseCb(uv_handle_t* handle) {
SHttpClient* cli = handle->data;
destroyHttpClient(cli);
}
static FORCE_INLINE void clientAllocBuffCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
SHttpClient* cli = handle->data;
buf->base = cli->rbuf;
buf->len = HTTP_RECV_BUF_SIZE;
}
static FORCE_INLINE void clientRecvCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) {
SHttpClient* cli = handle->data;
if (nread < 0) {
uError("http-report recv error:%s", uv_err_name(nread));
tError("http-report recv error:%s", uv_err_name(nread));
} else {
uTrace("http-report succ to recv %d bytes", (int32_t)nread);
tTrace("http-report succ to recv %d bytes", (int32_t)nread);
}
if (!uv_is_closing((uv_handle_t*)&cli->tcp)) {
uv_close((uv_handle_t*)&cli->tcp, clientCloseCb);
} else {
destroyHttpClient(cli);
}
}
static void clientSentCb(uv_write_t* req, int32_t status) {
SHttpClient* cli = req->data;
if (status != 0) {
terrno = TAOS_SYSTEM_ERROR(status);
uError("http-report failed to send data %s", uv_strerror(status));
tError("http-report failed to send data, reason: %s, dst:%s:%d", uv_strerror(status), cli->addr, cli->port);
if (!uv_is_closing((uv_handle_t*)&cli->tcp)) {
uv_close((uv_handle_t*)&cli->tcp, clientCloseCb);
} else {
destroyHttpClient(cli);
}
return;
} else {
uTrace("http-report succ to send data");
tTrace("http-report succ to send data");
}
status = uv_read_start((uv_stream_t*)&cli->tcp, clientAllocBuffCb, clientRecvCb);
if (status != 0) {
terrno = TAOS_SYSTEM_ERROR(status);
uError("http-report failed to recv data,reason:%s, dst:%s:%d", uv_strerror(status), cli->addr, cli->port);
tError("http-report failed to recv data,reason:%s, dst:%s:%d", uv_strerror(status), cli->addr, cli->port);
if (!uv_is_closing((uv_handle_t*)&cli->tcp)) {
uv_close((uv_handle_t*)&cli->tcp, clientCloseCb);
} else {
destroyHttpClient(cli);
}
}
}
static void clientConnCb(uv_connect_t* req, int32_t status) {
SHttpClient* cli = req->data;
if (status != 0) {
terrno = TAOS_SYSTEM_ERROR(status);
uError("http-report failed to conn to server, reason:%s, dst:%s:%d", uv_strerror(status), cli->addr, cli->port);
tError("http-report failed to conn to server, reason:%s, dst:%s:%d", uv_strerror(status), cli->addr, cli->port);
if (!uv_is_closing((uv_handle_t*)&cli->tcp)) {
uv_close((uv_handle_t*)&cli->tcp, clientCloseCb);
} else {
destroyHttpClient(cli);
}
return;
}
status = uv_write(&cli->req, (uv_stream_t*)&cli->tcp, cli->wbuf, 2, clientSentCb);
if (0 != status) {
terrno = TAOS_SYSTEM_ERROR(status);
uError("http-report failed to send data,reason:%s, dst:%s:%d", uv_strerror(status), cli->addr, cli->port);
tError("http-report failed to send data,reason:%s, dst:%s:%d", uv_strerror(status), cli->addr, cli->port);
if (!uv_is_closing((uv_handle_t*)&cli->tcp)) {
uv_close((uv_handle_t*)&cli->tcp, clientCloseCb);
} else {
destroyHttpClient(cli);
}
}
}
static FORCE_INLINE int32_t taosBuildDstAddr(const char* server, uint16_t port, struct sockaddr_in* dest) {
uint32_t ip = taosGetIpv4FromFqdn(server);
if (ip == 0xffffffff) {
terrno = TAOS_SYSTEM_ERROR(errno);
uError("http-report failed to get http server:%s since %s", server, errno == 0 ? "invalid http server" : terrstr());
int32_t httpSendQuit() {
SHttpMsg* msg = taosMemoryCalloc(1, sizeof(SHttpMsg));
msg->quit = 1;
SHttpModule* load = atomic_load_ptr(&thttp);
if (load == NULL) {
httpDestroyMsg(msg);
tError("http-report already released");
return -1;
} else {
msg->http = load;
}
char buf[128] = {0};
tinet_ntoa(buf, ip);
uv_ip4_addr(buf, port, dest);
transAsyncSend(load->asyncPool, &(msg->q));
return 0;
}
int32_t taosSendHttpReport(const char* server, uint16_t port, char* pCont, int32_t contLen, EHttpCompFlag flag) {
struct sockaddr_in dest = {0};
if (taosBuildDstAddr(server, port, &dest) < 0) {
static int32_t taosSendHttpReportImpl(const char* server, uint16_t port, char* pCont, int32_t contLen,
EHttpCompFlag flag) {
SHttpMsg* msg = taosMemoryMalloc(sizeof(SHttpMsg));
msg->server = strdup(server);
msg->port = port;
msg->cont = taosMemoryMalloc(contLen);
memcpy(msg->cont, pCont, contLen);
msg->len = contLen;
msg->flag = flag;
msg->quit = 0;
SHttpModule* load = atomic_load_ptr(&thttp);
if (load == NULL) {
httpDestroyMsg(msg);
tError("http-report already released");
return -1;
} else {
msg->http = load;
transAsyncSend(load->asyncPool, &(msg->q));
}
return 0;
}
static void httpDestroyClientCb(uv_handle_t* handle) {
SHttpClient* http = handle->data;
destroyHttpClient(http);
}
static void httpWalkCb(uv_handle_t* handle, void* arg) {
// impl later
if (!uv_is_closing(handle)) {
uv_handle_type type = uv_handle_get_type(handle);
if (uv_handle_get_type(handle) == UV_TCP) {
uv_close(handle, httpDestroyClientCb);
} else {
uv_close(handle, NULL);
}
}
if (flag == HTTP_GZIP) {
int32_t dstLen = taosCompressHttpRport(pCont, contLen);
return;
}
static void httpHandleQuit(SHttpMsg* msg) {
SHttpModule* http = msg->http;
taosMemoryFree(msg);
uv_walk(http->loop, httpWalkCb, NULL);
}
static void httpHandleReq(SHttpMsg* msg) {
SHttpModule* http = msg->http;
struct sockaddr_in dest = {0};
if (taosBuildDstAddr(msg->server, msg->port, &dest) < 0) {
goto END;
}
if (msg->flag == HTTP_GZIP) {
int32_t dstLen = taosCompressHttpRport(msg->cont, msg->len);
if (dstLen > 0) {
contLen = dstLen;
msg->len = dstLen;
} else {
flag = HTTP_FLAT;
msg->flag = HTTP_FLAT;
}
if (dstLen < 0) {
goto END;
}
}
terrno = 0;
char header[2048] = {0};
int32_t headLen = taosBuildHttpHeader(server, contLen, header, sizeof(header), flag);
int32_t len = 2048;
char* header = taosMemoryCalloc(1, len);
int32_t headLen = taosBuildHttpHeader(msg->server, msg->len, header, len, msg->flag);
if (headLen < 0) {
taosMemoryFree(header);
goto END;
}
uv_buf_t* wb = taosMemoryCalloc(2, sizeof(uv_buf_t));
wb[0] = uv_buf_init((char*)header, headLen); // stack var
wb[1] = uv_buf_init((char*)pCont, contLen); // heap var
wb[0] = uv_buf_init((char*)header, strlen(header)); // heap var
wb[1] = uv_buf_init((char*)msg->cont, msg->len); // heap var
SHttpClient* cli = taosMemoryCalloc(1, sizeof(SHttpClient));
cli->conn.data = cli;
......@@ -243,41 +376,71 @@ int32_t taosSendHttpReport(const char* server, uint16_t port, char* pCont, int32
cli->req.data = cli;
cli->wbuf = wb;
cli->rbuf = taosMemoryCalloc(1, HTTP_RECV_BUF_SIZE);
cli->addr = tstrdup(server);
cli->port = port;
cli->addr = msg->server;
cli->port = msg->port;
cli->dest = dest;
taosMemoryFree(msg);
uv_tcp_init(http->loop, &cli->tcp);
uv_loop_t* loop = taosMemoryMalloc(sizeof(uv_loop_t));
int err = uv_loop_init(loop);
if (err != 0) {
uError("http-report failed to init uv_loop, reason: %s", uv_strerror(err));
taosMemoryFree(loop);
terrno = TAOS_SYSTEM_ERROR(err);
destroyHttpClient(cli);
return terrno;
}
uv_tcp_init(loop, &cli->tcp);
// set up timeout to avoid stuck;
int32_t fd = taosCreateSocketWithTimeout(5);
int ret = uv_tcp_open((uv_tcp_t*)&cli->tcp, fd);
if (ret != 0) {
tError("http-report failed to open socket, reason:%s, dst:%s:%d", uv_strerror(ret), cli->addr, cli->port);
destroyHttpClient(cli);
return;
}
int ret = uv_tcp_open((uv_tcp_t*)&cli->tcp, fd);
ret = uv_tcp_connect(&cli->conn, &cli->tcp, (const struct sockaddr*)&cli->dest, clientConnCb);
if (ret != 0) {
uError("http-report failed to open socket, reason:%s, dst:%s:%d", uv_strerror(ret), cli->addr, cli->port);
tError("http-report failed to connect to http-server, reason:%s, dst:%s:%d", uv_strerror(ret), cli->addr,
cli->port);
destroyHttpClient(cli);
uv_stop(loop);
terrno = TAOS_SYSTEM_ERROR(ret);
} else {
ret = uv_tcp_connect(&cli->conn, &cli->tcp, (const struct sockaddr*)&dest, clientConnCb);
if (ret != 0) {
uError("http-report failed to connect to http-server, reason:%s, dst:%s:%d", uv_strerror(ret), cli->addr,
cli->port);
destroyHttpClient(cli);
uv_stop(loop);
terrno = TAOS_SYSTEM_ERROR(ret);
}
}
return;
END:
tError("http-report failed to report, reason: %s, addr: %s:%d", terrstr(), msg->server, msg->port);
httpDestroyMsg(msg);
}
int32_t taosSendHttpReport(const char* server, uint16_t port, char* pCont, int32_t contLen, EHttpCompFlag flag) {
taosThreadOnce(&transHttpInit, transHttpEnvInit);
return taosSendHttpReportImpl(server, port, pCont, contLen, flag);
}
static void transHttpEnvInit() {
SHttpModule* http = taosMemoryMalloc(sizeof(SHttpModule));
http->loop = taosMemoryMalloc(sizeof(uv_loop_t));
uv_loop_init(http->loop);
http->asyncPool = transAsyncPoolCreate(http->loop, 1, http, httpAsyncCb);
int err = taosThreadCreate(&http->thread, NULL, httpThread, (void*)http);
if (err != 0) {
taosMemoryFree(http->loop);
taosMemoryFree(http);
http = NULL;
}
atomic_store_ptr(&thttp, http);
}
void transHttpEnvDestroy() {
SHttpModule* load = atomic_load_ptr(&thttp);
if (load == NULL) {
return;
}
httpSendQuit();
taosThreadJoin(load->thread, NULL);
TRANS_DESTROY_ASYNC_POOL_MSG(load->asyncPool, SHttpMsg, httpDestroyMsg);
transAsyncPoolDestroy(load->asyncPool);
uv_loop_close(load->loop);
taosMemoryFree(load->loop);
taosMemoryFree(load);
uv_run(loop, UV_RUN_DEFAULT);
uv_loop_close(loop);
taosMemoryFree(loop);
return terrno;
atomic_store_ptr(&thttp, NULL);
}
source/libs/transport/src/trans.c
浏览文件 @ 94ae5544
......@@ -172,6 +172,8 @@ int32_t rpcInit() {
}
void rpcCleanup(void) {
transCleanup();
transHttpEnvDestroy();
return;
}
......
source/libs/transport/src/transCli.c
浏览文件 @ 94ae5544
......@@ -187,18 +187,8 @@ static void cliReleaseUnfinishedMsg(SCliConn* conn) {
snprintf(key, sizeof(key), "%s:%d", ip, (int)port); \
} while (0)
#define CONN_HOST_THREAD_IDX1(idx, exh, refId, pThrd) \
do { \
if (exh == NULL) { \
idx = -1; \
} else { \
ASYNC_CHECK_HANDLE((exh), refId); \
pThrd = (SCliThrd*)(exh)->pThrd; \
} \
} while (0)
#define CONN_PERSIST_TIME(para) ((para) <= 90000 ? 90000 : (para))
#define CONN_GET_HOST_THREAD(conn) (conn ? ((SCliConn*)conn)->hostThrd : NULL)
#define CONN_GET_INST_LABEL(conn) (((STrans*)(((SCliThrd*)(conn)->hostThrd)->pTransInst))->label)
#define CONN_PERSIST_TIME(para) ((para) <= 90000 ? 90000 : (para))
#define CONN_GET_INST_LABEL(conn) (((STrans*)(((SCliThrd*)(conn)->hostThrd)->pTransInst))->label)
#define CONN_GET_MSGCTX_BY_AHANDLE(conn, ahandle) \
do { \
......@@ -217,6 +207,7 @@ static void cliReleaseUnfinishedMsg(SCliConn* conn) {
tDebug("msg found, %" PRIu64 "", ahandle); \
} \
} while (0)
#define CONN_GET_NEXT_SENDMSG(conn) \
do { \
int i = 0; \
......@@ -231,21 +222,6 @@ static void cliReleaseUnfinishedMsg(SCliConn* conn) {
} \
} while (0)
#define CONN_HANDLE_THREAD_QUIT(thrd) \
do { \
if (thrd->quit) { \
return; \
} \
} while (0)
#define CONN_HANDLE_BROKEN(conn) \
do { \
if (conn->broken) { \
cliHandleExcept(conn); \
return; \
} \
} while (0)
#define CONN_SET_PERSIST_BY_APP(conn) \
do { \
if (conn->status == ConnNormal) { \
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或