未验证 提交 4c65b082 编写于 作者: S Shengliang Guan 提交者: GitHub

Merge pull request #8525 from taosdata/feature/dnode3

Feature/dnode3
...@@ -21,8 +21,8 @@ extern "C" { ...@@ -21,8 +21,8 @@ extern "C" {
#endif #endif
#include "dnodeInt.h" #include "dnodeInt.h"
int32_t dnodeInitMsg(); int32_t dnodeInitDnode();
void dnodeCleanupMsg(); void dnodeCleanupDnode();
void dnodeProcessStatusRsp(SRpcMsg *pMsg); void dnodeProcessStatusRsp(SRpcMsg *pMsg);
void dnodeProcessStartupReq(SRpcMsg *pMsg); void dnodeProcessStartupReq(SRpcMsg *pMsg);
void dnodeProcessConfigDnodeReq(SRpcMsg *pMsg); void dnodeProcessConfigDnodeReq(SRpcMsg *pMsg);
......
...@@ -24,7 +24,6 @@ extern "C" { ...@@ -24,7 +24,6 @@ extern "C" {
#include "tglobal.h" #include "tglobal.h"
#include "tlog.h" #include "tlog.h"
#include "trpc.h" #include "trpc.h"
#include "ttimer.h"
extern int32_t dDebugFlag; extern int32_t dDebugFlag;
......
...@@ -21,6 +21,9 @@ extern "C" { ...@@ -21,6 +21,9 @@ extern "C" {
#endif #endif
#include "dnodeInt.h" #include "dnodeInt.h"
int32_t dnodeInitMnode();
void dnodeCleanupMnode();
void dnodeProcessCreateMnodeReq(SRpcMsg *pMsg); void dnodeProcessCreateMnodeReq(SRpcMsg *pMsg);
#ifdef __cplusplus #ifdef __cplusplus
......
...@@ -21,12 +21,8 @@ extern "C" { ...@@ -21,12 +21,8 @@ extern "C" {
#endif #endif
#include "dnodeInt.h" #include "dnodeInt.h"
int32_t dnodeInitMsg(); int32_t dnodeInitVnodes();
void dnodeCleanupMsg(); void dnodeCleanupVnodes();
void dnodeProcessStatusRsp(SRpcMsg *pMsg);
void dnodeProcessStartupReq(SRpcMsg *pMsg);
void dnodeProcessCreateMnodeReq(SRpcMsg *pMsg);
void dnodeProcessConfigDnodeReq(SRpcMsg *pMsg);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
...@@ -30,7 +30,7 @@ static struct { ...@@ -30,7 +30,7 @@ static struct {
pthread_mutex_t mutex; pthread_mutex_t mutex;
} tsConfig; } tsConfig;
vstaticoid dnodeGetEpSetForPeer(SRpcEpSet *epSet) { void dnodeGetEpSetForPeer(SRpcEpSet *epSet) {
pthread_mutex_lock(&tsConfig.mutex); pthread_mutex_lock(&tsConfig.mutex);
*epSet = tsConfig.mnodeEpSetForPeer; *epSet = tsConfig.mnodeEpSetForPeer;
pthread_mutex_unlock(&tsConfig.mutex); pthread_mutex_unlock(&tsConfig.mutex);
......
...@@ -25,7 +25,7 @@ static struct { ...@@ -25,7 +25,7 @@ static struct {
pthread_t *threadId; pthread_t *threadId;
bool stop; bool stop;
uint32_t rebootTime; uint32_t rebootTime;
} tsMsg; } tsDnode;
static void dnodeSendStatusMsg() { static void dnodeSendStatusMsg() {
int32_t contLen = sizeof(SStatusMsg) + TSDB_MAX_VNODES * sizeof(SVnodeLoad); int32_t contLen = sizeof(SStatusMsg) + TSDB_MAX_VNODES * sizeof(SVnodeLoad);
...@@ -39,7 +39,7 @@ static void dnodeSendStatusMsg() { ...@@ -39,7 +39,7 @@ static void dnodeSendStatusMsg() {
pStatus->dnodeId = htonl(dnodeGetDnodeId()); pStatus->dnodeId = htonl(dnodeGetDnodeId());
tstrncpy(pStatus->dnodeEp, tsLocalEp, TSDB_EP_LEN); tstrncpy(pStatus->dnodeEp, tsLocalEp, TSDB_EP_LEN);
pStatus->clusterId = htobe64(dnodeGetClusterId()); pStatus->clusterId = htobe64(dnodeGetClusterId());
pStatus->lastReboot = htonl(tsMsg.rebootTime); pStatus->lastReboot = htonl(tsDnode.rebootTime);
pStatus->numOfCores = htonl(tsNumOfCores); pStatus->numOfCores = htonl(tsNumOfCores);
pStatus->diskAvailable = tsAvailDataDirGB; pStatus->diskAvailable = tsAvailDataDirGB;
...@@ -93,17 +93,17 @@ void dnodeProcessStatusRsp(SRpcMsg *pMsg) { ...@@ -93,17 +93,17 @@ void dnodeProcessStatusRsp(SRpcMsg *pMsg) {
static void *dnodeThreadRoutine(void *param) { static void *dnodeThreadRoutine(void *param) {
int32_t ms = tsStatusInterval * 1000; int32_t ms = tsStatusInterval * 1000;
while (!tsMsg.stop) { while (!tsDnode.stop) {
taosMsleep(ms); taosMsleep(ms);
dnodeSendStatusMsg(); dnodeSendStatusMsg();
} }
} }
int32_t dnodeInitMsg() { int32_t dnodeInitDnode() {
tsMsg.stop = false; tsDnode.stop = false;
tsMsg.rebootTime = taosGetTimestampSec(); tsDnode.rebootTime = taosGetTimestampSec();
tsMsg.threadId = taosCreateThread(dnodeThreadRoutine, NULL); tsDnode.threadId = taosCreateThread(dnodeThreadRoutine, NULL);
if (tsMsg.threadId == NULL) { if (tsDnode.threadId == NULL) {
return -1; return -1;
} }
...@@ -111,11 +111,11 @@ int32_t dnodeInitMsg() { ...@@ -111,11 +111,11 @@ int32_t dnodeInitMsg() {
return 0; return 0;
} }
void dnodeCleanupMsg() { void dnodeCleanupDnode() {
if (tsMsg.threadId != NULL) { if (tsDnode.threadId != NULL) {
tsMsg.stop = true; tsDnode.stop = true;
taosDestoryThread(tsMsg.threadId); taosDestoryThread(tsDnode.threadId);
tsMsg.threadId = NULL; tsDnode.threadId = NULL;
} }
dInfo("dnode msg is cleanuped"); dInfo("dnode msg is cleanuped");
......
...@@ -17,62 +17,44 @@ ...@@ -17,62 +17,44 @@
#include "dnodeCheck.h" #include "dnodeCheck.h"
#include "dnodeConfig.h" #include "dnodeConfig.h"
#include "dnodeDnode.h" #include "dnodeDnode.h"
#include "dnodeMnode.h"
#include "dnodeTransport.h" #include "dnodeTransport.h"
#include "mnode.h" #include "dnodeVnodes.h"
#include "sync.h" #include "sync.h"
#include "tcache.h" #include "tcache.h"
#include "tconfig.h" #include "tconfig.h"
#include "tnote.h" #include "tnote.h"
#include "tstep.h" #include "tstep.h"
#include "vnode.h"
#include "wal.h" #include "wal.h"
static struct { static struct {
EDnStat runStatus; EDnStat runStatus;
SStartupStep startup; SStartupStep startup;
SSteps *steps; SSteps *steps;
} tsDnode; } tsInt;
EDnStat dnodeGetRunStat() { return tsDnode.runStatus; } EDnStat dnodeGetRunStat() { return tsInt.runStatus; }
void dnodeSetRunStat(EDnStat stat) { tsDnode.runStatus = stat; } void dnodeSetRunStat(EDnStat stat) { tsInt.runStatus = stat; }
void dnodeReportStartup(char *name, char *desc) { static void dnodeReportStartup(char *name, char *desc) {
SStartupStep *startup = &tsDnode.startup; SStartupStep *startup = &tsInt.startup;
tstrncpy(startup->name, name, strlen(startup->name)); tstrncpy(startup->name, name, strlen(startup->name));
tstrncpy(startup->desc, desc, strlen(startup->desc)); tstrncpy(startup->desc, desc, strlen(startup->desc));
startup->finished = 0; startup->finished = 0;
} }
static void dnodeReportStartupFinished(char *name, char *desc) { static void dnodeReportStartupFinished(char *name, char *desc) {
SStartupStep *startup = &tsDnode.startup; SStartupStep *startup = &tsInt.startup;
tstrncpy(startup->name, name, strlen(startup->name)); tstrncpy(startup->name, name, strlen(startup->name));
tstrncpy(startup->desc, desc, strlen(startup->desc)); tstrncpy(startup->desc, desc, strlen(startup->desc));
startup->finished = 1; startup->finished = 1;
} }
void dnodeGetStartup(SStartupStep *pStep) { memcpy(pStep, &tsDnode.startup, sizeof(SStartupStep)); } void dnodeGetStartup(SStartupStep *pStep) { memcpy(pStep, &tsInt.startup, sizeof(SStartupStep)); }
static int32_t dnodeInitVnode() {
return vnodeInit();
}
static int32_t dnodeInitMnode() {
SMnodePara para;
para.fp.GetDnodeEp = dnodeGetEp;
para.fp.SendMsgToDnode = dnodeSendMsgToDnode;
para.fp.SendMsgToMnode = dnodeSendMsgToMnode;
para.fp.SendRedirectMsg = dnodeSendRedirectMsg;
para.dnodeId = dnodeGetDnodeId();
para.clusterId = dnodeGetClusterId();
return mnodeInit(para);
}
static int32_t dnodeInitTfs() {}
static int32_t dnodeInitMain() { static int32_t dnodeInitMain() {
tsDnode.runStatus = DN_RUN_STAT_STOPPED; tsInt.runStatus = DN_RUN_STAT_STOPPED;
tscEmbedded = 1; tscEmbedded = 1;
taosIgnSIGPIPE(); taosIgnSIGPIPE();
taosBlockSIGPIPE(); taosBlockSIGPIPE();
...@@ -168,17 +150,17 @@ int32_t dnodeInit() { ...@@ -168,17 +150,17 @@ int32_t dnodeInit() {
taosStepAdd(steps, "dnode-dir", dnodeInitDir, dnodeCleanupDir); taosStepAdd(steps, "dnode-dir", dnodeInitDir, dnodeCleanupDir);
taosStepAdd(steps, "dnode-check", dnodeInitCheck, dnodeCleanupCheck); taosStepAdd(steps, "dnode-check", dnodeInitCheck, dnodeCleanupCheck);
taosStepAdd(steps, "dnode-rpc", rpcInit, rpcCleanup); taosStepAdd(steps, "dnode-rpc", rpcInit, rpcCleanup);
taosStepAdd(steps, "dnode-tfs", dnodeInitTfs, NULL); taosStepAdd(steps, "dnode-tfs", NULL, NULL);
taosStepAdd(steps, "dnode-wal", walInit, walCleanUp); taosStepAdd(steps, "dnode-wal", walInit, walCleanUp);
taosStepAdd(steps, "dnode-sync", syncInit, syncCleanUp); taosStepAdd(steps, "dnode-sync", syncInit, syncCleanUp);
taosStepAdd(steps, "dnode-eps", dnodeInitConfig, dnodeCleanupConfig); taosStepAdd(steps, "dnode-config", dnodeInitConfig, dnodeCleanupConfig);
taosStepAdd(steps, "dnode-vnode", dnodeInitVnode, vnodeCleanup); taosStepAdd(steps, "dnode-vnodes", dnodeInitVnodes, dnodeCleanupVnodes);
taosStepAdd(steps, "dnode-mnode", dnodeInitMnode, mnodeCleanup); taosStepAdd(steps, "dnode-mnode", dnodeInitMnode, dnodeCleanupMnode);
taosStepAdd(steps, "dnode-trans", dnodeInitTrans, dnodeCleanupTrans); taosStepAdd(steps, "dnode-trans", dnodeInitTrans, dnodeCleanupTrans);
taosStepAdd(steps, "dnode-msg", dnodeInitMsg, dnodeCleanupMsg); taosStepAdd(steps, "dnode-dnode", dnodeInitDnode, dnodeCleanupDnode);
tsDnode.steps = steps; tsInt.steps = steps;
taosStepExec(tsDnode.steps); taosStepExec(tsInt.steps);
dnodeSetRunStat(DN_RUN_STAT_RUNNING); dnodeSetRunStat(DN_RUN_STAT_RUNNING);
dnodeReportStartupFinished("TDengine", "initialized successfully"); dnodeReportStartupFinished("TDengine", "initialized successfully");
...@@ -190,15 +172,7 @@ int32_t dnodeInit() { ...@@ -190,15 +172,7 @@ int32_t dnodeInit() {
void dnodeCleanup() { void dnodeCleanup() {
if (dnodeGetRunStat() != DN_RUN_STAT_STOPPED) { if (dnodeGetRunStat() != DN_RUN_STAT_STOPPED) {
dnodeSetRunStat(DN_RUN_STAT_STOPPED); dnodeSetRunStat(DN_RUN_STAT_STOPPED);
taosStepCleanup(tsDnode.steps); taosStepCleanup(tsInt.steps);
tsDnode.steps = NULL; tsInt.steps = NULL;
} }
} }
// tsVnode.msgFp[TSDB_MSG_TYPE_MD_CREATE_VNODE] = vnodeProcessMgmtMsg;
// tsVnode.msgFp[TSDB_MSG_TYPE_MD_ALTER_VNODE] = vnodeProcessMgmtMsg;
// tsVnode.msgFp[TSDB_MSG_TYPE_MD_SYNC_VNODE] = vnodeProcessMgmtMsg;
// tsVnode.msgFp[TSDB_MSG_TYPE_MD_COMPACT_VNODE] = vnodeProcessMgmtMsg;
// tsVnode.msgFp[TSDB_MSG_TYPE_MD_DROP_VNODE] = vnodeProcessMgmtMsg;
// tsVnode.msgFp[TSDB_MSG_TYPE_MD_ALTER_STREAM] = vnodeProcessMgmtMsg;
\ No newline at end of file
...@@ -12,8 +12,8 @@ ...@@ -12,8 +12,8 @@
* You should have received a copy of the GNU Affero General Public License * You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#include "os.h"
#include "ulog.h" #define _DEFAULT_SOURCE
#include "dnodeInt.h" #include "dnodeInt.h"
static bool stop = false; static bool stop = false;
...@@ -35,8 +35,6 @@ int main(int argc, char const *argv[]) { ...@@ -35,8 +35,6 @@ int main(int argc, char const *argv[]) {
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
} }
dInfo("Started TDengine service successfully.");
while (!stop) { while (!stop) {
taosMsleep(100); taosMsleep(100);
} }
......
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "dnodeMnode.h"
#include "dnodeConfig.h"
#include "dnodeTransport.h"
#include "mnode.h"
int32_t dnodeInitMnode() {
SMnodePara para;
para.fp.GetDnodeEp = dnodeGetEp;
para.fp.SendMsgToDnode = dnodeSendMsgToDnode;
para.fp.SendMsgToMnode = dnodeSendMsgToMnode;
para.fp.SendRedirectMsg = dnodeSendRedirectMsg;
para.dnodeId = dnodeGetDnodeId();
para.clusterId = dnodeGetClusterId();
return mnodeInit(para);
}
void dnodeCleanupMnode() { mnodeCleanup(); }
\ No newline at end of file
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "dnodeDnode.h"
#include "vnode.h"
int32_t dnodeInitVnodes() { return vnodeInit(); }
void dnodeCleanupVnodes() { vnodeCleanup(); }
\ No newline at end of file
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册