#3534

Committed 21 Nov 2024 07:36AM UTC coverage: 60.825% (+2.0%) from 58.848%

Build # #3534

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #28810 from taosdata/ehn/add-sync-heartbeat-sent-time-to-log

ehn:add-sync-heartbeat-sent-time-to-log

Run Details

120023 of 252376 branches covered (47.56%)

Branch coverage included in aggregate %.

43 of 47 new or added lines in 3 files covered. (91.49%)

2254 existing lines in 162 files now uncovered.

200876 of 275203 relevant lines covered (72.99%)

16110754.39 hits per line

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

80.62

/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    100

typedef struct SBuildScanWalMsgParam {
  int64_t metaId;
  int32_t numOfTasks;
} SBuildScanWalMsgParam;

static int32_t doScanWalForAllTasks(SStreamMeta* pStreamMeta, bool* pScanIdle);
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);

// 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;
  bool         shouldIdle = true;

  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, &shouldIdle);
  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;
  STQ*    pTq = NULL;
  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;
  }

  vgId = pMeta->vgId;
  pTq = pMeta->ahandle;

  tqDebug("vgId:%d create msg to start wal scan, numOfTasks:%d, vnd restored:%d", vgId, pParam->numOfTasks,
          pTq->pVnode->restored);

  code = streamTaskSchedTask(&pTq->pVnode->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;

  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) {
  int32_t      vgId = TD_VID(pTq->pVnode);
  SStreamMeta* pMeta = pTq->pStreamMeta;
  bool         alreadyRestored = pTq->pVnode->restored;
  int32_t      numOfTasks = 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);

  numOfTasks = taosArrayGetSize(pMeta->pTaskList);
  if (numOfTasks == 0) {
    tqDebug("vgId:%d no stream tasks existed to run", vgId);
    streamMetaWUnLock(pMeta);
    return 0;
  }

  if (pMeta->startInfo.startAllTasks) {
    tqTrace("vgId:%d in restart procedure, not scan wal", vgId);
    streamMetaWUnLock(pMeta);
    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);
    streamMetaWUnLock(pMeta);
    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;
    streamMetaWUnLock(pMeta);
    return 0;
  }

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

  int32_t code = streamTaskSchedTask(&pTq->pVnode->msgCb, vgId, 0, 0, STREAM_EXEC_T_EXTRACT_WAL_DATA);
  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, bool* pScanIdle) {
  *pScanIdle = true;
  bool    noDataInWal = true;
  int32_t vgId = pStreamMeta->vgId;

  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
  SArray* pTaskList = NULL;
  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;
    }

    *pScanIdle = false;

    // 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) {
      noDataInWal = false;
      code = streamTrySchedExec(pTask);
      if (code != TSDB_CODE_SUCCESS) {
        streamMetaReleaseTask(pStreamMeta, pTask);
        taosArrayDestroy(pTaskList);
        return code;
      }
    }

    streamMetaReleaseTask(pStreamMeta, pTask);
  }

  // all wal are checked, and no new data available in wal.
  if (noDataInWal) {
    *pScanIdle = true;
  }

  taosArrayDestroy(pTaskList);
  return TSDB_CODE_SUCCESS;
}

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 100
21
22	typedef struct SBuildScanWalMsgParam {
23	int64_t metaId;
24	int32_t numOfTasks;
25	} SBuildScanWalMsgParam;
26
27	static int32_t doScanWalForAllTasks(SStreamMeta* pStreamMeta, bool* pScanIdle);
28	static int32_t setWalReaderStartOffset(SStreamTask* pTask, int32_t vgId);
29	static bool handleFillhistoryScanComplete(SStreamTask* pTask, int64_t ver);
30	static bool taskReadyForDataFromWal(SStreamTask* pTask);
31	static int32_t doPutDataIntoInputQ(SStreamTask* pTask, int64_t maxVer, int32_t* numOfItems, bool* pSucc);
32	static int32_t tqScanWalInFuture(STQ* pTq, int32_t numOfTasks, int32_t idleDuration);
33
34	// extract data blocks(submit/delete) from WAL, and add them into the input queue for all the sources tasks.
35	int32_t tqScanWal(STQ* pTq) {	104,999✔
36	SStreamMeta* pMeta = pTq->pStreamMeta;	104,999✔
37	int32_t vgId = pMeta->vgId;	104,999✔
38	int64_t st = taosGetTimestampMs();	105,030✔
39	int32_t numOfTasks = 0;	105,030✔
40	bool shouldIdle = true;	105,030✔
41
42	tqDebug("vgId:%d continue to check if data in wal are available, scanCounter:%d", vgId, pMeta->scanInfo.scanCounter);	105,030✔
43
44	// check all tasks
45	int32_t code = doScanWalForAllTasks(pMeta, &shouldIdle);	105,033✔
46	if (code) {	105,011!
47	tqError("vgId:%d failed to start all tasks, try next time, code:%s", vgId, tstrerror(code));	×
48	return code;	×
49	}
50
51	streamMetaWLock(pMeta);	105,011✔
52	int32_t times = (--pMeta->scanInfo.scanCounter);	105,017✔
53	if (times < 0) {	105,017!
54	tqError("vgId:%d invalid scan counter:%d, reset to 0", vgId, times);	×
55	times = 0;	×
56	}
57
58	numOfTasks = taosArrayGetSize(pMeta->pTaskList);	105,017✔
59	streamMetaWUnLock(pMeta);	105,009✔
60
61	int64_t el = (taosGetTimestampMs() - st);	105,025✔
62	tqDebug("vgId:%d scan wal for stream tasks completed, elapsed time:%" PRId64 " ms", vgId, el);	105,025✔
63
64	if (times > 0) {	105,025✔
65	tqDebug("vgId:%d scan wal for stream tasks for %d times in %dms", vgId, times, SCAN_WAL_IDLE_DURATION);	37,432✔
66	code = tqScanWalInFuture(pTq, numOfTasks, SCAN_WAL_IDLE_DURATION);	37,432✔
67	if (code) {	37,442!
68	tqError("vgId:%d sched scan wal in %dms failed, ignore this failure", vgId, SCAN_WAL_IDLE_DURATION);	×
69	}
70	}
71
72	return code;	105,018✔
73	}
74
75	static void doStartScanWal(void* param, void* tmrId) {	37,442✔
76	int32_t vgId = 0;	37,442✔
77	STQ* pTq = NULL;	37,442✔
78	int32_t code = 0;	37,442✔
79
80	SBuildScanWalMsgParam* pParam = (SBuildScanWalMsgParam*)param;	37,442✔
81
82	SStreamMeta* pMeta = taosAcquireRef(streamMetaRefPool, pParam->metaId);	37,442✔
83	if (pMeta == NULL) {	37,442✔
84	tqError("metaRid:%" PRId64 " not valid now, stream meta has been freed", pParam->metaId);	1!
85	taosMemoryFree(pParam);	1✔
86	return;	1✔
87	}
88
89	vgId = pMeta->vgId;	37,441✔
90	pTq = pMeta->ahandle;	37,441✔
91
92	tqDebug("vgId:%d create msg to start wal scan, numOfTasks:%d, vnd restored:%d", vgId, pParam->numOfTasks,	37,441✔
93	pTq->pVnode->restored);
94
95	code = streamTaskSchedTask(&pTq->pVnode->msgCb, vgId, 0, 0, STREAM_EXEC_T_EXTRACT_WAL_DATA);	37,441✔
96	if (code) {	37,441✔
97	tqError("vgId:%d failed sched task to scan wal, code:%s", vgId, tstrerror(code));	202!
98	}
99
100	code = taosReleaseRef(streamMetaRefPool, pParam->metaId);	37,441✔
101	if (code) {	37,441!
102	tqError("vgId:%d failed to release ref for streamMeta, rid:%" PRId64 " code:%s", vgId, pParam->metaId,	×
103	tstrerror(code));
104	}
105
106	taosMemoryFree(pParam);	37,441✔
107	}
108
109	int32_t tqScanWalInFuture(STQ* pTq, int32_t numOfTasks, int32_t idleDuration) {	37,419✔
110	SStreamMeta* pMeta = pTq->pStreamMeta;	37,419✔
111	int32_t code = 0;	37,419✔
112	int32_t vgId = TD_VID(pTq->pVnode);	37,419✔
113	tmr_h pTimer = NULL;	37,419✔
114	SBuildScanWalMsgParam* pParam = NULL;	37,419✔
115
116	pParam = taosMemoryMalloc(sizeof(SBuildScanWalMsgParam));	37,419✔
117	if (pParam == NULL) {	37,434!
118	return terrno;	×
119	}
120
121	pParam->metaId = pMeta->rid;	37,434✔
122	pParam->numOfTasks = numOfTasks;	37,434✔
123
124	code = streamTimerGetInstance(&pTimer);	37,434✔
125	if (code) {	37,425!
126	tqError("vgId:%d failed to get tmr ctrl during sched scan wal", vgId);	×
127	taosMemoryFree(pParam);	×
128	} else {
129	streamTmrStart(doStartScanWal, idleDuration, pParam, pTimer, &pMeta->scanInfo.scanTimer, vgId, "scan-wal-fut");	37,425✔
130	}
131
132	return code;	37,442✔
133	}
134
135	int32_t tqScanWalAsync(STQ* pTq, bool ckPause) {	557,304✔
136	int32_t vgId = TD_VID(pTq->pVnode);	557,304✔
137	SStreamMeta* pMeta = pTq->pStreamMeta;	557,304✔
138	bool alreadyRestored = pTq->pVnode->restored;	557,304✔
139	int32_t numOfTasks = 0;	557,304✔
140
141	// do not launch the stream tasks, if it is a follower or not restored vnode.
142	if (!(vnodeIsRoleLeader(pTq->pVnode) && alreadyRestored)) {	557,304✔
143	return TSDB_CODE_SUCCESS;	46,246✔
144	}
145
146	streamMetaWLock(pMeta);	511,124✔
147
148	numOfTasks = taosArrayGetSize(pMeta->pTaskList);	511,072✔
149	if (numOfTasks == 0) {	511,089✔
150	tqDebug("vgId:%d no stream tasks existed to run", vgId);	9!
151	streamMetaWUnLock(pMeta);	9✔
152	return 0;	9✔
153	}
154
155	if (pMeta->startInfo.startAllTasks) {	511,080✔
156	tqTrace("vgId:%d in restart procedure, not scan wal", vgId);	17,481✔
157	streamMetaWUnLock(pMeta);	17,481✔
158	return 0;	17,481✔
159	}
160
161	pMeta->scanInfo.scanCounter += 1;	493,599✔
162	if (pMeta->scanInfo.scanCounter > MAX_REPEAT_SCAN_THRESHOLD) {	493,599✔
163	pMeta->scanInfo.scanCounter = MAX_REPEAT_SCAN_THRESHOLD;	387,370✔
164	}
165
166	if (pMeta->scanInfo.scanCounter > 1) {	493,599✔
167	tqDebug("vgId:%d wal read task has been launched, remain scan times:%d", vgId, pMeta->scanInfo.scanCounter);	425,147✔
168	streamMetaWUnLock(pMeta);	425,149✔
169	return 0;	425,164✔
170	}
171
172	int32_t numOfPauseTasks = pMeta->numOfPausedTasks;	68,452✔
173	if (ckPause && numOfTasks == numOfPauseTasks) {	68,452✔
174	tqDebug("vgId:%d ignore all submit, all streams had been paused, reset the walScanCounter", vgId);	125✔
175
176	// reset the counter value, since we do not launch the scan wal operation.
177	pMeta->scanInfo.scanCounter = 0;	125✔
178	streamMetaWUnLock(pMeta);	125✔
179	return 0;	125✔
180	}
181
182	tqDebug("vgId:%d create msg to start wal scan to launch stream tasks, numOfTasks:%d, vnd restored:%d", vgId,	68,327✔
183	numOfTasks, alreadyRestored);
184
185	int32_t code = streamTaskSchedTask(&pTq->pVnode->msgCb, vgId, 0, 0, STREAM_EXEC_T_EXTRACT_WAL_DATA);	68,327✔
186	streamMetaWUnLock(pMeta);	68,337✔
187
188	return code;	68,343✔
189	}
190
191	int32_t tqStopStreamTasksAsync(STQ* pTq) {	4,877✔
192	SStreamMeta* pMeta = pTq->pStreamMeta;	4,877✔
193	int32_t vgId = pMeta->vgId;	4,877✔
194	return streamTaskSchedTask(&pTq->pVnode->msgCb, vgId, 0, 0, STREAM_EXEC_T_STOP_ALL_TASKS);	4,877✔
195	}
196
197	int32_t setWalReaderStartOffset(SStreamTask* pTask, int32_t vgId) {	120,956✔
198	// seek the stored version and extract data from WAL
199	int64_t firstVer = walReaderGetValidFirstVer(pTask->exec.pWalReader);	120,956✔
200	if (pTask->chkInfo.nextProcessVer < firstVer) {	120,981✔
201	tqWarn("vgId:%d s-task:%s ver:%" PRId64 " earlier than the first ver of wal range %" PRId64 ", forward to %" PRId64,	18!
202	vgId, pTask->id.idStr, pTask->chkInfo.nextProcessVer, firstVer, firstVer);
203
204	pTask->chkInfo.nextProcessVer = firstVer;	18✔
205
206	// todo need retry if failed
207	int32_t code = walReaderSeekVer(pTask->exec.pWalReader, pTask->chkInfo.nextProcessVer);	18✔
208	if (code != TSDB_CODE_SUCCESS) {	18!
209	return code;	×
210	}
211
212	// append the data for the stream
213	tqDebug("vgId:%d s-task:%s wal reader seek to ver:%" PRId64, vgId, pTask->id.idStr, pTask->chkInfo.nextProcessVer);	18✔
214	} else {
215	int64_t currentVer = walReaderGetCurrentVer(pTask->exec.pWalReader);	120,963✔
216	if (currentVer == -1) { // we only seek the read for the first time	120,945✔
217	int32_t code = walReaderSeekVer(pTask->exec.pWalReader, pTask->chkInfo.nextProcessVer);	5,509✔
218	if (code != TSDB_CODE_SUCCESS) { // no data in wal, quit	5,506!
219	return code;	×
220	}
221
222	// append the data for the stream
223	tqDebug("vgId:%d s-task:%s wal reader initial seek to ver:%" PRId64, vgId, pTask->id.idStr,	5,506✔
224	pTask->chkInfo.nextProcessVer);
225	}
226	}
227
228	int64_t skipToVer = walReaderGetSkipToVersion(pTask->exec.pWalReader);	120,960✔
229	if (skipToVer != 0 && skipToVer > pTask->chkInfo.nextProcessVer) {	120,939✔
230	int32_t code = walReaderSeekVer(pTask->exec.pWalReader, skipToVer);	295✔
231	if (code != TSDB_CODE_SUCCESS) { // no data in wal, quit	295!
232	return code;	×
233	}
234
235	tqDebug("vgId:%d s-task:%s wal reader jump to ver:%" PRId64, vgId, pTask->id.idStr, skipToVer);	295✔
236	}
237
238	return TSDB_CODE_SUCCESS;	120,928✔
239	}
240
241	// todo handle memory error
242	bool handleFillhistoryScanComplete(SStreamTask* pTask, int64_t ver) {	576,640✔
243	const char* id = pTask->id.idStr;	576,640✔
244	int64_t maxVer = pTask->step2Range.maxVer;	576,640✔
245
246	if ((pTask->info.fillHistory == 1) && ver > maxVer) {	576,640✔
247	if (!pTask->status.appendTranstateBlock) {	1,504!
248	qWarn("s-task:%s fill-history scan WAL, nextProcessVer:%" PRId64 " out of the maximum ver:%" PRId64	1,504!
249	", not scan wal anymore, add transfer-state block into inputQ",
250	id, ver, maxVer);
251
252	double el = (taosGetTimestampMs() - pTask->execInfo.step2Start) / 1000.0;	1,504✔
253	qDebug("s-task:%s scan-history from WAL stage(step 2) ended, range:%" PRId64 "-%" PRId64 ", elapsed time:%.2fs",	1,504✔
254	id, pTask->step2Range.minVer, maxVer, el);
255	int32_t code = streamTaskPutTranstateIntoInputQ(pTask);	1,504✔
256	if (code) {	1,504!
257	qError("s-task:%s failed to put trans-state into inputQ", id);	×
258	}
259
260	return true;	1,504✔
261	} else {
262	qWarn("s-task:%s fill-history scan WAL, nextProcessVer:%" PRId64 " out of the ver range:%" PRId64 "-%" PRId64	×
263	", not scan wal",
264	id, ver, pTask->step2Range.minVer, maxVer);
265	}
266	}
267
268	return false;	575,121✔
269	}
270
271	bool taskReadyForDataFromWal(SStreamTask* pTask) {	422,772✔
272	// non-source or fill-history tasks don't need to response the WAL scan action.
273	SSTaskBasicInfo* pInfo = &pTask->info;	422,772✔
274	if ((pInfo->taskLevel != TASK_LEVEL__SOURCE) \|\| (pTask->status.downstreamReady == 0)) {	422,772✔
275	return false;	224,331✔
276	}
277
278	if (pInfo->taskLevel == TASK_LEVEL__SOURCE && pInfo->trigger == STREAM_TRIGGER_FORCE_WINDOW_CLOSE) {	198,441!
279	return false;	2,248✔
280	}
281
282	// not in ready state, do not handle the data from wal
283	SStreamTaskState pState = streamTaskGetStatus(pTask);	196,193✔
284	if (pState.state != TASK_STATUS__READY) {	196,173✔
285	tqTrace("s-task:%s not ready for submit block in wal, status:%s", pTask->id.idStr, pState.name);	59,607✔
286	return false;	59,607✔
287	}
288
289	// fill-history task has entered into the last phase, no need to anything
290	if ((pInfo->fillHistory == 1) && pTask->status.appendTranstateBlock) {	136,566✔
291	// the maximum version of data in the WAL has reached already, the step2 is done
292	tqDebug("s-task:%s fill-history reach the maximum ver:%" PRId64 ", not scan wal anymore", pTask->id.idStr,	15,582✔
293	pTask->dataRange.range.maxVer);
294	return false;	15,583✔
295	}
296
297	// check if input queue is full or not
298	if (streamQueueIsFull(pTask->inputq.queue)) {	120,984!
UNCOV 299	tqTrace("s-task:%s input queue is full, do nothing", pTask->id.idStr);	×
UNCOV 300	return false;	×
301	}
302
303	// the input queue of downstream task is full, so the output is blocked, stopped for a while
304	if (pTask->inputq.status == TASK_INPUT_STATUS__BLOCKED) {	121,019!
305	tqDebug("s-task:%s inputQ is blocked, do nothing", pTask->id.idStr);	×
306	return false;	×
307	}
308
309	return true;	121,019✔
310	}
311
312	int32_t doPutDataIntoInputQ(SStreamTask* pTask, int64_t maxVer, int32_t* numOfItems, bool* pSucc) {	120,938✔
313	const char* id = pTask->id.idStr;	120,938✔
314	int32_t numOfNewItems = 0;	120,938✔
315	int32_t code = 0;	120,938✔
316	*pSucc = false;	120,938✔
317
318	while (1) {	455,784✔
319	if ((pTask->info.fillHistory == 1) && pTask->status.appendTranstateBlock) {	576,722!
320	*numOfItems += numOfNewItems;	×
321	return numOfNewItems > 0;	×
322	}
323
324	SStreamQueueItem* pItem = NULL;	576,722✔
325	code = extractMsgFromWal(pTask->exec.pWalReader, (void**)&pItem, maxVer, id);	576,722✔
326	if (code != TSDB_CODE_SUCCESS \|\| pItem == NULL) { // failed, continue	576,680✔
327	int64_t currentVer = walReaderGetCurrentVer(pTask->exec.pWalReader);	120,625✔
328	bool itemInFillhistory = handleFillhistoryScanComplete(pTask, currentVer);	120,653✔
329	if (itemInFillhistory) {	120,617✔
330	numOfNewItems += 1;	1,259✔
331	}
332	break;	120,617✔
333	}
334
335	if (pItem != NULL) {	456,055✔
336	code = streamTaskPutDataIntoInputQ(pTask, pItem);	456,054✔
337	if (code == TSDB_CODE_SUCCESS) {	456,062✔
338	numOfNewItems += 1;	456,060✔
339	int64_t ver = walReaderGetCurrentVer(pTask->exec.pWalReader);	456,060✔
340	pTask->chkInfo.nextProcessVer = ver;	456,042✔
341	tqDebug("s-task:%s set ver:%" PRId64 " for reader after extract data from WAL", id, ver);	456,042✔
342
343	bool itemInFillhistory = handleFillhistoryScanComplete(pTask, ver);	456,042✔
344	if (itemInFillhistory) {	456,028✔
345	break;	245✔
346	}
347	} else {
348	if (code == TSDB_CODE_OUT_OF_MEMORY) {	2!
349	tqError("s-task:%s failed to put data into inputQ, since out of memory", id);	×
350	} else {
351	tqTrace("s-task:%s append input queue failed, code:inputQ is full, ver:%" PRId64, id,	2!
352	pTask->chkInfo.nextProcessVer);
353	code = walReaderSeekVer(pTask->exec.pWalReader, pTask->chkInfo.nextProcessVer);	2✔
354	if (code) {	2!
355	tqError("s-task:%s failed to seek ver to:%" PRId64 " in wal", id, pTask->chkInfo.nextProcessVer);	×
356	}
357
358	code = 0; // reset the error code	2✔
359	}
360
361	break;	2✔
362	}
363	}
364	}
365
366	*numOfItems += numOfNewItems;	120,864✔
367	*pSucc = (numOfNewItems > 0);	120,864✔
368	return code;	120,864✔
369	}
370
371	int32_t doScanWalForAllTasks(SStreamMeta* pStreamMeta, bool* pScanIdle) {	105,011✔
372	*pScanIdle = true;	105,011✔
373	bool noDataInWal = true;	105,011✔
374	int32_t vgId = pStreamMeta->vgId;	105,011✔
375
376	int32_t numOfTasks = taosArrayGetSize(pStreamMeta->pTaskList);	105,011✔
377	if (numOfTasks == 0) {	105,030✔
378	return TSDB_CODE_SUCCESS;	907✔
379	}
380
381	// clone the task list, to avoid the task update during scan wal files
382	SArray* pTaskList = NULL;	104,123✔
383	streamMetaWLock(pStreamMeta);	104,123✔
384	pTaskList = taosArrayDup(pStreamMeta->pTaskList, NULL);	104,113✔
385	streamMetaWUnLock(pStreamMeta);	104,122✔
386	if (pTaskList == NULL) {	104,089!
387	tqError("vgId:%d failed to create task list dup, code:%s", vgId, tstrerror(terrno));	×
388	return terrno;	×
389	}
390
391	tqDebug("vgId:%d start to check wal to extract new submit block for %d tasks", vgId, numOfTasks);	104,089✔
392
393	// update the new task number
394	numOfTasks = taosArrayGetSize(pTaskList);	104,089✔
395
396	for (int32_t i = 0; i < numOfTasks; ++i) {	527,666✔
397	STaskId* pTaskId = taosArrayGet(pTaskList, i);	423,485✔
398	if (pTaskId == NULL) {	423,393!
399	continue;	302,553✔
400	}
401
402	SStreamTask* pTask = NULL;	423,393✔
403	int32_t code = streamMetaAcquireTask(pStreamMeta, pTaskId->streamId, pTaskId->taskId, &pTask);	423,393✔
404	if (pTask == NULL \|\| code != 0) {	423,412!
405	continue;	557✔
406	}
407
408	if (!taskReadyForDataFromWal(pTask)) {	422,855✔
409	streamMetaReleaseTask(pStreamMeta, pTask);	301,758✔
410	continue;	301,959✔
411	}
412
413	*pScanIdle = false;	120,965✔
414
415	// seek the stored version and extract data from WAL
416	code = setWalReaderStartOffset(pTask, vgId);	120,965✔
417	if (code != TSDB_CODE_SUCCESS) {	120,919!
418	streamMetaReleaseTask(pStreamMeta, pTask);	×
419	continue;	×
420	}
421
422	int32_t numOfItems = streamQueueGetNumOfItems(pTask->inputq.queue);	120,919✔
423	int64_t maxVer = (pTask->info.fillHistory == 1) ? pTask->step2Range.maxVer : INT64_MAX;	121,021✔
424
425	streamMutexLock(&pTask->lock);	121,021✔
426
427	SStreamTaskState state = streamTaskGetStatus(pTask);	121,017✔
428	if (state.state != TASK_STATUS__READY) {	120,998✔
429	tqDebug("s-task:%s not ready for submit block from wal, status:%s", pTask->id.idStr, state.name);	28✔
430	streamMutexUnlock(&pTask->lock);	28✔
431	streamMetaReleaseTask(pStreamMeta, pTask);	28✔
432	continue;	37✔
433	}
434
435	bool hasNewData = false;	120,970✔
436	code = doPutDataIntoInputQ(pTask, maxVer, &numOfItems, &hasNewData);	120,970✔
437	streamMutexUnlock(&pTask->lock);	120,859✔
438
439	if ((numOfItems > 0) \|\| hasNewData) {	120,985!
440	noDataInWal = false;	40,267✔
441	code = streamTrySchedExec(pTask);	40,267✔
442	if (code != TSDB_CODE_SUCCESS) {	40,272!
443	streamMetaReleaseTask(pStreamMeta, pTask);	×
444	taosArrayDestroy(pTaskList);	×
445	return code;	×
446	}
447	}
448
449	streamMetaReleaseTask(pStreamMeta, pTask);	120,990✔
450	}
451
452	// all wal are checked, and no new data available in wal.
453	if (noDataInWal) {	104,181✔
454	*pScanIdle = true;	77,936✔
455	}
456
457	taosArrayDestroy(pTaskList);	104,181✔
458	return TSDB_CODE_SUCCESS;	104,105✔
459	}

taosdata / TDengine / #3534

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