• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

taosdata / TDengine / #4506

15 Jul 2025 12:33AM UTC coverage: 62.026% (-0.7%) from 62.706%
#4506

push

travis-ci

web-flow
docs: update stream docs (#31874)

155391 of 320094 branches covered (48.55%)

Branch coverage included in aggregate %.

240721 of 318525 relevant lines covered (75.57%)

6529048.03 hits per line

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

52.32
/source/libs/sync/src/syncIndexMgr.c
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 "syncIndexMgr.h"
18
#include "syncUtil.h"
19

20
SSyncIndexMgr *syncIndexMgrCreate(SSyncNode *pNode) {
34,080✔
21
  SSyncIndexMgr *pIndexMgr = taosMemoryCalloc(1, sizeof(SSyncIndexMgr));
34,080!
22
  if (pIndexMgr == NULL) {
34,083!
23
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
24
    return NULL;
×
25
  }
26

27
  pIndexMgr->replicas = &pNode->replicasId;
34,083✔
28
  pIndexMgr->replicaNum = pNode->replicaNum;
34,083✔
29
  pIndexMgr->totalReplicaNum = pNode->totalReplicaNum;
34,083✔
30
  pIndexMgr->pNode = pNode;
34,083✔
31
  syncIndexMgrClear(pIndexMgr);
34,083✔
32

33
  return pIndexMgr;
34,083✔
34
}
35

36
void syncIndexMgrUpdate(SSyncIndexMgr *pIndexMgr, SSyncNode *pNode) {
718✔
37
  pIndexMgr->replicas = &pNode->replicasId;
718✔
38
  pIndexMgr->replicaNum = pNode->replicaNum;
718✔
39
  pIndexMgr->totalReplicaNum = pNode->totalReplicaNum;
718✔
40
  pIndexMgr->pNode = pNode;
718✔
41
  syncIndexMgrClear(pIndexMgr);
718✔
42
}
718✔
43

44
void syncIndexMgrDestroy(SSyncIndexMgr *pIndexMgr) {
34,072✔
45
  if (pIndexMgr != NULL) {
34,072!
46
    taosMemoryFree(pIndexMgr);
34,074!
47
  }
48
}
34,069✔
49

50
void syncIndexMgrClear(SSyncIndexMgr *pIndexMgr) {
34,802✔
51
  memset(pIndexMgr->index, 0, sizeof(pIndexMgr->index));
34,802✔
52
  memset(pIndexMgr->privateTerm, 0, sizeof(pIndexMgr->privateTerm));
34,802✔
53

54
  int64_t timeNow = taosGetTimestampMs();
34,801✔
55
  for (int i = 0; i < pIndexMgr->totalReplicaNum; ++i) {
89,910✔
56
    pIndexMgr->startTimeArr[i] = 0;
55,109✔
57
    pIndexMgr->recvHBTimeArr[i] = timeNow;
55,109✔
58
    pIndexMgr->recvHBCountArr[i] = 0;
55,109✔
59
  }
60
}
34,801✔
61

62
void syncIndexMgrSetIndex(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, SyncIndex index) {
3,418,542✔
63
  for (int i = 0; i < pIndexMgr->totalReplicaNum; ++i) {
4,583,662!
64
    if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) {
4,583,913✔
65
      (pIndexMgr->index)[i] = index;
3,418,962✔
66
      return;
3,418,962✔
67
    }
68
  }
69

70
  sError("vgId:%d, indexmgr set index:%" PRId64 " for dnode:%d cluster:%d failed", pIndexMgr->pNode->vgId, index,
×
71
         DID(pRaftId), CID(pRaftId));
72
}
73

