dav1ddec: Move state to Mutex from Arc

This is needed because we may now access state concurrently with and
outside of the streaming thread during a LATENCY query.

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

video/dav1d/Cargo.toml

@@ -9,7 +9,6 @@ license = "MIT OR Apache-2.0"
 description = "Dav1d Plugin"
 
 [dependencies]
-atomic_refcell = "0.1"
 dav1d = "0.8"
 gst = { package = "gstreamer", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs" }
 gst-base = { package = "gstreamer-base", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs" }

video/dav1d/src/dav1ddec/imp.rs

@@ -16,9 +16,8 @@ use gst_video::subclass::prelude::*;
 
 use once_cell::sync::Lazy;
 
-use atomic_refcell::{AtomicRefCell, AtomicRefMut};
 use std::i32;
-use std::sync::Mutex;
+use std::sync::{Mutex, MutexGuard};
 
 const DEFAULT_N_THREADS: u32 = 0;
 const DEFAULT_MAX_FRAME_DELAY: i64 = -1;
@@ -48,7 +47,7 @@ impl Default for Settings {
 
 #[derive(Default)]
 pub struct Dav1dDec {
-    state: AtomicRefCell<Option<State>>,
+    state: Mutex<Option<State>>,
     settings: Mutex<Settings>,
 }
 
@@ -128,10 +127,10 @@ impl Dav1dDec {
     fn handle_resolution_change<'s>(
         &'s self,
         element: &super::Dav1dDec,
-        mut state_guard: AtomicRefMut<'s, Option<State>>,
+        mut state_guard: MutexGuard<'s, Option<State>>,
         pic: &dav1d::Picture,
-    ) -> Result<AtomicRefMut<'s, Option<State>>, gst::FlowError> {
+    ) -> Result<MutexGuard<'s, Option<State>>, gst::FlowError> {
-        let state = state_guard.as_mut().unwrap();
+        let state = state_guard.as_ref().unwrap();
 
         let format = self.gst_video_format_from_dav1d_picture(element, pic);
         if format == gst_video::VideoFormat::Unknown {
@@ -168,24 +167,23 @@ impl Dav1dDec {
         element.negotiate(output_state)?;
         let out_state = element.output_state().unwrap();
 
-        state_guard = self.state.borrow_mut();
+        state_guard = self.state.lock().unwrap();
         let state = state_guard.as_mut().unwrap();
         state.output_info = Some(out_state.info());
 
         Ok(state_guard)
     }
 
-    fn flush_decoder(&self, element: &super::Dav1dDec, state_guard: &mut Option<State>) {
+    fn flush_decoder(&self, element: &super::Dav1dDec, state: &mut State) {
         gst::info!(CAT, obj: element, "Flushing decoder");
 
-        let state = state_guard.as_mut().unwrap();
         state.decoder.flush();
     }
 
     fn send_data(
         &self,
         element: &super::Dav1dDec,
-        state_guard: &mut AtomicRefMut<Option<State>>,
+        state_guard: &mut MutexGuard<Option<State>>,
         input_buffer: gst::Buffer,
         frame: gst_video::VideoCodecFrame,
     ) -> Result<std::ops::ControlFlow<(), ()>, gst::FlowError> {
@@ -236,7 +234,7 @@ impl Dav1dDec {
     fn send_pending_data(
         &self,
         element: &super::Dav1dDec,
-        state_guard: &mut AtomicRefMut<Option<State>>,
+        state_guard: &mut MutexGuard<Option<State>>,
     ) -> Result<std::ops::ControlFlow<(), ()>, gst::FlowError> {
         gst::trace!(CAT, obj: element, "Sending pending data to decoder");
 
@@ -267,7 +265,7 @@ impl Dav1dDec {
     fn decoded_picture_as_buffer(
         &self,
         element: &super::Dav1dDec,
-        state_guard: &mut AtomicRefMut<Option<State>>,
+        state_guard: &mut MutexGuard<Option<State>>,
         pic: &dav1d::Picture,
         output_state: gst_video::VideoCodecState<gst_video::video_codec_state::Readable>,
     ) -> Result<gst::Buffer, gst::FlowError> {
@@ -362,9 +360,9 @@ impl Dav1dDec {
     fn handle_picture<'s>(
         &'s self,
         element: &super::Dav1dDec,
-        mut state_guard: AtomicRefMut<'s, Option<State>>,
+        mut state_guard: MutexGuard<'s, Option<State>>,
         pic: &dav1d::Picture,
-    ) -> Result<AtomicRefMut<'s, Option<State>>, gst::FlowError> {
+    ) -> Result<MutexGuard<'s, Option<State>>, gst::FlowError> {
         gst::trace!(CAT, obj: element, "Handling picture {}", pic.offset());
 
         state_guard = self.handle_resolution_change(element, state_guard, pic)?;
@@ -380,14 +378,18 @@ impl Dav1dDec {
             frame.set_output_buffer(output_buffer);
             drop(state_guard);
             element.finish_frame(frame)?;
-            Ok(self.state.borrow_mut())
+            Ok(self.state.lock().unwrap())
         } else {
             gst::warning!(CAT, obj: element, "No frame found for offset {}", offset);
             Ok(state_guard)
         }
     }
 
-    fn drop_decoded_pictures(&self, element: &super::Dav1dDec, state_guard: &mut Option<State>) {
+    fn drop_decoded_pictures(
+        &self,
+        element: &super::Dav1dDec,
+        state_guard: &mut MutexGuard<Option<State>>,
+    ) {
         while let Ok(Some(pic)) = self.pending_pictures(element, state_guard) {
             gst::debug!(CAT, obj: element, "Dropping picture {}", pic.offset());
             drop(pic);
@@ -397,7 +399,7 @@ impl Dav1dDec {
     fn pending_pictures(
         &self,
         element: &super::Dav1dDec,
-        state_guard: &mut Option<State>,
+        state_guard: &mut MutexGuard<Option<State>>,
     ) -> Result<Option<dav1d::Picture>, gst::FlowError> {
         gst::trace!(CAT, obj: element, "Retrieving pending picture");
 
@@ -434,8 +436,8 @@ impl Dav1dDec {
     fn forward_pending_pictures<'s>(
         &'s self,
         element: &super::Dav1dDec,
-        mut state_guard: AtomicRefMut<'s, Option<State>>,
+        mut state_guard: MutexGuard<'s, Option<State>>,
-    ) -> Result<AtomicRefMut<Option<State>>, gst::FlowError> {
+    ) -> Result<MutexGuard<Option<State>>, gst::FlowError> {
         while let Some(pic) = self.pending_pictures(element, &mut state_guard)? {
             state_guard = self.handle_picture(element, state_guard, &pic)?;
         }
@@ -614,7 +616,7 @@ impl ElementImpl for Dav1dDec {
 impl VideoDecoderImpl for Dav1dDec {
     fn src_query(&self, element: &Self::Type, query: &mut gst::QueryRef) -> bool {
         if let gst::QueryViewMut::Latency(q) = query.view_mut() {
-            let state_guard = self.state.borrow();
+            let state_guard = self.state.lock().unwrap();
             let max_frame_delay = {
                 let settings = self.settings.lock().unwrap();
                 settings.max_frame_delay
@@ -669,7 +671,7 @@ impl VideoDecoderImpl for Dav1dDec {
 
     fn stop(&self, element: &Self::Type) -> Result<(), gst::ErrorMessage> {
         {
-            let mut state_guard = self.state.borrow_mut();
+            let mut state_guard = self.state.lock().unwrap();
             *state_guard = None;
         }
 
@@ -681,7 +683,7 @@ impl VideoDecoderImpl for Dav1dDec {
         element: &Self::Type,
         input_state: &gst_video::VideoCodecState<'static, gst_video::video_codec_state::Readable>,
     ) -> Result<(), gst::LoggableError> {
-        let mut state_guard = self.state.borrow_mut();
+        let mut state_guard = self.state.lock().unwrap();
         let settings = self.settings.lock().unwrap();
         let mut decoder_settings = dav1d::Settings::new();
         let max_frame_delay: u32;
@@ -738,7 +740,7 @@ impl VideoDecoderImpl for Dav1dDec {
             .expect("frame without input buffer");
 
         {
-            let mut state_guard = self.state.borrow_mut();
+            let mut state_guard = self.state.lock().unwrap();
             state_guard = self.forward_pending_pictures(element, state_guard)?;
             if self.send_data(element, &mut state_guard, input_buffer, frame)?
                 == std::ops::ControlFlow::Continue(())
@@ -762,9 +764,10 @@ impl VideoDecoderImpl for Dav1dDec {
         gst::info!(CAT, obj: element, "Flushing");
 
         {
-            let mut state_guard = self.state.borrow_mut();
+            let mut state_guard = self.state.lock().unwrap();
             if state_guard.is_some() {
-                self.flush_decoder(element, &mut state_guard);
+                let state = state_guard.as_mut().unwrap();
+                self.flush_decoder(element, state);
                 self.drop_decoded_pictures(element, &mut state_guard);
             }
         }
@@ -776,9 +779,10 @@ impl VideoDecoderImpl for Dav1dDec {
         gst::info!(CAT, obj: element, "Draining");
 
         {
-            let mut state_guard = self.state.borrow_mut();
+            let mut state_guard = self.state.lock().unwrap();
             if state_guard.is_some() {
-                self.flush_decoder(element, &mut state_guard);
+                let state = state_guard.as_mut().unwrap();
+                self.flush_decoder(element, state);
                 let _state_guard = self.forward_pending_pictures(element, state_guard)?;
             }
         }
@@ -790,9 +794,10 @@ impl VideoDecoderImpl for Dav1dDec {
         gst::info!(CAT, obj: element, "Finishing");
 
         {
-            let mut state_guard = self.state.borrow_mut();
+            let mut state_guard = self.state.lock().unwrap();
             if state_guard.is_some() {
-                self.flush_decoder(element, &mut state_guard);
+                let state = state_guard.as_mut().unwrap();
+                self.flush_decoder(element, state);
                 let _state_guard = self.forward_pending_pictures(element, state_guard)?;
             }
         }
@@ -806,7 +811,7 @@ impl VideoDecoderImpl for Dav1dDec {
         query: &mut gst::query::Allocation,
     ) -> Result<(), gst::LoggableError> {
         {
-            let mut state_guard = self.state.borrow_mut();
+            let mut state_guard = self.state.lock().unwrap();
             let state = state_guard.as_mut().unwrap();
             state.video_meta_supported = query
                 .find_allocation_meta::<gst_video::VideoMeta>()