lotus/storage/sectorblocks/blocks.go

257 lines
5.5 KiB
Go
Raw Normal View History

2019-08-27 18:45:21 +00:00
package sectorblocks
import (
"context"
"errors"
2019-08-27 18:45:21 +00:00
"github.com/filecoin-project/go-lotus/api"
"github.com/filecoin-project/go-lotus/lib/sectorbuilder"
"github.com/filecoin-project/go-lotus/node/modules/dtypes"
"github.com/ipfs/go-datastore/namespace"
"github.com/ipfs/go-datastore/query"
2019-08-27 23:03:22 +00:00
blockstore "github.com/ipfs/go-ipfs-blockstore"
2019-08-27 18:45:21 +00:00
ipld "github.com/ipfs/go-ipld-format"
"github.com/ipfs/go-unixfs"
2019-08-27 18:45:21 +00:00
"sync"
"github.com/ipfs/go-cid"
"github.com/ipfs/go-datastore"
dshelp "github.com/ipfs/go-ipfs-ds-help"
files "github.com/ipfs/go-ipfs-files"
cbor "github.com/ipfs/go-ipld-cbor"
"github.com/filecoin-project/go-lotus/storage/sector"
)
type SealSerialization uint8
const (
SerializationUnixfs0 SealSerialization = 'u'
)
var dsPrefix = datastore.NewKey("/sealedblocks")
2019-08-27 23:03:22 +00:00
var imBlocksPrefix = datastore.NewKey("/intermediate")
2019-08-27 18:45:21 +00:00
var ErrNotFound = errors.New("not found")
2019-08-27 18:45:21 +00:00
type SectorBlocks struct {
*sector.Store
2019-08-27 23:03:22 +00:00
intermediate blockstore.Blockstore // holds intermediate nodes TODO: consider combining with the staging blockstore
2019-08-27 18:45:21 +00:00
unsealed *unsealedBlocks
2019-08-27 19:54:39 +00:00
keys datastore.Batching
keyLk sync.Mutex
2019-08-27 18:45:21 +00:00
}
func NewSectorBlocks(sectst *sector.Store, ds dtypes.MetadataDS, sb *sectorbuilder.SectorBuilder) *SectorBlocks {
sbc := &SectorBlocks{
Store: sectst,
2019-08-27 23:03:22 +00:00
intermediate: blockstore.NewBlockstore(namespace.Wrap(ds, imBlocksPrefix)),
keys: namespace.Wrap(ds, dsPrefix),
2019-08-27 18:45:21 +00:00
}
unsealed := &unsealedBlocks{ // TODO: untangle this
2019-08-27 23:03:22 +00:00
sb: sb,
2019-08-27 19:54:39 +00:00
unsealed: map[string][]byte{},
2019-08-27 18:45:21 +00:00
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 {
2019-08-27 23:03:22 +00:00
blockReader UnixfsReader
writeRef func(cid cid.Cid, pieceRef string, offset uint64, size uint32) error
2019-08-27 23:03:22 +00:00
intermediate blockstore.Blockstore
2019-08-27 18:45:21 +00:00
pieceRef string
remaining []byte
}
func (st *SectorBlocks) writeRef(cid cid.Cid, pieceRef string, offset uint64, size uint32) 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(dshelp.CidToDsKey(cid))
if err == datastore.ErrNotFound {
err = nil
}
if err != nil {
return err
}
var refs []api.SealedRef
if len(v) > 0 {
if err := cbor.DecodeInto(v, &refs); err != nil {
return err
}
}
refs = append(refs, api.SealedRef{
Piece: pieceRef,
2019-08-27 18:45:21 +00:00
Offset: offset,
Size: size,
})
newRef, err := cbor.DumpObject(&refs)
if err != nil {
return 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 {
return 0, err
}
if len(data) == 0 {
2019-08-27 23:03:22 +00:00
// TODO: batch
// TODO: GC
if err := r.intermediate.Put(nd); err != nil {
return 0, err
}
continue
2019-08-27 18:45:21 +00:00
}
if err := r.writeRef(nd.Cid(), r.pieceRef, offset, uint32(len(data))); err != nil {
2019-08-27 18:45:21 +00:00
return 0, 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, keepAtLeast uint64) (sectorID uint64, err error) {
size, err := r.Size()
if err != nil {
return 0, err
}
2019-08-27 23:03:22 +00:00
refst := &refStorer{
blockReader: r,
pieceRef: string(SerializationUnixfs0) + ref.String(),
writeRef: st.writeRef,
intermediate: st.intermediate,
}
2019-08-27 18:45:21 +00:00
return st.Store.AddPiece(refst.pieceRef, uint64(size), refst)
}
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.SealedRef
if err := cbor.DecodeInto(ent.Value, &refs); err != nil {
return nil, err
}
out[refCid] = 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
}
2019-08-27 18:45:21 +00:00
if err != nil {
return nil, err
}
var refs []api.SealedRef
if err := cbor.DecodeInto(ent, &refs); err != nil {
return nil, err
}
return 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
}
2019-08-27 18:45:21 +00:00
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{
2019-08-27 23:03:22 +00:00
intermediate: st.intermediate,
2019-08-27 18:45:21 +00:00
sectorBlocks: st,
approveUnseal: approveUnseal,
}
}