etcd-v3.5release-(2)-STM

a.b.c体现a文件里的b类的方法c，注意a不愿定是包名，由于文件名不愿定即是包名
！！！！etcd在put的过程中使用的batchTxBuffered，这个事故是写bbolt数据库使用的事故，是对bbolt.Tx的一个封装，就是攒一批事故作为一个大事故，一次性提交到bbolt。etcd对外服务得像一个数据库，但是外部看到的不是bbolt，也就是说外部对etcd提交的事故和etcd用来写bbolt时使用的事故完满是两个东西，etcd使用的事故是STM（软件事故内存）
！！！etcd smt在单个事故中对同一key的修改在提交的时间只有终极的修改有效，由于他用了一个rset/wset分别体现客户端在事故中要读取哪些key和修改哪些key，在一条path中对某个key只答应有一个put使用，假犹如一条path里有对同一个key的两个put使用，那么clientsdk就会报错
STM、ReadView
！！！！总的来说，etcd的stm事故模子是在客户端实现的，但事故的实行是在etcd集群中的成员节点上完成的。
apply的时间是一条一条的应用的，但是如果要原子的同时apply一批日志，那么就必须使用STM
STM内部包罗一个wset和一个rset，生存了本次事故访问到的数据和版本，然后在提交的时间如果发现事故变了，那么就会报错
commited时日志的索引次序和apply时revision次序是划一的
！！！wal固然是次序写入的，apply的时间也是次序提交到bbolt的，apply的时间他是会把从chan收到的数据包装成一个job，然后丢到一个fifo队列，这个fifo队列会等当前使命运行完毕才会运行下一个job，也就是apply的时间也肯定是按次序apply的，也就是说只有上一个事故提交到bbolt才会开始提交下一个事故（看了源码），如许在开始x事故的时间x从前的日志都已经写完了，也就是说如果x事故读取的数据被x-k版本的事故修改了，那么x事故是可以检测出来的，如果检测到辩论，那么就不会把该数据写入数据库，也便是说纵然日志已经写入了磁盘，但是数据是不愿定会写入数据库的。举个例子：一个事故t1，先读取key a，此时a的版本(即revision)为v1，然后事故t1在当地实行了很久才提交事故，然后t1提交到etcd的时间raft日志索引为index1，然后apply的时间对应的revision为 rev1，然后实行的时间检测到key a此时的版本是v1+x，但是他读的时间a的版本为v1，以是此时就检测到了辩论（从treeIndex检测，由于treeIndex包罗全部key和key对应的revision，比力一下revision就行），那么事故就会实行失败，固然事故实行失败，但是日志是apply乐成的，也就是说事故实行的结果就是数据不写入数据库。再举个例子，假设事故t1对应的raft日志已经写入磁盘，然后apply但是还没apply完成的时间体系瓦解了，这个不要紧，重做就是了，当再次重做到事故t1对应的日志的时间，由于t1从前的日志肯定已经重做完了，以是此时还是可以再次检测到辩论的，以是只要wal日志已经长期化了，那么瓦解就对这些已经长期化的日志无影响。
！！！区分revision和applyIndex：etcd的修改使用比如txn大概put，底层都是提交一个事故到bbolt，但是底层是batchTxBuffered即批量提交的，以是就是说上层实行完一个事故后，事故缓存在batchTxBuffered中并没有提交，但是只要把事故提交到了batchTxBuffered，s.currentRev就会+1，也就是说下一次put大概writetxn事故分配的revision就是s.currentRev了，但是！！！在底层batchTxBuffered提交事故前，新增的数据对外部是不可见的，假设batchTxBuffered提交时，本批次最大的revision为x，一旦提交以后，那么rev<=x的素有key都对外可见了，但是x<rev<=s.currentRev之间的而全部key此时对外是不可见的，也就是说可见的数据是落后于已做的数据（由于底层没有提交，以是这些数据对外不可见），这与applyIndex落后于commiteIndex原理是划一的，已经apply的数据对外可见，但是已经commite但是还没有apply的数据不可见
STM内对同一个值多次修改，那么只有末了一次写会乐成，由于他是先在当地实行函数，然后得到一个终极的结果，再在commited时间直接提交终极的结果
！！mvcc写，也就是说多个版本
Server上Txn哀求流程,分Read和Write
！！！etcd直接提供的if-then-else不支持嵌套事故，但是不知道提供的NewSTM支不支持嵌套事故，有待确定，但是etcd server是肯定支持嵌套事故的
batchtxn内部有一个batchtx体现末了一个事故
！！！stm原理：

