Skip to content

TDengine
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
TDengine
TDengine
未验证 提交 bddfff72 编写于 作者: S Shengliang Guan 提交者： GitHub
浏览文件
操作

Merge pull request #9225 from taosdata/feature/dnode3 

create vnode in dnode
上级 3fe3222f 0be539d5
展开全部 显示空白变更内容
内联 并排
Showing with 739 addition and 493 deletion
include/common/taosmsg.h
浏览文件 @ bddfff72
......@@ -591,8 +591,8 @@ typedef struct {
int32_t daysToKeep0;
int32_t daysToKeep1;
int32_t daysToKeep2;
int32_t minRowsPerFileBlock;
int32_t maxRowsPerFileBlock;
int32_t minRows;
int32_t maxRows;
int32_t commitTime;
int32_t fsyncPeriod;
int8_t walLevel;
......@@ -706,7 +706,7 @@ typedef struct {
SVnodeLoad data[];
} SVnodeLoads;
typedef struct SStatusMsg {
typedef struct {
int32_t sver;
int32_t dnodeId;
int32_t clusterId;
......@@ -756,6 +756,7 @@ typedef struct {
int32_t dnodeId;
char db[TSDB_FULL_DB_NAME_LEN];
uint64_t dbUid;
int32_t vgVersion;
int32_t cacheBlockSize;
int32_t totalBlocks;
int32_t daysPerFile;
......
include/dnode/vnode/vnode.h
浏览文件 @ bddfff72
......@@ -197,8 +197,8 @@ void *vnodeParseReq(void *buf, SVnodeReq *pReq, uint8_t type);
// int32_t daysToKeep0;
// int32_t daysToKeep1;
// int32_t daysToKeep2;
// int32_t minRowsPerFileBlock;
// int32_t maxRowsPerFileBlock;
// int32_t minRows;
// int32_t maxRows;
// int8_t precision; // time resolution
// int8_t compression;
// int8_t cacheLastRow;
......
include/util/tthread.h
浏览文件 @ bddfff72
......@@ -24,7 +24,7 @@ extern "C" {
#include "tdef.h"
// create new thread
pthread_t* taosCreateThread( void *(*__start_routine) (void *), void* param);
pthread_t* taosCreateThread(void* (*__start_routine)(void*), void* param);
// destory thread
bool taosDestoryThread(pthread_t* pthread);
// thread running return true
......
source/dnode/mgmt/impl/src/dnode.c
浏览文件 @ bddfff72
......@@ -176,6 +176,12 @@ SDnode *dndInit(SDnodeOpt *pOption) {
return NULL;
}
if (vnodeInit(1) != 0) {
dError("failed to init vnode env");
dndCleanup(pDnode);
return NULL;
}
if (dndInitDnode(pDnode) != 0) {
dError("failed to init dnode");
dndCleanup(pDnode);
......@@ -222,8 +228,10 @@ void dndCleanup(SDnode *pDnode) {
dndCleanupMnode(pDnode);
dndCleanupVnodes(pDnode);
dndCleanupDnode(pDnode);
vnodeClear();
walCleanUp();
rpcCleanup();
dndCleanupEnv(pDnode);
free(pDnode);
dInfo("TDengine is cleaned up successfully");
......
source/dnode/mgmt/impl/test/db/db.cpp
浏览文件 @ bddfff72
......@@ -210,8 +210,8 @@ TEST_F(DndTestDb, 02_Create_Alter_Drop_Db) {
pReq->daysToKeep0 = htonl(3650);
pReq->daysToKeep1 = htonl(3650);
pReq->daysToKeep2 = htonl(3650);
pReq->minRowsPerFileBlock = htonl(100);
pReq->maxRowsPerFileBlock = htonl(4096);
pReq->minRows = htonl(100);
pReq->maxRows = htonl(4096);
pReq->commitTime = htonl(3600);
pReq->fsyncPeriod = htonl(3000);
pReq->walLevel = 1;
......@@ -375,8 +375,8 @@ TEST_F(DndTestDb, 03_Create_Use_Restart_Use_Db) {
pReq->daysToKeep0 = htonl(3650);
pReq->daysToKeep1 = htonl(3650);
pReq->daysToKeep2 = htonl(3650);
pReq->minRowsPerFileBlock = htonl(100);
pReq->maxRowsPerFileBlock = htonl(4096);
pReq->minRows = htonl(100);
pReq->maxRows = htonl(4096);
pReq->commitTime = htonl(3600);
pReq->fsyncPeriod = htonl(3000);
pReq->walLevel = 1;
......
source/dnode/mgmt/impl/test/stb/stb.cpp
浏览文件 @ bddfff72
......@@ -187,8 +187,8 @@ TEST_F(DndTestStb, 01_Create_Show_Meta_Drop_Restart_Stb) {
pReq->daysToKeep0 = htonl(3650);
pReq->daysToKeep1 = htonl(3650);
pReq->daysToKeep2 = htonl(3650);
pReq->minRowsPerFileBlock = htonl(100);
pReq->maxRowsPerFileBlock = htonl(4096);
pReq->minRows = htonl(100);
pReq->maxRows = htonl(4096);
pReq->commitTime = htonl(3600);
pReq->fsyncPeriod = htonl(3000);
pReq->walLevel = 1;
......
source/dnode/mgmt/impl/test/sut/deploy.cpp
浏览文件 @ bddfff72
......@@ -16,7 +16,7 @@
#include "deploy.h"
void initLog(const char* path) {
dDebugFlag = 143;
dDebugFlag = 207;
vDebugFlag = 0;
mDebugFlag = 207;
cDebugFlag = 0;
......
source/dnode/mgmt/impl/test/vgroup/vgroup.cpp
浏览文件 @ bddfff72
......@@ -176,14 +176,15 @@ SServer* DndTestVgroup::pServer;
SClient* DndTestVgroup::pClient;
int32_t DndTestVgroup::connId;
TEST_F(DndTestVgroup, 01_Create_Restart_Drop_Vnode) {
{
for (int i = 0; i < 3; ++i) {
SCreateVnodeMsg* pReq = (SCreateVnodeMsg*)rpcMallocCont(sizeof(SCreateVnodeMsg));
pReq->vgId = htonl(2);
pReq->dnodeId = htonl(1);
strcpy(pReq->db, "1.d1");
pReq->dbUid = htobe64(9527);
pReq->vgVersion = htonl(1);
pReq->cacheBlockSize = htonl(16);
pReq->totalBlocks = htonl(10);
pReq->daysPerFile = htonl(10);
......@@ -217,8 +218,70 @@ TEST_F(DndTestVgroup, 01_Create_Restart_Drop_Vnode) {
SRpcMsg* pMsg = pClient->pRsp;
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, 0);
taosMsleep(1000000);
}
}
}
{
for (int i = 0; i < 3; ++i) {
SAlterVnodeMsg* pReq = (SAlterVnodeMsg*)rpcMallocCont(sizeof(SAlterVnodeMsg));
pReq->vgId = htonl(2);
pReq->dnodeId = htonl(1);
strcpy(pReq->db, "1.d1");
pReq->dbUid = htobe64(9527);
pReq->vgVersion = htonl(2);
pReq->cacheBlockSize = htonl(16);
pReq->totalBlocks = htonl(10);
pReq->daysPerFile = htonl(10);
pReq->daysToKeep0 = htonl(3650);
pReq->daysToKeep1 = htonl(3650);
pReq->daysToKeep2 = htonl(3650);
pReq->minRows = htonl(100);
pReq->minRows = htonl(4096);
pReq->commitTime = htonl(3600);
pReq->fsyncPeriod = htonl(3000);
pReq->walLevel = 1;
pReq->precision = 0;
pReq->compression = 2;
pReq->replica = 1;
pReq->quorum = 1;
pReq->update = 0;
pReq->cacheLastRow = 0;
pReq->selfIndex = 0;
for (int r = 0; r < pReq->replica; ++r) {
SReplica* pReplica = &pReq->replicas[r];
pReplica->id = htonl(1);
pReplica->port = htons(9150);
}
SRpcMsg rpcMsg = {0};
rpcMsg.pCont = pReq;
rpcMsg.contLen = sizeof(SAlterVnodeMsg);
rpcMsg.msgType = TSDB_MSG_TYPE_ALTER_VNODE_IN;
sendMsg(pClient, &rpcMsg);
SRpcMsg* pMsg = pClient->pRsp;
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, 0);
}
}
{
for (int i = 0; i < 3; ++i) {
SDropVnodeMsg* pReq = (SDropVnodeMsg*)rpcMallocCont(sizeof(SDropVnodeMsg));
pReq->vgId = htonl(2);
pReq->dnodeId = htonl(1);
strcpy(pReq->db, "1.d1");
pReq->dbUid = htobe64(9527);
SRpcMsg rpcMsg = {0};
rpcMsg.pCont = pReq;
rpcMsg.contLen = sizeof(SDropVnodeMsg);
rpcMsg.msgType = TSDB_MSG_TYPE_DROP_VNODE_IN;
sendMsg(pClient, &rpcMsg);
SRpcMsg* pMsg = pClient->pRsp;
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, 0);
}
}
}
source/dnode/mnode/impl/inc/mndDef.h
浏览文件 @ bddfff72
......@@ -61,15 +61,14 @@ typedef enum {
} EAuthOp;
typedef enum {
TRN_STAGE_PREPARE = 1,
TRN_STAGE_EXECUTE = 2,
TRN_STAGE_PREPARE = 0,
TRN_STAGE_EXECUTE = 1,
TRN_STAGE_ROLLBACK = 2,
TRN_STAGE_COMMIT = 3,
TRN_STAGE_ROLLBACK = 4,
TRN_STAGE_RETRY = 5,
TRN_STAGE_OVER = 6,
TRN_STAGE_OVER = 4,
} ETrnStage;
typedef enum { TRN_POLICY_ROLLBACK = 1, TRN_POLICY_RETRY = 2 } ETrnPolicy;
typedef enum { TRN_POLICY_ROLLBACK = 0, TRN_POLICY_RETRY = 1 } ETrnPolicy;
typedef enum {
DND_STATUS_OFFLINE = 0,
......
source/dnode/mnode/impl/inc/mndInt.h
浏览文件 @ bddfff72
......@@ -70,6 +70,7 @@ typedef struct SMnode {
tmr_h timer;
char *path;
SMnodeCfg cfg;
int64_t checkTime;
SSdb *pSdb;
SDnode *pDnode;
SArray *pSteps;
......
source/dnode/mnode/impl/inc/mndTrans.h
浏览文件 @ bddfff72
......@@ -22,6 +22,16 @@
extern "C" {
#endif
typedef struct {
SEpSet epSet;
int8_t msgType;
int8_t msgSent;
int8_t msgReceived;
int32_t errCode;
int32_t contLen;
void *pCont;
} STransAction;
int32_t mndInitTrans(SMnode *pMnode);
void mndCleanupTrans(SMnode *pMnode);
......@@ -30,10 +40,13 @@ void mndTransDrop(STrans *pTrans);
int32_t mndTransAppendRedolog(STrans *pTrans, SSdbRaw *pRaw);
int32_t mndTransAppendUndolog(STrans *pTrans, SSdbRaw *pRaw);
int32_t mndTransAppendCommitlog(STrans *pTrans, SSdbRaw *pRaw);
int32_t mndTransAppendRedoAction(STrans *pTrans, SEpSet *pEpSet, int8_t msgType, int32_t contLen, void *pCont);
int32_t mndTransAppendUndoAction(STrans *pTrans, SEpSet *pEpSet, int8_t msgType, int32_t contLen, void *pCont);
int32_t mndTransAppendRedoAction(STrans *pTrans, STransAction *pAction);
int32_t mndTransAppendUndoAction(STrans *pTrans, STransAction *pAction);
int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans);
void mndTransApply(SMnode *pMnode, SSdbRaw *pRaw, STransMsg *pMsg, int32_t code);
void mndTransHandleActionRsp(SMnodeMsg *pMsg);
char *mndTransStageStr(ETrnStage stage);
char *mndTransPolicyStr(ETrnPolicy policy);
......
source/dnode/mnode/impl/src/mndDb.c
浏览文件 @ bddfff72
此差异已折叠。
source/dnode/mnode/impl/src/mndDnode.c
浏览文件 @ bddfff72
......@@ -18,10 +18,10 @@
#include "mndMnode.h"
#include "mndShow.h"
#include "mndTrans.h"
#include "ttime.h"
#include "tep.h"
#include "ttime.h"
#define TSDB_DNODE_VER 1
#define TSDB_DNODE_VER_NUMBER 1
#define TSDB_DNODE_RESERVE_SIZE 64
#define TSDB_CONFIG_OPTION_LEN 16
#define TSDB_CONIIG_VALUE_LEN 48
......@@ -101,14 +101,14 @@ static int32_t mndCreateDefaultDnode(SMnode *pMnode) {
SSdbRaw *pRaw = mndDnodeActionEncode(&dnodeObj);
if (pRaw == NULL) return -1;
sdbSetRawStatus(pRaw, SDB_STATUS_READY);
if (sdbSetRawStatus(pRaw, SDB_STATUS_READY) != 0) return -1;
mDebug("dnode:%d, will be created while deploy sdb", dnodeObj.id);
return sdbWrite(pMnode->pSdb, pRaw);
}
static SSdbRaw *mndDnodeActionEncode(SDnodeObj *pDnode) {
SSdbRaw *pRaw = sdbAllocRaw(SDB_DNODE, TSDB_DNODE_VER, sizeof(SDnodeObj) + TSDB_DNODE_RESERVE_SIZE);
SSdbRaw *pRaw = sdbAllocRaw(SDB_DNODE, TSDB_DNODE_VER_NUMBER, sizeof(SDnodeObj) + TSDB_DNODE_RESERVE_SIZE);
if (pRaw == NULL) return NULL;
int32_t dataPos = 0;
......@@ -127,7 +127,7 @@ static SSdbRow *mndDnodeActionDecode(SSdbRaw *pRaw) {
int8_t sver = 0;
if (sdbGetRawSoftVer(pRaw, &sver) != 0) return NULL;
if (sver != TSDB_DNODE_VER) {
if (sver != TSDB_DNODE_VER_NUMBER) {
terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
mError("failed to decode dnode since %s", terrstr());
return NULL;
......@@ -150,21 +150,8 @@ static SSdbRow *mndDnodeActionDecode(SSdbRaw *pRaw) {
static int32_t mndDnodeActionInsert(SSdb *pSdb, SDnodeObj *pDnode) {
mTrace("dnode:%d, perform insert action", pDnode->id);
pDnode->rebootTime = 0;
pDnode->lastAccessTime = 0;
pDnode->accessTimes = 0;
pDnode->numOfMnodes = 0;
pDnode->numOfVnodes = 0;
pDnode->numOfQnodes = 0;
pDnode->numOfSupportMnodes = 0;
pDnode->numOfSupportVnodes = 0;
pDnode->numOfSupportQnodes = 0;
pDnode->numOfCores = 0;
pDnode->status = DND_STATUS_OFFLINE;
pDnode->offlineReason = DND_REASON_STATUS_NOT_RECEIVED;
snprintf(pDnode->ep, TSDB_EP_LEN, "%s:%u", pDnode->fqdn, pDnode->port);
return 0;
}
......@@ -225,7 +212,7 @@ int32_t mndGetDnodeSize(SMnode *pMnode) {
bool mndIsDnodeInReadyStatus(SMnode *pMnode, SDnodeObj *pDnode) {
int64_t ms = taosGetTimestampMs();
int64_t interval = ABS(pDnode->lastAccessTime - ms);
if (interval > 3000 * pMnode->cfg.statusInterval) {
if (interval > 3500 * pMnode->cfg.statusInterval) {
return false;
}
return true;
......@@ -267,12 +254,9 @@ static int32_t mndCheckClusterCfgPara(SMnode *pMnode, const SClusterCfg *pCfg) {
return DND_REASON_STATUS_INTERVAL_NOT_MATCH;
}
int64_t checkTime = 0;
char timestr[32] = "1970-01-01 00:00:00.00";
(void)taosParseTime(timestr, &checkTime, (int32_t)strlen(timestr), TSDB_TIME_PRECISION_MILLI, 0);
if ((0 != strcasecmp(pCfg->timezone, pMnode->cfg.timezone)) && (checkTime != pCfg->checkTime)) {
if ((0 != strcasecmp(pCfg->timezone, pMnode->cfg.timezone)) && (pMnode->checkTime != pCfg->checkTime)) {
mError("timezone [%s - %s] [%" PRId64 " - %" PRId64 "] cfg inconsistent", pCfg->timezone, pMnode->cfg.timezone,
pCfg->checkTime, checkTime);
pCfg->checkTime, pMnode->checkTime);
return DND_REASON_TIME_ZONE_NOT_MATCH;
}
......
source/dnode/mnode/impl/src/mndTrans.c
浏览文件 @ bddfff72
......@@ -21,13 +21,6 @@
#define TSDB_TRN_ARRAY_SIZE 8
#define TSDB_TRN_RESERVE_SIZE 64
typedef struct {
SEpSet epSet;
int8_t msgType;
int32_t contLen;
void *pCont;
} STransAction;
static SSdbRaw *mndTransActionEncode(STrans *pTrans);
static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw);
static int32_t mndTransActionInsert(SSdb *pSdb, STrans *pTrans);
......@@ -37,11 +30,11 @@ static int32_t mndTransActionDelete(SSdb *pSdb, STrans *pTrans);
static void mndTransSetRpcHandle(STrans *pTrans, void *rpcHandle);
static void mndTransSendRpcRsp(STrans *pTrans, int32_t code);
static int32_t mndTransAppendLog(SArray *pArray, SSdbRaw *pRaw);
static int32_t mndTransAppendAction(SArray *pArray, SEpSet *pEpSet, int8_t msgType, int32_t contLen, void *pCont);
static int32_t mndTransAppendAction(SArray *pArray, STransAction *pAction);
static void mndTransDropLogs(SArray *pArray);
static void mndTransDropActions(SArray *pArray);
static int32_t mndTransExecuteLogs(SMnode *pMnode, SArray *pArray);
static int32_t mndTransExecuteActions(SMnode *pMnode, SArray *pArray);
static int32_t mndTransExecuteActions(SMnode *pMnode, STrans *pTrans, SArray *pArray);
static int32_t mndTransExecuteRedoLogs(SMnode *pMnode, STrans *pTrans);
static int32_t mndTransExecuteUndoLogs(SMnode *pMnode, STrans *pTrans);
static int32_t mndTransExecuteCommitLogs(SMnode *pMnode, STrans *pTrans);
......@@ -343,10 +336,8 @@ char *mndTransStageStr(ETrnStage stage) {
return "commit";
case TRN_STAGE_ROLLBACK:
return "rollback";
case TRN_STAGE_RETRY:
return "retry";
case TRN_STAGE_OVER:
return "stop";
return "over";
default:
return "undefined";
}
......@@ -388,7 +379,7 @@ STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, void *rpcHandle) {
return NULL;
}
mDebug("trans:%d, data:%p is created", pTrans->id, pTrans);
mDebug("trans:%d, is created", pTrans->id);
return pTrans;
}
......@@ -417,7 +408,7 @@ void mndTransDrop(STrans *pTrans) {
mndTransDropActions(pTrans->redoActions);
mndTransDropActions(pTrans->undoActions);
mDebug("trans:%d, data:%p is dropped", pTrans->id, pTrans);
// mDebug("trans:%d, is dropped, data:%p", pTrans->id, pTrans);
tfree(pTrans);
}
......@@ -459,10 +450,8 @@ int32_t mndTransAppendCommitlog(STrans *pTrans, SSdbRaw *pRaw) {
return code;
}
static int32_t mndTransAppendAction(SArray *pArray, SEpSet *pEpSet, int8_t msgType, int32_t contLen, void *pCont) {
STransAction action = {.epSet = *pEpSet, .msgType = msgType, .contLen = contLen, .pCont = pCont};
void *ptr = taosArrayPush(pArray, &action);
static int32_t mndTransAppendAction(SArray *pArray, STransAction *pAction) {
void *ptr = taosArrayPush(pArray, pAction);
if (ptr == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
......@@ -471,16 +460,12 @@ static int32_t mndTransAppendAction(SArray *pArray, SEpSet *pEpSet, int8_t msgTy
return 0;
}
int32_t mndTransAppendRedoAction(STrans *pTrans, SEpSet *pEpSet, int8_t msgType, int32_t contLen, void *pCont) {
int32_t code = mndTransAppendAction(pTrans->redoActions, pEpSet, msgType, contLen, pCont);
mTrace("trans:%d, msg:%s len:%d append to redo actions", pTrans->id, taosMsg[msgType], contLen);
return code;
int32_t mndTransAppendRedoAction(STrans *pTrans, STransAction *pAction) {
return mndTransAppendAction(pTrans->redoActions, pAction);
}
int32_t mndTransAppendUndoAction(STrans *pTrans, SEpSet *pEpSet, int8_t msgType, int32_t contLen, void *pCont) {
int32_t code = mndTransAppendAction(pTrans->undoActions, pEpSet, msgType, contLen, pCont);
mTrace("trans:%d, msg:%s len:%d append to undo actions", pTrans->id, taosMsg[msgType], contLen);
return code;
int32_t mndTransAppendUndoAction(STrans *pTrans, STransAction *pAction) {
return mndTransAppendAction(pTrans->undoActions, pAction);
}
int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans) {
......@@ -493,7 +478,7 @@ int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans) {
}
sdbSetRawStatus(pRaw, SDB_STATUS_READY);
mTrace("trans:%d, start sync", pTrans->id);
mTrace("trans:%d, sync to other nodes", pTrans->id);
int32_t code = mndSyncPropose(pMnode, pRaw);
if (code != 0) {
mError("trans:%d, failed to sync since %s", pTrans->id, terrstr());
......@@ -533,7 +518,7 @@ int32_t mndTransCommit(SMnode *pMnode, STrans *pTrans) {
sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED);
if (taosArrayGetSize(pTrans->commitLogs) != 0) {
mTrace("trans:%d, start sync", pTrans->id);
mTrace("trans:%d, sync to other nodes", pTrans->id);
int32_t code = mndSyncPropose(pMnode, pRaw);
if (code != 0) {
mError("trans:%d, failed to sync since %s", pTrans->id, terrstr());
......@@ -563,7 +548,7 @@ int32_t mndTransRollback(SMnode *pMnode, STrans *pTrans) {
}
sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED);
mTrace("trans:%d, start sync", pTrans->id);
mTrace("trans:%d, sync to other nodes", pTrans->id);
int32_t code = mndSyncPropose(pMnode, pRaw);
if (code != 0) {
mError("trans:%d, failed to sync since %s", pTrans->id, terrstr());
......@@ -596,6 +581,50 @@ void mndTransApply(SMnode *pMnode, SSdbRaw *pRaw, STransMsg *pMsg, int32_t code)
// todo
}
void mndTransHandleActionRsp(SMnodeMsg *pMsg) {
SMnode *pMnode = pMsg->pMnode;
int64_t sig = (int64_t)(pMsg->rpcMsg.ahandle);
int32_t transId = (int32_t)(sig >> 32);
int32_t action = (int32_t)((sig << 32) >> 32);
STrans *pTrans = mndAcquireTrans(pMnode, transId);
if (pTrans == NULL) {
mError("trans:%d, failed to get transId from vnode rsp since %s", transId, terrstr());
goto HANDLE_ACTION_RSP_OVER;
}
SArray *pArray = NULL;
if (pTrans->stage == TRN_STAGE_EXECUTE) {
pArray = pTrans->redoActions;
} else if (pTrans->stage == TRN_STAGE_ROLLBACK) {
pArray = pTrans->undoActions;
} else {
}
if (pArray == NULL) {
mError("trans:%d, invalid trans stage:%s", transId, mndTransStageStr(pTrans->stage));
goto HANDLE_ACTION_RSP_OVER;
}
int32_t actionNum = taosArrayGetSize(pTrans->redoActions);
if (action < 0 || action > actionNum) {
mError("trans:%d, invalid action:%d", transId, action);
goto HANDLE_ACTION_RSP_OVER;
}
STransAction *pAction = taosArrayGet(pArray, action);
if (pAction != NULL) {
pAction->msgReceived = 1;
pAction->errCode = pMsg->code;
}
mDebug("trans:%d, action:%d response is received, code:0x%x", transId, action, pMsg->code);
mndTransExecute(pMnode, pTrans);
HANDLE_ACTION_RSP_OVER:
mndReleaseTrans(pMnode, pTrans);
}
static int32_t mndTransExecuteLogs(SMnode *pMnode, SArray *pArray) {
SSdb *pSdb = pMnode->pSdb;
int32_t arraySize = taosArrayGetSize(pArray);
......@@ -618,7 +647,7 @@ static int32_t mndTransExecuteRedoLogs(SMnode *pMnode, STrans *pTrans) {
if (code != 0) {
mError("trans:%d, failed to execute redo logs since %s", pTrans->id, terrstr())
} else {
mTrace("trans:%d, execute redo logs finished", pTrans->id)
mDebug("trans:%d, execute redo logs finished", pTrans->id)
}
}
......@@ -632,7 +661,7 @@ static int32_t mndTransExecuteUndoLogs(SMnode *pMnode, STrans *pTrans) {
if (code != 0) {
mError("trans:%d, failed to execute undo logs since %s", pTrans->id, terrstr())
} else {
mTrace("trans:%d, execute undo logs finished", pTrans->id)
mDebug("trans:%d, execute undo logs finished", pTrans->id)
}
}
......@@ -646,47 +675,70 @@ static int32_t mndTransExecuteCommitLogs(SMnode *pMnode, STrans *pTrans) {
if (code != 0) {
mError("trans:%d, failed to execute commit logs since %s", pTrans->id, terrstr())
} else {
mTrace("trans:%d, execute commit logs finished", pTrans->id)
mDebug("trans:%d, execute commit logs finished", pTrans->id)
}
}
return code;
}
static int32_t mndTransExecuteActions(SMnode *pMnode, SArray *pArray) {
#if 0
int32_t arraySize = taosArrayGetSize(pArray);
for (int32_t i = 0; i < arraySize; ++i) {
STransAction *pAction = taosArrayGet(pArray, i);
static int32_t mndTransExecuteActions(SMnode *pMnode, STrans *pTrans, SArray *pArray) {
int32_t numOfActions = taosArrayGetSize(pArray);
if (numOfActions == 0) return 0;
for (int32_t action = 0; action < numOfActions; ++action) {
STransAction *pAction = taosArrayGet(pArray, action);
if (pAction == NULL) continue;
if (pAction->msgSent) continue;
SRpcMsg rpcMsg = {.msgType = pAction->msgType, .contLen = pAction->contLen};
int64_t signature = pTrans->id;
signature = (signature << 32);
signature += action;
SRpcMsg rpcMsg = {.msgType = pAction->msgType, .contLen = pAction->contLen, .ahandle = (void *)signature};
rpcMsg.pCont = rpcMallocCont(pAction->contLen);
if (rpcMsg.pCont == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
memcpy(rpcMsg.pCont, pAction->pCont, pAction->contLen);
pAction->msgSent = 1;
pAction->msgReceived = 0;
pAction->errCode = 0;
mDebug("trans:%d, action:%d is sent", pTrans->id, action);
mndSendMsgToDnode(pMnode, &pAction->epSet, &rpcMsg);
}
int32_t numOfReceivedMsgs = 0;
int32_t errorCode = 0;
for (int32_t action = 0; action < numOfActions; ++action) {
STransAction *pAction = taosArrayGet(pArray, action);
if (pAction == NULL) continue;
if (pAction->msgSent && pAction->msgReceived) {
numOfReceivedMsgs++;
if (pAction->errCode != 0) {
errorCode = pAction->errCode;
}
}
}
if (numOfReceivedMsgs == numOfActions) {
mDebug("trans:%d, all %d actions executed, code:0x%x", pTrans->id, numOfActions, errorCode);
terrno = errorCode;
return errorCode;
} else {
return TSDB_CODE_MND_ACTION_IN_PROGRESS;
#else
return 0;
#endif
}
}
static int32_t mndTransExecuteRedoActions(SMnode *pMnode, STrans *pTrans) {
if (taosArrayGetSize(pTrans->redoActions) <= 0) return 0;
mTrace("trans:%d, start to execute redo actions", pTrans->id);
return mndTransExecuteActions(pMnode, pTrans->redoActions);
return mndTransExecuteActions(pMnode, pTrans, pTrans->redoActions);
}
static int32_t mndTransExecuteUndoActions(SMnode *pMnode, STrans *pTrans) {
if (taosArrayGetSize(pTrans->undoActions) <= 0) return 0;
mTrace("trans:%d, start to execute undo actions", pTrans->id);
return mndTransExecuteActions(pMnode, pTrans->undoActions);
return mndTransExecuteActions(pMnode, pTrans, pTrans->undoActions);
}
static int32_t mndTransPerformPrepareStage(SMnode *pMnode, STrans *pTrans) {
......@@ -694,7 +746,7 @@ static int32_t mndTransPerformPrepareStage(SMnode *pMnode, STrans *pTrans) {
if (code == 0) {
pTrans->stage = TRN_STAGE_EXECUTE;
mTrace("trans:%d, stage from prepare to execute", pTrans->id);
mDebug("trans:%d, stage from prepare to execute", pTrans->id);
} else {
pTrans->stage = TRN_STAGE_ROLLBACK;
mError("trans:%d, stage from prepare to rollback since %s", pTrans->id, terrstr());
......@@ -708,17 +760,17 @@ static int32_t mndTransPerformExecuteStage(SMnode *pMnode, STrans *pTrans) {
if (code == 0) {
pTrans->stage = TRN_STAGE_COMMIT;
mTrace("trans:%d, stage from execute to commit", pTrans->id);
mDebug("trans:%d, stage from execute to commit", pTrans->id);
} else if (code == TSDB_CODE_MND_ACTION_IN_PROGRESS) {
mTrace("trans:%d, stage keep on execute since %s", pTrans->id, terrstr(code));
mDebug("trans:%d, stage keep on execute since %s", pTrans->id, tstrerror(code));
return code;
} else {
if (pTrans->policy == TRN_POLICY_ROLLBACK) {
pTrans->stage = TRN_STAGE_ROLLBACK;
mError("trans:%d, stage from execute to rollback since %s", pTrans->id, terrstr());
} else {
pTrans->stage = TRN_STAGE_RETRY;
mError("trans:%d, stage from execute to retry since %s", pTrans->id, terrstr());
pTrans->stage = TRN_STAGE_EXECUTE;
mError("trans:%d, stage keep on execute since %s", pTrans->id, terrstr());
}
}
......@@ -726,29 +778,16 @@ static int32_t mndTransPerformExecuteStage(SMnode *pMnode, STrans *pTrans) {
}
static int32_t mndTransPerformCommitStage(SMnode *pMnode, STrans *pTrans) {
int32_t code = mndTransExecuteCommitLogs(pMnode, pTrans);
if (code == 0) {
mndTransExecuteCommitLogs(pMnode, pTrans);
pTrans->stage = TRN_STAGE_OVER;
mTrace("trans:%d, commit stage finished", pTrans->id);
} else {
if (pTrans->policy == TRN_POLICY_ROLLBACK) {
pTrans->stage = TRN_STAGE_ROLLBACK;
mError("trans:%d, stage from commit to rollback since %s", pTrans->id, terrstr());
} else {
pTrans->stage = TRN_STAGE_RETRY;
mError("trans:%d, stage from commit to retry since %s", pTrans->id, terrstr());
}
}
return code;
return 0;
}
static int32_t mndTransPerformRollbackStage(SMnode *pMnode, STrans *pTrans) {
int32_t code = mndTransExecuteUndoActions(pMnode, pTrans);
if (code == 0) {
mTrace("trans:%d, rollbacked", pTrans->id);
mDebug("trans:%d, rollbacked", pTrans->id);
} else {
pTrans->stage = TRN_STAGE_ROLLBACK;
mError("trans:%d, stage keep on rollback since %s", pTrans->id, terrstr());
......@@ -757,20 +796,6 @@ static int32_t mndTransPerformRollbackStage(SMnode *pMnode, STrans *pTrans) {
return code;
}
static int32_t mndTransPerformRetryStage(SMnode *pMnode, STrans *pTrans) {
int32_t code = mndTransExecuteRedoActions(pMnode, pTrans);
if (code == 0) {
pTrans->stage = TRN_STAGE_COMMIT;
mTrace("trans:%d, stage from retry to commit", pTrans->id);
} else {
pTrans->stage = TRN_STAGE_RETRY;
mError("trans:%d, stage keep on retry since %s", pTrans->id, terrstr());
}
return code;
}
static void mndTransExecute(SMnode *pMnode, STrans *pTrans) {
int32_t code = 0;
......@@ -785,7 +810,7 @@ static void mndTransExecute(SMnode *pMnode, STrans *pTrans) {
case TRN_STAGE_COMMIT:
code = mndTransCommit(pMnode, pTrans);
if (code == 0) {
code = mndTransPerformCommitStage(pMnode, pTrans);
mndTransPerformCommitStage(pMnode, pTrans);
}
break;
case TRN_STAGE_ROLLBACK:
......@@ -794,9 +819,6 @@ static void mndTransExecute(SMnode *pMnode, STrans *pTrans) {
code = mndTransRollback(pMnode, pTrans);
}
break;
case TRN_STAGE_RETRY:
code = mndTransPerformRetryStage(pMnode, pTrans);
break;
default:
mndTransSendRpcRsp(pTrans, 0);
return;
......
source/dnode/mnode/impl/src/mndVgroup.c
浏览文件 @ bddfff72
......@@ -21,7 +21,7 @@
#include "mndShow.h"
#include "mndTrans.h"
#define TSDB_VGROUP_VER_NUM 1
#define TSDB_VGROUP_VER_NUMBER 1
#define TSDB_VGROUP_RESERVE_SIZE 64
static SSdbRow *mndVgroupActionDecode(SSdbRaw *pRaw);
......@@ -70,7 +70,7 @@ int32_t mndInitVgroup(SMnode *pMnode) {
void mndCleanupVgroup(SMnode *pMnode) {}
SSdbRaw *mndVgroupActionEncode(SVgObj *pVgroup) {
SSdbRaw *pRaw = sdbAllocRaw(SDB_VGROUP, TSDB_VGROUP_VER_NUM, sizeof(SVgObj) + TSDB_VGROUP_RESERVE_SIZE);
SSdbRaw *pRaw = sdbAllocRaw(SDB_VGROUP, TSDB_VGROUP_VER_NUMBER, sizeof(SVgObj) + TSDB_VGROUP_RESERVE_SIZE);
if (pRaw == NULL) return NULL;
int32_t dataPos = 0;
......@@ -98,7 +98,7 @@ SSdbRow *mndVgroupActionDecode(SSdbRaw *pRaw) {
int8_t sver = 0;
if (sdbGetRawSoftVer(pRaw, &sver) != 0) return NULL;
if (sver != TSDB_VGROUP_VER_NUM) {
if (sver != TSDB_VGROUP_VER_NUMBER) {
mError("failed to decode vgroup since %s", terrstr());
terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
return NULL;
......@@ -142,6 +142,8 @@ static int32_t mndVgroupActionUpdate(SSdb *pSdb, SVgObj *pOldVgroup, SVgObj *pNe
mTrace("vgId:%d, perform update action", pOldVgroup->vgId);
pOldVgroup->updateTime = pNewVgroup->updateTime;
pOldVgroup->version = pNewVgroup->version;
pOldVgroup->hashBegin = pNewVgroup->hashBegin;
pOldVgroup->hashEnd = pNewVgroup->hashEnd;
pOldVgroup->replica = pNewVgroup->replica;
memcpy(pOldVgroup->vnodeGid, pNewVgroup->vnodeGid, TSDB_MAX_REPLICA * sizeof(SVnodeGid));
return 0;
......@@ -149,7 +151,11 @@ static int32_t mndVgroupActionUpdate(SSdb *pSdb, SVgObj *pOldVgroup, SVgObj *pNe
SVgObj *mndAcquireVgroup(SMnode *pMnode, int32_t vgId) {
SSdb *pSdb = pMnode->pSdb;
return sdbAcquire(pSdb, SDB_VGROUP, &vgId);
SVgObj *pVgroup = sdbAcquire(pSdb, SDB_VGROUP, &vgId);
if (pVgroup == NULL) {
terrno = TSDB_CODE_MND_VGROUP_NOT_EXIST;
}
return pVgroup;
}
void mndReleaseVgroup(SMnode *pMnode, SVgObj *pVgroup) {
......@@ -158,16 +164,17 @@ void mndReleaseVgroup(SMnode *pMnode, SVgObj *pVgroup) {
}
SCreateVnodeMsg *mndBuildCreateVnodeMsg(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup) {
SCreateVnodeMsg *pCreate = malloc(sizeof(SCreateVnodeMsg));
SCreateVnodeMsg *pCreate = calloc(1, sizeof(SCreateVnodeMsg));
if (pCreate == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
pCreate->dnodeId = htonl(pDnode->id);
pCreate->vgId = htonl(pVgroup->vgId);
pCreate->dnodeId = htonl(pDnode->id);
memcpy(pCreate->db, pDb->name, TSDB_FULL_DB_NAME_LEN);
pCreate->dbUid = htobe64(pDb->uid);
pCreate->vgVersion = htonl(pVgroup->version);
pCreate->cacheBlockSize = htonl(pDb->cfg.cacheBlockSize);
pCreate->totalBlocks = htonl(pDb->cfg.totalBlocks);
pCreate->daysPerFile = htonl(pDb->cfg.daysPerFile);
......@@ -193,7 +200,6 @@ SCreateVnodeMsg *mndBuildCreateVnodeMsg(SMnode *pMnode, SDnodeObj *pDnode, SDbOb
SDnodeObj *pVgidDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
if (pVgidDnode == NULL) {
free(pCreate);
terrno = TSDB_CODE_MND_APP_ERROR;
return NULL;
}
......@@ -217,7 +223,7 @@ SCreateVnodeMsg *mndBuildCreateVnodeMsg(SMnode *pMnode, SDnodeObj *pDnode, SDbOb
}
SDropVnodeMsg *mndBuildDropVnodeMsg(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup) {
SDropVnodeMsg *pDrop = malloc(sizeof(SDropVnodeMsg));
SDropVnodeMsg *pDrop = calloc(1, sizeof(SDropVnodeMsg));
if (pDrop == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
......@@ -269,7 +275,7 @@ int32_t mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups) {
return -1;
}
int32_t alloceVgroups = 0;
int32_t allocedVgroups = 0;
int32_t maxVgId = sdbGetMaxId(pMnode->pSdb, SDB_VGROUP);
uint32_t hashMin = 0;
uint32_t hashMax = UINT32_MAX;
......@@ -281,7 +287,6 @@ int32_t mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups) {
pVgroup->createdTime = taosGetTimestampMs();
pVgroup->updateTime = pVgroups->createdTime;
pVgroup->version = 1;
pVgroup->dbUid = pDb->uid;
pVgroup->hashBegin = hashMin + hashInterval * v;
if (v == pDb->cfg.numOfVgroups - 1) {
pVgroup->hashEnd = hashMax;
......@@ -290,6 +295,7 @@ int32_t mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups) {
}
memcpy(pVgroup->dbName, pDb->name, TSDB_FULL_DB_NAME_LEN);
pVgroup->dbUid = pDb->uid;
pVgroup->replica = pDb->cfg.replications;
if (mndGetAvailableDnode(pMnode, pVgroup) != 0) {
......@@ -298,14 +304,18 @@ int32_t mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups) {
return -1;
}
alloceVgroups++;
allocedVgroups++;
}
*ppVgroups = pVgroups;
return 0;
}
static int32_t mndProcessCreateVnodeRsp(SMnodeMsg *pMsg) { return 0; }
static int32_t mndProcessCreateVnodeRsp(SMnodeMsg *pMsg) {
mndTransHandleActionRsp(pMsg);
return 0;
}
static int32_t mndProcessAlterVnodeRsp(SMnodeMsg *pMsg) { return 0; }
static int32_t mndProcessDropVnodeRsp(SMnodeMsg *pMsg) { return 0; }
static int32_t mndProcessSyncVnodeRsp(SMnodeMsg *pMsg) { return 0; }
......@@ -313,7 +323,6 @@ static int32_t mndProcessCompactVnodeRsp(SMnodeMsg *pMsg) { return 0; }
static int32_t mndGetVgroupMaxReplica(SMnode *pMnode, char *dbName, int8_t *pReplica, int32_t *pNumOfVgroups) {
SSdb *pSdb = pMnode->pSdb;
SDbObj *pDb = mndAcquireDb(pMnode, dbName);
if (pDb == NULL) {
terrno = TSDB_CODE_MND_DB_NOT_SELECTED;
......@@ -329,7 +338,7 @@ static int32_t mndGetVgroupMaxReplica(SMnode *pMnode, char *dbName, int8_t *pRep
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
if (pIter == NULL) break;
if (strcmp(pVgroup->dbName, dbName) == 0) {
if (pVgroup->dbUid == pDb->uid) {
replica = MAX(replica, pVgroup->replica);
numOfVgroups++;
}
......@@ -441,11 +450,25 @@ static void mndCancelGetNextVgroup(SMnode *pMnode, void *pIter) {
}
static int32_t mndGetVnodesNum(SMnode *pMnode, int32_t dnodeId) {
if (dnodeId == 0) {
return 0;
SSdb *pSdb = pMnode->pSdb;
int32_t numOfVnodes = 0;
void *pIter = NULL;
while (1) {
SVgObj *pVgroup = NULL;
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
if (pIter == NULL) break;
for (int32_t v = 0; v < pVgroup->replica; ++v) {
if (pVgroup->vnodeGid[v].dnodeId == dnodeId) {
numOfVnodes++;
}
}
return 0;
sdbRelease(pSdb, pVgroup);
}
return numOfVnodes;
}
static int32_t mndGetVnodeMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaMsg *pMeta) {
......
source/dnode/mnode/impl/src/mnode.c
浏览文件 @ bddfff72
......@@ -237,6 +237,9 @@ SMnode *mndOpen(const char *path, const SMnodeOpt *pOption) {
return NULL;
}
char timestr[24] = "1970-01-01 00:00:00.00";
(void)taosParseTime(timestr, &pMnode->checkTime, (int32_t)strlen(timestr), TSDB_TIME_PRECISION_MILLI, 0);
pMnode->pSteps = taosArrayInit(24, sizeof(SMnodeStep));
if (pMnode->pSteps == NULL) {
free(pMnode);
......
source/dnode/mnode/sdb/src/sdbRaw.c
浏览文件 @ bddfff72
......@@ -27,12 +27,12 @@ SSdbRaw *sdbAllocRaw(ESdbType type, int8_t sver, int32_t dataLen) {
pRaw->sver = sver;
pRaw->dataLen = dataLen;
mTrace("raw:%p, is created, len:%d", pRaw, dataLen);
// mTrace("raw:%p, is created, len:%d", pRaw, dataLen);
return pRaw;
}
void sdbFreeRaw(SSdbRaw *pRaw) {
mTrace("raw:%p, is freed", pRaw);
// mTrace("raw:%p, is freed", pRaw);
free(pRaw);
}
......
source/libs/catalog/test/catalogTests.cpp
浏览文件 @ bddfff72
......@@ -61,8 +61,8 @@ void sendCreateDbMsg(void *shandle, SEpSet *pEpSet) {
pReq->daysToKeep0 = htonl(3650);
pReq->daysToKeep1 = htonl(3650);
pReq->daysToKeep2 = htonl(3650);
pReq->minRowsPerFileBlock = htonl(100);
pReq->maxRowsPerFileBlock = htonl(4096);
pReq->minRows = htonl(100);
pReq->maxRows = htonl(4096);
pReq->commitTime = htonl(3600);
pReq->fsyncPeriod = htonl(3000);
pReq->walLevel = 1;
......
source/libs/parser/src/astToMsg.c
浏览文件 @ bddfff72
......@@ -134,8 +134,8 @@ static void doSetDbOptions(SCreateDbMsg* pMsg, const SCreateDbInfo* pCreateDb) {
pMsg->totalBlocks = htonl(pCreateDb->numOfBlocks);
pMsg->daysPerFile = htonl(pCreateDb->daysPerFile);
pMsg->commitTime = htonl((int32_t)pCreateDb->commitTime);
pMsg->minRowsPerFileBlock = htonl(pCreateDb->minRowsPerBlock);
pMsg->maxRowsPerFileBlock = htonl(pCreateDb->maxRowsPerBlock);
pMsg->minRows = htonl(pCreateDb->minRowsPerBlock);
pMsg->maxRows = htonl(pCreateDb->maxRowsPerBlock);
pMsg->fsyncPeriod = htonl(pCreateDb->fsyncPeriod);
pMsg->compression = pCreateDb->compressionLevel;
pMsg->walLevel = (char)pCreateDb->walLevel;
......
source/util/src/tthread.c
浏览文件 @ bddfff72
......@@ -13,15 +13,15 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "os.h"
#include "tthread.h"
#include "os.h"
#include "taoserror.h"
#include "tdef.h"
#include "tutil.h"
#include "ulog.h"
#include "taoserror.h"
// create new thread
pthread_t* taosCreateThread( void *(*__start_routine) (void *), void* param) {
pthread_t* taosCreateThread(void* (*__start_routine)(void*), void* param) {
pthread_t* pthread = (pthread_t*)malloc(sizeof(pthread_t));
pthread_attr_t thattr;
pthread_attr_init(&thattr);
......@@ -38,8 +38,8 @@ pthread_t* taosCreateThread( void *(*__start_routine) (void *), void* param) {
// destory thread
bool taosDestoryThread(pthread_t* pthread) {
if(pthread == NULL) return false;
if(taosThreadRunning(pthread)) {
if (pthread == NULL) return false;
if (taosThreadRunning(pthread)) {
pthread_cancel(*pthread);
pthread_join(*pthread, NULL);
}
......@@ -50,12 +50,10 @@ bool taosDestoryThread(pthread_t* pthread) {
// thread running return true
bool taosThreadRunning(pthread_t* pthread) {
if(pthread == NULL) return false;
if (pthread == NULL) return false;
int ret = pthread_kill(*pthread, 0);
if(ret == ESRCH)
return false;
if(ret == EINVAL)
return false;
if (ret == ESRCH) return false;
if (ret == EINVAL) return false;
// alive
return true;
}
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或