#4873

Committed 04 Dec 2025 01:55AM UTC coverage: 64.558% (-0.1%) from 64.678%

Build # #4873

Build Type

push

travis-ci

Committed by

guanshengliang

Commit Message

Merge branch '3.0' into cover/3.0

Run Details

719 of 2219 new or added lines in 36 files covered. (32.4%)

6363 existing lines in 135 files now uncovered.

159381 of 246882 relevant lines covered (64.56%)

108937395.15 hits per line

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

76.62

/source/dnode/vnode/src/tsdb/tsdbRetentionMonitor.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 "tsdb.h"
#include "tsdbInt.h"
#include "vnd.h"

bool tsdbRetentionMonHasTask(STsdb *tsdb) { return (TARRAY2_SIZE(&tsdb->pRetentionMonitor->stateArr) > 0); }

int32_t tsdbOpenRetentionMonitor(STsdb *tsdb) {
  tsdb->pRetentionMonitor = (SRetentionMonitor *)taosMemoryCalloc(1, sizeof(SRetentionMonitor));
  if (tsdb->pRetentionMonitor == NULL) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }
  return 0;
}

void tsdbCloseRetentionMonitor(STsdb *tsdb) {
  if (tsdb->pRetentionMonitor) {
    TARRAY2_DESTROY(&tsdb->pRetentionMonitor->stateArr, NULL);
    taosMemoryFreeClear(tsdb->pRetentionMonitor);
  }
}

int32_t tsdbAddRetentionMonitorTask(STsdb *tsdb, int32_t fid, SVATaskID *taskId, int64_t fileSize) {
  if (TARRAY2_SIZE(&tsdb->pRetentionMonitor->stateArr) == 0) {
    tsdb->pRetentionMonitor->startTimeSec = taosGetTimestampSec();
    tsdb->pRetentionMonitor->totalTasks = 0;
    tsdb->pRetentionMonitor->totalSize = 0;
    tsdb->pRetentionMonitor->finishedSize = 0;
    tsdb->pRetentionMonitor->lastUpdateFinishedSizeTime = tsdb->pRetentionMonitor->startTimeSec;
    tsdb->pRetentionMonitor->killed = 0;
  }

  SRetentionState state = {
      .fid = fid,
      .taskId = *taskId,
      .fileSize = fileSize,
  };

  int32_t code = TARRAY2_APPEND(&tsdb->pRetentionMonitor->stateArr, state);
  if (code) return code;
  tsdb->pRetentionMonitor->totalTasks++;
  tsdb->pRetentionMonitor->totalSize += fileSize;
  tsdbDebug("vid:%d, fid:%d, taskId:%" PRId64 " is added to retention monitor, number of tasks:%d", TD_VID(tsdb->pVnode),
            fid, taskId->id, TARRAY2_SIZE(&tsdb->pRetentionMonitor->stateArr));
  return 0;
}

int32_t tsdbUpdateRetentionMonitorTask(STsdb *tsdb, int32_t fid, SVATaskID *taskId, int64_t fileSize) {
  for (int32_t i = 0; i < TARRAY2_SIZE(&tsdb->pRetentionMonitor->stateArr); i++) {
    SRetentionState *state = TARRAY2_GET_PTR(&tsdb->pRetentionMonitor->stateArr, i);
    if (state->fid == fid) {
      tsdb->pRetentionMonitor->totalSize -= state->fileSize;
      tsdb->pRetentionMonitor->totalSize += fileSize;
      state->fileSize = fileSize;
      tsdbDebug("vid:%d, fid:%d, taskId:%" PRId64 " is updated in retention monitor, number of tasks:%d",
                TD_VID(tsdb->pVnode), fid, taskId->id, TARRAY2_SIZE(&tsdb->pRetentionMonitor->stateArr));
      return 0;
    }
  }
  return 0;
}

