Codis原始碼解析——處理slot操作（1）
阿新 • • 發佈：2018-12-25
上一篇我們講了slot在叢集中的分配方式，重點講了auto-rebalance的原理。之前我們說過，再啟動dashboard的時候，有一個goroutine專門用來處理slot的操作。這一篇我們就來看看slot的操作是如何進行的。我們這裡舉例也是用叢集中有兩個group和1024個從未分配的slot。
首先複習一下，在slot還處於未分配狀態的時候，上下文中的1024個SlotMapping如下所示。後面就是以SlotMapping為單位進行處理，所以這裡一定要對其結構掌握清楚
接下來，當我們使用auto-rebalance對叢集進行處理後，每個slot都被指定了相應的遷移計劃
func 一個slot共有七種狀態，分別是：
nothing（用空字串表示）、pending、preparing、prepared、migrating、finished、syncing
在看每個slot具體的操作之前，可以先看一下SlotActionPrepareFilter這個方法。
func (s *Topom) SlotActionPrepareFilter(accept, update func(m *models.SlotMapping) bool) (int, bool, error) {
s.mu.Lock()
defer s.mu.Unlock()
//載入上下文
ctx, err := s.newContext()
if err != nil {
return 0, false, err
}
//找到所有Action.State既不為空也不是pending的SlotMapping中Action.Index最小的SlotMapping
var minActionIndex = func(filter func(m *models.SlotMapping) bool) (picked *models.SlotMapping) {
for _, m := range ctx.slots {
if m.Action.State == models.ActionNothing {
continue
}
if filter(m) {
if picked != nil && picked.Action.Index < m.Action.Index {
continue
}
//只有一個slot沒有執行過update方法，accept才會返回true；也就是說，一個slot只會被處理一次
if accept == nil || accept(m) {
picked = m
}
}
}
return picked
}
var m = func() *models.SlotMapping {
var picked = minActionIndex(func(m *models.SlotMapping) bool {
return m.Action.State != models.ActionPending
})
if picked != nil {
return picked
}
if s.action.disabled.IsTrue() {
return nil
}
//如果前面找不到Action.State既不為空也不是pending的SlotMapping中Action.Index最小的SlotMapping
//就去找Action.State為pending的SlotMapping中Action.Index最小的SlotMapping
return minActionIndex(func(m *models.SlotMapping) bool {
return m.Action.State == models.ActionPending
})
}()
if m == nil {
return 0, false, nil
}
if update != nil && !update(m) {
return 0, false, nil
}
log.Warnf("slot-[%d] action prepare:\n%s", m.Id, m.Encode())
//變更每個SlotMapping的action.state，並與zk互動
//另外，Action.state符合preparing或者prepared的時候，要根據SlotMapping的引數同步到Slot
switch m.Action.State {
case models.ActionPending:
defer s.dirtySlotsCache(m.Id)
//Action.State指向下一階段
m.Action.State = models.ActionPreparing
//只是更新zk
if err := s.storeUpdateSlotMapping(m); err != nil {
return 0, false, err
}
fallthrough
case models.ActionPreparing:
defer s.dirtySlotsCache(m.Id)
log.Warnf("slot-[%d] resync to prepared", m.Id)
m.Action.State = models.ActionPrepared
//同步SlotMapping操作，後面會有介紹
if err := s.resyncSlotMappings(ctx, m); err != nil {
log.Warnf("slot-[%d] resync-rollback to preparing", m.Id)
m.Action.State = models.ActionPreparing
s.resyncSlotMappings(ctx, m)
log.Warnf("slot-[%d] resync-rollback to preparing, done", m.Id)
return 0, false, err
}
if err := s.storeUpdateSlotMapping(m); err != nil {
return 0, false, err
}
fallthrough
case models.ActionPrepared:
defer s.dirtySlotsCache(m.Id)
log.Warnf("slot-[%d] resync to migrating", m.Id)
m.Action.State = models.ActionMigrating
if err := s.resyncSlotMappings(ctx, m); err != nil {
log.Warnf("slot-[%d] resync to migrating failed", m.Id)
return 0, false, err
}
if err := s.storeUpdateSlotMapping(m); err != nil {
return 0, false, err
}
fallthrough
case models.ActionMigrating:
return m.Id, true, nil
case models.ActionFinished:
return m.Id, true, nil
//如果不屬於以上任何一種情況，直接返回invalid
default:
return 0, false, errors.Errorf("slot-[%d] action state is invalid", m.Id)
}
}很顯然，上面的方法取出的最小的Action.State的Slotmapping是
當一個SlotMapping處於preparing和prepared轉檯的時候，會將其狀態推進到下一階段，並同步SlotMapping，根據[]*models.SlotMapping建立1024個models.Slot，再填充1024個pkg/proxy/slots.go中的Slot，此過程中Router為每個Slot都分配了對應的backendConn。下面就來看看這個同步方法。
func (s *Topom) resyncSlotMappings(ctx *context, slots ...*models.SlotMapping) error {
if len(slots) == 0 {
return nil
}
var fut sync2.Future
for _, p := range ctx.proxy {
fut.Add()
go func(p *models.Proxy) {
//ApiClient中儲存了proxy的address以及xauth資訊。其中xauth是根據ProductName，ProductAuth以及proxy的token生成的
err := s.newProxyClient(p).FillSlots(ctx.toSlotSlice(slots, p)...)
if err != nil {
log.ErrorErrorf(err, "proxy-[%s] resync slots failed", p.Token)
}
fut.Done(p.Token, err)
}(p)
}
for t, v := range fut.Wait() {
switch err := v.(type) {
case error:
if err != nil {
return errors.Errorf("proxy-[%s] resync slots failed", t)
}
}
}
return nil
}同步的過程中有兩個方法比較複雜，分別是FillSlots和toSlotSlice。這一節我們先來看toSlotSlice。這個方法實際上就是將SlotMapping切片轉化為Slot切片，在Slot結構體重記錄了這個Slot在遷移的不同階段，接到的請求由哪個BackendAddr進行處理。
type Slot struct {
Id int `json:"id"`
Locked bool `json:"locked,omitempty"`
BackendAddr string `json:"backend_addr,omitempty"`
BackendAddrGroupId int `json:"backend_addr_group_id,omitempty"`
MigrateFrom string `json:"migrate_from,omitempty"`
MigrateFromGroupId int `json:"migrate_from_group_id,omitempty"`
ForwardMethod int `json:"forward_method,omitempty"`
ReplicaGroups [][]string `json:"replica_groups,omitempty"`
}func (ctx *context) toSlotSlice(slots []*models.SlotMapping, p *models.Proxy) []*models.Slot {
var slice = make([]*models.Slot, len(slots))
for i, m := range slots {
slice[i] = ctx.toSlot(m, p)
}
return slice
}
func (ctx *context) toSlot(m *models.SlotMapping, p *models.Proxy) *models.Slot {
slot := &models.Slot{
Id: m.Id,
Locked: ctx.isSlotLocked(m),
ForwardMethod: ctx.method,
}
switch m.Action.State {
case models.ActionNothing, models.ActionPending:
//這個getGroupMaster實際上就是從每個Group中取出第一臺，因為codis中認定group中新增的第一臺是主伺服器
slot.BackendAddr = ctx.getGroupMaster(m.GroupId)
slot.BackendAddrGroupId = m.GroupId
slot.ReplicaGroups = ctx.toReplicaGroups(m.GroupId, p)
case models.ActionPreparing:
slot.BackendAddr = ctx.getGroupMaster(m.GroupId)
slot.BackendAddrGroupId = m.GroupId
case models.ActionPrepared:
fallthrough
case models.ActionMigrating:
slot.BackendAddr = ctx.getGroupMaster(m.Action.TargetId)
slot.BackendAddrGroupId = m.Action.TargetId
slot.MigrateFrom = ctx.getGroupMaster(m.GroupId)
slot.MigrateFromGroupId = m.GroupId
case models.ActionFinished:
slot.BackendAddr = ctx.getGroupMaster(m.Action.TargetId)
slot.BackendAddrGroupId = m.Action.TargetId
default:
log.Panicf("slot-[%d] action state is invalid:\n%s", m.Id, m.Encode())
}
return slot
}其中如果slot處於migrating狀態，migrate.bc就不為空，如果恰好有請求發到這個slot，proxy就會執行一次SLOTSMGRTTAGONE讓這個slot遷移完成，再由其backend.bc來執行請求
下面的ReplicaGroups是專門為了主從讀寫分離設定的。從優先順序我們可以看到，讀請求會優先轉發到和proxy在一臺伺服器上的codis-server，也就是優先順序最高的0
func (ctx *context) toReplicaGroups(gid int, p *models.Proxy) [][]string {
g := ctx.group[gid]
switch {
case g == nil:
return nil
case g.Promoting.State != models.ActionNothing:
return nil
case len(g.Servers) <= 1:
return nil
}
var dc string
var ip net.IP
if p != nil {
dc = p.DataCenter
ip = ctx.lookupIPAddr(p.AdminAddr)
}
//replica的訪問優先順序
getPriority := func(s *models.GroupServer) int {
if ip == nil || dc != s.DataCenter {
return 2
}
if ip.Equal(ctx.lookupIPAddr(s.Addr)) {
return 0
} else {
return 1
}
}
var groups [3][]string
for _, s := range g.Servers {
if s.ReplicaGroup {
p := getPriority(s)
groups[p] = append(groups[p], s.Addr)
}
}
var replicas [][]string
for _, l := range groups {
if len(l) != 0 {
replicas = append(replicas, l)
}
}
return replicas
}有關於FillSlot和每個槽的處理方法processSlotAction，我們會在下一篇講