diff --git a/source/client/inc/clientStmt2.h b/source/client/inc/clientStmt2.h index 74eb198930..4e9a09c082 100644 --- a/source/client/inc/clientStmt2.h +++ b/source/client/inc/clientStmt2.h @@ -150,10 +150,10 @@ typedef struct { SStmtExecInfo exec; SStmtBindInfo bInfo; - int64_t reqid; - int32_t errCode; - tsem_t asyncQuerySem; - + int64_t reqid; + int32_t errCode; + tsem_t asyncQuerySem; + bool semWaited; SStmtStatInfo stat; } STscStmt2; /* diff --git a/source/client/src/clientMain.c b/source/client/src/clientMain.c index 5b9684fb5e..cc7a7f9489 100644 --- a/source/client/src/clientMain.c +++ b/source/client/src/clientMain.c @@ -1317,8 +1317,8 @@ void doAsyncQuery(SRequestObj *pRequest, bool updateMetaForce) { tscDebug("0x%" PRIx64 " client retry to handle the error, code:%d - %s, tryCount:%d,QID:0x%" PRIx64, pRequest->self, code, tstrerror(code), pRequest->retry, pRequest->requestId); if (TSDB_CODE_SUCCESS != refreshMeta(pRequest->pTscObj, pRequest)) { - tscWarn("0x%" PRIx64 " refresh meta failed, code:%d - %s,QID:0x%" PRIx64, pRequest->self, code, - tstrerror(code), pRequest->requestId); + tscWarn("0x%" PRIx64 " refresh meta failed, code:%d - %s,QID:0x%" PRIx64, pRequest->self, code, tstrerror(code), + pRequest->requestId); } pRequest->prevCode = code; doAsyncQuery(pRequest, true); @@ -1369,7 +1369,7 @@ typedef struct SAsyncFetchParam { void *param; } SAsyncFetchParam; -static int32_t doAsyncFetch(void* pParam) { +static int32_t doAsyncFetch(void *pParam) { SAsyncFetchParam *param = pParam; taosAsyncFetchImpl(param->pReq, param->fp, param->param); taosMemoryFree(param); @@ -1393,7 +1393,7 @@ void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) { return; } - SAsyncFetchParam* pParam = taosMemoryCalloc(1, sizeof(SAsyncFetchParam)); + SAsyncFetchParam *pParam = taosMemoryCalloc(1, sizeof(SAsyncFetchParam)); if (!pParam) { fp(param, res, terrno); return; @@ -1983,6 +1983,12 @@ int taos_stmt2_bind_param(TAOS_STMT2 *stmt, TAOS_STMT2_BINDV *bindv, int32_t col return terrno; } + STscStmt2 *pStmt = (STscStmt2 *)stmt; + if (pStmt->options.asyncExecFn && !pStmt->semWaited) { + (void)tsem_wait(&pStmt->asyncQuerySem); + pStmt->semWaited = true; + } + int32_t code = 0; for (int i = 0; i < bindv->count; ++i) { if (bindv->tbnames && bindv->tbnames[i]) { diff --git a/source/client/src/clientStmt2.c b/source/client/src/clientStmt2.c index 01a7c4be4c..ce99cd7deb 100644 --- a/source/client/src/clientStmt2.c +++ b/source/client/src/clientStmt2.c @@ -818,6 +818,7 @@ TAOS_STMT2* stmtInit2(STscObj* taos, TAOS_STMT2_OPTION* pOptions) { if (pStmt->options.asyncExecFn) { (void)tsem_init(&pStmt->asyncQuerySem, 0, 1); } + pStmt->semWaited = false; STMT_LOG_SEQ(STMT_INIT); @@ -1262,10 +1263,6 @@ int stmtBindBatch2(TAOS_STMT2* stmt, TAOS_STMT2_BIND* bind, int32_t colIdx) { STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND)); - if (pStmt->options.asyncExecFn) { - (void)tsem_wait(&pStmt->asyncQuerySem); - } - if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 && STMT_TYPE_MULTI_INSERT != pStmt->sql.type) { pStmt->bInfo.needParse = false; @@ -1666,7 +1663,6 @@ int stmtExec2(TAOS_STMT2* stmt, int* affected_rows) { STMT_ERR_RET(stmtCleanExecInfo(pStmt, (code ? false : true), false)); ++pStmt->sql.runTimes; - } else { SSqlCallbackWrapper* pWrapper = taosMemoryCalloc(1, sizeof(SSqlCallbackWrapper)); if (pWrapper == NULL) { @@ -1682,6 +1678,7 @@ int stmtExec2(TAOS_STMT2* stmt, int* affected_rows) { pRequest->body.queryFp = asyncQueryCb; ((SSyncQueryParam*)(pRequest)->body.interParam)->userParam = pStmt; + pStmt->semWaited = false; launchAsyncQuery(pRequest, pStmt->sql.pQuery, NULL, pWrapper); } @@ -1703,6 +1700,10 @@ int stmtClose2(TAOS_STMT2* stmt) { pStmt->bindThreadInUse = false; } + if (pStmt->options.asyncExecFn && !pStmt->semWaited) { + (void)tsem_wait(&pStmt->asyncQuerySem); + } + STMT_DLOG("stmt %p closed, stbInterlaceMode: %d, statInfo: ctgGetTbMetaNum=>%" PRId64 ", getCacheTbInfo=>%" PRId64 ", parseSqlNum=>%" PRId64 ", pStmt->stat.bindDataNum=>%" PRId64 ", settbnameAPI:%u, bindAPI:%u, addbatchAPI:%u, execAPI:%u" diff --git a/tests/script/api/stmt2-nohole.c b/tests/script/api/stmt2-nohole.c index b29dd3e826..5954f3660b 100644 --- a/tests/script/api/stmt2-nohole.c +++ b/tests/script/api/stmt2-nohole.c @@ -14,12 +14,12 @@ int64_t genReqid() { return count; } -sem_t sem; +// sem_t sem; void stmtAsyncQueryCb(void* param, TAOS_RES* pRes, int code) { int affected_rows = taos_affected_rows(pRes); printf("\033[31maffected rows:%d\033[0m\n", affected_rows); - (void)sem_post(&sem); + //(void)sem_post(&sem); return; /* SSP_CB_PARAM* qParam = (SSP_CB_PARAM*)param; @@ -319,7 +319,7 @@ _bind_again: taos_stmt2_free_fields(stmt, fields); */ // if (taos_stmt_execute(stmt) != 0) { - (void)sem_init(&sem, 0, 0); + //(void)sem_init(&sem, 0, 0); start = clock(); // if (taos_stmt2_exec(stmt, NULL, stmtAsyncQueryCb, NULL) != 0) { if (taos_stmt2_exec(stmt, NULL) != 0) { @@ -330,9 +330,9 @@ _bind_again: end = clock(); printf("exec time:%f\n", (double)(end - start) / CLOCKS_PER_SEC); - sem_wait(&sem); - (void)sem_destroy(&sem); - if (++run_time < 2) { + // sem_wait(&sem); + //(void)sem_destroy(&sem); + if (++run_time < 20) { goto _bind_again; } taos_stmt2_close(stmt);