diff --git a/source/client/src/clientMain.c b/source/client/src/clientMain.c index 184f73c664..9a555682e6 100644 --- a/source/client/src/clientMain.c +++ b/source/client/src/clientMain.c @@ -2272,7 +2272,10 @@ int taos_stmt2_bind_param_a(TAOS_STMT2 *stmt, TAOS_STMT2_BINDV *bindv, int32_t c (void)atomic_add_fetch_8(&pStmt->asyncBindParam.asyncBindNum, 1); int code_s = taosStmt2AsyncBind(stmtAsyncBindThreadFunc, (void *)args); if (code_s != TSDB_CODE_SUCCESS) { + (void)taosThreadMutexLock(&(pStmt->asyncBindParam.mutex)); + (void)taosThreadCondSignal(&(pStmt->asyncBindParam.waitCond)); (void)atomic_sub_fetch_8(&pStmt->asyncBindParam.asyncBindNum, 1); + (void)taosThreadMutexUnlock(&(pStmt->asyncBindParam.mutex)); // terrno = TAOS_SYSTEM_ERROR(errno); } diff --git a/source/client/src/clientStmt2.c b/source/client/src/clientStmt2.c index 4462df740f..fc056a5648 100644 --- a/source/client/src/clientStmt2.c +++ b/source/client/src/clientStmt2.c @@ -1711,11 +1711,11 @@ int stmtExec2(TAOS_STMT2* stmt, int* affected_rows) { return pStmt->errCode; } - taosThreadMutexLock(&pStmt->asyncBindParam.mutex); - if (atomic_load_8((int8_t*)&pStmt->asyncBindParam.asyncBindNum) > 0) { + (void)taosThreadMutexLock(&pStmt->asyncBindParam.mutex); + while (atomic_load_8((int8_t*)&pStmt->asyncBindParam.asyncBindNum) > 0) { (void)taosThreadCondWait(&pStmt->asyncBindParam.waitCond, &pStmt->asyncBindParam.mutex); } - taosThreadMutexUnlock(&pStmt->asyncBindParam.mutex); + (void)taosThreadMutexUnlock(&pStmt->asyncBindParam.mutex); if (pStmt->sql.stbInterlaceMode) { STMT_ERR_RET(stmtAddBatch2(pStmt)); @@ -1817,11 +1817,12 @@ int stmtClose2(TAOS_STMT2* stmt) { pStmt->bindThreadInUse = false; } - taosThreadMutexLock(&pStmt->asyncBindParam.mutex); - if (atomic_load_8((int8_t*)&pStmt->asyncBindParam.asyncBindNum) > 0) { + (void)taosThreadMutexLock(&pStmt->asyncBindParam.mutex); + while (atomic_load_8((int8_t*)&pStmt->asyncBindParam.asyncBindNum) > 0) { (void)taosThreadCondWait(&pStmt->asyncBindParam.waitCond, &pStmt->asyncBindParam.mutex); } - taosThreadMutexUnlock(&pStmt->asyncBindParam.mutex); + (void)taosThreadMutexUnlock(&pStmt->asyncBindParam.mutex); + (void)taosThreadCondDestroy(&pStmt->queue.waitCond); (void)taosThreadMutexDestroy(&pStmt->queue.mutex); diff --git a/source/client/test/stmt2Test.cpp b/source/client/test/stmt2Test.cpp index eaed84ed49..6bae063124 100644 --- a/source/client/test/stmt2Test.cpp +++ b/source/client/test/stmt2Test.cpp @@ -15,6 +15,9 @@ #include #include +#include +#include +#include #include "clientInt.h" #include "geosWrapper.h" #include "osSemaphore.h" @@ -1706,7 +1709,7 @@ void stmtAsyncBindCb2(void* param, TAOS_RES* pRes, int code) { return; } -TEST(stmt2Case, async_order) { +void stmt2_async_test(std::atomic& stop_task) { int CTB_NUMS = 2; int ROW_NUMS = 2; int CYC_NUMS = 2; @@ -1912,6 +1915,27 @@ TEST(stmt2Case, async_order) { taosMemoryFree(tbs[i]); } taosMemoryFree(tbs); + stop_task = true; +} + +TEST(stmt2Case, async_order) { + std::atomic stop_task(false); + std::thread t(stmt2_async_test, std::ref(stop_task)); + + // 等待 60 秒钟 + auto start_time = std::chrono::steady_clock::now(); + while (!stop_task) { + auto elapsed_time = std::chrono::steady_clock::now() - start_time; + if (std::chrono::duration_cast(elapsed_time).count() > 60) { + FAIL() << "Test[stmt2_async_test] timed out"; + t.detach(); + break; + } + std::this_thread::sleep_for(std::chrono::seconds(1)); // 每 1s 检查一次 + } + if (t.joinable()) { + t.join(); + } } TEST(stmt2Case, rowformat_bind) { @@ -2067,4 +2091,5 @@ TEST(stmt2Case, rowformat_bind) { do_query(taos, "drop database if exists stmt2_testdb_16"); taos_close(taos); } + #pragma GCC diagnostic pop diff --git a/tests/system-test/3-enterprise/restore/kill_restore_dnode.py b/tests/system-test/3-enterprise/restore/kill_restore_dnode.py index d29a11f159..478c0db2a6 100644 --- a/tests/system-test/3-enterprise/restore/kill_restore_dnode.py +++ b/tests/system-test/3-enterprise/restore/kill_restore_dnode.py @@ -75,6 +75,12 @@ class TDTestCase: def restoreDnodeThread(self, p, newTdSql): sleep(1) + count = 0 + while count < 100: + newTdSql.query('show dnodes') + if newTdSql.queryResult[1][4] == "ready": + break + count+=1 sql = f"restore dnode 2" tdLog.info(sql)