implement an HAMT for unixfs directory sharding

License: MIT
Signed-off-by: Jeromy <[email protected]>

Author: whyrusleeping

assets/assets.go

@@ -53,17 +53,27 @@ func addAssetList(nd *core.IpfsNode, l []string) (*cid.Cid, error) {
 		}
 
 		fname := filepath.Base(p)
+
 		c, err := cid.Decode(s)
 		if err != nil {
 			return nil, err
 		}
 
-		if err := dirb.AddChild(nd.Context(), fname, c); err != nil {
+		node, err := nd.DAG.Get(nd.Context(), c)
+		if err != nil {
+			return nil, err
+		}
+
+		if err := dirb.AddChild(nd.Context(), fname, node); err != nil {
 			return nil, fmt.Errorf("assets: could not add '%s' as a child: %s", fname, err)
 		}
 	}
 
-	dir := dirb.GetNode()
+	dir, err := dirb.GetNode()
+	if err != nil {
+		return nil, err
+	}
+
 	dcid, err := nd.DAG.Add(dir)
 	if err != nil {
 		return nil, fmt.Errorf("assets: DAG.Add(dir) failed: %s", err)

commands/request.go

@@ -2,6 +2,7 @@ package commands
 
 import (
 	"bufio"
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -10,7 +11,6 @@ import (
 	"strconv"
 	"time"
 
-	context "context"
 	"github.com/ipfs/go-ipfs/commands/files"
 	"github.com/ipfs/go-ipfs/core"
 	"github.com/ipfs/go-ipfs/repo/config"

core/commands/files/files.go

@@ -265,16 +265,7 @@ func getNodeFromPath(ctx context.Context, node *core.IpfsNode, p string) (node.N
 			ResolveOnce: uio.ResolveUnixfsOnce,
 		}
 
-		nd, err := core.Resolve(ctx, node.Namesys, resolver, np)
-		if err != nil {
-			return nil, err
-		}
-		pbnd, ok := nd.(*dag.ProtoNode)
-		if !ok {
-			return nil, dag.ErrNotProtobuf
-		}
-
-		return pbnd, nil
+		return core.Resolve(ctx, node.Namesys, resolver, np)
 	default:
 		fsn, err := mfs.Lookup(node.FilesRoot, p)
 		if err != nil {
@@ -357,7 +348,13 @@ Examples:
 		case *mfs.Directory:
 			if !long {
 				var output []mfs.NodeListing
-				for _, name := range fsn.ListNames() {
+				names, err := fsn.ListNames()
+				if err != nil {
+					res.SetError(err, cmds.ErrNormal)
+					return
+				}
+
+				for _, name := range names {
 					output = append(output, mfs.NodeListing{
 						Name: name,
 					})

core/corehttp/gateway_handler.go

@@ -403,13 +403,14 @@ func (i *gatewayHandler) putHandler(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 
-		pbnd, ok := rnode.(*dag.ProtoNode)
+		pbrnode, ok := rnode.(*dag.ProtoNode)
 		if !ok {
-			webError(w, "Cannot read non protobuf nodes through gateway", dag.ErrNotProtobuf, http.StatusBadRequest)
+			webError(w, "putHandler: Can only operate on protobuf nodes", dag.ErrNotProtobuf, http.StatusInternalServerError)
 			return
 		}
 
-		e := dagutils.NewDagEditor(pbnd, i.node.DAG)
+		e := dagutils.NewDagEditor(pbrnode, i.node.DAG)
+
 		err = e.InsertNodeAtPath(ctx, newPath, newnode, ft.EmptyDirNode)
 		if err != nil {
 			webError(w, "putHandler: InsertNodeAtPath failed", err, http.StatusInternalServerError)

core/coreunix/add.go

@@ -226,7 +226,12 @@ func (adder *Adder) outputDirs(path string, fsn mfs.FSNode) error {
 	case *mfs.File:
 		return nil
 	case *mfs.Directory:
-		for _, name := range fsn.ListNames() {
+		names, err := fsn.ListNames()
+		if err != nil {
+			return err
+		}
+
+		for _, name := range names {
 			child, err := fsn.Child(name)
 			if err != nil {
 				return err

mfs/dir.go

@@ -12,6 +12,7 @@ import (
 
 	dag "github.com/ipfs/go-ipfs/merkledag"
 	ft "github.com/ipfs/go-ipfs/unixfs"
+	uio "github.com/ipfs/go-ipfs/unixfs/io"
 	ufspb "github.com/ipfs/go-ipfs/unixfs/pb"
 
 	node "gx/ipfs/QmU7bFWQ793qmvNy7outdCaMfSDNk8uqhx4VNrxYj5fj5g/go-ipld-node"
@@ -29,25 +30,31 @@ type Directory struct {
 	files     map[string]*File
 
 	lock sync.Mutex
-	node *dag.ProtoNode
 	ctx  context.Context
 
+	dirbuilder *uio.Directory
+
 	modTime time.Time
 
 	name string
 }
 
-func NewDirectory(ctx context.Context, name string, node *dag.ProtoNode, parent childCloser, dserv dag.DAGService) *Directory {
-	return &Directory{
-		dserv:     dserv,
-		ctx:       ctx,
-		name:      name,
-		node:      node,
-		parent:    parent,
-		childDirs: make(map[string]*Directory),
-		files:     make(map[string]*File),
-		modTime:   time.Now(),
+func NewDirectory(ctx context.Context, name string, node node.Node, parent childCloser, dserv dag.DAGService) (*Directory, error) {
+	db, err := uio.NewDirectoryFromNode(dserv, node)
+	if err != nil {
+		return nil, err
 	}
+
+	return &Directory{
+		dserv:      dserv,
+		ctx:        ctx,
+		name:       name,
+		dirbuilder: db,
+		parent:     parent,
+		childDirs:  make(map[string]*Directory),
+		files:      make(map[string]*File),
+		modTime:    time.Now(),
+	}, nil
 }
 
 // closeChild updates the child by the given name to the dag node 'nd'
@@ -81,21 +88,26 @@ func (d *Directory) closeChildUpdate(name string, nd *dag.ProtoNode, sync bool)
 }
 
 func (d *Directory) flushCurrentNode() (*dag.ProtoNode, error) {
-	_, err := d.dserv.Add(d.node)
+	nd, err := d.dirbuilder.GetNode()
 	if err != nil {
 		return nil, err
 	}
 
-	return d.node.Copy(), nil
-}
+	_, err = d.dserv.Add(nd)
+	if err != nil {
+		return nil, err
+	}
 
-func (d *Directory) updateChild(name string, nd node.Node) error {
-	err := d.node.RemoveNodeLink(name)
-	if err != nil && err != dag.ErrNotFound {
-		return err
+	pbnd, ok := nd.(*dag.ProtoNode)
+	if !ok {
+		return nil, dag.ErrNotProtobuf
 	}
 
-	err = d.node.AddNodeLinkClean(name, nd)
+	return pbnd.Copy(), nil
+}
+
+func (d *Directory) updateChild(name string, nd node.Node) error {
+	err := d.dirbuilder.AddChild(d.ctx, name, nd)
 	if err != nil {
 		return err
 	}
@@ -130,8 +142,12 @@ func (d *Directory) cacheNode(name string, nd node.Node) (FSNode, error) {
 		}
 
 		switch i.GetType() {
-		case ufspb.Data_Directory:
-			ndir := NewDirectory(d.ctx, name, nd, d, d.dserv)
+		case ufspb.Data_Directory, ufspb.Data_HAMTShard:
+			ndir, err := NewDirectory(d.ctx, name, nd, d, d.dserv)
+			if err != nil {
+				return nil, err
+			}
+
 			d.childDirs[name] = ndir
 			return ndir, nil
 		case ufspb.Data_File, ufspb.Data_Raw, ufspb.Data_Symlink:
@@ -175,15 +191,7 @@ func (d *Directory) Uncache(name string) {
 // childFromDag searches through this directories dag node for a child link
 // with the given name
 func (d *Directory) childFromDag(name string) (node.Node, error) {
-	pbn, err := d.node.GetLinkedNode(d.ctx, d.dserv, name)
-	switch err {
-	case nil:
-		return pbn, nil
-	case dag.ErrLinkNotFound:
-		return nil, os.ErrNotExist
-	default:
-		return nil, err
-	}
+	return d.dirbuilder.Find(d.ctx, name)
 }
 
 // childUnsync returns the child under this directory by the given name
@@ -209,7 +217,7 @@ type NodeListing struct {
 	Hash string
 }
 
-func (d *Directory) ListNames() []string {
+func (d *Directory) ListNames() ([]string, error) {
 	d.lock.Lock()
 	defer d.lock.Unlock()
 
@@ -221,7 +229,12 @@ func (d *Directory) ListNames() []string {
 		names[n] = struct{}{}
 	}
 
-	for _, l := range d.node.Links() {
+	links, err := d.dirbuilder.Links()
+	if err != nil {
+		return nil, err
+	}
+
+	for _, l := range links {
 		names[l.Name] = struct{}{}
 	}
 
@@ -231,15 +244,21 @@ func (d *Directory) ListNames() []string {
 	}
 	sort.Strings(out)
 
-	return out
+	return out, nil
 }
 
 func (d *Directory) List() ([]NodeListing, error) {
 	d.lock.Lock()
 	defer d.lock.Unlock()
 
 	var out []NodeListing
-	for _, l := range d.node.Links() {
+
+	links, err := d.dirbuilder.Links()
+	if err != nil {
+		return nil, err
+	}
+
+	for _, l := range links {
 		child := NodeListing{}
 		child.Name = l.Name
 
@@ -285,20 +304,23 @@ func (d *Directory) Mkdir(name string) (*Directory, error) {
 		}
 	}
 
-	ndir := new(dag.ProtoNode)
-	ndir.SetData(ft.FolderPBData())
+	ndir := ft.EmptyDirNode()
 
 	_, err = d.dserv.Add(ndir)
 	if err != nil {
 		return nil, err
 	}
 
-	err = d.node.AddNodeLinkClean(name, ndir)
+	err = d.dirbuilder.AddChild(d.ctx, name, ndir)
+	if err != nil {
+		return nil, err
+	}
+
+	dirobj, err := NewDirectory(d.ctx, name, ndir, d, d.dserv)
 	if err != nil {
 		return nil, err
 	}
 
-	dirobj := NewDirectory(d.ctx, name, ndir, d, d.dserv)
 	d.childDirs[name] = dirobj
 	return dirobj, nil
 }
@@ -310,12 +332,7 @@ func (d *Directory) Unlink(name string) error {
 	delete(d.childDirs, name)
 	delete(d.files, name)
 
-	err := d.node.RemoveNodeLink(name)
-	if err != nil {
-		return err
-	}
-
-	_, err = d.dserv.Add(d.node)
+	err := d.dirbuilder.RemoveChild(d.ctx, name)
 	if err != nil {
 		return err
 	}
@@ -350,7 +367,7 @@ func (d *Directory) AddChild(name string, nd node.Node) error {
 		return err
 	}
 
-	err = d.node.AddNodeLinkClean(name, nd)
+	err = d.dirbuilder.AddChild(d.ctx, name, nd)
 	if err != nil {
 		return err
 	}
@@ -406,10 +423,15 @@ func (d *Directory) GetNode() (node.Node, error) {
 		return nil, err
 	}
 
-	_, err = d.dserv.Add(d.node)
+	nd, err := d.dirbuilder.GetNode()
+	if err != nil {
+		return nil, err
+	}
+
+	_, err = d.dserv.Add(nd)
 	if err != nil {
 		return nil, err
 	}
 
-	return d.node.Copy(), nil
+	return nd, err
 }