lotus/storage/sectorblocks/blocks.go

172 lines
3.9 KiB
Go
Raw Normal View History

2019-08-27 18:45:21 +00:00
package sectorblocks
import (
2019-11-06 12:22:08 +00:00
"bytes"
2019-08-27 18:45:21 +00:00
"context"
"encoding/binary"
"errors"
2019-11-06 17:38:42 +00:00
"io"
2019-08-27 18:45:21 +00:00
"sync"
2020-02-11 01:10:50 +00:00
"github.com/filecoin-project/go-padreader"
2020-02-08 02:18:32 +00:00
"github.com/filecoin-project/specs-actors/actors/abi"
2019-08-27 18:45:21 +00:00
"github.com/ipfs/go-datastore"
2019-11-06 12:22:08 +00:00
"github.com/ipfs/go-datastore/namespace"
"github.com/ipfs/go-datastore/query"
2019-08-27 18:45:21 +00:00
dshelp "github.com/ipfs/go-ipfs-ds-help"
2019-11-06 12:22:08 +00:00
"golang.org/x/xerrors"
2019-08-27 18:45:21 +00:00
cborutil "github.com/filecoin-project/go-cbor-util"
2020-02-11 01:10:50 +00:00
2020-01-07 14:00:10 +00:00
"github.com/filecoin-project/lotus/api"
2019-11-06 12:22:08 +00:00
"github.com/filecoin-project/lotus/node/modules/dtypes"
2019-11-07 18:22:59 +00:00
"github.com/filecoin-project/lotus/storage"
2019-08-27 18:45:21 +00:00
)
type SealSerialization uint8
const (
SerializationUnixfs0 SealSerialization = 'u'
)
var dsPrefix = datastore.NewKey("/sealedblocks")
var ErrNotFound = errors.New("not found")
2020-02-08 02:18:32 +00:00
func DealIDToDsKey(dealID abi.DealID) datastore.Key {
buf := make([]byte, binary.MaxVarintLen64)
2020-02-08 02:18:32 +00:00
size := binary.PutUvarint(buf, uint64(dealID))
return dshelp.NewKeyFromBinary(buf[:size])
}
func DsKeyToDealID(key datastore.Key) (uint64, error) {
buf, err := dshelp.BinaryFromDsKey(key)
if err != nil {
return 0, err
}
dealID, _ := binary.Uvarint(buf)
return dealID, nil
}
2019-08-27 18:45:21 +00:00
type SectorBlocks struct {
2019-11-07 18:22:59 +00:00
*storage.Miner
2019-08-27 18:45:21 +00:00
2019-12-01 17:58:31 +00:00
keys datastore.Batching
keyLk sync.Mutex
2019-08-27 18:45:21 +00:00
}
2020-03-03 22:19:22 +00:00
func NewSectorBlocks(miner *storage.Miner, ds dtypes.MetadataDS) *SectorBlocks {
2019-08-27 18:45:21 +00:00
sbc := &SectorBlocks{
2019-11-07 18:22:59 +00:00
Miner: miner,
keys: namespace.Wrap(ds, dsPrefix),
2019-08-27 18:45:21 +00:00
}
return sbc
}
2020-02-08 02:18:32 +00:00
func (st *SectorBlocks) writeRef(dealID abi.DealID, sectorID abi.SectorNumber, offset uint64, size abi.UnpaddedPieceSize) error {
2019-08-27 18:45:21 +00:00
st.keyLk.Lock() // TODO: make this multithreaded
defer st.keyLk.Unlock()
v, err := st.keys.Get(DealIDToDsKey(dealID))
2019-08-27 18:45:21 +00:00
if err == datastore.ErrNotFound {
err = nil
}
if err != nil {
2019-11-06 12:04:33 +00:00
return xerrors.Errorf("getting existing refs: %w", err)
2019-08-27 18:45:21 +00:00
}
2019-11-06 12:22:08 +00:00
var refs api.SealedRefs
2019-08-27 18:45:21 +00:00
if len(v) > 0 {
2019-11-07 14:11:39 +00:00
if err := cborutil.ReadCborRPC(bytes.NewReader(v), &refs); err != nil {
2019-11-06 12:04:33 +00:00
return xerrors.Errorf("decoding existing refs: %w", err)
2019-08-27 18:45:21 +00:00
}
}
2019-11-06 12:22:08 +00:00
refs.Refs = append(refs.Refs, api.SealedRef{
2019-12-01 17:58:31 +00:00
SectorID: sectorID,
Offset: offset,
Size: size,
2019-08-27 18:45:21 +00:00
})
2019-11-07 14:11:39 +00:00
newRef, err := cborutil.Dump(&refs)
2019-08-27 18:45:21 +00:00
if err != nil {
2019-11-06 12:04:33 +00:00
return xerrors.Errorf("serializing refs: %w", err)
2019-08-27 18:45:21 +00:00
}
return st.keys.Put(DealIDToDsKey(dealID), newRef) // TODO: batch somehow
2019-08-27 18:45:21 +00:00
}
2020-02-08 02:18:32 +00:00
func (st *SectorBlocks) AddPiece(ctx context.Context, size abi.UnpaddedPieceSize, r io.Reader, dealID abi.DealID) (sectorID abi.SectorNumber, err error) {
2019-12-01 17:58:31 +00:00
sectorID, pieceOffset, err := st.Miner.AllocatePiece(padreader.PaddedSize(uint64(size)))
if err != nil {
return 0, err
}
err = st.writeRef(dealID, sectorID, pieceOffset, size)
if err != nil {
return 0, err
}
2019-12-01 17:58:31 +00:00
return sectorID, st.Miner.SealPiece(ctx, size, r, sectorID, dealID)
2019-08-27 18:45:21 +00:00
}
func (st *SectorBlocks) List() (map[uint64][]api.SealedRef, error) {
2019-08-27 18:45:21 +00:00
res, err := st.keys.Query(query.Query{})
if err != nil {
return nil, err
}
ents, err := res.Rest()
if err != nil {
return nil, err
}
out := map[uint64][]api.SealedRef{}
2019-08-27 18:45:21 +00:00
for _, ent := range ents {
dealID, err := DsKeyToDealID(datastore.RawKey(ent.Key))
2019-08-27 18:45:21 +00:00
if err != nil {
return nil, err
}
2019-11-06 12:22:08 +00:00
var refs api.SealedRefs
2019-11-07 14:11:39 +00:00
if err := cborutil.ReadCborRPC(bytes.NewReader(ent.Value), &refs); err != nil {
2019-08-27 18:45:21 +00:00
return nil, err
}
out[dealID] = refs.Refs
2019-08-27 18:45:21 +00:00
}
return out, nil
}
2020-02-08 02:18:32 +00:00
func (st *SectorBlocks) GetRefs(dealID abi.DealID) ([]api.SealedRef, error) { // TODO: track local sectors
ent, err := st.keys.Get(DealIDToDsKey(dealID))
if err == datastore.ErrNotFound {
err = ErrNotFound
}
2019-08-27 18:45:21 +00:00
if err != nil {
return nil, err
}
2019-11-06 12:22:08 +00:00
var refs api.SealedRefs
2019-11-07 14:11:39 +00:00
if err := cborutil.ReadCborRPC(bytes.NewReader(ent), &refs); err != nil {
2019-08-27 18:45:21 +00:00
return nil, err
}
2019-11-06 12:22:08 +00:00
return refs.Refs, nil
2019-08-27 18:45:21 +00:00
}
2020-02-08 02:18:32 +00:00
func (st *SectorBlocks) GetSize(dealID abi.DealID) (uint64, error) {
refs, err := st.GetRefs(dealID)
if err != nil {
return 0, err
}
return uint64(refs[0].Size), nil
}
2020-02-08 02:18:32 +00:00
func (st *SectorBlocks) Has(dealID abi.DealID) (bool, error) {
2019-08-27 18:45:21 +00:00
// TODO: ensure sector is still there
return st.keys.Has(DealIDToDsKey(dealID))
2019-08-27 18:45:21 +00:00
}