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

Merge branch '3.0' into fix/TD-31899

This commit is contained in:
WANG MINGMING 2024-09-13 14:31:53 +08:00 committed by GitHub
parent 1854663f92 ec2f2b68d9
commit bf3dbc1d4d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
57 changed files with 766 additions and 358 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
include/os/os.h
View File

@ -129,6 +129,14 @@ extern int64_t tsRandErrDivisor;
extern int64_t tsRandErrScope;
extern threadlocal bool tsEnableRandErr;
#define TAOS_UNUSED(expr) (void)(expr)
#define TAOS_SKIP_ERROR(expr) \
{ \
int32_t _code = terrno; \
(void)(expr); \
terrno = _code; \
}
#ifdef __cplusplus
}
#endif

2
include/os/osEnv.h
View File

@ -55,7 +55,7 @@ extern SDiskSpace tsLogSpace;
extern SDiskSpace tsTempSpace;
int32_t osDefaultInit();
void osUpdate();
int32_t osUpdate();
void osCleanup();
bool osLogSpaceAvailable();

30
include/os/osSemaphore.h
View File

@ -27,11 +27,11 @@ extern "C" {
// typedef struct tsem_s *tsem_t;
typedef dispatch_semaphore_t tsem_t;
int tsem_init(tsem_t *sem, int pshared, unsigned int value);
int tsem_wait(tsem_t *sem);
int tsem_timewait(tsem_t *sim, int64_t milis);
int tsem_post(tsem_t *sem);
int tsem_destroy(tsem_t *sem);
int32_t tsem_init(tsem_t *sem, int pshared, unsigned int value);
int32_t tsem_wait(tsem_t *sem);
int32_t tsem_timewait(tsem_t *sim, int64_t milis);
int32_t tsem_post(tsem_t *sem);
int32_t tsem_destroy(tsem_t *sem);
#define tsem2_t tsem_t
#define tsem2_init tsem_init
@ -45,11 +45,11 @@ int tsem_destroy(tsem_t *sem);
#define tsem_t HANDLE
int tsem_init(tsem_t *sem, int pshared, unsigned int value);
int tsem_wait(tsem_t *sem);
int tsem_timewait(tsem_t *sim, int64_t milis);
int tsem_post(tsem_t *sem);
int tsem_destroy(tsem_t *sem);
int32_t tsem_init(tsem_t *sem, int pshared, unsigned int value);
int32_t tsem_wait(tsem_t *sem);
int32_t tsem_timewait(tsem_t *sim, int64_t milis);
int32_t tsem_post(tsem_t *sem);
int32_t tsem_destroy(tsem_t *sem);
#define tsem2_t tsem_t
#define tsem2_init tsem_init
@ -61,11 +61,11 @@ int tsem_destroy(tsem_t *sem);
#else
#define tsem_t sem_t
#define tsem_init sem_init
int tsem_wait(tsem_t *sem);
int tsem_timewait(tsem_t *sim, int64_t milis);
#define tsem_post sem_post
#define tsem_destroy sem_destroy
int32_t tsem_init(tsem_t *sem, int pshared, unsigned int value);
int32_t tsem_wait(tsem_t *sem);
int32_t tsem_timewait(tsem_t *sim, int64_t milis);
int32_t tsem_post(tsem_t *sem);
int32_t tsem_destroy(tsem_t *sem);
typedef struct tsem2_t {
TdThreadMutex mutex;

5
include/os/osString.h
View File

@ -65,7 +65,10 @@ typedef enum { M2C = 0, C2M } ConvType;
char *tstrdup(const char *src);
int32_t taosUcs4len(TdUcs4 *ucs4);
int64_t taosStr2int64(const char *str);
int32_t taosStr2int64(const char *str, int64_t *val);
int32_t taosStr2int16(const char *str, int16_t *val);
int32_t taosStr2int32(const char *str, int32_t *val);
int32_t taosStr2int8(const char *str, int8_t *val);
int32_t taosConvInit(void);
void taosConvDestroy();

10
include/os/osSysinfo.h
View File

@ -39,15 +39,17 @@ int64_t taosGetOsUptime();
int32_t taosGetEmail(char *email, int32_t maxLen);
int32_t taosGetOsReleaseName(char *releaseName, char* sName, char* ver, int32_t maxLen);
int32_t taosGetCpuInfo(char *cpuModel, int32_t maxLen, float *numOfCores);
void taosGetCpuCores(float *numOfCores, bool physical);
void taosGetCpuUsage(double *cpu_system, double *cpu_engine);
int32_t taosGetCpuCores(float *numOfCores, bool physical);
int32_t taosGetCpuUsage(double *cpu_system, double *cpu_engine);
int32_t taosGetCpuInstructions(char* sse42, char* avx, char* avx2, char* fma, char* avx512);
int32_t taosGetTotalMemory(int64_t *totalKB);
int32_t taosGetProcMemory(int64_t *usedKB);
int32_t taosGetSysMemory(int64_t *usedKB);
int32_t taosGetDiskSize(char *dataDir, SDiskSize *diskSize);
void taosGetProcIODelta(int64_t *rchars, int64_t *wchars, int64_t *read_bytes, int64_t *write_bytes);
void taosGetCardInfoDelta(int64_t *receive_bytes, int64_t *transmit_bytes);
int32_t taosGetProcIODelta(int64_t *rchars, int64_t *wchars, int64_t *read_bytes, int64_t *write_bytes);
void taosSetDefaultProcIODelta(int64_t *rchars, int64_t *wchars, int64_t *read_bytes, int64_t *write_bytes);
int32_t taosGetCardInfoDelta(int64_t *receive_bytes, int64_t *transmit_bytes);
void taosSetDefaultCardInfoDelta(int64_t *receive_bytes, int64_t *transmit_bytes);
void taosKillSystem();
int32_t taosGetSystemUUID(char *uid, int32_t uidlen);

1
include/util/taoserror.h
View File

@ -156,6 +156,7 @@ int32_t taosGetErrSize();
#define TSDB_CODE_OUT_OF_BUFFER TAOS_DEF_ERROR_CODE(0, 0x0137)
#define TSDB_CODE_INTERNAL_ERROR TAOS_DEF_ERROR_CODE(0, 0x0138)
#define TSDB_CODE_SOCKET_ERROR TAOS_DEF_ERROR_CODE(0, 0x0139)
#define TSDB_CODE_UNSUPPORT_OS TAOS_DEF_ERROR_CODE(0, 0x013A)
//client
#define TSDB_CODE_TSC_INVALID_OPERATION TAOS_DEF_ERROR_CODE(0, 0x0200)

8
source/client/inc/clientStmt2.h
View File

@ -150,10 +150,10 @@ typedef struct {
SStmtExecInfo exec;
SStmtBindInfo bInfo;
int64_t reqid;
int32_t errCode;
tsem_t asyncQuerySem;
int64_t reqid;
int32_t errCode;
tsem_t asyncQuerySem;
bool semWaited;
SStmtStatInfo stat;
} STscStmt2;
/*

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

@ -2868,7 +2868,6 @@ TAOS_RES* taosQueryImpl(TAOS* taos, const char* sql, bool validateOnly, int8_t s
}
int32_t code = tsem_init(&param->sem, 0, 0);
if (TSDB_CODE_SUCCESS != code) {
terrno = code;
taosMemoryFree(param);
return NULL;
}
@ -2876,7 +2875,6 @@ TAOS_RES* taosQueryImpl(TAOS* taos, const char* sql, bool validateOnly, int8_t s
taosAsyncQueryImpl(*(int64_t*)taos, sql, syncQueryFn, param, validateOnly, source);
code = tsem_wait(&param->sem);
if (TSDB_CODE_SUCCESS != code) {
terrno = code;
taosMemoryFree(param);
return NULL;
}
@ -2907,7 +2905,6 @@ TAOS_RES* taosQueryImplWithReqid(TAOS* taos, const char* sql, bool validateOnly,
}
int32_t code = tsem_init(&param->sem, 0, 0);
if (TSDB_CODE_SUCCESS != code) {
terrno = code;
taosMemoryFree(param);
return NULL;
}
@ -2915,7 +2912,6 @@ TAOS_RES* taosQueryImplWithReqid(TAOS* taos, const char* sql, bool validateOnly,
taosAsyncQueryImplWithReqid(*(int64_t*)taos, sql, syncQueryFn, param, validateOnly, reqid);
code = tsem_wait(&param->sem);
if (TSDB_CODE_SUCCESS != code) {
terrno = code;
taosMemoryFree(param);
return NULL;
}

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

@ -1370,7 +1370,7 @@ typedef struct SAsyncFetchParam {
void *param;
} SAsyncFetchParam;
static int32_t doAsyncFetch(void* pParam) {
static int32_t doAsyncFetch(void *pParam) {
SAsyncFetchParam *param = pParam;
taosAsyncFetchImpl(param->pReq, param->fp, param->param);
taosMemoryFree(param);
@ -1394,7 +1394,7 @@ void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {
return;
}
SAsyncFetchParam* pParam = taosMemoryCalloc(1, sizeof(SAsyncFetchParam));
SAsyncFetchParam *pParam = taosMemoryCalloc(1, sizeof(SAsyncFetchParam));
if (!pParam) {
fp(param, res, terrno);
return;
@ -1984,6 +1984,12 @@ int taos_stmt2_bind_param(TAOS_STMT2 *stmt, TAOS_STMT2_BINDV *bindv, int32_t col
return terrno;
}
STscStmt2 *pStmt = (STscStmt2 *)stmt;
if (pStmt->options.asyncExecFn && !pStmt->semWaited) {
(void)tsem_wait(&pStmt->asyncQuerySem);
pStmt->semWaited = true;
}
int32_t code = 0;
for (int i = 0; i < bindv->count; ++i) {
if (bindv->tbnames && bindv->tbnames[i]) {

8
source/client/src/clientMonitor.c
View File

@ -34,19 +34,19 @@ static void processFileInTheEnd(TdFilePtr pFile, char* path) {
return;
}
if (taosFtruncateFile(pFile, 0) != 0) {
tscError("failed to truncate file:%s, errno:%d", path, terrno);
tscError("failed to truncate file:%s, terrno:%d", path, terrno);
return;
}
if (taosUnLockFile(pFile) != 0) {
tscError("failed to unlock file:%s, errno:%d", path, terrno);
tscError("failed to unlock file:%s, terrno:%d", path, terrno);
return;
}
if (taosCloseFile(&(pFile)) != 0) {
tscError("failed to close file:%s, errno:%d", path, errno);
tscError("failed to close file:%s, terrno:%d", path, terrno);
return;
}
if (taosRemoveFile(path) != 0) {
tscError("failed to remove file:%s, errno:%d", path, errno);
tscError("failed to remove file:%s, terrno:%d", path, terrno);
return;
}
}

11
source/client/src/clientStmt2.c
View File

@ -818,6 +818,7 @@ TAOS_STMT2* stmtInit2(STscObj* taos, TAOS_STMT2_OPTION* pOptions) {
if (pStmt->options.asyncExecFn) {
(void)tsem_init(&pStmt->asyncQuerySem, 0, 1);
}
pStmt->semWaited = false;
STMT_LOG_SEQ(STMT_INIT);
@ -1262,10 +1263,6 @@ int stmtBindBatch2(TAOS_STMT2* stmt, TAOS_STMT2_BIND* bind, int32_t colIdx) {
STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND));
if (pStmt->options.asyncExecFn) {
(void)tsem_wait(&pStmt->asyncQuerySem);
}
if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
pStmt->bInfo.needParse = false;
@ -1666,7 +1663,6 @@ int stmtExec2(TAOS_STMT2* stmt, int* affected_rows) {
STMT_ERR_RET(stmtCleanExecInfo(pStmt, (code ? false : true), false));
++pStmt->sql.runTimes;
} else {
SSqlCallbackWrapper* pWrapper = taosMemoryCalloc(1, sizeof(SSqlCallbackWrapper));
if (pWrapper == NULL) {
@ -1682,6 +1678,7 @@ int stmtExec2(TAOS_STMT2* stmt, int* affected_rows) {
pRequest->body.queryFp = asyncQueryCb;
((SSyncQueryParam*)(pRequest)->body.interParam)->userParam = pStmt;
pStmt->semWaited = false;
launchAsyncQuery(pRequest, pStmt->sql.pQuery, NULL, pWrapper);
}
@ -1703,6 +1700,10 @@ int stmtClose2(TAOS_STMT2* stmt) {
pStmt->bindThreadInUse = false;
}
if (pStmt->options.asyncExecFn && !pStmt->semWaited) {
(void)tsem_wait(&pStmt->asyncQuerySem);
}
STMT_DLOG("stmt %p closed, stbInterlaceMode: %d, statInfo: ctgGetTbMetaNum=>%" PRId64 ", getCacheTbInfo=>%" PRId64
", parseSqlNum=>%" PRId64 ", pStmt->stat.bindDataNum=>%" PRId64
", settbnameAPI:%u, bindAPI:%u, addbatchAPI:%u, execAPI:%u"

34
source/client/src/clientTmq.c
View File

@ -298,6 +298,7 @@ void tmq_conf_destroy(tmq_conf_t* conf) {
}
tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value) {
int32_t code = 0;
if (conf == NULL || key == NULL || value == NULL) {
return TMQ_CONF_INVALID;
}
@ -324,8 +325,9 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
}
if (strcasecmp(key, "auto.commit.interval.ms") == 0) {
int64_t tmp = taosStr2int64(value);
if (tmp < 0 || EINVAL == errno || ERANGE == errno) {
int64_t tmp;
code = taosStr2int64(value, &tmp);
if (tmp < 0 || code != 0) {
return TMQ_CONF_INVALID;
}
conf->autoCommitInterval = (tmp > INT32_MAX ? INT32_MAX : tmp);
@ -333,8 +335,9 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
}
if (strcasecmp(key, "session.timeout.ms") == 0) {
int64_t tmp = taosStr2int64(value);
if (tmp < 6000 || tmp > 1800000) {
int64_t tmp;
code = taosStr2int64(value, &tmp);
if (tmp < 6000 || tmp > 1800000 || code != 0) {
return TMQ_CONF_INVALID;
}
conf->sessionTimeoutMs = tmp;
@ -342,8 +345,9 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
}
if (strcasecmp(key, "heartbeat.interval.ms") == 0) {
int64_t tmp = taosStr2int64(value);
if (tmp < 1000 || tmp >= conf->sessionTimeoutMs) {
int64_t tmp;
code = taosStr2int64(value, &tmp);
if (tmp < 1000 || tmp >= conf->sessionTimeoutMs || code != 0) {
return TMQ_CONF_INVALID;
}
conf->heartBeatIntervalMs = tmp;
@ -351,8 +355,9 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
}
if (strcasecmp(key, "max.poll.interval.ms") == 0) {
int64_t tmp = taosStr2int64(value);
if (tmp < 1000 || tmp > INT32_MAX) {
int32_t tmp;
code = taosStr2int32(value, &tmp);
if (tmp < 1000 || code != 0) {
return TMQ_CONF_INVALID;
}
conf->maxPollIntervalMs = tmp;
@ -414,8 +419,9 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
}
if (strcasecmp(key, "td.connect.port") == 0) {
int64_t tmp = taosStr2int64(value);
if (tmp <= 0 || tmp > 65535) {
int64_t tmp;
code = taosStr2int64(value, &tmp);
if (tmp <= 0 || tmp > 65535 || code != 0) {
return TMQ_CONF_INVALID;
}
@ -435,7 +441,9 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
}
}
if (strcasecmp(key, "msg.consume.excluded") == 0) {
conf->sourceExcluded = (taosStr2int64(value) != 0) ? TD_REQ_FROM_TAOX : 0;
int64_t tmp;
code = taosStr2int64(value, &tmp);
conf->sourceExcluded = (0 == code && tmp != 0) ? TD_REQ_FROM_TAOX : 0;
return TMQ_CONF_OK;
}
@ -444,7 +452,9 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
}
if (strcasecmp(key, "msg.enable.batchmeta") == 0) {
conf->enableBatchMeta = (taosStr2int64(value) != 0) ? true : false;
int64_t tmp;
code = taosStr2int64(value, &tmp);
conf->enableBatchMeta = (0 == code && tmp != 0) ? true : false;
return TMQ_CONF_OK;
}

3
source/common/src/cos.c
View File

@ -1412,7 +1412,7 @@ static int32_t s3GetObjectToFileByEp(const char *object_name, const char *fileNa
TdFilePtr pFile = taosOpenFile(fileName, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pFile == NULL) {
uError("[s3] open file error, errno:%d, fileName:%s", terrno, fileName);
uError("[s3] open file error, terrno:%d, fileName:%s", terrno, fileName);
TAOS_RETURN(terrno);
}
@ -1871,7 +1871,6 @@ void s3EvictCache(const char *path, long object_size) {
taosDirName(dir_name);
if (taosGetDiskSize((char *)dir_name, &disk_size) < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
vError("failed to get disk:%s size since %s", path, terrstr());
return;
}

1
source/common/src/tmisce.c
View File

@ -197,7 +197,6 @@ int32_t taosGenCrashJsonMsg(int signum, char** pMsg, int64_t clusterId, int64_t
code = taosGetAppName(tmp, NULL);
if (code != 0) {
code = TAOS_SYSTEM_ERROR(errno);
TAOS_CHECK_GOTO(code, NULL, _exit);
}
TAOS_CHECK_GOTO(tjsonAddStringToObject(pJson, "appName", tmp), NULL, _exit);

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

@ -155,7 +155,11 @@ int32_t tNameGetDbName(const SName* name, char* dst) {
const char* tNameGetDbNameP(const SName* name) { return &name->dbname[0]; }
int32_t tNameGetFullDbName(const SName* name, char* dst) {
return snprintf(dst, TSDB_DB_FNAME_LEN, "%d.%s", name->acctId, name->dbname);
if (name == NULL || dst == NULL) {
return TSDB_CODE_INVALID_PARA;
}
(void)snprintf(dst, TSDB_DB_FNAME_LEN, "%d.%s", name->acctId, name->dbname);
return 0;
}
bool tNameIsEmpty(const SName* name) { return name->type == 0 || name->acctId == 0; }

6
source/dnode/mgmt/node_mgmt/src/dmEnv.c
View File

@ -89,9 +89,13 @@ static bool dmDataSpaceAvailable() {
}
static int32_t dmCheckDiskSpace() {
osUpdate();
// availability
int32_t code = 0;
code = osUpdate();
if(code != 0) {
code = 0; // ignore the error, just log it
dError("failed to update os info since %s", tstrerror(code));
}
if (!dmDataSpaceAvailable()) {
code = TSDB_CODE_NO_DISKSPACE;
return code;

4
source/dnode/mgmt/node_mgmt/src/dmNodes.c
View File

@ -159,7 +159,9 @@ int32_t dmRunDnode(SDnode *pDnode) {
}
if (count == 10) {
osUpdate();
if(osUpdate() != 0) {
dError("failed to update os info");
}
count = 0;
} else {
count++;

34
source/dnode/mgmt/node_util/src/dmUtil.c
View File

@ -55,14 +55,36 @@ void *dmSetMgmtHandle(SArray *pArray, tmsg_t msgType, void *nodeMsgFp, bool need
}
void dmGetMonitorSystemInfo(SMonSysInfo *pInfo) {
taosGetCpuUsage(&pInfo->cpu_system, &pInfo->cpu_engine);
taosGetCpuCores(&pInfo->cpu_cores, false);
(void)taosGetProcMemory(&pInfo->mem_engine);
(void)taosGetSysMemory(&pInfo->mem_system);
int32_t code = 0;
code = taosGetCpuUsage(&pInfo->cpu_system, &pInfo->cpu_engine);
if (code != 0) {
dError("failed to get cpu usage since %s", tstrerror(code));
}
code = taosGetCpuCores(&pInfo->cpu_cores, false);
if (code != 0) {
dError("failed to get cpu cores since %s", tstrerror(code));
}
code = taosGetProcMemory(&pInfo->mem_engine);
if (code != 0) {
dError("failed to get proc memory since %s", tstrerror(code));
}
code = taosGetSysMemory(&pInfo->mem_system);
if (code != 0) {
dError("failed to get sys memory since %s", tstrerror(code));
}
pInfo->mem_total = tsTotalMemoryKB;
pInfo->disk_engine = 0;
pInfo->disk_used = tsDataSpace.size.used;
pInfo->disk_total = tsDataSpace.size.total;
taosGetCardInfoDelta(&pInfo->net_in, &pInfo->net_out);
taosGetProcIODelta(&pInfo->io_read, &pInfo->io_write, &pInfo->io_read_disk, &pInfo->io_write_disk);
code = taosGetCardInfoDelta(&pInfo->net_in, &pInfo->net_out);
if (code != 0) {
dError("failed to get card info since %s", tstrerror(code));
taosSetDefaultCardInfoDelta(&pInfo->net_in, &pInfo->net_out);
}
code = taosGetProcIODelta(&pInfo->io_read, &pInfo->io_write, &pInfo->io_read_disk, &pInfo->io_write_disk);
if (code != 0) {
dError("failed to get proc io delta since %s", tstrerror(code));
taosSetDefaultProcIODelta(&pInfo->io_read, &pInfo->io_write, &pInfo->io_read_disk, &pInfo->io_write_disk);
}
return;
}

13
source/dnode/mnode/impl/inc/mndShow.h
View File

@ -32,6 +32,19 @@ extern "C" {
} \
} while (0)
#define RETRIEVE_CHECK_GOTO(CMD, pObj, LINO, LABEL) \
do { \
code = (CMD); \
if (code != TSDB_CODE_SUCCESS) { \
if (LINO) { \
*((int32_t *)(LINO)) = __LINE__; \
} \
if (pObj) sdbRelease(pSdb, (pObj)); \
if (pObj) sdbCancelFetch(pSdb, (pObj)); \
goto LABEL; \
} \
} while (0)
int32_t mndInitShow(SMnode *pMnode);
void mndCleanupShow(SMnode *pMnode);
void mndAddShowRetrieveHandle(SMnode *pMnode, EShowType showType, ShowRetrieveFp fp);

6
source/dnode/mnode/impl/inc/mndStb.h
View File

@ -37,9 +37,9 @@ int32_t mndAddStbToTrans(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *p
void mndFreeStb(SStbObj *pStb);
int32_t mndBuildSMCreateStbRsp(SMnode *pMnode, char *dbFName, char *stbFName, void **pCont, int32_t *pLen);
void mndExtractDbNameFromStbFullName(const char *stbFullName, char *dst);
void mndExtractShortDbNameFromStbFullName(const char *stbFullName, char *dst);
void mndExtractShortDbNameFromDbFullName(const char *stbFullName, char *dst);
int32_t mndExtractDbNameFromStbFullName(const char *stbFullName, char *dst);
int32_t mndExtractShortDbNameFromStbFullName(const char *stbFullName, char *dst);
int32_t mndExtractShortDbNameFromDbFullName(const char *stbFullName, char *dst);
void mndExtractTbNameFromStbFullName(const char *stbFullName, char *dst, int32_t dstSize);
const char *mndGetStbStr(const char *src);

4
source/dnode/mnode/impl/inc/mndVgroup.h
View File

@ -55,8 +55,8 @@ void *mndBuildCreateVnodeReq(SMnode *, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *p
void *mndBuildDropVnodeReq(SMnode *, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen);
bool mndVgroupInDb(SVgObj *pVgroup, int64_t dbUid);
bool mndVgroupInDnode(SVgObj *pVgroup, int32_t dnodeId);
int32_t mndBuildRestoreAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *db, SVgObj *pVgroup,
SDnodeObj *pDnode);
int32_t mndBuildRestoreAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *db, SVgObj *pVgroup, SDnodeObj *pDnode,
SDnodeObj *pAnotherDnode);
int32_t mndSplitVgroup(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SVgObj *pVgroup);

41
source/dnode/mnode/impl/src/mndCompact.c
View File

@ -268,7 +268,11 @@ int32_t mndAddCompactToTran(SMnode *pMnode, STrans *pTrans, SCompactObj *pCompac
sdbFreeRaw(pVgRaw);
TAOS_RETURN(code);
}
(void)sdbSetRawStatus(pVgRaw, SDB_STATUS_READY);
if ((code = sdbSetRawStatus(pVgRaw, SDB_STATUS_READY)) != 0) {
sdbFreeRaw(pVgRaw);
TAOS_RETURN(code);
}
rsp->compactId = pCompact->compactId;
@ -357,7 +361,10 @@ static void *mndBuildKillCompactReq(SMnode *pMnode, SVgObj *pVgroup, int32_t *pC
pHead->contLen = htonl(contLen);
pHead->vgId = htonl(pVgroup->vgId);
(void)tSerializeSVKillCompactReq((char *)pReq + sizeof(SMsgHead), contLen, &req);
if ((contLen = tSerializeSVKillCompactReq((char *)pReq + sizeof(SMsgHead), contLen, &req)) < 0) {
terrno = contLen;
return NULL;
}
*pContLen = contLen;
return pReq;
}
@ -421,7 +428,10 @@ static int32_t mndKillCompact(SMnode *pMnode, SRpcMsg *pReq, SCompactObj *pCompa
mndTransDrop(pTrans);
TAOS_RETURN(code);
}
(void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
if ((code = sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY)) != 0) {
mndTransDrop(pTrans);
TAOS_RETURN(code);
}
void *pIter = NULL;
while (1) {
@ -614,7 +624,10 @@ void mndCompactSendProgressReq(SMnode *pMnode, SCompactObj *pCompact) {
pHead->contLen = htonl(contLen);
pHead->vgId = htonl(pDetail->vgId);
(void)tSerializeSQueryCompactProgressReq((char *)pHead + sizeof(SMsgHead), contLen - sizeof(SMsgHead), &req);
if (tSerializeSQueryCompactProgressReq((char *)pHead + sizeof(SMsgHead), contLen - sizeof(SMsgHead), &req) <= 0) {
sdbRelease(pMnode->pSdb, pDetail);
continue;
}
SRpcMsg rpcMsg = {.msgType = TDMT_VND_QUERY_COMPACT_PROGRESS, .contLen = contLen};
@ -719,7 +732,12 @@ static int32_t mndSaveCompactProgress(SMnode *pMnode, int32_t compactId) {
mndTransDrop(pTrans);
TAOS_RETURN(code);
}
(void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
if ((code = sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY)) != 0) {
sdbCancelFetch(pMnode->pSdb, pIter);
sdbRelease(pMnode->pSdb, pDetail);
mndTransDrop(pTrans);
TAOS_RETURN(code);
}
}
sdbRelease(pMnode->pSdb, pDetail);
@ -782,7 +800,12 @@ static int32_t mndSaveCompactProgress(SMnode *pMnode, int32_t compactId) {
mndTransDrop(pTrans);
TAOS_RETURN(code);
}
(void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
if ((code = sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED)) != 0) {
sdbCancelFetch(pMnode->pSdb, pIter);
sdbRelease(pMnode->pSdb, pDetail);
mndTransDrop(pTrans);
TAOS_RETURN(code);
}
mInfo("compact:%d, add drop compactdetail action", pDetail->compactDetailId);
}
@ -809,7 +832,11 @@ static int32_t mndSaveCompactProgress(SMnode *pMnode, int32_t compactId) {
mndTransDrop(pTrans);
TAOS_RETURN(code);
}
(void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
if ((code = sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED)) != 0) {
mError("compact:%d, trans:%d, failed to append commit log since %s", compactId, pTrans->id, terrstr());
mndTransDrop(pTrans);
TAOS_RETURN(code);
}
mInfo("compact:%d, add drop compact action", pCompact->compactId);
}

45
source/dnode/mnode/impl/src/mndDnode.c
View File

@ -182,7 +182,7 @@ static int32_t mndCreateDefaultDnode(SMnode *pMnode) {
goto _OVER;
}
TAOS_CHECK_GOTO(mndTransAppendCommitlog(pTrans, pRaw), NULL, _OVER);
(void)sdbSetRawStatus(pRaw, SDB_STATUS_READY);
TAOS_CHECK_GOTO(sdbSetRawStatus(pRaw, SDB_STATUS_READY), NULL, _OVER);
pRaw = NULL;
TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
@ -621,7 +621,10 @@ static int32_t mndUpdateDnodeObj(SMnode *pMnode, SDnodeObj *pDnode) {
TAOS_RETURN(terrno);
}
(void)tSerializeSDnodeInfoReq(pReq, contLen, &infoReq);
if ((contLen = tSerializeSDnodeInfoReq(pReq, contLen, &infoReq)) <= 0) {
code = contLen;
goto _exit;
}
SRpcMsg rpcMsg = {.msgType = TDMT_MND_UPDATE_DNODE_INFO, .pCont = pReq, .contLen = contLen};
TAOS_CHECK_EXIT(tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg));
@ -661,7 +664,7 @@ static int32_t mndProcessUpdateDnodeInfoReq(SRpcMsg *pReq) {
mError("trans:%d, failed to append commit log since %s", pTrans->id, tstrerror(code));
TAOS_CHECK_EXIT(code);
}
(void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
TAOS_CHECK_EXIT(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));
pCommitRaw = NULL;
if ((code = mndTransPrepare(pMnode, pTrans)) != 0) {
@ -874,8 +877,12 @@ static int32_t mndProcessStatusReq(SRpcMsg *pReq) {
int32_t contLen = tSerializeSStatusRsp(NULL, 0, &statusRsp);
void *pHead = rpcMallocCont(contLen);
(void)tSerializeSStatusRsp(pHead, contLen, &statusRsp);
contLen = tSerializeSStatusRsp(pHead, contLen, &statusRsp);
taosArrayDestroy(statusRsp.pDnodeEps);
if (contLen < 0) {
code = contLen;
goto _OVER;
}
pReq->info.rspLen = contLen;
pReq->info.rsp = pHead;
@ -888,8 +895,7 @@ static int32_t mndProcessStatusReq(SRpcMsg *pReq) {
_OVER:
mndReleaseDnode(pMnode, pDnode);
taosArrayDestroy(statusReq.pVloads);
(void)mndUpdClusterInfo(pReq);
return code;
return mndUpdClusterInfo(pReq);
}
static int32_t mndProcessNotifyReq(SRpcMsg *pReq) {
@ -955,7 +961,7 @@ static int32_t mndCreateDnode(SMnode *pMnode, SRpcMsg *pReq, SCreateDnodeReq *pC
goto _OVER;
}
TAOS_CHECK_GOTO(mndTransAppendCommitlog(pTrans, pRaw), NULL, _OVER);
(void)sdbSetRawStatus(pRaw, SDB_STATUS_READY);
TAOS_CHECK_GOTO(sdbSetRawStatus(pRaw, SDB_STATUS_READY), NULL, _OVER);
pRaw = NULL;
TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
@ -1010,7 +1016,10 @@ static int32_t mndProcessDnodeListReq(SRpcMsg *pReq) {
goto _OVER;
}
(void)tSerializeSDnodeListRsp(pRsp, rspLen, &rsp);
if ((rspLen = tSerializeSDnodeListRsp(pRsp, rspLen, &rsp)) <= 0) {
code = rspLen;
goto _OVER;
}
pReq->info.rspLen = rspLen;
pReq->info.rsp = pRsp;
@ -1151,7 +1160,10 @@ static int32_t mndProcessShowVariablesReq(SRpcMsg *pReq) {
goto _OVER;
}
(void)tSerializeSShowVariablesRsp(pRsp, rspLen, &rsp);
if ((rspLen = tSerializeSShowVariablesRsp(pRsp, rspLen, &rsp)) <= 0) {
code = rspLen;
goto _OVER;
}
pReq->info.rspLen = rspLen;
pReq->info.rsp = pRsp;
@ -1247,7 +1259,7 @@ static int32_t mndDropDnode(SMnode *pMnode, SRpcMsg *pReq, SDnodeObj *pDnode, SM
goto _OVER;
}
TAOS_CHECK_GOTO(mndTransAppendRedolog(pTrans, pRaw), NULL, _OVER);
(void)sdbSetRawStatus(pRaw, SDB_STATUS_DROPPING);
TAOS_CHECK_GOTO(sdbSetRawStatus(pRaw, SDB_STATUS_DROPPING), NULL, _OVER);
pRaw = NULL;
pRaw = mndDnodeActionEncode(pDnode);
@ -1257,7 +1269,7 @@ static int32_t mndDropDnode(SMnode *pMnode, SRpcMsg *pReq, SDnodeObj *pDnode, SM
goto _OVER;
}
TAOS_CHECK_GOTO(mndTransAppendCommitlog(pTrans, pRaw), NULL, _OVER);
(void)sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED);
TAOS_CHECK_GOTO(sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED), NULL, _OVER);
pRaw = NULL;
if (pMObj != NULL) {
@ -1450,7 +1462,10 @@ static int32_t mndSendCfgDnodeReq(SMnode *pMnode, int32_t dnodeId, SDCfgDnodeReq
void *pBuf = rpcMallocCont(bufLen);
if (pBuf != NULL) {
(void)tSerializeSDCfgDnodeReq(pBuf, bufLen, pDcfgReq);
if ((bufLen = tSerializeSDCfgDnodeReq(pBuf, bufLen, pDcfgReq)) <= 0) {
code = bufLen;
return code;
}
mInfo("dnode:%d, send config req to dnode, config:%s value:%s", dnodeId, pDcfgReq->config, pDcfgReq->value);
SRpcMsg rpcMsg = {.msgType = TDMT_DND_CONFIG_DNODE, .pCont = pBuf, .contLen = bufLen};
code = tmsgSendReq(&epSet, &rpcMsg);
@ -1590,7 +1605,11 @@ static int32_t mndProcessCreateEncryptKeyReqImpl(SRpcMsg *pReq, int32_t dnodeId,
void *pBuf = rpcMallocCont(bufLen);
if (pBuf != NULL) {
(void)tSerializeSDCfgDnodeReq(pBuf, bufLen, pDcfgReq);
if ((bufLen = tSerializeSDCfgDnodeReq(pBuf, bufLen, pDcfgReq)) <= 0) {
code = bufLen;
sdbRelease(pSdb, pDnode);
goto _exit;
}
SRpcMsg rpcMsg = {.msgType = TDMT_DND_CREATE_ENCRYPT_KEY, .pCont = pBuf, .contLen = bufLen};
if (0 == tmsgSendReq(&epSet, &rpcMsg)) {
(void)atomic_add_fetch_16(&pMnode->encryptMgmt.nEncrypt, 1);

11
source/dnode/mnode/impl/src/mndFunc.c
View File

@ -375,7 +375,7 @@ static int32_t mndDropFunc(SMnode *pMnode, SRpcMsg *pReq, SFuncObj *pFunc) {
goto _OVER;
}
TAOS_CHECK_GOTO(mndTransAppendRedolog(pTrans, pRedoRaw), NULL, _OVER);
(void)sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPING);
TAOS_CHECK_GOTO(sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPING), NULL, _OVER);
SSdbRaw *pUndoRaw = mndFuncActionEncode(pFunc);
if (pUndoRaw == NULL) {
@ -384,7 +384,7 @@ static int32_t mndDropFunc(SMnode *pMnode, SRpcMsg *pReq, SFuncObj *pFunc) {
goto _OVER;
}
TAOS_CHECK_GOTO(mndTransAppendUndolog(pTrans, pUndoRaw), NULL, _OVER);
(void)sdbSetRawStatus(pUndoRaw, SDB_STATUS_READY);
TAOS_CHECK_GOTO(sdbSetRawStatus(pUndoRaw, SDB_STATUS_READY), NULL, _OVER);
SSdbRaw *pCommitRaw = mndFuncActionEncode(pFunc);
if (pCommitRaw == NULL) {
@ -393,7 +393,7 @@ static int32_t mndDropFunc(SMnode *pMnode, SRpcMsg *pReq, SFuncObj *pFunc) {
goto _OVER;
}
TAOS_CHECK_GOTO(mndTransAppendCommitlog(pTrans, pCommitRaw), NULL, _OVER);
(void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
TAOS_CHECK_GOTO(sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED), NULL, _OVER);
TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
@ -598,7 +598,10 @@ static int32_t mndProcessRetrieveFuncReq(SRpcMsg *pReq) {
goto RETRIEVE_FUNC_OVER;
}
(void)tSerializeSRetrieveFuncRsp(pRsp, contLen, &retrieveRsp);
if ((contLen = tSerializeSRetrieveFuncRsp(pRsp, contLen, &retrieveRsp)) <= 0) {
code = contLen;
goto RETRIEVE_FUNC_OVER;
}
pReq->info.rsp = pRsp;
pReq->info.rspLen = contLen;

11
source/dnode/mnode/impl/src/mndProfile.c
View File

@ -65,6 +65,8 @@ typedef struct {
int64_t ipWhiteListVer;
} SConnPreparedObj;
#define CACHE_OBJ_KEEP_TIME 3 // s
static SConnObj *mndCreateConn(SMnode *pMnode, const char *user, int8_t connType, uint32_t ip, uint16_t port,
int32_t pid, const char *app, int64_t startTime);
static void mndFreeConn(SConnObj *pConn);
@ -161,9 +163,8 @@ static SConnObj *mndCreateConn(SMnode *pMnode, const char *user, int8_t connType
tstrncpy(connObj.user, user, TSDB_USER_LEN);
tstrncpy(connObj.app, app, TSDB_APP_NAME_LEN);
int32_t keepTime = tsShellActivityTimer * 3;
SConnObj *pConn =
taosCachePut(pMgmt->connCache, &connId, sizeof(uint32_t), &connObj, sizeof(connObj), keepTime * 1000);
taosCachePut(pMgmt->connCache, &connId, sizeof(uint32_t), &connObj, sizeof(connObj), CACHE_OBJ_KEEP_TIME * 1000);
if (pConn == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
mError("conn:%d, failed to put into cache since %s, user:%s", connId, user, terrstr());
@ -376,8 +377,7 @@ static SAppObj *mndCreateApp(SMnode *pMnode, uint32_t clientIp, SAppHbReq *pReq)
(void)memcpy(&app.summary, &pReq->summary, sizeof(pReq->summary));
app.lastAccessTimeMs = taosGetTimestampMs();
const int32_t keepTime = tsShellActivityTimer * 3;
SAppObj *pApp = taosCachePut(pMgmt->appCache, &pReq->appId, sizeof(pReq->appId), &app, sizeof(app), keepTime * 1000);
SAppObj *pApp = taosCachePut(pMgmt->appCache, &pReq->appId, sizeof(pReq->appId), &app, sizeof(app), CACHE_OBJ_KEEP_TIME * 1000);
if (pApp == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
mError("failed to app %" PRIx64 " into cache since %s", pReq->appId, terrstr());
@ -842,7 +842,6 @@ static int32_t mndRetrieveConns(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
int32_t numOfRows = 0;
int32_t cols = 0;
SConnObj *pConn = NULL;
int32_t keepTime = tsShellActivityTimer * 3;
if (pShow->pIter == NULL) {
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
@ -856,7 +855,7 @@ static int32_t mndRetrieveConns(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
break;
}
if ((taosGetTimestampMs() - pConn->lastAccessTimeMs) > ((int64_t)keepTime * 1000)) {
if ((taosGetTimestampMs() - pConn->lastAccessTimeMs) > ((int64_t)CACHE_OBJ_KEEP_TIME * 1000)) {
continue;
}

184
source/dnode/mnode/impl/src/mndStb.c
View File

@ -482,7 +482,7 @@ SDbObj *mndAcquireDbByStb(SMnode *pMnode, const char *stbName) {
if ((terrno = tNameFromString(&name, stbName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0) return NULL;
char db[TSDB_TABLE_FNAME_LEN] = {0};
(void)tNameGetFullDbName(&name, db);
if ((terrno = tNameGetFullDbName(&name, db)) != 0) return NULL;
return mndAcquireDb(pMnode, db);
}
@ -507,7 +507,9 @@ void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb, int3
goto _err;
}
char dbFName[TSDB_DB_FNAME_LEN] = {0};
(void)tNameGetFullDbName(&name, dbFName);
if ((terrno = tNameGetFullDbName(&name, dbFName)) != 0) {
goto _err;
};
req.name = (char *)tNameGetTableName(&name);
req.suid = pStb->uid;
@ -625,8 +627,9 @@ static void *mndBuildVDropStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb,
void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));
tEncoderInit(&encoder, pBuf, contLen - sizeof(SMsgHead));
(void)tEncodeSVDropStbReq(&encoder, &req);
terrno = tEncodeSVDropStbReq(&encoder, &req);
tEncoderClear(&encoder);
if (terrno != 0) return NULL;
*pContLen = contLen;
return pHead;
@ -1003,7 +1006,7 @@ static int32_t mndCreateStb(SMnode *pMnode, SRpcMsg *pReq, SMCreateStbReq *pCrea
_OVER:
mndTransDrop(pTrans);
(void)mndStbActionDelete(pMnode->pSdb, &stbObj);
if (mndStbActionDelete(pMnode->pSdb, &stbObj) != 0) mError("failed to mndStbActionDelete");
TAOS_RETURN(code);
}
@ -1032,6 +1035,7 @@ static int32_t mndProcessTtlTimer(SRpcMsg *pReq) {
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
if (pIter == NULL) break;
int32_t code = 0;
SMsgHead *pHead = rpcMallocCont(contLen);
if (pHead == NULL) {
sdbRelease(pSdb, pVgroup);
@ -1039,12 +1043,16 @@ static int32_t mndProcessTtlTimer(SRpcMsg *pReq) {
}
pHead->contLen = htonl(contLen);
pHead->vgId = htonl(pVgroup->vgId);
(void)tSerializeSVDropTtlTableReq((char *)pHead + sizeof(SMsgHead), reqLen, &ttlReq);
if ((code = tSerializeSVDropTtlTableReq((char *)pHead + sizeof(SMsgHead), reqLen, &ttlReq)) < 0) {
mError("vgId:%d, failed to serialize drop ttl table request since %s", pVgroup->vgId, tstrerror(code));
sdbRelease(pSdb, pVgroup);
continue;
}
SRpcMsg rpcMsg = {
.msgType = TDMT_VND_FETCH_TTL_EXPIRED_TBS, .pCont = pHead, .contLen = contLen, .info = pReq->info};
SEpSet epSet = mndGetVgroupEpset(pMnode, pVgroup);
int32_t code = tmsgSendReq(&epSet, &rpcMsg);
code = tmsgSendReq(&epSet, &rpcMsg);
if (code != 0) {
mError("vgId:%d, failed to send drop ttl table request to vnode since 0x%x", pVgroup->vgId, code);
} else {
@ -1069,6 +1077,8 @@ static int32_t mndProcessTrimDbTimer(SRpcMsg *pReq) {
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
if (pIter == NULL) break;
int32_t code = 0;
SMsgHead *pHead = rpcMallocCont(contLen);
if (pHead == NULL) {
sdbCancelFetch(pSdb, pVgroup);
@ -1077,11 +1087,13 @@ static int32_t mndProcessTrimDbTimer(SRpcMsg *pReq) {
}
pHead->contLen = htonl(contLen);
pHead->vgId = htonl(pVgroup->vgId);
(void)tSerializeSVTrimDbReq((char *)pHead + sizeof(SMsgHead), reqLen, &trimReq);
if ((code = tSerializeSVTrimDbReq((char *)pHead + sizeof(SMsgHead), reqLen, &trimReq)) < 0) {
mError("vgId:%d, failed to serialize trim db request since %s", pVgroup->vgId, tstrerror(code));
}
SRpcMsg rpcMsg = {.msgType = TDMT_VND_TRIM, .pCont = pHead, .contLen = contLen};
SEpSet epSet = mndGetVgroupEpset(pMnode, pVgroup);
int32_t code = tmsgSendReq(&epSet, &rpcMsg);
code = tmsgSendReq(&epSet, &rpcMsg);
if (code != 0) {
mError("vgId:%d, timer failed to send vnode-trim request to vnode since 0x%x", pVgroup->vgId, code);
} else {
@ -1106,19 +1118,24 @@ static int32_t mndProcessS3MigrateDbTimer(SRpcMsg *pReq) {
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
if (pIter == NULL) break;
int32_t code = 0;
SMsgHead *pHead = rpcMallocCont(contLen);
if (pHead == NULL) {
sdbCancelFetch(pSdb, pVgroup);
sdbRelease(pSdb, pVgroup);
continue;
}
pHead->contLen = htonl(contLen);
pHead->vgId = htonl(pVgroup->vgId);
(void)tSerializeSVS3MigrateDbReq((char *)pHead + sizeof(SMsgHead), reqLen, &s3migrateReq);
if ((code = tSerializeSVS3MigrateDbReq((char *)pHead + sizeof(SMsgHead), reqLen, &s3migrateReq)) < 0) {
mError("vgId:%d, failed to serialize s3migrate db request since %s", pVgroup->vgId, tstrerror(code));
sdbRelease(pSdb, pVgroup);
continue;
}
SRpcMsg rpcMsg = {.msgType = TDMT_VND_S3MIGRATE, .pCont = pHead, .contLen = contLen};
SEpSet epSet = mndGetVgroupEpset(pMnode, pVgroup);
int32_t code = tmsgSendReq(&epSet, &rpcMsg);
code = tmsgSendReq(&epSet, &rpcMsg);
if (code != 0) {
mError("vgId:%d, timer failed to send vnode-s3migrate request to vnode since 0x%x", pVgroup->vgId, code);
} else {
@ -2336,11 +2353,13 @@ static int32_t mndBuildSMAlterStbRsp(SDbObj *pDb, SStbObj *pObj, void **pCont, i
void *cont = taosMemoryMalloc(contLen);
tEncoderInit(&ec, cont, contLen);
(void)tEncodeSMAlterStbRsp(&ec, &alterRsp);
code = tEncodeSMAlterStbRsp(&ec, &alterRsp);
tEncoderClear(&ec);
tFreeSMAlterStbRsp(&alterRsp);
if (code < 0) TAOS_RETURN(code);
*pCont = cont;
*pLen = contLen;
@ -2389,7 +2408,7 @@ int32_t mndBuildSMCreateStbRsp(SMnode *pMnode, char *dbFName, char *stbFName, vo
void *cont = taosMemoryMalloc(contLen);
tEncoderInit(&ec, cont, contLen);
(void)tEncodeSMCreateStbRsp(&ec, &stbRsp);
TAOS_CHECK_GOTO(tEncodeSMCreateStbRsp(&ec, &stbRsp), NULL, _OVER);
tEncoderClear(&ec);
tFreeSMCreateStbRsp(&stbRsp);
@ -2628,8 +2647,9 @@ static int32_t mndProcessAlterStbReq(SRpcMsg *pReq) {
if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
SName name = {0};
// TODO check return value
(void)tNameFromString(&name, alterReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
int32_t ret = 0;
if ((ret = tNameFromString(&name, alterReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0)
mError("stb:%s, failed to tNameFromString since %s", alterReq.name, tstrerror(ret));
auditRecord(pReq, pMnode->clusterId, "alterStb", name.dbname, name.tname, alterReq.sql, alterReq.sqlLen);
@ -2916,8 +2936,9 @@ static int32_t mndProcessDropStbReq(SRpcMsg *pReq) {
if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
SName name = {0};
// TODO check return value
(void)tNameFromString(&name, dropReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
int32_t ret = 0;
if ((ret = tNameFromString(&name, dropReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0)
mError("stb:%s, failed to tNameFromString since %s", dropReq.name, tstrerror(ret));
auditRecord(pReq, pMnode->clusterId, "dropStb", name.dbname, name.tname, dropReq.sql, dropReq.sqlLen);
@ -2968,7 +2989,10 @@ static int32_t mndProcessTableMetaReq(SRpcMsg *pReq) {
goto _OVER;
}
(void)tSerializeSTableMetaRsp(pRsp, rspLen, &metaRsp);
if ((rspLen = tSerializeSTableMetaRsp(pRsp, rspLen, &metaRsp)) < 0) {
code = rspLen;
goto _OVER;
}
pReq->info.rsp = pRsp;
pReq->info.rspLen = rspLen;
code = 0;
@ -2995,7 +3019,7 @@ static int32_t mndProcessTableCfgReq(SRpcMsg *pReq) {
TAOS_CHECK_GOTO(tDeserializeSTableCfgReq(pReq->pCont, pReq->contLen, &cfgReq), NULL, _OVER);
char dbName[TSDB_DB_NAME_LEN] = {0};
mndExtractShortDbNameFromDbFullName(cfgReq.dbFName, dbName);
TAOS_CHECK_GOTO(mndExtractShortDbNameFromDbFullName(cfgReq.dbFName, dbName), NULL, _OVER);
if (0 == strcmp(dbName, TSDB_INFORMATION_SCHEMA_DB)) {
mInfo("information_schema table:%s.%s, start to retrieve cfg", cfgReq.dbFName, cfgReq.tbName);
TAOS_CHECK_GOTO(mndBuildInsTableCfg(pMnode, cfgReq.dbFName, cfgReq.tbName, &cfgRsp), NULL, _OVER);
@ -3019,7 +3043,10 @@ static int32_t mndProcessTableCfgReq(SRpcMsg *pReq) {
goto _OVER;
}
(void)tSerializeSTableCfgRsp(pRsp, rspLen, &cfgRsp);
if ((rspLen = tSerializeSTableCfgRsp(pRsp, rspLen, &cfgRsp)) < 0) {
code = rspLen;
goto _OVER;
}
pReq->info.rsp = pRsp;
pReq->info.rspLen = rspLen;
code = 0;
@ -3140,8 +3167,9 @@ int32_t mndValidateStbInfo(SMnode *pMnode, SSTableVersion *pStbVersions, int32_t
TAOS_RETURN(code);
}
(void)tSerializeSSTbHbRsp(pRsp, rspLen, &hbRsp);
rspLen = tSerializeSSTbHbRsp(pRsp, rspLen, &hbRsp);
tFreeSSTbHbRsp(&hbRsp);
if (rspLen < 0) return rspLen;
*ppRsp = pRsp;
*pRspLen = rspLen;
TAOS_RETURN(code);
@ -3175,25 +3203,31 @@ int32_t mndGetNumOfStbs(SMnode *pMnode, char *dbName, int32_t *pNumOfStbs) {
TAOS_RETURN(code);
}
void mndExtractDbNameFromStbFullName(const char *stbFullName, char *dst) {
int32_t mndExtractDbNameFromStbFullName(const char *stbFullName, char *dst) {
SName name = {0};
(void)tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
TAOS_CHECK_RETURN(tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE));
(void)tNameGetFullDbName(&name, dst);
TAOS_CHECK_RETURN(tNameGetFullDbName(&name, dst));
return 0;
}
void mndExtractShortDbNameFromStbFullName(const char *stbFullName, char *dst) {
int32_t mndExtractShortDbNameFromStbFullName(const char *stbFullName, char *dst) {
SName name = {0};
(void)tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
TAOS_CHECK_RETURN(tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE));
(void)tNameGetDbName(&name, dst);
TAOS_CHECK_RETURN(tNameGetDbName(&name, dst));
return 0;
}
void mndExtractShortDbNameFromDbFullName(const char *stbFullName, char *dst) {
int32_t mndExtractShortDbNameFromDbFullName(const char *stbFullName, char *dst) {
SName name = {0};
(void)tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB);
TAOS_CHECK_RETURN(tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB));
(void)tNameGetDbName(&name, dst);
TAOS_CHECK_RETURN(tNameGetDbName(&name, dst));
return 0;
}
void mndExtractTbNameFromStbFullName(const char *stbFullName, char *dst, int32_t dstSize) {
@ -3491,6 +3525,8 @@ static int32_t mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBloc
int32_t numOfRows = 0;
SStbObj *pStb = NULL;
int32_t cols = 0;
int32_t lino = 0;
int32_t code = 0;
SDbObj *pDb = NULL;
if (strlen(pShow->db) > 0) {
@ -3515,42 +3551,44 @@ static int32_t mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBloc
cols = 0;
SName name = {0};
char stbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
mndExtractTbNameFromStbFullName(pStb->name, &stbName[VARSTR_HEADER_SIZE], TSDB_TABLE_NAME_LEN);
varDataSetLen(stbName, strlen(&stbName[VARSTR_HEADER_SIZE]));
SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)stbName, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stbName, false), pStb, &lino, _ERROR);
char db[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
(void)tNameFromString(&name, pStb->db, T_NAME_ACCT | T_NAME_DB);
(void)tNameGetDbName(&name, varDataVal(db));
RETRIEVE_CHECK_GOTO(tNameFromString(&name, pStb->db, T_NAME_ACCT | T_NAME_DB), pStb, &lino, _ERROR);
RETRIEVE_CHECK_GOTO(tNameGetDbName(&name, varDataVal(db)), pStb, &lino, _ERROR);
varDataSetLen(db, strlen(varDataVal(db)));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)db, false), pStb, &lino, _ERROR);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)db, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->createdTime, false), pStb, &lino,
_ERROR);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->createdTime, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->numOfColumns, false), pStb, &lino,
_ERROR);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->numOfColumns, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->numOfTags, false), pStb, &lino, _ERROR);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->numOfTags, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->updateTime, false); // number of tables
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->updateTime, false), pStb, &lino,
_ERROR); // number of tables
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
if (pStb->commentLen > 0) {
char comment[TSDB_TB_COMMENT_LEN + VARSTR_HEADER_SIZE] = {0};
STR_TO_VARSTR(comment, pStb->comment);
(void)colDataSetVal(pColInfo, numOfRows, comment, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, comment, false), pStb, &lino, _ERROR);
} else if (pStb->commentLen == 0) {
char comment[VARSTR_HEADER_SIZE + VARSTR_HEADER_SIZE] = {0};
STR_TO_VARSTR(comment, "");
(void)colDataSetVal(pColInfo, numOfRows, comment, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, comment, false), pStb, &lino, _ERROR);
} else {
colDataSetNULL(pColInfo, numOfRows);
}
@ -3560,14 +3598,14 @@ static int32_t mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBloc
varDataSetLen(watermark, strlen(varDataVal(watermark)));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)watermark, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)watermark, false), pStb, &lino, _ERROR);
char maxDelay[64 + VARSTR_HEADER_SIZE] = {0};
sprintf(varDataVal(maxDelay), "%" PRId64 "a,%" PRId64 "a", pStb->maxdelay[0], pStb->maxdelay[1]);
varDataSetLen(maxDelay, strlen(varDataVal(maxDelay)));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)maxDelay, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)maxDelay, false), pStb, &lino, _ERROR);
char rollup[160 + VARSTR_HEADER_SIZE] = {0};
int32_t rollupNum = (int32_t)taosArrayGetSize(pStb->pFuncs);
@ -3586,7 +3624,7 @@ static int32_t mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBloc
varDataSetLen(rollup, strlen(varDataVal(rollup)));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)rollup, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)rollup, false), pStb, &lino, _ERROR);
numOfRows++;
sdbRelease(pSdb, pStb);
@ -3596,6 +3634,13 @@ static int32_t mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBloc
mndReleaseDb(pMnode, pDb);
}
goto _OVER;
_ERROR:
mError("show:0x%" PRIx64 ", failed to retrieve data at %s:%d since %s", pShow->id, __FUNCTION__, lino,
tstrerror(code));
_OVER:
pShow->numOfRows += numOfRows;
return numOfRows;
}
@ -3606,6 +3651,8 @@ static int32_t buildDbColsInfoBlock(const SSDataBlock *p, const SSysTableMeta *p
char dName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
char typeName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
int32_t numOfRows = p->info.rows;
int32_t lino = 0;
int32_t code = 0;
STR_TO_VARSTR(dName, dbName);
STR_TO_VARSTR(typeName, "SYSTEM_TABLE");
@ -3624,20 +3671,20 @@ static int32_t buildDbColsInfoBlock(const SSDataBlock *p, const SSysTableMeta *p
for (int32_t j = 0; j < pm->colNum; j++) {
// table name
SColumnInfoData *pColInfoData = taosArrayGet(p->pDataBlock, 0);
(void)colDataSetVal(pColInfoData, numOfRows, tName, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, tName, false), &lino, _OVER);
// database name
pColInfoData = taosArrayGet(p->pDataBlock, 1);
(void)colDataSetVal(pColInfoData, numOfRows, dName, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, dName, false), &lino, _OVER);
pColInfoData = taosArrayGet(p->pDataBlock, 2);
(void)colDataSetVal(pColInfoData, numOfRows, typeName, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, typeName, false), &lino, _OVER);
// col name
char colName[TSDB_COL_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
STR_TO_VARSTR(colName, pm->schema[j].name);
pColInfoData = taosArrayGet(p->pDataBlock, 3);
(void)colDataSetVal(pColInfoData, numOfRows, colName, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, colName, false), &lino, _OVER);
// col type
int8_t colType = pm->schema[j].type;
@ -3652,10 +3699,10 @@ static int32_t buildDbColsInfoBlock(const SSDataBlock *p, const SSysTableMeta *p
(int32_t)((pm->schema[j].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));
}
varDataSetLen(colTypeStr, colTypeLen);
(void)colDataSetVal(pColInfoData, numOfRows, (char *)colTypeStr, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, (char *)colTypeStr, false), &lino, _OVER);
pColInfoData = taosArrayGet(p->pDataBlock, 5);
(void)colDataSetVal(pColInfoData, numOfRows, (const char *)&pm->schema[j].bytes, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, (const char *)&pm->schema[j].bytes, false), &lino, _OVER);
for (int32_t k = 6; k <= 8; ++k) {
pColInfoData = taosArrayGet(p->pDataBlock, k);
colDataSetNULL(pColInfoData, numOfRows);
@ -3664,7 +3711,8 @@ static int32_t buildDbColsInfoBlock(const SSDataBlock *p, const SSysTableMeta *p
numOfRows += 1;
}
}
_OVER:
mError("failed at %s:%d since %s", __FUNCTION__, lino, tstrerror(code));
return numOfRows;
}
#define BUILD_COL_FOR_INFO_DB 1
@ -3712,6 +3760,8 @@ static int32_t mndRetrieveStbCol(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
SSdb *pSdb = pMnode->pSdb;
SStbObj *pStb = NULL;
int32_t numOfRows = 0;
int32_t lino = 0;
int32_t code = 0;
buildWhichDBs = determineBuildColForWhichDBs(pShow->db);
@ -3771,26 +3821,26 @@ static int32_t mndRetrieveStbCol(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
mDebug("mndRetrieveStbCol get stable cols, stable name:%s, db:%s", pStb->name, pStb->db);
char db[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
(void)tNameFromString(&name, pStb->db, T_NAME_ACCT | T_NAME_DB);
(void)tNameGetDbName(&name, varDataVal(db));
RETRIEVE_CHECK_GOTO(tNameFromString(&name, pStb->db, T_NAME_ACCT | T_NAME_DB), pStb, &lino, _OVER);
RETRIEVE_CHECK_GOTO(tNameGetDbName(&name, varDataVal(db)), pStb, &lino, _OVER);
varDataSetLen(db, strlen(varDataVal(db)));
for (int i = 0; i < pStb->numOfColumns; i++) {
int32_t cols = 0;
SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)stbName, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stbName, false), pStb, &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)db, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)db, false), pStb, &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, typeName, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, typeName, false), pStb, &lino, _OVER);
// col name
char colName[TSDB_COL_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
STR_TO_VARSTR(colName, pStb->pColumns[i].name);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, colName, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, colName, false), pStb, &lino, _OVER);
// col type
int8_t colType = pStb->pColumns[i].type;
@ -3805,10 +3855,11 @@ static int32_t mndRetrieveStbCol(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
(int32_t)((pStb->pColumns[i].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));
}
varDataSetLen(colTypeStr, colTypeLen);
(void)colDataSetVal(pColInfo, numOfRows, (char *)colTypeStr, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (char *)colTypeStr, false), pStb, &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->pColumns[i].bytes, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->pColumns[i].bytes, false), pStb,
&lino, _OVER);
while (cols < pShow->numOfColumns) {
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetNULL(pColInfo, numOfRows);
@ -3824,9 +3875,15 @@ static int32_t mndRetrieveStbCol(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
}
}
pShow->numOfRows += numOfRows;
mDebug("mndRetrieveStbCol success, rows:%d, pShow->numOfRows:%d", numOfRows, pShow->numOfRows);
goto _OVER;
_ERROR:
mError("failed to mndRetrieveStbCol, rows:%d, pShow->numOfRows:%d, at %s:%d since %s", numOfRows, pShow->numOfRows,
__FUNCTION__, lino, tstrerror(code));
_OVER:
pShow->numOfRows += numOfRows;
return numOfRows;
}
@ -4111,8 +4168,9 @@ static void *mndBuildVDropTbsReq(SMnode *pMnode, const SVgroupInfo *pVgInfo, con
void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));
tEncoderInit(&encoder, pBuf, contLen - sizeof(SMsgHead));
(void)tEncodeSVDropTbBatchReq(&encoder, pReq);
int32_t code = tEncodeSVDropTbBatchReq(&encoder, pReq);
tEncoderClear(&encoder);
if (code != 0) return NULL;
*len = contLen;
return pHead;

43
source/dnode/mnode/impl/src/mndSync.c
View File

@ -238,7 +238,9 @@ static int32_t mndPostMgmtCode(SMnode *pMnode, int32_t code) {
pMgmt->transSec = 0;
pMgmt->transSeq = 0;
pMgmt->errCode = code;
(void)tsem_post(&pMgmt->syncSem);
if (tsem_post(&pMgmt->syncSem) < 0) {
mError("trans:%d, failed to post sem", transId);
}
if (pMgmt->errCode != 0) {
mError("trans:%d, failed to propose since %s, post sem", transId, tstrerror(pMgmt->errCode));
@ -310,11 +312,15 @@ void mndRestoreFinish(const SSyncFSM *pFsm, const SyncIndex commitIdx) {
} else {
mInfo("vgId:1, sync restore finished");
}
(void)mndRefreshUserIpWhiteList(pMnode);
int32_t code = mndRefreshUserIpWhiteList(pMnode);
if (code != 0) {
mError("vgId:1, failed to refresh user ip white list since %s", tstrerror(code));
mndSetRestored(pMnode, false);
}
SyncIndex fsmIndex = mndSyncAppliedIndex(pFsm);
if (commitIdx != fsmIndex) {
mError("vgId:1, sync restore finished, but commitIdx:%" PRId64 " is not equal to appliedIdx:%" PRId64, commitIdx,
mError("vgId:1, failed to sync restore, commitIdx:%" PRId64 " is not equal to appliedIdx:%" PRId64, commitIdx,
fsmIndex);
mndSetRestored(pMnode, false);
}
@ -368,7 +374,9 @@ static void mndBecomeFollower(const SSyncFSM *pFsm) {
pMgmt->transSec = 0;
pMgmt->transSeq = 0;
pMgmt->errCode = TSDB_CODE_SYN_NOT_LEADER;
(void)tsem_post(&pMgmt->syncSem);
if (tsem_post(&pMgmt->syncSem) < 0) {
mError("failed to post sem");
}
}
(void)taosThreadMutexUnlock(&pMgmt->lock);
@ -387,7 +395,9 @@ static void mndBecomeLearner(const SSyncFSM *pFsm) {
pMgmt->transSec = 0;
pMgmt->transSeq = 0;
pMgmt->errCode = TSDB_CODE_SYN_NOT_LEADER;
(void)tsem_post(&pMgmt->syncSem);
if (tsem_post(&pMgmt->syncSem) < 0) {
mError("failed to post sem");
}
}
(void)taosThreadMutexUnlock(&pMgmt->lock);
}
@ -487,13 +497,18 @@ int32_t mndInitSync(SMnode *pMnode) {
pNode->nodePort = pMgmt->replicas[i].port;
tstrncpy(pNode->nodeFqdn, pMgmt->replicas[i].fqdn, sizeof(pNode->nodeFqdn));
pNode->nodeRole = pMgmt->nodeRoles[i];
(void)tmsgUpdateDnodeInfo(&pNode->nodeId, &pNode->clusterId, pNode->nodeFqdn, &pNode->nodePort);
if (tmsgUpdateDnodeInfo(&pNode->nodeId, &pNode->clusterId, pNode->nodeFqdn, &pNode->nodePort) != true) {
mError("failed to open sync, tmsgUpdateDnodeInfo is false");
}
mInfo("vgId:1, index:%d ep:%s:%u dnode:%d cluster:%" PRId64, i, pNode->nodeFqdn, pNode->nodePort, pNode->nodeId,
pNode->clusterId);
}
int32_t code = 0;
(void)tsem_init(&pMgmt->syncSem, 0, 0);
if ((code = tsem_init(&pMgmt->syncSem, 0, 0)) < 0) {
mError("failed to open sync, tsem_init, since %s", tstrerror(code));
TAOS_RETURN(code);
}
pMgmt->sync = syncOpen(&syncInfo, 1); // always check
if (pMgmt->sync <= 0) {
if (terrno != 0) code = terrno;
@ -511,7 +526,9 @@ void mndCleanupSync(SMnode *pMnode) {
syncStop(pMgmt->sync);
mInfo("mnode-sync is stopped, id:%" PRId64, pMgmt->sync);
(void)tsem_destroy(&pMgmt->syncSem);
if (tsem_destroy(&pMgmt->syncSem) < 0) {
mError("failed to destroy sem");
}
(void)taosThreadMutexDestroy(&pMgmt->lock);
memset(pMgmt, 0, sizeof(SSyncMgmt));
}
@ -531,7 +548,9 @@ void mndSyncCheckTimeout(SMnode *pMnode) {
pMgmt->transSeq = 0;
terrno = TSDB_CODE_SYN_TIMEOUT;
pMgmt->errCode = TSDB_CODE_SYN_TIMEOUT;
(void)tsem_post(&pMgmt->syncSem);
if (tsem_post(&pMgmt->syncSem) < 0) {
mError("failed to post sem");
}
} else {
mDebug("trans:%d, waiting for sync confirm, start:%d cur:%d delta:%d seq:%" PRId64, pMgmt->transId,
pMgmt->transSec, curSec, curSec - pMgmt->transSec, pMgmt->transSeq);
@ -572,7 +591,7 @@ int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw, int32_t transId) {
mInfo("trans:%d, is proposing and wait sem, seq:%" PRId64, transId, seq);
pMgmt->transSeq = seq;
(void)taosThreadMutexUnlock(&pMgmt->lock);
(void)tsem_wait(&pMgmt->syncSem);
code = tsem_wait(&pMgmt->syncSem);
} else if (code > 0) {
mInfo("trans:%d, confirm at once since replica is 1, continue execute", transId);
pMgmt->transId = 0;
@ -624,7 +643,9 @@ void mndSyncStop(SMnode *pMnode) {
pMgmt->transId = 0;
pMgmt->transSec = 0;
pMgmt->errCode = TSDB_CODE_APP_IS_STOPPING;
(void)tsem_post(&pMgmt->syncSem);
if (tsem_post(&pMgmt->syncSem) < 0) {
mError("failed to post sem");
}
}
(void)taosThreadMutexUnlock(&pMgmt->lock);
}

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

@ -2800,7 +2800,7 @@ static int32_t mndLoopHash(SHashObj *hash, char *priType, SSDataBlock *pBlock, i
void *key = taosHashGetKey(value, &keyLen);
char dbName[TSDB_DB_NAME_LEN] = {0};
mndExtractShortDbNameFromStbFullName(key, dbName);
(void)mndExtractShortDbNameFromStbFullName(key, dbName);
char dbNameContent[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
STR_WITH_MAXSIZE_TO_VARSTR(dbNameContent, dbName, pShow->pMeta->pSchemas[cols].bytes);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

33
source/dnode/mnode/impl/src/mndVgroup.c
View File

@ -2847,8 +2847,8 @@ int32_t mndBuildRaftAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *pO
TAOS_RETURN(code);
}
int32_t mndBuildRestoreAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *db, SVgObj *pVgroup,
SDnodeObj *pDnode) {
int32_t mndBuildRestoreAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *db, SVgObj *pVgroup, SDnodeObj *pDnode,
SDnodeObj *pAnotherDnode) {
int32_t code = 0;
SVgObj newVgroup = {0};
memcpy(&newVgroup, pVgroup, sizeof(SVgObj));
@ -2864,7 +2864,33 @@ int32_t mndBuildRestoreAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj
}
}
TAOS_CHECK_RETURN(mndAddCreateVnodeAction(pMnode, pTrans, db, &newVgroup, &newVgroup.vnodeGid[selected]));
} else if (newVgroup.replica == 2 || newVgroup.replica == 3) {
} else if (newVgroup.replica == 2) {
for (int i = 0; i < newVgroup.replica; i++) {
if (newVgroup.vnodeGid[i].dnodeId == pDnode->id) {
newVgroup.vnodeGid[i].nodeRole = TAOS_SYNC_ROLE_LEARNER;
} else {
newVgroup.vnodeGid[i].nodeRole = TAOS_SYNC_ROLE_VOTER;
}
}
TAOS_CHECK_RETURN(mndRestoreAddAlterVnodeTypeAction(pMnode, pTrans, db, &newVgroup, pAnotherDnode));
for (int i = 0; i < newVgroup.replica; i++) {
if (newVgroup.vnodeGid[i].dnodeId == pDnode->id) {
newVgroup.vnodeGid[i].nodeRole = TAOS_SYNC_ROLE_LEARNER;
} else {
newVgroup.vnodeGid[i].nodeRole = TAOS_SYNC_ROLE_VOTER;
}
}
TAOS_CHECK_RETURN(mndRestoreAddCreateVnodeAction(pMnode, pTrans, db, &newVgroup, pDnode));
for (int i = 0; i < newVgroup.replica; i++) {
newVgroup.vnodeGid[i].nodeRole = TAOS_SYNC_ROLE_VOTER;
if (newVgroup.vnodeGid[i].dnodeId == pDnode->id) {
}
}
TAOS_CHECK_RETURN(mndRestoreAddAlterVnodeTypeAction(pMnode, pTrans, db, &newVgroup, pDnode));
TAOS_CHECK_RETURN(mndRestoreAddAlterVnodeTypeAction(pMnode, pTrans, db, &newVgroup, pAnotherDnode));
} else if (newVgroup.replica == 3) {
for (int i = 0; i < newVgroup.replica; i++) {
if (newVgroup.vnodeGid[i].dnodeId == pDnode->id) {
newVgroup.vnodeGid[i].nodeRole = TAOS_SYNC_ROLE_LEARNER;
@ -2881,7 +2907,6 @@ int32_t mndBuildRestoreAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj
}
TAOS_CHECK_RETURN(mndRestoreAddAlterVnodeTypeAction(pMnode, pTrans, db, &newVgroup, pDnode));
}
SSdbRaw *pVgRaw = mndVgroupActionEncode(&newVgroup);
if (pVgRaw == NULL) {
code = TSDB_CODE_MND_RETURN_VALUE_NULL;

2
source/dnode/vnode/src/meta/metaTable.c
View File

@ -232,7 +232,7 @@ int metaDelJsonVarFromIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry, const SSche
}
int32_t len = taosUcs4ToMbs((TdUcs4 *)pTagVal->pData, pTagVal->nData, val + VARSTR_HEADER_SIZE);
if (len < 0) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, NULL, _exception);
TAOS_CHECK_GOTO(len, NULL, _exception);
}
memcpy(val, (uint16_t *)&len, VARSTR_HEADER_SIZE);
type = TSDB_DATA_TYPE_VARCHAR;

15
source/dnode/vnode/src/tsdb/tsdbCache.c
View File

@ -1841,9 +1841,9 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
TAOS_CHECK_EXIT(code);
} else {
if (!remainCols) {
remainCols = taosArrayInit(numCols * 2, sizeof(SLastKey));
remainCols = taosArrayInit(numCols * 2, sizeof(SIdxKey));
}
if (!taosArrayPush(remainCols, &lastKey)) {
if (!taosArrayPush(remainCols, &(SIdxKey){i, lastKey})) {
TAOS_CHECK_EXIT(TSDB_CODE_OUT_OF_MEMORY);
}
}
@ -1871,7 +1871,9 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
code = terrno;
goto _exit;
}
((SLastKey *)key)[0] = *(SLastKey *)taosArrayGet(remainCols, i);
SIdxKey* idxKey = taosArrayGet(remainCols, i);
((SLastKey *)key)[0] = idxKey->key;
keys_list[i] = key;
keys_list_sizes[i] = klen;
@ -1887,10 +1889,11 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
for (int i = 0; i < numKeys; ++i) {
SLastCol *pLastCol = NULL;
(void)tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol);
SLastKey *pLastKey = (SLastKey *)keys_list[i];
SIdxKey* idxKey = taosArrayGet(remainCols, i);
SLastKey *pLastKey = &idxKey->key;
if (NULL != pLastCol && (pLastCol->rowKey.ts <= eKey && pLastCol->rowKey.ts >= sKey)) {
SLastCol noCacheCol = {.rowKey.ts = TSKEY_MIN,
.colVal = COL_VAL_NONE(pLastKey->cid, pTSchema->columns[i].type),
.colVal = COL_VAL_NONE(pLastKey->cid, pTSchema->columns[idxKey->idx].type),
.cacheStatus = TSDB_LAST_CACHE_NO_CACHE};
if ((code = tsdbCachePutToRocksdb(pTsdb, pLastKey, &noCacheCol)) != TSDB_CODE_SUCCESS) {
@ -1938,7 +1941,7 @@ _exit:
int32_t tsdbOpenCache(STsdb *pTsdb) {
int32_t code = 0, lino = 0;
size_t cfgCapacity = pTsdb->pVnode->config.cacheLastSize * 1024 * 1024;
size_t cfgCapacity = (size_t)pTsdb->pVnode->config.cacheLastSize * 1024 * 1024;
SLRUCache *pCache = taosLRUCacheInit(cfgCapacity, 0, .5);
if (pCache == NULL) {

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

@ -853,7 +853,7 @@ int32_t catalogInit(SCatalogCfg* cfg) {
}
if (tsem_init(&gCtgMgmt.queue.reqSem, 0, 0)) {
qError("tsem_init failed, error:%s", tstrerror(TAOS_SYSTEM_ERROR(errno)));
qError("tsem_init failed, terror:%s", tstrerror(terrno));
CTG_ERR_RET(TSDB_CODE_CTG_SYS_ERROR);
}

1
source/libs/executor/src/exchangeoperator.c
View File

@ -565,7 +565,6 @@ int32_t loadRemoteDataCallback(void* param, SDataBuf* pMsg, int32_t code) {
pSourceDataInfo->status = EX_SOURCE_DATA_READY;
code = tsem_post(&pExchangeInfo->ready);
if (code != TSDB_CODE_SUCCESS) {
code = TAOS_SYSTEM_ERROR(code);
qError("failed to invoke post when fetch rsp is ready, code:%s, %p", tstrerror(code), pExchangeInfo);
return code;
}

2
source/libs/index/inc/indexComm.h
View File

@ -30,7 +30,7 @@ char* idxPackJsonData(SIndexTerm* itm);
char* idxPackJsonDataPrefix(SIndexTerm* itm, int32_t* skip);
char* idxPackJsonDataPrefixNoType(SIndexTerm* itm, int32_t* skip);
typedef enum { MATCH, CONTINUE, BREAK } TExeCond;
typedef enum { MATCH, CONTINUE, BREAK, FAILED } TExeCond;
typedef TExeCond (*_cache_range_compare)(void* a, void* b, int8_t type);

6
source/libs/index/src/indexCache.c
View File

@ -153,9 +153,8 @@ static int32_t cacheSearchCompareFunc(void* cache, SIndexTerm* term, SIdxTRslt*
break;
}
CacheTerm* c = (CacheTerm*)SL_GET_NODE_DATA(node);
terrno = TSDB_CODE_SUCCESS;
TExeCond cond = cmpFn(c->colVal, pCt->colVal, pCt->colType);
if (terrno != TSDB_CODE_SUCCESS) {
if (cond == FAILED) {
code = terrno;
goto _return;
}
@ -335,10 +334,9 @@ static int32_t cacheSearchCompareFunc_JSON(void* cache, SIndexTerm* term, SIdxTR
goto _return;
}
memcpy(p, c->colVal, strlen(c->colVal));
terrno = TSDB_CODE_SUCCESS;
cond = cmpFn(p + skip, term->colVal, dType);
taosMemoryFree(p);
if (terrno != TSDB_CODE_SUCCESS) {
if (cond == FAILED) {
code = terrno;
goto _return;
}

56
source/libs/index/src/indexComm.c
View File

@ -118,6 +118,14 @@ static FORCE_INLINE TExeCond tCompareContains(void* a, void* b, int8_t type) {
}
return tCompare(func, QUERY_TERM, a, b, type);
}
#define CHECKCOMERROR(expr) \
do { \
if ((expr) != 0) { \
return FAILED; \
} \
} while (0)
static FORCE_INLINE TExeCond tCompareEqual(void* a, void* b, int8_t type) {
__compar_fn_t func = idxGetCompar(type);
if (func == NULL) {
@ -132,36 +140,52 @@ TExeCond tCompare(__compar_fn_t func, int8_t cmptype, void* a, void* b, int8_t d
}
#if 1
if (dtype == TSDB_DATA_TYPE_TIMESTAMP) {
int64_t va = taosStr2int64(a);
int64_t vb = taosStr2int64(b);
int64_t va;
CHECKCOMERROR(taosStr2int64(a, &va));
int64_t vb;
CHECKCOMERROR(taosStr2int64(b, &vb));
return tDoCompare(func, cmptype, &va, &vb);
} else if (dtype == TSDB_DATA_TYPE_BOOL || dtype == TSDB_DATA_TYPE_UTINYINT) {
uint8_t va = taosStr2int64(a);
uint8_t vb = taosStr2int64(b);
uint8_t va;
CHECKCOMERROR(taosStr2int8(a, &va));
uint8_t vb;
CHECKCOMERROR(taosStr2int8(b, &vb));
return tDoCompare(func, cmptype, &va, &vb);
} else if (dtype == TSDB_DATA_TYPE_TINYINT) {
int8_t va = taosStr2int64(a);
int8_t vb = taosStr2int64(b);
int8_t va;
CHECKCOMERROR(taosStr2int8(a, &va));
int8_t vb;
CHECKCOMERROR(taosStr2int8(b, &vb));
return tDoCompare(func, cmptype, &va, &vb);
} else if (dtype == TSDB_DATA_TYPE_SMALLINT) {
int16_t va = taosStr2int64(a);
int16_t vb = taosStr2int64(b);
int16_t va;
CHECKCOMERROR(taosStr2int16(a, &va));
int16_t vb;
CHECKCOMERROR(taosStr2int16(b, &vb));
return tDoCompare(func, cmptype, &va, &vb);
} else if (dtype == TSDB_DATA_TYPE_USMALLINT) {
uint16_t va = taosStr2int64(a);
uint16_t vb = taosStr2int64(b);
uint16_t va;
CHECKCOMERROR(taosStr2int16(a, &va));
uint16_t vb;
CHECKCOMERROR(taosStr2int16(b, &vb));
return tDoCompare(func, cmptype, &va, &vb);
} else if (dtype == TSDB_DATA_TYPE_INT) {
int32_t va = taosStr2int64(a);
int32_t vb = taosStr2int64(b);
int32_t va;
CHECKCOMERROR(taosStr2int32(a, &va));
int32_t vb;
CHECKCOMERROR(taosStr2int32(b, &vb));
return tDoCompare(func, cmptype, &va, &vb);
} else if (dtype == TSDB_DATA_TYPE_UINT) {
uint32_t va = taosStr2int64(a);
uint32_t vb = taosStr2int64(b);
uint32_t va;
CHECKCOMERROR(taosStr2int32(a, &va));
uint32_t vb;
CHECKCOMERROR(taosStr2int32(b, &vb));
return tDoCompare(func, cmptype, &va, &vb);
} else if (dtype == TSDB_DATA_TYPE_BIGINT) {
int64_t va = taosStr2int64(a);
int64_t vb = taosStr2int64(b);
int64_t va;
CHECKCOMERROR(taosStr2int64(a, &va));
int64_t vb;
CHECKCOMERROR(taosStr2int64(b, &vb));
return tDoCompare(func, cmptype, &va, &vb);
} else if (dtype == TSDB_DATA_TYPE_UBIGINT) {
uint64_t va, vb;

6
source/libs/index/src/indexTfile.c
View File

@ -367,9 +367,8 @@ static int32_t tfSearchCompareFunc(void* reader, SIndexTerm* tem, SIdxTRslt* tr,
FstSlice* s = &rt->data;
char* ch = (char*)fstSliceData(s, NULL);
terrno = TSDB_CODE_SUCCESS;
TExeCond cond = cmpFn(ch, p, tem->colType);
if (TSDB_CODE_SUCCESS != terrno) {
if (FAILED == cond) {
swsResultDestroy(rt);
code = terrno;
goto _return;
@ -520,10 +519,9 @@ static int32_t tfSearchCompareFunc_JSON(void* reader, SIndexTerm* tem, SIdxTRslt
goto _return;
}
memcpy(tBuf, ch, sz);
terrno = TSDB_CODE_SUCCESS;
cond = cmpFn(tBuf + skip, tem->colVal, IDX_TYPE_GET_TYPE(tem->colType));
taosMemoryFree(tBuf);
if (TSDB_CODE_SUCCESS != terrno) {
if (FAILED == cond) {
swsResultDestroy(rt);
code = terrno;
goto _return;

8
source/libs/index/test/utilUT.cc
View File

@ -328,13 +328,17 @@ TEST_F(UtilEnv, testFill) {
int64_t val = i;
char buf[65] = {0};
idxInt2str(val, buf, 1);
EXPECT_EQ(val, taosStr2int64(buf));
int64_t ret = 0;
taosStr2int64(buf, &ret);
EXPECT_EQ(val, ret);
}
for (int i = 0; i < 1000000; i++) {
int64_t val = 0 - i;
char buf[65] = {0};
idxInt2str(val, buf, -1);
EXPECT_EQ(val, taosStr2int64(buf));
int64_t ret = 0;
taosStr2int64(buf, &ret);
EXPECT_EQ(val, ret);
}
}
TEST_F(UtilEnv, TempResult) {

10
source/libs/parser/src/parInsertSml.c
View File

@ -143,13 +143,13 @@ static int32_t smlBuildTagRow(SArray* cols, SBoundColInfo* tags, SSchema* pSchem
goto end;
}
if (!taosMbsToUcs4(kv->value, kv->length, (TdUcs4*)(p), kv->length * TSDB_NCHAR_SIZE, &output)) {
if (errno == E2BIG) {
if (terrno == TAOS_SYSTEM_ERROR(E2BIG)) {
taosMemoryFree(p);
code = generateSyntaxErrMsg(msg, TSDB_CODE_PAR_VALUE_TOO_LONG, pTagSchema->name);
goto end;
}
char buf[512] = {0};
(void)snprintf(buf, tListLen(buf), " taosMbsToUcs4 error:%s", strerror(errno));
(void)snprintf(buf, tListLen(buf), " taosMbsToUcs4 error:%s", strerror(terrno));
taosMemoryFree(p);
code = buildSyntaxErrMsg(msg, buf, kv->value);
goto end;
@ -257,7 +257,7 @@ int32_t smlBuildCol(STableDataCxt* pTableCxt, SSchema* schema, void* data, int32
goto end;
}
if (!taosMbsToUcs4(kv->value, kv->length, (TdUcs4*)pUcs4, size, &len)) {
if (errno == E2BIG) {
if (terrno == TAOS_SYSTEM_ERROR(E2BIG)) {
taosMemoryFree(pUcs4);
ret = TSDB_CODE_PAR_VALUE_TOO_LONG;
goto end;
@ -404,14 +404,14 @@ int32_t smlBindData(SQuery* query, bool dataFormat, SArray* tags, SArray* colsSc
goto end;
}
if (!taosMbsToUcs4(kv->value, kv->length, (TdUcs4*)pUcs4, pColSchema->bytes - VARSTR_HEADER_SIZE, &len)) {
if (errno == E2BIG) {
if (terrno == TAOS_SYSTEM_ERROR(E2BIG)) {
uError("sml bind taosMbsToUcs4 error, kv length:%d, bytes:%d, kv->value:%s", (int)kv->length,
pColSchema->bytes, kv->value);
(void)buildInvalidOperationMsg(&pBuf, "value too long");
ret = TSDB_CODE_PAR_VALUE_TOO_LONG;
goto end;
}
ret = buildInvalidOperationMsg(&pBuf, strerror(errno));
ret = buildInvalidOperationMsg(&pBuf, strerror(terrno));
goto end;
}
pVal->value.pData = pUcs4;

8
source/libs/parser/src/parInsertSql.c
View File

@ -682,12 +682,12 @@ static int32_t parseTagToken(const char** end, SToken* pToken, SSchema* pSchema,
return TSDB_CODE_OUT_OF_MEMORY;
}
if (!taosMbsToUcs4(pToken->z, pToken->n, (TdUcs4*)(p), realLen, &output)) {
if (errno == E2BIG) {
if (terrno == TAOS_SYSTEM_ERROR(E2BIG)) {
taosMemoryFree(p);
return generateSyntaxErrMsg(pMsgBuf, TSDB_CODE_PAR_VALUE_TOO_LONG, pSchema->name);
}
char buf[512] = {0};
snprintf(buf, tListLen(buf), " taosMbsToUcs4 error:%s", strerror(errno));
snprintf(buf, tListLen(buf), " taosMbsToUcs4 error:%s", strerror(terrno));
taosMemoryFree(p);
return buildSyntaxErrMsg(pMsgBuf, buf, pToken->z);
}
@ -1616,11 +1616,11 @@ static int32_t parseValueTokenImpl(SInsertParseContext* pCxt, const char** pSql,
}
if (!taosMbsToUcs4(pToken->z, pToken->n, (TdUcs4*)pUcs4, realLen, &len)) {
taosMemoryFree(pUcs4);
if (errno == E2BIG) {
if (terrno == TAOS_SYSTEM_ERROR(E2BIG)) {
return generateSyntaxErrMsg(&pCxt->msg, TSDB_CODE_PAR_VALUE_TOO_LONG, pSchema->name);
}
char buf[512] = {0};
snprintf(buf, tListLen(buf), "%s", strerror(errno));
snprintf(buf, tListLen(buf), "%s", strerror(terrno));
return buildSyntaxErrMsg(&pCxt->msg, buf, pToken->z);
}
pVal->value.pData = pUcs4;

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

@ -194,13 +194,13 @@ int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, const ch
goto end;
}
if (!taosMbsToUcs4(bind[c].buffer, colLen, (TdUcs4*)(p), colLen * TSDB_NCHAR_SIZE, &output)) {
if (errno == E2BIG) {
if (terrno == TAOS_SYSTEM_ERROR(E2BIG)) {
taosMemoryFree(p);
code = generateSyntaxErrMsg(&pBuf, TSDB_CODE_PAR_VALUE_TOO_LONG, pTagSchema->name);
goto end;
}
char buf[512] = {0};
snprintf(buf, tListLen(buf), " taosMbsToUcs4 error:%s", strerror(errno));
snprintf(buf, tListLen(buf), " taosMbsToUcs4 error:%s", strerror(terrno));
taosMemoryFree(p);
code = buildSyntaxErrMsg(&pBuf, buf, bind[c].buffer);
goto end;
@ -274,11 +274,11 @@ int32_t convertStmtNcharCol(SMsgBuf* pMsgBuf, SSchema* pSchema, TAOS_MULTI_BIND*
if (!taosMbsToUcs4(((char*)src->buffer) + src->buffer_length * i, src->length[i],
(TdUcs4*)(((char*)dst->buffer) + dst->buffer_length * i), dst->buffer_length, &output)) {
if (errno == E2BIG) {
if (terrno == TAOS_SYSTEM_ERROR(E2BIG)) {
return generateSyntaxErrMsg(pMsgBuf, TSDB_CODE_PAR_VALUE_TOO_LONG, pSchema->name);
}
char buf[512] = {0};
snprintf(buf, tListLen(buf), "%s", strerror(errno));
snprintf(buf, tListLen(buf), "%s", strerror(terrno));
return buildSyntaxErrMsg(pMsgBuf, buf, NULL);
}
@ -534,13 +534,13 @@ int32_t qBindStmtTagsValue2(void* pBlock, void* boundTags, int64_t suid, const c
goto end;
}
if (!taosMbsToUcs4(bind[c].buffer, colLen, (TdUcs4*)(p), colLen * TSDB_NCHAR_SIZE, &output)) {
if (errno == E2BIG) {
if (terrno == TAOS_SYSTEM_ERROR(E2BIG)) {
taosMemoryFree(p);
code = generateSyntaxErrMsg(&pBuf, TSDB_CODE_PAR_VALUE_TOO_LONG, pTagSchema->name);
goto end;
}
char buf[512] = {0};
snprintf(buf, tListLen(buf), " taosMbsToUcs4 error:%s", strerror(errno));
snprintf(buf, tListLen(buf), " taosMbsToUcs4 error:%s", strerror(terrno));
taosMemoryFree(p);
code = buildSyntaxErrMsg(&pBuf, buf, bind[c].buffer);
goto end;
@ -610,11 +610,11 @@ static int32_t convertStmtStbNcharCol2(SMsgBuf* pMsgBuf, SSchema* pSchema, TAOS_
}
if (!taosMbsToUcs4(src_buf, src->length[i], (TdUcs4*)dst_buf, max_buf_len, &output)) {
if (errno == E2BIG) {
if (terrno == TAOS_SYSTEM_ERROR(E2BIG)) {
return generateSyntaxErrMsg(pMsgBuf, TSDB_CODE_PAR_VALUE_TOO_LONG, pSchema->name);
}
char buf[512] = {0};
snprintf(buf, tListLen(buf), "%s", strerror(errno));
snprintf(buf, tListLen(buf), "%s", strerror(terrno));
return buildSyntaxErrMsg(pMsgBuf, buf, NULL);
}
@ -740,11 +740,11 @@ static int32_t convertStmtNcharCol2(SMsgBuf* pMsgBuf, SSchema* pSchema, TAOS_STM
/*if (!taosMbsToUcs4(((char*)src->buffer) + src->buffer_length * i, src->length[i],
(TdUcs4*)(((char*)dst->buffer) + dst->buffer_length * i), dst->buffer_length, &output)) {*/
if (!taosMbsToUcs4(src_buf, src->length[i], (TdUcs4*)dst_buf, max_buf_len, &output)) {
if (errno == E2BIG) {
if (terrno == TAOS_SYSTEM_ERROR(E2BIG)) {
return generateSyntaxErrMsg(pMsgBuf, TSDB_CODE_PAR_VALUE_TOO_LONG, pSchema->name);
}
char buf[512] = {0};
snprintf(buf, tListLen(buf), "%s", strerror(errno));
snprintf(buf, tListLen(buf), "%s", strerror(terrno));
return buildSyntaxErrMsg(pMsgBuf, buf, NULL);
}

2
source/libs/parser/src/parUtil.c
View File

@ -469,7 +469,7 @@ int32_t parseJsontoTagData(const char* json, SArray* pTagVals, STag** ppTag, voi
val.type = TSDB_DATA_TYPE_NCHAR;
if (valLen > 0 && !taosMbsToUcs4(jsonValue, valLen, (TdUcs4*)tmp, (int32_t)(valLen * TSDB_NCHAR_SIZE), &valLen)) {
uError("charset:%s to %s. val:%s, errno:%s, convert failed.", DEFAULT_UNICODE_ENCODEC, tsCharset, jsonValue,
strerror(errno));
strerror(terrno));
retCode = buildSyntaxErrMsg(pMsgBuf, "charset convert json error", jsonValue);
taosMemoryFree(tmp);
goto end;

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

@ -198,7 +198,7 @@ static int32_t setValueByBindParam(SValueNode* pVal, TAOS_MULTI_BIND* pParam) {
int32_t output = 0;
if (!taosMbsToUcs4(pParam->buffer, inputSize, (TdUcs4*)varDataVal(pVal->datum.p), pVal->node.resType.bytes,
&output)) {
return errno;
return terrno;
}
varDataSetLen(pVal->datum.p, output);
pVal->node.resType.bytes = output + VARSTR_HEADER_SIZE;
@ -486,7 +486,7 @@ static int32_t setValueByBindParam2(SValueNode* pVal, TAOS_STMT2_BIND* pParam) {
int32_t output = 0;
if (!taosMbsToUcs4(pParam->buffer, inputSize, (TdUcs4*)varDataVal(pVal->datum.p), pVal->node.resType.bytes,
&output)) {
return errno;
return terrno;
}
varDataSetLen(pVal->datum.p, output);
pVal->node.resType.bytes = output + VARSTR_HEADER_SIZE;

4
source/libs/tfs/src/tfsDisk.c
View File

@ -33,7 +33,7 @@ int32_t tfsNewDisk(int32_t level, int32_t id, int8_t disable, const char *path,
pDisk->id = id;
pDisk->disable = disable;
if (taosGetDiskSize(pDisk->path, &pDisk->size) < 0) {
code = TAOS_SYSTEM_ERROR(errno); // TODO: refactor this line
code = terrno;
TAOS_CHECK_GOTO(code, &lino, _exit);
}
_exit:
@ -57,7 +57,7 @@ STfsDisk *tfsFreeDisk(STfsDisk *pDisk) {
int32_t tfsUpdateDiskSize(STfsDisk *pDisk) {
if (taosGetDiskSize(pDisk->path, &pDisk->size) < 0) {
int32_t code = TAOS_SYSTEM_ERROR(errno); // TODO: refactor this line
int32_t code = terrno;
fError("failed to get disk:%s size, level:%d id:%d since %s", pDisk->path, pDisk->level, pDisk->id,
tstrerror(code));
TAOS_RETURN(code);

8
source/libs/transport/src/transCli.c
View File

@ -3061,7 +3061,7 @@ int32_t transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STra
code = tsem_init(sem, 0, 0);
if (code != 0) {
taosMemoryFree(sem);
TAOS_CHECK_GOTO(TAOS_SYSTEM_ERROR(errno), NULL, _RETURN1);
TAOS_CHECK_GOTO(terrno, NULL, _RETURN1);
}
if (pReq->info.traceId.msgId == 0) TRACE_SET_MSGID(&pReq->info.traceId, tGenIdPI64());
@ -3224,9 +3224,8 @@ int32_t transSendRecvWithTimeout(void* shandle, SEpSet* pEpSet, STransMsg* pReq,
}
code = tsem2_timewait(pSyncMsg->pSem, timeoutMs);
if (code < 0) {
pRsp->code = TSDB_CODE_TIMEOUT_ERROR;
code = TSDB_CODE_TIMEOUT_ERROR;
if (code != 0) {
pRsp->code = code;
} else {
memcpy(pRsp, pSyncMsg->pRsp, sizeof(STransMsg));
pSyncMsg->pRsp->pCont = NULL;
@ -3234,7 +3233,6 @@ int32_t transSendRecvWithTimeout(void* shandle, SEpSet* pEpSet, STransMsg* pReq,
epsetAssign(pEpSet, &pSyncMsg->epSet);
*epUpdated = 1;
}
code = 0;
}
_RETURN:
(void)transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);

8
source/os/src/osDir.c
View File

@ -104,7 +104,7 @@ void taosRemoveDir(const char *dirname) {
if (taosDirEntryIsDir(de)) {
taosRemoveDir(filename);
} else {
(void)taosRemoveFile(filename);
TAOS_UNUSED(taosRemoveFile(filename));
// printf("file:%s is removed\n", filename);
}
}
@ -315,7 +315,7 @@ void taosRemoveOldFiles(const char *dirname, int32_t keepDays) {
if (fileSec <= 100) continue;
int32_t days = (int32_t)(TABS(sec - fileSec) / 86400 + 1);
if (days > keepDays) {
(void)taosRemoveFile(filename);
TAOS_UNUSED(taosRemoveFile(filename));
uInfo("file:%s is removed, days:%d keepDays:%d, sed:%"PRId64, filename, days, keepDays, fileSec);
} else {
// printf("file:%s won't be removed, days:%d keepDays:%d", filename, days, keepDays);
@ -323,8 +323,8 @@ void taosRemoveOldFiles(const char *dirname, int32_t keepDays) {
}
}
(void)taosCloseDir(&pDir);
(void)rmdir(dirname);
TAOS_UNUSED(taosCloseDir(&pDir));
TAOS_UNUSED(rmdir(dirname));
}
int32_t taosExpandDir(const char *dirname, char *outname, int32_t maxlen) {

10
source/os/src/osEnv.c
View File

@ -94,16 +94,18 @@ int32_t osDefaultInit() {
return code;
}
void osUpdate() {
int32_t osUpdate() {
int code = 0;
if (tsLogDir[0] != 0) {
(void)taosGetDiskSize(tsLogDir, &tsLogSpace.size);
code = taosGetDiskSize(tsLogDir, &tsLogSpace.size);
}
if (tsDataDir[0] != 0) {
(void)taosGetDiskSize(tsDataDir, &tsDataSpace.size);
code = taosGetDiskSize(tsDataDir, &tsDataSpace.size);
}
if (tsTempDir[0] != 0) {
(void)taosGetDiskSize(tsTempDir, &tsTempSpace.size);
code = taosGetDiskSize(tsTempDir, &tsTempSpace.size);
}
return code;
}
void osCleanup() {}

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

@ -174,9 +174,12 @@ int64_t taosCopyFile(const char *from, const char *to) {
}
code = taosFsyncFile(pFileTo);
if (code != 0) {
goto _err;
}
(void)taosCloseFile(&pFileFrom);
(void)taosCloseFile(&pFileTo);
TAOS_UNUSED(taosCloseFile(&pFileFrom));
TAOS_UNUSED(taosCloseFile(&pFileTo));
if (code != 0) {
terrno = code;
@ -187,10 +190,10 @@ int64_t taosCopyFile(const char *from, const char *to) {
_err:
if (pFileFrom != NULL) (void)taosCloseFile(&pFileFrom);
if (pFileTo != NULL) (void)taosCloseFile(&pFileTo);
if (pFileFrom != NULL) TAOS_SKIP_ERROR(taosCloseFile(&pFileFrom));
if (pFileTo != NULL) TAOS_SKIP_ERROR(taosCloseFile(&pFileTo));
/* coverity[+retval] */
(void)taosRemoveFile(to);
TAOS_SKIP_ERROR(taosRemoveFile(to));
terrno = code;
return -1;
@ -1120,8 +1123,8 @@ int32_t taosCloseFile(TdFilePtr *ppFile) {
(void)taosThreadRwlockWrlock(&((*ppFile)->rwlock));
#endif
if ((*ppFile)->fp != NULL) {
(void)fflush((*ppFile)->fp);
(void)fclose((*ppFile)->fp);
TAOS_UNUSED(fflush((*ppFile)->fp));
TAOS_UNUSED(fclose((*ppFile)->fp));
(*ppFile)->fp = NULL;
}
#ifdef WINDOWS
@ -1471,21 +1474,25 @@ int32_t taosCompressFile(char *srcFileName, char *destFileName) {
while (!feof(pSrcFile->fp)) {
len = (int32_t)fread(data, 1, compressSize, pSrcFile->fp);
if (len > 0) {
(void)gzwrite(dstFp, data, len);
if(gzwrite(dstFp, data, len) == 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
ret = terrno;
goto cmp_end;
}
}
}
cmp_end:
if (fd >= 0) {
(void)close(fd);
TAOS_SKIP_ERROR(close(fd));
}
if (pSrcFile) {
(void)taosCloseFile(&pSrcFile);
TAOS_SKIP_ERROR(taosCloseFile(&pSrcFile));
}
if (dstFp) {
(void)gzclose(dstFp);
TAOS_SKIP_ERROR(gzclose(dstFp));
}
taosMemoryFree(data);

9
source/os/src/osLocale.c
View File

@ -135,9 +135,14 @@ void taosGetSystemLocale(char *outLocale, char *outCharset) {
str++;
char *revisedCharset = taosCharsetReplace(str);
tstrncpy(outCharset, revisedCharset, TD_CHARSET_LEN);
taosMemoryFree(revisedCharset);
if (NULL == revisedCharset) {
(void)strcpy(outCharset, "UTF-8");
} else {
tstrncpy(outCharset, revisedCharset, TD_CHARSET_LEN);
taosMemoryFree(revisedCharset);
}
// printf("charset not configured, set to system default:%s", outCharset);
} else {
strcpy(outCharset, "UTF-8");

98
source/os/src/osSemaphore.c
View File

@ -48,7 +48,10 @@ int32_t taosGetPId() { return GetCurrentProcessId(); }
int32_t taosGetAppName(char* name, int32_t* len) {
char filepath[1024] = {0};
GetModuleFileName(NULL, filepath, MAX_PATH);
if (GetModuleFileName(NULL, filepath, MAX_PATH) == 0) {
terrno = TAOS_SYSTEM_WINAPI_ERROR(GetLastError());
return terrno;
}
char* sub = strrchr(filepath, '.');
if (sub != NULL) {
*sub = '\0';
@ -70,7 +73,12 @@ int32_t taosGetAppName(char* name, int32_t* len) {
}
int32_t tsem_wait(tsem_t* sem) {
return WaitForSingleObject(*sem, INFINITE);
DWORD ret = WaitForSingleObject(*sem, INFINITE);
if(ret == WAIT_OBJECT_0) {
return 0;
} else {
return TAOS_SYSTEM_WINAPI_ERROR(GetLastError());
}
}
int32_t tsem_timewait(tsem_t* sem, int64_t timeout_ms) {
@ -78,61 +86,65 @@ int32_t tsem_timewait(tsem_t* sem, int64_t timeout_ms) {
if (result == WAIT_OBJECT_0) {
return 0; // Semaphore acquired
} else if (result == WAIT_TIMEOUT) {
return -1; // Timeout reached
return TSDB_CODE_TIMEOUT_ERROR; // Timeout reached
} else {
return result;
return TAOS_SYSTEM_WINAPI_ERROR(GetLastError());
}
}
// Inter-process sharing is not currently supported. The pshared parameter is invalid.
int tsem_init(tsem_t* sem, int pshared, unsigned int value) {
int32_t tsem_init(tsem_t* sem, int pshared, unsigned int value) {
*sem = CreateSemaphore(NULL, value, LONG_MAX, NULL);
return (*sem != NULL) ? 0 : -1;
return (*sem != NULL) ? 0 : TAOS_SYSTEM_WINAPI_ERROR(GetLastError());
}
int tsem_post(tsem_t* sem) {
int32_t tsem_post(tsem_t* sem) {
if (ReleaseSemaphore(*sem, 1, NULL)) return 0;
return -1;
return TAOS_SYSTEM_WINAPI_ERROR(GetLastError());
}
int tsem_destroy(tsem_t* sem) {
int32_t tsem_destroy(tsem_t* sem) {
if (CloseHandle(*sem)) return 0;
return -1;
return TAOS_SYSTEM_WINAPI_ERROR(GetLastError());
}
#elif defined(_TD_DARWIN_64)
#include <libproc.h>
int tsem_init(tsem_t *psem, int flags, unsigned int count) {
int32_t tsem_init(tsem_t *psem, int flags, unsigned int count) {
*psem = dispatch_semaphore_create(count);
if (*psem == NULL) return -1;
if (*psem == NULL) return TAOS_SYSTEM_ERROR(errno);
return 0;
}
int tsem_destroy(tsem_t *psem) {
if (psem == NULL || *psem == NULL) return -1;
int32_t tsem_destroy(tsem_t *psem) {
// if (psem == NULL || *psem == NULL) return -1;
// dispatch_release(*psem);
// *psem = NULL;
return 0;
}
int tsem_post(tsem_t *psem) {
int32_t tsem_post(tsem_t *psem) {
if (psem == NULL || *psem == NULL) return -1;
dispatch_semaphore_signal(*psem);
(void)dispatch_semaphore_signal(*psem);
return 0;
}
int tsem_wait(tsem_t *psem) {
int32_t tsem_wait(tsem_t *psem) {
if (psem == NULL || *psem == NULL) return -1;
dispatch_semaphore_wait(*psem, DISPATCH_TIME_FOREVER);
return 0;
}
int tsem_timewait(tsem_t *psem, int64_t milis) {
int32_t tsem_timewait(tsem_t *psem, int64_t milis) {
if (psem == NULL || *psem == NULL) return -1;
dispatch_time_t time = dispatch_time(DISPATCH_TIME_NOW, (int64_t)(milis * USEC_PER_SEC));
return dispatch_semaphore_wait(*psem, time);
if(dispatch_semaphore_wait(*psem, time) == 0) {
return 0;
} else {
return TSDB_CODE_TIMEOUT_ERROR;
}
}
bool taosCheckPthreadValid(TdThread thread) { return thread != 0; }
@ -216,6 +228,14 @@ int32_t taosGetAppName(char* name, int32_t* len) {
return 0;
}
int32_t tsem_init(tsem_t *psem, int flags, unsigned int count) {
if(sem_init(psem, flags, count) == 0) {
return 0;
} else {
return TAOS_SYSTEM_ERROR(errno);
}
}
int32_t tsem_timewait(tsem_t* sem, int64_t ms) {
int ret = 0;
@ -230,16 +250,18 @@ int32_t tsem_timewait(tsem_t* sem, int64_t ms) {
ts.tv_sec += ts.tv_nsec / 1000000000;
ts.tv_nsec %= 1000000000;
while ((ret = sem_timedwait(sem, &ts)) == -1 && errno == EINTR) {
continue;
while ((ret = sem_timedwait(sem, &ts)) == -1) {
if(errno == EINTR) {
continue;
} else if(errno == ETIMEDOUT) {
return TSDB_CODE_TIMEOUT_ERROR;
} else {
terrno = TAOS_SYSTEM_ERROR(errno);
return terrno;
}
}
if (-1 == ret) {
terrno = TAOS_SYSTEM_ERROR(errno);
return terrno;
}
return ret;
return 0;
}
int32_t tsem_wait(tsem_t* sem) {
@ -285,6 +307,22 @@ int tsem2_init(tsem2_t* sem, int pshared, unsigned int value) {
return 0;
}
int32_t tsem_post(tsem_t* psem) {
if (sem_post(psem) == 0) {
return 0;
} else {
return TAOS_SYSTEM_ERROR(errno);
}
}
int32_t tsem_destroy(tsem_t *sem) {
if (sem_destroy(sem) == 0) {
return 0;
} else {
return TAOS_SYSTEM_ERROR(errno);
}
}
int tsem2_post(tsem2_t *sem) {
int32_t code = taosThreadMutexLock(&sem->mutex);
if (code) {
@ -364,7 +402,11 @@ int32_t tsem2_timewait(tsem2_t* sem, int64_t ms) {
ret = taosThreadCondTimedWait(&sem->cond, &sem->mutex, &ts);
if (ret != 0) {
(void)taosThreadMutexUnlock(&sem->mutex);
return ret;
if (errno == ETIMEDOUT) {
return TSDB_CODE_TIMEOUT_ERROR;
} else {
return TAOS_SYSTEM_ERROR(errno);
}
}
}
}

82
source/os/src/osString.c
View File

@ -38,6 +38,8 @@ char *tstrdup(const char *str) {
}
#ifdef WINDOWS
// No errors are expected to occur
char *strsep(char **stringp, const char *delim) {
char *s;
const char *spanp;
@ -84,9 +86,59 @@ char *stpncpy(char *dest, const char *src, int n) {
}
#endif
int64_t taosStr2int64(const char *str) {
char *endptr = NULL;
return strtoll(str, &endptr, 10);
int32_t taosStr2int64(const char *str, int64_t *val) {
if (str == NULL || val == NULL) {
return TSDB_CODE_INVALID_PARA;
}
char *endptr = NULL;
int64_t ret = strtoll(str, &endptr, 10);
if (errno == ERANGE && (ret == LLONG_MAX || ret == LLONG_MIN)) {
return TAOS_SYSTEM_ERROR(errno);
} else if (errno == EINVAL && ret == 0) {
return TSDB_CODE_INVALID_PARA;
} else {
*val = ret;
return 0;
}
}
int32_t taosStr2int16(const char *str, int16_t *val) {
int64_t tmp = 0;
int32_t code = taosStr2int64(str, &tmp);
if (code) {
return code;
} else if (tmp > INT16_MAX || tmp < INT16_MIN) {
return TAOS_SYSTEM_ERROR(ERANGE);
} else {
*val = (int16_t)tmp;
return 0;
}
}
int32_t taosStr2int32(const char *str, int32_t *val) {
int64_t tmp = 0;
int32_t code = taosStr2int64(str, &tmp);
if (code) {
return code;
} else if (tmp > INT32_MAX || tmp < INT32_MIN) {
return TAOS_SYSTEM_ERROR(ERANGE);
} else {
*val = (int32_t)tmp;
return 0;
}
}
int32_t taosStr2int8(const char *str, int8_t *val) {
int64_t tmp = 0;
int32_t code = taosStr2int64(str, &tmp);
if (code) {
return code;
} else if (tmp > INT8_MAX || tmp < INT8_MIN) {
return TAOS_SYSTEM_ERROR(ERANGE);
} else {
*val = (int8_t)tmp;
return 0;
}
}
int32_t tasoUcs4Compare(TdUcs4 *f1_ucs4, TdUcs4 *f2_ucs4, int32_t bytes) {
@ -253,12 +305,12 @@ void taosReleaseConv(int32_t idx, iconv_t conv, ConvType type) {
bool taosMbsToUcs4(const char *mbs, size_t mbsLength, TdUcs4 *ucs4, int32_t ucs4_max_len, int32_t *len) {
#ifdef DISALLOW_NCHAR_WITHOUT_ICONV
printf("Nchar cannot be read and written without iconv, please install iconv library and recompile.\n");
return -1;
terrno = TSDB_CODE_APP_ERROR;
return false;
#else
(void)memset(ucs4, 0, ucs4_max_len);
int32_t idx = -1;
int32_t code = 0;
iconv_t conv = taosAcquireConv(&idx, M2C);
if ((iconv_t)-1 == conv || (iconv_t)0 == conv) {
return false;
@ -267,9 +319,8 @@ bool taosMbsToUcs4(const char *mbs, size_t mbsLength, TdUcs4 *ucs4, int32_t ucs4
size_t ucs4_input_len = mbsLength;
size_t outLeft = ucs4_max_len;
if (iconv(conv, (char **)&mbs, &ucs4_input_len, (char **)&ucs4, &outLeft) == -1) {
code = TAOS_SYSTEM_ERROR(errno);
terrno = TAOS_SYSTEM_ERROR(errno);
taosReleaseConv(idx, conv, M2C);
terrno = code;
return false;
}
@ -277,6 +328,8 @@ bool taosMbsToUcs4(const char *mbs, size_t mbsLength, TdUcs4 *ucs4, int32_t ucs4
if (len != NULL) {
*len = (int32_t)(ucs4_max_len - outLeft);
if (*len < 0) {
// can not happen
terrno = TSDB_CODE_APP_ERROR;
return false;
}
}
@ -285,17 +338,20 @@ bool taosMbsToUcs4(const char *mbs, size_t mbsLength, TdUcs4 *ucs4, int32_t ucs4
#endif
}
// if success, return the number of bytes written to mbs ( >= 0)
// otherwise return error code ( < 0)
int32_t taosUcs4ToMbs(TdUcs4 *ucs4, int32_t ucs4_max_len, char *mbs) {
#ifdef DISALLOW_NCHAR_WITHOUT_ICONV
printf("Nchar cannot be read and written without iconv, please install iconv library and recompile.\n");
return -1;
terrno = TSDB_CODE_APP_ERROR;
return terrno;
#else
int32_t idx = -1;
int32_t code = 0;
iconv_t conv = taosAcquireConv(&idx, C2M);
if ((iconv_t)-1 == conv || (iconv_t)0 == conv) {
return false;
return TSDB_CODE_APP_ERROR;
}
size_t ucs4_input_len = ucs4_max_len;
@ -313,10 +369,13 @@ int32_t taosUcs4ToMbs(TdUcs4 *ucs4, int32_t ucs4_max_len, char *mbs) {
#endif
}
// if success, return the number of bytes written to mbs ( >= 0)
// otherwise return error code ( < 0)
int32_t taosUcs4ToMbsEx(TdUcs4 *ucs4, int32_t ucs4_max_len, char *mbs, iconv_t conv) {
#ifdef DISALLOW_NCHAR_WITHOUT_ICONV
printf("Nchar cannot be read and written without iconv, please install iconv library and recompile.\n");
return -1;
terrno = TSDB_CODE_APP_ERROR;
return terrno;
#else
size_t ucs4_input_len = ucs4_max_len;
@ -333,7 +392,8 @@ int32_t taosUcs4ToMbsEx(TdUcs4 *ucs4, int32_t ucs4_max_len, char *mbs, iconv_t c
bool taosValidateEncodec(const char *encodec) {
#ifdef DISALLOW_NCHAR_WITHOUT_ICONV
printf("Nchar cannot be read and written without iconv, please install iconv library and recompile.\n");
return true;
terrno = TSDB_CODE_APP_ERROR;
return false;
#else
iconv_t cd = iconv_open(encodec, DEFAULT_UNICODE_ENCODEC);
if (cd == (iconv_t)(-1)) {

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

@ -179,10 +179,8 @@ static int32_t taosGetSysCpuInfo(SysCpuInfo *cpuInfo) {
char line[1024];
ssize_t bytes = taosGetsFile(pFile, sizeof(line), line);
if (bytes < 0) {
code = terrno;
(void)taosCloseFile(&pFile);
terrno = code;
return code;
TAOS_SKIP_ERROR(taosCloseFile(&pFile));
return terrno;
}
char cpu[10] = {0};
@ -230,9 +228,7 @@ static int32_t taosGetProcCpuInfo(ProcCpuInfo *cpuInfo) {
char line[1024] = {0};
ssize_t bytes = taosGetsFile(pFile, sizeof(line), line);
if (bytes < 0) {
code = terrno;
(void)taosCloseFile(&pFile);
terrno = code;
TAOS_SKIP_ERROR(taosCloseFile(&pFile));
return code;
}
@ -250,7 +246,7 @@ static int32_t taosGetProcCpuInfo(ProcCpuInfo *cpuInfo) {
}
}
(void)taosCloseFile(&pFile);
TAOS_SKIP_ERROR(taosCloseFile(&pFile));
#endif
return 0;
@ -427,7 +423,7 @@ int32_t taosGetOsReleaseName(char *releaseName, char* sName, char* ver, int32_t
if (++cnt >= 3) break;
}
(void)taosCloseFile(&pFile);
TAOS_SKIP_ERROR(taosCloseFile(&pFile));
return code;
#endif
}
@ -497,13 +493,13 @@ int32_t taosGetCpuInfo(char *cpuModel, int32_t maxLen, float *numOfCores) {
if (strncmp(line, "processor", 9) == 0) coreCount += 1;
}
(void)taosCloseFile(&pFile);
TAOS_SKIP_ERROR(taosCloseFile(&pFile));
if (code != 0 && (done & 1) == 0) {
TdFilePtr pFile1 = taosOpenFile("/proc/device-tree/model", TD_FILE_READ | TD_FILE_STREAM);
if (pFile1 != NULL) {
ssize_t bytes = taosGetsFile(pFile1, maxLen, cpuModel);
(void)taosCloseFile(&pFile);
TAOS_SKIP_ERROR(taosCloseFile(&pFile));
if (bytes > 0) {
code = 0;
done |= 1;
@ -535,9 +531,9 @@ int32_t taosGetCpuInfo(char *cpuModel, int32_t maxLen, float *numOfCores) {
// Returns the container's CPU quota if successful, otherwise returns the physical CPU cores
static int32_t taosCntrGetCpuCores(float *numOfCores) {
#ifdef WINDOWS
return -1;
return TSDB_CODE_UNSUPPORT_OS;
#elif defined(_TD_DARWIN_64)
return -1;
return TSDB_CODE_UNSUPPORT_OS;
#else
TdFilePtr pFile = NULL;
if (!(pFile = taosOpenFile(tsCpuQuotaFile, TD_FILE_READ | TD_FILE_STREAM))) {
@ -545,11 +541,11 @@ static int32_t taosCntrGetCpuCores(float *numOfCores) {
}
char qline[32] = {0};
if (taosGetsFile(pFile, sizeof(qline), qline) <= 0) {
(void)taosCloseFile(&pFile);
TAOS_SKIP_ERROR(taosCloseFile(&pFile));
goto _sys;
}
(void)taosCloseFile(&pFile);
TAOS_SKIP_ERROR(taosCloseFile(&pFile));
float quota = taosStr2Float(qline, NULL);
if (quota < 0) {
goto _sys;
@ -561,11 +557,11 @@ static int32_t taosCntrGetCpuCores(float *numOfCores) {
char pline[32] = {0};
if (taosGetsFile(pFile, sizeof(pline), pline) <= 0) {
(void)taosCloseFile(&pFile);
TAOS_SKIP_ERROR(taosCloseFile(&pFile));
goto _sys;
}
(void)taosCloseFile(&pFile);
TAOS_SKIP_ERROR(taosCloseFile(&pFile));
float period = taosStr2Float(pline, NULL);
float quotaCores = quota / period;
@ -575,10 +571,16 @@ static int32_t taosCntrGetCpuCores(float *numOfCores) {
} else {
*numOfCores = sysCores;
}
if(*numOfCores <= 0) {
return TAOS_SYSTEM_ERROR(errno);
}
goto _end;
_sys:
*numOfCores = sysconf(_SC_NPROCESSORS_ONLN);
if(*numOfCores <= 0) {
return TAOS_SYSTEM_ERROR(errno);
}
_end:
return 0;
@ -586,26 +588,35 @@ _end:
#endif
}
void taosGetCpuCores(float *numOfCores, bool physical) {
int32_t taosGetCpuCores(float *numOfCores, bool physical) {
#ifdef WINDOWS
SYSTEM_INFO info;
GetSystemInfo(&info);
*numOfCores = info.dwNumberOfProcessors;
return;
return 0;
#elif defined(_TD_DARWIN_64)
*numOfCores = sysconf(_SC_NPROCESSORS_ONLN);
return;
if(*numOfCores <= 0) {
return TAOS_SYSTEM_ERROR(errno);
}
return 0;
#else
if (physical) {
*numOfCores = sysconf(_SC_NPROCESSORS_ONLN);
if(*numOfCores <= 0) {
return TAOS_SYSTEM_ERROR(errno);
}
} else {
(void)taosCntrGetCpuCores(numOfCores);
int code= taosCntrGetCpuCores(numOfCores);
if(code != 0) {
return code;
}
}
return;
return 0;
#endif
}
void taosGetCpuUsage(double *cpu_system, double *cpu_engine) {
int32_t taosGetCpuUsage(double *cpu_system, double *cpu_engine) {
static int64_t lastSysUsed = -1;
static int64_t lastSysTotal = -1;
static int64_t lastProcTotal = -1;
@ -639,6 +650,7 @@ void taosGetCpuUsage(double *cpu_system, double *cpu_engine) {
lastSysTotal = curSysTotal;
lastProcTotal = curProcTotal;
}
return 0;
}
#define __cpuid_fix(level, a, b, c, d) \
@ -779,15 +791,15 @@ int32_t taosGetDiskSize(char *dataDir, SDiskSize *diskSize) {
return 0;
} else {
// printf("failed to get disk size, dataDir:%s errno:%s", tsDataDir, strerror(errno));
// terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
terrno = TAOS_SYSTEM_WINAPI_ERROR(GetLastError());
return terrno;
}
#elif defined(_TD_DARWIN_64)
struct statvfs info;
if (statvfs(dataDir, &info)) {
// printf("failed to get disk size, dataDir:%s errno:%s", tsDataDir, strerror(errno));
// terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
terrno = TAOS_SYSTEM_ERROR(errno);
return terrno;
} else {
diskSize->total = info.f_blocks * info.f_frsize;
diskSize->avail = info.f_bavail * info.f_frsize;
@ -860,7 +872,7 @@ int32_t taosGetProcIO(int64_t *rchars, int64_t *wchars, int64_t *read_bytes, int
if (readIndex >= 4) break;
}
(void)taosCloseFile(&pFile);
TAOS_SKIP_ERROR(taosCloseFile(&pFile));
if (readIndex < 4) {
return -1;
@ -870,7 +882,7 @@ int32_t taosGetProcIO(int64_t *rchars, int64_t *wchars, int64_t *read_bytes, int
#endif
}
void taosGetProcIODelta(int64_t *rchars, int64_t *wchars, int64_t *read_bytes, int64_t *write_bytes) {
int32_t taosGetProcIODelta(int64_t *rchars, int64_t *wchars, int64_t *read_bytes, int64_t *write_bytes) {
static int64_t last_rchars = -1;
static int64_t last_wchars = -1;
static int64_t last_read_bytes = -1;
@ -879,7 +891,8 @@ void taosGetProcIODelta(int64_t *rchars, int64_t *wchars, int64_t *read_bytes, i
static int64_t cur_wchars = 0;
static int64_t cur_read_bytes = 0;
static int64_t cur_write_bytes = 0;
if (taosGetProcIO(&cur_rchars, &cur_wchars, &cur_read_bytes, &cur_write_bytes) == 0) {
int32_t code = taosGetProcIO(&cur_rchars, &cur_wchars, &cur_read_bytes, &cur_write_bytes);
if (code == 0) {
if(last_rchars >=0 && last_wchars >=0 && last_read_bytes >=0 && last_write_bytes >= 0){
*rchars = cur_rchars - last_rchars;
*wchars = cur_wchars - last_wchars;
@ -897,11 +910,15 @@ void taosGetProcIODelta(int64_t *rchars, int64_t *wchars, int64_t *read_bytes, i
last_read_bytes = cur_read_bytes;
last_write_bytes = cur_write_bytes;
} else {
*rchars = 0;
*wchars = 0;
*read_bytes = 0;
*write_bytes = 0;
return code;
}
return 0;
}
void taosSetDefaultProcIODelta(int64_t *rchars, int64_t *wchars, int64_t *read_bytes, int64_t *write_bytes) {
*rchars = 0;
*wchars = 0;
*read_bytes = 0;
*write_bytes = 0;
}
int32_t taosGetCardInfo(int64_t *receive_bytes, int64_t *transmit_bytes) {
@ -953,18 +970,19 @@ int32_t taosGetCardInfo(int64_t *receive_bytes, int64_t *transmit_bytes) {
*transmit_bytes += o_tbytes;
}
(void)taosCloseFile(&pFile);
TAOS_SKIP_ERROR(taosCloseFile(&pFile));
return 0;
#endif
}
void taosGetCardInfoDelta(int64_t *receive_bytes, int64_t *transmit_bytes) {
int32_t taosGetCardInfoDelta(int64_t *receive_bytes, int64_t *transmit_bytes) {
static int64_t last_receive_bytes = -1;
static int64_t last_transmit_bytes = -1;
int64_t cur_receive_bytes = 0;
int64_t cur_transmit_bytes = 0;
if (taosGetCardInfo(&cur_receive_bytes, &cur_transmit_bytes) == 0) {
int32_t code = taosGetCardInfo(&cur_receive_bytes, &cur_transmit_bytes);
if (code == 0) {
if(last_receive_bytes >= 0 && last_transmit_bytes >= 0){
*receive_bytes = cur_receive_bytes - last_receive_bytes;
*transmit_bytes = cur_transmit_bytes - last_transmit_bytes;
@ -977,9 +995,13 @@ void taosGetCardInfoDelta(int64_t *receive_bytes, int64_t *transmit_bytes) {
last_receive_bytes = cur_receive_bytes;
last_transmit_bytes = cur_transmit_bytes;
} else {
*receive_bytes = 0;
*transmit_bytes = 0;
return code;
}
return 0;
}
void taosSetDefaultCardInfoDelta(int64_t *receive_bytes, int64_t *transmit_bytes) {
*receive_bytes = 0;
*transmit_bytes = 0;
}
void taosKillSystem() {

16
source/os/test/osSemaphoreTests.cpp
View File

@ -59,18 +59,26 @@ TEST(osSemaphoreTests, Destroy) {
TEST(osSemaphoreTests, WaitTime0) {
tsem_t sem;
(void)tsem_init(&sem, 0, 0);
EXPECT_NE(tsem_timewait(&sem, 1000), 0);
EXPECT_EQ(tsem_timewait(&sem, 1000), TSDB_CODE_TIMEOUT_ERROR);
(void)tsem_destroy(&sem);
}
TEST(osSemaphoreTests, WaitTime1) {
tsem_t sem;
(void)tsem_init(&sem, 0, 1);
EXPECT_EQ(tsem_timewait(&sem, 1000), 0);
EXPECT_NE(tsem_timewait(&sem, 1000), 0);
EXPECT_EQ(tsem_timewait(&sem, 10000), 0);
EXPECT_EQ(tsem_timewait(&sem, 1000), TSDB_CODE_TIMEOUT_ERROR);
(void)tsem_destroy(&sem);
}
TEST(osSemaphoreTests, WaitTime2) {
tsem2_t sem;
(void)tsem2_init(&sem, 0, 1);
EXPECT_EQ(tsem2_timewait(&sem, 10000), 0);
EXPECT_EQ(tsem2_timewait(&sem, 1000), TSDB_CODE_TIMEOUT_ERROR);
(void)tsem2_destroy(&sem);
}
TEST(osSemaphoreTests, WaitAndPost) {
tsem_t sem;
int result = tsem_init(&sem, 0, 0);
@ -217,7 +225,7 @@ TEST(osSemaphoreTests, Performance4_1) {
(void)tsem_post(&sem);
}).detach();
(void)tsem_timewait(&sem, 1000);
EXPECT_EQ(tsem_timewait(&sem, 1000),0);
(void)tsem_destroy(&sem);
}

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

@ -215,7 +215,7 @@ int32_t taosInitSlowLog() {
int32_t taosInitLog(const char *logName, int32_t maxFiles, bool tsc) {
if (atomic_val_compare_exchange_8(&tsLogInited, 0, 1) != 0) return 0;
osUpdate();
TAOS_CHECK_RETURN(osUpdate());
TAOS_CHECK_RETURN(taosInitNormalLog(logName, maxFiles));
if (tsc){
@ -908,7 +908,7 @@ static void *taosAsyncOutputLog(void *param) {
updateCron++;
taosMsleep(writeInterval);
if (count > 1000) {
osUpdate();
TAOS_UNUSED(osUpdate());
count = 0;
}

8
source/util/src/tsched.c
View File

@ -150,7 +150,7 @@ void *taosProcessSchedQueue(void *scheduler) {
while (1) {
if ((ret = tsem_wait(&pSched->fullSem)) != 0) {
uFatal("wait %s fullSem failed(%s)", pSched->label, strerror(errno));
uFatal("wait %s fullSem failed(%s)", pSched->label, strerror(terrno));
}
if (atomic_load_8(&pSched->stop)) {
break;
@ -169,7 +169,7 @@ void *taosProcessSchedQueue(void *scheduler) {
}
if ((ret = tsem_post(&pSched->emptySem)) != 0) {
uFatal("post %s emptySem failed(%s)", pSched->label, strerror(errno));
uFatal("post %s emptySem failed(%s)", pSched->label, strerror(terrno));
}
if (msg.fp)
@ -197,7 +197,7 @@ int taosScheduleTask(void *queueScheduler, SSchedMsg *pMsg) {
}
if ((ret = tsem_wait(&pSched->emptySem)) != 0) {
uFatal("wait %s emptySem failed(%s)", pSched->label, strerror(errno));
uFatal("wait %s emptySem failed(%s)", pSched->label, strerror(terrno));
}
if ((ret = taosThreadMutexLock(&pSched->queueMutex)) != 0) {
@ -212,7 +212,7 @@ int taosScheduleTask(void *queueScheduler, SSchedMsg *pMsg) {
}
if ((ret = tsem_post(&pSched->fullSem)) != 0) {
uFatal("post %s fullSem failed(%s)", pSched->label, strerror(errno));
uFatal("post %s fullSem failed(%s)", pSched->label, strerror(terrno));
}
return ret;
}

16
tests/army/query/last/test_last.py
View File

@ -174,6 +174,19 @@ class TDTestCase(TBase):
tdSql.checkData(0, 2, None)
tdLog.info("Finish test_last_with_primarykey_str_rt")
def test_ts5389(self):
"""add test case to cover the crash issue of ts-5389
"""
tdSql.execute("create database db_ts5389;")
tdSql.execute("use db_ts5389;")
tdSql.execute("create stable trackers(ts timestamp, reg_firmware_rev double) tags(site nchar(8), tracker nchar(16), zone nchar(2));")
tdSql.execute("create table tr1 using trackers tags ('MI-01', 'N29-26', '12');")
tdSql.execute("create table tr2 using trackers tags ('MI-01', 'N29-6', '11');")
tdSql.execute("insert into tr1 values(now,null);")
tdSql.execute("insert into tr2 values(now,null);")
tdSql.query("select distinct site,zone,tracker,last(reg_firmware_rev) from trackers where ts > now() -1h and site='MI-01' partition by site;")
tdSql.checkRows(1)
def run(self):
self.prepare_data()
# regular table
@ -182,9 +195,12 @@ class TDTestCase(TBase):
# child tables
self.test_last_with_primarykey_int_ct()
self.test_last_with_primarykey_str_ct()
# ts-5389
self.test_ts5389()
def stop(self):
tdSql.execute("drop database db_td30816;")
tdSql.execute("drop database db_ts5389;")
tdSql.close()
tdLog.success("%s successfully executed" % __file__)

12
tests/script/api/stmt2-nohole.c
View File

@ -14,12 +14,12 @@ int64_t genReqid() {
return count;
}
sem_t sem;
// sem_t sem;
void stmtAsyncQueryCb(void* param, TAOS_RES* pRes, int code) {
int affected_rows = taos_affected_rows(pRes);
printf("\033[31maffected rows:%d\033[0m\n", affected_rows);
(void)sem_post(&sem);
//(void)sem_post(&sem);
return;
/*
SSP_CB_PARAM* qParam = (SSP_CB_PARAM*)param;
@ -319,7 +319,7 @@ _bind_again:
taos_stmt2_free_fields(stmt, fields);
*/
// if (taos_stmt_execute(stmt) != 0) {
(void)sem_init(&sem, 0, 0);
//(void)sem_init(&sem, 0, 0);
start = clock();
// if (taos_stmt2_exec(stmt, NULL, stmtAsyncQueryCb, NULL) != 0) {
if (taos_stmt2_exec(stmt, NULL) != 0) {
@ -330,9 +330,9 @@ _bind_again:
end = clock();
printf("exec time:%f\n", (double)(end - start) / CLOCKS_PER_SEC);
sem_wait(&sem);
(void)sem_destroy(&sem);
if (++run_time < 2) {
// sem_wait(&sem);
//(void)sem_destroy(&sem);
if (++run_time < 20) {
goto _bind_again;
}
taos_stmt2_close(stmt);