《MMO 服务器开发》8.2 并发修改与事务

// file: internal/inventory/bag.go
// tier: L1
// verify: go test -race ./internal/inventory/
package inventory

import (
	"sync"
	"sync/atomic"
)

// clone 深拷贝状态。槽位指针必须复制成新对象，否则副本上的改动会顺着
// 指针改到原状态，回滚就失效了。
func (s *bagState) clone() *bagState {
	ns := &bagState{
		slots:    make([]*ItemStack, len(s.slots)),
		equip:    make(map[EquipSlot]*ItemStack, len(s.equip)),
		index:    make(map[ItemInstID]int, len(s.index)),
		nextInst: s.nextInst,
	}
	for i, st := range s.slots {
		if st != nil {
			cp := *st
			ns.slots[i] = &cp
		}
	}
	for slot, st := range s.equip {
		cp := *st
		ns.equip[slot] = &cp
	}
	for k, v := range s.index {
		ns.index[k] = v
	}
	return ns
}

// Reader 是背包的只读视图。所有方法只读，不得保存底层状态引用。
type Reader interface {
	Count(id ItemID) int
	Snapshot() []ItemStack
}

// Bag 是并发安全的背包：写操作走 Do（事务），读操作走 View（无锁）。
//
// 状态是一个 *bagState，放在 atomic.Pointer 后面。写事务在副本上完成后再
// 一次性换指针，所以读路径完全不需要加锁，也不会读到「改了一半」的中间态。
type Bag struct {
	cat   *Catalog
	mu    sync.Mutex               // 串行化写事务；读路径不经过它
	state atomic.Pointer[bagState] // 当前已提交状态
	seen  map[uint64]struct{}      // 已应用的幂等操作号
}

// NewBag 建一个容量为 slots 的背包。
func NewBag(cat *Catalog, slots int) *Bag {
	b := &Bag{cat: cat, seen: make(map[uint64]struct{})}
	b.state.Store(newBagState(slots))
	return b
}

// Tx 是一次事务里的可写视图。它只改私有副本，提交时才换进 Bag。
type Tx struct {
	cat *Catalog
	st  *bagState
}

// Add 向背包加入 count 个 id。
func (t *Tx) Add(id ItemID, count int) error { return t.st.add(t.cat, id, count) }

// Remove 从背包扣减 count 个 id。
func (t *Tx) Remove(id ItemID, count int) error { return t.st.remove(t.cat, id, count) }

// Count 读当前事务副本里的数量，能看到同一事务内前面几步的改动。
func (t *Tx) Count(id ItemID) int { return t.st.Count(id) }

// Do 在一个事务里执行 fn：先复制当前状态，在副本上跑 fn；
// fn 返回 nil 才把副本换进 Bag，返回错误则整份丢弃——这就是天然的回滚。
//
// opID 非 0 时启用幂等：同一个 opID 只会真正生效一次，重复调用直接返回 nil。
// 幂等号通常取「客户端请求的 seq」或上游事件 ID，用来挡住重发与重放。
func (b *Bag) Do(opID uint64, fn func(tx *Tx) error) error {
	b.mu.Lock()
	defer b.mu.Unlock()
	if opID != 0 {
		if _, ok := b.seen[opID]; ok {
			return nil // 幂等：这次请求之前已生效
		}
	}
	next := b.state.Load().clone()
	tx := &Tx{cat: b.cat, st: next}
	if err := fn(tx); err != nil {
		return err // 副本丢弃，已提交状态纹丝不动
	}
	b.state.Store(next)
	if opID != 0 {
		b.seen[opID] = struct{}{}
	}
	return nil
}

// View 在无锁前提下把当前状态交给只读函数。fn 里只能读，
// 且不得把参数存起来——它指向的是某个已提交状态的快照。
func (b *Bag) View(fn func(v Reader)) { fn(b.state.Load()) }

// Count 是 View 的快捷方式，读单个物品数量。
func (b *Bag) Count(id ItemID) int {
	var n int
	b.View(func(v Reader) { n = v.Count(id) })
	return n
}

// Snapshot 是 View 的快捷方式，导出背包槽位。
func (b *Bag) Snapshot() []ItemStack { return b.state.Load().Snapshot() }

// file: internal/inventory/bag_test.go
// tier: L1
// verify: go test -race ./internal/inventory/
package inventory

import (
	"errors"
	"sync"
	"testing"
)

func TestTransactionRollback(t *testing.T) {
	b := NewBag(testCatalog(), 4)
	_ = b.Do(1, func(tx *Tx) error { return tx.Add(1001, 50) })
	// 事务中途失败：前面加的东西必须一并撤销。
	boom := errors.New("boom")
	err := b.Do(2, func(tx *Tx) error {
		if err := tx.Add(1001, 50); err != nil {
			return err
		}
		return boom
	})
	if !errors.Is(err, boom) {
		t.Fatalf("err=%v，期望 boom", err)
	}
	if got := b.Count(1001); got != 50 {
		t.Fatalf("回滚后 count=%d，期望 50", got)
	}
}

func TestIdempotentByOpID(t *testing.T) {
	b := NewBag(testCatalog(), 4)
	const op = 0xABCD
	for i := 0; i < 3; i++ {
		if err := b.Do(op, func(tx *Tx) error { return tx.Add(1001, 10) }); err != nil {
			t.Fatalf("do: %v", err)
		}
	}
	// 同一 opID 三次只应生效一次。
	if got := b.Count(1001); got != 10 {
		t.Fatalf("count=%d，期望 10（幂等）", got)
	}
}

func TestConcurrentAddNoLostUpdate(t *testing.T) {
	b := NewBag(testCatalog(), 64)
	const (
		workers = 32
		perW    = 50
	)
	var wg sync.WaitGroup
	for w := 0; w < workers; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for i := 0; i < perW; i++ {
				if err := b.Do(0, func(tx *Tx) error { return tx.Add(2001, 1) }); err != nil {
					t.Errorf("add: %v", err)
					return
				}
			}
		}()
	}
	wg.Wait()
	if got, want := b.Count(2001), workers*perW; got != want {
		t.Fatalf("count=%d，期望 %d（有丢更新）", got, want)
	}
}

func TestViewReadsCommittedState(t *testing.T) {
	b := NewBag(testCatalog(), 4)
	_ = b.Do(0, func(tx *Tx) error { return tx.Add(1001, 7) })
	// 读路径不经锁，读到的必须是某个已提交状态。
	var seen int
	b.View(func(v Reader) { seen = v.Count(1001) })
	if seen != 7 {
		t.Fatalf("View count=%d，期望 7", seen)
	}
}