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

taosdata / TDengine / #4416

03 Jul 2025 10:49AM UTC coverage: 61.007% (-1.2%) from 62.241%
#4416

push

travis-ci

GitHub
Merge pull request #31575 from taosdata/fix/huoh/taos_log

150735 of 316232 branches covered (47.67%)

Branch coverage included in aggregate %.

233783 of 314057 relevant lines covered (74.44%)

6782670.15 hits per line

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

70.85
/source/libs/sync/src/syncMessage.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 "syncMessage.h"
18
#include "syncRaftEntry.h"
19
#include "syncRaftStore.h"
20

21
int32_t syncBuildTimeout(SRpcMsg* pMsg, ESyncTimeoutType timeoutType, uint64_t logicClock, int32_t timerMS,
75,821✔
22
                         SSyncNode* pNode) {
23
  int32_t bytes = sizeof(SyncTimeout);
75,821✔
24
  pMsg->pCont = rpcMallocCont(bytes);
75,821✔
25
  pMsg->msgType = (timeoutType == SYNC_TIMEOUT_ELECTION) ? TDMT_SYNC_TIMEOUT_ELECTION : TDMT_SYNC_TIMEOUT;
75,821✔
26
  pMsg->contLen = bytes;
75,821✔
27
  if (pMsg->pCont == NULL) {
75,821!
28
    return terrno;
×
29
  }
30

31
  SyncTimeout* pTimeout = pMsg->pCont;
75,821✔
32
  pTimeout->bytes = bytes;
75,821✔
33
  pTimeout->msgType = pMsg->msgType;
75,821✔
34
  pTimeout->vgId = pNode->vgId;
75,821✔
35
  pTimeout->timeoutType = timeoutType;
75,821✔
36
  pTimeout->logicClock = logicClock;
75,821✔
37
  pTimeout->timerMS = timerMS;
75,821✔
38
  pTimeout->timeStamp = taosGetTimestampMs();
75,821✔
39
  pTimeout->data = pNode;
75,821✔
40
  return 0;
75,821✔
41
}
42

43
int32_t syncBuildClientRequest(SRpcMsg* pMsg, const SRpcMsg* pOriginal, uint64_t seqNum, bool isWeak, int32_t vgId) {
318,521✔
44
  int32_t bytes = sizeof(SyncClientRequest) + pOriginal->contLen;
318,521✔
45

46
  pMsg->pCont = rpcMallocCont(bytes);
318,521✔
47
  if (pMsg->pCont == NULL) {
318,521✔
48
    return terrno;
1✔
49
  }
50
  pMsg->msgType = TDMT_SYNC_CLIENT_REQUEST;
318,520✔
51
  pMsg->contLen = bytes;
318,520✔
52

53
  SyncClientRequest* pClientRequest = pMsg->pCont;
318,520✔
54
  pClientRequest->bytes = bytes;
318,520✔
55
  pClientRequest->vgId = vgId;
318,520✔
56
  pClientRequest->msgType = TDMT_SYNC_CLIENT_REQUEST;
318,520✔
57
  pClientRequest->originalRpcType = pOriginal->msgType;
318,520✔
58
  pClientRequest->seqNum = seqNum;
318,520✔
59
  pClientRequest->isWeak = isWeak;
318,520✔
60
  pClientRequest->dataLen = pOriginal->contLen;
318,520✔
61
  memcpy(pClientRequest->data, (char*)pOriginal->pCont, pOriginal->contLen);
318,520✔
62

63
  return 0;
318,520✔
64
}
65

66
int32_t syncBuildClientRequestFromNoopEntry(SRpcMsg* pMsg, const SSyncRaftEntry* pEntry, int32_t vgId) {
×
67
  int32_t bytes = sizeof(SyncClientRequest) + pEntry->bytes;
×
68
  pMsg->pCont = rpcMallocCont(bytes);
×
69
  pMsg->msgType = TDMT_SYNC_CLIENT_REQUEST;
×
70
  pMsg->contLen = bytes;
×
71
  if (pMsg->pCont == NULL) {
×
72
    return terrno;
×
73
  }
74

75
  SyncClientRequest* pClientRequest = pMsg->pCont;
×
76
  pClientRequest->bytes = bytes;
×
77
  pClientRequest->vgId = vgId;
×
78
  pClientRequest->msgType = TDMT_SYNC_CLIENT_REQUEST;
×
79
  pClientRequest->originalRpcType = TDMT_SYNC_NOOP;
×
80
  pClientRequest->dataLen = pEntry->bytes;
×
81
  memcpy(pClientRequest->data, (char*)pEntry, pEntry->bytes);
×
82

83
  return 0;
×
84
}
85

