Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
提交 5f21e2e7 编写于 作者: H Hongze Cheng
浏览文件
操作

Merge branch 'develop' into hotfix/TD-1429

上级 f41b47e1 d0935c02
展开全部 显示空白变更内容
内联 并排
Showing with 1518 addition and 612 deletion
Jenkinsfile
浏览文件 @ 5f21e2e7
pipeline {
agent any
agent none
environment{
WK = '/var/lib/jenkins/workspace/TDinternal'
WKC= '/var/lib/jenkins/workspace/TDinternal/community'
}
stages {
stage('build TDengine') {
agent{label 'master'}
steps {
sh '''cd ${WORKSPACE}
export TZ=Asia/Harbin
date
rm -rf ${WORKSPACE}/debug
mkdir debug
cd debug
cmake .. > /dev/null
make > /dev/null
cd ${WORKSPACE}/debug'''
sh '''
cd ${WKC}
git checkout develop
git pull
git submodule update
cd ${WK}
git checkout develop
git pull
export TZ=Asia/Harbin
date
rm -rf ${WK}/debug
mkdir debug
cd debug
#cmake .. > /dev/null
#make > /dev/null
'''
}
}
stage('test_tsim') {
stage('Parallel test stage') {
parallel {
stage('test') {
stage('pytest') {
agent{label 'master'}
steps {
sh '''
date
cd ${WKC}
git checkout develop
git pull
git submodule update
cd ${WK}
git checkout develop
git pull
export TZ=Asia/Harbin
date
rm -rf ${WK}/debug
mkdir debug
cd debug
cmake .. > /dev/null
make > /dev/null
cd ${WKC}/tests
#./test-all.sh smoke
./test-all.sh pytest
date'''
}
}
stage('test_b1') {
agent{label '184'}
steps {
sh '''cd ${WORKSPACE}/tests
#./test-all.sh smoke
sudo ./test-all.sh full'''
sh '''
date
cd ${WKC}
git checkout develop
git pull
git submodule update
cd ${WK}
git checkout develop
git pull
export TZ=Asia/Harbin
date
rm -rf ${WK}/debug
mkdir debug
cd debug
cmake .. > /dev/null
make > /dev/null
cd ${WKC}/tests
#./test-all.sh smoke
./test-all.sh b1
date'''
}
}
stage('test_crash_gen') {
agent{label "185"}
steps {
sh '''cd ${WORKSPACE}/tests/pytest
sudo ./crash_gen.sh -a -p -t 4 -s 2000'''
sh '''
cd ${WKC}
git checkout develop
git pull
git submodule update
cd ${WK}
git checkout develop
git pull
export TZ=Asia/Harbin
rm -rf ${WK}/debug
mkdir debug
cd debug
cmake .. > /dev/null
make > /dev/null
cd ${WKC}/tests/pytest
./crash_gen.sh -a -p -t 4 -s 2000
date
cd ${WKC}/tests
./test-all.sh b2
date
'''
}
}
stage('test_valgrind') {
agent{label "186"}
steps {
sh '''cd ${WORKSPACE}/tests/pytest
sudo ./valgrind-test.sh 2>&1 > mem-error-out.log
grep \'start to execute\\|ERROR SUMMARY\' mem-error-out.log|grep -v \'grep\'|uniq|tee uniq-mem-error-out.log
sh '''
date
cd ${WKC}
git checkout develop
git pull
git submodule update
cd ${WK}
git checkout develop
git pull
export TZ=Asia/Harbin
date
rm -rf ${WK}/debug
mkdir debug
cd debug
cmake .. > /dev/null
make > /dev/null
cd ${WKC}/tests/pytest
./valgrind-test.sh 2>&1 > mem-error-out.log
./handle_val_log.sh
for memError in `grep \'ERROR SUMMARY\' uniq-mem-error-out.log | awk \'{print $4}\'`
do
if [ -n "$memError" ]; then
if [ "$memError" -gt 12 ]; then
echo -e "${RED} ## Memory errors number valgrind reports is $memError.\\
More than our threshold! ## ${NC}"
travis_terminate $memError
fi
fi
done
grep \'start to execute\\|definitely lost:\' mem-error-out.log|grep -v \'grep\'|uniq|tee uniq-definitely-lost-out.log
for defiMemError in `grep \'definitely lost:\' uniq-definitely-lost-out.log | awk \'{print $7}\'`
do
if [ -n "$defiMemError" ]; then
if [ "$defiMemError" -gt 13 ]; then
echo -e "${RED} ## Memory errors number valgrind reports \\
Definitely lost is $defiMemError. More than our threshold! ## ${NC}"
travis_terminate $defiMemError
fi
fi
done'''
date
cd ${WKC}/tests
./test-all.sh b3
date'''
}
}
......@@ -67,4 +143,5 @@ done'''
}
}
}
\ No newline at end of file
cmake/env.inc
浏览文件 @ 5f21e2e7
......@@ -45,7 +45,7 @@ IF (${CMAKE_BUILD_TYPE} MATCHES "Debug")
ELSEIF (${CMAKE_BUILD_TYPE} MATCHES "Release")
MESSAGE(STATUS "Build Release Version")
ELSE ()
IF (TD_WINDOWS_64)
IF (TD_WINDOWS)
SET(CMAKE_BUILD_TYPE "Release")
MESSAGE(STATUS "Build Release Version in Windows as default")
ELSE ()
......
cmake/install.inc
浏览文件 @ 5f21e2e7
......@@ -16,6 +16,7 @@ ELSEIF (TD_WINDOWS)
INSTALL(DIRECTORY ${TD_COMMUNITY_DIR}/tests/examples DESTINATION .)
INSTALL(DIRECTORY ${TD_COMMUNITY_DIR}/packaging/cfg DESTINATION .)
INSTALL(FILES ${TD_COMMUNITY_DIR}/src/inc/taos.h DESTINATION include)
INSTALL(FILES ${TD_COMMUNITY_DIR}/src/inc/taoserror.h DESTINATION include)
INSTALL(FILES ${LIBRARY_OUTPUT_PATH}/taos.lib DESTINATION driver)
INSTALL(FILES ${LIBRARY_OUTPUT_PATH}/taos.exp DESTINATION driver)
INSTALL(FILES ${LIBRARY_OUTPUT_PATH}/taos.dll DESTINATION driver)
......
deps/zlib-1.2.11/inc/zconf.h
浏览文件 @ 5f21e2e7
......@@ -472,7 +472,7 @@ typedef uint64_t z_crc_t;
#endif
#ifndef Z_SOLO
#if defined(Z_HAVE_UNISTD_H) || defined(_LARGEFILE64_SOURCE)
#if (_WIN64)
#if defined(_WIN64) || defined(_WIN32)
#include <io.h>
#include <process.h>
#else
......
deps/zlib-1.2.11/src/zconf.h
浏览文件 @ 5f21e2e7
......@@ -472,7 +472,7 @@ typedef uLong FAR uLongf;
#endif
#ifndef Z_SOLO
# if defined(Z_HAVE_UNISTD_H) || defined(_LARGEFILE64_SOURCE)
#if (_WIN64)
#if defined(_WIN64) || defined(_WIN32)
#include <io.h>
#include <process.h>
#else
......
documentation20/webdocs/markdowndocs/administrator-ch.md
浏览文件 @ 5f21e2e7
......@@ -381,7 +381,7 @@ KILL STREAM <stream-id>;
## 系统监控
TDengine启动后，会自动创建一个监测数据库SYS，并自动将服务器的CPU、内存、硬盘空间、带宽、请求数、磁盘读写速度、慢查询等信息定时写入该数据库。TDengine还将重要的系统操作（比如登录、创建、删除数据库等）日志以及各种错误报警信息记录下来存放在SYS库里。系统管理员可以从CLI直接查看这个数据库，也可以在WEB通过图形化界面查看这些监测信息。
TDengine启动后，会自动创建一个监测数据库log，并自动将服务器的CPU、内存、硬盘空间、带宽、请求数、磁盘读写速度、慢查询等信息定时写入该数据库。TDengine还将重要的系统操作（比如登录、创建、删除数据库等）日志以及各种错误报警信息记录下来存放在log库里。系统管理员可以从CLI直接查看这个数据库，也可以在WEB通过图形化界面查看这些监测信息。
这些监测信息的采集缺省是打开的，但可以修改配置文件里的选项enableMonitor将其关闭或打开。
......
documentation20/webdocs/markdowndocs/connector-java-ch.md
浏览文件 @ 5f21e2e7
......@@ -72,38 +72,34 @@ maven 项目中使用如下 pom.xml 配置即可：
### 获取连接
如下所示配置即可获取 TDengine Connection：
#### 通过JdbcUrl获取连接
通过指定的jdbcUrl获取连接，如下所示：
```java
Class.forName("com.taosdata.jdbc.TSDBDriver");
String jdbcUrl = "jdbc:TAOS://127.0.0.1:6030/log?user=root&password=taosdata";
String jdbcUrl = "jdbc:TAOS://taosdemo.com:6030/test?user=root&password=taosdata";
Connection conn = DriverManager.getConnection(jdbcUrl);
```
> 端口 6030 为默认连接端口，JDBC URL 中的 log 为系统本身的监控数据库。
以上示例，建立了到hostname为taosdemo.com，端口为6030（TDengine的默认端口），数据库名为test的连接。这个url中指定用户名（user）为root，密码（password）为taosdata。
TDengine 的 JDBC URL 规范格式为：
`jdbc:TAOS://{host_ip}:{port}/[database_name]?[user={user}|&password={password}|&charset={charset}|&cfgdir={config_dir}|&locale={locale}|&timezone={timezone}]`
其中，`{}` 中的内容必须，`[]` 中为可选。配置参数说明如下：
`jdbc:TAOS://[host_name]:[port]/[database_name]?[user={user}|&password={password}|&charset={charset}|&cfgdir={config_dir}|&locale={locale}|&timezone={timezone}]`
url中的配置参数如下：
* user：登录 TDengine 用户名，默认值 root。
* password：用户登录密码，默认值 taosdata。
* charset：客户端使用的字符集，默认值为系统字符集。
* cfgdir：客户端配置文件目录路径，Linux OS 上默认值 /etc/taos ，Windows OS 上默认值 C:/TDengine/cfg。
* charset：客户端使用的字符集，默认值为系统字符集。
* locale：客户端语言环境，默认值系统当前 locale。
* timezone：客户端使用的时区，默认值为系统当前时区。
以上参数可以在 3 处配置，`优先级由高到低`分别如下：
1. JDBC URL 参数
如上所述，可以在 JDBC URL 的参数中指定。
2. java.sql.DriverManager.getConnection(String jdbcUrl, Properties connProps)
#### 使用JdbcUrl和Properties获取连接
除了通过指定的jdbcUrl获取连接，还可以使用Properties指定建立连接时的参数，如下所示：
```java
public Connection getConn() throws Exception{
Class.forName("com.taosdata.jdbc.TSDBDriver");
String jdbcUrl = "jdbc:TAOS://127.0.0.1:0/log?user=root&password=taosdata";
String jdbcUrl = "jdbc:TAOS://taosdemo.com:6030/test?user=root&password=taosdata";
Properties connProps = new Properties();
connProps.setProperty(TSDBDriver.PROPERTY_KEY_USER, "root");
connProps.setProperty(TSDBDriver.PROPERTY_KEY_PASSWORD, "taosdata");
connProps.setProperty(TSDBDriver.PROPERTY_KEY_CONFIG_DIR, "/etc/taos");
connProps.setProperty(TSDBDriver.PROPERTY_KEY_CHARSET, "UTF-8");
connProps.setProperty(TSDBDriver.PROPERTY_KEY_LOCALE, "en_US.UTF-8");
connProps.setProperty(TSDBDriver.PROPERTY_KEY_TIME_ZONE, "UTC-8");
......@@ -111,16 +107,39 @@ public Connection getConn() throws Exception{
return conn;
}
```
以上示例，建立一个到hostname为taosdemo.com，端口为6030，数据库名为test的连接。这个连接在url中指定了用户名(user)为root，密码（password）为taosdata，并在connProps中指定了使用的字符集、语言环境、时区等信息。
properties中的配置参数如下：
* TSDBDriver.PROPERTY_KEY_USER：登录 TDengine 用户名，默认值 root。
* TSDBDriver.PROPERTY_KEY_PASSWORD：用户登录密码，默认值 taosdata。
* TSDBDriver.PROPERTY_KEY_CONFIG_DIR：客户端配置文件目录路径，Linux OS 上默认值 /etc/taos ，Windows OS 上默认值 C:/TDengine/cfg。
* TSDBDriver.PROPERTY_KEY_CHARSET：客户端使用的字符集，默认值为系统字符集。
* TSDBDriver.PROPERTY_KEY_LOCALE：客户端语言环境，默认值系统当前 locale。
* TSDBDriver.PROPERTY_KEY_TIME_ZONE：客户端使用的时区，默认值为系统当前时区。
#### 使用客户端配置文件建立连接
当使用JDBC连接TDengine集群时，可以使用客户端配置文件，在客户端配置文件中指定集群的firstEp、secondEp参数。
如下所示：
1. 在java中不指定hostname和port
```java
public Connection getConn() throws Exception{
Class.forName("com.taosdata.jdbc.TSDBDriver");
String jdbcUrl = "jdbc:TAOS://:/test?user=root&password=taosdata";
Properties connProps = new Properties();
connProps.setProperty(TSDBDriver.PROPERTY_KEY_CHARSET, "UTF-8");
connProps.setProperty(TSDBDriver.PROPERTY_KEY_LOCALE, "en_US.UTF-8");
connProps.setProperty(TSDBDriver.PROPERTY_KEY_TIME_ZONE, "UTC-8");
Connection conn = DriverManager.getConnection(jdbcUrl, connProps);
return conn;
}
```
2. 在配置文件中指定firstEp和secondEp
```
# first fully qualified domain name (FQDN) for TDengine system
firstEp cluster_node1:6030
3. 客户端配置文件 taos.cfg
linux 系统默认配置文件为 /var/lib/taos/taos.cfg，windows 系统默认配置文件路径为 C:\TDengine\cfg\taos.cfg。
```properties
# client default username
# defaultUser root
# client default password
# defaultPass taosdata
# second fully qualified domain name (FQDN) for TDengine system, for cluster only
secondEp cluster_node2:6030
# default system charset
# charset UTF-8
......@@ -128,6 +147,19 @@ public Connection getConn() throws Exception{
# system locale
# locale en_US.UTF-8
```
以上示例，jdbc会使用客户端的配置文件，建立到hostname为cluster_node1，端口为6030，数据库名为test的连接。当集群中firstEp节点失效时，JDBC会尝试使用secondEp连接集群。
TDengine中，只要保证firstEp和secondEp中一个节点有效，就可以正常建立到集群的连接。
> 注意：这里的配置文件指的是调用JDBC Connector的应用程序所在机器上的配置文件，Linux OS 上默认值 /etc/taos/taos.cfg ，Windows OS 上默认值 C://TDengine/cfg/taos.cfg。
#### 配置参数的优先级
通过以上3种方式获取连接，如果配置参数在url、Properties、客户端配置文件中有重复，则参数的`优先级由高到低`分别如下：
1. JDBC URL 参数，如上所述，可以在 JDBC URL 的参数中指定。
2. Properties connProps
3. 客户端配置文件 taos.cfg
例如：在url中指定了password为taosdata，在Properties中指定了password为taosdemo，那么，JDBC会使用url中的password建立连接。
> 更多详细配置请参考[客户端配置][13]
### 创建数据库和表
......
packaging/cfg/taos.cfg
浏览文件 @ 5f21e2e7
......@@ -6,10 +6,7 @@
########################################################
# first fully qualified domain name (FQDN) for TDengine system
# firstEp hostname1:6030
# second fully qualified domain name (FQDN) for TDengine system, for cluster only
# secondEp cluster_hostname2:6030
# firstEp hostname:6030
# local fully qualified domain name (FQDN)
# fqdn hostname
......@@ -29,12 +26,18 @@
# number of threads per CPU core
# numOfThreadsPerCore 1.0
# the proportion of total threads responsible for query
# ratioOfQueryThreads 0.5
# number of management nodes in the system
# numOfMnodes 3
# enable/disable backuping vnode directory when removing dnode
# vnodeBak 1
# if report installation / use information
# telemetryReporting 1
# enable/disable load balancing
# balance 1
......@@ -89,9 +92,6 @@
# max number of tables per vnode
# maxTablesPerVnode 1000000
# step size of increasing table number in a vnode
# tableIncStepPerVnode 1000
# cache block size (Mbyte)
# cache 16
......@@ -110,6 +110,9 @@
# maximum rows of records in file block
# maxRows 4096
# the number of acknowledgments required for successful data writing
# quorum 1
# enable/disable compression
# comp 2
......@@ -122,15 +125,6 @@
# number of replications, for cluster only
# replica 1
# mqtt hostname
# mqttHostName test.mosquitto.org
# mqtt port
# mqttPort 1883
# mqtt topic
# mqttTopic /test
# the compressed rpc message, option:
# -1 (no compression)
# 0 (all message compressed),
......@@ -167,12 +161,12 @@
# stop writing data when the disk size of the log folder is less than this value
# minimalDataDirGB 0.1
# One mnode is equal to the number of vnode consumed
# mnodeEqualVnodeNum 4
# enbale/disable http service
# http 1
# enable/disable muqq service
# mqtt 0
# enable/disable system monitor
# monitor 1
......@@ -189,11 +183,12 @@
# max number of rows per log filters
# numOfLogLines 10000000
# enable/disable async log
# asyncLog 1
# time of keeping log files, days
# logKeepDays 0
# enable/disable async log
# asyncLog 1
# The following parameters are used for debug purpose only.
# debugFlag 8 bits mask: FILE-SCREEN-UNUSED-HeartBeat-DUMP-TRACE_WARN-ERROR
......@@ -231,18 +226,12 @@
# debug flag for JNI
# jniDebugflag 131
# debug flag for ODBC
# odbcDebugflag 131
# debug flag for storage
# uDebugflag 131
# debug flag for http server
# httpDebugFlag 131
# debug flag for mqtt
# mqttDebugFlag 131
# debug flag for monitor
# monitorDebugFlag 131
......@@ -255,6 +244,9 @@
# debug flag for http server
# tsdbDebugFlag 131
# debug flag for continue query
# cqDebugFlag 131
# enable/disable recording the SQL in taos client
# tscEnableRecordSql 0
......
packaging/docker/Dockerfile
浏览文件 @ 5f21e2e7
FROM centos:7
FROM ubuntu:16
WORKDIR /root
......
snap/snapcraft.yaml
浏览文件 @ 5f21e2e7
name: tdengine
base: core18
version: 'RELEASE_VERSION'
version: '2.0.5.1'
icon: snap/gui/t-dengine.svg
summary: an open-source big data platform designed and optimized for IoT.
description: |
......@@ -72,7 +72,7 @@ parts:
- usr/bin/taosd
- usr/bin/taos
- usr/bin/taosdemo
- usr/lib/libtaos.so.RELEASE_VERSION
- usr/lib/libtaos.so.2.0.5.1
- usr/lib/libtaos.so.1
- usr/lib/libtaos.so
......
src/balance/src/balance.c
浏览文件 @ 5f21e2e7
......@@ -937,6 +937,7 @@ static int32_t balanceRetrieveScores(SShowObj *pShow, char *data, int32_t rows,
mnodeDecDnodeRef(pDnode);
}
mnodeVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);
pShow->numOfReads += numOfRows;
return numOfRows;
}
......@@ -960,8 +961,13 @@ static void balanceMonitorDnodeModule() {
mLInfo("dnode:%d, numOfMnodes:%d expect:%d, create mnode in this dnode", pDnode->dnodeId, numOfMnodes, tsNumOfMnodes);
mnodeCreateMnode(pDnode->dnodeId, pDnode->dnodeEp, true);
#if 0
// Only create one mnode each time
return;
#else
numOfMnodes = mnodeGetMnodesNum();
if (numOfMnodes >= tsNumOfMnodes) return;
#endif
}
}
......
src/client/inc/tscLocalMerge.h
浏览文件 @ 5f21e2e7
......@@ -72,17 +72,10 @@ typedef struct SLocalReducer {
bool orderPrjOnSTable; // projection query on stable
} SLocalReducer;
typedef struct SSubqueryState {
int32_t numOfRemain; // the number of remain unfinished subquery
int32_t numOfTotal; // the number of total sub-queries
uint64_t numOfRetrievedRows; // total number of points in this query
} SSubqueryState;
typedef struct SRetrieveSupport {
tExtMemBuffer ** pExtMemBuffer; // for build loser tree
tOrderDescriptor *pOrderDescriptor;
SColumnModel * pFinalColModel; // colModel for final result
SSubqueryState * pState;
int32_t subqueryIndex; // index of current vnode in vnode list
SSqlObj * pParentSql;
tFilePage * localBuffer; // temp buffer, there is a buffer for each vnode to
......
src/client/inc/tscUtil.h
浏览文件 @ 5f21e2e7
......@@ -66,7 +66,6 @@ typedef struct STidTags {
#pragma pack(pop)
typedef struct SJoinSupporter {
SSubqueryState* pState;
SSqlObj* pObj; // parent SqlObj
int32_t subqueryIndex; // index of sub query
SInterval interval;
......@@ -207,8 +206,6 @@ void tscTagCondRelease(STagCond* pCond);
void tscGetSrcColumnInfo(SSrcColumnInfo* pColInfo, SQueryInfo* pQueryInfo);
void tscSetFreeHeatBeat(STscObj* pObj);
bool tscShouldFreeHeartBeat(SSqlObj* pHb);
bool tscShouldBeFreed(SSqlObj* pSql);
STableMetaInfo* tscGetTableMetaInfoFromCmd(SSqlCmd *pCmd, int32_t subClauseIndex, int32_t tableIndex);
......
src/client/inc/tsclient.h
浏览文件 @ 5f21e2e7
......@@ -80,6 +80,8 @@ enum {
DATA_FROM_DATA_FILE = 2,
};
typedef void (*__async_cb_func_t)(void *param, TAOS_RES *tres, int32_t numOfRows);
typedef struct STableComInfo {
uint8_t numOfTags;
uint8_t precision;
......@@ -226,7 +228,7 @@ typedef struct STableDataBlocks {
typedef struct SQueryInfo {
int16_t command; // the command may be different for each subclause, so keep it seperately.
uint32_t type; // query/insert type
// TODO refactor
STimeWindow window; // query time window
SInterval interval;
......@@ -334,6 +336,12 @@ typedef struct STscObj {
T_REF_DECLARE()
} STscObj;
typedef struct SSubqueryState {
int32_t numOfRemain; // the number of remain unfinished subquery
int32_t numOfSub; // the number of total sub-queries
uint64_t numOfRetrievedRows; // total number of points in this query
} SSubqueryState;
typedef struct SSqlObj {
void *signature;
pthread_t owner; // owner of sql object, by which it is executed
......@@ -355,10 +363,11 @@ typedef struct SSqlObj {
tsem_t rspSem;
SSqlCmd cmd;
SSqlRes res;
uint16_t numOfSubs;
SSubqueryState subState;
struct SSqlObj **pSubs;
struct SSqlObj * prev, *next;
struct SSqlObj *prev, *next;
struct SSqlObj **self;
} SSqlObj;
......@@ -433,19 +442,20 @@ void tscPartiallyFreeSqlObj(SSqlObj *pSql);
* @param pObj
*/
void tscFreeSqlObj(SSqlObj *pSql);
void tscFreeSqlObjInCache(void *pSql);
void tscFreeRegisteredSqlObj(void *pSql);
void tscCloseTscObj(STscObj *pObj);
// todo move to taos? or create a new file: taos_internal.h
TAOS *taos_connect_a(char *ip, char *user, char *pass, char *db, uint16_t port, void (*fp)(void *, TAOS_RES *, int),
void *param, void **taos);
void waitForQueryRsp(void *param, TAOS_RES *tres, int code) ;
TAOS_RES* taos_query_h(TAOS* taos, const char *sqlstr, TAOS_RES** res);
void waitForQueryRsp(void *param, TAOS_RES *tres, int code);
void doAsyncQuery(STscObj *pObj, SSqlObj *pSql, void (*fp)(), void *param, const char *sqlstr, size_t sqlLen);
void doAsyncQuery(STscObj *pObj, SSqlObj *pSql, __async_cb_func_t fp, void *param, const char *sqlstr, size_t sqlLen);
void tscProcessMultiVnodesImportFromFile(SSqlObj *pSql);
void tscKillSTableQuery(SSqlObj *pSql);
void tscInitResObjForLocalQuery(SSqlObj *pObj, int32_t numOfRes, int32_t rowLen);
bool tscIsUpdateQuery(SSqlObj* pSql);
bool tscHasReachLimitation(SQueryInfo *pQueryInfo, SSqlRes *pRes);
......@@ -510,8 +520,6 @@ extern SRpcCorEpSet tscMgmtEpSet;
extern int (*tscBuildMsg[TSDB_SQL_MAX])(SSqlObj *pSql, SSqlInfo *pInfo);
typedef void (*__async_cb_func_t)(void *param, TAOS_RES *tres, int numOfRows);
int32_t tscCompareTidTags(const void* p1, const void* p2);
void tscBuildVgroupTableInfo(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo, SArray* tables);
......
src/client/src/tscAsync.c
浏览文件 @ 5f21e2e7
......@@ -40,7 +40,7 @@ static void tscProcessAsyncRetrieveImpl(void *param, TAOS_RES *tres, int numOfRo
static void tscAsyncFetchRowsProxy(void *param, TAOS_RES *tres, int numOfRows);
static void tscAsyncFetchSingleRowProxy(void *param, TAOS_RES *tres, int numOfRows);
void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const char* sqlstr, size_t sqlLen) {
void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, __async_cb_func_t fp, void* param, const char* sqlstr, size_t sqlLen) {
SSqlCmd* pCmd = &pSql->cmd;
pSql->signature = pSql;
......@@ -361,15 +361,6 @@ void tscProcessFetchRow(SSchedMsg *pMsg) {
(*pSql->fetchFp)(pSql->param, pSql, pRes->tsrow);
}
void tscProcessAsyncRes(SSchedMsg *pMsg) {
SSqlObj *pSql = (SSqlObj *)pMsg->ahandle;
SSqlRes *pRes = &pSql->res;
assert(pSql->fp != NULL && pSql->fetchFp != NULL);
pSql->fp = pSql->fetchFp;
(*pSql->fp)(pSql->param, pSql, pRes->code);
}
// this function will be executed by queue task threads, so the terrno is not valid
static void tscProcessAsyncError(SSchedMsg *pMsg) {
void (*fp)() = pMsg->ahandle;
......@@ -393,22 +384,15 @@ void tscQueueAsyncRes(SSqlObj *pSql) {
if (pSql == NULL || pSql->signature != pSql) {
tscDebug("%p SqlObj is freed, not add into queue async res", pSql);
return;
} else {
tscError("%p add into queued async res, code:%s", pSql, tstrerror(pSql->res.code));
}
SSchedMsg schedMsg = { 0 };
schedMsg.fp = tscProcessAsyncRes;
schedMsg.ahandle = pSql;
schedMsg.thandle = (void *)1;
schedMsg.msg = NULL;
taosScheduleTask(tscQhandle, &schedMsg);
}
tscError("%p add into queued async res, code:%s", pSql, tstrerror(pSql->res.code));
void tscProcessAsyncFree(SSchedMsg *pMsg) {
SSqlObj *pSql = (SSqlObj *)pMsg->ahandle;
tscDebug("%p sql is freed", pSql);
taos_free_result(pSql);
SSqlRes *pRes = &pSql->res;
assert(pSql->fp != NULL && pSql->fetchFp != NULL);
pSql->fp = pSql->fetchFp;
(*pSql->fp)(pSql->param, pSql, pRes->code);
}
int tscSendMsgToServer(SSqlObj *pSql);
......
src/client/src/tscLocalMerge.c
浏览文件 @ 5f21e2e7
......@@ -639,7 +639,7 @@ int32_t tscLocalReducerEnvCreate(SSqlObj *pSql, tExtMemBuffer ***pMemBuffer, tOr
SQueryInfo * pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
(*pMemBuffer) = (tExtMemBuffer **)malloc(POINTER_BYTES * pSql->numOfSubs);
(*pMemBuffer) = (tExtMemBuffer **)malloc(POINTER_BYTES * pSql->subState.numOfSub);
if (*pMemBuffer == NULL) {
tscError("%p failed to allocate memory", pSql);
pRes->code = TSDB_CODE_TSC_OUT_OF_MEMORY;
......@@ -742,6 +742,7 @@ void tscLocalReducerEnvDestroy(tExtMemBuffer **pMemBuffer, tOrderDescriptor *pDe
int32_t numOfVnodes) {
destroyColumnModel(pFinalModel);
tOrderDescDestroy(pDesc);
for (int32_t i = 0; i < numOfVnodes; ++i) {
pMemBuffer[i] = destoryExtMemBuffer(pMemBuffer[i]);
}
......
src/client/src/tscProfile.c
浏览文件 @ 5f21e2e7
......@@ -151,10 +151,12 @@ void tscKillQuery(STscObj *pObj, uint32_t killId) {
pthread_mutex_unlock(&pObj->mutex);
if (pSql == NULL) return;
if (pSql == NULL) {
tscError("failed to kill query, id:%d, it may have completed/terminated", killId);
} else {
tscDebug("%p query is killed, queryId:%d", pSql, killId);
taos_stop_query(pSql);
}
}
void tscAddIntoStreamList(SSqlStream *pStream) {
......@@ -242,6 +244,7 @@ int tscBuildQueryStreamDesc(void *pMsg, STscObj *pObj) {
pQdesc->stime = htobe64(pSql->stime);
pQdesc->queryId = htonl(pSql->queryId);
pQdesc->useconds = htobe64(pSql->res.useconds);
pQdesc->qHandle = htobe64(pSql->res.qhandle);
pHeartbeat->numOfQueries++;
pQdesc++;
......
src/client/src/tscSQLParser.c
浏览文件 @ 5f21e2e7
......@@ -1200,6 +1200,10 @@ int32_t setObjFullName(char* fullName, const char* account, SStrToken* pDB, SStr
return (totalLen < TSDB_TABLE_FNAME_LEN) ? TSDB_CODE_SUCCESS : TSDB_CODE_TSC_INVALID_SQL;
}
static void tscInsertPrimaryTSSourceColumn(SQueryInfo* pQueryInfo, SColumnIndex* pIndex) {
SColumnIndex tsCol = {.tableIndex = pIndex->tableIndex, .columnIndex = PRIMARYKEY_TIMESTAMP_COL_INDEX};
tscColumnListInsert(pQueryInfo->colList, &tsCol);
}
static int32_t handleArithmeticExpr(SSqlCmd* pCmd, int32_t clauseIndex, int32_t exprIndex, tSQLExprItem* pItem) {
const char* msg1 = "invalid column name, or illegal column type";
const char* msg2 = "invalid arithmetic expression in select clause";
......@@ -1275,6 +1279,8 @@ static int32_t handleArithmeticExpr(SSqlCmd* pCmd, int32_t clauseIndex, int32_t
addExprParams(pExpr, c, TSDB_DATA_TYPE_BINARY, (int32_t)len, index.tableIndex);
insertResultField(pQueryInfo, exprIndex, &columnList, sizeof(double), TSDB_DATA_TYPE_DOUBLE, pExpr->aliasName, pExpr);
// add ts column
tscInsertPrimaryTSSourceColumn(pQueryInfo, &index);
tbufCloseWriter(&bw);
taosArrayDestroy(colList);
......@@ -1317,10 +1323,6 @@ static int32_t handleArithmeticExpr(SSqlCmd* pCmd, int32_t clauseIndex, int32_t
return TSDB_CODE_SUCCESS;
}
static void tscInsertPrimaryTSSourceColumn(SQueryInfo* pQueryInfo, SColumnIndex* pIndex) {
SColumnIndex tsCol = {.tableIndex = pIndex->tableIndex, .columnIndex = PRIMARYKEY_TIMESTAMP_COL_INDEX};
tscColumnListInsert(pQueryInfo->colList, &tsCol);
}
static void addProjectQueryCol(SQueryInfo* pQueryInfo, int32_t startPos, SColumnIndex* pIndex, tSQLExprItem* pItem) {
SSqlExpr* pExpr = doAddProjectCol(pQueryInfo, startPos, pIndex->columnIndex, pIndex->tableIndex);
......@@ -1603,8 +1605,8 @@ int32_t addProjectionExprAndResultField(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, t
return TSDB_CODE_SUCCESS;
}
static int32_t setExprInfoForFunctions(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SSchema* pSchema, SConvertFunc cvtFunc, char* aliasName,
int32_t resColIdx, SColumnIndex* pColIndex) {
static int32_t setExprInfoForFunctions(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SSchema* pSchema, SConvertFunc cvtFunc,
char* aliasName, int32_t resColIdx, SColumnIndex* pColIndex, bool finalResult) {
const char* msg1 = "not support column types";
int16_t type = 0;
......@@ -1650,8 +1652,13 @@ static int32_t setExprInfoForFunctions(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SS
SColumnIndex index = {.tableIndex = pColIndex->tableIndex, .columnIndex = PRIMARYKEY_TIMESTAMP_COL_INDEX};
tscColumnListInsert(pQueryInfo->colList, &index);
// if it is not in the final result, do not add it
SColumnList ids = getColumnList(1, pColIndex->tableIndex, pColIndex->columnIndex);
if (finalResult) {
insertResultField(pQueryInfo, resColIdx, &ids, bytes, (int8_t)type, columnName, pExpr);
} else {
tscColumnListInsert(pQueryInfo->colList, &(ids.ids[0]));
}
return TSDB_CODE_SUCCESS;
}
......@@ -1926,7 +1933,7 @@ int32_t addExprAndResultField(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, int32_t col
for (int32_t j = 0; j < tscGetNumOfColumns(pTableMetaInfo->pTableMeta); ++j) {
index.columnIndex = j;
if (setExprInfoForFunctions(pCmd, pQueryInfo, pSchema, cvtFunc, pItem->aliasName, colIndex++, &index) != 0) {
if (setExprInfoForFunctions(pCmd, pQueryInfo, pSchema, cvtFunc, pItem->aliasName, colIndex++, &index, finalResult) != 0) {
return TSDB_CODE_TSC_INVALID_SQL;
}
}
......@@ -1943,7 +1950,8 @@ int32_t addExprAndResultField(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, int32_t col
if ((index.columnIndex >= tscGetNumOfColumns(pTableMetaInfo->pTableMeta)) || (index.columnIndex < 0)) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg6);
}
if (setExprInfoForFunctions(pCmd, pQueryInfo, pSchema, cvtFunc, pItem->aliasName, colIndex + i, &index) != 0) {
if (setExprInfoForFunctions(pCmd, pQueryInfo, pSchema, cvtFunc, pItem->aliasName, colIndex + i, &index, finalResult) != 0) {
return TSDB_CODE_TSC_INVALID_SQL;
}
......@@ -1980,7 +1988,7 @@ int32_t addExprAndResultField(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, int32_t col
for (int32_t i = 0; i < tscGetNumOfColumns(pTableMetaInfo->pTableMeta); ++i) {
SColumnIndex index = {.tableIndex = j, .columnIndex = i};
if (setExprInfoForFunctions(pCmd, pQueryInfo, pSchema, cvtFunc, pItem->aliasName, colIndex, &index) != 0) {
if (setExprInfoForFunctions(pCmd, pQueryInfo, pSchema, cvtFunc, pItem->aliasName, colIndex, &index, finalResult) != 0) {
return TSDB_CODE_TSC_INVALID_SQL;
}
......
src/client/src/tscServer.c
浏览文件 @ 5f21e2e7
......@@ -24,7 +24,6 @@
#include "tschemautil.h"
#include "tsclient.h"
#include "ttimer.h"
#include "tutil.h"
#include "tlockfree.h"
SRpcCorEpSet tscMgmtEpSet;
......@@ -198,16 +197,20 @@ void tscProcessActivityTimer(void *handle, void *tmrId) {
return;
}
if (tscShouldFreeHeartBeat(pHB)) {
tscDebug("%p free HB object and release connection", pHB);
pObj->pHb = 0;
taos_free_result(pHB);
} else {
void** p = taosCacheAcquireByKey(tscObjCache, &pHB, sizeof(TSDB_CACHE_PTR_TYPE));
if (p == NULL) {
tscWarn("%p HB object has been released already", pHB);
return;
}
assert(*pHB->self == pHB);
int32_t code = tscProcessSql(pHB);
taosCacheRelease(tscObjCache, (void**) &p, false);
if (code != TSDB_CODE_SUCCESS) {
tscError("%p failed to sent HB to server, reason:%s", pHB, tstrerror(code));
}
}
}
int tscSendMsgToServer(SSqlObj *pSql) {
......@@ -464,35 +467,6 @@ int tscProcessSql(SSqlObj *pSql) {
return doProcessSql(pSql);
}
void tscKillSTableQuery(SSqlObj *pSql) {
SSqlCmd* pCmd = &pSql->cmd;
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
if (!tscIsTwoStageSTableQuery(pQueryInfo, 0)) {
return;
}
pSql->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
for (int i = 0; i < pSql->numOfSubs; ++i) {
// NOTE: pSub may have been released already here
SSqlObj *pSub = pSql->pSubs[i];
if (pSub == NULL) {
continue;
}
pSub->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
if (pSub->pRpcCtx != NULL) {
rpcCancelRequest(pSub->pRpcCtx);
pSub->pRpcCtx = NULL;
}
tscQueueAsyncRes(pSub); // async res? not other functions?
}
tscDebug("%p super table query cancelled", pSql);
}
int tscBuildFetchMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
SRetrieveTableMsg *pRetrieveMsg = (SRetrieveTableMsg *) pSql->cmd.payload;
pRetrieveMsg->qhandle = htobe64(pSql->res.qhandle);
......@@ -1451,7 +1425,7 @@ int tscProcessLocalRetrieveRsp(SSqlObj *pSql) {
int tscProcessRetrieveLocalMergeRsp(SSqlObj *pSql) {
SSqlRes *pRes = &pSql->res;
SSqlCmd *pCmd = &pSql->cmd;
SSqlCmd* pCmd = &pSql->cmd;
int32_t code = pRes->code;
if (pRes->code != TSDB_CODE_SUCCESS) {
......@@ -1494,6 +1468,7 @@ int tscBuildConnectMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
SCMConnectMsg *pConnect = (SCMConnectMsg*)pCmd->payload;
// TODO refactor full_name
char *db; // ugly code to move the space
db = strstr(pObj->db, TS_PATH_DELIMITER);
db = (db == NULL) ? pObj->db : db + 1;
......@@ -1501,6 +1476,9 @@ int tscBuildConnectMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
tstrncpy(pConnect->clientVersion, version, sizeof(pConnect->clientVersion));
tstrncpy(pConnect->msgVersion, "", sizeof(pConnect->msgVersion));
pConnect->pid = htonl(taosGetPId());
taosGetCurrentAPPName(pConnect->appName, NULL);
return TSDB_CODE_SUCCESS;
}
......@@ -1653,6 +1631,10 @@ int tscBuildHeartBeatMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
SCMHeartBeatMsg *pHeartbeat = (SCMHeartBeatMsg *)pCmd->payload;
pHeartbeat->numOfQueries = numOfQueries;
pHeartbeat->numOfStreams = numOfStreams;
pHeartbeat->pid = htonl(taosGetPId());
taosGetCurrentAPPName(pHeartbeat->appName, NULL);
int msgLen = tscBuildQueryStreamDesc(pHeartbeat, pObj);
pthread_mutex_unlock(&pObj->mutex);
......
src/client/src/tscSql.c
浏览文件 @ 5f21e2e7
......@@ -20,14 +20,15 @@
#include "tcache.h"
#include "tnote.h"
#include "trpc.h"
#include "ttimer.h"
#include "tscLog.h"
#include "tscSubquery.h"
#include "tscUtil.h"
#include "tsclient.h"
#include "ttokendef.h"
#include "tutil.h"
#include "ttimer.h"
#include "tscProfile.h"
#include "ttimer.h"
static bool validImpl(const char* str, size_t maxsize) {
if (str == NULL) {
......@@ -257,10 +258,18 @@ TAOS *taos_connect_a(char *ip, char *user, char *pass, char *db, uint16_t port,
void taos_close(TAOS *taos) {
STscObj *pObj = (STscObj *)taos;
if (pObj == NULL || pObj->signature != pObj) {
if (pObj == NULL) {
tscDebug("(null) try to free tscObj and close dnodeConn");
return;
}
tscDebug("%p try to free tscObj and close dnodeConn:%p", pObj, pObj->pDnodeConn);
if (pObj->signature != pObj) {
tscDebug("%p already closed or invalid tscObj", pObj);
return;
}
// make sure that the close connection can only be executed once.
pObj->signature = NULL;
taosTmrStopA(&(pObj->pTimer));
......@@ -299,7 +308,7 @@ static void waitForRetrieveRsp(void *param, TAOS_RES *tres, int numOfRows) {
tsem_post(&pSql->rspSem);
}
TAOS_RES* taos_query_c(TAOS *taos, const char *sqlstr, uint32_t sqlLen) {
TAOS_RES* taos_query_c(TAOS *taos, const char *sqlstr, uint32_t sqlLen, TAOS_RES** res) {
STscObj *pObj = (STscObj *)taos;
if (pObj == NULL || pObj->signature != pObj) {
terrno = TSDB_CODE_TSC_DISCONNECTED;
......@@ -324,12 +333,20 @@ TAOS_RES* taos_query_c(TAOS *taos, const char *sqlstr, uint32_t sqlLen) {
tsem_init(&pSql->rspSem, 0, 0);
doAsyncQuery(pObj, pSql, waitForQueryRsp, taos, sqlstr, sqlLen);
if (res != NULL) {
*res = pSql;
}
tsem_wait(&pSql->rspSem);
return pSql;
}
TAOS_RES* taos_query(TAOS *taos, const char *sqlstr) {
return taos_query_c(taos, sqlstr, (uint32_t)strlen(sqlstr));
return taos_query_c(taos, sqlstr, (uint32_t)strlen(sqlstr), NULL);
}
TAOS_RES* taos_query_h(TAOS* taos, const char *sqlstr, TAOS_RES** res) {
return taos_query_c(taos, sqlstr, (uint32_t) strlen(sqlstr), res);
}
int taos_result_precision(TAOS_RES *res) {
......@@ -463,6 +480,7 @@ TAOS_ROW taos_fetch_row(TAOS_RES *res) {
SSqlRes *pRes = &pSql->res;
if (pRes->qhandle == 0 ||
pRes->code == TSDB_CODE_TSC_QUERY_CANCELLED ||
pCmd->command == TSDB_SQL_RETRIEVE_EMPTY_RESULT ||
pCmd->command == TSDB_SQL_INSERT) {
return NULL;
......@@ -526,7 +544,7 @@ int taos_fetch_block(TAOS_RES *res, TAOS_ROW *rows) {
pRes->numOfClauseTotal = 0;
pRes->rspType = 0;
pSql->numOfSubs = 0;
pSql->subState.numOfSub = 0;
taosTFree(pSql->pSubs);
assert(pSql->fp == NULL);
......@@ -680,6 +698,45 @@ int* taos_fetch_lengths(TAOS_RES *res) {
char *taos_get_client_info() { return version; }
static void tscKillSTableQuery(SSqlObj *pSql) {
SSqlCmd* pCmd = &pSql->cmd;
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
if (!tscIsTwoStageSTableQuery(pQueryInfo, 0)) {
return;
}
// set the master sqlObj flag to cancel query
pSql->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
for (int i = 0; i < pSql->subState.numOfSub; ++i) {
// NOTE: pSub may have been released already here
SSqlObj *pSub = pSql->pSubs[i];
if (pSub == NULL) {
continue;
}
void** p = taosCacheAcquireByKey(tscObjCache, &pSub, sizeof(TSDB_CACHE_PTR_TYPE));
if (p == NULL) {
continue;
}
SSqlObj* pSubObj = (SSqlObj*) (*p);
assert(pSubObj->self == (SSqlObj**) p);
pSubObj->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
if (pSubObj->pRpcCtx != NULL) {
rpcCancelRequest(pSubObj->pRpcCtx);
pSubObj->pRpcCtx = NULL;
}
tscQueueAsyncRes(pSubObj);
taosCacheRelease(tscObjCache, (void**) &p, false);
}
tscDebug("%p super table query cancelled", pSql);
}
void taos_stop_query(TAOS_RES *res) {
SSqlObj *pSql = (SSqlObj *)res;
if (pSql == NULL || pSql->signature != pSql) {
......@@ -689,23 +746,26 @@ void taos_stop_query(TAOS_RES *res) {
tscDebug("%p start to cancel query", res);
SSqlCmd *pCmd = &pSql->cmd;
// TODO there are multi-thread problem.
// It may have been released by the other thread already.
// The ref count may fix this problem.
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
// set the error code for master pSqlObj firstly
pSql->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
if (tscIsTwoStageSTableQuery(pQueryInfo, 0)) {
assert(pSql->pRpcCtx == NULL);
tscKillSTableQuery(pSql);
} else {
if (pSql->cmd.command < TSDB_SQL_LOCAL) {
/*
* There is multi-thread problem here, since pSql->pRpcCtx may have been
* reset and freed in the processMsgFromServer function, and causes the invalid
* write problem for rpcCancelRequest.
*/
if (pSql->pRpcCtx != NULL) {
rpcCancelRequest(pSql->pRpcCtx);
pSql->pRpcCtx = NULL;
}
tscQueueAsyncRes(pSql);
}
}
......@@ -834,6 +894,8 @@ int taos_validate_sql(TAOS *taos, const char *sql) {
pSql->fp = asyncCallback;
pSql->fetchFp = asyncCallback;
pSql->param = pSql;
registerSqlObj(pSql);
int code = tsParseSql(pSql, true);
if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
tsem_wait(&pSql->rspSem);
......
src/client/src/tscStream.c
浏览文件 @ 5f21e2e7
......@@ -274,7 +274,7 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
taosCacheRelease(tscMetaCache, (void**)&(pTableMetaInfo->pTableMeta), false);
tscFreeSqlResult(pSql);
taosTFree(pSql->pSubs);
pSql->numOfSubs = 0;
pSql->subState.numOfSub = 0;
taosTFree(pTableMetaInfo->vgroupList);
tscSetNextLaunchTimer(pStream, pSql);
}
......
src/client/src/tscSubquery.c
浏览文件 @ 5f21e2e7
此差异已折叠。
src/client/src/tscSystem.c
浏览文件 @ 5f21e2e7
......@@ -141,7 +141,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, NULL, "tableMeta");
tscObjCache = taosCacheInit(TSDB_CACHE_PTR_KEY, refreshTime / 2, false, tscFreeSqlObjInCache, "sqlObj");
tscObjCache = taosCacheInit(TSDB_CACHE_PTR_KEY, refreshTime / 2, false, tscFreeRegisteredSqlObj, "sqlObj");
}
tscDebug("client is initialized successfully");
......
src/client/src/tscUtil.c
浏览文件 @ 5f21e2e7
......@@ -360,26 +360,26 @@ void tscPartiallyFreeSqlObj(SSqlObj* pSql) {
tscFreeSqlResult(pSql);
taosTFree(pSql->pSubs);
pSql->numOfSubs = 0;
pSql->subState.numOfSub = 0;
pSql->self = 0;
tscResetSqlCmdObj(pCmd, false);
}
static UNUSED_FUNC void tscFreeSubobj(SSqlObj* pSql) {
if (pSql->numOfSubs == 0) {
static void tscFreeSubobj(SSqlObj* pSql) {
if (pSql->subState.numOfSub == 0) {
return;
}
tscDebug("%p start to free sub SqlObj, numOfSub:%d", pSql, pSql->numOfSubs);
tscDebug("%p start to free sub SqlObj, numOfSub:%d", pSql, pSql->subState.numOfSub);
for(int32_t i = 0; i < pSql->numOfSubs; ++i) {
for(int32_t i = 0; i < pSql->subState.numOfSub; ++i) {
tscDebug("%p free sub SqlObj:%p, index:%d", pSql, pSql->pSubs[i], i);
taos_free_result(pSql->pSubs[i]);
pSql->pSubs[i] = NULL;
}
pSql->numOfSubs = 0;
pSql->subState.numOfSub = 0;
}
/**
......@@ -389,7 +389,7 @@ static UNUSED_FUNC void tscFreeSubobj(SSqlObj* pSql) {
*
* @param pSql
*/
void tscFreeSqlObjInCache(void *pSql) {
void tscFreeRegisteredSqlObj(void *pSql) {
assert(pSql != NULL);
SSqlObj** p = (SSqlObj**)pSql;
......@@ -415,7 +415,9 @@ void tscFreeSqlObj(SSqlObj* pSql) {
tscDebug("%p start to free sqlObj", pSql);
pSql->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
tscFreeSubobj(pSql);
tscPartiallyFreeSqlObj(pSql);
pSql->signature = NULL;
......@@ -1516,13 +1518,6 @@ void tscSetFreeHeatBeat(STscObj* pObj) {
pQueryInfo->type = TSDB_QUERY_TYPE_FREE_RESOURCE;
}
bool tscShouldFreeHeartBeat(SSqlObj* pHb) {
assert(pHb == pHb->signature);
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pHb->cmd, 0);
return pQueryInfo->type == TSDB_QUERY_TYPE_FREE_RESOURCE;
}
/*
* the following four kinds of SqlObj should not be freed
* 1. SqlObj for stream computing
......@@ -2291,7 +2286,7 @@ void tscTryQueryNextVnode(SSqlObj* pSql, __async_cb_func_t fp) {
*
* For super table join with projection query, if anyone of the subquery is exhausted, the query completed.
*/
pSql->numOfSubs = 0;
pSql->subState.numOfSub = 0;
pCmd->command = TSDB_SQL_SELECT;
tscResetForNextRetrieve(pRes);
......@@ -2323,7 +2318,7 @@ void tscTryQueryNextClause(SSqlObj* pSql, __async_cb_func_t fp) {
pRes->numOfTotal = num;
taosTFree(pSql->pSubs);
pSql->numOfSubs = 0;
pSql->subState.numOfSub = 0;
pSql->fp = fp;
tscDebug("%p try data in the next subclause:%d, total subclause:%d", pSql, pCmd->clauseIndex, pCmd->numOfClause);
......
src/common/src/tglobal.c
浏览文件 @ 5f21e2e7
......@@ -308,6 +308,8 @@ bool taosCfgDynamicOptions(char *msg) {
static void doInitGlobalConfig(void) {
osInit();
srand(taosSafeRand());
SGlobalCfg cfg = {0};
// ip address
......
go @ 8c58c512
比较 8d7bf743...8c58c512
Subproject commit 8d7bf743852897110cbdcc7c4322cd7a74d4167b
Subproject commit 8c58c512b6acda8bcdfa48fdc7140227b5221766
src/dnode/src/dnodeVWrite.c
浏览文件 @ 5f21e2e7
......@@ -210,12 +210,12 @@ void dnodeSendRpcVnodeWriteRsp(void *pVnode, void *param, int32_t code) {
static void *dnodeProcessWriteQueue(void *param) {
SWriteWorker *pWorker = (SWriteWorker *)param;
SWriteMsg *pWrite;
SWalHead *pHead;
SWriteMsg * pWrite;
SWalHead * pHead;
int32_t numOfMsgs;
int type;
void *pVnode, *item;
SRspRet *pRspRet;
void * pVnode, *item;
SRspRet * pRspRet;
dDebug("write worker:%d is running", pWorker->workerId);
......@@ -237,13 +237,18 @@ static void *dnodeProcessWriteQueue(void *param) {
pHead->msgType = pWrite->rpcMsg.msgType;
pHead->version = 0;
pHead->len = pWrite->contLen;
dDebug("%p, rpc msg:%s will be processed in vwrite queue", pWrite->rpcMsg.ahandle, taosMsg[pWrite->rpcMsg.msgType]);
dDebug("%p, rpc msg:%s will be processed in vwrite queue", pWrite->rpcMsg.ahandle,
taosMsg[pWrite->rpcMsg.msgType]);
} else {
pHead = (SWalHead *)item;
dTrace("%p, wal msg:%s will be processed in vwrite queue, version:%" PRIu64, pHead, taosMsg[pHead->msgType], pHead->version);
dTrace("%p, wal msg:%s will be processed in vwrite queue, version:%" PRIu64, pHead, taosMsg[pHead->msgType],
pHead->version);
}
int32_t code = vnodeProcessWrite(pVnode, type, pHead, pRspRet);
dTrace("%p, msg:%s is processed in vwrite queue, version:%" PRIu64 ", result:%s", pHead, taosMsg[pHead->msgType],
pHead->version, tstrerror(code));
if (pWrite) {
pWrite->rpcMsg.code = code;
if (code <= 0) pWrite->processedCount = 1;
......
src/inc/taosdef.h
浏览文件 @ 5f21e2e7
......@@ -253,8 +253,10 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size, void* buf
#define TSDB_COL_NAME_LEN 65
#define TSDB_MAX_SAVED_SQL_LEN TSDB_MAX_COLUMNS * 64
#define TSDB_MAX_SQL_LEN TSDB_PAYLOAD_SIZE
#define TSDB_MAX_SQL_SHOW_LEN 256
#define TSDB_MAX_ALLOWED_SQL_LEN (1*1024*1024U) // sql length should be less than 8mb
#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_APPNAME_LEN TSDB_UNI_LEN
#define TSDB_MAX_BYTES_PER_ROW 16384
#define TSDB_MAX_TAGS_LEN 16384
......@@ -282,7 +284,7 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size, void* buf
#define TSDB_SHELL_VNODE_BITS 24
#define TSDB_SHELL_SID_MASK 0xFF
#define TSDB_HTTP_TOKEN_LEN 20
#define TSDB_SHOW_SQL_LEN 64
#define TSDB_SHOW_SQL_LEN 512
#define TSDB_SLOW_QUERY_SQL_LEN 512
#define TSDB_MQTT_HOSTNAME_LEN 64
......
src/inc/taoserror.h
浏览文件 @ 5f21e2e7
......@@ -193,6 +193,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_VND_NO_DISK_PERMISSIONS, 0, 0x0506, "No write p
TAOS_DEFINE_ERROR(TSDB_CODE_VND_NO_SUCH_FILE_OR_DIR, 0, 0x0507, "Missing data file")
TAOS_DEFINE_ERROR(TSDB_CODE_VND_OUT_OF_MEMORY, 0, 0x0508, "Out of memory")
TAOS_DEFINE_ERROR(TSDB_CODE_VND_APP_ERROR, 0, 0x0509, "Unexpected generic error in vnode")
TAOS_DEFINE_ERROR(TSDB_CODE_VND_INVALID_VRESION_FILE, 0, 0x050A, "Invalid version file")
TAOS_DEFINE_ERROR(TSDB_CODE_VND_NOT_SYNCED, 0, 0x0511, "Database suspended")
TAOS_DEFINE_ERROR(TSDB_CODE_VND_NO_WRITE_AUTH, 0, 0x0512, "Write operation denied")
......@@ -247,6 +248,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_GRANT_CPU_LIMITED, 0, 0x080B, "CPU cores
// sync
TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INVALID_CONFIG, 0, 0x0900, "Invalid Sync Configuration")
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")
// wal
TAOS_DEFINE_ERROR(TSDB_CODE_WAL_APP_ERROR, 0, 0x1000, "Unexpected generic error in wal")
......
src/inc/taosmsg.h
浏览文件 @ 5f21e2e7
......@@ -305,6 +305,8 @@ typedef struct {
char clientVersion[TSDB_VERSION_LEN];
char msgVersion[TSDB_VERSION_LEN];
char db[TSDB_TABLE_FNAME_LEN];
char appName[TSDB_APPNAME_LEN];
int32_t pid;
} SCMConnectMsg;
typedef struct {
......@@ -746,6 +748,7 @@ typedef struct {
uint32_t queryId;
int64_t useconds;
int64_t stime;
uint64_t qHandle;
} SQueryDesc;
typedef struct {
......@@ -761,8 +764,10 @@ typedef struct {
typedef struct {
uint32_t connId;
int32_t pid;
int32_t numOfQueries;
int32_t numOfStreams;
char appName[TSDB_APPNAME_LEN];
char pData[];
} SCMHeartBeatMsg;
......
src/inc/twal.h
浏览文件 @ 5f21e2e7
......@@ -51,6 +51,7 @@ int walWrite(twalh, SWalHead *);
void walFsync(twalh);
int walRestore(twalh, void *pVnode, FWalWrite writeFp);
int walGetWalFile(twalh, char *name, uint32_t *index);
int64_t walGetVersion(twalh);
extern int wDebugFlag;
......
src/kit/shell/inc/shell.h
浏览文件 @ 5f21e2e7
......@@ -20,6 +20,7 @@
#include "taos.h"
#include "taosdef.h"
#include "stdbool.h"
#include "tsclient.h"
#define MAX_USERNAME_SIZE 64
#define MAX_DBNAME_SIZE 64
......
src/kit/shell/src/shellEngine.c
浏览文件 @ 5f21e2e7
......@@ -294,9 +294,7 @@ void shellRunCommandOnServer(TAOS *con, char command[]) {
st = taosGetTimestampUs();
TAOS_RES* pSql = taos_query(con, command);
atomic_store_ptr(&result, pSql); // set the global TAOS_RES pointer
TAOS_RES* pSql = taos_query_h(con, command, &result);
if (taos_errno(pSql)) {
taos_error(pSql);
return;
......
src/mnode/inc/mnodeProfile.h
浏览文件 @ 5f21e2e7
......@@ -23,6 +23,8 @@ extern "C" {
typedef struct {
char user[TSDB_USER_LEN];
char appName[TSDB_APPNAME_LEN]; // app name that invokes taosc
uint32_t pid; // pid of app that invokes taosc
int8_t killed;
uint16_t port;
uint32_t ip;
......@@ -40,7 +42,7 @@ typedef struct {
int32_t mnodeInitProfile();
void mnodeCleanupProfile();
SConnObj *mnodeCreateConn(char *user, uint32_t ip, uint16_t port);
SConnObj *mnodeCreateConn(char *user, uint32_t ip, uint16_t port, int32_t pid, const char* app);
SConnObj *mnodeAccquireConn(int32_t connId, char *user, uint32_t ip, uint16_t port);
void mnodeReleaseConn(SConnObj *pConn);
int32_t mnodeSaveQueryStreamList(SConnObj *pConn, SCMHeartBeatMsg *pHBMsg);
......
src/mnode/src/mnodeCluster.c
浏览文件 @ 5f21e2e7
......@@ -224,7 +224,8 @@ static int32_t mnodeRetrieveClusters(SShowObj *pShow, char *data, int32_t rows,
mnodeDecClusterRef(pCluster);
numOfRows++;
}
mnodeVacuumResult(data, cols, numOfRows, rows, pShow);
mnodeVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);
pShow->numOfReads += numOfRows;
return numOfRows;
}
src/mnode/src/mnodeDb.c
浏览文件 @ 5f21e2e7
......@@ -648,8 +648,12 @@ static int32_t mnodeRetrieveDbs(SShowObj *pShow, char *data, int32_t rows, void
while (numOfRows < rows) {
pShow->pIter = mnodeGetNextDb(pShow->pIter, &pDb);
if (pDb == NULL) break;
if (pDb->pAcct != pUser->pAcct) continue;
if (pDb->pAcct != pUser->pAcct) {
mnodeDecDbRef(pDb);
continue;
}
cols = 0;
......@@ -760,7 +764,7 @@ static int32_t mnodeRetrieveDbs(SShowObj *pShow, char *data, int32_t rows, void
}
pShow->numOfReads += numOfRows;
mnodeVacuumResult(data, cols, numOfRows, rows, pShow);
mnodeVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);
mnodeDecUserRef(pUser);
return numOfRows;
......
src/mnode/src/mnodeDnode.c
浏览文件 @ 5f21e2e7
......@@ -790,6 +790,7 @@ static int32_t mnodeRetrieveDnodes(SShowObj *pShow, char *data, int32_t rows, vo
mnodeDecDnodeRef(pDnode);
}
mnodeVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);
pShow->numOfReads += numOfRows;
return numOfRows;
}
......@@ -891,8 +892,8 @@ int32_t mnodeRetrieveModules(SShowObj *pShow, char *data, int32_t rows, void *pC
mnodeDecDnodeRef(pDnode);
}
mnodeVacuumResult(data, cols, numOfRows, rows, pShow);
mnodeVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);
pShow->numOfReads += numOfRows;
return numOfRows;
}
......@@ -992,6 +993,7 @@ static int32_t mnodeRetrieveConfigs(SShowObj *pShow, char *data, int32_t rows, v
}
}
mnodeVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);
pShow->numOfReads += numOfRows;
return numOfRows;
}
......@@ -1083,8 +1085,8 @@ static int32_t mnodeRetrieveVnodes(SShowObj *pShow, char *data, int32_t rows, vo
} else {
numOfRows = 0;
}
mnodeVacuumResult(data, cols, numOfRows, rows, pShow);
mnodeVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);
pShow->numOfReads += numOfRows;
return numOfRows;
}
......
src/mnode/src/mnodeMnode.c
浏览文件 @ 5f21e2e7
......@@ -480,8 +480,8 @@ static int32_t mnodeRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, vo
mnodeDecMnodeRef(pMnode);
}
mnodeVacuumResult(data, cols, numOfRows, rows, pShow);
mnodeVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);
pShow->numOfReads += numOfRows;
return numOfRows;
......
src/mnode/src/mnodeProfile.c
浏览文件 @ 5f21e2e7
......@@ -24,15 +24,9 @@
#include "mnode.h"
#include "mnodeDef.h"
#include "mnodeInt.h"
#include "mnodeAcct.h"
#include "mnodeDnode.h"
#include "mnodeDb.h"
#include "mnodeMnode.h"
#include "mnodeProfile.h"
#include "mnodeShow.h"
#include "mnodeTable.h"
#include "mnodeUser.h"
#include "mnodeVgroup.h"
#include "mnodeWrite.h"
#define CONN_KEEP_TIME (tsShellActivityTimer * 3)
......@@ -78,7 +72,7 @@ void mnodeCleanupProfile() {
}
}
SConnObj *mnodeCreateConn(char *user, uint32_t ip, uint16_t port) {
SConnObj *mnodeCreateConn(char *user, uint32_t ip, uint16_t port, int32_t pid, const char* app) {
#if 0
int32_t connSize = taosHashGetSize(tsMnodeConnCache->pHashTable);
if (connSize > tsMaxShellConns) {
......@@ -96,10 +90,13 @@ SConnObj *mnodeCreateConn(char *user, uint32_t ip, uint16_t port) {
.ip = ip,
.port = port,
.connId = connId,
.stime = taosGetTimestampMs()
.stime = taosGetTimestampMs(),
.pid = pid,
};
tstrncpy(connObj.user, user, sizeof(connObj.user));
tstrncpy(connObj.user, user, tListLen(connObj.user));
tstrncpy(connObj.appName, app, tListLen(connObj.appName));
connObj.lastAccess = connObj.stime;
SConnObj *pConn = taosCachePut(tsMnodeConnCache, &connId, sizeof(int32_t), &connObj, sizeof(connObj), CONN_KEEP_TIME * 1000);
......@@ -183,6 +180,20 @@ static int32_t mnodeGetConnsMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pC
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
// app name
pShow->bytes[cols] = TSDB_APPNAME_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "app_name");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
// app pid
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "pid");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = TSDB_IPv4ADDR_LEN + 6 + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "ip:port");
......@@ -191,13 +202,13 @@ static int32_t mnodeGetConnsMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pC
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "login time");
strcpy(pSchema[cols].name, "login_time");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "last access");
strcpy(pSchema[cols].name, "last_access");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
......@@ -236,6 +247,16 @@ static int32_t mnodeRetrieveConns(SShowObj *pShow, char *data, int32_t rows, voi
STR_WITH_MAXSIZE_TO_VARSTR(pWrite, pConnObj->user, pShow->bytes[cols]);
cols++;
// app name
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
STR_WITH_MAXSIZE_TO_VARSTR(pWrite, pConnObj->appName, pShow->bytes[cols]);
cols++;
// app pid
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(int32_t*)pWrite = pConnObj->pid;
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
snprintf(ipStr, sizeof(ipStr), "%s:%u", taosIpStr(pConnObj->ip), pConnObj->port);
STR_WITH_MAXSIZE_TO_VARSTR(pWrite, ipStr, pShow->bytes[cols]);
......@@ -254,8 +275,7 @@ static int32_t mnodeRetrieveConns(SShowObj *pShow, char *data, int32_t rows, voi
}
pShow->numOfReads += numOfRows;
const int32_t NUM_OF_COLUMNS = 5;
mnodeVacuumResult(data, NUM_OF_COLUMNS, numOfRows, rows, pShow);
mnodeVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);
return numOfRows;
}
......@@ -299,7 +319,7 @@ static int32_t mnodeGetQueryMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pC
pShow->bytes[cols] = QUERY_ID_SIZE + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "queryId");
strcpy(pSchema[cols].name, "query_id");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
......@@ -315,9 +335,15 @@ static int32_t mnodeGetQueryMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pC
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 24;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "qhandle");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "created time");
strcpy(pSchema[cols].name, "created_time");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
......@@ -352,7 +378,7 @@ static int32_t mnodeRetrieveQueries(SShowObj *pShow, char *data, int32_t rows, v
SConnObj *pConnObj = NULL;
int32_t cols = 0;
char * pWrite;
char ipStr[TSDB_IPv4ADDR_LEN + 6];
char str[TSDB_IPv4ADDR_LEN + 6] = {0};
while (numOfRows < rows) {
pShow->pIter = mnodeGetNextConn(pShow->pIter, &pConnObj);
......@@ -362,9 +388,9 @@ static int32_t mnodeRetrieveQueries(SShowObj *pShow, char *data, int32_t rows, v
SQueryDesc *pDesc = pConnObj->pQueries + i;
cols = 0;
snprintf(ipStr, QUERY_ID_SIZE + 1, "%u:%u", pConnObj->connId, htonl(pDesc->queryId));
snprintf(str, QUERY_ID_SIZE + 1, "%u:%u", pConnObj->connId, htonl(pDesc->queryId));
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
STR_WITH_MAXSIZE_TO_VARSTR(pWrite, ipStr, pShow->bytes[cols]);
STR_WITH_MAXSIZE_TO_VARSTR(pWrite, str, pShow->bytes[cols]);
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
......@@ -372,8 +398,15 @@ static int32_t mnodeRetrieveQueries(SShowObj *pShow, char *data, int32_t rows, v
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
snprintf(ipStr, sizeof(ipStr), "%s:%u", taosIpStr(pConnObj->ip), pConnObj->port);
STR_WITH_MAXSIZE_TO_VARSTR(pWrite, ipStr, pShow->bytes[cols]);
snprintf(str, tListLen(str), "%s:%u", taosIpStr(pConnObj->ip), pConnObj->port);
STR_WITH_MAXSIZE_TO_VARSTR(pWrite, str, pShow->bytes[cols]);
cols++;
char handleBuf[24] = {0};
snprintf(handleBuf, tListLen(handleBuf), "%p", (void*)htobe64(pDesc->qHandle));
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
STR_WITH_MAXSIZE_TO_VARSTR(pWrite, handleBuf, pShow->bytes[cols]);
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
......@@ -393,8 +426,7 @@ static int32_t mnodeRetrieveQueries(SShowObj *pShow, char *data, int32_t rows, v
}
pShow->numOfReads += numOfRows;
const int32_t NUM_OF_COLUMNS = 6;
mnodeVacuumResult(data, NUM_OF_COLUMNS, numOfRows, rows, pShow);
mnodeVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);
return numOfRows;
}
......@@ -522,8 +554,7 @@ static int32_t mnodeRetrieveStreams(SShowObj *pShow, char *data, int32_t rows, v
}
pShow->numOfReads += numOfRows;
const int32_t NUM_OF_COLUMNS = 8;
mnodeVacuumResult(data, NUM_OF_COLUMNS, numOfRows, rows, pShow);
mnodeVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);
return numOfRows;
}
......
src/mnode/src/mnodeSdb.c
浏览文件 @ 5f21e2e7
......@@ -594,7 +594,7 @@ static int sdbWrite(void *param, void *data, int type) {
pthread_mutex_unlock(&tsSdbObj.mutex);
sdbError("table:%s, failed to restore %s record:%s from source(%d), ver:%" PRId64 " too large, sdb ver:%" PRId64,
pTable->tableName, sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), type, pHead->version, tsSdbObj.version);
return TSDB_CODE_MND_APP_ERROR;
return TSDB_CODE_SYN_INVALID_VERSION;
} else {
tsSdbObj.version = pHead->version;
}
......
src/mnode/src/mnodeShow.c
浏览文件 @ 5f21e2e7
......@@ -186,7 +186,7 @@ static int32_t mnodeProcessRetrieveMsg(SMnodeMsg *pMsg) {
rowsToRead = pShow->numOfRows - pShow->numOfReads;
}
/* return no more than 100 meters in one round trip */
/* return no more than 100 tables in one round trip */
if (rowsToRead > 100) rowsToRead = 100;
/*
......@@ -244,7 +244,8 @@ static int32_t mnodeProcessHeartBeatMsg(SMnodeMsg *pMsg) {
int32_t connId = htonl(pHBMsg->connId);
SConnObj *pConn = mnodeAccquireConn(connId, connInfo.user, connInfo.clientIp, connInfo.clientPort);
if (pConn == NULL) {
pConn = mnodeCreateConn(connInfo.user, connInfo.clientIp, connInfo.clientPort);
pHBMsg->pid = htonl(pHBMsg->pid);
pConn = mnodeCreateConn(connInfo.user, connInfo.clientIp, connInfo.clientPort, pHBMsg->pid, pHBMsg->appName);
}
if (pConn == NULL) {
......@@ -325,7 +326,8 @@ static int32_t mnodeProcessConnectMsg(SMnodeMsg *pMsg) {
goto connect_over;
}
SConnObj *pConn = mnodeCreateConn(connInfo.user, connInfo.clientIp, connInfo.clientPort);
pConnectMsg->pid = htonl(pConnectMsg->pid);
SConnObj *pConn = mnodeCreateConn(connInfo.user, connInfo.clientIp, connInfo.clientPort, pConnectMsg->pid, pConnectMsg->appName);
if (pConn == NULL) {
code = terrno;
} else {
......
src/mnode/src/mnodeTable.c
浏览文件 @ 5f21e2e7
......@@ -63,27 +63,27 @@ static int32_t mnodeRetrieveShowSuperTables(SShowObj *pShow, char *data, int32_t
static int32_t mnodeGetStreamTableMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
static int32_t mnodeRetrieveStreamTables(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static int32_t mnodeProcessCreateTableMsg(SMnodeMsg *mnodeMsg);
static int32_t mnodeProcessCreateTableMsg(SMnodeMsg *pMsg);
static int32_t mnodeProcessCreateSuperTableMsg(SMnodeMsg *pMsg);
static int32_t mnodeProcessCreateChildTableMsg(SMnodeMsg *pMsg);
static void mnodeProcessCreateChildTableRsp(SRpcMsg *rpcMsg);
static int32_t mnodeProcessDropTableMsg(SMnodeMsg *mnodeMsg);
static int32_t mnodeProcessDropTableMsg(SMnodeMsg *pMsg);
static int32_t mnodeProcessDropSuperTableMsg(SMnodeMsg *pMsg);
static void mnodeProcessDropSuperTableRsp(SRpcMsg *rpcMsg);
static int32_t mnodeProcessDropChildTableMsg(SMnodeMsg *pMsg);
static void mnodeProcessDropChildTableRsp(SRpcMsg *rpcMsg);
static int32_t mnodeProcessSuperTableVgroupMsg(SMnodeMsg *mnodeMsg);
static int32_t mnodeProcessMultiTableMetaMsg(SMnodeMsg *mnodeMsg);
static int32_t mnodeProcessTableCfgMsg(SMnodeMsg *mnodeMsg);
static int32_t mnodeProcessSuperTableVgroupMsg(SMnodeMsg *pMsg);
static int32_t mnodeProcessMultiTableMetaMsg(SMnodeMsg *pMsg);
static int32_t mnodeProcessTableCfgMsg(SMnodeMsg *pMsg);
static int32_t mnodeProcessTableMetaMsg(SMnodeMsg *mnodeMsg);
static int32_t mnodeProcessTableMetaMsg(SMnodeMsg *pMsg);
static int32_t mnodeGetSuperTableMeta(SMnodeMsg *pMsg);
static int32_t mnodeGetChildTableMeta(SMnodeMsg *pMsg);
static int32_t mnodeAutoCreateChildTable(SMnodeMsg *pMsg);
static int32_t mnodeProcessAlterTableMsg(SMnodeMsg *mnodeMsg);
static int32_t mnodeProcessAlterTableMsg(SMnodeMsg *pMsg);
static void mnodeProcessAlterTableRsp(SRpcMsg *rpcMsg);
static int32_t mnodeFindSuperTableColumnIndex(SSuperTableObj *pStable, char *colName);
......@@ -131,6 +131,8 @@ static int32_t mnodeChildTableActionInsert(SSdbOper *pOper) {
mnodeAddTableIntoStable(pTable->superTable, pTable);
grantAdd(TSDB_GRANT_TIMESERIES, pTable->superTable->numOfColumns - 1);
if (pAcct) pAcct->acctInfo.numOfTimeSeries += (pTable->superTable->numOfColumns - 1);
} else {
mError("table:%s:%p, correspond stable not found suid:%" PRIu64, pTable->info.tableId, pTable, pTable->suid);
}
} else {
grantAdd(TSDB_GRANT_TIMESERIES, pTable->numOfColumns - 1);
......@@ -460,12 +462,14 @@ static int32_t mnodeSuperTableActionUpdate(SSdbOper *pOper) {
void *oldSchema = pTable->schema;
void *oldVgHash = pTable->vgHash;
int32_t oldRefCount = pTable->refCount;
int32_t oldNumOfTables = pTable->numOfTables;
memcpy(pTable, pNew, sizeof(SSuperTableObj));
pTable->vgHash = oldVgHash;
pTable->refCount = oldRefCount;
pTable->schema = pNew->schema;
pTable->numOfTables = oldNumOfTables;
free(pNew);
free(oldTableId);
free(oldSchema);
......@@ -837,10 +841,11 @@ static int32_t mnodeProcessCreateSuperTableMsg(SMnodeMsg *pMsg) {
return TSDB_CODE_MND_OUT_OF_MEMORY;
}
int64_t us = taosGetTimestampUs();
pStable->info.tableId = strdup(pCreate->tableId);
pStable->info.type = TSDB_SUPER_TABLE;
pStable->createdTime = taosGetTimestampMs();
pStable->uid = (((uint64_t) pStable->createdTime) << 16) + (sdbGetVersion() & ((1ul << 16) - 1ul));
pStable->uid = (us << 24) + ((sdbGetVersion() & ((1ul << 16) - 1ul)) << 8) + (taosRand() & ((1ul << 8) - 1ul));
pStable->sversion = 0;
pStable->tversion = 0;
pStable->numOfColumns = htons(pCreate->numOfColumns);
......@@ -1384,9 +1389,8 @@ int32_t mnodeRetrieveShowSuperTables(SShowObj *pShow, char *data, int32_t rows,
}
pShow->numOfReads += numOfRows;
const int32_t NUM_OF_COLUMNS = 5;
mnodeVacuumResult(data, NUM_OF_COLUMNS, numOfRows, rows, pShow);
mnodeVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);
mnodeDecDbRef(pDb);
return numOfRows;
......@@ -1495,7 +1499,7 @@ static int32_t mnodeProcessSuperTableVgroupMsg(SMnodeMsg *pMsg) {
continue;
}
if (pTable->vgHash == NULL) {
mError("app:%p:%p, stable:%s, no vgroup exist while get stable vgroup info", pMsg->rpcMsg.ahandle, pMsg,
mDebug("app:%p:%p, stable:%s, no vgroup exist while get stable vgroup info", pMsg->rpcMsg.ahandle, pMsg,
stableName);
mnodeDecTableRef(pTable);
......@@ -1706,7 +1710,7 @@ static int32_t mnodeDoCreateChildTable(SMnodeMsg *pMsg, int32_t tid) {
pTable->vgId = pVgroup->vgId;
if (pTable->info.type == TSDB_CHILD_TABLE) {
STagData *pTagData = (STagData *) pCreate->schema; // it is a tag key
STagData *pTagData = (STagData *)pCreate->schema; // it is a tag key
if (pMsg->pSTable == NULL) pMsg->pSTable = mnodeGetSuperTable(pTagData->name);
if (pMsg->pSTable == NULL) {
mError("app:%p:%p, table:%s, corresponding super table:%s does not exist", pMsg->rpcMsg.ahandle, pMsg,
......@@ -2251,7 +2255,8 @@ static void mnodeDropAllChildTablesInStable(SSuperTableObj *pStable) {
int32_t numOfTables = 0;
SChildTableObj *pTable = NULL;
mInfo("stable:%s, all child tables:%d will dropped from sdb", pStable->info.tableId, pStable->numOfTables);
mInfo("stable:%s uid:%" PRIu64 ", all child tables:%d will be dropped from sdb", pStable->info.tableId, pStable->uid,
pStable->numOfTables);
while (1) {
pIter = mnodeGetNextChildTable(pIter, &pTable);
......@@ -2543,6 +2548,25 @@ static int32_t mnodeGetShowTableMeta(STableMetaMsg *pMeta, SShowObj *pShow, void
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 8; // table uid
pSchema[cols].type = TSDB_DATA_TYPE_BIGINT;
strcpy(pSchema[cols].name, "uid");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "tid");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "vgId");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pMeta->numOfColumns = htons(cols);
pShow->numOfColumns = cols;
......@@ -2568,6 +2592,7 @@ static int32_t mnodeRetrieveShowTables(SShowObj *pShow, char *data, int32_t rows
return 0;
}
int32_t cols = 0;
int32_t numOfRows = 0;
SChildTableObj *pTable = NULL;
SPatternCompareInfo info = PATTERN_COMPARE_INFO_INITIALIZER;
......@@ -2608,8 +2633,7 @@ static int32_t mnodeRetrieveShowTables(SShowObj *pShow, char *data, int32_t rows
continue;
}
int32_t cols = 0;
cols = 0;
char *pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
STR_WITH_MAXSIZE_TO_VARSTR(pWrite, tableName, pShow->bytes[cols]);
......@@ -2638,14 +2662,29 @@ static int32_t mnodeRetrieveShowTables(SShowObj *pShow, char *data, int32_t rows
cols++;
// uid
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(int64_t*) pWrite = pTable->uid;
cols++;
// tid
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(int32_t*) pWrite = pTable->sid;
cols++;
//vgid
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(int32_t*) pWrite = pTable->vgId;
cols++;
numOfRows++;
mnodeDecTableRef(pTable);
}
pShow->numOfReads += numOfRows;
const int32_t NUM_OF_COLUMNS = 4;
mnodeVacuumResult(data, NUM_OF_COLUMNS, numOfRows, rows, pShow);
mnodeVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);
mnodeDecDbRef(pDb);
free(pattern);
......@@ -2843,9 +2882,8 @@ static int32_t mnodeRetrieveStreamTables(SShowObj *pShow, char *data, int32_t ro
}
pShow->numOfReads += numOfRows;
const int32_t NUM_OF_COLUMNS = 4;
mnodeVacuumResult(data, NUM_OF_COLUMNS, numOfRows, rows, pShow);
mnodeVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);
mnodeDecDbRef(pDb);
return numOfRows;
......
src/mnode/src/mnodeUser.c
浏览文件 @ 5f21e2e7
......@@ -385,8 +385,8 @@ static int32_t mnodeRetrieveUsers(SShowObj *pShow, char *data, int32_t rows, voi
numOfRows++;
mnodeDecUserRef(pUser);
}
mnodeVacuumResult(data, cols, numOfRows, rows, pShow);
mnodeVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);
pShow->numOfReads += numOfRows;
return numOfRows;
}
......
src/mnode/src/mnodeVgroup.c
浏览文件 @ 5f21e2e7
......@@ -723,8 +723,16 @@ static int32_t mnodeRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, v
while (numOfRows < rows) {
pShow->pIter = mnodeGetNextVgroup(pShow->pIter, &pVgroup);
if (pVgroup == NULL) break;
if (pVgroup->pDb != pDb) continue;
if (!mnodeFilterVgroups(pVgroup, pTable)) continue;
if (pVgroup->pDb != pDb) {
mnodeDecVgroupRef(pVgroup);
continue;
}
if (!mnodeFilterVgroups(pVgroup, pTable)) {
mnodeDecVgroupRef(pVgroup);
continue;
}
cols = 0;
......@@ -771,7 +779,8 @@ static int32_t mnodeRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, v
mnodeDecVgroupRef(pVgroup);
numOfRows++;
}
mnodeVacuumResult(data, cols, numOfRows, rows, pShow);
mnodeVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);
pShow->numOfReads += numOfRows;
mnodeDecTableRef(pTable);
......
src/os/inc/osSemphone.h
浏览文件 @ 5f21e2e7
......@@ -33,6 +33,8 @@ bool taosCheckPthreadValid(pthread_t thread);
int64_t taosGetPthreadId();
void taosResetPthread(pthread_t *thread);
bool taosComparePthread(pthread_t first, pthread_t second);
int32_t taosGetPId();
int32_t taosGetCurrentAPPName(char *name, int32_t* len);
#ifdef __cplusplus
}
......
src/os/src/detail/osSemphone.c
浏览文件 @ 5f21e2e7
......@@ -34,5 +34,31 @@ bool taosCheckPthreadValid(pthread_t thread) { return thread != 0; }
int64_t taosGetPthreadId() { return (int64_t)pthread_self(); }
void taosResetPthread(pthread_t *thread) { *thread = 0; }
bool taosComparePthread(pthread_t first, pthread_t second) { return first == second; }
int32_t taosGetPId() { return getpid(); }
int32_t taosGetCurrentAPPName(char *name, int32_t* len) {
const char* self = "/proc/self/exe";
char path[PATH_MAX] = {0};
if (readlink(self, path, PATH_MAX) <= 0) {
return -1;
}
path[PATH_MAX - 1] = 0;
char* end = strrchr(path, '/');
if (end == NULL) {
return -1;
}
++end;
strcpy(name, end);
if (len != NULL) {
*len = strlen(name);
}
return 0;
}
#endif
\ No newline at end of file
src/os/src/detail/osSysinfo.c
浏览文件 @ 5f21e2e7
......@@ -203,7 +203,7 @@ static void taosGetSystemTimezone() {
snprintf(tsTimezone, TSDB_TIMEZONE_LEN, "%s (%s, %s%02d00)", buf, tzname[daylight], tz >= 0 ? "+" : "-", abs(tz));
// cfg_timezone->cfgStatus = TAOS_CFG_CSTATUS_DEFAULT;
uInfo("timezone not configured, set to system default:%s", tsTimezone);
uWarn("timezone not configured, set to system default:%s", tsTimezone);
}
/*
......@@ -235,7 +235,7 @@ static void taosGetSystemLocale() { // get and set default locale
strcpy(tsLocale, "en_US.UTF-8");
} else {
tstrncpy(tsLocale, locale, TSDB_LOCALE_LEN);
uError("locale not configured, set to system default:%s", tsLocale);
uWarn("locale not configured, set to system default:%s", tsLocale);
}
}
......
src/os/src/linux/CMakeLists.txt
浏览文件 @ 5f21e2e7
......@@ -4,4 +4,4 @@ PROJECT(TDengine)
AUX_SOURCE_DIRECTORY(. SRC)
ADD_LIBRARY(os ${SRC})
TARGET_LINK_LIBRARIES(os m rt)
TARGET_LINK_LIBRARIES(os m rt z)
src/os/src/windows/wSemphone.c
浏览文件 @ 5f21e2e7
......@@ -36,3 +36,21 @@ int64_t taosGetPthreadId() {
bool taosComparePthread(pthread_t first, pthread_t second) {
return first.p == second.p;
}
int32_t taosGetPId() {
return GetCurrentProcessId();
}
int32_t taosGetCurrentAPPName(char *name, int32_t* len) {
char filepath[1024] = {0};
GetModuleFileName(NULL, filepath, MAX_PATH);
*strrchr(filepath,'.') = '\0';
strcpy(name, filepath);
if (len != NULL) {
*len = (int32_t) strlen(filepath);
}
return 0;
}
\ No newline at end of file
src/rpc/src/rpcMain.c
浏览文件 @ 5f21e2e7
......@@ -1051,6 +1051,9 @@ static void *rpcProcessMsgFromPeer(SRecvInfo *pRecv) {
if (code != 0) { // parsing error
if (rpcIsReq(pHead->msgType)) {
rpcSendErrorMsgToPeer(pRecv, code);
if (code == TSDB_CODE_RPC_INVALID_TIME_STAMP || code == TSDB_CODE_RPC_AUTH_FAILURE) {
rpcCloseConn(pConn);
}
tDebug("%s %p %p, %s is sent with error code:0x%x", pRpc->label, pConn, (void *)pHead->ahandle, taosMsg[pHead->msgType+1], code);
}
} else { // msg is passed to app only parsing is ok
......
src/sync/src/syncMain.c
浏览文件 @ 5f21e2e7
......@@ -311,9 +311,20 @@ int32_t syncForwardToPeer(void *param, void *data, void *mhandle, int qtype) {
if (pNode == NULL) return 0;
if (nodeRole == TAOS_SYNC_ROLE_SLAVE && pWalHead->version != nodeVersion + 1) {
sError("vgId:%d, received ver:%" PRIu64 ", inconsistent with last ver:%" PRIu64 ", restart connection", pNode->vgId,
pWalHead->version, nodeVersion);
for (int i = 0; i < pNode->replica; ++i) {
pPeer = pNode->peerInfo[i];
syncRestartConnection(pPeer);
}
return TSDB_CODE_SYN_INVALID_VERSION;
}
// always update version
nodeVersion = pWalHead->version;
sDebug("replica:%d nodeRole:%d qtype:%d", pNode->replica, nodeRole, qtype);
sDebug("vgId:%d, replica:%d nodeRole:%s qtype:%d ver:%" PRIu64, pNode->vgId, pNode->replica, syncRole[nodeRole],
qtype, pWalHead->version);
if (pNode->replica == 1 || nodeRole != TAOS_SYNC_ROLE_MASTER) return 0;
......@@ -883,7 +894,7 @@ static void syncProcessPeersStatusMsg(char *cont, SSyncPeer *pPeer) {
SSyncNode * pNode = pPeer->pSyncNode;
SPeersStatus *pPeersStatus = (SPeersStatus *)cont;
sDebug("%s, status msg received, self:%s ver:%" PRIu64 " peer:%s ver:%" PRIu64 ", ack:%d", pPeer->id,
sDebug("%s, status msg is received, self:%s ver:%" PRIu64 " peer:%s ver:%" PRIu64 ", ack:%d", pPeer->id,
syncRole[nodeRole], nodeVersion, syncRole[pPeersStatus->role], pPeersStatus->version, pPeersStatus->ack);
pPeer->version = pPeersStatus->version;
......@@ -970,7 +981,8 @@ static void syncSendPeersStatusMsgToPeer(SSyncPeer *pPeer, char ack) {
int retLen = write(pPeer->peerFd, msg, statusMsgLen);
if (retLen == statusMsgLen) {
sDebug("%s, status msg is sent", pPeer->id);
sDebug("%s, status msg is sent, self:%s ver:%" PRIu64 ", ack:%d", pPeer->id, syncRole[pPeersStatus->role],
pPeersStatus->version, pPeersStatus->ack);
} else {
sDebug("%s, failed to send status msg, restart", pPeer->id);
syncRestartConnection(pPeer);
......
src/util/src/tcache.c
浏览文件 @ 5f21e2e7
......@@ -107,12 +107,14 @@ static FORCE_INLINE void taosCacheReleaseNode(SCacheObj *pCacheObj, SCacheDataNo
free(pNode);
}
static FORCE_INLINE void doRemoveElemInTrashcan(SCacheObj* pCacheObj, STrashElem *pElem) {
static FORCE_INLINE STrashElem* doRemoveElemInTrashcan(SCacheObj* pCacheObj, STrashElem *pElem) {
if (pElem->pData->signature != (uint64_t) pElem->pData) {
uWarn("key:sig:0x%" PRIx64 " %p data has been released, ignore", pElem->pData->signature, pElem->pData);
return;
return NULL;
}
STrashElem* next = pElem->next;
pCacheObj->numOfElemsInTrash--;
if (pElem->prev) {
pElem->prev->next = pElem->next;
......@@ -120,9 +122,15 @@ static FORCE_INLINE void doRemoveElemInTrashcan(SCacheObj* pCacheObj, STrashElem
pCacheObj->pTrash = pElem->next;
}
if (pElem->next) {
pElem->next->prev = pElem->prev;
if (next) {
next->prev = pElem->prev;
}
if (pCacheObj->numOfElemsInTrash == 0) {
assert(pCacheObj->pTrash == NULL);
}
return next;
}
static FORCE_INLINE void doDestroyTrashcanElem(SCacheObj* pCacheObj, STrashElem *pElem) {
......@@ -550,8 +558,8 @@ void taosAddToTrashcan(SCacheObj *pCacheObj, SCacheDataNode *pNode) {
pCacheObj->numOfElemsInTrash++;
__cache_unlock(pCacheObj);
uDebug("cache:%s key:%p, %p move to trashcan, numOfElem in trashcan:%d", pCacheObj->name, pNode->key, pNode->data,
pCacheObj->numOfElemsInTrash);
uDebug("cache:%s key:%p, %p move to trashcan, pTrashElem:%p, numOfElem in trashcan:%d", pCacheObj->name,
pNode->key, pNode->data, pElem, pCacheObj->numOfElemsInTrash);
}
void taosTrashcanEmpty(SCacheObj *pCacheObj, bool force) {
......@@ -559,31 +567,30 @@ void taosTrashcanEmpty(SCacheObj *pCacheObj, bool force) {
if (pCacheObj->numOfElemsInTrash == 0) {
if (pCacheObj->pTrash != NULL) {
pCacheObj->pTrash = NULL;
uError("cache:%s, key:inconsistency data in cache, numOfElem in trashcan:%d", pCacheObj->name, pCacheObj->numOfElemsInTrash);
}
pCacheObj->pTrash = NULL;
__cache_unlock(pCacheObj);
return;
}
STrashElem *pElem = pCacheObj->pTrash;
const char* stat[] = {"false", "true"};
uDebug("cache:%s start to cleanup trashcan, numOfElem in trashcan:%d, free:%s", pCacheObj->name,
pCacheObj->numOfElemsInTrash, (force? stat[1]:stat[0]));
STrashElem *pElem = pCacheObj->pTrash;
while (pElem) {
T_REF_VAL_CHECK(pElem->pData);
if (pElem->next == pElem) {
pElem->next = NULL;
}
assert(pElem->next != pElem && pElem->prev != pElem);
if (force || (T_REF_VAL_GET(pElem->pData) == 0)) {
uDebug("cache:%s, key:%p, %p removed from trashcan. numOfElem in trashcan:%d", pCacheObj->name, pElem->pData->key, pElem->pData->data,
pCacheObj->numOfElemsInTrash - 1);
STrashElem *p = pElem;
pElem = pElem->next;
doRemoveElemInTrashcan(pCacheObj, p);
doDestroyTrashcanElem(pCacheObj, p);
doRemoveElemInTrashcan(pCacheObj, pElem);
doDestroyTrashcanElem(pCacheObj, pElem);
pElem = pCacheObj->pTrash;
} else {
pElem = pElem->next;
}
......
src/vnode/src/vnodeMain.c
浏览文件 @ 5f21e2e7
......@@ -239,8 +239,10 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
code = vnodeReadVersion(pVnode);
if (code != TSDB_CODE_SUCCESS) {
vnodeCleanUp(pVnode);
return code;
vError("vgId:%d, failed to read version, generate it from data file", pVnode->vgId);
// Allow vnode start even when read version fails, set version as walVersion or zero
// vnodeCleanUp(pVnode);
// return code;
}
pVnode->fversion = pVnode->version;
......@@ -292,6 +294,9 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
}
walRestore(pVnode->wal, pVnode, vnodeWriteToQueue);
if (pVnode->version == 0) {
pVnode->version = walGetVersion(pVnode->wal);
}
SSyncInfo syncInfo;
syncInfo.vgId = pVnode->vgId;
......@@ -947,6 +952,7 @@ static int32_t vnodeSaveVersion(SVnodeObj *pVnode) {
len += snprintf(content + len, maxLen - len, "}\n");
fwrite(content, 1, len, fp);
fflush(fp);
fclose(fp);
vInfo("vgId:%d, save vnode version:%" PRId64 " succeed", pVnode->vgId, pVnode->fversion);
......@@ -960,7 +966,7 @@ static int32_t vnodeReadVersion(SVnodeObj *pVnode) {
cJSON *root = NULL;
int maxLen = 100;
terrno = TSDB_CODE_VND_APP_ERROR;
terrno = TSDB_CODE_VND_INVALID_VRESION_FILE;
sprintf(versionFile, "%s/vnode%d/version.json", tsVnodeDir, pVnode->vgId);
FILE *fp = fopen(versionFile, "r");
if (!fp) {
......@@ -999,6 +1005,6 @@ static int32_t vnodeReadVersion(SVnodeObj *pVnode) {
PARSE_OVER:
taosTFree(content);
cJSON_Delete(root);
if(fp) fclose(fp);
if (fp) fclose(fp);
return terrno;
}
src/vnode/src/vnodeRead.c
浏览文件 @ 5f21e2e7
......@@ -54,13 +54,13 @@ int32_t vnodeProcessRead(void *param, SReadMsg *pReadMsg) {
// tsdb may be in reset state
if (pVnode->tsdb == NULL) return TSDB_CODE_APP_NOT_READY;
if (pVnode->status == TAOS_VN_STATUS_CLOSING)
return TSDB_CODE_APP_NOT_READY;
if (pVnode->status == TAOS_VN_STATUS_CLOSING) return TSDB_CODE_APP_NOT_READY;
// TODO: Later, let slave to support query
// if (pVnode->syncCfg.replica > 1 && pVnode->role != TAOS_SYNC_ROLE_MASTER) {
if (pVnode->role != TAOS_SYNC_ROLE_SLAVE && pVnode->role != TAOS_SYNC_ROLE_MASTER) {
vDebug("vgId:%d, msgType:%s not processed, replica:%d role:%d", pVnode->vgId, taosMsg[msgType], pVnode->syncCfg.replica, pVnode->role);
vDebug("vgId:%d, msgType:%s not processed, replica:%d role:%s", pVnode->vgId, taosMsg[msgType],
pVnode->syncCfg.replica, syncRole[pVnode->role]);
return TSDB_CODE_APP_NOT_READY;
}
......@@ -80,7 +80,7 @@ static void vnodePutItemIntoReadQueue(SVnodeObj *pVnode, void **qhandle) {
taosWriteQitem(pVnode->rqueue, TAOS_QTYPE_QUERY, pRead);
}
static int32_t vnodeDumpQueryResult(SRspRet *pRet, void* pVnode, void** handle, bool* freeHandle) {
static int32_t vnodeDumpQueryResult(SRspRet *pRet, void *pVnode, void **handle, bool *freeHandle) {
bool continueExec = false;
int32_t code = TSDB_CODE_SUCCESS;
......@@ -106,54 +106,55 @@ static int32_t vnodeDumpQueryResult(SRspRet *pRet, void* pVnode, void** handle,
return code;
}
static void vnodeBuildNoResultQueryRsp(SRspRet* pRet) {
static void vnodeBuildNoResultQueryRsp(SRspRet *pRet) {
pRet->rsp = (SRetrieveTableRsp *)rpcMallocCont(sizeof(SRetrieveTableRsp));
pRet->len = sizeof(SRetrieveTableRsp);
memset(pRet->rsp, 0, sizeof(SRetrieveTableRsp));
SRetrieveTableRsp* pRsp = pRet->rsp;
SRetrieveTableRsp *pRsp = pRet->rsp;
pRsp->completed = true;
}
static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
void *pCont = pReadMsg->pCont;
void * pCont = pReadMsg->pCont;
int32_t contLen = pReadMsg->contLen;
SRspRet *pRet = &pReadMsg->rspRet;
SQueryTableMsg* pQueryTableMsg = (SQueryTableMsg*) pCont;
SQueryTableMsg *pQueryTableMsg = (SQueryTableMsg *)pCont;
memset(pRet, 0, sizeof(SRspRet));
// qHandle needs to be freed correctly
if (pReadMsg->rpcMsg.code == TSDB_CODE_RPC_NETWORK_UNAVAIL) {
SRetrieveTableMsg* killQueryMsg = (SRetrieveTableMsg*) pReadMsg->pCont;
SRetrieveTableMsg *killQueryMsg = (SRetrieveTableMsg *)pReadMsg->pCont;
killQueryMsg->free = htons(killQueryMsg->free);
killQueryMsg->qhandle = htobe64(killQueryMsg->qhandle);
vWarn("QInfo:%p connection %p broken, kill query", (void*) killQueryMsg->qhandle, pReadMsg->rpcMsg.handle);
vWarn("QInfo:%p connection %p broken, kill query", (void *)killQueryMsg->qhandle, pReadMsg->rpcMsg.handle);
assert(pReadMsg->rpcMsg.contLen > 0 && killQueryMsg->free == 1);
void** qhandle = qAcquireQInfo(pVnode->qMgmt, (uint64_t) killQueryMsg->qhandle);
void **qhandle = qAcquireQInfo(pVnode->qMgmt, (uint64_t)killQueryMsg->qhandle);
if (qhandle == NULL || *qhandle == NULL) {
vWarn("QInfo:%p invalid qhandle, no matched query handle, conn:%p", (void*) killQueryMsg->qhandle, pReadMsg->rpcMsg.handle);
vWarn("QInfo:%p invalid qhandle, no matched query handle, conn:%p", (void *)killQueryMsg->qhandle,
pReadMsg->rpcMsg.handle);
} else {
assert(*qhandle == (void*) killQueryMsg->qhandle);
assert(*qhandle == (void *)killQueryMsg->qhandle);
qKillQuery(*qhandle);
qReleaseQInfo(pVnode->qMgmt, (void**) &qhandle, true);
qReleaseQInfo(pVnode->qMgmt, (void **)&qhandle, true);
}
return TSDB_CODE_TSC_QUERY_CANCELLED;
}
int32_t code = TSDB_CODE_SUCCESS;
void** handle = NULL;
void ** handle = NULL;
if (contLen != 0) {
qinfo_t pQInfo = NULL;
code = qCreateQueryInfo(pVnode->tsdb, pVnode->vgId, pQueryTableMsg, &pQInfo);
SQueryTableRsp *pRsp = (SQueryTableRsp *) rpcMallocCont(sizeof(SQueryTableRsp));
SQueryTableRsp *pRsp = (SQueryTableRsp *)rpcMallocCont(sizeof(SQueryTableRsp));
pRsp->code = code;
pRsp->qhandle = 0;
......@@ -163,7 +164,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
// current connect is broken
if (code == TSDB_CODE_SUCCESS) {
handle = qRegisterQInfo(pVnode->qMgmt, (uint64_t) pQInfo);
handle = qRegisterQInfo(pVnode->qMgmt, (uint64_t)pQInfo);
if (handle == NULL) { // failed to register qhandle, todo add error test case
vError("vgId:%d QInfo:%p register qhandle failed, return to app, code:%s", pVnode->vgId, (void *)pQInfo,
tstrerror(pRsp->code));
......@@ -171,13 +172,15 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
qDestroyQueryInfo(pQInfo); // destroy it directly
} else {
assert(*handle == pQInfo);
pRsp->qhandle = htobe64((uint64_t) pQInfo);
pRsp->qhandle = htobe64((uint64_t)pQInfo);
}
if (handle != NULL && vnodeNotifyCurrentQhandle(pReadMsg->rpcMsg.handle, *handle, pVnode->vgId) != TSDB_CODE_SUCCESS) {
vError("vgId:%d, QInfo:%p, query discarded since link is broken, %p", pVnode->vgId, *handle, pReadMsg->rpcMsg.handle);
if (handle != NULL &&
vnodeNotifyCurrentQhandle(pReadMsg->rpcMsg.handle, *handle, pVnode->vgId) != TSDB_CODE_SUCCESS) {
vError("vgId:%d, QInfo:%p, query discarded since link is broken, %p", pVnode->vgId, *handle,
pReadMsg->rpcMsg.handle);
pRsp->code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
qReleaseQInfo(pVnode->qMgmt, (void**) &handle, true);
qReleaseQInfo(pVnode->qMgmt, (void **)&handle, true);
return pRsp->code;
}
} else {
......@@ -190,7 +193,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
}
} else {
assert(pCont != NULL);
void** qhandle = (void**) pCont;
void **qhandle = (void **)pCont;
vDebug("vgId:%d, QInfo:%p, dnode continues to exec query", pVnode->vgId, *qhandle);
......@@ -233,15 +236,16 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
pRetrieve->free = htons(pRetrieve->free);
pRetrieve->qhandle = htobe64(pRetrieve->qhandle);
vDebug("vgId:%d, QInfo:%p, retrieve msg is disposed, free:%d, conn:%p", pVnode->vgId, (void*) pRetrieve->qhandle, pRetrieve->free, pReadMsg->rpcMsg.handle);
vDebug("vgId:%d, QInfo:%p, retrieve msg is disposed, free:%d, conn:%p", pVnode->vgId, (void *)pRetrieve->qhandle,
pRetrieve->free, pReadMsg->rpcMsg.handle);
memset(pRet, 0, sizeof(SRspRet));
int32_t code = TSDB_CODE_SUCCESS;
void** handle = qAcquireQInfo(pVnode->qMgmt, pRetrieve->qhandle);
if (handle == NULL || (*handle) != (void*) pRetrieve->qhandle) {
void ** handle = qAcquireQInfo(pVnode->qMgmt, pRetrieve->qhandle);
if (handle == NULL || (*handle) != (void *)pRetrieve->qhandle) {
code = TSDB_CODE_QRY_INVALID_QHANDLE;
vDebug("vgId:%d, invalid qhandle in retrieving result, QInfo:%p", pVnode->vgId, (void*) pRetrieve->qhandle);
vDebug("vgId:%d, invalid qhandle in retrieving result, QInfo:%p", pVnode->vgId, (void *)pRetrieve->qhandle);
vnodeBuildNoResultQueryRsp(pRet);
return code;
......@@ -250,7 +254,7 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
if (pRetrieve->free == 1) {
vWarn("vgId:%d, QInfo:%p, retrieve msg received to kill query and free qhandle", pVnode->vgId, *handle);
qKillQuery(*handle);
qReleaseQInfo(pVnode->qMgmt, (void**) &handle, true);
qReleaseQInfo(pVnode->qMgmt, (void **)&handle, true);
vnodeBuildNoResultQueryRsp(pRet);
code = TSDB_CODE_TSC_QUERY_CANCELLED;
......@@ -259,9 +263,11 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
// register the qhandle to connect to quit query immediate if connection is broken
if (vnodeNotifyCurrentQhandle(pReadMsg->rpcMsg.handle, *handle, pVnode->vgId) != TSDB_CODE_SUCCESS) {
vError("vgId:%d, QInfo:%p, retrieve discarded since link is broken, %p", pVnode->vgId, *handle, pReadMsg->rpcMsg.handle);
vError("vgId:%d, QInfo:%p, retrieve discarded since link is broken, %p", pVnode->vgId, *handle,
pReadMsg->rpcMsg.handle);
code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
qReleaseQInfo(pVnode->qMgmt, (void**) &handle, true);
qKillQuery(*handle);
qReleaseQInfo(pVnode->qMgmt, (void **)&handle, true);
return code;
}
......@@ -270,14 +276,14 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
code = qRetrieveQueryResultInfo(*handle, &buildRes, pReadMsg->rpcMsg.handle);
if (code != TSDB_CODE_SUCCESS) {
//TODO handle malloc failure
// TODO handle malloc failure
pRet->rsp = (SRetrieveTableRsp *)rpcMallocCont(sizeof(SRetrieveTableRsp));
pRet->len = sizeof(SRetrieveTableRsp);
memset(pRet->rsp, 0, sizeof(SRetrieveTableRsp));
freeHandle = true;
} else { // result is not ready, return immediately
if (!buildRes) {
qReleaseQInfo(pVnode->qMgmt, (void**) &handle, false);
qReleaseQInfo(pVnode->qMgmt, (void **)&handle, false);
return TSDB_CODE_QRY_NOT_READY;
}
......@@ -287,7 +293,7 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
// If qhandle is not added into vread queue, the query should be completed already or paused with error.
// Here free qhandle immediately
if (freeHandle) {
qReleaseQInfo(pVnode->qMgmt, (void**) &handle, true);
qReleaseQInfo(pVnode->qMgmt, (void **)&handle, true);
}
return code;
......@@ -295,13 +301,13 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
// notify connection(handle) that current qhandle is created, if current connection from
// client is broken, the query needs to be killed immediately.
int32_t vnodeNotifyCurrentQhandle(void* handle, void* qhandle, int32_t vgId) {
SRetrieveTableMsg* killQueryMsg = rpcMallocCont(sizeof(SRetrieveTableMsg));
killQueryMsg->qhandle = htobe64((uint64_t) qhandle);
int32_t vnodeNotifyCurrentQhandle(void *handle, void *qhandle, int32_t vgId) {
SRetrieveTableMsg *killQueryMsg = rpcMallocCont(sizeof(SRetrieveTableMsg));
killQueryMsg->qhandle = htobe64((uint64_t)qhandle);
killQueryMsg->free = htons(1);
killQueryMsg->header.vgId = htonl(vgId);
killQueryMsg->header.contLen = htonl(sizeof(SRetrieveTableMsg));
vDebug("QInfo:%p register qhandle to connect:%p", qhandle, handle);
return rpcReportProgress(handle, (char*) killQueryMsg, sizeof(SRetrieveTableMsg));
return rpcReportProgress(handle, (char *)killQueryMsg, sizeof(SRetrieveTableMsg));
}
src/vnode/src/vnodeWrite.c
浏览文件 @ 5f21e2e7
......@@ -47,7 +47,7 @@ void vnodeInitWriteFp(void) {
int32_t vnodeProcessWrite(void *param1, int qtype, void *param2, void *item) {
int32_t code = 0;
SVnodeObj *pVnode = (SVnodeObj *)param1;
SWalHead *pHead = param2;
SWalHead * pHead = param2;
if (vnodeProcessWriteMsgFp[pHead->msgType] == NULL) {
vDebug("vgId:%d, msgType:%s not processed, no handle", pVnode->vgId, taosMsg[pHead->msgType]);
......@@ -61,40 +61,40 @@ int32_t vnodeProcessWrite(void *param1, int qtype, void *param2, void *item) {
// tsdb may be in reset state
if (pVnode->tsdb == NULL) return TSDB_CODE_APP_NOT_READY;
if (pVnode->status == TAOS_VN_STATUS_CLOSING)
return TSDB_CODE_APP_NOT_READY;
if (pVnode->status == TAOS_VN_STATUS_CLOSING) return TSDB_CODE_APP_NOT_READY;
if (pHead->version == 0) { // from client or CQ
if (pVnode->status != TAOS_VN_STATUS_READY) {
vDebug("vgId:%d, msgType:%s not processed, vnode status is %d", pVnode->vgId, taosMsg[pHead->msgType], pVnode->status);
vDebug("vgId:%d, msgType:%s not processed, vnode status is %d", pVnode->vgId, taosMsg[pHead->msgType],
pVnode->status);
return TSDB_CODE_APP_NOT_READY; // it may be in deleting or closing state
}
if (pVnode->role != TAOS_SYNC_ROLE_MASTER) {
vDebug("vgId:%d, msgType:%s not processed, replica:%d role:%d", pVnode->vgId, taosMsg[pHead->msgType], pVnode->syncCfg.replica, pVnode->role);
vDebug("vgId:%d, msgType:%s not processed, replica:%d role:%s", pVnode->vgId, taosMsg[pHead->msgType],
pVnode->syncCfg.replica, syncRole[pVnode->role]);
return TSDB_CODE_APP_NOT_READY;
}
// assign version
pVnode->version++;
pHead->version = pVnode->version;
if (pVnode->delay) usleep(pVnode->delay*1000);
pHead->version = pVnode->version + 1;
if (pVnode->delay) usleep(pVnode->delay * 1000);
} else { // from wal or forward
// for data from WAL or forward, version may be smaller
if (pHead->version <= pVnode->version) return 0;
}
pVnode->version = pHead->version;
// forward to peers, even it is WAL/FWD, it shall be called to update version in sync
int32_t syncCode = 0;
syncCode = syncForwardToPeer(pVnode->sync, pHead, item, qtype);
if (syncCode < 0) return syncCode;
// write into WAL
code = walWrite(pVnode->wal, pHead);
if (code < 0) return code;
// forward to peers, even it is WAL/FWD, it shall be called to update version in sync
int32_t syncCode = 0;
syncCode = syncForwardToPeer(pVnode->sync, pHead, item, qtype);
if (syncCode < 0) return syncCode;
pVnode->version = pHead->version;
// write data locally
code = (*vnodeProcessWriteMsgFp[pHead->msgType])(pVnode, pHead->cont, item);
......@@ -191,7 +191,7 @@ static int32_t vnodeProcessUpdateTagValMsg(SVnodeObj *pVnode, void *pCont, SRspR
int vnodeWriteToQueue(void *param, void *data, int type) {
SVnodeObj *pVnode = param;
SWalHead *pHead = data;
SWalHead * pHead = data;
int size = sizeof(SWalHead) + pHead->len;
SWalHead *pWal = (SWalHead *)taosAllocateQitem(size);
......@@ -204,4 +204,3 @@ int vnodeWriteToQueue(void *param, void *data, int type) {
return 0;
}
src/wal/src/walMain.c
浏览文件 @ 5f21e2e7
......@@ -120,10 +120,9 @@ void *walOpen(const char *path, const SWalCfg *pCfg) {
if (pCfg->keep == 1) return pWal;
if (walHandleExistingFiles(path) == 0)
walRenew(pWal);
if (walHandleExistingFiles(path) == 0) walRenew(pWal);
if (pWal && pWal->fd <0) {
if (pWal && pWal->fd < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
wError("wal:%s, failed to open(%s)", path, strerror(errno));
walRelease(pWal);
......@@ -154,7 +153,7 @@ int walAlter(twalh wal, const SWalCfg *pCfg) {
pWal->fsyncPeriod = pCfg->fsyncPeriod;
if (walNeedFsyncTimer(pWal)) {
wInfo("wal:%s, reset fsync timer, walLevel:%d fsyncPeriod:%d", pWal->name, pWal->level, pWal->fsyncPeriod);
taosTmrReset(walProcessFsyncTimer, pWal->fsyncPeriod, pWal, &pWal->timer,walTmrCtrl);
taosTmrReset(walProcessFsyncTimer, pWal->fsyncPeriod, pWal, &pWal->timer, walTmrCtrl);
} else {
wInfo("wal:%s, stop fsync timer, walLevel:%d fsyncPeriod:%d", pWal->name, pWal->level, pWal->fsyncPeriod);
taosTmrStop(pWal->timer);
......@@ -174,9 +173,9 @@ void walClose(void *handle) {
if (pWal->keep == 0) {
// remove all files in the directory
for (int i=0; i<pWal->num; ++i) {
snprintf(pWal->name, sizeof(pWal->name), "%s/%s%d", pWal->path, walPrefix, pWal->id-i);
if (remove(pWal->name) <0) {
for (int i = 0; i < pWal->num; ++i) {
snprintf(pWal->name, sizeof(pWal->name), "%s/%s%d", pWal->path, walPrefix, pWal->id - i);
if (remove(pWal->name) < 0) {
wError("wal:%s, failed to remove", pWal->name);
} else {
wDebug("wal:%s, it is removed", pWal->name);
......@@ -197,7 +196,7 @@ int walRenew(void *handle) {
pthread_mutex_lock(&pWal->mutex);
if (pWal->fd >=0) {
if (pWal->fd >= 0) {
close(pWal->fd);
pWal->id++;
wDebug("wal:%s, it is closed", pWal->name);
......@@ -218,7 +217,7 @@ int walRenew(void *handle) {
// remove the oldest wal file
char name[TSDB_FILENAME_LEN * 3];
snprintf(name, sizeof(name), "%s/%s%d", pWal->path, walPrefix, pWal->id - pWal->max);
if (remove(name) <0) {
if (remove(name) < 0) {
wError("wal:%s, failed to remove(%s)", name, strerror(errno));
} else {
wDebug("wal:%s, it is removed", name);
......@@ -247,7 +246,7 @@ int walWrite(void *handle, SWalHead *pHead) {
taosCalcChecksumAppend(0, (uint8_t *)pHead, sizeof(SWalHead));
int contLen = pHead->len + sizeof(SWalHead);
if(taosTWrite(pWal->fd, pHead, contLen) != contLen) {
if (taosTWrite(pWal->fd, pHead, contLen) != contLen) {
wError("wal:%s, failed to write(%s)", pWal->name, strerror(errno));
terrno = TAOS_SYSTEM_ERROR(errno);
} else {
......@@ -258,7 +257,6 @@ int walWrite(void *handle, SWalHead *pHead) {
}
void walFsync(void *handle) {
SWal *pWal = handle;
if (pWal == NULL || pWal->level != TAOS_WAL_FSYNC || pWal->fd < 0) return;
......@@ -277,11 +275,10 @@ int walRestore(void *handle, void *pVnode, int (*writeFp)(void *, void *, int))
terrno = 0;
int plen = strlen(walPrefix);
char opath[TSDB_FILENAME_LEN+5];
char opath[TSDB_FILENAME_LEN + 5];
int slen = snprintf(opath, sizeof(opath), "%s", pWal->path);
if ( pWal->keep == 0)
strcpy(opath+slen, "/old");
if (pWal->keep == 0) strcpy(opath + slen, "/old");
DIR *dir = opendir(opath);
if (dir == NULL && errno == ENOENT) return 0;
......@@ -290,8 +287,8 @@ int walRestore(void *handle, void *pVnode, int (*writeFp)(void *, void *, int))
return terrno;
}
while ((ent = readdir(dir))!= NULL) {
if ( strncmp(ent->d_name, walPrefix, plen) == 0) {
while ((ent = readdir(dir)) != NULL) {
if (strncmp(ent->d_name, walPrefix, plen) == 0) {
index = atol(ent->d_name + plen);
if (index > maxId) maxId = index;
if (index < minId) minId = index;
......@@ -306,13 +303,13 @@ int walRestore(void *handle, void *pVnode, int (*writeFp)(void *, void *, int))
return terrno;
}
if ( count != (maxId-minId+1) ) {
if (count != (maxId - minId + 1)) {
wError("wal:%s, messed up, count:%d max:%d min:%d", opath, count, maxId, minId);
terrno = TSDB_CODE_WAL_APP_ERROR;
} else {
wDebug("wal:%s, %d files will be restored", opath, count);
for (index = minId; index<=maxId; ++index) {
for (index = minId; index <= maxId; ++index) {
snprintf(pWal->name, sizeof(pWal->name), "%s/%s%d", opath, walPrefix, index);
terrno = walRestoreWalFile(pWal, pVnode, writeFp);
if (terrno < 0) break;
......@@ -345,7 +342,7 @@ int walRestore(void *handle, void *pVnode, int (*writeFp)(void *, void *, int))
}
int walGetWalFile(void *handle, char *name, uint32_t *index) {
SWal *pWal = handle;
SWal * pWal = handle;
int code = 1;
int32_t first = 0;
......@@ -361,7 +358,7 @@ int walGetWalFile(void *handle, char *name, uint32_t *index) {
code = -1; // index out of range
} else {
sprintf(name, "wal/%s%d", walPrefix, *index);
code = (*index == pWal->id) ? 0:1;
code = (*index == pWal->id) ? 0 : 1;
}
pthread_mutex_unlock(&(pWal->mutex));
......@@ -370,7 +367,6 @@ int walGetWalFile(void *handle, char *name, uint32_t *index) {
}
static void walRelease(SWal *pWal) {
pthread_mutex_destroy(&pWal->mutex);
pWal->signature = NULL;
free(pWal);
......@@ -487,8 +483,8 @@ int walHandleExistingFiles(const char *path) {
} else {
// move all files to old directory
int count = 0;
while ((ent = readdir(dir))!= NULL) {
if ( strncmp(ent->d_name, walPrefix, plen) == 0) {
while ((ent = readdir(dir)) != NULL) {
if (strncmp(ent->d_name, walPrefix, plen) == 0) {
snprintf(oname, sizeof(oname), "%s/%s", path, ent->d_name);
snprintf(nname, sizeof(nname), "%s/old/%s", path, ent->d_name);
if (taosMkDir(opath, 0755) != 0) {
......@@ -528,10 +524,10 @@ static int walRemoveWalFiles(const char *path) {
return terrno;
}
while ((ent = readdir(dir))!= NULL) {
if ( strncmp(ent->d_name, walPrefix, plen) == 0) {
while ((ent = readdir(dir)) != NULL) {
if (strncmp(ent->d_name, walPrefix, plen) == 0) {
snprintf(name, sizeof(name), "%s/%s", path, ent->d_name);
if (remove(name) <0) {
if (remove(name) < 0) {
wError("wal:%s, failed to remove(%s)", name, strerror(errno));
terrno = TAOS_SYSTEM_ERROR(errno);
}
......@@ -561,3 +557,10 @@ static void walProcessFsyncTimer(void *param, void *tmrId) {
pWal->timer = NULL;
}
}
int64_t walGetVersion(twalh param) {
SWal *pWal = param;
if (pWal == 0) return 0;
return pWal->version;
}
\ No newline at end of file
tests/pytest/crash_gen.sh
浏览文件 @ 5f21e2e7
......@@ -35,12 +35,13 @@ CURR_DIR=`pwd`
IN_TDINTERNAL="community"
if [[ "$CURR_DIR" == *"$IN_TDINTERNAL"* ]]; then
TAOS_DIR=$CURR_DIR/../../..
TAOSD_DIR=`find $TAOS_DIR -name "taosd"|grep bin|head -n1`
LIB_DIR=`echo $TAOSD_DIR|rev|cut -d '/' -f 3,4,5,6,7|rev`/lib
else
TAOS_DIR=$CURR_DIR/../..
TAOSD_DIR=`find $TAOS_DIR -name "taosd"|grep bin|head -n1`
LIB_DIR=`echo $TAOSD_DIR|rev|cut -d '/' -f 3,4,5,6|rev`/lib
fi
TAOSD_DIR=`find $TAOS_DIR -name "taosd"|grep bin|head -n1`
LIB_DIR=`echo $TAOSD_DIR|rev|cut -d '/' -f 3,4,5,6|rev`/lib
# Now getting ready to execute Python
# The following is the default of our standard dev env (Ubuntu 20.04), modify/adjust at your own risk
......
tests/pytest/handle_val_log.sh 0 → 100755
浏览文件 @ 5f21e2e7
# Color setting
RED='\033[0;31m'
GREEN='\033[1;32m'
GREEN_DARK='\033[0;32m'
GREEN_UNDERLINE='\033[4;32m'
NC='\033[0m'
grep 'start to execute\|ERROR SUMMARY' mem-error-out.log|grep -v 'grep'|uniq|tee uniq-mem-error-out.log
for memError in `grep 'ERROR SUMMARY' uniq-mem-error-out.log | awk '{print $4}'`
do
if [ -n "$memError" ]; then
if [ "$memError" -gt 12 ]; then
echo -e "${RED} ## Memory errors number valgrind reports is $memError.\
More than our threshold! ## ${NC}"
fi
fi
done
grep 'start to execute\|definitely lost:' mem-error-out.log|grep -v 'grep'|uniq|tee uniq-definitely-lost-out.log
for defiMemError in `grep 'definitely lost:' uniq-definitely-lost-out.log | awk '{print $7}'`
do
if [ -n "$defiMemError" ]; then
if [ "$defiMemError" -gt 13 ]; then
echo -e "${RED} ## Memory errors number valgrind reports \
Definitely lost is $defiMemError. More than our threshold! ## ${NC}"
fi
fi
done
\ No newline at end of file
tests/pytest/valgrind-test.sh
浏览文件 @ 5f21e2e7
#!/bin/bash
IN_TDINTERNAL="community"
TDIR=`pwd`
if [[ "$tests_dir" == *"$IN_TDINTERNAL"* ]]; then
cd ../..
else
cd ../../..
fi
TOP_DIR=`pwd`
TAOSLIB_DIR=`find . -name "libtaos.so"|grep -w lib|head -n1`
if [[ "$TAOSLIB_DIR" == *"$IN_TDINTERNAL"* ]]; then
LIB_DIR=`find . -name "libtaos.so"|grep -w lib|head -n1|cut -d '/' --fields=2,3,4,5`
else
LIB_DIR=`find . -name "libtaos.so"|grep -w lib|head -n1|cut -d '/' --fields=2,3,4`
fi
if [ ! $LD_LIBRARY_PATH ]; then
export LD_LIBRARY_PATH=$TOP_DIR/$LIB_DIR
else
export LD_LIBRARY_PATH=$TOP_DIR/$LIB_DIR:$LD_LIBRARY_PATH
fi
cd $TDIR
# client
PYTHONMALLOC=malloc python3 ./test.py -g -f client/client.py
PYTHONMALLOC=malloc python3 ./test.py -g -s && sleep 1
......
tests/script/jenkins/basic.txt
浏览文件 @ 5f21e2e7
......@@ -272,6 +272,7 @@ cd ../../../debug; make
./test.sh -f unique/db/replica_part.sim
./test.sh -f unique/dnode/alternativeRole.sim
./test.sh -f unique/dnode/simple.sim
./test.sh -f unique/dnode/balance1.sim
./test.sh -f unique/dnode/balance2.sim
./test.sh -f unique/dnode/balance3.sim
......
tests/script/jenkins/basic_1.txt 0 → 100644
浏览文件 @ 5f21e2e7
./test.sh -f general/alter/cached_schema_after_alter.sim
./test.sh -f general/alter/count.sim
./test.sh -f general/alter/dnode.sim
./test.sh -f general/alter/import.sim
./test.sh -f general/alter/insert1.sim
./test.sh -f general/alter/insert2.sim
./test.sh -f general/alter/metrics.sim
./test.sh -f general/alter/table.sim
./test.sh -f general/cache/new_metrics.sim
./test.sh -f general/cache/restart_metrics.sim
./test.sh -f general/cache/restart_table.sim
./test.sh -f general/connection/connection.sim
./test.sh -f general/column/commit.sim
./test.sh -f general/column/metrics.sim
./test.sh -f general/column/table.sim
./test.sh -f general/compress/commitlog.sim
./test.sh -f general/compress/compress.sim
./test.sh -f general/compress/compress2.sim
./test.sh -f general/compress/uncompress.sim
./test.sh -f general/compute/avg.sim
./test.sh -f general/compute/bottom.sim
./test.sh -f general/compute/count.sim
./test.sh -f general/compute/diff.sim
./test.sh -f general/compute/diff2.sim
./test.sh -f general/compute/first.sim
./test.sh -f general/compute/interval.sim
./test.sh -f general/compute/last.sim
./test.sh -f general/compute/leastsquare.sim
./test.sh -f general/compute/max.sim
./test.sh -f general/compute/min.sim
./test.sh -f general/compute/null.sim
./test.sh -f general/compute/percentile.sim
./test.sh -f general/compute/stddev.sim
./test.sh -f general/compute/sum.sim
./test.sh -f general/compute/top.sim
./test.sh -f general/db/alter_option.sim
./test.sh -f general/db/alter_tables_d2.sim
./test.sh -f general/db/alter_tables_v1.sim
./test.sh -f general/db/alter_tables_v4.sim
./test.sh -f general/db/alter_vgroups.sim
./test.sh -f general/db/basic.sim
./test.sh -f general/db/basic1.sim
./test.sh -f general/db/basic2.sim
./test.sh -f general/db/basic3.sim
./test.sh -f general/db/basic4.sim
./test.sh -f general/db/basic5.sim
./test.sh -f general/db/delete_reuse1.sim
./test.sh -f general/db/delete_reuse2.sim
./test.sh -f general/db/delete_reusevnode.sim
./test.sh -f general/db/delete_reusevnode2.sim
./test.sh -f general/db/delete_writing1.sim
./test.sh -f general/db/delete_writing2.sim
./test.sh -f general/db/delete.sim
./test.sh -f general/db/len.sim
./test.sh -f general/db/repeat.sim
./test.sh -f general/db/tables.sim
./test.sh -f general/db/vnodes.sim
./test.sh -f general/field/2.sim
./test.sh -f general/field/3.sim
./test.sh -f general/field/4.sim
./test.sh -f general/field/5.sim
./test.sh -f general/field/6.sim
./test.sh -f general/field/bigint.sim
./test.sh -f general/field/binary.sim
./test.sh -f general/field/bool.sim
./test.sh -f general/field/single.sim
./test.sh -f general/field/smallint.sim
./test.sh -f general/field/tinyint.sim
./test.sh -f general/http/autocreate.sim
./test.sh -f general/http/chunked.sim
./test.sh -f general/http/gzip.sim
./test.sh -f general/http/restful.sim
./test.sh -f general/http/restful_insert.sim
./test.sh -f general/http/restful_limit.sim
./test.sh -f general/http/restful_full.sim
./test.sh -f general/http/prepare.sim
./test.sh -f general/http/telegraf.sim
./test.sh -f general/http/grafana_bug.sim
./test.sh -f general/http/grafana.sim
./test.sh -f general/import/basic.sim
./test.sh -f general/import/commit.sim
./test.sh -f general/import/large.sim
./test.sh -f general/import/replica1.sim
./test.sh -f general/insert/basic.sim
./test.sh -f general/insert/insert_drop.sim
./test.sh -f general/insert/query_block1_memory.sim
./test.sh -f general/insert/query_block2_memory.sim
./test.sh -f general/insert/query_block1_file.sim
./test.sh -f general/insert/query_block2_file.sim
./test.sh -f general/insert/query_file_memory.sim
./test.sh -f general/insert/query_multi_file.sim
./test.sh -f general/insert/tcp.sim
./test.sh -f general/parser/alter.sim
./test.sh -f general/parser/alter1.sim
./test.sh -f general/parser/alter_stable.sim
./test.sh -f general/parser/auto_create_tb.sim
./test.sh -f general/parser/auto_create_tb_drop_tb.sim
./test.sh -f general/parser/col_arithmetic_operation.sim
./test.sh -f general/parser/columnValue.sim
./test.sh -f general/parser/commit.sim
./test.sh -f general/parser/create_db.sim
./test.sh -f general/parser/create_mt.sim
./test.sh -f general/parser/create_tb.sim
./test.sh -f general/parser/dbtbnameValidate.sim
./test.sh -f general/parser/import_commit1.sim
./test.sh -f general/parser/import_commit2.sim
./test.sh -f general/parser/import_commit3.sim
./test.sh -f general/parser/insert_tb.sim
./test.sh -f general/parser/first_last.sim
./test.sh -f general/parser/lastrow.sim
./test.sh -f general/parser/nchar.sim
./test.sh -f general/parser/null_char.sim
./test.sh -f general/parser/single_row_in_tb.sim
./test.sh -f general/parser/select_from_cache_disk.sim
./test.sh -f general/parser/mixed_blocks.sim
./test.sh -f general/parser/selectResNum.sim
./test.sh -f general/parser/limit.sim
./test.sh -f general/parser/limit1.sim
./test.sh -f general/parser/limit1_tblocks100.sim
./test.sh -f general/parser/select_across_vnodes.sim
./test.sh -f general/parser/slimit1.sim
./test.sh -f general/parser/tbnameIn.sim
./test.sh -f general/parser/projection_limit_offset.sim
./test.sh -f general/parser/limit2.sim
./test.sh -f general/parser/fill.sim
./test.sh -f general/parser/fill_stb.sim
./test.sh -f general/parser/where.sim
./test.sh -f general/parser/slimit.sim
./test.sh -f general/parser/select_with_tags.sim
./test.sh -f general/parser/interp.sim
./test.sh -f general/parser/tags_dynamically_specifiy.sim
./test.sh -f general/parser/groupby.sim
./test.sh -f general/parser/set_tag_vals.sim
./test.sh -f general/parser/tags_filter.sim
./test.sh -f general/parser/slimit_alter_tags.sim
./test.sh -f general/parser/join.sim
./test.sh -f general/parser/join_multivnode.sim
./test.sh -f general/parser/binary_escapeCharacter.sim
./test.sh -f general/parser/repeatAlter.sim
./test.sh -f general/parser/union.sim
./test.sh -f general/parser/topbot.sim
./test.sh -f general/stable/disk.sim
./test.sh -f general/stable/dnode3.sim
./test.sh -f general/stable/metrics.sim
./test.sh -f general/stable/refcount.sim
./test.sh -f general/stable/show.sim
./test.sh -f general/stable/values.sim
./test.sh -f general/stable/vnode3.sim
./test.sh -f general/table/autocreate.sim
./test.sh -f general/table/basic1.sim
./test.sh -f general/table/basic2.sim
./test.sh -f general/table/basic3.sim
./test.sh -f general/table/bigint.sim
./test.sh -f general/table/binary.sim
./test.sh -f general/table/bool.sim
./test.sh -f general/table/column_name.sim
./test.sh -f general/table/column_num.sim
./test.sh -f general/table/column_value.sim
./test.sh -f general/table/column2.sim
./test.sh -f general/table/date.sim
./test.sh -f general/table/db.table.sim
./test.sh -f general/table/delete_reuse1.sim
./test.sh -f general/table/delete_reuse2.sim
./test.sh -f general/table/delete_writing.sim
./test.sh -f general/table/describe.sim
./test.sh -f general/table/double.sim
./test.sh -f general/table/fill.sim
./test.sh -f general/table/float.sim
./test.sh -f general/table/int.sim
./test.sh -f general/table/limit.sim
./test.sh -f general/table/smallint.sim
./test.sh -f general/table/table_len.sim
./test.sh -f general/table/table.sim
./test.sh -f general/table/tinyint.sim
./test.sh -f general/table/vgroup.sim
./test.sh -f unique/dnode/alternativeRole.sim
./test.sh -f unique/dnode/balance1.sim
./test.sh -f unique/dnode/balance2.sim
./test.sh -f unique/dnode/balance3.sim
./test.sh -f unique/dnode/balancex.sim
./test.sh -f unique/dnode/offline1.sim
./test.sh -f unique/dnode/offline2.sim
./test.sh -f unique/dnode/reason.sim
./test.sh -f unique/dnode/remove1.sim
./test.sh -f unique/dnode/remove2.sim
./test.sh -f unique/dnode/vnode_clean.sim
./test.sh -f unique/http/admin.sim
./test.sh -f unique/http/opentsdb.sim
\ No newline at end of file
tests/script/jenkins/basic_2.txt 0 → 100644
浏览文件 @ 5f21e2e7
cd ../../../debug; cmake ..
cd ../../../debug; make
./test.sh -f general/tag/3.sim
./test.sh -f general/tag/4.sim
./test.sh -f general/tag/5.sim
./test.sh -f general/tag/6.sim
./test.sh -f general/tag/add.sim
./test.sh -f general/tag/bigint.sim
./test.sh -f general/tag/binary_binary.sim
./test.sh -f general/tag/binary.sim
./test.sh -f general/tag/bool_binary.sim
./test.sh -f general/tag/bool_int.sim
./test.sh -f general/tag/bool.sim
./test.sh -f general/tag/change.sim
./test.sh -f general/tag/column.sim
./test.sh -f general/tag/commit.sim
./test.sh -f general/tag/create.sim
./test.sh -f general/tag/delete.sim
./test.sh -f general/tag/double.sim
./test.sh -f general/tag/filter.sim
./test.sh -f general/tag/float.sim
./test.sh -f general/tag/int_binary.sim
./test.sh -f general/tag/int_float.sim
./test.sh -f general/tag/int.sim
./test.sh -f general/tag/set.sim
./test.sh -f general/tag/smallint.sim
./test.sh -f general/tag/tinyint.sim
./test.sh -f general/user/authority.sim
./test.sh -f general/user/monitor.sim
./test.sh -f general/user/pass_alter.sim
./test.sh -f general/user/pass_len.sim
./test.sh -f general/user/user_create.sim
./test.sh -f general/user/user_len.sim
./test.sh -f general/vector/metrics_field.sim
./test.sh -f general/vector/metrics_mix.sim
./test.sh -f general/vector/metrics_query.sim
./test.sh -f general/vector/metrics_tag.sim
./test.sh -f general/vector/metrics_time.sim
./test.sh -f general/vector/multi.sim
./test.sh -f general/vector/single.sim
./test.sh -f general/vector/table_field.sim
./test.sh -f general/vector/table_mix.sim
./test.sh -f general/vector/table_query.sim
./test.sh -f general/vector/table_time.sim
./test.sh -f unique/account/account_create.sim
./test.sh -f unique/account/account_delete.sim
./test.sh -f unique/account/account_len.sim
./test.sh -f unique/account/authority.sim
./test.sh -f unique/account/basic.sim
./test.sh -f unique/account/paras.sim
./test.sh -f unique/account/pass_alter.sim
./test.sh -f unique/account/pass_len.sim
./test.sh -f unique/account/usage.sim
./test.sh -f unique/account/user_create.sim
./test.sh -f unique/account/user_len.sim
./test.sh -f unique/big/balance.sim
./test.sh -f unique/big/maxvnodes.sim
./test.sh -f unique/big/tcp.sim
./test.sh -f unique/cluster/alter.sim
./test.sh -f unique/cluster/balance1.sim
./test.sh -f unique/cluster/balance2.sim
./test.sh -f unique/cluster/balance3.sim
./test.sh -f unique/cluster/cache.sim
./test.sh -f unique/cluster/vgroup100.sim
./test.sh -f unique/column/replica3.sim
./test.sh -f unique/db/commit.sim
./test.sh -f unique/db/delete.sim
./test.sh -f unique/db/delete_part.sim
./test.sh -f unique/db/replica_add12.sim
./test.sh -f unique/db/replica_add13.sim
./test.sh -f unique/db/replica_add23.sim
./test.sh -f unique/db/replica_reduce21.sim
./test.sh -f unique/db/replica_reduce32.sim
./test.sh -f unique/db/replica_reduce31.sim
./test.sh -f unique/db/replica_part.sim
tests/script/jenkins/basic_3.txt 0 → 100644
浏览文件 @ 5f21e2e7
./test.sh -f unique/import/replica2.sim
./test.sh -f unique/import/replica3.sim
./test.sh -f unique/stable/balance_replica1.sim
./test.sh -f unique/stable/dnode2_stop.sim
./test.sh -f unique/stable/dnode2.sim
./test.sh -f unique/stable/dnode3.sim
./test.sh -f unique/stable/replica2_dnode4.sim
./test.sh -f unique/stable/replica2_vnode3.sim
./test.sh -f unique/stable/replica3_dnode6.sim
./test.sh -f unique/stable/replica3_vnode3.sim
./test.sh -f unique/mnode/mgmt20.sim
./test.sh -f unique/mnode/mgmt21.sim
./test.sh -f unique/mnode/mgmt22.sim
./test.sh -f unique/mnode/mgmt23.sim
./test.sh -f unique/mnode/mgmt24.sim
#./test.sh -f unique/mnode/mgmt25.sim
#./test.sh -f unique/mnode/mgmt26.sim
./test.sh -f unique/mnode/mgmt33.sim
./test.sh -f unique/mnode/mgmt34.sim
./test.sh -f unique/mnode/mgmtr2.sim
./test.sh -f unique/vnode/many.sim
./test.sh -f unique/vnode/replica2_basic2.sim
./test.sh -f unique/vnode/replica2_repeat.sim
./test.sh -f unique/vnode/replica3_basic.sim
./test.sh -f unique/vnode/replica3_repeat.sim
./test.sh -f unique/vnode/replica3_vgroup.sim
./test.sh -f general/parser/stream_on_sys.sim
./test.sh -f general/stream/metrics_del.sim
./test.sh -f general/stream/metrics_n.sim
./test.sh -f general/stream/metrics_replica1_vnoden.sim
./test.sh -f general/stream/restart_stream.sim
./test.sh -f general/stream/stream_3.sim
./test.sh -f general/stream/stream_restart.sim
./test.sh -f general/stream/table_1.sim
./test.sh -f general/stream/table_del.sim
./test.sh -f general/stream/table_n.sim
./test.sh -f general/stream/table_replica1_vnoden.sim
./test.sh -f unique/arbitrator/check_cluster_cfg_para.sim
#./test.sh -f unique/arbitrator/dn2_mn1_cache_file_sync.sim
./test.sh -f unique/arbitrator/dn3_mn1_full_createTableFail.sim
./test.sh -f unique/arbitrator/dn3_mn1_multiCreateDropTable.sim
#./test.sh -f unique/arbitrator/dn3_mn1_nw_disable_timeout_autoDropDnode.sim
#./test.sh -f unique/arbitrator/dn3_mn1_replica2_wal1_AddDelDnode.sim
./test.sh -f unique/arbitrator/dn3_mn1_replica_change_dropDnod.sim
./test.sh -f unique/arbitrator/dn3_mn1_replica_change.sim
#./test.sh -f unique/arbitrator/dn3_mn1_stopDnode_timeout.sim
# lower the priority while file corruption
#./test.sh -f unique/arbitrator/dn3_mn1_vnode_change.sim
#./test.sh -f unique/arbitrator/dn3_mn1_vnode_corruptFile_offline.sim
#./test.sh -f unique/arbitrator/dn3_mn1_vnode_corruptFile_online.sim
#./test.sh -f unique/arbitrator/dn3_mn1_vnode_createErrData_online.sim
./test.sh -f unique/arbitrator/dn3_mn1_vnode_noCorruptFile_offline.sim
./test.sh -f unique/arbitrator/dn3_mn1_vnode_delDir.sim
./test.sh -f unique/arbitrator/dn3_mn1_r2_vnode_delDir.sim
./test.sh -f unique/arbitrator/dn3_mn1_r3_vnode_delDir.sim
./test.sh -f unique/arbitrator/dn3_mn1_vnode_nomaster.sim
./test.sh -f unique/arbitrator/dn3_mn2_killDnode.sim
./test.sh -f unique/arbitrator/insert_duplicationTs.sim
./test.sh -f unique/arbitrator/offline_replica2_alterTable_online.sim
./test.sh -f unique/arbitrator/offline_replica2_alterTag_online.sim
./test.sh -f unique/arbitrator/offline_replica2_createTable_online.sim
./test.sh -f unique/arbitrator/offline_replica2_dropDb_online.sim
./test.sh -f unique/arbitrator/offline_replica2_dropTable_online.sim
./test.sh -f unique/arbitrator/offline_replica3_alterTable_online.sim
./test.sh -f unique/arbitrator/offline_replica3_alterTag_online.sim
./test.sh -f unique/arbitrator/offline_replica3_createTable_online.sim
./test.sh -f unique/arbitrator/offline_replica3_dropDb_online.sim
./test.sh -f unique/arbitrator/offline_replica3_dropTable_online.sim
./test.sh -f unique/arbitrator/replica_changeWithArbitrator.sim
./test.sh -f unique/arbitrator/sync_replica2_alterTable_add.sim
./test.sh -f unique/arbitrator/sync_replica2_alterTable_drop.sim
./test.sh -f unique/arbitrator/sync_replica2_dropDb.sim
./test.sh -f unique/arbitrator/sync_replica2_dropTable.sim
./test.sh -f unique/arbitrator/sync_replica3_alterTable_add.sim
./test.sh -f unique/arbitrator/sync_replica3_alterTable_drop.sim
./test.sh -f unique/arbitrator/sync_replica3_dropDb.sim
./test.sh -f unique/arbitrator/sync_replica3_dropTable.sim
./test.sh -f unique/migrate/mn2_vn2_repl2_rmMnodeDir.sim
./test.sh -f unique/migrate/mn2_vn2_repl2_rmMnodeVnodeDir.sim
./test.sh -f unique/migrate/mn2_vn2_repl2_rmMnodeVnodeDir_stopAll_starAll.sim
./test.sh -f unique/migrate/mn2_vn2_repl2_rmVnodeDir.sim
tests/script/unique/dnode/simple.sim 0 → 100644
浏览文件 @ 5f21e2e7
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/deploy.sh -n dnode2 -i 2
system sh/deploy.sh -n dnode3 -i 3
system sh/deploy.sh -n dnode4 -i 4
print ========== step1
system sh/exec.sh -n dnode1 -s start
sql connect
sql create dnode $hostname2
system sh/exec.sh -n dnode2 -s start
sql create dnode $hostname3
system sh/exec.sh -n dnode3 -s start
sleep 3000
sql create database d1 replica 2
sql create table d1.t1 (t timestamp, i int)
sql insert into d1.t1 values(now+1s, 15)
sql insert into d1.t1 values(now+2s, 14)
sql insert into d1.t1 values(now+3s, 13)
sql insert into d1.t1 values(now+4s, 12)
sql insert into d1.t1 values(now+5s, 11)
sql show dnodes
print dnode1 openVnodes $data2_1
print dnode2 openVnodes $data2_2
print dnode3 openVnodes $data2_3
print dnode4 openVnodes $data2_4
if $data2_1 != 0 then
return -1
endi
if $data2_2 != 1 then
return -1
endi
if $data2_3 != 1 then
return -1
endi
if $data2_4 != null then
return -1
endi
print ========== step2
sql create dnode $hostname4
system sh/exec.sh -n dnode4 -s start
sleep 3000
sql show dnodes
print dnode1 openVnodes $data2_1
print dnode2 openVnodes $data2_2
print dnode3 openVnodes $data2_3
print dnode4 openVnodes $data2_4
if $data2_1 != 0 then
return -1
endi
if $data2_2 != 1 then
return -1
endi
if $data2_3 != 1 then
return -1
endi
if $data2_4 != 0 then
return -1
endi
print ========== step3
sql drop dnode $hostname2
$x = 0
show3:
$x = $x + 1
sleep 2000
if $x == 10 then
return -1
endi
sql show dnodes
print dnode1 openVnodes $data2_1
print dnode2 openVnodes $data2_2
print dnode3 openVnodes $data2_3
print dnode4 openVnodes $data2_4
if $data2_1 != 0 then
goto show3
endi
if $data2_2 != null then
goto show3
endi
if $data2_3 != 1 then
goto show3
endi
if $data2_4 != 1 then
goto show3
endi
print ========== step4
sql drop dnode $hostname3
$x = 0
show4:
$x = $x + 1
sleep 2000
if $x == 10 then
return -1
endi
sql show dnodes
print dnode1 openVnodes $data2_1
print dnode2 openVnodes $data2_2
print dnode3 openVnodes $data2_3
print dnode4 openVnodes $data2_4
if $data2_1 != 1 then
goto show4
endi
if $data2_2 != null then
goto show4
endi
if $data2_3 != null then
goto show4
endi
if $data2_4 != 1 then
goto show4
endi
print ========== step5
sql select * from d1.t1 order by t desc
print $data01 $data11 $data21 $data31 $data41
if $data01 != 11 then
return -1
endi
if $data11 != 12 then
return -1
endi
if $data21 != 13 then
return -1
endi
if $data31 != 14 then
return -1
endi
if $data41 != 15 then
return -1
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT
system sh/exec.sh -n dnode2 -s stop -x SIGINT
system sh/exec.sh -n dnode3 -s stop -x SIGINT
system sh/exec.sh -n dnode4 -s stop -x SIGINT
tests/test-all.sh
浏览文件 @ 5f21e2e7
......@@ -56,6 +56,15 @@ if [ "$2" != "python" ]; then
elif [ "$1" == "full" ]; then
echo "### run TSIM full test ###"
runSimCaseOneByOne jenkins/basic.txt
elif [ "$1" == "b1" ]; then
echo "### run TSIM b1 test ###"
runSimCaseOneByOne jenkins/basic_1.txt
elif [ "$1" == "b2" ]; then
echo "### run TSIM b2 test ###"
runSimCaseOneByOne jenkins/basic_2.txt
elif [ "$1" == "b3" ]; then
echo "### run TSIM b3 test ###"
runSimCaseOneByOne jenkins/basic_3.txt
elif [ "$1" == "smoke" ] || [ -z "$1" ]; then
echo "### run TSIM smoke test ###"
runSimCaseOneByOne basicSuite.sim
......@@ -112,6 +121,11 @@ if [ "$2" != "sim" ]; then
elif [ "$1" == "full" ]; then
echo "### run Python full test ###"
runPyCaseOneByOne fulltest.sh
elif [ "$1" == "pytest" ]; then
echo "### run Python full test ###"
runPyCaseOneByOne fulltest.sh
elif [ "$1" == "b2" ] || [ "$1" == "b3" ]; then
exit $(($totalFailed + $totalPyFailed))
elif [ "$1" == "smoke" ] || [ -z "$1" ]; then
echo "### run Python smoke test ###"
runPyCaseOneByOne smoketest.sh
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或