Can zbus methods call be async? How does it work? #277

Closed zeenix closed 1 year ago

In GitLab by @kannarfr on Jun 10, 2022, 16:59

I'm experiencing bugs using zbus + pulsar-rs (can be a pulsar-rs issue).

The notify zbus method is async, as pulsar (message broker) send is async:

impl Manager {
    ...
    async fn do_notify(&mut self, topic_name: &str, message: &str) -> Result<()> {
        let topic = format!(
            "{}/{}/{}",
            self.config.pulsar.tenant, self.config.pulsar.namespace, topic_name
        );
        let producer = if let Some(producer) = self.producers.get_mut(&topic) {
            producer
        } else {
            let producer = self.client.producer(&topic).await?;
            self.producers.entry(topic).or_insert(producer)
        };

        producer.send(message).await?;

        Ok(())
    }
}

#[dbus_interface(name = "com.company.tako.Manager")]
impl Manager {
    async fn notify(&mut self, topic_name: &str, message: &str) -> bool {
        match self.do_notify(topic_name, message).await {
            Ok(()) => {
                info!("Sucessfully notified");
                true
            }
            Err(e) => {
                error!(?e, "Couldn't notify");
                false
            }
        }
    }
}

I'm investigating connection error with pulsar-rs but can't reproduce it easily. So I'm questioning myself about the bad usage of zbus I possibly have done.

As async method_call are not documented in zbus documentation. But as rustc authorizes it I'd like to know if I'm doing bad things or not.

(The notify method can be called by different process at the same

If it's possible, how zbus works this way? Should I take care of limitations?

As async method_call are not documented in zbus documentation.

Excuse me? :)

I'm investigating connection error with pulsar-rs but can't reproduce it easily.

If you can tell what problem exactly you're facing, I can make some educated guess about the cause.

In GitLab by @kannarfr on Jun 10, 2022, 18:38

Sorry, didn't read enough my bad.

Well, I'll be glad to get ur guess,

pulsar-rs connection-pool management seems flaky. When a connection doesn't ping it seems reused by producer/consumers and the application gets stuck. But I can't reproduce it. So I'm currently running 40 instances of random message producing to a test cluster and trying stuff like shutdown cluster node, kill TCP connection, etc, but not reproduced yet. Still investigating.

No worries.

The most common reason for getting hangs is when you've stream that you don't read from. In the code you showed, there are no stream but maybe elsehwere in your code there are some? By stream, I mean one of the zbus streams/iterators. This issue is documented here:

Connection keeps an internal queue of incoming message. The maximum capacity of this queue is configurable through the set_max_queued method. The default size is 64. When the queue is full, no more messages can be received until room is created for more. This is why it’s important to ensure that all crate::MessageStream and crate::blocking::MessageIterator instances are continuously polled and iterated on, respectively.

also, another thing is that while a notify method call is in progress, your Manager instance is locked. This is because you're taking a &mut self and there can't be multiple mutable refs to the same resource in Rust. So I'd check that out too and ensure notify doesn't take long.

In GitLab by @kannarfr on Jun 10, 2022, 19:03

notify can be retried as long as the pulsar cluster is experiencing issue, so maybe the zbus usage here is globally not a good idea, or i should add a timeout.

In GitLab by @kannarfr on Jun 10, 2022, 19:03

Thanks a lot for your help.

np, let us know if you find a (possible) bug in zbus and reopen. I'll close this for now.

dbus2 / zbus-old

Can zbus methods call be async? How does it work? #277