Skip to content

Add parallel tracing for trace block #60

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Add parallel tracing for trace block #60

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
278 changes: 207 additions & 71 deletions eth/tracers/api.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -615,70 +615,7 @@ func (api *API) traceBlock(ctx context.Context, block *types.Block, metadata []t
// core.ProcessParentBlockHash(block.ParentHash(), evm)
// }

// JS tracers have high overhead. In this case run a parallel
// process that generates states in one thread and traces txes
// in separate worker threads.
if config != nil && config.Tracer != nil && *config.Tracer != "" {
if isJS := DefaultDirectory.IsJS(*config.Tracer); isJS {
return api.traceBlockParallel(ctx, block, statedb, config)
}
}
// Native tracers have low overhead
blockCtx, err := api.backend.GetBlockContext(ctx, block, statedb, api.backend)
if err != nil {
return nil, fmt.Errorf("cannot get block context: %w", err)
}
var (
txs = block.Transactions()
blockHash = block.Hash()
signer = types.MakeSigner(api.backend.ChainConfig(), block.Number(), block.Time())
results = make([]*TxTraceResult, len(txs))
)
if len(metadata) == 0 {
for i, tx := range txs {
// Generate the next state snapshot fast without tracing
msg, _ := core.TransactionToMessage(tx, signer, block.BaseFee())
txctx := &Context{
BlockHash: blockHash,
BlockNumber: block.Number(),
TxIndex: i,
TxHash: tx.Hash(),
}
res, err := api.traceTx(ctx, tx, msg, txctx, blockCtx, statedb, config, nil)
if err != nil {
results[i] = &TxTraceResult{TxHash: tx.Hash(), Error: err.Error()}
} else {
results[i] = &TxTraceResult{TxHash: tx.Hash(), Result: res}
}
}
return results, nil
}
for _, md := range metadata {
if md.ShouldIncludeInTraceResult {
i := md.IdxInEthBlock
tx := txs[i]
// Generate the next state snapshot fast without tracing
msg, _ := core.TransactionToMessage(tx, signer, block.BaseFee())
txctx := &Context{
BlockHash: blockHash,
BlockNumber: block.Number(),
TxIndex: i,
TxHash: tx.Hash(),
}
res, err := api.traceTx(ctx, tx, msg, txctx, blockCtx, statedb, config, nil)
if err != nil {
results[i] = &TxTraceResult{TxHash: tx.Hash(), Error: err.Error()}
statedb.RevertToSnapshot(0)
} else {
results[i] = &TxTraceResult{TxHash: tx.Hash(), Result: res}
}
} else {
// should not be included in result but still needs to be run because
// these txs may affect cumulative state
md.TraceRunnable(statedb)
}
}
return results, nil
return api.traceBlockFullyParallel(ctx, block, statedb, config, metadata)
}

// traceBlockParallel is for tracers that have a high overhead (read JS tracers). One thread
Expand All @@ -698,11 +635,20 @@ func (api *API) traceBlockParallel(ctx context.Context, block *types.Block, stat
threads = len(txs)
}
jobs := make(chan *txTraceTask, threads)

// Get block context and precompiles once
blockCtx, err := api.backend.GetBlockContext(ctx, block, statedb, api.backend)
if err != nil {
return nil, err
}
precompiles := api.backend.GetCustomPrecompiles(block.Number().Int64())

