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

Merge branch main into FIX/TD-21218-main

This commit is contained in:
Benguang Zhao 2023-01-18 10:15:16 +08:00
parent d611793537 78d674e9c7
commit e55a76e25a
29 changed files with 156 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Jenkinsfile2
View File

@ -430,7 +430,7 @@ pipeline {
date date
rm -rf ${WKC}/debug rm -rf ${WKC}/debug
cd ${WKC}/tests/parallel_test cd ${WKC}/tests/parallel_test
time ./container_build.sh -w ${WKDIR} -t 10 -e time ./container_build.sh -w ${WKDIR} -e
''' '''
def extra_param = "" def extra_param = ""
def log_server_file = "/home/log_server.json" def log_server_file = "/home/log_server.json"

2
include/libs/sync/sync.h
View File

@ -193,7 +193,7 @@ typedef struct SSyncLogStore {
SyncIndex (*syncLogLastIndex)(struct SSyncLogStore* pLogStore); SyncIndex (*syncLogLastIndex)(struct SSyncLogStore* pLogStore);
SyncTerm (*syncLogLastTerm)(struct SSyncLogStore* pLogStore); SyncTerm (*syncLogLastTerm)(struct SSyncLogStore* pLogStore);
int32_t (*syncLogAppendEntry)(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry); int32_t (*syncLogAppendEntry)(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry, bool forcSync);
int32_t (*syncLogGetEntry)(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncRaftEntry** ppEntry); int32_t (*syncLogGetEntry)(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncRaftEntry** ppEntry);
int32_t (*syncLogTruncate)(struct SSyncLogStore* pLogStore, SyncIndex fromIndex); int32_t (*syncLogTruncate)(struct SSyncLogStore* pLogStore, SyncIndex fromIndex);

2
packaging/release.sh
View File

@ -2,7 +2,7 @@
# #
# Generate the deb package for ubuntu, or rpm package for centos, or tar.gz package for other linux os # Generate the deb package for ubuntu, or rpm package for centos, or tar.gz package for other linux os
set -e # set -e
# set -x # set -x
# release.sh -v [cluster | edge] # release.sh -v [cluster | edge]

70
source/client/src/clientEnv.c
View File

@ -357,6 +357,7 @@ void doDestroyRequest(void *p) {
taosMemoryFreeClear(pRequest->pDb); taosMemoryFreeClear(pRequest->pDb);
doFreeReqResultInfo(&pRequest->body.resInfo); doFreeReqResultInfo(&pRequest->body.resInfo);
tsem_destroy(&pRequest->body.rspSem);
taosArrayDestroy(pRequest->tableList); taosArrayDestroy(pRequest->tableList);
taosArrayDestroy(pRequest->dbList); taosArrayDestroy(pRequest->dbList);
@ -371,6 +372,9 @@ void doDestroyRequest(void *p) {
} }
if (pRequest->syncQuery) { if (pRequest->syncQuery) {
if (pRequest->body.param){
tsem_destroy(&((SSyncQueryParam*)pRequest->body.param)->sem);
}
taosMemoryFree(pRequest->body.param); taosMemoryFree(pRequest->body.param);
} }
@ -388,45 +392,6 @@ void destroyRequest(SRequestObj *pRequest) {
removeRequest(pRequest->self); removeRequest(pRequest->self);
} }
void taosClientCrash(int signum, void *sigInfo, void *context) {
taosIgnSignal(SIGTERM);
taosIgnSignal(SIGHUP);
taosIgnSignal(SIGINT);
taosIgnSignal(SIGBREAK);
#if !defined(WINDOWS)
taosIgnSignal(SIGBUS);
#endif
taosIgnSignal(SIGABRT);
taosIgnSignal(SIGFPE);
taosIgnSignal(SIGSEGV);
char *pMsg = NULL;
const char *flags = "UTL FATAL ";
ELogLevel level = DEBUG_FATAL;
int32_t dflag = 255;
int64_t msgLen= -1;
if (tsEnableCrashReport) {
if (taosGenCrashJsonMsg(signum, &pMsg, lastClusterId, appInfo.startTime)) {
taosPrintLog(flags, level, dflag, "failed to generate crash json msg");
goto _return;
} else {
msgLen = strlen(pMsg);
}
}
_return:
taosLogCrashInfo("taos", pMsg, msgLen, signum, sigInfo);
#ifdef _TD_DARWIN_64
exit(signum);
#elif defined(WINDOWS)
exit(signum);
#endif
}
void crashReportThreadFuncUnexpectedStopped(void) { atomic_store_32(&clientStop, -1); } void crashReportThreadFuncUnexpectedStopped(void) { atomic_store_32(&clientStop, -1); }
static void *tscCrashReportThreadFp(void *param) { static void *tscCrashReportThreadFp(void *param) {
@ -523,14 +488,25 @@ void tscStopCrashReport() {
} }
} }
static void tscSetSignalHandle() {
#if !defined(WINDOWS) void tscWriteCrashInfo(int signum, void *sigInfo, void *context) {
taosSetSignal(SIGBUS, taosClientCrash); char *pMsg = NULL;
#endif const char *flags = "UTL FATAL ";
taosSetSignal(SIGABRT, taosClientCrash); ELogLevel level = DEBUG_FATAL;
taosSetSignal(SIGFPE, taosClientCrash); int32_t dflag = 255;
taosSetSignal(SIGSEGV, taosClientCrash); int64_t msgLen= -1;
if (tsEnableCrashReport) {
if (taosGenCrashJsonMsg(signum, &pMsg, lastClusterId, appInfo.startTime)) {
taosPrintLog(flags, level, dflag, "failed to generate crash json msg");
} else {
msgLen = strlen(pMsg);
} }
}
taosLogCrashInfo("taos", pMsg, msgLen, signum, sigInfo);
}
void taos_init_imp(void) { void taos_init_imp(void) {
// In the APIs of other program language, taos_cleanup is not available yet. // In the APIs of other program language, taos_cleanup is not available yet.
@ -555,8 +531,6 @@ void taos_init_imp(void) {
return; return;
} }
tscSetSignalHandle();
initQueryModuleMsgHandle(); initQueryModuleMsgHandle();
if (taosConvInit() != 0) { if (taosConvInit() != 0) {

21
source/client/src/clientImpl.c
View File

@ -159,6 +159,12 @@ STscObj* taos_connect_internal(const char* ip, const char* user, const char* pas
return taosConnectImpl(user, &secretEncrypt[0], localDb, NULL, NULL, *pInst, connType); return taosConnectImpl(user, &secretEncrypt[0], localDb, NULL, NULL, *pInst, connType);
} }
void freeQueryParam(SSyncQueryParam* param) {
if (param == NULL) return;
tsem_destroy(&param->sem);
taosMemoryFree(param);
}
int32_t buildRequest(uint64_t connId, const char* sql, int sqlLen, void* param, bool validateSql, int32_t buildRequest(uint64_t connId, const char* sql, int sqlLen, void* param, bool validateSql,
SRequestObj** pRequest, int64_t reqid) { SRequestObj** pRequest, int64_t reqid) {
*pRequest = createRequest(connId, TSDB_SQL_SELECT, reqid); *pRequest = createRequest(connId, TSDB_SQL_SELECT, reqid);
@ -180,17 +186,18 @@ int32_t buildRequest(uint64_t connId, const char* sql, int sqlLen, void* param,
(*pRequest)->sqlLen = sqlLen; (*pRequest)->sqlLen = sqlLen;
(*pRequest)->validateOnly = validateSql; (*pRequest)->validateOnly = validateSql;
SSyncQueryParam* newpParam;
if (param == NULL) { if (param == NULL) {
SSyncQueryParam* pParam = taosMemoryCalloc(1, sizeof(SSyncQueryParam)); newpParam = taosMemoryCalloc(1, sizeof(SSyncQueryParam));
if (pParam == NULL) { if (newpParam == NULL) {
destroyRequest(*pRequest); destroyRequest(*pRequest);
*pRequest = NULL; *pRequest = NULL;
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
} }
tsem_init(&pParam->sem, 0, 0); tsem_init(&newpParam->sem, 0, 0);
pParam->pRequest = (*pRequest); newpParam->pRequest = (*pRequest);
param = pParam; param = newpParam;
} }
(*pRequest)->body.param = param; (*pRequest)->body.param = param;
@ -201,8 +208,7 @@ int32_t buildRequest(uint64_t connId, const char* sql, int sqlLen, void* param,
if (err) { if (err) {
tscError("%" PRId64 " failed to add to request container, reqId:0x%" PRIx64 ", conn:%" PRId64 ", %s", tscError("%" PRId64 " failed to add to request container, reqId:0x%" PRIx64 ", conn:%" PRId64 ", %s",
(*pRequest)->self, (*pRequest)->requestId, pTscObj->id, sql); (*pRequest)->self, (*pRequest)->requestId, pTscObj->id, sql);
freeQueryParam(newpParam);
taosMemoryFree(param);
destroyRequest(*pRequest); destroyRequest(*pRequest);
*pRequest = NULL; *pRequest = NULL;
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
@ -214,6 +220,7 @@ int32_t buildRequest(uint64_t connId, const char* sql, int sqlLen, void* param,
nodesCreateAllocator((*pRequest)->requestId, tsQueryNodeChunkSize, &((*pRequest)->allocatorRefId))) { nodesCreateAllocator((*pRequest)->requestId, tsQueryNodeChunkSize, &((*pRequest)->allocatorRefId))) {
tscError("%" PRId64 " failed to create node allocator, reqId:0x%" PRIx64 ", conn:%" PRId64 ", %s", tscError("%" PRId64 " failed to create node allocator, reqId:0x%" PRIx64 ", conn:%" PRId64 ", %s",
(*pRequest)->self, (*pRequest)->requestId, pTscObj->id, sql); (*pRequest)->self, (*pRequest)->requestId, pTscObj->id, sql);
freeQueryParam(newpParam);
destroyRequest(*pRequest); destroyRequest(*pRequest);
*pRequest = NULL; *pRequest = NULL;
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;

3
source/client/src/clientMain.c
View File

@ -509,9 +509,8 @@ void taos_stop_query(TAOS_RES *res) {
SRequestObj *pRequest = (SRequestObj *)res; SRequestObj *pRequest = (SRequestObj *)res;
pRequest->killed = true; pRequest->killed = true;
int32_t numOfFields = taos_num_fields(pRequest);
// It is not a query, no need to stop. // It is not a query, no need to stop.
if (numOfFields == 0) { if (NULL == pRequest->pQuery || QUERY_EXEC_MODE_SCHEDULE != pRequest->pQuery->execMode) {
tscDebug("request 0x%" PRIx64 " no need to be killed since not query", pRequest->requestId); tscDebug("request 0x%" PRIx64 " no need to be killed since not query", pRequest->requestId);
return; return;
} }

4
source/client/src/clientStmt.c
View File

@ -300,11 +300,7 @@ int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable, bool deepClean) {
continue; continue;
} }
if (STMT_TYPE_MULTI_INSERT == pStmt->sql.type) {
qFreeStmtDataBlock(pBlocks);
} else {
qDestroyStmtDataBlock(pBlocks); qDestroyStmtDataBlock(pBlocks);
}
taosHashRemove(pStmt->exec.pBlockHash, key, keyLen); taosHashRemove(pStmt->exec.pBlockHash, key, keyLen);
pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter); pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);

10
source/dnode/vnode/src/tsdb/tsdbCacheRead.c
View File

@ -241,7 +241,11 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
taosArrayPush(pLastCols, &p); taosArrayPush(pLastCols, &p);
} }
tsdbTakeReadSnap(pr->pVnode->pTsdb, &pr->pReadSnap, "cache-l"); code = tsdbTakeReadSnap(pr->pVnode->pTsdb, &pr->pReadSnap, "cache-l");
if (code != TSDB_CODE_SUCCESS) {
goto _end;
}
pr->pDataFReader = NULL; pr->pDataFReader = NULL;
pr->pDataFReaderLast = NULL; pr->pDataFReaderLast = NULL;
@ -252,7 +256,7 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
code = doExtractCacheRow(pr, lruCache, pKeyInfo->uid, &pRow, &h); code = doExtractCacheRow(pr, lruCache, pKeyInfo->uid, &pRow, &h);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
return code; goto _end;
} }
if (h == NULL) { if (h == NULL) {
@ -321,7 +325,7 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
STableKeyInfo* pKeyInfo = &pr->pTableList[i]; STableKeyInfo* pKeyInfo = &pr->pTableList[i];
code = doExtractCacheRow(pr, lruCache, pKeyInfo->uid, &pRow, &h); code = doExtractCacheRow(pr, lruCache, pKeyInfo->uid, &pRow, &h);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
return code; goto _end;
} }
if (h == NULL) { if (h == NULL) {

4
source/dnode/vnode/src/tsdb/tsdbRead.c
View File

@ -3884,7 +3884,7 @@ int32_t tsdbReaderOpen(SVnode* pVnode, SQueryTableDataCond* pCond, void* pTableL
if (pReader->type == TIMEWINDOW_RANGE_CONTAINED) { if (pReader->type == TIMEWINDOW_RANGE_CONTAINED) {
code = doOpenReaderImpl(pReader); code = doOpenReaderImpl(pReader);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
return code; goto _err;
} }
} else { } else {
STsdbReader* pPrevReader = pReader->innerReader[0]; STsdbReader* pPrevReader = pReader->innerReader[0];
@ -3905,7 +3905,7 @@ int32_t tsdbReaderOpen(SVnode* pVnode, SQueryTableDataCond* pCond, void* pTableL
code = doOpenReaderImpl(pPrevReader); code = doOpenReaderImpl(pPrevReader);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
return code; goto _err;
} }
} }
} }

