#3597

Committed 05 Feb 2025 01:41AM UTC coverage: 63.546% (-0.02%) from 63.562%

Build # #3597

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #29639 from taosdata/feat/3.0/TS-5795

Enh(tsdb):print fid while data file corrupted.

Run Details

141230 of 285630 branches covered (49.45%)

Branch coverage included in aggregate %.

2 of 4 new or added lines in 1 file covered. (50.0%)

398 existing lines in 102 files now uncovered.

220015 of 282846 relevant lines covered (77.79%)

18937864.43 hits per line

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

74.16

/source/libs/stream/src/streamQueue.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 "streamInt.h"

#define MAX_STREAM_EXEC_BATCH_NUM 32
#define MAX_SMOOTH_BURST_RATIO    5  // 5 sec

// todo refactor:
// read data from input queue
typedef struct SQueueReader {
  SStreamQueue* pQueue;
  int32_t       taskLevel;
  int32_t       maxBlocks;     // maximum block in one batch
  int32_t       waitDuration;  // maximum wait time to format several block into a batch to process, unit: ms
} SQueueReader;

#define streamQueueCurItem(_q) ((_q)->qItem)

static bool streamTaskExtractAvailableToken(STokenBucket* pBucket, const char* id);
static void streamTaskPutbackToken(STokenBucket* pBucket);
static void streamTaskConsumeQuota(STokenBucket* pBucket, int32_t bytes);

static void streamQueueCleanup(SStreamQueue* pQueue) {
  SStreamQueueItem* qItem = NULL;
  while (1) {
    streamQueueNextItem(pQueue, &qItem);
    if (qItem == NULL) {
      break;
    }
    streamFreeQitem(qItem);
  }
  pQueue->status = STREAM_QUEUE__SUCESS;
}

int32_t streamQueueOpen(int64_t cap, SStreamQueue** pQ) {
  *pQ = NULL;
  int32_t code = 0;

  SStreamQueue* pQueue = taosMemoryCalloc(1, sizeof(SStreamQueue));
  if (pQueue == NULL) {
    return terrno;
  }

  code = taosOpenQueue(&pQueue->pQueue);
  if (code) {
    taosMemoryFreeClear(pQueue);
    return code;
  }

  code = taosAllocateQall(&pQueue->qall);
  if (code) {
    taosCloseQueue(pQueue->pQueue);
    taosMemoryFree(pQueue);
    return code;
  }

  pQueue->status = STREAM_QUEUE__SUCESS;
  taosSetQueueCapacity(pQueue->pQueue, cap);
  taosSetQueueMemoryCapacity(pQueue->pQueue, cap * 1024);

  *pQ = pQueue;
  return code;
}

void streamQueueClose(SStreamQueue* pQueue, int32_t taskId) {
  stDebug("s-task:0x%x free the queue:%p, items in queue:%d", taskId, pQueue->pQueue,
          taosQueueItemSize(pQueue->pQueue));
  streamQueueCleanup(pQueue);

  taosFreeQall(pQueue->qall);
  taosCloseQueue(pQueue->pQueue);
  taosMemoryFree(pQueue);
}

void streamQueueNextItem(SStreamQueue* pQueue, SStreamQueueItem** pItem) {
  *pItem = NULL;
  int8_t flag = atomic_exchange_8(&pQueue->status, STREAM_QUEUE__PROCESSING);

  if (flag == STREAM_QUEUE__FAILED) {
    *pItem = streamQueueCurItem(pQueue);
  } else {
    pQueue->qItem = NULL;
    (void) taosGetQitem(pQueue->qall, &pQueue->qItem);
    if (pQueue->qItem == NULL) {
      (void) taosReadAllQitems(pQueue->pQueue, pQueue->qall);
      (void) taosGetQitem(pQueue->qall, &pQueue->qItem);
    }

    *pItem = streamQueueCurItem(pQueue);
  }
}

void streamQueueProcessSuccess(SStreamQueue* queue) {
  if (atomic_load_8(&queue->status) != STREAM_QUEUE__PROCESSING) {
    stError("invalid queue status:%d, expect:%d", atomic_load_8(&queue->status), STREAM_QUEUE__PROCESSING);
    return;
  }

  queue->qItem = NULL;
  atomic_store_8(&queue->status, STREAM_QUEUE__SUCESS);
}

void streamQueueProcessFail(SStreamQueue* queue) {
  if (atomic_load_8(&queue->status) != STREAM_QUEUE__PROCESSING) {
    stError("invalid queue status:%d, expect:%d", atomic_load_8(&queue->status), STREAM_QUEUE__PROCESSING);
    return;
  }
  atomic_store_8(&queue->status, STREAM_QUEUE__FAILED);
}

bool streamQueueIsFull(const SStreamQueue* pQueue) {
  int32_t numOfItems = streamQueueGetNumOfItems(pQueue);
  if (numOfItems >= STREAM_TASK_QUEUE_CAPACITY) {
    return true;
  }

  return (SIZE_IN_MiB(taosQueueMemorySize(pQueue->pQueue)) >= STREAM_TASK_QUEUE_CAPACITY_IN_SIZE);
}

int32_t streamQueueGetNumOfItems(const SStreamQueue* pQueue) {
  int32_t numOfItems1 = taosQueueItemSize(pQueue->pQueue);
  int32_t numOfItems2 = taosQallItemSize(pQueue->qall);

  return numOfItems1 + numOfItems2;
}

int32_t streamQueueGetNumOfUnAccessedItems(const SStreamQueue* pQueue) {
  int32_t numOfItems1 = taosQueueItemSize(pQueue->pQueue);
  int32_t numOfItems2 = taosQallUnAccessedItemSize(pQueue->qall);

  return numOfItems1 + numOfItems2;
}

int32_t streamQueueGetItemSize(const SStreamQueue* pQueue) {
  return taosQueueMemorySize(pQueue->pQueue) + taosQallUnAccessedMemSize(pQueue->qall);
}

int32_t streamQueueItemGetSize(const SStreamQueueItem* pItem) {
  STaosQnode* p = (STaosQnode*)((char*)pItem - sizeof(STaosQnode));
  return p->dataSize;
}

