Merge pull request #27837 from taosdata/fix/TD-30813-4

fix(stmt2/close): wait asnyc cb completed to free itself

source/client/inc/clientStmt2.h

@@ -150,10 +150,10 @@ typedef struct {
   SStmtExecInfo exec;
   SStmtBindInfo bInfo;
 
-  int64_t reqid;
+  int64_t       reqid;
-  int32_t errCode;
+  int32_t       errCode;
-  tsem_t  asyncQuerySem;
+  tsem_t        asyncQuerySem;
-
+  bool          semWaited;
   SStmtStatInfo stat;
 } STscStmt2;
 /*

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,
+        tscWarn("0x%" PRIx64 " refresh meta failed, code:%d - %s,QID:0x%" PRIx64, pRequest->self, code, tstrerror(code),
-                tstrerror(code), pRequest->requestId);
+                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]) {

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"

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);
+  // sem_wait(&sem);
-  (void)sem_destroy(&sem);
+  //(void)sem_destroy(&sem);
-  if (++run_time < 2) {
+  if (++run_time < 20) {
     goto _bind_again;
   }
   taos_stmt2_close(stmt);