TinyKv Project3 PartB Multi-raft KV

message Region {
  uint64 id = 1;
  // Region key range [start_key, end_key).
  bytes start_key = 2;
  bytes end_key = 3;
  RegionEpoch region_epoch = 4;
  repeated Peer peers = 5
}

//将 client 的请求包装成 entry 传递给 raft 层
func (d *peerMsgHandler) proposeRaftCommand(msg 
  // Your Code Here (2B).
  if msg.AdminRequest != nil {
    d.proposeAdminRequest(msg, cb)
  } else {
    d.proposeRequest(msg, cb)
  }
}
func (d *peerMsgHandler) proposeAdminRequest(msg *raft_cmdpb.RaftCmdRequest, cb *message.Callback) {
  case raft_cmdpb.AdminCmdType_CompactLog: // 日志压缩需要提交到 raft 同步
  case raft_cmdpb.AdminCmdType_TransferLeader: // 领导权禅让直接执行，不需要提交到 raft
  case raft_cmdpb.AdminCmdType_ChangePeer: // 集群成员变更，需要提交到 raft，并处理 proposal 回调
  case raft_cmdpb.AdminCmdType_Split: // Region 分裂
}

case raft_cmdpb.AdminCmdType_TransferLeader: // 领导权禅让直接执行，不需要提交到 raft
  // 执行领导权禅让
  d.RaftGroup.TransferLeader(msg.AdminRequest.TransferLeader.Peer.Id)
  // 返回 response
  adminResp := &raft_cmdpb.AdminResponse{
    CmdType:        raft_cmdpb.AdminCmdType_TransferLeader,
    TransferLeader: &raft_cmdpb.TransferLeaderResponse{},
  }
  cb.Done(&raft_cmdpb.RaftCmdResponse{
    Header:        &raft_cmdpb.RaftResponseHeader{},
    AdminResponse: adminResp,
  })

case raft_cmdpb.AdminCmdType_ChangePeer: // 集群成员变更，需要提交到 raft，并处理 proposal 回调
  // 单步成员变更：前一步成员变更被提交之后才可以执行下一步成员变更
  if d.peerStorage.AppliedIndex() >= d.RaftGroup.Raft.PendingConfIndex {
    // 如果 region 只有两个节点，并且需要 remove leader，则需要先完成 transferLeader
    if len(d.Region().Peers) == 2 && msg.AdminRequest.ChangePeer.ChangeType == pb.ConfChangeType_RemoveNode && msg.AdminRequest.ChangePeer.Peer.Id == d.PeerId() {
      for _, p := range d.Region().Peers {
        if p.Id != d.PeerId() {
          d.RaftGroup.TransferLeader(p.Id)
          break
        }
      }
    }
    // 1. 创建 proposal
    d.proposals = append(d.proposals, &proposal{
      index: d.nextProposalIndex(),
      term:  d.Term(),
      cb:    cb,
    })
    // 2. 提交到 raft
    context, _ := msg.Marshal()
    d.RaftGroup.ProposeConfChange(pb.ConfChange{
      ChangeType: msg.AdminRequest.ChangePeer.ChangeType, // 变更类型
      NodeId:     msg.AdminRequest.ChangePeer.Peer.Id,    // 变更成员 id
      Context:    context,                                // request data
    })
  }

case raft_cmdpb.AdminCmdType_Split: // Region 分裂
  // 如果收到的 Region Split 请求是一条过期的请求，则不应该提交到 Raft
  if err := util.CheckRegionEpoch(msg, d.Region(), true); err != nil {
    log.Infof("[AdminCmdType_Split] Region %v Split, a expired request", d.Region())
    cb.Done(ErrResp(err))
    return
  }
  if err := util.CheckKeyInRegion(msg.AdminRequest.Split.SplitKey, d.Region()); err != nil {
    cb.Done(ErrResp(err))
    return
  }
  log.Infof("[AdminCmdType_Split Propose] Region %v Split, entryIndex %v", d.Region(), d.nextProposalIndex())
  // 否则的话 Region 还没有开始分裂，则将请求提交到 Raft
  d.proposals = append(d.proposals, &proposal{
    index: d.nextProposalIndex(),
    term:  d.Term(),
    cb:    cb,
  })
  data, _ := msg.Marshal()
  d.RaftGroup.Propose(data)
}

// HandleRaftReady 处理 rawNode 传递来的 Ready
// HandleRaftReady 对这些 entries 进行 apply（即执行底层读写命令）
// 每执行完一次 apply，都需要对 proposals 中的相应 Index 的 proposal 进行 callback 回应（调用 cb.Done()）
// 然后从中删除这个 proposal。
func (d *peerMsgHandler) HandleRaftReady() {
  // Your Code Here (2B).
  //1. 判断是否有新的 Ready，没有就什么都不处理；
  //2. 调用 SaveReadyState 将 Ready 中需要持久化的内容保存到 badger。
  //3. 调用 d.Send() 方法将 Ready 中的 Msg 发送出去；
  //4. 应用待apply的日志，即实际去执行
  kvWB = d.processCommittedEntry(&ent, kvWB)
  //5. 调用 d.RaftGroup.Advance() 推进 RawNode,更新 raft 状态
}
func (d *peerMsgHandler) processCommittedEntry(entry *pb.Entry, kvWB *engine_util.WriteBatch) *engine_util.WriteBatch {
  // 检查日志是否是配置变更日志
  if entry.EntryType == pb.EntryType_EntryConfChange {
    return d.processConfChange(entry, cc, kvWB)
  }
  // 判断是 AdminRequest 还是普通的 Request
  if requests.AdminRequest != nil {
    return d.processAdminRequest(entry, requests, kvWB)
  } else {
    return d.processRequest(entry, requests, kvWB)
  }
}

