Skip to content

fix: make StartConditions sync, drop block_in_place under ctrl_lock #2779

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
YuanYuYuan wants to merge 1 commit into
base: main
Choose a base branch
from
+42 −41
Open

fix: make StartConditions sync, drop block_in_place under ctrl_lock #2779

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 12 additions & 10 deletions zenoh/src/net/protocol/gossip.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -394,26 +394,28 @@ impl Gossip {
.await
.is_none()
{
runtime.start_conditions().add_peer_connector_zid(zid).await;
runtime.start_conditions().add_peer_connector_zid(zid);
if runtime.connect_peer(&zid, &locators).await
&& ((!wait_declares) || whatami != WhatAmI::Peer)
{
runtime
.start_conditions()
.terminate_peer_connector_zid(zid)
.await;
runtime.start_conditions().terminate_peer_connector_zid(zid);
}
}
});
}
}
}
// NOTE: link_states() is called from OAM handling with the router
// ctrl_lock and the tables write lock held. It must never block on
// work scheduled on another runtime (e.g. via block_in_place): the
// Net runtime concurrently runs autoconnect connectors that block
// synchronously on the ctrl_lock in Router::new_transport_unicast,
// which deadlocks the whole session. terminate_peer_connector_zid
// is a sync call precisely so that it is safe to invoke here.
if (!self.wait_declares) || src_whatami != WhatAmI::Peer {
zenoh_runtime::ZRuntime::Net.block_in_place(
strong_runtime
.start_conditions()
.terminate_peer_connector_zid(src),
);
strong_runtime
.start_conditions()
.terminate_peer_connector_zid(src);
}
}

Expand Down
20 changes: 10 additions & 10 deletions zenoh/src/net/routing/hat/peer/interests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -280,17 +280,17 @@ impl HatInterestTrait for Hat {
return Noop;
}

zenoh_runtime::ZRuntime::Net.block_in_place(async move {
if let Some(runtime) = &ctx.tables.runtime {
if let Some(runtime) = runtime.upgrade() {
tracing::debug!("Terminating peer connector");
runtime
.start_conditions()
.terminate_peer_connector_zid(ctx.src_face.zid)
.await
}
// NOTE: called with the router ctrl_lock and the tables write lock
// held -- must not block on another runtime (see gossip.rs
// link_states); terminate_peer_connector_zid is sync on purpose.
if let Some(runtime) = &ctx.tables.runtime {
if let Some(runtime) = runtime.upgrade() {
tracing::debug!("Terminating peer connector");
runtime
.start_conditions()
.terminate_peer_connector_zid(ctx.src_face.zid)
}
});
}

Noop
} else {
Expand Down
41 changes: 20 additions & 21 deletions zenoh/src/net/runtime/orchestrator.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ use futures::{prelude::*, stream::FuturesUnordered};
use socket2::{Domain, Socket, Type};
use tokio::{
net::UdpSocket,
sync::{futures::Notified, Mutex, Notify},
sync::{futures::Notified, Notify},
};
use tokio_util::sync::CancellationToken;
use zenoh_buffers::{
Expand Down Expand Up @@ -91,22 +91,27 @@ pub(crate) struct PeerConnector {
#[derive(Default, Debug)]
pub(crate) struct StartConditions {
notify: Notify,
peer_connectors: Mutex<Vec<PeerConnector>>,
// NOTE: this is a sync mutex on purpose: its critical sections never await
// and it is locked from RX/routing contexts that hold the router
// ctrl_lock. An async (fair) mutex here can hand ownership to a connector
// task parked on a starved runtime and deadlock the whole session (see
// gossip.rs link_states / hat/peer/interests.rs route_declare_final).
peer_connectors: std::sync::Mutex<Vec<PeerConnector>>,
}

impl StartConditions {
pub(crate) fn notified(&self) -> Notified<'_> {
self.notify.notified()
}

pub(crate) async fn add_peer_connector(&self) -> usize {
let mut peer_connectors = self.peer_connectors.lock().await;
pub(crate) fn add_peer_connector(&self) -> usize {
let mut peer_connectors = zlock!(self.peer_connectors);
peer_connectors.push(PeerConnector::default());
peer_connectors.len() - 1
}

pub(crate) async fn add_peer_connector_zid(&self, zid: ZenohIdProto) {
let mut peer_connectors = self.peer_connectors.lock().await;
pub(crate) fn add_peer_connector_zid(&self, zid: ZenohIdProto) {
let mut peer_connectors = zlock!(self.peer_connectors);
if !peer_connectors.iter().any(|pc| pc.zid == Some(zid)) {
peer_connectors.push(PeerConnector {
zid: Some(zid),
Expand All @@ -115,15 +120,15 @@ impl StartConditions {
}
}

pub(crate) async fn set_peer_connector_zid(&self, idx: usize, zid: ZenohIdProto) {
let mut peer_connectors = self.peer_connectors.lock().await;
pub(crate) fn set_peer_connector_zid(&self, idx: usize, zid: ZenohIdProto) {
let mut peer_connectors = zlock!(self.peer_connectors);
if let Some(peer_connector) = peer_connectors.get_mut(idx) {
peer_connector.zid = Some(zid);
}
}

pub(crate) async fn terminate_peer_connector(&self, idx: usize) {
let mut peer_connectors = self.peer_connectors.lock().await;
pub(crate) fn terminate_peer_connector(&self, idx: usize) {
let mut peer_connectors = zlock!(self.peer_connectors);
if let Some(peer_connector) = peer_connectors.get_mut(idx) {
peer_connector.terminated = true;
}
Expand All @@ -132,8 +137,8 @@ impl StartConditions {
}
}

pub(crate) async fn terminate_peer_connector_zid(&self, zid: ZenohIdProto) {
let mut peer_connectors = self.peer_connectors.lock().await;
pub(crate) fn terminate_peer_connector_zid(&self, zid: ZenohIdProto) {
let mut peer_connectors = zlock!(self.peer_connectors);
if let Some(peer_connector) = peer_connectors.iter_mut().find(|pc| pc.zid == Some(zid)) {
peer_connector.terminated = true;
} else {
Expand Down Expand Up @@ -867,24 +872,18 @@ impl Runtime {
.await?
{
let this = self.clone();
let idx = self.state.start_conditions.add_peer_connector().await;
let idx = self.state.start_conditions.add_peer_connector();
let config_guard = this.config().lock();
let config = &config_guard;
let gossip = unwrap_or_default!(config.scouting().gossip().enabled());
let wait_declares = unwrap_or_default!(config.open().return_conditions().declares());
drop(config_guard);
self.spawn(async move {
if let Ok(zid) = this.peer_connector_retry(peer).await {
this.state
.start_conditions
.set_peer_connector_zid(idx, zid)
.await;
this.state.start_conditions.set_peer_connector_zid(idx, zid);
}
if !gossip && (!wait_declares || this.whatami() != WhatAmI::Peer) {
this.state
.start_conditions
.terminate_peer_connector(idx)
.await;
this.state.start_conditions.terminate_peer_connector(idx);
}
});
Ok(())
Expand Down
Loading