Add builder for unixfs dags

data/builder/dir_test.go

@@ -0,0 +1,72 @@
+package builder
+
+import (
+	"bytes"
+	"fmt"
+	"testing"
+
+	"github.com/ipfs/go-unixfsnode"
+	dagpb "github.com/ipld/go-codec-dagpb"
+	"github.com/ipld/go-ipld-prime"
+	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
+)
+
+func mkEntries(cnt int, ls *ipld.LinkSystem) ([]dagpb.PBLink, error) {
+	entries := make([]dagpb.PBLink, 0, cnt)
+	for i := 0; i < cnt; i++ {
+		r := bytes.NewBufferString(fmt.Sprintf("%d", i))
+		f, s, err := BuildUnixFSFile(r, "", ls)
+		if err != nil {
+			return nil, err
+		}
+		e, err := BuildUnixFSDirectoryEntry(fmt.Sprintf("file %d", i), int64(s), f)
+		if err != nil {
+			return nil, err
+		}
+		entries = append(entries, e)
+	}
+	return entries, nil
+}
+
+func TestBuildUnixFSDirectory(t *testing.T) {
+	ls := cidlink.DefaultLinkSystem()
+	storage := cidlink.Memory{}
+	ls.StorageReadOpener = storage.OpenRead
+	ls.StorageWriteOpener = storage.OpenWrite
+
+	testSizes := []int{100, 1000, 50000}
+	for _, cnt := range testSizes {
+		entries, err := mkEntries(cnt, &ls)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		dl, err := BuildUnixFSDirectory(entries, &ls)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		pbn, err := ls.Load(ipld.LinkContext{}, dl, dagpb.Type.PBNode)
+		if err != nil {
+			t.Fatal(err)
+		}
+		ufd, err := unixfsnode.Reify(ipld.LinkContext{}, pbn, &ls)
+		if err != nil {
+			t.Fatal(err)
+		}
+		observedCnt := 0
+
+		li := ufd.MapIterator()
+		for !li.Done() {
+			_, _, err := li.Next()
+			if err != nil {
+				t.Fatal(err)
+			}
+			observedCnt++
+		}
+		if observedCnt != cnt {
+			fmt.Printf("%+v\n", ufd)
+			t.Fatalf("unexpected number of dir entries %d vs %d", observedCnt, cnt)
+		}
+	}
+}

data/builder/directory.go

