pathdb: add bloomfilter to pathdb difflayer

triedb/pathdb/difflayer.go

@@ -17,15 +17,91 @@
 package pathdb
 
 import (
+	"encoding/binary"
 	"fmt"
+	"math"
+	"math/rand"
 	"sync"
+	"time"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/trie/trienode"
 	"github.com/ethereum/go-ethereum/trie/triestate"
+	bloomfilter "github.com/holiman/bloomfilter/v2"
 )
 
+var (
+	// aggregatorMemoryLimit is the maximum size of the bottom-most diff layer
+	// that aggregates the writes from above until it's flushed into the disk
+	// layer.
+	//
+	// Note, bumping this up might drastically increase the size of the bloom
+	// filters that's stored in every diff layer. Don't do that without fully
+	// understanding all the implications.
+	aggregatorMemoryLimit = uint64(4 * 1024 * 1024)
+
+	// aggregatorItemLimit is an approximate number of items that will end up
+	// in the aggregator layer before it's flushed out to disk. A plain account
+	// weighs around 14B (+hash), a storage slot 32B (+hash), a deleted slot
+	// 0B (+hash). Slots are mostly set/unset in lockstep, so that average at
+	// 16B (+hash). All in all, the average entry seems to be 15+32=47B. Use a
+	// smaller number to be on the safe side.
+	aggregatorItemLimit = aggregatorMemoryLimit / 42
+
+	// bloomTargetError is the target false positive rate when the aggregator
+	// layer is at its fullest. The actual value will probably move around up
+	// and down from this number, it's mostly a ballpark figure.
+	//
+	// Note, dropping this down might drastically increase the size of the bloom
+	// filters that's stored in every diff layer. Don't do that without fully
+	// understanding all the implications.
+	bloomTargetError = 0.02
+
+	// bloomSize is the ideal bloom filter size given the maximum number of items
+	// it's expected to hold and the target false positive error rate.
+	bloomSize = math.Ceil(float64(aggregatorItemLimit) * math.Log(bloomTargetError) / math.Log(1/math.Pow(2, math.Log(2))))
+
+	// bloomFuncs is the ideal number of bits a single entry should set in the
+	// bloom filter to keep its size to a minimum (given it's size and maximum
+	// entry count).
+	bloomFuncs = math.Round((bloomSize / float64(aggregatorItemLimit)) * math.Log(2))
+
+	// the bloom offsets are runtime constants which determines which part of the
+	// account/storage hash the hasher functions looks at, to determine the
+	// bloom key for an account/slot. This is randomized at init(), so that the
+	// global population of nodes do not all display the exact same behaviour with
+	// regards to bloom content
+	bloomNodeHasherOffset = 0
+)
+
+func init() {
+	// Init the bloom offsets in the range [0:24] (requires 8 bytes)
+	bloomNodeHasherOffset = rand.Intn(25)
+}
+
+func nodeBloomHash(h common.Hash, p []byte) uint64 {
+	return binary.BigEndian.Uint64(h[bloomNodeHasherOffset:bloomNodeHasherOffset+8]) ^ pathBloomHash(p)
+}
+
+func pathBloomHash(p []byte) uint64 {
+	if len(p)&1 != 0 {
+		panic("can't convert hex key of odd length")
+	}
+	hashLen := len(p) / 2
+	if hashLen > 8 {
+		panic("hash value too long")
+	}
+
+	var hashValue uint64
+	for i := 0; i < hashLen; i++ {
+		hashValue <<= 8
+		hashValue |= uint64(p[i*2])<<4 | uint64(p[i*2+1])
+	}
+
+	return (uint64(hashLen) << 32) | (hashValue << 1)
+}
+
 // diffLayer represents a collection of modifications made to the in-memory tries
 // along with associated state changes after running a block on top.
 //
@@ -42,6 +118,10 @@ type diffLayer struct {
 
 	parent layer        // Parent layer modified by this one, never nil, **can be changed**
 	lock   sync.RWMutex // Lock used to protect parent
+
+	origin     *diskLayer
+	diffed     *bloomfilter.Filter // Bloom filter tracking all the diffed items up to the disk layer
+	selfDiffed *bloomfilter.Filter // Bloom filter tracking all the diffed items of its own
 }
 
 // newDiffLayer creates a new diff layer on top of an existing layer.
