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

Merge branch '3.0' into enh/refactorBackend

This commit is contained in:
yihaoDeng 2023-11-23 17:41:37 +08:00
parent a2a68ce256 421a154d2a
commit 6129ebc294
40 changed files with 1026 additions and 343 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
cmake/cmake.define
View File

@ -97,7 +97,15 @@ ENDIF()
SET(JEMALLOC_ENABLED OFF) SET(JEMALLOC_ENABLED OFF)
IF (TD_WINDOWS) IF (TD_WINDOWS)
MESSAGE("${Yellow} set compiler flag for Windows! ${ColourReset}") MESSAGE("${Yellow} set compiler flag for Windows! ${ColourReset}")
IF (${CMAKE_BUILD_TYPE} MATCHES "Release")
MESSAGE("${Green} will build Release version! ${ColourReset}")
SET(COMMON_FLAGS "/W3 /D_WIN32 /DWIN32 /Zi- /O2 /GL /MD")
ELSE ()
MESSAGE("${Green} will build Debug version! ${ColourReset}")
SET(COMMON_FLAGS "/w /D_WIN32 /DWIN32 /Zi /MTd") SET(COMMON_FLAGS "/w /D_WIN32 /DWIN32 /Zi /MTd")
ENDIF()
SET(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} /MANIFEST:NO") SET(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} /MANIFEST:NO")
# IF (MSVC AND (MSVC_VERSION GREATER_EQUAL 1900)) # IF (MSVC AND (MSVC_VERSION GREATER_EQUAL 1900))
# SET(COMMON_FLAGS "${COMMON_FLAGS} /Wv:18") # SET(COMMON_FLAGS "${COMMON_FLAGS} /Wv:18")

26
contrib/CMakeLists.txt
View File

@ -317,7 +317,8 @@ if (${BUILD_WITH_ROCKSDB})
SET(CMAKE_BUILD_TYPE Release) SET(CMAKE_BUILD_TYPE Release)
endif() endif()
endif(${TD_LINUX}) endif(${TD_LINUX})
MESSAGE(STATUS "CXXXX STATUS CONFIG: " ${CMAKE_CXX_FLAGS}) MESSAGE(STATUS "ROCKSDB CXXXX STATUS CONFIG: " ${CMAKE_CXX_FLAGS})
MESSAGE(STATUS "ROCKSDB C STATUS CONFIG: " ${CMAKE_C_FLAGS})
if(${TD_DARWIN}) if(${TD_DARWIN})
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-error=maybe-uninitialized") SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-error=maybe-uninitialized")
@ -329,8 +330,12 @@ if (${BUILD_WITH_ROCKSDB})
if (${TD_WINDOWS}) if (${TD_WINDOWS})
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /wd4244 /wd4819") SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /wd4244 /wd4819")
SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} /wd4244 /wd4819")
option(WITH_JNI "" OFF) option(WITH_JNI "" OFF)
if(CMAKE_C_FLAGS MATCHES "/MT" OR CMAKE_C_FLAGS MATCHES "/MTd")
message("Rocksdb build runtime lib use /MT or /MTd")
option(WITH_MD_LIBRARY "build with MD" OFF) option(WITH_MD_LIBRARY "build with MD" OFF)
endif()
set(SYSTEM_LIBS ${SYSTEM_LIBS} shlwapi.lib rpcrt4.lib) set(SYSTEM_LIBS ${SYSTEM_LIBS} shlwapi.lib rpcrt4.lib)
endif(${TD_WINDOWS}) endif(${TD_WINDOWS})
@ -361,9 +366,11 @@ if (${BUILD_WITH_ROCKSDB})
) )
else() else()
if (NOT ${TD_LINUX}) if (NOT ${TD_LINUX})
MESSAGE(STATUS "CXXXX STATUS CONFIG: " ${CMAKE_CXX_FLAGS}) MESSAGE(STATUS "ROCKSDB CXX STATUS CONFIG: " ${CMAKE_CXX_FLAGS})
MESSAGE(STATUS "ROCKSDB C STATUS CONFIG: " ${CMAKE_C_FLAGS})
if(${TD_DARWIN}) if(${TD_DARWIN})
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-error=maybe-uninitialized") SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-error=maybe-uninitialized")
SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Wno-error=maybe-uninitialized")
endif(${TD_DARWIN}) endif(${TD_DARWIN})
if (${TD_DARWIN_ARM64}) if (${TD_DARWIN_ARM64})
@ -372,8 +379,12 @@ if (${BUILD_WITH_ROCKSDB})
if (${TD_WINDOWS}) if (${TD_WINDOWS})
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /wd4244 /wd4819") SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /wd4244 /wd4819")
SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} /wd4244 /wd4819")
option(WITH_JNI "" OFF) option(WITH_JNI "" OFF)
if(CMAKE_C_FLAGS MATCHES "/MT" OR CMAKE_C_FLAGS MATCHES "/MTd")
message("Rocksdb build runtime lib use /MT or /MTd")
option(WITH_MD_LIBRARY "build with MD" OFF) option(WITH_MD_LIBRARY "build with MD" OFF)
endif()
set(SYSTEM_LIBS ${SYSTEM_LIBS} shlwapi.lib rpcrt4.lib) set(SYSTEM_LIBS ${SYSTEM_LIBS} shlwapi.lib rpcrt4.lib)
endif(${TD_WINDOWS}) endif(${TD_WINDOWS})
@ -456,7 +467,9 @@ endif(${BUILD_WITH_NURAFT})
# pthread # pthread
if(${BUILD_PTHREAD}) if(${BUILD_PTHREAD})
set(CMAKE_BUILD_TYPE debug) if ("${CMAKE_BUILD_TYPE}" STREQUAL "")
SET(CMAKE_BUILD_TYPE Release)
endif()
add_definitions(-DPTW32_STATIC_LIB) add_definitions(-DPTW32_STATIC_LIB)
add_subdirectory(pthread EXCLUDE_FROM_ALL) add_subdirectory(pthread EXCLUDE_FROM_ALL)
set_target_properties(libpthreadVC3 PROPERTIES OUTPUT_NAME pthread) set_target_properties(libpthreadVC3 PROPERTIES OUTPUT_NAME pthread)
@ -640,13 +653,18 @@ if(${BUILD_GEOS})
if(${TD_LINUX}) if(${TD_LINUX})
set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS_REL}") set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS_REL}")
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS_REL}") set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS_REL}")
IF ("${CMAKE_BUILD_TYPE}" STREQUAL "") if ("${CMAKE_BUILD_TYPE}" STREQUAL "")
SET(CMAKE_BUILD_TYPE Release) SET(CMAKE_BUILD_TYPE Release)
endif() endif()
endif(${TD_LINUX}) endif(${TD_LINUX})
option(BUILD_SHARED_LIBS "Build GEOS with shared libraries" OFF) option(BUILD_SHARED_LIBS "Build GEOS with shared libraries" OFF)
add_subdirectory(geos EXCLUDE_FROM_ALL) add_subdirectory(geos EXCLUDE_FROM_ALL)
if (${TD_WINDOWS})
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS}")
SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS}")
else ()
unset(CMAKE_CXX_STANDARD CACHE) # undo libgeos's setting of global CMAKE_CXX_STANDARD unset(CMAKE_CXX_STANDARD CACHE) # undo libgeos's setting of global CMAKE_CXX_STANDARD
endif(${TD_WINDOWS})
target_include_directories( target_include_directories(
geos_c geos_c
PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/geos/include> PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/geos/include>

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

@ -34,11 +34,6 @@ extern "C" {
#define SIZE_IN_MiB(_v) ((_v) / ONE_MiB_F) #define SIZE_IN_MiB(_v) ((_v) / ONE_MiB_F)
#define SIZE_IN_KiB(_v) ((_v) / ONE_KiB_F) #define SIZE_IN_KiB(_v) ((_v) / ONE_KiB_F)
#define TASK_DOWNSTREAM_READY 0x0
#define TASK_DOWNSTREAM_NOT_READY 0x1
#define TASK_DOWNSTREAM_NOT_LEADER 0x2
#define TASK_SELF_NEW_STAGE 0x3
#define TASK_DOWNSTREAM_READY 0x0 #define TASK_DOWNSTREAM_READY 0x0
#define TASK_DOWNSTREAM_NOT_READY 0x1 #define TASK_DOWNSTREAM_NOT_READY 0x1
#define TASK_DOWNSTREAM_NOT_LEADER 0x2 #define TASK_DOWNSTREAM_NOT_LEADER 0x2
@ -315,6 +310,7 @@ typedef struct SCheckpointInfo {
int64_t checkpointingId; int64_t checkpointingId;
int32_t downstreamAlignNum; int32_t downstreamAlignNum;
int32_t checkpointNotReadyTasks; int32_t checkpointNotReadyTasks;
bool dispatchCheckpointTrigger;
int64_t msgVer; int64_t msgVer;
} SCheckpointInfo; } SCheckpointInfo;
@ -466,7 +462,6 @@ typedef struct STaskStartInfo {
int64_t startTs; int64_t startTs;
int64_t readyTs; int64_t readyTs;
int32_t tasksWillRestart; int32_t tasksWillRestart;
int32_t taskStarting; // restart flag, sentinel to guard the restart procedure. int32_t taskStarting; // restart flag, sentinel to guard the restart procedure.
SHashObj* pReadyTaskSet; // tasks that are all ready for running stream processing SHashObj* pReadyTaskSet; // tasks that are all ready for running stream processing
SHashObj* pFailedTaskSet; // tasks that are done the check downstream process, may be successful or failed SHashObj* pFailedTaskSet; // tasks that are done the check downstream process, may be successful or failed
@ -774,7 +769,8 @@ void initRpcMsg(SRpcMsg* pMsg, int32_t msgType, void* pCont, int32_t contLen);
// recover and fill history // recover and fill history
void streamTaskCheckDownstream(SStreamTask* pTask); void streamTaskCheckDownstream(SStreamTask* pTask);
int32_t streamTaskCheckStatus(SStreamTask* pTask, int32_t upstreamTaskId, int32_t vgId, int64_t stage, int64_t* oldStage); int32_t streamTaskCheckStatus(SStreamTask* pTask, int32_t upstreamTaskId, int32_t vgId, int64_t stage,
int64_t* oldStage);
int32_t streamTaskUpdateEpsetInfo(SStreamTask* pTask, SArray* pNodeList); int32_t streamTaskUpdateEpsetInfo(SStreamTask* pTask, SArray* pNodeList);
void streamTaskResetUpstreamStageInfo(SStreamTask* pTask); void streamTaskResetUpstreamStageInfo(SStreamTask* pTask);
bool streamTaskAllUpstreamClosed(SStreamTask* pTask); bool streamTaskAllUpstreamClosed(SStreamTask* pTask);
@ -803,7 +799,6 @@ int32_t streamQueueGetNumOfItems(const SStreamQueue* pQueue);
int32_t streamRestoreParam(SStreamTask* pTask); int32_t streamRestoreParam(SStreamTask* pTask);
void streamTaskPause(SStreamTask* pTask, SStreamMeta* pMeta); void streamTaskPause(SStreamTask* pTask, SStreamMeta* pMeta);
void streamTaskResume(SStreamTask* pTask); void streamTaskResume(SStreamTask* pTask);
void streamTaskEnablePause(SStreamTask* pTask);
int32_t streamTaskSetUpstreamInfo(SStreamTask* pTask, const SStreamTask* pUpstreamTask); int32_t streamTaskSetUpstreamInfo(SStreamTask* pTask, const SStreamTask* pUpstreamTask);
void streamTaskUpdateUpstreamInfo(SStreamTask* pTask, int32_t nodeId, const SEpSet* pEpSet); void streamTaskUpdateUpstreamInfo(SStreamTask* pTask, int32_t nodeId, const SEpSet* pEpSet);
void streamTaskUpdateDownstreamInfo(SStreamTask* pTask, int32_t nodeId, const SEpSet* pEpSet); void streamTaskUpdateDownstreamInfo(SStreamTask* pTask, int32_t nodeId, const SEpSet* pEpSet);
@ -859,7 +854,7 @@ void streamMetaResetStartInfo(STaskStartInfo* pMeta);
int32_t streamProcessCheckpointSourceReq(SStreamTask* pTask, SStreamCheckpointSourceReq* pReq); int32_t streamProcessCheckpointSourceReq(SStreamTask* pTask, SStreamCheckpointSourceReq* pReq);
int32_t streamProcessCheckpointReadyMsg(SStreamTask* pTask); int32_t streamProcessCheckpointReadyMsg(SStreamTask* pTask);
int32_t streamTaskBuildCheckpoint(SStreamTask* pTask); int32_t streamTaskBuildCheckpoint(SStreamTask* pTask);
void streamTaskClearCheckInfo(SStreamTask* pTask); void streamTaskClearCheckInfo(SStreamTask* pTask, bool clearChkpReadyMsg);
int32_t streamAlignTransferState(SStreamTask* pTask); int32_t streamAlignTransferState(SStreamTask* pTask);
int32_t streamBuildAndSendDropTaskMsg(SMsgCb* pMsgCb, int32_t vgId, SStreamTaskId* pTaskId); int32_t streamBuildAndSendDropTaskMsg(SMsgCb* pMsgCb, int32_t vgId, SStreamTaskId* pTaskId);
int32_t streamAddCheckpointSourceRspMsg(SStreamCheckpointSourceReq* pReq, SRpcHandleInfo* pRpcInfo, SStreamTask* pTask, int32_t streamAddCheckpointSourceRspMsg(SStreamCheckpointSourceReq* pReq, SRpcHandleInfo* pRpcInfo, SStreamTask* pTask,

1
include/os/osFile.h
View File

@ -63,6 +63,7 @@ typedef struct TdFile *TdFilePtr;
#define TD_FILE_AUTO_DEL 0x0040 #define TD_FILE_AUTO_DEL 0x0040
#define TD_FILE_EXCL 0x0080 #define TD_FILE_EXCL 0x0080
#define TD_FILE_STREAM 0x0100 // Only support taosFprintfFile, taosGetLineFile, taosEOFFile #define TD_FILE_STREAM 0x0100 // Only support taosFprintfFile, taosGetLineFile, taosEOFFile
#define TD_FILE_WRITE_THROUGH 0x0200
TdFilePtr taosOpenFile(const char *path, int32_t tdFileOptions); TdFilePtr taosOpenFile(const char *path, int32_t tdFileOptions);
TdFilePtr taosCreateFile(const char *path, int32_t tdFileOptions); TdFilePtr taosCreateFile(const char *path, int32_t tdFileOptions);

6
source/common/src/tdataformat.c
View File

@ -3590,9 +3590,9 @@ void (*tColDataCalcSMA[])(SColData *pColData, int64_t *sum, int64_t *max, int64_
tColDataCalcSMAUInt, // TSDB_DATA_TYPE_UINT tColDataCalcSMAUInt, // TSDB_DATA_TYPE_UINT
tColDataCalcSMAUBigInt, // TSDB_DATA_TYPE_UBIGINT tColDataCalcSMAUBigInt, // TSDB_DATA_TYPE_UBIGINT
tColDataCalcSMAVarType, // TSDB_DATA_TYPE_JSON tColDataCalcSMAVarType, // TSDB_DATA_TYPE_JSON
NULL, // TSDB_DATA_TYPE_VARBINARY tColDataCalcSMAVarType, // TSDB_DATA_TYPE_VARBINARY
NULL, // TSDB_DATA_TYPE_DECIMAL tColDataCalcSMAVarType, // TSDB_DATA_TYPE_DECIMAL
NULL, // TSDB_DATA_TYPE_BLOB tColDataCalcSMAVarType, // TSDB_DATA_TYPE_BLOB
NULL, // TSDB_DATA_TYPE_MEDIUMBLOB NULL, // TSDB_DATA_TYPE_MEDIUMBLOB
tColDataCalcSMAVarType // TSDB_DATA_TYPE_GEOMETRY tColDataCalcSMAVarType // TSDB_DATA_TYPE_GEOMETRY
}; };

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

@ -354,11 +354,16 @@ static int32_t taosLoadCfg(SConfig *pCfg, const char **envCmd, const char *input
char cfgFile[PATH_MAX + 100] = {0}; char cfgFile[PATH_MAX + 100] = {0};
taosExpandDir(inputCfgDir, cfgDir, PATH_MAX); taosExpandDir(inputCfgDir, cfgDir, PATH_MAX);
char lastC = cfgDir[strlen(cfgDir) - 1];
char *tdDirsep = TD_DIRSEP;
if (lastC == '\\' || lastC == '/') {
tdDirsep = "";
}
if (taosIsDir(cfgDir)) { if (taosIsDir(cfgDir)) {
#ifdef CUS_PROMPT #ifdef CUS_PROMPT
snprintf(cfgFile, sizeof(cfgFile), "%s" TD_DIRSEP "%s.cfg", cfgDir, CUS_PROMPT); snprintf(cfgFile, sizeof(cfgFile), "%s" "%s" "%s.cfg", cfgDir, tdDirsep, CUS_PROMPT);
#else #else
snprintf(cfgFile, sizeof(cfgFile), "%s" TD_DIRSEP "taos.cfg", cfgDir); snprintf(cfgFile, sizeof(cfgFile), "%s" "%s" "taos.cfg", cfgDir, tdDirsep);
#endif #endif
} else { } else {
tstrncpy(cfgFile, cfgDir, sizeof(cfgDir)); tstrncpy(cfgFile, cfgDir, sizeof(cfgDir));
@ -508,6 +513,7 @@ static int32_t taosAddClientCfg(SConfig *pCfg) {
tsNumOfTaskQueueThreads = tsNumOfCores / 2; tsNumOfTaskQueueThreads = tsNumOfCores / 2;
tsNumOfTaskQueueThreads = TMAX(tsNumOfTaskQueueThreads, 4); tsNumOfTaskQueueThreads = TMAX(tsNumOfTaskQueueThreads, 4);
if (tsNumOfTaskQueueThreads >= 50) { if (tsNumOfTaskQueueThreads >= 50) {
tsNumOfTaskQueueThreads = 50; tsNumOfTaskQueueThreads = 50;
} }

2
source/dnode/mgmt/mgmt_mnode/src/mmFile.c
View File

@ -169,7 +169,7 @@ int32_t mmWriteFile(const char *path, const SMnodeOpt *pOption) {
if (buffer == NULL) goto _OVER; if (buffer == NULL) goto _OVER;
terrno = 0; terrno = 0;
pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC); pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
if (pFile == NULL) goto _OVER; if (pFile == NULL) goto _OVER;
int32_t len = strlen(buffer); int32_t len = strlen(buffer);

2
source/dnode/mgmt/mgmt_vnode/inc/vmInt.h
View File

@ -84,8 +84,10 @@ typedef struct {
} SVnodeThread; } SVnodeThread;
// vmInt.c // vmInt.c
int32_t vmGetPrimaryDisk(SVnodeMgmt *pMgmt, int32_t vgId);
int32_t vmAllocPrimaryDisk(SVnodeMgmt *pMgmt, int32_t vgId); int32_t vmAllocPrimaryDisk(SVnodeMgmt *pMgmt, int32_t vgId);
SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId); SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId);
SVnodeObj *vmAcquireVnodeImpl(SVnodeMgmt *pMgmt, int32_t vgId, bool strict);
void vmReleaseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode); void vmReleaseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode);
int32_t vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl); int32_t vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl);
void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode, bool commitAndRemoveWal); void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode, bool commitAndRemoveWal);

