diff --git a/db/state/temporal_mem_batch.go b/db/state/temporal_mem_batch.go index 5f642e846b8..79136f761b2 100644 --- a/db/state/temporal_mem_batch.go +++ b/db/state/temporal_mem_batch.go @@ -64,8 +64,13 @@ type TemporalMemBatch struct { inMemHistoryReads bool latestStateLock sync.RWMutex - domains [kv.DomainLen]map[string][]dataWithTxNum - storage *btree2.Map[string, []dataWithTxNum] // TODO: replace hardcoded domain name to per-config configuration of available Guarantees/AccessMethods (range vs get) + // commitmentLock guards domains[kv.CommitmentDomain] only. Commitment branches + // are written by the calculator goroutine during the fold; keeping them off + // latestStateLock lets worker state reads run without contending those writes. + // Multi-domain ops (Flush/Unwind) take both, always latestStateLock first. + commitmentLock sync.RWMutex + domains [kv.DomainLen]map[string][]dataWithTxNum + storage *btree2.Map[string, []dataWithTxNum] // TODO: replace hardcoded domain name to per-config configuration of available Guarantees/AccessMethods (range vs get) domainWriters [kv.DomainLen]*DomainBufferedWriter iiWriters []*InvertedIndexBufferedWriter @@ -149,11 +154,26 @@ func (sd *TemporalMemBatch) putHistory(domain kv.Domain, k, v []byte, txNum uint return sd.domainWriters[domain].PutWithPrev(k, v, txNum, preval) } +// lockFor returns the latest-state lock guarding a single domain: CommitmentDomain +// has its own lock, everything else shares latestStateLock. +func (sd *TemporalMemBatch) lockFor(domain kv.Domain) *sync.RWMutex { + if domain == kv.CommitmentDomain { + return &sd.commitmentLock + } + return &sd.latestStateLock +} + +// lockBoth/unlockBoth guard operations spanning all domains (Flush, Unwind). +// Order is fixed — state before commitment — to avoid deadlock. +func (sd *TemporalMemBatch) lockBoth() { sd.latestStateLock.Lock(); sd.commitmentLock.Lock() } +func (sd *TemporalMemBatch) unlockBoth() { sd.commitmentLock.Unlock(); sd.latestStateLock.Unlock() } + // putLatest reports whether this write is a same-txNum update of the key, // replacing the key's last entry in place instead of appending a version. func (sd *TemporalMemBatch) putLatest(domain kv.Domain, key string, val []byte, txNum uint64) (sameTxNumUpdate bool) { - sd.latestStateLock.Lock() - defer sd.latestStateLock.Unlock() + l := sd.lockFor(domain) + l.Lock() + defer l.Unlock() var updateMetrics = func(domain kv.Domain, putKeySize int, putValueSize int) { sd.metrics.Lock() @@ -234,14 +254,15 @@ func (sd *TemporalMemBatch) putLatest(domain kv.Domain, key string, val []byte, } func (sd *TemporalMemBatch) GetLatest(domain kv.Domain, key []byte) (v []byte, step kv.Step, ok bool) { - sd.latestStateLock.RLock() - defer sd.latestStateLock.RUnlock() + l := sd.lockFor(domain) + l.RLock() + defer l.RUnlock() return sd.getLatest(domain, key) } // getLatest is the lock-free implementation of GetLatest. -// The caller must already hold latestStateLock (either RLock or Lock), -// e.g. from within an IteratePrefix callback. +// The caller must already hold the domain's lock (see lockFor), either RLock or +// Lock, e.g. from within an IteratePrefix callback. func (sd *TemporalMemBatch) getLatest(domain kv.Domain, key []byte) (v []byte, step kv.Step, ok bool) { var unwoundLatest = func(domain kv.Domain, key string) (v []byte, step kv.Step, ok bool) { if sd.unwindChangeset != nil { @@ -290,8 +311,9 @@ func (sd *TemporalMemBatch) GetAsOf(domain kv.Domain, key []byte, ts uint64) (v if !sd.inMemHistoryReads && domain != kv.ReceiptDomain { return nil, false, errors.New("GetAsOf called on TemporalMemBatch with inMemHistoryReads disabled") } - sd.latestStateLock.RLock() - defer sd.latestStateLock.RUnlock() + l := sd.lockFor(domain) + l.RLock() + defer l.RUnlock() // unwoundLatest returns the pre-unwound-block value for a key that was // modified by the unwound block. Only fires when ts is at-or-after the @@ -363,8 +385,9 @@ func (sd *TemporalMemBatch) SizeEstimate() uint64 { } func (sd *TemporalMemBatch) IteratePrefix(domain kv.Domain, prefix []byte, roTx kv.Tx, it func(k []byte, v []byte) (cont bool, err error)) error { - sd.latestStateLock.RLock() - defer sd.latestStateLock.RUnlock() + l := sd.lockFor(domain) + l.RLock() + defer l.RUnlock() var ramIter btree2.MapIter[string, []dataWithTxNum] if domain == kv.StorageDomain { ramIter = sd.storage.Iter() @@ -420,8 +443,9 @@ func (sd *TemporalMemBatch) HasPrefix(domain kv.Domain, prefix []byte, roTx kv.T // for the given domain whose key starts with prefix. It never touches disk or // segment files — only the in-memory btree (StorageDomain) or the domain map. func (sd *TemporalMemBatch) HasPrefixInRAM(domain kv.Domain, prefix []byte) bool { - sd.latestStateLock.RLock() - defer sd.latestStateLock.RUnlock() + l := sd.lockFor(domain) + l.RLock() + defer l.RUnlock() if domain == kv.StorageDomain { prefixStr := common.ToStringZeroCopy(prefix) @@ -526,8 +550,8 @@ func (sd *TemporalMemBatch) GetDiffset(tx kv.RwTx, blockHash common.Hash, blockN // Unwind drops [unwindToTxNum, ∞) func (sd *TemporalMemBatch) Unwind(unwindToTxNum uint64, changeset *[kv.DomainLen][]kv.DomainEntryDiff) { - sd.latestStateLock.Lock() - defer sd.latestStateLock.Unlock() + sd.lockBoth() + defer sd.unlockBoth() sd.unwindToTxNum = unwindToTxNum @@ -763,11 +787,11 @@ func (sd *TemporalMemBatch) flushLocked(ctx context.Context, tx kv.RwTx) error { // Flush writes the mem-batch to tx. With kv.WithFlushCallback options, the // registered per-domain callback is invoked for every (key, value, step, txNum) // tuple after the MDBX write succeeds, so a downstream cache can never be left -// ahead of MDBX. Runs under latestStateLock so the callback's snapshot matches -// flush-time state. +// ahead of MDBX. Runs under both latest-state locks so the callback's snapshot +// matches flush-time state. func (sd *TemporalMemBatch) Flush(ctx context.Context, tx kv.RwTx, opts ...kv.FlushOption) error { - sd.latestStateLock.Lock() - defer sd.latestStateLock.Unlock() + sd.lockBoth() + defer sd.unlockBoth() if err := sd.flushLocked(ctx, tx); err != nil { return err @@ -809,8 +833,8 @@ func (sd *TemporalMemBatch) Flush(ctx context.Context, tx kv.RwTx, opts ...kv.Fl // FlushWithCommitmentCallback flushes the batch then invokes cb per // commitment-domain tuple under the lock. func (sd *TemporalMemBatch) FlushWithCommitmentCallback(ctx context.Context, tx kv.RwTx, cb execctx.CommitmentFlushCallback) error { - sd.latestStateLock.Lock() - defer sd.latestStateLock.Unlock() + sd.lockBoth() + defer sd.unlockBoth() if err := sd.flushLocked(ctx, tx); err != nil { return err diff --git a/db/state/temporal_mem_batch_test.go b/db/state/temporal_mem_batch_test.go index 9aa8689897b..bfdcb64a7d2 100644 --- a/db/state/temporal_mem_batch_test.go +++ b/db/state/temporal_mem_batch_test.go @@ -17,13 +17,17 @@ package state import ( + "sync" "testing" + btree2 "github.com/tidwall/btree" + "github.com/stretchr/testify/require" "github.com/erigontech/erigon/common" "github.com/erigontech/erigon/db/kv" "github.com/erigontech/erigon/db/state/changeset" + "github.com/erigontech/erigon/db/state/kvmetrics" ) // A reorg unwind restores domain values from the diffset, so a domain missing @@ -45,3 +49,56 @@ func TestGetDiffsetCoversAllDomains(t *testing.T) { require.NotEmpty(t, diffs[d], "domain %s missing from GetDiffset", d) } } + +// CommitmentDomain is on its own lock, so the calculator can write commitment +// branches while workers read the state domains without contending. Run under +// -race: commitment write vs state read, state read/write, commitment +// read/write, and the both-lock Unwind path (deadlock-freedom). +func TestSplitLock_ConcurrentCommitmentWriteVsStateRead(t *testing.T) { + sd := &TemporalMemBatch{ + stepSize: 1, + storage: btree2.NewMap[string, []dataWithTxNum](128), + metrics: kvmetrics.NewDomainMetrics(), + } + for i := range sd.domains { + sd.domains[i] = map[string][]dataWithTxNum{} + } + const nKeys = 512 + mkKey := func(i int) string { return string([]byte{byte(i), byte(i >> 8)}) } + for i := range nKeys { + sd.putLatest(kv.AccountsDomain, mkKey(i), []byte{1}, uint64(i)) + } + + stop := make(chan struct{}) + var writers, readers sync.WaitGroup + + loopUntilStop := func(f func(i int)) { + defer writers.Done() + for i := 0; ; i++ { + select { + case <-stop: + return + default: + f(i) + } + } + } + writers.Add(3) + go loopUntilStop(func(i int) { sd.putLatest(kv.CommitmentDomain, mkKey(i&(nKeys-1)), []byte{byte(i)}, uint64(i)) }) + go loopUntilStop(func(i int) { sd.putLatest(kv.AccountsDomain, mkKey(i&(nKeys-1)), []byte{byte(i)}, uint64(i)) }) + go loopUntilStop(func(i int) { sd.Unwind(uint64(1_000_000+i), nil) }) // both-lock path; txNum far above data so nothing is pruned + + readers.Add(8) + for range 8 { + go func() { + defer readers.Done() + for i := range 20_000 { + sd.GetLatest(kv.AccountsDomain, []byte(mkKey(i&(nKeys-1)))) + sd.GetLatest(kv.CommitmentDomain, []byte(mkKey(i&(nKeys-1)))) + } + }() + } + readers.Wait() + close(stop) + writers.Wait() +}