`bound` option on `spsc_queue.rs` seems to have no effect

[MeirShpilraien]

Hey,

I am using channels on my rust code which behind the scene uses stream::Packet that uses sync::mpsc::spsc_queue.

I notice that my application reaches peaks on the queue size and the memory usage is not dropped after those peaks even though I see that the bound option is set to 128 elements:

rust/library/std/src/sync/mpsc/stream.rs

Line 70 in fe5b13d

spsc::Queue::with_additions(

.

I looked at the spsc_queue.rs code and I notice that the cached_nodes is not increased when adding an element to the cache:

rust/library/std/src/sync/mpsc/spsc_queue.rs

Line 180 in fe5b13d

self.consumer.cached_nodes.store(cached_nodes, Ordering::Relaxed);

I tried locally the following change and it fixed the problem:

- self.consumer.cached_nodes.store(cached_nodes, Ordering::Relaxed);
+ self.consumer.cached_nodes.store(cached_nodes + 1, Ordering::Relaxed);

I am not sure if this is a right fix or whether my analysis is correct (maybe I got it all wrong?). Please let me know what you think or if there is anything else I might be missing?

Thanks.

[MeirShpilraien]

Any inputs here?

[Enselic]

Triage: Hi. Please provide a minimal reproducer that demonstrates the problem. Thanks!

[bardiharborow]

This reproduces the issue on 1.66.1:

use std::alloc::{GlobalAlloc, Layout, System};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::mpsc;

struct Counting;

static ALLOCATED: AtomicUsize = AtomicUsize::new(0);

unsafe impl GlobalAlloc for Counting {
    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
        let p = System.alloc(layout);
        if !p.is_null() {
            ALLOCATED.fetch_add(layout.size(), Ordering::Relaxed);
        }
        p
    }
    unsafe fn dealloc(&self, p: *mut u8, layout: Layout) {
        System.dealloc(p, layout);
        ALLOCATED.fetch_sub(layout.size(), Ordering::Relaxed);
    }
}

#[global_allocator]
static GLOBAL: Counting = Counting;

const MESSAGES: usize = 100_000;
const ROUNDS: usize = 3;

fn allocated() -> usize {
    ALLOCATED.load(Ordering::Relaxed)
}

fn main() {
    // One sender, one receiver: this is the "stream" flavour backed by
    // spsc_queue::Queue::with_additions(128, ...).
    let (tx, rx) = mpsc::channel::<[u8; 1024]>();

    let baseline = allocated();
    println!("baseline:                 {:>12} bytes", baseline);

    for round in 1..=ROUNDS {
        // Burst: fill the channel far past the 128-node cache bound.
        for _ in 0..MESSAGES {
            tx.send([0u8; 1024]).unwrap();
        }
        let peak = allocated() - baseline;

        // Drain it completely.
        let mut received = 0;
        while rx.try_recv().is_ok() {
            received += 1;
        }
        assert_eq!(received, MESSAGES);
        let after_drain = allocated() - baseline;

        println!(
            "round {}: peak {:>12} bytes, after drain {:>12} bytes ({:.1}% retained)",
            round,
            peak,
            after_drain,
            100.0 * after_drain as f64 / peak as f64
        );
    }

    drop(tx);
    drop(rx);
    println!("after dropping channel:   {:>12} bytes", allocated() - baseline);
}

This issue was fixed by #93563 in Rust 1.67.

@rustbot label -S-needs-repro +S-has-mcve +A-concurrency +S-candidate-for-closing