diff --git a/tests/TestSuite_checksum_offload.py b/tests/TestSuite_checksum_offload.py
index 0214231c..9ba06a19 100644
--- a/tests/TestSuite_checksum_offload.py
+++ b/tests/TestSuite_checksum_offload.py
@@ -756,60 +756,6 @@ class TestChecksumOffload(TestCase):
 
         self.result_table_add(result)
 
-    def test_perf_checksum_throughtput(self):
-        """
-        Test checksum offload performance.
-        """
-        # Verify that enough ports are available
-        self.verify(len(self.dut_ports) >= 2, "Insufficient ports for testing")
-        self.dut.send_expect("quit", "#")
-
-        # sizes = [64, 128, 256, 512, 1024]
-        sizes = [64, 128]
-        pkts = {
-            "IP/UDP": 'Ether(dst="%s", src="52:00:00:00:00:00")/IP()/UDP()/("X"*(%d-46))',
-            "IP/TCP": 'Ether(dst="%s", src="52:00:00:00:00:00")/IP()/TCP()/("X"*(%d-58))',
-            "IP/SCTP": 'Ether(dst="%s", src="52:00:00:00:00:00")/IP()/SCTP()/("X"*(%d-50+2))',
-            "IPv6/UDP": 'Ether(dst="%s", src="52:00:00:00:00:00")/IPv6()/UDP()/("X"* (lambda x: x - 66 if x > 66 else 0)(%d))',
-            "IPv6/TCP": 'Ether(dst="%s", src="52:00:00:00:00:00")/IPv6()/TCP()/("X"* (lambda x: x - 78 if x > 78 else 0)(%d))',
-        }
-
-        if self.kdriver in DRIVER_TEST_LACK_CAPA["sctp_tx_offload"]:
-            del pkts["IP/SCTP"]
-
-        lcore = "1S/2C/1T"
-        portMask = utils.create_mask([self.dut_ports[0], self.dut_ports[1]])
-        for mode in ["sw", "hw"]:
-            self.logger.info("%s performance" % mode)
-            tblheader = ["Ports", "S/C/T", "Packet Type", "Mode"]
-            for size in sizes:
-                tblheader.append("%sB mpps" % str(size))
-                tblheader.append("%sB %%   " % str(size))
-            self.result_table_create(tblheader)
-            self.pmdout.start_testpmd(
-                lcore,
-                "--portmask=%s" % self.portMask
-                + " --enable-rx-cksum "
-                + "--port-topology=loop",
-                socket=self.ports_socket,
-            )
-
-            self.dut.send_expect("set fwd csum", "testpmd> ")
-            if mode == "hw":
-                self.checksum_enablehw(self.dut_ports[0])
-                self.checksum_enablehw(self.dut_ports[1])
-            else:
-                self.checksum_enablesw(self.dut_ports[0])
-                self.checksum_enablesw(self.dut_ports[1])
-
-            self.dut.send_expect("start", "testpmd> ", 3)
-            for ptype in list(pkts.keys()):
-                self.benchmark(lcore, ptype, mode, pkts[ptype], sizes, self.nic)
-
-            self.dut.send_expect("stop", "testpmd> ")
-            self.dut.send_expect("quit", "#", 10)
-            self.result_table_print()
-
     def test_hardware_checksum_check_ip_rx(self):
         self.tester.send_expect("scapy", ">>>")
         self.checksum_enablehw(self.dut_ports[0])
