Skip to content

T
TDengine

慢慢CG / TDengine
与 Fork 源项目一致

Fork自 taosdata / TDengine

通知 1
Star 0
Fork 0
T
TDengine
提交 c6479a14 编写于 作者: H Haojun Liao
浏览文件
操作

[td-225] merge develop

上级 25ee3b48 b604ed13
展开全部 隐藏空白更改
内联 并排
Showing with 18196 addition and 475 deletion
.travis.yml
浏览文件 @ c6479a14
...@@ -54,6 +54,7 @@ matrix: ...@@ -54,6 +54,7 @@ matrix:
py3ver=`python3 --version|awk '{print $2}'|cut -d "." -f 1,2` && apt install python$py3ver-dev py3ver=`python3 --version|awk '{print $2}'|cut -d "." -f 1,2` && apt install python$py3ver-dev
pip3 install psutil pip3 install psutil
pip3 install guppy3
pip3 install --user ${TRAVIS_BUILD_DIR}/src/connector/python/linux/python3/ pip3 install --user ${TRAVIS_BUILD_DIR}/src/connector/python/linux/python3/
cd ${TRAVIS_BUILD_DIR}/tests cd ${TRAVIS_BUILD_DIR}/tests
......
cmake/version.inc
浏览文件 @ c6479a14
...@@ -4,7 +4,7 @@ PROJECT(TDengine) ...@@ -4,7 +4,7 @@ PROJECT(TDengine)
IF (DEFINED VERNUMBER) IF (DEFINED VERNUMBER)
SET(TD_VER_NUMBER ${VERNUMBER}) SET(TD_VER_NUMBER ${VERNUMBER})
ELSE () ELSE ()
SET(TD_VER_NUMBER "2.0.4.0") SET(TD_VER_NUMBER "2.0.5.1")
ENDIF () ENDIF ()
IF (DEFINED VERCOMPATIBLE) IF (DEFINED VERCOMPATIBLE)
......
documentation20/webdocs/markdowndocs/Queries-ch.md
浏览文件 @ c6479a14
...@@ -29,7 +29,7 @@ Query OK, 2 row(s) in set (0.001100s) ...@@ -29,7 +29,7 @@ Query OK, 2 row(s) in set (0.001100s)
具体的查询语法请看<a href="https://www.taosdata.com/cn/documentation20/taos-sql/">TAOS SQL </a>。 具体的查询语法请看<a href="https://www.taosdata.com/cn/documentation20/taos-sql/">TAOS SQL </a>。
## 多表聚合查询 ## 多表聚合查询
物联网场景中，往往同一个类型的数据采集点有多个。TDengine采用超级表(STable)的概念来描述某一个类型的数据采集点，一张普通的表来描述一个具体的数据采集点。同时TDengine使用标签来描述数据采集点的静态属性，一个具体的数据采集点有具体的标签值。通过指定标签的过滤条件，TDengine提供了一高效的方法将超级表(某一类型的数据采集点)所属的子表进行聚合查询。对普通表的聚合函数以及绝大部分操作都适用于超级表，语法完全一样。 物联网场景中，往往同一个类型的数据采集点有多个。TDengine采用超级表(STable)的概念来描述某一个类型的数据采集点，一张普通的表来描述一个具体的数据采集点。同时TDengine使用标签来描述数据采集点的静态属性，一个具体的数据采集点有具体的标签值。通过指定标签的过滤条件，TDengine提供了一高效的方法将超级表(某一类型的数据采集点)所属的子表进行聚合查询。对普通表的聚合函数以及绝大部分操作都适用于超级表，语法完全一样。
**示例1**：在TAOS Shell，查找北京所有智能电表采集的电压平均值，并按照location分组 **示例1**：在TAOS Shell，查找北京所有智能电表采集的电压平均值，并按照location分组
```mysql ```mysql
......
documentation20/webdocs/markdowndocs/administrator-ch.md
浏览文件 @ c6479a14
...@@ -82,8 +82,7 @@ TDengine系统后台服务由taosd提供，可以在配置文件taos.cfg里修 ...@@ -82,8 +82,7 @@ TDengine系统后台服务由taosd提供，可以在配置文件taos.cfg里修
下面仅仅列出一些重要的配置参数，更多的参数请看配置文件里的说明。各个参数的详细介绍及作用请看前述章节，而且这些参数的缺省配置都是工作的，一般无需设置。**注意：配置修改后，需要重启*taosd*服务才能生效。** 下面仅仅列出一些重要的配置参数，更多的参数请看配置文件里的说明。各个参数的详细介绍及作用请看前述章节，而且这些参数的缺省配置都是工作的，一般无需设置。**注意：配置修改后，需要重启*taosd*服务才能生效。**
- firstEp: taosd启动时，主动连接的集群中第一个dnode的end point, 默认值为localhost:6030。 - firstEp: taosd启动时，主动连接的集群中首个dnode的end point, 默认值为localhost:6030。
- secondEp: taosd启动时，如果first连接不上，尝试连接集群中第二个dnode的end point, 默认值为空。
- fqdn：数据节点的FQDN，缺省为操作系统配置的第一个hostname。如果习惯IP地址访问，可设置为该节点的IP地址。 - fqdn：数据节点的FQDN，缺省为操作系统配置的第一个hostname。如果习惯IP地址访问，可设置为该节点的IP地址。
- serverPort：taosd启动后，对外服务的端口号，默认值为6030。 - serverPort：taosd启动后，对外服务的端口号，默认值为6030。
- httpPort: RESTful服务使用的端口号，所有的HTTP请求（TCP）都需要向该接口发起查询/写入请求, 默认值为6041。 - httpPort: RESTful服务使用的端口号，所有的HTTP请求（TCP）都需要向该接口发起查询/写入请求, 默认值为6041。
...@@ -156,76 +155,80 @@ TDengine系统的前台交互客户端应用程序为taos，它与taosd共享同 ...@@ -156,76 +155,80 @@ TDengine系统的前台交互客户端应用程序为taos，它与taosd共享同
客户端配置参数 客户端配置参数
- firstEp: taos启动时，主动连接的集群中第一个taosd实例的end point, 缺省值为 localhost:6030。 - firstEp: taos启动时，主动连接的集群中第一个taosd实例的end point, 缺省值为 localhost:6030。
- secondEp: taos启动时，如果first连接不上，尝试连接集群中第二个taosd实例的end point, 缺省值为空。
- locale - locale
> 默认值：系统中动态获取，如果自动获取失败，需要用户在配置文件设置或通过API设置 默认值：系统中动态获取，如果自动获取失败，需要用户在配置文件设置或通过API设置
TDengine为存储中文、日文、韩文等非ASCII编码的宽字符，提供一种专门的字段类型nchar。写入nchar字段的数据将统一采用UCS4-LE格式进行编码并发送到服务器。需要注意的是，编码正确性是客户端来保证。因此，如果用户想要正常使用nchar字段来存储诸如中文、日文、韩文等非ASCII字符，需要正确设置客户端的编码格式。 TDengine为存储中文、日文、韩文等非ASCII编码的宽字符，提供一种专门的字段类型nchar。写入nchar字段的数据将统一采用UCS4-LE格式进行编码并发送到服务器。需要注意的是，编码正确性是客户端来保证。因此，如果用户想要正常使用nchar字段来存储诸如中文、日文、韩文等非ASCII字符，需要正确设置客户端的编码格式。
客户端的输入的字符均采用操作系统当前默认的编码格式，在Linux系统上多为UTF-8，部分中文系统编码则可能是GB18030或GBK等。在docker环境中默认的编码是POSIX。在中文版Windows系统中，编码则是CP936。客户端需要确保正确设置自己所使用的字符集，即客户端运行的操作系统当前编码字符集，才能保证nchar中的数据正确转换为UCS4-LE编码格式。 客户端的输入的字符均采用操作系统当前默认的编码格式，在Linux系统上多为UTF-8，部分中文系统编码则可能是GB18030或GBK等。在docker环境中默认的编码是POSIX。在中文版Windows系统中，编码则是CP936。客户端需要确保正确设置自己所使用的字符集，即客户端运行的操作系统当前编码字符集，才能保证nchar中的数据正确转换为UCS4-LE编码格式。
在 Linux 中 locale 的命名规则为: <语言>_<地区>.<字符集编码> 如：zh_CN.UTF-8，zh代表中文，CN代表大陆地区，UTF-8表示字符集。字符集编码为客户端正确解析本地字符串提供编码转换的说明。Linux系统与 Mac OSX 系统可以通过设置locale来确定系统的字符编码，由于Windows使用的locale中不是POSIX标准的locale格式，因此在Windows下需要采用另一个配置参数charset来指定字符编码。在Linux 系统中也可以使用charset来指定字符编码。 在 Linux 中 locale 的命名规则为: <语言>_<地区>.<字符集编码> 如：zh_CN.UTF-8，zh代表中文，CN代表大陆地区，UTF-8表示字符集。字符集编码为客户端正确解析本地字符串提供编码转换的说明。Linux系统与 Mac OSX 系统可以通过设置locale来确定系统的字符编码，由于Windows使用的locale中不是POSIX标准的locale格式，因此在Windows下需要采用另一个配置参数charset来指定字符编码。在Linux 系统中也可以使用charset来指定字符编码。
- charset - charset
> 默认值：系统中动态获取，如果自动获取失败，需要用户在配置文件设置或通过API设置 默认值：系统中动态获取，如果自动获取失败，需要用户在配置文件设置或通过API设置
如果配置文件中不设置charset，在Linux系统中，taos在启动时候，自动读取系统当前的locale信息，并从locale信息中解析提取charset编码格式。如果自动读取locale信息失败，则尝试读取charset配置，如果读取charset配置也失败，则中断启动过程。 如果配置文件中不设置charset，在Linux系统中，taos在启动时候，自动读取系统当前的locale信息，并从locale信息中解析提取charset编码格式。如果自动读取locale信息失败，则尝试读取charset配置，如果读取charset配置也失败，则中断启动过程。
在Linux系统中，locale信息包含了字符编码信息，因此正确设置了Linux系统locale以后可以不用再单独设置charset。例如： 在Linux系统中，locale信息包含了字符编码信息，因此正确设置了Linux系统locale以后可以不用再单独设置charset。例如：
``` ```
locale zh_CN.UTF-8 locale zh_CN.UTF-8
``` ```
在Windows系统中，无法从locale获取系统当前编码。如果无法从配置文件中读取字符串编码信息，taos默认设置为字符编码为CP936。其等效在配置文件中添加如下配置： 在Windows系统中，无法从locale获取系统当前编码。如果无法从配置文件中读取字符串编码信息，taos默认设置为字符编码为CP936。其等效在配置文件中添加如下配置：
``` ```
charset CP936 charset CP936
``` ```
如果需要调整字符编码，请查阅当前操作系统使用的编码，并在配置文件中正确设置。 如果需要调整字符编码，请查阅当前操作系统使用的编码，并在配置文件中正确设置。
在Linux系统中，如果用户同时设置了locale和字符集编码charset，并且locale和charset的不一致，后设置的值将覆盖前面设置的值。 在Linux系统中，如果用户同时设置了locale和字符集编码charset，并且locale和charset的不一致，后设置的值将覆盖前面设置的值。
``` ```
locale zh_CN.UTF-8 locale zh_CN.UTF-8
charset GBK charset GBK
``` ```
则charset的有效值是GBK。 则charset的有效值是GBK。
``` ```
charset GBK charset GBK
locale zh_CN.UTF-8 locale zh_CN.UTF-8
``` ```
charset的有效值是UTF-8。 charset的有效值是UTF-8。
日志的配置参数，与server 的配置参数完全一样。 日志的配置参数，与server 的配置参数完全一样。
- timezone - timezone
默认值：从系统中动态获取当前的时区设置 默认值：从系统中动态获取当前的时区设置
客户端运行系统所在的时区。为应对多时区的数据写入和查询问题，TDengine 采用 Unix 时间戳(Unix Timestamp)来记录和存储时间戳。Unix 时间戳的特点决定了任一时刻不论在任何时区，产生的时间戳均一致。需要注意的是，Unix时间戳是在客户端完成转换和记录。为了确保客户端其他形式的时间转换为正确的 Unix 时间戳，需要设置正确的时区。 客户端运行系统所在的时区。为应对多时区的数据写入和查询问题，TDengine 采用 Unix 时间戳(Unix Timestamp)来记录和存储时间戳。Unix 时间戳的特点决定了任一时刻不论在任何时区，产生的时间戳均一致。需要注意的是，Unix时间戳是在客户端完成转换和记录。为了确保客户端其他形式的时间转换为正确的 Unix 时间戳，需要设置正确的时区。
在Linux系统中，客户端会自动读取系统设置的时区信息。用户也可以采用多种方式在配置文件设置时区。例如： 在Linux系统中，客户端会自动读取系统设置的时区信息。用户也可以采用多种方式在配置文件设置时区。例如：
``` ```
timezone UTC-8 timezone UTC-8
timezone GMT-8 timezone GMT-8
timezone Asia/Shanghai timezone Asia/Shanghai
``` ```
均是合法的设置东八区时区的格式。 均是合法的设置东八区时区的格式。
时区的设置对于查询和写入SQL语句中非Unix时间戳的内容（时间戳字符串、关键词now的解析）产生影响。例如： 时区的设置对于查询和写入SQL语句中非Unix时间戳的内容（时间戳字符串、关键词now的解析）产生影响。例如：
``` ```
SELECT count(*) FROM table_name WHERE TS<'2019-04-11 12:01:08'; SELECT count(*) FROM table_name WHERE TS<'2019-04-11 12:01:08';
``` ```
在东八区，SQL语句等效于 在东八区，SQL语句等效于
``` ```
SELECT count(*) FROM table_name WHERE TS<1554955268000; SELECT count(*) FROM table_name WHERE TS<1554955268000;
``` ```
在UTC时区，SQL语句等效于 在UTC时区，SQL语句等效于
``` ```
SELECT count(*) FROM table_name WHERE TS<1554984068000; SELECT count(*) FROM table_name WHERE TS<1554984068000;
``` ```
为了避免使用字符串时间格式带来的不确定性，也可以直接使用Unix时间戳。此外，还可以在SQL语句中使用带有时区的时间戳字符串，例如：RFC3339格式的时间戳字符串，2013-04-12T15:52:01.123+08:00或者ISO-8601格式时间戳字符串2013-04-12T15:52:01.123+0800。上述两个字符串转化为Unix时间戳不受系统所在时区的影响。 为了避免使用字符串时间格式带来的不确定性，也可以直接使用Unix时间戳。此外，还可以在SQL语句中使用带有时区的时间戳字符串，例如：RFC3339格式的时间戳字符串，2013-04-12T15:52:01.123+08:00或者ISO-8601格式时间戳字符串2013-04-12T15:52:01.123+0800。上述两个字符串转化为Unix时间戳不受系统所在时区的影响。
启动taos时，也可以从命令行指定一个taosd实例的end point，否则就从taos.cfg读取。
- maxBinaryDisplayWidth
启动taos时，也可以从命令行指定一个taosd实例的end point，否则就从taos.cfg读取。 Shell中binary 和 nchar字段的显示宽度上限，超过此限制的部分将被隐藏。默认值：30。可在 shell 中通过命令 set max_binary_display_width nn 动态修改此选项。
## 用户管理 ## 用户管理
...@@ -408,5 +411,4 @@ TDengine的所有可执行文件默认存放在 _/usr/local/taos/bin_ 目录下 ...@@ -408,5 +411,4 @@ TDengine的所有可执行文件默认存放在 _/usr/local/taos/bin_ 目录下
您可以通过修改系统配置文件taos.cfg来配置不同的数据目录和日志目录。 您可以通过修改系统配置文件taos.cfg来配置不同的数据目录和日志目录。
##
packaging/docker/Dockerfile
浏览文件 @ c6479a14
...@@ -2,9 +2,11 @@ FROM centos:7 ...@@ -2,9 +2,11 @@ FROM centos:7
WORKDIR /root WORKDIR /root
ARG version
RUN echo $version
COPY tdengine.tar.gz /root/ COPY tdengine.tar.gz /root/
RUN tar -zxf tdengine.tar.gz RUN tar -zxf tdengine.tar.gz
WORKDIR /root/TDengine-server/ WORKDIR /root/TDengine-server-$version/
RUN sh install.sh -e no RUN sh install.sh -e no
......
packaging/docker/dockerbuild.sh
浏览文件 @ c6479a14
#!/bin/bash #!/bin/bash
set -x set -x
docker build --rm -f "Dockerfile" -t tdengine/tdengine:$1 "." docker build --rm -f "Dockerfile" -t tdengine/tdengine:$1 "." --build-arg version=$1
docker login -u tdengine -p $2 #replace the docker registry username and password docker login -u tdengine -p $2 #replace the docker registry username and password
docker push tdengine/tdengine:$1 docker push tdengine/tdengine:$1
packaging/tools/install.sh
浏览文件 @ c6479a14
...@@ -272,6 +272,29 @@ function install_config() { ...@@ -272,6 +272,29 @@ function install_config() {
break break
fi fi
done done
# user email
#EMAIL_PATTERN='^[A-Za-z0-9\u4e00-\u9fa5]+@[a-zA-Z0-9_-]+(\.[a-zA-Z0-9_-]+)+$'
#EMAIL_PATTERN='^[\w-]+(\.[\w-]+)*@[\w-]+(\.[\w-]+)+$'
#EMAIL_PATTERN="^[\w-]+(\.[\w-]+)*@[\w-]+(\.[\w-]+)+$"
echo
echo -e -n "${GREEN}Enter your email address for priority support or enter empty to skip${NC}: "
read emailAddr
while true; do
if [ ! -z "$emailAddr" ]; then
# check the format of the emailAddr
#if [[ "$emailAddr" =~ $EMAIL_PATTERN ]]; then
# Write the email address to temp file
email_file="${install_main_dir}/email"
${csudo} bash -c "echo $emailAddr > ${email_file}"
break
#else
# read -p "Please enter the correct email address: " emailAddr
#fi
else
break
fi
done
} }
......
src/balance/src/balance.c
浏览文件 @ c6479a14
...@@ -957,11 +957,11 @@ static void balanceMonitorDnodeModule() { ...@@ -957,11 +957,11 @@ static void balanceMonitorDnodeModule() {
continue; continue;
} }
mLInfo("dnode:%d, numOfMnodes:%d expect:%d, add mnode in this dnode", pDnode->dnodeId, numOfMnodes, tsNumOfMnodes); mLInfo("dnode:%d, numOfMnodes:%d expect:%d, create mnode in this dnode", pDnode->dnodeId, numOfMnodes, tsNumOfMnodes);
mnodeAddMnode(pDnode->dnodeId); mnodeCreateMnode(pDnode->dnodeId, pDnode->dnodeEp, true);
numOfMnodes = mnodeGetMnodesNum(); // Only create one mnode each time
if (numOfMnodes >= tsNumOfMnodes) return; return;
} }
} }
......
src/client/src/tscSql.c
浏览文件 @ c6479a14
...@@ -26,6 +26,7 @@ ...@@ -26,6 +26,7 @@
#include "tsclient.h" #include "tsclient.h"
#include "ttokendef.h" #include "ttokendef.h"
#include "tutil.h" #include "tutil.h"
#include "ttimer.h"
#include "tscProfile.h" #include "tscProfile.h"
static bool validImpl(const char* str, size_t maxsize) { static bool validImpl(const char* str, size_t maxsize) {
...@@ -568,18 +569,10 @@ static UNUSED_FUNC bool tscKillQueryInDnode(SSqlObj* pSql) { ...@@ -568,18 +569,10 @@ static UNUSED_FUNC bool tscKillQueryInDnode(SSqlObj* pSql) {
} }
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0); SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
if (pQueryInfo == NULL) { if ((pQueryInfo == NULL) || tscIsTwoStageSTableQuery(pQueryInfo, 0)) {
return true; return true;
} }
if (pSql->pRpcCtx != NULL) {
rpcCancelRequest(pSql->pRpcCtx);
pSql->pRpcCtx = NULL;
return true;
} else {
pSql->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
}
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
tscRemoveFromSqlList(pSql); tscRemoveFromSqlList(pSql);
...@@ -705,10 +698,12 @@ void taos_stop_query(TAOS_RES *res) { ...@@ -705,10 +698,12 @@ void taos_stop_query(TAOS_RES *res) {
if (tscIsTwoStageSTableQuery(pQueryInfo, 0)) { if (tscIsTwoStageSTableQuery(pQueryInfo, 0)) {
assert(pSql->pRpcCtx == NULL); assert(pSql->pRpcCtx == NULL);
tscKillSTableQuery(pSql); tscKillSTableQuery(pSql);
} else { // TODO multithreads bug } else {
if (pSql->cmd.command < TSDB_SQL_LOCAL && pSql->pRpcCtx != NULL) { if (pSql->cmd.command < TSDB_SQL_LOCAL) {
rpcCancelRequest(pSql->pRpcCtx); if (pSql->pRpcCtx != NULL) {
pSql->pRpcCtx = NULL; rpcCancelRequest(pSql->pRpcCtx);
pSql->pRpcCtx = NULL;
}
} }
} }
......
src/common/inc/tglobal.h
浏览文件 @ c6479a14
...@@ -34,6 +34,7 @@ extern int32_t tsStatusInterval; ...@@ -34,6 +34,7 @@ extern int32_t tsStatusInterval;
extern int32_t tsNumOfMnodes; extern int32_t tsNumOfMnodes;
extern int32_t tsEnableVnodeBak; extern int32_t tsEnableVnodeBak;
extern int32_t tsEnableTelemetryReporting; extern int32_t tsEnableTelemetryReporting;
extern char tsEmail[];
// common // common
extern int tsRpcTimer; extern int tsRpcTimer;
......
src/common/src/tglobal.c
浏览文件 @ c6479a14
...@@ -42,6 +42,7 @@ int32_t tsStatusInterval = 1; // second ...@@ -42,6 +42,7 @@ int32_t tsStatusInterval = 1; // second
int32_t tsNumOfMnodes = 3; int32_t tsNumOfMnodes = 3;
int32_t tsEnableVnodeBak = 1; int32_t tsEnableVnodeBak = 1;
int32_t tsEnableTelemetryReporting = 1; int32_t tsEnableTelemetryReporting = 1;
char tsEmail[TSDB_FQDN_LEN] = {0};
// common // common
int32_t tsRpcTimer = 1000; int32_t tsRpcTimer = 1000;
......
src/dnode/src/dnodeMPeer.c
浏览文件 @ c6479a14
...@@ -33,7 +33,8 @@ typedef struct { ...@@ -33,7 +33,8 @@ typedef struct {
} SMPeerWorker; } SMPeerWorker;
typedef struct { typedef struct {
int32_t num; int32_t curNum;
int32_t maxNum;
SMPeerWorker *peerWorker; SMPeerWorker *peerWorker;
} SMPeerWorkerPool; } SMPeerWorkerPool;
...@@ -46,37 +47,44 @@ static void *dnodeProcessMnodePeerQueue(void *param); ...@@ -46,37 +47,44 @@ static void *dnodeProcessMnodePeerQueue(void *param);
int32_t dnodeInitMnodePeer() { int32_t dnodeInitMnodePeer() {
tsMPeerQset = taosOpenQset(); tsMPeerQset = taosOpenQset();
tsMPeerPool.num = 1; tsMPeerPool.maxNum = 1;
tsMPeerPool.peerWorker = (SMPeerWorker *)calloc(sizeof(SMPeerWorker), tsMPeerPool.num); tsMPeerPool.curNum = 0;
tsMPeerPool.peerWorker = (SMPeerWorker *)calloc(sizeof(SMPeerWorker), tsMPeerPool.maxNum);
if (tsMPeerPool.peerWorker == NULL) return -1; if (tsMPeerPool.peerWorker == NULL) return -1;
for (int32_t i = 0; i < tsMPeerPool.num; ++i) { for (int32_t i = 0; i < tsMPeerPool.maxNum; ++i) {
SMPeerWorker *pWorker = tsMPeerPool.peerWorker + i; SMPeerWorker *pWorker = tsMPeerPool.peerWorker + i;
pWorker->workerId = i; pWorker->workerId = i;
dDebug("dnode mpeer worker:%d is created", i);
} }
dInfo("dnode mpeer is opened"); dDebug("dnode mpeer is opened, workers:%d qset:%p", tsMPeerPool.maxNum, tsMPeerQset);
return 0; return 0;
} }
void dnodeCleanupMnodePeer() { void dnodeCleanupMnodePeer() {
for (int32_t i = 0; i < tsMPeerPool.num; ++i) { for (int32_t i = 0; i < tsMPeerPool.maxNum; ++i) {
SMPeerWorker *pWorker = tsMPeerPool.peerWorker + i; SMPeerWorker *pWorker = tsMPeerPool.peerWorker + i;
if (pWorker->thread) { if (pWorker->thread) {
taosQsetThreadResume(tsMPeerQset); taosQsetThreadResume(tsMPeerQset);
} }
dDebug("dnode mpeer worker:%d is closed", i);
} }
for (int32_t i = 0; i < tsMPeerPool.num; ++i) { for (int32_t i = 0; i < tsMPeerPool.maxNum; ++i) {
SMPeerWorker *pWorker = tsMPeerPool.peerWorker + i; SMPeerWorker *pWorker = tsMPeerPool.peerWorker + i;
dDebug("dnode mpeer worker:%d start to join", i);
if (pWorker->thread) { if (pWorker->thread) {
pthread_join(pWorker->thread, NULL); pthread_join(pWorker->thread, NULL);
} }
dDebug("dnode mpeer worker:%d join success", i);
} }
dDebug("dnode mpeer is closed, qset:%p", tsMPeerQset);
taosCloseQset(tsMPeerQset); taosCloseQset(tsMPeerQset);
tsMPeerQset = NULL;
taosTFree(tsMPeerPool.peerWorker); taosTFree(tsMPeerPool.peerWorker);
dInfo("dnode mpeer is closed");
} }
int32_t dnodeAllocateMnodePqueue() { int32_t dnodeAllocateMnodePqueue() {
...@@ -85,7 +93,7 @@ int32_t dnodeAllocateMnodePqueue() { ...@@ -85,7 +93,7 @@ int32_t dnodeAllocateMnodePqueue() {
taosAddIntoQset(tsMPeerQset, tsMPeerQueue, NULL); taosAddIntoQset(tsMPeerQset, tsMPeerQueue, NULL);
for (int32_t i = 0; i < tsMPeerPool.num; ++i) { for (int32_t i = tsMPeerPool.curNum; i < tsMPeerPool.maxNum; ++i) {
SMPeerWorker *pWorker = tsMPeerPool.peerWorker + i; SMPeerWorker *pWorker = tsMPeerPool.peerWorker + i;
pWorker->workerId = i; pWorker->workerId = i;
...@@ -98,7 +106,9 @@ int32_t dnodeAllocateMnodePqueue() { ...@@ -98,7 +106,9 @@ int32_t dnodeAllocateMnodePqueue() {
} }
pthread_attr_destroy(&thAttr); pthread_attr_destroy(&thAttr);
dDebug("dnode mpeer worker:%d is launched, total:%d", pWorker->workerId, tsMPeerPool.num);
tsMPeerPool.curNum = i + 1;
dDebug("dnode mpeer worker:%d is launched, total:%d", pWorker->workerId, tsMPeerPool.maxNum);
} }
dDebug("dnode mpeer queue:%p is allocated", tsMPeerQueue); dDebug("dnode mpeer queue:%p is allocated", tsMPeerQueue);
...@@ -106,6 +116,7 @@ int32_t dnodeAllocateMnodePqueue() { ...@@ -106,6 +116,7 @@ int32_t dnodeAllocateMnodePqueue() {
} }
void dnodeFreeMnodePqueue() { void dnodeFreeMnodePqueue() {
dDebug("dnode mpeer queue:%p is freed", tsMPeerQueue);
taosCloseQueue(tsMPeerQueue); taosCloseQueue(tsMPeerQueue);
tsMPeerQueue = NULL; tsMPeerQueue = NULL;
} }
...@@ -148,7 +159,7 @@ static void *dnodeProcessMnodePeerQueue(void *param) { ...@@ -148,7 +159,7 @@ static void *dnodeProcessMnodePeerQueue(void *param) {
while (1) { while (1) {
if (taosReadQitemFromQset(tsMPeerQset, &type, (void **)&pPeerMsg, &unUsed) == 0) { if (taosReadQitemFromQset(tsMPeerQset, &type, (void **)&pPeerMsg, &unUsed) == 0) {
dDebug("dnodeProcessMnodePeerQueue: got no message from qset, exiting..."); dDebug("qset:%p, mnode peer got no message from qset, exiting", tsMPeerQset);
break; break;
} }
......
src/dnode/src/dnodeMRead.c
浏览文件 @ c6479a14
...@@ -33,7 +33,8 @@ typedef struct { ...@@ -33,7 +33,8 @@ typedef struct {
} SMReadWorker; } SMReadWorker;
typedef struct { typedef struct {
int32_t num; int32_t curNum;
int32_t maxNum;
SMReadWorker *readWorker; SMReadWorker *readWorker;
} SMReadWorkerPool; } SMReadWorkerPool;
...@@ -46,40 +47,46 @@ static void *dnodeProcessMnodeReadQueue(void *param); ...@@ -46,40 +47,46 @@ static void *dnodeProcessMnodeReadQueue(void *param);
int32_t dnodeInitMnodeRead() { int32_t dnodeInitMnodeRead() {
tsMReadQset = taosOpenQset(); tsMReadQset = taosOpenQset();
tsMReadPool.num = tsNumOfCores * tsNumOfThreadsPerCore / 2; tsMReadPool.maxNum = tsNumOfCores * tsNumOfThreadsPerCore / 2;
tsMReadPool.num = MAX(2, tsMReadPool.num); tsMReadPool.maxNum = MAX(2, tsMReadPool.maxNum);
tsMReadPool.num = MIN(4, tsMReadPool.num); tsMReadPool.maxNum = MIN(4, tsMReadPool.maxNum);
tsMReadPool.readWorker = (SMReadWorker *)calloc(sizeof(SMReadWorker), tsMReadPool.num); tsMReadPool.curNum = 0;
tsMReadPool.readWorker = (SMReadWorker *)calloc(sizeof(SMReadWorker), tsMReadPool.maxNum);
if (tsMReadPool.readWorker == NULL) return -1; if (tsMReadPool.readWorker == NULL) return -1;
for (int32_t i = 0; i < tsMReadPool.num; ++i) { for (int32_t i = 0; i < tsMReadPool.maxNum; ++i) {
SMReadWorker *pWorker = tsMReadPool.readWorker + i; SMReadWorker *pWorker = tsMReadPool.readWorker + i;
pWorker->workerId = i; pWorker->workerId = i;
dDebug("dnode mread worker:%d is created", i);
} }
dInfo("dnode mread is opened"); dDebug("dnode mread is opened, workers:%d qset:%p", tsMReadPool.maxNum, tsMReadQset);
return 0; return 0;
} }
void dnodeCleanupMnodeRead() { void dnodeCleanupMnodeRead() {
for (int32_t i = 0; i < tsMReadPool.num; ++i) { for (int32_t i = 0; i < tsMReadPool.maxNum; ++i) {
SMReadWorker *pWorker = tsMReadPool.readWorker + i; SMReadWorker *pWorker = tsMReadPool.readWorker + i;
if (pWorker->thread) { if (pWorker->thread) {
taosQsetThreadResume(tsMReadQset); taosQsetThreadResume(tsMReadQset);
} }
dDebug("dnode mread worker:%d is closed", i);
} }
for (int32_t i = 0; i < tsMReadPool.num; ++i) { for (int32_t i = 0; i < tsMReadPool.maxNum; ++i) {
SMReadWorker *pWorker = tsMReadPool.readWorker + i; SMReadWorker *pWorker = tsMReadPool.readWorker + i;
dDebug("dnode mread worker:%d start to join", i);
if (pWorker->thread) { if (pWorker->thread) {
pthread_join(pWorker->thread, NULL); pthread_join(pWorker->thread, NULL);
} }
dDebug("dnode mread worker:%d start to join", i);
} }
dDebug("dnode mread is closed, qset:%p", tsMReadQset);
taosCloseQset(tsMReadQset); taosCloseQset(tsMReadQset);
tsMReadQset = NULL;
free(tsMReadPool.readWorker); free(tsMReadPool.readWorker);
dInfo("dnode mread is closed");
} }
int32_t dnodeAllocateMnodeRqueue() { int32_t dnodeAllocateMnodeRqueue() {
...@@ -88,7 +95,7 @@ int32_t dnodeAllocateMnodeRqueue() { ...@@ -88,7 +95,7 @@ int32_t dnodeAllocateMnodeRqueue() {
taosAddIntoQset(tsMReadQset, tsMReadQueue, NULL); taosAddIntoQset(tsMReadQset, tsMReadQueue, NULL);
for (int32_t i = 0; i < tsMReadPool.num; ++i) { for (int32_t i = tsMReadPool.curNum; i < tsMReadPool.maxNum; ++i) {
SMReadWorker *pWorker = tsMReadPool.readWorker + i; SMReadWorker *pWorker = tsMReadPool.readWorker + i;
pWorker->workerId = i; pWorker->workerId = i;
...@@ -101,7 +108,8 @@ int32_t dnodeAllocateMnodeRqueue() { ...@@ -101,7 +108,8 @@ int32_t dnodeAllocateMnodeRqueue() {
} }
pthread_attr_destroy(&thAttr); pthread_attr_destroy(&thAttr);
dDebug("dnode mread worker:%d is launched, total:%d", pWorker->workerId, tsMReadPool.num); tsMReadPool.curNum = i + 1;
dDebug("dnode mread worker:%d is launched, total:%d", pWorker->workerId, tsMReadPool.maxNum);
} }
dDebug("dnode mread queue:%p is allocated", tsMReadQueue); dDebug("dnode mread queue:%p is allocated", tsMReadQueue);
...@@ -109,6 +117,7 @@ int32_t dnodeAllocateMnodeRqueue() { ...@@ -109,6 +117,7 @@ int32_t dnodeAllocateMnodeRqueue() {
} }
void dnodeFreeMnodeRqueue() { void dnodeFreeMnodeRqueue() {
dDebug("dnode mread queue:%p is freed", tsMReadQueue);
taosCloseQueue(tsMReadQueue); taosCloseQueue(tsMReadQueue);
tsMReadQueue = NULL; tsMReadQueue = NULL;
} }
...@@ -156,7 +165,7 @@ static void *dnodeProcessMnodeReadQueue(void *param) { ...@@ -156,7 +165,7 @@ static void *dnodeProcessMnodeReadQueue(void *param) {
while (1) { while (1) {
if (taosReadQitemFromQset(tsMReadQset, &type, (void **)&pReadMsg, &unUsed) == 0) { if (taosReadQitemFromQset(tsMReadQset, &type, (void **)&pReadMsg, &unUsed) == 0) {
dDebug("dnodeProcessMnodeReadQueue: got no message from qset, exiting..."); dDebug("qset:%p, mnode read got no message from qset, exiting", tsMReadQset);
break; break;
} }
......
src/dnode/src/dnodeMWrite.c
浏览文件 @ c6479a14
...@@ -34,7 +34,8 @@ typedef struct { ...@@ -34,7 +34,8 @@ typedef struct {
} SMWriteWorker; } SMWriteWorker;
typedef struct { typedef struct {
int32_t num; int32_t curNum;
int32_t maxNum;
SMWriteWorker *writeWorker; SMWriteWorker *writeWorker;
} SMWriteWorkerPool; } SMWriteWorkerPool;
...@@ -47,38 +48,45 @@ static void *dnodeProcessMnodeWriteQueue(void *param); ...@@ -47,38 +48,45 @@ static void *dnodeProcessMnodeWriteQueue(void *param);
int32_t dnodeInitMnodeWrite() { int32_t dnodeInitMnodeWrite() {
tsMWriteQset = taosOpenQset(); tsMWriteQset = taosOpenQset();
tsMWritePool.num = 1; tsMWritePool.maxNum = 1;
tsMWritePool.writeWorker = (SMWriteWorker *)calloc(sizeof(SMWriteWorker), tsMWritePool.num); tsMWritePool.curNum = 0;
tsMWritePool.writeWorker = (SMWriteWorker *)calloc(sizeof(SMWriteWorker), tsMWritePool.maxNum);
if (tsMWritePool.writeWorker == NULL) return -1; if (tsMWritePool.writeWorker == NULL) return -1;
for (int32_t i = 0; i < tsMWritePool.num; ++i) { for (int32_t i = 0; i < tsMWritePool.maxNum; ++i) {
SMWriteWorker *pWorker = tsMWritePool.writeWorker + i; SMWriteWorker *pWorker = tsMWritePool.writeWorker + i;
pWorker->workerId = i; pWorker->workerId = i;
dDebug("dnode mwrite worker:%d is created", i);
} }
dInfo("dnode mwrite is opened"); dDebug("dnode mwrite is opened, workers:%d qset:%p", tsMWritePool.maxNum, tsMWriteQset);
return 0; return 0;
} }
void dnodeCleanupMnodeWrite() { void dnodeCleanupMnodeWrite() {
for (int32_t i = 0; i < tsMWritePool.num; ++i) { for (int32_t i = 0; i < tsMWritePool.maxNum; ++i) {
SMWriteWorker *pWorker = tsMWritePool.writeWorker + i; SMWriteWorker *pWorker = tsMWritePool.writeWorker + i;
if (pWorker->thread) { if (pWorker->thread) {
taosQsetThreadResume(tsMWriteQset); taosQsetThreadResume(tsMWriteQset);
} }
dDebug("dnode mwrite worker:%d is closed", i);
} }
for (int32_t i = 0; i < tsMWritePool.num; ++i) { for (int32_t i = 0; i < tsMWritePool.maxNum; ++i) {
SMWriteWorker *pWorker = tsMWritePool.writeWorker + i; SMWriteWorker *pWorker = tsMWritePool.writeWorker + i;
dDebug("dnode mwrite worker:%d start to join", i);
if (pWorker->thread) { if (pWorker->thread) {
pthread_join(pWorker->thread, NULL); pthread_join(pWorker->thread, NULL);
} }
dDebug("dnode mwrite worker:%d join success", i);
} }
dDebug("dnode mwrite is closed, qset:%p", tsMWriteQset);
taosCloseQset(tsMWriteQset); taosCloseQset(tsMWriteQset);
tsMWriteQset = NULL;
taosTFree(tsMWritePool.writeWorker); taosTFree(tsMWritePool.writeWorker);
dInfo("dnode mwrite is closed");
} }
int32_t dnodeAllocateMnodeWqueue() { int32_t dnodeAllocateMnodeWqueue() {
...@@ -87,7 +95,7 @@ int32_t dnodeAllocateMnodeWqueue() { ...@@ -87,7 +95,7 @@ int32_t dnodeAllocateMnodeWqueue() {
taosAddIntoQset(tsMWriteQset, tsMWriteQueue, NULL); taosAddIntoQset(tsMWriteQset, tsMWriteQueue, NULL);
for (int32_t i = 0; i < tsMWritePool.num; ++i) { for (int32_t i = tsMWritePool.curNum; i < tsMWritePool.maxNum; ++i) {
SMWriteWorker *pWorker = tsMWritePool.writeWorker + i; SMWriteWorker *pWorker = tsMWritePool.writeWorker + i;
pWorker->workerId = i; pWorker->workerId = i;
...@@ -100,7 +108,8 @@ int32_t dnodeAllocateMnodeWqueue() { ...@@ -100,7 +108,8 @@ int32_t dnodeAllocateMnodeWqueue() {
} }
pthread_attr_destroy(&thAttr); pthread_attr_destroy(&thAttr);
dDebug("dnode mwrite worker:%d is launched, total:%d", pWorker->workerId, tsMWritePool.num); tsMWritePool.curNum = i + 1;
dDebug("dnode mwrite worker:%d is launched, total:%d", pWorker->workerId, tsMWritePool.maxNum);
} }
dDebug("dnode mwrite queue:%p is allocated", tsMWriteQueue); dDebug("dnode mwrite queue:%p is allocated", tsMWriteQueue);
...@@ -108,6 +117,7 @@ int32_t dnodeAllocateMnodeWqueue() { ...@@ -108,6 +117,7 @@ int32_t dnodeAllocateMnodeWqueue() {
} }
void dnodeFreeMnodeWqueue() { void dnodeFreeMnodeWqueue() {
dDebug("dnode mwrite queue:%p is freed", tsMWriteQueue);
taosCloseQueue(tsMWriteQueue); taosCloseQueue(tsMWriteQueue);
tsMWriteQueue = NULL; tsMWriteQueue = NULL;
} }
...@@ -122,11 +132,15 @@ void dnodeDispatchToMnodeWriteQueue(SRpcMsg *pMsg) { ...@@ -122,11 +132,15 @@ void dnodeDispatchToMnodeWriteQueue(SRpcMsg *pMsg) {
SMnodeMsg *pWrite = (SMnodeMsg *)taosAllocateQitem(sizeof(SMnodeMsg)); SMnodeMsg *pWrite = (SMnodeMsg *)taosAllocateQitem(sizeof(SMnodeMsg));
mnodeCreateMsg(pWrite, pMsg); mnodeCreateMsg(pWrite, pMsg);
dDebug("app:%p:%p, msg:%s is put into mwrite queue", pWrite->rpcMsg.ahandle, pWrite, taosMsg[pWrite->rpcMsg.msgType]); dDebug("app:%p:%p, msg:%s is put into mwrite queue:%p", pWrite->rpcMsg.ahandle, pWrite,
taosMsg[pWrite->rpcMsg.msgType], tsMWriteQueue);
taosWriteQitem(tsMWriteQueue, TAOS_QTYPE_RPC, pWrite); taosWriteQitem(tsMWriteQueue, TAOS_QTYPE_RPC, pWrite);
} }
static void dnodeFreeMnodeWriteMsg(SMnodeMsg *pWrite) { static void dnodeFreeMnodeWriteMsg(SMnodeMsg *pWrite) {
dDebug("app:%p:%p, msg:%s is freed from mwrite queue:%p", pWrite->rpcMsg.ahandle, pWrite,
taosMsg[pWrite->rpcMsg.msgType], tsMWriteQueue);
mnodeCleanupMsg(pWrite); mnodeCleanupMsg(pWrite);
taosFreeQitem(pWrite); taosFreeQitem(pWrite);
} }
...@@ -158,7 +172,7 @@ static void *dnodeProcessMnodeWriteQueue(void *param) { ...@@ -158,7 +172,7 @@ static void *dnodeProcessMnodeWriteQueue(void *param) {
while (1) { while (1) {
if (taosReadQitemFromQset(tsMWriteQset, &type, (void **)&pWrite, &unUsed) == 0) { if (taosReadQitemFromQset(tsMWriteQset, &type, (void **)&pWrite, &unUsed) == 0) {
dDebug("dnodeProcessMnodeWriteQueue: got no message from qset, exiting..."); dDebug("qset:%p, mnode write got no message from qset, exiting", tsMWriteQset);
break; break;
} }
...@@ -182,8 +196,8 @@ void dnodeReprocessMnodeWriteMsg(void *pMsg) { ...@@ -182,8 +196,8 @@ void dnodeReprocessMnodeWriteMsg(void *pMsg) {
dnodeSendRedirectMsg(pMsg, true); dnodeSendRedirectMsg(pMsg, true);
dnodeFreeMnodeWriteMsg(pWrite); dnodeFreeMnodeWriteMsg(pWrite);
} else { } else {
dDebug("app:%p:%p, msg:%s is reput into mwrite queue, retry times:%d", pWrite->rpcMsg.ahandle, pWrite, dDebug("app:%p:%p, msg:%s is reput into mwrite queue:%p, retry times:%d", pWrite->rpcMsg.ahandle, pWrite,
taosMsg[pWrite->rpcMsg.msgType], pWrite->retry); taosMsg[pWrite->rpcMsg.msgType], tsMWriteQueue, pWrite->retry);
taosWriteQitem(tsMWriteQueue, TAOS_QTYPE_RPC, pWrite); taosWriteQitem(tsMWriteQueue, TAOS_QTYPE_RPC, pWrite);
} }
......
src/dnode/src/dnodeMgmt.c
浏览文件 @ c6479a14
...@@ -74,14 +74,16 @@ static int32_t dnodeProcessAlterVnodeMsg(SRpcMsg *pMsg); ...@@ -74,14 +74,16 @@ static int32_t dnodeProcessAlterVnodeMsg(SRpcMsg *pMsg);
static int32_t dnodeProcessDropVnodeMsg(SRpcMsg *pMsg); static int32_t dnodeProcessDropVnodeMsg(SRpcMsg *pMsg);
static int32_t dnodeProcessAlterStreamMsg(SRpcMsg *pMsg); static int32_t dnodeProcessAlterStreamMsg(SRpcMsg *pMsg);
static int32_t dnodeProcessConfigDnodeMsg(SRpcMsg *pMsg); static int32_t dnodeProcessConfigDnodeMsg(SRpcMsg *pMsg);
static int32_t dnodeProcessCreateMnodeMsg(SRpcMsg *pMsg);
static int32_t (*dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *pMsg); static int32_t (*dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *pMsg);
int32_t dnodeInitMgmt() { int32_t dnodeInitMgmt() {
dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_CREATE_VNODE] = dnodeProcessCreateVnodeMsg; dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_CREATE_VNODE] = dnodeProcessCreateVnodeMsg;
dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_ALTER_VNODE] = dnodeProcessAlterVnodeMsg; dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_ALTER_VNODE] = dnodeProcessAlterVnodeMsg;
dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_DROP_VNODE] = dnodeProcessDropVnodeMsg; dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_DROP_VNODE] = dnodeProcessDropVnodeMsg;
dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_ALTER_STREAM] = dnodeProcessAlterStreamMsg; dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_ALTER_STREAM] = dnodeProcessAlterStreamMsg;
dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_CONFIG_DNODE] = dnodeProcessConfigDnodeMsg; dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_CONFIG_DNODE] = dnodeProcessConfigDnodeMsg;
dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_CREATE_MNODE] = dnodeProcessCreateMnodeMsg;
dnodeAddClientRspHandle(TSDB_MSG_TYPE_DM_STATUS_RSP, dnodeProcessStatusRsp); dnodeAddClientRspHandle(TSDB_MSG_TYPE_DM_STATUS_RSP, dnodeProcessStatusRsp);
dnodeReadDnodeCfg(); dnodeReadDnodeCfg();
...@@ -226,7 +228,7 @@ static void *dnodeProcessMgmtQueue(void *param) { ...@@ -226,7 +228,7 @@ static void *dnodeProcessMgmtQueue(void *param) {
while (1) { while (1) {
if (taosReadQitemFromQset(tsMgmtQset, &type, (void **) &pMsg, &handle) == 0) { if (taosReadQitemFromQset(tsMgmtQset, &type, (void **) &pMsg, &handle) == 0) {
dDebug("dnode mgmt got no message from qset, exit ..."); dDebug("qset:%p, dnode mgmt got no message from qset, exit", tsMgmtQset);
break; break;
} }
...@@ -451,10 +453,34 @@ static int32_t dnodeProcessAlterStreamMsg(SRpcMsg *pMsg) { ...@@ -451,10 +453,34 @@ static int32_t dnodeProcessAlterStreamMsg(SRpcMsg *pMsg) {
} }
static int32_t dnodeProcessConfigDnodeMsg(SRpcMsg *pMsg) { static int32_t dnodeProcessConfigDnodeMsg(SRpcMsg *pMsg) {
SMDCfgDnodeMsg *pCfg = (SMDCfgDnodeMsg *)pMsg->pCont; SMDCfgDnodeMsg *pCfg = pMsg->pCont;
return taosCfgDynamicOptions(pCfg->config); return taosCfgDynamicOptions(pCfg->config);
} }
static int32_t dnodeProcessCreateMnodeMsg(SRpcMsg *pMsg) {
SMDCreateMnodeMsg *pCfg = pMsg->pCont;
pCfg->dnodeId = htonl(pCfg->dnodeId);
if (pCfg->dnodeId != dnodeGetDnodeId()) {
dError("dnodeId:%d, in create mnode msg is not equal with saved dnodeId:%d", pCfg->dnodeId, dnodeGetDnodeId());
return TSDB_CODE_MND_DNODE_ID_NOT_CONFIGURED;
}
if (strcmp(pCfg->dnodeEp, tsLocalEp) != 0) {
dError("dnodeEp:%s, in create mnode msg is not equal with saved dnodeEp:%s", pCfg->dnodeEp, tsLocalEp);
return TSDB_CODE_MND_DNODE_EP_NOT_CONFIGURED;
}
dDebug("dnodeId:%d, create mnode msg is received from mnodes, numOfMnodes:%d", pCfg->dnodeId, pCfg->mnodes.nodeNum);
for (int i = 0; i < pCfg->mnodes.nodeNum; ++i) {
pCfg->mnodes.nodeInfos[i].nodeId = htonl(pCfg->mnodes.nodeInfos[i].nodeId);
dDebug("mnode index:%d, mnode:%d:%s", i, pCfg->mnodes.nodeInfos[i].nodeId, pCfg->mnodes.nodeInfos[i].nodeEp);
}
dnodeStartMnode(&pCfg->mnodes);
return TSDB_CODE_SUCCESS;
}
void dnodeUpdateMnodeEpSetForPeer(SRpcEpSet *pEpSet) { void dnodeUpdateMnodeEpSetForPeer(SRpcEpSet *pEpSet) {
if (pEpSet->numOfEps <= 0) { if (pEpSet->numOfEps <= 0) {
dError("mnode EP list for peer is changed, but content is invalid, discard it"); dError("mnode EP list for peer is changed, but content is invalid, discard it");
...@@ -465,29 +491,6 @@ void dnodeUpdateMnodeEpSetForPeer(SRpcEpSet *pEpSet) { ...@@ -465,29 +491,6 @@ void dnodeUpdateMnodeEpSetForPeer(SRpcEpSet *pEpSet) {
for (int i = 0; i < pEpSet->numOfEps; ++i) { for (int i = 0; i < pEpSet->numOfEps; ++i) {
pEpSet->port[i] -= TSDB_PORT_DNODEDNODE; pEpSet->port[i] -= TSDB_PORT_DNODEDNODE;
dInfo("mnode index:%d %s:%u", i, pEpSet->fqdn[i], pEpSet->port[i]); dInfo("mnode index:%d %s:%u", i, pEpSet->fqdn[i], pEpSet->port[i]);
if (!mnodeIsRunning()) {
if (strcmp(pEpSet->fqdn[i], tsLocalFqdn) == 0 && pEpSet->port[i] == tsServerPort) {
dInfo("mnode index:%d %s:%u should work as mnode", i, pEpSet->fqdn[i], pEpSet->port[i]);
bool find = false;
for (int i = 0; i < tsDMnodeInfos.nodeNum; ++i) {
if (tsDMnodeInfos.nodeInfos[i].nodeId == dnodeGetDnodeId()) {
dInfo("localEp found in mnode infos");
find = true;
break;
}
}
if (!find) {
dInfo("localEp not found in mnode infos, will set into mnode infos");
tstrncpy(tsDMnodeInfos.nodeInfos[tsDMnodeInfos.nodeNum].nodeEp, tsLocalEp, TSDB_EP_LEN);
tsDMnodeInfos.nodeInfos[tsDMnodeInfos.nodeNum].nodeId = dnodeGetDnodeId();
tsDMnodeInfos.nodeNum++;
}
dnodeStartMnode();
}
}
} }
tsDMnodeEpSet = *pEpSet; tsDMnodeEpSet = *pEpSet;
...@@ -532,7 +535,9 @@ static void dnodeProcessStatusRsp(SRpcMsg *pMsg) { ...@@ -532,7 +535,9 @@ static void dnodeProcessStatusRsp(SRpcMsg *pMsg) {
} }
vnodeSetAccess(pStatusRsp->vgAccess, pCfg->numOfVnodes); vnodeSetAccess(pStatusRsp->vgAccess, pCfg->numOfVnodes);
dnodeProcessModuleStatus(pCfg->moduleStatus);
// will not set mnode in status msg
// dnodeProcessModuleStatus(pCfg->moduleStatus);
dnodeUpdateDnodeCfg(pCfg); dnodeUpdateDnodeCfg(pCfg);
dnodeUpdateMnodeInfos(pMnodes); dnodeUpdateMnodeInfos(pMnodes);
...@@ -576,7 +581,7 @@ static void dnodeUpdateMnodeInfos(SDMMnodeInfos *pMnodes) { ...@@ -576,7 +581,7 @@ static void dnodeUpdateMnodeInfos(SDMMnodeInfos *pMnodes) {
} }
dnodeSaveMnodeInfos(); dnodeSaveMnodeInfos();
sdbUpdateSync(); sdbUpdateAsync();
} }
static bool dnodeReadMnodeInfos() { static bool dnodeReadMnodeInfos() {
......
src/dnode/src/dnodeModule.c
浏览文件 @ c6479a14
...@@ -146,7 +146,9 @@ void dnodeProcessModuleStatus(uint32_t moduleStatus) { ...@@ -146,7 +146,9 @@ void dnodeProcessModuleStatus(uint32_t moduleStatus) {
} }
} }
bool dnodeStartMnode() { bool dnodeStartMnode(void *pMnodes) {
SDMMnodeInfos *mnodes = pMnodes;
if (tsModuleStatus & (1 << TSDB_MOD_MNODE)) { if (tsModuleStatus & (1 << TSDB_MOD_MNODE)) {
dDebug("mnode module is already started, module status:%d", tsModuleStatus); dDebug("mnode module is already started, module status:%d", tsModuleStatus);
return false; return false;
...@@ -156,6 +158,7 @@ bool dnodeStartMnode() { ...@@ -156,6 +158,7 @@ bool dnodeStartMnode() {
dInfo("start mnode module, module status:%d, new status:%d", tsModuleStatus, moduleStatus); dInfo("start mnode module, module status:%d, new status:%d", tsModuleStatus, moduleStatus);
dnodeProcessModuleStatus(moduleStatus); dnodeProcessModuleStatus(moduleStatus);
sdbUpdateSync(); sdbUpdateSync(mnodes);
return true; return true;
} }
src/dnode/src/dnodePeer.c
浏览文件 @ c6479a14
...@@ -48,6 +48,7 @@ int32_t dnodeInitServer() { ...@@ -48,6 +48,7 @@ int32_t dnodeInitServer() {
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_DROP_VNODE] = dnodeDispatchToMgmtQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_DROP_VNODE] = dnodeDispatchToMgmtQueue;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_ALTER_STREAM] = dnodeDispatchToMgmtQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_ALTER_STREAM] = dnodeDispatchToMgmtQueue;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CONFIG_DNODE] = dnodeDispatchToMgmtQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CONFIG_DNODE] = dnodeDispatchToMgmtQueue;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CREATE_MNODE] = dnodeDispatchToMgmtQueue;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_CONFIG_TABLE] = dnodeDispatchToMnodePeerQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_CONFIG_TABLE] = dnodeDispatchToMnodePeerQueue;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_CONFIG_VNODE] = dnodeDispatchToMnodePeerQueue; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_CONFIG_VNODE] = dnodeDispatchToMnodePeerQueue;
...@@ -170,8 +171,12 @@ void dnodeSendMsgToDnode(SRpcEpSet *epSet, SRpcMsg *rpcMsg) { ...@@ -170,8 +171,12 @@ void dnodeSendMsgToDnode(SRpcEpSet *epSet, SRpcMsg *rpcMsg) {
rpcSendRequest(tsDnodeClientRpc, epSet, rpcMsg); rpcSendRequest(tsDnodeClientRpc, epSet, rpcMsg);
} }
void dnodeSendMsgToDnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp) { void dnodeSendMsgToMnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp) {
SRpcEpSet epSet = {0}; SRpcEpSet epSet = {0};
dnodeGetMnodeEpSetForPeer(&epSet); dnodeGetMnodeEpSetForPeer(&epSet);
rpcSendRecv(tsDnodeClientRpc, &epSet, rpcMsg, rpcRsp); rpcSendRecv(tsDnodeClientRpc, &epSet, rpcMsg, rpcRsp);
} }
void dnodeSendMsgToDnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp, SRpcEpSet *epSet) {
rpcSendRecv(tsDnodeClientRpc, epSet, rpcMsg, rpcRsp);
}
\ No newline at end of file
src/dnode/src/dnodeShell.c
浏览文件 @ c6479a14
...@@ -156,7 +156,7 @@ static int dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char ...@@ -156,7 +156,7 @@ static int dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char
dDebug("user:%s, send auth msg to mnodes", user); dDebug("user:%s, send auth msg to mnodes", user);
SRpcMsg rpcRsp = {0}; SRpcMsg rpcRsp = {0};
dnodeSendMsgToDnodeRecv(&rpcMsg, &rpcRsp); dnodeSendMsgToMnodeRecv(&rpcMsg, &rpcRsp);
if (rpcRsp.code != 0) { if (rpcRsp.code != 0) {
dError("user:%s, auth msg received from mnodes, error:%s", user, tstrerror(rpcRsp.code)); dError("user:%s, auth msg received from mnodes, error:%s", user, tstrerror(rpcRsp.code));
...@@ -189,7 +189,7 @@ void *dnodeSendCfgTableToRecv(int32_t vgId, int32_t sid) { ...@@ -189,7 +189,7 @@ void *dnodeSendCfgTableToRecv(int32_t vgId, int32_t sid) {
rpcMsg.msgType = TSDB_MSG_TYPE_DM_CONFIG_TABLE; rpcMsg.msgType = TSDB_MSG_TYPE_DM_CONFIG_TABLE;
SRpcMsg rpcRsp = {0}; SRpcMsg rpcRsp = {0};
dnodeSendMsgToDnodeRecv(&rpcMsg, &rpcRsp); dnodeSendMsgToMnodeRecv(&rpcMsg, &rpcRsp);
terrno = rpcRsp.code; terrno = rpcRsp.code;
if (rpcRsp.code != 0) { if (rpcRsp.code != 0) {
......
src/dnode/src/dnodeTelemetry.c
浏览文件 @ c6479a14
...@@ -177,7 +177,8 @@ static void addMemoryInfo(SBufferWriter* bw) { ...@@ -177,7 +177,8 @@ static void addMemoryInfo(SBufferWriter* bw) {
static void addVersionInfo(SBufferWriter* bw) { static void addVersionInfo(SBufferWriter* bw) {
addStringField(bw, "version", version); addStringField(bw, "version", version);
addStringField(bw, "buildInfo", buildinfo); addStringField(bw, "buildInfo", buildinfo);
addStringField(bw, "gitInfo", gitinfo); addStringField(bw, "gitInfo", gitinfo);
addStringField(bw, "email", tsEmail);
} }
static void addRuntimeInfo(SBufferWriter* bw) { static void addRuntimeInfo(SBufferWriter* bw) {
...@@ -261,11 +262,27 @@ static void* telemetryThread(void* param) { ...@@ -261,11 +262,27 @@ static void* telemetryThread(void* param) {
return NULL; return NULL;
} }
static void dnodeGetEmail(char* filepath) {
int fd = open(filepath, O_RDONLY);
if (fd < 0) {
return;
}
if (taosTRead(fd, (void *)tsEmail, TSDB_FQDN_LEN) < 0) {
dError("failed to read %d bytes from file %s since %s", TSDB_FQDN_LEN, filepath, strerror(errno));
}
close(fd);
}
int32_t dnodeInitTelemetry() { int32_t dnodeInitTelemetry() {
if (!tsEnableTelemetryReporting) { if (!tsEnableTelemetryReporting) {
return 0; return 0;
} }
dnodeGetEmail("/usr/local/taos/email");
if (tsem_init(&tsExitSem, 0, 0) == -1) { if (tsem_init(&tsExitSem, 0, 0) == -1) {
// just log the error, it is ok for telemetry to fail // just log the error, it is ok for telemetry to fail
dTrace("failed to create semaphore for telemetry, reason:%s", strerror(errno)); dTrace("failed to create semaphore for telemetry, reason:%s", strerror(errno));
......
src/dnode/src/dnodeVRead.c
浏览文件 @ c6479a14
...@@ -199,7 +199,7 @@ static void *dnodeProcessReadQueue(void *param) { ...@@ -199,7 +199,7 @@ static void *dnodeProcessReadQueue(void *param) {
while (1) { while (1) {
if (taosReadQitemFromQset(readQset, &type, (void **)&pReadMsg, &pVnode) == 0) { if (taosReadQitemFromQset(readQset, &type, (void **)&pReadMsg, &pVnode) == 0) {
dDebug("dnodeProcessReadQueee: got no message from qset, exiting..."); dDebug("qset:%p dnode read got no message from qset, exiting", readQset);
break; break;
} }
......
src/dnode/src/dnodeVWrite.c
浏览文件 @ c6479a14
...@@ -222,7 +222,7 @@ static void *dnodeProcessWriteQueue(void *param) { ...@@ -222,7 +222,7 @@ static void *dnodeProcessWriteQueue(void *param) {
while (1) { while (1) {
numOfMsgs = taosReadAllQitemsFromQset(pWorker->qset, pWorker->qall, &pVnode); numOfMsgs = taosReadAllQitemsFromQset(pWorker->qset, pWorker->qall, &pVnode);
if (numOfMsgs == 0) { if (numOfMsgs == 0) {
dDebug("dnodeProcessWriteQueee: got no message from qset, exiting..."); dDebug("qset:%p, dnode write got no message from qset, exiting", pWorker->qset);
break; break;
} }
......
src/inc/dnode.h
浏览文件 @ c6479a14
...@@ -43,11 +43,12 @@ void dnodeGetMnodeEpSetForPeer(void *epSet); ...@@ -43,11 +43,12 @@ void dnodeGetMnodeEpSetForPeer(void *epSet);
void dnodeGetMnodeEpSetForShell(void *epSet); void dnodeGetMnodeEpSetForShell(void *epSet);
void * dnodeGetMnodeInfos(); void * dnodeGetMnodeInfos();
int32_t dnodeGetDnodeId(); int32_t dnodeGetDnodeId();
bool dnodeStartMnode(); bool dnodeStartMnode(void *pModes);
void dnodeAddClientRspHandle(uint8_t msgType, void (*fp)(SRpcMsg *rpcMsg)); void dnodeAddClientRspHandle(uint8_t msgType, void (*fp)(SRpcMsg *rpcMsg));
void dnodeSendMsgToDnode(SRpcEpSet *epSet, SRpcMsg *rpcMsg); void dnodeSendMsgToDnode(SRpcEpSet *epSet, SRpcMsg *rpcMsg);
void dnodeSendMsgToDnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp); void dnodeSendMsgToMnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp);
void dnodeSendMsgToDnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp, SRpcEpSet *epSet);
void *dnodeSendCfgTableToRecv(int32_t vgId, int32_t sid); void *dnodeSendCfgTableToRecv(int32_t vgId, int32_t sid);
void *dnodeAllocateVnodeWqueue(void *pVnode); void *dnodeAllocateVnodeWqueue(void *pVnode);
......
src/inc/mnode.h
浏览文件 @ c6479a14
...@@ -60,7 +60,8 @@ int32_t mnodeInitSystem(); ...@@ -60,7 +60,8 @@ int32_t mnodeInitSystem();
int32_t mnodeStartSystem(); int32_t mnodeStartSystem();
void mnodeCleanupSystem(); void mnodeCleanupSystem();
void mnodeStopSystem(); void mnodeStopSystem();
void sdbUpdateSync(); void sdbUpdateAsync();
void sdbUpdateSync(void *pMnodes);
bool mnodeIsRunning(); bool mnodeIsRunning();
int32_t mnodeProcessRead(SMnodeMsg *pMsg); int32_t mnodeProcessRead(SMnodeMsg *pMsg);
int32_t mnodeProcessWrite(SMnodeMsg *pMsg); int32_t mnodeProcessWrite(SMnodeMsg *pMsg);
......
src/inc/taoserror.h
浏览文件 @ c6479a14
...@@ -139,6 +139,8 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_VGROUP_ALREADY_IN_DNODE, 0, 0x0339, "Vgroup alr ...@@ -139,6 +139,8 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_VGROUP_ALREADY_IN_DNODE, 0, 0x0339, "Vgroup alr
TAOS_DEFINE_ERROR(TSDB_CODE_MND_DNODE_NOT_FREE, 0, 0x033A, "Dnode not avaliable") TAOS_DEFINE_ERROR(TSDB_CODE_MND_DNODE_NOT_FREE, 0, 0x033A, "Dnode not avaliable")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_CLUSTER_ID, 0, 0x033B, "Cluster id not match") TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_CLUSTER_ID, 0, 0x033B, "Cluster id not match")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_NOT_READY, 0, 0x033C, "Cluster not ready") TAOS_DEFINE_ERROR(TSDB_CODE_MND_NOT_READY, 0, 0x033C, "Cluster not ready")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_DNODE_ID_NOT_CONFIGURED, 0, 0x033D, "Dnode Id not configured")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_DNODE_EP_NOT_CONFIGURED, 0, 0x033E, "Dnode Ep not configured")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_ACCT_ALREADY_EXIST, 0, 0x0340, "Account already exists") TAOS_DEFINE_ERROR(TSDB_CODE_MND_ACCT_ALREADY_EXIST, 0, 0x0340, "Account already exists")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_ACCT, 0, 0x0341, "Invalid account") TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_ACCT, 0, 0x0341, "Invalid account")
......
src/inc/taosmsg.h
浏览文件 @ c6479a14
...@@ -59,7 +59,7 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_DROP_STABLE, "drop-stable" ) ...@@ -59,7 +59,7 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_DROP_STABLE, "drop-stable" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_ALTER_STREAM, "alter-stream" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_ALTER_STREAM, "alter-stream" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_CONFIG_DNODE, "config-dnode" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_CONFIG_DNODE, "config-dnode" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_ALTER_VNODE, "alter-vnode" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_ALTER_VNODE, "alter-vnode" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DUMMY5, "dummy5" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_CREATE_MNODE, "create-mnode" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DUMMY6, "dummy6" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DUMMY6, "dummy6" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DUMMY7, "dummy7" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DUMMY7, "dummy7" )
...@@ -721,6 +721,12 @@ typedef struct { ...@@ -721,6 +721,12 @@ typedef struct {
char ep[TSDB_EP_LEN]; // end point, hostname:port char ep[TSDB_EP_LEN]; // end point, hostname:port
} SCMCreateDnodeMsg, SCMDropDnodeMsg; } SCMCreateDnodeMsg, SCMDropDnodeMsg;
typedef struct {
int32_t dnodeId;
char dnodeEp[TSDB_EP_LEN]; // end point, hostname:port
SDMMnodeInfos mnodes;
} SMDCreateMnodeMsg;
typedef struct { typedef struct {
int32_t dnodeId; int32_t dnodeId;
int32_t vgId; int32_t vgId;
......
src/mnode/inc/mnodeMnode.h
浏览文件 @ c6479a14
...@@ -31,7 +31,7 @@ typedef enum { ...@@ -31,7 +31,7 @@ typedef enum {
int32_t mnodeInitMnodes(); int32_t mnodeInitMnodes();
void mnodeCleanupMnodes(); void mnodeCleanupMnodes();
int32_t mnodeAddMnode(int32_t dnodeId); void mnodeCreateMnode(int32_t dnodeId, char *dnodeEp, bool needConfirm);
int32_t mnodeDropMnode(int32_t dnodeId); int32_t mnodeDropMnode(int32_t dnodeId);
void mnodeDropMnodeLocal(int32_t dnodeId); void mnodeDropMnodeLocal(int32_t dnodeId);
......
src/mnode/src/mnodeCluster.c
浏览文件 @ c6479a14
...@@ -224,7 +224,7 @@ static int32_t mnodeRetrieveClusters(SShowObj *pShow, char *data, int32_t rows, ...@@ -224,7 +224,7 @@ static int32_t mnodeRetrieveClusters(SShowObj *pShow, char *data, int32_t rows,
mnodeDecClusterRef(pCluster); mnodeDecClusterRef(pCluster);
numOfRows++; numOfRows++;
} }
mnodeVacuumResult(data, cols, numOfRows, rows, pShow);
pShow->numOfReads += numOfRows; pShow->numOfReads += numOfRows;
return numOfRows; return numOfRows;
} }
src/mnode/src/mnodeDnode.c
浏览文件 @ c6479a14
...@@ -147,7 +147,7 @@ static int32_t mnodeDnodeActionRestored() { ...@@ -147,7 +147,7 @@ static int32_t mnodeDnodeActionRestored() {
mnodeCreateDnode(tsLocalEp, NULL); mnodeCreateDnode(tsLocalEp, NULL);
SDnodeObj *pDnode = mnodeGetDnodeByEp(tsLocalEp); SDnodeObj *pDnode = mnodeGetDnodeByEp(tsLocalEp);
if (pDnode != NULL) { if (pDnode != NULL) {
mnodeAddMnode(pDnode->dnodeId); mnodeCreateMnode(pDnode->dnodeId, pDnode->dnodeEp, false);
mnodeDecDnodeRef(pDnode); mnodeDecDnodeRef(pDnode);
} }
} }
...@@ -857,6 +857,7 @@ int32_t mnodeRetrieveModules(SShowObj *pShow, char *data, int32_t rows, void *pC ...@@ -857,6 +857,7 @@ int32_t mnodeRetrieveModules(SShowObj *pShow, char *data, int32_t rows, void *pC
char* pWrite; char* pWrite;
char* moduleName[5] = { "MNODE", "HTTP", "MONITOR", "MQTT", "UNKNOWN" }; char* moduleName[5] = { "MNODE", "HTTP", "MONITOR", "MQTT", "UNKNOWN" };
int32_t cols;
while (numOfRows < rows) { while (numOfRows < rows) {
SDnodeObj *pDnode = NULL; SDnodeObj *pDnode = NULL;
...@@ -864,7 +865,7 @@ int32_t mnodeRetrieveModules(SShowObj *pShow, char *data, int32_t rows, void *pC ...@@ -864,7 +865,7 @@ int32_t mnodeRetrieveModules(SShowObj *pShow, char *data, int32_t rows, void *pC
if (pDnode == NULL) break; if (pDnode == NULL) break;
for (int32_t moduleType = 0; moduleType < TSDB_MOD_MAX; ++moduleType) { for (int32_t moduleType = 0; moduleType < TSDB_MOD_MAX; ++moduleType) {
int32_t cols = 0; cols = 0;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows; pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(int16_t *)pWrite = pDnode->dnodeId; *(int16_t *)pWrite = pDnode->dnodeId;
...@@ -890,6 +891,7 @@ int32_t mnodeRetrieveModules(SShowObj *pShow, char *data, int32_t rows, void *pC ...@@ -890,6 +891,7 @@ int32_t mnodeRetrieveModules(SShowObj *pShow, char *data, int32_t rows, void *pC
mnodeDecDnodeRef(pDnode); mnodeDecDnodeRef(pDnode);
} }
mnodeVacuumResult(data, cols, numOfRows, rows, pShow);
pShow->numOfReads += numOfRows; pShow->numOfReads += numOfRows;
return numOfRows; return numOfRows;
...@@ -1081,6 +1083,7 @@ static int32_t mnodeRetrieveVnodes(SShowObj *pShow, char *data, int32_t rows, vo ...@@ -1081,6 +1083,7 @@ static int32_t mnodeRetrieveVnodes(SShowObj *pShow, char *data, int32_t rows, vo
} else { } else {
numOfRows = 0; numOfRows = 0;
} }
mnodeVacuumResult(data, cols, numOfRows, rows, pShow);
pShow->numOfReads += numOfRows; pShow->numOfReads += numOfRows;
return numOfRows; return numOfRows;
......
src/mnode/src/mnodeMain.c
浏览文件 @ c6479a14
...@@ -109,7 +109,7 @@ int32_t mnodeStartSystem() { ...@@ -109,7 +109,7 @@ int32_t mnodeStartSystem() {
mInfo("mnode is initialized successfully"); mInfo("mnode is initialized successfully");
sdbUpdateSync(); sdbUpdateSync(NULL);
return 0; return 0;
} }
......
src/mnode/src/mnodeMnode.c
浏览文件 @ c6479a14
...@@ -23,6 +23,8 @@ ...@@ -23,6 +23,8 @@
#include "tutil.h" #include "tutil.h"
#include "tsocket.h" #include "tsocket.h"
#include "tdataformat.h" #include "tdataformat.h"
#include "dnode.h"
#include "mnode.h"
#include "mnodeDef.h" #include "mnodeDef.h"
#include "mnodeInt.h" #include "mnodeInt.h"
#include "mnodeMnode.h" #include "mnodeMnode.h"
...@@ -30,6 +32,7 @@ ...@@ -30,6 +32,7 @@
#include "mnodeSdb.h" #include "mnodeSdb.h"
#include "mnodeShow.h" #include "mnodeShow.h"
#include "mnodeUser.h" #include "mnodeUser.h"
#include "mnodeVgroup.h"
static void * tsMnodeSdb = NULL; static void * tsMnodeSdb = NULL;
static int32_t tsMnodeUpdateSize = 0; static int32_t tsMnodeUpdateSize = 0;
...@@ -266,25 +269,87 @@ void mnodeGetMnodeInfos(void *mnodeInfos) { ...@@ -266,25 +269,87 @@ void mnodeGetMnodeInfos(void *mnodeInfos) {
mnodeMnodeUnLock(); mnodeMnodeUnLock();
} }
int32_t mnodeAddMnode(int32_t dnodeId) { static int32_t mnodeSendCreateMnodeMsg(int32_t dnodeId, char *dnodeEp) {
mDebug("dnode:%d, send create mnode msg to dnode %s", dnodeId, dnodeEp);
SMDCreateMnodeMsg *pCreate = rpcMallocCont(sizeof(SMDCreateMnodeMsg));
if (pCreate == NULL) {
return TSDB_CODE_MND_OUT_OF_MEMORY;
} else {
pCreate->dnodeId = htonl(dnodeId);
tstrncpy(pCreate->dnodeEp, dnodeEp, sizeof(pCreate->dnodeEp));
pCreate->mnodes = tsMnodeInfos;
bool found = false;
for (int i = 0; i < pCreate->mnodes.nodeNum; ++i) {
if (pCreate->mnodes.nodeInfos[i].nodeId == htonl(dnodeId)) {
found = true;
}
}
if (!found) {
pCreate->mnodes.nodeInfos[pCreate->mnodes.nodeNum].nodeId = htonl(dnodeId);
tstrncpy(pCreate->mnodes.nodeInfos[pCreate->mnodes.nodeNum].nodeEp, dnodeEp, sizeof(pCreate->dnodeEp));
pCreate->mnodes.nodeNum++;
}
}
SRpcMsg rpcMsg = {0};
rpcMsg.pCont = pCreate;
rpcMsg.contLen = sizeof(SMDCreateMnodeMsg);
rpcMsg.msgType = TSDB_MSG_TYPE_MD_CREATE_MNODE;
SRpcMsg rpcRsp = {0};
SRpcEpSet epSet = mnodeGetEpSetFromIp(pCreate->dnodeEp);
dnodeSendMsgToDnodeRecv(&rpcMsg, &rpcRsp, &epSet);
if (rpcRsp.code != TSDB_CODE_SUCCESS) {
mError("dnode:%d, failed to send create mnode msg, ep:%s reason:%s", dnodeId, dnodeEp, tstrerror(rpcRsp.code));
} else {
mDebug("dnode:%d, create mnode msg is disposed, mnode is created in dnode", dnodeId);
}
rpcFreeCont(rpcRsp.pCont);
return rpcRsp.code;
}
static int32_t mnodeCreateMnodeCb(SMnodeMsg *pMsg, int32_t code) {
if (code != TSDB_CODE_SUCCESS) {
mError("failed to create mnode, reason:%s", tstrerror(code));
} else {
mDebug("mnode is created successfully");
mnodeUpdateMnodeEpSet();
sdbUpdateAsync();
}
return code;
}
void mnodeCreateMnode(int32_t dnodeId, char *dnodeEp, bool needConfirm) {
SMnodeObj *pMnode = calloc(1, sizeof(SMnodeObj)); SMnodeObj *pMnode = calloc(1, sizeof(SMnodeObj));
pMnode->mnodeId = dnodeId; pMnode->mnodeId = dnodeId;
pMnode->createdTime = taosGetTimestampMs(); pMnode->createdTime = taosGetTimestampMs();
SSdbOper oper = { SSdbOper oper = {
.type = SDB_OPER_GLOBAL, .type = SDB_OPER_GLOBAL,
.table = tsMnodeSdb, .table = tsMnodeSdb,
.pObj = pMnode, .pObj = pMnode,
.writeCb = mnodeCreateMnodeCb
}; };
int32_t code = sdbInsertRow(&oper); int32_t code = TSDB_CODE_SUCCESS;
if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_MND_ACTION_IN_PROGRESS) { if (needConfirm) {
taosTFree(pMnode); code = mnodeSendCreateMnodeMsg(dnodeId, dnodeEp);
} }
mnodeUpdateMnodeEpSet(); if (code != TSDB_CODE_SUCCESS) {
taosTFree(pMnode);
return;
}
return code; code = sdbInsertRow(&oper);
if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_MND_ACTION_IN_PROGRESS) {
mError("dnode:%d, failed to create mnode, ep:%s reason:%s", dnodeId, dnodeEp, tstrerror(code));
taosTFree(pMnode);
}
} }
void mnodeDropMnodeLocal(int32_t dnodeId) { void mnodeDropMnodeLocal(int32_t dnodeId) {
...@@ -296,6 +361,7 @@ void mnodeDropMnodeLocal(int32_t dnodeId) { ...@@ -296,6 +361,7 @@ void mnodeDropMnodeLocal(int32_t dnodeId) {
} }
mnodeUpdateMnodeEpSet(); mnodeUpdateMnodeEpSet();
sdbUpdateAsync();
} }
int32_t mnodeDropMnode(int32_t dnodeId) { int32_t mnodeDropMnode(int32_t dnodeId) {
...@@ -315,6 +381,7 @@ int32_t mnodeDropMnode(int32_t dnodeId) { ...@@ -315,6 +381,7 @@ int32_t mnodeDropMnode(int32_t dnodeId) {
sdbDecRef(tsMnodeSdb, pMnode); sdbDecRef(tsMnodeSdb, pMnode);
mnodeUpdateMnodeEpSet(); mnodeUpdateMnodeEpSet();
sdbUpdateAsync();
return code; return code;
} }
...@@ -413,6 +480,7 @@ static int32_t mnodeRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, vo ...@@ -413,6 +480,7 @@ static int32_t mnodeRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, vo
mnodeDecMnodeRef(pMnode); mnodeDecMnodeRef(pMnode);
} }
mnodeVacuumResult(data, cols, numOfRows, rows, pShow);
pShow->numOfReads += numOfRows; pShow->numOfReads += numOfRows;
......
src/mnode/src/mnodePeer.c
浏览文件 @ c6479a14
...@@ -58,10 +58,15 @@ int32_t mnodeProcessPeerReq(SMnodeMsg *pMsg) { ...@@ -58,10 +58,15 @@ int32_t mnodeProcessPeerReq(SMnodeMsg *pMsg) {
rpcRsp->rsp = epSet; rpcRsp->rsp = epSet;
rpcRsp->len = sizeof(SRpcEpSet); rpcRsp->len = sizeof(SRpcEpSet);
mDebug("%p, msg:%s in mpeer queue, will be redireced, numOfEps:%d inUse:%d", pMsg->rpcMsg.ahandle, mDebug("%p, msg:%s in mpeer queue will be redirected, numOfEps:%d inUse:%d", pMsg->rpcMsg.ahandle,
taosMsg[pMsg->rpcMsg.msgType], epSet->numOfEps, epSet->inUse); taosMsg[pMsg->rpcMsg.msgType], epSet->numOfEps, epSet->inUse);
for (int32_t i = 0; i < epSet->numOfEps; ++i) { for (int32_t i = 0; i < epSet->numOfEps; ++i) {
mDebug("mnode index:%d ep:%s:%d", i, epSet->fqdn[i], htons(epSet->port[i])); if (strcmp(epSet->fqdn[i], tsLocalFqdn) == 0 && htons(epSet->port[i]) == tsServerPort + TSDB_PORT_DNODEDNODE) {
epSet->inUse = (i + 1) % epSet->numOfEps;
mDebug("mnode index:%d ep:%s:%u, set inUse to %d", i, epSet->fqdn[i], htons(epSet->port[i]), epSet->inUse);
} else {
mDebug("mnode index:%d ep:%s:%u", i, epSet->fqdn[i], htons(epSet->port[i]));
}
} }
return TSDB_CODE_RPC_REDIRECT; return TSDB_CODE_RPC_REDIRECT;
......
src/mnode/src/mnodeRead.c
浏览文件 @ c6479a14
...@@ -51,14 +51,21 @@ int32_t mnodeProcessRead(SMnodeMsg *pMsg) { ...@@ -51,14 +51,21 @@ int32_t mnodeProcessRead(SMnodeMsg *pMsg) {
SMnodeRsp *rpcRsp = &pMsg->rpcRsp; SMnodeRsp *rpcRsp = &pMsg->rpcRsp;
SRpcEpSet *epSet = rpcMallocCont(sizeof(SRpcEpSet)); SRpcEpSet *epSet = rpcMallocCont(sizeof(SRpcEpSet));
mnodeGetMnodeEpSetForShell(epSet); mnodeGetMnodeEpSetForShell(epSet);
rpcRsp->rsp = epSet;
rpcRsp->len = sizeof(SRpcEpSet);
mDebug("%p, msg:%s in mread queue, will be redireced, inUse:%d", pMsg->rpcMsg.ahandle, taosMsg[pMsg->rpcMsg.msgType], epSet->inUse); mDebug("%p, msg:%s in mread queue will be redirected, numOfEps:%d inUse:%d", pMsg->rpcMsg.ahandle,
taosMsg[pMsg->rpcMsg.msgType], epSet->numOfEps, epSet->inUse);
for (int32_t i = 0; i < epSet->numOfEps; ++i) { for (int32_t i = 0; i < epSet->numOfEps; ++i) {
mDebug("mnode index:%d ep:%s:%d", i, epSet->fqdn[i], htons(epSet->port[i])); if (strcmp(epSet->fqdn[i], tsLocalFqdn) == 0 && htons(epSet->port[i]) == tsServerPort) {
epSet->inUse = (i + 1) % epSet->numOfEps;
mDebug("mnode index:%d ep:%s:%u, set inUse to %d", i, epSet->fqdn[i], htons(epSet->port[i]), epSet->inUse);
} else {
mDebug("mnode index:%d ep:%s:%u", i, epSet->fqdn[i], htons(epSet->port[i]));
}
} }
rpcRsp->rsp = epSet;
rpcRsp->len = sizeof(SRpcEpSet);
return TSDB_CODE_RPC_REDIRECT; return TSDB_CODE_RPC_REDIRECT;
} }
......
src/mnode/src/mnodeSdb.c
浏览文件 @ c6479a14
...@@ -91,6 +91,7 @@ typedef struct { ...@@ -91,6 +91,7 @@ typedef struct {
} SSdbWriteWorkerPool; } SSdbWriteWorkerPool;
extern void * tsMnodeTmr; extern void * tsMnodeTmr;
static void * tsUpdateSyncTmr;
static SSdbObject tsSdbObj = {0}; static SSdbObject tsSdbObj = {0};
static taos_qset tsSdbWriteQset; static taos_qset tsSdbWriteQset;
static taos_qall tsSdbWriteQall; static taos_qall tsSdbWriteQall;
...@@ -297,27 +298,25 @@ static void sdbConfirmForward(void *ahandle, void *param, int32_t code) { ...@@ -297,27 +298,25 @@ static void sdbConfirmForward(void *ahandle, void *param, int32_t code) {
taosFreeQitem(pOper); taosFreeQitem(pOper);
} }
void sdbUpdateSync() { static void sdbUpdateSyncTmrFp(void *param, void *tmrId) { sdbUpdateSync(NULL); }
void sdbUpdateAsync() {
taosTmrReset(sdbUpdateSyncTmrFp, 200, NULL, tsMnodeTmr, &tsUpdateSyncTmr);
}
void sdbUpdateSync(void *pMnodes) {
SDMMnodeInfos *mnodes = pMnodes;
if (!mnodeIsRunning()) { if (!mnodeIsRunning()) {
mDebug("mnode not start yet, update sync info later"); mDebug("mnode not start yet, update sync config later");
return; return;
} }
mDebug("update sync info in sdb"); mDebug("update sync config in sync module, mnodes:%p", pMnodes);
SSyncCfg syncCfg = {0}; SSyncCfg syncCfg = {0};
int32_t index = 0; int32_t index = 0;
SDMMnodeInfos *mnodes = dnodeGetMnodeInfos(); if (mnodes == NULL) {
for (int32_t i = 0; i < mnodes->nodeNum; ++i) {
SDMMnodeInfo *node = &mnodes->nodeInfos[i];
syncCfg.nodeInfo[i].nodeId = node->nodeId;
taosGetFqdnPortFromEp(node->nodeEp, syncCfg.nodeInfo[i].nodeFqdn, &syncCfg.nodeInfo[i].nodePort);
syncCfg.nodeInfo[i].nodePort += TSDB_PORT_SYNC;
index++;
}
if (index == 0) {
void *pIter = NULL; void *pIter = NULL;
while (1) { while (1) {
SMnodeObj *pMnode = NULL; SMnodeObj *pMnode = NULL;
...@@ -337,9 +336,19 @@ void sdbUpdateSync() { ...@@ -337,9 +336,19 @@ void sdbUpdateSync() {
mnodeDecMnodeRef(pMnode); mnodeDecMnodeRef(pMnode);
} }
sdbFreeIter(pIter); sdbFreeIter(pIter);
syncCfg.replica = index;
mDebug("mnodes info not input, use infos in sdb, numOfMnodes:%d", syncCfg.replica);
} else {
for (index = 0; index < mnodes->nodeNum; ++index) {
SDMMnodeInfo *node = &mnodes->nodeInfos[index];
syncCfg.nodeInfo[index].nodeId = node->nodeId;
taosGetFqdnPortFromEp(node->nodeEp, syncCfg.nodeInfo[index].nodeFqdn, &syncCfg.nodeInfo[index].nodePort);
syncCfg.nodeInfo[index].nodePort += TSDB_PORT_SYNC;
}
syncCfg.replica = index;
mDebug("mnodes info input, numOfMnodes:%d", syncCfg.replica);
} }
syncCfg.replica = index;
syncCfg.quorum = (syncCfg.replica == 1) ? 1 : 2; syncCfg.quorum = (syncCfg.replica == 1) ? 1 : 2;
bool hasThisDnode = false; bool hasThisDnode = false;
...@@ -350,8 +359,15 @@ void sdbUpdateSync() { ...@@ -350,8 +359,15 @@ void sdbUpdateSync() {
} }
} }
if (!hasThisDnode) return; if (!hasThisDnode) {
if (memcmp(&syncCfg, &tsSdbObj.cfg, sizeof(SSyncCfg)) == 0) return; sdbDebug("update sync config, this dnode not exist");
return;
}
if (memcmp(&syncCfg, &tsSdbObj.cfg, sizeof(SSyncCfg)) == 0) {
sdbDebug("update sync config, info not changed");
return;
}
sdbInfo("work as mnode, replica:%d", syncCfg.replica); sdbInfo("work as mnode, replica:%d", syncCfg.replica);
for (int32_t i = 0; i < syncCfg.replica; ++i) { for (int32_t i = 0; i < syncCfg.replica; ++i) {
...@@ -1038,7 +1054,7 @@ static void *sdbWorkerFp(void *param) { ...@@ -1038,7 +1054,7 @@ static void *sdbWorkerFp(void *param) {
while (1) { while (1) {
numOfMsgs = taosReadAllQitemsFromQset(tsSdbWriteQset, tsSdbWriteQall, &unUsed); numOfMsgs = taosReadAllQitemsFromQset(tsSdbWriteQset, tsSdbWriteQall, &unUsed);
if (numOfMsgs == 0) { if (numOfMsgs == 0) {
sdbDebug("sdbWorkerFp: got no message from qset, exiting..."); sdbDebug("qset:%p, sdb got no message from qset, exiting", tsSdbWriteQset);
break; break;
} }
......
src/mnode/src/mnodeUser.c
浏览文件 @ c6479a14
...@@ -385,6 +385,7 @@ static int32_t mnodeRetrieveUsers(SShowObj *pShow, char *data, int32_t rows, voi ...@@ -385,6 +385,7 @@ static int32_t mnodeRetrieveUsers(SShowObj *pShow, char *data, int32_t rows, voi
numOfRows++; numOfRows++;
mnodeDecUserRef(pUser); mnodeDecUserRef(pUser);
} }
mnodeVacuumResult(data, cols, numOfRows, rows, pShow);
pShow->numOfReads += numOfRows; pShow->numOfReads += numOfRows;
return numOfRows; return numOfRows;
......
src/mnode/src/mnodeVgroup.c
浏览文件 @ c6479a14
...@@ -310,7 +310,7 @@ void mnodeUpdateVgroupStatus(SVgObj *pVgroup, SDnodeObj *pDnode, SVnodeLoad *pVl ...@@ -310,7 +310,7 @@ void mnodeUpdateVgroupStatus(SVgObj *pVgroup, SDnodeObj *pDnode, SVnodeLoad *pVl
for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) { for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
SVnodeGid *pVgid = &pVgroup->vnodeGid[i]; SVnodeGid *pVgid = &pVgroup->vnodeGid[i];
if (pVgid->pDnode == pDnode) { if (pVgid->pDnode == pDnode) {
mTrace("dnode:%d, receive status from dnode, vgId:%d status is %d", pDnode->dnodeId, pVgroup->vgId, pVgid->role); mTrace("dnode:%d, receive status from dnode, vgId:%d status is %d:%s", pDnode->dnodeId, pVgroup->vgId, pVgid->role, syncRole[pVgid->role]);
pVgid->role = pVload->role; pVgid->role = pVload->role;
if (pVload->role == TAOS_SYNC_ROLE_MASTER) { if (pVload->role == TAOS_SYNC_ROLE_MASTER) {
pVgroup->inUse = i; pVgroup->inUse = i;
...@@ -771,6 +771,7 @@ static int32_t mnodeRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, v ...@@ -771,6 +771,7 @@ static int32_t mnodeRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, v
mnodeDecVgroupRef(pVgroup); mnodeDecVgroupRef(pVgroup);
numOfRows++; numOfRows++;
} }
mnodeVacuumResult(data, cols, numOfRows, rows, pShow);
pShow->numOfReads += numOfRows; pShow->numOfReads += numOfRows;
mnodeDecTableRef(pTable); mnodeDecTableRef(pTable);
......
src/mnode/src/mnodeWrite.c
浏览文件 @ c6479a14
...@@ -54,11 +54,17 @@ int32_t mnodeProcessWrite(SMnodeMsg *pMsg) { ...@@ -54,11 +54,17 @@ int32_t mnodeProcessWrite(SMnodeMsg *pMsg) {
rpcRsp->rsp = epSet; rpcRsp->rsp = epSet;
rpcRsp->len = sizeof(SRpcEpSet); rpcRsp->len = sizeof(SRpcEpSet);
mDebug("app:%p:%p, msg:%s will be redireced inUse:%d", pMsg->rpcMsg.ahandle, pMsg, taosMsg[pMsg->rpcMsg.msgType], mDebug("app:%p:%p, msg:%s in write queue, will be redirected, numOfEps:%d inUse:%d", pMsg->rpcMsg.ahandle, pMsg,
epSet->inUse); taosMsg[pMsg->rpcMsg.msgType], epSet->numOfEps, epSet->inUse);
for (int32_t i = 0; i < epSet->numOfEps; ++i) { for (int32_t i = 0; i < epSet->numOfEps; ++i) {
mDebug("app:%p:%p, mnode index:%d ep:%s:%d", pMsg->rpcMsg.ahandle, pMsg, i, epSet->fqdn[i], if (strcmp(epSet->fqdn[i], tsLocalFqdn) == 0 && htons(epSet->port[i]) == tsServerPort) {
htons(epSet->port[i])); epSet->inUse = (i + 1) % epSet->numOfEps;
mDebug("app:%p:%p, mnode index:%d ep:%s:%d, set inUse to %d", pMsg->rpcMsg.ahandle, pMsg, i, epSet->fqdn[i],
htons(epSet->port[i]), epSet->inUse);
} else {
mDebug("app:%p:%p, mnode index:%d ep:%s:%d", pMsg->rpcMsg.ahandle, pMsg, i, epSet->fqdn[i],
htons(epSet->port[i]));
}
} }
return TSDB_CODE_RPC_REDIRECT; return TSDB_CODE_RPC_REDIRECT;
......
src/plugins/http/src/httpQueue.c
浏览文件 @ c6479a14
...@@ -67,7 +67,7 @@ static void *httpProcessResultQueue(void *param) { ...@@ -67,7 +67,7 @@ static void *httpProcessResultQueue(void *param) {
while (1) { while (1) {
if (taosReadQitemFromQset(tsHttpQset, &type, (void **)&pMsg, &unUsed) == 0) { if (taosReadQitemFromQset(tsHttpQset, &type, (void **)&pMsg, &unUsed) == 0) {
httpDebug("httpResultQueue: got no message from qset, exiting..."); httpDebug("qset:%p, http queue got no message from qset, exiting", tsHttpQset);
break; break;
} }
......
src/query/src/qExecutor.c
浏览文件 @ c6479a14
...@@ -4511,7 +4511,6 @@ int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, void *tsdb, int32_t vgId, bo ...@@ -4511,7 +4511,6 @@ int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, void *tsdb, int32_t vgId, bo
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
SQuery *pQuery = pQInfo->runtimeEnv.pQuery; SQuery *pQuery = pQInfo->runtimeEnv.pQuery;
pQuery->precision = tsdbGetCfg(tsdb)->precision;
pRuntimeEnv->topBotQuery = isTopBottomQuery(pQuery); pRuntimeEnv->topBotQuery = isTopBottomQuery(pQuery);
pRuntimeEnv->hasTagResults = hasTagValOutput(pQuery); pRuntimeEnv->hasTagResults = hasTagValOutput(pQuery);
...@@ -6323,6 +6322,8 @@ static int32_t initQInfo(SQueryTableMsg *pQueryMsg, void *tsdb, int32_t vgId, SQ ...@@ -6323,6 +6322,8 @@ static int32_t initQInfo(SQueryTableMsg *pQueryMsg, void *tsdb, int32_t vgId, SQ
bool ret = tsBufNextPos(pTSBuf); bool ret = tsBufNextPos(pTSBuf);
UNUSED(ret); UNUSED(ret);
} }
pQuery->precision = tsdbGetCfg(tsdb)->precision;
if ((QUERY_IS_ASC_QUERY(pQuery) && (pQuery->window.skey > pQuery->window.ekey)) || if ((QUERY_IS_ASC_QUERY(pQuery) && (pQuery->window.skey > pQuery->window.ekey)) ||
(!QUERY_IS_ASC_QUERY(pQuery) && (pQuery->window.ekey > pQuery->window.skey))) { (!QUERY_IS_ASC_QUERY(pQuery) && (pQuery->window.ekey > pQuery->window.skey))) {
......
src/rpc/src/rpcMain.c
浏览文件 @ c6479a14
...@@ -542,10 +542,7 @@ void rpcCancelRequest(void *handle) { ...@@ -542,10 +542,7 @@ void rpcCancelRequest(void *handle) {
if (pContext->pConn) { if (pContext->pConn) {
tDebug("%s, app tries to cancel request", pContext->pConn->info); tDebug("%s, app tries to cancel request", pContext->pConn->info);
pContext->pConn->pReqMsg = NULL;
rpcCloseConn(pContext->pConn); rpcCloseConn(pContext->pConn);
pContext->pConn = NULL;
rpcFreeCont(pContext->pCont);
} }
} }
...@@ -613,8 +610,10 @@ static void rpcReleaseConn(SRpcConn *pConn) { ...@@ -613,8 +610,10 @@ static void rpcReleaseConn(SRpcConn *pConn) {
if (pConn->pReqMsg) rpcFreeCont(pConn->pReqMsg); // do not use rpcFreeMsg if (pConn->pReqMsg) rpcFreeCont(pConn->pReqMsg); // do not use rpcFreeMsg
} else { } else {
// if there is an outgoing message, free it // if there is an outgoing message, free it
if (pConn->outType && pConn->pReqMsg) if (pConn->outType && pConn->pReqMsg) {
if (pConn->pContext) pConn->pContext->pConn = NULL;
rpcFreeMsg(pConn->pReqMsg); rpcFreeMsg(pConn->pReqMsg);
}
} }
// memset could not be used, since lockeBy can not be reset // memset could not be used, since lockeBy can not be reset
...@@ -1121,9 +1120,13 @@ static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead, SRpcReqConte ...@@ -1121,9 +1120,13 @@ static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead, SRpcReqConte
SRpcEpSet *pEpSet = (SRpcEpSet*)pHead->content; SRpcEpSet *pEpSet = (SRpcEpSet*)pHead->content;
if (pEpSet->numOfEps > 0) { if (pEpSet->numOfEps > 0) {
memcpy(&pContext->epSet, pHead->content, sizeof(pContext->epSet)); memcpy(&pContext->epSet, pHead->content, sizeof(pContext->epSet));
tDebug("%s, redirect is received, numOfEps:%d", pConn->info, pContext->epSet.numOfEps); tDebug("%s, redirect is received, numOfEps:%d inUse:%d", pConn->info, pContext->epSet.numOfEps,
for (int i=0; i<pContext->epSet.numOfEps; ++i) pContext->epSet.inUse);
for (int i = 0; i < pContext->epSet.numOfEps; ++i) {
pContext->epSet.port[i] = htons(pContext->epSet.port[i]); pContext->epSet.port[i] = htons(pContext->epSet.port[i]);
tDebug("%s, redirect is received, index:%d ep:%s:%u", pConn->info, i, pContext->epSet.fqdn[i],
pContext->epSet.port[i]);
}
} }
rpcSendReqToServer(pRpc, pContext); rpcSendReqToServer(pRpc, pContext);
rpcFreeCont(rpcMsg.pCont); rpcFreeCont(rpcMsg.pCont);
......
src/rpc/src/rpcTcp.c
浏览文件 @ c6479a14
...@@ -525,7 +525,7 @@ static void *taosProcessTcpData(void *param) { ...@@ -525,7 +525,7 @@ static void *taosProcessTcpData(void *param) {
while (pThreadObj->pHead) { while (pThreadObj->pHead) {
SFdObj *pFdObj = pThreadObj->pHead; SFdObj *pFdObj = pThreadObj->pHead;
pThreadObj->pHead = pFdObj->next; pThreadObj->pHead = pFdObj->next;
taosFreeFdObj(pFdObj); taosReportBrokenLink(pFdObj);
} }
pthread_mutex_destroy(&(pThreadObj->mutex)); pthread_mutex_destroy(&(pThreadObj->mutex));
......
src/sync/src/syncMain.c
浏览文件 @ c6479a14
...@@ -215,6 +215,9 @@ void syncStop(void *param) { ...@@ -215,6 +215,9 @@ void syncStop(void *param) {
pthread_mutex_lock(&(pNode->mutex)); pthread_mutex_lock(&(pNode->mutex));
if (vgIdHash) taosHashRemove(vgIdHash, (const char *)&pNode->vgId, sizeof(int32_t));
if (pNode->pFwdTimer) taosTmrStop(pNode->pFwdTimer);
for (int i = 0; i < pNode->replica; ++i) { for (int i = 0; i < pNode->replica; ++i) {
pPeer = pNode->peerInfo[i]; pPeer = pNode->peerInfo[i];
if (pPeer) syncRemovePeer(pPeer); if (pPeer) syncRemovePeer(pPeer);
...@@ -223,9 +226,6 @@ void syncStop(void *param) { ...@@ -223,9 +226,6 @@ void syncStop(void *param) {
pPeer = pNode->peerInfo[TAOS_SYNC_MAX_REPLICA]; pPeer = pNode->peerInfo[TAOS_SYNC_MAX_REPLICA];
if (pPeer) syncRemovePeer(pPeer); if (pPeer) syncRemovePeer(pPeer);
if (vgIdHash) taosHashRemove(vgIdHash, (const char *)&pNode->vgId, sizeof(int32_t));
if (pNode->pFwdTimer) taosTmrStop(pNode->pFwdTimer);
pthread_mutex_unlock(&(pNode->mutex)); pthread_mutex_unlock(&(pNode->mutex));
syncDecNodeRef(pNode); syncDecNodeRef(pNode);
...@@ -313,6 +313,8 @@ int32_t syncForwardToPeer(void *param, void *data, void *mhandle, int qtype) { ...@@ -313,6 +313,8 @@ int32_t syncForwardToPeer(void *param, void *data, void *mhandle, int qtype) {
// always update version // always update version
nodeVersion = pWalHead->version; nodeVersion = pWalHead->version;
sDebug("replica:%d nodeRole:%d qtype:%d", pNode->replica, nodeRole, qtype);
if (pNode->replica == 1 || nodeRole != TAOS_SYNC_ROLE_MASTER) return 0; if (pNode->replica == 1 || nodeRole != TAOS_SYNC_ROLE_MASTER) return 0;
// only pkt from RPC or CQ can be forwarded // only pkt from RPC or CQ can be forwarded
...@@ -1189,6 +1191,8 @@ static void syncProcessFwdAck(SSyncNode *pNode, SFwdInfo *pFwdInfo, int32_t code ...@@ -1189,6 +1191,8 @@ static void syncProcessFwdAck(SSyncNode *pNode, SFwdInfo *pFwdInfo, int32_t code
static void syncMonitorFwdInfos(void *param, void *tmrId) { static void syncMonitorFwdInfos(void *param, void *tmrId) {
SSyncNode *pNode = param; SSyncNode *pNode = param;
SSyncFwds *pSyncFwds = pNode->pSyncFwds; SSyncFwds *pSyncFwds = pNode->pSyncFwds;
if (pSyncFwds == NULL) return;
uint64_t time = taosGetTimestampMs(); uint64_t time = taosGetTimestampMs();
if (pSyncFwds->fwds > 0) { if (pSyncFwds->fwds > 0) {
......
src/tsdb/src/tsdbRead.c
浏览文件 @ c6479a14
...@@ -2639,8 +2639,7 @@ int32_t tsdbGetTableGroupFromIdList(TSDB_REPO_T* tsdb, SArray* pTableIdList, STa ...@@ -2639,8 +2639,7 @@ int32_t tsdbGetTableGroupFromIdList(TSDB_REPO_T* tsdb, SArray* pTableIdList, STa
pGroupInfo->pGroupList = taosArrayInit(1, POINTER_BYTES); pGroupInfo->pGroupList = taosArrayInit(1, POINTER_BYTES);
SArray* group = taosArrayInit(1, sizeof(STableKeyInfo)); SArray* group = taosArrayInit(1, sizeof(STableKeyInfo));
int32_t i = 0; for(int32_t i = 0; i < size; ++i) {
for(; i < size; ++i) {
STableIdInfo *id = taosArrayGet(pTableIdList, i); STableIdInfo *id = taosArrayGet(pTableIdList, i);
STable* pTable = tsdbGetTableByUid(tsdbGetMeta(tsdb), id->uid); STable* pTable = tsdbGetTableByUid(tsdbGetMeta(tsdb), id->uid);
...@@ -2665,8 +2664,12 @@ int32_t tsdbGetTableGroupFromIdList(TSDB_REPO_T* tsdb, SArray* pTableIdList, STa ...@@ -2665,8 +2664,12 @@ int32_t tsdbGetTableGroupFromIdList(TSDB_REPO_T* tsdb, SArray* pTableIdList, STa
return terrno; return terrno;
} }
pGroupInfo->numOfTables = i; pGroupInfo->numOfTables = taosArrayGetSize(group);
taosArrayPush(pGroupInfo->pGroupList, &group); if (pGroupInfo->numOfTables > 0) {
taosArrayPush(pGroupInfo->pGroupList, &group);
} else {
taosArrayDestroy(group);
}
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
......
src/util/src/tqueue.c
浏览文件 @ c6479a14
...@@ -263,6 +263,7 @@ void taosCloseQset(taos_qset param) { ...@@ -263,6 +263,7 @@ void taosCloseQset(taos_qset param) {
// thread to exit. // thread to exit.
void taosQsetThreadResume(taos_qset param) { void taosQsetThreadResume(taos_qset param) {
STaosQset *qset = (STaosQset *)param; STaosQset *qset = (STaosQset *)param;
uDebug("qset:%p, it will exit", qset);
tsem_post(&qset->sem); tsem_post(&qset->sem);
} }
......
src/wal/src/walMain.c
浏览文件 @ c6479a14
...@@ -427,7 +427,7 @@ static int walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp) { ...@@ -427,7 +427,7 @@ static int walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp) {
if (!taosCheckChecksumWhole((uint8_t *)pHead, sizeof(SWalHead))) { if (!taosCheckChecksumWhole((uint8_t *)pHead, sizeof(SWalHead))) {
wWarn("wal:%s, cksum is messed up, skip the rest of file", name); wWarn("wal:%s, cksum is messed up, skip the rest of file", name);
terrno = TSDB_CODE_WAL_FILE_CORRUPTED; terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
ASSERT(false); // ASSERT(false);
break; break;
} }
......
tests/examples/JDBC/calciteDemo/pom.xml
浏览文件 @ c6479a14
...@@ -44,7 +44,7 @@ ...@@ -44,7 +44,7 @@
<dependency> <dependency>
<groupId>com.taosdata.jdbc</groupId> <groupId>com.taosdata.jdbc</groupId>
<artifactId>taos-jdbcdriver</artifactId> <artifactId>taos-jdbcdriver</artifactId>
<version>2.0.7</version> <version>2.0.8</version>
</dependency> </dependency>
</dependencies> </dependencies>
......
tests/pytest/crash_gen.sh
浏览文件 @ c6479a14
...@@ -42,11 +42,37 @@ TAOSD_DIR=`find $TAOS_DIR -name "taosd"|grep bin|head -n1` ...@@ -42,11 +42,37 @@ 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 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
PYTHON_EXEC=python3.8
# First we need to set up a path for Python to find our own TAOS modules, so that "import" can work. # First we need to set up a path for Python to find our own TAOS modules, so that "import" can work.
export PYTHONPATH=$(pwd)/../../src/connector/python/linux/python3 export PYTHONPATH=$(pwd)/../../src/connector/python/linux/python3:$(pwd)
# Then let us set up the library path so that our compiled SO file can be loaded by Python # Then let us set up the library path so that our compiled SO file can be loaded by Python
export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:$LIB_DIR export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:$LIB_DIR
# Now we are all let, and let's see if we can find a crash. Note we pass all params # Now we are all let, and let's see if we can find a crash. Note we pass all params
python3.8 ./crash_gen.py $@ if [[ $1 == '--valgrind' ]]; then
shift
export PYTHONMALLOC=malloc
VALGRIND_OUT=valgrind.out
VALGRIND_ERR=valgrind.err
# How to generate valgrind suppression file: https://stackoverflow.com/questions/17159578/generating-suppressions-for-memory-leaks
# valgrind --leak-check=full --gen-suppressions=all --log-fd=9 python3.8 ./crash_gen.py $@ 9>>memcheck.log
echo Executing under VALGRIND, with STDOUT/ERR going to $VALGRIND_OUT and $VALGRIND_ERR, please watch them from a different terminal.
valgrind \
--leak-check=yes \
--suppressions=crash_gen/valgrind_taos.supp \
$PYTHON_EXEC \
./crash_gen/crash_gen.py $@ > $VALGRIND_OUT 2> $VALGRIND_ERR
elif [[ $1 == '--helgrind' ]]; then
shift
valgrind \
--tool=helgrind \
$PYTHON_EXEC \
./crash_gen/crash_gen.py $@
else
$PYTHON_EXEC ./crash_gen/crash_gen.py $@
fi
tests/pytest/crash_gen/valgrind_taos.supp 0 → 100644
浏览文件 @ c6479a14
此差异已折叠。
tests/script/sh/deploy.sh
浏览文件 @ c6479a14
...@@ -111,24 +111,24 @@ echo "serverPort ${NODE}" >> $TAOS_CFG ...@@ -111,24 +111,24 @@ echo "serverPort ${NODE}" >> $TAOS_CFG
echo "dataDir $DATA_DIR" >> $TAOS_CFG echo "dataDir $DATA_DIR" >> $TAOS_CFG
echo "logDir $LOG_DIR" >> $TAOS_CFG echo "logDir $LOG_DIR" >> $TAOS_CFG
echo "debugFlag 0" >> $TAOS_CFG echo "debugFlag 0" >> $TAOS_CFG
echo "mDebugFlag 135" >> $TAOS_CFG echo "mDebugFlag 143" >> $TAOS_CFG
echo "sdbDebugFlag 135" >> $TAOS_CFG echo "sdbDebugFlag 143" >> $TAOS_CFG
echo "dDebugFlag 135" >> $TAOS_CFG echo "dDebugFlag 143" >> $TAOS_CFG
echo "vDebugFlag 135" >> $TAOS_CFG echo "vDebugFlag 143" >> $TAOS_CFG
echo "tsdbDebugFlag 135" >> $TAOS_CFG echo "tsdbDebugFlag 143" >> $TAOS_CFG
echo "cDebugFlag 135" >> $TAOS_CFG echo "cDebugFlag 143" >> $TAOS_CFG
echo "jnidebugFlag 135" >> $TAOS_CFG echo "jnidebugFlag 143" >> $TAOS_CFG
echo "odbcdebugFlag 135" >> $TAOS_CFG echo "odbcdebugFlag 143" >> $TAOS_CFG
echo "httpDebugFlag 135" >> $TAOS_CFG echo "httpDebugFlag 143" >> $TAOS_CFG
echo "monitorDebugFlag 135" >> $TAOS_CFG echo "monitorDebugFlag 143" >> $TAOS_CFG
echo "mqttDebugFlag 135" >> $TAOS_CFG echo "mqttDebugFlag 143" >> $TAOS_CFG
echo "qdebugFlag 135" >> $TAOS_CFG echo "qdebugFlag 143" >> $TAOS_CFG
echo "rpcDebugFlag 135" >> $TAOS_CFG echo "rpcDebugFlag 143" >> $TAOS_CFG
echo "tmrDebugFlag 131" >> $TAOS_CFG echo "tmrDebugFlag 131" >> $TAOS_CFG
echo "udebugFlag 135" >> $TAOS_CFG echo "udebugFlag 143" >> $TAOS_CFG
echo "sdebugFlag 135" >> $TAOS_CFG echo "sdebugFlag 143" >> $TAOS_CFG
echo "wdebugFlag 135" >> $TAOS_CFG echo "wdebugFlag 143" >> $TAOS_CFG
echo "cqdebugFlag 135" >> $TAOS_CFG echo "cqdebugFlag 143" >> $TAOS_CFG
echo "monitor 0" >> $TAOS_CFG echo "monitor 0" >> $TAOS_CFG
echo "monitorInterval 1" >> $TAOS_CFG echo "monitorInterval 1" >> $TAOS_CFG
echo "http 0" >> $TAOS_CFG echo "http 0" >> $TAOS_CFG
......
tests/script/test.sh
浏览文件 @ c6479a14
...@@ -109,15 +109,10 @@ echo "dataDir $DATA_DIR" >> $TAOS_CFG ...@@ -109,15 +109,10 @@ echo "dataDir $DATA_DIR" >> $TAOS_CFG
echo "logDir $LOG_DIR" >> $TAOS_CFG echo "logDir $LOG_DIR" >> $TAOS_CFG
echo "scriptDir ${CODE_DIR}/../script" >> $TAOS_CFG echo "scriptDir ${CODE_DIR}/../script" >> $TAOS_CFG
echo "numOfLogLines 100000000" >> $TAOS_CFG echo "numOfLogLines 100000000" >> $TAOS_CFG
echo "dDebugFlag 135" >> $TAOS_CFG echo "rpcDebugFlag 143" >> $TAOS_CFG
echo "mDebugFlag 135" >> $TAOS_CFG
echo "sdbDebugFlag 135" >> $TAOS_CFG
echo "rpcDebugFlag 135" >> $TAOS_CFG
echo "tmrDebugFlag 131" >> $TAOS_CFG echo "tmrDebugFlag 131" >> $TAOS_CFG
echo "cDebugFlag 135" >> $TAOS_CFG echo "cDebugFlag 143" >> $TAOS_CFG
echo "httpDebugFlag 135" >> $TAOS_CFG echo "udebugFlag 143" >> $TAOS_CFG
echo "monitorDebugFlag 135" >> $TAOS_CFG
echo "udebugFlag 135" >> $TAOS_CFG
echo "tablemetakeeptimer 5" >> $TAOS_CFG echo "tablemetakeeptimer 5" >> $TAOS_CFG
echo "wal 0" >> $TAOS_CFG echo "wal 0" >> $TAOS_CFG
echo "asyncLog 0" >> $TAOS_CFG echo "asyncLog 0" >> $TAOS_CFG
......
tests/script/unique/cluster/vgroup100.sim
浏览文件 @ c6479a14
...@@ -27,7 +27,16 @@ system sh/exec.sh -n dnode2 -s start ...@@ -27,7 +27,16 @@ system sh/exec.sh -n dnode2 -s start
sql create dnode $hostname3 sql create dnode $hostname3
system sh/exec.sh -n dnode3 -s start system sh/exec.sh -n dnode3 -s start
sleep 5000 sleep 3000
$x = 0
show2:
$x = $x + 1
sleep 2000
if $x == 10 then
return -1
endi
sql show mnodes sql show mnodes
$dnode1Role = $data2_1 $dnode1Role = $data2_1
$dnode2Role = $data2_2 $dnode2Role = $data2_2
...@@ -37,6 +46,16 @@ print $dnode1Role ...@@ -37,6 +46,16 @@ print $dnode1Role
print $dnode2Role print $dnode2Role
print $dnode3Role print $dnode3Role
if $dnode1Role != master then
goto show2
endi
if $dnode2Role != slave then
goto show2
endi
if $dnode3Role != slave then
goto show2
endi
print ============================== step3 print ============================== step3
$count = 2 $count = 2
while $count < 102 while $count < 102
......
tests/script/unique/mnode/mgmt21.sim
浏览文件 @ c6479a14
...@@ -26,11 +26,11 @@ $x = 0 ...@@ -26,11 +26,11 @@ $x = 0
show2: show2:
$x = $x + 1 $x = $x + 1
sleep 2000 sleep 2000
if $x == 10 then if $x == 5 then
return -1 return -1
endi endi
sql show mnodes sql show mnodes -x show2
print dnode1 ==> $data2_1 print dnode1 ==> $data2_1
print dnode2 ==> $data2_2 print dnode2 ==> $data2_2
if $data2_1 != master then if $data2_1 != master then
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或