NC-1880 High TX volume swamps block processing (#337)

* NC-1880 High TX volume swamps block processing

Move transaction processing into its own thread(s).

Size of txWorkerExecutor thread pool can be independently configured.

consensus/ibftlegacy/src/main/java/tech/pegasys/pantheon/consensus/ibftlegacy/protocol/Istanbul64ProtocolManager.java

@@ -29,8 +29,9 @@ public class Istanbul64ProtocolManager extends EthProtocolManager {
       final Blockchain blockchain,
       final int networkId,
       final boolean fastSyncEnabled,
-      final int workers) {
+      final int syncWorkers,
-    super(blockchain, networkId, fastSyncEnabled, workers);
+      final int txWorkers) {
+    super(blockchain, networkId, fastSyncEnabled, syncWorkers, txWorkers);
   }
 
   @Override

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/manager/AbstractEthTask.java

@@ -100,7 +100,7 @@ public abstract class AbstractEthTask<T> implements EthTask<T> {
    */
   protected final <S> CompletableFuture<S> executeWorkerSubTask(
       final EthScheduler scheduler, final Supplier<CompletableFuture<S>> subTask) {
-    return executeSubTask(() -> scheduler.scheduleWorkerTask(subTask));
+    return executeSubTask(() -> scheduler.scheduleSyncWorkerTask(subTask));
   }
 
   public final T result() {

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/manager/EthProtocolManager.java

@@ -90,17 +90,24 @@ public class EthProtocolManager implements ProtocolManager, MinedBlockObserver {
       final Blockchain blockchain,
       final int networkId,
       final boolean fastSyncEnabled,
-      final int workers,
+      final int syncWorkers,
+      final int txWorkers,
       final int requestLimit) {
-    this(blockchain, networkId, fastSyncEnabled, requestLimit, new EthScheduler(workers));
+    this(
+        blockchain,
+        networkId,
+        fastSyncEnabled,
+        requestLimit,
+        new EthScheduler(syncWorkers, txWorkers));
   }
 
   public EthProtocolManager(
       final Blockchain blockchain,
       final int networkId,
       final boolean fastSyncEnabled,
-      final int workers) {
+      final int syncWorkers,
-    this(blockchain, networkId, fastSyncEnabled, workers, DEFAULT_REQUEST_LIMIT);
+      final int txWorkers) {
+    this(blockchain, networkId, fastSyncEnabled, syncWorkers, txWorkers, DEFAULT_REQUEST_LIMIT);
   }
 
   public EthContext ethContext() {

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/manager/EthScheduler.java

@@ -41,13 +41,14 @@ public class EthScheduler {
   private final AtomicBoolean stopped = new AtomicBoolean(false);
   private final CountDownLatch shutdown = new CountDownLatch(1);
 
-  protected final ExecutorService workerExecutor;
+  protected final ExecutorService syncWorkerExecutor;
   protected final ScheduledExecutorService scheduler;
+  protected final ExecutorService txWorkerExecutor;
 
-  EthScheduler(final int workerCount) {
+  EthScheduler(final int syncWorkerCount, final int txWorkerCount) {
     this(
         Executors.newFixedThreadPool(
-            workerCount,
+            syncWorkerCount,
             new ThreadFactoryBuilder()
                 .setNameFormat(EthScheduler.class.getSimpleName() + "-Workers")
                 .build()),
@@ -56,19 +57,28 @@ public class EthScheduler {
             new ThreadFactoryBuilder()
                 .setDaemon(true)
                 .setNameFormat(EthScheduler.class.getSimpleName() + "Timer")
+                .build()),
+        Executors.newFixedThreadPool(
+            txWorkerCount,
+            new ThreadFactoryBuilder()
+                .setNameFormat(EthScheduler.class.getSimpleName() + "-Transactions")
                 .build()));
   }
 
   protected EthScheduler(
-      final ExecutorService workerExecutor, final ScheduledExecutorService scheduler) {
+      final ExecutorService syncWorkerExecutor,
-    this.workerExecutor = workerExecutor;
+      final ScheduledExecutorService scheduler,
+      final ExecutorService txWorkerExecutor) {
+    this.syncWorkerExecutor = syncWorkerExecutor;
     this.scheduler = scheduler;
+    this.txWorkerExecutor = txWorkerExecutor;
   }
 
-  public <T> CompletableFuture<T> scheduleWorkerTask(final Supplier<CompletableFuture<T>> future) {
+  public <T> CompletableFuture<T> scheduleSyncWorkerTask(
+      final Supplier<CompletableFuture<T>> future) {
     final CompletableFuture<T> promise = new CompletableFuture<>();
     final Future<?> workerFuture =
-        workerExecutor.submit(
+        syncWorkerExecutor.submit(
             () -> {
               future
                   .get()
@@ -91,8 +101,12 @@ public class EthScheduler {
     return promise;
   }
 
-  public Future<?> scheduleWorkerTask(final Runnable command) {
+  public Future<?> scheduleSyncWorkerTask(final Runnable command) {
-    return workerExecutor.submit(command);
+    return syncWorkerExecutor.submit(command);
+  }
+
+  public Future<?> scheduleTxWorkerTask(final Runnable command) {
+    return txWorkerExecutor.submit(command);
   }
 
   public CompletableFuture<Void> scheduleFutureTask(
@@ -179,7 +193,7 @@ public class EthScheduler {
   public void stop() {
     if (stopped.compareAndSet(false, true)) {
       LOG.trace("Stopping " + getClass().getSimpleName());
-      workerExecutor.shutdown();
+      syncWorkerExecutor.shutdown();
       scheduler.shutdown();
       shutdown.countDown();
     } else {
@@ -189,10 +203,10 @@ public class EthScheduler {
 
   public void awaitStop() throws InterruptedException {
     shutdown.await();
-    if (!workerExecutor.awaitTermination(2L, TimeUnit.MINUTES)) {
+    if (!syncWorkerExecutor.awaitTermination(2L, TimeUnit.MINUTES)) {
       LOG.error("{} worker executor did not shutdown cleanly.", this.getClass().getSimpleName());
-      workerExecutor.shutdownNow();
+      syncWorkerExecutor.shutdownNow();
-      workerExecutor.awaitTermination(2L, TimeUnit.MINUTES);
+      syncWorkerExecutor.awaitTermination(2L, TimeUnit.MINUTES);
     }
     if (!scheduler.awaitTermination(2L, TimeUnit.MINUTES)) {
       LOG.error("{} scheduler did not shutdown cleanly.", this.getClass().getSimpleName());

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/BlockPropagationManager.java

@@ -128,7 +128,7 @@ public class BlockPropagationManager<C> {
               protocolSchedule, protocolContext, readyForImport, HeaderValidationMode.FULL);
       ethContext
           .getScheduler()
-          .scheduleWorkerTask(importBlocksTask)
+          .scheduleSyncWorkerTask(importBlocksTask)
           .whenComplete(
               (r, t) -> {
                 if (r != null) {
@@ -255,7 +255,7 @@ public class BlockPropagationManager<C> {
     final OperationTimer.TimingContext blockTimer = announcedBlockIngestTimer.startTimer();
     return ethContext
         .getScheduler()
-        .scheduleWorkerTask(importTask::run)
+        .scheduleSyncWorkerTask(importTask::run)
         .whenComplete(
             (r, t) -> {
               if (t != null) {

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/SynchronizerConfiguration.java

@@ -52,6 +52,7 @@ public class SynchronizerConfiguration {
   private final long trailingPeerBlocksBehindThreshold;
   private final int maxTrailingPeers;
   private final int downloaderParallelism;
+  private final int transactionsParallelism;
 
   private SynchronizerConfiguration(
       final SyncMode requestedSyncMode,
@@ -67,7 +68,8 @@ public class SynchronizerConfiguration {
       final int downloaderChainSegmentSize,
       final long trailingPeerBlocksBehindThreshold,
       final int maxTrailingPeers,
-      final int downloaderParallelism) {
+      final int downloaderParallelism,
+      final int transactionsParallelism) {
     this.requestedSyncMode = requestedSyncMode;
     this.fastSyncPivotDistance = fastSyncPivotDistance;
     this.fastSyncFullValidationRate = fastSyncFullValidationRate;
@@ -82,6 +84,7 @@ public class SynchronizerConfiguration {
     this.trailingPeerBlocksBehindThreshold = trailingPeerBlocksBehindThreshold;
     this.maxTrailingPeers = maxTrailingPeers;
     this.downloaderParallelism = downloaderParallelism;
+    this.transactionsParallelism = transactionsParallelism;
   }
 
   /**
@@ -122,7 +125,8 @@ public class SynchronizerConfiguration {
         downloaderChainSegmentSize,
         trailingPeerBlocksBehindThreshold,
         maxTrailingPeers,
-        downloaderParallelism);
+        downloaderParallelism,
+        transactionsParallelism);
   }
 
   public static Builder builder() {
@@ -203,6 +207,10 @@ public class SynchronizerConfiguration {
     return downloaderParallelism;
   }
 
+  public int transactionsParallelism() {
+    return transactionsParallelism;
+  }
+
   /**
    * The rate at which blocks should be fully validated during fast sync. At a rate of 1f, all
    * blocks are fully validated. At rates less than 1f, a subset of blocks will undergo light-weight
@@ -228,6 +236,7 @@ public class SynchronizerConfiguration {
     private long trailingPeerBlocksBehindThreshold;
     private int maxTrailingPeers = Integer.MAX_VALUE;
     private int downloaderParallelism = 2;
+    private int transactionsParallelism = 2;
 
     public Builder fastSyncPivotDistance(final int distance) {
       fastSyncPivotDistance = distance;
@@ -299,6 +308,11 @@ public class SynchronizerConfiguration {
       return this;
     }
 
+    public Builder transactionsParallelism(final int transactionsParallelism) {
+      this.transactionsParallelism = transactionsParallelism;
+      return this;
+    }
+
     public SynchronizerConfiguration build() {
       return new SynchronizerConfiguration(
           syncMode,
@@ -314,7 +328,8 @@ public class SynchronizerConfiguration {
           downloaderChainSegmentSize,
           trailingPeerBlocksBehindThreshold,
           maxTrailingPeers,
-          downloaderParallelism);
+          downloaderParallelism,
+          transactionsParallelism);
     }
   }
 }

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/transactions/TransactionSender.java

@@ -42,6 +42,6 @@ class TransactionSender implements TransactionBatchAddedListener {
                     transaction -> transactionTracker.addToPeerSendQueue(peer, transaction)));
     ethContext
         .getScheduler()
-        .scheduleWorkerTask(transactionsMessageSender::sendTransactionsToPeers);
+        .scheduleSyncWorkerTask(transactionsMessageSender::sendTransactionsToPeers);
   }
 }

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/transactions/TransactionsMessageHandler.java

@@ -32,7 +32,7 @@ class TransactionsMessageHandler implements MessageCallback {
   @Override
   public void exec(final EthMessage message) {
     final TransactionsMessage transactionsMessage = TransactionsMessage.readFrom(message.getData());
-    scheduler.scheduleWorkerTask(
+    scheduler.scheduleTxWorkerTask(
         () ->
             transactionsMessageProcessor.processTransactionsMessage(
                 message.getPeer(), transactionsMessage));

ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/manager/DeterministicEthScheduler.java

@@ -27,18 +27,22 @@ public class DeterministicEthScheduler extends EthScheduler {
   }
 
   DeterministicEthScheduler(final TimeoutPolicy timeoutPolicy) {
-    super(new MockExecutorService(), new MockScheduledExecutor());
+    super(new MockExecutorService(), new MockScheduledExecutor(), new MockExecutorService());
     this.timeoutPolicy = timeoutPolicy;
   }
 
-  MockExecutorService mockWorkerExecutor() {
+  MockExecutorService mockSyncWorkerExecutor() {
-    return (MockExecutorService) workerExecutor;
+    return (MockExecutorService) syncWorkerExecutor;
   }
 
   MockScheduledExecutor mockScheduledExecutor() {
     return (MockScheduledExecutor) scheduler;
   }
 
+  MockScheduledExecutor mockTransactionsExecutor() {
+    return (MockScheduledExecutor) txWorkerExecutor;
+  }
+
   @Override
   public <T> void failAfterTimeout(final CompletableFuture<T> promise, final Duration timeout) {
     if (timeoutPolicy.shouldTimeout()) {

ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/manager/EthProtocolManagerTest.java

@@ -13,18 +13,22 @@
 package tech.pegasys.pantheon.ethereum.eth.manager;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.reset;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyZeroInteractions;
 import static tech.pegasys.pantheon.ethereum.core.InMemoryStorageProvider.createInMemoryBlockchain;
 
+import tech.pegasys.pantheon.ethereum.ProtocolContext;
 import tech.pegasys.pantheon.ethereum.chain.Blockchain;
 import tech.pegasys.pantheon.ethereum.chain.MutableBlockchain;
 import tech.pegasys.pantheon.ethereum.core.Block;
 import tech.pegasys.pantheon.ethereum.core.BlockBody;
 import tech.pegasys.pantheon.ethereum.core.BlockHeader;
 import tech.pegasys.pantheon.ethereum.core.Hash;
+import tech.pegasys.pantheon.ethereum.core.Transaction;
 import tech.pegasys.pantheon.ethereum.core.TransactionReceipt;
 import tech.pegasys.pantheon.ethereum.eth.EthProtocol;
 import tech.pegasys.pantheon.ethereum.eth.EthProtocol.EthVersion;
@@ -40,12 +44,15 @@ import tech.pegasys.pantheon.ethereum.eth.messages.GetReceiptsMessage;
 import tech.pegasys.pantheon.ethereum.eth.messages.NewBlockMessage;
 import tech.pegasys.pantheon.ethereum.eth.messages.ReceiptsMessage;
 import tech.pegasys.pantheon.ethereum.eth.messages.StatusMessage;
+import tech.pegasys.pantheon.ethereum.eth.messages.TransactionsMessage;
+import tech.pegasys.pantheon.ethereum.eth.transactions.TransactionPoolFactory;
 import tech.pegasys.pantheon.ethereum.mainnet.MainnetProtocolSchedule;
 import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
 import tech.pegasys.pantheon.ethereum.p2p.api.MessageData;
 import tech.pegasys.pantheon.ethereum.p2p.api.PeerConnection;
 import tech.pegasys.pantheon.ethereum.p2p.wire.Capability;
 import tech.pegasys.pantheon.ethereum.p2p.wire.DefaultMessage;
+import tech.pegasys.pantheon.ethereum.p2p.wire.RawMessage;
 import tech.pegasys.pantheon.ethereum.testutil.BlockDataGenerator;
 import tech.pegasys.pantheon.util.uint.UInt256;
 
@@ -57,6 +64,8 @@ import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
@@ -72,6 +81,7 @@ public final class EthProtocolManagerTest {
   private static Blockchain blockchain;
   private static ProtocolSchedule<Void> protocolSchedule;
   private static BlockDataGenerator gen;
+  private static ProtocolContext<Void> protocolContext;
 
   @BeforeClass
   public static void setup() {
@@ -80,12 +90,13 @@ public final class EthProtocolManagerTest {
     blockchainSetupUtil.importAllBlocks();
     blockchain = blockchainSetupUtil.getBlockchain();
     protocolSchedule = blockchainSetupUtil.getProtocolSchedule();
+    protocolContext = blockchainSetupUtil.getProtocolContext();
     assert (blockchainSetupUtil.getMaxBlockNumber() >= 20L);
   }
 
   @Test
   public void disconnectOnUnsolicitedMessage() {
-    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1)) {
+    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1, 1)) {
       final MessageData messageData =
           BlockHeadersMessage.create(Collections.singletonList(blockchain.getBlockHeader(1).get()));
       final MockPeerConnection peer = setupPeer(ethManager, (cap, msg, conn) -> {});
@@ -96,7 +107,7 @@ public final class EthProtocolManagerTest {
 
   @Test
   public void disconnectOnFailureToSendStatusMessage() {
-    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1)) {
+    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1, 1)) {
       final MessageData messageData =
           BlockHeadersMessage.create(Collections.singletonList(blockchain.getBlockHeader(1).get()));
       final MockPeerConnection peer =
@@ -108,7 +119,7 @@ public final class EthProtocolManagerTest {
 
   @Test
   public void disconnectOnWrongChainId() {
-    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1)) {
+    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1, 1)) {
       final MessageData messageData =
           BlockHeadersMessage.create(Collections.singletonList(blockchain.getBlockHeader(1).get()));
       final MockPeerConnection peer =
@@ -131,7 +142,7 @@ public final class EthProtocolManagerTest {
 
   @Test
   public void disconnectOnWrongGenesisHash() {
-    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1)) {
+    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1, 1)) {
       final MessageData messageData =
           BlockHeadersMessage.create(Collections.singletonList(blockchain.getBlockHeader(1).get()));
       final MockPeerConnection peer =
@@ -154,7 +165,7 @@ public final class EthProtocolManagerTest {
 
   @Test(expected = ConditionTimeoutException.class)
   public void doNotDisconnectOnValidMessage() {
-    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1)) {
+    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1, 1)) {
       final MessageData messageData =
           GetBlockBodiesMessage.create(Collections.singletonList(gen.hash()));
       final MockPeerConnection peer = setupPeer(ethManager, (cap, msg, conn) -> {});
@@ -169,7 +180,7 @@ public final class EthProtocolManagerTest {
   @Test
   public void respondToGetHeaders() throws ExecutionException, InterruptedException {
     final CompletableFuture<Void> done = new CompletableFuture<>();
-    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1)) {
+    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1, 1)) {
       final long startBlock = 5L;
       final int blockCount = 5;
       final MessageData messageData =
@@ -201,7 +212,7 @@ public final class EthProtocolManagerTest {
     final CompletableFuture<Void> done = new CompletableFuture<>();
     final int limit = 5;
     try (final EthProtocolManager ethManager =
-        new EthProtocolManager(blockchain, 1, true, 1, limit)) {
+        new EthProtocolManager(blockchain, 1, true, 1, 1, limit)) {
       final long startBlock = 5L;
       final int blockCount = 10;
       final MessageData messageData =
@@ -231,7 +242,7 @@ public final class EthProtocolManagerTest {
   @Test
   public void respondToGetHeadersReversed() throws ExecutionException, InterruptedException {
     final CompletableFuture<Void> done = new CompletableFuture<>();
-    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1)) {
+    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1, 1)) {
       final long endBlock = 10L;
       final int blockCount = 5;
       final MessageData messageData = GetBlockHeadersMessage.create(endBlock, blockCount, 0, true);
@@ -260,7 +271,7 @@ public final class EthProtocolManagerTest {
   @Test
   public void respondToGetHeadersWithSkip() throws ExecutionException, InterruptedException {
     final CompletableFuture<Void> done = new CompletableFuture<>();
-    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1)) {
+    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1, 1)) {
       final long startBlock = 5L;
       final int blockCount = 5;
       final int skip = 1;
@@ -292,7 +303,7 @@ public final class EthProtocolManagerTest {
   public void respondToGetHeadersReversedWithSkip()
       throws ExecutionException, InterruptedException {
     final CompletableFuture<Void> done = new CompletableFuture<>();
-    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1)) {
+    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1, 1)) {
       final long endBlock = 10L;
       final int blockCount = 5;
       final int skip = 1;
@@ -336,7 +347,7 @@ public final class EthProtocolManagerTest {
 
   private MockPeerConnection setupPeerWithoutStatusExchange(
       final EthProtocolManager ethManager, final PeerSendHandler onSend) {
-    final Set<Capability> caps = new HashSet<>(Arrays.asList(EthProtocol.ETH63));
+    final Set<Capability> caps = new HashSet<>(Collections.singletonList(EthProtocol.ETH63));
     final MockPeerConnection peer = new MockPeerConnection(caps, onSend);
     ethManager.handleNewConnection(peer);
     return peer;
@@ -345,7 +356,7 @@ public final class EthProtocolManagerTest {
   @Test
   public void respondToGetHeadersPartial() throws ExecutionException, InterruptedException {
     final CompletableFuture<Void> done = new CompletableFuture<>();
-    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1)) {
+    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1, 1)) {
       final long startBlock = blockchain.getChainHeadBlockNumber() - 1L;
       final int blockCount = 5;
       final MessageData messageData =
@@ -375,7 +386,7 @@ public final class EthProtocolManagerTest {
   @Test
   public void respondToGetHeadersEmpty() throws ExecutionException, InterruptedException {
     final CompletableFuture<Void> done = new CompletableFuture<>();
-    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1)) {
+    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1, 1)) {
       final long startBlock = blockchain.getChainHeadBlockNumber() + 1;
       final int blockCount = 5;
       final MessageData messageData =
@@ -402,7 +413,7 @@ public final class EthProtocolManagerTest {
   @Test
   public void respondToGetBodies() throws ExecutionException, InterruptedException {
     final CompletableFuture<Void> done = new CompletableFuture<>();
-    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1)) {
+    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1, 1)) {
       // Setup blocks query
       final long startBlock = blockchain.getChainHeadBlockNumber() - 5;
       final int blockCount = 2;
@@ -446,7 +457,7 @@ public final class EthProtocolManagerTest {
     final CompletableFuture<Void> done = new CompletableFuture<>();
     final int limit = 5;
     try (final EthProtocolManager ethManager =
-        new EthProtocolManager(blockchain, 1, true, 1, limit)) {
+        new EthProtocolManager(blockchain, 1, true, 1, 1, limit)) {
       // Setup blocks query
       final int blockCount = 10;
       final long startBlock = blockchain.getChainHeadBlockNumber() - blockCount;
@@ -488,7 +499,7 @@ public final class EthProtocolManagerTest {
   @Test
   public void respondToGetBodiesPartial() throws ExecutionException, InterruptedException {
     final CompletableFuture<Void> done = new CompletableFuture<>();
-    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1)) {
+    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1, 1)) {
       // Setup blocks query
       final long expectedBlockNumber = blockchain.getChainHeadBlockNumber() - 1;
       final BlockHeader header = blockchain.getBlockHeader(expectedBlockNumber).get();
@@ -524,7 +535,7 @@ public final class EthProtocolManagerTest {
   @Test
   public void respondToGetReceipts() throws ExecutionException, InterruptedException {
     final CompletableFuture<Void> done = new CompletableFuture<>();
-    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1)) {
+    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1, 1)) {
       // Setup blocks query
       final long startBlock = blockchain.getChainHeadBlockNumber() - 5;
       final int blockCount = 2;
@@ -567,7 +578,7 @@ public final class EthProtocolManagerTest {
     final CompletableFuture<Void> done = new CompletableFuture<>();
     final int limit = 5;
     try (final EthProtocolManager ethManager =
-        new EthProtocolManager(blockchain, 1, true, 1, limit)) {
+        new EthProtocolManager(blockchain, 1, true, 1, 1, limit)) {
       // Setup blocks query
       final int blockCount = 10;
       final long startBlock = blockchain.getChainHeadBlockNumber() - blockCount;
@@ -608,7 +619,7 @@ public final class EthProtocolManagerTest {
   @Test
   public void respondToGetReceiptsPartial() throws ExecutionException, InterruptedException {
     final CompletableFuture<Void> done = new CompletableFuture<>();
-    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1)) {
+    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1, 1)) {
       // Setup blocks query
       final long blockNumber = blockchain.getChainHeadBlockNumber() - 5;
       final BlockHeader header = blockchain.getBlockHeader(blockNumber).get();
@@ -643,7 +654,7 @@ public final class EthProtocolManagerTest {
 
   @Test
   public void newBlockMinedSendsNewBlockMessageToAllPeers() {
-    final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1);
+    final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1, 1);
 
     // Define handler to validate response
     final PeerSendHandler onSend = mock(PeerSendHandler.class);
@@ -705,7 +716,7 @@ public final class EthProtocolManagerTest {
     blockchain.appendBlock(block, receipts);
 
     final CompletableFuture<Void> done = new CompletableFuture<>();
-    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1)) {
+    try (final EthProtocolManager ethManager = new EthProtocolManager(blockchain, 1, true, 1, 1)) {
       final long startBlock = 1L;
       final int requestedBlockCount = 13;
       final int receivedBlockCount = 2;
@@ -728,7 +739,7 @@ public final class EthProtocolManagerTest {
             done.complete(null);
           };
 
-      final Set<Capability> caps = new HashSet<>(Arrays.asList(EthProtocol.ETH63));
+      final Set<Capability> caps = new HashSet<>(Collections.singletonList(EthProtocol.ETH63));
       final MockPeerConnection peer = new MockPeerConnection(caps, onSend);
       ethManager.handleNewConnection(peer);
       final StatusMessage statusMessage =
@@ -744,4 +755,38 @@ public final class EthProtocolManagerTest {
       done.get();
     }
   }
+
+  @Test
+  public void transactionMessagesGoToTheCorrectExecutor() {
+    // Create a mock ethScheduler to hold our mock executors.
+    final ExecutorService worker = mock(ExecutorService.class);
+    final ScheduledExecutorService scheduled = mock(ScheduledExecutorService.class);
+    final ExecutorService transactions = mock(ExecutorService.class);
+    final EthScheduler ethScheduler = new EthScheduler(worker, scheduled, transactions);
+
+    // Create the fake TransactionMessage to feed to the EthManager.
+    final BlockDataGenerator gen = new BlockDataGenerator(1);
+    final List<Transaction> txes = Collections.singletonList(gen.transaction());
+    final MessageData initialMessage = TransactionsMessage.create(txes);
+    final MessageData raw = new RawMessage(EthPV62.TRANSACTIONS, initialMessage.getData());
+    final TransactionsMessage transactionMessage = TransactionsMessage.readFrom(raw);
+
+    try (final EthProtocolManager ethManager =
+        new EthProtocolManager(blockchain, 1, true, 1, ethScheduler)) {
+
+      // Create a transaction pool.  This has a side effect of registring a listener for the
+      // transactions message.
+      TransactionPoolFactory.createTransactionPool(
+          protocolSchedule, protocolContext, ethManager.ethContext());
+
+      // Send just a transaction message.
+      final PeerConnection peer = setupPeer(ethManager, (cap, msg, connection) -> {});
+      ethManager.processMessage(EthProtocol.ETH63, new DefaultMessage(peer, transactionMessage));
+
+      // Verify the regular message executor and scheduled executor got nothing to execute.
+      verifyZeroInteractions(worker, scheduled);
+      // Verify our transactions executor got something to execute.
+      verify(transactions).submit((Runnable) any());
+    }
+  }
 }

ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/manager/EthSchedulerTest.java

@@ -31,7 +31,7 @@ import org.junit.Test;
 public class EthSchedulerTest {
 
   private DeterministicEthScheduler ethScheduler;
-  private MockExecutorService workerExecutor;
+  private MockExecutorService syncWorkerExecutor;
   private MockScheduledExecutor scheduledExecutor;
   private AtomicBoolean shouldTimeout;
 
@@ -39,14 +39,14 @@ public class EthSchedulerTest {
   public void setup() {
     shouldTimeout = new AtomicBoolean(false);
     ethScheduler = new DeterministicEthScheduler(shouldTimeout::get);
-    workerExecutor = ethScheduler.mockWorkerExecutor();
+    syncWorkerExecutor = ethScheduler.mockSyncWorkerExecutor();
     scheduledExecutor = ethScheduler.mockScheduledExecutor();
   }
 
   @Test
   public void scheduleWorkerTask_completesWhenScheduledTaskCompletes() {
     final CompletableFuture<Object> future = new CompletableFuture<>();
-    final CompletableFuture<Object> result = ethScheduler.scheduleWorkerTask(() -> future);
+    final CompletableFuture<Object> result = ethScheduler.scheduleSyncWorkerTask(() -> future);
 
     assertThat(result.isDone()).isFalse();
     future.complete("bla");
@@ -58,7 +58,7 @@ public class EthSchedulerTest {
   @Test
   public void scheduleWorkerTask_completesWhenScheduledTaskFails() {
     final CompletableFuture<Object> future = new CompletableFuture<>();
-    final CompletableFuture<Object> result = ethScheduler.scheduleWorkerTask(() -> future);
+    final CompletableFuture<Object> result = ethScheduler.scheduleSyncWorkerTask(() -> future);
 
     assertThat(result.isDone()).isFalse();
     future.completeExceptionally(new RuntimeException("whoops"));
@@ -70,7 +70,7 @@ public class EthSchedulerTest {
   @Test
   public void scheduleWorkerTask_completesWhenScheduledTaskIsCancelled() {
     final CompletableFuture<Object> future = new CompletableFuture<>();
-    final CompletableFuture<Object> result = ethScheduler.scheduleWorkerTask(() -> future);
+    final CompletableFuture<Object> result = ethScheduler.scheduleSyncWorkerTask(() -> future);
 
     assertThat(result.isDone()).isFalse();
     future.cancel(false);
@@ -82,10 +82,10 @@ public class EthSchedulerTest {
   @Test
   public void scheduleWorkerTask_cancelsScheduledFutureWhenResultIsCancelled() {
     final CompletableFuture<Object> result =
-        ethScheduler.scheduleWorkerTask(() -> new CompletableFuture<>());
+        ethScheduler.scheduleSyncWorkerTask(() -> new CompletableFuture<>());
 
-    assertThat(workerExecutor.getScheduledFutures().size()).isEqualTo(1);
+    assertThat(syncWorkerExecutor.getScheduledFutures().size()).isEqualTo(1);
-    final Future<?> future = workerExecutor.getScheduledFutures().get(0);
+    final Future<?> future = syncWorkerExecutor.getScheduledFutures().get(0);
 
     verify(future, times(0)).cancel(anyBoolean());
     result.cancel(true);

ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/transactions/TestNode.java

@@ -99,7 +99,8 @@ public class TestNode implements Closeable {
     genesisState.writeStateTo(worldStateArchive.getMutable());
     final ProtocolContext<Void> protocolContext =
         new ProtocolContext<>(blockchain, worldStateArchive, null);
-    final EthProtocolManager ethProtocolManager = new EthProtocolManager(blockchain, 1, false, 1);
+    final EthProtocolManager ethProtocolManager =
+        new EthProtocolManager(blockchain, 1, false, 1, 1);
 
     final NetworkRunner networkRunner =
         NetworkRunner.builder()

pantheon/src/main/java/tech/pegasys/pantheon/controller/CliquePantheonController.java

@@ -144,7 +144,8 @@ public class CliquePantheonController implements PantheonController<CliqueContex
             protocolContext.getBlockchain(),
             networkId,
             fastSyncEnabled,
-            syncConfig.downloaderParallelism());
+            syncConfig.downloaderParallelism(),
+            syncConfig.transactionsParallelism());
     final SyncState syncState =
         new SyncState(
             protocolContext.getBlockchain(), ethProtocolManager.ethContext().getEthPeers());

pantheon/src/main/java/tech/pegasys/pantheon/controller/IbftLegacyPantheonController.java

@@ -144,7 +144,8 @@ public class IbftLegacyPantheonController implements PantheonController<IbftCont
               protocolContext.getBlockchain(),
               networkId,
               fastSyncEnabled,
-              syncConfig.downloaderParallelism());
+              syncConfig.downloaderParallelism(),
+              syncConfig.transactionsParallelism());
     } else {
       ethSubProtocol = EthProtocol.get();
       ethProtocolManager =
@@ -152,7 +153,8 @@ public class IbftLegacyPantheonController implements PantheonController<IbftCont
               protocolContext.getBlockchain(),
               networkId,
               fastSyncEnabled,
-              syncConfig.downloaderParallelism());
+              syncConfig.downloaderParallelism(),
+              syncConfig.transactionsParallelism());
     }
 
     final SyncState syncState =

pantheon/src/main/java/tech/pegasys/pantheon/controller/IbftPantheonController.java

@@ -150,7 +150,8 @@ public class IbftPantheonController implements PantheonController<IbftContext> {
             protocolContext.getBlockchain(),
             networkId,
             fastSyncEnabled,
-            syncConfig.downloaderParallelism());
+            syncConfig.downloaderParallelism(),
+            syncConfig.transactionsParallelism());
     final SubProtocol ethSubProtocol = EthProtocol.get();
 
     final SyncState syncState =

pantheon/src/main/java/tech/pegasys/pantheon/controller/MainnetPantheonController.java

@@ -117,7 +117,8 @@ public class MainnetPantheonController implements PantheonController<Void> {
                 .getChainId()
                 .orElse(MainnetProtocolSchedule.DEFAULT_CHAIN_ID),
             fastSyncEnabled,
-            syncConfig.downloaderParallelism());
+            syncConfig.downloaderParallelism(),
+            syncConfig.transactionsParallelism());
     final SyncState syncState =
         new SyncState(
             protocolContext.getBlockchain(), ethProtocolManager.ethContext().getEthPeers());