Possible Memory Leak

[jsouto18]

I've been using this library and recently I've noticed that when using a Delegate, memory seems to leak. I have the following Minimum Reproducible Example where if you run this with say 5MB messages on the queue you will notice that the memory of your process will never stop increasing. Am I doing something wrong?

MRE:

use lapin::{
    options::{BasicAckOptions, BasicConsumeOptions, BasicQosOptions},
    types::FieldTable,
    Connection, ConnectionProperties, ConsumerDelegate,
};

#[tokio::main]
async fn main() {
    let amqp_connection = Connection::connect(
        "amqp://127.0.0.1:5672/%2f",
        ConnectionProperties::default()
            .with_executor(tokio_executor_trait::Tokio::current())
            .with_reactor(tokio_reactor_trait::Tokio),
    )
    .await
    .unwrap();
    let consumer_channel = amqp_connection.create_channel().await.unwrap();

    consumer_channel
        .basic_qos(1, BasicQosOptions::default())
        .await
        .unwrap();

    consumer_channel
        .basic_consume(
            "entity_cache",
            "",
            BasicConsumeOptions::default(),
            FieldTable::default(),
        )
        .await
        .unwrap()
        .set_delegate(Test {});

    std::future::pending::<()>().await;
}

struct Test;
impl ConsumerDelegate for Test {
    fn on_new_delivery(
        &self,
        delivery: lapin::message::DeliveryResult,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>> {
        let delivery = delivery.expect("error in consumer").expect("???");

        Box::pin(async move {
            delivery.ack(BasicAckOptions::default()).await.expect("ack");
        })
    }
}

[jsouto18]

@Keruspe can I somehow help with the problem? I'm happy to put a bounty on it or help solve it.

[Keruspe]

Could you send me an email? <first-name>@<lastname>.com

[RobinDBL]

Is there more info about this? I'm having the same issue. I think the issue lies with the acknowledgement of the messages.

[ginkcode]

The snippet code above quite smells, especially inside the delegate. I think the first step is removing all of .expect() or .unwrap() inside it. Handle the error properly, make sure the delegate always return correctly.

If it still happens then start thinking about memory leak

[jsouto18]

The snippet code above quite smells, especially inside the delegate. I think the first step is removing all of .expect() or .unwrap() inside it. Handle the error properly, make sure the delegate always return correctly.

If it still happens then start thinking about memory leak

This is meant to be a "Minimum Reproducible Example" so error handling is not a priority nor a necessity. The error handling is not the cause here nor do I understand how you would assume that...

[jsouto18]

@Keruspe I did some investigations and this seems to be related to https://github.com/tokio-rs/tokio/issues/4406. I tested with jemallocator and the memory allocation is no longer retained. I believe we can close this issue as it's not related with lapin

[Keruspe]

Ah ! Thanks for digging this out 👍

Glad it's not because of lapin in the end