Skip to content
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

Enhance subscribers, queryables and liveliness tokens propagation to improve scalability for peers subsystems #1044

Merged
merged 7 commits into from
May 31, 2024
Merged

Enhance subscribers, queryables and liveliness tokens propagation to improve scalability for peers subsystems #1044

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
3c5d4de
Implement interest protocol between peers and routers
OlivierHecart May 22, 2024
d7bb636
Add logger init in matching test
OlivierHecart May 22, 2024
555f012
Peers send subscribers interests to newly connected routers
OlivierHecart May 24, 2024
8a7987c
Remove commented code
OlivierHecart May 24, 2024
f7bf8be
Remove empty functions
OlivierHecart May 30, 2024
69a70be
Merge branch 'dev/1.0.0' into peers_interests
Mallets May 30, 2024
3fa2c56
Fix code fmt
Mallets May 30, 2024
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
13 changes: 5 additions & 8 deletions zenoh/src/net/routing/hat/linkstate_peer/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,8 +41,8 @@ use zenoh_transport::unicast::TransportUnicast;

use self::{
network::Network,
pubsub::{pubsub_new_face, pubsub_remove_node, undeclare_client_subscription},
queries::{queries_new_face, queries_remove_node, undeclare_client_queryable},
pubsub::{pubsub_remove_node, undeclare_client_subscription},
queries::{queries_remove_node, undeclare_client_queryable},
};
use super::{
super::dispatcher::{
Expand Down Expand Up @@ -212,12 +212,11 @@ impl HatBaseTrait for HatCode {

fn new_local_face(
&self,
tables: &mut Tables,
_tables: &mut Tables,
_tables_ref: &Arc<TablesLock>,
face: &mut Face,
_face: &mut Face,
) -> ZResult<()> {
pubsub_new_face(tables, &mut face.state);
queries_new_face(tables, &mut face.state);
// Nothing to do
Ok(())
}

Expand All @@ -239,8 +238,6 @@ impl HatBaseTrait for HatCode {
};

face_hat_mut!(&mut face.state).link_id = link_id;
pubsub_new_face(tables, &mut face.state);
queries_new_face(tables, &mut face.state);

if face.state.whatami != WhatAmI::Client {
hat_mut!(tables).schedule_compute_trees(tables_ref.clone());
Expand Down
4 changes: 0 additions & 4 deletions zenoh/src/net/routing/hat/linkstate_peer/pubsub.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -554,10 +554,6 @@ fn forget_client_subscription(
}
}

pub(super) fn pubsub_new_face(_tables: &mut Tables, _face: &mut Arc<FaceState>) {
// Nothing to do
}

pub(super) fn pubsub_remove_node(tables: &mut Tables, node: &ZenohId) {
for mut res in hat!(tables)
.peer_subs
Expand Down
4 changes: 0 additions & 4 deletions zenoh/src/net/routing/hat/linkstate_peer/queries.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -584,10 +584,6 @@ fn forget_client_queryable(
}
}

pub(super) fn queries_new_face(_tables: &mut Tables, _face: &mut Arc<FaceState>) {
// Nothing to do
}

pub(super) fn queries_remove_node(tables: &mut Tables, node: &ZenohId) {
let mut qabls = vec![];
for res in hat!(tables).peer_qabls.iter() {
Expand Down
266 changes: 148 additions & 118 deletions zenoh/src/net/routing/hat/p2p_peer/pubsub.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,8 @@ use zenoh_protocol::{
common::ext::WireExprType, ext, subscriber::ext::SubscriberInfo, Declare, DeclareBody,
DeclareSubscriber, SubscriberId, UndeclareSubscriber,
},
interest::{InterestId, InterestMode},
interest::{InterestId, InterestMode, InterestOptions},
Interest,
},
};
use zenoh_sync::get_mut_unchecked;
Expand All @@ -34,11 +35,11 @@ use crate::{
key_expr::KeyExpr,
net::routing::{
dispatcher::{
face::FaceState,
face::{FaceState, InterestState},
resource::{NodeId, Resource, SessionContext},
tables::{Route, RoutingExpr, Tables},
},
hat::{CurrentFutureTrait, HatPubSubTrait, Sources},
hat::{HatPubSubTrait, Sources},
router::{update_data_routes_from, RoutesIndexes},
RoutingContext, PREFIX_LIVELINESS,
},
Expand Down Expand Up @@ -341,7 +342,7 @@ pub(super) fn pubsub_new_face(tables: &mut Tables, face: &mut Arc<FaceState>) {
let sub_info = SubscriberInfo {
reliability: Reliability::Reliable, // @TODO compute proper reliability to propagate from reliability of known subscribers
};
for src_face in tables
for mut src_face in tables
.faces
.values()
.cloned()
Expand All @@ -356,6 +357,33 @@ pub(super) fn pubsub_new_face(tables: &mut Tables, face: &mut Arc<FaceState>) {
&mut src_face.clone(),
);
}
if face.whatami == WhatAmI::Router {
for (res, _) in face_hat_mut!(&mut src_face).remote_sub_interests.values() {
let id = face_hat!(face).next_id.fetch_add(1, Ordering::SeqCst);
let options = InterestOptions::KEYEXPRS + InterestOptions::SUBSCRIBERS;
get_mut_unchecked(face).local_interests.insert(
id,
InterestState {
options,
res: res.as_ref().map(|res| (*res).clone()),
finalized: false,
},
);
let wire_expr = res.as_ref().map(|res| Resource::decl_key(res, face));
face.primitives.send_interest(RoutingContext::with_expr(
Interest {
id,
mode: InterestMode::CurrentFuture,
options,
wire_expr,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
},
res.as_ref().map(|res| res.expr()).unwrap_or_default(),
));
}
}
}
}
// recompute routes
Expand All @@ -374,133 +402,91 @@ impl HatPubSubTrait for HatCode {
mode: InterestMode,
aggregate: bool,
) {
if mode.current() && face.whatami == WhatAmI::Client {
let interest_id = (!mode.future()).then_some(id);
let sub_info = SubscriberInfo {
reliability: Reliability::Reliable, // @TODO compute proper reliability to propagate from reliability of known subscribers
};
if let Some(res) = res.as_ref() {
if aggregate {
if tables.faces.values().any(|src_face| {
src_face.id != face.id
&& face_hat!(src_face)
.remote_subs
.values()
.any(|sub| sub.context.is_some() && sub.matches(res))
}) {
let id = if mode.future() {
let id = face_hat!(face).next_id.fetch_add(1, Ordering::SeqCst);
face_hat_mut!(face).local_subs.insert((*res).clone(), id);
id
} else {
0
};
let wire_expr = Resource::decl_key(res, face);
face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
body: DeclareBody::DeclareSubscriber(DeclareSubscriber {
id,
wire_expr,
ext_info: sub_info,
}),
},
res.expr(),
));
}
} else {
for src_face in tables
.faces
face_hat_mut!(face)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If I understand correctly. Now we receive all the Subscriber declarations from the Router connected to one of the node in the peer-to-peer subsystem. Thus we need only forward peer Interest declarations to the Router and expect to receive Subscriber declarations.

I write this just to confirm that I understand the purpose of this change.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Before, when receiving interest from a client, a peer only had to answer its interest with its local knowledge of subscribers. Now it needs to propagate this interest to Routers.

.remote_sub_interests
.insert(id, (res.as_ref().map(|res| (*res).clone()), aggregate));
for dst_face in tables
.faces
.values_mut()
.filter(|f| f.whatami == WhatAmI::Router)
{
let id = face_hat!(dst_face).next_id.fetch_add(1, Ordering::SeqCst);
let options = InterestOptions::KEYEXPRS + InterestOptions::SUBSCRIBERS;
get_mut_unchecked(dst_face).local_interests.insert(
id,
InterestState {
options,
res: res.as_ref().map(|res| (*res).clone()),
finalized: mode == InterestMode::Future,
},
);
let wire_expr = res.as_ref().map(|res| Resource::decl_key(res, dst_face));
dst_face.primitives.send_interest(RoutingContext::with_expr(
Interest {
id,
mode,
options,
wire_expr,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
},
res.as_ref().map(|res| res.expr()).unwrap_or_default(),
));
}
}

fn undeclare_sub_interest(
&self,
tables: &mut Tables,
face: &mut Arc<FaceState>,
id: InterestId,
) {
if let Some(interest) = face_hat_mut!(face).remote_sub_interests.remove(&id) {
if !tables.faces.values().any(|f| {
f.whatami == WhatAmI::Client
&& face_hat!(f)
.remote_sub_interests
.values()
.cloned()
.collect::<Vec<Arc<FaceState>>>()
{
if src_face.id != face.id {
for sub in face_hat!(src_face).remote_subs.values() {
if sub.context.is_some() && sub.matches(res) {
let id = if mode.future() {
let id =
face_hat!(face).next_id.fetch_add(1, Ordering::SeqCst);
face_hat_mut!(face).local_subs.insert(sub.clone(), id);
id
} else {
0
};
let wire_expr = Resource::decl_key(sub, face);
face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
body: DeclareBody::DeclareSubscriber(
DeclareSubscriber {
id,
wire_expr,
ext_info: sub_info,
},
),
},
sub.expr(),
));
}
}
}
}
}
} else {
for src_face in tables
.any(|i| *i == interest)
}) {
for dst_face in tables
.faces
.values()
.cloned()
.collect::<Vec<Arc<FaceState>>>()
.values_mut()
.filter(|f| f.whatami == WhatAmI::Router)
{
if src_face.id != face.id {
for sub in face_hat!(src_face).remote_subs.values() {
let id = if mode.future() {
let id = face_hat!(face).next_id.fetch_add(1, Ordering::SeqCst);
face_hat_mut!(face).local_subs.insert(sub.clone(), id);
id
} else {
0
};
let wire_expr = Resource::decl_key(sub, face);
face.primitives.send_declare(RoutingContext::with_expr(
Declare {
interest_id,
for id in dst_face
.local_interests
.keys()
.cloned()
.collect::<Vec<InterestId>>()
{
let local_interest = dst_face.local_interests.get(&id).unwrap();
if local_interest.options.subscribers()
&& (local_interest.res == interest.0)
{
dst_face.primitives.send_interest(RoutingContext::with_expr(
Interest {
id,
mode: InterestMode::Final,
options: InterestOptions::empty(),
wire_expr: None,
ext_qos: ext::QoSType::DECLARE,
ext_tstamp: None,
ext_nodeid: ext::NodeIdType::DEFAULT,
body: DeclareBody::DeclareSubscriber(DeclareSubscriber {
id,
wire_expr,
ext_info: sub_info,
}),
},
sub.expr(),
local_interest
.res
.as_ref()
.map(|res| res.expr())
.unwrap_or_default(),
));
get_mut_unchecked(dst_face).local_interests.remove(&id);
}
}
}
}
}
if mode.future() {
face_hat_mut!(face)
.remote_sub_interests
.insert(id, (res.cloned(), aggregate));
}
}

fn undeclare_sub_interest(
&self,
_tables: &mut Tables,
face: &mut Arc<FaceState>,
id: InterestId,
) {
face_hat_mut!(face).remote_sub_interests.remove(&id);
}

fn declare_subscription(
Expand Down Expand Up @@ -570,6 +556,50 @@ impl HatPubSubTrait for HatCode {
}
};

for face in tables
.faces
.values()
.filter(|f| f.whatami == WhatAmI::Router)
{
if face.local_interests.values().any(|interest| {
interest.finalized
&& interest.options.subscribers()
&& interest
.res
.as_ref()
.map(|res| {
KeyExpr::try_from(res.expr())
.and_then(|intres| {
KeyExpr::try_from(expr.full_expr())
.map(|putres| intres.includes(&putres))
})
.unwrap_or(false)
})
.unwrap_or(true)
}) {
if face_hat!(face).remote_subs.values().any(|sub| {
KeyExpr::try_from(sub.expr())
.and_then(|subres| {
KeyExpr::try_from(expr.full_expr())
.map(|putres| subres.intersects(&putres))
})
.unwrap_or(false)
}) {
let key_expr = Resource::get_best_key(expr.prefix, expr.suffix, face.id);
route.insert(
face.id,
(face.clone(), key_expr.to_owned(), NodeId::default()),
);
}
} else {
let key_expr = Resource::get_best_key(expr.prefix, expr.suffix, face.id);
route.insert(
face.id,
(face.clone(), key_expr.to_owned(), NodeId::default()),
);
}
}

for face in tables.faces.values().filter(|f| {
f.whatami == WhatAmI::Peer
&& !f
Expand Down
10 changes: 10 additions & 0 deletions zenoh/src/net/routing/hat/p2p_peer/queries.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -554,6 +554,16 @@ impl HatQueriesTrait for HatCode {
}
};

// TODO: BNestMatching: What if there is a local compete ?
if let Some(face) = tables.faces.values().find(|f| f.whatami == WhatAmI::Router) {
let key_expr = Resource::get_best_key(expr.prefix, expr.suffix, face.id);
route.push(QueryTargetQabl {
direction: (face.clone(), key_expr.to_owned(), NodeId::default()),
complete: 0,
distance: f64::MAX,
});
}

for face in tables.faces.values().filter(|f| {
f.whatami == WhatAmI::Peer
&& !f
Expand Down
Loading