1
source/libs/catalog/inc/catalogInt.h
View File

@ -805,6 +805,7 @@ int32_t ctgMakeVgArray(SDBVgInfo* dbInfo);
int32_t ctgAcquireVgMetaFromCache(SCatalog *pCtg, const char *dbFName, const char *tbName, SCtgDBCache **pDb, SCtgTbCache **pTb); int32_t ctgAcquireVgMetaFromCache(SCatalog *pCtg, const char *dbFName, const char *tbName, SCtgDBCache **pDb, SCtgTbCache **pTb);
int32_t ctgCopyTbMeta(SCatalog *pCtg, SCtgTbMetaCtx *ctx, SCtgDBCache **pDb, SCtgTbCache **pTb, STableMeta **pTableMeta, char* dbFName); int32_t ctgCopyTbMeta(SCatalog *pCtg, SCtgTbMetaCtx *ctx, SCtgDBCache **pDb, SCtgTbCache **pTb, STableMeta **pTableMeta, char* dbFName);
void ctgReleaseVgMetaToCache(SCatalog *pCtg, SCtgDBCache *dbCache, SCtgTbCache *pCache); void ctgReleaseVgMetaToCache(SCatalog *pCtg, SCtgDBCache *dbCache, SCtgTbCache *pCache);
void ctgReleaseTbMetaToCache(SCatalog *pCtg, SCtgDBCache *dbCache, SCtgTbCache *pCache);
extern SCatalogMgmt gCtgMgmt; extern SCatalogMgmt gCtgMgmt;
extern SCtgDebug gCTGDebug; extern SCtgDebug gCTGDebug;

