Merge pull request #10590 from filecoin-project/vyzo/feat/exec-lanes-2

VM Execution Lanes Part II: Hide the lock

chain/consensus/compute_state.go

@@ -93,7 +93,7 @@ func (t *TipSetExecutor) ApplyBlocks(ctx context.Context,
 	}()
 
 	ctx = blockstore.WithHotView(ctx)
-	makeVm := func(base cid.Cid, e abi.ChainEpoch, timestamp uint64) (vm.Executor, error) {
+	makeVm := func(base cid.Cid, e abi.ChainEpoch, timestamp uint64) (vm.Interface, error) {
 		vmopt := &vm.VMOpts{
 			StateBase:      base,
 			Epoch:          e,
@@ -117,7 +117,7 @@ func (t *TipSetExecutor) ApplyBlocks(ctx context.Context,
 
 	var cronGas int64
 
-	runCron := func(vmCron vm.Executor, epoch abi.ChainEpoch) error {
+	runCron := func(vmCron vm.Interface, epoch abi.ChainEpoch) error {
 		cronMsg := &types.Message{
 			To:         cron.Address,
 			From:       builtin.SystemActorAddr,
@@ -170,17 +170,13 @@ func (t *TipSetExecutor) ApplyBlocks(ctx context.Context,
 
 			// run cron for null rounds if any
 			if err = runCron(vmCron, i); err != nil {
-				vmCron.Done()
 				return cid.Undef, cid.Undef, xerrors.Errorf("running cron: %w", err)
 			}
 
 			pstate, err = vmCron.Flush(ctx)
 			if err != nil {
-				vmCron.Done()
 				return cid.Undef, cid.Undef, xerrors.Errorf("flushing cron vm: %w", err)
 			}
-
-			vmCron.Done()
 		}
 
 		// handle state forks
@@ -201,7 +197,6 @@ func (t *TipSetExecutor) ApplyBlocks(ctx context.Context,
 	if err != nil {
 		return cid.Undef, cid.Undef, xerrors.Errorf("making vm: %w", err)
 	}
-	defer vmi.Done()
 
 	var (
 		receipts      []*types.MessageReceipt

chain/gen/genesis/genesis.go

@@ -496,7 +496,6 @@ func VerifyPreSealedData(ctx context.Context, cs *store.ChainStore, sys vm.Sysca
 	if err != nil {
 		return cid.Undef, xerrors.Errorf("failed to create VM: %w", err)
 	}
-	defer vm.Done()
 
 	for mi, m := range template.Miners {
 		for si, s := range m.Sectors {

chain/gen/genesis/miners.go

@@ -88,7 +88,7 @@ func SetupStorageMiners(ctx context.Context, cs *store.ChainStore, sys vm.Syscal
 		return big.Zero(), nil
 	}
 
-	newVM := func(base cid.Cid) (vm.Executor, error) {
+	newVM := func(base cid.Cid) (vm.Interface, error) {
 		vmopt := &vm.VMOpts{
 			StateBase:      base,
 			Epoch:          0,
@@ -108,8 +108,6 @@ func SetupStorageMiners(ctx context.Context, cs *store.ChainStore, sys vm.Syscal
 	if err != nil {
 		return cid.Undef, fmt.Errorf("creating vm: %w", err)
 	}
-	// Note: genesisVm is mutated, so this has to happen in a deferred func; go horror show.
-	defer func() { genesisVm.Done() }()
 
 	if len(miners) == 0 {
 		return cid.Undef, xerrors.New("no genesis miners")
@@ -340,7 +338,6 @@ func SetupStorageMiners(ctx context.Context, cs *store.ChainStore, sys vm.Syscal
 			return cid.Undef, xerrors.Errorf("flushing state tree: %w", err)
 		}
 
-		genesisVm.Done()
 		genesisVm, err = newVM(nh)
 		if err != nil {
 			return cid.Undef, fmt.Errorf("creating new vm: %w", err)
@@ -413,7 +410,6 @@ func SetupStorageMiners(ctx context.Context, cs *store.ChainStore, sys vm.Syscal
 					return cid.Undef, xerrors.Errorf("flushing state tree: %w", err)
 				}
 
-				genesisVm.Done()
 				genesisVm, err = newVM(nh)
 				if err != nil {
 					return cid.Undef, fmt.Errorf("creating new vm: %w", err)
@@ -521,7 +517,6 @@ func SetupStorageMiners(ctx context.Context, cs *store.ChainStore, sys vm.Syscal
 						return cid.Undef, xerrors.Errorf("flushing state tree: %w", err)
 					}
 
-					genesisVm.Done()
 					genesisVm, err = newVM(nh)
 					if err != nil {
 						return cid.Undef, fmt.Errorf("creating new vm: %w", err)

chain/stmgr/call.go

@@ -159,8 +159,6 @@ func (sm *StateManager) callInternal(ctx context.Context, msg *types.Message, pr
 	if err != nil {
 		return nil, xerrors.Errorf("failed to set up vm: %w", err)
 	}
-	// Note: vmi is mutated, so this has to happen in a deferred func; go horror show.
-	defer func() { vmi.Done() }()
 
 	for i, m := range priorMsgs {
 		_, err = vmi.ApplyMessage(ctx, m)
@@ -194,7 +192,6 @@ func (sm *StateManager) callInternal(ctx context.Context, msg *types.Message, pr
 		vmopt.BaseFee = big.Zero()
 		vmopt.StateBase = stateCid
 
-		vmi.Done()
 		vmi, err = sm.newVM(ctx, vmopt)
 		if err != nil {
 			return nil, xerrors.Errorf("failed to set up estimation vm: %w", err)

chain/stmgr/forks_test.go

@@ -56,12 +56,6 @@ const testForkHeight = 40
 type testActor struct {
 }
 
-type mockExecutor struct {
-	vm.Interface
-}
-
-func (*mockExecutor) Done() {}
-
 // must use existing actor that an account is allowed to exec.
 func (testActor) Code() cid.Cid  { return builtin0.PaymentChannelActorCodeID }
 func (testActor) State() cbor.Er { return new(testActorState) }
@@ -184,13 +178,13 @@ func TestForkHeightTriggers(t *testing.T) {
 	registry := builtin.MakeRegistryLegacy([]rtt.VMActor{testActor{}})
 	inv.Register(actorstypes.Version0, nil, registry)
 
-	sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Executor, error) {
+	sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Interface, error) {
 		nvm, err := vm.NewLegacyVM(ctx, vmopt)
 		if err != nil {
 			return nil, err
 		}
 		nvm.SetInvoker(inv)
-		return &mockExecutor{nvm}, nil
+		return nvm, nil
 	})
 
 	cg.SetStateManager(sm)
@@ -302,13 +296,13 @@ func testForkRefuseCall(t *testing.T, nullsBefore, nullsAfter int) {
 	registry := builtin.MakeRegistryLegacy([]rtt.VMActor{testActor{}})
 	inv.Register(actorstypes.Version0, nil, registry)
 
-	sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Executor, error) {
+	sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Interface, error) {
 		nvm, err := vm.NewLegacyVM(ctx, vmopt)
 		if err != nil {
 			return nil, err
 		}
 		nvm.SetInvoker(inv)
-		return &mockExecutor{nvm}, nil
+		return nvm, nil
 	})
 
 	cg.SetStateManager(sm)
@@ -524,13 +518,13 @@ func TestForkPreMigration(t *testing.T) {
 	registry := builtin.MakeRegistryLegacy([]rtt.VMActor{testActor{}})
 	inv.Register(actorstypes.Version0, nil, registry)
 
-	sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Executor, error) {
+	sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Interface, error) {
 		nvm, err := vm.NewLegacyVM(ctx, vmopt)
 		if err != nil {
 			return nil, err
 		}
 		nvm.SetInvoker(inv)
-		return &mockExecutor{nvm}, nil
+		return nvm, nil
 	})
 
 	cg.SetStateManager(sm)
@@ -598,11 +592,11 @@ func TestDisablePreMigration(t *testing.T) {
 	registry := builtin.MakeRegistryLegacy([]rtt.VMActor{testActor{}})
 	inv.Register(actorstypes.Version0, nil, registry)
 
-	sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Executor, error) {
+	sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Interface, error) {
 		nvm, err := vm.NewLegacyVM(ctx, vmopt)
 		require.NoError(t, err)
 		nvm.SetInvoker(inv)
-		return &mockExecutor{nvm}, nil
+		return nvm, nil
 	})
 
 	cg.SetStateManager(sm)
@@ -653,11 +647,11 @@ func TestMigrtionCache(t *testing.T) {
 	registry := builtin.MakeRegistryLegacy([]rtt.VMActor{testActor{}})
 	inv.Register(actorstypes.Version0, nil, registry)
 
-	sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Executor, error) {
+	sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Interface, error) {
 		nvm, err := vm.NewLegacyVM(ctx, vmopt)
 		require.NoError(t, err)
 		nvm.SetInvoker(inv)
-		return &mockExecutor{nvm}, nil
+		return nvm, nil
 	})
 
 	cg.SetStateManager(sm)
@@ -697,11 +691,11 @@ func TestMigrtionCache(t *testing.T) {
 			index.DummyMsgIndex,
 		)
 		require.NoError(t, err)
-		sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Executor, error) {
+		sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Interface, error) {
 			nvm, err := vm.NewLegacyVM(ctx, vmopt)
 			require.NoError(t, err)
 			nvm.SetInvoker(inv)
-			return &mockExecutor{nvm}, nil
+			return nvm, nil
 		})
 
 		ctx := context.Background()

chain/stmgr/stmgr.go

@@ -125,7 +125,7 @@ type StateManager struct {
 	compWait            map[string]chan struct{}
 	stlk                sync.Mutex
 	genesisMsigLk       sync.Mutex
-	newVM               func(context.Context, *vm.VMOpts) (vm.Executor, error)
+	newVM               func(context.Context, *vm.VMOpts) (vm.Interface, error)
 	Syscalls            vm.SyscallBuilder
 	preIgnitionVesting  []msig0.State
 	postIgnitionVesting []msig0.State
@@ -439,12 +439,12 @@ func (sm *StateManager) ValidateChain(ctx context.Context, ts *types.TipSet) err
 	return nil
 }
 
-func (sm *StateManager) SetVMConstructor(nvm func(context.Context, *vm.VMOpts) (vm.Executor, error)) {
+func (sm *StateManager) SetVMConstructor(nvm func(context.Context, *vm.VMOpts) (vm.Interface, error)) {
 	sm.newVM = nvm
 }
 
-func (sm *StateManager) VMConstructor() func(context.Context, *vm.VMOpts) (vm.Executor, error) {
+func (sm *StateManager) VMConstructor() func(context.Context, *vm.VMOpts) (vm.Interface, error) {
-	return func(ctx context.Context, opts *vm.VMOpts) (vm.Executor, error) {
+	return func(ctx context.Context, opts *vm.VMOpts) (vm.Interface, error) {
 		return sm.newVM(ctx, opts)
 	}
 }

chain/stmgr/utils.go

@@ -106,7 +106,6 @@ func ComputeState(ctx context.Context, sm *StateManager, height abi.ChainEpoch,
 	if err != nil {
 		return cid.Undef, nil, err
 	}
-	defer vmi.Done()
 
 	for i, msg := range msgs {
 		// TODO: Use the signed message length for secp messages

chain/vm/execution.go

@@ -33,62 +33,34 @@ var execution *executionEnv
 
 // implementation of vm executor with simple sanity check preventing use after free.
 type vmExecutor struct {
-	lk    sync.RWMutex
 	vmi  Interface
-	token *executionToken
+	lane ExecutionLane
-	done  bool
 }
 
-var _ Executor = (*vmExecutor)(nil)
+var _ Interface = (*vmExecutor)(nil)
 
-func newVMExecutor(vmi Interface, token *executionToken) Executor {
+func newVMExecutor(vmi Interface, lane ExecutionLane) Interface {
-	return &vmExecutor{vmi: vmi, token: token}
+	return &vmExecutor{vmi: vmi, lane: lane}
 }
 
 func (e *vmExecutor) ApplyMessage(ctx context.Context, cmsg types.ChainMsg) (*ApplyRet, error) {
-	e.lk.RLock()
+	token := execution.getToken(e.lane)
-	defer e.lk.RUnlock()
+	defer token.Done()
-
-	if e.done {
-		return nil, ErrExecutorDone
-	}
 
 	return e.vmi.ApplyMessage(ctx, cmsg)
 }
 
 func (e *vmExecutor) ApplyImplicitMessage(ctx context.Context, msg *types.Message) (*ApplyRet, error) {
-	e.lk.RLock()
+	token := execution.getToken(e.lane)
-	defer e.lk.RUnlock()
+	defer token.Done()
-
-	if e.done {
-		return nil, ErrExecutorDone
-	}
 
 	return e.vmi.ApplyImplicitMessage(ctx, msg)
 }
 
 func (e *vmExecutor) Flush(ctx context.Context) (cid.Cid, error) {
-	e.lk.RLock()
-	defer e.lk.RUnlock()
-
-	if e.done {
-		return cid.Undef, ErrExecutorDone
-	}
-
 	return e.vmi.Flush(ctx)
 }
 
-func (e *vmExecutor) Done() {
-	e.lk.Lock()
-	defer e.lk.Unlock()
-
-	if !e.done {
-		e.token.Done()
-		e.token = nil
-		e.done = true
-	}
-}
-
 type executionToken struct {
 	lane     ExecutionLane
 	reserved int

chain/vm/vmi.go

@@ -37,17 +37,6 @@ type Interface interface {
 	Flush(ctx context.Context) (cid.Cid, error)
 }
 
-// Executor is the general vm execution interface, which is prioritized according to execution lanes.
-// User must call Done when it is done with this executor to release resource holds by the execution
-// environment
-type Executor interface {
-	Interface
-
-	// Done must be called when done with the executor to release resource holds.
-	// It is an error to invoke Interface methods after Done has been called.
-	Done()
-}
-
 // WARNING: You will not affect your node's execution by misusing this feature, but you will confuse yourself thoroughly!
 // An envvar that allows the user to specify debug actors bundles to be used by the FVM
 // alongside regular execution. This is basically only to be used to print out specific logging information.
@@ -65,20 +54,17 @@ func makeVM(ctx context.Context, opts *VMOpts) (Interface, error) {
 	return NewLegacyVM(ctx, opts)
 }
 
-func NewVM(ctx context.Context, opts *VMOpts) (Executor, error) {
+func NewVM(ctx context.Context, opts *VMOpts) (Interface, error) {
 	switch opts.ExecutionLane {
 	case ExecutionLaneDefault, ExecutionLanePriority:
 	default:
 		return nil, fmt.Errorf("invalid execution lane: %d", opts.ExecutionLane)
 	}
 
-	token := execution.getToken(opts.ExecutionLane)
-
 	vmi, err := makeVM(ctx, opts)
 	if err != nil {
-		token.Done()
 		return nil, err
 	}
 
-	return newVMExecutor(vmi, token), nil
+	return newVMExecutor(vmi, opts.ExecutionLane), nil
 }

conformance/driver.go

@@ -158,7 +158,7 @@ func (d *Driver) ExecuteTipset(bs blockstore.Blockstore, ds ds.Batching, params
 		results:  []*vm.ApplyRet{},
 	}
 
-	sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Executor, error) {
+	sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Interface, error) {
 		vmopt.CircSupplyCalc = func(context.Context, abi.ChainEpoch, *state.StateTree) (abi.TokenAmount, error) {
 			return big.Zero(), nil
 		}