@@ -58,6 +138,15 @@ func newDiffLayer(parent layer, root common.Hash, id uint64, block uint64, nodes
 		states: states,
 		parent: parent,
 	}
+	switch l := parent.(type) {
+	case *diskLayer:
+		dl.rebloom(l)
+	case *diffLayer:
+		dl.rebloom(l.origin)
+	default:
+		panic("unknown parent type")
+	}
+
 	for _, subset := range nodes {
 		for path, n := range subset {
 			dl.memory += uint64(n.Size() + len(path))
@@ -75,6 +164,53 @@ func newDiffLayer(parent layer, root common.Hash, id uint64, block uint64, nodes
 	return dl
 }
 
+// rebloom discards the layer's current bloom and rebuilds it from scratch based
+// on the parent's and the local diffs.
+func (dl *diffLayer) rebloom(origin *diskLayer) {
+	dl.lock.Lock()
+	defer dl.lock.Unlock()
+
+	defer func(start time.Time) {
+		bloomIndexTimer.Update(time.Since(start))
+	}(time.Now())
+
+	// Inject the new origin that triggered the rebloom
+	dl.origin = origin
+
+	// Retrieve the parent bloom or create a fresh empty one
+	if parent, ok := dl.parent.(*diffLayer); ok {
+		parent.lock.RLock()
+		dl.diffed, _ = parent.diffed.Copy()
+		parent.lock.RUnlock()
+	} else {
+		if dl.selfDiffed == nil {
+			dl.diffed, _ = bloomfilter.New(uint64(bloomSize), uint64(bloomFuncs))
+		} else {
+			dl.diffed, _ = dl.selfDiffed.NewCompatible()
+		}
+	}
+
+	if dl.selfDiffed == nil {
+		dl.selfDiffed, _ = dl.diffed.NewCompatible()
+		for owner, subset := range dl.nodes {
+			for path, _ := range subset {
+				dl.selfDiffed.AddHash(nodeBloomHash(owner, []byte(path)))
+			}
+		}
+	}
+	err := dl.diffed.UnionInPlace(dl.selfDiffed)
+	if err != nil {
+		log.Error("diff layer bloom filter failed to union in place", "id", dl.id, "err", err)
+	}
+	// Calculate the current false positive rate and update the error rate meter.
+	// This is a bit cheating because subsequent layers will overwrite it, but it
+	// should be fine, we're only interested in ballpark figures.
+	k := float64(dl.diffed.K())
+	n := float64(dl.diffed.N())
+	m := float64(dl.diffed.M())
+	bloomErrorGauge.Update(math.Pow(1.0-math.Exp((-k)*(n+0.5)/(m-1)), k))
+}
+
 // rootHash implements the layer interface, returning the root hash of
 // corresponding state.
 func (dl *diffLayer) rootHash() common.Hash {
@@ -97,7 +233,7 @@ func (dl *diffLayer) parentLayer() layer {
 
 // node implements the layer interface, retrieving the trie node blob with the
 // provided node information. No error will be returned if the node is not found.
-func (dl *diffLayer) node(owner common.Hash, path []byte, depth int) ([]byte, common.Hash, *nodeLoc, error) {
+func (dl *diffLayer) nodeInternal(owner common.Hash, path []byte, depth int) ([]byte, common.Hash, *nodeLoc, error) {
 	// Hold the lock, ensure the parent won't be changed during the
 	// state accessing.
 	dl.lock.RLock()
@@ -114,10 +250,33 @@ func (dl *diffLayer) node(owner common.Hash, path []byte, depth int) ([]byte, co
 			return n.Blob, n.Hash, &nodeLoc{loc: locDiffLayer, depth: depth}, nil
 		}
 	}
+
 	// Trie node unknown to this layer, resolve from parent
+	if diff, ok := dl.parent.(*diffLayer); ok {
+		return diff.nodeInternal(owner, path, depth+1)
+	}
+	// Failed to resolve through diff layers, fallback to disk layer
 	return dl.parent.node(owner, path, depth+1)
 }
 
+// Node implements the layer interface, retrieving the trie node blob with the
+// provided node information. No error will be returned if the node is not found.
+func (dl *diffLayer) node(owner common.Hash, path []byte, depth int) ([]byte, common.Hash, *nodeLoc, error) {
+	dl.lock.RLock()
+	defer dl.lock.RUnlock()
+
+	var origin *diskLayer
+	hit := dl.diffed.ContainsHash(nodeBloomHash(owner, path))
+	if !hit {
+		origin = dl.origin // extract origin while holding the lock
+	}
+
+	if origin != nil {
+		return origin.node(owner, path, depth+1)
+	}
+	return dl.nodeInternal(owner, path, 0)
+}
+
 // update implements the layer interface, creating a new layer on top of the
 // existing layer tree with the specified data items.
 func (dl *diffLayer) update(root common.Hash, id uint64, block uint64, nodes map[common.Hash]map[string]*trienode.Node, states *triestate.Set) *diffLayer {

triedb/pathdb/difflayer_test.go

@@ -18,6 +18,7 @@ package pathdb
 
 import (
 	"bytes"
+	"fmt"
 	"testing"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -25,6 +26,7 @@ import (
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/internal/testrand"
 	"github.com/ethereum/go-ethereum/trie/trienode"
+	"github.com/stretchr/testify/assert"
 )
 
 func emptyLayer() *diskLayer {
@@ -170,3 +172,20 @@ func BenchmarkJournal(b *testing.B) {
 		layer.journal(new(bytes.Buffer))
 	}
 }
+
+func TestBloomHash(t *testing.T) {
+	n1 := nodeBloomHash(common.Hash{}, common.Hex2Bytes("0101"))
+	n2 := nodeBloomHash(common.Hash{}, common.Hex2Bytes("00000101"))
+	n3 := nodeBloomHash(common.Hash{}, common.Hex2Bytes("0000000000000000000000000101"))
+	n4 := nodeBloomHash(common.HexToHash("0xfffffffffffffffffffffffffffffffffffffffffffffff1ffffdfffffffffaf"), common.Hex2Bytes("0101"))
+	n5 := nodeBloomHash(common.HexToHash("0xf1234ffffffffffffffffffffffffffffffffffffffffff1ffffdfffffffffaf"), common.Hex2Bytes("00000101"))
+	fmt.Printf("hash: %x\n", n1)
+	fmt.Printf("hash: %x\n", n2)
+	fmt.Printf("hash: %x\n", n3)
+	fmt.Printf("hash: %x\n", n4)
+	fmt.Printf("hash: %x\n", n5)
+	assert.NotEqual(t, n1, n2)
+	assert.NotEqual(t, n1, n3)
+	assert.NotEqual(t, n1, n4)
+	assert.NotEqual(t, n4, n5)
+}

triedb/pathdb/layertree.go

@@ -145,6 +145,7 @@ func (tree *layerTree) cap(root common.Hash, layers int) error {
 			return nil
 		}
 	}
+	var persisted *diskLayer
 	// We're out of layers, flatten anything below, stopping if it's the disk or if
 	// the memory limit is not yet exceeded.
 	switch parent := diff.parentLayer().(type) {
@@ -163,6 +164,7 @@ func (tree *layerTree) cap(root common.Hash, layers int) error {
 		}
 		tree.layers[base.rootHash()] = base
 		diff.parent = base
+		persisted = base.(*diskLayer)
 
 		diff.lock.Unlock()
 
@@ -190,6 +192,19 @@ func (tree *layerTree) cap(root common.Hash, layers int) error {
 			remove(root)
 		}
 	}
+	// If the disk layer was modified, regenerate all the cumulative blooms
+	if persisted != nil {
+		var rebloom func(root common.Hash)
+		rebloom = func(root common.Hash) {
+			if diff, ok := tree.layers[root].(*diffLayer); ok {
+				diff.rebloom(persisted)
+			}
+			for _, child := range children[root] {
+				rebloom(child)
+			}
+		}
+		rebloom(persisted.root)
+	}
 	return nil
 }
 

triedb/pathdb/metrics.go

@@ -48,4 +48,7 @@ var (
 	historyBuildTimeMeter  = metrics.NewRegisteredTimer("pathdb/history/time", nil)
 	historyDataBytesMeter  = metrics.NewRegisteredMeter("pathdb/history/bytes/data", nil)
 	historyIndexBytesMeter = metrics.NewRegisteredMeter("pathdb/history/bytes/index", nil)
+
+	bloomIndexTimer = metrics.NewRegisteredResettingTimer("pathdb/bloom/index", nil)
+	bloomErrorGauge = metrics.NewRegisteredGaugeFloat64("pathdb/bloom/error", nil)
 )