8
source/libs/catalog/src/catalog.c
View File

@ -598,10 +598,16 @@ int32_t ctgGetCachedTbVgMeta(SCatalog* pCtg, const SName* pTableName, SVgroupInf
CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCtg, dbCache->vgCache.vgInfo, pTableName, pVgroup)); CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCtg, dbCache->vgCache.vgInfo, pTableName, pVgroup));
ctgRUnlockVgInfo(dbCache);
SCtgTbMetaCtx ctx = {0}; SCtgTbMetaCtx ctx = {0};
ctx.pName = (SName*)pTableName; ctx.pName = (SName*)pTableName;
ctx.flag = CTG_FLAG_UNKNOWN_STB; ctx.flag = CTG_FLAG_UNKNOWN_STB;
CTG_ERR_JRET(ctgCopyTbMeta(pCtg, &ctx, &dbCache, &tbCache, pTableMeta, db)); code = ctgCopyTbMeta(pCtg, &ctx, &dbCache, &tbCache, pTableMeta, db);
ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
CTG_RET(code);
_return: _return:

2
source/libs/catalog/src/ctgAsync.c
View File

@ -999,6 +999,7 @@ int32_t ctgHandleGetTbMetaRsp(SCtgTaskReq* tReq, int32_t reqType, const SDataBuf
CTG_ERR_JRET(ctgGetTbMetaFromVnode(pCtg, pConn, pName, &vgInfo, NULL, tReq)); CTG_ERR_JRET(ctgGetTbMetaFromVnode(pCtg, pConn, pName, &vgInfo, NULL, tReq));
ctgReleaseVgInfoToCache(pCtg, dbCache); ctgReleaseVgInfoToCache(pCtg, dbCache);
dbCache = NULL;
} else { } else {
SBuildUseDBInput input = {0}; SBuildUseDBInput input = {0};
@ -1168,6 +1169,7 @@ int32_t ctgHandleGetTbMetasRsp(SCtgTaskReq* tReq, int32_t reqType, const SDataBu
CTG_ERR_JRET(ctgGetTbMetaFromVnode(pCtg, pConn, pName, &vgInfo, NULL, tReq)); CTG_ERR_JRET(ctgGetTbMetaFromVnode(pCtg, pConn, pName, &vgInfo, NULL, tReq));
ctgReleaseVgInfoToCache(pCtg, dbCache); ctgReleaseVgInfoToCache(pCtg, dbCache);
dbCache = NULL;
} else { } else {
SBuildUseDBInput input = {0}; SBuildUseDBInput input = {0};

2
source/libs/catalog/src/ctgCache.c
View File

@ -2118,7 +2118,7 @@ int32_t ctgOpUpdateEpset(SCtgCacheOperation *operation) {
_return: _return:
if (dbCache) { if (code == TSDB_CODE_SUCCESS && dbCache) {
ctgWUnlockVgInfo(dbCache); ctgWUnlockVgInfo(dbCache);
} }

2
source/libs/executor/src/executil.c
View File

@ -954,7 +954,7 @@ static int32_t optimizeTbnameInCondImpl(void* metaHandle, int64_t suid, SArray*
return -1; return -1;
} }
} else { } else {
qWarn("failed to get tableIds from by table name: %s, reason: %s", name, tstrerror(terrno)); // qWarn("failed to get tableIds from by table name: %s, reason: %s", name, tstrerror(terrno));
terrno = 0; terrno = 0;
} }
} }

