#3719

Committed 26 Mar 2025 07:55AM UTC coverage: 1.208% (-17.7%) from 18.94%

Build # #3719

Build Type

push

travis-ci

Committed by GitHub

Commit Message

enh: update go.mod

Coverage Stats

3392 of 467393 branches covered (0.73%)

Branch coverage included in aggregate %.

7280 of 416407 relevant lines covered (1.75%)

25.6 hits per line

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

0.0

/enterprise/src/plugins/vnode/src/tsdbCompactMonitor.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 "vnd.h"

typedef struct SCompState SCompState;

struct SCompState {
  int32_t   fid;
  SVATaskID taskId;
  int64_t   compactSize;
};

struct SCompMonitor {
  int32_t totalCompTasks;
  int64_t startTimeSec;  // start time of seconds
  TARRAY2(SCompState) stateArr;
  int64_t totalCompactSize;
  int64_t finishedCompactSize;
  int64_t lastUpdateFinishedSizeTime;
};

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

int32_t tsdbOpenCompMonitor(STsdb *tsdb) {
  tsdb->pCompMonitor = (SCompMonitor *)taosMemoryCalloc(1, sizeof(SCompMonitor));
  if (tsdb->pCompMonitor == NULL) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }
  return 0;
}

void tsdbCloseCompMonitor(STsdb *tsdb) {
  if (tsdb->pCompMonitor) {
    TARRAY2_DESTROY(&tsdb->pCompMonitor->stateArr, NULL);
    taosMemoryFree(tsdb->pCompMonitor);
    tsdb->pCompMonitor = NULL;
  }
}

int32_t tsdbAddCompMonitorTask(STsdb *tsdb, int32_t fid, SVATaskID *taskId, int64_t compactSize) {
  if (TARRAY2_SIZE(&tsdb->pCompMonitor->stateArr) == 0) {
    tsdb->pCompMonitor->startTimeSec = taosGetTimestampSec();
    tsdb->pCompMonitor->totalCompTasks = 0;
    tsdb->pCompMonitor->totalCompactSize = 0;
    tsdb->pCompMonitor->finishedCompactSize = 0;
    tsdb->pCompMonitor->lastUpdateFinishedSizeTime = tsdb->pCompMonitor->startTimeSec;
  }

  SCompState state = {
      .fid = fid,
      .taskId = *taskId,
      .compactSize = compactSize,
  };

  int32_t code = TARRAY2_APPEND(&tsdb->pCompMonitor->stateArr, state);
  if (code) return code;
  tsdb->pCompMonitor->totalCompTasks++;
  tsdb->pCompMonitor->totalCompactSize += compactSize;
  tsdbDebug("vid:%d, fid:%d, taskId:%" PRId64 " is added to compact monitor, number of tasks:%d", TD_VID(tsdb->pVnode),
            fid, taskId->id, TARRAY2_SIZE(&tsdb->pCompMonitor->stateArr));
  return 0;
}

int32_t tsdbUpdateCompMonitorTask(STsdb *tsdb, int32_t fid, SVATaskID *taskId, int64_t compactSize) {
  for (int32_t i = 0; i < TARRAY2_SIZE(&tsdb->pCompMonitor->stateArr); i++) {
    SCompState *state = TARRAY2_GET_PTR(&tsdb->pCompMonitor->stateArr, i);
    if (state->fid == fid) {
      tsdb->pCompMonitor->totalCompactSize -= state->compactSize;
      tsdb->pCompMonitor->totalCompactSize += compactSize;
      state->compactSize = compactSize;
      tsdbDebug("vid:%d, fid:%d, taskId:%" PRId64 " is updated in compact monitor, number of tasks:%d",
                TD_VID(tsdb->pVnode), fid, taskId->id, TARRAY2_SIZE(&tsdb->pCompMonitor->stateArr));
      return 0;
    }
  }
  return 0;
}

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

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

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

