From 80fc12ea79afdbfed7f0ee8c86a20b0bafa3fd32 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Tue, 18 May 2021 23:17:53 +0800 Subject: [PATCH 01/19] [td-4231]: report error if only tags of the table are retrieved in the select clause when the filter condition only includes normal columns. --- src/client/src/tscSQLParser.c | 11 +++++++++++ tests/script/general/parser/select_with_tags.sim | 8 +++++++- 2 files changed, 18 insertions(+), 1 deletion(-) diff --git a/src/client/src/tscSQLParser.c b/src/client/src/tscSQLParser.c index f73556a890..5fc92dd860 100644 --- a/src/client/src/tscSQLParser.c +++ b/src/client/src/tscSQLParser.c @@ -7119,6 +7119,7 @@ int32_t doValidateSqlNode(SSqlObj* pSql, SQuerySqlNode* pQuerySqlNode, int32_t i const char* msg6 = "too many tables in from clause"; const char* msg7 = "invalid table alias name"; const char* msg8 = "alias name too long"; + const char* msg9 = "only tag query not compatible with normal column filter"; int32_t code = TSDB_CODE_SUCCESS; @@ -7288,6 +7289,16 @@ int32_t doValidateSqlNode(SSqlObj* pSql, SQuerySqlNode* pQuerySqlNode, int32_t i if (hasUnsupportFunctionsForSTableQuery(pCmd, pQueryInfo)) { return TSDB_CODE_TSC_INVALID_SQL; } + + if(tscQueryTags(pQueryInfo)) { + int32_t numOfCols = taosArrayGetSize(pQueryInfo->colList); + for(int32_t i = 0; i < numOfCols; ++i) { + SColumn* pCols = taosArrayGetP(pQueryInfo->colList, i); + if (pCols->numOfFilters > 0) { + return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg9); + } + } + } } if (parseSessionClause(pCmd, pQueryInfo, pQuerySqlNode) != TSDB_CODE_SUCCESS) { diff --git a/tests/script/general/parser/select_with_tags.sim b/tests/script/general/parser/select_with_tags.sim index 9f445649e1..11b1d0bc4b 100644 --- a/tests/script/general/parser/select_with_tags.sim +++ b/tests/script/general/parser/select_with_tags.sim @@ -847,10 +847,16 @@ sql_error select tbname, t1 from select_tags_mt0 interval(1y); #valid sql: select first(c1), last(c2), count(*) from select_tags_mt0 group by tbname, t1; #valid sql: select first(c1), tbname, t1 from select_tags_mt0 group by t2; +print ==================================>TD-4231 +sql_error select t1,tbname from select_tags_mt0 where c1<0 +sql_error select t1,tbname from select_tags_mt0 where c1<0 and tbname in ('select_tags_tb12') + +sql select tbname from select_tags_mt0 where tbname in ('select_tags_tb12'); + sql_error select first(c1), last(c2), t1 from select_tags_mt0 group by tbname; sql_error select first(c1), last(c2), tbname, t2 from select_tags_mt0 group by tbname; sql_error select first(c1), count(*), t2, t1, tbname from select_tags_mt0 group by tbname; -# this sql is valid: select first(c1), t2 from select_tags_mt0 group by tbname; +#valid sql: select first(c1), t2 from select_tags_mt0 group by tbname; #sql select first(ts), tbname from select_tags_mt0 group by tbname; #sql select count(c1) from select_tags_mt0 where c1=99 group by tbname; From 61ea20334f06cff53ddc4796aa67efa514932fb4 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Tue, 18 May 2021 23:27:16 +0800 Subject: [PATCH 02/19] [td-4231]fix compiler error. --- src/client/src/tscSQLParser.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/client/src/tscSQLParser.c b/src/client/src/tscSQLParser.c index 5fc92dd860..722ce7a231 100644 --- a/src/client/src/tscSQLParser.c +++ b/src/client/src/tscSQLParser.c @@ -7291,7 +7291,7 @@ int32_t doValidateSqlNode(SSqlObj* pSql, SQuerySqlNode* pQuerySqlNode, int32_t i } if(tscQueryTags(pQueryInfo)) { - int32_t numOfCols = taosArrayGetSize(pQueryInfo->colList); + int32_t numOfCols = (int32_t) taosArrayGetSize(pQueryInfo->colList); for(int32_t i = 0; i < numOfCols; ++i) { SColumn* pCols = taosArrayGetP(pQueryInfo->colList, i); if (pCols->numOfFilters > 0) { From 617a7f71896369a3ac39e9eee6efe01ed7e04e7c Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Thu, 20 May 2021 10:01:24 +0800 Subject: [PATCH 03/19] TD-4241 --- src/util/src/tcache.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/util/src/tcache.c b/src/util/src/tcache.c index c0cc8ce339..a8a1f69c1b 100644 --- a/src/util/src/tcache.c +++ b/src/util/src/tcache.c @@ -613,7 +613,7 @@ void doCleanupDataCache(SCacheObj *pCacheObj) { // todo memory leak if there are object with refcount greater than 0 in hash table? taosHashCleanup(pCacheObj->pHashTable); - taosTrashcanEmpty(pCacheObj, true); + taosTrashcanEmpty(pCacheObj, false); __cache_lock_destroy(pCacheObj); From 6c9df9d519bc59a84cb83818b93cc77ffd0f460f Mon Sep 17 00:00:00 2001 From: Minglei Jin Date: Fri, 21 May 2021 21:44:51 +0800 Subject: [PATCH 04/19] [TD-4279]: fix mnode single online as master --- src/sync/src/syncMain.c | 6 ++++-- tests/script/unique/arbitrator/dn3_mn1_vnode_nomaster.sim | 2 +- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/src/sync/src/syncMain.c b/src/sync/src/syncMain.c index e5f2d94c4a..1785f066d4 100644 --- a/src/sync/src/syncMain.c +++ b/src/sync/src/syncMain.c @@ -709,7 +709,7 @@ static void syncChooseMaster(SSyncNode *pNode) { } static SSyncPeer *syncCheckMaster(SSyncNode *pNode) { - int32_t onlineNum = 0; + int32_t onlineNum = 0, arbOnlineNum = 0; int32_t masterIndex = -1; int32_t replica = pNode->replica; @@ -723,13 +723,15 @@ static SSyncPeer *syncCheckMaster(SSyncNode *pNode) { SSyncPeer *pArb = pNode->peerInfo[TAOS_SYNC_MAX_REPLICA]; if (pArb && pArb->role != TAOS_SYNC_ROLE_OFFLINE) { onlineNum++; + ++arbOnlineNum; replica = pNode->replica + 1; } if (onlineNum <= replica * 0.5) { if (nodeRole != TAOS_SYNC_ROLE_UNSYNCED) { - if (nodeRole == TAOS_SYNC_ROLE_MASTER && onlineNum == replica * 0.5 && onlineNum >= 1) { + if (nodeRole == TAOS_SYNC_ROLE_MASTER && onlineNum == replica * 0.5 && ((replica > 2 && onlineNum - arbOnlineNum > 1) || pNode->replica < 3)) { sInfo("vgId:%d, self keep work as master, online:%d replica:%d", pNode->vgId, onlineNum, replica); + masterIndex = pNode->selfIndex; } else { nodeRole = TAOS_SYNC_ROLE_UNSYNCED; sInfo("vgId:%d, self change to unsynced state, online:%d replica:%d", pNode->vgId, onlineNum, replica); diff --git a/tests/script/unique/arbitrator/dn3_mn1_vnode_nomaster.sim b/tests/script/unique/arbitrator/dn3_mn1_vnode_nomaster.sim index 8e15c4f527..b9ee508f78 100644 --- a/tests/script/unique/arbitrator/dn3_mn1_vnode_nomaster.sim +++ b/tests/script/unique/arbitrator/dn3_mn1_vnode_nomaster.sim @@ -158,7 +158,7 @@ if $dnode4Vtatus != offline then sleep 2000 goto wait_dnode4_vgroup_offline endi -if $dnode3Vtatus != master then +if $dnode3Vtatus != unsynced then sleep 2000 goto wait_dnode4_vgroup_offline endi From ccdd3213e62cbc73e50dc0d8a25e7412b71ce3ba Mon Sep 17 00:00:00 2001 From: Shuduo Sang Date: Sat, 22 May 2021 21:07:40 +0800 Subject: [PATCH 05/19] Hotfix/sangshuduo/td 4240 taosdemo subscribe morethan 100 (#6202) * [TD-4240]: taosdemo subscribe more than max query sql count. * [TD-4240]: taosdemo subscribe more than 100. fix tsub sequence bug. * [TD-4240]: taosdemo subscribe more than 100. fix auto create table. * [TD-4240]: taosdemo subscribe more than 100. remove the condition should not be there. Co-authored-by: Shuduo Sang --- src/kit/taosdemo/taosdemo.c | 9 --------- 1 file changed, 9 deletions(-) diff --git a/src/kit/taosdemo/taosdemo.c b/src/kit/taosdemo/taosdemo.c index 02d72fc445..d4faecf1f0 100644 --- a/src/kit/taosdemo/taosdemo.c +++ b/src/kit/taosdemo/taosdemo.c @@ -6612,15 +6612,6 @@ static void *superSubscribe(void *sarg) { exit(-1); } - if (g_queryInfo.superQueryInfo.sqlCount * pThreadInfo->ntables > MAX_QUERY_SQL_COUNT) { - errorPrint("The number %"PRId64" of sql count(%"PRIu64") multiple the table number(%"PRId64") of the thread is more than max query sql count: %d\n", - g_queryInfo.superQueryInfo.sqlCount * pThreadInfo->ntables, - g_queryInfo.superQueryInfo.sqlCount, - pThreadInfo->ntables, - MAX_QUERY_SQL_COUNT); - exit(-1); - } - if (pThreadInfo->taos == NULL) { TAOS * taos = NULL; taos = taos_connect(g_queryInfo.host, From 94d45c3976ee24a8a3232275c197de3ab22b0f49 Mon Sep 17 00:00:00 2001 From: Shuduo Sang Date: Mon, 24 May 2021 11:27:11 +0800 Subject: [PATCH 06/19] [TD-3197]: taosdemo and taosdump coverity scan issues. (#6207) Co-authored-by: Shuduo Sang --- src/kit/taosdemo/taosdemo.c | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/src/kit/taosdemo/taosdemo.c b/src/kit/taosdemo/taosdemo.c index d4faecf1f0..81f48067d0 100644 --- a/src/kit/taosdemo/taosdemo.c +++ b/src/kit/taosdemo/taosdemo.c @@ -1114,21 +1114,21 @@ static int queryDbExec(TAOS *taos, char *command, QUERY_TYPE type, bool quiet) { static void appendResultBufToFile(char *resultBuf, char *resultFile) { - FILE *fp = NULL; - if (resultFile[0] != 0) { - fp = fopen(resultFile, "at"); - if (fp == NULL) { - errorPrint( - "%s() LN%d, failed to open result file: %s, result will not save to file\n", - __func__, __LINE__, resultFile); - return; + FILE *fp = NULL; + if (resultFile[0] != 0) { + fp = fopen(resultFile, "at"); + if (fp == NULL) { + errorPrint( + "%s() LN%d, failed to open result file: %s, result will not save to file\n", + __func__, __LINE__, resultFile); + return; + } + fprintf(fp, "%s", resultBuf); + tmfclose(fp); } - } - - fprintf(fp, "%s", resultBuf); - tmfclose(fp); } + static void appendResultToFile(TAOS_RES *res, char* resultFile) { TAOS_ROW row = NULL; int num_rows = 0; From 5772028d6e8807cf2f617987ed5c300cb46ca8ce Mon Sep 17 00:00:00 2001 From: zyyang <69311263+zyyang-taosdata@users.noreply.github.com> Date: Mon, 24 May 2021 11:45:23 +0800 Subject: [PATCH 07/19] change version number (#6204)https://github.com/taosdata/TDengine/tree/release/ver-2.0.20.5 Co-authored-by: plum-lihui --- .gitmodules | 6 +++--- cmake/version.inc | 2 +- snap/snapcraft.yaml | 4 ++-- src/connector/go | 2 +- 4 files changed, 7 insertions(+), 7 deletions(-) diff --git a/.gitmodules b/.gitmodules index 74afbbf997..0e65b02221 100644 --- a/.gitmodules +++ b/.gitmodules @@ -1,12 +1,12 @@ [submodule "src/connector/go"] path = src/connector/go - url = https://github.com/taosdata/driver-go + url = git@github.com:taosdata/driver-go.git [submodule "src/connector/grafanaplugin"] path = src/connector/grafanaplugin - url = https://github.com/taosdata/grafanaplugin + url = git@github.com:taosdata/grafanaplugin.git [submodule "src/connector/hivemq-tdengine-extension"] path = src/connector/hivemq-tdengine-extension - url = https://github.com/huskar-t/hivemq-tdengine-extension.git + url = git@github.com:taosdata/hivemq-tdengine-extension.git [submodule "tests/examples/rust"] path = tests/examples/rust url = https://github.com/songtianyi/tdengine-rust-bindings.git diff --git a/cmake/version.inc b/cmake/version.inc index a560c7f598..79eacb2d05 100755 --- a/cmake/version.inc +++ b/cmake/version.inc @@ -4,7 +4,7 @@ PROJECT(TDengine) IF (DEFINED VERNUMBER) SET(TD_VER_NUMBER ${VERNUMBER}) ELSE () - SET(TD_VER_NUMBER "2.0.20.2") + SET(TD_VER_NUMBER "2.0.20.5") ENDIF () IF (DEFINED VERCOMPATIBLE) diff --git a/snap/snapcraft.yaml b/snap/snapcraft.yaml index 9c7400c616..49eadd26d7 100644 --- a/snap/snapcraft.yaml +++ b/snap/snapcraft.yaml @@ -1,6 +1,6 @@ name: tdengine base: core18 -version: '2.0.20.2' +version: '2.0.20.5' icon: snap/gui/t-dengine.svg summary: an open-source big data platform designed and optimized for IoT. description: | @@ -72,7 +72,7 @@ parts: - usr/bin/taosd - usr/bin/taos - usr/bin/taosdemo - - usr/lib/libtaos.so.2.0.20.2 + - usr/lib/libtaos.so.2.0.20.5 - usr/lib/libtaos.so.1 - usr/lib/libtaos.so diff --git a/src/connector/go b/src/connector/go index 050667e5b4..8ce6d86558 160000 --- a/src/connector/go +++ b/src/connector/go @@ -1 +1 @@ -Subproject commit 050667e5b4d0eafa5387e4283e713559b421203f +Subproject commit 8ce6d86558afc8c0b50c10f990fd2b4270cf06fc From df588afe847a158f653729d997dd658a512c3104 Mon Sep 17 00:00:00 2001 From: Shuduo Sang Date: Mon, 24 May 2021 13:42:04 +0800 Subject: [PATCH 08/19] [TD-4322]: taosdemo query then fetch result. (#6210) Co-authored-by: Shuduo Sang --- src/kit/taosdemo/taosdemo.c | 57 ++++++++++++++++++------------------- 1 file changed, 28 insertions(+), 29 deletions(-) diff --git a/src/kit/taosdemo/taosdemo.c b/src/kit/taosdemo/taosdemo.c index 81f48067d0..5be7aea941 100644 --- a/src/kit/taosdemo/taosdemo.c +++ b/src/kit/taosdemo/taosdemo.c @@ -1114,22 +1114,20 @@ static int queryDbExec(TAOS *taos, char *command, QUERY_TYPE type, bool quiet) { static void appendResultBufToFile(char *resultBuf, char *resultFile) { - FILE *fp = NULL; - if (resultFile[0] != 0) { - fp = fopen(resultFile, "at"); - if (fp == NULL) { - errorPrint( - "%s() LN%d, failed to open result file: %s, result will not save to file\n", - __func__, __LINE__, resultFile); - return; - } - fprintf(fp, "%s", resultBuf); - tmfclose(fp); - } + FILE *fp = NULL; + fp = fopen(resultFile, "at"); + if (fp == NULL) { + errorPrint( + "%s() LN%d, failed to open result file: %s, result will not save to file\n", + __func__, __LINE__, resultFile); + return; + } + + fprintf(fp, "%s", resultBuf); + tmfclose(fp); } - -static void appendResultToFile(TAOS_RES *res, char* resultFile) { +static void fetchResult(TAOS_RES *res, char* resultFile) { TAOS_ROW row = NULL; int num_rows = 0; int num_fields = taos_field_count(res); @@ -1147,10 +1145,11 @@ static void appendResultToFile(TAOS_RES *res, char* resultFile) { // fetch the records row by row while((row = taos_fetch_row(res))) { - if (totalLen >= 100*1024*1024 - 32000) { - appendResultBufToFile(databuf, resultFile); - totalLen = 0; - memset(databuf, 0, 100*1024*1024); + if ((resultFile) + && (totalLen >= 100*1024*1024 - 32000)) { + appendResultBufToFile(databuf, resultFile); + totalLen = 0; + memset(databuf, 0, 100*1024*1024); } num_rows++; int len = taos_print_row(temp, row, fields, num_fields); @@ -1162,7 +1161,9 @@ static void appendResultToFile(TAOS_RES *res, char* resultFile) { verbosePrint("%s() LN%d, databuf=%s resultFile=%s\n", __func__, __LINE__, databuf, resultFile); - appendResultBufToFile(databuf, resultFile); + if (resultFile) { + appendResultBufToFile(databuf, resultFile); + } free(databuf); } @@ -1178,9 +1179,7 @@ static void selectAndGetResult( return; } - if ((strlen(pThreadInfo->fp))) { - appendResultToFile(res, pThreadInfo->fp); - } + fetchResult(res, pThreadInfo->fp); taos_free_result(res); } else if (0 == strncasecmp(g_queryInfo.queryMode, "rest", strlen("rest"))) { @@ -2017,13 +2016,13 @@ static void printfQuerySystemInfo(TAOS * taos) { // show variables res = taos_query(taos, "show variables;"); - //appendResultToFile(res, filename); + //fetchResult(res, filename); xDumpResultToFile(filename, res); // show dnodes res = taos_query(taos, "show dnodes;"); xDumpResultToFile(filename, res); - //appendResultToFile(res, filename); + //fetchResult(res, filename); // show databases res = taos_query(taos, "show databases;"); @@ -6546,7 +6545,7 @@ static void stable_sub_callback( } if (param) - appendResultToFile(res, ((threadInfo *)param)->fp); + fetchResult(res, ((threadInfo *)param)->fp); // tao_unscribe() will free result. } @@ -6559,7 +6558,7 @@ static void specified_sub_callback( } if (param) - appendResultToFile(res, ((threadInfo *)param)->fp); + fetchResult(res, ((threadInfo *)param)->fp); // tao_unscribe() will free result. } @@ -6700,13 +6699,13 @@ static void *superSubscribe(void *sarg) { sprintf(pThreadInfo->fp, "%s-%d", g_queryInfo.superQueryInfo.result[pThreadInfo->querySeq], pThreadInfo->threadID); - appendResultToFile(res, pThreadInfo->fp); + fetchResult(res, pThreadInfo->fp); } if (g_queryInfo.superQueryInfo.result[pThreadInfo->querySeq][0] != 0) { sprintf(pThreadInfo->fp, "%s-%d", g_queryInfo.superQueryInfo.result[pThreadInfo->querySeq], pThreadInfo->threadID); - appendResultToFile(res, pThreadInfo->fp); + fetchResult(res, pThreadInfo->fp); } consumed[tsubSeq] ++; @@ -6807,7 +6806,7 @@ static void *specifiedSubscribe(void *sarg) { sprintf(pThreadInfo->fp, "%s-%d", g_queryInfo.specifiedQueryInfo.result[pThreadInfo->querySeq], pThreadInfo->threadID); - appendResultToFile(res, pThreadInfo->fp); + fetchResult(res, pThreadInfo->fp); } consumed ++; From 20fa37c2538667849efb70191130ceb306c1938f Mon Sep 17 00:00:00 2001 From: Steven Li Date: Mon, 24 May 2021 06:11:39 +0000 Subject: [PATCH 09/19] Enhanced crash_gen tool with -k option to produce valgrind result --- tests/pytest/crash_gen/service_manager.py | 58 +++++++++++------------ tests/pytest/crash_gen/shared/types.py | 4 +- 2 files changed, 31 insertions(+), 31 deletions(-) diff --git a/tests/pytest/crash_gen/service_manager.py b/tests/pytest/crash_gen/service_manager.py index 95507f0142..1cd65c1dde 100644 --- a/tests/pytest/crash_gen/service_manager.py +++ b/tests/pytest/crash_gen/service_manager.py @@ -22,7 +22,7 @@ from queue import Queue, Empty from .shared.config import Config from .shared.db import DbTarget, DbConn from .shared.misc import Logging, Helper, CrashGenError, Status, Progress, Dice -from .shared.types import DirPath +from .shared.types import DirPath, IpcStream # from crash_gen.misc import CrashGenError, Dice, Helper, Logging, Progress, Status # from crash_gen.db import DbConn, DbTarget @@ -177,13 +177,12 @@ quorum 2 return "127.0.0.1" def getServiceCmdLine(self): # to start the instance - cmdLine = [] if Config.getConfig().track_memory_leaks: Logging.info("Invoking VALGRIND on service...") - cmdLine = ['valgrind', '--leak-check=yes'] - # TODO: move "exec -c" into Popen(), we can both "use shell" and NOT fork so ask to lose kill control - cmdLine += ["exec " + self.getExecFile(), '-c', self.getCfgDir()] # used in subproce.Popen() - return cmdLine + return ['exec /usr/bin/valgrind', '--leak-check=yes', self.getExecFile(), '-c', self.getCfgDir()] + else: + # TODO: move "exec -c" into Popen(), we can both "use shell" and NOT fork so ask to lose kill control + return ["exec " + self.getExecFile(), '-c', self.getCfgDir()] # used in subproce.Popen() def _getDnodes(self, dbc): dbc.query("show dnodes") @@ -281,16 +280,16 @@ class TdeSubProcess: return '[TdeSubProc: pid = {}, status = {}]'.format( self.getPid(), self.getStatus() ) - def getStdOut(self) -> BinaryIO : + def getIpcStdOut(self) -> IpcStream : if self._popen.universal_newlines : # alias of text_mode raise CrashGenError("We need binary mode for STDOUT IPC") # Logging.info("Type of stdout is: {}".format(type(self._popen.stdout))) - return typing.cast(BinaryIO, self._popen.stdout) + return typing.cast(IpcStream, self._popen.stdout) - def getStdErr(self) -> BinaryIO : + def getIpcStdErr(self) -> IpcStream : if self._popen.universal_newlines : # alias of text_mode raise CrashGenError("We need binary mode for STDERR IPC") - return typing.cast(BinaryIO, self._popen.stderr) + return typing.cast(IpcStream, self._popen.stderr) # Now it's always running, since we matched the life cycle # def isRunning(self): @@ -301,11 +300,6 @@ class TdeSubProcess: def _start(self, cmdLine) -> Popen : ON_POSIX = 'posix' in sys.builtin_module_names - - # Sanity check - # if self.subProcess: # already there - # raise RuntimeError("Corrupt process state") - # Prepare environment variables for coverage information # Ref: https://stackoverflow.com/questions/2231227/python-subprocess-popen-with-a-modified-environment @@ -314,9 +308,8 @@ class TdeSubProcess: # print(myEnv) # print("Starting TDengine with env: ", myEnv.items()) - # print("Starting TDengine via Shell: {}".format(cmdLineStr)) + print("Starting TDengine: {}".format(cmdLine)) - # useShell = True # Needed to pass environments into it return Popen( ' '.join(cmdLine), # ' '.join(cmdLine) if useShell else cmdLine, shell=True, # Always use shell, since we need to pass ENV vars @@ -732,19 +725,19 @@ class ServiceManagerThread: self._ipcQueue = Queue() # type: Queue self._thread = threading.Thread( # First thread captures server OUTPUT target=self.svcOutputReader, - args=(subProc.getStdOut(), self._ipcQueue, logDir)) + args=(subProc.getIpcStdOut(), self._ipcQueue, logDir)) self._thread.daemon = True # thread dies with the program self._thread.start() time.sleep(0.01) if not self._thread.is_alive(): # What happened? - Logging.info("Failed to started process to monitor STDOUT") + Logging.info("Failed to start process to monitor STDOUT") self.stop() raise CrashGenError("Failed to start thread to monitor STDOUT") Logging.info("Successfully started process to monitor STDOUT") self._thread2 = threading.Thread( # 2nd thread captures server ERRORs target=self.svcErrorReader, - args=(subProc.getStdErr(), self._ipcQueue, logDir)) + args=(subProc.getIpcStdErr(), self._ipcQueue, logDir)) self._thread2.daemon = True # thread dies with the program self._thread2.start() time.sleep(0.01) @@ -887,14 +880,19 @@ class ServiceManagerThread: print("\nNon-UTF8 server output: {}\n".format(bChunk.decode('cp437'))) return None - def _textChunkGenerator(self, streamIn: BinaryIO, logDir: str, logFile: str + def _textChunkGenerator(self, streamIn: IpcStream, logDir: str, logFile: str ) -> Generator[TextChunk, None, None]: ''' - Take an input stream with binary data, produced a generator of decoded - "text chunks", and also save the original binary data in a log file. + Take an input stream with binary data (likely from Popen), produced a generator of decoded + "text chunks". + + Side effect: it also save the original binary data in a log file. ''' os.makedirs(logDir, exist_ok=True) logF = open(os.path.join(logDir, logFile), 'wb') + if logF is None: + Logging.error("Failed to open log file (binary write): {}/{}".format(logDir, logFile)) + return for bChunk in iter(streamIn.readline, b''): logF.write(bChunk) # Write to log file immediately tChunk = self._decodeBinaryChunk(bChunk) # decode @@ -902,14 +900,14 @@ class ServiceManagerThread: yield tChunk # TODO: split into actual text lines # At the end... - streamIn.close() # Close the stream - logF.close() # Close the output file + streamIn.close() # Close the incoming stream + logF.close() # Close the log file - def svcOutputReader(self, stdOut: BinaryIO, queue, logDir: str): + def svcOutputReader(self, ipcStdOut: IpcStream, queue, logDir: str): ''' The infinite routine that processes the STDOUT stream for the sub process being managed. - :param stdOut: the IO stream object used to fetch the data from + :param ipcStdOut: the IO stream object used to fetch the data from :param queue: the queue where we dump the roughly parsed chunk-by-chunk text data :param logDir: where we should dump a verbatim output file ''' @@ -917,7 +915,7 @@ class ServiceManagerThread: # Important Reference: https://stackoverflow.com/questions/375427/non-blocking-read-on-a-subprocess-pipe-in-python # print("This is the svcOutput Reader...") # stdOut.readline() # Skip the first output? TODO: remove? - for tChunk in self._textChunkGenerator(stdOut, logDir, 'stdout.log') : + for tChunk in self._textChunkGenerator(ipcStdOut, logDir, 'stdout.log') : queue.put(tChunk) # tChunk garanteed not to be None self._printProgress("_i") @@ -940,12 +938,12 @@ class ServiceManagerThread: Logging.info("EOF found TDengine STDOUT, marking the process as terminated") self.setStatus(Status.STATUS_STOPPED) - def svcErrorReader(self, stdErr: BinaryIO, queue, logDir: str): + def svcErrorReader(self, ipcStdErr: IpcStream, queue, logDir: str): # os.makedirs(logDir, exist_ok=True) # logFile = os.path.join(logDir,'stderr.log') # fErr = open(logFile, 'wb') # for line in iter(err.readline, b''): - for tChunk in self._textChunkGenerator(stdErr, logDir, 'stderr.log') : + for tChunk in self._textChunkGenerator(ipcStdErr, logDir, 'stderr.log') : queue.put(tChunk) # tChunk garanteed not to be None # fErr.write(line) Logging.info("TDengine STDERR: {}".format(tChunk)) diff --git a/tests/pytest/crash_gen/shared/types.py b/tests/pytest/crash_gen/shared/types.py index 814a821917..42fd2a1617 100644 --- a/tests/pytest/crash_gen/shared/types.py +++ b/tests/pytest/crash_gen/shared/types.py @@ -1,4 +1,4 @@ -from typing import Any, List, Dict, NewType +from typing import Any, BinaryIO, List, Dict, NewType from enum import Enum DirPath = NewType('DirPath', str) @@ -26,3 +26,5 @@ class TdDataType(Enum): TdColumns = Dict[str, TdDataType] TdTags = Dict[str, TdDataType] + +IpcStream = NewType('IpcStream', BinaryIO) \ No newline at end of file From 0626a0914df776b336767d8fbd24021d9373e988 Mon Sep 17 00:00:00 2001 From: Shuduo Sang Date: Mon, 24 May 2021 16:32:11 +0800 Subject: [PATCH 10/19] Hotfix/sangshuduo/td 4296 taosdemo subscribe keepprogress (#6216) * [TD-4296]: taosdemo sub keepProgress. * make print sqlcount quiet. Co-authored-by: Shuduo Sang --- src/kit/taosdemo/taosdemo.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/kit/taosdemo/taosdemo.c b/src/kit/taosdemo/taosdemo.c index 5be7aea941..26088191cb 100644 --- a/src/kit/taosdemo/taosdemo.c +++ b/src/kit/taosdemo/taosdemo.c @@ -6915,7 +6915,7 @@ static int subscribeTestProcess() { //==== create threads for super table query if (g_queryInfo.superQueryInfo.sqlCount <= 0) { - printf("%s() LN%d, super table query sqlCount %"PRIu64".\n", + debugPrint("%s() LN%d, super table query sqlCount %"PRIu64".\n", __func__, __LINE__, g_queryInfo.superQueryInfo.sqlCount); } else { From 0923ccb3ad5edf7558cb57e0a2b32b1ae3e4aa0f Mon Sep 17 00:00:00 2001 From: liuyq-617 Date: Mon, 24 May 2021 17:08:37 +0800 Subject: [PATCH 11/19] [TD-4327]jenkins migration --- tests/Jenkinsfile | 6 +- tests/mas/Jenkinsfile | 309 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 312 insertions(+), 3 deletions(-) create mode 100644 tests/mas/Jenkinsfile diff --git a/tests/Jenkinsfile b/tests/Jenkinsfile index 64a71ef8f9..e785c8e807 100644 --- a/tests/Jenkinsfile +++ b/tests/Jenkinsfile @@ -37,7 +37,7 @@ pipeline { stage('Parallel test stage') { parallel { stage('pytest') { - agent{label '184'} + agent{label 'slad1'} steps { pre_test() sh ''' @@ -62,7 +62,7 @@ pipeline { } stage('test_crash_gen') { - agent{label "185"} + agent{label "slad2"} steps { pre_test() sh ''' @@ -149,7 +149,7 @@ pipeline { } stage('test_valgrind') { - agent{label "186"} + agent{label "slad3"} steps { pre_test() diff --git a/tests/mas/Jenkinsfile b/tests/mas/Jenkinsfile new file mode 100644 index 0000000000..b2a1a5e116 --- /dev/null +++ b/tests/mas/Jenkinsfile @@ -0,0 +1,309 @@ +def pre_test(){ + + sh ''' + sudo rmtaos||echo 'no taosd installed' + ''' + sh ''' + cd ${WKC} + git reset --hard + git checkout $BRANCH_NAME + git pull + git submodule update + cd ${WK} + git reset --hard + git checkout $BRANCH_NAME + git pull + export TZ=Asia/Harbin + date + rm -rf ${WK}/debug + mkdir debug + cd debug + cmake .. > /dev/null + make > /dev/null + make install > /dev/null + pip3 install ${WKC}/src/connector/python/linux/python3/ + ''' + return 1 +} +pipeline { + agent none + environment{ + + WK = '/var/lib/jenkins/workspace/TDinternal' + WKC= '/var/lib/jenkins/workspace/TDinternal/community' + } + + stages { + stage('Parallel test stage') { + parallel { + stage('pytest') { + agent{label 'slam1'} + steps { + pre_test() + sh ''' + cd ${WKC}/tests + find pytest -name '*'sql|xargs rm -rf + ./test-all.sh pytest + date''' + } + } + stage('test_b1') { + agent{label 'slam2'} + steps { + pre_test() + + sh ''' + cd ${WKC}/tests + ./test-all.sh b1 + date''' + + + } + } + + stage('test_crash_gen') { + agent{label "slam3"} + steps { + pre_test() + sh ''' + cd ${WKC}/tests/pytest + ''' + catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') { + sh ''' + cd ${WKC}/tests/pytest + ./crash_gen.sh -a -p -t 4 -s 2000 + ''' + } + catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') { + sh ''' + cd ${WKC}/tests/pytest + rm -rf /var/lib/taos/* + rm -rf /var/log/taos/* + ./handle_crash_gen_val_log.sh + ''' + } + catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') { + sh ''' + cd ${WKC}/tests/pytest + rm -rf /var/lib/taos/* + rm -rf /var/log/taos/* + ./handle_taosd_val_log.sh + ''' + } + + sh''' + systemctl start taosd + sleep 10 + ''' + catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') { + sh ''' + cd ${WKC}/tests/gotest + bash batchtest.sh + ''' + } + catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') { + sh ''' + cd ${WKC}/tests/examples/python/PYTHONConnectorChecker + python3 PythonChecker.py + ''' + } + catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') { + sh ''' + cd ${WKC}/tests/examples/JDBC/JDBCDemo/ + mvn clean package assembly:single -DskipTests >/dev/null + java -jar target/JDBCDemo-SNAPSHOT-jar-with-dependencies.jar -host 127.0.0.1 + ''' + } + catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') { + sh ''' + cd ${WKC}/src/connector/jdbc + mvn clean package -Dmaven.test.skip=true >/dev/null + cd ${WKC}/tests/examples/JDBC/JDBCDemo/ + java --class-path=../../../../src/connector/jdbc/target:$JAVA_HOME/jre/lib/ext -jar target/JDBCDemo-SNAPSHOT-jar-with-dependencies.jar -host 127.0.0.1 + ''' + } + catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') { + sh ''' + cp -rf ${WKC}/tests/examples/nodejs ${JENKINS_HOME}/workspace/ + cd ${JENKINS_HOME}/workspace/nodejs + node nodejsChecker.js host=localhost + ''' + } + catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') { + sh ''' + cd ${JENKINS_HOME}/workspace/C#NET/src/CheckC# + dotnet run + ''' + } + sh ''' + systemctl stop taosd + cd ${WKC}/tests + ./test-all.sh b2 + date + ''' + sh ''' + cd ${WKC}/tests + ./test-all.sh full unit + date''' + } + } + + stage('test_valgrind') { + agent{label "slam4"} + + steps { + pre_test() + catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') { + sh ''' + cd ${WKC}/tests/pytest + nohup taosd >/dev/null & + sleep 10 + python3 concurrent_inquiry.py -c 1 + + ''' + } + sh ''' + cd ${WKC}/tests + ./test-all.sh full jdbc + date''' + sh ''' + cd ${WKC}/tests/pytest + ./valgrind-test.sh 2>&1 > mem-error-out.log + ./handle_val_log.sh + + date + cd ${WKC}/tests + ./test-all.sh b3 + date''' + sh ''' + date + cd ${WKC}/tests + ./test-all.sh full example + date''' + } + } + + stage('arm64_build'){ + agent{label 'arm64'} + steps{ + sh ''' + cd ${WK} + git fetch + git checkout develop + git pull + cd ${WKC} + git fetch + git checkout develop + git pull + git submodule update + cd ${WKC}/packaging + ./release.sh -v cluster -c aarch64 -n 2.0.0.0 -m 2.0.0.0 + + ''' + } + } + stage('arm32_build'){ + agent{label 'arm32'} + steps{ + catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') { + sh ''' + cd ${WK} + git fetch + git checkout develop + git pull + cd ${WKC} + git fetch + git checkout develop + git pull + git submodule update + cd ${WKC}/packaging + ./release.sh -v cluster -c aarch32 -n 2.0.0.0 -m 2.0.0.0 + + ''' + } + + } + } + } + } + + } + post { + success { + emailext ( + subject: "SUCCESSFUL: Job '${env.JOB_NAME} [${env.BUILD_NUMBER}]'", + body: ''' + + + + + + + + + + + + +

+ 构建信息 +
+
    +
    +
  • 构建名称>>分支：${PROJECT_NAME}
    • +
  • 构建结果： Successful
    • +
  • 构建编号：${BUILD_NUMBER}
    • +
  • 触发用户：${CAUSE}
    • +
  • 变更概要：${CHANGES}
    • +
  • 构建地址：${BUILD_URL}
    • +
  • 构建日志：${BUILD_URL}console
    • +
  • 变更集：${JELLY_SCRIPT}
    • +
    +
+
+ + ''', + to: "yqliu@taosdata.com,pxiao@taosdata.com", + from: "support@taosdata.com" + ) + } + failure { + emailext ( + subject: "FAILED: Job '${env.JOB_NAME} [${env.BUILD_NUMBER}]'", + body: ''' + + + + + + + + + + + + +

+ 构建信息 +
+
    +
    +
  • 构建名称>>分支：${PROJECT_NAME}
    • +
  • 构建结果： Successful
    • +
  • 构建编号：${BUILD_NUMBER}
    • +
  • 触发用户：${CAUSE}
    • +
  • 变更概要：${CHANGES}
    • +
  • 构建地址：${BUILD_URL}
    • +
  • 构建日志：${BUILD_URL}console
    • +
  • 变更集：${JELLY_SCRIPT}
    • +
    +
+
+ + ''', + to: "yqliu@taosdata.com,pxiao@taosdata.com", + from: "support@taosdata.com" + ) + } + } +} \ No newline at end of file From 09f3c72455ef52ed637369f94d2244cac290f055 Mon Sep 17 00:00:00 2001 From: Shuduo Sang Date: Mon, 24 May 2021 20:12:20 +0800 Subject: [PATCH 12/19] Hotfix/sangshuduo/td 4322 taosdemo fetch result (#6219) * [TD-4322]: taosdemo query then fetch result. * [TD-4322]: taosdemo fetch result. check result file name length. Co-authored-by: Shuduo Sang --- src/kit/taosdemo/taosdemo.c | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/kit/taosdemo/taosdemo.c b/src/kit/taosdemo/taosdemo.c index 26088191cb..b37a3370c5 100644 --- a/src/kit/taosdemo/taosdemo.c +++ b/src/kit/taosdemo/taosdemo.c @@ -1145,7 +1145,7 @@ static void fetchResult(TAOS_RES *res, char* resultFile) { // fetch the records row by row while((row = taos_fetch_row(res))) { - if ((resultFile) + if ((resultFile) && (strlen(resultFile) > 0) && (totalLen >= 100*1024*1024 - 32000)) { appendResultBufToFile(databuf, resultFile); totalLen = 0; @@ -1161,7 +1161,7 @@ static void fetchResult(TAOS_RES *res, char* resultFile) { verbosePrint("%s() LN%d, databuf=%s resultFile=%s\n", __func__, __LINE__, databuf, resultFile); - if (resultFile) { + if ((resultFile) && (strlen(resultFile) > 0)) { appendResultBufToFile(databuf, resultFile); } free(databuf); @@ -2194,7 +2194,7 @@ static int postProceSql(char *host, struct sockaddr_in *pServAddr, uint16_t port response_buf[RESP_BUF_LEN - 1] = '\0'; printf("Response:\n%s\n", response_buf); - if (resultFile) { + if ((resultFile) && (strlen(resultFile) > 0)) { appendResultBufToFile(response_buf, resultFile); } From 40269fba16740ce096493ac7365b1d6e098f933a Mon Sep 17 00:00:00 2001 From: Shuduo Sang Date: Mon, 24 May 2021 23:04:03 +0800 Subject: [PATCH 13/19] Hotfix/sangshuduo/td 4296 taosdemo subscribe keepprogress (#6221) * [TD-4296]: taosdemo sub keepProgress. * make print sqlcount quiet. * [TD-4296]: taosdemo subscribe keepprogress. specified query topic thread safe. * [TD-4296]: taosdemo subscribe keepProgress. specified query thread safe. * fix compile warning for mac. Co-authored-by: Shuduo Sang --- src/kit/taosdemo/taosdemo.c | 394 ++++++++++++++++++------------------ 1 file changed, 199 insertions(+), 195 deletions(-) diff --git a/src/kit/taosdemo/taosdemo.c b/src/kit/taosdemo/taosdemo.c index b37a3370c5..b162be7739 100644 --- a/src/kit/taosdemo/taosdemo.c +++ b/src/kit/taosdemo/taosdemo.c @@ -363,7 +363,7 @@ typedef struct SDbs_S { typedef struct SpecifiedQueryInfo_S { uint64_t queryInterval; // 0: unlimit > 0 loop/s - uint64_t concurrent; + uint32_t concurrent; uint64_t sqlCount; uint32_t asyncMode; // 0: sync, 1: async uint64_t subscribeInterval; // ms @@ -374,6 +374,9 @@ typedef struct SpecifiedQueryInfo_S { char result[MAX_QUERY_SQL_COUNT][MAX_FILE_NAME_LEN+1]; int resubAfterConsume[MAX_QUERY_SQL_COUNT]; TAOS_SUB* tsub[MAX_QUERY_SQL_COUNT]; + char topic[MAX_QUERY_SQL_COUNT][32]; + int consumed[MAX_QUERY_SQL_COUNT]; + TAOS_RES* res[MAX_QUERY_SQL_COUNT]; uint64_t totalQueried; } SpecifiedQueryInfo; @@ -418,7 +421,8 @@ typedef struct SThreadInfo_S { int threadID; char db_name[MAX_DB_NAME_SIZE+1]; uint32_t time_precision; - char fp[4096]; + char filePath[4096]; + FILE *fp; char tb_prefix[MAX_TB_NAME_SIZE]; uint64_t start_table_from; uint64_t end_table_to; @@ -451,6 +455,7 @@ typedef struct SThreadInfo_S { // seq of query or subscribe uint64_t querySeq; // sequence number of sql command + TAOS_SUB* tsub; } threadInfo; @@ -532,7 +537,7 @@ static int createDatabasesAndStables(); static void createChildTables(); static int queryDbExec(TAOS *taos, char *command, QUERY_TYPE type, bool quiet); static int postProceSql(char *host, struct sockaddr_in *pServAddr, - uint16_t port, char* sqlstr, char *resultFile); + uint16_t port, char* sqlstr, threadInfo *pThreadInfo); /* ************ Global variables ************ */ @@ -1112,22 +1117,22 @@ static int queryDbExec(TAOS *taos, char *command, QUERY_TYPE type, bool quiet) { return 0; } -static void appendResultBufToFile(char *resultBuf, char *resultFile) +static void appendResultBufToFile(char *resultBuf, threadInfo *pThreadInfo) { - FILE *fp = NULL; - fp = fopen(resultFile, "at"); - if (fp == NULL) { + pThreadInfo->fp = fopen(pThreadInfo->filePath, "at"); + if (pThreadInfo->fp == NULL) { errorPrint( "%s() LN%d, failed to open result file: %s, result will not save to file\n", - __func__, __LINE__, resultFile); + __func__, __LINE__, pThreadInfo->filePath); return; } - fprintf(fp, "%s", resultBuf); - tmfclose(fp); + fprintf(pThreadInfo->fp, "%s", resultBuf); + tmfclose(pThreadInfo->fp); + pThreadInfo->fp = NULL; } -static void fetchResult(TAOS_RES *res, char* resultFile) { +static void fetchResult(TAOS_RES *res, threadInfo* pThreadInfo) { TAOS_ROW row = NULL; int num_rows = 0; int num_fields = taos_field_count(res); @@ -1145,9 +1150,9 @@ static void fetchResult(TAOS_RES *res, char* resultFile) { // fetch the records row by row while((row = taos_fetch_row(res))) { - if ((resultFile) && (strlen(resultFile) > 0) + if ((strlen(pThreadInfo->filePath) > 0) && (totalLen >= 100*1024*1024 - 32000)) { - appendResultBufToFile(databuf, resultFile); + appendResultBufToFile(databuf, pThreadInfo); totalLen = 0; memset(databuf, 0, 100*1024*1024); } @@ -1160,9 +1165,9 @@ static void fetchResult(TAOS_RES *res, char* resultFile) { } verbosePrint("%s() LN%d, databuf=%s resultFile=%s\n", - __func__, __LINE__, databuf, resultFile); - if ((resultFile) && (strlen(resultFile) > 0)) { - appendResultBufToFile(databuf, resultFile); + __func__, __LINE__, databuf, pThreadInfo->filePath); + if (strlen(pThreadInfo->filePath) > 0) { + appendResultBufToFile(databuf, pThreadInfo); } free(databuf); } @@ -1179,14 +1184,14 @@ static void selectAndGetResult( return; } - fetchResult(res, pThreadInfo->fp); + fetchResult(res, pThreadInfo); taos_free_result(res); } else if (0 == strncasecmp(g_queryInfo.queryMode, "rest", strlen("rest"))) { int retCode = postProceSql( g_queryInfo.host, &(g_queryInfo.serv_addr), g_queryInfo.port, command, - pThreadInfo->fp); + pThreadInfo); if (0 != retCode) { printf("====restful return fail, threadID[%d]\n", pThreadInfo->threadID); } @@ -1719,7 +1724,7 @@ static void printfQueryMeta() { printf("query interval: \033[33m%"PRIu64" ms\033[0m\n", g_queryInfo.specifiedQueryInfo.queryInterval); printf("top query times:\033[33m%"PRIu64"\033[0m\n", g_args.query_times); - printf("concurrent: \033[33m%"PRIu64"\033[0m\n", + printf("concurrent: \033[33m%d\033[0m\n", g_queryInfo.specifiedQueryInfo.concurrent); printf("mod: \033[33m%s\033[0m\n", (g_queryInfo.specifiedQueryInfo.asyncMode)?"async":"sync"); @@ -2058,7 +2063,7 @@ static void printfQuerySystemInfo(TAOS * taos) { } static int postProceSql(char *host, struct sockaddr_in *pServAddr, uint16_t port, - char* sqlstr, char *resultFile) + char* sqlstr, threadInfo *pThreadInfo) { char *req_fmt = "POST %s HTTP/1.1\r\nHost: %s:%d\r\nAccept: */*\r\nAuthorization: Basic %s\r\nContent-Length: %d\r\nContent-Type: application/x-www-form-urlencoded\r\n\r\n%s"; @@ -2194,8 +2199,8 @@ static int postProceSql(char *host, struct sockaddr_in *pServAddr, uint16_t port response_buf[RESP_BUF_LEN - 1] = '\0'; printf("Response:\n%s\n", response_buf); - if ((resultFile) && (strlen(resultFile) > 0)) { - appendResultBufToFile(response_buf, resultFile); + if (strlen(pThreadInfo->filePath) > 0) { + appendResultBufToFile(response_buf, pThreadInfo); } free(request_buf); @@ -2958,18 +2963,18 @@ static int startMultiThreadCreateChildTable( b = ntables % threads; for (int64_t i = 0; i < threads; i++) { - threadInfo *t_info = infos + i; - t_info->threadID = i; - tstrncpy(t_info->db_name, db_name, MAX_DB_NAME_SIZE); - t_info->superTblInfo = superTblInfo; + threadInfo *pThreadInfo = infos + i; + pThreadInfo->threadID = i; + tstrncpy(pThreadInfo->db_name, db_name, MAX_DB_NAME_SIZE); + pThreadInfo->superTblInfo = superTblInfo; verbosePrint("%s() %d db_name: %s\n", __func__, __LINE__, db_name); - t_info->taos = taos_connect( + pThreadInfo->taos = taos_connect( g_Dbs.host, g_Dbs.user, g_Dbs.password, db_name, g_Dbs.port); - if (t_info->taos == NULL) { + if (pThreadInfo->taos == NULL) { errorPrint( "%s() LN%d, Failed to connect to TDengine, reason:%s\n", __func__, __LINE__, taos_errstr(NULL)); free(pids); @@ -2977,14 +2982,14 @@ static int startMultiThreadCreateChildTable( return -1; } - t_info->start_table_from = startFrom; - t_info->ntables = iend_table_to = i < b ? startFrom + a : startFrom + a - 1; - startFrom = t_info->end_table_to + 1; - t_info->use_metric = true; - t_info->cols = cols; - t_info->minDelay = UINT64_MAX; - pthread_create(pids + i, NULL, createTable, t_info); + pThreadInfo->start_table_from = startFrom; + pThreadInfo->ntables = iend_table_to = i < b ? startFrom + a : startFrom + a - 1; + startFrom = pThreadInfo->end_table_to + 1; + pThreadInfo->use_metric = true; + pThreadInfo->cols = cols; + pThreadInfo->minDelay = UINT64_MAX; + pthread_create(pids + i, NULL, createTable, pThreadInfo); } for (int i = 0; i < threads; i++) { @@ -2992,8 +2997,8 @@ static int startMultiThreadCreateChildTable( } for (int i = 0; i < threads; i++) { - threadInfo *t_info = infos + i; - taos_close(t_info->taos); + threadInfo *pThreadInfo = infos + i; + taos_close(pThreadInfo->taos); } free(pids); @@ -4198,7 +4203,7 @@ static bool getMetaFromQueryJsonFile(cJSON* root) { if (concurrent && concurrent->type == cJSON_Number) { if (concurrent->valueint <= 0) { errorPrint( - "%s() LN%d, query sqlCount %"PRIu64" or concurrent %"PRIu64" is not correct.\n", + "%s() LN%d, query sqlCount %"PRIu64" or concurrent %d is not correct.\n", __func__, __LINE__, g_queryInfo.specifiedQueryInfo.sqlCount, g_queryInfo.specifiedQueryInfo.concurrent); @@ -4265,24 +4270,28 @@ static bool getMetaFromQueryJsonFile(cJSON* root) { } // sqls - cJSON* superSqls = cJSON_GetObjectItem(specifiedQuery, "sqls"); - if (!superSqls) { + cJSON* specifiedSqls = cJSON_GetObjectItem(specifiedQuery, "sqls"); + if (!specifiedSqls) { g_queryInfo.specifiedQueryInfo.sqlCount = 0; - } else if (superSqls->type != cJSON_Array) { + } else if (specifiedSqls->type != cJSON_Array) { errorPrint("%s() LN%d, failed to read json, super sqls not found\n", __func__, __LINE__); goto PARSE_OVER; } else { - int superSqlSize = cJSON_GetArraySize(superSqls); - if (superSqlSize > MAX_QUERY_SQL_COUNT) { - errorPrint("%s() LN%d, failed to read json, query sql size overflow, max is %d\n", - __func__, __LINE__, MAX_QUERY_SQL_COUNT); + int superSqlSize = cJSON_GetArraySize(specifiedSqls); + if (superSqlSize * g_queryInfo.specifiedQueryInfo.concurrent + > MAX_QUERY_SQL_COUNT) { + errorPrint("%s() LN%d, failed to read json, query sql(%d) * concurrent(%d) overflow, max is %d\n", + __func__, __LINE__, + superSqlSize, + g_queryInfo.specifiedQueryInfo.concurrent, + MAX_QUERY_SQL_COUNT); goto PARSE_OVER; } g_queryInfo.specifiedQueryInfo.sqlCount = superSqlSize; for (int j = 0; j < superSqlSize; ++j) { - cJSON* sql = cJSON_GetArrayItem(superSqls, j); + cJSON* sql = cJSON_GetArrayItem(specifiedSqls, j); if (sql == NULL) continue; cJSON *sqlStr = cJSON_GetObjectItem(sql, "sql"); @@ -4458,16 +4467,16 @@ static bool getMetaFromQueryJsonFile(cJSON* root) { g_queryInfo.superQueryInfo.resubAfterConsume = 1; } - // sqls - cJSON* subsqls = cJSON_GetObjectItem(superQuery, "sqls"); - if (!subsqls) { + // supert table sqls + cJSON* superSqls = cJSON_GetObjectItem(superQuery, "sqls"); + if (!superSqls) { g_queryInfo.superQueryInfo.sqlCount = 0; - } else if (subsqls->type != cJSON_Array) { + } else if (superSqls->type != cJSON_Array) { errorPrint("%s() LN%d: failed to read json, super sqls not found\n", __func__, __LINE__); goto PARSE_OVER; } else { - int superSqlSize = cJSON_GetArraySize(subsqls); + int superSqlSize = cJSON_GetArraySize(superSqls); if (superSqlSize > MAX_QUERY_SQL_COUNT) { errorPrint("%s() LN%d, failed to read json, query sql size overflow, max is %d\n", __func__, __LINE__, MAX_QUERY_SQL_COUNT); @@ -4476,7 +4485,7 @@ static bool getMetaFromQueryJsonFile(cJSON* root) { g_queryInfo.superQueryInfo.sqlCount = superSqlSize; for (int j = 0; j < superSqlSize; ++j) { - cJSON* sql = cJSON_GetArrayItem(subsqls, j); + cJSON* sql = cJSON_GetArrayItem(superSqls, j); if (sql == NULL) continue; cJSON *sqlStr = cJSON_GetObjectItem(sql, "sql"); @@ -5822,49 +5831,49 @@ static void startMultiThreadInsertData(int threads, char* db_name, } for (int i = 0; i < threads; i++) { - threadInfo *t_info = infos + i; - t_info->threadID = i; - tstrncpy(t_info->db_name, db_name, MAX_DB_NAME_SIZE); - t_info->time_precision = timePrec; - t_info->superTblInfo = superTblInfo; + threadInfo *pThreadInfo = infos + i; + pThreadInfo->threadID = i; + tstrncpy(pThreadInfo->db_name, db_name, MAX_DB_NAME_SIZE); + pThreadInfo->time_precision = timePrec; + pThreadInfo->superTblInfo = superTblInfo; - t_info->start_time = start_time; - t_info->minDelay = UINT64_MAX; + pThreadInfo->start_time = start_time; + pThreadInfo->minDelay = UINT64_MAX; if ((NULL == superTblInfo) || (0 == strncasecmp(superTblInfo->insertMode, "taosc", 5))) { - //t_info->taos = taos; - t_info->taos = taos_connect( + //pThreadInfo->taos = taos; + pThreadInfo->taos = taos_connect( g_Dbs.host, g_Dbs.user, g_Dbs.password, db_name, g_Dbs.port); - if (NULL == t_info->taos) { + if (NULL == pThreadInfo->taos) { errorPrint( "connect to server fail from insert sub thread, reason: %s\n", taos_errstr(NULL)); exit(-1); } } else { - t_info->taos = NULL; + pThreadInfo->taos = NULL; } /* if ((NULL == superTblInfo) || (0 == superTblInfo->multiThreadWriteOneTbl)) { */ - t_info->start_table_from = startFrom; - t_info->ntables = iend_table_to = i < b ? startFrom + a : startFrom + a - 1; - startFrom = t_info->end_table_to + 1; + pThreadInfo->start_table_from = startFrom; + pThreadInfo->ntables = iend_table_to = i < b ? startFrom + a : startFrom + a - 1; + startFrom = pThreadInfo->end_table_to + 1; /* } else { - t_info->start_table_from = 0; - t_info->ntables = superTblInfo->childTblCount; - t_info->start_time = t_info->start_time + rand_int() % 10000 - rand_tinyint(); + pThreadInfo->start_table_from = 0; + pThreadInfo->ntables = superTblInfo->childTblCount; + pThreadInfo->start_time = pThreadInfo->start_time + rand_int() % 10000 - rand_tinyint(); } */ - tsem_init(&(t_info->lock_sem), 0, 0); + tsem_init(&(pThreadInfo->lock_sem), 0, 0); if (ASYNC_MODE == g_Dbs.asyncMode) { - pthread_create(pids + i, NULL, asyncWrite, t_info); + pthread_create(pids + i, NULL, asyncWrite, pThreadInfo); } else { - pthread_create(pids + i, NULL, syncWrite, t_info); + pthread_create(pids + i, NULL, syncWrite, pThreadInfo); } } @@ -5879,27 +5888,27 @@ static void startMultiThreadInsertData(int threads, char* db_name, double avgDelay = 0; for (int i = 0; i < threads; i++) { - threadInfo *t_info = infos + i; + threadInfo *pThreadInfo = infos + i; - tsem_destroy(&(t_info->lock_sem)); - taos_close(t_info->taos); + tsem_destroy(&(pThreadInfo->lock_sem)); + taos_close(pThreadInfo->taos); debugPrint("%s() LN%d, [%d] totalInsert=%"PRIu64" totalAffected=%"PRIu64"\n", __func__, __LINE__, - t_info->threadID, t_info->totalInsertRows, - t_info->totalAffectedRows); + pThreadInfo->threadID, pThreadInfo->totalInsertRows, + pThreadInfo->totalAffectedRows); if (superTblInfo) { - superTblInfo->totalAffectedRows += t_info->totalAffectedRows; - superTblInfo->totalInsertRows += t_info->totalInsertRows; + superTblInfo->totalAffectedRows += pThreadInfo->totalAffectedRows; + superTblInfo->totalInsertRows += pThreadInfo->totalInsertRows; } else { - g_args.totalAffectedRows += t_info->totalAffectedRows; - g_args.totalInsertRows += t_info->totalInsertRows; + g_args.totalAffectedRows += pThreadInfo->totalAffectedRows; + g_args.totalInsertRows += pThreadInfo->totalInsertRows; } - totalDelay += t_info->totalDelay; - cntDelay += t_info->cntDelay; - if (t_info->maxDelay > maxDelay) maxDelay = t_info->maxDelay; - if (t_info->minDelay < minDelay) minDelay = t_info->minDelay; + totalDelay += pThreadInfo->totalDelay; + cntDelay += pThreadInfo->cntDelay; + if (pThreadInfo->maxDelay > maxDelay) maxDelay = pThreadInfo->maxDelay; + if (pThreadInfo->minDelay < minDelay) minDelay = pThreadInfo->minDelay; } cntDelay -= 1; @@ -5955,26 +5964,26 @@ static void startMultiThreadInsertData(int threads, char* db_name, static void *readTable(void *sarg) { #if 1 - threadInfo *rinfo = (threadInfo *)sarg; - TAOS *taos = rinfo->taos; + threadInfo *pThreadInfo = (threadInfo *)sarg; + TAOS *taos = pThreadInfo->taos; char command[BUFFER_SIZE] = "\0"; - uint64_t sTime = rinfo->start_time; - char *tb_prefix = rinfo->tb_prefix; - FILE *fp = fopen(rinfo->fp, "a"); + uint64_t sTime = pThreadInfo->start_time; + char *tb_prefix = pThreadInfo->tb_prefix; + FILE *fp = fopen(pThreadInfo->filePath, "a"); if (NULL == fp) { - errorPrint( "fopen %s fail, reason:%s.\n", rinfo->fp, strerror(errno)); + errorPrint( "fopen %s fail, reason:%s.\n", pThreadInfo->filePath, strerror(errno)); return NULL; } int64_t num_of_DPT; -/* if (rinfo->superTblInfo) { - num_of_DPT = rinfo->superTblInfo->insertRows; // nrecords_per_table; +/* if (pThreadInfo->superTblInfo) { + num_of_DPT = pThreadInfo->superTblInfo->insertRows; // nrecords_per_table; } else { */ num_of_DPT = g_args.num_of_DPT; // } - int64_t num_of_tables = rinfo->ntables; // rinfo->end_table_to - rinfo->start_table_from + 1; + int64_t num_of_tables = pThreadInfo->ntables; // rinfo->end_table_to - rinfo->start_table_from + 1; int64_t totalData = num_of_DPT * num_of_tables; bool do_aggreFunc = g_Dbs.do_aggreFunc; @@ -6027,17 +6036,17 @@ static void *readTable(void *sarg) { static void *readMetric(void *sarg) { #if 1 - threadInfo *rinfo = (threadInfo *)sarg; - TAOS *taos = rinfo->taos; + threadInfo *pThreadInfo = (threadInfo *)sarg; + TAOS *taos = pThreadInfo->taos; char command[BUFFER_SIZE] = "\0"; - FILE *fp = fopen(rinfo->fp, "a"); + FILE *fp = fopen(pThreadInfo->filePath, "a"); if (NULL == fp) { - printf("fopen %s fail, reason:%s.\n", rinfo->fp, strerror(errno)); + printf("fopen %s fail, reason:%s.\n", pThreadInfo->filePath, strerror(errno)); return NULL; } - int64_t num_of_DPT = rinfo->superTblInfo->insertRows; - int64_t num_of_tables = rinfo->ntables; // rinfo->end_table_to - rinfo->start_table_from + 1; + int64_t num_of_DPT = pThreadInfo->superTblInfo->insertRows; + int64_t num_of_tables = pThreadInfo->ntables; // rinfo->end_table_to - rinfo->start_table_from + 1; int64_t totalData = num_of_DPT * num_of_tables; bool do_aggreFunc = g_Dbs.do_aggreFunc; @@ -6236,8 +6245,8 @@ static void *specifiedTableQuery(void *sarg) { uint64_t lastPrintTime = taosGetTimestampMs(); uint64_t startTs = taosGetTimestampMs(); - if (g_queryInfo.specifiedQueryInfo.result[pThreadInfo->querySeq][0] != 0) { - sprintf(pThreadInfo->fp, "%s-%d", + if (g_queryInfo.specifiedQueryInfo.result[pThreadInfo->querySeq] != NULL) { + sprintf(pThreadInfo->filePath, "%s-%d", g_queryInfo.specifiedQueryInfo.result[pThreadInfo->querySeq], pThreadInfo->threadID); } @@ -6337,8 +6346,8 @@ static void *superTableQuery(void *sarg) { for (int j = 0; j < g_queryInfo.superQueryInfo.sqlCount; j++) { memset(sqlstr,0,sizeof(sqlstr)); replaceChildTblName(g_queryInfo.superQueryInfo.sql[j], sqlstr, i); - if (g_queryInfo.superQueryInfo.result[j][0] != 0) { - sprintf(pThreadInfo->fp, "%s-%d", + if (g_queryInfo.superQueryInfo.result[j] != NULL) { + sprintf(pThreadInfo->filePath, "%s-%d", g_queryInfo.superQueryInfo.result[j], pThreadInfo->threadID); } @@ -6428,9 +6437,9 @@ static int queryTestProcess() { for (uint64_t i = 0; i < nSqlCount; i++) { for (int j = 0; j < nConcurrent; j++) { uint64_t seq = i * nConcurrent + j; - threadInfo *t_info = infos + seq; - t_info->threadID = seq; - t_info->querySeq = i; + threadInfo *pThreadInfo = infos + seq; + pThreadInfo->threadID = seq; + pThreadInfo->querySeq = i; if (0 == strncasecmp(g_queryInfo.queryMode, "taosc", 5)) { @@ -6447,10 +6456,10 @@ static int queryTestProcess() { } } - t_info->taos = NULL;// TODO: workaround to use separate taos connection; + pThreadInfo->taos = NULL;// TODO: workaround to use separate taos connection; pthread_create(pids + seq, NULL, specifiedTableQuery, - t_info); + pThreadInfo); } } } else { @@ -6490,15 +6499,15 @@ static int queryTestProcess() { uint64_t startFrom = 0; for (int i = 0; i < threads; i++) { - threadInfo *t_info = infosOfSub + i; - t_info->threadID = i; + threadInfo *pThreadInfo = infosOfSub + i; + pThreadInfo->threadID = i; - t_info->start_table_from = startFrom; - t_info->ntables = iend_table_to = i < b ? startFrom + a : startFrom + a - 1; - startFrom = t_info->end_table_to + 1; - t_info->taos = NULL; // TODO: workaround to use separate taos connection; - pthread_create(pidsOfSub + i, NULL, superTableQuery, t_info); + pThreadInfo->start_table_from = startFrom; + pThreadInfo->ntables = iend_table_to = i < b ? startFrom + a : startFrom + a - 1; + startFrom = pThreadInfo->end_table_to + 1; + pThreadInfo->taos = NULL; // TODO: workaround to use separate taos connection; + pthread_create(pidsOfSub + i, NULL, superTableQuery, pThreadInfo); } g_queryInfo.superQueryInfo.threadCnt = threads; @@ -6545,7 +6554,7 @@ static void stable_sub_callback( } if (param) - fetchResult(res, ((threadInfo *)param)->fp); + fetchResult(res, (threadInfo *)param); // tao_unscribe() will free result. } @@ -6558,7 +6567,7 @@ static void specified_sub_callback( } if (param) - fetchResult(res, ((threadInfo *)param)->fp); + fetchResult(res, (threadInfo *)param); // tao_unscribe() will free result. } @@ -6612,18 +6621,15 @@ static void *superSubscribe(void *sarg) { } if (pThreadInfo->taos == NULL) { - TAOS * taos = NULL; - taos = taos_connect(g_queryInfo.host, + pThreadInfo->taos = taos_connect(g_queryInfo.host, g_queryInfo.user, g_queryInfo.password, g_queryInfo.dbName, g_queryInfo.port); - if (taos == NULL) { + if (pThreadInfo->taos == NULL) { errorPrint("[%d] Failed to connect to TDengine, reason:%s\n", pThreadInfo->threadID, taos_errstr(NULL)); return NULL; - } else { - pThreadInfo->taos = taos; } } @@ -6653,7 +6659,7 @@ static void *superSubscribe(void *sarg) { g_queryInfo.superQueryInfo.sql[pThreadInfo->querySeq], subSqlstr, i); if (g_queryInfo.superQueryInfo.result[pThreadInfo->querySeq][0] != 0) { - sprintf(pThreadInfo->fp, "%s-%d", + sprintf(pThreadInfo->filePath, "%s-%d", g_queryInfo.superQueryInfo.result[pThreadInfo->querySeq], pThreadInfo->threadID); } @@ -6696,16 +6702,16 @@ static void *superSubscribe(void *sarg) { if (res) { if (g_queryInfo.superQueryInfo.result[pThreadInfo->querySeq][0] != 0) { - sprintf(pThreadInfo->fp, "%s-%d", + sprintf(pThreadInfo->filePath, "%s-%d", g_queryInfo.superQueryInfo.result[pThreadInfo->querySeq], pThreadInfo->threadID); - fetchResult(res, pThreadInfo->fp); + fetchResult(res, pThreadInfo); } if (g_queryInfo.superQueryInfo.result[pThreadInfo->querySeq][0] != 0) { - sprintf(pThreadInfo->fp, "%s-%d", + sprintf(pThreadInfo->filePath, "%s-%d", g_queryInfo.superQueryInfo.result[pThreadInfo->querySeq], pThreadInfo->threadID); - fetchResult(res, pThreadInfo->fp); + fetchResult(res, pThreadInfo); } consumed[tsubSeq] ++; @@ -6746,21 +6752,18 @@ static void *superSubscribe(void *sarg) { static void *specifiedSubscribe(void *sarg) { threadInfo *pThreadInfo = (threadInfo *)sarg; - TAOS_SUB* tsub = NULL; +// TAOS_SUB* tsub = NULL; if (pThreadInfo->taos == NULL) { - TAOS * taos = NULL; - taos = taos_connect(g_queryInfo.host, + pThreadInfo->taos = taos_connect(g_queryInfo.host, g_queryInfo.user, g_queryInfo.password, g_queryInfo.dbName, g_queryInfo.port); - if (taos == NULL) { + if (pThreadInfo->taos == NULL) { errorPrint("[%d] Failed to connect to TDengine, reason:%s\n", pThreadInfo->threadID, taos_errstr(NULL)); return NULL; - } else { - pThreadInfo->taos = taos; } } @@ -6772,69 +6775,70 @@ static void *specifiedSubscribe(void *sarg) { return NULL; } - char topic[32] = {0}; - sprintf(topic, "taosdemo-subscribe-%"PRIu64"", pThreadInfo->querySeq); - if (g_queryInfo.specifiedQueryInfo.result[pThreadInfo->querySeq][0] != 0) { - sprintf(pThreadInfo->fp, "%s-%d", + sprintf(g_queryInfo.specifiedQueryInfo.topic[pThreadInfo->threadID], + "taosdemo-subscribe-%"PRIu64"-%d", + pThreadInfo->querySeq, + pThreadInfo->threadID); + if (g_queryInfo.specifiedQueryInfo.result[pThreadInfo->querySeq] != NULL) { + sprintf(pThreadInfo->filePath, "%s-%d", g_queryInfo.specifiedQueryInfo.result[pThreadInfo->querySeq], pThreadInfo->threadID); } - tsub = subscribeImpl( - SPECIFIED_CLASS, pThreadInfo, + g_queryInfo.specifiedQueryInfo.tsub[pThreadInfo->threadID] = subscribeImpl( + SPECIFIED_CLASS, pThreadInfo, g_queryInfo.specifiedQueryInfo.sql[pThreadInfo->querySeq], - topic, + g_queryInfo.specifiedQueryInfo.topic[pThreadInfo->threadID], g_queryInfo.specifiedQueryInfo.subscribeRestart, g_queryInfo.specifiedQueryInfo.subscribeInterval); - if (NULL == tsub) { + if (NULL == g_queryInfo.specifiedQueryInfo.tsub[pThreadInfo->threadID]) { taos_close(pThreadInfo->taos); return NULL; } // start loop to consume result - TAOS_RES* res = NULL; - - int consumed; + g_queryInfo.specifiedQueryInfo.consumed[pThreadInfo->threadID] = 0; while(1) { if (ASYNC_MODE == g_queryInfo.specifiedQueryInfo.asyncMode) { continue; } - res = taos_consume(tsub); - if (res) { + g_queryInfo.specifiedQueryInfo.res[pThreadInfo->threadID] = taos_consume( + g_queryInfo.specifiedQueryInfo.tsub[pThreadInfo->threadID]); + if (g_queryInfo.specifiedQueryInfo.res[pThreadInfo->threadID]) { if (g_queryInfo.specifiedQueryInfo.result[pThreadInfo->querySeq][0] != 0) { - sprintf(pThreadInfo->fp, "%s-%d", + sprintf(pThreadInfo->filePath, "%s-%d", g_queryInfo.specifiedQueryInfo.result[pThreadInfo->querySeq], pThreadInfo->threadID); - fetchResult(res, pThreadInfo->fp); + fetchResult(g_queryInfo.specifiedQueryInfo.res[pThreadInfo->threadID], pThreadInfo); } - consumed ++; + g_queryInfo.specifiedQueryInfo.consumed[pThreadInfo->threadID] ++; if ((g_queryInfo.specifiedQueryInfo.subscribeKeepProgress) - && (consumed >= + && (g_queryInfo.specifiedQueryInfo.consumed[pThreadInfo->threadID] >= g_queryInfo.specifiedQueryInfo.resubAfterConsume[pThreadInfo->querySeq])) { printf("keepProgress:%d, resub specified query: %"PRIu64"\n", g_queryInfo.specifiedQueryInfo.subscribeKeepProgress, pThreadInfo->querySeq); - consumed = 0; - taos_unsubscribe(tsub, + g_queryInfo.specifiedQueryInfo.consumed[pThreadInfo->threadID] = 0; + taos_unsubscribe(g_queryInfo.specifiedQueryInfo.tsub[pThreadInfo->threadID], g_queryInfo.specifiedQueryInfo.subscribeKeepProgress); - tsub = subscribeImpl( + g_queryInfo.specifiedQueryInfo.tsub[pThreadInfo->threadID] = subscribeImpl( SPECIFIED_CLASS, pThreadInfo, g_queryInfo.specifiedQueryInfo.sql[pThreadInfo->querySeq], - topic, + g_queryInfo.specifiedQueryInfo.topic[pThreadInfo->threadID], g_queryInfo.specifiedQueryInfo.subscribeRestart, g_queryInfo.specifiedQueryInfo.subscribeInterval); - if (NULL == tsub) { + if (NULL == g_queryInfo.specifiedQueryInfo.tsub[pThreadInfo->threadID]) { taos_close(pThreadInfo->taos); return NULL; } } } } - taos_free_result(res); - taos_unsubscribe(tsub, 0); + taos_free_result(g_queryInfo.specifiedQueryInfo.res[pThreadInfo->threadID]); + taos_unsubscribe(g_queryInfo.specifiedQueryInfo.tsub[pThreadInfo->querySeq], 0); taos_close(pThreadInfo->taos); return NULL; @@ -6904,11 +6908,11 @@ static int subscribeTestProcess() { for (int i = 0; i < g_queryInfo.specifiedQueryInfo.sqlCount; i++) { for (int j = 0; j < g_queryInfo.specifiedQueryInfo.concurrent; j++) { uint64_t seq = i * g_queryInfo.specifiedQueryInfo.concurrent + j; - threadInfo *t_info = infos + seq; - t_info->threadID = seq; - t_info->querySeq = i; - t_info->taos = NULL; // TODO: workaround to use separate taos connection; - pthread_create(pids + seq, NULL, specifiedSubscribe, t_info); + threadInfo *pThreadInfo = infos + seq; + pThreadInfo->threadID = seq; + pThreadInfo->querySeq = i; + pThreadInfo->taos = NULL; // TODO: workaround to use separate taos connection; + pthread_create(pids + seq, NULL, specifiedSubscribe, pThreadInfo); } } } @@ -6954,17 +6958,17 @@ static int subscribeTestProcess() { uint64_t startFrom = 0; for (int j = 0; j < threads; j++) { uint64_t seq = i * threads + j; - threadInfo *t_info = infosOfStable + seq; - t_info->threadID = seq; - t_info->querySeq = i; + threadInfo *pThreadInfo = infosOfStable + seq; + pThreadInfo->threadID = seq; + pThreadInfo->querySeq = i; - t_info->start_table_from = startFrom; - t_info->ntables = jend_table_to = jend_table_to + 1; - t_info->taos = NULL; // TODO: workaround to use separate taos connection; + pThreadInfo->start_table_from = startFrom; + pThreadInfo->ntables = jend_table_to = jend_table_to + 1; + pThreadInfo->taos = NULL; // TODO: workaround to use separate taos connection; pthread_create(pidsOfStable + seq, - NULL, superSubscribe, t_info); + NULL, superSubscribe, pThreadInfo); } } @@ -7243,47 +7247,47 @@ static void queryResult() { // query data pthread_t read_id; - threadInfo *rInfo = malloc(sizeof(threadInfo)); - assert(rInfo); - rInfo->start_time = 1500000000000; // 2017-07-14 10:40:00.000 - rInfo->start_table_from = 0; + threadInfo *pThreadInfo = malloc(sizeof(threadInfo)); + assert(pThreadInfo); + pThreadInfo->start_time = 1500000000000; // 2017-07-14 10:40:00.000 + pThreadInfo->start_table_from = 0; - //rInfo->do_aggreFunc = g_Dbs.do_aggreFunc; + //pThreadInfo->do_aggreFunc = g_Dbs.do_aggreFunc; if (g_args.use_metric) { - rInfo->ntables = g_Dbs.db[0].superTbls[0].childTblCount; - rInfo->end_table_to = g_Dbs.db[0].superTbls[0].childTblCount - 1; - rInfo->superTblInfo = &g_Dbs.db[0].superTbls[0]; - tstrncpy(rInfo->tb_prefix, + pThreadInfo->ntables = g_Dbs.db[0].superTbls[0].childTblCount; + pThreadInfo->end_table_to = g_Dbs.db[0].superTbls[0].childTblCount - 1; + pThreadInfo->superTblInfo = &g_Dbs.db[0].superTbls[0]; + tstrncpy(pThreadInfo->tb_prefix, g_Dbs.db[0].superTbls[0].childTblPrefix, MAX_TB_NAME_SIZE); } else { - rInfo->ntables = g_args.num_of_tables; - rInfo->end_table_to = g_args.num_of_tables -1; - tstrncpy(rInfo->tb_prefix, g_args.tb_prefix, MAX_TB_NAME_SIZE); + pThreadInfo->ntables = g_args.num_of_tables; + pThreadInfo->end_table_to = g_args.num_of_tables -1; + tstrncpy(pThreadInfo->tb_prefix, g_args.tb_prefix, MAX_TB_NAME_SIZE); } - rInfo->taos = taos_connect( + pThreadInfo->taos = taos_connect( g_Dbs.host, g_Dbs.user, g_Dbs.password, g_Dbs.db[0].dbName, g_Dbs.port); - if (rInfo->taos == NULL) { + if (pThreadInfo->taos == NULL) { errorPrint( "Failed to connect to TDengine, reason:%s\n", taos_errstr(NULL)); - free(rInfo); + free(pThreadInfo); exit(-1); } - tstrncpy(rInfo->fp, g_Dbs.resultFile, MAX_FILE_NAME_LEN); + tstrncpy(pThreadInfo->filePath, g_Dbs.resultFile, MAX_FILE_NAME_LEN); if (!g_Dbs.use_metric) { - pthread_create(&read_id, NULL, readTable, rInfo); + pthread_create(&read_id, NULL, readTable, pThreadInfo); } else { - pthread_create(&read_id, NULL, readMetric, rInfo); + pthread_create(&read_id, NULL, readMetric, pThreadInfo); } pthread_join(read_id, NULL); - taos_close(rInfo->taos); - free(rInfo); + taos_close(pThreadInfo->taos); + free(pThreadInfo); } static void testCmdLine() { From 6488456b86ed88ef2819a64381115c92b6d41dcb Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Tue, 25 May 2021 15:11:11 +0800 Subject: [PATCH 14/19] [td-4231]fix bug by regression test. --- src/client/src/tscSQLParser.c | 14 +++++++++----- src/client/src/tscSub.c | 6 +++--- 2 files changed, 12 insertions(+), 8 deletions(-) diff --git a/src/client/src/tscSQLParser.c b/src/client/src/tscSQLParser.c index 722ce7a231..b020432a2e 100644 --- a/src/client/src/tscSQLParser.c +++ b/src/client/src/tscSQLParser.c @@ -7291,11 +7291,15 @@ int32_t doValidateSqlNode(SSqlObj* pSql, SQuerySqlNode* pQuerySqlNode, int32_t i } if(tscQueryTags(pQueryInfo)) { - int32_t numOfCols = (int32_t) taosArrayGetSize(pQueryInfo->colList); - for(int32_t i = 0; i < numOfCols; ++i) { - SColumn* pCols = taosArrayGetP(pQueryInfo->colList, i); - if (pCols->numOfFilters > 0) { - return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg9); + SSqlExpr* pExpr1 = tscSqlExprGet(pQueryInfo, 0); + + if (pExpr1->functionId != TSDB_FUNC_TID_TAG) { + int32_t numOfCols = (int32_t)taosArrayGetSize(pQueryInfo->colList); + for (int32_t i = 0; i < numOfCols; ++i) { + SColumn* pCols = taosArrayGetP(pQueryInfo->colList, i); + if (pCols->numOfFilters > 0) { + return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg9); + } } } } diff --git a/src/client/src/tscSub.c b/src/client/src/tscSub.c index 7da98c7885..327ccd0294 100644 --- a/src/client/src/tscSub.c +++ b/src/client/src/tscSub.c @@ -215,7 +215,7 @@ static void tscProcessSubscriptionTimer(void *handle, void *tmrId) { taosTmrReset(tscProcessSubscriptionTimer, pSub->interval, pSub, tscTmr, &pSub->pTimer); } - +//TODO refactor: extract table list name not simply from the sql static SArray* getTableList( SSqlObj* pSql ) { const char* p = strstr( pSql->sqlstr, " from " ); assert(p != NULL); // we are sure this is a 'select' statement @@ -224,11 +224,11 @@ static SArray* getTableList( SSqlObj* pSql ) { SSqlObj* pNew = taos_query(pSql->pTscObj, sql); if (pNew == NULL) { - tscError("0x%"PRIx64"failed to retrieve table id: cannot create new sql object.", pSql->self); + tscError("0x%"PRIx64" failed to retrieve table id: cannot create new sql object.", pSql->self); return NULL; } else if (taos_errno(pNew) != TSDB_CODE_SUCCESS) { - tscError("0x%"PRIx64"failed to retrieve table id,error: %s", pSql->self, tstrerror(taos_errno(pNew))); + tscError("0x%"PRIx64" failed to retrieve table id,error: %s", pSql->self, tstrerror(taos_errno(pNew))); return NULL; } From 3da561521fdc2a42ee33d41138576a7ae8e16ae4 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Tue, 25 May 2021 20:37:58 +0800 Subject: [PATCH 15/19] [TD-4323]: the deleted vnode does not need to commit, so as to speed up the deletion and avoid crash while balance --- src/vnode/src/vnodeMain.c | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/vnode/src/vnodeMain.c b/src/vnode/src/vnodeMain.c index ee28be3d2f..f4515b6688 100644 --- a/src/vnode/src/vnodeMain.c +++ b/src/vnode/src/vnodeMain.c @@ -419,7 +419,11 @@ void vnodeDestroy(SVnodeObj *pVnode) { } if (pVnode->tsdb) { - code = tsdbCloseRepo(pVnode->tsdb, 1); + // the deleted vnode does not need to commit, so as to speed up the deletion + int toCommit = 1; + if (pVnode->dropped) toCommit = 0; + + code = tsdbCloseRepo(pVnode->tsdb, toCommit); pVnode->tsdb = NULL; } From 29746411b57a7949fbd46c631ed4fb4aa730b9e9 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Tue, 25 May 2021 22:19:27 +0800 Subject: [PATCH 16/19] [td-225] add log. --- src/client/src/tscSQLParser.c | 2 -- src/client/src/tscServer.c | 8 ++++---- src/mnode/src/mnodeTable.c | 4 ++-- 3 files changed, 6 insertions(+), 8 deletions(-) diff --git a/src/client/src/tscSQLParser.c b/src/client/src/tscSQLParser.c index b020432a2e..c472b08dc0 100644 --- a/src/client/src/tscSQLParser.c +++ b/src/client/src/tscSQLParser.c @@ -6494,7 +6494,6 @@ int32_t doCheckForCreateFromStable(SSqlObj* pSql, SSqlInfo* pInfo) { size_t valSize = taosArrayGetSize(pValList); - // too long tag values will return invalid sql, not be truncated automatically SSchema *pTagSchema = tscGetTableTagSchema(pStableMetaInfo->pTableMeta); STagData *pTag = &pCreateTableInfo->tagdata; @@ -6504,7 +6503,6 @@ int32_t doCheckForCreateFromStable(SSqlObj* pSql, SSqlInfo* pInfo) { return TSDB_CODE_TSC_OUT_OF_MEMORY; } - SArray* pNameList = NULL; size_t nameSize = 0; int32_t schemaSize = tscGetNumOfTags(pStableMetaInfo->pTableMeta); diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c index 0e16369cad..648b95180e 100644 --- a/src/client/src/tscServer.c +++ b/src/client/src/tscServer.c @@ -2020,8 +2020,9 @@ int tscProcessTableMetaRsp(SSqlObj *pSql) { } } - tscDebug("0x%"PRIx64" recv table meta, uid:%" PRIu64 ", tid:%d, name:%s", pSql->self, pTableMeta->id.uid, pTableMeta->id.tid, - tNameGetTableName(&pTableMetaInfo->name)); + tscDebug("0x%"PRIx64" recv table meta, uid:%" PRIu64 ", tid:%d, name:%s, numOfCols:%d, numOfTags:%d", pSql->self, + pTableMeta->id.uid, pTableMeta->id.tid, tNameGetTableName(&pTableMetaInfo->name), pTableMeta->tableInfo.numOfColumns, + pTableMeta->tableInfo.numOfTags); free(pTableMeta); return TSDB_CODE_SUCCESS; @@ -2164,8 +2165,7 @@ int tscProcessSTableVgroupRsp(SSqlObj *pSql) { pInfo->vgroupList->numOfVgroups = pVgroupMsg->numOfVgroups; if (pInfo->vgroupList->numOfVgroups <= 0) { - //tfree(pInfo->vgroupList); - tscError("0x%"PRIx64" empty vgroup info", pSql->self); + tscDebug("0x%"PRIx64" empty vgroup info", pSql->self); } else { for (int32_t j = 0; j < pInfo->vgroupList->numOfVgroups; ++j) { // just init, no need to lock diff --git a/src/mnode/src/mnodeTable.c b/src/mnode/src/mnodeTable.c index 2a8e941fcb..f098356e5c 100644 --- a/src/mnode/src/mnodeTable.c +++ b/src/mnode/src/mnodeTable.c @@ -1189,8 +1189,8 @@ static int32_t mnodeFindSuperTableTagIndex(SSTableObj *pStable, const char *tagN static int32_t mnodeAddSuperTableTagCb(SMnodeMsg *pMsg, int32_t code) { SSTableObj *pStable = (SSTableObj *)pMsg->pTable; - mLInfo("msg:%p, app:%p stable %s, add tag result:%s", pMsg, pMsg->rpcMsg.ahandle, pStable->info.tableId, - tstrerror(code)); + mLInfo("msg:%p, app:%p stable %s, add tag result:%s, numOfTags:%d", pMsg, pMsg->rpcMsg.ahandle, pStable->info.tableId, + tstrerror(code), pStable->numOfTags); return code; } From 04d464bdc92d3d62acbca50f3babf94e19036f30 Mon Sep 17 00:00:00 2001 From: Shuduo Sang Date: Wed, 26 May 2021 09:34:45 +0800 Subject: [PATCH 17/19] Hotfix/sangshuduo/td 3783 taosdemo invalid rand string (#6228) * [TD-3783]: taosdemo for windows generates rand string invalid. * refactor disorder timestamp. * refactor debug print. Co-authored-by: Shuduo Sang --- src/kit/taosdemo/taosdemo.c | 125 +++++++++++++++++------------------- 1 file changed, 58 insertions(+), 67 deletions(-) diff --git a/src/kit/taosdemo/taosdemo.c b/src/kit/taosdemo/taosdemo.c index b162be7739..54dd68de44 100644 --- a/src/kit/taosdemo/taosdemo.c +++ b/src/kit/taosdemo/taosdemo.c @@ -538,6 +538,8 @@ static void createChildTables(); static int queryDbExec(TAOS *taos, char *command, QUERY_TYPE type, bool quiet); static int postProceSql(char *host, struct sockaddr_in *pServAddr, uint16_t port, char* sqlstr, threadInfo *pThreadInfo); +static int64_t getTSRandTail(int64_t timeStampStep, int32_t seq, + int disorderRatio, int disorderRange); /* ************ Global variables ************ */ @@ -1097,9 +1099,9 @@ static int queryDbExec(TAOS *taos, char *command, QUERY_TYPE type, bool quiet) { } } + verbosePrint("%s() LN%d - command: %s\n", __func__, __LINE__, command); if (code != 0) { if (!quiet) { - debugPrint("%s() LN%d - command: %s\n", __func__, __LINE__, command); errorPrint("Failed to execute %s, reason: %s\n", command, taos_errstr(res)); } taos_free_result(res); @@ -2692,8 +2694,6 @@ static int createSuperTable( snprintf(command, BUFFER_SIZE, "create table if not exists %s.%s (ts timestamp%s) tags %s", dbName, superTbl->sTblName, cols, tags); - verbosePrint("%s() LN%d: %s\n", __func__, __LINE__, command); - if (0 != queryDbExec(taos, command, NO_INSERT_TYPE, false)) { errorPrint( "create supertable %s failed!\n\n", superTbl->sTblName); @@ -2716,7 +2716,6 @@ static int createDatabasesAndStables() { for (int i = 0; i < g_Dbs.dbCount; i++) { if (g_Dbs.db[i].drop) { sprintf(command, "drop database if exists %s;", g_Dbs.db[i].dbName); - verbosePrint("%s() %d command: %s\n", __func__, __LINE__, command); if (0 != queryDbExec(taos, command, NO_INSERT_TYPE, false)) { taos_close(taos); return -1; @@ -2789,7 +2788,6 @@ static int createDatabasesAndStables() { " precision \'%s\';", g_Dbs.db[i].dbCfg.precision); } - debugPrint("%s() %d command: %s\n", __func__, __LINE__, command); if (0 != queryDbExec(taos, command, NO_INSERT_TYPE, false)) { taos_close(taos); errorPrint( "\ncreate database %s failed!\n\n", g_Dbs.db[i].dbName); @@ -2806,8 +2804,6 @@ static int createDatabasesAndStables() { for (int j = 0; j < g_Dbs.db[i].superTblCount; j++) { sprintf(command, "describe %s.%s;", g_Dbs.db[i].dbName, g_Dbs.db[i].superTbls[j].sTblName); - verbosePrint("%s() %d command: %s\n", __func__, __LINE__, command); - ret = queryDbExec(taos, command, NO_INSERT_TYPE, true); if ((ret != 0) || (g_Dbs.db[i].drop)) { @@ -2911,7 +2907,6 @@ static void* createTable(void *sarg) } len = 0; - verbosePrint("%s() LN%d %s\n", __func__, __LINE__, buffer); if (0 != queryDbExec(pThreadInfo->taos, buffer, NO_INSERT_TYPE, false)){ errorPrint( "queryDbExec() failed. buffer:\n%s\n", buffer); free(buffer); @@ -2927,7 +2922,6 @@ static void* createTable(void *sarg) } if (0 != len) { - verbosePrint("%s() %d buffer: %s\n", __func__, __LINE__, buffer); if (0 != queryDbExec(pThreadInfo->taos, buffer, NO_INSERT_TYPE, false)) { errorPrint( "queryDbExec() failed. buffer:\n%s\n", buffer); } @@ -4877,6 +4871,14 @@ static int64_t generateDataTail( verbosePrint("%s() LN%d batch=%"PRIu64"\n", __func__, __LINE__, batch); + bool tsRand; + if ((superTblInfo) && (0 == strncasecmp(superTblInfo->dataSource, + "rand", strlen("rand")))) { + tsRand = true; + } else { + tsRand = false; + } + uint64_t k = 0; for (k = 0; k < batch;) { char data[MAX_DATA_SIZE]; @@ -4885,71 +4887,47 @@ static int64_t generateDataTail( int64_t retLen = 0; if (superTblInfo) { - if (0 == strncasecmp(superTblInfo->dataSource, - "sample", strlen("sample"))) { + if (tsRand) { + retLen = generateRowData( + data, + startTime + getTSRandTail( + superTblInfo->timeStampStep, k, + superTblInfo->disorderRatio, + superTblInfo->disorderRange), + superTblInfo); + } else { retLen = getRowDataFromSample( data, remainderBufLen, startTime + superTblInfo->timeStampStep * k, superTblInfo, pSamplePos); - } else if (0 == strncasecmp(superTblInfo->dataSource, - "rand", strlen("rand"))) { - - int64_t randTail = superTblInfo->timeStampStep * k; - if (superTblInfo->disorderRatio > 0) { - int rand_num = taosRandom() % 100; - if(rand_num < superTblInfo->disorderRatio) { - randTail = (randTail + (taosRandom() % superTblInfo->disorderRange + 1)) * (-1); - debugPrint("rand data generated, back %"PRId64"\n", randTail); - } + } + if (retLen > remainderBufLen) { + break; } - int64_t d = startTime - + randTail; - retLen = generateRowData( - data, - d, - superTblInfo); - } - - if (retLen > remainderBufLen) { - break; - } - - pstr += snprintf(pstr , retLen + 1, "%s", data); - k++; - len += retLen; - remainderBufLen -= retLen; + pstr += snprintf(pstr , retLen + 1, "%s", data); + k++; + len += retLen; + remainderBufLen -= retLen; } else { - char **data_type = g_args.datatype; - int lenOfBinary = g_args.len_of_binary; + char **data_type = g_args.datatype; + int lenOfBinary = g_args.len_of_binary; + retLen = generateData(data, data_type, + ncols_per_record, + startTime + getTSRandTail( + DEFAULT_TIMESTAMP_STEP, k, + g_args.disorderRatio, + g_args.disorderRange), + lenOfBinary); + if (len > remainderBufLen) + break; - int64_t randTail = DEFAULT_TIMESTAMP_STEP * k; - - if (g_args.disorderRatio != 0) { - int rand_num = taosRandom() % 100; - if (rand_num < g_args.disorderRatio) { - randTail = (randTail + (taosRandom() % g_args.disorderRange + 1)) * (-1); - - debugPrint("rand data generated, back %"PRId64"\n", randTail); - } - } else { - randTail = DEFAULT_TIMESTAMP_STEP * k; - } - - retLen = generateData(data, data_type, - ncols_per_record, - startTime + randTail, - lenOfBinary); - - if (len > remainderBufLen) - break; - - pstr += sprintf(pstr, "%s", data); - k++; - len += retLen; - remainderBufLen -= retLen; + pstr += sprintf(pstr, "%s", data); + k++; + len += retLen; + remainderBufLen -= retLen; } verbosePrint("%s() LN%d len=%"PRIu64" k=%"PRIu64" \nbuffer=%s\n", @@ -5092,6 +5070,22 @@ static int64_t generateInterlaceDataBuffer( return k; } +static int64_t getTSRandTail(int64_t timeStampStep, int32_t seq, + int disorderRatio, int disorderRange) +{ + int64_t randTail = timeStampStep * seq; + if (disorderRatio > 0) { + int rand_num = taosRandom() % 100; + if(rand_num < disorderRatio) { + randTail = (randTail + + (taosRandom() % disorderRange + 1)) * (-1); + debugPrint("rand data generated, back %"PRId64"\n", randTail); + } + } + + return randTail; +} + static int64_t generateProgressiveDataBuffer( char *tableName, int64_t tableSeq, @@ -6445,7 +6439,6 @@ static int queryTestProcess() { char sqlStr[MAX_TB_NAME_SIZE*2]; sprintf(sqlStr, "use %s", g_queryInfo.dbName); - verbosePrint("%s() %d sqlStr: %s\n", __func__, __LINE__, sqlStr); if (0 != queryDbExec(taos, sqlStr, NO_INSERT_TYPE, false)) { taos_close(taos); free(infos); @@ -6769,7 +6762,6 @@ static void *specifiedSubscribe(void *sarg) { char sqlStr[MAX_TB_NAME_SIZE*2]; sprintf(sqlStr, "use %s", g_queryInfo.dbName); - debugPrint("%s() %d sqlStr: %s\n", __func__, __LINE__, sqlStr); if (0 != queryDbExec(pThreadInfo->taos, sqlStr, NO_INSERT_TYPE, false)) { taos_close(pThreadInfo->taos); return NULL; @@ -7197,7 +7189,6 @@ static void querySqlFile(TAOS* taos, char* sqlFile) } memcpy(cmd + cmd_len, line, read_len); - verbosePrint("%s() LN%d cmd: %s\n", __func__, __LINE__, cmd); if (0 != queryDbExec(taos, cmd, NO_INSERT_TYPE, false)) { errorPrint("%s() LN%d, queryDbExec %s failed!\n", __func__, __LINE__, cmd); From a566615eb7d3d38cb08130ba671f900ba0b77db9 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Wed, 26 May 2021 10:28:58 +0800 Subject: [PATCH 18/19] TD-3279 TD-4328 --- src/inc/taoserror.h | 1 + src/mnode/src/mnodeSdb.c | 10 +++++----- src/vnode/src/vnodeSync.c | 7 ++++++- 3 files changed, 12 insertions(+), 6 deletions(-) diff --git a/src/inc/taoserror.h b/src/inc/taoserror.h index ce6f7c4f22..936e9860dd 100644 --- a/src/inc/taoserror.h +++ b/src/inc/taoserror.h @@ -219,6 +219,7 @@ int32_t* taosGetErrno(); #define TSDB_CODE_VND_NO_WRITE_AUTH TAOS_DEF_ERROR_CODE(0, 0x0512) //"Database write operation denied") #define TSDB_CODE_VND_IS_SYNCING TAOS_DEF_ERROR_CODE(0, 0x0513) //"Database is syncing") #define TSDB_CODE_VND_INVALID_TSDB_STATE TAOS_DEF_ERROR_CODE(0, 0x0514) //"Invalid tsdb state") +#define TSDB_CODE_VND_IS_CLOSING TAOS_DEF_ERROR_CODE(0, 0x0515) //"Database is closing") // tsdb #define TSDB_CODE_TDB_INVALID_TABLE_ID TAOS_DEF_ERROR_CODE(0, 0x0600) //"Invalid table ID") diff --git a/src/mnode/src/mnodeSdb.c b/src/mnode/src/mnodeSdb.c index 505d3c519c..cc088e3409 100644 --- a/src/mnode/src/mnodeSdb.c +++ b/src/mnode/src/mnodeSdb.c @@ -712,13 +712,13 @@ static int32_t sdbProcessWrite(void *wparam, void *hparam, int32_t qtype, void * if (action == SDB_ACTION_INSERT) { return sdbPerformInsertAction(pHead, pTable); } else if (action == SDB_ACTION_DELETE) { - if (qtype == TAOS_QTYPE_FWD) { + //if (qtype == TAOS_QTYPE_FWD) { // Drop database/stable may take a long time and cause a timeout, so we confirm first then reput it into queue - sdbWriteFwdToQueue(1, hparam, TAOS_QTYPE_QUERY, unused); - return TSDB_CODE_SUCCESS; - } else { + // sdbWriteFwdToQueue(1, hparam, TAOS_QTYPE_QUERY, unused); + // return TSDB_CODE_SUCCESS; + //} else { return sdbPerformDeleteAction(pHead, pTable); - } + //} } else if (action == SDB_ACTION_UPDATE) { return sdbPerformUpdateAction(pHead, pTable); } else { diff --git a/src/vnode/src/vnodeSync.c b/src/vnode/src/vnodeSync.c index e5a1964915..22c2dfffe6 100644 --- a/src/vnode/src/vnodeSync.c +++ b/src/vnode/src/vnodeSync.c @@ -126,11 +126,16 @@ void vnodeStopSyncFile(int32_t vgId, uint64_t fversion) { } void vnodeConfirmForard(int32_t vgId, void *wparam, int32_t code) { - void *pVnode = vnodeAcquire(vgId); + SVnodeObj *pVnode = vnodeAcquire(vgId); if (pVnode == NULL) { vError("vgId:%d, vnode not found while confirm forward", vgId); } + if (code == TSDB_CODE_SYN_CONFIRM_EXPIRED && pVnode->status == TAOS_VN_STATUS_CLOSING) { + vDebug("vgId:%d, db:%s, vnode is closing while confirm forward", vgId, pVnode->db); + code = TSDB_CODE_VND_IS_DROPPING; + } + dnodeSendRpcVWriteRsp(pVnode, wparam, code); vnodeRelease(pVnode); } From c96fb6b0f88f2a595e67bbc313be84e43ab6e709 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Wed, 26 May 2021 11:18:29 +0800 Subject: [PATCH 19/19] TD-4328 --- src/inc/taoserror.h | 2 +- src/vnode/src/vnodeSync.c | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/inc/taoserror.h b/src/inc/taoserror.h index 936e9860dd..b3e5b59627 100644 --- a/src/inc/taoserror.h +++ b/src/inc/taoserror.h @@ -219,7 +219,7 @@ int32_t* taosGetErrno(); #define TSDB_CODE_VND_NO_WRITE_AUTH TAOS_DEF_ERROR_CODE(0, 0x0512) //"Database write operation denied") #define TSDB_CODE_VND_IS_SYNCING TAOS_DEF_ERROR_CODE(0, 0x0513) //"Database is syncing") #define TSDB_CODE_VND_INVALID_TSDB_STATE TAOS_DEF_ERROR_CODE(0, 0x0514) //"Invalid tsdb state") -#define TSDB_CODE_VND_IS_CLOSING TAOS_DEF_ERROR_CODE(0, 0x0515) //"Database is closing") +#define TSDB_CODE_VND_IS_CLOSING TAOS_DEF_ERROR_CODE(0, 0x0515) //"Database is closing") // tsdb #define TSDB_CODE_TDB_INVALID_TABLE_ID TAOS_DEF_ERROR_CODE(0, 0x0600) //"Invalid table ID") diff --git a/src/vnode/src/vnodeSync.c b/src/vnode/src/vnodeSync.c index 22c2dfffe6..4197428fec 100644 --- a/src/vnode/src/vnodeSync.c +++ b/src/vnode/src/vnodeSync.c @@ -133,7 +133,7 @@ void vnodeConfirmForard(int32_t vgId, void *wparam, int32_t code) { if (code == TSDB_CODE_SYN_CONFIRM_EXPIRED && pVnode->status == TAOS_VN_STATUS_CLOSING) { vDebug("vgId:%d, db:%s, vnode is closing while confirm forward", vgId, pVnode->db); - code = TSDB_CODE_VND_IS_DROPPING; + code = TSDB_CODE_VND_IS_CLOSING; } dnodeSendRpcVWriteRsp(pVnode, wparam, code);