#5011

Committed 03 Apr 2026 03:59PM UTC coverage: 72.3% (+0.008%) from 72.292%

Build # #5011

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

merge: from main to 3.0 branch #35067

Coverage Stats

4053 of 5985 new or added lines in 68 files covered. (67.72%)

732 existing lines in 143 files now uncovered.

257430 of 356056 relevant lines covered (72.3%)

131834103.52 hits per line

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

88.0

/source/libs/new-stream/src/stream.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 "executor.h"
#include "streamInt.h"
#include "tmisce.h"
#include "tref.h"
#include "stream.h"
#include "ttimer.h"
#include "dataSink.h"

SStreamMgmtInfo gStreamMgmt = {0};

void streamSetSnodeEnabled(  SMsgCb* msgCb) {
  if (tsDisableStream) {
    return;
  }
  
  gStreamMgmt.snodeEnabled = true;
  gStreamMgmt.msgCb = *msgCb;
  stInfo("snode %d enabled", (*gStreamMgmt.getDnode)(gStreamMgmt.dnode));
}

void streamSetSnodeDisabled(bool cleanup) {
  if (tsDisableStream) {
    return;
  }
  
  stInfo("snode disabled");
  gStreamMgmt.snodeEnabled = false;
  smUndeploySnodeTasks(cleanup);
}

void streamMgmtCleanup() {
  taosArrayDestroy(gStreamMgmt.vgLeaders);
  gStreamMgmt.vgLeaders = NULL;
  taosHashCleanup(gStreamMgmt.taskMap);
  gStreamMgmt.taskMap = NULL;
  taosHashCleanup(gStreamMgmt.vgroupMap);
  gStreamMgmt.vgroupMap = NULL;
  for (int32_t i = 0; i < STREAM_MAX_GROUP_NUM; ++i) {
    taosHashCleanup(gStreamMgmt.stmGrp[i]);
    gStreamMgmt.stmGrp[i] = NULL;
  }
}

void streamCleanup(void) {
  if (tsDisableStream) {
    return;
  }
  
  stInfo("stream cleanup start");
  stTriggerTaskEnvCleanup();
  streamTimerCleanUp();
  smUndeployAllTasks();
  destroyDataSinkMgr();
  streamMgmtCleanup();
  destroyInserterGrpInfo();
  stInfo("stream cleanup end");
}

int32_t streamInit(void* pDnode, getDnodeId_f getDnode, getMnodeEpset_f getMnode, getSynEpset_f getSynEpset) {
  if (tsDisableStream) {
    stInfo("stream disabled");
    return TSDB_CODE_SUCCESS;
  }
  
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;

  gStreamMgmt.dnode = pDnode;
  gStreamMgmt.getMnode = getMnode;
  gStreamMgmt.getDnode = getDnode;
  gStreamMgmt.getSynEpset = getSynEpset;

  gStreamMgmt.vgLeaders = taosArrayInit(20, sizeof(int32_t));
  TSDB_CHECK_NULL(gStreamMgmt.vgLeaders, code, lino, _exit, terrno);

  gStreamMgmt.taskMap = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
  TSDB_CHECK_NULL(gStreamMgmt.taskMap, code, lino, _exit, terrno);

  gStreamMgmt.vgroupMap = taosHashInit(20, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_ENTRY_LOCK);
  TSDB_CHECK_NULL(gStreamMgmt.vgroupMap, code, lino, _exit, terrno);

//  gStreamMgmt.snodeTasks = taosArrayInit(20, POINTER_BYTES);
//  TSDB_CHECK_NULL(gStreamMgmt.snodeTasks, code, lino, _exit, terrno);
  
  TAOS_CHECK_EXIT(streamTimerInit(&gStreamMgmt.timer));

  TAOS_CHECK_EXIT(streamHbInit(&gStreamMgmt.hb));

  TAOS_CHECK_EXIT(stTriggerTaskEnvInit());

  TAOS_CHECK_EXIT(initInserterGrpInfo());

  TAOS_CHECK_EXIT(initStreamDataSink());

_exit:

  if (code) {
    terrno = code;
    stError("%s failed at line %d, error:%s", __func__, lino, tstrerror(code));
  }

  return code;
}