func (r *Raft) handleTransferLeader(m pb.Message) {
  // 判断 transferee 是否在集群中
  if _, ok := r.Prs[m.From]; !ok {
    return
  }
  // 如果 transferee 就是 leader 自身，则无事发生
  if m.From == r.id {
    return
  }
  // 判断是否有转让流程正在进行，如果是相同节点的转让流程就返回，否则的话终止上一个转让流程
  if r.leadTransferee != None {
    if r.leadTransferee == m.From {
      return
    }
    r.leadTransferee = None
  }
  r.leadTransferee = m.From
  r.transferElapsed = 0
  if r.Prs[m.From].Match == r.RaftLog.LastIndex() {
    // 日志是最新的，直接发送 TimeoutNow 消息
    r.sendTimeoutNow(m.From)
  } else {
    // 日志不是最新的，则帮助 leadTransferee 匹配最新的日志
    r.sendAppend(m.From)
  }
}

// processConfChange 处理配置变更日志
func (d *peerMsgHandler) processConfChange(entry *pb.Entry, cc *pb.ConfChange, kvWB *engine_util.WriteBatch) *engine_util.WriteBatch {
  // 获取 ConfChange Command Request
  msg := &raft_cmdpb.RaftCmdRequest{}
  if err := msg.Unmarshal(cc.Context); err != nil {
    log.Panic(err)
  }
  region := d.Region()
  changePeerReq := msg.AdminRequest.ChangePeer
  // 检查 Command Request 中的 RegionEpoch 是否是过期的，以此判定是不是一个重复的请求
  // 实验指导书中提到，测试程序可能会多次提交同一个 ConfChange 直到 ConfChange 被应用
  // CheckRegionEpoch 检查 RaftCmdRequest 头部携带的 RegionEpoch 是不是和 currentRegionEpoch 匹配
  if err, ok := util.CheckRegionEpoch(msg, region, true).(*util.ErrEpochNotMatch); ok {
    log.Infof("[processConfChange] %v RegionEpoch not match", d.PeerId())
    d.handleProposal(entry, ErrResp(err))
    return kvWB
  }
  switch cc.ChangeType {
  case pb.ConfChangeType_AddNode: // 添加一个节点
    log.Infof("[AddNode] %v add %v", d.PeerId(), cc.NodeId)
    // 待添加的节点必须原先在 Region 中不存在
    if d.searchPeerWithId(cc.NodeId) == len(region.Peers) {
      // region 中追加新的 peer
      region.Peers = append(region.Peers, changePeerReq.Peer)
      region.RegionEpoch.ConfVer++
      meta.WriteRegionState(kvWB, region, rspb.PeerState_Normal) // PeerState 用来表示当前 Peer 是否在 region 中
      // 更新 metaStore 中的 region 信息
      d.updateStoreMeta(region)
      // 更新 peerCache，peerCache 保存了 peerId -> Peer 的映射
      // 当前 raft_store 上的 peer 需要发送消息给同一个 region 中的别的节点的时候，需要获取别的节点所在 storeId
      // peerCache 里面就保存了属于同一个 region 的所有 peer 的元信息（peerId, storeId）
      d.insertPeerCache(changePeerReq.Peer)
    }
  case pb.ConfChangeType_RemoveNode: // 删除一个节点
    log.Infof("[RemoveNode] %v remove %v", d.PeerId(), cc.NodeId)
    // 如果目标节点是自身，那么直接销毁并返回：从 raft_store 上删除所属 region 的所有信息
    if cc.NodeId == d.PeerId() {
      d.destroyPeer()
      log.Infof("[RemoveNode] destory %v compeleted", cc.NodeId)
      return kvWB
    }
    // 待删除的节点必须存在于 region 中
    n := d.searchPeerWithId(cc.NodeId)
    if n != len(region.Peers) {
      // 删除节点 RaftGroup 中的第 n 个 peer（注意，这里并不是编号为 n 的 peer，而是第 n 个 peer）
      region.Peers = append(region.Peers[:n], region.Peers[n+1:]...)
      region.RegionEpoch.ConfVer++
      meta.WriteRegionState(kvWB, region, rspb.PeerState_Normal) // PeerState 用来表示当前 Peer 是否在 region 中
      // 更新 metaStore 中的 region 信息
      d.updateStoreMeta(region)
      // 更新 peerCache
      d.removePeerCache(cc.NodeId)
    }
  }
  // 更新 raft 层的配置信息
  d.RaftGroup.ApplyConfChange(*cc)
  // 处理 proposal
  d.handleProposal(entry, &raft_cmdpb.RaftCmdResponse{
    Header: &raft_cmdpb.RaftResponseHeader{},
    AdminResponse: &raft_cmdpb.AdminResponse{
      CmdType:    raft_cmdpb.AdminCmdType_ChangePeer,
      ChangePeer: &raft_cmdpb.ChangePeerResponse{Region: region},
    },
  })
  // 新增加的 peer 是通过 leader 的心跳完成的
  if d.IsLeader() {
    d.HeartbeatScheduler(d.ctx.schedulerTaskSender)
  }
  return kvWB
}

