#3612

Committed 14 Feb 2025 06:43AM UTC coverage: 63.513% (+0.06%) from 63.456%

Build # #3612

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #29757 from taosdata/feat/TS-5486

feat: support interp function with time range and time interval

Run Details

141465 of 286269 branches covered (49.42%)

Branch coverage included in aggregate %.

220292 of 283307 relevant lines covered (77.76%)

18570887.65 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

78.56

/source/dnode/vnode/src/tq/tqStreamTask.c

/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include "tq.h"
#include "vnd.h"

#define MAX_REPEAT_SCAN_THRESHOLD 3
#define SCAN_WAL_IDLE_DURATION    500    // idle for 500ms to do next wal scan

typedef struct SBuildScanWalMsgParam {
  int64_t metaId;
  int32_t numOfTasks;
  int8_t  restored;
  SMsgCb  msgCb;
} SBuildScanWalMsgParam;

static int32_t doScanWalForAllTasks(SStreamMeta* pStreamMeta);
static int32_t setWalReaderStartOffset(SStreamTask* pTask, int32_t vgId);
static bool    handleFillhistoryScanComplete(SStreamTask* pTask, int64_t ver);
static bool    taskReadyForDataFromWal(SStreamTask* pTask);
static int32_t doPutDataIntoInputQ(SStreamTask* pTask, int64_t maxVer, int32_t* numOfItems, bool* pSucc);
static int32_t tqScanWalInFuture(STQ* pTq, int32_t numOfTasks, int32_t idleDuration);
static int32_t doScanWalAsync(STQ* pTq, bool ckPause);

// extract data blocks(submit/delete) from WAL, and add them into the input queue for all the sources tasks.
int32_t tqScanWal(STQ* pTq) {
  SStreamMeta* pMeta = pTq->pStreamMeta;
  int32_t      vgId = pMeta->vgId;
  int64_t      st = taosGetTimestampMs();
  int32_t      numOfTasks = 0;

  tqDebug("vgId:%d continue to check if data in wal are available, scanCounter:%d", vgId, pMeta->scanInfo.scanCounter);

  // check all tasks
  int32_t code = doScanWalForAllTasks(pMeta);
  if (code) {
    tqError("vgId:%d failed to start all tasks, try next time, code:%s", vgId, tstrerror(code));
    return code;
  }

  streamMetaWLock(pMeta);
  int32_t times = (--pMeta->scanInfo.scanCounter);
  if (times < 0) {
    tqError("vgId:%d invalid scan counter:%d, reset to 0", vgId, times);
    times = 0;
  }

  numOfTasks = taosArrayGetSize(pMeta->pTaskList);
  streamMetaWUnLock(pMeta);

  int64_t el = (taosGetTimestampMs() - st);
  tqDebug("vgId:%d scan wal for stream tasks completed, elapsed time:%" PRId64 " ms", vgId, el);

  if (times > 0) {
    tqDebug("vgId:%d scan wal for stream tasks for %d times in %dms", vgId, times, SCAN_WAL_IDLE_DURATION);
    code = tqScanWalInFuture(pTq, numOfTasks, SCAN_WAL_IDLE_DURATION);
    if (code) {
      tqError("vgId:%d sched scan wal in %dms failed, ignore this failure", vgId, SCAN_WAL_IDLE_DURATION);
    }
  }

  return code;
}

static void doStartScanWal(void* param, void* tmrId) {
  int32_t vgId = 0;
  int32_t code = 0;

  SBuildScanWalMsgParam* pParam = (SBuildScanWalMsgParam*)param;

  SStreamMeta* pMeta = taosAcquireRef(streamMetaRefPool, pParam->metaId);
  if (pMeta == NULL) {
    tqError("metaRid:%" PRId64 " not valid now, stream meta has been freed", pParam->metaId);
    taosMemoryFree(pParam);
    return;
  }

  if (pMeta->closeFlag) {
    code = taosReleaseRef(streamMetaRefPool, pParam->metaId);
    if (code == TSDB_CODE_SUCCESS) {
      tqDebug("vgId:%d jump out of scan wal timer since closed", vgId);
    } else {
      tqError("vgId:%d failed to release ref for streamMeta, rid:%" PRId64 " code:%s", vgId, pParam->metaId,
              tstrerror(code));
    }

    taosMemoryFree(pParam);
    return;
  }

  vgId = pMeta->vgId;

  tqDebug("vgId:%d create msg to start wal scan, numOfTasks:%d, vnd restored:%d", vgId, pParam->numOfTasks,
          pParam->restored);
#if 0
  // wait for the vnode is freed, and invalid read may occur.
  taosMsleep(10000);
#endif

  code = streamTaskSchedTask(&pParam->msgCb, vgId, 0, 0, STREAM_EXEC_T_EXTRACT_WAL_DATA);
  if (code) {
    tqError("vgId:%d failed sched task to scan wal, code:%s", vgId, tstrerror(code));
  }

  code = taosReleaseRef(streamMetaRefPool, pParam->metaId);
  if (code) {
    tqError("vgId:%d failed to release ref for streamMeta, rid:%" PRId64 " code:%s", vgId, pParam->metaId,
            tstrerror(code));
  }

  taosMemoryFree(pParam);
}

