BFD: fix timing in the main loop

Properly wakeup the main process loop only when necessary to avoid
missed events. Improve BFD unit test reliability and speed. Add
timestamps to logs and replace Queue with more efficient deque.

Change-Id: I01255a02057d3146917b43b669ccdc74f31ef0c8
Signed-off-by: Klement Sekera <ksekera@cisco.com>

5 files changed, 86 insertions, 69 deletions

diff --git a/test/bfd.py b/test/bfd.py
index beacd80f687..fe63264e731 100644
--- a/test/bfd.py
+++ b/test/bfd.py
@@ -161,7 +161,8 @@ class VppBFDUDPSession(VppObject):
     def detect_mult(self):
         return self._detect_mult
 
-    def __init__(self, test, interface, peer_addr, local_addr=None, af=AF_INET):
+    def __init__(self, test, interface, peer_addr, local_addr=None, af=AF_INET,
+                 desired_min_tx=100000, required_min_rx=100000, detect_mult=3):
         self._test = test
         self._interface = interface
         self._af = af
@@ -169,9 +170,9 @@ class VppBFDUDPSession(VppObject):
         self._peer_addr = peer_addr
         self._peer_addr_n = socket.inet_pton(af, peer_addr)
         self._bs_index = None
-        self._desired_min_tx = 200000  # 0.2s
-        self._required_min_rx = 200000  # 0.2s
-        self._detect_mult = 3  # 3 packets need to be missed
+        self._desired_min_tx = desired_min_tx
+        self._required_min_rx = required_min_rx
+        self._detect_mult = detect_mult
 
     def add_vpp_config(self):
         is_ipv6 = 1 if AF_INET6 == self.af else 0
diff --git a/test/framework.py b/test/framework.py
index e1f3ba54b57..c91f34f973e 100644
--- a/test/framework.py
+++ b/test/framework.py
@@ -6,7 +6,7 @@ import tempfile
 import time
 import resource
 from time import sleep
-from Queue import Queue
+from collections import deque
 from threading import Thread
 from inspect import getdoc
 from hook import StepHook, PollHook
@@ -41,9 +41,9 @@ class _PacketInfo(object):
     data = None
 
 
-def pump_output(out, queue):
+def pump_output(out, deque):
     for line in iter(out.readline, b''):
-        queue.put(line)
+        deque.append(line)
 
 
 class VppTestCase(unittest.TestCase):
@@ -185,13 +185,13 @@ class VppTestCase(unittest.TestCase):
         # doesn't get called and we might end with a zombie vpp
         try:
             cls.run_vpp()
-            cls.vpp_stdout_queue = Queue()
+            cls.vpp_stdout_deque = deque()
             cls.vpp_stdout_reader_thread = Thread(target=pump_output, args=(
-                cls.vpp.stdout, cls.vpp_stdout_queue))
+                cls.vpp.stdout, cls.vpp_stdout_deque))
             cls.vpp_stdout_reader_thread.start()
-            cls.vpp_stderr_queue = Queue()
+            cls.vpp_stderr_deque = deque()
             cls.vpp_stderr_reader_thread = Thread(target=pump_output, args=(
-                cls.vpp.stderr, cls.vpp_stderr_queue))
+                cls.vpp.stderr, cls.vpp_stderr_deque))
             cls.vpp_stderr_reader_thread.start()
             cls.vapi = VppPapiProvider(cls.shm_prefix, cls.shm_prefix, cls)
             if cls.step:
@@ -239,27 +239,26 @@ class VppTestCase(unittest.TestCase):
                 cls.vpp.terminate()
             del cls.vpp
 
-        if hasattr(cls, 'vpp_stdout_queue'):
+        if hasattr(cls, 'vpp_stdout_deque'):
             cls.logger.info(single_line_delim)
             cls.logger.info('VPP output to stdout while running %s:',
                             cls.__name__)
             cls.logger.info(single_line_delim)
             f = open(cls.tempdir + '/vpp_stdout.txt', 'w')
-            while not cls.vpp_stdout_queue.empty():
-                line = cls.vpp_stdout_queue.get_nowait()
-                f.write(line)
-                cls.logger.info('VPP stdout: %s' % line.rstrip('\n'))
+            vpp_output = "".join(cls.vpp_stdout_deque)
+            f.write(vpp_output)
+            cls.logger.info('\n%s', vpp_output)
+            cls.logger.info(single_line_delim)
 
-        if hasattr(cls, 'vpp_stderr_queue'):
+        if hasattr(cls, 'vpp_stderr_deque'):
             cls.logger.info(single_line_delim)
             cls.logger.info('VPP output to stderr while running %s:',
                             cls.__name__)
             cls.logger.info(single_line_delim)
             f = open(cls.tempdir + '/vpp_stderr.txt', 'w')
