add tx/rx

Signed-off-by: TXXT <[email protected]>

Author: hunterlxt

tests/tests.rs

@@ -7,8 +7,50 @@ use std::*;
 
 use fail::fail_point;
 
+#[test]
+#[cfg_attr(not(feature = "failpoints"), ignore)]
+fn test_pause() {
+    let local_registry = fail::new_fail_group();
+    local_registry.register_current();
+    let f = || {
+        fail_point!("pause");
+    };
+    f();
+
+    fail::cfg("pause", "pause").unwrap();
+    let (tx, rx) = mpsc::channel();
+    // control `f()` is executed before next failpoint config
+    let (tx_before, rx_before) = mpsc::channel();
+    let thread_registry = local_registry.clone();
+    thread::spawn(move || {
+        thread_registry.register_current();
+        // pause
+        tx_before.send(()).unwrap();
+        tx.send(f()).unwrap();
+        // woken up by new order pause, and then pause again.
+        tx_before.send(()).unwrap();
+        tx.send(f()).unwrap();
+        // woken up by remove, and then quit immediately.
+        tx.send(f()).unwrap();
+    });
+
+    rx_before.recv().unwrap();
+    assert!(rx.recv_timeout(Duration::from_millis(800)).is_err());
+    fail::cfg("pause", "pause").unwrap();
+    rx.recv_timeout(Duration::from_millis(800)).unwrap();
+
+    assert!(rx.recv_timeout(Duration::from_millis(800)).is_err());
+    fail::remove("pause");
+    rx_before.recv().unwrap();
+    rx.recv_timeout(Duration::from_millis(800)).unwrap();
+    rx.recv_timeout(Duration::from_millis(800)).unwrap();
+}
+
 #[test]
 fn test_off() {
+    let local_registry = fail::new_fail_group();
+    local_registry.register_current();
+
     let f = || {
         fail_point!("off", |_| 2);
         0
@@ -22,6 +64,9 @@ fn test_off() {
 #[test]
 #[cfg_attr(not(feature = "failpoints"), ignore)]
 fn test_return() {
+    let local_registry = fail::new_fail_group();
+    local_registry.register_current();
+
     let f = || {
         fail_point!("return", |s: Option<String>| s
             .map_or(2, |s| s.parse().unwrap()));
@@ -39,6 +84,9 @@ fn test_return() {
 #[test]
 #[cfg_attr(not(feature = "failpoints"), ignore)]
 fn test_sleep() {
+    let local_registry = fail::new_fail_group();
+    local_registry.register_current();
+
     let f = || {
         fail_point!("sleep");
     };
@@ -56,6 +104,9 @@ fn test_sleep() {
 #[should_panic]
 #[cfg_attr(not(feature = "failpoints"), ignore)]
 fn test_panic() {
+    let local_registry = fail::new_fail_group();
+    local_registry.register_current();
+
     let f = || {
         fail_point!("panic");
     };
@@ -66,6 +117,9 @@ fn test_panic() {
 #[test]
 #[cfg_attr(not(feature = "failpoints"), ignore)]
 fn test_print() {
+    let local_registry = fail::new_fail_group();
+    local_registry.register_current();
+
     struct LogCollector(Arc<Mutex<Vec<String>>>);
     impl log::Log for LogCollector {
         fn enabled(&self, _: &log::Metadata) -> bool {
@@ -97,38 +151,11 @@ fn test_print() {
     assert_eq!(msg, "failpoint print executed.");
 }
 
-#[test]
-#[cfg_attr(not(feature = "failpoints"), ignore)]
-fn test_pause() {
-    let f = || {
-        fail_point!("pause");
-    };
-    f();
-
-    fail::cfg("pause", "pause").unwrap();
-    let (tx, rx) = mpsc::channel();
-    thread::spawn(move || {
-        // pause
-        tx.send(f()).unwrap();
-        // woken up by new order pause, and then pause again.
-        tx.send(f()).unwrap();
-        // woken up by remove, and then quit immediately.
-        tx.send(f()).unwrap();
-    });
-
-    assert!(rx.recv_timeout(Duration::from_millis(500)).is_err());
-    fail::cfg("pause", "pause").unwrap();
-    rx.recv_timeout(Duration::from_millis(500)).unwrap();
-
-    assert!(rx.recv_timeout(Duration::from_millis(500)).is_err());
-    fail::remove("pause");
-    rx.recv_timeout(Duration::from_millis(500)).unwrap();
-
-    rx.recv_timeout(Duration::from_millis(500)).unwrap();
-}
-
 #[test]
 fn test_yield() {
+    let local_registry = fail::new_fail_group();
+    local_registry.register_current();
+
     let f = || {
         fail_point!("yield");
     };
@@ -139,6 +166,9 @@ fn test_yield() {
 #[test]
 #[cfg_attr(not(feature = "failpoints"), ignore)]
 fn test_callback() {
+    let local_registry = fail::new_fail_group();
+    local_registry.register_current();
+
     let f1 = || {
         fail_point!("cb");
     };
@@ -160,6 +190,9 @@ fn test_callback() {
 #[test]
 #[cfg_attr(not(feature = "failpoints"), ignore)]
 fn test_delay() {
+    let local_registry = fail::new_fail_group();
+    local_registry.register_current();
+
     let f = || fail_point!("delay");
     let timer = Instant::now();
     fail::cfg("delay", "delay(1000)").unwrap();
@@ -170,6 +203,9 @@ fn test_delay() {
 #[test]
 #[cfg_attr(not(feature = "failpoints"), ignore)]
 fn test_freq_and_count() {
+    let local_registry = fail::new_fail_group();
+    local_registry.register_current();
+
     let f = || {
         fail_point!("freq_and_count", |s: Option<String>| s
             .map_or(2, |s| s.parse().unwrap()));
@@ -191,6 +227,9 @@ fn test_freq_and_count() {
 #[test]
 #[cfg_attr(not(feature = "failpoints"), ignore)]
 fn test_condition() {
+    let local_registry = fail::new_fail_group();
+    local_registry.register_current();
+
     let f = |_enabled| {
         fail_point!("condition", _enabled, |_| 2);
         0
@@ -205,9 +244,61 @@ fn test_condition() {
 
 #[test]
 fn test_list() {
+    let local_registry = fail::new_fail_group();
+    local_registry.register_current();
+
     assert!(!fail::list().contains(&("list".to_string(), "off".to_string())));
     fail::cfg("list", "off").unwrap();
     assert!(fail::list().contains(&("list".to_string(), "off".to_string())));
     fail::cfg("list", "return").unwrap();
     assert!(fail::list().contains(&("list".to_string(), "return".to_string())));
 }
+
+#[test]
+fn test_multiple_threads_cleanup() {
+    let local_registry = fail::new_fail_group();
+    local_registry.register_current();
+
+    let (tx, rx) = mpsc::channel();
+    thread::spawn(move || {
+        local_registry.register_current();
+        fail::cfg("thread_point", "sleep(10)").unwrap();
+        tx.send(()).unwrap();
+    });
+    rx.recv().unwrap();
+    let l = fail::list();
+    assert!(
+        l.iter()
+            .find(|&x| x == &("thread_point".to_owned(), "sleep(10)".to_owned()))
+            .is_some()
+            && l.len() == 1
+    );
+
+    let (tx, rx) = mpsc::channel();
+    let t = thread::spawn(move || {
+        let local_registry = fail::new_fail_group();
+        local_registry.register_current();
+        fail::cfg("thread_point", "panic").unwrap();
+        let l = fail::list();
+        assert!(
+            l.iter()
+                .find(|&x| x == &("thread_point".to_owned(), "panic".to_owned()))
+                .is_some()
+                && l.len() == 1
+        );
+        rx.recv().unwrap();
+        local_registry.cleanup();
+        let l = fail::list();
+        assert!(l.is_empty());
+    });
+
+    tx.send(()).unwrap();
+    let l = fail::list();
+    assert!(
+        l.iter()
+            .find(|&x| x == &("thread_point".to_owned(), "sleep(10)".to_owned()))
+            .is_some()
+            && l.len() == 1
+    );
+    t.join().unwrap();
+}