Merge pull request #1158 from filecoin-project/feat/par-pledge

sealing: Parallel CommP calc in pledge sector
This commit is contained in:
Whyrusleeping 2020-01-29 12:15:22 -08:00 committed by GitHub
commit cb318cf2e9
3 changed files with 64 additions and 6 deletions

View File

@ -6,6 +6,7 @@ import (
"io"
"math"
"math/rand"
"runtime"
sectorbuilder "github.com/filecoin-project/go-sectorbuilder"
"golang.org/x/xerrors"
@ -14,6 +15,17 @@ import (
"github.com/filecoin-project/lotus/chain/types"
)
func (m *Sealing) pledgeReader(size uint64, parts uint64) io.Reader {
piece := sectorbuilder.UserBytesForSectorSize((size/127 + size) / parts)
readers := make([]io.Reader, parts)
for i := range readers {
readers[i] = io.LimitReader(rand.New(rand.NewSource(42+int64(i))), int64(piece))
}
return io.MultiReader(readers...)
}
func (m *Sealing) pledgeSector(ctx context.Context, sectorID uint64, existingPieceSizes []uint64, sizes ...uint64) ([]Piece, error) {
if len(sizes) == 0 {
return nil, nil
@ -21,10 +33,7 @@ func (m *Sealing) pledgeSector(ctx context.Context, sectorID uint64, existingPie
deals := make([]actors.StorageDealProposal, len(sizes))
for i, size := range sizes {
release := m.sb.RateLimit()
commP, err := sectorbuilder.GeneratePieceCommitment(io.LimitReader(rand.New(rand.NewSource(42)), int64(size)), size)
release()
commP, err := m.fastPledgeCommitment(size, uint64(runtime.NumCPU()))
if err != nil {
return nil, err
}
@ -81,7 +90,7 @@ func (m *Sealing) pledgeSector(ctx context.Context, sectorID uint64, existingPie
out := make([]Piece, len(sizes))
for i, size := range sizes {
ppi, err := m.sb.AddPiece(size, sectorID, io.LimitReader(rand.New(rand.NewSource(42)), int64(size)), existingPieceSizes)
ppi, err := m.sb.AddPiece(size, sectorID, m.pledgeReader(size, uint64(runtime.NumCPU())), existingPieceSizes)
if err != nil {
return nil, err
}

View File

@ -1,7 +1,12 @@
package sealing
import (
"io"
"math/bits"
"math/rand"
"sync"
"github.com/hashicorp/go-multierror"
sectorbuilder "github.com/filecoin-project/go-sectorbuilder"
)
@ -42,6 +47,41 @@ func fillersFromRem(toFill uint64) ([]uint64, error) {
return out, nil
}
func (m *Sealing) fastPledgeCommitment(size uint64, parts uint64) (commP [sectorbuilder.CommLen]byte, err error) {
parts = 1 << bits.Len64(parts) // round down to nearest power of 2
piece := sectorbuilder.UserBytesForSectorSize(size / parts)
out := make([]sectorbuilder.PublicPieceInfo, parts)
var lk sync.Mutex
var wg sync.WaitGroup
wg.Add(int(parts))
for i := uint64(0); i < parts; i++ {
go func(i uint64) {
defer wg.Done()
commP, perr := sectorbuilder.GeneratePieceCommitment(io.LimitReader(rand.New(rand.NewSource(42+int64(i))), int64(piece)), piece)
lk.Lock()
if perr != nil {
err = multierror.Append(err, perr)
}
out[i] = sectorbuilder.PublicPieceInfo{
Size: piece,
CommP: commP,
}
lk.Unlock()
}(i)
}
wg.Wait()
if err != nil {
return [32]byte{}, err
}
return sectorbuilder.GenerateDataCommitment(m.sb.SectorSize(), out)
}
func (m *Sealing) ListSectors() ([]SectorInfo, error) {
var sectors []SectorInfo
if err := m.sectors.List(&sectors); err != nil {

View File

@ -1,6 +1,7 @@
package sealing
import (
"github.com/filecoin-project/lotus/storage/sbmock"
"testing"
"github.com/stretchr/testify/assert"
@ -42,5 +43,13 @@ func TestFillersFromRem(t *testing.T) {
ub = sectorbuilder.UserBytesForSectorSize(uint64(9) << i)
testFill(t, ub, []uint64{ub1, ub4})
}
}
func TestFastPledge(t *testing.T) {
sz := uint64(16 << 20)
s := Sealing{sb: sbmock.NewMockSectorBuilder(0, sz)}
if _, err := s.fastPledgeCommitment(sz, 5); err != nil {
t.Fatalf("%+v", err)
}
}