diff --git a/core/txpool/blobpool/blobpool.go b/core/txpool/blobpool/blobpool.go index d06b4bd21a..cca4eeb206 100644 --- a/core/txpool/blobpool/blobpool.go +++ b/core/txpool/blobpool/blobpool.go @@ -38,7 +38,6 @@ import ( "github.com/ethereum/go-ethereum/core/txpool" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto/kzg4844" - "github.com/ethereum/go-ethereum/eth/protocols/eth" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/metrics" @@ -291,7 +290,8 @@ func encodeForNetwork(storedRLP []byte, version uint) ([]byte, error) { // 5. Build the [blobs] field for the wire format. var blobsField []byte - if version >= eth.ETH72 { + // todo - Didn't use eth.ETH72 due to circular import error in test + if version >= 72 { // eth/72 omits the blob payload; peers fetch cells separately via GetCells. blobsField = []byte{0xc0} // RLP-encoded empty list } else { @@ -2241,6 +2241,8 @@ func (p *BlobPool) drop() { // // The transactions can also be pre-filtered by the dynamic fee components to // reduce allocations and load on downstream subsystems. +// The transactions without enough cells to recover thier blobs would be skipped +// in the response. func (p *BlobPool) Pending(filter txpool.PendingFilter) (map[common.Address][]*txpool.LazyTransaction, int) { // If only plain transactions are requested, this pool is unsuitable as it // contains none, don't even bother. @@ -2295,7 +2297,7 @@ func (p *BlobPool) Pending(filter txpool.PendingFilter) (map[common.Address][]*t } } // Skip transactions without enough cells to recover blobs - if tx.custody != nil && tx.custody.OneCount() < kzg4844.DataPerBlob { + if !filter.PartialCells && tx.custody != nil && tx.custody.OneCount() < kzg4844.DataPerBlob { break // not enough cells to build a full payload, discard rest of txs from the account } // Transaction was accepted according to the filter, append to the pending list diff --git a/core/txpool/blobpool/cache.go b/core/txpool/blobpool/cache.go index 325297a4e5..23f53f0fc1 100644 --- a/core/txpool/blobpool/cache.go +++ b/core/txpool/blobpool/cache.go @@ -46,12 +46,14 @@ var ( // missing on the lower level (in this case, the blobpool). The amount that // we failed to predict) can be calculated with the telemetry span // (blobs.filled - cache.hit). - cacheHitMeter = metrics.NewRegisteredMeter("blobpool/cache/hit", nil) - cacheMissMeter = metrics.NewRegisteredMeter("blobpool/cache/miss", nil) - cacheBlobsGauge = metrics.NewRegisteredGauge("blobpool/cache/blobs", nil) + cacheHitMeter = metrics.NewRegisteredMeter("blobpool/cache/hit", nil) + cacheMissMeter = metrics.NewRegisteredMeter("blobpool/cache/miss", nil) + cacheEntriesGauge = metrics.NewRegisteredGauge("blobpool/cache/entries", nil) ) type cachedBlob struct { + cell []kzg4844.Cell + custody types.CustodyBitmap blob *kzg4844.Blob commitment kzg4844.Commitment proofs []kzg4844.Proof @@ -74,11 +76,16 @@ type Cache struct { mu sync.Mutex entries map[common.Hash]*cachedBlob + // needCell is owned by the loop goroutine; it is only read and written + // there, so it needs no synchronization. It is flipped on via enableCellCh. + needCell bool + // channels into loop - quit chan struct{} - topkRequest chan struct{} - topkTimer mclock.Timer - hasBlobsCh chan []common.Hash // list of tx hashes that should be pinned + quit chan struct{} + topkRequest chan struct{} + topkTimer mclock.Timer + hasBlobsCh chan []common.Hash // list of tx hashes that should be pinned + enableCellCh chan struct{} // signals the loop to switch to cell mode step func() // test hook fired after each loop iteration @@ -96,13 +103,14 @@ func NewCache(p *BlobPool) *Cache { // It allows injecting a clock and a step hook. func newCache(p *BlobPool, clock mclock.Clock, step func()) *Cache { c := &Cache{ - entries: make(map[common.Hash]*cachedBlob), - blobpool: p, - hasBlobsCh: make(chan []common.Hash, 1), - clock: clock, - step: step, - quit: make(chan struct{}), - topkRequest: make(chan struct{}, 1), + entries: make(map[common.Hash]*cachedBlob), + blobpool: p, + hasBlobsCh: make(chan []common.Hash, 1), + clock: clock, + step: step, + quit: make(chan struct{}), + topkRequest: make(chan struct{}, 1), + enableCellCh: make(chan struct{}, 1), } c.wg.Add(1) @@ -197,13 +205,10 @@ func (c *Cache) GetBlobs(ctx context.Context, vhashes []common.Hash, version byt c.mu.Lock() for vhash, idxs := range indices { n := len(idxs) - - cached := c.entries[vhash] - if cached == nil || cached.version != version { + cached, ok := c.entries[vhash] + if !ok || cached.version != version || cached.blob == nil { cacheMiss += n - if cached == nil { - misses = append(misses, vhash) - } + misses = append(misses, vhash) continue } cacheHits += n @@ -241,6 +246,78 @@ func (c *Cache) GetBlobs(ctx context.Context, vhashes []common.Hash, version byt return blobs, commitments, proofs, nil } +func (c *Cache) GetCells(vhashes []common.Hash, mask types.CustodyBitmap) ([][]*kzg4844.Cell, [][]*kzg4844.Proof, error) { + var ( + cells = make([][]*kzg4844.Cell, len(vhashes)) + proofs = make([][]*kzg4844.Proof, len(vhashes)) + indices = make(map[common.Hash][]int) + misses []common.Hash + ) + for i, h := range vhashes { + indices[h] = append(indices[h], i) + } + requested := mask.Indices() + + c.mu.Lock() + for vhash, idxs := range indices { + cached, ok := c.entries[vhash] + if !ok { + misses = append(misses, vhash) + continue + } + stored := cached.custody.Indices() + blobCells := make([]*kzg4844.Cell, len(requested)) + blobProofs := make([]*kzg4844.Proof, len(requested)) + for i, cellIdx := range requested { + pos := -1 + for k, s := range stored { + if s == cellIdx { + pos = k + break + } + } + if pos >= 0 && pos < len(cached.cell) { + cell := cached.cell[pos] + blobCells[i] = &cell + if int(cellIdx) < len(cached.proofs) { + pf := cached.proofs[cellIdx] + blobProofs[i] = &pf + } + } + } + for _, idx := range idxs { + cells[idx] = blobCells + proofs[idx] = blobProofs + } + } + c.mu.Unlock() + + if len(misses) > 0 { + mc, mp, err := c.blobpool.GetBlobCells(misses, mask) + if err != nil { + return nil, nil, err + } + for j, vhash := range misses { + for _, idx := range indices[vhash] { + cells[idx] = mc[j] + proofs[idx] = mp[j] + } + } + } + return cells, proofs, nil +} + +// EnableCell allows the cache to store only cells without recovering +// blobs. This means we can also cache cells that lack enough blobs to +// recover. It signals the loop to switch to cell mode and re-select +// transactions from this wider pool. +func (c *Cache) EnableCell() { + select { + case c.enableCellCh <- struct{}{}: + case <-c.quit: + } +} + func (c *Cache) loop() { defer c.wg.Done() @@ -258,6 +335,16 @@ func (c *Cache) loop() { c.update(want) c.triggerTopKAfter(topKTimeout) + case <-c.enableCellCh: + // CL supports cell-based blob retrieval; switch to cell mode and + // re-select immediately over the now-wider pool. needCell is only + // touched here in the loop, so a fresh selection and update observe + // a consistent value. + if !c.needCell { + c.needCell = true + c.triggerTopK() + } + case <-c.quit: c.cancelUpdate() if c.topkTimer != nil { @@ -285,6 +372,8 @@ func (c *Cache) cancelUpdate() { // are no longer wanted and loads the missing ones from the blobpool in the // background. func (c *Cache) update(want []common.Hash) { + cellMode := c.needCell + wantSet := make(map[common.Hash]struct{}, len(want)) for _, vh := range want { wantSet[vh] = struct{}{} @@ -298,7 +387,13 @@ func (c *Cache) update(want []common.Hash) { c.mu.Lock() var missing []common.Hash for vh := range wantSet { - if _, ok := c.entries[vh]; !ok { + e, ok := c.entries[vh] + if ok && ((cellMode && e.cell == nil) || (!cellMode && e.blob == nil)) { + delete(c.entries, vh) + cacheEntriesGauge.Dec(1) + ok = false + } + if !ok { missing = append(missing, vh) } } @@ -307,7 +402,7 @@ func (c *Cache) update(want []common.Hash) { continue } delete(c.entries, vh) - cacheBlobsGauge.Dec(1) + cacheEntriesGauge.Dec(1) } c.mu.Unlock() @@ -330,44 +425,92 @@ func (c *Cache) update(want []common.Hash) { if ptx == nil { continue } - sidecar, err := ptx.sidecar() - if err != nil || sidecar == nil { - continue + if cellMode { + c.loadCells(ptx, wantSet) + } else { + c.loadBlobs(ptx, wantSet) } - - c.mu.Lock() - for i, v := range sidecar.BlobHashes() { - if _, ok := wantSet[v]; !ok { - continue - } - if _, exists := c.entries[v]; exists { - continue // recompute only new entries - } - var pf []kzg4844.Proof - switch sidecar.Version { - case types.BlobSidecarVersion0: - pf = []kzg4844.Proof{sidecar.Proofs[i]} - case types.BlobSidecarVersion1: - cellProofs, err := sidecar.CellProofsAt(i) - if err != nil { - log.Error("Failed to get cell proofs", "txhash", ptx.Tx.Hash(), "err", err) - continue - } - pf = cellProofs - } - c.entries[v] = &cachedBlob{ - blob: &sidecar.Blobs[i], - commitment: sidecar.Commitments[i], - proofs: pf, - version: sidecar.Version, - } - cacheBlobsGauge.Inc(1) - } - c.mu.Unlock() } }() } +// loadCells loads the cells whose vhash is in wantSet from ptx. +func (c *Cache) loadCells(ptx *BlobTxForPool, wantSet map[common.Hash]struct{}) { + cs := ptx.CellSidecar + cellsPerBlob := cs.Custody.OneCount() + c.mu.Lock() + defer c.mu.Unlock() + for i, v := range ptx.Tx.BlobHashes() { + if _, ok := wantSet[v]; !ok { + continue + } + if _, exists := c.entries[v]; exists { + continue + } + cellStart := i * cellsPerBlob + if cellStart+cellsPerBlob > len(cs.Cells) { + continue + } + blobCells := make([]kzg4844.Cell, cellsPerBlob) + copy(blobCells, cs.Cells[cellStart:cellStart+cellsPerBlob]) + + var pf []kzg4844.Proof + if ps := i * kzg4844.CellProofsPerBlob; ps+kzg4844.CellProofsPerBlob <= len(cs.Proofs) { + pf = make([]kzg4844.Proof, kzg4844.CellProofsPerBlob) + copy(pf, cs.Proofs[ps:ps+kzg4844.CellProofsPerBlob]) + } + c.entries[v] = &cachedBlob{ + cell: blobCells, + custody: cs.Custody, + commitment: cs.Commitments[i], + proofs: pf, + version: cs.Version, + } + cacheEntriesGauge.Inc(1) + } +} + +// loadBlobs loads the blobs whose vhash is in wantSet from ptx. +func (c *Cache) loadBlobs(ptx *BlobTxForPool, wantSet map[common.Hash]struct{}) { + if ptx.CellSidecar.Custody.OneCount() < kzg4844.DataPerBlob { + return + } + // blobs will be computed inside of sidecar() + sidecar, err := ptx.sidecar() + if err != nil || sidecar == nil { + return + } + c.mu.Lock() + defer c.mu.Unlock() + for i, v := range sidecar.BlobHashes() { + if _, ok := wantSet[v]; !ok { + continue + } + if _, exists := c.entries[v]; exists { + continue + } + var pf []kzg4844.Proof + switch sidecar.Version { + case types.BlobSidecarVersion0: + pf = []kzg4844.Proof{sidecar.Proofs[i]} + case types.BlobSidecarVersion1: + cellProofs, err := sidecar.CellProofsAt(i) + if err != nil { + log.Error("Failed to get cell proofs", "txhash", ptx.Tx.Hash(), "err", err) + continue + } + pf = cellProofs + } + c.entries[v] = &cachedBlob{ + blob: &sidecar.Blobs[i], + commitment: sidecar.Commitments[i], + proofs: pf, + version: sidecar.Version, + } + cacheEntriesGauge.Inc(1) + } +} + // selectTopTxs returns the vhashes of the top K most profitable pending blob // transactions, up to the active fork's maxBlobsPerBlock. func (c *Cache) selectTopTxs() []common.Hash { @@ -380,8 +523,9 @@ func (c *Cache) selectTopTxs() []common.Hash { baseFee := eip1559.CalcBaseFee(config, head) filter := txpool.PendingFilter{ - BlobTxs: true, - BaseFee: uint256.MustFromBig(baseFee), + BlobTxs: true, + BaseFee: uint256.MustFromBig(baseFee), + PartialCells: c.needCell, } if head.ExcessBlobGas != nil { filter.BlobFee = uint256.MustFromBig(eip4844.CalcBlobFee(config, head)) diff --git a/core/txpool/subpool.go b/core/txpool/subpool.go index 3c5f15e4bf..70a2d9300f 100644 --- a/core/txpool/subpool.go +++ b/core/txpool/subpool.go @@ -80,8 +80,9 @@ type PendingFilter struct { // When BlobTxs true, return only blob transactions (block blob-space filling) // when false, return only non-blob txs (peer-join announces, block space filling) - BlobTxs bool - BlobVersion byte // Blob tx version to include. 0 means pre-Osaka, 1 means Osaka and later + BlobTxs bool + PartialCells bool + BlobVersion byte // Blob tx version to include. 0 means pre-Osaka, 1 means Osaka and later } // TxMetadata denotes the metadata of a transaction. diff --git a/eth/catalyst/api.go b/eth/catalyst/api.go index 841bced1a3..f485633f0a 100644 --- a/eth/catalyst/api.go +++ b/eth/catalyst/api.go @@ -724,7 +724,7 @@ func (api *ConsensusAPI) GetBlobsV4(hashes []common.Hash, indicesBitarray types. if len(hashes) > 128 { return nil, engine.TooLargeRequest.With(fmt.Errorf("requested blob count too large: %v", len(hashes))) } - cells, proofs, err := api.eth.BlobTxPool().GetBlobCells(hashes, indicesBitarray) + cells, proofs, err := api.eth.BlobCache().GetCells(hashes, indicesBitarray) if err != nil { return nil, engine.InvalidParams.With(err) } @@ -1195,17 +1195,30 @@ func (api *ConsensusAPI) checkFork(timestamp uint64, forks ...forks.Fork) bool { } // ExchangeCapabilities returns the current methods provided by this node. -func (api *ConsensusAPI) ExchangeCapabilities([]string) []string { +func (api *ConsensusAPI) ExchangeCapabilities(caps []string) []string { valueT := reflect.TypeOf(api) - caps := make([]string, 0, valueT.NumMethod()) + + for _, cap := range caps { + if cap == "engine_getBlobsV4" { + // If the CL supports getBlobsV4, we call EnableCell() on the + // blob cache to skip the blob recovery process. This is a + // one-directional toggle, which assumes that once the CL + // supports getBlobsV4, it will not fall back to getBlobsV3 + // again. + api.eth.BlobCache().EnableCell() + break + } + } + + ourCaps := make([]string, 0, valueT.NumMethod()) for i := 0; i < valueT.NumMethod(); i++ { name := []rune(valueT.Method(i).Name) if string(name) == "ExchangeCapabilities" { continue } - caps = append(caps, "engine_"+string(unicode.ToLower(name[0]))+string(name[1:])) + ourCaps = append(ourCaps, "engine_"+string(unicode.ToLower(name[0]))+string(name[1:])) } - return caps + return ourCaps } // GetClientVersionV1 exchanges client version data of this node.