[pfcwd] add an option to support C code send_packets function in pfc_gen.py

tests/common/helpers/pfc_gen.py

@@ -2,7 +2,6 @@
 
 """
 Script to generate PFC packets.
-
 """
 import binascii
 import sys
@@ -11,8 +10,52 @@
 import logging.handlers
 import time
 import multiprocessing
-
 from socket import socket, AF_PACKET, SOCK_RAW
+from ctypes import (
+    CDLL,
+    POINTER,
+    Structure,
+    c_int,
+    c_size_t,
+    c_uint,
+    c_uint32,
+    c_void_p,
+    cast,
+    get_errno,
+    pointer,
+)
+
+
+class struct_iovec(Structure):
+    _fields_ = [
+        ("iov_base", c_void_p),
+        ("iov_len", c_size_t),
+    ]
+
+
+class struct_msghdr(Structure):
+    _fields_ = [
+        ("msg_name", c_void_p),
+        ("msg_namelen", c_uint32),
+        ("msg_iov", POINTER(struct_iovec)),
+        ("msg_iovlen", c_size_t),
+        ("msg_control", c_void_p),
+        ("msg_controllen", c_size_t),
+        ("msg_flags", c_int),
+    ]
+
+
+class struct_mmsghdr(Structure):
+    _fields_ = [
+        ("msg_hdr", struct_msghdr),
+        ("msg_len", c_uint)
+    ]
+
+
+libc = CDLL("libc.so.6")
+_sendmmsg = libc.sendmmsg
+_sendmmsg.argtypes = [c_int, POINTER(struct_mmsghdr), c_uint, c_int]
+_sendmmsg.restype = c_int
 
 logger = logging.getLogger('MyLogger')
 logger.setLevel(logging.DEBUG)
@@ -25,9 +68,12 @@ class PacketSender():
     """
     A class to send PFC pause frames
     """
-    def __init__(self, interfaces, packet, num, interval):
+    def __init__(self, interfaces, packet, num, interval, use_c_function):
         # Create RAW socket to send PFC pause frames
         self.sockets = []
+        self.interfaces = interfaces
+        self.use_c_function = use_c_function
+
         try:
             for interface in interfaces:
                 s = socket(AF_PACKET, SOCK_RAW)
@@ -36,6 +82,7 @@ def __init__(self, interfaces, packet, num, interval):
         except Exception as e:
             print("Unable to create socket. Check your permissions: %s" % e)
             sys.exit(1)
+
         self.packet_num = num
         self.packet_interval = interval
         self.process = None
@@ -50,8 +97,41 @@ def send_packets(self):
                     time.sleep(self.packet_interval)
             iteration -= 1
 
+    def send_packets_mmsg(self):
+        total_packets = self.packet_num
+        batch_size = 1024
+        m_msghdr = (struct_mmsghdr * batch_size)()
+
+        iov = struct_iovec(cast(self.packet, c_void_p), len(self.packet))
+
+        msg_iov = pointer(iov)
+        msg_iovlen = 1
+        msg_control = 0
+        msg_controllen = 0
+
+        msg_namelen = 0
+        msg_name = cast(None, c_void_p)
+
+        for i in range(batch_size):
+            msghdr = struct_msghdr(
+                        msg_name, msg_namelen, msg_iov, msg_iovlen,
+                        msg_control, msg_controllen, 0)
+            m_msghdr[i] = struct_mmsghdr(msghdr)
+
+        for s in self.sockets:
+            packets_sent = 0
+            while packets_sent < total_packets:
+                num_sent = _sendmmsg(s.fileno(), m_msghdr, min(batch_size, total_packets - packets_sent), 0)
+                if num_sent < 0:
+                    errno = get_errno()
+                    print('sendmmsg got errno {} for socket {}'.format(errno, s.getsockname()))
+                    break
+                else:
+                    packets_sent += num_sent
+
     def start(self):
