parallel bloom calculation

[steventranjs]

Description

• Trigger receipt.Bloom calculation in a new thread channel out of applyTransaction

Rationale

• Enhance applyTransaction processing

Example

add an example CLI or API response...

Changes

Notable changes:

• add each change in a bullet point here
• ...

Preflight checks

• build passed (make build)
• tests passed (make test)
• manual transaction test passed

Already reviewed by

...

Related issues

... reference related issue #'s here ...

[j75689]

I have an idea. Maybe we can define an interface to process some receipt data, like.

package core

import (
	"bytes"
	"sync"

	"github.com/ethereum/go-ethereum/core/types"
)

type ReceiptProcesser interface {
	Apply(receipt *types.Receipt)
}

var (
	_ ReceiptProcesser = (*ReceiptBloomGenertor)(nil)
	_ ReceiptProcesser = (*AsyncReceiptBloomGenertor)(nil)
)

func NewReceiptBloomGenertor() *ReceiptBloomGenertor {
	return &ReceiptBloomGenertor{}
}

type ReceiptBloomGenertor struct {
}

func (p *ReceiptBloomGenertor) Apply(receipt *types.Receipt) {
	receipt.Bloom = types.CreateBloom(types.Receipts{receipt})
}

func NewAsyncReceiptBloomGenertor(length, workerSize int) *AsyncReceiptBloomGenertor {
	generator := &AsyncReceiptBloomGenertor{
		receipts: make(chan *types.Receipt, length),
		wg:       sync.WaitGroup{},
	}
	generator.startWorker(workerSize)
	return generator
}

type AsyncReceiptBloomGenertor struct {
	receipts chan *types.Receipt
	wg       sync.WaitGroup
}

func (p *AsyncReceiptBloomGenertor) startWorker(workerSize int) {
	p.wg.Add(workerSize)
	for i := 0; i < workerSize; i++ {
		go func() {
			defer p.wg.Done()
			for receipt := range p.receipts {
				if receipt != nil && bytes.Equal(receipt.Bloom[:], types.EmptyBloom[:]) {
					receipt.Bloom = types.CreateBloom(types.Receipts{receipt})
				}
			}
		}()
	}
}

func (p *AsyncReceiptBloomGenertor) Apply(receipt *types.Receipt) {
	p.receipts <- receipt
}

func (p *AsyncReceiptBloomGenertor) Close() {
	close(p.receipts)
	p.wg.Wait()
}

And change the interface of applyTransaction

func applyTransaction(msg types.Message, config *params.ChainConfig, bc ChainContext, author *common.Address, gp *GasPool, statedb *state.StateDB, header *types.Header, tx *types.Transaction, usedGas *uint64, evm *vm.EVM, receiptProcessers ...ReceiptProcesser) (*types.Receipt, error){
...
    // Set the receipt logs and create the bloom filter.
    receipt.Logs = statedb.GetLogs(tx.Hash())
    receipt.BlockHash = statedb.BlockHash()
    receipt.BlockNumber = header.Number
    receipt.TransactionIndex = uint(statedb.TxIndex())
    for _, receiptProcesser := range receiptProcessers {
    	receiptProcesser.Apply(receipt)
    }
}

And pass the bloomProcesser

// initilise bloom processors
bloomProcessors := core.NewAsyncReceiptBloomGenertor(txs.CurrentSize(), gopool.Threads(txs.CurrentSize()))
...

This may be easier to expand in the future, you can see the detail of this commit -> j75689@81f2a3e