From dc9be3b60ed4644ed38aae9ce3faa22da0001a0b Mon Sep 17 00:00:00 2001
From: Albert Sjolund <alberts@axis.com>
Date: Wed, 26 Jun 2024 14:36:52 +0200
Subject: [PATCH] utils: streamproducer: wrap atomic operations

In order to support more platforms, especially MIPS,
add a small wrapper for necessary atomic(u64) instructions.
It exposes the necessary functions, and falls back to
mutex if needed.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gstreamer-rs/-/merge_requests/1505>
---
 gstreamer-utils/src/streamproducer.rs | 75 ++++++++++++++++++++++-----
 1 file changed, 63 insertions(+), 12 deletions(-)

diff --git a/gstreamer-utils/src/streamproducer.rs b/gstreamer-utils/src/streamproducer.rs
index ea4914a8a..e0a9d8142 100644
--- a/gstreamer-utils/src/streamproducer.rs
+++ b/gstreamer-utils/src/streamproducer.rs
@@ -8,6 +8,57 @@ use gst::{glib, prelude::*};
 use once_cell::sync::Lazy;
 use thiserror::Error;
 
+// Small wrapper around AtomicU64 and a Mutex, to allow it to run regular AtomicU64
+// operations where supported, and fallback to a mutex where it is not. The wrapper methods
+// are the ones that are needed, and not all are exposed.
+#[derive(Debug)]
+struct WrappedAtomicU64 {
+    #[cfg(not(target_has_atomic = "64"))]
+    atomic: Mutex<u64>,
+    #[cfg(target_has_atomic = "64")]
+    atomic: atomic::AtomicU64,
+}
+
+#[cfg(target_has_atomic = "64")]
+impl WrappedAtomicU64 {
+    fn new(value: u64) -> WrappedAtomicU64 {
+        WrappedAtomicU64 {
+            atomic: atomic::AtomicU64::new(value),
+        }
+    }
+    fn fetch_add(&self, value: u64, order: atomic::Ordering) -> u64 {
+        self.atomic.fetch_add(value, order)
+    }
+    fn store(&self, value: u64, order: atomic::Ordering) {
+        self.atomic.store(value, order);
+    }
+
+    fn load(&self, order: atomic::Ordering) -> u64 {
+        self.atomic.load(order)
+    }
+}
+
+#[cfg(not(target_has_atomic = "64"))]
+impl WrappedAtomicU64 {
+    fn new(value: u64) -> WrappedAtomicU64 {
+        WrappedAtomicU64 {
+            atomic: Mutex::new(value),
+        }
+    }
+    fn fetch_add(&self, value: u64, _order: atomic::Ordering) -> u64 {
+        let mut guard = self.atomic.lock().unwrap();
+        let old = *guard;
+        *guard += value;
+        old
+    }
+    fn store(&self, value: u64, _order: atomic::Ordering) {
+        *self.atomic.lock().unwrap() = value;
+    }
+    fn load(&self, _order: atomic::Ordering) -> u64 {
+        *self.atomic.lock().unwrap()
+    }
+}
+
 static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
     gst::DebugCategory::new(
         "utilsrs-stream-producer",
@@ -45,9 +96,9 @@ pub struct ConsumptionLink {
     consumer: gst_app::AppSrc,
     producer: Option<StreamProducer>,
     /// number of buffers dropped because `consumer` internal queue was full
-    dropped: Arc<atomic::AtomicU64>,
+    dropped: Arc<WrappedAtomicU64>,
     /// number of buffers pushed through `consumer`
-    pushed: Arc<atomic::AtomicU64>,
+    pushed: Arc<WrappedAtomicU64>,
     /// if buffers should not be pushed to the `consumer` right now
     discard: Arc<atomic::AtomicBool>,
 }
@@ -58,8 +109,8 @@ impl ConsumptionLink {
         ConsumptionLink {
             consumer,
             producer: None,
-            dropped: Arc::new(atomic::AtomicU64::new(0)),
-            pushed: Arc::new(atomic::AtomicU64::new(0)),
+            dropped: Arc::new(WrappedAtomicU64::new(0)),
+            pushed: Arc::new(WrappedAtomicU64::new(0)),
             discard: Arc::new(atomic::AtomicBool::new(false)),
         }
     }
@@ -157,8 +208,8 @@ impl StreamProducer {
         &self,
         consumer: &gst_app::AppSrc,
     ) -> Result<ConsumptionLink, AddConsumerError> {
-        let dropped = Arc::new(atomic::AtomicU64::new(0));
-        let pushed = Arc::new(atomic::AtomicU64::new(0));
+        let dropped = Arc::new(WrappedAtomicU64::new(0));
+        let pushed = Arc::new(WrappedAtomicU64::new(0));
         let discard = Arc::new(atomic::AtomicBool::new(false));
 
         self.add_consumer_internal(consumer, dropped.clone(), pushed.clone(), discard.clone())?;
@@ -175,8 +226,8 @@ impl StreamProducer {
     fn add_consumer_internal(
         &self,
         consumer: &gst_app::AppSrc,
-        dropped: Arc<atomic::AtomicU64>,
-        pushed: Arc<atomic::AtomicU64>,
+        dropped: Arc<WrappedAtomicU64>,
+        pushed: Arc<WrappedAtomicU64>,
         discard: Arc<atomic::AtomicBool>,
     ) -> Result<(), AddConsumerError> {
         let mut consumers = self.consumers.lock().unwrap();
@@ -622,9 +673,9 @@ struct StreamConsumer {
     /// streams.
     needs_keyframe: Arc<atomic::AtomicBool>,
     /// number of buffers dropped because `appsrc` internal queue was full
-    dropped: Arc<atomic::AtomicU64>,
+    dropped: Arc<WrappedAtomicU64>,
     /// number of buffers pushed through `appsrc`
-    pushed: Arc<atomic::AtomicU64>,
+    pushed: Arc<WrappedAtomicU64>,
     /// if buffers should not be pushed to the `appsrc` right now
     discard: Arc<atomic::AtomicBool>,
 }
@@ -634,8 +685,8 @@ impl StreamConsumer {
     fn new(
         appsrc: &gst_app::AppSrc,
         fku_probe_id: gst::PadProbeId,
-        dropped: Arc<atomic::AtomicU64>,
-        pushed: Arc<atomic::AtomicU64>,
+        dropped: Arc<WrappedAtomicU64>,
+        pushed: Arc<WrappedAtomicU64>,
         discard: Arc<atomic::AtomicBool>,
     ) -> Self {
         let needs_keyframe = Arc::new(atomic::AtomicBool::new(true));