-            while not cls.vpp_stderr_queue.empty():
-                line = cls.vpp_stderr_queue.get_nowait()
-                f.write(line)
-                cls.logger.info('VPP stderr: %s' % line.rstrip('\n'))
+            vpp_output = "".join(cls.vpp_stderr_deque)
+            f.write(vpp_output)
+            cls.logger.info('\n%s', vpp_output)
             cls.logger.info(single_line_delim)
 
     @classmethod
@@ -280,6 +279,15 @@ class VppTestCase(unittest.TestCase):
         """ Clear trace before running each test"""
         if self.vpp_dead:
             raise Exception("VPP is dead when setting up the test")
+        time.sleep(.1)
+        self.vpp_stdout_deque.append(
+            "--- test setUp() for %s.%s(%s) starts here ---\n" %
+            (self.__class__.__name__, self._testMethodName,
+             self._testMethodDoc))
+        self.vpp_stderr_deque.append(
+            "--- test setUp() for %s.%s(%s) starts here ---\n" %
+            (self.__class__.__name__, self._testMethodName,
+             self._testMethodDoc))
         self.vapi.cli("clear trace")
         # store the test instance inside the test class - so that objects
         # holding the class can access instance methods (like assertEqual)
@@ -297,14 +305,14 @@ class VppTestCase(unittest.TestCase):
             i.enable_capture()
 
     @classmethod
-    def pg_start(cls):
+    def pg_start(cls, sleep_time=1):
         """
         Enable the packet-generator and send all prepared packet streams
         Remove the packet streams afterwards
         """
         cls.vapi.cli("trace add pg-input 50")  # 50 is maximum
         cls.vapi.cli('packet-generator enable')
-        sleep(1)  # give VPP some time to process the packets
+        sleep(sleep_time)  # give VPP some time to process the packets
         for stream in cls.pg_streams:
             cls.vapi.cli('packet-generator delete %s' % stream)
         cls.pg_streams = []
diff --git a/test/log.py b/test/log.py
index 38610b7ee8b..5a6219ce973 100644
--- a/test/log.py
+++ b/test/log.py
@@ -26,7 +26,8 @@ class ColorFormatter(logging.Formatter):
         return message
 
 handler = logging.StreamHandler(sys.stdout)
-handler.setFormatter(ColorFormatter())
+handler.setFormatter(ColorFormatter(fmt='%(asctime)s,%(msecs)03d %(message)s',
+                                    datefmt="%H:%M:%S"))
 
 global_logger = logging.getLogger()
 global_logger.addHandler(handler)
diff --git a/test/test_bfd.py b/test/test_bfd.py
index 58f1264543c..b48c3cc4e46 100644
--- a/test/test_bfd.py
+++ b/test/test_bfd.py
@@ -8,12 +8,12 @@ from framework import *
 from util import ppp
 
 
-class BFDCLITestCase(VppTestCase):
-    """Bidirectional Forwarding Detection (BFD) - CLI"""
+class BFDAPITestCase(VppTestCase):
+    """Bidirectional Forwarding Detection (BFD) - API"""
 
     @classmethod
     def setUpClass(cls):
-        super(BFDCLITestCase, cls).setUpClass()
+        super(BFDAPITestCase, cls).setUpClass()
 
         try:
             cls.create_pg_interfaces([0])
@@ -21,7 +21,7 @@ class BFDCLITestCase(VppTestCase):
             cls.pg0.resolve_arp()
 
         except Exception:
-            super(BFDCLITestCase, cls).tearDownClass()
+            super(BFDAPITestCase, cls).tearDownClass()
             raise
 
     def test_add_bfd(self):
@@ -80,7 +80,7 @@ class BFDTestSession(object):
         self.interface = interface
         self.bfd_values = {
             'my_discriminator': 0,
-            'desired_min_tx_interval': 500000,
+            'desired_min_tx_interval': 100000,
             'detect_mult': detect_mult,
             'diag': BFDDiagCode.no_diagnostic,
         }
@@ -98,7 +98,7 @@ class BFDTestSession(object):
         p = self.create_packet()
         self.test.logger.debug(ppp("Sending packet:", p))
         self.test.pg0.add_stream([p])
-        self.test.pg_start()
+        self.test.pg_start(sleep_time=0)
 
     def verify_packet(self, packet):
         """ Verify correctness of BFD layer. """
@@ -116,7 +116,7 @@ class BFDTestCase(VppTestCase):
     def setUpClass(cls):
         super(BFDTestCase, cls).setUpClass()
         try:
