[INDY-1238] Limit zmq queue with 10000 msgs

stp_core/config.py

@@ -34,7 +34,7 @@
 MAX_SOCKETS = 16384 if sys.platform != 'win32' else None
 ENABLE_HEARTBEATS = False
 HEARTBEAT_FREQ = 5      # seconds
-ZMQ_INTERNAL_QUEUE_SIZE = 0  # messages (0 - no limit)
+ZMQ_INTERNAL_QUEUE_SIZE = 10000  # messages (0 - no limit)
 
 
 # All messages exceeding the limit will be rejected without processing

stp_zmq/test/conftest.py

@@ -47,9 +47,16 @@ def looper(tdirAndLooper):
     return tdirAndLooper[1]
 
 
+BIG_NUM_OF_MSGS = 100000
+
+
 @pytest.fixture()
 def tconf():
-    return getConfig()
+    tmp = getConfig()
+    old_num = tmp.ZMQ_INTERNAL_QUEUE_SIZE
+    tmp.ZMQ_INTERNAL_QUEUE_SIZE = BIG_NUM_OF_MSGS
+    yield tmp
+    tmp.ZMQ_INTERNAL_QUEUE_SIZE = old_num
 
 
 @pytest.fixture(scope="module")

stp_zmq/test/test_quotas.py

@@ -2,12 +2,12 @@
 
 import pytest
 from stp_core.loop.eventually import eventually
-
 from stp_core.crypto.util import randomSeed
 from stp_core.network.port_dispenser import genHa
 from stp_core.test.helper import Printer, prepStacks, CollectingMsgsHandler, CounterMsgsHandler, MessageSender
 from stp_zmq.test.helper import genKeys
 from stp_zmq.zstack import ZStack
+from stp_zmq.test.conftest import BIG_NUM_OF_MSGS
 
 
 def testMessageQuota(set_info_log_level, tdir, looper):
@@ -38,7 +38,7 @@ def checkAllReceived():
                           timeout=5))
 
 
-def testManyMessages(set_info_log_level, tdir, looper):
+def testManyMessages(set_info_log_level, tdir, looper, tconf):
     names = ['Alpha', 'Beta']
     genKeys(tdir, names)
     alphaP = Printer(names[0])
@@ -58,7 +58,7 @@ def testManyMessages(set_info_log_level, tdir, looper):
 
     looper.runFor(1)
 
-    msgNum = 100000
+    msgNum = BIG_NUM_OF_MSGS
     msgSender = MessageSender(msgNum, alpha, beta.name)
     looper.add(msgSender)
 

stp_zmq/test/test_zstack.py

@@ -8,6 +8,7 @@
     check_stacks_communicating, get_file_permission_mask, get_zstack_key_paths
 from stp_zmq.zstack import ZStack
 from stp_core.common.util import adict
+from stp_zmq.test.conftest import BIG_NUM_OF_MSGS
 
 
 def testRestricted2ZStackCommunication(tdir, looper, tconf):
@@ -132,7 +133,7 @@ def test_high_load(set_info_log_level, tdir, looper, tconf):
                'V', 'W', 'X', 'Y', 'Z']
 
     num_of_senders = 3
-    num_of_requests_per_sender = 100000
+    num_of_requests_per_sender = BIG_NUM_OF_MSGS
 
     expected_messages = []
     received_messages = []
@@ -150,13 +151,12 @@ def create_stack(name, handler=None):
     gamma = create_stack("Gamma", handler)
     prepStacks(looper, *senders, gamma, connect=True, useKeys=True)
 
-    for i in range(num_of_requests_per_sender):
-        for sender in senders:
+    for sender in senders:
+        for i in range(num_of_requests_per_sender):
             msg = {sender.name: i}
             expected_messages.append(msg)
             sender.send(msg, gamma.name)
-
-    looper.runFor(5)
+        looper.runFor(5)
 
     assert len(received_messages) != 0
     assert len(expected_messages) == len(received_messages), \