aboutsummaryrefslogtreecommitdiffstats
path: root/eth
diff options
context:
space:
mode:
authorJeffrey Wilcke <jeffrey@ethereum.org>2015-10-17 03:35:24 +0800
committerJeffrey Wilcke <jeffrey@ethereum.org>2015-10-17 03:35:24 +0800
commit10ed107ba2001d1aabba3d319ba88c5ce6e8fdc0 (patch)
tree7c79e36cadcc57a8c936f06838a061cdf7e3fd8a /eth
parentc5ef2afda56c069fda5528c48bd3d831b81455dc (diff)
parent6dc14788a238f3e0ec786c6c04d476a3b957e645 (diff)
downloaddexon-10ed107ba2001d1aabba3d319ba88c5ce6e8fdc0.tar
dexon-10ed107ba2001d1aabba3d319ba88c5ce6e8fdc0.tar.gz
dexon-10ed107ba2001d1aabba3d319ba88c5ce6e8fdc0.tar.bz2
dexon-10ed107ba2001d1aabba3d319ba88c5ce6e8fdc0.tar.lz
dexon-10ed107ba2001d1aabba3d319ba88c5ce6e8fdc0.tar.xz
dexon-10ed107ba2001d1aabba3d319ba88c5ce6e8fdc0.tar.zst
dexon-10ed107ba2001d1aabba3d319ba88c5ce6e8fdc0.zip
Merge pull request #1899 from obscuren/mipmap-bloom
core, eth/filters, miner, xeth: Optimised log filtering
Diffstat (limited to 'eth')
-rw-r--r--eth/backend.go46
-rw-r--r--eth/backend_test.go67
-rw-r--r--eth/filters/filter.go120
-rw-r--r--eth/filters/filter_test.go267
4 files changed, 447 insertions, 53 deletions
diff --git a/eth/backend.go b/eth/backend.go
index dc16ba4cf..9ec3c1440 100644
--- a/eth/backend.go
+++ b/eth/backend.go
@@ -47,6 +47,7 @@ import (
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/nat"
+ "github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/whisper"
)
@@ -288,6 +289,9 @@ func New(config *Config) (*Ethereum, error) {
if err := upgradeChainDatabase(chainDb); err != nil {
return nil, err
}
+ if err := addMipmapBloomBins(chainDb); err != nil {
+ return nil, err
+ }
dappDb, err := newdb(filepath.Join(config.DataDir, "dapp"))
if err != nil {
@@ -758,3 +762,45 @@ func upgradeChainDatabase(db ethdb.Database) error {
}
return nil
}
+
+func addMipmapBloomBins(db ethdb.Database) (err error) {
+ const mipmapVersion uint = 2
+
+ // check if the version is set. We ignore data for now since there's
+ // only one version so we can easily ignore it for now
+ var data []byte
+ data, _ = db.Get([]byte("setting-mipmap-version"))
+ if len(data) > 0 {
+ var version uint
+ if err := rlp.DecodeBytes(data, &version); err == nil && version == mipmapVersion {
+ return nil
+ }
+ }
+
+ defer func() {
+ if err == nil {
+ var val []byte
+ val, err = rlp.EncodeToBytes(mipmapVersion)
+ if err == nil {
+ err = db.Put([]byte("setting-mipmap-version"), val)
+ }
+ return
+ }
+ }()
+ latestBlock := core.GetBlock(db, core.GetHeadBlockHash(db))
+ if latestBlock == nil { // clean database
+ return
+ }
+
+ tstart := time.Now()
+ glog.V(logger.Info).Infoln("upgrading db log bloom bins")
+ for i := uint64(0); i <= latestBlock.NumberU64(); i++ {
+ hash := core.GetCanonicalHash(db, i)
+ if (hash == common.Hash{}) {
+ return fmt.Errorf("chain db corrupted. Could not find block %d.", i)
+ }
+ core.WriteMipmapBloom(db, i, core.GetBlockReceipts(db, hash))
+ }
+ glog.V(logger.Info).Infoln("upgrade completed in", time.Since(tstart))
+ return nil
+}
diff --git a/eth/backend_test.go b/eth/backend_test.go
new file mode 100644
index 000000000..220426c17
--- /dev/null
+++ b/eth/backend_test.go
@@ -0,0 +1,67 @@
+package eth
+
+import (
+ "math/big"
+ "testing"
+
+ "github.com/ethereum/go-ethereum/common"
+ "github.com/ethereum/go-ethereum/core"
+ "github.com/ethereum/go-ethereum/core/types"
+ "github.com/ethereum/go-ethereum/core/vm"
+ "github.com/ethereum/go-ethereum/ethdb"
+)
+
+func TestMipmapUpgrade(t *testing.T) {
+ db, _ := ethdb.NewMemDatabase()
+ addr := common.BytesToAddress([]byte("jeff"))
+ genesis := core.WriteGenesisBlockForTesting(db)
+
+ chain := core.GenerateChain(genesis, db, 10, func(i int, gen *core.BlockGen) {
+ var receipts types.Receipts
+ switch i {
+ case 1:
+ receipt := types.NewReceipt(nil, new(big.Int))
+ receipt.SetLogs(vm.Logs{&vm.Log{Address: addr}})
+ gen.AddUncheckedReceipt(receipt)
+ receipts = types.Receipts{receipt}
+ case 2:
+ receipt := types.NewReceipt(nil, new(big.Int))
+ receipt.SetLogs(vm.Logs{&vm.Log{Address: addr}})
+ gen.AddUncheckedReceipt(receipt)
+ receipts = types.Receipts{receipt}
+ }
+
+ // store the receipts
+ err := core.PutReceipts(db, receipts)
+ if err != nil {
+ t.Fatal(err)
+ }
+ })
+ for _, block := range chain {
+ core.WriteBlock(db, block)
+ if err := core.WriteCanonicalHash(db, block.Hash(), block.NumberU64()); err != nil {
+ t.Fatalf("failed to insert block number: %v", err)
+ }
+ if err := core.WriteHeadBlockHash(db, block.Hash()); err != nil {
+ t.Fatalf("failed to insert block number: %v", err)
+ }
+ if err := core.PutBlockReceipts(db, block, block.Receipts()); err != nil {
+ t.Fatal("error writing block receipts:", err)
+ }
+ }
+
+ err := addMipmapBloomBins(db)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ bloom := core.GetMipmapBloom(db, 1, core.MIPMapLevels[0])
+ if (bloom == types.Bloom{}) {
+ t.Error("got empty bloom filter")
+ }
+
+ data, _ := db.Get([]byte("setting-mipmap-version"))
+ if len(data) == 0 {
+ t.Error("setting-mipmap-version not written to database")
+ }
+}
diff --git a/eth/filters/filter.go b/eth/filters/filter.go
index 2bcf20d0c..2e81ea177 100644
--- a/eth/filters/filter.go
+++ b/eth/filters/filter.go
@@ -32,13 +32,10 @@ type AccountChange struct {
// Filtering interface
type Filter struct {
- db ethdb.Database
- earliest int64
- latest int64
- skip int
- address []common.Address
- max int
- topics [][]common.Hash
+ db ethdb.Database
+ begin, end int64
+ addresses []common.Address
+ topics [][]common.Hash
BlockCallback func(*types.Block, vm.Logs)
TransactionCallback func(*types.Transaction)
@@ -54,61 +51,82 @@ func New(db ethdb.Database) *Filter {
// Set the earliest and latest block for filtering.
// -1 = latest block (i.e., the current block)
// hash = particular hash from-to
-func (self *Filter) SetEarliestBlock(earliest int64) {
- self.earliest = earliest
+func (self *Filter) SetBeginBlock(begin int64) {
+ self.begin = begin
}
-func (self *Filter) SetLatestBlock(latest int64) {
- self.latest = latest
+func (self *Filter) SetEndBlock(end int64) {
+ self.end = end
}
-func (self *Filter) SetAddress(addr []common.Address) {
- self.address = addr
+func (self *Filter) SetAddresses(addr []common.Address) {
+ self.addresses = addr
}
func (self *Filter) SetTopics(topics [][]common.Hash) {
self.topics = topics
}
-func (self *Filter) SetMax(max int) {
- self.max = max
-}
-
-func (self *Filter) SetSkip(skip int) {
- self.skip = skip
-}
-
// Run filters logs with the current parameters set
func (self *Filter) Find() vm.Logs {
- earliestBlock := core.GetBlock(self.db, core.GetHeadBlockHash(self.db))
- var earliestBlockNo uint64 = uint64(self.earliest)
- if self.earliest == -1 {
- earliestBlockNo = earliestBlock.NumberU64()
+ latestBlock := core.GetBlock(self.db, core.GetHeadBlockHash(self.db))
+ var beginBlockNo uint64 = uint64(self.begin)
+ if self.begin == -1 {
+ beginBlockNo = latestBlock.NumberU64()
}
- var latestBlockNo uint64 = uint64(self.latest)
- if self.latest == -1 {
- latestBlockNo = earliestBlock.NumberU64()
+ var endBlockNo uint64 = uint64(self.end)
+ if self.end == -1 {
+ endBlockNo = latestBlock.NumberU64()
}
- var (
- logs vm.Logs
- block *types.Block
- )
- hash := core.GetCanonicalHash(self.db, latestBlockNo)
- if hash != (common.Hash{}) {
- block = core.GetBlock(self.db, hash)
+ // if no addresses are present we can't make use of fast search which
+ // uses the mipmap bloom filters to check for fast inclusion and uses
+ // higher range probability in order to ensure at least a false positive
+ if len(self.addresses) == 0 {
+ return self.getLogs(beginBlockNo, endBlockNo)
+ }
+ return self.mipFind(beginBlockNo, endBlockNo, 0)
+}
+
+func (self *Filter) mipFind(start, end uint64, depth int) (logs vm.Logs) {
+ level := core.MIPMapLevels[depth]
+ // normalise numerator so we can work in level specific batches and
+ // work with the proper range checks
+ for num := start / level * level; num <= end; num += level {
+ // find addresses in bloom filters
+ bloom := core.GetMipmapBloom(self.db, num, level)
+ for _, addr := range self.addresses {
+ if bloom.TestBytes(addr[:]) {
+ // range check normalised values and make sure that
+ // we're resolving the correct range instead of the
+ // normalised values.
+ start := uint64(math.Max(float64(num), float64(start)))
+ end := uint64(math.Min(float64(num+level-1), float64(end)))
+ if depth+1 == len(core.MIPMapLevels) {
+ logs = append(logs, self.getLogs(start, end)...)
+ } else {
+ logs = append(logs, self.mipFind(start, end, depth+1)...)
+ }
+ // break so we don't check the same range for each
+ // possible address. Checks on multiple addresses
+ // are handled further down the stack.
+ break
+ }
+ }
}
-done:
- for i := 0; block != nil; i++ {
- // Quit on latest
- switch {
- case block.NumberU64() == 0:
- break done
- case block.NumberU64() < earliestBlockNo:
- break done
- case self.max <= len(logs):
- break done
+ return logs
+}
+
+func (self *Filter) getLogs(start, end uint64) (logs vm.Logs) {
+ var block *types.Block
+
+ for i := start; i <= end; i++ {
+ hash := core.GetCanonicalHash(self.db, i)
+ if hash != (common.Hash{}) {
+ block = core.GetBlock(self.db, hash)
+ } else { // block not found
+ return logs
}
// Use bloom filtering to see if this block is interesting given the
@@ -124,13 +142,9 @@ done:
}
logs = append(logs, self.FilterLogs(unfiltered)...)
}
-
- block = core.GetBlock(self.db, block.ParentHash())
}
- skip := int(math.Min(float64(len(logs)), float64(self.skip)))
-
- return logs[skip:]
+ return logs
}
func includes(addresses []common.Address, a common.Address) bool {
@@ -149,7 +163,7 @@ func (self *Filter) FilterLogs(logs vm.Logs) vm.Logs {
// Filter the logs for interesting stuff
Logs:
for _, log := range logs {
- if len(self.address) > 0 && !includes(self.address, log.Address) {
+ if len(self.addresses) > 0 && !includes(self.addresses, log.Address) {
continue
}
@@ -185,9 +199,9 @@ Logs:
}
func (self *Filter) bloomFilter(block *types.Block) bool {
- if len(self.address) > 0 {
+ if len(self.addresses) > 0 {
var included bool
- for _, addr := range self.address {
+ for _, addr := range self.addresses {
if types.BloomLookup(block.Bloom(), addr) {
included = true
break
diff --git a/eth/filters/filter_test.go b/eth/filters/filter_test.go
new file mode 100644
index 000000000..9e7538fac
--- /dev/null
+++ b/eth/filters/filter_test.go
@@ -0,0 +1,267 @@
+package filters
+
+import (
+ "io/ioutil"
+ "math/big"
+ "os"
+ "testing"
+
+ "github.com/ethereum/go-ethereum/common"
+ "github.com/ethereum/go-ethereum/core"
+ "github.com/ethereum/go-ethereum/core/types"
+ "github.com/ethereum/go-ethereum/core/vm"
+ "github.com/ethereum/go-ethereum/crypto"
+ "github.com/ethereum/go-ethereum/ethdb"
+)
+
+func makeReceipt(addr common.Address) *types.Receipt {
+ receipt := types.NewReceipt(nil, new(big.Int))
+ receipt.SetLogs(vm.Logs{
+ &vm.Log{Address: addr},
+ })
+ receipt.Bloom = types.CreateBloom(types.Receipts{receipt})
+ return receipt
+}
+
+func BenchmarkMipmaps(b *testing.B) {
+ dir, err := ioutil.TempDir("", "mipmap")
+ if err != nil {
+ b.Fatal(err)
+ }
+ defer os.RemoveAll(dir)
+
+ var (
+ db, _ = ethdb.NewLDBDatabase(dir, 16)
+ key1, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
+ addr1 = crypto.PubkeyToAddress(key1.PublicKey)
+ addr2 = common.BytesToAddress([]byte("jeff"))
+ addr3 = common.BytesToAddress([]byte("ethereum"))
+ addr4 = common.BytesToAddress([]byte("random addresses please"))
+ )
+ defer db.Close()
+
+ genesis := core.WriteGenesisBlockForTesting(db, core.GenesisAccount{addr1, big.NewInt(1000000)})
+ chain := core.GenerateChain(genesis, db, 100010, func(i int, gen *core.BlockGen) {
+ var receipts types.Receipts
+ switch i {
+ case 2403:
+ receipt := makeReceipt(addr1)
+ receipts = types.Receipts{receipt}
+ gen.AddUncheckedReceipt(receipt)
+ case 1034:
+ receipt := makeReceipt(addr2)
+ receipts = types.Receipts{receipt}
+ gen.AddUncheckedReceipt(receipt)
+ case 34:
+ receipt := makeReceipt(addr3)
+ receipts = types.Receipts{receipt}
+ gen.AddUncheckedReceipt(receipt)
+ case 99999:
+ receipt := makeReceipt(addr4)
+ receipts = types.Receipts{receipt}
+ gen.AddUncheckedReceipt(receipt)
+
+ }
+
+ // store the receipts
+ err := core.PutReceipts(db, receipts)
+ if err != nil {
+ b.Fatal(err)
+ }
+ core.WriteMipmapBloom(db, uint64(i+1), receipts)
+ })
+ for _, block := range chain {
+ core.WriteBlock(db, block)
+ if err := core.WriteCanonicalHash(db, block.Hash(), block.NumberU64()); err != nil {
+ b.Fatalf("failed to insert block number: %v", err)
+ }
+ if err := core.WriteHeadBlockHash(db, block.Hash()); err != nil {
+ b.Fatalf("failed to insert block number: %v", err)
+ }
+ if err := core.PutBlockReceipts(db, block, block.Receipts()); err != nil {
+ b.Fatal("error writing block receipts:", err)
+ }
+ }
+
+ b.ResetTimer()
+
+ filter := New(db)
+ filter.SetAddresses([]common.Address{addr1, addr2, addr3, addr4})
+ filter.SetBeginBlock(0)
+ filter.SetEndBlock(-1)
+
+ for i := 0; i < b.N; i++ {
+ logs := filter.Find()
+ if len(logs) != 4 {
+ b.Fatal("expected 4 log, got", len(logs))
+ }
+ }
+}
+
+func TestFilters(t *testing.T) {
+ dir, err := ioutil.TempDir("", "mipmap")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer os.RemoveAll(dir)
+
+ var (
+ db, _ = ethdb.NewLDBDatabase(dir, 16)
+ key1, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
+ addr = crypto.PubkeyToAddress(key1.PublicKey)
+
+ hash1 = common.BytesToHash([]byte("topic1"))
+ hash2 = common.BytesToHash([]byte("topic2"))
+ hash3 = common.BytesToHash([]byte("topic3"))
+ hash4 = common.BytesToHash([]byte("topic4"))
+ )
+ defer db.Close()
+
+ genesis := core.WriteGenesisBlockForTesting(db, core.GenesisAccount{addr, big.NewInt(1000000)})
+ chain := core.GenerateChain(genesis, db, 1000, func(i int, gen *core.BlockGen) {
+ var receipts types.Receipts
+ switch i {
+ case 1:
+ receipt := types.NewReceipt(nil, new(big.Int))
+ receipt.SetLogs(vm.Logs{
+ &vm.Log{
+ Address: addr,
+ Topics: []common.Hash{hash1},
+ },
+ })
+ gen.AddUncheckedReceipt(receipt)
+ receipts = types.Receipts{receipt}
+ case 2:
+ receipt := types.NewReceipt(nil, new(big.Int))
+ receipt.SetLogs(vm.Logs{
+ &vm.Log{
+ Address: addr,
+ Topics: []common.Hash{hash2},
+ },
+ })
+ gen.AddUncheckedReceipt(receipt)
+ receipts = types.Receipts{receipt}
+ case 998:
+ receipt := types.NewReceipt(nil, new(big.Int))
+ receipt.SetLogs(vm.Logs{
+ &vm.Log{
+ Address: addr,
+ Topics: []common.Hash{hash3},
+ },
+ })
+ gen.AddUncheckedReceipt(receipt)
+ receipts = types.Receipts{receipt}
+ case 999:
+ receipt := types.NewReceipt(nil, new(big.Int))
+ receipt.SetLogs(vm.Logs{
+ &vm.Log{
+ Address: addr,
+ Topics: []common.Hash{hash4},
+ },
+ })
+ gen.AddUncheckedReceipt(receipt)
+ receipts = types.Receipts{receipt}
+ }
+
+ // store the receipts
+ err := core.PutReceipts(db, receipts)
+ if err != nil {
+ t.Fatal(err)
+ }
+ // i is used as block number for the writes but since the i
+ // starts at 0 and block 0 (genesis) is already present increment
+ // by one
+ core.WriteMipmapBloom(db, uint64(i+1), receipts)
+ })
+ for _, block := range chain {
+ core.WriteBlock(db, block)
+ if err := core.WriteCanonicalHash(db, block.Hash(), block.NumberU64()); err != nil {
+ t.Fatalf("failed to insert block number: %v", err)
+ }
+ if err := core.WriteHeadBlockHash(db, block.Hash()); err != nil {
+ t.Fatalf("failed to insert block number: %v", err)
+ }
+ if err := core.PutBlockReceipts(db, block, block.Receipts()); err != nil {
+ t.Fatal("error writing block receipts:", err)
+ }
+ }
+
+ filter := New(db)
+ filter.SetAddresses([]common.Address{addr})
+ filter.SetTopics([][]common.Hash{[]common.Hash{hash1, hash2, hash3, hash4}})
+ filter.SetBeginBlock(0)
+ filter.SetEndBlock(-1)
+
+ logs := filter.Find()
+ if len(logs) != 4 {
+ t.Error("expected 4 log, got", len(logs))
+ }
+
+ filter = New(db)
+ filter.SetAddresses([]common.Address{addr})
+ filter.SetTopics([][]common.Hash{[]common.Hash{hash3}})
+ filter.SetBeginBlock(900)
+ filter.SetEndBlock(999)
+ logs = filter.Find()
+ if len(logs) != 1 {
+ t.Error("expected 1 log, got", len(logs))
+ }
+ if len(logs) > 0 && logs[0].Topics[0] != hash3 {
+ t.Errorf("expected log[0].Topics[0] to be %x, got %x", hash3, logs[0].Topics[0])
+ }
+
+ filter = New(db)
+ filter.SetAddresses([]common.Address{addr})
+ filter.SetTopics([][]common.Hash{[]common.Hash{hash3}})
+ filter.SetBeginBlock(990)
+ filter.SetEndBlock(-1)
+ logs = filter.Find()
+ if len(logs) != 1 {
+ t.Error("expected 1 log, got", len(logs))
+ }
+ if len(logs) > 0 && logs[0].Topics[0] != hash3 {
+ t.Errorf("expected log[0].Topics[0] to be %x, got %x", hash3, logs[0].Topics[0])
+ }
+
+ filter = New(db)
+ filter.SetTopics([][]common.Hash{[]common.Hash{hash1, hash2}})
+ filter.SetBeginBlock(1)
+ filter.SetEndBlock(10)
+
+ logs = filter.Find()
+ if len(logs) != 2 {
+ t.Error("expected 2 log, got", len(logs))
+ }
+
+ failHash := common.BytesToHash([]byte("fail"))
+ filter = New(db)
+ filter.SetTopics([][]common.Hash{[]common.Hash{failHash}})
+ filter.SetBeginBlock(0)
+ filter.SetEndBlock(-1)
+
+ logs = filter.Find()
+ if len(logs) != 0 {
+ t.Error("expected 0 log, got", len(logs))
+ }
+
+ failAddr := common.BytesToAddress([]byte("failmenow"))
+ filter = New(db)
+ filter.SetAddresses([]common.Address{failAddr})
+ filter.SetBeginBlock(0)
+ filter.SetEndBlock(-1)
+
+ logs = filter.Find()
+ if len(logs) != 0 {
+ t.Error("expected 0 log, got", len(logs))
+ }
+
+ filter = New(db)
+ filter.SetTopics([][]common.Hash{[]common.Hash{failHash}, []common.Hash{hash1}})
+ filter.SetBeginBlock(0)
+ filter.SetEndBlock(-1)
+
+ logs = filter.Find()
+ if len(logs) != 0 {
+ t.Error("expected 0 log, got", len(logs))
+ }
+}