diff --git a/db/state/aggregator.go b/db/state/aggregator.go index 03db19fef9e..368f6bed53d 100644 --- a/db/state/aggregator.go +++ b/db/state/aggregator.go @@ -158,7 +158,7 @@ func newAggregator(ctx context.Context, dirs datadir.Dirs, reorgBlockDepth uint6 leakDetector: dbg.NewLeakDetector("agg", dbg.SlowTx()), backgroundProgress: background.NewProgressSet(), logger: logger, - workers: workersCfg{merge: 1, collateAndBuild: 1}, + workers: workersCfg{merge: dbg.MergeWorkers, collateAndBuild: dbg.CollateWorkers}, produce: true, } @@ -242,6 +242,7 @@ func (a *Aggregator) RegisterDomain(cfg statecfg.DomainCfg, salt *uint32, dirs d } a.d[cfg.Name].salt.Store(salt) a.AddDependencyBtwnHistoryII(cfg.Name) + a.applyEnvCompressWorkers() return nil } @@ -259,6 +260,7 @@ func (a *Aggregator) RegisterII(cfg statecfg.InvIdxCfg, salt *uint32, dirs datad } a.iis[a.iisCount] = ii a.iisCount++ + a.applyEnvCompressWorkers() return nil } @@ -787,6 +789,16 @@ func (a *Aggregator) setBuildAccessorsWorkers(i int) { }) } +// applyEnvCompressWorkers covers the window before the first preset lands: a restart with +// files on disk merges with whatever the compressor config holds at registration. +func (a *Aggregator) applyEnvCompressWorkers() { + if dbg.CompressWorkers <= 0 { + return + } + a.setCompressWorkers(dbg.CompressWorkers) + a.setBuildAccessorsWorkers(dbg.CompressWorkers) +} + func (a *Aggregator) setCompressWorkers(i int) { a.workers.trySet(func() { for _, d := range a.d { diff --git a/db/state/aggregator2.go b/db/state/aggregator2.go index 37849ba8d0e..43ffffca76d 100644 --- a/db/state/aggregator2.go +++ b/db/state/aggregator2.go @@ -144,9 +144,10 @@ func (opts AggOpts) WithErigonDBSettings(s *ErigonDBSettings) AggOpts { //nolint type workersCfg struct { mu sync.Mutex - editLocks int // >0 while background build/merge pins config; Preset* writes are no-ops + editLocks int // >0 while background build/merge pins config merge int // usually 1 collateAndBuild int + pending []func() // requests that arrived while pinned } func (w *workersCfg) getMerge() int { @@ -156,11 +157,7 @@ func (w *workersCfg) getMerge() int { } func (w *workersCfg) setMerge(n int) { - w.mu.Lock() - defer w.mu.Unlock() - if w.editLocks == 0 { - w.merge = n - } + w.trySet(func() { w.merge = n }) } func (w *workersCfg) getCollateAndBuild() int { @@ -170,20 +167,19 @@ func (w *workersCfg) getCollateAndBuild() int { } func (w *workersCfg) setCollateAndBuild(n int) { - w.mu.Lock() - defer w.mu.Unlock() - if w.editLocks == 0 { - w.collateAndBuild = n - } + w.trySet(func() { w.collateAndBuild = n }) } -// trySet runs fn under mu only while editing is unlocked (no background op holds it). +// trySet runs fn under mu, or queues it for the last unlockEditing while a background op pins +// the config: dropping the request loses it for the process — a restart merges before any preset. func (w *workersCfg) trySet(fn func()) { w.mu.Lock() defer w.mu.Unlock() - if w.editLocks == 0 { - fn() + if w.editLocks > 0 { + w.pending = append(w.pending, fn) + return } + fn() } // lockEditing is reentrant: overlapping build/merge ops each hold a lock, and @@ -200,6 +196,13 @@ func (w *workersCfg) unlockEditing() { if w.editLocks > 0 { w.editLocks-- } + if w.editLocks != 0 { + return + } + for _, fn := range w.pending { + fn() + } + w.pending = nil } func CheckSnapshotsCompatibility(d datadir.Dirs) error { diff --git a/db/state/aggregator2_test.go b/db/state/aggregator2_test.go index 332363c21bd..5fbcf1e12c5 100644 --- a/db/state/aggregator2_test.go +++ b/db/state/aggregator2_test.go @@ -49,3 +49,57 @@ func TestWorkersCfgEditingLockIsReentrant(t *testing.T) { w.unlockEditing() require.True(t, editable(), "extra unlock must not underflow and disable editing") } + +func TestWorkersCfgAppliesRequestDeferredWhilePinned(t *testing.T) { + t.Parallel() + w := &workersCfg{merge: 1, collateAndBuild: 1} + compress := 0 + + w.lockEditing() + w.setMerge(2) + w.setCollateAndBuild(4) + w.trySet(func() { compress = 8 }) + require.Equal(t, 1, w.getMerge(), "must not change under a running build or merge") + require.Equal(t, 1, w.getCollateAndBuild()) + require.Zero(t, compress) + + w.unlockEditing() + require.Equal(t, 2, w.getMerge(), "held request applies on release") + require.Equal(t, 4, w.getCollateAndBuild()) + require.Equal(t, 8, compress, "every queued request runs, not just the last") +} + +func TestWorkersCfgHoldsRequestUntilLastPinReleases(t *testing.T) { + t.Parallel() + w := &workersCfg{merge: 1, collateAndBuild: 1} + + w.lockEditing() + w.lockEditing() + w.setMerge(2) + w.setCollateAndBuild(4) + + w.unlockEditing() + require.Equal(t, 1, w.getCollateAndBuild(), "one of two overlapping pins released is still pinned") + + w.unlockEditing() + require.Equal(t, 4, w.getCollateAndBuild()) +} + +func TestWorkersCfgKeepsOnlyTheNewestRequest(t *testing.T) { + t.Parallel() + w := &workersCfg{merge: 1, collateAndBuild: 1} + + w.lockEditing() + w.setMerge(2) + w.setCollateAndBuild(4) + w.setMerge(3) + w.setCollateAndBuild(6) + w.unlockEditing() + require.Equal(t, 3, w.getMerge()) + require.Equal(t, 6, w.getCollateAndBuild()) + + // A pin/release cycle with no request in between must not resurrect a stale value. + w.lockEditing() + w.unlockEditing() + require.Equal(t, 6, w.getCollateAndBuild()) +}