Merge branch 'develop' into feature/TD-2502-v2
This commit is contained in:
Minglei Jin
commit
1a8d66ceee
|
|
@ -265,8 +265,14 @@ function install_config() {
|
|||
[ -f ${cfg_dir}/taos.cfg ] && ${csudo} cp ${cfg_dir}/taos.cfg ${cfg_install_dir}
|
||||
${csudo} chmod 644 ${cfg_install_dir}/*
|
||||
fi
|
||||
|
||||
# Save standard input to 6 and open / dev / TTY on standard input
|
||||
exec 6<&0 0</dev/tty
|
||||
|
||||
local_fqdn_check
|
||||
|
||||
# restore the backup standard input, and turn off 6
|
||||
exec 0<&6 6<&-
|
||||
|
||||
${csudo} mv ${cfg_dir}/taos.cfg ${cfg_dir}/taos.cfg.org
|
||||
${csudo} ln -s ${cfg_install_dir}/taos.cfg ${cfg_dir}
|
||||
|
|
@ -422,7 +428,7 @@ function install_service() {
|
|||
}
|
||||
|
||||
function install_TDengine() {
|
||||
echo -e "${GREEN}Start to install TDEngine...${NC}"
|
||||
echo -e "${GREEN}Start to install TDengine...${NC}"
|
||||
|
||||
#install log and data dir , then ln to /usr/local/taos
|
||||
${csudo} mkdir -p ${log_dir} && ${csudo} chmod 777 ${log_dir}
|
||||
|
|
|
|||
|
|
@ -119,4 +119,4 @@ if ((${service_mod}==2)); then
|
|||
kill_taosd
|
||||
fi
|
||||
|
||||
echo -e "${GREEN}TDEngine is removed successfully!${NC}"
|
||||
echo -e "${GREEN}TDengine is removed successfully!${NC}"
|
||||
|
|
|
|||
|
|
@ -2,19 +2,39 @@
|
|||
#
|
||||
# This file is used to set config for core when taosd crash
|
||||
|
||||
# Color setting
|
||||
RED='\033[0;31m'
|
||||
GREEN='\033[1;32m'
|
||||
GREEN_DARK='\033[0;32m'
|
||||
GREEN_UNDERLINE='\033[4;32m'
|
||||
NC='\033[0m'
|
||||
|
||||
set -e
|
||||
# set -x
|
||||
corePath=$1
|
||||
|
||||
csudo=""
|
||||
if command -v sudo > /dev/null; then
|
||||
csudo="sudo"
|
||||
fi
|
||||
|
||||
#ulimit -c unlimited
|
||||
if [[ ! -n ${corePath} ]]; then
|
||||
echo -e -n "${GREEN}Please enter a file directory to save the coredump file${NC}:"
|
||||
read corePath
|
||||
while true; do
|
||||
if [[ ! -z "$corePath" ]]; then
|
||||
break
|
||||
else
|
||||
read -p "Please enter a file directory to save the coredump file:" corePath
|
||||
fi
|
||||
done
|
||||
fi
|
||||
|
||||
ulimit -c unlimited
|
||||
${csudo} sed -i '/ulimit -c unlimited/d' /etc/profile ||:
|
||||
${csudo} sed -i '$a\ulimit -c unlimited' /etc/profile ||:
|
||||
source /etc/profile
|
||||
|
||||
${csudo} mkdir -p /coredump ||:
|
||||
${csudo} sysctl -w kernel.core_pattern='/coredump/core-%e-%p' ||:
|
||||
${csudo} echo '/coredump/core-%e-%p' | ${csudo} tee /proc/sys/kernel/core_pattern ||:
|
||||
${csudo} mkdir -p ${corePath} ||:
|
||||
${csudo} sysctl -w kernel.core_pattern=${corePath}/core-%e-%p ||:
|
||||
${csudo} echo "${corePath}/core-%e-%p" | ${csudo} tee /proc/sys/kernel/core_pattern ||:
|
||||
|
|
|
|||
|
|
@ -330,7 +330,7 @@ void bnReset() {
|
|||
tsAccessSquence = 0;
|
||||
}
|
||||
|
||||
static int32_t bnMonitorVgroups() {
|
||||
static bool bnMonitorVgroups() {
|
||||
void * pIter = NULL;
|
||||
SVgObj *pVgroup = NULL;
|
||||
bool hasUpdatingVgroup = false;
|
||||
|
|
@ -489,6 +489,7 @@ void bnCheckStatus() {
|
|||
mInfo("dnode:%d, set to offline state, access seq:%d last seq:%d laststat:%d", pDnode->dnodeId, tsAccessSquence,
|
||||
pDnode->lastAccess, pDnode->status);
|
||||
bnSetVgroupOffline(pDnode);
|
||||
bnStartTimer(3000);
|
||||
}
|
||||
}
|
||||
mnodeDecDnodeRef(pDnode);
|
||||
|
|
|
|||
|
|
@ -31,7 +31,10 @@ static void *bnThreadFunc(void *arg) {
|
|||
}
|
||||
|
||||
pthread_cond_wait(&tsBnThread.cond, &tsBnThread.mutex);
|
||||
mDebug("balance thread wakes up to work");
|
||||
bool updateSoon = bnStart();
|
||||
mDebug("balance thread finished this poll, updateSoon:%d", updateSoon);
|
||||
|
||||
bnStartTimer(updateSoon ? 1000 : -1);
|
||||
pthread_mutex_unlock(&(tsBnThread.mutex));
|
||||
}
|
||||
|
|
@ -101,8 +104,8 @@ static void bnProcessTimer(void *handle, void *tmrId) {
|
|||
tsBnThread.timer = NULL;
|
||||
tsAccessSquence++;
|
||||
|
||||
bnCheckStatus();
|
||||
bnStartTimer(-1);
|
||||
bnCheckStatus();
|
||||
|
||||
if (handle == NULL) {
|
||||
if (tsAccessSquence % tsBalanceInterval == 0) {
|
||||
|
|
@ -121,6 +124,7 @@ void bnStartTimer(int64_t mseconds) {
|
|||
|
||||
bool updateSoon = (mseconds != -1);
|
||||
if (updateSoon) {
|
||||
mTrace("balance function will be called after %" PRId64 " ms", mseconds);
|
||||
taosTmrReset(bnProcessTimer, mseconds, (void *)mseconds, tsMnodeTmr, &tsBnThread.timer);
|
||||
} else {
|
||||
taosTmrReset(bnProcessTimer, tsStatusInterval * 1000, NULL, tsMnodeTmr, &tsBnThread.timer);
|
||||
|
|
|
|||
|
|
@ -388,10 +388,10 @@ void tscQueueAsyncRes(SSqlObj *pSql) {
|
|||
return;
|
||||
}
|
||||
|
||||
assert(pSql->res.code != TSDB_CODE_SUCCESS);
|
||||
tscError("%p add into queued async res, code:%s", pSql, tstrerror(pSql->res.code));
|
||||
|
||||
SSqlRes *pRes = &pSql->res;
|
||||
|
||||
if (pSql->fp == NULL || pSql->fetchFp == NULL){
|
||||
return;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2597,14 +2597,23 @@ static void percentile_next_step(SQLFunctionCtx *pCtx) {
|
|||
}
|
||||
|
||||
//////////////////////////////////////////////////////////////////////////////////
|
||||
static void buildHistogramInfo(SAPercentileInfo* pInfo) {
|
||||
pInfo->pHisto = (SHistogramInfo*) ((char*) pInfo + sizeof(SAPercentileInfo));
|
||||
pInfo->pHisto->elems = (SHistBin*) ((char*)pInfo->pHisto + sizeof(SHistogramInfo));
|
||||
}
|
||||
|
||||
static SAPercentileInfo *getAPerctInfo(SQLFunctionCtx *pCtx) {
|
||||
SResultRowCellInfo *pResInfo = GET_RES_INFO(pCtx);
|
||||
|
||||
SAPercentileInfo* pInfo = NULL;
|
||||
|
||||
if (pCtx->stableQuery && pCtx->currentStage != SECONDARY_STAGE_MERGE) {
|
||||
return (SAPercentileInfo*) pCtx->aOutputBuf;
|
||||
pInfo = (SAPercentileInfo*) pCtx->aOutputBuf;
|
||||
} else {
|
||||
return GET_ROWCELL_INTERBUF(pResInfo);
|
||||
pInfo = GET_ROWCELL_INTERBUF(pResInfo);
|
||||
}
|
||||
|
||||
buildHistogramInfo(pInfo);
|
||||
return pInfo;
|
||||
}
|
||||
|
||||
static bool apercentile_function_setup(SQLFunctionCtx *pCtx) {
|
||||
|
|
@ -2616,6 +2625,7 @@ static bool apercentile_function_setup(SQLFunctionCtx *pCtx) {
|
|||
|
||||
char *tmp = (char *)pInfo + sizeof(SAPercentileInfo);
|
||||
pInfo->pHisto = tHistogramCreateFrom(tmp, MAX_HISTOGRAM_BIN);
|
||||
printf("%p, %p\n", pInfo->pHisto, pInfo->pHisto->elems);
|
||||
return true;
|
||||
}
|
||||
|
||||
|
|
@ -2624,6 +2634,8 @@ static void apercentile_function(SQLFunctionCtx *pCtx) {
|
|||
|
||||
SResultRowCellInfo * pResInfo = GET_RES_INFO(pCtx);
|
||||
SAPercentileInfo *pInfo = getAPerctInfo(pCtx);
|
||||
|
||||
assert(pInfo->pHisto->elems != NULL);
|
||||
|
||||
for (int32_t i = 0; i < pCtx->size; ++i) {
|
||||
char *data = GET_INPUT_CHAR_INDEX(pCtx, i);
|
||||
|
|
|
|||
|
|
@ -911,6 +911,13 @@ static void genFinalResWithoutFill(SSqlRes* pRes, SLocalReducer *pLocalReducer,
|
|||
}
|
||||
}
|
||||
|
||||
if (pRes->numOfRowsGroup >= pQueryInfo->limit.limit && pQueryInfo->limit.limit > 0) {
|
||||
pRes->numOfRows = 0;
|
||||
pBeforeFillData->num = 0;
|
||||
pLocalReducer->discard = true;
|
||||
return;
|
||||
}
|
||||
|
||||
pRes->numOfRowsGroup += pRes->numOfRows;
|
||||
|
||||
// impose the limitation of output rows on the final result
|
||||
|
|
|
|||
|
|
@ -284,16 +284,18 @@ void taos_close(TAOS *taos) {
|
|||
return;
|
||||
}
|
||||
|
||||
SSqlObj* pHb = (SSqlObj*)taosAcquireRef(tscObjRef, pObj->hbrid);
|
||||
if (pHb != NULL) {
|
||||
if (pHb->rpcRid > 0) { // wait for rsp from dnode
|
||||
rpcCancelRequest(pHb->rpcRid);
|
||||
pHb->rpcRid = -1;
|
||||
}
|
||||
if (RID_VALID(pObj->hbrid)) {
|
||||
SSqlObj* pHb = (SSqlObj*)taosAcquireRef(tscObjRef, pObj->hbrid);
|
||||
if (pHb != NULL) {
|
||||
if (RID_VALID(pHb->rpcRid)) { // wait for rsp from dnode
|
||||
rpcCancelRequest(pHb->rpcRid);
|
||||
pHb->rpcRid = -1;
|
||||
}
|
||||
|
||||
tscDebug("%p HB is freed", pHb);
|
||||
taosReleaseRef(tscObjRef, pHb->self);
|
||||
taos_free_result(pHb);
|
||||
tscDebug("%p HB is freed", pHb);
|
||||
taosReleaseRef(tscObjRef, pHb->self);
|
||||
taos_free_result(pHb);
|
||||
}
|
||||
}
|
||||
|
||||
tscDebug("%p all sqlObj are freed, free tscObj and close dnodeConn:%p", pObj, pObj->pDnodeConn);
|
||||
|
|
|
|||
|
|
@ -101,7 +101,8 @@ extern int32_t tsAlternativeRole;
|
|||
extern int32_t tsBalanceInterval;
|
||||
extern int32_t tsOfflineThreshold;
|
||||
extern int32_t tsMnodeEqualVnodeNum;
|
||||
extern int32_t tsFlowCtrl;
|
||||
extern int32_t tsEnableFlowCtrl;
|
||||
extern int32_t tsEnableSlaveQuery;
|
||||
|
||||
// restful
|
||||
extern int32_t tsEnableHttpModule;
|
||||
|
|
|
|||
|
|
@ -138,7 +138,8 @@ int32_t tsAlternativeRole = 0;
|
|||
int32_t tsBalanceInterval = 300; // seconds
|
||||
int32_t tsOfflineThreshold = 86400*100; // seconds 10days
|
||||
int32_t tsMnodeEqualVnodeNum = 4;
|
||||
int32_t tsFlowCtrl = 1;
|
||||
int32_t tsEnableFlowCtrl = 1;
|
||||
int32_t tsEnableSlaveQuery = 1;
|
||||
|
||||
// restful
|
||||
int32_t tsEnableHttpModule = 1;
|
||||
|
|
@ -221,7 +222,7 @@ int32_t uDebugFlag = 131;
|
|||
int32_t debugFlag = 0;
|
||||
int32_t sDebugFlag = 135;
|
||||
int32_t wDebugFlag = 135;
|
||||
int32_t tsdbDebugFlag = 131;
|
||||
uint32_t tsdbDebugFlag = 131;
|
||||
int32_t cqDebugFlag = 131;
|
||||
|
||||
int32_t (*monStartSystemFp)() = NULL;
|
||||
|
|
@ -542,7 +543,7 @@ static void doInitGlobalConfig(void) {
|
|||
cfg.ptr = &tsOfflineThreshold;
|
||||
cfg.valType = TAOS_CFG_VTYPE_INT32;
|
||||
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW;
|
||||
cfg.minValue = 5;
|
||||
cfg.minValue = 3;
|
||||
cfg.maxValue = 7200000;
|
||||
cfg.ptrLength = 0;
|
||||
cfg.unitType = TAOS_CFG_UTYPE_SECOND;
|
||||
|
|
@ -1004,7 +1005,17 @@ static void doInitGlobalConfig(void) {
|
|||
|
||||
// module configs
|
||||
cfg.option = "flowctrl";
|
||||
cfg.ptr = &tsFlowCtrl;
|
||||
cfg.ptr = &tsEnableFlowCtrl;
|
||||
cfg.valType = TAOS_CFG_VTYPE_INT32;
|
||||
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW;
|
||||
cfg.minValue = 0;
|
||||
cfg.maxValue = 1;
|
||||
cfg.ptrLength = 0;
|
||||
cfg.unitType = TAOS_CFG_UTYPE_NONE;
|
||||
taosInitConfigOption(cfg);
|
||||
|
||||
cfg.option = "slaveQuery";
|
||||
cfg.ptr = &tsEnableSlaveQuery;
|
||||
cfg.valType = TAOS_CFG_VTYPE_INT32;
|
||||
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW;
|
||||
cfg.minValue = 0;
|
||||
|
|
|
|||
|
|
@ -1 +1 @@
|
|||
Subproject commit 32e2c97a4cf7bedaa99f5d6dd8cb036e7f4470df
|
||||
Subproject commit ec77d9049a719dabfd1a7c1122a209e201861944
|
||||
|
|
@ -349,11 +349,12 @@ CTaosInterface.prototype.useResult = function useResult(result) {
|
|||
return fields;
|
||||
}
|
||||
CTaosInterface.prototype.fetchBlock = function fetchBlock(result, fields) {
|
||||
let pblock = ref.ref(ref.ref(ref.NULL)); // equal to our raw data
|
||||
let num_of_rows = this.libtaos.taos_fetch_block(result, pblock)
|
||||
if (num_of_rows == 0) {
|
||||
//let pblock = ref.ref(ref.ref(ref.NULL)); // equal to our raw data
|
||||
let pblock = this.libtaos.taos_fetch_row(result);
|
||||
if (pblock == null) {
|
||||
return {block:null, num_of_rows:0};
|
||||
}
|
||||
|
||||
var fieldL = this.libtaos.taos_fetch_lengths(result);
|
||||
|
||||
let isMicro = (this.libtaos.taos_result_precision(result) == FieldTypes.C_TIMESTAMP_MICRO);
|
||||
|
|
@ -361,7 +362,6 @@ CTaosInterface.prototype.fetchBlock = function fetchBlock(result, fields) {
|
|||
var fieldlens = [];
|
||||
|
||||
if (ref.isNull(fieldL) == false) {
|
||||
|
||||
for (let i = 0; i < fields.length; i ++) {
|
||||
let plen = ref.reinterpret(fieldL, 4, i*4);
|
||||
let len = plen.readInt32LE(0);
|
||||
|
|
|
|||
|
|
@ -113,6 +113,7 @@ static void dnodeCleanupTmr() {
|
|||
int32_t dnodeInitSystem() {
|
||||
dnodeSetRunStatus(TSDB_RUN_STATUS_INITIALIZE);
|
||||
tscEmbedded = 1;
|
||||
taosIgnSIGPIPE();
|
||||
taosBlockSIGPIPE();
|
||||
taosResolveCRC();
|
||||
taosInitGlobalCfg();
|
||||
|
|
@ -120,7 +121,6 @@ int32_t dnodeInitSystem() {
|
|||
taosSetCoreDump();
|
||||
taosInitNotes();
|
||||
dnodeInitTmr();
|
||||
signal(SIGPIPE, SIG_IGN);
|
||||
|
||||
if (dnodeCreateDir(tsLogDir) < 0) {
|
||||
printf("failed to create dir: %s, reason: %s\n", tsLogDir, strerror(errno));
|
||||
|
|
|
|||
|
|
@ -54,6 +54,7 @@ void dnodeCleanupVRead() {
|
|||
void dnodeDispatchToVReadQueue(SRpcMsg *pMsg) {
|
||||
int32_t queuedMsgNum = 0;
|
||||
int32_t leftLen = pMsg->contLen;
|
||||
int32_t code = TSDB_CODE_VND_INVALID_VGROUP_ID;
|
||||
char * pCont = pMsg->pCont;
|
||||
|
||||
while (leftLen > 0) {
|
||||
|
|
@ -64,7 +65,7 @@ void dnodeDispatchToVReadQueue(SRpcMsg *pMsg) {
|
|||
assert(pHead->contLen > 0);
|
||||
void *pVnode = vnodeAcquire(pHead->vgId);
|
||||
if (pVnode != NULL) {
|
||||
int32_t code = vnodeWriteToRQueue(pVnode, pCont, pHead->contLen, TAOS_QTYPE_RPC, pMsg);
|
||||
code = vnodeWriteToRQueue(pVnode, pCont, pHead->contLen, TAOS_QTYPE_RPC, pMsg);
|
||||
if (code == TSDB_CODE_SUCCESS) queuedMsgNum++;
|
||||
vnodeRelease(pVnode);
|
||||
}
|
||||
|
|
@ -74,7 +75,7 @@ void dnodeDispatchToVReadQueue(SRpcMsg *pMsg) {
|
|||
}
|
||||
|
||||
if (queuedMsgNum == 0) {
|
||||
SRpcMsg rpcRsp = {.handle = pMsg->handle, .code = TSDB_CODE_VND_INVALID_VGROUP_ID};
|
||||
SRpcMsg rpcRsp = {.handle = pMsg->handle, .code = code};
|
||||
rpcSendResponse(&rpcRsp);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -188,6 +188,7 @@ static void *dnodeProcessVWriteQueue(void *wparam) {
|
|||
int32_t numOfMsgs;
|
||||
int32_t qtype;
|
||||
|
||||
taosBlockSIGPIPE();
|
||||
dDebug("dnode vwrite worker:%d is running", pWorker->workerId);
|
||||
|
||||
while (1) {
|
||||
|
|
|
|||
|
|
@ -28,7 +28,7 @@ extern "C" {
|
|||
default: \
|
||||
(_v) = (_finalType)GET_INT32_VAL(_data); \
|
||||
break; \
|
||||
};
|
||||
}
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
|
|
|
|||
|
|
@ -332,6 +332,9 @@ static error_t parse_opt(int key, char *arg, struct argp_state *state) {
|
|||
break;
|
||||
case 'N':
|
||||
arguments->data_batch = atoi(arg);
|
||||
if (arguments->data_batch >= INT16_MAX) {
|
||||
arguments->data_batch = INT16_MAX - 1;
|
||||
}
|
||||
break;
|
||||
case 'L':
|
||||
{
|
||||
|
|
|
|||
|
|
@ -377,6 +377,24 @@ static int32_t mnodeCreateMnodeCb(SMnodeMsg *pMsg, int32_t code) {
|
|||
return code;
|
||||
}
|
||||
|
||||
static bool mnodeAllOnline() {
|
||||
void *pIter = NULL;
|
||||
bool allOnline = true;
|
||||
|
||||
while (1) {
|
||||
SMnodeObj *pMnode = NULL;
|
||||
pIter = mnodeGetNextMnode(pIter, &pMnode);
|
||||
if (pMnode == NULL) break;
|
||||
if (pMnode->role != TAOS_SYNC_ROLE_MASTER && pMnode->role != TAOS_SYNC_ROLE_SLAVE) {
|
||||
allOnline = false;
|
||||
mnodeDecMnodeRef(pMnode);
|
||||
}
|
||||
}
|
||||
mnodeCancelGetNextMnode(pIter);
|
||||
|
||||
return allOnline;
|
||||
}
|
||||
|
||||
void mnodeCreateMnode(int32_t dnodeId, char *dnodeEp, bool needConfirm) {
|
||||
SMnodeObj *pMnode = calloc(1, sizeof(SMnodeObj));
|
||||
pMnode->mnodeId = dnodeId;
|
||||
|
|
@ -389,6 +407,11 @@ void mnodeCreateMnode(int32_t dnodeId, char *dnodeEp, bool needConfirm) {
|
|||
.fpRsp = mnodeCreateMnodeCb
|
||||
};
|
||||
|
||||
if (needConfirm && !mnodeAllOnline()) {
|
||||
mDebug("wait all mnode online then create new mnode");
|
||||
return;
|
||||
}
|
||||
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
if (needConfirm) {
|
||||
code = mnodeSendCreateMnodeMsg(dnodeId, dnodeEp);
|
||||
|
|
|
|||
|
|
@ -1081,6 +1081,8 @@ static void *sdbWorkerFp(void *pWorker) {
|
|||
int32_t qtype;
|
||||
void * unUsed;
|
||||
|
||||
taosBlockSIGPIPE();
|
||||
|
||||
while (1) {
|
||||
int32_t numOfMsgs = taosReadAllQitemsFromQset(tsSdbWQset, tsSdbWQall, &unUsed);
|
||||
if (numOfMsgs == 0) {
|
||||
|
|
|
|||
|
|
@ -659,7 +659,7 @@ static int32_t mnodeGetVgroupMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *p
|
|||
|
||||
pShow->bytes[cols] = 4;
|
||||
pSchema[cols].type = TSDB_DATA_TYPE_INT;
|
||||
strcpy(pSchema[cols].name, "onlineVnodes");
|
||||
strcpy(pSchema[cols].name, "onlines");
|
||||
pSchema[cols].bytes = htons(pShow->bytes[cols]);
|
||||
cols++;
|
||||
|
||||
|
|
@ -674,13 +674,13 @@ static int32_t mnodeGetVgroupMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *p
|
|||
for (int32_t i = 0; i < pShow->maxReplica; ++i) {
|
||||
pShow->bytes[cols] = 2;
|
||||
pSchema[cols].type = TSDB_DATA_TYPE_SMALLINT;
|
||||
snprintf(pSchema[cols].name, TSDB_COL_NAME_LEN, "v%dDnode", i + 1);
|
||||
snprintf(pSchema[cols].name, TSDB_COL_NAME_LEN, "v%d_dnode", i + 1);
|
||||
pSchema[cols].bytes = htons(pShow->bytes[cols]);
|
||||
cols++;
|
||||
|
||||
pShow->bytes[cols] = 9 + VARSTR_HEADER_SIZE;
|
||||
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
|
||||
snprintf(pSchema[cols].name, TSDB_COL_NAME_LEN, "v%dStatus", i + 1);
|
||||
snprintf(pSchema[cols].name, TSDB_COL_NAME_LEN, "v%d_status", i + 1);
|
||||
pSchema[cols].bytes = htons(pShow->bytes[cols]);
|
||||
cols++;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -59,6 +59,7 @@ extern "C" {
|
|||
|
||||
// TAOS_OS_FUNC_SOCKET
|
||||
int32_t taosSetNonblocking(SOCKET sock, int32_t on);
|
||||
void taosIgnSIGPIPE();
|
||||
void taosBlockSIGPIPE();
|
||||
|
||||
// TAOS_OS_FUNC_SOCKET_SETSOCKETOPT
|
||||
|
|
|
|||
|
|
@ -39,6 +39,10 @@ int32_t taosSetNonblocking(SOCKET sock, int32_t on) {
|
|||
return 0;
|
||||
}
|
||||
|
||||
void taosIgnSIGPIPE() {
|
||||
signal(SIGPIPE, SIG_IGN);
|
||||
}
|
||||
|
||||
void taosBlockSIGPIPE() {
|
||||
sigset_t signal_mask;
|
||||
sigemptyset(&signal_mask);
|
||||
|
|
|
|||
|
|
@ -46,6 +46,7 @@ int32_t taosSetNonblocking(SOCKET sock, int32_t on) {
|
|||
return 0;
|
||||
}
|
||||
|
||||
void taosIgnSIGPIPE() {}
|
||||
void taosBlockSIGPIPE() {}
|
||||
|
||||
int32_t taosSetSockOpt(SOCKET socketfd, int32_t level, int32_t optname, void *optval, int32_t optlen) {
|
||||
|
|
|
|||
|
|
@ -33,13 +33,6 @@ struct SColumnFilterElem;
|
|||
typedef bool (*__filter_func_t)(struct SColumnFilterElem* pFilter, char* val1, char* val2);
|
||||
typedef int32_t (*__block_search_fn_t)(char* data, int32_t num, int64_t key, int32_t order);
|
||||
|
||||
typedef struct SGroupResInfo {
|
||||
int32_t groupId;
|
||||
int32_t numOfDataPages;
|
||||
int32_t pageId;
|
||||
int32_t rowId;
|
||||
} SGroupResInfo;
|
||||
|
||||
typedef struct SResultRowPool {
|
||||
int32_t elemSize;
|
||||
int32_t blockSize;
|
||||
|
|
@ -72,6 +65,12 @@ typedef struct SResultRow {
|
|||
union {STimeWindow win; char* key;}; // start key of current time window
|
||||
} SResultRow;
|
||||
|
||||
typedef struct SGroupResInfo {
|
||||
int32_t rowId;
|
||||
int32_t index;
|
||||
SArray* pRows; // SArray<SResultRow*>
|
||||
} SGroupResInfo;
|
||||
|
||||
/**
|
||||
* If the number of generated results is greater than this value,
|
||||
* query query will be halt and return results to client immediate.
|
||||
|
|
@ -89,7 +88,6 @@ typedef struct SResultRowInfo {
|
|||
int32_t size:24; // number of result set
|
||||
int32_t capacity; // max capacity
|
||||
int32_t curIndex; // current start active index
|
||||
int64_t startTime; // start time of the first time window for sliding query
|
||||
int64_t prevSKey; // previous (not completed) sliding window start key
|
||||
} SResultRowInfo;
|
||||
|
||||
|
|
|
|||
|
|
@ -67,7 +67,7 @@ void tHistogramDestroy(SHistogramInfo** pHisto);
|
|||
|
||||
void tHistogramPrint(SHistogramInfo* pHisto);
|
||||
|
||||
int32_t vnodeHistobinarySearch(SHistBin* pEntry, int32_t len, double val);
|
||||
int32_t histoBinarySearch(SHistBin* pEntry, int32_t len, double val);
|
||||
|
||||
SHeapEntry* tHeapCreate(int32_t numOfEntries);
|
||||
void tHeapSort(SHeapEntry* pEntry, int32_t len);
|
||||
|
|
|
|||
|
|
@ -77,7 +77,6 @@ void* destroyResultRowPool(SResultRowPool* p);
|
|||
int32_t getNumOfAllocatedResultRows(SResultRowPool* p);
|
||||
int32_t getNumOfUsedResultRows(SResultRowPool* p);
|
||||
|
||||
uint64_t getResultInfoUId(SQueryRuntimeEnv* pRuntimeEnv);
|
||||
bool isPointInterpoQuery(SQuery *pQuery);
|
||||
|
||||
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load Diff
|
|
@ -158,8 +158,8 @@ int32_t tHistogramAdd(SHistogramInfo** pHisto, double val) {
|
|||
}
|
||||
|
||||
#if defined(USE_ARRAYLIST)
|
||||
int32_t idx = vnodeHistobinarySearch((*pHisto)->elems, (*pHisto)->numOfEntries, val);
|
||||
assert(idx >= 0 && idx <= (*pHisto)->maxEntries);
|
||||
int32_t idx = histoBinarySearch((*pHisto)->elems, (*pHisto)->numOfEntries, val);
|
||||
assert(idx >= 0 && idx <= (*pHisto)->maxEntries && (*pHisto)->elems != NULL);
|
||||
|
||||
if ((*pHisto)->elems[idx].val == val && idx >= 0) {
|
||||
(*pHisto)->elems[idx].num += 1;
|
||||
|
|
@ -356,7 +356,7 @@ int32_t tHistogramAdd(SHistogramInfo** pHisto, double val) {
|
|||
return 0;
|
||||
}
|
||||
|
||||
int32_t vnodeHistobinarySearch(SHistBin* pEntry, int32_t len, double val) {
|
||||
int32_t histoBinarySearch(SHistBin* pEntry, int32_t len, double val) {
|
||||
int32_t end = len - 1;
|
||||
int32_t start = 0;
|
||||
|
||||
|
|
@ -466,7 +466,7 @@ void tHistogramPrint(SHistogramInfo* pHisto) {
|
|||
*/
|
||||
int64_t tHistogramSum(SHistogramInfo* pHisto, double v) {
|
||||
#if defined(USE_ARRAYLIST)
|
||||
int32_t slotIdx = vnodeHistobinarySearch(pHisto->elems, pHisto->numOfEntries, v);
|
||||
int32_t slotIdx = histoBinarySearch(pHisto->elems, pHisto->numOfEntries, v);
|
||||
if (pHisto->elems[slotIdx].val != v) {
|
||||
slotIdx -= 1;
|
||||
|
||||
|
|
|
|||
|
|
@ -96,8 +96,6 @@ void resetResultRowInfo(SQueryRuntimeEnv *pRuntimeEnv, SResultRowInfo *pResultRo
|
|||
|
||||
pResultRowInfo->curIndex = -1;
|
||||
pResultRowInfo->size = 0;
|
||||
|
||||
pResultRowInfo->startTime = TSKEY_INITIAL_VAL;
|
||||
pResultRowInfo->prevSKey = TSKEY_INITIAL_VAL;
|
||||
}
|
||||
|
||||
|
|
@ -110,7 +108,7 @@ void popFrontResultRow(SQueryRuntimeEnv *pRuntimeEnv, SResultRowInfo *pResultRow
|
|||
assert(num >= 0 && num <= numOfClosed);
|
||||
|
||||
int16_t type = pResultRowInfo->type;
|
||||
int64_t uid = getResultInfoUId(pRuntimeEnv);
|
||||
int64_t uid = 0;
|
||||
|
||||
char *key = NULL;
|
||||
int16_t bytes = -1;
|
||||
|
|
@ -181,11 +179,12 @@ void closeAllResultRows(SResultRowInfo *pResultRowInfo) {
|
|||
assert(pResultRowInfo->size >= 0 && pResultRowInfo->capacity >= pResultRowInfo->size);
|
||||
|
||||
for (int32_t i = 0; i < pResultRowInfo->size; ++i) {
|
||||
if (pResultRowInfo->pResult[i]->closed) {
|
||||
SResultRow* pRow = pResultRowInfo->pResult[i];
|
||||
if (pRow->closed) {
|
||||
continue;
|
||||
}
|
||||
|
||||
pResultRowInfo->pResult[i]->closed = true;
|
||||
pRow->closed = true;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -383,18 +382,4 @@ void* destroyResultRowPool(SResultRowPool* p) {
|
|||
|
||||
tfree(p);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
uint64_t getResultInfoUId(SQueryRuntimeEnv* pRuntimeEnv) {
|
||||
if (!pRuntimeEnv->stableQuery) {
|
||||
return 0; // for simple table query, the uid is always set to be 0;
|
||||
}
|
||||
|
||||
SQuery* pQuery = pRuntimeEnv->pQuery;
|
||||
if (pQuery->interval.interval == 0 || isPointInterpoQuery(pQuery) || pRuntimeEnv->groupbyNormalCol) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
STableId* id = TSDB_TABLEID(pRuntimeEnv->pQuery->current->pTable);
|
||||
return id->uid;
|
||||
}
|
||||
|
|
@ -21,19 +21,19 @@ TEST(testCase, histogram_binary_search) {
|
|||
pHisto->elems[i].val = i;
|
||||
}
|
||||
|
||||
int32_t idx = vnodeHistobinarySearch(pHisto->elems, pHisto->numOfEntries, 1);
|
||||
int32_t idx = histoBinarySearch(pHisto->elems, pHisto->numOfEntries, 1);
|
||||
assert(idx == 1);
|
||||
|
||||
idx = vnodeHistobinarySearch(pHisto->elems, pHisto->numOfEntries, 9);
|
||||
idx = histoBinarySearch(pHisto->elems, pHisto->numOfEntries, 9);
|
||||
assert(idx == 9);
|
||||
|
||||
idx = vnodeHistobinarySearch(pHisto->elems, pHisto->numOfEntries, 20);
|
||||
idx = histoBinarySearch(pHisto->elems, pHisto->numOfEntries, 20);
|
||||
assert(idx == 10);
|
||||
|
||||
idx = vnodeHistobinarySearch(pHisto->elems, pHisto->numOfEntries, -1);
|
||||
idx = histoBinarySearch(pHisto->elems, pHisto->numOfEntries, -1);
|
||||
assert(idx == 0);
|
||||
|
||||
idx = vnodeHistobinarySearch(pHisto->elems, pHisto->numOfEntries, 3.9);
|
||||
idx = histoBinarySearch(pHisto->elems, pHisto->numOfEntries, 3.9);
|
||||
assert(idx == 4);
|
||||
|
||||
free(pHisto);
|
||||
|
|
|
|||
|
|
@ -38,7 +38,7 @@ extern "C" {
|
|||
#define SYNC_MAX_FWDS 512
|
||||
#define SYNC_FWD_TIMER 300
|
||||
#define SYNC_ROLE_TIMER 15000 // ms
|
||||
#define SYNC_CHECK_INTERVAL 1 // ms
|
||||
#define SYNC_CHECK_INTERVAL 1000 // ms
|
||||
#define SYNC_WAIT_AFTER_CHOOSE_MASTER 10 // ms
|
||||
|
||||
#define nodeRole pNode->peerInfo[pNode->selfIndex]->role
|
||||
|
|
@ -86,9 +86,10 @@ typedef struct SsyncPeer {
|
|||
int32_t peerFd; // forward FD
|
||||
int32_t numOfRetrieves; // number of retrieves tried
|
||||
int32_t fileChanged; // a flag to indicate file is changed during retrieving process
|
||||
int32_t refCount;
|
||||
int64_t rid;
|
||||
void * timer;
|
||||
void * pConn;
|
||||
int32_t refCount; // reference count
|
||||
struct SSyncNode *pSyncNode;
|
||||
} SSyncPeer;
|
||||
|
||||
|
|
@ -98,6 +99,7 @@ typedef struct SSyncNode {
|
|||
int8_t quorum;
|
||||
int8_t selfIndex;
|
||||
uint32_t vgId;
|
||||
int32_t refCount;
|
||||
int64_t rid;
|
||||
SSyncPeer * peerInfo[TAOS_SYNC_MAX_REPLICA + 1]; // extra one for arbitrator
|
||||
SSyncPeer * pMaster;
|
||||
|
|
@ -121,13 +123,13 @@ extern int32_t tsSyncNum;
|
|||
extern char tsNodeFqdn[TSDB_FQDN_LEN];
|
||||
extern char * syncStatus[];
|
||||
|
||||
void *syncRetrieveData(void *param);
|
||||
void *syncRestoreData(void *param);
|
||||
int32_t syncSaveIntoBuffer(SSyncPeer *pPeer, SWalHead *pHead);
|
||||
void syncRestartConnection(SSyncPeer *pPeer);
|
||||
void syncBroadcastStatus(SSyncNode *pNode);
|
||||
void syncAddPeerRef(SSyncPeer *pPeer);
|
||||
int32_t syncDecPeerRef(SSyncPeer *pPeer);
|
||||
void * syncRetrieveData(void *param);
|
||||
void * syncRestoreData(void *param);
|
||||
int32_t syncSaveIntoBuffer(SSyncPeer *pPeer, SWalHead *pHead);
|
||||
void syncRestartConnection(SSyncPeer *pPeer);
|
||||
void syncBroadcastStatus(SSyncNode *pNode);
|
||||
SSyncPeer *syncAcquirePeer(int64_t rid);
|
||||
void syncReleasePeer(SSyncPeer *pPeer);
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
|
|
|
|||
|
|
@ -25,14 +25,14 @@ typedef struct {
|
|||
uint32_t serverIp;
|
||||
int16_t port;
|
||||
int32_t bufferSize;
|
||||
void (*processBrokenLink)(void *ahandle);
|
||||
int32_t (*processIncomingMsg)(void *ahandle, void *buffer);
|
||||
void (*processBrokenLink)(int64_t handleId);
|
||||
int32_t (*processIncomingMsg)(int64_t handleId, void *buffer);
|
||||
void (*processIncomingConn)(int32_t fd, uint32_t ip);
|
||||
} SPoolInfo;
|
||||
|
||||
void *syncOpenTcpThreadPool(SPoolInfo *pInfo);
|
||||
void syncCloseTcpThreadPool(void *);
|
||||
void *syncAllocateTcpConn(void *, void *ahandle, int32_t connFd);
|
||||
void *syncAllocateTcpConn(void *, int64_t rid, int32_t connFd);
|
||||
void syncFreeTcpConn(void *);
|
||||
|
||||
#ifdef __cplusplus
|
||||
|
|
|
|||
|
|
@ -29,8 +29,8 @@
|
|||
|
||||
static void arbSignalHandler(int32_t signum, siginfo_t *sigInfo, void *context);
|
||||
static void arbProcessIncommingConnection(int32_t connFd, uint32_t sourceIp);
|
||||
static void arbProcessBrokenLink(void *param);
|
||||
static int32_t arbProcessPeerMsg(void *param, void *buffer);
|
||||
static void arbProcessBrokenLink(int64_t rid);
|
||||
static int32_t arbProcessPeerMsg(int64_t rid, void *buffer);
|
||||
static tsem_t tsArbSem;
|
||||
static void * tsArbTcpPool;
|
||||
|
||||
|
|
@ -138,20 +138,20 @@ static void arbProcessIncommingConnection(int32_t connFd, uint32_t sourceIp) {
|
|||
|
||||
sDebug("%s, arbitrator request is accepted", pNode->id);
|
||||
pNode->nodeFd = connFd;
|
||||
pNode->pConn = syncAllocateTcpConn(tsArbTcpPool, pNode, connFd);
|
||||
pNode->pConn = syncAllocateTcpConn(tsArbTcpPool, (int64_t)pNode, connFd);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
static void arbProcessBrokenLink(void *param) {
|
||||
SNodeConn *pNode = param;
|
||||
static void arbProcessBrokenLink(int64_t rid) {
|
||||
SNodeConn *pNode = (SNodeConn *)rid;
|
||||
|
||||
sDebug("%s, TCP link is broken since %s, close connection", pNode->id, strerror(errno));
|
||||
tfree(pNode);
|
||||
}
|
||||
|
||||
static int32_t arbProcessPeerMsg(void *param, void *buffer) {
|
||||
SNodeConn *pNode = param;
|
||||
static int32_t arbProcessPeerMsg(int64_t rid, void *buffer) {
|
||||
SNodeConn *pNode = (SNodeConn *)rid;
|
||||
SSyncHead head;
|
||||
int32_t bytes = 0;
|
||||
char * cont = (char *)buffer;
|
||||
|
|
|
|||
|
|
@ -35,19 +35,21 @@ char tsNodeFqdn[TSDB_FQDN_LEN] = {0};
|
|||
static void * tsTcpPool = NULL;
|
||||
static void * tsSyncTmrCtrl = NULL;
|
||||
static void * tsVgIdHash = NULL;
|
||||
static int32_t tsSyncRefId = -1;
|
||||
static int32_t tsNodeRefId = -1;
|
||||
static int32_t tsPeerRefId = -1;
|
||||
|
||||
// local functions
|
||||
static void syncProcessSyncRequest(char *pMsg, SSyncPeer *pPeer);
|
||||
static void syncRecoverFromMaster(SSyncPeer *pPeer);
|
||||
static void syncCheckPeerConnection(void *param, void *tmrId);
|
||||
static void syncSendPeersStatusMsgToPeer(SSyncPeer *pPeer, char ack, int8_t type, uint16_t tranId);
|
||||
static void syncProcessBrokenLink(void *param);
|
||||
static int32_t syncProcessPeerMsg(void *param, void *buffer);
|
||||
static int32_t syncSendPeersStatusMsgToPeer(SSyncPeer *pPeer, char ack, int8_t type, uint16_t tranId);
|
||||
static void syncProcessBrokenLink(int64_t rid);
|
||||
static int32_t syncProcessPeerMsg(int64_t rid, void *buffer);
|
||||
static void syncProcessIncommingConnection(int32_t connFd, uint32_t sourceIp);
|
||||
static void syncRemovePeer(SSyncPeer *pPeer);
|
||||
static void syncAddArbitrator(SSyncNode *pNode);
|
||||
static void syncFreeNode(void *);
|
||||
static void syncFreePeer(void *);
|
||||
static void syncRemoveConfirmedFwdInfo(SSyncNode *pNode);
|
||||
static void syncMonitorFwdInfos(void *param, void *tmrId);
|
||||
static void syncMonitorNodeRole(void *param, void *tmrId);
|
||||
|
|
@ -55,7 +57,12 @@ static void syncProcessFwdAck(SSyncNode *pNode, SFwdInfo *pFwdInfo, int32_t c
|
|||
static int32_t syncSaveFwdInfo(SSyncNode *pNode, uint64_t version, void *mhandle);
|
||||
static void syncRestartPeer(SSyncPeer *pPeer);
|
||||
static int32_t syncForwardToPeerImpl(SSyncNode *pNode, void *data, void *mhandle, int32_t qtyp);
|
||||
|
||||
static SSyncPeer *syncAddPeer(SSyncNode *pNode, const SNodeInfo *pInfo);
|
||||
static void syncStartCheckPeerConn(SSyncPeer *pPeer);
|
||||
static void syncStopCheckPeerConn(SSyncPeer *pPeer);
|
||||
static SSyncNode *syncAcquireNode(int64_t rid);
|
||||
static void syncReleaseNode(SSyncNode *pNode);
|
||||
|
||||
char* syncRole[] = {
|
||||
"offline",
|
||||
|
|
@ -87,29 +94,34 @@ int32_t syncInit() {
|
|||
tsTcpPool = syncOpenTcpThreadPool(&info);
|
||||
if (tsTcpPool == NULL) {
|
||||
sError("failed to init tcpPool");
|
||||
syncCleanUp();
|
||||
return -1;
|
||||
}
|
||||
|
||||
tsSyncTmrCtrl = taosTmrInit(1000, 50, 10000, "SYNC");
|
||||
if (tsSyncTmrCtrl == NULL) {
|
||||
sError("failed to init tmrCtrl");
|
||||
syncCloseTcpThreadPool(tsTcpPool);
|
||||
tsTcpPool = NULL;
|
||||
syncCleanUp();
|
||||
return -1;
|
||||
}
|
||||
|
||||
tsVgIdHash = taosHashInit(TSDB_MIN_VNODES, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
|
||||
if (tsVgIdHash == NULL) {
|
||||
sError("failed to init vgIdHash");
|
||||
taosTmrCleanUp(tsSyncTmrCtrl);
|
||||
syncCloseTcpThreadPool(tsTcpPool);
|
||||
tsTcpPool = NULL;
|
||||
tsSyncTmrCtrl = NULL;
|
||||
syncCleanUp();
|
||||
return -1;
|
||||
}
|
||||
|
||||
tsSyncRefId = taosOpenRef(200, syncFreeNode);
|
||||
if (tsSyncRefId < 0) {
|
||||
tsNodeRefId = taosOpenRef(200, syncFreeNode);
|
||||
if (tsNodeRefId < 0) {
|
||||
sError("failed to init node ref");
|
||||
syncCleanUp();
|
||||
return -1;
|
||||
}
|
||||
|
||||
tsPeerRefId = taosOpenRef(1000, syncFreePeer);
|
||||
if (tsPeerRefId < 0) {
|
||||
sError("failed to init peer ref");
|
||||
syncCleanUp();
|
||||
return -1;
|
||||
}
|
||||
|
|
@ -121,12 +133,12 @@ int32_t syncInit() {
|
|||
}
|
||||
|
||||
void syncCleanUp() {
|
||||
if (tsTcpPool) {
|
||||
if (tsTcpPool != NULL) {
|
||||
syncCloseTcpThreadPool(tsTcpPool);
|
||||
tsTcpPool = NULL;
|
||||
}
|
||||
|
||||
if (tsSyncTmrCtrl) {
|
||||
if (tsSyncTmrCtrl != NULL) {
|
||||
taosTmrCleanUp(tsSyncTmrCtrl);
|
||||
tsSyncTmrCtrl = NULL;
|
||||
}
|
||||
|
|
@ -136,8 +148,15 @@ void syncCleanUp() {
|
|||
tsVgIdHash = NULL;
|
||||
}
|
||||
|
||||
taosCloseRef(tsSyncRefId);
|
||||
tsSyncRefId = -1;
|
||||
if (tsNodeRefId != -1) {
|
||||
taosCloseRef(tsNodeRefId);
|
||||
tsNodeRefId = -1;
|
||||
}
|
||||
|
||||
if (tsPeerRefId != -1) {
|
||||
taosCloseRef(tsPeerRefId);
|
||||
tsPeerRefId = -1;
|
||||
}
|
||||
|
||||
sInfo("sync module is cleaned up");
|
||||
}
|
||||
|
|
@ -170,7 +189,8 @@ int64_t syncStart(const SSyncInfo *pInfo) {
|
|||
pNode->quorum = pCfg->quorum;
|
||||
if (pNode->quorum > pNode->replica) pNode->quorum = pNode->replica;
|
||||
|
||||
pNode->rid = taosAddRef(tsSyncRefId, pNode);
|
||||
pNode->refCount = 1;
|
||||
pNode->rid = taosAddRef(tsNodeRefId, pNode);
|
||||
if (pNode->rid < 0) {
|
||||
syncFreeNode(pNode);
|
||||
return -1;
|
||||
|
|
@ -232,13 +252,18 @@ int64_t syncStart(const SSyncInfo *pInfo) {
|
|||
(*pNode->notifyRole)(pNode->vgId, nodeRole);
|
||||
}
|
||||
|
||||
syncStartCheckPeerConn(pNode->peerInfo[TAOS_SYNC_MAX_REPLICA]); // arb
|
||||
for (int32_t index = 0; index < pNode->replica; ++index) {
|
||||
syncStartCheckPeerConn(pNode->peerInfo[index]);
|
||||
}
|
||||
|
||||
return pNode->rid;
|
||||
}
|
||||
|
||||
void syncStop(int64_t rid) {
|
||||
SSyncPeer *pPeer;
|
||||
|
||||
SSyncNode *pNode = taosAcquireRef(tsSyncRefId, rid);
|
||||
SSyncNode *pNode = syncAcquireNode(rid);
|
||||
if (pNode == NULL) return;
|
||||
|
||||
sInfo("vgId:%d, cleanup sync", pNode->vgId);
|
||||
|
|
@ -259,14 +284,14 @@ void syncStop(int64_t rid) {
|
|||
|
||||
pthread_mutex_unlock(&pNode->mutex);
|
||||
|
||||
taosReleaseRef(tsSyncRefId, rid);
|
||||
taosRemoveRef(tsSyncRefId, rid);
|
||||
syncReleaseNode(pNode);
|
||||
taosRemoveRef(tsNodeRefId, rid);
|
||||
}
|
||||
|
||||
int32_t syncReconfig(int64_t rid, const SSyncCfg *pNewCfg) {
|
||||
int32_t i, j;
|
||||
|
||||
SSyncNode *pNode = taosAcquireRef(tsSyncRefId, rid);
|
||||
SSyncNode *pNode = syncAcquireNode(rid);
|
||||
if (pNode == NULL) return TSDB_CODE_SYN_INVALID_CONFIG;
|
||||
|
||||
sInfo("vgId:%d, reconfig, role:%s replica:%d old:%d", pNode->vgId, syncRole[nodeRole], pNewCfg->replica,
|
||||
|
|
@ -274,6 +299,11 @@ int32_t syncReconfig(int64_t rid, const SSyncCfg *pNewCfg) {
|
|||
|
||||
pthread_mutex_lock(&pNode->mutex);
|
||||
|
||||
syncStopCheckPeerConn(pNode->peerInfo[TAOS_SYNC_MAX_REPLICA]); // arb
|
||||
for (int32_t index = 0; index < pNode->replica; ++index) {
|
||||
syncStopCheckPeerConn(pNode->peerInfo[index]);
|
||||
}
|
||||
|
||||
for (i = 0; i < pNode->replica; ++i) {
|
||||
for (j = 0; j < pNewCfg->replica; ++j) {
|
||||
if ((strcmp(pNode->peerInfo[i]->fqdn, pNewCfg->nodeInfo[j].nodeFqdn) == 0) &&
|
||||
|
|
@ -330,28 +360,32 @@ int32_t syncReconfig(int64_t rid, const SSyncCfg *pNewCfg) {
|
|||
(*pNode->notifyRole)(pNode->vgId, nodeRole);
|
||||
}
|
||||
|
||||
syncStartCheckPeerConn(pNode->peerInfo[TAOS_SYNC_MAX_REPLICA]); // arb
|
||||
for (int32_t index = 0; index < pNode->replica; ++index) {
|
||||
syncStartCheckPeerConn(pNode->peerInfo[index]);
|
||||
}
|
||||
|
||||
pthread_mutex_unlock(&pNode->mutex);
|
||||
|
||||
sInfo("vgId:%d, %d replicas are configured, quorum:%d", pNode->vgId, pNode->replica, pNode->quorum);
|
||||
syncBroadcastStatus(pNode);
|
||||
|
||||
taosReleaseRef(tsSyncRefId, rid);
|
||||
syncReleaseNode(pNode);
|
||||
return 0;
|
||||
}
|
||||
|
||||
int32_t syncForwardToPeer(int64_t rid, void *data, void *mhandle, int32_t qtype) {
|
||||
SSyncNode *pNode = taosAcquireRef(tsSyncRefId, rid);
|
||||
if (pNode == NULL) return 0;
|
||||
SSyncNode *pNode = syncAcquireNode(rid);
|
||||
if (pNode == NULL) return 0;
|
||||
|
||||
int32_t code = syncForwardToPeerImpl(pNode, data, mhandle, qtype);
|
||||
|
||||
taosReleaseRef(tsSyncRefId, rid);
|
||||
|
||||
syncReleaseNode(pNode);
|
||||
return code;
|
||||
}
|
||||
|
||||
void syncConfirmForward(int64_t rid, uint64_t version, int32_t code) {
|
||||
SSyncNode *pNode = taosAcquireRef(tsSyncRefId, rid);
|
||||
SSyncNode *pNode = syncAcquireNode(rid);
|
||||
if (pNode == NULL) return;
|
||||
|
||||
SSyncPeer *pPeer = pNode->pMaster;
|
||||
|
|
@ -367,14 +401,14 @@ void syncConfirmForward(int64_t rid, uint64_t version, int32_t code) {
|
|||
}
|
||||
}
|
||||
|
||||
taosReleaseRef(tsSyncRefId, rid);
|
||||
syncReleaseNode(pNode);
|
||||
}
|
||||
|
||||
#if 0
|
||||
void syncRecover(int64_t rid) {
|
||||
SSyncPeer *pPeer;
|
||||
|
||||
SSyncNode *pNode = taosAcquireRef(tsSyncRefId, rid);
|
||||
SSyncNode *pNode = syncAcquireNode(rid);
|
||||
if (pNode == NULL) return;
|
||||
|
||||
// to do: add a few lines to check if recover is OK
|
||||
|
|
@ -395,12 +429,12 @@ void syncRecover(int64_t rid) {
|
|||
|
||||
pthread_mutex_unlock(&pNode->mutex);
|
||||
|
||||
taosReleaseRef(tsSyncRefId, rid);
|
||||
syncReleaseNode(pNode);
|
||||
}
|
||||
#endif
|
||||
|
||||
int32_t syncGetNodesRole(int64_t rid, SNodesRole *pNodesRole) {
|
||||
SSyncNode *pNode = taosAcquireRef(tsSyncRefId, rid);
|
||||
SSyncNode *pNode = syncAcquireNode(rid);
|
||||
if (pNode == NULL) return -1;
|
||||
|
||||
pNodesRole->selfIndex = pNode->selfIndex;
|
||||
|
|
@ -409,8 +443,7 @@ int32_t syncGetNodesRole(int64_t rid, SNodesRole *pNodesRole) {
|
|||
pNodesRole->role[i] = pNode->peerInfo[i]->role;
|
||||
}
|
||||
|
||||
taosReleaseRef(tsSyncRefId, rid);
|
||||
|
||||
syncReleaseNode(pNode);
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
|
@ -446,24 +479,61 @@ static void syncAddArbitrator(SSyncNode *pNode) {
|
|||
static void syncFreeNode(void *param) {
|
||||
SSyncNode *pNode = param;
|
||||
|
||||
int32_t refCount = atomic_sub_fetch_32(&pNode->refCount, 1);
|
||||
sDebug("vgId:%d, syncnode is freed, refCount:%d", pNode->vgId, refCount);
|
||||
|
||||
pthread_mutex_destroy(&pNode->mutex);
|
||||
tfree(pNode->pRecv);
|
||||
tfree(pNode->pSyncFwds);
|
||||
tfree(pNode);
|
||||
}
|
||||
|
||||
void syncAddPeerRef(SSyncPeer *pPeer) { atomic_add_fetch_32(&pPeer->refCount, 1); }
|
||||
|
||||
int32_t syncDecPeerRef(SSyncPeer *pPeer) {
|
||||
if (atomic_sub_fetch_32(&pPeer->refCount, 1) == 0) {
|
||||
taosReleaseRef(tsSyncRefId, pPeer->pSyncNode->rid);
|
||||
|
||||
sDebug("%s, resource is freed", pPeer->id);
|
||||
tfree(pPeer);
|
||||
return 0;
|
||||
SSyncNode *syncAcquireNode(int64_t rid) {
|
||||
SSyncNode *pNode = taosAcquireRef(tsNodeRefId, rid);
|
||||
if (pNode == NULL) {
|
||||
sDebug("failed to acquire syncnode from refId:%" PRId64, rid);
|
||||
} else {
|
||||
int32_t refCount = atomic_add_fetch_32(&pNode->refCount, 1);
|
||||
sTrace("vgId:%d, acquire syncnode refId:%" PRId64 ", refCount:%d", pNode->vgId, rid, refCount);
|
||||
}
|
||||
|
||||
return 1;
|
||||
return pNode;
|
||||
}
|
||||
|
||||
void syncReleaseNode(SSyncNode *pNode) {
|
||||
int32_t refCount = atomic_sub_fetch_32(&pNode->refCount, 1);
|
||||
sTrace("vgId:%d, dec syncnode refId:%" PRId64 " refCount:%d", pNode->vgId, pNode->rid, refCount);
|
||||
|
||||
taosReleaseRef(tsNodeRefId, pNode->rid);
|
||||
}
|
||||
|
||||
static void syncFreePeer(void *param) {
|
||||
SSyncPeer *pPeer = param;
|
||||
|
||||
int32_t refCount = atomic_sub_fetch_32(&pPeer->refCount, 1);
|
||||
sDebug("%s, peer is freed, refCount:%d", pPeer->id, refCount);
|
||||
|
||||
syncReleaseNode(pPeer->pSyncNode);
|
||||
tfree(pPeer);
|
||||
}
|
||||
|
||||
SSyncPeer *syncAcquirePeer(int64_t rid) {
|
||||
SSyncPeer *pPeer = taosAcquireRef(tsPeerRefId, rid);
|
||||
if (pPeer == NULL) {
|
||||
sDebug("failed to acquire peer from refId:%" PRId64, rid);
|
||||
} else {
|
||||
int32_t refCount = atomic_add_fetch_32(&pPeer->refCount, 1);
|
||||
sTrace("%s, acquire peer refId:%" PRId64 ", refCount:%d", pPeer->id, rid, refCount);
|
||||
}
|
||||
|
||||
return pPeer;
|
||||
}
|
||||
|
||||
void syncReleasePeer(SSyncPeer *pPeer) {
|
||||
int32_t refCount = atomic_sub_fetch_32(&pPeer->refCount, 1);
|
||||
sTrace("%s, dec peer refId:%" PRId64 ", refCount:%d", pPeer->id, pPeer->rid, refCount);
|
||||
|
||||
taosReleaseRef(tsPeerRefId, pPeer->rid);
|
||||
}
|
||||
|
||||
static void syncClosePeerConn(SSyncPeer *pPeer) {
|
||||
|
|
@ -473,7 +543,8 @@ static void syncClosePeerConn(SSyncPeer *pPeer) {
|
|||
taosClose(pPeer->syncFd);
|
||||
if (pPeer->peerFd >= 0) {
|
||||
pPeer->peerFd = -1;
|
||||
syncFreeTcpConn(pPeer->pConn);
|
||||
void *pConn = pPeer->pConn;
|
||||
if (pConn != NULL) syncFreeTcpConn(pPeer->pConn);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -482,7 +553,28 @@ static void syncRemovePeer(SSyncPeer *pPeer) {
|
|||
|
||||
pPeer->ip = 0;
|
||||
syncClosePeerConn(pPeer);
|
||||
syncDecPeerRef(pPeer);
|
||||
//taosRemoveRef(tsPeerRefId, pPeer->rid);
|
||||
syncReleasePeer(pPeer);
|
||||
}
|
||||
|
||||
static void syncStartCheckPeerConn(SSyncPeer *pPeer) {
|
||||
if (pPeer == NULL) return;
|
||||
SSyncNode *pNode = pPeer->pSyncNode;
|
||||
|
||||
int32_t ret = strcmp(pPeer->fqdn, tsNodeFqdn);
|
||||
if (pPeer->nodeId == 0 || (ret > 0) || (ret == 0 && pPeer->port > tsSyncPort)) {
|
||||
int32_t checkMs = 100 + (pNode->vgId * 10) % 100;
|
||||
if (pNode->vgId > 1) checkMs = tsStatusInterval * 1000 + checkMs;
|
||||
sDebug("%s, check peer connection after %d ms", pPeer->id, checkMs);
|
||||
taosTmrReset(syncCheckPeerConnection, checkMs, (void *)pPeer->rid, tsSyncTmrCtrl, &pPeer->timer);
|
||||
}
|
||||
}
|
||||
|
||||
static void syncStopCheckPeerConn(SSyncPeer *pPeer) {
|
||||
if (pPeer == NULL) return;
|
||||
|
||||
taosTmrStopA(&pPeer->timer);
|
||||
sDebug("%s, stop check peer connection", pPeer->id);
|
||||
}
|
||||
|
||||
static SSyncPeer *syncAddPeer(SSyncNode *pNode, const SNodeInfo *pInfo) {
|
||||
|
|
@ -508,17 +600,11 @@ static SSyncPeer *syncAddPeer(SSyncNode *pNode, const SNodeInfo *pInfo) {
|
|||
pPeer->role = TAOS_SYNC_ROLE_OFFLINE;
|
||||
pPeer->pSyncNode = pNode;
|
||||
pPeer->refCount = 1;
|
||||
pPeer->rid = taosAddRef(tsPeerRefId, pPeer);
|
||||
|
||||
sInfo("%s, it is configured, ep:%s:%u", pPeer->id, pPeer->fqdn, pPeer->port);
|
||||
int32_t ret = strcmp(pPeer->fqdn, tsNodeFqdn);
|
||||
if (pPeer->nodeId == 0 || (ret > 0) || (ret == 0 && pPeer->port > tsSyncPort)) {
|
||||
int32_t checkMs = 100 + (pNode->vgId * 10) % 100;
|
||||
if (pNode->vgId > 1) checkMs = tsStatusInterval * 1000 + checkMs;
|
||||
sDebug("%s, check peer connection after %d ms", pPeer->id, checkMs);
|
||||
taosTmrReset(syncCheckPeerConnection, checkMs, pPeer, tsSyncTmrCtrl, &pPeer->timer);
|
||||
}
|
||||
sInfo("%s, %p it is configured, ep:%s:%u rid:%" PRId64, pPeer->id, pPeer, pPeer->fqdn, pPeer->port, pPeer->rid);
|
||||
|
||||
taosAcquireRef(tsSyncRefId, pNode->rid);
|
||||
(void)syncAcquireNode(pNode->rid);
|
||||
return pPeer;
|
||||
}
|
||||
|
||||
|
|
@ -560,6 +646,9 @@ static void syncChooseMaster(SSyncNode *pNode) {
|
|||
index = i;
|
||||
}
|
||||
}
|
||||
sDebug("vgId:%d, master:%s may be choosed, index:%d", pNode->vgId, pNode->peerInfo[index]->id, index);
|
||||
} else {
|
||||
sDebug("vgId:%d, no master election since onlineNum:%d replica:%d", pNode->vgId, onlineNum, pNode->replica);
|
||||
}
|
||||
|
||||
// add arbitrator connection
|
||||
|
|
@ -580,6 +669,11 @@ static void syncChooseMaster(SSyncNode *pNode) {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (index >= 0) {
|
||||
sDebug("vgId:%d, master:%s may be choosed, index:%d onlineNum(arb):%d replica:%d", pNode->vgId,
|
||||
pNode->peerInfo[index]->id, index, onlineNum, replica);
|
||||
}
|
||||
}
|
||||
|
||||
if (index >= 0) {
|
||||
|
|
@ -621,9 +715,13 @@ static SSyncPeer *syncCheckMaster(SSyncNode *pNode) {
|
|||
|
||||
if (onlineNum <= replica * 0.5) {
|
||||
if (nodeRole != TAOS_SYNC_ROLE_UNSYNCED) {
|
||||
nodeRole = TAOS_SYNC_ROLE_UNSYNCED;
|
||||
if (nodeRole == TAOS_SYNC_ROLE_MASTER && onlineNum == replica * 0.5 && onlineNum >= 1) {
|
||||
sInfo("vgId:%d, self keep work as master, online:%d replica:%d", pNode->vgId, onlineNum, replica);
|
||||
} else {
|
||||
nodeRole = TAOS_SYNC_ROLE_UNSYNCED;
|
||||
sInfo("vgId:%d, self change to unsynced state, online:%d replica:%d", pNode->vgId, onlineNum, replica);
|
||||
}
|
||||
(*pNode->notifyRole)(pNode->vgId, nodeRole);
|
||||
sInfo("vgId:%d, self change to unsynced state, online:%d replica:%d", pNode->vgId, onlineNum, replica);
|
||||
}
|
||||
} else {
|
||||
for (int32_t index = 0; index < pNode->replica; ++index) {
|
||||
|
|
@ -678,7 +776,7 @@ static void syncCheckRole(SSyncPeer *pPeer, SPeerStatus* peersStatus, int8_t new
|
|||
if (pMaster) {
|
||||
// master is there
|
||||
pNode->pMaster = pMaster;
|
||||
sDebug("%s, it is the master, sver:%" PRIu64, pMaster->id, pMaster->version);
|
||||
sDebug("%s, it is the master, replica:%d sver:%" PRIu64, pMaster->id, pNode->replica, pMaster->version);
|
||||
|
||||
if (syncValidateMaster(pPeer) < 0) return;
|
||||
|
||||
|
|
@ -711,10 +809,10 @@ static void syncCheckRole(SSyncPeer *pPeer, SPeerStatus* peersStatus, int8_t new
|
|||
}
|
||||
|
||||
if (consistent) {
|
||||
sDebug("vgId:%d, choose master", pNode->vgId);
|
||||
sDebug("vgId:%d, choose master, replica:%d", pNode->vgId, pNode->replica);
|
||||
syncChooseMaster(pNode);
|
||||
} else {
|
||||
sDebug("vgId:%d, cannot choose master since roles inequality", pNode->vgId);
|
||||
sDebug("vgId:%d, cannot choose master since roles inequality, replica:%d", pNode->vgId, pNode->replica);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -743,7 +841,7 @@ static void syncRestartPeer(SSyncPeer *pPeer) {
|
|||
int32_t ret = strcmp(pPeer->fqdn, tsNodeFqdn);
|
||||
if (ret > 0 || (ret == 0 && pPeer->port > tsSyncPort)) {
|
||||
sDebug("%s, check peer connection in 1000 ms", pPeer->id);
|
||||
taosTmrReset(syncCheckPeerConnection, SYNC_CHECK_INTERVAL, pPeer, tsSyncTmrCtrl, &pPeer->timer);
|
||||
taosTmrReset(syncCheckPeerConnection, SYNC_CHECK_INTERVAL, (void *)pPeer->rid, tsSyncTmrCtrl, &pPeer->timer);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -772,25 +870,30 @@ static void syncProcessSyncRequest(char *msg, SSyncPeer *pPeer) {
|
|||
}
|
||||
|
||||
// start a new thread to retrieve the data
|
||||
syncAddPeerRef(pPeer);
|
||||
(void)syncAcquirePeer(pPeer->rid);
|
||||
|
||||
pthread_attr_t thattr;
|
||||
pthread_t thread;
|
||||
pthread_attr_init(&thattr);
|
||||
pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_DETACHED);
|
||||
int32_t ret = pthread_create(&thread, &thattr, syncRetrieveData, pPeer);
|
||||
int32_t ret = pthread_create(&thread, &thattr, syncRetrieveData, (void *)pPeer->rid);
|
||||
pthread_attr_destroy(&thattr);
|
||||
|
||||
if (ret != 0) {
|
||||
sError("%s, failed to create sync thread since %s", pPeer->id, strerror(errno));
|
||||
syncDecPeerRef(pPeer);
|
||||
} else {
|
||||
pPeer->sstatus = TAOS_SYNC_STATUS_START;
|
||||
sDebug("%s, thread is created to retrieve data, set sstatus:%s", pPeer->id, syncStatus[pPeer->sstatus]);
|
||||
}
|
||||
|
||||
syncReleasePeer(pPeer);
|
||||
}
|
||||
|
||||
static void syncNotStarted(void *param, void *tmrId) {
|
||||
SSyncPeer *pPeer = param;
|
||||
int64_t rid = (int64_t)param;
|
||||
SSyncPeer *pPeer = syncAcquirePeer(rid);
|
||||
if (pPeer == NULL) return;
|
||||
|
||||
SSyncNode *pNode = pPeer->pSyncNode;
|
||||
|
||||
pthread_mutex_lock(&pNode->mutex);
|
||||
|
|
@ -799,15 +902,22 @@ static void syncNotStarted(void *param, void *tmrId) {
|
|||
sInfo("%s, sync conn is still not up, restart and set sstatus:%s", pPeer->id, syncStatus[pPeer->sstatus]);
|
||||
syncRestartConnection(pPeer);
|
||||
pthread_mutex_unlock(&pNode->mutex);
|
||||
|
||||
syncReleasePeer(pPeer);
|
||||
}
|
||||
|
||||
static void syncTryRecoverFromMaster(void *param, void *tmrId) {
|
||||
SSyncPeer *pPeer = param;
|
||||
int64_t rid = (int64_t)param;
|
||||
SSyncPeer *pPeer = syncAcquirePeer(rid);
|
||||
if (pPeer == NULL) return;
|
||||
|
||||
SSyncNode *pNode = pPeer->pSyncNode;
|
||||
|
||||
pthread_mutex_lock(&pNode->mutex);
|
||||
syncRecoverFromMaster(pPeer);
|
||||
pthread_mutex_unlock(&pNode->mutex);
|
||||
|
||||
syncReleasePeer(pPeer);
|
||||
}
|
||||
|
||||
static void syncRecoverFromMaster(SSyncPeer *pPeer) {
|
||||
|
|
@ -823,7 +933,7 @@ static void syncRecoverFromMaster(SSyncPeer *pPeer) {
|
|||
// Ensure the sync of mnode not interrupted
|
||||
if (pNode->vgId != 1 && tsSyncNum >= SYNC_MAX_NUM) {
|
||||
sInfo("%s, %d syncs are in process, try later", pPeer->id, tsSyncNum);
|
||||
taosTmrReset(syncTryRecoverFromMaster, 500 + (pNode->vgId * 10) % 200, pPeer, tsSyncTmrCtrl, &pPeer->timer);
|
||||
taosTmrReset(syncTryRecoverFromMaster, 500 + (pNode->vgId * 10) % 200, (void *)pPeer->rid, tsSyncTmrCtrl, &pPeer->timer);
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
@ -832,7 +942,7 @@ static void syncRecoverFromMaster(SSyncPeer *pPeer) {
|
|||
SSyncMsg msg;
|
||||
syncBuildSyncReqMsg(&msg, pNode->vgId);
|
||||
|
||||
taosTmrReset(syncNotStarted, SYNC_CHECK_INTERVAL, pPeer, tsSyncTmrCtrl, &pPeer->timer);
|
||||
taosTmrReset(syncNotStarted, SYNC_CHECK_INTERVAL, (void *)pPeer->rid, tsSyncTmrCtrl, &pPeer->timer);
|
||||
|
||||
if (taosWriteMsg(pPeer->peerFd, &msg, sizeof(SSyncMsg)) != sizeof(SSyncMsg)) {
|
||||
sError("%s, failed to send sync-req to peer", pPeer->id);
|
||||
|
|
@ -920,8 +1030,10 @@ static int32_t syncReadPeerMsg(SSyncPeer *pPeer, SSyncHead *pHead) {
|
|||
return 0;
|
||||
}
|
||||
|
||||
static int32_t syncProcessPeerMsg(void *param, void *buffer) {
|
||||
SSyncPeer *pPeer = param;
|
||||
static int32_t syncProcessPeerMsg(int64_t rid, void *buffer) {
|
||||
SSyncPeer *pPeer = syncAcquirePeer(rid);
|
||||
if (pPeer == NULL) return -1;
|
||||
|
||||
SSyncHead *pHead = buffer;
|
||||
SSyncNode *pNode = pPeer->pSyncNode;
|
||||
|
||||
|
|
@ -942,12 +1054,17 @@ static int32_t syncProcessPeerMsg(void *param, void *buffer) {
|
|||
}
|
||||
|
||||
pthread_mutex_unlock(&pNode->mutex);
|
||||
syncReleasePeer(pPeer);
|
||||
|
||||
return code;
|
||||
}
|
||||
|
||||
static void syncSendPeersStatusMsgToPeer(SSyncPeer *pPeer, char ack, int8_t type, uint16_t tranId) {
|
||||
if (pPeer->peerFd < 0 || pPeer->ip == 0) return;
|
||||
static int32_t syncSendPeersStatusMsgToPeer(SSyncPeer *pPeer, char ack, int8_t type, uint16_t tranId) {
|
||||
if (pPeer->peerFd < 0 || pPeer->ip == 0) {
|
||||
sDebug("%s, failed to send status msg, restart fd:%d", pPeer->id, pPeer->peerFd);
|
||||
syncRestartConnection(pPeer);
|
||||
return -1;
|
||||
}
|
||||
|
||||
SSyncNode * pNode = pPeer->pSyncNode;
|
||||
SPeersStatus msg;
|
||||
|
|
@ -970,9 +1087,11 @@ static void syncSendPeersStatusMsgToPeer(SSyncPeer *pPeer, char ack, int8_t type
|
|||
sDebug("%s, status is sent, self:%s:%s:%" PRIu64 ", peer:%s:%s:%" PRIu64 ", ack:%d tranId:%u type:%s pfd:%d",
|
||||
pPeer->id, syncRole[nodeRole], syncStatus[nodeSStatus], nodeVersion, syncRole[pPeer->role],
|
||||
syncStatus[pPeer->sstatus], pPeer->version, ack, tranId, statusType[type], pPeer->peerFd);
|
||||
return 0;
|
||||
} else {
|
||||
sDebug("%s, failed to send status msg, restart", pPeer->id);
|
||||
syncRestartConnection(pPeer);
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -989,7 +1108,7 @@ static void syncSetupPeerConnection(SSyncPeer *pPeer) {
|
|||
int32_t connFd = taosOpenTcpClientSocket(pPeer->ip, pPeer->port, 0);
|
||||
if (connFd < 0) {
|
||||
sDebug("%s, failed to open tcp socket since %s", pPeer->id, strerror(errno));
|
||||
taosTmrReset(syncCheckPeerConnection, SYNC_CHECK_INTERVAL, pPeer, tsSyncTmrCtrl, &pPeer->timer);
|
||||
taosTmrReset(syncCheckPeerConnection, SYNC_CHECK_INTERVAL, (void *)pPeer->rid, tsSyncTmrCtrl, &pPeer->timer);
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
@ -1000,17 +1119,19 @@ static void syncSetupPeerConnection(SSyncPeer *pPeer) {
|
|||
sDebug("%s, connection to peer server is setup, pfd:%d sfd:%d tranId:%u", pPeer->id, connFd, pPeer->syncFd, msg.tranId);
|
||||
pPeer->peerFd = connFd;
|
||||
pPeer->role = TAOS_SYNC_ROLE_UNSYNCED;
|
||||
pPeer->pConn = syncAllocateTcpConn(tsTcpPool, pPeer, connFd);
|
||||
syncAddPeerRef(pPeer);
|
||||
pPeer->pConn = syncAllocateTcpConn(tsTcpPool, pPeer->rid, connFd);
|
||||
} else {
|
||||
sDebug("%s, failed to setup peer connection to server since %s, try later", pPeer->id, strerror(errno));
|
||||
taosClose(connFd);
|
||||
taosTmrReset(syncCheckPeerConnection, SYNC_CHECK_INTERVAL, pPeer, tsSyncTmrCtrl, &pPeer->timer);
|
||||
taosTmrReset(syncCheckPeerConnection, SYNC_CHECK_INTERVAL, (void *)pPeer->rid, tsSyncTmrCtrl, &pPeer->timer);
|
||||
}
|
||||
}
|
||||
|
||||
static void syncCheckPeerConnection(void *param, void *tmrId) {
|
||||
SSyncPeer *pPeer = param;
|
||||
int64_t rid = (int64_t)param;
|
||||
SSyncPeer *pPeer = syncAcquirePeer(rid);
|
||||
if (pPeer == NULL) return;
|
||||
|
||||
SSyncNode *pNode = pPeer->pSyncNode;
|
||||
|
||||
pthread_mutex_lock(&pNode->mutex);
|
||||
|
|
@ -1019,6 +1140,8 @@ static void syncCheckPeerConnection(void *param, void *tmrId) {
|
|||
syncSetupPeerConnection(pPeer);
|
||||
|
||||
pthread_mutex_unlock(&pNode->mutex);
|
||||
|
||||
syncReleasePeer(pPeer);
|
||||
}
|
||||
|
||||
static void syncCreateRestoreDataThread(SSyncPeer *pPeer) {
|
||||
|
|
@ -1029,8 +1152,9 @@ static void syncCreateRestoreDataThread(SSyncPeer *pPeer) {
|
|||
pthread_attr_init(&thattr);
|
||||
pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_DETACHED);
|
||||
|
||||
syncAddPeerRef(pPeer);
|
||||
int32_t ret = pthread_create(&(thread), &thattr, (void *)syncRestoreData, pPeer);
|
||||
(void)syncAcquirePeer(pPeer->rid);
|
||||
|
||||
int32_t ret = pthread_create(&(thread), &thattr, (void *)syncRestoreData, (void *)pPeer->rid);
|
||||
pthread_attr_destroy(&thattr);
|
||||
|
||||
if (ret < 0) {
|
||||
|
|
@ -1038,10 +1162,11 @@ static void syncCreateRestoreDataThread(SSyncPeer *pPeer) {
|
|||
nodeSStatus = TAOS_SYNC_STATUS_INIT;
|
||||
sError("%s, failed to create sync thread, set sstatus:%s", pPeer->id, syncStatus[nodeSStatus]);
|
||||
taosClose(pPeer->syncFd);
|
||||
syncDecPeerRef(pPeer);
|
||||
} else {
|
||||
sInfo("%s, sync connection is up", pPeer->id);
|
||||
}
|
||||
|
||||
syncReleasePeer(pPeer);
|
||||
}
|
||||
|
||||
static void syncProcessIncommingConnection(int32_t connFd, uint32_t sourceIp) {
|
||||
|
|
@ -1073,7 +1198,7 @@ static void syncProcessIncommingConnection(int32_t connFd, uint32_t sourceIp) {
|
|||
return;
|
||||
}
|
||||
|
||||
sDebug("vgId:%d, sync msg is received, tranId:%u", vgId, msg.tranId);
|
||||
sDebug("vgId:%d, sync connection is incomming, tranId:%u", vgId, msg.tranId);
|
||||
|
||||
SSyncNode *pNode = *ppNode;
|
||||
pthread_mutex_lock(&pNode->mutex);
|
||||
|
|
@ -1101,8 +1226,7 @@ static void syncProcessIncommingConnection(int32_t connFd, uint32_t sourceIp) {
|
|||
sDebug("%s, TCP connection is up, pfd:%d sfd:%d, old pfd:%d", pPeer->id, connFd, pPeer->syncFd, pPeer->peerFd);
|
||||
syncClosePeerConn(pPeer);
|
||||
pPeer->peerFd = connFd;
|
||||
pPeer->pConn = syncAllocateTcpConn(tsTcpPool, pPeer, connFd);
|
||||
syncAddPeerRef(pPeer);
|
||||
pPeer->pConn = syncAllocateTcpConn(tsTcpPool, pPeer->rid, connFd);
|
||||
sDebug("%s, ready to exchange data", pPeer->id);
|
||||
syncSendPeersStatusMsgToPeer(pPeer, 1, SYNC_STATUS_EXCHANGE_DATA, syncGenTranId());
|
||||
}
|
||||
|
|
@ -1111,23 +1235,21 @@ static void syncProcessIncommingConnection(int32_t connFd, uint32_t sourceIp) {
|
|||
pthread_mutex_unlock(&pNode->mutex);
|
||||
}
|
||||
|
||||
static void syncProcessBrokenLink(void *param) {
|
||||
if (param == NULL) return; // the connection for arbitrator
|
||||
SSyncPeer *pPeer = param;
|
||||
static void syncProcessBrokenLink(int64_t rid) {
|
||||
SSyncPeer *pPeer = syncAcquirePeer(rid);
|
||||
if (pPeer == NULL) return;
|
||||
|
||||
SSyncNode *pNode = pPeer->pSyncNode;
|
||||
|
||||
if (taosAcquireRef(tsSyncRefId, pNode->rid) == NULL) return;
|
||||
pthread_mutex_lock(&pNode->mutex);
|
||||
|
||||
sDebug("%s, TCP link is broken since %s, pfd:%d sfd:%d", pPeer->id, strerror(errno), pPeer->peerFd, pPeer->syncFd);
|
||||
pPeer->peerFd = -1;
|
||||
|
||||
if (syncDecPeerRef(pPeer) != 0) {
|
||||
syncRestartConnection(pPeer);
|
||||
}
|
||||
|
||||
syncRestartConnection(pPeer);
|
||||
pthread_mutex_unlock(&pNode->mutex);
|
||||
taosReleaseRef(tsSyncRefId, pNode->rid);
|
||||
|
||||
syncReleasePeer(pPeer);
|
||||
}
|
||||
|
||||
static int32_t syncSaveFwdInfo(SSyncNode *pNode, uint64_t version, void *mhandle) {
|
||||
|
|
@ -1198,7 +1320,7 @@ static void syncProcessFwdAck(SSyncNode *pNode, SFwdInfo *pFwdInfo, int32_t code
|
|||
|
||||
static void syncMonitorNodeRole(void *param, void *tmrId) {
|
||||
int64_t rid = (int64_t)param;
|
||||
SSyncNode *pNode = taosAcquireRef(tsSyncRefId, rid);
|
||||
SSyncNode *pNode = syncAcquireNode(rid);
|
||||
if (pNode == NULL) return;
|
||||
|
||||
for (int32_t index = 0; index < pNode->replica; index++) {
|
||||
|
|
@ -1215,12 +1337,12 @@ static void syncMonitorNodeRole(void *param, void *tmrId) {
|
|||
}
|
||||
|
||||
pNode->pRoleTimer = taosTmrStart(syncMonitorNodeRole, SYNC_ROLE_TIMER, (void *)pNode->rid, tsSyncTmrCtrl);
|
||||
taosReleaseRef(tsSyncRefId, rid);
|
||||
syncReleaseNode(pNode);
|
||||
}
|
||||
|
||||
static void syncMonitorFwdInfos(void *param, void *tmrId) {
|
||||
int64_t rid = (int64_t)param;
|
||||
SSyncNode *pNode = taosAcquireRef(tsSyncRefId, rid);
|
||||
SSyncNode *pNode = syncAcquireNode(rid);
|
||||
if (pNode == NULL) return;
|
||||
|
||||
SSyncFwds *pSyncFwds = pNode->pSyncFwds;
|
||||
|
|
@ -1246,7 +1368,7 @@ static void syncMonitorFwdInfos(void *param, void *tmrId) {
|
|||
pNode->pFwdTimer = taosTmrStart(syncMonitorFwdInfos, SYNC_FWD_TIMER, (void *)pNode->rid, tsSyncTmrCtrl);
|
||||
}
|
||||
|
||||
taosReleaseRef(tsSyncRefId, rid);
|
||||
syncReleaseNode(pNode);
|
||||
}
|
||||
|
||||
static int32_t syncForwardToPeerImpl(SSyncNode *pNode, void *data, void *mhandle, int32_t qtype) {
|
||||
|
|
|
|||
|
|
@ -90,15 +90,18 @@ static int32_t syncRestoreFile(SSyncPeer *pPeer, uint64_t *fversion) {
|
|||
break;
|
||||
}
|
||||
|
||||
sDebug("%s, file:%s info is received from master, index:%d size:%" PRId64 " fver:%" PRIu64 " magic:%d", pPeer->id,
|
||||
minfo.name, minfo.index, minfo.size, minfo.fversion, minfo.magic);
|
||||
|
||||
// remove extra files on slave between the current and last index
|
||||
syncRemoveExtraFile(pPeer, pindex + 1, minfo.index - 1);
|
||||
pindex = minfo.index;
|
||||
|
||||
// check the file info
|
||||
sinfo = minfo;
|
||||
sDebug("%s, get file:%s info size:%" PRId64, pPeer->id, minfo.name, minfo.size);
|
||||
sinfo.magic = (*pNode->getFileInfo)(pNode->vgId, sinfo.name, &sinfo.index, TAOS_SYNC_MAX_INDEX, &sinfo.size,
|
||||
&sinfo.fversion);
|
||||
sinfo.magic = (*pNode->getFileInfo)(pNode->vgId, sinfo.name, &sinfo.index, TAOS_SYNC_MAX_INDEX, &sinfo.size, &sinfo.fversion);
|
||||
sDebug("%s, local file:%s info, index:%d size:%" PRId64 " fver:%" PRIu64 " magic:%d", pPeer->id, sinfo.name,
|
||||
sinfo.index, sinfo.size, sinfo.fversion, sinfo.magic);
|
||||
|
||||
// if file not there or magic is not the same, file shall be synced
|
||||
memset(&fileAck, 0, sizeof(SFileAck));
|
||||
|
|
@ -116,6 +119,8 @@ static int32_t syncRestoreFile(SSyncPeer *pPeer, uint64_t *fversion) {
|
|||
if (fileAck.sync == 0) {
|
||||
sDebug("%s, %s is the same", pPeer->id, minfo.name);
|
||||
continue;
|
||||
} else {
|
||||
sDebug("%s, %s will be received, size:%" PRId64, pPeer->id, minfo.name, minfo.size);
|
||||
}
|
||||
|
||||
// if sync is required, open file, receive from master, and write to file
|
||||
|
|
@ -155,7 +160,7 @@ static int32_t syncRestoreFile(SSyncPeer *pPeer, uint64_t *fversion) {
|
|||
return code;
|
||||
}
|
||||
|
||||
static int32_t syncRestoreWal(SSyncPeer *pPeer) {
|
||||
static int32_t syncRestoreWal(SSyncPeer *pPeer, uint64_t *wver) {
|
||||
SSyncNode *pNode = pPeer->pSyncNode;
|
||||
int32_t ret, code = -1;
|
||||
uint64_t lastVer = 0;
|
||||
|
|
@ -198,6 +203,7 @@ static int32_t syncRestoreWal(SSyncPeer *pPeer) {
|
|||
}
|
||||
|
||||
free(pHead);
|
||||
*wver = lastVer;
|
||||
return code;
|
||||
}
|
||||
|
||||
|
|
@ -321,12 +327,19 @@ static int32_t syncRestoreDataStepByStep(SSyncPeer *pPeer) {
|
|||
|
||||
nodeVersion = fversion;
|
||||
|
||||
sInfo("%s, start to restore wal", pPeer->id);
|
||||
if (syncRestoreWal(pPeer) < 0) {
|
||||
sError("%s, failed to restore wal", pPeer->id);
|
||||
sInfo("%s, start to restore wal, fver:%" PRIu64, pPeer->id, nodeVersion);
|
||||
uint64_t wver = 0;
|
||||
code = syncRestoreWal(pPeer, &wver); // lastwar
|
||||
if (code < 0) {
|
||||
sError("%s, failed to restore wal, code:%d", pPeer->id, code);
|
||||
return -1;
|
||||
}
|
||||
|
||||
if (wver != 0) {
|
||||
nodeVersion = wver;
|
||||
sDebug("%s, restore wal finished, set sver:%" PRIu64, pPeer->id, nodeVersion);
|
||||
}
|
||||
|
||||
nodeSStatus = TAOS_SYNC_STATUS_CACHE;
|
||||
sInfo("%s, start to insert buffered points, set sstatus:%s", pPeer->id, syncStatus[nodeSStatus]);
|
||||
if (syncProcessBufferedFwd(pPeer) < 0) {
|
||||
|
|
@ -338,7 +351,10 @@ static int32_t syncRestoreDataStepByStep(SSyncPeer *pPeer) {
|
|||
}
|
||||
|
||||
void *syncRestoreData(void *param) {
|
||||
SSyncPeer *pPeer = param;
|
||||
int64_t rid = (int64_t)param;
|
||||
SSyncPeer *pPeer = syncAcquirePeer(rid);
|
||||
if (pPeer == NULL) return NULL;
|
||||
|
||||
SSyncNode *pNode = pPeer->pSyncNode;
|
||||
|
||||
taosBlockSIGPIPE();
|
||||
|
|
@ -369,7 +385,7 @@ void *syncRestoreData(void *param) {
|
|||
taosClose(pPeer->syncFd);
|
||||
syncCloseRecvBuffer(pNode);
|
||||
__sync_fetch_and_sub(&tsSyncNum, 1);
|
||||
syncDecPeerRef(pPeer);
|
||||
syncReleasePeer(pPeer);
|
||||
|
||||
return NULL;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -104,7 +104,8 @@ static int32_t syncRetrieveFile(SSyncPeer *pPeer) {
|
|||
fileInfo.magic = (*pNode->getFileInfo)(pNode->vgId, fileInfo.name, &fileInfo.index, TAOS_SYNC_MAX_INDEX,
|
||||
&fileInfo.size, &fileInfo.fversion);
|
||||
syncBuildFileInfo(&fileInfo, pNode->vgId);
|
||||
sDebug("%s, file:%s info is sent, size:%" PRId64, pPeer->id, fileInfo.name, fileInfo.size);
|
||||
sDebug("%s, file:%s info is sent, index:%d size:%" PRId64 " fver:%" PRIu64 " magic:%d", pPeer->id, fileInfo.name,
|
||||
fileInfo.index, fileInfo.size, fileInfo.fversion, fileInfo.magic);
|
||||
|
||||
// send the file info
|
||||
int32_t ret = taosWriteMsg(pPeer->syncFd, &(fileInfo), sizeof(SFileInfo));
|
||||
|
|
@ -144,6 +145,8 @@ static int32_t syncRetrieveFile(SSyncPeer *pPeer) {
|
|||
fileInfo.index++;
|
||||
sDebug("%s, %s is the same", pPeer->id, fileInfo.name);
|
||||
continue;
|
||||
} else {
|
||||
sDebug("%s, %s will be sent", pPeer->id, fileInfo.name);
|
||||
}
|
||||
|
||||
// get the full path to file
|
||||
|
|
@ -461,7 +464,10 @@ static int32_t syncRetrieveDataStepByStep(SSyncPeer *pPeer) {
|
|||
}
|
||||
|
||||
void *syncRetrieveData(void *param) {
|
||||
SSyncPeer *pPeer = (SSyncPeer *)param;
|
||||
int64_t rid = (int64_t)param;
|
||||
SSyncPeer *pPeer = syncAcquirePeer(rid);
|
||||
if (pPeer == NULL) return NULL;
|
||||
|
||||
SSyncNode *pNode = pPeer->pSyncNode;
|
||||
taosBlockSIGPIPE();
|
||||
|
||||
|
|
@ -490,7 +496,7 @@ void *syncRetrieveData(void *param) {
|
|||
|
||||
pPeer->fileChanged = 0;
|
||||
taosClose(pPeer->syncFd);
|
||||
syncDecPeerRef(pPeer);
|
||||
syncReleasePeer(pPeer);
|
||||
|
||||
return NULL;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -42,7 +42,7 @@ typedef struct SPoolObj {
|
|||
|
||||
typedef struct {
|
||||
SThreadObj *pThread;
|
||||
void * ahandle;
|
||||
int64_t handleId;
|
||||
int32_t fd;
|
||||
int32_t closedByApp;
|
||||
} SConnObj;
|
||||
|
|
@ -112,7 +112,7 @@ void syncCloseTcpThreadPool(void *param) {
|
|||
tfree(pPool);
|
||||
}
|
||||
|
||||
void *syncAllocateTcpConn(void *param, void *pPeer, int32_t connFd) {
|
||||
void *syncAllocateTcpConn(void *param, int64_t rid, int32_t connFd) {
|
||||
struct epoll_event event;
|
||||
SPoolObj *pPool = param;
|
||||
|
||||
|
|
@ -130,7 +130,7 @@ void *syncAllocateTcpConn(void *param, void *pPeer, int32_t connFd) {
|
|||
|
||||
pConn->fd = connFd;
|
||||
pConn->pThread = pThread;
|
||||
pConn->ahandle = pPeer;
|
||||
pConn->handleId = rid;
|
||||
pConn->closedByApp = 0;
|
||||
|
||||
event.events = EPOLLIN | EPOLLRDHUP;
|
||||
|
|
@ -164,7 +164,7 @@ static void taosProcessBrokenLink(SConnObj *pConn) {
|
|||
SPoolInfo * pInfo = &pPool->info;
|
||||
|
||||
if (pConn->closedByApp == 0) shutdown(pConn->fd, SHUT_WR);
|
||||
(*pInfo->processBrokenLink)(pConn->ahandle);
|
||||
(*pInfo->processBrokenLink)(pConn->handleId);
|
||||
|
||||
pThread->numOfFds--;
|
||||
epoll_ctl(pThread->pollFd, EPOLL_CTL_DEL, pConn->fd, NULL);
|
||||
|
|
@ -221,7 +221,7 @@ static void *syncProcessTcpData(void *param) {
|
|||
}
|
||||
|
||||
if (pConn->closedByApp == 0) {
|
||||
if ((*pInfo->processIncomingMsg)(pConn->ahandle, buffer) < 0) {
|
||||
if ((*pInfo->processIncomingMsg)(pConn->handleId, buffer) < 0) {
|
||||
syncFreeTcpConn(pConn);
|
||||
continue;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -31,19 +31,19 @@
|
|||
extern "C" {
|
||||
#endif
|
||||
|
||||
extern int tsdbDebugFlag;
|
||||
extern uint32_t tsdbDebugFlag;
|
||||
|
||||
#define tsdbFatal(...) { if (tsdbDebugFlag & DEBUG_FATAL) { taosPrintLog("TDB FATAL ", 255, __VA_ARGS__); }}
|
||||
#define tsdbError(...) { if (tsdbDebugFlag & DEBUG_ERROR) { taosPrintLog("TDB ERROR ", 255, __VA_ARGS__); }}
|
||||
#define tsdbWarn(...) { if (tsdbDebugFlag & DEBUG_WARN) { taosPrintLog("TDB WARN ", 255, __VA_ARGS__); }}
|
||||
#define tsdbInfo(...) { if (tsdbDebugFlag & DEBUG_INFO) { taosPrintLog("TDB ", 255, __VA_ARGS__); }}
|
||||
#define tsdbDebug(...) { if (tsdbDebugFlag & DEBUG_DEBUG) { taosPrintLog("TDB ", tsdbDebugFlag, __VA_ARGS__); }}
|
||||
#define tsdbTrace(...) { if (tsdbDebugFlag & DEBUG_TRACE) { taosPrintLog("TDB ", tsdbDebugFlag, __VA_ARGS__); }}
|
||||
#define tsdbFatal(...) do { if (tsdbDebugFlag & DEBUG_FATAL) { taosPrintLog("TDB FATAL ", 255, __VA_ARGS__); }} while(0)
|
||||
#define tsdbError(...) do { if (tsdbDebugFlag & DEBUG_ERROR) { taosPrintLog("TDB ERROR ", 255, __VA_ARGS__); }} while(0)
|
||||
#define tsdbWarn(...) do { if (tsdbDebugFlag & DEBUG_WARN) { taosPrintLog("TDB WARN ", 255, __VA_ARGS__); }} while(0)
|
||||
#define tsdbInfo(...) do { if (tsdbDebugFlag & DEBUG_INFO) { taosPrintLog("TDB ", 255, __VA_ARGS__); }} while(0)
|
||||
#define tsdbDebug(...) do { if (tsdbDebugFlag & DEBUG_DEBUG) { taosPrintLog("TDB ", tsdbDebugFlag, __VA_ARGS__); }} while(0)
|
||||
#define tsdbTrace(...) do { if (tsdbDebugFlag & DEBUG_TRACE) { taosPrintLog("TDB ", tsdbDebugFlag, __VA_ARGS__); }} while(0)
|
||||
|
||||
#define TSDB_MAX_TABLE_SCHEMAS 16
|
||||
#define TSDB_FILE_HEAD_SIZE 512
|
||||
#define TSDB_FILE_DELIMITER 0xF00AFA0F
|
||||
#define TSDB_FILE_INIT_MAGIC 0xFFFFFFFF
|
||||
#define TSDB_FILE_HEAD_SIZE 512
|
||||
#define TSDB_FILE_DELIMITER 0xF00AFA0F
|
||||
#define TSDB_FILE_INIT_MAGIC 0xFFFFFFFF
|
||||
|
||||
#define TAOS_IN_RANGE(key, keyMin, keyLast) (((key) >= (keyMin)) && ((key) <= (keyMax)))
|
||||
|
||||
|
|
|
|||
|
|
@ -161,6 +161,11 @@ _err:
|
|||
|
||||
static void tsdbEndCommit(STsdbRepo *pRepo, int eno) {
|
||||
if (pRepo->appH.notifyStatus) pRepo->appH.notifyStatus(pRepo->appH.appH, TSDB_STATUS_COMMIT_OVER, eno);
|
||||
SMemTable *pIMem = pRepo->imem;
|
||||
tsdbLockRepo(pRepo);
|
||||
pRepo->imem = NULL;
|
||||
tsdbUnlockRepo(pRepo);
|
||||
tsdbUnRefMemTable(pRepo, pIMem);
|
||||
sem_post(&(pRepo->readyToCommit));
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@
|
|||
#include "tsdbMain.h"
|
||||
|
||||
#define TSDB_DATA_SKIPLIST_LEVEL 5
|
||||
#define TSDB_MAX_INSERT_BATCH 512
|
||||
|
||||
static SMemTable * tsdbNewMemTable(STsdbRepo *pRepo);
|
||||
static void tsdbFreeMemTable(SMemTable *pMemTable);
|
||||
|
|
@ -205,7 +206,7 @@ void *tsdbAllocBytes(STsdbRepo *pRepo, int bytes) {
|
|||
int tsdbAsyncCommit(STsdbRepo *pRepo) {
|
||||
if (pRepo->mem == NULL) return 0;
|
||||
|
||||
SMemTable *pIMem = pRepo->imem;
|
||||
ASSERT(pRepo->imem == NULL);
|
||||
|
||||
sem_wait(&(pRepo->readyToCommit));
|
||||
|
||||
|
|
@ -220,8 +221,6 @@ int tsdbAsyncCommit(STsdbRepo *pRepo) {
|
|||
tsdbScheduleCommit(pRepo);
|
||||
if (tsdbUnlockRepo(pRepo) < 0) return -1;
|
||||
|
||||
if (tsdbUnRefMemTable(pRepo, pIMem) < 0) return -1;
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
|
@ -606,19 +605,13 @@ static int tsdbInsertDataToTable(STsdbRepo *pRepo, SSubmitBlk *pBlock, int32_t *
|
|||
STable * pTable = NULL;
|
||||
SSubmitBlkIter blkIter = {0};
|
||||
SDataRow row = NULL;
|
||||
void ** rows = NULL;
|
||||
void * rows[TSDB_MAX_INSERT_BATCH] = {0};
|
||||
int rowCounter = 0;
|
||||
|
||||
ASSERT(pBlock->tid < pMeta->maxTables);
|
||||
pTable = pMeta->tables[pBlock->tid];
|
||||
ASSERT(pTable != NULL && TABLE_UID(pTable) == pBlock->uid);
|
||||
|
||||
rows = (void **)calloc(pBlock->numOfRows, sizeof(void *));
|
||||
if (rows == NULL) {
|
||||
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
|
||||
return -1;
|
||||
}
|
||||
|
||||
tsdbInitSubmitBlkIter(pBlock, &blkIter);
|
||||
while ((row = tsdbGetSubmitBlkNext(&blkIter)) != NULL) {
|
||||
if (tsdbCopyRowToMem(pRepo, row, pTable, &(rows[rowCounter])) < 0) {
|
||||
|
|
@ -632,9 +625,18 @@ static int tsdbInsertDataToTable(STsdbRepo *pRepo, SSubmitBlk *pBlock, int32_t *
|
|||
if (rows[rowCounter] != NULL) {
|
||||
rowCounter++;
|
||||
}
|
||||
|
||||
if (rowCounter == TSDB_MAX_INSERT_BATCH) {
|
||||
if (tsdbInsertDataToTableImpl(pRepo, pTable, rows, rowCounter) < 0) {
|
||||
goto _err;
|
||||
}
|
||||
|
||||
rowCounter = 0;
|
||||
memset(rows, 0, sizeof(rows));
|
||||
}
|
||||
}
|
||||
|
||||
if (tsdbInsertDataToTableImpl(pRepo, pTable, rows, rowCounter) < 0) {
|
||||
if (rowCounter > 0 && tsdbInsertDataToTableImpl(pRepo, pTable, rows, rowCounter) < 0) {
|
||||
goto _err;
|
||||
}
|
||||
|
||||
|
|
@ -642,11 +644,9 @@ static int tsdbInsertDataToTable(STsdbRepo *pRepo, SSubmitBlk *pBlock, int32_t *
|
|||
pRepo->stat.pointsWritten += points * schemaNCols(pSchema);
|
||||
pRepo->stat.totalStorage += points * schemaVLen(pSchema);
|
||||
|
||||
free(rows);
|
||||
return 0;
|
||||
|
||||
_err:
|
||||
free(rows);
|
||||
return -1;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -956,9 +956,9 @@ static int32_t loadFileDataBlock(STsdbQueryHandle* pQueryHandle, SCompBlock* pBl
|
|||
return code;
|
||||
}
|
||||
|
||||
SDataCols* pTSCol = pQueryHandle->rhelper.pDataCols[0];
|
||||
SDataCols* pTsCol = pQueryHandle->rhelper.pDataCols[0];
|
||||
if (pCheckInfo->lastKey < pBlock->keyLast) {
|
||||
cur->pos = binarySearchForKey(pTSCol->cols[0].pData, pBlock->numOfRows, pCheckInfo->lastKey, pQueryHandle->order);
|
||||
cur->pos = binarySearchForKey(pTsCol->cols[0].pData, pBlock->numOfRows, pCheckInfo->lastKey, pQueryHandle->order);
|
||||
} else {
|
||||
cur->pos = pBlock->numOfRows - 1;
|
||||
}
|
||||
|
|
@ -1704,7 +1704,32 @@ static int32_t createDataBlocksInfo(STsdbQueryHandle* pQueryHandle, int32_t numO
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static int32_t getDataBlocksInFilesImpl(STsdbQueryHandle* pQueryHandle, bool* exists) {
|
||||
static int32_t getFirstFileDataBlock(STsdbQueryHandle* pQueryHandle, bool* exists);
|
||||
|
||||
static int32_t getDataBlockRv(STsdbQueryHandle* pQueryHandle, STableBlockInfo* pNext, bool *exists) {
|
||||
int32_t step = ASCENDING_TRAVERSE(pQueryHandle->order)? 1 : -1;
|
||||
SQueryFilePos* cur = &pQueryHandle->cur;
|
||||
|
||||
while(1) {
|
||||
int32_t code = loadFileDataBlock(pQueryHandle, pNext->compBlock, pNext->pTableCheckInfo, exists);
|
||||
if (code != TSDB_CODE_SUCCESS || *exists) {
|
||||
return code;
|
||||
}
|
||||
|
||||
if ((cur->slot == pQueryHandle->numOfBlocks - 1 && ASCENDING_TRAVERSE(pQueryHandle->order)) ||
|
||||
(cur->slot == 0 && !ASCENDING_TRAVERSE(pQueryHandle->order))) {
|
||||
// all data blocks in current file has been checked already, try next file if exists
|
||||
return getFirstFileDataBlock(pQueryHandle, exists);
|
||||
} else { // next block of the same file
|
||||
cur->slot += step;
|
||||
cur->mixBlock = false;
|
||||
cur->blockCompleted = false;
|
||||
pNext = &pQueryHandle->pDataBlockInfo[cur->slot];
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
static int32_t getFirstFileDataBlock(STsdbQueryHandle* pQueryHandle, bool* exists) {
|
||||
pQueryHandle->numOfBlocks = 0;
|
||||
SQueryFilePos* cur = &pQueryHandle->cur;
|
||||
|
||||
|
|
@ -1789,7 +1814,23 @@ static int32_t getDataBlocksInFilesImpl(STsdbQueryHandle* pQueryHandle, bool* ex
|
|||
cur->fid = pQueryHandle->pFileGroup->fileId;
|
||||
|
||||
STableBlockInfo* pBlockInfo = &pQueryHandle->pDataBlockInfo[cur->slot];
|
||||
return loadFileDataBlock(pQueryHandle, pBlockInfo->compBlock, pBlockInfo->pTableCheckInfo, exists);
|
||||
return getDataBlockRv(pQueryHandle, pBlockInfo, exists);
|
||||
}
|
||||
|
||||
static bool isEndFileDataBlock(SQueryFilePos* cur, int32_t numOfBlocks, bool ascTrav) {
|
||||
assert(cur != NULL && numOfBlocks > 0);
|
||||
return (cur->slot == numOfBlocks - 1 && ascTrav) || (cur->slot == 0 && !ascTrav);
|
||||
}
|
||||
|
||||
static void moveToNextDataBlockInCurrentFile(STsdbQueryHandle* pQueryHandle) {
|
||||
int32_t step = ASCENDING_TRAVERSE(pQueryHandle->order)? 1 : -1;
|
||||
|
||||
SQueryFilePos* cur = &pQueryHandle->cur;
|
||||
assert(cur->slot < pQueryHandle->numOfBlocks && cur->slot >= 0);
|
||||
|
||||
cur->slot += step;
|
||||
cur->mixBlock = false;
|
||||
cur->blockCompleted = false;
|
||||
}
|
||||
|
||||
static int32_t getDataBlocksInFiles(STsdbQueryHandle* pQueryHandle, bool* exists) {
|
||||
|
|
@ -1800,14 +1841,14 @@ static int32_t getDataBlocksInFiles(STsdbQueryHandle* pQueryHandle, bool* exists
|
|||
if (!pQueryHandle->locateStart) {
|
||||
pQueryHandle->locateStart = true;
|
||||
STsdbCfg* pCfg = &pQueryHandle->pTsdb->config;
|
||||
int32_t fid = getFileIdFromKey(pQueryHandle->window.skey, pCfg->daysPerFile, pCfg->precision);
|
||||
int32_t fid = getFileIdFromKey(pQueryHandle->window.skey, pCfg->daysPerFile, pCfg->precision);
|
||||
|
||||
pthread_rwlock_rdlock(&pQueryHandle->pTsdb->tsdbFileH->fhlock);
|
||||
tsdbInitFileGroupIter(pFileHandle, &pQueryHandle->fileIter, pQueryHandle->order);
|
||||
tsdbSeekFileGroupIter(&pQueryHandle->fileIter, fid);
|
||||
pthread_rwlock_unlock(&pQueryHandle->pTsdb->tsdbFileH->fhlock);
|
||||
|
||||
return getDataBlocksInFilesImpl(pQueryHandle, exists);
|
||||
return getFirstFileDataBlock(pQueryHandle, exists);
|
||||
} else {
|
||||
// check if current file block is all consumed
|
||||
STableBlockInfo* pBlockInfo = &pQueryHandle->pDataBlockInfo[cur->slot];
|
||||
|
|
@ -1815,27 +1856,26 @@ static int32_t getDataBlocksInFiles(STsdbQueryHandle* pQueryHandle, bool* exists
|
|||
|
||||
// current block is done, try next
|
||||
if ((!cur->mixBlock) || cur->blockCompleted) {
|
||||
if ((cur->slot == pQueryHandle->numOfBlocks - 1 && ASCENDING_TRAVERSE(pQueryHandle->order)) ||
|
||||
(cur->slot == 0 && !ASCENDING_TRAVERSE(pQueryHandle->order))) {
|
||||
// all data blocks in current file has been checked already, try next file if exists
|
||||
return getDataBlocksInFilesImpl(pQueryHandle, exists);
|
||||
} else {
|
||||
// next block of the same file
|
||||
int32_t step = ASCENDING_TRAVERSE(pQueryHandle->order) ? 1 : -1;
|
||||
cur->slot += step;
|
||||
|
||||
cur->mixBlock = false;
|
||||
cur->blockCompleted = false;
|
||||
|
||||
STableBlockInfo* pNext = &pQueryHandle->pDataBlockInfo[cur->slot];
|
||||
return loadFileDataBlock(pQueryHandle, pNext->compBlock, pNext->pTableCheckInfo, exists);
|
||||
}
|
||||
// all data blocks in current file has been checked already, try next file if exists
|
||||
} else {
|
||||
tsdbDebug("%p continue in current data block, index:%d, pos:%d, %p", pQueryHandle, cur->slot, cur->pos, pQueryHandle->qinfo);
|
||||
tsdbDebug("%p continue in current data block, index:%d, pos:%d, %p", pQueryHandle, cur->slot, cur->pos,
|
||||
pQueryHandle->qinfo);
|
||||
int32_t code = handleDataMergeIfNeeded(pQueryHandle, pBlockInfo->compBlock, pCheckInfo);
|
||||
*exists = pQueryHandle->realNumOfRows > 0;
|
||||
*exists = (pQueryHandle->realNumOfRows > 0);
|
||||
|
||||
return code;
|
||||
if (code != TSDB_CODE_SUCCESS || *exists) {
|
||||
return code;
|
||||
}
|
||||
}
|
||||
|
||||
// current block is empty, try next block in file
|
||||
// all data blocks in current file has been checked already, try next file if exists
|
||||
if (isEndFileDataBlock(cur, pQueryHandle->numOfBlocks, ASCENDING_TRAVERSE(pQueryHandle->order))) {
|
||||
return getFirstFileDataBlock(pQueryHandle, exists);
|
||||
} else {
|
||||
moveToNextDataBlockInCurrentFile(pQueryHandle);
|
||||
STableBlockInfo* pNext = &pQueryHandle->pDataBlockInfo[cur->slot];
|
||||
return getDataBlockRv(pQueryHandle, pNext, exists);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -52,6 +52,8 @@ void *taosIterateRef(int rsetId, int64_t rid);
|
|||
// return the number of references in system
|
||||
int taosListRef();
|
||||
|
||||
#define RID_VALID(x) ((x) > 0)
|
||||
|
||||
/* sample code to iterate the refs
|
||||
|
||||
void demoIterateRefs(int rsetId) {
|
||||
|
|
|
|||
|
|
@ -547,13 +547,14 @@ void taosAddToTrashcan(SCacheObj *pCacheObj, SCacheDataNode *pNode) {
|
|||
return;
|
||||
}
|
||||
|
||||
__cache_wr_lock(pCacheObj);
|
||||
STrashElem *pElem = calloc(1, sizeof(STrashElem));
|
||||
pElem->pData = pNode;
|
||||
pElem->prev = NULL;
|
||||
pElem->prev = NULL;
|
||||
pElem->next = NULL;
|
||||
pNode->inTrashcan = true;
|
||||
pNode->pTNodeHeader = pElem;
|
||||
|
||||
__cache_wr_lock(pCacheObj);
|
||||
pElem->next = pCacheObj->pTrash;
|
||||
if (pCacheObj->pTrash) {
|
||||
pCacheObj->pTrash->prev = pElem;
|
||||
|
|
@ -563,8 +564,8 @@ void taosAddToTrashcan(SCacheObj *pCacheObj, SCacheDataNode *pNode) {
|
|||
pCacheObj->numOfElemsInTrash++;
|
||||
__cache_unlock(pCacheObj);
|
||||
|
||||
uDebug("cache:%s key:%p, %p move to trashcan, pTrashElem:%p, numOfElem in trashcan:%d", pCacheObj->name,
|
||||
pNode->key, pNode->data, pElem, pCacheObj->numOfElemsInTrash);
|
||||
uDebug("cache:%s key:%p, %p move to trashcan, pTrashElem:%p, numOfElem in trashcan:%d", pCacheObj->name, pNode->key,
|
||||
pNode->data, pElem, pCacheObj->numOfElemsInTrash);
|
||||
}
|
||||
|
||||
void taosTrashcanEmpty(SCacheObj *pCacheObj, bool force) {
|
||||
|
|
|
|||
|
|
@ -280,10 +280,13 @@ bool tSkipListIterNext(SSkipListIterator *iter) {
|
|||
tSkipListRLock(pSkipList);
|
||||
|
||||
if (iter->order == TSDB_ORDER_ASC) {
|
||||
if (iter->cur == pSkipList->pTail) {
|
||||
// no data in the skip list
|
||||
if (iter->cur == pSkipList->pTail || iter->next == NULL) {
|
||||
iter->cur = pSkipList->pTail;
|
||||
tSkipListUnlock(pSkipList);
|
||||
return false;
|
||||
}
|
||||
|
||||
iter->cur = SL_NODE_GET_FORWARD_POINTER(iter->cur, 0);
|
||||
|
||||
// a new node is inserted into between iter->cur and iter->next, ignore it
|
||||
|
|
@ -295,9 +298,11 @@ bool tSkipListIterNext(SSkipListIterator *iter) {
|
|||
iter->step++;
|
||||
} else {
|
||||
if (iter->cur == pSkipList->pHead) {
|
||||
iter->cur = pSkipList->pHead;
|
||||
tSkipListUnlock(pSkipList);
|
||||
return false;
|
||||
}
|
||||
|
||||
iter->cur = SL_NODE_GET_BACKWARD_POINTER(iter->cur, 0);
|
||||
|
||||
// a new node is inserted into between iter->cur and iter->next, ignore it
|
||||
|
|
|
|||
|
|
@ -18,6 +18,10 @@
|
|||
#include "tsocket.h"
|
||||
#include "taoserror.h"
|
||||
|
||||
#ifndef SIGPIPE
|
||||
#define SIGPIPE EPIPE
|
||||
#endif
|
||||
|
||||
int32_t taosGetFqdn(char *fqdn) {
|
||||
char hostname[1024];
|
||||
hostname[1023] = '\0';
|
||||
|
|
@ -115,6 +119,10 @@ int32_t taosWriteMsg(SOCKET fd, void *buf, int32_t nbytes) {
|
|||
nleft -= nwritten;
|
||||
ptr += nwritten;
|
||||
}
|
||||
|
||||
if (errno == SIGPIPE || errno == EPIPE) {
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
|
||||
return (nbytes - nleft);
|
||||
|
|
@ -142,6 +150,10 @@ int32_t taosReadMsg(SOCKET fd, void *buf, int32_t nbytes) {
|
|||
nleft -= nread;
|
||||
ptr += nread;
|
||||
}
|
||||
|
||||
if (errno == SIGPIPE || errno == EPIPE) {
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
|
||||
return (nbytes - nleft);
|
||||
|
|
|
|||
|
|
@ -106,9 +106,10 @@ int32_t vnodeReadCfg(SVnodeObj *pVnode) {
|
|||
cJSON *vgCfgVersion = cJSON_GetObjectItem(root, "vgCfgVersion");
|
||||
if (!vgCfgVersion || vgCfgVersion->type != cJSON_Number) {
|
||||
vError("vgId:%d, failed to read %s, vgCfgVersion not found", pVnode->vgId, file);
|
||||
goto PARSE_VCFG_ERROR;
|
||||
vnodeMsg.cfg.vgCfgVersion = 0;
|
||||
} else {
|
||||
vnodeMsg.cfg.vgCfgVersion = vgCfgVersion->valueint;
|
||||
}
|
||||
vnodeMsg.cfg.vgCfgVersion = vgCfgVersion->valueint;
|
||||
|
||||
cJSON *cacheBlockSize = cJSON_GetObjectItem(root, "cacheBlockSize");
|
||||
if (!cacheBlockSize || cacheBlockSize->type != cJSON_Number) {
|
||||
|
|
|
|||
|
|
@ -89,7 +89,10 @@ static void vnodeIncRef(void *ptNode) {
|
|||
}
|
||||
|
||||
void *vnodeAcquire(int32_t vgId) {
|
||||
SVnodeObj **ppVnode = taosHashGetCB(tsVnodesHash, &vgId, sizeof(int32_t), vnodeIncRef, NULL, sizeof(void *));
|
||||
SVnodeObj **ppVnode = NULL;
|
||||
if (tsVnodesHash != NULL) {
|
||||
ppVnode = taosHashGetCB(tsVnodesHash, &vgId, sizeof(int32_t), vnodeIncRef, NULL, sizeof(void *));
|
||||
}
|
||||
|
||||
if (ppVnode == NULL || *ppVnode == NULL) {
|
||||
terrno = TSDB_CODE_VND_INVALID_VGROUP_ID;
|
||||
|
|
|
|||
|
|
@ -65,13 +65,17 @@ static int32_t vnodeCheckRead(SVnodeObj *pVnode) {
|
|||
return TSDB_CODE_APP_NOT_READY;
|
||||
}
|
||||
|
||||
if (pVnode->role != TAOS_SYNC_ROLE_SLAVE && pVnode->role != TAOS_SYNC_ROLE_MASTER) {
|
||||
vDebug("vgId:%d, replica:%d role:%s, refCount:%d pVnode:%p", pVnode->vgId, pVnode->syncCfg.replica,
|
||||
syncRole[pVnode->role], pVnode->refCount, pVnode);
|
||||
return TSDB_CODE_APP_NOT_READY;
|
||||
if (pVnode->role == TAOS_SYNC_ROLE_MASTER) {
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
return TSDB_CODE_SUCCESS;
|
||||
if (tsEnableSlaveQuery && pVnode->role == TAOS_SYNC_ROLE_SLAVE) {
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
vDebug("vgId:%d, replica:%d role:%s, refCount:%d pVnode:%p, cant provide query service", pVnode->vgId, pVnode->syncCfg.replica,
|
||||
syncRole[pVnode->role], pVnode->refCount, pVnode);
|
||||
return TSDB_CODE_APP_NOT_READY;
|
||||
}
|
||||
|
||||
void vnodeFreeFromRQueue(void *vparam, SVReadMsg *pRead) {
|
||||
|
|
@ -303,8 +307,11 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pRead) {
|
|||
// NOTE: set return code to be TSDB_CODE_QRY_HAS_RSP to notify dnode to return msg to client
|
||||
code = TSDB_CODE_QRY_HAS_RSP;
|
||||
} else {
|
||||
void *h1 = qGetResultRetrieveMsg(*qhandle);
|
||||
assert(h1 == NULL);
|
||||
//void *h1 = qGetResultRetrieveMsg(*qhandle);
|
||||
|
||||
/* remove this assert, one possible case that will cause h1 not NULL: query thread unlock pQInfo->lock, and then FETCH thread execute twice before query thread reach here */
|
||||
//assert(h1 == NULL);
|
||||
|
||||
freehandle = qQueryCompleted(*qhandle);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -297,7 +297,7 @@ static int32_t vnodePerformFlowCtrl(SVWriteMsg *pWrite) {
|
|||
if (pWrite->qtype != TAOS_QTYPE_RPC) return 0;
|
||||
if (pVnode->queuedWMsg < MAX_QUEUED_MSG_NUM && pVnode->flowctrlLevel <= 0) return 0;
|
||||
|
||||
if (tsFlowCtrl == 0) {
|
||||
if (tsEnableFlowCtrl == 0) {
|
||||
int32_t ms = pow(2, pVnode->flowctrlLevel + 2);
|
||||
if (ms > 100) ms = 100;
|
||||
vTrace("vgId:%d, msg:%p, app:%p, perform flowctrl for %d ms", pVnode->vgId, pWrite, pWrite->rpcMsg.ahandle, ms);
|
||||
|
|
|
|||
|
|
@ -50,12 +50,7 @@ pipeline {
|
|||
agent{label 'master'}
|
||||
steps {
|
||||
pre_test()
|
||||
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
|
||||
sh '''
|
||||
cd ${WKC}/tests/pytest
|
||||
python3 concurrent_inquiry.py -c 1
|
||||
'''
|
||||
}
|
||||
|
||||
sh '''
|
||||
cd ${WKC}/tests
|
||||
./test-all.sh b1
|
||||
|
|
@ -82,7 +77,37 @@ pipeline {
|
|||
./handle_crash_gen_val_log.sh
|
||||
'''
|
||||
}
|
||||
sh'''
|
||||
systemctl start taosd
|
||||
sleep 10
|
||||
'''
|
||||
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
|
||||
sh '''
|
||||
cd ${WKC}/tests/gotest
|
||||
bash batchtest.sh
|
||||
'''
|
||||
}
|
||||
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
|
||||
sh '''
|
||||
cd ${WKC}/tests/examples/python/PYTHONConnectorChecker
|
||||
python3 PythonChecker.py
|
||||
'''
|
||||
}
|
||||
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
|
||||
sh '''
|
||||
cd ${WKC}/tests/examples/JDBC/JDBCDemo/
|
||||
mvn clean package assembly:single -DskipTests >/dev/null
|
||||
java -jar target/jdbcChecker-SNAPSHOT-jar-with-dependencies.jar -host 127.0.0.1
|
||||
'''
|
||||
}
|
||||
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
|
||||
sh '''
|
||||
cd ${JENKINS_HOME}/workspace/C#NET/src/CheckC#
|
||||
dotnet run
|
||||
'''
|
||||
}
|
||||
sh '''
|
||||
systemctl stop taosd
|
||||
cd ${WKC}/tests
|
||||
./test-all.sh b2
|
||||
date
|
||||
|
|
@ -95,6 +120,15 @@ pipeline {
|
|||
|
||||
steps {
|
||||
pre_test()
|
||||
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
|
||||
sh '''
|
||||
cd ${WKC}/tests/pytest
|
||||
nohup taosd >/dev/null &
|
||||
sleep 10
|
||||
python3 concurrent_inquiry.py -c 1
|
||||
|
||||
'''
|
||||
}
|
||||
sh '''
|
||||
cd ${WKC}/tests/pytest
|
||||
./valgrind-test.sh 2>&1 > mem-error-out.log
|
||||
|
|
@ -106,41 +140,7 @@ pipeline {
|
|||
date'''
|
||||
}
|
||||
}
|
||||
stage('connector'){
|
||||
agent{label "release"}
|
||||
steps{
|
||||
sh'''
|
||||
cd ${WORKSPACE}
|
||||
git checkout develop
|
||||
'''
|
||||
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
|
||||
sh '''
|
||||
cd ${WORKSPACE}/tests/gotest
|
||||
bash batchtest.sh
|
||||
'''
|
||||
}
|
||||
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
|
||||
sh '''
|
||||
cd ${WORKSPACE}/tests/examples/python/PYTHONConnectorChecker
|
||||
python3 PythonChecker.py
|
||||
'''
|
||||
}
|
||||
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
|
||||
sh '''
|
||||
cd ${WORKSPACE}/tests/examples/JDBC/JDBCDemo/
|
||||
mvn clean package assembly:single >/dev/null
|
||||
java -jar target/jdbcChecker-SNAPSHOT-jar-with-dependencies.jar -host 127.0.0.1
|
||||
'''
|
||||
}
|
||||
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
|
||||
sh '''
|
||||
cd ${JENKINS_HOME}/workspace/C#NET/src/CheckC#
|
||||
dotnet run
|
||||
'''
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
stage('arm64_build'){
|
||||
agent{label 'arm64'}
|
||||
steps{
|
||||
|
|
|
|||
|
|
@ -388,7 +388,9 @@ class ConcurrentInquiry:
|
|||
print(
|
||||
"Failure thread%d, sql: %s \nexception: %s" %
|
||||
(threadID, str(sql),str(e)))
|
||||
#exit(-1)
|
||||
err_uec='Unable to establish connection'
|
||||
if err_uec in str(e) and loop >0:
|
||||
exit(-1)
|
||||
loop -= 1
|
||||
if loop == 0: break
|
||||
|
||||
|
|
@ -415,7 +417,9 @@ class ConcurrentInquiry:
|
|||
print(
|
||||
"Failure thread%d, sql: %s \nexception: %s" %
|
||||
(threadID, str(sql),str(e)))
|
||||
#exit(-1)
|
||||
err_uec='Unable to establish connection'
|
||||
if err_uec in str(e) and loop >0:
|
||||
exit(-1)
|
||||
loop -= 1
|
||||
if loop == 0: break
|
||||
|
||||
|
|
|
|||
|
|
@ -2224,22 +2224,25 @@ class ClientManager:
|
|||
if svcMgr: # gConfig.auto_start_service:
|
||||
svcMgr.stopTaosServices()
|
||||
svcMgr = None
|
||||
|
||||
|
||||
# Release global variables
|
||||
gConfig = None
|
||||
gSvcMgr = None
|
||||
logger = None
|
||||
|
||||
thPool = None
|
||||
dbManager.cleanUp() # destructor wouldn't run in time
|
||||
dbManager = None
|
||||
|
||||
# Print exec status, etc., AFTER showing messages from the server
|
||||
self.conclude()
|
||||
# print("TC failed (2) = {}".format(self.tc.isFailed()))
|
||||
# Linux return code: ref https://shapeshed.com/unix-exit-codes/
|
||||
ret = 1 if self.tc.isFailed() else 0
|
||||
self.tc.cleanup()
|
||||
|
||||
# Release global variables
|
||||
gConfig = None
|
||||
gSvcMgr = None
|
||||
logger = None
|
||||
|
||||
# Release variables here
|
||||
self.tc = None
|
||||
thPool = None
|
||||
dbManager = None
|
||||
|
||||
gc.collect() # force garbage collection
|
||||
# h = hpy()
|
||||
|
|
|
|||
|
|
@ -394,6 +394,7 @@ class DbManager():
|
|||
cType == 'native') else DbConn.createRest(dbTarget)
|
||||
try:
|
||||
self._dbConn.open() # may throw taos.error.ProgrammingError: disconnected
|
||||
Logging.debug("DbManager opened DB connection...")
|
||||
except taos.error.ProgrammingError as err:
|
||||
# print("Error type: {}, msg: {}, value: {}".format(type(err), err.msg, err))
|
||||
if (err.msg == 'client disconnected'): # cannot open DB connection
|
||||
|
|
@ -412,6 +413,10 @@ class DbManager():
|
|||
# Moved to Database()
|
||||
# self._stateMachine = StateMechine(self._dbConn)
|
||||
|
||||
def __del__(self):
|
||||
''' Release the underlying DB connection upon deletion of DbManager '''
|
||||
self.cleanUp()
|
||||
|
||||
def getDbConn(self):
|
||||
return self._dbConn
|
||||
|
||||
|
|
@ -438,5 +443,8 @@ class DbManager():
|
|||
return "table_{}".format(tblNum)
|
||||
|
||||
def cleanUp(self):
|
||||
self._dbConn.close()
|
||||
if self._dbConn:
|
||||
self._dbConn.close()
|
||||
self._dbConn = None
|
||||
Logging.debug("DbManager closed DB connection...")
|
||||
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ import threading
|
|||
import random
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
|
||||
import taos
|
||||
|
||||
|
|
@ -53,7 +54,7 @@ class Logging:
|
|||
# global misc.logger
|
||||
_logger = logging.getLogger('CrashGen') # real logger
|
||||
_logger.addFilter(LoggingFilter())
|
||||
ch = logging.StreamHandler()
|
||||
ch = logging.StreamHandler(sys.stdout) # Ref: https://stackoverflow.com/questions/14058453/making-python-loggers-output-all-messages-to-stdout-in-addition-to-log-file
|
||||
_logger.addHandler(ch)
|
||||
|
||||
# Logging adapter, to be used as a logger
|
||||
|
|
|
|||
|
|
@ -19,5 +19,5 @@ if __name__ == "__main__":
|
|||
mExec.init()
|
||||
exitCode = mExec.run()
|
||||
|
||||
print("Exiting with code: {}".format(exitCode))
|
||||
print("\nCrash_Gen is now exiting with status code: {}".format(exitCode))
|
||||
sys.exit(exitCode)
|
||||
|
|
|
|||
|
|
@ -5,9 +5,10 @@ GREEN='\033[1;32m'
|
|||
GREEN_DARK='\033[0;32m'
|
||||
GREEN_UNDERLINE='\033[4;32m'
|
||||
NC='\033[0m'
|
||||
nohup /var/lib/jenkins/workspace/TDinternal/debug/build/bin/taosd -c /var/lib/jenkins/workspace/TDinternal/community/sim/dnode1/cfg >/dev/null &
|
||||
#nohup /var/lib/jenkins/workspace/TDinternal/debug/build/bin/taosd -c /var/lib/jenkins/workspace/TDinternal/community/sim/dnode1/cfg >/dev/null &
|
||||
nohup /root/TDinternal/debug/build/bin/taosd -c /root/TDinternal/community/sim/dnode1/cfg >/dev/null &
|
||||
./crash_gen.sh --valgrind -p -t 10 -s 250 -b 4
|
||||
pidof taosd|xargs kill
|
||||
pidof taosd|xargs kill -9
|
||||
grep 'start to execute\|ERROR SUMMARY' valgrind.err|grep -v 'grep'|uniq|tee crash_gen_mem_err.log
|
||||
|
||||
for memError in `grep 'ERROR SUMMARY' crash_gen_mem_err.log | awk '{print $4}'`
|
||||
|
|
@ -31,4 +32,4 @@ if [ -n "$defiMemError" ]; then
|
|||
exit 8
|
||||
fi
|
||||
fi
|
||||
done
|
||||
done
|
||||
|
|
|
|||
|
|
@ -55,7 +55,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -55,7 +55,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -57,7 +57,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -55,7 +55,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -55,7 +55,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -55,7 +55,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -55,7 +55,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -57,7 +57,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -59,7 +59,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -60,7 +60,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -60,7 +60,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -62,7 +62,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -59,7 +59,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -59,7 +59,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -61,7 +61,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -59,7 +59,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -32,7 +32,7 @@ class TDTestCase:
|
|||
tdDnodes.stop(1)
|
||||
tdDnodes.deploy(1)
|
||||
tdDnodes.start(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
|
||||
tdSql.execute('reset query cache')
|
||||
tdSql.execute('drop database if exists db')
|
||||
|
|
@ -60,9 +60,9 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
tdLog.info("import 10 data totally repetitive")
|
||||
|
|
|
|||
|
|
@ -55,7 +55,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -55,7 +55,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -57,7 +57,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -55,7 +55,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -60,7 +60,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -55,7 +55,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -55,7 +55,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -57,9 +57,9 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
tdLog.info("import 20 data later with partly overlap")
|
||||
|
|
|
|||
|
|
@ -57,7 +57,7 @@ class TDTestCase:
|
|||
tdLog.info("================= step5")
|
||||
tdDnodes.forcestop(1)
|
||||
tdDnodes.start(1)
|
||||
tdLog.sleep(10)
|
||||
#tdLog.sleep(10)
|
||||
|
||||
tdLog.info("================= step6")
|
||||
tdSql.query('select * from tb1')
|
||||
|
|
|
|||
|
|
@ -62,7 +62,7 @@ class TDTestCase:
|
|||
tdLog.info("================= step5")
|
||||
tdDnodes.forcestop(1)
|
||||
tdDnodes.start(1)
|
||||
tdLog.sleep(10)
|
||||
#tdLog.sleep(10)
|
||||
|
||||
tdLog.info("================= step6")
|
||||
tdSql.query('select * from tb1')
|
||||
|
|
|
|||
|
|
@ -54,7 +54,7 @@ class TDTestCase:
|
|||
tdLog.info("================= step5")
|
||||
tdDnodes.forcestop(1)
|
||||
tdDnodes.start(1)
|
||||
tdLog.sleep(10)
|
||||
#tdLog.sleep(10)
|
||||
|
||||
tdLog.info("================= step6")
|
||||
tdSql.query('select * from tb1')
|
||||
|
|
|
|||
|
|
@ -61,7 +61,7 @@ class TDTestCase:
|
|||
tdLog.info("================= step5")
|
||||
tdDnodes.stop(1)
|
||||
tdDnodes.start(1)
|
||||
tdLog.sleep(10)
|
||||
#tdLog.sleep(10)
|
||||
|
||||
tdLog.info("================= step6")
|
||||
tdLog.info("import 100 sequential data again")
|
||||
|
|
|
|||
|
|
@ -53,7 +53,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -53,7 +53,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -55,7 +55,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -53,7 +53,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -53,7 +53,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -57,7 +57,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -57,7 +57,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -59,7 +59,7 @@ class TDTestCase:
|
|||
|
||||
tdLog.info("================= step4")
|
||||
tdDnodes.stop(1)
|
||||
tdLog.sleep(5)
|
||||
#tdLog.sleep(5)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdLog.info("================= step5")
|
||||
|
|
|
|||
|
|
@ -64,7 +64,7 @@ class TDTestCase:
|
|||
tdLog.info("================= step5")
|
||||
tdDnodes.forcestop(1)
|
||||
tdDnodes.start(1)
|
||||
tdLog.sleep(10)
|
||||
#tdLog.sleep(10)
|
||||
|
||||
tdLog.info("================= step6")
|
||||
tdSql.query('select * from tb1')
|
||||
|
|
|
|||
|
|
@ -64,7 +64,7 @@ class TDTestCase:
|
|||
tdLog.info("================= step5")
|
||||
tdDnodes.forcestop(1)
|
||||
tdDnodes.start(1)
|
||||
tdLog.sleep(10)
|
||||
#tdLog.sleep(10)
|
||||
|
||||
tdLog.info("================= step6")
|
||||
tdSql.query('select * from tb1')
|
||||
|
|
|
|||
|
|
@ -64,7 +64,7 @@ class TDTestCase:
|
|||
tdLog.info("================= step5")
|
||||
tdDnodes.forcestop(1)
|
||||
tdDnodes.start(1)
|
||||
tdLog.sleep(10)
|
||||
#tdLog.sleep(10)
|
||||
|
||||
tdLog.info("================= step6")
|
||||
tdSql.query('select * from tb1')
|
||||
|
|
|
|||
|
|
@ -68,7 +68,7 @@ class TDTestCase:
|
|||
tdLog.info("================= step5")
|
||||
tdDnodes.forcestop(1)
|
||||
tdDnodes.start(1)
|
||||
tdLog.sleep(10)
|
||||
#tdLog.sleep(10)
|
||||
|
||||
tdLog.info("================= step6")
|
||||
tdSql.query('select * from tb1')
|
||||
|
|
|
|||
|
|
@ -61,7 +61,7 @@ class TDTestCase:
|
|||
tdLog.info("================= step5")
|
||||
tdDnodes.forcestop(1)
|
||||
tdDnodes.start(1)
|
||||
tdLog.sleep(10)
|
||||
#tdLog.sleep(10)
|
||||
|
||||
tdLog.info("================= step6")
|
||||
tdSql.query('select * from tb1')
|
||||
|
|
|
|||
|
|
@ -16,7 +16,7 @@ import taos
|
|||
from util.log import tdLog
|
||||
from util.cases import tdCases
|
||||
from util.sql import tdSql
|
||||
|
||||
from util.dnodes import tdDnodes
|
||||
|
||||
class TDTestCase:
|
||||
def init(self, conn, logSql):
|
||||
|
|
@ -44,6 +44,25 @@ class TDTestCase:
|
|||
tdSql.query("select * from db.st where ts='2020-05-13 10:00:00.000'")
|
||||
tdSql.checkRows(1)
|
||||
|
||||
## test case for https://jira.taosdata.com:18080/browse/TD-2488
|
||||
tdSql.execute("create table m1(ts timestamp, k int) tags(a int)")
|
||||
tdSql.execute("create table t1 using m1 tags(1)")
|
||||
tdSql.execute("create table t2 using m1 tags(2)")
|
||||
tdSql.execute("insert into t1 values('2020-1-1 1:1:1', 1)")
|
||||
tdSql.execute("insert into t1 values('2020-1-1 1:10:1', 2)")
|
||||
tdSql.execute("insert into t2 values('2020-1-1 1:5:1', 99)")
|
||||
|
||||
tdSql.query("select count(*) from m1 where ts = '2020-1-1 1:5:1' ")
|
||||
tdSql.checkRows(1)
|
||||
tdSql.checkData(0, 0, 1)
|
||||
|
||||
tdDnodes.stop(1)
|
||||
tdDnodes.start(1)
|
||||
|
||||
tdSql.query("select count(*) from m1 where ts = '2020-1-1 1:5:1' ")
|
||||
tdSql.checkRows(1)
|
||||
tdSql.checkData(0, 0, 1)
|
||||
|
||||
def stop(self):
|
||||
tdSql.close()
|
||||
tdLog.success("%s successfully executed" % __file__)
|
||||
|
|
|
|||
Some files were not shown because too many files have changed in this diff Show More
Loading…
Reference in New Issue