refact vnode code
This commit is contained in:
Hongze Cheng
parent
e990ff5ade
commit
c78f3e357f
|
|
@ -1,4 +1,33 @@
|
|||
add_subdirectory(meta)
|
||||
add_subdirectory(tq)
|
||||
add_subdirectory(tsdb)
|
||||
add_subdirectory(impl)
|
||||
aux_source_directory(src/meta META_SRC)
|
||||
aux_source_directory(src/tq TQ_SRC)
|
||||
aux_source_directory(src/tsdb TSDB_SRC)
|
||||
aux_source_directory(src/vnd VND_SRC)
|
||||
list(APPEND
|
||||
VNODE_SRC
|
||||
${META_SRC}
|
||||
${TQ_SRC}
|
||||
${TSDB_SRC}
|
||||
${VND_SRC}
|
||||
)
|
||||
|
||||
add_library(vnode STATIC ${VNODE_SRC})
|
||||
target_include_directories(
|
||||
vnode
|
||||
PUBLIC inc
|
||||
PRIVATE src/inc
|
||||
)
|
||||
target_link_libraries(
|
||||
vnode
|
||||
PUBLIC os
|
||||
PUBLIC util
|
||||
PUBLIC common
|
||||
PUBLIC transport
|
||||
PUBLIC bdb
|
||||
PUBLIC tfs
|
||||
PUBLIC wal
|
||||
PUBLIC qworker
|
||||
)
|
||||
|
||||
if(${BUILD_TEST})
|
||||
# add_subdirectory(test)
|
||||
endif(${BUILD_TEST})
|
||||
|
|
|
|||
|
|
@ -1,12 +0,0 @@
|
|||
# Vnode API test
|
||||
add_executable(vnodeApiTests "")
|
||||
target_sources(vnodeApiTests
|
||||
PRIVATE
|
||||
"vnodeApiTests.cpp"
|
||||
)
|
||||
target_link_libraries(vnodeApiTests vnode gtest gtest_main)
|
||||
|
||||
add_test(
|
||||
NAME vnode_api_tests
|
||||
COMMAND ${CMAKE_CURRENT_BINARY_DIR}/vnodeApiTests
|
||||
)
|
||||
|
|
@ -1,2 +0,0 @@
|
|||
// https://stackoverflow.com/questions/8565666/benchmarking-with-googletest
|
||||
// https://github.com/google/benchmark
|
||||
|
|
@ -1,285 +0,0 @@
|
|||
/**
|
||||
* @file vnodeApiTests.cpp
|
||||
* @author hzcheng (hzcheng@taosdata.com)
|
||||
* @brief VNODE module API tests
|
||||
* @version 0.1
|
||||
* @date 2021-12-13
|
||||
*
|
||||
* @copyright Copyright (c) 2021
|
||||
*
|
||||
*/
|
||||
|
||||
#include <gtest/gtest.h>
|
||||
#include <iostream>
|
||||
|
||||
#include "vnode.h"
|
||||
|
||||
static STSchema *vtCreateBasicSchema() {
|
||||
STSchemaBuilder sb;
|
||||
STSchema * pSchema = NULL;
|
||||
|
||||
tdInitTSchemaBuilder(&sb, 0);
|
||||
|
||||
tdAddColToSchema(&sb, TSDB_DATA_TYPE_TIMESTAMP, 0, 0);
|
||||
for (int i = 1; i < 10; i++) {
|
||||
tdAddColToSchema(&sb, TSDB_DATA_TYPE_INT, i, 0);
|
||||
}
|
||||
|
||||
pSchema = tdGetSchemaFromBuilder(&sb);
|
||||
|
||||
tdDestroyTSchemaBuilder(&sb);
|
||||
|
||||
return pSchema;
|
||||
}
|
||||
|
||||
static STSchema *vtCreateBasicTagSchema() {
|
||||
STSchemaBuilder sb;
|
||||
STSchema * pSchema = NULL;
|
||||
|
||||
tdInitTSchemaBuilder(&sb, 0);
|
||||
|
||||
tdAddColToSchema(&sb, TSDB_DATA_TYPE_TIMESTAMP, 0, 0);
|
||||
for (int i = 10; i < 12; i++) {
|
||||
tdAddColToSchema(&sb, TSDB_DATA_TYPE_BINARY, i, 20);
|
||||
}
|
||||
|
||||
pSchema = tdGetSchemaFromBuilder(&sb);
|
||||
|
||||
tdDestroyTSchemaBuilder(&sb);
|
||||
|
||||
return pSchema;
|
||||
}
|
||||
|
||||
static SKVRow vtCreateBasicTag() {
|
||||
SKVRowBuilder rb;
|
||||
SKVRow pTag;
|
||||
|
||||
tdInitKVRowBuilder(&rb);
|
||||
|
||||
for (int i = 0; i < 2; i++) {
|
||||
void *pVal = malloc(sizeof(VarDataLenT) + strlen("foo"));
|
||||
varDataLen(pVal) = strlen("foo");
|
||||
memcpy(varDataVal(pVal), "foo", strlen("foo"));
|
||||
|
||||
tdAddColToKVRow(&rb, i, TSDB_DATA_TYPE_BINARY, pVal);
|
||||
free(pVal);
|
||||
}
|
||||
|
||||
pTag = tdGetKVRowFromBuilder(&rb);
|
||||
tdDestroyKVRowBuilder(&rb);
|
||||
|
||||
return pTag;
|
||||
}
|
||||
|
||||
static void vtBuildCreateStbReq(tb_uid_t suid, char *tbname, SRpcMsg **ppMsg) {
|
||||
SRpcMsg * pMsg;
|
||||
STSchema *pSchema;
|
||||
STSchema *pTagSchema;
|
||||
int zs;
|
||||
void * pBuf;
|
||||
|
||||
pSchema = vtCreateBasicSchema();
|
||||
pTagSchema = vtCreateBasicTagSchema();
|
||||
|
||||
SVnodeReq vCreateSTbReq;
|
||||
vnodeSetCreateStbReq(&vCreateSTbReq, tbname, UINT32_MAX, UINT32_MAX, suid, pSchema, pTagSchema);
|
||||
|
||||
zs = vnodeBuildReq(NULL, &vCreateSTbReq, TDMT_VND_CREATE_STB);
|
||||
pMsg = (SRpcMsg *)malloc(sizeof(SRpcMsg) + zs);
|
||||
pMsg->msgType = TDMT_VND_CREATE_STB;
|
||||
pMsg->contLen = zs;
|
||||
pMsg->pCont = POINTER_SHIFT(pMsg, sizeof(SRpcMsg));
|
||||
|
||||
pBuf = pMsg->pCont;
|
||||
vnodeBuildReq(&pBuf, &vCreateSTbReq, TDMT_VND_CREATE_STB);
|
||||
META_CLEAR_TB_CFG(&vCreateSTbReq);
|
||||
|
||||
tdFreeSchema(pSchema);
|
||||
tdFreeSchema(pTagSchema);
|
||||
|
||||
*ppMsg = pMsg;
|
||||
}
|
||||
|
||||
static void vtBuildCreateCtbReq(tb_uid_t suid, char *tbname, SRpcMsg **ppMsg) {
|
||||
SRpcMsg *pMsg;
|
||||
int tz;
|
||||
SKVRow pTag = vtCreateBasicTag();
|
||||
|
||||
SVnodeReq vCreateCTbReq;
|
||||
vnodeSetCreateCtbReq(&vCreateCTbReq, tbname, UINT32_MAX, UINT32_MAX, suid, pTag);
|
||||
|
||||
tz = vnodeBuildReq(NULL, &vCreateCTbReq, TDMT_VND_CREATE_TABLE);
|
||||
pMsg = (SRpcMsg *)malloc(sizeof(SRpcMsg) + tz);
|
||||
pMsg->msgType = TDMT_VND_CREATE_TABLE;
|
||||
pMsg->contLen = tz;
|
||||
pMsg->pCont = POINTER_SHIFT(pMsg, sizeof(*pMsg));
|
||||
void *pBuf = pMsg->pCont;
|
||||
|
||||
vnodeBuildReq(&pBuf, &vCreateCTbReq, TDMT_VND_CREATE_TABLE);
|
||||
META_CLEAR_TB_CFG(&vCreateCTbReq);
|
||||
free(pTag);
|
||||
|
||||
*ppMsg = pMsg;
|
||||
}
|
||||
|
||||
static void vtBuildCreateNtbReq(char *tbname, SRpcMsg **ppMsg) {
|
||||
// TODO
|
||||
}
|
||||
|
||||
static void vtBuildSubmitReq(SRpcMsg **ppMsg) {
|
||||
SRpcMsg * pMsg;
|
||||
SSubmitMsg *pSubmitMsg;
|
||||
SSubmitBlk *pSubmitBlk;
|
||||
int tz = 1024; // TODO
|
||||
|
||||
pMsg = (SRpcMsg *)malloc(sizeof(*pMsg) + tz);
|
||||
pMsg->msgType = TDMT_VND_SUBMIT;
|
||||
pMsg->contLen = tz;
|
||||
pMsg->pCont = POINTER_SHIFT(pMsg, sizeof(*pMsg));
|
||||
|
||||
// For submit msg header
|
||||
pSubmitMsg = (SSubmitMsg *)(pMsg->pCont);
|
||||
// pSubmitMsg->header.contLen = 0;
|
||||
// pSubmitMsg->header.vgId = 0;
|
||||
// pSubmitMsg->length = 0;
|
||||
pSubmitMsg->numOfBlocks = 1;
|
||||
|
||||
// For submit blk
|
||||
pSubmitBlk = (SSubmitBlk *)(pSubmitMsg->blocks);
|
||||
pSubmitBlk->uid = 0;
|
||||
pSubmitBlk->tid = 0;
|
||||
pSubmitBlk->padding = 0;
|
||||
pSubmitBlk->sversion = 0;
|
||||
pSubmitBlk->dataLen = 0;
|
||||
pSubmitBlk->numOfRows = 0;
|
||||
|
||||
// For row batch
|
||||
|
||||
*ppMsg = pMsg;
|
||||
}
|
||||
|
||||
static void vtClearMsgBatch(SArray *pMsgArr) {
|
||||
SRpcMsg *pMsg;
|
||||
for (size_t i = 0; i < taosArrayGetSize(pMsgArr); i++) {
|
||||
pMsg = *(SRpcMsg **)taosArrayGet(pMsgArr, i);
|
||||
free(pMsg);
|
||||
}
|
||||
|
||||
taosArrayClear(pMsgArr);
|
||||
}
|
||||
|
||||
static void vtProcessAndApplyReqs(SVnode *pVnode, SArray *pMsgArr) {
|
||||
int rcode;
|
||||
SRpcMsg *pReq;
|
||||
SRpcMsg *pRsp;
|
||||
|
||||
rcode = vnodeProcessWMsgs(pVnode, pMsgArr);
|
||||
GTEST_ASSERT_EQ(rcode, 0);
|
||||
|
||||
for (size_t i = 0; i < taosArrayGetSize(pMsgArr); i++) {
|
||||
pReq = *(SRpcMsg **)taosArrayGet(pMsgArr, i);
|
||||
rcode = vnodeApplyWMsg(pVnode, pReq, NULL);
|
||||
GTEST_ASSERT_EQ(rcode, 0);
|
||||
}
|
||||
}
|
||||
|
||||
TEST(vnodeApiTest, vnode_simple_create_table_test) {
|
||||
tb_uid_t suid = 1638166374163;
|
||||
SRpcMsg *pMsg;
|
||||
SArray * pMsgArr = NULL;
|
||||
SVnode * pVnode;
|
||||
int rcode;
|
||||
int ntables = 1000000;
|
||||
int batch = 10;
|
||||
char tbname[128];
|
||||
|
||||
pMsgArr = (SArray *)taosArrayInit(batch, sizeof(pMsg));
|
||||
|
||||
vnodeDestroy("vnode1");
|
||||
GTEST_ASSERT_GE(vnodeInit(2), 0);
|
||||
|
||||
// CREATE AND OPEN A VNODE
|
||||
pVnode = vnodeOpen("vnode1", NULL);
|
||||
ASSERT_NE(pVnode, nullptr);
|
||||
|
||||
// CREATE A SUPER TABLE
|
||||
sprintf(tbname, "st");
|
||||
vtBuildCreateStbReq(suid, tbname, &pMsg);
|
||||
taosArrayPush(pMsgArr, &pMsg);
|
||||
vtProcessAndApplyReqs(pVnode, pMsgArr);
|
||||
vtClearMsgBatch(pMsgArr);
|
||||
|
||||
// CREATE A LOT OF CHILD TABLES
|
||||
for (int i = 0; i < ntables / batch; i++) {
|
||||
// Build request batch
|
||||
for (int j = 0; j < batch; j++) {
|
||||
sprintf(tbname, "ct%d", i * batch + j + 1);
|
||||
vtBuildCreateCtbReq(suid, tbname, &pMsg);
|
||||
taosArrayPush(pMsgArr, &pMsg);
|
||||
}
|
||||
|
||||
// Process request batch
|
||||
vtProcessAndApplyReqs(pVnode, pMsgArr);
|
||||
|
||||
// Clear request batch
|
||||
vtClearMsgBatch(pMsgArr);
|
||||
}
|
||||
|
||||
// CLOSE THE VNODE
|
||||
vnodeClose(pVnode);
|
||||
vnodeCleanup();
|
||||
|
||||
taosArrayDestroy(pMsgArr);
|
||||
}
|
||||
|
||||
TEST(vnodeApiTest, vnode_simple_insert_test) {
|
||||
const char *vname = "vnode2";
|
||||
char tbname[128];
|
||||
tb_uid_t suid = 1638166374163;
|
||||
SRpcMsg * pMsg;
|
||||
SArray * pMsgArr;
|
||||
int rcode;
|
||||
SVnode * pVnode;
|
||||
int batch = 1;
|
||||
int loop = 1000000;
|
||||
|
||||
pMsgArr = (SArray *)taosArrayInit(0, sizeof(pMsg));
|
||||
|
||||
vnodeDestroy(vname);
|
||||
|
||||
GTEST_ASSERT_GE(vnodeInit(2), 0);
|
||||
|
||||
// Open a vnode
|
||||
pVnode = vnodeOpen(vname, NULL);
|
||||
GTEST_ASSERT_NE(pVnode, nullptr);
|
||||
|
||||
// 1. CREATE A SUPER TABLE
|
||||
sprintf(tbname, "st");
|
||||
vtBuildCreateStbReq(suid, tbname, &pMsg);
|
||||
taosArrayPush(pMsgArr, &pMsg);
|
||||
vtProcessAndApplyReqs(pVnode, pMsgArr);
|
||||
vtClearMsgBatch(pMsgArr);
|
||||
|
||||
// 2. CREATE A CHILD TABLE
|
||||
sprintf(tbname, "t0");
|
||||
vtBuildCreateCtbReq(suid, tbname, &pMsg);
|
||||
taosArrayPush(pMsgArr, &pMsg);
|
||||
vtProcessAndApplyReqs(pVnode, pMsgArr);
|
||||
vtClearMsgBatch(pMsgArr);
|
||||
|
||||
// 3. WRITE A LOT OF TIME-SERIES DATA
|
||||
for (int j = 0; j < loop; j++) {
|
||||
for (int i = 0; i < batch; i++) {
|
||||
vtBuildSubmitReq(&pMsg);
|
||||
taosArrayPush(pMsgArr, &pMsg);
|
||||
}
|
||||
vtProcessAndApplyReqs(pVnode, pMsgArr);
|
||||
vtClearMsgBatch(pMsgArr);
|
||||
}
|
||||
|
||||
// Close the vnode
|
||||
vnodeClose(pVnode);
|
||||
vnodeCleanup();
|
||||
|
||||
taosArrayDestroy(pMsgArr);
|
||||
}
|
||||
|
|
@ -1,212 +0,0 @@
|
|||
/*
|
||||
* 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 "metaDef.h"
|
||||
#include "sqlite3.h"
|
||||
|
||||
struct SMetaDB {
|
||||
sqlite3 *pDB;
|
||||
};
|
||||
|
||||
int metaOpenDB(SMeta *pMeta) {
|
||||
char dir[128];
|
||||
int rc;
|
||||
char *err = NULL;
|
||||
|
||||
pMeta->pDB = (SMetaDB *)calloc(1, sizeof(SMetaDB));
|
||||
if (pMeta->pDB == NULL) {
|
||||
// TODO: handle error
|
||||
return -1;
|
||||
}
|
||||
|
||||
sprintf(dir, "%s/meta.db", pMeta->path);
|
||||
rc = sqlite3_open(dir, &(pMeta->pDB->pDB));
|
||||
if (rc != SQLITE_OK) {
|
||||
// TODO: handle error
|
||||
printf("failed to open meta.db\n");
|
||||
}
|
||||
|
||||
// For all tables
|
||||
rc = sqlite3_exec(pMeta->pDB->pDB,
|
||||
"CREATE TABLE IF NOT EXISTS tb ("
|
||||
" tbname VARCHAR(256) NOT NULL UNIQUE,"
|
||||
" tb_uid INTEGER NOT NULL UNIQUE "
|
||||
");",
|
||||
NULL, NULL, &err);
|
||||
if (rc != SQLITE_OK) {
|
||||
// TODO: handle error
|
||||
printf("failed to create meta table tb since %s\n", err);
|
||||
}
|
||||
|
||||
// For super tables
|
||||
rc = sqlite3_exec(pMeta->pDB->pDB,
|
||||
"CREATE TABLE IF NOT EXISTS stb ("
|
||||
" tb_uid INTEGER NOT NULL UNIQUE,"
|
||||
" tbname VARCHAR(256) NOT NULL UNIQUE,"
|
||||
" tb_schema BLOB NOT NULL,"
|
||||
" tag_schema BLOB NOT NULL"
|
||||
");",
|
||||
NULL, NULL, &err);
|
||||
if (rc != SQLITE_OK) {
|
||||
// TODO: handle error
|
||||
printf("failed to create meta table stb since %s\n", err);
|
||||
}
|
||||
|
||||
// For normal tables
|
||||
rc = sqlite3_exec(pMeta->pDB->pDB,
|
||||
"CREATE TABLE IF NOT EXISTS ntb ("
|
||||
" tb_uid INTEGER NOT NULL UNIQUE,"
|
||||
" tbname VARCHAR(256) NOT NULL,"
|
||||
" tb_schema BLOB NOT NULL"
|
||||
");",
|
||||
NULL, NULL, &err);
|
||||
if (rc != SQLITE_OK) {
|
||||
// TODO: handle error
|
||||
printf("failed to create meta table ntb since %s\n", err);
|
||||
}
|
||||
|
||||
sqlite3_exec(pMeta->pDB->pDB, "BEGIN;", NULL, NULL, &err);
|
||||
|
||||
tfree(err);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
void metaCloseDB(SMeta *pMeta) {
|
||||
if (pMeta->pDB) {
|
||||
sqlite3_exec(pMeta->pDB->pDB, "COMMIT;", NULL, NULL, NULL);
|
||||
sqlite3_close(pMeta->pDB->pDB);
|
||||
free(pMeta->pDB);
|
||||
pMeta->pDB = NULL;
|
||||
}
|
||||
|
||||
// TODO
|
||||
}
|
||||
|
||||
int metaSaveTableToDB(SMeta *pMeta, const STbCfg *pTbCfg) {
|
||||
char sql[256];
|
||||
char * err = NULL;
|
||||
int rc;
|
||||
tb_uid_t uid;
|
||||
sqlite3_stmt *stmt;
|
||||
char buf[256];
|
||||
void * pBuf;
|
||||
|
||||
switch (pTbCfg->type) {
|
||||
case META_SUPER_TABLE:
|
||||
uid = pTbCfg->stbCfg.suid;
|
||||
sprintf(sql,
|
||||
"INSERT INTO tb VALUES (\'%s\', %" PRIu64
|
||||
");"
|
||||
"CREATE TABLE IF NOT EXISTS stb_%" PRIu64
|
||||
" ("
|
||||
" tb_uid INTEGER NOT NULL UNIQUE,"
|
||||
" tbname VARCHAR(256),"
|
||||
" tag1 INTEGER);",
|
||||
pTbCfg->name, uid, uid);
|
||||
rc = sqlite3_exec(pMeta->pDB->pDB, sql, NULL, NULL, &err);
|
||||
if (rc != SQLITE_OK) {
|
||||
printf("failed to create normal table since %s\n", err);
|
||||
}
|
||||
|
||||
sprintf(sql, "INSERT INTO stb VALUES (%" PRIu64 ", %s, ?, ?)", uid, pTbCfg->name);
|
||||
sqlite3_prepare_v2(pMeta->pDB->pDB, sql, -1, &stmt, NULL);
|
||||
|
||||
pBuf = buf;
|
||||
tdEncodeSchema(&pBuf, pTbCfg->stbCfg.pSchema);
|
||||
sqlite3_bind_blob(stmt, 1, buf, POINTER_DISTANCE(pBuf, buf), NULL);
|
||||
pBuf = buf;
|
||||
tdEncodeSchema(&pBuf, pTbCfg->stbCfg.pTagSchema);
|
||||
sqlite3_bind_blob(stmt, 2, buf, POINTER_DISTANCE(pBuf, buf), NULL);
|
||||
|
||||
sqlite3_step(stmt);
|
||||
|
||||
sqlite3_finalize(stmt);
|
||||
|
||||
#if 0
|
||||
sprintf(sql,
|
||||
"INSERT INTO tb VALUES (?, ?);"
|
||||
// "INSERT INTO stb VALUES (?, ?, ?, ?);"
|
||||
// "CREATE TABLE IF NOT EXISTS stb_%" PRIu64
|
||||
// " ("
|
||||
// " tb_uid INTEGER NOT NULL UNIQUE,"
|
||||
// " tbname VARCHAR(256),"
|
||||
// " tag1 INTEGER);"
|
||||
,
|
||||
uid);
|
||||
rc = sqlite3_prepare_v2(pMeta->pDB->pDB, sql, -1, &stmt, NULL);
|
||||
if (rc != SQLITE_OK) {
|
||||
return -1;
|
||||
}
|
||||
sqlite3_bind_text(stmt, 1, pTbCfg->name, -1, SQLITE_TRANSIENT);
|
||||
sqlite3_bind_int64(stmt, 2, uid);
|
||||
sqlite3_step(stmt);
|
||||
sqlite3_finalize(stmt);
|
||||
|
||||
|
||||
// sqlite3_bind_int64(stmt, 3, uid);
|
||||
// sqlite3_bind_text(stmt, 4, pTbCfg->name, -1, SQLITE_TRANSIENT);
|
||||
// pBuf = buf;
|
||||
// tdEncodeSchema(&pBuf, pTbCfg->stbCfg.pSchema);
|
||||
// sqlite3_bind_blob(stmt, 5, buf, POINTER_DISTANCE(pBuf, buf), NULL);
|
||||
// pBuf = buf;
|
||||
// tdEncodeSchema(&pBuf, pTbCfg->stbCfg.pTagSchema);
|
||||
// sqlite3_bind_blob(stmt, 6, buf, POINTER_DISTANCE(pBuf, buf), NULL);
|
||||
|
||||
rc = sqliteVjj3_step(stmt);
|
||||
if (rc != SQLITE_OK) {
|
||||
printf("failed to create normal table since %s\n", sqlite3_errmsg(pMeta->pDB->pDB));
|
||||
}
|
||||
sqlite3_finalize(stmt);
|
||||
#endif
|
||||
break;
|
||||
case META_NORMAL_TABLE:
|
||||
// uid = metaGenerateUid(pMeta);
|
||||
// sprintf(sql,
|
||||
// "INSERT INTO tb VALUES (\'%s\', %" PRIu64
|
||||
// ");"
|
||||
// "INSERT INTO ntb VALUES (%" PRIu64 ", \'%s\', );",
|
||||
// pTbCfg->name, uid, uid, pTbCfg->name, );
|
||||
|
||||
// rc = sqlite3_exec(pMeta->pDB->pDB, sql, NULL, NULL, &err);
|
||||
// if (rc != SQLITE_OK) {
|
||||
// printf("failed to create normal table since %s\n", err);
|
||||
// }
|
||||
break;
|
||||
case META_CHILD_TABLE:
|
||||
#if 0
|
||||
uid = metaGenerateUid(pMeta);
|
||||
// sprintf(sql, "INSERT INTO tb VALUES (\'%s\', %" PRIu64
|
||||
// ");"
|
||||
// "INSERT INTO stb_%" PRIu64 " VALUES (%" PRIu64 ", \'%s\', );");
|
||||
rc = sqlite3_exec(pMeta->pDB->pDB, sql, NULL, NULL, &err);
|
||||
if (rc != SQLITE_OK) {
|
||||
printf("failed to create child table since %s\n", err);
|
||||
}
|
||||
#endif
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
|
||||
tfree(err);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
int metaRemoveTableFromDb(SMeta *pMeta, tb_uid_t uid) {
|
||||
/* TODO */
|
||||
return 0;
|
||||
}
|
||||
|
|
@ -1,24 +0,0 @@
|
|||
# add_executable(metaTest "")
|
||||
# target_sources(metaTest
|
||||
# PRIVATE
|
||||
# "../src/metaMain.c"
|
||||
# "../src/metaUid.c"
|
||||
# "metaTests.cpp"
|
||||
# )
|
||||
# target_include_directories(metaTest
|
||||
# PUBLIC
|
||||
# "${CMAKE_SOURCE_DIR}/include/server/vnode/meta"
|
||||
# "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
|
||||
# )
|
||||
# target_link_libraries(metaTest
|
||||
# os
|
||||
# util
|
||||
# common
|
||||
# gtest_main
|
||||
# tkv
|
||||
# )
|
||||
# enable_testing()
|
||||
# add_test(
|
||||
# NAME meta_test
|
||||
# COMMAND metaTest
|
||||
# )
|
||||
|
|
@ -1,105 +0,0 @@
|
|||
#if 0
|
||||
#include <gtest/gtest.h>
|
||||
#include <string.h>
|
||||
#include <iostream>
|
||||
|
||||
#include "meta.h"
|
||||
|
||||
static STSchema *metaGetSimpleSchema() {
|
||||
STSchema * pSchema = NULL;
|
||||
STSchemaBuilder sb = {0};
|
||||
|
||||
tdInitTSchemaBuilder(&sb, 0);
|
||||
tdAddColToSchema(&sb, TSDB_DATA_TYPE_TIMESTAMP, 0, 8);
|
||||
tdAddColToSchema(&sb, TSDB_DATA_TYPE_INT, 1, 4);
|
||||
|
||||
pSchema = tdGetSchemaFromBuilder(&sb);
|
||||
tdDestroyTSchemaBuilder(&sb);
|
||||
|
||||
return pSchema;
|
||||
}
|
||||
|
||||
static SKVRow metaGetSimpleTags() {
|
||||
SKVRowBuilder kvrb = {0};
|
||||
SKVRow row;
|
||||
|
||||
tdInitKVRowBuilder(&kvrb);
|
||||
int64_t ts = 1634287978000;
|
||||
int32_t a = 10;
|
||||
|
||||
tdAddColToKVRow(&kvrb, 0, TSDB_DATA_TYPE_TIMESTAMP, (void *)(&ts));
|
||||
tdAddColToKVRow(&kvrb, 0, TSDB_DATA_TYPE_INT, (void *)(&a));
|
||||
|
||||
row = tdGetKVRowFromBuilder(&kvrb);
|
||||
|
||||
tdDestroyKVRowBuilder(&kvrb);
|
||||
|
||||
return row;
|
||||
}
|
||||
|
||||
TEST(MetaTest, DISABLED_meta_create_1m_normal_tables_test) {
|
||||
// Open Meta
|
||||
SMeta *meta = metaOpen(NULL, NULL);
|
||||
std::cout << "Meta is opened!" << std::endl;
|
||||
|
||||
// Create 1000000 normal tables
|
||||
META_TABLE_OPTS_DECLARE(tbOpts);
|
||||
STSchema *pSchema = metaGetSimpleSchema();
|
||||
char tbname[128];
|
||||
|
||||
for (size_t i = 0; i < 1000000; i++) {
|
||||
sprintf(tbname, "ntb%ld", i);
|
||||
metaNormalTableOptsInit(&tbOpts, tbname, pSchema);
|
||||
metaCreateTable(meta, &tbOpts);
|
||||
metaTableOptsClear(&tbOpts);
|
||||
}
|
||||
|
||||
tdFreeSchema(pSchema);
|
||||
|
||||
// Close Meta
|
||||
metaClose(meta);
|
||||
std::cout << "Meta is closed!" << std::endl;
|
||||
|
||||
// Destroy Meta
|
||||
metaDestroy("meta");
|
||||
std::cout << "Meta is destroyed!" << std::endl;
|
||||
}
|
||||
|
||||
TEST(MetaTest, meta_create_1m_child_tables_test) {
|
||||
// Open Meta
|
||||
SMeta *meta = metaOpen(NULL);
|
||||
std::cout << "Meta is opened!" << std::endl;
|
||||
|
||||
// Create a super tables
|
||||
tb_uid_t uid = 477529885843758ul;
|
||||
META_TABLE_OPTS_DECLARE(tbOpts);
|
||||
STSchema *pSchema = metaGetSimpleSchema();
|
||||
STSchema *pTagSchema = metaGetSimpleSchema();
|
||||
|
||||
metaSuperTableOptsInit(&tbOpts, "st", uid, pSchema, pTagSchema);
|
||||
metaCreateTable(meta, &tbOpts);
|
||||
metaTableOptsClear(&tbOpts);
|
||||
|
||||
tdFreeSchema(pSchema);
|
||||
tdFreeSchema(pTagSchema);
|
||||
|
||||
// Create 1000000 child tables
|
||||
char name[128];
|
||||
SKVRow row = metaGetSimpleTags();
|
||||
for (size_t i = 0; i < 1000000; i++) {
|
||||
sprintf(name, "ctb%ld", i);
|
||||
metaChildTableOptsInit(&tbOpts, name, uid, row);
|
||||
metaCreateTable(meta, &tbOpts);
|
||||
metaTableOptsClear(&tbOpts);
|
||||
}
|
||||
kvRowFree(row);
|
||||
|
||||
// Close Meta
|
||||
metaClose(meta);
|
||||
std::cout << "Meta is closed!" << std::endl;
|
||||
|
||||
// Destroy Meta
|
||||
metaDestroy("meta");
|
||||
std::cout << "Meta is destroyed!" << std::endl;
|
||||
}
|
||||
#endif
|
||||
|
|
@ -17,7 +17,7 @@
|
|||
#define _TD_VNODE_DEF_H_
|
||||
|
||||
#include "mallocator.h"
|
||||
#include "sync.h"
|
||||
// #include "sync.h"
|
||||
#include "tcoding.h"
|
||||
#include "tlist.h"
|
||||
#include "tlockfree.h"
|
||||
|
|
@ -19,7 +19,7 @@
|
|||
#include "vnode.h"
|
||||
|
||||
#include "meta.h"
|
||||
#include "sync.h"
|
||||
// #include "sync.h"
|
||||
#include "tlog.h"
|
||||
#include "tq.h"
|
||||
#include "tsdb.h"
|
||||
|
|
@ -16,7 +16,7 @@
|
|||
#ifndef _TD_VNODE_SYNC_H_
|
||||
#define _TD_VNODE_SYNC_H_
|
||||
|
||||
#include "sync.h"
|
||||
// #include "sync.h"
|
||||
|
||||
#ifdef __cplusplus
|
||||
extern "C" {
|
||||
|
|
@ -13,7 +13,9 @@
|
|||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
#ifdef USE_INVERTED_INDEX
|
||||
#include "index.h"
|
||||
#endif
|
||||
#include "metaDef.h"
|
||||
|
||||
struct SMetaIdx {
|
||||
|
|
@ -0,0 +1,98 @@
|
|||
/*
|
||||
* 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 "os.h"
|
||||
// #include "tmsg.h"
|
||||
// #include "tarray.h"
|
||||
// #include "query.h"
|
||||
// #include "tglobal.h"
|
||||
// #include "tlist.h"
|
||||
// #include "tsdbint.h"
|
||||
// #include "tsdbBuffer.h"
|
||||
// #include "tsdbLog.h"
|
||||
// #include "tsdbHealth.h"
|
||||
// #include "ttimer.h"
|
||||
// #include "tthread.h"
|
||||
|
||||
|
||||
// // return malloc new block count
|
||||
// int32_t tsdbInsertNewBlock(STsdbRepo * pRepo) {
|
||||
// STsdbBufPool *pPool = pRepo->pPool;
|
||||
// int32_t cnt = 0;
|
||||
|
||||
// if(tsdbAllowNewBlock(pRepo)) {
|
||||
// STsdbBufBlock *pBufBlock = tsdbNewBufBlock(pPool->bufBlockSize);
|
||||
// if (pBufBlock) {
|
||||
// if (tdListAppend(pPool->bufBlockList, (void *)(&pBufBlock)) < 0) {
|
||||
// // append error
|
||||
// tsdbFreeBufBlock(pBufBlock);
|
||||
// } else {
|
||||
// pPool->nElasticBlocks ++;
|
||||
// cnt ++ ;
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
// return cnt;
|
||||
// }
|
||||
|
||||
// // switch anther thread to run
|
||||
// void* cbKillQueryFree(void* param) {
|
||||
// STsdbRepo* pRepo = (STsdbRepo*)param;
|
||||
// // vnode
|
||||
// if(pRepo->appH.notifyStatus) {
|
||||
// pRepo->appH.notifyStatus(pRepo->appH.appH, TSDB_STATUS_COMMIT_NOBLOCK, TSDB_CODE_SUCCESS);
|
||||
// }
|
||||
|
||||
// // free
|
||||
// if(pRepo->pthread){
|
||||
// void* p = pRepo->pthread;
|
||||
// pRepo->pthread = NULL;
|
||||
// free(p);
|
||||
// }
|
||||
|
||||
// return NULL;
|
||||
// }
|
||||
|
||||
// // return true do free , false do nothing
|
||||
// bool tsdbUrgeQueryFree(STsdbRepo * pRepo) {
|
||||
// // check previous running
|
||||
// if(pRepo->pthread && taosThreadRunning(pRepo->pthread)) {
|
||||
// tsdbWarn("vgId:%d pre urge thread is runing. nBlocks=%d nElasticBlocks=%d", REPO_ID(pRepo), pRepo->pPool->nBufBlocks, pRepo->pPool->nElasticBlocks);
|
||||
// return false;
|
||||
// }
|
||||
// // create new
|
||||
// pRepo->pthread = taosCreateThread(cbKillQueryFree, pRepo);
|
||||
// if(pRepo->pthread == NULL) {
|
||||
// tsdbError("vgId:%d create urge thread error.", REPO_ID(pRepo));
|
||||
// return false;
|
||||
// }
|
||||
// return true;
|
||||
// }
|
||||
|
||||
// bool tsdbAllowNewBlock(STsdbRepo* pRepo) {
|
||||
// int32_t nMaxElastic = pRepo->config.totalBlocks/3;
|
||||
// STsdbBufPool* pPool = pRepo->pPool;
|
||||
// if(pPool->nElasticBlocks >= nMaxElastic) {
|
||||
// tsdbWarn("vgId:%d tsdbAllowNewBlock return fasle. nElasticBlock(%d) >= MaxElasticBlocks(%d)", REPO_ID(pRepo), pPool->nElasticBlocks, nMaxElastic);
|
||||
// return false;
|
||||
// }
|
||||
// return true;
|
||||
// }
|
||||
|
||||
// bool tsdbNoProblem(STsdbRepo* pRepo) {
|
||||
// if(listNEles(pRepo->pPool->bufBlockList) == 0)
|
||||
// return false;
|
||||
// return true;
|
||||
// }
|
||||
|
|
@ -28,7 +28,7 @@
|
|||
|
||||
#include "taosdef.h"
|
||||
#include "tlosertree.h"
|
||||
#include "tsdbint.h"
|
||||
#include "tsdbDef.h"
|
||||
#include "tmsg.h"
|
||||
|
||||
#define EXTRA_BYTES 2
|
||||
|
|
@ -13,18 +13,18 @@
|
|||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
#include "tsdbRowMergeBuf.h"
|
||||
#include "tdataformat.h"
|
||||
// #include "tsdbRowMergeBuf.h"
|
||||
// #include "tdataformat.h"
|
||||
|
||||
// row1 has higher priority
|
||||
SMemRow tsdbMergeTwoRows(SMergeBuf *pBuf, SMemRow row1, SMemRow row2, STSchema *pSchema1, STSchema *pSchema2) {
|
||||
if(row2 == NULL) return row1;
|
||||
if(row1 == NULL) return row2;
|
||||
ASSERT(pSchema1->version == memRowVersion(row1));
|
||||
ASSERT(pSchema2->version == memRowVersion(row2));
|
||||
// // row1 has higher priority
|
||||
// SMemRow tsdbMergeTwoRows(SMergeBuf *pBuf, SMemRow row1, SMemRow row2, STSchema *pSchema1, STSchema *pSchema2) {
|
||||
// if(row2 == NULL) return row1;
|
||||
// if(row1 == NULL) return row2;
|
||||
// ASSERT(pSchema1->version == memRowVersion(row1));
|
||||
// ASSERT(pSchema2->version == memRowVersion(row2));
|
||||
|
||||
if(tsdbMergeBufMakeSureRoom(pBuf, pSchema1, pSchema2) < 0) {
|
||||
return NULL;
|
||||
}
|
||||
return mergeTwoMemRows(*pBuf, row1, row2, pSchema1, pSchema2);
|
||||
}
|
||||
// if(tsdbMergeBufMakeSureRoom(pBuf, pSchema1, pSchema2) < 0) {
|
||||
// return NULL;
|
||||
// }
|
||||
// return mergeTwoMemRows(*pBuf, row1, row2, pSchema1, pSchema2);
|
||||
// }
|
||||
|
|
@ -13,9 +13,9 @@
|
|||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
#include "tsdbint.h"
|
||||
|
||||
#if 0
|
||||
#include "tsdbint.h"
|
||||
#ifndef _TSDB_PLUGINS
|
||||
|
||||
int tsdbScanFGroup(STsdbScanHandle* pScanHandle, char* rootDir, int fid) { return 0; }
|
||||
|
|
@ -1,149 +0,0 @@
|
|||
/*
|
||||
* 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/>.
|
||||
*/
|
||||
|
||||
#ifndef _TD_TSDB_INT_H_
|
||||
#define _TD_TSDB_INT_H_
|
||||
|
||||
#if 0
|
||||
// // TODO: remove the include
|
||||
// #include <errno.h>
|
||||
// #include <fcntl.h>
|
||||
// #include <limits.h>
|
||||
// #include <inttypes.h>
|
||||
// #include <sys/stat.h>
|
||||
// #include <sys/types.h>
|
||||
// #include <semaphore.h>
|
||||
// #include <dirent.h>
|
||||
|
||||
#include "hash.h"
|
||||
#include "os.h"
|
||||
#include "taosdef.h"
|
||||
#include "taoserror.h"
|
||||
#include "tarray.h"
|
||||
#include "tchecksum.h"
|
||||
#include "tcoding.h"
|
||||
#include "tcompression.h"
|
||||
#include "tdataformat.h"
|
||||
#include "tfs.h"
|
||||
#include "tlist.h"
|
||||
#include "tlockfree.h"
|
||||
#include "tlog.h"
|
||||
#include "tskiplist.h"
|
||||
#include "tsocket.h"
|
||||
|
||||
#include "tsdb.h"
|
||||
|
||||
#ifdef __cplusplus
|
||||
extern "C" {
|
||||
#endif
|
||||
|
||||
// Log
|
||||
#include "tsdbLog.h"
|
||||
// Meta
|
||||
#include "tsdbMeta.h"
|
||||
// Buffer
|
||||
#include "tsdbBuffer.h"
|
||||
// MemTable
|
||||
#include "tsdbMemTable.h"
|
||||
// File
|
||||
#include "tsdbFile.h"
|
||||
// FS
|
||||
#include "tsdbFS.h"
|
||||
// ReadImpl
|
||||
#include "tsdbReadImpl.h"
|
||||
// Commit
|
||||
#include "tsdbCommit.h"
|
||||
// Compact
|
||||
#include "tsdbCompact.h"
|
||||
// Commit Queue
|
||||
#include "tsdbCommitQueue.h"
|
||||
|
||||
#include "tsdbRowMergeBuf.h"
|
||||
// Main definitions
|
||||
struct STsdbRepo {
|
||||
uint8_t state;
|
||||
|
||||
STsdbCfg config;
|
||||
|
||||
STsdbCfg save_config; // save apply config
|
||||
bool config_changed; // config changed flag
|
||||
pthread_mutex_t save_mutex; // protect save config
|
||||
|
||||
uint8_t hasCachedLastColumn;
|
||||
|
||||
STsdbAppH appH;
|
||||
STsdbStat stat;
|
||||
STsdbMeta* tsdbMeta;
|
||||
STsdbBufPool* pPool;
|
||||
SMemTable* mem;
|
||||
SMemTable* imem;
|
||||
STsdbFS* fs;
|
||||
SRtn rtn;
|
||||
tsem_t readyToCommit;
|
||||
pthread_mutex_t mutex;
|
||||
bool repoLocked;
|
||||
int32_t code; // Commit code
|
||||
|
||||
SMergeBuf mergeBuf; //used when update=2
|
||||
int8_t compactState; // compact state: inCompact/noCompact/waitingCompact?
|
||||
pthread_t* pthread;
|
||||
};
|
||||
|
||||
#define REPO_ID(r) (r)->config.tsdbId
|
||||
#define REPO_CFG(r) (&((r)->config))
|
||||
#define REPO_FS(r) ((r)->fs)
|
||||
#define IS_REPO_LOCKED(r) (r)->repoLocked
|
||||
#define TSDB_SUBMIT_MSG_HEAD_SIZE sizeof(SSubmitMsg)
|
||||
|
||||
int tsdbLockRepo(STsdbRepo* pRepo);
|
||||
int tsdbUnlockRepo(STsdbRepo* pRepo);
|
||||
STsdbMeta* tsdbGetMeta(STsdbRepo* pRepo);
|
||||
int tsdbCheckCommit(STsdbRepo* pRepo);
|
||||
int tsdbRestoreInfo(STsdbRepo* pRepo);
|
||||
int tsdbCacheLastData(STsdbRepo *pRepo, STsdbCfg* oldCfg);
|
||||
void tsdbGetRootDir(int repoid, char dirName[]);
|
||||
void tsdbGetDataDir(int repoid, char dirName[]);
|
||||
|
||||
static FORCE_INLINE STsdbBufBlock* tsdbGetCurrBufBlock(STsdbRepo* pRepo) {
|
||||
ASSERT(pRepo != NULL);
|
||||
if (pRepo->mem == NULL) return NULL;
|
||||
|
||||
SListNode* pNode = listTail(pRepo->mem->bufBlockList);
|
||||
if (pNode == NULL) return NULL;
|
||||
|
||||
STsdbBufBlock* pBufBlock = NULL;
|
||||
tdListNodeGetData(pRepo->mem->bufBlockList, pNode, (void*)(&pBufBlock));
|
||||
|
||||
return pBufBlock;
|
||||
}
|
||||
|
||||
static FORCE_INLINE int tsdbGetNextMaxTables(int tid) {
|
||||
ASSERT(tid >= 1 && tid <= TSDB_MAX_TABLES);
|
||||
int maxTables = TSDB_INIT_NTABLES;
|
||||
while (true) {
|
||||
maxTables = MIN(maxTables, TSDB_MAX_TABLES);
|
||||
if (tid <= maxTables) break;
|
||||
maxTables *= 2;
|
||||
}
|
||||
|
||||
return maxTables + 1;
|
||||
}
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
#endif
|
||||
|
||||
#endif
|
||||
#endif /* _TD_TSDB_INT_H_ */
|
||||
|
|
@ -1,98 +0,0 @@
|
|||
/*
|
||||
* 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 "os.h"
|
||||
#include "tmsg.h"
|
||||
#include "tarray.h"
|
||||
#include "query.h"
|
||||
#include "tglobal.h"
|
||||
#include "tlist.h"
|
||||
#include "tsdbint.h"
|
||||
#include "tsdbBuffer.h"
|
||||
#include "tsdbLog.h"
|
||||
#include "tsdbHealth.h"
|
||||
#include "ttimer.h"
|
||||
#include "tthread.h"
|
||||
|
||||
|
||||
// return malloc new block count
|
||||
int32_t tsdbInsertNewBlock(STsdbRepo * pRepo) {
|
||||
STsdbBufPool *pPool = pRepo->pPool;
|
||||
int32_t cnt = 0;
|
||||
|
||||
if(tsdbAllowNewBlock(pRepo)) {
|
||||
STsdbBufBlock *pBufBlock = tsdbNewBufBlock(pPool->bufBlockSize);
|
||||
if (pBufBlock) {
|
||||
if (tdListAppend(pPool->bufBlockList, (void *)(&pBufBlock)) < 0) {
|
||||
// append error
|
||||
tsdbFreeBufBlock(pBufBlock);
|
||||
} else {
|
||||
pPool->nElasticBlocks ++;
|
||||
cnt ++ ;
|
||||
}
|
||||
}
|
||||
}
|
||||
return cnt;
|
||||
}
|
||||
|
||||
// switch anther thread to run
|
||||
void* cbKillQueryFree(void* param) {
|
||||
STsdbRepo* pRepo = (STsdbRepo*)param;
|
||||
// vnode
|
||||
if(pRepo->appH.notifyStatus) {
|
||||
pRepo->appH.notifyStatus(pRepo->appH.appH, TSDB_STATUS_COMMIT_NOBLOCK, TSDB_CODE_SUCCESS);
|
||||
}
|
||||
|
||||
// free
|
||||
if(pRepo->pthread){
|
||||
void* p = pRepo->pthread;
|
||||
pRepo->pthread = NULL;
|
||||
free(p);
|
||||
}
|
||||
|
||||
return NULL;
|
||||
}
|
||||
|
||||
// return true do free , false do nothing
|
||||
bool tsdbUrgeQueryFree(STsdbRepo * pRepo) {
|
||||
// check previous running
|
||||
if(pRepo->pthread && taosThreadRunning(pRepo->pthread)) {
|
||||
tsdbWarn("vgId:%d pre urge thread is runing. nBlocks=%d nElasticBlocks=%d", REPO_ID(pRepo), pRepo->pPool->nBufBlocks, pRepo->pPool->nElasticBlocks);
|
||||
return false;
|
||||
}
|
||||
// create new
|
||||
pRepo->pthread = taosCreateThread(cbKillQueryFree, pRepo);
|
||||
if(pRepo->pthread == NULL) {
|
||||
tsdbError("vgId:%d create urge thread error.", REPO_ID(pRepo));
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
bool tsdbAllowNewBlock(STsdbRepo* pRepo) {
|
||||
int32_t nMaxElastic = pRepo->config.totalBlocks/3;
|
||||
STsdbBufPool* pPool = pRepo->pPool;
|
||||
if(pPool->nElasticBlocks >= nMaxElastic) {
|
||||
tsdbWarn("vgId:%d tsdbAllowNewBlock return fasle. nElasticBlock(%d) >= MaxElasticBlocks(%d)", REPO_ID(pRepo), pPool->nElasticBlocks, nMaxElastic);
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
bool tsdbNoProblem(STsdbRepo* pRepo) {
|
||||
if(listNEles(pRepo->pPool->bufBlockList) == 0)
|
||||
return false;
|
||||
return true;
|
||||
}
|
||||
|
|
@ -1,724 +0,0 @@
|
|||
/*
|
||||
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
|
||||
*
|
||||
* This program is free software: you can use, redistribute, and/or modify
|
||||
* it under the terms of the GNU Affero General Public License, version 3
|
||||
* or later ("AGPL"), as published by the Free Software Foundation.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful, but WITHOUT
|
||||
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
|
||||
* FITNESS FOR A PARTICULAR PURPOSE.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
#define _DEFAULT_SOURCE
|
||||
#include "os.h"
|
||||
#include "taoserror.h"
|
||||
#include "tsdbint.h"
|
||||
|
||||
// Sync handle
|
||||
typedef struct {
|
||||
STsdbRepo *pRepo;
|
||||
SRtn rtn;
|
||||
SOCKET socketFd;
|
||||
void * pBuf;
|
||||
bool mfChanged;
|
||||
SMFile * pmf;
|
||||
SMFile mf;
|
||||
SDFileSet df;
|
||||
SDFileSet *pdf;
|
||||
} SSyncH;
|
||||
|
||||
#define SYNC_BUFFER(sh) ((sh)->pBuf)
|
||||
|
||||
static void tsdbInitSyncH(SSyncH *pSyncH, STsdbRepo *pRepo, SOCKET socketFd);
|
||||
static void tsdbDestroySyncH(SSyncH *pSyncH);
|
||||
static int32_t tsdbSyncSendMeta(SSyncH *pSynch);
|
||||
static int32_t tsdbSyncRecvMeta(SSyncH *pSynch);
|
||||
static int32_t tsdbSendMetaInfo(SSyncH *pSynch);
|
||||
static int32_t tsdbRecvMetaInfo(SSyncH *pSynch);
|
||||
static int32_t tsdbSendDecision(SSyncH *pSynch, bool toSend);
|
||||
static int32_t tsdbRecvDecision(SSyncH *pSynch, bool *toSend);
|
||||
static int32_t tsdbSyncSendDFileSetArray(SSyncH *pSynch);
|
||||
static int32_t tsdbSyncRecvDFileSetArray(SSyncH *pSynch);
|
||||
static bool tsdbIsTowFSetSame(SDFileSet *pSet1, SDFileSet *pSet2);
|
||||
static int32_t tsdbSyncSendDFileSet(SSyncH *pSynch, SDFileSet *pSet);
|
||||
static int32_t tsdbSendDFileSetInfo(SSyncH *pSynch, SDFileSet *pSet);
|
||||
static int32_t tsdbRecvDFileSetInfo(SSyncH *pSynch);
|
||||
static int tsdbReload(STsdbRepo *pRepo, bool isMfChanged);
|
||||
|
||||
int32_t tsdbSyncSend(void *tsdb, SOCKET socketFd) {
|
||||
STsdbRepo *pRepo = (STsdbRepo *)tsdb;
|
||||
SSyncH synch = {0};
|
||||
|
||||
tsdbInitSyncH(&synch, pRepo, socketFd);
|
||||
// Disable TSDB commit
|
||||
tsem_wait(&(pRepo->readyToCommit));
|
||||
|
||||
if (tsdbSyncSendMeta(&synch) < 0) {
|
||||
tsdbError("vgId:%d, failed to send metafile since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
goto _err;
|
||||
}
|
||||
|
||||
if (tsdbSyncSendDFileSetArray(&synch) < 0) {
|
||||
tsdbError("vgId:%d, failed to send filesets since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
goto _err;
|
||||
}
|
||||
|
||||
// Enable TSDB commit
|
||||
tsem_post(&(pRepo->readyToCommit));
|
||||
tsdbDestroySyncH(&synch);
|
||||
return 0;
|
||||
|
||||
_err:
|
||||
tsem_post(&(pRepo->readyToCommit));
|
||||
tsdbDestroySyncH(&synch);
|
||||
return -1;
|
||||
}
|
||||
|
||||
int32_t tsdbSyncRecv(void *tsdb, SOCKET socketFd) {
|
||||
STsdbRepo *pRepo = (STsdbRepo *)tsdb;
|
||||
SSyncH synch = {0};
|
||||
|
||||
pRepo->state = TSDB_STATE_OK;
|
||||
|
||||
tsdbInitSyncH(&synch, pRepo, socketFd);
|
||||
tsem_wait(&(pRepo->readyToCommit));
|
||||
tsdbStartFSTxn(pRepo, 0, 0);
|
||||
|
||||
if (tsdbSyncRecvMeta(&synch) < 0) {
|
||||
tsdbError("vgId:%d, failed to recv metafile since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
goto _err;
|
||||
}
|
||||
|
||||
if (tsdbSyncRecvDFileSetArray(&synch) < 0) {
|
||||
tsdbError("vgId:%d, failed to recv filesets since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
goto _err;
|
||||
}
|
||||
|
||||
tsdbEndFSTxn(pRepo);
|
||||
tsem_post(&(pRepo->readyToCommit));
|
||||
tsdbDestroySyncH(&synch);
|
||||
|
||||
// Reload file change
|
||||
tsdbReload(pRepo, synch.mfChanged);
|
||||
|
||||
return 0;
|
||||
|
||||
_err:
|
||||
tsdbEndFSTxnWithError(REPO_FS(pRepo));
|
||||
tsem_post(&(pRepo->readyToCommit));
|
||||
tsdbDestroySyncH(&synch);
|
||||
return -1;
|
||||
}
|
||||
|
||||
static void tsdbInitSyncH(SSyncH *pSyncH, STsdbRepo *pRepo, SOCKET socketFd) {
|
||||
pSyncH->pRepo = pRepo;
|
||||
pSyncH->socketFd = socketFd;
|
||||
tsdbGetRtnSnap(pRepo, &(pSyncH->rtn));
|
||||
}
|
||||
|
||||
static void tsdbDestroySyncH(SSyncH *pSyncH) { taosTZfree(pSyncH->pBuf); }
|
||||
|
||||
static int32_t tsdbSyncSendMeta(SSyncH *pSynch) {
|
||||
STsdbRepo *pRepo = pSynch->pRepo;
|
||||
bool toSendMeta = false;
|
||||
SMFile mf;
|
||||
|
||||
// Send meta info to remote
|
||||
tsdbInfo("vgId:%d, metainfo will be sent", REPO_ID(pRepo));
|
||||
if (tsdbSendMetaInfo(pSynch) < 0) {
|
||||
tsdbError("vgId:%d, failed to send metainfo since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
if (pRepo->fs->cstatus->pmf == NULL) {
|
||||
// No meta file, not need to wait to retrieve meta file
|
||||
tsdbInfo("vgId:%d, metafile not exist, no need to send", REPO_ID(pRepo));
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (tsdbRecvDecision(pSynch, &toSendMeta) < 0) {
|
||||
tsdbError("vgId:%d, failed to recv decision while send meta since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
if (toSendMeta) {
|
||||
tsdbInitMFileEx(&mf, pRepo->fs->cstatus->pmf);
|
||||
if (tsdbOpenMFile(&mf, O_RDONLY) < 0) {
|
||||
tsdbError("vgId:%d, failed to open file while send metafile since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
int64_t writeLen = mf.info.size;
|
||||
tsdbInfo("vgId:%d, metafile:%s will be sent, size:%" PRId64, REPO_ID(pRepo), mf.f.aname, writeLen);
|
||||
|
||||
int64_t ret = taosSendFile(pSynch->socketFd, TSDB_FILE_FD(&mf), 0, writeLen);
|
||||
if (ret != writeLen) {
|
||||
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||
tsdbError("vgId:%d, failed to send metafile since %s, ret:%" PRId64 " writeLen:%" PRId64, REPO_ID(pRepo),
|
||||
tstrerror(terrno), ret, writeLen);
|
||||
tsdbCloseMFile(&mf);
|
||||
return -1;
|
||||
}
|
||||
|
||||
tsdbCloseMFile(&mf);
|
||||
tsdbInfo("vgId:%d, metafile is sent", REPO_ID(pRepo));
|
||||
} else {
|
||||
tsdbInfo("vgId:%d, metafile is same, no need to send", REPO_ID(pRepo));
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int32_t tsdbSyncRecvMeta(SSyncH *pSynch) {
|
||||
STsdbRepo *pRepo = pSynch->pRepo;
|
||||
SMFile * pLMFile = pRepo->fs->cstatus->pmf;
|
||||
|
||||
// Recv meta info from remote
|
||||
if (tsdbRecvMetaInfo(pSynch) < 0) {
|
||||
tsdbError("vgId:%d, failed to recv metainfo since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
// No meta file, do nothing (rm local meta file)
|
||||
if (pSynch->pmf == NULL) {
|
||||
if (pLMFile == NULL) {
|
||||
pSynch->mfChanged = false;
|
||||
} else {
|
||||
pSynch->mfChanged = true;
|
||||
}
|
||||
tsdbInfo("vgId:%d, metafile not exist in remote, no need to recv", REPO_ID(pRepo));
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (pLMFile == NULL || pSynch->pmf->info.size != pLMFile->info.size ||
|
||||
pSynch->pmf->info.magic != pLMFile->info.magic || TSDB_FILE_IS_BAD(pLMFile)) {
|
||||
// Local has no meta file or has a different meta file, need to copy from remote
|
||||
pSynch->mfChanged = true;
|
||||
|
||||
if (tsdbSendDecision(pSynch, true) < 0) {
|
||||
tsdbError("vgId:%d, failed to send decision while recv metafile since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
tsdbInfo("vgId:%d, metafile will be received", REPO_ID(pRepo));
|
||||
|
||||
// Recv from remote
|
||||
SMFile mf;
|
||||
SDiskID did = {.level = TFS_PRIMARY_LEVEL, .id = TFS_PRIMARY_ID};
|
||||
tsdbInitMFile(&mf, did, REPO_ID(pRepo), FS_TXN_VERSION(REPO_FS(pRepo)));
|
||||
if (tsdbCreateMFile(&mf, false) < 0) {
|
||||
tsdbError("vgId:%d, failed to create file while recv metafile since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
tsdbInfo("vgId:%d, metafile:%s is created", REPO_ID(pRepo), mf.f.aname);
|
||||
|
||||
int64_t readLen = pSynch->pmf->info.size;
|
||||
int64_t ret = taosCopyFds(pSynch->socketFd, TSDB_FILE_FD(&mf), readLen);
|
||||
if (ret != readLen) {
|
||||
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||
tsdbError("vgId:%d, failed to recv metafile since %s, ret:%" PRId64 " readLen:%" PRId64, REPO_ID(pRepo),
|
||||
tstrerror(terrno), ret, readLen);
|
||||
tsdbCloseMFile(&mf);
|
||||
tsdbRemoveMFile(&mf);
|
||||
return -1;
|
||||
}
|
||||
|
||||
tsdbInfo("vgId:%d, metafile is received, size:%" PRId64, REPO_ID(pRepo), readLen);
|
||||
|
||||
mf.info = pSynch->pmf->info;
|
||||
tsdbCloseMFile(&mf);
|
||||
tsdbUpdateMFile(REPO_FS(pRepo), &mf);
|
||||
} else {
|
||||
pSynch->mfChanged = false;
|
||||
tsdbInfo("vgId:%d, metafile is same, no need to recv", REPO_ID(pRepo));
|
||||
if (tsdbSendDecision(pSynch, false) < 0) {
|
||||
tsdbError("vgId:%d, failed to send decision while recv metafile since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
tsdbUpdateMFile(REPO_FS(pRepo), pLMFile);
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int32_t tsdbSendMetaInfo(SSyncH *pSynch) {
|
||||
STsdbRepo *pRepo = pSynch->pRepo;
|
||||
uint32_t tlen = 0;
|
||||
SMFile * pMFile = pRepo->fs->cstatus->pmf;
|
||||
|
||||
if (pMFile) {
|
||||
tlen = tlen + tsdbEncodeSMFileEx(NULL, pMFile) + sizeof(TSCKSUM);
|
||||
}
|
||||
|
||||
if (tsdbMakeRoom((void **)(&SYNC_BUFFER(pSynch)), tlen + sizeof(tlen)) < 0) {
|
||||
tsdbError("vgId:%d, failed to makeroom while send metainfo since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
void *ptr = SYNC_BUFFER(pSynch);
|
||||
taosEncodeFixedU32(&ptr, tlen);
|
||||
void *tptr = ptr;
|
||||
if (pMFile) {
|
||||
tsdbEncodeSMFileEx(&ptr, pMFile);
|
||||
taosCalcChecksumAppend(0, (uint8_t *)tptr, tlen);
|
||||
}
|
||||
|
||||
int32_t writeLen = tlen + sizeof(uint32_t);
|
||||
int32_t ret = taosWriteMsg(pSynch->socketFd, SYNC_BUFFER(pSynch), writeLen);
|
||||
if (ret != writeLen) {
|
||||
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||
tsdbError("vgId:%d, failed to send metainfo since %s, ret:%d writeLen:%d", REPO_ID(pRepo), tstrerror(terrno), ret,
|
||||
writeLen);
|
||||
return -1;
|
||||
}
|
||||
|
||||
tsdbInfo("vgId:%d, metainfo is sent, tlen:%d, writeLen:%d", REPO_ID(pRepo), tlen, writeLen);
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int32_t tsdbRecvMetaInfo(SSyncH *pSynch) {
|
||||
STsdbRepo *pRepo = pSynch->pRepo;
|
||||
uint32_t tlen = 0;
|
||||
char buf[64] = {0};
|
||||
|
||||
int32_t readLen = sizeof(uint32_t);
|
||||
int32_t ret = taosReadMsg(pSynch->socketFd, buf, readLen);
|
||||
if (ret != readLen) {
|
||||
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||
tsdbError("vgId:%d, failed to recv metalen, ret:%d readLen:%d", REPO_ID(pRepo), ret, readLen);
|
||||
return -1;
|
||||
}
|
||||
|
||||
taosDecodeFixedU32(buf, &tlen);
|
||||
|
||||
tsdbInfo("vgId:%d, metalen is received, readLen:%d, tlen:%d", REPO_ID(pRepo), readLen, tlen);
|
||||
if (tlen == 0) {
|
||||
pSynch->pmf = NULL;
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (tsdbMakeRoom((void **)(&SYNC_BUFFER(pSynch)), tlen) < 0) {
|
||||
tsdbError("vgId:%d, failed to makeroom while recv metainfo since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
ret = taosReadMsg(pSynch->socketFd, SYNC_BUFFER(pSynch), tlen);
|
||||
if (ret != tlen) {
|
||||
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||
tsdbError("vgId:%d, failed to recv metainfo, ret:%d tlen:%d", REPO_ID(pRepo), ret, tlen);
|
||||
return -1;
|
||||
}
|
||||
|
||||
tsdbInfo("vgId:%d, metainfo is received, tlen:%d", REPO_ID(pRepo), tlen);
|
||||
if (!taosCheckChecksumWhole((uint8_t *)SYNC_BUFFER(pSynch), tlen)) {
|
||||
terrno = TSDB_CODE_TDB_MESSED_MSG;
|
||||
tsdbError("vgId:%d, failed to checksum while recv metainfo since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
pSynch->pmf = &(pSynch->mf);
|
||||
tsdbDecodeSMFileEx(SYNC_BUFFER(pSynch), pSynch->pmf);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int32_t tsdbSendDecision(SSyncH *pSynch, bool toSend) {
|
||||
STsdbRepo *pRepo = pSynch->pRepo;
|
||||
uint8_t decision = toSend;
|
||||
|
||||
int32_t writeLen = sizeof(uint8_t);
|
||||
int32_t ret = taosWriteMsg(pSynch->socketFd, (void *)(&decision), writeLen);
|
||||
if (ret != writeLen) {
|
||||
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||
tsdbError("vgId:%d, failed to send decison, ret:%d writeLen:%d", REPO_ID(pRepo), ret, writeLen);
|
||||
return -1;
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int32_t tsdbRecvDecision(SSyncH *pSynch, bool *toSend) {
|
||||
STsdbRepo *pRepo = pSynch->pRepo;
|
||||
uint8_t decision = 0;
|
||||
|
||||
int32_t readLen = sizeof(uint8_t);
|
||||
int32_t ret = taosReadMsg(pSynch->socketFd, (void *)(&decision), readLen);
|
||||
if (ret != readLen) {
|
||||
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||
tsdbError("vgId:%d, failed to recv decison, ret:%d readLen:%d", REPO_ID(pRepo), ret, readLen);
|
||||
return -1;
|
||||
}
|
||||
|
||||
*toSend = decision;
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int32_t tsdbSyncSendDFileSetArray(SSyncH *pSynch) {
|
||||
STsdbRepo *pRepo = pSynch->pRepo;
|
||||
STsdbFS * pfs = REPO_FS(pRepo);
|
||||
SFSIter fsiter;
|
||||
SDFileSet *pSet;
|
||||
|
||||
tsdbFSIterInit(&fsiter, pfs, TSDB_FS_ITER_FORWARD);
|
||||
|
||||
do {
|
||||
pSet = tsdbFSIterNext(&fsiter);
|
||||
if (tsdbSyncSendDFileSet(pSynch, pSet) < 0) {
|
||||
tsdbError("vgId:%d, failed to send fileset:%d since %s", REPO_ID(pRepo), pSet ? pSet->fid : -1,
|
||||
tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
// No more file set to send, jut break
|
||||
if (pSet == NULL) {
|
||||
tsdbInfo("vgId:%d, no filesets any more", REPO_ID(pRepo));
|
||||
break;
|
||||
}
|
||||
} while (true);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int32_t tsdbSyncRecvDFileSetArray(SSyncH *pSynch) {
|
||||
STsdbRepo *pRepo = pSynch->pRepo;
|
||||
STsdbFS * pfs = REPO_FS(pRepo);
|
||||
SFSIter fsiter;
|
||||
SDFileSet *pLSet; // Local file set
|
||||
|
||||
tsdbFSIterInit(&fsiter, pfs, TSDB_FS_ITER_FORWARD);
|
||||
|
||||
pLSet = tsdbFSIterNext(&fsiter);
|
||||
if (tsdbRecvDFileSetInfo(pSynch) < 0) {
|
||||
tsdbError("vgId:%d, failed to recv fileset since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
while (true) {
|
||||
if (pLSet == NULL && pSynch->pdf == NULL) {
|
||||
tsdbInfo("vgId:%d, all filesets is disposed", REPO_ID(pRepo));
|
||||
break;
|
||||
} else {
|
||||
tsdbInfo("vgId:%d, fileset local:%d remote:%d, will be disposed", REPO_ID(pRepo), pLSet != NULL ? pLSet->fid : -1,
|
||||
pSynch->pdf != NULL ? pSynch->pdf->fid : -1);
|
||||
}
|
||||
|
||||
if (pLSet && (pSynch->pdf == NULL || pLSet->fid < pSynch->pdf->fid)) {
|
||||
// remote not has pLSet->fid set, just remove local (do nothing to remote the fset)
|
||||
tsdbInfo("vgId:%d, fileset:%d smaller than remote:%d, remove it", REPO_ID(pRepo), pLSet->fid,
|
||||
pSynch->pdf != NULL ? pSynch->pdf->fid : -1);
|
||||
pLSet = tsdbFSIterNext(&fsiter);
|
||||
} else {
|
||||
if (pLSet && pSynch->pdf && pLSet->fid == pSynch->pdf->fid && tsdbIsTowFSetSame(pLSet, pSynch->pdf) &&
|
||||
tsdbFSetIsOk(pLSet)) {
|
||||
// Just keep local files and notify remote not to send
|
||||
tsdbInfo("vgId:%d, fileset:%d is same and no need to recv", REPO_ID(pRepo), pLSet->fid);
|
||||
|
||||
if (tsdbUpdateDFileSet(pfs, pLSet) < 0) {
|
||||
tsdbError("vgId:%d, failed to update fileset since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
if (tsdbSendDecision(pSynch, false) < 0) {
|
||||
tsdbError("vgId:%d, failed to send decision since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
} else {
|
||||
// Need to copy from remote
|
||||
int fidLevel = tsdbGetFidLevel(pSynch->pdf->fid, &(pSynch->rtn));
|
||||
if (fidLevel < 0) { // expired fileset
|
||||
tsdbInfo("vgId:%d, fileset:%d will be skipped as expired", REPO_ID(pRepo), pSynch->pdf->fid);
|
||||
if (tsdbSendDecision(pSynch, false) < 0) {
|
||||
tsdbError("vgId:%d, failed to send decision since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
// Move forward
|
||||
if (tsdbRecvDFileSetInfo(pSynch) < 0) {
|
||||
tsdbError("vgId:%d, failed to recv fileset since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
if (pLSet) {
|
||||
pLSet = tsdbFSIterNext(&fsiter);
|
||||
}
|
||||
// Next loop
|
||||
continue;
|
||||
} else {
|
||||
tsdbInfo("vgId:%d, fileset:%d will be received", REPO_ID(pRepo), pSynch->pdf->fid);
|
||||
// Notify remote to send there file here
|
||||
if (tsdbSendDecision(pSynch, true) < 0) {
|
||||
tsdbError("vgId:%d, failed to send decision since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
|
||||
// Create local files and copy from remote
|
||||
SDiskID did;
|
||||
SDFileSet fset;
|
||||
|
||||
tfsAllocDisk(fidLevel, &(did.level), &(did.id));
|
||||
if (did.level == TFS_UNDECIDED_LEVEL) {
|
||||
terrno = TSDB_CODE_TDB_NO_AVAIL_DISK;
|
||||
tsdbError("vgId:%d, failed allc disk since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
tsdbInitDFileSet(&fset, did, REPO_ID(pRepo), pSynch->pdf->fid, FS_TXN_VERSION(pfs));
|
||||
|
||||
// Create new FSET
|
||||
if (tsdbCreateDFileSet(&fset, false) < 0) {
|
||||
tsdbError("vgId:%d, failed to create fileset since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
|
||||
SDFile *pDFile = TSDB_DFILE_IN_SET(&fset, ftype); // local file
|
||||
SDFile *pRDFile = TSDB_DFILE_IN_SET(pSynch->pdf, ftype); // remote file
|
||||
|
||||
tsdbInfo("vgId:%d, file:%s will be received, osize:%" PRIu64 " rsize:%" PRIu64, REPO_ID(pRepo),
|
||||
pDFile->f.aname, pDFile->info.size, pRDFile->info.size);
|
||||
|
||||
int64_t writeLen = pRDFile->info.size;
|
||||
int64_t ret = taosCopyFds(pSynch->socketFd, pDFile->fd, writeLen);
|
||||
if (ret != writeLen) {
|
||||
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||
tsdbError("vgId:%d, failed to recv file:%s since %s, ret:%" PRId64 " writeLen:%" PRId64, REPO_ID(pRepo),
|
||||
pDFile->f.aname, tstrerror(terrno), ret, writeLen);
|
||||
tsdbCloseDFileSet(&fset);
|
||||
tsdbRemoveDFileSet(&fset);
|
||||
return -1;
|
||||
}
|
||||
|
||||
// Update new file info
|
||||
pDFile->info = pRDFile->info;
|
||||
tsdbInfo("vgId:%d, file:%s is received, size:%" PRId64, REPO_ID(pRepo), pDFile->f.aname, writeLen);
|
||||
}
|
||||
|
||||
tsdbCloseDFileSet(&fset);
|
||||
if (tsdbUpdateDFileSet(pfs, &fset) < 0) {
|
||||
tsdbInfo("vgId:%d, fileset:%d failed to update since %s", REPO_ID(pRepo), fset.fid, tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
tsdbInfo("vgId:%d, fileset:%d is received", REPO_ID(pRepo), pSynch->pdf->fid);
|
||||
}
|
||||
|
||||
// Move forward
|
||||
if (tsdbRecvDFileSetInfo(pSynch) < 0) {
|
||||
tsdbError("vgId:%d, failed to recv fileset since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
if (pLSet) {
|
||||
pLSet = tsdbFSIterNext(&fsiter);
|
||||
}
|
||||
}
|
||||
|
||||
#if 0
|
||||
if (pLSet == NULL) {
|
||||
// Copy from remote >>>>>>>>>>>
|
||||
} else {
|
||||
if (pSynch->pdf == NULL) {
|
||||
// Remove local file, just ignore ++++++++++++++
|
||||
pLSet = tsdbFSIterNext(&fsiter);
|
||||
} else {
|
||||
if (pLSet->fid < pSynch->pdf->fid) {
|
||||
// Remove local file, just ignore ++++++++++++
|
||||
pLSet = tsdbFSIterNext(&fsiter);
|
||||
} else if (pLSet->fid > pSynch->pdf->fid){
|
||||
// Copy from remote >>>>>>>>>>>>>>
|
||||
if (tsdbRecvDFileSetInfo(pSynch) < 0) {
|
||||
// TODO
|
||||
return -1;
|
||||
}
|
||||
} else {
|
||||
if (true/*TODO: is same fset*/) {
|
||||
// No need to copy ---------------------
|
||||
} else {
|
||||
// copy from remote >>>>>>>>>>>>>.
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
#endif
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
static bool tsdbIsTowFSetSame(SDFileSet *pSet1, SDFileSet *pSet2) {
|
||||
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
|
||||
SDFile *pDFile1 = TSDB_DFILE_IN_SET(pSet1, ftype);
|
||||
SDFile *pDFile2 = TSDB_DFILE_IN_SET(pSet2, ftype);
|
||||
|
||||
if (pDFile1->info.size != pDFile2->info.size || pDFile1->info.magic != pDFile2->info.magic) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
static int32_t tsdbSyncSendDFileSet(SSyncH *pSynch, SDFileSet *pSet) {
|
||||
STsdbRepo *pRepo = pSynch->pRepo;
|
||||
bool toSend = false;
|
||||
|
||||
// skip expired fileset
|
||||
if (pSet && tsdbGetFidLevel(pSet->fid, &(pSynch->rtn)) < 0) {
|
||||
tsdbInfo("vgId:%d, don't sync send since fileset:%d smaller than minFid:%d", REPO_ID(pRepo), pSet->fid,
|
||||
pSynch->rtn.minFid);
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (tsdbSendDFileSetInfo(pSynch, pSet) < 0) {
|
||||
tsdbError("vgId:%d, failed to send fileset:%d info since %s", REPO_ID(pRepo), pSet ? pSet->fid : -1, tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
// No file any more, no need to send file, just return
|
||||
if (pSet == NULL) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (tsdbRecvDecision(pSynch, &toSend) < 0) {
|
||||
tsdbError("vgId:%d, failed to recv decision while send fileset:%d since %s", REPO_ID(pRepo), pSet->fid,
|
||||
tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
if (toSend) {
|
||||
tsdbInfo("vgId:%d, fileset:%d will be sent", REPO_ID(pRepo), pSet->fid);
|
||||
|
||||
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
|
||||
SDFile df = *TSDB_DFILE_IN_SET(pSet, ftype);
|
||||
|
||||
if (tsdbOpenDFile(&df, O_RDONLY) < 0) {
|
||||
tsdbError("vgId:%d, failed to file:%s since %s", REPO_ID(pRepo), df.f.aname, tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
int64_t writeLen = df.info.size;
|
||||
tsdbInfo("vgId:%d, file:%s will be sent, size:%" PRId64, REPO_ID(pRepo), df.f.aname, writeLen);
|
||||
|
||||
int64_t ret = taosSendFile(pSynch->socketFd, TSDB_FILE_FD(&df), 0, writeLen);
|
||||
if (ret != writeLen) {
|
||||
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||
tsdbError("vgId:%d, failed to send file:%s since %s, ret:%" PRId64 " writeLen:%" PRId64, REPO_ID(pRepo),
|
||||
df.f.aname, tstrerror(terrno), ret, writeLen);
|
||||
tsdbCloseDFile(&df);
|
||||
return -1;
|
||||
}
|
||||
|
||||
tsdbInfo("vgId:%d, file:%s is sent", REPO_ID(pRepo), df.f.aname);
|
||||
tsdbCloseDFile(&df);
|
||||
}
|
||||
|
||||
tsdbInfo("vgId:%d, fileset:%d is sent", REPO_ID(pRepo), pSet->fid);
|
||||
} else {
|
||||
tsdbInfo("vgId:%d, fileset:%d is same, no need to send", REPO_ID(pRepo), pSet->fid);
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int32_t tsdbSendDFileSetInfo(SSyncH *pSynch, SDFileSet *pSet) {
|
||||
STsdbRepo *pRepo = pSynch->pRepo;
|
||||
uint32_t tlen = 0;
|
||||
|
||||
if (pSet) {
|
||||
tlen = tsdbEncodeDFileSetEx(NULL, pSet) + sizeof(TSCKSUM);
|
||||
}
|
||||
|
||||
if (tsdbMakeRoom((void **)(&SYNC_BUFFER(pSynch)), tlen + sizeof(tlen)) < 0) {
|
||||
tsdbError("vgId:%d, failed to makeroom while send fileinfo since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
void *ptr = SYNC_BUFFER(pSynch);
|
||||
taosEncodeFixedU32(&ptr, tlen);
|
||||
void *tptr = ptr;
|
||||
if (pSet) {
|
||||
tsdbEncodeDFileSetEx(&ptr, pSet);
|
||||
taosCalcChecksumAppend(0, (uint8_t *)tptr, tlen);
|
||||
}
|
||||
|
||||
int32_t writeLen = tlen + sizeof(uint32_t);
|
||||
int32_t ret = taosWriteMsg(pSynch->socketFd, SYNC_BUFFER(pSynch), writeLen);
|
||||
if (ret != writeLen) {
|
||||
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||
tsdbError("vgId:%d, failed to send fileinfo, ret:%d writeLen:%d", REPO_ID(pRepo), ret, writeLen);
|
||||
return -1;
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int32_t tsdbRecvDFileSetInfo(SSyncH *pSynch) {
|
||||
STsdbRepo *pRepo = pSynch->pRepo;
|
||||
uint32_t tlen;
|
||||
char buf[64] = {0};
|
||||
|
||||
int32_t readLen = sizeof(uint32_t);
|
||||
int32_t ret = taosReadMsg(pSynch->socketFd, buf, readLen);
|
||||
if (ret != readLen) {
|
||||
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||
return -1;
|
||||
}
|
||||
|
||||
taosDecodeFixedU32(buf, &tlen);
|
||||
|
||||
tsdbInfo("vgId:%d, fileinfo len:%d is received", REPO_ID(pRepo), tlen);
|
||||
if (tlen == 0) {
|
||||
pSynch->pdf = NULL;
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (tsdbMakeRoom((void **)(&SYNC_BUFFER(pSynch)), tlen) < 0) {
|
||||
tsdbError("vgId:%d, failed to makeroom while recv fileinfo since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
ret = taosReadMsg(pSynch->socketFd, SYNC_BUFFER(pSynch), tlen);
|
||||
if (ret != tlen) {
|
||||
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||
tsdbError("vgId:%d, failed to recv fileinfo, ret:%d readLen:%d", REPO_ID(pRepo), ret, tlen);
|
||||
return -1;
|
||||
}
|
||||
|
||||
if (!taosCheckChecksumWhole((uint8_t *)SYNC_BUFFER(pSynch), tlen)) {
|
||||
terrno = TSDB_CODE_TDB_MESSED_MSG;
|
||||
tsdbError("vgId:%d, failed to checksum while recv fileinfo since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
pSynch->pdf = &(pSynch->df);
|
||||
tsdbDecodeDFileSetEx(SYNC_BUFFER(pSynch), pSynch->pdf);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int tsdbReload(STsdbRepo *pRepo, bool isMfChanged) {
|
||||
// TODO: may need to stop and restart stream
|
||||
// if (isMfChanged) {
|
||||
tsdbCloseMeta(pRepo);
|
||||
tsdbFreeMeta(pRepo->tsdbMeta);
|
||||
pRepo->tsdbMeta = tsdbNewMeta(REPO_CFG(pRepo));
|
||||
tsdbOpenMeta(pRepo);
|
||||
tsdbLoadMetaCache(pRepo, true);
|
||||
// }
|
||||
|
||||
tsdbUnRefMemTable(pRepo, pRepo->mem);
|
||||
tsdbUnRefMemTable(pRepo, pRepo->imem);
|
||||
pRepo->mem = NULL;
|
||||
pRepo->imem = NULL;
|
||||
|
||||
if (tsdbRestoreInfo(pRepo) < 0) {
|
||||
tsdbError("vgId:%d failed to restore info from file since %s", REPO_ID(pRepo), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
Some files were not shown because too many files have changed in this diff Show More
Loading…
Reference in New Issue