void tsdbRemoveRetentionMonitorTask(STsdb *tsdb, SVATaskID *taskId) {
  (void)taosThreadMutexLock(&tsdb->mutex);

  for (int32_t i = 0; i < TARRAY2_SIZE(&tsdb->pRetentionMonitor->stateArr); i++) {
    SRetentionState *state = TARRAY2_GET_PTR(&tsdb->pRetentionMonitor->stateArr, i);
    if (state->taskId.async == taskId->async && state->taskId.id == taskId->id) {
      tsdbInfo("vid:%d, fid:%d, taskId:%" PRId64 " is removed from retention monitor, number of tasks:%d",
               TD_VID(tsdb->pVnode), state->fid, taskId->id, TARRAY2_SIZE(&tsdb->pRetentionMonitor->stateArr));
      tsdb->pRetentionMonitor->finishedSize += state->fileSize;
      tsdb->pRetentionMonitor->lastUpdateFinishedSizeTime = taosGetTimestampSec();
      TARRAY2_REMOVE(&tsdb->pRetentionMonitor->stateArr, i, NULL);
      break;
    }
  }

  (void)taosThreadMutexUnlock(&tsdb->mutex);
}

void tsdbStopAllRetentionTask(STsdb *tsdb) {
  int32_t i;

  (void)taosThreadMutexLock(&tsdb->mutex);

  if (tsdb->pRetentionMonitor == NULL) {
    (void)taosThreadMutexUnlock(&tsdb->mutex);
    return;
  }

  atomic_store_8(&tsdb->pRetentionMonitor->killed, 1);

  i = 0;
  while (i < TARRAY2_SIZE(&tsdb->pRetentionMonitor->stateArr)) {
    SRetentionState *state = TARRAY2_GET_PTR(&tsdb->pRetentionMonitor->stateArr, i);
    if (vnodeACancel(&state->taskId) == 0) {
      TARRAY2_REMOVE(&tsdb->pRetentionMonitor->stateArr, i, NULL);
    } else {
      i++;
    }
  }

  (void)taosThreadMutexUnlock(&tsdb->mutex);
  return;
}

