No RLP decoding of Bodies during the sync (fix #8242)

CHANGELOG.md

@@ -14,6 +14,7 @@
 - Add option to trace reference tests during execution [#8878](https://github.com/hyperledger/besu/pull/8878)
 - Expose methods to query hardfork by block header or for the next block in the Plugin API [#8909](https://github.com/hyperledger/besu/pull/8909)
 - Generate distribution dependencies catalog [#8987](https://github.com/hyperledger/besu/pull/8987)
+- Improve the sync performance by not RLP decoding bodies during sync. This means we are using less memory and CPU, allowing us to increase the parallelism of the download pipeline, which has been increased from 4 to 8. Can be reduced again with  `--Xsynchronizer-downloader-parallelism=4` [#8959]
 
 #### Fusaka devnets
 - EIP-7910 - `eth_config` JSON-RPC Method [#8417](https://github.com/hyperledger/besu/pull/8417), [#8946](https://github.com/hyperledger/besu/pull/8946)

app/src/main/java/org/hyperledger/besu/RunnerBuilder.java

@@ -721,7 +721,7 @@ public Runner build() {
         .getBlockchain()
         .observeBlockAdded(
             blockAddedEvent -> {
-              if (protocolSchedule.isOnMilestoneBoundary(blockAddedEvent.getBlock().getHeader())) {
+              if (protocolSchedule.isOnMilestoneBoundary(blockAddedEvent.getHeader())) {
                 network.updateNodeRecord();
               }
             });

app/src/main/java/org/hyperledger/besu/controller/CliqueBesuControllerBuilder.java

@@ -114,7 +114,7 @@ protected MiningCoordinator createMiningCoordinator(
             o ->
                 miningConfiguration.setBlockPeriodSeconds(
                     forksSchedule
-                        .getFork(o.getBlock().getHeader().getNumber() + 1)
+                        .getFork(o.getHeader().getNumber() + 1)
                         .getValue()
                         .getBlockPeriodSeconds()));
 

app/src/main/java/org/hyperledger/besu/controller/IbftBesuControllerBuilder.java

@@ -247,7 +247,7 @@ protected MiningCoordinator createMiningCoordinator(
             o ->
                 miningConfiguration.setBlockPeriodSeconds(
                     forksSchedule
-                        .getFork(o.getBlock().getHeader().getNumber() + 1)
+                        .getFork(o.getHeader().getNumber() + 1)
                         .getValue()
                         .getBlockPeriodSeconds()));
 

app/src/main/java/org/hyperledger/besu/controller/MergeBesuControllerBuilder.java

@@ -224,7 +224,7 @@ protected MergeContext createConsensusContext(
     blockchain.observeBlockAdded(
         blockAddedEvent ->
             blockchain
-                .getTotalDifficultyByHash(blockAddedEvent.getBlock().getHeader().getHash())
+                .getTotalDifficultyByHash(blockAddedEvent.getHeader().getHash())
                 .ifPresent(mergeContext::setIsPostMerge));
 
     return mergeContext;

app/src/main/java/org/hyperledger/besu/controller/QbftBesuControllerBuilder.java

@@ -332,12 +332,12 @@ protected MiningCoordinator createMiningCoordinator(
             o -> {
               miningConfiguration.setBlockPeriodSeconds(
                   qbftForksSchedule
-                      .getFork(o.getBlock().getHeader().getNumber() + 1)
+                      .getFork(o.getHeader().getNumber() + 1)
                       .getValue()
                       .getBlockPeriodSeconds());
               miningConfiguration.setEmptyBlockPeriodSeconds(
                   qbftForksSchedule
-                      .getFork(o.getBlock().getHeader().getNumber() + 1)
+                      .getFork(o.getHeader().getNumber() + 1)
                       .getValue()
                       .getEmptyBlockPeriodSeconds());
             });

app/src/main/java/org/hyperledger/besu/services/BesuEventsImpl.java

@@ -88,8 +88,8 @@ public long addBlockAddedListener(final BlockAddedListener listener) {
         event ->
             listener.onBlockAdded(
                 blockAddedContext(
-                    event.getBlock()::getHeader,
-                    event.getBlock()::getBody,
+                    event::getHeader,
+                    () -> event.getBlock().getBody(),
                     event::getTransactionReceipts)));
   }
 

consensus/common/src/main/java/org/hyperledger/besu/consensus/common/MigratingMiningCoordinator.java

@@ -132,7 +132,7 @@ public void changeTargetGasLimit(final Long targetGasLimit) {
 
   @Override
   public void onBlockAdded(final BlockAddedEvent event) {
-    final long currentBlock = event.getBlock().getHeader().getNumber();
+    final long currentBlock = event.getHeader().getNumber();
     final MiningCoordinator nextMiningCoordinator =
         miningCoordinatorSchedule.getFork(currentBlock + 1).getValue();
 

consensus/common/src/main/java/org/hyperledger/besu/consensus/common/bft/blockcreation/BftMiningCoordinator.java

@@ -258,7 +258,7 @@ public void changeTargetGasLimit(final Long targetGasLimit) {
   public void onBlockAdded(final BlockAddedEvent event) {
     if (event.isNewCanonicalHead()) {
       LOG.trace("New canonical head detected");
-      eventQueue.add(new NewChainHead(event.getBlock().getHeader()));
+      eventQueue.add(new NewChainHead(event.getHeader()));
     }
   }
 

consensus/merge/src/test/java/org/hyperledger/besu/consensus/merge/blockcreation/MergeCoordinatorTest.java

@@ -905,7 +905,8 @@ public void onBlockAdded(final BlockAddedEvent event) {
         prevParent, genesisState.getBlock().getHash(), genesisState.getBlock().getHash());
     assertThat(lastBlockAddedEvent.get().getCommonAncestorHash()).isEqualTo(expectedCommonAncestor);
     assertThat(lastBlockAddedEvent.get().getEventType()).isEqualTo(EventType.CHAIN_REORG);
-    assertThat(lastBlockAddedEvent.get().getBlock().getHash()).isEqualTo(prevParent.getBlockHash());
+    assertThat(lastBlockAddedEvent.get().getHeader().getHash())
+        .isEqualTo(prevParent.getBlockHash());
   }
 
   @Test

consensus/merge/src/test/java/org/hyperledger/besu/consensus/merge/blockcreation/MergeReorgTest.java

@@ -109,7 +109,7 @@ public void setUp() {
     blockchain.observeBlockAdded(
         blockAddedEvent ->
             blockchain
-                .getTotalDifficultyByHash(blockAddedEvent.getBlock().getHeader().getHash())
+                .getTotalDifficultyByHash(blockAddedEvent.getHeader().getHash())
                 .ifPresent(mergeContext::setIsPostMerge));
   }
 

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/filter/FilterManager.java

@@ -127,7 +127,7 @@ public boolean uninstallFilter(final String filterId) {
   }
 
   public void recordBlockEvent(final BlockAddedEvent event) {
-    final Hash blockHash = event.getBlock().getHash();
+    final Hash blockHash = event.getHeader().getHash();
     final Collection<BlockFilter> blockFilters =
         filterRepository.getFiltersOfType(BlockFilter.class);
     blockFilters.forEach(
@@ -144,7 +144,7 @@ public void recordBlockEvent(final BlockAddedEvent event) {
             filter -> {
               final Optional<Long> maybeToBlockNumber = filter.getToBlock().getNumber();
               return maybeToBlockNumber.isEmpty()
-                  || maybeToBlockNumber.get() >= event.getBlock().getHeader().getNumber();
+                  || maybeToBlockNumber.get() >= event.getHeader().getNumber();
             })
         .forEach(
             filter -> {

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/websocket/subscription/blockheaders/NewBlockHeadersSubscriptionService.java

@@ -47,7 +47,7 @@ public NewBlockHeadersSubscriptionService(
   public void onBlockAdded(final BlockAddedEvent event) {
     if (event.isNewCanonicalHead()) {
       final List<BlockHeader> blocks = new ArrayList<>();
-      BlockHeader blockPtr = event.getBlock().getHeader();
+      BlockHeader blockPtr = event.getHeader();
 
       while (!blockPtr.getHash().equals(event.getCommonAncestorHash())) {
         blocks.add(blockPtr);
@@ -56,7 +56,8 @@ public void onBlockAdded(final BlockAddedEvent event) {
             blockchainQueries
                 .getBlockchain()
                 .getBlockHeader(blockPtr.getParentHash())
-                .orElseThrow(() -> new IllegalStateException("The block was on a orphaned chain."));
+                .orElseThrow(
+                    () -> new IllegalStateException("The block was on an orphaned chain."));
       }
 
       Collections.reverse(blocks);

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/query/cache/AutoTransactionLogBloomCachingService.java

@@ -67,7 +67,7 @@ public void start() {
               blockchain.observeBlockAdded(
                   event -> {
                     if (event.isNewCanonicalHead()) {
-                      final BlockHeader eventBlockHeader = event.getBlock().getHeader();
+                      final BlockHeader eventBlockHeader = event.getHeader();
                       final Optional<BlockHeader> commonAncestorBlockHeader =
                           blockchain.getBlockHeader(event.getCommonAncestorHash());
                       transactionLogBloomCacher.cacheLogsBloomForBlockHeader(

ethereum/blockcreation/src/main/java/org/hyperledger/besu/ethereum/blockcreation/AbstractMiningCoordinator.java

@@ -172,8 +172,7 @@ protected void haltMiner(final M miner) {
   @Override
   public void onBlockAdded(final BlockAddedEvent event) {
     synchronized (this) {
-      if (event.isNewCanonicalHead()
-          && newChainHeadInvalidatesMiningOperation(event.getBlock().getHeader())) {
+      if (event.isNewCanonicalHead() && newChainHeadInvalidatesMiningOperation(event.getHeader())) {
         haltCurrentMiningOperation();
         startMiningIfPossible();
       }

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/chain/BlockAddedEvent.java

@@ -16,22 +16,25 @@
 
 import org.hyperledger.besu.datatypes.Hash;
 import org.hyperledger.besu.ethereum.core.Block;
+import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.core.LogWithMetadata;
 import org.hyperledger.besu.ethereum.core.Transaction;
 import org.hyperledger.besu.ethereum.core.TransactionReceipt;
 
 import java.util.Collections;
 import java.util.List;
+import java.util.function.Supplier;
 
 public class BlockAddedEvent {
 
-  private final Block block;
+  private final Supplier<Block> blockSupplier;
   private final List<Transaction> addedTransactions;
   private final List<Transaction> removedTransactions;
   private final List<TransactionReceipt> transactionReceipts;
   private final EventType eventType;
   private final List<LogWithMetadata> logsWithMetadata;
   private final Hash commonAncestorHash;
+  private final BlockHeader header;
 
   public enum EventType {
     HEAD_ADVANCED,
@@ -42,28 +45,47 @@ public enum EventType {
 
   private BlockAddedEvent(
       final EventType eventType,
-      final Block block,
+      final Supplier<Block> blockSupplier,
+      final BlockHeader header,
       final List<Transaction> addedTransactions,
       final List<Transaction> removedTransactions,
       final List<TransactionReceipt> transactionReceipts,
       final List<LogWithMetadata> logsWithMetadata,
       final Hash commonAncestorHash) {
     this.eventType = eventType;
-    this.block = block;
+    this.blockSupplier = blockSupplier;
+    this.header = header;
     this.addedTransactions = addedTransactions;
     this.removedTransactions = removedTransactions;
     this.transactionReceipts = transactionReceipts;
     this.logsWithMetadata = logsWithMetadata;
     this.commonAncestorHash = commonAncestorHash;
   }
 
+  public static BlockAddedEvent createForSyncHeadAdvancement(
+      final BlockHeader blockHeader,
+      final Supplier<Block> blockSupplier,
+      final List<LogWithMetadata> logsWithMetadata,
+      final List<TransactionReceipt> transactionReceipts) {
+    return new BlockAddedEvent(
+        EventType.HEAD_ADVANCED,
+        blockSupplier,
+        blockHeader,
+        Collections.emptyList(),
+        Collections.emptyList(),
+        transactionReceipts,
+        logsWithMetadata,
+        blockHeader.getParentHash());
+  }
+
   public static BlockAddedEvent createForHeadAdvancement(
       final Block block,
       final List<LogWithMetadata> logsWithMetadata,
       final List<TransactionReceipt> transactionReceipts) {
     return new BlockAddedEvent(
         EventType.HEAD_ADVANCED,
-        block,
+        () -> block,
+        block.getHeader(),
         block.getBody().getTransactions(),
         Collections.emptyList(),
         transactionReceipts,
@@ -80,7 +102,8 @@ public static BlockAddedEvent createForChainReorg(
       final Hash commonAncestorHash) {
     return new BlockAddedEvent(
         EventType.CHAIN_REORG,
-        block,
+        () -> block,
+        block.getHeader(),
         addedTransactions,
         removedTransactions,
         transactionReceipts,
@@ -91,7 +114,8 @@ public static BlockAddedEvent createForChainReorg(
   public static BlockAddedEvent createForFork(final Block block) {
     return new BlockAddedEvent(
         EventType.FORK,
-        block,
+        () -> block,
+        block.getHeader(),
         Collections.emptyList(),
         Collections.emptyList(),
         Collections.emptyList(),
@@ -102,7 +126,8 @@ public static BlockAddedEvent createForFork(final Block block) {
   public static BlockAddedEvent createForStoredOnly(final Block block) {
     return new BlockAddedEvent(
         EventType.STORED_ONLY,
-        block,
+        () -> block,
+        block.getHeader(),
         Collections.emptyList(),
         Collections.emptyList(),
         Collections.emptyList(),
@@ -111,7 +136,11 @@ public static BlockAddedEvent createForStoredOnly(final Block block) {
   }
 
   public Block getBlock() {
-    return block;
+    return blockSupplier.get();
+  }
+
+  public BlockHeader getHeader() {
+    return header;
   }
 
   public boolean isNewCanonicalHead() {
@@ -148,7 +177,7 @@ public String toString() {
         + "eventType="
         + eventType
         + ", block="
-        + block.toLogString()
+        + header.toLogString()
         + ", commonAncestorHash="
         + commonAncestorHash
         + ", addedTransactions count="

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/chain/BlockchainStorage.java

@@ -18,6 +18,7 @@
 import org.hyperledger.besu.ethereum.core.BlockBody;
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.core.Difficulty;
+import org.hyperledger.besu.ethereum.core.SyncBlockBody;
 import org.hyperledger.besu.ethereum.core.TransactionReceipt;
 
 import java.util.Collection;
@@ -54,6 +55,8 @@ interface Updater {
 
     void putBlockBody(Hash blockHash, BlockBody blockBody);
 
+    void putSyncBlockBody(Hash blockHash, SyncBlockBody blockBody);
+
     void putTransactionLocation(Hash transactionHash, TransactionLocation transactionLocation);
 
     void putTransactionReceipts(Hash blockHash, List<TransactionReceipt> transactionReceipts);

ethereum/core/src/main/java/org/hyperledger/besu/ethereum/chain/ChainDataPruner.java

@@ -74,7 +74,7 @@ public void onBlockAdded(final BlockAddedEvent event) {
   }
 
   private void chainPrunerAction(final BlockAddedEvent event) {
-    final long blockNumber = event.getBlock().getHeader().getNumber();
+    final long blockNumber = event.getHeader().getNumber();
     final long storedPruningMark = prunerStorage.getPruningMark().orElse(blockNumber);
     if (blockNumber < storedPruningMark) {
       LOG.warn(
@@ -90,7 +90,7 @@ private void chainPrunerAction(final BlockAddedEvent event) {
     final KeyValueStorageTransaction recordBlockHashesTransaction =
         prunerStorage.startTransaction();
     final Collection<Hash> forkBlocks = prunerStorage.getForkBlocks(blockNumber);
-    forkBlocks.add(event.getBlock().getHash());
+    forkBlocks.add(event.getHeader().getHash());
     prunerStorage.setForkBlocks(recordBlockHashesTransaction, blockNumber, forkBlocks);
     recordBlockHashesTransaction.commit();