Skip to content

Commit 93f93d2

Browse files
OlivierHecartOlivierHecart
authored
Release tables locks before propagating subscribers and queryables declarations to void dead locks (#1150)
* Send simple sub and qabl declarations using a given function * Send simple sub and qabl declarations after releasing tables lock * Send simple sub and qabl declarations after releasing tables lock (missing places)
1 parent 7adad94 commit 93f93d2

File tree

20 files changed

+1194
-647
lines changed

20 files changed

+1194
-647
lines changed

‎zenoh/src/net/primitives/demux.rs

Lines changed: 22 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -72,9 +72,21 @@ impl TransportPeerEventHandler for DeMux {
7272
NetworkBody::ResponseFinal(m) => self.face.send_response_final(m),
7373
NetworkBody::OAM(m) => {
7474
if let Some(transport) = self.transport.as_ref() {
75+
let mut declares = vec![];
7576
let ctrl_lock = zlock!(self.face.tables.ctrl_lock);
7677
let mut tables = zwrite!(self.face.tables.tables);
77-
ctrl_lock.handle_oam(&mut tables, &self.face.tables, m, transport)?
78+
ctrl_lock.handle_oam(
79+
&mut tables,
80+
&self.face.tables,
81+
m,
82+
transport,
83+
&mut |p, m| declares.push((p.clone(), m)),
84+
)?;
85+
drop(tables);
86+
drop(ctrl_lock);
87+
for (p, m) in declares {
88+
p.send_declare(m);
89+
}
7890
}
7991
}
8092
}
@@ -89,9 +101,17 @@ impl TransportPeerEventHandler for DeMux {
89101
fn closing(&self) {
90102
self.face.send_close();
91103
if let Some(transport) = self.transport.as_ref() {
104+
let mut declares = vec![];
92105
let ctrl_lock = zlock!(self.face.tables.ctrl_lock);
93106
let mut tables = zwrite!(self.face.tables.tables);
94-
let _ = ctrl_lock.closing(&mut tables, &self.face.tables, transport);
107+
let _ = ctrl_lock.closing(&mut tables, &self.face.tables, transport, &mut |p, m| {
108+
declares.push((p.clone(), m))
109+
});
110+
drop(tables);
111+
drop(ctrl_lock);
112+
for (p, m) in declares {
113+
p.send_declare(m);
114+
}
95115
}
96116
}
97117

‎zenoh/src/net/routing/dispatcher/face.rs

Lines changed: 24 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -195,50 +195,73 @@ impl Primitives for Face {
195195
unregister_expr(&self.tables, &mut self.state.clone(), m.id);
196196
}
197197
zenoh_protocol::network::DeclareBody::DeclareSubscriber(m) => {
198+
let mut declares = vec![];
198199
declare_subscription(
199200
ctrl_lock.as_ref(),
200201
&self.tables,
201202
&mut self.state.clone(),
202203
&m.wire_expr,
203204
&m.ext_info,
204205
msg.ext_nodeid.node_id,
206+
&mut |p, m| declares.push((p.clone(), m)),
205207
);
208+
drop(ctrl_lock);
209+
for (p, m) in declares {
210+
p.send_declare(m);
211+
}
206212
}
207213
zenoh_protocol::network::DeclareBody::UndeclareSubscriber(m) => {
214+
let mut declares = vec![];
208215
undeclare_subscription(
209216
ctrl_lock.as_ref(),
210217
&self.tables,
211218
&mut self.state.clone(),
212219
&m.ext_wire_expr.wire_expr,
213220
msg.ext_nodeid.node_id,
221+
&mut |p, m| declares.push((p.clone(), m)),
214222
);
223+
drop(ctrl_lock);
224+
for (p, m) in declares {
225+
p.send_declare(m);
226+
}
215227
}
216228
zenoh_protocol::network::DeclareBody::DeclareQueryable(m) => {
229+
let mut declares = vec![];
217230
declare_queryable(
218231
ctrl_lock.as_ref(),
219232
&self.tables,
220233
&mut self.state.clone(),
221234
&m.wire_expr,
222235
&m.ext_info,
223236
msg.ext_nodeid.node_id,
237+
&mut |p, m| declares.push((p.clone(), m)),
224238
);
239+
drop(ctrl_lock);
240+
for (p, m) in declares {
241+
p.send_declare(m);
242+
}
225243
}
226244
zenoh_protocol::network::DeclareBody::UndeclareQueryable(m) => {
245+
let mut declares = vec![];
227246
undeclare_queryable(
228247
ctrl_lock.as_ref(),
229248
&self.tables,
230249
&mut self.state.clone(),
231250
&m.ext_wire_expr.wire_expr,
232251
msg.ext_nodeid.node_id,
252+
&mut |p, m| declares.push((p.clone(), m)),
233253
);
254+
drop(ctrl_lock);
255+
for (p, m) in declares {
256+
p.send_declare(m);
257+
}
234258
}
235259
zenoh_protocol::network::DeclareBody::DeclareToken(_m) => todo!(),
236260
zenoh_protocol::network::DeclareBody::UndeclareToken(_m) => todo!(),
237261
zenoh_protocol::network::DeclareBody::DeclareInterest(_m) => todo!(),
238262
zenoh_protocol::network::DeclareBody::FinalInterest(_m) => todo!(),
239263
zenoh_protocol::network::DeclareBody::UndeclareInterest(_m) => todo!(),
240264
}
241-
drop(ctrl_lock);
242265
}
243266

