Skip to content

TDengine
TDengine

taosdata / TDengine
10 个月 前同步成功

通知 1177
Star 22014
Fork 4786
TDengine
TDengine

前往新版Gitcode，体验更适合开发者的 AI 搜索 >>

提交 c063ea69 编写于 作者: dengyihao's avatar dengyihao
浏览文件
操作

add client

上级 71729218
隐藏空白更改
内联 并排
Showing with 460 addition and 100 deletion
source/libs/transport/inc/transComm.h
浏览文件 @ c063ea69
......@@ -102,38 +102,101 @@ typedef void* queue[2];
#define QUEUE_DATA(e, type, field) ((type*)((void*)((char*)(e)-offsetof(type, field))))
typedef struct {
SRpcInfo* pRpc; // associated SRpcInfo
SEpSet epSet; // ip list provided by app
void* ahandle; // handle provided by app
struct SRpcConn* pConn; // pConn allocated
tmsg_t msgType; // message type
uint8_t* pCont; // content provided by app
int32_t contLen; // content length
int32_t code; // error code
int16_t numOfTry; // number of try for different servers
int8_t oldInUse; // server EP inUse passed by app
int8_t redirect; // flag to indicate redirect
int8_t connType; // connection type
int64_t rid; // refId returned by taosAddRef
SRpcMsg* pRsp; // for synchronous API
tsem_t* pSem; // for synchronous API
SEpSet* pSet; // for synchronous API
char msg[0]; // RpcHead starts from here
SRpcInfo* pRpc; // associated SRpcInfo
SEpSet epSet; // ip list provided by app
void* ahandle; // handle provided by app
// struct SRpcConn* pConn; // pConn allocated
tmsg_t msgType; // message type
uint8_t* pCont; // content provided by app
int32_t contLen; // content length
// int32_t code; // error code
// int16_t numOfTry; // number of try for different servers
// int8_t oldInUse; // server EP inUse passed by app
// int8_t redirect; // flag to indicate redirect
int8_t connType; // connection type
int64_t rid; // refId returned by taosAddRef
SRpcMsg* pRsp; // for synchronous API
tsem_t* pSem; // for synchronous API
char* ip;
uint32_t port;
// SEpSet* pSet; // for synchronous API
} SRpcReqContext;
typedef struct {
SRpcInfo* pRpc; // associated SRpcInfo
SEpSet epSet; // ip list provided by app
void* ahandle; // handle provided by app
// struct SRpcConn* pConn; // pConn allocated
tmsg_t msgType; // message type
uint8_t* pCont; // content provided by app
int32_t contLen; // content length
// int32_t code; // error code
// int16_t numOfTry; // number of try for different servers
// int8_t oldInUse; // server EP inUse passed by app
// int8_t redirect; // flag to indicate redirect
int8_t connType; // connection type
int64_t rid; // refId returned by taosAddRef
SRpcMsg* pRsp; // for synchronous API
tsem_t* pSem; // for synchronous API
char* ip;
uint32_t port;
// SEpSet* pSet; // for synchronous API
} STransConnCtx;
#pragma pack(push, 1)
typedef struct {
char version : 4; // RPC version
char comp : 4; // compression algorithm, 0:no compression 1:lz4
char resflag : 2; // reserved bits
char spi : 3; // security parameter index
char encrypt : 3; // encrypt algorithm, 0: no encryption
uint32_t code; // del later
uint32_t msgType;
int32_t msgLen;
uint8_t content[0]; // message body starts from here
} STransMsgHead;
typedef struct {
int32_t reserved;
int32_t contLen;
} STransCompMsg;
typedef struct {
uint32_t timeStamp;
uint8_t auth[TSDB_AUTH_LEN];
} STransDigestMsg;
#pragma pack(pop)
#define container_of(ptr, type, member) ((type*)((char*)(ptr)-offsetof(type, member)))
#define RPC_RESERVE_SIZE (sizeof(SRpcReqContext))
#define RPC_RESERVE_SIZE (sizeof(STranConnCtx))
#define RPC_MSG_OVERHEAD (sizeof(SRpcReqContext) + sizeof(SRpcHead) + sizeof(SRpcDigest))
#define RPC_MSG_OVERHEAD (sizeof(SRpcHead) + sizeof(SRpcDigest))
#define rpcHeadFromCont(cont) ((SRpcHead*)((char*)cont - sizeof(SRpcHead)))
#define rpcContFromHead(msg) (msg + sizeof(SRpcHead))
#define rpcMsgLenFromCont(contLen) (contLen + sizeof(SRpcHead))
#define rpcContLenFromMsg(msgLen) (msgLen - sizeof(SRpcHead))
#define rpcIsReq(type) (type & 1U)
#define TRANS_RESERVE_SIZE (sizeof(STranConnCtx))
#define TRANS_MSG_OVERHEAD (sizeof(STransMsgHead) + sizeof(STransDigestMsg))
#define transHeadFromCont(cont) ((STransMsgHead*)((char*)cont - sizeof(STransMsgHead)))
#define transContFromHead(msg) (msg + sizeof(STransMsgHead))
#define transMsgLenFromCont(contLen) (contLen + sizeof(STransMsgHead))
#define transContLenFromMsg(msgLen) (msgLen - sizeof(STransMsgHead));
#define transIsReq(type) (type & 1U)
int rpcAuthenticateMsg(void* pMsg, int msgLen, void* pAuth, void* pKey);
void rpcBuildAuthHead(void* pMsg, int msgLen, void* pAuth, void* pKey);
int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen);
SRpcHead* rpcDecompressRpcMsg(SRpcHead* pHead);
int transAuthenticateMsg(void* pMsg, int msgLen, void* pAuth, void* pKey);
void transBuildAuthHead(void* pMsg, int msgLen, void* pAuth, void* pKey);
bool transCompressMsg(char* msg, int32_t len, int32_t* flen);
bool transDecompressMsg(char* msg, int32_t len, int32_t* flen);
#endif
source/libs/transport/inc/transportInt.h
浏览文件 @ c063ea69
......@@ -45,6 +45,9 @@ extern "C" {
void* taosInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle);
void* taosInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle);
void taosCloseServer(void* arg);
void taosCloseClient(void* arg);
typedef struct {
int sessions; // number of sessions allowed
int numOfThreads; // number of threads to process incoming messages
......
source/libs/transport/src/trans.c
浏览文件 @ c063ea69
......@@ -24,8 +24,9 @@ typedef struct SConnBuffer {
int left;
} SConnBuffer;
void* (*taosHandle[])(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle) = {
void* (*taosInitHandle[])(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle) = {
taosInitServer, taosInitClient};
void (*taosCloseHandle[])(void* arg) = {taosCloseServer, taosCloseClient};
void* rpcOpen(const SRpcInit* pInit) {
SRpcInfo* pRpc = calloc(1, sizeof(SRpcInfo));
......@@ -38,11 +39,15 @@ void* rpcOpen(const SRpcInit* pInit) {
pRpc->cfp = pInit->cfp;
pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads;
pRpc->connType = pInit->connType;
pRpc->tcphandle = (*taosHandle[pRpc->connType])(0, pInit->localPort, pRpc->label, pRpc->numOfThreads, NULL, pRpc);
pRpc->tcphandle = (*taosInitHandle[pRpc->connType])(0, pInit->localPort, pRpc->label, pRpc->numOfThreads, NULL, pRpc);
return pRpc;
}
void rpcClose(void* arg) { return; }
void rpcClose(void* arg) {
SRpcInfo* pRpc = (SRpcInfo*)arg;
(*taosCloseHandle[pRpc->connType])(pRpc->tcphandle);
return;
}
void* rpcMallocCont(int contLen) {
int size = contLen + RPC_MSG_OVERHEAD;
......@@ -53,7 +58,7 @@ void* rpcMallocCont(int contLen) {
} else {
tTrace("malloc mem:%p size:%d", start, size);
}
return start + sizeof(SRpcReqContext) + sizeof(SRpcHead);
return start + sizeof(STransMsgHead);
}
void rpcFreeCont(void* cont) { return; }
void* rpcReallocCont(void* ptr, int contLen) { return NULL; }
......
source/libs/transport/src/transCli.c
浏览文件 @ c063ea69
......@@ -26,10 +26,12 @@ typedef struct SCliConn {
char spi;
char secured;
} SCliConn;
typedef struct SCliMsg {
SRpcReqContext* context;
queue q;
uint64_t st;
STransConnCtx* ctx;
SRpcMsg msg;
queue q;
uint64_t st;
} SCliMsg;
typedef struct SCliThrdObj {
......@@ -74,6 +76,7 @@ static void clientReadCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf
if (nread > 0) {
return;
}
//
uv_close((uv_handle_t*)handle, clientDestroy);
}
......@@ -101,10 +104,12 @@ static void clientWriteCb(uv_write_t* req, int status) {
}
static void clientWrite(SCliConn* pConn) {
SCliMsg* pMsg = pConn->data;
SRpcHead* pHead = rpcHeadFromCont(pMsg->context->pCont);
int msgLen = rpcMsgLenFromCont(pMsg->context->contLen);
char* msg = (char*)(pHead);
SCliMsg* pCliMsg = pConn->data;
SRpcMsg* pMsg = (SRpcMsg*)(&pCliMsg->msg);
STransMsgHead* pHead = transHeadFromCont(pMsg->pCont);
int msgLen = transMsgLenFromCont(pMsg->contLen);
char* msg = (char*)(pHead);
uv_buf_t wb = uv_buf_init(msg, msgLen);
uv_write(pConn->writeReq, (uv_stream_t*)pConn->stream, &wb, 1, clientWriteCb);
......@@ -118,17 +123,17 @@ static void clientConnCb(uv_connect_t* req, int status) {
return;
}
SCliMsg* pMsg = pConn->data;
SEpSet* pEpSet = &pMsg->context->epSet;
SRpcMsg rpcMsg;
SCliMsg* pMsg = pConn->data;
STransConnCtx* pCtx = ((SCliMsg*)(pConn->data))->ctx;
SRpcMsg rpcMsg;
rpcMsg.ahandle = pCtx->ahandle;
char* fqdn = pEpSet->fqdn[pEpSet->inUse];
uint32_t port = pEpSet->port[pEpSet->inUse];
if (status != 0) {
// call user fp later
tError("failed to connect server(%s, %d), errmsg: %s", fqdn, port, uv_strerror(status));
SRpcInfo* pRpc = pMsg->context->pRpc;
(pRpc->cfp)(NULL, &rpcMsg, pEpSet);
tError("failed to connect server(%s, %d), errmsg: %s", pCtx->ip, pCtx->port, uv_strerror(status));
SRpcInfo* pRpc = pMsg->ctx->pRpc;
(pRpc->cfp)(NULL, &rpcMsg, NULL);
uv_close((uv_handle_t*)req->handle, clientDestroy);
return;
}
......@@ -146,22 +151,18 @@ static void addConnToCache(void* cache, char* ip, uint32_t port, SCliConn* conn)
}
static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
SEpSet* pEpSet = &pMsg->context->epSet;
char* fqdn = pEpSet->fqdn[pEpSet->inUse];
uint32_t port = pEpSet->port[pEpSet->inUse];
uint64_t el = taosGetTimestampUs() - pMsg->st;
uint64_t et = taosGetTimestampUs();
uint64_t el = et - pMsg->st;
tDebug("msg tran time cost: %" PRIu64 "", el);
et = taosGetTimestampUs();
SCliConn* conn = getConnFromCache(pThrd->cache, fqdn, port);
STransConnCtx* pCtx = pMsg->ctx;
SCliConn* conn = getConnFromCache(pThrd->cache, pCtx->ip, pCtx->port);
if (conn != NULL) {
// impl later
conn->data = pMsg;
conn->writeReq->data = conn;
clientWrite(conn);
// uv_buf_t wb;
// uv_write(conn->writeReq, (uv_stream_t*)conn->stream, &wb, 1, clientWriteCb);
} else {
SCliConn* conn = malloc(sizeof(SCliConn));
......@@ -171,23 +172,11 @@ static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
conn->connReq.data = conn;
conn->data = pMsg;
struct sockaddr_in addr;
uv_ip4_addr(fqdn, port, &addr);
uv_ip4_addr(pMsg->ctx->ip, pMsg->ctx->port, &addr);
// handle error in callback if fail to connect
uv_tcp_connect(&conn->connReq, (uv_tcp_t*)(conn->stream), (const struct sockaddr*)&addr, clientConnCb);
// SRpcMsg rpcMsg;
// SEpSet* pEpSet = &pMsg->context->epSet;
// SRpcInfo* pRpc = pMsg->context->pRpc;
//// rpcMsg.ahandle = pMsg->context->ahandle;
// rpcMsg.pCont = NULL;
// rpcMsg.ahandle = pMsg->context->ahandle;
// uint64_t el1 = taosGetTimestampUs() - et;
// tError("msg tran back first: time cost: %" PRIu64 "", el1);
// et = taosGetTimestampUs();
//(pRpc->cfp)(NULL, &rpcMsg, pEpSet);
// uint64_t el2 = taosGetTimestampUs() - et;
// tError("msg tran back second: time cost: %" PRIu64 "", el2);
}
}
static void clientAsyncCb(uv_async_t* handle) {
......@@ -204,7 +193,8 @@ static void clientAsyncCb(uv_async_t* handle) {
while (!QUEUE_IS_EMPTY(&wq)) {
queue* h = QUEUE_HEAD(&wq);
QUEUE_REMOVE(h);
pMsg = QUEUE_DATA(h, SCliMsg, q);
SCliMsg* pMsg = QUEUE_DATA(h, SCliMsg, q);
clientHandleReq(pMsg, pThrd);
count++;
if (count >= 2) {
......@@ -220,6 +210,7 @@ static void* clientThread(void* arg) {
void* taosInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads, void* fp, void* shandle) {
SClientObj* cli = calloc(1, sizeof(SClientObj));
memcpy(cli->label, label, strlen(label));
cli->numOfThreads = numOfThreads;
cli->pThreadObj = (SCliThrdObj**)calloc(cli->numOfThreads, sizeof(SCliThrdObj*));
......@@ -244,22 +235,31 @@ void* taosInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads,
}
return cli;
}
void taosCloseClient(void* arg) {
// impl later
SClientObj* cli = arg;
}
void rpcSendRequest(void* shandle, const SEpSet* pEpSet, SRpcMsg* pMsg, int64_t* pRid) {
// impl later
char* ip = (char*)(pEpSet->fqdn[pEpSet->inUse]);
uint32_t port = pEpSet->port[pEpSet->inUse];
SRpcInfo* pRpc = (SRpcInfo*)shandle;
int len = rpcCompressRpcMsg(pMsg->pCont, pMsg->contLen);
SRpcReqContext* pContext;
pContext = (SRpcReqContext*)((char*)pMsg->pCont - sizeof(SRpcHead) - sizeof(SRpcReqContext));
pContext->ahandle = pMsg->ahandle;
pContext->pRpc = (SRpcInfo*)shandle;
pContext->epSet = *pEpSet;
pContext->contLen = len;
pContext->pCont = pMsg->pCont;
pContext->msgType = pMsg->msgType;
pContext->oldInUse = pEpSet->inUse;
STransConnCtx* pCtx = calloc(1, sizeof(STransConnCtx));
pCtx->pRpc = (SRpcInfo*)shandle;
pCtx->ahandle = pMsg->ahandle;
// pContext->contLen = len;
// pContext->pCont = pMsg->pCont;
pCtx->msgType = pMsg->msgType;
pCtx->ip = strdup(ip);
pCtx->port = port;
// pContext->epSet = *pEpSet;
// pContext->oldInUse = pEpSet->inUse;
assert(pRpc->connType == TAOS_CONN_CLIENT);
// atomic or not
......@@ -267,14 +267,15 @@ void rpcSendRequest(void* shandle, const SEpSet* pEpSet, SRpcMsg* pMsg, int64_t*
if (pRpc->index++ >= pRpc->numOfThreads) {
pRpc->index = 0;
}
SCliMsg* msg = malloc(sizeof(SCliMsg));
msg->context = pContext;
msg->st = taosGetTimestampUs();
SCliMsg* cliMsg = malloc(sizeof(SCliMsg));
cliMsg->ctx = pCtx;
cliMsg->msg = *pMsg;
cliMsg->st = taosGetTimestampUs();
SCliThrdObj* thrd = ((SClientObj*)pRpc->tcphandle)->pThreadObj[index % pRpc->numOfThreads];
pthread_mutex_lock(&thrd->msgMtx);
QUEUE_PUSH(&thrd->msg, &msg->q);
QUEUE_PUSH(&thrd->msg, &cliMsg->q);
pthread_mutex_unlock(&thrd->msgMtx);
uv_async_send(thrd->cliAsync);
......
source/libs/transport/src/transComm.c
浏览文件 @ c063ea69
......@@ -30,6 +30,20 @@ int rpcAuthenticateMsg(void* pMsg, int msgLen, void* pAuth, void* pKey) {
return ret;
}
int transAuthenticateMsg(void* pMsg, int msgLen, void* pAuth, void* pKey) {
T_MD5_CTX context;
int ret = -1;
tMD5Init(&context);
tMD5Update(&context, (uint8_t*)pKey, TSDB_PASSWORD_LEN);
tMD5Update(&context, (uint8_t*)pMsg, msgLen);
tMD5Update(&context, (uint8_t*)pKey, TSDB_PASSWORD_LEN);
tMD5Final(&context);
if (memcmp(context.digest, pAuth, sizeof(context.digest)) == 0) ret = 0;
return ret;
}
void rpcBuildAuthHead(void* pMsg, int msgLen, void* pAuth, void* pKey) {
T_MD5_CTX context;
......@@ -41,6 +55,17 @@ void rpcBuildAuthHead(void* pMsg, int msgLen, void* pAuth, void* pKey) {
memcpy(pAuth, context.digest, sizeof(context.digest));
}
void transBuildAuthHead(void* pMsg, int msgLen, void* pAuth, void* pKey) {
T_MD5_CTX context;
tMD5Init(&context);
tMD5Update(&context, (uint8_t*)pKey, TSDB_PASSWORD_LEN);
tMD5Update(&context, (uint8_t*)pMsg, msgLen);
tMD5Update(&context, (uint8_t*)pKey, TSDB_PASSWORD_LEN);
tMD5Final(&context);
memcpy(pAuth, context.digest, sizeof(context.digest));
}
int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen) {
SRpcHead* pHead = rpcHeadFromCont(pCont);
......@@ -81,6 +106,53 @@ int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen) {
return finalLen;
}
bool transCompressMsg(char* msg, int32_t len, int32_t* flen) {
// SRpcHead* pHead = rpcHeadFromCont(pCont);
bool succ = false;
int overhead = sizeof(STransCompMsg);
if (!NEEDTO_COMPRESSS_MSG(len)) {
return succ;
}
char* buf = malloc(len + overhead + 8); // 8 extra bytes
if (buf == NULL) {
tError("failed to allocate memory for rpc msg compression, contLen:%d", len);
*flen = len;
return succ;
}
int32_t clen = LZ4_compress_default(msg, buf, len, len + overhead);
tDebug("compress rpc msg, before:%d, after:%d, overhead:%d", len, clen, overhead);
/*
* only the compressed size is less than the value of contLen - overhead, the compression is applied
* The first four bytes is set to 0, the second four bytes are utilized to keep the original length of message
*/
if (clen > 0 && clen < len - overhead) {
STransCompMsg* pComp = (STransCompMsg*)msg;
pComp->reserved = 0;
pComp->contLen = htonl(len);
memcpy(msg + overhead, buf, clen);
tDebug("compress rpc msg, before:%d, after:%d", len, clen);
*flen = clen + overhead;
succ = true;
} else {
*flen = len;
succ = false;
}
free(buf);
return succ;
}
bool transDecompressMsg(char* msg, int32_t len, int32_t* flen) {
// impl later
return false;
STransCompMsg* pComp = (STransCompMsg*)msg;
int overhead = sizeof(STransCompMsg);
int clen = 0;
return false;
}
SRpcHead* rpcDecompressRpcMsg(SRpcHead* pHead) {
int overhead = sizeof(SRpcComp);
SRpcHead* pNewHead = NULL;
......
source/libs/transport/src/transSrv.c
浏览文件 @ c063ea69
......@@ -107,24 +107,24 @@ void uvAllocReadBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* b
SConn* conn = handle->data;
SConnBuffer* pBuf = &conn->connBuf;
if (pBuf->cap == 0) {
pBuf->buf = (char*)calloc(CAPACITY + RPC_RESERVE_SIZE, sizeof(char));
pBuf->buf = (char*)calloc(CAPACITY, sizeof(char));
pBuf->len = 0;
pBuf->cap = CAPACITY;
pBuf->left = -1;
buf->base = pBuf->buf + RPC_RESERVE_SIZE;
buf->base = pBuf->buf;
buf->len = CAPACITY;
} else {
if (pBuf->len >= pBuf->cap) {
if (pBuf->left == -1) {
pBuf->cap *= 2;
pBuf->buf = realloc(pBuf->buf, pBuf->cap + RPC_RESERVE_SIZE);
pBuf->buf = realloc(pBuf->buf, pBuf->cap);
} else if (pBuf->len + pBuf->left > pBuf->cap) {
pBuf->cap = pBuf->len + pBuf->left;
pBuf->buf = realloc(pBuf->buf, pBuf->len + pBuf->left + RPC_RESERVE_SIZE);
pBuf->buf = realloc(pBuf->buf, pBuf->len + pBuf->left);
}
}
buf->base = pBuf->buf + pBuf->len + RPC_RESERVE_SIZE;
buf->base = pBuf->buf + pBuf->len;
buf->len = pBuf->cap - pBuf->len;
}
}
......@@ -133,11 +133,12 @@ void uvAllocReadBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* b
//
static bool readComplete(SConnBuffer* data) {
// TODO(yihao): handle pipeline later
SRpcHead rpcHead;
int32_t headLen = sizeof(rpcHead);
// SRpcHead rpcHead;
STransMsgHead head;
int32_t headLen = sizeof(head);
if (data->len >= headLen) {
memcpy((char*)&rpcHead, data->buf + RPC_RESERVE_SIZE, headLen);
int32_t msgLen = (int32_t)htonl((uint32_t)rpcHead.msgLen);
memcpy((char*)&head, data->buf, headLen);
int32_t msgLen = (int32_t)htonl((uint32_t)head.msgLen);
if (msgLen > data->len) {
data->left = msgLen - data->len;
return false;
......@@ -150,21 +151,21 @@ static bool readComplete(SConnBuffer* data) {
}
static void uvDoProcess(SRecvInfo* pRecv) {
SRpcHead* pHead = (SRpcHead*)pRecv->msg;
SRpcInfo* pRpc = (SRpcInfo*)pRecv->shandle;
SConn* pConn = pRecv->thandle;
// impl later
STransMsgHead* pHead = (STransMsgHead*)pRecv->msg;
SRpcInfo* pRpc = (SRpcInfo*)pRecv->shandle;
SConn* pConn = pRecv->thandle;
tDump(pRecv->msg, pRecv->msgLen);
terrno = 0;
SRpcReqContext* pContest;
// SRpcReqContext* pContest;
// do auth and check
}
static int uvAuthMsg(SConn* pConn, char* msg, int len) {
SRpcHead* pHead = (SRpcHead*)msg;
int code = 0;
STransMsgHead* pHead = (STransMsgHead*)msg;
int code = 0;
if ((pConn->secured && pHead->spi == 0) || (pHead->spi == 0 && pConn->spi == 0)) {
// secured link, or no authentication
......@@ -224,7 +225,7 @@ static void uvProcessData(SConn* pConn) {
SRecvInfo info;
SRecvInfo* p = &info;
SConnBuffer* pBuf = &pConn->connBuf;
p->msg = pBuf->buf + RPC_RESERVE_SIZE;
p->msg = pBuf->buf;
p->msgLen = pBuf->len;
p->ip = 0;
p->port = 0;
......@@ -233,11 +234,10 @@ static void uvProcessData(SConn* pConn) {
p->chandle = NULL;
//
SRpcHead* pHead = (SRpcHead*)p->msg;
assert(rpcIsReq(pHead->msgType));
STransMsgHead* pHead = (STransMsgHead*)p->msg;
assert(transIsReq(pHead->msgType));
SRpcInfo* pRpc = (SRpcInfo*)p->shandle;
pConn->ahandle = (void*)pHead->ahandle;
// auth here
int8_t code = uvAuthMsg(pConn, (char*)pHead, p->msgLen);
......@@ -247,14 +247,19 @@ static void uvProcessData(SConn* pConn) {
}
pHead->code = htonl(pHead->code);
int32_t dlen = 0;
SRpcMsg rpcMsg;
pHead = rpcDecompressRpcMsg(pHead);
if (transDecompressMsg(NULL, 0, NULL)) {
// add compress later
// pHead = rpcDecompressRpcMsg(pHead);
} else {
// impl later
}
rpcMsg.contLen = rpcContLenFromMsg(pHead->msgLen);
rpcMsg.pCont = pHead->content;
rpcMsg.msgType = pHead->msgType;
rpcMsg.code = pHead->code;
rpcMsg.ahandle = pConn->ahandle;
rpcMsg.ahandle = NULL;
rpcMsg.handle = pConn;
(*(pRpc->cfp))(pRpc->parent, &rpcMsg, NULL);
......@@ -423,7 +428,7 @@ void* workerThread(void* arg) {
uv_loop_init(pThrd->loop);
// SRpcInfo* pRpc = pThrd->shandle;
uv_pipe_init(pThrd->loop, pThrd->pipe, 1);
uv_pipe_init(pThrd->loop, pThrd->pipe, 0);
uv_pipe_open(pThrd->pipe, pThrd->fd);
pThrd->pipe->data = pThrd;
......@@ -522,6 +527,10 @@ void* taosInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
return srv;
}
void taosCloseServer(void* arg) {
// impl later
SServerObj* srv = arg;
}
void rpcSendResponse(const SRpcMsg* pMsg) {
SConn* pConn = pMsg->handle;
......
source/libs/transport/test/uv.c 0 → 100644
浏览文件 @ c063ea69
#include <assert.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "task.h"
#include <uv.h>
#define NUM_OF_THREAD 1
#define TIMEOUT 10000
typedef struct SThreadObj {
pthread_t thread;
uv_pipe_t *pipe;
uv_loop_t *loop;
uv_async_t *workerAsync; //
int fd;
} SThreadObj;
typedef struct SServerObj {
uv_tcp_t server;
uv_loop_t *loop;
int workerIdx;
int numOfThread;
SThreadObj **pThreadObj;
uv_pipe_t **pipe;
} SServerObj;
typedef struct SConnCtx {
uv_tcp_t *pClient;
uv_timer_t *pTimer;
uv_async_t *pWorkerAsync;
int ref;
} SConnCtx;
void echo_write(uv_write_t *req, int status) {
if (status < 0) {
fprintf(stderr, "Write error %s\n", uv_err_name(status));
}
printf("write data to client\n");
free(req);
}
void echo_read(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) {
SConnCtx *pConn = container_of(client, SConnCtx, pClient);
pConn->ref += 1;
printf("read data %d\n", nread, buf->base, buf->len);
if (nread > 0) {
uv_write_t *req = (uv_write_t *)malloc(sizeof(uv_write_t));
// dispatch request to database other process thread
// just write out
uv_buf_t write_out;
write_out.base = buf->base;
write_out.len = nread;
uv_write((uv_write_t *)req, client, &write_out, 1, echo_write);
free(buf->base);
return;
}
if (nread < 0) {
if (nread != UV_EOF)
fprintf(stderr, "Read error %s\n", uv_err_name(nread));
uv_close((uv_handle_t *)client, NULL);
}
free(buf->base);
}
void alloc_buffer(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf) {
buf->base = malloc(suggested_size);
buf->len = suggested_size;
}
void on_new_connection(uv_stream_t *s, int status) {
if (status == -1) {
// error!
return;
}
SServerObj *pObj = container_of(s, SServerObj, server);
printf("new_connection from client\n");
uv_tcp_t *client = (uv_tcp_t *)malloc(sizeof(uv_tcp_t));
uv_tcp_init(pObj->loop, client);
if (uv_accept(s, (uv_stream_t *)client) == 0) {
uv_write_t *write_req = (uv_write_t *)malloc(sizeof(uv_write_t));
uv_buf_t dummy_buf = uv_buf_init("a", 1);
// despatch to worker thread
pObj->workerIdx = (pObj->workerIdx + 1) % pObj->numOfThread;
uv_write2(write_req, (uv_stream_t *)&(pObj->pipe[pObj->workerIdx][0]),
&dummy_buf, 1, (uv_stream_t *)client, echo_write);
} else {
uv_close((uv_handle_t *)client, NULL);
}
}
void child_on_new_connection(uv_stream_t *q, ssize_t nread,
const uv_buf_t *buf) {
printf("x child_on_new_connection \n");
if (nread < 0) {
if (nread != UV_EOF)
fprintf(stderr, "Read error %s\n", uv_err_name(nread));
uv_close((uv_handle_t *)q, NULL);
return;
}
SThreadObj *pObj = (SThreadObj *)container_of(q, struct SThreadObj, pipe);
uv_pipe_t *pipe = (uv_pipe_t *)q;
if (!uv_pipe_pending_count(pipe)) {
fprintf(stderr, "No pending count\n");
return;
}
uv_handle_type pending = uv_pipe_pending_type(pipe);
assert(pending == UV_TCP);
SConnCtx *pConn = malloc(sizeof(SConnCtx));
/* init conn timer*/
pConn->pTimer = malloc(sizeof(uv_timer_t));
uv_timer_init(pObj->loop, pConn->pTimer);
pConn->pClient = (uv_tcp_t *)malloc(sizeof(uv_tcp_t));
pConn->pWorkerAsync = pObj->workerAsync; // thread safty
uv_tcp_init(pObj->loop, pConn->pClient);
if (uv_accept(q, (uv_stream_t *)(pConn->pClient)) == 0) {
uv_os_fd_t fd;
uv_fileno((const uv_handle_t *)pConn->pClient, &fd);
fprintf(stderr, "Worker Accepted fd %d\n", fd);
uv_timer_start(pConn->pTimer, timeOutCallBack, TIMEOUT, 0);
uv_read_start((uv_stream_t *)(pConn->pClient), alloc_buffer, echo_read);
} else {
uv_timer_stop(pConn->pTimer);
free(pConn->pTimer);
uv_close((uv_handle_t *)pConn->pClient, NULL);
free(pConn->pClient);
free(pConn);
}
}
static void workerAsyncCallback(uv_async_t *handle) {
SThreadObj *pObj = container_of(handle, SThreadObj, workerAsync);
// do nothing
}
void *worker_thread(void *arg) {
SThreadObj *pObj = (SThreadObj *)arg;
int fd = pObj->fd;
pObj->loop = (uv_loop_t *)malloc(sizeof(uv_loop_t));
uv_loop_init(pObj->loop);
uv_pipe_init(pObj->loop, pObj->pipe, 1);
uv_pipe_open(pObj->pipe, fd);
pObj->workerAsync = malloc(sizeof(uv_async_t));
uv_async_init(pObj->loop, pObj->workerAsync, workerAsyncCallback);
uv_read_start((uv_stream_t *)pObj->pipe, alloc_buffer,
child_on_new_connection);
uv_run(pObj->loop, UV_RUN_DEFAULT);
}
int main() {
SServerObj *server = calloc(1, sizeof(SServerObj));
server->loop = (uv_loop_t *)malloc(sizeof(uv_loop_t));
server->numOfThread = NUM_OF_THREAD;
server->workerIdx = 0;
server->pThreadObj =
(SThreadObj **)calloc(server->numOfThread, sizeof(SThreadObj *));
server->pipe = (uv_pipe_t **)calloc(server->numOfThread, sizeof(uv_pipe_t *));
uv_loop_init(server->loop);
for (int i = 0; i < server->numOfThread; i++) {
server->pThreadObj[i] = (SThreadObj *)calloc(1, sizeof(SThreadObj));
server->pipe[i] = (uv_pipe_t *)calloc(2, sizeof(uv_pipe_t));
int fds[2];
if (uv_socketpair(AF_UNIX, SOCK_STREAM, fds, UV_NONBLOCK_PIPE,
UV_NONBLOCK_PIPE) != 0) {
return -1;
}
uv_pipe_init(server->loop, &(server->pipe[i][0]), 1);
uv_pipe_open(&(server->pipe[i][0]), fds[1]); // init write
server->pThreadObj[i]->fd = fds[0];
server->pThreadObj[i]->pipe = &(server->pipe[i][1]); // init read
int err = pthread_create(&(server->pThreadObj[i]->thread), NULL,
worker_thread, (void *)(server->pThreadObj[i]));
if (err == 0) {
printf("thread %d create\n", i);
} else {
printf("thread %d create failed", i);
}
uv_tcp_init(server->loop, &server->server);
struct sockaddr_in bind_addr;
uv_ip4_addr("0.0.0.0", 7000, &bind_addr);
uv_tcp_bind(&server->server, (const struct sockaddr *)&bind_addr, 0);
int err = 0;
if ((err = uv_listen((uv_stream_t *)&server->server, 128,
on_new_connection)) != 0) {
fprintf(stderr, "Listen error %s\n", uv_err_name(err));
return 2;
}
uv_run(server->loop, UV_RUN_DEFAULT);
return 0;
}
}
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或