-            cls.create_pg_interfaces([0, 1])
+            cls.create_pg_interfaces([0])
             cls.pg0.config_ip4()
             cls.pg0.generate_remote_hosts()
             cls.pg0.configure_ipv4_neighbors()
@@ -137,6 +137,7 @@ class BFDTestCase(VppTestCase):
 
     def tearDown(self):
         self.vapi.want_bfd_events(enable_disable=0)
+        self.vapi.collect_events()  # clear the event queue
         if not self.vpp_dead:
             self.vpp_session.remove_vpp_config()
         super(BFDTestCase, self).tearDown()
@@ -144,7 +145,7 @@ class BFDTestCase(VppTestCase):
     def verify_event(self, event, expected_state):
         """ Verify correctness of event values. """
         e = event
-        self.logger.debug("Event: %s" % repr(e))
+        self.logger.debug("BFD: Event: %s" % repr(e))
         self.assert_equal(e.bs_index, self.vpp_session.bs_index,
                           "BFD session index")
         self.assert_equal(e.sw_if_index, self.vpp_session.interface.sw_if_index,
@@ -166,6 +167,7 @@ class BFDTestCase(VppTestCase):
         self.assert_equal(e.state, expected_state, BFDState)
 
     def wait_for_bfd_packet(self, timeout=1):
+        self.logger.info("BFD: Waiting for BFD packet")
         p = self.pg0.wait_for_packet(timeout=timeout)
         bfd = p[BFD]
         if bfd is None:
@@ -178,22 +180,23 @@ class BFDTestCase(VppTestCase):
         return p
 
     def test_slow_timer(self):
-        """ Slow timer """
-
+        """ verify slow periodic control frames while session down """
         self.pg_enable_capture([self.pg0])
-        expected_packets = 10
-        self.logger.info("Waiting for %d BFD packets" % expected_packets)
+        expected_packets = 3
+        self.logger.info("BFD: Waiting for %d BFD packets" % expected_packets)
         self.wait_for_bfd_packet()
         for i in range(expected_packets):
             before = time.time()
             self.wait_for_bfd_packet()
             after = time.time()
+            # spec says the range should be <0.75, 1>, allow extra 0.05 margin
+            # to work around timing issues
             self.assert_in_range(
-                after - before, 0.75, 1, "time between slow packets")
+                after - before, 0.70, 1.05, "time between slow packets")
             before = after
 
     def test_zero_remote_min_rx(self):
-        """ Zero RemoteMinRxInterval """
+        """ no packets when zero BFD RemoteMinRxInterval """
         self.pg_enable_capture([self.pg0])
         p = self.wait_for_bfd_packet()
         self.test_session.update(my_discriminator=randint(0, 40000000),
@@ -210,50 +213,48 @@ class BFDTestCase(VppTestCase):
             return
         raise Exception(ppp("Received unexpected BFD packet:", p))
 
-    def bfd_conn_up(self):
+    def bfd_session_up(self):
         self.pg_enable_capture([self.pg0])
-        self.logger.info("Waiting for slow hello")
+        self.logger.info("BFD: Waiting for slow hello")
         p = self.wait_for_bfd_packet()
-        self.logger.info("Sending Init")
+        self.logger.info("BFD: Sending Init")
         self.test_session.update(my_discriminator=randint(0, 40000000),
                                  your_discriminator=p[BFD].my_discriminator,
                                  state=BFDState.init,
-                                 required_min_rx_interval=500000)
+                                 required_min_rx_interval=100000)
         self.test_session.send_packet()
-        self.logger.info("Waiting for event")
+        self.logger.info("BFD: Waiting for event")
         e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
         self.verify_event(e, expected_state=BFDState.up)
-        self.logger.info("Session is Up")
+        self.logger.info("BFD: Session is Up")
         self.test_session.update(state=BFDState.up)
 
-    def test_conn_up(self):
-        """ Basic connection up """
-        self.bfd_conn_up()
+    def test_session_up(self):
+        """ bring BFD session up """
+        self.bfd_session_up()
 
     def test_hold_up(self):
-        """ Hold BFD up """
-        self.bfd_conn_up()
+        """ hold BFD session up """
+        self.bfd_session_up()
         for i in range(5):
             self.wait_for_bfd_packet()
             self.test_session.send_packet()
 
     def test_conn_down(self):
-        """ Session down after inactivity """
-        self.bfd_conn_up()
+        """ verify session goes down after inactivity """
+        self.bfd_session_up()
         self.wait_for_bfd_packet()
