Skip to content

G
Geth-Modification
Commit 0ff35e17 authored by Péter Szilágyi's avatar Péter Szilágyi Committed by GitHub
Browse files
Options

core: remove redundant storage of transactions and receipts (#14801) 

* core: remove redundant storage of transactions and receipts

* core, eth, internal: new transaction schema usage polishes

* eth: implement upgrade mechanism for db deduplication

* core, eth: drop old sequential key db upgrader

* eth: close last iterator on successful db upgrage

* core: prefix the lookup entries to make their purpose clearer
parent 8d6a5a35
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 274 additions and 628 deletions
accounts/abi/bind/backends/simulated.go
View file @ 0ff35e17
......@@ -144,7 +144,8 @@ func (b *SimulatedBackend) StorageAt(ctx context.Context, contract common.Addres
// TransactionReceipt returns the receipt of a transaction.
func (b *SimulatedBackend) TransactionReceipt(ctx context.Context, txHash common.Hash) (*types.Receipt, error) {
return core.GetReceipt(b.database, txHash), nil
receipt, _, _, _ := core.GetReceipt(b.database, txHash)
return receipt, nil
}
// PendingCodeAt returns the code associated with an account in the pending state.
......
core/blockchain.go
View file @ 0ff35e17
......@@ -759,16 +759,10 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [
log.Crit("Failed to write log blooms", "err", err)
return
}
if err := WriteTransactions(bc.chainDb, block); err != nil {
errs[index] = fmt.Errorf("failed to write individual transactions: %v", err)
if err := WriteTxLookupEntries(bc.chainDb, block); err != nil {
errs[index] = fmt.Errorf("failed to write lookup metadata: %v", err)
atomic.AddInt32(&failed, 1)
log.Crit("Failed to write individual transactions", "err", err)
return
}
if err := WriteReceipts(bc.chainDb, receipts); err != nil {
errs[index] = fmt.Errorf("failed to write individual receipts: %v", err)
atomic.AddInt32(&failed, 1)
log.Crit("Failed to write individual receipts", "err", err)
log.Crit("Failed to write lookup metadata", "err", err)
return
}
atomic.AddInt32(&stats.processed, 1)
......@@ -1002,12 +996,8 @@ func (bc *BlockChain) InsertChain(chain types.Blocks) (int, error) {
blockInsertTimer.UpdateSince(bstart)
events = append(events, ChainEvent{block, block.Hash(), logs})
// This puts transactions in a extra db for rpc
if err := WriteTransactions(bc.chainDb, block); err != nil {
return i, err
}
// store the receipts
if err := WriteReceipts(bc.chainDb, receipts); err != nil {
// Write the positional metadata for transaction and receipt lookups
if err := WriteTxLookupEntries(bc.chainDb, block); err != nil {
return i, err
}
// Write map map bloom filters
......@@ -1167,16 +1157,12 @@ func (bc *BlockChain) reorg(oldBlock, newBlock *types.Block) error {
for _, block := range newChain {
// insert the block in the canonical way, re-writing history
bc.insert(block)
// write canonical receipts and transactions
if err := WriteTransactions(bc.chainDb, block); err != nil {
return err
}
receipts := GetBlockReceipts(bc.chainDb, block.Hash(), block.NumberU64())
// write receipts
if err := WriteReceipts(bc.chainDb, receipts); err != nil {
// write lookup entries for hash based transaction/receipt searches
if err := WriteTxLookupEntries(bc.chainDb, block); err != nil {
return err
}
// Write map map bloom filters
receipts := GetBlockReceipts(bc.chainDb, block.Hash(), block.NumberU64())
if err := WriteMipmapBloom(bc.chainDb, block.NumberU64(), receipts); err != nil {
return err
}
......@@ -1188,8 +1174,7 @@ func (bc *BlockChain) reorg(oldBlock, newBlock *types.Block) error {
// When transactions get deleted from the database that means the
// receipts that were created in the fork must also be deleted
for _, tx := range diff {
DeleteReceipt(bc.chainDb, tx.Hash())
DeleteTransaction(bc.chainDb, tx.Hash())
DeleteTxLookupEntry(bc.chainDb, tx.Hash())
}
// Must be posted in a goroutine because of the transaction pool trying
// to acquire the chain manager lock
......
core/blockchain_test.go
View file @ 0ff35e17
......@@ -806,8 +806,8 @@ func TestChainTxReorgs(t *testing.T) {
if txn, _, _, _ := GetTransaction(db, tx.Hash()); txn != nil {
t.Errorf("drop %d: tx %v found while shouldn't have been", i, txn)
}
if GetReceipt(db, tx.Hash()) != nil {
t.Errorf("drop %d: receipt found while shouldn't have been", i)
if rcpt, _, _, _ := GetReceipt(db, tx.Hash()); rcpt != nil {
t.Errorf("drop %d: receipt %v found while shouldn't have been", i, rcpt)
}
}
// added tx
......@@ -815,7 +815,7 @@ func TestChainTxReorgs(t *testing.T) {
if txn, _, _, _ := GetTransaction(db, tx.Hash()); txn == nil {
t.Errorf("add %d: expected tx to be found", i)
}
if GetReceipt(db, tx.Hash()) == nil {
if rcpt, _, _, _ := GetReceipt(db, tx.Hash()); rcpt == nil {
t.Errorf("add %d: expected receipt to be found", i)
}
}
......@@ -824,7 +824,7 @@ func TestChainTxReorgs(t *testing.T) {
if txn, _, _, _ := GetTransaction(db, tx.Hash()); txn == nil {
t.Errorf("share %d: expected tx to be found", i)
}
if GetReceipt(db, tx.Hash()) == nil {
if rcpt, _, _, _ := GetReceipt(db, tx.Hash()); rcpt == nil {
t.Errorf("share %d: expected receipt to be found", i)
}
}
......
core/database_util.go
View file @ 0ff35e17
This diff is collapsed.
core/database_util_test.go
View file @ 0ff35e17
......@@ -290,8 +290,8 @@ func TestHeadStorage(t *testing.T) {
}
}
// Tests that transactions and associated metadata can be stored and retrieved.
func TestTransactionStorage(t *testing.T) {
// Tests that positional lookup metadata can be stored and retrieved.
func TestLookupStorage(t *testing.T) {
db, _ := ethdb.NewMemDatabase()
tx1 := types.NewTransaction(1, common.BytesToAddress([]byte{0x11}), big.NewInt(111), big.NewInt(1111), big.NewInt(11111), []byte{0x11, 0x11, 0x11})
......@@ -308,7 +308,10 @@ func TestTransactionStorage(t *testing.T) {
}
}
// Insert all the transactions into the database, and verify contents
if err := WriteTransactions(db, block); err != nil {
if err := WriteBlock(db, block); err != nil {
t.Fatalf("failed to write block contents: %v", err)
}
if err := WriteTxLookupEntries(db, block); err != nil {
t.Fatalf("failed to write transactions: %v", err)
}
for i, tx := range txs {
......@@ -325,72 +328,13 @@ func TestTransactionStorage(t *testing.T) {
}
// Delete the transactions and check purge
for i, tx := range txs {
DeleteTransaction(db, tx.Hash())
DeleteTxLookupEntry(db, tx.Hash())
if txn, _, _, _ := GetTransaction(db, tx.Hash()); txn != nil {
t.Fatalf("tx #%d [%x]: deleted transaction returned: %v", i, tx.Hash(), txn)
}
}
}
// Tests that receipts can be stored and retrieved.
func TestReceiptStorage(t *testing.T) {
db, _ := ethdb.NewMemDatabase()
receipt1 := &types.Receipt{
PostState: []byte{0x01},
CumulativeGasUsed: big.NewInt(1),
Logs: []*types.Log{
{Address: common.BytesToAddress([]byte{0x11})},
{Address: common.BytesToAddress([]byte{0x01, 0x11})},
},
TxHash: common.BytesToHash([]byte{0x11, 0x11}),
ContractAddress: common.BytesToAddress([]byte{0x01, 0x11, 0x11}),
GasUsed: big.NewInt(111111),
}
receipt2 := &types.Receipt{
PostState: []byte{0x02},
CumulativeGasUsed: big.NewInt(2),
Logs: []*types.Log{
{Address: common.BytesToAddress([]byte{0x22})},
{Address: common.BytesToAddress([]byte{0x02, 0x22})},
},
TxHash: common.BytesToHash([]byte{0x22, 0x22}),
ContractAddress: common.BytesToAddress([]byte{0x02, 0x22, 0x22}),
GasUsed: big.NewInt(222222),
}
receipts := []*types.Receipt{receipt1, receipt2}
// Check that no receipt entries are in a pristine database
for i, receipt := range receipts {
if r := GetReceipt(db, receipt.TxHash); r != nil {
t.Fatalf("receipt #%d [%x]: non existent receipt returned: %v", i, receipt.TxHash, r)
}
}
// Insert all the receipts into the database, and verify contents
if err := WriteReceipts(db, receipts); err != nil {
t.Fatalf("failed to write receipts: %v", err)
}
for i, receipt := range receipts {
if r := GetReceipt(db, receipt.TxHash); r == nil {
t.Fatalf("receipt #%d [%x]: receipt not found", i, receipt.TxHash)
} else {
rlpHave, _ := rlp.EncodeToBytes(r)
rlpWant, _ := rlp.EncodeToBytes(receipt)
if !bytes.Equal(rlpHave, rlpWant) {
t.Fatalf("receipt #%d [%x]: receipt mismatch: have %v, want %v", i, receipt.TxHash, r, receipt)
}
}
}
// Delete the receipts and check purge
for i, receipt := range receipts {
DeleteReceipt(db, receipt.TxHash)
if r := GetReceipt(db, receipt.TxHash); r != nil {
t.Fatalf("receipt #%d [%x]: deleted receipt returned: %v", i, receipt.TxHash, r)
}
}
}
// Tests that receipts associated with a single block can be stored and retrieved.
func TestBlockReceiptStorage(t *testing.T) {
db, _ := ethdb.NewMemDatabase()
......@@ -530,10 +474,6 @@ func TestMipmapChain(t *testing.T) {
}
// store the receipts
err := WriteReceipts(db, receipts)
if err != nil {
t.Fatal(err)
}
WriteMipmapBloom(db, uint64(i+1), receipts)
})
for i, block := range chain {
......
eth/backend.go
View file @ 0ff35e17
......@@ -59,8 +59,8 @@ type LesServer interface {
type Ethereum struct {
chainConfig *params.ChainConfig
// Channel for shutting down the service
shutdownChan chan bool // Channel for shutting down the ethereum
stopDbUpgrade func() // stop chain db sequential key upgrade
shutdownChan chan bool // Channel for shutting down the ethereum
stopDbUpgrade func() error // stop chain db sequential key upgrade
// Handlers
txPool *core.TxPool
blockchain *core.BlockChain
......@@ -103,7 +103,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
if err != nil {
return nil, err
}
stopDbUpgrade := upgradeSequentialKeys(chainDb)
stopDbUpgrade := upgradeDeduplicateData(chainDb)
chainConfig, genesisHash, genesisErr := core.SetupGenesisBlock(chainDb, config.Genesis)
if _, ok := genesisErr.(*params.ConfigCompatError); genesisErr != nil && !ok {
return nil, genesisErr
......
eth/backend_test.go
View file @ 0ff35e17
......@@ -33,24 +33,15 @@ func TestMipmapUpgrade(t *testing.T) {
genesis := new(core.Genesis).MustCommit(db)
chain, receipts := core.GenerateChain(params.TestChainConfig, 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.Logs = []*types.Log{{Address: addr}}
gen.AddUncheckedReceipt(receipt)
receipts = types.Receipts{receipt}
case 2:
receipt := types.NewReceipt(nil, new(big.Int))
receipt.Logs = []*types.Log{{Address: addr}}
gen.AddUncheckedReceipt(receipt)
receipts = types.Receipts{receipt}
}
// store the receipts
err := core.WriteReceipts(db, receipts)
if err != nil {
t.Fatal(err)
}
})
for i, block := range chain {
......
eth/db_upgrade.go
View file @ 0ff35e17
......@@ -19,237 +19,120 @@ package eth
import (
"bytes"
"encoding/binary"
"fmt"
"math/big"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/rlp"
)
var useSequentialKeys = []byte("dbUpgrade_20160530sequentialKeys")
var deduplicateData = []byte("dbUpgrade_20170714deduplicateData")
// upgradeSequentialKeys checks the chain database version and
// upgradeDeduplicateData checks the chain database version and
// starts a background process to make upgrades if necessary.
// Returns a stop function that blocks until the process has
// been safely stopped.
func upgradeSequentialKeys(db ethdb.Database) (stopFn func()) {
data, _ := db.Get(useSequentialKeys)
func upgradeDeduplicateData(db ethdb.Database) func() error {
// If the database is already converted or empty, bail out
data, _ := db.Get(deduplicateData)
if len(data) > 0 && data[0] == 42 {
return nil // already converted
return nil
}
if data, _ := db.Get([]byte("LastHeader")); len(data) == 0 {
db.Put(useSequentialKeys, []byte{42})
return nil // empty database, nothing to do
db.Put(deduplicateData, []byte{42})
return nil
}
log.Warn("Upgrading chain database to use sequential keys")
stopChn := make(chan struct{})
stoppedChn := make(chan struct{})
// Start the deduplication upgrade on a new goroutine
log.Warn("Upgrading database to use lookup entries")
stop := make(chan chan error)
go func() {
stopFn := func() bool {
select {
case <-time.After(time.Microsecond * 100): // make sure other processes don't get starved
case <-stopChn:
return true
}
return false
}
err, stopped := upgradeSequentialCanonicalNumbers(db, stopFn)
if err == nil && !stopped {
err, stopped = upgradeSequentialBlocks(db, stopFn)
}
if err == nil && !stopped {
err, stopped = upgradeSequentialOrphanedReceipts(db, stopFn)
}
if err == nil && !stopped {
log.Info("Database conversion successful")
db.Put(useSequentialKeys, []byte{42})
}
if err != nil {
log.Error("Database conversion failed", "err", err)
}
close(stoppedChn)
}()
return func() {
close(stopChn)
<-stoppedChn
}
}
// upgradeSequentialCanonicalNumbers reads all old format canonical numbers from
// the database, writes them in new format and deletes the old ones if successful.
func upgradeSequentialCanonicalNumbers(db ethdb.Database, stopFn func() bool) (error, bool) {
prefix := []byte("block-num-")
it := db.(*ethdb.LDBDatabase).NewIterator()
defer func() {
it.Release()
}()
it.Seek(prefix)
cnt := 0
for bytes.HasPrefix(it.Key(), prefix) {
keyPtr := it.Key()
if len(keyPtr) < 20 {
cnt++
if cnt%100000 == 0 {
// Create an iterator to read the entire database and covert old lookup entires
it := db.(*ethdb.LDBDatabase).NewIterator()
defer func() {
if it != nil {
it.Release()
it = db.(*ethdb.LDBDatabase).NewIterator()
it.Seek(keyPtr)
log.Info("Converting canonical numbers", "count", cnt)
}
number := big.NewInt(0).SetBytes(keyPtr[10:]).Uint64()
newKey := []byte("h12345678n")
binary.BigEndian.PutUint64(newKey[1:9], number)
if err := db.Put(newKey, it.Value()); err != nil {
return err, false
}()
var (
converted uint64
failed error
)
for failed == nil && it.Next() {
// Skip any entries that don't look like old transaction meta entires (<hash>0x01)
key := it.Key()
if len(key) != common.HashLength+1 || key[common.HashLength] != 0x01 {
continue
}
if err := db.Delete(keyPtr); err != nil {
return err, false
// Skip any entries that don't contain metadata (name clash between <hash>0x01 and <some-prefix><hash>)
var meta struct {
BlockHash common.Hash
BlockIndex uint64
Index uint64
}
}
if stopFn() {
return nil, true
}
it.Next()
}
if cnt > 0 {
log.Info("converted canonical numbers", "count", cnt)
}
return nil, false
}
// upgradeSequentialBlocks reads all old format block headers, bodies, TDs and block
// receipts from the database, writes them in new format and deletes the old ones
// if successful.
func upgradeSequentialBlocks(db ethdb.Database, stopFn func() bool) (error, bool) {
prefix := []byte("block-")
it := db.(*ethdb.LDBDatabase).NewIterator()
defer func() {
it.Release()
}()
it.Seek(prefix)
cnt := 0
for bytes.HasPrefix(it.Key(), prefix) {
keyPtr := it.Key()
if len(keyPtr) >= 38 {
cnt++
if cnt%10000 == 0 {
it.Release()
it = db.(*ethdb.LDBDatabase).NewIterator()
it.Seek(keyPtr)
log.Info("Converting blocks", "count", cnt)
if err := rlp.DecodeBytes(it.Value(), &meta); err != nil {
continue
}
// convert header, body, td and block receipts
var keyPrefix [38]byte
copy(keyPrefix[:], keyPtr[0:38])
hash := keyPrefix[6:38]
if err := upgradeSequentialBlockData(db, hash); err != nil {
return err, false
}
// delete old db entries belonging to this hash
for bytes.HasPrefix(it.Key(), keyPrefix[:]) {
if err := db.Delete(it.Key()); err != nil {
return err, false
// Skip any already upgraded entries (clash due to <hash> ending with 0x01 (old suffix))
hash := key[:common.HashLength]
if hash[0] == byte('l') {
// Potential clash, the "old" `hash` must point to a live transaction.
if tx, _, _, _ := core.GetTransaction(db, common.BytesToHash(hash)); tx == nil || !bytes.Equal(tx.Hash().Bytes(), hash) {
continue
}
it.Next()
}
if err := db.Delete(append([]byte("receipts-block-"), hash...)); err != nil {
return err, false
// Convert the old metadata to a new lookup entry, delete duplicate data
if failed = db.Put(append([]byte("l"), hash...), it.Value()); failed == nil { // Write the new looku entry
if failed = db.Delete(hash); failed == nil { // Delete the duplicate transaction data
if failed = db.Delete(append([]byte("receipts-"), hash...)); failed == nil { // Delete the duplicate receipt data
if failed = db.Delete(key); failed != nil { // Delete the old transaction metadata
break
}
}
}
}
} else {
it.Next()
}
if stopFn() {
return nil, true
}
}
if cnt > 0 {
log.Info("Converted blocks", "count", cnt)
}
return nil, false
}
// Bump the conversion counter, and recreate the iterator occasionally to
// avoid too high memory consumption.
converted++
if converted%100000 == 0 {
it.Release()
it = db.(*ethdb.LDBDatabase).NewIterator()
it.Seek(key)
// upgradeSequentialOrphanedReceipts removes any old format block receipts from the
// database that did not have a corresponding block
func upgradeSequentialOrphanedReceipts(db ethdb.Database, stopFn func() bool) (error, bool) {
prefix := []byte("receipts-block-")
it := db.(*ethdb.LDBDatabase).NewIterator()
defer it.Release()
it.Seek(prefix)
cnt := 0
for bytes.HasPrefix(it.Key(), prefix) {
// phase 2 already converted receipts belonging to existing
// blocks, just remove if there's anything left
cnt++
if err := db.Delete(it.Key()); err != nil {
return err, false
log.Info("Deduplicating database entries", "deduped", converted)
}
// Check for termination, or continue after a bit of a timeout
select {
case errc := <-stop:
errc <- nil
return
case <-time.After(time.Microsecond * 100):
}
}
if stopFn() {
return nil, true
// Upgrade finished, mark a such and terminate
if failed == nil {
log.Info("Database deduplication successful", "deduped", converted)
db.Put(deduplicateData, []byte{42})
} else {
log.Error("Database deduplication failed", "deduped", converted, "err", failed)
}
it.Next()
}
if cnt > 0 {
log.Info("Removed orphaned block receipts", "count", cnt)
}
return nil, false
}
it.Release()
it = nil
// upgradeSequentialBlockData upgrades the header, body, td and block receipts
// database entries belonging to a single hash (doesn't delete old data).
func upgradeSequentialBlockData(db ethdb.Database, hash []byte) error {
// get old chain data and block number
headerRLP, _ := db.Get(append(append([]byte("block-"), hash...), []byte("-header")...))
if len(headerRLP) == 0 {
return nil
}
header := new(types.Header)
if err := rlp.Decode(bytes.NewReader(headerRLP), header); err != nil {
return err
}
number := header.Number.Uint64()
bodyRLP, _ := db.Get(append(append([]byte("block-"), hash...), []byte("-body")...))
tdRLP, _ := db.Get(append(append([]byte("block-"), hash...), []byte("-td")...))
receiptsRLP, _ := db.Get(append([]byte("receipts-block-"), hash...))
// store new hash -> number association
encNum := make([]byte, 8)
binary.BigEndian.PutUint64(encNum, number)
if err := db.Put(append([]byte("H"), hash...), encNum); err != nil {
return err
}
// store new chain data
if err := db.Put(append(append([]byte("h"), encNum...), hash...), headerRLP); err != nil {
return err
}
if len(tdRLP) != 0 {
if err := db.Put(append(append(append([]byte("h"), encNum...), hash...), []byte("t")...), tdRLP); err != nil {
return err
}
}
if len(bodyRLP) != 0 {
if err := db.Put(append(append([]byte("b"), encNum...), hash...), bodyRLP); err != nil {
return err
}
}
if len(receiptsRLP) != 0 {
if err := db.Put(append(append([]byte("r"), encNum...), hash...), receiptsRLP); err != nil {
return err
}
errc := <-stop
errc <- failed
}()
// Assembly the cancellation callback
return func() error {
errc := make(chan error)
stop <- errc
return <-errc
}
return nil
}
func addMipmapBloomBins(db ethdb.Database) (err error) {
......
eth/filters/filter_test.go
View file @ 0ff35e17
......@@ -82,12 +82,6 @@ func BenchmarkMipmaps(b *testing.B) {
gen.AddUncheckedReceipt(receipt)
}
// store the receipts
err := core.WriteReceipts(db, receipts)
if err != nil {
b.Fatal(err)
}
core.WriteMipmapBloom(db, uint64(i+1), receipts)
})
for i, block := range chain {
......@@ -183,12 +177,6 @@ func TestFilters(t *testing.T) {
gen.AddUncheckedReceipt(receipt)
receipts = types.Receipts{receipt}
}
// store the receipts
err := core.WriteReceipts(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
......
internal/ethapi/api.go
View file @ 0ff35e17
This diff is collapsed.
light/txpool.go
View file @ 0ff35e17
......@@ -130,19 +130,6 @@ type txBlockData struct {
Index uint64
}
// storeTxBlockData stores the block position of a mined tx in the local db
func (pool *TxPool) storeTxBlockData(txh common.Hash, tbd txBlockData) {
//fmt.Println("storeTxBlockData", txh, tbd)
data, _ := rlp.EncodeToBytes(tbd)
pool.chainDb.Put(append(txh[:], byte(1)), data)
}
// removeTxBlockData removes the stored block position of a rolled back tx
func (pool *TxPool) removeTxBlockData(txh common.Hash) {
//fmt.Println("removeTxBlockData", txh)
pool.chainDb.Delete(append(txh[:], byte(1)))
}
// txStateChanges stores the recent changes between pending/mined states of
// transactions. True means mined, false means rolled back, no entry means no change
type txStateChanges map[common.Hash]bool
......@@ -172,59 +159,48 @@ func (txc txStateChanges) getLists() (mined []common.Hash, rollback []common.Has
// checkMinedTxs checks newly added blocks for the currently pending transactions
// and marks them as mined if necessary. It also stores block position in the db
// and adds them to the received txStateChanges map.
func (pool *TxPool) checkMinedTxs(ctx context.Context, hash common.Hash, idx uint64, txc txStateChanges) error {
//fmt.Println("checkMinedTxs")
func (pool *TxPool) checkMinedTxs(ctx context.Context, hash common.Hash, number uint64, txc txStateChanges) error {
// If no transactions are pending, we don't care about anything
if len(pool.pending) == 0 {
return nil
}
//fmt.Println("len(pool) =", len(pool.pending))
block, err := GetBlock(ctx, pool.odr, hash, idx)
var receipts types.Receipts
block, err := GetBlock(ctx, pool.odr, hash, number)
if err != nil {
//fmt.Println(err)
return err
}
//fmt.Println("len(block.Transactions()) =", len(block.Transactions()))
// Gather all the local transaction mined in this block
list := pool.mined[hash]
for i, tx := range block.Transactions() {
txHash := tx.Hash()
//fmt.Println(" txHash:", txHash)
if tx, ok := pool.pending[txHash]; ok {
//fmt.Println("TX FOUND")
if receipts == nil {
receipts, err = GetBlockReceipts(ctx, pool.odr, hash, idx)
if err != nil {
return err
}
if len(receipts) != len(block.Transactions()) {
panic(nil) // should never happen if hashes did match
}
core.SetReceiptsData(pool.config, block, receipts)
}
//fmt.Println("WriteReceipt", receipts[i].TxHash)
core.WriteReceipt(pool.chainDb, receipts[i])
pool.storeTxBlockData(txHash, txBlockData{hash, idx, uint64(i)})
delete(pool.pending, txHash)
for _, tx := range block.Transactions() {
if _, ok := pool.pending[tx.Hash()]; ok {
list = append(list, tx)
txc.setState(txHash, true)
}
}
// If some transactions have been mined, write the needed data to disk and update
if list != nil {
// Retrieve all the receipts belonging to this block and write the loopup table
if _, err := GetBlockReceipts(ctx, pool.odr, hash, number); err != nil { // ODR caches, ignore results
return err
}
if err := core.WriteTxLookupEntries(pool.chainDb, block); err != nil {
return err
}
// Update the transaction pool's state
for _, tx := range list {
delete(pool.pending, tx.Hash())
txc.setState(tx.Hash(), true)
}
pool.mined[hash] = list
}
return nil
}
// rollbackTxs marks the transactions contained in recently rolled back blocks
// as rolled back. It also removes block position info from the db and adds them
// to the received txStateChanges map.
// as rolled back. It also removes any positional lookup entries.
func (pool *TxPool) rollbackTxs(hash common.Hash, txc txStateChanges) {
if list, ok := pool.mined[hash]; ok {
for _, tx := range list {
txHash := tx.Hash()
pool.removeTxBlockData(txHash)
core.DeleteTxLookupEntry(pool.chainDb, txHash)
pool.pending[txHash] = tx
txc.setState(txHash, false)
}
......
miner/worker.go
View file @ 0ff35e17
......@@ -293,9 +293,7 @@ func (self *worker) wait() {
// check if canon block and write transactions
if stat == core.CanonStatTy {
// This puts transactions in a extra db for rpc
core.WriteTransactions(self.chainDb, block)
// store the receipts
core.WriteReceipts(self.chainDb, work.receipts)
core.WriteTxLookupEntries(self.chainDb, block)
// Write map map bloom filters
core.WriteMipmapBloom(self.chainDb, block.NumberU64(), work.receipts)
// implicit by posting ChainHeadEvent
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment