Limit length of batches (#367)

* Very basic implementation

* Finish implementation

* Enable splitting for the Node

* Add logging

* Refactor tests

Author: AlexanderShekhovcov

plenum/common/batched.py

@@ -2,6 +2,7 @@
 from typing import Any, Iterable
 
 from plenum.common.constants import BATCH, OP_FIELD_NAME
+from plenum.common.prepare_batch import split_messages_on_batches
 from stp_core.common.constants import CONNECTION_PREFIX
 from stp_core.crypto.signer import Signer
 from stp_core.common.log import getlogger
@@ -98,23 +99,24 @@ def flushOutBoxes(self) -> None:
                         "{} batching {} msgs to {} into one transmission".
                         format(self, len(msgs), dest))
                     logger.trace("    messages: {}".format(msgs))
-                    batch = Batch(list(msgs), None)
+                    batches = split_messages_on_batches(list(msgs),
+                                                        self._make_batch,
+                                                        self._test_batch_len,
+                                                        )
                     msgs.clear()
-                    # don't need to sign the batch, when the composed msgs are
-                    # signed
-                    payload, err_msg = self.signAndSerialize(batch)
-                    if payload is not None:
-                        logger.trace("{} sending payload to {}: {}".format(
-                            self, dest, payload))
-                        # Setting timeout to never expire
-                        self.transmit(
-                            payload,
-                            rid,
-                            timeout=self.messageTimeout,
-                            serialized=True)
+                    if batches:
+                        for batch in batches:
+                            logger.trace("{} sending payload to {}: {}".format(
+                                self, dest, batch))
+                            # Setting timeout to never expire
+                            self.transmit(
+                                batch,
+                                rid,
+                                timeout=self.messageTimeout,
+                                serialized=True)
                     else:
-                        logger.warning("{} error {}. tried to {}: {}".format(
-                            self, err_msg, dest, payload))
+                        logger.warning("Cannot create batch(es) for {}".format(
+                            self, dest))
         for rid in removedRemotes:
             logger.warning("{}{} rid {} has been removed"
                            .format(CONNECTION_PREFIX, self, rid),
@@ -127,6 +129,14 @@ def flushOutBoxes(self) -> None:
                              logMethod=logger.debug)
             del self.outBoxes[rid]
 
+    def _make_batch(self, msgs):
+        batch = Batch(msgs, None)
+        serialized_batch, _ = self.signAndSerialize(batch)
+        return serialized_batch
+
+    def _test_batch_len(self, batch_len):
+        return self.msg_len_val.is_len_less_than_limit(batch_len)
+
     def doProcessReceived(self, msg, frm, ident):
         if OP_FIELD_NAME in msg and msg[OP_FIELD_NAME] == BATCH:
             if f.MSGS.nm in msg and isinstance(msg[f.MSGS.nm], list):

plenum/common/prepare_batch.py

@@ -0,0 +1,45 @@
+from stp_core.common.log import getlogger
+
+SPLIT_STEPS_LIMIT = 8
+
+logger = getlogger()
+
+
+def split_messages_on_batches(msgs, make_batch_func, is_batch_len_under_limit, step_num=0):
+
+    def split(rec_depth):
+        l = len(msgs) // 2
+        left_batch = split_messages_on_batches(msgs[:l], make_batch_func, is_batch_len_under_limit, rec_depth)
+        right_batch = split_messages_on_batches(msgs[l:], make_batch_func, is_batch_len_under_limit, rec_depth)
+        return left_batch + right_batch if left_batch and right_batch else None
+
+    if step_num > SPLIT_STEPS_LIMIT:
+        logger.warning('Too many split steps '
+                       'were done {}. Batches were not created'.format(step_num))
+        return None
+
+    # precondition for case when total length is greater than limit
+    # helps skip extra serialization step
+    tt_len = sum(len(m) for m in msgs)
+    if not is_batch_len_under_limit(tt_len):
+        for m in msgs:
+            if not is_batch_len_under_limit(len(m)):
+                logger.warning('The message {} is to long ({}). '
+                               'Batches were not created'.format(m, len(m)))
+                return
+        step_num += 1
+        return split(step_num)
+
+    # make a batch and check its length
+    batch = make_batch_func(msgs)
+    if is_batch_len_under_limit(len(batch)):
+        return [batch]  # success split
+    else:
+        if len(msgs) == 1:
+            # a batch with this message greater than limit so split fails
+            logger.warning('The message {} is less than limit '
+                           'but the batch which contains only this '
+                           'message is greater than limit'.format(msgs))
+            return None
+        step_num += 1
+        return split(step_num)

plenum/test/batching_3pc/test_basic_batching.py

@@ -2,11 +2,11 @@
 
 import pytest
 
-from stp_core.loop.eventually import eventually
 from plenum.common.exceptions import UnauthorizedClientRequest
 from plenum.test.batching_3pc.helper import checkNodesHaveSameRoots
 from plenum.test.helper import checkReqNackWithReason, sendRandomRequests, \
     checkRejectWithReason, waitForSufficientRepliesForRequests
