cancel the waiting injector

Signed-off-by: YangKeao <yangkeao@chunibyo.icu>

Cargo.lock

@@ -23,7 +23,7 @@ version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ee49baf6cb617b853aa8d93bf420db2383fab46d314482ca2803b40d5fde979b"
 dependencies = [
- "winapi",
+ "winapi 0.3.9",
 ]
 
 [[package]]
@@ -32,7 +32,7 @@ version = "0.12.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d52a9bb7ec0cf484c551830a7ce27bd20d67eac647e1befb56b0be4ee39a55d2"
 dependencies = [
- "winapi",
+ "winapi 0.3.9",
 ]
 
 [[package]]
@@ -60,7 +60,7 @@ checksum = "d9b39be18770d11421cdb1b9947a45dd3f37e93092cbf377614828a319d5fee8"
 dependencies = [
  "hermit-abi",
  "libc",
- "winapi",
+ "winapi 0.3.9",
 ]
 
 [[package]]
@@ -87,6 +87,12 @@ version = "0.5.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0e4cec68f03f32e44924783795810fa50a7035d8c8ebe78580ad7e6c703fba38"
 
+[[package]]
+name = "bytes"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c4872d67bab6358e59559027aa3b9157c53d9358c51423c17554809a8858e0f8"
+
 [[package]]
 name = "cc"
 version = "1.0.66"
@@ -115,7 +121,7 @@ dependencies = [
  "num-integer",
  "num-traits",
  "time",
- "winapi",
+ "winapi 0.3.9",
 ]
 
 [[package]]
@@ -204,6 +210,22 @@ version = "1.0.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"
 
+[[package]]
+name = "fuchsia-zircon"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2e9763c69ebaae630ba35f74888db465e49e259ba1bc0eda7d06f4a067615d82"
+dependencies = [
+ "bitflags",
+ "fuchsia-zircon-sys",
+]
+
+[[package]]
+name = "fuchsia-zircon-sys"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3dcaa9ae7725d12cdb85b3ad99a434db70b468c09ded17e012d86b5c1010f7a7"
+
 [[package]]
 name = "fuser"
 version = "0.6.0"
@@ -368,6 +390,15 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "iovec"
+version = "0.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b2b3ea6ff95e175473f8ffe6a7eb7c00d054240321b84c57051175fe3c1e075e"
+dependencies = [
+ "libc",
+]
+
 [[package]]
 name = "itertools"
 version = "0.9.0"
@@ -383,6 +414,16 @@ version = "0.4.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "dd25036021b0de88a0aff6b850051563c6516d0bf53f8638938edbb9de732736"
 
+[[package]]
+name = "kernel32-sys"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7507624b29483431c0ba2d82aece8ca6cdba9382bff4ddd0f7490560c056098d"
+dependencies = [
+ "winapi 0.2.8",
+ "winapi-build",
+]
+
 [[package]]
 name = "lazy_static"
 version = "1.4.0"
@@ -446,6 +487,80 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "mio"
+version = "0.6.23"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4afd66f5b91bf2a3bc13fad0e21caedac168ca4c707504e75585648ae80e4cc4"
+dependencies = [
+ "cfg-if 0.1.10",
+ "fuchsia-zircon",
+ "fuchsia-zircon-sys",
+ "iovec",
+ "kernel32-sys",
+ "libc",
+ "log",
+ "miow 0.2.2",
+ "net2",
+ "slab",
+ "winapi 0.2.8",
+]
+
+[[package]]
+name = "mio-named-pipes"
+version = "0.1.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0840c1c50fd55e521b247f949c241c9997709f23bd7f023b9762cd561e935656"
+dependencies = [
+ "log",
+ "mio",
+ "miow 0.3.7",
+ "winapi 0.3.9",
+]
+
+[[package]]
+name = "mio-uds"
+version = "0.6.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "afcb699eb26d4332647cc848492bbc15eafb26f08d0304550d5aa1f612e066f0"
+dependencies = [
+ "iovec",
+ "libc",
+ "mio",
+]
+
+[[package]]
+name = "miow"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ebd808424166322d4a38da87083bfddd3ac4c131334ed55856112eb06d46944d"
+dependencies = [
+ "kernel32-sys",
+ "net2",
+ "winapi 0.2.8",
+ "ws2_32-sys",
+]
+
+[[package]]
+name = "miow"
+version = "0.3.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9f1c5b025cda876f66ef43a113f91ebc9f4ccef34843000e0adf6ebbab84e21"
+dependencies = [
+ "winapi 0.3.9",
+]
+
+[[package]]
+name = "net2"
+version = "0.2.37"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "391630d12b68002ae1e25e8f974306474966550ad82dac6886fb8910c19568ae"
+dependencies = [
+ "cfg-if 0.1.10",
+ "libc",
+ "winapi 0.3.9",
+]
+
 [[package]]
 name = "nix"
 version = "0.18.0"