1. stm if xx then yy else zz
2. stm执行时先判断if里面的条件是否成立，一旦成立就表示true就执行then 路径，一旦失败就表示false就执行else路径，then和else都叫做txnpath即事务路径。stm是允许嵌套stm的比如if里面有一个子if-then-else或者then里面还有if-then-else或者else里也有子if-then-else，当然，也允许子stm里还有子stm。then-else里面的叫做操作即op,因为可能有很多，所以叫做ops。所以stm对于嵌套事务的执行是这样的：先执行最顶层if，然后根据if结果true或者false选择对应的路径，然后执行路径下的op，每个op可以是子事务也可以是简单的put/get/range等操作，然后就根据op执行对应的操作就行，如果是子事务，就再执行一遍if-then-else流程就行

复制代码

etcd stm代码使用样例（抄自网络）:

1. func getEtcdClient() *v3.Client {
2.         // 配置 etcd 客户端
3.         cfg := v3.Config{
4.                 Endpoints:   []string{"30.128.101.96:2379"}, // 集群中的任意一个节点
5.                 DialTimeout: 5 * time.Second,
6.         }
8.         // 创建 etcd 客户端
9.         cli, err := v3.New(cfg)
10.         if err != nil {
11.                 log.Fatal(err)
12.         }
13.         return cli
15. }
17. func stm() {
18.         client := getEtcdClient()
19.         txnTransfer(client, "k1", "k2")
20. }
21. func txnTransfer(etcd *v3.Client, sender, receiver string) error {
22.         // 失败重试
23.         for {
24.                 if ok, err := doTxn(etcd, sender, receiver); err != nil {
25.                         fmt.Println("dotxn failed:", err.Error())
26.                 } else if ok {
27.                         fmt.Println("dotxn success")
28.                 }
29.         }
30. }
31. func doTxn(etcd *v3.Client, sender, receiver string) (bool, error) {
32.         fmt.Println("\n开始事务前的读取操作:")
33.         getresp, err := etcd.Txn(context.TODO()).Then(v3.OpGet(sender), v3.OpGet(receiver)).Commit()
34.         if err != nil {
35.                 return false, err
36.         }
37.         senderKV := getresp.Responses[0].GetResponseRange().Kvs[0]
38.         receiverKV := getresp.Responses[0].GetResponseRange().Kvs[0]
39.         // 发起转账事务，冲突判断 ModRevision 是否发生变化
40.         var s string
41.         fmt.Println("senderKv: revision=" + strconv.FormatInt(getresp.Header.Revision, 10) +
42.                 " key=" + string(senderKV.Key) + "  value=" + string(senderKV.Value))
43.         fmt.Println("senderKv: revision=" + strconv.FormatInt(getresp.Header.Revision, 10) +
44.                 " key=" + string(receiverKV.Key) + "  value=" + string(receiverKV.Value))
45.         fmt.Println("请输入newValue:")
46.         _, err = fmt.Scan(&s)
47.         if err != nil {
48.                 fmt.Println("scan error:", err.Error())
49.                 return false, err
50.         }
51.         fmt.Println("开始执行事务：修改key的值为newValue......")
52.         txn := etcd.Txn(context.TODO()).If(                                     #etcd stm是通过if来判断是否可以执行
53.                                                                             #一旦if里面设置的条件通过了
54.                                                                             #那么就说明该事物没有发生冲突可以执行
55.                                                                             #这是if的两个compare即两个比较器
56.                 v3.Compare(v3.ModRevision(sender), "=", senderKV.ModRevision),      #compare1: key=k1,比较字段=ModRevision,result="="
57.                                                                             #result表示我们用什么比较方式
58.                                                                             #即比较看k1这个key的ModRevision这个字段
59.                                                                             #在事务执行时是否和旧的ModRevision相等
60.                                                                             #如果不相等则说明在这期间被其他人编辑了，
61.                                                                             #就说明if失败即事务失败
62.                 v3.Compare(v3.ModRevision(receiver), "=", receiverKV.ModRevision))  #compare2: key=k2,比较字段=ModRevision,result="="
63.                                                                             #即比较看k2这个key的ModRevision这个字段
64.                                                                             #在事务执行时是否和旧的ModRevision相等
65.                                                                             #如果不相等则说明在这期间被其他人编辑了
66.                                                                             #！！！比较哪些key是任意的，是自己选的
67.                                                                             #！！！也支持一次比较多个key  
68.                                                                             #！！！当然还支持比较key的其他字段比如xx
69.                                                                             #！！！当然也支持其他result即>、<、=、!=等
70.                                                                             #！！！我们这里只比较ModRevision且用的是=
72.     txn = txn.Then(                                                         #path then里面又有两个op，每个op又是一个put操作
73.                 v3.OpPut(sender, s),   
74.                 v3.OpPut(receiver, s),
75.         ).Else(                                                                 #path else里面又有两个op，每个op又是一个put操作
76.                 v3.OpPut(sender, "faild to set"),
77.                 v3.OpPut(receiver, "faild to set"),
78.         )  
79.         resp, err := txn.Commit()
80.         fmt.Println("事务执行完成")
81.         if err != nil {
82.                 return false, err
83.         }
84.         return resp.Succeeded, nil
85. }

