diff --git a/vnt-jni/java/top/wherewego/vnt/jni/Vnt.java b/vnt-jni/java/top/wherewego/vnt/jni/Vnt.java index a2bd45a..e931a56 100644 --- a/vnt-jni/java/top/wherewego/vnt/jni/Vnt.java +++ b/vnt-jni/java/top/wherewego/vnt/jni/Vnt.java @@ -11,7 +11,7 @@ import java.io.IOException; public class Vnt implements Closeable { private final long raw; - public Vnt(Config config, CallBack callBack) throws Exception{ + public Vnt(Config config, CallBack callBack) throws Exception { this.raw = new0(config, callBack); if (this.raw == 0) { throw new RuntimeException(); @@ -26,6 +26,10 @@ public class Vnt implements Closeable { wait0(raw); } + public boolean awaitTimeout(long ms) { + return waitTimeout0(raw, ms); + } + public PeerRouteInfo[] list() { return list0(raw); } @@ -36,6 +40,8 @@ public class Vnt implements Closeable { private native void wait0(long raw); + private native boolean waitTimeout0(long raw, long ms); + private native void drop0(long raw); private native PeerRouteInfo[] list0(long raw); diff --git a/vnt-jni/src/vnt.rs b/vnt-jni/src/vnt.rs index d1a0196..03fe4c5 100644 --- a/vnt-jni/src/vnt.rs +++ b/vnt-jni/src/vnt.rs @@ -1,8 +1,9 @@ use std::ptr; +use std::time::Duration; use jni::errors::Error; use jni::objects::{JClass, JObject, JValue}; -use jni::sys::{jint, jlong, jobject, jobjectArray, jsize}; +use jni::sys::{jboolean, jint, jlong, jobject, jobjectArray, jsize}; use jni::JNIEnv; use vnt::channel::Route; @@ -74,6 +75,16 @@ pub unsafe extern "C" fn Java_top_wherewego_vnt_jni_Vnt_wait0( let vnt = raw_vnt as *mut Vnt; let _ = (&*vnt).wait(); } +#[no_mangle] +pub unsafe extern "C" fn Java_top_wherewego_vnt_jni_Vnt_waitTimeout0( + _env: JNIEnv, + _class: JClass, + raw_vnt: jlong, + time: jlong, +) -> jboolean { + let vnt = raw_vnt as *mut Vnt; + (&*vnt).wait_timeout(Duration::from_millis(time as _)) as _ +} #[no_mangle] pub unsafe extern "C" fn Java_top_wherewego_vnt_jni_Vnt_drop0( diff --git a/vnt/src/core/conn.rs b/vnt/src/core/conn.rs index e51600a..6feb6b3 100644 --- a/vnt/src/core/conn.rs +++ b/vnt/src/core/conn.rs @@ -412,4 +412,7 @@ impl Vnt { pub fn wait(&self) { self.stop_manager.wait() } + pub fn wait_timeout(&self, dur: Duration) -> bool { + self.stop_manager.wait_timeout(dur) + } } diff --git a/vnt/src/handle/tun_tap/tun_handler.rs b/vnt/src/handle/tun_tap/tun_handler.rs index c918b01..bfa18b9 100644 --- a/vnt/src/handle/tun_tap/tun_handler.rs +++ b/vnt/src/handle/tun_tap/tun_handler.rs @@ -86,15 +86,16 @@ pub fn start( mut up_counter: SingleU64Adder, device_list: Arc)>>, ) -> io::Result<()> { + #[cfg(any(target_os = "windows", target_os = "linux", target_os = "macos"))] let worker = { - #[cfg(any(target_os = "macos", target_os = "android"))] + #[cfg(target_os = "macos")] let current_device = current_device.clone(); let device = device.clone(); stop_manager.add_listener("tun_device".into(), move || { if let Err(e) = device.shutdown() { log::warn!("{:?}", e); } - #[cfg(any(target_os = "macos", target_os = "android"))] + #[cfg(target_os = "macos")] { let ip = current_device.load().virtual_ip; if let Ok(udp) = std::net::UdpSocket::bind("0.0.0.0:0") { @@ -150,6 +151,7 @@ pub fn start( if let Err(e) = start_multi(stop_manager, device, sender, &mut up_counter) { log::warn!("stop:{}", e); } + #[cfg(any(target_os = "windows", target_os = "linux", target_os = "macos"))] worker.stop_all(); })?; } else { @@ -171,6 +173,7 @@ pub fn start( ) { log::warn!("stop:{}", e); } + #[cfg(any(target_os = "windows", target_os = "linux", target_os = "macos"))] worker.stop_all(); })?; } diff --git a/vnt/src/util/notify.rs b/vnt/src/util/notify.rs index bc9d242..5e8ec68 100644 --- a/vnt/src/util/notify.rs +++ b/vnt/src/util/notify.rs @@ -1,6 +1,7 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Arc; use std::thread::Thread; +use std::time::Duration; use std::{io, thread}; use parking_lot::Mutex; @@ -31,6 +32,9 @@ impl StopManager { pub fn wait(&self) { self.inner.wait(); } + pub fn wait_timeout(&self, dur: Duration) -> bool { + self.inner.wait_timeout(dur) + } pub fn is_stop(&self) -> bool { self.inner.state.load(Ordering::Acquire) } @@ -103,6 +107,18 @@ impl StopManagerInner { thread::park() } } + fn wait_timeout(&self, dur: Duration) -> bool { + { + let mut guard = self.park_threads.lock(); + guard.push(thread::current()); + drop(guard); + } + if self.worker_num.load(Ordering::Acquire) == 0 { + return true; + } + thread::park_timeout(dur); + self.worker_num.load(Ordering::Acquire) == 0 + } fn stop_call(&self) { if let Some(call) = self.stop_call.lock().take() { call();