lotus/storage/sectorblocks/blocks.go
2019-11-06 20:00:04 +01:00

266 lines
5.9 KiB
Go

package sectorblocks
import (
"bytes"
"context"
"errors"
"io"
"sync"
"github.com/ipfs/go-cid"
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-datastore/namespace"
"github.com/ipfs/go-datastore/query"
blockstore "github.com/ipfs/go-ipfs-blockstore"
dshelp "github.com/ipfs/go-ipfs-ds-help"
files "github.com/ipfs/go-ipfs-files"
ipld "github.com/ipfs/go-ipld-format"
"github.com/ipfs/go-unixfs"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/lib/cborrpc"
"github.com/filecoin-project/lotus/lib/padreader"
"github.com/filecoin-project/lotus/lib/sectorbuilder"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/storage/sector"
)
type SealSerialization uint8
const (
SerializationUnixfs0 SealSerialization = 'u'
)
var dsPrefix = datastore.NewKey("/sealedblocks")
var imBlocksPrefix = datastore.NewKey("/intermediate")
var ErrNotFound = errors.New("not found")
type SectorBlocks struct {
*sector.Store
intermediate blockstore.Blockstore // holds intermediate nodes TODO: consider combining with the staging blockstore
unsealed *unsealedBlocks
keys datastore.Batching
keyLk sync.Mutex
}
func NewSectorBlocks(sectst *sector.Store, ds dtypes.MetadataDS, sb *sectorbuilder.SectorBuilder) *SectorBlocks {
sbc := &SectorBlocks{
Store: sectst,
intermediate: blockstore.NewBlockstore(namespace.Wrap(ds, imBlocksPrefix)),
keys: namespace.Wrap(ds, dsPrefix),
}
unsealed := &unsealedBlocks{ // TODO: untangle this
sb: sb,
unsealed: map[string][]byte{},
unsealing: map[string]chan struct{}{},
}
sbc.unsealed = unsealed
return sbc
}
type UnixfsReader interface {
files.File
// ReadBlock reads data from a single unixfs block. Data is nil
// for intermediate nodes
ReadBlock(context.Context) (data []byte, offset uint64, nd ipld.Node, err error)
}
type refStorer struct {
blockReader UnixfsReader
writeRef func(cid cid.Cid, pieceRef string, offset uint64, size uint64) error
intermediate blockstore.Blockstore
pieceRef string
remaining []byte
}
func (st *SectorBlocks) writeRef(cid cid.Cid, pieceRef string, offset uint64, size uint64) error {
st.keyLk.Lock() // TODO: make this multithreaded
defer st.keyLk.Unlock()
v, err := st.keys.Get(dshelp.CidToDsKey(cid))
if err == datastore.ErrNotFound {
err = nil
}
if err != nil {
return xerrors.Errorf("getting existing refs: %w", err)
}
var refs api.SealedRefs
if len(v) > 0 {
if err := cborrpc.ReadCborRPC(bytes.NewReader(v), &refs); err != nil {
return xerrors.Errorf("decoding existing refs: %w", err)
}
}
refs.Refs = append(refs.Refs, api.SealedRef{
Piece: pieceRef,
Offset: offset,
Size: size,
})
newRef, err := cborrpc.Dump(&refs)
if err != nil {
return xerrors.Errorf("serializing refs: %w", err)
}
return st.keys.Put(dshelp.CidToDsKey(cid), newRef) // TODO: batch somehow
}
func (r *refStorer) Read(p []byte) (n int, err error) {
offset := 0
if len(r.remaining) > 0 {
offset += len(r.remaining)
read := copy(p, r.remaining)
if read == len(r.remaining) {
r.remaining = nil
} else {
r.remaining = r.remaining[read:]
}
return read, nil
}
for {
data, offset, nd, err := r.blockReader.ReadBlock(context.TODO())
if err != nil {
if err == io.EOF {
return 0, io.EOF
}
return 0, xerrors.Errorf("reading block: %w", err)
}
if len(data) == 0 {
// TODO: batch
// TODO: GC
if err := r.intermediate.Put(nd); err != nil {
return 0, xerrors.Errorf("storing intermediate node: %w", err)
}
continue
}
if err := r.writeRef(nd.Cid(), r.pieceRef, offset, uint64(len(data))); err != nil {
return 0, xerrors.Errorf("writing ref: %w", err)
}
read := copy(p, data)
if read < len(data) {
r.remaining = data[read:]
}
// TODO: read multiple
return read, nil
}
}
func (st *SectorBlocks) AddUnixfsPiece(ref cid.Cid, r UnixfsReader, dealID uint64) (sectorID uint64, err error) {
size, err := r.Size()
if err != nil {
return 0, err
}
refst := &refStorer{
blockReader: r,
pieceRef: string(SerializationUnixfs0) + ref.String(),
writeRef: st.writeRef,
intermediate: st.intermediate,
}
pr, psize := padreader.New(r, uint64(size))
return st.Store.AddPiece(refst.pieceRef, psize, pr, dealID)
}
func (st *SectorBlocks) List() (map[cid.Cid][]api.SealedRef, error) {
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[cid.Cid][]api.SealedRef{}
for _, ent := range ents {
refCid, err := dshelp.DsKeyToCid(datastore.RawKey(ent.Key))
if err != nil {
return nil, err
}
var refs api.SealedRefs
if err := cborrpc.ReadCborRPC(bytes.NewReader(ent.Value), &refs); err != nil {
return nil, err
}
out[refCid] = refs.Refs
}
return out, nil
}
func (st *SectorBlocks) GetRefs(k cid.Cid) ([]api.SealedRef, error) { // TODO: track local sectors
ent, err := st.keys.Get(dshelp.CidToDsKey(k))
if err == datastore.ErrNotFound {
err = ErrNotFound
}
if err != nil {
return nil, err
}
var refs api.SealedRefs
if err := cborrpc.ReadCborRPC(bytes.NewReader(ent), &refs); err != nil {
return nil, err
}
return refs.Refs, nil
}
func (st *SectorBlocks) GetSize(k cid.Cid) (uint64, error) {
blk, err := st.intermediate.Get(k)
if err == blockstore.ErrNotFound {
refs, err := st.GetRefs(k)
if err != nil {
return 0, err
}
return uint64(refs[0].Size), nil
}
if err != nil {
return 0, err
}
nd, err := ipld.Decode(blk)
if err != nil {
return 0, err
}
fsn, err := unixfs.ExtractFSNode(nd)
if err != nil {
return 0, err
}
return fsn.FileSize(), nil
}
func (st *SectorBlocks) Has(k cid.Cid) (bool, error) {
// TODO: ensure sector is still there
return st.keys.Has(dshelp.CidToDsKey(k))
}
func (st *SectorBlocks) SealedBlockstore(approveUnseal func() error) *SectorBlockStore {
return &SectorBlockStore{
intermediate: st.intermediate,
sectorBlocks: st,
approveUnseal: approveUnseal,
}
}