aws: Fix race condition when unlocking

It would be possible that there is no cancellable yet when unlock() is
called, then a new future is executed and it wouldn't have any
information that it is not supposed to run at all.

To solve this remember if unlock() was called and reset this in
unlock_stop().

Also implement actual unlocking in s3hlssink.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/-/merge_requests/1602>

net/aws/src/s3hlssink/imp.rs

@@ -8,7 +8,6 @@
 //
 // SPDX-License-Identifier: MPL-2.0
 
-use futures::future;
 use once_cell::sync::Lazy;
 use std::io::Write;
 use std::str::FromStr;
@@ -90,7 +89,7 @@ pub struct S3HlsSink {
     settings: Mutex<Settings>,
     state: Mutex<State>,
     hlssink: gst::Element,
-    canceller: Mutex<Option<future::AbortHandle>>,
+    canceller: Mutex<s3utils::Canceller>,
 }
 
 static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
@@ -459,7 +458,7 @@ impl ObjectSubclass for S3HlsSink {
             settings: Mutex::new(Settings::default()),
             state: Mutex::new(State::Stopped),
             hlssink,
-            canceller: Mutex::new(None),
+            canceller: Mutex::new(s3utils::Canceller::default()),
         }
     }
 }
@@ -803,10 +802,19 @@ impl ElementImpl for S3HlsSink {
         PAD_TEMPLATES.as_ref()
     }
 
