ProperyStream::next() panics

[zeenix]

In GitLab by @MaxVerevkin on Nov 21, 2021, 06:14

Minimal example:

use futures::stream::StreamExt;

#[zbus::dbus_proxy(interface = "org.bluez.Device1", default_service = "org.bluez")]
trait Device1 {
    #[dbus_proxy(property)]
    fn connected(&self) -> zbus::Result<bool>;
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let conn = zbus::Connection::system().await?;
    let proxy = Device1Proxy::builder(&conn)
        .path("/org/bluez/hci0/dev_00_18_09_92_1B_BA")?
        .build()
        .await?;
    let mut stream = proxy.receive_connected_changed().await;
    while let Some(event) = stream.next().await {
        dbg!(event.name());
        dbg!(event.get().await);
    }
    Ok(())
}

Produces this:

$ RUST_BACKTRACE=1 target/debug/zbt
[src/main.rs:18] event.name() = "Connected"
[src/main.rs:19] event.get().await = Ok(
    true,
)
thread 'main' panicked at 'internal error: entered unreachable code: cannot poll a completed `EventListener` future', /home/maxv/.cargo/registry/src/github.com-1ecc6299db9ec823/event-listener-2.5.1/src/lib.rs:665:21
stack backtrace:
   0: rust_begin_unwind
             at /rustc/0727994435c75fdedd3e9d226cf434089b0ab585/library/std/src/panicking.rs:498:5
   1: core::panicking::panic_fmt
             at /rustc/0727994435c75fdedd3e9d226cf434089b0ab585/library/core/src/panicking.rs:106:14
   2: <event_listener::EventListener as core::future::future::Future>::poll
             at /home/maxv/.cargo/registry/src/github.com-1ecc6299db9ec823/event-listener-2.5.1/src/lib.rs:665:21
   3: <zbus::proxy::PropertyStream<T> as futures_core::stream::Stream>::poll_next
             at /home/maxv/.cargo/git/checkouts/zbus-e0109d69954ffaa9/3a6f647/zbus/src/proxy.rs:244:16
   4: futures_util::stream::stream::StreamExt::poll_next_unpin
             at /home/maxv/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-util-0.3.17/src/stream/stream/mod.rs:1474:9
   5: <futures_util::stream::stream::next::Next<St> as core::future::future::Future>::poll
             at /home/maxv/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-util-0.3.17/src/stream/stream/next.rs:32:9
   6: zbt::main::{{closure}}
             at ./src/main.rs:17:29
   7: <core::future::from_generator::GenFuture<T> as core::future::future::Future>::poll
             at /rustc/0727994435c75fdedd3e9d226cf434089b0ab585/library/core/src/future/mod.rs:80:19
   8: tokio::park::thread::CachedParkThread::block_on::{{closure}}
             at /home/maxv/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.14.0/src/park/thread.rs:263:54
   9: tokio::coop::with_budget::{{closure}}
             at /home/maxv/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.14.0/src/coop.rs:106:9
  10: std::thread::local::LocalKey<T>::try_with
             at /rustc/0727994435c75fdedd3e9d226cf434089b0ab585/library/std/src/thread/local.rs:399:16
  11: std::thread::local::LocalKey<T>::with
             at /rustc/0727994435c75fdedd3e9d226cf434089b0ab585/library/std/src/thread/local.rs:375:9
  12: tokio::coop::with_budget
             at /home/maxv/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.14.0/src/coop.rs:99:5
  13: tokio::coop::budget
             at /home/maxv/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.14.0/src/coop.rs:76:5
  14: tokio::park::thread::CachedParkThread::block_on
             at /home/maxv/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.14.0/src/park/thread.rs:263:31
  15: tokio::runtime::enter::Enter::block_on
             at /home/maxv/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.14.0/src/runtime/enter.rs:151:13
  16: tokio::runtime::thread_pool::ThreadPool::block_on
             at /home/maxv/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.14.0/src/runtime/thread_pool/mod.rs:77:9
  17: tokio::runtime::Runtime::block_on
             at /home/maxv/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.14.0/src/runtime/mod.rs:463:43
  18: zbt::main
             at ./src/main.rs:22:5
  19: core::ops::function::FnOnce::call_once
             at /rustc/0727994435c75fdedd3e9d226cf434089b0ab585/library/core/src/ops/function.rs:227:5
note: Some details are omitted, run with `RUST_BACKTRACE=full` for a verbose backtrace.

[zeenix]

In GitLab by @MaxVerevkin on Nov 21, 2021, 08:58

Looks like this was introduced in 06d2d983

[zeenix]

In GitLab by @MaxVerevkin on Nov 21, 2021, 10:50

m.event should be recreated after https://gitlab.freedesktop.org/dbus/zbus/-/blob/main/zbus/src/proxy.rs#L244

The stream must either lock the mutex and get the event:

diff --git a/zbus/src/proxy.rs b/zbus/src/proxy.rs
index e8cd19b..e7a30bd 100644
--- a/zbus/src/proxy.rs
+++ b/zbus/src/proxy.rs
@@ -243,6 +243,15 @@ where
         };
         ready!(Pin::new(&mut m.event).poll(cx));

+        m.event = properties
+            .values
+            .lock()
+            .unwrap()
+            .get(m.name)
+            .unwrap()
+            .event
+            .listen();
+
         let prop_changed = PropertyChanged {
             name: m.name,
             properties,

or hold Arc<Event>. Which is better?

[zeenix]

@MaxVerevkin thanks so much for reporting and even figuring out the solution. :thumbsup:

Which is better?

The former (locking) is good and sort of what we had before 06d2d983. Only now the lock will be only held for the cloning. Maybe it'd be good idea to switch to RwLock here.

Since you've a test case for the code, if you decide to provide an MR for this, please do add the testcase (just modify the existing test for property stream) too so we don't break this in the future. :)

[zeenix]

mentioned in commit f8b63eb35283ac0c0486f16a16fec5d8599a8a92