void tsdbStopAllCompTask(STsdb *tsdb) {
  int32_t i;

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

  if (tsdb->pCompMonitor == NULL) {
    TAOS_UNUSED(taosThreadMutexUnlock(&tsdb->mutex));
    return;
  }

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

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

int32_t tsdbCompMonitorGetInfo(STsdb *tsdb, SQueryCompactProgressRsp *rsp) {
  TAOS_UNUSED(taosThreadMutexLock(&tsdb->mutex));
  rsp->compactId = 0;  // TODO
  rsp->vgId = TD_VID(tsdb->pVnode);
  rsp->numberFileset = tsdb->pCompMonitor->totalCompTasks;
  rsp->finished = rsp->numberFileset - TARRAY2_SIZE(&tsdb->pCompMonitor->stateArr);
  // calculate progress
  if (tsdb->pCompMonitor->totalCompactSize > 0) {
    rsp->progress = tsdb->pCompMonitor->finishedCompactSize * 100 / tsdb->pCompMonitor->totalCompactSize;
  } else {
    rsp->progress = 0;
  }
  // calculate estimated remaining time
  int64_t elapsed = tsdb->pCompMonitor->lastUpdateFinishedSizeTime - tsdb->pCompMonitor->startTimeSec;
  if (rsp->progress > 0 && elapsed > 0) {
    rsp->remainingTime = elapsed * (100 - rsp->progress) / rsp->progress;
  } else {
    rsp->remainingTime = tsdb->pCompMonitor->totalCompactSize / (20 * 1024 * 1024);  // suppose 20MB/s
  }
  TAOS_UNUSED(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 "vnd.h"
18
19	typedef struct SCompState SCompState;
20
21	struct SCompState {
22	int32_t fid;
23	SVATaskID taskId;
24	int64_t compactSize;
25	};
26
27	struct SCompMonitor {
28	int32_t totalCompTasks;
29	int64_t startTimeSec; // start time of seconds
30	TARRAY2(SCompState) stateArr;
31	int64_t totalCompactSize;
32	int64_t finishedCompactSize;
33	int64_t lastUpdateFinishedSizeTime;
34	};
35
36	bool tsdbCompMonHasTask(STsdb *tsdb) { return (TARRAY2_SIZE(&tsdb->pCompMonitor->stateArr) > 0); }	×
37
38	int32_t tsdbOpenCompMonitor(STsdb *tsdb) {	×
39	tsdb->pCompMonitor = (SCompMonitor *)taosMemoryCalloc(1, sizeof(SCompMonitor));	×
40	if (tsdb->pCompMonitor == NULL) {	×
41	return TSDB_CODE_OUT_OF_MEMORY;	×
42	}
43	return 0;	×
44	}
45
46	void tsdbCloseCompMonitor(STsdb *tsdb) {	×
47	if (tsdb->pCompMonitor) {	×
48	TARRAY2_DESTROY(&tsdb->pCompMonitor->stateArr, NULL);	×
49	taosMemoryFree(tsdb->pCompMonitor);	×
50	tsdb->pCompMonitor = NULL;	×
51	}
52	}	×
53
54	int32_t tsdbAddCompMonitorTask(STsdb tsdb, int32_t fid, SVATaskID taskId, int64_t compactSize) {	×
55	if (TARRAY2_SIZE(&tsdb->pCompMonitor->stateArr) == 0) {	×
56	tsdb->pCompMonitor->startTimeSec = taosGetTimestampSec();	×
57	tsdb->pCompMonitor->totalCompTasks = 0;	×
58	tsdb->pCompMonitor->totalCompactSize = 0;	×
59	tsdb->pCompMonitor->finishedCompactSize = 0;	×
60	tsdb->pCompMonitor->lastUpdateFinishedSizeTime = tsdb->pCompMonitor->startTimeSec;	×
61	}
62
63	SCompState state = {	×
64	.fid = fid,
65	.taskId = *taskId,
66	.compactSize = compactSize,
67	};
68
69	int32_t code = TARRAY2_APPEND(&tsdb->pCompMonitor->stateArr, state);	×
70	if (code) return code;	×
71	tsdb->pCompMonitor->totalCompTasks++;	×
72	tsdb->pCompMonitor->totalCompactSize += compactSize;	×
73	tsdbDebug("vid:%d, fid:%d, taskId:%" PRId64 " is added to compact monitor, number of tasks:%d", TD_VID(tsdb->pVnode),	×
74	fid, taskId->id, TARRAY2_SIZE(&tsdb->pCompMonitor->stateArr));
75	return 0;	×
76	}
77
78	int32_t tsdbUpdateCompMonitorTask(STsdb tsdb, int32_t fid, SVATaskID taskId, int64_t compactSize) {	×
79	for (int32_t i = 0; i < TARRAY2_SIZE(&tsdb->pCompMonitor->stateArr); i++) {	×
80	SCompState *state = TARRAY2_GET_PTR(&tsdb->pCompMonitor->stateArr, i);	×
81	if (state->fid == fid) {	×
82	tsdb->pCompMonitor->totalCompactSize -= state->compactSize;	×
83	tsdb->pCompMonitor->totalCompactSize += compactSize;	×
84	state->compactSize = compactSize;	×
85	tsdbDebug("vid:%d, fid:%d, taskId:%" PRId64 " is updated in compact monitor, number of tasks:%d",	×
86	TD_VID(tsdb->pVnode), fid, taskId->id, TARRAY2_SIZE(&tsdb->pCompMonitor->stateArr));
87	return 0;	×
88	}
89	}
90	return 0;	×
91	}
92
93	void tsdbRemoveCompMonitorTask(STsdb tsdb, SVATaskID taskId) {	×
94	TAOS_UNUSED(taosThreadMutexLock(&tsdb->mutex));	×
95
96	for (int32_t i = 0; i < TARRAY2_SIZE(&tsdb->pCompMonitor->stateArr); i++) {	×
97	SCompState *state = TARRAY2_GET_PTR(&tsdb->pCompMonitor->stateArr, i);	×
98	if (state->taskId.async == taskId->async && state->taskId.id == taskId->id) {	×
99	tsdbInfo("vid:%d, fid:%d, taskId:%" PRId64 " is removed from compact monitor, number of tasks:%d",	×
100	TD_VID(tsdb->pVnode), state->fid, taskId->id, TARRAY2_SIZE(&tsdb->pCompMonitor->stateArr));
101	tsdb->pCompMonitor->finishedCompactSize += state->compactSize;	×
102	tsdb->pCompMonitor->lastUpdateFinishedSizeTime = taosGetTimestampSec();	×
103	TARRAY2_REMOVE(&tsdb->pCompMonitor->stateArr, i, NULL);	×
104	break;	×
105	}
106	}
107
108	TAOS_UNUSED(taosThreadMutexUnlock(&tsdb->mutex));	×
109	}	×
110
111	void tsdbStopAllCompTask(STsdb *tsdb) {	×
112	int32_t i;
113
114	TAOS_UNUSED(taosThreadMutexLock(&tsdb->mutex));	×
115
116	if (tsdb->pCompMonitor == NULL) {	×
117	TAOS_UNUSED(taosThreadMutexUnlock(&tsdb->mutex));	×
118	return;	×
119	}
120
121	i = 0;	×
122	while (i < TARRAY2_SIZE(&tsdb->pCompMonitor->stateArr)) {	×
123	SCompState *state = TARRAY2_GET_PTR(&tsdb->pCompMonitor->stateArr, i);	×
124	if (vnodeACancel(&state->taskId) == 0) {	×
125	TARRAY2_REMOVE(&tsdb->pCompMonitor->stateArr, i, NULL);	×
126	} else {
127	i++;	×
128	}
129	}
130
131	TAOS_UNUSED(taosThreadMutexUnlock(&tsdb->mutex));	×
132	return;	×
133	}
134
135	int32_t tsdbCompMonitorGetInfo(STsdb tsdb, SQueryCompactProgressRsp rsp) {	×
136	TAOS_UNUSED(taosThreadMutexLock(&tsdb->mutex));	×
137	rsp->compactId = 0; // TODO	×
138	rsp->vgId = TD_VID(tsdb->pVnode);	×
139	rsp->numberFileset = tsdb->pCompMonitor->totalCompTasks;	×
140	rsp->finished = rsp->numberFileset - TARRAY2_SIZE(&tsdb->pCompMonitor->stateArr);	×
141	// calculate progress
142	if (tsdb->pCompMonitor->totalCompactSize > 0) {	×
143	rsp->progress = tsdb->pCompMonitor->finishedCompactSize * 100 / tsdb->pCompMonitor->totalCompactSize;	×
144	} else {
145	rsp->progress = 0;	×
146	}
147	// calculate estimated remaining time
148	int64_t elapsed = tsdb->pCompMonitor->lastUpdateFinishedSizeTime - tsdb->pCompMonitor->startTimeSec;	×
149	if (rsp->progress > 0 && elapsed > 0) {	×
150	rsp->remainingTime = elapsed * (100 - rsp->progress) / rsp->progress;	×
151	} else {
152	rsp->remainingTime = tsdb->pCompMonitor->totalCompactSize / (20 * 1024 * 1024); // suppose 20MB/s	×
153	}
154	TAOS_UNUSED(taosThreadMutexUnlock(&tsdb->mutex));	×
155	return 0;	×
156	}

taosdata / TDengine / #3719

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