int32_t tqScanWalInFuture(STQ* pTq, int32_t numOfTasks, int32_t idleDuration) {
  SStreamMeta*           pMeta = pTq->pStreamMeta;
  int32_t                code = 0;
  int32_t                vgId = TD_VID(pTq->pVnode);
  tmr_h                  pTimer = NULL;
  SBuildScanWalMsgParam* pParam = NULL;

  pParam = taosMemoryMalloc(sizeof(SBuildScanWalMsgParam));
  if (pParam == NULL) {
    return terrno;
  }

  pParam->metaId = pMeta->rid;
  pParam->numOfTasks = numOfTasks;
  pParam->restored = pTq->pVnode->restored;
  pParam->msgCb = pTq->pVnode->msgCb;

  code = streamTimerGetInstance(&pTimer);
  if (code) {
    tqError("vgId:%d failed to get tmr ctrl during sched scan wal", vgId);
    taosMemoryFree(pParam);
  } else {
    streamTmrStart(doStartScanWal, idleDuration, pParam, pTimer, &pMeta->scanInfo.scanTimer, vgId, "scan-wal-fut");
  }

  return code;
}

int32_t tqScanWalAsync(STQ* pTq, bool ckPause) {
  SStreamMeta* pMeta = pTq->pStreamMeta;
  bool         alreadyRestored = pTq->pVnode->restored;
  int32_t      code = 0;

  // do not launch the stream tasks, if it is a follower or not restored vnode.
  if (!(vnodeIsRoleLeader(pTq->pVnode) && alreadyRestored)) {
    return TSDB_CODE_SUCCESS;
  }

  streamMetaWLock(pMeta);
  code = doScanWalAsync(pTq, ckPause);
  streamMetaWUnLock(pMeta);
  return code;
}

int32_t tqStopStreamTasksAsync(STQ* pTq) {
  SStreamMeta* pMeta = pTq->pStreamMeta;
  int32_t      vgId = pMeta->vgId;
  return streamTaskSchedTask(&pTq->pVnode->msgCb, vgId, 0, 0, STREAM_EXEC_T_STOP_ALL_TASKS);
}

int32_t setWalReaderStartOffset(SStreamTask* pTask, int32_t vgId) {
  // seek the stored version and extract data from WAL
  int64_t firstVer = walReaderGetValidFirstVer(pTask->exec.pWalReader);
  if (pTask->chkInfo.nextProcessVer < firstVer) {
    tqWarn("vgId:%d s-task:%s ver:%" PRId64 " earlier than the first ver of wal range %" PRId64 ", forward to %" PRId64,
           vgId, pTask->id.idStr, pTask->chkInfo.nextProcessVer, firstVer, firstVer);

    pTask->chkInfo.nextProcessVer = firstVer;

    // todo need retry if failed
    int32_t code = walReaderSeekVer(pTask->exec.pWalReader, pTask->chkInfo.nextProcessVer);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

    // append the data for the stream
    tqDebug("vgId:%d s-task:%s wal reader seek to ver:%" PRId64, vgId, pTask->id.idStr, pTask->chkInfo.nextProcessVer);
  } else {
    int64_t currentVer = walReaderGetCurrentVer(pTask->exec.pWalReader);
    if (currentVer == -1) {  // we only seek the read for the first time
      int32_t code = walReaderSeekVer(pTask->exec.pWalReader, pTask->chkInfo.nextProcessVer);
      if (code != TSDB_CODE_SUCCESS) {  // no data in wal, quit
        return code;
      }

      // append the data for the stream
      tqDebug("vgId:%d s-task:%s wal reader initial seek to ver:%" PRId64, vgId, pTask->id.idStr,
              pTask->chkInfo.nextProcessVer);
    }
  }

  int64_t skipToVer = walReaderGetSkipToVersion(pTask->exec.pWalReader);
  if (skipToVer != 0 && skipToVer > pTask->chkInfo.nextProcessVer) {
    int32_t code = walReaderSeekVer(pTask->exec.pWalReader, skipToVer);
    if (code != TSDB_CODE_SUCCESS) {  // no data in wal, quit
      return code;
    }

    tqDebug("vgId:%d s-task:%s wal reader jump to ver:%" PRId64, vgId, pTask->id.idStr, skipToVer);
  }

  return TSDB_CODE_SUCCESS;
}

// todo handle memory error
bool handleFillhistoryScanComplete(SStreamTask* pTask, int64_t ver) {
  const char* id = pTask->id.idStr;
  int64_t     maxVer = pTask->step2Range.maxVer;

  if ((pTask->info.fillHistory == 1) && ver > maxVer) {
    if (!pTask->status.appendTranstateBlock) {
      qWarn("s-task:%s fill-history scan WAL, nextProcessVer:%" PRId64 " out of the maximum ver:%" PRId64
            ", not scan wal anymore, add transfer-state block into inputQ",
            id, ver, maxVer);

      double el = (taosGetTimestampMs() - pTask->execInfo.step2Start) / 1000.0;
      qDebug("s-task:%s scan-history from WAL stage(step 2) ended, range:%" PRId64 "-%" PRId64 ", elapsed time:%.2fs",
             id, pTask->step2Range.minVer, maxVer, el);
      int32_t code = streamTaskPutTranstateIntoInputQ(pTask);
      if (code) {
        qError("s-task:%s failed to put trans-state into inputQ", id);
      }

      return true;
    } else {
      qWarn("s-task:%s fill-history scan WAL, nextProcessVer:%" PRId64 " out of the ver range:%" PRId64 "-%" PRId64
            ", not scan wal",
            id, ver, pTask->step2Range.minVer, maxVer);
    }
  }

  return false;
}