void streamQueueItemIncSize(const SStreamQueueItem* pItem, int32_t size) {
  STaosQnode* p = (STaosQnode*)((char*)pItem - sizeof(STaosQnode));
  p->dataSize += size;
}

const char* streamQueueItemGetTypeStr(int32_t type) {
  switch (type) {
    case STREAM_INPUT__CHECKPOINT:
      return "checkpoint";
    case STREAM_INPUT__CHECKPOINT_TRIGGER:
      return "checkpoint-trigger";
    case STREAM_INPUT__TRANS_STATE:
      return "trans-state";
    case STREAM_INPUT__REF_DATA_BLOCK:
      return "ref-block";
    default:
      return "datablock";
  }
}

EExtractDataCode streamTaskGetDataFromInputQ(SStreamTask* pTask, SStreamQueueItem** pInput, int32_t* numOfBlocks,
                                             int32_t* blockSize) {
  const char* id = pTask->id.idStr;
  int32_t     taskLevel = pTask->info.taskLevel;

  *pInput = NULL;
  *numOfBlocks = 0;
  *blockSize = 0;

  // no available token in bucket for sink task, let's wait for a little bit
  if (taskLevel == TASK_LEVEL__SINK && (!streamTaskExtractAvailableToken(pTask->outputInfo.pTokenBucket, id))) {
    stDebug("s-task:%s no available token in bucket for sink data, wait for 10ms", id);
    return EXEC_AFTER_IDLE;
  }

  while (1) {
    if (streamTaskShouldPause(pTask) || streamTaskShouldStop(pTask)) {
      stDebug("s-task:%s task should pause, extract input blocks:%d", id, *numOfBlocks);
      return EXEC_CONTINUE;
    }

    SStreamQueueItem* qItem = NULL;
    streamQueueNextItem(pTask->inputq.queue, (SStreamQueueItem**)&qItem);
    if (qItem == NULL) {
      // restore the token to bucket
      if (*numOfBlocks > 0) {
        *blockSize = streamQueueItemGetSize(*pInput);
        if (taskLevel == TASK_LEVEL__SINK) {
          streamTaskConsumeQuota(pTask->outputInfo.pTokenBucket, *blockSize);
        }
      } else {
        streamTaskPutbackToken(pTask->outputInfo.pTokenBucket);
      }

      return EXEC_CONTINUE;
    }

    // do not merge blocks for sink node and check point data block
    int8_t type = qItem->type;
    if (type == STREAM_INPUT__CHECKPOINT || type == STREAM_INPUT__CHECKPOINT_TRIGGER ||
        type == STREAM_INPUT__TRANS_STATE || type == STREAM_INPUT__REF_DATA_BLOCK) {
      const char* p = streamQueueItemGetTypeStr(type);

      if (*pInput == NULL) {
        stDebug("s-task:%s %s msg extracted, start to process immediately", id, p);

        // restore the token to bucket in case of checkpoint/trans-state msg
        streamTaskPutbackToken(pTask->outputInfo.pTokenBucket);
        *blockSize = 0;
        *numOfBlocks = 1;
        *pInput = qItem;
        return EXEC_CONTINUE;
      } else {  // previous existed blocks needs to be handle, before handle the checkpoint msg block
        stDebug("s-task:%s %s msg extracted, handle previous blocks, numOfBlocks:%d", id, p, *numOfBlocks);
        *blockSize = streamQueueItemGetSize(*pInput);
        if (taskLevel == TASK_LEVEL__SINK) {
          streamTaskConsumeQuota(pTask->outputInfo.pTokenBucket, *blockSize);
        }

        streamQueueProcessFail(pTask->inputq.queue);
        return EXEC_CONTINUE;
      }
    } else {
      if (*pInput == NULL) {
        *pInput = qItem;
      } else { // merge current block failed, let's handle the already merged blocks.
        void*   newRet = NULL;
        int32_t code = streamQueueMergeQueueItem(*pInput, qItem, (SStreamQueueItem**)&newRet);
        if (newRet == NULL) {
          if (code != -1) {
            stError("s-task:%s failed to merge blocks from inputQ, numOfBlocks:%d, code:%s", id, *numOfBlocks,
                    tstrerror(code));
          }

          *blockSize = streamQueueItemGetSize(*pInput);
          if (taskLevel == TASK_LEVEL__SINK) {
            streamTaskConsumeQuota(pTask->outputInfo.pTokenBucket, *blockSize);
          }

          streamQueueProcessFail(pTask->inputq.queue);
          return EXEC_CONTINUE;
        }

        *pInput = newRet;
      }

      *numOfBlocks += 1;
      streamQueueProcessSuccess(pTask->inputq.queue);

      if (*numOfBlocks >= MAX_STREAM_EXEC_BATCH_NUM) {
        stDebug("s-task:%s batch size limit:%d reached, start to process blocks", id, MAX_STREAM_EXEC_BATCH_NUM);

        *blockSize = streamQueueItemGetSize(*pInput);
        if (taskLevel == TASK_LEVEL__SINK) {
          streamTaskConsumeQuota(pTask->outputInfo.pTokenBucket, *blockSize);
        }

        return EXEC_CONTINUE;
      }
    }
  }
}

