[WIP] Filestore Implementation

blocks/blocks.go

@@ -13,14 +13,15 @@ import (
 
 var ErrWrongHash = errors.New("data did not match given hash!")
 
+// Block is a singular block of data in ipfs
+
 type Block interface {
 	RawData() []byte
 	Cid() *cid.Cid
 	String() string
 	Loggable() map[string]interface{}
 }
 
-// Block is a singular block of data in ipfs
 type BasicBlock struct {
 	cid  *cid.Cid
 	data []byte

blocks/blockstore/blockstore.go

@@ -21,7 +21,9 @@ import (
 var log = logging.Logger("blockstore")
 
 // BlockPrefix namespaces blockstore datastores
-var BlockPrefix = ds.NewKey("blocks")
+const DefaultPrefix = "/blocks"
+
+var blockPrefix = ds.NewKey(DefaultPrefix)
 
 var ValueTypeMismatch = errors.New("the retrieved value is not a Block")
 var ErrHashMismatch = errors.New("block in storage has different hash than requested")
@@ -71,20 +73,23 @@ type gcBlockstore struct {
 }
 
 func NewBlockstore(d ds.Batching) *blockstore {
+	return NewBlockstoreWPrefix(d, DefaultPrefix)
+}
+
+func NewBlockstoreWPrefix(d ds.Batching, prefix string) *blockstore {
 	var dsb ds.Batching
-	dd := dsns.Wrap(d, BlockPrefix)
+	prefixKey := ds.NewKey(prefix)
+	dd := dsns.Wrap(d, prefixKey)
 	dsb = dd
 	return &blockstore{
 		datastore: dsb,
+		prefix:    prefixKey,
 	}
 }
 
 type blockstore struct {
 	datastore ds.Batching
-
-	lk      sync.RWMutex
-	gcreq   int32
-	gcreqlk sync.Mutex
+	prefix    ds.Key
 
 	rehash bool
 }
@@ -130,11 +135,8 @@ func (bs *blockstore) Get(k *cid.Cid) (blocks.Block, error) {
 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.
-	}
+	// Note: The Has Check is now done by the MultiBlockstore
+
 	return bs.datastore.Put(k, block.RawData())
 }
 
@@ -145,11 +147,6 @@ func (bs *blockstore) PutMany(blocks []blocks.Block) error {
 	}
 	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
@@ -175,7 +172,7 @@ 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()
+	q.Prefix = bs.prefix.String()
 	res, err := bs.datastore.Query(q)
 	if err != nil {
 		return nil, err

blocks/blockstore/blockstore_test.go

@@ -170,7 +170,7 @@ func TestAllKeysRespectsContext(t *testing.T) {
 	default:
 	}
 
-	e := dsq.Entry{Key: BlockPrefix.ChildString("foo").String()}
+	e := dsq.Entry{Key: blockPrefix.ChildString("foo").String()}
 	resultChan <- dsq.Result{Entry: e} // let it go.
 	close(resultChan)
 	<-done                       // should be done now.
@@ -190,7 +190,7 @@ func TestValueTypeMismatch(t *testing.T) {
 	block := blocks.NewBlock([]byte("some data"))
 
 	datastore := ds.NewMapDatastore()
-	k := BlockPrefix.Child(dshelp.CidToDsKey(block.Cid()))
+	k := blockPrefix.Child(dshelp.CidToDsKey(block.Cid()))
 	datastore.Put(k, "data that isn't a block!")
 
 	blockstore := NewBlockstore(ds_sync.MutexWrap(datastore))

blocks/blockstore/bloom_cache_test.go

@@ -104,11 +104,11 @@ func TestHasIsBloomCached(t *testing.T) {
 	block := blocks.NewBlock([]byte("newBlock"))
 
 	cachedbs.PutMany([]blocks.Block{block})
-	if cacheFails != 2 {
-		t.Fatalf("expected two datastore hits: %d", cacheFails)
+	if cacheFails != 1 {
+		t.Fatalf("expected datastore hits: %d", cacheFails)
 	}
 	cachedbs.Put(block)
-	if cacheFails != 3 {
+	if cacheFails != 2 {
 		t.Fatalf("expected datastore hit: %d", cacheFails)
 	}
 

blocks/blockstore/multi.go

@@ -0,0 +1,158 @@
+package blockstore
+
+// A very simple multi-blockstore that analogous to a unionfs Put and
+// DeleteBlock only go to the first blockstore all others are
+// considered readonly.
+
+import (
+	//"errors"
+	"context"
+
+	blocks "github.com/ipfs/go-ipfs/blocks"
+	cid "gx/ipfs/QmXfiyr2RWEXpVDdaYnD2HNiBk6UBddsvEP4RPfXb6nGqY/go-cid"
+	dsq "gx/ipfs/QmbzuUusHqaLLoNTDEVLcSF6vZDHZDLPC7p4bztRvvkXxU/go-datastore/query"
+)
+
+type LocateInfo struct {
+	Prefix string
+	Error  error
+}
+
+type MultiBlockstore interface {
+	Blockstore
+	GCLocker
+	FirstMount() Blockstore
+	Mounts() []string
+	Mount(prefix string) Blockstore
+	Locate(*cid.Cid) []LocateInfo
+}
+
+type Mount struct {
+	Prefix string
+	Blocks Blockstore
+}
+
+func NewMultiBlockstore(mounts ...Mount) *multiblockstore {
+	return &multiblockstore{
+		mounts: mounts,
+	}
+}
+
+type multiblockstore struct {
+	mounts []Mount
+	gclocker
+}
+
+func (bs *multiblockstore) FirstMount() Blockstore {
+	return bs.mounts[0].Blocks
+}
+
+func (bs *multiblockstore) Mounts() []string {
+	mounts := make([]string, 0, len(bs.mounts))
+	for _, mnt := range bs.mounts {
+		mounts = append(mounts, mnt.Prefix)
+	}
+	return mounts
+}
+
+func (bs *multiblockstore) Mount(prefix string) Blockstore {
+	for _, m := range bs.mounts {
+		if m.Prefix == prefix {
+			return m.Blocks
+		}
+	}
+	return nil
+}
+
+func (bs *multiblockstore) DeleteBlock(key *cid.Cid) error {
+	return bs.mounts[0].Blocks.DeleteBlock(key)
+}
+
+func (bs *multiblockstore) Has(c *cid.Cid) (bool, error) {
+	var firstErr error
+	for _, m := range bs.mounts {
+		have, err := m.Blocks.Has(c)
+		if have && err == nil {
+			return have, nil
+		}
+		if err != nil && firstErr == nil {
+			firstErr = err
+		}
+	}
+	return false, firstErr
+}
+
+func (bs *multiblockstore) Get(c *cid.Cid) (blocks.Block, error) {
+	var firstErr error
+	for _, m := range bs.mounts {
+		blk, err := m.Blocks.Get(c)
+		if err == nil {
+			return blk, nil
+		}
+		if firstErr == nil || firstErr == ErrNotFound {
+			firstErr = err
+		}
+	}
+	return nil, firstErr
+}
+
+func (bs *multiblockstore) Locate(c *cid.Cid) []LocateInfo {
+	res := make([]LocateInfo, 0, len(bs.mounts))
+	for _, m := range bs.mounts {
+		_, err := m.Blocks.Get(c)
+		res = append(res, LocateInfo{m.Prefix, err})
+	}
+	return res
+}
+
+func (bs *multiblockstore) Put(blk blocks.Block) error {
+	// First call Has() to make sure the block doesn't exist in any of
+	// the sub-blockstores, otherwise we could end with data being
+	// duplicated in two blockstores.
+	exists, err := bs.Has(blk.Cid())
+	if err == nil && exists {
+		return nil // already stored
+	}
+	return bs.mounts[0].Blocks.Put(blk)
+}
+
+func (bs *multiblockstore) PutMany(blks []blocks.Block) error {
+	stilladd := make([]blocks.Block, 0, len(blks))
+	// First call Has() to make sure the block doesn't exist in any of
+	// the sub-blockstores, otherwise we could end with data being
+	// duplicated in two blockstores.
+	for _, blk := range blks {
+		exists, err := bs.Has(blk.Cid())
+		if err == nil && exists {
+			continue // already stored
+		}
+		stilladd = append(stilladd, blk)
+	}
+	if len(stilladd) == 0 {
+		return nil
+	}
+	return bs.mounts[0].Blocks.PutMany(stilladd)
+}
+
+func (bs *multiblockstore) AllKeysChan(ctx context.Context) (<-chan *cid.Cid, error) {
+	//return bs.mounts[0].Blocks.AllKeysChan(ctx)
+	//return nil, errors.New("Unimplemented")
+	in := make([]<-chan *cid.Cid, 0, len(bs.mounts))
+	for _, m := range bs.mounts {
+		ch, err := m.Blocks.AllKeysChan(ctx)
+		if err != nil {
+			return nil, err
+		}
+		in = append(in, ch)
+	}
+	out := make(chan *cid.Cid, dsq.KeysOnlyBufSize)
+	go func() {
+		defer close(out)
+		for _, in0 := range in {
+			for key := range in0 {
+				out <- key
+			}
+		}
+	}()
+	return out, nil
+}

blocks/blockstore/util/remove.go

@@ -27,14 +27,23 @@ type RmBlocksOpts struct {
 	Force  bool
 }
 
-func RmBlocks(blocks bs.GCBlockstore, pins pin.Pinner, out chan<- interface{}, cids []*cid.Cid, opts RmBlocksOpts) error {
+func RmBlocks(mbs bs.MultiBlockstore, pins pin.Pinner, out chan<- interface{}, cids []*cid.Cid, opts RmBlocksOpts) error {
+	prefix := opts.Prefix
+	if prefix == "" {
+		prefix = mbs.Mounts()[0]
+	}
+	blocks := mbs.Mount(prefix)
+	if blocks == nil {
+		return fmt.Errorf("Could not find blockstore: %s\n", prefix)
+	}
+
 	go func() {
 		defer close(out)
 
-		unlocker := blocks.GCLock()
+		unlocker := mbs.GCLock()
 		defer unlocker.Unlock()
 
-		stillOkay := FilterPinned(pins, out, cids)
+		stillOkay := FilterPinned(mbs, pins, out, cids, prefix)
 
 		for _, c := range stillOkay {
 			err := blocks.DeleteBlock(c)
@@ -50,15 +59,15 @@ func RmBlocks(blocks bs.GCBlockstore, pins pin.Pinner, out chan<- interface{}, c
 	return nil
 }
 
-func FilterPinned(pins pin.Pinner, out chan<- interface{}, cids []*cid.Cid) []*cid.Cid {
+func FilterPinned(mbs bs.MultiBlockstore, pins pin.Pinner, out chan<- interface{}, cids []*cid.Cid, prefix string) []*cid.Cid {
 	stillOkay := make([]*cid.Cid, 0, len(cids))
 	res, err := pins.CheckIfPinned(cids...)
 	if err != nil {
 		out <- &RemovedBlock{Error: fmt.Sprintf("pin check failed: %s", err)}
 		return nil
 	}
 	for _, r := range res {
-		if !r.Pinned() {
+		if !r.Pinned() || AvailableElsewhere(mbs, prefix, r.Key) {
 			stillOkay = append(stillOkay, r.Key)
 		} else {
 			out <- &RemovedBlock{
@@ -70,6 +79,16 @@ func FilterPinned(pins pin.Pinner, out chan<- interface{}, cids []*cid.Cid) []*c
 	return stillOkay
 }
 
+func AvailableElsewhere(mbs bs.MultiBlockstore, prefix string, c *cid.Cid) bool {
+	locations := mbs.Locate(c)
+	for _, loc := range locations {
+		if loc.Error == nil && loc.Prefix != prefix {
+			return true
+		}
+	}
+	return false
+}
+
 func ProcRmOutput(in <-chan interface{}, sout io.Writer, serr io.Writer) error {
 	someFailed := false
 	for res := range in {

cmd/ipfs/ipfs.go

@@ -108,4 +108,6 @@ var cmdDetailsMap = map[*cmds.Command]cmdDetails{
 	commands.ActiveReqsCmd:                {cannotRunOnClient: true},
 	commands.RepoFsckCmd:                  {cannotRunOnDaemon: true},
 	commands.ConfigCmd.Subcommand("edit"): {cannotRunOnDaemon: true, doesNotUseRepo: true},
+	commands.FilestoreEnable:              {cannotRunOnDaemon: true},
+	commands.FilestoreDisable:             {cannotRunOnDaemon: true},
 }

commands/files/slicefile.go

@@ -23,6 +23,10 @@ func (f *SliceFile) IsDirectory() bool {
 	return true
 }
 
+func (f *SliceFile) NumFiles() int {
+	return len(f.files)
+}
+
 func (f *SliceFile) NextFile() (File, error) {
 	if f.n >= len(f.files) {
 		return nil, io.EOF