1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

// Schedule is used to ensure the Tick is performed periodically. This
// function is safe to call multiple times. If the memberlist is already
// scheduled, then it won't do anything.
func (m *Memberlist) schedule() {m.tickerLock.Lock()defer m.tickerLock.Unlock()// 创建probe goroutineif m.config.ProbeInterval > 0 {t := time.NewTicker(m.config.ProbeInterval)go m.triggerFunc(m.config.ProbeInterval, t.C, stopCh, m.probe)m.tickers = append(m.tickers, t)}// 创建push/pull goroutineif m.config.PushPullInterval > 0 {go m.pushPullTrigger(stopCh)}// 创建gossip goroutineif m.config.GossipInterval > 0 && m.config.GossipNodes > 0 {t := time.NewTicker(m.config.GossipInterval)go m.triggerFunc(m.config.GossipInterval, t.C, stopCh, m.gossip)m.tickers = append(m.tickers, t)}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

HANDLE_REMOTE_FAILURE:// Get some random live nodes.m.nodeLock.RLock()kNodes := kRandomNodes(m.config.IndirectChecks, m.nodes, func(n *nodeState) bool {return n.Name == m.config.Name ||n.Name == node.Name ||n.State != StateAlive})m.nodeLock.RUnlock()// Attempt an indirect ping.expectedNacks := 0selfAddr, selfPort = m.getAdvertise()ind := indirectPingReq{SeqNo:      ping.SeqNo,Target:     node.Addr,Port:       node.Port,Node:       node.Name,SourceAddr: selfAddr,SourcePort: selfPort,SourceNode: m.config.Name,}for _, peer := range kNodes {// We only expect nack to be sent from peers who understand// version 4 of the protocol.if ind.Nack = peer.PMax >= 4; ind.Nack {expectedNacks++}if err := m.encodeAndSendMsg(peer.FullAddress(), indirectPingMsg, &ind); err != nil {m.logger.Printf("[ERR] memberlist: Failed to send indirect ping: %s", err)}}// Also make an attempt to contact the node directly over TCP. This// helps prevent confused clients who get isolated from UDP traffic// but can still speak TCP (which also means they can possibly report// misinformation to other nodes via anti-entropy), avoiding flapping in// the cluster.//// This is a little unusual because we will attempt a TCP ping to any// member who understands version 3 of the protocol, regardless of// which protocol version we are speaking. That's why we've included a// config option to turn this off if desired.fallbackCh := make(chan bool, 1)// Wait for the acks or timeout. Note that we don't check the fallback// channel here because we want to issue a warning below if that's the// *only* way we hear back from the peer, so we have to let this time// out first to allow the normal UDP-based acks to come in.select {case v := <-ackCh:if v.Complete == true {return}}// Finally, poll the fallback channel. The timeouts are set such that// the channel will have something or be closed without having to wait// any additional time here.for didContact := range fallbackCh {if didContact {m.logger.Printf("[WARN] memberlist: Was able to connect to %s but other probes failed, network may be misconfigured", node.Name)return}}// Update our self-awareness based on the results of this failed probe.// If we don't have peers who will send nacks then we penalize for any// failed probe as a simple health metric. If we do have peers to nack// verify, then we can use that as a more sophisticated measure of self-// health because we assume them to be working, and they can help us// decide if the probed node was really dead or if it was something wrong// with ourselves.awarenessDelta = 0if expectedNacks > 0 {if nackCount := len(nackCh); nackCount < expectedNacks {awarenessDelta += (expectedNacks - nackCount)}} else {awarenessDelta += 1}// No acks received from target, suspect it as failed.m.logger.Printf("[INFO] memberlist: Suspect %s has failed, no acks received", node.Name)s := suspect{Incarnation: node.Incarnation, Node: node.Name, From: m.config.Name}m.suspectNode(&s)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

// gossip is invoked every GossipInterval period to broadcast our gossip
// messages to a few random nodes.
func (m *Memberlist) gossip() {// Get some random live, suspect, or recently dead nodesm.nodeLock.RLock()kNodes := kRandomNodes(m.config.GossipNodes, m.nodes, func(n *nodeState) bool {if n.Name == m.config.Name {return true}switch n.State {case StateAlive, StateSuspect:return falsecase StateDead:return time.Since(n.StateChange) > m.config.GossipToTheDeadTimedefault:return true}})m.nodeLock.RUnlock()for _, node := range kNodes {// Get any pending broadcastsmsgs := m.getBroadcasts(compoundOverhead, bytesAvail)if len(msgs) == 0 {return}addr := node.Address()if len(msgs) == 1 {// Send single message as isif err := m.rawSendMsgPacket(node.FullAddress(), &node.Node, msgs[0]); err != nil {m.logger.Printf("[ERR] memberlist: Failed to send gossip to %s: %s", addr, err)}} else {// Otherwise create and send a compound messagecompound := makeCompoundMessage(msgs)if err := m.rawSendMsgPacket(node.FullAddress(), &node.Node, compound.Bytes()); err != nil {m.logger.Printf("[ERR] memberlist: Failed to send gossip to %s: %s", addr, err)}}}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

// pushPull is invoked periodically to randomly perform a complete state
// exchange. Used to ensure a high level of convergence, but is also
// reasonably expensive as the entire state of this node is exchanged
// with the other node.
func (m *Memberlist) pushPull() {// Get a random live nodem.nodeLock.RLock()// 随机抽取一个Alive的Node进行pushPullnodes := kRandomNodes(1, m.nodes, func(n *nodeState) bool {return n.Name == m.config.Name ||n.State != StateAlive})m.nodeLock.RUnlock()// If no nodes, bailif len(nodes) == 0 {return}node := nodes[0]// Attempt a push pullif err := m.pushPullNode(node.FullAddress(), false); err != nil {m.logger.Printf("[ERR] memberlist: Push/Pull with %s failed: %s", node.Name, err)}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

// pushPullNode does a complete state exchange with a specific node.
func (m *Memberlist) pushPullNode(a Address, join bool) error {defer metrics.MeasureSince([]string{"memberlist", "pushPullNode"}, time.Now())// 发送自己的状态信息并获取目标Node的状态信息remote, userState, err := m.sendAndReceiveState(a, join)if err != nil {return err}// 合并状态信息if err := m.mergeRemoteState(join, remote, userState); err != nil {return err}return nil
}