core, eth: add/use separate header/body accessors

This commit is contained in:
Péter Szilágyi 2015-09-01 14:29:54 +03:00
parent fce218f7f8
commit a06c4c964d
7 changed files with 257 additions and 146 deletions

View file

@ -48,6 +48,8 @@ var (
) )
const ( const (
headerCacheLimit = 256
bodyCacheLimit = 256
blockCacheLimit = 256 blockCacheLimit = 256
maxFutureBlocks = 256 maxFutureBlocks = 256
maxTimeFutureBlocks = 30 maxTimeFutureBlocks = 30
@ -71,7 +73,10 @@ type ChainManager struct {
lastBlockHash common.Hash lastBlockHash common.Hash
currentGasLimit *big.Int currentGasLimit *big.Int
cache *lru.Cache // cache is the LRU caching headerCache *lru.Cache // Cache for the most recent block headers
bodyCache *lru.Cache // Cache for the most recent block bodies
bodyRLPCache *lru.Cache // Cache for the most recent block bodies in RLP encoded format
blockCache *lru.Cache // Cache for the most recent entire blocks
futureBlocks *lru.Cache // future blocks are blocks added for later processing futureBlocks *lru.Cache // future blocks are blocks added for later processing
quit chan struct{} quit chan struct{}
@ -84,12 +89,21 @@ type ChainManager struct {
} }
func NewChainManager(chainDb common.Database, pow pow.PoW, mux *event.TypeMux) (*ChainManager, error) { func NewChainManager(chainDb common.Database, pow pow.PoW, mux *event.TypeMux) (*ChainManager, error) {
cache, _ := lru.New(blockCacheLimit) headerCache, _ := lru.New(headerCacheLimit)
bodyCache, _ := lru.New(bodyCacheLimit)
bodyRLPCache, _ := lru.New(bodyCacheLimit)
blockCache, _ := lru.New(blockCacheLimit)
futureBlocks, _ := lru.New(maxFutureBlocks)
bc := &ChainManager{ bc := &ChainManager{
chainDb: chainDb, chainDb: chainDb,
eventMux: mux, eventMux: mux,
quit: make(chan struct{}), quit: make(chan struct{}),
cache: cache, headerCache: headerCache,
bodyCache: bodyCache,
bodyRLPCache: bodyRLPCache,
blockCache: blockCache,
futureBlocks: futureBlocks,
pow: pow, pow: pow,
} }
@ -105,11 +119,9 @@ func NewChainManager(chainDb common.Database, pow pow.PoW, mux *event.TypeMux) (
} }
glog.V(logger.Info).Infoln("WARNING: Wrote default ethereum genesis block") glog.V(logger.Info).Infoln("WARNING: Wrote default ethereum genesis block")
} }
if err := bc.setLastState(); err != nil { if err := bc.setLastState(); err != nil {
return nil, err return nil, err
} }
// Check the current state of the block hashes and make sure that we do not have any of the bad blocks in our chain // Check the current state of the block hashes and make sure that we do not have any of the bad blocks in our chain
for hash, _ := range BadHashes { for hash, _ := range BadHashes {
if block := bc.GetBlock(hash); block != nil { if block := bc.GetBlock(hash); block != nil {
@ -123,14 +135,8 @@ func NewChainManager(chainDb common.Database, pow pow.PoW, mux *event.TypeMux) (
glog.V(logger.Error).Infoln("Chain reorg was successfull. Resuming normal operation") glog.V(logger.Error).Infoln("Chain reorg was successfull. Resuming normal operation")
} }
} }
// Take ownership of this particular state // Take ownership of this particular state
bc.futureBlocks, _ = lru.New(maxFutureBlocks)
bc.makeCache()
go bc.update() go bc.update()
return bc, nil return bc, nil
} }
@ -139,13 +145,15 @@ func (bc *ChainManager) SetHead(head *types.Block) {
defer bc.mu.Unlock() defer bc.mu.Unlock()
for block := bc.currentBlock; block != nil && block.Hash() != head.Hash(); block = bc.GetBlock(block.ParentHash()) { for block := bc.currentBlock; block != nil && block.Hash() != head.Hash(); block = bc.GetBlock(block.ParentHash()) {
bc.removeBlock(block) DeleteBlock(bc.chainDb, block.Hash())
} }
bc.headerCache.Purge()
bc.bodyCache.Purge()
bc.bodyRLPCache.Purge()
bc.blockCache.Purge()
bc.futureBlocks.Purge()
bc.cache, _ = lru.New(blockCacheLimit)
bc.currentBlock = head bc.currentBlock = head
bc.makeCache()
bc.setTotalDifficulty(head.Td) bc.setTotalDifficulty(head.Td)
bc.insert(head) bc.insert(head)
bc.setLastState() bc.setLastState()
@ -199,11 +207,9 @@ func (bc *ChainManager) recover() bool {
if len(data) != 0 { if len(data) != 0 {
block := bc.GetBlock(common.BytesToHash(data)) block := bc.GetBlock(common.BytesToHash(data))
if block != nil { if block != nil {
err := bc.chainDb.Put([]byte("LastBlock"), block.Hash().Bytes()) if err := WriteHead(bc.chainDb, block); err != nil {
if err != nil { glog.Fatalf("failed to write database head: %v", err)
glog.Fatalln("db write err:", err)
} }
bc.currentBlock = block bc.currentBlock = block
bc.lastBlockHash = block.Hash() bc.lastBlockHash = block.Hash()
return true return true
@ -213,14 +219,14 @@ func (bc *ChainManager) recover() bool {
} }
func (bc *ChainManager) setLastState() error { func (bc *ChainManager) setLastState() error {
data, _ := bc.chainDb.Get([]byte("LastBlock")) head := GetHeadHash(bc.chainDb)
if len(data) != 0 { if head != (common.Hash{}) {
block := bc.GetBlock(common.BytesToHash(data)) block := bc.GetBlock(head)
if block != nil { if block != nil {
bc.currentBlock = block bc.currentBlock = block
bc.lastBlockHash = block.Hash() bc.lastBlockHash = block.Hash()
} else { } else {
glog.Infof("LastBlock (%x) not found. Recovering...\n", data) glog.Infof("LastBlock (%x) not found. Recovering...\n", head)
if bc.recover() { if bc.recover() {
glog.Infof("Recover successful") glog.Infof("Recover successful")
} else { } else {
@ -240,63 +246,37 @@ func (bc *ChainManager) setLastState() error {
return nil return nil
} }
func (bc *ChainManager) makeCache() { // Reset purges the entire blockchain, restoring it to its genesis state.
bc.cache, _ = lru.New(blockCacheLimit)
// load in last `blockCacheLimit` - 1 blocks. Last block is the current.
bc.cache.Add(bc.genesisBlock.Hash(), bc.genesisBlock)
for _, block := range bc.GetBlocksFromHash(bc.currentBlock.Hash(), blockCacheLimit) {
bc.cache.Add(block.Hash(), block)
}
}
func (bc *ChainManager) Reset() { func (bc *ChainManager) Reset() {
bc.ResetWithGenesisBlock(bc.genesisBlock)
}
// ResetWithGenesisBlock purges the entire blockchain, restoring it to the
// specified genesis state.
func (bc *ChainManager) ResetWithGenesisBlock(genesis *types.Block) {
bc.mu.Lock() bc.mu.Lock()
defer bc.mu.Unlock() defer bc.mu.Unlock()
// Dump the entire block chain and purge the caches
for block := bc.currentBlock; block != nil; block = bc.GetBlock(block.ParentHash()) { for block := bc.currentBlock; block != nil; block = bc.GetBlock(block.ParentHash()) {
bc.removeBlock(block) DeleteBlock(bc.chainDb, block.Hash())
} }
bc.headerCache.Purge()
bc.bodyCache.Purge()
bc.bodyRLPCache.Purge()
bc.blockCache.Purge()
bc.futureBlocks.Purge()
bc.cache, _ = lru.New(blockCacheLimit) // Prepare the genesis block and reinitialize the chain
bc.genesisBlock = genesis
bc.genesisBlock.Td = genesis.Difficulty()
// Prepare the genesis block if err := WriteBlock(bc.chainDb, bc.genesisBlock); err != nil {
err := WriteBlock(bc.chainDb, bc.genesisBlock) glog.Fatalf("failed to write genesis block: %v", err)
if err != nil {
glog.Fatalln("db err:", err)
} }
bc.insert(bc.genesisBlock) bc.insert(bc.genesisBlock)
bc.currentBlock = bc.genesisBlock bc.currentBlock = bc.genesisBlock
bc.makeCache() bc.setTotalDifficulty(genesis.Difficulty())
bc.setTotalDifficulty(common.Big("0"))
}
func (bc *ChainManager) removeBlock(block *types.Block) {
bc.chainDb.Delete(append(blockHashPre, block.Hash().Bytes()...))
}
func (bc *ChainManager) ResetWithGenesisBlock(gb *types.Block) {
bc.mu.Lock()
defer bc.mu.Unlock()
for block := bc.currentBlock; block != nil; block = bc.GetBlock(block.ParentHash()) {
bc.removeBlock(block)
}
// Prepare the genesis block
gb.Td = gb.Difficulty()
bc.genesisBlock = gb
err := WriteBlock(bc.chainDb, bc.genesisBlock)
if err != nil {
glog.Fatalln("db err:", err)
}
bc.insert(bc.genesisBlock)
bc.currentBlock = bc.genesisBlock
bc.makeCache()
bc.td = gb.Difficulty()
} }
// Export writes the active chain to the given writer. // Export writes the active chain to the given writer.
@ -359,56 +339,130 @@ func (bc *ChainManager) Genesis() *types.Block {
return bc.genesisBlock return bc.genesisBlock
} }
// Block fetching methods // HasHeader checks if a block header is present in the database or not, caching
// it if present.
func (bc *ChainManager) HasHeader(hash common.Hash) bool {
return bc.GetHeader(hash) != nil
}
// GetHeader retrieves a block header from the database by hash, caching it if
// found.
func (self *ChainManager) GetHeader(hash common.Hash) *types.Header {
// Short circuit if the header's already in the cache, retrieve otherwise
if header, ok := self.headerCache.Get(hash); ok {
return header.(*types.Header)
}
header := GetHeaderByHash(self.chainDb, hash)
if header == nil {
return nil
}
// Cache the found header for next time and return
self.headerCache.Add(header.Hash(), header)
return header
}
// GetHeaderByNumber retrieves a block header from the database by number,
// caching it (associated with its hash) if found.
func (self *ChainManager) GetHeaderByNumber(number uint64) *types.Header {
hash := GetHashByNumber(self.chainDb, number)
if hash == (common.Hash{}) {
return nil
}
return self.GetHeader(hash)
}
// GetBody retrieves a block body (transactions, uncles and total difficulty)
// from the database by hash, caching it if found. The resion for the peculiar
// pointer-to-slice return type is to differentiate between empty and inexistent
// bodies.
func (self *ChainManager) GetBody(hash common.Hash) (*[]*types.Transaction, *[]*types.Header) {
// Short circuit if the body's already in the cache, retrieve otherwise
if cached, ok := self.bodyCache.Get(hash); ok {
body := cached.(*storageBody)
return &body.Transactions, &body.Uncles
}
transactions, uncles, td := GetBodyByHash(self.chainDb, hash)
if td == nil {
return nil, nil
}
// Cache the found body for next time and return
self.bodyCache.Add(hash, &storageBody{
Transactions: transactions,
Uncles: uncles,
})
return &transactions, &uncles
}
// GetBodyRLP retrieves a block body in RLP encoding from the database by hash,
// caching it if found.
func (self *ChainManager) GetBodyRLP(hash common.Hash) []byte {
// Short circuit if the body's already in the cache, retrieve otherwise
if cached, ok := self.bodyRLPCache.Get(hash); ok {
return cached.([]byte)
}
body, td := GetBodyRLPByHash(self.chainDb, hash)
if td == nil {
return nil
}
// Cache the found body for next time and return
self.bodyRLPCache.Add(hash, body)
return body
}
// HasBlock checks if a block is fully present in the database or not, caching
// it if present.
func (bc *ChainManager) HasBlock(hash common.Hash) bool { func (bc *ChainManager) HasBlock(hash common.Hash) bool {
return bc.GetBlock(hash) != nil return bc.GetBlock(hash) != nil
} }
func (self *ChainManager) GetBlockHashesFromHash(hash common.Hash, max uint64) (chain []common.Hash) { // GetBlock retrieves a block from the database by hash, caching it if found.
block := self.GetBlock(hash)
if block == nil {
return
}
// XXX Could be optimised by using a different database which only holds hashes (i.e., linked list)
for i := uint64(0); i < max; i++ {
block = self.GetBlock(block.ParentHash())
if block == nil {
break
}
chain = append(chain, block.Hash())
if block.Number().Cmp(common.Big0) <= 0 {
break
}
}
return
}
func (self *ChainManager) GetBlock(hash common.Hash) *types.Block { func (self *ChainManager) GetBlock(hash common.Hash) *types.Block {
if block, ok := self.cache.Get(hash); ok { // Short circuit if the block's already in the cache, retrieve otherwise
if block, ok := self.blockCache.Get(hash); ok {
return block.(*types.Block) return block.(*types.Block)
} }
block := GetBlockByHash(self.chainDb, hash) block := GetBlockByHash(self.chainDb, hash)
if block == nil { if block == nil {
return nil return nil
} }
// Cache the found block for next time and return
// Add the block to the cache self.blockCache.Add(block.Hash(), block)
self.cache.Add(hash, (*types.Block)(block)) return block
return (*types.Block)(block)
} }
func (self *ChainManager) GetBlockByNumber(num uint64) *types.Block { // GetBlockByNumber retrieves a block from the database by number, caching it
self.mu.RLock() // (associated with its hash) if found.
defer self.mu.RUnlock() func (self *ChainManager) GetBlockByNumber(number uint64) *types.Block {
hash := GetHashByNumber(self.chainDb, number)
return self.getBlockByNumber(num) if hash == (common.Hash{}) {
return nil
}
return self.GetBlock(hash)
} }
// GetBlockHashesFromHash retrieves a number of block hashes starting at a given
// hash, fetching towards the genesis block.
func (self *ChainManager) GetBlockHashesFromHash(hash common.Hash, max uint64) []common.Hash {
// Get the origin header from which to fetch
header := self.GetHeader(hash)
if header == nil {
return nil
}
// Iterate the headers until enough is collected or the genesis reached
chain := make([]common.Hash, 0, max)
for i := uint64(0); i < max; i++ {
if header = self.GetHeader(header.ParentHash); header == nil {
break
}
chain = append(chain, header.Hash())
if header.Number.Cmp(common.Big0) <= 0 {
break
}
}
return chain
}
// [deprecated by eth/62]
// GetBlocksFromHash returns the block corresponding to hash and up to n-1 ancestors. // GetBlocksFromHash returns the block corresponding to hash and up to n-1 ancestors.
func (self *ChainManager) GetBlocksFromHash(hash common.Hash, n int) (blocks []*types.Block) { func (self *ChainManager) GetBlocksFromHash(hash common.Hash, n int) (blocks []*types.Block) {
for i := 0; i < n; i++ { for i := 0; i < n; i++ {
@ -422,11 +476,6 @@ func (self *ChainManager) GetBlocksFromHash(hash common.Hash, n int) (blocks []*
return return
} }
// non blocking version
func (self *ChainManager) getBlockByNumber(num uint64) *types.Block {
return GetBlockByNumber(self.chainDb, num)
}
func (self *ChainManager) GetUnclesInChain(block *types.Block, length int) (uncles []*types.Header) { func (self *ChainManager) GetUnclesInChain(block *types.Block, length int) (uncles []*types.Header) {
for i := 0; block != nil && i < length; i++ { for i := 0; block != nil && i < length; i++ {
uncles = append(uncles, block.Uncles()...) uncles = append(uncles, block.Uncles()...)

View file

@ -388,7 +388,10 @@ func makeChainWithDiff(genesis *types.Block, d []int, seed byte) []*types.Block
func chm(genesis *types.Block, db common.Database) *ChainManager { func chm(genesis *types.Block, db common.Database) *ChainManager {
var eventMux event.TypeMux var eventMux event.TypeMux
bc := &ChainManager{chainDb: db, genesisBlock: genesis, eventMux: &eventMux, pow: FakePow{}} bc := &ChainManager{chainDb: db, genesisBlock: genesis, eventMux: &eventMux, pow: FakePow{}}
bc.cache, _ = lru.New(100) bc.headerCache, _ = lru.New(100)
bc.bodyCache, _ = lru.New(100)
bc.bodyRLPCache, _ = lru.New(100)
bc.blockCache, _ = lru.New(100)
bc.futureBlocks, _ = lru.New(100) bc.futureBlocks, _ = lru.New(100)
bc.processor = bproc{} bc.processor = bproc{}
bc.ResetWithGenesisBlock(genesis) bc.ResetWithGenesisBlock(genesis)

View file

@ -29,6 +29,8 @@ import (
) )
var ( var (
headKey = []byte("LastBlock")
headerHashPre = []byte("header-hash-") headerHashPre = []byte("header-hash-")
bodyHashPre = []byte("body-hash-") bodyHashPre = []byte("body-hash-")
blockNumPre = []byte("block-num-") blockNumPre = []byte("block-num-")
@ -120,6 +122,24 @@ type storageBody struct {
Uncles []*types.Header Uncles []*types.Header
} }
// GetHashByNumber retrieves a hash assigned to a canonical block number.
func GetHashByNumber(db common.Database, number uint64) common.Hash {
data, _ := db.Get(append(blockNumPre, big.NewInt(int64(number)).Bytes()...))
if len(data) == 0 {
return common.Hash{}
}
return common.BytesToHash(data)
}
// GetHeadHash retrieves the hash of the current canonical head block.
func GetHeadHash(db common.Database) common.Hash {
data, _ := db.Get(headKey)
if len(data) == 0 {
return common.Hash{}
}
return common.BytesToHash(data)
}
// GetHeaderRLPByHash retrieves a block header in its raw RLP database encoding, // GetHeaderRLPByHash retrieves a block header in its raw RLP database encoding,
// or nil if the header's not found. // or nil if the header's not found.
func GetHeaderRLPByHash(db common.Database, hash common.Hash) []byte { func GetHeaderRLPByHash(db common.Database, hash common.Hash) []byte {
@ -202,24 +222,24 @@ func GetBlockByNumber(db common.Database, number uint64) *types.Block {
return GetBlockByHash(db, common.BytesToHash(key)) return GetBlockByHash(db, common.BytesToHash(key))
} }
// WriteCanonNumber writes the canonical hash for the given block // WriteCanonNumber stores the canonical hash for the given block number.
func WriteCanonNumber(db common.Database, block *types.Block) error { func WriteCanonNumber(db common.Database, hash common.Hash, number uint64) error {
key := append(blockNumPre, block.Number().Bytes()...) key := append(blockNumPre, big.NewInt(int64(number)).Bytes()...)
err := db.Put(key, block.Hash().Bytes()) if err := db.Put(key, hash.Bytes()); err != nil {
if err != nil { glog.Fatalf("failed to store number to hash mapping into database: %v", err)
return err return err
} }
return nil return nil
} }
// WriteHead force writes the current head // WriteHead updates the head block of the chain database.
func WriteHead(db common.Database, block *types.Block) error { func WriteHead(db common.Database, block *types.Block) error {
err := WriteCanonNumber(db, block) if err := WriteCanonNumber(db, block.Hash(), block.NumberU64()); err != nil {
if err != nil { glog.Fatalf("failed to store canonical number into database: %v", err)
return err return err
} }
err = db.Put([]byte("LastBlock"), block.Hash().Bytes()) if err := db.Put(headKey, block.Hash().Bytes()); err != nil {
if err != nil { glog.Fatalf("failed to store last block into database: %v", err)
return err return err
} }
return nil return nil
@ -272,6 +292,22 @@ func WriteBlock(db common.Database, block *types.Block) error {
return nil return nil
} }
// DeleteHeader removes all block header data associated with a hash.
func DeleteHeader(db common.Database, hash common.Hash) {
db.Delete(append(headerHashPre, hash.Bytes()...))
}
// DeleteBody removes all block body data associated with a hash.
func DeleteBody(db common.Database, hash common.Hash) {
db.Delete(append(bodyHashPre, hash.Bytes()...))
}
// DeleteBlock removes all block data associated with a hash.
func DeleteBlock(db common.Database, hash common.Hash) {
DeleteHeader(db, hash)
DeleteBody(db, hash)
}
// [deprecated by eth/63] // [deprecated by eth/63]
// GetBlockByHashOld returns the old combined block corresponding to the hash // GetBlockByHashOld returns the old combined block corresponding to the hash
// or nil if not found. This method is only used by the upgrade mechanism to // or nil if not found. This method is only used by the upgrade mechanism to

View file

@ -86,7 +86,7 @@ func WriteGenesisBlock(chainDb common.Database, reader io.Reader) (*types.Block,
if block := GetBlockByHash(chainDb, block.Hash()); block != nil { if block := GetBlockByHash(chainDb, block.Hash()); block != nil {
glog.V(logger.Info).Infoln("Genesis block already in chain. Writing canonical number") glog.V(logger.Info).Infoln("Genesis block already in chain. Writing canonical number")
err := WriteCanonNumber(chainDb, block) err := WriteCanonNumber(chainDb, block.Hash(), block.NumberU64())
if err != nil { if err != nil {
return nil, err return nil, err
} }

View file

@ -345,33 +345,33 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if err := msg.Decode(&query); err != nil { if err := msg.Decode(&query); err != nil {
return errResp(ErrDecode, "%v: %v", msg, err) return errResp(ErrDecode, "%v: %v", msg, err)
} }
// Gather blocks until the fetch or network limits is reached // Gather headers until the fetch or network limits is reached
var ( var (
bytes common.StorageSize bytes common.StorageSize
headers []*types.Header headers []*types.Header
unknown bool unknown bool
) )
for !unknown && len(headers) < int(query.Amount) && bytes < softResponseLimit && len(headers) < downloader.MaxHeaderFetch { for !unknown && len(headers) < int(query.Amount) && bytes < softResponseLimit && len(headers) < downloader.MaxHeaderFetch {
// Retrieve the next block satisfying the query // Retrieve the next header satisfying the query
var origin *types.Block var origin *types.Header
if query.Origin.Hash != (common.Hash{}) { if query.Origin.Hash != (common.Hash{}) {
origin = pm.chainman.GetBlock(query.Origin.Hash) origin = pm.chainman.GetHeader(query.Origin.Hash)
} else { } else {
origin = pm.chainman.GetBlockByNumber(query.Origin.Number) origin = pm.chainman.GetHeaderByNumber(query.Origin.Number)
} }
if origin == nil { if origin == nil {
break break
} }
headers = append(headers, origin.Header()) headers = append(headers, origin)
bytes += origin.Size() bytes += 500 // Approximate, should be good enough estimate
// Advance to the next block of the query // Advance to the next header of the query
switch { switch {
case query.Origin.Hash != (common.Hash{}) && query.Reverse: case query.Origin.Hash != (common.Hash{}) && query.Reverse:
// Hash based traversal towards the genesis block // Hash based traversal towards the genesis block
for i := 0; i < int(query.Skip)+1; i++ { for i := 0; i < int(query.Skip)+1; i++ {
if block := pm.chainman.GetBlock(query.Origin.Hash); block != nil { if header := pm.chainman.GetHeader(query.Origin.Hash); header != nil {
query.Origin.Hash = block.ParentHash() query.Origin.Hash = header.ParentHash
} else { } else {
unknown = true unknown = true
break break
@ -379,9 +379,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
} }
case query.Origin.Hash != (common.Hash{}) && !query.Reverse: case query.Origin.Hash != (common.Hash{}) && !query.Reverse:
// Hash based traversal towards the leaf block // Hash based traversal towards the leaf block
if block := pm.chainman.GetBlockByNumber(origin.NumberU64() + query.Skip + 1); block != nil { if header := pm.chainman.GetHeaderByNumber(origin.Number.Uint64() + query.Skip + 1); header != nil {
if pm.chainman.GetBlockHashesFromHash(block.Hash(), query.Skip+1)[query.Skip] == query.Origin.Hash { if pm.chainman.GetBlockHashesFromHash(header.Hash(), query.Skip+1)[query.Skip] == query.Origin.Hash {
query.Origin.Hash = block.Hash() query.Origin.Hash = header.Hash()
} else { } else {
unknown = true unknown = true
} }
@ -452,23 +452,24 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
// Gather blocks until the fetch or network limits is reached // Gather blocks until the fetch or network limits is reached
var ( var (
hash common.Hash hash common.Hash
bytes common.StorageSize bytes int
bodies []*blockBody bodies []*blockBodyRLP
) )
for bytes < softResponseLimit && len(bodies) < downloader.MaxBlockFetch { for bytes < softResponseLimit && len(bodies) < downloader.MaxBlockFetch {
//Retrieve the hash of the next block // Retrieve the hash of the next block
if err := msgStream.Decode(&hash); err == rlp.EOL { if err := msgStream.Decode(&hash); err == rlp.EOL {
break break
} else if err != nil { } else if err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err) return errResp(ErrDecode, "msg %v: %v", msg, err)
} }
// Retrieve the requested block, stopping if enough was found // Retrieve the requested block body, stopping if enough was found
if block := pm.chainman.GetBlock(hash); block != nil { if data := pm.chainman.GetBodyRLP(hash); len(data) != 0 {
bodies = append(bodies, &blockBody{Transactions: block.Transactions(), Uncles: block.Uncles()}) body := blockBodyRLP(data)
bytes += block.Size() bodies = append(bodies, &body)
bytes += len(body)
} }
} }
return p.SendBlockBodies(bodies) return p.SendBlockBodiesRLP(bodies)
case p.version >= eth63 && msg.Code == GetNodeDataMsg: case p.version >= eth63 && msg.Code == GetNodeDataMsg:
// Decode the retrieval message // Decode the retrieval message

View file

@ -184,6 +184,12 @@ func (p *peer) SendBlockBodies(bodies []*blockBody) error {
return p2p.Send(p.rw, BlockBodiesMsg, blockBodiesData(bodies)) return p2p.Send(p.rw, BlockBodiesMsg, blockBodiesData(bodies))
} }
// SendBlockBodiesRLP sends a batch of block contents to the remote peer from
// an already RLP encoded format.
func (p *peer) SendBlockBodiesRLP(bodies []*blockBodyRLP) error {
return p2p.Send(p.rw, BlockBodiesMsg, blockBodiesRLPData(bodies))
}
// SendNodeData sends a batch of arbitrary internal data, corresponding to the // SendNodeData sends a batch of arbitrary internal data, corresponding to the
// hashes requested. // hashes requested.
func (p *peer) SendNodeData(data [][]byte) error { func (p *peer) SendNodeData(data [][]byte) error {

View file

@ -213,6 +213,22 @@ type blockBody struct {
// blockBodiesData is the network packet for block content distribution. // blockBodiesData is the network packet for block content distribution.
type blockBodiesData []*blockBody type blockBodiesData []*blockBody
// blockBodyRLP represents the RLP encoded data content of a single block.
type blockBodyRLP []byte
// EncodeRLP is a specialized encoder for a block body to pass the already
// encoded body RLPs from the database on, without double encoding.
func (b *blockBodyRLP) EncodeRLP(w io.Writer) error {
if _, err := w.Write([]byte(*b)); err != nil {
return err
}
return nil
}
// blockBodiesRLPData is the network packet for block content distribution
// based on original RLP formatting (i.e. skip the db-decode/proto-encode).
type blockBodiesRLPData []*blockBodyRLP
// nodeDataData is the network response packet for a node data retrieval. // nodeDataData is the network response packet for a node data retrieval.
type nodeDataData []struct { type nodeDataData []struct {
Value []byte Value []byte