提交 c166bcff 编写于 作者: dengyihao's avatar dengyihao

enh: refactor trans code

上级 e7fe3577
...@@ -76,7 +76,6 @@ void* rpcOpen(const SRpcInit* pInit) { ...@@ -76,7 +76,6 @@ void* rpcOpen(const SRpcInit* pInit) {
if (pInit->user) { if (pInit->user) {
memcpy(pRpc->user, pInit->user, strlen(pInit->user)); memcpy(pRpc->user, pInit->user, strlen(pInit->user));
} }
// pRpc->refMgt = transOpenExHandleMgt(50000);
return pRpc; return pRpc;
} }
void rpcClose(void* arg) { void rpcClose(void* arg) {
......
...@@ -15,6 +15,9 @@ ...@@ -15,6 +15,9 @@
#ifdef USE_UV #ifdef USE_UV
#include "transComm.h" #include "transComm.h"
static int32_t transSCliInst = 0;
static int32_t refMgt = 0;
typedef struct SCliConn { typedef struct SCliConn {
T_REF_DECLARE() T_REF_DECLARE()
uv_connect_t connReq; uv_connect_t connReq;
...@@ -846,6 +849,11 @@ void* transInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads, ...@@ -846,6 +849,11 @@ void* transInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads,
} }
cli->pThreadObj[i] = pThrd; cli->pThreadObj[i] = pThrd;
} }
int ref = atomic_add_fetch_32(&transSCliInst, 1);
if (ref == 1) {
refMgt = transOpenExHandleMgt(50000);
}
return cli; return cli;
} }
...@@ -1019,6 +1027,10 @@ void transCloseClient(void* arg) { ...@@ -1019,6 +1027,10 @@ void transCloseClient(void* arg) {
} }
taosMemoryFree(cli->pThreadObj); taosMemoryFree(cli->pThreadObj);
taosMemoryFree(cli); taosMemoryFree(cli);
int ref = atomic_sub_fetch_32(&transSCliInst, 1);
if (ref == 0) {
transCloseExHandleMgt(refMgt);
}
} }
void transRefCliHandle(void* handle) { void transRefCliHandle(void* handle) {
if (handle == NULL) { if (handle == NULL) {
......
...@@ -472,8 +472,8 @@ bool transEpSetIsEqual(SEpSet* a, SEpSet* b) { ...@@ -472,8 +472,8 @@ bool transEpSetIsEqual(SEpSet* a, SEpSet* b) {
} }
void transInitEnv() { void transInitEnv() {
//
uv_os_setenv("UV_TCP_SINGLE_ACCEPT", "1"); uv_os_setenv("UV_TCP_SINGLE_ACCEPT", "1");
// uvOpenExHandleMgt(10000);
} }
int32_t transOpenExHandleMgt(int size) { int32_t transOpenExHandleMgt(int size) {
// added into once later // added into once later
......
...@@ -20,7 +20,7 @@ ...@@ -20,7 +20,7 @@
static TdThreadOnce transModuleInit = PTHREAD_ONCE_INIT; static TdThreadOnce transModuleInit = PTHREAD_ONCE_INIT;
static char* notify = "a"; static char* notify = "a";
static int tranSSvrInst = 0; static int32_t tranSSvrInst = 0;
static int32_t refMgt = 0; static int32_t refMgt = 0;
typedef struct { typedef struct {
...@@ -878,8 +878,8 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads, ...@@ -878,8 +878,8 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
uv_loop_init(srv->loop); uv_loop_init(srv->loop);
// taosThreadOnce(&transModuleInit, uvInitEnv); // taosThreadOnce(&transModuleInit, uvInitEnv);
tranSSvrInst++; int ref = atomic_add_fetch_32(&tranSSvrInst, 1);
if (tranSSvrInst == 1) { if (ref == 1) {
refMgt = transOpenExHandleMgt(50000); refMgt = transOpenExHandleMgt(50000);
} }
...@@ -941,43 +941,6 @@ End: ...@@ -941,43 +941,6 @@ End:
return NULL; return NULL;
} }
// void uvInitEnv() {
// uv_os_setenv("UV_TCP_SINGLE_ACCEPT", "1");
// uvOpenExHandleMgt(10000);
//}
// void uvOpenExHandleMgt(int size) {
// // added into once later
// exHandlesMgt = taosOpenRef(size, uvDestoryExHandle);
//}
// void uvCloseExHandleMgt() {
// // close ref
// taosCloseRef(exHandlesMgt);
//}
// int64_t uvAddExHandle(void* p) {
// // acquire extern handle
// return taosAddRef(exHandlesMgt, p);
//}
// int32_t uvRemoveExHandle(int64_t refId) {
// // acquire extern handle
// return taosRemoveRef(exHandlesMgt, refId);
//}
//
// SExHandle* uvAcquireExHandle(int64_t refId) {
// // acquire extern handle
// return (SExHandle*)taosAcquireRef(exHandlesMgt, refId);
//}
//
// int32_t uvReleaseExHandle(int64_t refId) {
// // release extern handle
// return taosReleaseRef(exHandlesMgt, refId);
//}
// void uvDestoryExHandle(void* handle) {
// if (handle == NULL) {
// return;
// }
// taosMemoryFree(handle);
//}
void uvHandleQuit(SSvrMsg* msg, SWorkThrdObj* thrd) { void uvHandleQuit(SSvrMsg* msg, SWorkThrdObj* thrd) {
thrd->quit = true; thrd->quit = true;
if (QUEUE_IS_EMPTY(&thrd->conn)) { if (QUEUE_IS_EMPTY(&thrd->conn)) {
...@@ -1072,8 +1035,8 @@ void transCloseServer(void* arg) { ...@@ -1072,8 +1035,8 @@ void transCloseServer(void* arg) {
taosMemoryFree(srv); taosMemoryFree(srv);
tranSSvrInst--; int ref = atomic_sub_fetch_32(&tranSSvrInst, 1);
if (tranSSvrInst == 0) { if (ref == 0) {
// TdThreadOnce tmpInit = PTHREAD_ONCE_INIT; // TdThreadOnce tmpInit = PTHREAD_ONCE_INIT;
// memcpy(&transModuleInit, &tmpInit, sizeof(TdThreadOnce)); // memcpy(&transModuleInit, &tmpInit, sizeof(TdThreadOnce));
transCloseExHandleMgt(refMgt); transCloseExHandleMgt(refMgt);
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册