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

Merge pull request #1617 from taosdata/feature/mpeer 

Feature/mpeer
This commit is contained in:
slguan 2020-04-15 23:20:28 +08:00 committed by GitHub
parent 6b2c5ba368 6b94643111
commit d8472ba302
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 324 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
tests/test/c/CMakeLists.txt
View File

@ -6,6 +6,9 @@ INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/util/inc)
INCLUDE_DIRECTORIES(${TD_OS_DIR}/inc)
IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
add_executable(benchmarkPerTable benchmarkPerTable.c)
target_link_libraries(benchmarkPerTable taos_static pthread)
add_executable(insertPerTable insertPerTable.c)
target_link_libraries(insertPerTable taos_static pthread)
add_executable(insertPerRow insertPerRow.c)
target_link_libraries(insertPerRow taos_static pthread)
ENDIF()

316
tests/test/c/insertPerRow.c Normal file
View File

@ -0,0 +1,316 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "os.h"
#include "taos.h"
#include "tlog.h"
#include "ttimer.h"
#include "tutil.h"
#define MAX_RANDOM_POINTS 20000
#define GREEN "\033[1;32m"
#define NC "\033[0m"
typedef struct {
int64_t rowsPerTable;
int64_t pointsPerTable;
int64_t tableBeginIndex;
int64_t tableEndIndex;
int threadIndex;
char dbName[32];
char stableName[64];
pthread_t thread;
} SInfo;
void *syncTest(void *param);
void generateRandomPoints();
void shellParseArgument(int argc, char *argv[]);
void createDbAndTable();
void insertData();
int32_t randomData[MAX_RANDOM_POINTS];
int64_t rowsPerTable = 10000;
int64_t pointsPerTable = 1;
int64_t numOfThreads = 1;
int64_t numOfTablesPerThread = 200;
char dbName[32] = "db";
char stableName[64] = "st";
int32_t cache = 16384;
int32_t tables = 1000;
int main(int argc, char *argv[]) {
shellParseArgument(argc, argv);
generateRandomPoints();
taos_init();
createDbAndTable();
insertData();
}
void createDbAndTable() {
dPrint("start to create table");
TAOS * con;
struct timeval systemTime;
int64_t st, et;
char qstr[64000];
con = taos_connect(tsMasterIp, tsDefaultUser, tsDefaultPass, NULL, 0);
if (con == NULL) {
dError("failed to connect to DB, reason:%s", taos_errstr(con));
exit(1);
}
sprintf(qstr, "create database if not exists %s cache %d tables %d", dbName, cache, tables);
if (taos_query(con, qstr)) {
dError("failed to create database:%s, code:%d reason:%s", dbName, taos_errno(con), taos_errstr(con));
exit(0);
}
sprintf(qstr, "use %s", dbName);
if (taos_query(con, qstr)) {
dError("failed to use db, code:%d reason:%s", taos_errno(con), taos_errstr(con));
exit(0);
}
gettimeofday(&systemTime, NULL);
st = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
int64_t totalTables = numOfTablesPerThread * numOfThreads;
if (strcmp(stableName, "no") != 0) {
int len = sprintf(qstr, "create table if not exists %s(ts timestamp", stableName);
for (int64_t f = 0; f < pointsPerTable; ++f) {
len += sprintf(qstr + len, ", f%ld double", f);
}
sprintf(qstr + len, ") tags(t int)");
if (taos_query(con, qstr)) {
dError("failed to create stable, code:%d reason:%s", taos_errno(con), taos_errstr(con));
exit(0);
}
for (int64_t t = 0; t < totalTables; ++t) {
sprintf(qstr, "create table if not exists %s%ld using %s tags(%ld)", stableName, t, stableName, t);
if (taos_query(con, qstr)) {
dError("failed to create table %s%d, reason:%s", stableName, t, taos_errstr(con));
exit(0);
}
}
} else {
for (int64_t t = 0; t < totalTables; ++t) {
int len = sprintf(qstr, "create table if not exists %s%ld(ts timestamp", stableName, t);
for (int64_t f = 0; f < pointsPerTable; ++f) {
len += sprintf(qstr + len, ", f%ld double", f);
}
sprintf(qstr + len, ")");
if (taos_query(con, qstr)) {
dError("failed to create table %s%ld, reason:%s", stableName, t, taos_errstr(con));
exit(0);
}
}
}
gettimeofday(&systemTime, NULL);
et = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
dPrint("%.1f seconds to create %ld tables", (et - st) / 1000.0 / 1000.0, totalTables);
}
void insertData() {
struct timeval systemTime;
int64_t st, et;
gettimeofday(&systemTime, NULL);
st = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
dPrint("%d threads are spawned to insert data", numOfThreads);
pthread_attr_t thattr;
pthread_attr_init(&thattr);
pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE);
SInfo *pInfo = (SInfo *)malloc(sizeof(SInfo) * numOfThreads);
// Start threads to write
for (int i = 0; i < numOfThreads; ++i) {
pInfo[i].rowsPerTable = rowsPerTable;
pInfo[i].pointsPerTable = pointsPerTable;
pInfo[i].tableBeginIndex = i * numOfTablesPerThread;
pInfo[i].tableEndIndex = (i + 1) * numOfTablesPerThread;
pInfo[i].threadIndex = i;
strcpy(pInfo[i].dbName, dbName);
strcpy(pInfo[i].stableName, stableName);
pthread_create(&(pInfo[i].thread), &thattr, syncTest, (void *)(pInfo + i));
}
taosMsleep(300);
for (int i = 0; i < numOfThreads; i++) {
pthread_join(pInfo[i].thread, NULL);
}
gettimeofday(&systemTime, NULL);
et = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
double seconds = (et - st) / 1000.0 / 1000.0;
int64_t totalTables = numOfTablesPerThread * numOfThreads;
int64_t totalRows = totalTables * rowsPerTable;
int64_t totalPoints = totalTables * rowsPerTable * pointsPerTable;
double speedOfRows = totalRows / seconds;
double speedOfPoints = totalPoints / seconds;
dPrint(
"%sall threads:%ld finished, use %.1lf seconds, tables:%.ld rows:%ld points:%ld, speed RowsPerSecond:%.1lf "
"PointsPerSecond:%.1lf%s",
GREEN, numOfThreads, seconds, totalTables, totalRows, totalPoints, speedOfRows, speedOfPoints, NC);
dPrint("threads exit");
pthread_attr_destroy(&thattr);
free(pInfo);
}
void *syncTest(void *param) {
TAOS * con;
SInfo * pInfo = (SInfo *)param;
struct timeval systemTime;
int64_t st, et;
char qstr[65000];
int maxBytes = 60000;
dPrint("thread:%d, start to run", pInfo->threadIndex);
con = taos_connect(tsMasterIp, tsDefaultUser, tsDefaultPass, NULL, 0);
if (con == NULL) {
dError("index:%d, failed to connect to DB, reason:%s", pInfo->threadIndex, taos_errstr(con));
exit(1);
}
sprintf(qstr, "use %s", pInfo->dbName);
taos_query(con, qstr);
gettimeofday(&systemTime, NULL);
st = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
int64_t start = 1430000000000;
int64_t interval = 1000; // 1000 ms
char *sql = qstr;
char inserStr[] = "insert into";
int len = sprintf(sql, "%s", inserStr);
for (int64_t row = 0; row < pInfo->rowsPerTable; row++) {
for (int64_t table = pInfo->tableBeginIndex; table < pInfo->tableEndIndex; ++table) {
len += sprintf(sql + len, " %s%ld values", pInfo->stableName, table);
len += sprintf(sql + len, "(%ld", start + row * interval);
for (int64_t point = 0; point < pInfo->pointsPerTable; ++point) {
len += sprintf(sql + len, ",%d", randomData[(123 * table + 456 * row + 789 * point) % MAX_RANDOM_POINTS]);
// len += sprintf(sql + len, ",%ld", row);
}
len += sprintf(sql + len, ")");
if (len > maxBytes) {
if (taos_query(con, qstr)) {
dError("thread:%d, failed to insert table:%s%ld row:%ld, reason:%s", pInfo->threadIndex, pInfo->stableName,
table, row, taos_errstr(con));
}
// "insert into"
len = sprintf(sql, "%s", inserStr);
}
}
}
if (len != strlen(inserStr)) {
taos_query(con, qstr);
}
gettimeofday(&systemTime, NULL);
et = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
int64_t totalTables = pInfo->tableEndIndex - pInfo->tableBeginIndex;
int64_t totalRows = totalTables * pInfo->rowsPerTable;
int64_t totalPoints = totalRows * pInfo->pointsPerTable;
dPrint("thread:%d, insert finished, use %.2f seconds, tables:%ld rows:%ld points:%ld", pInfo->threadIndex,
(et - st) / 1000.0 / 1000.0, totalTables, totalRows, totalPoints);
return NULL;
}
void generateRandomPoints() {
for (int r = 0; r < MAX_RANDOM_POINTS; ++r) {
randomData[r] = rand() % 1000;
}
}
void printHelp() {
char indent[10] = " ";
printf("Used to test the performance of TDengine\n After writing one row of data to all tables, write the next row\n");
printf("%s%s\n", indent, "-d");
printf("%s%s%s%s\n", indent, indent, "The name of the database to be created, default is ", dbName);
printf("%s%s\n", indent, "-s");
printf("%s%s%s%s%s\n", indent, indent, "The name of the super table to be created, default is ", stableName, ", if 'no' then create normal table");
printf("%s%s\n", indent, "-c");
printf("%s%s%s%s\n", indent, indent, "Configuration directory, default is ", configDir);
printf("%s%s\n", indent, "-r");
printf("%s%s%s%ld\n", indent, indent, "Number of records to write to each table, default is ", rowsPerTable);
printf("%s%s\n", indent, "-p");
printf("%s%s%s%" PRId64 "\n", indent, indent, "Number of columns per table, default is ", pointsPerTable);
printf("%s%s\n", indent, "-t");
printf("%s%s%s%" PRId64 "\n", indent, indent, "Number of threads to be used, default is ", numOfThreads);
printf("%s%s\n", indent, "-n");
printf("%s%s%s%" PRId64 "\n", indent, indent, "Number of tables per thread, default is ", numOfTablesPerThread);
printf("%s%s\n", indent, "-tables");
printf("%s%s%s%d\n", indent, indent, "Database parameters tables, default is ", tables);
printf("%s%s\n", indent, "-cache");
printf("%s%s%s%d\n", indent, indent, "Database parameters cache, default is ", cache);
exit(EXIT_SUCCESS);
}
void shellParseArgument(int argc, char *argv[]) {
for (int i = 1; i < argc; i++) {
if (strcmp(argv[i], "-h") == 0 || strcmp(argv[i], "--help") == 0) {
printHelp();
exit(0);
} else if (strcmp(argv[i], "-d") == 0) {
strcpy(dbName, argv[++i]);
} else if (strcmp(argv[i], "-c") == 0) {
strcpy(configDir, argv[++i]);
} else if (strcmp(argv[i], "-s") == 0) {
strcpy(stableName, argv[++i]);
} else if (strcmp(argv[i], "-r") == 0) {
rowsPerTable = atoi(argv[++i]);
} else if (strcmp(argv[i], "-p") == 0) {
pointsPerTable = atoi(argv[++i]);
} else if (strcmp(argv[i], "-t") == 0) {
numOfThreads = atoi(argv[++i]);
} else if (strcmp(argv[i], "-n") == 0) {
numOfTablesPerThread = atoi(argv[++i]);
} else if (strcmp(argv[i], "-tables") == 0) {
tables = atoi(argv[++i]);
} else if (strcmp(argv[i], "-cache") == 0) {
cache = atoi(argv[++i]);
} else {
}
}
dPrint("%srowsPerTable:%" PRId64 "%s", GREEN, rowsPerTable, NC);
dPrint("%spointsPerTable:%" PRId64 "%s", GREEN, pointsPerTable, NC);
dPrint("%snumOfThreads:%" PRId64 "%s", GREEN, numOfThreads, NC);
dPrint("%snumOfTablesPerThread:%" PRId64 "%s", GREEN, numOfTablesPerThread, NC);
dPrint("%scache:%" PRId64 "%s", GREEN, cache, NC);
dPrint("%stables:%" PRId64 "%s", GREEN, tables, NC);
dPrint("%sdbName:%s%s", GREEN, dbName, NC);
dPrint("%stableName:%s%s", GREEN, stableName, NC);
dPrint("%sstart to run%s", GREEN, NC);
}

4
tests/test/c/benchmarkPerTable.c → tests/test/c/insertPerTable.c
View File

@ -258,7 +258,7 @@ void generateRandomPoints() {
void printHelp() {
char indent[10] = " ";
printf("Used to test the performance of TDengine, the insert method is table-by-table\n");
printf("Used to test the performance of TDengine\n After writing all the data in one table, start the next table\n");
printf("%s%s\n", indent, "-d");
printf("%s%s%s%s\n", indent, indent, "The name of the database to be created, default is ", dbName);
@ -289,6 +289,8 @@ void shellParseArgument(int argc, char *argv[]) {
exit(0);
} else if (strcmp(argv[i], "-d") == 0) {
strcpy(dbName, argv[++i]);
} else if (strcmp(argv[i], "-c") == 0) {
strcpy(configDir, argv[++i]);
} else if (strcmp(argv[i], "-s") == 0) {
strcpy(stableName, argv[++i]);
} else if (strcmp(argv[i], "-r") == 0) {