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

Merge pull request #29935 from taosdata/fix/3.0/TD-33858 

fix(stmt2)[TD-33858]:fix stmt2 unit test deadlock
This commit is contained in:
Mario Peng 2025-02-27 10:45:47 +08:00 committed by GitHub
parent 016e2c7ba3 b1467e33bf
commit 03bd3190b4
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 36 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
source/client/src/clientMain.c
View File

@ -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); (void)atomic_add_fetch_8(&pStmt->asyncBindParam.asyncBindNum, 1);
int code_s = taosStmt2AsyncBind(stmtAsyncBindThreadFunc, (void *)args); int code_s = taosStmt2AsyncBind(stmtAsyncBindThreadFunc, (void *)args);
if (code_s != TSDB_CODE_SUCCESS) { 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)atomic_sub_fetch_8(&pStmt->asyncBindParam.asyncBindNum, 1);
(void)taosThreadMutexUnlock(&(pStmt->asyncBindParam.mutex));
// terrno = TAOS_SYSTEM_ERROR(errno); // terrno = TAOS_SYSTEM_ERROR(errno);
} }

13
source/client/src/clientStmt2.c
View File

@ -1690,11 +1690,11 @@ int stmtExec2(TAOS_STMT2* stmt, int* affected_rows) {
return pStmt->errCode; return pStmt->errCode;
} }
taosThreadMutexLock(&pStmt->asyncBindParam.mutex); (void)taosThreadMutexLock(&pStmt->asyncBindParam.mutex);
if (atomic_load_8((int8_t*)&pStmt->asyncBindParam.asyncBindNum) > 0) { while (atomic_load_8((int8_t*)&pStmt->asyncBindParam.asyncBindNum) > 0) {
(void)taosThreadCondWait(&pStmt->asyncBindParam.waitCond, &pStmt->asyncBindParam.mutex); (void)taosThreadCondWait(&pStmt->asyncBindParam.waitCond, &pStmt->asyncBindParam.mutex);
} }
taosThreadMutexUnlock(&pStmt->asyncBindParam.mutex); (void)taosThreadMutexUnlock(&pStmt->asyncBindParam.mutex);
if (pStmt->sql.stbInterlaceMode) { if (pStmt->sql.stbInterlaceMode) {
STMT_ERR_RET(stmtAddBatch2(pStmt)); STMT_ERR_RET(stmtAddBatch2(pStmt));
@ -1796,11 +1796,12 @@ int stmtClose2(TAOS_STMT2* stmt) {
pStmt->bindThreadInUse = false; pStmt->bindThreadInUse = false;
} }
taosThreadMutexLock(&pStmt->asyncBindParam.mutex); (void)taosThreadMutexLock(&pStmt->asyncBindParam.mutex);
if (atomic_load_8((int8_t*)&pStmt->asyncBindParam.asyncBindNum) > 0) { while (atomic_load_8((int8_t*)&pStmt->asyncBindParam.asyncBindNum) > 0) {
(void)taosThreadCondWait(&pStmt->asyncBindParam.waitCond, &pStmt->asyncBindParam.mutex); (void)taosThreadCondWait(&pStmt->asyncBindParam.waitCond, &pStmt->asyncBindParam.mutex);
} }
taosThreadMutexUnlock(&pStmt->asyncBindParam.mutex); (void)taosThreadMutexUnlock(&pStmt->asyncBindParam.mutex);
(void)taosThreadCondDestroy(&pStmt->queue.waitCond); (void)taosThreadCondDestroy(&pStmt->queue.waitCond);
(void)taosThreadMutexDestroy(&pStmt->queue.mutex); (void)taosThreadMutexDestroy(&pStmt->queue.mutex);

27
source/client/test/stmt2Test.cpp
View File

@ -15,6 +15,9 @@
#include <gtest/gtest.h> #include <gtest/gtest.h>
#include <string.h> #include <string.h>
#include <atomic>
#include <chrono>
#include <thread>
#include "clientInt.h" #include "clientInt.h"
#include "geosWrapper.h" #include "geosWrapper.h"
#include "osSemaphore.h" #include "osSemaphore.h"
@ -1662,7 +1665,7 @@ void stmtAsyncBindCb2(void* param, TAOS_RES* pRes, int code) {
return; return;
} }
TEST(stmt2Case, async_order) { void stmt2_async_test(std::atomic<bool>& stop_task) {
int CTB_NUMS = 2; int CTB_NUMS = 2;
int ROW_NUMS = 2; int ROW_NUMS = 2;
int CYC_NUMS = 2; int CYC_NUMS = 2;
@ -1868,5 +1871,27 @@ TEST(stmt2Case, async_order) {
taosMemoryFree(tbs[i]); taosMemoryFree(tbs[i]);
} }
taosMemoryFree(tbs); taosMemoryFree(tbs);
stop_task = true;
} }
TEST(stmt2Case, async_order) {
std::atomic<bool> 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<std::chrono::seconds>(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();
}
}
#pragma GCC diagnostic pop #pragma GCC diagnostic pop