Memory leak with Rayon and Indicatif in Rust

Question

So, I'm trying to do a exhaustive search of a hash. The hash itself is not important here. As I want to use all processing power of my CPU, I'm using Rayon to get a thread pool and lots of tasks. The search algorithm is the following:

let (tx, rx) = mpsc::channel();

let original_hash = String::from(original_hash);

rayon::spawn(move || {
    let mut i = 0;
    let mut iter = SenhaIterator::from(initial_pwd);
    while i < max_iteracoes {
        let pwd = iter.next().unwrap();

        let clone_tx = tx.clone();
        rayon::spawn(move || {
            let hash = calcula_hash(&pwd);
            clone_tx.send((pwd, hash)).unwrap();
        });

        i += 1;
    }
});

let mut last_pwd = None;
let bar = ProgressBar::new(max_iteracoes as u64);

while let Ok((pwd, hash)) = rx.recv() {
    last_pwd = Some(pwd);
    if hash == original_hash {
        bar.finish();
        return last_pwd.map_or(ResultadoSenha::SenhaNaoEncontrada(None), |s| {
            ResultadoSenha::SenhaEncontrada(s)
        });
    }
    bar.inc(1);
}
bar.finish();
ResultadoSenha::SenhaNaoEncontrada(last_pwd)

Just a high level explanation: as the tasks go completing their work, they send a pair of (password, hash) to the main thread, which will compare the hash with the original hash (the one I'm trying to find a password for). If they match, great, I return to main with an enum value that indicates success, and the password that produces the original hash. After all iterations end, I'll return to main with an enum value that indicates that no hash was found, but with the last password, so I can retry from this point in some future run.

I'm trying to use Indicatif to show a progress bar, so I can get a glimpse of the progress.

But my problem is that the program is growing it's memory usage without a clear reason why. If I make it run with, let's say, 1 billion iterations, it goes slowly adding memory until it fills all available system memory.

But when I comment the line bar.inc(1);, the program behaves as expected, with normal memory usage.

I've created a test program, with Rayon and Indicatif, but without the hash calculation and it works correctly, no memory misbehavior.

That makes me think that I'm doing something wrong with memory management in my code, but I can't see anything obvious.

Can you trim down your program to the smallest possible runnable version that still exhibits the problem? That would be of great help in diagnosing and fixing the issue. — user4815162342
well, that wouldn't surprise me that indicatif have memory leak, the code is not very good. — Stargateur
Does calling bar.reset_eta() after bar.inc(1) solve the memory leak? I would guess the cause is indicatif collecting progress metrics the whole time (see here). — L. Riemer
@L.Riemer, no, it didn't change anything. Thanks for pointing it, though. I was looking into the source of Indicatif and got suspicious of that feature too. — Denis Falqueto
Just for the record, I've switched from indicatif to pbr and my program behaves the same way. So there's something wrong in my code, I just can't see what... — Denis Falqueto

Denis Falqueto Denis Falqueto · Accepted Answer · 2020-11-08T16:28:15

I found a solution, but I'm still not sure why it solves the original problem.

What I did to solve it is to transfer the progress code to the first spawn closure. Look at lines 6 and 19 below:

let (tx, rx) = mpsc::channel();

let original_hash = String::from(original_hash);

rayon::spawn(move || {
    let mut bar = ProgressBar::new(max_iteracoes as u64);
    let mut i = 0;
    let mut iter = SenhaIterator::from(initial_pwd);
    while i < max_iteracoes {
        let pwd = iter.next().unwrap();

        let clone_tx = tx.clone();
        rayon::spawn(move || {
            let hash = calcula_hash(&pwd);
            clone_tx.send((pwd, hash)).unwrap();
        });

        i += 1;
        bar.inc();
    }
    bar.finish();
});

let mut latest_pwd = None;

while let Ok((pwd, hash)) = rx.recv() {
    latest_pwd = Some(pwd);
    if hash == original_hash {
        return latest_pwd.map_or(PasswordOutcome::PasswordNotFound(None), |s| {
            PasswordOutcome::PasswordFound(s)
        })
    }
}
PasswordOutcome::PasswordNotFound(latest_pwd)

That first spawn closure has the role of fetching the next password to try and pass it over to a worker task, which calculates the corresponding hash and sends the pair (password, hash) to the main thread. The main thread will wait for the pairs from the rx channel and compare with the expected hash.

What is still missing from me is why tracking the progress on the outer thread leaks memory. I couldn't identify what is really leaking. But it's working now and I'm happy with the result.

Memory leak with Rayon and Indicatif in Rust

1 Answers