bool taskReadyForDataFromWal(SStreamTask* pTask) {
  // non-source or fill-history tasks don't need to response the WAL scan action.
  SSTaskBasicInfo* pInfo = &pTask->info;
  if ((pInfo->taskLevel != TASK_LEVEL__SOURCE) || (pTask->status.downstreamReady == 0)) {
    return false;
  }

  if (pInfo->taskLevel == TASK_LEVEL__SOURCE && pInfo->trigger == STREAM_TRIGGER_FORCE_WINDOW_CLOSE) {
    return false;
  }

  // not in ready state, do not handle the data from wal
  SStreamTaskState pState = streamTaskGetStatus(pTask);
  if (pState.state != TASK_STATUS__READY) {
    tqTrace("s-task:%s not ready for submit block in wal, status:%s", pTask->id.idStr, pState.name);
    return false;
  }

  // fill-history task has entered into the last phase, no need to anything
  if ((pInfo->fillHistory == 1) && pTask->status.appendTranstateBlock) {
    // the maximum version of data in the WAL has reached already, the step2 is done
    tqDebug("s-task:%s fill-history reach the maximum ver:%" PRId64 ", not scan wal anymore", pTask->id.idStr,
            pTask->dataRange.range.maxVer);
    return false;
  }

  // check if input queue is full or not
  if (streamQueueIsFull(pTask->inputq.queue)) {
    tqTrace("s-task:%s input queue is full, do nothing", pTask->id.idStr);
    return false;
  }

  // the input queue of downstream task is full, so the output is blocked, stopped for a while
  if (pTask->inputq.status == TASK_INPUT_STATUS__BLOCKED) {
    tqDebug("s-task:%s inputQ is blocked, do nothing", pTask->id.idStr);
    return false;
  }

  return true;
}

int32_t doPutDataIntoInputQ(SStreamTask* pTask, int64_t maxVer, int32_t* numOfItems, bool* pSucc) {
  const char* id = pTask->id.idStr;
  int32_t     numOfNewItems = 0;
  int32_t     code = 0;
  *pSucc = false;

  while (1) {
    if ((pTask->info.fillHistory == 1) && pTask->status.appendTranstateBlock) {
      *numOfItems += numOfNewItems;
      return numOfNewItems > 0;
    }

    SStreamQueueItem* pItem = NULL;
    code = extractMsgFromWal(pTask->exec.pWalReader, (void**)&pItem, maxVer, id);
    if (code != TSDB_CODE_SUCCESS || pItem == NULL) {  // failed, continue
      int64_t currentVer = walReaderGetCurrentVer(pTask->exec.pWalReader);
      bool    itemInFillhistory = handleFillhistoryScanComplete(pTask, currentVer);
      if (itemInFillhistory) {
        numOfNewItems += 1;
      }
      break;
    }

    if (pItem != NULL) {
      code = streamTaskPutDataIntoInputQ(pTask, pItem);
      if (code == TSDB_CODE_SUCCESS) {
        numOfNewItems += 1;
        int64_t ver = walReaderGetCurrentVer(pTask->exec.pWalReader);
        pTask->chkInfo.nextProcessVer = ver;
        tqDebug("s-task:%s set ver:%" PRId64 " for reader after extract data from WAL", id, ver);

        bool itemInFillhistory = handleFillhistoryScanComplete(pTask, ver);
        if (itemInFillhistory) {
          break;
        }
      } else {
        if (code == TSDB_CODE_OUT_OF_MEMORY) {
          tqError("s-task:%s failed to put data into inputQ, since out of memory", id);
        } else {
          tqTrace("s-task:%s append input queue failed, code:inputQ is full, ver:%" PRId64, id,
                  pTask->chkInfo.nextProcessVer);
          code = walReaderSeekVer(pTask->exec.pWalReader, pTask->chkInfo.nextProcessVer);
          if (code) {
            tqError("s-task:%s failed to seek ver to:%" PRId64 " in wal", id, pTask->chkInfo.nextProcessVer);
          }

          code = 0;  // reset the error code
        }

        break;
      }
    }
  }

  *numOfItems += numOfNewItems;
  *pSucc = (numOfNewItems > 0);
  return code;
}

int32_t doScanWalForAllTasks(SStreamMeta* pStreamMeta) {
  int32_t vgId = pStreamMeta->vgId;
  SArray* pTaskList = NULL;
  int32_t numOfTasks = taosArrayGetSize(pStreamMeta->pTaskList);
  if (numOfTasks == 0) {
    return TSDB_CODE_SUCCESS;
  }

  // clone the task list, to avoid the task update during scan wal files
  streamMetaWLock(pStreamMeta);
  pTaskList = taosArrayDup(pStreamMeta->pTaskList, NULL);
  streamMetaWUnLock(pStreamMeta);
  if (pTaskList == NULL) {
    tqError("vgId:%d failed to create task list dup, code:%s", vgId, tstrerror(terrno));
    return terrno;
  }

  tqDebug("vgId:%d start to check wal to extract new submit block for %d tasks", vgId, numOfTasks);

  // update the new task number
  numOfTasks = taosArrayGetSize(pTaskList);

  for (int32_t i = 0; i < numOfTasks; ++i) {
    STaskId* pTaskId = taosArrayGet(pTaskList, i);
    if (pTaskId == NULL) {
      continue;
    }

    SStreamTask* pTask = NULL;
    int32_t      code = streamMetaAcquireTask(pStreamMeta, pTaskId->streamId, pTaskId->taskId, &pTask);
    if (pTask == NULL || code != 0) {
      continue;
    }

    if (!taskReadyForDataFromWal(pTask)) {
      streamMetaReleaseTask(pStreamMeta, pTask);
      continue;
    }

    // seek the stored version and extract data from WAL
    code = setWalReaderStartOffset(pTask, vgId);
    if (code != TSDB_CODE_SUCCESS) {
      streamMetaReleaseTask(pStreamMeta, pTask);
      continue;
    }

    int32_t numOfItems = streamQueueGetNumOfItems(pTask->inputq.queue);
    int64_t maxVer = (pTask->info.fillHistory == 1) ? pTask->step2Range.maxVer : INT64_MAX;

    streamMutexLock(&pTask->lock);

    SStreamTaskState state = streamTaskGetStatus(pTask);
    if (state.state != TASK_STATUS__READY) {
      tqDebug("s-task:%s not ready for submit block from wal, status:%s", pTask->id.idStr, state.name);
      streamMutexUnlock(&pTask->lock);
      streamMetaReleaseTask(pStreamMeta, pTask);
      continue;
    }

    bool hasNewData = false;
    code = doPutDataIntoInputQ(pTask, maxVer, &numOfItems, &hasNewData);
    streamMutexUnlock(&pTask->lock);

    if ((numOfItems > 0) || hasNewData) {
      code = streamTrySchedExec(pTask);
      if (code != TSDB_CODE_SUCCESS) {
        streamMetaReleaseTask(pStreamMeta, pTask);
        taosArrayDestroy(pTaskList);
        return code;
      }
    }

    streamMetaReleaseTask(pStreamMeta, pTask);
  }

  taosArrayDestroy(pTaskList);
  return TSDB_CODE_SUCCESS;
}