86
int32_t syncBuildRequestVote(SRpcMsg* pMsg, int32_t vgId) {
4,305✔
87
  int32_t bytes = sizeof(SyncRequestVote);
4,305✔
88
  pMsg->pCont = rpcMallocCont(bytes);
4,305✔
89
  pMsg->msgType = TDMT_SYNC_REQUEST_VOTE;
4,305✔
90
  pMsg->contLen = bytes;
4,305✔
91
  if (pMsg->pCont == NULL) {
4,305!
92
    return terrno;
×
93
  }
94

95
  SyncRequestVote* pRequestVote = pMsg->pCont;
4,305✔
96
  pRequestVote->bytes = bytes;
4,305✔
97
  pRequestVote->msgType = TDMT_SYNC_REQUEST_VOTE;
4,305✔
98
  pRequestVote->vgId = vgId;
4,305✔
99
  return 0;
4,305✔
100
}
101

102
int32_t syncBuildRequestVoteReply(SRpcMsg* pMsg, int32_t vgId) {
5,654✔
103
  int32_t bytes = sizeof(SyncRequestVoteReply);
5,654✔
104
  pMsg->pCont = rpcMallocCont(bytes);
5,654✔
105
  pMsg->msgType = TDMT_SYNC_REQUEST_VOTE_REPLY;
5,654✔
106
  pMsg->contLen = bytes;
5,654✔
107
  if (pMsg->pCont == NULL) {
5,654!
108
    return terrno;
×
109
  }
110

111
  SyncRequestVoteReply* pRequestVoteReply = pMsg->pCont;
5,654✔
112
  pRequestVoteReply->bytes = bytes;
5,654✔
113
  pRequestVoteReply->msgType = TDMT_SYNC_REQUEST_VOTE_REPLY;
5,654✔
114
  pRequestVoteReply->vgId = vgId;
5,654✔
115
  return 0;
5,654✔
116
}
117

118
int32_t syncBuildAppendEntries(SRpcMsg* pMsg, int32_t dataLen, int32_t vgId) {
×
119
  int32_t bytes = sizeof(SyncAppendEntries) + dataLen;
×
120
  pMsg->pCont = rpcMallocCont(bytes);
×
121
  pMsg->msgType = TDMT_SYNC_APPEND_ENTRIES;
×
122
  pMsg->contLen = bytes;
×
123
  if (pMsg->pCont == NULL) {
×
124
    return terrno;
×
125
  }
126

127
  SyncAppendEntries* pAppendEntries = pMsg->pCont;
×
128
  pAppendEntries->bytes = bytes;
×
129
  pAppendEntries->vgId = vgId;
×
130
  pAppendEntries->msgType = TDMT_SYNC_APPEND_ENTRIES;
×
131
  pAppendEntries->dataLen = dataLen;
×
132
  return 0;
×
133
}
134

135
int32_t syncBuildAppendEntriesReply(SRpcMsg* pMsg, int32_t vgId) {
479,099✔
136
  int32_t bytes = sizeof(SyncAppendEntriesReply);
479,099✔
137
  pMsg->pCont = rpcMallocCont(bytes);
479,099✔
138
  pMsg->msgType = TDMT_SYNC_APPEND_ENTRIES_REPLY;
479,106✔
139
  pMsg->contLen = bytes;
479,106✔
140
  if (pMsg->pCont == NULL) {
479,106✔
141
    return terrno;
2✔
142
  }
143

144
  SyncAppendEntriesReply* pAppendEntriesReply = pMsg->pCont;
479,104✔
145
  pAppendEntriesReply->bytes = bytes;
479,104✔
146
  pAppendEntriesReply->msgType = TDMT_SYNC_APPEND_ENTRIES_REPLY;
479,104✔
147
  pAppendEntriesReply->vgId = vgId;
479,104✔
148
  return 0;
479,104✔
149
}
150

