Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
T
TDengine
未验证 提交 c41d23da 编写于 作者: sangshuduo's avatar sangshuduo 提交者： GitHub
浏览文件
操作

copy from develop branch. stmt: 0, nanosec: 0 (#6955) 

* copy from develop branch. stmt: 0, nanosec: 0

* set thread name disabled on master branch.
上级 d9cc1eb7
显示空白变更内容
内联 并排
Showing with 1020 addition and 907 deletion
src/kit/taosdemo/taosdemo.c
浏览文件 @ c41d23da
......@@ -55,6 +55,11 @@
#define STMT_IFACE_ENABLED 0
#define NANO_SECOND_ENABLED 0
#define SET_THREADNAME_ENABLED 0
#if SET_THREADNAME_ENABLED == 0
#define setThreadName(name)
#endif
#define REQ_EXTRA_BUF_LEN 1024
#define RESP_BUF_LEN 4096
......@@ -90,7 +95,7 @@ extern char configDir[];
#define MAX_SUPER_TABLE_COUNT 200
#define MAX_QUERY_SQL_COUNT 100
#define MAX_QUERY_SQL_LENGTH 1024
#define MAX_QUERY_SQL_LENGTH BUFFER_SIZE
#define MAX_DATABASE_COUNT 256
#define INPUT_BUF_LEN 256
......@@ -2367,7 +2372,25 @@ static char* getTagValueFromTagSample(SSuperTable* stbInfo, int tagUsePos) {
return dataBuf;
}
static char* generateTagValuesForStb(SSuperTable* stbInfo, int32_t tableSeq) {
static char *generateBinaryNCharTagValues(int64_t tableSeq, uint32_t len)
{
char* buf = (char*)calloc(len, 1);
if (NULL == buf) {
printf("calloc failed! size:%d\n", len);
return NULL;
}
if (tableSeq % 2) {
tstrncpy(buf, "beijing", len);
} else {
tstrncpy(buf, "shanghai", len);
}
//rand_string(buf, stbInfo->tags[i].dataLen);
return buf;
}
static char* generateTagValuesForStb(SSuperTable* stbInfo, int64_t tableSeq) {
char* dataBuf = (char*)calloc(TSDB_MAX_SQL_LEN+1, 1);
if (NULL == dataBuf) {
printf("calloc failed! size:%d\n", TSDB_MAX_SQL_LEN+1);
......@@ -2388,20 +2411,12 @@ static char* generateTagValuesForStb(SSuperTable* stbInfo, int32_t tableSeq) {
return NULL;
}
int tagBufLen = stbInfo->tags[i].dataLen + 1;
char* buf = (char*)calloc(tagBufLen, 1);
int32_t tagBufLen = stbInfo->tags[i].dataLen + 1;
char *buf = generateBinaryNCharTagValues(tableSeq, tagBufLen);
if (NULL == buf) {
printf("calloc failed! size:%d\n", stbInfo->tags[i].dataLen);
tmfree(dataBuf);
return NULL;
}
if (tableSeq % 2) {
tstrncpy(buf, "beijing", tagBufLen);
} else {
tstrncpy(buf, "shanghai", tagBufLen);
}
//rand_string(buf, stbInfo->tags[i].dataLen);
dataLen += snprintf(dataBuf + dataLen, TSDB_MAX_SQL_LEN - dataLen,
"\'%s\',", buf);
tmfree(buf);
......@@ -2410,11 +2425,11 @@ static char* generateTagValuesForStb(SSuperTable* stbInfo, int32_t tableSeq) {
if ((g_args.demo_mode) && (i == 0)) {
dataLen += snprintf(dataBuf + dataLen,
TSDB_MAX_SQL_LEN - dataLen,
"%d,", tableSeq % 10);
"%"PRId64",", tableSeq % 10);
} else {
dataLen += snprintf(dataBuf + dataLen,
TSDB_MAX_SQL_LEN - dataLen,
"%d,", tableSeq);
"%"PRId64",", tableSeq);
}
} else if (0 == strncasecmp(stbInfo->tags[i].dataType,
"bigint", strlen("bigint"))) {
......@@ -2445,7 +2460,7 @@ static char* generateTagValuesForStb(SSuperTable* stbInfo, int32_t tableSeq) {
dataLen += snprintf(dataBuf + dataLen, TSDB_MAX_SQL_LEN - dataLen,
"%"PRId64",", rand_bigint());
} else {
printf("No support data type: %s\n", stbInfo->tags[i].dataType);
errorPrint("No support data type: %s\n", stbInfo->tags[i].dataType);
tmfree(dataBuf);
return NULL;
}
......@@ -2734,7 +2749,7 @@ static int createSuperTable(
} else if (strcasecmp(dataType, "INT") == 0) {
if ((g_args.demo_mode) && (colIndex == 1)) {
len += snprintf(cols + len, COL_BUFFER_LEN - len,
",VOLTAGE INT");
", VOLTAGE INT");
} else {
len += snprintf(cols + len, COL_BUFFER_LEN - len, ",C%d %s", colIndex, "INT");
}
......@@ -2756,9 +2771,9 @@ static int createSuperTable(
} else if (strcasecmp(dataType, "FLOAT") == 0) {
if (g_args.demo_mode) {
if (colIndex == 0) {
len += snprintf(cols + len, COL_BUFFER_LEN - len, ",CURRENT FLOAT");
len += snprintf(cols + len, COL_BUFFER_LEN - len, ", CURRENT FLOAT");
} else if (colIndex == 2) {
len += snprintf(cols + len, COL_BUFFER_LEN - len, ",PHASE FLOAT");
len += snprintf(cols + len, COL_BUFFER_LEN - len, ", PHASE FLOAT");
}
} else {
len += snprintf(cols + len, COL_BUFFER_LEN - len, ",C%d %s", colIndex, "FLOAT");
......@@ -3025,10 +3040,11 @@ static void* createTable(void *sarg)
threadInfo *pThreadInfo = (threadInfo *)sarg;
SSuperTable* superTblInfo = pThreadInfo->superTblInfo;
setThreadName("createTable");
uint64_t lastPrintTime = taosGetTimestampMs();
int buff_len;
buff_len = BUFFER_SIZE;
int buff_len = BUFFER_SIZE;
pThreadInfo->buffer = calloc(buff_len, 1);
if (pThreadInfo->buffer == NULL) {
......@@ -5318,7 +5334,7 @@ static int generateSQLHeadWithoutStb(char *tableName,
static int generateStbSQLHead(
SSuperTable* superTblInfo,
char *tableName, int32_t tableSeq,
char *tableName, int64_t tableSeq,
char *dbName,
char *buffer, int remainderBufLen)
{
......@@ -5650,8 +5666,7 @@ static int32_t prepareStmtBindArrayByType(TAOS_BIND *bind,
*ptr += bind->buffer_length;
} else {
errorPrint( "No support data type: %s\n",
dataType);
errorPrint( "No support data type: %s\n", dataType);
return -1;
}
......@@ -5737,64 +5752,40 @@ static int32_t prepareStmtWithoutStb(
return k;
}
static int32_t prepareStbStmt(
SSuperTable *stbInfo,
TAOS_STMT *stmt,
char *tableName, uint32_t batch,
uint64_t insertRows,
uint64_t recordFrom,
int64_t startTime,
int64_t *pSamplePos)
static int32_t prepareStbStmtBind(
char *bindArray, SSuperTable *stbInfo, bool sourceRand,
int64_t startTime, int32_t recSeq,
bool isColumn)
{
int ret = taos_stmt_set_tbname(stmt, tableName);
if (ret != 0) {
errorPrint("failed to execute taos_stmt_set_tbname(%s). return 0x%x. reason: %s\n",
tableName, ret, taos_errstr(NULL));
return ret;
}
char *bindArray = malloc(sizeof(TAOS_BIND) * (stbInfo->columnCount + 1));
if (bindArray == NULL) {
errorPrint("%s() LN%d, Failed to allocate %d bind params\n",
__func__, __LINE__, (stbInfo->columnCount + 1));
return -1;
}
bool sourceRand;
if (0 == strncasecmp(stbInfo->dataSource, "rand", strlen("rand"))) {
sourceRand = true;
} else {
sourceRand = false; // from sample data file
}
char *bindBuffer = malloc(g_args.len_of_binary);
char *bindBuffer = calloc(1, g_args.len_of_binary);
if (bindBuffer == NULL) {
errorPrint("%s() LN%d, Failed to allocate %d bind buffer\n",
__func__, __LINE__, g_args.len_of_binary);
free(bindArray);
return -1;
}
uint32_t k;
for (k = 0; k < batch;) {
/* columnCount + 1 (ts) */
char data[MAX_DATA_SIZE];
memset(data, 0, MAX_DATA_SIZE);
char *ptr = data;
TAOS_BIND *bind = (TAOS_BIND *)(bindArray + 0);
TAOS_BIND *bind;
if (isColumn) {
for (int i = 0; i < stbInfo->columnCount + 1; i ++) {
bind = (TAOS_BIND *)((char *)bindArray + (sizeof(TAOS_BIND) * i));
if (i == 0) {
int64_t *bind_ts;
bind_ts = (int64_t *)ptr;
bind->buffer_type = TSDB_DATA_TYPE_TIMESTAMP;
if (stbInfo->disorderRatio) {
*bind_ts = startTime + getTSRandTail(
stbInfo->timeStampStep, k,
stbInfo->timeStampStep, recSeq,
stbInfo->disorderRatio,
stbInfo->disorderRange);
} else {
*bind_ts = startTime + stbInfo->timeStampStep * k;
*bind_ts = startTime + stbInfo->timeStampStep * recSeq;
}
bind->buffer_length = sizeof(int64_t);
bind->buffer = bind_ts;
......@@ -5802,19 +5793,16 @@ static int32_t prepareStbStmt(
bind->is_null = NULL;
ptr += bind->buffer_length;
} else {
int cursor = 0;
for (int i = 0; i < stbInfo->columnCount; i ++) {
bind = (TAOS_BIND *)((char *)bindArray + (sizeof(TAOS_BIND) * (i + 1)));
if (sourceRand) {
if ( -1 == prepareStmtBindArrayByType(
bind,
stbInfo->columns[i].dataType,
stbInfo->columns[i].dataLen,
stbInfo->columns[i-1].dataType,
stbInfo->columns[i-1].dataLen,
&ptr,
NULL)) {
free(bindArray);
free(bindBuffer);
return -1;
}
......@@ -5835,16 +5823,120 @@ static int32_t prepareStbStmt(
if ( -1 == prepareStmtBindArrayByType(
bind,
stbInfo->columns[i].dataType,
stbInfo->columns[i].dataLen,
stbInfo->columns[i-1].dataType,
stbInfo->columns[i-1].dataLen,
&ptr,
bindBuffer)) {
free(bindArray);
free(bindBuffer);
return -1;
}
}
}
}
} else {
TAOS_BIND *tag;
for (int t = 0; t < stbInfo->tagCount; t ++) {
tag = (TAOS_BIND *)((char *)bindArray + (sizeof(TAOS_BIND) * t));
if ( -1 == prepareStmtBindArrayByType(
tag,
stbInfo->tags[t].dataType,
stbInfo->tags[t].dataLen,
&ptr,
NULL)) {
free(bindBuffer);
return -1;
}
}
}
return 0;
}
static int32_t prepareStbStmt(
SSuperTable *stbInfo,
TAOS_STMT *stmt,
char *tableName,
int64_t tableSeq,
uint32_t batch,
uint64_t insertRows,
uint64_t recordFrom,
int64_t startTime,
int64_t *pSamplePos)
{
int ret;
bool sourceRand;
if (0 == strncasecmp(stbInfo->dataSource, "rand", strlen("rand"))) {
sourceRand = true;
} else {
sourceRand = false; // from sample data file
}
if (AUTO_CREATE_SUBTBL == stbInfo->autoCreateTable) {
char* tagsValBuf = NULL;
bool tagRand;
if (0 == stbInfo->tagSource) {
tagRand = true;
tagsValBuf = generateTagValuesForStb(stbInfo, tableSeq);
} else {
tagRand = false;
tagsValBuf = getTagValueFromTagSample(
stbInfo,
tableSeq % stbInfo->tagSampleCount);
}
if (NULL == tagsValBuf) {
errorPrint("%s() LN%d, tag buf failed to allocate memory\n",
__func__, __LINE__);
return -1;
}
char *tagsArray = calloc(1, sizeof(TAOS_BIND) * stbInfo->tagCount);
if (NULL == tagsArray) {
tmfree(tagsValBuf);
errorPrint("%s() LN%d, tag buf failed to allocate memory\n",
__func__, __LINE__);
return -1;
}
if (-1 == prepareStbStmtBind(
tagsArray, stbInfo, tagRand, -1, -1, false /* is tag */)) {
free(tagsArray);
return -1;
}
ret = taos_stmt_set_tbname_tags(stmt, tableName, (TAOS_BIND *)tagsArray);
tmfree(tagsValBuf);
tmfree((char *)tagsArray);
} else {
ret = taos_stmt_set_tbname(stmt, tableName);
}
if (ret != 0) {
errorPrint("failed to execute taos_stmt_set_tbname(%s). return 0x%x. reason: %s\n",
tableName, ret, taos_errstr(NULL));
return ret;
}
char *bindArray = calloc(1, sizeof(TAOS_BIND) * (stbInfo->columnCount + 1));
if (bindArray == NULL) {
errorPrint("%s() LN%d, Failed to allocate %d bind params\n",
__func__, __LINE__, (stbInfo->columnCount + 1));
return -1;
}
uint32_t k;
for (k = 0; k < batch;) {
/* columnCount + 1 (ts) */
if (-1 == prepareStbStmtBind(bindArray, stbInfo, sourceRand,
startTime, k, true /* is column */)) {
free(bindArray);
return -1;
}
taos_stmt_bind_param(stmt, (TAOS_BIND *)bindArray);
// if msg > 3MB, break
taos_stmt_add_batch(stmt);
......@@ -5861,7 +5953,6 @@ static int32_t prepareStbStmt(
}
}
free(bindBuffer);
free(bindArray);
return k;
}
......@@ -5869,7 +5960,9 @@ static int32_t prepareStbStmt(
static int32_t prepareStbStmtInterlace(
SSuperTable *stbInfo,
TAOS_STMT *stmt,
char *tableName, uint32_t batch,
char *tableName,
int64_t tableSeq,
uint32_t batch,
uint64_t insertRows,
uint64_t recordFrom,
int64_t startTime,
......@@ -5879,6 +5972,7 @@ static int32_t prepareStbStmtInterlace(
stbInfo,
stmt,
tableName,
tableSeq,
batch,
insertRows, 0, startTime,
pSamplePos);
......@@ -5887,7 +5981,9 @@ static int32_t prepareStbStmtInterlace(
static int32_t prepareStbStmtProgressive(
SSuperTable *stbInfo,
TAOS_STMT *stmt,
char *tableName, uint32_t batch,
char *tableName,
int64_t tableSeq,
uint32_t batch,
uint64_t insertRows,
uint64_t recordFrom,
int64_t startTime,
......@@ -5897,6 +5993,7 @@ static int32_t prepareStbStmtProgressive(
stbInfo,
stmt,
tableName,
tableSeq,
g_args.num_of_RPR,
insertRows, recordFrom, startTime,
pSamplePos);
......@@ -6097,6 +6194,7 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) {
superTblInfo,
pThreadInfo->stmt,
tableName,
tableSeq,
batchPerTbl,
insertRows, i,
startTime,
......@@ -6326,6 +6424,7 @@ static void* syncWriteProgressive(threadInfo *pThreadInfo) {
superTblInfo,
pThreadInfo->stmt,
tableName,
tableSeq,
g_args.num_of_RPR,
insertRows, i, start_time,
&(pThreadInfo->samplePos));
......@@ -6428,6 +6527,8 @@ static void* syncWrite(void *sarg) {
threadInfo *pThreadInfo = (threadInfo *)sarg;
SSuperTable* superTblInfo = pThreadInfo->superTblInfo;
setThreadName("syncWrite");
uint32_t interlaceRows;
if (superTblInfo) {
......@@ -6513,6 +6614,8 @@ static void *asyncWrite(void *sarg) {
threadInfo *pThreadInfo = (threadInfo *)sarg;
SSuperTable* superTblInfo = pThreadInfo->superTblInfo;
setThreadName("asyncWrite");
pThreadInfo->st = 0;
pThreadInfo->et = 0;
pThreadInfo->lastTs = pThreadInfo->start_time;
......@@ -6754,7 +6857,7 @@ static void startMultiThreadInsertData(int threads, char* db_name,
exit(-1);
}
char buffer[3000];
char buffer[BUFFER_SIZE];
char *pstr = buffer;
if ((superTblInfo)
......@@ -6913,6 +7016,7 @@ static void *readTable(void *sarg) {
#if 1
threadInfo *pThreadInfo = (threadInfo *)sarg;
TAOS *taos = pThreadInfo->taos;
setThreadName("readTable");
char command[BUFFER_SIZE] = "\0";
uint64_t sTime = pThreadInfo->start_time;
char *tb_prefix = pThreadInfo->tb_prefix;
......@@ -6985,6 +7089,7 @@ static void *readMetric(void *sarg) {
#if 1
threadInfo *pThreadInfo = (threadInfo *)sarg;
TAOS *taos = pThreadInfo->taos;
setThreadName("readMetric");
char command[BUFFER_SIZE] = "\0";
FILE *fp = fopen(pThreadInfo->filePath, "a");
if (NULL == fp) {
......@@ -7161,6 +7266,8 @@ static int insertTestProcess() {
static void *specifiedTableQuery(void *sarg) {
threadInfo *pThreadInfo = (threadInfo *)sarg;
setThreadName("specTableQuery");
if (pThreadInfo->taos == NULL) {
TAOS * taos = NULL;
taos = taos_connect(g_queryInfo.host,
......@@ -7260,6 +7367,8 @@ static void *superTableQuery(void *sarg) {
char sqlstr[MAX_QUERY_SQL_LENGTH];
threadInfo *pThreadInfo = (threadInfo *)sarg;
setThreadName("superTableQuery");
if (pThreadInfo->taos == NULL) {
TAOS * taos = NULL;
taos = taos_connect(g_queryInfo.host,
......@@ -7562,6 +7671,8 @@ static void *superSubscribe(void *sarg) {
TAOS_SUB* tsub[MAX_QUERY_SQL_COUNT] = {0};
uint64_t tsubSeq;
setThreadName("superSub");
if (pThreadInfo->ntables > MAX_QUERY_SQL_COUNT) {
errorPrint("The table number(%"PRId64") of the thread is more than max query sql count: %d\n",
pThreadInfo->ntables, MAX_QUERY_SQL_COUNT);
......@@ -7708,6 +7819,8 @@ static void *specifiedSubscribe(void *sarg) {
threadInfo *pThreadInfo = (threadInfo *)sarg;
// TAOS_SUB* tsub = NULL;
setThreadName("specSub");
if (pThreadInfo->taos == NULL) {
pThreadInfo->taos = taos_connect(g_queryInfo.host,
g_queryInfo.user,
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或