Sitelet https://github.com/Fantom-foundation/go-opera/commit/e5491ba68057f431a16320dfbf2f701940467cb7
Skip to content

Commit e5491ba

Browse files
uprendisuprendis
committed
de-synchronize flushes between nodes
1 parent d932b9a commit e5491ba

4 files changed

Lines changed: 37 additions & 8 deletions

File tree

‎gossip/config.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -269,7 +269,7 @@ func DefaultStoreConfig(scale cachescale.Func) StoreConfig {
269269
LlrEpochVotesIndexes: scale.I(5),
270270
},
271271
EVM: evmstore.DefaultStoreConfig(scale),
272-
MaxNonFlushedSize: 20*opt.MiB + scale.I(2*opt.MiB),
272+
MaxNonFlushedSize: 21*opt.MiB + scale.I(2*opt.MiB),
273273
MaxNonFlushedPeriod: 30 * time.Minute,
274274
}
275275
}

‎gossip/service.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -310,13 +310,13 @@ func (s *Service) makePeriodicFlusher() PeriodicFlusher {
310310
commitNeeded: func() bool {
311311
// use slightly higher size threshold to avoid locking the mutex/wg pair and hurting events/blocks concurrency
312312
// PeriodicFlusher should mostly commit only data generated by async EVM snapshots generation
313-
return s.store.isCommitNeeded(120, 100)
313+
return s.store.isCommitNeeded(1200, 1000)
314314
},
315315
commit: func() {
316316
s.engineMu.Lock()
317317
defer s.engineMu.Unlock()
318318
// Note: blockProcWg.Wait() is already called by s.commit
319-
if s.store.isCommitNeeded(120, 100) {
319+
if s.store.isCommitNeeded(1200, 1000) {
320320
s.commit(false)
321321
}
322322
},

‎gossip/store.go‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ import (
1818
"github.com/Fantom-foundation/go-opera/logger"
1919
"github.com/Fantom-foundation/go-opera/utils/adapters/snap2kvdb"
2020
"github.com/Fantom-foundation/go-opera/utils/eventid"
21+
"github.com/Fantom-foundation/go-opera/utils/randat"
2122
"github.com/Fantom-foundation/go-opera/utils/rlpstore"
2223
"github.com/Fantom-foundation/go-opera/utils/switchable"
2324
)
@@ -165,14 +166,16 @@ func (s *Store) Close() {
165166
}
166167

167168
func (s *Store) IsCommitNeeded() bool {
168-
return s.isCommitNeeded(100, 100)
169+
// randomize flushing criteria for each epoch so that nodes would desynchronize flushes
170+
ratio := 900 + randat.RandAt(uint64(s.GetEpoch()))%100
171+
return s.isCommitNeeded(ratio, ratio)
169172
}
170173

171-
func (s *Store) isCommitNeeded(sc, tc int) bool {
172-
period := s.cfg.MaxNonFlushedPeriod * time.Duration(sc) / 100
173-
size := (s.cfg.MaxNonFlushedSize / 2) * tc / 100
174+
func (s *Store) isCommitNeeded(sc, tc uint64) bool {
175+
period := s.cfg.MaxNonFlushedPeriod * time.Duration(sc) / 1000
176+
size := (uint64(s.cfg.MaxNonFlushedSize) / 2) * tc / 1000
174177
return time.Since(s.prevFlushTime) > period ||
175-
s.dbs.NotFlushedSizeEst() > size
178+
uint64(s.dbs.NotFlushedSizeEst()) > size
176179
}
177180

178181
// commitEVM commits EVM storage

‎utils/randat/rand_at.go‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
package randat
2+
3+
import (
4+
"math/rand"
5+
)
6+
7+
type cached struct {
8+
seed uint64
9+
r uint64
10+
}
11+
12+
var (
13+
gSeed = rand.Int63()
14+
cache = cached{}
15+
)
16+
17+
// RandAt returns random number with seed
18+
// Not safe for concurrent use
19+
func RandAt(seed uint64) uint64 {
20+
if seed != 0 && cache.seed == seed {
21+
return cache.r
22+
}
23+
cache.seed = seed
24+
cache.r = rand.New(rand.NewSource(gSeed ^ int64(seed))).Uint64()
25+
return cache.r
26+
}

0 commit comments

Comments
 (0)