@@ -0,0 +1,123 @@
+package builder
+
+import (
+	"fmt"
+	"io/fs"
+	"os"
+	"path"
+
+	"github.com/ipfs/go-unixfsnode/data"
+	dagpb "github.com/ipld/go-codec-dagpb"
+	"github.com/ipld/go-ipld-prime"
+	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
+	"github.com/multiformats/go-multihash"
+)
+
+// https://github.com/ipfs/go-ipfs/pull/8114/files#diff-eec963b47a6e1080d9d8023b4e438e6e3591b4154f7379a7e728401d2055374aR319
+const shardSplitThreshold = 262144
+
+// https://github.com/ipfs/go-unixfs/blob/ec6bb5a4c5efdc3a5bce99151b294f663ee9c08d/io/directory.go#L29
+const defaultShardWidth = 256
+
+// BuildUnixFSRecursive returns a link pointing to the UnixFS node representing
+// the file or directory tree pointed to by `root`
+func BuildUnixFSRecursive(root string, ls *ipld.LinkSystem) (ipld.Link, uint64, error) {
+	info, err := os.Lstat(root)
+	if err != nil {
+		return nil, 0, err
+	}
+
+	m := info.Mode()
+	switch {
+	case m.IsDir():
+		entries, err := os.ReadDir(root)
+		if err != nil {
+			return nil, 0, err
+		}
+		lnks := make([]dagpb.PBLink, 0, len(entries))
+		for _, e := range entries {
+			lnk, sz, err := BuildUnixFSRecursive(path.Join(root, e.Name()), ls)
+			if err != nil {
+				return nil, 0, err
+			}
+			entry, err := BuildUnixFSDirectoryEntry(e.Name(), int64(sz), lnk)
+			if err != nil {
+				return nil, 0, err
+			}
+			lnks = append(lnks, entry)
+		}
+		outLnk, err := BuildUnixFSDirectory(lnks, ls)
+		return outLnk, 0, err
+	case m.Type() == fs.ModeSymlink:
+		content, err := os.Readlink(root)
+		if err != nil {
+			return nil, 0, err
+		}
+		return BuildUnixFSSymlink(content, ls)
+	case m.IsRegular():
+		fp, err := os.Open(root)
+		if err != nil {
+			return nil, 0, err
+		}
+		defer fp.Close()
+		return BuildUnixFSFile(fp, "", ls)
+	default:
+		return nil, 0, fmt.Errorf("cannot encode non regular file: %s", root)
+	}
+}
+
+// estimateDirSize estimates if a directory is big enough that it warrents sharding.
+// The estimate is the sum over the len(linkName) + bytelen(linkHash)
+// https://github.com/ipfs/go-unixfs/blob/master/io/directory.go#L152-L162
+func estimateDirSize(entries []dagpb.PBLink) int {
+	s := 0
+	for _, e := range entries {
+		lnk := e.Hash.Link()
+		cl, ok := lnk.(cidlink.Link)
+		if ok {
+			s += len(e.Name.Must().String()) + cl.ByteLen()
+		} else {
+			s += len(e.Name.Must().String()) + len(lnk.String())
+		}
+	}
+	return s
+}
+
+// BuildUnixFSDirectory creates a directory link over a collection of entries.
+func BuildUnixFSDirectory(entries []dagpb.PBLink, ls *ipld.LinkSystem) (ipld.Link, error) {
+	if estimateDirSize(entries) > shardSplitThreshold {
+		return BuildUnixFSShardedDirectory(defaultShardWidth, multihash.MURMUR3_128, entries, ls)
+	}
+	ufd, err := BuildUnixFS(func(b *Builder) {
+		DataType(b, data.Data_Directory)
+	})
+	if err != nil {
+		return nil, err
+	}
+	pbb := dagpb.Type.PBNode.NewBuilder()
+	pbm, err := pbb.BeginMap(2)
+	if err != nil {
+		return nil, err
+	}
+	pbm.AssembleKey().AssignString("Data")
+	pbm.AssembleValue().AssignBytes(data.EncodeUnixFSData(ufd))
+	pbm.AssembleKey().AssignString("Links")
+	lnks, err := pbm.AssembleValue().BeginList(int64(len(entries)))
+	if err != nil {
+		return nil, err
+	}
+	// sorting happens in codec-dagpb
+	for _, e := range entries {
+		if err := lnks.AssembleValue().AssignNode(e); err != nil {
+			return nil, err
+		}
+	}
+	if err := lnks.Finish(); err != nil {
+		return nil, err
+	}
+	if err := pbm.Finish(); err != nil {
+		return nil, err
+	}
+	node := pbb.Build()
+	return ls.Store(ipld.LinkContext{}, fileLinkProto, node)
+}

data/builder/dirshard.go

