Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

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

[TD-6448]<fix>: taosdemo stmt rand race. (#7719) 

Co-authored-by: NShuduo Sang <sdsang@taosdata.com>
上级 2483985d
隐藏空白更改
内联 并排
Showing with 71 addition and 102 deletion
src/kit/taosdemo/taosdemo.c
浏览文件 @ 56776b62
......@@ -291,7 +291,6 @@ typedef struct SSuperTable_S {
uint64_t lenOfTagOfOneRow;
char* sampleDataBuf;
char* sampleBindArray;
//int sampleRowCount;
//int sampleUsePos;
......@@ -438,7 +437,8 @@ typedef struct SQueryMetaInfo_S {
typedef struct SThreadInfo_S {
TAOS * taos;
TAOS_STMT *stmt;
int64_t *bind_ts;
char* sampleBindArray;
int64_t *bind_ts;
int threadID;
char db_name[TSDB_DB_NAME_LEN];
uint32_t time_precision;
......@@ -5738,20 +5738,6 @@ static void postFreeResource() {
free(g_Dbs.db[i].superTbls[j].sampleDataBuf);
g_Dbs.db[i].superTbls[j].sampleDataBuf = NULL;
}
if (g_Dbs.db[i].superTbls[j].sampleBindArray) {
for (int k = 0; k < MAX_SAMPLES_ONCE_FROM_FILE; k++) {
uintptr_t *tmp = (uintptr_t *)(*(uintptr_t *)(
g_Dbs.db[i].superTbls[j].sampleBindArray
+ sizeof(uintptr_t *) * k));
for (int c = 1; c < g_Dbs.db[i].superTbls[j].columnCount + 1; c++) {
TAOS_BIND *bind = (TAOS_BIND *)((char *)tmp + (sizeof(TAOS_BIND) * c));
if (bind)
tmfree(bind->buffer);
}
tmfree((char *)tmp);
}
}
tmfree((char *)g_Dbs.db[i].superTbls[j].sampleBindArray);
if (0 != g_Dbs.db[i].superTbls[j].tagDataBuf) {
free(g_Dbs.db[i].superTbls[j].tagDataBuf);
......@@ -6085,9 +6071,6 @@ static int32_t execInsert(threadInfo *pThreadInfo, uint32_t k)
int32_t affectedRows;
SSuperTable* stbInfo = pThreadInfo->stbInfo;
verbosePrint("[%d] %s() LN%d %s\n", pThreadInfo->threadID,
__func__, __LINE__, pThreadInfo->buffer);
uint16_t iface;
if (stbInfo)
iface = stbInfo->iface;
......@@ -6105,12 +6088,18 @@ static int32_t execInsert(threadInfo *pThreadInfo, uint32_t k)
switch(iface) {
case TAOSC_IFACE:
verbosePrint("[%d] %s() LN%d %s\n", pThreadInfo->threadID,
__func__, __LINE__, pThreadInfo->buffer);
affectedRows = queryDbExec(
pThreadInfo->taos,
pThreadInfo->buffer, INSERT_TYPE, false);
break;
case REST_IFACE:
verbosePrint("[%d] %s() LN%d %s\n", pThreadInfo->threadID,
__func__, __LINE__, pThreadInfo->buffer);
if (0 != postProceSql(g_Dbs.host, &g_Dbs.serv_addr, g_Dbs.port,
pThreadInfo->buffer, pThreadInfo)) {
affectedRows = -1;
......@@ -7088,12 +7077,12 @@ static int32_t prepareStbStmtBindRand(
return 0;
}
static int32_t prepareStbStmtBindWithSample(
static int32_t prepareStbStmtBindStartTime(
char *tableName,
int64_t *ts,
char *bindArray, SSuperTable *stbInfo,
int64_t startTime, int32_t recSeq,
int32_t timePrec,
int64_t samplePos)
int32_t timePrec)
{
TAOS_BIND *bind;
......@@ -7110,6 +7099,10 @@ static int32_t prepareStbStmtBindWithSample(
} else {
*bind_ts = startTime + stbInfo->timeStampStep * recSeq;
}
verbosePrint("%s() LN%d, tableName: %s, bind_ts=%"PRId64"\n",
__func__, __LINE__, tableName, *bind_ts);
bind->buffer_length = sizeof(int64_t);
bind->buffer = bind_ts;
bind->length = &bind->buffer_length;
......@@ -7118,7 +7111,7 @@ static int32_t prepareStbStmtBindWithSample(
return 0;
}
static int32_t prepareStbStmtRand(
UNUSED_FUNC static int32_t prepareStbStmtRand(
threadInfo *pThreadInfo,
char *tableName,
int64_t tableSeq,
......@@ -7299,14 +7292,14 @@ static int32_t prepareStbStmtWithSample(
uint32_t k;
for (k = 0; k < batch;) {
char *bindArray = (char *)(*((uintptr_t *)
(stbInfo->sampleBindArray + (sizeof(char *)) * (*pSamplePos))));
(pThreadInfo->sampleBindArray + (sizeof(char *)) * (*pSamplePos))));
/* columnCount + 1 (ts) */
if (-1 == prepareStbStmtBindWithSample(
if (-1 == prepareStbStmtBindStartTime(
tableName,
pThreadInfo->bind_ts,
bindArray, stbInfo,
startTime, k,
pThreadInfo->time_precision,
*pSamplePos
pThreadInfo->time_precision
/* is column */)) {
return -1;
}
......@@ -7427,8 +7420,6 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) {
int64_t nTimeStampStep;
uint64_t insert_interval;
bool sourceRand;
SSuperTable* stbInfo = pThreadInfo->stbInfo;
if (stbInfo) {
......@@ -7443,18 +7434,12 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) {
maxSqlLen = stbInfo->maxSqlLen;
nTimeStampStep = stbInfo->timeStampStep;
insert_interval = stbInfo->insertInterval;
if (0 == strncasecmp(stbInfo->dataSource, "rand", 4)) {
sourceRand = true;
} else {
sourceRand = false; // from sample data file
}
} else {
insertRows = g_args.num_of_DPT;
interlaceRows = g_args.interlace_rows;
maxSqlLen = g_args.max_sql_len;
nTimeStampStep = g_args.timestamp_step;
insert_interval = g_args.insert_interval;
sourceRand = true;
}
debugPrint("[%d] %s() LN%d: start_table_from=%"PRIu64" ntables=%"PRId64" insertRows=%"PRIu64"\n",
......@@ -7539,25 +7524,14 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) {
int32_t generated;
if (stbInfo) {
if (stbInfo->iface == STMT_IFACE) {
if (sourceRand) {
generated = prepareStbStmtRand(
pThreadInfo,
tableName,
tableSeq,
batchPerTbl,
insertRows, 0,
startTime
);
} else {
generated = prepareStbStmtWithSample(
pThreadInfo,
tableName,
tableSeq,
batchPerTbl,
insertRows, 0,
startTime,
&(pThreadInfo->samplePos));
}
generated = prepareStbStmtWithSample(
pThreadInfo,
tableName,
tableSeq,
batchPerTbl,
insertRows, 0,
startTime,
&(pThreadInfo->samplePos));
} else {
generated = generateStbInterlaceData(
pThreadInfo,
......@@ -7747,17 +7721,6 @@ static void* syncWriteProgressive(threadInfo *pThreadInfo) {
pThreadInfo->totalInsertRows = 0;
pThreadInfo->totalAffectedRows = 0;
bool sourceRand;
if (stbInfo) {
if (0 == strncasecmp(stbInfo->dataSource, "rand", 4)) {
sourceRand = true;
} else {
sourceRand = false; // from sample data file
}
} else {
sourceRand = true;
}
pThreadInfo->samplePos = 0;
int percentComplete = 0;
......@@ -7796,32 +7759,13 @@ static void* syncWriteProgressive(threadInfo *pThreadInfo) {
int32_t generated;
if (stbInfo) {
if (stbInfo->iface == STMT_IFACE) {
if (sourceRand) {
/* generated = prepareStbStmtRand(
pThreadInfo,
tableName,
tableSeq,
g_args.num_of_RPR,
insertRows,
i, start_time
);
*/
generated = prepareStbStmtWithSample(
pThreadInfo,
tableName,
tableSeq,
g_args.num_of_RPR,
insertRows, i, start_time,
&(pThreadInfo->samplePos));
} else {
generated = prepareStbStmtWithSample(
pThreadInfo,
tableName,
tableSeq,
g_args.num_of_RPR,
insertRows, i, start_time,
&(pThreadInfo->samplePos));
}
generated = prepareStbStmtWithSample(
pThreadInfo,
tableName,
tableSeq,
g_args.num_of_RPR,
insertRows, i, start_time,
&(pThreadInfo->samplePos));
} else {
generated = generateStbProgressiveData(
stbInfo,
......@@ -7849,6 +7793,11 @@ static void* syncWriteProgressive(threadInfo *pThreadInfo) {
&remainderBufLen);
}
}
verbosePrint("[%d] %s() LN%d generated=%d\n",
pThreadInfo->threadID,
__func__, __LINE__, generated);
if (generated > 0)
i += generated;
else
......@@ -8059,17 +8008,22 @@ static int convertHostToServAddr(char *host, uint16_t port, struct sockaddr_in *
return 0;
}
static int parseSampleFileToStmt(SSuperTable *stbInfo, uint32_t timePrec)
static int parseSampleFileToStmt(
threadInfo *pThreadInfo,
SSuperTable *stbInfo, uint32_t timePrec)
{
stbInfo->sampleBindArray = calloc(1, sizeof(char *) * MAX_SAMPLES_ONCE_FROM_FILE);
if (stbInfo->sampleBindArray == NULL) {
pThreadInfo->sampleBindArray =
calloc(1, sizeof(char *) * MAX_SAMPLES_ONCE_FROM_FILE);
if (pThreadInfo->sampleBindArray == NULL) {
errorPrint2("%s() LN%d, Failed to allocate %"PRIu64" bind array buffer\n",
__func__, __LINE__, (uint64_t)sizeof(char *) * MAX_SAMPLES_ONCE_FROM_FILE);
__func__, __LINE__,
(uint64_t)sizeof(char *) * MAX_SAMPLES_ONCE_FROM_FILE);
return -1;
}
for (int i=0; i < MAX_SAMPLES_ONCE_FROM_FILE; i++) {
char *bindArray = calloc(1, sizeof(TAOS_BIND) * (stbInfo->columnCount + 1));
char *bindArray =
calloc(1, sizeof(TAOS_BIND) * (stbInfo->columnCount + 1));
if (bindArray == NULL) {
errorPrint2("%s() LN%d, Failed to allocate %d bind params\n",
__func__, __LINE__, (stbInfo->columnCount + 1));
......@@ -8122,7 +8076,8 @@ static int parseSampleFileToStmt(SSuperTable *stbInfo, uint32_t timePrec)
free(bindBuffer);
}
}
*((uintptr_t *)(stbInfo->sampleBindArray + (sizeof(char *)) * i)) = (uintptr_t)bindArray;
*((uintptr_t *)(pThreadInfo->sampleBindArray + (sizeof(char *)) * i)) =
(uintptr_t)bindArray;
}
return 0;
......@@ -8312,10 +8267,6 @@ static void startMultiThreadInsertData(int threads, char* db_name,
pstr += sprintf(pstr, ")");
debugPrint("%s() LN%d, stmtBuffer: %s", __func__, __LINE__, stmtBuffer);
if (stbInfo) {
parseSampleFileToStmt(stbInfo, timePrec);
}
}
for (int i = 0; i < threads; i++) {
......@@ -8348,7 +8299,6 @@ static void startMultiThreadInsertData(int threads, char* db_name,
|| ((stbInfo)
&& (stbInfo->iface == STMT_IFACE))) {
pThreadInfo->stmt = taos_stmt_init(pThreadInfo->taos);
if (NULL == pThreadInfo->stmt) {
free(pids);
......@@ -8370,6 +8320,10 @@ static void startMultiThreadInsertData(int threads, char* db_name,
exit(EXIT_FAILURE);
}
pThreadInfo->bind_ts = malloc(sizeof(int64_t));
if (stbInfo) {
parseSampleFileToStmt(pThreadInfo, stbInfo, timePrec);
}
}
} else {
pThreadInfo->taos = NULL;
......@@ -8420,6 +8374,21 @@ static void startMultiThreadInsertData(int threads, char* db_name,
tsem_destroy(&(pThreadInfo->lock_sem));
taos_close(pThreadInfo->taos);
if (pThreadInfo->sampleBindArray) {
for (int k = 0; k < MAX_SAMPLES_ONCE_FROM_FILE; k++) {
uintptr_t *tmp = (uintptr_t *)(*(uintptr_t *)(
pThreadInfo->sampleBindArray
+ sizeof(uintptr_t *) * k));
for (int c = 1; c < pThreadInfo->stbInfo->columnCount + 1; c++) {
TAOS_BIND *bind = (TAOS_BIND *)((char *)tmp + (sizeof(TAOS_BIND) * c));
if (bind)
tmfree(bind->buffer);
}
tmfree((char *)tmp);
}
tmfree(pThreadInfo->sampleBindArray);
}
debugPrint("%s() LN%d, [%d] totalInsert=%"PRIu64" totalAffected=%"PRIu64"\n",
__func__, __LINE__,
pThreadInfo->threadID, pThreadInfo->totalInsertRows,
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或