Relax locking contention

baseapp/baseapp.go

@@ -165,7 +165,8 @@
 
 	TmConfig *tmcfg.Config
 
-	TracingInfo *tracing.Info
+	TracingInfo    *tracing.Info
+	TracingEnabled bool
 
 	concurrencyWorkers int
 	occEnabled         bool
@@ -244,7 +245,8 @@
 	tp := trace.NewNoopTracerProvider()
 	otel.SetTracerProvider(trace.NewNoopTracerProvider())
 	tr := tp.Tracer("component-main")
-	if tracingEnabled := cast.ToBool(appOpts.Get(tracing.FlagTracing)); tracingEnabled {
+	tracingEnabled := cast.ToBool(appOpts.Get(tracing.FlagTracing))
+	if tracingEnabled {
 		tp, err := tracing.DefaultTracerProvider()
 		if err != nil {
 			panic(err)
@@ -267,8 +269,9 @@
 			grpcQueryRouter:  NewGRPCQueryRouter(),
 			msgServiceRouter: NewMsgServiceRouter(),
 		},
-		txDecoder: txDecoder,
-		TmConfig:  tmConfig,
+		txDecoder:      txDecoder,
+		TmConfig:       tmConfig,
+		TracingEnabled: tracingEnabled,
 		TracingInfo: &tracing.Info{
 			Tracer: &tr,
 		},
@@ -316,8 +319,8 @@
 }
 
 // ConcurrencyWorkers returns the number of concurrent workers for the BaseApp.
 func (app *BaseApp) ConcurrencyWorkers() int {
 	return app.concurrencyWorkers
 }
 
 // OccEnabled returns the whether OCC is enabled for the BaseApp.
@@ -613,8 +616,8 @@
 	}
 }
 
 func (app *BaseApp) prepareProcessProposalState(headerHash []byte) {
 	app.processProposalState.SetContext(app.processProposalState.Context().
 		WithHeaderHash(headerHash).
 		WithConsensusParams(app.GetConsensusParams(app.processProposalState.Context())))
 
@@ -844,11 +847,13 @@
 	// resources are acceessed by the ante handlers and message handlers.
 	defer acltypes.SendAllSignalsForTx(ctx.TxCompletionChannels())
 	acltypes.WaitForAllSignalsForTx(ctx.TxBlockingChannels())
-	// check for existing parent tracer, and if applicable, use it
-	spanCtx, span := app.TracingInfo.StartWithContext("RunTx", ctx.TraceSpanContext())
-	defer span.End()
-	ctx = ctx.WithTraceSpanContext(spanCtx)
-	span.SetAttributes(attribute.String("txHash", fmt.Sprintf("%X", sha256.Sum256(txBytes))))
+	if app.TracingEnabled {
+		// check for existing parent tracer, and if applicable, use it
+		spanCtx, span := app.TracingInfo.StartWithContext("RunTx", ctx.TraceSpanContext())
+		defer span.End()
+		ctx = ctx.WithTraceSpanContext(spanCtx)
+		span.SetAttributes(attribute.String("txHash", fmt.Sprintf("%X", sha256.Sum256(txBytes))))
+	}
 
 	// NOTE: GasWanted should be returned by the AnteHandler. GasUsed is
 	// determined by the GasMeter. We need access to the context to get the gas
@@ -881,9 +886,12 @@
 	}
 
 	if app.anteHandler != nil {
-		// trace AnteHandler
-		_, anteSpan := app.TracingInfo.StartWithContext("AnteHandler", ctx.TraceSpanContext())
-		defer anteSpan.End()
+		var anteSpan trace.Span
+		if app.TracingEnabled {
+			// trace AnteHandler
+			_, anteSpan := app.TracingInfo.StartWithContext("AnteHandler", ctx.TraceSpanContext())
+			defer anteSpan.End()
+		}
 		var (
 			anteCtx sdk.Context
 			msCache sdk.CacheMultiStore
@@ -927,7 +935,7 @@
 			storeAccessOpEvents := msCache.GetEvents()
 			accessOps := ctx.TxMsgAccessOps()[acltypes.ANTE_MSG_INDEX]
 
 			// TODO: (occ) This is an example of where we do our current validation. Note that this validation operates on the declared dependencies for a TX / antehandler + the utilized dependencies, whereas the validation
 			missingAccessOps := ctx.MsgValidator().ValidateAccessOperations(accessOps, storeAccessOpEvents)
 			if len(missingAccessOps) != 0 {
 				for op := range missingAccessOps {
@@ -942,7 +950,9 @@
 		priority = ctx.Priority()
 		msCache.Write()
 		anteEvents = events.ToABCIEvents()
-		anteSpan.End()
+		if app.TracingEnabled {
+			anteSpan.End()
+		}
 	}
 
 	// Create a new Context based off of the existing Context with a MultiStore branch
@@ -987,9 +997,11 @@
 			panic(err)
 		}
 	}()
-	spanCtx, span := app.TracingInfo.StartWithContext("RunMsgs", ctx.TraceSpanContext())
-	defer span.End()
-	ctx = ctx.WithTraceSpanContext(spanCtx)
+	if app.TracingEnabled {
+		spanCtx, span := app.TracingInfo.StartWithContext("RunMsgs", ctx.TraceSpanContext())
+		defer span.End()
+		ctx = ctx.WithTraceSpanContext(spanCtx)
+	}
 	msgLogs := make(sdk.ABCIMessageLogs, 0, len(msgs))
 	events := sdk.EmptyEvents()
 	txMsgData := &sdk.TxMsgData{
@@ -1069,8 +1081,8 @@
 		storeAccessOpEvents := msgMsCache.GetEvents()
 		accessOps := ctx.TxMsgAccessOps()[i]
 		missingAccessOps := ctx.MsgValidator().ValidateAccessOperations(accessOps, storeAccessOpEvents)
 		// TODO: (occ) This is where we are currently validating our per message dependencies,
 		// whereas validation will be done holistically based on the mvkv for OCC approach
 		if len(missingAccessOps) != 0 {
 			for op := range missingAccessOps {
 				ctx.Logger().Info((fmt.Sprintf("eventMsgName=%s Missing Access Operation:%s ", eventMsgName, op.String())))

store/cachekv/mergeiterator.go

@@ -16,11 +16,10 @@ import (
 //
 // TODO: Optimize by memoizing.
 type cacheMergeIterator struct {
-	parent       types.Iterator
-	cache        types.Iterator
-	ascending    bool
-	storeKey     sdktypes.StoreKey
-	eventManager *sdktypes.EventManager
+	parent    types.Iterator
+	cache     types.Iterator
+	ascending bool
+	storeKey  sdktypes.StoreKey
 }
 
 var _ types.Iterator = (*cacheMergeIterator)(nil)
@@ -29,14 +28,12 @@ func NewCacheMergeIterator(
 	parent, cache types.Iterator,
 	ascending bool,
 	storeKey sdktypes.StoreKey,
-	eventManager *sdktypes.EventManager,
 ) *cacheMergeIterator {
 	iter := &cacheMergeIterator{
-		parent:       parent,
-		cache:        cache,
-		ascending:    ascending,
-		storeKey:     storeKey,
-		eventManager: eventManager,
+		parent:    parent,
+		cache:     cache,
+		ascending: ascending,
+		storeKey:  storeKey,
 	}
 
 	return iter
@@ -138,14 +135,12 @@ func (iter *cacheMergeIterator) Value() []byte {
 	// If parent is invalid, get the cache value.
 	if !iter.parent.Valid() {
 		value := iter.cache.Value()
-		iter.eventManager.EmitResourceAccessReadEvent("iterator", iter.storeKey, iter.cache.Key(), value)
 		return value
 	}
 
 	// If cache is invalid, get the parent value.
 	if !iter.cache.Valid() {
 		value := iter.parent.Value()
-		iter.eventManager.EmitResourceAccessReadEvent("iterator", iter.storeKey, iter.parent.Key(), value)
 		return value
 	}
 
@@ -156,11 +151,9 @@ func (iter *cacheMergeIterator) Value() []byte {
 	switch cmp {
 	case -1: // parent < cache
 		value := iter.parent.Value()
-		iter.eventManager.EmitResourceAccessReadEvent("iterator", iter.storeKey, keyP, value)
 		return value
 	case 0, 1: // parent >= cache
 		value := iter.cache.Value()
-		iter.eventManager.EmitResourceAccessReadEvent("iterator", iter.storeKey, keyC, value)
 		return value
 	default:
 		panic("invalid comparison result")

store/cachekv/mergeiterator_test.go

@@ -6,15 +6,13 @@ import (
 	"github.com/cosmos/cosmos-sdk/store/cachekv"
 	"github.com/cosmos/cosmos-sdk/store/dbadapter"
 	"github.com/cosmos/cosmos-sdk/store/types"
-	sdktypes "github.com/cosmos/cosmos-sdk/types"
 	"github.com/stretchr/testify/require"
 	dbm "github.com/tendermint/tm-db"
 )
 
 func TestMangerIterator(t *testing.T) {
 	// initiate mock kvstore
 	mem := dbadapter.Store{DB: dbm.NewMemDB()}
-	eventManager := sdktypes.NewEventManager()
 	kvstore := cachekv.NewStore(mem, types.NewKVStoreKey("CacheKvTest"), types.DefaultCacheSizeLimit)
 	value := randSlice(defaultValueSizeBz)
 	startKey := randSlice(32)
@@ -29,27 +27,13 @@ func TestMangerIterator(t *testing.T) {
 	cache := kvstore.Iterator(nil, nil)
 	for ; cache.Valid(); cache.Next() {
 	}
-	iter := cachekv.NewCacheMergeIterator(parent, cache, true, types.NewKVStoreKey("CacheKvTest"), eventManager)
+	iter := cachekv.NewCacheMergeIterator(parent, cache, true, types.NewKVStoreKey("CacheKvTest"))
 
-	// get the next value
-	iter.Value()
-
-	// assert the resource access is still emitted correctly when the cache store is unavailable
-	require.Equal(t, "access_type", string(eventManager.Events()[0].Attributes[0].Key))
-	require.Equal(t, "read", string(eventManager.Events()[0].Attributes[0].Value))
-	require.Equal(t, "store_key", string(eventManager.Events()[0].Attributes[1].Key))
-	require.Equal(t, "CacheKvTest", string(eventManager.Events()[0].Attributes[1].Value))
-
-	// assert event emission when cache is available
-	cache = kvstore.Iterator(keys[1], keys[2])
-	iter = cachekv.NewCacheMergeIterator(parent, cache, true, types.NewKVStoreKey("CacheKvTest"), eventManager)
+	// get the next value and it should not be nil
+	nextValue := iter.Value()
+	require.NotNil(t, nextValue)
 
 	// get the next value
-	iter.Value()
-
-	// assert the resource access is still emitted correctly when the cache store is available
-	require.Equal(t, "access_type", string(eventManager.Events()[0].Attributes[0].Key))
-	require.Equal(t, "read", string(eventManager.Events()[0].Attributes[0].Value))
-	require.Equal(t, "store_key", string(eventManager.Events()[0].Attributes[1].Key))
-	require.Equal(t, "CacheKvTest", string(eventManager.Events()[0].Attributes[1].Value))
+	nextValue = iter.Value()
+	require.NotNil(t, nextValue)
 }

store/cachekv/store.go

@@ -56,8 +56,6 @@ func (store *Store) GetEvents() []abci.Event {
 
 // Implements Store
 func (store *Store) ResetEvents() {
-	store.mtx.Lock()
-	defer store.mtx.Unlock()
 	store.eventManager = sdktypes.NewEventManager()
 }
 
@@ -77,7 +75,6 @@ func (store *Store) getFromCache(key []byte) []byte {
 // Get implements types.KVStore.
 func (store *Store) Get(key []byte) (value []byte) {
 	types.AssertValidKey(key)
-	store.eventManager.EmitResourceAccessReadEvent("get", store.storeKey, key, value)
 	return store.getFromCache(key)
 }
 
@@ -86,13 +83,11 @@ func (store *Store) Set(key []byte, value []byte) {
 	types.AssertValidKey(key)
 	types.AssertValidValue(value)
 	store.setCacheValue(key, value, false, true)
-	store.eventManager.EmitResourceAccessWriteEvent("set", store.storeKey, key, value)
 }
 
 // Has implements types.KVStore.
 func (store *Store) Has(key []byte) bool {
 	value := store.Get(key)
-	store.eventManager.EmitResourceAccessReadEvent("has", store.storeKey, key, value)
 	return value != nil
 }
 
@@ -194,7 +189,7 @@ func (store *Store) iterator(start, end []byte, ascending bool) types.Iterator {
 	}()
 	store.dirtyItems(start, end)
 	cache = newMemIterator(start, end, store.sortedCache, store.deleted, ascending, store.eventManager, store.storeKey)
-	return NewCacheMergeIterator(parent, cache, ascending, store.storeKey, store.eventManager)
+	return NewCacheMergeIterator(parent, cache, ascending, store.storeKey)
 }
 
 func findStartIndex(strL []string, startQ string) int {