Skip to content
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.

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

eth/catalyst: move block commit into its own go-routine to avoid deadlock #29657

Closed
wants to merge 6 commits into from
Closed
Show file tree
Hide file tree
Changes from 5 commits
Commits
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
22 changes: 5 additions & 17 deletions eth/catalyst/api.go
Original file line number Diff line number Diff line change
Expand Up @@ -184,7 +184,7 @@ func (api *ConsensusAPI) ForkchoiceUpdatedV1(update engine.ForkchoiceStateV1, pa
return engine.STATUS_INVALID, engine.InvalidParams.With(errors.New("forkChoiceUpdateV1 called post-shanghai"))
}
}
return api.forkchoiceUpdated(update, payloadAttributes, engine.PayloadV1, false)
return api.forkchoiceUpdated(update, payloadAttributes, engine.PayloadV1)
}

// ForkchoiceUpdatedV2 is equivalent to V1 with the addition of withdrawals in the payload
Expand All @@ -207,7 +207,7 @@ func (api *ConsensusAPI) ForkchoiceUpdatedV2(update engine.ForkchoiceStateV1, pa
return engine.STATUS_INVALID, engine.UnsupportedFork.With(errors.New("forkchoiceUpdatedV2 must only be called with paris and shanghai payloads"))
}
}
return api.forkchoiceUpdated(update, params, engine.PayloadV2, false)
return api.forkchoiceUpdated(update, params, engine.PayloadV2)
}

// ForkchoiceUpdatedV3 is equivalent to V2 with the addition of parent beacon block root
Expand All @@ -228,10 +228,10 @@ func (api *ConsensusAPI) ForkchoiceUpdatedV3(update engine.ForkchoiceStateV1, pa
// hash, even if params are wrong. To do this we need to split up
// forkchoiceUpdate into a function that only updates the head and then a
// function that kicks off block construction.
return api.forkchoiceUpdated(update, params, engine.PayloadV3, false)
return api.forkchoiceUpdated(update, params, engine.PayloadV3)
}

