// package blockstore implements a thin wrapper over a datastore, giving a // clean interface for Getting and Putting block objects. package blockstore import ( "context" "errors" "sync" "sync/atomic" blocks "github.com/ipfs/go-ipfs/blocks" dshelp "github.com/ipfs/go-ipfs/thirdparty/ds-help" ds "gx/ipfs/QmRWDav6mzWseLWeYfVd5fvUKiVe9xNH29YfMF438fG364/go-datastore" dsns "gx/ipfs/QmRWDav6mzWseLWeYfVd5fvUKiVe9xNH29YfMF438fG364/go-datastore/namespace" dsq "gx/ipfs/QmRWDav6mzWseLWeYfVd5fvUKiVe9xNH29YfMF438fG364/go-datastore/query" logging "gx/ipfs/QmSpJByNKFX1sCsHBEp3R73FL4NF6FnQTEGyNAXHm2GS52/go-log" cid "gx/ipfs/QmcTcsTvfaeEBRFo1TkFgT8sRmgi1n1LTZpecfVP8fzpGD/go-cid" ) var log = logging.Logger("blockstore") // BlockPrefix namespaces blockstore datastores var BlockPrefix = ds.NewKey("blocks") var ValueTypeMismatch = errors.New("the retrieved value is not a Block") var ErrHashMismatch = errors.New("block in storage has different hash than requested") var ErrNotFound = errors.New("blockstore: block not found") // Blockstore wraps a Datastore type Blockstore interface { DeleteBlock(*cid.Cid) error Has(*cid.Cid) (bool, error) Get(*cid.Cid) (blocks.Block, error) Put(blocks.Block) error PutMany([]blocks.Block) error AllKeysChan(ctx context.Context) (<-chan *cid.Cid, error) } type GCLocker interface { // GCLock locks the blockstore for garbage collection. No operations // that expect to finish with a pin should ocurr simultaneously. // Reading during GC is safe, and requires no lock. GCLock() Unlocker // PinLock locks the blockstore for sequences of puts expected to finish // with a pin (before GC). Multiple put->pin sequences can write through // at the same time, but no GC should not happen simulatenously. // Reading during Pinning is safe, and requires no lock. PinLock() Unlocker // GcRequested returns true if GCLock has been called and is waiting to // take the lock GCRequested() bool } type GCBlockstore interface { Blockstore GCLocker } func NewGCBlockstore(bs Blockstore, gcl GCLocker) GCBlockstore { return gcBlockstore{bs, gcl} } type gcBlockstore struct { Blockstore GCLocker } func NewBlockstore(d ds.Batching) *blockstore { var dsb ds.Batching dd := dsns.Wrap(d, BlockPrefix) dsb = dd return &blockstore{ datastore: dsb, } } type blockstore struct { datastore ds.Batching lk sync.RWMutex gcreq int32 gcreqlk sync.Mutex rehash bool } func (bs *blockstore) HashOnRead(enabled bool) { bs.rehash = enabled } func (bs *blockstore) Get(k *cid.Cid) (blocks.Block, error) { if k == nil { log.Error("nil cid in blockstore") return nil, ErrNotFound } maybeData, err := bs.datastore.Get(dshelp.CidToDsKey(k)) if err == ds.ErrNotFound { return nil, ErrNotFound } if err != nil { return nil, err } bdata, ok := maybeData.([]byte) if !ok { return nil, ValueTypeMismatch } if bs.rehash { rbcid, err := k.Prefix().Sum(bdata) if err != nil { return nil, err } if !rbcid.Equals(k) { return nil, ErrHashMismatch } return blocks.NewBlockWithCid(bdata, rbcid) } else { return blocks.NewBlockWithCid(bdata, k) } } func (bs *blockstore) Put(block blocks.Block) error { k := dshelp.CidToDsKey(block.Cid()) // Has is cheaper than Put, so see if we already have it exists, err := bs.datastore.Has(k) if err == nil && exists { return nil // already stored. } return bs.datastore.Put(k, block.RawData()) } func (bs *blockstore) PutMany(blocks []blocks.Block) error { t, err := bs.datastore.Batch() if err != nil { return err } for _, b := range blocks { k := dshelp.CidToDsKey(b.Cid()) exists, err := bs.datastore.Has(k) if err == nil && exists { continue } err = t.Put(k, b.RawData()) if err != nil { return err } } return t.Commit() } func (bs *blockstore) Has(k *cid.Cid) (bool, error) { return bs.datastore.Has(dshelp.CidToDsKey(k)) } func (s *blockstore) DeleteBlock(k *cid.Cid) error { return s.datastore.Delete(dshelp.CidToDsKey(k)) } // AllKeysChan runs a query for keys from the blockstore. // this is very simplistic, in the future, take dsq.Query as a param? // // AllKeysChan respects context func (bs *blockstore) AllKeysChan(ctx context.Context) (<-chan *cid.Cid, error) { // KeysOnly, because that would be _a lot_ of data. q := dsq.Query{KeysOnly: true} // datastore/namespace does *NOT* fix up Query.Prefix q.Prefix = BlockPrefix.String() res, err := bs.datastore.Query(q) if err != nil { return nil, err } // this function is here to compartmentalize get := func() (*cid.Cid, bool) { select { case <-ctx.Done(): return nil, false case e, more := <-res.Next(): if !more { return nil, false } if e.Error != nil { log.Debug("blockstore.AllKeysChan got err:", e.Error) return nil, false } // need to convert to key.Key using key.KeyFromDsKey. c, err := dshelp.DsKeyToCid(ds.RawKey(e.Key)) if err != nil { log.Warningf("error parsing key from DsKey: ", err) return nil, true } return c, true } } output := make(chan *cid.Cid, dsq.KeysOnlyBufSize) go func() { defer func() { res.Process().Close() // ensure exit (signals early exit, too) close(output) }() for { k, ok := get() if !ok { return } if k == nil { continue } select { case <-ctx.Done(): return case output <- k: } } }() return output, nil } func NewGCLocker() *gclocker { return &gclocker{} } type gclocker struct { lk sync.RWMutex gcreq int32 gcreqlk sync.Mutex } type Unlocker interface { Unlock() } type unlocker struct { unlock func() } func (u *unlocker) Unlock() { u.unlock() u.unlock = nil // ensure its not called twice } func (bs *gclocker) GCLock() Unlocker { atomic.AddInt32(&bs.gcreq, 1) bs.lk.Lock() atomic.AddInt32(&bs.gcreq, -1) return &unlocker{bs.lk.Unlock} } func (bs *gclocker) PinLock() Unlocker { bs.lk.RLock() return &unlocker{bs.lk.RUnlock} } func (bs *gclocker) GCRequested() bool { return atomic.LoadInt32(&bs.gcreq) > 0 }