74
void syncIndexMgrCopyIfExist(SSyncIndexMgr * pNewIndex, SSyncIndexMgr * pOldIndex, SRaftId *oldReplicasId){
×
75
  for(int j = 0; j < pOldIndex->totalReplicaNum; ++j){
×
76
    sDebug("old Index j:%d, index:%"PRId64, j, pOldIndex->index[j]);
×
77
  }
78
  
79
  for (int i = 0; i < pNewIndex->totalReplicaNum; ++i) {
×
80
    for(int j = 0; j < pOldIndex->totalReplicaNum; ++j){
×
81
      if (syncUtilSameId(/*(const SRaftId*)*/&((oldReplicasId[j])), &((*(pNewIndex->replicas))[i]))) {
×
82
        pNewIndex->index[i] = pOldIndex->index[j];
×
83
        pNewIndex->privateTerm[i] = pOldIndex->privateTerm[j];
×
84
        pNewIndex->startTimeArr[i] = pOldIndex->startTimeArr[j];
×
85
        pNewIndex->recvHBTimeArr[i] = pOldIndex->recvHBTimeArr[j];
×
86
        pNewIndex->recvHBCountArr[i] = pOldIndex->recvHBCountArr[j];   
×
87
      }
88
    }
89
  }
90

91
  for (int i = 0; i < pNewIndex->totalReplicaNum; ++i){
×
92
    sDebug("new index i:%d, index:%"PRId64, i, pNewIndex->index[i]);
×
93
  }
94
}
×
95

96
SSyncLogReplMgr *syncNodeGetLogReplMgr(SSyncNode *pNode, SRaftId *pRaftId) {
1,001,399✔
97
  for (int i = 0; i < pNode->totalReplicaNum; i++) {
2,182,912✔
98
    if (syncUtilSameId(&pNode->replicasId[i], pRaftId)) {
2,182,911✔
99
      return pNode->logReplMgrs[i];
1,001,399✔
100
    }
101
  }
102

103
  terrno = TSDB_CODE_SYN_INVALID_ID;
1✔
104
  sError("vgId:%d, indexmgr get replmgr from dnode:%d cluster:%d failed", pNode->vgId, DID(pRaftId), CID(pRaftId));
4!
105
  return NULL;
4✔
106
}
107

108
SyncIndex syncIndexMgrGetIndex(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId) {
560,340✔
109
  for (int i = 0; i < pIndexMgr->totalReplicaNum; ++i) {
1,215,371!
110
    if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) {
1,215,373✔
111
      SyncIndex idx = (pIndexMgr->index)[i];
560,341✔
112
      return idx;
560,341✔
113
    }
114
  }
115

116
  terrno = TSDB_CODE_SYN_INVALID_ID;
×
117
  sError("vgId:%d, indexmgr get index from dnode:%d cluster:%d failed", pIndexMgr->pNode->vgId, DID(pRaftId),
×
118
         CID(pRaftId));
119
  return SYNC_INDEX_INVALID;
×
120
}
121

122
void syncIndexMgrSetStartTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, int64_t startTime) {
×
123
  for (int i = 0; i < pIndexMgr->totalReplicaNum; ++i) {
×
124
    if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) {
×
125
      (pIndexMgr->startTimeArr)[i] = startTime;
×
126
      return;
×
127
    }
128
  }
129

130
  sError("vgId:%d, indexmgr set start-time:%" PRId64 " for dnode:%d cluster:%d failed", pIndexMgr->pNode->vgId,
×
131
         startTime, DID(pRaftId), CID(pRaftId));
132
}
133

134
int64_t syncIndexMgrGetStartTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId) {
×
135
  for (int i = 0; i < pIndexMgr->totalReplicaNum; ++i) {
×
136
    if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) {
×
137
      int64_t startTime = (pIndexMgr->startTimeArr)[i];
×
138
      return startTime;
×
139
    }
140
  }
141

142
  sError("vgId:%d, indexmgr get start-time from dnode:%d cluster:%d failed", pIndexMgr->pNode->vgId, DID(pRaftId),
×
143
         CID(pRaftId));
144
  return TSDB_CODE_SYN_INVALID_ID;
×
145
  ;
146
}
147

148
void syncIndexMgrSetRecvTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, int64_t recvTime) {
104,484✔
149
  for (int i = 0; i < pIndexMgr->totalReplicaNum; ++i) {
204,109✔
150
    if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) {
204,091✔
151
      (pIndexMgr->recvHBTimeArr)[i] = recvTime;
104,466✔
152
      return;
104,466✔
153
    }
154
  }
155

156
  sError("vgId:%d, indexmgr set recv-time:%" PRId64 " for dnode:%d cluster:%d failed", pIndexMgr->pNode->vgId, recvTime,
18!
157
         DID(pRaftId), CID(pRaftId));
158
}
159

