feat: introduce vmpool to make each vm run on single thread (#272)

* introduce vmpool to make each vm run on single thread

* fix lint

* use index to rotate

* use atomic

x/move/keeper/genesis.go

@@ -46,10 +46,13 @@ func (k Keeper) Initialize(
 		_allowedPublishers[i] = addr
 	}
 
-	// The default upgrade policy is compatible when it's not set,
-	// so skip the registration at initialize.
+	// TODO - remove this after loader v2 is installed
+	vm := k.acquireVM(ctx)
+	defer k.releaseVM()
+
+	// The default upgrade policy is compatible when it's not set.
 	vmStore := types.NewVMStore(ctx, k.VMStore)
-	execRes, err := k.moveVM.Initialize(vmStore, api, env, vmtypes.NewModuleBundle(modules...), _allowedPublishers)
+	execRes, err := vm.Initialize(vmStore, api, env, vmtypes.NewModuleBundle(modules...), _allowedPublishers)
 	if err != nil {
 		return err
 	}

x/move/keeper/handler.go

@@ -164,6 +164,10 @@ func (k Keeper) executeEntryFunction(
 	args [][]byte,
 	isJSON bool,
 ) error {
+	// TODO - remove this after loader v2 is installed
+	vm := k.acquireVM(ctx)
+	defer k.releaseVM()
+
 	payload, err := types.BuildExecuteEntryFunctionPayload(
 		moduleAddr,
 		moduleName,
@@ -200,7 +204,7 @@ func (k Keeper) executeEntryFunction(
 
 	// run vm
 	gasBalance := gasForRuntime
-	execRes, err := k.moveVM.ExecuteEntryFunction(
+	execRes, err := vm.ExecuteEntryFunction(
 		&gasBalance,
 		types.NewVMStore(sdkCtx, k.VMStore),
 		NewApi(k, sdkCtx),
@@ -280,6 +284,10 @@ func (k Keeper) executeScript(
 	args [][]byte,
 	isJSON bool,
 ) error {
+	// TODO - remove this after loader v2 is installed
+	vm := k.acquireVM(ctx)
+	defer k.releaseVM()
+
 	// prepare payload
 	payload, err := types.BuildExecuteScriptPayload(
 		byteCodes,
@@ -315,7 +323,7 @@ func (k Keeper) executeScript(
 
 	// run vm
 	gasBalance := gasForRuntime
-	execRes, err := k.moveVM.ExecuteScript(
+	execRes, err := vm.ExecuteScript(
 		&gasBalance,
 		types.NewVMStore(sdkCtx, k.VMStore),
 		NewApi(k, sdkCtx),
@@ -557,6 +565,10 @@ func (k Keeper) executeViewFunction(
 	args [][]byte,
 	isJSON bool,
 ) (vmtypes.ViewOutput, uint64, error) {
+	// TODO - remove this after loader v2 is installed
+	vm := k.acquireVM(ctx)
+	defer k.releaseVM()
+
 	payload, err := types.BuildExecuteViewFunctionPayload(
 		moduleAddr,
 		moduleName,
@@ -586,7 +598,7 @@ func (k Keeper) executeViewFunction(
 	gasForRuntime := gasMeter.Limit() - gasMeter.GasConsumedToLimit()
 
 	gasBalance := gasForRuntime
-	viewRes, err := k.moveVM.ExecuteViewFunction(
+	viewRes, err := vm.ExecuteViewFunction(
 		&gasBalance,
 		types.NewVMStore(ctx, k.VMStore),
 		api,

x/move/keeper/keeper.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"errors"
 
+	"golang.org/x/sync/semaphore"
+
 	"cosmossdk.io/collections"
 	"cosmossdk.io/core/address"
 	corestoretypes "cosmossdk.io/core/store"
@@ -41,8 +43,10 @@ type Keeper struct {
 
 	config moveconfig.MoveConfig
 
-	// moveVM instance
-	moveVM types.VMEngine
+	// TODO - remove after loader v2
+	moveVMs         []types.VMEngine
+	moveVMIdx       *uint64
+	moveVMSemaphore *semaphore.Weighted
 
 	feeCollector string
 	authority    string
@@ -85,12 +89,19 @@ func NewKeeper(
 		moveConfig.ContractSimulationGasLimit = moveconfig.DefaultContractSimulationGasLimit
 	}
 
-	moveVM, err := vm.NewVM(vmtypes.InitiaVMConfig{
-		// TODO: check this before mainnet
-		AllowUnstable: true,
-	})
-	if err != nil {
-		panic(err)
+	vmCount := 10
+	moveVMIdx := uint64(0)
+	vms := make([]types.VMEngine, vmCount)
+	for i := 0; i < vmCount; i++ {
+		moveVM, err := vm.NewVM(vmtypes.InitiaVMConfig{
+			// TODO: check this before mainnet
+			AllowUnstable: true,
+		})
+		if err != nil {
+			panic(err)
+		}
+
+		vms[i] = &moveVM
 	}
 
 	sb := collections.NewSchemaBuilder(storeService)
@@ -103,7 +114,9 @@ func NewKeeper(
 		msgRouter:           msgRouter,
 		grpcRouter:          grpcRouter,
 		config:              moveConfig,
-		moveVM:              &moveVM,
+		moveVMs:             vms,
+		moveVMIdx:           &moveVMIdx,
+		moveVMSemaphore:     semaphore.NewWeighted(int64(vmCount)),
 		distrKeeper:         distrKeeper,
 		StakingKeeper:       stakingKeeper,
 		RewardKeeper:        rewardKeeper,

x/move/keeper/vmpool.go

@@ -0,0 +1,24 @@
+package keeper
+
+import (
+	"context"
+	"sync/atomic"
+
+	"github.com/initia-labs/initia/x/move/types"
+)
+
+func (k Keeper) acquireVM(ctx context.Context) (vm types.VMEngine) {
+	err := k.moveVMSemaphore.Acquire(ctx, 1)
+	if err != nil {
+		panic(err)
+	}
+
+	idx := atomic.AddUint64(k.moveVMIdx, 1)
+	vm = k.moveVMs[idx%uint64(len(k.moveVMs))]
+
+	return
+}
+
+func (k Keeper) releaseVM() {
+	k.moveVMSemaphore.Release(1)
+}