paramfetch: check checksums in parallel

api/struct.go

@@ -129,8 +129,8 @@ type StorageMinerStruct struct {
 
 		StoreGarbageData func(context.Context) (uint64, error) `perm:"write"`
 
-		SectorsStatus     func(context.Context, uint64) (sectorbuilder.SectorSealingStatus, error) `perm:"read"`
-		SectorsList       func(context.Context) ([]uint64, error)                                  `perm:"read"`
+		SectorsStatus func(context.Context, uint64) (sectorbuilder.SectorSealingStatus, error) `perm:"read"`
+		SectorsList   func(context.Context) ([]uint64, error)                                  `perm:"read"`
 
 		SectorsRefs func(context.Context) (map[string][]SealedRef, error) `perm:"read"`
 	}

build/paramfetch.go

@@ -8,10 +8,13 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 
 	rice "github.com/GeertJohan/go.rice"
 	logging "github.com/ipfs/go-log"
 	"github.com/minio/blake2b-simd"
+	"go.uber.org/multierr"
+	"golang.org/x/xerrors"
 	pb "gopkg.in/cheggaaa/pb.v1"
 )
 
@@ -27,6 +30,13 @@ type paramFile struct {
 	SectorSize uint64 `json:"sector_size"`
 }
 
+type fetch struct {
+	wg      sync.WaitGroup
+	fetchLk sync.Mutex
+
+	errs []error
+}
+
 func GetParams(storage bool) error {
 	if err := os.Mkdir(paramdir, 0755); err != nil && !os.IsExist(err) {
 		return err
@@ -39,6 +49,8 @@ func GetParams(storage bool) error {
 		return err
 	}
 
+	ft := &fetch{}
+
 	for name, info := range params {
 		if !SupportedSectorSize(info.SectorSize) {
 			continue
@@ -47,35 +59,61 @@ func GetParams(storage bool) error {
 			continue
 		}
 
-		if err := maybeFetch(name, info); err != nil {
-			return err
-		}
+		ft.maybeFetchAsync(name, info)
 	}
 
-	return nil
+	return ft.wait()
 }
 
-func maybeFetch(name string, info paramFile) error {
-	path := filepath.Join(paramdir, name)
+func (ft *fetch) maybeFetchAsync(name string, info paramFile) {
+	ft.wg.Add(1)
+
+	go func() {
+		defer ft.wg.Done()
+
+		path := filepath.Join(paramdir, name)
+
+		err := ft.checkFile(path, info)
+		if !os.IsNotExist(err) && err != nil {
+			log.Warn(err)
+		}
+		if err == nil {
+			return
+		}
+
+		ft.fetchLk.Lock()
+		defer ft.fetchLk.Unlock()
+
+		if err := doFetch(path, info); err != nil {
+			ft.errs = append(ft.errs, xerrors.Errorf("fetching file %s: %w", path, err))
+		}
+	}()
+}
+
+func (ft *fetch) checkFile(path string, info paramFile) error {
 	f, err := os.Open(path)
-	if err == nil {
-		defer f.Close()
+	if err != nil {
+		return err
+	}
+	defer f.Close()
 
-		h := blake2b.New512()
-		if _, err := io.Copy(h, f); err != nil {
-			return err
-		}
-
-		sum := h.Sum(nil)
-		strSum := hex.EncodeToString(sum[:16])
-		if strSum == info.Digest {
-			return nil
-		}
-
-		log.Warnf("Checksum mismatch in param file %s, %s != %s", name, strSum, info.Digest)
+	h := blake2b.New512()
+	if _, err := io.Copy(h, f); err != nil {
+		return err
 	}
 
-	return doFetch(path, info)
+	sum := h.Sum(nil)
+	strSum := hex.EncodeToString(sum[:16])
+	if strSum == info.Digest {
+		return nil
+	}
+
+	return xerrors.Errorf("checksum mismatch in param file %s, %s != %s", path, strSum, info.Digest)
+}
+
+func (ft *fetch) wait() error {
+	ft.wg.Wait()
+	return multierr.Combine(ft.errs...)
 }
 
 func doFetch(out string, info paramFile) error {

chain/gen/utils.go

@@ -372,7 +372,7 @@ func MakeGenesisBlock(bs bstore.Blockstore, balances map[address.Address]types.B
 	log.Infof("Empty Genesis root: %s", emptyroot)
 
 	genesisticket := &types.Ticket{
-		VRFProof: []byte("vrf proof"),
+		VRFProof: []byte("vrf proof0000000vrf proof0000000"),
 	}
 
 	b := &types.BlockHeader{

chain/types/blockheader_test.go

@@ -27,7 +27,7 @@ func testBlockHeader(t testing.TB) *BlockHeader {
 		ElectionProof: []byte("cats won the election"),
 		Tickets: []*Ticket{
 			&Ticket{
-				VRFProof: []byte("vrf proof"),
+				VRFProof: []byte("vrf proof0000000vrf proof0000000"),
 			},
 		},
 		Parents:               []cid.Cid{c, c},

go.mod

@@ -79,6 +79,7 @@ require (
 	go.uber.org/dig v1.7.0 // indirect
 	go.uber.org/fx v1.9.0
 	go.uber.org/goleak v0.10.0 // indirect
+	go.uber.org/multierr v1.1.0
 	go.uber.org/zap v1.10.0
 	go4.org v0.0.0-20190313082347-94abd6928b1d // indirect
 	golang.org/x/crypto v0.0.0-20190829043050-9756ffdc2472 // indirect

node/modules/storageminer.go

@@ -181,7 +181,6 @@ func SealTicketGen(api api.FullNode) sector.TicketFn {
 			return nil, xerrors.Errorf("unexpected randomness len: %d (expected %d)", n, sectorbuilder.CommLen)
 		}
 
-
 		return &sectorbuilder.SealTicket{
 			BlockHeight: ts.Height(),
 			TicketBytes: tkt,

storage/sector/store.go

@@ -38,7 +38,7 @@ type TicketFn func(context.Context) (*sectorbuilder.SealTicket, error)
 type Store struct {
 	waitingLk sync.Mutex
 
-	sb *sectorbuilder.SectorBuilder
+	sb    *sectorbuilder.SectorBuilder
 	tktFn TicketFn
 
 	dealsLk sync.Mutex
@@ -54,7 +54,7 @@ type Store struct {
 func NewStore(sb *sectorbuilder.SectorBuilder, ds dtypes.MetadataDS, tktFn TicketFn) *Store {
 	return &Store{
 		sb:      sb,
-		tktFn:tktFn,
+		tktFn:   tktFn,
 		deals:   namespace.Wrap(ds, sectorDealsPrefix),
 		waiting: map[uint64]chan struct{}{},
 		closeCh: make(chan struct{}),