23
source/libs/parser/src/parInsertStmt.c
View File

@ -425,6 +425,27 @@ int32_t qCloneStmtDataBlock(void** pDst, void* pSrc) {
pBlock->pTableMeta = pNewMeta; pBlock->pTableMeta = pNewMeta;
} }
if (pBlock->boundColumnInfo.boundColumns) {
int32_t size = pBlock->boundColumnInfo.numOfCols * sizeof(col_id_t);
void* tmp = taosMemoryMalloc(size);
memcpy(tmp, pBlock->boundColumnInfo.boundColumns, size);
pBlock->boundColumnInfo.boundColumns = tmp;
}
if (pBlock->boundColumnInfo.cols) {
int32_t size = pBlock->boundColumnInfo.numOfCols * sizeof(SBoundColumn);
void* tmp = taosMemoryMalloc(size);
memcpy(tmp, pBlock->boundColumnInfo.cols, size);
pBlock->boundColumnInfo.cols = tmp;
}
if (pBlock->boundColumnInfo.colIdxInfo) {
int32_t size = pBlock->boundColumnInfo.numOfBound * sizeof(SBoundIdxInfo);
void* tmp = taosMemoryMalloc(size);
memcpy(tmp, pBlock->boundColumnInfo.colIdxInfo, size);
pBlock->boundColumnInfo.colIdxInfo = tmp;
}
return qResetStmtDataBlock(*pDst, false); return qResetStmtDataBlock(*pDst, false);
} }
@ -437,7 +458,7 @@ int32_t qRebuildStmtDataBlock(void** pDst, void* pSrc, uint64_t uid, int32_t vgI
STableDataBlocks* pBlock = (STableDataBlocks*)*pDst; STableDataBlocks* pBlock = (STableDataBlocks*)*pDst;
pBlock->pData = taosMemoryMalloc(pBlock->nAllocSize); pBlock->pData = taosMemoryMalloc(pBlock->nAllocSize);
if (NULL == pBlock->pData) { if (NULL == pBlock->pData) {
qFreeStmtDataBlock(pBlock); qDestroyStmtDataBlock(pBlock);
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
} }

7
source/libs/qworker/inc/qwInt.h
View File

@ -228,9 +228,14 @@ typedef struct SQWorkerMgmt {
case QW_PHASE_POST_FETCH: \ case QW_PHASE_POST_FETCH: \
ctx->inFetch = 0; \ ctx->inFetch = 0; \
break; \ break; \
default: \ case QW_PHASE_PRE_QUERY: \
case QW_PHASE_POST_QUERY: \
case QW_PHASE_PRE_CQUERY: \
case QW_PHASE_POST_CQUERY: \
atomic_store_8(&(ctx)->phase, _value); \ atomic_store_8(&(ctx)->phase, _value); \
break; \ break; \
default: \
break; \
} \ } \
} while (0) } while (0)

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

