diff --git a/eth/catalyst/simulated_beacon.go b/eth/catalyst/simulated_beacon.go index 8bdf94b80e..052dc7e237 100644 --- a/eth/catalyst/simulated_beacon.go +++ b/eth/catalyst/simulated_beacon.go @@ -30,6 +30,7 @@ import ( "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto/kzg4844" "github.com/ethereum/go-ethereum/eth" + "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/params" @@ -41,36 +42,47 @@ const devEpochLength = 32 // withdrawalQueue implements a FIFO queue which holds withdrawals that are // pending inclusion. type withdrawalQueue struct { - pending chan *types.Withdrawal + pending types.Withdrawals + mu sync.Mutex + feed event.Feed + subs event.SubscriptionScope } +type newWithdrawalsEvent struct{ Withdrawals types.Withdrawals } + // add queues a withdrawal for future inclusion. -func (w *withdrawalQueue) add(withdrawal *types.Withdrawal) error { - select { - case w.pending <- withdrawal: - break - default: - return errors.New("withdrawal queue full") - } +func (w *withdrawalQueue) Add(withdrawal *types.Withdrawal) error { + w.mu.Lock() + defer w.mu.Unlock() + + w.pending = append(w.pending, withdrawal) + w.feed.Send(newWithdrawalsEvent{types.Withdrawals{withdrawal}}) + return nil } -// gatherPending returns a number of queued withdrawals up to a maximum count. -func (w *withdrawalQueue) gatherPending(maxCount int) []*types.Withdrawal { - withdrawals := []*types.Withdrawal{} - for { - select { - case withdrawal := <-w.pending: - withdrawals = append(withdrawals, withdrawal) - if len(withdrawals) == maxCount { - return withdrawals - } - default: - return withdrawals - } - } +// pop dequeues the specified number of withdrawals from the queue. +func (w *withdrawalQueue) Pop(count int) types.Withdrawals { + w.mu.Lock() + defer w.mu.Unlock() + + count = min(count, len(w.pending)) + popped := w.pending[0:count] + w.pending = w.pending[count:] + + return popped } +// subscribe allows a listener to be updated when new withdrawals are added to +// the queue. +func (w *withdrawalQueue) Subscribe(ch chan<- newWithdrawalsEvent) event.Subscription { + sub := w.feed.Subscribe(ch) + return w.subs.Track(sub) +} + +// SimulatedBeacon drives an Ethereum instance as if it were a real beacon +// client. It can run in period mode where it mines a new block every period +// (seconds) or on every transaction via Commit, Fork and AdjustTime. type SimulatedBeacon struct { shutdownCh chan struct{} eth *eth.Ethereum @@ -86,10 +98,6 @@ type SimulatedBeacon struct { } // NewSimulatedBeacon constructs a new simulated beacon chain. -// Period sets the period in which blocks should be produced. -// -// - If period is set to 0, a block is produced on every transaction. -// via Commit, Fork and AdjustTime. func NewSimulatedBeacon(period uint64, eth *eth.Ethereum) (*SimulatedBeacon, error) { block := eth.BlockChain().CurrentBlock() current := engine.ForkchoiceStateV1{ @@ -112,7 +120,6 @@ func NewSimulatedBeacon(period uint64, eth *eth.Ethereum) (*SimulatedBeacon, err engineAPI: engineAPI, lastBlockTime: block.Time, curForkchoiceState: current, - withdrawals: withdrawalQueue{make(chan *types.Withdrawal, 20)}, }, nil } @@ -171,6 +178,7 @@ func (c *SimulatedBeacon) sealBlock(withdrawals []*types.Withdrawal, timestamp u if fcResponse == engine.STATUS_SYNCING { return errors.New("chain rewind prevented invocation of payload creation") } + envelope, err := c.engineAPI.getPayload(*fcResponse.PayloadID, true) if err != nil { return err @@ -223,8 +231,7 @@ func (c *SimulatedBeacon) loop() { case <-c.shutdownCh: return case <-timer.C: - withdrawals := c.withdrawals.gatherPending(10) - if err := c.sealBlock(withdrawals, uint64(time.Now().Unix())); err != nil { + if err := c.sealBlock(c.withdrawals.Pop(10), uint64(time.Now().Unix())); err != nil { log.Warn("Error performing sealing work", "err", err) } else { timer.Reset(time.Second * time.Duration(c.period)) @@ -260,7 +267,7 @@ func (c *SimulatedBeacon) setCurrentState(headHash, finalizedHash common.Hash) { // Commit seals a block on demand. func (c *SimulatedBeacon) Commit() common.Hash { - withdrawals := c.withdrawals.gatherPending(10) + withdrawals := c.withdrawals.Pop(10) if err := c.sealBlock(withdrawals, uint64(time.Now().Unix())); err != nil { log.Warn("Error performing sealing work", "err", err) } @@ -301,12 +308,14 @@ func (c *SimulatedBeacon) AdjustTime(adjustment time.Duration) error { if parent == nil { return errors.New("parent not found") } - withdrawals := c.withdrawals.gatherPending(10) + withdrawals := c.withdrawals.Pop(10) return c.sealBlock(withdrawals, parent.Time+uint64(adjustment/time.Second)) } +// RegisterSimulatedBeaconAPIs registers the simulated beacon's API with the +// stack. func RegisterSimulatedBeaconAPIs(stack *node.Node, sim *SimulatedBeacon) { - api := &api{sim} + api := &simulatedBeaconAPI{sim: sim, doCommit: make(chan struct{}, 1)} if sim.period == 0 { // mine on demand if period is set to 0 go api.loop() diff --git a/eth/catalyst/simulated_beacon_api.go b/eth/catalyst/simulated_beacon_api.go index 73d0a5921d..b26316f858 100644 --- a/eth/catalyst/simulated_beacon_api.go +++ b/eth/catalyst/simulated_beacon_api.go @@ -18,44 +18,77 @@ package catalyst import ( "context" - "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" - "github.com/ethereum/go-ethereum/log" ) -type api struct { - sim *SimulatedBeacon +// simulatedBeaconAPI provides a RPC API for SimulatedBeacon. +type simulatedBeaconAPI struct { + sim *SimulatedBeacon + doCommit chan struct{} } -func (a *api) loop() { +// loop is the main loop for the API when it's running in period = 0 mode. It +// ensures that block production is triggered as soon as a new withdrawal or +// transaction is received. +func (a *simulatedBeaconAPI) loop() { var ( - newTxs = make(chan core.NewTxsEvent) - sub = a.sim.eth.TxPool().SubscribeTransactions(newTxs, true) + newTxs = make(chan core.NewTxsEvent) + newWxs = make(chan newWithdrawalsEvent) + newTxsSub = a.sim.eth.TxPool().SubscribeTransactions(newTxs, true) + newWxsSub = a.sim.withdrawals.Subscribe(newWxs) ) - defer sub.Unsubscribe() + defer newTxsSub.Unsubscribe() + defer newWxsSub.Unsubscribe() + + go a.worker() for { select { 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) - } + case <-newWxs: + a.commit() case <-newTxs: - a.sim.Commit() + a.commit() } } } -func (a *api) AddWithdrawal(ctx context.Context, withdrawal *types.Withdrawal) error { - return a.sim.withdrawals.add(withdrawal) +// commit is a non-blocking method to initate Commit() on the simulator. +func (a *simulatedBeaconAPI) commit() { + select { + case a.doCommit <- struct{}{}: + default: + } } -func (a *api) SetFeeRecipient(ctx context.Context, feeRecipient common.Address) { +// worker runs in the background and signals to the simulator when to commit +// based on messages over doCommit. +func (a *simulatedBeaconAPI) worker() { + for { + select { + case <-a.sim.shutdownCh: + return + case <-a.doCommit: + a.sim.Commit() + a.sim.eth.TxPool().Sync() + executable, _ := a.sim.eth.TxPool().Stats() + if executable != 0 { + a.commit() + } + } + } +} + +// AddWithdrawal adds a withdrawal to the pending queue. +func (a *simulatedBeaconAPI) AddWithdrawal(ctx context.Context, withdrawal *types.Withdrawal) error { + return a.sim.withdrawals.Add(withdrawal) +} + +// SetFeeRecipient sets the fee recipient for block building purposes. +func (a *simulatedBeaconAPI) SetFeeRecipient(ctx context.Context, feeRecipient common.Address) { a.sim.setFeeRecipient(feeRecipient) } diff --git a/eth/catalyst/simulated_beacon_test.go b/eth/catalyst/simulated_beacon_test.go index bb10938c35..723338a1af 100644 --- a/eth/catalyst/simulated_beacon_test.go +++ b/eth/catalyst/simulated_beacon_test.go @@ -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{ @@ -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) } @@ -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() @@ -98,7 +98,7 @@ func TestSimulatedBeaconSendWithdrawals(t *testing.T) { // generate some withdrawals for i := 0; i < 20; i++ { withdrawals = append(withdrawals, types.Withdrawal{Index: uint64(i)}) - if err := mock.withdrawals.add(&withdrawals[i]); err != nil { + if err := mock.withdrawals.Add(&withdrawals[i]); err != nil { t.Fatal("addWithdrawal failed", err) } } @@ -140,3 +140,68 @@ 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) + testKey, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") + testAddr = crypto.PubkeyToAddress(testKey.PublicKey) + gasLimit uint64 = 10_000_000 + genesis = core.DeveloperGenesisBlock(gasLimit, &testAddr) + node, eth, mock = startSimulatedBeaconEthService(t, genesis, 0) + signer = types.LatestSigner(eth.BlockChain().Config()) + chainHeadCh = make(chan core.ChainHeadEvent, 100) + sub = eth.BlockChain().SubscribeChainHeadEvent(chainHeadCh) + ) + defer node.Close() + defer sub.Unsubscribe() + + // start simulated beacon + api := &simulatedBeaconAPI{sim: mock, doCommit: make(chan struct{}, 1)} + go api.loop() + + // generate some withdrawals + for i := 0; i < 20; 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 + for i := 0; i < 20000; i++ { + tx, err := types.SignTx(types.NewTransaction(uint64(i), common.Address{byte(i), byte(1)}, big.NewInt(1000), params.TxGas, big.NewInt(params.InitialBaseFee*2), nil), signer, testKey) + if err != nil { + t.Fatal("error signing transaction", err) + } + txs[tx.Hash()] = tx + if err := eth.APIBackend.SendTx(context.Background(), tx); err != nil { + t.Fatal("error adding txs to pool", err) + } + } + + var ( + includedTxs = make(map[common.Hash]struct{}) + includedWxs []uint64 + ) + for { + select { + case evt := <-chainHeadCh: + for _, itx := range evt.Block.Transactions() { + includedTxs[itx.Hash()] = struct{}{} + } + for _, iwx := range evt.Block.Withdrawals() { + includedWxs = append(includedWxs, iwx.Index) + } + // ensure all withdrawals/txs included. this will take two blocks b/c number of withdrawals > 10 + if len(includedTxs) == len(txs) && len(includedWxs) == len(withdrawals) { + return + } + case <-time.After(10 * time.Second): + t.Fatalf("timed out without including all withdrawals/txs: have txs %d, want %d, have wxs %d, want %d", len(includedTxs), len(txs), len(includedWxs), len(withdrawals)) + } + } +}