151
int32_t syncBuildAppendEntriesFromRaftEntry(SSyncNode* pNode, SSyncRaftEntry* pEntry, SyncTerm prevLogTerm,
495,101✔
152
                                            SRpcMsg* pRpcMsg) {
153
  uint32_t dataLen = pEntry->bytes;
495,101✔
154
  uint32_t bytes = sizeof(SyncAppendEntries) + dataLen;
495,101✔
155
  pRpcMsg->info.traceId = pEntry->originRpcTraceId;
495,101✔
156
  pRpcMsg->contLen = bytes;
495,101✔
157
  pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
495,101✔
158
  if (pRpcMsg->pCont == NULL) {
495,093!
159
    return terrno;
×
160
  }
161

162
  SyncAppendEntries* pMsg = pRpcMsg->pCont;
495,093✔
163
  pMsg->bytes = pRpcMsg->contLen;
495,093✔
164
  pMsg->msgType = pRpcMsg->msgType = TDMT_SYNC_APPEND_ENTRIES;
495,093✔
165
  pMsg->dataLen = dataLen;
495,093✔
166

167
  (void)memcpy(pMsg->data, pEntry, dataLen);
495,093✔
168

169
  pMsg->prevLogIndex = pEntry->index - 1;
495,093✔
170
  pMsg->prevLogTerm = prevLogTerm;
495,093✔
171
  pMsg->vgId = pNode->vgId;
495,093✔
172
  pMsg->srcId = pNode->myRaftId;
495,093✔
173
  pMsg->term = raftStoreGetTerm(pNode);
495,093✔
174
  pMsg->commitIndex = pNode->commitIndex;
495,102✔
175
  pMsg->privateTerm = 0;
495,102✔
176
  return 0;
495,102✔
177
}
178

179
int32_t syncBuildHeartbeat(SRpcMsg* pMsg, int32_t vgId) {
67,694✔
180
  int32_t bytes = sizeof(SyncHeartbeat);
67,694✔
181
  pMsg->pCont = rpcMallocCont(bytes);
67,694✔
182
  pMsg->msgType = TDMT_SYNC_HEARTBEAT;
67,694✔
183
  pMsg->contLen = bytes;
67,694✔
184
  if (pMsg->pCont == NULL) {
67,694!
185
    return terrno;
×
186
  }
187

188
  SyncHeartbeat* pHeartbeat = pMsg->pCont;
67,694✔
189
  pHeartbeat->bytes = bytes;
67,694✔
190
  pHeartbeat->msgType = TDMT_SYNC_HEARTBEAT;
67,694✔
191
  pHeartbeat->vgId = vgId;
67,694✔
192
  return 0;
67,694✔
193
}
194

195
int32_t syncBuildHeartbeatReply(SRpcMsg* pMsg, int32_t vgId) {
60,000✔
196
  int32_t bytes = sizeof(SyncHeartbeatReply);
60,000✔
197
  pMsg->pCont = rpcMallocCont(bytes);
60,000✔
198
  pMsg->msgType = TDMT_SYNC_HEARTBEAT_REPLY;
60,000✔
199
  pMsg->contLen = bytes;
60,000✔
200
  if (pMsg->pCont == NULL) {
60,000!
201
    return terrno;
×
202
  }
203

204
  SyncHeartbeatReply* pHeartbeatReply = pMsg->pCont;
60,000✔
205
  pHeartbeatReply->bytes = bytes;
60,000✔
206
  pHeartbeatReply->msgType = TDMT_SYNC_HEARTBEAT_REPLY;
60,000✔
207
  pHeartbeatReply->vgId = vgId;
60,000✔
208
  return 0;
60,000✔
209
}
210

211
int32_t syncBuildSnapshotSend(SRpcMsg* pMsg, int32_t dataLen, int32_t vgId) {
1,283✔
212
  int32_t bytes = sizeof(SyncSnapshotSend) + dataLen;
1,283✔
213
  pMsg->pCont = rpcMallocCont(bytes);
1,283✔
214
  pMsg->msgType = TDMT_SYNC_SNAPSHOT_SEND;
1,283✔
215
  pMsg->contLen = bytes;
1,283✔
216
  if (pMsg->pCont == NULL) {
1,283!
217
    return terrno;
×
218
  }
219

220
  SyncSnapshotSend* pSnapshotSend = pMsg->pCont;
1,283✔
221
  pSnapshotSend->bytes = bytes;
1,283✔
222
  pSnapshotSend->vgId = vgId;
1,283✔
223
  pSnapshotSend->msgType = TDMT_SYNC_SNAPSHOT_SEND;
1,283✔
224
  pSnapshotSend->dataLen = dataLen;
1,283✔
225
  return 0;
1,283✔
226
}
227

