Skip to content

T
TDengine

慢慢CG / TDengine
与 Fork 源项目一致

Fork自 taosdata / TDengine

通知 1
Star 0
Fork 0
T
TDengine
提交 9598664f 编写于 作者: S Shengliang Guan
浏览文件
操作

Merge from develop

上级 26d168de 916a9f68
隐藏空白更改
内联 并排
Showing with 1417 addition and 566 deletion
cmake/install.inc
浏览文件 @ 9598664f
......@@ -32,7 +32,7 @@ ELSEIF (TD_WINDOWS)
#INSTALL(TARGETS taos RUNTIME DESTINATION driver)
#INSTALL(TARGETS shell RUNTIME DESTINATION .)
IF (TD_MVN_INSTALLED)
INSTALL(FILES ${LIBRARY_OUTPUT_PATH}/taos-jdbcdriver-2.0.14-dist.jar DESTINATION connector/jdbc)
INSTALL(FILES ${LIBRARY_OUTPUT_PATH}/taos-jdbcdriver-2.0.15-dist.jar DESTINATION connector/jdbc)
ENDIF ()
ELSEIF (TD_DARWIN)
SET(TD_MAKE_INSTALL_SH "${TD_COMMUNITY_DIR}/packaging/tools/make_install.sh")
......
src/balance/src/bnMain.c
浏览文件 @ 9598664f
......@@ -15,6 +15,7 @@
#define _DEFAULT_SOURCE
#include "os.h"
#include "tref.h"
#include "tsync.h"
#include "tglobal.h"
#include "dnode.h"
......@@ -28,7 +29,9 @@
#include "mnodeUser.h"
#include "mnodeVgroup.h"
static SBnMgmt tsBnMgmt;;
extern int64_t tsDnodeRid;
extern int64_t tsSdbRid;
static SBnMgmt tsBnMgmt;
static void bnMonitorDnodeModule();
static void bnLock() {
......@@ -529,6 +532,9 @@ void bnCheckStatus() {
void * pIter = NULL;
SDnodeObj *pDnode = NULL;
void *dnodeSdb = taosAcquireRef(tsSdbRid, tsDnodeRid);
if (dnodeSdb == NULL) return;
while (1) {
pIter = mnodeGetNextDnode(pIter, &pDnode);
if (pDnode == NULL) break;
......@@ -543,6 +549,8 @@ void bnCheckStatus() {
}
mnodeDecDnodeRef(pDnode);
}
taosReleaseRef(tsSdbRid, tsDnodeRid);
}
void bnCheckModules() {
......
src/client/src/tscAsync.c
浏览文件 @ 9598664f
......@@ -96,7 +96,7 @@ void taos_query_a(TAOS *taos, const char *sqlstr, __async_cb_func_t fp, void *pa
return;
}
taosNotePrintTsc(sqlstr);
nPrintTsc(sqlstr);
SSqlObj *pSql = (SSqlObj *)calloc(1, sizeof(SSqlObj));
if (pSql == NULL) {
......
src/client/src/tscSQLParser.c
浏览文件 @ 9598664f
......@@ -3282,7 +3282,12 @@ static int32_t extractColumnFilterInfo(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SC
((pSchema->type == TSDB_DATA_TYPE_BINARY || pSchema->type == TSDB_DATA_TYPE_NCHAR) ? 1 : 0);
if (pColFilter->filterstr) {
if (pExpr->nSQLOptr != TK_EQ && pExpr->nSQLOptr != TK_NE && pExpr->nSQLOptr != TK_LIKE) {
if (pExpr->nSQLOptr != TK_EQ
&& pExpr->nSQLOptr != TK_NE
&& pExpr->nSQLOptr != TK_ISNULL
&& pExpr->nSQLOptr != TK_NOTNULL
&& pExpr->nSQLOptr != TK_LIKE
) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg2);
}
} else {
......
src/client/src/tscSql.c
浏览文件 @ 9598664f
......@@ -344,7 +344,7 @@ TAOS_RES* taos_query_c(TAOS *taos, const char *sqlstr, uint32_t sqlLen, TAOS_RES
return NULL;
}
taosNotePrintTsc(sqlstr);
nPrintTsc(sqlstr);
SSqlObj* pSql = calloc(1, sizeof(SSqlObj));
if (pSql == NULL) {
......
src/client/src/tscSystem.c
浏览文件 @ 9598664f
......@@ -17,6 +17,7 @@
#include "taosmsg.h"
#include "tref.h"
#include "trpc.h"
#include "tnote.h"
#include "ttimer.h"
#include "tutil.h"
#include "tsched.h"
......@@ -40,7 +41,6 @@ int tscRefId = -1;
int tscNumOfThreads;
static pthread_once_t tscinit = PTHREAD_ONCE_INIT;
void taosInitNote(int numOfNoteLines, int maxNotes, char* lable);
//void tscUpdateEpSet(void *ahandle, SRpcEpSet *pEpSet);
void tscCheckDiskUsage(void *UNUSED_PARAM(para), void *UNUSED_PARAM(param)) {
......@@ -77,7 +77,6 @@ int32_t tscInitRpc(const char *user, const char *secretEncrypt, void **pDnodeCon
return 0;
}
void taos_init_imp(void) {
char temp[128] = {0};
......@@ -103,6 +102,7 @@ void taos_init_imp(void) {
taosReadGlobalCfg();
taosCheckGlobalCfg();
taosInitNotes();
rpcInit();
tscDebug("starting to initialize TAOS client ...");
......@@ -110,11 +110,6 @@ void taos_init_imp(void) {
}
taosSetCoreDump();
if (tsTscEnableRecordSql != 0) {
taosInitNote(tsNumOfLogLines / 10, 1, (char*)"tsc_note");
}
tscInitMsgsFp();
int queueSize = tsMaxConnections*2;
......
src/connector/jdbc/CMakeLists.txt
浏览文件 @ 9598664f
......@@ -8,7 +8,7 @@ IF (TD_MVN_INSTALLED)
ADD_CUSTOM_COMMAND(OUTPUT ${JDBC_CMD_NAME}
POST_BUILD
COMMAND mvn -Dmaven.test.skip=true install -f ${CMAKE_CURRENT_SOURCE_DIR}/pom.xml
COMMAND ${CMAKE_COMMAND} -E copy ${CMAKE_CURRENT_SOURCE_DIR}/target/taos-jdbcdriver-2.0.14-dist.jar ${LIBRARY_OUTPUT_PATH}
COMMAND ${CMAKE_COMMAND} -E copy ${CMAKE_CURRENT_SOURCE_DIR}/target/taos-jdbcdriver-2.0.15-dist.jar ${LIBRARY_OUTPUT_PATH}
COMMAND mvn -Dmaven.test.skip=true clean -f ${CMAKE_CURRENT_SOURCE_DIR}/pom.xml
COMMENT "build jdbc driver")
ADD_CUSTOM_TARGET(${JDBC_TARGET_NAME} ALL WORKING_DIRECTORY ${EXECUTABLE_OUTPUT_PATH} DEPENDS ${JDBC_CMD_NAME})
......
src/connector/jdbc/deploy-pom.xml
浏览文件 @ 9598664f
......@@ -5,7 +5,7 @@
<groupId>com.taosdata.jdbc</groupId>
<artifactId>taos-jdbcdriver</artifactId>
<version>2.0.14</version>
<version>2.0.15</version>
<packaging>jar</packaging>
<name>JDBCDriver</name>
......@@ -36,7 +36,6 @@
</developer>
</developers>
<dependencies>
<dependency>
<groupId>commons-logging</groupId>
......
src/connector/jdbc/pom.xml
浏览文件 @ 9598664f
......@@ -3,7 +3,7 @@
<modelVersion>4.0.0</modelVersion>
<groupId>com.taosdata.jdbc</groupId>
<artifactId>taos-jdbcdriver</artifactId>
<version>2.0.14</version>
<version>2.0.15</version>
<packaging>jar</packaging>
<name>JDBCDriver</name>
<url>https://github.com/taosdata/TDengine/tree/master/src/connector/jdbc</url>
......
src/connector/jdbc/src/main/java/com/taosdata/jdbc/TSDBStatement.java
浏览文件 @ 9598664f
......@@ -81,7 +81,7 @@ public class TSDBStatement implements Statement {
}
if (!this.connector.isUpdateQuery(pSql)) {
TSDBResultSet res = new TSDBResultSet(this.connector, resultSetPointer);
TSDBResultSet res = new TSDBResultSet(this.connector, resultSetPointer);
res.setBatchFetch(this.connection.getBatchFetch());
return res;
} else {
......@@ -125,7 +125,8 @@ public class TSDBStatement implements Statement {
}
public int getMaxFieldSize() throws SQLException {
throw new SQLException(TSDBConstants.UNSUPPORT_METHOD_EXCEPTIONZ_MSG);
return 0;
// throw new SQLException(TSDBConstants.UNSUPPORT_METHOD_EXCEPTIONZ_MSG);
}
public void setMaxFieldSize(int max) throws SQLException {
......@@ -218,7 +219,8 @@ public class TSDBStatement implements Statement {
}
public int getFetchDirection() throws SQLException {
throw new SQLException(TSDBConstants.UNSUPPORT_METHOD_EXCEPTIONZ_MSG);
return ResultSet.FETCH_FORWARD;
// throw new SQLException(TSDBConstants.UNSUPPORT_METHOD_EXCEPTIONZ_MSG);
}
/*
......
src/dnode/src/dnodeEps.c
浏览文件 @ 9598664f
......@@ -237,7 +237,7 @@ PRASE_EPS_OVER:
dnodeUpdateEp(dnodeGetDnodeId(), tsLocalEp, tsLocalFqdn, &tsServerPort);
#else
if (dnodeCheckEpChanged(dnodeGetDnodeId(), tsLocalEp)) {
dError("dnode:%d, localEp is changed to %s in dnodeEps.json and need reconfigured", dnodeGetDnodeId(), tsLocalEp);
dError("dnode:%d, localEp is different from %s in dnodeEps.json and need reconfigured", dnodeGetDnodeId(), tsLocalEp);
return -1;
}
#endif
......
src/dnode/src/dnodeMain.c
浏览文件 @ 9598664f
......@@ -16,6 +16,7 @@
#define _DEFAULT_SOURCE
#include "os.h"
#include "taos.h"
#include "tnote.h"
#include "tconfig.h"
#include "tfile.h"
#include "twal.h"
......@@ -98,6 +99,7 @@ int32_t dnodeInitSystem() {
taosInitGlobalCfg();
taosReadGlobalLogCfg();
taosSetCoreDump();
taosInitNotes();
signal(SIGPIPE, SIG_IGN);
if (dnodeCreateDir(tsLogDir) < 0) {
......
src/kit/shell/src/shellLinux.c
浏览文件 @ 9598664f
......@@ -46,7 +46,7 @@ static struct argp_option options[] = {
{"thread", 'T', "THREADNUM", 0, "Number of threads when using multi-thread to import data."},
{"database", 'd', "DATABASE", 0, "Database to use when connecting to the server."},
{"timezone", 't', "TIMEZONE", 0, "Time zone of the shell, default is local."},
{"netrole", 'n', "NETROLE", 0, "Net role when network connectivity test, default is NULL, options: client|server|rpc|startup."},
{"netrole", 'n', "NETROLE", 0, "Net role when network connectivity test, default is startup, options: client|server|rpc|startup."},
{"pktlen", 'l', "PKTLEN", 0, "Packet length used for net test, default is 1000 bytes."},
{0}};
......
src/kit/shell/src/shellWindows.c
浏览文件 @ 9598664f
......@@ -45,6 +45,10 @@ void printHelp() {
printf("%s%s%s\n", indent, indent, "Database to use when connecting to the server.");
printf("%s%s\n", indent, "-t");
printf("%s%s%s\n", indent, indent, "Time zone of the shell, default is local.");
printf("%s%s\n", indent, "-n");
printf("%s%s%s\n", indent, indent, "Net role when network connectivity test, default is startup, options: client|server|rpc|startup.");
printf("%s%s\n", indent, "-l");
printf("%s%s%s\n", indent, indent, "Packet length used for net test, default is 1000 bytes.");
exit(EXIT_SUCCESS);
}
......@@ -137,6 +141,24 @@ void shellParseArgument(int argc, char *argv[], SShellArguments *arguments) {
exit(EXIT_FAILURE);
}
}
// For time zone
else if (strcmp(argv[i], "-n") == 0) {
if (i < argc - 1) {
arguments->netTestRole = argv[++i];
} else {
fprintf(stderr, "option -n requires an argument\n");
exit(EXIT_FAILURE);
}
}
// For time zone
else if (strcmp(argv[i], "-l") == 0) {
if (i < argc - 1) {
arguments->pktLen = atoi(argv[++i]);
} else {
fprintf(stderr, "option -l requires an argument\n");
exit(EXIT_FAILURE);
}
}
// For temperory command TODO
else if (strcmp(argv[i], "--help") == 0) {
printHelp();
......
src/mnode/src/mnodeTable.c
浏览文件 @ 9598664f
......@@ -396,14 +396,15 @@ static void mnodeAddTableIntoStable(SSTableObj *pStable, SCTableObj *pCtable) {
atomic_add_fetch_32(&pStable->numOfTables, 1);
if (pStable->vgHash == NULL) {
pStable->vgHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
pStable->vgHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
mDebug("table:%s, create hash:%p", pStable->info.tableId, pStable->vgHash);
}
if (pStable->vgHash != NULL) {
if (taosHashGet(pStable->vgHash, &pCtable->vgId, sizeof(pCtable->vgId)) == NULL) {
taosHashPut(pStable->vgHash, &pCtable->vgId, sizeof(pCtable->vgId), &pCtable->vgId, sizeof(pCtable->vgId));
mDebug("table:%s, vgId:%d is put into stable vgList, sizeOfVgList:%d", pStable->info.tableId, pCtable->vgId,
(int32_t)taosHashGetSize(pStable->vgHash));
mDebug("table:%s, vgId:%d is put into stable hash:%p, sizeOfVgList:%d", pStable->info.tableId, pCtable->vgId,
pStable->vgHash, taosHashGetSize(pStable->vgHash));
}
}
}
......@@ -416,13 +417,14 @@ static void mnodeRemoveTableFromStable(SSTableObj *pStable, SCTableObj *pCtable)
SVgObj *pVgroup = mnodeGetVgroup(pCtable->vgId);
if (pVgroup == NULL) {
taosHashRemove(pStable->vgHash, &pCtable->vgId, sizeof(pCtable->vgId));
mDebug("table:%s, vgId:%d is remove from stable vgList, sizeOfVgList:%d", pStable->info.tableId, pCtable->vgId,
(int32_t)taosHashGetSize(pStable->vgHash));
mDebug("table:%s, vgId:%d is remove from stable hash:%p sizeOfVgList:%d", pStable->info.tableId, pCtable->vgId,
pStable->vgHash, taosHashGetSize(pStable->vgHash));
}
mnodeDecVgroupRef(pVgroup);
}
static void mnodeDestroySuperTable(SSTableObj *pStable) {
mDebug("table:%s, is destroyed, stable hash:%p", pStable->info.tableId, pStable->vgHash);
if (pStable->vgHash != NULL) {
taosHashCleanup(pStable->vgHash);
pStable->vgHash = NULL;
......@@ -464,6 +466,9 @@ static int32_t mnodeSuperTableActionUpdate(SSdbRow *pRow) {
SSTableObj *pNew = pRow->pObj;
SSTableObj *pTable = mnodeGetSuperTable(pNew->info.tableId);
if (pTable != NULL && pTable != pNew) {
mDebug("table:%s, will be updated, hash:%p sizeOfVgList:%d, new hash:%p sizeOfVgList:%d", pTable->info.tableId,
pTable->vgHash, taosHashGetSize(pTable->vgHash), pNew->vgHash, taosHashGetSize(pNew->vgHash));
void *oldTableId = pTable->info.tableId;
void *oldSchema = pTable->schema;
void *oldVgHash = pTable->vgHash;
......@@ -479,6 +484,9 @@ static int32_t mnodeSuperTableActionUpdate(SSdbRow *pRow) {
free(pNew);
free(oldTableId);
free(oldSchema);
mDebug("table:%s, update finished, hash:%p sizeOfVgList:%d", pTable->info.tableId, pTable->vgHash,
taosHashGetSize(pTable->vgHash));
}
mnodeDecTableRef(pTable);
......@@ -783,8 +791,8 @@ static int32_t mnodeProcessDropTableMsg(SMnodeMsg *pMsg) {
if (pMsg->pTable->type == TSDB_SUPER_TABLE) {
SSTableObj *pSTable = (SSTableObj *)pMsg->pTable;
mInfo("msg:%p, app:%p table:%s, start to drop stable, uid:%" PRIu64 ", numOfChildTables:%d, sizeOfVgList:%d",
pMsg, pMsg->rpcMsg.ahandle, pDrop->tableId, pSTable->uid, pSTable->numOfTables, (int32_t)taosHashGetSize(pSTable->vgHash));
mInfo("msg:%p, app:%p table:%s, start to drop stable, uid:%" PRIu64 ", numOfChildTables:%d, sizeOfVgList:%d", pMsg,
pMsg->rpcMsg.ahandle, pDrop->tableId, pSTable->uid, pSTable->numOfTables, taosHashGetSize(pSTable->vgHash));
return mnodeProcessDropSuperTableMsg(pMsg);
} else {
SCTableObj *pCTable = (SCTableObj *)pMsg->pTable;
......@@ -925,7 +933,10 @@ static int32_t mnodeProcessDropSuperTableMsg(SMnodeMsg *pMsg) {
if (pMsg == NULL) return TSDB_CODE_MND_APP_ERROR;
SSTableObj *pStable = (SSTableObj *)pMsg->pTable;
if (pStable->vgHash != NULL /*pStable->numOfTables != 0*/) {
mInfo("msg:%p, app:%p stable:%s will be dropped, hash:%p sizeOfVgList:%d", pMsg, pMsg->rpcMsg.ahandle,
pStable->info.tableId, pStable->vgHash, taosHashGetSize(pStable->vgHash));
if (pStable->vgHash != NULL /*pStable->numOfTables != 0*/) {
int32_t *pVgId = taosHashIterate(pStable->vgHash, NULL);
while (pVgId) {
SVgObj *pVgroup = mnodeGetVgroup(*pVgId);
......@@ -938,8 +949,9 @@ static int32_t mnodeProcessDropSuperTableMsg(SMnodeMsg *pMsg) {
pDrop->uid = htobe64(pStable->uid);
mnodeExtractTableName(pStable->info.tableId, pDrop->tableId);
mInfo("msg:%p, app:%p stable:%s, send drop stable msg to vgId:%d", pMsg, pMsg->rpcMsg.ahandle,
pStable->info.tableId, pVgroup->vgId);
mInfo("msg:%p, app:%p stable:%s, send drop stable msg to vgId:%d, hash:%p sizeOfVgList:%d", pMsg,
pMsg->rpcMsg.ahandle, pStable->info.tableId, pVgroup->vgId, pStable->vgHash,
taosHashGetSize(pStable->vgHash));
SRpcEpSet epSet = mnodeGetEpSetFromVgroup(pVgroup);
SRpcMsg rpcMsg = {.pCont = pDrop, .contLen = sizeof(SDropSTableMsg), .msgType = TSDB_MSG_TYPE_MD_DROP_STABLE};
dnodeSendMsgToDnode(&epSet, &rpcMsg);
......@@ -1482,8 +1494,8 @@ static int32_t mnodeGetSuperTableMeta(SMnodeMsg *pMsg) {
pMsg->rpcRsp.rsp = pMeta;
mDebug("msg:%p, app:%p stable:%s, uid:%" PRIu64 " table meta is retrieved", pMsg, pMsg->rpcMsg.ahandle,
pTable->info.tableId, pTable->uid);
mDebug("msg:%p, app:%p stable:%s, uid:%" PRIu64 " table meta is retrieved, sizeOfVgList:%d numOfTables:%d", pMsg,
pMsg->rpcMsg.ahandle, pTable->info.tableId, pTable->uid, taosHashGetSize(pTable->vgHash), pTable->numOfTables);
return TSDB_CODE_SUCCESS;
}
......@@ -1512,7 +1524,7 @@ static int32_t mnodeProcessSuperTableVgroupMsg(SMnodeMsg *pMsg) {
char *msg = (char *)pRsp + sizeof(SSTableVgroupRspMsg);
for (int32_t i = 0; i < numOfTable; ++i) {
char * stableName = (char *)pInfo + sizeof(SSTableVgroupMsg) + (TSDB_TABLE_FNAME_LEN)*i;
char *stableName = (char *)pInfo + sizeof(SSTableVgroupMsg) + (TSDB_TABLE_FNAME_LEN)*i;
SSTableObj *pTable = mnodeGetSuperTable(stableName);
if (pTable == NULL) {
mError("msg:%p, app:%p stable:%s, not exist while get stable vgroup info", pMsg, pMsg->rpcMsg.ahandle, stableName);
......@@ -1533,6 +1545,8 @@ static int32_t mnodeProcessSuperTableVgroupMsg(SMnodeMsg *pMsg) {
msg += sizeof(SVgroupsMsg);
} else {
SVgroupsMsg *pVgroupMsg = (SVgroupsMsg *)msg;
mDebug("msg:%p, app:%p stable:%s, hash:%p sizeOfVgList:%d will be returned", pMsg, pMsg->rpcMsg.ahandle,
pTable->info.tableId, pTable->vgHash, taosHashGetSize(pTable->vgHash));
int32_t *pVgId = taosHashIterate(pTable->vgHash, NULL);
int32_t vgSize = 0;
......
src/plugins/http/src/httpContext.c
浏览文件 @ 9598664f
......@@ -79,7 +79,7 @@ bool httpInitContexts() {
void httpCleanupContexts() {
if (tsHttpServer.contextCache != NULL) {
SCacheObj *cache = tsHttpServer.contextCache;
httpInfo("context cache is cleanuping, size:%" PRIzu "", taosHashGetSize(cache->pHashTable));
httpInfo("context cache is cleanuping, size:%d", taosHashGetSize(cache->pHashTable));
taosCacheCleanup(tsHttpServer.contextCache);
tsHttpServer.contextCache = NULL;
}
......
src/plugins/http/src/httpSession.c
浏览文件 @ 9598664f
......@@ -107,7 +107,7 @@ static void httpDestroySession(void *data) {
void httpCleanUpSessions() {
if (tsHttpServer.sessionCache != NULL) {
SCacheObj *cache = tsHttpServer.sessionCache;
httpInfo("session cache is cleanuping, size:%" PRIzu "", taosHashGetSize(cache->pHashTable));
httpInfo("session cache is cleanuping, size:%d", taosHashGetSize(cache->pHashTable));
taosCacheCleanup(tsHttpServer.sessionCache);
tsHttpServer.sessionCache = NULL;
}
......
src/plugins/http/src/httpSql.c
浏览文件 @ 9598664f
......@@ -181,7 +181,7 @@ void httpProcessMultiSql(HttpContext *pContext) {
char *sql = httpGetCmdsString(pContext, cmd->sql);
httpTraceL("context:%p, fd:%d, user:%s, process pos:%d, start query, sql:%s", pContext, pContext->fd, pContext->user,
multiCmds->pos, sql);
taosNotePrintHttp(sql);
nPrintHttp(sql);
taos_query_a(pContext->session->taos, sql, httpProcessMultiSqlCallBack, (void *)pContext);
}
......@@ -329,7 +329,7 @@ void httpProcessSingleSqlCmd(HttpContext *pContext) {
}
httpTraceL("context:%p, fd:%d, user:%s, start query, sql:%s", pContext, pContext->fd, pContext->user, sql);
taosNotePrintHttp(sql);
nPrintHttp(sql);
taos_query_a(pSession->taos, sql, httpProcessSingleSqlCallBack, (void *)pContext);
}
......
src/plugins/http/src/httpSystem.c
浏览文件 @ 9598664f
......@@ -37,7 +37,6 @@ void opInitHandle(HttpServer* pServer) {}
#endif
HttpServer tsHttpServer;
void taosInitNote(int32_t numOfNoteLines, int32_t maxNotes, char* lable);
int32_t httpInitSystem() {
strcpy(tsHttpServer.label, "rest");
......@@ -48,9 +47,6 @@ int32_t httpInitSystem() {
pthread_mutex_init(&tsHttpServer.serverMutex, NULL);
if (tsHttpEnableRecordSql != 0) {
taosInitNote(tsNumOfLogLines / 10, 1, (char*)"http_note");
}
restInitHandle(&tsHttpServer);
adminInitHandle(&tsHttpServer);
gcInitHandle(&tsHttpServer);
......
src/util/inc/hash.h
浏览文件 @ 9598664f
......@@ -82,7 +82,7 @@ SHashObj *taosHashInit(size_t capacity, _hash_fn_t fn, bool update, SHashLockTyp
* @param pHashObj
* @return
*/
size_t taosHashGetSize(const SHashObj *pHashObj);
int32_t taosHashGetSize(const SHashObj *pHashObj);
/**
* put element into hash table, if the element with the same key exists, update it
......
src/util/inc/tnote.h
浏览文件 @ 9598664f
......@@ -20,41 +20,42 @@
extern "C" {
#endif
#include "os.h"
#include "tutil.h"
#include "tglobal.h"
#define MAX_NOTE_LINE_SIZE 66000
#define NOTE_FILE_NAME_LEN 300
typedef struct _taosNoteInfo {
int taosNoteFileNum ;
int taosNoteMaxLines;
int taosNoteLines;
char taosNoteName[NOTE_FILE_NAME_LEN];
int taosNoteFlag;
int taosNoteFd;
int taosNoteOpenInProgress;
pthread_mutex_t taosNoteMutex;
}taosNoteInfo;
void taosNotePrint(taosNoteInfo * pNote, const char * const format, ...);
extern taosNoteInfo m_HttpNote;
extern taosNoteInfo m_TscNote;
extern int tsHttpEnableRecordSql;
extern int tsTscEnableRecordSql;
#define taosNotePrintHttp(...) \
typedef struct {
int32_t fileNum;
int32_t maxLines;
int32_t lines;
int32_t flag;
int32_t fd;
int32_t openInProgress;
char name[NOTE_FILE_NAME_LEN];
pthread_mutex_t mutex;
} SNoteObj;
extern SNoteObj tsHttpNote;
extern SNoteObj tsTscNote;
extern SNoteObj tsInfoNote;
void taosInitNotes();
void taosNotePrint(SNoteObj* pNote, const char* const format, ...);
void taosNotePrintBuffer(SNoteObj *pNote, char *buffer, int32_t len);
#define nPrintHttp(...) \
if (tsHttpEnableRecordSql) { \
taosNotePrint(&m_HttpNote, __VA_ARGS__); \
taosNotePrint(&tsHttpNote, __VA_ARGS__); \
}
#define nPrintTsc(...) \
if (tsTscEnableRecordSql) { \
taosNotePrint(&tsTscNote, __VA_ARGS__); \
}
#define nInfo(buffer, len) \
if (tscEmbedded == 1) { \
taosNotePrintBuffer(&tsInfoNote, buffer, len); \
}
#define taosNotePrintTsc(...) \
if (tsTscEnableRecordSql) { \
taosNotePrint(&m_TscNote, __VA_ARGS__); \
}
#ifdef __cplusplus
}
......
src/util/src/hash.c
浏览文件 @ 9598664f
......@@ -189,7 +189,7 @@ SHashObj *taosHashInit(size_t capacity, _hash_fn_t fn, bool update, SHashLockTyp
return pHashObj;
}
size_t taosHashGetSize(const SHashObj *pHashObj) { return (pHashObj == NULL) ? 0 : pHashObj->size; }
int32_t taosHashGetSize(const SHashObj *pHashObj) { return (int32_t)((pHashObj == NULL) ? 0 : pHashObj->size); }
int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, void *data, size_t size) {
uint32_t hashVal = (*pHashObj->hashFp)(key, (uint32_t)keyLen);
......
src/util/src/tlog.c
浏览文件 @ 9598664f
......@@ -17,6 +17,7 @@
#include "os.h"
#include "tulog.h"
#include "tlog.h"
#include "tnote.h"
#include "tutil.h"
#define MAX_LOGLINE_SIZE (1000)
......@@ -287,7 +288,6 @@ static int32_t taosOpenLogFile(char *fn, int32_t maxLines, int32_t maxFileNum) {
tsLogObj.fileNum = maxFileNum;
taosGetLogFileName(fn);
if (strlen(fn) < LOG_FILE_NAME_LEN + 50 - 2) {
strcpy(name, fn);
strcat(name, ".0");
......@@ -401,6 +401,7 @@ void taosPrintLog(const char *flags, int32_t dflag, const char *format, ...) {
}
if (dflag & DEBUG_SCREEN) taosWrite(1, buffer, (uint32_t)len);
if (dflag == 255) nInfo(buffer, len);
}
void taosDumpData(unsigned char *msg, int32_t len) {
......
src/util/src/tnettest.c
浏览文件 @ 9598664f
......@@ -43,12 +43,13 @@ static void *taosNetBindUdpPort(void *sarg) {
char buffer[BUFFER_SIZE];
int32_t iDataNum;
socklen_t sin_size;
int32_t bufSize = 1024000;
struct sockaddr_in server_addr;
struct sockaddr_in clientAddr;
if ((serverSocket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) < 0) {
uError("failed to create udp socket since %s", strerror(errno));
uError("failed to create UDP socket since %s", strerror(errno));
return NULL;
}
......@@ -58,11 +59,23 @@ static void *taosNetBindUdpPort(void *sarg) {
server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
if (bind(serverSocket, (struct sockaddr *)&server_addr, sizeof(server_addr)) < 0) {
uError("failed to bind udp port:%d since %s", port, strerror(errno));
uError("failed to bind UDP port:%d since %s", port, strerror(errno));
return NULL;
}
uInfo("udp server at port:%d is listening", port);
if (taosSetSockOpt(serverSocket, SOL_SOCKET, SO_SNDBUF, (void *)&bufSize, sizeof(bufSize)) != 0) {
uError("failed to set the send buffer size for UDP socket\n");
taosCloseSocket(serverSocket);
return NULL;
}
if (taosSetSockOpt(serverSocket, SOL_SOCKET, SO_RCVBUF, (void *)&bufSize, sizeof(bufSize)) != 0) {
uError("failed to set the receive buffer size for UDP socket\n");
taosCloseSocket(serverSocket);
return NULL;
}
uInfo("UDP server at port:%d is listening", port);
while (1) {
memset(buffer, 0, BUFFER_SIZE);
......@@ -74,10 +87,13 @@ static void *taosNetBindUdpPort(void *sarg) {
continue;
}
uInfo("UDP: recv:%d bytes from %s at %d", iDataNum, taosInetNtoa(clientAddr.sin_addr), port);
if (iDataNum > 0) {
uInfo("UDP: recv:%d bytes from %s:%d", iDataNum, taosInetNtoa(clientAddr.sin_addr), port);
sendto(serverSocket, buffer, iDataNum, 0, (struct sockaddr *)&clientAddr, (int32_t)sin_size);
iDataNum = taosSendto(serverSocket, buffer, iDataNum, 0, (struct sockaddr *)&clientAddr, (int32_t)sin_size);
}
uInfo("UDP: send:%d bytes to %s at %d", iDataNum, taosInetNtoa(clientAddr.sin_addr), port);
}
taosCloseSocket(serverSocket);
......@@ -94,10 +110,9 @@ static void *taosNetBindTcpPort(void *sarg) {
int32_t addr_len = sizeof(clientAddr);
SOCKET client;
char buffer[BUFFER_SIZE];
int32_t iDataNum = 0;
if ((serverSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0) {
uError("failed to create tcp socket since %s", strerror(errno));
uError("failed to create TCP socket since %s", strerror(errno));
return NULL;
}
......@@ -106,130 +121,103 @@ static void *taosNetBindTcpPort(void *sarg) {
server_addr.sin_port = htons(port);
server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
int32_t reuse = 1;
if (taosSetSockOpt(serverSocket, SOL_SOCKET, SO_REUSEADDR, (void *)&reuse, sizeof(reuse)) < 0) {
uError("setsockopt SO_REUSEADDR failed: %d (%s)", errno, strerror(errno));
taosCloseSocket(serverSocket);
return NULL;
}
if (bind(serverSocket, (struct sockaddr *)&server_addr, sizeof(server_addr)) < 0) {
uError("failed to bind tcp port:%d since %s", port, strerror(errno));
uError("failed to bind TCP port:%d since %s", port, strerror(errno));
return NULL;
}
if (listen(serverSocket, 5) < 0) {
uError("failed to listen tcp port:%d since %s", port, strerror(errno));
if (taosKeepTcpAlive(serverSocket) < 0) {
uError("failed to set tcp server keep-alive option since %s", strerror(errno));
taosCloseSocket(serverSocket);
return NULL;
}
uInfo("tcp server at port:%d is listening", port);
if (listen(serverSocket, 10) < 0) {
uError("failed to listen TCP port:%d since %s", port, strerror(errno));
return NULL;
}
uInfo("TCP server at port:%d is listening", port);
while (1) {
client = accept(serverSocket, (struct sockaddr *)&clientAddr, (socklen_t *)&addr_len);
if (client < 0) {
uDebug("failed to accept from tcp port:%d since %s", port, strerror(errno));
uDebug("TCP: failed to accept at port:%d since %s", port, strerror(errno));
continue;
}
iDataNum = 0;
memset(buffer, 0, BUFFER_SIZE);
int32_t nleft, nread;
char * ptr = buffer;
nleft = pinfo->pktLen;
while (nleft > 0) {
nread = recv(client, ptr, BUFFER_SIZE, 0);
if (nread == 0) {
break;
} else if (nread < 0) {
if (errno == EINTR) {
continue;
} else {
uError("failed to perform recv func at %d since %s", port, strerror(errno));
taosCloseSocket(serverSocket);
return NULL;
}
} else {
nleft -= nread;
ptr += nread;
iDataNum += nread;
}
int32_t ret = taosReadMsg(client, buffer, pinfo->pktLen);
if (ret < 0 || ret != pinfo->pktLen) {
uError("TCP: failed to read %d bytes at port:%d since %s", pinfo->pktLen, port, strerror(errno));
taosCloseSocket(serverSocket);
return NULL;
}
if (iDataNum > 0) {
uInfo("TCP: recv:%d bytes from %s:%d", iDataNum, taosInetNtoa(clientAddr.sin_addr), port);
send(client, buffer, iDataNum, 0);
uInfo("TCP: read:%d bytes from %s at %d", pinfo->pktLen, taosInetNtoa(clientAddr.sin_addr), port);
ret = taosWriteMsg(client, buffer, pinfo->pktLen);
if (ret < 0) {
uError("TCP: failed to write %d bytes at %d since %s", pinfo->pktLen, port, strerror(errno));
taosCloseSocket(serverSocket);
return NULL;
}
uInfo("TCP: write:%d bytes to %s at %d", pinfo->pktLen, taosInetNtoa(clientAddr.sin_addr), port);
}
taosCloseSocket(serverSocket);
return NULL;
}
static int32_t taosNetCheckTcpPort(STestInfo *info) {
SOCKET clientSocket;
char sendbuf[BUFFER_SIZE];
char recvbuf[BUFFER_SIZE];
int32_t iDataNum = 0;
SOCKET clientSocket;
char buffer[BUFFER_SIZE] = {0};
struct sockaddr_in serverAddr;
if ((clientSocket = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
uError("failed to create tcp client socket since %s", strerror(errno));
uError("failed to create TCP client socket since %s", strerror(errno));
return -1;
}
// set send and recv overtime
struct timeval timeout;
timeout.tv_sec = 2; // s
timeout.tv_usec = 0; // us
if (setsockopt(clientSocket, SOL_SOCKET, SO_SNDTIMEO, (char *)&timeout, sizeof(struct timeval)) == -1) {
uError("failed to setsockopt send timer since %s", strerror(errno));
}
if (setsockopt(clientSocket, SOL_SOCKET, SO_RCVTIMEO, (char *)&timeout, sizeof(struct timeval)) == -1) {
uError("failed to setsockopt recv timer since %s", strerror(errno));
int32_t reuse = 1;
if (taosSetSockOpt(clientSocket, SOL_SOCKET, SO_REUSEADDR, (void *)&reuse, sizeof(reuse)) < 0) {
uError("setsockopt SO_REUSEADDR failed: %d (%s)", errno, strerror(errno));
taosCloseSocket(clientSocket);
return -1;
}
struct sockaddr_in serverAddr;
memset((char *)&serverAddr, 0, sizeof(serverAddr));
serverAddr.sin_family = AF_INET;
serverAddr.sin_port = htons(info->port);
serverAddr.sin_port = (uint16_t)htons((uint16_t)info->port);
serverAddr.sin_addr.s_addr = info->hostIp;
if (connect(clientSocket, (struct sockaddr *)&serverAddr, sizeof(serverAddr)) < 0) {
uError("failed to connect port:%d since %s", info->port, strerror(errno));
uError("TCP: failed to connect port %s:%d since %s", taosIpStr(info->hostIp), info->port, strerror(errno));
return -1;
}
memset(sendbuf, 0, BUFFER_SIZE);
memset(recvbuf, 0, BUFFER_SIZE);
struct in_addr ipStr;
memcpy(&ipStr, &info->hostIp, 4);
sprintf(sendbuf, "client send tcp pkg to %s:%d, content: 1122334455", taosInetNtoa(ipStr), info->port);
sprintf(sendbuf + info->pktLen - 16, "1122334455667788");
send(clientSocket, sendbuf, info->pktLen, 0);
taosKeepTcpAlive(clientSocket);
memset(recvbuf, 0, BUFFER_SIZE);
int32_t nleft, nread;
char * ptr = recvbuf;
nleft = info->pktLen;
sprintf(buffer, "client send TCP pkg to %s:%d, content: 1122334455", taosIpStr(info->hostIp), info->port);
sprintf(buffer + info->pktLen - 16, "1122334455667788");
while (nleft > 0) {
nread = recv(clientSocket, ptr, BUFFER_SIZE, 0);;
if (nread == 0) {
break;
} else if (nread < 0) {
if (errno == EINTR) {
continue;
} else {
uError("faild to recv pkg from TCP port:%d since %s", info->port, strerror(errno));
taosCloseSocket(clientSocket);
return -1;
}
} else {
nleft -= nread;
ptr += nread;
iDataNum += nread;
}
int32_t ret = taosWriteMsg(clientSocket, buffer, info->pktLen);
if (ret < 0) {
uError("TCP: failed to write msg to %s:%d since %s", taosIpStr(info->hostIp), info->port, strerror(errno));
return -1;
}
if (iDataNum < info->pktLen) {
uError("TCP: received ack:%d bytes, less than send:%d bytes from port:%d", iDataNum, info->pktLen, info->port);
ret = taosReadMsg(clientSocket, buffer, info->pktLen);
if (ret < 0) {
uError("TCP: failed to read msg from %s:%d since %s", taosIpStr(info->hostIp), info->port, strerror(errno));
return -1;
}
......@@ -239,9 +227,9 @@ static int32_t taosNetCheckTcpPort(STestInfo *info) {
static int32_t taosNetCheckUdpPort(STestInfo *info) {
SOCKET clientSocket;
char sendbuf[BUFFER_SIZE];
char recvbuf[BUFFER_SIZE];
char buffer[BUFFER_SIZE] = {0};
int32_t iDataNum = 0;
int32_t bufSize = 1024000;
struct sockaddr_in serverAddr;
......@@ -250,41 +238,39 @@ static int32_t taosNetCheckUdpPort(STestInfo *info) {
return -1;
}
// set overtime
struct timeval timeout;
timeout.tv_sec = 2; // s
timeout.tv_usec = 0; // us
if (setsockopt(clientSocket, SOL_SOCKET, SO_SNDTIMEO, (char *)&timeout, sizeof(struct timeval)) == -1) {
uError("failed to setsockopt send timer since %s", strerror(errno));
if (taosSetSockOpt(clientSocket, SOL_SOCKET, SO_SNDBUF, (void *)&bufSize, sizeof(bufSize)) != 0) {
uError("failed to set the send buffer size for UDP socket\n");
return -1;
}
if (setsockopt(clientSocket, SOL_SOCKET, SO_RCVTIMEO, (char *)&timeout, sizeof(struct timeval)) == -1) {
uError("failed to setsockopt recv timer since %s", strerror(errno));
if (taosSetSockOpt(clientSocket, SOL_SOCKET, SO_RCVBUF, (void *)&bufSize, sizeof(bufSize)) != 0) {
uError("failed to set the receive buffer size for UDP socket\n");
return -1;
}
serverAddr.sin_family = AF_INET;
serverAddr.sin_port = htons(info->port);
serverAddr.sin_addr.s_addr = info->hostIp;
memset(sendbuf, 0, BUFFER_SIZE);
memset(recvbuf, 0, BUFFER_SIZE);
struct in_addr ipStr;
memcpy(&ipStr, &info->hostIp, 4);
sprintf(sendbuf, "client send udp pkg to %s:%d, content: 1122334455", taosInetNtoa(ipStr), info->port);
sprintf(sendbuf + info->pktLen - 16, "1122334455667788");
sprintf(buffer, "client send udp pkg to %s:%d, content: 1122334455", taosInetNtoa(ipStr), info->port);
sprintf(buffer + info->pktLen - 16, "1122334455667788");
socklen_t sin_size = sizeof(*(struct sockaddr *)&serverAddr);
int32_t code = sendto(clientSocket, sendbuf, info->pktLen, 0, (struct sockaddr *)&serverAddr, (int32_t)sin_size);
if (code < 0) {
uError("failed to perform sendto func since %s", strerror(errno));
iDataNum = taosSendto(clientSocket, buffer, info->pktLen, 0, (struct sockaddr *)&serverAddr, (int32_t)sin_size);
if (iDataNum < 0 || iDataNum != info->pktLen) {
uError("UDP: failed to perform sendto func since %s", strerror(errno));
return -1;
}
iDataNum = recvfrom(clientSocket, recvbuf, BUFFER_SIZE, 0, (struct sockaddr *)&serverAddr, &sin_size);
memset(buffer, 0, BUFFER_SIZE);
sin_size = sizeof(*(struct sockaddr *)&serverAddr);
iDataNum = recvfrom(clientSocket, buffer, BUFFER_SIZE, 0, (struct sockaddr *)&serverAddr, &sin_size);
if (iDataNum < info->pktLen) {
uError("UDP: received ack:%d bytes, less than send:%d bytes from port:%d", iDataNum, info->pktLen, info->port);
if (iDataNum < 0 || iDataNum != info->pktLen) {
uError("UDP: received ack:%d bytes(expect:%d) from port:%d since %s", iDataNum, info->pktLen, info->port, strerror(errno));
return -1;
}
......@@ -304,19 +290,18 @@ static void taosNetCheckPort(uint32_t hostIp, int32_t startPort, int32_t endPort
info.port = port;
ret = taosNetCheckTcpPort(&info);
if (ret != 0) {
uError("failed to test tcp port:%d", port);
uError("failed to test TCP port:%d", port);
} else {
uInfo("successed to test tcp port:%d", port);
uInfo("successed to test TCP port:%d", port);
}
ret = taosNetCheckUdpPort(&info);
if (ret != 0) {
uError("failed to test udp port:%d", port);
uError("failed to test UDP port:%d", port);
} else {
uInfo("successed to test udp port:%d", port);
uInfo("successed to test UDP port:%d", port);
}
}
return;
}
void *taosNetInitRpc(char *secretEncrypt, char spi) {
......@@ -440,9 +425,9 @@ static void taosNetTestRpc(char *host, int32_t startPort, int32_t pkgLen) {
int32_t ret = taosNetCheckRpc(host, port, sendpkgLen, spi, NULL);
if (ret < 0) {
uError("failed to test tcp port:%d", port);
uError("failed to test TCP port:%d", port);
} else {
uInfo("successed to test tcp port:%d", port);
uInfo("successed to test TCP port:%d", port);
}
if (pkgLen >= tsRpcMaxUdpSize) {
......@@ -453,9 +438,9 @@ static void taosNetTestRpc(char *host, int32_t startPort, int32_t pkgLen) {
ret = taosNetCheckRpc(host, port, pkgLen, spi, NULL);
if (ret < 0) {
uError("failed to test udp port:%d", port);
uError("failed to test UDP port:%d", port);
} else {
uInfo("successed to test udp port:%d", port);
uInfo("successed to test UDP port:%d", port);
}
}
}
......@@ -492,14 +477,15 @@ static void taosNetTestServer(char *host, int32_t startPort, int32_t pkgLen) {
tcpInfo->pktLen = pkgLen;
if (pthread_create(pids + i, NULL, taosNetBindTcpPort, tcpInfo) != 0) {
uInfo("failed to create tcp test thread, %s:%d", tcpInfo->hostFqdn, tcpInfo->port);
uInfo("failed to create TCP test thread, %s:%d", tcpInfo->hostFqdn, tcpInfo->port);
exit(-1);
}
STestInfo *udpInfo = uinfos + i;
udpInfo->port = (uint16_t)(port + i);
udpInfo->port = port + i;
tcpInfo->pktLen = pkgLen;
if (pthread_create(pids + num + i, NULL, taosNetBindUdpPort, udpInfo) != 0) {
uInfo("failed to create udp test thread, %s:%d", tcpInfo->hostFqdn, tcpInfo->port);
uInfo("failed to create UDP test thread, %s:%d", tcpInfo->hostFqdn, tcpInfo->port);
exit(-1);
}
}
......
src/util/src/tnote.c
浏览文件 @ 9598664f
......@@ -13,277 +13,260 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "os.h"
#include "tutil.h"
#include "tglobal.h"
#include "tnote.h"
taosNoteInfo m_HttpNote;
taosNoteInfo m_TscNote;
SNoteObj tsHttpNote;
SNoteObj tsTscNote;
SNoteObj tsInfoNote;
int taosOpenNoteWithMaxLines(char *fn, int maxLines, int maxNoteNum, taosNoteInfo * pNote);
static int32_t taosOpenNoteWithMaxLines(char *fn, int32_t maxLines, int32_t maxNoteNum, SNoteObj *pNote);
static void taosCloseNoteByFd(int32_t oldFd, SNoteObj *pNote);
void taosInitNote(int numOfNoteLines, int maxNotes, char* lable)
{
taosNoteInfo * pNote = NULL;
char temp[128] = { 0 };
static void taosInitNote(int32_t numOfLines, int32_t maxNotes, SNoteObj *pNote, char *name) {
memset(pNote, 0, sizeof(SNoteObj));
pNote->fileNum = 1;
pNote->fd = -1;
if (strcasecmp(lable, "http_note") == 0) {
pNote = &m_HttpNote;
sprintf(temp, "%s/httpnote", tsLogDir);
} else if (strcasecmp(lable, "tsc_note") == 0) {
pNote = &m_TscNote;
sprintf(temp, "%s/tscnote-%d", tsLogDir, getpid());
} else {
return;
}
if (taosOpenNoteWithMaxLines(name, numOfLines, maxNotes, pNote) < 0) {
fprintf(stderr, "failed to init note file\n");
}
taosNotePrint(pNote, "==================================================");
taosNotePrint(pNote, "=================== new note ===================");
taosNotePrint(pNote, "==================================================");
}
memset(pNote, 0, sizeof(taosNoteInfo));
pNote->taosNoteFileNum = 1;
//pNote->taosNoteMaxLines = 0;
//pNote->taosNoteLines = 0;
//pNote->taosNoteFlag = 0;
pNote->taosNoteFd = -1;
//pNote->taosNoteOpenInProgress = 0;
void taosInitNotes() {
char name[TSDB_FILENAME_LEN * 2] = {0};
if (taosOpenNoteWithMaxLines(temp, numOfNoteLines, maxNotes, pNote) < 0)
fprintf(stderr, "failed to init note file\n");
if (tsTscEnableRecordSql) {
snprintf(name, TSDB_FILENAME_LEN * 2, "%s/tscsql-%d", tsLogDir, taosGetPId());
taosInitNote(tsNumOfLogLines, 1, &tsTscNote, name);
}
taosNotePrint(pNote, "==================================================");
taosNotePrint(pNote, "=================== new note ===================");
taosNotePrint(pNote, "==================================================");
if (tsHttpEnableRecordSql) {
snprintf(name, TSDB_FILENAME_LEN * 2, "%s/httpsql", tsLogDir);
taosInitNote(tsNumOfLogLines, 1, &tsHttpNote, name);
}
if (tscEmbedded == 1) {
snprintf(name, TSDB_FILENAME_LEN * 2, "%s/taosinfo", tsLogDir);
taosInitNote(tsNumOfLogLines, 1, &tsInfoNote, name);
}
}
void taosCloseNoteByFd(int oldFd, taosNoteInfo * pNote);
bool taosLockNote(int fd, taosNoteInfo * pNote)
{
if (fd < 0) return false;
static bool taosLockNote(int32_t fd, SNoteObj *pNote) {
if (fd < 0) return false;
if (pNote->taosNoteFileNum > 1) {
int ret = (int)(flock(fd, LOCK_EX | LOCK_NB));
if (ret == 0) {
return true;
}
if (pNote->fileNum > 1) {
int32_t ret = (int32_t)(flock(fd, LOCK_EX | LOCK_NB));
if (ret == 0) {
return true;
}
}
return false;
return false;
}
void taosUnLockNote(int fd, taosNoteInfo * pNote)
{
if (fd < 0) return;
static void taosUnLockNote(int32_t fd, SNoteObj *pNote) {
if (fd < 0) return;
if (pNote->taosNoteFileNum > 1) {
flock(fd, LOCK_UN | LOCK_NB);
}
if (pNote->fileNum > 1) {
flock(fd, LOCK_UN | LOCK_NB);
}
}
void *taosThreadToOpenNewNote(void *param)
{
char name[NOTE_FILE_NAME_LEN * 2];
taosNoteInfo * pNote = (taosNoteInfo *)param;
static void *taosThreadToOpenNewNote(void *param) {
char name[NOTE_FILE_NAME_LEN * 2];
SNoteObj *pNote = (SNoteObj *)param;
pNote->taosNoteFlag ^= 1;
pNote->taosNoteLines = 0;
sprintf(name, "%s.%d", pNote->taosNoteName, pNote->taosNoteFlag);
pNote->flag ^= 1;
pNote->lines = 0;
sprintf(name, "%s.%d", pNote->name, pNote->flag);
umask(0);
umask(0);
int fd = open(name, O_WRONLY | O_CREAT | O_TRUNC, S_IRWXU | S_IRWXG | S_IRWXO);
if (fd < 0) {
return NULL;
}
int32_t fd = open(name, O_WRONLY | O_CREAT | O_TRUNC, S_IRWXU | S_IRWXG | S_IRWXO);
if (fd < 0) {
return NULL;
}
taosLockNote(fd, pNote);
(void)lseek(fd, 0, SEEK_SET);
taosLockNote(fd, pNote);
(void)lseek(fd, 0, SEEK_SET);
int oldFd = pNote->taosNoteFd;
pNote->taosNoteFd = fd;
pNote->taosNoteLines = 0;
pNote->taosNoteOpenInProgress = 0;
taosNotePrint(pNote, "=============== new note is opened =============");
int32_t oldFd = pNote->fd;
pNote->fd = fd;
pNote->lines = 0;
pNote->openInProgress = 0;
taosNotePrint(pNote, "=============== new note is opened =============");
taosCloseNoteByFd(oldFd, pNote);
return NULL;
taosCloseNoteByFd(oldFd, pNote);
return NULL;
}
int taosOpenNewNote(taosNoteInfo * pNote)
{
pthread_mutex_lock(&pNote->taosNoteMutex);
static int32_t taosOpenNewNote(SNoteObj *pNote) {
pthread_mutex_lock(&pNote->mutex);
if (pNote->taosNoteLines > pNote->taosNoteMaxLines && pNote->taosNoteOpenInProgress == 0) {
pNote->taosNoteOpenInProgress = 1;
if (pNote->lines > pNote->maxLines && pNote->openInProgress == 0) {
pNote->openInProgress = 1;
taosNotePrint(pNote, "=============== open new note ==================");
pthread_t pattern;
pthread_attr_t attr;
pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
taosNotePrint(pNote, "=============== open new note ==================");
pthread_t pattern;
pthread_attr_t attr;
pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
pthread_create(&pattern, &attr, taosThreadToOpenNewNote, (void*)pNote);
pthread_attr_destroy(&attr);
}
pthread_create(&pattern, &attr, taosThreadToOpenNewNote, (void *)pNote);
pthread_attr_destroy(&attr);
}
pthread_mutex_unlock(&pNote->taosNoteMutex);
pthread_mutex_unlock(&pNote->mutex);
return pNote->taosNoteFd;
return pNote->fd;
}
bool taosCheckNoteIsOpen(char *noteName, taosNoteInfo * pNote)
{
/*
int exist = access(noteName, F_OK);
if (exist != 0) {
return false;
}
*/
int fd = open(noteName, O_WRONLY | O_CREAT, S_IRWXU | S_IRWXG | S_IRWXO);
if (fd < 0) {
fprintf(stderr, "failed to open note:%s reason:%s\n", noteName, strerror(errno));
return true;
}
static bool taosCheckNoteIsOpen(char *noteName, SNoteObj *pNote) {
int32_t fd = open(noteName, O_WRONLY | O_CREAT, S_IRWXU | S_IRWXG | S_IRWXO);
if (fd < 0) {
fprintf(stderr, "failed to open note:%s reason:%s\n", noteName, strerror(errno));
return true;
}
if (taosLockNote(fd, pNote)) {
taosUnLockNote(fd, pNote);
close(fd);
return false;
}
else {
close(fd);
return true;
}
if (taosLockNote(fd, pNote)) {
taosUnLockNote(fd, pNote);
close(fd);
return false;
} else {
close(fd);
return true;
}
}
void taosGetNoteName(char *fn, taosNoteInfo * pNote)
{
if (pNote->taosNoteFileNum > 1) {
for (int i = 0; i < pNote->taosNoteFileNum; i++) {
char fileName[NOTE_FILE_NAME_LEN];
static void taosGetNoteName(char *fn, SNoteObj *pNote) {
if (pNote->fileNum > 1) {
for (int32_t i = 0; i < pNote->fileNum; i++) {
char fileName[NOTE_FILE_NAME_LEN];
sprintf(fileName, "%s%d.0", fn, i);
bool file1open = taosCheckNoteIsOpen(fileName, pNote);
sprintf(fileName, "%s%d.0", fn, i);
bool file1open = taosCheckNoteIsOpen(fileName, pNote);
sprintf(fileName, "%s%d.1", fn, i);
bool file2open = taosCheckNoteIsOpen(fileName, pNote);
sprintf(fileName, "%s%d.1", fn, i);
bool file2open = taosCheckNoteIsOpen(fileName, pNote);
if (!file1open && !file2open) {
sprintf(pNote->taosNoteName, "%s%d", fn, i);
return;
}
}
if (!file1open && !file2open) {
sprintf(pNote->name, "%s%d", fn, i);
return;
}
}
}
if (strlen(fn) < NOTE_FILE_NAME_LEN) {
strcpy(pNote->taosNoteName, fn);
}
if (strlen(fn) < NOTE_FILE_NAME_LEN) {
strcpy(pNote->name, fn);
}
}
int taosOpenNoteWithMaxLines(char *fn, int maxLines, int maxNoteNum, taosNoteInfo * pNote)
{
char name[NOTE_FILE_NAME_LEN * 2] = "\0";
struct stat notestat0, notestat1;
int size;
pNote->taosNoteMaxLines = maxLines;
pNote->taosNoteFileNum = maxNoteNum;
taosGetNoteName(fn, pNote);
static int32_t taosOpenNoteWithMaxLines(char *fn, int32_t maxLines, int32_t maxNoteNum, SNoteObj *pNote) {
char name[NOTE_FILE_NAME_LEN * 2] = {0};
int32_t size;
struct stat logstat0, logstat1;
if (strlen(fn) > NOTE_FILE_NAME_LEN * 2 - 2) {
fprintf(stderr, "the len of file name overflow:%s\n", fn);
return -1;
}
pNote->maxLines = maxLines;
pNote->fileNum = maxNoteNum;
taosGetNoteName(fn, pNote);
if (strlen(fn) < NOTE_FILE_NAME_LEN + 50 - 2) {
strcpy(name, fn);
strcat(name, ".0");
}
bool log0Exist = stat(name, &logstat0) >= 0;
// if none of the note files exist, open 0, if both exists, open the old one
if (stat(name, &notestat0) < 0) {
pNote->taosNoteFlag = 0;
} else {
strcpy(name, fn);
strcat(name, ".1");
if (stat(name, &notestat1) < 0) {
pNote->taosNoteFlag = 1;
}
else {
pNote->taosNoteFlag = (notestat0.st_mtime > notestat1.st_mtime) ? 0 : 1;
}
}
char noteName[NOTE_FILE_NAME_LEN * 2] = "\0";
sprintf(noteName, "%s.%d", pNote->taosNoteName, pNote->taosNoteFlag);
pthread_mutex_init(&pNote->taosNoteMutex, NULL);
umask(0);
pNote->taosNoteFd = open(noteName, O_WRONLY | O_CREAT, S_IRWXU | S_IRWXG | S_IRWXO);
if (pNote->taosNoteFd < 0) {
fprintf(stderr, "failed to open note file:%s reason:%s\n", noteName, strerror(errno));
return -1;
}
taosLockNote(pNote->taosNoteFd, pNote);
// only an estimate for number of lines
struct stat filestat;
if (fstat(pNote->taosNoteFd, &filestat) < 0) {
fprintf(stderr, "failed to fstat note file:%s reason:%s\n", noteName, strerror(errno));
return -1;
}
size = (int)filestat.st_size;
pNote->taosNoteLines = size / 60;
if (strlen(fn) < NOTE_FILE_NAME_LEN + 50 - 2) {
strcpy(name, fn);
strcat(name, ".1");
}
bool log1Exist = stat(name, &logstat1) >= 0;
if (!log0Exist && !log1Exist) {
pNote->flag = 0;
} else if (!log1Exist) {
pNote->flag = 0;
} else if (!log0Exist) {
pNote->flag = 1;
} else {
pNote->flag = (logstat0.st_mtime > logstat1.st_mtime) ? 0 : 1;
}
char noteName[NOTE_FILE_NAME_LEN * 2] = {0};
sprintf(noteName, "%s.%d", pNote->name, pNote->flag);
pthread_mutex_init(&pNote->mutex, NULL);
umask(0);
pNote->fd = open(noteName, O_WRONLY | O_CREAT, S_IRWXU | S_IRWXG | S_IRWXO);
if (pNote->fd < 0) {
fprintf(stderr, "failed to open note file:%s reason:%s\n", noteName, strerror(errno));
return -1;
}
taosLockNote(pNote->fd, pNote);
// only an estimate for number of lines
struct stat filestat;
if (fstat(pNote->fd, &filestat) < 0) {
fprintf(stderr, "failed to fstat note file:%s reason:%s\n", noteName, strerror(errno));
return -1;
}
size = (int32_t)filestat.st_size;
pNote->lines = size / 60;
lseek(pNote->fd, 0, SEEK_END);
return 0;
}
lseek(pNote->taosNoteFd, 0, SEEK_END);
void taosNotePrintBuffer(SNoteObj *pNote, char *buffer, int32_t len) {
if (pNote->fd <= 0) return;
taosWrite(pNote->fd, buffer, len);
return 0;
if (pNote->maxLines > 0) {
pNote->lines++;
if ((pNote->lines > pNote->maxLines) && (pNote->openInProgress == 0)) taosOpenNewNote(pNote);
}
}
void taosNotePrint(taosNoteInfo * pNote, const char * const format, ...)
{
va_list argpointer;
char buffer[MAX_NOTE_LINE_SIZE+2];
int len;
struct tm Tm, *ptm;
struct timeval timeSecs;
time_t curTime;
gettimeofday(&timeSecs, NULL);
curTime = timeSecs.tv_sec;
ptm = localtime_r(&curTime, &Tm);
#ifndef LINUX
len = sprintf(buffer, "%02d/%02d %02d:%02d:%02d.%06d 0x%lld ", ptm->tm_mon + 1, ptm->tm_mday, ptm->tm_hour,
ptm->tm_min, ptm->tm_sec, (int)timeSecs.tv_usec, taosGetPthreadId());
#else
len = sprintf(buffer, "%02d/%02d %02d:%02d:%02d.%06d %lx ", ptm->tm_mon + 1, ptm->tm_mday, ptm->tm_hour, ptm->tm_min,
ptm->tm_sec, (int)timeSecs.tv_usec, (unsigned long int)pthread_self());
#endif
va_start(argpointer, format);
len += vsnprintf(buffer + len, MAX_NOTE_LINE_SIZE - len, format, argpointer);
va_end(argpointer);
if (len >= MAX_NOTE_LINE_SIZE) len = MAX_NOTE_LINE_SIZE - 2;
buffer[len++] = '\n';
buffer[len] = 0;
if (pNote->taosNoteFd >= 0) {
taosWrite(pNote->taosNoteFd, buffer, (unsigned int)len);
if (pNote->taosNoteMaxLines > 0) {
pNote->taosNoteLines++;
if ((pNote->taosNoteLines > pNote->taosNoteMaxLines) && (pNote->taosNoteOpenInProgress == 0))
taosOpenNewNote(pNote);
}
}
void taosNotePrint(SNoteObj *pNote, const char *const format, ...) {
va_list argpointer;
char buffer[MAX_NOTE_LINE_SIZE + 2];
int32_t len;
struct tm Tm, *ptm;
struct timeval timeSecs;
time_t curTime;
gettimeofday(&timeSecs, NULL);
curTime = timeSecs.tv_sec;
ptm = localtime_r(&curTime, &Tm);
len = sprintf(buffer, "%02d/%02d %02d:%02d:%02d.%06d 0x%08" PRIx64 " ", ptm->tm_mon + 1, ptm->tm_mday, ptm->tm_hour,
ptm->tm_min, ptm->tm_sec, (int32_t)timeSecs.tv_usec, taosGetPthreadId());
va_start(argpointer, format);
len += vsnprintf(buffer + len, MAX_NOTE_LINE_SIZE - len, format, argpointer);
va_end(argpointer);
if (len >= MAX_NOTE_LINE_SIZE) len = MAX_NOTE_LINE_SIZE - 2;
buffer[len++] = '\n';
buffer[len] = 0;
taosNotePrintBuffer(pNote, buffer, len);
}
void taosCloseNote(taosNoteInfo * pNote)
{
taosCloseNoteByFd(pNote->taosNoteFd, pNote);
}
// static void taosCloseNote(SNoteObj *pNote) { taosCloseNoteByFd(pNote->fd, pNote); }
void taosCloseNoteByFd(int fd, taosNoteInfo * pNote)
{
if (fd >= 0) {
taosUnLockNote(fd, pNote);
close(fd);
}
static void taosCloseNoteByFd(int32_t fd, SNoteObj *pNote) {
if (fd >= 0) {
taosUnLockNote(fd, pNote);
close(fd);
}
}
src/util/src/tqueue.c
浏览文件 @ 9598664f
......@@ -366,7 +366,7 @@ int taosReadQitemFromQset(taos_qset param, int *type, void **pitem, void **phand
queue->numOfItems--;
atomic_sub_fetch_32(&qset->numOfItems, 1);
code = 1;
uTrace("item:%p is read out from queue:%p, type:%d items:%d", *pitem, queue, *type, queue->numOfItems);
uTrace("item:%p is read out from queue:%p, type:%d items:%d", *pitem, queue, pNode->type, queue->numOfItems);
}
pthread_mutex_unlock(&queue->mutex);
......
src/vnode/inc/vnodeMain.h
浏览文件 @ 9598664f
......@@ -28,6 +28,7 @@ int32_t vnodeAlter(void *pVnode, SCreateVnodeMsg *pVnodeCfg);
int32_t vnodeClose(int32_t vgId);
int32_t vnodeReset(SVnodeObj *pVnode);
void vnodeCleanUp(SVnodeObj *pVnode);
void vnodeDestroy(SVnodeObj *pVnode);
#ifdef __cplusplus
......
src/vnode/inc/vnodeWorker.h
浏览文件 @ 9598664f
......@@ -23,8 +23,8 @@ extern "C" {
int32_t vnodeInitMWorker();
void vnodeCleanupMWorker();
int32_t vnodeOpenInMWorker(int32_t vgId, void *rpcHandle);
int32_t vnodeCleanupInMWorker(int32_t vgId, void *rpcHandle);
int32_t vnodeCleanupInMWorker(SVnodeObj *pVnode);
int32_t vnodeDestroyInMWorker(SVnodeObj *pVnode);
#ifdef __cplusplus
}
......
src/vnode/src/vnodeMain.c
浏览文件 @ 9598664f
......@@ -26,8 +26,9 @@
#include "vnodeSync.h"
#include "vnodeVersion.h"
#include "vnodeMgmt.h"
#include "vnodeWorker.h"
#include "vnodeMain.h"
static void vnodeCleanUp(SVnodeObj *pVnode);
static int32_t vnodeProcessTsdbStatus(void *arg, int32_t status, int32_t eno);
int32_t vnodeCreate(SCreateVnodeMsg *pVnodeCfg) {
......@@ -99,8 +100,10 @@ int32_t vnodeDrop(int32_t vgId) {
vInfo("vgId:%d, vnode will be dropped, refCount:%d pVnode:%p", pVnode->vgId, pVnode->refCount, pVnode);
pVnode->dropped = 1;
// remove from hash, so new messages wont be consumed
vnodeRemoveFromHash(pVnode);
vnodeRelease(pVnode);
vnodeCleanUp(pVnode);
vnodeCleanupInMWorker(pVnode);
return TSDB_CODE_SUCCESS;
}
......@@ -298,6 +301,7 @@ int32_t vnodeOpen(int32_t vgId) {
if (pVnode->sync <= 0) {
vError("vgId:%d, failed to open sync, replica:%d reason:%s", pVnode->vgId, pVnode->syncCfg.replica,
tstrerror(terrno));
vnodeRemoveFromHash(pVnode);
vnodeCleanUp(pVnode);
return terrno;
}
......@@ -311,6 +315,7 @@ int32_t vnodeClose(int32_t vgId) {
if (pVnode == NULL) return 0;
vDebug("vgId:%d, vnode will be closed, pVnode:%p", pVnode->vgId, pVnode);
vnodeRemoveFromHash(pVnode);
vnodeRelease(pVnode);
vnodeCleanUp(pVnode);
......@@ -387,10 +392,7 @@ void vnodeDestroy(SVnodeObj *pVnode) {
tsdbDecCommitRef(vgId);
}
static void vnodeCleanUp(SVnodeObj *pVnode) {
// remove from hash, so new messages wont be consumed
vnodeRemoveFromHash(pVnode);
void vnodeCleanUp(SVnodeObj *pVnode) {
if (!vnodeInInitStatus(pVnode)) {
// it may be in updateing or reset state, then it shall wait
int32_t i = 0;
......@@ -428,7 +430,6 @@ static int32_t vnodeProcessTsdbStatus(void *arg, int32_t status, int32_t eno) {
if (status == TSDB_STATUS_COMMIT_START) {
pVnode->isCommiting = 1;
pVnode->fversion = pVnode->version;
vDebug("vgId:%d, start commit, fver:%" PRIu64 " vver:%" PRIu64, pVnode->vgId, pVnode->fversion, pVnode->version);
if (!vnodeInInitStatus(pVnode)) {
return walRenew(pVnode->wal);
......@@ -437,9 +438,10 @@ static int32_t vnodeProcessTsdbStatus(void *arg, int32_t status, int32_t eno) {
}
if (status == TSDB_STATUS_COMMIT_OVER) {
vDebug("vgId:%d, commit over, fver:%" PRIu64 " vver:%" PRIu64, pVnode->vgId, pVnode->fversion, pVnode->version);
pVnode->isCommiting = 0;
pVnode->isFull = 0;
pVnode->fversion = pVnode->version;
vDebug("vgId:%d, commit over, fver:%" PRIu64 " vver:%" PRIu64, pVnode->vgId, pVnode->fversion, pVnode->version);
if (!vnodeInInitStatus(pVnode)) {
walRemoveOneOldFile(pVnode->wal);
}
......
src/vnode/src/vnodeMgmt.c
浏览文件 @ 9598664f
......@@ -114,7 +114,7 @@ void vnodeRelease(void *vparam) {
}
} else {
vDebug("vgId:%d, vnode will be destroyed, refCount:%d pVnode:%p", pVnode->vgId, refCount, pVnode);
vnodeDestroy(pVnode);
vnodeDestroyInMWorker(pVnode);
int32_t count = taosHashGetSize(tsVnodesHash);
vDebug("vgId:%d, vnode is destroyed, vnodes:%d", pVnode->vgId, count);
}
......
src/vnode/src/vnodeWorker.c
浏览文件 @ 9598664f
......@@ -21,10 +21,11 @@
#include "tqueue.h"
#include "tglobal.h"
#include "vnodeWorker.h"
#include "vnodeMain.h"
typedef enum {
VNODE_WORKER_ACTION_CREATE,
VNODE_WORKER_ACTION_DELETE
VNODE_WORKER_ACTION_CLEANUP,
VNODE_WORKER_ACTION_DESTROUY
} EVMWorkerAction;
typedef struct {
......@@ -132,14 +133,11 @@ void vnodeCleanupMWorker() {
vnodeStopMWorker();
}
static int32_t vnodeWriteIntoMWorker(int32_t vgId, EVMWorkerAction action,void *rpcHandle) {
static int32_t vnodeWriteIntoMWorker(SVnodeObj *pVnode, EVMWorkerAction action, void *rpcHandle) {
SVMWorkerMsg *pMsg = taosAllocateQitem(sizeof(SVMWorkerMsg));
if (pMsg == NULL) return TSDB_CODE_VND_OUT_OF_MEMORY;
SVnodeObj *pVnode = vnodeAcquire(vgId);
if (pVnode == NULL) return TSDB_CODE_VND_INVALID_VGROUP_ID;
pMsg->vgId = vgId;
pMsg->vgId = pVnode->vgId;
pMsg->pVnode = pVnode;
pMsg->rpcHandle = rpcHandle;
pMsg->action = action;
......@@ -150,29 +148,27 @@ static int32_t vnodeWriteIntoMWorker(int32_t vgId, EVMWorkerAction action,void *
return code;
}
int32_t vnodeOpenInMWorker(int32_t vgId, void *rpcHandle) {
vTrace("vgId:%d, will open in vmworker", vgId);
return vnodeWriteIntoMWorker(vgId, VNODE_WORKER_ACTION_CREATE, rpcHandle);
int32_t vnodeCleanupInMWorker(SVnodeObj *pVnode) {
vTrace("vgId:%d, will cleanup in vmworker", pVnode->vgId);
return vnodeWriteIntoMWorker(pVnode, VNODE_WORKER_ACTION_CLEANUP, NULL);
}
int32_t vnodeCleanupInMWorker(int32_t vgId, void *rpcHandle) {
vTrace("vgId:%d, will cleanup in vmworker", vgId);
return vnodeWriteIntoMWorker(vgId, VNODE_WORKER_ACTION_DELETE, rpcHandle);
int32_t vnodeDestroyInMWorker(SVnodeObj *pVnode) {
vTrace("vgId:%d, will destroy in vmworker", pVnode->vgId);
return vnodeWriteIntoMWorker(pVnode, VNODE_WORKER_ACTION_DESTROUY, NULL);
}
static void vnodeFreeMWorkerMsg(SVMWorkerMsg *pMsg) {
vTrace("vgId:%d, disposed in vmworker", pMsg->vgId);
vnodeRelease(pMsg->pVnode);
taosFreeQitem(pMsg);
}
static void vnodeSendVMWorkerRpcRsp(SVMWorkerMsg *pMsg) {
SRpcMsg rpcRsp = {
.handle = pMsg->rpcHandle,
.code = pMsg->code,
};
if (pMsg->rpcHandle != NULL) {
SRpcMsg rpcRsp = {.handle = pMsg->rpcHandle, .code = pMsg->code};
rpcSendResponse(&rpcRsp);
}
rpcSendResponse(&rpcRsp);
vnodeFreeMWorkerMsg(pMsg);
}
......@@ -180,11 +176,11 @@ static void vnodeProcessMWorkerMsg(SVMWorkerMsg *pMsg) {
pMsg->code = 0;
switch (pMsg->action) {
case VNODE_WORKER_ACTION_CREATE:
pMsg->code = vnodeOpen(pMsg->vgId);
case VNODE_WORKER_ACTION_CLEANUP:
vnodeCleanUp(pMsg->pVnode);
break;
case VNODE_WORKER_ACTION_DELETE:
pMsg->code = vnodeDrop(pMsg->vgId);
case VNODE_WORKER_ACTION_DESTROUY:
vnodeDestroy(pMsg->pVnode);
break;
default:
break;
......
tests/Jenkinsfile 0 → 100644
浏览文件 @ 9598664f
properties([pipelineTriggers([githubPush()])])
node {
git url: 'https://github.com/liuyq-617/TDengine'
}
def pre_test(){
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
sh '''
sudo rmtaos
'''
}
sh '''
cd ${WKC}
rm -rf *
cd ${WK}
git reset --hard
git checkout develop
git pull
cd ${WKC}
rm -rf *
mv ${WORKSPACE}/* .
cd ${WK}
export TZ=Asia/Harbin
date
rm -rf ${WK}/debug
mkdir debug
cd debug
cmake .. > /dev/null
make > /dev/null
make install > /dev/null
cd ${WKC}/tests
'''
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('python p1') {
agent{label 'p1'}
steps {
pre_test()
sh '''
cd ${WKC}/tests
./test-all.sh p1
date'''
}
}
stage('test_b1') {
agent{label 'master'}
steps {
pre_test()
sh '''
cd ${WKC}/tests
./test-all.sh b1
date'''
}
}
stage('test_crash_gen') {
agent{label "b2"}
steps {
pre_test()
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
./handle_crash_gen_val_log.sh
'''
}
sh '''
date
cd ${WKC}/tests
./test-all.sh b2
date
'''
}
}
stage('test_valgrind') {
agent{label "b3"}
steps {
pre_test()
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
sh '''
cd ${WKC}/tests/pytest
./valgrind-test.sh 2>&1 > mem-error-out.log
./handle_val_log.sh
'''
}
sh '''
date
cd ${WKC}/tests
./test-all.sh b3
date'''
}
}
stage('python p2'){
agent{label "p2"}
steps{
pre_test()
sh '''
date
cd ${WKC}/tests
./test-all.sh p2
date
'''
}
}
}
}
}
}
tests/examples/JDBC/mybatisplus-demo/pom.xml
浏览文件 @ 9598664f
......@@ -38,11 +38,16 @@
<artifactId>h2</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>druid</artifactId>
<version>1.1.17</version>
</dependency>
<dependency>
<groupId>com.taosdata.jdbc</groupId>
<artifactId>taos-jdbcdriver</artifactId>
<version>2.0.11</version>
<version>2.0.14</version>
</dependency>
<!-- https://mvnrepository.com/artifact/mysql/mysql-connector-java -->
......
tests/examples/JDBC/mybatisplus-demo/src/main/resources/application.yml
浏览文件 @ 9598664f
spring:
datasource:
# driver-class-name: org.h2.Driver
# schema: classpath:db/schema-mysql.sql
# data: classpath:db/data-mysql.sql
# url: jdbc:h2:mem:test
# username: root
# password: test
# driver-class-name: com.mysql.jdbc.Driver
# url: jdbc:mysql://master:3306/test?useSSL=false
# username: root
# password: 123456
driver-class-name: com.taosdata.jdbc.TSDBDriver
url: jdbc:TAOS://localhost:6030/mp_test
user: root
......@@ -20,6 +8,12 @@ spring:
locale: en_US.UTF-8
timezone: UTC-8
druid:
initial-size: 5
min-idle: 5
max-active: 5
mybatis-plus:
configuration:
map-underscore-to-camel-case: false
......
tests/perftest-scripts/perftest-query.sh
浏览文件 @ 9598664f
......@@ -65,7 +65,18 @@ function runQueryPerfTest {
echoInfo "Run Performance Test"
cd $WORK_DIR/TDengine/tests/pytest
python3 query/queryPerformance.py 0 | tee -a $PERFORMANCE_TEST_REPORT
python3 query/queryPerformance.py -c $LOCAL_COMMIT | tee -a $PERFORMANCE_TEST_REPORT
python3 insert/insertFromCSVPerformance.py -c $LOCAL_COMMIT | tee -a $PERFORMANCE_TEST_REPORT
yes | taosdemo -c /etc/taosperf/ -d taosdemo_insert_test -t 1000 -n 1000 > taosdemoperf.txt
CREATETABLETIME=`grep 'Spent' taosdemoperf.txt | awk 'NR==1{print $2}'`
INSERTRECORDSTIME=`grep 'Spent' taosdemoperf.txt | awk 'NR==2{print $2}'`
REQUESTSPERSECOND=`grep 'Spent' taosdemoperf.txt | awk 'NR==2{print $13}'`
python3 tools/taosdemoPerformance.py -c $LOCAL_COMMIT -t $CREATETABLETIME -i $INSERTRECORDSTIME -r $REQUESTSPERSECOND | tee -a $PERFORMANCE_TEST_REPORT
[ -f taosdemoperf.txt ] && rm taosdemoperf.txt
}
......
tests/pytest/bug2265.py 0 → 100644
浏览文件 @ 9598664f
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import sys
from util.log import *
from util.cases import *
from util.sql import *
from util.dnodes import *
import taos
if __name__ == "__main__":
logSql = True
deployPath = ""
testCluster = False
valgrind = 0
print("start to execute %s" % __file__)
tdDnodes.init(deployPath)
tdDnodes.setTestCluster(testCluster)
tdDnodes.setValgrind(valgrind)
tdDnodes.stopAll()
tdDnodes.addSimExtraCfg("maxSQLLength", "1048576")
tdDnodes.deploy(1)
tdDnodes.start(1)
host = '127.0.0.1'
tdLog.info("Procedures for tdengine deployed in %s" % (host))
tdCases.logSql(logSql)
print('1')
conn = taos.connect(
host,
config=tdDnodes.getSimCfgPath())
tdSql.init(conn.cursor(), True)
print("==========step1")
print("create table ")
tdSql.execute("create database db")
tdSql.execute("use db")
tdSql.execute("create table t1 (ts timestamp, c1 int,c2 int ,c3 int)")
print("==========step2")
print("insert maxSQLLength data ")
data = 'insert into t1 values'
ts = 1604298064000
i = 0
while ((len(data)<(1024*1024)) & (i < 32767 - 1) ):
data += '(%s,%d,%d,%d)'%(ts+i,i%1000,i%1000,i%1000)
i+=1
tdSql.execute(data)
print("==========step4")
print("insert data batch larger than 32767 ")
i = 0
while ((len(data)<(1024*1024)) & (i < 32767) ):
data += '(%s,%d,%d,%d)'%(ts+i,i%1000,i%1000,i%1000)
i+=1
tdSql.error(data)
print("==========step4")
print("insert data larger than maxSQLLength ")
tdSql.execute("create table t2 (ts timestamp, c1 binary(50))")
data = 'insert into t2 values'
i = 0
while ((len(data)<(1024*1024)) & (i < 32767 - 1 ) ):
data += '(%s,%s)'%(ts+i,'a'*50)
i+=1
tdSql.error(data)
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tests/pytest/client/noConnectionErrorTest.py 0 → 100644
浏览文件 @ 9598664f
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import sys
from util.log import *
from util.cases import *
from util.sql import *
from util.dnodes import tdDnodes
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
def run(self):
tdSql.prepare()
tdDnodes.stop(1)
sql = "use db"
try:
tdSql.execute(sql)
except Exception as e:
expectError = 'Unable to establish connection'
if expectError in str(e):
pass
else:
caller = inspect.getframeinfo(inspect.stack()[1][1])
tdLog.exit("%s(%d) failed: sql:%s, expect error not occured" % (caller.filename, caller.lineno, sql))
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())
tests/pytest/fulltest.sh
浏览文件 @ 9598664f
......@@ -20,6 +20,7 @@ python3 insert/retentionpolicy.py
python3 ./test.py -f insert/alterTableAndInsert.py
python3 ./test.py -f insert/insertIntoTwoTables.py
python3 ./test.py -f insert/before_1970.py
python3 bug2265.py
python3 ./test.py -f table/alter_wal0.py
python3 ./test.py -f table/column_name.py
......@@ -27,6 +28,8 @@ python3 ./test.py -f table/column_num.py
python3 ./test.py -f table/db_table.py
python3 ./test.py -f table/create_sensitive.py
#python3 ./test.py -f table/tablename-boundary.py
python3 ./test.py -f table/max_table_length.py
# tag
python3 ./test.py -f tag_lite/filter.py
......@@ -162,8 +165,10 @@ python3 ./test.py -f query/bug1876.py
python3 ./test.py -f query/bug2218.py
python3 ./test.py -f query/bug2117.py
python3 ./test.py -f query/bug2143.py
python3 ./test.py -f query/sliding.py
python3 ./test.py -f query/sliding.py
python3 ./test.py -f query/unionAllTest.py
python3 ./test.py -f query/bug2281.py
python3 ./test.py -f query/bug2119.py
#stream
python3 ./test.py -f stream/metric_1.py
python3 ./test.py -f stream/new.py
......@@ -179,6 +184,7 @@ python3 ./test.py -f alter/alter_table_crash.py
python3 ./test.py -f client/client.py
python3 ./test.py -f client/version.py
python3 ./test.py -f client/alterDatabase.py
python3 ./test.py -f client/noConnectionErrorTest.py
# Misc
python3 testCompress.py
......@@ -213,6 +219,7 @@ python3 test.py -f query/queryFillTest.py
python3 test.py -f tools/taosdemoTest.py
python3 test.py -f tools/taosdumpTest.py
python3 test.py -f tools/lowaTest.py
python3 test.py -f tools/taosdemoTest2.py
# subscribe
python3 test.py -f subscribe/singlemeter.py
......
tests/pytest/insert/insertFromCSVPerformance.py 0 → 100644
浏览文件 @ 9598664f
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import sys
import taos
import time
import datetime
import csv
import random
import pandas as pd
import argparse
import os.path
class insertFromCSVPerformace:
def __init__(self, commitID, dbName, stbName, branchName):
self.commitID = commitID
self.dbName = dbName
self.stbName = stbName
self.branchName = branchName
self.ts = 1500074556514
self.host = "127.0.0.1"
self.user = "root"
self.password = "taosdata"
self.config = "/etc/taosperf"
self.conn = taos.connect(
self.host,
self.user,
self.password,
self.config)
def writeCSV(self):
with open('test3.csv','w', encoding='utf-8', newline='') as csvFile:
writer = csv.writer(csvFile, dialect='excel')
for i in range(1000000):
newTimestamp = self.ts + random.randint(10000000, 10000000000) + random.randint(1000, 10000000) + random.randint(1, 1000)
d = datetime.datetime.fromtimestamp(newTimestamp / 1000)
dt = str(d.strftime("%Y-%m-%d %H:%M:%S.%f"))
writer.writerow(["'%s'" % dt, random.randint(1, 100), random.uniform(1, 100), random.randint(1, 100), random.randint(1, 100)])
def removCSVHeader(self):
data = pd.read_csv("ordered.csv")
data = data.drop([0])
data.to_csv("ordered.csv", header = False, index = False)
def createTables(self):
cursor = self.conn.cursor()
cursor.execute("create database if not exists %s" % self.dbName)
cursor.execute("use %s" % self.dbName)
cursor.execute("create table if not exists %s(ts timestamp, in_order_time float, out_of_order_time float, commit_id binary(50)) tags(branch binary(50))" % self.stbName)
cursor.execute("create table if not exists %s using %s tags('%s')" % (self.branchName, self.stbName, self.branchName))
cursor.execute("create table if not exists t1(ts timestamp, c1 int, c2 float, c3 int, c4 int)")
cursor.execute("create table if not exists t2(ts timestamp, c1 int, c2 float, c3 int, c4 int)")
cursor.close()
def run(self):
cursor = self.conn.cursor()
cursor.execute("use %s" % self.dbName)
print("==================== CSV insert performance ====================")
totalTime = 0
for i in range(10):
cursor.execute("create table if not exists t1(ts timestamp, c1 int, c2 float, c3 int, c4 int)")
startTime = time.time()
cursor.execute("insert into t1 file 'outoforder.csv'")
totalTime += time.time() - startTime
cursor.execute("drop table if exists t1")
out_of_order_time = (float) (totalTime / 10)
print("Out of Order - Insert time: %f" % out_of_order_time)
totalTime = 0
for i in range(10):
cursor.execute("create table if not exists t2(ts timestamp, c1 int, c2 float, c3 int, c4 int)")
startTime = time.time()
cursor.execute("insert into t2 file 'ordered.csv'")
totalTime += time.time() - startTime
cursor.execute("drop table if exists t2")
in_order_time = (float) (totalTime / 10)
print("In order - Insert time: %f" % in_order_time)
cursor.execute("insert into %s values(now, %f, %f, '%s')" % (self.branchName, in_order_time, out_of_order_time, self.commitID))
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument(
'-c',
'--commit-id',
action='store',
default='null',
type=str,
help='git commit id (default: null)')
parser.add_argument(
'-d',
'--database-name',
action='store',
default='perf',
type=str,
help='Database name to be created (default: perf)')
parser.add_argument(
'-t',
'--stable-name',
action='store',
default='csv_insert',
type=str,
help='Database name to be created (default: csv_insert)')
parser.add_argument(
'-b',
'--branch-name',
action='store',
default='develop',
type=str,
help='branch name (default: develop)')
args = parser.parse_args()
perftest = insertFromCSVPerformace(args.commit_id, args.database_name, args.stable_name, args.branch_name)
perftest.createTables()
perftest.run()
\ No newline at end of file
tests/pytest/pytest_1.sh
浏览文件 @ 9598664f
......@@ -26,6 +26,7 @@ python3 ./test.py -f table/column_num.py
python3 ./test.py -f table/db_table.py
python3 ./test.py -f table/create_sensitive.py
#python3 ./test.py -f table/tablename-boundary.py
python3 ./test.py -f table/max_table_length.py
# tag
python3 ./test.py -f tag_lite/filter.py
......@@ -159,7 +160,9 @@ python3 ./test.py -f query/bug1874.py
python3 ./test.py -f query/bug1875.py
python3 ./test.py -f query/bug1876.py
python3 ./test.py -f query/bug2218.py
python3 ./test.py -f query/bug2281.py
python3 ./test.py -f query/bug2119.py
python3 bug2265.py
#stream
python3 ./test.py -f stream/metric_1.py
python3 ./test.py -f stream/new.py
......
tests/pytest/query/bug2119.py 0 → 100644
浏览文件 @ 9598664f
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import sys
from util.log import *
from util.cases import *
from util.sql import *
from util.dnodes import *
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
def run(self):
tdSql.prepare()
print("==========step1")
print("create table && insert data")
tdSql.execute("create table t1 (ts timestamp, c1 int, c2 float)")
print("==========step2")
print("query percentile from blank table")
tdSql.query('select percentile(c1,1) from t1')
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())
\ No newline at end of file
tests/pytest/query/bug2281.py 0 → 100644
浏览文件 @ 9598664f
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import sys
from util.log import *
from util.cases import *
from util.sql import *
from util.dnodes import *
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
def run(self):
tdSql.prepare()
print("==========step1")
print("create table && insert data")
tdSql.execute("create table t1 (ts timestamp, c1 int, c2 float)")
insertRows = 10
t0 = 1604298064000
tdLog.info("insert %d rows" % (insertRows))
for i in range(insertRows):
ret = tdSql.execute(
"insert into t1 values (%d , %d,%d)" %
(t0+i,i%100,i/2.0))
print("==========step2")
print("query diff && top")
tdSql.error('select diff(c1),top(c2) from t1')
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())
\ No newline at end of file
tests/pytest/query/queryPerformance.py
浏览文件 @ 9598664f
......@@ -16,10 +16,16 @@ import sys
import os
import taos
import time
import argparse
class taosdemoQueryPerformace:
def initConnection(self):
def __init__(self, clearCache, commitID, dbName, stbName, tbPerfix):
self.clearCache = clearCache
self.commitID = commitID
self.dbName = dbName
self.stbName = stbName
self.tbPerfix = tbPerfix
self.host = "127.0.0.1"
self.user = "root"
self.password = "taosdata"
......@@ -30,92 +36,109 @@ class taosdemoQueryPerformace:
self.password,
self.config)
def createPerfTables(self):
cursor = self.conn.cursor()
cursor.execute("create database if not exists %s" % self.dbName)
cursor.execute("use %s" % self.dbName)
cursor.execute("create table if not exists %s(ts timestamp, query_time float, commit_id binary(50)) tags(query_id int, query_sql binary(300))" % self.stbName)
sql = "select count(*) from test.meters"
tableid = 1
cursor.execute("create table if not exists %s%d using %s tags(%d, '%s')" % (self.tbPerfix, tableid, self.stbName, tableid, sql))
sql = "select avg(f1), max(f2), min(f3) from test.meters"
tableid = 2
cursor.execute("create table if not exists %s%d using %s tags(%d, '%s')" % (self.tbPerfix, tableid, self.stbName, tableid, sql))
sql = "select count(*) from test.meters where loc='beijing'"
tableid = 3
cursor.execute("create table if not exists %s%d using %s tags(%d, \"%s\")" % (self.tbPerfix, tableid, self.stbName, tableid, sql))
sql = "select avg(f1), max(f2), min(f3) from test.meters where areaid=10"
tableid = 4
cursor.execute("create table if not exists %s%d using %s tags(%d, '%s')" % (self.tbPerfix, tableid, self.stbName, tableid, sql))
sql = "select avg(f1), max(f2), min(f3) from test.t10 interval(10s)"
tableid = 5
cursor.execute("create table if not exists %s%d using %s tags(%d, '%s')" % (self.tbPerfix, tableid, self.stbName, tableid, sql))
sql = "select last_row(*) from meters"
tableid = 6
cursor.execute("create table if not exists %s%d using %s tags(%d, '%s')" % (self.tbPerfix, tableid, self.stbName, tableid, sql))
sql = "select * from meters"
tableid = 7
cursor.execute("create table if not exists %s%d using %s tags(%d, '%s')" % (self.tbPerfix, tableid, self.stbName, tableid, sql))
sql = "select avg(f1), max(f2), min(f3) from meters where ts <= '2017-07-15 10:40:01.000' and ts <= '2017-07-15 14:00:40.000'"
tableid = 8
cursor.execute("create table if not exists %s%d using %s tags(%d, \"%s\")" % (self.tbPerfix, tableid, self.stbName, tableid, sql))
cursor.close()
def query(self):
cursor = self.conn.cursor()
cursor.execute("use test")
totalTime = 0
for i in range(100):
if(sys.argv[1] == '1'):
# root permission is required
os.system("echo 3 > /proc/sys/vm/drop_caches")
startTime = time.time()
cursor.execute("select count(*) from test.meters")
totalTime += time.time() - startTime
print("query time for: select count(*) from test.meters %f seconds" % (totalTime / 100))
totalTime = 0
for i in range(100):
if(sys.argv[1] == '1'):
# root permission is required
os.system("echo 3 > /proc/sys/vm/drop_caches")
startTime = time.time()
cursor.execute("select avg(f1), max(f2), min(f3) from test.meters")
totalTime += time.time() - startTime
print("query time for: select avg(f1), max(f2), min(f3) from test.meters %f seconds" % (totalTime / 100))
totalTime = 0
for i in range(100):
if(sys.argv[1] == '1'):
# root permission is required
os.system("echo 3 > /proc/sys/vm/drop_caches")
startTime = time.time()
cursor.execute("select count(*) from test.meters where loc='beijing'")
totalTime += time.time() - startTime
print("query time for: select count(*) from test.meters where loc='beijing' %f seconds" % (totalTime / 100))
totalTime = 0
for i in range(100):
if(sys.argv[1] == '1'):
# root permission is required
os.system("echo 3 > /proc/sys/vm/drop_caches")
startTime = time.time()
cursor.execute("select avg(f1), max(f2), min(f3) from test.meters where areaid=10")
totalTime += time.time() - startTime
print("query time for: select avg(f1), max(f2), min(f3) from test.meters where areaid=10 %f seconds" % (totalTime / 100))
totalTime = 0
for i in range(100):
if(sys.argv[1] == '1'):
# root permission is required
os.system("echo 3 > /proc/sys/vm/drop_caches")
startTime = time.time()
cursor.execute("select avg(f1), max(f2), min(f3) from test.t10 interval(10s)")
totalTime += time.time() - startTime
print("query time for: select avg(f1), max(f2), min(f3) from test.t10 interval(10s) %f seconds" % (totalTime / 100))
totalTime = 0
for i in range(100):
if(sys.argv[1] == '1'):
# root permission is required
os.system("echo 3 > /proc/sys/vm/drop_caches")
startTime = time.time()
cursor.execute("select last_row(*) from meters")
totalTime += time.time() - startTime
print("query time for: select last_row(*) from meters %f seconds" % (totalTime / 100))
totalTime = 0
for i in range(100):
if(sys.argv[1] == '1'):
# root permission is required
os.system("echo 3 > /proc/sys/vm/drop_caches")
startTime = time.time()
cursor.execute("select * from meters")
totalTime += time.time() - startTime
print("query time for: select * from meters %f seconds" % (totalTime / 100))
totalTime = 0
for i in range(100):
if(sys.argv[1] == '1'):
# root permission is required
os.system("echo 3 > /proc/sys/vm/drop_caches")
startTime = time.time()
cursor.execute("select avg(f1), max(f2), min(f3) from meters where ts <= '2017-07-15 10:40:01.000' and ts <= '2017-07-15 14:00:40.000'")
totalTime += time.time() - startTime
print("query time for: select avg(f1), max(f2), min(f3) from meters where ts <= '2017-07-15 10:40:01.000' and ts <= '2017-07-15 14:00:40.000' %f seconds" % (totalTime / 100))
cursor = self.conn.cursor()
print("==================== query performance ====================")
cursor.execute("use %s" % self.dbName)
cursor.execute("select tbname, query_id, query_sql from %s" % self.stbName)
for data in cursor:
table_name = data[0]
query_id = data[1]
sql = data[2]
totalTime = 0
cursor2 = self.conn.cursor()
cursor2.execute("use test")
for i in range(100):
if(self.clearCache == True):
# root permission is required
os.system("echo 3 > /proc/sys/vm/drop_caches")
startTime = time.time()
cursor2.execute(sql)
totalTime += time.time() - startTime
cursor2.close()
print("query time for: %s %f seconds" % (sql, totalTime / 100))
cursor3 = self.conn.cursor()
cursor3.execute("insert into %s.%s values(now, %f, '%s')" % (self.dbName, table_name, totalTime / 100, self.commitID))
cursor3.close()
cursor.close()
if __name__ == '__main__':
perftest = taosdemoQueryPerformace()
perftest.initConnection()
perftest.query()
\ No newline at end of file
parser = argparse.ArgumentParser()
parser.add_argument(
'-r',
'--remove-cache',
action='store_true',
default=False,
help='clear cache before query (default: False)')
parser.add_argument(
'-c',
'--commit-id',
action='store',
default='null',
type=str,
help='git commit id (default: null)')
parser.add_argument(
'-d',
'--database-name',
action='store',
default='perf',
type=str,
help='Database name to be created (default: perf)')
parser.add_argument(
'-t',
'--stable-name',
action='store',
default='query_tb',
type=str,
help='table name to be created (default: query_tb)')
parser.add_argument(
'-p',
'--table-perfix',
action='store',
default='q',
type=str,
help='table name perfix (default: q)')
args = parser.parse_args()
perftest = taosdemoQueryPerformace(args.remove_cache, args.commit_id, args.database_name, args.stable_name, args.table_perfix)
perftest.createPerfTables()
perftest.query()
tests/pytest/query/unionAllTest.py 0 → 100644
浏览文件 @ 9598664f
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import sys
import taos
from util.log import tdLog
from util.cases import tdCases
from util.sql import tdSql
import random
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
self.ts = 1500000000000
self.num = 10
def run(self):
tdSql.prepare()
tdSql.execute("create table st(ts timestamp, c1 int) tags(loc nchar(20))")
tdSql.execute("create table t0 using st tags('nchar0')")
tdSql.execute("create table t1 using st tags('nchar1')")
tdSql.execute("create table t2 using st tags('nchar2')")
tdSql.execute("create table t3 using st tags('nchar3')")
tdSql.execute("create table t4 using st tags('nchar4')")
tdSql.execute("create table t5 using st tags('nchar5')")
for i in range(self.num):
tdSql.execute("insert into t0 values(%d, %d)" % (self.ts + i, i))
tdSql.execute("insert into t1 values(%d, %d)" % (self.ts + i, i))
tdSql.execute("insert into t2 values(%d, %d)" % (self.ts + i, i))
tdSql.execute("insert into t3 values(%d, %d)" % (self.ts + i, i))
tdSql.execute("insert into t4 values(%d, %d)" % (self.ts + i, i))
tdSql.execute("insert into t5 values(%d, %d)" % (self.ts + i, i))
sql = ''' select * from st where loc = 'nchar0' limit 1 union all select * from st where loc = 'nchar1' limit 1 union all select * from st where loc = 'nchar2' limit 1
union all select * from st where loc = 'nchar3' limit 1 union all select * from st where loc = 'nchar4' limit 1'''
tdSql.query(sql)
tdSql.checkRows(5)
sql = ''' select * from st where loc = 'nchar0' limit 1 union all select * from st where loc = 'nchar1' limit 1 union all select * from st where loc = 'nchar2' limit 1
union all select * from st where loc = 'nchar3' limit 1 union all select * from st where loc = 'nchar4' limit 1 union all select * from st where loc = 'nchar5' limit 1'''
tdSql.query(sql)
tdSql.checkRows(6)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())
\ No newline at end of file
tests/pytest/table/max_table_length.py 0 → 100644
浏览文件 @ 9598664f
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, db_test.stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import sys
import taos
from util.log import tdLog
from util.cases import tdCases
from util.sql import tdSql
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
def run(self):
tdSql.prepare()
print("==============step1")
tdLog.info("check nchar")
tdSql.error("create database anal (ts timestamp ,i nchar(4094))")
tdSql.execute(
"create table anal (ts timestamp ,i nchar(4093))")
print("==============step2")
tdLog.info("check binary")
tdSql.error("create database anal (ts timestamp ,i binary(16375))")
tdSql.execute(
"create table anal1 (ts timestamp ,i binary(16374))")
print("==============step3")
tdLog.info("check int & binary")
tdSql.error("create table anal2 (ts timestamp ,i binary(16371),j int)")
tdSql.execute("create table anal2 (ts timestamp ,i binary(16370),j int)")
tdSql.execute("create table anal3 (ts timestamp ,i binary(16366), j int, k int)")
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())
tests/pytest/tools/taosdemoPerformance.py 0 → 100644
浏览文件 @ 9598664f
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import sys
import taos
import time
import datetime
import csv
import random
import pandas as pd
import argparse
import os.path
class taosdemoPerformace:
def __init__(self, commitID, dbName, createTableTime, insertRecordsTime, recordsPerSecond):
self.commitID = commitID
self.dbName = dbName
self.createTableTime = createTableTime
self.insertRecordsTime = insertRecordsTime
self.recordsPerSecond = recordsPerSecond
self.host = "127.0.0.1"
self.user = "root"
self.password = "taosdata"
self.config = "/etc/taosperf"
self.conn = taos.connect(
self.host,
self.user,
self.password,
self.config)
def createTablesAndStoreData(self):
cursor = self.conn.cursor()
cursor.execute("create database if not exists %s" % self.dbName)
cursor.execute("use %s" % self.dbName)
cursor.execute("create table if not exists taosdemo_perf (ts timestamp, create_table_time float, insert_records_time float, records_per_second float, commit_id binary(50))")
print("==================== taosdemo performance ====================")
print("create tables time: %f" % self.createTableTime)
print("insert records time: %f" % self.insertRecordsTime)
print("records per second: %f" % self.recordsPerSecond)
cursor.execute("insert into taosdemo_perf values(now, %f, %f, %f, '%s')" % (self.createTableTime, self.insertRecordsTime, self.recordsPerSecond, self.commitID))
cursor.execute("drop database if exists taosdemo_insert_test")
cursor.close()
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument(
'-c',
'--commit-id',
action='store',
type=str,
help='git commit id (default: null)')
parser.add_argument(
'-d',
'--database-name',
action='store',
default='perf',
type=str,
help='Database name to be created (default: perf)')
parser.add_argument(
'-t',
'--create-table',
action='store',
type=float,
help='create table time')
parser.add_argument(
'-i',
'--insert-records',
action='store',
type=float,
help='insert records time')
parser.add_argument(
'-r',
'---records-per-second',
action='store',
type=float,
help='records per request')
args = parser.parse_args()
perftest = taosdemoPerformace(args.commit_id, args.database_name, args.create_table, args.insert_records, args.records_per_second)
perftest.createTablesAndStoreData()
\ No newline at end of file
tests/pytest/tools/taosdemoTest2.py 0 → 100644
浏览文件 @ 9598664f
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import sys
import os
from util.log import *
from util.cases import *
from util.sql import *
from util.dnodes import *
import threading
import time
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
self.numberOfTables = 10
self.numberOfRecords = 1000000
def insertDataAndAlterTable(self, threadID):
if(threadID == 0):
os.system("yes | taosdemo -t %d -n %d" % (self.numberOfTables, self.numberOfRecords))
if(threadID == 1):
print("use test")
tdSql.execute("use test")
print("alter table test.meters add column f4 int")
tdSql.execute("alter table test.meters add column f4 int")
print("insert into test.t0 values (now, 1, 2, 3, 4)")
tdSql.execute("insert into test.t0 values (now, 1, 2, 3, 4)")
def run(self):
tdSql.prepare()
t1 = threading.Thread(target=self.insertDataAndAlterTable, args=(0, ))
t2 = threading.Thread(target=self.insertDataAndAlterTable, args=(1, ))
t1.start()
time.sleep(2)
t2.start()
t1.join()
t2.join()
tdSql.query("select count(*) from test.meters")
tdSql.checkData(0, 0, self.numberOfRecords * self.numberOfTables + 1)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())
\ No newline at end of file
tests/test/c/hashPerformance.c
浏览文件 @ 9598664f
......@@ -24,15 +24,17 @@
#define GREEN "\033[1;32m"
#define NC "\033[0m"
int32_t capacity = 100000;
int32_t q1Times = 1;
int32_t q2Times = 1;
int32_t capacity = 128;
int32_t q1Times = 10;
int32_t q2Times = 10;
int32_t keyNum = 100000;
int32_t printInterval = 10000;
int32_t printInterval = 1000;
void * hashHandle;
pthread_t thread;
typedef struct HashTestRow {
int32_t size;
void * ptr;
int32_t keySize;
char key[100];
} HashTestRow;
void shellParseArgument(int argc, char *argv[]);
......@@ -40,7 +42,7 @@ void shellParseArgument(int argc, char *argv[]);
void testHashPerformance() {
int64_t initialMs = taosGetTimestampMs();
_hash_fn_t hashFp = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
void * hashHandle = taosHashInit(capacity, hashFp, true);
hashHandle = taosHashInit(128, hashFp, true, HASH_NO_LOCK);
int64_t startMs = taosGetTimestampMs();
float seconds = (startMs - initialMs) / 1000.0;
......@@ -48,17 +50,25 @@ void testHashPerformance() {
for (int32_t t = 1; t <= keyNum; ++t) {
HashTestRow row = {0};
char key[100] = {0};
int32_t keySize = sprintf(key, "0.db.st%d", t);
row.keySize = sprintf(row.key, "0.db.st%d", t);
for (int32_t q = 0; q < q1Times; q++) {
taosHashGet(hashHandle, &key, keySize);
taosHashGet(hashHandle, row.key, row.keySize);
}
taosHashPut(hashHandle, key, keySize, &row, sizeof(HashTestRow));
taosHashPut(hashHandle, row.key, row.keySize, &row, sizeof(HashTestRow));
for (int32_t q = 0; q < q2Times; q++) {
taosHashGet(hashHandle, &key, keySize);
taosHashGet(hashHandle, row.key, row.keySize);
}
// test iterator
{
HashTestRow *row = taosHashIterate(hashHandle, NULL);
while (row) {
taosHashGet(hashHandle, row->key, row->keySize);
row = taosHashIterate(hashHandle, row);
}
}
if (t % printInterval == 0) {
......@@ -80,9 +90,35 @@ void testHashPerformance() {
taosHashCleanup(hashHandle);
}
void *multiThreadFunc(void *param) {
for (int i = 0; i < 100; ++i) {
taosMsleep(1000);
HashTestRow *row = taosHashIterate(hashHandle, NULL);
while (row) {
taosHashGet(hashHandle, row->key, row->keySize);
row = taosHashIterate(hashHandle, row);
}
int64_t hashSize = taosHashGetSize(hashHandle);
pPrint("i:%d hashSize:%ld", i, hashSize);
}
return NULL;
}
void multiThreadTest() {
pthread_attr_t thattr;
pthread_attr_init(&thattr);
pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE);
// Start threads to write
pthread_create(&thread, &thattr, multiThreadFunc, NULL);
}
int main(int argc, char *argv[]) {
shellParseArgument(argc, argv);
multiThreadTest();
testHashPerformance();
pthread_join(thread, NULL);
}
void printHelp() {
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或