int32_t streamTaskPutDataIntoInputQ(SStreamTask* pTask, SStreamQueueItem* pItem) {
  int8_t      type = pItem->type;
  STaosQueue* pQueue = pTask->inputq.queue->pQueue;
  int32_t     total = streamQueueGetNumOfItems(pTask->inputq.queue) + 1;

  if (type == STREAM_INPUT__DATA_SUBMIT) {
    SStreamDataSubmit* px = (SStreamDataSubmit*)pItem;
    if ((pTask->info.taskLevel == TASK_LEVEL__SOURCE) && streamQueueIsFull(pTask->inputq.queue)) {
      double size = SIZE_IN_MiB(taosQueueMemorySize(pQueue));
      stTrace(
          "s-task:%s inputQ is full, capacity(size:%d num:%dMiB), current(blocks:%d, size:%.2fMiB) stop to push data",
          pTask->id.idStr, STREAM_TASK_QUEUE_CAPACITY, STREAM_TASK_QUEUE_CAPACITY_IN_SIZE, total, size);
      streamDataSubmitDestroy(px);
      return TSDB_CODE_STREAM_INPUTQ_FULL;
    }

    int32_t msgLen = px->submit.msgLen;
    int64_t ver = px->submit.ver;

    int32_t code = taosWriteQitem(pQueue, pItem);
    if (code != TSDB_CODE_SUCCESS) {
      streamDataSubmitDestroy(px);
      return code;
    }

    double size = SIZE_IN_MiB(taosQueueMemorySize(pQueue));

    // use the local variable to avoid the pItem be freed by other threads, since it has been put into queue already.
    stDebug("s-task:%s submit enqueue msgLen:%d ver:%" PRId64 ", total in queue:%d, size:%.2fMiB", pTask->id.idStr,
            msgLen, ver, total, size + SIZE_IN_MiB(msgLen));
  } else if (type == STREAM_INPUT__DATA_BLOCK || type == STREAM_INPUT__REF_DATA_BLOCK) {
    if (streamQueueIsFull(pTask->inputq.queue)) {
      double size = SIZE_IN_MiB(taosQueueMemorySize(pQueue));

      stTrace("s-task:%s input queue is full, capacity:%d size:%d MiB, current(blocks:%d, size:%.2fMiB) abort",
              pTask->id.idStr, STREAM_TASK_QUEUE_CAPACITY, STREAM_TASK_QUEUE_CAPACITY_IN_SIZE, total, size);
      streamFreeQitem(pItem);
      return TSDB_CODE_STREAM_INPUTQ_FULL;
    }

    int32_t code = taosWriteQitem(pQueue, pItem);
    if (code != TSDB_CODE_SUCCESS) {
      streamFreeQitem(pItem);
      return code;
    }

    double size = SIZE_IN_MiB(taosQueueMemorySize(pQueue));
    stDebug("s-task:%s blockdata enqueue, total in queue:%d, size:%.2fMiB", pTask->id.idStr, total, size);
  } else if (type == STREAM_INPUT__CHECKPOINT || type == STREAM_INPUT__CHECKPOINT_TRIGGER ||
             type == STREAM_INPUT__TRANS_STATE || type == STREAM_INPUT__DATA_RETRIEVE) {
    int32_t code = taosWriteQitem(pQueue, pItem);
    if (code != TSDB_CODE_SUCCESS) {
      streamFreeQitem(pItem);
      return code;
    }

    double size = SIZE_IN_MiB(taosQueueMemorySize(pQueue));
    stDebug("s-task:%s level:%d %s blockdata enqueue, total in queue:%d, size:%.2fMiB", pTask->id.idStr,
            pTask->info.taskLevel, streamQueueItemGetTypeStr(type), total, size);
  } else if (type == STREAM_INPUT__GET_RES) {
    // use the default memory limit, refactor later.
    int32_t code = taosWriteQitem(pQueue, pItem);
    if (code != TSDB_CODE_SUCCESS) {
      streamFreeQitem(pItem);
      return code;
    }

    double size = SIZE_IN_MiB(taosQueueMemorySize(pQueue));
    stDebug("s-task:%s data res enqueue, current(blocks:%d, size:%.2fMiB)", pTask->id.idStr, total, size);
  } else {
    stError("s-task:%s invalid type:%d to put in inputQ", pTask->id.idStr, type);
    return TSDB_CODE_INVALID_PARA;
  }

  if (type != STREAM_INPUT__GET_RES && type != STREAM_INPUT__CHECKPOINT && type != STREAM_INPUT__CHECKPOINT_TRIGGER &&
      (pTask->info.delaySchedParam != 0)) {
    (void)atomic_val_compare_exchange_8(&pTask->schedInfo.status, TASK_TRIGGER_STATUS__INACTIVE,
                                        TASK_TRIGGER_STATUS__MAY_ACTIVE);
    stDebug("s-task:%s new data arrived, active the sched-trigger, triggerStatus:%d", pTask->id.idStr,
            pTask->schedInfo.status);
  }

  return 0;
}

int32_t streamTaskPutTranstateIntoInputQ(SStreamTask* pTask) {
  int32_t           code = 0;
  SStreamDataBlock* pTranstate = NULL;
  SSDataBlock*      pBlock = NULL;

  code = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, sizeof(SSDataBlock), (void**)&pTranstate);
  if (code) {
    return code;
  }

  pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
  if (pBlock == NULL) {
    code = terrno;
    goto _err;
  }

  pTranstate->type = STREAM_INPUT__TRANS_STATE;

  pBlock->info.type = STREAM_TRANS_STATE;
  pBlock->info.rows = 1;
  pBlock->info.childId = pTask->info.selfChildId;

  pTranstate->blocks = taosArrayInit(4, sizeof(SSDataBlock));  // pBlock;
  if (pTranstate->blocks == NULL) {
    code = terrno;
    goto _err;
  }

  void* p = taosArrayPush(pTranstate->blocks, pBlock);
  if (p == NULL) {
    code = terrno;
    goto _err;
  }

  taosMemoryFree(pBlock);
  if (streamTaskPutDataIntoInputQ(pTask, (SStreamQueueItem*)pTranstate) < 0) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }

  pTask->status.appendTranstateBlock = true;
  return TSDB_CODE_SUCCESS;

_err:
  taosMemoryFree(pBlock);
  taosFreeQitem(pTranstate);
  return code;
}

// the result should be put into the outputQ in any cases, the result may be lost otherwise.
int32_t streamTaskPutDataIntoOutputQ(SStreamTask* pTask, SStreamDataBlock* pBlock) {
  STaosQueue* pQueue = pTask->outputq.queue->pQueue;
  int32_t     code = taosWriteQitem(pQueue, pBlock);

  int32_t total = streamQueueGetNumOfItems(pTask->outputq.queue);
  double  size = SIZE_IN_MiB(taosQueueMemorySize(pQueue));
  if (code != 0) {
    stError("s-task:%s failed to put res into outputQ, outputQ items:%d, size:%.2fMiB code:%s, result lost",
            pTask->id.idStr, total + 1, size, tstrerror(code));
  } else {
    if (streamQueueIsFull(pTask->outputq.queue)) {
      stWarn(
          "s-task:%s outputQ is full(outputQ items:%d, size:%.2fMiB), set the output status BLOCKING, wait for 500ms "
          "after handle this batch of blocks",
          pTask->id.idStr, total, size);
    } else {
      stDebug("s-task:%s data put into outputQ, outputQ items:%d, size:%.2fMiB", pTask->id.idStr, total, size);
    }
  }

  return code;
}

