提交 dac59f16 编写于 作者: haoranc's avatar haoranc

Merge branch 'main' of https://github.com/taosdata/TDengine into test/main/TD-22157

......@@ -2,7 +2,7 @@
# taos-tools
ExternalProject_Add(taos-tools
GIT_REPOSITORY https://github.com/taosdata/taos-tools.git
GIT_TAG 01195d6
GIT_TAG 63635fc
SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools"
BINARY_DIR ""
#BUILD_IN_SOURCE TRUE
......
......@@ -31,21 +31,49 @@
extern "C" {
#endif
#if defined(CUS_NAME) || defined(CUS_PROMPT) || defined(CUS_EMAIL)
#include "cus_name.h"
#endif
#ifdef WINDOWS
#define TD_TMP_DIR_PATH "C:\\Windows\\Temp\\"
#ifdef CUS_NAME
#define TD_CFG_DIR_PATH "C:\\"CUS_NAME"\\cfg\\"
#define TD_DATA_DIR_PATH "C:\\"CUS_NAME"\\data\\"
#define TD_LOG_DIR_PATH "C:\\"CUS_NAME"\\log\\"
#else
#define TD_CFG_DIR_PATH "C:\\TDengine\\cfg\\"
#define TD_DATA_DIR_PATH "C:\\TDengine\\data\\"
#define TD_LOG_DIR_PATH "C:\\TDengine\\log\\"
#endif // CUS_NAME
#elif defined(_TD_DARWIN_64)
#ifdef CUS_PROMPT
#define TD_TMP_DIR_PATH "/tmp/"CUS_PROMPT"d/"
#define TD_CFG_DIR_PATH "/etc/"CUS_PROMPT"/"
#define TD_DATA_DIR_PATH "/var/lib/"CUS_PROMPT"/"
#define TD_LOG_DIR_PATH "/var/log/"CUS_PROMPT"/"
#else
#define TD_TMP_DIR_PATH "/tmp/taosd/"
#define TD_CFG_DIR_PATH "/etc/taos/"
#define TD_DATA_DIR_PATH "/var/lib/taos/"
#define TD_LOG_DIR_PATH "/var/log/taos/"
#endif // CUS_PROMPT
#else
#define TD_TMP_DIR_PATH "/tmp/"
#ifdef CUS_PROMPT
#define TD_CFG_DIR_PATH "/etc/"CUS_PROMPT"/"
#define TD_DATA_DIR_PATH "/var/lib/"CUS_PROMPT"/"
#define TD_LOG_DIR_PATH "/var/log/"CUS_PROMPT"/"
#else
#define TD_CFG_DIR_PATH "/etc/taos/"
#define TD_DATA_DIR_PATH "/var/lib/taos/"
#define TD_LOG_DIR_PATH "/var/log/taos/"
#endif // CUS_PROMPT
#endif
typedef struct TdDir *TdDirPtr;
......
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _CUS_NAME_H_
#define _CUS_NAME_H_
#ifndef CUS_NAME
#define CUS_NAME "TDengine"
#endif
#ifndef CUS_PROMPT
#define CUS_PROMPT "taos"
#endif
#ifndef CUS_EMAIL
#define CUS_EMAIL "<support@taosdata.com>"
#endif
#endif // _CUS_NAME_H_
########################################################
# #
# Configuration #
# Any questions, please email support@taosdata.com #
# #
########################################################
......@@ -13,7 +12,7 @@
############### 1. Cluster End point ############################
# The end point of the first dnode in the cluster to be connected to when this dnode or a CLI `taos` is started
# The end point of the first dnode in the cluster to be connected to when this dnode or the CLI utility is started
# firstEp hostname:6030
# The end point of the second dnode to be connected to if the firstEp is not available
......@@ -25,7 +24,7 @@
# The FQDN of the host on which this dnode will be started. It can be IP address
# fqdn hostname
# The port for external access after this dnode is started
# The port for external access after this dnode is started
# serverPort 6030
# The maximum number of connections a dnode can accept
......@@ -96,7 +95,7 @@
# if free disk space is less than this value, this dnode will fail to start
# minimalDataDirGB 2.0
# enable/disable system monitor
# enable/disable system monitor
# monitor 1
# The following parameter is used to limit the maximum number of lines in log files.
......@@ -114,8 +113,8 @@
# The following parameters are used for debug purpose only by this dnode.
# debugFlag is a 8 bits mask: FILE-SCREEN-UNUSED-HeartBeat-DUMP-TRACE_WARN-ERROR
# Available debug levels are:
# 131: output warning and error
# Available debug levels are:
# 131: output warning and error
# 135: output debug, warning and error
# 143: output trace, debug, warning and error to log
# 199: output debug, warning and error to both screen and file
......@@ -130,7 +129,7 @@
# debug flag for util
# uDebugFlag 131
# debug flag for rpc
# debug flag for rpc
# rpcDebugFlag 131
# debug flag for jni
......@@ -139,7 +138,7 @@
# debug flag for query
# qDebugFlag 131
# debug flag for taosc driver
# debug flag for client driver
# cDebugFlag 131
# debug flag for dnode messages
......
[Unit]
Description=TDengine server service
Description=server service
After=network-online.target
Wants=network-online.target
......
此差异已折叠。
......@@ -23,9 +23,12 @@ clientName2="${12}"
productName="TDengine"
clientName="taos"
benchmarkName="taosBenchmark"
configFile="taos.cfg"
tarName="package.tar.gz"
benchmarkName2="${clientName2}Benchmark"
if [ "$osType" != "Darwin" ]; then
script_dir="$(dirname $(readlink -f $0))"
top_dir="$(readlink -f ${script_dir}/../..)"
......@@ -53,11 +56,12 @@ fi
# Directories and files.
if [ "$verMode" == "cluster" ]; then
sed -i 's/verMode=edge/verMode=cluster/g' ${script_dir}/remove_client.sh
sed -i "s/clientName2=\"taos\"/clientName2=\"${clientName2}\"/g" ${script_dir}/remove_client.sh
sed -i "s/productName2=\"TDengine\"/productName2=\"${productName2}\"/g" ${script_dir}/remove_client.sh
fi
#if [ "$verMode" == "cluster" ]; then
# sed -i 's/verMode=edge/verMode=cluster/g' ${script_dir}/remove_client.sh
# sed -i "s/clientName2=\"taos\"/clientName2=\"${clientName2}\"/g" ${script_dir}/remove_client.sh
# sed -i "s/configFile2=\"taos\"/configFile2=\"${clientName2}\"/g" ${script_dir}/remove_client.sh
# sed -i "s/productName2=\"TDengine\"/productName2=\"${productName2}\"/g" ${script_dir}/remove_client.sh
#fi
if [ "$osType" != "Darwin" ]; then
if [ "$pagMode" == "lite" ]; then
......@@ -66,6 +70,7 @@ if [ "$osType" != "Darwin" ]; then
${script_dir}/remove_client.sh"
else
bin_files="${build_dir}/bin/${clientName} \
${build_dir}/bin/${benchmarkName} \
${script_dir}/remove_client.sh \
${script_dir}/set_core.sh \
${script_dir}/get_client.sh"
......@@ -153,6 +158,7 @@ if [ "$verMode" == "cluster" ]; then
sed -i 's/verMode=edge/verMode=cluster/g' install_client_temp.sh
sed -i "s/serverName2=\"taosd\"/serverName2=\"${serverName2}\"/g" install_client_temp.sh
sed -i "s/clientName2=\"taos\"/clientName2=\"${clientName2}\"/g" install_client_temp.sh
sed -i "s/configFile2=\"taos.cfg\"/configFile2=\"${clientName2}.cfg\"/g" install_client_temp.sh
sed -i "s/productName2=\"TDengine\"/productName2=\"${productName2}\"/g" install_client_temp.sh
sed -i "s/emailName2=\"taosdata.com\"/emailName2=\"${cusEmail2}\"/g" install_client_temp.sh
......
......@@ -51,9 +51,9 @@ fi
if [ -d ${top_dir}/tools/taos-tools/packaging/deb ]; then
cd ${top_dir}/tools/taos-tools/packaging/deb
taostools_ver=$(git for-each-ref --sort=taggerdate --format '%(tag)' refs/tags|grep -v taos | tail -1)
[ -z "$taos_tools_ver" ] && taos_tools_ver="0.1.0"
[ -z "$taostools_ver" ] && taostools_ver="0.1.0"
taostools_install_dir="${release_dir}/${clientName2}Tools-${taostools_ver}"
cd ${curr_dir}
......@@ -96,7 +96,7 @@ else
${taostools_bin_files} \
${taosx_bin} \
${explorer_bin_files} \
${build_dir}/bin/taosadapter \
${build_dir}/bin/${clientName}adapter \
${build_dir}/bin/udfd \
${script_dir}/remove.sh \
${script_dir}/set_core.sh \
......@@ -135,12 +135,12 @@ mkdir -p ${install_dir}/inc && cp ${header_files} ${install_dir}/inc
mkdir -p ${install_dir}/cfg && cp ${cfg_dir}/${configFile} ${install_dir}/cfg/${configFile}
if [ -f "${compile_dir}/test/cfg/taosadapter.toml" ]; then
cp ${compile_dir}/test/cfg/taosadapter.toml ${install_dir}/cfg || :
if [ -f "${compile_dir}/test/cfg/${clientName}adapter.toml" ]; then
cp ${compile_dir}/test/cfg/${clientName}adapter.toml ${install_dir}/cfg || :
fi
if [ -f "${compile_dir}/test/cfg/taosadapter.service" ]; then
cp ${compile_dir}/test/cfg/taosadapter.service ${install_dir}/cfg || :
if [ -f "${compile_dir}/test/cfg/${clientName}adapter.service" ]; then
cp ${compile_dir}/test/cfg/${clientName}adapter.service ${install_dir}/cfg || :
fi
if [ -f "${cfg_dir}/${serverName}.service" ]; then
......@@ -152,16 +152,16 @@ mkdir -p ${install_dir}/init.d && cp ${init_file_deb} ${install_dir}/init.d/${se
mkdir -p ${install_dir}/init.d && cp ${init_file_rpm} ${install_dir}/init.d/${serverName}.rpm
if [ $adapterName != "taosadapter" ]; then
mv ${install_dir}/cfg/taosadapter.toml ${install_dir}/cfg/$adapterName.toml
mv ${install_dir}/cfg/${clientName2}adapter.toml ${install_dir}/cfg/$adapterName.toml
sed -i "s/path = \"\/var\/log\/taos\"/path = \"\/var\/log\/${productName}\"/g" ${install_dir}/cfg/$adapterName.toml
sed -i "s/password = \"taosdata\"/password = \"${defaultPasswd}\"/g" ${install_dir}/cfg/$adapterName.toml
mv ${install_dir}/cfg/taosadapter.service ${install_dir}/cfg/$adapterName.service
mv ${install_dir}/cfg/${clientName2}adapter.service ${install_dir}/cfg/$adapterName.service
sed -i "s/TDengine/${productName}/g" ${install_dir}/cfg/$adapterName.service
sed -i "s/taosAdapter/${adapterName}/g" ${install_dir}/cfg/$adapterName.service
sed -i "s/taosadapter/${adapterName}/g" ${install_dir}/cfg/$adapterName.service
mv ${install_dir}/bin/taosadapter ${install_dir}/bin/${adapterName}
mv ${install_dir}/bin/${clientName2}adapter ${install_dir}/bin/${adapterName}
mv ${install_dir}/bin/taosd-dump-cfg.gdb ${install_dir}/bin/${serverName}-dump-cfg.gdb
fi
......@@ -233,8 +233,10 @@ if [ "$verMode" == "cluster" ]; then
sed 's/verMode=edge/verMode=cluster/g' ${install_dir}/bin/remove.sh >>remove_temp.sh
sed -i "s/serverName2=\"taosd\"/serverName2=\"${serverName2}\"/g" remove_temp.sh
sed -i "s/clientName2=\"taos\"/clientName2=\"${clientName2}\"/g" remove_temp.sh
sed -i "s/configFile2=\"taos.cfg\"/configFile2=\"${clientName2}.cfg\"/g" remove_temp.sh
sed -i "s/productName2=\"TDengine\"/productName2=\"${productName2}\"/g" remove_temp.sh
sed -i "s/emailName2=\"taosdata.com\"/emailName2=\"${cusEmail2}\"/g" remove_temp.sh
cusDomain=`echo "${cusEmail2}" | sed 's/^[^@]*@//'`
sed -i "s/emailName2=\"taosdata.com\"/emailName2=\"${cusDomain}\"/g" remove_temp.sh
mv remove_temp.sh ${install_dir}/bin/remove.sh
fi
if [ "$verMode" == "cloud" ]; then
......@@ -262,8 +264,10 @@ if [ "$verMode" == "cluster" ]; then
sed -i 's/verMode=edge/verMode=cluster/g' install_temp.sh
sed -i "s/serverName2=\"taosd\"/serverName2=\"${serverName2}\"/g" install_temp.sh
sed -i "s/clientName2=\"taos\"/clientName2=\"${clientName2}\"/g" install_temp.sh
sed -i "s/configFile2=\"taos.cfg\"/configFile2=\"${clientName2}.cfg\"/g" install_temp.sh
sed -i "s/productName2=\"TDengine\"/productName2=\"${productName2}\"/g" install_temp.sh
sed -i "s/emailName2=\"taosdata.com\"/emailName2=\"${cusEmail2}\"/g" install_temp.sh
cusDomain=`echo "${cusEmail2}" | sed 's/^[^@]*@//'`
sed -i "s/emailName2=\"taosdata.com\"/emailName2=\"${cusDomain}\"/g" install_temp.sh
mv install_temp.sh ${install_dir}/install.sh
fi
if [ "$verMode" == "cloud" ]; then
......
......@@ -40,11 +40,16 @@ serverName2="taosd"
clientName2="taos"
productName2="TDengine"
adapterName2="${clientName2}adapter"
demoName2="${clientName2}demo"
benchmarkName2="${clientName2}Benchmark"
dumpName2="${clientName2}dump"
keeperName2="${clientName2}keeper"
xName2="${clientName2}x"
explorerName2="${clientName2}-explorer"
uninstallScript2="rm${clientName2}"
installDir="/usr/local/${clientName}"
installDir="/usr/local/${clientName2}"
#install main path
install_main_dir=${installDir}
......@@ -55,8 +60,8 @@ local_bin_link_dir="/usr/local/bin"
service_config_dir="/etc/systemd/system"
taos_service_name=${serverName}
taosadapter_service_name="taosadapter"
taos_service_name=${serverName2}
taosadapter_service_name="${clientName2}adapter"
tarbitrator_service_name="tarbitratord"
csudo=""
if command -v sudo >/dev/null; then
......@@ -84,14 +89,14 @@ else
fi
function kill_taosadapter() {
pid=$(ps -ef | grep "taosadapter" | grep -v "grep" | awk '{print $2}')
pid=$(ps -ef | grep "${adapterName2}" | grep -v "grep" | awk '{print $2}')
if [ -n "$pid" ]; then
${csudo}kill -9 $pid || :
fi
}
function kill_taosd() {
pid=$(ps -ef | grep ${serverName} | grep -v "grep" | awk '{print $2}')
pid=$(ps -ef | grep ${serverName2} | grep -v "grep" | awk '{print $2}')
if [ -n "$pid" ]; then
${csudo}kill -9 $pid || :
fi
......@@ -109,17 +114,17 @@ function clean_bin() {
${csudo}rm -f ${bin_link_dir}/${clientName} || :
${csudo}rm -f ${bin_link_dir}/${serverName} || :
${csudo}rm -f ${bin_link_dir}/udfd || :
${csudo}rm -f ${bin_link_dir}/taosadapter || :
${csudo}rm -f ${bin_link_dir}/taosBenchmark || :
${csudo}rm -f ${bin_link_dir}/taosdemo || :
${csudo}rm -f ${bin_link_dir}/taosdump || :
${csudo}rm -f ${bin_link_dir}/${uninstallScript} || :
${csudo}rm -f ${bin_link_dir}/${adapterName2} || :
${csudo}rm -f ${bin_link_dir}/${benchmarkName2} || :
${csudo}rm -f ${bin_link_dir}/${demoName2} || :
${csudo}rm -f ${bin_link_dir}/${dumpName2} || :
${csudo}rm -f ${bin_link_dir}/${uninstallScript} || :
${csudo}rm -f ${bin_link_dir}/tarbitrator || :
${csudo}rm -f ${bin_link_dir}/set_core || :
${csudo}rm -f ${bin_link_dir}/TDinsight.sh || :
${csudo}rm -f ${bin_link_dir}/taoskeeper || :
${csudo}rm -f ${bin_link_dir}/taosx || :
${csudo}rm -f ${bin_link_dir}/taos-explorer || :
${csudo}rm -f ${bin_link_dir}/${keeperName2} || :
${csudo}rm -f ${bin_link_dir}/${xName2} || :
${csudo}rm -f ${bin_link_dir}/${explorerName2} || :
if [ "$verMode" == "cluster" ] && [ "$clientName" != "$clientName2" ]; then
${csudo}rm -f ${bin_link_dir}/${clientName2} || :
......@@ -130,8 +135,8 @@ function clean_bin() {
}
function clean_local_bin() {
${csudo}rm -f ${local_bin_link_dir}/taosBenchmark || :
${csudo}rm -f ${local_bin_link_dir}/taosdemo || :
${csudo}rm -f ${local_bin_link_dir}/${benchmarkName2} || :
${csudo}rm -f ${local_bin_link_dir}/${demoName2} || :
}
function clean_lib() {
......@@ -173,7 +178,7 @@ function clean_service_on_systemd() {
${csudo}systemctl disable ${taos_service_name} &>/dev/null || echo &>/dev/null
${csudo}rm -f ${taosd_service_config}
taosadapter_service_config="${service_config_dir}/taosadapter.service"
taosadapter_service_config="${service_config_dir}/${clientName2}adapter.service"
if systemctl is-active --quiet ${taosadapter_service_name}; then
echo "${productName2} ${clientName2}Adapter is running, stopping it..."
${csudo}systemctl stop ${taosadapter_service_name} &>/dev/null || echo &>/dev/null
......@@ -235,8 +240,8 @@ function clean_service_on_sysvinit() {
function clean_service_on_launchctl() {
${csudouser}launchctl unload -w /Library/LaunchDaemons/com.taosdata.taosd.plist > /dev/null 2>&1 || :
${csudo}rm /Library/LaunchDaemons/com.taosdata.taosd.plist > /dev/null 2>&1 || :
${csudouser}launchctl unload -w /Library/LaunchDaemons/com.taosdata.taosadapter.plist > /dev/null 2>&1 || :
${csudo}rm /Library/LaunchDaemons/com.taosdata.taosadapter.plist > /dev/null 2>&1 || :
${csudouser}launchctl unload -w /Library/LaunchDaemons/com.taosdata.${clientName2}adapter.plist > /dev/null 2>&1 || :
${csudo}rm /Library/LaunchDaemons/com.taosdata.${clientName2}adapter.plist > /dev/null 2>&1 || :
}
function clean_service() {
......
......@@ -15,11 +15,12 @@ uninstallScript="rmtaos"
clientName2="taos"
productName2="TDengine"
benchmarkName2="${clientName}Benchmark"
dumpName2="${clientName}dump"
uninstallScript2="rm${clientName}"
benchmarkName2="${clientName2}Benchmark"
demoName2="${clientName2}demo"
dumpName2="${clientName2}dump"
uninstallScript2="rm${clientName2}"
installDir="/usr/local/${clientName}"
installDir="/usr/local/${clientName2}"
#install main path
install_main_dir=${installDir}
......@@ -44,14 +45,17 @@ function kill_client() {
function clean_bin() {
# Remove link
${csudo}rm -f ${bin_link_dir}/${clientName} || :
${csudo}rm -f ${bin_link_dir}/taosdemo || :
${csudo}rm -f ${bin_link_dir}/taosdump || :
${csudo}rm -f ${bin_link_dir}/${clientName2} || :
${csudo}rm -f ${bin_link_dir}/${demoName2} || :
${csudo}rm -f ${bin_link_dir}/${benchmarkName2} || :
${csudo}rm -f ${bin_link_dir}/${dumpName2} || :
${csudo}rm -f ${bin_link_dir}/${uninstallScript} || :
${csudo}rm -f ${bin_link_dir}/set_core || :
if [ "$verMode" == "cluster" ] && [ "$clientName" != "$clientName2" ]; then
${csudo}rm -f ${bin_link_dir}/${clientName2} || :
${csudo}rm -f ${bin_link_dir}/${demoName2} || :
${csudo}rm -f ${bin_link_dir}/${benchmarkName2} || :
${csudo}rm -f ${bin_link_dir}/${dumpName2} || :
${csudo}rm -f ${bin_link_dir}/${uninstallScript2} || :
fi
......
......@@ -30,6 +30,10 @@
#include "tsched.h"
#include "ttime.h"
#if defined(CUS_NAME) || defined(CUS_PROMPT) || defined(CUS_EMAIL)
#include "cus_name.h"
#endif
#define TSC_VAR_NOT_RELEASE 1
#define TSC_VAR_RELEASED 0
......@@ -542,9 +546,15 @@ void taos_init_imp(void) {
deltaToUtcInitOnce();
if (taosCreateLog("taoslog", 10, configDir, NULL, NULL, NULL, NULL, 1) != 0) {
char logDirName[64] = {0};
#ifdef CUS_PROMPT
snprintf(logDirName, 64, "%slog", CUS_PROMPT);
#else
snprintf(logDirName, 64, "taoslog");
#endif
if (taosCreateLog(logDirName, 10, configDir, NULL, NULL, NULL, NULL, 1) != 0) {
// ignore create log failed, only print
printf(" WARING: Create taoslog failed:%s. configDir=%s\n", strerror(errno), configDir);
printf(" WARING: Create %s failed:%s. configDir=%s\n", logDirName, strerror(errno), configDir);
}
if (taosInitCfg(configDir, NULL, NULL, NULL, NULL, 1) != 0) {
......
......@@ -228,7 +228,11 @@ static int32_t taosLoadCfg(SConfig *pCfg, const char **envCmd, const char *input
taosExpandDir(inputCfgDir, cfgDir, PATH_MAX);
if (taosIsDir(cfgDir)) {
#ifdef CUS_PROMPT
snprintf(cfgFile, sizeof(cfgFile), "%s" TD_DIRSEP "%s.cfg", CUS_PROMPT, cfgDir);
#else
snprintf(cfgFile, sizeof(cfgFile), "%s" TD_DIRSEP "taos.cfg", cfgDir);
#endif
} else {
tstrncpy(cfgFile, cfgDir, sizeof(cfgDir));
}
......
......@@ -20,6 +20,21 @@
#include "tglobal.h"
#include "version.h"
#if defined(CUS_NAME) || defined(CUS_PROMPT) || defined(CUS_EMAIL)
#include "cus_name.h"
#else
#ifndef CUS_NAME
#define CUS_NAME "TDengine"
#endif
#ifndef CUS_PROMPT
#define CUS_PROMPT "taos"
#endif
#ifndef CUS_EMAIL
#define CUS_EMAIL "<support@taosdata.com>"
#endif
#endif
// clang-format off
#define DM_APOLLO_URL "The apollo string to use when configuring the server, such as: -a 'jsonFile:./tests/cfg.json', cfg.json text can be '{\"fqdn\":\"td1\"}'."
#define DM_CFG_DIR "Configuration directory."
......@@ -232,7 +247,7 @@ static void dmDumpCfg() {
}
static int32_t dmInitLog() {
return taosCreateLog("taosdlog", 1, configDir, global.envCmd, global.envFile, global.apolloUrl, global.pArgs, 0);
return taosCreateLog(CUS_PROMPT"dlog", 1, configDir, global.envCmd, global.envFile, global.apolloUrl, global.pArgs, 0);
}
static void taosCleanupArgs() {
......
......@@ -123,12 +123,12 @@ int32_t tsdbRowIterOpen(STSDBRowIter *pIter, TSDBROW *pRow, STSchema *pTSchema)
void tsdbRowClose(STSDBRowIter *pIter);
SColVal *tsdbRowIterNext(STSDBRowIter *pIter);
// SRowMerger
int32_t tsdbRowMergerInit2(SRowMerger *pMerger, STSchema *pResTSchema, TSDBROW *pRow, STSchema *pTSchema);
int32_t tsdbRowMergerInit(SRowMerger *pMerger, STSchema *pResTSchema, TSDBROW *pRow, STSchema *pTSchema);
int32_t tsdbRowMergerAdd(SRowMerger *pMerger, TSDBROW *pRow, STSchema *pTSchema);
int32_t tsdbRowMergerInit(SRowMerger *pMerger, TSDBROW *pRow, STSchema *pTSchema);
void tsdbRowMergerClear(SRowMerger *pMerger);
int32_t tsdbRowMerge(SRowMerger *pMerger, TSDBROW *pRow);
// int32_t tsdbRowMergerInit(SRowMerger *pMerger, TSDBROW *pRow, STSchema *pTSchema);
void tsdbRowMergerClear(SRowMerger *pMerger);
// int32_t tsdbRowMerge(SRowMerger *pMerger, TSDBROW *pRow);
int32_t tsdbRowMergerGetRow(SRowMerger *pMerger, SRow **ppRow);
// TABLEID
int32_t tTABLEIDCmprFn(const void *p1, const void *p2);
......@@ -224,7 +224,7 @@ int32_t tsdbTbDataIterCreate(STbData *pTbData, TSDBKEY *pFrom, int8_t backward,
void *tsdbTbDataIterDestroy(STbDataIter *pIter);
void tsdbTbDataIterOpen(STbData *pTbData, TSDBKEY *pFrom, int8_t backward, STbDataIter *pIter);
bool tsdbTbDataIterNext(STbDataIter *pIter);
void tsdbMemTableCountRows(SMemTable *pMemTable, SHashObj* pTableMap, int64_t *rowsNum);
void tsdbMemTableCountRows(SMemTable *pMemTable, SHashObj *pTableMap, int64_t *rowsNum);
// STbData
int32_t tsdbGetNRowsInTbData(STbData *pTbData);
......
......@@ -170,7 +170,7 @@ int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMeta
qStreamExtractOffset(task, &tmp->rspOffset);
*pMetaRsp = *tmp;
tqDebug("tmqsnap task get data");
tqDebug("tmqsnap task get meta");
break;
}
......
......@@ -638,13 +638,17 @@ SColVal *tsdbRowIterNext(STSDBRowIter *pIter) {
// SRowMerger ======================================================
int32_t tsdbRowMergerInit2(SRowMerger *pMerger, STSchema *pResTSchema, TSDBROW *pRow, STSchema *pTSchema) {
int32_t tsdbRowMergerInit(SRowMerger *pMerger, STSchema *pResTSchema, TSDBROW *pRow, STSchema *pTSchema) {
int32_t code = 0;
TSDBKEY key = TSDBROW_KEY(pRow);
SColVal *pColVal = &(SColVal){0};
STColumn *pTColumn;
int32_t iCol, jCol = 0;
if (NULL == pResTSchema) {
pResTSchema = pTSchema;
}
pMerger->pTSchema = pResTSchema;
pMerger->version = key.version;
......@@ -712,6 +716,9 @@ int32_t tsdbRowMergerAdd(SRowMerger *pMerger, TSDBROW *pRow, STSchema *pTSchema)
STColumn *pTColumn;
int32_t iCol, jCol = 1;
if (NULL == pTSchema) {
pTSchema = pMerger->pTSchema;
}
ASSERT(((SColVal *)pMerger->pArray->pData)->value.val == key.ts);
for (iCol = 1; iCol < pMerger->pTSchema->numOfCols && jCol < pTSchema->numOfCols; ++iCol) {
......@@ -727,23 +734,6 @@ int32_t tsdbRowMergerAdd(SRowMerger *pMerger, TSDBROW *pRow, STSchema *pTSchema)
tsdbRowGetColVal(pRow, pTSchema, jCol++, pColVal);
if (key.version > pMerger->version) {
#if 0
if (!COL_VAL_IS_NONE(pColVal)) {
if ((!COL_VAL_IS_NULL(pColVal)) && IS_VAR_DATA_TYPE(pColVal->type)) {
SColVal *tColVal = taosArrayGet(pMerger->pArray, iCol);
code = tRealloc(&tColVal->value.pData, pColVal->value.nData);
if (code) return code;
tColVal->value.nData = pColVal->value.nData;
if (pColVal->value.nData) {
memcpy(tColVal->value.pData, pColVal->value.pData, pColVal->value.nData);
}
tColVal->flag = 0;
} else {
taosArraySet(pMerger->pArray, iCol, pColVal);
}
}
#endif
if (!COL_VAL_IS_NONE(pColVal)) {
if (IS_VAR_DATA_TYPE(pColVal->type)) {
SColVal *pTColVal = taosArrayGet(pMerger->pArray, iCol);
......@@ -758,7 +748,6 @@ int32_t tsdbRowMergerAdd(SRowMerger *pMerger, TSDBROW *pRow, STSchema *pTSchema)
pTColVal->flag = 0;
} else {
tFree(pTColVal->value.pData);
pTColVal->value.pData = NULL;
taosArraySet(pMerger->pArray, iCol, pColVal);
}
} else {
......@@ -789,7 +778,7 @@ int32_t tsdbRowMergerAdd(SRowMerger *pMerger, TSDBROW *pRow, STSchema *pTSchema)
pMerger->version = key.version;
return code;
}
/*
int32_t tsdbRowMergerInit(SRowMerger *pMerger, TSDBROW *pRow, STSchema *pTSchema) {
int32_t code = 0;
TSDBKEY key = TSDBROW_KEY(pRow);
......@@ -840,7 +829,7 @@ int32_t tsdbRowMergerInit(SRowMerger *pMerger, TSDBROW *pRow, STSchema *pTSchema
_exit:
return code;
}
*/
void tsdbRowMergerClear(SRowMerger *pMerger) {
for (int32_t iCol = 1; iCol < pMerger->pTSchema->numOfCols; iCol++) {
SColVal *pTColVal = taosArrayGet(pMerger->pArray, iCol);
......@@ -851,7 +840,7 @@ void tsdbRowMergerClear(SRowMerger *pMerger) {
taosArrayDestroy(pMerger->pArray);
}
/*
int32_t tsdbRowMerge(SRowMerger *pMerger, TSDBROW *pRow) {
int32_t code = 0;
TSDBKEY key = TSDBROW_KEY(pRow);
......@@ -916,7 +905,7 @@ int32_t tsdbRowMerge(SRowMerger *pMerger, TSDBROW *pRow) {
_exit:
return code;
}
*/
int32_t tsdbRowMergerGetRow(SRowMerger *pMerger, SRow **ppRow) {
return tRowBuild(pMerger->pArray, pMerger->pTSchema, ppRow);
}
......
......@@ -1200,7 +1200,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
tableListClear(pTableListInfo);
if (mtInfo.uid == 0) {
return 0; // no data
goto end; // no data
}
initQueryTableDataCondForTmq(&pTaskInfo->streamInfo.tableCond, sContext, &mtInfo);
......@@ -1234,6 +1234,8 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
qDebug("tmqsnap qStreamPrepareScan snapshot log, %s", id);
}
}
end:
pTaskInfo->streamInfo.currentOffset = *pOffset;
return 0;
......
......@@ -125,6 +125,12 @@ static int32_t smlBuildTagRow(SArray* cols, SBoundColInfo* tags, SSchema* pSchem
SSchema* pTagSchema = &pSchema[tags->pColIndex[i]];
SSmlKv* kv = taosArrayGet(cols, i);
if(kv->keyLen != strlen(pTagSchema->name) || memcmp(kv->key, pTagSchema->name, kv->keyLen) != 0 || kv->type != pTagSchema->type){
code = TSDB_CODE_SML_INVALID_DATA;
uError("SML smlBuildCol error col not same %s", pTagSchema->name);
goto end;
}
taosArrayPush(*tagName, pTagSchema->name);
STagVal val = {.cid = pTagSchema->colId, .type = pTagSchema->type};
// strcpy(val.colName, pTagSchema->name);
......
......@@ -59,36 +59,36 @@ typedef struct SSyncLogBuffer {
} SSyncLogBuffer;
// SSyncLogRepMgr
SSyncLogReplMgr* syncLogReplMgrCreate();
void syncLogReplMgrDestroy(SSyncLogReplMgr* pMgr);
void syncLogReplMgrReset(SSyncLogReplMgr* pMgr);
SSyncLogReplMgr* syncLogReplCreate();
void syncLogReplDestroy(SSyncLogReplMgr* pMgr);
void syncLogReplReset(SSyncLogReplMgr* pMgr);
int32_t syncNodeLogReplMgrInit(SSyncNode* pNode);
void syncNodeLogReplMgrDestroy(SSyncNode* pNode);
int32_t syncNodeLogReplInit(SSyncNode* pNode);
void syncNodeLogReplDestroy(SSyncNode* pNode);
// access
static FORCE_INLINE int64_t syncLogGetRetryBackoffTimeMs(SSyncLogReplMgr* pMgr) {
static FORCE_INLINE int64_t syncLogReplGetRetryBackoffTimeMs(SSyncLogReplMgr* pMgr) {
return ((int64_t)1 << pMgr->retryBackoff) * SYNC_LOG_REPL_RETRY_WAIT_MS;
}
static FORCE_INLINE int32_t syncLogGetNextRetryBackoff(SSyncLogReplMgr* pMgr) {
static FORCE_INLINE int32_t syncLogReplGetNextRetryBackoff(SSyncLogReplMgr* pMgr) {
return TMIN(pMgr->retryBackoff + 1, SYNC_MAX_RETRY_BACKOFF);
}
SyncTerm syncLogReplMgrGetPrevLogTerm(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncIndex index);
SyncTerm syncLogReplGetPrevLogTerm(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncIndex index);
int32_t syncLogReplMgrReplicateOnce(SSyncLogReplMgr* pMgr, SSyncNode* pNode);
int32_t syncLogReplMgrReplicateOneTo(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncIndex index, SyncTerm* pTerm,
SRaftId* pDestId, bool* pBarrier);
int32_t syncLogReplMgrReplicateAttempt(SSyncLogReplMgr* pMgr, SSyncNode* pNode);
int32_t syncLogReplMgrReplicateProbe(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncIndex index);
int32_t syncLogReplReplicateOnce(SSyncLogReplMgr* pMgr, SSyncNode* pNode);
int32_t syncLogReplReplicateOneTo(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncIndex index, SyncTerm* pTerm,
SRaftId* pDestId, bool* pBarrier);
int32_t syncLogReplReplicateAttempt(SSyncLogReplMgr* pMgr, SSyncNode* pNode);
int32_t syncLogReplReplicateProbe(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncIndex index);
int32_t syncLogReplMgrProcessReply(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncAppendEntriesReply* pMsg);
int32_t syncLogReplMgrProcessReplyAsRecovery(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncAppendEntriesReply* pMsg);
int32_t syncLogReplMgrProcessReplyAsNormal(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncAppendEntriesReply* pMsg);
int32_t syncLogReplProcessReply(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncAppendEntriesReply* pMsg);
int32_t syncLogReplProcessReplyAsRecovery(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncAppendEntriesReply* pMsg);
int32_t syncLogReplProcessReplyAsNormal(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncAppendEntriesReply* pMsg);
int32_t syncLogReplMgrProcessHeartbeatReply(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncHeartbeatReply* pMsg);
int32_t syncLogReplMgrRetryOnNeed(SSyncLogReplMgr* pMgr, SSyncNode* pNode);
int32_t syncLogReplProcessHeartbeatReply(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncHeartbeatReply* pMsg);
int32_t syncLogReplRetryOnNeed(SSyncLogReplMgr* pMgr, SSyncNode* pNode);
// SSyncLogBuffer
SSyncLogBuffer* syncLogBufferCreate();
......@@ -111,8 +111,8 @@ SSyncRaftEntry* syncLogBufferGetOneEntry(SSyncLogBuffer* pBuf, SSyncNode* pNode,
int32_t syncLogBufferValidate(SSyncLogBuffer* pBuf);
int32_t syncLogBufferRollback(SSyncLogBuffer* pBuf, SSyncNode* pNode, SyncIndex toIndex);
int32_t syncLogFsmExecute(SSyncNode* pNode, SSyncFSM* pFsm, ESyncState role, SyncTerm term, SSyncRaftEntry* pEntry,
int32_t applyCode);
int32_t syncFsmExecute(SSyncNode* pNode, SSyncFSM* pFsm, ESyncState role, SyncTerm term, SSyncRaftEntry* pEntry,
int32_t applyCode);
#ifdef __cplusplus
}
#endif
......
......@@ -85,7 +85,7 @@ int32_t syncNodeOnAppendEntriesReply(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
sError("vgId:%d, failed to get log repl mgr for src addr: 0x%016" PRIx64, ths->vgId, pMsg->srcId.addr);
return -1;
}
(void)syncLogReplMgrProcessReply(pMgr, ths, pMsg);
(void)syncLogReplProcessReply(pMgr, ths, pMsg);
}
return 0;
}
......@@ -949,7 +949,7 @@ SSyncNode* syncNodeOpen(SSyncInfo* pSyncInfo) {
pSyncNode->changing = false;
// replication mgr
if (syncNodeLogReplMgrInit(pSyncNode) < 0) {
if (syncNodeLogReplInit(pSyncNode) < 0) {
sError("vgId:%d, failed to init repl mgr since %s.", pSyncNode->vgId, terrstr());
goto _error;
}
......@@ -1122,7 +1122,7 @@ void syncNodeClose(SSyncNode* pSyncNode) {
syncNodeStopPingTimer(pSyncNode);
syncNodeStopElectTimer(pSyncNode);
syncNodeStopHeartbeatTimer(pSyncNode);
syncNodeLogReplMgrDestroy(pSyncNode);
syncNodeLogReplDestroy(pSyncNode);
syncRespMgrDestroy(pSyncNode->pSyncRespMgr);
pSyncNode->pSyncRespMgr = NULL;
......@@ -2164,7 +2164,7 @@ int32_t syncNodeAppend(SSyncNode* ths, SSyncRaftEntry* pEntry) {
if (syncLogBufferAppend(ths->pLogBuf, ths, pEntry) < 0) {
sError("vgId:%d, failed to enqueue sync log buffer, index:%" PRId64, ths->vgId, pEntry->index);
ASSERT(terrno != 0);
(void)syncLogFsmExecute(ths, ths->pFsm, ths->state, raftStoreGetTerm(ths), pEntry, terrno);
(void)syncFsmExecute(ths, ths->pFsm, ths->state, raftStoreGetTerm(ths), pEntry, terrno);
syncEntryDestroy(pEntry);
return -1;
}
......@@ -2374,7 +2374,7 @@ int32_t syncNodeOnHeartbeatReply(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
syncIndexMgrSetRecvTime(ths->pMatchIndex, &pMsg->srcId, tsMs);
return syncLogReplMgrProcessHeartbeatReply(pMgr, ths, pMsg);
return syncLogReplProcessHeartbeatReply(pMgr, ths, pMsg);
}
int32_t syncNodeOnHeartbeatReplyOld(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
......
......@@ -86,7 +86,7 @@ _err:
return -1;
}
SyncTerm syncLogReplMgrGetPrevLogTerm(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncIndex index) {
SyncTerm syncLogReplGetPrevLogTerm(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncIndex index) {
SSyncLogBuffer* pBuf = pNode->pLogBuf;
SSyncRaftEntry* pEntry = NULL;
SyncIndex prevIndex = index - 1;
......@@ -316,7 +316,7 @@ int32_t syncLogBufferAccept(SSyncLogBuffer* pBuf, SSyncNode* pNode, SSyncRaftEnt
" %" PRId64 ", %" PRId64 ")",
pNode->vgId, pEntry->index, pEntry->term, pBuf->startIndex, pBuf->commitIndex, pBuf->matchIndex,
pBuf->endIndex);
SyncTerm term = syncLogReplMgrGetPrevLogTerm(NULL, pNode, index + 1);
SyncTerm term = syncLogReplGetPrevLogTerm(NULL, pNode, index + 1);
ASSERT(pEntry->term >= 0);
if (term == pEntry->term) {
ret = 0;
......@@ -351,7 +351,7 @@ int32_t syncLogBufferAccept(SSyncLogBuffer* pBuf, SSyncNode* pNode, SSyncRaftEnt
" %" PRId64 " %" PRId64 ", %" PRId64 ")",
pNode->vgId, pEntry->index, pEntry->term, pBuf->startIndex, pBuf->commitIndex, pBuf->matchIndex,
pBuf->endIndex);
SyncTerm existPrevTerm = syncLogReplMgrGetPrevLogTerm(NULL, pNode, index);
SyncTerm existPrevTerm = syncLogReplGetPrevLogTerm(NULL, pNode, index);
ASSERT(pEntry->term == pExist->term && (pEntry->index > pBuf->matchIndex || prevTerm == existPrevTerm));
ret = 0;
goto _out;
......@@ -482,8 +482,8 @@ _out:
return matchIndex;
}
int32_t syncLogFsmExecute(SSyncNode* pNode, SSyncFSM* pFsm, ESyncState role, SyncTerm term, SSyncRaftEntry* pEntry,
int32_t applyCode) {
int32_t syncFsmExecute(SSyncNode* pNode, SSyncFSM* pFsm, ESyncState role, SyncTerm term, SSyncRaftEntry* pEntry,
int32_t applyCode) {
if (pNode->replicaNum == 1 && pNode->restoreFinish && pNode->vgId != 1) {
return 0;
}
......@@ -564,7 +564,7 @@ int32_t syncLogBufferCommit(SSyncLogBuffer* pBuf, SSyncNode* pNode, int64_t comm
pEntry->term, TMSG_INFO(pEntry->originalRpcType));
}
if (syncLogFsmExecute(pNode, pFsm, role, currentTerm, pEntry, 0) != 0) {
if (syncFsmExecute(pNode, pFsm, role, currentTerm, pEntry, 0) != 0) {
sError("vgId:%d, failed to execute sync log entry. index:%" PRId64 ", term:%" PRId64
", role:%d, current term:%" PRId64,
vgId, pEntry->index, pEntry->term, role, currentTerm);
......@@ -611,7 +611,7 @@ _out:
return ret;
}
void syncLogReplMgrReset(SSyncLogReplMgr* pMgr) {
void syncLogReplReset(SSyncLogReplMgr* pMgr) {
if (pMgr == NULL) return;
ASSERT(pMgr->startIndex >= 0);
......@@ -625,14 +625,14 @@ void syncLogReplMgrReset(SSyncLogReplMgr* pMgr) {
pMgr->retryBackoff = 0;
}
int32_t syncLogReplMgrRetryOnNeed(SSyncLogReplMgr* pMgr, SSyncNode* pNode) {
int32_t syncLogReplRetryOnNeed(SSyncLogReplMgr* pMgr, SSyncNode* pNode) {
if (pMgr->endIndex <= pMgr->startIndex) {
return 0;
}
SRaftId* pDestId = &pNode->replicasId[pMgr->peerId];
if (pMgr->retryBackoff == SYNC_MAX_RETRY_BACKOFF) {
syncLogReplMgrReset(pMgr);
syncLogReplReset(pMgr);
sWarn("vgId:%d, reset sync log repl mgr since retry backoff exceeding limit. peer:%" PRIx64, pNode->vgId,
pDestId->addr);
return -1;
......@@ -640,7 +640,7 @@ int32_t syncLogReplMgrRetryOnNeed(SSyncLogReplMgr* pMgr, SSyncNode* pNode) {
int32_t ret = -1;
bool retried = false;
int64_t retryWaitMs = syncLogGetRetryBackoffTimeMs(pMgr);
int64_t retryWaitMs = syncLogReplGetRetryBackoffTimeMs(pMgr);
int64_t nowMs = taosGetMonoTimestampMs();
int count = 0;
int64_t firstIndex = -1;
......@@ -657,7 +657,7 @@ int32_t syncLogReplMgrRetryOnNeed(SSyncLogReplMgr* pMgr, SSyncNode* pNode) {
if (pMgr->states[pos].acked) {
if (pMgr->matchIndex < index && pMgr->states[pos].timeMs + (syncGetRetryMaxWaitMs() << 3) < nowMs) {
syncLogReplMgrReset(pMgr);
syncLogReplReset(pMgr);
sWarn("vgId:%d, reset sync log repl mgr since stagnation. index:%" PRId64 ", peer:%" PRIx64, pNode->vgId,
index, pDestId->addr);
goto _out;
......@@ -666,7 +666,7 @@ int32_t syncLogReplMgrRetryOnNeed(SSyncLogReplMgr* pMgr, SSyncNode* pNode) {
}
bool barrier = false;
if (syncLogReplMgrReplicateOneTo(pMgr, pNode, index, &term, pDestId, &barrier) < 0) {
if (syncLogReplReplicateOneTo(pMgr, pNode, index, &term, pDestId, &barrier) < 0) {
sError("vgId:%d, failed to replicate sync log entry since %s. index:%" PRId64 ", dest:%" PRIx64 "", pNode->vgId,
terrstr(), index, pDestId->addr);
goto _out;
......@@ -687,7 +687,7 @@ int32_t syncLogReplMgrRetryOnNeed(SSyncLogReplMgr* pMgr, SSyncNode* pNode) {
ret = 0;
_out:
if (retried) {
pMgr->retryBackoff = syncLogGetNextRetryBackoff(pMgr);
pMgr->retryBackoff = syncLogReplGetNextRetryBackoff(pMgr);
SSyncLogBuffer* pBuf = pNode->pLogBuf;
sInfo("vgId:%d, resend %d sync log entries. dest:%" PRIx64 ", indexes:%" PRId64 " ..., terms: ... %" PRId64
", retryWaitMs:%" PRId64 ", mgr: [%" PRId64 " %" PRId64 ", %" PRId64 "), buffer: [%" PRId64 " %" PRId64
......@@ -698,7 +698,7 @@ _out:
return ret;
}
int32_t syncLogReplMgrProcessReplyAsRecovery(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncAppendEntriesReply* pMsg) {
int32_t syncLogReplProcessReplyAsRecovery(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncAppendEntriesReply* pMsg) {
SSyncLogBuffer* pBuf = pNode->pLogBuf;
SRaftId destId = pMsg->srcId;
ASSERT(pMgr->restored == false);
......@@ -716,7 +716,7 @@ int32_t syncLogReplMgrProcessReplyAsRecovery(SSyncLogReplMgr* pMgr, SSyncNode* p
}
} else {
if (pMsg->lastSendIndex < pMgr->startIndex || pMsg->lastSendIndex >= pMgr->endIndex) {
syncLogReplMgrRetryOnNeed(pMgr, pNode);
syncLogReplRetryOnNeed(pMgr, pNode);
return 0;
}
......@@ -750,7 +750,7 @@ int32_t syncLogReplMgrProcessReplyAsRecovery(SSyncLogReplMgr* pMgr, SSyncNode* p
SyncIndex index = TMIN(pMsg->matchIndex, pNode->pLogBuf->matchIndex);
if (pMsg->matchIndex < pNode->pLogBuf->matchIndex) {
term = syncLogReplMgrGetPrevLogTerm(pMgr, pNode, index + 1);
term = syncLogReplGetPrevLogTerm(pMgr, pNode, index + 1);
if ((index + 1 < firstVer) || (term < 0) ||
(term != pMsg->lastMatchTerm && (index + 1 == firstVer || index == firstVer))) {
ASSERT(term >= 0 || terrno == TSDB_CODE_WAL_LOG_NOT_EXIST);
......@@ -773,53 +773,53 @@ int32_t syncLogReplMgrProcessReplyAsRecovery(SSyncLogReplMgr* pMgr, SSyncNode* p
}
// attempt to replicate the raft log at index
(void)syncLogReplMgrReset(pMgr);
return syncLogReplMgrReplicateProbe(pMgr, pNode, index);
(void)syncLogReplReset(pMgr);
return syncLogReplReplicateProbe(pMgr, pNode, index);
}
int32_t syncLogReplMgrProcessHeartbeatReply(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncHeartbeatReply* pMsg) {
int32_t syncLogReplProcessHeartbeatReply(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncHeartbeatReply* pMsg) {
SSyncLogBuffer* pBuf = pNode->pLogBuf;
taosThreadMutexLock(&pBuf->mutex);
if (pMsg->startTime != 0 && pMsg->startTime != pMgr->peerStartTime) {
sInfo("vgId:%d, reset sync log repl mgr in heartbeat. peer:%" PRIx64 ", start time:%" PRId64 ", old:%" PRId64 "",
pNode->vgId, pMsg->srcId.addr, pMsg->startTime, pMgr->peerStartTime);
syncLogReplMgrReset(pMgr);
syncLogReplReset(pMgr);
pMgr->peerStartTime = pMsg->startTime;
}
taosThreadMutexUnlock(&pBuf->mutex);
return 0;
}
int32_t syncLogReplMgrProcessReply(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncAppendEntriesReply* pMsg) {
int32_t syncLogReplProcessReply(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncAppendEntriesReply* pMsg) {
SSyncLogBuffer* pBuf = pNode->pLogBuf;
taosThreadMutexLock(&pBuf->mutex);
if (pMsg->startTime != pMgr->peerStartTime) {
sInfo("vgId:%d, reset sync log repl mgr in appendlog reply. peer:%" PRIx64 ", start time:%" PRId64
", old:%" PRId64,
pNode->vgId, pMsg->srcId.addr, pMsg->startTime, pMgr->peerStartTime);
syncLogReplMgrReset(pMgr);
syncLogReplReset(pMgr);
pMgr->peerStartTime = pMsg->startTime;
}
if (pMgr->restored) {
(void)syncLogReplMgrProcessReplyAsNormal(pMgr, pNode, pMsg);
(void)syncLogReplProcessReplyAsNormal(pMgr, pNode, pMsg);
} else {
(void)syncLogReplMgrProcessReplyAsRecovery(pMgr, pNode, pMsg);
(void)syncLogReplProcessReplyAsRecovery(pMgr, pNode, pMsg);
}
taosThreadMutexUnlock(&pBuf->mutex);
return 0;
}
int32_t syncLogReplMgrReplicateOnce(SSyncLogReplMgr* pMgr, SSyncNode* pNode) {
int32_t syncLogReplReplicateOnce(SSyncLogReplMgr* pMgr, SSyncNode* pNode) {
if (pMgr->restored) {
(void)syncLogReplMgrReplicateAttempt(pMgr, pNode);
(void)syncLogReplReplicateAttempt(pMgr, pNode);
} else {
(void)syncLogReplMgrReplicateProbe(pMgr, pNode, pNode->pLogBuf->matchIndex);
(void)syncLogReplReplicateProbe(pMgr, pNode, pNode->pLogBuf->matchIndex);
}
return 0;
}
int32_t syncLogReplMgrReplicateProbe(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncIndex index) {
int32_t syncLogReplReplicateProbe(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncIndex index) {
ASSERT(!pMgr->restored);
ASSERT(pMgr->startIndex >= 0);
int64_t retryMaxWaitMs = syncGetRetryMaxWaitMs();
......@@ -829,12 +829,12 @@ int32_t syncLogReplMgrReplicateProbe(SSyncLogReplMgr* pMgr, SSyncNode* pNode, Sy
nowMs < pMgr->states[pMgr->startIndex % pMgr->size].timeMs + retryMaxWaitMs) {
return 0;
}
(void)syncLogReplMgrReset(pMgr);
(void)syncLogReplReset(pMgr);
SRaftId* pDestId = &pNode->replicasId[pMgr->peerId];
bool barrier = false;
SyncTerm term = -1;
if (syncLogReplMgrReplicateOneTo(pMgr, pNode, index, &term, pDestId, &barrier) < 0) {
if (syncLogReplReplicateOneTo(pMgr, pNode, index, &term, pDestId, &barrier) < 0) {
sError("vgId:%d, failed to replicate log entry since %s. index:%" PRId64 ", dest: 0x%016" PRIx64 "", pNode->vgId,
terrstr(), index, pDestId->addr);
return -1;
......@@ -857,7 +857,7 @@ int32_t syncLogReplMgrReplicateProbe(SSyncLogReplMgr* pMgr, SSyncNode* pNode, Sy
return 0;
}
int32_t syncLogReplMgrReplicateAttempt(SSyncLogReplMgr* pMgr, SSyncNode* pNode) {
int32_t syncLogReplReplicateAttempt(SSyncLogReplMgr* pMgr, SSyncNode* pNode) {
ASSERT(pMgr->restored);
SRaftId* pDestId = &pNode->replicasId[pMgr->peerId];
......@@ -879,7 +879,7 @@ int32_t syncLogReplMgrReplicateAttempt(SSyncLogReplMgr* pMgr, SSyncNode* pNode)
SRaftId* pDestId = &pNode->replicasId[pMgr->peerId];
bool barrier = false;
SyncTerm term = -1;
if (syncLogReplMgrReplicateOneTo(pMgr, pNode, index, &term, pDestId, &barrier) < 0) {
if (syncLogReplReplicateOneTo(pMgr, pNode, index, &term, pDestId, &barrier) < 0) {
sError("vgId:%d, failed to replicate log entry since %s. index:%" PRId64 ", dest: 0x%016" PRIx64 "", pNode->vgId,
terrstr(), index, pDestId->addr);
return -1;
......@@ -902,7 +902,7 @@ int32_t syncLogReplMgrReplicateAttempt(SSyncLogReplMgr* pMgr, SSyncNode* pNode)
}
}
syncLogReplMgrRetryOnNeed(pMgr, pNode);
syncLogReplRetryOnNeed(pMgr, pNode);
SSyncLogBuffer* pBuf = pNode->pLogBuf;
sTrace("vgId:%d, replicated %d msgs to peer:%" PRIx64 ". indexes:%" PRId64 "..., terms: ...%" PRId64
......@@ -913,7 +913,7 @@ int32_t syncLogReplMgrReplicateAttempt(SSyncLogReplMgr* pMgr, SSyncNode* pNode)
return 0;
}
int32_t syncLogReplMgrProcessReplyAsNormal(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncAppendEntriesReply* pMsg) {
int32_t syncLogReplProcessReplyAsNormal(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncAppendEntriesReply* pMsg) {
ASSERT(pMgr->restored == true);
if (pMgr->startIndex <= pMsg->lastSendIndex && pMsg->lastSendIndex < pMgr->endIndex) {
if (pMgr->startIndex < pMgr->matchIndex && pMgr->retryBackoff > 0) {
......@@ -932,10 +932,10 @@ int32_t syncLogReplMgrProcessReplyAsNormal(SSyncLogReplMgr* pMgr, SSyncNode* pNo
pMgr->startIndex = pMgr->matchIndex;
}
return syncLogReplMgrReplicateAttempt(pMgr, pNode);
return syncLogReplReplicateAttempt(pMgr, pNode);
}
SSyncLogReplMgr* syncLogReplMgrCreate() {
SSyncLogReplMgr* syncLogReplCreate() {
SSyncLogReplMgr* pMgr = taosMemoryCalloc(1, sizeof(SSyncLogReplMgr));
if (pMgr == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
......@@ -949,7 +949,7 @@ SSyncLogReplMgr* syncLogReplMgrCreate() {
return pMgr;
}
void syncLogReplMgrDestroy(SSyncLogReplMgr* pMgr) {
void syncLogReplDestroy(SSyncLogReplMgr* pMgr) {
if (pMgr == NULL) {
return;
}
......@@ -957,10 +957,10 @@ void syncLogReplMgrDestroy(SSyncLogReplMgr* pMgr) {
return;
}
int32_t syncNodeLogReplMgrInit(SSyncNode* pNode) {
int32_t syncNodeLogReplInit(SSyncNode* pNode) {
for (int i = 0; i < TSDB_MAX_REPLICA; i++) {
ASSERT(pNode->logReplMgrs[i] == NULL);
pNode->logReplMgrs[i] = syncLogReplMgrCreate();
pNode->logReplMgrs[i] = syncLogReplCreate();
if (pNode->logReplMgrs[i] == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
......@@ -970,9 +970,9 @@ int32_t syncNodeLogReplMgrInit(SSyncNode* pNode) {
return 0;
}
void syncNodeLogReplMgrDestroy(SSyncNode* pNode) {
void syncNodeLogReplDestroy(SSyncNode* pNode) {
for (int i = 0; i < TSDB_MAX_REPLICA; i++) {
syncLogReplMgrDestroy(pNode->logReplMgrs[i]);
syncLogReplDestroy(pNode->logReplMgrs[i]);
pNode->logReplMgrs[i] = NULL;
}
}
......@@ -1103,7 +1103,7 @@ int32_t syncLogBufferReset(SSyncLogBuffer* pBuf, SSyncNode* pNode) {
// reset repl mgr
for (int i = 0; i < pNode->replicaNum; i++) {
SSyncLogReplMgr* pMgr = pNode->logReplMgrs[i];
syncLogReplMgrReset(pMgr);
syncLogReplReset(pMgr);
}
syncLogBufferValidate(pBuf);
taosThreadMutexUnlock(&pBuf->mutex);
......@@ -1127,8 +1127,8 @@ SSyncRaftEntry* syncLogBufferGetOneEntry(SSyncLogBuffer* pBuf, SSyncNode* pNode,
return pEntry;
}
int32_t syncLogReplMgrReplicateOneTo(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncIndex index, SyncTerm* pTerm,
SRaftId* pDestId, bool* pBarrier) {
int32_t syncLogReplReplicateOneTo(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncIndex index, SyncTerm* pTerm,
SRaftId* pDestId, bool* pBarrier) {
SSyncRaftEntry* pEntry = NULL;
SRpcMsg msgOut = {0};
bool inBuf = false;
......@@ -1143,14 +1143,14 @@ int32_t syncLogReplMgrReplicateOneTo(SSyncLogReplMgr* pMgr, SSyncNode* pNode, Sy
if (pMgr) {
sInfo("vgId:%d, reset sync log repl mgr of peer:%" PRIx64 " since %s. index:%" PRId64, pNode->vgId,
pDestId->addr, terrstr(), index);
(void)syncLogReplMgrReset(pMgr);
(void)syncLogReplReset(pMgr);
}
}
goto _err;
}
*pBarrier = syncLogIsReplicationBarrier(pEntry);
prevLogTerm = syncLogReplMgrGetPrevLogTerm(pMgr, pNode, index);
prevLogTerm = syncLogReplGetPrevLogTerm(pMgr, pNode, index);
if (prevLogTerm < 0) {
sError("vgId:%d, failed to get prev log term since %s. index:%" PRId64 "", pNode->vgId, terrstr(), index);
goto _err;
......
......@@ -52,7 +52,7 @@ int32_t syncNodeReplicateReset(SSyncNode* pNode, SRaftId* pDestId) {
SSyncLogBuffer* pBuf = pNode->pLogBuf;
taosThreadMutexLock(&pBuf->mutex);
SSyncLogReplMgr* pMgr = syncNodeGetLogReplMgr(pNode, pDestId);
syncLogReplMgrReset(pMgr);
syncLogReplReset(pMgr);
taosThreadMutexUnlock(&pBuf->mutex);
return 0;
}
......@@ -74,7 +74,7 @@ int32_t syncNodeReplicateWithoutLock(SSyncNode* pNode) {
continue;
}
SSyncLogReplMgr* pMgr = pNode->logReplMgrs[i];
(void)syncLogReplMgrReplicateOnce(pMgr, pNode);
(void)syncLogReplReplicateOnce(pMgr, pNode);
}
return 0;
}
......
......@@ -125,7 +125,7 @@ static void syncLogBufferStates2Str(SSyncNode* pSyncNode, char* buf, int32_t buf
pBuf->commitIndex, pBuf->matchIndex, pBuf->endIndex);
}
static void syncLogReplMgrStates2Str(SSyncNode* pSyncNode, char* buf, int32_t bufLen) {
static void syncLogReplStates2Str(SSyncNode* pSyncNode, char* buf, int32_t bufLen) {
int len = 0;
len += snprintf(buf + len, bufLen - len, "%s", "{");
for (int32_t i = 0; i < pSyncNode->replicaNum; i++) {
......@@ -178,7 +178,7 @@ void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNo
syncCfg2SimpleStr(&pNode->raftCfg.cfg, cfgStr, sizeof(cfgStr));
char replMgrStatesStr[1024] = "";
syncLogReplMgrStates2Str(pNode, replMgrStatesStr, sizeof(replMgrStatesStr));
syncLogReplStates2Str(pNode, replMgrStatesStr, sizeof(replMgrStatesStr));
char bufferStatesStr[256] = "";
syncLogBufferStates2Str(pNode, bufferStatesStr, sizeof(bufferStatesStr));
......
......@@ -17,6 +17,10 @@
#include "os.h"
#include "taoserror.h"
#if defined(CUS_NAME) || defined(CUS_PROMPT) || defined(CUS_EMAIL)
#include "cus_name.h"
#endif
#define PROCESS_ITEM 12
#define UUIDLEN37 37
......@@ -252,7 +256,11 @@ int32_t taosGetEmail(char *email, int32_t maxLen) {
#ifdef WINDOWS
// ASSERT(0);
#elif defined(_TD_DARWIN_64)
#ifdef CUS_PROMPT
const char *filepath = "/usr/local/"CUS_PROMPT"/email";
#else
const char *filepath = "/usr/local/taos/email";
#endif // CUS_PROMPT
TdFilePtr pFile = taosOpenFile(filepath, TD_FILE_READ);
if (pFile == NULL) return false;
......@@ -264,8 +272,12 @@ int32_t taosGetEmail(char *email, int32_t maxLen) {
taosCloseFile(&pFile);
return 0;
#else
#ifdef CUS_PROMPT
const char *filepath = "/usr/local/"CUS_PROMPT"/email";
#else
const char *filepath = "/usr/local/taos/email";
#endif // CUS_PROMPT
TdFilePtr pFile = taosOpenFile(filepath, TD_FILE_READ);
if (pFile == NULL) return false;
......
......@@ -26,10 +26,6 @@ ELSE ()
SET(LINK_WEBSOCKET "")
ENDIF ()
IF (CUS_NAME OR CUS_PROMPT OR CUS_EMAIL)
ADD_DEFINITIONS(-I${CMAKE_CURRENT_SOURCE_DIR}/../../../enterprise/packaging)
ENDIF (CUS_NAME OR CUS_PROMPT OR CUS_EMAIL)
IF (TD_LINUX AND TD_ALPINE)
SET(LINK_ARGP "/usr/lib/libargp.a")
ELSE ()
......
......@@ -59,8 +59,8 @@ char cusEmail[] = "<support@taosdata.com>";
#define SHELL_VERSION "Print program version."
#ifdef WEBSOCKET
#define SHELL_DSN "The dsn to use when connecting to cloud server."
#define SHELL_REST "Use restful mode when connecting."
#define SHELL_DSN "Use dsn to connect to the cloud server or to a remote server which provides WebSocket connection."
#define SHELL_REST "Use RESTful mode when connecting."
#define SHELL_TIMEOUT "Set the timeout for websocket query in seconds, default is 30."
#endif
......@@ -93,7 +93,11 @@ void shellPrintHelp() {
#endif
printf("%s%s%s%s\r\n", indent, "-w,", indent, SHELL_WIDTH);
printf("%s%s%s%s\r\n", indent, "-V,", indent, SHELL_VERSION);
printf("\r\n\r\nReport bugs to %s.\r\n", cusEmail);
#ifdef CUS_EMAIL
printf("\r\n\r\nReport bugs to %s.\r\n", CUS_EMAIL);
#else
printf("\r\n\r\nReport bugs to %s.\r\n", "support@taosdata.com");
#endif
}
#ifdef LINUX
......@@ -105,7 +109,11 @@ void shellPrintHelp() {
#endif
const char *argp_program_version = version;
const char *argp_program_bug_address = cusEmail;
#ifdef CUS_EMAIL
const char *argp_program_bug_address = CUS_EMAIL;
#else
const char *argp_program_bug_address = "support@taosdata.com";
#endif
static struct argp_option shellOptions[] = {
{"host", 'h', "HOST", 0, SHELL_HOST},
......@@ -411,10 +419,19 @@ int32_t shellParseArgs(int32_t argc, char *argv[]) {
shell.info.clientVersion =
"Welcome to the %s Command Line Interface, Client Version:%s\r\n"
"Copyright (c) 2022 by %s, all rights reserved.\r\n\r\n";
strcpy(shell.info.cusName, cusName);
sprintf(shell.info.promptHeader, "%s> ", cusPrompt);
#ifdef CUS_NAME
strcpy(shell.info.cusName, CUS_NAME);
#else
strcpy(shell.info.cusName, "TDengine");
#endif
char promptContinueFormat[32] = {0};
sprintf(promptContinueFormat, "%%%zus> ", strlen(cusPrompt));
#ifdef CUS_PROMPT
sprintf(shell.info.promptHeader, "%s> ", CUS_PROMPT);
sprintf(promptContinueFormat, "%%%zus> ", strlen(CUS_PROMPT));
#else
sprintf(shell.info.promptHeader, "taos> ");
sprintf(promptContinueFormat, "%%%zus> ", strlen("taos"));
#endif
sprintf(shell.info.promptContinue, promptContinueFormat, " ");
shell.info.promptSize = strlen(shell.info.promptHeader);
#ifdef TD_ENTERPRISE
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册