lotus/extern/sector-storage/piece_provider_test.go

325 lines
9.5 KiB
Go
Raw Normal View History

2021-05-19 20:36:13 +00:00
package sectorstorage
import (
2021-05-21 05:30:17 +00:00
"bytes"
2021-05-19 20:36:13 +00:00
"context"
"io/ioutil"
2021-05-21 05:30:17 +00:00
"math/rand"
"net"
"net/http"
2021-05-19 20:36:13 +00:00
"testing"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/go-statestore"
2021-05-21 05:30:17 +00:00
"github.com/filecoin-project/lotus/extern/sector-storage/sealtasks"
2021-05-19 20:36:13 +00:00
specstorage "github.com/filecoin-project/specs-storage/storage"
2021-05-21 05:30:17 +00:00
"github.com/gorilla/mux"
"github.com/ipfs/go-cid"
2021-05-19 20:36:13 +00:00
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-datastore/namespace"
ds_sync "github.com/ipfs/go-datastore/sync"
2021-05-21 05:30:17 +00:00
logging "github.com/ipfs/go-log/v2"
2021-05-19 20:36:13 +00:00
"github.com/stretchr/testify/require"
"github.com/filecoin-project/lotus/extern/sector-storage/stores"
"github.com/filecoin-project/lotus/extern/sector-storage/storiface"
)
// TestPieceProviderReadPiece verifies that the ReadPiece method works correctly
2021-05-21 05:30:17 +00:00
// only uses miner and does NOT use any remote worker.
func TestPieceProviderSimpleNoRemoteWorker(t *testing.T) {
2021-05-19 20:36:13 +00:00
runTest := func(t *testing.T, alreadyUnsealed bool) {
// Set up sector storage manager
sealerCfg := SealerConfig{
ParallelFetchLimit: 10,
AllowAddPiece: true,
AllowPreCommit1: true,
AllowPreCommit2: true,
AllowCommit: true,
AllowUnseal: true,
}
2021-05-21 05:30:17 +00:00
ppt := newPieceProviderTestHarness(t, sealerCfg, abi.RegisteredSealProof_StackedDrg8MiBV1)
defer ppt.shutdown(t)
2021-05-19 20:36:13 +00:00
2021-05-21 05:30:17 +00:00
// Create some padded data that aligns with the piece boundaries.
pieceData := generatePieceData(8 * 127 * 1024 * 8)
2021-05-19 20:36:13 +00:00
size := abi.UnpaddedPieceSize(len(pieceData))
2021-05-21 05:30:17 +00:00
ppt.addPiece(t, pieceData)
2021-05-19 20:36:13 +00:00
// pre-commit 1
2021-05-21 05:30:17 +00:00
preCommit1 := ppt.preCommit1(t)
2021-05-19 20:36:13 +00:00
// pre-commit 2
2021-05-21 05:30:17 +00:00
ppt.preCommit2(t, preCommit1)
2021-05-19 20:36:13 +00:00
// If we want to test what happens when the data must be unsealed
// (ie there is not an unsealed copy already available)
if !alreadyUnsealed {
// Remove the unsealed copy from local storage
2021-05-21 05:30:17 +00:00
ppt.removeAllUnsealedSectorFiles(t)
2021-05-19 20:36:13 +00:00
}
// Read the piece
2021-05-21 05:30:17 +00:00
ppt.readPiece(t, storiface.UnpaddedByteIndex(0), size,
!alreadyUnsealed, pieceData)
2021-05-19 20:36:13 +00:00
}
t.Run("already unsealed", func(t *testing.T) {
runTest(t, true)
})
t.Run("requires unseal", func(t *testing.T) {
runTest(t, false)
})
}
2021-05-21 05:30:17 +00:00
func TestReadPieceRemoteWorkers(t *testing.T) {
logging.SetAllLoggers(logging.LevelDebug)
// miner's worker can only add pieces to an unsealed sector.
sealerCfg := SealerConfig{
ParallelFetchLimit: 10,
AllowAddPiece: true,
AllowPreCommit1: false,
AllowPreCommit2: false,
AllowCommit: false,
AllowUnseal: false,
}
// test harness for an 8M sector.
ppt := newPieceProviderTestHarness(t, sealerCfg, abi.RegisteredSealProof_StackedDrg8MiBV1)
defer ppt.shutdown(t)
// worker 2 will ONLY help with the sealing by first fetching
// the unsealed file from the miner.
ppt.addRemoteWorker(t, []sealtasks.TaskType{
sealtasks.TTPreCommit1, sealtasks.TTPreCommit2, sealtasks.TTCommit1,
sealtasks.TTFetch, sealtasks.TTFinalize,
})
// create a worker that can ONLY unseal and fetch
ppt.addRemoteWorker(t, []sealtasks.TaskType{
sealtasks.TTUnseal, sealtasks.TTFetch,
})
// run the test
// add one piece that aligns with the padding/piece boundaries.
pd1 := generatePieceData(8 * 127 * 4 * 1024)
pi1 := ppt.addPiece(t, pd1)
pd1size := pi1.Size.Unpadded()
pd2 := generatePieceData(8 * 127 * 4 * 1024)
pi2 := ppt.addPiece(t, pd2)
pd2size := pi2.Size.Unpadded()
// pre-commit 1
pC1 := ppt.preCommit1(t)
// pre-commit 2
ppt.preCommit2(t, pC1)
// finalize the sector so we declare to the index we have the sealed file
// so the unsealing worker can later look it up and fetch it if needed
// sending nil here will remove all unsealed files after sector is finalized.
ppt.finalizeSector(t, nil)
// Read the piece -> have to unseal since we removed the file.
ppt.readPiece(t, storiface.UnpaddedByteIndex(0), pd1size,
true, pd1)
// Read the same piece again -> will NOT have to unseal.
ppt.readPiece(t, storiface.UnpaddedByteIndex(0), pd1size, false, pd1)
// remove the unsealed file and read again -> will have to unseal.
ppt.removeAllUnsealedSectorFiles(t)
ppt.readPiece(t, storiface.UnpaddedByteIndex(0), pd1size,
true, pd1)
// Read Piece 2 -> no unsealing as it got unsealed above.
ppt.readPiece(t, storiface.UnpaddedByteIndex(pd1size), pd2size, false, pd2)
// remove all unseal files -> Read Piece 2 -> will have to Unseal.
ppt.removeAllUnsealedSectorFiles(t)
ppt.readPiece(t, storiface.UnpaddedByteIndex(pd1size), pd2size, true, pd2)
}
type pieceProviderTestHarness struct {
ctx context.Context
index *stores.Index
pp PieceProvider
sector specstorage.SectorRef
mgr *Manager
ticket abi.SealRandomness
commD cid.Cid
localStores []*stores.Local
servers []*http.Server
addedPieces []abi.PieceInfo
}
func generatePieceData(size uint64) []byte {
bz := make([]byte, size)
rand.Read(bz)
return bz
}
func newPieceProviderTestHarness(t *testing.T, mgrConfig SealerConfig, sectorProofType abi.RegisteredSealProof) *pieceProviderTestHarness {
ctx := context.Background()
// listen on tcp socket to create an http server later
address := "0.0.0.0:0"
nl, err := net.Listen("tcp", address)
require.NoError(t, err)
// create index, storage, local store & remote store.
index := stores.NewIndex()
storage := newTestStorage(t)
localStore, err := stores.NewLocal(ctx, storage, index, []string{"http://" + nl.Addr().String() + "/remote"})
require.NoError(t, err)
remoteStore := stores.NewRemote(localStore, index, nil, 6000, &stores.DefaultPartialFileHandler{})
// data stores for state tracking.
dstore := ds_sync.MutexWrap(datastore.NewMapDatastore())
wsts := statestore.New(namespace.Wrap(dstore, datastore.NewKey("/worker/calls")))
smsts := statestore.New(namespace.Wrap(dstore, datastore.NewKey("/stmgr/calls")))
mgr, err := New(ctx, localStore, remoteStore, storage, index, mgrConfig, wsts, smsts)
require.NoError(t, err)
// start a http server on the manager to serve sector file requests.
svc := &http.Server{
Addr: nl.Addr().String(),
Handler: mgr,
}
go func() {
_ = svc.Serve(nl)
}()
pp := NewPieceProvider(remoteStore, index, mgr)
sector := specstorage.SectorRef{
ID: abi.SectorID{
Miner: 100,
Number: 10,
},
ProofType: sectorProofType,
}
ticket := abi.SealRandomness{9, 9, 9, 9, 9, 9, 9, 9}
ppt := &pieceProviderTestHarness{
ctx: ctx,
index: index,
pp: pp,
sector: sector,
mgr: mgr,
ticket: ticket,
}
ppt.servers = append(ppt.servers, svc)
ppt.localStores = append(ppt.localStores, localStore)
return ppt
}
func (p *pieceProviderTestHarness) addRemoteWorker(t *testing.T, tasks []sealtasks.TaskType) {
// start an http Server
address := "0.0.0.0:0"
nl, err := net.Listen("tcp", address)
require.NoError(t, err)
localStore, err := stores.NewLocal(p.ctx, newTestStorage(t), p.index, []string{"http://" + nl.Addr().String() + "/remote"})
require.NoError(t, err)
fh := &stores.FetchHandler{
Local: localStore,
PfHandler: &stores.DefaultPartialFileHandler{},
}
mux := mux.NewRouter()
mux.PathPrefix("/remote").HandlerFunc(fh.ServeHTTP)
svc := &http.Server{
Addr: nl.Addr().String(),
Handler: mux,
}
go func() {
_ = svc.Serve(nl)
}()
remote := stores.NewRemote(localStore, p.index, nil, 1000,
&stores.DefaultPartialFileHandler{})
dstore := ds_sync.MutexWrap(datastore.NewMapDatastore())
csts := statestore.New(namespace.Wrap(dstore, datastore.NewKey("/stmgr/calls")))
worker := newLocalWorker(nil, WorkerConfig{
TaskTypes: tasks,
}, remote, localStore, p.index, p.mgr, csts)
p.servers = append(p.servers, svc)
p.localStores = append(p.localStores, localStore)
// register self with manager
require.NoError(t, p.mgr.AddWorker(p.ctx, worker))
}
func (p *pieceProviderTestHarness) removeAllUnsealedSectorFiles(t *testing.T) {
for i := range p.localStores {
ls := p.localStores[i]
require.NoError(t, ls.Remove(p.ctx, p.sector.ID, storiface.FTUnsealed, false))
}
}
func (p *pieceProviderTestHarness) addPiece(t *testing.T, pieceData []byte) abi.PieceInfo {
var existing []abi.UnpaddedPieceSize
for _, pi := range p.addedPieces {
existing = append(existing, pi.Size.Unpadded())
}
size := abi.UnpaddedPieceSize(len(pieceData))
pieceInfo, err := p.mgr.AddPiece(p.ctx, p.sector, existing, size, bytes.NewReader(pieceData))
require.NoError(t, err)
p.addedPieces = append(p.addedPieces, pieceInfo)
return pieceInfo
}
func (p *pieceProviderTestHarness) preCommit1(t *testing.T) specstorage.PreCommit1Out {
preCommit1, err := p.mgr.SealPreCommit1(p.ctx, p.sector, p.ticket, p.addedPieces)
require.NoError(t, err)
return preCommit1
}
func (p *pieceProviderTestHarness) preCommit2(t *testing.T, pc1 specstorage.PreCommit1Out) {
sectorCids, err := p.mgr.SealPreCommit2(p.ctx, p.sector, pc1)
require.NoError(t, err)
commD := sectorCids.Unsealed
p.commD = commD
}
func (p *pieceProviderTestHarness) readPiece(t *testing.T, offset storiface.UnpaddedByteIndex, size abi.UnpaddedPieceSize,
expectedHadToUnseal bool, expectedBytes []byte) {
rd, isUnsealed, err := p.pp.ReadPiece(p.ctx, p.sector, offset, size, p.ticket, p.commD)
require.NoError(t, err)
require.NotNil(t, rd)
require.Equal(t, expectedHadToUnseal, isUnsealed)
defer func() { _ = rd.Close() }()
// Make sure the input matches the output
readData, err := ioutil.ReadAll(rd)
require.NoError(t, err)
require.Equal(t, expectedBytes, readData)
}
func (p *pieceProviderTestHarness) finalizeSector(t *testing.T, keepUnseal []specstorage.Range) {
require.NoError(t, p.mgr.FinalizeSector(p.ctx, p.sector, keepUnseal))
}
func (p *pieceProviderTestHarness) shutdown(t *testing.T) {
for _, svc := range p.servers {
s := svc
require.NoError(t, s.Shutdown(p.ctx))
}
}