【ETCD】【源码阅读】深入分析 storeTxnWrite.Put方法源码

该方法是 storeTxnWrite 类型中的核心方法，负责将键值对存储到数据库，同时处理键的元数据（如版本、修订号、租约）并管理租约关联。

- - 一、完整代码
  - 二、方法详解
  - 方法签名
  - 1. 计算修订号并初始化变量
  - 2. 检查键是否已存在
  - 3. 生成索引修订号与字节表示
  - 4. 更新版本号并构建新的键值对
  - 5. 序列化键值对并存储
  - 6. 处理旧租约
  - 7. 绑定新租约
  - 三、总结
  - - 方法的主要功能：
    - 核心步骤：

一、完整代码

func (tw *storeTxnWrite) put(key, value []byte, leaseID lease.LeaseID) {
	rev := tw.beginRev + 1
	c := rev
	oldLease := lease.NoLease

	// if the key exists before, use its previous created and
	// get its previous leaseID
	_, created, ver, err := tw.s.kvindex.Get(key, rev)
	if err == nil {
		c = created.main
		oldLease = tw.s.le.GetLease(lease.LeaseItem{Key: string(key)})
		tw.trace.Step("get key's previous created_revision and leaseID")
	}
	ibytes := newRevBytes()
	idxRev := revision{main: rev, sub: int64(len(tw.changes))}
	revToBytes(idxRev, ibytes)

	ver = ver + 1
	kv := mvccpb.KeyValue{
		Key:            key,
		Value:          value,
		CreateRevision: c,
		ModRevision:    rev,
		Version:        ver,
		Lease:          int64(leaseID),
	}

	d, err := kv.Marshal()
	if err != nil {
		tw.storeTxnRead.s.lg.Fatal(
			"failed to marshal mvccpb.KeyValue",
			zap.Error(err),
		)
	}

	tw.trace.Step("marshal mvccpb.KeyValue")
	tw.tx.UnsafeSeqPut(buckets.Key, ibytes, d)
	tw.s.kvindex.Put(key, idxRev)
	tw.changes = append(tw.changes, kv)
	tw.trace.Step("store kv pair into bolt db")

	if oldLease != lease.NoLease {
		if tw.s.le == nil {
			panic("no lessor to detach lease")
		}
		err = tw.s.le.Detach(oldLease, []lease.LeaseItem{{Key: string(key)}})
		if err != nil {
			tw.storeTxnRead.s.lg.Error(
				"failed to detach old lease from a key",
				zap.Error(err),
			)
		}
	}
	if leaseID != lease.NoLease {
		if tw.s.le == nil {
			panic("no lessor to attach lease")
		}
		err = tw.s.le.Attach(leaseID, []lease.LeaseItem{{Key: string(key)}})
		if err != nil {
			panic("unexpected error from lease Attach")
		}
	}
	tw.trace.Step("attach lease to kv pair")
}

二、方法详解

方法签名

func (tw *storeTxnWrite) put(key, value []byte, leaseID lease.LeaseID)

key, value []byte：键值对的字节切片。
leaseID lease.LeaseID：与键关联的租约 ID。如果为 lease.NoLease，表示没有租约。

1. 计算修订号并初始化变量

rev := tw.beginRev + 1
c := rev
oldLease := lease.NoLease

rev：新键值对的修订号，比事务起始修订号 beginRev 大 1。
c：表示键的 CreateRevision，初始值为当前修订号。
oldLease：存储键之前的租约 ID，初始值设为 lease.NoLease（无租约）。

2. 检查键是否已存在

_, created, ver, err := tw.s.kvindex.Get(key, rev)
if err == nil {
    c = created.main
    oldLease = tw.s.le.GetLease(lease.LeaseItem{Key: string(key)})
    tw.trace.Step("get key's previous created_revision and leaseID")
}

tw.s.kvindex.Get：
- 在 kvindex 中查找当前键是否存在（按当前修订号 rev）。
- 如果键存在，获取：
  - created.main：键的初始创建修订号。
  - ver：键的当前版本号。
c = created.main：如果键已存在，将其初始创建修订号 created.main 赋值给 c。
oldLease = tw.s.le.GetLease：查找该键关联的旧租约（如果有）。
tw.trace.Step：记录 “获取键的创建修订号和租约 ID” 这一步操作。

3. 生成索引修订号与字节表示

ibytes := newRevBytes()
idxRev := revision{main: rev, sub: int64(len(tw.changes))}
revToBytes(idxRev, ibytes)

idxRev：为键生成索引修订号：
- main：当前修订号 rev。
- sub：当前事务中已变更的键值对数量（len(tw.changes)）。
revToBytes：将修订号 idxRev 转换成字节数组 ibytes，用于存储到底层数据库。

4. 更新版本号并构建新的键值对

ver = ver + 1
kv := mvccpb.KeyValue{
    Key:            key,
    Value:          value,
    CreateRevision: c,
    ModRevision:    rev,
    Version:        ver,
    Lease:          int64(leaseID),
}

ver = ver + 1：增加键的版本号。
构建新的 mvccpb.KeyValue 对象：
- Key：键。
- Value：值。
- CreateRevision：键的创建修订号 c。
- ModRevision：键的最新修改修订号 rev。
- Version：键的版本号。
- Lease：关联的租约 ID。

5. 序列化键值对并存储

d, err := kv.Marshal()
if err != nil {
    tw.storeTxnRead.s.lg.Fatal(
        "failed to marshal mvccpb.KeyValue",
        zap.Error(err),
    )
}

tw.trace.Step("marshal mvccpb.KeyValue")
tw.tx.UnsafeSeqPut(buckets.Key, ibytes, d)
tw.s.kvindex.Put(key, idxRev)
tw.changes = append(tw.changes, kv)
tw.trace.Step("store kv pair into bolt db")

序列化：将 kv 通过 Marshal 序列化为字节数组 d。
存储：
- 调用 UnsafeSeqPut 将键值对存储到 Bolt DB 中。
- 使用 kvindex.Put 更新键的索引，记录索引修订号 idxRev。
- 将 kv 添加到事务变更列表 tw.changes 中。
tw.trace.Step：记录 “序列化键值对” 和 “存储到 Bolt DB” 的步骤。

6. 处理旧租约

if oldLease != lease.NoLease {
    if tw.s.le == nil {
        panic("no lessor to detach lease")
    }
    err = tw.s.le.Detach(oldLease, []lease.LeaseItem{{Key: string(key)}})
    if err != nil {
        tw.storeTxnRead.s.lg.Error(
            "failed to detach old lease from a key",
            zap.Error(err),
        )
    }
}

如果键之前绑定了租约 oldLease，调用 Detach 方法解除该键与旧租约的关联。
如果 Detach 失败，记录错误日志。

7. 绑定新租约

if leaseID != lease.NoLease {
    if tw.s.le == nil {
        panic("no lessor to attach lease")
    }
    err = tw.s.le.Attach(leaseID, []lease.LeaseItem{{Key: string(key)}})
    if err != nil {
        panic("unexpected error from lease Attach")
    }
}
tw.trace.Step("attach lease to kv pair")