@@ -742,6 +857,15 @@ dependencies = [
  "lazy_static",
 ]
 
+[[package]]
+name = "signal-hook-registry"
+version = "1.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e51e73328dc4ac0c7ccbda3a494dfa03df1de2f46018127f60c693f2648455b0"
+dependencies = [
+ "libc",
+]
+
 [[package]]
 name = "slab"
 version = "0.4.2"
@@ -850,7 +974,7 @@ checksum = "6db9e6914ab8b1ae1c260a4ae7a49b6c5611b40328a735b21862567685e73255"
 dependencies = [
  "libc",
  "wasi 0.10.0+wasi-snapshot-preview1",
- "winapi",
+ "winapi 0.3.9",
 ]
 
 [[package]]
@@ -880,7 +1004,8 @@ dependencies = [
  "structopt",
  "thiserror",
  "time",
- "tokio",
+ "tokio 0.2.24",
+ "tokio-util",
  "tracing",
  "tracing-futures",
  "tracing-subscriber",
@@ -892,12 +1017,56 @@ version = "0.2.24"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "099837d3464c16a808060bb3f02263b412f6fafcb5d01c533d309985fbeebe48"
 dependencies = [
- "bytes",
+ "bytes 0.5.6",
  "fnv",
+ "futures-core",
+ "iovec",
+ "lazy_static",
+ "libc",
  "memchr",
+ "mio",
+ "mio-named-pipes",
+ "mio-uds",
  "num_cpus",
  "pin-project-lite 0.1.11",
+ "signal-hook-registry",
  "slab",
+ "tokio-macros",
+ "winapi 0.3.9",
+]
+
+[[package]]
+name = "tokio"
+version = "1.15.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fbbf1c778ec206785635ce8ad57fe52b3009ae9e0c9f574a728f3049d3e55838"
+dependencies = [
+ "pin-project-lite 0.2.4",
+]
+
+[[package]]
+name = "tokio-macros"
+version = "0.2.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e44da00bfc73a25f814cd8d7e57a68a5c31b74b3152a0a1d1f590c97ed06265a"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "tokio-util"
+version = "0.6.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9e99e1983e5d376cd8eb4b66604d2e99e79f5bd988c3055891dcd8c9e2604cc0"
+dependencies = [
+ "bytes 1.1.0",
+ "futures-core",
+ "futures-sink",
+ "log",
+ "pin-project-lite 0.2.4",
+ "tokio 1.15.0",
 ]
 
 [[package]]
@@ -1037,6 +1206,12 @@ version = "0.10.0+wasi-snapshot-preview1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1a143597ca7c7793eff794def352d41792a93c481eb1042423ff7ff72ba2c31f"
 
+[[package]]
+name = "winapi"
+version = "0.2.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "167dc9d6949a9b857f3451275e911c3f44255842c1f7a76f33c55103a909087a"
+
 [[package]]
 name = "winapi"
 version = "0.3.9"
@@ -1047,6 +1222,12 @@ dependencies = [
  "winapi-x86_64-pc-windows-gnu",
 ]
 
+[[package]]
+name = "winapi-build"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2d315eee3b34aca4797b2da6b13ed88266e6d612562a0c46390af8299fc699bc"
+
 [[package]]
 name = "winapi-i686-pc-windows-gnu"
 version = "0.4.0"
@@ -1059,7 +1240,7 @@ version = "0.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "70ec6ce85bb158151cae5e5c87f95a8e97d2c0c4b001223f33a334e3ce5de178"
 dependencies = [
- "winapi",
+ "winapi 0.3.9",
 ]
 
 [[package]]
@@ -1067,3 +1248,13 @@ name = "winapi-x86_64-pc-windows-gnu"
 version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
+
+[[package]]
+name = "ws2_32-sys"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d59cefebd0c892fa2dd6de581e937301d8552cb44489cdff035c6187cb63fa5e"
+dependencies = [
+ "winapi 0.2.8",
+ "winapi-build",
+]

Cargo.toml

@@ -14,7 +14,8 @@ fuser = {version = "0.6", features = ["abi-7-19"]}
 time = "0.1"
 libc = "0.2"
 async-trait = "0.1"
-tokio = {version = "0.2", features = ["rt-core", "rt-threaded", "sync", "fs", "time", "blocking"]}
+tokio = {version = "0.2", features = ["rt-core", "rt-threaded", "sync", "fs", "time", "blocking", "macros", "full"]}
+tokio-util = "0.6"
 thiserror = "1.0"
 futures = "0.3"
 derive_more = "0.99.9"

src/hookfs/mod.rs

@@ -23,13 +23,13 @@ use nix::fcntl::{open, readlink, renameat, OFlag};
 use nix::sys::stat;
 use nix::sys::statfs;
 use nix::unistd::{
-    fchownat, fsync, linkat, mkdir, symlinkat, truncate, unlink, AccessFlags, FchownatFlags, Gid,
+    close, fchownat, fsync, linkat, mkdir, symlinkat, truncate, unlink, AccessFlags, FchownatFlags,
-    LinkatFlags, Uid, close,
+    Gid, LinkatFlags, Uid,
 };
 
 use tracing::{debug, error, instrument, trace};
 
-use std::{collections::{HashMap, LinkedList}};
+use std::collections::{HashMap, LinkedList};
 use std::ffi::{CString, OsStr, OsString};
 use std::os::unix::ffi::OsStrExt;
 use std::os::unix::io::RawFd;
@@ -81,8 +81,7 @@ macro_rules! inject_with_fh {
 }
 
 macro_rules! inject_write_data {
-    ($self:ident, $fh:ident, $data:ident) => {
+    ($self:ident, $fh:ident, $data:ident) => {{
-        {
         let opened_files = $self.opened_files.read().await;
         if let Ok(file) = opened_files.get($fh as usize) {
             let path = file.original_path().to_owned();
@@ -92,8 +91,7 @@ macro_rules! inject_write_data {
                 .inject_write_data($self.rebuild_path(path)?.as_path(), &mut $data)?;
             trace!("Write data after inject {:?}", $data);
         }
-        }
+    }};
-    };
 }
 
 macro_rules! inject_with_dir_fh {
@@ -326,6 +324,8 @@ impl HookFs {
 
     pub fn disable_injection(&self) {
         self.enable_injection.store(false, Ordering::SeqCst);
+
+        self.injector.interrupt();
     }
 
     pub fn rebuild_path<P: AsRef<Path>>(&self, path: P) -> Result<PathBuf> {

src/injector/filter.rs

@@ -109,8 +109,7 @@ impl Filter {
             })
             .unwrap_or(Method::all());
 
-        let path_filter = conf
+        let path_filter = conf.path.and_then(|path| -> Option<Pattern> {
-            .path.and_then(|path| -> Option<Pattern> {
             if !path.is_empty() {
                 Pattern::new(&path).ok()
             } else {

src/injector/injector_config.rs

@@ -88,7 +88,7 @@ pub struct Timespec {
 #[serde(rename_all = "camelCase")]
 pub enum MistakeType {
     Zero,
-    Random
+    Random,
 }
 
 #[derive(Serialize, Deserialize, Clone, Debug)]

src/injector/latency_injector.rs

@@ -8,13 +8,17 @@ use super::injector_config::LatencyConfig;
 use super::Injector;
 use crate::hookfs::Result;
 
+use tokio::select;
 use tokio::time::delay_for;
+use tokio_util::sync::CancellationToken;
 use tracing::{debug, trace};
 
 #[derive(Debug)]
 pub struct LatencyInjector {
     latency: Duration,
     filter: filter::Filter,
+
+    cancel_token: CancellationToken,
 }
 
 #[async_trait]
@@ -22,13 +26,27 @@ impl Injector for LatencyInjector {
     async fn inject(&self, method: &filter::Method, path: &Path) -> Result<()> {
         trace!("test for filter");
         if self.filter.filter(method, path) {
-            debug!("inject io delay {:?}", self.latency);
+            let token = self.cancel_token.clone();
-            delay_for(self.latency).await;
+            let latency = self.latency;
+            debug!("inject io delay {:?}", latency);
+
+            select! {
+                _ = delay_for(latency) => {}
+                _ = token.cancelled() => {
+                    debug!("cancelled");
+                }
+            }
+
             debug!("latency finished");
         }
 
         Ok(())
     }
+
+    fn interrupt(&self) {
+        debug!("interrupt latency");
+        self.cancel_token.cancel();
+    }
 }
 
 impl LatencyInjector {
@@ -38,6 +56,7 @@ impl LatencyInjector {
         Ok(Self {
             latency: conf.latency,
             filter: filter::Filter::build(conf.filter)?,
+            cancel_token: CancellationToken::new(),
         })
     }
 }

src/injector/mod.rs

@@ -14,7 +14,7 @@ use crate::hookfs::{Reply, Result};
 use async_trait::async_trait;
 use fuser::FileAttr;
 
-use std::{path::Path};
+use std::path::Path;
 
 #[async_trait]
 pub trait Injector: Send + Sync + std::fmt::Debug {
@@ -28,13 +28,11 @@ pub trait Injector: Send + Sync + std::fmt::Debug {
     ) -> Result<()> {
         Ok(())
     }
-    fn inject_write_data(
+    fn inject_write_data(&self, _path: &Path, _data: &mut Vec<u8>) -> Result<()> {
-        &self,
-        _path: &Path,
-        _data: &mut Vec<u8>,
-    ) -> Result<()> {
         Ok(())
     }
 
     fn inject_attr(&self, _attr: &mut FileAttr, _path: &Path) {}
+
+    fn interrupt(&self) {}
 }

src/injector/multi_injector.rs

@@ -69,14 +69,16 @@ impl Injector for MultiInjector {
         }
     }
 
-    fn inject_write_data(
+    fn inject_write_data(&self, path: &Path, data: &mut Vec<u8>) -> Result<()> {
-        &self,
-        path: &Path,
-        data: &mut Vec<u8>,
-    ) -> Result<()> {
         for injector in self.injectors.iter() {
             injector.inject_write_data(path, data)?;
         }
         Ok(())
     }
+
+    fn interrupt(&self) {
+        for injector in self.injectors.iter() {
+            injector.interrupt();
+        }
+    }
 }

src/ptrace/mod.rs

@@ -1,4 +1,3 @@
-use Error::Internal;
 use anyhow::{anyhow, Result};
 use nix::sys::ptrace;
 use nix::sys::signal::Signal;
@@ -10,16 +9,21 @@ use nix::{
     sys::mman::{MapFlags, ProtFlags},
     Error::Sys,
 };
+use Error::Internal;
 
-use procfs::{ProcError, process::Task};
+use procfs::{process::Task, ProcError};
-use retry::{Error::{self, Operation}, OperationResult, delay::Fixed};
+use retry::{
+    delay::Fixed,
+    Error::{self, Operation},
+    OperationResult,
+};
 use tracing::{error, info, instrument, trace, warn};
 
-use std::{cell::RefCell, collections::HashSet};
 use std::collections::HashMap;
 use std::ffi::CString;
 use std::os::unix::ffi::OsStrExt;
 use std::path::Path;
+use std::{cell::RefCell, collections::HashSet};
 
 // There should be only one PtraceManager in one thread. But as we don't implement TLS
 // , we cannot use thread-local variables safely.
@@ -56,8 +60,8 @@ fn attach_task(task: &Task) -> Result<()> {
         }
         Err(err) => {
             warn!("attach error: {:?}", err);
-            return Err(err.into())
+            return Err(err.into());
-        },
+        }
         _ => {}
     }
     info!("attach task: {} successfully", task.tid);
@@ -93,7 +97,7 @@ impl PtraceManager {
                     let process = procfs::process::Process::new(raw_pid)?;
                     for task in (process.tasks()?).flatten() {
                         if traced_tasks.contains(&task.tid) {
-                            continue
+                            continue;
                         }
 
                         if let Ok(()) = attach_task(&task) {
@@ -127,8 +131,9 @@ impl PtraceManager {
                     counter_ref.remove(&pid);
 
                     info!("detach process: {}", pid);
-                    if let Err(err) = retry::retry::<_, _, _, anyhow::Error, _>(Fixed::from_millis(500).take(20), || {
+                    if let Err(err) = retry::retry::<_, _, _, anyhow::Error, _>(
-                        match procfs::process::Process::new(pid) {
+                        Fixed::from_millis(500).take(20),
+                        || match procfs::process::Process::new(pid) {
                             Err(ProcError::NotFound(_)) => {
                                 info!("process {} not found", pid);
                                 OperationResult::Ok(())
@@ -137,11 +142,8 @@ impl PtraceManager {
                                 warn!("fail to detach task: {}, retry", pid);
                                 OperationResult::Retry(err.into())
                             }
-                            Ok(process) => {
+                            Ok(process) => match process.tasks() {
-                                match process.tasks() {
+                                Err(err) => OperationResult::Retry(err.into()),
-                                    Err(err) => {
-                                        OperationResult::Retry(err.into())
-                                    }
                                 Ok(tasks) => {
                                     for task in tasks.flatten() {
                                         match ptrace::detach(Pid::from_raw(task.tid), None) {
@@ -161,16 +163,16 @@ impl PtraceManager {
                                     info!("detach process: {} successfully", pid);
                                     OperationResult::Ok(())
                                 }
-                                }
+                            },
-                                
+                        },
-                            }
+                    ) {
-                        }
-                    }) {
                         warn!("fail to detach: {:?}", err);
                         match err {
-                            Operation {error: e, total_delay:_, tries:_, } => {
+                            Operation {
-                                return Err(e)
+                                error: e,
-                            },
+                                total_delay: _,
+                                tries: _,
+                            } => return Err(e),
                             Internal(err) => {
                                 error!("internal error: {:?}", err)
                             }