Add bundle merging with multiple workers

.github/workflows/go.yml

@@ -53,6 +53,7 @@ jobs:
       uses: actions/checkout@v2
       with:
         repository: flashbots/mev-geth-demo
+        ref: jason/v0.2
         path: e2e
 
     - run: cd e2e && yarn install

cmd/geth/main.go

@@ -125,6 +125,7 @@ var (
 		utils.MinerExtraDataFlag,
 		utils.MinerRecommitIntervalFlag,
 		utils.MinerNoVerfiyFlag,
+		utils.MinerMaxMergedBundles,
 		utils.NATFlag,
 		utils.NoDiscoverFlag,
 		utils.DiscoveryV5Flag,

cmd/geth/usage.go

@@ -188,6 +188,7 @@ var AppHelpFlagGroups = []flags.FlagGroup{
 			utils.MinerExtraDataFlag,
 			utils.MinerRecommitIntervalFlag,
 			utils.MinerNoVerfiyFlag,
+			utils.MinerMaxMergedBundles,
 		},
 	},
 	{

cmd/utils/flags.go

@@ -463,6 +463,11 @@ var (
 		Usage: "Time interval to recreate the block being mined",
 		Value: ethconfig.Defaults.Miner.Recommit,
 	}
+	MinerMaxMergedBundles = cli.IntFlag{
+		Name:  "miner.maxmergedbundles",
+		Usage: "flashbots - The maximum amount of bundles to merge. The miner will run this many workers in parallel to calculate if the full block is more profitable with these additional bundles.",
+		Value: 3,
+	}
 	MinerNoVerfiyFlag = cli.BoolFlag{
 		Name:  "miner.noverify",
 		Usage: "Disable remote sealing verification",
@@ -1391,6 +1396,8 @@ func setMiner(ctx *cli.Context, cfg *miner.Config) {
 	if ctx.GlobalIsSet(MinerNoVerfiyFlag.Name) {
 		cfg.Noverify = ctx.GlobalBool(MinerNoVerfiyFlag.Name)
 	}
+
+	cfg.MaxMergedBundles = ctx.GlobalInt(MinerMaxMergedBundles.Name)
 }
 
 func setWhitelist(ctx *cli.Context, cfg *ethconfig.Config) {

internal/ethapi/api.go

@@ -2145,7 +2145,7 @@ func (s *PrivateTxBundleAPI) SendBundle(ctx context.Context, args SendBundleArgs
 
 	for _, encodedTx := range args.Txs {
 		tx := new(types.Transaction)
-		if err := rlp.DecodeBytes(encodedTx, tx); err != nil {
+		if err := tx.UnmarshalBinary(encodedTx); err != nil {
 			return err
 		}
 		txs = append(txs, tx)

miner/miner.go

@@ -42,15 +42,16 @@ type Backend interface {
 
 // Config is the configuration parameters of mining.
 type Config struct {
-	Etherbase  common.Address `toml:",omitempty"` // Public address for block mining rewards (default = first account)
-	Notify     []string       `toml:",omitempty"` // HTTP URL list to be notified of new work packages (only useful in ethash).
-	NotifyFull bool           `toml:",omitempty"` // Notify with pending block headers instead of work packages
-	ExtraData  hexutil.Bytes  `toml:",omitempty"` // Block extra data set by the miner
-	GasFloor   uint64         // Target gas floor for mined blocks.
-	GasCeil    uint64         // Target gas ceiling for mined blocks.
-	GasPrice   *big.Int       // Minimum gas price for mining a transaction
-	Recommit   time.Duration  // The time interval for miner to re-create mining work.
-	Noverify   bool           // Disable remote mining solution verification(only useful in ethash).
+	Etherbase        common.Address `toml:",omitempty"` // Public address for block mining rewards (default = first account)
+	Notify           []string       `toml:",omitempty"` // HTTP URL list to be notified of new work packages (only useful in ethash).
+	NotifyFull       bool           `toml:",omitempty"` // Notify with pending block headers instead of work packages
+	ExtraData        hexutil.Bytes  `toml:",omitempty"` // Block extra data set by the miner
+	GasFloor         uint64         // Target gas floor for mined blocks.
+	GasCeil          uint64         // Target gas ceiling for mined blocks.
+	GasPrice         *big.Int       // Minimum gas price for mining a transaction
+	Recommit         time.Duration  // The time interval for miner to re-create mining work.
+	Noverify         bool           // Disable remote mining solution verification(only useful in ethash).
+	MaxMergedBundles int
 }
 
 // Miner creates blocks and searches for proof-of-work values.

miner/multi_worker.go

@@ -7,74 +7,100 @@ import (
 	"github.com/ethereum/go-ethereum/consensus"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/params"
 )
 
 type multiWorker struct {
-	regularWorker   *worker
-	flashbotsWorker *worker
+	workers       []*worker
+	regularWorker *worker
 }
 
 func (w *multiWorker) stop() {
-	w.regularWorker.stop()
-	w.flashbotsWorker.stop()
+	for _, worker := range w.workers {
+		worker.stop()
+	}
 }
 
 func (w *multiWorker) start() {
-	w.regularWorker.start()
-	w.flashbotsWorker.start()
+	for _, worker := range w.workers {
+		worker.start()
+	}
 }
 
 func (w *multiWorker) close() {
-	w.regularWorker.close()
-	w.flashbotsWorker.close()
+	for _, worker := range w.workers {
+		worker.close()
+	}
 }
 
 func (w *multiWorker) isRunning() bool {
-	return w.regularWorker.isRunning() || w.flashbotsWorker.isRunning()
+	for _, worker := range w.workers {
+		if worker.isRunning() {
+			return true
+		}
+	}
+	return false
 }
 
 func (w *multiWorker) setExtra(extra []byte) {
-	w.regularWorker.setExtra(extra)
-	w.flashbotsWorker.setExtra(extra)
+	for _, worker := range w.workers {
+		worker.setExtra(extra)
+	}
 }
 
 func (w *multiWorker) setRecommitInterval(interval time.Duration) {
-	w.regularWorker.setRecommitInterval(interval)
-	w.flashbotsWorker.setRecommitInterval(interval)
+	for _, worker := range w.workers {
+		worker.setRecommitInterval(interval)
+	}
 }
 
 func (w *multiWorker) setEtherbase(addr common.Address) {
-	w.regularWorker.setEtherbase(addr)
-	w.flashbotsWorker.setEtherbase(addr)
+	for _, worker := range w.workers {
+		worker.setEtherbase(addr)
+	}
 }
 
 func (w *multiWorker) enablePreseal() {
-	w.regularWorker.enablePreseal()
-	w.flashbotsWorker.enablePreseal()
+	for _, worker := range w.workers {
+		worker.enablePreseal()
+	}
 }
 
 func (w *multiWorker) disablePreseal() {
-	w.regularWorker.disablePreseal()
-	w.flashbotsWorker.disablePreseal()
+	for _, worker := range w.workers {
+		worker.disablePreseal()
+	}
 }
 
 func newMultiWorker(config *Config, chainConfig *params.ChainConfig, engine consensus.Engine, eth Backend, mux *event.TypeMux, isLocalBlock func(*types.Block) bool, init bool) *multiWorker {
 	queue := make(chan *task)
 
+	regularWorker := newWorker(config, chainConfig, engine, eth, mux, isLocalBlock, init, &flashbotsData{
+		isFlashbots: false,
+		queue:       queue,
+	})
+
+	workers := []*worker{regularWorker}
+
+	for i := 1; i <= config.MaxMergedBundles; i++ {
+		workers = append(workers,
+			newWorker(config, chainConfig, engine, eth, mux, isLocalBlock, init, &flashbotsData{
+				isFlashbots:      true,
+				queue:            queue,
+				maxMergedBundles: i,
+			}))
+	}
+
+	log.Info("creating multi worker", "config.MaxMergedBundles", config.MaxMergedBundles, "worker", len(workers))
 	return &multiWorker{
-		regularWorker: newWorker(config, chainConfig, engine, eth, mux, isLocalBlock, init, &flashbotsData{
-			isFlashbots: false,
-			queue:       queue,
-		}),
-		flashbotsWorker: newWorker(config, chainConfig, engine, eth, mux, isLocalBlock, init, &flashbotsData{
-			isFlashbots: true,
-			queue:       queue,
-		}),
+		regularWorker: regularWorker,
+		workers:       workers,
 	}
 }
 
 type flashbotsData struct {
-	isFlashbots bool
-	queue       chan *task
+	isFlashbots      bool
+	queue            chan *task
+	maxMergedBundles int
 }