+    #[allow(clippy::single_match)]
     fn change_state(
         &self,
         transition: gst::StateChange,
     ) -> Result<gst::StateChangeSuccess, gst::StateChangeError> {
+        match transition {
+            gst::StateChange::PausedToReady => {
+                let mut canceller = self.canceller.lock().unwrap();
+                canceller.abort();
+            }
+            _ => (),
+        }
+
         let ret = self.parent_change_state(transition)?;
         /*
          * The settings lock must not be taken before the parent state change.
@@ -850,6 +858,11 @@ impl ElementImpl for S3HlsSink {
                 }
             }
 
+            gst::StateChange::PausedToReady => {
+                let mut canceller = self.canceller.lock().unwrap();
+                *canceller = s3utils::Canceller::None;
+            }
+
             gst::StateChange::ReadyToNull => {
                 drop(settings);
                 /*

net/aws/src/s3sink/multipartsink.rs

@@ -26,7 +26,6 @@ use aws_sdk_s3::{
     Client,
 };
 
-use futures::future;
 use once_cell::sync::Lazy;
 use std::collections::HashMap;
 use std::convert::From;
@@ -186,8 +185,8 @@ pub struct S3Sink {
     url: Mutex<Option<GstS3Url>>,
     settings: Mutex<Settings>,
     state: Mutex<State>,
-    canceller: Mutex<Option<future::AbortHandle>>,
-    abort_multipart_canceller: Mutex<Option<future::AbortHandle>>,
+    canceller: Mutex<s3utils::Canceller>,
+    abort_multipart_canceller: Mutex<s3utils::Canceller>,
 }
 
 static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
@@ -618,19 +617,6 @@ impl S3Sink {
         Ok(())
     }
 
-    fn cancel(&self) {
-        let mut canceller = self.canceller.lock().unwrap();
-        let mut abort_canceller = self.abort_multipart_canceller.lock().unwrap();
-
-        if let Some(c) = abort_canceller.take() {
-            c.abort()
-        };
-
-        if let Some(c) = canceller.take() {
-            c.abort()
-        };
-    }
-
     fn set_uri(self: &S3Sink, url_str: Option<&str>) -> Result<(), glib::Error> {
         let state = self.state.lock().unwrap();
 
@@ -1103,8 +1089,18 @@ impl BaseSinkImpl for S3Sink {
     }
 
     fn unlock(&self) -> Result<(), gst::ErrorMessage> {
-        self.cancel();
+        let mut canceller = self.canceller.lock().unwrap();
+        let mut abort_canceller = self.abort_multipart_canceller.lock().unwrap();
+        canceller.abort();
+        abort_canceller.abort();
+        Ok(())
+    }
 
+    fn unlock_stop(&self) -> Result<(), gst::ErrorMessage> {
+        let mut canceller = self.canceller.lock().unwrap();
+        let mut abort_canceller = self.abort_multipart_canceller.lock().unwrap();
+        *canceller = s3utils::Canceller::None;
+        *abort_canceller = s3utils::Canceller::None;
         Ok(())
     }
 

net/aws/src/s3sink/putobjectsink.rs

@@ -21,7 +21,6 @@ use aws_sdk_s3::{
     Client,
 };
 
-use futures::future;
 use once_cell::sync::Lazy;
 use std::collections::HashMap;
 use std::convert::From;
@@ -158,7 +157,7 @@ pub struct S3PutObjectSink {
     url: Mutex<Option<GstS3Url>>,
     settings: Mutex<Settings>,
     state: Mutex<State>,
-    canceller: Mutex<Option<future::AbortHandle>>,
+    canceller: Mutex<s3utils::Canceller>,
 }
 
 static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
@@ -328,14 +327,6 @@ impl S3PutObjectSink {
         Ok(())
     }
 
-    fn cancel(&self) {
-        let mut canceller = self.canceller.lock().unwrap();
-
-        if let Some(c) = canceller.take() {
-            c.abort()
-        };
-    }
-
     fn set_uri(self: &S3PutObjectSink, url_str: Option<&str>) -> Result<(), glib::Error> {
         let state = self.state.lock().unwrap();
 
@@ -756,8 +747,14 @@ impl BaseSinkImpl for S3PutObjectSink {
     }
 
     fn unlock(&self) -> Result<(), gst::ErrorMessage> {
-        self.cancel();
+        let mut canceller = self.canceller.lock().unwrap();
+        canceller.abort();
+        Ok(())
+    }
 
+    fn unlock_stop(&self) -> Result<(), gst::ErrorMessage> {
+        let mut canceller = self.canceller.lock().unwrap();
+        *canceller = s3utils::Canceller::None;
         Ok(())
     }
 

net/aws/src/s3src/imp.rs

@@ -7,7 +7,6 @@
 // SPDX-License-Identifier: MPL-2.0
 
 use bytes::Bytes;
-use futures::future;
 use once_cell::sync::Lazy;
 use std::sync::Mutex;
 use std::time::Duration;
@@ -77,7 +76,7 @@ impl Default for Settings {
 pub struct S3Src {
     settings: Mutex<Settings>,
     state: Mutex<StreamingState>,
-    canceller: Mutex<Option<future::AbortHandle>>,
+    canceller: Mutex<s3utils::Canceller>,
 }
 
 static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
@@ -89,14 +88,6 @@ static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
 });
 
 impl S3Src {
-    fn cancel(&self) {
-        let mut canceller = self.canceller.lock().unwrap();
-
-        if let Some(c) = canceller.take() {
-            c.abort()
-        };
-    }
-
     fn connect(self: &S3Src, url: &GstS3Url) -> Result<Client, gst::ErrorMessage> {
         let settings = self.settings.lock().unwrap();
         let timeout_config = s3utils::timeout_config(settings.request_timeout);
@@ -521,9 +512,6 @@ impl BaseSrcImpl for S3Src {
     }
 
     fn stop(&self) -> Result<(), gst::ErrorMessage> {
-        // First, stop any asynchronous tasks if we're running, as they will have the state lock
-        self.cancel();
-
         let mut state = self.state.lock().unwrap();
 
         if let StreamingState::Stopped = *state {
@@ -587,7 +575,14 @@ impl BaseSrcImpl for S3Src {
     }
 
     fn unlock(&self) -> Result<(), gst::ErrorMessage> {
-        self.cancel();
+        let mut canceller = self.canceller.lock().unwrap();
+        canceller.abort();
+        Ok(())
+    }
+
+    fn unlock_stop(&self) -> Result<(), gst::ErrorMessage> {
+        let mut canceller = self.canceller.lock().unwrap();
+        *canceller = s3utils::Canceller::None;
         Ok(())
     }
 }

net/aws/src/s3utils.rs

@@ -51,21 +51,38 @@ impl<E: ProvideErrorMetadata + std::error::Error> fmt::Display for WaitError<E>
     }
 }
 
-pub fn wait<F, T, E>(
-    canceller: &Mutex<Option<future::AbortHandle>>,
-    future: F,
-) -> Result<T, WaitError<E>>
+#[derive(Default)]
+pub enum Canceller {
+    #[default]
+    None,
+    Handle(future::AbortHandle),
+    Cancelled,
+}
+
+impl Canceller {
+    pub fn abort(&mut self) {
+        if let Canceller::Handle(ref canceller) = *self {
+            canceller.abort();
+        }
+
+        *self = Canceller::Cancelled;
+    }
+}
+
+pub fn wait<F, T, E>(canceller_mutex: &Mutex<Canceller>, future: F) -> Result<T, WaitError<E>>
 where
     F: Send + Future<Output = Result<T, E>>,
     F::Output: Send,
     T: Send,
     E: Send,
 {
-    let mut canceller_guard = canceller.lock().unwrap();
+    let mut canceller = canceller_mutex.lock().unwrap();
+    if matches!(*canceller, Canceller::Cancelled) {
+        return Err(WaitError::Cancelled);
+    }
     let (abort_handle, abort_registration) = future::AbortHandle::new_pair();
-
-    canceller_guard.replace(abort_handle);
-    drop(canceller_guard);
+    *canceller = Canceller::Handle(abort_handle);
+    drop(canceller);
 
     let abortable_future = future::Abortable::new(future, abort_registration);
 
@@ -86,17 +103,21 @@ where
     };
 
     /* Clear out the canceller */
-    canceller_guard = canceller.lock().unwrap();
-    *canceller_guard = None;
+    let mut canceller = canceller_mutex.lock().unwrap();
+    if matches!(*canceller, Canceller::Cancelled) {
+        return Err(WaitError::Cancelled);
+    }
+    *canceller = Canceller::None;
+    drop(canceller);
 
     res
 }
 
 pub fn wait_stream(
-    canceller: &Mutex<Option<future::AbortHandle>>,
+    canceller_mutex: &Mutex<Canceller>,
     stream: &mut ByteStream,
 ) -> Result<Bytes, WaitError<ByteStreamError>> {
-    wait(canceller, async move {
+    wait(canceller_mutex, async move {
         let mut collect = BytesMut::new();
 
         // Loop over the stream and collect till we're done
@@ -116,7 +137,7 @@ pub fn timeout_config(request_timeout: Duration) -> TimeoutConfig {
 }
 
 pub fn wait_config(
-    canceller: &Mutex<Option<future::AbortHandle>>,
+    canceller_mutex: &Mutex<Canceller>,
     region: Region,
     timeout_config: TimeoutConfig,
     credentials: Option<Credentials>,
@@ -136,11 +157,13 @@ pub fn wait_config(
             .load(),
     };
 
-    let mut canceller_guard = canceller.lock().unwrap();
+    let mut canceller = canceller_mutex.lock().unwrap();
+    if matches!(*canceller, Canceller::Cancelled) {
+        return Err(WaitError::Cancelled);
+    }
     let (abort_handle, abort_registration) = future::AbortHandle::new_pair();
-
-    canceller_guard.replace(abort_handle);
-    drop(canceller_guard);
+    *canceller = Canceller::Handle(abort_handle);
+    drop(canceller);
 
     let abortable_future = future::Abortable::new(config_future, abort_registration);
 
@@ -157,8 +180,12 @@ pub fn wait_config(
     };
 
     /* Clear out the canceller */
-    canceller_guard = canceller.lock().unwrap();
-    *canceller_guard = None;
+    let mut canceller = canceller_mutex.lock().unwrap();
+    if matches!(*canceller, Canceller::Cancelled) {
+        return Err(WaitError::Cancelled);
+    }
+    *canceller = Canceller::None;
+    drop(canceller);
 
     res
 }