int32_t tsdbRetentionMonitorGetInfo(STsdb *tsdb, SQueryCompactProgressRsp *rsp) {
  (void)taosThreadMutexLock(&tsdb->mutex);
  rsp->compactId = 0;  // TODO
  rsp->vgId = TD_VID(tsdb->pVnode);
  rsp->numberFileset = tsdb->pRetentionMonitor->totalTasks;
  rsp->finished = rsp->numberFileset - TARRAY2_SIZE(&tsdb->pRetentionMonitor->stateArr);
  // calculate progress
  if (tsdb->pRetentionMonitor->totalSize > 0) {
    rsp->progress = tsdb->pRetentionMonitor->finishedSize * 100 / tsdb->pRetentionMonitor->totalSize;
  } else {
    rsp->progress = 0;
  }
  // calculate estimated remaining time
  int64_t elapsed = tsdb->pRetentionMonitor->lastUpdateFinishedSizeTime - tsdb->pRetentionMonitor->startTimeSec;
  if (rsp->progress > 0 && elapsed > 0) {
    rsp->remainingTime = elapsed * (100 - rsp->progress) / rsp->progress;
  } else {
    rsp->remainingTime = tsdb->pRetentionMonitor->totalSize / (20 * 1024 * 1024);  // suppose 20MB/s
  }
  (void)taosThreadMutexUnlock(&tsdb->mutex);
  return 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 "tsdb.h"
17	#include "tsdbInt.h"
18	#include "vnd.h"
19
20	bool tsdbRetentionMonHasTask(STsdb *tsdb) { return (TARRAY2_SIZE(&tsdb->pRetentionMonitor->stateArr) > 0); }	×
21
22	int32_t tsdbOpenRetentionMonitor(STsdb *tsdb) {	4,048,073✔
23	tsdb->pRetentionMonitor = (SRetentionMonitor *)taosMemoryCalloc(1, sizeof(SRetentionMonitor));	4,048,073✔
24	if (tsdb->pRetentionMonitor == NULL) {	4,063,592✔
25	return TSDB_CODE_OUT_OF_MEMORY;	×
26	}
27	return 0;	4,063,634✔
28	}
29
30	void tsdbCloseRetentionMonitor(STsdb *tsdb) {	4,064,148✔
31	if (tsdb->pRetentionMonitor) {	4,064,148✔
32	TARRAY2_DESTROY(&tsdb->pRetentionMonitor->stateArr, NULL);	4,064,224✔
33	taosMemoryFreeClear(tsdb->pRetentionMonitor);	4,064,223✔
34	}
35	}	4,065,271✔
36
37	int32_t tsdbAddRetentionMonitorTask(STsdb tsdb, int32_t fid, SVATaskID taskId, int64_t fileSize) {	213,430✔
38	if (TARRAY2_SIZE(&tsdb->pRetentionMonitor->stateArr) == 0) {	213,430✔
39	tsdb->pRetentionMonitor->startTimeSec = taosGetTimestampSec();	66,229✔
40	tsdb->pRetentionMonitor->totalTasks = 0;	66,229✔
41	tsdb->pRetentionMonitor->totalSize = 0;	66,229✔
42	tsdb->pRetentionMonitor->finishedSize = 0;	66,229✔
43	tsdb->pRetentionMonitor->lastUpdateFinishedSizeTime = tsdb->pRetentionMonitor->startTimeSec;	66,229✔
44	tsdb->pRetentionMonitor->killed = 0;	66,229✔
45	}
46
47	SRetentionState state = {	213,430✔
48	.fid = fid,
49	.taskId = *taskId,
50	.fileSize = fileSize,
51	};
52
53	int32_t code = TARRAY2_APPEND(&tsdb->pRetentionMonitor->stateArr, state);	213,430✔
54	if (code) return code;	213,430✔
55	tsdb->pRetentionMonitor->totalTasks++;	213,430✔
56	tsdb->pRetentionMonitor->totalSize += fileSize;	213,430✔
57	tsdbDebug("vid:%d, fid:%d, taskId:%" PRId64 " is added to retention monitor, number of tasks:%d", TD_VID(tsdb->pVnode),	213,430✔
58	fid, taskId->id, TARRAY2_SIZE(&tsdb->pRetentionMonitor->stateArr));
59	return 0;	213,430✔
60	}
61
62	int32_t tsdbUpdateRetentionMonitorTask(STsdb tsdb, int32_t fid, SVATaskID taskId, int64_t fileSize) {	×
63	for (int32_t i = 0; i < TARRAY2_SIZE(&tsdb->pRetentionMonitor->stateArr); i++) {	×
64	SRetentionState *state = TARRAY2_GET_PTR(&tsdb->pRetentionMonitor->stateArr, i);	×
65	if (state->fid == fid) {	×
66	tsdb->pRetentionMonitor->totalSize -= state->fileSize;	×
67	tsdb->pRetentionMonitor->totalSize += fileSize;	×
68	state->fileSize = fileSize;	×
69	tsdbDebug("vid:%d, fid:%d, taskId:%" PRId64 " is updated in retention monitor, number of tasks:%d",	×
70	TD_VID(tsdb->pVnode), fid, taskId->id, TARRAY2_SIZE(&tsdb->pRetentionMonitor->stateArr));
71	return 0;	×
72	}
73	}
74	return 0;	×
75	}
76
77	void tsdbRemoveRetentionMonitorTask(STsdb tsdb, SVATaskID taskId) {	213,430✔
78	(void)taosThreadMutexLock(&tsdb->mutex);	213,430✔
79
80	for (int32_t i = 0; i < TARRAY2_SIZE(&tsdb->pRetentionMonitor->stateArr); i++) {	213,430✔
81	SRetentionState *state = TARRAY2_GET_PTR(&tsdb->pRetentionMonitor->stateArr, i);	213,430✔
82	if (state->taskId.async == taskId->async && state->taskId.id == taskId->id) {	213,430✔
83	tsdbInfo("vid:%d, fid:%d, taskId:%" PRId64 " is removed from retention monitor, number of tasks:%d",	213,430✔
84	TD_VID(tsdb->pVnode), state->fid, taskId->id, TARRAY2_SIZE(&tsdb->pRetentionMonitor->stateArr));
85	tsdb->pRetentionMonitor->finishedSize += state->fileSize;	213,430✔
86	tsdb->pRetentionMonitor->lastUpdateFinishedSizeTime = taosGetTimestampSec();	213,430✔
87	TARRAY2_REMOVE(&tsdb->pRetentionMonitor->stateArr, i, NULL);	213,430✔
88	break;	213,430✔
89	}
90	}
91
92	(void)taosThreadMutexUnlock(&tsdb->mutex);	213,430✔
93	}	213,430✔
94
95	void tsdbStopAllRetentionTask(STsdb *tsdb) {	4,077,715✔
96	int32_t i;
97
98	(void)taosThreadMutexLock(&tsdb->mutex);	4,077,715✔
99
100	if (tsdb->pRetentionMonitor == NULL) {	4,085,030✔
101	(void)taosThreadMutexUnlock(&tsdb->mutex);	×
102	return;	×
103	}
104
105	atomic_store_8(&tsdb->pRetentionMonitor->killed, 1);	4,084,346✔
106
107	i = 0;	4,085,030✔
108	while (i < TARRAY2_SIZE(&tsdb->pRetentionMonitor->stateArr)) {	4,085,030✔
UNCOV 109	SRetentionState *state = TARRAY2_GET_PTR(&tsdb->pRetentionMonitor->stateArr, i);	×
UNCOV 110	if (vnodeACancel(&state->taskId) == 0) {	×
111	TARRAY2_REMOVE(&tsdb->pRetentionMonitor->stateArr, i, NULL);	×
112	} else {
UNCOV 113	i++;	×
114	}
115	}
116
117	(void)taosThreadMutexUnlock(&tsdb->mutex);	4,083,208✔
118	return;	4,083,954✔
119	}
120
121	int32_t tsdbRetentionMonitorGetInfo(STsdb tsdb, SQueryCompactProgressRsp rsp) {	118,578✔
122	(void)taosThreadMutexLock(&tsdb->mutex);	118,578✔
123	rsp->compactId = 0; // TODO	120,116✔
124	rsp->vgId = TD_VID(tsdb->pVnode);	120,116✔
125	rsp->numberFileset = tsdb->pRetentionMonitor->totalTasks;	120,116✔
126	rsp->finished = rsp->numberFileset - TARRAY2_SIZE(&tsdb->pRetentionMonitor->stateArr);	120,116✔
127	// calculate progress
128	if (tsdb->pRetentionMonitor->totalSize > 0) {	120,116✔
129	rsp->progress = tsdb->pRetentionMonitor->finishedSize * 100 / tsdb->pRetentionMonitor->totalSize;	116,023✔
130	} else {
131	rsp->progress = 0;	3,324✔
132	}
133	// calculate estimated remaining time
134	int64_t elapsed = tsdb->pRetentionMonitor->lastUpdateFinishedSizeTime - tsdb->pRetentionMonitor->startTimeSec;	120,116✔
135	if (rsp->progress > 0 && elapsed > 0) {	120,116✔
136	rsp->remainingTime = elapsed * (100 - rsp->progress) / rsp->progress;	9,231✔
137	} else {
138	rsp->remainingTime = tsdb->pRetentionMonitor->totalSize / (20 * 1024 * 1024); // suppose 20MB/s	110,116✔
139	}
140	(void)taosThreadMutexUnlock(&tsdb->mutex);	119,347✔
141	return 0;	120,116✔
142	}

taosdata / TDengine / #4873

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