diff --git a/cmake/taostools_CMakeLists.txt.in b/cmake/taostools_CMakeLists.txt.in index 5683a4358e..7247200fe7 100644 --- a/cmake/taostools_CMakeLists.txt.in +++ b/cmake/taostools_CMakeLists.txt.in @@ -2,7 +2,7 @@ # taos-tools ExternalProject_Add(taos-tools GIT_REPOSITORY https://github.com/taosdata/taos-tools.git - GIT_TAG cac24d3 + GIT_TAG 4a4027c SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools" BINARY_DIR "" #BUILD_IN_SOURCE TRUE diff --git a/include/client/taos.h b/include/client/taos.h index 69774b750f..379363c51d 100644 --- a/include/client/taos.h +++ b/include/client/taos.h @@ -149,7 +149,7 @@ DLL_EXPORT TAOS *taos_connect(const char *ip, const char *user, const char DLL_EXPORT TAOS *taos_connect_auth(const char *ip, const char *user, const char *auth, const char *db, uint16_t port); DLL_EXPORT void taos_close(TAOS *taos); -const char *taos_data_type(int type); +DLL_EXPORT const char *taos_data_type(int type); DLL_EXPORT TAOS_STMT *taos_stmt_init(TAOS *taos); DLL_EXPORT TAOS_STMT *taos_stmt_init_with_reqid(TAOS *taos, int64_t reqid); diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 14d58ef58c..3537ba687d 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -70,6 +70,11 @@ static inline bool vnodeIsMsgBlock(tmsg_t type) { return (type == TDMT_VND_CREATE_TABLE) || (type == TDMT_VND_ALTER_TABLE) || (type == TDMT_VND_DROP_TABLE) || (type == TDMT_VND_UPDATE_TAG_VAL); } + +static inline bool syncUtilUserCommit(tmsg_t msgType) { + return msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_LEADER_TRANSFER; +} + /* ------------------------ OTHER DEFINITIONS ------------------------ */ // IE type #define TSDB_IE_TYPE_SEC 1 diff --git a/include/libs/sync/sync.h b/include/libs/sync/sync.h index 900bc88c41..4cf4800472 100644 --- a/include/libs/sync/sync.h +++ b/include/libs/sync/sync.h @@ -47,6 +47,7 @@ extern "C" { #define SYNC_HEARTBEAT_SLOW_MS 1500 #define SYNC_HEARTBEAT_REPLY_SLOW_MS 1500 +#define SYNC_SNAP_RESEND_MS 1000 * 60 #define SYNC_MAX_BATCH_SIZE 1 #define SYNC_INDEX_BEGIN 0 diff --git a/include/os/os.h b/include/os/os.h index 0688eeb9ad..b27fa84406 100644 --- a/include/os/os.h +++ b/include/os/os.h @@ -27,6 +27,7 @@ extern "C" { #if !defined(WINDOWS) #include +#include #include #include #include diff --git a/include/os/osDef.h b/include/os/osDef.h index 0bf9c6184e..c18728c9a7 100644 --- a/include/os/osDef.h +++ b/include/os/osDef.h @@ -120,12 +120,6 @@ void syslog(int unused, const char *format, ...); #define POINTER_SHIFT(p, b) ((void *)((char *)(p) + (b))) #define POINTER_DISTANCE(p1, p2) ((char *)(p1) - (char *)(p2)) -#ifndef NDEBUG -#define ASSERT(x) assert(x) -#else -#define ASSERT(x) -#endif - #ifndef UNUSED #define UNUSED(x) ((void)(x)) #endif diff --git a/include/os/osDir.h b/include/os/osDir.h index 40012f8141..73871602c5 100644 --- a/include/os/osDir.h +++ b/include/os/osDir.h @@ -62,6 +62,7 @@ int32_t taosRealPath(char *dirname, char *realPath, int32_t maxlen); bool taosIsDir(const char *dirname); char *taosDirName(char *dirname); char *taosDirEntryBaseName(char *dirname); +void taosGetCwd(char *buf, int32_t len); TdDirPtr taosOpenDir(const char *dirname); TdDirEntryPtr taosReadDir(TdDirPtr pDir); diff --git a/include/os/osString.h b/include/os/osString.h index 4c3fbe46c3..b609c9d351 100644 --- a/include/os/osString.h +++ b/include/os/osString.h @@ -62,7 +62,7 @@ typedef int32_t TdUcs4; int32_t taosUcs4len(TdUcs4 *ucs4); int64_t taosStr2int64(const char *str); -void taosConvInit(void); +int32_t taosConvInit(void); void taosConvDestroy(); int32_t taosUcs4ToMbs(TdUcs4 *ucs4, int32_t ucs4_max_len, char *mbs); bool taosMbsToUcs4(const char *mbs, size_t mbs_len, TdUcs4 *ucs4, int32_t ucs4_max_len, int32_t *len); diff --git a/include/os/osSystem.h b/include/os/osSystem.h index eca984c41a..8f1f5c58d5 100644 --- a/include/os/osSystem.h +++ b/include/os/osSystem.h @@ -46,6 +46,29 @@ void taosSetTerminalMode(); int32_t taosGetOldTerminalMode(); void taosResetTerminalMode(); +#if !defined(WINDOWS) +#define taosPrintTrace(flags, level, dflag) \ + { \ + void* array[100]; \ + int32_t size = backtrace(array, 100); \ + char** strings = backtrace_symbols(array, size); \ + if (strings != NULL) { \ + taosPrintLog(flags, level, dflag, "obtained %d stack frames", size); \ + for (int32_t i = 0; i < size; i++) { \ + taosPrintLog(flags, level, dflag, "frame:%d, %s", i, strings[i]); \ + } \ + } \ + \ + taosMemoryFree(strings); \ + } +#else +#define taosPrintTrace(flags, level, dflag) \ + { \ + taosPrintLog(flags, level, dflag, \ + "backtrace not implemented on windows, so detailed stack information cannot be printed"); \ + } +#endif + #ifdef __cplusplus } #endif diff --git a/include/util/tcoding.h b/include/util/tcoding.h index 38eb0d8fc6..b1bd09e123 100644 --- a/include/util/tcoding.h +++ b/include/util/tcoding.h @@ -17,6 +17,7 @@ #define _TD_UTIL_CODING_H_ #include "os.h" +#include "tlog.h" #ifdef __cplusplus extern "C" { diff --git a/include/util/tdef.h b/include/util/tdef.h index 15133630f8..2bc8ef2788 100644 --- a/include/util/tdef.h +++ b/include/util/tdef.h @@ -308,8 +308,9 @@ typedef enum ELogicConditionType { #define TSDB_MIN_DURATION_PER_FILE 60 // unit minute #define TSDB_MAX_DURATION_PER_FILE (3650 * 1440) #define TSDB_DEFAULT_DURATION_PER_FILE (10 * 1440) -#define TSDB_MIN_KEEP (1 * 1440) // data in db to be reserved. unit minute -#define TSDB_MAX_KEEP (365000 * 1440) // data in db to be reserved. +#define TSDB_MIN_KEEP (1 * 1440) // data in db to be reserved. unit minute +#define TSDB_MAX_KEEP (365000 * 1440) // data in db to be reserved. +#define TSDB_MAX_KEEP_NS (365 * 292 * 1440) // data in db to be reserved. #define TSDB_DEFAULT_KEEP (3650 * 1440) // ten years #define TSDB_MIN_MINROWS_FBLOCK 10 #define TSDB_MAX_MINROWS_FBLOCK 1000 diff --git a/include/util/tlog.h b/include/util/tlog.h index 68b004cda7..e6ef7f388f 100644 --- a/include/util/tlog.h +++ b/include/util/tlog.h @@ -38,6 +38,7 @@ typedef void (*LogFp)(int64_t ts, ELogLevel level, const char *content); extern bool tsLogEmbedded; extern bool tsAsyncLog; +extern bool tsAssert; extern int32_t tsNumOfLogLines; extern int32_t tsLogKeepDays; extern LogFp tsLogFp; @@ -82,6 +83,10 @@ void taosPrintLongString(const char *flags, ELogLevel level, int32_t dflag, cons #endif ; +bool taosAssert(bool condition, const char *file, int32_t line, const char *format, ...); +#define ASSERTS(condition, ...) taosAssert(condition, __FILE__, __LINE__, __VA_ARGS__) +#define ASSERT(condition) ASSERTS(condition, "assert info not provided") + // clang-format off #define uFatal(...) { if (uDebugFlag & DEBUG_FATAL) { taosPrintLog("UTL FATAL", DEBUG_FATAL, tsLogEmbedded ? 255 : uDebugFlag, __VA_ARGS__); }} #define uError(...) { if (uDebugFlag & DEBUG_ERROR) { taosPrintLog("UTL ERROR ", DEBUG_ERROR, tsLogEmbedded ? 255 : uDebugFlag, __VA_ARGS__); }} diff --git a/packaging/tools/make_install.sh b/packaging/tools/make_install.sh index 9034fd85f5..6a82322a12 100755 --- a/packaging/tools/make_install.sh +++ b/packaging/tools/make_install.sh @@ -497,27 +497,27 @@ function install_service_on_systemd() { taosd_service_config="${service_config_dir}/${serverName}.service" - ${csudo}bash -c "echo '[Unit]' >> ${taosd_service_config}" - ${csudo}bash -c "echo 'Description=${productName} server service' >> ${taosd_service_config}" - ${csudo}bash -c "echo 'After=network-online.target' >> ${taosd_service_config}" - ${csudo}bash -c "echo 'Wants=network-online.target' >> ${taosd_service_config}" + ${csudo}bash -c "echo [Unit] >> ${taosd_service_config}" + ${csudo}bash -c "echo Description=${productName} server service >> ${taosd_service_config}" + ${csudo}bash -c "echo After=network-online.target >> ${taosd_service_config}" + ${csudo}bash -c "echo Wants=network-online.target >> ${taosd_service_config}" ${csudo}bash -c "echo >> ${taosd_service_config}" - ${csudo}bash -c "echo '[Service]' >> ${taosd_service_config}" - ${csudo}bash -c "echo 'Type=simple' >> ${taosd_service_config}" - ${csudo}bash -c "echo 'ExecStart=/usr/bin/${serverName}' >> ${taosd_service_config}" - ${csudo}bash -c "echo 'ExecStartPre=${installDir}/bin/startPre.sh' >> ${taosd_service_config}" - ${csudo}bash -c "echo 'TimeoutStopSec=1000000s' >> ${taosd_service_config}" - ${csudo}bash -c "echo 'LimitNOFILE=infinity' >> ${taosd_service_config}" - ${csudo}bash -c "echo 'LimitNPROC=infinity' >> ${taosd_service_config}" - ${csudo}bash -c "echo 'LimitCORE=infinity' >> ${taosd_service_config}" - ${csudo}bash -c "echo 'TimeoutStartSec=0' >> ${taosd_service_config}" - ${csudo}bash -c "echo 'StandardOutput=null' >> ${taosd_service_config}" - ${csudo}bash -c "echo 'Restart=always' >> ${taosd_service_config}" - ${csudo}bash -c "echo 'StartLimitBurst=3' >> ${taosd_service_config}" - ${csudo}bash -c "echo 'StartLimitInterval=60s' >> ${taosd_service_config}" + ${csudo}bash -c "echo [Service] >> ${taosd_service_config}" + ${csudo}bash -c "echo Type=simple >> ${taosd_service_config}" + ${csudo}bash -c "echo ExecStart=/usr/bin/${serverName} >> ${taosd_service_config}" + ${csudo}bash -c "echo ExecStartPre=${installDir}/bin/startPre.sh >> ${taosd_service_config}" + ${csudo}bash -c "echo TimeoutStopSec=1000000s >> ${taosd_service_config}" + ${csudo}bash -c "echo LimitNOFILE=infinity >> ${taosd_service_config}" + ${csudo}bash -c "echo LimitNPROC=infinity >> ${taosd_service_config}" + ${csudo}bash -c "echo LimitCORE=infinity >> ${taosd_service_config}" + ${csudo}bash -c "echo TimeoutStartSec=0 >> ${taosd_service_config}" + ${csudo}bash -c "echo StandardOutput=null >> ${taosd_service_config}" + ${csudo}bash -c "echo Restart=always >> ${taosd_service_config}" + ${csudo}bash -c "echo StartLimitBurst=3 >> ${taosd_service_config}" + ${csudo}bash -c "echo StartLimitInterval=60s >> ${taosd_service_config}" ${csudo}bash -c "echo >> ${taosd_service_config}" - ${csudo}bash -c "echo '[Install]' >> ${taosd_service_config}" - ${csudo}bash -c "echo 'WantedBy=multi-user.target' >> ${taosd_service_config}" + ${csudo}bash -c "echo [Install] >> ${taosd_service_config}" + ${csudo}bash -c "echo WantedBy=multi-user.target >> ${taosd_service_config}" ${csudo}systemctl enable ${serverName} } diff --git a/source/client/src/clientEnv.c b/source/client/src/clientEnv.c index 06a334db9c..d3d713c643 100644 --- a/source/client/src/clientEnv.c +++ b/source/client/src/clientEnv.c @@ -413,7 +413,9 @@ void taos_init_imp(void) { initQueryModuleMsgHandle(); - taosConvInit(); + if (taosConvInit() != 0) { + ASSERTS(0, "failed to init conv"); + } rpcInit(); diff --git a/source/client/src/clientTmq.c b/source/client/src/clientTmq.c index ade0c95227..766a94caf1 100644 --- a/source/client/src/clientTmq.c +++ b/source/client/src/clientTmq.c @@ -875,6 +875,7 @@ void tmqFreeImpl(void* handle) { tmq_t* tmq = (tmq_t*)handle; // TODO stop timer + tmqClearUnhandleMsg(tmq); if (tmq->mqueue) taosCloseQueue(tmq->mqueue); if (tmq->delayedTask) taosCloseQueue(tmq->delayedTask); if (tmq->qall) taosFreeQall(tmq->qall); @@ -884,8 +885,7 @@ void tmqFreeImpl(void* handle) { int32_t sz = taosArrayGetSize(tmq->clientTopics); for (int32_t i = 0; i < sz; i++) { SMqClientTopic* pTopic = taosArrayGet(tmq->clientTopics, i); - if (pTopic->schema.nCols) taosMemoryFreeClear(pTopic->schema.pSchema); - int32_t vgSz = taosArrayGetSize(pTopic->vgs); + taosMemoryFreeClear(pTopic->schema.pSchema); taosArrayDestroy(pTopic->vgs); } taosArrayDestroy(tmq->clientTopics); @@ -1304,7 +1304,6 @@ bool tmqUpdateEp(tmq_t* tmq, int32_t epoch, const SMqAskEpRsp* pRsp) { for (int32_t i = 0; i < sz; i++) { SMqClientTopic* pTopic = taosArrayGet(tmq->clientTopics, i); if (pTopic->schema.nCols) taosMemoryFreeClear(pTopic->schema.pSchema); - int32_t vgSz = taosArrayGetSize(pTopic->vgs); taosArrayDestroy(pTopic->vgs); } taosArrayDestroy(tmq->clientTopics); @@ -1410,7 +1409,7 @@ int32_t tmqAskEp(tmq_t* tmq, bool async) { return -1; } void* pReq = taosMemoryCalloc(1, tlen); - if (tlen < 0) { + if (pReq == NULL) { tscError("failed to malloc askEpReq msg, size:%d", tlen); return -1; } @@ -1738,7 +1737,7 @@ void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) { taosFreeQitem(pollRspWrapper); return pRsp; } else { - tscDebug("msg discard since epoch mismatch: msg epoch %d, consumer epoch %d\n", + tscDebug("msg discard since epoch mismatch: msg epoch %d, consumer epoch %d", pollRspWrapper->taosxRsp.head.epoch, consumerEpoch); taosFreeQitem(pollRspWrapper); } diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index d1a3f38143..7e9b28939b 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -333,6 +333,7 @@ static int32_t taosAddSystemCfg(SConfig *pCfg) { if (cfgAddTimezone(pCfg, "timezone", tsTimezoneStr) != 0) return -1; if (cfgAddLocale(pCfg, "locale", tsLocale) != 0) return -1; if (cfgAddCharset(pCfg, "charset", tsCharset) != 0) return -1; + if (cfgAddBool(pCfg, "assert", 1, 1) != 0) return -1; if (cfgAddBool(pCfg, "enableCoreFile", 1, 1) != 0) return -1; if (cfgAddFloat(pCfg, "numOfCores", tsNumOfCores, 1, 100000, 1) != 0) return -1; @@ -693,6 +694,8 @@ static void taosSetSystemCfg(SConfig *pCfg) { bool enableCore = cfgGetItem(pCfg, "enableCoreFile")->bval; taosSetCoreDump(enableCore); + tsAssert = cfgGetItem(pCfg, "assert")->bval; + // todo tsVersion = 30000000; } @@ -788,6 +791,8 @@ int32_t taosSetCfg(SConfig *pCfg, char *name) { case 'a': { if (strcasecmp("asyncLog", name) == 0) { tsAsyncLog = cfgGetItem(pCfg, "asyncLog")->bval; + } else if (strcasecmp("assert", name) == 0) { + tsAssert = cfgGetItem(pCfg, "assert")->bval; } break; } diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index 22ed4e4aa3..95a72843e8 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -28,6 +28,8 @@ #undef TD_MSG_SEG_CODE_ #include "tmsgdef.h" +#include "tlog.h" + int32_t tInitSubmitMsgIter(const SSubmitReq *pMsg, SSubmitMsgIter *pIter) { if (pMsg == NULL) { terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP; diff --git a/source/common/src/trow.c b/source/common/src/trow.c index 52ebd7f879..ca2c056743 100644 --- a/source/common/src/trow.c +++ b/source/common/src/trow.c @@ -15,6 +15,7 @@ #define _DEFAULT_SOURCE #include "trow.h" +#include "tlog.h" const uint8_t tdVTypeByte[2][3] = {{ // 2 bits diff --git a/source/common/src/ttime.c b/source/common/src/ttime.c index a106a09a69..ec05ef4c44 100644 --- a/source/common/src/ttime.c +++ b/source/common/src/ttime.c @@ -23,6 +23,8 @@ #define _DEFAULT_SOURCE #include "ttime.h" +#include "tlog.h" + /* * mktime64 - Converts date to seconds. * Converts Gregorian date to seconds since 1970-01-01 00:00:00. diff --git a/source/dnode/mgmt/exe/dmMain.c b/source/dnode/mgmt/exe/dmMain.c index 188677656a..a8103351b4 100644 --- a/source/dnode/mgmt/exe/dmMain.c +++ b/source/dnode/mgmt/exe/dmMain.c @@ -17,6 +17,7 @@ #include "dmMgmt.h" #include "mnode.h" #include "tconfig.h" +#include "tglobal.h" // 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\"}'." @@ -45,9 +46,30 @@ static struct { SArray *pArgs; // SConfigPair } global = {0}; -static void dmStopDnode(int signum, void *info, void *ctx) { dmStop(); } +static void dmSetDebugFlag(int32_t signum, void *sigInfo, void *context) { taosSetAllDebugFlag(143, true); } +static void dmSetAssert(int32_t signum, void *sigInfo, void *context) { tsAssert = 1; } + +static void dmStopDnode(int signum, void *sigInfo, void *context) { + // taosIgnSignal(SIGUSR1); + // taosIgnSignal(SIGUSR2); + taosIgnSignal(SIGTERM); + taosIgnSignal(SIGHUP); + taosIgnSignal(SIGINT); + taosIgnSignal(SIGABRT); + taosIgnSignal(SIGBREAK); + + dInfo("shut down signal is %d", signum); +#ifndef WINDOWS + dInfo("sender PID:%d cmdline:%s", ((siginfo_t *)sigInfo)->si_pid, + taosGetCmdlineByPID(((siginfo_t *)sigInfo)->si_pid)); +#endif + + dmStop(); +} static void dmSetSignalHandle() { + taosSetSignal(SIGUSR1, dmSetDebugFlag); + taosSetSignal(SIGUSR2, dmSetAssert); taosSetSignal(SIGTERM, dmStopDnode); taosSetSignal(SIGHUP, dmStopDnode); taosSetSignal(SIGINT, dmStopDnode); @@ -105,6 +127,19 @@ static int32_t dmParseArgs(int32_t argc, char const *argv[]) { return 0; } +static void dmPrintArgs(int32_t argc, char const *argv[]) { + char path[1024] = {0}; + taosGetCwd(path, sizeof(path)); + + char args[1024] = {0}; + int32_t arglen = snprintf(args, sizeof(args), "%s", argv[0]); + for (int32_t i = 1; i < argc; ++i) { + arglen = arglen + snprintf(args + arglen, sizeof(args) - arglen, " %s", argv[i]); + } + + dInfo("startup path:%s args:%s", path, args); +} + static void dmGenerateGrant() { mndGenerateMachineCode(); } static void dmPrintVersion() { @@ -194,6 +229,8 @@ int mainWindows(int argc, char **argv) { return -1; } + dmPrintArgs(argc, argv); + if (taosInitCfg(configDir, global.envCmd, global.envFile, global.apolloUrl, global.pArgs, 0) != 0) { dError("failed to start since read config error"); taosCloseLog(); @@ -201,7 +238,12 @@ int mainWindows(int argc, char **argv) { return -1; } - taosConvInit(); + if (taosConvInit() != 0) { + dError("failed to init conv"); + taosCloseLog(); + taosCleanupArgs(); + return -1; + } if (global.dumpConfig) { dmDumpCfg(); diff --git a/source/dnode/mgmt/mgmt_dnode/src/dmHandle.c b/source/dnode/mgmt/mgmt_dnode/src/dmHandle.c index a7ad983b0c..39eeeb32ea 100644 --- a/source/dnode/mgmt/mgmt_dnode/src/dmHandle.c +++ b/source/dnode/mgmt/mgmt_dnode/src/dmHandle.c @@ -103,7 +103,12 @@ void dmSendStatusReq(SDnodeMgmt *pMgmt) { tSerializeSStatusReq(pHead, contLen, &req); tFreeSStatusReq(&req); - SRpcMsg rpcMsg = {.pCont = pHead, .contLen = contLen, .msgType = TDMT_MND_STATUS, .info.ahandle = (void *)0x9527}; + SRpcMsg rpcMsg = {.pCont = pHead, + .contLen = contLen, + .msgType = TDMT_MND_STATUS, + .info.ahandle = (void *)0x9527, + .info.refId = 0, + .info.noResp = 0}; SRpcMsg rpcRsp = {0}; dTrace("send status req to mnode, dnodeVer:%" PRId64 " statusSeq:%d", req.dnodeVer, req.statusSeq); @@ -150,7 +155,8 @@ static void dmGetServerRunStatus(SDnodeMgmt *pMgmt, SServerStatusRsp *pStatus) { SServerStatusRsp statusRsp = {0}; SMonMloadInfo minfo = {0}; (*pMgmt->getMnodeLoadsFp)(&minfo); - if (minfo.isMnode && (minfo.load.syncState == TAOS_SYNC_STATE_ERROR || minfo.load.syncState == TAOS_SYNC_STATE_OFFLINE)) { + if (minfo.isMnode && + (minfo.load.syncState == TAOS_SYNC_STATE_ERROR || minfo.load.syncState == TAOS_SYNC_STATE_OFFLINE)) { pStatus->statusCode = TSDB_SRV_STATUS_SERVICE_DEGRADED; snprintf(pStatus->details, sizeof(pStatus->details), "mnode sync state is %s", syncStr(minfo.load.syncState)); return; diff --git a/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c b/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c index 30ef7b9542..80c040a5e8 100644 --- a/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c +++ b/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c @@ -20,7 +20,9 @@ static void *dmStatusThreadFp(void *param) { SDnodeMgmt *pMgmt = param; int64_t lastTime = taosGetTimestampMs(); setThreadName("dnode-status"); - + + const static int16_t TRIM_FREQ = 30; + int32_t trimCount = 0; while (1) { taosMsleep(200); if (pMgmt->pData->dropped || pMgmt->pData->stopped) break; @@ -28,9 +30,13 @@ static void *dmStatusThreadFp(void *param) { int64_t curTime = taosGetTimestampMs(); float interval = (curTime - lastTime) / 1000.0f; if (interval >= tsStatusInterval) { - taosMemoryTrim(0); dmSendStatusReq(pMgmt); lastTime = curTime; + + trimCount = (trimCount + 1) % TRIM_FREQ; + if (trimCount == 0) { + taosMemoryTrim(0); + } } } diff --git a/source/dnode/mgmt/mgmt_snode/src/smWorker.c b/source/dnode/mgmt/mgmt_snode/src/smWorker.c index 2d2a121795..2e66aeae37 100644 --- a/source/dnode/mgmt/mgmt_snode/src/smWorker.c +++ b/source/dnode/mgmt/mgmt_snode/src/smWorker.c @@ -139,7 +139,7 @@ int32_t smPutMsgToQueue(SSnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) { SSnode *pSnode = pMgmt->pSnode; if (pSnode == NULL) { - dError("snode: msg:%p failed to put into vnode queue since %s, type:%s qtype:%d", pMsg, terrstr(), + dError("msg:%p failed to put into snode queue since %s, type:%s qtype:%d", pMsg, terrstr(), TMSG_INFO(pMsg->msgType), qtype); taosFreeQitem(pMsg); rpcFreeCont(pRpc->pCont); @@ -161,7 +161,8 @@ int32_t smPutMsgToQueue(SSnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) { smPutNodeMsgToWriteQueue(pMgmt, pMsg); break; default: - ASSERT(0); + ASSERTS(0, "msg:%p failed to put into snode queue since %s, type:%s qtype:%d", pMsg, terrstr(), + TMSG_INFO(pMsg->msgType), qtype); } return 0; } diff --git a/source/dnode/mgmt/node_mgmt/src/dmTransport.c b/source/dnode/mgmt/node_mgmt/src/dmTransport.c index 2383a09343..99f8bd002a 100644 --- a/source/dnode/mgmt/node_mgmt/src/dmTransport.c +++ b/source/dnode/mgmt/node_mgmt/src/dmTransport.c @@ -301,6 +301,7 @@ int32_t dmInitServer(SDnode *pDnode) { rpcInit.connType = TAOS_CONN_SERVER; rpcInit.idleTime = tsShellActivityTimer * 1000; rpcInit.parent = pDnode; + rpcInit.compressSize = tsCompressMsgSize; pTrans->serverRpc = rpcOpen(&rpcInit); if (pTrans->serverRpc == NULL) { diff --git a/source/dnode/mnode/impl/src/mndConsumer.c b/source/dnode/mnode/impl/src/mndConsumer.c index 972907cc95..4aaa96cb71 100644 --- a/source/dnode/mnode/impl/src/mndConsumer.c +++ b/source/dnode/mnode/impl/src/mndConsumer.c @@ -555,6 +555,12 @@ static int32_t mndProcessSubscribeReq(SRpcMsg *pMsg) { } if (mndCheckDbPrivilegeByName(pMnode, pMsg->info.conn.user, MND_OPER_READ_DB, pTopic->db) != 0) { + mndReleaseTopic(pMnode, pTopic); + goto SUBSCRIBE_OVER; + } + + if (mndCheckTopicPrivilege(pMnode, pMsg->info.conn.user, MND_OPER_SUBSCRIBE, pTopic) != 0) { + mndReleaseTopic(pMnode, pTopic); goto SUBSCRIBE_OVER; } diff --git a/source/dnode/mnode/impl/src/mndDb.c b/source/dnode/mnode/impl/src/mndDb.c index 5fa86dffe2..43155124c1 100644 --- a/source/dnode/mnode/impl/src/mndDb.c +++ b/source/dnode/mnode/impl/src/mndDb.c @@ -825,7 +825,13 @@ static int32_t mndProcessAlterDbReq(SRpcMsg *pReq) { dbObj.cfgVersion++; dbObj.updateTime = taosGetTimestampMs(); code = mndAlterDb(pMnode, pReq, pDb, &dbObj); - if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS; + + if (dbObj.cfg.replications != pDb->cfg.replications) { + // return quickly, operation executed asynchronously + mInfo("db:%s, alter db replica from %d to %d", pDb->name, pDb->cfg.replications, dbObj.cfg.replications); + } else { + if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS; + } _OVER: if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) { diff --git a/source/dnode/mnode/impl/src/mndSync.c b/source/dnode/mnode/impl/src/mndSync.c index 895d7fedc4..10b1e36496 100644 --- a/source/dnode/mnode/impl/src/mndSync.c +++ b/source/dnode/mnode/impl/src/mndSync.c @@ -119,7 +119,13 @@ int32_t mndProcessWriteMsg(const SSyncFSM *pFsm, SRpcMsg *pMsg, const SFsmCbMeta } int32_t mndSyncCommitMsg(const SSyncFSM *pFsm, SRpcMsg *pMsg, const SFsmCbMeta *pMeta) { - int32_t code = mndProcessWriteMsg(pFsm, pMsg, pMeta); + int32_t code = 0; + if (!syncUtilUserCommit(pMsg->msgType)) { + goto _out; + } + code = mndProcessWriteMsg(pFsm, pMsg, pMeta); + +_out: rpcFreeCont(pMsg->pCont); pMsg->pCont = NULL; return code; @@ -143,9 +149,13 @@ void mndRestoreFinish(const SSyncFSM *pFsm) { SMnode *pMnode = pFsm->data; if (!pMnode->deploy) { - mInfo("vgId:1, sync restore finished, and will handle outstanding transactions"); - mndTransPullup(pMnode); - mndSetRestored(pMnode, true); + if (!pMnode->restored) { + mInfo("vgId:1, sync restore finished, and will handle outstanding transactions"); + mndTransPullup(pMnode); + mndSetRestored(pMnode, true); + } else { + mInfo("vgId:1, sync restore finished, repeat call"); + } } else { mInfo("vgId:1, sync restore finished"); } diff --git a/source/dnode/vnode/inc/vnode.h b/source/dnode/vnode/inc/vnode.h index 83392049b8..03069d5404 100644 --- a/source/dnode/vnode/inc/vnode.h +++ b/source/dnode/vnode/inc/vnode.h @@ -174,7 +174,7 @@ int32_t tsdbReaderOpen(SVnode *pVnode, SQueryTableDataCond *pCond, void *pTableL void tsdbReaderClose(STsdbReader *pReader); bool tsdbNextDataBlock(STsdbReader *pReader); void tsdbRetrieveDataBlockInfo(const STsdbReader *pReader, int32_t *rows, uint64_t *uid, STimeWindow *pWindow); -int32_t tsdbRetrieveDatablockSMA(STsdbReader *pReader, SColumnDataAgg ***pBlockSMA, bool *allHave); +int32_t tsdbRetrieveDatablockSMA(STsdbReader *pReader, SSDataBlock* pDataBlock, bool *allHave); SSDataBlock *tsdbRetrieveDataBlock(STsdbReader *pTsdbReadHandle, SArray *pColumnIdList); int32_t tsdbReaderReset(STsdbReader *pReader, SQueryTableDataCond *pCond); int32_t tsdbGetFileBlocksDistInfo(STsdbReader *pReader, STableBlockDistInfo *pTableBlockInfo); diff --git a/source/dnode/vnode/src/meta/metaCache.c b/source/dnode/vnode/src/meta/metaCache.c index 6a704d0425..0ed68dce95 100644 --- a/source/dnode/vnode/src/meta/metaCache.c +++ b/source/dnode/vnode/src/meta/metaCache.c @@ -454,7 +454,7 @@ int32_t metaGetCachedTableUidList(SMeta* pMeta, tb_uid_t suid, const uint8_t* pK SListNode* pNode = NULL; while ((pNode = tdListNext(&iter)) != NULL) { - memcpy(pBuf + sizeof(suid), pNode->data, keyLen); + memcpy(&pBuf[1], pNode->data, keyLen); // check whether it is existed in LRU cache, and remove it from linked list if not. LRUHandle* pRes = taosLRUCacheLookup(pCache, pBuf, len); diff --git a/source/dnode/vnode/src/meta/metaTable.c b/source/dnode/vnode/src/meta/metaTable.c index 0b00f036de..722a93137c 100644 --- a/source/dnode/vnode/src/meta/metaTable.c +++ b/source/dnode/vnode/src/meta/metaTable.c @@ -549,8 +549,8 @@ int metaTtlDropTable(SMeta *pMeta, int64_t ttl, SArray *tbUids) { } static void metaBuildTtlIdxKey(STtlIdxKey *ttlKey, const SMetaEntry *pME) { - int64_t ttlDays; - int64_t ctime; + int64_t ttlDays = 0; + int64_t ctime = 0; if (pME->type == TSDB_CHILD_TABLE) { ctime = pME->ctbEntry.ctime; ttlDays = pME->ctbEntry.ttlDays; @@ -1353,6 +1353,10 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) { goto end; } + if (stbEntry.stbEntry.schemaTag.pSchema == NULL) { + goto end; + } + pTagColumn = &stbEntry.stbEntry.schemaTag.pSchema[0]; STagVal tagVal = {.cid = pTagColumn->colId}; diff --git a/source/dnode/vnode/src/tsdb/tsdbCache.c b/source/dnode/vnode/src/tsdb/tsdbCache.c index 571fbb088b..8aca60fbd9 100644 --- a/source/dnode/vnode/src/tsdb/tsdbCache.c +++ b/source/dnode/vnode/src/tsdb/tsdbCache.c @@ -1025,7 +1025,10 @@ static int32_t nextRowIterOpen(CacheNextRowIter *pIter, tb_uid_t uid, STsdb *pTs SArray *pDelIdxArray = taosArrayInit(32, sizeof(SDelIdx)); code = tsdbReadDelIdx(pDelFReader, pDelIdxArray); - if (code) goto _err; + if (code) { + tsdbDelFReaderClose(&pDelFReader); + goto _err; + } SDelIdx *delIdx = taosArraySearch(pDelIdxArray, &(SDelIdx){.suid = suid, .uid = uid}, tCmprDelIdx, TD_EQ); diff --git a/source/dnode/vnode/src/tsdb/tsdbCommit.c b/source/dnode/vnode/src/tsdb/tsdbCommit.c index 40932f077c..92451dcf58 100644 --- a/source/dnode/vnode/src/tsdb/tsdbCommit.c +++ b/source/dnode/vnode/src/tsdb/tsdbCommit.c @@ -53,6 +53,7 @@ typedef struct { // -------------- TSKEY nextKey; // reset by each table commit int32_t commitFid; + int32_t expLevel; TSKEY minKey; TSKEY maxKey; // commit file data @@ -503,6 +504,7 @@ static int32_t tsdbCommitFileDataStart(SCommitter *pCommitter) { // memory pCommitter->commitFid = tsdbKeyFid(pCommitter->nextKey, pCommitter->minutes, pCommitter->precision); + pCommitter->expLevel = tsdbFidLevel(pCommitter->commitFid, &pCommitter->pTsdb->keepCfg, taosGetTimestampSec()); tsdbFidKeyRange(pCommitter->commitFid, pCommitter->minutes, pCommitter->precision, &pCommitter->minKey, &pCommitter->maxKey); #if 0 @@ -556,7 +558,10 @@ static int32_t tsdbCommitFileDataStart(SCommitter *pCommitter) { } } else { SDiskID did = {0}; - tfsAllocDisk(pTsdb->pVnode->pTfs, 0, &did); + if (tfsAllocDisk(pTsdb->pVnode->pTfs, pCommitter->expLevel, &did) < 0) { + code = terrno; + TSDB_CHECK_CODE(code, lino, _exit); + } tfsMkdirRecurAt(pTsdb->pVnode->pTfs, pTsdb->path, did); wSet.diskId = did; wSet.nSttF = 1; diff --git a/source/dnode/vnode/src/tsdb/tsdbFS.c b/source/dnode/vnode/src/tsdb/tsdbFS.c index 72d9c4f69e..7dc839773f 100644 --- a/source/dnode/vnode/src/tsdb/tsdbFS.c +++ b/source/dnode/vnode/src/tsdb/tsdbFS.c @@ -962,6 +962,7 @@ int32_t tsdbFSUpsertFSet(STsdbFS *pFS, SDFileSet *pSet) { } } + pDFileSet->diskId = pSet->diskId; goto _exit; } } diff --git a/source/dnode/vnode/src/tsdb/tsdbRead.c b/source/dnode/vnode/src/tsdb/tsdbRead.c index 823125a905..d6931e1b01 100644 --- a/source/dnode/vnode/src/tsdb/tsdbRead.c +++ b/source/dnode/vnode/src/tsdb/tsdbRead.c @@ -3128,7 +3128,7 @@ bool hasBeenDropped(const SArray* pDelList, int32_t* index, TSDBKEY* pKey, int32 return false; } else if (pKey->ts == last->ts) { TSDBKEY* prev = taosArrayGet(pDelList, num - 2); - return (prev->version >= pKey->version); + return (prev->version >= pKey->version && prev->version <= pVerRange->maxVer && prev->version >= pVerRange->minVer); } } else { TSDBKEY* pCurrent = taosArrayGet(pDelList, *index); @@ -4134,8 +4134,9 @@ static void doFillNullColSMA(SBlockLoadSuppInfo* pSup, int32_t numOfRows, int32_ } } -int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SColumnDataAgg*** pBlockSMA, bool* allHave) { +int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SSDataBlock* pDataBlock, bool* allHave) { int32_t code = 0; + SColumnDataAgg ***pBlockSMA = &pDataBlock->pBlockAgg; *allHave = false; if (pReader->type == TIMEWINDOW_RANGE_EXTERNAL) { @@ -4183,6 +4184,12 @@ int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SColumnDataAgg*** pBlockS int32_t i = 0, j = 0; size_t size = taosArrayGetSize(pSup->pColAgg); + // ensure capacity + if(pDataBlock->pDataBlock) { + size_t colsNum = taosArrayGetSize(pDataBlock->pDataBlock); + taosArrayEnsureCap(pSup->pColAgg, colsNum); + } + SSDataBlock* pResBlock = pReader->pResBlock; if (pResBlock->pBlockAgg == NULL) { size_t num = taosArrayGetSize(pResBlock->pDataBlock); diff --git a/source/dnode/vnode/src/tsdb/tsdbSnapshot.c b/source/dnode/vnode/src/tsdb/tsdbSnapshot.c index 048d3cc22b..bd17a2593b 100644 --- a/source/dnode/vnode/src/tsdb/tsdbSnapshot.c +++ b/source/dnode/vnode/src/tsdb/tsdbSnapshot.c @@ -155,7 +155,7 @@ static int32_t tsdbSnapReadOpenFile(STsdbSnapReader* pReader) { if (rowVer >= pReader->sver && rowVer <= pReader->ever) { pIter->rInfo.suid = pIter->bData.suid; - pIter->rInfo.uid = pIter->bData.uid; + pIter->rInfo.uid = pIter->bData.uid ? pIter->bData.uid : pIter->bData.aUid[pIter->iRow]; pIter->rInfo.row = tsdbRowFromBlockData(&pIter->bData, pIter->iRow); goto _add_iter; } @@ -179,16 +179,14 @@ _err: return code; } -static SRowInfo* tsdbSnapGetRow(STsdbSnapReader* pReader) { return pReader->pIter ? &pReader->pIter->rInfo : NULL; } - static int32_t tsdbSnapNextRow(STsdbSnapReader* pReader) { int32_t code = 0; if (pReader->pIter) { - SFDataIter* pIter = pReader->pIter; - + SFDataIter* pIter = NULL; while (true) { _find_row: + pIter = pReader->pIter; for (pIter->iRow++; pIter->iRow < pIter->bData.nRow; pIter->iRow++) { int64_t rowVer = pIter->bData.aVersion[pIter->iRow]; @@ -224,6 +222,7 @@ static int32_t tsdbSnapNextRow(STsdbSnapReader* pReader) { } pReader->pIter = NULL; + break; } else if (pIter->type == SNAP_STT_FILE_ITER) { for (pIter->iSttBlk++; pIter->iSttBlk < taosArrayGetSize(pIter->aSttBlk); pIter->iSttBlk++) { SSttBlk* pSttBlk = (SSttBlk*)taosArrayGet(pIter->aSttBlk, pIter->iSttBlk); @@ -238,6 +237,7 @@ static int32_t tsdbSnapNextRow(STsdbSnapReader* pReader) { } pReader->pIter = NULL; + break; } else { ASSERT(0); } @@ -269,6 +269,20 @@ _err: return code; } +static SRowInfo* tsdbSnapGetRow(STsdbSnapReader* pReader) { + if (pReader->pIter) { + return &pReader->pIter->rInfo; + } else { + tsdbSnapNextRow(pReader); + + if (pReader->pIter) { + return &pReader->pIter->rInfo; + } else { + return NULL; + } + } +} + static int32_t tsdbSnapCmprData(STsdbSnapReader* pReader, uint8_t** ppData) { int32_t code = 0; @@ -1356,7 +1370,7 @@ _exit: taosMemoryFree(pWriter); } } else { - tsdbDebug("vgId:%d, tsdb snapshot writer open for %s succeed", TD_VID(pTsdb->pVnode), pTsdb->path); + tsdbInfo("vgId:%d %s done", TD_VID(pTsdb->pVnode), __func__); *ppWriter = pWriter; } return code; @@ -1421,7 +1435,7 @@ int32_t tsdbSnapWriterClose(STsdbSnapWriter** ppWriter, int8_t rollback) { for (int32_t iBuf = 0; iBuf < sizeof(pWriter->aBuf) / sizeof(uint8_t*); iBuf++) { tFree(pWriter->aBuf[iBuf]); } - tsdbInfo("vgId:%d, vnode snapshot tsdb writer close for %s", TD_VID(pWriter->pTsdb->pVnode), pWriter->pTsdb->path); + tsdbInfo("vgId:%d %s done", TD_VID(pWriter->pTsdb->pVnode), __func__); taosMemoryFree(pWriter); *ppWriter = NULL; return code; diff --git a/source/dnode/vnode/src/tsdb/tsdbUtil.c b/source/dnode/vnode/src/tsdb/tsdbUtil.c index 63391ec0f4..1c92e922b0 100644 --- a/source/dnode/vnode/src/tsdb/tsdbUtil.c +++ b/source/dnode/vnode/src/tsdb/tsdbUtil.c @@ -637,6 +637,7 @@ SColVal *tsdbRowIterNext(STSDBRowIter *pIter) { } } else { ASSERT(0); + return NULL; // suppress error report by compiler } } diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c index cbfc16ce4a..bb5063cfea 100644 --- a/source/dnode/vnode/src/vnd/vnodeSvr.c +++ b/source/dnode/vnode/src/vnd/vnodeSvr.c @@ -210,9 +210,13 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRp version); ASSERT(pVnode->state.applyTerm <= pMsg->info.conn.applyTerm); + ASSERT(pVnode->state.applied + 1 == version); + pVnode->state.applied = version; pVnode->state.applyTerm = pMsg->info.conn.applyTerm; + if (!syncUtilUserCommit(pMsg->msgType)) goto _exit; + // skip header pReq = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)); len = pMsg->contLen - sizeof(SMsgHead); diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 0b198afb0d..1fcee1992e 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -224,7 +224,7 @@ static bool doFilterByBlockSMA(SFilterInfo* pFilterInfo, SColumnDataAgg** pColsA static bool doLoadBlockSMA(STableScanBase* pTableScanInfo, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo) { bool allColumnsHaveAgg = true; - int32_t code = tsdbRetrieveDatablockSMA(pTableScanInfo->dataReader, &pBlock->pBlockAgg, &allColumnsHaveAgg); + int32_t code = tsdbRetrieveDatablockSMA(pTableScanInfo->dataReader, pBlock, &allColumnsHaveAgg); if (code != TSDB_CODE_SUCCESS) { T_LONG_JMP(pTaskInfo->env, code); } diff --git a/source/libs/executor/src/timesliceoperator.c b/source/libs/executor/src/timesliceoperator.c index 2374d80bbf..072a96fa83 100644 --- a/source/libs/executor/src/timesliceoperator.c +++ b/source/libs/executor/src/timesliceoperator.c @@ -95,6 +95,8 @@ static void doKeepLinearInfo(STimeSliceOperatorInfo* pSliceInfo, const SSDataBlo // TODO: optimize to ignore null values for linear interpolation. if (!pLinearInfo->isStartSet) { if (!colDataIsNull_s(pColInfoData, rowIndex)) { + ASSERT(IS_MATHABLE_TYPE(pColInfoData->info.type)); + pLinearInfo->start.key = *(int64_t*)colDataGetData(pTsCol, rowIndex); memcpy(pLinearInfo->start.val, colDataGetData(pColInfoData, rowIndex), pLinearInfo->bytes); } diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index d9a011a892..6f1ab8ca65 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -863,19 +863,20 @@ static void removeResults(SArray* pWins, SHashObj* pUpdatedMap) { int32_t compareWinRes(void* pKey, void* data, int32_t index) { SArray* res = (SArray*)data; - SWinKey* pos = taosArrayGet(res, index); - SResKeyPos* pData = (SResKeyPos*)pKey; - if (*(int64_t*)pData->key == pos->ts) { - if (pData->groupId > pos->groupId) { - return 1; - } else if (pData->groupId < pos->groupId) { - return -1; - } - return 0; - } else if (*(int64_t*)pData->key > pos->ts) { + SWinKey* pDataPos = taosArrayGet(res, index); + SResKeyPos* pRKey = (SResKeyPos*)pKey; + if (pRKey->groupId > pDataPos->groupId) { return 1; + } else if (pRKey->groupId < pDataPos->groupId) { + return -1; } - return -1; + + if (*(int64_t*)pRKey->key > pDataPos->ts) { + return 1; + } else if (*(int64_t*)pRKey->key < pDataPos->ts){ + return -1; + } + return 0; } static void removeDeleteResults(SHashObj* pUpdatedMap, SArray* pDelWins) { @@ -1400,19 +1401,21 @@ static int32_t getAllIntervalWindow(SSHashObj* pHashMap, SHashObj* resWins) { int32_t compareWinKey(void* pKey, void* data, int32_t index) { SArray* res = (SArray*)data; - SWinKey* pos = taosArrayGet(res, index); - SWinKey* pData = (SWinKey*)pKey; - if (pData->ts == pos->ts) { - if (pData->groupId > pos->groupId) { - return 1; - } else if (pData->groupId < pos->groupId) { - return -1; - } - return 0; - } else if (pData->ts > pos->ts) { + SWinKey* pDataPos = taosArrayGet(res, index); + SWinKey* pWKey = (SWinKey*)pKey; + + if (pWKey->groupId > pDataPos->groupId) { return 1; + } else if (pWKey->groupId < pDataPos->groupId) { + return -1; } - return -1; + + if (pWKey->ts > pDataPos->ts) { + return 1; + } else if (pWKey->ts < pDataPos->ts) { + return -1; + } + return 0; } static int32_t closeStreamIntervalWindow(SSHashObj* pHashMap, STimeWindowAggSupp* pTwSup, SInterval* pInterval, diff --git a/source/libs/executor/src/tlinearhash.c b/source/libs/executor/src/tlinearhash.c index 4204692514..d97f81c994 100644 --- a/source/libs/executor/src/tlinearhash.c +++ b/source/libs/executor/src/tlinearhash.c @@ -17,6 +17,7 @@ #include "taoserror.h" #include "tdef.h" #include "tpagedbuf.h" +#include "tlog.h" #define LHASH_CAP_RATIO 0.85 diff --git a/source/libs/function/src/tpercentile.c b/source/libs/function/src/tpercentile.c index e5727f1472..9c3e1f5604 100644 --- a/source/libs/function/src/tpercentile.c +++ b/source/libs/function/src/tpercentile.c @@ -22,6 +22,7 @@ #include "tpagedbuf.h" #include "tpercentile.h" #include "ttypes.h" +#include "tlog.h" #define DEFAULT_NUM_OF_SLOT 1024 @@ -367,11 +368,13 @@ int32_t tMemBucketPut(tMemBucket *pBucket, const void *data, size_t size) { pSlot->info.data = NULL; } - SArray *pPageIdList = (SArray *)taosHashGet(pBucket->groupPagesMap, &groupId, sizeof(groupId)); - if (pPageIdList == NULL) { - SArray *pList = taosArrayInit(4, sizeof(int32_t)); - taosHashPut(pBucket->groupPagesMap, &groupId, sizeof(groupId), &pList, POINTER_BYTES); - pPageIdList = pList; + SArray *pPageIdList; + void *p = taosHashGet(pBucket->groupPagesMap, &groupId, sizeof(groupId)); + if (p == NULL) { + pPageIdList = taosArrayInit(4, sizeof(int32_t)); + taosHashPut(pBucket->groupPagesMap, &groupId, sizeof(groupId), &pPageIdList, POINTER_BYTES); + } else { + pPageIdList = *(SArray **)p; } pSlot->info.data = getNewBufPage(pBucket->pBuffer, &pageId); diff --git a/source/libs/function/src/tudf.c b/source/libs/function/src/tudf.c index c78ec5b999..32e57565d4 100644 --- a/source/libs/function/src/tudf.c +++ b/source/libs/function/src/tudf.c @@ -88,11 +88,13 @@ static int32_t udfSpawnUdfd(SUdfdData *pData) { } #ifdef WINDOWS if (strlen(path) == 0) { - strcat(path, "udfd.exe"); - } else { - strcat(path, "\\udfd.exe"); + strcat(path, "C:\\TDengine"); } + strcat(path, "\\udfd.exe"); #else + if (strlen(path) == 0) { + strcat(path, "/usr/bin"); + } strcat(path, "/udfd"); #endif char *argsUdfd[] = {path, "-c", configDir, NULL}; diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c index 85a4e70124..10e45901e5 100644 --- a/source/libs/parser/src/parTranslater.c +++ b/source/libs/parser/src/parTranslater.c @@ -3879,12 +3879,17 @@ static int32_t checkDbKeepOption(STranslateContext* pCxt, SDatabaseOptions* pOpt pOptions->keep[2] = getBigintFromValueNode((SValueNode*)nodesListGetNode(pOptions->pKeep, 2)); } + int64_t tsdbMaxKeep = TSDB_MAX_KEEP; + if (pOptions->precision == TSDB_TIME_PRECISION_NANO) { + tsdbMaxKeep = TSDB_MAX_KEEP_NS; + } + if (pOptions->keep[0] < TSDB_MIN_KEEP || pOptions->keep[1] < TSDB_MIN_KEEP || pOptions->keep[2] < TSDB_MIN_KEEP || - pOptions->keep[0] > TSDB_MAX_KEEP || pOptions->keep[1] > TSDB_MAX_KEEP || pOptions->keep[2] > TSDB_MAX_KEEP) { + pOptions->keep[0] > tsdbMaxKeep || pOptions->keep[1] > tsdbMaxKeep || pOptions->keep[2] > tsdbMaxKeep) { return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_DB_OPTION, "Invalid option keep: %" PRId64 ", %" PRId64 ", %" PRId64 " valid range: [%dm, %dm]", pOptions->keep[0], pOptions->keep[1], pOptions->keep[2], TSDB_MIN_KEEP, - TSDB_MAX_KEEP); + tsdbMaxKeep); } if (!((pOptions->keep[0] <= pOptions->keep[1]) && (pOptions->keep[1] <= pOptions->keep[2]))) { @@ -4036,7 +4041,10 @@ static int32_t checkDatabaseOptions(STranslateContext* pCxt, const char* pDbName TSDB_MAX_MINROWS_FBLOCK); } if (TSDB_CODE_SUCCESS == code) { - code = checkDbKeepOption(pCxt, pOptions); + code = checkDbPrecisionOption(pCxt, pOptions); + } + if (TSDB_CODE_SUCCESS == code) { + code = checkDbKeepOption(pCxt, pOptions); // use precision } if (TSDB_CODE_SUCCESS == code) { code = checkDbRangeOption(pCxt, "pages", pOptions->pages, TSDB_MIN_PAGES_PER_VNODE, TSDB_MAX_PAGES_PER_VNODE); @@ -4049,9 +4057,6 @@ static int32_t checkDatabaseOptions(STranslateContext* pCxt, const char* pDbName code = checkDbRangeOption(pCxt, "tsdbPagesize", pOptions->tsdbPageSize, TSDB_MIN_TSDB_PAGESIZE, TSDB_MAX_TSDB_PAGESIZE); } - if (TSDB_CODE_SUCCESS == code) { - code = checkDbPrecisionOption(pCxt, pOptions); - } if (TSDB_CODE_SUCCESS == code) { code = checkDbEnumOption(pCxt, "replications", pOptions->replica, TSDB_MIN_DB_REPLICA, TSDB_MAX_DB_REPLICA); } diff --git a/source/libs/sync/inc/syncSnapshot.h b/source/libs/sync/inc/syncSnapshot.h index 1f9675a3cd..ee83636192 100644 --- a/source/libs/sync/inc/syncSnapshot.h +++ b/source/libs/sync/inc/syncSnapshot.h @@ -44,6 +44,7 @@ typedef struct SSyncSnapshotSender { SyncTerm term; int64_t startTime; int64_t endTime; + int64_t lastSendTime; bool finish; // init when create diff --git a/source/libs/sync/inc/syncUtil.h b/source/libs/sync/inc/syncUtil.h index f198f3809d..be14ef91a4 100644 --- a/source/libs/sync/inc/syncUtil.h +++ b/source/libs/sync/inc/syncUtil.h @@ -79,7 +79,6 @@ char* syncUtilPrintBin2(char* ptr, uint32_t len); void syncUtilMsgHtoN(void* msg); void syncUtilMsgNtoH(void* msg); bool syncUtilUserPreCommit(tmsg_t msgType); -bool syncUtilUserCommit(tmsg_t msgType); bool syncUtilUserRollback(tmsg_t msgType); void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNode* pNode, const char* format, ...); diff --git a/source/libs/sync/src/syncCommit.c b/source/libs/sync/src/syncCommit.c index 07b1101256..5fdcbeb91c 100644 --- a/source/libs/sync/src/syncCommit.c +++ b/source/libs/sync/src/syncCommit.c @@ -84,7 +84,7 @@ void syncOneReplicaAdvance(SSyncNode* pSyncNode) { } void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) { - ASSERT(false && "deprecated"); + ASSERTS(false, "deprecated"); if (pSyncNode == NULL) { sError("pSyncNode is NULL"); return; diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 59e5e968e7..6fabab18cb 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -151,7 +151,7 @@ int32_t syncReconfig(int64_t rid, SSyncCfg* pNewCfg) { } syncNodeStartHeartbeatTimer(pSyncNode); - syncNodeReplicate(pSyncNode); + //syncNodeReplicate(pSyncNode); } syncNodeRelease(pSyncNode); @@ -791,9 +791,9 @@ static int32_t syncHbTimerStop(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) { } int32_t syncNodeLogStoreRestoreOnNeed(SSyncNode* pNode) { - ASSERT(pNode->pLogStore != NULL && "log store not created"); - ASSERT(pNode->pFsm != NULL && "pFsm not registered"); - ASSERT(pNode->pFsm->FpGetSnapshotInfo != NULL && "FpGetSnapshotInfo not registered"); + ASSERTS(pNode->pLogStore != NULL, "log store not created"); + ASSERTS(pNode->pFsm != NULL, "pFsm not registered"); + ASSERTS(pNode->pFsm->FpGetSnapshotInfo != NULL, "FpGetSnapshotInfo not registered"); SSnapshot snapshot; if (pNode->pFsm->FpGetSnapshotInfo(pNode->pFsm, &snapshot) < 0) { sError("vgId:%d, failed to get snapshot info since %s", pNode->vgId, terrstr()); @@ -1144,8 +1144,8 @@ void syncNodeMaybeUpdateCommitBySnapshot(SSyncNode* pSyncNode) { } int32_t syncNodeRestore(SSyncNode* pSyncNode) { - ASSERT(pSyncNode->pLogStore != NULL && "log store not created"); - ASSERT(pSyncNode->pLogBuf != NULL && "ring log buffer not created"); + ASSERTS(pSyncNode->pLogStore != NULL, "log store not created"); + ASSERTS(pSyncNode->pLogBuf != NULL, "ring log buffer not created"); SyncIndex lastVer = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore); SyncIndex commitIndex = pSyncNode->pLogStore->syncLogCommitIndex(pSyncNode->pLogStore); @@ -1839,7 +1839,8 @@ void syncNodeBecomeLeader(SSyncNode* pSyncNode, const char* debugStr) { #endif // close receiver - if (snapshotReceiverIsStart(pSyncNode->pNewNodeReceiver)) { + if (pSyncNode != NULL && pSyncNode->pNewNodeReceiver != NULL && + snapshotReceiverIsStart(pSyncNode->pNewNodeReceiver)) { snapshotReceiverForceStop(pSyncNode->pNewNodeReceiver); } @@ -2663,7 +2664,7 @@ int32_t syncNodeOnClientRequest(SSyncNode* ths, SRpcMsg* pMsg, SyncIndex* pRetIn int32_t code = syncNodeAppend(ths, pEntry); if (code < 0 && ths->vgId != 1 && vnodeIsMsgBlock(pEntry->originalRpcType)) { - ASSERT(false && "failed to append blocking msg"); + ASSERTS(false, "failed to append blocking msg"); } return code; } diff --git a/source/libs/sync/src/syncPipeline.c b/source/libs/sync/src/syncPipeline.c index 5d6905f99e..7bd8d75dd1 100644 --- a/source/libs/sync/src/syncPipeline.c +++ b/source/libs/sync/src/syncPipeline.c @@ -50,7 +50,7 @@ int32_t syncLogBufferAppend(SSyncLogBuffer* pBuf, SSyncNode* pNode, SSyncRaftEnt // initial log buffer with at least one item, e.g. commitIndex SSyncRaftEntry* pMatch = pBuf->entries[(index - 1 + pBuf->size) % pBuf->size].pItem; - ASSERT(pMatch != NULL && "no matched log entry"); + ASSERTS(pMatch != NULL, "no matched log entry"); ASSERT(pMatch->index + 1 == index); SSyncLogBufEntry tmp = {.pItem = pEntry, .prevLogIndex = pMatch->index, .prevLogTerm = pMatch->term}; @@ -86,14 +86,14 @@ SyncTerm syncLogReplMgrGetPrevLogTerm(SSyncLogReplMgr* pMgr, SSyncNode* pNode, S if (prevIndex >= pBuf->startIndex) { pEntry = pBuf->entries[(prevIndex + pBuf->size) % pBuf->size].pItem; - ASSERT(pEntry != NULL && "no log entry found"); + ASSERTS(pEntry != NULL, "no log entry found"); prevLogTerm = pEntry->term; return prevLogTerm; } if (pMgr && pMgr->startIndex <= prevIndex && prevIndex < pMgr->endIndex) { int64_t timeMs = pMgr->states[(prevIndex + pMgr->size) % pMgr->size].timeMs; - ASSERT(timeMs != 0 && "no log entry found"); + ASSERTS(timeMs != 0, "no log entry found"); prevLogTerm = pMgr->states[(prevIndex + pMgr->size) % pMgr->size].term; ASSERT(prevIndex == 0 || prevLogTerm != 0); return prevLogTerm; @@ -141,9 +141,9 @@ int32_t syncLogValidateAlignmentOfCommit(SSyncNode* pNode, SyncIndex commitIndex } int32_t syncLogBufferInitWithoutLock(SSyncLogBuffer* pBuf, SSyncNode* pNode) { - ASSERT(pNode->pLogStore != NULL && "log store not created"); - ASSERT(pNode->pFsm != NULL && "pFsm not registered"); - ASSERT(pNode->pFsm->FpGetSnapshotInfo != NULL && "FpGetSnapshotInfo not registered"); + ASSERTS(pNode->pLogStore != NULL, "log store not created"); + ASSERTS(pNode->pFsm != NULL, "pFsm not registered"); + ASSERTS(pNode->pFsm->FpGetSnapshotInfo != NULL, "FpGetSnapshotInfo not registered"); SSnapshot snapshot; if (pNode->pFsm->FpGetSnapshotInfo(pNode->pFsm, &snapshot) < 0) { @@ -437,7 +437,7 @@ _out: } int32_t syncLogFsmExecute(SSyncNode* pNode, SSyncFSM* pFsm, ESyncState role, SyncTerm term, SSyncRaftEntry* pEntry) { - ASSERT(pFsm->FpCommitCb != NULL && "No commit cb registered for the FSM"); + ASSERTS(pFsm->FpCommitCb != NULL, "No commit cb registered for the FSM"); if ((pNode->replicaNum == 1) && pNode->restoreFinish && pNode->vgId != 1) { return 0; @@ -513,13 +513,8 @@ int32_t syncLogBufferCommit(SSyncLogBuffer* pBuf, SSyncNode* pNode, int64_t comm if (!syncUtilUserCommit(pEntry->originalRpcType)) { sInfo("vgId:%d, commit sync barrier. index: %" PRId64 ", term:%" PRId64 ", type: %s", vgId, pEntry->index, pEntry->term, TMSG_INFO(pEntry->originalRpcType)); - pBuf->commitIndex = index; - if (!inBuf) { - syncEntryDestroy(pEntry); - pEntry = NULL; - } - continue; } + if (syncLogFsmExecute(pNode, pFsm, role, term, pEntry) != 0) { sError("vgId:%d, failed to execute sync log entry. index:%" PRId64 ", term:%" PRId64 ", role: %d, current term: %" PRId64, @@ -905,7 +900,7 @@ int32_t syncNodeLogReplMgrInit(SSyncNode* pNode) { ASSERT(pNode->logReplMgrs[i] == NULL); pNode->logReplMgrs[i] = syncLogReplMgrCreate(); pNode->logReplMgrs[i]->peerId = i; - ASSERT(pNode->logReplMgrs[i] != NULL && "Out of memory."); + ASSERTS(pNode->logReplMgrs[i] != NULL, "Out of memory."); } return 0; } diff --git a/source/libs/sync/src/syncSnapshot.c b/source/libs/sync/src/syncSnapshot.c index b8ecbe7515..540f40a4c0 100644 --- a/source/libs/sync/src/syncSnapshot.c +++ b/source/libs/sync/src/syncSnapshot.c @@ -103,6 +103,7 @@ int32_t snapshotSenderStart(SSyncSnapshotSender *pSender) { pSender->sendingMS = 0; pSender->term = pSender->pSyncNode->pRaftStore->currentTerm; pSender->startTime = taosGetTimestampMs(); + pSender->lastSendTime = pSender->startTime; pSender->finish = false; // build begin msg @@ -201,6 +202,8 @@ int32_t snapshotSend(SSyncSnapshotSender *pSender) { syncNodeSendMsgById(&pMsg->destId, pSender->pSyncNode, &rpcMsg); syncLogSendSyncSnapshotSend(pSender->pSyncNode, pMsg, ""); + pSender->lastSendTime = taosGetTimestampMs(); + // event log if (pSender->seq == SYNC_SNAPSHOT_SEQ_END) { sSTrace(pSender, "snapshot sender finish"); @@ -213,33 +216,36 @@ int32_t snapshotSend(SSyncSnapshotSender *pSender) { // send snapshot data from cache int32_t snapshotReSend(SSyncSnapshotSender *pSender) { // send current block data + + // build msg + SRpcMsg rpcMsg = {0}; + (void)syncBuildSnapshotSend(&rpcMsg, pSender->blockLen, pSender->pSyncNode->vgId); + + SyncSnapshotSend *pMsg = rpcMsg.pCont; + pMsg->srcId = pSender->pSyncNode->myRaftId; + pMsg->destId = (pSender->pSyncNode->replicasId)[pSender->replicaIndex]; + pMsg->term = pSender->pSyncNode->pRaftStore->currentTerm; + pMsg->beginIndex = pSender->snapshotParam.start; + pMsg->lastIndex = pSender->snapshot.lastApplyIndex; + pMsg->lastTerm = pSender->snapshot.lastApplyTerm; + pMsg->lastConfigIndex = pSender->snapshot.lastConfigIndex; + pMsg->lastConfig = pSender->lastConfig; + pMsg->seq = pSender->seq; + if (pSender->pCurrentBlock != NULL && pSender->blockLen > 0) { - // build msg - SRpcMsg rpcMsg = {0}; - (void)syncBuildSnapshotSend(&rpcMsg, pSender->blockLen, pSender->pSyncNode->vgId); - - SyncSnapshotSend *pMsg = rpcMsg.pCont; - pMsg->srcId = pSender->pSyncNode->myRaftId; - pMsg->destId = (pSender->pSyncNode->replicasId)[pSender->replicaIndex]; - pMsg->term = pSender->pSyncNode->pRaftStore->currentTerm; - pMsg->beginIndex = pSender->snapshotParam.start; - pMsg->lastIndex = pSender->snapshot.lastApplyIndex; - pMsg->lastTerm = pSender->snapshot.lastApplyTerm; - pMsg->lastConfigIndex = pSender->snapshot.lastConfigIndex; - pMsg->lastConfig = pSender->lastConfig; - pMsg->seq = pSender->seq; - // pMsg->privateTerm = pSender->privateTerm; memcpy(pMsg->data, pSender->pCurrentBlock, pSender->blockLen); - - // send msg - syncNodeSendMsgById(&pMsg->destId, pSender->pSyncNode, &rpcMsg); - syncLogSendSyncSnapshotSend(pSender->pSyncNode, pMsg, ""); - - // event log - sSTrace(pSender, "snapshot sender resend"); } + // send msg + syncNodeSendMsgById(&pMsg->destId, pSender->pSyncNode, &rpcMsg); + syncLogSendSyncSnapshotSend(pSender->pSyncNode, pMsg, ""); + + pSender->lastSendTime = taosGetTimestampMs(); + + // event log + sSTrace(pSender, "snapshot sender resend"); + return 0; } @@ -339,6 +345,8 @@ bool snapshotReceiverIsStart(SSyncSnapshotReceiver *pReceiver) { return pReceive void snapshotReceiverForceStop(SSyncSnapshotReceiver *pReceiver) { // force close, abandon incomplete data if (pReceiver->pWriter != NULL) { + // event log + sRTrace(pReceiver, "snapshot receiver force stop"); int32_t ret = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, false, &(pReceiver->snapshot)); ASSERT(ret == 0); @@ -348,7 +356,7 @@ void snapshotReceiverForceStop(SSyncSnapshotReceiver *pReceiver) { pReceiver->start = false; // event log - sRTrace(pReceiver, "snapshot receiver force stop"); + // sRTrace(pReceiver, "snapshot receiver force stop"); } int32_t snapshotReceiverStartWriter(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pBeginMsg) { @@ -669,6 +677,7 @@ static int32_t syncNodeOnSnapshotEnd(SSyncNode *pSyncNode, SyncSnapshotSend *pMs sNTrace(pSyncNode, "snapshot receiver finish waitting for true time, now:%" PRId64 ", stime:%" PRId64, timeNow, pMsg->startTime); taosMsleep(10); + timeNow = taosGetTimestampMs(); } int32_t code = snapshotReceiverFinish(pReceiver, pMsg); diff --git a/source/libs/sync/src/syncTimeout.c b/source/libs/sync/src/syncTimeout.c index 9c39fc1e8e..16e593d0e4 100644 --- a/source/libs/sync/src/syncTimeout.c +++ b/source/libs/sync/src/syncTimeout.c @@ -20,6 +20,7 @@ #include "syncRaftLog.h" #include "syncReplication.h" #include "syncRespMgr.h" +#include "syncSnapshot.h" #include "syncUtil.h" static void syncNodeCleanConfigIndex(SSyncNode* ths) { @@ -70,6 +71,20 @@ static int32_t syncNodeTimerRoutine(SSyncNode* ths) { } int64_t timeNow = taosGetTimestampMs(); + + for (int i = 0; i < ths->peersNum; ++i) { + SSyncSnapshotSender* pSender = syncNodeGetSnapshotSender(ths, &(ths->peersId[i])); + if (pSender != NULL) { + if (ths->isStart && ths->state == TAOS_SYNC_STATE_LEADER && pSender->start && + timeNow - pSender->lastSendTime > SYNC_SNAP_RESEND_MS) { + snapshotReSend(pSender); + } else { + sTrace("vgId:%d, do not resend: nstart%d, now:%" PRId64 ", lstsend:%" PRId64 ", diff:%" PRId64, ths->vgId, + ths->isStart, timeNow, pSender->lastSendTime, timeNow - pSender->lastSendTime); + } + } + } + if (atomic_load_64(&ths->snapshottingIndex) != SYNC_INDEX_INVALID) { // end timeout wal snapshot if (timeNow - ths->snapshottingTime > SYNC_DEL_WAL_MS && diff --git a/source/libs/sync/src/syncUtil.c b/source/libs/sync/src/syncUtil.c index 7787438dfe..a034e6ad83 100644 --- a/source/libs/sync/src/syncUtil.c +++ b/source/libs/sync/src/syncUtil.c @@ -160,8 +160,6 @@ void syncUtilMsgNtoH(void* msg) { bool syncUtilUserPreCommit(tmsg_t msgType) { return msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_LEADER_TRANSFER; } -bool syncUtilUserCommit(tmsg_t msgType) { return msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_LEADER_TRANSFER; } - bool syncUtilUserRollback(tmsg_t msgType) { return msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_LEADER_TRANSFER; } void syncCfg2SimpleStr(const SSyncCfg* pCfg, char* buf, int32_t bufLen) { @@ -568,7 +566,7 @@ void syncLogSendSyncSnapshotSend(SSyncNode* pSyncNode, const SyncSnapshotSend* p syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port); sNTrace(pSyncNode, - "send sync-snapshot-send from %s:%d {term:%" PRId64 ", begin:%" PRId64 ", end:%" PRId64 ", lterm:%" PRId64 + "send sync-snapshot-send to %s:%d {term:%" PRId64 ", begin:%" PRId64 ", end:%" PRId64 ", lterm:%" PRId64 ", stime:%" PRId64 ", seq:%d}, %s", host, port, pMsg->term, pMsg->beginIndex, pMsg->lastIndex, pMsg->lastTerm, pMsg->startTime, pMsg->seq, s); } @@ -595,7 +593,7 @@ void syncLogSendSyncSnapshotRsp(SSyncNode* pSyncNode, const SyncSnapshotRsp* pMs syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port); sNTrace(pSyncNode, - "send sync-snapshot-rsp from %s:%d {term:%" PRId64 ", begin:%" PRId64 ", lst:%" PRId64 ", lterm:%" PRId64 + "send sync-snapshot-rsp to %s:%d {term:%" PRId64 ", begin:%" PRId64 ", lst:%" PRId64 ", lterm:%" PRId64 ", stime:%" PRId64 ", ack:%d}, %s", host, port, pMsg->term, pMsg->snapBeginIndex, pMsg->lastIndex, pMsg->lastTerm, pMsg->startTime, pMsg->ack, s); } diff --git a/source/libs/sync/test/syncAppendEntriesReplyTest.cpp b/source/libs/sync/test/syncAppendEntriesReplyTest.cpp index 1eb803e846..e1a8a5c832 100644 --- a/source/libs/sync/test/syncAppendEntriesReplyTest.cpp +++ b/source/libs/sync/test/syncAppendEntriesReplyTest.cpp @@ -19,7 +19,7 @@ SyncAppendEntriesReply *createMsg() { pMsg->success = true; pMsg->matchIndex = 77; pMsg->term = 33; - pMsg->privateTerm = 44; + // pMsg->privateTerm = 44; pMsg->startTime = taosGetTimestampMs(); return pMsg; } diff --git a/source/libs/sync/test/syncTest.cpp b/source/libs/sync/test/syncTest.cpp index 9ae79e11fb..7b636085f2 100644 --- a/source/libs/sync/test/syncTest.cpp +++ b/source/libs/sync/test/syncTest.cpp @@ -1,5 +1,5 @@ #include "syncTest.h" -#include +// #include /* typedef enum { diff --git a/source/libs/sync/test/sync_test_lib/inc/syncIO.h b/source/libs/sync/test/sync_test_lib/inc/syncIO.h index 19f896182e..98b013b46d 100644 --- a/source/libs/sync/test/sync_test_lib/inc/syncIO.h +++ b/source/libs/sync/test/sync_test_lib/inc/syncIO.h @@ -81,6 +81,8 @@ int32_t syncIOQTimerStop(); int32_t syncIOPingTimerStart(); int32_t syncIOPingTimerStop(); +void syncEntryDestory(SSyncRaftEntry* pEntry); + #ifdef __cplusplus } #endif diff --git a/source/libs/sync/test/sync_test_lib/src/syncIO.c b/source/libs/sync/test/sync_test_lib/src/syncIO.c index 2c24451713..415fecd9f2 100644 --- a/source/libs/sync/test/sync_test_lib/src/syncIO.c +++ b/source/libs/sync/test/sync_test_lib/src/syncIO.c @@ -469,3 +469,5 @@ static void syncIOTickPing(void *param, void *tmrId) { taosTmrReset(syncIOTickPing, io->pingTimerMS, io, io->timerMgr, &io->pingTimer); } + +void syncEntryDestory(SSyncRaftEntry* pEntry) {} \ No newline at end of file diff --git a/source/libs/sync/test/sync_test_lib/src/syncMessageDebug.c b/source/libs/sync/test/sync_test_lib/src/syncMessageDebug.c index 1ea7629601..3b6007df6b 100644 --- a/source/libs/sync/test/sync_test_lib/src/syncMessageDebug.c +++ b/source/libs/sync/test/sync_test_lib/src/syncMessageDebug.c @@ -1583,8 +1583,8 @@ cJSON* syncAppendEntriesReply2Json(const SyncAppendEntriesReply* pMsg) { cJSON_AddNumberToObject(pDestId, "vgId", pMsg->destId.vgId); cJSON_AddItemToObject(pRoot, "destId", pDestId); - snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->privateTerm); - cJSON_AddStringToObject(pRoot, "privateTerm", u64buf); + // snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->privateTerm); + // cJSON_AddStringToObject(pRoot, "privateTerm", u64buf); snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pMsg->term); cJSON_AddStringToObject(pRoot, "term", u64buf); diff --git a/source/libs/tdb/inc/tdb.h b/source/libs/tdb/inc/tdb.h index c728e29641..10a99bb1fa 100644 --- a/source/libs/tdb/inc/tdb.h +++ b/source/libs/tdb/inc/tdb.h @@ -81,6 +81,8 @@ void tdbFree(void *); typedef struct hashset_st *hashset_t; +void hashset_destroy(hashset_t set); + struct STxn { int flags; int64_t txnId; diff --git a/source/libs/tdb/src/db/tdbBtree.c b/source/libs/tdb/src/db/tdbBtree.c index 8c62f89b64..5c1f264460 100644 --- a/source/libs/tdb/src/db/tdbBtree.c +++ b/source/libs/tdb/src/db/tdbBtree.c @@ -110,6 +110,7 @@ int tdbBtreeOpen(int keyLen, int valLen, SPager *pPager, char const *tbname, SPg ret = tdbBegin(pEnv, &txn, tdbDefaultMalloc, tdbDefaultFree, NULL, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED); if (ret < 0) { + tdbOsFree(pBt); return -1; } @@ -119,6 +120,7 @@ int tdbBtreeOpen(int keyLen, int valLen, SPager *pPager, char const *tbname, SPg ret = tdbPagerFetchPage(pPager, &pgno, &pPage, tdbBtreeInitPage, &zArg, txn); if (ret < 0) { tdbAbort(pEnv, txn); + tdbOsFree(pBt); return -1; } @@ -126,6 +128,7 @@ int tdbBtreeOpen(int keyLen, int valLen, SPager *pPager, char const *tbname, SPg if (ret < 0) { tdbError("failed to write page since %s", terrstr()); tdbAbort(pEnv, txn); + tdbOsFree(pBt); return -1; } @@ -138,6 +141,7 @@ int tdbBtreeOpen(int keyLen, int valLen, SPager *pPager, char const *tbname, SPg ret = tdbTbInsert(pPager->pEnv->pMainDb, tbname, strlen(tbname) + 1, &pBt->info, sizeof(pBt->info), txn); if (ret < 0) { tdbAbort(pEnv, txn); + tdbOsFree(pBt); return -1; } } diff --git a/source/libs/tdb/src/db/tdbPager.c b/source/libs/tdb/src/db/tdbPager.c index f638ec25a3..e3d886b046 100644 --- a/source/libs/tdb/src/db/tdbPager.c +++ b/source/libs/tdb/src/db/tdbPager.c @@ -379,9 +379,6 @@ int tdbPagerPostCommit(SPager *pPager, TXN *pTxn) { return -1; } - if (pTxn->jPageSet) { - hashset_destroy(pTxn->jPageSet); - } // pPager->inTran = 0; return 0; @@ -549,8 +546,6 @@ int tdbPagerAbort(SPager *pPager, TXN *pTxn) { return -1; } - hashset_destroy(pTxn->jPageSet); - // pPager->inTran = 0; return 0; @@ -922,6 +917,8 @@ int tdbPagerRestoreJournals(SPager *pPager, SBTree *pBt) { char *name = tdbDirEntryBaseName(tdbGetDirEntryName(pDirEntry)); if (strncmp(TDB_MAINDB_NAME "-journal", name, 16) == 0) { if (tdbPagerRestore(pPager, pBt, name) < 0) { + tdbCloseDir(&pDir); + tdbError("failed to restore file due to %s. jFileName:%s", strerror(errno), name); return -1; } @@ -946,6 +943,8 @@ int tdbPagerRollback(SPager *pPager) { if (strncmp(TDB_MAINDB_NAME "-journal", name, 16) == 0) { if (tdbOsRemove(name) < 0 && errno != ENOENT) { + tdbCloseDir(&pDir); + tdbError("failed to remove file due to %s. jFileName:%s", strerror(errno), name); terrno = TAOS_SYSTEM_ERROR(errno); return -1; diff --git a/source/libs/tdb/src/db/tdbTxn.c b/source/libs/tdb/src/db/tdbTxn.c index 77c87d18f2..055d9c7f98 100644 --- a/source/libs/tdb/src/db/tdbTxn.c +++ b/source/libs/tdb/src/db/tdbTxn.c @@ -30,6 +30,11 @@ int tdbTxnOpen(TXN *pTxn, int64_t txnid, void *(*xMalloc)(void *, size_t), void int tdbTxnClose(TXN *pTxn) { if (pTxn) { + if (pTxn->jPageSet) { + hashset_destroy(pTxn->jPageSet); + pTxn->jPageSet = NULL; + } + tdbOsFree(pTxn); } diff --git a/source/libs/tdb/test/tdbExOVFLTest.cpp b/source/libs/tdb/test/tdbExOVFLTest.cpp index f4f09b20b8..b16bc643d3 100644 --- a/source/libs/tdb/test/tdbExOVFLTest.cpp +++ b/source/libs/tdb/test/tdbExOVFLTest.cpp @@ -8,6 +8,7 @@ #include #include #include +#include "tlog.h" typedef struct SPoolMem { int64_t size; diff --git a/source/libs/tdb/test/tdbTest.cpp b/source/libs/tdb/test/tdbTest.cpp index 54e80a0009..cd02eb8d5e 100644 --- a/source/libs/tdb/test/tdbTest.cpp +++ b/source/libs/tdb/test/tdbTest.cpp @@ -8,6 +8,7 @@ #include #include #include +#include "tlog.h" typedef struct SPoolMem { int64_t size; diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index 50594f1a5a..060d1d3a98 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -1052,9 +1052,6 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) { cliMayCvtFqdnToIp(&pCtx->epSet, &pThrd->cvtAddr); - char tbuf[256] = {0}; - EPSET_DEBUG_STR(&pCtx->epSet, tbuf); - if (!EPSET_IS_VALID(&pCtx->epSet)) { tError("invalid epset"); destroyCmsg(pMsg); @@ -1493,18 +1490,35 @@ FORCE_INLINE bool cliTryExtractEpSet(STransMsg* pResp, SEpSet* dst) { bool cliResetEpset(STransConnCtx* pCtx, STransMsg* pResp, bool hasEpSet) { bool noDelay = true; if (hasEpSet == false) { - // assert(pResp->contLen == 0); if (pResp->contLen == 0) { if (pCtx->epsetRetryCnt >= pCtx->epSet.numOfEps) { noDelay = false; } else { EPSET_FORWARD_INUSE(&pCtx->epSet); } - } else { - if (pCtx->epsetRetryCnt >= pCtx->epSet.numOfEps) { - noDelay = false; + } else if (pResp->contLen != 0) { + SEpSet epSet; + int32_t valid = tDeserializeSEpSet(pResp->pCont, pResp->contLen, &epSet); + if (valid < 0) { + tDebug("get invalid epset, epset equal, continue"); + if (pCtx->epsetRetryCnt >= pCtx->epSet.numOfEps) { + noDelay = false; + } else { + EPSET_FORWARD_INUSE(&pCtx->epSet); + } } else { - EPSET_FORWARD_INUSE(&pCtx->epSet); + if (!transEpSetIsEqual(&pCtx->epSet, &epSet)) { + tDebug("epset not equal, retry new epset"); + pCtx->epSet = epSet; + noDelay = false; + } else { + if (pCtx->epsetRetryCnt >= pCtx->epSet.numOfEps) { + noDelay = false; + } else { + tDebug("epset equal, continue"); + EPSET_FORWARD_INUSE(&pCtx->epSet); + } + } } } } else { @@ -1584,7 +1598,7 @@ bool cliGenRetryRule(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) { addConnToPool(pThrd->pool, pConn); } else if (code == TSDB_CODE_SYN_RESTORING) { tTrace("code str %s, contlen:%d 0", tstrerror(code), pResp->contLen); - noDelay = cliResetEpset(pCtx, pResp, false); + noDelay = cliResetEpset(pCtx, pResp, true); addConnToPool(pThrd->pool, pConn); transFreeMsg(pResp->pCont); } else { diff --git a/source/libs/transport/src/transSvr.c b/source/libs/transport/src/transSvr.c index 73c6dc4011..2b1f68d5f6 100644 --- a/source/libs/transport/src/transSvr.c +++ b/source/libs/transport/src/transSvr.c @@ -1195,6 +1195,8 @@ void transCloseServer(void* arg) { sendQuitToWorkThrd(srv->pThreadObj[i]); destroyWorkThrd(srv->pThreadObj[i]); } + } else { + uv_loop_close(srv->loop); } taosMemoryFree(srv->pThreadObj); diff --git a/source/os/src/osDir.c b/source/os/src/osDir.c index 421901184b..9175067568 100644 --- a/source/os/src/osDir.c +++ b/source/os/src/osDir.c @@ -163,7 +163,7 @@ int32_t taosMulMkDir(const char *dirname) { code = mkdir(temp, 0755); #endif if (code < 0 && errno != EEXIST) { - terrno = TAOS_SYSTEM_ERROR(errno); + // terrno = TAOS_SYSTEM_ERROR(errno); return code; } *pos = TD_DIRSEP[0]; @@ -179,7 +179,7 @@ int32_t taosMulMkDir(const char *dirname) { code = mkdir(temp, 0755); #endif if (code < 0 && errno != EEXIST) { - terrno = TAOS_SYSTEM_ERROR(errno); + // terrno = TAOS_SYSTEM_ERROR(errno); return code; } } @@ -225,7 +225,7 @@ int32_t taosMulModeMkDir(const char *dirname, int mode) { code = mkdir(temp, mode); #endif if (code < 0 && errno != EEXIST) { - terrno = TAOS_SYSTEM_ERROR(errno); + // terrno = TAOS_SYSTEM_ERROR(errno); return code; } *pos = TD_DIRSEP[0]; @@ -241,7 +241,7 @@ int32_t taosMulModeMkDir(const char *dirname, int mode) { code = mkdir(temp, mode); #endif if (code < 0 && errno != EEXIST) { - terrno = TAOS_SYSTEM_ERROR(errno); + // terrno = TAOS_SYSTEM_ERROR(errno); return code; } } @@ -497,3 +497,11 @@ int32_t taosCloseDir(TdDirPtr *ppDir) { return 0; #endif } + +void taosGetCwd(char *buf, int32_t len) { +#if !defined(WINDOWS) + (void)getcwd(buf, len - 1); +#else + strncpy(buf, "not implemented on windows", len -1); +#endif +} diff --git a/source/os/src/osFile.c b/source/os/src/osFile.c index 68365be481..8c2170239f 100644 --- a/source/os/src/osFile.c +++ b/source/os/src/osFile.c @@ -313,7 +313,7 @@ TdFilePtr taosOpenFile(const char *path, int32_t tdFileOptions) { assert(!(tdFileOptions & TD_FILE_EXCL)); fp = fopen(path, mode); if (fp == NULL) { - terrno = TAOS_SYSTEM_ERROR(errno); + // terrno = TAOS_SYSTEM_ERROR(errno); return NULL; } } else { @@ -336,14 +336,14 @@ TdFilePtr taosOpenFile(const char *path, int32_t tdFileOptions) { fd = open(path, access, S_IRWXU | S_IRWXG | S_IRWXO); #endif if (fd == -1) { - terrno = TAOS_SYSTEM_ERROR(errno); + // terrno = TAOS_SYSTEM_ERROR(errno); return NULL; } } TdFilePtr pFile = (TdFilePtr)taosMemoryMalloc(sizeof(TdFile)); if (pFile == NULL) { - terrno = TSDB_CODE_OUT_OF_MEMORY; + // terrno = TSDB_CODE_OUT_OF_MEMORY; if (fd >= 0) close(fd); if (fp != NULL) fclose(fp); return NULL; diff --git a/source/os/src/osMemory.c b/source/os/src/osMemory.c index 230ca5b968..2e68bd5ccd 100644 --- a/source/os/src/osMemory.c +++ b/source/os/src/osMemory.c @@ -348,7 +348,7 @@ void taosMemoryTrim(int32_t size) { void* taosMemoryMallocAlign(uint32_t alignment, int64_t size) { #ifdef USE_TD_MEMORY - ASSERT(0); + assert(0); #else #if defined(LINUX) void* p = memalign(alignment, size); diff --git a/source/os/src/osString.c b/source/os/src/osString.c index e2d8ce95db..f03778de2f 100644 --- a/source/os/src/osString.c +++ b/source/os/src/osString.c @@ -143,15 +143,17 @@ SConv *gConv = NULL; int32_t convUsed = 0; int32_t gConvMaxNum = 0; -void taosConvInit(void) { +int32_t taosConvInit(void) { gConvMaxNum = 512; gConv = taosMemoryCalloc(gConvMaxNum, sizeof(SConv)); for (int32_t i = 0; i < gConvMaxNum; ++i) { gConv[i].conv = iconv_open(DEFAULT_UNICODE_ENCODEC, tsCharset); if ((iconv_t)-1 == gConv[i].conv || (iconv_t)0 == gConv[i].conv) { - ASSERT(0); + return -1; } } + + return 0; } void taosConvDestroy() { diff --git a/source/os/src/osSysinfo.c b/source/os/src/osSysinfo.c index 7ec1da0530..89b84ad395 100644 --- a/source/os/src/osSysinfo.c +++ b/source/os/src/osSysinfo.c @@ -97,6 +97,7 @@ LONG WINAPI exceptionHandler(LPEXCEPTION_POINTERS exception); #include #include +#include #else @@ -275,34 +276,34 @@ int32_t taosGetEmail(char *email, int32_t maxLen) { #endif } + + int32_t taosGetOsReleaseName(char *releaseName, int32_t maxLen) { #ifdef WINDOWS snprintf(releaseName, maxLen, "Windows"); return 0; #elif defined(_TD_DARWIN_64) - char line[1024]; - size_t size = 0; - int32_t code = -1; + char osversion[32]; + size_t osversion_len = sizeof(osversion) - 1; + int osversion_name[] = { CTL_KERN, KERN_OSRELEASE }; - TdFilePtr pFile = taosOpenFile("/etc/os-release", TD_FILE_READ | TD_FILE_STREAM); - if (pFile == NULL) return false; - - while ((size = taosGetsFile(pFile, sizeof(line), line)) != -1) { - line[size - 1] = '\0'; - if (strncmp(line, "PRETTY_NAME", 11) == 0) { - const char *p = strchr(line, '=') + 1; - if (*p == '"') { - p++; - line[size - 2] = 0; - } - tstrncpy(releaseName, p, maxLen); - code = 0; - break; - } + if (sysctl(osversion_name, 2, osversion, &osversion_len, NULL, 0) == -1) { + return -1; } - taosCloseFile(&pFile); - return code; + uint32_t major, minor; + if (sscanf(osversion, "%u.%u", &major, &minor) != 2) { + return -1; + } + if (major >= 20) { + major -= 9; // macOS 11 and newer + sprintf(releaseName, "macOS %u.%u", major, minor); + } else { + major -= 4; // macOS 10.1.1 and newer + sprintf(releaseName, "macOS 10.%d.%d", major, minor); + } + + return 0; #else char line[1024]; size_t size = 0; @@ -617,14 +618,14 @@ int32_t taosGetDiskSize(char *dataDir, SDiskSize *diskSize) { return 0; } else { // printf("failed to get disk size, dataDir:%s errno:%s", tsDataDir, strerror(errno)); - terrno = TAOS_SYSTEM_ERROR(errno); + // terrno = TAOS_SYSTEM_ERROR(errno); return -1; } #elif defined(_TD_DARWIN_64) struct statvfs info; if (statvfs(dataDir, &info)) { // printf("failed to get disk size, dataDir:%s errno:%s", tsDataDir, strerror(errno)); - terrno = TAOS_SYSTEM_ERROR(errno); + // terrno = TAOS_SYSTEM_ERROR(errno); return -1; } else { diskSize->total = info.f_blocks * info.f_frsize; @@ -635,7 +636,7 @@ int32_t taosGetDiskSize(char *dataDir, SDiskSize *diskSize) { #else struct statvfs info; if (statvfs(dataDir, &info)) { - terrno = TAOS_SYSTEM_ERROR(errno); + // terrno = TAOS_SYSTEM_ERROR(errno); return -1; } else { diskSize->total = info.f_blocks * info.f_frsize; diff --git a/source/util/src/talgo.c b/source/util/src/talgo.c index 4d6875d263..057d3a620c 100644 --- a/source/util/src/talgo.c +++ b/source/util/src/talgo.c @@ -15,6 +15,7 @@ #define _DEFAULT_SOURCE #include "talgo.h" +#include "tlog.h" #define doswap(__left, __right, __size, __buf) \ do { \ diff --git a/source/util/src/tjson.c b/source/util/src/tjson.c index cc823decf4..48638af8d5 100644 --- a/source/util/src/tjson.c +++ b/source/util/src/tjson.c @@ -181,7 +181,7 @@ int32_t tjsonGetObjectValueString(const SJson* pJson, char** pValueString) { int32_t tjsonGetStringValue(const SJson* pJson, const char* pName, char* pVal) { char* p = cJSON_GetStringValue(tjsonGetObjectItem((cJSON*)pJson, pName)); if (NULL == p) { - return TSDB_CODE_FAILED; + return TSDB_CODE_SUCCESS; } strcpy(pVal, p); return TSDB_CODE_SUCCESS; @@ -190,7 +190,7 @@ int32_t tjsonGetStringValue(const SJson* pJson, const char* pName, char* pVal) { int32_t tjsonDupStringValue(const SJson* pJson, const char* pName, char** pVal) { char* p = cJSON_GetStringValue(tjsonGetObjectItem((cJSON*)pJson, pName)); if (NULL == p) { - return TSDB_CODE_FAILED; + return TSDB_CODE_SUCCESS; } *pVal = strdup(p); return TSDB_CODE_SUCCESS; @@ -199,7 +199,7 @@ int32_t tjsonDupStringValue(const SJson* pJson, const char* pName, char** pVal) int32_t tjsonGetBigIntValue(const SJson* pJson, const char* pName, int64_t* pVal) { char* p = cJSON_GetStringValue(tjsonGetObjectItem((cJSON*)pJson, pName)); if (NULL == p) { - return TSDB_CODE_FAILED; + return TSDB_CODE_SUCCESS; } #ifdef WINDOWS sscanf(p, "%" PRId64, pVal); @@ -233,7 +233,7 @@ int32_t tjsonGetTinyIntValue(const SJson* pJson, const char* pName, int8_t* pVal int32_t tjsonGetUBigIntValue(const SJson* pJson, const char* pName, uint64_t* pVal) { char* p = cJSON_GetStringValue(tjsonGetObjectItem((cJSON*)pJson, pName)); if (NULL == p) { - return TSDB_CODE_FAILED; + return TSDB_CODE_SUCCESS; } #ifdef WINDOWS sscanf(p, "%" PRIu64, pVal); @@ -259,6 +259,9 @@ int32_t tjsonGetUTinyIntValue(const SJson* pJson, const char* pName, uint8_t* pV int32_t tjsonGetBoolValue(const SJson* pJson, const char* pName, bool* pVal) { const SJson* pObject = tjsonGetObjectItem(pJson, pName); + if (NULL == pObject) { + return TSDB_CODE_SUCCESS; + } if (!cJSON_IsBool(pObject)) { return TSDB_CODE_FAILED; } @@ -268,6 +271,9 @@ int32_t tjsonGetBoolValue(const SJson* pJson, const char* pName, bool* pVal) { int32_t tjsonGetDoubleValue(const SJson* pJson, const char* pName, double* pVal) { const SJson* pObject = tjsonGetObjectItem(pJson, pName); + if (NULL == pObject) { + return TSDB_CODE_SUCCESS; + } if (!cJSON_IsNumber(pObject)) { return TSDB_CODE_FAILED; } @@ -282,7 +288,7 @@ SJson* tjsonGetArrayItem(const SJson* pJson, int32_t index) { return cJSON_GetAr int32_t tjsonToObject(const SJson* pJson, const char* pName, FToObject func, void* pObj) { SJson* pJsonObj = tjsonGetObjectItem(pJson, pName); if (NULL == pJsonObj) { - return TSDB_CODE_FAILED; + return TSDB_CODE_SUCCESS; } return func(pJsonObj, pObj); } @@ -294,7 +300,7 @@ int32_t tjsonMakeObject(const SJson* pJson, const char* pName, FToObject func, v SJson* pJsonObj = tjsonGetObjectItem(pJson, pName); if (NULL == pJsonObj) { - return TSDB_CODE_FAILED; + return TSDB_CODE_SUCCESS; } *pObj = taosMemoryCalloc(1, objSize); if (NULL == *pObj) { diff --git a/source/util/src/tlog.c b/source/util/src/tlog.c index be1db74f1a..3825b91c6d 100644 --- a/source/util/src/tlog.c +++ b/source/util/src/tlog.c @@ -72,6 +72,7 @@ static int32_t tsDaylightActive; /* Currently in daylight saving time. */ bool tsLogEmbedded = 0; bool tsAsyncLog = true; +bool tsAssert = true; int32_t tsNumOfLogLines = 10000000; int32_t tsLogKeepDays = 0; LogFp tsLogFp = NULL; @@ -778,3 +779,37 @@ cmp_end: return ret; } + +bool taosAssert(bool condition, const char *file, int32_t line, const char *format, ...) { + if (condition) return false; + + const char *flags = "UTL FATAL "; + ELogLevel level = DEBUG_FATAL; + int32_t dflag = 255; // tsLogEmbedded ? 255 : uDebugFlag + char buffer[LOG_MAX_LINE_BUFFER_SIZE]; + int32_t len = taosBuildLogHead(buffer, flags); + + va_list argpointer; + va_start(argpointer, format); + len = len + vsnprintf(buffer + len, LOG_MAX_LINE_BUFFER_SIZE - len, format, argpointer); + va_end(argpointer); + buffer[len++] = '\n'; + buffer[len] = 0; + taosPrintLogImp(1, 255, buffer, len); + + taosPrintLog(flags, level, dflag, "tAssert at file %s:%d exit:%d", file, line, tsAssert); + taosPrintTrace(flags, level, dflag); + + if (tsAssert) { + taosCloseLog(); + taosMsleep(300); + +#ifdef NDEBUG + abort(); +#else + assert(0); +#endif + } + + return true; +} \ No newline at end of file diff --git a/source/util/test/hashTest.cpp b/source/util/test/hashTest.cpp index 76fb9c7067..00379a755b 100644 --- a/source/util/test/hashTest.cpp +++ b/source/util/test/hashTest.cpp @@ -6,6 +6,7 @@ #include "taos.h" #include "taosdef.h" #include "thash.h" +#include "tlog.h" namespace { diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task index c90b2499a4..28ba7ea60f 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -1022,9 +1022,9 @@ ,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/auto_create_table_json.py ,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/custom_col_tag.py ,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/default_json.py -#,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/demo.py +,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/demo.py ,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/insert_alltypes_json.py -#,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/invalid_commandline.py +,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/invalid_commandline.py ,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/json_tag.py ,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/query_json.py ,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/sample_csv_json.py diff --git a/tests/script/tsim/db/alter_replica_13.sim b/tests/script/tsim/db/alter_replica_13.sim index 4eafb86198..d75acb50ad 100644 --- a/tests/script/tsim/db/alter_replica_13.sim +++ b/tests/script/tsim/db/alter_replica_13.sim @@ -116,6 +116,25 @@ endi print ============= step4: alter database sql alter database db replica 3 +$wt = 0 +stepwt1: + $wt = $wt + 1 + sleep 1000 + if $wt == 200 then + print ====> dnode not ready! + return -1 + endi +sql show transactions +if $rows != 0 then + print wait 1 seconds to alter + goto stepwt1 +endi + +sql show db.vgroups +print ---> $data00 $data01 $data02 $data03 $data04 $data05 $data06 $data07 $data08 $data09 +print ---> $data10 $data11 $data12 $data13 $data14 $data15 $data16 $data27 $data28 $data29 +print ---> $data10 $data11 $data12 $data13 $data14 $data15 $data26 $data37 $data38 $data39 +print ---> $data10 $data11 $data12 $data13 $data14 $data15 $data36 $data47 $data48 $data49 $leaderIndex = 0 diff --git a/tests/script/tsim/db/alter_replica_31.sim b/tests/script/tsim/db/alter_replica_31.sim index 47e1fda79f..74e8007791 100644 --- a/tests/script/tsim/db/alter_replica_31.sim +++ b/tests/script/tsim/db/alter_replica_31.sim @@ -148,6 +148,26 @@ endi print ============= step3: alter database sql alter database db replica 1 +$wt = 0 + stepwt1: + $wt = $wt + 1 + sleep 1000 + if $wt == 200 then + print ====> dnode not ready! + return -1 + endi +sql show transactions +if $rows != 0 then + print wait 1 seconds to alter + goto stepwt1 +endi + +sql show db.vgroups +print ---> $data00 $data01 $data02 $data03 $data04 $data05 $data06 $data07 $data08 $data09 +print ---> $data10 $data11 $data12 $data13 $data14 $data15 $data16 $data27 $data28 $data29 +print ---> $data10 $data11 $data12 $data13 $data14 $data15 $data26 $data37 $data38 $data39 +print ---> $data10 $data11 $data12 $data13 $data14 $data15 $data36 $data47 $data48 $data49 + $hasleader = 0 $x = 0 diff --git a/tools/shell/src/shellWebsocket.c b/tools/shell/src/shellWebsocket.c index 94bb909e29..add3ccc51f 100644 --- a/tools/shell/src/shellWebsocket.c +++ b/tools/shell/src/shellWebsocket.c @@ -223,7 +223,7 @@ void shellRunSingleCommandWebsocketImp(char *command) { if (code == TSDB_CODE_WS_SEND_TIMEOUT || code == TSDB_CODE_WS_RECV_TIMEOUT) { fprintf(stderr, "Hint: use -t to increase the timeout in seconds\n"); } else if (code == TSDB_CODE_WS_INTERNAL_ERRO || code == TSDB_CODE_WS_CLOSED) { - fprintf(stderr, "TDengine server is down, will try to reconnect\n"); + fprintf(stderr, "TDengine server is disconnected, will try to reconnect\n"); shell.ws_conn = NULL; } ws_free_result(res); diff --git a/utils/test/c/sml_test.c b/utils/test/c/sml_test.c index cd0587bfa4..2597a26243 100644 --- a/utils/test/c/sml_test.c +++ b/utils/test/c/sml_test.c @@ -20,6 +20,7 @@ #include #include "taos.h" #include "types.h" +#include "tlog.h" int smlProcess_influx_Test() { TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);