int32_t streamVgIdSort(void const *lp, void const *rp) {
  int32_t* pVg1 = (int32_t*)lp;
  int32_t* pVg2 = (int32_t*)rp;

  if (*pVg1 < *pVg2) {
    return -1;
  } else if (*pVg1 > *pVg2) {
    return 1;
  }

  return 0;
}


void streamRemoveVnodeLeader(int32_t vgId) {
  if (tsDisableStream) {
    return;
  }
  
  taosWLockLatch(&gStreamMgmt.vgLeadersLock);
  int32_t idx = taosArraySearchIdx(gStreamMgmt.vgLeaders, &vgId, streamVgIdSort, TD_EQ);
  if (idx >= 0) {
    taosArrayRemove(gStreamMgmt.vgLeaders, idx);
  }
  taosWUnLockLatch(&gStreamMgmt.vgLeadersLock);
  
  if (idx >= 0) {
    stInfo("remove vgroup %d from vgroupLeaders succeed", vgId);
  } else {
    stWarn("remove vgroup %d from vgroupLeaders failed since not exists", vgId);
  }

  smUndeployVgTasks(vgId, false);
}

void streamAddVnodeLeader(int32_t vgId) {
  if (tsDisableStream) {
    return;
  }
  
  int32_t code = TSDB_CODE_SUCCESS;
  taosWLockLatch(&gStreamMgmt.vgLeadersLock);
  void* p = taosArrayPush(gStreamMgmt.vgLeaders, &vgId);
  if (p) {
    taosArraySort(gStreamMgmt.vgLeaders, streamVgIdSort);
  } else {
    code = terrno;
  }
  taosWUnLockLatch(&gStreamMgmt.vgLeadersLock);

  smEnableVgDeploy(vgId);
  
  if (p) {
    stInfo("add vgroup %d to vgroupLeaders succeed", vgId);
  } else {
    stError("add vgroup %d to vgroupLeaders failed, error:%s", vgId, tstrerror(code));
  }
}

int32_t streamAcquireTask(int64_t streamId, int64_t taskId, SStreamTask** ppTask, void** ppAddr) {
  int64_t key[2] = {streamId, taskId};

  SStreamTask** task = taosHashAcquire(gStreamMgmt.taskMap, key, sizeof(key));
  if (NULL == task) {
    stsWarn("task %" PRIx64 " not exists in taskMap", taskId);
    return TSDB_CODE_STREAM_TASK_NOT_EXIST;
  }

  SStreamTask* pTask = *task;
  if (taosRTryLockLatch(&pTask->entryLock)) {
    ST_TASK_DLOG("task entry lock failed since task dropping, entryLock:%x", pTask->entryLock);
    taosHashRelease(gStreamMgmt.taskMap, task);
    return TSDB_CODE_STREAM_TASK_NOT_EXIST;
  }

  *ppTask = pTask;
  *ppAddr = (void*)task;

  return TSDB_CODE_SUCCESS;
}

void streamReleaseTask(void* taskAddr) {
  if (NULL == taskAddr) {
    return;
  }
  
  SStreamTask* pTask = *(SStreamTask**)taskAddr;
  SRWLatch lock = taosRUnLockLatch_r(&pTask->entryLock);
  if (taosIsOnlyWLocked(&lock)) {
    switch (pTask->type) {
      case STREAM_READER_TASK:
        stReaderTaskUndeploy((SStreamReaderTask**)taskAddr, true);
        break;
      case STREAM_TRIGGER_TASK:
        stTriggerTaskUndeploy((SStreamTriggerTask**)taskAddr, true);
        break;
      case STREAM_RUNNER_TASK:
        stRunnerTaskUndeploy((SStreamRunnerTask**)taskAddr, true);
        break;
      default:
        break;
    }
  }
  
  taosHashRelease(gStreamMgmt.taskMap, taskAddr);
}

int32_t streamAcquireTriggerTask(int64_t streamId, SStreamTask** ppTask, void** ppAddr) {
  int32_t gid = STREAM_GID(streamId);
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  SHashObj* pGrp = gStreamMgmt.stmGrp[gid];

  SStreamInfo* pStream = taosHashAcquire(pGrp, &streamId, sizeof(streamId));
  if (NULL == pStream) {
    stsWarn("stream %" PRIx64 " not exists when acquire trigger task", streamId);
    return TSDB_CODE_MND_STREAM_NOT_EXIST;
  }

  TAOS_CHECK_EXIT(streamAcquireTask(streamId, pStream->triggerTaskId, ppTask, ppAddr));

_exit:

  taosHashRelease(pGrp, pStream);

  if (code) {
    stsWarn("%s failed at line %d, error:%s", __FUNCTION__, lino, tstrerror(code));
  }

  return code;
}


