Cache current chain head info (#1335)

Store the header for the current chain head and total difficulty to avoid RocksDB lookups when requesting those common values.

Also uses that cache to avoid a database lookup when checking if a block has already been imported if the block's parent is the current chain head.

ethereum/core/src/main/java/tech/pegasys/pantheon/ethereum/chain/DefaultMutableBlockchain.java

@@ -48,6 +48,9 @@ public class DefaultMutableBlockchain implements MutableBlockchain {
 
   private final Subscribers<BlockAddedObserver> blockAddedObservers = new Subscribers<>();
 
+  private volatile BlockHeader chainHeader;
+  private volatile UInt256 totalDifficulty;
+
   public DefaultMutableBlockchain(
       final Block genesisBlock,
       final BlockchainStorage blockchainStorage,
@@ -56,6 +59,10 @@ public DefaultMutableBlockchain(
     this.blockchainStorage = blockchainStorage;
     this.setGenesis(genesisBlock);
 
+    final Hash chainHead = blockchainStorage.getChainHead().get();
+    chainHeader = blockchainStorage.getBlockHeader(chainHead).get();
+    totalDifficulty = blockchainStorage.getTotalDifficulty(chainHead).get();
+
     metricsSystem.createGauge(
         MetricCategory.BLOCKCHAIN,
         "height",
@@ -72,25 +79,22 @@ public DefaultMutableBlockchain(
 
   @Override
   public ChainHead getChainHead() {
-    return blockchainStorage
-        .getChainHead()
-        .flatMap(h -> blockchainStorage.getTotalDifficulty(h).map(td -> new ChainHead(h, td)))
-        .get();
+    return new ChainHead(chainHeader.getHash(), totalDifficulty);
   }
 
   @Override
   public Hash getChainHeadHash() {
-    return blockchainStorage.getChainHead().get();
+    return chainHeader.getHash();
   }
 
   @Override
   public long getChainHeadBlockNumber() {
-    // Head should always be set, so we can call get()
-    return blockchainStorage
-        .getChainHead()
-        .flatMap(blockchainStorage::getBlockHeader)
-        .map(BlockHeader::getNumber)
-        .get();
+    return chainHeader.getNumber();
+  }
+
+  @Override
+  public BlockHeader getChainHeadHeader() {
+    return chainHeader;
   }
 
   @Override
@@ -171,6 +175,10 @@ private BlockAddedEvent appendBlockHelper(
     final BlockAddedEvent blockAddedEvent = updateCanonicalChainData(updater, block, td);
 
     updater.commit();
+    if (blockAddedEvent.isNewCanonicalHead()) {
+      chainHeader = block.getHeader();
+      totalDifficulty = td;
+    }
 
     return blockAddedEvent;
   }
@@ -368,11 +376,17 @@ protected void setGenesis(final Block genesisBlock) {
     }
   }
 
-  protected boolean blockIsAlreadyTracked(final Block block) {
+  private boolean blockIsAlreadyTracked(final Block block) {
+    if (block.getHeader().getParentHash().equals(chainHeader.getHash())) {
+      // If this block builds on our chain head it would have a higher TD and be the chain head
+      // but since it isn't we mustn't have imported it yet.
+      // Saves a db read for the most common case
+      return false;
+    }
     return blockchainStorage.getBlockHeader(block.getHash()).isPresent();
   }
 
-  protected boolean blockIsConnected(final Block block) {
+  private boolean blockIsConnected(final Block block) {
     return blockchainStorage.getBlockHeader(block.getHeader().getParentHash()).isPresent();
   }
 

ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/BlockPropagationManagerTest.java

@@ -14,6 +14,9 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
@@ -26,6 +29,7 @@
 import tech.pegasys.pantheon.ethereum.core.Block;
 import tech.pegasys.pantheon.ethereum.core.BlockDataGenerator;
 import tech.pegasys.pantheon.ethereum.core.BlockDataGenerator.BlockOptions;
+import tech.pegasys.pantheon.ethereum.core.BlockImporter;
 import tech.pegasys.pantheon.ethereum.eth.manager.EthContext;
 import tech.pegasys.pantheon.ethereum.eth.manager.EthMessages;
 import tech.pegasys.pantheon.ethereum.eth.manager.EthPeers;
@@ -41,6 +45,7 @@
 import tech.pegasys.pantheon.ethereum.eth.sync.state.PendingBlocks;
 import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncState;
 import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
+import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSpec;
 import tech.pegasys.pantheon.metrics.MetricsSystem;
 import tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem;
 import tech.pegasys.pantheon.util.uint.UInt256;
@@ -77,7 +82,7 @@ public static void setupSuite() {
   @Before
   public void setup() {
     blockchainUtil = BlockchainSetupUtil.forTesting();
-    blockchain = spy(blockchainUtil.getBlockchain());
+    blockchain = blockchainUtil.getBlockchain();
     protocolSchedule = blockchainUtil.getProtocolSchedule();
     final ProtocolContext<Void> tempProtocolContext = blockchainUtil.getProtocolContext();
     protocolContext =
@@ -290,6 +295,22 @@ public void importsMixedOutOfOrderMessages() {
 
   @Test
   public void handlesDuplicateAnnouncements() {
+    final ProtocolSchedule<Void> stubProtocolSchedule = spy(protocolSchedule);
+    final ProtocolSpec<Void> stubProtocolSpec = spy(protocolSchedule.getByBlockNumber(2));
+    final BlockImporter<Void> stubBlockImporter = spy(stubProtocolSpec.getBlockImporter());
+    doReturn(stubProtocolSpec).when(stubProtocolSchedule).getByBlockNumber(anyLong());
+    doReturn(stubBlockImporter).when(stubProtocolSpec).getBlockImporter();
+    final BlockPropagationManager<Void> blockPropagationManager =
+        new BlockPropagationManager<>(
+            syncConfig,
+            stubProtocolSchedule,
+            protocolContext,
+            ethProtocolManager.ethContext(),
+            syncState,
+            pendingBlocks,
+            metricsSystem,
+            blockBroadcaster);
+
     blockchainUtil.importFirstBlocks(2);
     final Block nextBlock = blockchainUtil.getBlock(2);
 
@@ -320,11 +341,26 @@ public void handlesDuplicateAnnouncements() {
     peer.respondWhile(responder, peer::hasOutstandingRequests);
 
     assertThat(blockchain.contains(nextBlock.getHash())).isTrue();
-    verify(blockchain, times(1)).appendBlock(any(), any());
+    verify(stubBlockImporter, times(1)).importBlock(eq(protocolContext), eq(nextBlock), any());
   }
 
   @Test
   public void handlesPendingDuplicateAnnouncements() {
+    final ProtocolSchedule<Void> stubProtocolSchedule = spy(protocolSchedule);
+    final ProtocolSpec<Void> stubProtocolSpec = spy(protocolSchedule.getByBlockNumber(2));
+    final BlockImporter<Void> stubBlockImporter = spy(stubProtocolSpec.getBlockImporter());
+    doReturn(stubProtocolSpec).when(stubProtocolSchedule).getByBlockNumber(anyLong());
+    doReturn(stubBlockImporter).when(stubProtocolSpec).getBlockImporter();
+    final BlockPropagationManager<Void> blockPropagationManager =
+        new BlockPropagationManager<>(
+            syncConfig,
+            stubProtocolSchedule,
+            protocolContext,
+            ethProtocolManager.ethContext(),
+            syncState,
+            pendingBlocks,
+            metricsSystem,
+            blockBroadcaster);
     blockchainUtil.importFirstBlocks(2);
     final Block nextBlock = blockchainUtil.getBlock(2);
 
@@ -352,7 +388,7 @@ public void handlesPendingDuplicateAnnouncements() {
     peer.respondWhile(responder, peer::hasOutstandingRequests);
 
     assertThat(blockchain.contains(nextBlock.getHash())).isTrue();
-    verify(blockchain, times(1)).appendBlock(any(), any());
+    verify(stubBlockImporter, times(1)).importBlock(eq(protocolContext), eq(nextBlock), any());
   }
 
   @Test

ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/fullsync/FullSyncChainDownloaderTest.java

@@ -13,13 +13,8 @@
 package tech.pegasys.pantheon.ethereum.eth.sync.fullsync;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.fail;
 import static org.assertj.core.api.Assumptions.assumeThatObject;
 import static org.awaitility.Awaitility.await;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
 import static tech.pegasys.pantheon.ethereum.core.InMemoryStorageProvider.createInMemoryBlockchain;
 
 import tech.pegasys.pantheon.ethereum.ProtocolContext;
@@ -58,6 +53,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 
+import org.awaitility.Awaitility;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -96,7 +92,7 @@ public static Object[][] params() {
   public void setupTest() {
     gen = new BlockDataGenerator();
     localBlockchainSetup = BlockchainSetupUtil.forTesting();
-    localBlockchain = spy(localBlockchainSetup.getBlockchain());
+    localBlockchain = localBlockchainSetup.getBlockchain();
     otherBlockchainSetup = BlockchainSetupUtil.forTesting();
     otherBlockchain = otherBlockchainSetup.getBlockchain();
 
@@ -229,7 +225,6 @@ public void doesNotSyncToWorseChain() {
     peer.respondWhileOtherThreadsWork(responder, peer::hasOutstandingRequests);
 
     assertThat(syncState.syncTarget()).isNotPresent();
-    verify(localBlockchain, times(0)).appendBlock(any(), any());
   }
 
   @Test
@@ -565,12 +560,18 @@ public void requestsCheckpointsFromSyncTarget() {
     assertThat(syncState.syncTarget()).isPresent();
     assertThat(syncState.syncTarget().get().peer()).isEqualTo(bestPeer.getEthPeer());
 
-    int count = 0;
     while (localBlockchain.getChainHeadBlockNumber() < bestPeerChainHead) {
-      if (count > 10_000) {
-        fail("Did not reach chain head soon enough");
-      }
-      count++;
+      // Wait until there is a request to respond to (or we reached chain head).
+      // If we don't get a new request within 30 seconds the test will fail because we've probably
+      // stalled.
+      Awaitility.await()
+          .atMost(30, TimeUnit.SECONDS)
+          .until(
+              () ->
+                  bestPeer.hasOutstandingRequests()
+                      || otherPeers.stream().anyMatch(RespondingEthPeer::hasOutstandingRequests)
+                      || localBlockchain.getChainHeadBlockNumber() >= bestPeerChainHead);
+
       // Check that any requests for checkpoint headers are only sent to the best peer
       final long checkpointRequestsToOtherPeers =
           otherPeers.stream()

ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/fullsync/IncrementerTest.java

@@ -13,7 +13,6 @@
 package tech.pegasys.pantheon.ethereum.eth.sync.fullsync;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.Mockito.spy;
 
 import tech.pegasys.pantheon.ethereum.ProtocolContext;
 import tech.pegasys.pantheon.ethereum.chain.Blockchain;
@@ -62,7 +61,7 @@ public void setUp() {
     metricsSystem = PrometheusMetricsSystem.init(metricsConfiguration);
 
     final BlockchainSetupUtil<Void> localBlockchainSetup = BlockchainSetupUtil.forTesting();
-    localBlockchain = spy(localBlockchainSetup.getBlockchain());
+    localBlockchain = localBlockchainSetup.getBlockchain();
     final BlockchainSetupUtil<Void> otherBlockchainSetup = BlockchainSetupUtil.forTesting();
     otherBlockchain = otherBlockchainSetup.getBlockchain();