int32_t doScanWalAsync(STQ* pTq, bool ckPause) {
  SStreamMeta* pMeta = pTq->pStreamMeta;
  bool         alreadyRestored = pTq->pVnode->restored;
  int32_t      vgId = pMeta->vgId;
  int32_t      numOfTasks = taosArrayGetSize(pMeta->pTaskList);

  if (numOfTasks == 0) {
    tqDebug("vgId:%d no stream tasks existed to run", vgId);
    return 0;
  }

  if (pMeta->startInfo.startAllTasks) {
    tqTrace("vgId:%d in restart procedure, not scan wal", vgId);
    return 0;
  }

  pMeta->scanInfo.scanCounter += 1;
  if (pMeta->scanInfo.scanCounter > MAX_REPEAT_SCAN_THRESHOLD) {
    pMeta->scanInfo.scanCounter = MAX_REPEAT_SCAN_THRESHOLD;
  }

  if (pMeta->scanInfo.scanCounter > 1) {
    tqDebug("vgId:%d wal read task has been launched, remain scan times:%d", vgId, pMeta->scanInfo.scanCounter);
    return 0;
  }

  int32_t numOfPauseTasks = pMeta->numOfPausedTasks;
  if (ckPause && numOfTasks == numOfPauseTasks) {
    tqDebug("vgId:%d ignore all submit, all streams had been paused, reset the walScanCounter", vgId);

    // reset the counter value, since we do not launch the scan wal operation.
    pMeta->scanInfo.scanCounter = 0;
    return 0;
  }

  tqDebug("vgId:%d create msg to start wal scan to launch stream tasks, numOfTasks:%d, vnd restored:%d", vgId,
          numOfTasks, alreadyRestored);

  return streamTaskSchedTask(&pTq->pVnode->msgCb, vgId, 0, 0, STREAM_EXEC_T_EXTRACT_WAL_DATA);
}

