Merge remote-tracking branch 'origin/master' into next

chain/events/state/ctxstore.go

@@ -0,0 +1,25 @@
+package state
+
+import (
+	"context"
+
+	"github.com/ipfs/go-cid"
+	cbor "github.com/ipfs/go-ipld-cbor"
+)
+
+type contextStore struct {
+	ctx context.Context
+	cst *cbor.BasicIpldStore
+}
+
+func (cs *contextStore) Context() context.Context {
+	return cs.ctx
+}
+
+func (cs *contextStore) Get(ctx context.Context, c cid.Cid, out interface{}) error {
+	return cs.cst.Get(ctx, c, out)
+}
+
+func (cs *contextStore) Put(ctx context.Context, v interface{}) (cid.Cid, error) {
+	return cs.cst.Put(ctx, v)
+}

chain/events/state/diff_adt.go

@@ -0,0 +1,55 @@
+package state
+
+import (
+	"github.com/filecoin-project/specs-actors/actors/util/adt"
+	typegen "github.com/whyrusleeping/cbor-gen"
+)
+
+// AdtArrayDiff generalizes adt.Array diffing by accepting a Deferred type that can unmarshalled to its corresponding struct
+// in an interface implantation.
+// Add should be called when a new k,v is added to the array
+// Modify should be called when a value is modified in the array
+// Remove should be called when a value is removed from the array
+type AdtArrayDiff interface {
+	Add(key uint64, val *typegen.Deferred) error
+	Modify(key uint64, from, to *typegen.Deferred) error
+	Remove(key uint64, val *typegen.Deferred) error
+}
+
+// TODO Performance can be improved by diffing the underlying IPLD graph, e.g. https://github.com/ipfs/go-merkledag/blob/749fd8717d46b4f34c9ce08253070079c89bc56d/dagutils/diff.go#L104
+// CBOR Marshaling will likely be the largest performance bottleneck here.
+
+// DiffAdtArray accepts two *adt.Array's and an AdtArrayDiff implementation. It does the following:
+// - All values that exist in preArr and not in curArr are passed to AdtArrayDiff.Remove()
+// - All values that exist in curArr nnd not in prevArr are passed to adtArrayDiff.Add()
+// - All values that exist in preArr and in curArr are passed to AdtArrayDiff.Modify()
+//  - It is the responsibility of AdtArrayDiff.Modify() to determine if the values it was passed have been modified.
+func DiffAdtArray(preArr, curArr *adt.Array, out AdtArrayDiff) error {
+	prevVal := new(typegen.Deferred)
+	if err := preArr.ForEach(prevVal, func(i int64) error {
+		curVal := new(typegen.Deferred)
+		found, err := curArr.Get(uint64(i), curVal)
+		if err != nil {
+			return err
+		}
+		if !found {
+			if err := out.Remove(uint64(i), prevVal); err != nil {
+				return err
+			}
+			return nil
+		}
+
+		if err := out.Modify(uint64(i), prevVal, curVal); err != nil {
+			return err
+		}
+
+		return curArr.Delete(uint64(i))
+	}); err != nil {
+		return err
+	}
+
+	curVal := new(typegen.Deferred)
+	return curArr.ForEach(curVal, func(i int64) error {
+		return out.Add(uint64(i), curVal)
+	})
+}

chain/events/state/diff_adt_test.go