244267
#[inline]

‎zenoh/src/net/routing/dispatcher/pubsub.rs

Lines changed: 18 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@
1414
use super::face::FaceState;
1515
use super::resource::{DataRoutes, Direction, PullCaches, Resource};
1616
use super::tables::{NodeId, Route, RoutingExpr, Tables, TablesLock};
17-
use crate::net::routing::hat::HatTrait;
17+
use crate::net::routing::hat::{HatTrait, SendDeclare};
1818
use std::borrow::Cow;
1919
use std::collections::HashMap;
2020
use std::sync::Arc;
@@ -37,6 +37,7 @@ pub(crate) fn declare_subscription(
3737
expr: &WireExpr,
3838
sub_info: &SubscriberInfo,
3939
node_id: NodeId,
40+
send_declare: &mut SendDeclare,
4041
) {
4142
tracing::debug!("Declare subscription {}", face);
4243
let rtables = zread!(tables.tables);
@@ -66,7 +67,14 @@ pub(crate) fn declare_subscription(
6667
(res, wtables)
6768
};
6869

69-
hat_code.declare_subscription(&mut wtables, face, &mut res, sub_info, node_id);
70+
hat_code.declare_subscription(
71+
&mut wtables,
72+
face,
73+
&mut res,
74+
sub_info,
75+
node_id,
76+
send_declare,
77+
);
7078

7179
disable_matches_data_routes(&mut wtables, &mut res);
7280
drop(wtables);
@@ -96,6 +104,7 @@ pub(crate) fn undeclare_subscription(
96104
face: &mut Arc<FaceState>,
97105
expr: &WireExpr,
98106
node_id: NodeId,
107+
send_declare: &mut SendDeclare,
99108
) {
100109
tracing::debug!("Undeclare subscription {}", face);
101110
let rtables = zread!(tables.tables);
@@ -105,7 +114,13 @@ pub(crate) fn undeclare_subscription(
105114
drop(rtables);
106115
let mut wtables = zwrite!(tables.tables);
107116

108-
hat_code.undeclare_subscription(&mut wtables, face, &mut res, node_id);
117+
hat_code.undeclare_subscription(
118+
&mut wtables,
119+
face,
120+
&mut res,
121+
node_id,
122+
send_declare,
123+
);
109124

110125
disable_matches_data_routes(&mut wtables, &mut res);
111126
drop(wtables);

‎zenoh/src/net/routing/dispatcher/queries.rs

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ use super::face::FaceState;
1515
use super::resource::{QueryRoute, QueryRoutes, QueryTargetQablSet, Resource};
1616
use super::tables::NodeId;
1717
use super::tables::{RoutingExpr, Tables, TablesLock};
18-
use crate::net::routing::hat::HatTrait;
18+
use crate::net::routing::hat::{HatTrait, SendDeclare};
1919
use crate::net::routing::RoutingContext;
2020
use async_trait::async_trait;
2121
use std::collections::HashMap;
@@ -56,6 +56,7 @@ pub(crate) fn declare_queryable(
5656
expr: &WireExpr,
5757
qabl_info: &QueryableInfo,
5858
node_id: NodeId,
59+
send_declare: &mut SendDeclare,
5960
) {
6061
tracing::debug!("Register queryable {}", face);
6162
let rtables = zread!(tables.tables);
@@ -85,7 +86,14 @@ pub(crate) fn declare_queryable(
8586
(res, wtables)
8687
};
8788

88-
hat_code.declare_queryable(&mut wtables, face, &mut res, qabl_info, node_id);
89+
hat_code.declare_queryable(
90+
&mut wtables,
91+
face,
92+
&mut res,
93+
qabl_info,
94+
node_id,
95+
send_declare,
96+
);
8997

9098
disable_matches_query_routes(&mut wtables, &mut res);
9199
drop(wtables);
@@ -112,6 +120,7 @@ pub(crate) fn undeclare_queryable(
112120
face: &mut Arc<FaceState>,
113121
expr: &WireExpr,
114122
node_id: NodeId,
123+
send_declare: &mut SendDeclare,
115124
) {
116125
let rtables = zread!(tables.tables);
117126
match rtables.get_mapping(face, &expr.scope, expr.mapping) {
@@ -120,7 +129,7 @@ pub(crate) fn undeclare_queryable(
120129
drop(rtables);
121130
let mut wtables = zwrite!(tables.tables);
122131

123-
hat_code.undeclare_queryable(&mut wtables, face, &mut res, node_id);
132+
hat_code.undeclare_queryable(&mut wtables, face, &mut res, node_id, send_declare);
124133

125134
disable_matches_query_routes(&mut wtables, &mut res);
126135
drop(wtables);

‎zenoh/src/net/routing/dispatcher/tables.rs

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -174,7 +174,13 @@ pub fn close_face(tables: &TablesLock, face: &Weak<FaceState>) {
174174
tracing::debug!("Close {}", face);
175175
face.task_controller.terminate_all(Duration::from_secs(10));
176176
finalize_pending_queries(tables, &mut face);
177-
zlock!(tables.ctrl_lock).close_face(tables, &mut face);
177+
let mut declares = vec![];
178+
let ctrl_lock = zlock!(tables.ctrl_lock);
179+
ctrl_lock.close_face(tables, &mut face, &mut |p, m| declares.push((p.clone(), m)));
180+
drop(ctrl_lock);
181+
for (p, m) in declares {
182+
p.send_declare(m);
183+
}
178184
}
179185
None => tracing::error!("Face already closed!"),
180186
}

‎zenoh/src/net/routing/hat/client/mod.rs

Lines changed: 17 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@ use super::{
3636
face::FaceState,
3737
tables::{NodeId, Resource, RoutingExpr, Tables, TablesLock},
3838
},
39-
HatBaseTrait, HatTrait,
39+
HatBaseTrait, HatTrait, SendDeclare,
4040
};
4141
use std::{
4242
any::Any,
@@ -97,9 +97,10 @@ impl HatBaseTrait for HatCode {
9797
tables: &mut Tables,
9898
_tables_ref: &Arc<TablesLock>,
9999
face: &mut Face,
100+
send_declare: &mut SendDeclare,
100101
) -> ZResult<()> {
101-
pubsub_new_face(tables, &mut face.state);
102-
queries_new_face(tables, &mut face.state);
102+
pubsub_new_face(tables, &mut face.state, send_declare);
103+
queries_new_face(tables, &mut face.state, send_declare);
103104
Ok(())
104105
}
105106

@@ -109,13 +110,19 @@ impl HatBaseTrait for HatCode {
109110
_tables_ref: &Arc<TablesLock>,
110111
face: &mut Face,
111112
_transport: &TransportUnicast,
113+
send_declare: &mut SendDeclare,
112114
) -> ZResult<()> {
113-
pubsub_new_face(tables, &mut face.state);
114-
queries_new_face(tables, &mut face.state);
115+
pubsub_new_face(tables, &mut face.state, send_declare);
116+
queries_new_face(tables, &mut face.state, send_declare);
115117
Ok(())
116118
}
117119

118-
fn close_face(&self, tables: &TablesLock, face: &mut Arc<FaceState>) {
120+
fn close_face(
121+
&self,
122+
tables: &TablesLock,
123+
face: &mut Arc<FaceState>,
124+
send_declare: &mut SendDeclare,
125+
) {
119126
let mut wtables = zwrite!(tables.tables);
120127
let mut face_clone = face.clone();
121128
let face = get_mut_unchecked(face);
@@ -139,7 +146,7 @@ impl HatBaseTrait for HatCode {
139146
.drain()
140147
{
141148
get_mut_unchecked(&mut res).session_ctxs.remove(&face.id);
142-
undeclare_client_subscription(&mut wtables, &mut face_clone, &mut res);
149+
undeclare_client_subscription(&mut wtables, &mut face_clone, &mut res, send_declare);
143150

144151
if res.context.is_some() {
145152
for match_ in &res.context().matches {
@@ -167,7 +174,7 @@ impl HatBaseTrait for HatCode {
167174
.drain()
168175
{
169176
get_mut_unchecked(&mut res).session_ctxs.remove(&face.id);
170-
undeclare_client_queryable(&mut wtables, &mut face_clone, &mut res);
177+
undeclare_client_queryable(&mut wtables, &mut face_clone, &mut res, send_declare);
171178

172179
if res.context.is_some() {
173180
for match_ in &res.context().matches {
@@ -229,6 +236,7 @@ impl HatBaseTrait for HatCode {
229236
_tables_ref: &Arc<TablesLock>,
230237
_oam: Oam,
231238
_transport: &TransportUnicast,
239+
_send_declare: &mut SendDeclare,
232240
) -> ZResult<()> {
233241
Ok(())
234242
}
@@ -248,6 +256,7 @@ impl HatBaseTrait for HatCode {
248256
_tables: &mut Tables,
249257
_tables_ref: &Arc<TablesLock>,
250258
_transport: &TransportUnicast,
259+
_send_declare: &mut SendDeclare,
251260
) -> ZResult<()> {
252261
Ok(())
253262
}

0 commit comments

Comments
 (0)