enh: synchronize iteration over sub queries with subState.mutex

src/client/inc/tsclient.h

@@ -440,6 +440,8 @@ typedef struct SSqlStream {
 } SSqlStream;
 
 SSqlObj* tscAllocSqlObj();
+SSqlObj* tscAcquireRefOfSubobj(SSqlObj *pSql, int32_t idx);
+void tscReleaseRefOfSubobj(SSqlObj *pSql);
 
 void tscSetStreamDestTable(SSqlStream* pStream, const char* dstTable);
 

src/client/src/tscDelete.c

@@ -210,7 +210,10 @@ int32_t executeDelete(SSqlObj* pSql, SQueryInfo* pQueryInfo) {
   tscDebug("0x%"PRIx64":CDEL retrieved query data from %d vnode(s)", pSql->self, pSql->subState.numOfSub);
   pRes->code = TSDB_CODE_SUCCESS;
   
-  int32_t i;
+  int32_t i = 0;
+
+  { pthread_mutex_lock(&pSql->subState.mutex);
+
   for (i = 0; i < pSql->subState.numOfSub; ++i) {
     // vgroup
     SVgroupMsg* pVgroupMsg = &pTableMetaInfo->vgroupList->vgroups[i];
@@ -239,6 +242,8 @@ int32_t executeDelete(SSqlObj* pSql, SQueryInfo* pQueryInfo) {
     pRes->code = TSDB_CODE_TSC_OUT_OF_MEMORY;
   }
 
+  pthread_mutex_unlock(&pSql->subState.mutex); }
+
   if (pRes->code != TSDB_CODE_SUCCESS) {
     doCleanupSubqueries(pSql);
     return pRes->code;

src/client/src/tscProfile.c

@@ -281,7 +281,8 @@ int tscBuildQueryStreamDesc(void *pMsg, STscObj *pObj) {
 //      } else {
 //        pQdesc->stableQuery = 0;
 //      }
-      pthread_mutex_lock(&pSql->subState.mutex);
+      { pthread_mutex_lock(&pSql->subState.mutex);
+
       if (pSql->pSubs != NULL && pSql->subState.states != NULL) {
         for (int32_t i = 0; i < pSql->subState.numOfSub; ++i) {
           // because subState maybe free on anytime by any thread, check validate from here
@@ -298,7 +299,8 @@ int tscBuildQueryStreamDesc(void *pMsg, STscObj *pObj) {
         }
       }
       pQdesc->numOfSub = pSql->subState.numOfSub;
-      pthread_mutex_unlock(&pSql->subState.mutex);
+
+      pthread_mutex_unlock(&pSql->subState.mutex); }
     }
 
     pQdesc->numOfSub = htonl(pQdesc->numOfSub);

src/client/src/tscServer.c

@@ -353,8 +353,8 @@ void checkBrokenQueries(STscObj *pTscObj) {
         pSql->lastAlive = taosGetTimestampMs();
       }
     } else {
-      // lock subs
-      pthread_mutex_lock(&pSql->subState.mutex);
+      { pthread_mutex_lock(&pSql->subState.mutex);
+
       if (pSql->pSubs) {
         // have sub sql
         for (int i = 0; i < pSql->subState.numOfSub; i++) {
@@ -375,8 +375,8 @@ void checkBrokenQueries(STscObj *pTscObj) {
           }
         }
       }
-      // unlock
-      pthread_mutex_unlock(&pSql->subState.mutex);
+
+      pthread_mutex_unlock(&pSql->subState.mutex); }
     }
 
     // kill query

src/client/src/tscSql.c

@@ -747,7 +747,9 @@ static void tscKillSTableQuery(SSqlObj *pSql) {
   pSql->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
 
   tscLockByThread(&pSql->squeryLock);
-  
+
+  { pthread_mutex_lock(&pSql->subState.mutex);
+
   for (int i = 0; i < pSql->subState.numOfSub; ++i) {
     // NOTE: pSub may have been released already here
     SSqlObj *pSub = pSql->pSubs[i];
@@ -767,6 +769,8 @@ static void tscKillSTableQuery(SSqlObj *pSql) {
     // taosRelekaseRef(tscObjRef, pSubObj->self);
   }
 
+  pthread_mutex_unlock(&pSql->subState.mutex); }
+
   if (pSql->subState.numOfSub <= 0) {
     tscAsyncResultOnError(pSql);
   }

src/client/src/tscSub.c

@@ -83,7 +83,6 @@ void tscUpdateSubscriptionProgress(void* sub, int64_t uid, TSKEY ts) {
   }
 }
 
-
 static void asyncCallback(void *param, TAOS_RES *tres, int code) {
   assert(param != NULL);
   SSub *pSub = ((SSub *)param);

src/client/src/tscUtil.c

@@ -1749,6 +1749,29 @@ SSqlObj* tscAllocSqlObj() {
     return pNew;
 }
 
+SSqlObj* tscAcquireRefOfSubobj(SSqlObj* pSql, int32_t idx) {
+    assert (pSql != NULL);
+    SSqlObj *pSub = NULL;
+
+    pthread_mutex_lock(&pSql->subState.mutex);
+    if (idx < 0 ||
+        idx >= pSql->subState.numOfSub ||
+        !pSql->pSubs[idx]) {
+        goto _out;
+    }
+    pSub = taosAcquireRef(tscObjRef, pSql->pSubs[idx]->self);
+    assert (pSql->pSubs[idx] == pSub && "Refcounted subquery obj mismatch");
+
+_out:
+    pthread_mutex_unlock(&pSql->subState.mutex);
+    return pSub;
+}
+
+void tscReleaseRefOfSubobj(SSqlObj* pSub) {
+    assert (pSub != NULL && pSub->self != 0 && "Subquery obj not refcounted");
+    taosReleaseRef(tscObjRef, pSub->self);
+}
+
 void tscFreeSqlObj(SSqlObj* pSql) {
   if (pSql == NULL || pSql->signature != pSql) {
     return;
@@ -4209,14 +4232,16 @@ int32_t doReInitSubState(SSqlObj* pSql, int32_t numOfSubqueries) {
   tscFreeSubobj(pSql);
   int32_t code = TSDB_CODE_SUCCESS;
 
-  pthread_mutex_lock(&pSql->subState.mutex);
+  { pthread_mutex_lock(&pSql->subState.mutex);
+
   pSql->pSubs = calloc(numOfSubqueries, POINTER_BYTES);
   pSql->subState.states = calloc(numOfSubqueries, sizeof(int8_t));
   if (pSql->pSubs == NULL || pSql->subState.states == NULL) {
     code = TSDB_CODE_TSC_OUT_OF_MEMORY;
   }
   pSql->subState.numOfSub = numOfSubqueries;
-  pthread_mutex_unlock(&pSql->subState.mutex);
+
+  pthread_mutex_unlock(&pSql->subState.mutex); }
   return code;
 }
 
@@ -4247,6 +4272,9 @@ void executeQuery(SSqlObj* pSql, SQueryInfo* pQueryInfo) {
       goto _error;
     }
 
+    int errflag = 0;
+    { pthread_mutex_lock(&pSql->subState.mutex);
+
     for(int32_t i = 0; i < pSql->subState.numOfSub; ++i) {
       SQueryInfo* pSub = taosArrayGetP(pQueryInfo->pUpstream, i);
 
@@ -4256,7 +4284,8 @@ void executeQuery(SSqlObj* pSql, SQueryInfo* pQueryInfo) {
       SSqlObj* pNew = tscAllocSqlObj();
       if (pNew == NULL) {
         code = TSDB_CODE_TSC_OUT_OF_MEMORY;
-        goto _error;
+        errflag = 1;
+        break;
       }
 
       pNew->pTscObj   = pSql->pTscObj;
@@ -4272,19 +4301,23 @@ void executeQuery(SSqlObj* pSql, SQueryInfo* pQueryInfo) {
       SRetrieveSupport* ps = calloc(1, sizeof(SRetrieveSupport));  // todo use object id
       if (ps == NULL) {
         tscFreeSqlObj(pNew);
-        goto _error;
+        errflag = 1;
+        break;
       }
 
       ps->pParentSql = pSql;
       ps->subqueryIndex = i;
-
       pNew->param = ps;
+
+      registerSqlObj(pNew);
+
       pSql->pSubs[i] = pNew;
 
       SSqlCmd* pCmd = &pNew->cmd;
       pCmd->command = TSDB_SQL_SELECT;
       if ((code = tscAddQueryInfo(pCmd)) != TSDB_CODE_SUCCESS) {
-        goto _error;
+        errflag = 1;
+        break;
       }
 
       SQueryInfo* pNewQueryInfo = tscGetQueryInfo(pCmd);
@@ -4294,9 +4327,12 @@ void executeQuery(SSqlObj* pSql, SQueryInfo* pQueryInfo) {
       numOfInit++;
     }
 
+    pthread_mutex_unlock(&pSql->subState.mutex); }
+    if (errflag) { goto _error; }
+
     for(int32_t i = 0; i < pSql->subState.numOfSub; ++i) {
-      SSqlObj* psub = pSql->pSubs[i];
-      registerSqlObj(psub);
+      SSqlObj* psub = tscAcquireRefOfSubobj(pSql, i);  // ACQ REF
+      if (!psub) continue;
 
       // create sub query to handle the sub query.
       SQueryInfo* pq = tscGetQueryInfo(&psub->cmd);
@@ -4306,6 +4342,8 @@ void executeQuery(SSqlObj* pSql, SQueryInfo* pQueryInfo) {
         psub->cmd.command = TSDB_SQL_RETRIEVE_EMPTY_RESULT;
       }
       executeQuery(psub, pq);
+
+      tscReleaseRefOfSubobj(psub); // REL REF
     }
 
     return;

src/util/src/tref.c

@@ -195,7 +195,6 @@ int taosRemoveRef(int rsetId, int64_t rid)
   return taosDecRefCount(rsetId, rid, 1);
 }
 
-// if rid is 0, return the first p in hash list, otherwise, return the next after current rid
 void *taosAcquireRef(int rsetId, int64_t rid)
 {
   int       hash;