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

Optimize RoutingContext keyexpr for Query and Response messages #1266

Merged
merged 6 commits into from
Jul 26, 2024
Merged

Optimize RoutingContext keyexpr for Query and Response messages #1266

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
7c21ce0
Optimize RoutingContext keyexpr for Query and Response messages
oteffahi Jul 25, 2024
9d08747
Update Reply(egress) keyexpr logic in ACL interceptor
oteffahi Jul 25, 2024
7ad6d88
Fix parameter names
oteffahi Jul 25, 2024
0aa8ed6
Change auth lowlatency test to TLS in order to dodge issue with Quic
oteffahi Jul 25, 2024
545cc4c
Merge branch 'dev/1.0.0' into routing/interceptor-keyexpr-optimization
oteffahi Jul 26, 2024
56c4397
Optimize ResponseFinal RoutingContext, set its RoutingContext key_exp…
oteffahi Jul 26, 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
12 changes: 6 additions & 6 deletions zenoh/src/api/session.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2799,18 +2799,18 @@ impl crate::net::primitives::EPrimitives for Session {
}

#[inline]
fn send_request(&self, ctx: crate::net::routing::RoutingContext<Request>) {
(self as &dyn Primitives).send_request(ctx.msg)
fn send_request(&self, msg: Request) {
(self as &dyn Primitives).send_request(msg)
}

#[inline]
fn send_response(&self, ctx: crate::net::routing::RoutingContext<Response>) {
(self as &dyn Primitives).send_response(ctx.msg)
fn send_response(&self, msg: Response) {
(self as &dyn Primitives).send_response(msg)
}

#[inline]
fn send_response_final(&self, ctx: crate::net::routing::RoutingContext<ResponseFinal>) {
(self as &dyn Primitives).send_response_final(ctx.msg)
fn send_response_final(&self, msg: ResponseFinal) {
(self as &dyn Primitives).send_response_final(msg)
}

fn as_any(&self) -> &dyn std::any::Any {
Expand Down
12 changes: 6 additions & 6 deletions zenoh/src/net/primitives/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,11 +49,11 @@ pub(crate) trait EPrimitives: Send + Sync {

fn send_push(&self, msg: Push);

fn send_request(&self, ctx: RoutingContext<Request>);
fn send_request(&self, msg: Request);

fn send_response(&self, ctx: RoutingContext<Response>);
fn send_response(&self, msg: Response);

fn send_response_final(&self, ctx: RoutingContext<ResponseFinal>);
fn send_response_final(&self, msg: ResponseFinal);
}

#[derive(Default)]
Expand Down Expand Up @@ -82,11 +82,11 @@ impl EPrimitives for DummyPrimitives {

fn send_push(&self, _msg: Push) {}

fn send_request(&self, _ctx: RoutingContext<Request>) {}
fn send_request(&self, _msg: Request) {}

fn send_response(&self, _ctx: RoutingContext<Response>) {}
fn send_response(&self, _msg: Response) {}

fn send_response_final(&self, _ctx: RoutingContext<ResponseFinal>) {}
fn send_response_final(&self, _msg: ResponseFinal) {}

fn as_any(&self) -> &dyn Any {
self
Expand Down
246 changes: 120 additions & 126 deletions zenoh/src/net/primitives/mux.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -271,78 +271,75 @@ impl EPrimitives for Mux {
}
}

fn send_request(&self, ctx: RoutingContext<Request>) {
let ctx = RoutingContext {
msg: NetworkMessage {
body: NetworkBody::Request(ctx.msg),
#[cfg(feature = "stats")]
size: None,
},
inface: ctx.inface,
outface: ctx.outface,
prefix: ctx.prefix,
full_expr: ctx.full_expr,
fn send_request(&self, msg: Request) {
let msg = NetworkMessage {
body: NetworkBody::Request(msg),
#[cfg(feature = "stats")]
size: None,
};
let prefix = ctx
.wire_expr()
.and_then(|we| (!we.has_suffix()).then(|| ctx.prefix()))
.flatten()
.cloned();
let cache = prefix
.as_ref()
.and_then(|p| p.get_egress_cache(ctx.outface.get().unwrap()));
if let Some(ctx) = self.interceptor.intercept(ctx, cache) {
let _ = self.handler.schedule(ctx.msg);
if self.interceptor.interceptors.is_empty() {
let _ = self.handler.schedule(msg);
} else if let Some(face) = self.face.get().and_then(|f| f.upgrade()) {
let ctx = RoutingContext::new_out(msg, face.clone());
let prefix = ctx
.wire_expr()
.and_then(|we| (!we.has_suffix()).then(|| ctx.prefix()))
.flatten()
.cloned();
let cache = prefix.as_ref().and_then(|p| p.get_egress_cache(&face));
if let Some(ctx) = self.interceptor.intercept(ctx, cache) {
let _ = self.handler.schedule(ctx.msg);
}
} else {
tracing::error!("Uninitialized multiplexer!");
}
}

fn send_response(&self, ctx: RoutingContext<Response>) {
let ctx = RoutingContext {
msg: NetworkMessage {
body: NetworkBody::Response(ctx.msg),
#[cfg(feature = "stats")]
size: None,
},
inface: ctx.inface,
outface: ctx.outface,
prefix: ctx.prefix,
full_expr: ctx.full_expr,
fn send_response(&self, msg: Response) {
let msg = NetworkMessage {
body: NetworkBody::Response(msg),
#[cfg(feature = "stats")]
size: None,
};
let prefix = ctx
.wire_expr()
.and_then(|we| (!we.has_suffix()).then(|| ctx.prefix()))
.flatten()
.cloned();
let cache = prefix
.as_ref()
.and_then(|p| p.get_egress_cache(ctx.outface.get().unwrap()));
if let Some(ctx) = self.interceptor.intercept(ctx, cache) {
let _ = self.handler.schedule(ctx.msg);
if self.interceptor.interceptors.is_empty() {
let _ = self.handler.schedule(msg);
} else if let Some(face) = self.face.get().and_then(|f| f.upgrade()) {
let ctx = RoutingContext::new_out(msg, face.clone());
let prefix = ctx
.wire_expr()
.and_then(|we| (!we.has_suffix()).then(|| ctx.prefix()))
.flatten()
.cloned();
let cache = prefix.as_ref().and_then(|p| p.get_egress_cache(&face));
if let Some(ctx) = self.interceptor.intercept(ctx, cache) {
let _ = self.handler.schedule(ctx.msg);
}
} else {
tracing::error!("Uninitialized multiplexer!");
}
}

fn send_response_final(&self, ctx: RoutingContext<ResponseFinal>) {
let ctx = RoutingContext {
msg: NetworkMessage {
body: NetworkBody::ResponseFinal(ctx.msg),
#[cfg(feature = "stats")]
size: None,
},
inface: ctx.inface,
outface: ctx.outface,
prefix: ctx.prefix,
full_expr: ctx.full_expr,
fn send_response_final(&self, msg: ResponseFinal) {
let msg = NetworkMessage {
body: NetworkBody::ResponseFinal(msg),
#[cfg(feature = "stats")]
size: None,
};
let prefix = ctx
.wire_expr()
.and_then(|we| (!we.has_suffix()).then(|| ctx.prefix()))
.flatten()
.cloned();
let cache = prefix
.as_ref()
.and_then(|p| p.get_egress_cache(ctx.outface.get().unwrap()));
if let Some(ctx) = self.interceptor.intercept(ctx, cache) {
let _ = self.handler.schedule(ctx.msg);
if self.interceptor.interceptors.is_empty() {
let _ = self.handler.schedule(msg);
} else if let Some(face) = self.face.get().and_then(|f| f.upgrade()) {
let ctx = RoutingContext::new_out(msg, face.clone());
let prefix = ctx
.wire_expr()
.and_then(|we| (!we.has_suffix()).then(|| ctx.prefix()))
.flatten()
.cloned();
let cache = prefix.as_ref().and_then(|p| p.get_egress_cache(&face));
if let Some(ctx) = self.interceptor.intercept(ctx, cache) {
let _ = self.handler.schedule(ctx.msg);
}
} else {
tracing::error!("Uninitialized multiplexer!");
}
}

Expand Down Expand Up @@ -592,78 +589,75 @@ impl EPrimitives for McastMux {
}
}

fn send_request(&self, ctx: RoutingContext<Request>) {
let ctx = RoutingContext {
msg: NetworkMessage {
body: NetworkBody::Request(ctx.msg),
#[cfg(feature = "stats")]
size: None,
},
inface: ctx.inface,
outface: ctx.outface,
prefix: ctx.prefix,
full_expr: ctx.full_expr,
fn send_request(&self, msg: Request) {
let msg = NetworkMessage {
body: NetworkBody::Request(msg),
#[cfg(feature = "stats")]
size: None,
};
let prefix = ctx
.wire_expr()
.and_then(|we| (!we.has_suffix()).then(|| ctx.prefix()))
.flatten()
.cloned();
let cache = prefix
.as_ref()
.and_then(|p| p.get_egress_cache(ctx.outface.get().unwrap()));
if let Some(ctx) = self.interceptor.intercept(ctx, cache) {
let _ = self.handler.schedule(ctx.msg);
if self.interceptor.interceptors.is_empty() {
let _ = self.handler.schedule(msg);
} else if let Some(face) = self.face.get() {
let ctx = RoutingContext::new_out(msg, face.clone());
let prefix = ctx
.wire_expr()
.and_then(|we| (!we.has_suffix()).then(|| ctx.prefix()))
.flatten()
.cloned();
let cache = prefix.as_ref().and_then(|p| p.get_egress_cache(face));
if let Some(ctx) = self.interceptor.intercept(ctx, cache) {
let _ = self.handler.schedule(ctx.msg);
}
} else {
tracing::error!("Uninitialized multiplexer!");
}
}

fn send_response(&self, ctx: RoutingContext<Response>) {
let ctx = RoutingContext {
msg: NetworkMessage {
body: NetworkBody::Response(ctx.msg),
#[cfg(feature = "stats")]
size: None,
},
inface: ctx.inface,
outface: ctx.outface,
prefix: ctx.prefix,
full_expr: ctx.full_expr,
fn send_response(&self, msg: Response) {
let msg = NetworkMessage {
body: NetworkBody::Response(msg),
#[cfg(feature = "stats")]
size: None,
};
let prefix = ctx
.wire_expr()
.and_then(|we| (!we.has_suffix()).then(|| ctx.prefix()))
.flatten()
.cloned();
let cache = prefix
.as_ref()
.and_then(|p| p.get_egress_cache(ctx.outface.get().unwrap()));
if let Some(ctx) = self.interceptor.intercept(ctx, cache) {
let _ = self.handler.schedule(ctx.msg);
if self.interceptor.interceptors.is_empty() {
let _ = self.handler.schedule(msg);
} else if let Some(face) = self.face.get() {
let ctx = RoutingContext::new_out(msg, face.clone());
let prefix = ctx
.wire_expr()
.and_then(|we| (!we.has_suffix()).then(|| ctx.prefix()))
.flatten()
.cloned();
let cache = prefix.as_ref().and_then(|p| p.get_egress_cache(face));
if let Some(ctx) = self.interceptor.intercept(ctx, cache) {
let _ = self.handler.schedule(ctx.msg);
}
} else {
tracing::error!("Uninitialized multiplexer!");
}
}

fn send_response_final(&self, ctx: RoutingContext<ResponseFinal>) {
let ctx = RoutingContext {
msg: NetworkMessage {
body: NetworkBody::ResponseFinal(ctx.msg),
#[cfg(feature = "stats")]
size: None,
},
inface: ctx.inface,
outface: ctx.outface,
prefix: ctx.prefix,
full_expr: ctx.full_expr,
fn send_response_final(&self, msg: ResponseFinal) {
let msg = NetworkMessage {
body: NetworkBody::ResponseFinal(msg),
#[cfg(feature = "stats")]
size: None,
};
let prefix = ctx
.wire_expr()
.and_then(|we| (!we.has_suffix()).then(|| ctx.prefix()))
.flatten()
.cloned();
let cache = prefix
.as_ref()
.and_then(|p| p.get_egress_cache(ctx.outface.get().unwrap()));
if let Some(ctx) = self.interceptor.intercept(ctx, cache) {
let _ = self.handler.schedule(ctx.msg);
if self.interceptor.interceptors.is_empty() {
let _ = self.handler.schedule(msg);
} else if let Some(face) = self.face.get() {
let ctx = RoutingContext::new_out(msg, face.clone());
let prefix = ctx
.wire_expr()
.and_then(|we| (!we.has_suffix()).then(|| ctx.prefix()))
.flatten()
.cloned();
let cache = prefix.as_ref().and_then(|p| p.get_egress_cache(face));
if let Some(ctx) = self.interceptor.intercept(ctx, cache) {
let _ = self.handler.schedule(ctx.msg);
}
} else {
tracing::error!("Uninitialized multiplexer!");
}
}

Expand Down
Loading