Faster vector clock

config/config.go

@@ -7,6 +7,7 @@ import (
  carmen "github.com/Fantom-foundation/Carmen/go/state"
  "github.com/Fantom-foundation/go-opera/config/flags"
  "github.com/Fantom-foundation/go-opera/gossip/evmstore"
+ "github.com/Fantom-foundation/go-opera/vecclock"
  "github.com/Fantom-foundation/go-opera/version"
  "github.com/ethereum/go-ethereum/common/fdlimit"
  "os"
@@ -30,7 +31,6 @@ import (
  "github.com/Fantom-foundation/go-opera/gossip/emitter"
  "github.com/Fantom-foundation/go-opera/integration"
  "github.com/Fantom-foundation/go-opera/utils/memory"
- "github.com/Fantom-foundation/go-opera/vecmt"
 )
 
 const (
@@ -65,7 +65,7 @@ type Config struct {
  OperaStore gossip.StoreConfig
  Lachesis abft.Config
  LachesisStore abft.StoreConfig
- VectorClock vecmt.IndexConfig
+ VectorClock vecclock.Config
  DBs integration.DBsConfig
 }
 
@@ -178,7 +178,7 @@ func gossipConfigWithFlags(ctx *cli.Context, src gossip.Config) gossip.Config {
  return cfg
 }
 
-func setEvmStore(ctx *cli.Context, datadir string, src  evmstore.StoreConfig) (evmstore.StoreConfig, error) {
+func setEvmStore(ctx *cli.Context, datadir string, src evmstore.StoreConfig) (evmstore.StoreConfig, error) {
  cfg := src
  cfg.StateDb.Directory = filepath.Join(datadir, "carmen")
 
@@ -222,7 +222,11 @@ func setDBConfig(cfg Config, cacheRatio cachescale.Func) (Config, error) {
  }
  cfg.DBs.RuntimeCache = integration.DBCacheConfig{
  Cache: cacheRatio.U64(480 * opt.MiB),
- Fdlimit: handles*480/1400 + 1,
+ Fdlimit: handles/2 + 16,
+ }
+ cfg.DBs.TmpDBCache = integration.DBCacheConfig{
+ Cache: cacheRatio.U64(24 * opt.MiB),
+ Fdlimit: handles/1000 + 16,
  }
  return cfg, nil
 }
@@ -231,13 +235,13 @@ const (
  // DefaultCacheSize is calculated as memory consumption in a worst case scenario with default configuration
  // Average memory consumption might be 3-5 times lower than the maximum
  DefaultCacheSize = 6 * 1024 // MB
- ConstantCacheSize = 400 // MB
+ ConstantCacheSize = 400  // MB
 )
 
 func cacheScaler(ctx *cli.Context) cachescale.Func {
  baseSize := DefaultCacheSize
  totalMemory := int(memory.TotalMemory() / opt.MiB)
- maxCache := totalMemory * 3 / 5  // max 60% of available memory
+ maxCache := totalMemory * 3 / 5 // max 60% of available memory
  if maxCache < baseSize {
  maxCache = baseSize
  }
@@ -274,7 +278,7 @@ func MakeAllConfigsFromFile(ctx *cli.Context, configFile string) (*Config, error
  OperaStore: gossip.DefaultStoreConfig(cacheRatio),
  Lachesis: abft.DefaultConfig(),
  LachesisStore: abft.DefaultStoreConfig(cacheRatio),
- VectorClock: vecmt.DefaultConfig(cacheRatio),
+ VectorClock: vecclock.DefaultConfig(cacheRatio),
  }
 
  if ctx.GlobalIsSet(FakeNetFlag.Name) {

config/config_custom_test.go

@@ -2,6 +2,7 @@ package config
 
 import (
  "bytes"
+ "github.com/Fantom-foundation/go-opera/vecclock"
  "testing"
 
  "github.com/Fantom-foundation/lachesis-base/abft"
@@ -12,7 +13,6 @@ import (
  "github.com/Fantom-foundation/go-opera/evmcore"
  "github.com/Fantom-foundation/go-opera/gossip"
  "github.com/Fantom-foundation/go-opera/gossip/emitter"
- "github.com/Fantom-foundation/go-opera/vecmt"
 )
 
 func TestConfigFile(t *testing.T) {
@@ -29,7 +29,7 @@ func TestConfigFile(t *testing.T) {
  OperaStore: gossip.DefaultStoreConfig(cacheRatio),
  Lachesis: abft.DefaultConfig(),
  LachesisStore: abft.DefaultStoreConfig(cacheRatio),
- VectorClock: vecmt.DefaultConfig(cacheRatio),
+ VectorClock: vecclock.DefaultConfig(cacheRatio),
  }
 
  canonical := func(nn []*enode.Node) []*enode.Node {

config/make_node.go

@@ -127,13 +127,9 @@ func MakeNode(ctx *cli.Context, cfg *Config) (*node.Node, *gossip.Service, func(
  if err != nil {
  return nil, nil, nil, fmt.Errorf("failed to create the gossip service: %w", err)
  }
- err = engine.Bootstrap(svc.GetConsensusCallbacks())
+ err = engine.StartFrom(svc.GetConsensusCallbacks(), gdb.GetEpoch(), gdb.GetValidators())
  if err != nil {
- return nil, nil, nil, fmt.Errorf("failed to bootstrap the consensus engine: %w", err)
- }
- err = engine.Reset(gdb.GetEpoch(), gdb.GetValidators())
- if err != nil {
- return nil, nil, nil, fmt.Errorf("failed to reset the consensus engine: %w", err)
+ return nil, nil, nil, fmt.Errorf("failed to start the consensus engine: %w", err)
  }
  svc.ReprocessEpochEvents()
 
@@ -164,7 +160,6 @@ func MakeNetworkStack(ctx *cli.Context, cfg *node.Config) (*node.Node, error) {
  return nil, fmt.Errorf("failed to create the protocol stack: %w", err)
  }
 
-
  keystoreDir, err := cfg.KeyDirConfig()
  if err != nil {
  return nil, fmt.Errorf("failed to setup account config: %w", err)

gossip/c_event_callbacks.go

@@ -2,16 +2,15 @@ package gossip
 
 import (
  "errors"
- "github.com/ethereum/go-ethereum/metrics"
- "math/big"
- "sync/atomic"
-
  "github.com/Fantom-foundation/lachesis-base/gossip/dagprocessor"
  "github.com/Fantom-foundation/lachesis-base/hash"
  "github.com/Fantom-foundation/lachesis-base/inter/dag"
  "github.com/Fantom-foundation/lachesis-base/inter/idx"
  "github.com/ethereum/go-ethereum/common"
  "github.com/ethereum/go-ethereum/log"
+ "github.com/ethereum/go-ethereum/metrics"
+ "math/big"
+ "sync/atomic"
 
  "github.com/Fantom-foundation/go-opera/eventcheck"
  "github.com/Fantom-foundation/go-opera/eventcheck/epochcheck"
@@ -43,11 +42,8 @@ func (s *Service) buildEvent(e *inter.MutableEventPayload, onIndexed func()) err
  }
 
  // indexing event without saving
- defer s.dagIndexer.DropNotFlushed()
- err := s.dagIndexer.Add(e)
- if err != nil {
- return err
- }
+ defer s.dagIndexer.Revert()
+ s.dagIndexer.Add(e)
 
  if onIndexed != nil {
  onIndexed()
@@ -74,10 +70,7 @@ func (s *Service) buildEvent(e *inter.MutableEventPayload, onIndexed func()) err
 
 // processSavedEvent performs processing which depends on event being saved in DB
 func (s *Service) processSavedEvent(e *inter.EventPayload, es *iblockproc.EpochState) error {
- err := s.dagIndexer.Add(e)
- if err != nil {
- return err
- }
+ s.dagIndexer.Add(e)
 
  // check median time
  if e.MedianTime() != s.dagIndexer.MedianTime(e.ID(), es.EpochStart) {
@@ -93,7 +86,7 @@ func (s *Service) saveAndProcessEvent(e *inter.EventPayload, es *iblockproc.Epoc
  fixEventTxHashes(e)
  // indexing event
  s.store.SetEvent(e)
- defer s.dagIndexer.DropNotFlushed()
+ defer s.dagIndexer.Revert()
 
  err := s.processSavedEvent(e, es)
  if err != nil {
@@ -102,7 +95,7 @@ func (s *Service) saveAndProcessEvent(e *inter.EventPayload, es *iblockproc.Epoc
  }
 
  // save event index after success
- s.dagIndexer.Flush()
+ s.dagIndexer.Commit()
  return nil
 }
 
@@ -128,8 +121,7 @@ func (s *Service) switchEpochTo(newEpoch idx.Epoch) {
  s.store.SetHighestLamport(0)
  // reset dag indexer
  s.store.resetEpochStore(newEpoch)
- es := s.store.getEpochStore(newEpoch)
- s.dagIndexer.Reset(s.store.GetValidators(), es.table.DagIndex, func(id hash.Event) dag.Event {
+ s.dagIndexer.Reset(s.store.GetValidators(), func(id hash.Event) dag.Event {
  return s.store.GetEvent(id)
  })
  // notify event checkers about new validation data
@@ -179,12 +171,9 @@ func (s *Service) ReprocessEpochEvents() {
  s.bootstrapping = true
  // reprocess epoch events, as epoch DBs don't survive restart
  s.store.ForEachEpochEvent(s.store.GetEpoch(), func(event *inter.EventPayload) bool {
- err := s.dagIndexer.Add(event)
- if err != nil {
- log.Crit("Failed to reindex epoch event", "event", event.String(), "err", err)
- }
- s.dagIndexer.Flush()
- err = s.engine.Process(event)
+ s.dagIndexer.Add(event)
+ s.dagIndexer.Commit()
+ err := s.engine.Process(event)
  if err != nil {
  log.Crit("Failed to reprocess epoch event", "event", event.String(), "err", err)
  }

gossip/common_test.go

@@ -5,6 +5,10 @@ import (
  "crypto/ecdsa"
  "errors"
  "fmt"
+ "github.com/Fantom-foundation/go-opera/vecclock"
+ "github.com/Fantom-foundation/lachesis-base/kvdb"
+ "github.com/Fantom-foundation/lachesis-base/kvdb/memorydb"
+ "github.com/ethereum/go-ethereum"
  "github.com/ethereum/go-ethereum/core/tracing"
  "math"
  "math/big"
@@ -17,7 +21,6 @@ import (
  "github.com/Fantom-foundation/lachesis-base/inter/dag"
  "github.com/Fantom-foundation/lachesis-base/inter/idx"
  "github.com/Fantom-foundation/lachesis-base/utils/cachescale"
- "github.com/ethereum/go-ethereum"
  "github.com/ethereum/go-ethereum/accounts/abi/bind"
  "github.com/ethereum/go-ethereum/common"
  "github.com/ethereum/go-ethereum/common/hexutil"
@@ -41,7 +44,6 @@ import (
  "github.com/Fantom-foundation/go-opera/utils"
  "github.com/Fantom-foundation/go-opera/utils/adapters/vecmt2dagidx"
  "github.com/Fantom-foundation/go-opera/valkeystore"
- "github.com/Fantom-foundation/go-opera/vecmt"
 )
 
 const (
@@ -87,13 +89,17 @@ func (g *testGossipStoreAdapter) GetEvent(id hash.Event) dag.Event {
  return e
 }
 
-func makeTestEngine(gdb *Store) (*abft.Lachesis, *vecmt.Index) {
+func makeTmpDB(name string) kvdb.Store {
+ return memorydb.New()
+}
+
+func makeTestEngine(gdb *Store) (*abft.Lachesis, *vecclock.Index) {
  cdb := abft.NewMemStore()
  _ = cdb.ApplyGenesis(&abft.Genesis{
  Epoch: gdb.GetEpoch(),
  Validators: gdb.GetValidators(),
  })
- vecClock := vecmt.NewIndex(panics("Vector clock"), vecmt.LiteConfig())
+ vecClock := vecclock.NewIndex(makeTmpDB, vecclock.LiteConfig())
  engine := abft.NewLachesis(cdb, &testGossipStoreAdapter{gdb}, vecmt2dagidx.Wrap(vecClock), panics("Lachesis"), abft.LiteConfig())
  return engine, vecClock
 }

gossip/emitter/emitter_test.go

@@ -1,6 +1,7 @@
 package emitter
 
 import (
+ "github.com/Fantom-foundation/go-opera/vecclock"
  "math/big"
  "testing"
  "time"
@@ -18,7 +19,6 @@ import (
  "github.com/Fantom-foundation/go-opera/inter"
  "github.com/Fantom-foundation/go-opera/opera"
  "github.com/Fantom-foundation/go-opera/utils/txtime"
- "github.com/Fantom-foundation/go-opera/vecmt"
 )
 
 //go:generate go run github.com/golang/mock/mockgen -package=mock -destination=mock/world.go github.com/Fantom-foundation/go-opera/gossip/emitter External,TxPool,TxSigner,Signer
@@ -44,7 +44,7 @@ func TestEmitter(t *testing.T) {
  external.EXPECT().Unlock().
  AnyTimes()
  external.EXPECT().DagIndex().
- Return((*vecmt.Index)(nil)).
+ Return((*vecclock.Index)(nil)).
  AnyTimes()
  external.EXPECT().IsSynced().
  Return(true).

gossip/emitter/world.go

@@ -2,6 +2,7 @@ package emitter
 
 import (
  "errors"
+ "github.com/Fantom-foundation/go-opera/vecclock"
  "sync"
 
  "github.com/Fantom-foundation/lachesis-base/hash"
@@ -14,7 +15,6 @@ import (
  "github.com/Fantom-foundation/go-opera/inter/state"
  "github.com/Fantom-foundation/go-opera/opera"
  "github.com/Fantom-foundation/go-opera/valkeystore"
- "github.com/Fantom-foundation/go-opera/vecmt"
 )
 
 var (
@@ -31,7 +31,7 @@ type (
  Process(*inter.EventPayload) error
  Broadcast(*inter.EventPayload)
  Build(*inter.MutableEventPayload, func()) error
- DagIndex() *vecmt.Index
+ DagIndex() *vecclock.Index
 
  IsBusy() bool
  IsSynced() bool

gossip/emitter_world.go

@@ -1,6 +1,7 @@
 package gossip
 
 import (
+ "github.com/Fantom-foundation/go-opera/vecclock"
  "sync/atomic"
 
  "github.com/Fantom-foundation/lachesis-base/hash"
@@ -12,7 +13,6 @@ import (
  "github.com/Fantom-foundation/go-opera/inter/state"
  "github.com/Fantom-foundation/go-opera/utils/wgmutex"
  "github.com/Fantom-foundation/go-opera/valkeystore"
- "github.com/Fantom-foundation/go-opera/vecmt"
 )
 
 type emitterWorldProc struct {
@@ -53,7 +53,7 @@ func (ew *emitterWorldProc) Build(e *inter.MutableEventPayload, onIndexed func()
  return ew.s.buildEvent(e, onIndexed)
 }
 
-func (ew *emitterWorldProc) DagIndex() *vecmt.Index {
+func (ew *emitterWorldProc) DagIndex() *vecclock.Index {
  return ew.s.dagIndexer
 }