void streamMetaFreeTQDuringScanWalError(STQ* pTq) {
  SBuildScanWalMsgParam* p = taosMemoryCalloc(1, sizeof(SBuildScanWalMsgParam));
  p->metaId = pTq->pStreamMeta->rid;
  p->numOfTasks = 0;

  doStartScanWal(p, 0);
}

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15
16	#include "tq.h"
17	#include "vnd.h"
18
19	#define MAX_REPEAT_SCAN_THRESHOLD 3
20	#define SCAN_WAL_IDLE_DURATION 500 // idle for 500ms to do next wal scan
21
22	typedef struct SBuildScanWalMsgParam {
23	int64_t metaId;
24	int32_t numOfTasks;
25	int8_t restored;
26	SMsgCb msgCb;
27	} SBuildScanWalMsgParam;
28
29	static int32_t doScanWalForAllTasks(SStreamMeta* pStreamMeta);
30	static int32_t setWalReaderStartOffset(SStreamTask* pTask, int32_t vgId);
31	static bool handleFillhistoryScanComplete(SStreamTask* pTask, int64_t ver);
32	static bool taskReadyForDataFromWal(SStreamTask* pTask);
33	static int32_t doPutDataIntoInputQ(SStreamTask* pTask, int64_t maxVer, int32_t* numOfItems, bool* pSucc);
34	static int32_t tqScanWalInFuture(STQ* pTq, int32_t numOfTasks, int32_t idleDuration);
35	static int32_t doScanWalAsync(STQ* pTq, bool ckPause);
36
37	// extract data blocks(submit/delete) from WAL, and add them into the input queue for all the sources tasks.
38	int32_t tqScanWal(STQ* pTq) {	73,297✔
39	SStreamMeta* pMeta = pTq->pStreamMeta;	73,297✔
40	int32_t vgId = pMeta->vgId;	73,297✔
41	int64_t st = taosGetTimestampMs();	73,314✔
42	int32_t numOfTasks = 0;	73,314✔
43
44	tqDebug("vgId:%d continue to check if data in wal are available, scanCounter:%d", vgId, pMeta->scanInfo.scanCounter);	73,314✔
45
46	// check all tasks
47	int32_t code = doScanWalForAllTasks(pMeta);	73,315✔
48	if (code) {	73,310✔
49	tqError("vgId:%d failed to start all tasks, try next time, code:%s", vgId, tstrerror(code));	1!
50	return code;	1✔
51	}
52
53	streamMetaWLock(pMeta);	73,309✔
54	int32_t times = (--pMeta->scanInfo.scanCounter);	73,305✔
55	if (times < 0) {	73,305!
56	tqError("vgId:%d invalid scan counter:%d, reset to 0", vgId, times);	×
57	times = 0;	×
58	}
59
60	numOfTasks = taosArrayGetSize(pMeta->pTaskList);	73,305✔
61	streamMetaWUnLock(pMeta);	73,300✔
62
63	int64_t el = (taosGetTimestampMs() - st);	73,310✔
64	tqDebug("vgId:%d scan wal for stream tasks completed, elapsed time:%" PRId64 " ms", vgId, el);	73,310✔
65
66	if (times > 0) {	73,310✔
67	tqDebug("vgId:%d scan wal for stream tasks for %d times in %dms", vgId, times, SCAN_WAL_IDLE_DURATION);	22,559✔
68	code = tqScanWalInFuture(pTq, numOfTasks, SCAN_WAL_IDLE_DURATION);	22,559✔
69	if (code) {	22,562!
70	tqError("vgId:%d sched scan wal in %dms failed, ignore this failure", vgId, SCAN_WAL_IDLE_DURATION);	×
71	}
72	}
73
74	return code;	73,270✔
75	}
76
77	static void doStartScanWal(void* param, void* tmrId) {	22,562✔
78	int32_t vgId = 0;	22,562✔
79	int32_t code = 0;	22,562✔
80
81	SBuildScanWalMsgParam* pParam = (SBuildScanWalMsgParam*)param;	22,562✔
82
83	SStreamMeta* pMeta = taosAcquireRef(streamMetaRefPool, pParam->metaId);	22,562✔
84	if (pMeta == NULL) {	22,562✔
85	tqError("metaRid:%" PRId64 " not valid now, stream meta has been freed", pParam->metaId);	201!
86	taosMemoryFree(pParam);	201!
87	return;	201✔
88	}
89
90	if (pMeta->closeFlag) {	22,361✔
91	code = taosReleaseRef(streamMetaRefPool, pParam->metaId);	20✔
92	if (code == TSDB_CODE_SUCCESS) {	20!
93	tqDebug("vgId:%d jump out of scan wal timer since closed", vgId);	20✔
94	} else {
95	tqError("vgId:%d failed to release ref for streamMeta, rid:%" PRId64 " code:%s", vgId, pParam->metaId,	×
96	tstrerror(code));
97	}
98
99	taosMemoryFree(pParam);	20!
100	return;	20✔
101	}
102
103	vgId = pMeta->vgId;	22,341✔
104
105	tqDebug("vgId:%d create msg to start wal scan, numOfTasks:%d, vnd restored:%d", vgId, pParam->numOfTasks,	22,341✔
106	pParam->restored);
107	#if 0
108	// wait for the vnode is freed, and invalid read may occur.
109	taosMsleep(10000);
110	#endif
111
112	code = streamTaskSchedTask(&pParam->msgCb, vgId, 0, 0, STREAM_EXEC_T_EXTRACT_WAL_DATA);	22,341✔
113	if (code) {	22,341✔
114	tqError("vgId:%d failed sched task to scan wal, code:%s", vgId, tstrerror(code));	527!
115	}
116
117	code = taosReleaseRef(streamMetaRefPool, pParam->metaId);	22,341✔
118	if (code) {	22,341!
119	tqError("vgId:%d failed to release ref for streamMeta, rid:%" PRId64 " code:%s", vgId, pParam->metaId,	×
120	tstrerror(code));
121	}
122
123	taosMemoryFree(pParam);	22,341!
124	}
125
126	int32_t tqScanWalInFuture(STQ* pTq, int32_t numOfTasks, int32_t idleDuration) {	22,556✔
127	SStreamMeta* pMeta = pTq->pStreamMeta;	22,556✔
128	int32_t code = 0;	22,556✔
129	int32_t vgId = TD_VID(pTq->pVnode);	22,556✔
130	tmr_h pTimer = NULL;	22,556✔
131	SBuildScanWalMsgParam* pParam = NULL;	22,556✔
132
133	pParam = taosMemoryMalloc(sizeof(SBuildScanWalMsgParam));	22,556!
134	if (pParam == NULL) {	22,560!
135	return terrno;	×
136	}
137
138	pParam->metaId = pMeta->rid;	22,560✔
139	pParam->numOfTasks = numOfTasks;	22,560✔
140	pParam->restored = pTq->pVnode->restored;	22,560✔
141	pParam->msgCb = pTq->pVnode->msgCb;	22,560✔
142
143	code = streamTimerGetInstance(&pTimer);	22,560✔
144	if (code) {	22,556!
145	tqError("vgId:%d failed to get tmr ctrl during sched scan wal", vgId);	×
146	taosMemoryFree(pParam);	×
147	} else {
148	streamTmrStart(doStartScanWal, idleDuration, pParam, pTimer, &pMeta->scanInfo.scanTimer, vgId, "scan-wal-fut");	22,556✔
149	}
150
151	return code;	22,562✔
152	}
153
154	int32_t tqScanWalAsync(STQ* pTq, bool ckPause) {	508,607✔
155	SStreamMeta* pMeta = pTq->pStreamMeta;	508,607✔
156	bool alreadyRestored = pTq->pVnode->restored;	508,607✔
157	int32_t code = 0;	508,607✔
158
159	// do not launch the stream tasks, if it is a follower or not restored vnode.
160	if (!(vnodeIsRoleLeader(pTq->pVnode) && alreadyRestored)) {	508,607✔
161	return TSDB_CODE_SUCCESS;	23,150✔
162	}
163
164	streamMetaWLock(pMeta);	485,549✔
165	code = doScanWalAsync(pTq, ckPause);	485,540✔
166	streamMetaWUnLock(pMeta);	485,514✔
167	return code;	485,547✔
168	}
169
170	int32_t tqStopStreamTasksAsync(STQ* pTq) {	4,678✔
171	SStreamMeta* pMeta = pTq->pStreamMeta;	4,678✔
172	int32_t vgId = pMeta->vgId;	4,678✔
173	return streamTaskSchedTask(&pTq->pVnode->msgCb, vgId, 0, 0, STREAM_EXEC_T_STOP_ALL_TASKS);	4,678✔
174	}
175
176	int32_t setWalReaderStartOffset(SStreamTask* pTask, int32_t vgId) {	73,179✔
177	// seek the stored version and extract data from WAL
178	int64_t firstVer = walReaderGetValidFirstVer(pTask->exec.pWalReader);	73,179✔
179	if (pTask->chkInfo.nextProcessVer < firstVer) {	73,203✔
180	tqWarn("vgId:%d s-task:%s ver:%" PRId64 " earlier than the first ver of wal range %" PRId64 ", forward to %" PRId64,	36!
181	vgId, pTask->id.idStr, pTask->chkInfo.nextProcessVer, firstVer, firstVer);
182
183	pTask->chkInfo.nextProcessVer = firstVer;	36✔
184
185	// todo need retry if failed
186	int32_t code = walReaderSeekVer(pTask->exec.pWalReader, pTask->chkInfo.nextProcessVer);	36✔
187	if (code != TSDB_CODE_SUCCESS) {	36!
188	return code;	×
189	}
190
191	// append the data for the stream
192	tqDebug("vgId:%d s-task:%s wal reader seek to ver:%" PRId64, vgId, pTask->id.idStr, pTask->chkInfo.nextProcessVer);	36!
193	} else {
194	int64_t currentVer = walReaderGetCurrentVer(pTask->exec.pWalReader);	73,167✔
195	if (currentVer == -1) { // we only seek the read for the first time	73,140✔
196	int32_t code = walReaderSeekVer(pTask->exec.pWalReader, pTask->chkInfo.nextProcessVer);	4,783✔
197	if (code != TSDB_CODE_SUCCESS) { // no data in wal, quit	4,783!
198	return code;	×
199	}
200
201	// append the data for the stream
202	tqDebug("vgId:%d s-task:%s wal reader initial seek to ver:%" PRId64, vgId, pTask->id.idStr,	4,783✔
203	pTask->chkInfo.nextProcessVer);
204	}
205	}
206
207	int64_t skipToVer = walReaderGetSkipToVersion(pTask->exec.pWalReader);	73,176✔
208	if (skipToVer != 0 && skipToVer > pTask->chkInfo.nextProcessVer) {	73,166✔
209	int32_t code = walReaderSeekVer(pTask->exec.pWalReader, skipToVer);	442✔
210	if (code != TSDB_CODE_SUCCESS) { // no data in wal, quit	442!
211	return code;	×
212	}
213
214	tqDebug("vgId:%d s-task:%s wal reader jump to ver:%" PRId64, vgId, pTask->id.idStr, skipToVer);	442✔
215	}
216
217	return TSDB_CODE_SUCCESS;	73,153✔
218	}
219
220	// todo handle memory error
221	bool handleFillhistoryScanComplete(SStreamTask* pTask, int64_t ver) {	607,624✔
222	const char* id = pTask->id.idStr;	607,624✔
223	int64_t maxVer = pTask->step2Range.maxVer;	607,624✔
224
225	if ((pTask->info.fillHistory == 1) && ver > maxVer) {	607,624✔
226	if (!pTask->status.appendTranstateBlock) {	792!
227	qWarn("s-task:%s fill-history scan WAL, nextProcessVer:%" PRId64 " out of the maximum ver:%" PRId64	792!
228	", not scan wal anymore, add transfer-state block into inputQ",
229	id, ver, maxVer);
230
231	double el = (taosGetTimestampMs() - pTask->execInfo.step2Start) / 1000.0;	792✔
232	qDebug("s-task:%s scan-history from WAL stage(step 2) ended, range:%" PRId64 "-%" PRId64 ", elapsed time:%.2fs",	792✔
233	id, pTask->step2Range.minVer, maxVer, el);
234	int32_t code = streamTaskPutTranstateIntoInputQ(pTask);	792✔
235	if (code) {	792!
236	qError("s-task:%s failed to put trans-state into inputQ", id);	×
237	}
238
239	return true;	792✔
240	} else {
241	qWarn("s-task:%s fill-history scan WAL, nextProcessVer:%" PRId64 " out of the ver range:%" PRId64 "-%" PRId64	×
242	", not scan wal",
243	id, ver, pTask->step2Range.minVer, maxVer);
244	}
245	}
246
247	return false;	606,838✔
248	}
249
250	bool taskReadyForDataFromWal(SStreamTask* pTask) {	280,917✔
251	// non-source or fill-history tasks don't need to response the WAL scan action.
252	SSTaskBasicInfo* pInfo = &pTask->info;	280,917✔
253	if ((pInfo->taskLevel != TASK_LEVEL__SOURCE) \|\| (pTask->status.downstreamReady == 0)) {	280,917✔
254	return false;	156,216✔
255	}
256
257	if (pInfo->taskLevel == TASK_LEVEL__SOURCE && pInfo->trigger == STREAM_TRIGGER_FORCE_WINDOW_CLOSE) {	124,701!
258	return false;	12,888✔
259	}
260
261	// not in ready state, do not handle the data from wal
262	SStreamTaskState pState = streamTaskGetStatus(pTask);	111,813✔
263	if (pState.state != TASK_STATUS__READY) {	111,804✔
264	tqTrace("s-task:%s not ready for submit block in wal, status:%s", pTask->id.idStr, pState.name);	32,973✔
265	return false;	32,974✔
266	}
267
268	// fill-history task has entered into the last phase, no need to anything
269	if ((pInfo->fillHistory == 1) && pTask->status.appendTranstateBlock) {	78,831✔
270	// the maximum version of data in the WAL has reached already, the step2 is done
271	tqDebug("s-task:%s fill-history reach the maximum ver:%" PRId64 ", not scan wal anymore", pTask->id.idStr,	5,628✔
272	pTask->dataRange.range.maxVer);
273	return false;	5,628✔
274	}
275
276	// check if input queue is full or not
277	if (streamQueueIsFull(pTask->inputq.queue)) {	73,203✔
278	tqTrace("s-task:%s input queue is full, do nothing", pTask->id.idStr);	30!
279	return false;	×
280	}
281
282	// the input queue of downstream task is full, so the output is blocked, stopped for a while
283	if (pTask->inputq.status == TASK_INPUT_STATUS__BLOCKED) {	73,192!
284	tqDebug("s-task:%s inputQ is blocked, do nothing", pTask->id.idStr);	×
285	return false;	×
286	}
287
288	return true;	73,192✔
289	}
290
291	int32_t doPutDataIntoInputQ(SStreamTask* pTask, int64_t maxVer, int32_t* numOfItems, bool* pSucc) {	73,182✔
292	const char* id = pTask->id.idStr;	73,182✔
293	int32_t numOfNewItems = 0;	73,182✔
294	int32_t code = 0;	73,182✔
295	*pSucc = false;	73,182✔
296
297	while (1) {	534,522✔
298	if ((pTask->info.fillHistory == 1) && pTask->status.appendTranstateBlock) {	607,704!
299	*numOfItems += numOfNewItems;	×
300	return numOfNewItems > 0;	×
301	}
302
303	SStreamQueueItem* pItem = NULL;	607,704✔
304	code = extractMsgFromWal(pTask->exec.pWalReader, (void**)&pItem, maxVer, id);	607,704✔
305	if (code != TSDB_CODE_SUCCESS \|\| pItem == NULL) { // failed, continue	607,620✔
306	int64_t currentVer = walReaderGetCurrentVer(pTask->exec.pWalReader);	72,957✔
307	bool itemInFillhistory = handleFillhistoryScanComplete(pTask, currentVer);	72,978✔
308	if (itemInFillhistory) {	72,990✔
309	numOfNewItems += 1;	657✔
310	}
311	break;	72,990✔
312	}
313
314	if (pItem != NULL) {	534,663!
315	code = streamTaskPutDataIntoInputQ(pTask, pItem);	534,663✔
316	if (code == TSDB_CODE_SUCCESS) {	534,684✔
317	numOfNewItems += 1;	534,662✔
318	int64_t ver = walReaderGetCurrentVer(pTask->exec.pWalReader);	534,662✔
319	pTask->chkInfo.nextProcessVer = ver;	534,659✔
320	tqDebug("s-task:%s set ver:%" PRId64 " for reader after extract data from WAL", id, ver);	534,659✔
321
322	bool itemInFillhistory = handleFillhistoryScanComplete(pTask, ver);	534,659✔
323	if (itemInFillhistory) {	534,657✔
324	break;	135✔
325	}
326	} else {
327	if (code == TSDB_CODE_OUT_OF_MEMORY) {	22!
328	tqError("s-task:%s failed to put data into inputQ, since out of memory", id);	×
329	} else {
330	tqTrace("s-task:%s append input queue failed, code:inputQ is full, ver:%" PRId64, id,	22!
331	pTask->chkInfo.nextProcessVer);
332	code = walReaderSeekVer(pTask->exec.pWalReader, pTask->chkInfo.nextProcessVer);	22✔
333	if (code) {	22!
334	tqError("s-task:%s failed to seek ver to:%" PRId64 " in wal", id, pTask->chkInfo.nextProcessVer);	×
335	}
336
337	code = 0; // reset the error code	22✔
338	}
339
340	break;	22✔
341	}
342	}
343	}
344
345	*numOfItems += numOfNewItems;	73,147✔
346	*pSucc = (numOfNewItems > 0);	73,147✔
347	return code;	73,147✔
348	}
349
350	int32_t doScanWalForAllTasks(SStreamMeta* pStreamMeta) {	73,303✔
351	int32_t vgId = pStreamMeta->vgId;	73,303✔
352	SArray* pTaskList = NULL;	73,303✔
353	int32_t numOfTasks = taosArrayGetSize(pStreamMeta->pTaskList);	73,303✔
354	if (numOfTasks == 0) {	73,311✔
355	return TSDB_CODE_SUCCESS;	600✔
356	}
357
358	// clone the task list, to avoid the task update during scan wal files
359	streamMetaWLock(pStreamMeta);	72,711✔
360	pTaskList = taosArrayDup(pStreamMeta->pTaskList, NULL);	72,715✔
361	streamMetaWUnLock(pStreamMeta);	72,716✔
362	if (pTaskList == NULL) {	72,718!
363	tqError("vgId:%d failed to create task list dup, code:%s", vgId, tstrerror(terrno));	×
364	return terrno;	×
365	}
366
367	tqDebug("vgId:%d start to check wal to extract new submit block for %d tasks", vgId, numOfTasks);	72,718✔
368
369	// update the new task number
370	numOfTasks = taosArrayGetSize(pTaskList);	72,718✔
371
372	for (int32_t i = 0; i < numOfTasks; ++i) {	354,431✔
373	STaskId* pTaskId = taosArrayGet(pTaskList, i);	281,680✔
374	if (pTaskId == NULL) {	281,654!
375	continue;	208,520✔
376	}
377
378	SStreamTask* pTask = NULL;	281,654✔
379	int32_t code = streamMetaAcquireTask(pStreamMeta, pTaskId->streamId, pTaskId->taskId, &pTask);	281,654✔
380	if (pTask == NULL \|\| code != 0) {	281,695✔
381	continue;	738✔
382	}
383
384	if (!taskReadyForDataFromWal(pTask)) {	280,957✔
385	streamMetaReleaseTask(pStreamMeta, pTask);	207,707✔
386	continue;	207,792✔
387	}
388
389	// seek the stored version and extract data from WAL
390	code = setWalReaderStartOffset(pTask, vgId);	73,171✔
391	if (code != TSDB_CODE_SUCCESS) {	73,151!
392	streamMetaReleaseTask(pStreamMeta, pTask);	×
393	continue;	×
394	}
395
396	int32_t numOfItems = streamQueueGetNumOfItems(pTask->inputq.queue);	73,151✔
397	int64_t maxVer = (pTask->info.fillHistory == 1) ? pTask->step2Range.maxVer : INT64_MAX;	73,213✔
398
399	streamMutexLock(&pTask->lock);	73,213✔
400
401	SStreamTaskState state = streamTaskGetStatus(pTask);	73,208✔
402	if (state.state != TASK_STATUS__READY) {	73,193✔
403	tqDebug("s-task:%s not ready for submit block from wal, status:%s", pTask->id.idStr, state.name);	16✔
404	streamMutexUnlock(&pTask->lock);	16✔
405	streamMetaReleaseTask(pStreamMeta, pTask);	16✔
406	continue;	×
407	}
408
409	bool hasNewData = false;	73,177✔
410	code = doPutDataIntoInputQ(pTask, maxVer, &numOfItems, &hasNewData);	73,177✔
411	streamMutexUnlock(&pTask->lock);	73,152✔
412
413	if ((numOfItems > 0) \|\| hasNewData) {	73,202!
414	code = streamTrySchedExec(pTask);	24,761✔
415	if (code != TSDB_CODE_SUCCESS) {	24,765✔
416	streamMetaReleaseTask(pStreamMeta, pTask);	1✔
417	taosArrayDestroy(pTaskList);	1✔
418	return code;	1✔
419	}
420	}
421
422	streamMetaReleaseTask(pStreamMeta, pTask);	73,205✔
423	}
424
425	taosArrayDestroy(pTaskList);	72,751✔
426	return TSDB_CODE_SUCCESS;	72,710✔
427	}
428
429	int32_t doScanWalAsync(STQ* pTq, bool ckPause) {	485,452✔
430	SStreamMeta* pMeta = pTq->pStreamMeta;	485,452✔
431	bool alreadyRestored = pTq->pVnode->restored;	485,452✔
432	int32_t vgId = pMeta->vgId;	485,452✔
433	int32_t numOfTasks = taosArrayGetSize(pMeta->pTaskList);	485,452✔
434
435	if (numOfTasks == 0) {	485,509✔
436	tqDebug("vgId:%d no stream tasks existed to run", vgId);	3!
437	return 0;	10✔
438	}
439
440	if (pMeta->startInfo.startAllTasks) {	485,506✔
441	tqTrace("vgId:%d in restart procedure, not scan wal", vgId);	12,198✔
442	return 0;	12,198✔
443	}
444
445	pMeta->scanInfo.scanCounter += 1;	473,308✔
446	if (pMeta->scanInfo.scanCounter > MAX_REPEAT_SCAN_THRESHOLD) {	473,308✔
447	pMeta->scanInfo.scanCounter = MAX_REPEAT_SCAN_THRESHOLD;	397,654✔
448	}
449
450	if (pMeta->scanInfo.scanCounter > 1) {	473,308✔
451	tqDebug("vgId:%d wal read task has been launched, remain scan times:%d", vgId, pMeta->scanInfo.scanCounter);	421,152✔
452	return 0;	421,157✔
453	}
454
455	int32_t numOfPauseTasks = pMeta->numOfPausedTasks;	52,156✔
456	if (ckPause && numOfTasks == numOfPauseTasks) {	52,156✔
457	tqDebug("vgId:%d ignore all submit, all streams had been paused, reset the walScanCounter", vgId);	113✔
458
459	// reset the counter value, since we do not launch the scan wal operation.
460	pMeta->scanInfo.scanCounter = 0;	113✔
461	return 0;	113✔
462	}
463
464	tqDebug("vgId:%d create msg to start wal scan to launch stream tasks, numOfTasks:%d, vnd restored:%d", vgId,	52,043✔
465	numOfTasks, alreadyRestored);
466
467	return streamTaskSchedTask(&pTq->pVnode->msgCb, vgId, 0, 0, STREAM_EXEC_T_EXTRACT_WAL_DATA);	52,043✔
468	}
469
470	void streamMetaFreeTQDuringScanWalError(STQ* pTq) {	×
471	SBuildScanWalMsgParam* p = taosMemoryCalloc(1, sizeof(SBuildScanWalMsgParam));	×
472	p->metaId = pTq->pStreamMeta->rid;	×
473	p->numOfTasks = 0;	×
474
475	doStartScanWal(p, 0);	×
476	}	×

taosdata / TDengine / #3612

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous