core/rawdb: new implementation of tx indexing/unindex using generic tx iterator and hashing rlp-data

core/rawdb/accessors_chain.go

@@ -327,6 +327,25 @@ func ReadBodyRLP(db ethdb.Reader, hash common.Hash, number uint64) rlp.RawValue
 	return nil // Can't find the data anywhere.
 }
 
+// ReadCanonicalBodyRLP retrieves the block body (transactions and uncles) for the canonical
+// bloc at number, in RLP encoding.
+func ReadCanonicalBodyRLP(db ethdb.Reader, number uint64) rlp.RawValue {
+	// if it's an ancient one, we don't need the canonical hash
+	data, _ := db.Ancient(freezerBodiesTable, number)
+	if len(data) == 0 {
+		// Need to get the hash
+		data, _ = db.Get(blockBodyKey(number, ReadCanonicalHash(db, number)))
+		// In the background freezer is moving data from leveldb to flatten files.
+		// So during the first check for ancient db, the data is not yet in there,
+		// but when we reach into leveldb, the data was already moved. That would
+		// result in a not found error.
+		if len(data) == 0 {
+			data, _ = db.Ancient(freezerBodiesTable, number)
+		}
+	}
+	return data
+}
+
 // WriteBodyRLP stores an RLP encoded block body into the database.
 func WriteBodyRLP(db ethdb.KeyValueWriter, hash common.Hash, number uint64, rlp rlp.RawValue) {
 	if err := db.Put(blockBodyKey(number, hash), rlp); err != nil {

core/rawdb/accessors_indexes.go

@@ -63,6 +63,16 @@ func WriteTxLookupEntries(db ethdb.KeyValueWriter, block *types.Block) {
 	}
 }
 
+// WriteTxLookupEntriesByHash is identical to WriteTxLookupEntries, but does not
+// require a full types.Block as input
+func WriteTxLookupEntriesByHash(db ethdb.KeyValueWriter, number []byte, hashes []common.Hash) {
+	for _, hash := range hashes {
+		if err := db.Put(txLookupKey(hash), number); err != nil {
+			log.Crit("Failed to store transaction lookup entry", "err", err)
+		}
+	}
+}
+
 // DeleteTxLookupEntry removes all transaction data associated with a hash.
 func DeleteTxLookupEntry(db ethdb.KeyValueWriter, hash common.Hash) {
 	if err := db.Delete(txLookupKey(hash)); err != nil {
@@ -71,9 +81,9 @@ func DeleteTxLookupEntry(db ethdb.KeyValueWriter, hash common.Hash) {
 }
 
 // DeleteTxLookupEntries removes all transaction lookups for a given block.
-func DeleteTxLookupEntries(db ethdb.KeyValueWriter, block *types.Block) {
-	for _, tx := range block.Transactions() {
-		if err := db.Delete(txLookupKey(tx.Hash())); err != nil {
+func DeleteTxLookupEntriesByHash(db ethdb.KeyValueWriter, hashes []common.Hash) {
+	for _, hash := range hashes {
+		if err := db.Delete(txLookupKey(hash)); err != nil {
 			log.Crit("Failed to delete transaction lookup entry", "err", err)
 		}
 	}

core/rawdb/chain_iterator.go

@@ -17,146 +17,20 @@
 package rawdb
 
 import (
-	"errors"
 	"math"
+	"math/big"
 	"runtime"
 	"sync/atomic"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/prque"
-	"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"
+	"golang.org/x/crypto/sha3"
 )
 
-type (
-	prepareCallback func(*types.Block)              // Callback for custom concurrent pre-computations on block data
-	actionCallback  func(ethdb.Batch, *types.Block) // Callback for custom sequential operations on block data.
-)
-
-// iterateCanonicalChain iterates the specified range of canonical blocks and applies
-// the given action callback. Note, both for forward and backward iteration, the range
-// is [from, to).
-func iterateCanonicalChain(db ethdb.Database, from uint64, to uint64, prepare prepareCallback, action actionCallback, reverse bool, progMsg, doneMsg string) error {
-	// Short circuit if the action is nil or if the range is invalid
-	if action == nil {
-		return nil
-	}
-	if from >= to {
-		return nil
-	}
-	// Spawn multi-routines, iterate over the specified blocks and invoke prepare
-	// callback concurrently.
-	var (
-		number  int64
-		results = make(chan *types.Block, 4*runtime.NumCPU())
-	)
-	if !reverse {
-		number = int64(from - 1)
-	} else {
-		number = int64(to)
-	}
-	abort := make(chan struct{})
-	defer close(abort)
-
-	threads := to - from
-	if cpus := runtime.NumCPU(); threads > uint64(cpus) {
-		threads = uint64(cpus)
-	}
-	for i := 0; i < int(threads); i++ {
-		go func() {
-			for {
-				// Fetch the next task number, terminating if everything's done
-				var n int64
-				if !reverse {
-					n = atomic.AddInt64(&number, 1)
-					if n >= int64(to) {
-						return
-					}
-				} else {
-					n = atomic.AddInt64(&number, -1)
-					if n < int64(from) {
-						return
-					}
-				}
-				block := ReadBlock(db, ReadCanonicalHash(db, uint64(n)), uint64(n))
-				if prepare != nil && block != nil {
-					prepare(block)
-				}
-				// Feed the block to the aggregator, or abort on interrupt
-				select {
-				case results <- block:
-				case <-abort:
-					return
-				}
-			}
-		}()
-	}
-	// Reassemble the blocks into a contiguous stream and apply the action callback
-	var (
-		next, first, last int64
-		queue             = prque.New(nil)
-
-		batch  = db.NewBatch()
-		start  = time.Now()
-		logged = start.Add(-7 * time.Second) // Unindex during import is fast, don't double log
-	)
-	if !reverse {
-		next, first, last = int64(from), int64(from), int64(to)
-	} else {
-		next, first, last = int64(to-1), int64(to-1), int64(from-1)
-	}
-	for i := from; i < to; i++ {
-		// Retrieve the next result and bail if it's nil
-		block := <-results
-		if block == nil {
-			return errors.New("broken database")
-		}
-		// Push the block into the import queue and process contiguous ranges
-		priority := -int64(block.NumberU64())
-		if reverse {
-			priority = int64(block.NumberU64())
-		}
-		queue.Push(block, priority)
-		for !queue.Empty() {
-			// If the next available item is gapped, return
-			if _, priority := queue.Peek(); !reverse && -priority != next || reverse && priority != next {
-				break
-			}
-			// Next block available, pop it off and index it
-			block = queue.PopItem().(*types.Block)
-
-			if !reverse {
-				next++
-			} else {
-				next--
-			}
-			// Invoke action to inject specified data into key-value database.
-			action(batch, block)
-
-			// If enough data was accumulated in memory or we're at the last block, dump to disk
-			if batch.ValueSize() > ethdb.IdealBatchSize || next == last {
-				if err := batch.Write(); err != nil {
-					return err
-				}
-				batch.Reset()
-			}
-			// If we've spent too much time already, notify the user of what we're doing
-			if time.Since(logged) > 8*time.Second {
-				log.Info(progMsg, "blocks", int64(math.Abs(float64(next-first))), "total", to-from, "tail", block.Number(), "hash", block.Hash(), "elapsed", common.PrettyDuration(time.Since(start)))
-				logged = time.Now()
-			}
-		}
-	}
-	tail := to
-	if reverse {
-		tail = from
-	}
-	log.Info(doneMsg, "blocks", to-from, "tail", tail, "elapsed", common.PrettyDuration(time.Since(start)))
-	return nil
-}
-
 // InitDatabaseFromFreezer reinitializes an empty database from a previous batch
 // of frozen ancient blocks. The method iterates over all the frozen blocks and
 // injects into the database the block hash->number mappings.
@@ -173,12 +47,15 @@ func InitDatabaseFromFreezer(db ethdb.Database) {
 		hash   common.Hash
 	)
 	for i := uint64(0); i < frozen; i++ {
+		// Since the freezer has all data in sequential order on a file,
+		// it would be 'neat' to read more data in one go, and let the
+		// freezerdb return N items (e.g up to 1000 items per go)
+		// That would require an API change in Ancients though
 		if h, err := db.Ancient(freezerHashTable, i); err != nil {
 			log.Crit("Failed to init database from freezer", "err", err)
 		} else {
 			hash = common.BytesToHash(h)
 		}
-
 		WriteHeaderNumber(batch, hash, i)
 		// If enough data was accumulated in memory or we're at the last block, dump to disk
 		if batch.ValueSize() > ethdb.IdealBatchSize {
@@ -189,7 +66,7 @@ func InitDatabaseFromFreezer(db ethdb.Database) {
 		}
 		// If we've spent too much time already, notify the user of what we're doing
 		if time.Since(logged) > 8*time.Second {
-			log.Info("Initializing database from freezer", "blocks", i, "total", frozen, "tail", i, "hash", hash, "elapsed", common.PrettyDuration(time.Since(start)))
+			log.Info("Initializing database from freezer", "total", frozen, "number", i, "hash", hash, "elapsed", common.PrettyDuration(time.Since(start)))
 			logged = time.Now()
 		}
 	}
@@ -200,7 +77,109 @@ func InitDatabaseFromFreezer(db ethdb.Database) {
 
 	WriteHeadHeaderHash(db, hash)
 	WriteHeadFastBlockHash(db, hash)
-	log.Info("Initialized database from freezer", "blocks", frozen, "tail", frozen, "elapsed", common.PrettyDuration(time.Since(start)))
+	log.Info("Initialized database from freezer", "blocks", frozen, "elapsed", common.PrettyDuration(time.Since(start)))
+}
+
+type blockTxHashes struct {
+	number uint64
+	hashes []common.Hash
+}
+
+// iterateTransactions iterates over all transactions in the (canon) block
+// number(s) given, and yields the hashes on a channel
+func iterateTransactions(db ethdb.Database, from uint64, to uint64, reverse bool) (chan *blockTxHashes, chan struct{}) {
+	// One thread sequentially reads data from db
+	type numberRlp struct {
+		number uint64
+		rlp    rlp.RawValue
+	}
+	if to == from {
+		return nil, nil
+	}
+	threads := to - from
+	if cpus := runtime.NumCPU(); threads > uint64(cpus) {
+		threads = uint64(cpus)
+	}
+	var (
+		rlpCh    = make(chan *numberRlp, threads*2)     // we send raw rlp over this channel
+		hashesCh = make(chan *blockTxHashes, threads*2) // send hashes over hashesCh
+		abortCh  = make(chan struct{})
+	)
+	// lookup runs in one instance
+	lookup := func() {
+		n, end := from, to
+		if reverse {
+			n, end = to-1, from-1
+		}
+		defer close(rlpCh)
+		for n != end {
+			data := ReadCanonicalBodyRLP(db, n)
+			// Feed the block to the aggregator, or abort on interrupt
+			select {
+			case rlpCh <- &numberRlp{n, data}:
+			case <-abortCh:
+				return
+			}
+			if reverse {
+				n--
+			} else {
+				n++
+			}
+		}
+	}
+	// process runs in parallell
+	nThreadsAlive := int32(threads)
+	process := func() {
+		defer func() {
+			// Last processor closes the result channel
+			if atomic.AddInt32(&nThreadsAlive, -1) == 0 {
+				close(hashesCh)
+			}
+		}()
+
+		var hasher = sha3.NewLegacyKeccak256()
+		for data := range rlpCh {
+			it, err := rlp.NewListIterator(data.rlp)
+			if err != nil {
+				log.Warn("tx iteration error", "error", err)
+				return
+			}
+			it.Next()
+			txs := it.Value()
+			txIt, err := rlp.NewListIterator(txs)
+			if err != nil {
+				log.Warn("tx iteration error", "error", err)
+				return
+			}
+			var hashes []common.Hash
+			for txIt.Next() {
+				if err := txIt.Err(); err != nil {
+					log.Warn("tx iteration error", "error", err)
+					return
+				}
+				var txHash common.Hash
+				hasher.Reset()
+				hasher.Write(txIt.Value())
+				hasher.Sum(txHash[:0])
+				hashes = append(hashes, txHash)
+			}
+			result := &blockTxHashes{
+				hashes: hashes,
+				number: data.number,
+			}
+			// Feed the block to the aggregator, or abort on interrupt
+			select {
+			case hashesCh <- result:
+			case <-abortCh:
+				return
+			}
+		}
+	}
+	go lookup() // start the sequential db accessor
+	for i := 0; i < int(threads); i++ {
+		go process()
+	}
+	return hashesCh, abortCh
 }
 
 // IndexTransactions creates txlookup indices of the specified block range.
@@ -209,41 +188,118 @@ func InitDatabaseFromFreezer(db ethdb.Database) {
 // We can write tx index tail flag periodically even without the whole indexing
 // procedure is finished. So that we can resume indexing procedure next time quickly.
 func IndexTransactions(db ethdb.Database, from uint64, to uint64) {
-	// hashTxs calculates transaction hash in advance using the multi-routine's
-	// concurrent computing power.
-	hashTxs := func(block *types.Block) {
-		for _, tx := range block.Transactions() {
-			tx.Hash()
+	// short circuit for invalid range
+	if from >= to {
+		return
+	}
+	var (
+		hashesCh, abortCh = iterateTransactions(db, from, to, true)
+		batch             = db.NewBatch()
+		start             = time.Now()
+		logged            = start.Add(-7 * time.Second)
+		//  Since we iterate in reverse, we expect the first number to come
+		// in to be [to-1]. Therefore, setting lastNum to means that the
+		// prqueue gap-evaluation will work correctly
+		lastNum = to
+		queue   = prque.New(nil)
+		// for stats reporting
+		blockCount, txCount = 0, 0
+	)
+	defer close(abortCh)
+
+	for chanDelivery := range hashesCh {
+		// Push the delivery into the queue and process contiguous ranges.
+		// Since we iterate in reverse, so lower numbers have lower prio, and
+		// we can use the number directly as prio marker
+		queue.Push(chanDelivery, int64(chanDelivery.number))
+		for !queue.Empty() {
+			// If the next available item is gapped, return
+			if _, priority := queue.Peek(); priority != int64(lastNum-1) {
+				break
+			}
+			// Next block available, pop it off and index it
+			delivery := queue.PopItem().(*blockTxHashes)
+			lastNum = delivery.number
+			number := new(big.Int).SetUint64(lastNum).Bytes()
+			WriteTxLookupEntriesByHash(batch, number, delivery.hashes)
+			blockCount++
+			txCount += len(delivery.hashes)
+			// If enough data was accumulated in memory or we're at the last block, dump to disk
+			if batch.ValueSize() > ethdb.IdealBatchSize {
+				// Also write the tail there
+				WriteTxIndexTail(batch, lastNum)
+				if err := batch.Write(); err != nil {
+					log.Crit("Failed writing batch to db", "error", err)
+					return
+				}
+				batch.Reset()
+			}
+			// If we've spent too much time already, notify the user of what we're doing
+			if time.Since(logged) > 8*time.Second {
+				log.Info("Indexing transactions", "blocks", blockCount, "txs", txCount, "current", lastNum, "total", to-from, "elapsed", common.PrettyDuration(time.Since(start)))
+				logged = time.Now()
+			}
 		}
 	}
-	// writeIndices injects txlookup indices into the database.
-	writeIndices := func(batch ethdb.Batch, block *types.Block) {
-		WriteTxLookupEntries(batch, block)
-		if block.NumberU64() == to-1 || block.NumberU64()%10000 == 0 {
-			WriteTxIndexTail(batch, block.NumberU64())
+	if lastNum < to {
+		WriteTxIndexTail(batch, lastNum)
+		// No need to write the batch if we never entered the loop above...
+		if err := batch.Write(); err != nil {
+			log.Crit("Failed writing batch to db", "error", err)
+			return
 		}
 	}
-	if err := iterateCanonicalChain(db, from, to, hashTxs, writeIndices, true, "Indexing transactions", "Indexed transactions"); err != nil {
-		log.Crit("Failed to index transactions", "err", err)
-	}
-	WriteTxIndexTail(db, from)
+	log.Info("Indexed transactions", "blocks", blockCount, "txs", txCount, "elapsed", common.PrettyDuration(time.Since(start)))
 }
 
 // UnindexTransactions removes txlookup indices of the specified block range.
 func UnindexTransactions(db ethdb.Database, from uint64, to uint64) {
+	// short circuit for invalid range
+	if from >= to {
+		return
+	}
 	// Write flag first and then unindex the transaction indices. Some indices
 	// will be left in the database if crash happens but it's fine.
 	WriteTxIndexTail(db, to)
-
 	// If only one block is unindexed, do it directly
-	if from+1 == to {
-		DeleteTxLookupEntries(db, ReadBlock(db, ReadCanonicalHash(db, from), from))
-		log.Info("Unindexed transactions", "blocks", 1, "tail", to)
+	//if from+1 == to {
+	//	data := ReadCanonicalBodyRLP(db, uint64(from))
+	//	DeleteTxLookupEntries(db, ReadBlock(db, ReadCanonicalHash(db, from), from))
+	//	log.Info("Unindexed transactions", "blocks", 1, "tail", to)
+	//	return
+	//}
+	// TODO @holiman, add this back (if we want it)
+	var (
+		hashesCh, abortCh = iterateTransactions(db, from, to, false)
+		batch             = db.NewBatch()
+		start             = time.Now()
+		logged            = start.Add(-7 * time.Second)
+	)
+	defer close(abortCh)
+	// Otherwise spin up the concurrent iterator and unindexer
+	count := 0
+	for delivery := range hashesCh {
+		DeleteTxLookupEntriesByHash(batch, delivery.hashes)
+		// If enough data was accumulated in memory or we're at the last block, dump to disk
+		// A batch counts the size of deletion as '1', so we need to flush more
+		// often than that.
+		count++
+		if count%1000 == 0 {
+			if err := batch.Write(); err != nil {
+				log.Crit("Failed writing batch to db", "error", err)
+				return
+			}
+			batch.Reset()
+		}
+		// If we've spent too much time already, notify the user of what we're doing
+		if time.Since(logged) > 8*time.Second {
+			log.Info("Unindexing transactions", "blocks", int64(math.Abs(float64(delivery.number-from))), "total", to-from, "elapsed", common.PrettyDuration(time.Since(start)))
+			logged = time.Now()
+		}
+	}
+	if err := batch.Write(); err != nil {
+		log.Crit("Failed writing batch to db", "error", err)
 		return
 	}
-	// Otherwise spin up the concurrent iterator and unindexer
-	deleteIndices := func(batch ethdb.Batch, block *types.Block) { DeleteTxLookupEntries(batch, block) }
-	if err := iterateCanonicalChain(db, from, to, nil, deleteIndices, false, "Unindexing transactions", "Unindexed transactions"); err != nil {
-		log.Crit("Failed to unindex transactions", "err", err)
-	}
+	log.Info("Indexed transactions", "blocks", int64(math.Abs(float64(to-from))), "total", to-from, "elapsed", common.PrettyDuration(time.Since(start)))
 }

core/rawdb/chain_iterator_test.go

@@ -19,11 +19,11 @@ package rawdb
 import (
 	"math/big"
 	"reflect"
+	"sort"
 	"testing"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
-	"github.com/ethereum/go-ethereum/ethdb"
 )
 
 func TestChainIterator(t *testing.T) {
@@ -31,11 +31,13 @@ func TestChainIterator(t *testing.T) {
 	chainDb := NewMemoryDatabase()
 
 	var block *types.Block
+	var txs []*types.Transaction
 	for i := uint64(0); i <= 10; i++ {
 		if i == 0 {
 			block = types.NewBlock(&types.Header{Number: big.NewInt(int64(i))}, nil, nil, nil) // Empty genesis block
 		} else {
 			tx := types.NewTransaction(i, common.BytesToAddress([]byte{0x11}), big.NewInt(111), 1111, big.NewInt(11111), []byte{0x11, 0x11, 0x11})
+			txs = append(txs, tx)
 			block = types.NewBlock(&types.Header{Number: big.NewInt(int64(i))}, []*types.Transaction{tx}, nil, nil)
 		}
 		WriteBlock(chainDb, block)
@@ -45,23 +47,36 @@ func TestChainIterator(t *testing.T) {
 	var cases = []struct {
 		from, to uint64
 		reverse  bool
-		expect   []uint64
+		expect   []int
 	}{
-		{0, 11, true, []uint64{10, 9, 8, 7, 6, 5, 4, 3, 2, 1, 0}},
+		{0, 11, true, []int{10, 9, 8, 7, 6, 5, 4, 3, 2, 1, 0}},
 		{0, 0, true, nil},
-		{10, 11, true, []uint64{10}},
-		{0, 11, false, []uint64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10}},
+		{0, 5, true, []int{4, 3, 2, 1, 0}},
+		{10, 11, true, []int{10}},
+		{0, 11, false, []int{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10}},
 		{0, 0, false, nil},
-		{10, 11, false, []uint64{10}},
+		{10, 11, false, []int{10}},
 	}
 	for i, c := range cases {
-		var visit []uint64
-		err := iterateCanonicalChain(chainDb, c.from, c.to, nil, func(db ethdb.Batch, b *types.Block) { visit = append(visit, b.NumberU64()) }, c.reverse, "", "")
-		if err != nil {
-			t.Fatalf("Case %d failed, err %v", i, err)
+		var numbers []int
+		hashCh, _ := iterateTransactions(chainDb, c.from, c.to, c.reverse)
+		if hashCh != nil {
+			for h := range hashCh {
+				numbers = append(numbers, int(h.number))
+				if len(h.hashes) > 0 {
+					if got, exp := h.hashes[0], txs[h.number-1].Hash(); got != exp {
+						t.Fatalf("hash wrong, got %x exp %x", got, exp)
+					}
+				}
+			}
 		}
-		if !reflect.DeepEqual(visit, c.expect) {
-			t.Fatalf("Case %d failed, visit element mismatch, want %v, got %v", i, c.expect, visit)
+		if !c.reverse {
+			sort.Ints(numbers)
+		} else {
+			sort.Sort(sort.Reverse(sort.IntSlice(numbers)))
+		}
+		if !reflect.DeepEqual(numbers, c.expect) {
+			t.Fatalf("Case %d failed, visit element mismatch, want %v, got %v", i, c.expect, numbers)
 		}
 	}
 }

core/rlp_test.go

@@ -17,7 +17,6 @@ package core
 
 import (
 	"fmt"
-	"golang.org/x/crypto/sha3"
 	"math/big"
 	"testing"
 
@@ -28,6 +27,7 @@ import (
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/ethereum/go-ethereum/rlp"
+	"golang.org/x/crypto/sha3"
 )
 
 func getBlock(transactions int, uncles int, dataSize int) *types.Block {
@@ -152,9 +152,13 @@ func BenchmarkHashing(b *testing.B) {
 	var got common.Hash
 	var hasher = sha3.NewLegacyKeccak256()
 	b.Run("iteratorhashing", func(b *testing.B) {
+		b.ResetTimer()
 		for i := 0; i < b.N; i++ {
 			var hash common.Hash
 			it, err := rlp.NewListIterator(bodyRlp)
+			if err != nil {
+				b.Fatal(err)
+			}
 			it.Next()
 			txs := it.Value()
 			txIt, err := rlp.NewListIterator(txs)
@@ -171,6 +175,7 @@ func BenchmarkHashing(b *testing.B) {
 	})
 	var exp common.Hash
 	b.Run("fullbodyhashing", func(b *testing.B) {
+		b.ResetTimer()
 		for i := 0; i < b.N; i++ {
 			var body types.Body
 			rlp.DecodeBytes(bodyRlp, &body)
@@ -180,6 +185,7 @@ func BenchmarkHashing(b *testing.B) {
 		}
 	})
 	b.Run("fullblockhashing", func(b *testing.B) {
+		b.ResetTimer()
 		for i := 0; i < b.N; i++ {
 			var block types.Block
 			rlp.DecodeBytes(blockRlp, &block)