Remove prefetch functionality (#596)

* remove prefetch functionality and write by multi-threading

* apply goimports

* fix a few trivial diff

baseapp/abci.go

@@ -21,8 +21,6 @@ import (
  "github.com/line/lbm-sdk/telemetry"
  sdk "github.com/line/lbm-sdk/types"
  sdkerrors "github.com/line/lbm-sdk/types/errors"
-
- iavlstore "github.com/line/lbm-sdk/store/iavl"
 )
 
 // InitChain implements the ABCI interface. It runs the initialization logic
@@ -280,13 +278,11 @@ func (app *BaseApp) CheckTxAsync(req abci.RequestCheckTx, callback abci.CheckTxC
 func (app *BaseApp) BeginRecheckTx(req abci.RequestBeginRecheckTx) abci.ResponseBeginRecheckTx {
  // NOTE: This is safe because Ostracon holds a lock on the mempool for Rechecking.
  app.setCheckState(req.Header)
- iavlstore.PausePrefetcher()
  return abci.ResponseBeginRecheckTx{Code: abci.CodeTypeOK}
 }
 
 // EndRecheckTx implements the ABCI interface.
 func (app *BaseApp) EndRecheckTx(req abci.RequestEndRecheckTx) abci.ResponseEndRecheckTx {
- iavlstore.ResumePrefetcher()
  return abci.ResponseEndRecheckTx{Code: abci.CodeTypeOK}
 }
 
@@ -344,9 +340,7 @@ func (app *BaseApp) Commit() (res abci.ResponseCommit) {
  // The write to the DeliverTx state writes all state transitions to the root
  // MultiStore (app.cms) so when Commit() is called is persists those values.
  app.deliverState.ms.Write()
- iavlstore.PausePrefetcher()
  commitID := app.cms.Commit()
- iavlstore.ResumePrefetcher()
  app.logger.Info("commit synced", "commit", fmt.Sprintf("%X", commitID))
 
  // iavl, db & disk stats

baseapp/grpcrouter.go

@@ -2,7 +2,6 @@ package baseapp
 
 import (
  "fmt"
- "sync"
 
  gogogrpc "github.com/gogo/protobuf/grpc"
  "google.golang.org/grpc"
@@ -20,7 +19,6 @@ var protoCodec = encoding.GetCodec(proto.Name)
 
 // GRPCQueryRouter routes ABCI Query requests to GRPC handlers
 type GRPCQueryRouter struct {
- lck sync.Mutex
  routes map[string]GRPCQueryHandler
  interfaceRegistry codectypes.InterfaceRegistry
  serviceData []serviceData
@@ -37,7 +35,6 @@ var _ gogogrpc.Server = &GRPCQueryRouter{}
 // NewGRPCQueryRouter creates a new GRPCQueryRouter
 func NewGRPCQueryRouter() *GRPCQueryRouter {
  return &GRPCQueryRouter{
- lck: sync.Mutex{},
  routes: map[string]GRPCQueryHandler{},
  }
 }
@@ -49,8 +46,6 @@ type GRPCQueryHandler = func(ctx sdk.Context, req abci.RequestQuery) (abci.Respo
 // Route returns the GRPCQueryHandler for a given query route path or nil
 // if not found
 func (qrt *GRPCQueryRouter) Route(path string) GRPCQueryHandler {
- qrt.lck.Lock()
- defer qrt.lck.Unlock()
  handler, found := qrt.routes[path]
  if !found {
  return nil
@@ -64,9 +59,6 @@ func (qrt *GRPCQueryRouter) Route(path string) GRPCQueryHandler {
 // This functions PANICS:
 // - if a protobuf service is registered twice.
 func (qrt *GRPCQueryRouter) RegisterService(sd *grpc.ServiceDesc, handler interface{}) {
- qrt.lck.Lock()
- defer qrt.lck.Unlock()
-
  // adds a top-level query handler based on the gRPC service name
  for _, method := range sd.Methods {
  fqName := fmt.Sprintf("/%s/%s", sd.ServiceName, method.MethodName)
@@ -127,9 +119,6 @@ func (qrt *GRPCQueryRouter) RegisterService(sd *grpc.ServiceDesc, handler interf
 // SetInterfaceRegistry sets the interface registry for the router. This will
 // also register the interface reflection gRPC service.
 func (qrt *GRPCQueryRouter) SetInterfaceRegistry(interfaceRegistry codectypes.InterfaceRegistry) {
- // qrt.lck.Lock()
- // defer qrt.lck.Unlock()
-
  qrt.interfaceRegistry = interfaceRegistry
  // Once we have an interface registry, we can register the interface
  // registry reflection gRPC service.

server/mock/store.go

@@ -180,10 +180,6 @@ func (kv kvStore) Delete(key []byte) {
  delete(kv.store, string(key))
 }
 
-func (kv kvStore) Prefetch(key []byte, forSet bool) (hits, misses int, value []byte) {
- return 0, 0, nil
-}
-
 func (kv kvStore) Prefix(prefix []byte) sdk.KVStore {
  panic("not implemented")
 }

server/mock/store_test.go

@@ -27,7 +27,6 @@ func TestStore(t *testing.T) {
  require.False(t, store.Has(k))
  store.Set(k, v)
  require.True(t, store.Has(k))
- store.Prefetch(k, false)
  require.Equal(t, v, store.Get(k))
  store.Delete(k)
  require.False(t, store.Has(k))

simapp/ante_handler.go

@@ -31,7 +31,7 @@ func NewAnteHandler(
  // ante.NewDeductFeeDecorator(ak, bankKeeper),
  ante.NewSigGasConsumeDecorator(ak, sigGasConsumer),
  ante.NewSigVerificationDecorator(ak, signModeHandler),
- ante.NewIncrementSequenceDecorator(ak, bankKeeper),
+ ante.NewIncrementSequenceDecorator(ak),
  ibcante.NewAnteDecorator(channelKeeper),
  )
 }

store/cachekv/store.go

@@ -94,15 +94,6 @@ func (store *Store) Delete(key []byte) {
  store.setCacheValue(key, nil, true, true)
 }
 
-// Prefetch implements types.KVStore.
-func (store *Store) Prefetch(key []byte, forSet bool) (hits, misses int, value []byte) {
- defer telemetry.MeasureSince(time.Now(), "store", "cachekv", "prefetch")
-
- // do not update cache
- types.AssertValidKey(key)
- return store.parent.Prefetch(key, forSet)
-}
-
 // Implements Cachetypes.KVStore.
 func (store *Store) Write() {
  store.mtx.Lock()

store/cachekv/store_test.go

@@ -33,7 +33,6 @@ func TestCacheKVStore(t *testing.T) {
  // put something in mem and in cache
  mem.Set(keyFmt(1), valFmt(1))
  st.Set(keyFmt(1), valFmt(1))
- st.Prefetch(keyFmt(1), false)
  require.Equal(t, valFmt(1), st.Get(keyFmt(1)))
 
  // update it in cache, shoudn't change mem

store/cachemulti/store.go

@@ -3,7 +3,6 @@ package cachemulti
 import (
  "fmt"
  "io"
- "sync"
 
  tmdb "github.com/line/tm-db/v2"
 
@@ -138,26 +137,8 @@ func (cms Store) GetStoreType() types.StoreType {
 // Write calls Write on each underlying store.
 func (cms Store) Write() {
  cms.db.Write()
- var wg sync.WaitGroup
- var panicMsg interface{}
  for _, store := range cms.stores {
- wg.Add(1)
- go func(s types.CacheWrap) {
- defer func() {
- if msg := recover(); msg != nil {
- if panicMsg == nil {
- panicMsg = msg
- }
- }
- wg.Done()
- }()
-
- s.Write()
- }(store)
- }
- wg.Wait()
- if panicMsg != nil {
- panic(panicMsg)
+ store.Write()
  }
 }
 

store/dbadapter/store.go

@@ -51,15 +51,6 @@ func (dsa Store) Delete(key []byte) {
  }
 }
 
-// Prefetch wraps the underlying DB's Get method panicing on error.
-func (dsa Store) Prefetch(key []byte, forSet bool) (hits, misses int, value []byte) {
- v, err := dsa.DB.Get(key)
- if err != nil {
- return 0, 0, nil
- }
- return 1, 1, v
-}
-
 // Iterator wraps the underlying DB's Iterator method panicing on error.
 func (dsa Store) Iterator(start, end []byte) types.Iterator {
  iter, err := dsa.DB.Iterator(start, end)

store/dbadapter/store_test.go

@@ -35,8 +35,6 @@ func TestAccessors(t *testing.T) {
  retFoo := []byte("xxx")
  mockDB.EXPECT().Get(gomock.Eq(key)).Times(1).Return(retFoo, nil)
  require.True(t, bytes.Equal(retFoo, store.Get(key)))
- mockDB.EXPECT().Get(gomock.Eq(key)).Times(1).Return(retFoo, nil)
- store.Prefetch(key, false)
 
  mockDB.EXPECT().Get(gomock.Eq(key)).Times(1).Return(nil, errFoo)
  require.Panics(t, func() { store.Get(key) })

store/gaskv/store.go

@@ -71,12 +71,6 @@ func (gs *Store) Delete(key []byte) {
  gs.parent.Delete(key)
 }
 
-// Implements KVStore.
-func (gs *Store) Prefetch(key []byte, forSet bool) (hits, misses int, value []byte) {
- defer telemetry.MeasureSince(time.Now(), "store", "gaskv", "load")
- return gs.parent.Prefetch(key, forSet)
-}
-
 // Iterator implements the KVStore interface. It returns an iterator which
 // incurs a flat gas cost for seeking to the first key/value pair and a variable
 // gas cost based on the current value's length if the iterator is valid.

store/gaskv/store_test.go

@@ -32,7 +32,6 @@ func TestGasKVStoreBasic(t *testing.T) {
  require.Panics(t, func() { st.Set([]byte(""), []byte("value")) }, "setting an empty key should panic")
 
  require.Empty(t, st.Get(keyFmt(1)), "Expected `key1` to be empty")
- st.Prefetch(keyFmt(1), false)
  st.Set(keyFmt(1), valFmt(1))
  require.Equal(t, valFmt(1), st.Get(keyFmt(1)))
  st.Delete(keyFmt(1))

store/iavl/metrics.go

@@ -15,11 +15,10 @@ const (
 
 // Metrics contains metrics exposed by this package.
 type Metrics struct {
- IAVLCacheHits metrics.Gauge
- IAVLCacheMisses metrics.Gauge
- IAVLCacheEntries metrics.Gauge
- IAVLCacheBytes metrics.Gauge
- IAVLCachePeakBytes metrics.Gauge
+ IAVLCacheHits metrics.Gauge
+ IAVLCacheMisses metrics.Gauge
+ IAVLCacheEntries metrics.Gauge
+ IAVLCacheBytes metrics.Gauge
 }
 
 // PrometheusMetrics returns Metrics build using Prometheus client library.
@@ -55,23 +54,16 @@ func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
  Name: "bytes_size",
  Help: "Cache bytes size of the iavl cache",
  }, labels).With(labelsAndValues...),
- IAVLCachePeakBytes: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
- Namespace: namespace,
- Subsystem: MetricsSubsystem,
- Name: "peak_bytes_size",
- Help: "Peak cache bytes size of the iavl cache",
- }, labels).With(labelsAndValues...),
  }
 }
 
 // NopMetrics returns no-op Metrics.
 func NopMetrics() *Metrics {
  return &Metrics{
- IAVLCacheHits: discard.NewGauge(),
- IAVLCacheMisses: discard.NewGauge(),
- IAVLCacheEntries: discard.NewGauge(),
- IAVLCacheBytes: discard.NewGauge(),
- IAVLCachePeakBytes: discard.NewGauge(),
+ IAVLCacheHits: discard.NewGauge(),
+ IAVLCacheMisses: discard.NewGauge(),
+ IAVLCacheEntries: discard.NewGauge(),
+ IAVLCacheBytes: discard.NewGauge(),
  }
 }