-        self.process = multiprocessing.Process(target=self.send_packets)
+        self.process = multiprocessing.Process(
+            target=self.send_packets_mmsg if self.use_c_function else self.send_packets)
         self.process.start()
 
     def stop(self, timeout=None):
@@ -65,7 +145,7 @@ def main():
     usage = "usage: %prog [options] arg1 arg2"
     parser = optparse.OptionParser(usage=usage)
     parser.add_option("-i", "--interface", type="string", dest="interface",
-                      help="Interface list to send packets, seperated by ','", metavar="Interface")
+                      help="Interface list to send packets, separated by ','", metavar="Interface")
     parser.add_option('-p', "--priority", type="int", dest="priority",
                       help="PFC class enable bitmap.", metavar="Priority", default=-1)
     parser.add_option("-t", "--time", type="int", dest="time",
@@ -78,6 +158,8 @@ def main():
                       help="Send global pause frames (not PFC)", default=False)
     parser.add_option("-s", "--send_pfc_frame_interval", type="float", dest="send_pfc_frame_interval",
                       help="Interval sending pfc frame", metavar="send_pfc_frame_interval", default=0)
+    parser.add_option("-c", "--use_c_function", action="store_true", dest="use_c_function",
+                      help="Use C function to send packets", default=False)
 
     (options, args) = parser.parse_args()
 
@@ -93,7 +175,6 @@ def main():
 
     if options.global_pf:
         # Send global pause frames
-        # -p option should not be set
         if options.priority != -1:
             print("'-p' option is not valid when sending global pause frames ('--global' / '-g')")
             parser.print_help()
@@ -172,19 +253,29 @@ def main():
                 packet = packet + b"\x00\x00"
 
     pre_str = 'GLOBAL_PF' if options.global_pf else 'PFC'
-    logger.debug(pre_str + '_STORM_START')
 
     # Start sending PFC pause frames
     senders = []
     interface_slices = [[] for i in range(MAX_PROCESS_NUM)]
     for i in range(0, len(interfaces)):
         interface_slices[i % MAX_PROCESS_NUM].append(interfaces[i])
 
+    logger.debug(pre_str + '_STORM_DEBUG')
+
     for interface_slice in interface_slices:
         if (interface_slice):
-            s = PacketSender(interface_slice, packet, options.num, options.send_pfc_frame_interval)
-            s.start()
-            senders.append(s)
+            if options.use_c_function and len(interface_slice) > 1:
+                for interface in interface_slice:
+                    s = PacketSender([interface], packet, options.num,
+                                     options.send_pfc_frame_interval, options.use_c_function)
+                    s.start()
+                    senders.append(s)
+            else:
+                s = PacketSender(interface_slice, packet, options.num,
+                                 options.send_pfc_frame_interval, options.use_c_function)
+                s.start()
+                senders.append(s)
+    logger.debug(pre_str + '_STORM_START')
 
     # Wait PFC packets to be sent
     for sender in senders:

tests/common/helpers/pfc_storm.py

@@ -41,6 +41,7 @@ def __init__(self, duthost, fanout_graph_facts, fanouthosts, **kwargs):
         self.fanout_info = fanout_graph_facts
         self.fanout_hosts = fanouthosts
         self.pfc_gen_file = kwargs.pop('pfc_gen_file', "pfc_gen.py")
+        self.pfc_gen_c_function_used = kwargs.pop('pfc_gen_c_function_used', False)
         self.pfc_queue_idx = kwargs.pop('pfc_queue_index', 3)
         self.pfc_frames_number = kwargs.pop('pfc_frames_number', 100000)
         self.send_pfc_frame_interval = kwargs.pop('send_pfc_frame_interval', 0)
@@ -115,7 +116,7 @@ def _create_pfc_gen(self):
 
     def deploy_pfc_gen(self):
         """
-        Deploy the pfc generation file on the fanout
+        Deploy the PFC generation file(s) on the fanout
         """
         if self.peer_device.os in ('eos', 'sonic'):
             src_pfc_gen_file = "common/helpers/{}".format(self.pfc_gen_file)
