[dualtor] Fix test_link_failure (#5771)

What is the motivation for this PR?
For recent failures of test_link_failure.py:

test_active_link_down_upstream
test_active_link_down_downstream_active
test_active_link_down_downstream_standby
test_standby_link_down_upstream
How did you do it?
The failure of those four test cases is caused that one of the fanout connects to upper ToR being unreachable. The first three cases will fail when they are trying to shut down the fanout ports connected to the failed fanout, but they doesn't recover those ports that are successfully shut down. So those ports will be left admin down, so test_standby_link_down_upstream will suffer from the data plane IO issue.
Improvements:

Fix those fixtures that shut down fanout ports so that even if one fanout fails, those fixtures could recover those ports on other fanouts.
Improve the data plane IO so if the action errors, the sender/sniffer will stop earlier.
How did you verify/test it?

tests/common/dualtor/data_plane_utils.py

@@ -163,7 +163,15 @@ def run_test(
         logger.info("Sender and sniffer threads started, ready to execute the "\
             "callback action")
         time.sleep(15)
-        action()
+
+        try:
+            action()
+        except Exception as error:
+            logging.error("Caught exception %s during action.", repr(error))
+            tor_IO.stop_early = True
+            send_and_sniff.join()
+            raise
+
     # do not time-wait the test, if early stop is not requested (when stop_after=None)
     if stop_after is not None:
         wait_until(timeout=stop_after, interval=0.5, delay=0, condition=\

tests/common/dualtor/dual_tor_utils.py

@@ -342,6 +342,9 @@ def lower_tor_fanouthosts(lower_tor_host, fanouthosts):
     return _get_tor_fanouthosts(lower_tor_host, fanouthosts)
 
 
+fanout_intfs_to_recover = defaultdict(list)
+
+
 def _shutdown_fanout_tor_intfs(tor_host, tor_fanouthosts, tbinfo, dut_intfs=None):
     """Helper function for shutting down fanout interfaces that are connected to specified DUT interfaces.
 
@@ -392,6 +395,7 @@ def _shutdown_fanout_tor_intfs(tor_host, tor_fanouthosts, tbinfo, dut_intfs=None
 
     for fanout_host, intf_list in fanout_shut_intfs.items():
         fanout_host.shutdown(intf_list)
+        fanout_intfs_to_recover[fanout_host].extend(intf_list)
 
     return fanout_shut_intfs
 
@@ -410,6 +414,7 @@ def shutdown_fanout_upper_tor_intfs(upper_tor_host, upper_tor_fanouthosts, tbinf
         function: A function for shutting down fanout interfaces connected to specified upper_tor interfaces
     """
     shut_fanouts = []
+    fanout_intfs_to_recover.clear()
 
     def shutdown(dut_intfs=None):
         logger.info('Shutdown fanout ports connected to upper_tor')
@@ -419,9 +424,9 @@ def shutdown(dut_intfs=None):
 
     logger.info('Recover fanout ports connected to upper_tor')
 
-    for instance in shut_fanouts:
-        for fanout_host, intf_list in instance.items():
-            fanout_host.no_shutdown(intf_list)
+    for fanout_host, intf_list in fanout_intfs_to_recover.items():
+        fanout_host.no_shutdown(intf_list)
+    fanout_intfs_to_recover.clear()
 
 
 @pytest.fixture
@@ -438,6 +443,7 @@ def shutdown_fanout_lower_tor_intfs(lower_tor_host, lower_tor_fanouthosts, tbinf
         function: A function for shutting down fanout interfaces connected to specified lower_tor interfaces
     """
     shut_fanouts = []
+    fanout_intfs_to_recover.clear()
 
     def shutdown(dut_intfs=None):
         logger.info('Shutdown fanout ports connected to lower_tor')
@@ -447,9 +453,9 @@ def shutdown(dut_intfs=None):
 
     logger.info('Recover fanout ports connected to lower_tor')
 
-    for instance in shut_fanouts:
-        for fanout_host, intf_list in instance.items():
-            fanout_host.no_shutdown(intf_list)
+    for fanout_host, intf_list in fanout_intfs_to_recover.items():
+        fanout_host.no_shutdown(intf_list)
+    fanout_intfs_to_recover.clear()
 
 
 @pytest.fixture
@@ -467,6 +473,7 @@ def shutdown_fanout_tor_intfs(upper_tor_host, upper_tor_fanouthosts, lower_tor_h
         function: A function for shutting down fanout interfaces connected to specified lower_tor interfaces
     """
     down_intfs = []
+    fanout_intfs_to_recover.clear()
 
     def shutdown(dut_intfs=None, upper=False, lower=False):
         if not upper and not lower:
@@ -484,8 +491,9 @@ def shutdown(dut_intfs=None, upper=False, lower=False):
     yield shutdown
 
     logger.info('Recover fanout ports connected to tor')
-    for fanout_host, fanout_intf in down_intfs:
-        fanout_host.no_shutdown(fanout_intf)
+    for fanout_host, intf_list in fanout_intfs_to_recover.items():
+        fanout_host.no_shutdown(intf_list)
+    fanout_intfs_to_recover.clear()
 
 
 def _shutdown_t1_tor_intfs(tor_host, nbrhosts, tbinfo, vm_names=None):