WIP

Author: schomatis

hamt/hamt.go

@@ -37,7 +37,11 @@ const (
 	HashMurmur3 uint64 = 0x22
 )
 
-func (ds *Shard) isValueNode() bool {
+// Hash function declared as global variable only for testing purposes.
+// FIXME: We shoul have a cleaner way to replace this during tests.
+var HAMTHashFunction = murmur3Hash
+
+func (ds *Shard) IsValueNode() bool {
 	return ds.key != "" && ds.val != nil
 }
 
@@ -47,13 +51,22 @@ type Shard struct {
 
 	childer *childer
 
+	// Entries per node (number of possible childs indexed by the partial key).
 	tableSize    int
+	// Bits needed to encode child indexes (log2 of number of entries). This is
+	// the number of bits taken from the hash key on each level of the tree.
 	tableSizeLg2 int
 
 	builder  cid.Builder
 	hashFunc uint64
 
+	// String format with number of zeros that will be present in the hexadecimal
+	// encoding of the child index to always reach the fixed maxpadlen chars.
+	// Example: maxpadlen = 4 => prefixPadStr: "%04X" (print number in hexadecimal
+	// format padding with zeros to always reach 4 characters).
 	prefixPadStr string
+	// Length in chars of string that encodes child indexes. We encode indexes
+	// as hexadecimal strings to this is log4 of number of entries.
 	maxpadlen    int
 
 	dserv ipld.DAGService
@@ -70,6 +83,7 @@ func NewShard(dserv ipld.DAGService, size int) (*Shard, error) {
 		return nil, err
 	}
 
+	// FIXME: Make this at least a static configuration for testing.
 	ds.hashFunc = HashMurmur3
 	return ds, nil
 }
@@ -214,7 +228,7 @@ func (ds *Shard) Set(ctx context.Context, name string, nd ipld.Node) error {
 // name key in this Shard or its children. It also returns the previous link
 // under that name key (if any).
 func (ds *Shard) SetAndPrevious(ctx context.Context, name string, node ipld.Node) (*ipld.Link, error) {
-	hv := &hashBits{b: hash([]byte(name))}
+	hv := newHashBits(name)
 	err := ds.dserv.Add(ctx, node)
 	if err != nil {
 		return nil, err
@@ -224,6 +238,9 @@ func (ds *Shard) SetAndPrevious(ctx context.Context, name string, node ipld.Node
 	if err != nil {
 		return nil, err
 	}
+
+	// FIXME: We don't need to set the name here, it will get overwritten.
+	//  This is confusing, confirm and remove this line.
 	lnk.Name = ds.linkNamePrefix(0) + name
 
 	return ds.setValue(ctx, hv, name, lnk)
@@ -239,13 +256,13 @@ func (ds *Shard) Remove(ctx context.Context, name string) error {
 // RemoveAndPrevious is similar to the public Remove but also returns the
 // old removed link (if it exists).
 func (ds *Shard) RemoveAndPrevious(ctx context.Context, name string) (*ipld.Link, error) {
-	hv := &hashBits{b: hash([]byte(name))}
+	hv := newHashBits(name)
 	return ds.setValue(ctx, hv, name, nil)
 }
 
 // Find searches for a child node by 'name' within this hamt
 func (ds *Shard) Find(ctx context.Context, name string) (*ipld.Link, error) {
-	hv := &hashBits{b: hash([]byte(name))}
+	hv := newHashBits(name)
 
 	var out *ipld.Link
 	err := ds.getValue(ctx, hv, name, func(sv *Shard) error {
@@ -279,7 +296,7 @@ func (ds *Shard) childLinkType(lnk *ipld.Link) (linkType, error) {
 
 // Link returns a merklelink to this shard node
 func (ds *Shard) Link() (*ipld.Link, error) {
-	if ds.isValueNode() {
+	if ds.IsValueNode() {
 		return ds.val, nil
 	}
 
@@ -308,7 +325,7 @@ func (ds *Shard) getValue(ctx context.Context, hv *hashBits, key string, cb func
 			return err
 		}
 
-		if child.isValueNode() {
+		if child.IsValueNode() {
 			if child.key == key {
 				return cb(child)
 			}
@@ -335,6 +352,20 @@ func (ds *Shard) EnumLinks(ctx context.Context) ([]*ipld.Link, error) {
 	return links, nil
 }
 
+func (ds *Shard) EnumAll(ctx context.Context) ([]*ipld.Link, error) {
+	var links []*ipld.Link
+
+	linkResults := ds.EnumAllAsync(ctx)
+
+	for linkResult := range linkResults {
+		if linkResult.Err != nil {
+			return links, linkResult.Err
+		}
+		links = append(links, linkResult.Link)
+	}
+	return links, nil
+}
+
 // ForEachLink walks the Shard and calls the given function.
 func (ds *Shard) ForEachLink(ctx context.Context, f func(*ipld.Link) error) error {
 	return ds.walkTrie(ctx, func(sv *Shard) error {
@@ -348,6 +379,26 @@ func (ds *Shard) ForEachLink(ctx context.Context, f func(*ipld.Link) error) erro
 // EnumLinksAsync returns a channel which will receive Links in the directory
 // as they are enumerated, where order is not guaranteed
 func (ds *Shard) EnumLinksAsync(ctx context.Context) <-chan format.LinkResult {
+	linkResults := make(chan format.LinkResult)
+	ctx, cancel := context.WithCancel(ctx)
+	go func() {
+		defer close(linkResults)
+		defer cancel()
+		getLinks := makeAsyncTrieGetLinks(ds.dserv, linkResults)
+		cset := cid.NewSet()
+		// FIXME: Make concurrency an option for testing.
+		//err := dag.Walk(ctx, getLinks, ds.cid, cset.Visit, dag.Concurrent())
+		err := dag.Walk(ctx, getLinks, ds.cid, cset.Visit)
+		if err != nil {
+			emitResult(ctx, linkResults, format.LinkResult{Link: nil, Err: err})
+		}
+	}()
+	return linkResults
+}
+
+// EnumLinksAsync returns a channel which will receive Links in the directory
+// as they are enumerated, where order is not guaranteed
+func (ds *Shard) EnumAllAsync(ctx context.Context) <-chan format.LinkResult {
 	linkResults := make(chan format.LinkResult)
 	ctx, cancel := context.WithCancel(ctx)
 	go func() {
@@ -403,6 +454,39 @@ func makeAsyncTrieGetLinks(dagService ipld.DAGService, linkResults chan<- format
 	}
 }
 
+// same as makeAsyncTrieGetLinks but return all
+// FIXME: Check how to abstract this.
+func makeAsyncTrieGetAll(dagService ipld.DAGService, linkResults chan<- format.LinkResult) dag.GetLinks {
+
+	return func(ctx context.Context, currentCid cid.Cid) ([]*ipld.Link, error) {
+		node, err := dagService.Get(ctx, currentCid)
+		if err != nil {
+			return nil, err
+		}
+		directoryShard, err := NewHamtFromDag(dagService, node)
+		if err != nil {
+			return nil, err
+		}
+
+		childShards := make([]*ipld.Link, 0, directoryShard.childer.length())
+		links := directoryShard.childer.links
+		for idx := range directoryShard.childer.children {
+			lnk := links[idx]
+			// We don't care about the link type (shard or value), just count
+			//  *all* nodes in this HAMT.
+			emitResult(ctx, linkResults, format.LinkResult{Link: lnk, Err: nil})
+			lnkLinkType, err := directoryShard.childLinkType(lnk)
+			if err != nil {
+				return nil, err
+			}
+			if lnkLinkType == shardLink {
+				childShards = append(childShards, lnk)
+			}
+		}
+		return childShards, nil
+	}
+}
+
 func emitResult(ctx context.Context, linkResults chan<- format.LinkResult, r format.LinkResult) {
 	// make sure that context cancel is processed first
 	// the reason is due to the concurrency of EnumerateChildrenAsync
@@ -421,7 +505,7 @@ func emitResult(ctx context.Context, linkResults chan<- format.LinkResult, r for
 
 func (ds *Shard) walkTrie(ctx context.Context, cb func(*Shard) error) error {
 	return ds.childer.each(ctx, func(s *Shard) error {
-		if s.isValueNode() {
+		if s.IsValueNode() {
 			if err := cb(s); err != nil {
 				return err
 			}
@@ -453,7 +537,7 @@ func (ds *Shard) setValue(ctx context.Context, hv *hashBits, key string, value *
 		return
 	}
 
-	if child.isValueNode() {
+	if child.IsValueNode() {
 		// Leaf node. This is the base case of this recursive function.
 		// FIXME: Misleading: the base case also includes a recursive call
 		//  in the case both keys share the same slot in the new child shard
@@ -493,10 +577,7 @@ func (ds *Shard) setValue(ctx context.Context, hv *hashBits, key string, value *
 			return nil, err
 		}
 		child.builder = ds.builder
-		chhv := &hashBits{
-			b:        hash([]byte(grandChild.key)),
-			consumed: hv.consumed,
-		}
+		chhv := newConsumedHashBits(grandChild.key, hv.consumed)
 
 		// We explicitly ignore the oldValue returned by the next two insertions
 		// (which will be nil) to highlight there is no overwrite here: they are
@@ -545,7 +626,7 @@ func (ds *Shard) setValue(ctx context.Context, hv *hashBits, key string, value *
 				// Have we loaded the child? Prefer that.
 				schild := child.childer.child(0)
 				if schild != nil {
-					if schild.isValueNode() {
+					if schild.IsValueNode() {
 						ds.childer.set(schild, i)
 					}
 					return

hamt/util.go

@@ -1,9 +1,13 @@
 package hamt
 
 import (
+	"context"
+	"encoding/binary"
 	"fmt"
-
+	ipld "github.com/ipfs/go-ipld-format"
+	"github.com/ipfs/go-unixfs"
 	"github.com/spaolacci/murmur3"
+	"math"
 	"math/bits"
 )
 
@@ -13,6 +17,16 @@ type hashBits struct {
 	consumed int
 }
 
+func newHashBits(val string) *hashBits {
+	return &hashBits{b: HAMTHashFunction([]byte(val))}
+}
+
+func newConsumedHashBits(val string, consumed int) *hashBits {
+	hv := &hashBits{b: HAMTHashFunction([]byte(val))}
+	hv.consumed = consumed
+	return hv
+}
+
 func mkmask(n int) byte {
 	return (1 << uint(n)) - 1
 }
@@ -61,8 +75,66 @@ func logtwo(v int) (int, error) {
 	return lg2, nil
 }
 
-func hash(val []byte) []byte {
+func murmur3Hash(val []byte) []byte {
 	h := murmur3.New64()
 	h.Write(val)
 	return h.Sum(nil)
 }
+
+// ONLY FOR TESTING: Return the same value as the hash.
+func IdHash(val []byte) []byte {
+	return val
+}
+
+// TESTING ONLY. Creates a full HAMT tree with ID hashes. This uses very
+// low-level functions. Curretnly used for Basic-HAMT transition testing,
+// not for production.
+// FIXME: HAMTHashFunction needs to be set to IdHash for it to work.
+//  We don't store hasher information in the node as the murmur3 is
+//  hard-coded at the moment.
+func CreateFullShard(ds ipld.DAGService, treeHeight int) (rootNode ipld.Node, err error) {
+	// FIXME: We need to review how to do this.
+	oldHashFunc := HAMTHashFunction
+	defer func() { HAMTHashFunction = oldHashFunc }()
+	HAMTHashFunction = IdHash
+	// FIXME: If we didn't rehash inside setValue then we could just generate
+	//  the fake hash as in SetAndPrevious and pass it as an argument making
+	//  the hash independent of the tree manipulation (which should be the
+	//  correct way to go) and we wouldn't need this.
+
+	// FIXME: for now leave the childsPerNode at the default of 256 (1 byte per
+	//  level)
+	//childsPerNode := io.DefaultShardWidth
+	childsPerNode := 256 // FIXME: 'import cycle not allowed' from io package
+	if treeHeight < 1 {
+		panic("treeHeight < 1")
+	}
+	if treeHeight > 8 {
+		panic("treeHeight > 8: we don't allow a key larger than what can be enconded in a 64-bit word")
+	}
+	root, err := NewShard(ds, childsPerNode)
+	if err != nil {
+		return
+	}
+	// FIXME: Do we need to set the CID builder? Not part of the NewShard
+	//  interface so it shouldn't be mandatory.
+
+	// Use set value with the IdHash and the bitfield.
+	// FIXME: Set the biggest value possible within fixed sizes, we don't allow
+	//  trees bigger than that.
+	totalChildren := int(math.Pow(float64(childsPerNode), float64(treeHeight)))
+	for i:= 0; i < totalChildren; i++ {
+		var hashbuf [8]byte
+		binary.LittleEndian.PutUint64(hashbuf[:], uint64(i))
+		_, err = root.SetAndPrevious(context.Background(), string(hashbuf[:treeHeight]), unixfs.EmptyFileNode())
+		if err != nil {
+			return
+		}
+	}
+
+	rootNode, err = root.Node()
+	if err != nil {
+		return
+	}
+	return
+}

hamt/util_test.go

@@ -1,6 +1,10 @@
 package hamt
 
 import (
+	"context"
+	mdtest "github.com/ipfs/go-merkledag/test"
+	"github.com/stretchr/testify/assert"
+	"math"
 	"testing"
 )
 
@@ -62,3 +66,22 @@ func TestHashBitsUneven(t *testing.T) {
 		t.Fatalf("expected 20269, but got %b (%d)", v, v)
 	}
 }
+
+func TestCreateFullShard(t *testing.T) {
+	ds := mdtest.Mock()
+	childsPerNode := 256
+	treeHeight := 2 // This is the limit of what we can fastly generate,
+	// the default width is too big (256). We may need to refine
+	// CreateFullShard encoding of the key to reduce the tableSize.
+	node, err := CreateFullShard(ds, treeHeight)
+	assert.NoError(t, err)
+
+	shard, err := NewHamtFromDag(ds, node)
+	links, err := shard.EnumAll(context.Background())
+	assert.NoError(t, err)
+
+	childNodes := int(math.Pow(float64(childsPerNode), float64(treeHeight)))
+	internalNodes := int(math.Pow(float64(childsPerNode), float64(treeHeight-1)))
+	totalNodes := childNodes + internalNodes
+	assert.Equal(t, totalNodes, len(links))
+}