diff --git a/core/rawdb/chain_iterator.go b/core/rawdb/chain_iterator.go index ecbc44e1f1..35a1ff8874 100644 --- a/core/rawdb/chain_iterator.go +++ b/core/rawdb/chain_iterator.go @@ -18,16 +18,11 @@ package rawdb import ( "encoding/binary" - "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" ) // InitDatabaseFromFreezer reinitializes an empty database from a previous batch @@ -84,285 +79,6 @@ func InitDatabaseFromFreezer(db ethdb.Database) { 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. If there is a signal -// received from interrupt channel, the iteration will be aborted and result -// channel will be closed. -func iterateTransactions(db ethdb.Database, from uint64, to uint64, reverse bool, interrupt chan struct{}) chan *blockTxHashes { - // One thread sequentially reads data from db - type numberRlp struct { - number uint64 - rlp rlp.RawValue - } - if to == from { - return 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 - ) - // 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 <-interrupt: - return - } - if reverse { - n-- - } else { - n++ - } - } - } - // process runs in parallel - var nThreadsAlive atomic.Int32 - nThreadsAlive.Store(int32(threads)) - process := func() { - defer func() { - // Last processor closes the result channel - if nThreadsAlive.Add(-1) == 0 { - close(hashesCh) - } - }() - for data := range rlpCh { - var body types.Body - if err := rlp.DecodeBytes(data.rlp, &body); err != nil { - log.Warn("Failed to decode block body", "block", data.number, "error", err) - return - } - var hashes []common.Hash - for _, tx := range body.Transactions { - hashes = append(hashes, tx.Hash()) - } - result := &blockTxHashes{ - hashes: hashes, - number: data.number, - } - // Feed the block to the aggregator, or abort on interrupt - select { - case hashesCh <- result: - case <-interrupt: - return - } - } - } - go lookup() // start the sequential db accessor - for i := 0; i < int(threads); i++ { - go process() - } - return hashesCh -} - -// indexTransactions creates txlookup indices of the specified block range. -// -// This function iterates canonical chain in reverse order, it has one main advantage: -// 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. -// -// There is a passed channel, the whole procedure will be interrupted if any -// signal received. -func indexTransactions(db ethdb.Database, from uint64, to uint64, interrupt chan struct{}, hook func(uint64) bool, report bool) { - // short circuit for invalid range - if from >= to { - return - } - var ( - hashesCh = iterateTransactions(db, from, to, true, interrupt) - 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 - // queue gap-evaluation will work correctly - lastNum = to - queue = prque.New[int64, *blockTxHashes](nil) - blocks, txs = 0, 0 // for stats reporting - ) - 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 - } - // For testing - if hook != nil && !hook(lastNum-1) { - break - } - // Next block available, pop it off and index it - delivery := queue.PopItem() - lastNum = delivery.number - WriteTxLookupEntries(batch, delivery.number, delivery.hashes) - blocks++ - txs += 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 { - WriteTxIndexTail(batch, lastNum) // Also write the tail here - 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", blocks, "txs", txs, "tail", lastNum, "total", to-from, "elapsed", common.PrettyDuration(time.Since(start))) - logged = time.Now() - } - } - } - // Flush the new indexing tail and the last committed data. It can also happen - // that the last batch is empty because nothing to index, but the tail has to - // be flushed anyway. - WriteTxIndexTail(batch, lastNum) - if err := batch.Write(); err != nil { - log.Crit("Failed writing batch to db", "error", err) - return - } - logger := log.Debug - if report { - logger = log.Info - } - select { - case <-interrupt: - logger("Transaction indexing interrupted", "blocks", blocks, "txs", txs, "tail", lastNum, "elapsed", common.PrettyDuration(time.Since(start))) - default: - logger("Indexed transactions", "blocks", blocks, "txs", txs, "tail", lastNum, "elapsed", common.PrettyDuration(time.Since(start))) - } -} - -// IndexTransactions creates txlookup indices of the specified block range. The from -// is included while to is excluded. -// -// This function iterates canonical chain in reverse order, it has one main advantage: -// 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. -// -// There is a passed channel, the whole procedure will be interrupted if any -// signal received. -func IndexTransactions(db ethdb.Database, from uint64, to uint64, interrupt chan struct{}, report bool) { - indexTransactions(db, from, to, interrupt, nil, report) -} - -// indexTransactionsForTesting is the internal debug version with an additional hook. -func indexTransactionsForTesting(db ethdb.Database, from uint64, to uint64, interrupt chan struct{}, hook func(uint64) bool) { - indexTransactions(db, from, to, interrupt, hook, false) -} - -// unindexTransactions removes txlookup indices of the specified block range. -// -// There is a passed channel, the whole procedure will be interrupted if any -// signal received. -func unindexTransactions(db ethdb.Database, from uint64, to uint64, interrupt chan struct{}, hook func(uint64) bool, report bool) { - // short circuit for invalid range - if from >= to { - return - } - var ( - hashesCh = iterateTransactions(db, from, to, false, interrupt) - batch = db.NewBatch() - start = time.Now() - logged = start.Add(-7 * time.Second) - - // we expect the first number to come in to be [from]. Therefore, setting - // nextNum to from means that the queue gap-evaluation will work correctly - nextNum = from - queue = prque.New[int64, *blockTxHashes](nil) - blocks, txs = 0, 0 // for stats reporting - ) - // Otherwise spin up the concurrent iterator and unindexer - for delivery := range hashesCh { - // Push the delivery into the queue and process contiguous ranges. - queue.Push(delivery, -int64(delivery.number)) - for !queue.Empty() { - // If the next available item is gapped, return - if _, priority := queue.Peek(); -priority != int64(nextNum) { - break - } - // For testing - if hook != nil && !hook(nextNum) { - break - } - delivery := queue.PopItem() - nextNum = delivery.number + 1 - DeleteTxLookupEntries(batch, delivery.hashes) - txs += len(delivery.hashes) - blocks++ - - // 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. - if blocks%1000 == 0 { - WriteTxIndexTail(batch, nextNum) - 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", blocks, "txs", txs, "total", to-from, "elapsed", common.PrettyDuration(time.Since(start))) - logged = time.Now() - } - } - } - // Flush the new indexing tail and the last committed data. It can also happen - // that the last batch is empty because nothing to unindex, but the tail has to - // be flushed anyway. - WriteTxIndexTail(batch, nextNum) - if err := batch.Write(); err != nil { - log.Crit("Failed writing batch to db", "error", err) - return - } - logger := log.Debug - if report { - logger = log.Info - } - select { - case <-interrupt: - logger("Transaction unindexing interrupted", "blocks", blocks, "txs", txs, "tail", to, "elapsed", common.PrettyDuration(time.Since(start))) - default: - logger("Unindexed transactions", "blocks", blocks, "txs", txs, "tail", to, "elapsed", common.PrettyDuration(time.Since(start))) - } -} - -// UnindexTransactions removes txlookup indices of the specified block range. -// The from is included while to is excluded. -// -// There is a passed channel, the whole procedure will be interrupted if any -// signal received. -func UnindexTransactions(db ethdb.Database, from uint64, to uint64, interrupt chan struct{}, report bool) { - unindexTransactions(db, from, to, interrupt, nil, report) -} - -// unindexTransactionsForTesting is the internal debug version with an additional hook. -func unindexTransactionsForTesting(db ethdb.Database, from uint64, to uint64, interrupt chan struct{}, hook func(uint64) bool) { - unindexTransactions(db, from, to, interrupt, hook, false) -} - // PruneTransactionIndex removes all tx index entries below a certain block number. func PruneTransactionIndex(db ethdb.Database, pruneBlock uint64) { tail := ReadTxIndexTail(db) diff --git a/core/rawdb/chain_iterator_test.go b/core/rawdb/chain_iterator_test.go index 75bd5a9a94..ecc35d9479 100644 --- a/core/rawdb/chain_iterator_test.go +++ b/core/rawdb/chain_iterator_test.go @@ -18,9 +18,6 @@ package rawdb import ( "math/big" - "reflect" - "sort" - "sync" "testing" "github.com/ethereum/go-ethereum/common" @@ -28,81 +25,6 @@ import ( "github.com/ethereum/go-ethereum/ethdb" ) -func TestChainIterator(t *testing.T) { - // Construct test chain db - chainDb := NewMemoryDatabase() - - var block *types.Block - var txs []*types.Transaction - to := common.BytesToAddress([]byte{0x11}) - block = types.NewBlock(&types.Header{Number: big.NewInt(int64(0))}, nil, nil, newTestHasher()) // Empty genesis block - WriteBlock(chainDb, block) - WriteCanonicalHash(chainDb, block.Hash(), block.NumberU64()) - for i := uint64(1); i <= 10; i++ { - var tx *types.Transaction - if i%2 == 0 { - tx = types.NewTx(&types.LegacyTx{ - Nonce: i, - GasPrice: big.NewInt(11111), - Gas: 1111, - To: &to, - Value: big.NewInt(111), - Data: []byte{0x11, 0x11, 0x11}, - }) - } else { - tx = types.NewTx(&types.AccessListTx{ - ChainID: big.NewInt(1337), - Nonce: i, - GasPrice: big.NewInt(11111), - Gas: 1111, - To: &to, - Value: big.NewInt(111), - Data: []byte{0x11, 0x11, 0x11}, - }) - } - txs = append(txs, tx) - block = types.NewBlock(&types.Header{Number: big.NewInt(int64(i))}, &types.Body{Transactions: types.Transactions{tx}}, nil, newTestHasher()) - WriteBlock(chainDb, block) - WriteCanonicalHash(chainDb, block.Hash(), block.NumberU64()) - } - - var cases = []struct { - from, to uint64 - reverse bool - expect []int - }{ - {0, 11, true, []int{10, 9, 8, 7, 6, 5, 4, 3, 2, 1, 0}}, - {0, 0, true, nil}, - {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, []int{10}}, - } - for i, c := range cases { - var numbers []int - hashCh := iterateTransactions(chainDb, c.from, c.to, c.reverse, nil) - 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("block %d: hash wrong, got %x exp %x", h.number, got, exp) - } - } - } - } - 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) - } - } -} - func initDatabaseWithTransactions(db ethdb.Database) ([]*types.Block, []*types.Transaction) { var blocks []*types.Block var txs []*types.Transaction @@ -147,84 +69,16 @@ func initDatabaseWithTransactions(db ethdb.Database) ([]*types.Block, []*types.T return blocks, txs } -func TestIndexTransactions(t *testing.T) { - // Construct test chain db - chainDB := NewMemoryDatabase() - - _, txs := initDatabaseWithTransactions(chainDB) - - // verify checks whether the tx indices in the range [from, to) - // is expected. - verify := func(from, to int, exist bool, tail uint64) { - for i := from; i < to; i++ { - if i == 0 { - continue - } - number := ReadTxLookupEntry(chainDB, txs[i-1].Hash()) - if exist && number == nil { - t.Fatalf("Transaction index %d missing", i) - } - if !exist && number != nil { - t.Fatalf("Transaction index %d is not deleted", i) - } - } - number := ReadTxIndexTail(chainDB) - if number == nil || *number != tail { - t.Fatalf("Transaction tail mismatch") - } - } - IndexTransactions(chainDB, 5, 11, nil, false) - verify(5, 11, true, 5) - verify(0, 5, false, 5) - - IndexTransactions(chainDB, 0, 5, nil, false) - verify(0, 11, true, 0) - - UnindexTransactions(chainDB, 0, 5, nil, false) - verify(5, 11, true, 5) - verify(0, 5, false, 5) - - UnindexTransactions(chainDB, 5, 11, nil, false) - verify(0, 11, false, 11) - - // Testing corner cases - signal := make(chan struct{}) - var once sync.Once - indexTransactionsForTesting(chainDB, 5, 11, signal, func(n uint64) bool { - if n <= 8 { - once.Do(func() { - close(signal) - }) - return false - } - return true - }) - verify(9, 11, true, 9) - verify(0, 9, false, 9) - IndexTransactions(chainDB, 0, 9, nil, false) - - signal = make(chan struct{}) - var once2 sync.Once - unindexTransactionsForTesting(chainDB, 0, 11, signal, func(n uint64) bool { - if n >= 8 { - once2.Do(func() { - close(signal) - }) - return false - } - return true - }) - verify(8, 11, true, 8) - verify(0, 8, false, 8) -} - func TestPruneTransactionIndex(t *testing.T) { chainDB := NewMemoryDatabase() blocks, _ := initDatabaseWithTransactions(chainDB) lastBlock := blocks[len(blocks)-1].NumberU64() pruneBlock := lastBlock - 3 - IndexTransactions(chainDB, 0, lastBlock+1, nil, false) + for i := uint64(0); i <= lastBlock; i++ { + WriteTxLookupEntriesByBlock(chainDB, blocks[i]) + } + WriteTxIndexTail(chainDB, 0) // Check all transactions are in index. for _, block := range blocks { diff --git a/core/txindexer.go b/core/txindexer.go index 587118ed7f..5658d716b9 100644 --- a/core/txindexer.go +++ b/core/txindexer.go @@ -18,12 +18,17 @@ package core import ( "fmt" + "runtime" "sync/atomic" + "time" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/prque" "github.com/ethereum/go-ethereum/core/rawdb" + "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/rlp" ) // TxIndexProgress is the struct describing the progress for transaction indexing. @@ -122,7 +127,7 @@ func (indexer *txIndexer) run(head uint64, stop chan struct{}, done chan struct{ from = head - indexer.limit + 1 } from = max(from, indexer.cutoff) - rawdb.IndexTransactions(indexer.db, from, head+1, stop, true) + indexer.index(from, head+1, stop, nil, true) return } // The tail flag is existent (which means indexes in [tail, head] should be @@ -130,7 +135,7 @@ func (indexer *txIndexer) run(head uint64, stop chan struct{}, done chan struct{ if indexer.limit == 0 || head < indexer.limit { if *tail > 0 { from := max(uint64(0), indexer.cutoff) - rawdb.IndexTransactions(indexer.db, from, *tail, stop, true) + indexer.index(from, *tail, stop, nil, true) } return } @@ -140,10 +145,10 @@ func (indexer *txIndexer) run(head uint64, stop chan struct{}, done chan struct{ from = max(from, indexer.cutoff) if from < *tail { // Reindex a part of missing indices and rewind index tail to HEAD-limit - rawdb.IndexTransactions(indexer.db, from, *tail, stop, true) + indexer.index(from, *tail, stop, nil, true) } else { // Unindex a part of stale indices and forward index tail to HEAD-limit - rawdb.UnindexTransactions(indexer.db, *tail, from, stop, false) + indexer.unindex(*tail, from, stop, nil, false) } } @@ -331,3 +336,250 @@ func (indexer *txIndexer) close() { case <-indexer.closed: } } + +type blockTxHashes struct { + number uint64 + hashes []common.Hash +} + +// iterate iterates over all transactions in the (canon) block +// number(s) given, and yields the hashes on a channel. If there is a signal +// received from interrupt channel, the iteration will be aborted and result +// channel will be closed. +func (indexer *txIndexer) iterate(from uint64, to uint64, reverse bool, interrupt chan struct{}) chan *blockTxHashes { + // One thread sequentially reads data from db + type numberRlp struct { + number uint64 + rlp rlp.RawValue + } + if to == from { + return 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 + ) + // 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 := rawdb.ReadCanonicalBodyRLP(indexer.db, n) + // Feed the block to the aggregator, or abort on interrupt + select { + case rlpCh <- &numberRlp{n, data}: + case <-interrupt: + return + } + if reverse { + n-- + } else { + n++ + } + } + } + // process runs in parallel + var nThreadsAlive atomic.Int32 + nThreadsAlive.Store(int32(threads)) + process := func() { + defer func() { + // Last processor closes the result channel + if nThreadsAlive.Add(-1) == 0 { + close(hashesCh) + } + }() + for data := range rlpCh { + var body types.Body + if err := rlp.DecodeBytes(data.rlp, &body); err != nil { + log.Warn("Failed to decode block body", "block", data.number, "error", err) + return + } + var hashes []common.Hash + for _, tx := range body.Transactions { + hashes = append(hashes, tx.Hash()) + } + result := &blockTxHashes{ + hashes: hashes, + number: data.number, + } + // Feed the block to the aggregator, or abort on interrupt + select { + case hashesCh <- result: + case <-interrupt: + return + } + } + } + go lookup() // start the sequential db accessor + for i := 0; i < int(threads); i++ { + go process() + } + return hashesCh +} + +// index creates txlookup indices of the specified block range. +// +// This function iterates canonical chain in reverse order, it has one main advantage: +// 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. +// +// There is a passed channel, the whole procedure will be interrupted if any +// signal received. +func (indexer *txIndexer) index(from uint64, to uint64, interrupt chan struct{}, hook func(uint64) bool, report bool) { + // short circuit for invalid range + if from >= to { + return + } + var ( + hashesCh = indexer.iterate(from, to, true, interrupt) + batch = indexer.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 + // queue gap-evaluation will work correctly + lastNum = to + queue = prque.New[int64, *blockTxHashes](nil) + blocks, txs = 0, 0 // for stats reporting + ) + 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 + } + // For testing + if hook != nil && !hook(lastNum-1) { + break + } + // Next block available, pop it off and index it + delivery := queue.PopItem() + lastNum = delivery.number + rawdb.WriteTxLookupEntries(batch, delivery.number, delivery.hashes) + blocks++ + txs += 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 { + rawdb.WriteTxIndexTail(batch, lastNum) // Also write the tail here + 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", blocks, "txs", txs, "tail", lastNum, "total", to-from, "elapsed", common.PrettyDuration(time.Since(start))) + logged = time.Now() + } + } + } + // Flush the new indexing tail and the last committed data. It can also happen + // that the last batch is empty because nothing to index, but the tail has to + // be flushed anyway. + rawdb.WriteTxIndexTail(batch, lastNum) + if err := batch.Write(); err != nil { + log.Crit("Failed writing batch to db", "error", err) + return + } + logger := log.Debug + if report { + logger = log.Info + } + select { + case <-interrupt: + logger("Transaction indexing interrupted", "blocks", blocks, "txs", txs, "tail", lastNum, "elapsed", common.PrettyDuration(time.Since(start))) + default: + logger("Indexed transactions", "blocks", blocks, "txs", txs, "tail", lastNum, "elapsed", common.PrettyDuration(time.Since(start))) + } +} + +// unindex removes txlookup indices of the specified block range. +// +// There is a passed channel, the whole procedure will be interrupted if any +// signal received. +func (indexer *txIndexer) unindex(from uint64, to uint64, interrupt chan struct{}, hook func(uint64) bool, report bool) { + // short circuit for invalid range + if from >= to { + return + } + var ( + hashesCh = indexer.iterate(from, to, false, interrupt) + batch = indexer.db.NewBatch() + start = time.Now() + logged = start.Add(-7 * time.Second) + + // we expect the first number to come in to be [from]. Therefore, setting + // nextNum to from means that the queue gap-evaluation will work correctly + nextNum = from + queue = prque.New[int64, *blockTxHashes](nil) + blocks, txs = 0, 0 // for stats reporting + ) + // Otherwise spin up the concurrent iterator and unindexer + for delivery := range hashesCh { + // Push the delivery into the queue and process contiguous ranges. + queue.Push(delivery, -int64(delivery.number)) + for !queue.Empty() { + // If the next available item is gapped, return + if _, priority := queue.Peek(); -priority != int64(nextNum) { + break + } + // For testing + if hook != nil && !hook(nextNum) { + break + } + delivery := queue.PopItem() + nextNum = delivery.number + 1 + rawdb.DeleteTxLookupEntries(batch, delivery.hashes) + txs += len(delivery.hashes) + blocks++ + + // 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. + if blocks%1000 == 0 { + rawdb.WriteTxIndexTail(batch, nextNum) + 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", blocks, "txs", txs, "total", to-from, "elapsed", common.PrettyDuration(time.Since(start))) + logged = time.Now() + } + } + } + // Flush the new indexing tail and the last committed data. It can also happen + // that the last batch is empty because nothing to unindex, but the tail has to + // be flushed anyway. + rawdb.WriteTxIndexTail(batch, nextNum) + if err := batch.Write(); err != nil { + log.Crit("Failed writing batch to db", "error", err) + return + } + logger := log.Debug + if report { + logger = log.Info + } + select { + case <-interrupt: + logger("Transaction unindexing interrupted", "blocks", blocks, "txs", txs, "tail", to, "elapsed", common.PrettyDuration(time.Since(start))) + default: + logger("Unindexed transactions", "blocks", blocks, "txs", txs, "tail", to, "elapsed", common.PrettyDuration(time.Since(start))) + } +}