func (api *ConsensusAPI) forkchoiceUpdated(update engine.ForkchoiceStateV1, payloadAttributes *engine.PayloadAttributes, payloadVersion engine.PayloadVersion, simulatorMode bool) (engine.ForkChoiceResponse, error) {
func (api *ConsensusAPI) forkchoiceUpdated(update engine.ForkchoiceStateV1, payloadAttributes *engine.PayloadAttributes, payloadVersion engine.PayloadVersion) (engine.ForkChoiceResponse, error) {
api.forkchoiceLock.Lock()
defer api.forkchoiceLock.Unlock()

Expand Down Expand Up @@ -374,19 +374,7 @@ func (api *ConsensusAPI) forkchoiceUpdated(update engine.ForkchoiceStateV1, payl
if api.localBlocks.has(id) {
return valid(&id), nil
}
// If the beacon chain is ran by a simulator, then transaction insertion,
// block insertion and block production will happen without any timing
// delay between them. This will cause flaky simulator executions due to
// the transaction pool running its internal reset operation on a back-
// ground thread. To avoid the racey behavior - in simulator mode - the
// pool will be explicitly blocked on its reset before continuing to the
// block production below.
if simulatorMode {
if err := api.eth.TxPool().Sync(); err != nil {
log.Error("Failed to sync transaction pool", "err", err)
return valid(nil), engine.InvalidPayloadAttributes.With(err)
}
}

payload, err := api.eth.Miner().BuildPayload(args)
if err != nil {
log.Error("Failed to build payload", "err", err)
Expand Down
2 changes: 1 addition & 1 deletion eth/catalyst/simulated_beacon.go
Original file line number Diff line number Diff line change
Expand Up @@ -164,7 +164,7 @@ func (c *SimulatedBeacon) sealBlock(withdrawals []*types.Withdrawal, timestamp u
Withdrawals: withdrawals,
Random: random,
BeaconRoot: &common.Hash{},
}, engine.PayloadV3, true)
}, engine.PayloadV3)
if err != nil {
return err
}
Expand Down
40 changes: 33 additions & 7 deletions eth/catalyst/simulated_beacon_api.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ package catalyst

import (
"context"
"sync"
"time"

"github.com/ethereum/go-ethereum/common"
Expand All @@ -32,8 +33,9 @@ type api struct {

func (a *api) loop() {
var (
newTxs = make(chan core.NewTxsEvent)
sub = a.sim.eth.TxPool().SubscribeTransactions(newTxs, true)
newTxs = make(chan core.NewTxsEvent)
sub = a.sim.eth.TxPool().SubscribeTransactions(newTxs, true)
commitMu = sync.Mutex{}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just fyi, the canonical way to initialize a variable with its zero value is just to declare it, i.e. you can write

var (
     commitMu sync.Mutex
)

Initializing with a zero literal is bad style.

)
defer sub.Unsubscribe()

Expand All @@ -42,12 +44,36 @@ func (a *api) loop() {
case <-a.sim.shutdownCh:
return
case w := <-a.sim.withdrawals.pending:
withdrawals := append(a.sim.withdrawals.gatherPending(9), w)
if err := a.sim.sealBlock(withdrawals, uint64(time.Now().Unix())); err != nil {
log.Warn("Error performing sealing work", "err", err)
}
go func() {
commitMu.Lock()
defer commitMu.Unlock()
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure if it matters, but this construction with the lock here will not preserve the receive order of select clauses. If both <-newTxs and <-a.sim.withdrawals.pending become ready to receive around the same time, they will both spawn new goroutines in quick succession. These goroutines will then begin to race for the lock, and it isn't guaranteed that the goroutine that got spawned first will also get the lock first.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It should be fine because eventually each go-routine will unblock and commit, making sure every pending new txs/withdrawals notification will trigger a commit.

// When the beacon chain is ran by a simulator, then transaction insertion,
// block insertion and block production will happen without any timing
// delay between them. This will cause flaky simulator executions due to
// the transaction pool running its internal reset operation on a back-
// ground thread. To avoid the racey behavior - in simulator mode - the
// pool will be explicitly blocked on its reset before continuing to the
// block production below.
if err := a.sim.eth.TxPool().Sync(); err != nil {
log.Error("Failed to sync transaction pool", "err", err)
return
}
withdrawals := append(a.sim.withdrawals.gatherPending(9), w)
if err := a.sim.sealBlock(withdrawals, uint64(time.Now().Unix())); err != nil {
log.Warn("Error performing sealing work", "err", err)
}
}()
case <-newTxs:
a.sim.Commit()
go func() {
commitMu.Lock()
defer commitMu.Unlock()

if err := a.sim.eth.TxPool().Sync(); err != nil {
log.Error("Failed to sync transaction pool", "err", err)
return
}
a.sim.Commit()
}()
}
}
}
Expand Down
80 changes: 77 additions & 3 deletions eth/catalyst/simulated_beacon_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ import (
"github.com/ethereum/go-ethereum/params"
)

func startSimulatedBeaconEthService(t *testing.T, genesis *core.Genesis) (*node.Node, *eth.Ethereum, *SimulatedBeacon) {
func startSimulatedBeaconEthService(t *testing.T, genesis *core.Genesis, period uint64) (*node.Node, *eth.Ethereum, *SimulatedBeacon) {
t.Helper()

n, err := node.New(&node.Config{
Expand All @@ -55,7 +55,7 @@ func startSimulatedBeaconEthService(t *testing.T, genesis *core.Genesis) (*node.
t.Fatal("can't create eth service:", err)
}

simBeacon, err := NewSimulatedBeacon(1, ethservice)
simBeacon, err := NewSimulatedBeacon(period, ethservice)
if err != nil {
t.Fatal("can't create simulated beacon:", err)
}
Expand Down Expand Up @@ -87,7 +87,7 @@ func TestSimulatedBeaconSendWithdrawals(t *testing.T) {
// short period (1 second) for testing purposes
var gasLimit uint64 = 10_000_000
genesis := core.DeveloperGenesisBlock(gasLimit, &testAddr)
node, ethService, mock := startSimulatedBeaconEthService(t, genesis)
node, ethService, mock := startSimulatedBeaconEthService(t, genesis, 1)
_ = mock
defer node.Close()

Expand Down Expand Up @@ -140,3 +140,77 @@ func TestSimulatedBeaconSendWithdrawals(t *testing.T) {
}
}
}

// Tests that zero-period dev mode can handle a lot of simultaneous
// transactions/withdrawals
func TestOnDemandSpam(t *testing.T) {
var withdrawals []types.Withdrawal
txs := make(map[common.Hash]*types.Transaction)

var (
// testKey is a private key to use for funding a tester account.
testKey, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")

// testAddr is the Ethereum address of the tester account.
testAddr = crypto.PubkeyToAddress(testKey.PublicKey)
)

// short period (1 second) for testing purposes
var gasLimit uint64 = 10_000_000
genesis := core.DeveloperGenesisBlock(gasLimit, &testAddr)
node, ethService, mock := startSimulatedBeaconEthService(t, genesis, 0)
_ = mock
defer node.Close()

api := &api{mock}
go api.loop()

chainHeadCh := make(chan core.ChainHeadEvent, 10)
subscription := ethService.BlockChain().SubscribeChainHeadEvent(chainHeadCh)
defer subscription.Unsubscribe()

// generate some withdrawals
for i := 0; i < 0; i++ {
withdrawals = append(withdrawals, types.Withdrawal{Index: uint64(i)})
if err := mock.withdrawals.add(&withdrawals[i]); err != nil {
t.Fatal("addWithdrawal failed", err)
}
}

// generate a bunch of transactions
signer := types.NewEIP155Signer(ethService.BlockChain().Config().ChainID)
for i := 0; i < 20000; i++ {
tx, err := types.SignTx(types.NewTransaction(uint64(i), common.Address{}, big.NewInt(1000), params.TxGas, big.NewInt(params.InitialBaseFee), nil), signer, testKey)
if err != nil {
t.Fatalf("error signing transaction, err=%v", err)
}
txs[tx.Hash()] = tx

if err := ethService.APIBackend.SendTx(context.Background(), tx); err != nil {
t.Fatal("SendTx failed", err)
}
}

includedTxs := make(map[common.Hash]struct{})
var includedWithdrawals []uint64

timer := time.NewTimer(20 * time.Second)
for {
select {
case evt := <-chainHeadCh:
for _, includedTx := range evt.Block.Transactions() {
includedTxs[includedTx.Hash()] = struct{}{}
}
for _, includedWithdrawal := range evt.Block.Withdrawals() {
includedWithdrawals = append(includedWithdrawals, includedWithdrawal.Index)
}

// ensure all withdrawals/txs included. this will take two blocks b/c number of withdrawals > 10
if len(includedTxs) == len(txs) && len(includedWithdrawals) == len(withdrawals) {
return
}
case <-timer.C:
t.Fatal("timed out without including all withdrawals/txs")
}
}
}