Compare commits

...
Author SHA1 Message Date
raulk 2911e76d2c Merge branch 'vyzo/feat/exec-lanes' into HEAD 2023-03-27 19:01:39 +01:00
vyzo ddebdfb37c add execution metrics to the chain node views 2023-03-27 20:56:07 +03:00
Maciej Witowski ad7c79d959 Merge branch '10504-cache-execution-traces' into feat/rpcenhancement-mirror 2023-03-24 20:36:23 +01:00
Maciej Witowski 29e9e782c9 Merge branch '10537-populate-index-after-snapshot' into feat/rpcenhancement-mirror 2023-03-24 20:32:48 +01:00
Maciej Witowski b284beaaef Merge branch 'vyzo/feat/exec-lanes' into feat/rpcenhancement-mirror 2023-03-24 20:31:09 +01:00
Fridrik Asmundsson b8137f6b4d Delete existing message index when loading from snapshot 2023-03-24 16:45:33 +00:00
Fridrik Asmundsson 48c57d394b Improve performance when populating message indax 2023-03-24 15:36:31 +00:00
vyzo 08134552a4 address review comments 2023-03-24 15:48:58 +02:00
vyzo 4b590e2102 add vm execution metrics 2023-03-24 15:16:22 +02:00
Fridrik Asmundsson 59640a8b22 Populate the index on snapshot import
Fixes: https://github.com/filecoin-project/lotus/issues/10537
2023-03-24 11:41:27 +00:00
41fce94db4 perf: eth: gas estimate set applyTsMessages false (#10546)
* have gas estimate call callInternal with applyTsMessages = false and other calls with applyTsMessages=true for gas caclulation optimization

* set applyTsMessages = true in CallWithGas call in shed

* update test with new callwithgas api optimization for eth call

* Update chain/stmgr/call.go

Co-authored-by: Łukasz Magiera <magik6k@users.noreply.github.com>

* env flag LOTUS_SKIP_APPLY_TS_MESSAGE_CALL_WITH_GAS must be 1 in order to have applyTsMessages change

* env flag LOTUS_SKIP_APPLY_TS_MESSAGE_CALL_WITH_GAS must be 1 in order to have applyTsMessages change

* make sure that even if we arent apply ts messages we grab ts messages from the particular user who is requesting gas estimation

---------

Co-authored-by: Jiaying Wang <42981373+jennijuju@users.noreply.github.com>
Co-authored-by: Łukasz Magiera <magik6k@users.noreply.github.com>
Co-authored-by: Ubuntu <ubuntu@ip-10-0-4-29.us-east-2.compute.internal>
2023-03-23 18:27:01 -04:00
Aayush Rajasekaran 4efc29aa64 Merge pull request #10513 from filecoin-project/feat/shed-chainwatch
feat: shed: incoming block-sub chainwatch tool
2023-03-23 17:50:41 -04:00
Aayush Rajasekaran 2cd3a052e3 Merge pull request #10553 from filecoin-project/asr/drop-genesis-marketfund
feat: stmgr: speed up calculation of genesis circ supply
2023-03-23 17:34:17 -04:00
Aayush bc87017ea5 feat: supply: only grab genesis msig locks for writes 2023-03-23 17:04:59 -04:00
Aayush dcdd1bc214 feat: supply: drop genesis market locked funds 2023-03-23 15:56:59 -04:00
vyzo f11a7f8940 fix incorrect deferred vm release 2023-03-23 19:51:23 +02:00
vyzo 317a87d669 add some sanity checks for execution concurrency parameters 2023-03-23 18:52:37 +02:00
vyzo 2a0660447a make token.Done idempotent 2023-03-23 17:38:36 +02:00
vyzo 2bb89d9c30 call Executor.Done where appropriate in stmgr uses 2023-03-23 17:34:59 +02:00
vyzo ee6c0f8570 only call Atoi on non empty strings 2023-03-23 17:28:08 +02:00
vyzo 7362556c02 update VM interface references to use the executor, and call Done where appropriate 2023-03-23 17:17:46 +02:00
vyzo 6550abdfcc introduce execution lanes 2023-03-23 16:53:50 +02:00
Łukasz Magiera ce17546a76 Merge pull request #10549 from filecoin-project/asr/gas-paych-special
fix: gas estimation: don't special case paych collects
2023-03-23 15:15:05 +01:00
Aayush 57d4c98d00 fix: gas estimation: don't special case paych collects 2023-03-23 09:50:28 -04:00
Fridrik Asmundsson c71680ded8 Use TipSetKey as key in cache and return copies 2023-03-23 13:14:49 +00:00
Łukasz Magiera 1022ac5e2b Merge pull request #10271 from iand/iand/fix-tracer-pid
fix: tracer: emit raw peer ids for compatibility with libp2p tracer
2023-03-23 12:41:31 +01:00
Łukasz Magiera b484d72c80 Merge pull request #10323 from mtelka/info-noadmin
fix: state: lotus-miner info should show deals info without admin permission
2023-03-23 12:39:28 +01:00
Łukasz Magiera e2ff9027f9 shed chainwatch: Appease the linter 2023-03-23 12:33:00 +01:00
Łukasz Magiera dbd7ac5315 Merge branch 'feat/new-gw-methods' 2023-03-23 11:26:58 +01:00
Łukasz Magiera 4ebc408a5a Merge remote-tracking branch 'origin/master' into feat/new-gw-methods 2023-03-23 11:26:25 +01:00
Łukasz Magiera 1152bb2e7b Merge pull request #10531 from Jorropo/bump-go-libipfs
chore: bump go-libipfs
2023-03-23 10:28:06 +01:00
vyzo 8abe0ea608 Merge pull request #10452 from filecoin-project/vyzo/feat/chain/msgindex
feat:chain: Message Index
2023-03-22 17:02:34 +02:00
Jorropo f48c6268f8 chore: all: bump go-libipfs 2023-03-22 03:26:34 +01:00
Fridrik Asmundsson 3738433abe feat: Add small cache to execution traces
This PR adds a small cache to calls to ExecutionTrace which helps
 improve performance for node operators like exchanges and block
explorers.

If items is in cache calls to this function will be 2-3x faster.

Fixes: https://github.com/filecoin-project/lotus/issues/10504
2023-03-20 15:34:09 +00:00
Łukasz Magiera 02db37bf37 feat: shed: incoming block-sub chainwatch tool 2023-03-20 10:37:48 +01:00
vyzo 883bbf8701 add sanity check in SearchForIndexedMsg call site
Check that receipt is non nil, and the message was indeed found.
2023-03-16 17:51:28 +02:00
vyzo 3710768910 add TODO for WaitForMessage to use the index 2023-03-16 17:45:16 +02:00
vyzo ef2f2b0f89 reword funny comment 2023-03-16 17:43:56 +02:00
vyzo 9c087cc52c second error variable name to avoid confusing they yushie. 2023-03-16 17:42:09 +02:00
vyzo 1e7f5c6a1e shut down the index if there is an error during head processing 2023-03-16 17:41:06 +02:00
vyzo ff22a46253 complain if head change processing is building backlog 2023-03-16 17:38:46 +02:00
vyzo 90c4763e00 fix typos 2023-03-16 17:33:12 +02:00
vyzo 8bceaadda8 chain errors in searchForIndexedMsg 2023-03-16 17:32:06 +02:00
vyzo 91fccc421a add ON CONFLICT REPLACE clause in messages 2023-03-16 17:25:20 +02:00
vyzo 1676d51484 shed: expand homedir in repo path for msgindex tools 2023-03-16 17:17:26 +02:00
vyzo 5ade40cbac increase coalesce delays for test to deflake on CI 2023-03-13 17:27:12 +02:00
vyzo 6deec4c5f4 lint 2023-03-13 17:26:20 +02:00
vyzo 06f93861bc make gen 2023-03-13 17:00:16 +02:00
vyzo cafa1eaba4 fix test for CI
test files are run individually...
2023-03-13 16:56:51 +02:00
vyzo 27dc4951eb lotus-shed tools for msgindex 2023-03-13 16:51:21 +02:00
vyzo 47646cbd69 add optimization TODO comment 2023-03-13 15:19:07 +02:00
vyzo 5113c72b3a make gen 2023-03-13 15:09:11 +02:00
vyzo 1970792445 fix bug in searchForIndexedMsg
Need to use and return the execution tipset
2023-03-13 14:56:00 +02:00
vyzo 4b1a405002 basic msgindex itest 2023-03-13 14:55:26 +02:00
vyzo 05cb2428c1 add confidence check for indexed message epoch 2023-03-13 12:33:19 +02:00
vyzo 0bf6a333b9 more gen 2023-03-13 12:24:35 +02:00
vyzo 5e011d536b enhance comment about lookup cid semantics 2023-03-13 12:14:30 +02:00
vyzo 0a5618d406 make gen 2023-03-13 12:10:36 +02:00
vyzo db8b593c9b make MsgIndex configurable, disabled by default. 2023-03-13 12:01:04 +02:00
vyzo 3c945e9e3c no need to return error from DummyMsgIndex DI constructor 2023-03-13 12:00:48 +02:00
vyzo a11032b10a adjust coalescer delays 2023-03-13 11:44:10 +02:00
vyzo bda7ef52da log rollback errors 2023-03-13 11:41:46 +02:00
vyzo b90cfff0aa wire in lifecycle context 2023-03-13 10:42:48 +02:00
vyzo 5461548b7e fix comment typo 2023-03-13 05:57:26 +02:00
vyzo 4a20c9b60f cosmetics 2023-03-13 05:54:32 +02:00
vyzo e7448b1bb5 fix tests 2023-03-13 05:51:33 +02:00
vyzo df6dfdf8a9 refactor database
- drop the execution index; we don't need it
- it is inclusion tipset
- use MessagesForTipset
- hoist db sql stuffs on top for clarity
- add index for tipset on messages
2023-03-13 05:51:24 +02:00
vyzo 88d7a4e610 more lint 2023-03-12 16:13:05 +02:00
vyzo 3b765a30d3 dummy index for itests 2023-03-12 16:02:51 +02:00
vyzo 1415391988 lint
lint

lint
2023-03-12 15:50:10 +02:00
vyzo 0077fa2a98 make gen 2023-03-12 15:30:05 +02:00
vyzo 171734ea31 hook the index into the rest of lotus 2023-03-12 15:25:07 +02:00
vyzo 0d274df977 use the transaction Luke! 2023-03-12 13:35:50 +02:00
vyzo d9ca214309 test test test 2023-03-12 13:21:03 +02:00
vyzo bf9ae23c98 implement Close 2023-03-11 22:09:31 +02:00
vyzo b5dd4e31ac implement reconciliation 2023-03-11 21:26:11 +02:00
vyzo d97c6b2f69 more implementation 2023-03-11 19:11:08 +02:00
vyzo 7fcf228bc4 better logger name 2023-03-11 18:23:05 +02:00
vyzo 3649ae373b implementation details 2023-03-11 18:21:16 +02:00
vyzo 9144f9cea7 abstract ChainStore interface 2023-03-11 17:56:23 +02:00
vyzo eab728c4e7 we don't need the block 2023-03-11 17:25:57 +02:00
vyzo cc83a7cd35 use tipset cid instead of key for posterity. 2023-03-11 17:12:28 +02:00
vyzo a0e95ebe97 scaffolding 2023-03-11 16:50:29 +02:00
Ales Dumikau 6e9a1f44ed Add new methods to gw 2023-03-11 10:27:42 +03:00
Ales Dumikau 8a3f91e655 Merge branch 'master' of github.com:protofire/lotus 2023-03-10 10:14:18 +03:00
dumikau c099317bb3 Merge pull request #1 from filecoin-project/master
Fastforward main branch
2023-03-09 14:46:03 +03:00
Ian Davis 5d3314f121 Touch source to trigger CI 2023-02-23 13:02:20 +00:00
Marcel Telka 7f33db90d8 lotus-miner info should show deals info without admin permission 2023-02-21 11:10:38 +01:00
Ian Davis 35730637dc fix: tracer: emit raw peer ids for compatibility with libp2p tracer 2023-02-14 16:17:03 +00:00
59 changed files with 2378 additions and 113 deletions
+6
View File
@@ -818,6 +818,12 @@ workflows:
- build
suite: itest-mpool_push_with_uuid
target: "./itests/mpool_push_with_uuid_test.go"
- test:
name: test-itest-msgindex
requires:
- build
suite: itest-msgindex
target: "./itests/msgindex_test.go"
- test:
name: test-itest-multisig
requires:
+3
View File
@@ -33,6 +33,9 @@ import (
// * Generate openrpc blobs
type Gateway interface {
StateMinerSectorCount(context.Context, address.Address, types.TipSetKey) (MinerSectors, error)
GasEstimateGasPremium(context.Context, uint64, address.Address, int64, types.TipSetKey) (types.BigInt, error)
StateReplay(context.Context, types.TipSetKey, cid.Cid) (*InvocResult, error)
ChainHasObj(context.Context, cid.Cid) (bool, error)
ChainPutObj(context.Context, blocks.Block) error
ChainHead(ctx context.Context) (*types.TipSet, error)
+39
View File
@@ -726,6 +726,8 @@ type GatewayMethods struct {
EthUnsubscribe func(p0 context.Context, p1 ethtypes.EthSubscriptionID) (bool, error) ``
GasEstimateGasPremium func(p0 context.Context, p1 uint64, p2 address.Address, p3 int64, p4 types.TipSetKey) (types.BigInt, error) ``
GasEstimateMessageGas func(p0 context.Context, p1 *types.Message, p2 *MessageSendSpec, p3 types.TipSetKey) (*types.Message, error) ``
MpoolGetNonce func(p0 context.Context, p1 address.Address) (uint64, error) ``
@@ -768,12 +770,16 @@ type GatewayMethods struct {
StateMinerProvingDeadline func(p0 context.Context, p1 address.Address, p2 types.TipSetKey) (*dline.Info, error) ``
StateMinerSectorCount func(p0 context.Context, p1 address.Address, p2 types.TipSetKey) (MinerSectors, error) ``
StateNetworkName func(p0 context.Context) (dtypes.NetworkName, error) ``
StateNetworkVersion func(p0 context.Context, p1 types.TipSetKey) (apitypes.NetworkVersion, error) ``
StateReadState func(p0 context.Context, p1 address.Address, p2 types.TipSetKey) (*ActorState, error) ``
StateReplay func(p0 context.Context, p1 types.TipSetKey, p2 cid.Cid) (*InvocResult, error) ``
StateSearchMsg func(p0 context.Context, p1 types.TipSetKey, p2 cid.Cid, p3 abi.ChainEpoch, p4 bool) (*MsgLookup, error) ``
StateSectorGetInfo func(p0 context.Context, p1 address.Address, p2 abi.SectorNumber, p3 types.TipSetKey) (*miner.SectorOnChainInfo, error) ``
@@ -4614,6 +4620,17 @@ func (s *GatewayStub) EthUnsubscribe(p0 context.Context, p1 ethtypes.EthSubscrip
return false, ErrNotSupported
}
func (s *GatewayStruct) GasEstimateGasPremium(p0 context.Context, p1 uint64, p2 address.Address, p3 int64, p4 types.TipSetKey) (types.BigInt, error) {
if s.Internal.GasEstimateGasPremium == nil {
return *new(types.BigInt), ErrNotSupported
}
return s.Internal.GasEstimateGasPremium(p0, p1, p2, p3, p4)
}
func (s *GatewayStub) GasEstimateGasPremium(p0 context.Context, p1 uint64, p2 address.Address, p3 int64, p4 types.TipSetKey) (types.BigInt, error) {
return *new(types.BigInt), ErrNotSupported
}
func (s *GatewayStruct) GasEstimateMessageGas(p0 context.Context, p1 *types.Message, p2 *MessageSendSpec, p3 types.TipSetKey) (*types.Message, error) {
if s.Internal.GasEstimateMessageGas == nil {
return nil, ErrNotSupported
@@ -4845,6 +4862,17 @@ func (s *GatewayStub) StateMinerProvingDeadline(p0 context.Context, p1 address.A
return nil, ErrNotSupported
}
func (s *GatewayStruct) StateMinerSectorCount(p0 context.Context, p1 address.Address, p2 types.TipSetKey) (MinerSectors, error) {
if s.Internal.StateMinerSectorCount == nil {
return *new(MinerSectors), ErrNotSupported
}
return s.Internal.StateMinerSectorCount(p0, p1, p2)
}
func (s *GatewayStub) StateMinerSectorCount(p0 context.Context, p1 address.Address, p2 types.TipSetKey) (MinerSectors, error) {
return *new(MinerSectors), ErrNotSupported
}
func (s *GatewayStruct) StateNetworkName(p0 context.Context) (dtypes.NetworkName, error) {
if s.Internal.StateNetworkName == nil {
return *new(dtypes.NetworkName), ErrNotSupported
@@ -4878,6 +4906,17 @@ func (s *GatewayStub) StateReadState(p0 context.Context, p1 address.Address, p2
return nil, ErrNotSupported
}
func (s *GatewayStruct) StateReplay(p0 context.Context, p1 types.TipSetKey, p2 cid.Cid) (*InvocResult, error) {
if s.Internal.StateReplay == nil {
return nil, ErrNotSupported
}
return s.Internal.StateReplay(p0, p1, p2)
}
func (s *GatewayStub) StateReplay(p0 context.Context, p1 types.TipSetKey, p2 cid.Cid) (*InvocResult, error) {
return nil, ErrNotSupported
}
func (s *GatewayStruct) StateSearchMsg(p0 context.Context, p1 types.TipSetKey, p2 cid.Cid, p3 abi.ChainEpoch, p4 bool) (*MsgLookup, error) {
if s.Internal.StateSearchMsg == nil {
return nil, ErrNotSupported
+3
View File
@@ -35,6 +35,9 @@ import (
// * Generate openrpc blobs
type Gateway interface {
StateMinerSectorCount(context.Context, address.Address, types.TipSetKey) (api.MinerSectors, error)
GasEstimateGasPremium(context.Context, uint64, address.Address, int64, types.TipSetKey) (types.BigInt, error)
StateReplay(context.Context, types.TipSetKey, cid.Cid) (*api.InvocResult, error)
ChainHasObj(context.Context, cid.Cid) (bool, error)
ChainPutObj(context.Context, blocks.Block) error
ChainHead(ctx context.Context) (*types.TipSet, error)
+39
View File
@@ -449,6 +449,8 @@ type GatewayMethods struct {
ChainReadObj func(p0 context.Context, p1 cid.Cid) ([]byte, error) ``
GasEstimateGasPremium func(p0 context.Context, p1 uint64, p2 address.Address, p3 int64, p4 types.TipSetKey) (types.BigInt, error) ``
GasEstimateMessageGas func(p0 context.Context, p1 *types.Message, p2 *api.MessageSendSpec, p3 types.TipSetKey) (*types.Message, error) ``
MpoolGetNonce func(p0 context.Context, p1 address.Address) (uint64, error) ``
@@ -487,10 +489,14 @@ type GatewayMethods struct {
StateMinerProvingDeadline func(p0 context.Context, p1 address.Address, p2 types.TipSetKey) (*dline.Info, error) ``
StateMinerSectorCount func(p0 context.Context, p1 address.Address, p2 types.TipSetKey) (api.MinerSectors, error) ``
StateNetworkName func(p0 context.Context) (dtypes.NetworkName, error) ``
StateNetworkVersion func(p0 context.Context, p1 types.TipSetKey) (abinetwork.Version, error) ``
StateReplay func(p0 context.Context, p1 types.TipSetKey, p2 cid.Cid) (*api.InvocResult, error) ``
StateSearchMsg func(p0 context.Context, p1 cid.Cid) (*api.MsgLookup, error) ``
StateSectorGetInfo func(p0 context.Context, p1 address.Address, p2 abi.SectorNumber, p3 types.TipSetKey) (*miner.SectorOnChainInfo, error) ``
@@ -2674,6 +2680,17 @@ func (s *GatewayStub) ChainReadObj(p0 context.Context, p1 cid.Cid) ([]byte, erro
return *new([]byte), ErrNotSupported
}
func (s *GatewayStruct) GasEstimateGasPremium(p0 context.Context, p1 uint64, p2 address.Address, p3 int64, p4 types.TipSetKey) (types.BigInt, error) {
if s.Internal.GasEstimateGasPremium == nil {
return *new(types.BigInt), ErrNotSupported
}
return s.Internal.GasEstimateGasPremium(p0, p1, p2, p3, p4)
}
func (s *GatewayStub) GasEstimateGasPremium(p0 context.Context, p1 uint64, p2 address.Address, p3 int64, p4 types.TipSetKey) (types.BigInt, error) {
return *new(types.BigInt), ErrNotSupported
}
func (s *GatewayStruct) GasEstimateMessageGas(p0 context.Context, p1 *types.Message, p2 *api.MessageSendSpec, p3 types.TipSetKey) (*types.Message, error) {
if s.Internal.GasEstimateMessageGas == nil {
return nil, ErrNotSupported
@@ -2883,6 +2900,17 @@ func (s *GatewayStub) StateMinerProvingDeadline(p0 context.Context, p1 address.A
return nil, ErrNotSupported
}
func (s *GatewayStruct) StateMinerSectorCount(p0 context.Context, p1 address.Address, p2 types.TipSetKey) (api.MinerSectors, error) {
if s.Internal.StateMinerSectorCount == nil {
return *new(api.MinerSectors), ErrNotSupported
}
return s.Internal.StateMinerSectorCount(p0, p1, p2)
}
func (s *GatewayStub) StateMinerSectorCount(p0 context.Context, p1 address.Address, p2 types.TipSetKey) (api.MinerSectors, error) {
return *new(api.MinerSectors), ErrNotSupported
}
func (s *GatewayStruct) StateNetworkName(p0 context.Context) (dtypes.NetworkName, error) {
if s.Internal.StateNetworkName == nil {
return *new(dtypes.NetworkName), ErrNotSupported
@@ -2905,6 +2933,17 @@ func (s *GatewayStub) StateNetworkVersion(p0 context.Context, p1 types.TipSetKey
return *new(abinetwork.Version), ErrNotSupported
}
func (s *GatewayStruct) StateReplay(p0 context.Context, p1 types.TipSetKey, p2 cid.Cid) (*api.InvocResult, error) {
if s.Internal.StateReplay == nil {
return nil, ErrNotSupported
}
return s.Internal.StateReplay(p0, p1, p2)
}
func (s *GatewayStub) StateReplay(p0 context.Context, p1 types.TipSetKey, p2 cid.Cid) (*api.InvocResult, error) {
return nil, ErrNotSupported
}
func (s *GatewayStruct) StateSearchMsg(p0 context.Context, p1 cid.Cid) (*api.MsgLookup, error) {
if s.Internal.StateSearchMsg == nil {
return nil, ErrNotSupported
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+9 -2
View File
@@ -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.Interface, error) {
makeVm := func(base cid.Cid, e abi.ChainEpoch, timestamp uint64) (vm.Executor, error) {
vmopt := &vm.VMOpts{
StateBase: base,
Epoch: e,
@@ -109,6 +109,7 @@ func (t *TipSetExecutor) ApplyBlocks(ctx context.Context,
TipSetGetter: stmgr.TipSetGetterForTipset(sm.ChainStore(), ts),
Tracing: vmTracing,
ReturnEvents: sm.ChainStore().IsStoringEvents(),
ExecutionLane: vm.ExecutionLanePriority,
}
return sm.VMConstructor()(ctx, vmopt)
@@ -116,7 +117,7 @@ func (t *TipSetExecutor) ApplyBlocks(ctx context.Context,
var cronGas int64
runCron := func(vmCron vm.Interface, epoch abi.ChainEpoch) error {
runCron := func(vmCron vm.Executor, epoch abi.ChainEpoch) error {
cronMsg := &types.Message{
To: cron.Address,
From: builtin.SystemActorAddr,
@@ -169,13 +170,17 @@ 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
@@ -191,10 +196,12 @@ func (t *TipSetExecutor) ApplyBlocks(ctx context.Context,
cronGas = 0
partDone = metrics.Timer(ctx, metrics.VMApplyMessages)
// TODO reorg the code to minimize the execution critical section
vmi, err := makeVm(pstate, epoch, ts.MinTimestamp())
if err != nil {
return cid.Undef, cid.Undef, xerrors.Errorf("making vm: %w", err)
}
defer vmi.Done()
var (
receipts []*types.MessageReceipt
+2 -1
View File
@@ -34,6 +34,7 @@ import (
"github.com/filecoin-project/lotus/chain/consensus"
"github.com/filecoin-project/lotus/chain/consensus/filcns"
genesis2 "github.com/filecoin-project/lotus/chain/gen/genesis"
"github.com/filecoin-project/lotus/chain/index"
"github.com/filecoin-project/lotus/chain/rand"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/store"
@@ -256,7 +257,7 @@ func NewGeneratorWithSectorsAndUpgradeSchedule(numSectors int, us stmgr.UpgradeS
//return nil, xerrors.Errorf("creating drand beacon: %w", err)
//}
sm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), sys, us, beac, ds)
sm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), sys, us, beac, ds, index.DummyMsgIndex)
if err != nil {
return nil, xerrors.Errorf("initing stmgr: %w", err)
}
+1
View File
@@ -496,6 +496,7 @@ 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 {
+6 -1
View File
@@ -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.Interface, error) {
newVM := func(base cid.Cid) (vm.Executor, error) {
vmopt := &vm.VMOpts{
StateBase: base,
Epoch: 0,
@@ -108,6 +108,8 @@ 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")
@@ -338,6 +340,7 @@ 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)
@@ -410,6 +413,7 @@ 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)
@@ -517,6 +521,7 @@ 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)
+45
View File
@@ -0,0 +1,45 @@
package index
import (
"context"
"errors"
"github.com/ipfs/go-cid"
"github.com/filecoin-project/go-state-types/abi"
)
var ErrNotFound = errors.New("message not found")
var ErrClosed = errors.New("index closed")
// MsgInfo is the Message metadata the index tracks.
type MsgInfo struct {
// the message this record refers to
Message cid.Cid
// the tipset where this message was included
TipSet cid.Cid
// the epoch where this message was included
Epoch abi.ChainEpoch
}
// MsgIndex is the interface to the message index
type MsgIndex interface {
// GetMsgInfo retrieves the message metadata through the index.
// The lookup is done using the onchain message Cid; that is the signed message Cid
// for SECP messages and unsigned message Cid for BLS messages.
GetMsgInfo(ctx context.Context, m cid.Cid) (MsgInfo, error)
// Close closes the index
Close() error
}
type dummyMsgIndex struct{}
func (dummyMsgIndex) GetMsgInfo(ctx context.Context, m cid.Cid) (MsgInfo, error) {
return MsgInfo{}, ErrNotFound
}
func (dummyMsgIndex) Close() error {
return nil
}
var DummyMsgIndex MsgIndex = dummyMsgIndex{}
+549
View File
@@ -0,0 +1,549 @@
package index
import (
"context"
"database/sql"
"errors"
"io/fs"
"os"
"path"
"sync"
"time"
"github.com/ipfs/go-cid"
logging "github.com/ipfs/go-log/v2"
_ "github.com/mattn/go-sqlite3"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
)
var log = logging.Logger("msgindex")
var dbName = "msgindex.db"
var dbDefs = []string{
`CREATE TABLE IF NOT EXISTS messages (
cid VARCHAR(80) PRIMARY KEY ON CONFLICT REPLACE,
tipset_cid VARCHAR(80) NOT NULL,
epoch INTEGER NOT NULL
)`,
`CREATE INDEX IF NOT EXISTS tipset_cids ON messages (tipset_cid)
`,
`CREATE TABLE IF NOT EXISTS _meta (
version UINT64 NOT NULL UNIQUE
)`,
`INSERT OR IGNORE INTO _meta (version) VALUES (1)`,
}
var dbPragmas = []string{}
const (
// prepared stmts
dbqGetMessageInfo = "SELECT tipset_cid, epoch FROM messages WHERE cid = ?"
dbqInsertMessage = "INSERT INTO messages VALUES (?, ?, ?)"
dbqDeleteTipsetMessages = "DELETE FROM messages WHERE tipset_cid = ?"
// reconciliation
dbqCountMessages = "SELECT COUNT(*) FROM messages"
dbqMinEpoch = "SELECT MIN(epoch) FROM messages"
dbqCountTipsetMessages = "SELECT COUNT(*) FROM messages WHERE tipset_cid = ?"
dbqDeleteMessagesByEpoch = "DELETE FROM messages WHERE epoch >= ?"
)
// coalescer configuration (TODO: use observer instead)
// these are exposed to make tests snappy
var (
CoalesceMinDelay = time.Second
CoalesceMaxDelay = 15 * time.Second
CoalesceMergeInterval = time.Second
)
// chain store interface; we could use store.ChainStore directly,
// but this simplifies unit testing.
type ChainStore interface {
SubscribeHeadChanges(f store.ReorgNotifee)
MessagesForTipset(ctx context.Context, ts *types.TipSet) ([]types.ChainMsg, error)
GetHeaviestTipSet() *types.TipSet
GetTipSetFromKey(ctx context.Context, tsk types.TipSetKey) (*types.TipSet, error)
}
var _ ChainStore = (*store.ChainStore)(nil)
type msgIndex struct {
cs ChainStore
db *sql.DB
selectMsgStmt *sql.Stmt
insertMsgStmt *sql.Stmt
deleteTipSetStmt *sql.Stmt
sema chan struct{}
mx sync.Mutex
pend []headChange
cancel func()
workers sync.WaitGroup
closeLk sync.RWMutex
closed bool
}
var _ MsgIndex = (*msgIndex)(nil)
type headChange struct {
rev []*types.TipSet
app []*types.TipSet
}
func NewMsgIndex(lctx context.Context, basePath string, cs ChainStore) (MsgIndex, error) {
var (
dbPath string
exists bool
err error
)
err = os.MkdirAll(basePath, 0755)
if err != nil {
return nil, xerrors.Errorf("error creating msgindex base directory: %w", err)
}
dbPath = path.Join(basePath, dbName)
_, err = os.Stat(dbPath)
switch {
case err == nil:
exists = true
case errors.Is(err, fs.ErrNotExist):
case err != nil:
return nil, xerrors.Errorf("error stating msgindex database: %w", err)
}
db, err := sql.Open("sqlite3", dbPath)
if err != nil {
// TODO [nice to have]: automaticaly delete corrupt databases
// but for now we can just error and let the operator delete.
return nil, xerrors.Errorf("error opening msgindex database: %w", err)
}
if err := prepareDB(db); err != nil {
return nil, xerrors.Errorf("error creating msgindex database: %w", err)
}
// TODO we may consider populating the index when first creating the db
if exists {
if err := reconcileIndex(db, cs); err != nil {
return nil, xerrors.Errorf("error reconciling msgindex database: %w", err)
}
}
ctx, cancel := context.WithCancel(lctx)
msgIndex := &msgIndex{
db: db,
cs: cs,
sema: make(chan struct{}, 1),
cancel: cancel,
}
err = msgIndex.prepareStatements()
if err != nil {
if err := db.Close(); err != nil {
log.Errorf("error closing msgindex database: %s", err)
}
return nil, xerrors.Errorf("error preparing msgindex database statements: %w", err)
}
rnf := store.WrapHeadChangeCoalescer(
msgIndex.onHeadChange,
CoalesceMinDelay,
CoalesceMaxDelay,
CoalesceMergeInterval,
)
cs.SubscribeHeadChanges(rnf)
msgIndex.workers.Add(1)
go msgIndex.background(ctx)
return msgIndex, nil
}
func PopulateAfterSnapshot(lctx context.Context, basePath string, cs ChainStore) error {
err := os.MkdirAll(basePath, 0755)
if err != nil {
return xerrors.Errorf("error creating msgindex base directory: %w", err)
}
dbPath := path.Join(basePath, dbName)
// if a database already exists, we try to delete it and create a new one
if _, err := os.Stat(dbPath); err == nil {
if err = os.Remove(dbPath); err != nil {
return xerrors.Errorf("msgindex already exists at %s and can't be deleted", dbPath)
}
}
db, err := sql.Open("sqlite3", dbPath)
if err != nil {
return xerrors.Errorf("error opening msgindex database: %w", err)
}
defer db.Close()
if err := prepareDB(db); err != nil {
return xerrors.Errorf("error creating msgindex database: %w", err)
}
tx, err := db.Begin()
if err != nil {
return xerrors.Errorf("error when starting transaction: %w", err)
}
rollback := func() {
if err := tx.Rollback(); err != nil {
log.Errorf("error in rollback: %s", err)
}
}
insertStmt, err := tx.Prepare(dbqInsertMessage)
if err != nil {
return xerrors.Errorf("error preparing insertMsgStmt: %w", err)
}
defer insertStmt.Close()
curTs := cs.GetHeaviestTipSet()
startHeight := curTs.Height()
for curTs != nil {
tscid, err := curTs.Key().Cid()
if err != nil {
rollback()
return xerrors.Errorf("error computing tipset cid: %w", err)
}
tskey := tscid.String()
epoch := int64(curTs.Height())
msgs, err := cs.MessagesForTipset(lctx, curTs)
if err != nil {
log.Infof("stopping import after %d tipsets", startHeight-curTs.Height())
break
}
for _, msg := range msgs {
key := msg.Cid().String()
if _, err := insertStmt.Exec(key, tskey, epoch); err != nil {
rollback()
return xerrors.Errorf("error inserting message: %w", err)
}
}
curTs, err = cs.GetTipSetFromKey(lctx, curTs.Parents())
if err != nil {
rollback()
return xerrors.Errorf("error walking chain: %w", err)
}
}
err = tx.Commit()
if err != nil {
return xerrors.Errorf("error commiting transaction: %w", err)
}
return nil
}
// init utilities
func prepareDB(db *sql.DB) error {
for _, stmt := range dbDefs {
if _, err := db.Exec(stmt); err != nil {
return xerrors.Errorf("error executing sql statement '%s': %w", stmt, err)
}
}
for _, stmt := range dbPragmas {
if _, err := db.Exec(stmt); err != nil {
return xerrors.Errorf("error executing sql statement '%s': %w", stmt, err)
}
}
return nil
}
func reconcileIndex(db *sql.DB, cs ChainStore) error {
// Invariant: after reconciliation, every tipset in the index is in the current chain; ie either
// the chain head or reachable by walking the chain.
// Algorithm:
// 1. Count messages in index; if none, trivially reconciled.
// TODO we may consider populating the index in that case
// 2. Find the minimum tipset in the index; this will mark the end of the reconciliation walk
// 3. Walk from current tipset until we find a tipset in the index.
// 4. Delete (revert!) all tipsets above the found tipset.
// 5. If the walk ends in the boundary epoch, then delete everything.
//
row := db.QueryRow(dbqCountMessages)
var result int64
if err := row.Scan(&result); err != nil {
return xerrors.Errorf("error counting messages: %w", err)
}
if result == 0 {
return nil
}
row = db.QueryRow(dbqMinEpoch)
if err := row.Scan(&result); err != nil {
return xerrors.Errorf("error finding boundary epoch: %w", err)
}
boundaryEpoch := abi.ChainEpoch(result)
countMsgsStmt, err := db.Prepare(dbqCountTipsetMessages)
if err != nil {
return xerrors.Errorf("error preparing statement: %w", err)
}
curTs := cs.GetHeaviestTipSet()
for curTs != nil && curTs.Height() >= boundaryEpoch {
tsCid, err := curTs.Key().Cid()
if err != nil {
return xerrors.Errorf("error computing tipset cid: %w", err)
}
key := tsCid.String()
row = countMsgsStmt.QueryRow(key)
if err := row.Scan(&result); err != nil {
return xerrors.Errorf("error counting messages: %w", err)
}
if result > 0 {
// found it!
boundaryEpoch = curTs.Height() + 1
break
}
// walk up
parents := curTs.Parents()
curTs, err = cs.GetTipSetFromKey(context.TODO(), parents)
if err != nil {
return xerrors.Errorf("error walking chain: %w", err)
}
}
// delete everything above the minEpoch
if _, err = db.Exec(dbqDeleteMessagesByEpoch, int64(boundaryEpoch)); err != nil {
return xerrors.Errorf("error deleting stale reorged out message: %w", err)
}
return nil
}
func (x *msgIndex) prepareStatements() error {
stmt, err := x.db.Prepare(dbqGetMessageInfo)
if err != nil {
return xerrors.Errorf("prepare selectMsgStmt: %w", err)
}
x.selectMsgStmt = stmt
stmt, err = x.db.Prepare(dbqInsertMessage)
if err != nil {
return xerrors.Errorf("prepare insertMsgStmt: %w", err)
}
x.insertMsgStmt = stmt
stmt, err = x.db.Prepare(dbqDeleteTipsetMessages)
if err != nil {
return xerrors.Errorf("prepare deleteTipSetStmt: %w", err)
}
x.deleteTipSetStmt = stmt
return nil
}
// head change notifee
func (x *msgIndex) onHeadChange(rev, app []*types.TipSet) error {
x.closeLk.RLock()
defer x.closeLk.RUnlock()
if x.closed {
return nil
}
// do it in the background to avoid blocking head change processing
x.mx.Lock()
x.pend = append(x.pend, headChange{rev: rev, app: app})
pendLen := len(x.pend)
x.mx.Unlock()
// complain loudly if this is building backlog
if pendLen > 10 {
log.Warnf("message index head change processing is building backlog: %d pending head changes", pendLen)
}
select {
case x.sema <- struct{}{}:
default:
}
return nil
}
func (x *msgIndex) background(ctx context.Context) {
defer x.workers.Done()
for {
select {
case <-x.sema:
err := x.processHeadChanges(ctx)
if err != nil {
// we can't rely on an inconsistent index, so shut it down.
log.Errorf("error processing head change notifications: %s; shutting down message index", err)
if err2 := x.Close(); err2 != nil {
log.Errorf("error shutting down index: %s", err2)
}
}
case <-ctx.Done():
return
}
}
}
func (x *msgIndex) processHeadChanges(ctx context.Context) error {
x.mx.Lock()
pend := x.pend
x.pend = nil
x.mx.Unlock()
tx, err := x.db.Begin()
if err != nil {
return xerrors.Errorf("error creating transaction: %w", err)
}
for _, hc := range pend {
for _, ts := range hc.rev {
if err := x.doRevert(ctx, tx, ts); err != nil {
if err2 := tx.Rollback(); err2 != nil {
log.Errorf("error rolling back transaction: %s", err2)
}
return xerrors.Errorf("error reverting %s: %w", ts, err)
}
}
for _, ts := range hc.app {
if err := x.doApply(ctx, tx, ts); err != nil {
if err2 := tx.Rollback(); err2 != nil {
log.Errorf("error rolling back transaction: %s", err2)
}
return xerrors.Errorf("error applying %s: %w", ts, err)
}
}
}
return tx.Commit()
}
func (x *msgIndex) doRevert(ctx context.Context, tx *sql.Tx, ts *types.TipSet) error {
tskey, err := ts.Key().Cid()
if err != nil {
return xerrors.Errorf("error computing tipset cid: %w", err)
}
key := tskey.String()
_, err = tx.Stmt(x.deleteTipSetStmt).Exec(key)
return err
}
func (x *msgIndex) doApply(ctx context.Context, tx *sql.Tx, ts *types.TipSet) error {
tscid, err := ts.Key().Cid()
if err != nil {
return xerrors.Errorf("error computing tipset cid: %w", err)
}
tskey := tscid.String()
epoch := int64(ts.Height())
msgs, err := x.cs.MessagesForTipset(ctx, ts)
if err != nil {
return xerrors.Errorf("error retrieving messages for tipset %s: %w", ts, err)
}
insertStmt := tx.Stmt(x.insertMsgStmt)
for _, msg := range msgs {
key := msg.Cid().String()
if _, err := insertStmt.Exec(key, tskey, epoch); err != nil {
return xerrors.Errorf("error inserting message: %w", err)
}
}
return nil
}
// interface
func (x *msgIndex) GetMsgInfo(ctx context.Context, m cid.Cid) (MsgInfo, error) {
x.closeLk.RLock()
defer x.closeLk.RUnlock()
if x.closed {
return MsgInfo{}, ErrClosed
}
var (
tipset string
epoch int64
)
key := m.String()
row := x.selectMsgStmt.QueryRow(key)
err := row.Scan(&tipset, &epoch)
switch {
case err == sql.ErrNoRows:
return MsgInfo{}, ErrNotFound
case err != nil:
return MsgInfo{}, xerrors.Errorf("error querying msgindex database: %w", err)
}
tipsetCid, err := cid.Decode(tipset)
if err != nil {
return MsgInfo{}, xerrors.Errorf("error decoding tipset cid: %w", err)
}
return MsgInfo{
Message: m,
TipSet: tipsetCid,
Epoch: abi.ChainEpoch(epoch),
}, nil
}
func (x *msgIndex) Close() error {
x.closeLk.Lock()
defer x.closeLk.Unlock()
if x.closed {
return nil
}
x.closed = true
x.cancel()
x.workers.Wait()
return x.db.Close()
}
// informal apis for itests; not exposed in the main interface
func (x *msgIndex) CountMessages() (int64, error) {
x.closeLk.RLock()
defer x.closeLk.RUnlock()
if x.closed {
return 0, ErrClosed
}
var result int64
row := x.db.QueryRow(dbqCountMessages)
err := row.Scan(&result)
return result, err
}
+298
View File
@@ -0,0 +1,298 @@
package index
import (
"context"
"errors"
"math/rand"
"os"
"testing"
"time"
"github.com/ipfs/go-cid"
"github.com/ipfs/go-libipfs/blocks"
"github.com/stretchr/testify/require"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/chain/types/mock"
)
func TestBasicMsgIndex(t *testing.T) {
// the most basic of tests:
// 1. Create an index with mock chain store
// 2. Advance the chain for a few tipsets
// 3. Verify that the index contains all messages with the correct tipset/epoch
cs := newMockChainStore()
cs.genesis()
tmp := t.TempDir()
t.Cleanup(func() { _ = os.RemoveAll(tmp) })
msgIndex, err := NewMsgIndex(context.Background(), tmp, cs)
require.NoError(t, err)
defer msgIndex.Close() //nolint
for i := 0; i < 10; i++ {
t.Logf("advance to epoch %d", i+1)
err := cs.advance()
require.NoError(t, err)
// wait for the coalescer to notify
time.Sleep(CoalesceMinDelay + 10*time.Millisecond)
}
t.Log("verifying index")
verifyIndex(t, cs, msgIndex)
}
func TestReorgMsgIndex(t *testing.T) {
// slightly more nuanced test that includes reorgs
// 1. Create an index with mock chain store
// 2. Advance/Reorg the chain for a few tipsets
// 3. Verify that the index contains all messages with the correct tipst/epoch
cs := newMockChainStore()
cs.genesis()
tmp := t.TempDir()
t.Cleanup(func() { _ = os.RemoveAll(tmp) })
msgIndex, err := NewMsgIndex(context.Background(), tmp, cs)
require.NoError(t, err)
defer msgIndex.Close() //nolint
for i := 0; i < 10; i++ {
t.Logf("advance to epoch %d", i+1)
err := cs.advance()
require.NoError(t, err)
// wait for the coalescer to notify
time.Sleep(CoalesceMinDelay + 10*time.Millisecond)
}
// a simple reorg
t.Log("doing reorg")
reorgme := cs.curTs
reorgmeParent, err := cs.GetTipSetFromKey(context.Background(), reorgme.Parents())
require.NoError(t, err)
cs.setHead(reorgmeParent)
reorgmeChild := cs.makeBlk()
err = cs.reorg([]*types.TipSet{reorgme}, []*types.TipSet{reorgmeChild})
require.NoError(t, err)
time.Sleep(CoalesceMinDelay + 10*time.Millisecond)
t.Log("verifying index")
verifyIndex(t, cs, msgIndex)
t.Log("verifying that reorged messages are not present")
verifyMissing(t, cs, msgIndex, reorgme)
}
func TestReconcileMsgIndex(t *testing.T) {
// test that exercises the reconciliation code paths
// 1. Create and populate a basic msgindex, similar to TestBasicMsgIndex.
// 2. Close it
// 3. Reorg the mock chain store
// 4. Reopen the index to trigger reconciliation
// 5. Enxure that only the stable messages remain.
cs := newMockChainStore()
cs.genesis()
tmp := t.TempDir()
t.Cleanup(func() { _ = os.RemoveAll(tmp) })
msgIndex, err := NewMsgIndex(context.Background(), tmp, cs)
require.NoError(t, err)
for i := 0; i < 10; i++ {
t.Logf("advance to epoch %d", i+1)
err := cs.advance()
require.NoError(t, err)
// wait for the coalescer to notify
time.Sleep(CoalesceMinDelay + 10*time.Millisecond)
}
// Close it and reorg
err = msgIndex.Close()
require.NoError(t, err)
cs.notify = nil
// a simple reorg
t.Log("doing reorg")
reorgme := cs.curTs
reorgmeParent, err := cs.GetTipSetFromKey(context.Background(), reorgme.Parents())
require.NoError(t, err)
cs.setHead(reorgmeParent)
reorgmeChild := cs.makeBlk()
err = cs.reorg([]*types.TipSet{reorgme}, []*types.TipSet{reorgmeChild})
require.NoError(t, err)
// reopen to reconcile
msgIndex, err = NewMsgIndex(context.Background(), tmp, cs)
require.NoError(t, err)
defer msgIndex.Close() //nolint
t.Log("verifying index")
// need to step one up because the last tipset is not known by the index
cs.setHead(reorgmeParent)
verifyIndex(t, cs, msgIndex)
t.Log("verifying that reorged and unknown messages are not present")
verifyMissing(t, cs, msgIndex, reorgme, reorgmeChild)
}
func verifyIndex(t *testing.T, cs *mockChainStore, msgIndex MsgIndex) {
for ts := cs.curTs; ts.Height() > 0; {
t.Logf("verify at height %d", ts.Height())
blks := ts.Blocks()
if len(blks) == 0 {
break
}
tsCid, err := ts.Key().Cid()
require.NoError(t, err)
msgs, err := cs.MessagesForTipset(context.Background(), ts)
require.NoError(t, err)
for _, m := range msgs {
minfo, err := msgIndex.GetMsgInfo(context.Background(), m.Cid())
require.NoError(t, err)
require.Equal(t, tsCid, minfo.TipSet)
require.Equal(t, ts.Height(), minfo.Epoch)
}
parents := ts.Parents()
ts, err = cs.GetTipSetFromKey(context.Background(), parents)
require.NoError(t, err)
}
}
func verifyMissing(t *testing.T, cs *mockChainStore, msgIndex MsgIndex, missing ...*types.TipSet) {
for _, ts := range missing {
msgs, err := cs.MessagesForTipset(context.Background(), ts)
require.NoError(t, err)
for _, m := range msgs {
_, err := msgIndex.GetMsgInfo(context.Background(), m.Cid())
require.Equal(t, ErrNotFound, err)
}
}
}
type mockChainStore struct {
notify store.ReorgNotifee
curTs *types.TipSet
tipsets map[types.TipSetKey]*types.TipSet
msgs map[types.TipSetKey][]types.ChainMsg
nonce uint64
}
var _ ChainStore = (*mockChainStore)(nil)
var systemAddr address.Address
var rng *rand.Rand
func init() {
systemAddr, _ = address.NewIDAddress(0)
rng = rand.New(rand.NewSource(314159))
// adjust those to make tests snappy
CoalesceMinDelay = 100 * time.Millisecond
CoalesceMaxDelay = time.Second
CoalesceMergeInterval = 100 * time.Millisecond
}
func newMockChainStore() *mockChainStore {
return &mockChainStore{
tipsets: make(map[types.TipSetKey]*types.TipSet),
msgs: make(map[types.TipSetKey][]types.ChainMsg),
}
}
func (cs *mockChainStore) genesis() {
genBlock := mock.MkBlock(nil, 0, 0)
genTs := mock.TipSet(genBlock)
cs.msgs[genTs.Key()] = nil
cs.setHead(genTs)
}
func (cs *mockChainStore) setHead(ts *types.TipSet) {
cs.curTs = ts
cs.tipsets[ts.Key()] = ts
}
func (cs *mockChainStore) advance() error {
ts := cs.makeBlk()
return cs.reorg(nil, []*types.TipSet{ts})
}
func (cs *mockChainStore) reorg(rev, app []*types.TipSet) error {
for _, ts := range rev {
parents := ts.Parents()
cs.curTs = cs.tipsets[parents]
}
for _, ts := range app {
cs.tipsets[ts.Key()] = ts
cs.curTs = ts
}
if cs.notify != nil {
return cs.notify(rev, app)
}
return nil
}
func (cs *mockChainStore) makeBlk() *types.TipSet {
height := cs.curTs.Height() + 1
blk := mock.MkBlock(cs.curTs, uint64(height), uint64(height))
blk.Messages = cs.makeGarbageCid()
ts := mock.TipSet(blk)
msg1 := cs.makeMsg()
msg2 := cs.makeMsg()
cs.msgs[ts.Key()] = []types.ChainMsg{msg1, msg2}
return ts
}
func (cs *mockChainStore) makeMsg() *types.Message {
nonce := cs.nonce
cs.nonce++
return &types.Message{To: systemAddr, From: systemAddr, Nonce: nonce}
}
func (cs *mockChainStore) makeGarbageCid() cid.Cid {
garbage := blocks.NewBlock([]byte{byte(rng.Intn(256)), byte(rng.Intn(256)), byte(rng.Intn(256))})
return garbage.Cid()
}
func (cs *mockChainStore) SubscribeHeadChanges(f store.ReorgNotifee) {
cs.notify = f
}
func (cs *mockChainStore) MessagesForTipset(ctx context.Context, ts *types.TipSet) ([]types.ChainMsg, error) {
msgs, ok := cs.msgs[ts.Key()]
if !ok {
return nil, errors.New("unknown tipset")
}
return msgs, nil
}
func (cs *mockChainStore) GetHeaviestTipSet() *types.TipSet {
return cs.curTs
}
func (cs *mockChainStore) GetTipSetFromKey(ctx context.Context, tsk types.TipSetKey) (*types.TipSet, error) {
ts, ok := cs.tipsets[tsk]
if !ok {
return nil, errors.New("unknown tipset")
}
return ts, nil
}
+20 -6
View File
@@ -52,8 +52,8 @@ func (sm *StateManager) Call(ctx context.Context, msg *types.Message, ts *types.
}
// CallWithGas calculates the state for a given tipset, and then applies the given message on top of that state.
func (sm *StateManager) CallWithGas(ctx context.Context, msg *types.Message, priorMsgs []types.ChainMsg, ts *types.TipSet) (*api.InvocResult, error) {
return sm.callInternal(ctx, msg, priorMsgs, ts, cid.Undef, sm.GetNetworkVersion, true, true)
func (sm *StateManager) CallWithGas(ctx context.Context, msg *types.Message, priorMsgs []types.ChainMsg, ts *types.TipSet, applyTsMessages bool) (*api.InvocResult, error) {
return sm.callInternal(ctx, msg, priorMsgs, ts, cid.Undef, sm.GetNetworkVersion, true, applyTsMessages)
}
// CallAtStateAndVersion allows you to specify a message to execute on the given stateCid and network version.
@@ -117,12 +117,22 @@ func (sm *StateManager) callInternal(ctx context.Context, msg *types.Message, pr
if stateCid == cid.Undef {
stateCid = ts.ParentState()
}
tsMsgs, err := sm.cs.MessagesForTipset(ctx, ts)
if err != nil {
return nil, xerrors.Errorf("failed to lookup messages for parent tipset: %w", err)
}
if applyTsMessages {
tsMsgs, err := sm.cs.MessagesForTipset(ctx, ts)
if err != nil {
return nil, xerrors.Errorf("failed to lookup messages for parent tipset: %w", err)
}
priorMsgs = append(tsMsgs, priorMsgs...)
} else {
var filteredTsMsgs []types.ChainMsg
for _, tsMsg := range tsMsgs {
//TODO we should technically be normalizing the filecoin address of from when we compare here
if tsMsg.VMMessage().From == msg.VMMessage().From {
filteredTsMsgs = append(filteredTsMsgs, tsMsg)
}
}
priorMsgs = append(filteredTsMsgs, priorMsgs...)
}
// Technically, the tipset we're passing in here should be ts+1, but that may not exist.
@@ -159,6 +169,9 @@ 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)
if err != nil {
@@ -191,6 +204,7 @@ 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)
+28
View File
@@ -128,10 +128,38 @@ func (sm *StateManager) ExecutionTraceWithMonitor(ctx context.Context, ts *types
}
func (sm *StateManager) ExecutionTrace(ctx context.Context, ts *types.TipSet) (cid.Cid, []*api.InvocResult, error) {
tsKey := ts.Key()
{
// check if we have the trace for this tipset in the cache
sm.execTraceCacheLock.Lock()
defer sm.execTraceCacheLock.Unlock()
if entry, ok := sm.execTraceCache.Get(tsKey); ok {
// we have to make a deep copy since caller can modify the invocTrace
return entry.postStateRoot, makeDeepCopy(entry.invocTrace), nil
}
}
var invocTrace []*api.InvocResult
st, err := sm.ExecutionTraceWithMonitor(ctx, ts, &InvocationTracer{trace: &invocTrace})
if err != nil {
return cid.Undef, nil, err
}
sm.execTraceCache.Add(tsKey, tipSetCacheEntry{st, makeDeepCopy(invocTrace)})
return st, invocTrace, nil
}
func makeDeepCopy(invocTrace []*api.InvocResult) []*api.InvocResult {
c := make([]*api.InvocResult, len(invocTrace))
for i, ir := range invocTrace {
if ir == nil {
continue
}
tmp := *ir
c[i] = &tmp
}
return c
}
+27 -17
View File
@@ -36,6 +36,7 @@ import (
"github.com/filecoin-project/lotus/chain/consensus"
"github.com/filecoin-project/lotus/chain/consensus/filcns"
"github.com/filecoin-project/lotus/chain/gen"
"github.com/filecoin-project/lotus/chain/index"
"github.com/filecoin-project/lotus/chain/stmgr"
. "github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/types"
@@ -55,6 +56,12 @@ 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) }
@@ -168,7 +175,7 @@ func TestForkHeightTriggers(t *testing.T) {
}
return st.Flush(ctx)
}}}, cg.BeaconSchedule(), datastore.NewMapDatastore())
}}}, cg.BeaconSchedule(), datastore.NewMapDatastore(), index.DummyMsgIndex)
if err != nil {
t.Fatal(err)
}
@@ -177,13 +184,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.Interface, error) {
sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Executor, error) {
nvm, err := vm.NewLegacyVM(ctx, vmopt)
if err != nil {
return nil, err
}
nvm.SetInvoker(inv)
return nvm, nil
return &mockExecutor{nvm}, nil
})
cg.SetStateManager(sm)
@@ -286,7 +293,7 @@ func testForkRefuseCall(t *testing.T, nullsBefore, nullsAfter int) {
root cid.Cid, height abi.ChainEpoch, ts *types.TipSet) (cid.Cid, error) {
migrationCount++
return root, nil
}}}, cg.BeaconSchedule(), datastore.NewMapDatastore())
}}}, cg.BeaconSchedule(), datastore.NewMapDatastore(), index.DummyMsgIndex)
if err != nil {
t.Fatal(err)
}
@@ -295,13 +302,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.Interface, error) {
sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Executor, error) {
nvm, err := vm.NewLegacyVM(ctx, vmopt)
if err != nil {
return nil, err
}
nvm.SetInvoker(inv)
return nvm, nil
return &mockExecutor{nvm}, nil
})
cg.SetStateManager(sm)
@@ -340,7 +347,7 @@ func testForkRefuseCall(t *testing.T, nullsBefore, nullsAfter int) {
currentHeight := ts.TipSet.TipSet().Height()
// CallWithGas calls on top of the given tipset.
ret, err := sm.CallWithGas(ctx, m, nil, ts.TipSet.TipSet())
ret, err := sm.CallWithGas(ctx, m, nil, ts.TipSet.TipSet(), true)
if parentHeight <= testForkHeight && currentHeight >= testForkHeight {
// If I had a fork, or I _will_ have a fork, it should fail.
require.Equal(t, ErrExpensiveFork, err)
@@ -361,7 +368,7 @@ func testForkRefuseCall(t *testing.T, nullsBefore, nullsAfter int) {
// Calls without a tipset should walk back to the last non-fork tipset.
// We _verify_ that the migration wasn't run multiple times at the end of the
// test.
ret, err = sm.CallWithGas(ctx, m, nil, nil)
ret, err = sm.CallWithGas(ctx, m, nil, nil, true)
require.NoError(t, err)
require.True(t, ret.MsgRct.ExitCode.IsSuccess())
@@ -504,7 +511,7 @@ func TestForkPreMigration(t *testing.T) {
return nil
},
}}},
}, cg.BeaconSchedule(), datastore.NewMapDatastore())
}, cg.BeaconSchedule(), datastore.NewMapDatastore(), index.DummyMsgIndex)
if err != nil {
t.Fatal(err)
}
@@ -517,13 +524,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.Interface, error) {
sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Executor, error) {
nvm, err := vm.NewLegacyVM(ctx, vmopt)
if err != nil {
return nil, err
}
nvm.SetInvoker(inv)
return nvm, nil
return &mockExecutor{nvm}, nil
})
cg.SetStateManager(sm)
@@ -579,6 +586,7 @@ func TestDisablePreMigration(t *testing.T) {
},
cg.BeaconSchedule(),
datastore.NewMapDatastore(),
index.DummyMsgIndex,
)
require.NoError(t, err)
require.NoError(t, sm.Start(context.Background()))
@@ -590,11 +598,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.Interface, error) {
sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Executor, error) {
nvm, err := vm.NewLegacyVM(ctx, vmopt)
require.NoError(t, err)
nvm.SetInvoker(inv)
return nvm, nil
return &mockExecutor{nvm}, nil
})
cg.SetStateManager(sm)
@@ -633,6 +641,7 @@ func TestMigrtionCache(t *testing.T) {
},
cg.BeaconSchedule(),
metadataDs,
index.DummyMsgIndex,
)
require.NoError(t, err)
require.NoError(t, sm.Start(context.Background()))
@@ -644,11 +653,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.Interface, error) {
sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Executor, error) {
nvm, err := vm.NewLegacyVM(ctx, vmopt)
require.NoError(t, err)
nvm.SetInvoker(inv)
return nvm, nil
return &mockExecutor{nvm}, nil
})
cg.SetStateManager(sm)
@@ -685,13 +694,14 @@ func TestMigrtionCache(t *testing.T) {
},
cg.BeaconSchedule(),
metadataDs,
index.DummyMsgIndex,
)
require.NoError(t, err)
sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Interface, error) {
sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Executor, error) {
nvm, err := vm.NewLegacyVM(ctx, vmopt)
require.NoError(t, err)
nvm.SetInvoker(inv)
return nvm, nil
return &mockExecutor{nvm}, nil
})
ctx := context.Background()
+61 -1
View File
@@ -10,6 +10,7 @@ import (
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/lotus/chain/index"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
)
@@ -18,6 +19,7 @@ import (
// happened, with an optional limit to how many epochs it will search. It guarantees that the message has been on
// chain for at least confidence epochs without being reverted before returning.
func (sm *StateManager) WaitForMessage(ctx context.Context, mcid cid.Cid, confidence uint64, lookbackLimit abi.ChainEpoch, allowReplaced bool) (*types.TipSet, *types.MessageReceipt, cid.Cid, error) {
// TODO use the index to speed this up.
ctx, cancel := context.WithCancel(ctx)
defer cancel()
@@ -145,7 +147,30 @@ func (sm *StateManager) SearchForMessage(ctx context.Context, head *types.TipSet
return head, r, foundMsg, nil
}
fts, r, foundMsg, err := sm.searchBackForMsg(ctx, head, msg, lookbackLimit, allowReplaced)
fts, r, foundMsg, err := sm.searchForIndexedMsg(ctx, mcid, msg)
switch {
case err == nil:
if r != nil && foundMsg.Defined() {
return fts, r, foundMsg, nil
}
// debug log this, it's noteworthy
if r == nil {
log.Debugf("missing receipt for message in index for %s", mcid)
}
if !foundMsg.Defined() {
log.Debugf("message %s not found", mcid)
}
case errors.Is(err, index.ErrNotFound):
// ok for the index to have incomplete data
default:
log.Warnf("error searching message index: %s", err)
}
fts, r, foundMsg, err = sm.searchBackForMsg(ctx, head, msg, lookbackLimit, allowReplaced)
if err != nil {
log.Warnf("failed to look back through chain for message %s", mcid)
@@ -159,6 +184,41 @@ func (sm *StateManager) SearchForMessage(ctx context.Context, head *types.TipSet
return fts, r, foundMsg, nil
}
func (sm *StateManager) searchForIndexedMsg(ctx context.Context, mcid cid.Cid, m types.ChainMsg) (*types.TipSet, *types.MessageReceipt, cid.Cid, error) {
minfo, err := sm.msgIndex.GetMsgInfo(ctx, mcid)
if err != nil {
return nil, nil, cid.Undef, xerrors.Errorf("error looking up message in index: %w", err)
}
// check the height against the current tipset; minimum execution confidence requires that the
// inclusion tipset height is lower than the current head + 1
curTs := sm.cs.GetHeaviestTipSet()
if curTs.Height() <= minfo.Epoch+1 {
return nil, nil, cid.Undef, xerrors.Errorf("indexed message does not appear before the current tipset; index epoch: %d, current epoch: %d", minfo.Epoch, curTs.Height())
}
// now get the execution tipset
// TODO optimization: the index should have it implicitly so we can return it in the msginfo.
xts, err := sm.cs.GetTipsetByHeight(ctx, minfo.Epoch+1, curTs, false)
if err != nil {
return nil, nil, cid.Undef, xerrors.Errorf("error looking up execution tipset: %w", err)
}
// check that the parent of the execution index is indeed the inclusion tipset
parent := xts.Parents()
parentCid, err := parent.Cid()
if err != nil {
return nil, nil, cid.Undef, xerrors.Errorf("error computing tipset cid: %w", err)
}
if !parentCid.Equals(minfo.TipSet) {
return nil, nil, cid.Undef, xerrors.Errorf("inclusion tipset mismatch: have %s, expected %s", parentCid, minfo.TipSet)
}
r, foundMsg, err := sm.tipsetExecutedMessage(ctx, xts, mcid, m.VMMessage(), false)
return xts, r, foundMsg, xerrors.Errorf("error in tipstExecutedMessage: %w", err)
}
// searchBackForMsg searches up to limit tipsets backwards from the given
// tipset for a message receipt.
// If limit is
+32 -10
View File
@@ -5,6 +5,7 @@ import (
"fmt"
"sync"
lru "github.com/hashicorp/golang-lru/v2"
"github.com/ipfs/go-cid"
dstore "github.com/ipfs/go-datastore"
cbor "github.com/ipfs/go-ipld-cbor"
@@ -25,6 +26,7 @@ import (
"github.com/filecoin-project/lotus/chain/actors/builtin/paych"
"github.com/filecoin-project/lotus/chain/actors/policy"
"github.com/filecoin-project/lotus/chain/beacon"
"github.com/filecoin-project/lotus/chain/index"
"github.com/filecoin-project/lotus/chain/rand"
"github.com/filecoin-project/lotus/chain/state"
"github.com/filecoin-project/lotus/chain/store"
@@ -38,6 +40,8 @@ import (
const LookbackNoLimit = api.LookbackNoLimit
const ReceiptAmtBitwidth = 3
const execTraceCacheSize = 16
var log = logging.Logger("statemgr")
type StateManagerAPI interface {
@@ -124,18 +128,24 @@ type StateManager struct {
compWait map[string]chan struct{}
stlk sync.Mutex
genesisMsigLk sync.Mutex
newVM func(context.Context, *vm.VMOpts) (vm.Interface, error)
newVM func(context.Context, *vm.VMOpts) (vm.Executor, error)
Syscalls vm.SyscallBuilder
preIgnitionVesting []msig0.State
postIgnitionVesting []msig0.State
postCalicoVesting []msig0.State
genesisPledge abi.TokenAmount
genesisMarketFunds abi.TokenAmount
genesisPledge abi.TokenAmount
tsExec Executor
tsExecMonitor ExecMonitor
beacon beacon.Schedule
// We keep a small cache for calls to ExecutionTrace which helps improve
// performance for node operators like exchanges and block explorers
execTraceCache *lru.ARCCache[types.TipSetKey, tipSetCacheEntry]
execTraceCacheLock sync.Mutex
msgIndex index.MsgIndex
}
// Caches a single state tree
@@ -144,7 +154,12 @@ type treeCache struct {
tree *state.StateTree
}
func NewStateManager(cs *store.ChainStore, exec Executor, sys vm.SyscallBuilder, us UpgradeSchedule, beacon beacon.Schedule, metadataDs dstore.Batching) (*StateManager, error) {
type tipSetCacheEntry struct {
postStateRoot cid.Cid
invocTrace []*api.InvocResult
}
func NewStateManager(cs *store.ChainStore, exec Executor, sys vm.SyscallBuilder, us UpgradeSchedule, beacon beacon.Schedule, metadataDs dstore.Batching, msgIndex index.MsgIndex) (*StateManager, error) {
// If we have upgrades, make sure they're in-order and make sense.
if err := us.Validate(); err != nil {
return nil, err
@@ -183,6 +198,11 @@ func NewStateManager(cs *store.ChainStore, exec Executor, sys vm.SyscallBuilder,
}
}
execTraceCache, err := lru.NewARC[types.TipSetKey, tipSetCacheEntry](execTraceCacheSize)
if err != nil {
return nil, err
}
return &StateManager{
networkVersions: networkVersions,
latestVersion: lastVersion,
@@ -198,12 +218,14 @@ func NewStateManager(cs *store.ChainStore, exec Executor, sys vm.SyscallBuilder,
root: cid.Undef,
tree: nil,
},
compWait: make(map[string]chan struct{}),
compWait: make(map[string]chan struct{}),
execTraceCache: execTraceCache,
msgIndex: msgIndex,
}, nil
}
func NewStateManagerWithUpgradeScheduleAndMonitor(cs *store.ChainStore, exec Executor, sys vm.SyscallBuilder, us UpgradeSchedule, b beacon.Schedule, em ExecMonitor, metadataDs dstore.Batching) (*StateManager, error) {
sm, err := NewStateManager(cs, exec, sys, us, b, metadataDs)
func NewStateManagerWithUpgradeScheduleAndMonitor(cs *store.ChainStore, exec Executor, sys vm.SyscallBuilder, us UpgradeSchedule, b beacon.Schedule, em ExecMonitor, metadataDs dstore.Batching, msgIndex index.MsgIndex) (*StateManager, error) {
sm, err := NewStateManager(cs, exec, sys, us, b, metadataDs, msgIndex)
if err != nil {
return nil, err
}
@@ -435,12 +457,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.Interface, error)) {
func (sm *StateManager) SetVMConstructor(nvm func(context.Context, *vm.VMOpts) (vm.Executor, error)) {
sm.newVM = nvm
}
func (sm *StateManager) VMConstructor() func(context.Context, *vm.VMOpts) (vm.Interface, error) {
return func(ctx context.Context, opts *vm.VMOpts) (vm.Interface, error) {
func (sm *StateManager) VMConstructor() func(context.Context, *vm.VMOpts) (vm.Executor, error) {
return func(ctx context.Context, opts *vm.VMOpts) (vm.Executor, error) {
return sm.newVM(ctx, opts)
}
}
+10 -12
View File
@@ -51,17 +51,13 @@ func (sm *StateManager) setupGenesisVestingSchedule(ctx context.Context) error {
return xerrors.Errorf("loading state tree: %w", err)
}
gmf, err := getFilMarketLocked(ctx, sTree)
if err != nil {
return xerrors.Errorf("setting up genesis market funds: %w", err)
}
gp, err := getFilPowerLocked(ctx, sTree)
if err != nil {
return xerrors.Errorf("setting up genesis pledge: %w", err)
}
sm.genesisMarketFunds = gmf
sm.genesisMsigLk.Lock()
defer sm.genesisMsigLk.Unlock()
sm.genesisPledge = gp
totalsByEpoch := make(map[abi.ChainEpoch]abi.TokenAmount)
@@ -128,6 +124,8 @@ func (sm *StateManager) setupPostIgnitionVesting(ctx context.Context) error {
totalsByEpoch[sixYears] = big.NewInt(100_000_000)
totalsByEpoch[sixYears] = big.Add(totalsByEpoch[sixYears], big.NewInt(300_000_000))
sm.genesisMsigLk.Lock()
defer sm.genesisMsigLk.Unlock()
sm.postIgnitionVesting = make([]msig0.State, 0, len(totalsByEpoch))
for k, v := range totalsByEpoch {
ns := msig0.State{
@@ -178,6 +176,9 @@ func (sm *StateManager) setupPostCalicoVesting(ctx context.Context) error {
totalsByEpoch[sixYears] = big.Add(totalsByEpoch[sixYears], big.NewInt(300_000_000))
totalsByEpoch[sixYears] = big.Add(totalsByEpoch[sixYears], big.NewInt(9_805_053))
sm.genesisMsigLk.Lock()
defer sm.genesisMsigLk.Unlock()
sm.postCalicoVesting = make([]msig0.State, 0, len(totalsByEpoch))
for k, v := range totalsByEpoch {
ns := msig0.State{
@@ -198,21 +199,20 @@ func (sm *StateManager) setupPostCalicoVesting(ctx context.Context) error {
func (sm *StateManager) GetFilVested(ctx context.Context, height abi.ChainEpoch) (abi.TokenAmount, error) {
vf := big.Zero()
sm.genesisMsigLk.Lock()
defer sm.genesisMsigLk.Unlock()
// TODO: combine all this?
if sm.preIgnitionVesting == nil || sm.genesisPledge.IsZero() || sm.genesisMarketFunds.IsZero() {
if sm.preIgnitionVesting == nil || sm.genesisPledge.IsZero() {
err := sm.setupGenesisVestingSchedule(ctx)
if err != nil {
return vf, xerrors.Errorf("failed to setup pre-ignition vesting schedule: %w", err)
}
}
if sm.postIgnitionVesting == nil {
err := sm.setupPostIgnitionVesting(ctx)
if err != nil {
return vf, xerrors.Errorf("failed to setup post-ignition vesting schedule: %w", err)
}
}
if sm.postCalicoVesting == nil {
err := sm.setupPostCalicoVesting(ctx)
@@ -246,8 +246,6 @@ func (sm *StateManager) GetFilVested(ctx context.Context, height abi.ChainEpoch)
if height <= build.UpgradeAssemblyHeight {
// continue to use preIgnitionGenInfos, nothing changed at the Ignition epoch
vf = big.Add(vf, sm.genesisPledge)
// continue to use preIgnitionGenInfos, nothing changed at the Ignition epoch
vf = big.Add(vf, sm.genesisMarketFunds)
}
return vf, nil
+1
View File
@@ -106,6 +106,7 @@ 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
+2 -1
View File
@@ -18,6 +18,7 @@ import (
"github.com/filecoin-project/lotus/chain/consensus"
"github.com/filecoin-project/lotus/chain/consensus/filcns"
"github.com/filecoin-project/lotus/chain/gen"
"github.com/filecoin-project/lotus/chain/index"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
@@ -214,7 +215,7 @@ func TestChainExportImportFull(t *testing.T) {
t.Fatal("imported chain differed from exported chain")
}
sm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), nil, filcns.DefaultUpgradeSchedule(), cg.BeaconSchedule(), ds)
sm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), nil, filcns.DefaultUpgradeSchedule(), cg.BeaconSchedule(), ds, index.DummyMsgIndex)
if err != nil {
t.Fatal(err)
}
+227
View File
@@ -0,0 +1,227 @@
package vm
import (
"context"
"errors"
"os"
"strconv"
"sync"
"go.opencensus.io/stats"
"go.opencensus.io/tag"
"github.com/ipfs/go-cid"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/metrics"
)
const (
// DefaultAvailableExecutionLanes is the number of available execution lanes; it is the bound of
// concurrent active executions.
// This is the default value in filecoin-ffi
DefaultAvailableExecutionLanes = 4
// DefaultPriorityExecutionLanes is the number of reserved execution lanes for priority computations.
// This is purely userspace, but we believe it is a reasonable default, even with more available
// lanes.
DefaultPriorityExecutionLanes = 2
)
var ErrExecutorDone = errors.New("executor has been released")
// the execution environment; see below for definition, methods, and initilization
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
done bool
}
var _ Executor = (*vmExecutor)(nil)
func newVMExecutor(vmi Interface, token *executionToken) Executor {
return &vmExecutor{vmi: vmi, token: token}
}
func (e *vmExecutor) ApplyMessage(ctx context.Context, cmsg types.ChainMsg) (*ApplyRet, error) {
e.lk.RLock()
defer e.lk.RUnlock()
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()
defer e.lk.RUnlock()
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
}
func (token *executionToken) Done() {
execution.putToken(token)
}
type executionEnv struct {
mx *sync.Mutex
cond *sync.Cond
// available executors
available int
// reserved executors
reserved int
}
func (e *executionEnv) getToken(lane ExecutionLane) *executionToken {
metricsUp(metrics.VMExecutionWaiting, lane)
defer metricsDown(metrics.VMExecutionWaiting, lane)
e.mx.Lock()
defer e.mx.Unlock()
switch lane {
case ExecutionLaneDefault:
for e.available <= e.reserved {
e.cond.Wait()
}
e.available--
metricsUp(metrics.VMExecutionRunning, lane)
return &executionToken{lane: lane, reserved: 0}
case ExecutionLanePriority:
for e.available == 0 {
e.cond.Wait()
}
e.available--
reserving := 0
if e.reserved > 0 {
e.reserved--
reserving = 1
}
metricsUp(metrics.VMExecutionRunning, lane)
return &executionToken{lane: lane, reserved: reserving}
default:
// already checked at interface boundary in NewVM, so this is appropriate
panic("bogus execution lane")
}
}
func (e *executionEnv) putToken(token *executionToken) {
e.mx.Lock()
defer e.mx.Unlock()
e.available++
e.reserved += token.reserved
e.cond.Broadcast()
metricsDown(metrics.VMExecutionRunning, token.lane)
}
func metricsUp(metric *stats.Int64Measure, lane ExecutionLane) {
metricsAdjust(metric, lane, 1)
}
func metricsDown(metric *stats.Int64Measure, lane ExecutionLane) {
metricsAdjust(metric, lane, -1)
}
func metricsAdjust(metric *stats.Int64Measure, lane ExecutionLane, delta int) {
laneName := "default"
if lane > ExecutionLaneDefault {
laneName = "priority"
}
ctx, _ := tag.New(
context.Background(),
tag.Upsert(metrics.ExecutionLane, laneName),
)
stats.Record(ctx, metric.M(int64(delta)))
}
func init() {
var available, priority int
var err error
concurrency := os.Getenv("LOTUS_FVM_CONCURRENCY")
if concurrency == "" {
available = DefaultAvailableExecutionLanes
} else {
available, err = strconv.Atoi(concurrency)
if err != nil {
panic(err)
}
}
reserved := os.Getenv("LOTUS_FVM_CONCURRENCY_RESERVED")
if reserved == "" {
priority = DefaultPriorityExecutionLanes
} else {
priority, err = strconv.Atoi(reserved)
if err != nil {
panic(err)
}
}
// some sanity checks
if available < 2 {
panic("insufficient execution concurrency")
}
if priority > available-1 {
panic("insufficient default execution concurrency")
}
mx := &sync.Mutex{}
cond := sync.NewCond(mx)
execution = &executionEnv{
mx: mx,
cond: cond,
available: available,
reserved: priority,
}
}
+2
View File
@@ -250,6 +250,8 @@ type VMOpts struct {
Tracing bool
// ReturnEvents decodes and returns emitted events.
ReturnEvents bool
// ExecutionLane specifies the execution priority of the created vm
ExecutionLane ExecutionLane
}
func NewLegacyVM(ctx context.Context, opts *VMOpts) (*LegacyVM, error) {
+40 -1
View File
@@ -2,6 +2,7 @@ package vm
import (
"context"
"fmt"
"os"
cid "github.com/ipfs/go-cid"
@@ -17,6 +18,15 @@ var (
StatApplied uint64
)
type ExecutionLane int
const (
// ExecutionLaneDefault signifies a default, non prioritized execution lane.
ExecutionLaneDefault ExecutionLane = iota
// ExecutionLanePriority signifies a prioritized execution lane with reserved resources.
ExecutionLanePriority
)
type Interface interface {
// Applies the given message onto the VM's current state, returning the result of the execution
ApplyMessage(ctx context.Context, cmsg types.ChainMsg) (*ApplyRet, error)
@@ -27,13 +37,24 @@ 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.
// Message failures, unexpected terminations,gas costs, etc. should all be ignored.
var useFvmDebug = os.Getenv("LOTUS_FVM_DEVELOPER_DEBUG") == "1"
func NewVM(ctx context.Context, opts *VMOpts) (Interface, error) {
func newVM(ctx context.Context, opts *VMOpts) (Interface, error) {
if opts.NetworkVersion >= network.Version16 {
if useFvmDebug {
return NewDualExecutionFVM(ctx, opts)
@@ -43,3 +64,21 @@ func NewVM(ctx context.Context, opts *VMOpts) (Interface, error) {
return NewLegacyVM(ctx, opts)
}
func NewVM(ctx context.Context, opts *VMOpts) (Executor, 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 := newVM(ctx, opts)
if err != nil {
token.Done()
return nil, err
}
return newVMExecutor(vmi, token), nil
}
+2 -1
View File
@@ -36,6 +36,7 @@ import (
badgerbs "github.com/filecoin-project/lotus/blockstore/badger"
"github.com/filecoin-project/lotus/chain/consensus"
"github.com/filecoin-project/lotus/chain/consensus/filcns"
"github.com/filecoin-project/lotus/chain/index"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
@@ -229,7 +230,7 @@ var importBenchCmd = &cli.Command{
defer cs.Close() //nolint:errcheck
// TODO: We need to supply the actual beacon after v14
stm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(verifier), filcns.DefaultUpgradeSchedule(), nil, metadataDs)
stm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(verifier), filcns.DefaultUpgradeSchedule(), nil, metadataDs, index.DummyMsgIndex)
if err != nil {
return err
}
+5 -7
View File
@@ -325,14 +325,12 @@ func handleMiningInfo(ctx context.Context, cctx *cli.Context, fullapi v1api.Full
}
}
{
fmt.Println()
ws, err := nodeApi.WorkerStats(ctx)
if err != nil {
return xerrors.Errorf("getting worker stats: %w", err)
}
fmt.Println()
ws, err := nodeApi.WorkerStats(ctx)
if err != nil {
fmt.Printf("ERROR: getting worker stats: %s\n", err)
} else {
workersByType := map[string]int{
sealtasks.WorkerSealing: 0,
sealtasks.WorkerWindowPoSt: 0,
+3 -2
View File
@@ -35,6 +35,7 @@ import (
"github.com/filecoin-project/lotus/chain/consensus"
"github.com/filecoin-project/lotus/chain/consensus/filcns"
"github.com/filecoin-project/lotus/chain/gen/genesis"
"github.com/filecoin-project/lotus/chain/index"
"github.com/filecoin-project/lotus/chain/state"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/store"
@@ -513,7 +514,7 @@ var chainBalanceStateCmd = &cli.Command{
cst := cbor.NewCborStore(bs)
store := adt.WrapStore(ctx, cst)
sm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(ffiwrapper.ProofVerifier), filcns.DefaultUpgradeSchedule(), nil, mds)
sm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(ffiwrapper.ProofVerifier), filcns.DefaultUpgradeSchedule(), nil, mds, index.DummyMsgIndex)
if err != nil {
return err
}
@@ -737,7 +738,7 @@ var chainPledgeCmd = &cli.Command{
cst := cbor.NewCborStore(bs)
store := adt.WrapStore(ctx, cst)
sm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(ffiwrapper.ProofVerifier), filcns.DefaultUpgradeSchedule(), nil, mds)
sm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(ffiwrapper.ProofVerifier), filcns.DefaultUpgradeSchedule(), nil, mds, index.DummyMsgIndex)
if err != nil {
return err
}
+383
View File
@@ -0,0 +1,383 @@
package main
import (
"container/list"
"context"
"database/sql"
"fmt"
"hash/crc32"
"strconv"
"sync"
"time"
"github.com/ipfs/go-cid"
"github.com/urfave/cli/v2"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
cliutil "github.com/filecoin-project/lotus/cli/util"
)
var chainwatchCmd = &cli.Command{
Name: "chainwatch",
Usage: "lotus chainwatch",
Flags: []cli.Flag{
&cli.StringFlag{
Name: "db",
EnvVars: []string{"CHAINWATCH_DB"},
Value: "./chainwatch.db",
},
},
Subcommands: []*cli.Command{
chainwatchRunCmd,
chainwatchDotCmd,
},
}
var chainwatchDotCmd = &cli.Command{
Name: "dot",
Usage: "generate dot graphs",
ArgsUsage: "<minHeight> <toseeHeight>",
Action: func(cctx *cli.Context) error {
st, err := cwOpenStorage(cctx.String("db"))
if err != nil {
return err
}
minH, err := strconv.ParseInt(cctx.Args().Get(0), 10, 32)
if err != nil {
return err
}
tosee, err := strconv.ParseInt(cctx.Args().Get(1), 10, 32)
if err != nil {
return err
}
maxH := minH + tosee
res, err := st.db.Query(`select block, parent, b.miner, b.height, p.height from block_parents
inner join blocks b on block_parents.block = b.cid
inner join blocks p on block_parents.parent = p.cid
where b.height > ? and b.height < ?`, minH, maxH)
if err != nil {
return err
}
fmt.Println("digraph D {")
for res.Next() {
var block, parent, miner string
var height, ph uint64
if err := res.Scan(&block, &parent, &miner, &height, &ph); err != nil {
return err
}
bc, err := cid.Parse(block)
if err != nil {
return err
}
has := st.hasBlock(bc)
col := crc32.Checksum([]byte(miner), crc32.MakeTable(crc32.Castagnoli))&0xc0c0c0c0 + 0x30303030
hasstr := ""
if !has {
//col = 0xffffffff
hasstr = " UNSYNCED"
}
nulls := height - ph - 1
for i := uint64(0); i < nulls; i++ {
name := block + "NP" + fmt.Sprint(i)
fmt.Printf("%s [label = \"NULL:%d\", fillcolor = \"#ffddff\", style=filled, forcelabels=true]\n%s -> %s\n",
name, height-nulls+i, name, parent)
parent = name
}
fmt.Printf("%s [label = \"%s:%d%s\", fillcolor = \"#%06x\", style=filled, forcelabels=true]\n%s -> %s\n", block, miner, height, hasstr, col, block, parent)
}
if res.Err() != nil {
return res.Err()
}
fmt.Println("}")
return nil
},
}
var chainwatchRunCmd = &cli.Command{
Name: "run",
Usage: "Start lotus chainwatch",
Action: func(cctx *cli.Context) error {
api, closer, err := cliutil.GetFullNodeAPIV1(cctx)
if err != nil {
return err
}
defer closer()
ctx := cliutil.ReqContext(cctx)
v, err := api.Version(ctx)
if err != nil {
return err
}
log.Infof("Remote version: %s", v.Version)
st, err := cwOpenStorage(cctx.String("db")) // todo flag
if err != nil {
return err
}
defer st.close() // nolint:errcheck
cwRunSyncer(ctx, api, st)
go cwSubBlocks(ctx, api, st)
<-ctx.Done()
return nil
},
}
func cwSubBlocks(ctx context.Context, api api.FullNode, st *cwStorage) {
sub, err := api.SyncIncomingBlocks(ctx)
if err != nil {
log.Error(err)
return
}
for bh := range sub {
err := st.storeHeaders(map[cid.Cid]*types.BlockHeader{
bh.Cid(): bh,
}, false)
if err != nil {
log.Error(err)
}
}
}
func cwRunSyncer(ctx context.Context, api api.FullNode, st *cwStorage) {
notifs, err := api.ChainNotify(ctx)
if err != nil {
panic(err)
}
go func() {
for notif := range notifs {
for _, change := range notif {
switch change.Type {
case store.HCCurrent:
fallthrough
case store.HCApply:
syncHead(ctx, api, st, change.Val)
case store.HCRevert:
log.Warnf("revert todo")
}
}
}
}()
}
func syncHead(ctx context.Context, api api.FullNode, st *cwStorage, ts *types.TipSet) {
log.Infof("Getting headers / actors")
toSync := map[cid.Cid]*types.BlockHeader{}
toVisit := list.New()
for _, header := range ts.Blocks() {
toVisit.PushBack(header)
}
for toVisit.Len() > 0 {
bh := toVisit.Remove(toVisit.Back()).(*types.BlockHeader)
if _, seen := toSync[bh.Cid()]; seen || st.hasBlock(bh.Cid()) {
continue
}
toSync[bh.Cid()] = bh
if len(toSync)%500 == 10 {
log.Infof("todo: (%d) %s", len(toSync), bh.Cid())
}
if len(bh.Parents) == 0 {
continue
}
if bh.Height <= 530000 {
continue
}
pts, err := api.ChainGetTipSet(ctx, types.NewTipSetKey(bh.Parents...))
if err != nil {
log.Error(err)
continue
}
for _, header := range pts.Blocks() {
toVisit.PushBack(header)
}
}
log.Infof("Syncing %d blocks", len(toSync))
log.Infof("Persisting headers")
if err := st.storeHeaders(toSync, true); err != nil {
log.Error(err)
return
}
log.Infof("Sync done")
}
type cwStorage struct {
db *sql.DB
headerLk sync.Mutex
}
func cwOpenStorage(dbSource string) (*cwStorage, error) {
db, err := sql.Open("sqlite3", dbSource)
if err != nil {
return nil, err
}
st := &cwStorage{db: db}
return st, st.setup()
}
func (st *cwStorage) setup() error {
tx, err := st.db.Begin()
if err != nil {
return err
}
_, err = tx.Exec(`
create table if not exists blocks
(
cid text not null
constraint blocks_pk
primary key,
parentWeight numeric not null,
parentStateRoot text not null,
height int not null,
miner text not null
constraint blocks_id_address_map_miner_fk
references id_address_map (address),
timestamp int not null,
vrfproof blob
);
create unique index if not exists block_cid_uindex
on blocks (cid);
create table if not exists blocks_synced
(
cid text not null
constraint blocks_synced_pk
primary key
constraint blocks_synced_blocks_cid_fk
references blocks,
add_ts int not null
);
create unique index if not exists blocks_synced_cid_uindex
on blocks_synced (cid);
create table if not exists block_parents
(
block text not null
constraint block_parents_blocks_cid_fk
references blocks,
parent text not null
constraint block_parents_blocks_cid_fk_2
references blocks
);
create unique index if not exists block_parents_block_parent_uindex
on block_parents (block, parent);
create unique index if not exists blocks_cid_uindex
on blocks (cid);
`)
if err != nil {
return err
}
return tx.Commit()
}
func (st *cwStorage) hasBlock(bh cid.Cid) bool {
var exitsts bool
err := st.db.QueryRow(`select exists (select 1 FROM blocks_synced where cid=?)`, bh.String()).Scan(&exitsts)
if err != nil {
log.Error(err)
return false
}
return exitsts
}
func (st *cwStorage) storeHeaders(bhs map[cid.Cid]*types.BlockHeader, sync bool) error {
st.headerLk.Lock()
defer st.headerLk.Unlock()
tx, err := st.db.Begin()
if err != nil {
return err
}
stmt, err := tx.Prepare(`insert into blocks (cid, parentWeight, parentStateRoot, height, miner, "timestamp", vrfproof) values (?, ?, ?, ?, ?, ?, ?) on conflict do nothing`)
if err != nil {
return err
}
defer stmt.Close() // nolint:errcheck
for _, bh := range bhs {
if _, err := stmt.Exec(bh.Cid().String(),
bh.ParentWeight.String(),
bh.ParentStateRoot.String(),
bh.Height,
bh.Miner.String(),
bh.Timestamp,
bh.Ticket.VRFProof,
); err != nil {
return err
}
}
stmt2, err := tx.Prepare(`insert into block_parents (block, parent) values (?, ?) on conflict do nothing`)
if err != nil {
return err
}
defer stmt2.Close() // nolint:errcheck
for _, bh := range bhs {
for _, parent := range bh.Parents {
if _, err := stmt2.Exec(bh.Cid().String(), parent.String()); err != nil {
return err
}
}
}
if sync {
stmt, err := tx.Prepare(`insert into blocks_synced (cid, add_ts) values (?, ?) on conflict do nothing`)
if err != nil {
return err
}
defer stmt.Close() // nolint:errcheck
now := time.Now().Unix()
for _, bh := range bhs {
if _, err := stmt.Exec(bh.Cid().String(), now); err != nil {
return err
}
}
}
return tx.Commit()
}
func (st *cwStorage) close() error {
return st.db.Close()
}
+4 -3
View File
@@ -20,6 +20,7 @@ import (
"github.com/filecoin-project/lotus/chain/beacon/drand"
"github.com/filecoin-project/lotus/chain/consensus"
"github.com/filecoin-project/lotus/chain/consensus/filcns"
"github.com/filecoin-project/lotus/chain/index"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
@@ -111,7 +112,7 @@ var gasTraceCmd = &cli.Command{
cs := store.NewChainStore(bs, bs, mds, filcns.Weight, nil)
defer cs.Close() //nolint:errcheck
sm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(ffiwrapper.ProofVerifier), filcns.DefaultUpgradeSchedule(), shd, mds)
sm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(ffiwrapper.ProofVerifier), filcns.DefaultUpgradeSchedule(), shd, mds, index.DummyMsgIndex)
if err != nil {
return err
}
@@ -212,7 +213,7 @@ var replayOfflineCmd = &cli.Command{
cs := store.NewChainStore(bs, bs, mds, filcns.Weight, nil)
defer cs.Close() //nolint:errcheck
sm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(ffiwrapper.ProofVerifier), filcns.DefaultUpgradeSchedule(), shd, mds)
sm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(ffiwrapper.ProofVerifier), filcns.DefaultUpgradeSchedule(), shd, mds, index.DummyMsgIndex)
if err != nil {
return err
}
@@ -240,7 +241,7 @@ var replayOfflineCmd = &cli.Command{
}
tw := tabwriter.NewWriter(os.Stdout, 8, 2, 2, ' ', tabwriter.AlignRight)
res, err := sm.CallWithGas(ctx, msg, []types.ChainMsg{}, executionTs)
res, err := sm.CallWithGas(ctx, msg, []types.ChainMsg{}, executionTs, true)
if err != nil {
return err
}
+2 -1
View File
@@ -21,6 +21,7 @@ import (
"github.com/filecoin-project/lotus/chain/actors"
"github.com/filecoin-project/lotus/chain/consensus"
"github.com/filecoin-project/lotus/chain/consensus/filcns"
"github.com/filecoin-project/lotus/chain/index"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
@@ -90,7 +91,7 @@ var invariantsCmd = &cli.Command{
cs := store.NewChainStore(bs, bs, mds, filcns.Weight, nil)
defer cs.Close() //nolint:errcheck
sm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(ffiwrapper.ProofVerifier), filcns.DefaultUpgradeSchedule(), nil, mds)
sm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(ffiwrapper.ProofVerifier), filcns.DefaultUpgradeSchedule(), nil, mds, index.DummyMsgIndex)
if err != nil {
return err
}
+2
View File
@@ -26,6 +26,7 @@ func main() {
base32Cmd,
base16Cmd,
bitFieldCmd,
chainwatchCmd,
cronWcCmd,
frozenMinersCmd,
dealLabelCmd,
@@ -83,6 +84,7 @@ func main() {
invariantsCmd,
gasTraceCmd,
replayOfflineCmd,
msgindexCmd,
}
app := &cli.App{
+2 -1
View File
@@ -41,6 +41,7 @@ import (
"github.com/filecoin-project/lotus/chain/actors/builtin/verifreg"
"github.com/filecoin-project/lotus/chain/consensus"
"github.com/filecoin-project/lotus/chain/consensus/filcns"
"github.com/filecoin-project/lotus/chain/index"
"github.com/filecoin-project/lotus/chain/state"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/store"
@@ -123,7 +124,7 @@ var migrationsCmd = &cli.Command{
defer cs.Close() //nolint:errcheck
// Note: we use a map datastore for the metadata to avoid writing / using cached migration results in the metadata store
sm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(ffiwrapper.ProofVerifier), filcns.DefaultUpgradeSchedule(), nil, datastore.NewMapDatastore())
sm, err := stmgr.NewStateManager(cs, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(ffiwrapper.ProofVerifier), filcns.DefaultUpgradeSchedule(), nil, datastore.NewMapDatastore(), index.DummyMsgIndex)
if err != nil {
return err
}
+221
View File
@@ -0,0 +1,221 @@
package main
import (
"database/sql"
"fmt"
"path"
"github.com/ipfs/go-cid"
_ "github.com/mattn/go-sqlite3"
"github.com/mitchellh/go-homedir"
"github.com/urfave/cli/v2"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-state-types/abi"
lcli "github.com/filecoin-project/lotus/cli"
)
var msgindexCmd = &cli.Command{
Name: "msgindex",
Usage: "Tools for managing the message index",
Subcommands: []*cli.Command{
msgindexBackfillCmd,
msgindexPruneCmd,
},
}
var msgindexBackfillCmd = &cli.Command{
Name: "backfill",
Usage: "Backfill the message index for a number of epochs starting from a specified height",
Flags: []cli.Flag{
&cli.IntFlag{
Name: "from",
Value: 0,
Usage: "height to start the backfill; uses the current head if omitted",
},
&cli.IntFlag{
Name: "epochs",
Value: 1800,
Usage: "number of epochs to backfill; defaults to 1800 (2 finalities)",
},
&cli.StringFlag{
Name: "repo",
Value: "~/.lotus",
Usage: "path to the repo",
},
},
Action: func(cctx *cli.Context) error {
api, closer, err := lcli.GetFullNodeAPI(cctx)
if err != nil {
return err
}
defer closer()
ctx := lcli.ReqContext(cctx)
curTs, err := api.ChainHead(ctx)
if err != nil {
return err
}
startHeight := int64(cctx.Int("from"))
if startHeight == 0 {
startHeight = int64(curTs.Height()) - 1
}
epochs := cctx.Int("epochs")
basePath, err := homedir.Expand(cctx.String("repo"))
if err != nil {
return err
}
dbPath := path.Join(basePath, "sqlite", "msgindex.db")
db, err := sql.Open("sqlite3", dbPath)
if err != nil {
return err
}
defer func() {
err := db.Close()
if err != nil {
fmt.Printf("ERROR: closing db: %s", err)
}
}()
tx, err := db.Begin()
if err != nil {
return err
}
insertStmt, err := tx.Prepare("INSERT INTO messages VALUES (?, ?, ?)")
if err != nil {
return err
}
insertMsg := func(cid, tsCid cid.Cid, epoch abi.ChainEpoch) error {
key := cid.String()
tskey := tsCid.String()
if _, err := insertStmt.Exec(key, tskey, int64(epoch)); err != nil {
return err
}
return nil
}
rollback := func() {
if err := tx.Rollback(); err != nil {
fmt.Printf("ERROR: rollback: %s", err)
}
}
for i := 0; i < epochs; i++ {
epoch := abi.ChainEpoch(startHeight - int64(i))
ts, err := api.ChainGetTipSetByHeight(ctx, epoch, curTs.Key())
if err != nil {
rollback()
return err
}
tsCid, err := ts.Key().Cid()
if err != nil {
rollback()
return err
}
msgs, err := api.ChainGetMessagesInTipset(ctx, ts.Key())
if err != nil {
rollback()
return err
}
for _, msg := range msgs {
if err := insertMsg(msg.Cid, tsCid, epoch); err != nil {
rollback()
return err
}
}
}
if err := tx.Commit(); err != nil {
return err
}
return nil
},
}
var msgindexPruneCmd = &cli.Command{
Name: "prune",
Usage: "Prune the message index for messages included before a given epoch",
Flags: []cli.Flag{
&cli.IntFlag{
Name: "from",
Usage: "height to start the prune; if negative it indicates epochs from current head",
},
&cli.StringFlag{
Name: "repo",
Value: "~/.lotus",
Usage: "path to the repo",
},
},
Action: func(cctx *cli.Context) error {
api, closer, err := lcli.GetFullNodeAPI(cctx)
if err != nil {
return err
}
defer closer()
ctx := lcli.ReqContext(cctx)
startHeight := int64(cctx.Int("from"))
if startHeight < 0 {
curTs, err := api.ChainHead(ctx)
if err != nil {
return err
}
startHeight += int64(curTs.Height())
if startHeight < 0 {
return xerrors.Errorf("bogus start height %d", startHeight)
}
}
basePath, err := homedir.Expand(cctx.String("repo"))
if err != nil {
return err
}
dbPath := path.Join(basePath, "sqlite", "msgindex.db")
db, err := sql.Open("sqlite3", dbPath)
if err != nil {
return err
}
defer func() {
err := db.Close()
if err != nil {
fmt.Printf("ERROR: closing db: %s", err)
}
}()
tx, err := db.Begin()
if err != nil {
return err
}
if _, err := tx.Exec("DELETE FROM messages WHERE epoch < ?", startHeight); err != nil {
if err := tx.Rollback(); err != nil {
fmt.Printf("ERROR: rollback: %s", err)
}
return err
}
if err := tx.Commit(); err != nil {
return err
}
return nil
},
}
+2 -1
View File
@@ -26,6 +26,7 @@ import (
"github.com/filecoin-project/lotus/chain/actors/builtin"
"github.com/filecoin-project/lotus/chain/consensus"
"github.com/filecoin-project/lotus/chain/consensus/filcns"
"github.com/filecoin-project/lotus/chain/index"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
@@ -308,7 +309,7 @@ to reduce the number of decode operations performed by caching the decoded objec
}
tsExec := consensus.NewTipSetExecutor(filcns.RewardFunc)
sm, err := stmgr.NewStateManager(cs, tsExec, vm.Syscalls(ffiwrapper.ProofVerifier), filcns.DefaultUpgradeSchedule(), nil, mds)
sm, err := stmgr.NewStateManager(cs, tsExec, vm.Syscalls(ffiwrapper.ProofVerifier), filcns.DefaultUpgradeSchedule(), nil, mds, index.DummyMsgIndex)
if err != nil {
return err
}
+3 -2
View File
@@ -12,6 +12,7 @@ import (
"github.com/filecoin-project/lotus/blockstore"
"github.com/filecoin-project/lotus/chain/consensus"
"github.com/filecoin-project/lotus/chain/consensus/filcns"
"github.com/filecoin-project/lotus/chain/index"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
@@ -106,7 +107,7 @@ func (nd *Node) LoadSim(ctx context.Context, name string) (*Simulation, error) {
if err != nil {
return nil, xerrors.Errorf("failed to create upgrade schedule for simulation %s: %w", name, err)
}
sim.StateManager, err = stmgr.NewStateManager(nd.Chainstore, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(mock.Verifier), us, nil, nd.MetadataDS)
sim.StateManager, err = stmgr.NewStateManager(nd.Chainstore, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(mock.Verifier), us, nil, nd.MetadataDS, index.DummyMsgIndex)
if err != nil {
return nil, xerrors.Errorf("failed to create state manager for simulation %s: %w", name, err)
}
@@ -125,7 +126,7 @@ func (nd *Node) CreateSim(ctx context.Context, name string, head *types.TipSet)
if err != nil {
return nil, err
}
sm, err := stmgr.NewStateManager(nd.Chainstore, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(mock.Verifier), filcns.DefaultUpgradeSchedule(), nil, nd.MetadataDS)
sm, err := stmgr.NewStateManager(nd.Chainstore, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(mock.Verifier), filcns.DefaultUpgradeSchedule(), nil, nd.MetadataDS, index.DummyMsgIndex)
if err != nil {
return nil, xerrors.Errorf("creating state manager: %w", err)
}
+2 -1
View File
@@ -17,6 +17,7 @@ import (
"github.com/filecoin-project/lotus/chain/consensus"
"github.com/filecoin-project/lotus/chain/consensus/filcns"
"github.com/filecoin-project/lotus/chain/index"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/chain/vm"
@@ -201,7 +202,7 @@ func (sim *Simulation) SetUpgradeHeight(nv network.Version, epoch abi.ChainEpoch
if err != nil {
return err
}
sm, err := stmgr.NewStateManager(sim.Node.Chainstore, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(mock.Verifier), newUpgradeSchedule, nil, sim.Node.MetadataDS)
sm, err := stmgr.NewStateManager(sim.Node.Chainstore, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(mock.Verifier), newUpgradeSchedule, nil, sim.Node.MetadataDS, index.DummyMsgIndex)
if err != nil {
return err
}
+9 -1
View File
@@ -13,6 +13,7 @@ import (
"io/ioutil"
"net/http"
"os"
"path"
"runtime/pprof"
"strings"
@@ -35,6 +36,7 @@ import (
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/consensus"
"github.com/filecoin-project/lotus/chain/consensus/filcns"
"github.com/filecoin-project/lotus/chain/index"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
@@ -540,7 +542,7 @@ func ImportChain(ctx context.Context, r repo.Repo, fname string, snapshot bool)
}
// TODO: We need to supply the actual beacon after v14
stm, err := stmgr.NewStateManager(cst, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(ffiwrapper.ProofVerifier), filcns.DefaultUpgradeSchedule(), nil, mds)
stm, err := stmgr.NewStateManager(cst, consensus.NewTipSetExecutor(filcns.RewardFunc), vm.Syscalls(ffiwrapper.ProofVerifier), filcns.DefaultUpgradeSchedule(), nil, mds, index.DummyMsgIndex)
if err != nil {
return err
}
@@ -557,5 +559,11 @@ func ImportChain(ctx context.Context, r repo.Repo, fname string, snapshot bool)
return err
}
log.Info("populating message index...")
if err := index.PopulateAfterSnapshot(ctx, path.Join(lr.Path(), "sqlite"), cst); err != nil {
return err
}
log.Info("populating message index done")
return nil
}
+3 -2
View File
@@ -22,6 +22,7 @@ import (
"github.com/filecoin-project/lotus/chain/actors/builtin"
"github.com/filecoin-project/lotus/chain/consensus"
"github.com/filecoin-project/lotus/chain/consensus/filcns"
"github.com/filecoin-project/lotus/chain/index"
"github.com/filecoin-project/lotus/chain/state"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/store"
@@ -108,7 +109,7 @@ func (d *Driver) ExecuteTipset(bs blockstore.Blockstore, ds ds.Batching, params
cs = store.NewChainStore(bs, bs, ds, filcns.Weight, nil)
tse = consensus.NewTipSetExecutor(filcns.RewardFunc)
sm, err = stmgr.NewStateManager(cs, tse, syscalls, filcns.DefaultUpgradeSchedule(), nil, ds)
sm, err = stmgr.NewStateManager(cs, tse, syscalls, filcns.DefaultUpgradeSchedule(), nil, ds, index.DummyMsgIndex)
)
if err != nil {
return nil, err
@@ -157,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.Interface, error) {
sm.SetVMConstructor(func(ctx context.Context, vmopt *vm.VMOpts) (vm.Executor, error) {
vmopt.CircSupplyCalc = func(context.Context, abi.ChainEpoch, *state.StateTree) (abi.TokenAmount, error) {
return big.Zero(), nil
}
@@ -390,3 +390,11 @@
#DatabasePath = ""
[Index]
# EnableMsgIndex enables indexing of messages on chain.
#
# type: bool
# env var: LOTUS_INDEX_ENABLEMSGINDEX
#EnableMsgIndex = false
+3
View File
@@ -43,6 +43,9 @@ const (
// TargetAPI defines the API methods that the Node depends on
// (to make it easy to mock for tests)
type TargetAPI interface {
GasEstimateGasPremium(context.Context, uint64, address.Address, int64, types.TipSetKey) (types.BigInt, error)
StateReplay(context.Context, types.TipSetKey, cid.Cid) (*api.InvocResult, error)
StateMinerSectorCount(context.Context, address.Address, types.TipSetKey) (api.MinerSectors, error)
Version(context.Context) (api.APIVersion, error)
ChainGetParentMessages(context.Context, cid.Cid) ([]api.Message, error)
ChainGetParentReceipts(context.Context, cid.Cid) ([]*types.MessageReceipt, error)
+30
View File
@@ -23,6 +23,36 @@ import (
"github.com/filecoin-project/lotus/node/modules/dtypes"
)
func (gw *Node) StateReplay(ctx context.Context, tsk types.TipSetKey, c cid.Cid) (*api.InvocResult, error) {
if err := gw.limit(ctx, chainRateLimitTokens); err != nil {
return nil, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return nil, err
}
return gw.target.StateReplay(ctx, tsk, c)
}
func (gw *Node) GasEstimateGasPremium(ctx context.Context, nblocksincl uint64, sender address.Address, gaslimit int64, tsk types.TipSetKey) (types.BigInt, error) {
if err := gw.limit(ctx, chainRateLimitTokens); err != nil {
return types.BigInt{}, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return types.BigInt{}, err
}
return gw.target.GasEstimateGasPremium(ctx, nblocksincl, sender, gaslimit, tsk)
}
func (gw *Node) StateMinerSectorCount(ctx context.Context, m address.Address, tsk types.TipSetKey) (api.MinerSectors, error) {
if err := gw.limit(ctx, chainRateLimitTokens); err != nil {
return api.MinerSectors{}, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return api.MinerSectors{}, err
}
return gw.target.StateMinerSectorCount(ctx, m, tsk)
}
func (gw *Node) Discover(ctx context.Context) (apitypes.OpenRPCDocument, error) {
return build.OpenRPCDiscoverJSON_Gateway(), nil
}
+3 -3
View File
@@ -78,7 +78,7 @@ require (
github.com/influxdata/influxdb1-client v0.0.0-20200827194710-b269163b24ab
github.com/ipfs/bbloom v0.0.4
github.com/ipfs/go-blockservice v0.5.0
github.com/ipfs/go-cid v0.3.2
github.com/ipfs/go-cid v0.4.0
github.com/ipfs/go-cidutil v0.1.0
github.com/ipfs/go-datastore v0.6.0
github.com/ipfs/go-ds-badger2 v0.1.3
@@ -97,7 +97,7 @@ require (
github.com/ipfs/go-ipfs-util v0.0.2
github.com/ipfs/go-ipld-cbor v0.0.6
github.com/ipfs/go-ipld-format v0.4.0
github.com/ipfs/go-libipfs v0.5.0
github.com/ipfs/go-libipfs v0.7.0
github.com/ipfs/go-log/v2 v2.5.1
github.com/ipfs/go-merkledag v0.9.0
github.com/ipfs/go-metrics-interface v0.0.1
@@ -244,7 +244,7 @@ require (
github.com/ipfs/go-ipld-legacy v0.1.1 // indirect
github.com/ipfs/go-ipns v0.3.0 // indirect
github.com/ipfs/go-log v1.0.5 // indirect
github.com/ipfs/go-path v0.3.0 // indirect
github.com/ipfs/go-path v0.3.1 // indirect
github.com/ipfs/go-peertaskqueue v0.8.1 // indirect
github.com/ipfs/go-verifcid v0.0.2 // indirect
github.com/ipld/go-ipld-adl-hamt v0.0.0-20220616142416-9004dbd839e0 // indirect
+6 -9
View File
@@ -698,8 +698,8 @@ github.com/ipfs/go-cid v0.0.6/go.mod h1:6Ux9z5e+HpkQdckYoX1PG/6xqKspzlEIR5SDmgqg
github.com/ipfs/go-cid v0.0.7/go.mod h1:6Ux9z5e+HpkQdckYoX1PG/6xqKspzlEIR5SDmgqgC/I=
github.com/ipfs/go-cid v0.1.0/go.mod h1:rH5/Xv83Rfy8Rw6xG+id3DYAMUVmem1MowoKwdXmN2o=
github.com/ipfs/go-cid v0.2.0/go.mod h1:P+HXFDF4CVhaVayiEb4wkAy7zBHxBwsJyt0Y5U6MLro=
github.com/ipfs/go-cid v0.3.2 h1:OGgOd+JCFM+y1DjWPmVH+2/4POtpDzwcr7VgnB7mZXc=
github.com/ipfs/go-cid v0.3.2/go.mod h1:gQ8pKqT/sUxGY+tIwy1RPpAojYu7jAyCp5Tz1svoupw=
github.com/ipfs/go-cid v0.4.0 h1:a4pdZq0sx6ZSxbCizebnKiMCx/xI/aBBFlB73IgH4rA=
github.com/ipfs/go-cid v0.4.0/go.mod h1:uQHwDeX4c6CtyrFwdqyhpNcxVewur1M7l7fNU7LKwZk=
github.com/ipfs/go-cidutil v0.1.0 h1:RW5hO7Vcf16dplUU60Hs0AKDkQAVPVplr7lk97CFL+Q=
github.com/ipfs/go-cidutil v0.1.0/go.mod h1:e7OEVBMIv9JaOxt9zaGEmAoSlXW9jdFZ5lP/0PwcfpA=
github.com/ipfs/go-datastore v0.0.1/go.mod h1:d4KVXhMt913cLBEI/PXAy6ko+W7e9AhyAKBGh803qeE=
@@ -731,7 +731,6 @@ github.com/ipfs/go-ds-leveldb v0.5.0 h1:s++MEBbD3ZKc9/8/njrn4flZLnCuY9I79v94gBUN
github.com/ipfs/go-ds-leveldb v0.5.0/go.mod h1:d3XG9RUDzQ6V4SHi8+Xgj9j1XuEk1z82lquxrVbml/Q=
github.com/ipfs/go-ds-measure v0.2.0 h1:sG4goQe0KDTccHMyT45CY1XyUbxe5VwTKpg2LjApYyQ=
github.com/ipfs/go-ds-measure v0.2.0/go.mod h1:SEUD/rE2PwRa4IQEC5FuNAmjJCyYObZr9UvVh8V3JxE=
github.com/ipfs/go-fetcher v1.6.1/go.mod h1:27d/xMV8bodjVs9pugh/RCjjK2OZ68UgAMspMdingNo=
github.com/ipfs/go-filestore v1.2.0 h1:O2wg7wdibwxkEDcl7xkuQsPvJFRBVgVSsOJ/GP6z3yU=
github.com/ipfs/go-filestore v1.2.0/go.mod h1:HLJrCxRXquTeEEpde4lTLMaE/MYJZD7WHLkp9z6+FF8=
github.com/ipfs/go-fs-lock v0.0.6/go.mod h1:OTR+Rj9sHiRubJh3dRhD15Juhd/+w6VPOY28L7zESmM=
@@ -805,8 +804,8 @@ github.com/ipfs/go-ipld-legacy v0.1.1 h1:BvD8PEuqwBHLTKqlGFTHSwrwFOMkVESEvwIYwR2
github.com/ipfs/go-ipld-legacy v0.1.1/go.mod h1:8AyKFCjgRPsQFf15ZQgDB8Din4DML/fOmKZkkFkrIEg=
github.com/ipfs/go-ipns v0.3.0 h1:ai791nTgVo+zTuq2bLvEGmWP1M0A6kGTXUsgv/Yq67A=
github.com/ipfs/go-ipns v0.3.0/go.mod h1:3cLT2rbvgPZGkHJoPO1YMJeh6LtkxopCkKFcio/wE24=
github.com/ipfs/go-libipfs v0.5.0 h1:gtvzeoTdUHPUN4B5izzyBS3Cxtpvi+l7hd2mmjN+teM=
github.com/ipfs/go-libipfs v0.5.0/go.mod h1:iZ9QyhzNr3AkxRXrbQYb//rv7iLyvZJX0GNuc3lJDiQ=
github.com/ipfs/go-libipfs v0.7.0 h1:Mi54WJTODaOL2/ZSm5loi3SwI3jI2OuFWUrQIkJ5cpM=
github.com/ipfs/go-libipfs v0.7.0/go.mod h1:KsIf/03CqhICzyRGyGo68tooiBE2iFbI/rXW7FhAYr0=
github.com/ipfs/go-log v0.0.1/go.mod h1:kL1d2/hzSpI0thNYjiKfjanbVNU+IIGA/WnNESY9leM=
github.com/ipfs/go-log v1.0.0/go.mod h1:JO7RzlMK6rA+CIxFMLOuB6Wf5b81GDiKElL7UPSIKjA=
github.com/ipfs/go-log v1.0.1/go.mod h1:HuWlQttfN6FWNHRhlY5yMk/lW7evQC0HHGOxEwMRR8I=
@@ -838,17 +837,15 @@ github.com/ipfs/go-metrics-interface v0.0.1 h1:j+cpbjYvu4R8zbleSs36gvB7jR+wsL2fG
github.com/ipfs/go-metrics-interface v0.0.1/go.mod h1:6s6euYU4zowdslK0GKHmqaIZ3j/b/tL7HTWtJ4VPgWY=
github.com/ipfs/go-metrics-prometheus v0.0.2 h1:9i2iljLg12S78OhC6UAiXi176xvQGiZaGVF1CUVdE+s=
github.com/ipfs/go-metrics-prometheus v0.0.2/go.mod h1:ELLU99AQQNi+zX6GCGm2lAgnzdSH3u5UVlCdqSXnEks=
github.com/ipfs/go-path v0.3.0 h1:tkjga3MtpXyM5v+3EbRvOHEoo+frwi4oumw5K+KYWyA=
github.com/ipfs/go-path v0.3.0/go.mod h1:NOScsVgxfC/eIw4nz6OiGwK42PjaSJ4Y/ZFPn1Xe07I=
github.com/ipfs/go-path v0.3.1 h1:wkeaCWE/NTuuPGlEkLTsED5UkzfKYZpxaFFPgk8ZVLE=
github.com/ipfs/go-path v0.3.1/go.mod h1:eNLsxJEEMxn/CDzUJ6wuNl+6No6tEUhOZcPKsZsYX0E=
github.com/ipfs/go-peertaskqueue v0.1.0/go.mod h1:Jmk3IyCcfl1W3jTW3YpghSwSEC6IJ3Vzz/jUmWw8Z0U=
github.com/ipfs/go-peertaskqueue v0.7.0/go.mod h1:M/akTIE/z1jGNXMU7kFB4TeSEFvj68ow0Rrb04donIU=
github.com/ipfs/go-peertaskqueue v0.8.1 h1:YhxAs1+wxb5jk7RvS0LHdyiILpNmRIRnZVztekOF0pg=
github.com/ipfs/go-peertaskqueue v0.8.1/go.mod h1:Oxxd3eaK279FxeydSPPVGHzbwVeHjatZ2GA8XD+KbPU=
github.com/ipfs/go-unixfs v0.2.2-0.20190827150610-868af2e9e5cb/go.mod h1:IwAAgul1UQIcNZzKPYZWOCijryFBeCV79cNubPzol+k=
github.com/ipfs/go-unixfs v0.2.4/go.mod h1:SUdisfUjNoSDzzhGVxvCL9QO/nKdwXdr+gbMUdqcbYw=
github.com/ipfs/go-unixfs v0.4.3 h1:EdDc1sNZNFDUlo4UrVAvvAofVI5EwTnKu8Nv8mgXkWQ=
github.com/ipfs/go-unixfs v0.4.3/go.mod h1:TSG7G1UuT+l4pNj91raXAPkX0BhJi3jST1FDTfQ5QyM=
github.com/ipfs/go-unixfsnode v1.1.2/go.mod h1:5dcE2x03pyjHk4JjamXmunTMzz+VUtqvPwZjIEkfV6s=
github.com/ipfs/go-unixfsnode v1.5.2 h1:CvsiTt58W2uR5dD8bqQv+aAY0c1qolmXmSyNbPHYiew=
github.com/ipfs/go-unixfsnode v1.5.2/go.mod h1:NlOebRwYx8lMCNMdhAhEspYPBD3obp7TE0LvBqHY+ks=
github.com/ipfs/go-verifcid v0.0.1/go.mod h1:5Hrva5KBeIog4A+UpqlaIU+DEstipcJYQQZc0g37pY0=
+124
View File
@@ -0,0 +1,124 @@
package itests
import (
"context"
"os"
"sync"
"testing"
"time"
"github.com/stretchr/testify/require"
"github.com/filecoin-project/go-state-types/big"
"github.com/filecoin-project/go-state-types/exitcode"
lapi "github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/chain/index"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/itests/kit"
"github.com/filecoin-project/lotus/node"
)
func init() {
// adjust those to make tests snappy
index.CoalesceMinDelay = time.Millisecond
index.CoalesceMaxDelay = 10 * time.Millisecond
index.CoalesceMergeInterval = time.Millisecond
}
func testMsgIndex(
t *testing.T,
name string,
run func(t *testing.T, makeMsgIndex func(cs *store.ChainStore) (index.MsgIndex, error)),
check func(t *testing.T, i int, msgIndex index.MsgIndex),
) {
// create the message indices in the test context
var mx sync.Mutex
var tmpDirs []string
var msgIndices []index.MsgIndex
t.Cleanup(func() {
for _, msgIndex := range msgIndices {
_ = msgIndex.Close()
}
for _, tmp := range tmpDirs {
_ = os.RemoveAll(tmp)
}
})
makeMsgIndex := func(cs *store.ChainStore) (index.MsgIndex, error) {
var err error
tmp := t.TempDir()
msgIndex, err := index.NewMsgIndex(context.Background(), tmp, cs)
if err == nil {
mx.Lock()
tmpDirs = append(tmpDirs, tmp)
msgIndices = append(msgIndices, msgIndex)
mx.Unlock()
}
return msgIndex, err
}
t.Run(name, func(t *testing.T) {
run(t, makeMsgIndex)
})
if len(msgIndices) == 0 {
t.Fatal("no message indices")
}
for i, msgIndex := range msgIndices {
check(t, i, msgIndex)
}
}
func checkNonEmptyMsgIndex(t *testing.T, _ int, msgIndex index.MsgIndex) {
mi, ok := msgIndex.(interface{ CountMessages() (int64, error) })
if !ok {
t.Fatal("index does not allow counting")
}
count, err := mi.CountMessages()
require.NoError(t, err)
require.NotEqual(t, count, 0)
}
func TestMsgIndex(t *testing.T) {
testMsgIndex(t, "testSearchMsg", testSearchMsgWithIndex, checkNonEmptyMsgIndex)
}
func testSearchMsgWithIndex(t *testing.T, makeMsgIndex func(cs *store.ChainStore) (index.MsgIndex, error)) {
// copy of apiSuite.testSearchMsgWith; needs to be copied or else CI is angry, tests are built individually there
ctx := context.Background()
full, _, ens := kit.EnsembleMinimal(t, kit.ConstructorOpts(node.Override(new(index.MsgIndex), makeMsgIndex)))
senderAddr, err := full.WalletDefaultAddress(ctx)
require.NoError(t, err)
msg := &types.Message{
From: senderAddr,
To: senderAddr,
Value: big.Zero(),
}
ens.BeginMining(100 * time.Millisecond)
sm, err := full.MpoolPushMessage(ctx, msg, nil)
require.NoError(t, err)
//stm: @CHAIN_STATE_WAIT_MSG_001
res, err := full.StateWaitMsg(ctx, sm.Cid(), 1, lapi.LookbackNoLimit, true)
require.NoError(t, err)
require.Equal(t, exitcode.Ok, res.Receipt.ExitCode, "message not successful")
//stm: @CHAIN_STATE_SEARCH_MSG_001
searchRes, err := full.StateSearchMsg(ctx, types.EmptyTSK, sm.Cid(), lapi.LookbackNoLimit, true)
require.NoError(t, err)
require.NotNil(t, searchRes)
require.Equalf(t, res.TipSet, searchRes.TipSet, "search ts: %s, different from wait ts: %s", searchRes.TipSet, res.TipSet)
}
+17
View File
@@ -58,6 +58,9 @@ var (
ProtocolID, _ = tag.NewKey("proto")
Direction, _ = tag.NewKey("direction")
UseFD, _ = tag.NewKey("use_fd")
// vm execution
ExecutionLane, _ = tag.NewKey("lane")
)
// Measures
@@ -121,6 +124,8 @@ var (
VMApplyFlush = stats.Float64("vm/applyblocks_flush", "Time spent flushing vm state", stats.UnitMilliseconds)
VMSends = stats.Int64("vm/sends", "Counter for sends processed by the VM", stats.UnitDimensionless)
VMApplied = stats.Int64("vm/applied", "Counter for messages (including internal messages) processed by the VM", stats.UnitDimensionless)
VMExecutionWaiting = stats.Int64("vm/execution_waiting", "Counter for VM executions waiting to be assigned to a lane", stats.UnitDimensionless)
VMExecutionRunning = stats.Int64("vm/execution_running", "Counter for running VM executions", stats.UnitDimensionless)
// miner
WorkerCallsStarted = stats.Int64("sealing/worker_calls_started", "Counter of started worker tasks", stats.UnitDimensionless)
@@ -363,6 +368,16 @@ var (
Measure: VMApplied,
Aggregation: view.LastValue(),
}
VMExecutionWaitingView = &view.View{
Measure: VMExecutionWaiting,
Aggregation: view.LastValue(),
TagKeys: []tag.Key{ExecutionLane},
}
VMExecutionRunningView = &view.View{
Measure: VMExecutionRunning,
Aggregation: view.LastValue(),
TagKeys: []tag.Key{ExecutionLane},
}
// miner
WorkerCallsStartedView = &view.View{
@@ -727,6 +742,8 @@ var ChainNodeViews = append([]*view.View{
VMApplyFlushView,
VMSendsView,
VMAppliedView,
VMExecutionWaitingView,
VMExecutionRunningView,
}, DefaultViews...)
var MinerNodeViews = append([]*view.View{
+2
View File
@@ -24,6 +24,7 @@ import (
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/beacon"
"github.com/filecoin-project/lotus/chain/index"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/journal"
"github.com/filecoin-project/lotus/journal/alerting"
@@ -390,6 +391,7 @@ func Test() Option {
Unset(new(*peermgr.PeerMgr)),
Override(new(beacon.Schedule), testing.RandomBeacon),
Override(new(*storageadapter.DealPublisher), storageadapter.NewDealPublisher(nil, storageadapter.PublishMsgConfig{})),
Override(new(index.MsgIndex), modules.DummyMsgIndex),
)
}
+5
View File
@@ -20,6 +20,7 @@ import (
"github.com/filecoin-project/lotus/chain/events"
"github.com/filecoin-project/lotus/chain/exchange"
"github.com/filecoin-project/lotus/chain/gen/slashfilter"
"github.com/filecoin-project/lotus/chain/index"
"github.com/filecoin-project/lotus/chain/market"
"github.com/filecoin-project/lotus/chain/messagepool"
"github.com/filecoin-project/lotus/chain/messagesigner"
@@ -275,6 +276,10 @@ func ConfigFullNode(c interface{}) Option {
Override(new(full.EthEventAPI), &full.EthModuleDummy{}),
),
),
// enable message index for full node when configured by the user, otherwise use dummy.
If(cfg.Index.EnableMsgIndex, Override(new(index.MsgIndex), modules.MsgIndex)),
If(!cfg.Index.EnableMsgIndex, Override(new(index.MsgIndex), modules.DummyMsgIndex)),
)
}
+14
View File
@@ -461,6 +461,20 @@ Set to 0 to keep all mappings`,
Comment: ``,
},
{
Name: "Index",
Type: "IndexConfig",
Comment: ``,
},
},
"IndexConfig": []DocField{
{
Name: "EnableMsgIndex",
Type: "bool",
Comment: `EnableMsgIndex enables indexing of messages on chain.`,
},
},
"IndexProviderConfig": []DocField{
{
+6
View File
@@ -28,6 +28,7 @@ type FullNode struct {
Chainstore Chainstore
Cluster UserRaftConfig
Fevm FevmConfig
Index IndexConfig
}
// // Common
@@ -726,3 +727,8 @@ type Events struct {
// Set a timeout for subscription clients
// Set upper bound on index size
}
type IndexConfig struct {
// EnableMsgIndex enables indexing of messages on chain.
EnableMsgIndex bool
}
+13 -2
View File
@@ -6,6 +6,7 @@ import (
"encoding/json"
"errors"
"fmt"
"os"
"sort"
"strconv"
"sync"
@@ -888,9 +889,14 @@ func (a *EthModule) applyMessage(ctx context.Context, msg *types.Message, tsk ty
return nil, xerrors.Errorf("cannot get tipset: %w", err)
}
applyTsMessages := true
if os.Getenv("LOTUS_SKIP_APPLY_TS_MESSAGE_CALL_WITH_GAS") == "1" {
applyTsMessages = false
}
// Try calling until we find a height with no migration.
for {
res, err = a.StateManager.CallWithGas(ctx, msg, []types.ChainMsg{}, ts)
res, err = a.StateManager.CallWithGas(ctx, msg, []types.ChainMsg{}, ts, applyTsMessages)
if err != stmgr.ErrExpensiveFork {
break
}
@@ -959,10 +965,15 @@ func gasSearch(
high := msg.GasLimit
low := msg.GasLimit
applyTsMessages := true
if os.Getenv("LOTUS_SKIP_APPLY_TS_MESSAGE_CALL_WITH_GAS") == "1" {
applyTsMessages = false
}
canSucceed := func(limit int64) (bool, error) {
msg.GasLimit = limit
res, err := smgr.CallWithGas(ctx, &msg, priorMsgs, ts)
res, err := smgr.CallWithGas(ctx, &msg, priorMsgs, ts, applyTsMessages)
if err != nil {
return false, xerrors.Errorf("CallWithGas failed: %w", err)
}
+7 -13
View File
@@ -4,6 +4,7 @@ import (
"context"
"math"
"math/rand"
"os"
"sort"
lru "github.com/hashicorp/golang-lru/v2"
@@ -276,10 +277,15 @@ func gasEstimateCallWithGas(
priorMsgs = append(priorMsgs, m)
}
applyTsMessages := true
if os.Getenv("LOTUS_SKIP_APPLY_TS_MESSAGE_CALL_WITH_GAS") == "1" {
applyTsMessages = false
}
// Try calling until we find a height with no migration.
var res *api.InvocResult
for {
res, err = smgr.CallWithGas(ctx, &msg, priorMsgs, ts)
res, err = smgr.CallWithGas(ctx, &msg, priorMsgs, ts, applyTsMessages)
if err != stmgr.ErrExpensiveFork {
break
}
@@ -372,18 +378,6 @@ func gasEstimateGasLimit(
}
ret = (ret * int64(transitionalMulti*1024)) >> 10
// Special case for PaymentChannel collect, which is deleting actor
// We ignore errors in this special case since they CAN occur,
// and we just want to detect existing payment channel actors
st, err := smgr.ParentState(ts)
if err == nil {
act, err := st.GetActor(msg.To)
if err == nil && lbuiltin.IsPaymentChannelActor(act.Code) && msgIn.Method == builtin.MethodsPaych.Collect {
// add the refunded gas for DestroyActor back into the gas used
ret += 76e3
}
}
return ret, nil
}
+2 -1
View File
@@ -21,6 +21,7 @@ import (
"github.com/filecoin-project/lotus/chain/consensus/filcns"
"github.com/filecoin-project/lotus/chain/exchange"
"github.com/filecoin-project/lotus/chain/gen/slashfilter"
"github.com/filecoin-project/lotus/chain/index"
"github.com/filecoin-project/lotus/chain/messagepool"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/store"
@@ -123,7 +124,7 @@ func NetworkName(mctx helpers.MetricsCtx,
ctx := helpers.LifecycleCtx(mctx, lc)
sm, err := stmgr.NewStateManager(cs, tsexec, syscalls, us, nil, nil)
sm, err := stmgr.NewStateManager(cs, tsexec, syscalls, us, nil, nil, index.DummyMsgIndex)
if err != nil {
return "", err
}
+36
View File
@@ -0,0 +1,36 @@
package modules
import (
"context"
"go.uber.org/fx"
"github.com/filecoin-project/lotus/chain/index"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/node/modules/helpers"
"github.com/filecoin-project/lotus/node/repo"
)
func MsgIndex(lc fx.Lifecycle, mctx helpers.MetricsCtx, cs *store.ChainStore, r repo.LockedRepo) (index.MsgIndex, error) {
basePath, err := r.SqlitePath()
if err != nil {
return nil, err
}
msgIndex, err := index.NewMsgIndex(helpers.LifecycleCtx(mctx, lc), basePath, cs)
if err != nil {
return nil, err
}
lc.Append(fx.Hook{
OnStop: func(_ context.Context) error {
return msgIndex.Close()
},
})
return msgIndex, nil
}
func DummyMsgIndex() index.MsgIndex {
return index.DummyMsgIndex
}
+3 -2
View File
@@ -4,14 +4,15 @@ import (
"go.uber.org/fx"
"github.com/filecoin-project/lotus/chain/beacon"
"github.com/filecoin-project/lotus/chain/index"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/vm"
"github.com/filecoin-project/lotus/node/modules/dtypes"
)
func StateManager(lc fx.Lifecycle, cs *store.ChainStore, exec stmgr.Executor, sys vm.SyscallBuilder, us stmgr.UpgradeSchedule, b beacon.Schedule, metadataDs dtypes.MetadataDS) (*stmgr.StateManager, error) {
sm, err := stmgr.NewStateManager(cs, exec, sys, us, b, metadataDs)
func StateManager(lc fx.Lifecycle, cs *store.ChainStore, exec stmgr.Executor, sys vm.SyscallBuilder, us stmgr.UpgradeSchedule, b beacon.Schedule, metadataDs dtypes.MetadataDS, msgIndex index.MsgIndex) (*stmgr.StateManager, error) {
sm, err := stmgr.NewStateManager(cs, exec, sys, us, b, metadataDs, msgIndex)
if err != nil {
return nil, err
}
+4 -5
View File
@@ -31,7 +31,7 @@ const (
type LotusTraceEvent struct {
Type pubsub_pb.TraceEvent_Type `json:"type,omitempty"`
PeerID string `json:"peerID,omitempty"`
PeerID []byte `json:"peerID,omitempty"`
Timestamp *int64 `json:"timestamp,omitempty"`
PeerScore TraceEventPeerScore `json:"peerScore,omitempty"`
SourceAuth string `json:"sourceAuth,omitempty"`
@@ -46,7 +46,7 @@ type TopicScore struct {
}
type TraceEventPeerScore struct {
PeerID string `json:"peerID"`
PeerID []byte `json:"peerID"`
Score float64 `json:"score"`
AppSpecificScore float64 `json:"appSpecificScore"`
IPColocationFactor float64 `json:"ipColocationFactor"`
@@ -77,11 +77,11 @@ func (lt *lotusTracer) PeerScores(scores map[peer.ID]*pubsub.PeerScoreSnapshot)
evt := &LotusTraceEvent{
Type: *TraceEventPeerScores.Enum(),
PeerID: lt.pid.Pretty(),
Timestamp: &now,
PeerID: []byte(lt.pid),
SourceAuth: lt.sa,
PeerScore: TraceEventPeerScore{
PeerID: pid.Pretty(),
PeerID: []byte(pid),
Score: score.Score,
AppSpecificScore: score.AppSpecificScore,
IPColocationFactor: score.IPColocationFactor,
@@ -104,7 +104,6 @@ func (lt *lotusTracer) TraceLotusEvent(evt *LotusTraceEvent) {
log.Errorf("error while transporting peer scores: %s", err)
}
}
}
func (lt *lotusTracer) Trace(evt *pubsub_pb.TraceEvent) {
+2 -4
View File
@@ -30,16 +30,15 @@ func (ttt *testTracerTransport) Transport(evt TracerTransportEvent) error {
}
func TestTracer_PeerScores(t *testing.T) {
testTransport := NewTestTraceTransport(t, func(t *testing.T, evt TracerTransportEvent) {
require.Equal(t, peerIDA.Pretty(), evt.lotusTraceEvent.PeerID)
require.Equal(t, []byte(peerIDA), evt.lotusTraceEvent.PeerID)
require.Equal(t, "source-auth-token-test", evt.lotusTraceEvent.SourceAuth)
require.Equal(t, float64(32), evt.lotusTraceEvent.PeerScore.Score)
n := time.Now().UnixNano()
require.LessOrEqual(t, *evt.lotusTraceEvent.Timestamp, n)
require.Equal(t, peerIDA.Pretty(), evt.lotusTraceEvent.PeerScore.PeerID)
require.Equal(t, []byte(peerIDA), evt.lotusTraceEvent.PeerScore.PeerID)
require.Equal(t, 1, len(evt.lotusTraceEvent.PeerScore.Topics))
topic := evt.lotusTraceEvent.PeerScore.Topics[0]
@@ -85,7 +84,6 @@ func TestTracer_PubSubTrace(t *testing.T) {
PeerID: []byte(peerIDA),
Timestamp: &n,
})
}
func TestTracer_MultipleTransports(t *testing.T) {