starred/EasyTier
1
0
Fork 0
mirror of https://github.com/EasyTier/EasyTier.git synced 2026-10-13 00:00:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: restore thread network namespaces and stabilize tunnel tests (#2683)

Browse Source
This commit is contained in:
KKRainbow
2026-10-12 20:04:01 +08:00
committed by GitHub
parent 1649ec0013
commit 4b4f0c3cb3
5 changed files with 162 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
easytier/src/common/netns.rs
+8 -3
View File
@@ -18,7 +18,7 @@ impl NetNSGuard {
pub fn new(ns: Option<String>) -> Box<Self> {
let old_ns = if ns.is_some() {
let old_ns = if cfg!(target_os = "linux") {
Some(std::fs::File::open("/proc/self/ns/net").unwrap())
Some(std::fs::File::open("/proc/thread-self/ns/net").unwrap())
} else {
None
};
@@ -101,8 +101,13 @@ impl NetNS {
{
// TODO: do we really need this lock
// let _lock = LOCK.lock().await;
let _guard = NetNSGuard::new(self.name.clone());
f().await
// Namespace membership belongs to a thread. Restore it before yielding,
// and re-enter on every poll even if the future moves to another thread.
// Drop the future in its namespace too, including on cancellation.
let mut future = guarden::guard::ContextGuard::new(Box::pin(self.run(f)), |future| {
self.run(|| drop(future));
});
std::future::poll_fn(|cx| self.run(|| future.as_mut().poll(cx))).await
}
pub fn run<F, Ret>(&self, f: F) -> Ret
easytier/src/tests/network.rs
+55
View File
@@ -93,3 +93,58 @@ impl Drop for TestNetwork {
}
}
}
#[test]
fn nested_namespace_guard_restores_current_thread() {
let mut network = TestNetwork::new();
network.create_empty_namespace("net_a");
network.create_empty_namespace("net_b");
let net_a = NetNS::new(Some(network.namespace("net_a").into()));
let net_b = NetNS::new(Some(network.namespace("net_b").into()));
let current = || std::fs::read_link("/proc/thread-self/ns/net").unwrap();
let original = current();
net_a.run(|| {
let expected = current();
assert_ne!(expected, original);
net_b.run(|| assert_ne!(current(), expected));
assert_eq!(current(), expected);
});
assert_eq!(current(), original);
}
#[tokio::test]
async fn async_namespace_restores_between_polls() {
let mut network = TestNetwork::new();
network.create_empty_namespace("net_a");
network.create_empty_namespace("net_b");
let net_a = NetNS::new(Some(network.namespace("net_a").into()));
let net_b = NetNS::new(Some(network.namespace("net_b").into()));
let current = || std::fs::read_link("/proc/thread-self/ns/net").unwrap();
let original = current();
let expected = net_a.run(current);
let mut future = Box::pin(net_a.run_async(|| async {
assert_eq!(current(), expected);
tokio::task::yield_now().await;
assert_eq!(current(), expected);
}));
assert!(futures::poll!(future.as_mut()).is_pending());
assert_eq!(current(), original);
net_b.run(|| assert_ne!(current(), expected));
future.await;
assert_eq!(current(), original);
let dropped_in = std::cell::RefCell::new(None);
let mut cancelled = Box::pin(net_a.run_async(|| async {
let _cleanup = guarden::guard::ContextGuard::new((), |_| {
*dropped_in.borrow_mut() = Some(current());
});
std::future::pending::<()>().await;
}));
assert!(futures::poll!(cancelled.as_mut()).is_pending());
assert_eq!(current(), original);
drop(cancelled);
assert_eq!(current(), original);
assert_eq!(*dropped_in.borrow(), Some(expected));
}
easytier/src/tests/three_node.rs
+7 -4
View File
@@ -2836,13 +2836,16 @@ pub async fn port_forward_test(
let mut buf = vec![0; buf_size as usize];
rand::thread_rng().fill(&mut buf[..]);
// Allow the UDP connector's three-second handshake deadline to complete.
// Echo and shutdown retain their separate one-second deadlines.
let connect_timeout = Duration::from_secs(5);
_tunnel_pingpong_netns_with_timeout(
tcp_listener,
tcp_connector,
NetNS::new(Some(network.namespace("net_c").into())),
NetNS::new(Some(network.namespace("net_a").into())),
buf,
Duration::from_secs(1),
connect_timeout,
)
.await
.unwrap();
@@ -2859,7 +2862,7 @@ pub async fn port_forward_test(
NetNS::new(Some(network.namespace("net_d").into())),
NetNS::new(Some(network.namespace("net_a").into())),
buf,
Duration::from_secs(1),
connect_timeout,
)
.await
.unwrap();
@@ -2876,7 +2879,7 @@ pub async fn port_forward_test(
NetNS::new(Some(network.namespace("net_c").into())),
NetNS::new(Some(network.namespace("net_a").into())),
buf,
Duration::from_secs(1),
connect_timeout,
)
.await
.unwrap();
@@ -2893,7 +2896,7 @@ pub async fn port_forward_test(
NetNS::new(Some(network.namespace("net_d").into())),
NetNS::new(Some(network.namespace("net_a").into())),
buf,
Duration::from_secs(1),
connect_timeout,
)
.await
.unwrap();
easytier/src/tunnel/common.rs
+74 -40
View File
@@ -252,8 +252,10 @@ pub(crate) mod tests {
use atomic_shim::AtomicU64;
use std::{sync::Arc, time::Instant};
use anyhow::Context as _;
use futures::{Future, SinkExt, StreamExt};
use tokio_util::bytes::{BufMut, Bytes, BytesMut};
use tokio_util::task::AbortOnDropHandle;
use easytier_core::{
connectivity::protocol::raw::TunnelDialer, packet::ZCPacket, socket::SocketListener,
@@ -341,7 +343,7 @@ pub(crate) mod tests {
if !once {
while let Some(item) = recv.next().await {
let Ok(msg) = item else {
continue;
break;
};
tracing::debug!(?msg, "recv a msg, try echo back");
if send.send(msg).await.is_err() {
@@ -368,23 +370,41 @@ pub(crate) mod tests {
tracing::warn!("echo server exit...");
}
#[tokio::test]
async fn echo_server_stops_on_read_error() {
use easytier_core::tunnel::{TunnelError, wrapper::TunnelWrapper};
let mut failed = false;
let reader = futures::stream::poll_fn(move |_| {
assert!(!failed, "a terminal read error must not be polled again");
failed = true;
std::task::Poll::Ready(Some(Err::<ZCPacket, _>(TunnelError::IOError(
std::io::ErrorKind::NotConnected.into(),
))))
});
let writer = futures::sink::drain().sink_map_err(|err| -> TunnelError { match err {} });
_tunnel_echo_server(Box::new(TunnelWrapper::new(reader, writer, None)), false).await;
}
async fn _tunnel_pingpong_netns<L, C>(
mut listener: L,
connector: C,
l_netns: NetNS,
c_netns: NetNS,
buf: Vec<u8>,
) where
timeout: std::time::Duration,
) -> anyhow::Result<()>
where
L: SocketListener<Accepted = Box<dyn Tunnel>> + Sync + 'static,
C: TunnelDialer,
{
l_netns
.run_async(|| async {
listener.listen().await.unwrap();
})
.await;
tokio::time::timeout(timeout, l_netns.run_async(|| listener.listen()))
.await
.context("listener setup timed out")?
.context("listener setup failed")?;
let lis = tokio::spawn(async move {
let lis = AbortOnDropHandle::new(tokio::spawn(async move {
let ret = listener.accept().await.unwrap();
println!("accept: {:?}", ret.info());
assert_eq!(
@@ -392,9 +412,12 @@ pub(crate) mod tests {
listener.local_url()
);
_tunnel_echo_server(ret, false).await
});
}));
let tunnel = c_netns.run_async(|| connector.connect()).await.unwrap();
let tunnel = tokio::time::timeout(timeout, c_netns.run_async(|| connector.connect()))
.await
.context("connect timed out")?
.context("connect failed")?;
println!("connect: {:?}", tunnel.info());
if connector.remote_url().scheme() == "faketcp" {
@@ -409,27 +432,37 @@ pub(crate) mod tests {
let (mut recv, mut send) = tunnel.split();
send.send(ZCPacket::new_with_payload(buf.as_slice()))
.await
.unwrap();
tokio::time::timeout(
timeout,
send.send(ZCPacket::new_with_payload(buf.as_slice())),
)
.await
.context("send timed out")?
.context("send failed")?;
let ret = tokio::time::timeout(tokio::time::Duration::from_secs(1), recv.next())
.await
.unwrap()
.unwrap()
.unwrap();
.context("echo timed out")?
.context("tunnel closed before echo")?
.context("echo read failed")?;
println!("echo back: {:?}", ret);
assert_eq!(ret.payload(), Bytes::from(buf));
send.close().await.unwrap();
tokio::time::timeout(std::time::Duration::from_secs(1), send.close())
.await
.context("client close timed out")?
.context("client close failed")?;
if ["udp", "wg"].contains(&connector.remote_url().scheme()) {
lis.abort();
let _ = lis.await;
} else {
// lis should finish in 1 second
let ret = tokio::time::timeout(tokio::time::Duration::from_secs(1), lis).await;
assert!(ret.is_ok());
ret.context("echo server shutdown timed out")?
.context("echo server task failed")?;
}
Ok(())
}
pub(crate) async fn _tunnel_pingpong_netns_with_timeout<L, C>(
@@ -444,30 +477,31 @@ pub(crate) mod tests {
L: SocketListener<Accepted = Box<dyn Tunnel>> + Sync + 'static,
C: TunnelDialer,
{
let handle = tokio::spawn(async move {
_tunnel_pingpong_netns(listener, connector, l_netns, c_netns, buf).await;
});
// Each phase has its own deadline; connection setup must not consume
// the echo or shutdown budget. Dropping the caller cancels both tasks.
let handle = AbortOnDropHandle::new(tokio::spawn(async move {
_tunnel_pingpong_netns(listener, connector, l_netns, c_netns, buf, timeout).await
}));
match tokio::time::timeout(timeout, handle).await {
Ok(join_res) => match join_res {
Ok(_) => Ok(()),
Err(join_err) => {
if join_err.is_panic() {
let payload = join_err.into_panic();
let msg = match payload.downcast::<String>() {
Ok(s) => *s,
Err(payload) => match payload.downcast::<&str>() {
Ok(s) => (*s).to_string(),
Err(_) => "non-string panic payload".to_string(),
},
};
Err(anyhow::anyhow!("task panicked: {}", msg))
} else {
Err(anyhow::anyhow!("task cancelled"))
}
match handle.await {
Ok(result) => {
result.inspect_err(|error| eprintln!("tunnel pingpong failed: {error:#}"))
}
Err(join_err) => {
if join_err.is_panic() {
let payload = join_err.into_panic();
let msg = match payload.downcast::<String>() {
Ok(s) => *s,
Err(payload) => match payload.downcast::<&str>() {
Ok(s) => (*s).to_string(),
Err(_) => "non-string panic payload".to_string(),
},
};
Err(anyhow::anyhow!("task panicked: {}", msg))
} else {
Err(anyhow::anyhow!("task cancelled"))
}
},
Err(elapsed) => Err(elapsed.into()),
}
}
}
easytier/src/tunnel/quic.rs
+18 -6
View File
@@ -673,7 +673,7 @@ mod tests {
socket::SocketListener,
socket::udp::{
UdpBindOptions, UdpSessionAcceptKind, UdpSessionListenRequest, UdpSessionProtocol,
VirtualUdpSocket,
UdpSessionSocket, VirtualUdpSocket,
},
};
use futures::{SinkExt, StreamExt};
@@ -692,7 +692,9 @@ mod tests {
let mut listener = new_runtime_udp_session_listener(
format!("quic://{bind_addr}").parse().unwrap(),
UdpSessionListenRequest::new(
UdpBindOptions::port_bound_listener(bind_addr).with_only_v6(false),
// Port-bound listeners enable address reuse on Unix. This test
// needs an exclusive ephemeral port across parallel processes.
UdpBindOptions::direct_connect().with_local_addr(Some(bind_addr)),
),
UdpSessionAcceptKind::Classified(UdpSessionProtocol::Quic),
NetNS::new(None),
@@ -716,7 +718,9 @@ mod tests {
Endpoint::new_with_abstract_socket(endpoint_config(), None, socket, runtime)
.unwrap();
endpoint.set_default_client_config(client_config());
let client_addr = endpoint.local_addr().unwrap();
let (first_accepted_tx, first_accepted_rx) = tokio::sync::oneshot::channel();
let server_task = tokio::spawn(async move {
let session = listener.accept().await.unwrap();
let admission = ServerProtocolAdmissionController::new(1, 2)
@@ -724,12 +728,17 @@ mod tests {
.unwrap();
let mut accepted = QuicAcceptedSession::new(session, local_url, admission).unwrap();
let first = accepted.accept().await.unwrap();
first_accepted_tx.send(()).unwrap();
let second = accepted.accept().await.unwrap();
let extra_session =
tokio::time::timeout(Duration::from_millis(50), listener.accept()).await;
assert!(
tokio::time::timeout(Duration::from_millis(50), listener.accept())
.await
.is_err(),
"both QUIC connections must use the first accepted UDP session"
extra_session.is_err(),
"both QUIC connections must use the first accepted UDP session; \
server={remote_addr}, client={client_addr}, extra_peer={:?}",
extra_session
.as_ref()
.map(|result| result.as_ref().map(|session| session.peer_addr()))
);
(first, second)
});
@@ -745,6 +754,9 @@ mod tests {
.send(ZCPacket::new_with_payload(b"first QUIC connection"))
.await
.unwrap();
// Keep both connections alive, but establish their server-side identity
// before starting the second handshake, which may complete out of order.
first_accepted_rx.await.unwrap();
let second_connection = endpoint
.connect(remote_addr, "localhost")
.unwrap()