Implement max message size rather than limiting with fixed number of transactions

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/messages/DefaultTransactionMessageLimiter.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2019 ConsenSys AG.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package tech.pegasys.pantheon.ethereum.eth.messages;
+
+import tech.pegasys.pantheon.ethereum.rlp.BytesValueRLPOutput;
+
+public class DefaultTransactionMessageLimiter implements TransactionsMessageLimiter {
+
+  private static final int LIMIT = 1048576;
+
+  @Override
+  public boolean isLastTransactionToIncludeWithoutExceedingLimit(final BytesValueRLPOutput tmp) {
+    return tmp.uncheckedEncodedSize() >= LIMIT;
+  }
+}

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/messages/LimitedTransactionsMessages.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2019 ConsenSys AG.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package tech.pegasys.pantheon.ethereum.eth.messages;
+
+import tech.pegasys.pantheon.ethereum.core.Transaction;
+
+import java.util.Set;
+
+public final class LimitedTransactionsMessages {
+
+  private final TransactionsMessage transactionsMessage;
+  private final Set<Transaction> includedTransactions;
+
+  public LimitedTransactionsMessages(
+      final TransactionsMessage transactionsMessage, final Set<Transaction> includedTransactions) {
+    this.transactionsMessage = transactionsMessage;
+    this.includedTransactions = includedTransactions;
+  }
+
+  public final TransactionsMessage getTransactionsMessage() {
+    return transactionsMessage;
+  }
+
+  public final Set<Transaction> getIncludedTransactions() {
+    return includedTransactions;
+  }
+}

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/messages/TransactionsMessage.java

@@ -20,7 +20,9 @@
 import tech.pegasys.pantheon.ethereum.rlp.RLPInput;
 import tech.pegasys.pantheon.util.bytes.BytesValue;
 
+import java.util.HashSet;
 import java.util.Iterator;
