Attempt to add with overflow: producer.rs line 234

[DataFrogman]

Describe the bug

I have run into an integer overflow error in producer.rs, it appears that the 'val' created on line 232 should be set to usize rather than determined by the compiler.

Reproduction steps

1. Create a queue with a max size of 25GB
2. Async create producers (max 16 at a time)
3. Push 50,000+ messages per producer
4. When queue length is greater than 300,000 eventually the error will occur

Expected behavior

Integers to be allocated of the maximum necessary size, potentially at usize maximum.

Additional context

No response

[Gsantomaggio]

Can you please provide the program to reproduce the issue?
GitRepo or Gist
Thank you

[DataFrogman]

fn producer_thread(rx: Receiver<Message>, environment: Environment)  -> Result<(), Box<dyn Error + Send + Sync>> {
    let runtime = tokio::runtime::Runtime::new().unwrap();
    let _guard = runtime.enter();
    let producer = match runtime.block_on(environment.producer().batch_delay(Duration::from_secs(1)).batch_size(100).build("test")) {
        Ok(x) => Arc::new(x),
        Err(e) => {
            println!("Producer creation error");
            return Err(e.to_string().into())
        }
    };
    loop {
        let mess = rx.recv()?;
        let temp = producer.clone();
        tokio::spawn(async move {temp.send_with_confirm(mess).await});
    }
}

Here is a function that will cause it. I have an external loop that is using the mpsc channel to feed messages to the producer to send into rabbitmq. Unfortunately after a couple of seconds the threads will start crashing with the error The application panicked (crashed). Message: attempt to add with overflow Location: /home/ubuntu/.cargo/git/checkouts/rabbitmq-stream-rust-client-09c996cc06a727bd/f335c59/src/producer.rs:239

[michaelklishin]

After discussing this with @Gsantomaggio, I arrive at the following conclusion: we can make the counter type explicit but cloning the producer for every message you receive over a MCSP channel is not how this library was meant to be used.

If you need all those those MC (multiple consumers) of the MCSP channel to publish, each consuming thread must use its own stream client connection. Cloning a connection, directly or indirectly, will not go well. Cloning and sharing [between threads] resources that use sockets does not sound safe at all.

On-the-wire interleaving of protocol frames is the most obvious outcome of such sharing. Which is exactly why all the AMQP 0-9-1 clients for RabbitMQ explicitly prohibit such sharing of channels for concurrent publishing.

[DataFrogman]

How would you recommend publishing messages that come from multiple async functions then? Opening a new producer for every function is not viable as it quickly depletes server resources and Tokio explicitly warns against making a parament loop in one of their threads.

[michaelklishin]

Aggregate them to a single channel where just one producer will do all the publishing. This client is pretty efficient and you likely can get to a good million of messages per second with just one stream, just like you can with Java and .NET stream clients.

If you need more than that, use multiple streams and multiple connections for as many threads.
Each thread will consume a subset of messages for publishing. What would be the best Rust concurrency primitive for that, I don't have an opinion about.

Perhaps Crossbeam would offer something more appropriate than the standard library?

[DataFrogman]

Okay, thank you very much!

[the-mikedavis]

I think the line that overflows is this one: val + 1 could overflow if val == usize::MAX. We could switch the check to val == self.capacity - 1 to be safe since self.capacity > 0.

[michaelklishin]

@Gsantomaggio I have enabled Discussions for this repository. This problem is not very well understood yet, and that's exactly how we use Discussions in every other repo in this org.

[michaelklishin]

Here is a PR to verify the original hypothesis #187.

[wolf4ood]

Hi @DataFrogman

i think in this case there are a lot tasks in the runtime trying to send messages in batch . I would in case try to reduce the number of talks spawned. You can try to make the producer_thread an async function and use the api producer#send with callback for better performance. Meanwhile we can also our something on ours side to mitigate this

Thanks