@@ -0,0 +1,155 @@
+package state
+
+import (
+	"bytes"
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	ds "github.com/ipfs/go-datastore"
+	ds_sync "github.com/ipfs/go-datastore/sync"
+	bstore "github.com/ipfs/go-ipfs-blockstore"
+	cbornode "github.com/ipfs/go-ipld-cbor"
+	typegen "github.com/whyrusleeping/cbor-gen"
+
+	"github.com/filecoin-project/specs-actors/actors/runtime"
+	"github.com/filecoin-project/specs-actors/actors/util/adt"
+)
+
+func TestDiffAdtArray(t *testing.T) {
+	ctxstoreA := newContextStore()
+	ctxstoreB := newContextStore()
+
+	arrA := adt.MakeEmptyArray(ctxstoreA)
+	arrB := adt.MakeEmptyArray(ctxstoreB)
+
+	require.NoError(t, arrA.Set(0, runtime.CBORBytes([]byte{0}))) // delete
+
+	require.NoError(t, arrA.Set(1, runtime.CBORBytes([]byte{0}))) // modify
+	require.NoError(t, arrB.Set(1, runtime.CBORBytes([]byte{1})))
+
+	require.NoError(t, arrA.Set(2, runtime.CBORBytes([]byte{1}))) // delete
+
+	require.NoError(t, arrA.Set(3, runtime.CBORBytes([]byte{0}))) // noop
+	require.NoError(t, arrB.Set(3, runtime.CBORBytes([]byte{0})))
+
+	require.NoError(t, arrA.Set(4, runtime.CBORBytes([]byte{0}))) // modify
+	require.NoError(t, arrB.Set(4, runtime.CBORBytes([]byte{6})))
+
+	require.NoError(t, arrB.Set(5, runtime.CBORBytes{8})) // add
+	require.NoError(t, arrB.Set(6, runtime.CBORBytes{9})) // add
+
+	changes := new(TestAdtDiff)
+
+	assert.NoError(t, DiffAdtArray(arrA, arrB, changes))
+	assert.NotNil(t, changes)
+
+	assert.Equal(t, 2, len(changes.Added))
+	// keys 5 and 6 were added
+	assert.EqualValues(t, uint64(5), changes.Added[0].key)
+	assert.EqualValues(t, []byte{8}, changes.Added[0].val)
+	assert.EqualValues(t, uint64(6), changes.Added[1].key)
+	assert.EqualValues(t, []byte{9}, changes.Added[1].val)
+
+	assert.Equal(t, 2, len(changes.Modified))
+	// keys 1 and 4 were modified
+	assert.EqualValues(t, uint64(1), changes.Modified[0].From.key)
+	assert.EqualValues(t, []byte{0}, changes.Modified[0].From.val)
+	assert.EqualValues(t, uint64(1), changes.Modified[0].To.key)
+	assert.EqualValues(t, []byte{1}, changes.Modified[0].To.val)
+	assert.EqualValues(t, uint64(4), changes.Modified[1].From.key)
+	assert.EqualValues(t, []byte{0}, changes.Modified[1].From.val)
+	assert.EqualValues(t, uint64(4), changes.Modified[1].To.key)
+	assert.EqualValues(t, []byte{6}, changes.Modified[1].To.val)
+
+	assert.Equal(t, 2, len(changes.Removed))
+	// keys 0 and 2 were deleted
+	assert.EqualValues(t, uint64(0), changes.Removed[0].key)
+	assert.EqualValues(t, []byte{0}, changes.Removed[0].val)
+	assert.EqualValues(t, uint64(2), changes.Removed[1].key)
+	assert.EqualValues(t, []byte{1}, changes.Removed[1].val)
+}
+
+type adtDiffResult struct {
+	key uint64
+	val runtime.CBORBytes
+}
+
+type TestAdtDiff struct {
+	Added    []adtDiffResult
+	Modified []TestAdtDiffModified
+	Removed  []adtDiffResult
+}
+
+var _ AdtArrayDiff = &TestAdtDiff{}
+
+type TestAdtDiffModified struct {
+	From adtDiffResult
+	To   adtDiffResult
+}
+
+func (t *TestAdtDiff) Add(key uint64, val *typegen.Deferred) error {
+	v := new(runtime.CBORBytes)
+	err := v.UnmarshalCBOR(bytes.NewReader(val.Raw))
+	if err != nil {
+		return err
+	}
+	t.Added = append(t.Added, adtDiffResult{
+		key: key,
+		val: *v,
+	})
+	return nil
+}
+
+func (t *TestAdtDiff) Modify(key uint64, from, to *typegen.Deferred) error {
+	vFrom := new(runtime.CBORBytes)
+	err := vFrom.UnmarshalCBOR(bytes.NewReader(from.Raw))
+	if err != nil {
+		return err
+	}
+
+	vTo := new(runtime.CBORBytes)
+	err = vTo.UnmarshalCBOR(bytes.NewReader(to.Raw))
+	if err != nil {
+		return err
+	}
+
+	if !bytes.Equal(*vFrom, *vTo) {
+		t.Modified = append(t.Modified, TestAdtDiffModified{
+			From: adtDiffResult{
+				key: key,
+				val: *vFrom,
+			},
+			To: adtDiffResult{
+				key: key,
+				val: *vTo,
+			},
+		})
+	}
+	return nil
+}
+
+func (t *TestAdtDiff) Remove(key uint64, val *typegen.Deferred) error {
+	v := new(runtime.CBORBytes)
+	err := v.UnmarshalCBOR(bytes.NewReader(val.Raw))
+	if err != nil {
+		return err
+	}
+	t.Removed = append(t.Removed, adtDiffResult{
+		key: key,
+		val: *v,
+	})
+	return nil
+}
+
+func newContextStore() *contextStore {
+	ctx := context.Background()
+	bs := bstore.NewBlockstore(ds_sync.MutexWrap(ds.NewMapDatastore()))
+	store := cbornode.NewCborStore(bs)
+	return &contextStore{
+		ctx: ctx,
+		cst: store,
+	}
+}

