ETCD数据库源码分析——rawnode简单封装

在etcd 3.6.0之前是没有rawnode模块，现在出现的rawnode.go仅仅是raft模块的简单封装，并将raft.Node run协程中每轮循环需要保存的raft的前一次硬软状态prevSoftSt和prevHardSt保存到rawnode结构体中。如下为rawnode结构体的定义：

// RawNode is a thread-unsafe Node. The methods of this struct correspond to the methods of Node and are described more fully there.
type RawNode struct {
	raft       *raft
	prevSoftSt *SoftState
	prevHardSt pb.HardState
}
1
2
3
4
5
6

rawnode提供的接口如下所示，粗体的函数会被其他模块调用，从这些接口函数我们可以看到作为raft模块的封装需要提供哪些接口：
NewRawNode实例化给定配置中的RawNode。有关引导初始状态的信息，请参阅Bootstrap（）；这取代了该方法的前“peers”参数（具有相同的行为）。但是，建议应用程序不调用引导，而是通过设置第一个索引大于1且将所需ConfState存储为其初始状态的存储器来手动引导其状态。这部分可以看node.go中的startNode和RestartNode函数。

// NewRawNode instantiates a RawNode from the given configuration. See Bootstrap() for bootstrapping an initial state; this replaces the former 'peers' argument to this method (with identical behavior). However, It is recommended that instead of calling Bootstrap, applications bootstrap their state manually by setting up a Storage that has a first index > 1 and which stores the desired ConfState as its InitialState.
func NewRawNode(config *Config) (*RawNode, error) {
	r := newRaft(config)
	rn := &RawNode{ raft: r, }
	rn.prevSoftSt = r.softState()
	rn.prevHardSt = r.hardState()
	return rn, nil
}
1
2
3
4
5
6
7
8

Tick将内部逻辑时钟提前一个Tick，可以查看ETCD数据库源码分析——从raftNode的start函数说起文章看查看逻辑时钟调用链原理。

func (rn *RawNode) Tick() { rn.raft.tick() } // Tick advances the internal logical clock by a single tick.
1

acceptReady:当RawNode的使用者决定继续处理Ready时，调用acceptReady。在这个调用和之前对Ready()的调用之间，任何东西都不能改变RawNode的状态。

// acceptReady is called when the consumer of the RawNode has decided to go
// ahead and handle a Ready. Nothing must alter the state of the RawNode between
// this call and the prior call to Ready().
func (rn *RawNode) acceptReady(rd Ready) {
	if rd.SoftState != nil { rn.prevSoftSt = rd.SoftState }
	if len(rd.ReadStates) != 0 { rn.raft.readStates = nil }
	rn.raft.msgs = nil
}
1
2
3
4
5
6
7
8

readyWithoutAccept返回Ready实例。这是一个只读操作，即没有必须处理就绪的义务。

// readyWithoutAccept returns a Ready. This is a read-only operation, i.e. there
// is no obligation that the Ready must be handled.
func (rn *RawNode) readyWithoutAccept() Ready {
	return newReady(rn.raft, rn.prevSoftSt, rn.prevHardSt)
}
1
2
3
4
5

HasReady在RawNode用户需要检查是否有任何准备就绪挂起时调用。此方法中的检查逻辑应与Ready.containsUpdates()一致。

// HasReady called when RawNode user need to check if any Ready pending.
// Checking logic in this method should be consistent with Ready.containsUpdates().
func (rn *RawNode) HasReady() bool {
	r := rn.raft
	if !r.softState().equal(rn.prevSoftSt) { return true }
	if hardSt := r.hardState(); !IsEmptyHardState(hardSt) && !isHardStateEqual(hardSt, rn.prevHardSt) { return true }
	if r.raftLog.hasPendingSnapshot() { return true }
	if len(r.msgs) > 0 || len(r.raftLog.unstableEntries()) > 0 || r.raftLog.hasNextEnts() { return true }
	if len(r.readStates) != 0 { return true }
	return false
}
1
2
3
4
5
6
7
8
9
10
11

Advance通知RawNode应用程序已应用并保存最后就绪结果中的进度。

// Advance notifies the RawNode that the application has applied and saved progress in the
// last Ready results.
func (rn *RawNode) Advance(rd Ready) {
	if !IsEmptyHardState(rd.HardState) { rn.prevHardSt = rd.HardState }
	rn.raft.advance(rd)
}
1
2
3
4
5
6

---

TickQuiesced将内部逻辑时钟提前一个滴答，而不执行任何其他状态机处理。当已知Raft组中的所有对等点处于同一状态时，调用方可以避免周期性心跳和选举。预期用法是根据组是“活动”还是“静止”定期调用Tick或tickquiesed。警告：使用此方法时要非常小心，因为它会破坏Raft状态机。你可能应该改用Tick。

// TickQuiesced advances the internal logical clock by a single tick without
// performing any other state machine processing. It allows the caller to avoid
// periodic heartbeats and elections when all of the peers in a Raft group are
// known to be at the same state. Expected usage is to periodically invoke Tick
// or TickQuiesced depending on whether the group is "active" or "quiesced".
//
// WARNING: Be very careful about using this method as it subverts the Raft
// state machine. You should probably be using Tick instead.
func (rn *RawNode) TickQuiesced() { rn.raft.electionElapsed++ }
1
2
3
4
5
6
7
8
9

Campaign导致此RawNode转换为候选状态。

// Campaign causes this RawNode to transition to candidate state.
func (rn *RawNode) Campaign() error { return rn.raft.Step(pb.Message{ Type: pb.MsgHup, }) }
1
2

