xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge branch 'main' into refact/streamsm

This commit is contained in:
Haojun Liao 2023-10-18 10:29:09 +08:00
parent 64ed7e49a2 5115a1ac88
commit 2904cfd4d5
19 changed files with 243 additions and 224 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
include/common/tglobal.h
View File

@ -193,7 +193,8 @@ extern int64_t tsWalFsyncDataSizeLimit;
// internal // internal
extern int32_t tsTransPullupInterval; extern int32_t tsTransPullupInterval;
extern int32_t tsMqRebalanceInterval; extern int32_t tsMqRebalanceInterval;
extern int32_t tsStreamCheckpointTickInterval; extern int32_t tsStreamCheckpointInterval;
extern float tsSinkDataRate;
extern int32_t tsStreamNodeCheckInterval; extern int32_t tsStreamNodeCheckInterval;
extern int32_t tsTtlUnit; extern int32_t tsTtlUnit;
extern int32_t tsTtlPushIntervalSec; extern int32_t tsTtlPushIntervalSec;
@ -202,9 +203,6 @@ extern int32_t tsTrimVDbIntervalSec;
extern int32_t tsGrantHBInterval; extern int32_t tsGrantHBInterval;
extern int32_t tsUptimeInterval; extern int32_t tsUptimeInterval;
extern int32_t tsRpcRetryLimit;
extern int32_t tsRpcRetryInterval;
extern bool tsDisableStream; extern bool tsDisableStream;
extern int64_t tsStreamBufferSize; extern int64_t tsStreamBufferSize;
extern bool tsFilterScalarMode; extern bool tsFilterScalarMode;

1
include/libs/stream/tstream.h
View File

@ -716,7 +716,6 @@ int32_t streamTaskScanHistoryDataComplete(SStreamTask* pTask);
int32_t streamStartScanHistoryAsync(SStreamTask* pTask, int8_t igUntreated); int32_t streamStartScanHistoryAsync(SStreamTask* pTask, int8_t igUntreated);
bool streamHistoryTaskSetVerRangeStep2(SStreamTask* pTask, int64_t latestVer); bool streamHistoryTaskSetVerRangeStep2(SStreamTask* pTask, int64_t latestVer);
int32_t streamQueueGetNumOfItems(const SStreamQueue* pQueue); int32_t streamQueueGetNumOfItems(const SStreamQueue* pQueue);
int32_t streamQueueGetAvailableSpace(const SStreamQueue* pQueue, int32_t* availNum, double* availSize);
// common // common
int32_t streamRestoreParam(SStreamTask* pTask); int32_t streamRestoreParam(SStreamTask* pTask);

2
include/util/tconfig.h
View File

