社区首页 >专栏 >etcd源码分析 - 4.【打通核心流程】processInternalRaftRequestOnce四个细节

etcd源码分析 - 4.【打通核心流程】processInternalRaftRequestOnce四个细节

junedayday

发布于 2022-12-02 11:12:44

58900

代码可运行

文章被收录于专栏：Go编程点滴Go编程点滴

运行总次数：0

代码可运行

在上一讲，我们继续梳理了PUT请求到EtcdServer这一层的逻辑，并大概阅读了其中的关键函数processInternalRaftRequestOnce。

这个方法里面有不少细节，我们今天就选择其中有价值的四点来看看。

1. entry索引 - appliedIndex与committedIndex

在etcd中，我们将每个客户端的操作（如PUT）抽象为一个日志项（entry）。如果这个操作生效，etcd就将这个entry项同步给其它etcd server，作为数据同步。

操作有顺序之分，于是服务端就保存了一个长entry数组，用一个关键的索引index来进行区分entry数组（即一个分界的标志），对entry状态进行分类：

entry处于状态A - 小于等于索引的entry项
entry处于状态B - 大于索引的entry项

一般状态A和B都是互补的，即是一种二分类状态。

而由于分布式的特性，entry不能立刻完成执行的，于是这里就区分出了两种状态，它们复用一个entry数组：

已应用 - applied
已提交 - committed

对应索引appliedIndex与committedIndex：

// 函数用atomic保证原子性
ai := s.getAppliedIndex()
ci := s.getCommittedIndex()
// 两者的差值，表示已应用但是未提交的entry数，不能太多
if ci > ai+maxGapBetweenApplyAndCommitIndex {
  return nil, ErrTooManyRequests
}

entry数组中的索引的一致性非常重要，尤其是在并发的场景下。而示例中的原子操作，其实是一种乐观锁的实现。

更多的细节就涉及到分布式相关了，这里就不展开。

2.id生成器 - idutil.Generator

Generator数据结构不复杂，它的设计详情都放在了备注里，我们可以自行阅读：

// Generator generates unique identifiers based on counters, timestamps, and
// a node member ID.
//
// The initial id is in this format:
// High order 2 bytes are from memberID, next 5 bytes are from timestamp,
// and low order one byte is a counter.
// | prefix   | suffix              |
// | 2 bytes  | 5 bytes   | 1 byte  |
// | memberID | timestamp | cnt     |

在很多分布式系统中，都需要有一套唯一id生成器。etcd的这个方案相对简单，就是成员id+时间戳的组合方案。

关于分布式唯一id，更全面的设计可以参考Snowflake，如 https://segmentfault.com/a/1190000020899379

3.认证模块 - auth.AuthStore

authInfo, err := s.AuthInfoFromCtx(ctx)

认证功能在成熟软件中非常常见。在etcd，被独立到了etcd/auth模块中。这个模块的内部调用不复杂，功能就是从context中提取出 用户名+版本信息。

这个提取过程中值得注意的是，AuthStore是从grpc的metadata提取出想要的认证信息，而metadata类似于HTTP1协议中的header，是一种用KV形式保存和提取数据的结构。

串联一下我们之前的思路，etcd通过grpc-gateway将HTTP1转化成了gRPC，那么就有一个 HTTP header到grpc metadata的映射过程，有兴趣的可以去研究一下。

总体来说，etcd的认证模块做得很简单，也方便其接入service-mesh。

4.多协程小工具 - wait.Wait

wait.Wait是一个很精巧的小工具，使用起来非常简单：

// 示例代码
ch := s.w.Register(id)
s.w.Trigger(id, nil)

我们可以在etcd/pkg/wait目录下看到它的具体实现，我提取了重点

// 通过id，来等待和触发对应的事件。
// 注意使用的顺序：先等待，再触发。
type Wait interface {
  // 等待，即注册一个id
 Register(id uint64) <-chan interface{}
 // 触发，用一个id
 Trigger(id uint64, x interface{})
 IsRegistered(id uint64) bool
}

// 实现：读写锁+map数据结构
type list struct {
 l sync.RWMutex
 m map[uint64]chan interface{}
}

// 注册一个id
func (w *list) Register(id uint64) <-chan interface{} {
 w.l.Lock()
 defer w.l.Unlock()
 ch := w.m[id]
 if ch == nil {
    // go官方建议带buffer的channel尽量设置大小为1
  ch = make(chan interface{}, 1)
  w.m[id] = ch
 } else {
    // 不允许重复
  log.Panicf("dup id %x", id)
 }
 return ch
}

// 触发id的channel
func (w *list) Trigger(id uint64, x interface{}) {
 w.l.Lock()
 ch := w.m[id]
 delete(w.m, id)
  // 取出ch后直接Unlock（可以思考一下与defer的区别）
 w.l.Unlock()
  // 如果触发的id不存在map里，就直接跳过这个判断
 if ch != nil {
  ch <- x
  close(ch)
 }
}

了解Wait的实现之后，我们就知道在正常情况下，Register和Trigger必须一一对应。

但是，我们再往下看processInternalRaftRequestOnce这部分代码，发现了一个异常点：

select {
  // 异常：没有找到Trigger，难道忘了？
 case x := <-ch:
  return x.(*applyResult), nil
  // 正常：用Trigger退出
 case <-cctx.Done():
  proposalsFailed.Inc()
  s.w.Trigger(id, nil) 
  return nil, s.parseProposeCtxErr(cctx.Err(), start)
  // 正常：整个server停止，此时不用关心单个Trigger了
 case <-s.done:
  return nil, ErrStopped
}