Skip to content

T
TDengine

慢慢CG / TDengine
与 Fork 源项目一致

Fork自 taosdata / TDengine

通知 1
Star 0
Fork 0
T
TDengine
提交 4bc9284a 编写于 作者: S Shengliang Guan
浏览文件
操作

TD-1915

上级 048d0a46
隐藏空白更改
内联 并排
Showing with 316 addition and 327 deletion
src/dnode/inc/dnodeMPeer.h
浏览文件 @ 4bc9284a
...@@ -20,11 +20,11 @@ ...@@ -20,11 +20,11 @@
extern "C" { extern "C" {
#endif #endif
int32_t dnodeInitMnodePeer(); int32_t dnodeInitMPeer();
void dnodeCleanupMnodePeer(); void dnodeCleanupMPeer();
int32_t dnodeAllocateMnodePqueue(); int32_t dnodeAllocateMPeerQueue();
void dnodeFreeMnodePqueue(); void dnodeFreeMPeerQueue();
void dnodeDispatchToMnodePeerQueue(SRpcMsg *pMsg); void dnodeDispatchToMPeerQueue(SRpcMsg *pMsg);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
src/dnode/inc/dnodeMRead.h
浏览文件 @ 4bc9284a
...@@ -20,11 +20,11 @@ ...@@ -20,11 +20,11 @@
extern "C" { extern "C" {
#endif #endif
int32_t dnodeInitMnodeRead(); int32_t dnodeInitMRead();
void dnodeCleanupMnodeRead(); void dnodeCleanupMRead();
int32_t dnodeAllocateMnodeRqueue(); int32_t dnodeAllocMReadQueue();
void dnodeFreeMnodeRqueue(); void dnodeFreeMReadQueue();
void dnodeDispatchToMnodeReadQueue(SRpcMsg *rpcMsg); void dnodeDispatchToMReadQueue(SRpcMsg *rpcMsg);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
src/dnode/inc/dnodeMWrite.h
浏览文件 @ 4bc9284a
...@@ -20,11 +20,11 @@ ...@@ -20,11 +20,11 @@
extern "C" { extern "C" {
#endif #endif
int32_t dnodeInitMnodeWrite(); int32_t dnodeInitMWrite();
void dnodeCleanupMnodeWrite(); void dnodeCleanupMWrite();
int32_t dnodeAllocateMnodeWqueue(); int32_t dnodeAllocMWritequeue();
void dnodeFreeMnodeWqueue(); void dnodeFreeMWritequeue();
void dnodeDispatchToMnodeWriteQueue(SRpcMsg *pMsg); void dnodeDispatchToMWriteQueue(SRpcMsg *pMsg);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
src/dnode/src/dnodeMPeer.c
浏览文件 @ 4bc9284a
...@@ -35,44 +35,44 @@ typedef struct { ...@@ -35,44 +35,44 @@ typedef struct {
typedef struct { typedef struct {
int32_t curNum; int32_t curNum;
int32_t maxNum; int32_t maxNum;
SMPeerWorker *peerWorker; SMPeerWorker *worker;
} SMPeerWorkerPool; } SMPeerWorkerPool;
static SMPeerWorkerPool tsMPeerPool; static SMPeerWorkerPool tsMPeerWP;
static taos_qset tsMPeerQset; static taos_qset tsMPeerQset;
static taos_queue tsMPeerQueue; static taos_queue tsMPeerQueue;
static void *dnodeProcessMnodePeerQueue(void *param); static void *dnodeProcessMPeerQueue(void *param);
int32_t dnodeInitMnodePeer() { int32_t dnodeInitMPeer() {
tsMPeerQset = taosOpenQset(); tsMPeerQset = taosOpenQset();
tsMPeerPool.maxNum = 1; tsMPeerWP.maxNum = 1;
tsMPeerPool.curNum = 0; tsMPeerWP.curNum = 0;
tsMPeerPool.peerWorker = (SMPeerWorker *)calloc(sizeof(SMPeerWorker), tsMPeerPool.maxNum); tsMPeerWP.worker = (SMPeerWorker *)calloc(sizeof(SMPeerWorker), tsMPeerWP.maxNum);
if (tsMPeerPool.peerWorker == NULL) return -1; if (tsMPeerWP.worker == NULL) return -1;
for (int32_t i = 0; i < tsMPeerPool.maxNum; ++i) { for (int32_t i = 0; i < tsMPeerWP.maxNum; ++i) {
SMPeerWorker *pWorker = tsMPeerPool.peerWorker + i; SMPeerWorker *pWorker = tsMPeerWP.worker + i;
pWorker->workerId = i; pWorker->workerId = i;
dDebug("dnode mpeer worker:%d is created", i); dDebug("dnode mpeer worker:%d is created", i);
} }
dDebug("dnode mpeer is initialized, workers:%d qset:%p", tsMPeerPool.maxNum, tsMPeerQset); dDebug("dnode mpeer is initialized, workers:%d qset:%p", tsMPeerWP.maxNum, tsMPeerQset);
return 0; return 0;
} }
void dnodeCleanupMnodePeer() { void dnodeCleanupMPeer() {
for (int32_t i = 0; i < tsMPeerPool.maxNum; ++i) { for (int32_t i = 0; i < tsMPeerWP.maxNum; ++i) {
SMPeerWorker *pWorker = tsMPeerPool.peerWorker + i; SMPeerWorker *pWorker = tsMPeerWP.worker + i;
if (pWorker->thread) { if (pWorker->thread) {
taosQsetThreadResume(tsMPeerQset); taosQsetThreadResume(tsMPeerQset);
} }
dDebug("dnode mpeer worker:%d is closed", i); dDebug("dnode mpeer worker:%d is closed", i);
} }
for (int32_t i = 0; i < tsMPeerPool.maxNum; ++i) { for (int32_t i = 0; i < tsMPeerWP.maxNum; ++i) {
SMPeerWorker *pWorker = tsMPeerPool.peerWorker + i; SMPeerWorker *pWorker = tsMPeerWP.worker + i;
dDebug("dnode mpeer worker:%d start to join", i); dDebug("dnode mpeer worker:%d start to join", i);
if (pWorker->thread) { if (pWorker->thread) {
pthread_join(pWorker->thread, NULL); pthread_join(pWorker->thread, NULL);
...@@ -84,44 +84,44 @@ void dnodeCleanupMnodePeer() { ...@@ -84,44 +84,44 @@ void dnodeCleanupMnodePeer() {
taosCloseQset(tsMPeerQset); taosCloseQset(tsMPeerQset);
tsMPeerQset = NULL; tsMPeerQset = NULL;
taosTFree(tsMPeerPool.peerWorker); taosTFree(tsMPeerWP.worker);
} }
int32_t dnodeAllocateMnodePqueue() { int32_t dnodeAllocateMPeerQueue() {
tsMPeerQueue = taosOpenQueue(); tsMPeerQueue = taosOpenQueue();
if (tsMPeerQueue == NULL) return TSDB_CODE_DND_OUT_OF_MEMORY; if (tsMPeerQueue == NULL) return TSDB_CODE_DND_OUT_OF_MEMORY;
taosAddIntoQset(tsMPeerQset, tsMPeerQueue, NULL); taosAddIntoQset(tsMPeerQset, tsMPeerQueue, NULL);
for (int32_t i = tsMPeerPool.curNum; i < tsMPeerPool.maxNum; ++i) { for (int32_t i = tsMPeerWP.curNum; i < tsMPeerWP.maxNum; ++i) {
SMPeerWorker *pWorker = tsMPeerPool.peerWorker + i; SMPeerWorker *pWorker = tsMPeerWP.worker + i;
pWorker->workerId = i; pWorker->workerId = i;
pthread_attr_t thAttr; pthread_attr_t thAttr;
pthread_attr_init(&thAttr); pthread_attr_init(&thAttr);
pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE); pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE);
if (pthread_create(&pWorker->thread, &thAttr, dnodeProcessMnodePeerQueue, pWorker) != 0) { if (pthread_create(&pWorker->thread, &thAttr, dnodeProcessMPeerQueue, pWorker) != 0) {
dError("failed to create thread to process mpeer queue, reason:%s", strerror(errno)); dError("failed to create thread to process mpeer queue, reason:%s", strerror(errno));
} }
pthread_attr_destroy(&thAttr); pthread_attr_destroy(&thAttr);
tsMPeerPool.curNum = i + 1; tsMPeerWP.curNum = i + 1;
dDebug("dnode mpeer worker:%d is launched, total:%d", pWorker->workerId, tsMPeerPool.maxNum); dDebug("dnode mpeer worker:%d is launched, total:%d", pWorker->workerId, tsMPeerWP.maxNum);
} }
dDebug("dnode mpeer queue:%p is allocated", tsMPeerQueue); dDebug("dnode mpeer queue:%p is allocated", tsMPeerQueue);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
void dnodeFreeMnodePqueue() { void dnodeFreeMPeerQueue() {
dDebug("dnode mpeer queue:%p is freed", tsMPeerQueue); dDebug("dnode mpeer queue:%p is freed", tsMPeerQueue);
taosCloseQueue(tsMPeerQueue); taosCloseQueue(tsMPeerQueue);
tsMPeerQueue = NULL; tsMPeerQueue = NULL;
} }
void dnodeDispatchToMnodePeerQueue(SRpcMsg *pMsg) { void dnodeDispatchToMPeerQueue(SRpcMsg *pMsg) {
if (!mnodeIsRunning() || tsMPeerQueue == NULL) { if (!mnodeIsRunning() || tsMPeerQueue == NULL) {
dnodeSendRedirectMsg(pMsg, false); dnodeSendRedirectMsg(pMsg, false);
rpcFreeCont(pMsg->pCont); rpcFreeCont(pMsg->pCont);
...@@ -133,12 +133,12 @@ void dnodeDispatchToMnodePeerQueue(SRpcMsg *pMsg) { ...@@ -133,12 +133,12 @@ void dnodeDispatchToMnodePeerQueue(SRpcMsg *pMsg) {
taosWriteQitem(tsMPeerQueue, TAOS_QTYPE_RPC, pPeer); taosWriteQitem(tsMPeerQueue, TAOS_QTYPE_RPC, pPeer);
} }
static void dnodeFreeMnodePeerMsg(SMnodeMsg *pPeer) { static void dnodeFreeMPeerMsg(SMnodeMsg *pPeer) {
mnodeCleanupMsg(pPeer); mnodeCleanupMsg(pPeer);
taosFreeQitem(pPeer); taosFreeQitem(pPeer);
} }
static void dnodeSendRpcMnodePeerRsp(SMnodeMsg *pPeer, int32_t code) { static void dnodeSendRpcMPeerRsp(SMnodeMsg *pPeer, int32_t code) {
if (code == TSDB_CODE_MND_ACTION_IN_PROGRESS) return; if (code == TSDB_CODE_MND_ACTION_IN_PROGRESS) return;
SRpcMsg rpcRsp = { SRpcMsg rpcRsp = {
...@@ -149,10 +149,10 @@ static void dnodeSendRpcMnodePeerRsp(SMnodeMsg *pPeer, int32_t code) { ...@@ -149,10 +149,10 @@ static void dnodeSendRpcMnodePeerRsp(SMnodeMsg *pPeer, int32_t code) {
}; };
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
dnodeFreeMnodePeerMsg(pPeer); dnodeFreeMPeerMsg(pPeer);
} }
static void *dnodeProcessMnodePeerQueue(void *param) { static void *dnodeProcessMPeerQueue(void *param) {
SMnodeMsg *pPeerMsg; SMnodeMsg *pPeerMsg;
int32_t type; int32_t type;
void * unUsed; void * unUsed;
...@@ -165,7 +165,7 @@ static void *dnodeProcessMnodePeerQueue(void *param) { ...@@ -165,7 +165,7 @@ static void *dnodeProcessMnodePeerQueue(void *param) {
dDebug("msg:%s will be processed in mpeer queue", taosMsg[pPeerMsg->rpcMsg.msgType]); dDebug("msg:%s will be processed in mpeer queue", taosMsg[pPeerMsg->rpcMsg.msgType]);
int32_t code = mnodeProcessPeerReq(pPeerMsg); int32_t code = mnodeProcessPeerReq(pPeerMsg);
dnodeSendRpcMnodePeerRsp(pPeerMsg, code); dnodeSendRpcMPeerRsp(pPeerMsg, code);
} }
return NULL; return NULL;
......
src/dnode/src/dnodeMRead.c
浏览文件 @ 4bc9284a
...@@ -35,46 +35,46 @@ typedef struct { ...@@ -35,46 +35,46 @@ typedef struct {
typedef struct { typedef struct {
int32_t curNum; int32_t curNum;
int32_t maxNum; int32_t maxNum;
SMReadWorker *readWorker; SMReadWorker *worker;
} SMReadWorkerPool; } SMReadWorkerPool;
static SMReadWorkerPool tsMReadPool; static SMReadWorkerPool tsMReadWP;
static taos_qset tsMReadQset; static taos_qset tsMReadQset;
static taos_queue tsMReadQueue; static taos_queue tsMReadQueue;
static void *dnodeProcessMnodeReadQueue(void *param); static void *dnodeProcessMReadQueue(void *param);
int32_t dnodeInitMnodeRead() { int32_t dnodeInitMRead() {
tsMReadQset = taosOpenQset(); tsMReadQset = taosOpenQset();
tsMReadPool.maxNum = tsNumOfCores * tsNumOfThreadsPerCore / 2; tsMReadWP.maxNum = tsNumOfCores * tsNumOfThreadsPerCore / 2;
tsMReadPool.maxNum = MAX(2, tsMReadPool.maxNum); tsMReadWP.maxNum = MAX(2, tsMReadWP.maxNum);
tsMReadPool.maxNum = MIN(4, tsMReadPool.maxNum); tsMReadWP.maxNum = MIN(4, tsMReadWP.maxNum);
tsMReadPool.curNum = 0; tsMReadWP.curNum = 0;
tsMReadPool.readWorker = (SMReadWorker *)calloc(sizeof(SMReadWorker), tsMReadPool.maxNum); tsMReadWP.worker = (SMReadWorker *)calloc(sizeof(SMReadWorker), tsMReadWP.maxNum);
if (tsMReadPool.readWorker == NULL) return -1; if (tsMReadWP.worker == NULL) return -1;
for (int32_t i = 0; i < tsMReadPool.maxNum; ++i) { for (int32_t i = 0; i < tsMReadWP.maxNum; ++i) {
SMReadWorker *pWorker = tsMReadPool.readWorker + i; SMReadWorker *pWorker = tsMReadWP.worker + i;
pWorker->workerId = i; pWorker->workerId = i;
dDebug("dnode mread worker:%d is created", i); dDebug("dnode mread worker:%d is created", i);
} }
dDebug("dnode mread is initialized, workers:%d qset:%p", tsMReadPool.maxNum, tsMReadQset); dDebug("dnode mread is initialized, workers:%d qset:%p", tsMReadWP.maxNum, tsMReadQset);
return 0; return 0;
} }
void dnodeCleanupMnodeRead() { void dnodeCleanupMRead() {
for (int32_t i = 0; i < tsMReadPool.maxNum; ++i) { for (int32_t i = 0; i < tsMReadWP.maxNum; ++i) {
SMReadWorker *pWorker = tsMReadPool.readWorker + i; SMReadWorker *pWorker = tsMReadWP.worker + i;
if (pWorker->thread) { if (pWorker->thread) {
taosQsetThreadResume(tsMReadQset); taosQsetThreadResume(tsMReadQset);
} }
dDebug("dnode mread worker:%d is closed", i); dDebug("dnode mread worker:%d is closed", i);
} }
for (int32_t i = 0; i < tsMReadPool.maxNum; ++i) { for (int32_t i = 0; i < tsMReadWP.maxNum; ++i) {
SMReadWorker *pWorker = tsMReadPool.readWorker + i; SMReadWorker *pWorker = tsMReadWP.worker + i;
dDebug("dnode mread worker:%d start to join", i); dDebug("dnode mread worker:%d start to join", i);
if (pWorker->thread) { if (pWorker->thread) {
pthread_join(pWorker->thread, NULL); pthread_join(pWorker->thread, NULL);
...@@ -86,43 +86,43 @@ void dnodeCleanupMnodeRead() { ...@@ -86,43 +86,43 @@ void dnodeCleanupMnodeRead() {
taosCloseQset(tsMReadQset); taosCloseQset(tsMReadQset);
tsMReadQset = NULL; tsMReadQset = NULL;
free(tsMReadPool.readWorker); free(tsMReadWP.worker);
} }
int32_t dnodeAllocateMnodeRqueue() { int32_t dnodeAllocMReadQueue() {
tsMReadQueue = taosOpenQueue(); tsMReadQueue = taosOpenQueue();
if (tsMReadQueue == NULL) return TSDB_CODE_DND_OUT_OF_MEMORY; if (tsMReadQueue == NULL) return TSDB_CODE_DND_OUT_OF_MEMORY;
taosAddIntoQset(tsMReadQset, tsMReadQueue, NULL); taosAddIntoQset(tsMReadQset, tsMReadQueue, NULL);
for (int32_t i = tsMReadPool.curNum; i < tsMReadPool.maxNum; ++i) { for (int32_t i = tsMReadWP.curNum; i < tsMReadWP.maxNum; ++i) {
SMReadWorker *pWorker = tsMReadPool.readWorker + i; SMReadWorker *pWorker = tsMReadWP.worker + i;
pWorker->workerId = i; pWorker->workerId = i;
pthread_attr_t thAttr; pthread_attr_t thAttr;
pthread_attr_init(&thAttr); pthread_attr_init(&thAttr);
pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE); pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE);
if (pthread_create(&pWorker->thread, &thAttr, dnodeProcessMnodeReadQueue, pWorker) != 0) { if (pthread_create(&pWorker->thread, &thAttr, dnodeProcessMReadQueue, pWorker) != 0) {
dError("failed to create thread to process mread queue, reason:%s", strerror(errno)); dError("failed to create thread to process mread queue, reason:%s", strerror(errno));
} }
pthread_attr_destroy(&thAttr); pthread_attr_destroy(&thAttr);
tsMReadPool.curNum = i + 1; tsMReadWP.curNum = i + 1;
dDebug("dnode mread worker:%d is launched, total:%d", pWorker->workerId, tsMReadPool.maxNum); dDebug("dnode mread worker:%d is launched, total:%d", pWorker->workerId, tsMReadWP.maxNum);
} }
dDebug("dnode mread queue:%p is allocated", tsMReadQueue); dDebug("dnode mread queue:%p is allocated", tsMReadQueue);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
void dnodeFreeMnodeRqueue() { void dnodeFreeMReadQueue() {
dDebug("dnode mread queue:%p is freed", tsMReadQueue); dDebug("dnode mread queue:%p is freed", tsMReadQueue);
taosCloseQueue(tsMReadQueue); taosCloseQueue(tsMReadQueue);
tsMReadQueue = NULL; tsMReadQueue = NULL;
} }
void dnodeDispatchToMnodeReadQueue(SRpcMsg *pMsg) { void dnodeDispatchToMReadQueue(SRpcMsg *pMsg) {
if (!mnodeIsRunning() || tsMReadQueue == NULL) { if (!mnodeIsRunning() || tsMReadQueue == NULL) {
dnodeSendRedirectMsg(pMsg, true); dnodeSendRedirectMsg(pMsg, true);
rpcFreeCont(pMsg->pCont); rpcFreeCont(pMsg->pCont);
...@@ -134,16 +134,16 @@ void dnodeDispatchToMnodeReadQueue(SRpcMsg *pMsg) { ...@@ -134,16 +134,16 @@ void dnodeDispatchToMnodeReadQueue(SRpcMsg *pMsg) {
taosWriteQitem(tsMReadQueue, TAOS_QTYPE_RPC, pRead); taosWriteQitem(tsMReadQueue, TAOS_QTYPE_RPC, pRead);
} }
static void dnodeFreeMnodeReadMsg(SMnodeMsg *pRead) { static void dnodeFreeMReadMsg(SMnodeMsg *pRead) {
mnodeCleanupMsg(pRead); mnodeCleanupMsg(pRead);
taosFreeQitem(pRead); taosFreeQitem(pRead);
} }
static void dnodeSendRpcMnodeReadRsp(SMnodeMsg *pRead, int32_t code) { static void dnodeSendRpcMReadRsp(SMnodeMsg *pRead, int32_t code) {
if (code == TSDB_CODE_MND_ACTION_IN_PROGRESS) return; if (code == TSDB_CODE_MND_ACTION_IN_PROGRESS) return;
if (code == TSDB_CODE_MND_ACTION_NEED_REPROCESSED) { if (code == TSDB_CODE_MND_ACTION_NEED_REPROCESSED) {
// may be a auto create req, should put into write queue // may be a auto create req, should put into write queue
dnodeReprocessMnodeWriteMsg(pRead); dnodeReprocessMWriteMsg(pRead);
return; return;
} }
...@@ -155,23 +155,23 @@ static void dnodeSendRpcMnodeReadRsp(SMnodeMsg *pRead, int32_t code) { ...@@ -155,23 +155,23 @@ static void dnodeSendRpcMnodeReadRsp(SMnodeMsg *pRead, int32_t code) {
}; };
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
dnodeFreeMnodeReadMsg(pRead); dnodeFreeMReadMsg(pRead);
} }
static void *dnodeProcessMnodeReadQueue(void *param) { static void *dnodeProcessMReadQueue(void *param) {
SMnodeMsg *pReadMsg; SMnodeMsg *pRead;
int32_t type; int32_t type;
void * unUsed; void * unUsed;
while (1) { while (1) {
if (taosReadQitemFromQset(tsMReadQset, &type, (void **)&pReadMsg, &unUsed) == 0) { if (taosReadQitemFromQset(tsMReadQset, &type, (void **)&pRead, &unUsed) == 0) {
dDebug("qset:%p, mnode read got no message from qset, exiting", tsMReadQset); dDebug("qset:%p, mnode read got no message from qset, exiting", tsMReadQset);
break; break;
} }
dDebug("%p, msg:%s will be processed in mread queue", pReadMsg->rpcMsg.ahandle, taosMsg[pReadMsg->rpcMsg.msgType]); dDebug("%p, msg:%s will be processed in mread queue", pRead->rpcMsg.ahandle, taosMsg[pRead->rpcMsg.msgType]);
int32_t code = mnodeProcessRead(pReadMsg); int32_t code = mnodeProcessRead(pRead);
dnodeSendRpcMnodeReadRsp(pReadMsg, code); dnodeSendRpcMReadRsp(pRead, code);
} }
return NULL; return NULL;
......
src/dnode/src/dnodeMWrite.c
浏览文件 @ 4bc9284a
...@@ -36,45 +36,45 @@ typedef struct { ...@@ -36,45 +36,45 @@ typedef struct {
typedef struct { typedef struct {
int32_t curNum; int32_t curNum;
int32_t maxNum; int32_t maxNum;
SMWriteWorker *writeWorker; SMWriteWorker *worker;
} SMWriteWorkerPool; } SMWriteWorkerPool;
static SMWriteWorkerPool tsMWritePool; static SMWriteWorkerPool tsMWriteWP;
static taos_qset tsMWriteQset; static taos_qset tsMWriteQset;
static taos_queue tsMWriteQueue; static taos_queue tsMWriteQueue;
extern void * tsDnodeTmr; extern void * tsDnodeTmr;
static void *dnodeProcessMnodeWriteQueue(void *param); static void *dnodeProcessMWriteQueue(void *param);
int32_t dnodeInitMnodeWrite() { int32_t dnodeInitMWrite() {
tsMWriteQset = taosOpenQset(); tsMWriteQset = taosOpenQset();
tsMWritePool.maxNum = 1; tsMWriteWP.maxNum = 1;
tsMWritePool.curNum = 0; tsMWriteWP.curNum = 0;
tsMWritePool.writeWorker = (SMWriteWorker *)calloc(sizeof(SMWriteWorker), tsMWritePool.maxNum); tsMWriteWP.worker = (SMWriteWorker *)calloc(sizeof(SMWriteWorker), tsMWriteWP.maxNum);
if (tsMWritePool.writeWorker == NULL) return -1; if (tsMWriteWP.worker == NULL) return -1;
for (int32_t i = 0; i < tsMWritePool.maxNum; ++i) { for (int32_t i = 0; i < tsMWriteWP.maxNum; ++i) {
SMWriteWorker *pWorker = tsMWritePool.writeWorker + i; SMWriteWorker *pWorker = tsMWriteWP.worker + i;
pWorker->workerId = i; pWorker->workerId = i;
dDebug("dnode mwrite worker:%d is created", i); dDebug("dnode mwrite worker:%d is created", i);
} }
dDebug("dnode mwrite is initialized, workers:%d qset:%p", tsMWritePool.maxNum, tsMWriteQset); dDebug("dnode mwrite is initialized, workers:%d qset:%p", tsMWriteWP.maxNum, tsMWriteQset);
return 0; return 0;
} }
void dnodeCleanupMnodeWrite() { void dnodeCleanupMWrite() {
for (int32_t i = 0; i < tsMWritePool.maxNum; ++i) { for (int32_t i = 0; i < tsMWriteWP.maxNum; ++i) {
SMWriteWorker *pWorker = tsMWritePool.writeWorker + i; SMWriteWorker *pWorker = tsMWriteWP.worker + i;
if (pWorker->thread) { if (pWorker->thread) {
taosQsetThreadResume(tsMWriteQset); taosQsetThreadResume(tsMWriteQset);
} }
dDebug("dnode mwrite worker:%d is closed", i); dDebug("dnode mwrite worker:%d is closed", i);
} }
for (int32_t i = 0; i < tsMWritePool.maxNum; ++i) { for (int32_t i = 0; i < tsMWriteWP.maxNum; ++i) {
SMWriteWorker *pWorker = tsMWritePool.writeWorker + i; SMWriteWorker *pWorker = tsMWriteWP.worker + i;
dDebug("dnode mwrite worker:%d start to join", i); dDebug("dnode mwrite worker:%d start to join", i);
if (pWorker->thread) { if (pWorker->thread) {
pthread_join(pWorker->thread, NULL); pthread_join(pWorker->thread, NULL);
...@@ -86,43 +86,43 @@ void dnodeCleanupMnodeWrite() { ...@@ -86,43 +86,43 @@ void dnodeCleanupMnodeWrite() {
taosCloseQset(tsMWriteQset); taosCloseQset(tsMWriteQset);
tsMWriteQset = NULL; tsMWriteQset = NULL;
taosTFree(tsMWritePool.writeWorker); taosTFree(tsMWriteWP.worker);
} }
int32_t dnodeAllocateMnodeWqueue() { int32_t dnodeAllocMWritequeue() {
tsMWriteQueue = taosOpenQueue(); tsMWriteQueue = taosOpenQueue();
if (tsMWriteQueue == NULL) return TSDB_CODE_DND_OUT_OF_MEMORY; if (tsMWriteQueue == NULL) return TSDB_CODE_DND_OUT_OF_MEMORY;
taosAddIntoQset(tsMWriteQset, tsMWriteQueue, NULL); taosAddIntoQset(tsMWriteQset, tsMWriteQueue, NULL);
for (int32_t i = tsMWritePool.curNum; i < tsMWritePool.maxNum; ++i) { for (int32_t i = tsMWriteWP.curNum; i < tsMWriteWP.maxNum; ++i) {
SMWriteWorker *pWorker = tsMWritePool.writeWorker + i; SMWriteWorker *pWorker = tsMWriteWP.worker + i;
pWorker->workerId = i; pWorker->workerId = i;
pthread_attr_t thAttr; pthread_attr_t thAttr;
pthread_attr_init(&thAttr); pthread_attr_init(&thAttr);
pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE); pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE);
if (pthread_create(&pWorker->thread, &thAttr, dnodeProcessMnodeWriteQueue, pWorker) != 0) { if (pthread_create(&pWorker->thread, &thAttr, dnodeProcessMWriteQueue, pWorker) != 0) {
dError("failed to create thread to process mwrite queue, reason:%s", strerror(errno)); dError("failed to create thread to process mwrite queue, reason:%s", strerror(errno));
} }
pthread_attr_destroy(&thAttr); pthread_attr_destroy(&thAttr);
tsMWritePool.curNum = i + 1; tsMWriteWP.curNum = i + 1;
dDebug("dnode mwrite worker:%d is launched, total:%d", pWorker->workerId, tsMWritePool.maxNum); dDebug("dnode mwrite worker:%d is launched, total:%d", pWorker->workerId, tsMWriteWP.maxNum);
} }
dDebug("dnode mwrite queue:%p is allocated", tsMWriteQueue); dDebug("dnode mwrite queue:%p is allocated", tsMWriteQueue);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
void dnodeFreeMnodeWqueue() { void dnodeFreeMWritequeue() {
dDebug("dnode mwrite queue:%p is freed", tsMWriteQueue); dDebug("dnode mwrite queue:%p is freed", tsMWriteQueue);
taosCloseQueue(tsMWriteQueue); taosCloseQueue(tsMWriteQueue);
tsMWriteQueue = NULL; tsMWriteQueue = NULL;
} }
void dnodeDispatchToMnodeWriteQueue(SRpcMsg *pMsg) { void dnodeDispatchToMWriteQueue(SRpcMsg *pMsg) {
if (!mnodeIsRunning() || tsMWriteQueue == NULL) { if (!mnodeIsRunning() || tsMWriteQueue == NULL) {
dnodeSendRedirectMsg(pMsg, true); dnodeSendRedirectMsg(pMsg, true);
rpcFreeCont(pMsg->pCont); rpcFreeCont(pMsg->pCont);
...@@ -137,7 +137,7 @@ void dnodeDispatchToMnodeWriteQueue(SRpcMsg *pMsg) { ...@@ -137,7 +137,7 @@ void dnodeDispatchToMnodeWriteQueue(SRpcMsg *pMsg) {
taosWriteQitem(tsMWriteQueue, TAOS_QTYPE_RPC, pWrite); taosWriteQitem(tsMWriteQueue, TAOS_QTYPE_RPC, pWrite);
} }
static void dnodeFreeMnodeWriteMsg(SMnodeMsg *pWrite) { static void dnodeFreeMWriteMsg(SMnodeMsg *pWrite) {
dDebug("app:%p:%p, msg:%s is freed from mwrite queue:%p", pWrite->rpcMsg.ahandle, pWrite, dDebug("app:%p:%p, msg:%s is freed from mwrite queue:%p", pWrite->rpcMsg.ahandle, pWrite,
taosMsg[pWrite->rpcMsg.msgType], tsMWriteQueue); taosMsg[pWrite->rpcMsg.msgType], tsMWriteQueue);
...@@ -145,12 +145,12 @@ static void dnodeFreeMnodeWriteMsg(SMnodeMsg *pWrite) { ...@@ -145,12 +145,12 @@ static void dnodeFreeMnodeWriteMsg(SMnodeMsg *pWrite) {
taosFreeQitem(pWrite); taosFreeQitem(pWrite);
} }
void dnodeSendRpcMnodeWriteRsp(void *pMsg, int32_t code) { void dnodeSendRpcMWriteRsp(void *pMsg, int32_t code) {
SMnodeMsg *pWrite = pMsg; SMnodeMsg *pWrite = pMsg;
if (pWrite == NULL) return; if (pWrite == NULL) return;
if (code == TSDB_CODE_MND_ACTION_IN_PROGRESS) return; if (code == TSDB_CODE_MND_ACTION_IN_PROGRESS) return;
if (code == TSDB_CODE_MND_ACTION_NEED_REPROCESSED) { if (code == TSDB_CODE_MND_ACTION_NEED_REPROCESSED) {
dnodeReprocessMnodeWriteMsg(pWrite); dnodeReprocessMWriteMsg(pWrite);
return; return;
} }
...@@ -162,10 +162,10 @@ void dnodeSendRpcMnodeWriteRsp(void *pMsg, int32_t code) { ...@@ -162,10 +162,10 @@ void dnodeSendRpcMnodeWriteRsp(void *pMsg, int32_t code) {
}; };
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
dnodeFreeMnodeWriteMsg(pWrite); dnodeFreeMWriteMsg(pWrite);
} }
static void *dnodeProcessMnodeWriteQueue(void *param) { static void *dnodeProcessMWriteQueue(void *param) {
SMnodeMsg *pWrite; SMnodeMsg *pWrite;
int32_t type; int32_t type;
void * unUsed; void * unUsed;
...@@ -180,13 +180,13 @@ static void *dnodeProcessMnodeWriteQueue(void *param) { ...@@ -180,13 +180,13 @@ static void *dnodeProcessMnodeWriteQueue(void *param) {
taosMsg[pWrite->rpcMsg.msgType]); taosMsg[pWrite->rpcMsg.msgType]);
int32_t code = mnodeProcessWrite(pWrite); int32_t code = mnodeProcessWrite(pWrite);
dnodeSendRpcMnodeWriteRsp(pWrite, code); dnodeSendRpcMWriteRsp(pWrite, code);
} }
return NULL; return NULL;
} }
void dnodeReprocessMnodeWriteMsg(void *pMsg) { void dnodeReprocessMWriteMsg(void *pMsg) {
SMnodeMsg *pWrite = pMsg; SMnodeMsg *pWrite = pMsg;
if (!mnodeIsRunning() || tsMWriteQueue == NULL) { if (!mnodeIsRunning() || tsMWriteQueue == NULL) {
...@@ -194,7 +194,7 @@ void dnodeReprocessMnodeWriteMsg(void *pMsg) { ...@@ -194,7 +194,7 @@ void dnodeReprocessMnodeWriteMsg(void *pMsg) {
taosMsg[pWrite->rpcMsg.msgType], pWrite->retry); taosMsg[pWrite->rpcMsg.msgType], pWrite->retry);
dnodeSendRedirectMsg(pMsg, true); dnodeSendRedirectMsg(pMsg, true);
dnodeFreeMnodeWriteMsg(pWrite); dnodeFreeMWriteMsg(pWrite);
} else { } else {
dDebug("app:%p:%p, msg:%s is reput into mwrite queue:%p, retry times:%d", pWrite->rpcMsg.ahandle, pWrite, dDebug("app:%p:%p, msg:%s is reput into mwrite queue:%p, retry times:%d", pWrite->rpcMsg.ahandle, pWrite,
taosMsg[pWrite->rpcMsg.msgType], tsMWriteQueue, pWrite->retry); taosMsg[pWrite->rpcMsg.msgType], tsMWriteQueue, pWrite->retry);
...@@ -203,12 +203,12 @@ void dnodeReprocessMnodeWriteMsg(void *pMsg) { ...@@ -203,12 +203,12 @@ void dnodeReprocessMnodeWriteMsg(void *pMsg) {
} }
} }
static void dnodeDoDelayReprocessMnodeWriteMsg(void *param, void *tmrId) { static void dnodeDoDelayReprocessMWriteMsg(void *param, void *tmrId) {
dnodeReprocessMnodeWriteMsg(param); dnodeReprocessMWriteMsg(param);
} }
void dnodeDelayReprocessMnodeWriteMsg(void *pMsg) { void dnodeDelayReprocessMWriteMsg(void *pMsg) {
SMnodeMsg *mnodeMsg = pMsg; SMnodeMsg *mnodeMsg = pMsg;
void *unUsed = NULL; void *unUsed = NULL;
taosTmrReset(dnodeDoDelayReprocessMnodeWriteMsg, 300, mnodeMsg, tsDnodeTmr, &unUsed); taosTmrReset(dnodeDoDelayReprocessMWriteMsg, 300, mnodeMsg, tsDnodeTmr, &unUsed);
} }
src/dnode/src/dnodeMain.c
浏览文件 @ 4bc9284a
...@@ -37,11 +37,11 @@ ...@@ -37,11 +37,11 @@
#include "dnodeShell.h" #include "dnodeShell.h"
#include "dnodeTelemetry.h" #include "dnodeTelemetry.h"
static SDnodeRunStatus tsDnodeRunStatus = TSDB_DNODE_RUN_STATUS_STOPPED; static SRunStatus tsRunStatus = TSDB_RUN_STATUS_STOPPED;
static int32_t dnodeInitStorage(); static int32_t dnodeInitStorage();
static void dnodeCleanupStorage(); static void dnodeCleanupStorage();
static void dnodeSetRunStatus(SDnodeRunStatus status); static void dnodeSetRunStatus(SRunStatus status);
static void dnodeCheckDataDirOpenned(char *dir); static void dnodeCheckDataDirOpenned(char *dir);
static int32_t dnodeInitComponents(); static int32_t dnodeInitComponents();
static void dnodeCleanupComponents(int32_t stepId); static void dnodeCleanupComponents(int32_t stepId);
...@@ -63,9 +63,9 @@ static const SDnodeComponent tsDnodeComponents[] = { ...@@ -63,9 +63,9 @@ static const SDnodeComponent tsDnodeComponents[] = {
{"check", dnodeInitCheck, dnodeCleanupCheck}, // NOTES: dnodeInitCheck must be behind the dnodeinitStorage component !!! {"check", dnodeInitCheck, dnodeCleanupCheck}, // NOTES: dnodeInitCheck must be behind the dnodeinitStorage component !!!
{"vread", dnodeInitVRead, dnodeCleanupVRead}, {"vread", dnodeInitVRead, dnodeCleanupVRead},
{"vwrite", dnodeInitVWrite, dnodeCleanupVWrite}, {"vwrite", dnodeInitVWrite, dnodeCleanupVWrite},
{"mread", dnodeInitMnodeRead, dnodeCleanupMnodeRead}, {"mread", dnodeInitMRead, dnodeCleanupMRead},
{"mwrite", dnodeInitMnodeWrite, dnodeCleanupMnodeWrite}, {"mwrite", dnodeInitMWrite, dnodeCleanupMWrite},
{"mpeer", dnodeInitMnodePeer, dnodeCleanupMnodePeer}, {"mpeer", dnodeInitMPeer, dnodeCleanupMPeer},
{"client", dnodeInitClient, dnodeCleanupClient}, {"client", dnodeInitClient, dnodeCleanupClient},
{"server", dnodeInitServer, dnodeCleanupServer}, {"server", dnodeInitServer, dnodeCleanupServer},
{"mgmt", dnodeInitMgmt, dnodeCleanupMgmt}, {"mgmt", dnodeInitMgmt, dnodeCleanupMgmt},
...@@ -104,7 +104,7 @@ static int32_t dnodeInitComponents() { ...@@ -104,7 +104,7 @@ static int32_t dnodeInitComponents() {
} }
int32_t dnodeInitSystem() { int32_t dnodeInitSystem() {
dnodeSetRunStatus(TSDB_DNODE_RUN_STATUS_INITIALIZE); dnodeSetRunStatus(TSDB_RUN_STATUS_INITIALIZE);
tscEmbedded = 1; tscEmbedded = 1;
taosBlockSIGPIPE(); taosBlockSIGPIPE();
taosResolveCRC(); taosResolveCRC();
...@@ -137,7 +137,7 @@ int32_t dnodeInitSystem() { ...@@ -137,7 +137,7 @@ int32_t dnodeInitSystem() {
} }
dnodeStartModules(); dnodeStartModules();
dnodeSetRunStatus(TSDB_DNODE_RUN_STATUS_RUNING); dnodeSetRunStatus(TSDB_RUN_STATUS_RUNING);
dInfo("TDengine is initialized successfully"); dInfo("TDengine is initialized successfully");
...@@ -145,20 +145,20 @@ int32_t dnodeInitSystem() { ...@@ -145,20 +145,20 @@ int32_t dnodeInitSystem() {
} }
void dnodeCleanUpSystem() { void dnodeCleanUpSystem() {
if (dnodeGetRunStatus() != TSDB_DNODE_RUN_STATUS_STOPPED) { if (dnodeGetRunStatus() != TSDB_RUN_STATUS_STOPPED) {
dnodeSetRunStatus(TSDB_DNODE_RUN_STATUS_STOPPED); dnodeSetRunStatus(TSDB_RUN_STATUS_STOPPED);
dnodeCleanupComponents(sizeof(tsDnodeComponents) / sizeof(tsDnodeComponents[0]) - 1); dnodeCleanupComponents(sizeof(tsDnodeComponents) / sizeof(tsDnodeComponents[0]) - 1);
taos_cleanup(); taos_cleanup();
taosCloseLog(); taosCloseLog();
} }
} }
SDnodeRunStatus dnodeGetRunStatus() { SRunStatus dnodeGetRunStatus() {
return tsDnodeRunStatus; return tsRunStatus;
} }
static void dnodeSetRunStatus(SDnodeRunStatus status) { static void dnodeSetRunStatus(SRunStatus status) {
tsDnodeRunStatus = status; tsRunStatus = status;
} }
static void dnodeCheckDataDirOpenned(char *dir) { static void dnodeCheckDataDirOpenned(char *dir) {
......
src/dnode/src/dnodePeer.c
浏览文件 @ 4bc9284a
...@@ -34,8 +34,8 @@ static void (*dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *); ...@@ -34,8 +34,8 @@ static void (*dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *);
static void dnodeProcessReqMsgFromDnode(SRpcMsg *pMsg, SRpcEpSet *); static void dnodeProcessReqMsgFromDnode(SRpcMsg *pMsg, SRpcEpSet *);
static void (*dnodeProcessRspMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *rpcMsg); static void (*dnodeProcessRspMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *rpcMsg);
static void dnodeProcessRspFromDnode(SRpcMsg *pMsg, SRpcEpSet *pEpSet); static void dnodeProcessRspFromDnode(SRpcMsg *pMsg, SRpcEpSet *pEpSet);
static void *tsDnodeServerRpc = NULL; static void *tsServerRpc = NULL;
static void *tsDnodeClientRpc = NULL; static void *tsClientRpc = NULL;
int32_t dnodeInitServer() { int32_t dnodeInitServer() {
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CREATE_TABLE] = dnodeDispatchToVWriteQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CREATE_TABLE] = dnodeDispatchToVWriteQueue;
...@@ -50,11 +50,11 @@ int32_t dnodeInitServer() { ...@@ -50,11 +50,11 @@ int32_t dnodeInitServer() {
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CONFIG_DNODE] = dnodeDispatchToMgmtQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CONFIG_DNODE] = dnodeDispatchToMgmtQueue;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CREATE_MNODE] = dnodeDispatchToMgmtQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CREATE_MNODE] = dnodeDispatchToMgmtQueue;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_CONFIG_TABLE] = dnodeDispatchToMnodePeerQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_CONFIG_TABLE] = dnodeDispatchToMPeerQueue;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_CONFIG_VNODE] = dnodeDispatchToMnodePeerQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_CONFIG_VNODE] = dnodeDispatchToMPeerQueue;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_AUTH] = dnodeDispatchToMnodePeerQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_AUTH] = dnodeDispatchToMPeerQueue;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_GRANT] = dnodeDispatchToMnodePeerQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_GRANT] = dnodeDispatchToMPeerQueue;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_STATUS] = dnodeDispatchToMnodePeerQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_STATUS] = dnodeDispatchToMPeerQueue;
SRpcInit rpcInit; SRpcInit rpcInit;
memset(&rpcInit, 0, sizeof(rpcInit)); memset(&rpcInit, 0, sizeof(rpcInit));
...@@ -66,8 +66,8 @@ int32_t dnodeInitServer() { ...@@ -66,8 +66,8 @@ int32_t dnodeInitServer() {
rpcInit.connType = TAOS_CONN_SERVER; rpcInit.connType = TAOS_CONN_SERVER;
rpcInit.idleTime = tsShellActivityTimer * 1000; rpcInit.idleTime = tsShellActivityTimer * 1000;
tsDnodeServerRpc = rpcOpen(&rpcInit); tsServerRpc = rpcOpen(&rpcInit);
if (tsDnodeServerRpc == NULL) { if (tsServerRpc == NULL) {
dError("failed to init inter-dnodes RPC server"); dError("failed to init inter-dnodes RPC server");
return -1; return -1;
} }
...@@ -77,9 +77,9 @@ int32_t dnodeInitServer() { ...@@ -77,9 +77,9 @@ int32_t dnodeInitServer() {
} }
void dnodeCleanupServer() { void dnodeCleanupServer() {
if (tsDnodeServerRpc) { if (tsServerRpc) {
rpcClose(tsDnodeServerRpc); rpcClose(tsServerRpc);
tsDnodeServerRpc = NULL; tsServerRpc = NULL;
dInfo("inter-dnodes RPC server is closed"); dInfo("inter-dnodes RPC server is closed");
} }
} }
...@@ -93,7 +93,7 @@ static void dnodeProcessReqMsgFromDnode(SRpcMsg *pMsg, SRpcEpSet *pEpSet) { ...@@ -93,7 +93,7 @@ static void dnodeProcessReqMsgFromDnode(SRpcMsg *pMsg, SRpcEpSet *pEpSet) {
if (pMsg->pCont == NULL) return; if (pMsg->pCont == NULL) return;
if (dnodeGetRunStatus() != TSDB_DNODE_RUN_STATUS_RUNING) { if (dnodeGetRunStatus() != TSDB_RUN_STATUS_RUNING) {
rspMsg.code = TSDB_CODE_APP_NOT_READY; rspMsg.code = TSDB_CODE_APP_NOT_READY;
rpcSendResponse(&rspMsg); rpcSendResponse(&rspMsg);
rpcFreeCont(pMsg->pCont); rpcFreeCont(pMsg->pCont);
...@@ -131,8 +131,8 @@ int32_t dnodeInitClient() { ...@@ -131,8 +131,8 @@ int32_t dnodeInitClient() {
rpcInit.ckey = "key"; rpcInit.ckey = "key";
rpcInit.secret = secret; rpcInit.secret = secret;
tsDnodeClientRpc = rpcOpen(&rpcInit); tsClientRpc = rpcOpen(&rpcInit);
if (tsDnodeClientRpc == NULL) { if (tsClientRpc == NULL) {
dError("failed to init mnode rpc client"); dError("failed to init mnode rpc client");
return -1; return -1;
} }
...@@ -142,9 +142,9 @@ int32_t dnodeInitClient() { ...@@ -142,9 +142,9 @@ int32_t dnodeInitClient() {
} }
void dnodeCleanupClient() { void dnodeCleanupClient() {
if (tsDnodeClientRpc) { if (tsClientRpc) {
rpcClose(tsDnodeClientRpc); rpcClose(tsClientRpc);
tsDnodeClientRpc = NULL; tsClientRpc = NULL;
dInfo("dnode inter-dnodes rpc client is closed"); dInfo("dnode inter-dnodes rpc client is closed");
} }
} }
...@@ -168,15 +168,15 @@ void dnodeAddClientRspHandle(uint8_t msgType, void (*fp)(SRpcMsg *rpcMsg)) { ...@@ -168,15 +168,15 @@ void dnodeAddClientRspHandle(uint8_t msgType, void (*fp)(SRpcMsg *rpcMsg)) {
} }
void dnodeSendMsgToDnode(SRpcEpSet *epSet, SRpcMsg *rpcMsg) { void dnodeSendMsgToDnode(SRpcEpSet *epSet, SRpcMsg *rpcMsg) {
rpcSendRequest(tsDnodeClientRpc, epSet, rpcMsg); rpcSendRequest(tsClientRpc, epSet, rpcMsg);
} }
void dnodeSendMsgToMnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp) { void dnodeSendMsgToMnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp) {
SRpcEpSet epSet = {0}; SRpcEpSet epSet = {0};
dnodeGetEpSetForPeer(&epSet); dnodeGetEpSetForPeer(&epSet);
rpcSendRecv(tsDnodeClientRpc, &epSet, rpcMsg, rpcRsp); rpcSendRecv(tsClientRpc, &epSet, rpcMsg, rpcRsp);
} }
void dnodeSendMsgToDnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp, SRpcEpSet *epSet) { void dnodeSendMsgToDnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp, SRpcEpSet *epSet) {
rpcSendRecv(tsDnodeClientRpc, epSet, rpcMsg, rpcRsp); rpcSendRecv(tsClientRpc, epSet, rpcMsg, rpcRsp);
} }
\ No newline at end of file
src/dnode/src/dnodeShell.c
浏览文件 @ 4bc9284a
...@@ -33,9 +33,9 @@ ...@@ -33,9 +33,9 @@
static void (*dnodeProcessShellMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *); static void (*dnodeProcessShellMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *);
static void dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcEpSet *); static void dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcEpSet *);
static int dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char *secret, char *ckey); static int dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char *secret, char *ckey);
static void * tsDnodeShellRpc = NULL; static void * tsShellRpc = NULL;
static int32_t tsDnodeQueryReqNum = 0; static int32_t tsQueryReqNum = 0;
static int32_t tsDnodeSubmitReqNum = 0; static int32_t tsSubmitReqNum = 0;
int32_t dnodeInitShell() { int32_t dnodeInitShell() {
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_SUBMIT] = dnodeDispatchToVWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_SUBMIT] = dnodeDispatchToVWriteQueue;
...@@ -44,35 +44,35 @@ int32_t dnodeInitShell() { ...@@ -44,35 +44,35 @@ int32_t dnodeInitShell() {
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_UPDATE_TAG_VAL] = dnodeDispatchToVWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_UPDATE_TAG_VAL] = dnodeDispatchToVWriteQueue;
// the following message shall be treated as mnode write // the following message shall be treated as mnode write
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_ACCT] = dnodeDispatchToMnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_ACCT] = dnodeDispatchToMWriteQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_ACCT] = dnodeDispatchToMnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_ACCT] = dnodeDispatchToMWriteQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_ACCT] = dnodeDispatchToMnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_ACCT] = dnodeDispatchToMWriteQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_USER] = dnodeDispatchToMnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_USER] = dnodeDispatchToMWriteQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_USER] = dnodeDispatchToMnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_USER] = dnodeDispatchToMWriteQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_USER] = dnodeDispatchToMnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_USER] = dnodeDispatchToMWriteQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_DNODE]= dnodeDispatchToMnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_DNODE]= dnodeDispatchToMWriteQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_DNODE] = dnodeDispatchToMnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_DNODE] = dnodeDispatchToMWriteQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_DB] = dnodeDispatchToMnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_DB] = dnodeDispatchToMWriteQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_DB] = dnodeDispatchToMnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_DB] = dnodeDispatchToMWriteQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_DB] = dnodeDispatchToMnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_DB] = dnodeDispatchToMWriteQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_TABLE]= dnodeDispatchToMnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_TABLE]= dnodeDispatchToMWriteQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_TABLE] = dnodeDispatchToMnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_TABLE] = dnodeDispatchToMWriteQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_TABLE] = dnodeDispatchToMnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_TABLE] = dnodeDispatchToMWriteQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_STREAM]= dnodeDispatchToMnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_STREAM]= dnodeDispatchToMWriteQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_KILL_QUERY] = dnodeDispatchToMnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_KILL_QUERY] = dnodeDispatchToMWriteQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_KILL_STREAM] = dnodeDispatchToMnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_KILL_STREAM] = dnodeDispatchToMWriteQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_KILL_CONN] = dnodeDispatchToMnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_KILL_CONN] = dnodeDispatchToMWriteQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CONFIG_DNODE]= dnodeDispatchToMnodeWriteQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CONFIG_DNODE]= dnodeDispatchToMWriteQueue;
// the following message shall be treated as mnode query // the following message shall be treated as mnode query
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_HEARTBEAT] = dnodeDispatchToMnodeReadQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_HEARTBEAT] = dnodeDispatchToMReadQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CONNECT] = dnodeDispatchToMnodeReadQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CONNECT] = dnodeDispatchToMReadQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_USE_DB] = dnodeDispatchToMnodeReadQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_USE_DB] = dnodeDispatchToMReadQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_TABLE_META] = dnodeDispatchToMnodeReadQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_TABLE_META] = dnodeDispatchToMReadQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_STABLE_VGROUP]= dnodeDispatchToMnodeReadQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_STABLE_VGROUP]= dnodeDispatchToMReadQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_TABLES_META] = dnodeDispatchToMnodeReadQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_TABLES_META] = dnodeDispatchToMReadQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_SHOW] = dnodeDispatchToMnodeReadQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_SHOW] = dnodeDispatchToMReadQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_RETRIEVE] = dnodeDispatchToMnodeReadQueue; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_RETRIEVE] = dnodeDispatchToMReadQueue;
int32_t numOfThreads = tsNumOfCores * tsNumOfThreadsPerCore; int32_t numOfThreads = tsNumOfCores * tsNumOfThreadsPerCore;
numOfThreads = (int32_t) ((1.0 - tsRatioOfQueryThreads) * numOfThreads / 2.0); numOfThreads = (int32_t) ((1.0 - tsRatioOfQueryThreads) * numOfThreads / 2.0);
...@@ -91,8 +91,8 @@ int32_t dnodeInitShell() { ...@@ -91,8 +91,8 @@ int32_t dnodeInitShell() {
rpcInit.idleTime = tsShellActivityTimer * 1000; rpcInit.idleTime = tsShellActivityTimer * 1000;
rpcInit.afp = dnodeRetrieveUserAuthInfo; rpcInit.afp = dnodeRetrieveUserAuthInfo;
tsDnodeShellRpc = rpcOpen(&rpcInit); tsShellRpc = rpcOpen(&rpcInit);
if (tsDnodeShellRpc == NULL) { if (tsShellRpc == NULL) {
dError("failed to init shell rpc server"); dError("failed to init shell rpc server");
return -1; return -1;
} }
...@@ -102,13 +102,13 @@ int32_t dnodeInitShell() { ...@@ -102,13 +102,13 @@ int32_t dnodeInitShell() {
} }
void dnodeCleanupShell() { void dnodeCleanupShell() {
if (tsDnodeShellRpc) { if (tsShellRpc) {
rpcClose(tsDnodeShellRpc); rpcClose(tsShellRpc);
tsDnodeShellRpc = NULL; tsShellRpc = NULL;
} }
} }
void dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcEpSet *pEpSet) { static void dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcEpSet *pEpSet) {
SRpcMsg rpcMsg = { SRpcMsg rpcMsg = {
.handle = pMsg->handle, .handle = pMsg->handle,
.pCont = NULL, .pCont = NULL,
...@@ -117,7 +117,7 @@ void dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcEpSet *pEpSet) { ...@@ -117,7 +117,7 @@ void dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcEpSet *pEpSet) {
if (pMsg->pCont == NULL) return; if (pMsg->pCont == NULL) return;
if (dnodeGetRunStatus() != TSDB_DNODE_RUN_STATUS_RUNING) { if (dnodeGetRunStatus() != TSDB_RUN_STATUS_RUNING) {
dError("RPC %p, shell msg:%s is ignored since dnode not running", pMsg->handle, taosMsg[pMsg->msgType]); dError("RPC %p, shell msg:%s is ignored since dnode not running", pMsg->handle, taosMsg[pMsg->msgType]);
rpcMsg.code = TSDB_CODE_APP_NOT_READY; rpcMsg.code = TSDB_CODE_APP_NOT_READY;
rpcSendResponse(&rpcMsg); rpcSendResponse(&rpcMsg);
...@@ -126,9 +126,9 @@ void dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcEpSet *pEpSet) { ...@@ -126,9 +126,9 @@ void dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcEpSet *pEpSet) {
} }
if (pMsg->msgType == TSDB_MSG_TYPE_QUERY) { if (pMsg->msgType == TSDB_MSG_TYPE_QUERY) {
atomic_fetch_add_32(&tsDnodeQueryReqNum, 1); atomic_fetch_add_32(&tsQueryReqNum, 1);
} else if (pMsg->msgType == TSDB_MSG_TYPE_SUBMIT) { } else if (pMsg->msgType == TSDB_MSG_TYPE_SUBMIT) {
atomic_fetch_add_32(&tsDnodeSubmitReqNum, 1); atomic_fetch_add_32(&tsSubmitReqNum, 1);
} else {} } else {}
if ( dnodeProcessShellMsgFp[pMsg->msgType] ) { if ( dnodeProcessShellMsgFp[pMsg->msgType] ) {
...@@ -211,12 +211,12 @@ void *dnodeSendCfgTableToRecv(int32_t vgId, int32_t tid) { ...@@ -211,12 +211,12 @@ void *dnodeSendCfgTableToRecv(int32_t vgId, int32_t tid) {
} }
} }
SDnodeStatisInfo dnodeGetStatisInfo() { SStatisInfo dnodeGetStatisInfo() {
SDnodeStatisInfo info = {0}; SStatisInfo info = {0};
if (dnodeGetRunStatus() == TSDB_DNODE_RUN_STATUS_RUNING) { if (dnodeGetRunStatus() == TSDB_RUN_STATUS_RUNING) {
info.httpReqNum = httpGetReqCount(); info.httpReqNum = httpGetReqCount();
info.queryReqNum = atomic_exchange_32(&tsDnodeQueryReqNum, 0); info.queryReqNum = atomic_exchange_32(&tsQueryReqNum, 0);
info.submitReqNum = atomic_exchange_32(&tsDnodeSubmitReqNum, 0); info.submitReqNum = atomic_exchange_32(&tsSubmitReqNum, 0);
} }
return info; return info;
......
src/dnode/src/dnodeVWrite.c
浏览文件 @ 4bc9284a
...@@ -113,7 +113,7 @@ void dnodeDispatchToVWriteQueue(SRpcMsg *pRpcMsg) { ...@@ -113,7 +113,7 @@ void dnodeDispatchToVWriteQueue(SRpcMsg *pRpcMsg) {
void *dnodeAllocVWriteQueue(void *pVnode) { void *dnodeAllocVWriteQueue(void *pVnode) {
pthread_mutex_lock(&tsVWriteWP.mutex); pthread_mutex_lock(&tsVWriteWP.mutex);
SVWriteWorker *pWorker = tsVWriteWP.worker + tsVWriteWP.nextId; SVWriteWorker *pWorker = tsVWriteWP.worker + tsVWriteWP.nextId;
void *queue = taosOpenQueue(); taos_queue *queue = taosOpenQueue();
if (queue == NULL) { if (queue == NULL) {
pthread_mutex_unlock(&tsVWriteWP.mutex); pthread_mutex_unlock(&tsVWriteWP.mutex);
return NULL; return NULL;
......
src/inc/dnode.h
浏览文件 @ 4bc9284a
...@@ -27,16 +27,16 @@ typedef struct { ...@@ -27,16 +27,16 @@ typedef struct {
int32_t queryReqNum; int32_t queryReqNum;
int32_t submitReqNum; int32_t submitReqNum;
int32_t httpReqNum; int32_t httpReqNum;
} SDnodeStatisInfo; } SStatisInfo;
typedef enum { typedef enum {
TSDB_DNODE_RUN_STATUS_INITIALIZE, TSDB_RUN_STATUS_INITIALIZE,
TSDB_DNODE_RUN_STATUS_RUNING, TSDB_RUN_STATUS_RUNING,
TSDB_DNODE_RUN_STATUS_STOPPED TSDB_RUN_STATUS_STOPPED
} SDnodeRunStatus; } SRunStatus;
SDnodeRunStatus dnodeGetRunStatus(); SRunStatus dnodeGetRunStatus();
SDnodeStatisInfo dnodeGetStatisInfo(); SStatisInfo dnodeGetStatisInfo();
bool dnodeIsFirstDeploy(); bool dnodeIsFirstDeploy();
bool dnodeIsMasterEp(char *ep); bool dnodeIsMasterEp(char *ep);
...@@ -59,15 +59,15 @@ void dnodeSendRpcVWriteRsp(void *pVnode, void *param, int32_t code); ...@@ -59,15 +59,15 @@ void dnodeSendRpcVWriteRsp(void *pVnode, void *param, int32_t code);
void *dnodeAllocVReadQueue(void *pVnode); void *dnodeAllocVReadQueue(void *pVnode);
void dnodeFreeVReadQueue(void *rqueue); void dnodeFreeVReadQueue(void *rqueue);
int32_t dnodeAllocateMnodePqueue(); int32_t dnodeAllocateMPeerQueue();
void dnodeFreeMnodePqueue(); void dnodeFreeMPeerQueue();
int32_t dnodeAllocateMnodeRqueue(); int32_t dnodeAllocMReadQueue();
void dnodeFreeMnodeRqueue(); void dnodeFreeMReadQueue();
int32_t dnodeAllocateMnodeWqueue(); int32_t dnodeAllocMWritequeue();
void dnodeFreeMnodeWqueue(); void dnodeFreeMWritequeue();
void dnodeSendRpcMnodeWriteRsp(void *pMsg, int32_t code); void dnodeSendRpcMWriteRsp(void *pMsg, int32_t code);
void dnodeReprocessMnodeWriteMsg(void *pMsg); void dnodeReprocessMWriteMsg(void *pMsg);
void dnodeDelayReprocessMnodeWriteMsg(void *pMsg); void dnodeDelayReprocessMWriteMsg(void *pMsg);
void dnodeSendStatusMsgToMnode(); void dnodeSendStatusMsgToMnode();
......
src/mnode/src/mnodeMain.c
浏览文件 @ 4bc9284a
...@@ -96,9 +96,9 @@ int32_t mnodeStartSystem() { ...@@ -96,9 +96,9 @@ int32_t mnodeStartSystem() {
return -1; return -1;
} }
dnodeAllocateMnodeWqueue(); dnodeAllocMWritequeue();
dnodeAllocateMnodeRqueue(); dnodeAllocMReadQueue();
dnodeAllocateMnodePqueue(); dnodeAllocateMPeerQueue();
if (mnodeInitComponents() != 0) { if (mnodeInitComponents() != 0) {
return -1; return -1;
...@@ -127,9 +127,9 @@ void mnodeCleanupSystem() { ...@@ -127,9 +127,9 @@ void mnodeCleanupSystem() {
mInfo("starting to clean up mnode"); mInfo("starting to clean up mnode");
tsMgmtIsRunning = false; tsMgmtIsRunning = false;
dnodeFreeMnodeWqueue(); dnodeFreeMWritequeue();
dnodeFreeMnodeRqueue(); dnodeFreeMReadQueue();
dnodeFreeMnodePqueue(); dnodeFreeMPeerQueue();
mnodeCleanupTimer(); mnodeCleanupTimer();
mnodeCleanupComponents(sizeof(tsMnodeComponents) / sizeof(tsMnodeComponents[0]) - 1); mnodeCleanupComponents(sizeof(tsMnodeComponents) / sizeof(tsMnodeComponents[0]) - 1);
......
src/mnode/src/mnodeSdb.c
浏览文件 @ 4bc9284a
...@@ -295,7 +295,7 @@ static void sdbConfirmForward(void *ahandle, void *param, int32_t code) { ...@@ -295,7 +295,7 @@ static void sdbConfirmForward(void *ahandle, void *param, int32_t code) {
if (pOper->writeCb != NULL) { if (pOper->writeCb != NULL) {
pOper->retCode = (*pOper->writeCb)(pMsg, pOper->retCode); pOper->retCode = (*pOper->writeCb)(pMsg, pOper->retCode);
} }
dnodeSendRpcMnodeWriteRsp(pMsg, pOper->retCode); dnodeSendRpcMWriteRsp(pMsg, pOper->retCode);
// if ahandle, means this func is called by sdb write // if ahandle, means this func is called by sdb write
if (ahandle == NULL) { if (ahandle == NULL) {
......
src/mnode/src/mnodeTable.c
浏览文件 @ 4bc9284a
...@@ -1679,12 +1679,12 @@ static int32_t mnodeDoCreateChildTableCb(SMnodeMsg *pMsg, int32_t code) { ...@@ -1679,12 +1679,12 @@ static int32_t mnodeDoCreateChildTableCb(SMnodeMsg *pMsg, int32_t code) {
pTable->info.tableId, pMsg->rpcMsg.handle); pTable->info.tableId, pMsg->rpcMsg.handle);
pMsg->retry = 0; pMsg->retry = 0;
dnodeReprocessMnodeWriteMsg(pMsg); dnodeReprocessMWriteMsg(pMsg);
} else { } else {
mDebug("app:%p:%p, table:%s, created in dnode, thandle:%p", pMsg->rpcMsg.ahandle, pMsg, pTable->info.tableId, mDebug("app:%p:%p, table:%s, created in dnode, thandle:%p", pMsg->rpcMsg.ahandle, pMsg, pTable->info.tableId,
pMsg->rpcMsg.handle); pMsg->rpcMsg.handle);
dnodeSendRpcMnodeWriteRsp(pMsg, TSDB_CODE_SUCCESS); dnodeSendRpcMWriteRsp(pMsg, TSDB_CODE_SUCCESS);
} }
return TSDB_CODE_MND_ACTION_IN_PROGRESS; return TSDB_CODE_MND_ACTION_IN_PROGRESS;
} else { } else {
...@@ -2351,14 +2351,14 @@ static void mnodeProcessDropChildTableRsp(SRpcMsg *rpcMsg) { ...@@ -2351,14 +2351,14 @@ static void mnodeProcessDropChildTableRsp(SRpcMsg *rpcMsg) {
mError("app:%p:%p, table:%s, failed to drop in dnode, vgId:%d sid:%d uid:%" PRIu64 ", reason:%s", mError("app:%p:%p, table:%s, failed to drop in dnode, vgId:%d sid:%d uid:%" PRIu64 ", reason:%s",
mnodeMsg->rpcMsg.ahandle, mnodeMsg, pTable->info.tableId, pTable->vgId, pTable->tid, pTable->uid, mnodeMsg->rpcMsg.ahandle, mnodeMsg, pTable->info.tableId, pTable->vgId, pTable->tid, pTable->uid,
tstrerror(rpcMsg->code)); tstrerror(rpcMsg->code));
dnodeSendRpcMnodeWriteRsp(mnodeMsg, rpcMsg->code); dnodeSendRpcMWriteRsp(mnodeMsg, rpcMsg->code);
return; return;
} }
if (mnodeMsg->pVgroup == NULL) mnodeMsg->pVgroup = mnodeGetVgroup(pTable->vgId); if (mnodeMsg->pVgroup == NULL) mnodeMsg->pVgroup = mnodeGetVgroup(pTable->vgId);
if (mnodeMsg->pVgroup == NULL) { if (mnodeMsg->pVgroup == NULL) {
mError("app:%p:%p, table:%s, failed to get vgroup", mnodeMsg->rpcMsg.ahandle, mnodeMsg, pTable->info.tableId); mError("app:%p:%p, table:%s, failed to get vgroup", mnodeMsg->rpcMsg.ahandle, mnodeMsg, pTable->info.tableId);
dnodeSendRpcMnodeWriteRsp(mnodeMsg, TSDB_CODE_MND_VGROUP_NOT_EXIST); dnodeSendRpcMWriteRsp(mnodeMsg, TSDB_CODE_MND_VGROUP_NOT_EXIST);
return; return;
} }
...@@ -2368,7 +2368,7 @@ static void mnodeProcessDropChildTableRsp(SRpcMsg *rpcMsg) { ...@@ -2368,7 +2368,7 @@ static void mnodeProcessDropChildTableRsp(SRpcMsg *rpcMsg) {
mnodeDropVgroup(mnodeMsg->pVgroup, NULL); mnodeDropVgroup(mnodeMsg->pVgroup, NULL);
} }
dnodeSendRpcMnodeWriteRsp(mnodeMsg, TSDB_CODE_SUCCESS); dnodeSendRpcMWriteRsp(mnodeMsg, TSDB_CODE_SUCCESS);
} }
/* /*
...@@ -2399,7 +2399,7 @@ static void mnodeProcessCreateChildTableRsp(SRpcMsg *rpcMsg) { ...@@ -2399,7 +2399,7 @@ static void mnodeProcessCreateChildTableRsp(SRpcMsg *rpcMsg) {
mnodeSendDropChildTableMsg(mnodeMsg, false); mnodeSendDropChildTableMsg(mnodeMsg, false);
rpcMsg->code = TSDB_CODE_SUCCESS; rpcMsg->code = TSDB_CODE_SUCCESS;
dnodeSendRpcMnodeWriteRsp(mnodeMsg, rpcMsg->code); dnodeSendRpcMWriteRsp(mnodeMsg, rpcMsg->code);
return; return;
} }
...@@ -2416,7 +2416,7 @@ static void mnodeProcessCreateChildTableRsp(SRpcMsg *rpcMsg) { ...@@ -2416,7 +2416,7 @@ static void mnodeProcessCreateChildTableRsp(SRpcMsg *rpcMsg) {
if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_MND_ACTION_IN_PROGRESS) { if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_MND_ACTION_IN_PROGRESS) {
mnodeMsg->pTable = NULL; mnodeMsg->pTable = NULL;
mnodeDestroyChildTable(pTable); mnodeDestroyChildTable(pTable);
dnodeSendRpcMnodeWriteRsp(mnodeMsg, code); dnodeSendRpcMWriteRsp(mnodeMsg, code);
} }
} else { } else {
if (mnodeMsg->retry++ < 10) { if (mnodeMsg->retry++ < 10) {
...@@ -2425,7 +2425,7 @@ static void mnodeProcessCreateChildTableRsp(SRpcMsg *rpcMsg) { ...@@ -2425,7 +2425,7 @@ static void mnodeProcessCreateChildTableRsp(SRpcMsg *rpcMsg) {
mnodeMsg->rpcMsg.ahandle, mnodeMsg, pTable->info.tableId, mnodeMsg->retry, pTable->vgId, pTable->tid, mnodeMsg->rpcMsg.ahandle, mnodeMsg, pTable->info.tableId, mnodeMsg->retry, pTable->vgId, pTable->tid,
pTable->uid, tstrerror(rpcMsg->code), mnodeMsg->rpcMsg.handle); pTable->uid, tstrerror(rpcMsg->code), mnodeMsg->rpcMsg.handle);
dnodeDelayReprocessMnodeWriteMsg(mnodeMsg); dnodeDelayReprocessMWriteMsg(mnodeMsg);
} else { } else {
mError("app:%p:%p, table:%s, failed to create in dnode, vgId:%d sid:%d uid:%" PRIu64 ", result:%s thandle:%p", mError("app:%p:%p, table:%s, failed to create in dnode, vgId:%d sid:%d uid:%" PRIu64 ", result:%s thandle:%p",
mnodeMsg->rpcMsg.ahandle, mnodeMsg, pTable->info.tableId, pTable->vgId, pTable->tid, pTable->uid, mnodeMsg->rpcMsg.ahandle, mnodeMsg, pTable->info.tableId, pTable->vgId, pTable->tid, pTable->uid,
...@@ -2434,7 +2434,7 @@ static void mnodeProcessCreateChildTableRsp(SRpcMsg *rpcMsg) { ...@@ -2434,7 +2434,7 @@ static void mnodeProcessCreateChildTableRsp(SRpcMsg *rpcMsg) {
SSdbOper oper = {.type = SDB_OPER_GLOBAL, .table = tsChildTableSdb, .pObj = pTable}; SSdbOper oper = {.type = SDB_OPER_GLOBAL, .table = tsChildTableSdb, .pObj = pTable};
sdbDeleteRow(&oper); sdbDeleteRow(&oper);
dnodeSendRpcMnodeWriteRsp(mnodeMsg, rpcMsg->code); dnodeSendRpcMWriteRsp(mnodeMsg, rpcMsg->code);
} }
} }
} }
...@@ -2452,18 +2452,18 @@ static void mnodeProcessAlterTableRsp(SRpcMsg *rpcMsg) { ...@@ -2452,18 +2452,18 @@ static void mnodeProcessAlterTableRsp(SRpcMsg *rpcMsg) {
mDebug("app:%p:%p, ctable:%s, altered in dnode, thandle:%p result:%s", mnodeMsg->rpcMsg.ahandle, mnodeMsg, mDebug("app:%p:%p, ctable:%s, altered in dnode, thandle:%p result:%s", mnodeMsg->rpcMsg.ahandle, mnodeMsg,
pTable->info.tableId, mnodeMsg->rpcMsg.handle, tstrerror(rpcMsg->code)); pTable->info.tableId, mnodeMsg->rpcMsg.handle, tstrerror(rpcMsg->code));
dnodeSendRpcMnodeWriteRsp(mnodeMsg, TSDB_CODE_SUCCESS); dnodeSendRpcMWriteRsp(mnodeMsg, TSDB_CODE_SUCCESS);
} else { } else {
if (mnodeMsg->retry++ < 3) { if (mnodeMsg->retry++ < 3) {
mDebug("app:%p:%p, table:%s, alter table rsp received, need retry, times:%d result:%s thandle:%p", mDebug("app:%p:%p, table:%s, alter table rsp received, need retry, times:%d result:%s thandle:%p",
mnodeMsg->rpcMsg.ahandle, mnodeMsg, pTable->info.tableId, mnodeMsg->retry, tstrerror(rpcMsg->code), mnodeMsg->rpcMsg.ahandle, mnodeMsg, pTable->info.tableId, mnodeMsg->retry, tstrerror(rpcMsg->code),
mnodeMsg->rpcMsg.handle); mnodeMsg->rpcMsg.handle);
dnodeDelayReprocessMnodeWriteMsg(mnodeMsg); dnodeDelayReprocessMWriteMsg(mnodeMsg);
} else { } else {
mError("app:%p:%p, table:%s, failed to alter in dnode, result:%s thandle:%p", mnodeMsg->rpcMsg.ahandle, mnodeMsg, mError("app:%p:%p, table:%s, failed to alter in dnode, result:%s thandle:%p", mnodeMsg->rpcMsg.ahandle, mnodeMsg,
pTable->info.tableId, tstrerror(rpcMsg->code), mnodeMsg->rpcMsg.handle); pTable->info.tableId, tstrerror(rpcMsg->code), mnodeMsg->rpcMsg.handle);
dnodeSendRpcMnodeWriteRsp(mnodeMsg, rpcMsg->code); dnodeSendRpcMWriteRsp(mnodeMsg, rpcMsg->code);
} }
} }
} }
......
src/mnode/src/mnodeVgroup.c
浏览文件 @ 4bc9284a
...@@ -529,7 +529,7 @@ static int32_t mnodeCreateVgroupCb(SMnodeMsg *pMsg, int32_t code) { ...@@ -529,7 +529,7 @@ static int32_t mnodeCreateVgroupCb(SMnodeMsg *pMsg, int32_t code) {
SSdbOper desc = {.type = SDB_OPER_GLOBAL, .pObj = pVgroup, .table = tsVgroupSdb}; SSdbOper desc = {.type = SDB_OPER_GLOBAL, .pObj = pVgroup, .table = tsVgroupSdb};
(void)sdbUpdateRow(&desc); (void)sdbUpdateRow(&desc);
dnodeReprocessMnodeWriteMsg(pMsg); dnodeReprocessMWriteMsg(pMsg);
return TSDB_CODE_MND_ACTION_IN_PROGRESS; return TSDB_CODE_MND_ACTION_IN_PROGRESS;
// if (pVgroup->status == TAOS_VG_STATUS_CREATING || pVgroup->status == TAOS_VG_STATUS_READY) { // if (pVgroup->status == TAOS_VG_STATUS_CREATING || pVgroup->status == TAOS_VG_STATUS_READY) {
// mInfo("app:%p:%p, vgId:%d, is created in sdb, db:%s replica:%d", pMsg->rpcMsg.ahandle, pMsg, pVgroup->vgId, // mInfo("app:%p:%p, vgId:%d, is created in sdb, db:%s replica:%d", pMsg->rpcMsg.ahandle, pMsg, pVgroup->vgId,
...@@ -537,7 +537,7 @@ static int32_t mnodeCreateVgroupCb(SMnodeMsg *pMsg, int32_t code) { ...@@ -537,7 +537,7 @@ static int32_t mnodeCreateVgroupCb(SMnodeMsg *pMsg, int32_t code) {
// pVgroup->status = TAOS_VG_STATUS_READY; // pVgroup->status = TAOS_VG_STATUS_READY;
// SSdbOper desc = {.type = SDB_OPER_GLOBAL, .pObj = pVgroup, .table = tsVgroupSdb}; // SSdbOper desc = {.type = SDB_OPER_GLOBAL, .pObj = pVgroup, .table = tsVgroupSdb};
// (void)sdbUpdateRow(&desc); // (void)sdbUpdateRow(&desc);
// dnodeReprocessMnodeWriteMsg(pMsg); // dnodeReprocessMWriteMsg(pMsg);
// return TSDB_CODE_MND_ACTION_IN_PROGRESS; // return TSDB_CODE_MND_ACTION_IN_PROGRESS;
// } else { // } else {
// mError("app:%p:%p, vgId:%d, is created in sdb, db:%s replica:%d, but vgroup is dropping", pMsg->rpcMsg.ahandle, // mError("app:%p:%p, vgId:%d, is created in sdb, db:%s replica:%d, but vgroup is dropping", pMsg->rpcMsg.ahandle,
...@@ -969,7 +969,7 @@ static void mnodeProcessCreateVnodeRsp(SRpcMsg *rpcMsg) { ...@@ -969,7 +969,7 @@ static void mnodeProcessCreateVnodeRsp(SRpcMsg *rpcMsg) {
if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_MND_ACTION_IN_PROGRESS) { if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_MND_ACTION_IN_PROGRESS) {
mnodeMsg->pVgroup = NULL; mnodeMsg->pVgroup = NULL;
mnodeDestroyVgroup(pVgroup); mnodeDestroyVgroup(pVgroup);
dnodeSendRpcMnodeWriteRsp(mnodeMsg, code); dnodeSendRpcMWriteRsp(mnodeMsg, code);
} }
} else { } else {
SSdbOper oper = { SSdbOper oper = {
...@@ -978,7 +978,7 @@ static void mnodeProcessCreateVnodeRsp(SRpcMsg *rpcMsg) { ...@@ -978,7 +978,7 @@ static void mnodeProcessCreateVnodeRsp(SRpcMsg *rpcMsg) {
.pObj = pVgroup .pObj = pVgroup
}; };
sdbDeleteRow(&oper); sdbDeleteRow(&oper);
dnodeSendRpcMnodeWriteRsp(mnodeMsg, mnodeMsg->code); dnodeSendRpcMWriteRsp(mnodeMsg, mnodeMsg->code);
} }
} }
...@@ -1040,7 +1040,7 @@ static void mnodeProcessDropVnodeRsp(SRpcMsg *rpcMsg) { ...@@ -1040,7 +1040,7 @@ static void mnodeProcessDropVnodeRsp(SRpcMsg *rpcMsg) {
code = TSDB_CODE_MND_SDB_ERROR; code = TSDB_CODE_MND_SDB_ERROR;
} }
dnodeReprocessMnodeWriteMsg(mnodeMsg); dnodeReprocessMWriteMsg(mnodeMsg);
} }
static int32_t mnodeProcessVnodeCfgMsg(SMnodeMsg *pMsg) { static int32_t mnodeProcessVnodeCfgMsg(SMnodeMsg *pMsg) {
......
src/plugins/monitor/src/monitorMain.c
浏览文件 @ 4bc9284a
...@@ -27,12 +27,12 @@ ...@@ -27,12 +27,12 @@
#include "monitor.h" #include "monitor.h"
#include "taoserror.h" #include "taoserror.h"
#define monitorFatal(...) { if (monitorDebugFlag & DEBUG_FATAL) { taosPrintLog("MON FATAL ", 255, __VA_ARGS__); }} #define mnFatal(...) { if (monitorDebugFlag & DEBUG_FATAL) { taosPrintLog("MON FATAL ", 255, __VA_ARGS__); }}
#define monitorError(...) { if (monitorDebugFlag & DEBUG_ERROR) { taosPrintLog("MON ERROR ", 255, __VA_ARGS__); }} #define mnError(...) { if (monitorDebugFlag & DEBUG_ERROR) { taosPrintLog("MON ERROR ", 255, __VA_ARGS__); }}
#define monitorWarn(...) { if (monitorDebugFlag & DEBUG_WARN) { taosPrintLog("MON WARN ", 255, __VA_ARGS__); }} #define mnWarn(...) { if (monitorDebugFlag & DEBUG_WARN) { taosPrintLog("MON WARN ", 255, __VA_ARGS__); }}
#define monitorInfo(...) { if (monitorDebugFlag & DEBUG_INFO) { taosPrintLog("MON ", 255, __VA_ARGS__); }} #define mnInfo(...) { if (monitorDebugFlag & DEBUG_INFO) { taosPrintLog("MON ", 255, __VA_ARGS__); }}
#define monitorDebug(...) { if (monitorDebugFlag & DEBUG_DEBUG) { taosPrintLog("MON ", monitorDebugFlag, __VA_ARGS__); }} #define mnDebug(...) { if (monitorDebugFlag & DEBUG_DEBUG) { taosPrintLog("MON ", monitorDebugFlag, __VA_ARGS__); }}
#define monitorTrace(...) { if (monitorDebugFlag & DEBUG_TRACE) { taosPrintLog("MON ", monitorDebugFlag, __VA_ARGS__); }} #define mnTrace(...) { if (monitorDebugFlag & DEBUG_TRACE) { taosPrintLog("MON ", monitorDebugFlag, __VA_ARGS__); }}
#define SQL_LENGTH 1030 #define SQL_LENGTH 1030
#define LOG_LEN_STR 100 #define LOG_LEN_STR 100
...@@ -91,12 +91,12 @@ int32_t monitorInitSystem() { ...@@ -91,12 +91,12 @@ int32_t monitorInitSystem() {
pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE); pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE);
if (pthread_create(&tsMonitor.thread, &thAttr, monitorThreadFunc, NULL)) { if (pthread_create(&tsMonitor.thread, &thAttr, monitorThreadFunc, NULL)) {
monitorError("failed to create thread to for monitor module, reason:%s", strerror(errno)); mnError("failed to create thread to for monitor module, reason:%s", strerror(errno));
return -1; return -1;
} }
pthread_attr_destroy(&thAttr); pthread_attr_destroy(&thAttr);
monitorDebug("monitor thread is launched"); mnDebug("monitor thread is launched");
monitorStartSystemFp = monitorStartSystem; monitorStartSystemFp = monitorStartSystem;
monitorStopSystemFp = monitorStopSystem; monitorStopSystemFp = monitorStopSystem;
...@@ -107,12 +107,12 @@ int32_t monitorStartSystem() { ...@@ -107,12 +107,12 @@ int32_t monitorStartSystem() {
taos_init(); taos_init();
tsMonitor.start = 1; tsMonitor.start = 1;
monitorExecuteSQLFp = monitorExecuteSQL; monitorExecuteSQLFp = monitorExecuteSQL;
monitorInfo("monitor module start"); mnInfo("monitor module start");
return 0; return 0;
} }
static void *monitorThreadFunc(void *param) { static void *monitorThreadFunc(void *param) {
monitorDebug("starting to initialize monitor module ..."); mnDebug("starting to initialize monitor module ...");
while (1) { while (1) {
static int32_t accessTimes = 0; static int32_t accessTimes = 0;
...@@ -121,7 +121,7 @@ static void *monitorThreadFunc(void *param) { ...@@ -121,7 +121,7 @@ static void *monitorThreadFunc(void *param) {
if (tsMonitor.quiting) { if (tsMonitor.quiting) {
tsMonitor.state = MON_STATE_NOT_INIT; tsMonitor.state = MON_STATE_NOT_INIT;
monitorInfo("monitor thread will quit, for taosd is quiting"); mnInfo("monitor thread will quit, for taosd is quiting");
break; break;
} else { } else {
taosGetDisk(); taosGetDisk();
...@@ -132,7 +132,7 @@ static void *monitorThreadFunc(void *param) { ...@@ -132,7 +132,7 @@ static void *monitorThreadFunc(void *param) {
} }
if (dnodeGetDnodeId() <= 0) { if (dnodeGetDnodeId() <= 0) {
monitorDebug("dnode not initialized, waiting for 3000 ms to start monitor module"); mnDebug("dnode not initialized, waiting for 3000 ms to start monitor module");
continue; continue;
} }
...@@ -140,10 +140,10 @@ static void *monitorThreadFunc(void *param) { ...@@ -140,10 +140,10 @@ static void *monitorThreadFunc(void *param) {
tsMonitor.state = MON_STATE_NOT_INIT; tsMonitor.state = MON_STATE_NOT_INIT;
tsMonitor.conn = taos_connect(NULL, "monitor", tsInternalPass, "", 0); tsMonitor.conn = taos_connect(NULL, "monitor", tsInternalPass, "", 0);
if (tsMonitor.conn == NULL) { if (tsMonitor.conn == NULL) {
monitorError("failed to connect to database, reason:%s", tstrerror(terrno)); mnError("failed to connect to database, reason:%s", tstrerror(terrno));
continue; continue;
} else { } else {
monitorDebug("connect to database success"); mnDebug("connect to database success");
} }
} }
...@@ -155,10 +155,10 @@ static void *monitorThreadFunc(void *param) { ...@@ -155,10 +155,10 @@ static void *monitorThreadFunc(void *param) {
taos_free_result(res); taos_free_result(res);
if (code != 0) { if (code != 0) {
monitorError("failed to exec sql:%s, reason:%s", tsMonitor.sql, tstrerror(code)); mnError("failed to exec sql:%s, reason:%s", tsMonitor.sql, tstrerror(code));
break; break;
} else { } else {
monitorDebug("successfully to exec sql:%s", tsMonitor.sql); mnDebug("successfully to exec sql:%s", tsMonitor.sql);
} }
} }
...@@ -174,7 +174,7 @@ static void *monitorThreadFunc(void *param) { ...@@ -174,7 +174,7 @@ static void *monitorThreadFunc(void *param) {
} }
} }
monitorInfo("monitor thread is stopped"); mnInfo("monitor thread is stopped");
return NULL; return NULL;
} }
...@@ -238,7 +238,7 @@ void monitorStopSystem() { ...@@ -238,7 +238,7 @@ void monitorStopSystem() {
tsMonitor.start = 0; tsMonitor.start = 0;
tsMonitor.state = MON_STATE_NOT_INIT; tsMonitor.state = MON_STATE_NOT_INIT;
monitorExecuteSQLFp = NULL; monitorExecuteSQLFp = NULL;
monitorInfo("monitor module stopped"); mnInfo("monitor module stopped");
} }
void monitorCleanUpSystem() { void monitorCleanUpSystem() {
...@@ -249,7 +249,7 @@ void monitorCleanUpSystem() { ...@@ -249,7 +249,7 @@ void monitorCleanUpSystem() {
taos_close(tsMonitor.conn); taos_close(tsMonitor.conn);
tsMonitor.conn = NULL; tsMonitor.conn = NULL;
} }
monitorInfo("monitor module is cleaned up"); mnInfo("monitor module is cleaned up");
} }
// unit is MB // unit is MB
...@@ -257,13 +257,13 @@ static int32_t monitorBuildMemorySql(char *sql) { ...@@ -257,13 +257,13 @@ static int32_t monitorBuildMemorySql(char *sql) {
float sysMemoryUsedMB = 0; float sysMemoryUsedMB = 0;
bool suc = taosGetSysMemory(&sysMemoryUsedMB); bool suc = taosGetSysMemory(&sysMemoryUsedMB);
if (!suc) { if (!suc) {
monitorDebug("failed to get sys memory info"); mnDebug("failed to get sys memory info");
} }
float procMemoryUsedMB = 0; float procMemoryUsedMB = 0;
suc = taosGetProcMemory(&procMemoryUsedMB); suc = taosGetProcMemory(&procMemoryUsedMB);
if (!suc) { if (!suc) {
monitorDebug("failed to get proc memory info"); mnDebug("failed to get proc memory info");
} }
return sprintf(sql, ", %f, %f, %d", procMemoryUsedMB, sysMemoryUsedMB, tsTotalMemoryMB); return sprintf(sql, ", %f, %f, %d", procMemoryUsedMB, sysMemoryUsedMB, tsTotalMemoryMB);
...@@ -274,7 +274,7 @@ static int32_t monitorBuildCpuSql(char *sql) { ...@@ -274,7 +274,7 @@ static int32_t monitorBuildCpuSql(char *sql) {
float sysCpuUsage = 0, procCpuUsage = 0; float sysCpuUsage = 0, procCpuUsage = 0;
bool suc = taosGetCpuUsage(&sysCpuUsage, &procCpuUsage); bool suc = taosGetCpuUsage(&sysCpuUsage, &procCpuUsage);
if (!suc) { if (!suc) {
monitorDebug("failed to get cpu usage"); mnDebug("failed to get cpu usage");
} }
if (sysCpuUsage <= procCpuUsage) { if (sysCpuUsage <= procCpuUsage) {
...@@ -294,14 +294,14 @@ static int32_t monitorBuildBandSql(char *sql) { ...@@ -294,14 +294,14 @@ static int32_t monitorBuildBandSql(char *sql) {
float bandSpeedKb = 0; float bandSpeedKb = 0;
bool suc = taosGetBandSpeed(&bandSpeedKb); bool suc = taosGetBandSpeed(&bandSpeedKb);
if (!suc) { if (!suc) {
monitorDebug("failed to get bandwidth speed"); mnDebug("failed to get bandwidth speed");
} }
return sprintf(sql, ", %f", bandSpeedKb); return sprintf(sql, ", %f", bandSpeedKb);
} }
static int32_t monitorBuildReqSql(char *sql) { static int32_t monitorBuildReqSql(char *sql) {
SDnodeStatisInfo info = dnodeGetStatisInfo(); SStatisInfo info = dnodeGetStatisInfo();
return sprintf(sql, ", %d, %d, %d)", info.httpReqNum, info.queryReqNum, info.submitReqNum); return sprintf(sql, ", %d, %d, %d)", info.httpReqNum, info.queryReqNum, info.submitReqNum);
} }
...@@ -309,7 +309,7 @@ static int32_t monitorBuildIoSql(char *sql) { ...@@ -309,7 +309,7 @@ static int32_t monitorBuildIoSql(char *sql) {
float readKB = 0, writeKB = 0; float readKB = 0, writeKB = 0;
bool suc = taosGetProcIO(&readKB, &writeKB); bool suc = taosGetProcIO(&readKB, &writeKB);
if (!suc) { if (!suc) {
monitorDebug("failed to get io info"); mnDebug("failed to get io info");
} }
return sprintf(sql, ", %f, %f", readKB, writeKB); return sprintf(sql, ", %f, %f", readKB, writeKB);
...@@ -332,19 +332,19 @@ static void monitorSaveSystemInfo() { ...@@ -332,19 +332,19 @@ static void monitorSaveSystemInfo() {
taos_free_result(res); taos_free_result(res);
if (code != 0) { if (code != 0) {
monitorError("failed to save system info, reason:%s, sql:%s", tstrerror(code), tsMonitor.sql); mnError("failed to save system info, reason:%s, sql:%s", tstrerror(code), tsMonitor.sql);
} else { } else {
monitorDebug("successfully to save system info, sql:%s", tsMonitor.sql); mnDebug("successfully to save system info, sql:%s", tsMonitor.sql);
} }
} }
static void montiorExecSqlCb(void *param, TAOS_RES *result, int32_t code) { static void montiorExecSqlCb(void *param, TAOS_RES *result, int32_t code) {
int32_t c = taos_errno(result); int32_t c = taos_errno(result);
if (c != TSDB_CODE_SUCCESS) { if (c != TSDB_CODE_SUCCESS) {
monitorError("save %s failed, reason:%s", (char *)param, tstrerror(c)); mnError("save %s failed, reason:%s", (char *)param, tstrerror(c));
} else { } else {
int32_t rows = taos_affected_rows(result); int32_t rows = taos_affected_rows(result);
monitorDebug("save %s succ, rows:%d", (char *)param, rows); mnDebug("save %s succ, rows:%d", (char *)param, rows);
} }
taos_free_result(result); taos_free_result(result);
...@@ -380,7 +380,7 @@ void monitorSaveAcctLog(SAcctMonitorObj *pMon) { ...@@ -380,7 +380,7 @@ void monitorSaveAcctLog(SAcctMonitorObj *pMon) {
pMon->totalConns, pMon->maxConns, pMon->totalConns, pMon->maxConns,
pMon->accessState); pMon->accessState);
monitorDebug("save account info, sql:%s", sql); mnDebug("save account info, sql:%s", sql);
taos_query_a(tsMonitor.conn, sql, montiorExecSqlCb, "account info"); taos_query_a(tsMonitor.conn, sql, montiorExecSqlCb, "account info");
} }
...@@ -401,13 +401,13 @@ void monitorSaveLog(int32_t level, const char *const format, ...) { ...@@ -401,13 +401,13 @@ void monitorSaveLog(int32_t level, const char *const format, ...) {
len += sprintf(sql + len, "', '%s')", tsLocalEp); len += sprintf(sql + len, "', '%s')", tsLocalEp);
sql[len++] = 0; sql[len++] = 0;
monitorDebug("save log, sql: %s", sql); mnDebug("save log, sql: %s", sql);
taos_query_a(tsMonitor.conn, sql, montiorExecSqlCb, "log"); taos_query_a(tsMonitor.conn, sql, montiorExecSqlCb, "log");
} }
void monitorExecuteSQL(char *sql) { void monitorExecuteSQL(char *sql) {
if (tsMonitor.state != MON_STATE_INITED) return; if (tsMonitor.state != MON_STATE_INITED) return;
monitorDebug("execute sql:%s", sql); mnDebug("execute sql:%s", sql);
taos_query_a(tsMonitor.conn, sql, montiorExecSqlCb, "sql"); taos_query_a(tsMonitor.conn, sql, montiorExecSqlCb, "sql");
} }
src/vnode/src/vnodeCfg.c
浏览文件 @ 4bc9284a
...@@ -22,7 +22,6 @@ ...@@ -22,7 +22,6 @@
#include "tsdb.h" #include "tsdb.h"
#include "dnode.h" #include "dnode.h"
#include "vnodeInt.h" #include "vnodeInt.h"
#include "vnodeVersion.h"
#include "vnodeCfg.h" #include "vnodeCfg.h"
static void vnodeLoadCfg(SVnodeObj *pVnode, SCreateVnodeMsg* vnodeMsg) { static void vnodeLoadCfg(SVnodeObj *pVnode, SCreateVnodeMsg* vnodeMsg) {
......
src/vnode/src/vnodeMain.c
浏览文件 @ 4bc9284a
...@@ -15,17 +15,11 @@ ...@@ -15,17 +15,11 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "os.h" #include "os.h"
#include "tcache.h"
#include "cJSON.h"
#include "dnode.h"
#include "hash.h"
#include "taoserror.h" #include "taoserror.h"
#include "taosmsg.h" #include "taosmsg.h"
#include "tglobal.h" #include "tglobal.h"
#include "trpc.h" #include "trpc.h"
#include "tsdb.h" #include "tsdb.h"
#include "ttimer.h"
#include "tutil.h" #include "tutil.h"
#include "vnode.h" #include "vnode.h"
#include "vnodeInt.h" #include "vnodeInt.h"
...@@ -34,7 +28,7 @@ ...@@ -34,7 +28,7 @@
#include "vnodeCfg.h" #include "vnodeCfg.h"
#include "vnodeVersion.h" #include "vnodeVersion.h"
static SHashObj*tsDnodeVnodesHash; static SHashObj*tsVnodesHash;
static void vnodeCleanUp(SVnodeObj *pVnode); static void vnodeCleanUp(SVnodeObj *pVnode);
static int vnodeProcessTsdbStatus(void *arg, int status); static int vnodeProcessTsdbStatus(void *arg, int status);
static uint32_t vnodeGetFileInfo(void *ahandle, char *name, uint32_t *index, uint32_t eindex, int64_t *size, uint64_t *fversion); static uint32_t vnodeGetFileInfo(void *ahandle, char *name, uint32_t *index, uint32_t eindex, int64_t *size, uint64_t *fversion);
...@@ -67,8 +61,8 @@ int32_t vnodeInitResources() { ...@@ -67,8 +61,8 @@ int32_t vnodeInitResources() {
vnodeInitWriteFp(); vnodeInitWriteFp();
vnodeInitReadFp(); vnodeInitReadFp();
tsDnodeVnodesHash = taosHashInit(TSDB_MIN_VNODES, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, true); tsVnodesHash = taosHashInit(TSDB_MIN_VNODES, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, true);
if (tsDnodeVnodesHash == NULL) { if (tsVnodesHash == NULL) {
vError("failed to init vnode list"); vError("failed to init vnode list");
return TSDB_CODE_VND_OUT_OF_MEMORY; return TSDB_CODE_VND_OUT_OF_MEMORY;
} }
...@@ -77,10 +71,10 @@ int32_t vnodeInitResources() { ...@@ -77,10 +71,10 @@ int32_t vnodeInitResources() {
} }
void vnodeCleanupResources() { void vnodeCleanupResources() {
if (tsDnodeVnodesHash != NULL) { if (tsVnodesHash != NULL) {
vDebug("vnode list is cleanup"); vDebug("vnode list is cleanup");
taosHashCleanup(tsDnodeVnodesHash); taosHashCleanup(tsVnodesHash);
tsDnodeVnodesHash = NULL; tsVnodesHash = NULL;
} }
syncCleanUp(); syncCleanUp();
...@@ -348,7 +342,7 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) { ...@@ -348,7 +342,7 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
pVnode->status = TAOS_VN_STATUS_READY; pVnode->status = TAOS_VN_STATUS_READY;
vDebug("vgId:%d, vnode is opened in %s, pVnode:%p", pVnode->vgId, rootDir, pVnode); vDebug("vgId:%d, vnode is opened in %s, pVnode:%p", pVnode->vgId, rootDir, pVnode);
taosHashPut(tsDnodeVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t), (char *)(&pVnode), sizeof(SVnodeObj *)); taosHashPut(tsVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t), (char *)(&pVnode), sizeof(SVnodeObj *));
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
...@@ -440,7 +434,7 @@ void vnodeRelease(void *pVnodeRaw) { ...@@ -440,7 +434,7 @@ void vnodeRelease(void *pVnodeRaw) {
tsem_destroy(&pVnode->sem); tsem_destroy(&pVnode->sem);
free(pVnode); free(pVnode);
int32_t count = taosHashGetSize(tsDnodeVnodesHash); int32_t count = taosHashGetSize(tsVnodesHash);
vDebug("vgId:%d, vnode is destroyed, vnodes:%d", vgId, count); vDebug("vgId:%d, vnode is destroyed, vnodes:%d", vgId, count);
} }
...@@ -457,7 +451,7 @@ static void vnodeIncRef(void *ptNode) { ...@@ -457,7 +451,7 @@ static void vnodeIncRef(void *ptNode) {
} }
void *vnodeAcquire(int32_t vgId) { void *vnodeAcquire(int32_t vgId) {
SVnodeObj **ppVnode = taosHashGetCB(tsDnodeVnodesHash, &vgId, sizeof(int32_t), vnodeIncRef, NULL, sizeof(void *)); SVnodeObj **ppVnode = taosHashGetCB(tsVnodesHash, &vgId, sizeof(int32_t), vnodeIncRef, NULL, sizeof(void *));
if (ppVnode == NULL || *ppVnode == NULL) { if (ppVnode == NULL || *ppVnode == NULL) {
terrno = TSDB_CODE_VND_INVALID_VGROUP_ID; terrno = TSDB_CODE_VND_INVALID_VGROUP_ID;
...@@ -496,7 +490,7 @@ static void vnodeBuildVloadMsg(SVnodeObj *pVnode, SStatusMsg *pStatus) { ...@@ -496,7 +490,7 @@ static void vnodeBuildVloadMsg(SVnodeObj *pVnode, SStatusMsg *pStatus) {
} }
int32_t vnodeGetVnodeList(int32_t vnodeList[], int32_t *numOfVnodes) { int32_t vnodeGetVnodeList(int32_t vnodeList[], int32_t *numOfVnodes) {
SHashMutableIterator *pIter = taosHashCreateIter(tsDnodeVnodesHash); SHashMutableIterator *pIter = taosHashCreateIter(tsVnodesHash);
while (taosHashIterNext(pIter)) { while (taosHashIterNext(pIter)) {
SVnodeObj **pVnode = taosHashIterGet(pIter); SVnodeObj **pVnode = taosHashIterGet(pIter);
if (pVnode == NULL) continue; if (pVnode == NULL) continue;
...@@ -517,7 +511,7 @@ int32_t vnodeGetVnodeList(int32_t vnodeList[], int32_t *numOfVnodes) { ...@@ -517,7 +511,7 @@ int32_t vnodeGetVnodeList(int32_t vnodeList[], int32_t *numOfVnodes) {
void vnodeBuildStatusMsg(void *param) { void vnodeBuildStatusMsg(void *param) {
SStatusMsg *pStatus = param; SStatusMsg *pStatus = param;
SHashMutableIterator *pIter = taosHashCreateIter(tsDnodeVnodesHash); SHashMutableIterator *pIter = taosHashCreateIter(tsVnodesHash);
while (taosHashIterNext(pIter)) { while (taosHashIterNext(pIter)) {
SVnodeObj **pVnode = taosHashIterGet(pIter); SVnodeObj **pVnode = taosHashIterGet(pIter);
...@@ -546,7 +540,7 @@ void vnodeSetAccess(SVgroupAccess *pAccess, int32_t numOfVnodes) { ...@@ -546,7 +540,7 @@ void vnodeSetAccess(SVgroupAccess *pAccess, int32_t numOfVnodes) {
static void vnodeCleanUp(SVnodeObj *pVnode) { static void vnodeCleanUp(SVnodeObj *pVnode) {
// remove from hash, so new messages wont be consumed // remove from hash, so new messages wont be consumed
taosHashRemove(tsDnodeVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t)); taosHashRemove(tsVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t));
if (pVnode->status != TAOS_VN_STATUS_INIT) { if (pVnode->status != TAOS_VN_STATUS_INIT) {
// it may be in updateing or reset state, then it shall wait // it may be in updateing or reset state, then it shall wait
......
src/vnode/src/vnodeRead.c
浏览文件 @ 4bc9284a
...@@ -15,13 +15,10 @@ ...@@ -15,13 +15,10 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#define _NON_BLOCKING_RETRIEVE 0 #define _NON_BLOCKING_RETRIEVE 0
#include "os.h" #include "os.h"
#include "tglobal.h" #include "tglobal.h"
#include "taoserror.h" #include "taoserror.h"
#include "taosmsg.h" #include "taosmsg.h"
#include "tcache.h"
#include "query.h" #include "query.h"
#include "trpc.h" #include "trpc.h"
#include "tsdb.h" #include "tsdb.h"
...@@ -29,9 +26,9 @@ ...@@ -29,9 +26,9 @@
#include "vnodeInt.h" #include "vnodeInt.h"
#include "tqueue.h" #include "tqueue.h"
static int32_t (*vnodeProcessReadMsgFp[TSDB_MSG_TYPE_MAX])(SVnodeObj *pVnode, SVReadMsg *pReadMsg); static int32_t (*vnodeProcessReadMsgFp[TSDB_MSG_TYPE_MAX])(SVnodeObj *pVnode, SVReadMsg *pRead);
static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg); static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pRead);
static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg); static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pRead);
static int32_t vnodeNotifyCurrentQhandle(void* handle, void* qhandle, int32_t vgId); static int32_t vnodeNotifyCurrentQhandle(void* handle, void* qhandle, int32_t vgId);
void vnodeInitReadFp(void) { void vnodeInitReadFp(void) {
...@@ -44,16 +41,16 @@ void vnodeInitReadFp(void) { ...@@ -44,16 +41,16 @@ void vnodeInitReadFp(void) {
// still required, or there will be a deadlock, so we don’t do any check here, but put the check codes before the // still required, or there will be a deadlock, so we don’t do any check here, but put the check codes before the
// request enters the queue // request enters the queue
// //
int32_t vnodeProcessRead(void *param, SVReadMsg *pReadMsg) { int32_t vnodeProcessRead(void *param, SVReadMsg *pRead) {
SVnodeObj *pVnode = (SVnodeObj *)param; SVnodeObj *pVnode = (SVnodeObj *)param;
int32_t msgType = pReadMsg->msgType; int32_t msgType = pRead->msgType;
if (vnodeProcessReadMsgFp[msgType] == NULL) { if (vnodeProcessReadMsgFp[msgType] == NULL) {
vDebug("vgId:%d, msgType:%s not processed, no handle", pVnode->vgId, taosMsg[msgType]); vDebug("vgId:%d, msgType:%s not processed, no handle", pVnode->vgId, taosMsg[msgType]);
return TSDB_CODE_VND_MSG_NOT_PROCESSED; return TSDB_CODE_VND_MSG_NOT_PROCESSED;
} }
return (*vnodeProcessReadMsgFp[msgType])(pVnode, pReadMsg); return (*vnodeProcessReadMsgFp[msgType])(pVnode, pRead);
} }
static int32_t vnodeCheckRead(void *param) { static int32_t vnodeCheckRead(void *param) {
...@@ -180,27 +177,27 @@ static void vnodeBuildNoResultQueryRsp(SRspRet *pRet) { ...@@ -180,27 +177,27 @@ static void vnodeBuildNoResultQueryRsp(SRspRet *pRet) {
pRsp->completed = true; pRsp->completed = true;
} }
static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pRead) {
void * pCont = pReadMsg->pCont; void * pCont = pRead->pCont;
int32_t contLen = pReadMsg->contLen; int32_t contLen = pRead->contLen;
SRspRet *pRet = &pReadMsg->rspRet; SRspRet *pRet = &pRead->rspRet;
SQueryTableMsg *pQueryTableMsg = (SQueryTableMsg *)pCont; SQueryTableMsg *pQueryTableMsg = (SQueryTableMsg *)pCont;
memset(pRet, 0, sizeof(SRspRet)); memset(pRet, 0, sizeof(SRspRet));
// qHandle needs to be freed correctly // qHandle needs to be freed correctly
if (pReadMsg->code == TSDB_CODE_RPC_NETWORK_UNAVAIL) { if (pRead->code == TSDB_CODE_RPC_NETWORK_UNAVAIL) {
SRetrieveTableMsg *killQueryMsg = (SRetrieveTableMsg *)pReadMsg->pCont; SRetrieveTableMsg *killQueryMsg = (SRetrieveTableMsg *)pRead->pCont;
killQueryMsg->free = htons(killQueryMsg->free); killQueryMsg->free = htons(killQueryMsg->free);
killQueryMsg->qhandle = htobe64(killQueryMsg->qhandle); killQueryMsg->qhandle = htobe64(killQueryMsg->qhandle);
vWarn("QInfo:%p connection %p broken, kill query", (void *)killQueryMsg->qhandle, pReadMsg->rpcHandle); vWarn("QInfo:%p connection %p broken, kill query", (void *)killQueryMsg->qhandle, pRead->rpcHandle);
assert(pReadMsg->contLen > 0 && killQueryMsg->free == 1); assert(pRead->contLen > 0 && killQueryMsg->free == 1);
void **qhandle = qAcquireQInfo(pVnode->qMgmt, (uint64_t)killQueryMsg->qhandle); void **qhandle = qAcquireQInfo(pVnode->qMgmt, (uint64_t)killQueryMsg->qhandle);
if (qhandle == NULL || *qhandle == NULL) { if (qhandle == NULL || *qhandle == NULL) {
vWarn("QInfo:%p invalid qhandle, no matched query handle, conn:%p", (void *)killQueryMsg->qhandle, vWarn("QInfo:%p invalid qhandle, no matched query handle, conn:%p", (void *)killQueryMsg->qhandle,
pReadMsg->rpcHandle); pRead->rpcHandle);
} else { } else {
assert(*qhandle == (void *)killQueryMsg->qhandle); assert(*qhandle == (void *)killQueryMsg->qhandle);
...@@ -242,9 +239,9 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { ...@@ -242,9 +239,9 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) {
} }
if (handle != NULL && if (handle != NULL &&
vnodeNotifyCurrentQhandle(pReadMsg->rpcHandle, *handle, pVnode->vgId) != TSDB_CODE_SUCCESS) { vnodeNotifyCurrentQhandle(pRead->rpcHandle, *handle, pVnode->vgId) != TSDB_CODE_SUCCESS) {
vError("vgId:%d, QInfo:%p, query discarded since link is broken, %p", pVnode->vgId, *handle, vError("vgId:%d, QInfo:%p, query discarded since link is broken, %p", pVnode->vgId, *handle,
pReadMsg->rpcHandle); pRead->rpcHandle);
pRsp->code = TSDB_CODE_RPC_NETWORK_UNAVAIL; pRsp->code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
qReleaseQInfo(pVnode->qMgmt, (void **)&handle, true); qReleaseQInfo(pVnode->qMgmt, (void **)&handle, true);
return pRsp->code; return pRsp->code;
...@@ -255,7 +252,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { ...@@ -255,7 +252,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) {
if (handle != NULL) { if (handle != NULL) {
vDebug("vgId:%d, QInfo:%p, dnode query msg disposed, create qhandle and returns to app", vgId, *handle); vDebug("vgId:%d, QInfo:%p, dnode query msg disposed, create qhandle and returns to app", vgId, *handle);
code = vnodePutItemIntoReadQueue(pVnode, handle, pReadMsg->rpcHandle); code = vnodePutItemIntoReadQueue(pVnode, handle, pRead->rpcHandle);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
pRsp->code = code; pRsp->code = code;
qReleaseQInfo(pVnode->qMgmt, (void **)&handle, true); qReleaseQInfo(pVnode->qMgmt, (void **)&handle, true);
...@@ -264,7 +261,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { ...@@ -264,7 +261,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) {
} }
} else { } else {
assert(pCont != NULL); assert(pCont != NULL);
void **qhandle = (void **)pReadMsg->qhandle; void **qhandle = (void **)pRead->qhandle;
vDebug("vgId:%d, QInfo:%p, dnode continues to exec query", pVnode->vgId, *qhandle); vDebug("vgId:%d, QInfo:%p, dnode continues to exec query", pVnode->vgId, *qhandle);
...@@ -276,14 +273,14 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { ...@@ -276,14 +273,14 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) {
// build query rsp, the retrieve request has reached here already // build query rsp, the retrieve request has reached here already
if (buildRes) { if (buildRes) {
// update the connection info according to the retrieve connection // update the connection info according to the retrieve connection
pReadMsg->rpcHandle = qGetResultRetrieveMsg(*qhandle); pRead->rpcHandle = qGetResultRetrieveMsg(*qhandle);
assert(pReadMsg->rpcHandle != NULL); assert(pRead->rpcHandle != NULL);
vDebug("vgId:%d, QInfo:%p, start to build retrieval rsp after query paused, %p", pVnode->vgId, *qhandle, vDebug("vgId:%d, QInfo:%p, start to build retrieval rsp after query paused, %p", pVnode->vgId, *qhandle,
pReadMsg->rpcHandle); pRead->rpcHandle);
// set the real rsp error code // set the real rsp error code
pReadMsg->code = vnodeDumpQueryResult(&pRead->rspRet, pVnode, qhandle, &freehandle, pReadMsg->rpcHandle); pRead->code = vnodeDumpQueryResult(&pRead->rspRet, pVnode, qhandle, &freehandle, pRead->rpcHandle);
// NOTE: set return code to be TSDB_CODE_QRY_HAS_RSP to notify dnode to return msg to client // NOTE: set return code to be TSDB_CODE_QRY_HAS_RSP to notify dnode to return msg to client
code = TSDB_CODE_QRY_HAS_RSP; code = TSDB_CODE_QRY_HAS_RSP;
...@@ -308,16 +305,16 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { ...@@ -308,16 +305,16 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) {
return code; return code;
} }
static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pRead) {
void * pCont = pReadMsg->pCont; void * pCont = pRead->pCont;
SRspRet *pRet = &pReadMsg->rspRet; SRspRet *pRet = &pRead->rspRet;
SRetrieveTableMsg *pRetrieve = pCont; SRetrieveTableMsg *pRetrieve = pCont;
pRetrieve->free = htons(pRetrieve->free); pRetrieve->free = htons(pRetrieve->free);
pRetrieve->qhandle = htobe64(pRetrieve->qhandle); pRetrieve->qhandle = htobe64(pRetrieve->qhandle);
vDebug("vgId:%d, QInfo:%p, retrieve msg is disposed, free:%d, conn:%p", pVnode->vgId, (void *)pRetrieve->qhandle, vDebug("vgId:%d, QInfo:%p, retrieve msg is disposed, free:%d, conn:%p", pVnode->vgId, (void *)pRetrieve->qhandle,
pRetrieve->free, pReadMsg->rpcHandle); pRetrieve->free, pRead->rpcHandle);
memset(pRet, 0, sizeof(SRspRet)); memset(pRet, 0, sizeof(SRspRet));
...@@ -348,8 +345,8 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { ...@@ -348,8 +345,8 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) {
} }
// register the qhandle to connect to quit query immediate if connection is broken // register the qhandle to connect to quit query immediate if connection is broken
if (vnodeNotifyCurrentQhandle(pReadMsg->rpcHandle, *handle, pVnode->vgId) != TSDB_CODE_SUCCESS) { if (vnodeNotifyCurrentQhandle(pRead->rpcHandle, *handle, pVnode->vgId) != TSDB_CODE_SUCCESS) {
vError("vgId:%d, QInfo:%p, retrieve discarded since link is broken, %p", pVnode->vgId, *handle, pReadMsg->rpcHandle); vError("vgId:%d, QInfo:%p, retrieve discarded since link is broken, %p", pVnode->vgId, *handle, pRead->rpcHandle);
code = TSDB_CODE_RPC_NETWORK_UNAVAIL; code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
qKillQuery(*handle); qKillQuery(*handle);
qReleaseQInfo(pVnode->qMgmt, (void **)&handle, true); qReleaseQInfo(pVnode->qMgmt, (void **)&handle, true);
...@@ -359,7 +356,7 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { ...@@ -359,7 +356,7 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) {
bool freeHandle = true; bool freeHandle = true;
bool buildRes = false; bool buildRes = false;
code = qRetrieveQueryResultInfo(*handle, &buildRes, pReadMsg->rpcHandle); code = qRetrieveQueryResultInfo(*handle, &buildRes, pRead->rpcHandle);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
// TODO handle malloc failure // TODO handle malloc failure
pRet->rsp = (SRetrieveTableRsp *)rpcMallocCont(sizeof(SRetrieveTableRsp)); pRet->rsp = (SRetrieveTableRsp *)rpcMallocCont(sizeof(SRetrieveTableRsp));
...@@ -370,7 +367,7 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { ...@@ -370,7 +367,7 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) {
assert(buildRes == true); assert(buildRes == true);
#if _NON_BLOCKING_RETRIEVE #if _NON_BLOCKING_RETRIEVE
if (!buildRes) { if (!buildRes) {
assert(pReadMsg->rpcHandle != NULL); assert(pRead->rpcHandle != NULL);
qReleaseQInfo(pVnode->qMgmt, (void **)&handle, false); qReleaseQInfo(pVnode->qMgmt, (void **)&handle, false);
return TSDB_CODE_QRY_NOT_READY; return TSDB_CODE_QRY_NOT_READY;
...@@ -378,7 +375,7 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) { ...@@ -378,7 +375,7 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pReadMsg) {
#endif #endif
// ahandle is the sqlObj pointer // ahandle is the sqlObj pointer
code = vnodeDumpQueryResult(pRet, pVnode, handle, &freeHandle, pReadMsg->rpcHandle); code = vnodeDumpQueryResult(pRet, pVnode, handle, &freeHandle, pRead->rpcHandle);
} }
// If qhandle is not added into vread queue, the query should be completed already or paused with error. // If qhandle is not added into vread queue, the query should be completed already or paused with error.
......
src/vnode/src/vnodeWrite.c
浏览文件 @ 4bc9284a
...@@ -19,7 +19,6 @@ ...@@ -19,7 +19,6 @@
#include "taoserror.h" #include "taoserror.h"
#include "tqueue.h" #include "tqueue.h"
#include "trpc.h" #include "trpc.h"
#include "tutil.h"
#include "tsdb.h" #include "tsdb.h"
#include "twal.h" #include "twal.h"
#include "tsync.h" #include "tsync.h"
...@@ -185,7 +184,7 @@ static int32_t vnodeProcessAlterTableMsg(SVnodeObj *pVnode, void *pCont, SRspRet ...@@ -185,7 +184,7 @@ static int32_t vnodeProcessAlterTableMsg(SVnodeObj *pVnode, void *pCont, SRspRet
static int32_t vnodeProcessDropStableMsg(SVnodeObj *pVnode, void *pCont, SRspRet *pRet) { static int32_t vnodeProcessDropStableMsg(SVnodeObj *pVnode, void *pCont, SRspRet *pRet) {
SDropSTableMsg *pTable = pCont; SDropSTableMsg *pTable = pCont;
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
vDebug("vgId:%d, stable:%s, start to drop", pVnode->vgId, pTable->tableId); vDebug("vgId:%d, stable:%s, start to drop", pVnode->vgId, pTable->tableId);
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或