提交 ea1ee014 编写于 作者: H Haojun Liao

Merge branch 'develop' into feature/query

......@@ -14,10 +14,12 @@ pipeline {
sh '''
date
cd ${WKC}
git reset --hard
git checkout develop
git pull
git submodule update
cd ${WK}
git reset --hard
git checkout develop
git pull
export TZ=Asia/Harbin
......@@ -39,11 +41,13 @@ pipeline {
steps {
sh '''
cd ${WKC}
git reset --hard
git checkout develop
git pull
git submodule update
cd ${WK}
git reset --hard
git checkout develop
git pull
export TZ=Asia/Harbin
......@@ -65,11 +69,13 @@ pipeline {
steps {
sh '''
cd ${WKC}
git reset --hard
git checkout develop
git pull
git submodule update
cd ${WK}
git reset --hard
git checkout develop
git pull
export TZ=Asia/Harbin
......@@ -108,11 +114,13 @@ pipeline {
steps {
sh '''
cd ${WKC}
git reset --hard
git checkout develop
git pull
git submodule update
cd ${WK}
git reset --hard
git checkout develop
git pull
export TZ=Asia/Harbin
......@@ -167,7 +175,47 @@ pipeline {
}
}
stage('arm64_build'){
agent{label 'arm64'}
steps{
sh '''
cd ${WK}
git fetch
git checkout develop
git pull
cd ${WKC}
git fetch
git checkout develop
git pull
git submodule update
cd ${WKC}/packaging
./release.sh -v cluster -c aarch64 -n 2.0.0.0 -m 2.0.0.0
'''
}
}
stage('arm32_build'){
agent{label 'arm32'}
steps{
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
sh '''
cd ${WK}
git fetch
git checkout develop
git pull
cd ${WKC}
git fetch
git checkout develop
git pull
git submodule update
cd ${WKC}/packaging
./release.sh -v cluster -c aarch32 -n 2.0.0.0 -m 2.0.0.0
'''
}
}
}
}
}
......
......@@ -95,6 +95,7 @@ TDengine系统后台服务由taosd提供,可以在配置文件taos.cfg里修
- logKeepDays:日志文件的最长保存时间。大于0时,日志文件会被重命名为taosdlog.xxx,其中xxx为日志文件最后修改的时间戳,单位为秒。默认值:0天。
- maxSQLLength:单条SQL语句允许最长限制。默认值:65380字节。
- telemetryReporting: 是否允许 TDengine 采集和上报基本使用信息,0表示不允许,1表示允许。 默认值:1。
- stream: 是否启用连续查询(流计算功能),0表示不允许,1表示允许。 默认值:1。
**注意:**对于端口,TDengine会使用从serverPort起13个连续的TCP和UDP端口号,请务必在防火墙打开。因此如果是缺省配置,需要打开从6030都6042共13个端口,而且必须TCP和UDP都打开。
......
......@@ -142,7 +142,7 @@ C/C++的API类似于MySQL的C API。应用程序使用时,需要包含TDengine
获取最近一次API调用失败的原因,返回值为错误代码。
**注意**:对于单个数据库连接,在同一时刻只能有一个线程使用该连接调用API,否则会有未定义的行为出现并可能导致客户端crash。客户端应用可以通过建立多个连接进行多线程的数据写入或查询处理
**注意**:对于每个数据库应用,2.0及以上版本 TDengine 推荐只建立一个连接。同时在应用中将该连接 (TAOS*) 结构体传递到不同的线程共享使用。基于 TAOS 结构体发出的查询、写入等操作具有多线程安全性。C 语言的连接器可以按照需求动态建立面向数据库的新连接(该过程对用户不可见),同时建议只有在程序最后退出的时候才调用 taos_close 关闭连接
### 异步查询API
......
......@@ -260,4 +260,7 @@
# maxBinaryDisplayWidth 30
# enable/disable telemetry reporting
# telemetryReporting 1
\ No newline at end of file
# telemetryReporting 1
# enable/disable stream (continuous query)
# stream 1
......@@ -172,6 +172,7 @@ function install_bin() {
${csudo} rm -f ${bin_link_dir}/taos || :
${csudo} rm -f ${bin_link_dir}/taosd || :
${csudo} rm -f ${bin_link_dir}/taosdemo || :
${csudo} rm -f ${bin_link_dir}/taosdump || :
${csudo} rm -f ${bin_link_dir}/rmtaos || :
${csudo} rm -f ${bin_link_dir}/tarbitrator || :
${csudo} rm -f ${bin_link_dir}/set_core || :
......@@ -182,6 +183,7 @@ function install_bin() {
[ -x ${install_main_dir}/bin/taos ] && ${csudo} ln -s ${install_main_dir}/bin/taos ${bin_link_dir}/taos || :
[ -x ${install_main_dir}/bin/taosd ] && ${csudo} ln -s ${install_main_dir}/bin/taosd ${bin_link_dir}/taosd || :
[ -x ${install_main_dir}/bin/taosdemo ] && ${csudo} ln -s ${install_main_dir}/bin/taosdemo ${bin_link_dir}/taosdemo || :
[ -x ${install_main_dir}/bin/taosdump ] && ${csudo} ln -s ${install_main_dir}/bin/taosdump ${bin_link_dir}/taosdump || :
[ -x ${install_main_dir}/bin/remove.sh ] && ${csudo} ln -s ${install_main_dir}/bin/remove.sh ${bin_link_dir}/rmtaos || :
[ -x ${install_main_dir}/bin/set_core.sh ] && ${csudo} ln -s ${install_main_dir}/bin/set_core.sh ${bin_link_dir}/set_core || :
[ -x ${install_main_dir}/bin/tarbitrator ] && ${csudo} ln -s ${install_main_dir}/bin/tarbitrator ${bin_link_dir}/tarbitrator || :
......
......@@ -84,8 +84,9 @@ function install_main_path() {
function install_bin() {
# Remove links
${csudo} rm -f ${bin_link_dir}/taos || :
if [ "$osType" == "Darwin" ]; then
if [ "$osType" != "Darwin" ]; then
${csudo} rm -f ${bin_link_dir}/taosdemo || :
${csudo} rm -f ${bin_link_dir}/taosdump || :
fi
${csudo} rm -f ${bin_link_dir}/rmtaos || :
${csudo} rm -f ${bin_link_dir}/set_core || :
......@@ -94,8 +95,9 @@ function install_bin() {
#Make link
[ -x ${install_main_dir}/bin/taos ] && ${csudo} ln -s ${install_main_dir}/bin/taos ${bin_link_dir}/taos || :
if [ "$osType" == "Darwin" ]; then
if [ "$osType" != "Darwin" ]; then
[ -x ${install_main_dir}/bin/taosdemo ] && ${csudo} ln -s ${install_main_dir}/bin/taosdemo ${bin_link_dir}/taosdemo || :
[ -x ${install_main_dir}/bin/taosdump ] && ${csudo} ln -s ${install_main_dir}/bin/taosdump ${bin_link_dir}/taosdump || :
fi
[ -x ${install_main_dir}/bin/remove_client.sh ] && ${csudo} ln -s ${install_main_dir}/bin/remove_client.sh ${bin_link_dir}/rmtaos || :
[ -x ${install_main_dir}/bin/set_core.sh ] && ${csudo} ln -s ${install_main_dir}/bin/set_core.sh ${bin_link_dir}/set_core || :
......
......@@ -84,8 +84,9 @@ function install_main_path() {
function install_bin() {
# Remove links
${csudo} rm -f ${bin_link_dir}/power || :
if [ "$osType" == "Darwin" ]; then
if [ "$osType" != "Darwin" ]; then
${csudo} rm -f ${bin_link_dir}/powerdemo || :
${csudo} rm -f ${bin_link_dir}/powerdump || :
fi
${csudo} rm -f ${bin_link_dir}/rmpower || :
${csudo} rm -f ${bin_link_dir}/set_core || :
......@@ -94,8 +95,9 @@ function install_bin() {
#Make link
[ -x ${install_main_dir}/bin/power ] && ${csudo} ln -s ${install_main_dir}/bin/power ${bin_link_dir}/power || :
if [ "$osType" == "Darwin" ]; then
if [ "$osType" != "Darwin" ]; then
[ -x ${install_main_dir}/bin/powerdemo ] && ${csudo} ln -s ${install_main_dir}/bin/powerdemo ${bin_link_dir}/powerdemo || :
[ -x ${install_main_dir}/bin/powerdump ] && ${csudo} ln -s ${install_main_dir}/bin/powerdump ${bin_link_dir}/powerdump || :
fi
[ -x ${install_main_dir}/bin/remove_client_power.sh ] && ${csudo} ln -s ${install_main_dir}/bin/remove_client_power.sh ${bin_link_dir}/rmpower || :
[ -x ${install_main_dir}/bin/set_core.sh ] && ${csudo} ln -s ${install_main_dir}/bin/set_core.sh ${bin_link_dir}/set_core || :
......
......@@ -172,6 +172,7 @@ function install_bin() {
${csudo} rm -f ${bin_link_dir}/power || :
${csudo} rm -f ${bin_link_dir}/powerd || :
${csudo} rm -f ${bin_link_dir}/powerdemo || :
${csudo} rm -f ${bin_link_dir}/powerdump || :
${csudo} rm -f ${bin_link_dir}/rmpower || :
${csudo} rm -f ${bin_link_dir}/tarbitrator || :
${csudo} rm -f ${bin_link_dir}/set_core || :
......@@ -182,6 +183,7 @@ function install_bin() {
[ -x ${install_main_dir}/bin/power ] && ${csudo} ln -s ${install_main_dir}/bin/power ${bin_link_dir}/power || :
[ -x ${install_main_dir}/bin/powerd ] && ${csudo} ln -s ${install_main_dir}/bin/powerd ${bin_link_dir}/powerd || :
[ -x ${install_main_dir}/bin/powerdemo ] && ${csudo} ln -s ${install_main_dir}/bin/powerdemo ${bin_link_dir}/powerdemo || :
[ -x ${install_main_dir}/bin/powerdump ] && ${csudo} ln -s ${install_main_dir}/bin/powerdump ${bin_link_dir}/powerdump || :
[ -x ${install_main_dir}/bin/remove_power.sh ] && ${csudo} ln -s ${install_main_dir}/bin/remove_power.sh ${bin_link_dir}/rmpower || :
[ -x ${install_main_dir}/bin/set_core.sh ] && ${csudo} ln -s ${install_main_dir}/bin/set_core.sh ${bin_link_dir}/set_core || :
[ -x ${install_main_dir}/bin/tarbitrator ] && ${csudo} ln -s ${install_main_dir}/bin/tarbitrator ${bin_link_dir}/tarbitrator || :
......
......@@ -92,6 +92,7 @@ function install_bin() {
${csudo} rm -f ${bin_link_dir}/taos || :
${csudo} rm -f ${bin_link_dir}/taosd || :
${csudo} rm -f ${bin_link_dir}/taosdemo || :
${csudo} rm -f ${bin_link_dir}/taosdump || :
${csudo} rm -f ${bin_link_dir}/rmtaos || :
${csudo} rm -f ${bin_link_dir}/set_core || :
......@@ -101,6 +102,7 @@ function install_bin() {
[ -x ${bin_dir}/taos ] && ${csudo} ln -s ${bin_dir}/taos ${bin_link_dir}/taos || :
[ -x ${bin_dir}/taosd ] && ${csudo} ln -s ${bin_dir}/taosd ${bin_link_dir}/taosd || :
[ -x ${bin_dir}/taosdemo ] && ${csudo} ln -s ${bin_dir}/taosdemo ${bin_link_dir}/taosdemo || :
[ -x ${bin_dir}/taosdump ] && ${csudo} ln -s ${bin_dir}/taosdump ${bin_link_dir}/taosdump || :
[ -x ${bin_dir}/set_core.sh ] && ${csudo} ln -s ${bin_dir}/set_core.sh ${bin_link_dir}/set_core || :
}
......
......@@ -72,6 +72,7 @@ function clean_bin() {
${csudo} rm -f ${bin_link_dir}/taos || :
${csudo} rm -f ${bin_link_dir}/taosd || :
${csudo} rm -f ${bin_link_dir}/taosdemo || :
${csudo} rm -f ${bin_link_dir}/taosdump || :
${csudo} rm -f ${bin_link_dir}/rmtaos || :
${csudo} rm -f ${bin_link_dir}/tarbitrator || :
${csudo} rm -f ${bin_link_dir}/set_core || :
......@@ -222,4 +223,4 @@ elif echo $osinfo | grep -qwi "centos" ; then
fi
echo -e "${GREEN}TDengine is removed successfully!${NC}"
echo
\ No newline at end of file
echo
......@@ -38,6 +38,7 @@ function clean_bin() {
# Remove link
${csudo} rm -f ${bin_link_dir}/taos || :
${csudo} rm -f ${bin_link_dir}/taosdemo || :
${csudo} rm -f ${bin_link_dir}/taosdump || :
${csudo} rm -f ${bin_link_dir}/rmtaos || :
${csudo} rm -f ${bin_link_dir}/set_core || :
}
......
......@@ -38,6 +38,7 @@ function clean_bin() {
# Remove link
${csudo} rm -f ${bin_link_dir}/power || :
${csudo} rm -f ${bin_link_dir}/powerdemo || :
${csudo} rm -f ${bin_link_dir}/powerdump || :
${csudo} rm -f ${bin_link_dir}/rmpower || :
${csudo} rm -f ${bin_link_dir}/set_core || :
}
......
......@@ -72,6 +72,7 @@ function clean_bin() {
${csudo} rm -f ${bin_link_dir}/power || :
${csudo} rm -f ${bin_link_dir}/powerd || :
${csudo} rm -f ${bin_link_dir}/powerdemo || :
${csudo} rm -f ${bin_link_dir}/powerdump || :
${csudo} rm -f ${bin_link_dir}/rmpower || :
${csudo} rm -f ${bin_link_dir}/tarbitrator || :
${csudo} rm -f ${bin_link_dir}/set_core || :
......@@ -223,4 +224,4 @@ fi
#fi
echo -e "${GREEN}PowerDB is removed successfully!${NC}"
echo
\ No newline at end of file
echo
......@@ -336,7 +336,7 @@ typedef struct STscObj {
char superAuth : 1;
uint32_t connId;
uint64_t rid; // ref ID returned by taosAddRef
struct SSqlObj * pHb;
int64_t hbrid;
struct SSqlObj * sqlList;
struct SSqlStream *streamList;
void* pDnodeConn;
......@@ -376,7 +376,7 @@ typedef struct SSqlObj {
struct SSqlObj **pSubs;
struct SSqlObj *prev, *next;
struct SSqlObj **self;
int64_t self;
} SSqlObj;
typedef struct SSqlStream {
......@@ -510,7 +510,7 @@ static FORCE_INLINE void tscGetResultColumnChr(SSqlRes* pRes, SFieldInfo* pField
}
extern SCacheObj* tscMetaCache;
extern SCacheObj* tscObjCache;
extern int tscObjRef;
extern void * tscTmr;
extern void * tscQhandle;
extern int tscKeepConn[];
......
......@@ -825,8 +825,11 @@ static int32_t tscProcessClientVer(SSqlObj *pSql) {
static int32_t tscProcessServStatus(SSqlObj *pSql) {
STscObj* pObj = pSql->pTscObj;
if (pObj->pHb != NULL) {
if (pObj->pHb->res.code == TSDB_CODE_RPC_NETWORK_UNAVAIL) {
SSqlObj* pHb = (SSqlObj*)taosAcquireRef(tscObjRef, pObj->hbrid);
if (pHb != NULL) {
int32_t code = pHb->res.code;
taosReleaseRef(tscObjRef, pObj->hbrid);
if (code == TSDB_CODE_RPC_NETWORK_UNAVAIL) {
pSql->res.code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
return pSql->res.code;
}
......
......@@ -175,10 +175,10 @@ void tscProcessHeartBeatRsp(void *param, TAOS_RES *tres, int code) {
if (pRsp->streamId) tscKillStream(pObj, htonl(pRsp->streamId));
}
} else {
tscDebug("%p heartbeat failed, code:%s", pObj->pHb, tstrerror(code));
tscDebug("%" PRId64 " heartbeat failed, code:%s", pObj->hbrid, tstrerror(code));
}
if (pObj->pHb != NULL) {
if (pObj->hbrid != 0) {
int32_t waitingDuring = tsShellActivityTimer * 500;
tscDebug("%p send heartbeat in %dms", pSql, waitingDuring);
......@@ -193,20 +193,12 @@ void tscProcessActivityTimer(void *handle, void *tmrId) {
STscObj *pObj = taosAcquireRef(tscRefId, rid);
if (pObj == NULL) return;
SSqlObj* pHB = pObj->pHb;
void** p = taosCacheAcquireByKey(tscObjCache, &pHB, sizeof(TSDB_CACHE_PTR_TYPE));
if (p == NULL) {
tscWarn("%p HB object has been released already", pHB);
taosReleaseRef(tscRefId, pObj->rid);
return;
}
assert(*pHB->self == pHB);
SSqlObj* pHB = taosAcquireRef(tscObjRef, pObj->hbrid);
assert(pHB->self == pObj->hbrid);
pHB->retry = 0;
int32_t code = tscProcessSql(pHB);
taosCacheRelease(tscObjCache, (void**) &p, false);
taosReleaseRef(tscObjRef, pObj->hbrid);
if (code != TSDB_CODE_SUCCESS) {
tscError("%p failed to sent HB to server, reason:%s", pHB, tstrerror(code));
......@@ -236,7 +228,7 @@ int tscSendMsgToServer(SSqlObj *pSql) {
.msgType = pSql->cmd.msgType,
.pCont = pMsg,
.contLen = pSql->cmd.payloadLen,
.ahandle = pSql,
.ahandle = (void*)pSql->self,
.handle = NULL,
.code = 0
};
......@@ -247,26 +239,24 @@ int tscSendMsgToServer(SSqlObj *pSql) {
void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
TSDB_CACHE_PTR_TYPE handle = (TSDB_CACHE_PTR_TYPE) rpcMsg->ahandle;
void** p = taosCacheAcquireByKey(tscObjCache, &handle, sizeof(TSDB_CACHE_PTR_TYPE));
if (p == NULL) {
SSqlObj* pSql = (SSqlObj*)taosAcquireRef(tscObjRef, handle);
if (pSql == NULL) {
rpcFreeCont(rpcMsg->pCont);
return;
}
SSqlObj* pSql = *p;
assert(pSql != NULL);
assert(pSql->self == handle);
STscObj *pObj = pSql->pTscObj;
SSqlRes *pRes = &pSql->res;
SSqlCmd *pCmd = &pSql->cmd;
assert(*pSql->self == pSql);
pSql->rpcRid = -1;
if (pObj->signature != pObj) {
tscDebug("%p DB connection is closed, cmd:%d pObj:%p signature:%p", pSql, pCmd->command, pObj, pObj->signature);
taosCacheRelease(tscObjCache, (void**) &p, true);
taosRemoveRef(tscObjRef, pSql->self);
taosReleaseRef(tscObjRef, pSql->self);
rpcFreeCont(rpcMsg->pCont);
return;
}
......@@ -276,10 +266,8 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
tscDebug("%p sqlObj needs to be released or DB connection is closed, cmd:%d type:%d, pObj:%p signature:%p",
pSql, pCmd->command, pQueryInfo->type, pObj, pObj->signature);
void** p1 = p;
taosCacheRelease(tscObjCache, (void**) &p1, false);
taosCacheRelease(tscObjCache, (void**) &p, true);
taosRemoveRef(tscObjRef, pSql->self);
taosReleaseRef(tscObjRef, pSql->self);
rpcFreeCont(rpcMsg->pCont);
return;
}
......@@ -322,7 +310,7 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
// if there is an error occurring, proceed to the following error handling procedure.
if (rpcMsg->code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
taosCacheRelease(tscObjCache, (void**) &p, false);
taosReleaseRef(tscObjRef, pSql->self);
rpcFreeCont(rpcMsg->pCont);
return;
}
......@@ -390,11 +378,10 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
(*pSql->fp)(pSql->param, pSql, rpcMsg->code);
}
void** p1 = p;
taosCacheRelease(tscObjCache, (void**) &p1, false);
taosReleaseRef(tscObjRef, pSql->self);
if (shouldFree) { // in case of table-meta/vgrouplist query, automatically free it
taosCacheRelease(tscObjCache, (void **)&p, true);
taosRemoveRef(tscObjRef, pSql->self);
tscDebug("%p sqlObj is automatically freed", pSql);
}
......@@ -2021,7 +2008,7 @@ int tscProcessShowRsp(SSqlObj *pSql) {
// TODO multithread problem
static void createHBObj(STscObj* pObj) {
if (pObj->pHb != NULL) {
if (pObj->hbrid != 0) {
return;
}
......@@ -2053,7 +2040,7 @@ static void createHBObj(STscObj* pObj) {
registerSqlObj(pSql);
tscDebug("%p HB is allocated, pObj:%p", pSql, pObj);
pObj->pHb = pSql;
pObj->hbrid = pSql->self;
}
int tscProcessConnectRsp(SSqlObj *pSql) {
......@@ -2072,6 +2059,10 @@ int tscProcessConnectRsp(SSqlObj *pSql) {
if (pConnect->epSet.numOfEps > 0) {
tscEpSetHtons(&pConnect->epSet);
tscUpdateMgmtEpSet(&pConnect->epSet);
for (int i = 0; i < pConnect->epSet.numOfEps; ++i) {
tscDebug("%p epSet.fqdn[%d]: %s, pObj:%p", pSql, i, pConnect->epSet.fqdn[i], pObj);
}
}
strcpy(pObj->sversion, pConnect->serverVersion);
......
......@@ -276,8 +276,8 @@ void taos_close(TAOS *taos) {
pObj->signature = NULL;
taosTmrStopA(&(pObj->pTimer));
SSqlObj* pHb = pObj->pHb;
if (pHb != NULL && atomic_val_compare_exchange_ptr(&pObj->pHb, pHb, 0) == pHb) {
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;
......@@ -285,6 +285,7 @@ void taos_close(TAOS *taos) {
tscDebug("%p HB is freed", pHb);
taos_free_result(pHb);
taosReleaseRef(tscObjRef, pHb->self);
}
int32_t ref = T_REF_DEC(pObj);
......@@ -606,8 +607,7 @@ void taos_free_result(TAOS_RES *res) {
bool freeNow = tscKillQueryInDnode(pSql);
if (freeNow) {
tscDebug("%p free sqlObj in cache", pSql);
SSqlObj** p = pSql->self;
taosCacheRelease(tscObjCache, (void**) &p, true);
taosReleaseRef(tscObjRef, pSql->self);
}
}
......@@ -700,13 +700,7 @@ static void tscKillSTableQuery(SSqlObj *pSql) {
continue;
}
void** p = taosCacheAcquireByKey(tscObjCache, &pSub, sizeof(TSDB_CACHE_PTR_TYPE));
if (p == NULL) {
continue;
}
SSqlObj* pSubObj = (SSqlObj*) (*p);
assert(pSubObj->self == (SSqlObj**) p);
SSqlObj* pSubObj = pSub;
pSubObj->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
if (pSubObj->rpcRid > 0) {
......@@ -715,7 +709,7 @@ static void tscKillSTableQuery(SSqlObj *pSql) {
}
tscQueueAsyncRes(pSubObj);
taosCacheRelease(tscObjCache, (void**) &p, false);
taosReleaseRef(tscObjRef, pSubObj->self);
}
tscDebug("%p super table query cancelled", pSql);
......
......@@ -157,7 +157,7 @@ static SSub* tscCreateSubscription(STscObj* pObj, const char* topic, const char*
registerSqlObj(pSql);
code = tsParseSql(pSql, false);
code = tsParseSql(pSql, true);
if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
tsem_wait(&pSub->sem);
code = pSql->res.code;
......@@ -168,7 +168,7 @@ static SSub* tscCreateSubscription(STscObj* pObj, const char* topic, const char*
goto fail;
}
if (pSql->cmd.command != TSDB_SQL_SELECT) {
if (pSql->cmd.command != TSDB_SQL_SELECT && pSql->cmd.command != TSDB_SQL_RETRIEVE_EMPTY_RESULT) {
line = __LINE__;
code = TSDB_CODE_TSC_INVALID_SQL;
goto fail;
......@@ -179,10 +179,10 @@ static SSub* tscCreateSubscription(STscObj* pObj, const char* topic, const char*
fail:
tscError("tscCreateSubscription failed at line %d, reason: %s", line, tstrerror(code));
if (pSql != NULL) {
if (pSql->self != NULL) {
taos_free_result(pSql);
if (pSql->self != 0) {
taosReleaseRef(tscObjRef, pSql->self);
} else {
tscFreeSqlObj(pSql);
tscFreeSqlObj(pSql);
}
pSql = NULL;
......@@ -401,9 +401,11 @@ TAOS_SUB *taos_subscribe(TAOS *taos, int restart, const char* topic, const char
tscLoadSubscriptionProgress(pSub);
}
if (!tscUpdateSubscription(pObj, pSub)) {
taos_unsubscribe(pSub, 1);
return NULL;
if (pSub->pSql->cmd.command == TSDB_SQL_SELECT) {
if (!tscUpdateSubscription(pObj, pSub)) {
taos_unsubscribe(pSub, 1);
return NULL;
}
}
pSub->interval = interval;
......@@ -417,10 +419,80 @@ TAOS_SUB *taos_subscribe(TAOS *taos, int restart, const char* topic, const char
return pSub;
}
SSqlObj* recreateSqlObj(SSub* pSub) {
SSqlObj* pSql = calloc(1, sizeof(SSqlObj));
if (pSql == NULL) {
return NULL;
}
pSql->signature = pSql;
pSql->pTscObj = pSub->taos;
SSqlCmd* pCmd = &pSql->cmd;
SSqlRes* pRes = &pSql->res;
if (tsem_init(&pSql->rspSem, 0, 0) == -1) {
tscFreeSqlObj(pSql);
return NULL;
}
pSql->param = pSub;
pSql->maxRetry = TSDB_MAX_REPLICA;
pSql->fp = asyncCallback;
pSql->fetchFp = asyncCallback;
pSql->sqlstr = strdup(pSub->pSql->sqlstr);
if (pSql->sqlstr == NULL) {
tscFreeSqlObj(pSql);
return NULL;
}
pRes->qhandle = 0;
pRes->numOfRows = 1;
int code = tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE);
if (code != TSDB_CODE_SUCCESS) {
tscFreeSqlObj(pSql);
return NULL;
}
registerSqlObj(pSql);
code = tsParseSql(pSql, true);
if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
tsem_wait(&pSub->sem);
code = pSql->res.code;
}
if (code != TSDB_CODE_SUCCESS) {
taosReleaseRef(tscObjRef, pSql->self);
return NULL;
}
if (pSql->cmd.command != TSDB_SQL_SELECT) {
taosReleaseRef(tscObjRef, pSql->self);
return NULL;
}
return pSql;
}
TAOS_RES *taos_consume(TAOS_SUB *tsub) {
SSub *pSub = (SSub *)tsub;
if (pSub == NULL) return NULL;
if (pSub->pSql->cmd.command == TSDB_SQL_RETRIEVE_EMPTY_RESULT) {
SSqlObj* pSql = recreateSqlObj(pSub);
if (pSql == NULL) {
return NULL;
}
if (pSub->pSql->self != 0) {
taosReleaseRef(tscObjRef, pSub->pSql->self);
} else {
tscFreeSqlObj(pSub->pSql);
}
pSub->pSql = pSql;
pSql->pSubscription = pSub;
}
tscSaveSubscriptionProgress(pSub);
SSqlObj *pSql = pSub->pSql;
......@@ -512,10 +584,13 @@ void taos_unsubscribe(TAOS_SUB *tsub, int keepProgress) {
}
if (pSub->pSql != NULL) {
taos_free_result(pSub->pSql);
if (pSub->pSql->self != 0) {
taosReleaseRef(tscObjRef, pSub->pSql->self);
} else {
tscFreeSqlObj(pSub->pSql);
}
}
tscFreeSqlObj(pSub->pSql);
taosArrayDestroy(pSub->progress);
tsem_destroy(&pSub->sem);
memset(pSub, 0, sizeof(*pSub));
......
......@@ -2198,6 +2198,9 @@ int32_t tscHandleInsertRetry(SSqlObj* pSql) {
STableDataBlocks* pTableDataBlock = taosArrayGetP(pCmd->pDataBlocks, pSupporter->index);
int32_t code = tscCopyDataBlockToPayload(pSql, pTableDataBlock);
// free the data block created from insert sql string
pCmd->pDataBlocks = tscDestroyBlockArrayList(pCmd->pDataBlocks);
if ((pRes->code = code)!= TSDB_CODE_SUCCESS) {
tscQueueAsyncRes(pSql);
return code; // here the pSql may have been released already.
......
......@@ -15,7 +15,7 @@
#include "os.h"
#include "taosmsg.h"
#include "tcache.h"
#include "tref.h"
#include "trpc.h"
#include "tsystem.h"
#include "ttimer.h"
......@@ -31,7 +31,7 @@
// global, not configurable
SCacheObj* tscMetaCache;
SCacheObj* tscObjCache;
int tscObjRef = -1;
void * tscTmr;
void * tscQhandle;
void * tscCheckDiskUsageTmr;
......@@ -144,7 +144,7 @@ void taos_init_imp(void) {
int64_t refreshTime = 10; // 10 seconds by default
if (tscMetaCache == NULL) {
tscMetaCache = taosCacheInit(TSDB_DATA_TYPE_BINARY, refreshTime, false, tscFreeTableMetaHelper, "tableMeta");
tscObjCache = taosCacheInit(TSDB_CACHE_PTR_KEY, refreshTime / 2, false, tscFreeRegisteredSqlObj, "sqlObj");
tscObjRef = taosOpenRef(4096, tscFreeRegisteredSqlObj);
}
tscRefId = taosOpenRef(200, tscCloseTscObj);
......@@ -167,9 +167,9 @@ void taos_cleanup(void) {
taosCacheCleanup(m);
}
m = tscObjCache;
if (m != NULL && atomic_val_compare_exchange_ptr(&tscObjCache, m, 0) == m) {
taosCacheCleanup(m);
int refId = atomic_exchange_32(&tscObjRef, -1);
if (refId != -1) {
taosCloseRef(refId);
}
m = tscQhandle;
......
......@@ -447,20 +447,18 @@ static void tscFreeSubobj(SSqlObj* pSql) {
void tscFreeRegisteredSqlObj(void *pSql) {
assert(pSql != NULL);
SSqlObj** p = (SSqlObj**)pSql;
STscObj* pTscObj = (*p)->pTscObj;
SSqlObj* p = *(SSqlObj**)pSql;
STscObj* pTscObj = p->pTscObj;
assert((*p)->self != 0 && (*p)->self == (p));
SSqlObj* ptr = *p;
tscFreeSqlObj(*p);
assert(p->self != 0);
tscFreeSqlObj(p);
int32_t ref = T_REF_DEC(pTscObj);
assert(ref >= 0);
tscDebug("%p free sqlObj completed, tscObj:%p ref:%d", ptr, pTscObj, ref);
tscDebug("%p free sqlObj completed, tscObj:%p ref:%d", p, pTscObj, ref);
if (ref == 0) {
tscDebug("%p all sqlObj freed, free tscObj:%p", ptr, pTscObj);
tscDebug("%p all sqlObj freed, free tscObj:%p", p, pTscObj);
taosRemoveRef(tscRefId, pTscObj->rid);
}
}
......@@ -840,7 +838,6 @@ int32_t tscMergeTableDataBlocks(SSqlObj* pSql, SArray* pTableDataBlockList) {
// the length does not include the SSubmitBlk structure
pBlocks->dataLen = htonl(finalLen);
dataBuf->numOfTables += 1;
}
......@@ -1565,19 +1562,6 @@ void tscGetSrcColumnInfo(SSrcColumnInfo* pColInfo, SQueryInfo* pQueryInfo) {
}
}
void tscSetFreeHeatBeat(STscObj* pObj) {
if (pObj == NULL || pObj->signature != pObj || pObj->pHb == NULL) {
return;
}
SSqlObj* pHeatBeat = pObj->pHb;
assert(pHeatBeat == pHeatBeat->signature);
// to denote the heart-beat timer close connection and free all allocated resources
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pHeatBeat->cmd, 0);
pQueryInfo->type = TSDB_QUERY_TYPE_FREE_RESOURCE;
}
/*
* the following four kinds of SqlObj should not be freed
* 1. SqlObj for stream computing
......@@ -1596,7 +1580,7 @@ bool tscShouldBeFreed(SSqlObj* pSql) {
}
STscObj* pTscObj = pSql->pTscObj;
if (pSql->pStream != NULL || pTscObj->pHb == pSql || pSql->pSubscription != NULL) {
if (pSql->pStream != NULL || pTscObj->hbrid == pSql->self || pSql->pSubscription != NULL) {
return false;
}
......@@ -1888,13 +1872,10 @@ void tscResetForNextRetrieve(SSqlRes* pRes) {
}
void registerSqlObj(SSqlObj* pSql) {
int32_t DEFAULT_LIFE_TIME = 2 * 600 * 1000; // 1200 sec
int32_t ref = T_REF_INC(pSql->pTscObj);
tscDebug("%p add to tscObj:%p, ref:%d", pSql, pSql->pTscObj, ref);
TSDB_CACHE_PTR_TYPE p = (TSDB_CACHE_PTR_TYPE) pSql;
pSql->self = taosCachePut(tscObjCache, &p, sizeof(TSDB_CACHE_PTR_TYPE), &p, sizeof(TSDB_CACHE_PTR_TYPE), DEFAULT_LIFE_TIME);
pSql->self = taosAddRef(tscObjRef, pSql);
}
SSqlObj* createSimpleSubObj(SSqlObj* pSql, void (*fp)(), void* param, int32_t cmd) {
......
......@@ -125,6 +125,9 @@ extern char tsMonitorDbName[];
extern char tsInternalPass[];
extern int32_t tsMonitorInterval;
// stream
extern int32_t tsEnableStream;
// internal
extern int32_t tsPrintAuth;
extern int32_t tscEmbedded;
......
......@@ -161,6 +161,9 @@ char tsMonitorDbName[TSDB_DB_NAME_LEN] = "log";
char tsInternalPass[] = "secretkey";
int32_t tsMonitorInterval = 30; // seconds
// stream
int32_t tsEnableStream = 1;
// internal
int32_t tsPrintAuth = 0;
int32_t tscEmbedded = 0;
......@@ -1015,6 +1018,16 @@ static void doInitGlobalConfig(void) {
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosInitConfigOption(cfg);
cfg.option = "stream";
cfg.ptr = &tsEnableStream;
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 = 1;
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosInitConfigOption(cfg);
cfg.option = "httpEnableRecordSql";
cfg.ptr = &tsHttpEnableRecordSql;
cfg.valType = TAOS_CFG_VTYPE_INT32;
......
......@@ -40,15 +40,14 @@
typedef struct {
int32_t vgId;
int32_t master;
int32_t num; // number of continuous streams
char user[TSDB_USER_LEN];
char pass[TSDB_PASSWORD_LEN];
char db[TSDB_DB_NAME_LEN];
FCqWrite cqWrite;
void *ahandle;
int32_t num; // number of continuous streams
struct SCqObj *pHead;
void *dbConn;
int32_t master;
void *tmrCtrl;
pthread_mutex_t mutex;
} SCqContext;
......@@ -70,6 +69,9 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row);
static void cqCreateStream(SCqContext *pContext, SCqObj *pObj);
void *cqOpen(void *ahandle, const SCqCfg *pCfg) {
if (tsEnableStream == 0) {
return NULL;
}
SCqContext *pContext = calloc(sizeof(SCqContext), 1);
if (pContext == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
......@@ -90,7 +92,6 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) {
tstrncpy(pContext->db, db, sizeof(pContext->db));
pContext->vgId = pCfg->vgId;
pContext->cqWrite = pCfg->cqWrite;
pContext->ahandle = ahandle;
tscEmbedded = 1;
pthread_mutex_init(&pContext->mutex, NULL);
......@@ -101,6 +102,9 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) {
}
void cqClose(void *handle) {
if (tsEnableStream == 0) {
return;
}
SCqContext *pContext = handle;
if (handle == NULL) return;
......@@ -131,6 +135,9 @@ void cqClose(void *handle) {
}
void cqStart(void *handle) {
if (tsEnableStream == 0) {
return;
}
SCqContext *pContext = handle;
if (pContext->dbConn || pContext->master) return;
......@@ -149,6 +156,9 @@ void cqStart(void *handle) {
}
void cqStop(void *handle) {
if (tsEnableStream == 0) {
return;
}
SCqContext *pContext = handle;
cInfo("vgId:%d, stop all CQs", pContext->vgId);
if (pContext->dbConn == NULL || pContext->master == 0) return;
......@@ -176,6 +186,9 @@ void cqStop(void *handle) {
}
void *cqCreate(void *handle, uint64_t uid, int32_t tid, char *sqlStr, STSchema *pSchema) {
if (tsEnableStream == 0) {
return NULL;
}
SCqContext *pContext = handle;
SCqObj *pObj = calloc(sizeof(SCqObj), 1);
......@@ -205,6 +218,9 @@ void *cqCreate(void *handle, uint64_t uid, int32_t tid, char *sqlStr, STSchema *
}
void cqDrop(void *handle) {
if (tsEnableStream == 0) {
return;
}
SCqObj *pObj = handle;
SCqContext *pContext = pObj->pContext;
......@@ -241,8 +257,12 @@ static void doCreateStream(void *param, TAOS_RES *result, int32_t code) {
SCqObj* pObj = (SCqObj*)param;
SCqContext* pContext = pObj->pContext;
SSqlObj* pSql = (SSqlObj*)result;
pContext->dbConn = pSql->pTscObj;
if (atomic_val_compare_exchange_ptr(&(pContext->dbConn), NULL, pSql->pTscObj) != NULL) {
taos_close(pSql->pTscObj);
}
pthread_mutex_lock(&pContext->mutex);
cqCreateStream(pContext, pObj);
pthread_mutex_unlock(&pContext->mutex);
}
static void cqProcessCreateTimer(void *param, void *tmrId) {
......@@ -253,7 +273,9 @@ static void cqProcessCreateTimer(void *param, void *tmrId) {
cDebug("vgId:%d, try connect to TDengine", pContext->vgId);
taos_connect_a(NULL, pContext->user, pContext->pass, pContext->db, 0, doCreateStream, param, NULL);
} else {
pthread_mutex_lock(&pContext->mutex);
cqCreateStream(pContext, pObj);
pthread_mutex_unlock(&pContext->mutex);
}
}
......@@ -267,12 +289,14 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
}
pObj->tmrId = 0;
pObj->pStream = taos_open_stream(pContext->dbConn, pObj->sqlStr, cqProcessStreamRes, 0, pObj, NULL);
if (pObj->pStream) {
pContext->num++;
cInfo("vgId:%d, id:%d CQ:%s is openned", pContext->vgId, pObj->tid, pObj->sqlStr);
} else {
cError("vgId:%d, id:%d CQ:%s, failed to open", pContext->vgId, pObj->tid, pObj->sqlStr);
if (pObj->pStream == NULL) {
pObj->pStream = taos_open_stream(pContext->dbConn, pObj->sqlStr, cqProcessStreamRes, 0, pObj, NULL);
if (pObj->pStream) {
pContext->num++;
cInfo("vgId:%d, id:%d CQ:%s is openned", pContext->vgId, pObj->tid, pObj->sqlStr);
} else {
cError("vgId:%d, id:%d CQ:%s, failed to open", pContext->vgId, pObj->tid, pObj->sqlStr);
}
}
}
......@@ -334,7 +358,7 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
pHead->version = 0;
// write into vnode write queue
pContext->cqWrite(pContext->ahandle, pHead, TAOS_QTYPE_CQ, NULL);
pContext->cqWrite(pContext->vgId, pHead, TAOS_QTYPE_CQ, NULL);
free(buffer);
}
......@@ -24,7 +24,7 @@
int64_t ver = 0;
void *pCq = NULL;
int writeToQueue(void *pVnode, void *data, int type, void *pMsg) {
int writeToQueue(int32_t vgId, void *data, int type, void *pMsg) {
return 0;
}
......
......@@ -444,12 +444,12 @@ static int32_t dnodeProcessCreateMnodeMsg(SRpcMsg *pMsg) {
SCreateMnodeMsg *pCfg = pMsg->pCont;
pCfg->dnodeId = htonl(pCfg->dnodeId);
if (pCfg->dnodeId != dnodeGetDnodeId()) {
dError("dnodeId:%d, in create mnode msg is not equal with saved dnodeId:%d", pCfg->dnodeId, dnodeGetDnodeId());
dDebug("dnodeId:%d, in create mnode msg is not equal with saved dnodeId:%d", pCfg->dnodeId, dnodeGetDnodeId());
return TSDB_CODE_MND_DNODE_ID_NOT_CONFIGURED;
}
if (strcmp(pCfg->dnodeEp, tsLocalEp) != 0) {
dError("dnodeEp:%s, in create mnode msg is not equal with saved dnodeEp:%s", pCfg->dnodeEp, tsLocalEp);
dDebug("dnodeEp:%s, in create mnode msg is not equal with saved dnodeEp:%s", pCfg->dnodeEp, tsLocalEp);
return TSDB_CODE_MND_DNODE_EP_NOT_CONFIGURED;
}
......
......@@ -257,7 +257,7 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size, void* buf
#define TSDB_MAX_SAVED_SQL_LEN TSDB_MAX_COLUMNS * 64
#define TSDB_MAX_SQL_LEN TSDB_PAYLOAD_SIZE
#define TSDB_MAX_SQL_SHOW_LEN 512
#define TSDB_MAX_ALLOWED_SQL_LEN (8*1024*1024U) // sql length should be less than 8mb
#define TSDB_MAX_ALLOWED_SQL_LEN (1*1024*1024U) // sql length should be less than 1mb
#define TSDB_APPNAME_LEN TSDB_UNI_LEN
......
......@@ -184,6 +184,9 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_DATABASES, 0, 0x0385, "Too many d
TAOS_DEFINE_ERROR(TSDB_CODE_MND_DB_IN_DROPPING, 0, 0x0386, "Database not available")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_VGROUP_NOT_READY, 0, 0x0387, "Database unsynced")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_DB_OPTION_DAYS, 0, 0x0390, "Invalid database option: days out of range")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_DB_OPTION_KEEP, 0, 0x0391, "Invalid database option: keep >= keep2 >= keep1 >= days")
// dnode
TAOS_DEFINE_ERROR(TSDB_CODE_DND_MSG_NOT_PROCESSED, 0, 0x0400, "Message not processed")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_OUT_OF_MEMORY, 0, 0x0401, "Dnode out of memory")
......@@ -261,9 +264,6 @@ TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INVALID_CONFIG, 0, 0x0900, "Invalid Sy
TAOS_DEFINE_ERROR(TSDB_CODE_SYN_NOT_ENABLED, 0, 0x0901, "Sync module not enabled")
TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INVALID_VERSION, 0, 0x0902, "Invalid Sync version")
TAOS_DEFINE_ERROR(TSDB_CODE_SYN_CONFIRM_EXPIRED, 0, 0x0903, "Sync confirm expired")
TAOS_DEFINE_ERROR(TSDB_CODE_SYN_VND_COMMITING, 0, 0x0904, "Vnode is commiting")
TAOS_DEFINE_ERROR(TSDB_CODE_SYN_FILE_CHNAGED, 0, 0x0905, "Vnode file is changed")
TAOS_DEFINE_ERROR(TSDB_CODE_SYN_APP_ERROR, 0, 0x1000, "Unexpected generic error in sync")
// wal
TAOS_DEFINE_ERROR(TSDB_CODE_WAL_APP_ERROR, 0, 0x1000, "Unexpected generic error in wal")
......
......@@ -21,7 +21,7 @@ extern "C" {
#include "tdataformat.h"
typedef int32_t (*FCqWrite)(void *ahandle, void *pHead, int32_t qtype, void *pMsg);
typedef int32_t (*FCqWrite)(int32_t vgId, void *pHead, int32_t qtype, void *pMsg);
typedef struct {
int32_t vgId;
......
......@@ -86,7 +86,7 @@ typedef void (*FNotifyFlowCtrl)(int32_t vgId, int32_t level);
typedef int32_t (*FNotifyFileSynced)(int32_t vgId, uint64_t fversion);
// get file version
typedef int32_t (*FGetFileVersion)(int32_t vgId, uint64_t *fver);
typedef int32_t (*FGetVersion)(int32_t vgId, uint64_t *fver, uint64_t *vver);
typedef struct {
int32_t vgId; // vgroup ID
......@@ -100,7 +100,7 @@ typedef struct {
FNotifyRole notifyRole;
FNotifyFlowCtrl notifyFlowCtrl;
FNotifyFileSynced notifyFileSynced;
FGetFileVersion getFileVersion;
FGetVersion getVersion;
} SSyncInfo;
typedef void *tsync_h;
......
......@@ -407,7 +407,11 @@ void get_history_path(char *history) { sprintf(history, "%s/%s", getpwuid(getuid
void clearScreen(int ecmd_pos, int cursor_pos) {
struct winsize w;
ioctl(0, TIOCGWINSZ, &w);
if (ioctl(0, TIOCGWINSZ, &w) < 0 || w.ws_col == 0 || w.ws_row == 0) {
//fprintf(stderr, "No stream device, and use default value(col 120, row 30)\n");
w.ws_col = 120;
w.ws_row = 30;
}
int cursor_x = cursor_pos / w.ws_col;
int cursor_y = cursor_pos % w.ws_col;
......@@ -425,8 +429,9 @@ void clearScreen(int ecmd_pos, int cursor_pos) {
void showOnScreen(Command *cmd) {
struct winsize w;
if (ioctl(0, TIOCGWINSZ, &w) < 0 || w.ws_col == 0 || w.ws_row == 0) {
fprintf(stderr, "No stream device\n");
exit(EXIT_FAILURE);
//fprintf(stderr, "No stream device\n");
w.ws_col = 120;
w.ws_row = 30;
}
wchar_t wc;
......
......@@ -413,7 +413,11 @@ void get_history_path(char *history) { snprintf(history, TSDB_FILENAME_LEN, "%s/
void clearScreen(int ecmd_pos, int cursor_pos) {
struct winsize w;
ioctl(0, TIOCGWINSZ, &w);
if (ioctl(0, TIOCGWINSZ, &w) < 0 || w.ws_col == 0 || w.ws_row == 0) {
//fprintf(stderr, "No stream device, and use default value(col 120, row 30)\n");
w.ws_col = 120;
w.ws_row = 30;
}
int cursor_x = cursor_pos / w.ws_col;
int cursor_y = cursor_pos % w.ws_col;
......@@ -431,8 +435,9 @@ void clearScreen(int ecmd_pos, int cursor_pos) {
void showOnScreen(Command *cmd) {
struct winsize w;
if (ioctl(0, TIOCGWINSZ, &w) < 0 || w.ws_col == 0 || w.ws_row == 0) {
fprintf(stderr, "No stream device\n");
exit(EXIT_FAILURE);
//fprintf(stderr, "No stream device\n");
w.ws_col = 120;
w.ws_row = 30;
}
wchar_t wc;
......
......@@ -79,10 +79,13 @@ typedef struct {
int32_t (*fpRestored)();
} SSdbTableDesc;
int32_t sdbInitRef();
void sdbCleanUpRef();
int32_t sdbInit();
void sdbCleanUp();
void * sdbOpenTable(SSdbTableDesc *desc);
void sdbCloseTable(void *handle);
int64_t sdbOpenTable(SSdbTableDesc *desc);
void sdbCloseTable(int64_t rid);
void* sdbGetTableByRid(int64_t rid);
bool sdbIsMaster();
bool sdbIsServing();
void sdbUpdateMnodeRoles();
......
......@@ -26,6 +26,7 @@
#include "mnodeUser.h"
#include "mnodeVgroup.h"
int64_t tsAcctRid = -1;
void * tsAcctSdb = NULL;
static int32_t tsAcctUpdateSize;
static int32_t mnodeCreateRootAcct();
......@@ -114,7 +115,8 @@ int32_t mnodeInitAccts() {
.fpRestored = mnodeAcctActionRestored
};
tsAcctSdb = sdbOpenTable(&desc);
tsAcctRid = sdbOpenTable(&desc);
tsAcctSdb = sdbGetTableByRid(tsAcctRid);
if (tsAcctSdb == NULL) {
mError("table:%s, failed to create hash", desc.name);
return -1;
......@@ -126,7 +128,7 @@ int32_t mnodeInitAccts() {
void mnodeCleanupAccts() {
acctCleanUp();
sdbCloseTable(tsAcctSdb);
sdbCloseTable(tsAcctRid);
tsAcctSdb = NULL;
}
......
......@@ -24,6 +24,7 @@
#include "mnodeShow.h"
#include "tglobal.h"
int64_t tsClusterRid = -1;
static void * tsClusterSdb = NULL;
static int32_t tsClusterUpdateSize;
static char tsClusterId[TSDB_CLUSTER_ID_LEN];
......@@ -101,9 +102,10 @@ int32_t mnodeInitCluster() {
.fpRestored = mnodeClusterActionRestored
};
tsClusterSdb = sdbOpenTable(&desc);
tsClusterRid = sdbOpenTable(&desc);
tsClusterSdb = sdbGetTableByRid(tsClusterRid);
if (tsClusterSdb == NULL) {
mError("table:%s, failed to create hash", desc.name);
mError("table:%s, rid:%" PRId64 ", failed to create hash", desc.name, tsClusterRid);
return -1;
}
......@@ -116,7 +118,7 @@ int32_t mnodeInitCluster() {
}
void mnodeCleanupCluster() {
sdbCloseTable(tsClusterSdb);
sdbCloseTable(tsClusterRid);
tsClusterSdb = NULL;
}
......
......@@ -38,6 +38,7 @@
#include "mnodeVgroup.h"
#define VG_LIST_SIZE 8
int64_t tsDbRid = -1;
static void * tsDbSdb = NULL;
static int32_t tsDbUpdateSize;
......@@ -160,7 +161,8 @@ int32_t mnodeInitDbs() {
.fpRestored = mnodeDbActionRestored
};
tsDbSdb = sdbOpenTable(&desc);
tsDbRid = sdbOpenTable(&desc);
tsDbSdb = sdbGetTableByRid(tsDbRid);
if (tsDbSdb == NULL) {
mError("failed to init db data");
return -1;
......@@ -234,30 +236,28 @@ static int32_t mnodeCheckDbCfg(SDbCfg *pCfg) {
if (pCfg->daysPerFile < TSDB_MIN_DAYS_PER_FILE || pCfg->daysPerFile > TSDB_MAX_DAYS_PER_FILE) {
mError("invalid db option daysPerFile:%d valid range: [%d, %d]", pCfg->daysPerFile, TSDB_MIN_DAYS_PER_FILE,
TSDB_MAX_DAYS_PER_FILE);
return TSDB_CODE_MND_INVALID_DB_OPTION;
return TSDB_CODE_MND_INVALID_DB_OPTION_DAYS;
}
if (pCfg->daysToKeep < TSDB_MIN_KEEP || pCfg->daysToKeep > TSDB_MAX_KEEP) {
mError("invalid db option daysToKeep:%d valid range: [%d, %d]", pCfg->daysToKeep, TSDB_MIN_KEEP, TSDB_MAX_KEEP);
return TSDB_CODE_MND_INVALID_DB_OPTION;
return TSDB_CODE_MND_INVALID_DB_OPTION_KEEP;
}
if (pCfg->daysToKeep < pCfg->daysPerFile) {
mError("invalid db option daysToKeep:%d should larger than daysPerFile:%d", pCfg->daysToKeep, pCfg->daysPerFile);
return TSDB_CODE_MND_INVALID_DB_OPTION;
return TSDB_CODE_MND_INVALID_DB_OPTION_KEEP;
}
#if 0
if (pCfg->daysToKeep2 < TSDB_MIN_KEEP || pCfg->daysToKeep2 > pCfg->daysToKeep) {
mError("invalid db option daysToKeep2:%d valid range: [%d, %d]", pCfg->daysToKeep, TSDB_MIN_KEEP, pCfg->daysToKeep);
return TSDB_CODE_MND_INVALID_DB_OPTION;
mError("invalid db option daysToKeep2:%d valid range: [%d, %d]", pCfg->daysToKeep2, TSDB_MIN_KEEP, pCfg->daysToKeep);
return TSDB_CODE_MND_INVALID_DB_OPTION_KEEP;
}
if (pCfg->daysToKeep1 < TSDB_MIN_KEEP || pCfg->daysToKeep1 > pCfg->daysToKeep2) {
mError("invalid db option daysToKeep1:%d valid range: [%d, %d]", pCfg->daysToKeep1, TSDB_MIN_KEEP, pCfg->daysToKeep2);
return TSDB_CODE_MND_INVALID_DB_OPTION;
return TSDB_CODE_MND_INVALID_DB_OPTION_KEEP;
}
#endif
if (pCfg->maxRowsPerFileBlock < TSDB_MIN_MAX_ROW_FBLOCK || pCfg->maxRowsPerFileBlock > TSDB_MAX_MAX_ROW_FBLOCK) {
mError("invalid db option maxRowsPerFileBlock:%d valid range: [%d, %d]", pCfg->maxRowsPerFileBlock,
......@@ -496,7 +496,7 @@ void mnodeRemoveVgroupFromDb(SVgObj *pVgroup) {
}
void mnodeCleanupDbs() {
sdbCloseTable(tsDbSdb);
sdbCloseTable(tsDbRid);
tsDbSdb = NULL;
}
......
......@@ -39,6 +39,7 @@
#include "mnodeCluster.h"
int32_t tsAccessSquence = 0;
int64_t tsDnodeRid = -1;
static void * tsDnodeSdb = NULL;
static int32_t tsDnodeUpdateSize = 0;
extern void * tsMnodeSdb;
......@@ -187,7 +188,8 @@ int32_t mnodeInitDnodes() {
.fpRestored = mnodeDnodeActionRestored
};
tsDnodeSdb = sdbOpenTable(&desc);
tsDnodeRid = sdbOpenTable(&desc);
tsDnodeSdb = sdbGetTableByRid(tsDnodeRid);
if (tsDnodeSdb == NULL) {
mError("failed to init dnodes data");
return -1;
......@@ -213,7 +215,7 @@ int32_t mnodeInitDnodes() {
}
void mnodeCleanupDnodes() {
sdbCloseTable(tsDnodeSdb);
sdbCloseTable(tsDnodeRid);
pthread_mutex_destroy(&tsDnodeEpsMutex);
free(tsDnodeEps);
tsDnodeEps = NULL;
......
......@@ -47,6 +47,7 @@ void *tsMnodeTmr = NULL;
static bool tsMgmtIsRunning = false;
static const SMnodeComponent tsMnodeComponents[] = {
{"sdbref", sdbInitRef, sdbCleanUpRef},
{"profile", mnodeInitProfile, mnodeCleanupProfile},
{"cluster", mnodeInitCluster, mnodeCleanupCluster},
{"accts", mnodeInitAccts, mnodeCleanupAccts},
......
......@@ -34,6 +34,7 @@
#include "mnodeUser.h"
#include "mnodeVgroup.h"
int64_t tsMnodeRid = -1;
static void * tsMnodeSdb = NULL;
static int32_t tsMnodeUpdateSize = 0;
static SRpcEpSet tsMnodeEpSetForShell;
......@@ -153,7 +154,8 @@ int32_t mnodeInitMnodes() {
.fpRestored = mnodeMnodeActionRestored
};
tsMnodeSdb = sdbOpenTable(&desc);
tsMnodeRid = sdbOpenTable(&desc);
tsMnodeSdb = sdbGetTableByRid(tsMnodeRid);
if (tsMnodeSdb == NULL) {
mError("failed to init mnodes data");
return -1;
......@@ -168,7 +170,7 @@ int32_t mnodeInitMnodes() {
}
void mnodeCleanupMnodes() {
sdbCloseTable(tsMnodeSdb);
sdbCloseTable(tsMnodeRid);
tsMnodeSdb = NULL;
mnodeMnodeDestroyLock();
}
......@@ -251,12 +253,30 @@ void mnodeGetMnodeEpSetForPeer(SRpcEpSet *epSet) {
mnodeMnodeRdLock();
*epSet = tsMnodeEpSetForPeer;
mnodeMnodeUnLock();
for (int32_t i = 0; i < epSet->numOfEps; ++i) {
if (strcmp(epSet->fqdn[i], tsLocalFqdn) == 0 && htons(epSet->port[i]) == tsServerPort + TSDB_PORT_DNODEDNODE) {
epSet->inUse = (i + 1) % epSet->numOfEps;
mTrace("mnode:%d, for peer ep:%s:%u, set inUse to %d", i, epSet->fqdn[i], htons(epSet->port[i]), epSet->inUse);
} else {
mTrace("mpeer:%d, for peer ep:%s:%u", i, epSet->fqdn[i], htons(epSet->port[i]));
}
}
}
void mnodeGetMnodeEpSetForShell(SRpcEpSet *epSet) {
mnodeMnodeRdLock();
*epSet = tsMnodeEpSetForShell;
mnodeMnodeUnLock();
for (int32_t i = 0; i < epSet->numOfEps; ++i) {
if (strcmp(epSet->fqdn[i], tsLocalFqdn) == 0 && htons(epSet->port[i]) == tsServerPort) {
epSet->inUse = (i + 1) % epSet->numOfEps;
mTrace("mnode:%d, for shell ep:%s:%u, set inUse to %d", i, epSet->fqdn[i], htons(epSet->port[i]), epSet->inUse);
} else {
mTrace("mnode:%d, for shell ep:%s:%u", i, epSet->fqdn[i], htons(epSet->port[i]));
}
}
}
char* mnodeGetMnodeMasterEp() {
......
......@@ -58,16 +58,8 @@ int32_t mnodeProcessPeerReq(SMnodeMsg *pMsg) {
rpcRsp->rsp = epSet;
rpcRsp->len = sizeof(SRpcEpSet);
mDebug("msg:%p, ahandle:%p type:%s in mpeer queue will be redirected, numOfEps:%d inUse:%d", pMsg,
pMsg->rpcMsg.ahandle, taosMsg[pMsg->rpcMsg.msgType], epSet->numOfEps, epSet->inUse);
for (int32_t i = 0; i < epSet->numOfEps; ++i) {
if (strcmp(epSet->fqdn[i], tsLocalFqdn) == 0 && htons(epSet->port[i]) == tsServerPort + TSDB_PORT_DNODEDNODE) {
epSet->inUse = (i + 1) % epSet->numOfEps;
mDebug("mpeer:%d ep:%s:%u, set inUse to %d", i, epSet->fqdn[i], htons(epSet->port[i]), epSet->inUse);
} else {
mDebug("mpeer:%d ep:%s:%u", i, epSet->fqdn[i], htons(epSet->port[i]));
}
}
mDebug("msg:%p, ahandle:%p type:%s in mpeer queue is redirected, numOfEps:%d inUse:%d", pMsg, pMsg->rpcMsg.ahandle,
taosMsg[pMsg->rpcMsg.msgType], epSet->numOfEps, epSet->inUse);
return TSDB_CODE_RPC_REDIRECT;
}
......
......@@ -34,7 +34,6 @@
#define QUERY_ID_SIZE 20
#define QUERY_STREAM_SAVE_SIZE 20
extern void *tsMnodeTmr;
static SCacheObj *tsMnodeConnCache = NULL;
static int32_t tsConnIndex = 0;
......
......@@ -51,21 +51,12 @@ int32_t mnodeProcessRead(SMnodeMsg *pMsg) {
SMnodeRsp *rpcRsp = &pMsg->rpcRsp;
SRpcEpSet *epSet = rpcMallocCont(sizeof(SRpcEpSet));
mnodeGetMnodeEpSetForShell(epSet);
mDebug("msg:%p, app:%p type:%s in mread queue will be redirected, numOfEps:%d inUse:%d", pMsg, pMsg->rpcMsg.ahandle,
taosMsg[pMsg->rpcMsg.msgType], epSet->numOfEps, epSet->inUse);
for (int32_t i = 0; i < epSet->numOfEps; ++i) {
if (strcmp(epSet->fqdn[i], tsLocalFqdn) == 0 && htons(epSet->port[i]) == tsServerPort) {
epSet->inUse = (i + 1) % epSet->numOfEps;
mDebug("mnode index:%d ep:%s:%u, set inUse to %d", i, epSet->fqdn[i], htons(epSet->port[i]), epSet->inUse);
} else {
mDebug("mnode index:%d ep:%s:%u", i, epSet->fqdn[i], htons(epSet->port[i]));
}
}
rpcRsp->rsp = epSet;
rpcRsp->len = sizeof(SRpcEpSet);
mDebug("msg:%p, app:%p type:%s in mread queue is redirected, numOfEps:%d inUse:%d", pMsg, pMsg->rpcMsg.ahandle,
taosMsg[pMsg->rpcMsg.msgType], epSet->numOfEps, epSet->inUse);
return TSDB_CODE_RPC_REDIRECT;
}
......
......@@ -18,6 +18,7 @@
#include "taoserror.h"
#include "hash.h"
#include "tutil.h"
#include "tref.h"
#include "tbalance.h"
#include "tqueue.h"
#include "twal.h"
......@@ -98,6 +99,7 @@ typedef struct {
SSdbWorker *worker;
} SSdbWorkerPool;
int32_t tsSdbRid;
extern void * tsMnodeTmr;
static void * tsSdbTmr;
static SSdbMgmt tsSdbMgmt = {0};
......@@ -118,6 +120,7 @@ static void sdbFreeQueue();
static int32_t sdbInsertHash(SSdbTable *pTable, SSdbRow *pRow);
static int32_t sdbUpdateHash(SSdbTable *pTable, SSdbRow *pRow);
static int32_t sdbDeleteHash(SSdbTable *pTable, SSdbRow *pRow);
static void sdbCloseTableObj(void *handle);
int32_t sdbGetId(void *pTable) {
return ((SSdbTable *)pTable)->autoIndex;
......@@ -248,6 +251,16 @@ static void sdbNotifyRole(int32_t vgId, int8_t role) {
sdbUpdateMnodeRoles();
}
static int32_t sdbNotifyFileSynced(int32_t vgId, uint64_t fversion) { return 0; }
static void sdbNotifyFlowCtrl(int32_t vgId, int32_t level) {}
static int32_t sdbGetSyncVersion(int32_t vgId, uint64_t *fver, uint64_t *vver) {
*fver = 0;
*vver = 0;
return 0;
}
// failed to forward, need revert insert
static void sdbHandleFailedConfirm(SSdbRow *pRow) {
SWalHead *pHead = pRow->pHead;
......@@ -369,11 +382,14 @@ void sdbUpdateSync(void *pMnodes) {
syncInfo.version = sdbGetVersion();
syncInfo.syncCfg = syncCfg;
sprintf(syncInfo.path, "%s", tsMnodeDir);
syncInfo.getWalInfo = sdbGetWalInfo;
syncInfo.getFileInfo = sdbGetFileInfo;
syncInfo.getWalInfo = sdbGetWalInfo;
syncInfo.writeToCache = sdbWriteFwdToQueue;
syncInfo.confirmForward = sdbConfirmForward;
syncInfo.notifyRole = sdbNotifyRole;
syncInfo.notifyFileSynced = sdbNotifyFileSynced;
syncInfo.notifyFlowCtrl = sdbNotifyFlowCtrl;
syncInfo.getVersion = sdbGetSyncVersion;
tsSdbMgmt.cfg = syncCfg;
if (tsSdbMgmt.sync) {
......@@ -385,6 +401,17 @@ void sdbUpdateSync(void *pMnodes) {
sdbUpdateMnodeRoles();
}
int32_t sdbInitRef() {
tsSdbRid = taosOpenRef(10, sdbCloseTableObj);
if (tsSdbRid <= 0) {
sdbError("failed to init sdb ref");
return -1;
}
return 0;
}
void sdbCleanUpRef() { taosCloseRef(tsSdbRid); }
int32_t sdbInit() {
pthread_mutex_init(&tsSdbMgmt.mutex, NULL);
......@@ -423,7 +450,7 @@ void sdbCleanUp() {
walClose(tsSdbMgmt.wal);
tsSdbMgmt.wal = NULL;
}
pthread_mutex_destroy(&tsSdbMgmt.mutex);
}
......@@ -506,7 +533,7 @@ static int32_t sdbInsertHash(SSdbTable *pTable, SSdbRow *pRow) {
atomic_add_fetch_32(&pTable->autoIndex, 1);
}
sdbDebug("vgId:1, sdb:%s, insert key:%s to hash, rowSize:%d rows:%" PRId64 ", msg:%p", pTable->name,
sdbTrace("vgId:1, sdb:%s, insert key:%s to hash, rowSize:%d rows:%" PRId64 ", msg:%p", pTable->name,
sdbGetRowStr(pTable, pRow->pObj), pRow->rowSize, pTable->numOfRows, pRow->pMsg);
int32_t code = (*pTable->fpInsert)(pRow);
......@@ -542,7 +569,7 @@ static int32_t sdbDeleteHash(SSdbTable *pTable, SSdbRow *pRow) {
atomic_sub_fetch_32(&pTable->numOfRows, 1);
sdbDebug("vgId:1, sdb:%s, delete key:%s from hash, numOfRows:%" PRId64 ", msg:%p", pTable->name,
sdbTrace("vgId:1, sdb:%s, delete key:%s from hash, numOfRows:%" PRId64 ", msg:%p", pTable->name,
sdbGetRowStr(pTable, pRow->pObj), pTable->numOfRows, pRow->pMsg);
sdbDecRef(pTable, pRow->pObj);
......@@ -551,7 +578,7 @@ static int32_t sdbDeleteHash(SSdbTable *pTable, SSdbRow *pRow) {
}
static int32_t sdbUpdateHash(SSdbTable *pTable, SSdbRow *pRow) {
sdbDebug("vgId:1, sdb:%s, update key:%s in hash, numOfRows:%" PRId64 ", msg:%p", pTable->name,
sdbTrace("vgId:1, sdb:%s, update key:%s in hash, numOfRows:%" PRId64 ", msg:%p", pTable->name,
sdbGetRowStr(pTable, pRow->pObj), pTable->numOfRows, pRow->pMsg);
(*pTable->fpUpdate)(pRow);
......@@ -649,7 +676,7 @@ static int32_t sdbProcessWrite(void *wparam, void *hparam, int32_t qtype, void *
return syncCode;
}
sdbDebug("vgId:1, sdb:%s, record from wal/fwd is disposed, action:%s key:%s hver:%" PRIu64, pTable->name,
sdbTrace("vgId:1, sdb:%s, record from %s is disposed, action:%s key:%s hver:%" PRIu64, pTable->name, qtypeStr[qtype],
actStr[action], sdbGetKeyStr(pTable, pHead->cont), pHead->version);
// even it is WAL/FWD, it shall be called to update version in sync
......@@ -801,10 +828,10 @@ void sdbFreeIter(void *tparam, void *pIter) {
taosHashCancelIterate(pTable->iHandle, pIter);
}
void *sdbOpenTable(SSdbTableDesc *pDesc) {
int64_t sdbOpenTable(SSdbTableDesc *pDesc) {
SSdbTable *pTable = (SSdbTable *)calloc(1, sizeof(SSdbTable));
if (pTable == NULL) return NULL;
if (pTable == NULL) return -1;
pthread_mutex_init(&pTable->mutex, NULL);
tstrncpy(pTable->name, pDesc->name, SDB_TABLE_LEN);
......@@ -829,10 +856,21 @@ void *sdbOpenTable(SSdbTableDesc *pDesc) {
tsSdbMgmt.numOfTables++;
tsSdbMgmt.tableList[pTable->id] = pTable;
return pTable;
return taosAddRef(tsSdbRid, pTable);
}
void sdbCloseTable(int64_t rid) {
taosRemoveRef(tsSdbRid, rid);
}
void *sdbGetTableByRid(int64_t rid) {
void *handle = taosAcquireRef(tsSdbRid, rid);
taosReleaseRef(tsSdbRid, rid);
return handle;
}
void sdbCloseTable(void *handle) {
static void sdbCloseTableObj(void *handle) {
SSdbTable *pTable = (SSdbTable *)handle;
if (pTable == NULL) return;
......@@ -855,6 +893,7 @@ void sdbCloseTable(void *handle) {
taosHashCancelIterate(pTable->iHandle, pIter);
taosHashCleanup(pTable->iHandle);
pTable->iHandle = NULL;
pthread_mutex_destroy(&pTable->mutex);
sdbDebug("vgId:1, sdb:%s, is closed, numOfTables:%d", pTable->name, tsSdbMgmt.numOfTables);
......
......@@ -52,7 +52,6 @@ static bool mnodeCheckShowFinished(SShowObj *pShow);
static void *mnodePutShowObj(SShowObj *pShow);
static void mnodeReleaseShowObj(SShowObj *pShow, bool forceRemove);
extern void *tsMnodeTmr;
static void *tsMnodeShowCache = NULL;
static int32_t tsShowObjIndex = 0;
static SShowMetaFp tsMnodeShowMetaFp[TSDB_MGMT_TABLE_MAX] = {0};
......
......@@ -49,7 +49,9 @@
#define CREATE_CTABLE_RETRY_TIMES 10
#define CREATE_CTABLE_RETRY_SEC 14
int64_t tsCTableRid = -1;
static void * tsChildTableSdb;
int64_t tsSTableRid = -1;
static void * tsSuperTableSdb;
static int32_t tsChildTableUpdateSize;
static int32_t tsSuperTableUpdateSize;
......@@ -350,7 +352,7 @@ static int32_t mnodeInitChildTables() {
SCTableObj tObj;
tsChildTableUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj.info.type;
SSdbTableDesc tableDesc = {
SSdbTableDesc desc = {
.id = SDB_TABLE_CTABLE,
.name = "ctables",
.hashSessions = TSDB_DEFAULT_CTABLES_HASH_SIZE,
......@@ -366,7 +368,8 @@ static int32_t mnodeInitChildTables() {
.fpRestored = mnodeChildTableActionRestored
};
tsChildTableSdb = sdbOpenTable(&tableDesc);
tsCTableRid = sdbOpenTable(&desc);
tsChildTableSdb = sdbGetTableByRid(tsCTableRid);
if (tsChildTableSdb == NULL) {
mError("failed to init child table data");
return -1;
......@@ -377,7 +380,7 @@ static int32_t mnodeInitChildTables() {
}
static void mnodeCleanupChildTables() {
sdbCloseTable(tsChildTableSdb);
sdbCloseTable(tsCTableRid);
tsChildTableSdb = NULL;
}
......@@ -543,7 +546,7 @@ static int32_t mnodeInitSuperTables() {
SSTableObj tObj;
tsSuperTableUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj.info.type;
SSdbTableDesc tableDesc = {
SSdbTableDesc desc = {
.id = SDB_TABLE_STABLE,
.name = "stables",
.hashSessions = TSDB_DEFAULT_STABLES_HASH_SIZE,
......@@ -559,7 +562,8 @@ static int32_t mnodeInitSuperTables() {
.fpRestored = mnodeSuperTableActionRestored
};
tsSuperTableSdb = sdbOpenTable(&tableDesc);
tsSTableRid = sdbOpenTable(&desc);
tsSuperTableSdb = sdbGetTableByRid(tsSTableRid);
if (tsSuperTableSdb == NULL) {
mError("failed to init stables data");
return -1;
......@@ -570,7 +574,7 @@ static int32_t mnodeInitSuperTables() {
}
static void mnodeCleanupSuperTables() {
sdbCloseTable(tsSuperTableSdb);
sdbCloseTable(tsSTableRid);
tsSuperTableSdb = NULL;
}
......
......@@ -33,6 +33,7 @@
#include "mnodeWrite.h"
#include "mnodePeer.h"
int64_t tsUserRid = -1;
static void * tsUserSdb = NULL;
static int32_t tsUserUpdateSize = 0;
static int32_t mnodeGetUserMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
......@@ -165,7 +166,8 @@ int32_t mnodeInitUsers() {
.fpRestored = mnodeUserActionRestored
};
tsUserSdb = sdbOpenTable(&desc);
tsUserRid = sdbOpenTable(&desc);
tsUserSdb = sdbGetTableByRid(tsUserRid);
if (tsUserSdb == NULL) {
mError("table:%s, failed to create hash", desc.name);
return -1;
......@@ -185,7 +187,7 @@ int32_t mnodeInitUsers() {
}
void mnodeCleanupUsers() {
sdbCloseTable(tsUserSdb);
sdbCloseTable(tsUserRid);
tsUserSdb = NULL;
}
......
......@@ -51,6 +51,7 @@ char* vgroupStatus[] = {
"updating"
};
int64_t tsVgroupRid = -1;
static void *tsVgroupSdb = NULL;
static int32_t tsVgUpdateSize = 0;
......@@ -222,7 +223,8 @@ int32_t mnodeInitVgroups() {
.fpRestored = mnodeVgroupActionRestored,
};
tsVgroupSdb = sdbOpenTable(&desc);
tsVgroupRid = sdbOpenTable(&desc);
tsVgroupSdb = sdbGetTableByRid(tsVgroupRid);
if (tsVgroupSdb == NULL) {
mError("failed to init vgroups data");
return -1;
......@@ -610,7 +612,7 @@ void mnodeDropVgroup(SVgObj *pVgroup, void *ahandle) {
}
void mnodeCleanupVgroups() {
sdbCloseTable(tsVgroupSdb);
sdbCloseTable(tsVgroupRid);
tsVgroupSdb = NULL;
}
......
......@@ -54,18 +54,8 @@ int32_t mnodeProcessWrite(SMnodeMsg *pMsg) {
rpcRsp->rsp = epSet;
rpcRsp->len = sizeof(SRpcEpSet);
mDebug("msg:%p, app:%p type:%s in write queue, will be redirected, numOfEps:%d inUse:%d", pMsg, pMsg->rpcMsg.ahandle,
taosMsg[pMsg->rpcMsg.msgType], epSet->numOfEps, epSet->inUse);
for (int32_t i = 0; i < epSet->numOfEps; ++i) {
if (strcmp(epSet->fqdn[i], tsLocalFqdn) == 0 && htons(epSet->port[i]) == tsServerPort) {
epSet->inUse = (i + 1) % epSet->numOfEps;
mDebug("msg:%p, app:%p mnode index:%d ep:%s:%d, set inUse to %d", pMsg, pMsg->rpcMsg.ahandle, i, epSet->fqdn[i],
htons(epSet->port[i]), epSet->inUse);
} else {
mDebug("msg:%p, app:%p mnode index:%d ep:%s:%d", pMsg, pMsg->rpcMsg.ahandle, i, epSet->fqdn[i],
htons(epSet->port[i]));
}
}
mDebug("msg:%p, app:%p type:%s in write queue, is redirected, numOfEps:%d inUse:%d", pMsg,
pMsg->rpcMsg.ahandle, taosMsg[pMsg->rpcMsg.msgType], epSet->numOfEps, epSet->inUse);
return TSDB_CODE_RPC_REDIRECT;
}
......
......@@ -630,8 +630,16 @@ static void rpcReleaseConn(SRpcConn *pConn) {
} else {
// if there is an outgoing message, free it
if (pConn->outType && pConn->pReqMsg) {
if (pConn->pContext) pConn->pContext->pConn = NULL;
taosRemoveRef(tsRpcRefId, pConn->pContext->rid);
SRpcReqContext *pContext = pConn->pContext;
if (pContext->pRsp) {
// for synchronous API, post semaphore to unblock app
pContext->pRsp->code = TSDB_CODE_RPC_APP_ERROR;
pContext->pRsp->pCont = NULL;
pContext->pRsp->contLen = 0;
tsem_post(pContext->pSem);
}
pContext->pConn = NULL;
taosRemoveRef(tsRpcRefId, pContext->rid);
}
}
......
......@@ -139,16 +139,14 @@ typedef struct SsyncPeer {
char id[TSDB_EP_LEN + 32]; // peer vgId + end point
uint64_t version;
uint64_t sversion; // track the peer version in retrieve process
uint64_t lastVer; // track the file version while retrieve
uint64_t lastFileVer; // track the file version while retrieve
uint64_t lastWalVer; // track the wal version while retrieve
int32_t syncFd;
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
void * timer;
void * pConn;
int32_t notifyFd;
int32_t watchNum;
int32_t *watchFd;
int32_t refCount; // reference count
struct SSyncNode *pSyncNode;
} SSyncPeer;
......@@ -173,7 +171,7 @@ typedef struct SSyncNode {
FNotifyRole notifyRole;
FNotifyFlowCtrl notifyFlowCtrl;
FNotifyFileSynced notifyFileSynced;
FGetFileVersion getFileVersion;
FGetVersion getVersion;
pthread_mutex_t mutex;
} SSyncNode;
......
......@@ -196,7 +196,7 @@ int64_t syncStart(const SSyncInfo *pInfo) {
pNode->confirmForward = pInfo->confirmForward;
pNode->notifyFlowCtrl = pInfo->notifyFlowCtrl;
pNode->notifyFileSynced = pInfo->notifyFileSynced;
pNode->getFileVersion = pInfo->getFileVersion;
pNode->getVersion = pInfo->getVersion;
pNode->selfIndex = -1;
pNode->vgId = pInfo->vgId;
......@@ -498,7 +498,6 @@ int32_t syncDecPeerRef(SSyncPeer *pPeer) {
taosReleaseRef(tsSyncRefId, pPeer->pSyncNode->rid);
sDebug("%s, resource is freed", pPeer->id);
tfree(pPeer->watchFd);
tfree(pPeer);
return 0;
}
......
......@@ -26,39 +26,78 @@
#include "tsync.h"
#include "syncInt.h"
static int32_t syncAreFilesModified(SSyncNode *pNode, SSyncPeer *pPeer) {
if (pNode->getFileVersion == NULL) return TSDB_CODE_SUCCESS;
static int32_t syncGetWalVersion(SSyncNode *pNode, SSyncPeer *pPeer) {
uint64_t fver, wver;
int32_t code = (*pNode->getVersion)(pNode->vgId, &fver, &wver);
if (code != 0) {
sDebug("%s, vnode is commiting while retrieve, last wver:%" PRIu64, pPeer->id, pPeer->lastWalVer);
return -1;
}
pPeer->lastWalVer = wver;
return code;
}
static bool syncIsWalModified(SSyncNode *pNode, SSyncPeer *pPeer) {
uint64_t fver, wver;
int32_t code = (*pNode->getVersion)(pNode->vgId, &fver, &wver);
if (code != 0) {
sDebug("%s, vnode is commiting while retrieve, last wver:%" PRIu64, pPeer->id, pPeer->lastWalVer);
return true;
}
if (wver != pPeer->lastWalVer) {
sDebug("%s, wal is modified while retrieve, wver:%" PRIu64 ", last:%" PRIu64, pPeer->id, wver, pPeer->lastWalVer);
return true;
}
return false;
}
uint64_t fver = 0;
int32_t code = (*pNode->getFileVersion)(pNode->vgId, &fver);
static int32_t syncGetFileVersion(SSyncNode *pNode, SSyncPeer *pPeer) {
uint64_t fver, wver;
int32_t code = (*pNode->getVersion)(pNode->vgId, &fver, &wver);
if (code != 0) {
sInfo("%s, vnode is commiting while retrieve, last fver:%" PRIu64, pPeer->id, pPeer->lastVer);
sDebug("%s, vnode is commiting while retrieve, last fver:%" PRIu64, pPeer->id, pPeer->lastFileVer);
return -1;
}
pPeer->lastFileVer = fver;
return code;
}
static bool syncAreFilesModified(SSyncNode *pNode, SSyncPeer *pPeer) {
uint64_t fver, wver;
int32_t code = (*pNode->getVersion)(pNode->vgId, &fver, &wver);
if (code != 0) {
sDebug("%s, vnode is commiting while retrieve, last fver:%" PRIu64, pPeer->id, pPeer->lastFileVer);
pPeer->fileChanged = 1;
return TSDB_CODE_SYN_VND_COMMITING;
return true;
}
if (fver != pPeer->lastVer) {
sInfo("%s, files are modified while retrieve, fver:%" PRIu64 ", last fver:%" PRIu64, pPeer->id, fver, pPeer->lastVer);
if (fver != pPeer->lastFileVer) {
sDebug("%s, files are modified while retrieve, fver:%" PRIu64 ", last:%" PRIu64, pPeer->id, fver, pPeer->lastFileVer);
pPeer->fileChanged = 1;
return TSDB_CODE_SYN_FILE_CHNAGED;
return true;
}
pPeer->fileChanged = 0;
return TSDB_CODE_SUCCESS;
return false;
}
static int32_t syncRetrieveFile(SSyncPeer *pPeer) {
SSyncNode *pNode = pPeer->pSyncNode;
SFileInfo fileInfo; memset(&fileInfo, 0, sizeof(SFileInfo));
SFileAck fileAck = {0};
int32_t code = TSDB_CODE_SYN_APP_ERROR;
int32_t code = -1;
char name[TSDB_FILENAME_LEN * 2] = {0};
if (pNode->getFileVersion) (*pNode->getFileVersion)(pNode->vgId, &pPeer->lastVer);
if (syncGetFileVersion(pNode, pPeer) < 0) return -1;
while (1) {
// retrieve file info
fileInfo.name[0] = 0;
fileInfo.size = 0;
fileInfo.magic = (*pNode->getFileInfo)(pNode->vgId, fileInfo.name, &fileInfo.index, TAOS_SYNC_MAX_INDEX,
&fileInfo.size, &fileInfo.fversion);
// fileInfo.size = htonl(size);
......@@ -67,14 +106,14 @@ static int32_t syncRetrieveFile(SSyncPeer *pPeer) {
// send the file info
int32_t ret = taosWriteMsg(pPeer->syncFd, &(fileInfo), sizeof(fileInfo));
if (ret < 0) {
code = TAOS_SYSTEM_ERROR(errno);
code = -1;
sError("%s, failed to write file:%s info while retrieve file since %s", pPeer->id, fileInfo.name, strerror(errno));
break;
}
// if no file anymore, break
if (fileInfo.magic == 0 || fileInfo.name[0] == 0) {
code = TSDB_CODE_SUCCESS;
code = 0;
sDebug("%s, no more files to sync", pPeer->id);
break;
}
......@@ -82,7 +121,7 @@ static int32_t syncRetrieveFile(SSyncPeer *pPeer) {
// wait for the ack from peer
ret = taosReadMsg(pPeer->syncFd, &fileAck, sizeof(fileAck));
if (ret < 0) {
code = TAOS_SYSTEM_ERROR(errno);
code = -1;
sError("%s, failed to read file:%s ack while retrieve file since %s", pPeer->id, fileInfo.name, strerror(errno));
break;
}
......@@ -103,7 +142,7 @@ static int32_t syncRetrieveFile(SSyncPeer *pPeer) {
// send the file to peer
int32_t sfd = open(name, O_RDONLY);
if (sfd < 0) {
code = TAOS_SYSTEM_ERROR(errno);
code = -1;
sError("%s, failed to open file:%s while retrieve file since %s", pPeer->id, fileInfo.name, strerror(errno));
break;
}
......@@ -111,7 +150,7 @@ static int32_t syncRetrieveFile(SSyncPeer *pPeer) {
ret = taosSendFile(pPeer->syncFd, sfd, NULL, fileInfo.size);
close(sfd);
if (ret < 0) {
code = TAOS_SYSTEM_ERROR(errno);
code = -1;
sError("%s, failed to send file:%s while retrieve file since %s", pPeer->id, fileInfo.name, strerror(errno));
break;
}
......@@ -120,128 +159,103 @@ static int32_t syncRetrieveFile(SSyncPeer *pPeer) {
fileInfo.index++;
// check if processed files are modified
code = syncAreFilesModified(pNode, pPeer);
if (code != TSDB_CODE_SUCCESS) break;
if (syncAreFilesModified(pNode, pPeer)) {
code = -1;
break;
}
}
if (code != TSDB_CODE_SUCCESS) {
sError("%s, failed to retrieve file since %s", pPeer->id, tstrerror(code));
sError("%s, failed to retrieve file, code:0x%x", pPeer->id, code);
}
return code;
}
/* if only a partial record is read out, set the IN_MODIFY flag in event,
so upper layer will reload the file to get a complete record */
static int32_t syncReadOneWalRecord(int32_t sfd, SWalHead *pHead, uint32_t *pEvent) {
int32_t ret;
// if only a partial record is read out, upper layer will reload the file to get a complete record
static int32_t syncReadOneWalRecord(int32_t sfd, SWalHead *pHead) {
int32_t ret = read(sfd, pHead, sizeof(SWalHead));
if (ret < 0) {
sError("sfd:%d, failed to read wal head since %s, ret:%d", sfd, strerror(errno), ret);
return -1;
}
ret = read(sfd, pHead, sizeof(SWalHead));
if (ret < 0) return -1;
if (ret == 0) return 0;
if (ret == 0) {
sTrace("sfd:%d, read to the end of file, ret:%d", sfd, ret);
return 0;
}
if (ret != sizeof(SWalHead)) {
// file is not at end yet, it shall be reloaded
*pEvent = *pEvent | IN_MODIFY;
sDebug("sfd:%d, a partial wal head is read out, ret:%d", sfd, ret);
return 0;
}
assert(pHead->len <= TSDB_MAX_WAL_SIZE);
ret = read(sfd, pHead->cont, pHead->len);
if (ret < 0) return -1;
if (ret < 0) {
sError("sfd:%d, failed to read wal content since %s, ret:%d", sfd, strerror(errno), ret);
return -1;
}
if (ret != pHead->len) {
// file is not at end yet, it shall be reloaded
*pEvent = *pEvent | IN_MODIFY;
sDebug("sfd:%d, a partial wal conetnt is read out, ret:%d", sfd, ret);
return 0;
}
return sizeof(SWalHead) + pHead->len;
}
static int32_t syncMonitorLastWal(SSyncPeer *pPeer, char *name) {
pPeer->watchNum = 0;
taosClose(pPeer->notifyFd);
pPeer->notifyFd = inotify_init1(IN_NONBLOCK);
if (pPeer->notifyFd < 0) {
sError("%s, failed to init inotify since %s", pPeer->id, strerror(errno));
return -1;
}
if (pPeer->watchFd == NULL) pPeer->watchFd = malloc(sizeof(int32_t) * tsMaxWatchFiles);
if (pPeer->watchFd == NULL) {
sError("%s, failed to allocate watchFd", pPeer->id);
return -1;
}
memset(pPeer->watchFd, -1, sizeof(int32_t) * tsMaxWatchFiles);
int32_t *wd = pPeer->watchFd;
*wd = inotify_add_watch(pPeer->notifyFd, name, IN_MODIFY | IN_CLOSE_WRITE);
if (*wd == -1) {
sError("%s, failed to watch last wal since %s", pPeer->id, strerror(errno));
static int32_t syncRetrieveLastWal(SSyncPeer *pPeer, char *name, uint64_t fversion, int64_t offset) {
int32_t sfd = open(name, O_RDONLY);
if (sfd < 0) {
sError("%s, failed to open wal:%s for retrieve since:%s", pPeer->id, name, tstrerror(errno));
return -1;
}
return 0;
}
static int32_t syncCheckLastWalChanges(SSyncPeer *pPeer, uint32_t *pEvent) {
char buf[2048];
int32_t len = read(pPeer->notifyFd, buf, sizeof(buf));
if (len < 0 && errno != EAGAIN) {
sError("%s, failed to read notify FD since %s", pPeer->id, strerror(errno));
int32_t code = taosLSeek(sfd, offset, SEEK_SET);
if (code < 0) {
sError("%s, failed to seek %" PRId64 " in wal:%s for retrieve since:%s", pPeer->id, offset, name, tstrerror(errno));
close(sfd);
return -1;
}
if (len == 0) return 0;
struct inotify_event *event;
for (char *ptr = buf; ptr < buf + len; ptr += sizeof(struct inotify_event) + event->len) {
event = (struct inotify_event *)ptr;
if (event->mask & IN_MODIFY) *pEvent = *pEvent | IN_MODIFY;
if (event->mask & IN_CLOSE_WRITE) *pEvent = *pEvent | IN_CLOSE_WRITE;
}
if (pEvent != 0) sDebug("%s, last wal event:0x%x", pPeer->id, *pEvent);
return 0;
}
sDebug("%s, retrieve last wal:%s, offset:%" PRId64 " fver:%" PRIu64, pPeer->id, name, offset, fversion);
static int32_t syncRetrieveLastWal(SSyncPeer *pPeer, char *name, uint64_t fversion, int64_t offset, uint32_t *pEvent) {
SWalHead *pHead = malloc(SYNC_MAX_SIZE);
int32_t code = -1;
int32_t bytes = 0;
int32_t sfd;
sfd = open(name, O_RDONLY);
if (sfd < 0) {
free(pHead);
return -1;
}
(void)lseek(sfd, offset, SEEK_SET);
sDebug("%s, retrieve last wal, offset:%" PRId64 " fver:%" PRIu64, pPeer->id, offset, fversion);
while (1) {
int32_t wsize = syncReadOneWalRecord(sfd, pHead, pEvent);
if (wsize < 0) break;
if (wsize == 0) {
code = 0;
code = syncReadOneWalRecord(sfd, pHead);
if (code < 0) {
sError("%s, failed to read one record from wal:%s", pPeer->id, name);
break;
}
if (code == 0) {
code = bytes;
sDebug("%s, read to the end of wal, bytes:%d", pPeer->id, bytes);
break;
}
sTrace("%s, last wal is forwarded, hver:%" PRIu64, pPeer->id, pHead->version);
int32_t wsize = code;
int32_t ret = taosWriteMsg(pPeer->syncFd, pHead, wsize);
if (ret != wsize) break;
pPeer->sversion = pHead->version;
if (ret != wsize) {
code = -1;
sError("%s, failed to forward wal since %s, hver:%" PRIu64, pPeer->id, strerror(errno), pHead->version);
break;
}
pPeer->sversion = pHead->version;
bytes += wsize;
if (pHead->version >= fversion && fversion > 0) {
code = 0;
bytes = 0;
sDebug("%s, retrieve wal finished, hver:%" PRIu64 " fver:%" PRIu64, pPeer->id, pHead->version, fversion);
break;
}
}
......@@ -249,92 +263,62 @@ static int32_t syncRetrieveLastWal(SSyncPeer *pPeer, char *name, uint64_t fversi
free(pHead);
close(sfd);
if (code == 0) return bytes;
return -1;
return code;
}
static int32_t syncProcessLastWal(SSyncPeer *pPeer, char *wname, int64_t index) {
SSyncNode *pNode = pPeer->pSyncNode;
int32_t code = -1;
int32_t once = 0; // last WAL has once ever been processed
int64_t offset = 0;
uint64_t fversion = 0;
char fname[TSDB_FILENAME_LEN * 2] = {0}; // full path to wal file
if (syncAreFilesModified(pNode, pPeer) != 0) return -1;
// get full path to wal file
snprintf(fname, sizeof(fname), "%s/%s", pNode->path, wname);
sDebug("%s, start to retrieve last wal:%s", pPeer->id, fname);
while (1) {
int32_t once = 0; // last WAL has once ever been processed
int64_t offset = 0;
uint64_t fversion = 0;
uint32_t event = 0;
// get full path to wal file
snprintf(fname, sizeof(fname), "%s/%s", pNode->path, wname);
sDebug("%s, start to retrieve last wal:%s", pPeer->id, fname);
// monitor last wal
if (syncMonitorLastWal(pPeer, fname) < 0) break;
while (1) {
int32_t bytes = syncRetrieveLastWal(pPeer, fname, fversion, offset, &event);
if (bytes < 0) break;
if (syncAreFilesModified(pNode, pPeer)) return -1;
if (syncGetWalVersion(pNode, pPeer) < 0) return -1;
// check file changes
if (syncCheckLastWalChanges(pPeer, &event) < 0) break;
// if file is not updated or updated once, set the fversion and sstatus
if (((event & IN_MODIFY) == 0) || once) {
if (fversion == 0) {
pPeer->sstatus = TAOS_SYNC_STATUS_CACHE; // start to forward pkt
sDebug("%s, fversion is 0 then set sstatus:%s", pPeer->id, syncStatus[pPeer->sstatus]);
fversion = nodeVersion; // must read data to fversion
}
}
int32_t bytes = syncRetrieveLastWal(pPeer, fname, fversion, offset);
if (bytes < 0) {
sDebug("%s, failed to retrieve last wal", pPeer->id);
return bytes;
}
// if all data up to fversion is read out, it is over
if (pPeer->sversion >= fversion && fversion > 0) {
code = 0;
sDebug("%s, data up to fver:%" PRIu64 " has been read out, bytes:%d", pPeer->id, fversion, bytes);
break;
}
// check file changes
bool walModified = syncIsWalModified(pNode, pPeer);
// if all data are read out, and no update
if ((bytes == 0) && ((event & IN_MODIFY) == 0)) {
// wal file is closed, break
if (event & IN_CLOSE_WRITE) {
code = 0;
sDebug("%s, current wal is closed", pPeer->id);
break;
}
// wal not closed, it means some data not flushed to disk, wait for a while
usleep(10000);
// if file is not updated or updated once, set the fversion and sstatus
if (!walModified || once) {
if (fversion == 0) {
pPeer->sstatus = TAOS_SYNC_STATUS_CACHE; // start to forward pkt
fversion = nodeVersion; // must read data to fversion
sDebug("%s, set sstatus:%s and fver:%" PRIu64, pPeer->id, syncStatus[pPeer->sstatus], fversion);
}
// if bytes>0, file is updated, or fversion is not reached but file still open, read again
once = 1;
offset += bytes;
sDebug("%s, retrieve last wal, bytes:%d", pPeer->id, bytes);
event = event & (~IN_MODIFY); // clear IN_MODIFY flag
}
if (code < 0) break;
if (pPeer->sversion >= fversion && fversion > 0) break;
// if all data up to fversion is read out, it is over
if (pPeer->sversion >= fversion && fversion > 0) {
sDebug("%s, data up to fver:%" PRIu64 " has been read out, bytes:%d sver:%" PRIu64, pPeer->id, fversion, bytes,
pPeer->sversion);
return 0;
}
index++;
wname[0] = 0;
code = (*pNode->getWalInfo)(pNode->vgId, wname, &index);
if (code < 0) break;
if (wname[0] == 0) {
code = 0;
break;
// if all data are read out, and no update
if (bytes == 0 && !walModified) {
// wal not closed, it means some data not flushed to disk, wait for a while
usleep(10000);
}
// current last wal is closed, there is a new one
sDebug("%s, last wal is closed, try new one", pPeer->id);
// if bytes > 0, file is updated, or fversion is not reached but file still open, read again
once = 1;
offset += bytes;
sDebug("%s, continue retrieve last wal, bytes:%d offset:%" PRId64, pPeer->id, bytes, offset);
}
taosClose(pPeer->notifyFd);
return code;
return -1;
}
static int32_t syncRetrieveWal(SSyncPeer *pPeer) {
......@@ -342,7 +326,6 @@ static int32_t syncRetrieveWal(SSyncPeer *pPeer) {
char fname[TSDB_FILENAME_LEN * 3];
char wname[TSDB_FILENAME_LEN * 2];
int32_t size;
struct stat fstat;
int32_t code = -1;
int64_t index = 0;
......@@ -350,9 +333,14 @@ static int32_t syncRetrieveWal(SSyncPeer *pPeer) {
// retrieve wal info
wname[0] = 0;
code = (*pNode->getWalInfo)(pNode->vgId, wname, &index);
if (code < 0) break; // error
if (code < 0) {
sError("%s, failed to get wal info since:%s, code:0x%x", pPeer->id, strerror(errno), code);
break;
}
if (wname[0] == 0) { // no wal file
sDebug("%s, no wal file", pPeer->id);
code = 0;
sDebug("%s, no wal file anymore", pPeer->id);
break;
}
......@@ -364,20 +352,35 @@ static int32_t syncRetrieveWal(SSyncPeer *pPeer) {
// get the full path to wal file
snprintf(fname, sizeof(fname), "%s/%s", pNode->path, wname);
// send wal file,
// inotify is not required, old wal file won't be modified, even remove is ok
if (stat(fname, &fstat) < 0) break;
size = fstat.st_size;
// send wal file, old wal file won't be modified, even remove is ok
struct stat fstat;
if (stat(fname, &fstat) < 0) {
code = -1;
sDebug("%s, failed to stat wal:%s for retrieve since %s, code:0x%x", pPeer->id, fname, strerror(errno), code);
break;
}
size = fstat.st_size;
sDebug("%s, retrieve wal:%s size:%d", pPeer->id, fname, size);
int32_t sfd = open(fname, O_RDONLY);
if (sfd < 0) break;
if (sfd < 0) {
code = -1;
sError("%s, failed to open wal:%s for retrieve since %s, code:0x%x", pPeer->id, fname, strerror(errno), code);
break;
}
code = taosSendFile(pPeer->syncFd, sfd, NULL, size);
close(sfd);
if (code < 0) break;
if (code < 0) {
sError("%s, failed to send wal:%s for retrieve since %s, code:0x%x", pPeer->id, fname, strerror(errno), code);
break;
}
if (syncAreFilesModified(pNode, pPeer) != 0) break;
if (syncAreFilesModified(pNode, pPeer)) {
code = -1;
break;
}
}
if (code == 0) {
......@@ -386,9 +389,9 @@ static int32_t syncRetrieveWal(SSyncPeer *pPeer) {
SWalHead walHead;
memset(&walHead, 0, sizeof(walHead));
code = taosWriteMsg(pPeer->syncFd, &walHead, sizeof(walHead));
taosWriteMsg(pPeer->syncFd, &walHead, sizeof(walHead));
} else {
sError("%s, failed to send wal since %s", pPeer->id, strerror(errno));
sError("%s, failed to send wal since %s, code:0x%x", pPeer->id, strerror(errno), code);
}
return code;
......@@ -428,7 +431,7 @@ static int32_t syncRetrieveDataStepByStep(SSyncPeer *pPeer) {
pPeer->sversion = 0;
pPeer->sstatus = TAOS_SYNC_STATUS_FILE;
sInfo("%s, start to retrieve files, set sstatus:%s", pPeer->id, syncStatus[pPeer->sstatus]);
if (syncRetrieveFile(pPeer) < 0) {
if (syncRetrieveFile(pPeer) != 0) {
sError("%s, failed to retrieve files", pPeer->id);
return -1;
}
......@@ -437,8 +440,9 @@ static int32_t syncRetrieveDataStepByStep(SSyncPeer *pPeer) {
if (pPeer->sversion == 0) pPeer->sversion = 1;
sInfo("%s, start to retrieve wals", pPeer->id);
if (syncRetrieveWal(pPeer) < 0) {
sError("%s, failed to retrieve wals", pPeer->id);
int32_t code = syncRetrieveWal(pPeer);
if (code != 0) {
sError("%s, failed to retrieve wals, code:0x%x", pPeer->id, code);
return -1;
}
......@@ -474,7 +478,6 @@ void *syncRetrieveData(void *param) {
}
pPeer->fileChanged = 0;
taosClose(pPeer->notifyFd);
taosClose(pPeer->syncFd);
syncDecPeerRef(pPeer);
......
......@@ -917,6 +917,8 @@ static int tsdbInsertSuperBlock(SRWHelper *pHelper, SCompBlock *pCompBlock, int
ASSERT(pHelper->pCompInfo->blocks[0].keyLast < pHelper->pCompInfo->blocks[1].keyFirst);
}
ASSERT((blkIdx == pIdx->numOfBlocks -1) || (!pCompBlock->last));
tsdbDebug("vgId:%d tid:%d a super block is inserted at index %d", REPO_ID(pHelper->pRepo), pHelper->tableInfo.tid,
blkIdx);
......@@ -1042,6 +1044,8 @@ static int tsdbUpdateSuperBlock(SRWHelper *pHelper, SCompBlock *pCompBlock, int
pIdx->maxKey = blockAtIdx(pHelper, pIdx->numOfBlocks - 1)->keyLast;
pIdx->hasLast = (uint32_t)blockAtIdx(pHelper, pIdx->numOfBlocks - 1)->last;
ASSERT((blkIdx == pIdx->numOfBlocks-1) || (!pCompBlock->last));
tsdbDebug("vgId:%d tid:%d a super block is updated at index %d", REPO_ID(pHelper->pRepo), pHelper->tableInfo.tid,
blkIdx);
......@@ -1622,11 +1626,7 @@ static int tsdbProcessMergeCommit(SRWHelper *pHelper, SCommitIter *pCommitIter,
pCfg->update);
if (pDataCols->numOfRows == 0) break;
if (tblkIdx == pIdx->numOfBlocks - 1) {
if (tsdbWriteBlockToProperFile(pHelper, pDataCols, &compBlock) < 0) return -1;
} else {
if (tsdbWriteBlockToFile(pHelper, helperDataF(pHelper), pDataCols, &compBlock, false, true) < 0) return -1;
}
if (tsdbWriteBlockToFile(pHelper, helperDataF(pHelper), pDataCols, &compBlock, false, true) < 0) return -1;
if (round == 0) {
if (oBlock.last && pHelper->hasOldLastBlock) pHelper->hasOldLastBlock = false;
......
......@@ -93,17 +93,23 @@ int64_t tfRead(int64_t tfd, void *buf, int64_t count) {
return ret;
}
int64_t tfFsync(int64_t tfd) {
int32_t tfFsync(int64_t tfd) {
void *p = taosAcquireRef(tsFileRsetId, tfd);
if (p == NULL) return -1;
int32_t fd = (int32_t)(uintptr_t)p;
return fsync(fd);
int32_t code = fsync(fd);
taosReleaseRef(tsFileRsetId, tfd);
return code;
}
bool tfValid(int64_t tfd) {
void *p = taosAcquireRef(tsFileRsetId, tfd);
return p != NULL;
if (p == NULL) return false;
taosReleaseRef(tsFileRsetId, tfd);
return true;
}
int64_t tfLseek(int64_t tfd, int64_t offset, int32_t whence) {
......@@ -111,7 +117,10 @@ int64_t tfLseek(int64_t tfd, int64_t offset, int32_t whence) {
if (p == NULL) return -1;
int32_t fd = (int32_t)(uintptr_t)p;
return taosLSeek(fd, offset, whence);
int64_t ret = taosLSeek(fd, offset, whence);
taosReleaseRef(tsFileRsetId, tfd);
return ret;
}
int32_t tfFtruncate(int64_t tfd, int64_t length) {
......@@ -119,5 +128,8 @@ int32_t tfFtruncate(int64_t tfd, int64_t length) {
if (p == NULL) return -1;
int32_t fd = (int32_t)(uintptr_t)p;
return taosFtruncate(fd, length);
int32_t code = taosFtruncate(fd, length);
taosReleaseRef(tsFileRsetId, tfd);
return code;
}
......@@ -38,7 +38,7 @@ static void vnodeCtrlFlow(int32_t vgId, int32_t level);
static int32_t vnodeNotifyFileSynced(int32_t vgId, uint64_t fversion);
static void vnodeConfirmForard(int32_t vgId, void *wparam, int32_t code);
static int32_t vnodeWriteToCache(int32_t vgId, void *wparam, int32_t qtype, void *rparam);
static int32_t vnodeGetFileVersion(int32_t vgId, uint64_t *fver);
static int32_t vnodeGetVersion(int32_t vgId, uint64_t *fver, uint64_t *wver);
#ifndef _SYNC
int64_t syncStart(const SSyncInfo *info) { return NULL; }
......@@ -267,16 +267,18 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
return terrno;
}
SCqCfg cqCfg = {0};
sprintf(cqCfg.user, "_root");
strcpy(cqCfg.pass, tsInternalPass);
strcpy(cqCfg.db, pVnode->db);
cqCfg.vgId = vnode;
cqCfg.cqWrite = vnodeWriteToWQueue;
pVnode->cq = cqOpen(pVnode, &cqCfg);
if (pVnode->cq == NULL) {
vnodeCleanUp(pVnode);
return terrno;
if (tsEnableStream) {
SCqCfg cqCfg = {0};
sprintf(cqCfg.user, "_root");
strcpy(cqCfg.pass, tsInternalPass);
strcpy(cqCfg.db, pVnode->db);
cqCfg.vgId = vnode;
cqCfg.cqWrite = vnodeWriteToCache;
pVnode->cq = cqOpen(pVnode, &cqCfg);
if (pVnode->cq == NULL) {
vnodeCleanUp(pVnode);
return terrno;
}
}
STsdbAppH appH = {0};
......@@ -353,7 +355,7 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
syncInfo.notifyRole = vnodeNotifyRole;
syncInfo.notifyFlowCtrl = vnodeCtrlFlow;
syncInfo.notifyFileSynced = vnodeNotifyFileSynced;
syncInfo.getFileVersion = vnodeGetFileVersion;
syncInfo.getVersion = vnodeGetVersion;
pVnode->sync = syncStart(&syncInfo);
#ifndef _SYNC
......@@ -771,7 +773,7 @@ static int32_t vnodeWriteToCache(int32_t vgId, void *wparam, int32_t qtype, void
return code;
}
static int32_t vnodeGetFileVersion(int32_t vgId, uint64_t *fver) {
static int32_t vnodeGetVersion(int32_t vgId, uint64_t *fver, uint64_t *wver) {
SVnodeObj *pVnode = vnodeAcquire(vgId);
if (pVnode == NULL) {
vError("vgId:%d, vnode not found while write to cache", vgId);
......@@ -780,10 +782,11 @@ static int32_t vnodeGetFileVersion(int32_t vgId, uint64_t *fver) {
int32_t code = 0;
if (pVnode->isCommiting) {
vDebug("vgId:%d, vnode is commiting while get file version", vgId);
vDebug("vgId:%d, vnode is commiting while get version", vgId);
code = -1;
} else {
*fver = pVnode->fversion;
*wver = pVnode->version;
}
vnodeRelease(pVnode);
......
......@@ -56,7 +56,7 @@
<dependency>
<groupId>com.taosdata.jdbc</groupId>
<artifactId>taos-jdbcdriver</artifactId>
<version>2.0.8</version>
<version>2.0.12</version>
</dependency>
<dependency>
<groupId>log4j</groupId>
......
......@@ -6,10 +6,24 @@ TDengine's JDBC demo project is organized in a Maven way so that users can easil
Make sure you have already installed a tdengine client on your current develop environment.
Download the tdengine package on our website: ``https://www.taosdata.com/cn/all-downloads/`` and install the client.
## How to run jdbcChecker
<pre>mvn clean compile exec:java -Dexec.mainClass="com.taosdata.example.JdbcChecker" -Dexec.args="-host localhost"</pre>
## How to run jdbcTaosDemo
run command:
<pre> mvn clean compile exec:java -Dexec.mainClass="com.taosdata.example.jdbcTaosdemo.JdbcTaosdemo"</pre>
and run with your customed args
<pre>mvn clean compile exec:java -Dexec.mainClass="com.taosdata.example.jdbcTaosdemo.JdbcTaosdemo" -Dexec.args="-host localhost"</pre>
## Compile the Demo Code and Run It
To compile the demo project, go to the source directory ``TDengine/tests/examples/JDBC/JDBCDemo`` and execute
<pre>mvn clean package assembly:single</pre>
<pre>
mvn clean package assembly:single
</pre>
The ``pom.xml`` is configured to package all the dependencies into one executable jar file.
To run it, go to ``examples/JDBC/JDBCDemo/target`` and execute
<pre>java -jar jdbcChecker-SNAPSHOT-jar-with-dependencies.jar -host localhost</pre>
\ No newline at end of file
<pre>java -jar jdbcChecker-SNAPSHOT-jar-with-dependencies.jar -host localhost</pre>
......@@ -14,9 +14,9 @@ public final class JdbcTaosdemoConfig {
//Destination database. Default is 'test'
private String dbName = "test";
//keep
private int keep = 3650;
private int keep = 36500;
//days
private int days = 10;
private int days = 120;
//Super table Name. Default is 'meters'
private String stbName = "meters";
......
......@@ -41,7 +41,7 @@ public class InsertTableTask implements Runnable {
long ts = start.toEpochMilli() + (j * timeGap);
// insert data into echo table
for (int i = startTbIndex; i < startTbIndex + tableNumber; i++) {
String sql = SqlSpeller.insertOneRowSQL(config.getDbName(), config.getTbPrefix(), i + 1, ts);
String sql = SqlSpeller.insertBatchSizeRowsSQL(config.getDbName(), config.getTbPrefix(), i + 1, ts, config.getNumberOfRecordsPerRequest());
logger.info(Thread.currentThread().getName() + ">>> " + sql);
Statement statement = connection.createStatement();
statement.execute(sql);
......
......@@ -32,9 +32,9 @@ class TDTestCase:
tdSql.query("show databases")
tdSql.checkData(0, 14, 2)
tdSql.execute("alter database db keep 365")
tdSql.execute("alter database db keep 365,365,365")
tdSql.query("show databases")
tdSql.checkData(0, 7, "3650,3650,365")
tdSql.checkData(0, 7, "365,365,365")
tdSql.execute("alter database db quorum 2")
tdSql.query("show databases")
......
......@@ -78,7 +78,7 @@ class DbConn:
if nRows != 1:
raise taos.error.ProgrammingError(
"Unexpected result for query: {}, rows = {}".format(sql, nRows),
(0x991 if nRows==0 else 0x992)
(CrashGenError.INVALID_EMPTY_RESULT if nRows==0 else CrashGenError.INVALID_MULTIPLE_RESULT)
)
if self.getResultRows() != 1 or self.getResultCols() != 1:
raise RuntimeError("Unexpected result set for query: {}".format(sql))
......@@ -349,7 +349,8 @@ class DbConnNative(DbConn):
def execute(self, sql):
if (not self.isOpen):
raise RuntimeError("Cannot execute database commands until connection is open")
raise CrashGenError(
"Cannot exec SQL unless db connection is open", CrashGenError.DB_CONNECTION_NOT_OPEN)
Logging.debug("[SQL] Executing SQL: {}".format(sql))
self._lastSql = sql
nRows = self._tdSql.execute(sql)
......@@ -360,8 +361,8 @@ class DbConnNative(DbConn):
def query(self, sql): # return rows affected
if (not self.isOpen):
raise RuntimeError(
"Cannot query database until connection is open")
raise CrashGenError(
"Cannot query database until connection is open, restarting?", CrashGenError.DB_CONNECTION_NOT_OPEN)
Logging.debug("[SQL] Executing SQL: {}".format(sql))
self._lastSql = sql
nRows = self._tdSql.query(sql)
......
......@@ -3,14 +3,20 @@ import random
import logging
import os
import taos
class CrashGenError(Exception):
def __init__(self, msg=None, errno=None):
self.msg = msg
self.errno = errno
def __str__(self):
return self.msg
class CrashGenError(taos.error.ProgrammingError):
INVALID_EMPTY_RESULT = 0x991
INVALID_MULTIPLE_RESULT = 0x992
DB_CONNECTION_NOT_OPEN = 0x993
# def __init__(self, msg=None, errno=None):
# self.msg = msg
# self.errno = errno
# def __str__(self):
# return self.msg
pass
class LoggingFilter(logging.Filter):
......@@ -168,6 +174,7 @@ class Progress:
SERVICE_RECONNECT_FAILURE = 6
SERVICE_START_NAP = 7
CREATE_TABLE_ATTEMPT = 8
QUERY_GROUP_BY = 9
tokens = {
STEP_BOUNDARY: '.',
......@@ -178,7 +185,8 @@ class Progress:
SERVICE_RECONNECT_SUCCESS: '.r>',
SERVICE_RECONNECT_FAILURE: '.xr>',
SERVICE_START_NAP: '_zz',
CREATE_TABLE_ATTEMPT: '_c',
CREATE_TABLE_ATTEMPT: 'c',
QUERY_GROUP_BY: 'g',
}
@classmethod
......
......@@ -51,10 +51,12 @@ class TdeInstance():
def prepareGcovEnv(cls, env):
# Ref: https://gcc.gnu.org/onlinedocs/gcc/Cross-profiling.html
bPath = cls._getBuildPath() # build PATH
numSegments = len(bPath.split('/')) - 1 # "/x/TDengine/build" should yield 3
numSegments = numSegments - 1 # DEBUG only
env['GCOV_PREFIX'] = bPath + '/svc_gcov'
numSegments = len(bPath.split('/')) # "/x/TDengine/build" should yield 3
# numSegments += 2 # cover "/src" after build
# numSegments = numSegments - 1 # DEBUG only
env['GCOV_PREFIX'] = bPath + '/src_s' # Server side source
env['GCOV_PREFIX_STRIP'] = str(numSegments) # Strip every element, plus, ENV needs strings
# VERY VERY important note: GCOV data collection NOT effective upon SIG_KILL
Logging.info("Preparing GCOV environement to strip {} elements and use path: {}".format(
numSegments, env['GCOV_PREFIX'] ))
......@@ -258,14 +260,15 @@ class TdeSubProcess:
TdeInstance.prepareGcovEnv(myEnv)
# print(myEnv)
# print(myEnv.items())
# print("Starting TDengine with env: ", myEnv.items())
# print("Starting TDengine via Shell: {}".format(cmdLineStr))
useShell = True
self.subProcess = subprocess.Popen(
' '.join(cmdLine) if useShell else cmdLine,
shell=useShell,
# svcCmdSingle, shell=True, # capture core dump?
# ' '.join(cmdLine) if useShell else cmdLine,
# shell=useShell,
' '.join(cmdLine),
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
# bufsize=1, # not supported in binary mode
......@@ -273,7 +276,8 @@ class TdeSubProcess:
env=myEnv
) # had text=True, which interferred with reading EOF
STOP_SIGNAL = signal.SIGKILL # What signal to use (in kill) to stop a taosd process?
STOP_SIGNAL = signal.SIGKILL # signal.SIGKILL/SIGINT # What signal to use (in kill) to stop a taosd process?
SIG_KILL_RETCODE = 137 # ref: https://stackoverflow.com/questions/43268156/process-finished-with-exit-code-137-in-pycharm
def stop(self):
"""
......@@ -320,8 +324,12 @@ class TdeSubProcess:
retCode = self.subProcess.returncode # should always be there
# May throw subprocess.TimeoutExpired exception above, therefore
# The process is guranteed to have ended by now
self.subProcess = None
if retCode != 0: # != (- signal.SIGINT):
self.subProcess = None
if retCode == self.SIG_KILL_RETCODE:
Logging.info("TSP.stop(): sub proc KILLED, as expected")
elif retCode == (- self.STOP_SIGNAL):
Logging.info("TSP.stop(), sub process STOPPED, as expected")
elif retCode != 0: # != (- signal.SIGINT):
Logging.error("TSP.stop(): Failed to stop sub proc properly w/ SIG {}, retCode={}".format(
self.STOP_SIGNAL, retCode))
else:
......
......@@ -19,6 +19,7 @@ python3 ./test.py -f insert/randomNullCommit.py
python3 insert/retentionpolicy.py
python3 ./test.py -f insert/alterTableAndInsert.py
python3 ./test.py -f insert/insertIntoTwoTables.py
python3 ./test.py -f insert/before_1970.py
python3 ./test.py -f table/alter_wal0.py
python3 ./test.py -f table/column_name.py
......@@ -228,6 +229,7 @@ python3 ./test.py -f update/merge_commit_data2.py
python3 ./test.py -f update/merge_commit_data2_update0.py
python3 ./test.py -f update/merge_commit_last-0.py
python3 ./test.py -f update/merge_commit_last.py
python3 ./test.py -f update/bug_td2279.py
# wal
python3 ./test.py -f wal/addOldWalTest.py
\ No newline at end of file
#!/usr/bin/python3
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# install pip
# pip install src/connector/python/linux/python2/
import sys
import os
import os.path
import time
import glob
import getopt
import subprocess
from shutil import which
from multipledispatch import dispatch
@dispatch(str, str)
def v_print(msg: str, arg: str):
if verbose:
print(msg % arg)
@dispatch(str, int)
def v_print(msg: str, arg: int):
if verbose:
print(msg % int(arg))
@dispatch(str, int, int)
def v_print(msg: str, arg1: int, arg2: int):
if verbose:
print(msg % (int(arg1), int(arg2)))
@dispatch(str, int, int, int)
def v_print(msg: str, arg1: int, arg2: int, arg3: int):
if verbose:
print(msg % (int(arg1), int(arg2), int(arg3)))
@dispatch(str, int, int, int, int)
def v_print(msg: str, arg1: int, arg2: int, arg3: int, arg4: int):
if verbose:
print(msg % (int(arg1), int(arg2), int(arg3), int(arg4)))
def isHiveMQInstalled():
v_print("%s", "Check if HiveMQ installed")
defaultHiveMQPath = "/opt/hivemq*"
hiveMQDir = glob.glob(defaultHiveMQPath)
if (len(hiveMQDir) == 0):
v_print("%s", "ERROR: HiveMQ NOT found")
return False
else:
v_print("HiveMQ installed at %s", hiveMQDir[0])
return True
def isMosquittoInstalled():
v_print("%s", "Check if mosquitto installed")
if not which('mosquitto_pub'):
v_print("%s", "ERROR: mosquitto is NOT installed")
return False
else:
return True
def installExtension():
currentDir = os.getcwd()
extDir = 'src/connector/hivemq-tdengine-extension'
os.chdir('../..')
os.system('git submodule update --init -- %s' % extDir)
os.chdir(extDir)
v_print("%s", "build extension..")
os.system('mvn clean package')
tdExtensionZip = 'target/hivemq-tdengine-extension*.zip'
tdExtensionZipDir = glob.glob(tdExtensionZip)
defaultHiveMQPath = "/opt/hivemq*"
hiveMQDir = glob.glob(defaultHiveMQPath)
extPath = hiveMQDir[0] + '/extensions'
tdExtDir = glob.glob(extPath + '/hivemq-tdengine-extension')
if len(tdExtDir):
v_print("%s", "delete exist extension..")
os.system('rm -rf %s' % tdExtDir[0])
v_print("%s", "unzip extension..")
os.system('unzip %s -d %s' % (tdExtensionZipDir[0], extPath))
os.chdir(currentDir)
def stopProgram(prog: str):
psCmd = "ps ax|grep -w %s| grep -v grep | awk '{print $1}'" % prog
processID = subprocess.check_output(
psCmd, shell=True).decode("utf-8")
while(processID):
killCmd = "kill -TERM %s > /dev/null 2>&1" % processID
os.system(killCmd)
time.sleep(1)
processID = subprocess.check_output(
psCmd, shell=True).decode("utf-8")
pass
def stopHiveMQ():
stopProgram("hivemq.jar")
v_print("%s", "ERROR: HiveMQ is NOT running")
def checkProgramRunning(prog: str):
psCmd = "ps ax|grep -w %s| grep -v grep | awk '{print $1}'" % prog
processID = subprocess.check_output(
psCmd, shell=True).decode("utf-8")
if not processID:
v_print("ERROR: %s is NOT running", prog)
return False
else:
return True
def runHiveMQ():
defaultHiveMQPath = "/opt/hivemq*"
hiveMQDir = glob.glob(defaultHiveMQPath)
runPath = hiveMQDir[0] + '/bin/run.sh > /dev/null &'
os.system(runPath)
time.sleep(10)
if not checkProgramRunning("hivemq.jar"):
return False
else:
v_print("%s", "hivemq is running")
return True
def getBuildPath():
selfPath = os.path.dirname(os.path.realpath(__file__))
binPath = ''
if ("community" in selfPath):
projPath = selfPath[:selfPath.find("community")]
else:
projPath = selfPath[:selfPath.find("tests")]
for root, dirs, files in os.walk(projPath):
if ("taosd" in files):
rootRealPath = os.path.dirname(os.path.realpath(root))
if ("packaging" not in rootRealPath):
binPath = root[:len(root) - len("/build/bin")]
break
return binPath
def runTDengine():
stopProgram("taosd")
buildPath = getBuildPath()
if (buildPath == ""):
v_print("%s", "ERROR: taosd NOT found!")
sys.exit(1)
else:
v_print("%s", "taosd found in %s" % buildPath)
binPath = buildPath + "/build/bin/taosd"
os.system('%s > /dev/null &' % binPath)
time.sleep(10)
if not checkProgramRunning("taosd"):
return False
else:
v_print("%s", "TDengine is running")
return True
def reCreateDatabase():
buildPath = getBuildPath()
binPath = buildPath + "/build/bin/taos"
os.system('%s -s "DROP DATABASE IF EXISTS hivemq"' % binPath)
os.system('%s -s "CREATE DATABASE IF NOT EXISTS hivemq"' % binPath)
def sendMqttMsg(topic: str, payload: str):
testStr = 'mosquitto_pub -t %s -m "%s"' % (topic, payload)
os.system(testStr)
time.sleep(3)
def checkTDengineData(topic: str, payload: str):
buildPath = getBuildPath()
binPath = buildPath + "/build/bin/taos"
output = subprocess.check_output(
'%s -s "select * from hivemq.mqtt_payload"' %
binPath, shell=True).decode('utf-8')
if (topic in output) and (payload in output):
v_print("%s", output)
return True
else:
v_print("%s", "ERROR: mqtt topic or payload NOT found")
return False
if __name__ == "__main__":
verbose = True
testTopic = 'test'
testPayload = 'hello world'
if not isHiveMQInstalled():
sys.exit(1)
if not isMosquittoInstalled():
sys.exit(1)
stopHiveMQ()
installExtension()
if not runTDengine():
sys.exit(1)
reCreateDatabase()
if not runHiveMQ():
sys.exit(1)
sendMqttMsg(testTopic, testPayload)
if not checkTDengineData(testTopic, testPayload):
sys.exit(1)
sys.exit(0)
import taos
import datetime
import random
import multiprocessing
def taos_excute(table, connect_host):
conn = taos.connect(host=connect_host, user="root", password="taosdata", config="/etc/taos", database='test')
cursor = conn.cursor()
for i in range(1000000):
pk = random.randint(100001, 300000)
time_now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
col1 = random.randint(1, 10000)
col2 = random.randint(1, 10000)
col3 = random.randint(1, 10000)
col4 = random.randint(1, 10000)
col5 = random.randint(1, 10000)
col6 = random.randint(1, 10000)
sql = f"INSERT INTO {table}_{pk} USING {table} TAGS ({pk}) VALUES ('{time_now}', {col1}, {col2}, {col3}, {col4}, {col5}, {col6})"
cursor.execute(sql)
cursor.close()
conn.close()
def taos_init(table, connect_host, pk):
conn = taos.connect(host=connect_host, user="root", password="taosdata", config="/etc/taos", database='test')
cursor = conn.cursor()
sql = f"CREATE TABLE {table}_{pk} USING {table} TAGS ({pk})"
cursor.execute(sql)
cursor.close()
conn.close()
print("init time:", datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
connect_list = ["node1", "node2", "node3", "node4", "node5"]
pool = multiprocessing.Pool(processes=108)
for pk in range(100001, 300000):
pool.apply_async(func=taos_init, args=("test", connect_list[pk % 5], pk, ))
print("start time:", datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
for i in range(10000):
pool.apply_async(func=taos_excute, args=("test", connect_list[i % 5],))
pool.close()
pool.join()
print("end time:", datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
\ No newline at end of file
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import sys
import taos
from util.log import tdLog
from util.cases import tdCases
from util.sql import tdSql
import time
import datetime
import csv
import random
import pandas as pd
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
self.ts = 1500074556514
def writeCSV(self):
with open('test3.csv','w', encoding='utf-8', newline='') as csvFile:
writer = csv.writer(csvFile, dialect='excel')
for i in range(1000000):
newTimestamp = self.ts + random.randint(10000000, 10000000000) + random.randint(1000, 10000000) + random.randint(1, 1000)
d = datetime.datetime.fromtimestamp(newTimestamp / 1000)
dt = str(d.strftime("%Y-%m-%d %H:%M:%S.%f"))
writer.writerow(["'%s'" % dt, random.randint(1, 100), random.uniform(1, 100), random.randint(1, 100), random.randint(1, 100)])
def removCSVHeader(self):
data = pd.read_csv("ordered.csv")
data = data.drop([0])
data.to_csv("ordered.csv", header = False, index = False)
def run(self):
tdSql.prepare()
tdSql.execute("create table t1(ts timestamp, c1 int, c2 float, c3 int, c4 int)")
startTime = time.time()
tdSql.execute("insert into t1 file 'outoforder.csv'")
duration = time.time() - startTime
print("Out of Order - Insert time: %d" % duration)
tdSql.query("select count(*) from t1")
rows = tdSql.getData(0, 0)
tdSql.execute("create table t2(ts timestamp, c1 int, c2 float, c3 int, c4 int)")
startTime = time.time()
tdSql.execute("insert into t2 file 'ordered.csv'")
duration = time.time() - startTime
print("Ordered - Insert time: %d" % duration)
tdSql.query("select count(*) from t2")
tdSql.checkData(0,0, rows)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())
\ No newline at end of file
......@@ -18,7 +18,7 @@ import time
import argparse
class RestfulInsert:
def __init__(self, host, startTimestamp, dbname, threads, tables, records, batchSize, tbNamePerfix, outOfOrder):
def __init__(self, host, startTimestamp, dbname, threads, tables, records, batchSize, tbNamePerfix, outOfOrder,tablePerbatch):
self.header = {'Authorization': 'Basic cm9vdDp0YW9zZGF0YQ=='}
self.url = "http://%s:6041/rest/sql" % host
self.ts = startTimestamp
......@@ -29,32 +29,71 @@ class RestfulInsert:
self.batchSize = batchSize
self.tableNamePerfix = tbNamePerfix
self.outOfOrder = outOfOrder
self.tablePerbatch = tablePerbatch
def createTable(self, threadID):
tablesPerThread = int (self.numOfTables / self.numOfThreads)
print("create table %d to %d" % (tablesPerThread * threadID, tablesPerThread * (threadID + 1) - 1))
for i in range(tablesPerThread):
tablesPerThread = int (self.numOfTables / self.numOfThreads)
loop = tablesPerThread if threadID != self.numOfThreads - 1 else self.numOfTables - tablesPerThread * threadID
print("create table %d to %d" % (tablesPerThread * threadID, tablesPerThread * threadID + loop - 1))
for i in range(loop):
tableID = threadID * tablesPerThread
if tableID + i >= self.numOfTables : break
name = 'beijing' if tableID % 2 == 0 else 'shanghai'
data = "create table if not exists %s.%s%d using %s.meters tags(%d, '%s')" % (self.dbname, self.tableNamePerfix, tableID + i, self.dbname, tableID + i, name)
response = requests.post(self.url, data, headers = self.header)
if response.status_code != 200:
print(response.content)
def insertData(self, threadID):
print("thread %d started" % threadID)
tablesPerThread = int (self.numOfTables / self.numOfThreads)
for i in range(tablesPerThread):
tableID = i + threadID * tablesPerThread
start = self.ts
for j in range(int(self.recordsPerTable / self.batchSize)):
data = "insert into %s.%s%d values" % (self.dbname, self.tableNamePerfix, tableID)
values = []
for k in range(self.batchSize):
data += "(%d, %d, %d, %d)" % (start + j * self.batchSize + k, random.randint(1, 100), random.randint(1, 100), random.randint(1, 100))
response = requests.post(self.url, data, headers = self.header)
if response.status_code != 200:
print(response.content)
tablesPerThread = int (self.numOfTables / self.numOfThreads)
loop = int(self.recordsPerTable / self.batchSize)
if self.tablePerbatch == 1 :
for i in range(tablesPerThread+1):
tableID = i + threadID * tablesPerThread
if tableID >= self.numOfTables: return
start = self.ts
start1=time.time()
for k in range(loop):
data = "insert into %s.%s%d values" % (self.dbname, self.tableNamePerfix, tableID)
values = []
bloop = self.batchSize if k != loop - 1 else self.recordsPerTable - self.batchSize * k
for l in range(bloop):
values.append("(%d, %d, %d, %d)" % (start + k * self.batchSize + l, random.randint(1, 100), random.randint(1, 100), random.randint(1, 100)))
if len(data) > 1048576 :
print ('batch size is larger than 1M')
exit(-1)
if self.outOfOrder :
random.shuffle(values)
data+=''.join(values)
response = requests.post(self.url, data, headers = self.header)
if response.status_code != 200:
print(response.content)
else:
for i in range(0,tablesPerThread+self.tablePerbatch,self.tablePerbatch):
for k in range(loop):
data = "insert into "
for j in range(self.tablePerbatch):
tableID = i + threadID * tablesPerThread+j
if tableID >= self.numOfTables: return
start = self.ts
data += "%s.%s%d values" % (self.dbname, self.tableNamePerfix, tableID)
values = []
bloop = self.batchSize if k != loop - 1 else self.recordsPerTable - self.batchSize * k
for l in range(bloop):
values.append("(%d, %d, %d, %d)" % (start + k * self.batchSize + l, random.randint(1, 100), random.randint(1, 100), random.randint(1, 100)))
if self.outOfOrder :
random.shuffle(values)
data+=''.join(values)
if len(data) > 1024*1024 :
print ('batch size is larger than 1M')
exit(-1)
response = requests.post(self.url, data, headers = self.header)
if response.status_code != 200:
print(response.content)
def insertUnlimitedData(self, threadID):
print("thread %d started" % threadID)
......@@ -85,7 +124,7 @@ class RestfulInsert:
if response.status_code != 200:
print(response.content)
def run(self):
def run(self):
data = "create database if not exists %s" % self.dbname
requests.post(self.url, data, headers = self.header)
data = "create table if not exists %s.meters(ts timestamp, f1 int, f2 int, f3 int) tags(id int, loc nchar(20))" % self.dbname
......@@ -114,7 +153,7 @@ class RestfulInsert:
for i in range(self.numOfThreads):
threads[i].join()
print("inserting %d records takes %d seconds" % (self.numOfTables * self.recordsPerTable, (time.time() - startTime)))
print("inserting %s records takes %d seconds" % (self.numOfTables * self.recordsPerTable, (time.time() - startTime)))
parser = argparse.ArgumentParser()
parser.add_argument(
......@@ -149,14 +188,14 @@ parser.add_argument(
'-T',
'--number-of-tables',
action='store',
default=1000,
default=10000,
type=int,
help='Number of tables to be created (default: 1000)')
parser.add_argument(
'-r',
'--number-of-records',
action='store',
default=1000,
default=10000,
type=int,
help='Number of record to be created for each table (default: 1000, -1 for unlimited records)')
parser.add_argument(
......@@ -178,7 +217,18 @@ parser.add_argument(
'--out-of-order',
action='store_true',
help='The order of test data (default: False)')
parser.add_argument(
'-b',
'--table-per-batch',
action='store',
default=1,
type=int,
help='the table per batch (default: 1)')
args = parser.parse_args()
ri = RestfulInsert(args.host_name, args.start_timestamp, args.db_name, args.number_of_threads, args.number_of_tables, args.number_of_records, args.batch_size, args.table_name_prefix, args.out_of_order)
ri = RestfulInsert(
args.host_name, args.start_timestamp, args.db_name, args.number_of_threads, args.number_of_tables,
args.number_of_records, args.batch_size, args.table_name_prefix, args.out_of_order, args.table_per_batch)
ri.run()
\ No newline at end of file
#!/bin/bash
ulimit -c unlimited
python3 ./test.py -f insert/basic.py
python3 ./test.py -f insert/int.py
python3 ./test.py -f insert/float.py
python3 ./test.py -f insert/bigint.py
python3 ./test.py -f insert/bool.py
python3 ./test.py -f insert/double.py
python3 ./test.py -f insert/smallint.py
python3 ./test.py -f insert/tinyint.py
python3 ./test.py -f insert/date.py
python3 ./test.py -f insert/binary.py
python3 ./test.py -f insert/nchar.py
#python3 ./test.py -f insert/nchar-boundary.py
python3 ./test.py -f insert/nchar-unicode.py
python3 ./test.py -f insert/multi.py
python3 ./test.py -f insert/randomNullCommit.py
python3 insert/retentionpolicy.py
python3 ./test.py -f insert/alterTableAndInsert.py
python3 ./test.py -f insert/insertIntoTwoTables.py
python3 ./test.py -f table/alter_wal0.py
python3 ./test.py -f table/column_name.py
python3 ./test.py -f table/column_num.py
python3 ./test.py -f table/db_table.py
python3 ./test.py -f table/create_sensitive.py
#python3 ./test.py -f table/tablename-boundary.py
# tag
python3 ./test.py -f tag_lite/filter.py
python3 ./test.py -f tag_lite/create-tags-boundary.py
python3 ./test.py -f tag_lite/3.py
python3 ./test.py -f tag_lite/4.py
python3 ./test.py -f tag_lite/5.py
python3 ./test.py -f tag_lite/6.py
python3 ./test.py -f tag_lite/add.py
python3 ./test.py -f tag_lite/bigint.py
python3 ./test.py -f tag_lite/binary_binary.py
python3 ./test.py -f tag_lite/binary.py
python3 ./test.py -f tag_lite/bool_binary.py
python3 ./test.py -f tag_lite/bool_int.py
python3 ./test.py -f tag_lite/bool.py
python3 ./test.py -f tag_lite/change.py
python3 ./test.py -f tag_lite/column.py
python3 ./test.py -f tag_lite/commit.py
python3 ./test.py -f tag_lite/create.py
python3 ./test.py -f tag_lite/datatype.py
python3 ./test.py -f tag_lite/datatype-without-alter.py
python3 ./test.py -f tag_lite/delete.py
python3 ./test.py -f tag_lite/double.py
python3 ./test.py -f tag_lite/float.py
python3 ./test.py -f tag_lite/int_binary.py
python3 ./test.py -f tag_lite/int_float.py
python3 ./test.py -f tag_lite/int.py
python3 ./test.py -f tag_lite/set.py
python3 ./test.py -f tag_lite/smallint.py
python3 ./test.py -f tag_lite/tinyint.py
#python3 ./test.py -f dbmgmt/database-name-boundary.py
python3 ./test.py -f import_merge/importBlock1HO.py
python3 ./test.py -f import_merge/importBlock1HPO.py
python3 ./test.py -f import_merge/importBlock1H.py
python3 ./test.py -f import_merge/importBlock1S.py
python3 ./test.py -f import_merge/importBlock1Sub.py
python3 ./test.py -f import_merge/importBlock1TO.py
python3 ./test.py -f import_merge/importBlock1TPO.py
python3 ./test.py -f import_merge/importBlock1T.py
python3 ./test.py -f import_merge/importBlock2HO.py
python3 ./test.py -f import_merge/importBlock2HPO.py
python3 ./test.py -f import_merge/importBlock2H.py
python3 ./test.py -f import_merge/importBlock2S.py
python3 ./test.py -f import_merge/importBlock2Sub.py
python3 ./test.py -f import_merge/importBlock2TO.py
python3 ./test.py -f import_merge/importBlock2TPO.py
python3 ./test.py -f import_merge/importBlock2T.py
python3 ./test.py -f import_merge/importBlockbetween.py
python3 ./test.py -f import_merge/importCacheFileHO.py
python3 ./test.py -f import_merge/importCacheFileHPO.py
python3 ./test.py -f import_merge/importCacheFileH.py
python3 ./test.py -f import_merge/importCacheFileS.py
python3 ./test.py -f import_merge/importCacheFileSub.py
python3 ./test.py -f import_merge/importCacheFileTO.py
python3 ./test.py -f import_merge/importCacheFileTPO.py
python3 ./test.py -f import_merge/importCacheFileT.py
python3 ./test.py -f import_merge/importDataH2.py
python3 ./test.py -f import_merge/importDataHO2.py
python3 ./test.py -f import_merge/importDataHO.py
python3 ./test.py -f import_merge/importDataHPO.py
python3 ./test.py -f import_merge/importDataLastHO.py
python3 ./test.py -f import_merge/importDataLastHPO.py
python3 ./test.py -f import_merge/importDataLastH.py
python3 ./test.py -f import_merge/importDataLastS.py
python3 ./test.py -f import_merge/importDataLastSub.py
python3 ./test.py -f import_merge/importDataLastTO.py
python3 ./test.py -f import_merge/importDataLastTPO.py
python3 ./test.py -f import_merge/importDataLastT.py
python3 ./test.py -f import_merge/importDataS.py
python3 ./test.py -f import_merge/importDataSub.py
python3 ./test.py -f import_merge/importDataTO.py
python3 ./test.py -f import_merge/importDataTPO.py
python3 ./test.py -f import_merge/importDataT.py
python3 ./test.py -f import_merge/importHeadOverlap.py
python3 ./test.py -f import_merge/importHeadPartOverlap.py
python3 ./test.py -f import_merge/importHead.py
python3 ./test.py -f import_merge/importHORestart.py
python3 ./test.py -f import_merge/importHPORestart.py
python3 ./test.py -f import_merge/importHRestart.py
python3 ./test.py -f import_merge/importLastHO.py
python3 ./test.py -f import_merge/importLastHPO.py
python3 ./test.py -f import_merge/importLastH.py
python3 ./test.py -f import_merge/importLastS.py
python3 ./test.py -f import_merge/importLastSub.py
python3 ./test.py -f import_merge/importLastTO.py
python3 ./test.py -f import_merge/importLastTPO.py
python3 ./test.py -f import_merge/importLastT.py
python3 ./test.py -f import_merge/importSpan.py
python3 ./test.py -f import_merge/importSRestart.py
python3 ./test.py -f import_merge/importSubRestart.py
python3 ./test.py -f import_merge/importTailOverlap.py
python3 ./test.py -f import_merge/importTailPartOverlap.py
python3 ./test.py -f import_merge/importTail.py
python3 ./test.py -f import_merge/importToCommit.py
python3 ./test.py -f import_merge/importTORestart.py
python3 ./test.py -f import_merge/importTPORestart.py
python3 ./test.py -f import_merge/importTRestart.py
python3 ./test.py -f import_merge/importInsertThenImport.py
python3 ./test.py -f import_merge/importCSV.py
# user
python3 ./test.py -f user/user_create.py
python3 ./test.py -f user/pass_len.py
# stable
python3 ./test.py -f stable/query_after_reset.py
# table
python3 ./test.py -f table/del_stable.py
#query
python3 ./test.py -f query/filter.py
python3 ./test.py -f query/filterCombo.py
python3 ./test.py -f query/queryNormal.py
python3 ./test.py -f query/queryError.py
python3 ./test.py -f query/filterAllIntTypes.py
python3 ./test.py -f query/filterFloatAndDouble.py
python3 ./test.py -f query/filterOtherTypes.py
python3 ./test.py -f query/querySort.py
python3 ./test.py -f query/queryJoin.py
python3 ./test.py -f query/select_last_crash.py
python3 ./test.py -f query/queryNullValueTest.py
python3 ./test.py -f query/queryInsertValue.py
python3 ./test.py -f query/queryConnection.py
python3 ./test.py -f query/queryCountCSVData.py
python3 ./test.py -f query/natualInterval.py
python3 ./test.py -f query/bug1471.py
#python3 ./test.py -f query/dataLossTest.py
python3 ./test.py -f query/bug1874.py
python3 ./test.py -f query/bug1875.py
python3 ./test.py -f query/bug1876.py
python3 ./test.py -f query/bug2218.py
#stream
python3 ./test.py -f stream/metric_1.py
python3 ./test.py -f stream/new.py
python3 ./test.py -f stream/stream1.py
python3 ./test.py -f stream/stream2.py
#python3 ./test.py -f stream/parser.py
python3 ./test.py -f stream/history.py
#alter table
python3 ./test.py -f alter/alter_table_crash.py
# client
python3 ./test.py -f client/client.py
python3 ./test.py -f client/version.py
python3 ./test.py -f client/alterDatabase.py
# Misc
python3 testCompress.py
python3 testNoCompress.py
python3 testMinTablesPerVnode.py
# functions
python3 ./test.py -f functions/function_avg.py -r 1
python3 ./test.py -f functions/function_bottom.py -r 1
python3 ./test.py -f functions/function_count.py -r 1
python3 ./test.py -f functions/function_diff.py -r 1
python3 ./test.py -f functions/function_first.py -r 1
python3 ./test.py -f functions/function_last.py -r 1
python3 ./test.py -f functions/function_last_row.py -r 1
python3 ./test.py -f functions/function_leastsquares.py -r 1
python3 ./test.py -f functions/function_max.py -r 1
python3 ./test.py -f functions/function_min.py -r 1
python3 ./test.py -f functions/function_operations.py -r 1
python3 ./test.py -f functions/function_percentile.py -r 1
python3 ./test.py -f functions/function_spread.py -r 1
python3 ./test.py -f functions/function_stddev.py -r 1
python3 ./test.py -f functions/function_sum.py -r 1
python3 ./test.py -f functions/function_top.py -r 1
#python3 ./test.py -f functions/function_twa.py -r 1
python3 queryCount.py
python3 ./test.py -f query/queryGroupbyWithInterval.py
python3 client/twoClients.py
python3 test.py -f query/queryInterval.py
python3 test.py -f query/queryFillTest.py
# tools
python3 test.py -f tools/taosdemoTest.py
python3 test.py -f tools/taosdumpTest.py
python3 test.py -f tools/lowaTest.py
# subscribe
python3 test.py -f subscribe/singlemeter.py
#python3 test.py -f subscribe/stability.py
python3 test.py -f subscribe/supertable.py
# update
python3 ./test.py -f update/allow_update.py
python3 ./test.py -f update/allow_update-0.py
python3 ./test.py -f update/append_commit_data.py
python3 ./test.py -f update/append_commit_last-0.py
python3 ./test.py -f update/append_commit_last.py
python3 ./test.py -f update/merge_commit_data.py
python3 ./test.py -f update/merge_commit_data-0.py
python3 ./test.py -f update/merge_commit_data2.py
python3 ./test.py -f update/merge_commit_data2_update0.py
python3 ./test.py -f update/merge_commit_last-0.py
python3 ./test.py -f update/merge_commit_last.py
# wal
python3 ./test.py -f wal/addOldWalTest.py
\ No newline at end of file
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import sys
from util.log import *
from util.cases import *
from util.sql import *
from util.dnodes import *
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
def run(self):
tdSql.prepare()
print("==========step1")
print("create table && insert data")
tdSql.execute("create table mt0 (ts timestamp, c1 int, c2 float, c3 bigint, c4 smallint, c5 tinyint, c6 double, c7 bool,c8 binary(20),c9 nchar(20))")
insertRows = 1000
t0 = 1604298064000
tdLog.info("insert %d rows" % (insertRows))
for i in range(insertRows):
ret = tdSql.execute(
"insert into mt0 values (%d , %d,%d,%d,%d,%d,%d,%d,'%s','%s')" %
(t0+i,i%100,i/2,i%41,i%100,i%100,i*1.0,i%2,'taos'+str(i%100),'涛思'+str(i%100)))
print("==========step2")
print("test last with group by normal_col ")
tdSql.query('select last(c1) from mt0 group by c3')
tdSql.checkData(0,0,84)
tdSql.checkData(0,1,85)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())
\ No newline at end of file
......@@ -38,12 +38,12 @@ class TDTestCase:
print("test col*1*1 desc ")
tdSql.query('select c1,c1*1*1,c2*1*1,c3*1*1,c4*1*1,c5*1*1,c6*1*1 from mt0 order by ts desc limit 2')
tdSql.checkData(0,0,99)
tdSql.checkData(0,1,0.0)
tdSql.checkData(0,2,0.0)
tdSql.checkData(0,3,0.0)
tdSql.checkData(0,4,0.0)
tdSql.checkData(0,5,0.0)
tdSql.checkData(0,6,0.0)
tdSql.checkData(0,1,99.0)
tdSql.checkData(0,2,499.0)
tdSql.checkData(0,3,99.0)
tdSql.checkData(0,4,99.0)
tdSql.checkData(0,5,99.0)
tdSql.checkData(0,6,999.0)
def stop(self):
......
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import sys
import taos
from util.log import *
from util.cases import *
from util.sql import *
from util.dnodes import *
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor())
self.ts = 1606700000000
def restartTaosd(self):
tdDnodes.stop(1)
tdDnodes.startWithoutSleep(1)
tdSql.execute("use db")
def run(self):
tdSql.prepare()
print("==============step1")
tdSql.execute("create table t (ts timestamp, a int)")
for i in range(3276):
tdSql.execute("insert into t values(%d, 0)" % (self.ts + i))
newTs = 1606700010000
for i in range(3275):
tdSql.execute("insert into t values(%d, 0)" % (self.ts + i))
tdSql.execute("insert into t values(%d, 0)" % 1606700013280)
self.restartTaosd()
for i in range(1606700003275, 1606700006609):
tdSql.execute("insert into t values(%d, 0)" % i)
tdSql.execute("insert into t values(%d, 0)" % 1606700006612)
self.restartTaosd()
tdSql.execute("insert into t values(%d, 0)" % 1606700006610)
tdSql.query("select * from t")
tdSql.checkRows(6612)
tdDnodes.stop(1)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())
......@@ -115,31 +115,31 @@ if $data7_db != 20,20,20 then
return -1
endi
sql alter database db keep 10
sql alter database db keep 20
sql show databases
print keep $data7_db
if $data7_db != 20,20,10 then
if $data7_db != 20,20,20 then
return -1
endi
sql alter database db keep 20
sql alter database db keep 30
sql show databases
print keep $data7_db
if $data7_db != 20,20,20 then
if $data7_db != 20,20,30 then
return -1
endi
sql alter database db keep 30
sql alter database db keep 40
sql show databases
print keep $data7_db
if $data7_db != 20,20,30 then
if $data7_db != 20,20,40 then
return -1
endi
sql alter database db keep 40
sql alter database db keep 30
sql alter database db keep 20
sql alter database db keep 10
sql_error alter database db keep 10
sql_error alter database db keep 9
sql_error alter database db keep 1
sql alter database db keep 0
......@@ -277,4 +277,4 @@ sql_error alter database db prec 'us'
print ============== step status
sql_error alter database db status 'delete'
system sh/exec.sh -n dnode1 -s stop -x SIGINT
\ No newline at end of file
system sh/exec.sh -n dnode1 -s stop -x SIGINT
......@@ -137,6 +137,12 @@ if [ "$2" != "sim" ]; then
elif [ "$1" == "pytest" ]; then
echo "### run Python full test ###"
runPyCaseOneByOne fulltest.sh
elif [ "$1" == "p1" ]; then
echo "### run Python_1 test ###"
runPyCaseOneByOne pytest_1.sh
elif [ "$1" == "p2" ]; then
echo "### run Python_2 test ###"
runPyCaseOneByOne pytest_2.sh
elif [ "$1" == "b2" ] || [ "$1" == "b3" ]; then
exit $(($totalFailed + $totalPyFailed))
elif [ "$1" == "smoke" ] || [ -z "$1" ]; then
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册