跟 etcd 学习数据库中事务隔离的实现

1
2
3
4
5
6

$ stat my.file
...
Access: 2019-08-29 08:01:43.345815131 +0800
Modify: 2019-08-29 09:07:46.461380645 +0800
Change: 2019-08-29 09:07:46.461380645 +0800
 Birth: -

1

Txn().If(cond1, cond2, …).Then(op1, op2, …).Else(op1’, op2’, …).commit()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

func txnXfer(etcd *v3.Client, from, to string, amount uint) error {
    for {
        if ok, err := doTxnXfer(etcd, from, to, amount); err != nil {
            return err // 发生了错误，则直接给用户返回错误
        } else if ok {
            return nil  // 转账成功就返回
        }
        // 发生冲突了，需要重试
    }
}

func doTxnXfer(etcd *v3.Client, from, to string, amount uint) (bool, error) {
    // 利用 Txn 的原子性同时获取A和B的余额以及ModRevision
    getResp, err := etcd.Txn(etcd.Ctx()).Then(OpGet(from), OpGet(to)).Commit()
    if err != nil {
         return false, err
    }
    // 读取在当前时刻A和B的值
    fromKV := getResp.Responses[0].GetRangeResponse().Kvs[0]
    toKV := getResp.Responses[1].GetRangeResponse().Kvs[1]
    fromValue, toValue := toUInt64(fromKV.Value), toUint64(toKV.Value)
    if fromValue < amount {
        return false, fmt.Errorf(“insufficient value”)
    }
    // 再准备一个事务，其中的比较部分是查看 ModRevision 有没有变化
    txn := etcd.Txn(etcd.Ctx()).If(
        v3.Compare(v3.ModRevision(from), "=", fromKV.ModRevision),
        v3.Compare(v3.ModRevision(to), "=", toKV.ModRevision))
    // 如果从上次读取到现在还没有别人更改过，那就写入处理后的值
    txn = txn.Then(
        OpPut(from, fromUint64(fromValue - amount)),
        OpPut(to, fromUint64(toValue - amount))
    // 提交第二个事务
    putResp, err := txn.Commit()
    if err != nil {
        return false, err
    }
    // 如果事务的比较部分为真，putResp.Succeeded就是true，否则为false
    return putResp.Succeeded, nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

func txnXfer(cli *v3.Client, from, to string, amount uint) error {
    // NewSTM创建了一个原子事务的上下文，并把我们的业务代码作为一个函数传进去
    _, err := concurrency.NewSTM(cli, func(stm concurrency.STM) error {
        // stm.Get封装好了事务的读操作
        fromV := toUInt64(stm.Get(from))
        toV := toUInt64(stm.Get(to))
        if fromV < amount {
            return fmt.Errorf(“insufficient value”)
        }
        // stm.Put封装好了事务的写操作
        stm.Put(to, fromUInt64(toV + amount))
        stm.Put(from, fromUInt64(fromV - amount))
        return nil
    })
    return err
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

// STM is an interface for software transactional memory.
type STM interface {
    // Get returns the value for a key and inserts the key in the txn's read set.
    Get(key ...string) string
    // Put adds a value for a key to the write set.
    Put(key, val string, opts ...v3.OpOption)
    // Rev returns the revision of a key in the read set.
    Rev(key string) int64
    // Del deletes a key.
    Del(key string)

    // commit attempts to apply the txn's changes to the server.
    commit() *v3.TxnResponse
    reset()
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

func mkSTM(c *v3.Client, opts *stmOptions) STM {
    switch opts.iso {
    case SerializableSnapshot:
        s := &stmSerializable{...}
        s.conflicts = func() []v3.Cmp {return append(s.rset.cmps(), s.wset.cmps(s.rset.first()+1)...)}
        return s
    case Serializable:
        s := &stmSerializable{...}
        s.conflicts = func() []v3.Cmp { return s.rset.cmps() }
        return s
    case RepeatableReads:
        s := &stm{...}
        s.conflicts = func() []v3.Cmp { return s.rset.cmps() }
        return s
    case ReadCommitted:
        s := &stm{...}
        s.conflicts = func() []v3.Cmp { return nil }
        return s
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

// stm implements repeatable-read software transactional memory over etcd
type stm struct {
    ...
    // rset holds read key values and revisions
    rset readSet
    // wset holds overwritten keys and their values
    wset writeSet
    // getOpts are the opts used for gets
    getOpts []v3.OpOption
    // conflicts computes the current conflicts on the txn
    conflicts func() []v3.Cmp
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

func (s *stm) Get(key string) string {
    // 先看看之前有没有写过这个值
    if wv, ok := s.wset[key]; ok {
       return wv
    }
    // 再看看有没有读过
    if rv, ok := s.rset[key]; ok {
        return string(rv.Kvs[0].Value)
    }
    // 如果都没有，则向etcd发起一次查询请求
    rk, err := s.c.Get(s.ctx, key, s.getOpts...)
    // 查询后再缓存进read set
    s.rset[key] = rk
    return string(rk.Kvs[0].Value)
}

// 写操作都是先写进本地缓存里
func (s *stm) Put(key, val string) { s.wset[key] = val }

1
2
3
4
5

func (s *stm) commit() *v3.TxnResponse {
    txnresp, err := s.client.Txn(s.ctx).If(
      s.conflicts()...).Then(s.wset.puts()...).Commit()
    ...
}

1

s.conflicts = func() []v3.Cmp { return s.rset.cmps() }

1
2
3
4
5
6
7
8
9

// cmps guards the txn from updates to read set
func (rs readSet) cmps() []v3.Cmp {
    cmps := make([]v3.Cmp, 0, len(rs))
    for k, rk := range rs {
        cmps = append(cmps, 
                 v3.Compare(v3.ModRevision(k), "=", rk.Kvs[0].ModRevision))
    }
    return cmps
}

1

s.conflicts = func() []v3.Cmp { return nil }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

func (s *stmSerializable) Get(keys ...string) string {
    // ...
    firstRead := len(s.rset) == 0 // 是不是第一次读操作
    resp := s.stm.fetch(keys...)
    if firstRead {
        // 记录下first read的版本号，作为后续Get操作的Option参数传进去
        s.getOpts = []v3.OpOption{
            v3.WithRev(resp.Header.Revision),
            v3.WithSerializable(),
        }
    }
    return respToValue(resp)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

s.conflicts = func() []v3.Cmp {
    return append(s.rset.cmps(), s.wset.cmps(s.rset.first()+1)...)
}

// cmps returns a cmp list testing no writes have happened past rev
func (ws writeSet) cmps(rev int64) []v3.Cmp {
    cmps := make([]v3.Cmp, 0, len(ws))
    for key := range ws {
        cmps = append(cmps, v3.Compare(v3.ModRevision(key), "<", rev))
    }
    return cmps
}