160
void syncIndexMgrIncRecvCount(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId) {
104,484✔
161
  for (int i = 0; i < pIndexMgr->totalReplicaNum; ++i) {
204,109✔
162
    if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) {
204,091✔
163
      (pIndexMgr->recvHBCountArr)[i] = (pIndexMgr->recvHBCountArr)[i] + 1;
104,466✔
164
      return;
104,466✔
165
    }
166
  }
167

168
  sError("vgId:%d, indexmgr inc recv for dnode:%d cluster:%d failed", pIndexMgr->pNode->vgId, DID(pRaftId), CID(pRaftId));
18!
169
}
170

171
void syncIndexMgrSetSentTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, int64_t sentTime) {
60,664✔
172
  for (int i = 0; i < pIndexMgr->totalReplicaNum; ++i) {
127,156!
173
    if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) {
127,156✔
174
      (pIndexMgr->sentHBTimeArr)[i] = sentTime;
60,664✔
175
      return;
60,664✔
176
    }
177
  }
178

179
  sError("vgId:%d, indexmgr set sent-time:%" PRId64 " for dnode:%d cluster:%d failed", pIndexMgr->pNode->vgId, sentTime,
×
180
         DID(pRaftId), CID(pRaftId));
181
}
182

183
int64_t syncIndexMgrGetRecvTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId) {
911,645✔
184
  for (int i = 0; i < pIndexMgr->totalReplicaNum; ++i) {
1,720,628!
185
    if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) {
1,720,631✔
186
      int64_t recvTime = (pIndexMgr->recvHBTimeArr)[i];
911,644✔
187
      return recvTime;
911,644✔
188
    }
189
  }
190

191
  sError("vgId:%d, indexmgr get recv-time from dnode:%d cluster:%d failed", pIndexMgr->pNode->vgId, DID(pRaftId),
×
192
         CID(pRaftId));
193
  return TSDB_CODE_SYN_INVALID_ID;
18✔
194
}
195

196
int64_t syncIndexMgrGetRecvCount(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId) {
607,069✔
197
  for (int i = 0; i < pIndexMgr->totalReplicaNum; ++i) {
1,001,894!
198
    if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) {
1,001,896✔
199
      int64_t count = (pIndexMgr->recvHBCountArr)[i];
607,065✔
200
      return count;
607,065✔
201
    }
202
  }
203

204
  sError("vgId:%d, indexmgr get recv-time from dnode:%d cluster:%d failed", pIndexMgr->pNode->vgId, DID(pRaftId),
×
205
         CID(pRaftId));
206
  return TSDB_CODE_SYN_INVALID_ID;
×
207
}
208

209
int64_t syncIndexMgrGetSentTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId) {
202,369✔
210
  for (int i = 0; i < pIndexMgr->totalReplicaNum; ++i) {
333,989✔
211
    if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) {
333,987✔
212
      int64_t recvTime = (pIndexMgr->sentHBTimeArr)[i];
202,371✔
213
      return recvTime;
202,371✔
214
    }
215
  }
216

217
  sError("vgId:%d, indexmgr get sent-time from dnode:%d cluster:%d failed", pIndexMgr->pNode->vgId, DID(pRaftId),
2!
218
         CID(pRaftId));
219
  return TSDB_CODE_SYN_INVALID_ID;
×
220
}
221

222
void syncIndexMgrSetTerm(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, SyncTerm term) {
×
223
  for (int i = 0; i < pIndexMgr->totalReplicaNum; ++i) {
×
224
    if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) {
×
225
      (pIndexMgr->privateTerm)[i] = term;
×
226
      return;
×
227
    }
228
  }
229

230
  sError("vgId:%d, indexmgr set term:%" PRId64 " for dnode:%d cluster:%d failed", pIndexMgr->pNode->vgId, term,
×
231
         DID(pRaftId), CID(pRaftId));
232
}
233

234
SyncTerm syncIndexMgrGetTerm(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId) {
×
235
  for (int i = 0; i < pIndexMgr->totalReplicaNum; ++i) {
×
236
    if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) {
×
237
      SyncTerm term = (pIndexMgr->privateTerm)[i];
×
238
      return term;
×
239
    }
240
  }
241

242
  sError("vgId:%d, indexmgr get term from dnode:%d cluster:%d failed", pIndexMgr->pNode->vgId, DID(pRaftId),
×
243
         CID(pRaftId));
244
  return TSDB_CODE_SYN_INVALID_ID;
×
245
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc