summaryrefslogtreecommitdiffhomepage
diff options
context:
space:
mode:
Diffstat
-rw-r--r--talpid-core/src/routing/unix.rs48
-rw-r--r--talpid-core/src/routing/windows.rs14
-rw-r--r--talpid-core/src/tunnel/openvpn/mod.rs21
-rw-r--r--talpid-core/src/tunnel/wireguard/mod.rs20
-rw-r--r--talpid-core/src/tunnel_state_machine/connected_state.rs5
-rw-r--r--talpid-core/src/tunnel_state_machine/connecting_state.rs5
-rw-r--r--talpid-core/src/tunnel_state_machine/mod.rs7
7 files changed, 58 insertions, 62 deletions
diff --git a/talpid-core/src/routing/unix.rs b/talpid-core/src/routing/unix.rs
index c1d2da522b..4b7cb8eed4 100644
--- a/talpid-core/src/routing/unix.rs
+++ b/talpid-core/src/routing/unix.rs
@@ -46,26 +46,25 @@ pub enum Error {
/// Handle to a route manager.
#[derive(Clone)]
pub struct RouteManagerHandle {
- runtime: tokio::runtime::Handle,
tx: UnboundedSender<RouteManagerCommand>,
}
impl RouteManagerHandle {
/// Applies the given routes while the route manager is running.
- pub fn add_routes(&self, routes: HashSet<RequiredRoute>) -> Result<(), Error> {
+ pub async fn add_routes(&self, routes: HashSet<RequiredRoute>) -> Result<(), Error> {
let (response_tx, response_rx) = oneshot::channel();
self.tx
.unbounded_send(RouteManagerCommand::AddRoutes(routes, response_tx))
.map_err(|_| Error::RouteManagerDown)?;
- self.runtime
- .block_on(response_rx)
+ response_rx
+ .await
.map_err(|_| Error::ManagerChannelDown)?
.map_err(Error::PlatformError)
}
/// Ensure that packets are routed using the correct tables.
#[cfg(target_os = "linux")]
- pub fn create_routing_rules(&self, enable_ipv6: bool) -> Result<(), Error> {
+ pub async fn create_routing_rules(&self, enable_ipv6: bool) -> Result<(), Error> {
let (response_tx, response_rx) = oneshot::channel();
self.tx
.unbounded_send(RouteManagerCommand::CreateRoutingRules(
@@ -73,21 +72,21 @@ impl RouteManagerHandle {
response_tx,
))
.map_err(|_| Error::RouteManagerDown)?;
- self.runtime
- .block_on(response_rx)
+ response_rx
+ .await
.map_err(|_| Error::ManagerChannelDown)?
.map_err(Error::PlatformError)
}
/// Remove any routing rules created by [`create_routing_rules`].
#[cfg(target_os = "linux")]
- pub fn clear_routing_rules(&self) -> Result<(), Error> {
+ pub async fn clear_routing_rules(&self) -> Result<(), Error> {
let (response_tx, response_rx) = oneshot::channel();
self.tx
.unbounded_send(RouteManagerCommand::ClearRoutingRules(response_tx))
.map_err(|_| Error::RouteManagerDown)?;
- self.runtime
- .block_on(response_rx)
+ response_rx
+ .await
.map_err(|_| Error::ManagerChannelDown)?
.map_err(Error::PlatformError)
}
@@ -120,12 +119,12 @@ impl RouteManager {
/// Constructs a RouteManager and applies the required routes.
/// Takes a set of network destinations and network nodes as an argument, and applies said
/// routes.
- pub fn new(
+ pub async fn new(
runtime: tokio::runtime::Handle,
required_routes: HashSet<RequiredRoute>,
) -> Result<Self, Error> {
let (manage_tx, manage_rx) = mpsc::unbounded();
- let manager = runtime.block_on(imp::RouteManagerImpl::new(required_routes))?;
+ let manager = imp::RouteManagerImpl::new(required_routes).await?;
runtime.spawn(manager.run(manage_rx));
Ok(Self {
@@ -135,7 +134,7 @@ impl RouteManager {
}
/// Stops RouteManager and removes all of the applied routes.
- pub fn stop(&mut self) {
+ pub async fn stop(&mut self) {
if let Some(tx) = self.manage_tx.take() {
let (wait_tx, wait_rx) = oneshot::channel();
@@ -147,14 +146,14 @@ impl RouteManager {
return;
}
- if self.runtime.block_on(wait_rx).is_err() {
+ if wait_rx.await.is_err() {
log::error!("{}", Error::ManagerChannelDown);
}
}
}
/// Applies the given routes until [`RouteManager::stop`] is called.
- pub fn add_routes(&mut self, routes: HashSet<RequiredRoute>) -> Result<(), Error> {
+ pub async fn add_routes(&mut self, routes: HashSet<RequiredRoute>) -> Result<(), Error> {
if let Some(tx) = &self.manage_tx {
let (result_tx, result_rx) = oneshot::channel();
if tx
@@ -164,8 +163,8 @@ impl RouteManager {
return Err(Error::RouteManagerDown);
}
- self.runtime
- .block_on(result_rx)
+ result_rx
+ .await
.map_err(|_| Error::ManagerChannelDown)?
.map_err(Error::PlatformError)
} else {
@@ -188,23 +187,20 @@ impl RouteManager {
/// Ensure that packets are routed using the correct tables.
#[cfg(target_os = "linux")]
- pub fn create_routing_rules(&mut self, enable_ipv6: bool) -> Result<(), Error> {
- self.handle()?.create_routing_rules(enable_ipv6)
+ pub async fn create_routing_rules(&mut self, enable_ipv6: bool) -> Result<(), Error> {
+ self.handle()?.create_routing_rules(enable_ipv6).await
}
/// Remove any routing rules created by [`create_routing_rules`].
#[cfg(target_os = "linux")]
- pub fn clear_routing_rules(&mut self) -> Result<(), Error> {
- self.handle()?.clear_routing_rules()
+ pub async fn clear_routing_rules(&mut self) -> Result<(), Error> {
+ self.handle()?.clear_routing_rules().await
}
/// Retrieve a sender directly to the command channel.
pub fn handle(&self) -> Result<RouteManagerHandle, Error> {
if let Some(tx) = &self.manage_tx {
- Ok(RouteManagerHandle {
- runtime: self.runtime.clone(),
- tx: tx.clone(),
- })
+ Ok(RouteManagerHandle { tx: tx.clone() })
} else {
Err(Error::RouteManagerDown)
}
@@ -219,6 +215,6 @@ impl RouteManager {
impl Drop for RouteManager {
fn drop(&mut self) {
- self.stop();
+ self.runtime.clone().block_on(self.stop());
}
}
diff --git a/talpid-core/src/routing/windows.rs b/talpid-core/src/routing/windows.rs
index ca5fbb8ea2..93574884f6 100644
--- a/talpid-core/src/routing/windows.rs
+++ b/talpid-core/src/routing/windows.rs
@@ -41,20 +41,17 @@ pub struct RouteManager {
/// Handle to a route manager.
#[derive(Clone)]
pub struct RouteManagerHandle {
- runtime: tokio::runtime::Handle,
tx: UnboundedSender<RouteManagerCommand>,
}
impl RouteManagerHandle {
/// Applies the given routes while the route manager is running.
- pub fn add_routes(&self, routes: HashSet<RequiredRoute>) -> Result<()> {
+ pub async fn add_routes(&self, routes: HashSet<RequiredRoute>) -> Result<()> {
let (response_tx, response_rx) = oneshot::channel();
self.tx
.unbounded_send(RouteManagerCommand::AddRoutes(routes, response_tx))
.map_err(|_| Error::RouteManagerDown)?;
- self.runtime
- .block_on(response_rx)
- .map_err(|_| Error::ManagerChannelDown)?
+ response_rx.await.map_err(|_| Error::ManagerChannelDown)?
}
}
@@ -67,7 +64,7 @@ pub enum RouteManagerCommand {
impl RouteManager {
/// Creates a new route manager that will apply the provided routes and ensure they exist until
/// it's stopped.
- pub fn new(
+ pub async fn new(
runtime: tokio::runtime::Handle,
required_routes: HashSet<RequiredRoute>,
) -> Result<Self> {
@@ -89,10 +86,7 @@ impl RouteManager {
/// Retrieve a sender directly to the command channel.
pub fn handle(&self) -> Result<RouteManagerHandle> {
if let Some(tx) = &self.manage_tx {
- Ok(RouteManagerHandle {
- runtime: self.runtime.clone(),
- tx: tx.clone(),
- })
+ Ok(RouteManagerHandle { tx: tx.clone() })
} else {
Err(Error::RouteManagerDown)
}
diff --git a/talpid-core/src/tunnel/openvpn/mod.rs b/talpid-core/src/tunnel/openvpn/mod.rs
index e3709b0183..87dbc55101 100644
--- a/talpid-core/src/tunnel/openvpn/mod.rs
+++ b/talpid-core/src/tunnel/openvpn/mod.rs
@@ -1028,19 +1028,14 @@ mod event_server {
.filter(|route| route.prefix.is_ipv4() || ipv6_enabled)
.collect();
- tokio::task::spawn_blocking(move || {
- if let Err(error) = route_handle.add_routes(routes) {
- log::error!("{}", error.display_chain());
- return Err(tonic::Status::failed_precondition("Failed to add routes"));
- }
- if let Err(error) = route_handle.create_routing_rules(ipv6_enabled) {
- log::error!("{}", error.display_chain());
- return Err(tonic::Status::failed_precondition("Failed to add routes"));
- }
- Ok(())
- })
- .await
- .map_err(|_| tonic::Status::internal("task failed to complete"))??;
+ if let Err(error) = route_handle.add_routes(routes).await {
+ log::error!("{}", error.display_chain());
+ return Err(tonic::Status::failed_precondition("Failed to add routes"));
+ }
+ if let Err(error) = route_handle.create_routing_rules(ipv6_enabled).await {
+ log::error!("{}", error.display_chain());
+ return Err(tonic::Status::failed_precondition("Failed to add routes"));
+ }
}
let tunnel_alias = env
diff --git a/talpid-core/src/tunnel/wireguard/mod.rs b/talpid-core/src/tunnel/wireguard/mod.rs
index 4d7fa478b1..899589dbc4 100644
--- a/talpid-core/src/tunnel/wireguard/mod.rs
+++ b/talpid-core/src/tunnel/wireguard/mod.rs
@@ -253,20 +253,24 @@ impl WireguardMonitor {
}
}
- let setup_iface_routes = move || -> Result<()> {
+ let setup_iface_routes = || -> Result<()> {
#[cfg(target_os = "windows")]
if !crate::winnet::add_device_ip_addresses(&iface_name, &config.tunnel.addresses) {
return Err(Error::SetIpAddressesError);
}
- #[cfg(target_os = "linux")]
- route_handle
- .create_routing_rules(config.enable_ipv6)
- .map_err(Error::SetupRoutingError)?;
+ runtime.block_on(async move {
+ #[cfg(target_os = "linux")]
+ route_handle
+ .create_routing_rules(config.enable_ipv6)
+ .await
+ .map_err(Error::SetupRoutingError)?;
- route_handle
- .add_routes(Self::get_routes(&iface_name, &config))
- .map_err(Error::SetupRoutingError)
+ route_handle
+ .add_routes(Self::get_routes(&iface_name, &config))
+ .await
+ .map_err(Error::SetupRoutingError)
+ })
};
if let Err(error) = setup_iface_routes() {
diff --git a/talpid-core/src/tunnel_state_machine/connected_state.rs b/talpid-core/src/tunnel_state_machine/connected_state.rs
index ea8ce8e992..72af2b6f38 100644
--- a/talpid-core/src/tunnel_state_machine/connected_state.rs
+++ b/talpid-core/src/tunnel_state_machine/connected_state.rs
@@ -139,7 +139,10 @@ impl ConnectedState {
log::error!("{}", error.display_chain_with_msg("Failed to clear routes"));
}
#[cfg(target_os = "linux")]
- if let Err(error) = shared_values.route_manager.clear_routing_rules() {
+ if let Err(error) = shared_values
+ .runtime
+ .block_on(shared_values.route_manager.clear_routing_rules())
+ {
log::error!(
"{}",
error.display_chain_with_msg("Failed to clear routing rules")
diff --git a/talpid-core/src/tunnel_state_machine/connecting_state.rs b/talpid-core/src/tunnel_state_machine/connecting_state.rs
index 8486ae0a3e..b0c87acdb4 100644
--- a/talpid-core/src/tunnel_state_machine/connecting_state.rs
+++ b/talpid-core/src/tunnel_state_machine/connecting_state.rs
@@ -204,7 +204,10 @@ impl ConnectingState {
log::error!("{}", error.display_chain_with_msg("Failed to clear routes"));
}
#[cfg(target_os = "linux")]
- if let Err(error) = shared_values.route_manager.clear_routing_rules() {
+ if let Err(error) = shared_values
+ .runtime
+ .block_on(shared_values.route_manager.clear_routing_rules())
+ {
log::error!(
"{}",
error.display_chain_with_msg("Failed to clear routing rules")
diff --git a/talpid-core/src/tunnel_state_machine/mod.rs b/talpid-core/src/tunnel_state_machine/mod.rs
index 1bcc6961d9..819d167733 100644
--- a/talpid-core/src/tunnel_state_machine/mod.rs
+++ b/talpid-core/src/tunnel_state_machine/mod.rs
@@ -113,7 +113,7 @@ pub async fn spawn(
let (startup_result_tx, startup_result_rx) = sync_mpsc::channel();
std::thread::spawn(move || {
- let state_machine = TunnelStateMachine::new(
+ let state_machine = runtime.block_on(TunnelStateMachine::new(
runtime.clone(),
allow_lan,
block_when_disconnected,
@@ -127,7 +127,7 @@ pub async fn spawn(
cache_dir,
command_rx,
reset_firewall,
- );
+ ));
let state_machine = match state_machine {
Ok(state_machine) => {
startup_result_tx.send(Ok(())).unwrap();
@@ -199,7 +199,7 @@ struct TunnelStateMachine {
}
impl TunnelStateMachine {
- fn new(
+ async fn new(
runtime: tokio::runtime::Handle,
allow_lan: bool,
block_when_disconnected: bool,
@@ -224,6 +224,7 @@ impl TunnelStateMachine {
let dns_monitor =
DnsMonitor::new(runtime.clone(), cache_dir).map_err(Error::InitDnsMonitorError)?;
let route_manager = RouteManager::new(runtime.clone(), HashSet::new())
+ .await
.map_err(Error::InitRouteManagerError)?;
let mut shared_values = SharedTunnelStateValues {
runtime,
Copyright © 2025 - 2026 Wayne Cole. WTFPL. Attribution appreciated. No warranty. Not liable for bodily damages.