Propose将数据附加到筏形日志中。

// Propose proposes data be appended to the raft log.
func (rn *RawNode) Propose(data []byte) error { return rn.raft.Step(pb.Message{Type: pb.MsgProp,From: rn.raft.id,Entries: []pb.Entry{{Data: data},}}) }
1
2

ProposeConfChange提出配置更改。

// ProposeConfChange proposes a config change. See (Node).ProposeConfChange for details.
func (rn *RawNode) ProposeConfChange(cc pb.ConfChangeI) error {
	m, err := confChangeToMsg(cc)
	if err != nil { return err }
	return rn.raft.Step(m)
}
1
2
3
4
5
6

ApplyConfChange将配置更改应用于本地节点。应用程序在应用配置更改时必须调用此选项，除非它决定拒绝配置更改，在这种情况下，不得调用。

// ApplyConfChange applies a config change to the local node. The app must call
// this when it applies a configuration change, except when it decides to reject
// the configuration change, in which case no call must take place.
func (rn *RawNode) ApplyConfChange(cc pb.ConfChangeI) *pb.ConfState {
	cs := rn.raft.applyConfChange(cc.AsV2())
	return &cs
}
1
2
3
4
5
6
7

Step使用给定消息推进状态机。

// Step advances the state machine using the given message.
func (rn *RawNode) Step(m pb.Message) error {
	// ignore unexpected local messages receiving over network
	if IsLocalMsg(m.Type) { return ErrStepLocalMsg }
	if pr := rn.raft.prs.Progress[m.From]; pr != nil || !IsResponseMsg(m.Type) { return rn.raft.Step(m) }
	return ErrStepPeerNotFound
}
1
2
3
4
5
6
7

ReportUnreachable报告上次发送时无法访问给定节点。ReportSnapshot报告已发送快照的状态。TransferLeader尝试将领导权转移给给定的受让人。ReadIndex请求读取状态。读取状态将设置为就绪。读取状态具有读取索引。一旦应用程序的进程超过读取索引，则可以安全地处理在读取请求之前发出的任何可线性化的读取请求。读取状态将连接相同的rctx。

// ReportUnreachable reports the given node is not reachable for the last send.
func (rn *RawNode) ReportUnreachable(id uint64) { _ = rn.raft.Step(pb.Message{Type: pb.MsgUnreachable, From: id}) }
// ReportSnapshot reports the status of the sent snapshot.
func (rn *RawNode) ReportSnapshot(id uint64, status SnapshotStatus) {
	rej := status == SnapshotFailure
	_ = rn.raft.Step(pb.Message{Type: pb.MsgSnapStatus, From: id, Reject: rej})
}
// TransferLeader tries to transfer leadership to the given transferee.
func (rn *RawNode) TransferLeader(transferee uint64) { _ = rn.raft.Step(pb.Message{Type: pb.MsgTransferLeader, From: transferee}) }

// ReadIndex requests a read state. The read state will be set in ready.
// Read State has a read index. Once the application advances further than the read
// index, any linearizable read requests issued before the read request can be
// processed safely. The read state will have the same rctx attached.
func (rn *RawNode) ReadIndex(rctx []byte) { _ = rn.raft.Step(pb.Message{Type: pb.MsgReadIndex, Entries: []pb.Entry{{Data: rctx}}}) }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

Ready返回应用程序需要处理的未完成工作。这包括附加和应用条目或快照、更新硬状态和发送消息。返回的Ready()必须处理，然后通过Advance()传回。

// Ready returns the outstanding work that the application needs to handle. This
// includes appending and applying entries or a snapshot, updating the HardState,
// and sending messages. The returned Ready() *must* be handled and subsequently
// passed back via Advance().
func (rn *RawNode) Ready() Ready {
	rd := rn.readyWithoutAccept()
	rn.acceptReady(rd)
	return rd
}
1
2
3
4
5
6
7
8
9

Status返回给定组的当前状态。这将进行分配，请参阅BasicStatus和WithProgress以获得更友好的分配选择。BasicStatus返回BasicStatus。值得注意的是，这并不包含进度图；请参阅WithProgress，了解无需分配的检查方法。

// Status returns the current status of the given group. This allocates, see
// BasicStatus and WithProgress for allocation-friendlier choices.
func (rn *RawNode) Status() Status {
	status := getStatus(rn.raft)
	return status
}
// BasicStatus returns a BasicStatus. Notably this does not contain the
// Progress map; see WithProgress for an allocation-free way to inspect it.
func (rn *RawNode) BasicStatus() BasicStatus { return getBasicStatus(rn.raft) }
1
2
3
4
5
6
7
8
9

WithProgress是一个帮助器，用于反思此节点及其对等节点的进度。

type ProgressType byte // ProgressType indicates the type of replica a Progress corresponds to.
const (	
	ProgressTypePeer ProgressType = iota // ProgressTypePeer accompanies a Progress for a regular peer replica.	
	ProgressTypeLearner // ProgressTypeLearner accompanies a Progress for a learner replica.
)

// WithProgress is a helper to introspect the Progress for this node and its peers.
func (rn *RawNode) WithProgress(visitor func(id uint64, typ ProgressType, pr tracker.Progress)) {
	rn.raft.prs.Visit(func(id uint64, pr *tracker.Progress) {
		typ := ProgressTypePeer
		if pr.IsLearner {
			typ = ProgressTypeLearner
		}
		p := *pr
		p.Inflights = nil
		visitor(id, typ, p)
	})
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18