+import java.util.Set;
 import java.util.function.Function;
 
 public class TransactionsMessage extends AbstractMessageData {
@@ -47,7 +49,24 @@ public static TransactionsMessage create(final Iterable<Transaction> transaction
     return new TransactionsMessage(tmp.encoded());
   }
 
-  private TransactionsMessage(final BytesValue data) {
+  public static LimitedTransactionsMessages createLimited(
+      final Iterable<Transaction> transactions, final TransactionsMessageLimiter limiter) {
+    final Set<Transaction> includedTransactions = new HashSet<>();
+    final BytesValueRLPOutput tmp = new BytesValueRLPOutput();
+    tmp.startList();
+    for (final Transaction transaction : transactions) {
+      transaction.writeTo(tmp);
+      includedTransactions.add(transaction);
+      if (limiter.isLastTransactionToIncludeWithoutExceedingLimit(tmp)) {
+        break;
+      }
+    }
+    tmp.endList();
+    return new LimitedTransactionsMessages(
+        new TransactionsMessage(tmp.encoded()), includedTransactions);
+  }
+
+  TransactionsMessage(final BytesValue data) {
     super(data);
   }
 

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/messages/TransactionsMessageLimiter.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2019 ConsenSys AG.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package tech.pegasys.pantheon.ethereum.eth.messages;
+
+import tech.pegasys.pantheon.ethereum.rlp.BytesValueRLPOutput;
+
+public interface TransactionsMessageLimiter {
+
+  boolean isLastTransactionToIncludeWithoutExceedingLimit(final BytesValueRLPOutput tmp);
+}

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

@@ -16,6 +16,7 @@
 import tech.pegasys.pantheon.ethereum.core.PendingTransactions;
 import tech.pegasys.pantheon.ethereum.core.TransactionPool;
 import tech.pegasys.pantheon.ethereum.eth.manager.EthContext;
+import tech.pegasys.pantheon.ethereum.eth.messages.DefaultTransactionMessageLimiter;
 import tech.pegasys.pantheon.ethereum.eth.messages.EthPV62;
 import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
 import tech.pegasys.pantheon.metrics.MetricsSystem;
@@ -36,7 +37,7 @@ public static TransactionPool createTransactionPool(
 
     final PeerTransactionTracker transactionTracker = new PeerTransactionTracker();
     final TransactionsMessageSender transactionsMessageSender =
-        new TransactionsMessageSender(transactionTracker);
+        new TransactionsMessageSender(transactionTracker, new DefaultTransactionMessageLimiter());
 
     final TransactionPool transactionPool =
         new TransactionPool(

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

@@ -12,36 +12,42 @@
  */
 package tech.pegasys.pantheon.ethereum.eth.transactions;
 
-import static java.util.stream.Collectors.toSet;
-
 import tech.pegasys.pantheon.ethereum.core.Transaction;
 import tech.pegasys.pantheon.ethereum.eth.manager.EthPeer;
+import tech.pegasys.pantheon.ethereum.eth.messages.LimitedTransactionsMessages;
 import tech.pegasys.pantheon.ethereum.eth.messages.TransactionsMessage;
+import tech.pegasys.pantheon.ethereum.eth.messages.TransactionsMessageLimiter;
 import tech.pegasys.pantheon.ethereum.p2p.api.PeerConnection.PeerNotConnected;
 
 import java.util.Set;
+import java.util.stream.StreamSupport;
 
 class TransactionsMessageSender {
 
-  private static final int MAX_BATCH_SIZE = 10;
   private final PeerTransactionTracker transactionTracker;
+  private final TransactionsMessageLimiter transactionsMessageLimiter;
 
-  public TransactionsMessageSender(final PeerTransactionTracker transactionTracker) {
+  public TransactionsMessageSender(
+      final PeerTransactionTracker transactionTracker,
+      final TransactionsMessageLimiter transactionsMessageLimiter) {
     this.transactionTracker = transactionTracker;
+    this.transactionsMessageLimiter = transactionsMessageLimiter;
   }
 
   public void sendTransactionsToPeers() {
-    transactionTracker.getEthPeersWithUnsentTransactions().forEach(this::sendTransactionsToPeer);
+    StreamSupport.stream(transactionTracker.getEthPeersWithUnsentTransactions().spliterator(), true)
+        .parallel()
+        .forEach(this::sendTransactionsToPeer);
   }
 
   private void sendTransactionsToPeer(final EthPeer peer) {
     final Set<Transaction> allTxToSend = transactionTracker.claimTransactionsToSendToPeer(peer);
     while (!allTxToSend.isEmpty()) {
-      final Set<Transaction> subsetToSend =
-          allTxToSend.stream().limit(MAX_BATCH_SIZE).collect(toSet());
-      allTxToSend.removeAll(subsetToSend);
+      final LimitedTransactionsMessages limitedTransactionsMessages =
+          TransactionsMessage.createLimited(allTxToSend, transactionsMessageLimiter);
+      allTxToSend.removeAll(limitedTransactionsMessages.getIncludedTransactions());
       try {
-        peer.send(TransactionsMessage.create(subsetToSend));
+        peer.send(limitedTransactionsMessages.getTransactionsMessage());
       } catch (final PeerNotConnected e) {
         return;
       }

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

@@ -15,17 +15,20 @@
 import static com.google.common.collect.Sets.newHashSet;
 import static java.util.stream.Collectors.toSet;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.argThat;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
 
 import tech.pegasys.pantheon.ethereum.core.BlockDataGenerator;
 import tech.pegasys.pantheon.ethereum.core.Transaction;
 import tech.pegasys.pantheon.ethereum.eth.manager.EthPeer;
 import tech.pegasys.pantheon.ethereum.eth.messages.EthPV62;
 import tech.pegasys.pantheon.ethereum.eth.messages.TransactionsMessage;
+import tech.pegasys.pantheon.ethereum.eth.messages.TransactionsMessageLimiter;
 import tech.pegasys.pantheon.ethereum.p2p.api.MessageData;
 
 import java.util.List;
@@ -40,14 +43,17 @@ public class TransactionsMessageSenderTest {
 
   private final EthPeer peer1 = mock(EthPeer.class);
   private final EthPeer peer2 = mock(EthPeer.class);
+  private final TransactionsMessageLimiter transactionsMessageLimiter =
+      mock(TransactionsMessageLimiter.class);
+
   private final BlockDataGenerator generator = new BlockDataGenerator();
   private final Transaction transaction1 = generator.transaction();
   private final Transaction transaction2 = generator.transaction();
   private final Transaction transaction3 = generator.transaction();
 
   private final PeerTransactionTracker transactionTracker = new PeerTransactionTracker();
   private final TransactionsMessageSender messageSender =
-      new TransactionsMessageSender(transactionTracker);
+      new TransactionsMessageSender(transactionTracker, transactionsMessageLimiter);
 
   @Test
   public void shouldSendTransactionsToEachPeer() throws Exception {
@@ -63,14 +69,16 @@ public void shouldSendTransactionsToEachPeer() throws Exception {
   }
 
   @Test
-  public void shouldSendTransactionsInBatches() throws Exception {
-    final Set<Transaction> fifteenTransactions =
-        IntStream.range(0, 15).mapToObj(number -> generator.transaction()).collect(toSet());
-    fifteenTransactions.forEach(
+  public void shouldSendTransactionsInBatchesWithLimit() throws Exception {
+    final Set<Transaction> threeTransactions =
+        IntStream.range(0, 3).mapToObj(number -> generator.transaction()).collect(toSet());
+    threeTransactions.forEach(
         transaction -> transactionTracker.addToPeerSendQueue(peer1, transaction));
 
+    when(transactionsMessageLimiter.isLastTransactionToIncludeWithoutExceedingLimit(any()))
+        .thenReturn(false)
+        .thenReturn(true);
     messageSender.sendTransactionsToPeers();
-
     final ArgumentCaptor<MessageData> messageDataArgumentCaptor =
         ArgumentCaptor.forClass(MessageData.class);
     verify(peer1, times(2)).send(messageDataArgumentCaptor.capture());
@@ -82,10 +90,10 @@ public void shouldSendTransactionsInBatches() throws Exception {
     final Set<Transaction> firstBatch = getTransactionsFromMessage(sentMessages.get(0));
     final Set<Transaction> secondBatch = getTransactionsFromMessage(sentMessages.get(1));
 
-    assertThat(firstBatch).hasSize(10);
-    assertThat(secondBatch).hasSize(5);
+    assertThat(firstBatch).hasSize(2);
+    assertThat(secondBatch).hasSize(1);
 
-    assertThat(Sets.union(firstBatch, secondBatch)).isEqualTo(fifteenTransactions);
+    assertThat(Sets.union(firstBatch, secondBatch)).isEqualTo(threeTransactions);
   }
 
   private MessageData transactionsMessageContaining(final Transaction... transactions) {

ethereum/rlp/src/main/java/tech/pegasys/pantheon/ethereum/rlp/AbstractRLPOutput.java

@@ -141,6 +141,10 @@ public int encodedSize() {
     return payloadSizes[0];
   }
 
+  public int uncheckedEncodedSize() {
+    return payloadSizes[0];
+  }
+
   /**
    * Write the rlp encoded value to the provided {@link MutableBytesValue}
    *