diff --git a/chain/store/index_test.go b/chain/store/index_test.go index 11ff4371f..89756a252 100644 --- a/chain/store/index_test.go +++ b/chain/store/index_test.go @@ -32,6 +32,7 @@ func TestIndexSeeks(t *testing.T) { nbs := blockstore.NewTemporarySync() cs := store.NewChainStore(nbs, nbs, syncds.MutexWrap(datastore.NewMapDatastore()), nil, nil) + defer cs.Close() //nolint:errcheck _, err = cs.Import(bytes.NewReader(gencar)) if err != nil { diff --git a/chain/store/store.go b/chain/store/store.go index bde179316..f4ce8112b 100644 --- a/chain/store/store.go +++ b/chain/store/store.go @@ -134,24 +134,30 @@ type ChainStore struct { evtTypes [1]journal.EventType journal journal.Journal + + cancelFn context.CancelFunc + wg sync.WaitGroup } // localbs is guaranteed to fail Get* if requested block isn't stored locally func NewChainStore(bs bstore.Blockstore, localbs bstore.Blockstore, ds dstore.Batching, vmcalls vm.SyscallBuilder, j journal.Journal) *ChainStore { - c, _ := lru.NewARC(DefaultMsgMetaCacheSize) - tsc, _ := lru.NewARC(DefaultTipSetCacheSize) + mmCache, _ := lru.NewARC(DefaultMsgMetaCacheSize) + tsCache, _ := lru.NewARC(DefaultTipSetCacheSize) if j == nil { j = journal.NilJournal() } + + ctx, cancel := context.WithCancel(context.Background()) cs := &ChainStore{ bs: bs, localbs: localbs, ds: ds, bestTips: pubsub.New(64), tipsets: make(map[abi.ChainEpoch][]cid.Cid), - mmCache: c, - tsCache: tsc, + mmCache: mmCache, + tsCache: tsCache, vmcalls: vmcalls, + cancelFn: cancel, journal: j, } @@ -191,19 +197,24 @@ func NewChainStore(bs bstore.Blockstore, localbs bstore.Blockstore, ds dstore.Ba } hcmetric := func(rev, app []*types.TipSet) error { - ctx := context.Background() for _, r := range app { - stats.Record(ctx, metrics.ChainNodeHeight.M(int64(r.Height()))) + stats.Record(context.Background(), metrics.ChainNodeHeight.M(int64(r.Height()))) } return nil } cs.reorgNotifeeCh = make(chan ReorgNotifee) - cs.reorgCh = cs.reorgWorker(context.TODO(), []ReorgNotifee{hcnf, hcmetric}) + cs.reorgCh = cs.reorgWorker(ctx, []ReorgNotifee{hcnf, hcmetric}) return cs } +func (cs *ChainStore) Close() error { + cs.cancelFn() + cs.wg.Wait() + return nil +} + func (cs *ChainStore) Load() error { head, err := cs.ds.Get(chainHeadKey) if err == dstore.ErrNotFound { @@ -383,7 +394,7 @@ func (cs *ChainStore) MaybeTakeHeavierTipSet(ctx context.Context, ts *types.TipS // particular tipset to carry out a benchmark, verification, etc. on a chain // segment. func (cs *ChainStore) ForceHeadSilent(_ context.Context, ts *types.TipSet) error { - log.Warnf("(!!!) forcing a new head silently; only use this only for testing; new head: %s", ts) + log.Warnf("(!!!) forcing a new head silently; new head: %s", ts) cs.heaviestLk.Lock() defer cs.heaviestLk.Unlock() @@ -406,7 +417,9 @@ func (cs *ChainStore) reorgWorker(ctx context.Context, initialNotifees []ReorgNo notifees := make([]ReorgNotifee, len(initialNotifees)) copy(notifees, initialNotifees) + cs.wg.Add(1) go func() { + defer cs.wg.Done() defer log.Warn("reorgWorker quit") for { diff --git a/chain/store/store_test.go b/chain/store/store_test.go index bef066024..5f1f336f3 100644 --- a/chain/store/store_test.go +++ b/chain/store/store_test.go @@ -71,6 +71,7 @@ func BenchmarkGetRandomness(b *testing.B) { } cs := store.NewChainStore(bs, bs, mds, nil, nil) + defer cs.Close() //nolint:errcheck b.ResetTimer() @@ -105,6 +106,7 @@ func TestChainExportImport(t *testing.T) { nbs := blockstore.NewTemporary() cs := store.NewChainStore(nbs, nbs, datastore.NewMapDatastore(), nil, nil) + defer cs.Close() //nolint:errcheck root, err := cs.Import(buf) if err != nil { @@ -139,6 +141,8 @@ func TestChainExportImportFull(t *testing.T) { nbs := blockstore.NewTemporary() cs := store.NewChainStore(nbs, nbs, datastore.NewMapDatastore(), nil, nil) + defer cs.Close() //nolint:errcheck + root, err := cs.Import(buf) if err != nil { t.Fatal(err) diff --git a/cmd/lotus-bench/import.go b/cmd/lotus-bench/import.go index c99bed158..f8752fc99 100644 --- a/cmd/lotus-bench/import.go +++ b/cmd/lotus-bench/import.go @@ -263,6 +263,8 @@ var importBenchCmd = &cli.Command{ metadataDs := datastore.NewMapDatastore() cs := store.NewChainStore(bs, bs, metadataDs, vm.Syscalls(verifier), nil) + defer cs.Close() //nolint:errcheck + stm := stmgr.NewStateManager(cs) startTime := time.Now() diff --git a/cmd/lotus-shed/balances.go b/cmd/lotus-shed/balances.go index 474dfe685..da1263408 100644 --- a/cmd/lotus-shed/balances.go +++ b/cmd/lotus-shed/balances.go @@ -189,6 +189,7 @@ var chainBalanceStateCmd = &cli.Command{ } cs := store.NewChainStore(bs, bs, mds, vm.Syscalls(ffiwrapper.ProofVerifier), nil) + defer cs.Close() //nolint:errcheck cst := cbor.NewCborStore(bs) store := adt.WrapStore(ctx, cst) @@ -409,6 +410,7 @@ var chainPledgeCmd = &cli.Command{ } cs := store.NewChainStore(bs, bs, mds, vm.Syscalls(ffiwrapper.ProofVerifier), nil) + defer cs.Close() //nolint:errcheck cst := cbor.NewCborStore(bs) store := adt.WrapStore(ctx, cst) diff --git a/cmd/lotus-shed/export.go b/cmd/lotus-shed/export.go index 42434f3d2..9b0703445 100644 --- a/cmd/lotus-shed/export.go +++ b/cmd/lotus-shed/export.go @@ -91,6 +91,8 @@ var exportChainCmd = &cli.Command{ } cs := store.NewChainStore(bs, bs, mds, nil, nil) + defer cs.Close() //nolint:errcheck + if err := cs.Load(); err != nil { return err } diff --git a/cmd/lotus-shed/genesis-verify.go b/cmd/lotus-shed/genesis-verify.go index e15a42374..20561eb5a 100644 --- a/cmd/lotus-shed/genesis-verify.go +++ b/cmd/lotus-shed/genesis-verify.go @@ -53,6 +53,7 @@ var genesisVerifyCmd = &cli.Command{ bs := blockstore.NewBlockstore(datastore.NewMapDatastore()) cs := store.NewChainStore(bs, bs, datastore.NewMapDatastore(), nil, nil) + defer cs.Close() //nolint:errcheck cf := cctx.Args().Get(0) f, err := os.Open(cf) diff --git a/cmd/lotus-shed/pruning.go b/cmd/lotus-shed/pruning.go index 76d283f6b..8728163c9 100644 --- a/cmd/lotus-shed/pruning.go +++ b/cmd/lotus-shed/pruning.go @@ -170,6 +170,8 @@ var stateTreePruneCmd = &cli.Command{ } cs := store.NewChainStore(bs, bs, mds, vm.Syscalls(ffiwrapper.ProofVerifier), nil) + defer cs.Close() //nolint:errcheck + if err := cs.Load(); err != nil { return fmt.Errorf("loading chainstore: %w", err) } diff --git a/cmd/lotus/daemon.go b/cmd/lotus/daemon.go index f49278a2b..1d13b4082 100644 --- a/cmd/lotus/daemon.go +++ b/cmd/lotus/daemon.go @@ -410,14 +410,6 @@ func ImportChain(r repo.Repo, fname string, snapshot bool) (err error) { return xerrors.Errorf("failed to open blockstore: %w", err) } - defer func() { - if c, ok := bs.(io.Closer); ok { - if err := c.Close(); err != nil { - log.Warnf("failed to close blockstore: %s", err) - } - } - }() - mds, err := lr.Datastore("/metadata") if err != nil { return err @@ -427,7 +419,9 @@ func ImportChain(r repo.Repo, fname string, snapshot bool) (err error) { if err != nil { return xerrors.Errorf("failed to open journal: %w", err) } + cst := store.NewChainStore(bs, bs, mds, vm.Syscalls(ffiwrapper.ProofVerifier), j) + defer cst.Close() //nolint:errcheck log.Infof("importing chain from %s...", fname) @@ -472,7 +466,7 @@ func ImportChain(r repo.Repo, fname string, snapshot bool) (err error) { } log.Infof("accepting %s as new head", ts.Cids()) - if err := cst.SetHead(ts); err != nil { + if err := cst.ForceHeadSilent(context.Background(), ts); err != nil { return err } diff --git a/conformance/driver.go b/conformance/driver.go index 91f461722..833d50d7b 100644 --- a/conformance/driver.go +++ b/conformance/driver.go @@ -91,6 +91,8 @@ func (d *Driver) ExecuteTipset(bs blockstore.Blockstore, ds ds.Batching, preroot sm = stmgr.NewStateManager(cs) ) + defer cs.Close() //nolint:errcheck + blocks := make([]store.BlockMessages, 0, len(tipset.Blocks)) for _, b := range tipset.Blocks { sb := store.BlockMessages{ diff --git a/go.mod b/go.mod index e97e1dbad..277af452a 100644 --- a/go.mod +++ b/go.mod @@ -68,7 +68,7 @@ require ( github.com/ipfs/go-filestore v1.0.0 github.com/ipfs/go-fs-lock v0.0.6 github.com/ipfs/go-graphsync v0.5.0 - github.com/ipfs/go-ipfs-blockstore v1.0.2 + github.com/ipfs/go-ipfs-blockstore v1.0.3-0.20201116142306-a33814d3b08f github.com/ipfs/go-ipfs-chunker v0.0.5 github.com/ipfs/go-ipfs-ds-help v1.0.0 github.com/ipfs/go-ipfs-exchange-interface v0.0.1 diff --git a/go.sum b/go.sum index f266b9c94..e6e56e37b 100644 --- a/go.sum +++ b/go.sum @@ -567,8 +567,8 @@ github.com/ipfs/go-ipfs-blockstore v0.1.4/go.mod h1:Jxm3XMVjh6R17WvxFEiyKBLUGr86 github.com/ipfs/go-ipfs-blockstore v1.0.0/go.mod h1:knLVdhVU9L7CC4T+T4nvGdeUIPAXlnd9zmXfp+9MIjU= github.com/ipfs/go-ipfs-blockstore v1.0.1 h1:fnuVj4XdZp4yExhd0CnUwAiMNJHiPnfInhiuwz4lW1w= github.com/ipfs/go-ipfs-blockstore v1.0.1/go.mod h1:MGNZlHNEnR4KGgPHM3/k8lBySIOK2Ve+0KjZubKlaOE= -github.com/ipfs/go-ipfs-blockstore v1.0.2 h1:Z8nUlBHK7wVKPKliQCQR9tLgUtz4J2QRbqFcJrqzM+E= -github.com/ipfs/go-ipfs-blockstore v1.0.2/go.mod h1:MGNZlHNEnR4KGgPHM3/k8lBySIOK2Ve+0KjZubKlaOE= +github.com/ipfs/go-ipfs-blockstore v1.0.3-0.20201116142306-a33814d3b08f h1:AQQb5zZj7KKTEFh9EaAUXc5Q+F7SbYkjfYogZnEzfUc= +github.com/ipfs/go-ipfs-blockstore v1.0.3-0.20201116142306-a33814d3b08f/go.mod h1:MGNZlHNEnR4KGgPHM3/k8lBySIOK2Ve+0KjZubKlaOE= github.com/ipfs/go-ipfs-blocksutil v0.0.1 h1:Eh/H4pc1hsvhzsQoMEP3Bke/aW5P5rVM1IWFJMcGIPQ= github.com/ipfs/go-ipfs-blocksutil v0.0.1/go.mod h1:Yq4M86uIOmxmGPUHv/uI7uKqZNtLb449gwKqXjIsnRk= github.com/ipfs/go-ipfs-chunker v0.0.1/go.mod h1:tWewYK0we3+rMbOh7pPFGDyypCtvGcBFymgY4rSDLAw= diff --git a/node/modules/chain.go b/node/modules/chain.go index fe0ebdfbb..095bb501c 100644 --- a/node/modules/chain.go +++ b/node/modules/chain.go @@ -111,13 +111,19 @@ func SetupFallbackBlockstore(cbs dtypes.ChainBlockstore, rem dtypes.ChainBitswap return nil } -func ChainStore(bs dtypes.ChainBlockstore, lbs dtypes.ChainRawBlockstore, ds dtypes.MetadataDS, syscalls vm.SyscallBuilder, j journal.Journal) *store.ChainStore { +func ChainStore(lc fx.Lifecycle, bs dtypes.ChainBlockstore, lbs dtypes.ChainRawBlockstore, ds dtypes.MetadataDS, syscalls vm.SyscallBuilder, j journal.Journal) *store.ChainStore { chain := store.NewChainStore(bs, lbs, ds, syscalls, j) if err := chain.Load(); err != nil { log.Warnf("loading chain state from disk: %s", err) } + lc.Append(fx.Hook{ + OnStop: func(_ context.Context) error { + return chain.Close() + }, + }) + return chain }