From 9945b702b82670354474cb9b5101cc1f430da5b0 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Sebastian=20Dr=C3=B6ge?= <sebastian@centricular.com>
Date: Mon, 3 Jun 2024 16:01:23 +0300
Subject: [PATCH] reqwesthttpsrc: 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().

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/-/merge_requests/1602>
---
 net/reqwest/src/reqwesthttpsrc/imp.rs | 45 ++++++++++++++++++++++-----
 1 file changed, 37 insertions(+), 8 deletions(-)

diff --git a/net/reqwest/src/reqwesthttpsrc/imp.rs b/net/reqwest/src/reqwesthttpsrc/imp.rs
index bdeb6077..98d94e91 100644
--- a/net/reqwest/src/reqwesthttpsrc/imp.rs
+++ b/net/reqwest/src/reqwesthttpsrc/imp.rs
@@ -139,13 +139,31 @@ enum State {
     },
 }
 
-#[derive(Debug, Default)]
+#[derive(Default)]
+enum Canceller {
+    #[default]
+    None,
+    Handle(future::AbortHandle),
+    Cancelled,
+}
+
+impl Canceller {
+    fn abort(&mut self) {
+        if let Canceller::Handle(ref canceller) = *self {
+            canceller.abort();
+        }
+
+        *self = Canceller::Cancelled;
+    }
+}
+
+#[derive(Default)]
 pub struct ReqwestHttpSrc {
     client: Mutex<Option<ClientContext>>,
     external_client: Mutex<Option<ClientContext>>,
     settings: Mutex<Settings>,
     state: Mutex<State>,
-    canceller: Mutex<Option<future::AbortHandle>>,
+    canceller: Mutex<Canceller>,
 }
 
 static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
@@ -617,8 +635,11 @@ impl ReqwestHttpSrc {
         let timeout = self.settings.lock().unwrap().timeout;
 
         let mut canceller = self.canceller.lock().unwrap();
+        if matches!(*canceller, Canceller::Cancelled) {
+            return Err(None);
+        }
         let (abort_handle, abort_registration) = future::AbortHandle::new_pair();
-        canceller.replace(abort_handle);
+        *canceller = Canceller::Handle(abort_handle);
         drop(canceller);
 
         // Wrap in a timeout
@@ -652,7 +673,11 @@ impl ReqwestHttpSrc {
         };
 
         /* Clear out the canceller */
-        let _ = self.canceller.lock().unwrap().take();
+        let mut canceller = self.canceller.lock().unwrap();
+        if matches!(*canceller, Canceller::Cancelled) {
+            return Err(None);
+        }
+        *canceller = Canceller::None;
 
         res
     }
@@ -1024,10 +1049,14 @@ impl BaseSrcImpl for ReqwestHttpSrc {
     }
 
     fn unlock(&self) -> Result<(), gst::ErrorMessage> {
-        let canceller = self.canceller.lock().unwrap();
-        if let Some(ref canceller) = *canceller {
-            canceller.abort();
-        }
+        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 = Canceller::None;
         Ok(())
     }