Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
T
TDengine
提交 f73bc53b 编写于 作者: H hzcheng
浏览文件
操作

Merge branch 'develop' into feature/2.0tsdb

上级 3f5b1415 89b72161
隐藏空白更改
内联 并排
Showing with 849 addition and 387 deletion
deps/CMakeLists.txt
浏览文件 @ f73bc53b
......@@ -5,3 +5,5 @@ ADD_SUBDIRECTORY(zlib-1.2.11)
ADD_SUBDIRECTORY(pthread)
ADD_SUBDIRECTORY(regex)
ADD_SUBDIRECTORY(iconv)
ADD_SUBDIRECTORY(lz4)
ADD_SUBDIRECTORY(cJson)
src/thirdparty/CMakeLists.txt → deps/cJson/CMakeLists.txt
浏览文件 @ f73bc53b
aux_source_directory(${CMAKE_CURRENT_SOURCE_DIR}/src SOURCE_LIST)
add_library(thirdparty ${SOURCE_LIST})
target_include_directories(thirdparty PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/inc)
\ No newline at end of file
add_library(cJson ${SOURCE_LIST})
target_include_directories(cJson PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/inc)
\ No newline at end of file
deps/lz4/CMakeLists.txt 0 → 100644
浏览文件 @ f73bc53b
aux_source_directory(${CMAKE_CURRENT_SOURCE_DIR}/src SOURCE_LIST)
add_library(lz4 ${SOURCE_LIST})
target_include_directories(lz4 PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/inc)
\ No newline at end of file
src/CMakeLists.txt
浏览文件 @ f73bc53b
......@@ -3,8 +3,6 @@ PROJECT(TDengine)
# Base compile
ADD_SUBDIRECTORY(os)
ADD_SUBDIRECTORY(thirdparty)
ADD_SUBDIRECTORY(common)
ADD_SUBDIRECTORY(util)
ADD_SUBDIRECTORY(rpc)
......@@ -14,5 +12,7 @@ ADD_SUBDIRECTORY(kit)
ADD_SUBDIRECTORY(plugins)
ADD_SUBDIRECTORY(mnode)
ADD_SUBDIRECTORY(vnode)
ADD_SUBDIRECTORY(tsdb)
ADD_SUBDIRECTORY(wal)
ADD_SUBDIRECTORY(dnode)
#ADD_SUBDIRECTORY(connector/jdbc)
src/dnode/CMakeLists.txt
浏览文件 @ f73bc53b
......@@ -7,14 +7,16 @@ IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/util/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/query/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/mnode/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/vnode/tsdb/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/tsdb/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/common/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/cJson/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/lz4/inc)
INCLUDE_DIRECTORIES(${TD_ENTERPRISE_DIR}/src/inc)
INCLUDE_DIRECTORIES(inc)
AUX_SOURCE_DIRECTORY(src SRC)
ADD_EXECUTABLE(taosd ${SRC})
TARGET_LINK_LIBRARIES(taosd mnode taos_static monitor http tsdb twal vnode)
TARGET_LINK_LIBRARIES(taosd mnode taos_static monitor http tsdb twal vnode cJson lz4)
IF (TD_ACCOUNT)
TARGET_LINK_LIBRARIES(taosd account)
......@@ -31,6 +33,10 @@ IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
TARGET_LINK_LIBRARIES(taosd balance sync)
ENDIF ()
IF (TD_MPEER)
TARGET_LINK_LIBRARIES(taosd mpeer sync)
ENDIF ()
SET(PREPARE_ENV_CMD "prepare_env_cmd")
SET(PREPARE_ENV_TARGET "prepare_env_target")
ADD_CUSTOM_COMMAND(OUTPUT ${PREPARE_ENV_CMD}
......
src/dnode/inc/dnodeMClient.h
浏览文件 @ f73bc53b
......@@ -25,6 +25,7 @@ void dnodeCleanupMClient();
void dnodeSendMsgToMnode(SRpcMsg *rpcMsg);
uint32_t dnodeGetMnodeMasteIp();
void * dnodeGetMpeerInfos();
int32_t dnodeGetDnodeId();
#ifdef __cplusplus
}
......
src/dnode/inc/dnodeMgmt.h
浏览文件 @ f73bc53b
......@@ -23,7 +23,6 @@ extern "C" {
int32_t dnodeInitMgmt();
void dnodeCleanupMgmt();
void dnodeMgmt(SRpcMsg *rpcMsg);
void dnodeUpdateDnodeId(int32_t dnodeId);
void* dnodeGetVnode(int32_t vgId);
int32_t dnodeGetVnodeStatus(void *pVnode);
......
src/dnode/src/dnodeMClient.c
浏览文件 @ f73bc53b
......@@ -21,30 +21,52 @@
#include "trpc.h"
#include "tutil.h"
#include "tsync.h"
#include "ttime.h"
#include "ttimer.h"
#include "dnode.h"
#include "dnodeMClient.h"
#include "dnodeModule.h"
#include "dnodeMgmt.h"
#include "vnode.h"
#include "mpeer.h"
#define MPEER_CONTENT_LEN 2000
static bool dnodeReadMnodeIpList();
static void dnodeSaveMnodeIpList();
static void dnodeReadDnodeInfo();
static void dnodeUpdateDnodeInfo(int32_t dnodeId);
static void dnodeProcessRspFromMnode(SRpcMsg *pMsg);
static void dnodeProcessStatusRsp(SRpcMsg *pMsg);
static void dnodeSendStatusMsg(void *handle, void *tmrId);
static void (*tsDnodeProcessMgmtRspFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *);
static void *tsDnodeMClientRpc = NULL;
static SRpcIpSet tsMnodeIpList = {0};
static SDMNodeInfos tsMnodeInfos = {0};
static void *tsDnodeTmr = NULL;
static void *tsStatusTimer = NULL;
static uint32_t tsRebootTime;
static int32_t tsDnodeId = 0;
static char tsDnodeName[TSDB_NODE_NAME_LEN];
int32_t dnodeInitMClient() {
dnodeReadDnodeInfo();
tsRebootTime = taosGetTimestampSec();
tsDnodeTmr = taosTmrInit(100, 200, 60000, "DND-DM");
if (tsDnodeTmr == NULL) {
dError("failed to init dnode timer");
return -1;
}
if (!dnodeReadMnodeIpList()) {
memset(&tsMnodeIpList, 0, sizeof(SRpcIpSet));
memset(&tsMnodeInfos, 0, sizeof(SDMNodeInfos));
tsMnodeIpList.port = tsMnodeDnodePort;
tsMnodeIpList.numOfIps = 1;
tsMnodeIpList.ip[0] = inet_addr(tsMasterIp);
if (tsSecondIp[0]) {
if (strcmp(tsSecondIp, tsMasterIp) != 0) {
tsMnodeIpList.numOfIps = 2;
tsMnodeIpList.ip[1] = inet_addr(tsSecondIp);
}
......@@ -57,8 +79,6 @@ int32_t dnodeInitMClient() {
}
}
tsDnodeProcessMgmtRspFp[TSDB_MSG_TYPE_DM_STATUS_RSP] = dnodeProcessStatusRsp;
SRpcInit rpcInit;
memset(&rpcInit, 0, sizeof(rpcInit));
rpcInit.localIp = tsAnyIp ? "0.0.0.0" : tsPrivateIp;
......@@ -79,11 +99,24 @@ int32_t dnodeInitMClient() {
return -1;
}
tsDnodeProcessMgmtRspFp[TSDB_MSG_TYPE_DM_STATUS_RSP] = dnodeProcessStatusRsp;
taosTmrReset(dnodeSendStatusMsg, 500, NULL, tsDnodeTmr, &tsStatusTimer);
dPrint("mnode rpc client is opened");
return 0;
}
void dnodeCleanupMClient() {
if (tsStatusTimer != NULL) {
taosTmrStopA(&tsStatusTimer);
tsStatusTimer = NULL;
}
if (tsDnodeTmr != NULL) {
taosTmrCleanUp(tsDnodeTmr);
tsDnodeTmr = NULL;
}
if (tsDnodeMClientRpc) {
rpcClose(tsDnodeMClientRpc);
tsDnodeMClientRpc = NULL;
......@@ -104,6 +137,7 @@ static void dnodeProcessRspFromMnode(SRpcMsg *pMsg) {
static void dnodeProcessStatusRsp(SRpcMsg *pMsg) {
if (pMsg->code != TSDB_CODE_SUCCESS) {
dError("status rsp is received, error:%s", tstrerror(pMsg->code));
taosTmrReset(dnodeSendStatusMsg, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);
return;
}
......@@ -111,9 +145,19 @@ static void dnodeProcessStatusRsp(SRpcMsg *pMsg) {
SDMNodeInfos *mpeers = &pStatusRsp->mpeers;
if (mpeers->nodeNum <= 0) {
dError("status msg is invalid, num of ips is %d", mpeers->nodeNum);
taosTmrReset(dnodeSendStatusMsg, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);
return;
}
SDnodeState *pState = &pStatusRsp->dnodeState;
pState->numOfVnodes = htonl(pState->numOfVnodes);
pState->moduleStatus = htonl(pState->moduleStatus);
pState->createdTime = htonl(pState->createdTime);
pState->dnodeId = htonl(pState->dnodeId);
dnodeProcessModuleStatus(pState->moduleStatus);
dnodeUpdateDnodeInfo(pState->dnodeId);
SRpcIpSet mgmtIpSet = {0};
mgmtIpSet.inUse = mpeers->inUse;
mgmtIpSet.numOfIps = mpeers->nodeNum;
......@@ -122,29 +166,25 @@ static void dnodeProcessStatusRsp(SRpcMsg *pMsg) {
mgmtIpSet.ip[i] = htonl(mpeers->nodeInfos[i].nodeIp);
}
if (memcmp(&mgmtIpSet, &tsMnodeIpList, sizeof(SRpcIpSet)) != 0) {
if (memcmp(&mgmtIpSet, &tsMnodeIpList, sizeof(SRpcIpSet)) != 0 || tsMnodeInfos.nodeNum == 0) {
memcpy(&tsMnodeIpList, &mgmtIpSet, sizeof(SRpcIpSet));
memcpy(&tsMnodeInfos, mpeers, sizeof(SDMNodeInfos));
tsMnodeInfos.inUse = mpeers->inUse;
tsMnodeInfos.nodeNum = mpeers->nodeNum;
dPrint("mnode ip list is changed, numOfIps:%d inUse:%d", tsMnodeInfos.nodeNum, tsMnodeInfos.inUse);
for (int32_t i = 0; i < mpeers->nodeNum; i++) {
tsMnodeInfos.nodeInfos[i].nodeId = htonl(mpeers->nodeInfos[i].nodeId);
tsMnodeInfos.nodeInfos[i].nodeIp = htonl(mpeers->nodeInfos[i].nodeIp);
tsMnodeInfos.nodeInfos[i].nodePort = htons(mpeers->nodeInfos[i].nodePort);
strcpy(tsMnodeInfos.nodeInfos[i].nodeName, mpeers->nodeInfos[i].nodeName);
dPrint("mnode:%d, ip:%s:%u name:%s", tsMnodeInfos.nodeInfos[i].nodeId,
taosIpStr(tsMnodeInfos.nodeInfos[i].nodeId), tsMnodeInfos.nodeInfos[i].nodePort,
taosIpStr(tsMnodeInfos.nodeInfos[i].nodeIp), tsMnodeInfos.nodeInfos[i].nodePort,
tsMnodeInfos.nodeInfos[i].nodeName);
}
dnodeSaveMnodeIpList();
mpeerUpdateSync();
}
SDnodeState *pState = &pStatusRsp->dnodeState;
pState->numOfVnodes = htonl(pState->numOfVnodes);
pState->moduleStatus = htonl(pState->moduleStatus);
pState->createdTime = htonl(pState->createdTime);
pState->dnodeId = htonl(pState->dnodeId);
dnodeProcessModuleStatus(pState->moduleStatus);
dnodeUpdateDnodeId(pState->dnodeId);
taosTmrReset(dnodeSendStatusMsg, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);
}
void dnodeSendMsgToMnode(SRpcMsg *rpcMsg) {
......@@ -294,4 +334,93 @@ uint32_t dnodeGetMnodeMasteIp() {
void* dnodeGetMpeerInfos() {
return &tsMnodeInfos;
}
static void dnodeSendStatusMsg(void *handle, void *tmrId) {
if (tsDnodeTmr == NULL) {
dError("dnode timer is already released");
return;
}
if (tsStatusTimer == NULL) {
taosTmrReset(dnodeSendStatusMsg, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);
dError("failed to start status timer");
return;
}
int32_t contLen = sizeof(SDMStatusMsg) + TSDB_MAX_VNODES * sizeof(SVnodeLoad);
SDMStatusMsg *pStatus = rpcMallocCont(contLen);
if (pStatus == NULL) {
taosTmrReset(dnodeSendStatusMsg, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);
dError("failed to malloc status message");
return;
}
strcpy(pStatus->dnodeName, tsDnodeName);
pStatus->version = htonl(tsVersion);
pStatus->dnodeId = htonl(tsDnodeId);
pStatus->privateIp = htonl(inet_addr(tsPrivateIp));
pStatus->publicIp = htonl(inet_addr(tsPublicIp));
pStatus->lastReboot = htonl(tsRebootTime);
pStatus->numOfTotalVnodes = htons((uint16_t) tsNumOfTotalVnodes);
pStatus->numOfCores = htons((uint16_t) tsNumOfCores);
pStatus->diskAvailable = tsAvailDataDirGB;
pStatus->alternativeRole = (uint8_t) tsAlternativeRole;
vnodeBuildStatusMsg(pStatus);
contLen = sizeof(SDMStatusMsg) + pStatus->openVnodes * sizeof(SVnodeLoad);
pStatus->openVnodes = htons(pStatus->openVnodes);
SRpcMsg rpcMsg = {
.pCont = pStatus,
.contLen = contLen,
.msgType = TSDB_MSG_TYPE_DM_STATUS
};
dnodeSendMsgToMnode(&rpcMsg);
}
static void dnodeReadDnodeInfo() {
char dnodeIdFile[TSDB_FILENAME_LEN] = {0};
sprintf(dnodeIdFile, "%s/dnodeId", tsDnodeDir);
FILE *fp = fopen(dnodeIdFile, "r");
if (!fp) return;
char option[32] = {0};
int32_t value = 0;
int32_t num = 0;
num = fscanf(fp, "%s %d", option, &value);
if (num != 2) return;
if (strcmp(option, "dnodeId") != 0) return;
tsDnodeId = value;;
fclose(fp);
dPrint("read dnodeId:%d successed", tsDnodeId);
}
static void dnodeSaveDnodeInfo() {
char dnodeIdFile[TSDB_FILENAME_LEN] = {0};
sprintf(dnodeIdFile, "%s/dnodeId", tsDnodeDir);
FILE *fp = fopen(dnodeIdFile, "w");
if (!fp) return;
fprintf(fp, "dnodeId %d\n", tsDnodeId);
fclose(fp);
dPrint("save dnodeId successed");
}
void dnodeUpdateDnodeInfo(int32_t dnodeId) {
if (tsDnodeId == 0) {
dPrint("dnodeId is set to %d", dnodeId);
tsDnodeId = dnodeId;
dnodeSaveDnodeInfo();
}
}
int32_t dnodeGetDnodeId() {
return tsDnodeId;
}
\ No newline at end of file
src/dnode/src/dnodeMain.c
浏览文件 @ f73bc53b
......@@ -159,10 +159,10 @@ static int32_t dnodeInitSystem() {
dPrint("starting to initialize TDengine ...");
if (dnodeInitStorage() != 0) return -1;
if (dnodeInitModules() != 0) return -1;
if (dnodeInitRead() != 0) return -1;
if (dnodeInitWrite() != 0) return -1;
if (dnodeInitMClient() != 0) return -1;
if (dnodeInitModules() != 0) return -1;
if (dnodeInitMnode() != 0) return -1;
if (dnodeInitMgmt() != 0) return -1;
if (dnodeInitShell() != 0) return -1;
......@@ -177,7 +177,6 @@ static int32_t dnodeInitSystem() {
static void dnodeCleanUpSystem() {
if (dnodeGetRunStatus() != TSDB_DNODE_RUN_STATUS_STOPPED) {
tclearModuleStatus(TSDB_MOD_MGMT);
dnodeSetRunStatus(TSDB_DNODE_RUN_STATUS_STOPPED);
dnodeCleanupShell();
dnodeCleanupMnode();
......
src/dnode/src/dnodeMgmt.c
浏览文件 @ f73bc53b
......@@ -21,8 +21,6 @@
#include "tlog.h"
#include "trpc.h"
#include "tsdb.h"
#include "ttime.h"
#include "ttimer.h"
#include "twal.h"
#include "dnodeMClient.h"
#include "dnodeMgmt.h"
......@@ -38,52 +36,23 @@ static int32_t dnodeProcessAlterVnodeMsg(SRpcMsg *pMsg);
static int32_t dnodeProcessAlterStreamMsg(SRpcMsg *pMsg);
static int32_t dnodeProcessConfigDnodeMsg(SRpcMsg *pMsg);
static int32_t (*dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *pMsg);
static void dnodeSendStatusMsg(void *handle, void *tmrId);
static void dnodeReadDnodeId();
static void *tsDnodeTmr = NULL;
static void *tsStatusTimer = NULL;
static uint32_t tsRebootTime;
static int32_t tsDnodeId = 0;
static char tsDnodeName[TSDB_NODE_NAME_LEN];
int32_t dnodeInitMgmt() {
dnodeReadDnodeId();
dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_CREATE_VNODE] = dnodeProcessCreateVnodeMsg;
dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_DROP_VNODE] = dnodeProcessDropVnodeMsg;
dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_ALTER_VNODE] = dnodeProcessAlterVnodeMsg;
dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_ALTER_STREAM] = dnodeProcessAlterStreamMsg;
dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_CONFIG_DNODE] = dnodeProcessConfigDnodeMsg;
tsRebootTime = taosGetTimestampSec();
tsDnodeTmr = taosTmrInit(100, 200, 60000, "DND-DM");
if (tsDnodeTmr == NULL) {
dError("failed to init dnode timer");
return -1;
}
int32_t code = dnodeOpenVnodes();
if (code != TSDB_CODE_SUCCESS) {
return -1;
}
taosTmrReset(dnodeSendStatusMsg, 500, NULL, tsDnodeTmr, &tsStatusTimer);
return TSDB_CODE_SUCCESS;
}
void dnodeCleanupMgmt() {
if (tsStatusTimer != NULL) {
taosTmrStopA(&tsStatusTimer);
tsStatusTimer = NULL;
}
if (tsDnodeTmr != NULL) {
taosTmrCleanUp(tsDnodeTmr);
tsDnodeTmr = NULL;
}
dnodeCloseVnodes();
}
......@@ -213,89 +182,3 @@ static int32_t dnodeProcessConfigDnodeMsg(SRpcMsg *pMsg) {
SMDCfgDnodeMsg *pCfg = (SMDCfgDnodeMsg *)pMsg->pCont;
return tsCfgDynamicOptions(pCfg->config);
}
static void dnodeSendStatusMsg(void *handle, void *tmrId) {
if (tsDnodeTmr == NULL) {
dError("dnode timer is already released");
return;
}
if (tsStatusTimer == NULL) {
taosTmrReset(dnodeSendStatusMsg, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);
dError("failed to start status timer");
return;
}
int32_t contLen = sizeof(SDMStatusMsg) + TSDB_MAX_VNODES * sizeof(SVnodeLoad);
SDMStatusMsg *pStatus = rpcMallocCont(contLen);
if (pStatus == NULL) {
taosTmrReset(dnodeSendStatusMsg, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);
dError("failed to malloc status message");
return;
}
strcpy(pStatus->dnodeName, tsDnodeName);
pStatus->version = htonl(tsVersion);
pStatus->dnodeId = htonl(tsDnodeId);
pStatus->privateIp = htonl(inet_addr(tsPrivateIp));
pStatus->publicIp = htonl(inet_addr(tsPublicIp));
pStatus->lastReboot = htonl(tsRebootTime);
pStatus->numOfTotalVnodes = htons((uint16_t) tsNumOfTotalVnodes);
pStatus->numOfCores = htons((uint16_t) tsNumOfCores);
pStatus->diskAvailable = tsAvailDataDirGB;
pStatus->alternativeRole = (uint8_t) tsAlternativeRole;
vnodeBuildStatusMsg(pStatus);
contLen = sizeof(SDMStatusMsg) + pStatus->openVnodes * sizeof(SVnodeLoad);
pStatus->openVnodes = htons(pStatus->openVnodes);
SRpcMsg rpcMsg = {
.pCont = pStatus,
.contLen = contLen,
.msgType = TSDB_MSG_TYPE_DM_STATUS
};
dnodeSendMsgToMnode(&rpcMsg);
taosTmrReset(dnodeSendStatusMsg, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);
}
static void dnodeReadDnodeId() {
char dnodeIdFile[TSDB_FILENAME_LEN] = {0};
sprintf(dnodeIdFile, "%s/dnodeId", tsDnodeDir);
FILE *fp = fopen(dnodeIdFile, "r");
if (!fp) return;
char option[32] = {0};
int32_t value = 0;
int32_t num = 0;
num = fscanf(fp, "%s %d", option, &value);
if (num != 2) return;
if (strcmp(option, "dnodeId") != 0) return;
tsDnodeId = value;;
fclose(fp);
dPrint("read dnodeId:%d successed", tsDnodeId);
}
static void dnodeSaveDnodeId() {
char dnodeIdFile[TSDB_FILENAME_LEN] = {0};
sprintf(dnodeIdFile, "%s/dnodeId", tsDnodeDir);
FILE *fp = fopen(dnodeIdFile, "w");
if (!fp) return;
fprintf(fp, "dnodeId %d\n", tsDnodeId);
fclose(fp);
dPrint("save dnodeId successed");
}
void dnodeUpdateDnodeId(int32_t dnodeId) {
if (tsDnodeId == 0) {
dPrint("dnodeId is set to %d", dnodeId);
tsDnodeId = dnodeId;
dnodeSaveDnodeId();
}
}
src/inc/mnode.h
浏览文件 @ f73bc53b
......@@ -45,16 +45,14 @@ struct _mnode_obj;
typedef struct _mnode_obj {
int32_t mnodeId;
int32_t dnodeId;
int64_t createdTime;
int8_t reserved[14];
int8_t updateEnd[1];
int32_t refCount;
int8_t role;
int8_t status;
uint16_t port;
uint32_t privateIp;
uint32_t publicIp;
uint16_t port;
int8_t role;
char mnodeName[TSDB_NODE_NAME_LEN + 1];
} SMnodeObj;
......
src/inc/mpeer.h
浏览文件 @ f73bc53b
......@@ -28,22 +28,29 @@ enum _TAOS_MN_STATUS {
TAOS_MN_STATUS_READY
};
// general implementation
int32_t mpeerInit();
void mpeerCleanup();
// special implementation
int32_t mpeerInitMnodes();
void mpeerCleanupMnodes();
int32_t mpeerAddMnode(int32_t dnodeId);
int32_t mpeerRemoveMnode(int32_t dnodeId);
void * mpeerGetMnode(int32_t mnodeId);
int32_t mpeerGetMnodesNum();
void * mpeerGetNextMnode(void *pNode, struct _mnode_obj **pMnode);
void mpeerReleaseMnode(struct _mnode_obj *pMnode);
bool mpeerInServerStatus();
bool mpeerIsMaster();
bool mpeerCheckRedirect();
void mpeerGetPrivateIpList(SRpcIpSet *ipSet);
void mpeerGetPublicIpList(SRpcIpSet *ipSet);
void mpeerGetMpeerInfos(void *mpeers);
char * mpeerGetMnodeStatusStr(int32_t status);
char * mpeerGetMnodeRoleStr(int32_t role);
int32_t mpeerForwardReqToPeer(void *pHead);
void mpeerUpdateSync();
#ifdef __cplusplus
}
......
src/inc/tbalance.h
浏览文件 @ f73bc53b
......@@ -31,6 +31,7 @@ struct _dnode_obj;
int32_t balanceInit();
void balanceCleanUp();
void balanceNotify();
void balanceReset();
int32_t balanceAllocVnodes(struct _vg_obj *pVgroup);
int32_t balanceDropDnode(struct _dnode_obj *pDnode);
......
src/inc/tcluster.h
浏览文件 @ f73bc53b
......@@ -37,6 +37,7 @@ int32_t clusterInit();
void clusterCleanUp();
char* clusterGetDnodeStatusStr(int32_t dnodeStatus);
bool clusterCheckModuleInDnode(struct _dnode_obj *pDnode, int moduleType);
void clusterMonitorDnodeModule();
int32_t clusterInitDnodes();
void clusterCleanupDnodes();
......
src/mnode/inc/mgmtSdb.h
浏览文件 @ f73bc53b
......@@ -21,8 +21,8 @@ extern "C" {
#endif
typedef enum {
SDB_TABLE_MNODE = 0,
SDB_TABLE_DNODE = 1,
SDB_TABLE_DNODE = 0,
SDB_TABLE_MNODE = 1,
SDB_TABLE_ACCOUNT = 2,
SDB_TABLE_USER = 3,
SDB_TABLE_DB = 4,
......@@ -34,6 +34,7 @@ typedef enum {
typedef enum {
SDB_KEY_STRING,
SDB_KEY_INT,
SDB_KEY_AUTO
} ESdbKeyType;
......@@ -63,14 +64,22 @@ typedef struct {
int32_t (*encodeFp)(SSdbOperDesc *pOper);
int32_t (*decodeFp)(SSdbOperDesc *pDesc);
int32_t (*destroyFp)(SSdbOperDesc *pDesc);
int32_t (*updateAllFp)();
int32_t (*restoredFp)();
} SSdbTableDesc;
typedef struct {
int64_t version;
void * wal;
pthread_mutex_t mutex;
} SSdbObject;
int32_t sdbInit();
void sdbCleanUp();
SSdbObject *sdbGetObj();
void * sdbOpenTable(SSdbTableDesc *desc);
void sdbCloseTable(void *handle);
int sdbProcessWrite(void *param, void *data, int type);
int32_t sdbInsertRow(SSdbOperDesc *pOper);
int32_t sdbDeleteRow(SSdbOperDesc *pOper);
......@@ -81,7 +90,7 @@ void *sdbFetchRow(void *handle, void *pNode, void **ppRow);
void sdbIncRef(void *thandle, void *pRow);
void sdbDecRef(void *thandle, void *pRow);
int64_t sdbGetNumOfRows(void *handle);
int64_t sdbGetId(void *handle);
int32_t sdbGetId(void *handle);
uint64_t sdbGetVersion();
#ifdef __cplusplus
......
src/mnode/src/mgmtDb.c
浏览文件 @ f73bc53b
......@@ -102,7 +102,7 @@ static int32_t mgmtDbActionDecode(SSdbOperDesc *pOper) {
return TSDB_CODE_SUCCESS;
}
static int32_t mgmtDbActionUpdateAll() {
static int32_t mgmtDbActionRestored() {
return 0;
}
......@@ -123,7 +123,7 @@ int32_t mgmtInitDbs() {
.encodeFp = mgmtDbActionEncode,
.decodeFp = mgmtDbActionDecode,
.destroyFp = mgmtDbActionDestroy,
.updateAllFp = mgmtDbActionUpdateAll
.restoredFp = mgmtDbActionRestored
};
tsDbSdb = sdbOpenTable(&tableDesc);
......
src/mnode/src/mgmtDnode.c
浏览文件 @ f73bc53b
......@@ -77,6 +77,7 @@ void * clusterGetDnode(int32_t dnodeId) { return dnodeId == 1 ? &tsDnodeObj : N
void * clusterGetDnodeByIp(uint32_t ip) { return &tsDnodeObj; }
void clusterReleaseDnode(struct _dnode_obj *pDnode) {}
void clusterUpdateDnode(struct _dnode_obj *pDnode) {}
void clusterMonitorDnodeModule() {}
#endif
......@@ -208,6 +209,7 @@ void clusterProcessDnodeStatusMsg(SRpcMsg *rpcMsg) {
mTrace("dnode:%d, from offline to online", pDnode->dnodeId);
pDnode->status = TAOS_DN_STATUS_READY;
balanceNotify();
clusterMonitorDnodeModule();
}
clusterReleaseDnode(pDnode);
......
src/mnode/src/mgmtMain.c
浏览文件 @ f73bc53b
......@@ -109,6 +109,11 @@ int32_t mgmtStartSystem() {
return -1;
}
if (mpeerInit() < 0) {
mError("failed to init mpeers");
return -1;
}
if (sdbInit() < 0) {
mError("failed to init sdb");
return -1;
......@@ -122,11 +127,6 @@ int32_t mgmtStartSystem() {
return -1;
}
if (mpeerInit() < 0) {
mError("failed to init mpeers");
return -1;
}
if (balanceInit() < 0) {
mError("failed to init dnode balance")
}
......
src/mnode/src/mgmtMnode.c
浏览文件 @ f73bc53b
......@@ -19,12 +19,9 @@
#include "trpc.h"
#include "tsync.h"
#include "mpeer.h"
#include "mgmtSdb.h"
#include "mgmtShell.h"
#include "mgmtUser.h"
extern int32_t mpeerInitMnodes();
extern void mpeerCleanupMnodes();
static int32_t mgmtGetMnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn);
......@@ -34,18 +31,25 @@ static SMnodeObj tsMnodeObj = {0};
int32_t mpeerInitMnodes() {
tsMnodeObj.mnodeId = 1;
tsMnodeObj.dnodeId = 1;
tsMnodeObj.privateIp = inet_addr(tsPrivateIp);
tsMnodeObj.publicIp = inet_addr(tsPublicIp);
tsMnodeObj.createdTime = taosGetTimestampMs();
tsMnodeObj.role = TAOS_SYNC_ROLE_MASTER;
tsMnodeObj.status = TAOS_MN_STATUS_READY;
tsMnodeObj.port = tsMnodeDnodePort;
sprintf(tsMnodeObj.mnodeName, "m%d", tsMnodeObj.mnodeId);
return TSDB_CODE_SUCCESS;
}
void mpeerCleanupMnodes() {}
int32_t mpeerAddMnode(int32_t dnodeId) { return TSDB_CODE_SUCCESS; }
int32_t mpeerRemoveMnode(int32_t dnodeId) { return TSDB_CODE_SUCCESS; }
void * mpeerGetMnode(int32_t mnodeId) { return &tsMnodeObj; }
int32_t mpeerGetMnodesNum() { return 1; }
void mpeerReleaseMnode(struct _mnode_obj *pMnode) {}
bool mpeerIsMaster() { return tsMnodeObj.role == TAOS_SYNC_ROLE_MASTER; }
void mpeerUpdateSync() {}
void *mpeerGetNextMnode(void *pNode, SMnodeObj **pMnode) {
if (*pMnode == NULL) {
*pMnode = &tsMnodeObj;
......@@ -58,20 +62,21 @@ void *mpeerGetNextMnode(void *pNode, SMnodeObj **pMnode) {
void mpeerGetPrivateIpList(SRpcIpSet *ipSet) {
ipSet->inUse = 0;
ipSet->port = htons(tsMnodeDnodePort);
ipSet->numOfIps = 1;
ipSet->port = htons(tsMnodeObj.port);
ipSet->ip[0] = htonl(tsMnodeObj.privateIp);
}
void mpeerGetPublicIpList(SRpcIpSet *ipSet) {
ipSet->inUse = 0;
ipSet->port = htons(tsMnodeDnodePort);
ipSet->numOfIps = 1;
ipSet->port = htons(tsMnodeObj.port);
ipSet->ip[0] = htonl(tsMnodeObj.publicIp);
}
void mpeerGetMpeerInfos(void *param) {
SDMNodeInfos *mpeers = param;
mpeers->inUse = 0;
mpeers->nodeNum = 1;
mpeers->nodeInfos[0].nodeId = htonl(tsMnodeObj.mnodeId);
mpeers->nodeInfos[0].nodeIp = htonl(tsMnodeObj.privateIp);
......@@ -79,40 +84,23 @@ void mpeerGetMpeerInfos(void *param) {
strcpy(mpeers->nodeInfos[0].nodeName, tsMnodeObj.mnodeName);
}
void mpeerCleanupDnodes() {}
int32_t mpeerGetMnodesNum() { return 1; }
void mpeerReleaseMnode(struct _mnode_obj *pMnode) {}
bool mpeerInServerStatus() { return tsMnodeObj.status == TAOS_MN_STATUS_READY; }
bool mpeerIsMaster() { return tsMnodeObj.role == TAOS_SYNC_ROLE_MASTER; }
bool mpeerCheckRedirect() { return false; }
int32_t mpeerForwardReqToPeer(void *pHead) {
return TSDB_CODE_SUCCESS;
}
#endif
int32_t mpeerInit() {
mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_MNODE, mgmtGetMnodeMeta);
mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_MNODE, mgmtRetrieveMnodes);
return mpeerInitMnodes();
}
void mpeerCleanup() {
mpeerCleanupDnodes();
}
char *mpeerGetMnodeStatusStr(int32_t status) {
switch (status) {
case TAOS_MN_STATUS_OFFLINE:
return "offline";
case TAOS_MN_STATUS_DROPPING:
return "dropping";
case TAOS_MN_STATUS_READY:
return "ready";
default:
return "undefined";
}
mpeerCleanupMnodes();
}
char *mpeerGetMnodeRoleStr(int32_t role) {
static char *mpeerGetMnodeRoleStr(int32_t role) {
switch (role) {
case TAOS_SYNC_ROLE_OFFLINE:
return "offline";
......@@ -160,12 +148,6 @@ static int32_t mgmtGetMnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pCo
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 10;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "status");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 10;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "role");
......@@ -220,14 +202,12 @@ static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, voi
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
strcpy(pWrite, mpeerGetMnodeStatusStr(pMnode->status));
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
strcpy(pWrite, mpeerGetMnodeRoleStr(pMnode->role));
cols++;
numOfRows++;
mpeerReleaseMnode(pMnode);
}
pShow->numOfReads += numOfRows;
......
src/mnode/src/mgmtSdb.c
浏览文件 @ f73bc53b
......@@ -15,28 +15,16 @@
#define _DEFAULT_SOURCE
#include "os.h"
#include "taosdef.h"
#include "taoserror.h"
#include "tchecksum.h"
#include "tglobalcfg.h"
#include "tlog.h"
#include "trpc.h"
#include "tutil.h"
#include "tqueue.h"
#include "twal.h"
#include "tsync.h"
#include "hashint.h"
#include "hashstr.h"
#include "mpeer.h"
#include "mgmtSdb.h"
typedef struct {
int32_t code;
int64_t version;
void * sync;
void * wal;
sem_t sem;
pthread_mutex_t mutex;
} SSdbSync;
typedef struct _SSdbTable {
char tableName[TSDB_DB_NAME_LEN + 1];
ESdbTable tableId;
......@@ -47,13 +35,13 @@ typedef struct _SSdbTable {
int32_t autoIndex;
int64_t numOfRows;
void * iHandle;
int32_t (*insertFp)(SSdbOperDesc *pDesc);
int32_t (*deleteFp)(SSdbOperDesc *pOper);
int32_t (*updateFp)(SSdbOperDesc *pOper);
int32_t (*decodeFp)(SSdbOperDesc *pOper);
int32_t (*encodeFp)(SSdbOperDesc *pOper);
int32_t (*destroyFp)(SSdbOperDesc *pOper);
int32_t (*updateAllFp)();
int32_t (*insertFp)(SSdbOperDesc *pDesc);
int32_t (*deleteFp)(SSdbOperDesc *pOper);
int32_t (*updateFp)(SSdbOperDesc *pOper);
int32_t (*decodeFp)(SSdbOperDesc *pOper);
int32_t (*encodeFp)(SSdbOperDesc *pOper);
int32_t (*destroyFp)(SSdbOperDesc *pOper);
int32_t (*restoredFp)();
pthread_mutex_t mutex;
} SSdbTable;
......@@ -70,18 +58,17 @@ typedef enum {
static SSdbTable *tsSdbTableList[SDB_TABLE_MAX] = {0};
static int32_t tsSdbNumOfTables = 0;
static SSdbSync * tsSdbSync;
static void *(*sdbInitIndexFp[])(int32_t maxRows, int32_t dataSize) = {sdbOpenStrHash, sdbOpenIntHash};
static void *(*sdbAddIndexFp[])(void *handle, void *key, void *data) = {sdbAddStrHash, sdbAddIntHash};
static void (*sdbDeleteIndexFp[])(void *handle, void *key) = {sdbDeleteStrHash, sdbDeleteIntHash};
static void *(*sdbGetIndexFp[])(void *handle, void *key) = {sdbGetStrHashData, sdbGetIntHashData};
static void (*sdbCleanUpIndexFp[])(void *handle) = {sdbCloseStrHash, sdbCloseIntHash};
static void *(*sdbFetchRowFp[])(void *handle, void *ptr, void **ppRow) = {sdbFetchStrHashData, sdbFetchIntHashData};
static int sdbProcessWrite(void *param, void *data, int type);
uint64_t sdbGetVersion() { return tsSdbSync->version; }
int64_t sdbGetId(void *handle) { return ((SSdbTable *)handle)->autoIndex; }
static SSdbObject * tsSdbObj;
static void *(*sdbInitIndexFp[])(int32_t maxRows, int32_t dataSize) = {sdbOpenStrHash, sdbOpenIntHash, sdbOpenIntHash};
static void *(*sdbAddIndexFp[])(void *handle, void *key, void *data) = {sdbAddStrHash, sdbAddIntHash, sdbAddIntHash};
static void (*sdbDeleteIndexFp[])(void *handle, void *key) = {sdbDeleteStrHash, sdbDeleteIntHash, sdbDeleteIntHash};
static void *(*sdbGetIndexFp[])(void *handle, void *key) = {sdbGetStrHashData, sdbGetIntHashData, sdbGetIntHashData};
static void (*sdbCleanUpIndexFp[])(void *handle) = {sdbCloseStrHash, sdbCloseIntHash, sdbCloseIntHash};
static void *(*sdbFetchRowFp[])(void *handle, void *ptr, void **ppRow) = {sdbFetchStrHashData, sdbFetchIntHashData, sdbFetchIntHashData};
uint64_t sdbGetVersion() { return tsSdbObj->version; }
int32_t sdbGetId(void *handle) { return ((SSdbTable *)handle)->autoIndex; }
int64_t sdbGetNumOfRows(void *handle) { return ((SSdbTable *)handle)->numOfRows; }
static char *sdbGetActionStr(int32_t action) {
......@@ -101,6 +88,7 @@ static char *sdbGetkeyStr(SSdbTable *pTable, void *row) {
switch (pTable->keyType) {
case SDB_KEY_STRING:
return (char *)row;
case SDB_KEY_INT:
case SDB_KEY_AUTO:
sprintf(str, "%d", *(int32_t *)row);
return str;
......@@ -113,44 +101,27 @@ static void *sdbGetTableFromId(int32_t tableId) {
return tsSdbTableList[tableId];
}
// static void mpeerConfirmForward(void *ahandle, void *param, int32_t code) {
// sem_post(&tsSdbSync->sem);
// mPrint("mpeerConfirmForward");
// }
static int32_t sdbForwardDbReqToPeer(SWalHead *pHead) {
// int32_t code = syncForwardToPeer(NULL, pHead, NULL);
// if (code < 0) {
// return code;
// }
// sem_wait(&tsSdbSync->sem);
// return tsSdbSync->code;
return TSDB_CODE_SUCCESS;
}
int32_t sdbInit() {
tsSdbSync = calloc(1, sizeof(SSdbSync));
sem_init(&tsSdbSync->sem, 0, 0);
pthread_mutex_init(&tsSdbSync->mutex, NULL);
tsSdbObj = calloc(1, sizeof(SSdbObject));
pthread_mutex_init(&tsSdbObj->mutex, NULL);
SWalCfg walCfg = {.commitLog = 2, .wals = 2, .keep = 1};
tsSdbSync->wal = walOpen(tsMnodeDir, &walCfg);
if (tsSdbSync->wal == NULL) {
tsSdbObj->wal = walOpen(tsMnodeDir, &walCfg);
if (tsSdbObj->wal == NULL) {
sdbError("failed to open sdb in %s", tsMnodeDir);
return -1;
}
sdbTrace("open sdb file for read");
walRestore(tsSdbSync->wal, tsSdbSync, sdbProcessWrite);
walRestore(tsSdbObj->wal, tsSdbObj, sdbProcessWrite);
int32_t totalRows = 0;
int32_t numOfTables = 0;
for (int32_t tableId = SDB_TABLE_DNODE; tableId < SDB_TABLE_MAX; ++tableId) {
SSdbTable *pTable = sdbGetTableFromId(tableId);
if (pTable == NULL) continue;
if (pTable->updateAllFp) {
(*pTable->updateAllFp)();
if (pTable->restoredFp) {
(*pTable->restoredFp)();
}
totalRows += pTable->numOfRows;
......@@ -158,20 +129,26 @@ int32_t sdbInit() {
sdbTrace("table:%s, is initialized, numOfRows:%d", pTable->tableName, pTable->numOfRows);
}
sdbTrace("sdb is initialized, version:%d totalRows:%d numOfTables:%d", tsSdbSync->version, totalRows, numOfTables);
sdbTrace("sdb is initialized, version:%d totalRows:%d numOfTables:%d", tsSdbObj->version, totalRows, numOfTables);
mpeerUpdateSync();
return TSDB_CODE_SUCCESS;
}
void sdbCleanUp() {
if (tsSdbSync) {
sem_destroy(&tsSdbSync->sem);
pthread_mutex_destroy(&tsSdbSync->mutex);
walClose(tsSdbSync->wal);
free(tsSdbSync);
tsSdbSync = NULL;
if (tsSdbObj) {
pthread_mutex_destroy(&tsSdbObj->mutex);
walClose(tsSdbObj->wal);
free(tsSdbObj);
tsSdbObj = NULL;
}
}
SSdbObject *sdbGetObj() {
return tsSdbObj;
}
void sdbIncRef(void *handle, void *pRow) {
if (pRow) {
SSdbTable *pTable = handle;
......@@ -241,6 +218,11 @@ static int32_t sdbInsertLocal(SSdbTable *pTable, SSdbOperDesc *pOper) {
(*sdbAddIndexFp[pTable->keyType])(pTable->iHandle, pOper->pObj, &rowMeta);
sdbIncRef(pTable, pOper->pObj);
pTable->numOfRows++;
if (pTable->keyType == SDB_KEY_AUTO) {
pTable->autoIndex = MAX(pTable->autoIndex, *((uint32_t *)pOper->pObj));
}
pthread_mutex_unlock(&pTable->mutex);
sdbTrace("table:%s, insert record:%s, numOfRows:%d", pTable->tableName, sdbGetkeyStr(pTable, pOper->pObj),
......@@ -278,20 +260,20 @@ static int32_t sdbUpdateLocal(SSdbTable *pTable, SSdbOperDesc *pOper) {
static int32_t sdbProcessWriteFromApp(SSdbTable *pTable, SWalHead *pHead, int32_t action) {
int32_t code = 0;
pthread_mutex_lock(&tsSdbSync->mutex);
tsSdbSync->version++;
pHead->version = tsSdbSync->version;
pthread_mutex_lock(&tsSdbObj->mutex);
tsSdbObj->version++;
pHead->version = tsSdbObj->version;
code = sdbForwardDbReqToPeer(pHead);
code = mpeerForwardReqToPeer(pHead);
if (code != TSDB_CODE_SUCCESS) {
pthread_mutex_unlock(&tsSdbSync->mutex);
pthread_mutex_unlock(&tsSdbObj->mutex);
sdbError("table:%s, failed to forward %s record:%s from file, version:%" PRId64 ", reason:%s", pTable->tableName,
sdbGetActionStr(action), sdbGetkeyStr(pTable, pHead->cont), pHead->version, tstrerror(code));
return code;
}
code = walWrite(tsSdbSync->wal, pHead);
pthread_mutex_unlock(&tsSdbSync->mutex);
code = walWrite(tsSdbObj->wal, pHead);
pthread_mutex_unlock(&tsSdbObj->mutex);
if (code < 0) {
sdbError("table:%s, failed to %s record:%s to file, version:%" PRId64 ", reason:%s", pTable->tableName,
......@@ -301,26 +283,25 @@ static int32_t sdbProcessWriteFromApp(SSdbTable *pTable, SWalHead *pHead, int32_
sdbGetkeyStr(pTable, pHead->cont), pHead->version);
}
walFsync(tsSdbSync->wal);
free(pHead);
walFsync(tsSdbObj->wal);
taosFreeQitem(pHead);
return code;
}
static int32_t sdbProcessWriteFromWal(SSdbTable *pTable, SWalHead *pHead, int32_t action) {
pthread_mutex_lock(&tsSdbSync->mutex);
if (pHead->version <= tsSdbSync->version) {
pthread_mutex_unlock(&tsSdbSync->mutex);
pthread_mutex_lock(&tsSdbObj->mutex);
if (pHead->version <= tsSdbObj->version) {
pthread_mutex_unlock(&tsSdbObj->mutex);
return TSDB_CODE_SUCCESS;
} else if (pHead->version != tsSdbSync->version + 1) {
pthread_mutex_unlock(&tsSdbSync->mutex);
} else if (pHead->version != tsSdbObj->version + 1) {
pthread_mutex_unlock(&tsSdbObj->mutex);
sdbError("table:%s, failed to restore %s record:%s from file, version:%" PRId64 " too large, sdb version:%" PRId64,
pTable->tableName, sdbGetActionStr(action), sdbGetkeyStr(pTable, pHead->cont), pHead->version,
tsSdbSync->version);
tsSdbObj->version);
return TSDB_CODE_OTHERS;
}
tsSdbSync->version = pHead->version;
tsSdbObj->version = pHead->version;
sdbTrace("table:%s, success to restore %s record:%s from file, version:%" PRId64, pTable->tableName,
sdbGetActionStr(action), sdbGetkeyStr(pTable, pHead->cont), pHead->version);
......@@ -335,7 +316,7 @@ static int32_t sdbProcessWriteFromWal(SSdbTable *pTable, SWalHead *pHead, int32_
if (code < 0) {
sdbTrace("table:%s, failed to decode %s record:%s from file, version:%" PRId64, pTable->tableName,
sdbGetActionStr(action), sdbGetkeyStr(pTable, pHead->cont), pHead->version);
pthread_mutex_unlock(&tsSdbSync->mutex);
pthread_mutex_unlock(&tsSdbObj->mutex);
return code;
}
......@@ -369,17 +350,17 @@ static int32_t sdbProcessWriteFromWal(SSdbTable *pTable, SWalHead *pHead, int32_
if (code < 0) {
sdbTrace("table:%s, failed to decode %s record:%s from file, version:%" PRId64, pTable->tableName,
sdbGetActionStr(action), sdbGetkeyStr(pTable, pHead->cont), pHead->version);
pthread_mutex_unlock(&tsSdbSync->mutex);
pthread_mutex_unlock(&tsSdbObj->mutex);
return code;
}
code = sdbInsertLocal(pTable, &oper2);
}
pthread_mutex_unlock(&tsSdbSync->mutex);
pthread_mutex_unlock(&tsSdbObj->mutex);
return code;
}
static int sdbProcessWrite(void *param, void *data, int type) {
int sdbProcessWrite(void *param, void *data, int type) {
SWalHead *pHead = data;
int32_t tableId = pHead->msgType / 10;
int32_t action = pHead->msgType % 10;
......@@ -417,7 +398,7 @@ int32_t sdbInsertRow(SSdbOperDesc *pOper) {
if (pOper->type == SDB_OPER_GLOBAL) {
int32_t size = sizeof(SWalHead) + pTable->maxRowSize;
SWalHead *pHead = calloc(1, size);
SWalHead *pHead = taosAllocateQitem(size);
pHead->version = 0;
pHead->len = pOper->rowSize;
pHead->msgType = pTable->tableId * 10 + SDB_ACTION_INSERT;
......@@ -426,7 +407,7 @@ int32_t sdbInsertRow(SSdbOperDesc *pOper) {
(*pTable->encodeFp)(pOper);
pHead->len = pOper->rowSize;
int32_t code = sdbProcessWrite(tsSdbSync, pHead, pHead->msgType);
int32_t code = sdbProcessWrite(tsSdbObj, pHead, pHead->msgType);
if (code < 0) return code;
}
......@@ -453,6 +434,7 @@ int32_t sdbDeleteRow(SSdbOperDesc *pOper) {
case SDB_KEY_STRING:
rowSize = strlen((char *)pOper->pObj) + 1;
break;
case SDB_KEY_INT:
case SDB_KEY_AUTO:
rowSize = sizeof(uint64_t);
break;
......@@ -461,13 +443,13 @@ int32_t sdbDeleteRow(SSdbOperDesc *pOper) {
}
int32_t size = sizeof(SWalHead) + rowSize;
SWalHead *pHead = calloc(1, size);
SWalHead *pHead = taosAllocateQitem(size);
pHead->version = 0;
pHead->len = rowSize;
pHead->msgType = pTable->tableId * 10 + SDB_ACTION_DELETE;
memcpy(pHead->cont, pOper->pObj, rowSize);
int32_t code = sdbProcessWrite(tsSdbSync, pHead, pHead->msgType);
int32_t code = sdbProcessWrite(tsSdbObj, pHead, pHead->msgType);
if (code < 0) return code;
}
......@@ -489,7 +471,7 @@ int32_t sdbUpdateRow(SSdbOperDesc *pOper) {
if (pOper->type == SDB_OPER_GLOBAL) {
int32_t size = sizeof(SWalHead) + pTable->maxRowSize;
SWalHead *pHead = calloc(1, size);
SWalHead *pHead = taosAllocateQitem(size);
pHead->version = 0;
pHead->msgType = pTable->tableId * 10 + SDB_ACTION_UPDATE;
......@@ -497,7 +479,7 @@ int32_t sdbUpdateRow(SSdbOperDesc *pOper) {
(*pTable->encodeFp)(pOper);
pHead->len = pOper->rowSize;
int32_t code = sdbProcessWrite(tsSdbSync, pHead, pHead->msgType);
int32_t code = sdbProcessWrite(tsSdbObj, pHead, pHead->msgType);
if (code < 0) return code;
}
......@@ -522,6 +504,7 @@ void *sdbFetchRow(void *handle, void *pNode, void **ppRow) {
void *sdbOpenTable(SSdbTableDesc *pDesc) {
SSdbTable *pTable = (SSdbTable *)calloc(1, sizeof(SSdbTable));
if (pTable == NULL) return NULL;
strcpy(pTable->tableName, pDesc->tableName);
......@@ -536,7 +519,7 @@ void *sdbOpenTable(SSdbTableDesc *pDesc) {
pTable->encodeFp = pDesc->encodeFp;
pTable->decodeFp = pDesc->decodeFp;
pTable->destroyFp = pDesc->destroyFp;
pTable->updateAllFp = pDesc->updateAllFp;
pTable->restoredFp = pDesc->restoredFp;
if (sdbInitIndexFp[pTable->keyType] != NULL) {
pTable->iHandle = (*sdbInitIndexFp[pTable->keyType])(pTable->maxRowSize, sizeof(SRowMeta));
......@@ -575,7 +558,7 @@ void sdbCloseTable(void *handle) {
}
pthread_mutex_destroy(&pTable->mutex);
sdbTrace("table:%s, is closed, numOfTables:%d", pTable->tableName, tsSdbNumOfTables);
free(pTable);
}
src/mnode/src/mgmtShell.c
浏览文件 @ f73bc53b
......@@ -42,7 +42,6 @@ static int mgmtShellRetriveAuth(char *user, char *spi, char *encrypt, char *sec
static bool mgmtCheckMsgReadOnly(SQueuedMsg *pMsg);
static void mgmtProcessMsgFromShell(SRpcMsg *pMsg);
static void mgmtProcessUnSupportMsg(SRpcMsg *rpcMsg);
static void mgmtProcessMsgWhileNotReady(SRpcMsg *rpcMsg);
static void mgmtProcessShowMsg(SQueuedMsg *queuedMsg);
static void mgmtProcessRetrieveMsg(SQueuedMsg *queuedMsg);
static void mgmtProcessHeartBeatMsg(SQueuedMsg *queuedMsg);
......@@ -142,19 +141,13 @@ static void mgmtProcessMsgFromShell(SRpcMsg *rpcMsg) {
return;
}
if (mpeerCheckRedirect()) {
if (!mpeerIsMaster()) {
// rpcSendRedirectRsp(rpcMsg->handle, mgmtGetMnodeIpListForRedirect());
mgmtSendSimpleResp(rpcMsg->handle, TSDB_CODE_NO_MASTER);
rpcFreeCont(rpcMsg->pCont);
return;
}
if (!mpeerInServerStatus()) {
mgmtProcessMsgWhileNotReady(rpcMsg);
rpcFreeCont(rpcMsg->pCont);
return;
}
if (grantCheck(TSDB_GRANT_TIME) != TSDB_CODE_SUCCESS) {
mgmtSendSimpleResp(rpcMsg->handle, TSDB_CODE_GRANT_EXPIRED);
rpcFreeCont(rpcMsg->pCont);
......@@ -501,18 +494,6 @@ static void mgmtProcessUnSupportMsg(SRpcMsg *rpcMsg) {
rpcSendResponse(&rpcRsp);
}
static void mgmtProcessMsgWhileNotReady(SRpcMsg *rpcMsg) {
mTrace("%s is ignored since SDB is not ready", taosMsg[rpcMsg->msgType]);
SRpcMsg rpcRsp = {
.msgType = 0,
.pCont = 0,
.contLen = 0,
.code = TSDB_CODE_NOT_READY,
.handle = rpcMsg->handle
};
rpcSendResponse(&rpcRsp);
}
void mgmtSendSimpleResp(void *thandle, int32_t code) {
SRpcMsg rpcRsp = {
.msgType = 0,
......
src/mnode/src/mgmtTable.c
浏览文件 @ f73bc53b
......@@ -220,7 +220,7 @@ static int32_t mgmtChildTableActionDecode(SSdbOperDesc *pOper) {
return TSDB_CODE_SUCCESS;
}
static int32_t mgmtChildTableActionUpdateAll() {
static int32_t mgmtChildTableActionRestored() {
void *pNode = NULL;
void *pLastNode = NULL;
SChildTableObj *pTable = NULL;
......@@ -320,7 +320,7 @@ static int32_t mgmtInitChildTables() {
.encodeFp = mgmtChildTableActionEncode,
.decodeFp = mgmtChildTableActionDecode,
.destroyFp = mgmtChildTableActionDestroy,
.updateAllFp = mgmtChildTableActionUpdateAll
.restoredFp = mgmtChildTableActionRestored
};
tsChildTableSdb = sdbOpenTable(&tableDesc);
......@@ -414,7 +414,7 @@ static int32_t mgmtSuperTableActionDecode(SSdbOperDesc *pOper) {
return TSDB_CODE_SUCCESS;
}
static int32_t mgmtSuperTableActionUpdateAll() {
static int32_t mgmtSuperTableActionRestored() {
return 0;
}
......@@ -435,7 +435,7 @@ static int32_t mgmtInitSuperTables() {
.encodeFp = mgmtSuperTableActionEncode,
.decodeFp = mgmtSuperTableActionDecode,
.destroyFp = mgmtSuperTableActionDestroy,
.updateAllFp = mgmtSuperTableActionUpdateAll
.restoredFp = mgmtSuperTableActionRestored
};
tsSuperTableSdb = sdbOpenTable(&tableDesc);
......
src/mnode/src/mgmtUser.c
浏览文件 @ f73bc53b
......@@ -84,12 +84,14 @@ static int32_t mgmtUserActionDecode(SSdbOperDesc *pOper) {
return TSDB_CODE_SUCCESS;
}
static int32_t mgmtUserActionUpdateAll() {
SAcctObj *pAcct = acctGetAcct("root");
mgmtCreateUser(pAcct, "root", "taosdata");
mgmtCreateUser(pAcct, "monitor", tsInternalPass);
mgmtCreateUser(pAcct, "_root", tsInternalPass);
acctReleaseAcct(pAcct);
static int32_t mgmtUserActionRestored() {
if (strcmp(tsMasterIp, tsPrivateIp) == 0) {
SAcctObj *pAcct = acctGetAcct("root");
mgmtCreateUser(pAcct, "root", "taosdata");
mgmtCreateUser(pAcct, "monitor", tsInternalPass);
mgmtCreateUser(pAcct, "_root", tsInternalPass);
acctReleaseAcct(pAcct);
}
return 0;
}
......@@ -111,7 +113,7 @@ int32_t mgmtInitUsers() {
.encodeFp = mgmtUserActionEncode,
.decodeFp = mgmtUserActionDecode,
.destroyFp = mgmtUserActionDestroy,
.updateAllFp = mgmtUserActionUpdateAll
.restoredFp = mgmtUserActionRestored
};
tsUserSdb = sdbOpenTable(&tableDesc);
......
src/mnode/src/mgmtVgroup.c
浏览文件 @ f73bc53b
......@@ -152,7 +152,7 @@ static int32_t mgmtVgroupActionDecode(SSdbOperDesc *pOper) {
return TSDB_CODE_SUCCESS;
}
static int32_t mgmtVgroupActionUpdateAll() {
static int32_t mgmtVgroupActionRestored() {
return 0;
}
......@@ -173,7 +173,7 @@ int32_t mgmtInitVgroups() {
.encodeFp = mgmtVgroupActionEncode,
.decodeFp = mgmtVgroupActionDecode,
.destroyFp = mgmtVgroupActionDestroy,
.updateAllFp = mgmtVgroupActionUpdateAll,
.restoredFp = mgmtVgroupActionRestored,
};
tsVgroupSdb = sdbOpenTable(&tableDesc);
......
src/plugins/http/CMakeLists.txt
浏览文件 @ f73bc53b
......@@ -7,6 +7,8 @@ IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/util/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/query/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/zlib-1.2.11/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/cJson/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/lz4/inc)
INCLUDE_DIRECTORIES(${TD_OS_DIR}/inc)
INCLUDE_DIRECTORIES(inc)
AUX_SOURCE_DIRECTORY(src SRC)
......
src/plugins/http/src/httpServer.c
浏览文件 @ f73bc53b
......@@ -270,7 +270,7 @@ void httpCleanUpConnect(HttpServer *pServer) {
for (i = 0; i < pServer->numOfThreads; ++i) {
pThread = pServer->pThreads + i;
taosCloseSocket(pThread->pollFd);
//taosCloseSocket(pThread->pollFd);
while (pThread->pHead) {
httpCleanUpContext(pThread->pHead, 0);
......@@ -591,7 +591,6 @@ void httpAcceptHttpConnection(void *arg) {
bool httpInitConnect(HttpServer *pServer) {
int i;
pthread_attr_t thattr;
HttpThread * pThread;
pServer->pThreads = (HttpThread *)malloc(sizeof(HttpThread) * (size_t)pServer->numOfThreads);
......@@ -601,8 +600,6 @@ bool httpInitConnect(HttpServer *pServer) {
}
memset(pServer->pThreads, 0, sizeof(HttpThread) * (size_t)pServer->numOfThreads);
pthread_attr_init(&thattr);
pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE);
pThread = pServer->pThreads;
for (i = 0; i < pServer->numOfThreads; ++i) {
sprintf(pThread->label, "%s%d", pServer->label, i);
......@@ -626,21 +623,27 @@ bool httpInitConnect(HttpServer *pServer) {
return false;
}
pthread_attr_t thattr;
pthread_attr_init(&thattr);
pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE);
if (pthread_create(&(pThread->thread), &thattr, (void *)httpProcessHttpData, (void *)(pThread)) != 0) {
httpError("http thread:%s, failed to create HTTP process data thread, reason:%s",
pThread->label, strerror(errno));
return false;
}
pthread_attr_destroy(&thattr);
httpTrace("http thread:%p:%s, initialized", pThread, pThread->label);
pThread++;
}
pthread_attr_t thattr;
pthread_attr_init(&thattr);
pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE);
if (pthread_create(&(pServer->thread), &thattr, (void *)httpAcceptHttpConnection, (void *)(pServer)) != 0) {
httpError("http server:%s, failed to create Http accept thread, reason:%s", pServer->label, strerror(errno));
return false;
}
pthread_attr_destroy(&thattr);
httpTrace("http server:%s, initialized, ip:%s:%u, numOfThreads:%d", pServer->label, pServer->serverIp,
......
src/plugins/http/src/httpSystem.c
浏览文件 @ f73bc53b
......@@ -54,7 +54,7 @@ static HttpServer *httpServer = NULL;
void taosInitNote(int numOfNoteLines, int maxNotes, char* lable);
int httpInitSystem() {
taos_init();
// taos_init();
httpServer = (HttpServer *)malloc(sizeof(HttpServer));
memset(httpServer, 0, sizeof(HttpServer));
......
src/query/CMakeLists.txt
浏览文件 @ f73bc53b
......@@ -5,7 +5,7 @@ INCLUDE_DIRECTORIES(${TD_OS_DIR}/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/util/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/common/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/vnode/tsdb/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/tsdb/inc)
INCLUDE_DIRECTORIES(inc)
IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
......
src/rpc/CMakeLists.txt
浏览文件 @ f73bc53b
......@@ -4,6 +4,8 @@ PROJECT(TDengine)
INCLUDE_DIRECTORIES(${TD_OS_DIR}/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/util/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/cJson/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/lz4/inc)
INCLUDE_DIRECTORIES(inc)
IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
......@@ -22,7 +24,7 @@ ELSEIF (TD_WINDOWS_64)
ENDIF ()
ADD_LIBRARY(trpc ${SRC})
TARGET_LINK_LIBRARIES(trpc tutil)
TARGET_LINK_LIBRARIES(trpc tutil lz4)
ADD_SUBDIRECTORY(test)
src/rpc/src/rpcMain.c
浏览文件 @ f73bc53b
......@@ -286,15 +286,15 @@ void *rpcOpen(const SRpcInit *pInit) {
void rpcClose(void *param) {
SRpcInfo *pRpc = (SRpcInfo *)param;
(*taosCleanUpConn[pRpc->connType | RPC_CONN_TCP])(pRpc->tcphandle);
(*taosCleanUpConn[pRpc->connType])(pRpc->udphandle);
for (int i = 0; i < pRpc->sessions; ++i) {
if (pRpc->connList && pRpc->connList[i].user[0]) {
rpcCloseConn((void *)(pRpc->connList + i));
}
}
(*taosCleanUpConn[pRpc->connType | RPC_CONN_TCP])(pRpc->tcphandle);
(*taosCleanUpConn[pRpc->connType])(pRpc->udphandle);
taosHashCleanup(pRpc->hash);
taosTmrCleanUp(pRpc->tmrCtrl);
taosIdPoolCleanUp(pRpc->idPool);
......@@ -521,11 +521,15 @@ static SRpcConn *rpcOpenConn(SRpcInfo *pRpc, char *peerIpStr, uint16_t peerPort,
static void rpcCloseConn(void *thandle) {
SRpcConn *pConn = (SRpcConn *)thandle;
SRpcInfo *pRpc = pConn->pRpc;
if (pConn->user[0] == 0) return;
rpcLockConn(pConn);
if (pConn->user[0] == 0) {
rpcUnlockConn(pConn);
return;
}
pConn->user[0] = 0;
if (taosCloseConn[pConn->connType]) (*taosCloseConn[pConn->connType])(pConn->chandle);
......
src/rpc/src/rpcTcp.c
浏览文件 @ f73bc53b
......@@ -158,8 +158,9 @@ void taosCleanUpTcpServer(void *handle) {
pThreadObj = pServerObj->pThreadObj + i;
while (pThreadObj->pHead) {
taosFreeFdObj(pThreadObj->pHead);
pThreadObj->pHead = pThreadObj->pHead;
SFdObj *pFdObj = pThreadObj->pHead;
pThreadObj->pHead = pFdObj->next;
taosFreeFdObj(pFdObj);
}
close(pThreadObj->pollFd);
......@@ -269,8 +270,9 @@ void taosCleanUpTcpClient(void *chandle) {
if (pThreadObj == NULL) return;
while (pThreadObj->pHead) {
taosFreeFdObj(pThreadObj->pHead);
pThreadObj->pHead = pThreadObj->pHead->next;
SFdObj *pFdObj = pThreadObj->pHead;
pThreadObj->pHead = pFdObj->next;
taosFreeFdObj(pFdObj);
}
close(pThreadObj->pollFd);
......@@ -456,14 +458,18 @@ static void taosFreeFdObj(SFdObj *pFdObj) {
if (pFdObj == NULL) return;
if (pFdObj->signature != pFdObj) return;
pFdObj->signature = NULL;
SThreadObj *pThreadObj = pFdObj->pThreadObj;
pthread_mutex_lock(&pThreadObj->mutex);
if (pFdObj->signature == NULL) {
pthread_mutex_unlock(&pThreadObj->mutex);
return;
}
pFdObj->signature = NULL;
close(pFdObj->fd);
epoll_ctl(pThreadObj->pollFd, EPOLL_CTL_DEL, pFdObj->fd, NULL);
pthread_mutex_lock(&pThreadObj->mutex);
pThreadObj->numOfFds--;
if (pThreadObj->numOfFds < 0)
......
src/util/CMakeLists.txt
浏览文件 @ f73bc53b
......@@ -6,9 +6,11 @@ INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/util/inc)
IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/cJson/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/lz4/inc)
AUX_SOURCE_DIRECTORY(src SRC)
ADD_LIBRARY(tutil ${SRC})
TARGET_LINK_LIBRARIES(tutil thirdparty pthread os m rt)
TARGET_LINK_LIBRARIES(tutil pthread os m rt)
FIND_PATH(ICONV_INCLUDE_EXIST iconv.h /usr/include/ /usr/local/include/)
IF (ICONV_INCLUDE_EXIST)
ADD_DEFINITIONS(-DUSE_LIBICONV)
......@@ -65,7 +67,7 @@ ELSEIF (TD_WINDOWS_64)
LIST(APPEND SRC ./src/tutil.c)
LIST(APPEND SRC ./src/version.c)
ADD_LIBRARY(tutil ${SRC})
TARGET_LINK_LIBRARIES(tutil thirdparty iconv regex pthread os winmm IPHLPAPI ws2_32)
TARGET_LINK_LIBRARIES(tutil iconv regex pthread os winmm IPHLPAPI ws2_32)
ELSEIF(TD_DARWIN_64)
ADD_DEFINITIONS(-DUSE_LIBICONV)
LIST(APPEND SRC ./src/hash.c)
......@@ -102,7 +104,7 @@ ELSEIF(TD_DARWIN_64)
LIST(APPEND SRC ./src/version.c)
LIST(APPEND SRC ./src/hash.c)
ADD_LIBRARY(tutil ${SRC})
TARGET_LINK_LIBRARIES(tutil thirdparty iconv pthread os)
TARGET_LINK_LIBRARIES(tutil iconv pthread os)
ENDIF()
#IF (TD_CLUSTER)
......
src/vnode/CMakeLists.txt
浏览文件 @ f73bc53b
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
PROJECT(TDengine)
ADD_SUBDIRECTORY(wal)
ADD_SUBDIRECTORY(tsdb)
ADD_SUBDIRECTORY(main)
IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
INCLUDE_DIRECTORIES(${TD_OS_DIR}/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/util/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/query/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/tsdb/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/common/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/dnode/inc)
INCLUDE_DIRECTORIES(${TD_ENTERPRISE_DIR}/src/inc)
INCLUDE_DIRECTORIES(inc)
AUX_SOURCE_DIRECTORY(src SRC)
ADD_LIBRARY(vnode ${SRC})
TARGET_LINK_LIBRARIES(vnode tsdb)
ENDIF ()
\ No newline at end of file
src/vnode/cache/inc/cache.h 已删除 100644 → 0
浏览文件 @ 3f5b1415
#if !defined(_TD_CACHE_H_)
#define _TD_CACHE_H_
#define TD_MIN_CACHE_BLOCK_SIZE 1024*1024 /* 1M */
#define TD_MAX_CACHE_BLOCK_SIZE 64*1024*1024 /* 64M */
typedef void cache_pool_t;
typedef struct SCacheBlock
{
int32_t blockId;
char data[];
} SCacheBlock;
#endif // _TD_CACHE_H_
src/vnode/main/CMakeLists.txt 已删除 100644 → 0
浏览文件 @ 3f5b1415
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
PROJECT(TDengine)
IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
INCLUDE_DIRECTORIES(${TD_OS_DIR}/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/util/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/query/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/vnode/tsdb/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/common/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/dnode/inc)
INCLUDE_DIRECTORIES(${TD_ENTERPRISE_DIR}/src/inc)
INCLUDE_DIRECTORIES(inc)
AUX_SOURCE_DIRECTORY(src SRC)
ADD_LIBRARY(vnode ${SRC})
TARGET_LINK_LIBRARIES(vnode tsdb)
ENDIF ()
\ No newline at end of file
src/vnode/main/src/vnodeMain.c → src/vnode/src/vnodeMain.c
浏览文件 @ f73bc53b
......@@ -160,7 +160,7 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
syncInfo.writeToCache = vnodeWriteToQueue;
syncInfo.confirmForward = dnodeSendRpcWriteRsp;
syncInfo.notifyRole = vnodeNotifyRole;
pVnode->sync = syncStart(&syncInfo);;
pVnode->sync = syncStart(&syncInfo);
pVnode->events = NULL;
pVnode->cq = NULL;
......
tests/script/tmp/dnode2.sim → tests/script/tmp/mnodes.sim
浏览文件 @ f73bc53b
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -m 192.168.0.1 -i 192.168.0.1
system sh/deploy.sh -n dnode2 -m 192.168.0.1 -i 192.168.0.2
system sh/exec_up.sh -n dnode1 -s start
system sh/exec_up.sh -n dnode2 -s start
sql connect
\ No newline at end of file
system sh/deploy.sh -n dnode3 -m 192.168.0.1 -i 192.168.0.3
system sh/cfg.sh -n dnode1 -c numOfMPeers -v 3
system sh/cfg.sh -n dnode2 -c numOfMPeers -v 3
system sh/cfg.sh -n dnode3 -c numOfMPeers -v 3
tests/test/c/CMakeLists.txt
浏览文件 @ f73bc53b
......@@ -6,6 +6,12 @@ INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/util/inc)
INCLUDE_DIRECTORIES(${TD_OS_DIR}/inc)
IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
add_executable(benchmarkPerTable benchmarkPerTable.c)
target_link_libraries(benchmarkPerTable taos_static pthread)
add_executable(insertPerTable insertPerTable.c)
target_link_libraries(insertPerTable taos_static pthread)
add_executable(insertPerRow insertPerRow.c)
target_link_libraries(insertPerRow taos_static pthread)
add_executable(importOneRow importOneRow.c)
target_link_libraries(importOneRow taos_static pthread)
ENDIF()
tests/test/c/importOneRow.c 0 → 100644
浏览文件 @ f73bc53b
/*
* 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 "taos.h"
#include "tlog.h"
#include "ttimer.h"
#include "tutil.h"
#define MAX_RANDOM_POINTS 20000
#define GREEN "\033[1;32m"
#define NC "\033[0m"
void taos_error(TAOS *taos);
void* taos_execute(void *param);
typedef struct {
pthread_t pid;
int index;
} ThreadObj;
int threadNum = 1;
int rowNum = 1000;
int replica = 1;
void printHelp() {
char indent[10] = " ";
printf("Used to test the performance of TDengine\n After writing one row of data to all tables, write the next row\n");
printf("%s%s\n", indent, "-r");
printf("%s%s%s%d\n", indent, indent, "Number of records to write table, default is ", rowNum);
printf("%s%s\n", indent, "-t");
printf("%s%s%s%d\n", indent, indent, "Number of threads to be used, default is ", threadNum);
printf("%s%s\n", indent, "-replica");
printf("%s%s%s%d\n", indent, indent, "Database parameters replica, default is ", replica);
exit(EXIT_SUCCESS);
}
void shellParseArgument(int argc, char *argv[]) {
for (int i = 1; i < argc; i++) {
if (strcmp(argv[i], "-h") == 0 || strcmp(argv[i], "--help") == 0) {
printHelp();
exit(0);
} else if (strcmp(argv[i], "-r") == 0) {
rowNum = atoi(argv[++i]);
} else if (strcmp(argv[i], "-t") == 0) {
threadNum = atoi(argv[++i]);
} else if (strcmp(argv[i], "-replica") == 0) {
replica = atoi(argv[++i]);
} else {
}
}
dPrint("%s rowNum:%d %s", GREEN, rowNum, NC);
dPrint("%s threadNum:%d %s", GREEN, threadNum, NC);
dPrint("%s replica:%d %s", GREEN, replica, NC);
}
int main(int argc, char *argv[]) {
shellParseArgument(argc, argv);
taos_init();
ThreadObj *threads = calloc(threadNum, sizeof(ThreadObj));
for (int i = 0; i < threadNum; ++i) {
ThreadObj * pthread = threads + i;
pthread_attr_t thattr;
pthread->index = i;
pthread_attr_init(&thattr);
pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE);
pthread_create(&pthread->pid, &thattr, taos_execute, pthread);
}
for (int i = 0; i < threadNum; i++) {
pthread_join(threads[i].pid, NULL);
}
printf("all finished\n");
return 0;
}
void taos_error(TAOS *con) {
fprintf(stderr, "TDengine error: %s\n", taos_errstr(con));
taos_close(con);
exit(1);
}
void* taos_execute(void *param) {
ThreadObj *pThread = (ThreadObj *)param;
void *taos = taos_connect(tsMasterIp, tsDefaultUser, tsDefaultPass, NULL, 0);
if (taos == NULL) taos_error(taos);
char sql[1024] = {0};
sprintf(sql, "create database if not exists db replica %d", replica);
taos_query(taos, sql);
sprintf(sql, "create table if not exists db.t%d (ts timestamp, i int, j float, k double)", pThread->index);
taos_query(taos, sql);
int64_t timestamp = 1530374400000L;
sprintf(sql, "insert into db.t%d values(%ld, %d, %d, %d)", pThread->index, timestamp, 0, 0, 0);
int code = taos_query(taos, sql);
if (code != 0) printf("error code:%d, sql:%s\n", code, sql);
int affectrows = taos_affected_rows(taos);
if (affectrows != 1) printf("affect rows:%d, sql:%s\n", affectrows, sql);
timestamp -= 1000;
int total_affect_rows = affectrows;
for (int i = 1; i < rowNum; ++i) {
sprintf(sql, "import into db.t%d values(%ld, %d, %d, %d)", pThread->index, timestamp, i, i, i);
code = taos_query(taos, sql);
if (code != 0) printf("error code:%d, sql:%s\n", code, sql);
int affectrows = taos_affected_rows(taos);
if (affectrows != 1) printf("affect rows:%d, sql:%s\n", affectrows, sql);
total_affect_rows += affectrows;
timestamp -= 1000;
}
printf("thread:%d run finished total_affect_rows:%d\n", pThread->index, total_affect_rows);
return NULL;
}
tests/test/c/insertPerRow.c 0 → 100644
浏览文件 @ f73bc53b
/*
* 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 "taos.h"
#include "tlog.h"
#include "ttimer.h"
#include "tutil.h"
#define MAX_RANDOM_POINTS 20000
#define GREEN "\033[1;32m"
#define NC "\033[0m"
typedef struct {
int64_t rowsPerTable;
int64_t pointsPerTable;
int64_t tableBeginIndex;
int64_t tableEndIndex;
int threadIndex;
char dbName[32];
char stableName[64];
pthread_t thread;
} SInfo;
void *syncTest(void *param);
void generateRandomPoints();
void shellParseArgument(int argc, char *argv[]);
void createDbAndTable();
void insertData();
int32_t randomData[MAX_RANDOM_POINTS];
int64_t rowsPerTable = 10000;
int64_t pointsPerTable = 1;
int64_t numOfThreads = 1;
int64_t numOfTablesPerThread = 200;
char dbName[32] = "db";
char stableName[64] = "st";
int32_t cache = 16384;
int32_t tables = 1000;
int main(int argc, char *argv[]) {
shellParseArgument(argc, argv);
generateRandomPoints();
taos_init();
createDbAndTable();
insertData();
}
void createDbAndTable() {
dPrint("start to create table");
TAOS * con;
struct timeval systemTime;
int64_t st, et;
char qstr[64000];
con = taos_connect(tsMasterIp, tsDefaultUser, tsDefaultPass, NULL, 0);
if (con == NULL) {
dError("failed to connect to DB, reason:%s", taos_errstr(con));
exit(1);
}
sprintf(qstr, "create database if not exists %s cache %d tables %d", dbName, cache, tables);
if (taos_query(con, qstr)) {
dError("failed to create database:%s, code:%d reason:%s", dbName, taos_errno(con), taos_errstr(con));
exit(0);
}
sprintf(qstr, "use %s", dbName);
if (taos_query(con, qstr)) {
dError("failed to use db, code:%d reason:%s", taos_errno(con), taos_errstr(con));
exit(0);
}
gettimeofday(&systemTime, NULL);
st = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
int64_t totalTables = numOfTablesPerThread * numOfThreads;
if (strcmp(stableName, "no") != 0) {
int len = sprintf(qstr, "create table if not exists %s(ts timestamp", stableName);
for (int64_t f = 0; f < pointsPerTable; ++f) {
len += sprintf(qstr + len, ", f%ld double", f);
}
sprintf(qstr + len, ") tags(t int)");
if (taos_query(con, qstr)) {
dError("failed to create stable, code:%d reason:%s", taos_errno(con), taos_errstr(con));
exit(0);
}
for (int64_t t = 0; t < totalTables; ++t) {
sprintf(qstr, "create table if not exists %s%ld using %s tags(%ld)", stableName, t, stableName, t);
if (taos_query(con, qstr)) {
dError("failed to create table %s%d, reason:%s", stableName, t, taos_errstr(con));
exit(0);
}
}
} else {
for (int64_t t = 0; t < totalTables; ++t) {
int len = sprintf(qstr, "create table if not exists %s%ld(ts timestamp", stableName, t);
for (int64_t f = 0; f < pointsPerTable; ++f) {
len += sprintf(qstr + len, ", f%ld double", f);
}
sprintf(qstr + len, ")");
if (taos_query(con, qstr)) {
dError("failed to create table %s%ld, reason:%s", stableName, t, taos_errstr(con));
exit(0);
}
}
}
gettimeofday(&systemTime, NULL);
et = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
dPrint("%.1f seconds to create %ld tables", (et - st) / 1000.0 / 1000.0, totalTables);
}
void insertData() {
struct timeval systemTime;
int64_t st, et;
gettimeofday(&systemTime, NULL);
st = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
dPrint("%d threads are spawned to insert data", numOfThreads);
pthread_attr_t thattr;
pthread_attr_init(&thattr);
pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE);
SInfo *pInfo = (SInfo *)malloc(sizeof(SInfo) * numOfThreads);
// Start threads to write
for (int i = 0; i < numOfThreads; ++i) {
pInfo[i].rowsPerTable = rowsPerTable;
pInfo[i].pointsPerTable = pointsPerTable;
pInfo[i].tableBeginIndex = i * numOfTablesPerThread;
pInfo[i].tableEndIndex = (i + 1) * numOfTablesPerThread;
pInfo[i].threadIndex = i;
strcpy(pInfo[i].dbName, dbName);
strcpy(pInfo[i].stableName, stableName);
pthread_create(&(pInfo[i].thread), &thattr, syncTest, (void *)(pInfo + i));
}
taosMsleep(300);
for (int i = 0; i < numOfThreads; i++) {
pthread_join(pInfo[i].thread, NULL);
}
gettimeofday(&systemTime, NULL);
et = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
double seconds = (et - st) / 1000.0 / 1000.0;
int64_t totalTables = numOfTablesPerThread * numOfThreads;
int64_t totalRows = totalTables * rowsPerTable;
int64_t totalPoints = totalTables * rowsPerTable * pointsPerTable;
double speedOfRows = totalRows / seconds;
double speedOfPoints = totalPoints / seconds;
dPrint(
"%sall threads:%ld finished, use %.1lf seconds, tables:%.ld rows:%ld points:%ld, speed RowsPerSecond:%.1lf "
"PointsPerSecond:%.1lf%s",
GREEN, numOfThreads, seconds, totalTables, totalRows, totalPoints, speedOfRows, speedOfPoints, NC);
dPrint("threads exit");
pthread_attr_destroy(&thattr);
free(pInfo);
}
void *syncTest(void *param) {
TAOS * con;
SInfo * pInfo = (SInfo *)param;
struct timeval systemTime;
int64_t st, et;
char qstr[65000];
int maxBytes = 60000;
dPrint("thread:%d, start to run", pInfo->threadIndex);
con = taos_connect(tsMasterIp, tsDefaultUser, tsDefaultPass, NULL, 0);
if (con == NULL) {
dError("index:%d, failed to connect to DB, reason:%s", pInfo->threadIndex, taos_errstr(con));
exit(1);
}
sprintf(qstr, "use %s", pInfo->dbName);
taos_query(con, qstr);
gettimeofday(&systemTime, NULL);
st = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
int64_t start = 1430000000000;
int64_t interval = 1000; // 1000 ms
char *sql = qstr;
char inserStr[] = "insert into";
int len = sprintf(sql, "%s", inserStr);
for (int64_t row = 0; row < pInfo->rowsPerTable; row++) {
for (int64_t table = pInfo->tableBeginIndex; table < pInfo->tableEndIndex; ++table) {
len += sprintf(sql + len, " %s%ld values", pInfo->stableName, table);
len += sprintf(sql + len, "(%ld", start + row * interval);
for (int64_t point = 0; point < pInfo->pointsPerTable; ++point) {
len += sprintf(sql + len, ",%d", randomData[(123 * table + 456 * row + 789 * point) % MAX_RANDOM_POINTS]);
// len += sprintf(sql + len, ",%ld", row);
}
len += sprintf(sql + len, ")");
if (len > maxBytes) {
if (taos_query(con, qstr)) {
dError("thread:%d, failed to insert table:%s%ld row:%ld, reason:%s", pInfo->threadIndex, pInfo->stableName,
table, row, taos_errstr(con));
}
// "insert into"
len = sprintf(sql, "%s", inserStr);
}
}
}
if (len != strlen(inserStr)) {
taos_query(con, qstr);
}
gettimeofday(&systemTime, NULL);
et = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
int64_t totalTables = pInfo->tableEndIndex - pInfo->tableBeginIndex;
int64_t totalRows = totalTables * pInfo->rowsPerTable;
int64_t totalPoints = totalRows * pInfo->pointsPerTable;
dPrint("thread:%d, insert finished, use %.2f seconds, tables:%ld rows:%ld points:%ld", pInfo->threadIndex,
(et - st) / 1000.0 / 1000.0, totalTables, totalRows, totalPoints);
return NULL;
}
void generateRandomPoints() {
for (int r = 0; r < MAX_RANDOM_POINTS; ++r) {
randomData[r] = rand() % 1000;
}
}
void printHelp() {
char indent[10] = " ";
printf("Used to test the performance of TDengine\n After writing one row of data to all tables, write the next row\n");
printf("%s%s\n", indent, "-d");
printf("%s%s%s%s\n", indent, indent, "The name of the database to be created, default is ", dbName);
printf("%s%s\n", indent, "-s");
printf("%s%s%s%s%s\n", indent, indent, "The name of the super table to be created, default is ", stableName, ", if 'no' then create normal table");
printf("%s%s\n", indent, "-c");
printf("%s%s%s%s\n", indent, indent, "Configuration directory, default is ", configDir);
printf("%s%s\n", indent, "-r");
printf("%s%s%s%ld\n", indent, indent, "Number of records to write to each table, default is ", rowsPerTable);
printf("%s%s\n", indent, "-p");
printf("%s%s%s%" PRId64 "\n", indent, indent, "Number of columns per table, default is ", pointsPerTable);
printf("%s%s\n", indent, "-t");
printf("%s%s%s%" PRId64 "\n", indent, indent, "Number of threads to be used, default is ", numOfThreads);
printf("%s%s\n", indent, "-n");
printf("%s%s%s%" PRId64 "\n", indent, indent, "Number of tables per thread, default is ", numOfTablesPerThread);
printf("%s%s\n", indent, "-tables");
printf("%s%s%s%d\n", indent, indent, "Database parameters tables, default is ", tables);
printf("%s%s\n", indent, "-cache");
printf("%s%s%s%d\n", indent, indent, "Database parameters cache, default is ", cache);
exit(EXIT_SUCCESS);
}
void shellParseArgument(int argc, char *argv[]) {
for (int i = 1; i < argc; i++) {
if (strcmp(argv[i], "-h") == 0 || strcmp(argv[i], "--help") == 0) {
printHelp();
exit(0);
} else if (strcmp(argv[i], "-d") == 0) {
strcpy(dbName, argv[++i]);
} else if (strcmp(argv[i], "-c") == 0) {
strcpy(configDir, argv[++i]);
} else if (strcmp(argv[i], "-s") == 0) {
strcpy(stableName, argv[++i]);
} else if (strcmp(argv[i], "-r") == 0) {
rowsPerTable = atoi(argv[++i]);
} else if (strcmp(argv[i], "-p") == 0) {
pointsPerTable = atoi(argv[++i]);
} else if (strcmp(argv[i], "-t") == 0) {
numOfThreads = atoi(argv[++i]);
} else if (strcmp(argv[i], "-n") == 0) {
numOfTablesPerThread = atoi(argv[++i]);
} else if (strcmp(argv[i], "-tables") == 0) {
tables = atoi(argv[++i]);
} else if (strcmp(argv[i], "-cache") == 0) {
cache = atoi(argv[++i]);
} else {
}
}
dPrint("%srowsPerTable:%" PRId64 "%s", GREEN, rowsPerTable, NC);
dPrint("%spointsPerTable:%" PRId64 "%s", GREEN, pointsPerTable, NC);
dPrint("%snumOfThreads:%" PRId64 "%s", GREEN, numOfThreads, NC);
dPrint("%snumOfTablesPerThread:%" PRId64 "%s", GREEN, numOfTablesPerThread, NC);
dPrint("%scache:%" PRId64 "%s", GREEN, cache, NC);
dPrint("%stables:%" PRId64 "%s", GREEN, tables, NC);
dPrint("%sdbName:%s%s", GREEN, dbName, NC);
dPrint("%stableName:%s%s", GREEN, stableName, NC);
dPrint("%sstart to run%s", GREEN, NC);
}
tests/test/c/benchmarkPerTable.c → tests/test/c/insertPerTable.c
浏览文件 @ f73bc53b
......@@ -258,7 +258,7 @@ void generateRandomPoints() {
void printHelp() {
char indent[10] = " ";
printf("Used to test the performance of TDengine, the insert method is table-by-table\n");
printf("Used to test the performance of TDengine\n After writing all the data in one table, start the next table\n");
printf("%s%s\n", indent, "-d");
printf("%s%s%s%s\n", indent, indent, "The name of the database to be created, default is ", dbName);
......@@ -289,6 +289,8 @@ void shellParseArgument(int argc, char *argv[]) {
exit(0);
} else if (strcmp(argv[i], "-d") == 0) {
strcpy(dbName, argv[++i]);
} else if (strcmp(argv[i], "-c") == 0) {
strcpy(configDir, argv[++i]);
} else if (strcmp(argv[i], "-s") == 0) {
strcpy(stableName, argv[++i]);
} else if (strcmp(argv[i], "-r") == 0) {
......
tests/tsim/CMakeLists.txt
浏览文件 @ f73bc53b
......@@ -3,6 +3,7 @@ PROJECT(TDengine)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/util/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/client/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/cJson/inc)
INCLUDE_DIRECTORIES(${TD_OS_DIR}/inc)
INCLUDE_DIRECTORIES(inc)
......@@ -12,4 +13,4 @@ ENDIF ()
AUX_SOURCE_DIRECTORY(src SRC)
ADD_EXECUTABLE(tsim ${SRC})
TARGET_LINK_LIBRARIES(tsim taos_static trpc tutil pthread )
TARGET_LINK_LIBRARIES(tsim taos_static trpc tutil pthread cJson)
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或