228
int32_t syncBuildSnapshotSendRsp(SRpcMsg* pMsg, int32_t dataLen, int32_t vgId) {
1,283✔
229
  int32_t bytes = sizeof(SyncSnapshotRsp) + dataLen;
1,283✔
230
  pMsg->pCont = rpcMallocCont(bytes);
1,283✔
231
  pMsg->msgType = TDMT_SYNC_SNAPSHOT_RSP;
1,283✔
232
  pMsg->contLen = bytes;
1,283✔
233
  if (pMsg->pCont == NULL) {
1,283!
234
    return terrno;
×
235
  }
236

237
  SyncSnapshotRsp* pPreSnapshotRsp = pMsg->pCont;
1,283✔
238
  pPreSnapshotRsp->bytes = bytes;
1,283✔
239
  pPreSnapshotRsp->msgType = TDMT_SYNC_SNAPSHOT_RSP;
1,283✔
240
  pPreSnapshotRsp->vgId = vgId;
1,283✔
241
  return 0;
1,283✔
242
}
243

244
int32_t syncBuildLeaderTransfer(SRpcMsg* pMsg, int32_t vgId) {
1,470✔
245
  int32_t bytes = sizeof(SyncLeaderTransfer);
1,470✔
246
  pMsg->pCont = rpcMallocCont(bytes);
1,470✔
247
  pMsg->msgType = TDMT_SYNC_LEADER_TRANSFER;
1,470✔
248
  pMsg->contLen = bytes;
1,470✔
249
  if (pMsg->pCont == NULL) {
1,470!
250
    return terrno;
×
251
  }
252

253
  SyncLeaderTransfer* pLeaderTransfer = pMsg->pCont;
1,470✔
254
  pLeaderTransfer->bytes = bytes;
1,470✔
255
  pLeaderTransfer->msgType = TDMT_SYNC_LEADER_TRANSFER;
1,470✔
256
  pLeaderTransfer->vgId = vgId;
1,470✔
257
  return 0;
1,470✔
258
}
259

260
int32_t syncBuildLocalCmd(SRpcMsg* pMsg, int32_t vgId) {
59,525✔
261
  int32_t bytes = sizeof(SyncLocalCmd);
59,525✔
262
  pMsg->pCont = rpcMallocCont(bytes);
59,525✔
263
  pMsg->msgType = TDMT_SYNC_LOCAL_CMD;
59,525✔
264
  pMsg->contLen = bytes;
59,525✔
265
  if (pMsg->pCont == NULL) {
59,525!
266
    return terrno;
×
267
  }
268

269
  SyncLocalCmd* pLocalCmd = pMsg->pCont;
59,525✔
270
  pLocalCmd->bytes = bytes;
59,525✔
271
  pLocalCmd->msgType = TDMT_SYNC_LOCAL_CMD;
59,525✔
272
  pLocalCmd->vgId = vgId;
59,525✔
273
  return 0;
59,525✔
274
}
275

276
const char* syncTimerTypeStr(enum ESyncTimeoutType timerType) {
4,554✔
277
  switch (timerType) {
4,554!
278
    case SYNC_TIMEOUT_PING:
4,554✔
279
      return "ping";
4,554✔
280
    case SYNC_TIMEOUT_ELECTION:
×
281
      return "elect";
×
282
    case SYNC_TIMEOUT_HEARTBEAT:
×
283
      return "heartbeat";
×
284
    default:
×
285
      return "unknown";
×
286
  }
287
}
288

289
const char* syncLocalCmdGetStr(ESyncLocalCmd cmd) {
×
290
  switch (cmd) {
×
291
    case SYNC_LOCAL_CMD_STEP_DOWN:
×
292
      return "step-down";
×
293
    case SYNC_LOCAL_CMD_FOLLOWER_CMT:
×
294
      return "follower-commit";
×
295
    default:
×
296
      return "unknown-local-cmd";
×
297
  }
298
}
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