Fix a few segfaults in the PulseAudio driver upon shutdown (#22)

A few segfaults were discovered in the PulseAudio driver during
application shutdown. While it's hard to say for certain whether these
changes resolve the discovered segfaults, they seem to greatly reduce
their incidence rate, at a minimum. The fixes involve:

* Implementing `Drop` for `StreamReadFuture`, `PulseFuture`, and
`InitializingFuture`, all of of which pass a `Waker` instance into a
PulseAudio callback. If `libpulse` attempts to run the callback after
the futures/tasks have been dropped, this results in a segfault.
* This segfault was only observed with `StreamReadFuture`, however it
was fixed for all three future types since they all follow the same
pattern.
* Unsetting the read callback on the audio stream when closing a
`StreamReadNotifier`. The exact cause isn't clear, but there appears to
be some condition where `libpulse` attempting to call the read callback
during application shutdown would result in a segfault (that is distinct
from the `Waker` issue described above)

src/snd/pulse/context/async_polyfill.rs

@@ -120,6 +120,17 @@ impl<S: Unpin, T: InitializingEntity<S> + Unpin> Future for InitializingFuture<S
     }
 }
 
+impl<S, T: InitializingEntity<S>> Drop for InitializingFuture<S, T> {
+    fn drop(&mut self) {
+        if let Some(entity) = &mut self.0 {
+            //In case the future is cancelled, clear out the Waker to prevent
+            //the stream read callback from potentially accidentally waking a
+            //nonexistent task
+            entity.wake_on_state_change(None)
+        }
+    }
+}
+
 impl InitializingEntity<PulseContextState> for PulseContext {
     fn get_initialization_state(&self) -> InitializingState<PulseContextState> {
         match self.get_state() {
@@ -179,6 +190,18 @@ impl StreamReadNotifier {
     pub fn await_data<'a>(&'a self) -> StreamReadFuture<'a> {
         StreamReadFuture::new(&self.0)
     }
+
+    /// Closes the notifier, and unsets the read callback on the stream. This
+    /// should be done instead of letting the notifier go out of scope
+    /// implicitly, as leaving the read callback set can lead to intermittent
+    /// segfaults.
+    ///
+    /// This is done as a dedicated function to avoid holding a mutable
+    /// reference to the stream during the notifier's lifetime when it is not
+    /// needed, which would also prevent reading data from the stream.
+    pub fn close(self, stream: &mut PulseStream) {
+        stream.set_read_callback(None);
+    }
 }
 
 //IMPL: StreamReadFuture *******************************************************
@@ -209,6 +232,15 @@ impl<'a> Future for StreamReadFuture<'a> {
     }
 }
 
+impl<'a> Drop for StreamReadFuture<'a> {
+    fn drop(&mut self) {
+        //In case the future is cancelled, clear out the Waker to prevent the
+        //stream read callback from potentially accidentally waking a
+        //nonexistent task
+        *self.waker.lock().unwrap() = None;
+    }
+}
+
 //IMPL: PulseFuture ************************************************************
 impl<T> PulseFutureInner<T> {
     /// Creates a new shared state object seeded with the given initial value.
@@ -315,3 +347,12 @@ impl<T: Default> Future for PulseFuture<T> {
         }
     }
 }
+
+impl<T> Drop for PulseFuture<T> {
+    fn drop(&mut self) {
+        //In case the future is cancelled, clear out the Waker to prevent the
+        //stream read callback from potentially accidentally waking a
+        //nonexistent task
+        self.0.lock().unwrap().waker = None
+    }
+}

src/snd/pulse/context/mod.rs

@@ -211,6 +211,7 @@ impl PulseContextWrapper {
             }
 
             debug!("Stream handler shutting down");
+            notify.close(&mut stream);
 
             //Tear down the record stream
             if let Err(e) = stream.disconnect() {