#5056

Committed 17 May 2026 01:15AM UTC coverage: 73.384% (+0.03%) from 73.355%

Build # #5056

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat (TDgpt): Dynamic Model Synchronization Enhancements (#35344)

* refactor: do some internal refactor.

* fix: fix multiprocess sync issue.

* feat: add dynamic anomaly detection and forecasting services

* fix: log error message for undeploying model in exception handling

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* fix: handle undeploy when model exists only on disk

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/286aafa0-c3ce-4c27-b803-2707571e9dc1

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: guard dynamic registry concurrent access

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/5e4db858-6458-40f4-ac28-d1b1b7f97c18

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: tighten service list locking scope

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/5e4db858-6458-40f4-ac28-d1b1b7f97c18

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: restore prophet support and update tests per review feedback

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/92298ae1-7da6-4d07-b20e-101c7cd0b26b

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: improve test name and move copy inside lock scope

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/92298ae1-7da6-4d07-b20e-101c7cd0b26b

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* Potential fix for pull request finding

Co-au... (continued)

Coverage Stats

281643 of 383795 relevant lines covered (73.38%)

135942701.67 hits per line

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

83.13

/source/libs/sync/src/syncEnv.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/>.
 */

#define _DEFAULT_SOURCE
#include "syncEnv.h"
#include "syncUtil.h"
#include "tref.h"

static SSyncEnv gSyncEnv = {0};
static int32_t  gNodeRefId = -1;
static int32_t  gHbDataRefId = -1;

SSyncEnv *syncEnv() { return &gSyncEnv; }

bool syncIsInit() { return atomic_load_8(&gSyncEnv.isStart); }

int32_t syncInit() {
  if (syncIsInit()) return 0;

  uint32_t seed = (uint32_t)(taosGetTimestampNs() & 0x00000000FFFFFFFF);
  taosSeedRand(seed);

  (void)memset(&gSyncEnv, 0, sizeof(SSyncEnv));
  gSyncEnv.pTimerManager = taosTmrInit(1000, 50, 10000, "SYNC-ENV");

  gNodeRefId = taosOpenRef(200, (RefFp)syncNodeClose);
  if (gNodeRefId < 0) {
    sError("failed to init node rset");
    syncCleanUp();
    return TSDB_CODE_SYN_WRONG_REF;
  }
  sDebug("sync node rset is open, rsetId:%d", gNodeRefId);

  gHbDataRefId = taosOpenRef(200, (RefFp)syncHbTimerDataFree);
  if (gHbDataRefId < 0) {
    sError("failed to init hbdata rset");
    syncCleanUp();
    return TSDB_CODE_SYN_WRONG_REF;
  }

  sDebug("sync hbdata rset is open, rsetId:%d", gHbDataRefId);

  atomic_store_8(&gSyncEnv.isStart, 1);
  return 0;
}

void syncCleanUp() {
  atomic_store_8(&gSyncEnv.isStart, 0);
  taosTmrCleanUp(gSyncEnv.pTimerManager);
  (void)memset(&gSyncEnv, 0, sizeof(SSyncEnv));

  if (gNodeRefId != -1) {
    sDebug("sync node rset is closed, rsetId:%d", gNodeRefId);
    taosCloseRef(gNodeRefId);
    gNodeRefId = -1;
  }

  if (gHbDataRefId != -1) {
    sDebug("sync hbdata rset is closed, rsetId:%d", gHbDataRefId);
    taosCloseRef(gHbDataRefId);
    gHbDataRefId = -1;
  }
}

int64_t syncNodeAdd(SSyncNode *pNode) {
  pNode->rid = taosAddRef(gNodeRefId, pNode);
  if (pNode->rid < 0) {
    return terrno = TSDB_CODE_SYN_WRONG_REF;
  }

  sDebug("vgId:%d, sync node refId:%" PRId64 " is added to rsetId:%d", pNode->vgId, pNode->rid, gNodeRefId);
  return pNode->rid;
}

void syncNodeRemove(int64_t rid) {
  sDebug("sync node refId:%" PRId64 " is removed from rsetId:%d", rid, gNodeRefId);
  if (rid > 0) {
    int32_t code = 0;
    if ((code = taosRemoveRef(gNodeRefId, rid)) != 0)
      sError("failed to remove sync node from refId:%" PRId64 ", rsetId:%d, since %s", rid, gNodeRefId,
             tstrerror(code));
  }
}

SSyncNode *syncNodeAcquire(int64_t rid) {
  SSyncNode *pNode = taosAcquireRef(gNodeRefId, rid);
  if (pNode == NULL) {
    sError("failed to acquire sync node from refId:%" PRId64 ", rsetId:%d", rid, gNodeRefId);
    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
  }

  return pNode;
}

void syncNodeRelease(SSyncNode *pNode) {
  if (pNode) {
    int32_t code = 0;
    if ((code = taosReleaseRef(gNodeRefId, pNode->rid)) != 0)
      sError("failed to release sync node from refId:%" PRId64 ", rsetId:%d, since %s", pNode->rid, gNodeRefId,
             tstrerror(code));
  }
}

int64_t syncHbTimerDataAdd(SSyncHbTimerData *pData) {
  pData->rid = taosAddRef(gHbDataRefId, pData);
  if (pData->rid < 0) {
    return terrno = TSDB_CODE_SYN_WRONG_REF;
  }

  return pData->rid;
}

void syncHbTimerDataRemove(int64_t rid) {
  if (rid > 0) {
    int32_t code = 0;
    if ((code = taosRemoveRef(gHbDataRefId, rid)) != 0)
      sError("failed to remove hbdata from refId:%" PRId64 ", rsetId:%d, since %s", rid, gHbDataRefId, tstrerror(code));
  }
}

SSyncHbTimerData *syncHbTimerDataAcquire(int64_t rid) {
  SSyncHbTimerData *pData = taosAcquireRef(gHbDataRefId, rid);
  if (pData == NULL && rid > 0) {
    sInfo("failed to acquire hbdata from refId:%" PRId64 ", rsetId:%d", rid, gHbDataRefId);
    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
  }

  return pData;
}

void syncHbTimerDataRelease(SSyncHbTimerData *pData) {
  if (pData) {
    int32_t code = 0;
    if ((code = taosReleaseRef(gHbDataRefId, pData->rid)) != 0) {
      sError("failed to release hbdata from refId:%" PRId64 ", rsetId:%d, since %s", pData->rid, gHbDataRefId,
             tstrerror(code));
    }
  }
}

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	#define _DEFAULT_SOURCE
17	#include "syncEnv.h"
18	#include "syncUtil.h"
19	#include "tref.h"
20
21	static SSyncEnv gSyncEnv = {0};
22	static int32_t gNodeRefId = -1;
23	static int32_t gHbDataRefId = -1;
24
25	SSyncEnv *syncEnv() { return &gSyncEnv; }	533,380,025✔
26
27	bool syncIsInit() { return atomic_load_8(&gSyncEnv.isStart); }	1,912,422,812✔
28
29	int32_t syncInit() {	1,282,660✔
30	if (syncIsInit()) return 0;	1,282,660✔
31
32	uint32_t seed = (uint32_t)(taosGetTimestampNs() & 0x00000000FFFFFFFF);	743,566✔
33	taosSeedRand(seed);	743,566✔
34
35	(void)memset(&gSyncEnv, 0, sizeof(SSyncEnv));	743,566✔
36	gSyncEnv.pTimerManager = taosTmrInit(1000, 50, 10000, "SYNC-ENV");	743,566✔
37
38	gNodeRefId = taosOpenRef(200, (RefFp)syncNodeClose);	743,566✔
39	if (gNodeRefId < 0) {	743,566✔
40	sError("failed to init node rset");	×
41	syncCleanUp();	×
42	return TSDB_CODE_SYN_WRONG_REF;	×
43	}
44	sDebug("sync node rset is open, rsetId:%d", gNodeRefId);	743,566✔
45
46	gHbDataRefId = taosOpenRef(200, (RefFp)syncHbTimerDataFree);	743,566✔
47	if (gHbDataRefId < 0) {	743,566✔
48	sError("failed to init hbdata rset");	×
49	syncCleanUp();	×
50	return TSDB_CODE_SYN_WRONG_REF;	×
51	}
52
53	sDebug("sync hbdata rset is open, rsetId:%d", gHbDataRefId);	743,566✔
54
55	atomic_store_8(&gSyncEnv.isStart, 1);	743,566✔
56	return 0;	743,566✔
57	}
58
59	void syncCleanUp() {	743,736✔
60	atomic_store_8(&gSyncEnv.isStart, 0);	743,736✔
61	taosTmrCleanUp(gSyncEnv.pTimerManager);	743,736✔
62	(void)memset(&gSyncEnv, 0, sizeof(SSyncEnv));	743,736✔
63
64	if (gNodeRefId != -1) {	743,736✔
65	sDebug("sync node rset is closed, rsetId:%d", gNodeRefId);	743,497✔
66	taosCloseRef(gNodeRefId);	743,497✔
67	gNodeRefId = -1;	743,497✔
68	}
69
70	if (gHbDataRefId != -1) {	743,736✔
71	sDebug("sync hbdata rset is closed, rsetId:%d", gHbDataRefId);	743,497✔
72	taosCloseRef(gHbDataRefId);	743,497✔
73	gHbDataRefId = -1;	743,497✔
74	}
75	}	743,736✔
76
77	int64_t syncNodeAdd(SSyncNode *pNode) {	5,531,008✔
78	pNode->rid = taosAddRef(gNodeRefId, pNode);	5,531,008✔
79	if (pNode->rid < 0) {	5,530,323✔
80	return terrno = TSDB_CODE_SYN_WRONG_REF;	×
81	}
82
83	sDebug("vgId:%d, sync node refId:%" PRId64 " is added to rsetId:%d", pNode->vgId, pNode->rid, gNodeRefId);	5,530,323✔
84	return pNode->rid;	5,530,323✔
85	}
86
87	void syncNodeRemove(int64_t rid) {	5,530,939✔
88	sDebug("sync node refId:%" PRId64 " is removed from rsetId:%d", rid, gNodeRefId);	5,530,939✔
89	if (rid > 0) {	5,530,939✔
90	int32_t code = 0;	5,530,939✔
91	if ((code = taosRemoveRef(gNodeRefId, rid)) != 0)	5,530,939✔
92	sError("failed to remove sync node from refId:%" PRId64 ", rsetId:%d, since %s", rid, gNodeRefId,	×
93	tstrerror(code));
94	}
95	}	5,530,939✔
96
97	SSyncNode *syncNodeAcquire(int64_t rid) {	2,147,483,647✔
98	SSyncNode *pNode = taosAcquireRef(gNodeRefId, rid);	2,147,483,647✔
99	if (pNode == NULL) {	2,147,483,647✔
100	sError("failed to acquire sync node from refId:%" PRId64 ", rsetId:%d", rid, gNodeRefId);	9,978✔
101	terrno = TSDB_CODE_SYN_INTERNAL_ERROR;	9,978✔
102	}
103
104	return pNode;	2,147,483,647✔
105	}
106
107	void syncNodeRelease(SSyncNode *pNode) {	2,147,483,647✔
108	if (pNode) {	2,147,483,647✔
109	int32_t code = 0;	2,147,483,647✔
110	if ((code = taosReleaseRef(gNodeRefId, pNode->rid)) != 0)	2,147,483,647✔
111	sError("failed to release sync node from refId:%" PRId64 ", rsetId:%d, since %s", pNode->rid, gNodeRefId,	×
112	tstrerror(code));
113	}
114	}	2,147,483,647✔
115
116	int64_t syncHbTimerDataAdd(SSyncHbTimerData *pData) {	1,148,064✔
117	pData->rid = taosAddRef(gHbDataRefId, pData);	1,148,064✔
118	if (pData->rid < 0) {	1,148,064✔
119	return terrno = TSDB_CODE_SYN_WRONG_REF;	×
120	}
121
122	return pData->rid;	1,148,064✔
123	}
124
125	void syncHbTimerDataRemove(int64_t rid) {	12,308,002✔
126	if (rid > 0) {	12,308,002✔
127	int32_t code = 0;	1,147,296✔
128	if ((code = taosRemoveRef(gHbDataRefId, rid)) != 0)	1,147,296✔
129	sError("failed to remove hbdata from refId:%" PRId64 ", rsetId:%d, since %s", rid, gHbDataRefId, tstrerror(code));	×
130	}
131	}	12,308,002✔
132
133	SSyncHbTimerData *syncHbTimerDataAcquire(int64_t rid) {	33,163,287✔
134	SSyncHbTimerData *pData = taosAcquireRef(gHbDataRefId, rid);	33,163,287✔
135	if (pData == NULL && rid > 0) {	33,163,287✔
136	sInfo("failed to acquire hbdata from refId:%" PRId64 ", rsetId:%d", rid, gHbDataRefId);	×
137	terrno = TSDB_CODE_SYN_INTERNAL_ERROR;	×
138	}
139
140	return pData;	33,163,287✔
141	}
142
143	void syncHbTimerDataRelease(SSyncHbTimerData *pData) {	32,015,204✔
144	if (pData) {	32,015,204✔
145	int32_t code = 0;	32,015,204✔
146	if ((code = taosReleaseRef(gHbDataRefId, pData->rid)) != 0) {	32,015,204✔
147	sError("failed to release hbdata from refId:%" PRId64 ", rsetId:%d, since %s", pData->rid, gHbDataRefId,	×
148	tstrerror(code));
149	}
150	}
151	}	32,015,204✔

taosdata / TDengine / #5056

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