From 41663f15efea57bfc6007670be168a88ca45ea7d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Sebastian=20Dr=C3=B6ge?= <sebastian@centricular.com>
Date: Tue, 28 Jan 2020 23:39:36 +0200
Subject: [PATCH] bus: Don't use the bus sync handler in the bus Stream to
 notify about messages being available

This is racy and can cause the consumer of the messages to never be
woken up anymore:

1. Waker is stored because no message on the bus
2. Sync handler is called, waker is woken up
3. Bus is polled again and no message is on it (yet),
   new waker is registered
4. Bus stores the message from 2. in its queue (after
   the sync handler has returned BusSyncReply::Pass)
5. No new message ever appears on the bus because all
   this happened for the very last message

Fixes https://gitlab.freedesktop.org/gstreamer/gstreamer-rs/issues/235
---
 gstreamer/src/bus.rs | 35 ++++++++++-------------------------
 1 file changed, 10 insertions(+), 25 deletions(-)

diff --git a/gstreamer/src/bus.rs b/gstreamer/src/bus.rs
index 14bd54db7..02b790332 100644
--- a/gstreamer/src/bus.rs
+++ b/gstreamer/src/bus.rs
@@ -6,8 +6,9 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
+use futures_channel::mpsc;
 use futures_core::stream::Stream;
-use futures_core::task::{Context, Poll, Waker};
+use futures_core::task::{Context, Poll};
 use glib;
 use glib::source::{Continue, Priority, SourceId};
 use glib::translate::*;
@@ -18,7 +19,6 @@ use std::cell::RefCell;
 use std::mem::transmute;
 use std::pin::Pin;
 use std::ptr;
-use std::sync::{Arc, Mutex};
 
 use Bus;
 use BusSyncReply;
@@ -234,26 +234,20 @@ impl<'a> Iterator for Iter<'a> {
 }
 
 #[derive(Debug)]
-pub struct BusStream(Bus, Arc<Mutex<Option<Waker>>>);
+pub struct BusStream(Bus, mpsc::UnboundedReceiver<Message>);
 
 impl BusStream {
     pub fn new(bus: &Bus) -> Self {
         skip_assert_initialized!();
-        let waker = Arc::new(Mutex::new(None));
-        let waker_clone = Arc::clone(&waker);
 
-        bus.set_sync_handler(move |_, _| {
-            let mut waker = waker_clone.lock().unwrap();
-            if let Some(waker) = waker.take() {
-                // FIXME: Force type...
-                let waker: Waker = waker;
-                waker.wake();
-            }
+        let (sender, receiver) = mpsc::unbounded();
 
-            BusSyncReply::Pass
+        bus.set_sync_handler(move |_, msg| {
+            let _ = sender.unbounded_send(msg.to_owned());
+            BusSyncReply::Drop
         });
 
-        BusStream(bus.clone(), waker)
+        BusStream(bus.clone(), receiver)
     }
 }
 
@@ -266,17 +260,8 @@ impl Drop for BusStream {
 impl Stream for BusStream {
     type Item = Message;
 
-    fn poll_next(self: Pin<&mut Self>, ctx: &mut Context) -> Poll<Option<Self::Item>> {
-        let BusStream(ref bus, ref waker) = *self;
-
-        let mut waker = waker.lock().unwrap();
-        let msg = bus.pop();
-        if let Some(msg) = msg {
-            Poll::Ready(Some(msg))
-        } else {
-            *waker = Some(ctx.waker().clone());
-            Poll::Pending
-        }
+    fn poll_next(mut self: Pin<&mut Self>, ctx: &mut Context) -> Poll<Option<Self::Item>> {
+        Pin::new(&mut self.1).poll_next(ctx)
     }
 }