@@ -0,0 +1,200 @@
+package builder
+
+import (
+	"fmt"
+	"hash"
+
+	bitfield "github.com/ipfs/go-bitfield"
+	"github.com/ipfs/go-unixfsnode/data"
+	"github.com/ipfs/go-unixfsnode/hamt"
+	dagpb "github.com/ipld/go-codec-dagpb"
+	"github.com/ipld/go-ipld-prime"
+	"github.com/multiformats/go-multihash"
+	"github.com/spaolacci/murmur3"
+)
+
+type shard struct {
+	// metadata about the shard
+	hasher  uint64
+	size    int
+	sizeLg2 int
+	width   int
+	depth   int
+
+	children map[int]entry
+}
+
+// a shard entry is either another shard, or a direct link.
+type entry struct {
+	*shard
+	*hamtLink
+}
+
+// a hamtLink is a member of the hamt - the file/directory pointed to, but
+// stored with it's hashed key used for addressing.
+type hamtLink struct {
+	hash hashBits
+	dagpb.PBLink
+}
+
+// BuildUnixFSShardedDirectory will build a hamt of unixfs hamt shards encoing a directory with more entries
+// than is typically allowed to fit in a standard IPFS single-block unixFS directory.
+func BuildUnixFSShardedDirectory(size int, hasher uint64, entries []dagpb.PBLink, ls *ipld.LinkSystem) (ipld.Link, error) {
+	// hash the entries
+	var h hash.Hash
+	var err error
+	// TODO: use the multihash registry once murmur3 behavior is encoded there.
+	// https://github.com/multiformats/go-multihash/pull/150
+	if hasher == hamt.HashMurmur3 {
+		h = murmur3.New64()
+	} else {
+		h, err = multihash.GetHasher(hasher)
+		if err != nil {
+			return nil, err
+		}
+	}
+	hamtEntries := make([]hamtLink, 0, len(entries))
+	for _, e := range entries {
+		name := e.Name.Must().String()
+		sum := h.Sum([]byte(name))
+		hamtEntries = append(hamtEntries, hamtLink{
+			sum,
+			e,
+		})
+	}
+
+	sizeLg2, err := logtwo(size)
+	if err != nil {
+		return nil, err
+	}
+
+	sharder := shard{
+		hasher:  hasher,
+		size:    size,
+		sizeLg2: sizeLg2,
+		width:   len(fmt.Sprintf("%X", size-1)),
+		depth:   0,
+
+		children: make(map[int]entry),
+	}
+
+	for _, entry := range hamtEntries {
+		err := sharder.add(entry)
+		if err != nil {
+			return nil, err
+		}
+	}
+	fmt.Printf("sharder to serialize, %d children (%d)\n", len(sharder.children), len(hamtEntries))
+
+	return sharder.serialize(ls)
+}
+
+func (s *shard) add(lnk hamtLink) error {
+	// get the bucket for lnk
+	bucket, err := lnk.hash.Slice(s.depth*s.sizeLg2, s.sizeLg2)
+	if err != nil {
+		return err
+	}
+
+	current, ok := s.children[bucket]
+	if !ok {
+		s.children[bucket] = entry{nil, &lnk}
+		return nil
+	} else if current.shard != nil {
+		return current.shard.add(lnk)
+	}
+	// make a shard for current and lnk
+	newShard := entry{
+		&shard{
+			hasher:   s.hasher,
+			size:     s.size,
+			sizeLg2:  s.sizeLg2,
+			width:    s.width,
+			depth:    s.depth + 1,
+			children: make(map[int]entry),
+		},
+		nil,
+	}
+	if err := newShard.add(*current.hamtLink); err != nil {
+		return err
+	}
+	s.children[bucket] = newShard
+	return newShard.add(lnk)
+}
+
+func (s *shard) formatLinkName(name string, idx int) string {
+	return fmt.Sprintf("%*X%s", s.width, idx, name)
+}
+
+// bitmap calculates the bitmap of which links in the shard are set.
+func (s *shard) bitmap() []byte {
+	bm := bitfield.NewBitfield(s.size)
+	for i := 0; i < s.size; i++ {
+		if _, ok := s.children[i]; ok {
+			bm.SetBit(i)
+		}
+	}
+	return bm.Bytes()
+}
+
+// serialize stores the concrete representation of this shard in the link system and
+// returns a link to it.
+func (s *shard) serialize(ls *ipld.LinkSystem) (ipld.Link, error) {
+	ufd, err := BuildUnixFS(func(b *Builder) {
+		DataType(b, data.Data_HAMTShard)
+		HashType(b, s.hasher)
+		Data(b, s.bitmap())
+		Fanout(b, uint64(s.size))
+	})
+	fmt.Printf("shard: %+v\n", ufd)
+	if err != nil {
+		return nil, err
+	}
+	pbb := dagpb.Type.PBNode.NewBuilder()
+	pbm, err := pbb.BeginMap(2)
+	if err != nil {
+		return nil, err
+	}
+	pbm.AssembleKey().AssignString("Data")
+	pbm.AssembleValue().AssignBytes(data.EncodeUnixFSData(ufd))
+	pbm.AssembleKey().AssignString("Links")
+
+	lnkBuilder := dagpb.Type.PBLinks.NewBuilder()
+	lnks, err := lnkBuilder.BeginList(int64(len(s.children)))
+	if err != nil {
+		return nil, err
+	}
+	// sorting happens in codec-dagpb
+	for idx, e := range s.children {
+		var lnk dagpb.PBLink
+		if e.shard != nil {
+			ipldLnk, err := e.shard.serialize(ls)
+			if err != nil {
+				return nil, err
+			}
+			fullName := s.formatLinkName("", idx)
+			lnk, err = BuildUnixFSDirectoryEntry(fullName, 0, ipldLnk)
+			if err != nil {
+				return nil, err
+			}
+		} else {
+			fullName := s.formatLinkName(e.Name.Must().String(), idx)
+			lnk, err = BuildUnixFSDirectoryEntry(fullName, e.Tsize.Must().Int(), e.Hash.Link())
+		}
+		if err != nil {
+			return nil, err
+		}
+		if err := lnks.AssembleValue().AssignNode(lnk); err != nil {
+			return nil, err
+		}
+	}
+	if err := lnks.Finish(); err != nil {
+		return nil, err
+	}
+	pbm.AssembleValue().AssignNode(lnkBuilder.Build())
+	if err := pbm.Finish(); err != nil {
+		return nil, err
+	}
+	node := pbb.Build()
+	return ls.Store(ipld.LinkContext{}, fileLinkProto, node)
+}