mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-19 02:12:23 +00:00
eth, les, light: make indexer configurable
This commit is contained in:
parent
526abe2736
commit
8849bc1a85
12 changed files with 77 additions and 84 deletions
|
|
@ -130,7 +130,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
|
|||
gasPrice: config.GasPrice,
|
||||
etherbase: config.Etherbase,
|
||||
bloomRequests: make(chan chan *bloombits.Retrieval),
|
||||
bloomIndexer: NewBloomIndexer(chainDb, params.BloomBitsBlocks),
|
||||
bloomIndexer: NewBloomIndexer(chainDb, params.BloomBitsBlocks, params.BloomConfirms),
|
||||
}
|
||||
|
||||
log.Info("Initialising Ethereum protocol", "versions", ProtocolVersions, "network", config.NetworkId)
|
||||
|
|
@ -385,7 +385,7 @@ func (s *Ethereum) Protocols() []p2p.Protocol {
|
|||
// Ethereum protocol implementation.
|
||||
func (s *Ethereum) Start(srvr *p2p.Server) error {
|
||||
// Start the bloom bits servicing goroutines
|
||||
s.startBloomHandlers()
|
||||
s.startBloomHandlers(params.BloomBitsBlocks)
|
||||
|
||||
// Start the RPC service
|
||||
s.netRPCService = ethapi.NewPublicNetAPI(srvr, s.NetVersion())
|
||||
|
|
|
|||
|
|
@ -26,7 +26,6 @@ import (
|
|||
"github.com/ethereum/go-ethereum/core/rawdb"
|
||||
"github.com/ethereum/go-ethereum/core/types"
|
||||
"github.com/ethereum/go-ethereum/ethdb"
|
||||
"github.com/ethereum/go-ethereum/params"
|
||||
)
|
||||
|
||||
const (
|
||||
|
|
@ -49,7 +48,7 @@ const (
|
|||
|
||||
// startBloomHandlers starts a batch of goroutines to accept bloom bit database
|
||||
// retrievals from possibly a range of filters and serving the data to satisfy.
|
||||
func (eth *Ethereum) startBloomHandlers() {
|
||||
func (eth *Ethereum) startBloomHandlers(sectionSize uint64) {
|
||||
for i := 0; i < bloomServiceThreads; i++ {
|
||||
go func() {
|
||||
for {
|
||||
|
|
@ -61,9 +60,9 @@ func (eth *Ethereum) startBloomHandlers() {
|
|||
task := <-request
|
||||
task.Bitsets = make([][]byte, len(task.Sections))
|
||||
for i, section := range task.Sections {
|
||||
head := rawdb.ReadCanonicalHash(eth.chainDb, (section+1)*params.BloomBitsBlocks-1)
|
||||
head := rawdb.ReadCanonicalHash(eth.chainDb, (section+1)*sectionSize-1)
|
||||
if compVector, err := rawdb.ReadBloomBits(eth.chainDb, task.Bit, section, head); err == nil {
|
||||
if blob, err := bitutil.DecompressBytes(compVector, int(params.BloomBitsBlocks)/8); err == nil {
|
||||
if blob, err := bitutil.DecompressBytes(compVector, int(sectionSize/8)); err == nil {
|
||||
task.Bitsets[i] = blob
|
||||
} else {
|
||||
task.Error = err
|
||||
|
|
@ -80,9 +79,6 @@ func (eth *Ethereum) startBloomHandlers() {
|
|||
}
|
||||
|
||||
const (
|
||||
// bloomConfirms is the number of confirmation blocks before a bloom section is
|
||||
// considered probably final and its rotated bits are calculated.
|
||||
bloomConfirms = 256
|
||||
|
||||
// bloomThrottling is the time to wait between processing two consecutive index
|
||||
// sections. It's useful during chain upgrades to prevent disk overload.
|
||||
|
|
@ -103,14 +99,14 @@ type BloomIndexer struct {
|
|||
|
||||
// NewBloomIndexer returns a chain indexer that generates bloom bits data for the
|
||||
// canonical chain for fast logs filtering.
|
||||
func NewBloomIndexer(db ethdb.Database, size uint64) *core.ChainIndexer {
|
||||
func NewBloomIndexer(db ethdb.Database, size, confirms uint64) *core.ChainIndexer {
|
||||
backend := &BloomIndexer{
|
||||
db: db,
|
||||
size: size,
|
||||
}
|
||||
table := ethdb.NewTable(db, string(rawdb.BloomBitsIndexPrefix))
|
||||
|
||||
return core.NewChainIndexer(db, table, backend, size, bloomConfirms, bloomThrottling, "bloombits")
|
||||
return core.NewChainIndexer(db, table, backend, size, confirms, bloomThrottling, "bloombits")
|
||||
}
|
||||
|
||||
// Reset implements core.ChainIndexerBackend, starting a new bloombits index
|
||||
|
|
|
|||
|
|
@ -106,9 +106,10 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) {
|
|||
shutdownChan: make(chan bool),
|
||||
networkId: config.NetworkId,
|
||||
bloomRequests: make(chan chan *bloombits.Retrieval),
|
||||
bloomIndexer: eth.NewBloomIndexer(chainDb, light.BloomTrieFrequency),
|
||||
chtIndexer: light.NewChtIndexer(chainDb, true),
|
||||
bloomTrieIndexer: light.NewBloomTrieIndexer(chainDb, true),
|
||||
bloomIndexer: eth.NewBloomIndexer(chainDb, light.BloomTrieFrequency, params.BloomConfirms),
|
||||
chtIndexer: light.NewChtIndexer(chainDb, light.CHTFrequencyClient, light.HelperTrieConfirmations),
|
||||
bloomTrieIndexer: light.NewBloomTrieIndexer(chainDb, light.BloomTrieFrequency, params.BloomConfirms,
|
||||
light.BloomTrieFrequency, light.HelperTrieConfirmations),
|
||||
}
|
||||
|
||||
leth.relay = NewLesTxRelay(peers, leth.reqDist)
|
||||
|
|
@ -222,7 +223,7 @@ func (s *LightEthereum) Protocols() []p2p.Protocol {
|
|||
// Start implements node.Service, starting all internal goroutines needed by the
|
||||
// Ethereum protocol implementation.
|
||||
func (s *LightEthereum) Start(srvr *p2p.Server) error {
|
||||
s.startBloomHandlers()
|
||||
s.startBloomHandlers(light.BloomTrieFrequency)
|
||||
log.Warn("Light client mode is an experimental feature")
|
||||
s.netRPCService = ethapi.NewPublicNetAPI(srvr, s.networkId)
|
||||
// clients are searching for the first advertised protocol in the list
|
||||
|
|
|
|||
|
|
@ -43,7 +43,7 @@ const (
|
|||
|
||||
// startBloomHandlers starts a batch of goroutines to accept bloom bit database
|
||||
// retrievals from possibly a range of filters and serving the data to satisfy.
|
||||
func (eth *LightEthereum) startBloomHandlers() {
|
||||
func (eth *LightEthereum) startBloomHandlers(sectionSize uint64) {
|
||||
for i := 0; i < bloomServiceThreads; i++ {
|
||||
go func() {
|
||||
for {
|
||||
|
|
@ -57,7 +57,7 @@ func (eth *LightEthereum) startBloomHandlers() {
|
|||
compVectors, err := light.GetBloomBits(task.Context, eth.odr, task.Bit, task.Sections)
|
||||
if err == nil {
|
||||
for i := range task.Sections {
|
||||
if blob, err := bitutil.DecompressBytes(compVectors[i], int(light.BloomTrieFrequency/8)); err == nil {
|
||||
if blob, err := bitutil.DecompressBytes(compVectors[i], int(sectionSize/8)); err == nil {
|
||||
task.Bitsets[i] = blob
|
||||
} else {
|
||||
task.Error = err
|
||||
|
|
|
|||
|
|
@ -1147,8 +1147,9 @@ func (pm *ProtocolManager) getAccount(statedb *state.StateDB, root, hash common.
|
|||
func (pm *ProtocolManager) getHelperTrie(id uint, idx uint64) (common.Hash, string) {
|
||||
switch id {
|
||||
case htCanonical:
|
||||
idxV2 := (idx+1)*(light.CHTFrequencyClient/light.CHTFrequencyServer) - 1
|
||||
sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, (idx+1)*light.CHTFrequencyClient-1)
|
||||
return light.GetChtV2Root(pm.chainDb, idx, sectionHead), light.ChtTablePrefix
|
||||
return light.GetChtRoot(pm.chainDb, idxV2, sectionHead), light.ChtTablePrefix
|
||||
case htBloomBits:
|
||||
sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, (idx+1)*light.BloomTrieFrequency-1)
|
||||
return light.GetBloomTrieRoot(pm.chainDb, idx, sectionHead), light.BloomTrieTablePrefix
|
||||
|
|
|
|||
|
|
@ -414,7 +414,7 @@ func testGetCHTProofs(t *testing.T, protocol int) {
|
|||
proofsV1[0].Proof = proof
|
||||
|
||||
case 2:
|
||||
root := light.GetChtV2Root(db, 0, bc.GetHeaderByNumber(frequency-1).Hash())
|
||||
root := light.GetChtRoot(db, (light.CHTFrequencyClient/light.CHTFrequencyServer)-1, bc.GetHeaderByNumber(frequency-1).Hash())
|
||||
trie, _ := trie.New(root, trie.NewDatabase(ethdb.NewTable(db, light.ChtTablePrefix)))
|
||||
trie.Prove(key, 0, &proofsV2.Proofs)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -156,12 +156,13 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor
|
|||
} else {
|
||||
blockchain, _ := core.NewBlockChain(db, nil, gspec.Config, engine, vm.Config{})
|
||||
|
||||
chtIndexer := light.NewChtIndexer(db, false)
|
||||
chtIndexer := light.NewChtIndexer(db, light.CHTFrequencyServer, light.HelperTrieProcessConfirmations)
|
||||
chtIndexer.Start(blockchain)
|
||||
|
||||
bbtIndexer := light.NewBloomTrieIndexer(db, false)
|
||||
bbtIndexer := light.NewBloomTrieIndexer(db, params.BloomBitsBlocks, params.BloomConfirms,
|
||||
light.BloomTrieFrequency, light.HelperTrieProcessConfirmations)
|
||||
|
||||
bloomIndexer := eth.NewBloomIndexer(db, params.BloomBitsBlocks)
|
||||
bloomIndexer := eth.NewBloomIndexer(db, params.BloomBitsBlocks, params.BloomConfirms)
|
||||
bloomIndexer.AddChildIndexer(bbtIndexer)
|
||||
bloomIndexer.Start(blockchain)
|
||||
|
||||
|
|
|
|||
|
|
@ -167,7 +167,9 @@ func testOdr(t *testing.T, protocol int, expFail uint64, fn odrTestFn) {
|
|||
rm := newRetrieveManager(peers, dist, nil)
|
||||
db := ethdb.NewMemDatabase()
|
||||
ldb := ethdb.NewMemDatabase()
|
||||
odr := NewLesOdr(ldb, light.NewChtIndexer(db, true), light.NewBloomTrieIndexer(db, true), eth.NewBloomIndexer(db, light.BloomTrieFrequency), rm)
|
||||
odr := NewLesOdr(ldb, light.NewChtIndexer(db, light.CHTFrequencyClient, light.HelperTrieConfirmations),
|
||||
light.NewBloomTrieIndexer(db, light.BloomTrieFrequency, params.BloomConfirms, light.BloomTrieFrequency, light.HelperTrieConfirmations),
|
||||
eth.NewBloomIndexer(db, light.BloomTrieFrequency, params.BloomConfirms), rm)
|
||||
pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, db)
|
||||
lpm := newTestProtocolManagerMust(t, true, 0, nil, peers, odr, ldb)
|
||||
_, err1, lpeer, err2 := newTestPeerPair("peer", protocol, pm, lpm)
|
||||
|
|
|
|||
|
|
@ -27,6 +27,7 @@ import (
|
|||
"github.com/ethereum/go-ethereum/eth"
|
||||
"github.com/ethereum/go-ethereum/ethdb"
|
||||
"github.com/ethereum/go-ethereum/light"
|
||||
"github.com/ethereum/go-ethereum/params"
|
||||
)
|
||||
|
||||
var testBankSecureTrieKey = secAddr(testBankAddress)
|
||||
|
|
@ -89,7 +90,9 @@ func testAccess(t *testing.T, protocol int, fn accessTestFn) {
|
|||
rm := newRetrieveManager(peers, dist, nil)
|
||||
db := ethdb.NewMemDatabase()
|
||||
ldb := ethdb.NewMemDatabase()
|
||||
odr := NewLesOdr(ldb, light.NewChtIndexer(db, true), light.NewBloomTrieIndexer(db, true), eth.NewBloomIndexer(db, light.BloomTrieFrequency), rm)
|
||||
odr := NewLesOdr(ldb, light.NewChtIndexer(db, light.CHTFrequencyClient, light.HelperTrieConfirmations),
|
||||
light.NewBloomTrieIndexer(db, light.BloomTrieFrequency, params.BloomConfirms, light.BloomTrieFrequency, light.HelperTrieConfirmations),
|
||||
eth.NewBloomIndexer(db, light.BloomTrieFrequency, params.BloomConfirms), rm)
|
||||
|
||||
pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, db)
|
||||
lpm := newTestProtocolManagerMust(t, true, 0, nil, peers, odr, ldb)
|
||||
|
|
|
|||
|
|
@ -34,6 +34,7 @@ import (
|
|||
"github.com/ethereum/go-ethereum/log"
|
||||
"github.com/ethereum/go-ethereum/p2p"
|
||||
"github.com/ethereum/go-ethereum/p2p/discv5"
|
||||
"github.com/ethereum/go-ethereum/params"
|
||||
"github.com/ethereum/go-ethereum/rlp"
|
||||
)
|
||||
|
||||
|
|
@ -67,8 +68,9 @@ func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) {
|
|||
protocolManager: pm,
|
||||
quitSync: quitSync,
|
||||
lesTopics: lesTopics,
|
||||
chtIndexer: light.NewChtIndexer(eth.ChainDb(), false),
|
||||
bloomTrieIndexer: light.NewBloomTrieIndexer(eth.ChainDb(), false),
|
||||
chtIndexer: light.NewChtIndexer(eth.ChainDb(), light.CHTFrequencyServer, light.HelperTrieProcessConfirmations),
|
||||
bloomTrieIndexer: light.NewBloomTrieIndexer(eth.ChainDb(), params.BloomBitsBlocks, params.BloomConfirms,
|
||||
light.BloomTrieFrequency, light.HelperTrieProcessConfirmations),
|
||||
}
|
||||
logger := log.New()
|
||||
|
||||
|
|
@ -80,7 +82,7 @@ func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) {
|
|||
// convert last LES/2 section index back to LES/1 index for chtIndexer.SectionHead
|
||||
chtLastSectionV1 := (chtLastSection+1)*(light.CHTFrequencyClient/light.CHTFrequencyServer) - 1
|
||||
chtSectionHead := srv.chtIndexer.SectionHead(chtLastSectionV1)
|
||||
chtRoot := light.GetChtV2Root(pm.chainDb, chtLastSection, chtSectionHead)
|
||||
chtRoot := light.GetChtRoot(pm.chainDb, chtLastSectionV1, chtSectionHead)
|
||||
logger.Info("Loaded CHT", "section", chtLastSection, "head", chtSectionHead, "root", chtRoot)
|
||||
}
|
||||
bloomTrieSectionCount, _, _ := srv.bloomTrieIndexer.Sections()
|
||||
|
|
|
|||
|
|
@ -94,7 +94,7 @@ type ChtNode struct {
|
|||
Td *big.Int
|
||||
}
|
||||
|
||||
// GetChtRoot reads the CHT root assoctiated to the given section from the database
|
||||
// GetChtRoot reads the CHT root associated to the given section from the database
|
||||
// Note that sectionIdx is specified according to LES/1 CHT section size
|
||||
func GetChtRoot(db ethdb.Database, sectionIdx uint64, sectionHead common.Hash) common.Hash {
|
||||
var encNumber [8]byte
|
||||
|
|
@ -103,13 +103,7 @@ func GetChtRoot(db ethdb.Database, sectionIdx uint64, sectionHead common.Hash) c
|
|||
return common.BytesToHash(data)
|
||||
}
|
||||
|
||||
// GetChtV2Root reads the CHT root assoctiated to the given section from the database
|
||||
// Note that sectionIdx is specified according to LES/2 CHT section size
|
||||
func GetChtV2Root(db ethdb.Database, sectionIdx uint64, sectionHead common.Hash) common.Hash {
|
||||
return GetChtRoot(db, (sectionIdx+1)*(CHTFrequencyClient/CHTFrequencyServer)-1, sectionHead)
|
||||
}
|
||||
|
||||
// StoreChtRoot writes the CHT root assoctiated to the given section into the database
|
||||
// StoreChtRoot writes the CHT root associated to the given section into the database
|
||||
// Note that sectionIdx is specified according to LES/1 CHT section size
|
||||
func StoreChtRoot(db ethdb.Database, sectionIdx uint64, sectionHead, root common.Hash) {
|
||||
var encNumber [8]byte
|
||||
|
|
@ -126,23 +120,15 @@ type ChtIndexerBackend struct {
|
|||
trie *trie.Trie
|
||||
}
|
||||
|
||||
// NewBloomTrieIndexer creates a BloomTrie chain indexer
|
||||
func NewChtIndexer(db ethdb.Database, clientMode bool) *core.ChainIndexer {
|
||||
var sectionSize, confirmReq uint64
|
||||
if clientMode {
|
||||
sectionSize = CHTFrequencyClient
|
||||
confirmReq = HelperTrieConfirmations
|
||||
} else {
|
||||
sectionSize = CHTFrequencyServer
|
||||
confirmReq = HelperTrieProcessConfirmations
|
||||
}
|
||||
// NewChtIndexer creates a Cht chain indexer
|
||||
func NewChtIndexer(db ethdb.Database, size, confirms uint64) *core.ChainIndexer {
|
||||
idb := ethdb.NewTable(db, "chtIndex-")
|
||||
backend := &ChtIndexerBackend{
|
||||
diskdb: db,
|
||||
triedb: trie.NewDatabase(ethdb.NewTable(db, ChtTablePrefix)),
|
||||
sectionSize: sectionSize,
|
||||
sectionSize: size,
|
||||
}
|
||||
return core.NewChainIndexer(db, idb, backend, sectionSize, confirmReq, time.Millisecond*100, "cht")
|
||||
return core.NewChainIndexer(db, idb, backend, size, confirms, time.Millisecond*100, "cht")
|
||||
}
|
||||
|
||||
// Reset implements core.ChainIndexerBackend
|
||||
|
|
@ -217,30 +203,27 @@ func StoreBloomTrieRoot(db ethdb.Database, sectionIdx uint64, sectionHead, root
|
|||
type BloomTrieIndexerBackend struct {
|
||||
diskdb ethdb.Database
|
||||
triedb *trie.Database
|
||||
section, parentSectionSize, bloomTrieRatio uint64
|
||||
section uint64
|
||||
parentSize uint64
|
||||
size uint64
|
||||
bloomTrieRatio uint64
|
||||
trie *trie.Trie
|
||||
sectionHeads []common.Hash
|
||||
}
|
||||
|
||||
// NewBloomTrieIndexer creates a BloomTrie chain indexer
|
||||
func NewBloomTrieIndexer(db ethdb.Database, clientMode bool) *core.ChainIndexer {
|
||||
func NewBloomTrieIndexer(db ethdb.Database, parentSize, parentConfirms, size, confirms uint64) *core.ChainIndexer {
|
||||
backend := &BloomTrieIndexerBackend{
|
||||
diskdb: db,
|
||||
triedb: trie.NewDatabase(ethdb.NewTable(db, BloomTrieTablePrefix)),
|
||||
parentSize: parentSize,
|
||||
size: size,
|
||||
}
|
||||
idb := ethdb.NewTable(db, "bltIndex-")
|
||||
|
||||
var confirmReq uint64
|
||||
if clientMode {
|
||||
backend.parentSectionSize = BloomTrieFrequency
|
||||
confirmReq = HelperTrieConfirmations
|
||||
} else {
|
||||
backend.parentSectionSize = ethBloomBitsSection
|
||||
confirmReq = HelperTrieProcessConfirmations
|
||||
}
|
||||
backend.bloomTrieRatio = BloomTrieFrequency / backend.parentSectionSize
|
||||
backend.bloomTrieRatio = size / backend.parentSize
|
||||
backend.sectionHeads = make([]common.Hash, backend.bloomTrieRatio)
|
||||
return core.NewChainIndexer(db, idb, backend, BloomTrieFrequency, confirmReq-ethBloomBitsConfirmations, time.Millisecond*100, "bloomtrie")
|
||||
return core.NewChainIndexer(db, idb, backend, size, confirms-parentConfirms, time.Millisecond*100, "bloomtrie")
|
||||
}
|
||||
|
||||
// Reset implements core.ChainIndexerBackend
|
||||
|
|
@ -257,9 +240,9 @@ func (b *BloomTrieIndexerBackend) Reset(section uint64, lastSectionHead common.H
|
|||
|
||||
// Process implements core.ChainIndexerBackend
|
||||
func (b *BloomTrieIndexerBackend) Process(header *types.Header) {
|
||||
num := header.Number.Uint64() - b.section*BloomTrieFrequency
|
||||
if (num+1)%b.parentSectionSize == 0 {
|
||||
b.sectionHeads[num/b.parentSectionSize] = header.Hash()
|
||||
num := header.Number.Uint64() - b.section*b.size
|
||||
if (num+1)%b.parentSize == 0 {
|
||||
b.sectionHeads[num/b.parentSize] = header.Hash()
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -277,7 +260,7 @@ func (b *BloomTrieIndexerBackend) Commit() error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
decompData, err2 := bitutil.DecompressBytes(data, int(b.parentSectionSize/8))
|
||||
decompData, err2 := bitutil.DecompressBytes(data, int(b.parentSize/8))
|
||||
if err2 != nil {
|
||||
return err2
|
||||
}
|
||||
|
|
|
|||
|
|
@ -23,4 +23,8 @@ const (
|
|||
// BloomBitsBlocks is the number of blocks a single bloom bit section vector
|
||||
// contains.
|
||||
BloomBitsBlocks uint64 = 4096
|
||||
|
||||
// BloomConfirms is the number of confirmation blocks before a bloom section is
|
||||
// considered probably final and its rotated bits are calculated.
|
||||
BloomConfirms = 256
|
||||
)
|
||||
|
|
|
|||
Loading…
Reference in a new issue