int32_t streamTaskInitTokenBucket(STokenBucket* pBucket, int32_t numCap, int32_t numRate, float quotaRate,
                                  const char* id) {
  if (numCap < 10 || numRate < 10 || pBucket == NULL) {
    stError("failed to init sink task bucket, cap:%d, rate:%d", numCap, numRate);
    return TSDB_CODE_INVALID_PARA;
  }

  pBucket->numCapacity = numCap;
  pBucket->numOfToken = numCap;
  pBucket->numRate = numRate;

  pBucket->quotaRate = quotaRate;
  pBucket->quotaCapacity = quotaRate * MAX_SMOOTH_BURST_RATIO;
  pBucket->quotaRemain = pBucket->quotaCapacity;

  pBucket->tokenFillTimestamp = taosGetTimestampMs();
  pBucket->quotaFillTimestamp = taosGetTimestampMs();
  stDebug("s-task:%s sink quotaRate:%.2fMiB, numRate:%d", id, quotaRate, numRate);
  return TSDB_CODE_SUCCESS;
}

static void fillTokenBucket(STokenBucket* pBucket, const char* id) {
  int64_t now = taosGetTimestampMs();

  int64_t deltaToken = now - pBucket->tokenFillTimestamp;
  if (pBucket->numOfToken < 0) {
    return;
  }

  int32_t incNum = (deltaToken / 1000.0) * pBucket->numRate;
  if (incNum > 0) {
    pBucket->numOfToken = TMIN(pBucket->numOfToken + incNum, pBucket->numCapacity);
    pBucket->tokenFillTimestamp = now;
  }

  // increase the new available quota as time goes on
  int64_t deltaQuota = now - pBucket->quotaFillTimestamp;
  double  incSize = (deltaQuota / 1000.0) * pBucket->quotaRate;
  if (incSize > 0) {
    pBucket->quotaRemain = TMIN(pBucket->quotaRemain + incSize, pBucket->quotaCapacity);
    pBucket->quotaFillTimestamp = now;
  }

  if (incNum > 0 || incSize > 0) {
    stTrace("token/quota available, token:%d inc:%d, token_TsDelta:%" PRId64
            ", quota:%.2fMiB inc:%.3fMiB quotaTs:%" PRId64 " now:%" PRId64 "ms, %s",
            pBucket->numOfToken, incNum, deltaToken, pBucket->quotaRemain, incSize, deltaQuota, now, id);
  }
}

bool streamTaskExtractAvailableToken(STokenBucket* pBucket, const char* id) {
  fillTokenBucket(pBucket, id);

  if (pBucket->numOfToken > 0) {
    if (pBucket->quotaRemain > 0) {
      pBucket->numOfToken -= 1;
      return true;
    } else {  // no available size quota now
      return false;
    }
  } else {
    return false;
  }
}

void streamTaskPutbackToken(STokenBucket* pBucket) {
  pBucket->numOfToken = TMIN(pBucket->numOfToken + 1, pBucket->numCapacity);
}

// size in KB
void streamTaskConsumeQuota(STokenBucket* pBucket, int32_t bytes) { pBucket->quotaRemain -= SIZE_IN_MiB(bytes); }