@@ -183,6 +184,8 @@ def _update_template_args(self):
             self.extra_vars.update({"pfc_asym": self.pfc_asym})
         if self.fanout_asic_type == 'mellanox' and self.peer_device.os == 'sonic':
             self.extra_vars.update({"pfc_fanout_label_port": self._generate_mellanox_label_ports()})
+        if self.pfc_gen_c_function_used:
+            self.extra_vars.update({"pfc_gen_c_function_used": True})
 
     def _prepare_start_template(self):
         """

tests/common/templates/pfc_storm_eos.j2

@@ -1,9 +1,9 @@
 bash
 cd /mnt/flash
 {% if (pfc_asym  is defined) and (pfc_asym == True) %}
-{% if pfc_storm_defer_time is defined %} sleep {{pfc_storm_defer_time}} &&{% endif %} sudo python {{pfc_gen_file}} -p {{pfc_queue_index}} -t 65535 -n {{pfc_frames_number}} -i {{pfc_fanout_interface | replace("Ethernet", "et") | replace("/", "_")}} &
+{% if pfc_storm_defer_time is defined %} sleep {{pfc_storm_defer_time}} &&{% endif %} sudo python {{pfc_gen_file}} {% if pfc_gen_c_function_used is defined %}-c {% endif %}-p {{pfc_queue_index}} -t 65535 -n {{pfc_frames_number}} -i {{pfc_fanout_interface | replace("Ethernet", "et") | replace("/", "_")}} &
 {% else %}
-{% if pfc_storm_defer_time is defined %} sleep {{pfc_storm_defer_time}} &&{% endif %} sudo python {{pfc_gen_file}} -p {{(1).__lshift__(pfc_queue_index)}} -t 65535 -n {{pfc_frames_number}} -i {{pfc_fanout_interface | replace("Ethernet", "et") | replace("/", "_")}} -r {{ansible_eth0_ipv4_addr}} &
+{% if pfc_storm_defer_time is defined %} sleep {{pfc_storm_defer_time}} &&{% endif %} sudo python {{pfc_gen_file}} {% if pfc_gen_c_function_used is defined %}-c {% endif %}-p {{(1).__lshift__(pfc_queue_index)}} -t 65535 -n {{pfc_frames_number}} -i {{pfc_fanout_interface | replace("Ethernet", "et") | replace("/", "_")}} -r {{ansible_eth0_ipv4_addr}} &
 {% endif %}
 exit
 exit

tests/common/templates/pfc_storm_stop_eos.j2

@@ -1,9 +1,9 @@
 bash
 cd /mnt/flash
 {% if (pfc_asym  is defined) and (pfc_asym == True) %}
-{% if pfc_storm_defer_time is defined %} sleep {{pfc_storm_defer_time}} &&{% endif %} sudo pkill -f "python {{pfc_gen_file}} -p {{pfc_queue_index}} -t 65535 -n {{pfc_frames_number}} -i {{pfc_fanout_interface | replace("Ethernet", "et") | replace("/", "_")}}" {{'&' if pfc_storm_stop_defer_time is defined else ''}}
+{% if pfc_storm_defer_time is defined %} sleep {{pfc_storm_defer_time}} &&{% endif %} sudo pkill -f "python {{pfc_gen_file}} {% if pfc_gen_c_function_used is defined %}-c {% endif %}-p {{pfc_queue_index}} -t 65535 -n {{pfc_frames_number}} -i {{pfc_fanout_interface | replace("Ethernet", "et") | replace("/", "_")}}" {{'&' if pfc_storm_stop_defer_time is defined else ''}}
 {% else %}
-{% if pfc_storm_stop_defer_time is defined %} sleep {{pfc_storm_stop_defer_time}} &&{% endif %} sudo pkill -f "python {{pfc_gen_file}} -p {{(1).__lshift__(pfc_queue_index)}} -t 65535 -n {{pfc_frames_number}} -i {{pfc_fanout_interface | replace("Ethernet", "et") | replace("/", "_")}} -r {{ansible_eth0_ipv4_addr}}" {{'&' if pfc_storm_stop_defer_time is defined else ''}}
+{% if pfc_storm_stop_defer_time is defined %} sleep {{pfc_storm_stop_defer_time}} &&{% endif %} sudo pkill -f "python {{pfc_gen_file}} {% if pfc_gen_c_function_used is defined %}-c {% endif %}-p {{(1).__lshift__(pfc_queue_index)}} -t 65535 -n {{pfc_frames_number}} -i {{pfc_fanout_interface | replace("Ethernet", "et") | replace("/", "_")}} -r {{ansible_eth0_ipv4_addr}}" {{'&' if pfc_storm_stop_defer_time is defined else ''}}
 {% endif %}
 exit
 exit

tests/pfcwd/test_pfcwd_timer_accuracy.py

@@ -1,8 +1,8 @@
 import logging
 import pytest
 import time
+import re
 
-from tests.common.errors import RunAnsibleModuleFail
 from tests.common.fixtures.conn_graph_facts import enum_fanout_graph_facts      # noqa F401
 from tests.common.helpers.assertions import pytest_assert
 from tests.common.helpers.pfc_storm import PFCStorm
@@ -153,15 +153,19 @@ def set_storm_params(dut, fanout_info, fanout, peer_params):
     pfc_queue_index = 4
     pfc_frames_count = 1000000
     peer_device = peer_params['peerdevice']
+    pfc_gen_c_function_used = False
     if dut.topo_type == 't2' and fanout[peer_device].os == 'sonic':
         pfc_gen_file = 'pfc_gen_t2.py'
         pfc_send_time = 8
     else:
         pfc_gen_file = 'pfc_gen.py'
+        pfc_gen_c_function_used = True
         pfc_send_time = None
+
     storm_handle = PFCStorm(dut, fanout_info, fanout, pfc_queue_idx=pfc_queue_index,
                             pfc_frames_number=pfc_frames_count, pfc_gen_file=pfc_gen_file,
-                            pfc_send_period=pfc_send_time, peer_info=peer_params)
+                            pfc_send_period=pfc_send_time, peer_info=peer_params,
+                            pfc_gen_c_function_used=pfc_gen_c_function_used)
     storm_handle.deploy_pfc_gen()
     return storm_handle
 
@@ -193,17 +197,32 @@ def run_test(self, setup_info):
         else:
             storm_start_ms = self.retrieve_timestamp("[P]FC_STORM_START")
             storm_detect_ms = self.retrieve_timestamp("[d]etected PFC storm")
+
+        if storm_detect_ms is None or storm_start_ms is None:
+            logger.info("PFC storm detect timestamp not found in syslog")
+            return
+
         logger.info("Wait for PFC storm end marker to appear in logs")
         time.sleep(16)
         if self.dut.topo_type == 't2' and self.storm_handle.peer_device.os == 'sonic':
             storm_restore_ms = self.retrieve_timestamp("[s]torm restored")
         else:
             storm_end_ms = self.retrieve_timestamp("[P]FC_STORM_END")
             storm_restore_ms = self.retrieve_timestamp("[s]torm restored")
+
+        if storm_restore_ms is None or storm_end_ms is None:
+            logger.info("PFC storm restore timestamp not found in syslog")
+            return
+
+        if self.dut.topo_type != 't2' or self.storm_handle.peer_device.os != 'sonic':
+            real_storm_duration_time = storm_end_ms - storm_start_ms
             real_detect_time = storm_detect_ms - storm_start_ms
             real_restore_time = storm_restore_ms - storm_end_ms
+            self.all_storm_duration_time.append(real_storm_duration_time)
             self.all_detect_time.append(real_detect_time)
             self.all_restore_time.append(real_restore_time)
+            logger.info("storm_duration_time {} detect_time {} restore_time {}".format(
+                real_storm_duration_time, real_detect_time, real_restore_time))
 
         dut_detect_restore_time = storm_restore_ms - storm_detect_ms
         logger.info(
@@ -218,11 +237,16 @@ def verify_pfcwd_timers(self):
         """
         Compare the timestamps obtained and verify the timer accuracy
         """
+        logger.info("all_detect_time {}".format(self.all_detect_time))
+        logger.info("all_restore_time {}".format(self.all_restore_time))
+        logger.info("all_storm_duration_time {}".format(self.all_storm_duration_time))
+
         self.all_detect_time.sort()
         self.all_restore_time.sort()
         logger.info("Verify that real detection time is not greater than configured")
-        logger.info("all detect time {}".format(self.all_detect_time))
-        logger.info("all restore time {}".format(self.all_restore_time))
+        logger.info("sorted all_detect_time {}".format(self.all_detect_time))
+        logger.info("sorted all_restore_time {}".format(self.all_restore_time))
+
         config_detect_time = self.timers['pfc_wd_detect_time'] + self.timers['pfc_wd_poll_time']
         err_msg = ("Real detection time is greater than configured: Real detect time: {} "
                    "Expected: {} (wd_detect_time + wd_poll_time)".format(self.all_detect_time[9],
@@ -278,17 +302,24 @@ def retrieve_timestamp(self, pattern):
         Returns:
             timestamp_ms (int): syslog timestamp in ms for the line matching the pattern
         """
