From ba7acaa5bcde42a098441a586c0a392ae431bf2c Mon Sep 17 00:00:00 2001 From: yifan hao Date: Fri, 8 May 2020 18:52:17 -0600 Subject: [PATCH 1/9] [Timer] Initialize the last tmrCtrls. While initialize timer module, tmrCtrls is constructed to be a free list. However, the last tmrCtrls is never initialized, which is allocated from malloc(). Since the code relies on tmrCtrls->next to tell if there's any free tmrCtrls left, the last tmrCtrls should be properly linked to NULL. --- src/util/src/ttimer.c | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/src/util/src/ttimer.c b/src/util/src/ttimer.c index 42fd13b2cd..e6ef73ef57 100644 --- a/src/util/src/ttimer.c +++ b/src/util/src/ttimer.c @@ -405,19 +405,19 @@ static bool doStopTimer(tmr_obj_t* timer, uint8_t state) { tmrTrace(fmt, timer->ctrl->label, timer->id, timer->fp, timer->param); return reusable; } - + if (state != TIMER_STATE_EXPIRED) { // timer already stopped or cancelled, has nothing to do in this case return false; } - + if (timer->executedBy == taosGetPthreadId()) { // taosTmrReset is called in the timer callback, should do nothing in this // case to avoid dead lock. note taosTmrReset must be the last statement // of the callback funtion, will be a bug otherwise. return false; } - + // timer callback is executing in another thread, we SHOULD wait it to stop, // BUT this may result in dead lock if current thread are holding a lock which // the timer callback need to acquire. so, we HAVE TO return directly. @@ -501,6 +501,7 @@ static void taosTmrModuleInit(void) { tmr_ctrl_t* ctrl = tmrCtrls + i; ctrl->next = ctrl + 1; } + (tmrCtrls + taosMaxTmrCtrl - 1)->next = NULL; unusedTmrCtrl = tmrCtrls; pthread_mutex_init(&tmrCtrlMutex, NULL); @@ -574,12 +575,12 @@ void taosTmrCleanUp(void* handle) { if (numOfTmrCtrl <=0) { taosUninitTimer(); - + taosCleanUpScheduler(tmrQhandle); for (int i = 0; i < tListLen(wheels); i++) { time_wheel_t* wheel = wheels + i; - pthread_mutex_destroy(&wheel->mutex); + pthread_mutex_destroy(&wheel->mutex); free(wheel->slots); } From 8bf4ef21fac840d56ef0e51f383ab3ae7bb556ca Mon Sep 17 00:00:00 2001 From: Bomin Zhang Date: Sat, 23 May 2020 15:00:22 +0800 Subject: [PATCH 2/9] fix td-317: max table name length now is 392 --- src/common/src/tname.c | 1 + src/mnode/src/mgmtTable.c | 4 ++-- tests/pytest/table/boundary.py | 37 ++++++++++++++++++++++++++++------ tests/pytest/util/sql.py | 3 +++ 4 files changed, 37 insertions(+), 8 deletions(-) diff --git a/src/common/src/tname.c b/src/common/src/tname.c index 29236ed0ff..3566f26abd 100644 --- a/src/common/src/tname.c +++ b/src/common/src/tname.c @@ -29,6 +29,7 @@ void extractTableName(const char* tableId, char* name) { size_t s2 = strcspn(&tableId[s1 + 1], &TS_PATH_DELIMITER[0]); strncpy(name, &tableId[s1 + s2 + 2], TSDB_TABLE_NAME_LEN); + name[TSDB_TABLE_NAME_LEN] = 0; } char* extractDBName(const char* tableId, char* name) { diff --git a/src/mnode/src/mgmtTable.c b/src/mnode/src/mgmtTable.c index f0c55449a3..87243d25c0 100644 --- a/src/mnode/src/mgmtTable.c +++ b/src/mnode/src/mgmtTable.c @@ -1139,7 +1139,7 @@ int32_t mgmtRetrieveShowSuperTables(SShowObj *pShow, char *data, int32_t rows, v prefixLen = strlen(prefix); SPatternCompareInfo info = PATTERN_COMPARE_INFO_INITIALIZER; - char stableName[TSDB_TABLE_NAME_LEN] = {0}; + char stableName[TSDB_TABLE_NAME_LEN + 1] = {0}; while (numOfRows < rows) { pShow->pIter = mgmtGetNextSuperTable(pShow->pIter, &pTable); @@ -2148,7 +2148,7 @@ static int32_t mgmtRetrieveShowTables(SShowObj *pShow, char *data, int32_t rows, continue; } - char tableName[TSDB_TABLE_NAME_LEN] = {0}; + char tableName[TSDB_TABLE_NAME_LEN + 1] = {0}; // pattern compare for table name mgmtExtractTableName(pTable->info.tableId, tableName); diff --git a/tests/pytest/table/boundary.py b/tests/pytest/table/boundary.py index b68671c61a..29fdd5c475 100644 --- a/tests/pytest/table/boundary.py +++ b/tests/pytest/table/boundary.py @@ -10,7 +10,7 @@ from util.sql import * class TDTestCase: - def init(self, conn): + def init(self, conn, logSql): tdLog.debug("start to execute %s" % __file__) tdSql.init(conn.cursor(), logSql) @@ -95,17 +95,42 @@ class TDTestCase: maxTableNameLen = self.getLimitFromSourceCode('TSDB_TABLE_NAME_LEN') tdLog.notice("table name max length is %d" % maxTableNameLen) - name = self.generateString(maxTableNameLen - 1) - tdLog.info("table name is '%s'" % name) + # create a super table with name exceed max length + sname = self.generateString(maxTableNameLen + 1) + tdLog.info("create a super table with length %d" % len(sname)) + tdSql.error("create table %s (ts timestamp, value int) tags(id int)" % sname) - tdSql.execute("create table %s (ts timestamp, value int)" % name) - tdSql.execute("insert into %s values(now, 0)" % name) + # create a super table with name of max length + sname = self.generateString(maxTableNameLen) + tdLog.info("create a super table with length %d" % len(sname)) + tdSql.execute("create table %s (ts timestamp, value int) tags(id int)" % sname) + tdLog.info("check table count, should be one") + tdSql.query('show stables') + tdSql.checkRows(1) + # create a child table with name exceed max length + name = self.generateString(maxTableNameLen + 1) + tdLog.info("create a child table with length %d" % len(name)) + tdSql.error("create table %s using %s tags(0)" % (name, sname)) + + # create a child table with name of max length + name = self.generateString(maxTableNameLen) + tdLog.info("create a child table with length %d" % len(name)) + tdSql.execute("create table %s using %s tags(0)" % (name, sname)) tdSql.query('show tables') tdSql.checkRows(1) - tdSql.query('select * from %s' % name) + # insert one row + tdLog.info("insert one row of data") + tdSql.execute("insert into %s values(now, 0)" % name) + tdSql.query("select * from " + name) tdSql.checkRows(1) + tdSql.query("select * from " + sname) + tdSql.checkRows(1) + + name = name[:len(name) - 1] + tdSql.error("select * from " + name) + tdSql.checkRows(0) def checkRowBoundaries(self): tdLog.debug("checking row boundaries") diff --git a/tests/pytest/util/sql.py b/tests/pytest/util/sql.py index 245e4b0945..ec7ac117c0 100644 --- a/tests/pytest/util/sql.py +++ b/tests/pytest/util/sql.py @@ -58,6 +58,9 @@ class TDSql: "%s failed: sql:%s, expect error not occured" % (callerFilename, sql)) else: + self.queryRows = 0 + self.queryCols = 0 + self.queryResult = None tdLog.info("sql:%s, expect error occured" % (sql)) def query(self, sql): From 2a4fecebe647936b07d2eb778e5231148a1a5885 Mon Sep 17 00:00:00 2001 From: Jeff Tao Date: Sat, 23 May 2020 13:39:44 +0000 Subject: [PATCH 3/9] tune the code to make the vnode cleanup process more clear --- src/cq/src/cqMain.c | 9 ++- src/inc/tsync.h | 4 +- src/util/src/tqueue.c | 2 + src/vnode/src/vnodeMain.c | 163 +++++++++++++++++++++++--------------- src/wal/src/walMain.c | 12 ++- 5 files changed, 121 insertions(+), 69 deletions(-) diff --git a/src/cq/src/cqMain.c b/src/cq/src/cqMain.c index 7935bb7ff5..5cc3ce0159 100644 --- a/src/cq/src/cqMain.c +++ b/src/cq/src/cqMain.c @@ -18,6 +18,7 @@ #include #include #include +#include #include "taosdef.h" #include "taosmsg.h" #include "tglobal.h" @@ -64,7 +65,10 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj); void *cqOpen(void *ahandle, const SCqCfg *pCfg) { SCqContext *pContext = calloc(sizeof(SCqContext), 1); - if (pContext == NULL) return NULL; + if (pContext == NULL) { + terrno = TAOS_SYSTEM_ERROR(errno); + return NULL; + } strcpy(pContext->user, pCfg->user); strcpy(pContext->pass, pCfg->pass); @@ -82,6 +86,7 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) { void cqClose(void *handle) { SCqContext *pContext = handle; + if (handle == NULL) return; // stop all CQs cqStop(pContext); @@ -106,9 +111,9 @@ void cqClose(void *handle) { void cqStart(void *handle) { SCqContext *pContext = handle; - cTrace("vgId:%d, start all CQs", pContext->vgId); if (pContext->dbConn || pContext->master) return; + cTrace("vgId:%d, start all CQs", pContext->vgId); pthread_mutex_lock(&pContext->mutex); pContext->master = 1; diff --git a/src/inc/tsync.h b/src/inc/tsync.h index 05d1d93cf6..fcf26d22c3 100644 --- a/src/inc/tsync.h +++ b/src/inc/tsync.h @@ -94,8 +94,8 @@ typedef void* tsync_h; tsync_h syncStart(const SSyncInfo *); void syncStop(tsync_h shandle); -int syncReconfig(tsync_h shandle, const SSyncCfg *); -int syncForwardToPeer(tsync_h shandle, void *pHead, void *mhandle, int qtype); +int32_t syncReconfig(tsync_h shandle, const SSyncCfg *); +int32_t syncForwardToPeer(tsync_h shandle, void *pHead, void *mhandle, int qtype); void syncConfirmForward(tsync_h shandle, uint64_t version, int32_t code); void syncRecover(tsync_h shandle); // recover from other nodes: int syncGetNodesRole(tsync_h shandle, SNodesRole *); diff --git a/src/util/src/tqueue.c b/src/util/src/tqueue.c index 1e248c9e45..475941dbdb 100644 --- a/src/util/src/tqueue.c +++ b/src/util/src/tqueue.c @@ -65,6 +65,7 @@ taos_queue taosOpenQueue() { } void taosCloseQueue(taos_queue param) { + if (param == NULL) return; STaosQueue *queue = (STaosQueue *)param; STaosQnode *pTemp; STaosQnode *pNode = queue->head; @@ -224,6 +225,7 @@ taos_qset taosOpenQset() { } void taosCloseQset(taos_qset param) { + if (param == NULL) return; STaosQset *qset = (STaosQset *)param; pthread_mutex_destroy(&qset->mutex); tsem_destroy(&qset->sem); diff --git a/src/vnode/src/vnodeMain.c b/src/vnode/src/vnodeMain.c index 6f0b19b0c6..926eee6037 100644 --- a/src/vnode/src/vnodeMain.c +++ b/src/vnode/src/vnodeMain.c @@ -35,7 +35,7 @@ static void vnodeCleanUp(SVnodeObj *pVnode); static int32_t vnodeSaveCfg(SMDCreateVnodeMsg *pVnodeCfg); static int32_t vnodeReadCfg(SVnodeObj *pVnode); static int32_t vnodeSaveVersion(SVnodeObj *pVnode); -static bool vnodeReadVersion(SVnodeObj *pVnode); +static int32_t vnodeReadVersion(SVnodeObj *pVnode); static int vnodeProcessTsdbStatus(void *arg, int status); static uint32_t vnodeGetFileInfo(void *ahandle, char *name, uint32_t *index, int32_t *size, uint64_t *fversion); static int vnodeGetWalInfo(void *ahandle, char *name, uint32_t *index); @@ -46,9 +46,9 @@ static pthread_once_t vnodeModuleInit = PTHREAD_ONCE_INIT; #ifndef _SYNC tsync_h syncStart(const SSyncInfo *info) { return NULL; } -int syncForwardToPeer(tsync_h shandle, void *pHead, void *mhandle, int qtype) { return 0; } +int32_t syncForwardToPeer(tsync_h shandle, void *pHead, void *mhandle, int qtype) { return 0; } void syncStop(tsync_h shandle) {} -int syncReconfig(tsync_h shandle, const SSyncCfg * cfg) { return 0; } +int32_t syncReconfig(tsync_h shandle, const SSyncCfg * cfg) { return 0; } int syncGetNodesRole(tsync_h shandle, SNodesRole * cfg) { return 0; } void syncConfirmForward(tsync_h shandle, uint64_t version, int32_t code) {} #endif @@ -185,26 +185,40 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) { pthread_once(&vnodeModuleInit, vnodeInit); SVnodeObj *pVnode = calloc(sizeof(SVnodeObj), 1); + if (pVnode == NULL) { + vError("vgId:%d, failed to open vnode since no enough memory", vnode); + return TAOS_SYSTEM_ERROR(errno); + } + + atomic_add_fetch_32(&tsOpennedVnodes, 1); + atomic_add_fetch_32(&pVnode->refCount, 1); + pVnode->vgId = vnode; pVnode->status = TAOS_VN_STATUS_INIT; - pVnode->refCount = 1; pVnode->version = 0; pVnode->tsdbCfg.tsdbId = pVnode->vgId; pVnode->rootDir = strdup(rootDir); - taosHashPut(tsDnodeVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t), (char *)(&pVnode), sizeof(SVnodeObj *)); int32_t code = vnodeReadCfg(pVnode); if (code != TSDB_CODE_SUCCESS) { - vError("vgId:%d, failed to read cfg file", pVnode->vgId); - taosHashRemove(tsDnodeVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t)); + vnodeCleanUp(pVnode); + return code; + } + + code = vnodeReadVersion(pVnode); + if (code != TSDB_CODE_SUCCESS) { + vnodeCleanUp(pVnode); return code; } - vnodeReadVersion(pVnode); pVnode->fversion = pVnode->version; pVnode->wqueue = dnodeAllocateWqueue(pVnode); pVnode->rqueue = dnodeAllocateRqueue(pVnode); + if (pVnode->wqueue == NULL || pVnode->rqueue == NULL) { + vnodeCleanUp(pVnode); + return terrno; + } SCqCfg cqCfg = {0}; sprintf(cqCfg.user, "root"); @@ -212,22 +226,29 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) { cqCfg.vgId = vnode; cqCfg.cqWrite = vnodeWriteToQueue; pVnode->cq = cqOpen(pVnode, &cqCfg); + if (pVnode->cq == NULL) { + vnodeCleanUp(pVnode); + return terrno; + } STsdbAppH appH = {0}; appH.appH = (void *)pVnode; appH.notifyStatus = vnodeProcessTsdbStatus; appH.cqH = pVnode->cq; - sprintf(temp, "%s/tsdb", rootDir); pVnode->tsdb = tsdbOpenRepo(temp, &appH); if (pVnode->tsdb == NULL) { - vError("vgId:%d, failed to open tsdb at %s(%s)", pVnode->vgId, temp, tstrerror(terrno)); - taosHashRemove(tsDnodeVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t)); + vnodeCleanUp(pVnode); return terrno; } sprintf(temp, "%s/wal", rootDir); pVnode->wal = walOpen(temp, &pVnode->walCfg); + if (pVnode->wal == NULL) { + vnodeCleanUp(pVnode); + return terrno; + } + walRestore(pVnode->wal, pVnode, vnodeWriteToQueue); SSyncInfo syncInfo; @@ -243,6 +264,10 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) { syncInfo.notifyRole = vnodeNotifyRole; syncInfo.notifyFileSynced = vnodeNotifyFileSynced; pVnode->sync = syncStart(&syncInfo); + if (pVnode->sync == NULL) { + vnodeCleanUp(pVnode); + return terrno; + } #ifndef _SYNC pVnode->role = TAOS_SYNC_ROLE_MASTER; @@ -253,11 +278,11 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) { cqStart(pVnode->cq); pVnode->events = NULL; - pVnode->status = TAOS_VN_STATUS_READY; vTrace("vgId:%d, vnode is opened in %s, pVnode:%p", pVnode->vgId, rootDir, pVnode); - atomic_add_fetch_32(&tsOpennedVnodes, 1); + taosHashPut(tsDnodeVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t), (char *)(&pVnode), sizeof(SVnodeObj *)); + return TSDB_CODE_SUCCESS; } @@ -286,13 +311,6 @@ void vnodeRelease(void *pVnodeRaw) { } tfree(pVnode->rootDir); - // remove read queue - dnodeFreeRqueue(pVnode->rqueue); - pVnode->rqueue = NULL; - - // remove write queue - dnodeFreeWqueue(pVnode->wqueue); - pVnode->wqueue = NULL; if (pVnode->status == TAOS_VN_STATUS_DELETING) { char rootDir[TSDB_FILENAME_LEN] = {0}; @@ -387,15 +405,26 @@ static void vnodeCleanUp(SVnodeObj *pVnode) { pVnode->sync = NULL; } - cqClose(pVnode->cq); - pVnode->cq = NULL; - - tsdbCloseRepo(pVnode->tsdb, 1); - pVnode->tsdb = NULL; - - walClose(pVnode->wal); + if (pVnode->wal) + walClose(pVnode->wal); pVnode->wal = NULL; + if (pVnode->tsdb) + tsdbCloseRepo(pVnode->tsdb, 1); + pVnode->tsdb = NULL; + + if (pVnode->cq) + cqClose(pVnode->cq); + pVnode->cq = NULL; + + if (pVnode->wqueue) + dnodeFreeWqueue(pVnode->wqueue); + pVnode->wqueue = NULL; + + if (pVnode->rqueue) + dnodeFreeRqueue(pVnode->rqueue); + pVnode->rqueue = NULL; + vnodeRelease(pVnode); } @@ -512,27 +541,31 @@ static int32_t vnodeSaveCfg(SMDCreateVnodeMsg *pVnodeCfg) { } static int32_t vnodeReadCfg(SVnodeObj *pVnode) { - char cfgFile[TSDB_FILENAME_LEN + 30] = {0}; + cJSON *root = NULL; + char *content = NULL; + char cfgFile[TSDB_FILENAME_LEN + 30] = {0}; + int maxLen = 1000; + int32_t code = TSDB_CODE_OTHERS; + + terrno = TSDB_CODE_OTHERS; sprintf(cfgFile, "%s/vnode%d/config.json", tsVnodeDir, pVnode->vgId); FILE *fp = fopen(cfgFile, "r"); if (!fp) { - vError("vgId:%d, failed to open vnode cfg file for read, file:%s, error:%s", pVnode->vgId, + vError("vgId:%d, failed to open vnode cfg file:%s to read, error:%s", pVnode->vgId, cfgFile, strerror(errno)); + code = TAOS_SYSTEM_ERROR(errno); + goto PARSE_OVER; + } + + content = calloc(1, maxLen + 1); + if (content == NULL) goto PARSE_OVER; + int len = fread(content, 1, maxLen, fp); + if (len <= 0) { + vError("vgId:%d, failed to read vnode cfg, content is null", pVnode->vgId); return errno; } - int ret = TSDB_CODE_OTHERS; - int maxLen = 1000; - char *content = calloc(1, maxLen + 1); - int len = fread(content, 1, maxLen, fp); - if (len <= 0) { - free(content); - fclose(fp); - vError("vgId:%d, failed to read vnode cfg, content is null", pVnode->vgId); - return false; - } - - cJSON *root = cJSON_Parse(content); + root = cJSON_Parse(content); if (root == NULL) { vError("vgId:%d, failed to read vnode cfg, invalid json format", pVnode->vgId); goto PARSE_OVER; @@ -691,19 +724,19 @@ static int32_t vnodeReadCfg(SVnodeObj *pVnode) { pVnode->syncCfg.nodeInfo[i].nodePort += TSDB_PORT_SYNC; } - ret = 0; + code = TSDB_CODE_SUCCESS; - vPrint("vgId:%d, read vnode cfg successed, replcia:%d", pVnode->vgId, pVnode->syncCfg.replica); + vPrint("vgId:%d, read vnode cfg successfully, replcia:%d", pVnode->vgId, pVnode->syncCfg.replica); for (int32_t i = 0; i < pVnode->syncCfg.replica; i++) { vPrint("vgId:%d, dnode:%d, %s:%d", pVnode->vgId, pVnode->syncCfg.nodeInfo[i].nodeId, pVnode->syncCfg.nodeInfo[i].nodeFqdn, pVnode->syncCfg.nodeInfo[i].nodePort); } PARSE_OVER: - free(content); + tfree(content); cJSON_Delete(root); - fclose(fp); - return ret; + if (fp) fclose(fp); + return code; } static int32_t vnodeSaveVersion(SVnodeObj *pVnode) { @@ -713,7 +746,7 @@ static int32_t vnodeSaveVersion(SVnodeObj *pVnode) { if (!fp) { vError("vgId:%d, failed to open vnode version file for write, file:%s error:%s", pVnode->vgId, versionFile, strerror(errno)); - return errno; + return TAOS_SYSTEM_ERROR(errno); } int32_t len = 0; @@ -733,29 +766,33 @@ static int32_t vnodeSaveVersion(SVnodeObj *pVnode) { return 0; } -static bool vnodeReadVersion(SVnodeObj *pVnode) { - char versionFile[TSDB_FILENAME_LEN + 30] = {0}; +static int32_t vnodeReadVersion(SVnodeObj *pVnode) { + char versionFile[TSDB_FILENAME_LEN + 30] = {0}; + char *content = NULL; + cJSON *root = NULL; + int maxLen = 100; + int32_t code = TSDB_CODE_OTHERS; + sprintf(versionFile, "%s/vnode%d/version.json", tsVnodeDir, pVnode->vgId); FILE *fp = fopen(versionFile, "r"); if (!fp) { if (errno != ENOENT) { vError("vgId:%d, failed to open version file:%s error:%s", pVnode->vgId, versionFile, strerror(errno)); + code = TAOS_SYSTEM_ERROR(errno); + } else { + code = TSDB_CODE_SUCCESS; } - return false; + goto PARSE_OVER; } - bool ret = false; - int maxLen = 100; - char *content = calloc(1, maxLen + 1); + content = calloc(1, maxLen + 1); int len = fread(content, 1, maxLen, fp); if (len <= 0) { - free(content); - fclose(fp); - vPrint("vgId:%d, failed to read vnode version, content is null", pVnode->vgId); - return false; + vError("vgId:%d, failed to read vnode version, content is null", pVnode->vgId); + goto PARSE_OVER; } - cJSON *root = cJSON_Parse(content); + root = cJSON_Parse(content); if (root == NULL) { vError("vgId:%d, failed to read vnode version, invalid json format", pVnode->vgId); goto PARSE_OVER; @@ -768,13 +805,13 @@ static bool vnodeReadVersion(SVnodeObj *pVnode) { } pVnode->version = version->valueint; - ret = true; + code = TSDB_CODE_SUCCESS; - vPrint("vgId:%d, read vnode version succeed, version:%" PRId64, pVnode->vgId, pVnode->version); + vPrint("vgId:%d, read vnode version successfully, version:%" PRId64, pVnode->vgId, pVnode->version); PARSE_OVER: - free(content); + tfree(content); cJSON_Delete(root); - fclose(fp); - return ret; + if(fp) fclose(fp); + return code; } diff --git a/src/wal/src/walMain.c b/src/wal/src/walMain.c index 8d92fac926..ebfc9d98bb 100644 --- a/src/wal/src/walMain.c +++ b/src/wal/src/walMain.c @@ -25,6 +25,7 @@ #include "tlog.h" #include "tchecksum.h" #include "tutil.h" +#include "taoserror.h" #include "twal.h" #include "tqueue.h" @@ -56,7 +57,10 @@ static int walRemoveWalFiles(const char *path); void *walOpen(const char *path, const SWalCfg *pCfg) { SWal *pWal = calloc(sizeof(SWal), 1); - if (pWal == NULL) return NULL; + if (pWal == NULL) { + terrno = TAOS_SYSTEM_ERROR(errno); + return NULL; + } pWal->fd = -1; pWal->max = pCfg->wals; @@ -75,6 +79,7 @@ void *walOpen(const char *path, const SWalCfg *pCfg) { walRenew(pWal); if (pWal->fd <0) { + terrno = TAOS_SYSTEM_ERROR(errno); wError("wal:%s, failed to open", path); pthread_mutex_destroy(&pWal->mutex); free(pWal); @@ -112,9 +117,10 @@ void walClose(void *handle) { } int walRenew(void *handle) { + if (handle == NULL) return 0; SWal *pWal = handle; int code = 0; - + pthread_mutex_lock(&pWal->mutex); if (pWal->fd >=0) { @@ -156,6 +162,7 @@ int walRenew(void *handle) { int walWrite(void *handle, SWalHead *pHead) { SWal *pWal = handle; int code = 0; + if (pWal == NULL) return -1; // no wal if (pWal->level == TAOS_WAL_NOLOG) return 0; @@ -178,6 +185,7 @@ int walWrite(void *handle, SWalHead *pHead) { void walFsync(void *handle) { SWal *pWal = handle; + if (pWal == NULL) return; if (pWal->level == TAOS_WAL_FSYNC && pWal->fd >=0) { if (fsync(pWal->fd) < 0) { From b9c08aff60d4a591455ac4138440bd746f3e7425 Mon Sep 17 00:00:00 2001 From: Jeff Tao Date: Sat, 23 May 2020 13:54:36 +0000 Subject: [PATCH 4/9] make more changes on vnodeMain.c --- src/vnode/src/vnodeMain.c | 46 +++++++++++++-------------------------- 1 file changed, 15 insertions(+), 31 deletions(-) diff --git a/src/vnode/src/vnodeMain.c b/src/vnode/src/vnodeMain.c index 926eee6037..5d41fa6bc4 100644 --- a/src/vnode/src/vnodeMain.c +++ b/src/vnode/src/vnodeMain.c @@ -148,35 +148,20 @@ int32_t vnodeAlter(void *param, SMDCreateVnodeMsg *pVnodeCfg) { pVnode->status = TAOS_VN_STATUS_UPDATING; int32_t code = vnodeSaveCfg(pVnodeCfg); - if (code != TSDB_CODE_SUCCESS) { - vError("vgId:%d, failed to save vnode cfg, reason:%s", pVnodeCfg->cfg.vgId, tstrerror(code)); - return code; - } + if (code != TSDB_CODE_SUCCESS) return code; code = vnodeReadCfg(pVnode); - if (code != TSDB_CODE_SUCCESS) { - vError("vgId:%d, failed to read cfg file", pVnode->vgId); - taosHashRemove(tsDnodeVnodesHash, (const char *)&pVnode->vgId, sizeof(int32_t)); - return code; - } + if (code != TSDB_CODE_SUCCESS) return code; code = syncReconfig(pVnode->sync, &pVnode->syncCfg); - if (code != TSDB_CODE_SUCCESS) { - vTrace("vgId:%d, failed to alter vnode, canot reconfig sync, result:%s", pVnode->vgId, - tstrerror(code)); - return code; - } + if (code != TSDB_CODE_SUCCESS) return code; code = tsdbConfigRepo(pVnode->tsdb, &pVnode->tsdbCfg); - if (code != TSDB_CODE_SUCCESS) { - vTrace("vgId:%d, failed to alter vnode, canot reconfig tsdb, result:%s", pVnode->vgId, - tstrerror(code)); - return code; - } + if (code != TSDB_CODE_SUCCESS) return code; pVnode->status = TAOS_VN_STATUS_READY; - vTrace("vgId:%d, vnode is altered", pVnode->vgId); + return TSDB_CODE_SUCCESS; } @@ -491,7 +476,8 @@ static int32_t vnodeSaveCfg(SMDCreateVnodeMsg *pVnodeCfg) { if (!fp) { vError("vgId:%d, failed to open vnode cfg file for write, file:%s error:%s", pVnodeCfg->cfg.vgId, cfgFile, strerror(errno)); - return errno; + terrno = TAOS_SYSTEM_ERROR(errno); + return terrno; } int32_t len = 0; @@ -545,7 +531,6 @@ static int32_t vnodeReadCfg(SVnodeObj *pVnode) { char *content = NULL; char cfgFile[TSDB_FILENAME_LEN + 30] = {0}; int maxLen = 1000; - int32_t code = TSDB_CODE_OTHERS; terrno = TSDB_CODE_OTHERS; sprintf(cfgFile, "%s/vnode%d/config.json", tsVnodeDir, pVnode->vgId); @@ -553,7 +538,7 @@ static int32_t vnodeReadCfg(SVnodeObj *pVnode) { if (!fp) { vError("vgId:%d, failed to open vnode cfg file:%s to read, error:%s", pVnode->vgId, cfgFile, strerror(errno)); - code = TAOS_SYSTEM_ERROR(errno); + terrno = TAOS_SYSTEM_ERROR(errno); goto PARSE_OVER; } @@ -724,7 +709,7 @@ static int32_t vnodeReadCfg(SVnodeObj *pVnode) { pVnode->syncCfg.nodeInfo[i].nodePort += TSDB_PORT_SYNC; } - code = TSDB_CODE_SUCCESS; + terrno = TSDB_CODE_SUCCESS; vPrint("vgId:%d, read vnode cfg successfully, replcia:%d", pVnode->vgId, pVnode->syncCfg.replica); for (int32_t i = 0; i < pVnode->syncCfg.replica; i++) { @@ -736,7 +721,7 @@ PARSE_OVER: tfree(content); cJSON_Delete(root); if (fp) fclose(fp); - return code; + return terrno; } static int32_t vnodeSaveVersion(SVnodeObj *pVnode) { @@ -771,16 +756,16 @@ static int32_t vnodeReadVersion(SVnodeObj *pVnode) { char *content = NULL; cJSON *root = NULL; int maxLen = 100; - int32_t code = TSDB_CODE_OTHERS; + terrno = TSDB_CODE_OTHERS; sprintf(versionFile, "%s/vnode%d/version.json", tsVnodeDir, pVnode->vgId); FILE *fp = fopen(versionFile, "r"); if (!fp) { if (errno != ENOENT) { vError("vgId:%d, failed to open version file:%s error:%s", pVnode->vgId, versionFile, strerror(errno)); - code = TAOS_SYSTEM_ERROR(errno); + terrno = TAOS_SYSTEM_ERROR(errno); } else { - code = TSDB_CODE_SUCCESS; + terrno = TSDB_CODE_SUCCESS; } goto PARSE_OVER; } @@ -805,13 +790,12 @@ static int32_t vnodeReadVersion(SVnodeObj *pVnode) { } pVnode->version = version->valueint; - code = TSDB_CODE_SUCCESS; - + terrno = TSDB_CODE_SUCCESS; vPrint("vgId:%d, read vnode version successfully, version:%" PRId64, pVnode->vgId, pVnode->version); PARSE_OVER: tfree(content); cJSON_Delete(root); if(fp) fclose(fp); - return code; + return terrno; } From 19bc76109b1f0e26fb8ed4e734e10a4c407e9a45 Mon Sep 17 00:00:00 2001 From: Jeff Tao Date: Sat, 23 May 2020 14:47:49 +0000 Subject: [PATCH 5/9] compliing opton for sync --- src/vnode/src/vnodeMain.c | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/src/vnode/src/vnodeMain.c b/src/vnode/src/vnodeMain.c index 5d41fa6bc4..d548c0fc98 100644 --- a/src/vnode/src/vnodeMain.c +++ b/src/vnode/src/vnodeMain.c @@ -249,13 +249,14 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) { syncInfo.notifyRole = vnodeNotifyRole; syncInfo.notifyFileSynced = vnodeNotifyFileSynced; pVnode->sync = syncStart(&syncInfo); + +#ifndef _SYNC + pVnode->role = TAOS_SYNC_ROLE_MASTER; +#elif if (pVnode->sync == NULL) { vnodeCleanUp(pVnode); return terrno; } - -#ifndef _SYNC - pVnode->role = TAOS_SYNC_ROLE_MASTER; #endif // start continuous query From 4187ae43ab71726a2f02f29b9135addf500cc972 Mon Sep 17 00:00:00 2001 From: Jeff Tao Date: Sat, 23 May 2020 14:59:11 +0000 Subject: [PATCH 6/9] compiling option --- src/vnode/src/vnodeMain.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/vnode/src/vnodeMain.c b/src/vnode/src/vnodeMain.c index d548c0fc98..b8bc29550e 100644 --- a/src/vnode/src/vnodeMain.c +++ b/src/vnode/src/vnodeMain.c @@ -252,7 +252,7 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) { #ifndef _SYNC pVnode->role = TAOS_SYNC_ROLE_MASTER; -#elif +#else if (pVnode->sync == NULL) { vnodeCleanUp(pVnode); return terrno; From 3a7570d21b847704b37a19202e5c921dfb5f947a Mon Sep 17 00:00:00 2001 From: Jeff Tao Date: Sun, 24 May 2020 06:12:32 +0000 Subject: [PATCH 7/9] for TCP client, dont set socket option --- src/util/src/tsocket.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/util/src/tsocket.c b/src/util/src/tsocket.c index f2b89c8243..d92228a089 100644 --- a/src/util/src/tsocket.c +++ b/src/util/src/tsocket.c @@ -305,7 +305,7 @@ int taosOpenTcpClientSocket(uint32_t destIp, uint16_t destPort, uint32_t clientI sockFd = -1; } - taosKeepTcpAlive(sockFd); + // taosKeepTcpAlive(sockFd); return sockFd; } From a585b1d9be7f8b98f0d8f7f26d4c54ce6cd1cbc3 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Sun, 24 May 2020 20:58:54 +0800 Subject: [PATCH 8/9] [TD-413] dnode may be null if mnode list changed --- src/mnode/inc/mgmtDef.h | 1 - src/mnode/src/mgmtMnode.c | 40 +++++++++++++++++++++++++-------------- src/mnode/src/mgmtSdb.c | 12 +++++++++--- 3 files changed, 35 insertions(+), 18 deletions(-) diff --git a/src/mnode/inc/mgmtDef.h b/src/mnode/inc/mgmtDef.h index 58d16ce1b3..9d3e46205d 100644 --- a/src/mnode/inc/mgmtDef.h +++ b/src/mnode/inc/mgmtDef.h @@ -63,7 +63,6 @@ typedef struct SMnodeObj { int8_t updateEnd[1]; int32_t refCount; int8_t role; - SDnodeObj *pDnode; } SMnodeObj; typedef struct { diff --git a/src/mnode/src/mgmtMnode.c b/src/mnode/src/mgmtMnode.c index e9d14dc6e7..6471b7f182 100644 --- a/src/mnode/src/mgmtMnode.c +++ b/src/mnode/src/mgmtMnode.c @@ -65,7 +65,6 @@ static int32_t mgmtMnodeActionInsert(SSdbOper *pOper) { SDnodeObj *pDnode = mgmtGetDnode(pMnode->mnodeId); if (pDnode == NULL) return TSDB_CODE_DNODE_NOT_EXIST; - pMnode->pDnode = pDnode; pDnode->isMgmt = true; mgmtDecDnodeRef(pDnode); @@ -220,22 +219,27 @@ void mgmtUpdateMnodeIpSet() { pIter = mgmtGetNextMnode(pIter, &pMnode); if (pMnode == NULL) break; - strcpy(ipSet->fqdn[ipSet->numOfIps], pMnode->pDnode->dnodeFqdn); - ipSet->port[ipSet->numOfIps] = htons(pMnode->pDnode->dnodePort); + SDnodeObj *pDnode = mgmtGetDnode(pMnode->mnodeId); + if (pDnode != NULL) { + strcpy(ipSet->fqdn[ipSet->numOfIps], pDnode->dnodeFqdn); + ipSet->port[ipSet->numOfIps] = htons(pDnode->dnodePort); - mnodes->nodeInfos[index].nodeId = htonl(pMnode->mnodeId); - strcpy(mnodes->nodeInfos[index].nodeEp, pMnode->pDnode->dnodeEp); + mnodes->nodeInfos[index].nodeId = htonl(pMnode->mnodeId); + strcpy(mnodes->nodeInfos[index].nodeEp, pDnode->dnodeEp); - if (pMnode->role == TAOS_SYNC_ROLE_MASTER) { - ipSet->inUse = ipSet->numOfIps; - mnodes->inUse = index; + if (pMnode->role == TAOS_SYNC_ROLE_MASTER) { + ipSet->inUse = ipSet->numOfIps; + mnodes->inUse = index; + } + + mPrint("mnode:%d, ep:%s %s", index, pDnode->dnodeEp, + pMnode->role == TAOS_SYNC_ROLE_MASTER ? "master" : ""); + + ipSet->numOfIps++; + index++; } - mPrint("mnode:%d, ep:%s %s", index, pMnode->pDnode->dnodeEp, pMnode->role == TAOS_SYNC_ROLE_MASTER ? "master" : ""); - - ipSet->numOfIps++; - index++; - + mgmtDecDnodeRef(pDnode); mgmtDecMnodeRef(pMnode); } @@ -385,7 +389,15 @@ static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, voi cols++; pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows; - STR_WITH_MAXSIZE_TO_VARSTR(pWrite, pMnode->pDnode->dnodeEp, pShow->bytes[cols] - VARSTR_HEADER_SIZE); + + SDnodeObj *pDnode = mgmtGetDnode(pMnode->mnodeId); + if (pDnode != NULL) { + STR_WITH_MAXSIZE_TO_VARSTR(pWrite, pDnode->dnodeEp, pShow->bytes[cols] - VARSTR_HEADER_SIZE); + } else { + STR_WITH_MAXSIZE_TO_VARSTR(pWrite, "invalid ep", pShow->bytes[cols] - VARSTR_HEADER_SIZE); + } + mgmtDecDnodeRef(pDnode); + cols++; pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows; diff --git a/src/mnode/src/mgmtSdb.c b/src/mnode/src/mgmtSdb.c index 3d4e6fcab1..237d2ca499 100644 --- a/src/mnode/src/mgmtSdb.c +++ b/src/mnode/src/mgmtSdb.c @@ -28,6 +28,7 @@ #include "mgmtDef.h" #include "mgmtInt.h" #include "mgmtMnode.h" +#include "mgmtDnode.h" #include "mgmtSdb.h" typedef enum { @@ -259,10 +260,15 @@ void sdbUpdateSync() { if (pMnode == NULL) break; syncCfg.nodeInfo[index].nodeId = pMnode->mnodeId; - syncCfg.nodeInfo[index].nodePort = pMnode->pDnode->dnodePort + TSDB_PORT_SYNC; - strcpy(syncCfg.nodeInfo[index].nodeFqdn, pMnode->pDnode->dnodeEp); - index++; + SDnodeObj *pDnode = mgmtGetDnode(pMnode->mnodeId); + if (pDnode != NULL) { + syncCfg.nodeInfo[index].nodePort = pDnode->dnodePort + TSDB_PORT_SYNC; + strcpy(syncCfg.nodeInfo[index].nodeFqdn, pDnode->dnodeEp); + index++; + } + + mgmtDecDnodeRef(pDnode); mgmtDecMnodeRef(pMnode); } sdbFreeIter(pIter); From 54d0511ea57d52803763a8239055c09b65a35cb5 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Sun, 24 May 2020 21:07:24 +0800 Subject: [PATCH 9/9] script --- tests/script/tmp/prepare.sim | 16 +++++----------- 1 file changed, 5 insertions(+), 11 deletions(-) diff --git a/tests/script/tmp/prepare.sim b/tests/script/tmp/prepare.sim index 31d7839566..1db643c5c9 100644 --- a/tests/script/tmp/prepare.sim +++ b/tests/script/tmp/prepare.sim @@ -1,14 +1,4 @@ system sh/stop_dnodes.sh -system sh/deploy.sh -n dnode1 -i 1 -system sh/deploy.sh -n dnode2 -i 2 -system sh/deploy.sh -n dnode3 -i 3 - -system sh/cfg.sh -n dnode1 -c numOfMPeers -v 2 -system sh/cfg.sh -n dnode2 -c numOfMPeers -v 2 -system sh/cfg.sh -n dnode3 -c numOfMPeers -v 2 - -return -system sh/stop_dnodes.sh system sh/deploy.sh -n dnode1 -i 1 system sh/deploy.sh -n dnode2 -i 2 @@ -33,4 +23,8 @@ system sh/cfg.sh -n dnode4 -c mgmtEqualVnodeNum -v 4 system sh/cfg.sh -n dnode1 -c numOfTotalVnodes -v 4 system sh/cfg.sh -n dnode2 -c numOfTotalVnodes -v 4 system sh/cfg.sh -n dnode3 -c numOfTotalVnodes -v 4 -system sh/cfg.sh -n dnode4 -c numOfTotalVnodes -v 4 \ No newline at end of file +system sh/cfg.sh -n dnode4 -c numOfTotalVnodes -v 4 + +system sh/cfg.sh -n dnode1 -c http -v 1 +system sh/cfg.sh -n dnode2 -c http -v 1 +system sh/cfg.sh -n dnode3 -c http -v 1 \ No newline at end of file