复制代码

3. //txn事务有两种，一种是读事务，一种是写事务，通通都是通过这个函数来处理，读事务比较简单，重点在于后半段的写事务
4. //！！！！etcd只读事务不走raft流程，直接本地读
5. //！！！！个人疑惑：只读事务里面读不需要readIndex吗？
6. key.quotaKVServer.Txn
7.   key.kvServer.Txn
8.     if isTxnReadOnly                                  #如果是只读事务，那么不需要走raft请求，直接事务读就行
9.       if !isTxnSerializable(r):                       #判断是不是线性读，!isTxnSerializable表示线性读
10.         s.linearizableReadNotify(ctx)                 #如果需要线性读则阻塞等待直到读的条件满足
11.                                                       #就是记录当前的commitedIndex为cx，然后等待，直到applyIndex>=cx
12.       s.doSerialize                                   #此时applyIndex>=cx，所以可以执行序列化读
13.         apply.applierV3backend.Txn                    #事务读也是一个事务操作，只不过这个事务执行的是range读操作
14.                                                       #所以这里的具体操作就是读取操作
15.            ...txn函数在下面详解
16.     else:                       
17.       etcdserver.EtcdServer.raftRequest               #写事务因为对数据库有修改，所以必须先走一遍raft日志流程
18.                                                       #也就是先写一条raft日志，然后再apply这条日志
19.                                                       #是的，所有对数据库有修改的操作都叫apply，
20.                                                       #只不过apply可以进一步细分为put/txn等很多操作
21.         ...raft写日志流程略...                          #1:raft写日志流程，略
22.         etcdserver.apply                              #2:apply日志，这里就是执行写事务
23.           ......
24.             apply.applierV3backend.Apply              #所有修改数据库的操作都可以叫做apply
25.               case r.Txn != nil:                      #apply可以进一步细分，这里是实务操作，即txn
26.                         op = "Txn"  
27.                         apply_auth.authApplierV3.Txn          #鉴权准入
28.                   apply.quotaApplierV3.Txn            #配额准入
29.                     apply.applierV3backend.Txn        #执行事务
30.                       isWrite := !isTxnReadonly(rt)   #判断是不是写事务
31.                       if isWrite && a.s.Cfg.ExperimentalTxnModeWriteWithSharedBuffer:
32.                         readTx=kvstore_txn.store.Read(mvcc.SharedBufReadTxMode) #共享buf读事务，可以避免复制buffer
33.                                                                     #因为是共享buf，所以会对buf加读锁
34.                           s.mu.RLock()                              #kvstore代表数据库，kvstore是一个watchablestore对象
35.                                                                     #watchablestore对象内部又含有一个store对象
36.                                                                     #watchablestore和store对象都有一个mu对象
37.                                                                     #但是watchablestore对象没有Read函数而store对象有
38.                                                                     #所以go这里kvstore.Read代表的就是store的Read函数
39.                                                                     #所以store.Read函数里使用的肯定就是store.mu了
40.                                                                     #操蛋的go语法，刚开始没了解数据结构，迷惑了好久
41.                                                                     #！！！这里是给store.mu加读锁即给数据库加读锁
42.                                                                     #！！！加读锁后会阻止数据库写操作
43.                                                                     #！！！也就是说虽然读写会并发进行
44.                                                                     #！！！但是读的时候会阻止写
45.                                                                     #！！！注意：这里是store.mu后面还有把watchable.mu锁
46.    
47.                                                                     #！！！简单粗暴，s.mu.RLock会在txn.End中释放
48.                           s.revMu.RLock()                           #临时给revMu加读锁，
49.                                                                     #因为我们要读取curretnRev和compactMainRev
50.                           var tx backend.ReadTx                     #backend.ReadTx是对bolt readtx的一个封装
51.                           if mode == ConcurrentReadTxMode:          #根据mode参数创建对应的txn
52.                             tx = s.b.ConcurrentReadTx()             #conncurrentReadTx创建一个新的readTx,并且复制buf
53.                                                                     #readBuf中保存的是最新commit的数据
54.                               backend.readTx.RLock()                #给readBuf加读锁，因为事务提交的时候会把writebuf写回到readbuf
55.                               backend.readTx.txWg.Add(1)            #？？？不太懂，bbolt回滚和事务机制
57.                                                       
58.                                                          
59.                               curBuf := b.readTx.buf.unsafeCopy()   #复制readBuf，readTx.buf就是readBuf
60.                                                                     #提交事务的时候会把batchTxBuffered的writeBuffer
61.                                                                     #中的内容写到readTx.buf
62.                                                                     #查询操作先查readBuf，然后再查bbolt数据库
63.                               backend.readTx.RUnlock()              #！！！操作完后解锁
64.                                                                     #！！！也就是说复制完readBuf后就会释放事务读锁了
65.                                                                     #！！！后续读操作就是读readBuf的副本了
66.                                                                     #！！！所以不用担心commite的时候会用writebuf覆盖readBuf
67.                                                                     #！！！虽然concurrentRead创建后就会释放readTx的读锁
68.                                                                     #！！！但是创建concurrentReadTx的时候需要对readTx加读锁
69.                                                                     #！！！所以如果有地方已经加了写锁，比如提交事务的时候
70.                                                                     #！！！那么这里也会等到写事务完成后才能成功加读锁
71.                           else:
72.                             tx = backend.backend.ReadTx()           #如果是sharedBuff模式
73.                               return b.readTx                       #则直接返回backend内部的readTx而不是新创一个readTx
74.                           tx.RLock()                                #因为如果是sharedbuff模式则是共用一个readTx
75.                             tx.mu.RLock                             #所以tx.RLock是一个mutex.RLock操作
76.                                                                     #如果是concurrentReadTx则tx.RLock是一个空操作
77.                                                                     #！！！etcd自己包装了bbolt.tx，即包装出一个readTx
78.                                                                     #！！！这个readTx主要同步读事务和写事务的
79.                                                                     #！！！etcd读流程会先从readbuf读，读不到再去bbolt读
80.                                                                     #！！！sharedBuff模式下所有请求都使用同一个readTx
81.                                                                     #！！！即都会访问同一个readTx.buf，所以读写需要加锁
82.                                                                     #！！！对这个readTx加读写锁来表示是否有读写事务正在运行
83.                                                                     #！！！如果是读操作那么就对这个readTx加读锁
84.                                                                     #！！！比如这里compare操作需要加读锁阻止修改
85.                                                                     #！！！如果是写操作则对这个readTx加写锁
86.                                                                     #！！！比如commit的时候就要对readTx加写锁阻止后续读
87.                                                                     #！！！直到写操作完成
88.                                                                     #总结一下两种模式：
89.                                                                     #concurrent模式:加读锁-复制-解读锁-读，加写锁-写-解写锁，
90.                                                                     #因为锁住事件断，这样就可以读、写交替，但是有复制的开销
91.                                                                     #shared模式:加锁-读-读-读-读-解锁-写
92.                                                                     #无复制，但是这样就必须全部读事务都读完写事务才能写，
93.                                                                     #这样就会导致写阻塞的时间可能偏长
94.               
96.       
97.                           firstRev, rev := s.compactMainRev, s.currentRev    #读取共享变量 compactMainRev/currentRev
98.                                                                              #因为会有其他goroute并发修改，所以读之前需要加锁
99.                                                                              #currentRev就是当前的revision
100.                           s.revMu.RUnlock()                                  #释放revMu读锁
101.                           return newMetricsTxnRead(&storeTxnRead{s, tx, firstRev, rev, trace}) #创建事务对象
102.                         txn = mvcc.NewReadOnlyTxnWrite(readTx)   
103.                       else:  
104.                         readTx=kvstore_txn.store.Read(mvcc.ConcurrentReadTxMode)#并发读事务，创建时会复制buffer
105.                                                                                 #所以读取的时候不会加锁
106.                         txn = mvcc.NewReadOnlyTxnWrite(readTx) #并发读
107.                                                       #！！！stm事务在执行比较前会创建一个只读视图用来确保视图一致性
108.                                                       #！！！即compare期间看到的数据是不变的
109.                                                       #！！！虽然apply是串行apply的即写是串行的但是读与写却不是串行的而是并发的
110.                                                       #！！！即可能同时进行读写操作，因为compare操作需要读数据库
111.                                                       #！！！所有compare操作时必须通过一个只读视图来compare
112.                                                       #！！！（个人猜测）：换句话说bbolt只能读或者写不能同时读写因为会加读写锁
113.                               
114.                         return &txnReadWrite{txn}
115.                       txnPath := make([]bool, 1)
116.                       txnPath = apply.compareToPath(txn, rt)             #！！！执行比较，stm事务的核心就是执行比较
117.                                                                          #！！！rt就代表一个事务if-then-else
118.                                                                          #！！！这里会递归遍历所有事务及子事务
119.                                                                          #！！！所有事务的比较结果都会顺序存放到txnPath数组里面
120.                                                                          #！！！也就是说compareToPath这个函数
121.                                                                          #！！！就是用来确定所有事务该走哪条path true还是false
122.                                                                          #！！！也就是说不管有多少个嵌套事务，最终的路径一定是确定的
123.                                                                          #！！！所以txnPath[i]是事务i，
124.                                                                          #！！！那么txnPath[i+1]必定是执行路径上的下一个事务的compare结果
125.                                                                          #！！！也就是说不管有多少个嵌套事务，最终的路径一定是确定的
126.                         txnPath[0] = apply.applyCompares(rv, rt.Compare) #rt.Compare即if块，一个if可以包含很多个比较器
127.                           for _, c := range rt.Compare:                  #c代表compare中的一个比较器,这里遍历所有比较器
128.                                                                          #我们的例子中含有两个比较器
129.                                                                          #compare1: key=k1,比较字段=ModRevision,result="="
130.                                                                          #compare2: key=k2,比较字段=ModRevision,result="="
131.                             ok2=apply.applyCompare(rv, c):               #这里以compare1为例
132.                               metrics_txn.metricsTxnWrite.Range          #读取比较器中的key
133.                               for _, kv := range rr.KVs:                 #遍历compare1中的所有的key，我们这里只有一个key，即k1
134.                                         ok3=apply.compareKV(c, kv):              #对该key应用比较器，这里是对key k1应用比较器compare1
135.                                   switch c.Target:                       #target表示要比较哪个字段，它支持很多字段
136.                                                                          #比如value/createVersion/ModVersion......
137.                                                                          #我们这里是用的ModVersion这个字段
138.                                   case pb.Compare_VALUE:
139.                                             tv=c.TargetUnion.(*pb.Compare_Value)
140.                                     v=tv.Value
141.                                             result = bytes.Compare(ckv.Value, v)
142.                                       case pb.Compare_CREATE:
143.                                             tv=c.TargetUnion.(*pb.Compare_CreateRevision)
144.                                             rev = tv.CreateRevision
145.                                             result = compareInt64(ckv.CreateRevision, rev)
146.                                       case pb.Compare_MOD:                           #我们用的是ModVersion这个字段
147.                                             tv= c.TargetUnion.(*pb.Compare_ModRevision)  #TargetUnion是一个联合体，
148.                                             rev = tv.ModRevision                         #表示此时请求中的字段表示的含义是ModRevision
149.                                             result = compareInt64(ckv.ModRevision, rev)  #执行比较，然后得到一个结果-1(小于),0(等于),1(大于)
150.                                       case pb.Compare_VERSION:                       #version是每个key独有的，
151.                                             tv= c.TargetUnion.(*pb.Compare_Version)
152.                                             rev = tv.Version
153.                                             result = compareInt64(ckv.Version, rev)
154.                                       case pb.Compare_LEASE:
155.                                             tv= c.TargetUnion.(*pb.Compare_Lease)
156.                                             rev = tv.Lease
157.                                             result = compareInt64(ckv.Lease, rev)
159.                                       switch c.Result:                                 #Result表示我们选择的比较方式是什么
160.                                                                                   #支持很多，比如=、!=、>、<等
161.                                       case pb.Compare_EQUAL:                          #我们这里compare1的k1选择的是=比较
162.                                             return result == 0                            #所以我们返回result是否等于0
163.                                                                                   #true表示if成功，那么稍后执行的就是then路径
164.                                                                                   #false表示if失败，那么稍后执行的就行else路径
165.                                       case pb.Compare_NOT_EQUAL:
166.                                             return result != 0
167.                                       case pb.Compare_GREATER:
168.                                             return result > 0
169.                                       case pb.Compare_LESS:
170.                                             return result < 0
171.                                         if !ok3:                                            #只要有任意一个key比较失败，就返回false
172.                                   return false                                      #就返回false
173.                               return true
174.   
175.                             if !ok2                                                #只要有任意一个比较器返回false即比较失败
176.                               return false                                         #那么就立即返回false
177.                           return true
178.                         ops := rt.Success                           
179.                         if !txnPath[0]:                                            #如果比较失败就执行false路径，否则执行success路径
180.                                   ops = rt.Failure                                         #success对应then,failure对应else
181.                         for _, op := range ops {                                   #选择好执行路径后就遍历该路径下的所有op
182.                           tv, ok := op.Request.(*pb.RequestOp_RequestTxn)
183.                           if !ok || tv.RequestTxn == nil:                          #compareToPath只是为了确定事务及其子事务该走哪条path
184.                             continue                                               #所以如果该op里不包含事务，就跳过
185.                           txnPath = append(txnPath, compareToPath(rv, tv.RequestTxn)...) #递归调用compareToPath来确定所有事务的路径
187.                       if isWrite:
188.                         apply.checkRequests(txn, rt, txnPath, a.checkPut)        #检查写请求正确性
189.                       apply.checkRequests(txn, rt, txnPath, a.checkRange)        #检查range请求正确性
191.                       if isWrite:  
192.                         metrics_txn.metricsTxnWrite.End                   #结束我们为了compare而创建的只读事务
193.                          kvstore_txn.storeTxnRead.End
194.                            read_tx.readTx.RUnlock                         #释放backend对应的事务的读锁
195.                                                                           #一个backend对应一个readTx
196.                            store.mu.RUnlock                               #释放store.mu的读锁
197.                         txn = watchable_store_txn.watchableStore.Write    #compare完毕，可以写了，所以这里创建写事务
198.                           txn=kvstore_txn.store.Write  
199.                             s.mu.RLock()                                  #store代表kvstore，写事务会先加读锁
200.                                                                           #等所有读事务完成后释放读锁并加写锁                  
201.                             tx := s.b.BatchTx()                           #获取底层的batchtx
202.                                                                           #底层数据结构：store含有一个backend
203.                                                                           #backend内部有一个batchTxBuffered
204.                                                                           #batchTxBuffered内嵌一个batchTx
205.                                                                           #所以调用batchTxBuffered.LockInsideApply
206.                                                                           #实际就是调用内嵌的batchTx的LockInsideApply
207.                                                                           #！！！batchTx对应批量提交，内部含有一个bbolt.tx对象
208.                                                                           #！！！所有事务都是直接写这个bblot.tx
209.                                                                           #！！！写完以后这个bbolt.tx不会提交
210.                                                                           #！！！对于上层来说事务只要写到bbolt.tx就会返回
211.                                                                           #！！！上层会等到bbolt.tx提交以后才会返回结果给client
212.                                                                           #！！！也就是异步+batch
213.                                                                           #！！！这个bbolt.tx可以写很多次都不提交
214.                                                                           #！！！对于上层来说，这个bbolt.tx里包含了一批事务
215.                                                                           #！！！但是对于底层bbolt来说自始至终都是一个事务
216.                                                                           #！！！即上层看到的批量事务在底层实际就是一个事务
217.                                                                           #！！！所以只需要提交一个事务就行，即batchTx.bbolt.tx
218.                             tx.LockInsideApply()                          #batchtx加写锁，
219.                             tw := &storeTxnWrite{
220.                               storeTxnRead: storeTxnRead{s, tx, 0, 0, trace},
221.                               tx:           tx,
222.                               beginRev:     s.currentRev,                        #！！！beginRev记录的是当前最后一次操作后的revision
223.                                                                                  #！！！执行事务里面的操作比如put的时候
224.                                                                                  #！！！用的revision是beginRev+1
225.                               changes:      make([]mvccpb.KeyValue, 0, 4),
226.                             return newMetricsTxnWrite(tw)
227.                           return &watchableStoreTxnWrite{s.store.Write(trace), s}
229.                       apply.applierV3backend.applyTxn                          #执行事务
230.                         reqs := rt.Success                                     #这里判断是走success路径
231.                         if !txnPath[0]:
232.                           reqs = rt.Failure                                    #还是走failure路径
233.                                                                                #！！！stm中不存在事务冲突
234.                                                                                #！！！stm中只有path  
235.                                                                                #！！！如果没有发生冲突即比较成功
236.                                                                                #！！！那么就走success路径
237.                                                                                #！！！如果发生冲突，那就走failure路径
238.                                                                                #！！！即不管冲突还是没有冲突，都有path可走
239.                                                                                #！！！当然，如果走success路径，
240.                                                                                #！！！但我们没有设置对应的语句，
241.                                                                                #！！！那么就相当于一个空操作，什么也不做
242.                         for i, req := range reqs:                              #遍历该path中的所有op
243.                           switch tv := req.Request.(type):                     #该op有很多类型
244.                             case *pb.RequestOp_RequestRange:                   #如果是range读取操作，那么就执行range读取操作
245.                               apply.applierV3backend.Range(txn, tv.RequestRange)
246.                             case *pb.RequestOp_RequestPut:                     #如果是put操作,那么就执行put操作，这里以put为例
247.                               apply.applierV3backend.Put(txn, tv.RequestPut)   #！！！那么就直接执行普通的put流程
248.                                                                                #！！！即使是stm事务，本质上也算一个操作
249.                                                                                #！！！也就是说每个stm事务都对应一个版本号
250.                                                                                #！！！也就是说客户端一次stm事务操作和客户端一个put操作
251.                                                                                #！！！本质上没有任何区别，都是一个修改数据库的操作
252.                                                                                #！！！都是先写一条日志，然后apply执行日志完成修改
253.                                                                                #！！！注意，这里的一次stm事务是指一次完整的事务操作
254.                                                                                #！！！stm事务以及他嵌套的所有子事务用的是同一个版本号
255.                                                                                #！！！换句话说，不管是子事务还是put等操作
256.                                                                                #！！！同一个事务中的所有操作都用同一个版本号
257.                                                                   
258.                             case *pb.RequestOp_RequestDeleteRange:             #如果是deleteRange操作，那么就执行deleteRange操作
259.                               apply.applierV3backend.DeleteRange(txn, tv.RequestDeleteRange)  
260.                             case *pb.RequestOp_RequestTxn:                     #如果是子事务，就递归执行applyTxn
261.                                                                                #！！！子事务也是使用txn这个事务对象
262.                                                                                #！！！也就是说子事务的版本号和父事务是同一个
263.                               apply.applierV3backend.applyTxn(txn, tv.RequestTxn, txnPath[1:], resp)
264.                                              txns += applyTxns + 1
265.                                           txnPath = txnPath[applyTxns+1:]                   #txnpath保存了执行路径上所有事物的path
266.                                                                                 #所以txnPath[applyTxns+1]就表示下一个事务的path索引
267.                             default:                                            #如果是非预期的操作类型，则啥也不干
269.                                           no-op
271.                       rev := txn.Rev()                                #获取txn开始时最新的revision
272.                       if len(txn.Changes()) != 0:                     #每次修改操作都会改变revision，读操作不会
273.                         rev++
274.                       watchable_store_txn.watchableStoreTxnWrite.End  #结束事务，主要是释放锁和提交事务
275.                             tw.s.mu.Lock()                                #！！！我TM真找不到在哪释放的读锁，难道不是同一个store?
276.                                                                       #！！！破案了：tw.s代表watchable,tw.s.mu是另一把锁
277.                                                                       #！！！tw.s.store才表示kvstore，tw.s.store.mu才是数据库锁
278.                                                                       #！！！kvstore是一个watchablestore对象，就表示会有watch
279.                                                                       #！！！也就是说这边在end的时候那边可能会有一个watch在不断处理
280.                                                                       #！！！（不知道compaction会不会也需要访问这把锁）
281.                                                                       #！！！所以就需要临时锁一下kvstore，注意，这是watchablestore.mu
282.                                                                       #！！！etcdserver是一条一条apply日志的
283.                                                                       #！！！也就是说apply是串行执行的
284.                                                                       #！！！执行一次写事务就执行一次提交操作
285.                                                                       #！！！但是并不是每次提交都会真的提交
286.                                                                       #！！！而是每次提交都会把事务放到batchTx buf
287.                                                                       #！！！当达到一定条数后才会执行commit操作，
288.                                                                       #！！！由BatchLimit参数指定
289.                             tw.s.notify(rev, evs)                         #通知watch相关
290.                             metrics_txn.metricsTxnWrite.End               #更新一些metrics后继续调用内部事务的end
291.                           kvstore_txn.storeTxnWrite.End
292.                             tw.s.revMu.Lock()  
293.                             tw.s.currentRev++                         #更新server.currentRev，
294.                                                                       #因为一次writeTxn代表一次修改操作                        
295.                             batch_tx.batchTxBuffered.Unlock()         #解锁batchtx
296.                                                                       #注意:batchTxBufered加锁是直接调用内部batchTx.lock
297.                                                                       #但是解锁的时候不能只调用batchTx.Unlock
298.                                                                       #因为batchTxBufered还需要做其他事，所以得用自己的Unlock函数
299.                                                                       #batchTxBufered.unlock做三件事:1:把数据从writebuf写入readBuf
300.                                                                       #2:如果未提交事务数达到batchLimit就提交事务
301.                                                                       #3：做完上面两条后才能释放batchTx.lock
302.                                                                       #所以加锁用batchTx.lock即可
303.                                                                       #但是解锁则batchTxBufered需实现自己的Unlock函数
304.                               if t.pending != 0:
305.                                 t.backend.readTx.Lock()               #batchTxn的所有读请求都会对这个readTx加读锁，
306.                                                                       #会等到所有读请求完成并释放锁后才对buf加写锁  
307.                                 txWriteBuffer.writeback               #把writebuf中的内容写到readbuff
308.                                 t.backend.readTx.Unlock()             #释放buf锁
309.                                 if t.pending >= t.backend.batchLimit ||t.pendingDeleteOperations > 0   
310.                                                                       #batchTxn表示批量事务，如果pending的事务数超过了阈值
311.                                                                       #那么就提交,默认是1w或者挂起的delete操作数大于0也会提交
312.                                 {
313.                                   t.commit(false)                            #提交事务
314.                                     t.backend.readTx.Lock()                  #要提交事务必须先阻止所有读
315.                                     batch_tx.batchTxBuffered.unsafeCommit    #提交事务
316.                                       backend.hooks.OnPreCommitUnsafe(t)     #执行hook
317.                                       if t.backend.readTx.tx != nil:         #tx是bbolt.tx  
318.                                                                              #如果bbolt.readTx不为null
319.                                                                              #则需要等待这些读事务完成才能commit  
320.                                                                
321.                                         go func(tx *bolt.Tx, wg *sync.WaitGroup):
322.                                           wg.Wait()
323.                                           bbolt.tx.Rollback()
324.                                           t.backend.readTx.reset()
325.                                       batch_tx.batchTx.commit
326.                                                 bbolt.tx.Commit()                    #bbolt的事务提交sdk
327.                                         t.pending = 0
328.                                         t.pendingDeleteOperations = 0
329.                                     t.backend.readTx.Unlock()                #提交完毕，释放锁
330.                                 }
331.                                 t.batchTx.Unlock()                           #解锁batchtx
333.                             tw.s.revMu.Unlock()
334.                             tw.s.mu.RUnlock()                         #？？？释放读锁。我TM真找不到在哪加的读锁
335.                                                                       #下班 20241202 18:35
336.                                                                       #我靠，根本不是同一把锁,watchableStore有一把mu锁
337.                                                                       #watchablestore.store也有一把叫做mu的锁
338.                                                                       #这里是释放store上的读锁，
339.                                                                       #我们在创建txnWrite的时候会加读锁,在End里释放读锁
340.                                                                       #！！！store.mu是一把读写锁，因为采用的是mvcc。
341.                                                                       #！！！任何put/txn/delete等操作对于store来说
342.                                                                       #！！！都是一个读操作即只会追加数据不会修改旧数据
343.                                                                       #！！！所以此时加读锁就行了
344.                                                                       #！！！但是compact会压缩数据也就是修改旧数据
345.                                                                       #！！！所以此时就必须对store.mu加读锁
346.                                                                       #！！！捋一下：store.mu是保护compact和mvcc操作
347.                                                                       #！！！而watchablestore.mu则是保护watch与notify 操作
348.                                                                       #！！！所以txnwrite的时候只需要对store加读锁就行了
349.                                                                       #！！！不影响watch线程
350.                                                         
352.                               tw.s.mu.Unlock()                            #这里是watchablestore.mu.unlock
353.                                                                       #此后watch goroute就可以继续访问kvstore了
354.                       txnResp.Header.Revision = rev                   #返回给客户端的revision

复制代码

干系文章链接：
STM客户端api干系文章