博客

4.1.1   增加hash_slots

typedef struct redisDb {
    dict *dict;                 /* The keyspace for this DB */
    dict *expires;              /* Timeout of keys with a timeout set */
    ...
    dict *hash_slots[HASH_SLOTS_SIZE];
} redisDb;

initServer() {
    for (i = 0; i < HASH_SLOTS_SIZE; i ++) {
        server.db[j].hash_slots[i] = dictCreate(&hashSlotType, NULL);
    }

void dbAdd(redisDb *db, robj *key, robj *val) {
    sds copy = sdsdup(key->ptr);

    int retval = dictAdd(db->dict, copy, val);

    do {
        uint32_t crc;
        int slot = slots_num(key->ptr, &crc);
        dictAdd(db->hash_slots[slot], sdsdup(key->ptr), (void *)(long)crc);
    } while (0);
    ...
}

增加了一些命令:

{"slotsinfo",slotsinfoCommand,-1,"rF",0,NULL,0,0,0,0,0},
{"slotsdel",slotsdelCommand,-2,"w",0,NULL,1,-1,1,0,0},
{"slotsmgrtslot",slotsmgrtslotCommand,5,"aw",0,NULL,0,0,0,0,0},
{"slotsmgrtone",slotsmgrtoneCommand,5,"aw",0,NULL,0,0,0,0,0},
{"slotsmgrttagslot",slotsmgrttagslotCommand,5,"aw",0,NULL,0,0,0,0,0},
{"slotsmgrttagone",slotsmgrttagoneCommand,5,"aw",0,NULL,0,0,0,0,0},
{"slotshashkey",slotshashkeyCommand,-1,"rF",0,NULL,0,0,0,0,0},
{"slotscheck",slotscheckCommand,0,"r",0,NULL,0,0,0,0,0},
{"slotsrestore",slotsrestoreCommand,-4,"awm",0,NULL,1,1,1,0,0},

4.1.2   阻塞迁移

static int
slotsmgrt(redisClient *c, sds host, sds port, int fd, int dbid, int timeout, robj *keys[], robj *vals[], int n) {
    ...
    syncWrite(fd, buf + pos, towrite, timeout);
    syncReadLine(fd, buf1, sizeof(buf1), timeout)
}

4.1.3   问题

1. 它会记录每个slot有哪些key, 这就会导致key多存一份, 这在某些情况下带来的内存消耗大约是1.5倍.(详见后面测试)
   • 而对redis来说, 空间很珍贵, 这里可以用时间换空间, 会更好.

2. 阻塞迁移造成的短时拒绝服务(详见后面测试).

4.2.1   actions

• codis 对每个topo变化的操作, 都会记录到actions, 同时对于某些action, 会要求每个proxy ack确保更新到最新拓扑.
  • TODO: 如何实现确保ack

4.2.2   router.go

加载拓扑信息(一个slot指向那个group):

//use it in lock
func (s *Server) fillSlot(i int, force bool) {
    slotInfo, groupInfo, err := s.top.GetSlotByIndex(i)             # 从zk中获取slot信息

    slot := &Slot{
        slotInfo:  slotInfo,
        dst:       group.NewGroup(*groupInfo),
        groupInfo: groupInfo,
    }

    s.pools.AddPool(slot.dst.Master())                              # 准备连接池

    if slot.slotInfo.State.Status == models.SLOT_STATUS_MIGRATE {   # 处理MIGRATE状态.
        //get migrate src group and fill it
        from, err := s.top.GetGroup(slot.slotInfo.State.MigrateStatus.From)
        slot.migrateFrom = group.NewGroup(*from)
        s.pools.AddPool(slot.migrateFrom.Master())
    }

    s.slots[i] = slot
}

#如果状态是migrate 的slot, 发migrate命令
func (s *Server) handleMigrateState(slotIndex int, key []byte) error {
    ...
    err = WriteMigrateKeyCmd(redisConn.(*redispool.PooledConn), shd.dst.Master(), 30*1000, key)
    ...
}

转发逻辑, 读一个请求, 向后端转发一个:

func (s *Server) redisTunnel(c *session) error {
    resp, err := parser.Parse(c.r) // read client request

    op, k, err := getOpKey(resp)
    i := mapKey2Slot(k)

check_state:        # 这里是一个循环来检查, 等待SLOT_STATUS_PRE_MIGRATE结束
    s.mu.RLock()
    if s.slots[i] == nil {
        s.mu.Unlock()
        return errors.Errorf("should never happend, slot %d is empty", i)
    }
    //wait for state change, should be soon
    if s.slots[i].slotInfo.State.Status == models.SLOT_STATUS_PRE_MIGRATE {
        s.mu.RUnlock()
        time.Sleep(10 * time.Millisecond)
        goto check_state
    }

    s.handleMigrateState(i, k);

    //get redis connection
    redisConn, err := s.pools.GetConn(s.slots[i].dst.Master())
    redisErr, clientErr := forward(c, redisConn.(*redispool.PooledConn), resp)
}

func (s *Server) handleConn(c net.Conn) {
    for {
        err = s.redisTunnel(client)
        client.Ops++
    }
}

func (s *Server) Run() {
    log.Info("listening on", s.addr)
    listener, err := net.Listen("tcp", s.addr)
    for {
        conn, err := listener.Accept()
        go s.handleConn(conn)       #起一个
    }
}

4.2.3   更新路由

proxy 会watch action 树下的变更, 有变化时 重新加载路由:

func (s *Server) OnGroupChange(groupId int) {
    log.Warning("group changed", groupId)

    for i, slot := range s.slots {
        if slot.slotInfo.GroupId == groupId {
            s.fillSlot(i, true)
        }
    }
}

4.2.4   pipeline

上面代码是@ngaut同学pipeline优化前的代码,

一般来说, 实现pipeline可能存在下面两个问题, 不过测试发现:codis都没有问题
.

1. 返回乱序:

   get     k1 k2 k3 k4
   return  v2 v1 v3 vj
   

原因是k1,k2发到不同后端, 如果其中一个后端很慢, 而先返回的后端就先写客户端, 就是这个错误.

测试发现codis没有这个问题, 但是看代码没有看懂为什么. 涉及到多个channel中传递消息(TODO).

1. 乱序执行:

   lpush   lst 1
   lpush   lst 2
   lpush   lst 3
   lpush   lst 4
   
   lpop return 2 1 3 4
   

第二种情况是发到同一个后端, 但是如果向同一个后端有多个连接, 就可能出这个问题.
某个连接上的请求先执行.

codis一个后端只有一个TaskRunner(一个连接), 所以应该不会出这个问题.

4.2.5   mget

逐一访问:

func (oper *MultiOperator) mgetResults(mop *MulOp) ([]byte, error) {
    results := make([]interface{}, len(mop.keys))
    conn := oper.pool.Get()
    defer conn.Close()
    for i, key := range mop.keys {
        replys, err := redis.Values(conn.Do("mget", key))

        for _, reply := range replys {
            if reply != nil {
                results[i] = reply
            } else {
                results[i] = nil
            }
        }
    }

    b, err := respcoding.Marshal(results)
    return b, errors.Trace(err)
}

这是为了保证迁移过程中的一致性, 必须一个一个处理.

性能较差.

5.3.1   性能问题

spinlock同学的测试:

https://github.com/wandoulabs/codis/issues/63

CONC        PIPELINE        CODIS-LATENCY   REDIS-LATENCY
50      10          3.17            0.60
50      20          5.88            0.89
50      75          21.78           2.40p

@ngaut 同学在15年2月实现了pipeline的支持:https://github.com/wandoulabs/codis/pull/110

• 用go来实现pipeline的问题是:
  • pipeline并不是一个命令, pipeline实际上是异步处理所带来的一个好处, 只是一种使用方法.
  • 理论上, proxy/redis都不需要做任何事情来支持pipeline.
  • 但是codis采用了逐一转发的方法来处理请求, 就需要对pipeline专门处理.
  • 需要把go写成异步处理的形式, go 带来的编程模型简单的好处就没有了.
  • 优化后, 对所有的请求统一使用同样的方法(proxy并不知道自己处在一个pipeline中)

pipeline 遇到slot迁移状态性能又是一个严重的问题.