-        self.assert_equal(
-            0, len(self.vapi.collect_events()),
-            "number of bfd events")
+        self.assert_equal(len(self.vapi.collect_events()), 0,
+                          "number of bfd events")
         self.wait_for_bfd_packet()
-        self.assert_equal(
-            0, len(self.vapi.collect_events()),
-            "number of bfd events")
+        self.assert_equal(len(self.vapi.collect_events()), 0,
+                          "number of bfd events")
         e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
         self.verify_event(e, expected_state=BFDState.down)
 
     @unittest.skip("this test is not working yet")
     def test_large_required_min_rx(self):
-        self.bfd_conn_up()
+        self.bfd_session_up()
         interval = 5000000
         self.test_session.update(required_min_rx_interval=interval)
         self.test_session.send_packet()
diff --git a/test/vpp_papi_provider.py b/test/vpp_papi_provider.py
index 5e80a03e2c5..c45518bd2c1 100644
--- a/test/vpp_papi_provider.py
+++ b/test/vpp_papi_provider.py
@@ -2,6 +2,7 @@ import os
 import fnmatch
 import time
 from hook import Hook
+from collections import deque
 
 # Sphinx creates auto-generated documentation by importing the python source
 # files and collecting the docstrings from them. The NO_VPP_PAPI flag allows the
@@ -48,7 +49,7 @@ class VppPapiProvider(object):
                 jsonfiles.append(os.path.join(root, filename))
 
         self.papi = VPP(jsonfiles)
-        self._events = list()
+        self._events = deque()
 
     def register_hook(self, hook):
         """Replace hook registration with new hook
@@ -59,19 +60,25 @@ class VppPapiProvider(object):
         self.hook = hook
 
     def collect_events(self):
+        """ Collect all events from the internal queue and clear the queue. """
         e = self._events
-        self._events = list()
+        self._events = deque()
         return e
 
     def wait_for_event(self, timeout, name=None):
+        """ Wait for and return next event. """
+        if self._events:
+            self.test_class.logger.debug("Not waiting, event already queued")
         limit = time.time() + timeout
         while time.time() < limit:
             if self._events:
-                e = self._events.pop(0)
+                e = self._events.popleft()
                 if name and type(e).__name__ != name:
                     raise Exception(
                         "Unexpected event received: %s, expected: %s" %
                         (type(e).__name__, name))
+                self.test_class.logger.debug("Returning event %s:%s" %
+                                             (name, e))
                 return e
             time.sleep(0)  # yield
         if name is not None:
@@ -79,8 +86,10 @@ class VppPapiProvider(object):
         raise Exception("Event did not occur within timeout")
 
     def __call__(self, name, event):
+        """ Enqueue event in the internal event queue. """
         # FIXME use the name instead of relying on type(e).__name__ ?
         # FIXME #2 if this throws, it is eaten silently, Ole?
+        self.test_class.logger.debug("New event: %s: %s" % (name, event))
         self._events.append(event)
 
     def connect(self):
@@ -93,7 +102,7 @@ class VppPapiProvider(object):
         self.papi.disconnect()
 
     def api(self, api_fn, api_args, expected_retval=0):
-        """Call API function and check it's return value
+        """ Call API function and check it's return value.
         Call the appropriate hooks before and after the API call
 
         :param api_fn: API function to call
@@ -107,14 +116,13 @@ class VppPapiProvider(object):
         if hasattr(reply, 'retval') and reply.retval != expected_retval:
             msg = "API call failed, expected retval == %d, got %s" % (
                 expected_retval, repr(reply))
-            self.test_class.logger.error(msg)
+            self.test_class.logger.info(msg)
             raise Exception(msg)
         self.hook.after_api(api_fn.__name__, api_args)
         return reply
 
     def cli(self, cli):
-        """
-        Execute a CLI, calling the before/after hooks appropriately.
+        """ Execute a CLI, calling the before/after hooks appropriately.
 
         :param cli: CLI to execute
         :returns: CLI output
@@ -128,8 +136,7 @@ class VppPapiProvider(object):
             return r.reply.decode().rstrip('\x00')
 
     def ppcli(self, cli):
-        """
-        Helping method to print CLI command in case of info logging level.
+        """ Helper method to print CLI command in case of info logging level.
 
         :param cli: CLI to execute
         :returns: CLI output
@@ -165,8 +172,7 @@ class VppPapiProvider(object):
         return self.api(self.papi.sw_interface_dump, args)
 
     def sw_interface_set_table(self, sw_if_index, is_ipv6, table_id):
-        """
-          Set the IPvX Table-id for the Interface
+        """ Set the IPvX Table-id for the Interface
 
         :param sw_if_index:
         :param is_ipv6: