Page MenuHomePhabricator
Differential D13604

[tunnelbroker] Use AMQP wrapper in servers
ClosedPublic
Actions

Authored by bartek on Oct 3 2024, 1:36 PM.
Tags
None
Referenced Files
Unknown Object (File)
Sat, Apr 5, 1:01 PM
Unknown Object (File)
Mar 16 2025, 1:13 PM
Unknown Object (File)
Mar 11 2025, 4:16 PM
Unknown Object (File)
Mar 11 2025, 4:11 PM
Unknown Object (File)
Mar 11 2025, 4:10 PM
Unknown Object (File)
Mar 8 2025, 11:06 PM
Unknown Object (File)
Mar 7 2025, 8:09 PM
Unknown Object (File)
Mar 7 2025, 7:46 PM
View All 64 Files
Subscribers
ashoat
tomek

Details

Reviewers
kamil
varun
will
Commits
rCOMMe2324a90efb1: [tunnelbroker] Use AMQP wrapper in servers
Summary

Replacing legacy connection with the new wrapper struct

Depends on D13603

Test Plan

cargo build, tunnelbroker integration tests

Diff Detail

Repository
rCOMM Comm
Lint
Lint Not Applicable
Unit
Tests Not Applicable

Event Timeline

bartek created this revision.Oct 3 2024, 1:36 PM
bartek held this revision as a draft.
Herald added subscribers: tomek, ashoat. · View Herald TranscriptOct 3 2024, 1:36 PM
bartek added a child revision: D13605: [tunnelbroker] Refactor Websocket session init.Oct 3 2024, 1:36 PM
Harbormaster completed remote builds in B32055: Diff 44884.Oct 3 2024, 2:01 PM
bartek edited the summary of this revision. (Show Details)Oct 4 2024, 3:50 AM
bartek edited the test plan for this revision. (Show Details)
bartek published this revision for review.Oct 4 2024, 3:52 AM
bartek added inline comments.
services/tunnelbroker/src/grpc/mod.rs
61–64 ↗(On Diff #44884)

By retrieving it each time we make sure it's connected - the channel().await takes care of it

131–134 ↗(On Diff #44884)

This expect caused crashes when trying to create channels on already-broken connection

bartek updated this revision to Diff 44901.Oct 4 2024, 8:17 AM

Remove connection pool

Harbormaster completed remote builds in B32069: Diff 44901.Oct 4 2024, 8:39 AM
bartek mentioned this in D13621: [tunnelbroker] Implement lazy AMQP channel wrapper.Oct 6 2024, 11:51 PM
kamil accepted this revision.Oct 7 2024, 1:52 AM
This revision is now accepted and ready to land.Oct 7 2024, 1:52 AM
Closed by commit rCOMMe2324a90efb1: [tunnelbroker] Use AMQP wrapper in servers (authored by bartek). · Explain WhyOct 7 2024, 7:28 AM
This revision was automatically updated to reflect the committed changes.
bartek added a commit: rCOMMe2324a90efb1: [tunnelbroker] Use AMQP wrapper in servers.
bartek mentioned this in rCOMMb92865431259: [tunnelbroker] Implement lazy AMQP channel wrapper.

Revision Contents
Changeset List

PathSize
services/
tunnelbroker/
src/
grpc/
22 lines
4 lines
websockets/
26 lines

Diff 44933

View Options

services/tunnelbroker/src/grpc/mod.rs

Show First 20 Lines • Show All 90 Lines • ▼ Show 20 Lines Ok(Self {
.try_into() .try_into()
.expect("Channels vec size doesn't match array size"), .expect("Channels vec size doesn't match array size"),
}) })
} }
fn get_channel( fn get_channel(
&self, &self,
id_hash: impl std::hash::Hash, id_hash: impl std::hash::Hash,
) -> Result<lapin::Channel, lapin::Error> { ) -> Option<lapin::Channel> {
let channel_idx = Self::channel_idx_for_hash(id_hash); let channel_idx = Self::channel_idx_for_hash(id_hash);
let channel = self.channels[channel_idx].clone(); let channel = self.channels[channel_idx].clone();
let channel_id = channel.id(); let channel_id = channel.id();
tracing::trace!(channel_id, channel_idx, "Retrieving AMQP Channel"); tracing::trace!(channel_id, channel_idx, "Retrieving AMQP Channel");
Ok(channel)
if channel.status().connected() {
return Some(channel);
}
warn!(channel_id, channel_idx, "Channel is dead!");
None
}
async fn reset_channel(
&mut self,
id_hash: impl std::hash::Hash,
) -> Result<lapin::Channel, lapin::Error> {
let channel_idx = Self::channel_idx_for_hash(id_hash);
let existing_channel = &self.channels[channel_idx];
if existing_channel.status().connected() {
return Ok(existing_channel.clone());
}
let new_channel = self.conn.create_channel().await?;
debug!(
"Channel for idx={channel_idx} was recreated, new Id={}",
new_channel.id()
);
self.channels[channel_idx] = new_channel.clone();
Ok(new_channel)
} }
fn is_connected(&self) -> bool { fn is_connected(&self) -> bool {
self.conn.status().connected() self.conn.status().connected()
} }
fn channel_idx_for_hash(id_hash: impl std::hash::Hash) -> usize { fn channel_idx_for_hash(id_hash: impl std::hash::Hash) -> usize {
// We have channel pool and want to distribute them between connected // We have channel pool and want to distribute them between connected
Show All 22 Lines pub async fn connect() -> Result<Self, lapin::Error> {
let inner = Arc::new(RwLock::new(conn)); let inner = Arc::new(RwLock::new(conn));
info!("Connected to AMQP endpoint: {}", &CONFIG.amqp_uri); info!("Connected to AMQP endpoint: {}", &CONFIG.amqp_uri);
Ok(Self { inner }) Ok(Self { inner })
} }
pub async fn channel( pub async fn channel(
&self, &self,
id_hash: impl std::hash::Hash, id_hash: impl std::hash::Hash + Clone,
) -> Result<lapin::Channel, lapin::Error> { ) -> Result<lapin::Channel, lapin::Error> {
if !self.is_connected().await { if !self.is_connected().await {
warn!("AMQP disconnected while retrieving channel"); warn!("AMQP disconnected while retrieving channel");
self.reset_conn().await?; self.reset_conn().await?;
} }
self.inner.read().await.get_channel(id_hash)
{
let inner = self.inner.read().await;
if let Some(channel) = inner.get_channel(id_hash.clone()) {
return Ok(channel);
}
}
self.inner.write().await.reset_channel(id_hash).await
} }
async fn reset_conn(&self) -> Result<(), lapin::Error> { async fn reset_conn(&self) -> Result<(), lapin::Error> {
debug!("Resetting connection..."); debug!("Resetting connection...");
let mut inner = self.inner.write().await; let mut inner = self.inner.write().await;
if !inner.is_connected() { if !inner.is_connected() {
let new_conn = ConnectionInner::new().await?; let new_conn = ConnectionInner::new().await?;
*inner = new_conn; *inner = new_conn;
▲ Show 20 Lines • Show All 41 Lines • Show Last 20 Lines
View Options

services/tunnelbroker/src/main.rs

Show First 20 Lines • Show All 90 Lines • ▼ Show 20 Lines Ok(Self {
.try_into() .try_into()
.expect("Channels vec size doesn't match array size"), .expect("Channels vec size doesn't match array size"),
}) })
} }
fn get_channel( fn get_channel(
&self, &self,
id_hash: impl std::hash::Hash, id_hash: impl std::hash::Hash,
) -> Result<lapin::Channel, lapin::Error> { ) -> Option<lapin::Channel> {
let channel_idx = Self::channel_idx_for_hash(id_hash); let channel_idx = Self::channel_idx_for_hash(id_hash);
let channel = self.channels[channel_idx].clone(); let channel = self.channels[channel_idx].clone();
let channel_id = channel.id(); let channel_id = channel.id();
tracing::trace!(channel_id, channel_idx, "Retrieving AMQP Channel"); tracing::trace!(channel_id, channel_idx, "Retrieving AMQP Channel");
Ok(channel)
if channel.status().connected() {
return Some(channel);
}
warn!(channel_id, channel_idx, "Channel is dead!");
None
}
async fn reset_channel(
&mut self,
id_hash: impl std::hash::Hash,
) -> Result<lapin::Channel, lapin::Error> {
let channel_idx = Self::channel_idx_for_hash(id_hash);
let existing_channel = &self.channels[channel_idx];
if existing_channel.status().connected() {
return Ok(existing_channel.clone());
}
let new_channel = self.conn.create_channel().await?;
debug!(
"Channel for idx={channel_idx} was recreated, new Id={}",
new_channel.id()
);
self.channels[channel_idx] = new_channel.clone();
Ok(new_channel)
} }
fn is_connected(&self) -> bool { fn is_connected(&self) -> bool {
self.conn.status().connected() self.conn.status().connected()
} }
fn channel_idx_for_hash(id_hash: impl std::hash::Hash) -> usize { fn channel_idx_for_hash(id_hash: impl std::hash::Hash) -> usize {
// We have channel pool and want to distribute them between connected // We have channel pool and want to distribute them between connected
Show All 22 Lines pub async fn connect() -> Result<Self, lapin::Error> {
let inner = Arc::new(RwLock::new(conn)); let inner = Arc::new(RwLock::new(conn));
info!("Connected to AMQP endpoint: {}", &CONFIG.amqp_uri); info!("Connected to AMQP endpoint: {}", &CONFIG.amqp_uri);
Ok(Self { inner }) Ok(Self { inner })
} }
pub async fn channel( pub async fn channel(
&self, &self,
id_hash: impl std::hash::Hash, id_hash: impl std::hash::Hash + Clone,
) -> Result<lapin::Channel, lapin::Error> { ) -> Result<lapin::Channel, lapin::Error> {
if !self.is_connected().await { if !self.is_connected().await {
warn!("AMQP disconnected while retrieving channel"); warn!("AMQP disconnected while retrieving channel");
self.reset_conn().await?; self.reset_conn().await?;
} }
self.inner.read().await.get_channel(id_hash)
{
let inner = self.inner.read().await;
if let Some(channel) = inner.get_channel(id_hash.clone()) {
return Ok(channel);
}
}
self.inner.write().await.reset_channel(id_hash).await
} }
async fn reset_conn(&self) -> Result<(), lapin::Error> { async fn reset_conn(&self) -> Result<(), lapin::Error> {
debug!("Resetting connection..."); debug!("Resetting connection...");
let mut inner = self.inner.write().await; let mut inner = self.inner.write().await;
if !inner.is_connected() { if !inner.is_connected() {
let new_conn = ConnectionInner::new().await?; let new_conn = ConnectionInner::new().await?;
*inner = new_conn; *inner = new_conn;
▲ Show 20 Lines • Show All 41 Lines • Show Last 20 Lines
View Options

services/tunnelbroker/src/websockets/mod.rs

Show First 20 Lines • Show All 90 Lines • ▼ Show 20 Lines Ok(Self {
.try_into() .try_into()
.expect("Channels vec size doesn't match array size"), .expect("Channels vec size doesn't match array size"),
}) })
} }
fn get_channel( fn get_channel(
&self, &self,
id_hash: impl std::hash::Hash, id_hash: impl std::hash::Hash,
) -> Result<lapin::Channel, lapin::Error> { ) -> Option<lapin::Channel> {
let channel_idx = Self::channel_idx_for_hash(id_hash); let channel_idx = Self::channel_idx_for_hash(id_hash);
let channel = self.channels[channel_idx].clone(); let channel = self.channels[channel_idx].clone();
let channel_id = channel.id(); let channel_id = channel.id();
tracing::trace!(channel_id, channel_idx, "Retrieving AMQP Channel"); tracing::trace!(channel_id, channel_idx, "Retrieving AMQP Channel");
Ok(channel)
if channel.status().connected() {
return Some(channel);
}
warn!(channel_id, channel_idx, "Channel is dead!");
None
}
async fn reset_channel(
&mut self,
id_hash: impl std::hash::Hash,
) -> Result<lapin::Channel, lapin::Error> {
let channel_idx = Self::channel_idx_for_hash(id_hash);
let existing_channel = &self.channels[channel_idx];
if existing_channel.status().connected() {
return Ok(existing_channel.clone());
}
let new_channel = self.conn.create_channel().await?;
debug!(
"Channel for idx={channel_idx} was recreated, new Id={}",
new_channel.id()
);
self.channels[channel_idx] = new_channel.clone();
Ok(new_channel)
} }
fn is_connected(&self) -> bool { fn is_connected(&self) -> bool {
self.conn.status().connected() self.conn.status().connected()
} }
fn channel_idx_for_hash(id_hash: impl std::hash::Hash) -> usize { fn channel_idx_for_hash(id_hash: impl std::hash::Hash) -> usize {
// We have channel pool and want to distribute them between connected // We have channel pool and want to distribute them between connected
Show All 22 Lines pub async fn connect() -> Result<Self, lapin::Error> {
let inner = Arc::new(RwLock::new(conn)); let inner = Arc::new(RwLock::new(conn));
info!("Connected to AMQP endpoint: {}", &CONFIG.amqp_uri); info!("Connected to AMQP endpoint: {}", &CONFIG.amqp_uri);
Ok(Self { inner }) Ok(Self { inner })
} }
pub async fn channel( pub async fn channel(
&self, &self,
id_hash: impl std::hash::Hash, id_hash: impl std::hash::Hash + Clone,
) -> Result<lapin::Channel, lapin::Error> { ) -> Result<lapin::Channel, lapin::Error> {
if !self.is_connected().await { if !self.is_connected().await {
warn!("AMQP disconnected while retrieving channel"); warn!("AMQP disconnected while retrieving channel");
self.reset_conn().await?; self.reset_conn().await?;
} }
self.inner.read().await.get_channel(id_hash)
{
let inner = self.inner.read().await;
if let Some(channel) = inner.get_channel(id_hash.clone()) {
return Ok(channel);
}
}
self.inner.write().await.reset_channel(id_hash).await
} }
async fn reset_conn(&self) -> Result<(), lapin::Error> { async fn reset_conn(&self) -> Result<(), lapin::Error> {
debug!("Resetting connection..."); debug!("Resetting connection...");
let mut inner = self.inner.write().await; let mut inner = self.inner.write().await;
if !inner.is_connected() { if !inner.is_connected() {
let new_conn = ConnectionInner::new().await?; let new_conn = ConnectionInner::new().await?;
*inner = new_conn; *inner = new_conn;
▲ Show 20 Lines • Show All 41 Lines • Show Last 20 Lines