distributed tipset workers

chain/actors/adt/diff/array.go

@@ -4,12 +4,17 @@ import (
 	"context"
 
 	"github.com/filecoin-project/go-amt-ipld/v3"
-	adt2 "github.com/filecoin-project/lily/chain/actors/adt"
 	"github.com/filecoin-project/lotus/chain/actors/adt"
+	"go.opentelemetry.io/otel"
+
+	adt2 "github.com/filecoin-project/lily/chain/actors/adt"
 )
 
 // Amt returns a set of changes that transform `preArr` into `curArr`. opts are applied to both `preArr` and `curArr`.
 func Amt(ctx context.Context, preArr, curArr adt2.Array, preStore, curStore adt.Store, amtOpts ...amt.Option) ([]*amt.Change, error) {
+	ctx, span := otel.Tracer("").Start(ctx, "Amt.Diff")
+	defer span.End()
+
 	preRoot, err := preArr.Root()
 	if err != nil {
 		return nil, err

chain/actors/adt/diff/map.go

@@ -4,12 +4,17 @@ import (
 	"context"
 
 	"github.com/filecoin-project/go-hamt-ipld/v3"
-	adt2 "github.com/filecoin-project/lily/chain/actors/adt"
 	"github.com/filecoin-project/lotus/chain/actors/adt"
+	"go.opentelemetry.io/otel"
+
+	adt2 "github.com/filecoin-project/lily/chain/actors/adt"
 )
 
 // Hamt returns a set of changes that transform `preMap` into `curMap`. opts are applied to both `preMap` and `curMap`.
 func Hamt(ctx context.Context, preMap, curMap adt2.Map, preStore, curStore adt.Store, hamtOpts ...hamt.Option) ([]*hamt.Change, error) {
+	ctx, span := otel.Tracer("").Start(ctx, "Hamt.Diff")
+	defer span.End()
+
 	preRoot, err := preMap.Root()
 	if err != nil {
 		return nil, err

chain/actors/builtin/miner/actor.go.template

@@ -113,12 +113,13 @@ type State interface {
 
     SectorsAmtBitwidth() int
 
-	// Diff helpers. Used by Diff* functions internally.
-	sectors() (adt.Array, error)
-	decodeSectorOnChainInfo(*cbg.Deferred) (SectorOnChainInfo, error)
-	precommits() (adt.Map, error)
-	decodeSectorPreCommitOnChainInfo(*cbg.Deferred) (SectorPreCommitOnChainInfo, error)
 	GetState() interface{}
+
+	// Diff helpers. Used by Diff* functions internally.
+	SectorArray() (adt.Array, error)
+	DecodeSectorOnChainInfo(*cbg.Deferred) (SectorOnChainInfo, error)
+	PreCommitMap() (adt.Map, error)
+	DecodeSectorPreCommitOnChainInfo(*cbg.Deferred) (SectorPreCommitOnChainInfo, error)
 }
 
 type Deadline interface {

chain/actors/builtin/miner/diff.go

@@ -22,12 +22,12 @@ import (
 func DiffPreCommits(ctx context.Context, store adt.Store, pre, cur State) (*PreCommitChanges, error) {
 	ctx, span := otel.Tracer("").Start(ctx, "DiffPreCommits")
 	defer span.End()
-	prep, err := pre.precommits()
+	prep, err := pre.PreCommitMap()
 	if err != nil {
 		return nil, err
 	}
 
-	curp, err := cur.precommits()
+	curp, err := cur.PreCommitMap()
 	if err != nil {
 		return nil, err
 	}
@@ -42,7 +42,7 @@ func DiffPreCommits(ctx context.Context, store adt.Store, pre, cur State) (*PreC
 	}
 
 	diffContainer := NewPreCommitDiffContainer(pre, cur)
-	if mapRequiresLegacyDiffing(pre, cur, preOpts, curOpts) {
+	if MapRequiresLegacyDiffing(pre, cur, preOpts, curOpts) {
 		if span.IsRecording() {
 			span.SetAttributes(attribute.String("diff", "slow"))
 		}
@@ -102,7 +102,7 @@ func (m *preCommitDiffContainer) AsKey(key string) (abi.Keyer, error) {
 }
 
 func (m *preCommitDiffContainer) Add(key string, val *cbg.Deferred) error {
-	sp, err := m.after.decodeSectorPreCommitOnChainInfo(val)
+	sp, err := m.after.DecodeSectorPreCommitOnChainInfo(val)
 	if err != nil {
 		return xerrors.Errorf("pre commit diff container add: %w", err)
 	}
@@ -115,7 +115,7 @@ func (m *preCommitDiffContainer) Modify(key string, from, to *cbg.Deferred) erro
 }
 
 func (m *preCommitDiffContainer) Remove(key string, val *cbg.Deferred) error {
-	sp, err := m.pre.decodeSectorPreCommitOnChainInfo(val)
+	sp, err := m.pre.DecodeSectorPreCommitOnChainInfo(val)
 	if err != nil {
 		return xerrors.Errorf("pre commit diff container remove: %w", err)
 	}
@@ -126,20 +126,20 @@ func (m *preCommitDiffContainer) Remove(key string, val *cbg.Deferred) error {
 func DiffSectors(ctx context.Context, store adt.Store, pre, cur State) (*SectorChanges, error) {
 	ctx, span := otel.Tracer("").Start(ctx, "DiffSectors")
 	defer span.End()
-	pres, err := pre.sectors()
+	pres, err := pre.SectorArray()
 	if err != nil {
 		return nil, err
 	}
 
-	curs, err := cur.sectors()
+	curs, err := cur.SectorArray()
 	if err != nil {
 		return nil, err
 	}
 
 	preBw := pre.SectorsAmtBitwidth()
 	curBw := cur.SectorsAmtBitwidth()
 	diffContainer := NewSectorDiffContainer(pre, cur)
-	if arrayRequiresLegacyDiffing(pre, cur, preBw, curBw) {
+	if ArrayRequiresLegacyDiffing(pre, cur, preBw, curBw) {
 		if span.IsRecording() {
 			span.SetAttributes(attribute.String("diff", "slow"))
 		}
@@ -191,7 +191,7 @@ type sectorDiffContainer struct {
 }
 
 func (m *sectorDiffContainer) Add(key uint64, val *cbg.Deferred) error {
-	si, err := m.after.decodeSectorOnChainInfo(val)
+	si, err := m.after.DecodeSectorOnChainInfo(val)
 	if err != nil {
 		return xerrors.Errorf("sector diff container add: %w", err)
 	}
@@ -200,12 +200,12 @@ func (m *sectorDiffContainer) Add(key uint64, val *cbg.Deferred) error {
 }
 
 func (m *sectorDiffContainer) Modify(key uint64, from, to *cbg.Deferred) error {
-	siFrom, err := m.pre.decodeSectorOnChainInfo(from)
+	siFrom, err := m.pre.DecodeSectorOnChainInfo(from)
 	if err != nil {
 		return xerrors.Errorf("sector diff container modify from: %w", err)
 	}
 
-	siTo, err := m.after.decodeSectorOnChainInfo(to)
+	siTo, err := m.after.DecodeSectorOnChainInfo(to)
 	if err != nil {
 		return xerrors.Errorf("sector diff container modify to: %w", err)
 	}
@@ -229,15 +229,15 @@ func (m *sectorDiffContainer) Modify(key uint64, from, to *cbg.Deferred) error {
 }
 
 func (m *sectorDiffContainer) Remove(key uint64, val *cbg.Deferred) error {
-	si, err := m.pre.decodeSectorOnChainInfo(val)
+	si, err := m.pre.DecodeSectorOnChainInfo(val)
 	if err != nil {
 		return xerrors.Errorf("sector diff container remove: %w", err)
 	}
 	m.Results.Removed = append(m.Results.Removed, si)
 	return nil
 }
 
-func arrayRequiresLegacyDiffing(pre, cur State, pOpts, cOpts int) bool {
+func ArrayRequiresLegacyDiffing(pre, cur State, pOpts, cOpts int) bool {
 	// amt/v3 cannot read amt/v2 nodes. Their Pointers struct has changed cbor marshalers.
 	if pre.Code().Equals(builtin0.StorageMinerActorCodeID) {
 		return true
@@ -258,7 +258,7 @@ func arrayRequiresLegacyDiffing(pre, cur State, pOpts, cOpts int) bool {
 	return false
 }
 
-func mapRequiresLegacyDiffing(pre, cur State, pOpts, cOpts *adt.MapOpts) bool {
+func MapRequiresLegacyDiffing(pre, cur State, pOpts, cOpts *adt.MapOpts) bool {
 	// hamt/v3 cannot read hamt/v2 nodes. Their Pointers struct has changed cbor marshalers.
 	if pre.Code().Equals(builtin0.StorageMinerActorCodeID) {
 		return true

chain/actors/builtin/miner/state.go.template

@@ -432,11 +432,11 @@ func (s *state{{.v}}) DeadlineCronActive() (bool, error) {
 	return {{if (ge .v 4)}}s.State.DeadlineCronActive{{else}}true{{end}}, nil{{if (lt .v 4)}} // always active in this version{{end}}
 }
 
-func (s *state{{.v}}) sectors() (adt.Array, error) {
+func (s *state{{.v}}) SectorArray() (adt.Array, error) {
 	return adt{{.v}}.AsArray(s.store, s.Sectors{{if (ge .v 3)}}, miner{{.v}}.SectorsAmtBitwidth{{end}})
 }
 
-func (s *state{{.v}}) decodeSectorOnChainInfo(val *cbg.Deferred) (SectorOnChainInfo, error) {
+func (s *state{{.v}}) DecodeSectorOnChainInfo(val *cbg.Deferred) (SectorOnChainInfo, error) {
 	var si miner{{.v}}.SectorOnChainInfo
 	err := si.UnmarshalCBOR(bytes.NewReader(val.Raw))
 	if err != nil {
@@ -446,11 +446,11 @@ func (s *state{{.v}}) decodeSectorOnChainInfo(val *cbg.Deferred) (SectorOnChainI
 	return fromV{{.v}}SectorOnChainInfo(si), nil
 }
 
-func (s *state{{.v}}) precommits() (adt.Map, error) {
+func (s *state{{.v}}) PreCommitMap() (adt.Map, error) {
 	return adt{{.v}}.AsMap(s.store, s.PreCommittedSectors{{if (ge .v 3)}}, builtin{{.v}}.DefaultHamtBitwidth{{end}})
 }
 
-func (s *state{{.v}}) decodeSectorPreCommitOnChainInfo(val *cbg.Deferred) (SectorPreCommitOnChainInfo, error) {
+func (s *state{{.v}}) DecodeSectorPreCommitOnChainInfo(val *cbg.Deferred) (SectorPreCommitOnChainInfo, error) {
 	var sp miner{{.v}}.SectorPreCommitOnChainInfo
 	err := sp.UnmarshalCBOR(bytes.NewReader(val.Raw))
 	if err != nil {