docker/pgcat
1
0
Fork 0
mirror of https://github.com/postgresml/pgcat.git synced 2026-03-23 01:16:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

Reduce memory and CPU footprint of mirroring (#369)

Browse Source 
The experimental mirroring feature used a lot of memory and CPU when put under production traffic. This change attempts to reduce memory and CPU usage.

Memory footprint is reduced by making the channel smaller. CPU usage is reduced by avoiding allocations if the channel is full or is closed.

We might lose more messages this way if the mirror falls behind but that is more acceptable than crashing the entire process when it goes out-of-memory (OOM)
This commit is contained in:
Mostafa Abdelraouf
2023-03-15 17:58:45 -05:00
committed by GitHub
parent 9a668e584f
commit e5df179ac9
1 changed files with 16 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
src/mirrors.rs
View File

@@ -119,7 +119,7 @@ impl MirroringManager {
let mut exit_senders: Vec<Sender<()>> = vec![];
addresses.iter().for_each(|mirror| {
let (bytes_tx, bytes_rx) = channel::<Bytes>(500);
let (bytes_tx, bytes_rx) = channel::<Bytes>(10);
let (exit_tx, exit_rx) = channel::<()>(1);
let mut addr = mirror.clone();
addr.role = Role::Mirror;
@@ -142,15 +142,25 @@ impl MirroringManager {
}
pub fn send(self: &mut Self, bytes: &BytesMut) {
let cpy = bytes.clone().freeze();
self.byte_senders
.iter_mut()
.for_each(|sender| match sender.try_send(cpy.clone()) {
// We want to avoid performing an allocation if we won't be able to send the message
// There is a possibility of a race here where we check the capacity and then the channel is
// closed or the capacity is reduced to 0, but mirroring is best effort anyway
if self
.byte_senders
.iter()
.all(|sender| sender.capacity() == 0 || sender.is_closed())
{
return;
}
let immutable_bytes = bytes.clone().freeze();
self.byte_senders.iter_mut().for_each(|sender| {
match sender.try_send(immutable_bytes.clone()) {
Ok(_) => {}
Err(err) => {
warn!("Failed to send bytes to a mirror channel {}", err);
}
});
}
});
}
pub fn disconnect(self: &mut Self) {