chain/events/state/predicates.go

@@ -1,10 +1,12 @@
 package state
 
 import (
+	"bytes"
 	"context"
 
 	"github.com/ipfs/go-cid"
 	cbor "github.com/ipfs/go-ipld-cbor"
+	typegen "github.com/whyrusleeping/cbor-gen"
 
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-amt-ipld/v2"
@@ -171,15 +173,59 @@ type MinerSectorChanges struct {
 	Removed  []miner.SectorOnChainInfo
 }
 
+var _ AdtArrayDiff = &MinerSectorChanges{}
+
 type SectorExtensions struct {
 	From miner.SectorOnChainInfo
 	To   miner.SectorOnChainInfo
 }
 
+func (m *MinerSectorChanges) Add(key uint64, val *typegen.Deferred) error {
+	si := new(miner.SectorOnChainInfo)
+	err := si.UnmarshalCBOR(bytes.NewReader(val.Raw))
+	if err != nil {
+		return err
+	}
+	m.Added = append(m.Added, *si)
+	return nil
+}
+
+func (m *MinerSectorChanges) Modify(key uint64, from, to *typegen.Deferred) error {
+	siFrom := new(miner.SectorOnChainInfo)
+	err := siFrom.UnmarshalCBOR(bytes.NewReader(from.Raw))
+	if err != nil {
+		return err
+	}
+
+	siTo := new(miner.SectorOnChainInfo)
+	err = siTo.UnmarshalCBOR(bytes.NewReader(to.Raw))
+	if err != nil {
+		return err
+	}
+
+	if siFrom.Expiration != siTo.Expiration {
+		m.Extended = append(m.Extended, SectorExtensions{
+			From: *siFrom,
+			To:   *siTo,
+		})
+	}
+	return nil
+}
+
+func (m *MinerSectorChanges) Remove(key uint64, val *typegen.Deferred) error {
+	si := new(miner.SectorOnChainInfo)
+	err := si.UnmarshalCBOR(bytes.NewReader(val.Raw))
+	if err != nil {
+		return err
+	}
+	m.Removed = append(m.Removed, *si)
+	return nil
+}
+
 func (sp *StatePredicates) OnMinerSectorChange() DiffMinerActorStateFunc {
 	return func(ctx context.Context, oldState, newState *miner.State) (changed bool, user UserData, err error) {
 		ctxStore := &contextStore{
-			ctx: context.TODO(),
+			ctx: ctx,
 			cst: sp.cst,
 		}
 
@@ -204,42 +250,7 @@ func (sp *StatePredicates) OnMinerSectorChange() DiffMinerActorStateFunc {
 			return false, nil, err
 		}
 
-		var osi miner.SectorOnChainInfo
+		if err := DiffAdtArray(oldSectors, newSectors, sectorChanges); err != nil {
-
-		// find all sectors that were extended or removed
-		if err := oldSectors.ForEach(&osi, func(i int64) error {
-			var nsi miner.SectorOnChainInfo
-			found, err := newSectors.Get(uint64(osi.SectorNumber), &nsi)
-			if err != nil {
-				return err
-			}
-			if !found {
-				sectorChanges.Removed = append(sectorChanges.Removed, osi)
-				return nil
-			}
-
-			if nsi.Expiration != osi.Expiration {
-				sectorChanges.Extended = append(sectorChanges.Extended, SectorExtensions{
-					From: osi,
-					To:   nsi,
-				})
-			}
-
-			// we don't update miners state filed with `newSectors.Root()` so this operation is safe.
-			if err := newSectors.Delete(uint64(osi.SectorNumber)); err != nil {
-				return err
-			}
-			return nil
-		}); err != nil {
-			return false, nil, err
-		}
-
-		// all sectors that remain in newSectors are new
-		var nsi miner.SectorOnChainInfo
-		if err := newSectors.ForEach(&nsi, func(i int64) error {
-			sectorChanges.Added = append(sectorChanges.Added, nsi)
-			return nil
-		}); err != nil {
 			return false, nil, err
 		}
 
@@ -251,20 +262,3 @@ func (sp *StatePredicates) OnMinerSectorChange() DiffMinerActorStateFunc {
 		return true, sectorChanges, nil
 	}
 }
-
-type contextStore struct {
-	ctx context.Context
-	cst *cbor.BasicIpldStore
-}
-
-func (cs *contextStore) Context() context.Context {
-	return cs.ctx
-}
-
-func (cs *contextStore) Get(ctx context.Context, c cid.Cid, out interface{}) error {
-	return cs.cst.Get(ctx, c, out)
-}
-
-func (cs *contextStore) Put(ctx context.Context, v interface{}) (cid.Cid, error) {
-	return cs.cst.Put(ctx, v)
-}

cmd/lotus-storage-miner/info.go

@@ -44,7 +44,7 @@ var infoCmd = &cli.Command{
 
 		ctx := lcli.ReqContext(cctx)
 
-		maddr, err := nodeApi.ActorAddress(ctx)
+		maddr, err := getActorAddress(ctx, nodeApi, cctx.String("actor"))
 		if err != nil {
 			return err
 		}

cmd/lotus-storage-miner/main.go

@@ -1,12 +1,16 @@
 package main
 
 import (
+	"context"
 	"os"
 
 	logging "github.com/ipfs/go-log/v2"
 	"github.com/urfave/cli/v2"
 	"go.opencensus.io/trace"
+	"golang.org/x/xerrors"
 
+	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/build"
 	lcli "github.com/filecoin-project/lotus/cli"
 	"github.com/filecoin-project/lotus/lib/lotuslog"
@@ -62,6 +66,12 @@ func main() {
 		Version:              build.UserVersion(),
 		EnableBashCompletion: true,
 		Flags: []cli.Flag{
+			&cli.StringFlag{
+				Name:    "actor",
+				Value:   "",
+				Usage:   "specify other actor to check state for (read only)",
+				Aliases: []string{"a"},
+			},
 			&cli.StringFlag{
 				Name:    "repo",
 				EnvVars: []string{"LOTUS_PATH"},
@@ -85,3 +95,19 @@ func main() {
 		os.Exit(1)
 	}
 }
+
+func getActorAddress(ctx context.Context, nodeAPI api.StorageMiner, overrideMaddr string) (maddr address.Address, err error) {
+	if overrideMaddr != "" {
+		maddr, err = address.NewFromString(overrideMaddr)
+		if err != nil {
+			return maddr, err
+		}
+	}
+
+	maddr, err = nodeAPI.ActorAddress(ctx)
+	if err != nil {
+		return maddr, xerrors.Errorf("getting actor address: %w", err)
+	}
+
+	return maddr, nil
+}

cmd/lotus-storage-miner/proving.go

@@ -51,9 +51,9 @@ var provingFaultsCmd = &cli.Command{
 
 		ctx := lcli.ReqContext(cctx)
 
-		maddr, err := nodeApi.ActorAddress(ctx)
+		maddr, err := getActorAddress(ctx, nodeApi, cctx.String("actor"))
 		if err != nil {
-			return xerrors.Errorf("getting actor address: %w", err)
+			return err
 		}
 
 		var mas miner.State
@@ -120,9 +120,9 @@ var provingInfoCmd = &cli.Command{
 
 		ctx := lcli.ReqContext(cctx)
 
-		maddr, err := nodeApi.ActorAddress(ctx)
+		maddr, err := getActorAddress(ctx, nodeApi, cctx.String("actor"))
 		if err != nil {
-			return xerrors.Errorf("getting actor address: %w", err)
+			return err
 		}
 
 		head, err := api.ChainHead(ctx)
@@ -140,11 +140,6 @@ var provingInfoCmd = &cli.Command{
 			return xerrors.Errorf("getting miner deadlines: %w", err)
 		}
 
-		curDeadlineSectors, err := deadlines.Due[cd.Index].Count()
-		if err != nil {
-			return xerrors.Errorf("counting deadline sectors: %w", err)
-		}
-
 		var mas miner.State
 		{
 			mact, err := api.StateGetActor(ctx, maddr, types.EmptyTSK)
@@ -203,7 +198,15 @@ var provingInfoCmd = &cli.Command{
 		fmt.Printf("New Sectors: %d\n\n", newSectors)
 
 		fmt.Printf("Deadline Index:       %d\n", cd.Index)
+
+		if cd.Index < uint64(len(deadlines.Due)) {
+			curDeadlineSectors, err := deadlines.Due[cd.Index].Count()
+			if err != nil {
+				return xerrors.Errorf("counting deadline sectors: %w", err)
+			}
 			fmt.Printf("Deadline Sectors:     %d\n", curDeadlineSectors)
+		}
+
 		fmt.Printf("Deadline Open:        %s\n", epochTime(cd.CurrentEpoch, cd.Open))
 		fmt.Printf("Deadline Close:       %s\n", epochTime(cd.CurrentEpoch, cd.Close))
 		fmt.Printf("Deadline Challenge:   %s\n", epochTime(cd.CurrentEpoch, cd.Challenge))
@@ -243,9 +246,9 @@ var provingDeadlinesCmd = &cli.Command{
 
 		ctx := lcli.ReqContext(cctx)
 
-		maddr, err := nodeApi.ActorAddress(ctx)
+		maddr, err := getActorAddress(ctx, nodeApi, cctx.String("actor"))
 		if err != nil {
-			return xerrors.Errorf("getting actor address: %w", err)
+			return err
 		}
 
 		deadlines, err := api.StateMinerDeadlines(ctx, maddr, types.EmptyTSK)

cmd/lotus/rpc.go

@@ -63,6 +63,7 @@ func serveRPC(a api.FullNode, stop node.StopFunc, addr multiaddr.Multiaddr, shut
 	srv := &http.Server{Handler: http.DefaultServeMux}
 
 	sigCh := make(chan os.Signal, 2)
+	shutdownDone := make(chan struct{})
 	go func() {
 		select {
 		case <-sigCh:
@@ -77,10 +78,17 @@ func serveRPC(a api.FullNode, stop node.StopFunc, addr multiaddr.Multiaddr, shut
 			log.Errorf("graceful shutting down failed: %s", err)
 		}
 		log.Warn("Graceful shutdown successful")
+		_ = log.Sync() //nolint:errcheck
+		close(shutdownDone)
 	}()
 	signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT)
 
-	return srv.Serve(manet.NetListener(lst))
+	err = srv.Serve(manet.NetListener(lst))
+	if err == http.ErrServerClosed {
+		<-shutdownDone
+		return nil
+	}
+	return err
 }
 
 func handleImport(a *impl.FullNodeAPI) func(w http.ResponseWriter, r *http.Request) {

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/Gurpartap/async v0.0.0-20180927173644-4f7f499dd9ee
 	github.com/StackExchange/wmi v0.0.0-20190523213315-cbe66965904d // indirect
 	github.com/coreos/go-systemd/v22 v22.0.0
+	github.com/dgraph-io/badger/v2 v2.0.3
 	github.com/docker/go-units v0.4.0
 	github.com/drand/drand v0.9.2-0.20200616080806-a94e9c1636a4
 	github.com/drand/kyber v1.1.0
@@ -47,7 +48,7 @@ require (
 	github.com/ipfs/go-cid v0.0.6
 	github.com/ipfs/go-cidutil v0.0.2
 	github.com/ipfs/go-datastore v0.4.4
-	github.com/ipfs/go-ds-badger2 v0.1.0
+	github.com/ipfs/go-ds-badger2 v0.1.1-0.20200708190120-187fc06f714e
 	github.com/ipfs/go-ds-leveldb v0.4.2
 	github.com/ipfs/go-ds-measure v0.1.0
 	github.com/ipfs/go-filestore v1.0.0

go.sum

@@ -505,6 +505,8 @@ github.com/ipfs/go-ds-badger v0.2.3 h1:J27YvAcpuA5IvZUbeBxOcQgqnYHUPxoygc6Qxxkod
 github.com/ipfs/go-ds-badger v0.2.3/go.mod h1:pEYw0rgg3FIrywKKnL+Snr+w/LjJZVMTBRn4FS6UHUk=
 github.com/ipfs/go-ds-badger2 v0.1.0 h1:784py6lXkwlVF+K6XSuqmdMgy5l8GI6k60ngBokb9Fg=
 github.com/ipfs/go-ds-badger2 v0.1.0/go.mod h1:pbR1p817OZbdId9EvLOhKBgUVTM3BMCSTan78lDDVaw=
+github.com/ipfs/go-ds-badger2 v0.1.1-0.20200708190120-187fc06f714e h1:Xi1nil8K2lBOorBS6Ys7+hmUCzH8fr3U9ipdL/IrcEI=
+github.com/ipfs/go-ds-badger2 v0.1.1-0.20200708190120-187fc06f714e/go.mod h1:lJnws7amT9Ehqzta0gwMrRsURU04caT0iRPr1W8AsOU=
 github.com/ipfs/go-ds-leveldb v0.0.1/go.mod h1:feO8V3kubwsEF22n0YRQCffeb79OOYIykR4L04tMOYc=
 github.com/ipfs/go-ds-leveldb v0.1.0/go.mod h1:hqAW8y4bwX5LWcCtku2rFNX3vjDZCy5LZCg+cSZvYb8=
 github.com/ipfs/go-ds-leveldb v0.4.1/go.mod h1:jpbku/YqBSsBc1qgME8BkWS4AxzF2cEu1Ii2r79Hh9s=

node/impl/client/client.go

@@ -102,7 +102,7 @@ func (a *API) ClientStartDeal(ctx context.Context, params *api.StartDealParams)
 
 	md, err := a.StateMinerProvingDeadline(ctx, params.Miner, types.EmptyTSK)
 	if err != nil {
-		return nil, xerrors.Errorf("failed getting peer ID: %w", err)
+		return nil, xerrors.Errorf("failed getting miner's deadline info: %w", err)
 	}
 
 	rt, err := ffiwrapper.SealProofTypeFromSectorSize(mi.SectorSize)

node/repo/fsrepo_ds.go

@@ -1,16 +1,18 @@
 package repo
 
 import (
+	"os"
+	"path/filepath"
+
 	"github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/mount"
 	"github.com/ipfs/go-datastore/namespace"
 	"golang.org/x/xerrors"
-	"os"
-	"path/filepath"
 
+	dgbadger "github.com/dgraph-io/badger/v2"
 	badger "github.com/ipfs/go-ds-badger2"
 	levelds "github.com/ipfs/go-ds-leveldb"
-	"github.com/ipfs/go-ds-measure"
+	measure "github.com/ipfs/go-ds-measure"
 	ldbopts "github.com/syndtr/goleveldb/leveldb/opt"
 )
 
@@ -25,7 +27,8 @@ var fsDatastores = map[string]func(path string) (datastore.Batching, error){
 
 func badgerDs(path string) (datastore.Batching, error) {
 	opts := badger.DefaultOptions
-	opts.Truncate = true
+	opts.Options = dgbadger.DefaultOptions("").WithTruncate(true).
+		WithValueThreshold(1 << 10)
 
 	return badger.NewDatastore(path, &opts)
 }