()
| 250 | } |
| 251 | |
| 252 | func (n *Node) runRaftMessages() error { |
| 253 | n.wg.Add(1) |
| 254 | go func() { |
| 255 | ticker := time.NewTicker(time.Second) |
| 256 | defer func() { |
| 257 | ticker.Stop() |
| 258 | n.wg.Done() |
| 259 | }() |
| 260 | |
| 261 | for { |
| 262 | select { |
| 263 | case <-ticker.C: |
| 264 | n.raftNode.Tick() |
| 265 | case rd := <-n.raftNode.Ready(): |
| 266 | // Save to wal and storage first |
| 267 | if !raft.IsEmptySnap(rd.Snapshot) { |
| 268 | if err := n.dataStore.saveSnapshot(rd.Snapshot); err != nil { |
| 269 | n.logger.Error("Failed to save snapshot", zap.Error(err)) |
| 270 | } |
| 271 | } |
| 272 | if err := n.dataStore.wal.Save(rd.HardState, rd.Entries); err != nil { |
| 273 | n.logger.Error("Failed to save to wal", zap.Error(err)) |
| 274 | } |
| 275 | |
| 276 | // Replay the entries into the raft storage |
| 277 | if err := n.applySnapshot(rd.Snapshot); err != nil { |
| 278 | n.logger.Error("Failed to apply snapshot", zap.Error(err)) |
| 279 | } |
| 280 | _ = n.dataStore.raftStorage.Append(rd.Entries) |
| 281 | |
| 282 | for _, msg := range rd.Messages { |
| 283 | if msg.Type == raftpb.MsgApp { |
| 284 | msg.Snapshot.Metadata.ConfState = n.confState |
| 285 | } |
| 286 | } |
| 287 | n.transport.Send(rd.Messages) |
| 288 | |
| 289 | // Apply the committed entries to the state machine |
| 290 | n.applyEntries(rd.CommittedEntries) |
| 291 | if err := n.triggerSnapshotIfNeed(); err != nil { |
| 292 | n.logger.Error("Failed to trigger snapshot", zap.Error(err)) |
| 293 | } |
| 294 | n.raftNode.Advance() |
| 295 | case err := <-n.transport.ErrorC: |
| 296 | n.logger.Fatal("Found transport error", zap.Error(err)) |
| 297 | return |
| 298 | case <-n.shutdown: |
| 299 | n.logger.Info("Shutting down raft node") |
| 300 | return |
| 301 | } |
| 302 | } |
| 303 | }() |
| 304 | return nil |
| 305 | } |
| 306 | |
| 307 | func (n *Node) triggerSnapshotIfNeed() error { |
| 308 | if n.appliedIndex-n.snapshotIndex <= n.snapshotThreshold.Load() { |
no test coverage detected