diff --git a/tests/TestSuite_perf_checksum_offload.py b/tests/TestSuite_perf_checksum_offload.py
new file mode 100644
index 00000000..a2288613
--- /dev/null
+++ b/tests/TestSuite_perf_checksum_offload.py
@@ -0,0 +1,624 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2010-2014 Intel Corporation
+# Copyright(c) 2018-2019 The University of New Hampshire
+#
+
+"""
+DPDK Test suite.
+
+Test support of RX/TX Checksum Offload Features by Poll Mode Drivers.
+
+"""
+
+import os
+import re
+import subprocess
+import time
+from typing import List, Pattern, Tuple, Union
+
+from scapy.layers.inet import IP, TCP, UDP
+from scapy.layers.inet6 import IPv6
+from scapy.layers.l2 import GRE, Ether
+from scapy.layers.sctp import SCTP
+from scapy.layers.vxlan import VXLAN
+from scapy.packet import Raw
+from scapy.utils import rdpcap, wrpcap
+
+import framework.packet as packet
+import framework.utils as utils
+from framework.exception import VerifyFailure
+from framework.pktgen import PacketGeneratorHelper
+from framework.pmd_output import PmdOutput
+from framework.rst import RstReport
+from framework.settings import FOLDERS
+from framework.test_capabilities import DRIVER_TEST_LACK_CAPA
+from framework.test_case import TestCase
+
+l3_proto_classes = [IP, IPv6]
+
+l4_proto_classes = [
+    UDP,
+    TCP,
+]
+
+tunnelling_proto_classes = [
+    VXLAN,
+    GRE,
+]
+
+l3_protos = ["IP", "IPv6"]
+
+l4_protos = [
+    "UDP",
+    "TCP",
+    "SCTP",
+]
+
+
+class TestChecksumOffload(TestCase):
+    def set_up_all(self):
+        """
+        Run at the start of each test suite.
+        Checksum offload prerequisites.
+        """
+        # Based on h/w type, choose how many ports to use
+        self.dut_ports = self.dut.get_ports(self.nic)
+        # Verify that enough ports are available
+        self.verify(len(self.dut_ports) >= 1, "Insufficient ports for testing")
+        self.pmdout: PmdOutput = PmdOutput(self.dut)
+        self.portMask = utils.create_mask([self.dut_ports[0]])
+        self.ports_socket = self.dut.get_numa_id(self.dut_ports[0])
+        # get dts output path
+        if self.logger.log_path.startswith(os.sep):
+            self.output_path = self.logger.log_path
+        else:
+            cur_path = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
+            self.output_path = os.sep.join([cur_path, self.logger.log_path])
+        # log debug used
+        self.count = 0
+
+    def set_up(self):
+        """
+        Run before each test case.
+        """
+        self.pmdout.start_testpmd(
+            "Default",
+            "--portmask=%s " % (self.portMask)
+            + " --enable-rx-cksum "
+            + "--port-topology=loop",
+            socket=self.ports_socket,
+        )
+        self.dut.send_expect("set verbose 1", "testpmd>")
+        self.dut.send_expect("set fwd csum", "testpmd>")
+        self.dut.send_expect("csum mac-swap off 0", "testpmd>")
+
+    def checksum_enablehw(self, port):
+        self.dut.send_expect("port stop all", "testpmd>")
+        self.dut.send_expect("rx_vxlan_port add 4789 0 ", "testpmd>")
+        self.dut.send_expect("csum set ip hw %d" % port, "testpmd>")
+        self.dut.send_expect("csum set udp hw %d" % port, "testpmd>")
+        self.dut.send_expect("csum set tcp hw %d" % port, "testpmd>")
+        self.dut.send_expect("csum set sctp hw %d" % port, "testpmd>")
+        self.dut.send_expect("csum set outer-ip hw %d" % port, "testpmd>")
+        self.dut.send_expect("csum set outer-udp hw %d" % port, "testpmd>")
+        self.dut.send_expect("csum parse-tunnel on %d" % port, "testpmd>")
+        self.dut.send_expect("port start all", "testpmd>")
+
+    def checksum_enablesw(self, port):
+        self.dut.send_expect("port stop all", "testpmd>")
+        self.dut.send_expect("csum set ip sw %d" % port, "testpmd>")
+        self.dut.send_expect("csum set udp sw %d" % port, "testpmd>")
+        self.dut.send_expect("csum set tcp sw %d" % port, "testpmd>")
+        self.dut.send_expect("csum set sctp sw %d" % port, "testpmd>")
+        self.dut.send_expect("port start all", "testpmd>")
+
+    def get_chksum_values(self, packets_expected):
+        """
+        Validate the checksum flags.
+        """
+        checksum_pattern = re.compile("chksum.*=.*(0x[0-9a-z]+)")
+
+        chksum = dict()
+
+        self.tester.send_expect("scapy", ">>> ")
+
+        for packet_type in list(packets_expected.keys()):
+            self.tester.send_expect("p = %s" % packets_expected[packet_type], ">>>")
+            out = self.tester.send_command("p.show2()", timeout=1)
+            chksums = checksum_pattern.findall(out)
+            chksum[packet_type] = chksums
+
+        self.tester.send_expect("exit()", "#")
+
+        return chksum
+
+    def checksum_valid_flags(self, packets_sent, flag):
+        """
+        Sends packets and check the checksum valid-flags.
+        """
+        self.dut.send_expect("start", "testpmd>")
+        tx_interface = self.tester.get_interface(
+            self.tester.get_local_port(self.dut_ports[0])
+        )
+        for packet_type in list(packets_sent.keys()):
+            self.pkt = packet.Packet(pkt_str=packets_sent[packet_type])
+            self.pkt.send_pkt(self.tester, tx_interface, count=4)
+            out = self.dut.get_session_output(timeout=1)
+            lines = out.split("\r\n")
+
+            # collect the checksum result
+            for line in lines:
+                line = line.strip()
+                if len(line) != 0 and line.startswith("rx"):
+                    # IPv6 don't be checksum, so always show "GOOD"
+                    if packet_type.startswith("IPv6"):
+                        if "RTE_MBUF_F_RX_L4_CKSUM" not in line:
+                            self.verify(0, "There is no checksum flags appeared!")
+                        else:
+                            if flag == 1:
+                                self.verify(
+                                    "RTE_MBUF_F_RX_L4_CKSUM_GOOD" in line,
+                                    "Packet Rx L4 checksum valid-flags error!",
+                                )
+                            elif flag == 0:
+                                self.verify(
+                                    "RTE_MBUF_F_RX_L4_CKSUM_BAD" in line
+                                    or "RTE_MBUF_F_RX_L4_CKSUM_UNKNOWN" in line,
+                                    "Packet Rx L4 checksum valid-flags error!",
+                                )
+                    else:
+                        if "RTE_MBUF_F_RX_L4_CKSUM" not in line:
+                            self.verify(0, "There is no L4 checksum flags appeared!")
+                        elif "RTE_MBUF_F_RX_IP_CKSUM" not in line:
+                            self.verify(0, "There is no IP checksum flags appeared!")
+                        else:
+                            if flag == 1:
+                                self.verify(
+                                    "RTE_MBUF_F_RX_L4_CKSUM_GOOD" in line,
+                                    "Packet Rx L4 checksum valid-flags error!",
+                                )
+                                self.verify(
+                                    "RTE_MBUF_F_RX_IP_CKSUM_GOOD" in line,
+                                    "Packet Rx IP checksum valid-flags error!",
+                                )
+                            elif flag == 0:
+                                self.verify(
+                                    "RTE_MBUF_F_RX_L4_CKSUM_BAD" in line
+                                    or "RTE_MBUF_F_RX_L4_CKSUM_UNKNOWN" in line,
+                                    "Packet Rx L4 checksum valid-flags error!",
+                                )
+                                self.verify(
+                                    "RTE_MBUF_F_RX_IP_CKSUM_BAD" in line,
+                                    "Packet Rx IP checksum valid-flags error!",
+                                )
+
+        self.dut.send_expect("stop", "testpmd>")
+
+    def checksum_validate(self, packets_sent, packets_expected):
+        """
+        Validate the checksum.
+        """
+        tx_interface = self.tester.get_interface(
+            self.tester.get_local_port(self.dut_ports[0])
+        )
+        rx_interface = self.tester.get_interface(
+            self.tester.get_local_port(self.dut_ports[0])
+        )
+
+        sniff_src = "52:00:00:00:00:00"
+        result = dict()
+
+        chksum = self.get_chksum_values(packets_expected)
+        inst = self.tester.tcpdump_sniff_packets(
+            intf=rx_interface,
+            count=len(packets_sent) * 4,
+            filters=[{"layer": "ether", "config": {"src": sniff_src}}],
+        )
+
+        self.pkt = packet.Packet()
+        for packet_type in list(packets_sent.keys()):
+            self.pkt.append_pkt(packets_sent[packet_type])
+        self.pkt.send_pkt(crb=self.tester, tx_port=tx_interface, count=4)
+
+        p = self.tester.load_tcpdump_sniff_packets(inst)
+        nr_packets = len(p)
+        print(p)
+        packets_received = [
+            p[i].sprintf("%IP.chksum%;%TCP.chksum%;%UDP.chksum%;%SCTP.chksum%")
+            for i in range(nr_packets)
+        ]
+        print(len(packets_sent), len(packets_received))
+        self.verify(
+            len(packets_sent) * 4 == len(packets_received), "Unexpected Packets Drop"
+        )
+
+        for packet_received in packets_received:
+            (
+                ip_checksum,
+                tcp_checksum,
+                udp_checksum,
+                sctp_checksum,
+            ) = packet_received.split(";")
+
+            packet_type = ""
+            l4_checksum = ""
+            if tcp_checksum != "??":
+                packet_type = "TCP"
+                l4_checksum = tcp_checksum
+            elif udp_checksum != "??":
+                packet_type = "UDP"
+                l4_checksum = udp_checksum
+            elif sctp_checksum != "??":
+                packet_type = "SCTP"
+                l4_checksum = sctp_checksum
+
+            if ip_checksum != "??":
+                packet_type = "IP/" + packet_type
+                if chksum[packet_type] != [ip_checksum, l4_checksum]:
+                    result[packet_type] = packet_type + " checksum error"
+            else:
+                packet_type = "IPv6/" + packet_type
+                if chksum[packet_type] != [l4_checksum]:
+                    result[packet_type] = packet_type + " checksum error"
+
+        return result
+
+    def send_scapy_packet(self, packet: str):
+        itf = self.tester.get_interface(self.tester.get_local_port(self.dut_ports[0]))
+
+        self.tester.scapy_foreground()
+        self.tester.scapy_append(f'sendp({packet}, iface="{itf}")')
+        return self.tester.scapy_execute()
+
+    def get_pkt_rx_l4_cksum(self, testpmd_output: str) -> bool:
+        return self.checksum_flags_are_good("RTE_MBUF_F_RX_L4_CKSUM_", testpmd_output)
+
+    def get_pkt_rx_ip_cksum(self, testpmd_output: str) -> bool:
+        return self.checksum_flags_are_good("RTE_MBUF_F_RX_IP_CKSUM_", testpmd_output)
+
+    def send_pkt_expect_good_bad_from_flag(
+        self, pkt_str: str, flag: str, test_name: str, should_pass: bool = True
+    ):
+        self.pmdout.get_output(timeout=1)  # Remove any old output
+        self.scapy_exec(f"sendp({pkt_str}, iface=iface)")
+        testpmd_output: str = self.pmdout.get_output(timeout=1)
+        self.verify(
+            flag in testpmd_output,
+            f"Flag {flag[:-1]} not found for test {test_name}, please run test_rx_checksum_valid_flags.",
+        )
+        self.verify(
+            (flag + "UNKNOWN") not in testpmd_output,
+            f"Flag {flag[:-1]} was found to be unknown for test {test_name}, indicating a possible lack of support",
+        )
+        if should_pass:
+            if flag + "GOOD" in testpmd_output:
+                return
+            else:  # flag + "BAD" in testpmd_output
+                self.verify(
+                    False, f"{flag}BAD was found in output, expecting {flag}GOOD."
+                )
+        else:
+            if flag + "BAD" in testpmd_output:
+                return
+            else:  # flag + "GOOD" in testpmd_output
+                self.verify(
+                    False, f"{flag}GOOD was found in output, expecting {flag}BAD."
+                )
+
+    def send_pkt_expect_good_bad_from_flag_catch_failure(
+        self, pkt_str: str, flag: str, test_name: str, should_pass: bool = True
+    ) -> Union[VerifyFailure, None]:
+        try:
+            self.send_pkt_expect_good_bad_from_flag(
+                pkt_str, flag, test_name, should_pass=should_pass
+            )
+        except VerifyFailure as vf:
+            return vf
+
+        return None
+
+    def validate_checksum(self, pkt, pkt_type, inner_flag=False) -> bool:
+        """
+        @param pkt: The packet to validate the checksum of.
+        @return: Whether the checksum was valid.
+        """
+        if pkt is None:
+            return False
+        for i in range(0, len(l3_protos)):
+            if l3_protos[i] in pkt:
+                l3 = l3_protos[i]
+        for j in range(0, len(l4_protos)):
+            if l4_protos[j] in pkt:
+                layer = l4_proto_classes[j]
+                csum = pkt[layer].chksum
+                if csum is None:
+                    csum = 0
+                del pkt[layer].chksum
+                # Converting it to raw will calculate the checksum
+                correct_csum = layer(bytes(Raw(pkt[layer]))).chksum
+                if correct_csum == csum:
+                    # checksum value is correct
+                    return False
+                else:
+                    if inner_flag:
+                        print(
+                            "{} pkg[{}] VXLAN/{}/{} inner checksum {} is not correct {}".format(
+                                pkt_type,
+                                self.count,
+                                l3,
+                                l4_protos[j],
+                                hex(correct_csum),
+                                hex(csum),
+                            )
+                        )
+                    else:
+                        print(
+                            "{} pkg[{}] {}/{} outer checksum {} is not correct {}".format(
+                                pkt_type,
+                                self.count,
+                                l3,
+                                l4_protos[j],
+                                hex(correct_csum),
+                                hex(csum),
+                            )
+                        )
+                    return True
+        return False
+
+    def scapy_exec(self, cmd: str, timeout=1) -> str:
+        return self.tester.send_expect(cmd, ">>>", timeout=timeout)
+
+    def get_packets(self, dut_mac, tester_mac):
+        eth = Ether(dst=dut_mac, src=tester_mac)
+        packets = []
+        checksum_options = (
+            {},
+            {"chksum": 0xF},
+        )
+        # Untunneled
+        for l3 in l3_proto_classes:
+            for l4 in l4_proto_classes:
+                for chksum in checksum_options:
+                    # The packet's data can be used to identify how the packet was constructed so avoid any issues with
+                    # ordering
+                    pkt = eth / l3() / l4(**chksum) / (b"X" * 48)
+                    # Prevents the default behavior which adds DNS headers
+                    if l4 == UDP:
+                        pkt[UDP].dport, pkt[UDP].sport = 1001, 1001
+                    packets.append(pkt)
+
+        # Tunneled
+        # VXLAN
+        for l3 in l3_proto_classes:
+            for l4 in l4_proto_classes:
+                for outer_arg in checksum_options:
+                    for inner_arg in checksum_options:
+                        pkt = (
+                            eth
+                            / l3()
+                            / UDP(**outer_arg)
+                            / VXLAN()
+                            / Ether()
+                            / l3()
+                            / l4(**inner_arg)
+                            / (b"Y" * 48)
+                        )
+                        # Prevents the default behavior which adds DNS headers
+                        if l4 == UDP:
+                            pkt[VXLAN][UDP].dport, pkt[VXLAN][UDP].sport = 1001, 1001
+                        packets.append(pkt)
+        # GRE
+        for l3 in l3_proto_classes:
+            for l4 in l4_proto_classes:
+                for chksum in checksum_options:
+                    pkt = eth / l3() / GRE() / l3() / l4(**chksum) / (b"Z" * 48)
+                    # Prevents the default behavior which adds DNS headers
+                    if l4 == UDP:
+                        pkt[GRE][UDP].dport, pkt[GRE][UDP].sport = 1001, 1001
+                    packets.append(pkt)
+
+        return packets
+
+    def send_tx_package(
+        self, packet_file_path, capture_file_path, packets, iface, dut_mac
+    ):
+        if os.path.isfile(capture_file_path):
+            os.remove(capture_file_path)
+        src_mac = "52:00:00:00:00:00"
+        self.tester.send_expect(
+            f"tcpdump -i '{iface}' ether src {src_mac} -s 0 -w {capture_file_path} -Q in &",
+            "# ",
+        )
+
+        if os.path.isfile(packet_file_path):
+            os.remove(packet_file_path)
+        wrpcap(packet_file_path, packets)
+        self.tester.session.copy_file_to(packet_file_path, packet_file_path)
+
+        # send packet
+        self.tester.send_expect("scapy", ">>>")
+        self.scapy_exec(f"packets = rdpcap('{packet_file_path}')")
+        for i in range(0, len(packets)):
+            self.scapy_exec(f"packets[{i}].show")
+            self.scapy_exec(f"sendp(packets[{i}], iface='{iface}')")
+            self.pmdout.get_output(timeout=0.5)
+            self.dut.send_expect(
+                "show port stats {}".format(self.dut_ports[0]), "testpmd>"
+            )
+        self.tester.send_expect("quit()", "# ")
+
+        time.sleep(1)
+        self.tester.send_expect("killall tcpdump", "#")
+        time.sleep(1)
+        self.tester.send_expect('echo "Cleaning buffer"', "#")
+        time.sleep(1)
+        return
+
+    def validate_packet_list_checksums(self, packets):
+        error_messages = []
+        untunnelled_error_message = (
+            f"un-tunneled checksum state for pkg[%s] with a invalid checksum."
+        )
+        vxlan_error_message = (
+            f"VXLAN tunnelled checksum state for pkg[%s]  with a invalid checksum."
+        )
+        gre_error_message = (
+            f"GRE tunnelled checksum state for pkg[%s] with a invalid checksum."
+        )
+
+        for packet in packets:
+            self.count = self.count + 1
+            payload: str
+            # try:
+            payload = packet[Raw].load.decode("utf-8")
+            # # This error usually happens with tunneling protocols, and means that an additional cast is needed
+            # except UnicodeDecodeError:
+            #     for proto in tunnelling_proto_classes:
+            if "X" in payload:
+                if self.validate_checksum(packet, "un-tunneled"):
+                    error_messages.append(untunnelled_error_message % self.count)
+            elif "Y" in payload:
+                if self.validate_checksum(
+                    packet[VXLAN][Ether], "VXLAN", inner_flag=True
+                ):
+                    error_messages.append(vxlan_error_message % self.count)
+                # Intel® Ethernet 700 Series not support outer udp checksum
+                if self.is_eth_series_nic(700):
+                    continue
+                if self.validate_checksum(packet, "VXLAN"):
+                    error_messages.append(vxlan_error_message % self.count)
+            elif "Z" in payload:
+                if self.validate_checksum(packet, "GRE"):
+                    error_messages.append(gre_error_message % self.count)
+        return error_messages
+
+    #
+    #
+    #
+    # Test Cases
+    #
+    def benchmark(self, lcore, ptype, mode, flow_format, size_list, nic):
+        """
+        Test ans report checksum offload performance for given parameters.
+        """
+        Bps = dict()
+        Pps = dict()
+        Pct = dict()
+        dmac = self.dut.get_mac_address(self.dut_ports[0])
+        dmac1 = self.dut.get_mac_address(self.dut_ports[1])
+
+        result = [2, lcore, ptype, mode]
+        for size in size_list:
+            flow = flow_format % (dmac, size)
+            pcap = os.sep.join([self.output_path, "test.pcap"])
+            self.tester.scapy_append('wrpcap("%s", [%s])' % (pcap, flow))
+            self.tester.scapy_execute()
+            flow = flow_format % (dmac1, size)
+            pcap = os.sep.join([self.output_path, "test1.pcap"])
+            self.tester.scapy_append('wrpcap("%s", [%s])' % (pcap, flow))
+            self.tester.scapy_execute()
+
+            tgenInput = []
+            pcap = os.sep.join([self.output_path, "test.pcap"])
+            tgenInput.append(
+                (
+                    self.tester.get_local_port(self.dut_ports[0]),
+                    self.tester.get_local_port(self.dut_ports[1]),
+                    pcap,
+                )
+            )
+            pcap = os.sep.join([self.output_path, "test1.pcap"])
+            tgenInput.append(
+                (
+                    self.tester.get_local_port(self.dut_ports[1]),
+                    self.tester.get_local_port(self.dut_ports[0]),
+                    pcap,
+                )
+            )
+
+            # clear streams before add new streams
+            self.tester.pktgen.clear_streams()
+            # create an instance to set stream field setting
+            # Moved here because it messes with the ability of the functional tests to use scapy.
+            self.pktgen_helper = PacketGeneratorHelper()
+            # run packet generator
+            streams = self.pktgen_helper.prepare_stream_from_tginput(
+                tgenInput, 100, None, self.tester.pktgen
+            )
+            Bps[str(size)], Pps[str(size)] = self.tester.pktgen.measure_throughput(
+                stream_ids=streams
+            )
+            self.verify(Pps[str(size)] > 0, "No traffic detected")
+            Pps[str(size)] /= 1e6
+            Pct[str(size)] = (Pps[str(size)] * 100) / self.wirespeed(self.nic, size, 2)
+
+            result.append(Pps[str(size)])
+            result.append(Pct[str(size)])
+
+        self.result_table_add(result)
+
+    def test_perf_checksum_throughtput(self):
+        """
+        Test checksum offload performance.
+        """
+        # Verify that enough ports are available
+        self.verify(len(self.dut_ports) >= 2, "Insufficient ports for testing")
+        self.dut.send_expect("quit", "#")
+
+        # sizes = [64, 128, 256, 512, 1024]
+        sizes = [64, 128]
+        pkts = {
+            "IP/UDP": 'Ether(dst="%s", src="52:00:00:00:00:00")/IP()/UDP()/("X"*(%d-46))',
+            "IP/TCP": 'Ether(dst="%s", src="52:00:00:00:00:00")/IP()/TCP()/("X"*(%d-58))',
+            "IP/SCTP": 'Ether(dst="%s", src="52:00:00:00:00:00")/IP()/SCTP()/("X"*(%d-50+2))',
+            "IPv6/UDP": 'Ether(dst="%s", src="52:00:00:00:00:00")/IPv6()/UDP()/("X"* (lambda x: x - 66 if x > 66 else 0)(%d))',
+            "IPv6/TCP": 'Ether(dst="%s", src="52:00:00:00:00:00")/IPv6()/TCP()/("X"* (lambda x: x - 78 if x > 78 else 0)(%d))',
+        }
+
+        if self.kdriver in DRIVER_TEST_LACK_CAPA["sctp_tx_offload"]:
+            del pkts["IP/SCTP"]
+
+        lcore = "1S/2C/1T"
+        portMask = utils.create_mask([self.dut_ports[0], self.dut_ports[1]])
+        for mode in ["sw", "hw"]:
+            self.logger.info("%s performance" % mode)
+            tblheader = ["Ports", "S/C/T", "Packet Type", "Mode"]
+            for size in sizes:
+                tblheader.append("%sB mpps" % str(size))
+                tblheader.append("%sB %%   " % str(size))
+            self.result_table_create(tblheader)
+            self.pmdout.start_testpmd(
+                lcore,
+                "--portmask=%s" % self.portMask
+                + " --enable-rx-cksum "
+                + "--port-topology=loop",
+                socket=self.ports_socket,
+            )
+
+            self.dut.send_expect("set fwd csum", "testpmd> ")
+            if mode == "hw":
+                self.checksum_enablehw(self.dut_ports[0])
+                self.checksum_enablehw(self.dut_ports[1])
+            else:
+                self.checksum_enablesw(self.dut_ports[0])
+                self.checksum_enablesw(self.dut_ports[1])
+
+            self.dut.send_expect("start", "testpmd> ", 3)
+            for ptype in list(pkts.keys()):
+                self.benchmark(lcore, ptype, mode, pkts[ptype], sizes, self.nic)
+
+            self.dut.send_expect("stop", "testpmd> ")
+            self.dut.send_expect("quit", "#", 10)
+            self.result_table_print()
+
+    def tear_down(self):
+        """
+        Run after each test case.
+        """
+        self.dut.send_expect("quit", "#")
+
+    def tear_down_all(self):
+        """
+        Run after each test suite.
+        """
+        pass