@ -65,7 +65,6 @@ typedef struct SConfigItem {
union { union {
bool bval; bool bval;
float fval; float fval;
double dval;
int32_t i32; int32_t i32;
int64_t i64; int64_t i64;
char *str; char *str;
@ -104,7 +103,6 @@ int32_t cfgAddBool(SConfig *pCfg, const char *name, bool defaultVal, int8_t scop
int32_t cfgAddInt32(SConfig *pCfg, const char *name, int32_t defaultVal, int64_t minval, int64_t maxval, int8_t scope); int32_t cfgAddInt32(SConfig *pCfg, const char *name, int32_t defaultVal, int64_t minval, int64_t maxval, int8_t scope);
int32_t cfgAddInt64(SConfig *pCfg, const char *name, int64_t defaultVal, int64_t minval, int64_t maxval, int8_t scope); int32_t cfgAddInt64(SConfig *pCfg, const char *name, int64_t defaultVal, int64_t minval, int64_t maxval, int8_t scope);
int32_t cfgAddFloat(SConfig *pCfg, const char *name, float defaultVal, float minval, float maxval, int8_t scope); int32_t cfgAddFloat(SConfig *pCfg, const char *name, float defaultVal, float minval, float maxval, int8_t scope);
int32_t cfgAddDouble(SConfig *pCfg, const char *name, double defaultVal, double minval, double maxval, int8_t scope);
int32_t cfgAddString(SConfig *pCfg, const char *name, const char *defaultVal, int8_t scope); int32_t cfgAddString(SConfig *pCfg, const char *name, const char *defaultVal, int8_t scope);
int32_t cfgAddDir(SConfig *pCfg, const char *name, const char *defaultVal, int8_t scope); int32_t cfgAddDir(SConfig *pCfg, const char *name, const char *defaultVal, int8_t scope);
int32_t cfgAddLocale(SConfig *pCfg, const char *name, const char *defaultVal, int8_t scope); int32_t cfgAddLocale(SConfig *pCfg, const char *name, const char *defaultVal, int8_t scope);

193
packaging/tools/install.sh
View File

@ -373,42 +373,52 @@ function add_newHostname_to_hosts() {
return return
fi fi
done done
${csudo}echo "127.0.0.1 $1" >>/etc/hosts || :
if grep -q "127.0.0.1 $1" /etc/hosts; then
return
else
${csudo}echo "127.0.0.1 $1" >>/etc/hosts
fi
} }
function set_hostname() { function set_hostname() {
echo -e -n "${GREEN}Please enter one hostname(must not be 'localhost')${NC}:" echo -e -n "${GREEN}Enter the public accessible IP address or fully qualified domain name TDengine will expose to users or applications (must not be 'localhost') :${NC}"
read newHostname read newHostname
while true; do while true; do
if [[ ! -z "$newHostname" && "$newHostname" != "localhost" ]]; then if [[ ! -z "$newHostname" && "$newHostname" != "localhost" ]]; then
break break
else else
read -p "Please enter one hostname(must not be 'localhost'):" newHostname echo -e -n "${GREEN}Enter the public accessible IP address or fully qualified domain name TDengine will expose to users or applications (must not be 'localhost') :${NC}"
read newHostname
fi fi
done done
${csudo}hostname $newHostname || : # ${csudo}hostname $newHostname || :
retval=$(echo $?) # retval=$(echo $?)
if [[ $retval != 0 ]]; then # if [[ $retval != 0 ]]; then
echo # echo
echo "set hostname fail!" # echo "set hostname fail!"
return # return
fi # fi
#ubuntu/centos /etc/hostname # #ubuntu/centos /etc/hostname
if [[ -e /etc/hostname ]]; then # if [[ -e /etc/hostname ]]; then
${csudo}echo $newHostname >/etc/hostname || : # ${csudo}echo $newHostname >/etc/hostname || :
fi # fi
#debian: #HOSTNAME=yourname # #debian: #HOSTNAME=yourname
if [[ -e /etc/sysconfig/network ]]; then # if [[ -e /etc/sysconfig/network ]]; then
${csudo}sed -i -r "s/#*\s*(HOSTNAME=\s*).*/\1$newHostname/" /etc/sysconfig/network || : # ${csudo}sed -i -r "s/#*\s*(HOSTNAME=\s*).*/\1$newHostname/" /etc/sysconfig/network || :
fi # fi
if [ -f ${cfg_install_dir}/${configFile2} ]; then
${csudo}sed -i -r "s/#*\s*(fqdn\s*).*/\1$newHostname/" ${cfg_install_dir}/${configFile2} ${csudo}sed -i -r "s/#*\s*(fqdn\s*).*/\1$newHostname/" ${cfg_install_dir}/${configFile2}
else
${csudo}sed -i -r "s/#*\s*(fqdn\s*).*/\1$newHostname/" ${script_dir}/cfg/${configFile2}
fi
serverFqdn=$newHostname serverFqdn=$newHostname
if [[ -e /etc/hosts ]]; then if [[ -e /etc/hosts ]] && [[ ! $newHostname =~ ^[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}$ ]]; then
add_newHostname_to_hosts $newHostname add_newHostname_to_hosts $newHostname
fi fi
} }
@ -439,7 +449,12 @@ function set_ipAsFqdn() {
echo -e -n "${GREEN}Unable to get local ip, use 127.0.0.1${NC}" echo -e -n "${GREEN}Unable to get local ip, use 127.0.0.1${NC}"
localFqdn="127.0.0.1" localFqdn="127.0.0.1"
# Write the local FQDN to configuration file # Write the local FQDN to configuration file
if [ -f ${cfg_install_dir}/${configFile2} ]; then
${csudo}sed -i -r "s/#*\s*(fqdn\s*).*/\1$localFqdn/" ${cfg_install_dir}/${configFile2} ${csudo}sed -i -r "s/#*\s*(fqdn\s*).*/\1$localFqdn/" ${cfg_install_dir}/${configFile2}
else
${csudo}sed -i -r "s/#*\s*(fqdn\s*).*/\1$localFqdn/" ${script_dir}/cfg/${configFile2}
fi
serverFqdn=$localFqdn serverFqdn=$localFqdn
echo echo
return return
@ -461,7 +476,11 @@ function set_ipAsFqdn() {
read -p "Please choose an IP from local IP list:" localFqdn read -p "Please choose an IP from local IP list:" localFqdn
else else
# Write the local FQDN to configuration file # Write the local FQDN to configuration file
if [ -f ${cfg_install_dir}/${configFile2} ]; then
${csudo}sed -i -r "s/#*\s*(fqdn\s*).*/\1$localFqdn/" ${cfg_install_dir}/${configFile2} ${csudo}sed -i -r "s/#*\s*(fqdn\s*).*/\1$localFqdn/" ${cfg_install_dir}/${configFile2}
else
${csudo}sed -i -r "s/#*\s*(fqdn\s*).*/\1$localFqdn/" ${script_dir}/cfg/${configFile2}
fi
serverFqdn=$localFqdn serverFqdn=$localFqdn
break break
fi fi
@ -476,37 +495,13 @@ function local_fqdn_check() {
echo echo
echo -e -n "System hostname is: ${GREEN}$serverFqdn${NC}" echo -e -n "System hostname is: ${GREEN}$serverFqdn${NC}"
echo echo
if [[ "$serverFqdn" == "" ]] || [[ "$serverFqdn" == "localhost" ]]; then
echo -e -n "${GREEN}It is strongly recommended to configure a hostname for this machine ${NC}"
echo
while true; do
read -r -p "Set hostname now? [Y/n] " input
if [ ! -n "$input" ]; then
set_hostname set_hostname
break
else
case $input in
[yY][eE][sS] | [yY])
set_hostname
break
;;
[nN][oO] | [nN])
set_ipAsFqdn
break
;;
*)
echo "Invalid input..."
;;
esac
fi
done
fi
} }
function install_adapter_config() { function install_adapter_config() {
if [ -f ${script_dir}/cfg/${adapterName}.toml ]; then
${csudo}sed -i -r "s/localhost/${serverFqdn}/g" ${script_dir}/cfg/${adapterName}.toml
fi
if [ ! -f "${cfg_install_dir}/${adapterName}.toml" ]; then if [ ! -f "${cfg_install_dir}/${adapterName}.toml" ]; then
${csudo}mkdir -p ${cfg_install_dir} ${csudo}mkdir -p ${cfg_install_dir}
[ -f ${script_dir}/cfg/${adapterName}.toml ] && ${csudo}cp ${script_dir}/cfg/${adapterName}.toml ${cfg_install_dir} [ -f ${script_dir}/cfg/${adapterName}.toml ] && ${csudo}cp ${script_dir}/cfg/${adapterName}.toml ${cfg_install_dir}
@ -525,11 +520,19 @@ function install_adapter_config() {
function install_config() { function install_config() {
local_fqdn_check
if [ ! -f "${cfg_install_dir}/${configFile2}" ]; then if [ ! -f "${cfg_install_dir}/${configFile2}" ]; then
${csudo}mkdir -p ${cfg_install_dir} ${csudo}mkdir -p ${cfg_install_dir}
[ -f ${script_dir}/cfg/${configFile2} ] && ${csudo}cp ${script_dir}/cfg/${configFile2} ${cfg_install_dir} if [ -f ${script_dir}/cfg/${configFile2} ]; then
${csudo} echo "monitor 1" >> ${script_dir}/cfg/${configFile2}
${csudo} echo "monitorFQDN ${serverFqdn}" >> ${script_dir}/cfg/${configFile2}
${csudo}cp ${script_dir}/cfg/${configFile2} ${cfg_install_dir}
fi
${csudo}chmod 644 ${cfg_install_dir}/* ${csudo}chmod 644 ${cfg_install_dir}/*
else else
${csudo} echo "monitor 1" >> ${script_dir}/cfg/${configFile2}
${csudo} echo "monitorFQDN ${serverFqdn}" >> ${script_dir}/cfg/${configFile2}
${csudo}cp -f ${script_dir}/cfg/${configFile2} ${cfg_install_dir}/${configFile2}.new ${csudo}cp -f ${script_dir}/cfg/${configFile2} ${cfg_install_dir}/${configFile2}.new
fi fi
@ -537,15 +540,15 @@ function install_config() {
[ ! -z $1 ] && return 0 || : # only install client [ ! -z $1 ] && return 0 || : # only install client
if ((${update_flag} == 1)); then
return 0
fi
if [ "$interactiveFqdn" == "no" ]; then
return 0
fi
local_fqdn_check # if ((${update_flag} == 1)); then
# return 0
# fi
# if [ "$interactiveFqdn" == "no" ]; then
# return 0
# fi
echo echo
echo -e -n "${GREEN}Enter FQDN:port (like h1.${emailName2}:6030) of an existing ${productName2} cluster node to join${NC}" echo -e -n "${GREEN}Enter FQDN:port (like h1.${emailName2}:6030) of an existing ${productName2} cluster node to join${NC}"
@ -554,7 +557,11 @@ function install_config() {
read firstEp read firstEp
while true; do while true; do
if [ ! -z "$firstEp" ]; then if [ ! -z "$firstEp" ]; then
if [ -f ${cfg_install_dir}/${configFile2} ]; then
${csudo}sed -i -r "s/#*\s*(firstEp\s*).*/\1$firstEp/" ${cfg_install_dir}/${configFile2} ${csudo}sed -i -r "s/#*\s*(firstEp\s*).*/\1$firstEp/" ${cfg_install_dir}/${configFile2}
else
${csudo}sed -i -r "s/#*\s*(firstEp\s*).*/\1$firstEp/" ${script_dir}/cfg/${configFile2}
fi
break break
else else
break break
@ -629,7 +636,7 @@ function install_taosx() {
if [ -f "${script_dir}/taosx/install_taosx.sh" ]; then if [ -f "${script_dir}/taosx/install_taosx.sh" ]; then
cd ${script_dir}/taosx cd ${script_dir}/taosx
chmod a+x install_taosx.sh chmod a+x install_taosx.sh
bash install_taosx.sh bash install_taosx.sh -e $serverFqdn
fi fi
} }
@ -713,29 +720,6 @@ function clean_service_on_systemd() {
fi fi
${csudo}systemctl disable tarbitratord &>/dev/null || echo &>/dev/null ${csudo}systemctl disable tarbitratord &>/dev/null || echo &>/dev/null
${csudo}rm -f ${tarbitratord_service_config} ${csudo}rm -f ${tarbitratord_service_config}
if [ "$verMode" == "cluster" ] && [ "$clientName" != "$clientName2" ]; then
x_service_config="${service_config_dir}/${xName2}.service"
if [ -e "$x_service_config" ]; then
if systemctl is-active --quiet ${xName2}; then
echo "${productName2} ${xName2} is running, stopping it..."
${csudo}systemctl stop ${xName2} &>/dev/null || echo &>/dev/null
fi
${csudo}systemctl disable ${xName2} &>/dev/null || echo &>/dev/null
${csudo}rm -f ${x_service_config}
fi
explorer_service_config="${service_config_dir}/${explorerName2}.service"
if [ -e "$explorer_service_config" ]; then
if systemctl is-active --quiet ${explorerName2}; then
echo "${productName2} ${explorerName2} is running, stopping it..."
${csudo}systemctl stop ${explorerName2} &>/dev/null || echo &>/dev/null
fi
${csudo}systemctl disable ${explorerName2} &>/dev/null || echo &>/dev/null
${csudo}rm -f ${explorer_service_config}
${csudo}rm -f /etc/${clientName2}/explorer.toml
fi
fi
} }
function install_service_on_systemd() { function install_service_on_systemd() {
@ -756,15 +740,27 @@ function install_service_on_systemd() {
${csudo}systemctl daemon-reload ${csudo}systemctl daemon-reload
${csudo}systemctl enable ${serverName2} ${csudo}systemctl enable ${serverName2}
${csudo}systemctl daemon-reload ${csudo}systemctl daemon-reload
} }
function install_adapter_service() { function install_adapter_service() {
if ((${service_mod} == 0)); then if ((${service_mod} == 0)); then
[ -f ${script_dir}/cfg/${adapterName}.service ] && [ -f ${script_dir}/cfg/${adapterName2}.service ] &&
${csudo}cp ${script_dir}/cfg/${adapterName}.service \ ${csudo}cp ${script_dir}/cfg/${adapterName2}.service \
${service_config_dir}/ || : ${service_config_dir}/ || :
${csudo}systemctl enable ${adapterName2}
${csudo}systemctl daemon-reload
fi
}
function install_keeper_service() {
if ((${service_mod} == 0)); then
[ -f ${script_dir}/cfg/${clientName2}keeper.service ] &&
${csudo}cp ${script_dir}/cfg/${clientName2}keeper.service \
${service_config_dir}/ || :
${csudo}systemctl enable ${clientName2}keeper
${csudo}systemctl daemon-reload ${csudo}systemctl daemon-reload
fi fi
} }
@ -883,7 +879,7 @@ function updateProduct() {
tar -zxf ${tarName} tar -zxf ${tarName}
install_jemalloc install_jemalloc
echo -e "${GREEN}Start to update ${productName2}...${NC}" echo "Start to update ${productName2}..."
# Stop the service if running # Stop the service if running
if ps aux | grep -v grep | grep ${serverName2} &>/dev/null; then if ps aux | grep -v grep | grep ${serverName2} &>/dev/null; then
if ((${service_mod} == 0)); then if ((${service_mod} == 0)); then
@ -901,6 +897,7 @@ function updateProduct() {
install_log install_log
install_header install_header
install_lib install_lib
install_config
if [ "$verMode" == "cluster" ]; then if [ "$verMode" == "cluster" ]; then
install_connector install_connector
@ -913,8 +910,8 @@ function updateProduct() {
install_bin install_bin
install_service install_service
install_adapter_service install_adapter_service
install_config
install_adapter_config install_adapter_config
install_keeper_service
openresty_work=false openresty_work=false
@ -957,19 +954,18 @@ function updateProduct() {
# fi # fi
echo echo
echo -e "\033[44;32;1m${productName2} is updated successfully!${NC}" echo "${productName2} is updated successfully!"
echo echo
if [ "$verMode" == "cluster" ];then if [ "$verMode" == "cluster" ];then
echo -e "\033[44;32;1mTo start all the components \t: sudo ./start-all.sh${NC}" echo -e "\033[44;32;1mTo start all the components : sudo ./start-all.sh${NC}"
fi fi
echo -e "\033[44;32;1mTo access ${productName2} \t\t: ${clientName2} -h $serverFqdn${NC}" echo -e "\033[44;32;1mTo access ${productName2} : ${clientName2} -h $serverFqdn${NC}"
if [ "$verMode" == "cluster" ];then if [ "$verMode" == "cluster" ];then
echo -e "\033[44;32;1mTo access the management system \t: http://$serverFqdn:6060${NC}" echo -e "\033[44;32;1mTo access the management system : http://$serverFqdn:6060${NC}"
echo -e "\033[44;32;1mTo read the user manual \t: http://$serverFqdn:6060/docs${NC}" echo -e "\033[44;32;1mTo read the user manual : http://$serverFqdn:6060/docs${NC}"
fi fi
else else
install_bin install_bin
install_config
echo echo
echo -e "\033[44;32;1m${productName2} client is updated successfully!${NC}" echo -e "\033[44;32;1m${productName2} client is updated successfully!${NC}"
@ -987,7 +983,7 @@ function installProduct() {
fi fi
tar -zxf ${tarName} tar -zxf ${tarName}
echo -e "${GREEN}Start to install ${productName2}...${NC}" echo "Start to install ${productName2}..."
install_main_path install_main_path
@ -1001,6 +997,7 @@ function installProduct() {
install_jemalloc install_jemalloc
#install_avro lib #install_avro lib
#install_avro lib64 #install_avro lib64
install_config
if [ "$verMode" == "cluster" ]; then if [ "$verMode" == "cluster" ]; then
install_connector install_connector
@ -1014,10 +1011,10 @@ function installProduct() {
install_service install_service
install_adapter_service install_adapter_service
install_adapter_config install_adapter_config
install_keeper_service
openresty_work=false openresty_work=false
install_config
# Ask if to start the service # Ask if to start the service
echo echo
@ -1068,20 +1065,20 @@ function installProduct() {
# echo # echo
# fi # fi
echo echo
echo -e "\033[44;32;1m${productName2} is installed successfully!${NC}" echo "${productName2} is installed successfully!"
echo echo
if [ "$verMode" == "cluster" ];then if [ "$verMode" == "cluster" ];then
echo -e "\033[44;32;1mTo start all the components \t: sudo ./start-all.sh${NC}" echo -e "\033[44;32;1mTo start all the components : sudo ./start-all.sh${NC}"
fi fi
echo -e "\033[44;32;1mTo access ${productName2} \t\t: ${clientName2} -h $serverFqdn${NC}" echo -e "\033[44;32;1mTo access ${productName2} : ${clientName2} -h $serverFqdn${NC}"
if [ "$verMode" == "cluster" ];then if [ "$verMode" == "cluster" ];then
echo -e "\033[44;32;1mTo access the management system \t: http://$serverFqdn:6060${NC}" echo -e "\033[44;32;1mTo access the management system : http://$serverFqdn:6060${NC}"
echo -e "\033[44;32;1mTo read the user manual \t: http://$serverFqdn:6060/docs${NC}" echo -e "\033[44;32;1mTo read the user manual : http://$serverFqdn:6060/docs${NC}"
fi fi
echo echo
else # Only install client else # Only install client
install_bin install_bin
install_config
echo echo
echo -e "\033[44;32;1m${productName2} client is installed successfully!${NC}" echo -e "\033[44;32;1m${productName2} client is installed successfully!${NC}"
fi fi
@ -1118,3 +1115,5 @@ elif [ "$verType" == "client" ]; then
else else
echo "please input correct verType" echo "please input correct verType"
fi fi

1
packaging/tools/makepkg.sh
View File

@ -374,6 +374,7 @@ if [ "$verMode" == "cluster" ]; then
cp -r ${top_dir}/../enterprise/src/plugins/taosx/release/taosx ${install_dir} cp -r ${top_dir}/../enterprise/src/plugins/taosx/release/taosx ${install_dir}
cp ${top_dir}/../enterprise/packaging/install_taosx.sh ${install_dir}/taosx cp ${top_dir}/../enterprise/packaging/install_taosx.sh ${install_dir}/taosx
cp ${top_dir}/../enterprise/src/plugins/taosx/packaging/uninstall.sh ${install_dir}/taosx cp ${top_dir}/../enterprise/src/plugins/taosx/packaging/uninstall.sh ${install_dir}/taosx
sed -i 's/target=\"\"/target=\"taosx\"/g' ${install_dir}/taosx/uninstall.sh
else else
echo "taox package not found" echo "taox package not found"
exit 1 exit 1

104
packaging/tools/remove.sh
View File

@ -113,8 +113,10 @@ function clean_bin() {
# Remove link # Remove link
${csudo}rm -f ${bin_link_dir}/${clientName} || : ${csudo}rm -f ${bin_link_dir}/${clientName} || :
${csudo}rm -f ${bin_link_dir}/${serverName} || : ${csudo}rm -f ${bin_link_dir}/${serverName} || :
echo "${serverName} is removed successfully"
${csudo}rm -f ${bin_link_dir}/udfd || : ${csudo}rm -f ${bin_link_dir}/udfd || :
${csudo}rm -f ${bin_link_dir}/${adapterName2} || : ${csudo}rm -f ${bin_link_dir}/${adapterName2} || :
echo "${adapterName2} is removed successfully"
${csudo}rm -f ${bin_link_dir}/${benchmarkName2} || : ${csudo}rm -f ${bin_link_dir}/${benchmarkName2} || :
${csudo}rm -f ${bin_link_dir}/${demoName2} || : ${csudo}rm -f ${bin_link_dir}/${demoName2} || :
${csudo}rm -f ${bin_link_dir}/${dumpName2} || : ${csudo}rm -f ${bin_link_dir}/${dumpName2} || :
@ -175,7 +177,7 @@ function clean_log() {
function clean_service_on_systemd() { function clean_service_on_systemd() {
taosd_service_config="${service_config_dir}/${taos_service_name}.service" taosd_service_config="${service_config_dir}/${taos_service_name}.service"
if systemctl is-active --quiet ${taos_service_name}; then if systemctl is-active --quiet ${taos_service_name}; then
echo "${productName2} ${serverName2} is running, stopping it..." echo "${taos_service_name} is running, stopping it..."
${csudo}systemctl stop ${taos_service_name} &>/dev/null || echo &>/dev/null ${csudo}systemctl stop ${taos_service_name} &>/dev/null || echo &>/dev/null
fi fi
${csudo}systemctl disable ${taos_service_name} &>/dev/null || echo &>/dev/null ${csudo}systemctl disable ${taos_service_name} &>/dev/null || echo &>/dev/null
@ -183,7 +185,7 @@ function clean_service_on_systemd() {
taosadapter_service_config="${service_config_dir}/${clientName2}adapter.service" taosadapter_service_config="${service_config_dir}/${clientName2}adapter.service"
if systemctl is-active --quiet ${taosadapter_service_name}; then if systemctl is-active --quiet ${taosadapter_service_name}; then
echo "${productName2} ${clientName2}Adapter is running, stopping it..." echo "${clientName2}Adapter is running, stopping it..."
${csudo}systemctl stop ${taosadapter_service_name} &>/dev/null || echo &>/dev/null ${csudo}systemctl stop ${taosadapter_service_name} &>/dev/null || echo &>/dev/null
fi fi
${csudo}systemctl disable ${taosadapter_service_name} &>/dev/null || echo &>/dev/null ${csudo}systemctl disable ${taosadapter_service_name} &>/dev/null || echo &>/dev/null
@ -196,33 +198,11 @@ function clean_service_on_systemd() {
fi fi
${csudo}systemctl disable ${tarbitrator_service_name} &>/dev/null || echo &>/dev/null ${csudo}systemctl disable ${tarbitrator_service_name} &>/dev/null || echo &>/dev/null
if [ "$verMode" == "cluster" ] && [ "$clientName" != "$clientName2" ]; then
x_service_config="${service_config_dir}/${xName2}.service"
if [ -e "$x_service_config" ]; then
if systemctl is-active --quiet ${xName2}; then
echo "${productName2} ${xName2} is running, stopping it..."
${csudo}systemctl stop ${xName2} &>/dev/null || echo &>/dev/null
fi
${csudo}systemctl disable ${xName2} &>/dev/null || echo &>/dev/null
${csudo}rm -f ${x_service_config}
fi
explorer_service_config="${service_config_dir}/${explorerName2}.service"
if [ -e "$explorer_service_config" ]; then
if systemctl is-active --quiet ${explorerName2}; then
echo "${productName2} ${explorerName2} is running, stopping it..."
${csudo}systemctl stop ${explorerName2} &>/dev/null || echo &>/dev/null
fi
${csudo}systemctl disable ${explorerName2} &>/dev/null || echo &>/dev/null
${csudo}rm -f ${explorer_service_config}
${csudo}rm -f /etc/${clientName2}/explorer.toml
fi
fi
} }
function clean_service_on_sysvinit() { function clean_service_on_sysvinit() {
if ps aux | grep -v grep | grep ${serverName} &>/dev/null; then if ps aux | grep -v grep | grep ${serverName} &>/dev/null; then
echo "${productName2} ${serverName2} is running, stopping it..." echo "${serverName2} is running, stopping it..."
${csudo}service ${serverName} stop || : ${csudo}service ${serverName} stop || :
fi fi
@ -287,10 +267,7 @@ function clean_service() {
function uninstall_taosx() { function uninstall_taosx() {
if [ -f /usr/local/taosx/uninstall.sh ]; then if [ -f /usr/local/taosx/uninstall.sh ]; then
cd /usr/local/taosx cd /usr/local/taosx
bash uninstall.sh > /dev/null bash uninstall.sh
echo -e "${GREEN}${xName2} is removed successfully!${NC}"
echo -e "${GREEN}${explorerName2} is removed successfully!${NC}"
fi fi
} }
@ -332,8 +309,75 @@ if [ "$osType" = "Darwin" ]; then
${csudo}rm -rf /Applications/TDengine.app ${csudo}rm -rf /Applications/TDengine.app
fi fi
echo -e "${GREEN}${productName2} is removed successfully!${NC}" _kill_service_of() {
_service=$1
pid=$(ps -ef | grep "$_service" | grep -v "grep" | awk '{print $2}')
if [ -n "$pid" ]; then
${csudo}kill -9 $pid || :
fi
}
_clean_service_on_systemd_of() {
_service=$1
_service_config="${service_config_dir}/${_service}.service"
if systemctl is-active --quiet ${_service}; then
echo "taoskeeper is running, stopping it..."
${csudo}systemctl stop ${_service} &>/dev/null || echo &>/dev/null
fi
${csudo}systemctl disable ${_service} &>/dev/null || echo &>/dev/null
${csudo}rm -f ${_service_config}
}
_clean_service_on_sysvinit_of() {
_service=$1
if pidof ${_service} &>/dev/null; then
echo "${_service} is running, stopping it..."
${csudo}service ${_service} stop || :
fi
if ((${initd_mod} == 1)); then
if [ -e ${service_config_dir}/${_service} ]; then
${csudo}chkconfig --del ${_service} || :
fi
elif ((${initd_mod} == 2)); then
if [ -e ${service_config_dir}/${_service} ]; then
${csudo}insserv -r ${_service} || :
fi
elif ((${initd_mod} == 3)); then
if [ -e ${service_config_dir}/${_service} ]; then
${csudo}update-rc.d -f ${_service} remove || :
fi
fi
${csudo}rm -f ${service_config_dir}/${_service} || :
if $(which init &>/dev/null); then
${csudo}init q || :
fi
}
_clean_service_of() {
_service=$1
if ((${service_mod} == 0)); then
_clean_service_on_systemd_of $_service
elif ((${service_mod} == 1)); then
_clean_service_on_sysvinit_of $_service
else
_kill_service_of $_service
fi
}
remove_taoskeeper() {
# remove taoskeeper bin
_clean_service_of taoskeeper
[ -e "${bin_link_dir}/taoskeeper" ] && ${csudo}rm -rf ${bin_link_dir}/taoskeeper
[ -e "${cfg_link_dir}/metrics.toml" ] || ${csudo}rm -rf ${cfg_link_dir}/metrics.toml
echo "taosKeeper is removed successfully!"
}
remove_taoskeeper
if [ "$verMode" == "cluster" ]; then if [ "$verMode" == "cluster" ]; then
uninstall_taosx uninstall_taosx
fi fi
echo
echo "${productName2} is removed successfully!"
echo echo

54
source/common/src/tglobal.c
View File

@ -244,7 +244,8 @@ int32_t tsTtlBatchDropNum = 10000; // number of tables dropped per batch
// internal // internal
int32_t tsTransPullupInterval = 2; int32_t tsTransPullupInterval = 2;
int32_t tsMqRebalanceInterval = 2; int32_t tsMqRebalanceInterval = 2;
int32_t tsStreamCheckpointTickInterval = 300; int32_t tsStreamCheckpointInterval = 300;
float tsSinkDataRate = 2.0;
int32_t tsStreamNodeCheckInterval = 30; int32_t tsStreamNodeCheckInterval = 30;
int32_t tsTtlUnit = 86400; int32_t tsTtlUnit = 86400;
int32_t tsTtlPushIntervalSec = 10; int32_t tsTtlPushIntervalSec = 10;
@ -269,8 +270,6 @@ int8_t tsS3Enabled = false;
int32_t tsS3BlockSize = 4096; // number of tsdb pages int32_t tsS3BlockSize = 4096; // number of tsdb pages
int32_t tsS3BlockCacheSize = 16; // number of blocks int32_t tsS3BlockCacheSize = 16; // number of blocks
int32_t tsCheckpointInterval = 300;
#ifndef _STORAGE #ifndef _STORAGE
int32_t taosSetTfsCfg(SConfig *pCfg) { int32_t taosSetTfsCfg(SConfig *pCfg) {
SConfigItem *pItem = cfgGetItem(pCfg, "dataDir"); SConfigItem *pItem = cfgGetItem(pCfg, "dataDir");
@ -490,11 +489,11 @@ static int32_t taosAddSystemCfg(SConfig *pCfg) {
if (cfgAddBool(pCfg, "enableCoreFile", 1, CFG_SCOPE_BOTH) != 0) return -1; if (cfgAddBool(pCfg, "enableCoreFile", 1, CFG_SCOPE_BOTH) != 0) return -1;
if (cfgAddFloat(pCfg, "numOfCores", tsNumOfCores, 1, 100000, CFG_SCOPE_BOTH) != 0) return -1; if (cfgAddFloat(pCfg, "numOfCores", tsNumOfCores, 1, 100000, CFG_SCOPE_BOTH) != 0) return -1;
if (cfgAddBool(pCfg, "SSE42", tsSSE42Enable, CFG_SCOPE_BOTH) != 0) return -1; if (cfgAddBool(pCfg, "ssd42", tsSSE42Enable, CFG_SCOPE_BOTH) != 0) return -1;
if (cfgAddBool(pCfg, "AVX", tsAVXEnable, CFG_SCOPE_BOTH) != 0) return -1; if (cfgAddBool(pCfg, "avx", tsAVXEnable, CFG_SCOPE_BOTH) != 0) return -1;
if (cfgAddBool(pCfg, "AVX2", tsAVX2Enable, CFG_SCOPE_BOTH) != 0) return -1; if (cfgAddBool(pCfg, "avx2", tsAVX2Enable, CFG_SCOPE_BOTH) != 0) return -1;
if (cfgAddBool(pCfg, "FMA", tsFMAEnable, CFG_SCOPE_BOTH) != 0) return -1; if (cfgAddBool(pCfg, "fma", tsFMAEnable, CFG_SCOPE_BOTH) != 0) return -1;
if (cfgAddBool(pCfg, "SIMD-builtins", tsSIMDBuiltins, CFG_SCOPE_BOTH) != 0) return -1; if (cfgAddBool(pCfg, "simdEnable", tsSIMDBuiltins, CFG_SCOPE_BOTH) != 0) return -1;
if (cfgAddBool(pCfg, "tagFilterCache", tsTagFilterCache, CFG_SCOPE_BOTH) != 0) return -1; if (cfgAddBool(pCfg, "tagFilterCache", tsTagFilterCache, CFG_SCOPE_BOTH) != 0) return -1;
if (cfgAddInt64(pCfg, "openMax", tsOpenMax, 0, INT64_MAX, CFG_SCOPE_BOTH) != 0) return -1; if (cfgAddInt64(pCfg, "openMax", tsOpenMax, 0, INT64_MAX, CFG_SCOPE_BOTH) != 0) return -1;
@ -650,18 +649,18 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
if (cfgAddBool(pCfg, "disableStream", tsDisableStream, CFG_SCOPE_SERVER) != 0) return -1; if (cfgAddBool(pCfg, "disableStream", tsDisableStream, CFG_SCOPE_SERVER) != 0) return -1;
if (cfgAddInt64(pCfg, "streamBufferSize", tsStreamBufferSize, 0, INT64_MAX, CFG_SCOPE_SERVER) != 0) return -1; if (cfgAddInt64(pCfg, "streamBufferSize", tsStreamBufferSize, 0, INT64_MAX, CFG_SCOPE_SERVER) != 0) return -1;
if (cfgAddInt64(pCfg, "checkpointInterval", tsStreamCheckpointTickInterval, 60, 1200, CFG_SCOPE_SERVER) != 0) return -1; if (cfgAddInt64(pCfg, "checkpointInterval", tsStreamCheckpointInterval, 60, 1200, CFG_SCOPE_SERVER) != 0) return -1;
if (cfgAddFloat(pCfg, "streamSinkDataRate", tsSinkDataRate, 0.1, 5, CFG_SCOPE_SERVER) != 0) return -1;
if (cfgAddInt32(pCfg, "cacheLazyLoadThreshold", tsCacheLazyLoadThreshold, 0, 100000, CFG_SCOPE_SERVER) != 0) if (cfgAddInt32(pCfg, "cacheLazyLoadThreshold", tsCacheLazyLoadThreshold, 0, 100000, CFG_SCOPE_SERVER) != 0) return -1;
return -1;
if (cfgAddString(pCfg, "LossyColumns", tsLossyColumns, CFG_SCOPE_SERVER) != 0) return -1; if (cfgAddString(pCfg, "lossyColumns", tsLossyColumns, CFG_SCOPE_SERVER) != 0) return -1;
if (cfgAddFloat(pCfg, "FPrecision", tsFPrecision, 0.0f, 100000.0f, CFG_SCOPE_SERVER) != 0) return -1; if (cfgAddFloat(pCfg, "fPrecision", tsFPrecision, 0.0f, 100000.0f, CFG_SCOPE_SERVER) != 0) return -1;
if (cfgAddDouble(pCfg, "DPrecision", tsDPrecision, 0.0f, 1000000.0f, CFG_SCOPE_SERVER) != 0) return -1; if (cfgAddFloat(pCfg, "dPrecision", tsDPrecision, 0.0f, 1000000.0f, CFG_SCOPE_SERVER) != 0) return -1;
if (cfgAddInt32(pCfg, "MaxRange", tsMaxRange, 0, 65536, CFG_SCOPE_SERVER) != 0) return -1; if (cfgAddInt32(pCfg, "maxRange", tsMaxRange, 0, 65536, CFG_SCOPE_SERVER) != 0) return -1;
if (cfgAddInt32(pCfg, "CurRange", tsCurRange, 0, 65536, CFG_SCOPE_SERVER) != 0) return -1; if (cfgAddInt32(pCfg, "curRange", tsCurRange, 0, 65536, CFG_SCOPE_SERVER) != 0) return -1;
if (cfgAddBool(pCfg, "IfAdtFse", tsIfAdtFse, CFG_SCOPE_SERVER) != 0) return -1; if (cfgAddBool(pCfg, "ifAdtFse", tsIfAdtFse, CFG_SCOPE_SERVER) != 0) return -1;
if (cfgAddString(pCfg, "Compressor", tsCompressor, CFG_SCOPE_SERVER) != 0) return -1; if (cfgAddString(pCfg, "compressor", tsCompressor, CFG_SCOPE_SERVER) != 0) return -1;
if (cfgAddBool(pCfg, "filterScalarMode", tsFilterScalarMode, CFG_SCOPE_SERVER) != 0) return -1; if (cfgAddBool(pCfg, "filterScalarMode", tsFilterScalarMode, CFG_SCOPE_SERVER) != 0) return -1;
if (cfgAddInt32(pCfg, "maxStreamBackendCache", tsMaxStreamBackendCache, 16, 1024, CFG_SCOPE_SERVER) != 0) return -1; if (cfgAddInt32(pCfg, "maxStreamBackendCache", tsMaxStreamBackendCache, 16, 1024, CFG_SCOPE_SERVER) != 0) return -1;
@ -1022,7 +1021,7 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
tsNumOfSnodeWriteThreads = cfgGetItem(pCfg, "numOfSnodeUniqueThreads")->i32; tsNumOfSnodeWriteThreads = cfgGetItem(pCfg, "numOfSnodeUniqueThreads")->i32;
tsRpcQueueMemoryAllowed = cfgGetItem(pCfg, "rpcQueueMemoryAllowed")->i64; tsRpcQueueMemoryAllowed = cfgGetItem(pCfg, "rpcQueueMemoryAllowed")->i64;
tsSIMDBuiltins = (bool)cfgGetItem(pCfg, "SIMD-builtins")->bval; tsSIMDBuiltins = (bool)cfgGetItem(pCfg, "simdEnable")->bval;
tsTagFilterCache = (bool)cfgGetItem(pCfg, "tagFilterCache")->bval; tsTagFilterCache = (bool)cfgGetItem(pCfg, "tagFilterCache")->bval;
tsEnableMonitor = cfgGetItem(pCfg, "monitor")->bval; tsEnableMonitor = cfgGetItem(pCfg, "monitor")->bval;
@ -1080,17 +1079,18 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
tsCacheLazyLoadThreshold = cfgGetItem(pCfg, "cacheLazyLoadThreshold")->i32; tsCacheLazyLoadThreshold = cfgGetItem(pCfg, "cacheLazyLoadThreshold")->i32;
tstrncpy(tsLossyColumns, cfgGetItem(pCfg, "LossyColumns")->str, sizeof(tsLossyColumns)); tstrncpy(tsLossyColumns, cfgGetItem(pCfg, "lossyColumns")->str, sizeof(tsLossyColumns));
tsFPrecision = cfgGetItem(pCfg, "FPrecision")->fval; tsFPrecision = cfgGetItem(pCfg, "fPrecision")->fval;
tsDPrecision = cfgGetItem(pCfg, "DPrecision")->dval; tsDPrecision = cfgGetItem(pCfg, "dPrecision")->fval;
tsMaxRange = cfgGetItem(pCfg, "MaxRange")->i32; tsMaxRange = cfgGetItem(pCfg, "maxRange")->i32;
tsCurRange = cfgGetItem(pCfg, "CurRange")->i32; tsCurRange = cfgGetItem(pCfg, "curRange")->i32;
tsIfAdtFse = cfgGetItem(pCfg, "IfAdtFse")->bval; tsIfAdtFse = cfgGetItem(pCfg, "ifAdtFse")->bval;
tstrncpy(tsCompressor, cfgGetItem(pCfg, "Compressor")->str, sizeof(tsCompressor)); tstrncpy(tsCompressor, cfgGetItem(pCfg, "compressor")->str, sizeof(tsCompressor));
tsDisableStream = cfgGetItem(pCfg, "disableStream")->bval; tsDisableStream = cfgGetItem(pCfg, "disableStream")->bval;
tsStreamBufferSize = cfgGetItem(pCfg, "streamBufferSize")->i64; tsStreamBufferSize = cfgGetItem(pCfg, "streamBufferSize")->i64;
tsStreamCheckpointInterval = cfgGetItem(pCfg, "checkpointInterval")->i32;
tsSinkDataRate = cfgGetItem(pCfg, "streamSinkDataRate")->fval;
tsFilterScalarMode = cfgGetItem(pCfg, "filterScalarMode")->bval; tsFilterScalarMode = cfgGetItem(pCfg, "filterScalarMode")->bval;
tsMaxStreamBackendCache = cfgGetItem(pCfg, "maxStreamBackendCache")->i32; tsMaxStreamBackendCache = cfgGetItem(pCfg, "maxStreamBackendCache")->i32;

2
source/dnode/mnode/impl/src/mndMain.c
View File

@ -281,7 +281,7 @@ static void *mndThreadFp(void *param) {
mndCalMqRebalance(pMnode); mndCalMqRebalance(pMnode);
} }
if (sec % tsStreamCheckpointTickInterval == 0) { if (sec % tsStreamCheckpointInterval == 0) {
mndStreamCheckpointTick(pMnode, sec); mndStreamCheckpointTick(pMnode, sec);
} }

2
source/dnode/mnode/impl/src/mndStream.c
View File

@ -937,7 +937,7 @@ static int32_t mndBuildStreamCheckpointSourceReq2(void **pBuf, int32_t *pLen, in
} }
// static int32_t mndProcessStreamCheckpointTrans(SMnode *pMnode, SStreamObj *pStream, int64_t checkpointId) { // static int32_t mndProcessStreamCheckpointTrans(SMnode *pMnode, SStreamObj *pStream, int64_t checkpointId) {
// int64_t timestampMs = taosGetTimestampMs(); // int64_t timestampMs = taosGetTimestampMs();
// if (timestampMs - pStream->checkpointFreq < tsStreamCheckpointTickInterval * 1000) { // if (timestampMs - pStream->checkpointFreq < tsStreamCheckpointInterval * 1000) {
// return -1; // return -1;
// } // }

4
source/dnode/vnode/src/tq/tq.c
View File

@ -1196,7 +1196,11 @@ int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg) {
pStreamTask->status.keepTaskStatus = status; pStreamTask->status.keepTaskStatus = status;
pStreamTask->status.taskStatus = TASK_STATUS__HALT; pStreamTask->status.taskStatus = TASK_STATUS__HALT;
// wal scan not start yet, reset it to be the start position
nextProcessedVer = walReaderGetCurrentVer(pStreamTask->exec.pWalReader); nextProcessedVer = walReaderGetCurrentVer(pStreamTask->exec.pWalReader);
if (nextProcessedVer == -1) {
nextProcessedVer = pStreamTask->dataRange.range.maxVer + 1;
}
tqDebug("s-task:%s level:%d nextProcessedVer:%" PRId64 ", sched-status:%d is halt by fill-history task:%s", tqDebug("s-task:%s level:%d nextProcessedVer:%" PRId64 ", sched-status:%d is halt by fill-history task:%s",
pStreamTask->id.idStr, pStreamTask->info.taskLevel, nextProcessedVer, pStreamTask->status.schedStatus, pStreamTask->id.idStr, pStreamTask->info.taskLevel, nextProcessedVer, pStreamTask->status.schedStatus,

10
source/dnode/vnode/src/tq/tqStreamTaskSnap.c
View File

@ -198,8 +198,6 @@ int32_t streamTaskSnapWriterClose(SStreamTaskWriter* pWriter, int8_t rollback) {
taosWLockLatch(&pTq->pStreamMeta->lock); taosWLockLatch(&pTq->pStreamMeta->lock);
tqDebug("vgId:%d, vnode stream-task snapshot writer closed", TD_VID(pTq->pVnode)); tqDebug("vgId:%d, vnode stream-task snapshot writer closed", TD_VID(pTq->pVnode));
taosWLockLatch(&pTq->pStreamMeta->lock);
if (rollback) { if (rollback) {
tdbAbort(pTq->pStreamMeta->db, pTq->pStreamMeta->txn); tdbAbort(pTq->pStreamMeta->db, pTq->pStreamMeta->txn);
} else { } else {
@ -208,21 +206,13 @@ int32_t streamTaskSnapWriterClose(SStreamTaskWriter* pWriter, int8_t rollback) {
code = tdbPostCommit(pTq->pStreamMeta->db, pTq->pStreamMeta->txn); code = tdbPostCommit(pTq->pStreamMeta->db, pTq->pStreamMeta->txn);
if (code) goto _err; if (code) goto _err;
} }
if (tdbBegin(pTq->pStreamMeta->db, &pTq->pStreamMeta->txn, tdbDefaultMalloc, tdbDefaultFree, NULL, 0) < 0) {
code = -1;
goto _err;
}
taosWUnLockLatch(&pTq->pStreamMeta->lock);
if (tdbBegin(pTq->pStreamMeta->db, &pTq->pStreamMeta->txn, tdbDefaultMalloc, tdbDefaultFree, NULL, 0) < 0) { if (tdbBegin(pTq->pStreamMeta->db, &pTq->pStreamMeta->txn, tdbDefaultMalloc, tdbDefaultFree, NULL, 0) < 0) {
code = -1; code = -1;
taosMemoryFree(pWriter); taosMemoryFree(pWriter);
goto _err; goto _err;
} }
taosWUnLockLatch(&pTq->pStreamMeta->lock); taosWUnLockLatch(&pTq->pStreamMeta->lock);
taosMemoryFree(pWriter); taosMemoryFree(pWriter);
return code; return code;

7
source/libs/executor/src/streamtimewindowoperator.c
View File

@ -373,8 +373,11 @@ void clearGroupResInfo(SGroupResInfo* pGroupResInfo) {
if (pGroupResInfo->freeItem) { if (pGroupResInfo->freeItem) {
int32_t size = taosArrayGetSize(pGroupResInfo->pRows); int32_t size = taosArrayGetSize(pGroupResInfo->pRows);
for (int32_t i = pGroupResInfo->index; i < size; i++) { for (int32_t i = pGroupResInfo->index; i < size; i++) {
void* pVal = taosArrayGetP(pGroupResInfo->pRows, i); SRowBuffPos* pPos = taosArrayGetP(pGroupResInfo->pRows, i);
taosMemoryFree(pVal); if (!pPos->needFree && !pPos->pRowBuff) {
taosMemoryFreeClear(pPos->pKey);
taosMemoryFree(pPos);
}
} }
pGroupResInfo->freeItem = false; pGroupResInfo->freeItem = false;
} }

2
source/libs/stream/inc/streamInt.h
View File

@ -122,7 +122,7 @@ int32_t streamNotifyUpstreamContinue(SStreamTask* pTask);
int32_t streamTaskFillHistoryFinished(SStreamTask* pTask); int32_t streamTaskFillHistoryFinished(SStreamTask* pTask);
int32_t streamTransferStateToStreamTask(SStreamTask* pTask); int32_t streamTransferStateToStreamTask(SStreamTask* pTask);
int32_t streamTaskInitTokenBucket(STokenBucket* pBucket, int32_t numCap, int32_t numRate, int32_t quotaRate); int32_t streamTaskInitTokenBucket(STokenBucket* pBucket, int32_t numCap, int32_t numRate, float quotaRate);
STaskId streamTaskExtractKey(const SStreamTask* pTask); STaskId streamTaskExtractKey(const SStreamTask* pTask);
void streamTaskInitForLaunchHTask(SHistoryTaskInfo* pInfo); void streamTaskInitForLaunchHTask(SHistoryTaskInfo* pInfo);
void streamTaskSetRetryInfoForLaunch(SHistoryTaskInfo* pInfo); void streamTaskSetRetryInfoForLaunch(SHistoryTaskInfo* pInfo);

4
source/libs/stream/src/streamExec.c
View File

@ -309,7 +309,9 @@ int32_t streamDoTransferStateToStreamTask(SStreamTask* pTask) {
pStreamTask->id.idStr); pStreamTask->id.idStr);
} }
ASSERT(pStreamTask->hTaskInfo.id.taskId == pTask->id.taskId && pTask->status.appendTranstateBlock == true); ASSERT(((pStreamTask->status.taskStatus == TASK_STATUS__STOP) ||
(pStreamTask->hTaskInfo.id.taskId == pTask->id.taskId)) &&
pTask->status.appendTranstateBlock == true);
STimeWindow* pTimeWindow = &pStreamTask->dataRange.window; STimeWindow* pTimeWindow = &pStreamTask->dataRange.window;

19
source/libs/stream/src/streamQueue.c
View File

@ -18,7 +18,7 @@
#define MAX_STREAM_EXEC_BATCH_NUM 32 #define MAX_STREAM_EXEC_BATCH_NUM 32
#define MAX_SMOOTH_BURST_RATIO 5 // 5 sec #define MAX_SMOOTH_BURST_RATIO 5 // 5 sec
#define WAIT_FOR_DURATION 40 #define WAIT_FOR_DURATION 40
#define SINK_TASK_IDLE_DURATION 200 // 200 ms #define OUTPUT_QUEUE_FULL_WAIT_DURATION 500 // 500 ms
// todo refactor: // todo refactor:
// read data from input queue // read data from input queue
@ -119,14 +119,6 @@ int32_t streamQueueGetNumOfItems(const SStreamQueue* pQueue) {
return numOfItems1 + numOfItems2; return numOfItems1 + numOfItems2;
} }
int32_t streamQueueGetAvailableSpace(const SStreamQueue* pQueue, int32_t* availNum, double* availSize) {
int32_t num = streamQueueGetNumOfItems(pQueue);
*availNum = STREAM_TASK_QUEUE_CAPACITY - num;
*availSize = STREAM_TASK_QUEUE_CAPACITY_IN_SIZE - taosQueueMemorySize(pQueue->pQueue);
return 0;
}
// todo: fix it: data in Qall is not included here // todo: fix it: data in Qall is not included here
int32_t streamQueueGetItemSize(const SStreamQueue* pQueue) { int32_t streamQueueGetItemSize(const SStreamQueue* pQueue) {
return taosQueueMemorySize(pQueue->pQueue); return taosQueueMemorySize(pQueue->pQueue);
@ -362,9 +354,10 @@ int32_t streamTaskPutDataIntoOutputQ(SStreamTask* pTask, SStreamDataBlock* pBloc
int32_t total = streamQueueGetNumOfItems(pTask->outputq.queue); int32_t total = streamQueueGetNumOfItems(pTask->outputq.queue);
double size = SIZE_IN_MiB(taosQueueMemorySize(pQueue)); double size = SIZE_IN_MiB(taosQueueMemorySize(pQueue));
// let's wait for there are enough space to hold this result pBlock // let's wait for there are enough space to hold this result pBlock
stDebug("s-task:%s outputQ is full, wait for 500ms and retry, outputQ items:%d, size:%.2fMiB", pTask->id.idStr, stDebug("s-task:%s outputQ is full, wait for %dms and retry, outputQ items:%d, size:%.2fMiB", pTask->id.idStr,
total, size); OUTPUT_QUEUE_FULL_WAIT_DURATION, total, size);
taosMsleep(500);
taosMsleep(OUTPUT_QUEUE_FULL_WAIT_DURATION);
} }
int32_t code = taosWriteQitem(pQueue, pBlock); int32_t code = taosWriteQitem(pQueue, pBlock);
@ -381,7 +374,7 @@ int32_t streamTaskPutDataIntoOutputQ(SStreamTask* pTask, SStreamDataBlock* pBloc
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t streamTaskInitTokenBucket(STokenBucket* pBucket, int32_t numCap, int32_t numRate, int32_t quotaRate) { int32_t streamTaskInitTokenBucket(STokenBucket* pBucket, int32_t numCap, int32_t numRate, float quotaRate) {
if (numCap < 10 || numRate < 10 || pBucket == NULL) { if (numCap < 10 || numRate < 10 || pBucket == NULL) {
stError("failed to init sink task bucket, cap:%d, rate:%d", numCap, numRate); stError("failed to init sink task bucket, cap:%d, rate:%d", numCap, numRate);
return TSDB_CODE_INVALID_PARA; return TSDB_CODE_INVALID_PARA;

0
source/libs/stream/src/streamRecover.c → source/libs/stream/src/streamStart.c
View File

2
source/libs/stream/src/streamTask.c
View File

@ -437,7 +437,7 @@ int32_t streamTaskInit(SStreamTask* pTask, SStreamMeta* pMeta, SMsgCb* pMsgCb, i
// 2MiB per second for sink task // 2MiB per second for sink task
// 50 times sink operator per second // 50 times sink operator per second
streamTaskInitTokenBucket(pTask->outputInfo.pTokenBucket, 50, 50, 2); streamTaskInitTokenBucket(pTask->outputInfo.pTokenBucket, 50, 50, tsSinkDataRate);
TdThreadMutexAttr attr = {0}; TdThreadMutexAttr attr = {0};
int code = taosThreadMutexAttrInit(&attr); int code = taosThreadMutexAttrInit(&attr);

4
source/libs/stream/src/tstreamFileState.c
View File

@ -297,6 +297,10 @@ void popUsedBuffs(SStreamFileState* pFileState, SStreamSnapshot* pFlushList, uin
while ((pNode = tdListNext(&iter)) != NULL && i < max) { while ((pNode = tdListNext(&iter)) != NULL && i < max) {
SRowBuffPos* pPos = *(SRowBuffPos**)pNode->data; SRowBuffPos* pPos = *(SRowBuffPos**)pNode->data;
if (pPos->beUsed == used) { if (pPos->beUsed == used) {
if (used && !pPos->pRowBuff) {
ASSERT(pPos->needFree == true);
continue;
}
tdListAppend(pFlushList, &pPos); tdListAppend(pFlushList, &pPos);
pFileState->flushMark = TMAX(pFileState->flushMark, pFileState->getTs(pPos->pKey)); pFileState->flushMark = TMAX(pFileState->flushMark, pFileState->getTs(pPos->pKey));
pFileState->stateBuffRemoveByPosFn(pFileState, pPos); pFileState->stateBuffRemoveByPosFn(pFileState, pPos);

30
source/util/src/tconfig.c
View File

@ -416,16 +416,6 @@ int32_t cfgAddFloat(SConfig *pCfg, const char *name, float defaultVal, float min
return cfgAddItem(pCfg, &item, name); return cfgAddItem(pCfg, &item, name);
} }
int32_t cfgAddDouble(SConfig *pCfg, const char *name, double defaultVal, double minval, double maxval, int8_t scope) {
if (defaultVal < minval || defaultVal > maxval) {
terrno = TSDB_CODE_OUT_OF_RANGE;
return -1;
}
SConfigItem item = {.dtype = CFG_DTYPE_DOUBLE, .dval = defaultVal, .fmin = minval, .fmax = maxval, .scope = scope};
return cfgAddItem(pCfg, &item, name);
}
int32_t cfgAddString(SConfig *pCfg, const char *name, const char *defaultVal, int8_t scope) { int32_t cfgAddString(SConfig *pCfg, const char *name, const char *defaultVal, int8_t scope) {
SConfigItem item = {.dtype = CFG_DTYPE_STRING, .scope = scope}; SConfigItem item = {.dtype = CFG_DTYPE_STRING, .scope = scope};
item.str = taosStrdup(defaultVal); item.str = taosStrdup(defaultVal);
@ -611,8 +601,7 @@ void cfgDumpCfg(SConfig *pCfg, bool tsc, bool dump) {
switch (pItem->dtype) { switch (pItem->dtype) {
case CFG_DTYPE_BOOL: case CFG_DTYPE_BOOL:
if (dump) { if (dump) {
printf("%s %s %u", src, name, pItem->bval); printf("%s %s %u\n", src, name, pItem->bval);
printf("\n");
} else { } else {
uInfo("%s %s %u", src, name, pItem->bval); uInfo("%s %s %u", src, name, pItem->bval);
} }
@ -620,25 +609,22 @@ void cfgDumpCfg(SConfig *pCfg, bool tsc, bool dump) {
break; break;
case CFG_DTYPE_INT32: case CFG_DTYPE_INT32:
if (dump) { if (dump) {
printf("%s %s %d", src, name, pItem->i32); printf("%s %s %d\n", src, name, pItem->i32);
printf("\n");
} else { } else {
uInfo("%s %s %d", src, name, pItem->i32); uInfo("%s %s %d", src, name, pItem->i32);
} }
break; break;
case CFG_DTYPE_INT64: case CFG_DTYPE_INT64:
if (dump) { if (dump) {
printf("%s %s %" PRId64, src, name, pItem->i64); printf("%s %s %" PRId64"\n", src, name, pItem->i64);
printf("\n");
} else { } else {
uInfo("%s %s %" PRId64, src, name, pItem->i64); uInfo("%s %s %" PRId64, src, name, pItem->i64);
} }
break; break;
case CFG_DTYPE_FLOAT:
case CFG_DTYPE_DOUBLE: case CFG_DTYPE_DOUBLE:
case CFG_DTYPE_FLOAT:
if (dump) { if (dump) {
printf("%s %s %.2f", src, name, pItem->fval); printf("%s %s %.2f\n", src, name, pItem->fval);
printf("\n");
} else { } else {
uInfo("%s %s %.2f", src, name, pItem->fval); uInfo("%s %s %.2f", src, name, pItem->fval);
} }
@ -650,8 +636,7 @@ void cfgDumpCfg(SConfig *pCfg, bool tsc, bool dump) {
case CFG_DTYPE_TIMEZONE: case CFG_DTYPE_TIMEZONE:
case CFG_DTYPE_NONE: case CFG_DTYPE_NONE:
if (dump) { if (dump) {
printf("%s %s %s", src, name, pItem->str); printf("%s %s %s\n", src, name, pItem->str);
printf("\n");
} else { } else {
uInfo("%s %s %s", src, name, pItem->str); uInfo("%s %s %s", src, name, pItem->str);
} }
@ -660,8 +645,7 @@ void cfgDumpCfg(SConfig *pCfg, bool tsc, bool dump) {
} }
if (dump) { if (dump) {
printf("================================================================="); printf("=================================================================\n");
printf("\n");
} else { } else {
uInfo("================================================================="); uInfo("=================================================================");
} }