case raft_cmdpb.AdminCmdType_Split: // Region Split 请求处理
  if requests.Header.RegionId != d.regionId {
    regionNotFound := &util.ErrRegionNotFound{RegionId: requests.Header.RegionId}
    d.handleProposal(entry, ErrResp(regionNotFound))
    return kvWB
  }
  // error: 过期的请求
  if errEpochNotMatch, ok := util.CheckRegionEpoch(requests, d.Region(), true).(*util.ErrEpochNotMatch); ok {
    d.handleProposal(entry, ErrResp(errEpochNotMatch))
    return kvWB
  }
  // error: key 不在 oldRegion 中
  if err := util.CheckKeyInRegion(adminReq.Split.SplitKey, d.Region()); err != nil {
    d.handleProposal(entry, ErrResp(err))
    return kvWB
  }
  // error: Split Region 的 peers 和当前 oldRegion 的 peers 数量不相等，不知道为什么会出现这种原因
  if len(d.Region().Peers) != len(adminReq.Split.NewPeerIds) {
    d.handleProposal(entry, ErrRespStaleCommand(d.Term()))
    return kvWB
  }
  oldRegion, split := d.Region(), adminReq.Split
  oldRegion.RegionEpoch.Version++
  newRegion := d.createNewSplitRegion(split, oldRegion) // 创建新的 Region
  // 修改 storeMeta 信息
  storeMeta := d.ctx.storeMeta
  storeMeta.Lock()
  storeMeta.regionRanges.Delete(&regionItem{region: oldRegion})          // 删除 oldRegion 的数据范围
  oldRegion.EndKey = split.SplitKey                                      // 修改 oldRegion 的 range
  storeMeta.regionRanges.ReplaceOrInsert(&regionItem{region: oldRegion}) // 更新 oldRegion 的 range
  storeMeta.regionRanges.ReplaceOrInsert(&regionItem{region: newRegion}) // 创建 newRegion 的 range
  storeMeta.regions[newRegion.Id] = newRegion                            // 设置 regions 映射
  storeMeta.Unlock()
  // 持久化 oldRegion 和 newRegion
  meta.WriteRegionState(kvWB, oldRegion, rspb.PeerState_Normal)
  meta.WriteRegionState(kvWB, newRegion, rspb.PeerState_Normal)
  // 创建当前 store 上的 newRegion Peer，注册到 router，并启动
  peer, err := createPeer(d.storeID(), d.ctx.cfg, d.ctx.schedulerTaskSender, d.ctx.engine, newRegion)
  if err != nil {
    log.Panic(err)
  }
  d.ctx.router.register(peer)
  d.ctx.router.send(newRegion.Id, message.Msg{Type: message.MsgTypeStart})
  // 处理回调函数
  d.handleProposal(entry, &raft_cmdpb.RaftCmdResponse{
    Header: &raft_cmdpb.RaftResponseHeader{},
    AdminResponse: &raft_cmdpb.AdminResponse{
      CmdType: raft_cmdpb.AdminCmdType_Split,
      Split:   &raft_cmdpb.SplitResponse{Regions: []*metapb.Region{newRegion, oldRegion}},
    },
  })
  log.Infof("[AdminCmdType_Split Process] oldRegin %v, newRegion %v", oldRegion, newRegion)
  // 发送 heartbeat 给其他节点
  if d.IsLeader() {
    d.HeartbeatScheduler(d.ctx.schedulerTaskSender)
    d.notifyHeartbeatScheduler(newRegion, peer)
  }
}

func (d *storeWorker) onRaftMessage(msg *rspb.RaftMessage) error {
  created, err := d.maybeCreatePeer(regionID, msg)
}
func (d *storeWorker) maybeCreatePeer(regionID uint64, msg *rspb.RaftMessage) (bool, error) {
  if !util.IsInitialMsg(msg.Message) {
    log.Debugf("target peer %s doesn't exist", msg.ToPeer)
    return false, nil
  }
}
//这个部分用来判断 msg 是否初始化了，如果没有，那就直接 return 回去，不创建 peer 了
func IsInitialMsg(msg *eraftpb.Message) bool {
  return msg.MsgType == eraftpb.MessageType_MsgRequestVote ||
    // the peer has not been known to this leader, it may exist or not.
    (msg.MsgType == eraftpb.MessageType_MsgHeartbeat && msg.Commit == RaftInvalidIndex)
}