// Launch worker goroutines
for th := 0; th < threads; th++ {
pend.Add(1)
go func() {
defer pend.Done()
// Fetch and execute the next transaction trace tasks

for task := range jobs {
msg, _ := core.TransactionToMessage(txs[task.index], signer, block.BaseFee())
txctx := &Context{
Expand All @@ -720,7 +666,7 @@ func (api *API) traceBlockParallel(ctx context.Context, block *types.Block, stat
results[task.index] = &TxTraceResult{TxHash: txs[task.index].Hash(), Error: err.Error()}
continue
}
res, err := api.traceTx(ctx, txs[task.index], msg, txctx, blockCtx, task.statedb, config, nil)
res, err := api.traceTx(ctx, txs[task.index], msg, txctx, blockCtx, task.statedb, config, precompiles)
if err != nil {
results[task.index] = &TxTraceResult{TxHash: txs[task.index].Hash(), Error: err.Error()}
continue
Expand All @@ -732,11 +678,7 @@ func (api *API) traceBlockParallel(ctx context.Context, block *types.Block, stat

// Feed the transactions into the tracers and return
var failed error
blockCtx, err := api.backend.GetBlockContext(ctx, block, statedb, api.backend)
if err != nil {
return nil, err
}
evm := vm.NewEVM(blockCtx, statedb, api.backend.ChainConfig(), vm.Config{}, api.backend.GetCustomPrecompiles(block.Number().Int64()))
evm := vm.NewEVM(blockCtx, statedb, api.backend.ChainConfig(), vm.Config{}, precompiles)

txloop:
for i, tx := range txs {
Expand Down Expand Up @@ -771,6 +713,200 @@ txloop:
return results, nil
}

// traceBlockFullyParallel executes all transactions in a block with full parallelization.
// Unlike traceBlockParallel, this method parallelizes both the state generation and the
// tracing work, making it suitable for scenarios where maximum performance is needed.
// All ApplyMessage calls are executed in parallel with isolated state copies.
// traceBlockFullyParallel executes all transactions in a block with full parallelization.
// Unlike traceBlockParallel, this method parallelizes both the state generation and the
// tracing work, making it suitable for scenarios where maximum performance is needed.
// All ApplyMessage calls are executed in parallel with isolated state copies.
func (api *API) traceBlockFullyParallel(ctx context.Context, block *types.Block, statedb vm.StateDB, config *TraceConfig, metadata []tracersutils.TraceBlockMetadata) ([]*TxTraceResult, error) {
var (
txs = block.Transactions()
blockHash = block.Hash()
signer = types.MakeSigner(api.backend.ChainConfig(), block.Number(), block.Time())
results = make([]*TxTraceResult, len(txs))
pend sync.WaitGroup
)

// Get block context and precompiles once
blockCtx, err := api.backend.GetBlockContext(ctx, block, statedb, api.backend)
if err != nil {
return nil, err
}
precompiles := api.backend.GetCustomPrecompiles(block.Number().Int64())

// Handle case where no metadata is provided - trace all transactions
if len(metadata) == 0 {
// Number of worker threads
threads := runtime.NumCPU()
if threads > len(txs) {
threads = len(txs)
}

// Create a channel for transaction indices to process
jobs := make(chan int, len(txs))

// Launch worker goroutines
for th := 0; th < threads; th++ {
pend.Add(1)
go func() {
defer pend.Done()

for txIndex := range jobs {
// Create isolated state copy for this transaction
isolatedState := statedb.Copy()
tx := txs[txIndex]

// Create isolated EVM for building state
evm := vm.NewEVM(blockCtx, isolatedState, api.backend.ChainConfig(), vm.Config{}, precompiles)

// Process all transactions up to this index to build the correct state
for i := 0; i < txIndex; i++ {
prevTx := txs[i]
prevMsg, _ := core.TransactionToMessage(prevTx, signer, block.BaseFee())
isolatedState.SetTxContext(prevTx.Hash(), i)

if _, err := core.ApplyMessage(evm, prevMsg, new(core.GasPool).AddGas(prevMsg.GasLimit)); err != nil {
results[txIndex] = &TxTraceResult{TxHash: tx.Hash(), Error: err.Error()}
continue
}
isolatedState.Finalise(evm.ChainConfig().IsEIP158(block.Number()))
}

// Now trace the actual transaction
msg, _ := core.TransactionToMessage(tx, signer, block.BaseFee())
txctx := &Context{
BlockHash: blockHash,
BlockNumber: block.Number(),
TxIndex: txIndex,
TxHash: tx.Hash(),
}

// Reconstruct the block context for this transaction to avoid concurrent use issues
isolatedBlockCtx, err := api.backend.GetBlockContext(ctx, block, isolatedState, api.backend)
if err != nil {
results[txIndex] = &TxTraceResult{TxHash: tx.Hash(), Error: err.Error()}
continue
}

res, err := api.traceTx(ctx, tx, msg, txctx, isolatedBlockCtx, isolatedState, config, precompiles)
if err != nil {
results[txIndex] = &TxTraceResult{TxHash: tx.Hash(), Error: err.Error()}
continue
}
results[txIndex] = &TxTraceResult{TxHash: tx.Hash(), Result: res}
}
}()
}

// Queue all transaction indices for processing
for i := 0; i < len(txs); i++ {
select {
case <-ctx.Done():
close(jobs)
return nil, ctx.Err()
case jobs <- i:
}
}
close(jobs)

// Wait for all workers to complete
pend.Wait()

return results, nil
}

// Handle case where metadata is provided - only trace specific transactions
// First, collect indices of transactions that should be traced
var traceIndices []int
for _, md := range metadata {
if md.ShouldIncludeInTraceResult {
traceIndices = append(traceIndices, md.IdxInEthBlock)
}
}

// Number of worker threads
threads := runtime.NumCPU()
if threads > len(traceIndices) {
threads = len(traceIndices)
}

// Create a channel for transaction indices to process
jobs := make(chan int, len(traceIndices))

// Launch worker goroutines
for th := 0; th < threads; th++ {
pend.Add(1)
go func() {
defer pend.Done()

for txIndex := range jobs {
// Create isolated state copy for this transaction
isolatedState := statedb.Copy()
tx := txs[txIndex]

// Create isolated EVM for building state
evm := vm.NewEVM(blockCtx, isolatedState, api.backend.ChainConfig(), vm.Config{}, precompiles)

// Process all transactions up to this index to build the correct state
// We need to process ALL transactions (including those not traced) to maintain state consistency
for i := 0; i < txIndex; i++ {
prevTx := txs[i]
prevMsg, _ := core.TransactionToMessage(prevTx, signer, block.BaseFee())
isolatedState.SetTxContext(prevTx.Hash(), i)

if _, err := core.ApplyMessage(evm, prevMsg, new(core.GasPool).AddGas(prevMsg.GasLimit)); err != nil {
results[txIndex] = &TxTraceResult{TxHash: tx.Hash(), Error: err.Error()}
continue
}
isolatedState.Finalise(evm.ChainConfig().IsEIP158(block.Number()))
}

// Now trace the actual transaction
msg, _ := core.TransactionToMessage(tx, signer, block.BaseFee())
txctx := &Context{
BlockHash: blockHash,
BlockNumber: block.Number(),
TxIndex: txIndex,
TxHash: tx.Hash(),
}

// Reconstruct the block context for this transaction to avoid concurrent use issues
isolatedBlockCtx, err := api.backend.GetBlockContext(ctx, block, isolatedState, api.backend)
if err != nil {
results[txIndex] = &TxTraceResult{TxHash: tx.Hash(), Error: err.Error()}
continue
}

res, err := api.traceTx(ctx, tx, msg, txctx, isolatedBlockCtx, isolatedState, config, precompiles)
if err != nil {
results[txIndex] = &TxTraceResult{TxHash: tx.Hash(), Error: err.Error()}
continue
}
results[txIndex] = &TxTraceResult{TxHash: tx.Hash(), Result: res}
}
}()
}

// Queue transaction indices that should be traced
for _, txIndex := range traceIndices {
select {
case <-ctx.Done():
close(jobs)
return nil, ctx.Err()
case jobs <- txIndex:
}
}
close(jobs)

// Wait for all workers to complete
pend.Wait()

return results, nil
}

// standardTraceBlockToFile configures a new tracer which uses standard JSON output,
// and traces either a full block or an individual transaction. The return value will
// be one filename per transaction traced.
Expand Down
3 changes: 0 additions & 3 deletions go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,6 @@ require (
github.com/google/gofuzz v1.2.0
github.com/google/uuid v1.3.0
github.com/gorilla/websocket v1.4.2
github.com/graph-gophers/graphql-go v1.3.0
github.com/hashicorp/go-bexpr v0.1.10
github.com/holiman/billy v0.0.0-20240216141850-2abb0c79d3c4
github.com/holiman/bloomfilter/v2 v2.0.3
Expand Down Expand Up @@ -63,7 +62,6 @@ require (
github.com/syndtr/goleveldb v1.0.1-0.20210819022825-2ae1ddf74ef7
github.com/urfave/cli/v2 v2.27.5
go.uber.org/automaxprocs v1.5.2
go.uber.org/goleak v1.3.0
golang.org/x/crypto v0.35.0
golang.org/x/exp v0.0.0-20230626212559-97b1e661b5df
golang.org/x/sync v0.11.0
Expand Down Expand Up @@ -125,7 +123,6 @@ require (
github.com/mitchellh/pointerstructure v1.2.0 // indirect
github.com/mmcloughlin/addchain v0.4.0 // indirect
github.com/naoina/go-stringutil v0.1.0 // indirect
github.com/opentracing/opentracing-go v1.1.0 // indirect
github.com/pion/dtls/v2 v2.2.7 // indirect
github.com/pion/logging v0.2.2 // indirect
github.com/pion/transport/v2 v2.2.1 // indirect
Expand Down
6 changes: 0 additions & 6 deletions go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -288,8 +288,6 @@ github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5m
github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
github.com/graph-gophers/graphql-go v1.3.0 h1:Eb9x/q6MFpCLz7jBCiP/WTxjSDrYLR1QY41SORZyNJ0=
github.com/graph-gophers/graphql-go v1.3.0/go.mod h1:9CQHMSxwO4MprSdzoIEobiHpoLtHm77vfxsvsIN5Vuc=
github.com/hashicorp/go-bexpr v0.1.10 h1:9kuI5PFotCboP3dkDYFr/wi0gg0QVbSNz5oFRpxn4uE=
github.com/hashicorp/go-bexpr v0.1.10/go.mod h1:oxlubA2vC/gFVfX1A6JGp7ls7uCDlfJn732ehYYg+g0=
github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
Expand Down Expand Up @@ -410,8 +408,6 @@ github.com/onsi/ginkgo v1.14.0/go.mod h1:iSB4RoI2tjJc9BBv4NKIKWKya62Rps+oPG/Lv9k
github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY=
github.com/onsi/gomega v1.10.1 h1:o0+MgICZLuZ7xjH7Vx6zS/zcu93/BEp1VwkIW1mEXCE=
github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
github.com/opentracing/opentracing-go v1.1.0 h1:pWlfV3Bxv7k65HYwkikxat0+s3pV4bsqf19k25Ur8rU=
github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7 h1:oYW+YCJ1pachXTQmzR3rNLYGGz4g/UgFcjb28p/viDM=
github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7/go.mod h1:CRroGNssyjTd/qIG2FyxByd2S8JEAZXBl4qUrZf8GS0=
github.com/pingcap/errors v0.11.4 h1:lFuQV/oaUMGcD2tqt+01ROSmJs75VG1ToEOkZIZ4nE4=
Expand Down Expand Up @@ -528,8 +524,6 @@ go.opencensus.io v0.22.3/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
go.opencensus.io v0.22.4/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
go.uber.org/automaxprocs v1.5.2 h1:2LxUOGiR3O6tw8ui5sZa2LAaHnsviZdVOUZw4fvbnME=
go.uber.org/automaxprocs v1.5.2/go.mod h1:eRbA25aqJrxAbsLO0xy5jVwPt7FQnRgjW+efnwa1WM0=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
Expand Down