xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/3.0' into feature/scheduler

This commit is contained in:
dapan1121 2021-12-20 08:18:12 +08:00
parent 5579b50ac6 67ed87e2fa
commit fa66a89bcc
29 changed files with 875 additions and 217 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
cmake/cmake.options
View File

@ -37,6 +37,7 @@ option(
off off
) )
option( option(
BUILD_WITH_NURAFT BUILD_WITH_NURAFT
"If build with NuRaft" "If build with NuRaft"
@ -54,3 +55,9 @@ option(
"If use doxygen build documents" "If use doxygen build documents"
OFF OFF
) )
option(
BUILD_WITH_INVERTEDINDEX
"If use invertedIndex"
ON
)

19
include/common/taosmsg.h
View File

@ -752,24 +752,26 @@ typedef struct {
} SReplica; } SReplica;
typedef struct { typedef struct {
char db[TSDB_FULL_DB_NAME_LEN];
int32_t vgId; int32_t vgId;
int32_t dnodeId;
char db[TSDB_FULL_DB_NAME_LEN];
uint64_t dbUid;
int32_t cacheBlockSize; int32_t cacheBlockSize;
int32_t totalBlocks; int32_t totalBlocks;
int32_t daysPerFile; int32_t daysPerFile;
int32_t daysToKeep0; int32_t daysToKeep0;
int32_t daysToKeep1; int32_t daysToKeep1;
int32_t daysToKeep2; int32_t daysToKeep2;
int32_t minRowsPerFileBlock; int32_t minRows;
int32_t maxRowsPerFileBlock; int32_t maxRows;
int32_t commitTime;
int32_t fsyncPeriod; int32_t fsyncPeriod;
int8_t reserved[16]; int8_t walLevel;
int8_t precision; int8_t precision;
int8_t compression; int8_t compression;
int8_t cacheLastRow;
int8_t update;
int8_t walLevel;
int8_t quorum; int8_t quorum;
int8_t update;
int8_t cacheLastRow;
int8_t replica; int8_t replica;
int8_t selfIndex; int8_t selfIndex;
SReplica replicas[TSDB_MAX_REPLICA]; SReplica replicas[TSDB_MAX_REPLICA];
@ -777,6 +779,9 @@ typedef struct {
typedef struct { typedef struct {
int32_t vgId; int32_t vgId;
int32_t dnodeId;
char db[TSDB_FULL_DB_NAME_LEN];
uint64_t dbUid;
} SDropVnodeMsg, SSyncVnodeMsg, SCompactVnodeMsg; } SDropVnodeMsg, SSyncVnodeMsg, SCompactVnodeMsg;
typedef struct { typedef struct {

18
include/dnode/vnode/tsdb/tsdb.h
View File

@ -24,26 +24,26 @@ extern "C" {
// TYPES EXPOSED // TYPES EXPOSED
typedef struct STsdb STsdb; typedef struct STsdb STsdb;
typedef struct STsdbCfg STsdbCfg;
typedef struct STsdbCfg {
uint64_t lruCacheSize;
uint32_t keep0;
uint32_t keep1;
uint32_t keep2;
} STsdbCfg;
// STsdb // STsdb
STsdb *tsdbOpen(const char *path, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF); STsdb *tsdbOpen(const char *path, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF);
void tsdbClose(STsdb *); void tsdbClose(STsdb *);
void tsdbRemove(const char *path); void tsdbRemove(const char *path);
int tsdbInsertData(STsdb *pTsdb, SSubmitMsg *pMsg); int tsdbInsertData(STsdb *pTsdb, SSubmitMsg *pMsg);
int tsdbPrepareCommit(STsdb *pTsdb);
int tsdbCommit(STsdb *pTsdb);
// STsdbCfg // STsdbCfg
int tsdbOptionsInit(STsdbCfg *); int tsdbOptionsInit(STsdbCfg *);
void tsdbOptionsClear(STsdbCfg *); void tsdbOptionsClear(STsdbCfg *);
/* ------------------------ STRUCT DEFINITIONS ------------------------ */
struct STsdbCfg {
uint64_t lruCacheSize;
uint32_t keep0;
uint32_t keep1;
uint32_t keep2;
};
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

20
include/libs/index/index.h
View File

@ -24,6 +24,7 @@ extern "C" {
#endif #endif
typedef struct SIndex SIndex; typedef struct SIndex SIndex;
typedef struct SIndexTerm SIndexTerm;
typedef struct SIndexOpts SIndexOpts; typedef struct SIndexOpts SIndexOpts;
typedef struct SIndexMultiTermQuery SIndexMultiTermQuery; typedef struct SIndexMultiTermQuery SIndexMultiTermQuery;
typedef struct SArray SIndexMultiTerm; typedef struct SArray SIndexMultiTerm;
@ -35,7 +36,7 @@ typedef enum {
ADD_INDEX, // add index on specify column ADD_INDEX, // add index on specify column
DROP_INDEX, // drop existed index DROP_INDEX, // drop existed index
DROP_SATBLE // drop stable DROP_SATBLE // drop stable
} SIndexColumnType; } SIndexOperOnColumn;
typedef enum { MUST = 0, SHOULD = 1, NOT = 2 } EIndexOperatorType; typedef enum { MUST = 0, SHOULD = 1, NOT = 2 } EIndexOperatorType;
typedef enum { QUERY_TERM = 0, QUERY_PREFIX = 1, QUERY_SUFFIX = 2,QUERY_REGEX = 3} EIndexQueryType; typedef enum { QUERY_TERM = 0, QUERY_PREFIX = 1, QUERY_SUFFIX = 2,QUERY_REGEX = 3} EIndexQueryType;
@ -45,12 +46,12 @@ typedef enum { QUERY_TERM = 0, QUERY_PREFIX = 1, QUERY_SUFFIX = 2,QUERY_REGEX =
*/ */
SIndexMultiTermQuery *indexMultiTermQueryCreate(EIndexOperatorType oper); SIndexMultiTermQuery *indexMultiTermQueryCreate(EIndexOperatorType oper);
void indexMultiTermQueryDestroy(SIndexMultiTermQuery *pQuery); void indexMultiTermQueryDestroy(SIndexMultiTermQuery *pQuery);
int indexMultiTermQueryAdd(SIndexMultiTermQuery *pQuery, const char *field, int32_t nFields, const char *value, int32_t nValue, EIndexQueryType type); int indexMultiTermQueryAdd(SIndexMultiTermQuery *pQuery, SIndexTerm *term, EIndexQueryType type);
/* /*
* @param: * @param:
* @param: * @param:
*/ */
SIndex* indexOpen(SIndexOpts *opt, const char *path); int indexOpen(SIndexOpts *opt, const char *path, SIndex **index);
void indexClose(SIndex *index); void indexClose(SIndex *index);
int indexPut(SIndex *index, SIndexMultiTerm *terms, int uid); int indexPut(SIndex *index, SIndexMultiTerm *terms, int uid);
int indexDelete(SIndex *index, SIndexMultiTermQuery *query); int indexDelete(SIndex *index, SIndexMultiTermQuery *query);
@ -61,7 +62,7 @@ int indexRebuild(SIndex *index, SIndexOpts *opt);
* @param * @param
*/ */
SIndexMultiTerm *indexMultiTermCreate(); SIndexMultiTerm *indexMultiTermCreate();
int indexMultiTermAdd(SIndexMultiTerm *terms, const char *field, int32_t nFields, const char *value, int32_t nValue); int indexMultiTermAdd(SIndexMultiTerm *terms, SIndexTerm *term);
void indexMultiTermDestroy(SIndexMultiTerm *terms); void indexMultiTermDestroy(SIndexMultiTerm *terms);
/* /*
* @param: * @param:
@ -70,6 +71,17 @@ void indexMultiTermDestroy(SIndexMultiTerm *terms);
SIndexOpts *indexOptsCreate(); SIndexOpts *indexOptsCreate();
void indexOptsDestroy(SIndexOpts *opts); void indexOptsDestroy(SIndexOpts *opts);
/*
* @param:
* @param:
*/
SIndexTerm *indexTermCreate(int64_t suid, SIndexOperOnColumn operType, uint8_t colType,
const char *colName, int32_t nColName, const char *colVal, int32_t nColVal);
void indexTermDestroy(SIndexTerm *p);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

3
include/util/mallocator.h
View File

@ -39,6 +39,9 @@ typedef struct SMemAllocator {
TD_MEM_ALCT(SMemAllocator); TD_MEM_ALCT(SMemAllocator);
} SMemAllocator; } SMemAllocator;
#define tMalloc(pMA, SIZE) TD_MA_MALLOC(PMA, SIZE)
#define tFree(pMA, PTR) TD_MA_FREE(PMA, PTR)
typedef struct SMemAllocatorFactory { typedef struct SMemAllocatorFactory {
void *impl; void *impl;
SMemAllocator *(*create)(struct SMemAllocatorFactory *); SMemAllocator *(*create)(struct SMemAllocatorFactory *);

23
include/util/taoserror.h
View File

@ -120,17 +120,18 @@ int32_t* taosGetErrno();
#define TSDB_CODE_TSC_INVALID_INPUT TAOS_DEF_ERROR_CODE(0, 0X0224) //"Invalid tsc input") #define TSDB_CODE_TSC_INVALID_INPUT TAOS_DEF_ERROR_CODE(0, 0X0224) //"Invalid tsc input")
// mnode-common // mnode-common
#define TSDB_CODE_MND_NOT_READY TAOS_DEF_ERROR_CODE(0, 0x0300) #define TSDB_CODE_MND_APP_ERROR TAOS_DEF_ERROR_CODE(0, 0x0300)
#define TSDB_CODE_MND_MSG_NOT_PROCESSED TAOS_DEF_ERROR_CODE(0, 0x0301) #define TSDB_CODE_MND_NOT_READY TAOS_DEF_ERROR_CODE(0, 0x0301)
#define TSDB_CODE_MND_ACTION_IN_PROGRESS TAOS_DEF_ERROR_CODE(0, 0x0302) #define TSDB_CODE_MND_MSG_NOT_PROCESSED TAOS_DEF_ERROR_CODE(0, 0x0302)
#define TSDB_CODE_MND_ACTION_NEED_REPROCESSED TAOS_DEF_ERROR_CODE(0, 0x0303) #define TSDB_CODE_MND_ACTION_IN_PROGRESS TAOS_DEF_ERROR_CODE(0, 0x0303)
#define TSDB_CODE_MND_NO_RIGHTS TAOS_DEF_ERROR_CODE(0, 0x0304) #define TSDB_CODE_MND_ACTION_NEED_REPROCESSED TAOS_DEF_ERROR_CODE(0, 0x0304)
#define TSDB_CODE_MND_INVALID_OPTIONS TAOS_DEF_ERROR_CODE(0, 0x0305) #define TSDB_CODE_MND_NO_RIGHTS TAOS_DEF_ERROR_CODE(0, 0x0305)
#define TSDB_CODE_MND_INVALID_CONNECTION TAOS_DEF_ERROR_CODE(0, 0x0306) #define TSDB_CODE_MND_INVALID_OPTIONS TAOS_DEF_ERROR_CODE(0, 0x0306)
#define TSDB_CODE_MND_INVALID_MSG_VERSION TAOS_DEF_ERROR_CODE(0, 0x0307) #define TSDB_CODE_MND_INVALID_CONNECTION TAOS_DEF_ERROR_CODE(0, 0x0307)
#define TSDB_CODE_MND_INVALID_MSG_LEN TAOS_DEF_ERROR_CODE(0, 0x0308) #define TSDB_CODE_MND_INVALID_MSG_VERSION TAOS_DEF_ERROR_CODE(0, 0x0308)
#define TSDB_CODE_MND_INVALID_MSG_TYPE TAOS_DEF_ERROR_CODE(0, 0x0309) #define TSDB_CODE_MND_INVALID_MSG_LEN TAOS_DEF_ERROR_CODE(0, 0x0309)
#define TSDB_CODE_MND_TOO_MANY_SHELL_CONNS TAOS_DEF_ERROR_CODE(0, 0x030A) #define TSDB_CODE_MND_INVALID_MSG_TYPE TAOS_DEF_ERROR_CODE(0, 0x030A)
#define TSDB_CODE_MND_TOO_MANY_SHELL_CONNS TAOS_DEF_ERROR_CODE(0, 0x030B)
// mnode-show // mnode-show
#define TSDB_CODE_MND_INVALID_SHOWOBJ TAOS_DEF_ERROR_CODE(0, 0x0310) #define TSDB_CODE_MND_INVALID_SHOWOBJ TAOS_DEF_ERROR_CODE(0, 0x0310)

2
include/util/tdef.h
View File

@ -231,7 +231,7 @@ do { \
#define TSDB_DEFAULT_PAYLOAD_SIZE 5120 // default payload size, greater than PATH_MAX value #define TSDB_DEFAULT_PAYLOAD_SIZE 5120 // default payload size, greater than PATH_MAX value
#define TSDB_EXTRA_PAYLOAD_SIZE 128 // extra bytes for auth #define TSDB_EXTRA_PAYLOAD_SIZE 128 // extra bytes for auth
#define TSDB_CQ_SQL_SIZE 1024 #define TSDB_CQ_SQL_SIZE 1024
#define TSDB_MIN_VNODES 64 #define TSDB_MIN_VNODES 16
#define TSDB_MAX_VNODES 512 #define TSDB_MAX_VNODES 512
#define TSDB_MIN_VNODES_PER_DB 1 #define TSDB_MIN_VNODES_PER_DB 1
#define TSDB_MAX_VNODES_PER_DB 4096 #define TSDB_MAX_VNODES_PER_DB 4096

84
source/dnode/mgmt/impl/src/dndVnodes.c
View File

@ -95,7 +95,7 @@ static int32_t dndProcessCompactVnodeReq(SDnode *pDnode, SRpcMsg *rpcMsg);
static SVnodeObj *dndAcquireVnode(SDnode *pDnode, int32_t vgId) { static SVnodeObj *dndAcquireVnode(SDnode *pDnode, int32_t vgId) {
SVnodesMgmt *pMgmt = &pDnode->vmgmt; SVnodesMgmt *pMgmt = &pDnode->vmgmt;
SVnodeObj * pVnode = NULL; SVnodeObj *pVnode = NULL;
int32_t refCount = 0; int32_t refCount = 0;
taosRLockLatch(&pMgmt->latch); taosRLockLatch(&pMgmt->latch);
@ -107,23 +107,23 @@ static SVnodeObj *dndAcquireVnode(SDnode *pDnode, int32_t vgId) {
} }
taosRUnLockLatch(&pMgmt->latch); taosRUnLockLatch(&pMgmt->latch);
if (pVnode != NULL) {
dTrace("vgId:%d, acquire vnode, refCount:%d", pVnode->vgId, refCount); dTrace("vgId:%d, acquire vnode, refCount:%d", pVnode->vgId, refCount);
}
return pVnode; return pVnode;
} }
static void dndReleaseVnode(SDnode *pDnode, SVnodeObj *pVnode) { static void dndReleaseVnode(SDnode *pDnode, SVnodeObj *pVnode) {
if (pVnode == NULL) return;
SVnodesMgmt *pMgmt = &pDnode->vmgmt; SVnodesMgmt *pMgmt = &pDnode->vmgmt;
int32_t refCount = 0;
taosRLockLatch(&pMgmt->latch); taosRLockLatch(&pMgmt->latch);
if (pVnode != NULL) { int32_t refCount = atomic_sub_fetch_32(&pVnode->refCount, 1);
refCount = atomic_sub_fetch_32(&pVnode->refCount, 1);
}
taosRUnLockLatch(&pMgmt->latch); taosRUnLockLatch(&pMgmt->latch);
if (pVnode != NULL) {
dTrace("vgId:%d, release vnode, refCount:%d", pVnode->vgId, refCount); dTrace("vgId:%d, release vnode, refCount:%d", pVnode->vgId, refCount);
}
} }
static int32_t dndCreateVnodeWrapper(SDnode *pDnode, int32_t vgId, char *path, SVnode *pImpl) { static int32_t dndCreateVnodeWrapper(SDnode *pDnode, int32_t vgId, char *path, SVnode *pImpl) {
@ -457,7 +457,7 @@ static int32_t dndOpenVnodes(SDnode *pDnode) {
pMgmt->totalVnodes = numOfVnodes; pMgmt->totalVnodes = numOfVnodes;
int32_t threadNum = tsNumOfCores; int32_t threadNum = pDnode->opt.numOfCores;
int32_t vnodesPerThread = numOfVnodes / threadNum + 1; int32_t vnodesPerThread = numOfVnodes / threadNum + 1;
SVnodeThread *threads = calloc(threadNum, sizeof(SVnodeThread)); SVnodeThread *threads = calloc(threadNum, sizeof(SVnodeThread));
@ -525,33 +525,49 @@ static void dndCloseVnodes(SDnode *pDnode) {
static int32_t dndParseCreateVnodeReq(SRpcMsg *rpcMsg, int32_t *vgId, SVnodeCfg *pCfg) { static int32_t dndParseCreateVnodeReq(SRpcMsg *rpcMsg, int32_t *vgId, SVnodeCfg *pCfg) {
SCreateVnodeMsg *pCreate = rpcMsg->pCont; SCreateVnodeMsg *pCreate = rpcMsg->pCont;
*vgId = htonl(pCreate->vgId); pCreate->vgId = htonl(pCreate->vgId);
pCreate->dnodeId = htonl(pCreate->dnodeId);
pCreate->dbUid = htobe64(pCreate->dbUid);
pCreate->cacheBlockSize = htonl(pCreate->cacheBlockSize);
pCreate->totalBlocks = htonl(pCreate->totalBlocks);
pCreate->daysPerFile = htonl(pCreate->daysPerFile);
pCreate->daysToKeep0 = htonl(pCreate->daysToKeep0);
pCreate->daysToKeep1 = htonl(pCreate->daysToKeep1);
pCreate->daysToKeep2 = htonl(pCreate->daysToKeep2);
pCreate->minRows = htonl(pCreate->minRows);
pCreate->maxRows = htonl(pCreate->maxRows);
pCreate->commitTime = htonl(pCreate->commitTime);
pCreate->fsyncPeriod = htonl(pCreate->fsyncPeriod);
for (int r = 0; r < pCreate->replica; ++r) {
SReplica *pReplica = &pCreate->replicas[r];
pReplica->id = htonl(pReplica->id);
pReplica->port = htons(pReplica->port);
}
*vgId = pCreate->vgId;
#if 0 #if 0
tstrncpy(pCfg->db, pCreate->db, TSDB_FULL_DB_NAME_LEN); pCfg->wsize = pCreate->cacheBlockSize;
pCfg->cacheBlockSize = htonl(pCreate->cacheBlockSize); pCfg->ssize = pCreate->cacheBlockSize;
pCfg->totalBlocks = htonl(pCreate->totalBlocks); pCfg->wsize = pCreate->cacheBlockSize;
pCfg->daysPerFile = htonl(pCreate->daysPerFile); pCfg->lsize = pCreate->cacheBlockSize;
pCfg->daysToKeep0 = htonl(pCreate->daysToKeep0); pCfg->isHeapAllocator = true;
pCfg->daysToKeep1 = htonl(pCreate->daysToKeep1); pCfg->ttl = 4;
pCfg->daysToKeep2 = htonl(pCreate->daysToKeep2); pCfg->keep = pCreate->daysToKeep0;
pCfg->minRowsPerFileBlock = htonl(pCreate->minRowsPerFileBlock); pCfg->isWeak = true;
pCfg->maxRowsPerFileBlock = htonl(pCreate->maxRowsPerFileBlock); pCfg->tsdbCfg.keep0 = pCreate->daysToKeep0;
pCfg->precision = pCreate->precision; pCfg->tsdbCfg.keep1 = pCreate->daysToKeep2;
pCfg->compression = pCreate->compression; pCfg->tsdbCfg.keep2 = pCreate->daysToKeep0;
pCfg->cacheLastRow = pCreate->cacheLastRow; pCfg->tsdbCfg.lruCacheSize = pCreate->cacheBlockSize;
pCfg->update = pCreate->update; pCfg->metaCfg.lruSize = pCreate->cacheBlockSize;
pCfg->quorum = pCreate->quorum; pCfg->walCfg.fsyncPeriod = pCreate->fsyncPeriod;
pCfg->replica = pCreate->replica; pCfg->walCfg.level = pCreate->walLevel;
pCfg->walLevel = pCreate->walLevel; pCfg->walCfg.retentionPeriod = 10;
pCfg->fsyncPeriod = htonl(pCreate->fsyncPeriod); pCfg->walCfg.retentionSize = 128;
pCfg->walCfg.rollPeriod = 128;
for (int32_t i = 0; i < pCfg->replica; ++i) { pCfg->walCfg.segSize = 128;
pCfg->replicas[i].port = htons(pCreate->replicas[i].port); pCfg->walCfg.vgId = pCreate->vgId;
tstrncpy(pCfg->replicas[i].fqdn, pCreate->replicas[i].fqdn, TSDB_FQDN_LEN);
}
#endif #endif
return 0; return 0;
} }
@ -1016,7 +1032,7 @@ static int32_t dndInitVnodeWriteWorker(SDnode *pDnode) {
SVnodesMgmt * pMgmt = &pDnode->vmgmt; SVnodesMgmt * pMgmt = &pDnode->vmgmt;
SMWorkerPool *pPool = &pMgmt->writePool; SMWorkerPool *pPool = &pMgmt->writePool;
pPool->name = "vnode-write"; pPool->name = "vnode-write";
pPool->max = tsNumOfCores; pPool->max = pDnode->opt.numOfCores;
if (tMWorkerInit(pPool) != 0) { if (tMWorkerInit(pPool) != 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1; return -1;
@ -1050,7 +1066,7 @@ static void dndFreeVnodeSyncQueue(SDnode *pDnode, SVnodeObj *pVnode) {
} }
static int32_t dndInitVnodeSyncWorker(SDnode *pDnode) { static int32_t dndInitVnodeSyncWorker(SDnode *pDnode) {
int32_t maxThreads = tsNumOfCores / 2; int32_t maxThreads = pDnode->opt.numOfCores / 2;
if (maxThreads < 1) maxThreads = 1; if (maxThreads < 1) maxThreads = 1;
SVnodesMgmt *pMgmt = &pDnode->vmgmt; SVnodesMgmt *pMgmt = &pDnode->vmgmt;

2
source/dnode/mgmt/impl/test/CMakeLists.txt
View File

@ -15,6 +15,6 @@ add_subdirectory(stb)
# add_subdirectory(telem) # add_subdirectory(telem)
# add_subdirectory(trans) # add_subdirectory(trans)
add_subdirectory(user) add_subdirectory(user)
# add_subdirectory(vgroup) add_subdirectory(vgroup)
# add_subdirectory(common) # add_subdirectory(common)

1
source/dnode/mgmt/impl/test/db/db.cpp
View File

@ -232,6 +232,7 @@ TEST_F(DndTestDb, 02_Create_Alter_Drop_Db) {
SRpcMsg* pMsg = pClient->pRsp; SRpcMsg* pMsg = pClient->pRsp;
ASSERT_NE(pMsg, nullptr); ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, 0); ASSERT_EQ(pMsg->code, 0);
// taosMsleep(1000000);
} }
SendTheCheckShowMetaMsg(TSDB_MGMT_TABLE_DB, "show databases", 17, NULL); SendTheCheckShowMetaMsg(TSDB_MGMT_TABLE_DB, "show databases", 17, NULL);

27
source/dnode/mgmt/impl/test/vgroup/CMakeLists.txt Normal file
View File

@ -0,0 +1,27 @@
add_executable(dnode_test_vgroup "")
target_sources(dnode_test_vgroup
PRIVATE
"vgroup.cpp"
"../sut/deploy.cpp"
)
target_link_libraries(
dnode_test_vgroup
PUBLIC dnode
PUBLIC util
PUBLIC os
PUBLIC gtest_main
)
target_include_directories(dnode_test_vgroup
PUBLIC
"${CMAKE_SOURCE_DIR}/include/server/dnode/mgmt"
"${CMAKE_CURRENT_SOURCE_DIR}/../../inc"
"${CMAKE_CURRENT_SOURCE_DIR}/../sut"
)
add_test(
NAME dnode_test_vgroup
COMMAND dnode_test_vgroup
)

224
source/dnode/mgmt/impl/test/vgroup/vgroup.cpp Normal file
View File

@ -0,0 +1,224 @@
/**
* @file db.cpp
* @author slguan (slguan@taosdata.com)
* @brief DNODE module vgroup-msg tests
* @version 0.1
* @date 2021-12-20
*
* @copyright Copyright (c) 2021
*
*/
#include "deploy.h"
class DndTestVgroup : public ::testing::Test {
protected:
static SServer* CreateServer(const char* path, const char* fqdn, uint16_t port, const char* firstEp) {
SServer* pServer = createServer(path, fqdn, port, firstEp);
ASSERT(pServer);
return pServer;
}
static void SetUpTestSuite() {
initLog("/tmp/tdlog");
const char* fqdn = "localhost";
const char* firstEp = "localhost:9150";
pServer = CreateServer("/tmp/dnode_test_vgroup", fqdn, 9150, firstEp);
pClient = createClient("root", "taosdata", fqdn, 9150);
taosMsleep(1100);
}
static void TearDownTestSuite() {
stopServer(pServer);
dropClient(pClient);
pServer = NULL;
pClient = NULL;
}
static SServer* pServer;
static SClient* pClient;
static int32_t connId;
public:
void SetUp() override {}
void TearDown() override {}
void SendTheCheckShowMetaMsg(int8_t showType, const char* showName, int32_t columns, const char* db) {
SShowMsg* pShow = (SShowMsg*)rpcMallocCont(sizeof(SShowMsg));
pShow->type = showType;
if (db != NULL) {
strcpy(pShow->db, db);
}
SRpcMsg showRpcMsg = {0};
showRpcMsg.pCont = pShow;
showRpcMsg.contLen = sizeof(SShowMsg);
showRpcMsg.msgType = TSDB_MSG_TYPE_SHOW;
sendMsg(pClient, &showRpcMsg);
ASSERT_NE(pClient->pRsp, nullptr);
ASSERT_EQ(pClient->pRsp->code, 0);
ASSERT_NE(pClient->pRsp->pCont, nullptr);
SShowRsp* pShowRsp = (SShowRsp*)pClient->pRsp->pCont;
ASSERT_NE(pShowRsp, nullptr);
pShowRsp->showId = htonl(pShowRsp->showId);
pMeta = &pShowRsp->tableMeta;
pMeta->numOfTags = htonl(pMeta->numOfTags);
pMeta->numOfColumns = htonl(pMeta->numOfColumns);
pMeta->sversion = htonl(pMeta->sversion);
pMeta->tversion = htonl(pMeta->tversion);
pMeta->tuid = htobe64(pMeta->tuid);
pMeta->suid = htobe64(pMeta->suid);
showId = pShowRsp->showId;
EXPECT_NE(pShowRsp->showId, 0);
EXPECT_STREQ(pMeta->tbFname, showName);
EXPECT_EQ(pMeta->numOfTags, 0);
EXPECT_EQ(pMeta->numOfColumns, columns);
EXPECT_EQ(pMeta->precision, 0);
EXPECT_EQ(pMeta->tableType, 0);
EXPECT_EQ(pMeta->update, 0);
EXPECT_EQ(pMeta->sversion, 0);
EXPECT_EQ(pMeta->tversion, 0);
EXPECT_EQ(pMeta->tuid, 0);
EXPECT_EQ(pMeta->suid, 0);
}
void CheckSchema(int32_t index, int8_t type, int32_t bytes, const char* name) {
SSchema* pSchema = &pMeta->pSchema[index];
pSchema->bytes = htonl(pSchema->bytes);
EXPECT_EQ(pSchema->colId, 0);
EXPECT_EQ(pSchema->type, type);
EXPECT_EQ(pSchema->bytes, bytes);
EXPECT_STREQ(pSchema->name, name);
}
void SendThenCheckShowRetrieveMsg(int32_t rows) {
SRetrieveTableMsg* pRetrieve = (SRetrieveTableMsg*)rpcMallocCont(sizeof(SRetrieveTableMsg));
pRetrieve->showId = htonl(showId);
pRetrieve->free = 0;
SRpcMsg retrieveRpcMsg = {0};
retrieveRpcMsg.pCont = pRetrieve;
retrieveRpcMsg.contLen = sizeof(SRetrieveTableMsg);
retrieveRpcMsg.msgType = TSDB_MSG_TYPE_SHOW_RETRIEVE;
sendMsg(pClient, &retrieveRpcMsg);
ASSERT_NE(pClient->pRsp, nullptr);
ASSERT_EQ(pClient->pRsp->code, 0);
ASSERT_NE(pClient->pRsp->pCont, nullptr);
pRetrieveRsp = (SRetrieveTableRsp*)pClient->pRsp->pCont;
ASSERT_NE(pRetrieveRsp, nullptr);
pRetrieveRsp->numOfRows = htonl(pRetrieveRsp->numOfRows);
pRetrieveRsp->useconds = htobe64(pRetrieveRsp->useconds);
pRetrieveRsp->compLen = htonl(pRetrieveRsp->compLen);
EXPECT_EQ(pRetrieveRsp->numOfRows, rows);
EXPECT_EQ(pRetrieveRsp->useconds, 0);
// EXPECT_EQ(pRetrieveRsp->completed, completed);
EXPECT_EQ(pRetrieveRsp->precision, TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(pRetrieveRsp->compressed, 0);
EXPECT_EQ(pRetrieveRsp->compLen, 0);
pData = pRetrieveRsp->data;
pos = 0;
}
void CheckInt8(int8_t val) {
int8_t data = *((int8_t*)(pData + pos));
pos += sizeof(int8_t);
EXPECT_EQ(data, val);
}
void CheckInt16(int16_t val) {
int16_t data = *((int16_t*)(pData + pos));
pos += sizeof(int16_t);
EXPECT_EQ(data, val);
}
void CheckInt32(int32_t val) {
int32_t data = *((int32_t*)(pData + pos));
pos += sizeof(int32_t);
EXPECT_EQ(data, val);
}
void CheckInt64(int64_t val) {
int64_t data = *((int64_t*)(pData + pos));
pos += sizeof(int64_t);
EXPECT_EQ(data, val);
}
void CheckTimestamp() {
int64_t data = *((int64_t*)(pData + pos));
pos += sizeof(int64_t);
EXPECT_GT(data, 0);
}
void CheckBinary(const char* val, int32_t len) {
pos += sizeof(VarDataLenT);
char* data = (char*)(pData + pos);
pos += len;
EXPECT_STREQ(data, val);
}
int32_t showId;
STableMetaMsg* pMeta;
SRetrieveTableRsp* pRetrieveRsp;
char* pData;
int32_t pos;
};
SServer* DndTestVgroup::pServer;
SClient* DndTestVgroup::pClient;
int32_t DndTestVgroup::connId;
TEST_F(DndTestVgroup, 01_Create_Restart_Drop_Vnode) {
{
SCreateVnodeMsg* pReq = (SCreateVnodeMsg*)rpcMallocCont(sizeof(SCreateVnodeMsg));
pReq->vgId = htonl(2);
pReq->dnodeId = htonl(1);
strcpy(pReq->db, "1.d1");
pReq->dbUid = htobe64(9527);
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(SCreateVnodeMsg);
rpcMsg.msgType = TSDB_MSG_TYPE_CREATE_VNODE_IN;
sendMsg(pClient, &rpcMsg);
SRpcMsg* pMsg = pClient->pRsp;
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, 0);
taosMsleep(1000000);
}
}

4
source/dnode/mnode/impl/inc/mndTrans.h
View File

@ -30,8 +30,8 @@ void mndTransDrop(STrans *pTrans);
int32_t mndTransAppendRedolog(STrans *pTrans, SSdbRaw *pRaw); int32_t mndTransAppendRedolog(STrans *pTrans, SSdbRaw *pRaw);
int32_t mndTransAppendUndolog(STrans *pTrans, SSdbRaw *pRaw); int32_t mndTransAppendUndolog(STrans *pTrans, SSdbRaw *pRaw);
int32_t mndTransAppendCommitlog(STrans *pTrans, SSdbRaw *pRaw); int32_t mndTransAppendCommitlog(STrans *pTrans, SSdbRaw *pRaw);
int32_t mndTransAppendRedoAction(STrans *pTrans, SEpSet *, void *pMsg); int32_t mndTransAppendRedoAction(STrans *pTrans, SEpSet *pEpSet, int8_t msgType, int32_t contLen, void *pCont);
int32_t mndTransAppendUndoAction(STrans *pTrans, SEpSet *, void *pMsg); int32_t mndTransAppendUndoAction(STrans *pTrans, SEpSet *pEpSet, int8_t msgType, int32_t contLen, void *pCont);
int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans); int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans);
void mndTransApply(SMnode *pMnode, SSdbRaw *pRaw, STransMsg *pMsg, int32_t code); void mndTransApply(SMnode *pMnode, SSdbRaw *pRaw, STransMsg *pMsg, int32_t code);
char *mndTransStageStr(ETrnStage stage); char *mndTransStageStr(ETrnStage stage);

4
source/dnode/mnode/impl/inc/mndVgroup.h
View File

@ -28,7 +28,9 @@ SVgObj *mndAcquireVgroup(SMnode *pMnode, int32_t vgId);
void mndReleaseVgroup(SMnode *pMnode, SVgObj *pVgroup); void mndReleaseVgroup(SMnode *pMnode, SVgObj *pVgroup);
int32_t mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups); int32_t mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups);
SSdbRaw *mndVgroupActionEncode(SVgObj *pVgroup); SSdbRaw *mndVgroupActionEncode(SVgObj *pVgroup);
SSdbRow *mndVgroupActionDecode(SSdbRaw *pRaw);
SCreateVnodeMsg *mndBuildCreateVnodeMsg(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup);
SDropVnodeMsg *mndBuildDropVnodeMsg(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup);
#ifdef __cplusplus #ifdef __cplusplus
} }

50
source/dnode/mnode/impl/src/mndDb.c
View File

@ -285,10 +285,60 @@ static int32_t mndSetCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVg
} }
static int32_t mndSetRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroups) { static int32_t mndSetRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroups) {
for (int v = 0; v < pDb->cfg.numOfVgroups; ++v) {
SVgObj *pVgroup = pVgroups + v;
for (int32_t vn = 0; vn < pVgroup->replica; ++vn) {
SVnodeGid *pVgid = pVgroup->vnodeGid + vn;
SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
if (pDnode == NULL) {
return -1;
}
SEpSet epset = mndGetDnodeEpset(pDnode);
mndReleaseDnode(pMnode, pDnode);
SCreateVnodeMsg *pMsg = mndBuildCreateVnodeMsg(pMnode, pDnode, pDb, pVgroup);
if (pMsg == NULL) {
return -1;
}
if (mndTransAppendRedoAction(pTrans, &epset, TSDB_MSG_TYPE_ALTER_VNODE_IN, sizeof(SCreateVnodeMsg), pMsg) != 0) {
free(pMsg);
return -1;
}
}
}
return 0; return 0;
} }
static int32_t mndSetUndoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroups) { static int32_t mndSetUndoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroups) {
for (int v = 0; v < pDb->cfg.numOfVgroups; ++v) {
SVgObj *pVgroup = pVgroups + v;
for (int32_t vn = 0; vn < pVgroup->replica; ++vn) {
SVnodeGid *pVgid = pVgroup->vnodeGid + vn;
SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
if (pDnode == NULL) {
return -1;
}
SEpSet epset = mndGetDnodeEpset(pDnode);
mndReleaseDnode(pMnode, pDnode);
SDropVnodeMsg *pMsg = mndBuildDropVnodeMsg(pMnode, pDnode, pDb, pVgroup);
if (pMsg == NULL) {
return -1;
}
if (mndTransAppendUndoAction(pTrans, &epset, TSDB_MSG_TYPE_DROP_VNODE_IN, sizeof(SDropVnodeMsg), pMsg) != 0) {
free(pMsg);
return -1;
}
}
}
return 0; return 0;
} }

6
source/dnode/mnode/impl/src/mndDnode.c
View File

@ -181,7 +181,11 @@ static int32_t mndDnodeActionUpdate(SSdb *pSdb, SDnodeObj *pOldDnode, SDnodeObj
SDnodeObj *mndAcquireDnode(SMnode *pMnode, int32_t dnodeId) { SDnodeObj *mndAcquireDnode(SMnode *pMnode, int32_t dnodeId) {
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
return sdbAcquire(pSdb, SDB_DNODE, &dnodeId); SDnodeObj *pDnode = sdbAcquire(pSdb, SDB_DNODE, &dnodeId);
if (pDnode == NULL) {
terrno = TSDB_CODE_MND_DNODE_NOT_EXIST;
}
return pDnode;
} }
void mndReleaseDnode(SMnode *pMnode, SDnodeObj *pDnode) { void mndReleaseDnode(SMnode *pMnode, SDnodeObj *pDnode) {

208
source/dnode/mnode/impl/src/mndTrans.c
View File

@ -21,6 +21,13 @@
#define TSDB_TRN_ARRAY_SIZE 8 #define TSDB_TRN_ARRAY_SIZE 8
#define TSDB_TRN_RESERVE_SIZE 64 #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 SSdbRaw *mndTransActionEncode(STrans *pTrans);
static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw); static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw);
static int32_t mndTransActionInsert(SSdb *pSdb, STrans *pTrans); static int32_t mndTransActionInsert(SSdb *pSdb, STrans *pTrans);
@ -29,9 +36,12 @@ static int32_t mndTransActionDelete(SSdb *pSdb, STrans *pTrans);
static void mndTransSetRpcHandle(STrans *pTrans, void *rpcHandle); static void mndTransSetRpcHandle(STrans *pTrans, void *rpcHandle);
static void mndTransSendRpcRsp(STrans *pTrans, int32_t code); static void mndTransSendRpcRsp(STrans *pTrans, int32_t code);
static int32_t mndTransAppendArray(SArray *pArray, SSdbRaw *pRaw); static int32_t mndTransAppendLog(SArray *pArray, SSdbRaw *pRaw);
static void mndTransDropArray(SArray *pArray); static int32_t mndTransAppendAction(SArray *pArray, SEpSet *pEpSet, int8_t msgType, int32_t contLen, void *pCont);
static int32_t mndTransExecuteArray(SMnode *pMnode, SArray *pArray); 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 mndTransExecuteRedoLogs(SMnode *pMnode, STrans *pTrans); static int32_t mndTransExecuteRedoLogs(SMnode *pMnode, STrans *pTrans);
static int32_t mndTransExecuteUndoLogs(SMnode *pMnode, STrans *pTrans); static int32_t mndTransExecuteUndoLogs(SMnode *pMnode, STrans *pTrans);
static int32_t mndTransExecuteCommitLogs(SMnode *pMnode, STrans *pTrans); static int32_t mndTransExecuteCommitLogs(SMnode *pMnode, STrans *pTrans);
@ -58,7 +68,7 @@ int32_t mndInitTrans(SMnode *pMnode) {
void mndCleanupTrans(SMnode *pMnode) {} void mndCleanupTrans(SMnode *pMnode) {}
static SSdbRaw *mndTransActionEncode(STrans *pTrans) { static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
int32_t rawDataLen = 16 * sizeof(int32_t) + TSDB_TRN_RESERVE_SIZE; int32_t rawDataLen = sizeof(STrans) + TSDB_TRN_RESERVE_SIZE;
int32_t redoLogNum = taosArrayGetSize(pTrans->redoLogs); int32_t redoLogNum = taosArrayGetSize(pTrans->redoLogs);
int32_t undoLogNum = taosArrayGetSize(pTrans->undoLogs); int32_t undoLogNum = taosArrayGetSize(pTrans->undoLogs);
int32_t commitLogNum = taosArrayGetSize(pTrans->commitLogs); int32_t commitLogNum = taosArrayGetSize(pTrans->commitLogs);
@ -80,6 +90,16 @@ static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
rawDataLen += sdbGetRawTotalSize(pTmp); rawDataLen += sdbGetRawTotalSize(pTmp);
} }
for (int32_t i = 0; i < redoActionNum; ++i) {
STransAction *pAction = taosArrayGet(pTrans->redoActions, i);
rawDataLen += (sizeof(STransAction) + pAction->contLen);
}
for (int32_t i = 0; i < undoActionNum; ++i) {
STransAction *pAction = taosArrayGet(pTrans->undoActions, i);
rawDataLen += (sizeof(STransAction) + pAction->contLen);
}
SSdbRaw *pRaw = sdbAllocRaw(SDB_TRANS, TSDB_TRANS_VER, rawDataLen); SSdbRaw *pRaw = sdbAllocRaw(SDB_TRANS, TSDB_TRANS_VER, rawDataLen);
if (pRaw == NULL) { if (pRaw == NULL) {
mError("trans:%d, failed to alloc raw since %s", pTrans->id, terrstr()); mError("trans:%d, failed to alloc raw since %s", pTrans->id, terrstr());
@ -116,6 +136,22 @@ static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
SDB_SET_BINARY(pRaw, dataPos, (void *)pTmp, len) SDB_SET_BINARY(pRaw, dataPos, (void *)pTmp, len)
} }
for (int32_t i = 0; i < redoActionNum; ++i) {
STransAction *pAction = taosArrayGet(pTrans->redoActions, i);
SDB_SET_BINARY(pRaw, dataPos, (void *)&pAction->epSet, sizeof(SEpSet));
SDB_SET_INT8(pRaw, dataPos, pAction->msgType)
SDB_SET_INT32(pRaw, dataPos, pAction->contLen)
SDB_SET_BINARY(pRaw, dataPos, pAction->pCont, pAction->contLen);
}
for (int32_t i = 0; i < undoActionNum; ++i) {
STransAction *pAction = taosArrayGet(pTrans->undoActions, i);
SDB_SET_BINARY(pRaw, dataPos, (void *)&pAction->epSet, sizeof(SEpSet));
SDB_SET_INT8(pRaw, dataPos, pAction->msgType)
SDB_SET_INT32(pRaw, dataPos, pAction->contLen)
SDB_SET_BINARY(pRaw, dataPos, (void *)pAction->pCont, pAction->contLen);
}
SDB_SET_RESERVE(pRaw, dataPos, TSDB_TRN_RESERVE_SIZE) SDB_SET_RESERVE(pRaw, dataPos, TSDB_TRN_RESERVE_SIZE)
SDB_SET_DATALEN(pRaw, dataPos); SDB_SET_DATALEN(pRaw, dataPos);
mTrace("trans:%d, encode to raw:%p, len:%d", pTrans->id, pRaw, dataPos); mTrace("trans:%d, encode to raw:%p, len:%d", pTrans->id, pRaw, dataPos);
@ -147,8 +183,8 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
pTrans->redoLogs = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *)); pTrans->redoLogs = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *));
pTrans->undoLogs = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *)); pTrans->undoLogs = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *));
pTrans->commitLogs = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *)); pTrans->commitLogs = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *));
pTrans->redoActions = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *)); pTrans->redoActions = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(STransAction));
pTrans->undoActions = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *)); pTrans->undoActions = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(STransAction));
if (pTrans->redoLogs == NULL || pTrans->undoLogs == NULL || pTrans->commitLogs == NULL || if (pTrans->redoLogs == NULL || pTrans->undoLogs == NULL || pTrans->commitLogs == NULL ||
pTrans->redoActions == NULL || pTrans->undoActions == NULL) { pTrans->redoActions == NULL || pTrans->undoActions == NULL) {
@ -175,42 +211,77 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
for (int32_t i = 0; i < redoLogNum; ++i) { for (int32_t i = 0; i < redoLogNum; ++i) {
int32_t dataLen = 0; int32_t dataLen = 0;
SDB_GET_INT32(pRaw, pRow, dataPos, &dataLen) SDB_GET_INT32(pRaw, pRow, dataPos, &dataLen)
char *pData = malloc(dataLen); char *pData = malloc(dataLen);
SDB_GET_BINARY(pRaw, pRow, dataPos, pData, dataLen); SDB_GET_BINARY(pRaw, pRow, dataPos, pData, dataLen);
void *ret = taosArrayPush(pTrans->redoLogs, &pData); void *ret = taosArrayPush(pTrans->redoLogs, &pData);
if (ret == NULL) { if (ret == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto TRANS_DECODE_OVER; goto TRANS_DECODE_OVER;
break;
} }
} }
for (int32_t i = 0; i < undoLogNum; ++i) { for (int32_t i = 0; i < undoLogNum; ++i) {
int32_t dataLen = 0; int32_t dataLen = 0;
SDB_GET_INT32(pRaw, pRow, dataPos, &dataLen) SDB_GET_INT32(pRaw, pRow, dataPos, &dataLen)
char *pData = malloc(dataLen); char *pData = malloc(dataLen);
SDB_GET_BINARY(pRaw, pRow, dataPos, pData, dataLen); SDB_GET_BINARY(pRaw, pRow, dataPos, pData, dataLen);
void *ret = taosArrayPush(pTrans->undoLogs, &pData); void *ret = taosArrayPush(pTrans->undoLogs, &pData);
if (ret == NULL) { if (ret == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto TRANS_DECODE_OVER; goto TRANS_DECODE_OVER;
break;
} }
} }
for (int32_t i = 0; i < commitLogNum; ++i) { for (int32_t i = 0; i < commitLogNum; ++i) {
int32_t dataLen = 0; int32_t dataLen = 0;
SDB_GET_INT32(pRaw, pRow, dataPos, &dataLen) SDB_GET_INT32(pRaw, pRow, dataPos, &dataLen)
char *pData = malloc(dataLen); char *pData = malloc(dataLen);
SDB_GET_BINARY(pRaw, pRow, dataPos, pData, dataLen); SDB_GET_BINARY(pRaw, pRow, dataPos, pData, dataLen);
void *ret = taosArrayPush(pTrans->commitLogs, &pData); void *ret = taosArrayPush(pTrans->commitLogs, &pData);
if (ret == NULL) { if (ret == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto TRANS_DECODE_OVER; goto TRANS_DECODE_OVER;
break; }
}
for (int32_t i = 0; i < redoActionNum; ++i) {
STransAction action = {0};
SDB_GET_BINARY(pRaw, pRow, dataPos, (void *)&action.epSet, sizeof(SEpSet));
SDB_GET_INT8(pRaw, pRow, dataPos, &action.msgType)
SDB_GET_INT32(pRaw, pRow, dataPos, &action.contLen)
action.pCont = malloc(action.contLen);
if (action.pCont == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto TRANS_DECODE_OVER;
}
SDB_GET_BINARY(pRaw, pRow, dataPos, action.pCont, action.contLen);
void *ret = taosArrayPush(pTrans->redoActions, &action);
if (ret == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto TRANS_DECODE_OVER;
}
}
for (int32_t i = 0; i < undoActionNum; ++i) {
STransAction action = {0};
SDB_GET_BINARY(pRaw, pRow, dataPos, (void *)&action.epSet, sizeof(SEpSet));
SDB_GET_INT8(pRaw, pRow, dataPos, &action.msgType)
SDB_GET_INT32(pRaw, pRow, dataPos, &action.contLen)
action.pCont = malloc(action.contLen);
if (action.pCont == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto TRANS_DECODE_OVER;
}
SDB_GET_BINARY(pRaw, pRow, dataPos, action.pCont, action.contLen);
void *ret = taosArrayPush(pTrans->undoActions, &action);
if (ret == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto TRANS_DECODE_OVER;
} }
} }
@ -237,11 +308,11 @@ static int32_t mndTransActionInsert(SSdb *pSdb, STrans *pTrans) {
static int32_t mndTransActionDelete(SSdb *pSdb, STrans *pTrans) { static int32_t mndTransActionDelete(SSdb *pSdb, STrans *pTrans) {
mTrace("trans:%d, perform delete action, stage:%s", pTrans->id, mndTransStageStr(pTrans->stage)); mTrace("trans:%d, perform delete action, stage:%s", pTrans->id, mndTransStageStr(pTrans->stage));
mndTransDropArray(pTrans->redoLogs); mndTransDropLogs(pTrans->redoLogs);
mndTransDropArray(pTrans->undoLogs); mndTransDropLogs(pTrans->undoLogs);
mndTransDropArray(pTrans->commitLogs); mndTransDropLogs(pTrans->commitLogs);
mndTransDropArray(pTrans->redoActions); mndTransDropActions(pTrans->redoActions);
mndTransDropArray(pTrans->undoActions); mndTransDropActions(pTrans->undoActions);
return 0; return 0;
} }
@ -274,6 +345,8 @@ char *mndTransStageStr(ETrnStage stage) {
return "rollback"; return "rollback";
case TRN_STAGE_RETRY: case TRN_STAGE_RETRY:
return "retry"; return "retry";
case TRN_STAGE_OVER:
return "stop";
default: default:
return "undefined"; return "undefined";
} }
@ -305,8 +378,8 @@ STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, void *rpcHandle) {
pTrans->redoLogs = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *)); pTrans->redoLogs = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *));
pTrans->undoLogs = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *)); pTrans->undoLogs = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *));
pTrans->commitLogs = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *)); pTrans->commitLogs = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *));
pTrans->redoActions = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *)); pTrans->redoActions = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(STransAction));
pTrans->undoActions = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(void *)); pTrans->undoActions = taosArrayInit(TSDB_TRN_ARRAY_SIZE, sizeof(STransAction));
if (pTrans->redoLogs == NULL || pTrans->undoLogs == NULL || pTrans->commitLogs == NULL || if (pTrans->redoLogs == NULL || pTrans->undoLogs == NULL || pTrans->commitLogs == NULL ||
pTrans->redoActions == NULL || pTrans->undoActions == NULL) { pTrans->redoActions == NULL || pTrans->undoActions == NULL) {
@ -319,7 +392,7 @@ STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, void *rpcHandle) {
return pTrans; return pTrans;
} }
static void mndTransDropArray(SArray *pArray) { static void mndTransDropLogs(SArray *pArray) {
for (int32_t i = 0; i < pArray->size; ++i) { for (int32_t i = 0; i < pArray->size; ++i) {
SSdbRaw *pRaw = taosArrayGetP(pArray, i); SSdbRaw *pRaw = taosArrayGetP(pArray, i);
tfree(pRaw); tfree(pRaw);
@ -328,12 +401,21 @@ static void mndTransDropArray(SArray *pArray) {
taosArrayDestroy(pArray); taosArrayDestroy(pArray);
} }
static void mndTransDropActions(SArray *pArray) {
for (int32_t i = 0; i < pArray->size; ++i) {
STransAction *pAction = taosArrayGet(pArray, i);
free(pAction->pCont);
}
taosArrayDestroy(pArray);
}
void mndTransDrop(STrans *pTrans) { void mndTransDrop(STrans *pTrans) {
mndTransDropArray(pTrans->redoLogs); mndTransDropLogs(pTrans->redoLogs);
mndTransDropArray(pTrans->undoLogs); mndTransDropLogs(pTrans->undoLogs);
mndTransDropArray(pTrans->commitLogs); mndTransDropLogs(pTrans->commitLogs);
mndTransDropArray(pTrans->redoActions); mndTransDropActions(pTrans->redoActions);
mndTransDropArray(pTrans->undoActions); mndTransDropActions(pTrans->undoActions);
mDebug("trans:%d, data:%p is dropped", pTrans->id, pTrans); mDebug("trans:%d, data:%p is dropped", pTrans->id, pTrans);
tfree(pTrans); tfree(pTrans);
@ -344,7 +426,7 @@ static void mndTransSetRpcHandle(STrans *pTrans, void *rpcHandle) {
mTrace("trans:%d, set rpc handle:%p", pTrans->id, rpcHandle); mTrace("trans:%d, set rpc handle:%p", pTrans->id, rpcHandle);
} }
static int32_t mndTransAppendArray(SArray *pArray, SSdbRaw *pRaw) { static int32_t mndTransAppendLog(SArray *pArray, SSdbRaw *pRaw) {
if (pArray == NULL || pRaw == NULL) { if (pArray == NULL || pRaw == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1; return -1;
@ -360,32 +442,44 @@ static int32_t mndTransAppendArray(SArray *pArray, SSdbRaw *pRaw) {
} }
int32_t mndTransAppendRedolog(STrans *pTrans, SSdbRaw *pRaw) { int32_t mndTransAppendRedolog(STrans *pTrans, SSdbRaw *pRaw) {
int32_t code = mndTransAppendArray(pTrans->redoLogs, pRaw); int32_t code = mndTransAppendLog(pTrans->redoLogs, pRaw);
mTrace("trans:%d, raw:%p append to redo logs, code:0x%x", pTrans->id, pRaw, code); mTrace("trans:%d, raw:%p append to redo logs, code:0x%x", pTrans->id, pRaw, code);
return code; return code;
} }
int32_t mndTransAppendUndolog(STrans *pTrans, SSdbRaw *pRaw) { int32_t mndTransAppendUndolog(STrans *pTrans, SSdbRaw *pRaw) {
int32_t code = mndTransAppendArray(pTrans->undoLogs, pRaw); int32_t code = mndTransAppendLog(pTrans->undoLogs, pRaw);
mTrace("trans:%d, raw:%p append to undo logs, code:0x%x", pTrans->id, pRaw, code); mTrace("trans:%d, raw:%p append to undo logs, code:0x%x", pTrans->id, pRaw, code);
return code; return code;
} }
int32_t mndTransAppendCommitlog(STrans *pTrans, SSdbRaw *pRaw) { int32_t mndTransAppendCommitlog(STrans *pTrans, SSdbRaw *pRaw) {
int32_t code = mndTransAppendArray(pTrans->commitLogs, pRaw); int32_t code = mndTransAppendLog(pTrans->commitLogs, pRaw);
mTrace("trans:%d, raw:%p append to commit logs, code:0x%x", pTrans->id, pRaw, code); mTrace("trans:%d, raw:%p append to commit logs, code:0x%x", pTrans->id, pRaw, code);
return code; return code;
} }
int32_t mndTransAppendRedoAction(STrans *pTrans, SEpSet *pEpSet, void *pMsg) { static int32_t mndTransAppendAction(SArray *pArray, SEpSet *pEpSet, int8_t msgType, int32_t contLen, void *pCont) {
int32_t code = mndTransAppendArray(pTrans->redoActions, pMsg); STransAction action = {.epSet = *pEpSet, .msgType = msgType, .contLen = contLen, .pCont = pCont};
mTrace("trans:%d, msg:%p append to redo actions", pTrans->id, pMsg);
void *ptr = taosArrayPush(pArray, &action);
if (ptr == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
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; return code;
} }
int32_t mndTransAppendUndoAction(STrans *pTrans, SEpSet *pEpSet, void *pMsg) { int32_t mndTransAppendUndoAction(STrans *pTrans, SEpSet *pEpSet, int8_t msgType, int32_t contLen, void *pCont) {
int32_t code = mndTransAppendArray(pTrans->undoActions, pMsg); int32_t code = mndTransAppendAction(pTrans->undoActions, pEpSet, msgType, contLen, pCont);
mTrace("trans:%d, msg:%p append to undo actions", pTrans->id, pMsg); mTrace("trans:%d, msg:%s len:%d append to undo actions", pTrans->id, taosMsg[msgType], contLen);
return code; return code;
} }
@ -502,7 +596,7 @@ void mndTransApply(SMnode *pMnode, SSdbRaw *pRaw, STransMsg *pMsg, int32_t code)
// todo // todo
} }
static int32_t mndTransExecuteArray(SMnode *pMnode, SArray *pArray) { static int32_t mndTransExecuteLogs(SMnode *pMnode, SArray *pArray) {
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
int32_t arraySize = taosArrayGetSize(pArray); int32_t arraySize = taosArrayGetSize(pArray);
@ -520,7 +614,7 @@ static int32_t mndTransExecuteArray(SMnode *pMnode, SArray *pArray) {
static int32_t mndTransExecuteRedoLogs(SMnode *pMnode, STrans *pTrans) { static int32_t mndTransExecuteRedoLogs(SMnode *pMnode, STrans *pTrans) {
int32_t code = 0; int32_t code = 0;
if (taosArrayGetSize(pTrans->redoLogs) != 0) { if (taosArrayGetSize(pTrans->redoLogs) != 0) {
code = mndTransExecuteArray(pMnode, pTrans->redoLogs); code = mndTransExecuteLogs(pMnode, pTrans->redoLogs);
if (code != 0) { if (code != 0) {
mError("trans:%d, failed to execute redo logs since %s", pTrans->id, terrstr()) mError("trans:%d, failed to execute redo logs since %s", pTrans->id, terrstr())
} else { } else {
@ -534,7 +628,7 @@ static int32_t mndTransExecuteRedoLogs(SMnode *pMnode, STrans *pTrans) {
static int32_t mndTransExecuteUndoLogs(SMnode *pMnode, STrans *pTrans) { static int32_t mndTransExecuteUndoLogs(SMnode *pMnode, STrans *pTrans) {
int32_t code = 0; int32_t code = 0;
if (taosArrayGetSize(pTrans->undoLogs) != 0) { if (taosArrayGetSize(pTrans->undoLogs) != 0) {
code = mndTransExecuteArray(pMnode, pTrans->undoLogs); code = mndTransExecuteLogs(pMnode, pTrans->undoLogs);
if (code != 0) { if (code != 0) {
mError("trans:%d, failed to execute undo logs since %s", pTrans->id, terrstr()) mError("trans:%d, failed to execute undo logs since %s", pTrans->id, terrstr())
} else { } else {
@ -548,7 +642,7 @@ static int32_t mndTransExecuteUndoLogs(SMnode *pMnode, STrans *pTrans) {
static int32_t mndTransExecuteCommitLogs(SMnode *pMnode, STrans *pTrans) { static int32_t mndTransExecuteCommitLogs(SMnode *pMnode, STrans *pTrans) {
int32_t code = 0; int32_t code = 0;
if (taosArrayGetSize(pTrans->commitLogs) != 0) { if (taosArrayGetSize(pTrans->commitLogs) != 0) {
code = mndTransExecuteArray(pMnode, pTrans->commitLogs); code = mndTransExecuteLogs(pMnode, pTrans->commitLogs);
if (code != 0) { if (code != 0) {
mError("trans:%d, failed to execute commit logs since %s", pTrans->id, terrstr()) mError("trans:%d, failed to execute commit logs since %s", pTrans->id, terrstr())
} else { } else {
@ -559,18 +653,40 @@ static int32_t mndTransExecuteCommitLogs(SMnode *pMnode, STrans *pTrans) {
return code; return code;
} }
static int32_t mndTransExecuteRedoActions(SMnode *pMnode, STrans *pTrans) { static int32_t mndTransExecuteActions(SMnode *pMnode, SArray *pArray) {
if (taosArrayGetSize(pTrans->redoActions) != 0) { #if 0
mTrace("trans:%d, execute redo actions finished", pTrans->id); int32_t arraySize = taosArrayGetSize(pArray);
for (int32_t i = 0; i < arraySize; ++i) {
STransAction *pAction = taosArrayGet(pArray, i);
SRpcMsg rpcMsg = {.msgType = pAction->msgType, .contLen = pAction->contLen};
rpcMsg.pCont = rpcMallocCont(pAction->contLen);
if (rpcMsg.pCont == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
} }
memcpy(rpcMsg.pCont, pAction->pCont, pAction->contLen);
mndSendMsgToDnode(pMnode, &pAction->epSet, &rpcMsg);
}
return TSDB_CODE_MND_ACTION_IN_PROGRESS;
#else
return 0; 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);
} }
static int32_t mndTransExecuteUndoActions(SMnode *pMnode, STrans *pTrans) { static int32_t mndTransExecuteUndoActions(SMnode *pMnode, STrans *pTrans) {
if (taosArrayGetSize(pTrans->undoActions) != 0) { if (taosArrayGetSize(pTrans->undoActions) <= 0) return 0;
mTrace("trans:%d, execute undo actions finished", pTrans->id);
} mTrace("trans:%d, start to execute undo actions", pTrans->id);
return 0; return mndTransExecuteActions(pMnode, pTrans->undoActions);
} }
static int32_t mndTransPerformPrepareStage(SMnode *pMnode, STrans *pTrans) { static int32_t mndTransPerformPrepareStage(SMnode *pMnode, STrans *pTrans) {

75
source/dnode/mnode/impl/src/mndVgroup.c
View File

@ -24,6 +24,7 @@
#define TSDB_VGROUP_VER_NUM 1 #define TSDB_VGROUP_VER_NUM 1
#define TSDB_VGROUP_RESERVE_SIZE 64 #define TSDB_VGROUP_RESERVE_SIZE 64
static SSdbRow *mndVgroupActionDecode(SSdbRaw *pRaw);
static int32_t mndVgroupActionInsert(SSdb *pSdb, SVgObj *pVgroup); static int32_t mndVgroupActionInsert(SSdb *pSdb, SVgObj *pVgroup);
static int32_t mndVgroupActionDelete(SSdb *pSdb, SVgObj *pVgroup); static int32_t mndVgroupActionDelete(SSdb *pSdb, SVgObj *pVgroup);
static int32_t mndVgroupActionUpdate(SSdb *pSdb, SVgObj *pOldVgroup, SVgObj *pNewVgroup); static int32_t mndVgroupActionUpdate(SSdb *pSdb, SVgObj *pOldVgroup, SVgObj *pNewVgroup);
@ -156,6 +157,80 @@ void mndReleaseVgroup(SMnode *pMnode, SVgObj *pVgroup) {
sdbRelease(pSdb, pVgroup); sdbRelease(pSdb, pVgroup);
} }
SCreateVnodeMsg *mndBuildCreateVnodeMsg(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup) {
SCreateVnodeMsg *pCreate = malloc(sizeof(SCreateVnodeMsg));
if (pCreate == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
pCreate->dnodeId = htonl(pDnode->id);
pCreate->vgId = htonl(pVgroup->vgId);
memcpy(pCreate->db, pDb->name, TSDB_FULL_DB_NAME_LEN);
pCreate->dbUid = htobe64(pDb->uid);
pCreate->cacheBlockSize = htonl(pDb->cfg.cacheBlockSize);
pCreate->totalBlocks = htonl(pDb->cfg.totalBlocks);
pCreate->daysPerFile = htonl(pDb->cfg.daysPerFile);
pCreate->daysToKeep0 = htonl(pDb->cfg.daysToKeep0);
pCreate->daysToKeep1 = htonl(pDb->cfg.daysToKeep1);
pCreate->daysToKeep2 = htonl(pDb->cfg.daysToKeep2);
pCreate->minRows = htonl(pDb->cfg.minRows);
pCreate->maxRows = htonl(pDb->cfg.maxRows);
pCreate->commitTime = htonl(pDb->cfg.commitTime);
pCreate->fsyncPeriod = htonl(pDb->cfg.fsyncPeriod);
pCreate->walLevel = pDb->cfg.walLevel;
pCreate->precision = pDb->cfg.precision;
pCreate->compression = pDb->cfg.compression;
pCreate->quorum = pDb->cfg.quorum;
pCreate->update = pDb->cfg.update;
pCreate->cacheLastRow = pDb->cfg.cacheLastRow;
pCreate->replica = pVgroup->replica;
pCreate->selfIndex = -1;
for (int32_t v = 0; v < pVgroup->replica; ++v) {
SReplica *pReplica = &pCreate->replicas[v];
SVnodeGid *pVgid = &pVgroup->vnodeGid[v];
SDnodeObj *pVgidDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
if (pVgidDnode == NULL) {
free(pCreate);
terrno = TSDB_CODE_MND_APP_ERROR;
return NULL;
}
pReplica->id = htonl(pVgidDnode->id);
pReplica->port = htons(pVgidDnode->port);
memcpy(pReplica->fqdn, pVgidDnode->fqdn, TSDB_FQDN_LEN);
mndReleaseDnode(pMnode, pVgidDnode);
if (pDnode->id == pVgid->dnodeId) {
pCreate->selfIndex = v;
}
}
if (pCreate->selfIndex == -1) {
free(pCreate);
terrno = TSDB_CODE_MND_APP_ERROR;
return NULL;
}
return pCreate;
}
SDropVnodeMsg *mndBuildDropVnodeMsg(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup) {
SDropVnodeMsg *pDrop = malloc(sizeof(SDropVnodeMsg));
if (pDrop == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
pDrop->dnodeId = htonl(pDnode->id);
pDrop->vgId = htonl(pVgroup->vgId);
memcpy(pDrop->db, pDb->name, TSDB_FULL_DB_NAME_LEN);
pDrop->dbUid = htobe64(pDb->uid);
return pDrop;
}
static int32_t mndGetAvailableDnode(SMnode *pMnode, SVgObj *pVgroup) { static int32_t mndGetAvailableDnode(SMnode *pMnode, SVgObj *pVgroup) {
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
int32_t allocedVnodes = 0; int32_t allocedVnodes = 0;

8
source/dnode/vnode/impl/src/vnodeCommit.c
View File

@ -25,6 +25,10 @@ int vnodeAsyncCommit(SVnode *pVnode) {
pTask->execute = vnodeCommit; // TODO pTask->execute = vnodeCommit; // TODO
pTask->arg = pVnode; // TODO pTask->arg = pVnode; // TODO
tsdbPrepareCommit(pVnode->pTsdb);
// metaPrepareCommit(pVnode->pMeta);
// walPreapareCommit(pVnode->pWal);
vnodeScheduleTask(pTask); vnodeScheduleTask(pTask);
return 0; return 0;
} }
@ -32,6 +36,10 @@ int vnodeAsyncCommit(SVnode *pVnode) {
int vnodeCommit(void *arg) { int vnodeCommit(void *arg) {
SVnode *pVnode = (SVnode *)arg; SVnode *pVnode = (SVnode *)arg;
metaCommit(pVnode->pMeta);
tqCommit(pVnode->pTq);
tsdbCommit(pVnode->pTq);
vnodeBufPoolRecycle(pVnode); vnodeBufPoolRecycle(pVnode);
// TODO // TODO
return 0; return 0;

2
source/dnode/vnode/tq/src/tq.c
View File

@ -50,7 +50,7 @@ STQ* tqOpen(const char* path, STqCfg* tqConfig, STqLogReader* tqLogReader, SMemA
pTq->tqConfig = tqConfig; pTq->tqConfig = tqConfig;
pTq->tqLogReader = tqLogReader; pTq->tqLogReader = tqLogReader;
pTq->tqMemRef.pAlloctorFactory = allocFac; pTq->tqMemRef.pAlloctorFactory = allocFac;
pTq->tqMemRef.pAllocator = allocFac->create(allocFac); // pTq->tqMemRef.pAllocator = allocFac->create(allocFac);
if (pTq->tqMemRef.pAllocator == NULL) { if (pTq->tqMemRef.pAllocator == NULL) {
// TODO // TODO
} }

12
source/dnode/vnode/tsdb/src/tsdbCommit.c
View File

@ -15,7 +15,19 @@
#include "tsdbDef.h" #include "tsdbDef.h"
int tsdbPrepareCommit(STsdb *pTsdb) {
if (pTsdb->mem == NULL) return 0;
// tsem_wait(&(pTsdb->canCommit));
ASSERT(pTsdb->imem == NULL);
pTsdb->imem = pTsdb->mem;
pTsdb->mem = NULL;
}
int tsdbCommit(STsdb *pTsdb) { int tsdbCommit(STsdb *pTsdb) {
// TODO // TODO
pTsdb->imem = NULL;
// tsem_post(&(pTsdb->canCommit));
return 0; return 0;
} }

6
source/libs/index/CMakeLists.txt
View File

@ -22,9 +22,13 @@ if (${BUILD_WITH_LUCENE})
index index
PUBLIC lucene++ PUBLIC lucene++
) )
endif(${BUILD_WITH_LUCENE}) endif(${BUILD_WITH_LUCENE})
if (${BUILD_WITH_INVERTEDINDEX})
add_definitions(-DUSE_INVERTED_INDEX)
endif(${BUILD_WITH_INVERTEDINDEX})
if (${BUILD_TEST}) if (${BUILD_TEST})
add_subdirectory(test) add_subdirectory(test)
endif(${BUILD_TEST}) endif(${BUILD_TEST})

23
source/libs/index/inc/indexInt.h
View File

@ -37,10 +37,10 @@ struct SIndex {
#endif #endif
void *cache; void *cache;
void *tindex; void *tindex;
SHashObj *fieldObj;// < field name, field id> SHashObj *colObj;// < field name, field id>
int64_t suid; // current super table id, -1 is normal table int64_t suid; // current super table id, -1 is normal table
int fieldId; // field id allocated to cache int colId; // field id allocated to cache
int32_t cVersion; // current version allocated to cache int32_t cVersion; // current version allocated to cache
pthread_mutex_t mtx; pthread_mutex_t mtx;
}; };
@ -60,22 +60,21 @@ struct SIndexMultiTermQuery {
// field and key; // field and key;
typedef struct SIndexTerm { typedef struct SIndexTerm {
uint8_t type; // term data type, str/interger/json int64_t suid;
char *key; SIndexOperOnColumn operType; // oper type, add/del/update
int32_t nKey; uint8_t colType; // term data type, str/interger/json
char *val; char *colName;
int32_t nVal; int32_t nColName;
char *colVal;
int32_t nColVal;
} SIndexTerm; } SIndexTerm;
typedef struct SIndexTermQuery { typedef struct SIndexTermQuery {
SIndexTerm* field_value; SIndexTerm* term;
EIndexQueryType type; EIndexQueryType qType;
} SIndexTermQuery; } SIndexTermQuery;
SIndexTerm *indexTermCreate(const char *key, int32_t nKey, const char *val, int32_t nVal);
void indexTermDestroy(SIndexTerm *p);
#define indexFatal(...) do { if (sDebugFlag & DEBUG_FATAL) { taosPrintLog("index FATAL ", 255, __VA_ARGS__); }} while(0) #define indexFatal(...) do { if (sDebugFlag & DEBUG_FATAL) { taosPrintLog("index FATAL ", 255, __VA_ARGS__); }} while(0)
#define indexError(...) do { if (sDebugFlag & DEBUG_ERROR) { taosPrintLog("index ERROR ", 255, __VA_ARGS__); }} while(0) #define indexError(...) do { if (sDebugFlag & DEBUG_ERROR) { taosPrintLog("index ERROR ", 255, __VA_ARGS__); }} while(0)

8
source/libs/index/inc/index_cache.h
View File

@ -38,13 +38,13 @@ typedef struct IndexCache {
// //
IndexCache *indexCacheCreate(); IndexCache *indexCacheCreate();
void indexCacheDestroy(IndexCache *cache); void indexCacheDestroy(void *cache);
int indexCachePut(IndexCache *cache, int16_t fieldId, int16_t fieldType, const char *fieldValue, int32_t fvLen, int indexCachePut(void *cache, int16_t fieldId, int16_t fieldType, const char *fieldValue, int32_t fvLen,
uint32_t version, uint64_t uid, int8_t operType); uint32_t version, uint64_t uid, int8_t operType);
int indexCacheGet(IndexCache *cache, uint64_t *rst); int indexCacheGet(void *cache, uint64_t *rst);
int indexCacheSearch(IndexCache *cache, SIndexMultiTermQuery *query, SArray *result); int indexCacheSearch(void *cache, SIndexMultiTermQuery *query, SArray *result);
#ifdef __cplusplus #ifdef __cplusplus
} }

113
source/libs/index/src/index.c
View File

@ -22,11 +22,10 @@
#endif #endif
typedef struct SIdxFieldInfo { typedef struct SIdxColInfo {
int fieldId; // generated by index internal int colId; // generated by index internal
int cVersion; int cVersion;
int type; // field type } SIdxColInfo;
} SIdxFieldInfo;
static pthread_once_t isInit = PTHREAD_ONCE_INIT; static pthread_once_t isInit = PTHREAD_ONCE_INIT;
static void indexInit(); static void indexInit();
@ -38,9 +37,10 @@ static int indexMergeCacheIntoTindex(struct SIndex *sIdx) {
indexWarn("suid %" PRIu64 " merge cache into tindex", sIdx->suid); indexWarn("suid %" PRIu64 " merge cache into tindex", sIdx->suid);
return 0; return 0;
} }
SIndex *indexOpen(SIndexOpts *opts, const char *path) { int indexOpen(SIndexOpts *opts, const char *path, SIndex **index) {
pthread_once(&isInit, indexInit); pthread_once(&isInit, indexInit);
SIndex *sIdx = calloc(1, sizeof(SIndex)); SIndex *sIdx = calloc(1, sizeof(SIndex));
if (sIdx == NULL) { return -1; }
#ifdef USE_LUCENE #ifdef USE_LUCENE
index_t *index = index_open(path); index_t *index = index_open(path);
@ -49,11 +49,13 @@ SIndex *indexOpen(SIndexOpts *opts, const char *path) {
sIdx->cache = (void*)indexCacheCreate(); sIdx->cache = (void*)indexCacheCreate();
sIdx->tindex = NULL; sIdx->tindex = NULL;
sIdx->fieldObj = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK); sIdx->colObj = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
sIdx->fieldId = 1; sIdx->colId = 1;
sIdx->cVersion = 1; sIdx->cVersion = 1;
pthread_mutex_init(&sIdx->mtx, NULL); pthread_mutex_init(&sIdx->mtx, NULL);
return sIdx;
*index = sIdx;
return 0;
} }
void indexClose(SIndex *sIdx) { void indexClose(SIndex *sIdx) {
@ -61,14 +63,17 @@ void indexClose(SIndex *sIdx) {
index_close(sIdex->index); index_close(sIdex->index);
sIdx->index = NULL; sIdx->index = NULL;
#endif #endif
#ifdef USE_INVERTED_INDEX
indexCacheDestroy(sIdx->cache); indexCacheDestroy(sIdx->cache);
taosHashCleanup(sIdx->fieldObj); taosHashCleanup(sIdx->colObj);
pthread_mutex_destroy(&sIdx->mtx); pthread_mutex_destroy(&sIdx->mtx);
#endif
free(sIdx); free(sIdx);
return; return;
} }
int indexPut(SIndex *index, SArray* fVals, int uid) { int indexPut(SIndex *index, SIndexMultiTerm * fVals, int uid) {
#ifdef USE_LUCENE #ifdef USE_LUCENE
index_document_t *doc = index_document_create(); index_document_t *doc = index_document_create();
@ -86,32 +91,38 @@ int indexPut(SIndex *index, SArray* fVals, int uid) {
index_document_destroy(doc); index_document_destroy(doc);
#endif #endif
#ifdef USE_INVERTED_INDEX
//TODO(yihao): reduce the lock range //TODO(yihao): reduce the lock range
pthread_mutex_lock(&index->mtx); pthread_mutex_lock(&index->mtx);
for (int i = 0; i < taosArrayGetSize(fVals); i++) { for (int i = 0; i < taosArrayGetSize(fVals); i++) {
SIndexTerm *p = taosArrayGetP(fVals, i); SIndexTerm *p = taosArrayGetP(fVals, i);
SIdxFieldInfo *fi = taosHashGet(index->fieldObj, p->key, p->nKey); SIdxColInfo *fi = taosHashGet(index->colObj, p->colName, p->nColName);
if (fi == NULL) { if (fi == NULL) {
SIdxFieldInfo tfi = {.fieldId = index->fieldId, .type = p->type}; SIdxColInfo tfi = {.colId = index->colId};
index->cVersion++; index->cVersion++;
index->fieldId++; index->colId++;
taosHashPut(index->fieldObj, p->key, p->nKey, &tfi, sizeof(tfi)); taosHashPut(index->colObj, p->colName, p->nColName, &tfi, sizeof(tfi));
} else { } else {
//TODO, del //TODO, del
} }
} }
pthread_mutex_unlock(&index->mtx);
for (int i = 0; i < taosArrayGetSize(fVals); i++) { for (int i = 0; i < taosArrayGetSize(fVals); i++) {
SIndexTerm *p = taosArrayGetP(fVals, i); SIndexTerm *p = taosArrayGetP(fVals, i);
SIdxFieldInfo *fi = taosHashGet(index->fieldObj, p->key, p->nKey); SIdxColInfo *fi = taosHashGet(index->colObj, p->colName, p->nColName);
assert(fi != NULL); assert(fi != NULL);
int32_t fieldId = fi->fieldId; int32_t colId = fi->colId;
int32_t colType = fi->type;
int32_t version = index->cVersion; int32_t version = index->cVersion;
int ret = indexCachePut(index->cache, colId, p->colType, p->colVal, p->nColVal, version, uid, p->operType);
if (ret != 0) {
return ret;
} }
pthread_mutex_unlock(&index->mtx); }
return 1; #endif
return 0;
} }
int indexSearch(SIndex *index, SIndexMultiTermQuery *multiQuerys, SArray *result) { int indexSearch(SIndex *index, SIndexMultiTermQuery *multiQuerys, SArray *result) {
#ifdef USE_LUCENE #ifdef USE_LUCENE
@ -148,16 +159,26 @@ int indexSearch(SIndex *index, SIndexMultiTermQuery *multiQuerys, SArray *result
free(fields); free(fields);
free(keys); free(keys);
free(types); free(types);
#endif
#ifdef USE_INVERTED_INDEX
#endif #endif
return 1; return 1;
} }
int indexDelete(SIndex *index, SIndexMultiTermQuery *query) { int indexDelete(SIndex *index, SIndexMultiTermQuery *query) {
#ifdef USE_INVERTED_INDEX
#endif
return 1; return 1;
} }
int indexRebuild(SIndex *index, SIndexOpts *opts); int indexRebuild(SIndex *index, SIndexOpts *opts) {
#ifdef USE_INVERTED_INDEX
#endif
}
SIndexOpts *indexOptsCreate() { SIndexOpts *indexOptsCreate() {
@ -184,53 +205,55 @@ SIndexMultiTermQuery *indexMultiTermQueryCreate(EIndexOperatorType opera) {
void indexMultiTermQueryDestroy(SIndexMultiTermQuery *pQuery) { void indexMultiTermQueryDestroy(SIndexMultiTermQuery *pQuery) {
for (int i = 0; i < taosArrayGetSize(pQuery->query); i++) { for (int i = 0; i < taosArrayGetSize(pQuery->query); i++) {
SIndexTermQuery *p = (SIndexTermQuery *)taosArrayGet(pQuery->query, i); SIndexTermQuery *p = (SIndexTermQuery *)taosArrayGet(pQuery->query, i);
indexTermDestroy(p->field_value); indexTermDestroy(p->term);
} }
taosArrayDestroy(pQuery->query); taosArrayDestroy(pQuery->query);
free(pQuery); free(pQuery);
}; };
int indexMultiTermQueryAdd(SIndexMultiTermQuery *pQuery, const char *field, int32_t nFields, const char *value, int32_t nValue, EIndexQueryType type){ int indexMultiTermQueryAdd(SIndexMultiTermQuery *pQuery, SIndexTerm *term, EIndexQueryType qType){
SIndexTerm *t = indexTermCreate(field, nFields, value, nValue); SIndexTermQuery q = {.qType = qType, .term = term};
if (t == NULL) {return -1;}
SIndexTermQuery q = {.type = type, .field_value = t};
taosArrayPush(pQuery->query, &q); taosArrayPush(pQuery->query, &q);
return 0; return 0;
} }
SIndexTerm *indexTermCreate(const char *key, int32_t nKey, const char *val, int32_t nVal) { SIndexTerm *indexTermCreate(int64_t suid, SIndexOperOnColumn oper, uint8_t colType, const char *colName, int32_t nColName, const char *colVal, int32_t nColVal) {
SIndexTerm *t = (SIndexTerm *)malloc(sizeof(SIndexTerm)); SIndexTerm *t = (SIndexTerm *)calloc(1, (sizeof(SIndexTerm)));
t->key = (char *)calloc(nKey + 1, 1); if (t == NULL) { return NULL; }
memcpy(t->key, key, nKey);
t->nKey = nKey;
t->val = (char *)calloc(nVal + 1, 1); t->suid = suid;
memcpy(t->val, val, nVal); t->operType= oper;
t->nVal = nVal; t->colType = colType;
t->colName = (char *)calloc(1, nColName + 1);
memcpy(t->colName, colName, nColName);
t->nColName = nColName;
t->colVal = (char *)calloc(1, nColVal + 1);
memcpy(t->colVal, colVal, nColVal);
t->nColVal = nColVal;
return t; return t;
} }
void indexTermDestroy(SIndexTerm *p) { void indexTermDestroy(SIndexTerm *p) {
free(p->key); free(p->colName);
free(p->val); free(p->colVal);
free(p); free(p);
} }
SArray *indexMultiTermCreate() { SIndexMultiTerm *indexMultiTermCreate() {
return taosArrayInit(4, sizeof(SIndexTerm *)); return taosArrayInit(4, sizeof(SIndexTerm *));
} }
int indexMultiTermAdd(SArray *array, const char *field, int32_t nField, const char *val, int32_t nVal) { int indexMultiTermAdd(SIndexMultiTerm *terms, SIndexTerm *term) {
SIndexTerm *term = indexTermCreate(field, nField, val, nVal); taosArrayPush(terms, &term);
if (term == NULL) { return -1; }
taosArrayPush(array, &term);
return 0; return 0;
} }
void indexMultiTermDestroy(SArray *array) { void indexMultiTermDestroy(SIndexMultiTerm *terms) {
for (int32_t i = 0; i < taosArrayGetSize(array); i++) { for (int32_t i = 0; i < taosArrayGetSize(terms); i++) {
SIndexTerm *p = taosArrayGetP(array, i); SIndexTerm *p = taosArrayGetP(terms, i);
indexTermDestroy(p); indexTermDestroy(p);
} }
taosArrayDestroy(array); taosArrayDestroy(terms);
} }
void indexInit() { void indexInit() {

30
source/libs/index/src/index_cache.c
View File

@ -16,7 +16,7 @@
#include "index_cache.h" #include "index_cache.h"
#include "tcompare.h" #include "tcompare.h"
#define MAX_INDEX_KEY_LEN 128 // test only, change later #define MAX_INDEX_KEY_LEN 256// test only, change later
static char* getIndexKey(const void *pData) { static char* getIndexKey(const void *pData) {
return NULL; return NULL;
@ -96,16 +96,19 @@ IndexCache *indexCacheCreate() {
} }
void indexCacheDestroy(IndexCache *cache) { void indexCacheDestroy(void *cache) {
if (cache == NULL) { return; } IndexCache *pCache = cache;
tSkipListDestroy(cache->skiplist); if (pCache == NULL) { return; }
free(cache); tSkipListDestroy(pCache->skiplist);
free(pCache);
} }
int indexCachePut(IndexCache *cache, int16_t fieldId, int16_t fieldType, const char *fieldValue, int32_t fvLen, int indexCachePut(void *cache, int16_t fieldId, int16_t fieldType, const char *fieldValue, int32_t fvLen,
uint32_t version, uint64_t uid, int8_t operType) { uint32_t version, uint64_t uid, int8_t operType) {
if (cache == NULL) { return -1;} if (cache == NULL) { return -1;}
IndexCache *pCache = cache;
// encode data // encode data
int32_t total = sizeof(int32_t) + sizeof(fieldId) + sizeof(fieldType) + sizeof(fvLen) + fvLen + sizeof(version) + sizeof(uid) + sizeof(operType); int32_t total = sizeof(int32_t) + sizeof(fieldId) + sizeof(fieldType) + sizeof(fvLen) + fvLen + sizeof(version) + sizeof(uid) + sizeof(operType);
@ -135,20 +138,15 @@ int indexCachePut(IndexCache *cache, int16_t fieldId, int16_t fieldType, const c
memcpy(p, &operType, sizeof(operType)); memcpy(p, &operType, sizeof(operType));
p += sizeof(operType); p += sizeof(operType);
tSkipListPut(cache->skiplist, (void *)buf); tSkipListPut(pCache->skiplist, (void *)buf);
// encode end // encode end
} }
int indexCacheDel(IndexCache *cache, int32_t fieldId, const char *fieldValue, int32_t fvlen, uint64_t uid, int8_t operType) { int indexCacheDel(void *cache, int32_t fieldId, const char *fieldValue, int32_t fvlen, uint64_t uid, int8_t operType) {
IndexCache *pCache = cache;
return 0;
} }
int indexCacheSearch(IndexCache *cache, SIndexMultiTermQuery *query, SArray *result) { int indexCacheSearch(void *cache, SIndexMultiTermQuery *query, SArray *result) {
return 0; return 0;
} }

2
source/libs/index/test/CMakeLists.txt
View File

@ -1,7 +1,7 @@
add_executable(indexTest "") add_executable(indexTest "")
target_sources(indexTest target_sources(indexTest
PRIVATE PRIVATE
"indexTests.cpp" "indexTests.cc"
) )
target_include_directories ( indexTest target_include_directories ( indexTest
PUBLIC PUBLIC

84
source/libs/index/test/indexTests.cpp → source/libs/index/test/indexTests.cc
View File

@ -1,3 +1,17 @@
/*
* 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/>.
*/
#include <gtest/gtest.h> #include <gtest/gtest.h>
#include <string> #include <string>
#include <iostream> #include <iostream>
@ -279,15 +293,71 @@ void validateFst() {
delete m; delete m;
} }
class IndexEnv : public ::testing::Test {
protected:
virtual void SetUp() {
taosRemoveDir(path);
opts = indexOptsCreate();
int ret = indexOpen(opts, path, &index);
assert(ret == 0);
}
virtual void TearDown() {
indexClose(index);
indexOptsDestroy(opts);
}
int main(int argc, char** argv) { const char *path = "/tmp/tindex";
checkFstPerf(); SIndexOpts *opts;
//checkFstPrefixSearch(); SIndex *index;
return 1; };
TEST_F(IndexEnv, testPut) {
// single index column
{
std::string colName("tag1"), colVal("Hello world");
SIndexTerm *term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size());
SIndexMultiTerm *terms = indexMultiTermCreate();
indexMultiTermAdd(terms, term);
for (size_t i = 0; i < 100; i++) {
int tableId = i;
int ret = indexPut(index, terms, tableId);
assert(ret == 0);
}
indexMultiTermDestroy(terms);
}
// multi index column
{
SIndexMultiTerm *terms = indexMultiTermCreate();
{
std::string colName("tag1"), colVal("Hello world");
SIndexTerm *term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size());
indexMultiTermAdd(terms, term);
}
{
std::string colName("tag2"), colVal("Hello world");
SIndexTerm *term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size());
indexMultiTermAdd(terms, term);
}
for (int i = 0; i < 100; i++) {
int tableId = i;
int ret = indexPut(index, terms, tableId);
assert(ret == 0);
}
indexMultiTermDestroy(terms);
}
//
} }
//TEST(IndexFstBuilder, IndexFstInput) { TEST_F(IndexEnv, testDel) {
//
//} }

1
source/util/src/terror.c
View File

@ -130,6 +130,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_TSC_VALUE_OUT_OF_RANGE, "Value out of range")
TAOS_DEFINE_ERROR(TSDB_CODE_TSC_INVALID_INPUT, "Invalid tsc input") TAOS_DEFINE_ERROR(TSDB_CODE_TSC_INVALID_INPUT, "Invalid tsc input")
// mnode-common // mnode-common
TAOS_DEFINE_ERROR(TSDB_CODE_MND_APP_ERROR, "Mnode internal error")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_NOT_READY, "Cluster not ready") TAOS_DEFINE_ERROR(TSDB_CODE_MND_NOT_READY, "Cluster not ready")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_MSG_NOT_PROCESSED, "Message not processed") TAOS_DEFINE_ERROR(TSDB_CODE_MND_MSG_NOT_PROCESSED, "Message not processed")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_ACTION_IN_PROGRESS, "Message is progressing") TAOS_DEFINE_ERROR(TSDB_CODE_MND_ACTION_IN_PROGRESS, "Message is progressing")