-        cmd = "grep \"{}\" /var/log/syslog".format(pattern)
-        syslog_msg_list = self.dut.shell(cmd)['stdout'].split()
         try:
-            timestamp_ms = float(self.dut.shell("date -d \"{}\" +%s%3N".format(syslog_msg_list[3]))['stdout'])
-        except RunAnsibleModuleFail:
-            timestamp_ms = float(self.dut.shell("date -d \"{}\" +%s%3N".format(syslog_msg_list[2]))['stdout'])
-        except Exception as e:
-            logging.error("Error when parsing syslog message timestamp: {}".format(repr(e)))
-            pytest.fail("Failed to parse syslog message timestamp")
+            cmd = "grep \"{}\" /var/log/syslog".format(pattern)
+            syslog_msg = self.dut.shell(cmd)['stdout']
+
+            # Regular expressions for the two timestamp formats
+            regex1 = re.compile(r'^[A-Za-z]{3} \d{2} \d{2}:\d{2}:\d{2}\.\d{6}')
+            regex2 = re.compile(r'^\d{4} [A-Za-z]{3} \d{2} \d{2}:\d{2}:\d{2}\.\d{6}')
 
-        return int(timestamp_ms)
+            if regex1.match(syslog_msg):
+                timestamp = syslog_msg.replace('  ', ' ').split(' ')[2]
+            elif regex2.match(syslog_msg):
+                timestamp = syslog_msg.replace('  ', ' ').split(' ')[3]
+
+            timestamp_ms = self.dut.shell("date -d {} +%s%3N".format(timestamp))['stdout']
+            return int(timestamp_ms)
+        except Exception as e:
+            logger.warning("Get {} timestamp: An unexpected error occurred: {}".format(pattern, str(e)))
+            return None
 
     def test_pfcwd_timer_accuracy(self, duthosts, ptfhost, enum_rand_one_per_hwsku_frontend_hostname,
                                   pfcwd_timer_setup_restore):
@@ -305,6 +336,7 @@ def test_pfcwd_timer_accuracy(self, duthosts, ptfhost, enum_rand_one_per_hwsku_f
         self.timers = setup_info['timers']
         self.dut = duthost
         self.ptf = ptfhost
+        self.all_storm_duration_time = list()
         self.all_detect_time = list()
         self.all_restore_time = list()
         self.all_dut_detect_restore_time = list()