void streamHandleTaskError(int64_t streamId, int64_t taskId, int32_t errCode) {
  int64_t key[2] = {streamId, taskId};

  SStreamTask** task = taosHashGet(gStreamMgmt.taskMap, key, sizeof(key));
  if (NULL == task) {
    stError("stream %" PRIx64 " task %" PRIx64 " not exists in taskMap", streamId, taskId);
    return;
  }

  atomic_store_32(&(*task)->errorCode, errCode);
  atomic_store_32((int32_t*)&(*task)->status, STREAM_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 "executor.h"
17	#include "streamInt.h"
18	#include "tmisce.h"
19	#include "tref.h"
20	#include "stream.h"
21	#include "ttimer.h"
22	#include "dataSink.h"
23
24	SStreamMgmtInfo gStreamMgmt = {0};
25
26	void streamSetSnodeEnabled( SMsgCb* msgCb) {	77,719✔
27	if (tsDisableStream) {	77,719✔
28	return;	×
29	}
30
31	gStreamMgmt.snodeEnabled = true;	77,719✔
32	gStreamMgmt.msgCb = *msgCb;	77,719✔
33	stInfo("snode %d enabled", (*gStreamMgmt.getDnode)(gStreamMgmt.dnode));	77,719✔
34	}
35
36	void streamSetSnodeDisabled(bool cleanup) {	77,719✔
37	if (tsDisableStream) {	77,719✔
38	return;	×
39	}
40
41	stInfo("snode disabled");	77,719✔
42	gStreamMgmt.snodeEnabled = false;	77,719✔
43	smUndeploySnodeTasks(cleanup);	77,719✔
44	}
45
46	void streamMgmtCleanup() {	645,321✔
47	taosArrayDestroy(gStreamMgmt.vgLeaders);	645,321✔
48	gStreamMgmt.vgLeaders = NULL;	645,321✔
49	taosHashCleanup(gStreamMgmt.taskMap);	645,321✔
50	gStreamMgmt.taskMap = NULL;	645,321✔
51	taosHashCleanup(gStreamMgmt.vgroupMap);	645,321✔
52	gStreamMgmt.vgroupMap = NULL;	645,321✔
53	for (int32_t i = 0; i < STREAM_MAX_GROUP_NUM; ++i) {	3,871,926✔
54	taosHashCleanup(gStreamMgmt.stmGrp[i]);	3,226,605✔
55	gStreamMgmt.stmGrp[i] = NULL;	3,226,605✔
56	}
57	}	645,321✔
58
59	void streamCleanup(void) {	645,321✔
60	if (tsDisableStream) {	645,321✔
61	return;	×
62	}
63
64	stInfo("stream cleanup start");	645,321✔
65	stTriggerTaskEnvCleanup();	645,321✔
66	streamTimerCleanUp();	645,321✔
67	smUndeployAllTasks();	645,321✔
68	destroyDataSinkMgr();	645,321✔
69	streamMgmtCleanup();	645,321✔
70	destroyInserterGrpInfo();	645,321✔
71	stInfo("stream cleanup end");	645,321✔
72	}
73
74	int32_t streamInit(void* pDnode, getDnodeId_f getDnode, getMnodeEpset_f getMnode, getSynEpset_f getSynEpset) {	645,321✔
75	if (tsDisableStream) {	645,321✔
76	stInfo("stream disabled");	×
77	return TSDB_CODE_SUCCESS;	×
78	}
79
80	int32_t code = TSDB_CODE_SUCCESS;	645,321✔
81	int32_t lino = 0;	645,321✔
82
83	gStreamMgmt.dnode = pDnode;	645,321✔
84	gStreamMgmt.getMnode = getMnode;	645,321✔
85	gStreamMgmt.getDnode = getDnode;	645,321✔
86	gStreamMgmt.getSynEpset = getSynEpset;	645,321✔
87
88	gStreamMgmt.vgLeaders = taosArrayInit(20, sizeof(int32_t));	645,321✔
89	TSDB_CHECK_NULL(gStreamMgmt.vgLeaders, code, lino, _exit, terrno);	645,321✔
90
91	gStreamMgmt.taskMap = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);	645,321✔
92	TSDB_CHECK_NULL(gStreamMgmt.taskMap, code, lino, _exit, terrno);	645,321✔
93
94	gStreamMgmt.vgroupMap = taosHashInit(20, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_ENTRY_LOCK);	645,321✔
95	TSDB_CHECK_NULL(gStreamMgmt.vgroupMap, code, lino, _exit, terrno);	645,321✔
96
97	// gStreamMgmt.snodeTasks = taosArrayInit(20, POINTER_BYTES);
98	// TSDB_CHECK_NULL(gStreamMgmt.snodeTasks, code, lino, _exit, terrno);
99
100	TAOS_CHECK_EXIT(streamTimerInit(&gStreamMgmt.timer));	645,321✔
101
102	TAOS_CHECK_EXIT(streamHbInit(&gStreamMgmt.hb));	645,321✔
103
104	TAOS_CHECK_EXIT(stTriggerTaskEnvInit());	645,321✔
105
106	TAOS_CHECK_EXIT(initInserterGrpInfo());	645,321✔
107
108	TAOS_CHECK_EXIT(initStreamDataSink());	645,321✔
109
110	_exit:	645,321✔
111
112	if (code) {	645,321✔
113	terrno = code;	×
114	stError("%s failed at line %d, error:%s", __func__, lino, tstrerror(code));	×
115	}
116
117	return code;	645,321✔
118	}
119
120	int32_t streamVgIdSort(void const lp, void const rp) {	42,147,470✔
121	int32_t* pVg1 = (int32_t*)lp;	42,147,470✔
122	int32_t* pVg2 = (int32_t*)rp;	42,147,470✔
123
124	if (pVg1 < pVg2) {	42,147,470✔
125	return -1;	32,216,229✔
126	} else if (pVg1 > pVg2) {	9,931,241✔
127	return 1;	6,033,485✔
128	}
129
130	return 0;	3,897,756✔
131	}
132
133
134	void streamRemoveVnodeLeader(int32_t vgId) {	7,289,533✔
135	if (tsDisableStream) {	7,289,533✔
136	return;	×
137	}
138
139	taosWLockLatch(&gStreamMgmt.vgLeadersLock);	7,289,533✔
140	int32_t idx = taosArraySearchIdx(gStreamMgmt.vgLeaders, &vgId, streamVgIdSort, TD_EQ);	7,290,275✔
141	if (idx >= 0) {	7,290,275✔
142	taosArrayRemove(gStreamMgmt.vgLeaders, idx);	3,897,738✔
143	}
144	taosWUnLockLatch(&gStreamMgmt.vgLeadersLock);	7,290,275✔
145
146	if (idx >= 0) {	7,290,275✔
147	stInfo("remove vgroup %d from vgroupLeaders succeed", vgId);	3,897,738✔
148	} else {
149	stWarn("remove vgroup %d from vgroupLeaders failed since not exists", vgId);	3,392,537✔
150	}
151
152	smUndeployVgTasks(vgId, false);	7,290,275✔
153	}
154
155	void streamAddVnodeLeader(int32_t vgId) {	3,897,756✔
156	if (tsDisableStream) {	3,897,756✔
157	return;	×
158	}
159
160	int32_t code = TSDB_CODE_SUCCESS;	3,897,756✔
161	taosWLockLatch(&gStreamMgmt.vgLeadersLock);	3,897,756✔
162	void* p = taosArrayPush(gStreamMgmt.vgLeaders, &vgId);	3,897,756✔
163	if (p) {	3,897,756✔
164	taosArraySort(gStreamMgmt.vgLeaders, streamVgIdSort);	3,897,756✔
165	} else {
166	code = terrno;	×
167	}
168	taosWUnLockLatch(&gStreamMgmt.vgLeadersLock);	3,897,756✔
169
170	smEnableVgDeploy(vgId);	3,897,756✔
171
172	if (p) {	3,897,713✔
173	stInfo("add vgroup %d to vgroupLeaders succeed", vgId);	3,897,713✔
174	} else {
175	stError("add vgroup %d to vgroupLeaders failed, error:%s", vgId, tstrerror(code));	×
176	}
177	}
178
179	int32_t streamAcquireTask(int64_t streamId, int64_t taskId, SStreamTask ppTask, void ppAddr) {	130,849,372✔
180	int64_t key[2] = {streamId, taskId};	130,849,372✔
181
182	SStreamTask** task = taosHashAcquire(gStreamMgmt.taskMap, key, sizeof(key));	130,854,864✔
183	if (NULL == task) {	130,836,277✔
184	stsWarn("task %" PRIx64 " not exists in taskMap", taskId);	26,809✔
185	return TSDB_CODE_STREAM_TASK_NOT_EXIST;	26,809✔
186	}
187
188	SStreamTask* pTask = *task;	130,809,468✔
189	if (taosRTryLockLatch(&pTask->entryLock)) {	130,822,304✔
UNCOV 190	ST_TASK_DLOG("task entry lock failed since task dropping, entryLock:%x", pTask->entryLock);	×
UNCOV 191	taosHashRelease(gStreamMgmt.taskMap, task);	×
192	return TSDB_CODE_STREAM_TASK_NOT_EXIST;	×
193	}
194
195	*ppTask = pTask;	130,822,123✔
196	ppAddr = (void)task;	130,816,530✔
197
198	return TSDB_CODE_SUCCESS;	130,822,058✔
199	}
200
201	void streamReleaseTask(void* taskAddr) {	130,833,934✔
202	if (NULL == taskAddr) {	130,833,934✔
203	return;	20,094✔
204	}
205
206	SStreamTask* pTask = (SStreamTask*)taskAddr;	130,813,840✔
207	SRWLatch lock = taosRUnLockLatch_r(&pTask->entryLock);	130,827,075✔
208	if (taosIsOnlyWLocked(&lock)) {	130,828,885✔
209	switch (pTask->type) {	3,939✔
210	case STREAM_READER_TASK:	60✔
211	stReaderTaskUndeploy((SStreamReaderTask**)taskAddr, true);	60✔
212	break;	60✔
213	case STREAM_TRIGGER_TASK:	506✔
214	stTriggerTaskUndeploy((SStreamTriggerTask**)taskAddr, true);	506✔
215	break;	506✔
216	case STREAM_RUNNER_TASK:	3,373✔
217	stRunnerTaskUndeploy((SStreamRunnerTask**)taskAddr, true);	3,373✔
218	break;	3,373✔
219	default:	×
220	break;	×
221	}
222	}
223
224	taosHashRelease(gStreamMgmt.taskMap, taskAddr);	130,824,068✔
225	}
226
227	int32_t streamAcquireTriggerTask(int64_t streamId, SStreamTask ppTask, void ppAddr) {	5,087✔
228	int32_t gid = STREAM_GID(streamId);	5,087✔
229	int32_t code = TSDB_CODE_SUCCESS;	5,087✔
230	int32_t lino = 0;	5,087✔
231	SHashObj* pGrp = gStreamMgmt.stmGrp[gid];	5,087✔
232
233	SStreamInfo* pStream = taosHashAcquire(pGrp, &streamId, sizeof(streamId));	5,087✔
234	if (NULL == pStream) {	5,087✔
235	stsWarn("stream %" PRIx64 " not exists when acquire trigger task", streamId);	×
236	return TSDB_CODE_MND_STREAM_NOT_EXIST;	×
237	}
238
239	TAOS_CHECK_EXIT(streamAcquireTask(streamId, pStream->triggerTaskId, ppTask, ppAddr));	5,087✔
240
241	_exit:	4,152✔
242
243	taosHashRelease(pGrp, pStream);	5,087✔
244
245	if (code) {	5,087✔
246	stsWarn("%s failed at line %d, error:%s", __FUNCTION__, lino, tstrerror(code));	935✔
247	}
248
249	return code;	5,087✔
250	}
251
252
253	void streamHandleTaskError(int64_t streamId, int64_t taskId, int32_t errCode) {	214,148✔
254	int64_t key[2] = {streamId, taskId};	214,148✔
255
256	SStreamTask** task = taosHashGet(gStreamMgmt.taskMap, key, sizeof(key));	214,148✔
257	if (NULL == task) {	214,148✔
258	stError("stream %" PRIx64 " task %" PRIx64 " not exists in taskMap", streamId, taskId);	39,040✔
259	return;	39,040✔
260	}
261
262	atomic_store_32(&(*task)->errorCode, errCode);	175,108✔
263	atomic_store_32((int32_t)&(task)->status, STREAM_STATUS_FAILED);	175,108✔
264	}
265

taosdata / TDengine / #5011

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