void streamTaskInputFail(SStreamTask* pTask) { atomic_store_8(&pTask->inputq.status, TASK_INPUT_STATUS__FAILED); }

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 "streamInt.h"
17
18	#define MAX_STREAM_EXEC_BATCH_NUM 32
19	#define MAX_SMOOTH_BURST_RATIO 5 // 5 sec
20
21	// todo refactor:
22	// read data from input queue
23	typedef struct SQueueReader {
24	SStreamQueue* pQueue;
25	int32_t taskLevel;
26	int32_t maxBlocks; // maximum block in one batch
27	int32_t waitDuration; // maximum wait time to format several block into a batch to process, unit: ms
28	} SQueueReader;
29
30	#define streamQueueCurItem(_q) ((_q)->qItem)
31
32	static bool streamTaskExtractAvailableToken(STokenBucket* pBucket, const char* id);
33	static void streamTaskPutbackToken(STokenBucket* pBucket);
34	static void streamTaskConsumeQuota(STokenBucket* pBucket, int32_t bytes);
35
36	static void streamQueueCleanup(SStreamQueue* pQueue) {	29,108✔
37	SStreamQueueItem* qItem = NULL;	29,108✔
38	while (1) {
39	streamQueueNextItem(pQueue, &qItem);	29,664✔
40	if (qItem == NULL) {	29,710✔
41	break;	29,154✔
42	}
43	streamFreeQitem(qItem);	556✔
44	}
45	pQueue->status = STREAM_QUEUE__SUCESS;	29,154✔
46	}	29,154✔
47
48	int32_t streamQueueOpen(int64_t cap, SStreamQueue** pQ) {	29,634✔
49	*pQ = NULL;	29,634✔
50	int32_t code = 0;	29,634✔
51
52	SStreamQueue* pQueue = taosMemoryCalloc(1, sizeof(SStreamQueue));	29,634!
53	if (pQueue == NULL) {	29,657!
54	return terrno;	×
55	}
56
57	code = taosOpenQueue(&pQueue->pQueue);	29,657✔
58	if (code) {	29,639!
59	taosMemoryFreeClear(pQueue);	×
60	return code;	×
61	}
62
63	code = taosAllocateQall(&pQueue->qall);	29,639✔
64	if (code) {	29,647!
65	taosCloseQueue(pQueue->pQueue);	×
66	taosMemoryFree(pQueue);	×
67	return code;	×
68	}
69
70	pQueue->status = STREAM_QUEUE__SUCESS;	29,647✔
71	taosSetQueueCapacity(pQueue->pQueue, cap);	29,647✔
72	taosSetQueueMemoryCapacity(pQueue->pQueue, cap * 1024);	29,649✔
73
74	*pQ = pQueue;	29,652✔
75	return code;	29,652✔
76	}
77
78	void streamQueueClose(SStreamQueue* pQueue, int32_t taskId) {	29,110✔
79	stDebug("s-task:0x%x free the queue:%p, items in queue:%d", taskId, pQueue->pQueue,	29,110✔
80	taosQueueItemSize(pQueue->pQueue));
81	streamQueueCleanup(pQueue);	29,110✔
82
83	taosFreeQall(pQueue->qall);	29,153✔
84	taosCloseQueue(pQueue->pQueue);	29,150✔
85	taosMemoryFree(pQueue);	29,146!
86	}	29,145✔
87
88	void streamQueueNextItem(SStreamQueue* pQueue, SStreamQueueItem** pItem) {	357,932✔
89	*pItem = NULL;	357,932✔
90	int8_t flag = atomic_exchange_8(&pQueue->status, STREAM_QUEUE__PROCESSING);	357,932✔
91
92	if (flag == STREAM_QUEUE__FAILED) {	358,199✔
93	*pItem = streamQueueCurItem(pQueue);	3,306✔
94	} else {
95	pQueue->qItem = NULL;	354,893✔
96	(void) taosGetQitem(pQueue->qall, &pQueue->qItem);	354,893✔
97	if (pQueue->qItem == NULL) {	354,798✔
98	(void) taosReadAllQitems(pQueue->pQueue, pQueue->qall);	213,779✔
99	(void) taosGetQitem(pQueue->qall, &pQueue->qItem);	213,856✔
100	}
101
102	*pItem = streamQueueCurItem(pQueue);	354,872✔
103	}
104	}	358,178✔
105
106	void streamQueueProcessSuccess(SStreamQueue* queue) {	169,570✔
107	if (atomic_load_8(&queue->status) != STREAM_QUEUE__PROCESSING) {	169,570!
108	stError("invalid queue status:%d, expect:%d", atomic_load_8(&queue->status), STREAM_QUEUE__PROCESSING);	×
109	return;	×
110	}
111
112	queue->qItem = NULL;	169,568✔
113	atomic_store_8(&queue->status, STREAM_QUEUE__SUCESS);	169,568✔
114	}
115
116	void streamQueueProcessFail(SStreamQueue* queue) {	3,305✔
117	if (atomic_load_8(&queue->status) != STREAM_QUEUE__PROCESSING) {	3,305!
118	stError("invalid queue status:%d, expect:%d", atomic_load_8(&queue->status), STREAM_QUEUE__PROCESSING);	×
119	return;	×
120	}
121	atomic_store_8(&queue->status, STREAM_QUEUE__FAILED);	3,305✔
122	}
123
124	bool streamQueueIsFull(const SStreamQueue* pQueue) {	401,818✔
125	int32_t numOfItems = streamQueueGetNumOfItems(pQueue);	401,818✔
126	if (numOfItems >= STREAM_TASK_QUEUE_CAPACITY) {	401,964!
UNCOV 127	return true;	×
128	}
129
130	return (SIZE_IN_MiB(taosQueueMemorySize(pQueue->pQueue)) >= STREAM_TASK_QUEUE_CAPACITY_IN_SIZE);	401,964✔
131	}
132
133	int32_t streamQueueGetNumOfItems(const SStreamQueue* pQueue) {	795,617✔
134	int32_t numOfItems1 = taosQueueItemSize(pQueue->pQueue);	795,617✔
135	int32_t numOfItems2 = taosQallItemSize(pQueue->qall);	796,090✔
136
137	return numOfItems1 + numOfItems2;	796,015✔
138	}
139
140	int32_t streamQueueGetNumOfUnAccessedItems(const SStreamQueue* pQueue) {	37,106✔
141	int32_t numOfItems1 = taosQueueItemSize(pQueue->pQueue);	37,106✔
142	int32_t numOfItems2 = taosQallUnAccessedItemSize(pQueue->qall);	37,107✔
143
144	return numOfItems1 + numOfItems2;	37,107✔
145	}
146
147	int32_t streamQueueGetItemSize(const SStreamQueue* pQueue) {	61,777✔
148	return taosQueueMemorySize(pQueue->pQueue) + taosQallUnAccessedMemSize(pQueue->qall);	61,777✔
149	}
150
151	int32_t streamQueueItemGetSize(const SStreamQueueItem* pItem) {	169,568✔
152	STaosQnode* p = (STaosQnode)((char)pItem - sizeof(STaosQnode));	169,568✔
153	return p->dataSize;	169,568✔
154	}
155
156	void streamQueueItemIncSize(const SStreamQueueItem* pItem, int32_t size) {	126,847✔
157	STaosQnode* p = (STaosQnode)((char)pItem - sizeof(STaosQnode));	126,847✔
158	p->dataSize += size;	126,847✔
159	}	126,847✔
160
161	const char* streamQueueItemGetTypeStr(int32_t type) {	64,530✔
162	switch (type) {	64,530✔
163	case STREAM_INPUT__CHECKPOINT:	4,692✔
164	return "checkpoint";	4,692✔
165	case STREAM_INPUT__CHECKPOINT_TRIGGER:	18,487✔
166	return "checkpoint-trigger";	18,487✔
167	case STREAM_INPUT__TRANS_STATE:	16,243✔
168	return "trans-state";	16,243✔
169	case STREAM_INPUT__REF_DATA_BLOCK:	5,974✔
170	return "ref-block";	5,974✔
171	default:	19,134✔
172	return "datablock";	19,134✔
173	}
174	}
175
176	EExtractDataCode streamTaskGetDataFromInputQ(SStreamTask* pTask, SStreamQueueItem** pInput, int32_t* numOfBlocks,	131,159✔
177	int32_t* blockSize) {
178	const char* id = pTask->id.idStr;	131,159✔
179	int32_t taskLevel = pTask->info.taskLevel;	131,159✔
180
181	*pInput = NULL;	131,159✔
182	*numOfBlocks = 0;	131,159✔
183	*blockSize = 0;	131,159✔
184
185	// no available token in bucket for sink task, let's wait for a little bit
186	if (taskLevel == TASK_LEVEL__SINK && (!streamTaskExtractAvailableToken(pTask->outputInfo.pTokenBucket, id))) {	131,159!
187	stDebug("s-task:%s no available token in bucket for sink data, wait for 10ms", id);	×
188	return EXEC_AFTER_IDLE;	×
189	}
190
191	while (1) {	167,184✔
192	if (streamTaskShouldPause(pTask) \|\| streamTaskShouldStop(pTask)) {	298,333!
193	stDebug("s-task:%s task should pause, extract input blocks:%d", id, *numOfBlocks);	×
194	return EXEC_CONTINUE;	131,206✔
195	}
196
197	SStreamQueueItem* qItem = NULL;	298,260✔
198	streamQueueNextItem(pTask->inputq.queue, (SStreamQueueItem**)&qItem);	298,260✔
199	if (qItem == NULL) {	298,401✔
200	// restore the token to bucket
201	if (*numOfBlocks > 0) {	99,344✔
202	blockSize = streamQueueItemGetSize(pInput);	37,009✔
203	if (taskLevel == TASK_LEVEL__SINK) {	37,006✔
204	streamTaskConsumeQuota(pTask->outputInfo.pTokenBucket, *blockSize);	11,078✔
205	}
206	} else {
207	streamTaskPutbackToken(pTask->outputInfo.pTokenBucket);	62,335✔
208	}
209
210	return EXEC_CONTINUE;	99,329✔
211	}
212
213	// do not merge blocks for sink node and check point data block
214	int8_t type = qItem->type;	199,057✔
215	if (type == STREAM_INPUT__CHECKPOINT \|\| type == STREAM_INPUT__CHECKPOINT_TRIGGER \|\|	199,057✔
216	type == STREAM_INPUT__TRANS_STATE \|\| type == STREAM_INPUT__REF_DATA_BLOCK) {	174,316✔
217	const char* p = streamQueueItemGetTypeStr(type);	29,320✔
218
219	if (*pInput == NULL) {	29,329✔
220	stDebug("s-task:%s %s msg extracted, start to process immediately", id, p);	26,167✔
221
222	// restore the token to bucket in case of checkpoint/trans-state msg
223	streamTaskPutbackToken(pTask->outputInfo.pTokenBucket);	26,167✔
224	*blockSize = 0;	26,160✔
225	*numOfBlocks = 1;	26,160✔
226	*pInput = qItem;	26,160✔
227	return EXEC_CONTINUE;	26,160✔
228	} else { // previous existed blocks needs to be handle, before handle the checkpoint msg block
229	stDebug("s-task:%s %s msg extracted, handle previous blocks, numOfBlocks:%d", id, p, *numOfBlocks);	3,162✔
230	blockSize = streamQueueItemGetSize(pInput);	3,162✔
231	if (taskLevel == TASK_LEVEL__SINK) {	3,164✔
232	streamTaskConsumeQuota(pTask->outputInfo.pTokenBucket, *blockSize);	882✔
233	}
234
235	streamQueueProcessFail(pTask->inputq.queue);	3,163✔
236	return EXEC_CONTINUE;	3,166✔
237	}
238	} else {
239	if (*pInput == NULL) {	169,737✔
240	*pInput = qItem;	42,708✔
241	} else { // merge current block failed, let's handle the already merged blocks.
242	void* newRet = NULL;	127,029✔
243	int32_t code = streamQueueMergeQueueItem(pInput, qItem, (SStreamQueueItem*)&newRet);	127,029✔
244	if (newRet == NULL) {	127,024✔
245	if (code != -1) {	139!
246	stError("s-task:%s failed to merge blocks from inputQ, numOfBlocks:%d, code:%s", id, *numOfBlocks,	×
247	tstrerror(code));
248	}
249
250	blockSize = streamQueueItemGetSize(pInput);	139✔
251	if (taskLevel == TASK_LEVEL__SINK) {	140!
252	streamTaskConsumeQuota(pTask->outputInfo.pTokenBucket, *blockSize);	×
253	}
254
255	streamQueueProcessFail(pTask->inputq.queue);	140✔
256	return EXEC_CONTINUE;	140✔
257	}
258
259	*pInput = newRet;	126,885✔
260	}
261
262	*numOfBlocks += 1;	169,593✔
263	streamQueueProcessSuccess(pTask->inputq.queue);	169,593✔
264
265	if (*numOfBlocks >= MAX_STREAM_EXEC_BATCH_NUM) {	169,593✔
266	stDebug("s-task:%s batch size limit:%d reached, start to process blocks", id, MAX_STREAM_EXEC_BATCH_NUM);	2,409✔
267
268	blockSize = streamQueueItemGetSize(pInput);	2,409✔
269	if (taskLevel == TASK_LEVEL__SINK) {	2,411✔
270	streamTaskConsumeQuota(pTask->outputInfo.pTokenBucket, *blockSize);	25✔
271	}
272
273	return EXEC_CONTINUE;	2,411✔
274	}
275	}
276	}
277	}
278
279	int32_t streamTaskPutDataIntoInputQ(SStreamTask* pTask, SStreamQueueItem* pItem) {	196,255✔
280	int8_t type = pItem->type;	196,255✔
281	STaosQueue* pQueue = pTask->inputq.queue->pQueue;	196,255✔
282	int32_t total = streamQueueGetNumOfItems(pTask->inputq.queue) + 1;	196,255✔
283
284	if (type == STREAM_INPUT__DATA_SUBMIT) {	196,301✔
285	SStreamDataSubmit* px = (SStreamDataSubmit*)pItem;	146,537✔
286	if ((pTask->info.taskLevel == TASK_LEVEL__SOURCE) && streamQueueIsFull(pTask->inputq.queue)) {	146,537!
UNCOV 287	double size = SIZE_IN_MiB(taosQueueMemorySize(pQueue));	×
UNCOV 288	stTrace(	×
289	"s-task:%s inputQ is full, capacity(size:%d num:%dMiB), current(blocks:%d, size:%.2fMiB) stop to push data",
290	pTask->id.idStr, STREAM_TASK_QUEUE_CAPACITY, STREAM_TASK_QUEUE_CAPACITY_IN_SIZE, total, size);
UNCOV 291	streamDataSubmitDestroy(px);	×
UNCOV 292	return TSDB_CODE_STREAM_INPUTQ_FULL;	×
293	}
294
295	int32_t msgLen = px->submit.msgLen;	146,535✔
296	int64_t ver = px->submit.ver;	146,535✔
297
298	int32_t code = taosWriteQitem(pQueue, pItem);	146,535✔
299	if (code != TSDB_CODE_SUCCESS) {	146,538!
300	streamDataSubmitDestroy(px);	×
301	return code;	×
302	}
303
304	double size = SIZE_IN_MiB(taosQueueMemorySize(pQueue));	146,538✔
305
306	// use the local variable to avoid the pItem be freed by other threads, since it has been put into queue already.
307	stDebug("s-task:%s submit enqueue msgLen:%d ver:%" PRId64 ", total in queue:%d, size:%.2fMiB", pTask->id.idStr,	146,540✔
308	msgLen, ver, total, size + SIZE_IN_MiB(msgLen));
309	} else if (type == STREAM_INPUT__DATA_BLOCK \|\| type == STREAM_INPUT__REF_DATA_BLOCK) {	70,822✔
310	if (streamQueueIsFull(pTask->inputq.queue)) {	21,065!
311	double size = SIZE_IN_MiB(taosQueueMemorySize(pQueue));	×
312
313	stTrace("s-task:%s input queue is full, capacity:%d size:%d MiB, current(blocks:%d, size:%.2fMiB) abort",	×
314	pTask->id.idStr, STREAM_TASK_QUEUE_CAPACITY, STREAM_TASK_QUEUE_CAPACITY_IN_SIZE, total, size);
315	streamFreeQitem(pItem);	×
316	return TSDB_CODE_STREAM_INPUTQ_FULL;	×
317	}
318
319	int32_t code = taosWriteQitem(pQueue, pItem);	21,059✔
320	if (code != TSDB_CODE_SUCCESS) {	21,057!
321	streamFreeQitem(pItem);	×
322	return code;	×
323	}
324
325	double size = SIZE_IN_MiB(taosQueueMemorySize(pQueue));	21,057✔
326	stDebug("s-task:%s blockdata enqueue, total in queue:%d, size:%.2fMiB", pTask->id.idStr, total, size);	21,058✔
327	} else if (type == STREAM_INPUT__CHECKPOINT \|\| type == STREAM_INPUT__CHECKPOINT_TRIGGER \|\|	28,699✔
328	type == STREAM_INPUT__TRANS_STATE \|\| type == STREAM_INPUT__DATA_RETRIEVE) {	29,266✔
329	int32_t code = taosWriteQitem(pQueue, pItem);	24,257✔
330	if (code != TSDB_CODE_SUCCESS) {	24,267!
331	streamFreeQitem(pItem);	×
332	return code;	×
333	}
334
335	double size = SIZE_IN_MiB(taosQueueMemorySize(pQueue));	24,267✔
336	stDebug("s-task:%s level:%d %s blockdata enqueue, total in queue:%d, size:%.2fMiB", pTask->id.idStr,	24,274✔
337	pTask->info.taskLevel, streamQueueItemGetTypeStr(type), total, size);
338	} else if (type == STREAM_INPUT__GET_RES) {	4,442!
339	// use the default memory limit, refactor later.
340	int32_t code = taosWriteQitem(pQueue, pItem);	4,442✔
341	if (code != TSDB_CODE_SUCCESS) {	4,442!
342	streamFreeQitem(pItem);	×
343	return code;	×
344	}
345
346	double size = SIZE_IN_MiB(taosQueueMemorySize(pQueue));	4,442✔
347	stDebug("s-task:%s data res enqueue, current(blocks:%d, size:%.2fMiB)", pTask->id.idStr, total, size);	4,442✔
348	} else {
349	stError("s-task:%s invalid type:%d to put in inputQ", pTask->id.idStr, type);	×
350	return TSDB_CODE_INVALID_PARA;	×
351	}
352
353	if (type != STREAM_INPUT__GET_RES && type != STREAM_INPUT__CHECKPOINT && type != STREAM_INPUT__CHECKPOINT_TRIGGER &&	196,300✔
354	(pTask->info.delaySchedParam != 0)) {	177,355✔
355	(void)atomic_val_compare_exchange_8(&pTask->schedInfo.status, TASK_TRIGGER_STATUS__INACTIVE,	11,363✔
356	TASK_TRIGGER_STATUS__MAY_ACTIVE);
357	stDebug("s-task:%s new data arrived, active the sched-trigger, triggerStatus:%d", pTask->id.idStr,	11,363✔
358	pTask->schedInfo.status);
359	}
360
361	return 0;	196,294✔
362	}
363
364	int32_t streamTaskPutTranstateIntoInputQ(SStreamTask* pTask) {	2,258✔
365	int32_t code = 0;	2,258✔
366	SStreamDataBlock* pTranstate = NULL;	2,258✔
367	SSDataBlock* pBlock = NULL;	2,258✔
368
369	code = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, sizeof(SSDataBlock), (void**)&pTranstate);	2,258✔
370	if (code) {	2,258!
371	return code;	×
372	}
373
374	pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));	2,258!
375	if (pBlock == NULL) {	2,258!
376	code = terrno;	×
377	goto _err;	×
378	}
379
380	pTranstate->type = STREAM_INPUT__TRANS_STATE;	2,258✔
381
382	pBlock->info.type = STREAM_TRANS_STATE;	2,258✔
383	pBlock->info.rows = 1;	2,258✔
384	pBlock->info.childId = pTask->info.selfChildId;	2,258✔
385
386	pTranstate->blocks = taosArrayInit(4, sizeof(SSDataBlock)); // pBlock;	2,258✔
387	if (pTranstate->blocks == NULL) {	2,258!
388	code = terrno;	×
389	goto _err;	×
390	}
391
392	void* p = taosArrayPush(pTranstate->blocks, pBlock);	2,258✔
393	if (p == NULL) {	2,258!
394	code = terrno;	×
395	goto _err;	×
396	}
397
398	taosMemoryFree(pBlock);	2,258!
399	if (streamTaskPutDataIntoInputQ(pTask, (SStreamQueueItem*)pTranstate) < 0) {	2,258!
400	code = TSDB_CODE_OUT_OF_MEMORY;	×
401	goto _err;	×
402	}
403
404	pTask->status.appendTranstateBlock = true;	2,258✔
405	return TSDB_CODE_SUCCESS;	2,258✔
406
407	_err:	×
408	taosMemoryFree(pBlock);	×
409	taosFreeQitem(pTranstate);	×
410	return code;	×
411	}
412
413	// the result should be put into the outputQ in any cases, the result may be lost otherwise.
414	int32_t streamTaskPutDataIntoOutputQ(SStreamTask* pTask, SStreamDataBlock* pBlock) {	14,314✔
415	STaosQueue* pQueue = pTask->outputq.queue->pQueue;	14,314✔
416	int32_t code = taosWriteQitem(pQueue, pBlock);	14,314✔
417
418	int32_t total = streamQueueGetNumOfItems(pTask->outputq.queue);	14,315✔
419	double size = SIZE_IN_MiB(taosQueueMemorySize(pQueue));	14,314✔
420	if (code != 0) {	14,314!
421	stError("s-task:%s failed to put res into outputQ, outputQ items:%d, size:%.2fMiB code:%s, result lost",	×
422	pTask->id.idStr, total + 1, size, tstrerror(code));
423	} else {
424	if (streamQueueIsFull(pTask->outputq.queue)) {	14,314!
425	stWarn(	×
426	"s-task:%s outputQ is full(outputQ items:%d, size:%.2fMiB), set the output status BLOCKING, wait for 500ms "
427	"after handle this batch of blocks",
428	pTask->id.idStr, total, size);
429	} else {
430	stDebug("s-task:%s data put into outputQ, outputQ items:%d, size:%.2fMiB", pTask->id.idStr, total, size);	14,315✔
431	}
432	}
433
434	return code;	14,313✔
435	}
436
437	int32_t streamTaskInitTokenBucket(STokenBucket* pBucket, int32_t numCap, int32_t numRate, float quotaRate,	14,823✔
438	const char* id) {
439	if (numCap < 10 \|\| numRate < 10 \|\| pBucket == NULL) {	14,823!
440	stError("failed to init sink task bucket, cap:%d, rate:%d", numCap, numRate);	×
441	return TSDB_CODE_INVALID_PARA;	×
442	}
443
444	pBucket->numCapacity = numCap;	14,830✔
445	pBucket->numOfToken = numCap;	14,830✔
446	pBucket->numRate = numRate;	14,830✔
447
448	pBucket->quotaRate = quotaRate;	14,830✔
449	pBucket->quotaCapacity = quotaRate * MAX_SMOOTH_BURST_RATIO;	14,830✔
450	pBucket->quotaRemain = pBucket->quotaCapacity;	14,830✔
451
452	pBucket->tokenFillTimestamp = taosGetTimestampMs();	14,827✔
453	pBucket->quotaFillTimestamp = taosGetTimestampMs();	14,827✔
454	stDebug("s-task:%s sink quotaRate:%.2fMiB, numRate:%d", id, quotaRate, numRate);	14,827✔
455	return TSDB_CODE_SUCCESS;	14,827✔
456	}
457
458	static void fillTokenBucket(STokenBucket* pBucket, const char* id) {	47,262✔
459	int64_t now = taosGetTimestampMs();	47,260✔
460
461	int64_t deltaToken = now - pBucket->tokenFillTimestamp;	47,260✔
462	if (pBucket->numOfToken < 0) {	47,260!
463	return;	×
464	}
465
466	int32_t incNum = (deltaToken / 1000.0) * pBucket->numRate;	47,260✔
467	if (incNum > 0) {	47,260✔
468	pBucket->numOfToken = TMIN(pBucket->numOfToken + incNum, pBucket->numCapacity);	25,010✔
469	pBucket->tokenFillTimestamp = now;	25,010✔
470	}
471
472	// increase the new available quota as time goes on
473	int64_t deltaQuota = now - pBucket->quotaFillTimestamp;	47,260✔
474	double incSize = (deltaQuota / 1000.0) * pBucket->quotaRate;	47,260✔
475	if (incSize > 0) {	47,260✔
476	pBucket->quotaRemain = TMIN(pBucket->quotaRemain + incSize, pBucket->quotaCapacity);	29,928✔
477	pBucket->quotaFillTimestamp = now;	29,928✔
478	}
479
480	if (incNum > 0 \|\| incSize > 0) {	47,260✔
481	stTrace("token/quota available, token:%d inc:%d, token_TsDelta:%" PRId64	29,914✔
482	", quota:%.2fMiB inc:%.3fMiB quotaTs:%" PRId64 " now:%" PRId64 "ms, %s",
483	pBucket->numOfToken, incNum, deltaToken, pBucket->quotaRemain, incSize, deltaQuota, now, id);
484	}
485	}
486
487	bool streamTaskExtractAvailableToken(STokenBucket* pBucket, const char* id) {	47,261✔
488	fillTokenBucket(pBucket, id);	47,261✔
489
490	if (pBucket->numOfToken > 0) {	47,259!
491	if (pBucket->quotaRemain > 0) {	47,260!
492	pBucket->numOfToken -= 1;	47,260✔
493	return true;	47,260✔
494	} else { // no available size quota now
495	return false;	×
496	}
497	} else {
498	return false;	×
499	}
500	}
501
502	void streamTaskPutbackToken(STokenBucket* pBucket) {	88,469✔
503	pBucket->numOfToken = TMIN(pBucket->numOfToken + 1, pBucket->numCapacity);	88,469✔
504	}	88,469✔
505
506	// size in KB
507	void streamTaskConsumeQuota(STokenBucket* pBucket, int32_t bytes) { pBucket->quotaRemain -= SIZE_IN_MiB(bytes); }	11,983✔
508
509	void streamTaskInputFail(SStreamTask* pTask) { atomic_store_8(&pTask->inputq.status, TASK_INPUT_STATUS__FAILED); }	×

taosdata / TDengine / #3597

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