例子加深理解
你的工作是实现一个分布式的MapReduce，包括两个程序，master 和 worker。只有一个master进程 以及 一个或多个worker进程并行执行。在真实的系统中，工作人员将在许多不同的机器上运行，但在本实验室中，您将在一台机器上运行所有的worker。worker将通过RPC与master通信。每个worker进程将向master请求任务，从一个或多个文件读取任务的输入，执行任务，并将任务的输出写入一个或多个文件。如果一个worker没有在合理的时间内完成任务(对于这个实验室，使用10秒)，master应该注意，并把同样的任务交给另一个worker。
开始已经给了一些代码。main/mrcoordinator和main/mrworker.go分别是master和worker 的主例程，不要修改这些代码。
应该在mr/master.go、mr/worker.go和mr/rpc.go中给出实现。
lab1任务，干三件事
如何将大文件保存到一个单机中：
如何将一个大文件存到不同机器上
流程如下图所示：
WAL（WriteAhead Logging），先写日志，再写硬盘。保证原子性和持久性
如果写错了，就告诉client说写错了，让客户端重试，分布式系统不要想着修复，往往越修复错误越多
但是索引保存的信息：服务器id、偏移量、chunkid这种保存方式会很麻烦，一旦哪个文件修改了，都需要劳烦master更改。所以，为了减少master数据和流量，将偏移量保存在服务器上，master只保存chunk对应的服务器信息。如下图所示：
chunk(64MB)由一系列的block(64KB)组成，每个block都会保存校验和（checksum），每次读取时通过校验和来判断数据是否损坏，如下图所示
通过将chunk保存在3个不同的chunkServer上（也就是创建3个副本），来避免chunk服务器挂掉带来的损失。
chunkServer4中丢失（损坏）了chunk03的数据，向Master求助，Master告诉chunkServer4：CS3和CS5也保存了；CS4就会去找CS3或者CS5请求Chunk03。如下图所示：
会有一个修复进程，记录每个少于3个ChunkServer的数据，并按照副本的多少从少到多排序（先修复副本最少的数据）。如下图所示：
热点平衡进程会记录每个chunk的访问频率，以及每个chunkServer的空余空间、带宽等信息。当某个副本过于繁忙，成为热点，会将其复制到更多的ChunkServer上。
那如何选择ChunkServer呢？基于带宽和剩余空间来选择
复制（replication）能解决fail-stop故障。比如：拔掉网线，硬件故障等，都可以通过复制来解决
fail-stop：如果某些东西出了故障，比如说计算机，那么它会单纯的停止运行。
但是复制不能处理软件中的bug和硬件设计中的缺陷。
有两种办法实现复制
state tranfer：就是发送Primary的状态。比如Primary将自己的完整状态，例如内存，拷贝并发送给Backup。 Backup会保存收到的最近一次状态。当Primary故障了，Backup就可以从它所保存的最新状态开始运行。
replicated state machine：类似于redis的主从复制，在获取到初始化的状态后，将每一次输入（称外部操作或者外部事件）也发送给Backup，从而实现同步。 注意：通常来说，如果有两台计算机，如果它们从相同的状态开始，并且它们以相同的顺序，在相同的时间，看到了相同的输入，那么它们会一直互为副本（数据保持一致）。
replicated state machine更常用的原因是：外部事件或操作肯定比整个服务器的状态要小。
但是replicated state machine他会更加复杂一点，需要考虑的东西比较多。而state transfer只是简单的复制整个状态就行，不需要考虑其他事
❓： 随机操作在replicated state machine怎么处理?
对于这一类问题的统一答案是，Primary会执行这些指令，并将结果发送给Backup。Backup不会执行这些指令，而是在应该执行指令的地方，等着Primary告诉它，正确的答案是什么。
需要关注的知识点：
绝大多数的复制都是类似于GFS的软件层级的复制（更高效），而VMware FT（这篇论文）的复制是最底层的复制（系统层面）（包括复制内存的内容，复制寄存器的内容等），所以VMware FT需要考虑更加细节的东西（比如说需要确保中断在Primary和Backup的相同位置执行）。
但是VMware FT的优势在于：它从底层机器级别实现复制，因此它不关心你在机器上运行什么样的软件，它就是复制底层的寄存器和内存。所以基本可以实现所有软件的复制。VMware FT就是可以让任何软件都具备容错性的魔法棒。
事实上，Primary和Backup是运行在两个不同的物理服务器的VMM（虚拟机软件）上：在一个物理服务器上有一个虚拟机，这个虚拟机跑了某个操作系统（或者某个应用程序），我们将之指定为Primary。在第二个物理服务器上，运行了相同的VMM，和一个相同的虚拟机作为Backup。
他们处在同一个局域网内（LAN），该局域网还有一些客户端（用于与多副本服务进行交互的计算机），如下图所示
基本的工作流程是，我们假设这两个副本，或者说这两个虚拟机：Primary和Backup，互为副本。某些我们服务的客户端，向Primary发送了一个请求，这个请求以网络数据包（network packet）的形式发出。
这个网络数据包产生一个中断（interrupt），之后这个中断送到了Primary的VMM。VMM看到热点（a hot）可以发现这是一个发给我们的多副本服务的一个输入，所以这里VMM会做两件事情：
replicated state machine）。Backup虚拟机所在的VMM收到网络数据包后也会发送给backup。Log Channel：将Primary到Backup之间同步的数据流通道称之为Log Channel。
Log Event/entry：从Primary发往Backup的事件被称为Log Event/Entry。
如何回复应答？
注意：当网卡有新的数据包到达时，VMM不会允许直接将网卡的数据包通过DMA传到Primary虚拟机中，因为这样就失去了对Primary的时序控制。 准确的做法是：
在Primary中，会通过周期性的时钟中断生成Log Entry来发送给Backup。在某个时间段内，如果没有收到中断，说明Primary故障停止（fail-stop）了。
此时，Backup虚拟机会上线，他不会在等待来自Primary的Log Channel的事件。并且，Backup的VMM会让后续的客户端请求发往Backup虚拟机，而不是Primary虚拟机，Backup的VMM不再会丢弃Backup虚拟机的输出。此时Backup虚拟机接管了服务，并称为Primary。
非确定性事件有很多，主要有以下几种：
在这个系统中，唯一的输出就是对于客户端请求的响应。
Output Rule：直到Backup的VMM确认收到了相应的Log entry后，Primary虚拟机才允许将生成输出发送。 其实核心思想就是确保Backup能接收到了Log entry，才发送输出信息。
所以，几乎每一个复制系统都有这个问题，在某个时间点，Primary必须要停下来等待Backup，这对于性能是实打实的限制。每次生成一个输出时，都需要至少等待5毫秒，等Backup确认收到了这个Log entry，然后VMM才能将Primary输出发送到网络。
这就是VMwareFT的缺陷，所以大家更喜欢在软件层面实现复制。因为这样的复制可以理解每一个操作含义，这样就不需要在每个网络数据包到达时都暂停一下（例如，如果是软件层面的复制，在读操作就可以不暂停，甚至写操作也可以放缓冲区，主机持续运行，参考redis）
Test-and-Set服务器作用是：只要某个副本想上线接管服务时，他就需要去询问Test-and-Set服务器。主要是为了解决脑裂的。
为什么会产生脑裂？
Primary和Backup都在运行，但是它们之间的网络出现了问题，同时它们各自又能够与一些客户端通信。这时，它们都会以为对方挂了，自己需要上线并接管服务。
只要是涉及到网络，都有可能出现脑裂的现象，这不是系统设计的问题。所以需要通过Test-and-Set服务解决：向一个外部的第三方权威机构求证，来决定Primary还是Backup允许上线。这里的第三方就是Test-and-Set服务。
具体来说：Test-and-Set服务不运行在Primary和Backup的物理服务器上，VMware FT需要通过网络支持Test-and-Set服务。这个服务会在内存中保留一些标志位，当你向它发送一个Test-and-Set请求，它会设置标志位，并且返回旧的值。Primary和Backup都需要获取Test-and-Set标志位，这有点像一个锁。为了能够上线，它们或许会同时发送一个Test-and-Set请求，给Test-and-Set服务。当第一个请求送达时，Test-and-Set服务会说，这个标志位之前是0，现在是1。第二个请求送达时，Test-and-Set服务会说，标志位已经是1了，你不允许成为Primary。对于这个Test-and-Set服务，我们可以认为运行在单台服务器。当网络出现故障，并且两个副本都认为对方已经挂了时，Test-and-Set服务就是一个仲裁官，决定了两个副本中哪一个应该上线。
go主要的同步方式有四种
var wg sync.WaitGroup
wg.Add(1)// wg 信号量加1
wg.Done()// wg 信号量-1,注意 wg.Add(-1)和wg.Done是完全等价的
wg.Wait()//当wg不为0时，阻塞。只有当wg为0时，才继续执行
func main(){
var wg sync.WaitGroup
for i:=0; i<5; i++{
wg.Add(1)
//goroutine 匿名函数（闭包）
go func(x int){
// 协程执行任务
sendRPC(x)
wg.Done() // 减少计数器
}(i)//将参数传入匿名函数
// 阻塞主进程，直到所有协程完成任务（计数器为0）
wg.Wait()
}
func sendRPC(i int){
println(i)
}
上述代码可以用在raft选举中，等待所有候选人投完票，主协程在继续往后执行。或者利用条件变量（效率更高，不需要等待所有人投完票才看结果，只需要yes占半数就能胜任leader，如下代码所示）
func main() {
count:=1//投yes的数量。注意自己也会投自己一票
finished:=1//已完成投票的数量。
total:=10//假设节点总数为total
var mu sync.Mutex
cond := sync.NewCond(&mu)
//开始进行选举
for i:=0; i<total-1; i++{
//协程运行
go func() {
vote := RequestVote()//投票RPC
mu.Lock()
defer mu.Unlock()
if vote{
count++
}
finished++
cond.Broadcast()
}()
}
mu.Lock()
for count <= total/2 && finished<total{
cond.Wait()//每当一个协程完成投票，就会解除阻塞一次，判断是否满足条件，如果满足条件了，就跳出循环
}
if(count>total/2){
fmt.Println("success!","Yes=", count)
//改变该follower状态
}else {
fmt.Println("lost!","Yes=", count)
}
mu.Unlock()
}
func RequestVote() bool {
rand.Seed(time.Now().UnixNano()) // 设置随机数种子
return rand.Intn(2) == 0 // 生成 0 或者 1，对应 true 或者 false
}
在某个函数调用rpc调用函数不要使用锁锁住rpc调用函数（也就是一般来讲不要在 RPC 调用期间持有锁），会造成死锁。
意思是在等待其他peer响应时（调用rpc等对方回复消息），就不应该持有锁
方法1
在 util.go 文件中对打印 log 进行了封装，只有 Debug 参数等于 1 时才输出 log。
好处就是在代码中任何地方都可以加打印 log(如何用请看下面示例)，例如每个节点的变化。而需要关闭时只改一个参数即可（我之前都是一个个删除）。
~~some code
DPrintf("这里打印输出信息，只有当把util.go的Debug参数=1时，才会输出这行信息")
方法2
ctrl + \，Go 会终止程序，退出所有协程并打印 stacktrace，从stacktrace 中就能找到可能出问题的地方。
方法3
打开race检测看是否存在并发冲突
go test -run 2A // 正常运行测试案例 go test -race -run 2A // 开启race检测
共识：分布式系统中多个节点之间，彼此对某个状态达成一致结果的过程。换句话说就是，一群分布式的机器得在一些问题上统一口径，例如现在所有节点都认同，现在A正拥有一个锁。
例如：
通过共识算法实现分布式锁
如何实现复制？
首先，领导者通过日志复制（AppendEntries）RPC消息，将日志项复制到集群其他节点上
接着，如果领导者接收到大多数的复制成功响应后，它将日志项应用到它的状态机，并返回成功给客户端。如果领导者没有接收到大多数的复制成功响应，那么就返回错误给客户端
领导者将日志项应用到它的状态机，怎么没通知跟随者应用日志项呢？
因为领导者的日志复制RPC消息或心跳消息，包含了当前最大的、将会被提交（应用到状态机）的日志项索引值。所以通过日志复制RPC消息或心跳消息，跟随者就可以知道领导者的日志提交位置信息
日志复制过程：
在Raft算法中，服务器节点间的沟通联络采用的是远程过程调用（RPC），在领导者选举中，需要用到这两类的RPC：
State：每个节点所需要存储的变量
基本不怎么变的变量：
常变的变量：
Leader才需要的变量，每次选举后都需要重新初始化的变量：
Raft算法中每个任期由单调递增的数字（任期编号）标识，任期编号是随着选举的举行而变化的
该方法是由领导者发起，用来复制日志和提供心跳消息
发送的变量包括：
发送后，返回给leader的变量：
规则：
如何实现follower日志和leader日志保持一致性？
在Raft算法中，领导者通过强制跟随者直接复制自己的日志项，处理不一致日志。也就是说，Raft是通过以领导者的日志为准，来实现各节点日志的一致性的
领导者通过日志复制RPC一致性检查，找到跟随者节点上与自己相同日志项的最大索引值，然后复制并更新覆盖该索引值之后的日志项，实现了各节点日志的一致。跟随者中的不一致日志项会被领导者的日志覆盖，而且领导者从来不会覆盖或者删除自己的日志
发送的变量：
返回的结果：
注意：日志完整性更高的follower（日志索引项比candidate更大）是不会投票给candidate的。
解决服务器故障的两种方式
持久化的存储可以确保当服务器重启时，服务器可以找到相应的数据，并将其加载到内存中。这样可以使得服务器在故障并重启后，继续重启之前的状态。 当你更改了被标记为持久化的某个数据，服务器应该将更新写入到磁盘。
需要持久化的数据有：
1和2是为了防止产生2个ledaer，而3是为了保存应用程序（使用raft接口的应用）的状态
每次添加一个Log entry，更新currentTerm或者更新votedFor，你或许都需要持久化存储这些数据
在leader将log entry应用到状态机后，才会返回响应发送给client
如果你赢得了选举，你需要立刻发送一条AppendEntries RPC消息给其他所有的服务器。
这一部分简单介绍一下应用层和Raft层之间的接口。假设我们的应用程序是一个key-value数据库，下面一层是Raft层。
在Raft集群中，每一个副本上，这两层之间主要有两个接口。
接口一（KV层->raft）：key-value层用来转发客户端请求的接口。如果客户端发送一个请求给key-value层，key-value层会将这个请求转发给Raft层，并说：请将这个请求存放在Log中的某处。如下图所示
这个接口实际上是个函数调用，称之为Start函数。这个函数只接收一个参数，就是客户端请求。key-value层说：我接到了这个请求，请把它存在Log中，并在committed之后告诉我。
接口二（raft->KV层）：当大多数服务器（超半数）将请求保存到大多数服务器的log时，Raft层会通知key-value层：你刚刚在Start函数中传给我的请求已经commit了。 Raft层通知的，不一定是最近一次Start函数传入的请求。如下图所示
这个向上的接口时通过向go语言中的channel发送信息来实现的， raft从channel的一端写，key-value层从另一端读。所以这里有个叫做applyChannel的channel，通过它raft可以发送apply消息告知key-value层可以执行该请求了！ 。
注意：在raft发送apply消息消息之前，kv层还并没有真正执行客户端的命令，他必须等到raft接口发送了apply信息后，才能执行。然后返回reply给客户端
当然，key-value层需要知道从applyChannel中读取的消息，匹配之前调用的哪个Start函数，所以Start函数的返回需要有足够的信息给key-value层，这样才能完成对应。Start函数的返回值包括，这个请求将会存放在Log中的位置（index）。这个请求不一定能commit成功，但是如果commit成功的话，会存放在这个index位置。同时，它还会返回当前的任期号（term）和一些其它我们现在还不太关心的内容。
在apply消息中，将会包含请求（command）和对应的Log位置（index）
每次follower收到AppendEntries消息（心跳）都会重置raft节点的选举定时器，以压制其成为candidate。
如何设置每个节点的超时时间 （如果该结点在规定时间内没有收到心跳了，该结点就发送投票RPC，请求成为leader）
考虑一个特殊场景：
假设网线故障了，旧的Leader在一个网络分区中，这个网络分区中有一些客户端和少数（未过半）的服务器。在网络的另一个分区中，有着过半的服务器，这些服务器选出了一个新的Leader。旧的Leader会怎样，或者说为什么旧的Leader不会执行错误的操作？
这里看起来有两个潜在的问题。第一个问题是，如果一个Leader在一个网络分区中，并且这个网络分区没有过半的服务器。那么下次客户端发送请求时，这个在少数分区的Leader，它会发出AppendEntries消息。但是因为它在少数分区，即使包括它自己，它也凑不齐过半服务器，所以它永远不会commit这个客户端请求，它永远不会执行这个请求，它也永远不会响应客户端，并告诉客户端它已经执行了这个请求。所以，如果一个旧的Leader在一个不同的网络分区中，客户端或许会发送一个请求给这个旧的Leader，但是客户端永远也不能从这个Leader获得响应。所以没有客户端会认为这个旧的Leader执行了任何操作。另一个更奇怪的问题是，有可能Leader在向一部分Followers发完AppendEntries消息之后就故障了，所以这个Leader还没决定commit这个请求。
实现一个领导者选举和心跳：选举一个领导者，如果出现超时，则新leader接管
go test -run 2A运行测试RequestVote RPC，与选举相关的服务器规则以及与领导者选举相关的状态。raft struct中。还需要定义一个结构来保存日志信息RequestVoteArgs和RequestVoteReply结构。修改make函数：当一段时间没有收到其他peer信息，将通过发送requestvote rpc来启动leader 选举。AppendEntries RPC结构（尽管您可能还不需要所有参数）,并让leader定期发送它们。编写一个AppendEntries RPC处理程序方法，该方法可以重置选举超时。请务必通过2A测试，以便您看到以下内容：
$ go test -run 2A
Test (2A): initial election ...
... Passed -- 3.5 3 58 16840 0
Test (2A): election after network failure ...
... Passed -- 5.4 3 118 25269 0
Test (2A): multiple elections ...
... Passed -- 7.3 7 624 138014 0
PASS
ok 6.5840/raft 16.265s
$
所有测试花费超过600秒，单独的测试超过120秒都算失败。
难点：
由于我设计思路是：A节点发送投票rpc后，主线程就会沉睡，等待半数赞成票或者全部投票投完才会苏醒。这样的结果就是某个节点可能断开链接了，导致A节点没办法收到所有的投票结果，并且没法半数赞成，导致一直阻塞。所以需要设置一个协程负责监督计时，当超过选举时间了，那么就唤醒主线程成为Follower
//该协程负责监控这次选举是否超时，如果超时就通知主线程不阻塞，
go func() {
timer :=time.NewTimer(300 * time.Millisecond)
<-timer.C
rf.mu.Lock()
//把节点设置回Follower
if rf.state !=Leader {
rf.state = Follower
rf.votedFor = -1
rf.BroadcastHeartbeat(false)
}
cond.Broadcast()//时间到了通知主线程
rf.mu.Unlock()
}()
难点：
log复制的相关问题特别难：具体来说：ABC三个节点，A为Leader，此时日志状态为
A [1, 1, 1, 1, 1]
B [1]
C [1]
A在发送完第一个entry后，就由于网络问题，自己划为一个分区，此时B节点由于选举超时，成为leader，发送entry给C（A由于还在分区内，无法接收到B的rpc）
A [1, 1, 1, 1, 1]
B [1, 2]
C [1, 2]
此时A成功恢复网络，B将修改A内的日志
A [1, 2]
B [1, 2]
C [1, 2]
而此时B陷入沉睡，C由于选举超时成为Ledaer，并发送entry给A
A [1, 2, 3]
B [1, 2]
C [1, 2, 3]
此时，B突然苏醒，给A发送entry，如果此时没有处理，那么A就会修改索引为2的位置，将3改为2。而这个行为是不对的，因为B已经是老的leader了，没资格修改term为3的数据。这个bug找了修改了好久！
A [1, 2, 2] ❌
B [1, 2, 2]
C [1, 2, 3]
rf.currentTerm、rf.votedFor、rf.log的地方，都调用rf.persist()就可以了。2C比较简单，补充下函数，再调用下rf.persist()就过了。没什么难点。
如果我们能构造满足下面两个要求的这么一个序列，并且生成的图没有带环，那么就说明是强一致的
如下图所示，就是一个强一致性的例子
相同请求不执行2次原则：（可以当作难点）
客户端在发送请求时，可能该请求由于网络或者其他原因，导致结果没有回传，那么客户端将会重新发送该请求给同一个leader/或者其他服务器。
所以，要求服务器有两个功能：
- 有能力过滤出重复的请求
- 能记住第一次请求的回复
解决方法是，服务器会根据请求的唯一号或者其他的客户端信息来保存一个表。这样服务器可以记住，哦，我之前看过这个请求，并且执行过它，我会发送一个相同的回复给它，因为我不想执行相同的请求两次。
这将在Lab3中完成这样的机制，服务器发现了重复的请求，并将之前的回复重新发给客户端。
Zookeeper是读写分离：主服务器负责写，从服务器负责读。
那么如何保证读请求一定返回最新的数据（线性一致）？
实际上，Zookeeper并不要求返回最新的写入数据。 Zookeeper允许客户端将读请求发送给任意副本，并由副本根据自己的状态来响应读请求。副本的Log可能并没有拥有最新的条目，所以尽管系统中可能有一些更新的数据，这个副本可能还是会返回旧的数据。
有两个主要的要求来保证一致性：
更进一步，FIFO客户端请求序列是对一个客户端的所有读请求，写请求生效。所以，如果我发送一个写请求给Leader，在Leader commit这个请求之前需要消耗一些时间，所以我现在给Leader发了一个写请求，而Leader还没有处理完它，或者commit它。之后，我发送了一个读请求给某个副本。这个读请求需要暂缓一下，以确保FIFO客户端请求序列。读请求需要暂缓，直到这个副本发现之前的写请求已经执行了。这是FIFO客户端请求序列的必然结果，（对于某个特定的客户端）读写请求是线性一致的。
可以通过使用命令sync，先把所有写操作复制到副本后，再发送读请求，这样读到的数据一定是最新的。
要求：
我的实现是服务器协程将日志放入 raft 层去同步后即注册一个 channel 去阻塞等待，接着 apply 协程监控 applyCh，在得到 raft 层已经 commit 的日志后，apply 协程首先将其 apply 到状态机中，接着根据 index 得到对应的 channel ，最后将状态机执行的结果 push 到 channel 中，这使得服务器协程能够解除阻塞并回复结果给客户端。
客户端：
客户端通过RPC把需要put/append/get的信息发送给服务器。
服务器：
每个服务器开启一个raft协议，客户端通过调用服务器的函数将信息通过调用raft的start函数下发到raft中，raft在通过一致性协议，将信息通过日志的形式保存在raft日志中。接着，raft会通过applyChannel告诉服务器，我已经将信息保存在raft日志中，服务器中的一个负责监控applyChannel的协程就会接收到，该协程会做两件事:
相同请求不执行2次原则：（可以当作难点）
客户端在发送请求时，可能该请求由于网络或者其他原因，导致结果没有回传，那么客户端将会重新发送该请求给同一个leader/或者其他服务器。
所以，要求服务器有两个功能：
- 有能力过滤出重复的请求
- 能记住第一次请求的回复
解决方法是，服务器会记录每个命令的序号（序号是递增的），并且通过map结构保存每个客户端已经执行命令的最大序号。这样服务器可以记住，哦，我之前执行过这个请求，我会发送一个相同的回复给它，因为我不想执行相同的请求两次。