Backport: core: remove redundant storage of transactions and receipts

This commit is contained in:
Julian Yap 2018-07-03 12:48:03 -10:00
parent d6682ebcb9
commit f0992e40b2
12 changed files with 267 additions and 714 deletions

View file

@ -144,7 +144,8 @@ func (b *SimulatedBackend) StorageAt(ctx context.Context, contract common.Addres
// TransactionReceipt returns the receipt of a transaction. // TransactionReceipt returns the receipt of a transaction.
func (b *SimulatedBackend) TransactionReceipt(ctx context.Context, txHash common.Hash) (*types.Receipt, error) { 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. // PendingCodeAt returns the code associated with an account in the pending state.

View file

@ -779,16 +779,10 @@ func (self *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain
glog.Fatal(errs[index]) glog.Fatal(errs[index])
return return
} }
if err := WriteTransactions(self.chainDb, block); err != nil { if err := WriteTxLookupEntries(self.chainDb, block); err != nil {
errs[index] = fmt.Errorf("failed to write individual transactions: %v", err) errs[index] = fmt.Errorf("failed to write lookup metadata: %v", err)
atomic.AddInt32(&failed, 1) atomic.AddInt32(&failed, 1)
glog.Fatal(errs[index]) glog.Fatal("Failed to write lookup metadata", "err", err)
return
}
if err := WriteReceipts(self.chainDb, receipts); err != nil {
errs[index] = fmt.Errorf("failed to write individual receipts: %v", err)
atomic.AddInt32(&failed, 1)
glog.Fatal(errs[index])
return return
} }
atomic.AddInt32(&stats.processed, 1) atomic.AddInt32(&stats.processed, 1)
@ -1027,12 +1021,8 @@ func (self *BlockChain) InsertChain(chain types.Blocks) (int, error) {
blockInsertTimer.UpdateSince(bstart) blockInsertTimer.UpdateSince(bstart)
events = append(events, ChainEvent{block, block.Hash(), logs}) events = append(events, ChainEvent{block, block.Hash(), logs})
// This puts transactions in a extra db for rpc // Write the positional metadata for transaction and receipt lookups
if err := WriteTransactions(self.chainDb, block); err != nil { if err := WriteTxLookupEntries(self.chainDb, block); err != nil {
return i, err
}
// store the receipts
if err := WriteReceipts(self.chainDb, receipts); err != nil {
return i, err return i, err
} }
// Write map map bloom filters // Write map map bloom filters
@ -1201,16 +1191,12 @@ func (self *BlockChain) reorg(oldBlock, newBlock *types.Block) error {
for _, block := range newChain { for _, block := range newChain {
// insert the block in the canonical way, re-writing history // insert the block in the canonical way, re-writing history
self.insert(block) self.insert(block)
// write canonical receipts and transactions // write lookup entries for hash based transaction/receipt searches
if err := WriteTransactions(self.chainDb, block); err != nil { if err := WriteTxLookupEntries(self.chainDb, block); err != nil {
return err
}
receipts := GetBlockReceipts(self.chainDb, block.Hash(), block.NumberU64())
// write receipts
if err := WriteReceipts(self.chainDb, receipts); err != nil {
return err return err
} }
// Write map map bloom filters // Write map map bloom filters
receipts := GetBlockReceipts(self.chainDb, block.Hash(), block.NumberU64())
if err := WriteMipmapBloom(self.chainDb, block.NumberU64(), receipts); err != nil { if err := WriteMipmapBloom(self.chainDb, block.NumberU64(), receipts); err != nil {
return err return err
} }
@ -1222,8 +1208,7 @@ func (self *BlockChain) reorg(oldBlock, newBlock *types.Block) error {
// When transactions get deleted from the database that means the // When transactions get deleted from the database that means the
// receipts that were created in the fork must also be deleted // receipts that were created in the fork must also be deleted
for _, tx := range diff { for _, tx := range diff {
DeleteReceipt(self.chainDb, tx.Hash()) DeleteTxLookupEntry(self.chainDb, tx.Hash())
DeleteTransaction(self.chainDb, tx.Hash())
} }
// Must be posted in a goroutine because of the transaction pool trying // Must be posted in a goroutine because of the transaction pool trying
// to acquire the chain manager lock // to acquire the chain manager lock

View file

@ -28,6 +28,7 @@ import (
"time" "time"
"github.com/ethereum/ethash" "github.com/ethereum/ethash"
"github.com/hashicorp/golang-lru"
"github.com/ubiq/go-ubiq/common" "github.com/ubiq/go-ubiq/common"
"github.com/ubiq/go-ubiq/core/state" "github.com/ubiq/go-ubiq/core/state"
"github.com/ubiq/go-ubiq/core/types" "github.com/ubiq/go-ubiq/core/types"
@ -38,7 +39,6 @@ import (
"github.com/ubiq/go-ubiq/params" "github.com/ubiq/go-ubiq/params"
"github.com/ubiq/go-ubiq/pow" "github.com/ubiq/go-ubiq/pow"
"github.com/ubiq/go-ubiq/rlp" "github.com/ubiq/go-ubiq/rlp"
"github.com/hashicorp/golang-lru"
) )
func init() { func init() {
@ -949,8 +949,8 @@ func TestChainTxReorgs(t *testing.T) {
if txn, _, _, _ := GetTransaction(db, tx.Hash()); txn != nil { if txn, _, _, _ := GetTransaction(db, tx.Hash()); txn != nil {
t.Errorf("drop %d: tx %v found while shouldn't have been", i, txn) t.Errorf("drop %d: tx %v found while shouldn't have been", i, txn)
} }
if GetReceipt(db, tx.Hash()) != nil { if rcpt, _, _, _ := GetReceipt(db, tx.Hash()); rcpt != nil {
t.Errorf("drop %d: receipt found while shouldn't have been", i) t.Errorf("drop %d: receipt %v found while shouldn't have been", i, rcpt)
} }
} }
// added tx // added tx
@ -958,7 +958,7 @@ func TestChainTxReorgs(t *testing.T) {
if txn, _, _, _ := GetTransaction(db, tx.Hash()); txn == nil { if txn, _, _, _ := GetTransaction(db, tx.Hash()); txn == nil {
t.Errorf("add %d: expected tx to be found", i) 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) t.Errorf("add %d: expected receipt to be found", i)
} }
} }
@ -967,7 +967,7 @@ func TestChainTxReorgs(t *testing.T) {
if txn, _, _, _ := GetTransaction(db, tx.Hash()); txn == nil { if txn, _, _, _ := GetTransaction(db, tx.Hash()); txn == nil {
t.Errorf("share %d: expected tx to be found", i) 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) t.Errorf("share %d: expected receipt to be found", i)
} }
} }

View file

@ -46,24 +46,17 @@ var (
blockHashPrefix = []byte("H") // blockHashPrefix + hash -> num (uint64 big endian) blockHashPrefix = []byte("H") // blockHashPrefix + hash -> num (uint64 big endian)
bodyPrefix = []byte("b") // bodyPrefix + num (uint64 big endian) + hash -> block body bodyPrefix = []byte("b") // bodyPrefix + num (uint64 big endian) + hash -> block body
blockReceiptsPrefix = []byte("r") // blockReceiptsPrefix + num (uint64 big endian) + hash -> block receipts blockReceiptsPrefix = []byte("r") // blockReceiptsPrefix + num (uint64 big endian) + hash -> block receipts
lookupPrefix = []byte("l") // lookupPrefix + hash -> transaction/receipt lookup metadata
preimagePrefix = "secure-key-" // preimagePrefix + hash -> preimage preimagePrefix = "secure-key-" // preimagePrefix + hash -> preimage
txMetaSuffix = []byte{0x01}
receiptsPrefix = []byte("receipts-")
mipmapPre = []byte("mipmap-log-bloom-") mipmapPre = []byte("mipmap-log-bloom-")
MIPMapLevels = []uint64{1000000, 500000, 100000, 50000, 1000} MIPMapLevels = []uint64{1000000, 500000, 100000, 50000, 1000}
configPrefix = []byte("ubiq-config-") // config prefix for the db configPrefix = []byte("ubiq-config-") // config prefix for the db
// used by old (non-sequential keys) db, now only used for conversion // used by old db, now only used for conversion
oldBlockPrefix = []byte("block-") oldReceiptsPrefix = []byte("receipts-")
oldHeaderSuffix = []byte("-header") oldTxMetaSuffix = []byte{0x01}
oldTdSuffix = []byte("-td") // headerPrefix + num (uint64 big endian) + hash + tdSuffix -> td
oldBodySuffix = []byte("-body")
oldBlockNumPrefix = []byte("block-num-")
oldBlockReceiptsPrefix = []byte("receipts-block-")
oldBlockHashPrefix = []byte("block-hash-") // [deprecated by the header/block split, remove eventually]
ChainConfigNotFoundErr = errors.New("ChainConfig not found") // general config not found error ChainConfigNotFoundErr = errors.New("ChainConfig not found") // general config not found error
@ -73,6 +66,14 @@ var (
preimageHitCounter = metrics.NewCounter("db/preimage/hits") preimageHitCounter = metrics.NewCounter("db/preimage/hits")
) )
// txLookupEntry is a positional metadata to help looking up the data content of
// a transaction or receipt given only its hash.
type txLookupEntry struct {
BlockHash common.Hash
BlockIndex uint64
Index uint64
}
// encodeBlockNumber encodes a block number as big endian uint64 // encodeBlockNumber encodes a block number as big endian uint64
func encodeBlockNumber(number uint64) []byte { func encodeBlockNumber(number uint64) []byte {
enc := make([]byte, 8) enc := make([]byte, 8)
@ -83,12 +84,9 @@ func encodeBlockNumber(number uint64) []byte {
// GetCanonicalHash retrieves a hash assigned to a canonical block number. // GetCanonicalHash retrieves a hash assigned to a canonical block number.
func GetCanonicalHash(db ethdb.Database, number uint64) common.Hash { func GetCanonicalHash(db ethdb.Database, number uint64) common.Hash {
data, _ := db.Get(append(append(headerPrefix, encodeBlockNumber(number)...), numSuffix...)) data, _ := db.Get(append(append(headerPrefix, encodeBlockNumber(number)...), numSuffix...))
if len(data) == 0 {
data, _ = db.Get(append(oldBlockNumPrefix, big.NewInt(int64(number)).Bytes()...))
if len(data) == 0 { if len(data) == 0 {
return common.Hash{} return common.Hash{}
} }
}
return common.BytesToHash(data) return common.BytesToHash(data)
} }
@ -101,16 +99,8 @@ const missingNumber = uint64(0xffffffffffffffff)
func GetBlockNumber(db ethdb.Database, hash common.Hash) uint64 { func GetBlockNumber(db ethdb.Database, hash common.Hash) uint64 {
data, _ := db.Get(append(blockHashPrefix, hash.Bytes()...)) data, _ := db.Get(append(blockHashPrefix, hash.Bytes()...))
if len(data) != 8 { if len(data) != 8 {
data, _ := db.Get(append(append(oldBlockPrefix, hash.Bytes()...), oldHeaderSuffix...))
if len(data) == 0 {
return missingNumber return missingNumber
} }
header := new(types.Header)
if err := rlp.Decode(bytes.NewReader(data), header); err != nil {
glog.Fatalf("failed to decode block header: %v", err)
}
return header.Number.Uint64()
}
return binary.BigEndian.Uint64(data) return binary.BigEndian.Uint64(data)
} }
@ -152,9 +142,6 @@ func GetHeadFastBlockHash(db ethdb.Database) common.Hash {
// if the header's not found. // if the header's not found.
func GetHeaderRLP(db ethdb.Database, hash common.Hash, number uint64) rlp.RawValue { func GetHeaderRLP(db ethdb.Database, hash common.Hash, number uint64) rlp.RawValue {
data, _ := db.Get(append(append(headerPrefix, encodeBlockNumber(number)...), hash.Bytes()...)) data, _ := db.Get(append(append(headerPrefix, encodeBlockNumber(number)...), hash.Bytes()...))
if len(data) == 0 {
data, _ = db.Get(append(append(oldBlockPrefix, hash.Bytes()...), oldHeaderSuffix...))
}
return data return data
} }
@ -176,9 +163,6 @@ func GetHeader(db ethdb.Database, hash common.Hash, number uint64) *types.Header
// GetBodyRLP retrieves the block body (transactions and uncles) in RLP encoding. // GetBodyRLP retrieves the block body (transactions and uncles) in RLP encoding.
func GetBodyRLP(db ethdb.Database, hash common.Hash, number uint64) rlp.RawValue { func GetBodyRLP(db ethdb.Database, hash common.Hash, number uint64) rlp.RawValue {
data, _ := db.Get(append(append(bodyPrefix, encodeBlockNumber(number)...), hash.Bytes()...)) data, _ := db.Get(append(append(bodyPrefix, encodeBlockNumber(number)...), hash.Bytes()...))
if len(data) == 0 {
data, _ = db.Get(append(append(oldBlockPrefix, hash.Bytes()...), oldBodySuffix...))
}
return data return data
} }
@ -201,12 +185,9 @@ func GetBody(db ethdb.Database, hash common.Hash, number uint64) *types.Body {
// none found. // none found.
func GetTd(db ethdb.Database, hash common.Hash, number uint64) *big.Int { func GetTd(db ethdb.Database, hash common.Hash, number uint64) *big.Int {
data, _ := db.Get(append(append(append(headerPrefix, encodeBlockNumber(number)...), hash[:]...), tdSuffix...)) data, _ := db.Get(append(append(append(headerPrefix, encodeBlockNumber(number)...), hash[:]...), tdSuffix...))
if len(data) == 0 {
data, _ = db.Get(append(append(oldBlockPrefix, hash.Bytes()...), oldTdSuffix...))
if len(data) == 0 { if len(data) == 0 {
return nil return nil
} }
}
td := new(big.Int) td := new(big.Int)
if err := rlp.Decode(bytes.NewReader(data), td); err != nil { if err := rlp.Decode(bytes.NewReader(data), td); err != nil {
glog.V(logger.Error).Infof("invalid block total difficulty RLP for hash %x: %v", hash, err) glog.V(logger.Error).Infof("invalid block total difficulty RLP for hash %x: %v", hash, err)
@ -239,12 +220,9 @@ func GetBlock(db ethdb.Database, hash common.Hash, number uint64) *types.Block {
// in a block given by its hash. // in a block given by its hash.
func GetBlockReceipts(db ethdb.Database, hash common.Hash, number uint64) types.Receipts { func GetBlockReceipts(db ethdb.Database, hash common.Hash, number uint64) types.Receipts {
data, _ := db.Get(append(append(blockReceiptsPrefix, encodeBlockNumber(number)...), hash[:]...)) data, _ := db.Get(append(append(blockReceiptsPrefix, encodeBlockNumber(number)...), hash[:]...))
if len(data) == 0 {
data, _ = db.Get(append(oldBlockReceiptsPrefix, hash.Bytes()...))
if len(data) == 0 { if len(data) == 0 {
return nil return nil
} }
}
storageReceipts := []*types.ReceiptForStorage{} storageReceipts := []*types.ReceiptForStorage{}
if err := rlp.DecodeBytes(data, &storageReceipts); err != nil { if err := rlp.DecodeBytes(data, &storageReceipts); err != nil {
glog.V(logger.Error).Infof("invalid receipt array RLP for hash %x: %v", hash, err) glog.V(logger.Error).Infof("invalid receipt array RLP for hash %x: %v", hash, err)
@ -257,6 +235,23 @@ func GetBlockReceipts(db ethdb.Database, hash common.Hash, number uint64) types.
return receipts return receipts
} }
// GetTxLookupEntry retrieves the positional metadata associated with a transaction
// hash to allow retrieving the transaction or receipt by hash.
func GetTxLookupEntry(db ethdb.Database, hash common.Hash) (common.Hash, uint64, uint64) {
// Load the positional metadata from disk and bail if it fails
data, _ := db.Get(append(lookupPrefix, hash.Bytes()...))
if len(data) == 0 {
return common.Hash{}, 0, 0
}
// Parse and return the contents of the lookup entry
var entry txLookupEntry
if err := rlp.DecodeBytes(data, &entry); err != nil {
glog.V(logger.Error).Infof("Invalid lookup entry RLP", "hash", hash, "err", err)
return common.Hash{}, 0, 0
}
return entry.BlockHash, entry.BlockIndex, entry.Index
}
// GetTransaction retrieves a specific transaction from the database, along with // GetTransaction retrieves a specific transaction from the database, along with
// its added positional metadata. // its added positional metadata.
func GetTransaction(db ethdb.Database, hash common.Hash) (*types.Transaction, common.Hash, uint64, uint64) { func GetTransaction(db ethdb.Database, hash common.Hash) (*types.Transaction, common.Hash, uint64, uint64) {
@ -270,33 +265,42 @@ func GetTransaction(db ethdb.Database, hash common.Hash) (*types.Transaction, co
return nil, common.Hash{}, 0, 0 return nil, common.Hash{}, 0, 0
} }
// Retrieve the blockchain positional metadata // Retrieve the blockchain positional metadata
data, _ = db.Get(append(hash.Bytes(), txMetaSuffix...)) data, _ = db.Get(append(hash.Bytes(), oldTxMetaSuffix...))
if len(data) == 0 { if len(data) == 0 {
return nil, common.Hash{}, 0, 0 return nil, common.Hash{}, 0, 0
} }
var meta struct { var entry txLookupEntry
BlockHash common.Hash if err := rlp.DecodeBytes(data, &entry); err != nil {
BlockIndex uint64
Index uint64
}
if err := rlp.DecodeBytes(data, &meta); err != nil {
return nil, common.Hash{}, 0, 0 return nil, common.Hash{}, 0, 0
} }
return &tx, meta.BlockHash, meta.BlockIndex, meta.Index return &tx, entry.BlockHash, entry.BlockIndex, entry.Index
} }
// GetReceipt returns a receipt by hash // GetReceipt retrieves a specific transaction receipt from the database, along with
func GetReceipt(db ethdb.Database, txHash common.Hash) *types.Receipt { // its added positional metadata.
data, _ := db.Get(append(receiptsPrefix, txHash[:]...)) func GetReceipt(db ethdb.Database, hash common.Hash) (*types.Receipt, common.Hash, uint64, uint64) {
// Retrieve the lookup metadata and resolve the receipt from the receipts
blockHash, blockNumber, receiptIndex := GetTxLookupEntry(db, hash)
if blockHash != (common.Hash{}) {
receipts := GetBlockReceipts(db, blockHash, blockNumber)
if len(receipts) <= int(receiptIndex) {
glog.V(logger.Error).Infof("Receipt refereced missing", "number", blockNumber, "hash", blockHash, "index", receiptIndex)
return nil, common.Hash{}, 0, 0
}
return receipts[receiptIndex], blockHash, blockNumber, receiptIndex
}
// Old receipt representation, load the receipt and set an unknown metadata
data, _ := db.Get(append(oldReceiptsPrefix, hash[:]...))
if len(data) == 0 { if len(data) == 0 {
return nil return nil, common.Hash{}, 0, 0
} }
var receipt types.ReceiptForStorage var receipt types.ReceiptForStorage
err := rlp.DecodeBytes(data, &receipt) err := rlp.DecodeBytes(data, &receipt)
if err != nil { if err != nil {
glog.V(logger.Core).Infoln("GetReceipt err:", err) glog.V(logger.Core).Infoln("GetReceipt err:", err)
} }
return (*types.Receipt)(&receipt) return (*types.Receipt)(&receipt), common.Hash{}, 0, 0
} }
// WriteCanonicalHash stores the canonical hash for the given block number. // WriteCanonicalHash stores the canonical hash for the given block number.
@ -421,76 +425,29 @@ func WriteBlockReceipts(db ethdb.Database, hash common.Hash, number uint64, rece
return nil return nil
} }
// WriteTransactions stores the transactions associated with a specific block // WriteTxLookupEntries stores a positional metadata for every transaction from
// into the given database. Beside writing the transaction, the function also // a block, enabling hash based transaction and receipt lookups.
// stores a metadata entry along with the transaction, detailing the position func WriteTxLookupEntries(db ethdb.Database, block *types.Block) error {
// of this within the blockchain.
func WriteTransactions(db ethdb.Database, block *types.Block) error {
batch := db.NewBatch() batch := db.NewBatch()
// Iterate over each transaction and encode it with its metadata // Iterate over each transaction and encode its metadata
for i, tx := range block.Transactions() { for i, tx := range block.Transactions() {
// Encode and queue up the transaction for storage entry := txLookupEntry{
data, err := rlp.EncodeToBytes(tx)
if err != nil {
return err
}
if err := batch.Put(tx.Hash().Bytes(), data); err != nil {
return err
}
// Encode and queue up the transaction metadata for storage
meta := struct {
BlockHash common.Hash
BlockIndex uint64
Index uint64
}{
BlockHash: block.Hash(), BlockHash: block.Hash(),
BlockIndex: block.NumberU64(), BlockIndex: block.NumberU64(),
Index: uint64(i), Index: uint64(i),
} }
data, err = rlp.EncodeToBytes(meta) data, err := rlp.EncodeToBytes(entry)
if err != nil { if err != nil {
return err return err
} }
if err := batch.Put(append(tx.Hash().Bytes(), txMetaSuffix...), data); err != nil { if err := batch.Put(append(lookupPrefix, tx.Hash().Bytes()...), data); err != nil {
return err return err
} }
} }
// Write the scheduled data into the database // Write the scheduled data into the database
if err := batch.Write(); err != nil { if err := batch.Write(); err != nil {
glog.Fatalf("failed to store transactions into database: %v", err) glog.Fatalf("Failed to store lookup entries", "err", err)
}
return nil
}
// WriteReceipt stores a single transaction receipt into the database.
func WriteReceipt(db ethdb.Database, receipt *types.Receipt) error {
storageReceipt := (*types.ReceiptForStorage)(receipt)
data, err := rlp.EncodeToBytes(storageReceipt)
if err != nil {
return err
}
return db.Put(append(receiptsPrefix, receipt.TxHash.Bytes()...), data)
}
// WriteReceipts stores a batch of transaction receipts into the database.
func WriteReceipts(db ethdb.Database, receipts types.Receipts) error {
batch := db.NewBatch()
// Iterate over all the receipts and queue them for database injection
for _, receipt := range receipts {
storageReceipt := (*types.ReceiptForStorage)(receipt)
data, err := rlp.EncodeToBytes(storageReceipt)
if err != nil {
return err
}
if err := batch.Put(append(receiptsPrefix, receipt.TxHash.Bytes()...), data); err != nil {
return err
}
}
// Write the scheduled data into the database
if err := batch.Write(); err != nil {
glog.Fatalf("failed to store receipts into database: %v", err)
} }
return nil return nil
} }
@ -529,33 +486,9 @@ func DeleteBlockReceipts(db ethdb.Database, hash common.Hash, number uint64) {
db.Delete(append(append(blockReceiptsPrefix, encodeBlockNumber(number)...), hash.Bytes()...)) db.Delete(append(append(blockReceiptsPrefix, encodeBlockNumber(number)...), hash.Bytes()...))
} }
// DeleteTransaction removes all transaction data associated with a hash. // DeleteTxLookupEntry removes all transaction data associated with a hash.
func DeleteTransaction(db ethdb.Database, hash common.Hash) { func DeleteTxLookupEntry(db ethdb.Database, hash common.Hash) {
db.Delete(hash.Bytes()) db.Delete(append(lookupPrefix, hash.Bytes()...))
db.Delete(append(hash.Bytes(), txMetaSuffix...))
}
// DeleteReceipt removes all receipt data associated with a transaction hash.
func DeleteReceipt(db ethdb.Database, hash common.Hash) {
db.Delete(append(receiptsPrefix, hash.Bytes()...))
}
// [deprecated by the header/block split, remove eventually]
// 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
// access the old combined block representation. It will be dropped after the
// network transitions to eth/63.
func GetBlockByHashOld(db ethdb.Database, hash common.Hash) *types.Block {
data, _ := db.Get(append(oldBlockHashPrefix, hash[:]...))
if len(data) == 0 {
return nil
}
var block types.StorageBlock
if err := rlp.Decode(bytes.NewReader(data), &block); err != nil {
glog.V(logger.Error).Infof("invalid block RLP for hash %x: %v", hash, err)
return nil
}
return (*types.Block)(&block)
} }
// returns a formatted MIP mapped key by adding prefix, canonical number and level // returns a formatted MIP mapped key by adding prefix, canonical number and level

View file

@ -343,8 +343,8 @@ func TestHeadStorage(t *testing.T) {
} }
} }
// Tests that transactions and associated metadata can be stored and retrieved. // Tests that positional lookup metadata can be stored and retrieved.
func TestTransactionStorage(t *testing.T) { func TestLookupStorage(t *testing.T) {
db, _ := ethdb.NewMemDatabase() db, _ := ethdb.NewMemDatabase()
tx1 := types.NewTransaction(1, common.BytesToAddress([]byte{0x11}), big.NewInt(111), big.NewInt(1111), big.NewInt(11111), []byte{0x11, 0x11, 0x11}) tx1 := types.NewTransaction(1, common.BytesToAddress([]byte{0x11}), big.NewInt(111), big.NewInt(1111), big.NewInt(11111), []byte{0x11, 0x11, 0x11})
@ -361,7 +361,10 @@ func TestTransactionStorage(t *testing.T) {
} }
} }
// Insert all the transactions into the database, and verify contents // 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) t.Fatalf("failed to write transactions: %v", err)
} }
for i, tx := range txs { for i, tx := range txs {
@ -378,72 +381,13 @@ func TestTransactionStorage(t *testing.T) {
} }
// Delete the transactions and check purge // Delete the transactions and check purge
for i, tx := range txs { for i, tx := range txs {
DeleteTransaction(db, tx.Hash()) DeleteTxLookupEntry(db, tx.Hash())
if txn, _, _, _ := GetTransaction(db, tx.Hash()); txn != nil { if txn, _, _, _ := GetTransaction(db, tx.Hash()); txn != nil {
t.Fatalf("tx #%d [%x]: deleted transaction returned: %v", i, tx.Hash(), txn) 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. // Tests that receipts associated with a single block can be stored and retrieved.
func TestBlockReceiptStorage(t *testing.T) { func TestBlockReceiptStorage(t *testing.T) {
db, _ := ethdb.NewMemDatabase() db, _ := ethdb.NewMemDatabase()
@ -579,10 +523,6 @@ func TestMipmapChain(t *testing.T) {
} }
// store the receipts // store the receipts
err := WriteReceipts(db, receipts)
if err != nil {
t.Fatal(err)
}
WriteMipmapBloom(db, uint64(i+1), receipts) WriteMipmapBloom(db, uint64(i+1), receipts)
}) })
for i, block := range chain { for i, block := range chain {

View file

@ -114,7 +114,7 @@ type Ethereum struct {
chainConfig *params.ChainConfig chainConfig *params.ChainConfig
// Channel for shutting down the service // Channel for shutting down the service
shutdownChan chan bool // Channel for shutting down the ethereum shutdownChan chan bool // Channel for shutting down the ethereum
stopDbUpgrade func() // stop chain db sequential key upgrade stopDbUpgrade func() error // stop chain db sequential key upgrade
// Handlers // Handlers
txPool *core.TxPool txPool *core.TxPool
txMu sync.Mutex txMu sync.Mutex
@ -154,7 +154,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
if err != nil { if err != nil {
return nil, err return nil, err
} }
stopDbUpgrade := upgradeSequentialKeys(chainDb) stopDbUpgrade := upgradeDeduplicateData(chainDb)
if err := SetupGenesisBlock(&chainDb, config); err != nil { if err := SetupGenesisBlock(&chainDb, config); err != nil {
return nil, err return nil, err
} }
@ -177,9 +177,6 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
solcPath: config.SolcPath, solcPath: config.SolcPath,
} }
if err := upgradeChainDatabase(chainDb); err != nil {
return nil, err
}
if err := addMipmapBloomBins(chainDb); err != nil { if err := addMipmapBloomBins(chainDb); err != nil {
return nil, err return nil, err
} }

View file

@ -33,24 +33,15 @@ func TestMipmapUpgrade(t *testing.T) {
genesis := core.WriteGenesisBlockForTesting(db) genesis := core.WriteGenesisBlockForTesting(db)
chain, receipts := core.GenerateChain(params.TestChainConfig, genesis, db, 10, func(i int, gen *core.BlockGen) { chain, receipts := core.GenerateChain(params.TestChainConfig, genesis, db, 10, func(i int, gen *core.BlockGen) {
var receipts types.Receipts
switch i { switch i {
case 1: case 1:
receipt := types.NewReceipt(nil, new(big.Int)) receipt := types.NewReceipt(nil, new(big.Int))
receipt.Logs = []*types.Log{{Address: addr}} receipt.Logs = []*types.Log{{Address: addr}}
gen.AddUncheckedReceipt(receipt) gen.AddUncheckedReceipt(receipt)
receipts = types.Receipts{receipt}
case 2: case 2:
receipt := types.NewReceipt(nil, new(big.Int)) receipt := types.NewReceipt(nil, new(big.Int))
receipt.Logs = []*types.Log{{Address: addr}} receipt.Logs = []*types.Log{{Address: addr}}
gen.AddUncheckedReceipt(receipt) 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 { for i, block := range chain {

View file

@ -19,297 +19,121 @@ package eth
import ( import (
"bytes" "bytes"
"encoding/binary"
"fmt" "fmt"
"math/big"
"time" "time"
"github.com/ubiq/go-ubiq/common" "github.com/ubiq/go-ubiq/common"
"github.com/ubiq/go-ubiq/core" "github.com/ubiq/go-ubiq/core"
"github.com/ubiq/go-ubiq/core/types"
"github.com/ubiq/go-ubiq/ethdb" "github.com/ubiq/go-ubiq/ethdb"
"github.com/ubiq/go-ubiq/logger" "github.com/ubiq/go-ubiq/logger"
"github.com/ubiq/go-ubiq/logger/glog" "github.com/ubiq/go-ubiq/logger/glog"
"github.com/ubiq/go-ubiq/rlp" "github.com/ubiq/go-ubiq/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. // starts a background process to make upgrades if necessary.
// Returns a stop function that blocks until the process has // Returns a stop function that blocks until the process has
// been safely stopped. // been safely stopped.
func upgradeSequentialKeys(db ethdb.Database) (stopFn func()) { func upgradeDeduplicateData(db ethdb.Database) func() error {
data, _ := db.Get(useSequentialKeys) // If the database is already converted or empty, bail out
data, _ := db.Get(deduplicateData)
if len(data) > 0 && data[0] == 42 { if len(data) > 0 && data[0] == 42 {
return nil // already converted return nil
} }
if data, _ := db.Get([]byte("LastHeader")); len(data) == 0 { if data, _ := db.Get([]byte("LastHeader")); len(data) == 0 {
db.Put(useSequentialKeys, []byte{42}) db.Put(deduplicateData, []byte{42})
return nil // empty database, nothing to do return nil
} }
// Start the deduplication upgrade on a new goroutine
glog.V(logger.Info).Infof("Upgrading chain database to use sequential keys") glog.V(logger.Info).Infof("Upgrading database to use lookup entries")
stop := make(chan chan error)
stopChn := make(chan struct{})
stoppedChn := make(chan struct{})
go func() { go func() {
stopFn := func() bool { // 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()
}
}()
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
}
// 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 err := rlp.DecodeBytes(it.Value(), &meta); err != nil {
continue
}
// 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
}
}
// 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
}
}
}
}
// 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)
glog.V(logger.Info).Infof("Deduplicating database entries. deduped: %d", converted)
}
// Check for termination, or continue after a bit of a timeout
select { select {
case <-time.After(time.Microsecond * 100): // make sure other processes don't get starved case errc := <-stop:
case <-stopChn: errc <- nil
return true return
} case <-time.After(time.Microsecond * 100):
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 {
glog.V(logger.Info).Infof("Database conversion successful")
db.Put(useSequentialKeys, []byte{42})
}
if err != nil {
glog.V(logger.Error).Infof("Database conversion failed: %v", 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 {
it.Release()
it = db.(*ethdb.LDBDatabase).NewIterator()
it.Seek(keyPtr)
glog.V(logger.Info).Infof("converting %d canonical numbers...", 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
}
if err := db.Delete(keyPtr); err != nil {
return err, false
} }
} }
// Upgrade finished, mark a such and terminate
if stopFn() { if failed == nil {
return nil, true glog.V(logger.Info).Infof("Database deduplication successful. deduped: %d", converted)
} db.Put(deduplicateData, []byte{42})
it.Next()
}
if cnt > 0 {
glog.V(logger.Info).Infof("converted %d canonical numbers...", 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)
glog.V(logger.Info).Infof("converting %d blocks...", cnt)
}
// 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
}
it.Next()
}
if err := db.Delete(append([]byte("receipts-block-"), hash...)); err != nil {
return err, false
}
} else { } else {
it.Next() glog.V(logger.Error).Infof("Database deduplication failed. deduped: %d", converted, "err", failed)
} }
it.Release()
it = nil
if stopFn() { errc := <-stop
return nil, true errc <- failed
}()
// Assembly the cancellation callback
return func() error {
errc := make(chan error)
stop <- errc
return <-errc
} }
}
if cnt > 0 {
glog.V(logger.Info).Infof("converted %d blocks...", cnt)
}
return nil, false
}
// 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
}
if stopFn() {
return nil, true
}
it.Next()
}
if cnt > 0 {
glog.V(logger.Info).Infof("removed %d orphaned block receipts...", cnt)
}
return nil, false
}
// 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
}
}
return nil
}
// upgradeChainDatabase ensures that the chain database stores block split into
// separate header and body entries.
func upgradeChainDatabase(db ethdb.Database) error {
// Short circuit if the head block is stored already as separate header and body
data, err := db.Get([]byte("LastBlock"))
if err != nil {
return nil
}
head := common.BytesToHash(data)
if block := core.GetBlockByHashOld(db, head); block == nil {
return nil
}
// At least some of the database is still the old format, upgrade (skip the head block!)
glog.V(logger.Info).Info("Old database detected, upgrading...")
if db, ok := db.(*ethdb.LDBDatabase); ok {
blockPrefix := []byte("block-hash-")
for it := db.NewIterator(); it.Next(); {
// Skip anything other than a combined block
if !bytes.HasPrefix(it.Key(), blockPrefix) {
continue
}
// Skip the head block (merge last to signal upgrade completion)
if bytes.HasSuffix(it.Key(), head.Bytes()) {
continue
}
// Load the block, split and serialize (order!)
block := core.GetBlockByHashOld(db, common.BytesToHash(bytes.TrimPrefix(it.Key(), blockPrefix)))
if err := core.WriteTd(db, block.Hash(), block.NumberU64(), block.DeprecatedTd()); err != nil {
return err
}
if err := core.WriteBody(db, block.Hash(), block.NumberU64(), block.Body()); err != nil {
return err
}
if err := core.WriteHeader(db, block.Header()); err != nil {
return err
}
if err := db.Delete(it.Key()); err != nil {
return err
}
}
// Lastly, upgrade the head block, disabling the upgrade mechanism
current := core.GetBlockByHashOld(db, head)
if err := core.WriteTd(db, current.Hash(), current.NumberU64(), current.DeprecatedTd()); err != nil {
return err
}
if err := core.WriteBody(db, current.Hash(), current.NumberU64(), current.Body()); err != nil {
return err
}
if err := core.WriteHeader(db, current.Header()); err != nil {
return err
}
}
return nil
} }
func addMipmapBloomBins(db ethdb.Database) (err error) { func addMipmapBloomBins(db ethdb.Database) (err error) {

View file

@ -83,12 +83,6 @@ func BenchmarkMipmaps(b *testing.B) {
gen.AddUncheckedReceipt(receipt) gen.AddUncheckedReceipt(receipt)
} }
// store the receipts
err := core.WriteReceipts(db, receipts)
if err != nil {
b.Fatal(err)
}
core.WriteMipmapBloom(db, uint64(i+1), receipts) core.WriteMipmapBloom(db, uint64(i+1), receipts)
}) })
for i, block := range chain { for i, block := range chain {
@ -184,12 +178,6 @@ func TestFilters(t *testing.T) {
gen.AddUncheckedReceipt(receipt) gen.AddUncheckedReceipt(receipt)
receipts = types.Receipts{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 // i is used as block number for the writes but since the i
// starts at 0 and block 0 (genesis) is already present increment // starts at 0 and block 0 (genesis) is already present increment
// by one // by one

View file

@ -17,7 +17,6 @@
package ethapi package ethapi
import ( import (
"bytes"
"encoding/hex" "encoding/hex"
"errors" "errors"
"fmt" "fmt"
@ -37,7 +36,6 @@ import (
"github.com/ubiq/go-ubiq/core/types" "github.com/ubiq/go-ubiq/core/types"
"github.com/ubiq/go-ubiq/core/vm" "github.com/ubiq/go-ubiq/core/vm"
"github.com/ubiq/go-ubiq/crypto" "github.com/ubiq/go-ubiq/crypto"
"github.com/ubiq/go-ubiq/ethdb"
"github.com/ubiq/go-ubiq/logger" "github.com/ubiq/go-ubiq/logger"
"github.com/ubiq/go-ubiq/logger/glog" "github.com/ubiq/go-ubiq/logger/glog"
"github.com/ubiq/go-ubiq/p2p" "github.com/ubiq/go-ubiq/p2p"
@ -743,7 +741,7 @@ func (s *PublicBlockChainAPI) rpcOutputBlock(b *types.Block, inclTx bool, fullTx
if fullTx { if fullTx {
formatTx = func(tx *types.Transaction) (interface{}, error) { formatTx = func(tx *types.Transaction) (interface{}, error) {
return newRPCTransaction(b, tx.Hash()) return newRPCTransactionFromBlockHash(b, tx.Hash()), nil
} }
} }
@ -786,15 +784,17 @@ type RPCTransaction struct {
S *hexutil.Big `json:"s"` S *hexutil.Big `json:"s"`
} }
// newRPCPendingTransaction returns a pending transaction that will serialize to the RPC representation // newRPCTransaction returns a transaction that will serialize to the RPC
func newRPCPendingTransaction(tx *types.Transaction) *RPCTransaction { // representation, with the given location metadata set (if available).
func newRPCTransaction(tx *types.Transaction, blockHash common.Hash, blockNumber uint64, index uint64) *RPCTransaction {
var signer types.Signer = types.FrontierSigner{} var signer types.Signer = types.FrontierSigner{}
if tx.Protected() { if tx.Protected() {
signer = types.NewEIP155Signer(tx.ChainId()) signer = types.NewEIP155Signer(tx.ChainId())
} }
from, _ := types.Sender(signer, tx) from, _ := types.Sender(signer, tx)
v, r, s := tx.RawSignatureValues() v, r, s := tx.RawSignatureValues()
return &RPCTransaction{
result := &RPCTransaction{
From: from, From: from,
Gas: (*hexutil.Big)(tx.Gas()), Gas: (*hexutil.Big)(tx.Gas()),
GasPrice: (*hexutil.Big)(tx.GasPrice()), GasPrice: (*hexutil.Big)(tx.GasPrice()),
@ -807,58 +807,46 @@ func newRPCPendingTransaction(tx *types.Transaction) *RPCTransaction {
R: (*hexutil.Big)(r), R: (*hexutil.Big)(r),
S: (*hexutil.Big)(s), S: (*hexutil.Big)(s),
} }
if blockHash != (common.Hash{}) {
result.BlockHash = blockHash
result.BlockNumber = (*hexutil.Big)(new(big.Int).SetUint64(blockNumber))
result.TransactionIndex = hexutil.Uint(index)
}
return result
} }
// newRPCTransaction returns a transaction that will serialize to the RPC representation. // newRPCPendingTransaction returns a pending transaction that will serialize to the RPC representation
func newRPCTransactionFromBlockIndex(b *types.Block, txIndex uint) (*RPCTransaction, error) { func newRPCPendingTransaction(tx *types.Transaction) *RPCTransaction {
if txIndex < uint(len(b.Transactions())) { return newRPCTransaction(tx, common.Hash{}, 0, 0)
tx := b.Transactions()[txIndex] }
var signer types.Signer = types.FrontierSigner{}
if tx.Protected() {
signer = types.NewEIP155Signer(tx.ChainId())
}
from, _ := types.Sender(signer, tx)
v, r, s := tx.RawSignatureValues()
return &RPCTransaction{
BlockHash: b.Hash(),
BlockNumber: (*hexutil.Big)(b.Number()),
From: from,
Gas: (*hexutil.Big)(tx.Gas()),
GasPrice: (*hexutil.Big)(tx.GasPrice()),
Hash: tx.Hash(),
Input: hexutil.Bytes(tx.Data()),
Nonce: hexutil.Uint64(tx.Nonce()),
To: tx.To(),
TransactionIndex: hexutil.Uint(txIndex),
Value: (*hexutil.Big)(tx.Value()),
V: (*hexutil.Big)(v),
R: (*hexutil.Big)(r),
S: (*hexutil.Big)(s),
}, nil
}
return nil, nil // newRPCTransactionFromBlockIndex returns a transaction that will serialize to the RPC representation.
func newRPCTransactionFromBlockIndex(b *types.Block, index uint64) *RPCTransaction {
txs := b.Transactions()
if index >= uint64(len(txs)) {
return nil
}
return newRPCTransaction(txs[index], b.Hash(), b.NumberU64(), index)
} }
// newRPCRawTransactionFromBlockIndex returns the bytes of a transaction given a block and a transaction index. // newRPCRawTransactionFromBlockIndex returns the bytes of a transaction given a block and a transaction index.
func newRPCRawTransactionFromBlockIndex(b *types.Block, txIndex uint) (hexutil.Bytes, error) { func newRPCRawTransactionFromBlockIndex(b *types.Block, index uint64) hexutil.Bytes {
if txIndex < uint(len(b.Transactions())) { txs := b.Transactions()
tx := b.Transactions()[txIndex] if index >= uint64(len(txs)) {
return rlp.EncodeToBytes(tx) return nil
} }
blob, _ := rlp.EncodeToBytes(txs[index])
return nil, nil return blob
} }
// newRPCTransaction returns a transaction that will serialize to the RPC representation. // newRPCTransactionFromBlockHash returns a transaction that will serialize to the RPC representation.
func newRPCTransaction(b *types.Block, txHash common.Hash) (*RPCTransaction, error) { func newRPCTransactionFromBlockHash(b *types.Block, hash common.Hash) *RPCTransaction {
for idx, tx := range b.Transactions() { for idx, tx := range b.Transactions() {
if tx.Hash() == txHash { if tx.Hash() == hash {
return newRPCTransactionFromBlockIndex(b, uint(idx)) return newRPCTransactionFromBlockIndex(b, uint64(idx))
} }
} }
return nil
return nil, nil
} }
// PublicTransactionPoolAPI exposes methods for the RPC interface // PublicTransactionPoolAPI exposes methods for the RPC interface
@ -871,24 +859,6 @@ func NewPublicTransactionPoolAPI(b Backend) *PublicTransactionPoolAPI {
return &PublicTransactionPoolAPI{b} return &PublicTransactionPoolAPI{b}
} }
func getTransaction(chainDb ethdb.Database, b Backend, txHash common.Hash) (*types.Transaction, bool, error) {
txData, err := chainDb.Get(txHash.Bytes())
isPending := false
tx := new(types.Transaction)
if err == nil && len(txData) > 0 {
if err := rlp.DecodeBytes(txData, tx); err != nil {
return nil, isPending, err
}
} else {
// pending transaction?
tx = b.GetPoolTransaction(txHash)
isPending = true
}
return tx, isPending, nil
}
// GetBlockTransactionCountByNumber returns the number of transactions in the block with the given block number. // GetBlockTransactionCountByNumber returns the number of transactions in the block with the given block number.
func (s *PublicTransactionPoolAPI) GetBlockTransactionCountByNumber(ctx context.Context, blockNr rpc.BlockNumber) *hexutil.Uint { func (s *PublicTransactionPoolAPI) GetBlockTransactionCountByNumber(ctx context.Context, blockNr rpc.BlockNumber) *hexutil.Uint {
if block, _ := s.b.BlockByNumber(ctx, blockNr); block != nil { if block, _ := s.b.BlockByNumber(ctx, blockNr); block != nil {
@ -908,35 +878,35 @@ func (s *PublicTransactionPoolAPI) GetBlockTransactionCountByHash(ctx context.Co
} }
// GetTransactionByBlockNumberAndIndex returns the transaction for the given block number and index. // GetTransactionByBlockNumberAndIndex returns the transaction for the given block number and index.
func (s *PublicTransactionPoolAPI) GetTransactionByBlockNumberAndIndex(ctx context.Context, blockNr rpc.BlockNumber, index hexutil.Uint) (*RPCTransaction, error) { func (s *PublicTransactionPoolAPI) GetTransactionByBlockNumberAndIndex(ctx context.Context, blockNr rpc.BlockNumber, index hexutil.Uint) *RPCTransaction {
if block, _ := s.b.BlockByNumber(ctx, blockNr); block != nil { if block, _ := s.b.BlockByNumber(ctx, blockNr); block != nil {
return newRPCTransactionFromBlockIndex(block, uint(index)) return newRPCTransactionFromBlockIndex(block, uint64(index))
} }
return nil, nil return nil
} }
// GetTransactionByBlockHashAndIndex returns the transaction for the given block hash and index. // GetTransactionByBlockHashAndIndex returns the transaction for the given block hash and index.
func (s *PublicTransactionPoolAPI) GetTransactionByBlockHashAndIndex(ctx context.Context, blockHash common.Hash, index hexutil.Uint) (*RPCTransaction, error) { func (s *PublicTransactionPoolAPI) GetTransactionByBlockHashAndIndex(ctx context.Context, blockHash common.Hash, index hexutil.Uint) *RPCTransaction {
if block, _ := s.b.GetBlock(ctx, blockHash); block != nil { if block, _ := s.b.GetBlock(ctx, blockHash); block != nil {
return newRPCTransactionFromBlockIndex(block, uint(index)) return newRPCTransactionFromBlockIndex(block, uint64(index))
} }
return nil, nil return nil
} }
// GetRawTransactionByBlockNumberAndIndex returns the bytes of the transaction for the given block number and index. // GetRawTransactionByBlockNumberAndIndex returns the bytes of the transaction for the given block number and index.
func (s *PublicTransactionPoolAPI) GetRawTransactionByBlockNumberAndIndex(ctx context.Context, blockNr rpc.BlockNumber, index hexutil.Uint) (hexutil.Bytes, error) { func (s *PublicTransactionPoolAPI) GetRawTransactionByBlockNumberAndIndex(ctx context.Context, blockNr rpc.BlockNumber, index hexutil.Uint) hexutil.Bytes {
if block, _ := s.b.BlockByNumber(ctx, blockNr); block != nil { if block, _ := s.b.BlockByNumber(ctx, blockNr); block != nil {
return newRPCRawTransactionFromBlockIndex(block, uint(index)) return newRPCRawTransactionFromBlockIndex(block, uint64(index))
} }
return nil, nil return nil
} }
// GetRawTransactionByBlockHashAndIndex returns the bytes of the transaction for the given block hash and index. // GetRawTransactionByBlockHashAndIndex returns the bytes of the transaction for the given block hash and index.
func (s *PublicTransactionPoolAPI) GetRawTransactionByBlockHashAndIndex(ctx context.Context, blockHash common.Hash, index hexutil.Uint) (hexutil.Bytes, error) { func (s *PublicTransactionPoolAPI) GetRawTransactionByBlockHashAndIndex(ctx context.Context, blockHash common.Hash, index hexutil.Uint) hexutil.Bytes {
if block, _ := s.b.GetBlock(ctx, blockHash); block != nil { if block, _ := s.b.GetBlock(ctx, blockHash); block != nil {
return newRPCRawTransactionFromBlockIndex(block, uint(index)) return newRPCRawTransactionFromBlockIndex(block, uint64(index))
} }
return nil, nil return nil
} }
// GetTransactionCount returns the number of transactions the given address has sent for the given block number // GetTransactionCount returns the number of transactions the given address has sent for the given block number
@ -952,92 +922,42 @@ func (s *PublicTransactionPoolAPI) GetTransactionCount(ctx context.Context, addr
return (*hexutil.Uint64)(&nonce), nil return (*hexutil.Uint64)(&nonce), nil
} }
// getTransactionBlockData fetches the meta data for the given transaction from the chain database. This is useful to
// retrieve block information for a hash. It returns the block hash, block index and transaction index.
func getTransactionBlockData(chainDb ethdb.Database, txHash common.Hash) (common.Hash, uint64, uint64, error) {
var txBlock struct {
BlockHash common.Hash
BlockIndex uint64
Index uint64
}
blockData, err := chainDb.Get(append(txHash.Bytes(), 0x0001))
if err != nil {
return common.Hash{}, uint64(0), uint64(0), err
}
reader := bytes.NewReader(blockData)
if err = rlp.Decode(reader, &txBlock); err != nil {
return common.Hash{}, uint64(0), uint64(0), err
}
return txBlock.BlockHash, txBlock.BlockIndex, txBlock.Index, nil
}
// GetTransactionByHash returns the transaction for the given hash // GetTransactionByHash returns the transaction for the given hash
func (s *PublicTransactionPoolAPI) GetTransactionByHash(ctx context.Context, txHash common.Hash) (*RPCTransaction, error) { func (s *PublicTransactionPoolAPI) GetTransactionByHash(ctx context.Context, hash common.Hash) *RPCTransaction {
var tx *types.Transaction // Try to return an already finalized transaction
var isPending bool if tx, blockHash, blockNumber, index := core.GetTransaction(s.b.ChainDb(), hash); tx != nil {
var err error return newRPCTransaction(tx, blockHash, blockNumber, index)
if tx, isPending, err = getTransaction(s.b.ChainDb(), s.b, txHash); err != nil {
glog.V(logger.Debug).Infof("%v\n", err)
return nil, nil
} else if tx == nil {
return nil, nil
} }
// No finalized transaction, try to retrieve it from the pool
if isPending { if tx := s.b.GetPoolTransaction(hash); tx != nil {
return newRPCPendingTransaction(tx), nil return newRPCPendingTransaction(tx)
} }
// Transaction unknown, return as such
blockHash, _, _, err := getTransactionBlockData(s.b.ChainDb(), txHash) return nil
if err != nil {
glog.V(logger.Debug).Infof("%v\n", err)
return nil, nil
}
if block, _ := s.b.GetBlock(ctx, blockHash); block != nil {
return newRPCTransaction(block, txHash)
}
return nil, nil
} }
// GetRawTransactionByHash returns the bytes of the transaction for the given hash. // GetRawTransactionByHash returns the bytes of the transaction for the given hash.
func (s *PublicTransactionPoolAPI) GetRawTransactionByHash(ctx context.Context, txHash common.Hash) (hexutil.Bytes, error) { func (s *PublicTransactionPoolAPI) GetRawTransactionByHash(ctx context.Context, hash common.Hash) (hexutil.Bytes, error) {
var tx *types.Transaction var tx *types.Transaction
var err error
if tx, _, err = getTransaction(s.b.ChainDb(), s.b, txHash); err != nil { // Retrieve a finalized transaction, or a pooled otherwise
glog.V(logger.Debug).Infof("%v\n", err) if tx, _, _, _ = core.GetTransaction(s.b.ChainDb(), hash); tx == nil {
return nil, nil if tx = s.b.GetPoolTransaction(hash); tx == nil {
} else if tx == nil { // Transaction not found anywhere, abort
return nil, nil return nil, nil
} }
}
// Serialize to RLP and return
return rlp.EncodeToBytes(tx) return rlp.EncodeToBytes(tx)
} }
// GetTransactionReceipt returns the transaction receipt for the given transaction hash. // GetTransactionReceipt returns the transaction receipt for the given transaction hash.
func (s *PublicTransactionPoolAPI) GetTransactionReceipt(txHash common.Hash) (map[string]interface{}, error) { func (s *PublicTransactionPoolAPI) GetTransactionReceipt(hash common.Hash) (map[string]interface{}, error) {
receipt := core.GetReceipt(s.b.ChainDb(), txHash) tx, blockHash, blockNumber, index := core.GetTransaction(s.b.ChainDb(), hash)
if receipt == nil { if tx == nil {
glog.V(logger.Debug).Infof("receipt not found for transaction %s", txHash.Hex())
return nil, nil
}
tx, _, err := getTransaction(s.b.ChainDb(), s.b, txHash)
if err != nil {
glog.V(logger.Debug).Infof("%v\n", err)
return nil, nil
}
txBlock, blockIndex, index, err := getTransactionBlockData(s.b.ChainDb(), txHash)
if err != nil {
glog.V(logger.Debug).Infof("%v\n", err)
return nil, nil return nil, nil
} }
receipt, _, _, _ := core.GetReceipt(s.b.ChainDb(), hash) // Old receipts don't have the lookup data available
var signer types.Signer = types.FrontierSigner{} var signer types.Signer = types.FrontierSigner{}
if tx.Protected() { if tx.Protected() {
@ -1047,9 +967,9 @@ func (s *PublicTransactionPoolAPI) GetTransactionReceipt(txHash common.Hash) (ma
fields := map[string]interface{}{ fields := map[string]interface{}{
"root": hexutil.Bytes(receipt.PostState), "root": hexutil.Bytes(receipt.PostState),
"blockHash": txBlock, "blockHash": blockHash,
"blockNumber": hexutil.Uint64(blockIndex), "blockNumber": hexutil.Uint64(blockNumber),
"transactionHash": txHash, "transactionHash": hash,
"transactionIndex": hexutil.Uint64(index), "transactionIndex": hexutil.Uint64(index),
"from": from, "from": from,
"to": tx.To(), "to": tx.To(),

View file

@ -129,19 +129,6 @@ type txBlockData struct {
Index uint64 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 // txStateChanges stores the recent changes between pending/mined states of
// transactions. True means mined, false means rolled back, no entry means no change // transactions. True means mined, false means rolled back, no entry means no change
type txStateChanges map[common.Hash]bool type txStateChanges map[common.Hash]bool
@ -171,59 +158,48 @@ func (txc txStateChanges) getLists() (mined []common.Hash, rollback []common.Has
// checkMinedTxs checks newly added blocks for the currently pending transactions // 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 marks them as mined if necessary. It also stores block position in the db
// and adds them to the received txStateChanges map. // and adds them to the received txStateChanges map.
func (pool *TxPool) checkMinedTxs(ctx context.Context, hash common.Hash, idx uint64, txc txStateChanges) error { func (pool *TxPool) checkMinedTxs(ctx context.Context, hash common.Hash, number uint64, txc txStateChanges) error {
//fmt.Println("checkMinedTxs") // If no transactions are pending, we don't care about anything
if len(pool.pending) == 0 { if len(pool.pending) == 0 {
return nil return nil
} }
//fmt.Println("len(pool) =", len(pool.pending)) block, err := GetBlock(ctx, pool.odr, hash, number)
block, err := GetBlock(ctx, pool.odr, hash, idx)
var receipts types.Receipts
if err != nil { if err != nil {
//fmt.Println(err)
return err return err
} }
//fmt.Println("len(block.Transactions()) =", len(block.Transactions())) // Gather all the local transaction mined in this block
list := pool.mined[hash] list := pool.mined[hash]
for i, tx := range block.Transactions() { for _, tx := range block.Transactions() {
txHash := tx.Hash() if _, ok := pool.pending[tx.Hash()]; ok {
//fmt.Println(" txHash:", txHash) list = append(list, tx)
if tx, ok := pool.pending[txHash]; ok { }
//fmt.Println("TX FOUND") }
if receipts == nil { // If some transactions have been mined, write the needed data to disk and update
receipts, err = GetBlockReceipts(ctx, pool.odr, hash, idx) if list != nil {
if err != 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 return err
} }
if len(receipts) != len(block.Transactions()) { if err := core.WriteTxLookupEntries(pool.chainDb, block); err != nil {
panic(nil) // should never happen if hashes did match return err
} }
core.SetReceiptsData(pool.config, block, receipts) // Update the transaction pool's state
for _, tx := range list {
delete(pool.pending, tx.Hash())
txc.setState(tx.Hash(), true)
} }
//fmt.Println("WriteReceipt", receipts[i].TxHash)
core.WriteReceipt(pool.chainDb, receipts[i])
pool.storeTxBlockData(txHash, txBlockData{hash, idx, uint64(i)})
delete(pool.pending, txHash)
list = append(list, tx)
txc.setState(txHash, true)
}
}
if list != nil {
pool.mined[hash] = list pool.mined[hash] = list
} }
return nil return nil
} }
// rollbackTxs marks the transactions contained in recently rolled back blocks // 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 // as rolled back. It also removes any positional lookup entries.
// to the received txStateChanges map.
func (pool *TxPool) rollbackTxs(hash common.Hash, txc txStateChanges) { func (pool *TxPool) rollbackTxs(hash common.Hash, txc txStateChanges) {
if list, ok := pool.mined[hash]; ok { if list, ok := pool.mined[hash]; ok {
for _, tx := range list { for _, tx := range list {
txHash := tx.Hash() txHash := tx.Hash()
pool.removeTxBlockData(txHash) core.DeleteTxLookupEntry(pool.chainDb, txHash)
pool.pending[txHash] = tx pool.pending[txHash] = tx
txc.setState(txHash, false) txc.setState(txHash, false)
} }

View file

@ -314,9 +314,7 @@ func (self *worker) wait() {
// check if canon block and write transactions // check if canon block and write transactions
if stat == core.CanonStatTy { if stat == core.CanonStatTy {
// This puts transactions in a extra db for rpc // This puts transactions in a extra db for rpc
core.WriteTransactions(self.chainDb, block) core.WriteTxLookupEntries(self.chainDb, block)
// store the receipts
core.WriteReceipts(self.chainDb, work.receipts)
// Write map map bloom filters // Write map map bloom filters
core.WriteMipmapBloom(self.chainDb, block.NumberU64(), work.receipts) core.WriteMipmapBloom(self.chainDb, block.NumberU64(), work.receipts)
// implicit by posting ChainHeadEvent // implicit by posting ChainHeadEvent