@ -551,7 +551,9 @@ _return:
if (ctx) { if (ctx) {
QW_UPDATE_RSP_CODE(ctx, code); QW_UPDATE_RSP_CODE(ctx, code);
if (QW_PHASE_POST_CQUERY != phase) {
QW_SET_PHASE(ctx, phase); QW_SET_PHASE(ctx, phase);
}
QW_UNLOCK(QW_WRITE, &ctx->lock); QW_UNLOCK(QW_WRITE, &ctx->lock);
qwReleaseTaskCtx(mgmt, ctx); qwReleaseTaskCtx(mgmt, ctx);
@ -758,7 +760,7 @@ int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
QW_LOCK(QW_WRITE, &ctx->lock); QW_LOCK(QW_WRITE, &ctx->lock);
if (qComplete || (queryStop && (0 == atomic_load_8((int8_t *)&ctx->queryContinue))) || code) { if (qComplete || (queryStop && (0 == atomic_load_8((int8_t *)&ctx->queryContinue))) || code) {
// Note: query is not running anymore // Note: query is not running anymore
QW_SET_PHASE(ctx, 0); QW_SET_PHASE(ctx, QW_PHASE_POST_CQUERY);
QW_UNLOCK(QW_WRITE, &ctx->lock); QW_UNLOCK(QW_WRITE, &ctx->lock);
break; break;
} }

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

@ -2387,7 +2387,7 @@ static int32_t syncNodeAppendNoopOld(SSyncNode* ths) {
LRUHandle* h = NULL; LRUHandle* h = NULL;
if (ths->state == TAOS_SYNC_STATE_LEADER) { if (ths->state == TAOS_SYNC_STATE_LEADER) {
int32_t code = ths->pLogStore->syncLogAppendEntry(ths->pLogStore, pEntry); int32_t code = ths->pLogStore->syncLogAppendEntry(ths->pLogStore, pEntry, false);
if (code != 0) { if (code != 0) {
sError("append noop error"); sError("append noop error");
return -1; return -1;

11
source/libs/sync/src/syncPipeline.c
View File

@ -364,7 +364,11 @@ _out:
return ret; return ret;
} }
int32_t syncLogStorePersist(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry) { static inline bool syncLogStoreNeedFlush(SSyncRaftEntry* pEntry, int32_t replicaNum) {
return (replicaNum > 1) && (pEntry->originalRpcType == TDMT_VND_COMMIT);
}
int32_t syncLogStorePersist(SSyncLogStore* pLogStore, SSyncNode* pNode, SSyncRaftEntry* pEntry) {
ASSERT(pEntry->index >= 0); ASSERT(pEntry->index >= 0);
SyncIndex lastVer = pLogStore->syncLogLastIndex(pLogStore); SyncIndex lastVer = pLogStore->syncLogLastIndex(pLogStore);
if (lastVer >= pEntry->index && pLogStore->syncLogTruncate(pLogStore, pEntry->index) < 0) { if (lastVer >= pEntry->index && pLogStore->syncLogTruncate(pLogStore, pEntry->index) < 0) {
@ -374,7 +378,8 @@ int32_t syncLogStorePersist(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry) {
lastVer = pLogStore->syncLogLastIndex(pLogStore); lastVer = pLogStore->syncLogLastIndex(pLogStore);
ASSERT(pEntry->index == lastVer + 1); ASSERT(pEntry->index == lastVer + 1);
if (pLogStore->syncLogAppendEntry(pLogStore, pEntry) < 0) { bool doFsync = syncLogStoreNeedFlush(pEntry, pNode->replicaNum);
if (pLogStore->syncLogAppendEntry(pLogStore, pEntry, doFsync) < 0) {
sError("failed to append sync log entry since %s. index:%" PRId64 ", term:%" PRId64 "", terrstr(), pEntry->index, sError("failed to append sync log entry since %s. index:%" PRId64 ", term:%" PRId64 "", terrstr(), pEntry->index,
pEntry->term); pEntry->term);
return -1; return -1;
@ -436,7 +441,7 @@ int64_t syncLogBufferProceed(SSyncLogBuffer* pBuf, SSyncNode* pNode, SyncTerm* p
(void)syncNodeReplicateWithoutLock(pNode); (void)syncNodeReplicateWithoutLock(pNode);
// persist // persist
if (syncLogStorePersist(pLogStore, pEntry) < 0) { if (syncLogStorePersist(pLogStore, pNode, pEntry) < 0) {
sError("vgId:%d, failed to persist sync log entry from buffer since %s. index:%" PRId64, pNode->vgId, terrstr(), sError("vgId:%d, failed to persist sync log entry from buffer since %s. index:%" PRId64, pNode->vgId, terrstr(),
pEntry->index); pEntry->index);
goto _out; goto _out;

7
source/libs/sync/src/syncRaftLog.c
View File

@ -23,7 +23,7 @@
// public function // public function
static int32_t raftLogRestoreFromSnapshot(struct SSyncLogStore* pLogStore, SyncIndex snapshotIndex); static int32_t raftLogRestoreFromSnapshot(struct SSyncLogStore* pLogStore, SyncIndex snapshotIndex);
static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry); static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry, bool forceSync);
static int32_t raftLogTruncate(struct SSyncLogStore* pLogStore, SyncIndex fromIndex); static int32_t raftLogTruncate(struct SSyncLogStore* pLogStore, SyncIndex fromIndex);
static bool raftLogExist(struct SSyncLogStore* pLogStore, SyncIndex index); static bool raftLogExist(struct SSyncLogStore* pLogStore, SyncIndex index);
static int32_t raftLogUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index); static int32_t raftLogUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index);
@ -192,9 +192,7 @@ SyncTerm raftLogLastTerm(struct SSyncLogStore* pLogStore) {
return SYNC_TERM_INVALID; return SYNC_TERM_INVALID;
} }
static inline bool raftLogForceSync(SSyncRaftEntry* pEntry) { return (pEntry->originalRpcType == TDMT_VND_COMMIT); } static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry, bool forceSync) {
static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry) {
SSyncLogStoreData* pData = pLogStore->data; SSyncLogStoreData* pData = pLogStore->data;
SWal* pWal = pData->pWal; SWal* pWal = pData->pWal;
@ -221,7 +219,6 @@ static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntr
ASSERT(pEntry->index == index); ASSERT(pEntry->index == index);
bool forceSync = raftLogForceSync(pEntry);
walFsync(pWal, forceSync); walFsync(pWal, forceSync);
sNTrace(pData->pSyncNode, "write index:%" PRId64 ", type:%s, origin type:%s, elapsed:%" PRId64, pEntry->index, sNTrace(pData->pSyncNode, "write index:%" PRId64 ", type:%s, origin type:%s, elapsed:%" PRId64, pEntry->index,

6
source/os/src/osSysinfo.c
View File

@ -834,7 +834,11 @@ int32_t taosGetSystemUUID(char *uid, int32_t uidlen) {
uuid_generate(uuid); uuid_generate(uuid);
// it's caller's responsibility to make enough space for `uid`, that's 36-char + 1-null // it's caller's responsibility to make enough space for `uid`, that's 36-char + 1-null
uuid_unparse_lower(uuid, buf); uuid_unparse_lower(uuid, buf);
memcpy(uid, buf, uidlen); int n = snprintf(uid, uidlen, "%.*s", (int)sizeof(buf), buf); // though less performance, much safer
if (n >= uidlen) {
// target buffer is too small
return -1;
}
return 0; return 0;
#else #else
int len = 0; int len = 0;

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

@ -897,6 +897,7 @@ void taosLogCrashInfo(char* nodeType, char* pMsg, int64_t msgLen, int signum, vo
pFile = taosOpenFile(filepath, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND); pFile = taosOpenFile(filepath, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND);
if (pFile == NULL) { if (pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
taosPrintLog(flags, level, dflag, "failed to open file:%s since %s", filepath, terrstr()); taosPrintLog(flags, level, dflag, "failed to open file:%s since %s", filepath, terrstr());
goto _return; goto _return;
} }

1
tests/parallel_test/cases.task
View File

@ -145,6 +145,7 @@
,,y,script,./test.sh -f tsim/parser/precision_ns.sim ,,y,script,./test.sh -f tsim/parser/precision_ns.sim
,,y,script,./test.sh -f tsim/parser/projection_limit_offset.sim ,,y,script,./test.sh -f tsim/parser/projection_limit_offset.sim
,,y,script,./test.sh -f tsim/parser/regex.sim ,,y,script,./test.sh -f tsim/parser/regex.sim
,,y,script,./test.sh -f tsim/parser/regressiontest.sim
,,y,script,./test.sh -f tsim/parser/select_across_vnodes.sim ,,y,script,./test.sh -f tsim/parser/select_across_vnodes.sim
,,y,script,./test.sh -f tsim/parser/select_distinct_tag.sim ,,y,script,./test.sh -f tsim/parser/select_distinct_tag.sim
,,y,script,./test.sh -f tsim/parser/select_from_cache_disk.sim ,,y,script,./test.sh -f tsim/parser/select_from_cache_disk.sim

10
tests/parallel_test/container_build.sh
View File

@ -37,9 +37,9 @@ if [ -z "$WORKDIR" ]; then
usage usage
exit 1 exit 1
fi fi
if [ -z "$THREAD_COUNT" ]; then # if [ -z "$THREAD_COUNT" ]; then
THREAD_COUNT=1 # THREAD_COUNT=1
fi # fi
ulimit -c unlimited ulimit -c unlimited
@ -55,7 +55,7 @@ fi
date date
docker run \ docker run \
-v $REP_MOUNT_PARAM \ -v $REP_MOUNT_PARAM \
--rm --ulimit core=-1 taos_test:v1.0 sh -c "cd $REP_DIR;rm -rf debug;mkdir -p debug;cd debug;cmake .. -DBUILD_HTTP=false -DBUILD_TOOLS=true -DBUILD_TEST=true -DWEBSOCKET=true;make -j $THREAD_COUNT || exit 1" --rm --ulimit core=-1 taos_test:v1.0 sh -c "cd $REP_DIR;rm -rf debug;mkdir -p debug;cd debug;cmake .. -DBUILD_HTTP=false -DBUILD_TOOLS=true -DBUILD_TEST=true -DWEBSOCKET=true;make -j || exit 1"
if [[ -d ${WORKDIR}/debugNoSan ]] ;then if [[ -d ${WORKDIR}/debugNoSan ]] ;then
echo "delete ${WORKDIR}/debugNoSan" echo "delete ${WORKDIR}/debugNoSan"
@ -70,7 +70,7 @@ mv ${REP_REAL_PATH}/debug ${WORKDIR}/debugNoSan
date date
docker run \ docker run \
-v $REP_MOUNT_PARAM \ -v $REP_MOUNT_PARAM \
--rm --ulimit core=-1 taos_test:v1.0 sh -c "cd $REP_DIR;rm -rf debug;mkdir -p debug;cd debug;cmake .. -DBUILD_HTTP=false -DBUILD_TOOLS=true -DBUILD_TEST=true -DWEBSOCKET=true -DBUILD_SANITIZER=1 -DTOOLS_SANITIZE=true -DTOOLS_BUILD_TYPE=Debug;make -j $THREAD_COUNT || exit 1 " --rm --ulimit core=-1 taos_test:v1.0 sh -c "cd $REP_DIR;rm -rf debug;mkdir -p debug;cd debug;cmake .. -DBUILD_HTTP=false -DBUILD_TOOLS=true -DBUILD_TEST=true -DWEBSOCKET=true -DBUILD_SANITIZER=1 -DTOOLS_SANITIZE=true -DTOOLS_BUILD_TYPE=Debug;make -j || exit 1 "
mv ${REP_REAL_PATH}/debug ${WORKDIR}/debugSan mv ${REP_REAL_PATH}/debug ${WORKDIR}/debugSan

4
tests/parallel_test/run.sh
View File

@ -184,6 +184,10 @@ function run_thread() {
if [ $? -eq 0 ]; then if [ $? -eq 0 ]; then
case_file=`echo "$case_cmd"|grep -o ".*\.py"|awk '{print $NF}'` case_file=`echo "$case_cmd"|grep -o ".*\.py"|awk '{print $NF}'`
fi fi
echo "$case_cmd"|grep -q "^./pytest.sh"
if [ $? -eq 0 ]; then
case_file=`echo "$case_cmd"|grep -o ".*\.py"|awk '{print $NF}'`
fi
echo "$case_cmd"|grep -q "\.sim" echo "$case_cmd"|grep -q "\.sim"
if [ $? -eq 0 ]; then if [ $? -eq 0 ]; then
case_file=`echo "$case_cmd"|grep -o ".*\.sim"|awk '{print $NF}'` case_file=`echo "$case_cmd"|grep -o ".*\.sim"|awk '{print $NF}'`

2
tests/script/api/batchprepare.c
View File

@ -2828,7 +2828,7 @@ void runAll(TAOS *taos) {
printf("%s Begin\n", gCaseCtrl.caseCatalog); printf("%s Begin\n", gCaseCtrl.caseCatalog);
runCaseList(taos); runCaseList(taos);
#if 0 #if 1
strcpy(gCaseCtrl.caseCatalog, "Micro DB precision Test"); strcpy(gCaseCtrl.caseCatalog, "Micro DB precision Test");
printf("%s Begin\n", gCaseCtrl.caseCatalog); printf("%s Begin\n", gCaseCtrl.caseCatalog);
gCaseCtrl.precision = TIME_PRECISION_MICRO; gCaseCtrl.precision = TIME_PRECISION_MICRO;

1
tools/shell/inc/shellInt.h
View File

@ -147,5 +147,6 @@ void shellRunSingleCommandWebsocketImp(char *command);
// shellMain.c // shellMain.c
extern SShellObj shell; extern SShellObj shell;
extern void tscWriteCrashInfo(int signum, void *sigInfo, void *context);
#endif /*_TD_SHELL_INT_H_*/ #endif /*_TD_SHELL_INT_H_*/

2
tools/shell/src/shellEngine.c
View File

@ -1136,8 +1136,6 @@ int32_t shellExecute() {
taosSetSignal(SIGTERM, shellQueryInterruptHandler); taosSetSignal(SIGTERM, shellQueryInterruptHandler);
taosSetSignal(SIGHUP, shellQueryInterruptHandler); taosSetSignal(SIGHUP, shellQueryInterruptHandler);
taosSetSignal(SIGABRT, shellQueryInterruptHandler);
taosSetSignal(SIGINT, shellQueryInterruptHandler); taosSetSignal(SIGINT, shellQueryInterruptHandler);
#ifdef WEBSOCKET #ifdef WEBSOCKET

30
tools/shell/src/shellMain.c
View File

@ -19,6 +19,29 @@
SShellObj shell = {0}; SShellObj shell = {0};
void shellCrashHandler(int signum, void *sigInfo, void *context) {
taosIgnSignal(SIGTERM);
taosIgnSignal(SIGHUP);
taosIgnSignal(SIGINT);
taosIgnSignal(SIGBREAK);
#if !defined(WINDOWS)
taosIgnSignal(SIGBUS);
#endif
taosIgnSignal(SIGABRT);
taosIgnSignal(SIGFPE);
taosIgnSignal(SIGSEGV);
tscWriteCrashInfo(signum, sigInfo, context);
#ifdef _TD_DARWIN_64
exit(signum);
#elif defined(WINDOWS)
exit(signum);
#endif
}
int main(int argc, char *argv[]) { int main(int argc, char *argv[]) {
shell.exit = false; shell.exit = false;
#ifdef WEBSOCKET #ifdef WEBSOCKET
@ -26,6 +49,13 @@ int main(int argc, char *argv[]) {
shell.args.cloud = true; shell.args.cloud = true;
#endif #endif
#if !defined(WINDOWS)
taosSetSignal(SIGBUS, shellCrashHandler);
#endif
taosSetSignal(SIGABRT, shellCrashHandler);
taosSetSignal(SIGFPE, shellCrashHandler);
taosSetSignal(SIGSEGV, shellCrashHandler);
if (shellCheckIntSize() != 0) { if (shellCheckIntSize() != 0) {
return -1; return -1;
} }