2
source/dnode/mgmt/mgmt_vnode/src/vmFile.c
View File

@ -200,7 +200,7 @@ int32_t vmWriteVnodeListToFile(SVnodeMgmt *pMgmt) {
if (buffer == NULL) goto _OVER; if (buffer == NULL) goto _OVER;
terrno = 0; terrno = 0;
pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC); pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
if (pFile == NULL) goto _OVER; if (pFile == NULL) goto _OVER;
int32_t len = strlen(buffer); int32_t len = strlen(buffer);

23
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
View File

@ -282,7 +282,7 @@ int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
vmGenerateWrapperCfg(pMgmt, &req, &wrapperCfg); vmGenerateWrapperCfg(pMgmt, &req, &wrapperCfg);
SVnodeObj *pVnode = vmAcquireVnode(pMgmt, req.vgId); SVnodeObj *pVnode = vmAcquireVnode(pMgmt, req.vgId);
if (pVnode != NULL && !pVnode->failed) { if (pVnode != NULL) {
dError("vgId:%d, already exist", req.vgId); dError("vgId:%d, already exist", req.vgId);
tFreeSCreateVnodeReq(&req); tFreeSCreateVnodeReq(&req);
vmReleaseVnode(pMgmt, pVnode); vmReleaseVnode(pMgmt, pVnode);
@ -291,10 +291,11 @@ int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
return 0; return 0;
} }
ASSERT(pVnode == NULL || pVnode->failed); int32_t diskPrimary = vmGetPrimaryDisk(pMgmt, vnodeCfg.vgId);
if (diskPrimary < 0) {
wrapperCfg.diskPrimary = pVnode ? pVnode->diskPrimary : vmAllocPrimaryDisk(pMgmt, vnodeCfg.vgId); diskPrimary = vmAllocPrimaryDisk(pMgmt, vnodeCfg.vgId);
int32_t diskPrimary = wrapperCfg.diskPrimary; }
wrapperCfg.diskPrimary = diskPrimary;
snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, vnodeCfg.vgId); snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, vnodeCfg.vgId);
@ -371,7 +372,7 @@ int32_t vmProcessAlterVnodeTypeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
TMSG_INFO(pMsg->msgType)); TMSG_INFO(pMsg->msgType));
SVnodeObj *pVnode = vmAcquireVnode(pMgmt, req.vgId); SVnodeObj *pVnode = vmAcquireVnode(pMgmt, req.vgId);
if (pVnode == NULL || pVnode->failed) { if (pVnode == NULL) {
dError("vgId:%d, failed to alter vnode type since %s", req.vgId, terrstr()); dError("vgId:%d, failed to alter vnode type since %s", req.vgId, terrstr());
terrno = TSDB_CODE_VND_NOT_EXIST; terrno = TSDB_CODE_VND_NOT_EXIST;
if (pVnode) vmReleaseVnode(pMgmt, pVnode); if (pVnode) vmReleaseVnode(pMgmt, pVnode);
@ -489,7 +490,7 @@ int32_t vmProcessCheckLearnCatchupReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
req.vgId, TMSG_INFO(pMsg->msgType)); req.vgId, TMSG_INFO(pMsg->msgType));
SVnodeObj *pVnode = vmAcquireVnode(pMgmt, req.vgId); SVnodeObj *pVnode = vmAcquireVnode(pMgmt, req.vgId);
if (pVnode == NULL || pVnode->failed) { if (pVnode == NULL) {
dError("vgId:%d, failed to alter vnode type since %s", req.vgId, terrstr()); dError("vgId:%d, failed to alter vnode type since %s", req.vgId, terrstr());
terrno = TSDB_CODE_VND_NOT_EXIST; terrno = TSDB_CODE_VND_NOT_EXIST;
if (pVnode) vmReleaseVnode(pMgmt, pVnode); if (pVnode) vmReleaseVnode(pMgmt, pVnode);
@ -532,7 +533,7 @@ int32_t vmProcessDisableVnodeWriteReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
dInfo("vgId:%d, vnode write disable:%d", req.vgId, req.disable); dInfo("vgId:%d, vnode write disable:%d", req.vgId, req.disable);
SVnodeObj *pVnode = vmAcquireVnode(pMgmt, req.vgId); SVnodeObj *pVnode = vmAcquireVnode(pMgmt, req.vgId);
if (pVnode == NULL || pVnode->failed) { if (pVnode == NULL) {
dError("vgId:%d, failed to disable write since %s", req.vgId, terrstr()); dError("vgId:%d, failed to disable write since %s", req.vgId, terrstr());
terrno = TSDB_CODE_VND_NOT_EXIST; terrno = TSDB_CODE_VND_NOT_EXIST;
if (pVnode) vmReleaseVnode(pMgmt, pVnode); if (pVnode) vmReleaseVnode(pMgmt, pVnode);
@ -565,7 +566,7 @@ int32_t vmProcessAlterHashRangeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
dInfo("vgId:%d, start to alter vnode hashrange:[%u, %u], dstVgId:%d", req.srcVgId, req.hashBegin, req.hashEnd, dInfo("vgId:%d, start to alter vnode hashrange:[%u, %u], dstVgId:%d", req.srcVgId, req.hashBegin, req.hashEnd,
req.dstVgId); req.dstVgId);
pVnode = vmAcquireVnode(pMgmt, srcVgId); pVnode = vmAcquireVnode(pMgmt, srcVgId);
if (pVnode == NULL || pVnode->failed) { if (pVnode == NULL) {
dError("vgId:%d, failed to alter hashrange since %s", srcVgId, terrstr()); dError("vgId:%d, failed to alter hashrange since %s", srcVgId, terrstr());
terrno = TSDB_CODE_VND_NOT_EXIST; terrno = TSDB_CODE_VND_NOT_EXIST;
if (pVnode) vmReleaseVnode(pMgmt, pVnode); if (pVnode) vmReleaseVnode(pMgmt, pVnode);
@ -680,7 +681,7 @@ int32_t vmProcessAlterVnodeReplicaReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
} }
SVnodeObj *pVnode = vmAcquireVnode(pMgmt, vgId); SVnodeObj *pVnode = vmAcquireVnode(pMgmt, vgId);
if (pVnode == NULL || pVnode->failed) { if (pVnode == NULL) {
dError("vgId:%d, failed to alter replica since %s", vgId, terrstr()); dError("vgId:%d, failed to alter replica since %s", vgId, terrstr());
terrno = TSDB_CODE_VND_NOT_EXIST; terrno = TSDB_CODE_VND_NOT_EXIST;
if (pVnode) vmReleaseVnode(pMgmt, pVnode); if (pVnode) vmReleaseVnode(pMgmt, pVnode);
@ -748,7 +749,7 @@ int32_t vmProcessDropVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
return -1; return -1;
} }
SVnodeObj *pVnode = vmAcquireVnode(pMgmt, vgId); SVnodeObj *pVnode = vmAcquireVnodeImpl(pMgmt, vgId, false);
if (pVnode == NULL) { if (pVnode == NULL) {
dInfo("vgId:%d, failed to drop since %s", vgId, terrstr()); dInfo("vgId:%d, failed to drop since %s", vgId, terrstr());
terrno = TSDB_CODE_VND_NOT_EXIST; terrno = TSDB_CODE_VND_NOT_EXIST;

39
source/dnode/mgmt/mgmt_vnode/src/vmInt.c
View File

@ -19,6 +19,19 @@
#include "vnd.h" #include "vnd.h"
#include "libs/function/tudf.h" #include "libs/function/tudf.h"
int32_t vmGetPrimaryDisk(SVnodeMgmt *pMgmt, int32_t vgId) {
int32_t diskId = -1;
SVnodeObj *pVnode = NULL;
taosThreadRwlockRdlock(&pMgmt->lock);
taosHashGetDup(pMgmt->hash, &vgId, sizeof(int32_t), (void *)&pVnode);
if (pVnode != NULL) {
diskId = pVnode->diskPrimary;
}
taosThreadRwlockUnlock(&pMgmt->lock);
return diskId;
}
int32_t vmAllocPrimaryDisk(SVnodeMgmt *pMgmt, int32_t vgId) { int32_t vmAllocPrimaryDisk(SVnodeMgmt *pMgmt, int32_t vgId) {
STfs *pTfs = pMgmt->pTfs; STfs *pTfs = pMgmt->pTfs;
int32_t diskId = 0; int32_t diskId = 0;
@ -74,12 +87,12 @@ int32_t vmAllocPrimaryDisk(SVnodeMgmt *pMgmt, int32_t vgId) {
return diskId; return diskId;
} }
SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId) { SVnodeObj *vmAcquireVnodeImpl(SVnodeMgmt *pMgmt, int32_t vgId, bool strict) {
SVnodeObj *pVnode = NULL; SVnodeObj *pVnode = NULL;
taosThreadRwlockRdlock(&pMgmt->lock); taosThreadRwlockRdlock(&pMgmt->lock);
taosHashGetDup(pMgmt->hash, &vgId, sizeof(int32_t), (void *)&pVnode); taosHashGetDup(pMgmt->hash, &vgId, sizeof(int32_t), (void *)&pVnode);
if (pVnode == NULL || pVnode->dropped) { if (pVnode == NULL || strict && (pVnode->dropped || pVnode->failed)) {
terrno = TSDB_CODE_VND_INVALID_VGROUP_ID; terrno = TSDB_CODE_VND_INVALID_VGROUP_ID;
pVnode = NULL; pVnode = NULL;
} else { } else {
@ -91,6 +104,8 @@ SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId) {
return pVnode; return pVnode;
} }
SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId) { return vmAcquireVnodeImpl(pMgmt, vgId, true); }
void vmReleaseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) { void vmReleaseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
if (pVnode == NULL) return; if (pVnode == NULL) return;
@ -100,6 +115,15 @@ void vmReleaseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
taosThreadRwlockUnlock(&pMgmt->lock); taosThreadRwlockUnlock(&pMgmt->lock);
} }
static void vmFreeVnodeObj(SVnodeObj **ppVnode) {
if (!ppVnode || !(*ppVnode)) return;
SVnodeObj *pVnode = *ppVnode;
taosMemoryFree(pVnode->path);
taosMemoryFree(pVnode);
ppVnode[0] = NULL;
}
int32_t vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl) { int32_t vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl) {
SVnodeObj *pVnode = taosMemoryCalloc(1, sizeof(SVnodeObj)); SVnodeObj *pVnode = taosMemoryCalloc(1, sizeof(SVnodeObj));
if (pVnode == NULL) { if (pVnode == NULL) {
@ -134,6 +158,12 @@ int32_t vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl) {
} }
taosThreadRwlockWrlock(&pMgmt->lock); taosThreadRwlockWrlock(&pMgmt->lock);
SVnodeObj *pOld = NULL;
taosHashGetDup(pMgmt->hash, &pVnode->vgId, sizeof(int32_t), (void *)&pOld);
if (pOld) {
ASSERT(pOld->failed);
vmFreeVnodeObj(&pOld);
}
int32_t code = taosHashPut(pMgmt->hash, &pVnode->vgId, sizeof(int32_t), &pVnode, sizeof(SVnodeObj *)); int32_t code = taosHashPut(pMgmt->hash, &pVnode->vgId, sizeof(int32_t), &pVnode, sizeof(SVnodeObj *));
taosThreadRwlockUnlock(&pMgmt->lock); taosThreadRwlockUnlock(&pMgmt->lock);
@ -223,8 +253,7 @@ _closed:
vnodeDestroy(pVnode->vgId, path, pMgmt->pTfs); vnodeDestroy(pVnode->vgId, path, pMgmt->pTfs);
} }
taosMemoryFree(pVnode->path); vmFreeVnodeObj(&pVnode);
taosMemoryFree(pVnode);
} }
static int32_t vmRestoreVgroupId(SWrapperCfg *pCfg, STfs *pTfs) { static int32_t vmRestoreVgroupId(SWrapperCfg *pCfg, STfs *pTfs) {
@ -621,7 +650,7 @@ static void *vmRestoreVnodeInThread(void *param) {
for (int32_t v = 0; v < pThread->vnodeNum; ++v) { for (int32_t v = 0; v < pThread->vnodeNum; ++v) {
SVnodeObj *pVnode = pThread->ppVnodes[v]; SVnodeObj *pVnode = pThread->ppVnodes[v];
if (pVnode->failed) { if (pVnode->failed) {
dError("vgId:%d, skip restoring vnode in failure mode.", pVnode->vgId); dError("vgId:%d, cannot restore a vnode in failed mode.", pVnode->vgId);
continue; continue;
} }

4
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
View File

@ -187,7 +187,7 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp
pHead->vgId = ntohl(pHead->vgId); pHead->vgId = ntohl(pHead->vgId);
SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId); SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
if (pVnode == NULL || pVnode->failed) { if (pVnode == NULL) {
dGDebug("vgId:%d, msg:%p failed to put into vnode queue since %s, type:%s qtype:%d contLen:%d", pHead->vgId, pMsg, dGDebug("vgId:%d, msg:%p failed to put into vnode queue since %s, type:%s qtype:%d contLen:%d", pHead->vgId, pMsg,
terrstr(), TMSG_INFO(pMsg->msgType), qtype, pHead->contLen); terrstr(), TMSG_INFO(pMsg->msgType), qtype, pHead->contLen);
terrno = (terrno != 0) ? terrno : -1; terrno = (terrno != 0) ? terrno : -1;
@ -316,7 +316,7 @@ int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) {
int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) { int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) {
int32_t size = -1; int32_t size = -1;
SVnodeObj *pVnode = vmAcquireVnode(pMgmt, vgId); SVnodeObj *pVnode = vmAcquireVnode(pMgmt, vgId);
if (pVnode != NULL && !pVnode->failed) { if (pVnode != NULL) {
switch (qtype) { switch (qtype) {
case WRITE_QUEUE: case WRITE_QUEUE:
size = taosQueueItemSize(pVnode->pWriteW.queue); size = taosQueueItemSize(pVnode->pWriteW.queue);

2
source/dnode/mgmt/node_util/src/dmEps.c
View File

@ -232,7 +232,7 @@ int32_t dmWriteEps(SDnodeData *pData) {
if (buffer == NULL) goto _OVER; if (buffer == NULL) goto _OVER;
terrno = 0; terrno = 0;
pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC); pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
if (pFile == NULL) goto _OVER; if (pFile == NULL) goto _OVER;
int32_t len = strlen(buffer); int32_t len = strlen(buffer);

2
source/dnode/mgmt/node_util/src/dmFile.c
View File

@ -120,7 +120,7 @@ int32_t dmWriteFile(const char *path, const char *name, bool deployed) {
if (buffer == NULL) goto _OVER; if (buffer == NULL) goto _OVER;
terrno = 0; terrno = 0;
pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC); pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
if (pFile == NULL) goto _OVER; if (pFile == NULL) goto _OVER;
int32_t len = strlen(buffer); int32_t len = strlen(buffer);

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

@ -605,7 +605,7 @@ void mndDumpSdb() {
char *pCont = tjsonToString(json); char *pCont = tjsonToString(json);
int32_t contLen = strlen(pCont); int32_t contLen = strlen(pCont);
char file[] = "sdb.json"; char file[] = "sdb.json";
TdFilePtr pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC); TdFilePtr pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC| TD_FILE_WRITE_THROUGH);
if (pFile == NULL) { if (pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
mError("failed to write %s since %s", file, terrstr()); mError("failed to write %s since %s", file, terrstr());

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

@ -167,6 +167,7 @@ int32_t mndInitStream(SMnode *pMnode) {
void mndCleanupStream(SMnode *pMnode) { void mndCleanupStream(SMnode *pMnode) {
taosArrayDestroy(execInfo.pTaskList); taosArrayDestroy(execInfo.pTaskList);
taosHashCleanup(execInfo.pTaskMap); taosHashCleanup(execInfo.pTaskMap);
taosHashCleanup(execInfo.transMgmt.pDBTrans);
taosThreadMutexDestroy(&execInfo.lock); taosThreadMutexDestroy(&execInfo.lock);
taosHashCleanup(execInfo.transMgmt.pDBTrans); taosHashCleanup(execInfo.transMgmt.pDBTrans);
taosHashCleanup(execInfo.transMgmt.pWaitingList); taosHashCleanup(execInfo.transMgmt.pWaitingList);

2
source/dnode/mnode/impl/test/trans/trans1.cpp
View File

@ -38,7 +38,7 @@ class MndTestTrans1 : public ::testing::Test {
test.ServerStop(); test.ServerStop();
pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC); pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
int32_t writeLen = taosWriteFile(pFile, buffer, readLen); int32_t writeLen = taosWriteFile(pFile, buffer, readLen);
if (writeLen < 0 || writeLen == readLen) { if (writeLen < 0 || writeLen == readLen) {
ASSERT(1); ASSERT(1);

1
source/dnode/snode/src/snode.c
View File

@ -325,6 +325,7 @@ int32_t sndInit(SSnode *pSnode) {
} }
void sndClose(SSnode *pSnode) { void sndClose(SSnode *pSnode) {
stopRsync();
streamMetaNotifyClose(pSnode->pMeta); streamMetaNotifyClose(pSnode->pMeta);
streamMetaCommit(pSnode->pMeta); streamMetaCommit(pSnode->pMeta);
streamMetaClose(pSnode->pMeta); streamMetaClose(pSnode->pMeta);

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

@ -2015,8 +2015,7 @@ int32_t tqProcessTaskResetReq(STQ* pTq, SRpcMsg* pMsg) {
// clear flag set during do checkpoint, and open inputQ for all upstream tasks // clear flag set during do checkpoint, and open inputQ for all upstream tasks
if (streamTaskGetStatus(pTask, NULL) == TASK_STATUS__CK) { if (streamTaskGetStatus(pTask, NULL) == TASK_STATUS__CK) {
streamTaskClearCheckInfo(pTask); streamTaskClearCheckInfo(pTask, true);
taosArrayClear(pTask->pReadyMsgList);
streamTaskSetStatusReady(pTask); streamTaskSetStatusReady(pTask);
} }

2
source/dnode/vnode/src/tsdb/tsdbFS.c
View File

@ -104,7 +104,7 @@ static int32_t tsdbSaveFSToFile(STsdbFS *pFS, const char *fname) {
taosCalcChecksumAppend(0, pData, size); taosCalcChecksumAppend(0, pData, size);
// save to file // save to file
TdFilePtr pFD = taosOpenFile(fname, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC); TdFilePtr pFD = taosOpenFile(fname, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
if (pFD == NULL) { if (pFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);

11
source/dnode/vnode/src/tsdb/tsdbFS2.c
View File

@ -85,7 +85,7 @@ static int32_t save_json(const cJSON *json, const char *fname) {
char *data = cJSON_PrintUnformatted(json); char *data = cJSON_PrintUnformatted(json);
if (data == NULL) return TSDB_CODE_OUT_OF_MEMORY; if (data == NULL) return TSDB_CODE_OUT_OF_MEMORY;
TdFilePtr fp = taosOpenFile(fname, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC); TdFilePtr fp = taosOpenFile(fname, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
if (fp == NULL) { if (fp == NULL) {
code = TAOS_SYSTEM_ERROR(code); code = TAOS_SYSTEM_ERROR(code);
goto _exit; goto _exit;
@ -1186,11 +1186,6 @@ const char *gFSBgTaskName[] = {NULL, "MERGE", "RETENTION", "COMPACT"};
static int32_t tsdbFSRunBgTask(void *arg) { static int32_t tsdbFSRunBgTask(void *arg) {
STFSBgTask *task = (STFSBgTask *)arg; STFSBgTask *task = (STFSBgTask *)arg;
STFileSystem *fs = task->fs; STFileSystem *fs = task->fs;
STFileSet *fset;
tsdbFSGetFSet(fs, task->fid, &fset);
ASSERT(fset != NULL && fset->bgTaskRunning == task);
task->launchTime = taosGetTimestampMs(); task->launchTime = taosGetTimestampMs();
task->run(task->arg); task->run(task->arg);
@ -1203,6 +1198,10 @@ static int32_t tsdbFSRunBgTask(void *arg) {
taosThreadMutexLock(&fs->tsdb->mutex); taosThreadMutexLock(&fs->tsdb->mutex);
STFileSet *fset = NULL;
tsdbFSGetFSet(fs, task->fid, &fset);
ASSERT(fset != NULL && fset->bgTaskRunning == task);
// free last // free last
tsdbDoDoneBgTask(fs, task); tsdbDoDoneBgTask(fs, task);
fset->bgTaskRunning = NULL; fset->bgTaskRunning = NULL;

2
source/dnode/vnode/src/tsdb/tsdbFile.c
View File

@ -177,7 +177,7 @@ int32_t tsdbDFileRollback(STsdb *pTsdb, SDFileSet *pSet, EDataFileT ftype) {
taosCalcChecksumAppend(0, hdr, TSDB_FHDR_SIZE); taosCalcChecksumAppend(0, hdr, TSDB_FHDR_SIZE);
// open // open
pFD = taosOpenFile(fname, TD_FILE_WRITE); pFD = taosOpenFile(fname, TD_FILE_WRITE | TD_FILE_WRITE_THROUGH);
if (pFD == NULL) { if (pFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;

2
source/dnode/vnode/src/vnd/vnodeCommit.c
View File

@ -177,7 +177,7 @@ int vnodeSaveInfo(const char *dir, const SVnodeInfo *pInfo) {
} }
// save info to a vnode_tmp.json // save info to a vnode_tmp.json
pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC); pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
if (pFile == NULL) { if (pFile == NULL) {
vError("failed to open info file:%s for write:%s", fname, terrstr()); vError("failed to open info file:%s for write:%s", fname, terrstr());
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);

1
source/libs/index/src/indexFilter.c
View File

@ -328,6 +328,7 @@ static int32_t sifInitParam(SNode *node, SIFParam *param, SIFCtx *ctx) {
SIF_ERR_RET(scalarGenerateSetFromList((void **)&param->pFilter, node, nl->node.resType.type)); SIF_ERR_RET(scalarGenerateSetFromList((void **)&param->pFilter, node, nl->node.resType.type));
if (taosHashPut(ctx->pRes, &node, POINTER_BYTES, param, sizeof(*param))) { if (taosHashPut(ctx->pRes, &node, POINTER_BYTES, param, sizeof(*param))) {
taosHashCleanup(param->pFilter); taosHashCleanup(param->pFilter);
param->pFilter = NULL;
indexError("taosHashPut nodeList failed, size:%d", (int32_t)sizeof(*param)); indexError("taosHashPut nodeList failed, size:%d", (int32_t)sizeof(*param));
SIF_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); SIF_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
} }

12
source/libs/parser/src/parTranslater.c
View File

@ -1433,14 +1433,18 @@ static int32_t dataTypeComp(const SDataType* l, const SDataType* r) {
static EDealRes translateOperator(STranslateContext* pCxt, SOperatorNode* pOp) { static EDealRes translateOperator(STranslateContext* pCxt, SOperatorNode* pOp) {
if (isMultiResFunc(pOp->pLeft)) { if (isMultiResFunc(pOp->pLeft)) {
return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, ((SExprNode*)(pOp->pLeft))->aliasName); generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, ((SExprNode*)(pOp->pLeft))->aliasName);
return DEAL_RES_ERROR;
} }
if (isMultiResFunc(pOp->pRight)) { if (isMultiResFunc(pOp->pRight)) {
return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, ((SExprNode*)(pOp->pRight))->aliasName); generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, ((SExprNode*)(pOp->pRight))->aliasName);
return DEAL_RES_ERROR;
} }
if (TSDB_CODE_SUCCESS != scalarGetOperatorResultType(pOp)) { int32_t res = scalarGetOperatorResultType(pOp);
return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, pOp->node.aliasName); if (TSDB_CODE_SUCCESS != res) {
pCxt->errCode = res;
return DEAL_RES_ERROR;
} }
return DEAL_RES_CONTINUE; return DEAL_RES_CONTINUE;

4
source/libs/scalar/src/scalar.c
View File

@ -208,6 +208,7 @@ void sclFreeParam(SScalarParam *param) {
if (param->columnData != NULL) { if (param->columnData != NULL) {
colDataDestroy(param->columnData); colDataDestroy(param->columnData);
taosMemoryFreeClear(param->columnData); taosMemoryFreeClear(param->columnData);
param->columnData = NULL;
} }
if (param->pHashFilter != NULL) { if (param->pHashFilter != NULL) {
@ -845,6 +846,7 @@ int32_t sclExecOperator(SOperatorNode *node, SScalarCtx *ctx, SScalarParam *outp
SScalarParam *params = NULL; SScalarParam *params = NULL;
int32_t rowNum = 0; int32_t rowNum = 0;
int32_t code = 0; int32_t code = 0;
int32_t paramNum = 0;
// json not support in in operator // json not support in in operator
if (nodeType(node->pLeft) == QUERY_NODE_VALUE) { if (nodeType(node->pLeft) == QUERY_NODE_VALUE) {
@ -865,7 +867,7 @@ int32_t sclExecOperator(SOperatorNode *node, SScalarCtx *ctx, SScalarParam *outp
_bin_scalar_fn_t OperatorFn = getBinScalarOperatorFn(node->opType); _bin_scalar_fn_t OperatorFn = getBinScalarOperatorFn(node->opType);
int32_t paramNum = scalarGetOperatorParamNum(node->opType); paramNum = scalarGetOperatorParamNum(node->opType);
SScalarParam *pLeft = &params[0]; SScalarParam *pLeft = &params[0];
SScalarParam *pRight = paramNum > 1 ? &params[1] : NULL; SScalarParam *pRight = paramNum > 1 ? &params[1] : NULL;

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

@ -138,6 +138,8 @@ 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);
void streamClearChkptReadyMsg(SStreamTask* pTask);
int32_t streamTaskInitTokenBucket(STokenBucket* pBucket, int32_t numCap, int32_t numRate, float quotaRate, const char*); int32_t streamTaskInitTokenBucket(STokenBucket* pBucket, int32_t numCap, int32_t numRate, float quotaRate, const char*);
STaskId streamTaskExtractKey(const SStreamTask* pTask); STaskId streamTaskExtractKey(const SStreamTask* pTask);
void streamTaskInitForLaunchHTask(SHistoryTaskInfo* pInfo); void streamTaskInitForLaunchHTask(SHistoryTaskInfo* pInfo);

13
source/libs/stream/src/streamCheckpoint.c
View File

@ -165,6 +165,7 @@ static int32_t continueDispatchCheckpointBlock(SStreamDataBlock* pBlock, SStream
int32_t code = taosWriteQitem(pTask->outputq.queue->pQueue, pBlock); int32_t code = taosWriteQitem(pTask->outputq.queue->pQueue, pBlock);
if (code == 0) { if (code == 0) {
ASSERT(pTask->chkInfo.dispatchCheckpointTrigger == false);
streamDispatchStreamBlock(pTask); streamDispatchStreamBlock(pTask);
} else { } else {
stError("s-task:%s failed to put checkpoint into outputQ, code:%s", pTask->id.idStr, tstrerror(code)); stError("s-task:%s failed to put checkpoint into outputQ, code:%s", pTask->id.idStr, tstrerror(code));
@ -267,13 +268,17 @@ int32_t streamProcessCheckpointReadyMsg(SStreamTask* pTask) {
return 0; return 0;
} }
void streamTaskClearCheckInfo(SStreamTask* pTask) { void streamTaskClearCheckInfo(SStreamTask* pTask, bool clearChkpReadyMsg) {
pTask->chkInfo.checkpointingId = 0; // clear the checkpoint id pTask->chkInfo.checkpointingId = 0; // clear the checkpoint id
pTask->chkInfo.failedId = 0; pTask->chkInfo.failedId = 0;
pTask->chkInfo.startTs = 0; // clear the recorded start time pTask->chkInfo.startTs = 0; // clear the recorded start time
pTask->chkInfo.downstreamAlignNum = 0;
pTask->chkInfo.checkpointNotReadyTasks = 0; pTask->chkInfo.checkpointNotReadyTasks = 0;
// pTask->chkInfo.checkpointAlignCnt = 0;
pTask->chkInfo.dispatchCheckpointTrigger = false;
streamTaskOpenAllUpstreamInput(pTask); // open inputQ for all upstream tasks streamTaskOpenAllUpstreamInput(pTask); // open inputQ for all upstream tasks
if (clearChkpReadyMsg) {
streamClearChkptReadyMsg(pTask);
}
} }
int32_t streamSaveTaskCheckpointInfo(SStreamTask* p, int64_t checkpointId) { int32_t streamSaveTaskCheckpointInfo(SStreamTask* p, int64_t checkpointId) {
@ -293,7 +298,7 @@ int32_t streamSaveTaskCheckpointInfo(SStreamTask* p, int64_t checkpointId) {
p->chkInfo.checkpointId = p->chkInfo.checkpointingId; p->chkInfo.checkpointId = p->chkInfo.checkpointingId;
p->chkInfo.checkpointVer = p->chkInfo.processedVer; p->chkInfo.checkpointVer = p->chkInfo.processedVer;
streamTaskClearCheckInfo(p); streamTaskClearCheckInfo(p, false);
char* str = NULL; char* str = NULL;
streamTaskGetStatus(p, &str); streamTaskGetStatus(p, &str);
@ -469,7 +474,7 @@ int32_t streamTaskBuildCheckpoint(SStreamTask* pTask) {
if (code != TSDB_CODE_SUCCESS) { // clear the checkpoint info if failed if (code != TSDB_CODE_SUCCESS) { // clear the checkpoint info if failed
taosThreadMutexLock(&pTask->lock); taosThreadMutexLock(&pTask->lock);
streamTaskClearCheckInfo(pTask); streamTaskClearCheckInfo(pTask, false);
code = streamTaskHandleEvent(pTask->status.pSM, TASK_EVENT_CHECKPOINT_DONE); code = streamTaskHandleEvent(pTask->status.pSM, TASK_EVENT_CHECKPOINT_DONE);
taosThreadMutexUnlock(&pTask->lock); taosThreadMutexUnlock(&pTask->lock);

124
source/libs/stream/src/streamDispatch.c
View File

@ -14,9 +14,9 @@
*/ */
#include "streamInt.h" #include "streamInt.h"
#include "tmisce.h"
#include "trpc.h" #include "trpc.h"
#include "ttimer.h" #include "ttimer.h"
#include "tmisce.h"
typedef struct SBlockName { typedef struct SBlockName {
uint32_t hashValue; uint32_t hashValue;
@ -343,7 +343,8 @@ static int32_t doBuildDispatchMsg(SStreamTask* pTask, const SStreamDataBlock* pD
SSDataBlock* pDataBlock = taosArrayGet(pData->blocks, i); SSDataBlock* pDataBlock = taosArrayGet(pData->blocks, i);
// TODO: do not use broadcast // TODO: do not use broadcast
if (pDataBlock->info.type == STREAM_DELETE_RESULT || pDataBlock->info.type == STREAM_CHECKPOINT || pDataBlock->info.type == STREAM_TRANS_STATE) { if (pDataBlock->info.type == STREAM_DELETE_RESULT || pDataBlock->info.type == STREAM_CHECKPOINT ||
pDataBlock->info.type == STREAM_TRANS_STATE) {
for (int32_t j = 0; j < numOfVgroups; j++) { for (int32_t j = 0; j < numOfVgroups; j++) {
code = streamAddBlockIntoDispatchMsg(pDataBlock, &pReqs[j]); code = streamAddBlockIntoDispatchMsg(pDataBlock, &pReqs[j]);
if (code != 0) { if (code != 0) {
@ -371,7 +372,8 @@ static int32_t doBuildDispatchMsg(SStreamTask* pTask, const SStreamDataBlock* pD
pTask->msgInfo.pData = pReqs; pTask->msgInfo.pData = pReqs;
} }
stDebug("s-task:%s build dispatch msg success, msgId:%d, stage:%" PRId64, pTask->id.idStr, pTask->execInfo.dispatch, pTask->pMeta->stage); stDebug("s-task:%s build dispatch msg success, msgId:%d, stage:%" PRId64, pTask->id.idStr, pTask->execInfo.dispatch,
pTask->pMeta->stage);
return code; return code;
} }
@ -393,8 +395,8 @@ static int32_t sendDispatchMsg(SStreamTask* pTask, SStreamDispatchReq* pDispatch
SArray* vgInfo = pTask->outputInfo.shuffleDispatcher.dbInfo.pVgroupInfos; SArray* vgInfo = pTask->outputInfo.shuffleDispatcher.dbInfo.pVgroupInfos;
int32_t numOfVgroups = taosArrayGetSize(vgInfo); int32_t numOfVgroups = taosArrayGetSize(vgInfo);
stDebug("s-task:%s (child taskId:%d) start to shuffle-dispatch blocks to %d vgroup(s), msgId:%d", stDebug("s-task:%s (child taskId:%d) start to shuffle-dispatch blocks to %d vgroup(s), msgId:%d", id,
id, pTask->info.selfChildId, numOfVgroups, msgId); pTask->info.selfChildId, numOfVgroups, msgId);
for (int32_t i = 0; i < numOfVgroups; i++) { for (int32_t i = 0; i < numOfVgroups; i++) {
if (pDispatchMsg[i].blockNum > 0) { if (pDispatchMsg[i].blockNum > 0) {
@ -409,7 +411,8 @@ static int32_t sendDispatchMsg(SStreamTask* pTask, SStreamDispatchReq* pDispatch
} }
} }
stDebug("s-task:%s complete shuffle-dispatch blocks to all %d vnodes, msgId:%d", pTask->id.idStr, numOfVgroups, msgId); stDebug("s-task:%s complete shuffle-dispatch blocks to all %d vnodes, msgId:%d", pTask->id.idStr, numOfVgroups,
msgId);
} }
return code; return code;
@ -441,8 +444,8 @@ static void doRetryDispatchData(void* param, void* tmrId) {
int32_t numOfVgroups = taosArrayGetSize(vgInfo); int32_t numOfVgroups = taosArrayGetSize(vgInfo);
int32_t numOfFailed = taosArrayGetSize(pList); int32_t numOfFailed = taosArrayGetSize(pList);
stDebug("s-task:%s (child taskId:%d) retry shuffle-dispatch blocks to %d vgroup(s), msgId:%d", stDebug("s-task:%s (child taskId:%d) retry shuffle-dispatch blocks to %d vgroup(s), msgId:%d", id,
id, pTask->info.selfChildId, numOfFailed, msgId); pTask->info.selfChildId, numOfFailed, msgId);
for (int32_t i = 0; i < numOfFailed; i++) { for (int32_t i = 0; i < numOfFailed; i++) {
int32_t vgId = *(int32_t*)taosArrayGet(pList, i); int32_t vgId = *(int32_t*)taosArrayGet(pList, i);
@ -461,7 +464,8 @@ static void doRetryDispatchData(void* param, void* tmrId) {
} }
} }
stDebug("s-task:%s complete re-try shuffle-dispatch blocks to all %d vnodes, msgId:%d", pTask->id.idStr, numOfFailed, msgId); stDebug("s-task:%s complete re-try shuffle-dispatch blocks to all %d vnodes, msgId:%d", pTask->id.idStr,
numOfFailed, msgId);
} else { } else {
int32_t vgId = pTask->outputInfo.fixedDispatcher.nodeId; int32_t vgId = pTask->outputInfo.fixedDispatcher.nodeId;
SEpSet* pEpSet = &pTask->outputInfo.fixedDispatcher.epSet; SEpSet* pEpSet = &pTask->outputInfo.fixedDispatcher.epSet;
@ -531,10 +535,12 @@ int32_t streamSearchAndAddBlock(SStreamTask* pTask, SStreamDispatchReq* pReqs, S
} }
if (pDataBlock->info.parTbName[0]) { if (pDataBlock->info.parTbName[0]) {
snprintf(ctbName, TSDB_TABLE_NAME_LEN, "%s.%s", pTask->outputInfo.shuffleDispatcher.dbInfo.db, pDataBlock->info.parTbName); snprintf(ctbName, TSDB_TABLE_NAME_LEN, "%s.%s", pTask->outputInfo.shuffleDispatcher.dbInfo.db,
pDataBlock->info.parTbName);
} else { } else {
buildCtbNameByGroupIdImpl(pTask->outputInfo.shuffleDispatcher.stbFullName, groupId, pDataBlock->info.parTbName); buildCtbNameByGroupIdImpl(pTask->outputInfo.shuffleDispatcher.stbFullName, groupId, pDataBlock->info.parTbName);
snprintf(ctbName, TSDB_TABLE_NAME_LEN, "%s.%s", pTask->outputInfo.shuffleDispatcher.dbInfo.db, pDataBlock->info.parTbName); snprintf(ctbName, TSDB_TABLE_NAME_LEN, "%s.%s", pTask->outputInfo.shuffleDispatcher.dbInfo.db,
pDataBlock->info.parTbName);
} }
/*uint32_t hashValue = MurmurHash3_32(ctbName, strlen(ctbName));*/ /*uint32_t hashValue = MurmurHash3_32(ctbName, strlen(ctbName));*/
@ -576,13 +582,15 @@ int32_t streamSearchAndAddBlock(SStreamTask* pTask, SStreamDispatchReq* pReqs, S
} }
int32_t streamDispatchStreamBlock(SStreamTask* pTask) { int32_t streamDispatchStreamBlock(SStreamTask* pTask) {
ASSERT((pTask->outputInfo.type == TASK_OUTPUT__FIXED_DISPATCH || pTask->outputInfo.type == TASK_OUTPUT__SHUFFLE_DISPATCH)); ASSERT((pTask->outputInfo.type == TASK_OUTPUT__FIXED_DISPATCH ||
pTask->outputInfo.type == TASK_OUTPUT__SHUFFLE_DISPATCH));
const char* id = pTask->id.idStr; const char* id = pTask->id.idStr;
int32_t numOfElems = streamQueueGetNumOfItems(pTask->outputq.queue); int32_t numOfElems = streamQueueGetNumOfItems(pTask->outputq.queue);
if (numOfElems > 0) { if (numOfElems > 0) {
double size = SIZE_IN_MiB(taosQueueMemorySize(pTask->outputq.queue->pQueue)); double size = SIZE_IN_MiB(taosQueueMemorySize(pTask->outputq.queue->pQueue));
stDebug("s-task:%s start to dispatch intermediate block to downstream, elem in outputQ:%d, size:%.2fMiB", id, numOfElems, size); stDebug("s-task:%s start to dispatch intermediate block to downstream, elem in outputQ:%d, size:%.2fMiB", id,
numOfElems, size);
} }
// to make sure only one dispatch is running // to make sure only one dispatch is running
@ -593,6 +601,12 @@ int32_t streamDispatchStreamBlock(SStreamTask* pTask) {
return 0; return 0;
} }
if (pTask->chkInfo.dispatchCheckpointTrigger) {
stDebug("s-task:%s already send checkpoint trigger, not dispatch anymore", id);
atomic_store_8(&pTask->outputq.status, TASK_OUTPUT_STATUS__NORMAL);
return 0;
}
ASSERT(pTask->msgInfo.pData == NULL); ASSERT(pTask->msgInfo.pData == NULL);
stDebug("s-task:%s start to dispatch msg, set output status:%d", id, pTask->outputq.status); stDebug("s-task:%s start to dispatch msg, set output status:%d", id, pTask->outputq.status);
@ -635,7 +649,8 @@ int32_t streamDispatchStreamBlock(SStreamTask* pTask) {
if (++retryCount > MAX_CONTINUE_RETRY_COUNT) { // add to timer to retry if (++retryCount > MAX_CONTINUE_RETRY_COUNT) { // add to timer to retry
int32_t ref = atomic_add_fetch_32(&pTask->status.timerActive, 1); int32_t ref = atomic_add_fetch_32(&pTask->status.timerActive, 1);
stDebug("s-task:%s failed to dispatch msg to downstream for %d times, code:%s, add timer to retry in %dms, ref:%d", stDebug(
"s-task:%s failed to dispatch msg to downstream for %d times, code:%s, add timer to retry in %dms, ref:%d",
pTask->id.idStr, retryCount, tstrerror(terrno), DISPATCH_RETRY_INTERVAL_MS, ref); pTask->id.idStr, retryCount, tstrerror(terrno), DISPATCH_RETRY_INTERVAL_MS, ref);
streamRetryDispatchData(pTask, DISPATCH_RETRY_INTERVAL_MS); streamRetryDispatchData(pTask, DISPATCH_RETRY_INTERVAL_MS);
@ -659,7 +674,8 @@ int32_t streamDispatchScanHistoryFinishMsg(SStreamTask* pTask) {
if (pTask->outputInfo.type == TASK_OUTPUT__FIXED_DISPATCH) { if (pTask->outputInfo.type == TASK_OUTPUT__FIXED_DISPATCH) {
req.downstreamTaskId = pTask->outputInfo.fixedDispatcher.taskId; req.downstreamTaskId = pTask->outputInfo.fixedDispatcher.taskId;
pTask->notReadyTasks = 1; pTask->notReadyTasks = 1;
doDispatchScanHistoryFinishMsg(pTask, &req, pTask->outputInfo.fixedDispatcher.nodeId, &pTask->outputInfo.fixedDispatcher.epSet); doDispatchScanHistoryFinishMsg(pTask, &req, pTask->outputInfo.fixedDispatcher.nodeId,
&pTask->outputInfo.fixedDispatcher.epSet);
} else if (pTask->outputInfo.type == TASK_OUTPUT__SHUFFLE_DISPATCH) { } else if (pTask->outputInfo.type == TASK_OUTPUT__SHUFFLE_DISPATCH) {
SArray* vgInfo = pTask->outputInfo.shuffleDispatcher.dbInfo.pVgroupInfos; SArray* vgInfo = pTask->outputInfo.shuffleDispatcher.dbInfo.pVgroupInfos;
int32_t numOfVgs = taosArrayGetSize(vgInfo); int32_t numOfVgs = taosArrayGetSize(vgInfo);
@ -667,8 +683,8 @@ int32_t streamDispatchScanHistoryFinishMsg(SStreamTask* pTask) {
char* p = NULL; char* p = NULL;
streamTaskGetStatus(pTask, &p); streamTaskGetStatus(pTask, &p);
stDebug("s-task:%s send scan-history data complete msg to downstream (shuffle-dispatch) %d tasks, status:%s", pTask->id.idStr, stDebug("s-task:%s send scan-history data complete msg to downstream (shuffle-dispatch) %d tasks, status:%s",
numOfVgs, p); pTask->id.idStr, numOfVgs, p);
for (int32_t i = 0; i < numOfVgs; i++) { for (int32_t i = 0; i < numOfVgs; i++) {
SVgroupInfo* pVgInfo = taosArrayGet(vgInfo, i); SVgroupInfo* pVgInfo = taosArrayGet(vgInfo, i);
req.downstreamTaskId = pVgInfo->taskId; req.downstreamTaskId = pVgInfo->taskId;
@ -696,7 +712,8 @@ int32_t streamTaskSendCheckpointReadyMsg(SStreamTask* pTask) {
} }
taosArrayClear(pTask->pReadyMsgList); taosArrayClear(pTask->pReadyMsgList);
stDebug("s-task:%s level:%d checkpoint ready msg sent to all %d upstreams", pTask->id.idStr, pTask->info.taskLevel, num); stDebug("s-task:%s level:%d checkpoint ready msg sent to all %d upstreams", pTask->id.idStr, pTask->info.taskLevel,
num);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -866,8 +883,8 @@ int32_t buildCheckpointSourceRsp(SStreamCheckpointSourceReq* pReq, SRpcHandleInf
return 0; return 0;
} }
int32_t streamAddCheckpointSourceRspMsg(SStreamCheckpointSourceReq* pReq, SRpcHandleInfo* pRpcInfo, int32_t streamAddCheckpointSourceRspMsg(SStreamCheckpointSourceReq* pReq, SRpcHandleInfo* pRpcInfo, SStreamTask* pTask,
SStreamTask* pTask, int8_t isSucceed) { int8_t isSucceed) {
SStreamChkptReadyInfo info = {0}; SStreamChkptReadyInfo info = {0};
buildCheckpointSourceRsp(pReq, pRpcInfo, &info.msg, isSucceed); buildCheckpointSourceRsp(pReq, pRpcInfo, &info.msg, isSucceed);
@ -876,7 +893,8 @@ int32_t streamAddCheckpointSourceRspMsg(SStreamCheckpointSourceReq* pReq, SRpcHa
} }
taosArrayPush(pTask->pReadyMsgList, &info); taosArrayPush(pTask->pReadyMsgList, &info);
stDebug("s-task:%s add checkpoint source rsp msg, total:%d", pTask->id.idStr, (int32_t)taosArrayGetSize(pTask->pReadyMsgList)); stDebug("s-task:%s add checkpoint source rsp msg, total:%d", pTask->id.idStr,
(int32_t)taosArrayGetSize(pTask->pReadyMsgList));
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -926,8 +944,10 @@ int32_t streamAddCheckpointReadyMsg(SStreamTask* pTask, int32_t upstreamTaskId,
initRpcMsg(&info.msg, TDMT_STREAM_TASK_CHECKPOINT_READY, buf, tlen + sizeof(SMsgHead)); initRpcMsg(&info.msg, TDMT_STREAM_TASK_CHECKPOINT_READY, buf, tlen + sizeof(SMsgHead));
info.msg.info.noResp = 1; // refactor later. info.msg.info.noResp = 1; // refactor later.
stDebug("s-task:%s (level:%d) prepare checkpoint ready msg to upstream s-task:0x%" PRIx64 ":0x%x (vgId:%d) idx:%d, vgId:%d", stDebug("s-task:%s (level:%d) prepare checkpoint ready msg to upstream s-task:0x%" PRIx64
pTask->id.idStr, pTask->info.taskLevel, req.streamId, req.upstreamTaskId, req.upstreamNodeId, index, req.upstreamNodeId); ":0x%x (vgId:%d) idx:%d, vgId:%d",
pTask->id.idStr, pTask->info.taskLevel, req.streamId, req.upstreamTaskId, req.upstreamNodeId, index,
req.upstreamNodeId);
if (pTask->pReadyMsgList == NULL) { if (pTask->pReadyMsgList == NULL) {
pTask->pReadyMsgList = taosArrayInit(4, sizeof(SStreamChkptReadyInfo)); pTask->pReadyMsgList = taosArrayInit(4, sizeof(SStreamChkptReadyInfo));
@ -937,6 +957,16 @@ int32_t streamAddCheckpointReadyMsg(SStreamTask* pTask, int32_t upstreamTaskId,
return 0; return 0;
} }
void streamClearChkptReadyMsg(SStreamTask* pTask) {
if (pTask->pReadyMsgList == NULL) return;
for (int i = 0; i < taosArrayGetSize(pTask->pReadyMsgList); i++) {
SStreamChkptReadyInfo* pInfo = taosArrayGet(pTask->pReadyMsgList, i);
rpcFreeCont(pInfo->msg.pCont);
}
taosArrayClear(pTask->pReadyMsgList);
}
int32_t tEncodeCompleteHistoryDataMsg(SEncoder* pEncoder, const SStreamCompleteHistoryMsg* pReq) { int32_t tEncodeCompleteHistoryDataMsg(SEncoder* pEncoder, const SStreamCompleteHistoryMsg* pReq) {
if (tStartEncode(pEncoder) < 0) return -1; if (tStartEncode(pEncoder) < 0) return -1;
if (tEncodeI64(pEncoder, pReq->streamId) < 0) return -1; if (tEncodeI64(pEncoder, pReq->streamId) < 0) return -1;
@ -959,7 +989,8 @@ int32_t tDecodeCompleteHistoryDataMsg(SDecoder* pDecoder, SStreamCompleteHistory
return 0; return 0;
} }
int32_t streamTaskBuildScanhistoryRspMsg(SStreamTask* pTask, SStreamScanHistoryFinishReq* pReq, void** pBuffer, int32_t* pLen) { int32_t streamTaskBuildScanhistoryRspMsg(SStreamTask* pTask, SStreamScanHistoryFinishReq* pReq, void** pBuffer,
int32_t* pLen) {
int32_t len = 0; int32_t len = 0;
int32_t code = 0; int32_t code = 0;
SEncoder encoder; SEncoder encoder;
@ -1039,30 +1070,14 @@ int32_t streamNotifyUpstreamContinue(SStreamTask* pTask) {
return 0; return 0;
} }
static void dispatchDataInFuture(void* param, void* tmrId) {
SStreamTask* pTask = param;
if (streamTaskShouldStop(pTask)) {
int32_t ref = atomic_sub_fetch_32(&pTask->status.timerActive, 1);
stDebug("s-task:%s should stop, abort from timer, ref:%d", pTask->id.idStr, ref);
return;
}
ETaskStatus status = streamTaskGetStatus(pTask, NULL);
if (status == TASK_STATUS__CK) {
stDebug("s-task:%s in checkpoint status, wait for 500ms to dispatch data downstream", pTask->id.idStr);
taosTmrReset(dispatchDataInFuture, 500, pTask, streamEnv.timer, &pTask->msgInfo.pTimer);
} else {
int32_t ref = atomic_sub_fetch_32(&pTask->status.timerActive, 1);
stDebug("s-task:%s start to dispatch data, jump out of timer, ref:%d", pTask->id.idStr, ref);
streamDispatchStreamBlock(pTask);
}
}
// this message has been sent successfully, let's try next one. // this message has been sent successfully, let's try next one.
static int32_t handleDispatchSuccessRsp(SStreamTask* pTask, int32_t downstreamId) { static int32_t handleDispatchSuccessRsp(SStreamTask* pTask, int32_t downstreamId) {
destroyDispatchMsg(pTask->msgInfo.pData, getNumOfDispatchBranch(pTask)); destroyDispatchMsg(pTask->msgInfo.pData, getNumOfDispatchBranch(pTask));
bool delayDispatch = (pTask->msgInfo.dispatchMsgType == STREAM_INPUT__CHECKPOINT_TRIGGER); bool delayDispatch = (pTask->msgInfo.dispatchMsgType == STREAM_INPUT__CHECKPOINT_TRIGGER);
if (delayDispatch) {
pTask->chkInfo.dispatchCheckpointTrigger = true;
}
pTask->msgInfo.pData = NULL; pTask->msgInfo.pData = NULL;
pTask->msgInfo.dispatchMsgType = 0; pTask->msgInfo.dispatchMsgType = 0;
@ -1083,13 +1098,7 @@ static int32_t handleDispatchSuccessRsp(SStreamTask* pTask, int32_t downstreamId
// otherwise, continue dispatch the first block to down stream task in pipeline // otherwise, continue dispatch the first block to down stream task in pipeline
if (delayDispatch) { if (delayDispatch) {
int32_t ref = atomic_add_fetch_32(&pTask->status.timerActive, 1); return 0;
stDebug("s-task:%s in checkpoint status, add in timer, try dispatch data in 500ms, ref:%d", pTask->id.idStr, ref);
if (pTask->msgInfo.pTimer != NULL) {
taosTmrReset(dispatchDataInFuture, 500, pTask, streamEnv.timer, &pTask->msgInfo.pTimer);
} else {
pTask->msgInfo.pTimer = taosTmrStart(dispatchDataInFuture, 500, pTask, streamEnv.timer);
}
} else { } else {
streamDispatchStreamBlock(pTask); streamDispatchStreamBlock(pTask);
} }
@ -1111,7 +1120,8 @@ int32_t streamProcessDispatchRsp(SStreamTask* pTask, SStreamDispatchRsp* pRsp, i
// follower not handle the dispatch rsp // follower not handle the dispatch rsp
if ((pTask->pMeta->role == NODE_ROLE_FOLLOWER) || (pTask->status.downstreamReady != 1)) { if ((pTask->pMeta->role == NODE_ROLE_FOLLOWER) || (pTask->status.downstreamReady != 1)) {
stError("s-task:%s vgId:%d is follower or task just re-launched, not handle the dispatch rsp, discard it", id, vgId); stError("s-task:%s vgId:%d is follower or task just re-launched, not handle the dispatch rsp, discard it", id,
vgId);
return TSDB_CODE_STREAM_TASK_NOT_EXIST; return TSDB_CODE_STREAM_TASK_NOT_EXIST;
} }
@ -1129,8 +1139,8 @@ int32_t streamProcessDispatchRsp(SStreamTask* pTask, SStreamDispatchRsp* pRsp, i
// flag. Here we need to retry dispatch this message to downstream task immediately. handle the case the failure // flag. Here we need to retry dispatch this message to downstream task immediately. handle the case the failure
// happened too fast. // happened too fast.
if (code == TSDB_CODE_STREAM_TASK_NOT_EXIST) { // destination task does not exist, not retry anymore if (code == TSDB_CODE_STREAM_TASK_NOT_EXIST) { // destination task does not exist, not retry anymore
stError("s-task:%s failed to dispatch msg to task:0x%x(vgId:%d), msgId:%d no retry, since task destroyed already", id, stError("s-task:%s failed to dispatch msg to task:0x%x(vgId:%d), msgId:%d no retry, since task destroyed already",
pRsp->downstreamTaskId, pRsp->downstreamNodeId, msgId); id, pRsp->downstreamTaskId, pRsp->downstreamNodeId, msgId);
} else { } else {
stError("s-task:%s failed to dispatch msgId:%d to task:0x%x(vgId:%d), code:%s, add to retry list", id, msgId, stError("s-task:%s failed to dispatch msgId:%d to task:0x%x(vgId:%d), code:%s, add to retry list", id, msgId,
pRsp->downstreamTaskId, pRsp->downstreamNodeId, tstrerror(code)); pRsp->downstreamTaskId, pRsp->downstreamNodeId, tstrerror(code));
@ -1174,7 +1184,9 @@ int32_t streamProcessDispatchRsp(SStreamTask* pTask, SStreamDispatchRsp* pRsp, i
ASSERT(leftRsp >= 0); ASSERT(leftRsp >= 0);
if (leftRsp > 0) { if (leftRsp > 0) {
stDebug( "s-task:%s recv dispatch rsp, msgId:%d from 0x%x(vgId:%d), downstream task input status:%d code:%s, waiting for %d rsp", stDebug(
"s-task:%s recv dispatch rsp, msgId:%d from 0x%x(vgId:%d), downstream task input status:%d code:%s, waiting "
"for %d rsp",
id, msgId, pRsp->downstreamTaskId, pRsp->downstreamNodeId, pRsp->inputStatus, tstrerror(code), leftRsp); id, msgId, pRsp->downstreamTaskId, pRsp->downstreamNodeId, pRsp->inputStatus, tstrerror(code), leftRsp);
} else { } else {
stDebug( stDebug(
@ -1182,8 +1194,8 @@ int32_t streamProcessDispatchRsp(SStreamTask* pTask, SStreamDispatchRsp* pRsp, i
id, msgId, pRsp->downstreamTaskId, pRsp->downstreamNodeId, pRsp->inputStatus, tstrerror(code)); id, msgId, pRsp->downstreamTaskId, pRsp->downstreamNodeId, pRsp->inputStatus, tstrerror(code));
} }
} else { } else {
stDebug("s-task:%s recv fix-dispatch rsp, msgId:%d from 0x%x(vgId:%d), downstream task input status:%d code:%s", stDebug("s-task:%s recv fix-dispatch rsp, msgId:%d from 0x%x(vgId:%d), downstream task input status:%d code:%s", id,
id, msgId, pRsp->downstreamTaskId, pRsp->downstreamNodeId, pRsp->inputStatus, tstrerror(code)); msgId, pRsp->downstreamTaskId, pRsp->downstreamNodeId, pRsp->inputStatus, tstrerror(code));
} }
ASSERT(leftRsp >= 0); ASSERT(leftRsp >= 0);

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

@ -48,6 +48,7 @@ static int32_t doOutputResultBlockImpl(SStreamTask* pTask, SStreamDataBlock* pBl
return code; return code;
} }
// checkpoint trigger will be checked
streamDispatchStreamBlock(pTask); streamDispatchStreamBlock(pTask);
} }

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

@ -15,11 +15,11 @@
#include "executor.h" #include "executor.h"
#include "streamInt.h" #include "streamInt.h"
#include "streamsm.h"
#include "tmisce.h" #include "tmisce.h"
#include "tstream.h" #include "tstream.h"
#include "ttimer.h" #include "ttimer.h"
#include "wal.h" #include "wal.h"
#include "streamsm.h"
static void streamTaskDestroyUpstreamInfo(SUpstreamInfo* pUpstreamInfo); static void streamTaskDestroyUpstreamInfo(SUpstreamInfo* pUpstreamInfo);
@ -357,7 +357,9 @@ void tFreeStreamTask(SStreamTask* pTask) {
walCloseReader(pTask->exec.pWalReader); walCloseReader(pTask->exec.pWalReader);
} }
streamClearChkptReadyMsg(pTask);
pTask->pReadyMsgList = taosArrayDestroy(pTask->pReadyMsgList); pTask->pReadyMsgList = taosArrayDestroy(pTask->pReadyMsgList);
if (pTask->msgInfo.pData != NULL) { if (pTask->msgInfo.pData != NULL) {
destroyDispatchMsg(pTask->msgInfo.pData, getNumOfDispatchBranch(pTask)); destroyDispatchMsg(pTask->msgInfo.pData, getNumOfDispatchBranch(pTask));
pTask->msgInfo.pData = NULL; pTask->msgInfo.pData = NULL;

2
source/libs/sync/src/syncRaftCfg.c
View File

@ -103,7 +103,7 @@ int32_t syncWriteCfgFile(SSyncNode *pNode) {
if (buffer == NULL) goto _OVER; if (buffer == NULL) goto _OVER;
terrno = 0; terrno = 0;
pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC); pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
if (pFile == NULL) goto _OVER; if (pFile == NULL) goto _OVER;
int32_t len = strlen(buffer); int32_t len = strlen(buffer);

2
source/libs/sync/src/syncRaftStore.c
View File

@ -128,7 +128,7 @@ int32_t raftStoreWriteFile(SSyncNode *pNode) {
if (buffer == NULL) goto _OVER; if (buffer == NULL) goto _OVER;
terrno = 0; terrno = 0;
pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC); pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
if (pFile == NULL) goto _OVER; if (pFile == NULL) goto _OVER;
int32_t len = strlen(buffer); int32_t len = strlen(buffer);

2
source/libs/wal/src/walMeta.c
View File

@ -873,7 +873,7 @@ int walSaveMeta(SWal* pWal) {
return -1; return -1;
} }
TdFilePtr pMetaFile = taosOpenFile(tmpFnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC); TdFilePtr pMetaFile = taosOpenFile(tmpFnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
if (pMetaFile == NULL) { if (pMetaFile == NULL) {
wError("vgId:%d, failed to open file due to %s. file:%s", pWal->cfg.vgId, strerror(errno), tmpFnameStr); wError("vgId:%d, failed to open file due to %s. file:%s", pWal->cfg.vgId, strerror(errno), tmpFnameStr);
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);

5
source/libs/wal/src/walWrite.c
View File

@ -133,6 +133,7 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
} }
walBuildIdxName(pWal, walGetCurFileFirstVer(pWal), fnameStr); walBuildIdxName(pWal, walGetCurFileFirstVer(pWal), fnameStr);
taosCloseFile(&pWal->pIdxFile);
TdFilePtr pIdxFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ | TD_FILE_APPEND); TdFilePtr pIdxFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ | TD_FILE_APPEND);
if (pIdxFile == NULL) { if (pIdxFile == NULL) {
@ -153,6 +154,7 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
} }
walBuildLogName(pWal, walGetCurFileFirstVer(pWal), fnameStr); walBuildLogName(pWal, walGetCurFileFirstVer(pWal), fnameStr);
taosCloseFile(&pWal->pLogFile);
TdFilePtr pLogFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ | TD_FILE_APPEND); TdFilePtr pLogFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ | TD_FILE_APPEND);
wDebug("vgId:%d, wal truncate file %s", pWal->cfg.vgId, fnameStr); wDebug("vgId:%d, wal truncate file %s", pWal->cfg.vgId, fnameStr);
if (pLogFile == NULL) { if (pLogFile == NULL) {
@ -204,6 +206,7 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
pWal->vers.lastVer = ver - 1; pWal->vers.lastVer = ver - 1;
((SWalFileInfo *)taosArrayGetLast(pWal->fileInfoSet))->lastVer = ver - 1; ((SWalFileInfo *)taosArrayGetLast(pWal->fileInfoSet))->lastVer = ver - 1;
((SWalFileInfo *)taosArrayGetLast(pWal->fileInfoSet))->fileSize = entry.offset; ((SWalFileInfo *)taosArrayGetLast(pWal->fileInfoSet))->fileSize = entry.offset;
taosCloseFile(&pIdxFile); taosCloseFile(&pIdxFile);
taosCloseFile(&pLogFile); taosCloseFile(&pLogFile);
@ -605,7 +608,7 @@ int32_t walWriteWithSyncInfo(SWal *pWal, int64_t index, tmsg_t msgType, SWalSync
return -1; return -1;
} }
if (pWal->pIdxFile == NULL || pWal->pIdxFile == NULL || pWal->writeCur < 0) { if (pWal->pIdxFile == NULL || pWal->pLogFile == NULL || pWal->writeCur < 0) {
if (walInitWriteFile(pWal) < 0) { if (walInitWriteFile(pWal) < 0) {
taosThreadMutexUnlock(&pWal->mutex); taosThreadMutexUnlock(&pWal->mutex);
return -1; return -1;

671
source/os/src/osFile.c
View File

@ -21,11 +21,12 @@
#include <io.h> #include <io.h>
#include <WinBase.h> #include <WinBase.h>
#include <ktmw32.h> #include <ktmw32.h>
#include <windows.h>
#define F_OK 0 #define F_OK 0
#define W_OK 2 #define W_OK 2
#define R_OK 4 #define R_OK 4
#define _SEND_FILE_STEP_ 1000 #define _SEND_FILE_STEP_ 1024
#else #else
#include <fcntl.h> #include <fcntl.h>
@ -44,12 +45,22 @@
typedef int32_t FileFd; typedef int32_t FileFd;
#ifdef WINDOWS
typedef struct TdFile {
TdThreadRwlock rwlock;
int refId;
HANDLE hFile;
FILE* fp;
int32_t tdFileOptions;
} TdFile;
#else
typedef struct TdFile { typedef struct TdFile {
TdThreadRwlock rwlock; TdThreadRwlock rwlock;
int refId; int refId;
FileFd fd; FileFd fd;
FILE *fp; FILE *fp;
} TdFile; } TdFile;
#endif // WINDOWS
#define FILE_WITH_LOCK 1 #define FILE_WITH_LOCK 1
@ -240,15 +251,12 @@ int32_t taosStatFile(const char *path, int64_t *size, int32_t *mtime, int32_t *a
return 0; return 0;
} }
int32_t taosDevInoFile(TdFilePtr pFile, int64_t *stDev, int64_t *stIno) { int32_t taosDevInoFile(TdFilePtr pFile, int64_t *stDev, int64_t *stIno) {
if (pFile == NULL || pFile->fd < 0) { #ifdef WINDOWS
if (pFile == NULL || pFile->hFile == NULL) {
return -1; return -1;
} }
#ifdef WINDOWS
BY_HANDLE_FILE_INFORMATION bhfi; BY_HANDLE_FILE_INFORMATION bhfi;
HANDLE handle = (HANDLE)_get_osfhandle(pFile->fd); if (GetFileInformationByHandle(pFile->hFile, &bhfi) == FALSE) {
if (GetFileInformationByHandle(handle, &bhfi) == FALSE) {
printf("taosFStatFile get file info fail."); printf("taosFStatFile get file info fail.");
return -1; return -1;
} }
@ -262,7 +270,9 @@ int32_t taosDevInoFile(TdFilePtr pFile, int64_t *stDev, int64_t *stIno) {
} }
#else #else
if (pFile == NULL || pFile->fd < 0) {
return -1;
}
struct stat fileStat; struct stat fileStat;
int32_t code = fstat(pFile->fd, &fileStat); int32_t code = fstat(pFile->fd, &fileStat);
if (code < 0) { if (code < 0) {
@ -282,10 +292,7 @@ int32_t taosDevInoFile(TdFilePtr pFile, int64_t *stDev, int64_t *stIno) {
return 0; return 0;
} }
TdFilePtr taosOpenFile(const char *path, int32_t tdFileOptions) { FILE *taosOpenFileForStream(const char *path, int32_t tdFileOptions) {
int fd = -1;
FILE *fp = NULL;
if (tdFileOptions & TD_FILE_STREAM) {
char *mode = NULL; char *mode = NULL;
if (tdFileOptions & TD_FILE_APPEND) { if (tdFileOptions & TD_FILE_APPEND) {
mode = (tdFileOptions & TD_FILE_TEXT) ? "at+" : "ab+"; mode = (tdFileOptions & TD_FILE_TEXT) ? "at+" : "ab+";
@ -300,11 +307,333 @@ TdFilePtr taosOpenFile(const char *path, int32_t tdFileOptions) {
if (tdFileOptions & TD_FILE_EXCL) { if (tdFileOptions & TD_FILE_EXCL) {
return NULL; return NULL;
} }
fp = fopen(path, mode); return fopen(path, mode);
if (fp == NULL) { }
return NULL;
#ifdef WINDOWS
HANDLE taosOpenFileNotStream(const char *path, int32_t tdFileOptions) {
DWORD openMode = 0;
DWORD access = 0;
DWORD fileFlag = FILE_ATTRIBUTE_NORMAL;
DWORD shareMode = FILE_SHARE_READ;
openMode = OPEN_EXISTING;
if (tdFileOptions & TD_FILE_CREATE) {
openMode = OPEN_ALWAYS;
} else if (tdFileOptions & TD_FILE_EXCL) {
openMode = CREATE_NEW;
} else if ((tdFileOptions & TD_FILE_TRUNC)) {
openMode = TRUNCATE_EXISTING;
access |= GENERIC_WRITE;
}
if (tdFileOptions & TD_FILE_APPEND) {
access |= FILE_APPEND_DATA;
}
if (tdFileOptions & TD_FILE_WRITE) {
access |= GENERIC_WRITE;
}
shareMode |= FILE_SHARE_WRITE;
access |= GENERIC_READ;
if (tdFileOptions & TD_FILE_AUTO_DEL) {
fileFlag |= FILE_ATTRIBUTE_TEMPORARY;
}
if (tdFileOptions & TD_FILE_WRITE_THROUGH) {
fileFlag |= FILE_FLAG_WRITE_THROUGH;
}
HANDLE h = CreateFile(path, access, shareMode, NULL, openMode, fileFlag, NULL);
if (h != INVALID_HANDLE_VALUE && (tdFileOptions & TD_FILE_APPEND) && (tdFileOptions & TD_FILE_WRITE)) {
SetFilePointer(h, 0, NULL, FILE_END);
}
if (h == INVALID_HANDLE_VALUE) {
DWORD dwError = GetLastError();
LPVOID lpMsgBuf;
FormatMessage(FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM, NULL, dwError,
0,
(LPTSTR)&lpMsgBuf, 0, NULL);
printf("CreateFile failed with error %d: %s", dwError, (char*)lpMsgBuf);
LocalFree(lpMsgBuf);
}
return h;
}
int64_t taosReadFile(TdFilePtr pFile, void *buf, int64_t count) {
#if FILE_WITH_LOCK
taosThreadRwlockRdlock(&(pFile->rwlock));
#endif
if (pFile->hFile == NULL) {
#if FILE_WITH_LOCK
taosThreadRwlockUnlock(&(pFile->rwlock));
#endif
return -1;
}
DWORD bytesRead;
if (!ReadFile(pFile->hFile, buf, count, &bytesRead, NULL)) {
bytesRead = -1;
}
#if FILE_WITH_LOCK
taosThreadRwlockUnlock(&(pFile->rwlock));
#endif
return bytesRead;
}
int64_t taosWriteFile(TdFilePtr pFile, const void *buf, int64_t count) {
if (pFile == NULL || pFile->hFile == NULL) {
return 0;
}
#if FILE_WITH_LOCK
taosThreadRwlockWrlock(&(pFile->rwlock));
#endif
DWORD bytesWritten;
if (!WriteFile(pFile->hFile, buf, count, &bytesWritten, NULL)) {
bytesWritten = -1;
}
#if FILE_WITH_LOCK
taosThreadRwlockUnlock(&(pFile->rwlock));
#endif
return bytesWritten;
}
int64_t taosPWriteFile(TdFilePtr pFile, const void *buf, int64_t count, int64_t offset) {
if (pFile == NULL) {
return 0;
}
#if FILE_WITH_LOCK
taosThreadRwlockWrlock(&(pFile->rwlock));
#endif
ASSERT(pFile->hFile != NULL); // Please check if you have closed the file.
if (pFile->hFile == NULL) {
#if FILE_WITH_LOCK
taosThreadRwlockUnlock(&(pFile->rwlock));
#endif
return 0;
}
DWORD ret = 0;
OVERLAPPED ol = {0};
ol.OffsetHigh = (uint32_t)((offset & 0xFFFFFFFF00000000LL) >> 0x20);
ol.Offset = (uint32_t)(offset & 0xFFFFFFFFLL);
SetLastError(0);
BOOL result = WriteFile(pFile->hFile, buf, count, &ret, &ol);
if (!result) {
errno = GetLastError();
ret = -1;
}
#if FILE_WITH_LOCK
taosThreadRwlockUnlock(&(pFile->rwlock));
#endif
return ret;
}
int64_t taosLSeekFile(TdFilePtr pFile, int64_t offset, int32_t whence) {
if (pFile == NULL || pFile->hFile == NULL) {
return -1;
}
#if FILE_WITH_LOCK
taosThreadRwlockRdlock(&(pFile->rwlock));
#endif
LARGE_INTEGER liOffset;
liOffset.QuadPart = offset;
if (!SetFilePointerEx(pFile->hFile, liOffset, NULL, whence)) {
return -1;
}
liOffset.QuadPart = 0;
if (!SetFilePointerEx(pFile->hFile, liOffset, &liOffset, FILE_CURRENT)) {
return -1;
}
#if FILE_WITH_LOCK
taosThreadRwlockUnlock(&(pFile->rwlock));
#endif
return liOffset.QuadPart;
}
int32_t taosFStatFile(TdFilePtr pFile, int64_t *size, int32_t *mtime) {
if (pFile == NULL || pFile->hFile == NULL) {
return 0;
}
if (size != NULL) {
LARGE_INTEGER fileSize;
if (!GetFileSizeEx(pFile->hFile, &fileSize)) {
return -1; // Error getting file size
}
*size = fileSize.QuadPart;
}
if (mtime != NULL) {
FILETIME creationTime, lastAccessTime, lastWriteTime;
if (!GetFileTime(pFile->hFile, &creationTime, &lastAccessTime, &lastWriteTime)) {
return -1; // Error getting file time
}
// Convert the FILETIME structure to a time_t value
ULARGE_INTEGER ull;
ull.LowPart = lastWriteTime.dwLowDateTime;
ull.HighPart = lastWriteTime.dwHighDateTime;
*mtime = (int32_t)((ull.QuadPart - 116444736000000000ULL) / 10000000ULL);
}
return 0;
}
int32_t taosLockFile(TdFilePtr pFile) {
if (pFile == NULL || pFile->hFile == NULL) {
return -1;
}
BOOL fSuccess = FALSE;
LARGE_INTEGER fileSize;
OVERLAPPED overlapped = {0};
fSuccess = LockFileEx(pFile->hFile, LOCKFILE_EXCLUSIVE_LOCK | LOCKFILE_FAIL_IMMEDIATELY,
0, // reserved
~0, // number of bytes to lock low
~0, // number of bytes to lock high
&overlapped // overlapped structure
);
if (!fSuccess) {
return GetLastError();
}
return 0;
}
int32_t taosUnLockFile(TdFilePtr pFile) {
if (pFile == NULL || pFile->hFile == NULL) {
return 0;
}
BOOL fSuccess = FALSE;
OVERLAPPED overlapped = {0};
fSuccess = UnlockFileEx(pFile->hFile, 0, ~0, ~0, &overlapped);
if (!fSuccess) {
return GetLastError();
}
return 0;
}
int32_t taosFtruncateFile(TdFilePtr pFile, int64_t l_size) {
if (pFile == NULL) {
return 0;
}
if (pFile->hFile == NULL) {
printf("Ftruncate file error, hFile was null\n");
return -1;
}
LARGE_INTEGER li_0;
li_0.QuadPart = (int64_t)0;
BOOL cur = SetFilePointerEx(pFile->hFile, li_0, NULL, FILE_CURRENT);
if (!cur) {
printf("SetFilePointerEx Error getting current position in file.\n");
return -1;
}
LARGE_INTEGER li_size;
li_size.QuadPart = l_size;
BOOL cur2 = SetFilePointerEx(pFile->hFile, li_size, NULL, FILE_BEGIN);
if (cur2 == 0) {
int error = GetLastError();
printf("SetFilePointerEx GetLastError is: %d\n", error);
switch (error) {
case ERROR_INVALID_HANDLE:
errno = EBADF;
break;
default:
errno = EIO;
break;
}
return -1;
}
if (!SetEndOfFile(pFile->hFile)) {
int error = GetLastError();
printf("SetEndOfFile GetLastError is:%d", error);
switch (error) {
case ERROR_INVALID_HANDLE:
errno = EBADF;
break;
default:
errno = EIO;
break;
}
return -1;
}
return 0;
}
int64_t taosFSendFile(TdFilePtr pFileOut, TdFilePtr pFileIn, int64_t *offset, int64_t size) {
if (pFileOut == NULL || pFileIn == NULL) {
return 0;
}
if (pFileIn->hFile == NULL || pFileOut->hFile == NULL) {
return 0;
}
LARGE_INTEGER fileOffset;
fileOffset.QuadPart = *offset;
if (!SetFilePointerEx(pFileIn->hFile, fileOffset, &fileOffset, FILE_BEGIN)) {
return -1;
}
int64_t writeLen = 0;
uint8_t buffer[_SEND_FILE_STEP_] = {0};
DWORD bytesRead;
DWORD bytesWritten;
for (int64_t len = 0; len < (size - _SEND_FILE_STEP_); len += _SEND_FILE_STEP_) {
if (!ReadFile(pFileIn->hFile, buffer, _SEND_FILE_STEP_, &bytesRead, NULL)) {
return writeLen;
}
if (bytesRead <= 0) {
return writeLen;
} else if (bytesRead < _SEND_FILE_STEP_) {
if (!WriteFile(pFileOut->hFile, buffer, bytesRead, &bytesWritten, NULL)) {
return -1;
} else {
return (int64_t)(writeLen + bytesRead);
} }
} else { } else {
if (!WriteFile(pFileOut->hFile, buffer, _SEND_FILE_STEP_, &bytesWritten, NULL)) {
return -1;
} else {
writeLen += _SEND_FILE_STEP_;
}
}
}
int64_t remain = size - writeLen;
if (remain > 0) {
DWORD bytesRead;
if (!ReadFile(pFileIn->hFile, buffer, (DWORD)remain, &bytesRead, NULL)) {
return -1;
}
if (bytesRead <= 0) {
return writeLen;
} else {
DWORD bytesWritten;
if (!WriteFile(pFileOut->hFile, buffer, bytesRead, &bytesWritten, NULL)) {
return -1;
} else {
writeLen += bytesWritten;
}
}
}
return writeLen;
}
#else
int taosOpenFileNotStream(const char *path, int32_t tdFileOptions) {
int access = O_BINARY; int access = O_BINARY;
access |= (tdFileOptions & TD_FILE_CREATE) ? O_CREAT : 0; access |= (tdFileOptions & TD_FILE_CREATE) ? O_CREAT : 0;
if ((tdFileOptions & TD_FILE_WRITE) && (tdFileOptions & TD_FILE_READ)) { if ((tdFileOptions & TD_FILE_WRITE) && (tdFileOptions & TD_FILE_READ)) {
@ -318,80 +647,8 @@ TdFilePtr taosOpenFile(const char *path, int32_t tdFileOptions) {
access |= (tdFileOptions & TD_FILE_APPEND) ? O_APPEND : 0; access |= (tdFileOptions & TD_FILE_APPEND) ? O_APPEND : 0;
access |= (tdFileOptions & TD_FILE_TEXT) ? O_TEXT : 0; access |= (tdFileOptions & TD_FILE_TEXT) ? O_TEXT : 0;
access |= (tdFileOptions & TD_FILE_EXCL) ? O_EXCL : 0; access |= (tdFileOptions & TD_FILE_EXCL) ? O_EXCL : 0;
#ifdef WINDOWS int fd = open(path, access, S_IRWXU | S_IRWXG | S_IRWXO);
int32_t pmode = _S_IREAD | _S_IWRITE; return fd;
if (tdFileOptions & TD_FILE_AUTO_DEL) {
pmode |= _O_TEMPORARY;
}
fd = _open(path, access, pmode);
#else
fd = open(path, access, S_IRWXU | S_IRWXG | S_IRWXO);
#endif
if (fd == -1) {
return NULL;
}
}
TdFilePtr pFile = (TdFilePtr)taosMemoryMalloc(sizeof(TdFile));
if (pFile == NULL) {
if (fd >= 0) close(fd);
if (fp != NULL) fclose(fp);
return NULL;
}
#if FILE_WITH_LOCK
taosThreadRwlockInit(&(pFile->rwlock), NULL);
#endif
pFile->fd = fd;
pFile->fp = fp;
pFile->refId = 0;
if (tdFileOptions & TD_FILE_AUTO_DEL) {
#ifdef WINDOWS
// do nothing, since the property of pmode is set with _O_TEMPORARY; the OS will recycle
// the file handle, as well as the space on disk.
#else
// Remove it instantly, so when the program exits normally/abnormally, the file
// will be automatically remove by OS.
unlink(path);
#endif
}
return pFile;
}
int32_t taosCloseFile(TdFilePtr *ppFile) {
int32_t code = 0;
if (ppFile == NULL || *ppFile == NULL) {
return 0;
}
#if FILE_WITH_LOCK
taosThreadRwlockWrlock(&((*ppFile)->rwlock));
#endif
if ((*ppFile)->fp != NULL) {
fflush((*ppFile)->fp);
fclose((*ppFile)->fp);
(*ppFile)->fp = NULL;
}
if ((*ppFile)->fd >= 0) {
#ifdef WINDOWS
HANDLE h = (HANDLE)_get_osfhandle((*ppFile)->fd);
!FlushFileBuffers(h);
#else
// warning: never fsync silently in base lib
/*fsync((*ppFile)->fd);*/
#endif
code = close((*ppFile)->fd);
(*ppFile)->fd = -1;
}
(*ppFile)->refId = 0;
#if FILE_WITH_LOCK
taosThreadRwlockUnlock(&((*ppFile)->rwlock));
taosThreadRwlockDestroy(&((*ppFile)->rwlock));
#endif
taosMemoryFree(*ppFile);
*ppFile = NULL;
return code;
} }
int64_t taosReadFile(TdFilePtr pFile, void *buf, int64_t count) { int64_t taosReadFile(TdFilePtr pFile, void *buf, int64_t count) {
@ -441,42 +698,6 @@ int64_t taosReadFile(TdFilePtr pFile, void *buf, int64_t count) {
return count; return count;
} }
int64_t taosPReadFile(TdFilePtr pFile, void *buf, int64_t count, int64_t offset) {
if (pFile == NULL) {
return 0;
}
#if FILE_WITH_LOCK
taosThreadRwlockRdlock(&(pFile->rwlock));
#endif
ASSERT(pFile->fd >= 0); // Please check if you have closed the file.
if (pFile->fd < 0) {
#if FILE_WITH_LOCK
taosThreadRwlockUnlock(&(pFile->rwlock));
#endif
return -1;
}
#ifdef WINDOWS
DWORD ret = 0;
OVERLAPPED ol = {0};
ol.OffsetHigh = (uint32_t)((offset & 0xFFFFFFFF00000000LL) >> 0x20);
ol.Offset = (uint32_t)(offset & 0xFFFFFFFFLL);
HANDLE handle = (HANDLE)_get_osfhandle(pFile->fd);
SetLastError(0);
BOOL result = ReadFile(handle, buf, count, &ret, &ol);
if (!result && GetLastError() != ERROR_HANDLE_EOF) {
errno = GetLastError();
ret = -1;
}
#else
int64_t ret = pread(pFile->fd, buf, count, offset);
#endif
#if FILE_WITH_LOCK
taosThreadRwlockUnlock(&(pFile->rwlock));
#endif
return ret;
}
int64_t taosWriteFile(TdFilePtr pFile, const void *buf, int64_t count) { int64_t taosWriteFile(TdFilePtr pFile, const void *buf, int64_t count) {
if (pFile == NULL) { if (pFile == NULL) {
return 0; return 0;
@ -706,25 +927,6 @@ int32_t taosFtruncateFile(TdFilePtr pFile, int64_t l_size) {
#endif #endif
} }
int32_t taosFsyncFile(TdFilePtr pFile) {
if (pFile == NULL) {
return 0;
}
// this implementation is WRONG
// fflush is not a replacement of fsync
if (pFile->fp != NULL) return fflush(pFile->fp);
if (pFile->fd >= 0) {
#ifdef WINDOWS
HANDLE h = (HANDLE)_get_osfhandle(pFile->fd);
return !FlushFileBuffers(h);
#else
return fsync(pFile->fd);
#endif
}
return 0;
}
int64_t taosFSendFile(TdFilePtr pFileOut, TdFilePtr pFileIn, int64_t *offset, int64_t size) { int64_t taosFSendFile(TdFilePtr pFileOut, TdFilePtr pFileIn, int64_t *offset, int64_t size) {
if (pFileOut == NULL || pFileIn == NULL) { if (pFileOut == NULL || pFileIn == NULL) {
return 0; return 0;
@ -824,6 +1026,167 @@ int64_t taosFSendFile(TdFilePtr pFileOut, TdFilePtr pFileIn, int64_t *offset, in
#endif #endif
} }
#endif // WINDOWS
TdFilePtr taosOpenFile(const char *path, int32_t tdFileOptions) {
FILE *fp = NULL;
#ifdef WINDOWS
HANDLE hFile = NULL;
#else
int fd = -1;
#endif
if (tdFileOptions & TD_FILE_STREAM) {
fp = taosOpenFileForStream(path, tdFileOptions);
if (fp == NULL) return NULL;
} else {
#ifdef WINDOWS
hFile = taosOpenFileNotStream(path, tdFileOptions);
if (hFile == INVALID_HANDLE_VALUE) return NULL;
#else
fd = taosOpenFileNotStream(path, tdFileOptions);
if (fd == -1) return NULL;
#endif
}
TdFilePtr pFile = (TdFilePtr)taosMemoryMalloc(sizeof(TdFile));
if (pFile == NULL) {
#ifdef WINDOWS
if (hFile != NULL) CloseHandle(hFile);
#else
if (fd >= 0) close(fd);
#endif
if (fp != NULL) fclose(fp);
return NULL;
}
#if FILE_WITH_LOCK
taosThreadRwlockInit(&(pFile->rwlock), NULL);
#endif
pFile->fp = fp;
pFile->refId = 0;
#ifdef WINDOWS
pFile->hFile = hFile;
pFile->tdFileOptions = tdFileOptions;
// do nothing, since the property of pmode is set with _O_TEMPORARY; the OS will recycle
// the file handle, as well as the space on disk.
#else
pFile->fd = fd;
// Remove it instantly, so when the program exits normally/abnormally, the file
// will be automatically remove by OS.
if (tdFileOptions & TD_FILE_AUTO_DEL) {
unlink(path);
}
#endif
return pFile;
}
int32_t taosCloseFile(TdFilePtr *ppFile) {
int32_t code = 0;
if (ppFile == NULL || *ppFile == NULL) {
return 0;
}
#if FILE_WITH_LOCK
taosThreadRwlockWrlock(&((*ppFile)->rwlock));
#endif
if ((*ppFile)->fp != NULL) {
fflush((*ppFile)->fp);
fclose((*ppFile)->fp);
(*ppFile)->fp = NULL;
}
#ifdef WINDOWS
if ((*ppFile)->hFile != NULL) {
// FlushFileBuffers((*ppFile)->hFile);
if (!CloseHandle((*ppFile)->hFile)) {
code = -1;
}
(*ppFile)->hFile = NULL;
#else
if ((*ppFile)->fd >= 0) {
// warning: never fsync silently in base lib
/*fsync((*ppFile)->fd);*/
code = close((*ppFile)->fd);
(*ppFile)->fd = -1;
#endif
}
(*ppFile)->refId = 0;
#if FILE_WITH_LOCK
taosThreadRwlockUnlock(&((*ppFile)->rwlock));
taosThreadRwlockDestroy(&((*ppFile)->rwlock));
#endif
taosMemoryFree(*ppFile);
*ppFile = NULL;
return code;
}
int64_t taosPReadFile(TdFilePtr pFile, void *buf, int64_t count, int64_t offset) {
if (pFile == NULL) {
return 0;
}
#ifdef WINDOWS
#if FILE_WITH_LOCK
taosThreadRwlockRdlock(&(pFile->rwlock));
#endif
ASSERT(pFile->hFile != NULL); // Please check if you have closed the file.
if (pFile->hFile == NULL) {
#if FILE_WITH_LOCK
taosThreadRwlockUnlock(&(pFile->rwlock));
#endif
return -1;
}
DWORD ret = 0;
OVERLAPPED ol = {0};
ol.OffsetHigh = (uint32_t)((offset & 0xFFFFFFFF00000000LL) >> 0x20);
ol.Offset = (uint32_t)(offset & 0xFFFFFFFFLL);
SetLastError(0);
BOOL result = ReadFile(pFile->hFile, buf, count, &ret, &ol);
if (!result && GetLastError() != ERROR_HANDLE_EOF) {
errno = GetLastError();
ret = -1;
}
#else
#if FILE_WITH_LOCK
taosThreadRwlockRdlock(&(pFile->rwlock));
#endif
ASSERT(pFile->fd >= 0); // Please check if you have closed the file.
if (pFile->fd < 0) {
#if FILE_WITH_LOCK
taosThreadRwlockUnlock(&(pFile->rwlock));
#endif
return -1;
}
int64_t ret = pread(pFile->fd, buf, count, offset);
#endif
#if FILE_WITH_LOCK
taosThreadRwlockUnlock(&(pFile->rwlock));
#endif
return ret;
}
int32_t taosFsyncFile(TdFilePtr pFile) {
if (pFile == NULL) {
return 0;
}
// this implementation is WRONG
// fflush is not a replacement of fsync
if (pFile->fp != NULL) return fflush(pFile->fp);
#ifdef WINDOWS
if (pFile->hFile != NULL) {
if (pFile->tdFileOptions & TD_FILE_WRITE_THROUGH) {
return 0;
}
return !FlushFileBuffers(pFile->hFile);
#else
if (pFile->fd >= 0) {
return fsync(pFile->fd);
#endif
}
return 0;
}
void taosFprintfFile(TdFilePtr pFile, const char *format, ...) { void taosFprintfFile(TdFilePtr pFile, const char *format, ...) {
if (pFile == NULL || pFile->fp == NULL) { if (pFile == NULL || pFile->fp == NULL) {
return; return;
@ -834,7 +1197,13 @@ void taosFprintfFile(TdFilePtr pFile, const char *format, ...) {
va_end(ap); va_end(ap);
} }
bool taosValidFile(TdFilePtr pFile) { return pFile != NULL && pFile->fd > 0; } bool taosValidFile(TdFilePtr pFile) {
#ifdef WINDOWS
return pFile != NULL && pFile->hFile != NULL;
#else
return pFile != NULL && pFile->fd > 0;
#endif
}
int32_t taosUmaskFile(int32_t maskVal) { int32_t taosUmaskFile(int32_t maskVal) {
#ifdef WINDOWS #ifdef WINDOWS
@ -960,14 +1329,20 @@ int32_t taosCompressFile(char *srcFileName, char *destFileName) {
goto cmp_end; goto cmp_end;
} }
pFile = taosOpenFile(destFileName, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC); int access = O_BINARY | O_WRONLY | O_TRUNC | O_CREAT;
if (pFile == NULL) { #ifdef WINDOWS
int32_t pmode = _S_IREAD | _S_IWRITE;
#else
int32_t pmode = S_IRWXU | S_IRWXG | S_IRWXO;
#endif
int fd = open(destFileName, access, pmode);
if (fd < 0) {
ret = -2; ret = -2;
goto cmp_end; goto cmp_end;
} }
// Both gzclose() and fclose() will close the associated fd, so they need to have different fds. // Both gzclose() and fclose() will close the associated fd, so they need to have different fds.
FileFd gzFd = dup(pFile->fd); FileFd gzFd = dup(fd);
if (gzFd < 0) { if (gzFd < 0) {
ret = -4; ret = -4;
goto cmp_end; goto cmp_end;

183
source/os/test/osTests.cpp
View File

@ -15,6 +15,7 @@
#include <gtest/gtest.h> #include <gtest/gtest.h>
#include <iostream> #include <iostream>
#include <inttypes.h>
#pragma GCC diagnostic push #pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wwrite-strings" #pragma GCC diagnostic ignored "-Wwrite-strings"
@ -29,6 +30,10 @@
#include "os.h" #include "os.h"
#include "tlog.h" #include "tlog.h"
#ifdef WINDOWS
#include <windows.h>
#endif // WINDOWS
TEST(osTest, osSystem) { TEST(osTest, osSystem) {
const char *flags = "UTL FATAL "; const char *flags = "UTL FATAL ";
ELogLevel level = DEBUG_FATAL; ELogLevel level = DEBUG_FATAL;
@ -68,7 +73,8 @@ void fileOperateOnBusy(void *param) {
char * fname = (char *)param; char * fname = (char *)param;
TdFilePtr pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE); TdFilePtr pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE);
printf("On busy thread open file\n"); printf("On busy thread open file\n");
ASSERT_NE(pFile, nullptr); if (pFile == NULL) return;
// ASSERT_NE(pFile, nullptr);
int ret = taosLockFile(pFile); int ret = taosLockFile(pFile);
printf("On busy thread lock file ret:%d\n", ret); printf("On busy thread lock file ret:%d\n", ret);
@ -97,6 +103,7 @@ TEST(osTest, osFile) {
TdFilePtr pOutFD = taosCreateFile(fname, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC); TdFilePtr pOutFD = taosCreateFile(fname, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
ASSERT_NE(pOutFD, nullptr); ASSERT_NE(pOutFD, nullptr);
printf("create file success\n"); printf("create file success\n");
taosCloseFile(&pOutFD);
TdFilePtr pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE); TdFilePtr pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE);
printf("open file\n"); printf("open file\n");
@ -135,4 +142,178 @@ TEST(osTest, osFile) {
//printf("remove file success"); //printf("remove file success");
} }
#ifndef OSFILE_PERFORMANCE_TEST
#define MAX_WORDS 100
#define MAX_WORD_LENGTH 20
#define MAX_TEST_FILE_SIZE 100000
#define TESTTIMES 1000
char *getRandomWord() {
static char words[][MAX_WORD_LENGTH] = {
"Lorem", "ipsum", "dolor", "sit", "amet", "consectetur", "adipiscing", "elit",
"sed", "do", "eiusmod", "tempor", "incididunt", "ut", "labore", "et", "dolore", "magna",
"aliqua", "Ut", "enim", "ad", "minim", "veniam", "quis", "nostrud", "exercitation", "ullamco",
"Why", "do", "programmers", "prefer", "using", "dark", "mode?", "Because", "light", "attracts",
"bugs", "and", "they", "want", "to", "code", "in", "peace,", "like", "a", "ninja", "in", "the", "shadows."
"aliqua", "Ut", "enim", "ad", "minim", "veniam", "quis", "nostrud", "exercitation", "ullamco",
"laboris", "nisi", "ut", "aliquip", "ex", "ea", "commodo", "consequat", "Duis", "aute", "irure",
"dolor", "in", "reprehenderit", "in", "voluptate", "velit", "esse", "cillum", "dolore", "eu",
"fugiat", "nulla", "pariatur", "Excepteur", "sint", "occaecat", "cupidatat", "non", "proident",
"sunt", "in", "culpa", "qui", "officia", "deserunt", "mollit", "anim", "id", "est", "laborum"
};
return words[taosRand() % MAX_WORDS];
}
int64_t fillBufferWithRandomWords(char *buffer, int64_t maxBufferSize) {
int64_t len = 0;
while (len < maxBufferSize) {
char * word = getRandomWord();
size_t wordLen = strlen(word);
if (len + wordLen + 1 < maxBufferSize) {
strcat(buffer, word);
strcat(buffer, " ");
len += wordLen + 1;
} else {
break;
}
}
return len;
}
int64_t calculateAverage(int64_t arr[], int size) {
int64_t sum = 0;
for (int i = 0; i < size; i++) {
sum += arr[i];
}
return sum / size;
}
int64_t calculateMax(int64_t arr[], int size) {
int64_t max = arr[0];
for (int i = 1; i < size; i++) {
if (arr[i] > max) {
max = arr[i];
}
}
return max;
}
int64_t calculateMin(int64_t arr[], int size) {
int64_t min = arr[0];
for (int i = 1; i < size; i++) {
if (arr[i] < min) {
min = arr[i];
}
}
return min;
}
TEST(osTest, osFilePerformance) {
printf("os file performance testting...\n");
int64_t WriteFileCost;
int64_t ReadFileCost;
int64_t OpenForWriteCloseFileCost;
int64_t OpenForReadCloseFileCost;
char * buffer;
char * writeBuffer = (char *)taosMemoryCalloc(1, MAX_TEST_FILE_SIZE);
char * readBuffer = (char *)taosMemoryCalloc(1, MAX_TEST_FILE_SIZE);
int64_t size = fillBufferWithRandomWords(writeBuffer, MAX_TEST_FILE_SIZE);
char * fname = "./osFilePerformanceTest.txt";
TdFilePtr pOutFD = taosCreateFile(fname, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
ASSERT_NE(pOutFD, nullptr);
taosCloseFile(&pOutFD);
printf("os file performance start write...\n");
int64_t t1 = taosGetTimestampUs();
for (int i = 0; i < TESTTIMES; ++i) {
TdFilePtr pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_WRITE_THROUGH);
ASSERT_NE(pFile, nullptr);
taosWriteFile(pFile, writeBuffer, size);
taosFsyncFile(pFile);
taosCloseFile(&pFile);
}
int64_t t2 = taosGetTimestampUs();
WriteFileCost = t2 - t1;
printf("os file performance start read...\n");
for (int i = 0; i < TESTTIMES; ++i) {
TdFilePtr pFile = taosOpenFile(fname, TD_FILE_READ);
ASSERT_NE(pFile, nullptr);
taosReadFile(pFile, readBuffer, size);
taosCloseFile(&pFile);
int readLine = strlen(readBuffer);
ASSERT_EQ(size, readLine);
}
int64_t t3 = taosGetTimestampUs();
ReadFileCost = t3 - t2;
printf("os file performance start open1...\n");
for (int i = 0; i < TESTTIMES; ++i) {
TdFilePtr pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE);
ASSERT_NE(pFile, nullptr);
taosCloseFile(&pFile);
}
int64_t t4 = taosGetTimestampUs();
OpenForWriteCloseFileCost = t4 - t3;
printf("os file performance start open2...\n");
for (int i = 0; i < TESTTIMES; ++i) {
TdFilePtr pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_READ);
ASSERT_NE(pFile, nullptr);
taosCloseFile(&pFile);
}
int64_t t5 = taosGetTimestampUs();
OpenForReadCloseFileCost = t5 - t4;
#ifdef WINDOWS
printf("os file performance start window native...\n");
for (int i = 0; i < TESTTIMES; ++i) {
HANDLE hFile = CreateFile(fname, // 文件名
GENERIC_WRITE, // 写权限
FILE_SHARE_READ, // 不共享
NULL, // 默认安全描述符
OPEN_ALWAYS, // 打开已存在的文件
FILE_FLAG_WRITE_THROUGH, // 文件标志，可以根据实际需求调整
NULL // 模板文件句柄，对于创建新文件不需要
);
if (hFile == INVALID_HANDLE_VALUE) {
printf("Error opening file\n");
break;
}
// 写入数据
DWORD bytesWritten;
if (!WriteFile(hFile, writeBuffer, size, &bytesWritten, NULL)) {
// 处理错误
printf("Error writing to file\n");
CloseHandle(hFile);
break;
}
// 关闭文件
CloseHandle(hFile);
}
int64_t t6 = taosGetTimestampUs();
int64_t nativeWritCost = t6 - t5;
printf("Test Write file using native API %d times, cost: %" PRId64 "us\n", TESTTIMES, nativeWritCost);
#endif // WINDOWS
taosMemoryFree(writeBuffer);
taosMemoryFree(readBuffer);
printf("Test Write file %d times, cost: %" PRId64 "us\n", TESTTIMES, WriteFileCost);
printf("Test Read file %d times, cost: %" PRId64 "us\n", TESTTIMES, ReadFileCost);
printf("Test OpenForWrite & Close file %d times, cost: %" PRId64 "us\n", TESTTIMES, OpenForWriteCloseFileCost);
printf("Test OpenForRead & Close file %d times, cost: %" PRId64 "us\n", TESTTIMES, OpenForReadCloseFileCost);
}
#endif OSFILE_PERFORMANCE_TEST
#pragma GCC diagnostic pop #pragma GCC diagnostic pop

10
source/util/src/tlog.c
View File

@ -154,7 +154,12 @@ int32_t taosInitSlowLog() {
#endif #endif
if (strlen(tsLogDir) != 0) { if (strlen(tsLogDir) != 0) {
char lastC = tsLogDir[strlen(tsLogDir) - 1];
if (lastC == '\\' || lastC == '/') {
snprintf(fullName, PATH_MAX, "%s" "%s", tsLogDir, logFileName);
} else {
snprintf(fullName, PATH_MAX, "%s" TD_DIRSEP "%s", tsLogDir, logFileName); snprintf(fullName, PATH_MAX, "%s" TD_DIRSEP "%s", tsLogDir, logFileName);
}
} else { } else {
snprintf(fullName, PATH_MAX, "%s", logFileName); snprintf(fullName, PATH_MAX, "%s", logFileName);
} }
@ -178,7 +183,12 @@ int32_t taosInitLog(const char *logName, int32_t maxFiles) {
char fullName[PATH_MAX] = {0}; char fullName[PATH_MAX] = {0};
if (strlen(tsLogDir) != 0) { if (strlen(tsLogDir) != 0) {
char lastC = tsLogDir[strlen(tsLogDir) - 1];
if (lastC == '\\' || lastC == '/') {
snprintf(fullName, PATH_MAX, "%s" "%s", tsLogDir, logName);
} else {
snprintf(fullName, PATH_MAX, "%s" TD_DIRSEP "%s", tsLogDir, logName); snprintf(fullName, PATH_MAX, "%s" TD_DIRSEP "%s", tsLogDir, logName);
}
} else { } else {
snprintf(fullName, PATH_MAX, "%s", logName); snprintf(fullName, PATH_MAX, "%s", logName);
} }

2
tests/system-test/0-others/view/non_marterial_view/test_view.py
View File

@ -162,7 +162,7 @@ class TDTestCase:
assert('TIMESTAMP' in data_type_list and 'INT' in data_type_list and 'INT UNSIGNED' in data_type_list and 'BIGINT' in data_type_list and 'BIGINT UNSIGNED' in data_type_list and 'FLOAT' in data_type_list and 'DOUBLE' in data_type_list and 'VARCHAR' in data_type_list and 'SMALLINT' in data_type_list and 'SMALLINT UNSIGNED' in data_type_list and 'TINYINT' in data_type_list and 'TINYINT UNSIGNED' in data_type_list and 'BOOL' in data_type_list and 'VARCHAR' in data_type_list and 'NCHAR' in data_type_list and 'GEOMETRY' in data_type_list and 'VARBINARY' in data_type_list) assert('TIMESTAMP' in data_type_list and 'INT' in data_type_list and 'INT UNSIGNED' in data_type_list and 'BIGINT' in data_type_list and 'BIGINT UNSIGNED' in data_type_list and 'FLOAT' in data_type_list and 'DOUBLE' in data_type_list and 'VARCHAR' in data_type_list and 'SMALLINT' in data_type_list and 'SMALLINT UNSIGNED' in data_type_list and 'TINYINT' in data_type_list and 'TINYINT UNSIGNED' in data_type_list and 'BOOL' in data_type_list and 'VARCHAR' in data_type_list and 'NCHAR' in data_type_list and 'GEOMETRY' in data_type_list and 'VARBINARY' in data_type_list)
tdSql.execute("create view v2 as select * from tb where c1 >5 and c7 like '%ab%';") tdSql.execute("create view v2 as select * from tb where c1 >5 and c7 like '%ab%';")
self.check_view_num(2) self.check_view_num(2)
tdSql.error("create view v3 as select * from tb where c1 like '%ab%';", expectErrInfo='Invalid value type') tdSql.error("create view v3 as select * from tb where c1 like '%ab%';", expectErrInfo='Invalid operation')
tdSql.execute("create view v3 as select first(ts), sum(c1) from tb group by c2 having avg(c4) > 0;") tdSql.execute("create view v3 as select first(ts), sum(c1) from tb group by c2 having avg(c4) > 0;")
tdSql.execute("create view v4 as select _wstart,sum(c6) from tb interval(10s);") tdSql.execute("create view v4 as select _wstart,sum(c6) from tb interval(10s);")
tdSql.execute("create view v5 as select * from tb join v2 on tb.ts = v2.ts;") tdSql.execute("create view v5 as select * from tb join v2 on tb.ts = v2.ts;")

31
utils/test/c/tmq_taosx_ci.c
View File

@ -30,7 +30,7 @@ typedef struct {
int meta; int meta;
int srcVgroups; int srcVgroups;
int dstVgroups; int dstVgroups;
char dir[64]; char dir[256];
} Config; } Config;
Config g_conf = {0}; Config g_conf = {0};
@ -409,6 +409,30 @@ int buildStable(TAOS* pConn, TAOS_RES* pRes) {
} }
taos_free_result(pRes); taos_free_result(pRes);
#ifdef WINDOWS
pRes = taos_query(pConn,
"CREATE STABLE `meters_summary` (`_wstart` TIMESTAMP, `current` FLOAT, `groupid` INT, `location` VARCHAR(16)) TAGS (`group_id` BIGINT UNSIGNED)");
if (taos_errno(pRes) != 0) {
printf("failed to create super table meters_summary, reason:%s\n", taos_errstr(pRes));
return -1;
}
taos_free_result(pRes);
pRes = taos_query(pConn,
" CREATE TABLE `t_d2a450ee819dcf7576f0282d9ac22dbc` USING `meters_summary` (`group_id`) TAGS (13135550082773579308)");
if (taos_errno(pRes) != 0) {
printf("failed to create super table meters_summary, reason:%s\n", taos_errstr(pRes));
return -1;
}
taos_free_result(pRes);
pRes = taos_query(pConn, "insert into t_d2a450ee819dcf7576f0282d9ac22dbc values (now, 120, 1, 'San Francisco')");
if (taos_errno(pRes) != 0) {
printf("failed to insert into table d0, reason:%s\n", taos_errstr(pRes));
return -1;
}
taos_free_result(pRes);
#else
pRes = taos_query(pConn, pRes = taos_query(pConn,
"create stream meters_summary_s trigger at_once IGNORE EXPIRED 0 into meters_summary as select _wstart, max(current) as current, " "create stream meters_summary_s trigger at_once IGNORE EXPIRED 0 into meters_summary as select _wstart, max(current) as current, "
"groupid, location from meters partition by groupid, location interval(10m)"); "groupid, location from meters partition by groupid, location interval(10m)");
@ -417,6 +441,7 @@ int buildStable(TAOS* pConn, TAOS_RES* pRes) {
return -1; return -1;
} }
taos_free_result(pRes); taos_free_result(pRes);
#endif
pRes = taos_query(pConn, "insert into d0 (ts, current) values (now, 120)"); pRes = taos_query(pConn, "insert into d0 (ts, current) values (now, 120)");
if (taos_errno(pRes) != 0) { if (taos_errno(pRes) != 0) {
@ -598,8 +623,8 @@ void basic_consume_loop(tmq_t* tmq, tmq_list_t* topics) {
} }
void initLogFile() { void initLogFile() {
char f1[256] = {0}; char f1[1024] = {0};
char f2[256] = {0}; char f2[1024] = {0};
if (g_conf.snapShot) { if (g_conf.snapShot) {
sprintf(f1, "%s/../log/tmq_taosx_tmp_snapshot.source", g_conf.dir); sprintf(f1, "%s/../log/tmq_taosx_tmp_snapshot.source", g_conf.dir);