Skip to content

Commit

Permalink
Merge pull request filecoin-project#7966 from filecoin-project/asr/co…
Browse files Browse the repository at this point in the history
…mpute-state-vm

Fix: chain: create a new VM for each epoch
  • Loading branch information
arajasek authored Jan 18, 2022
2 parents 7dba12f + 9ec1abf commit c0f29ea
Show file tree
Hide file tree
Showing 3 changed files with 38 additions and 49 deletions.
56 changes: 25 additions & 31 deletions chain/consensus/filcns/compute_state.go
Original file line number Diff line number Diff line change
Expand Up @@ -92,29 +92,24 @@ func (t *TipSetExecutor) ApplyBlocks(ctx context.Context, sm *stmgr.StateManager
partDone()
}()

makeVmWithBaseState := func(base cid.Cid) (*vm.VM, error) {
makeVmWithBaseStateAndEpoch := func(base cid.Cid, e abi.ChainEpoch) (*vm.VM, error) {
vmopt := &vm.VMOpts{
StateBase: base,
Epoch: epoch,
Epoch: e,
Rand: r,
Bstore: sm.ChainStore().StateBlockstore(),
Actors: NewActorRegistry(),
Syscalls: sm.Syscalls,
CircSupplyCalc: sm.GetVMCirculatingSupply,
NetworkVersion: sm.GetNetworkVersion(ctx, epoch),
NetworkVersion: sm.GetNetworkVersion(ctx, e),
BaseFee: baseFee,
LookbackState: stmgr.LookbackStateGetterForTipset(sm, ts),
}

return sm.VMConstructor()(ctx, vmopt)
}

vmi, err := makeVmWithBaseState(pstate)
if err != nil {
return cid.Undef, cid.Undef, xerrors.Errorf("making vm: %w", err)
}

runCron := func(epoch abi.ChainEpoch) error {
runCron := func(vmCron *vm.VM, epoch abi.ChainEpoch) error {
cronMsg := &types.Message{
To: cron.Address,
From: builtin.SystemActorAddr,
Expand All @@ -126,59 +121,58 @@ func (t *TipSetExecutor) ApplyBlocks(ctx context.Context, sm *stmgr.StateManager
Method: cron.Methods.EpochTick,
Params: nil,
}
ret, err := vmi.ApplyImplicitMessage(ctx, cronMsg)
ret, err := vmCron.ApplyImplicitMessage(ctx, cronMsg)
if err != nil {
return err
return xerrors.Errorf("running cron: %w", err)
}

if em != nil {
if err := em.MessageApplied(ctx, ts, cronMsg.Cid(), cronMsg, ret, true); err != nil {
return xerrors.Errorf("callback failed on cron message: %w", err)
}
}
if ret.ExitCode != 0 {
return xerrors.Errorf("CheckProofSubmissions exit was non-zero: %d", ret.ExitCode)
return xerrors.Errorf("cron exit was non-zero: %d", ret.ExitCode)
}

return nil
}

for i := parentEpoch; i < epoch; i++ {
var err error
if i > parentEpoch {
vmCron, err := makeVmWithBaseStateAndEpoch(pstate, i)
if err != nil {
return cid.Undef, cid.Undef, xerrors.Errorf("making cron vm: %w", err)
}

// run cron for null rounds if any
if err := runCron(i); err != nil {
return cid.Undef, cid.Undef, err
if err = runCron(vmCron, i); err != nil {
return cid.Undef, cid.Undef, xerrors.Errorf("running cron: %w", err)
}

pstate, err = vmi.Flush(ctx)
pstate, err = vmCron.Flush(ctx)
if err != nil {
return cid.Undef, cid.Undef, xerrors.Errorf("flushing vm: %w", err)
return cid.Undef, cid.Undef, xerrors.Errorf("flushing cron vm: %w", err)
}
}

// handle state forks
// XXX: The state tree
newState, err := sm.HandleStateForks(ctx, pstate, i, em, ts)
pstate, err = sm.HandleStateForks(ctx, pstate, i, em, ts)
if err != nil {
return cid.Undef, cid.Undef, xerrors.Errorf("error handling state forks: %w", err)
}

if pstate != newState {
vmi, err = makeVmWithBaseState(newState)
if err != nil {
return cid.Undef, cid.Undef, xerrors.Errorf("making vm: %w", err)
}
}

if err = vmi.SetBlockHeight(ctx, i+1); err != nil {
return cid.Undef, cid.Undef, xerrors.Errorf("error advancing vm an epoch: %w", err)
}

pstate = newState
}

partDone()
partDone = metrics.Timer(ctx, metrics.VMApplyMessages)

vmi, err := makeVmWithBaseStateAndEpoch(pstate, epoch)
if err != nil {
return cid.Undef, cid.Undef, xerrors.Errorf("making vm: %w", err)
}

var receipts []cbg.CBORMarshaler
processedMsgs := make(map[cid.Cid]struct{})
for _, b := range bms {
Expand Down Expand Up @@ -246,7 +240,7 @@ func (t *TipSetExecutor) ApplyBlocks(ctx context.Context, sm *stmgr.StateManager
partDone()
partDone = metrics.Timer(ctx, metrics.VMApplyCron)

if err := runCron(epoch); err != nil {
if err := runCron(vmi, epoch); err != nil {
return cid.Cid{}, cid.Cid{}, err
}

Expand Down
20 changes: 13 additions & 7 deletions chain/consensus/filcns/filecoin.go
Original file line number Diff line number Diff line change
Expand Up @@ -458,7 +458,7 @@ func (filec *FilecoinEC) checkBlockMessages(ctx context.Context, b *types.FullBl

stateroot, _, err := filec.sm.TipSetState(ctx, baseTs)
if err != nil {
return err
return xerrors.Errorf("failed to compute tipsettate for %s: %w", baseTs.Key(), err)
}

st, err := state.LoadStateTree(filec.store.ActorStore(ctx), stateroot)
Expand All @@ -475,7 +475,7 @@ func (filec *FilecoinEC) checkBlockMessages(ctx context.Context, b *types.FullBl
// Phase 1: syntactic validation, as defined in the spec
minGas := pl.OnChainMessage(msg.ChainLength())
if err := m.ValidForBlockInclusion(minGas.Total(), nv); err != nil {
return err
return xerrors.Errorf("msg %s invalid for block inclusion: %w", m.Cid(), err)
}

// ValidForBlockInclusion checks if any single message does not exceed BlockGasLimit
Expand All @@ -491,7 +491,7 @@ func (filec *FilecoinEC) checkBlockMessages(ctx context.Context, b *types.FullBl
if filec.sm.GetNetworkVersion(ctx, b.Header.Height) >= network.Version13 {
sender, err = st.LookupID(m.From)
if err != nil {
return err
return xerrors.Errorf("failed to lookup sender %s: %w", m.From, err)
}
} else {
sender = m.From
Expand Down Expand Up @@ -574,28 +574,34 @@ func (filec *FilecoinEC) checkBlockMessages(ctx context.Context, b *types.FullBl

bmroot, err := bmArr.Root()
if err != nil {
return err
return xerrors.Errorf("failed to root bls msgs: %w", err)

}

smroot, err := smArr.Root()
if err != nil {
return err
return xerrors.Errorf("failed to root secp msgs: %w", err)
}

mrcid, err := tmpstore.Put(ctx, &types.MsgMeta{
BlsMessages: bmroot,
SecpkMessages: smroot,
})
if err != nil {
return err
return xerrors.Errorf("failed to put msg meta: %w", err)
}

if b.Header.Messages != mrcid {
return fmt.Errorf("messages didnt match message root in header")
}

// Finally, flush.
return vm.Copy(ctx, tmpbs, filec.store.ChainBlockstore(), mrcid)
err = vm.Copy(ctx, tmpbs, filec.store.ChainBlockstore(), mrcid)
if err != nil {
return xerrors.Errorf("failed to flush:%w", err)
}

return nil
}

func (filec *FilecoinEC) IsEpochBeyondCurrMax(epoch abi.ChainEpoch) bool {
Expand Down
11 changes: 0 additions & 11 deletions chain/vm/vm.go
Original file line number Diff line number Diff line change
Expand Up @@ -824,17 +824,6 @@ func (vm *VM) StateTree() types.StateTree {
return vm.cstate
}

func (vm *VM) SetBlockHeight(ctx context.Context, h abi.ChainEpoch) error {
vm.blockHeight = h
ncirc, err := vm.circSupplyCalc(ctx, vm.blockHeight, vm.cstate)
if err != nil {
return err
}

vm.baseCircSupply = ncirc
return nil
}

func (vm *VM) Invoke(act *types.Actor, rt *Runtime, method abi.MethodNum, params []byte) ([]byte, aerrors.ActorError) {
ctx, span := trace.StartSpan(rt.ctx, "vm.Invoke")
defer span.End()
Expand Down

0 comments on commit c0f29ea

Please sign in to comment.