+from stp_core.loop.eventually import eventually
 
 
 def testRequestStaticValidation(tconf, looper, txnPoolNodeSet, client,

plenum/test/common/test_prepare_batch.py

@@ -0,0 +1,59 @@
+from plenum.common.prepare_batch import split_messages_on_batches, SPLIT_STEPS_LIMIT
+
+
+LEN_LIMIT_BYTES = 100
+SERIALIZATION_OTHER_HEAD_BYTES = 10
+MAX_ONE_MSG_LEN = LEN_LIMIT_BYTES - SERIALIZATION_OTHER_HEAD_BYTES
+
+
+def make_batch_func(msgs):
+    overhead = b'1' * SERIALIZATION_OTHER_HEAD_BYTES
+    return b''.join(msgs + [overhead])
+
+
+def check_batch_len_func(length):
+    return length <= LEN_LIMIT_BYTES
+
+
+def split_ut(msgs):
+    return split_messages_on_batches(msgs, make_batch_func, check_batch_len_func)
+
+
+def test_empty_msgs_returns_one_batch():
+    assert len(split_ut([])) == 1
+
+
+def test_less_than_limit_returns_one_batch():
+    msgs = [b'1'] * 10
+    assert len(split_ut(msgs)) == 1
+
+
+def test_total_len_excesses_limit_two_batches():
+    msgs = [b'1'] * (LEN_LIMIT_BYTES + 1)
+    assert len(split_ut(msgs)) == 2
+
+
+def test_each_msg_almost_excesses_limit_one_msg_per_batch():
+    count = 100
+    msgs = [b'1' * MAX_ONE_MSG_LEN] * count
+    assert len(split_ut(msgs)) == count
+
+
+def test_small_msgs_with_one_huge_more_than_one_batch():
+    msgs = [b'1', b'1', b'1', b'1' * MAX_ONE_MSG_LEN, b'1']
+    assert len(split_ut(msgs)) == 4
+
+
+def test_one_msg_excesses_limit_split_fails():
+    msgs = [b'1' * (LEN_LIMIT_BYTES + 1)]
+    assert split_ut(msgs) is None
+
+
+def test_one_msg_almost_excesses_limit_split_fails():
+    msgs = [b'1' * (MAX_ONE_MSG_LEN + 1)]
+    assert split_ut(msgs) is None
+
+
+def test_excesses_limit_of_split_steps_split_fails():
+    msgs = [b'1' * MAX_ONE_MSG_LEN] * 2**(SPLIT_STEPS_LIMIT + 1)
+    assert split_ut(msgs) is None

plenum/test/node_request/test_split_non_3pc_messages_on_batches.py

@@ -0,0 +1,74 @@
+from functools import partial
+
+import pytest
+
+from plenum.test import waits
+
+from plenum.test.helper import sendRandomRequests, waitForSufficientRepliesForRequests, checkReqAck
+from plenum.test.pool_transactions.helper import buildPoolClientAndWallet
+from stp_core.loop.eventually import eventuallyAll
+from stp_core.validators.message_length_validator import MessageLenValidator
+
+from plenum.test.pool_transactions.conftest import looper, client1Connected  # noqa
+from plenum.test.pool_transactions.conftest import clientAndWallet1, client1, wallet1  # noqa
+
+
+def test_msg_max_length_check_node_to_node(tconf,
+                                           tdir,
+                                           looper,
+                                           txnPoolNodeSet,
+                                           client1,
+                                           wallet1,
+                                           client1Connected,
+                                           clientAndWallet2):
+    """
+    Two clients send 2*N requests each at the same time.
+    N < MSG_LEN_LIMIT but 2*N > MSG_LEN_LIMIT so the requests pass the max
+    length check for client-node requests but do not pass the check
+    for node-node requests.
+    """
+    N = 10
+    # it is an empirical value for N random requests
+    # it has to be adjusted if the world changed (see pydoc)
+    max_len_limit = 3000
+
+    patch_msg_len_validators(max_len_limit, txnPoolNodeSet)
+
+    client2, wallet2 = clientAndWallet2
+
+    reqs1 = sendRandomRequests(wallet1, client1, N)
+    reqs2 = sendRandomRequests(wallet2, client2, N)
+
+    check_reqacks(client1, looper, reqs1, txnPoolNodeSet)
+    check_reqacks(client2, looper, reqs2, txnPoolNodeSet)
+
+    waitForSufficientRepliesForRequests(looper, client1, requests=reqs1)
+    waitForSufficientRepliesForRequests(looper, client2, requests=reqs2)
+
+
+def patch_msg_len_validators(max_len_limit, txnPoolNodeSet):
+    for node in txnPoolNodeSet:
+        assert hasattr(node.nodestack, 'msgLenVal')
+        assert hasattr(node.nodestack, 'msg_len_val')
+        node.nodestack.msgLenVal = MessageLenValidator(max_len_limit)
+        node.nodestack.msg_len_val = MessageLenValidator(max_len_limit)
+
+
+def check_reqacks(client, looper, reqs, txnPoolNodeSet):
+    reqack_coros = []
+    for req in reqs:
+        reqack_coros.extend([partial(checkReqAck, client, node, req.identifier,
+                                     req.reqId, None) for node in txnPoolNodeSet])
+    timeout = waits.expectedReqAckQuorumTime()
+    looper.run(eventuallyAll(*reqack_coros, totalTimeout=timeout))
+
+
+@pytest.fixture(scope="module")
+def clientAndWallet2(looper, poolTxnClientData, tdirWithPoolTxns):
+    client, wallet = buildPoolClientAndWallet(poolTxnClientData,
+                                              tdirWithPoolTxns)
+
+    looper.add(client)
+    looper.run(client.ensureConnectedToNodes())
+    yield client, wallet
+    client.stop()

stp_core/validators/message_length_validator.py

@@ -7,6 +7,9 @@ def __init__(self, max_allowed: int):
 
     def validate(self, msg: bytes):
         has_len = len(msg)
-        if has_len > self.max_allowed:
+        if not self.is_len_less_than_limit(has_len):
             raise InvalidMessageExceedingSizeException(
                 self.max_allowed, has_len)
+
+    def is_len_less_than_limit(self, l):
+        return l <= self.max_allowed