Merge pull request #22465 from taosdata/fix/3_liaohj

fix(stream): add the back pressure for sink node.
taosdata · Aug 16, 2023 · f8f8e7a · f8f8e7a
2 parents f135907 + 59d5858
commit f8f8e7a
Show file tree

Hide file tree

Showing 3 changed files with 14 additions and 1 deletion.
diff --git a/source/dnode/vnode/src/tq/tqRestore.c b/source/dnode/vnode/src/tq/tqRestore.c
@@ -287,6 +287,13 @@ int32_t createStreamTaskRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle) {
       continue;
     }
 
+    // downstream task has blocked the output, stopped for a while
+    if (pTask->inputStatus == TASK_INPUT_STATUS__BLOCKED) {
+      tqDebug("s-task:%s inputQ is blocked, do nothing", pTask->id.idStr);
+      streamMetaReleaseTask(pStreamMeta, pTask);
+      continue;
+    }
+
     *pScanIdle = false;
 
     // seek the stored version and extract data from WAL

diff --git a/source/libs/stream/src/stream.c b/source/libs/stream/src/stream.c
@@ -362,7 +362,7 @@ int32_t tAppendDataToInputQueue(SStreamTask* pTask, SStreamQueueItem* pItem) {
            msgLen, ver, total, size + msgLen/1048576.0);
   } else if (type == STREAM_INPUT__DATA_BLOCK || type == STREAM_INPUT__DATA_RETRIEVE ||
              type == STREAM_INPUT__REF_DATA_BLOCK) {
-    if ((pTask->info.taskLevel == TASK_LEVEL__SOURCE) && (tInputQueueIsFull(pTask))) {
+    if (/*(pTask->info.taskLevel == TASK_LEVEL__SOURCE) && */(tInputQueueIsFull(pTask))) {
       qError("s-task:%s input queue is full, capacity:%d size:%d MiB, current(blocks:%d, size:%.2fMiB) abort",
              pTask->id.idStr, STREAM_TASK_INPUT_QUEUE_CAPACITY, STREAM_TASK_INPUT_QUEUE_CAPACITY_IN_SIZE, total,
              size);

diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c
@@ -191,6 +191,12 @@ int32_t streamScanExec(SStreamTask* pTask, int32_t batchSize) {
         return 0;
       }
 
+      if (pTask->inputStatus == TASK_INPUT_STATUS__BLOCKED) {
+        qDebug("s-task:%s inputQ is blocked, wait for 10sec and retry", pTask->id.idStr);
+        taosMsleep(10000);
+        continue;
+      }
+
       SSDataBlock* output = NULL;
       uint64_t     ts = 0;
       code = qExecTask(exec, &output, &ts);