trie: experimental parallel commit

trie: test which shows that the commit is erroneous
trie: make parallel committer correct
trie/trienode: better merge of set
fix error in bench

trie/committer.go

@@ -44,31 +44,17 @@ func newCommitter(nodes *trienode.NodeSet, tracer *tracer, collectLeaf bool, par
 	}
 }
 
-type wrapNode struct {
-	node     *trienode.Node
-	path     string
-	leafHash common.Hash // optional, the parent hash of the related leaf
-	leafBlob []byte      // optional, the blob of the related leaf
-}
-
 // Commit collapses a node down into a hash node.
 func (c *committer) Commit(n node) hashNode {
-	hn, wnodes := c.commit(nil, n, true)
-	for _, wn := range wnodes {
-		c.nodes.AddNode(wn.path, wn.node)
-		if wn.leafHash != (common.Hash{}) {
-			c.nodes.AddLeaf(wn.leafHash, wn.leafBlob)
-		}
-	}
-	return hn.(hashNode)
+	return c.commit(nil, n, true).(hashNode)
 }
 
 // commit collapses a node down into a hash node and returns it.
-func (c *committer) commit(path []byte, n node, topmost bool) (node, []*wrapNode) {
+func (c *committer) commit(path []byte, n node, topmost bool) node {
 	// if this path is clean, use available cached data
 	hash, dirty := n.cache()
 	if hash != nil && !dirty {
-		return hash, nil
+		return hash
 	}
 	// Commit children, then parent, and remove the dirty flag.
 	switch cn := n.(type) {
@@ -78,36 +64,29 @@ func (c *committer) commit(path []byte, n node, topmost bool) (node, []*wrapNode
 
 		// If the child is fullNode, recursively commit,
 		// otherwise it can only be hashNode or valueNode.
-		var nodes []*wrapNode
 		if _, ok := cn.Val.(*fullNode); ok {
-			collapsed.Val, nodes = c.commit(append(path, cn.Key...), cn.Val, false)
+			collapsed.Val = c.commit(append(path, cn.Key...), cn.Val, false)
 		}
 		// The key needs to be copied, since we're adding it to the
 		// modified nodeset.
 		collapsed.Key = hexToCompact(cn.Key)
-		hashedNode, wNode := c.store(path, collapsed)
-		if wNode != nil {
-			nodes = append(nodes, wNode)
-		}
+		hashedNode := c.store(path, collapsed)
 		if hn, ok := hashedNode.(hashNode); ok {
-			return hn, nodes
+			return hn
 		}
-		return collapsed, nodes
+		return collapsed
 	case *fullNode:
-		hashedKids, nodes := c.commitChildren(path, cn, topmost && c.parallel)
+		hashedKids := c.commitChildren(path, cn, topmost && c.parallel)
 		collapsed := cn.copy()
 		collapsed.Children = hashedKids
 
-		hashedNode, wNode := c.store(path, collapsed)
-		if wNode != nil {
-			nodes = append(nodes, wNode)
-		}
+		hashedNode := c.store(path, collapsed)
 		if hn, ok := hashedNode.(hashNode); ok {
-			return hn, nodes
+			return hn
 		}
-		return collapsed, nodes
+		return collapsed
 	case hashNode:
-		return cn, nil
+		return cn
 	default:
 		// nil, valuenode shouldn't be committed
 		panic(fmt.Sprintf("%T: invalid node: %v", n, n))
@@ -115,11 +94,10 @@ func (c *committer) commit(path []byte, n node, topmost bool) (node, []*wrapNode
 }
 
 // commitChildren commits the children of the given fullnode
-func (c *committer) commitChildren(path []byte, n *fullNode, parallel bool) ([17]node, []*wrapNode) {
+func (c *committer) commitChildren(path []byte, n *fullNode, parallel bool) [17]node {
 	var (
 		wg       sync.WaitGroup
 		children [17]node
-		results  [16][]*wrapNode
 	)
 	for i := 0; i < 16; i++ {
 		child := n.Children[i]
@@ -137,34 +115,34 @@ func (c *committer) commitChildren(path []byte, n *fullNode, parallel bool) ([17
 		// Note the returned node can be some embedded nodes, so it's
 		// possible the type is not hashNode.
 		if !parallel {
-			children[i], results[i] = c.commit(append(path, byte(i)), child, false)
+			children[i] = c.commit(append(path, byte(i)), child, false)
 		} else {
 			wg.Add(1)
 			go func(index int) {
-				defer wg.Done()
-				children[index], results[index] = c.commit(append(path, byte(index)), child, false)
+				p := append(path, byte(i))
+				set := trienode.NewNodeSet(c.nodes.Owner)
+				childComitter := newCommitter(set, c.tracer, c.collectLeaf, false)
+				h := childComitter.commit(p, child, false)
+				children[index] = h
+				c.nodes.MergeSet(set)
+				wg.Done()
 			}(i)
 		}
 	}
 	if parallel {
 		wg.Wait()
+
 	}
 	// For the 17th child, it's possible the type is valuenode.
 	if n.Children[16] != nil {
 		children[16] = n.Children[16]
 	}
-	var wnodes []*wrapNode
-	for i := 0; i < 16; i++ {
-		if results[i] != nil {
-			wnodes = append(wnodes, results[i]...)
-		}
-	}
-	return children, wnodes
+	return children
 }
 
 // store hashes the node n and adds it to the modified nodeset. If leaf collection
 // is enabled, leaf nodes will be tracked in the modified nodeset as well.
-func (c *committer) store(path []byte, n node) (node, *wrapNode) {
+func (c *committer) store(path []byte, n node) node {
 	// Larger nodes are replaced by their hash and stored in the database.
 	var hash, _ = n.cache()
 
@@ -178,30 +156,24 @@ func (c *committer) store(path []byte, n node) (node, *wrapNode) {
 		// deleted only if the node was existent in database before.
 		_, ok := c.tracer.accessList[string(path)]
 		if ok {
-			return n, &wrapNode{
-				path: string(path),
-				node: trienode.NewDeleted(),
+			c.nodes.AddNode(path, trienode.NewDeleted()) // TODO
 		}
-		}
-		return n, nil
+		return n
 	}
 	nhash := common.BytesToHash(hash)
-	wNode := &wrapNode{
-		path: string(path),
-		node: trienode.New(nhash, nodeToBytes(n)),
-	}
+	c.nodes.AddNode(path, trienode.New(nhash, nodeToBytes(n))) // TODO
+
 	// Collect the corresponding leaf node if it's required. We don't check
 	// full node since it's impossible to store value in fullNode. The key
 	// length of leaves should be exactly same.
 	if c.collectLeaf {
 		if sn, ok := n.(*shortNode); ok {
 			if val, ok := sn.Val.(valueNode); ok {
-				wNode.leafHash = nhash
-				wNode.leafBlob = val
+				c.nodes.AddLeaf(nhash, val) // TODO
 			}
 		}
 	}
-	return hash, wNode
+	return hash
 }
 
 // ForGatherChildren decodes the provided node and traverses the children inside.

trie/trie.go

@@ -613,7 +613,7 @@ func (t *Trie) Commit(collectLeaf bool) (common.Hash, *trienode.NodeSet) {
 		}
 		nodes := trienode.NewNodeSet(t.owner)
 		for _, path := range paths {
-			nodes.AddNode(path, trienode.NewDeleted())
+			nodes.AddNode([]byte(path), trienode.NewDeleted())
 		}
 		return types.EmptyRootHash, nodes // case (b)
 	}
@@ -631,7 +631,7 @@ func (t *Trie) Commit(collectLeaf bool) (common.Hash, *trienode.NodeSet) {
 	}
 	nodes := trienode.NewNodeSet(t.owner)
 	for _, path := range t.tracer.deletedNodes() {
-		nodes.AddNode(path, trienode.NewDeleted())
+		nodes.AddNode([]byte(path), trienode.NewDeleted())
 	}
 	// If the number of changes is below 100, we let one thread handle it
 	t.root = newCommitter(nodes, t.tracer, collectLeaf, t.mutate > 100).Commit(t.root)

trie/trie_test.go

@@ -40,6 +40,7 @@ import (
 	"github.com/ethereum/go-ethereum/trie/trienode"
 	"github.com/holiman/uint256"
 	"golang.org/x/crypto/sha3"
+	"strings"
 )
 
 func init() {
@@ -1236,11 +1237,11 @@ func FuzzTrie(f *testing.F) {
 // BenchmarkCommit/commit-5000nodes-parallel
 // BenchmarkCommit/commit-5000nodes-parallel-8      	     450	   2725071 ns/op	 6471357 B/op	   74938 allocs/op
 func BenchmarkCommit(b *testing.B) {
-	benchmarkCommit(b, 100)
-	benchmarkCommit(b, 200)
-	benchmarkCommit(b, 500)
-	benchmarkCommit(b, 1000)
-	benchmarkCommit(b, 2000)
+	//benchmarkCommit(b, 100)
+	//benchmarkCommit(b, 200)
+	//benchmarkCommit(b, 500)
+	//benchmarkCommit(b, 1000)
+	//benchmarkCommit(b, 2000)
 	benchmarkCommit(b, 5000)
 }
 
@@ -1269,8 +1270,73 @@ func testCommit(b *testing.B, n int, parallel bool) {
 	}
 	b.ResetTimer()
 	b.ReportAllocs()
-
 	for i := 0; i < len(tries); i++ {
 		tries[i].Commit(true)
 	}
 }
+
+func TestCommitCorrect(t *testing.T) {
+	var paraTrie = NewEmpty(nil)
+	var refTrie = NewEmpty(nil)
+
+	for j := 0; j < 5000; j++ {
+		key := testrand.Bytes(32)
+		val := testrand.Bytes(32)
+		paraTrie.Update(key, val)
+		refTrie.Update(common.CopyBytes(key), common.CopyBytes(val))
+	}
+	paraTrie.Hash()
+	//paraTrie.mutate = 0
+	refTrie.Hash()
+	refTrie.mutate = 0
+
+	haveRoot, haveNodes := paraTrie.Commit(true)
+	wantRoot, wantNodes := refTrie.Commit(true)
+
+	if haveRoot != wantRoot {
+		t.Fatalf("have %x want %x", haveRoot, wantRoot)
+	}
+	have := printSet(haveNodes)
+	want := printSet(wantNodes)
+	if have != want {
+		i := 0
+		for i = 0; i < len(have); i++ {
+			if have[i] != want[i] {
+				break
+			}
+		}
+		if i > 100 {
+			i -= 100
+		}
+		t.Fatalf("have != want\nhave %q\nwant %q", have[i:], want[i:])
+	}
+}
+func printSet(set *trienode.NodeSet) string {
+	var out = new(strings.Builder)
+	fmt.Fprintf(out, "nodeset owner: %v\n", set.Owner)
+	var paths []string
+	for k, _ := range set.Nodes {
+		paths = append(paths, k)
+	}
+	sort.Strings(paths)
+
+	for _, path := range paths {
+		n := set.Nodes[path]
+		// Deletion
+		if n.IsDeleted() {
+			fmt.Fprintf(out, "  [-]: %x\n", path)
+			continue
+		}
+		// Insertion or update
+		fmt.Fprintf(out, "  [+/*]: %x -> %v \n", path, n.Hash)
+	}
+	sort.Slice(set.Leaves, func(i, j int) bool {
+		a := set.Leaves[i]
+		b := set.Leaves[j]
+		return bytes.Compare(a.Parent[:], b.Parent[:]) < 0
+	})
+	for _, n := range set.Leaves {
+		fmt.Fprintf(out, "[leaf]: %v\n", n)
+	}
+	return out.String()
+}

trie/trienode/node.go

@@ -22,6 +22,7 @@ import (
 	"strings"
 
 	"github.com/ethereum/go-ethereum/common"
+	"sync"
 )
 
 // Node is a wrapper which contains the encoded blob of the trie node and its
@@ -59,6 +60,8 @@ type leaf struct {
 // NodeSet contains a set of nodes collected during the commit operation.
 // Each node is keyed by path. It's not thread-safe to use.
 type NodeSet struct {
+	mu sync.Mutex
+
 	Owner   common.Hash
 	Leaves  []*leaf
 	Nodes   map[string]*Node
@@ -90,13 +93,40 @@ func (set *NodeSet) ForEachWithOrder(callback func(path string, n *Node)) {
 }
 
 // AddNode adds the provided node into set.
-func (set *NodeSet) AddNode(path string, n *Node) {
+func (set *NodeSet) AddNode(path []byte, n *Node) {
 	if n.IsDeleted() {
 		set.deletes += 1
 	} else {
 		set.updates += 1
 	}
-	set.Nodes[path] = n
+	set.Nodes[string(path)] = n
+}
+
+func (set *NodeSet) MergeSet(other *NodeSet) error {
+	if set.Owner != other.Owner {
+		return fmt.Errorf("nodesets belong to different owner are not mergeable %x-%x", set.Owner, other.Owner)
+	}
+	set.mu.Lock()
+	defer set.mu.Unlock()
+	for path, node := range other.Nodes {
+		prev, ok := set.Nodes[path]
+		if ok {
+			// overwrite happens, revoke the counter
+			if prev.IsDeleted() {
+				set.deletes -= 1
+			} else {
+				set.updates -= 1
+			}
+		}
+		if node.IsDeleted() {
+			set.deletes += 1
+		} else {
+			set.updates += 1
+		}
+		set.Nodes[path] = node
+	}
+	set.Leaves = append(set.Leaves, other.Leaves...)
+	return nil
 }
 
 // Merge adds a set of nodes into the set.
@@ -104,6 +134,8 @@ func (set *NodeSet) Merge(owner common.Hash, nodes map[string]*Node) error {
 	if set.Owner != owner {
 		return fmt.Errorf("nodesets belong to different owner are not mergeable %x-%x", set.Owner, owner)
 	}
+	set.mu.Lock()
+	defer set.mu.Unlock()
 	for path, node := range nodes {
 		prev, ok := set.Nodes[path]
 		if ok {

trie/verkle.go

@@ -258,7 +258,7 @@ func (t *VerkleTrie) Commit(_ bool) (common.Hash, *trienode.NodeSet) {
 	nodeset := trienode.NewNodeSet(common.Hash{})
 	for _, node := range nodes {
 		// Hash parameter is not used in pathdb
-		nodeset.AddNode(string(node.Path), trienode.New(common.Hash{}, node.SerializedBytes))
+		nodeset.AddNode(node.Path, trienode.New(common.Hash{}, node.SerializedBytes))
 	}
 	// Serialize root commitment form
 	return t.Hash(), nodeset