Fix race condition with seek lock by enforcing an ack

backends/gstreamer/player.rs

@@ -13,7 +13,9 @@ use render::GStreamerRender;
 use servo_media_player::context::PlayerGLContext;
 use servo_media_player::frame::FrameRenderer;
 use servo_media_player::metadata::Metadata;
-use servo_media_player::{PlaybackState, Player, PlayerError, PlayerEvent, StreamType};
+use servo_media_player::{
+    PlaybackState, Player, PlayerError, PlayerEvent, SeekLock, SeekLockMsg, StreamType,
+};
 use servo_media_streams::registry::{get_stream, MediaStreamId};
 use servo_media_traits::Muteable;
 use source::{register_servo_src, ServoSrc};
@@ -298,21 +300,26 @@ macro_rules! player(
 );
 
 struct SeekChannel {
-    sender: IpcSender<bool>,
-    recv: IpcReceiver<bool>,
+    sender: SeekLock,
+    recv: IpcReceiver<SeekLockMsg>,
 }
 
 impl SeekChannel {
     fn new() -> Self {
-        let (sender, recv) = channel::<bool>().expect("Couldn't create IPC channel");
-        Self { sender, recv }
+        let (sender, recv) = channel::<SeekLockMsg>().expect("Couldn't create IPC channel");
+        Self {
+            sender: SeekLock {
+                lock_channel: sender,
+            },
+            recv,
+        }
     }
 
-    fn sender(&self) -> IpcSender<bool> {
+    fn sender(&self) -> SeekLock {
         self.sender.clone()
     }
 
-    fn await(&self) -> bool {
+    fn await(&self) -> SeekLockMsg {
         self.recv.recv().unwrap()
     }
 }
@@ -656,21 +663,25 @@ impl GStreamerPlayer {
                                     notify!(observer__, PlayerEvent::EnoughData);
                                 })
                                 .seek_data(move |_, offset| {
-                                    let ret = if servosrc_.set_seek_offset(offset) {
+                                    let (ret, ack_channel) = if servosrc_.set_seek_offset(offset) {
                                         notify!(
                                             observer___,
                                             PlayerEvent::SeekData(
                                                 offset,
                                                 seek_channel.lock().unwrap().sender()
                                             )
                                         );
-                                        let ret = seek_channel.lock().unwrap().await();
-                                        ret
+                                        let (ret, ack_channel) =
+                                            seek_channel.lock().unwrap().await();
+                                        (ret, Some(ack_channel))
                                     } else {
-                                        true
+                                        (true, None)
                                     };
 
                                     servosrc_.set_seek_done();
+                                    if let Some(ack_channel) = ack_channel {
+                                        ack_channel.send(()).unwrap();
+                                    }
                                     ret
                                 })
                                 .build(),

backends/gstreamer/render.rs

@@ -44,7 +44,7 @@ mod platform {
     pub struct RenderDummy();
     pub type Render = RenderDummy;
 
-    pub fn create_render(_: Box<PlayerGLContext>) -> Option<RenderDummy> {
+    pub fn create_render(_: Box<dyn PlayerGLContext>) -> Option<RenderDummy> {
         None
     }
 

backends/gstreamer/source.rs

@@ -55,10 +55,19 @@ mod imp {
         srcpad: gst::GhostPad,
         position: Mutex<Position>,
         seeking: AtomicBool,
+        size: Mutex<Option<i64>>,
     }
 
     impl ServoSrc {
         pub fn set_size(&self, size: i64) {
+            if self.seeking.load(Ordering::Relaxed) {
+                // We ignore set_size requests if we are seeking.
+                // The size value is temporarily stored so it
+                // is properly set once we are done seeking.
+                *self.size.lock().unwrap() = Some(size);
+                return;
+            }
+
             if self.appsrc.get_size() == -1 {
                 self.appsrc.set_size(size);
             }
@@ -86,6 +95,12 @@ mod imp {
         pub fn set_seek_done(&self) {
             self.seeking.store(false, Ordering::Relaxed);
 
+            if let Some(size) = self.size.lock().unwrap().take() {
+                if self.appsrc.get_size() == -1 {
+                    self.appsrc.set_size(size);
+                }
+            }
+
             let mut pos = self.position.lock().unwrap();
             pos.offset = pos.requested_offset;
             pos.requested_offset = 0;
@@ -267,6 +282,7 @@ mod imp {
                 srcpad: ghost_pad,
                 position: Mutex::new(Default::default()),
                 seeking: AtomicBool::new(false),
+                size: Mutex::new(None),
             }
         }
 

backends/gstreamer/webrtc.rs

@@ -310,7 +310,8 @@ impl GStreamerWebRtcController {
             let mut caps = gst::Caps::new_empty();
             let caps_mut = caps.get_mut().expect("Fresh caps should be uniquely owned");
             for format in media.formats() {
-                let pt = format.parse()
+                let pt = format
+                    .parse()
                     .expect("Gstreamer provided noninteger format");
                 caps_mut.append(
                     media

examples/player/app.rs

@@ -389,14 +389,13 @@ pub fn main_loop(mut app: App) -> Result<glutin::WindowedContext<glutin::Possibl
                         _ => (),
                     }
                 }
-                player::PlayerEvent::SeekData(offset, sender) => {
+                player::PlayerEvent::SeekData(offset, seek_lock) => {
                     input_eos = false;
-                    let ret = if let Ok(pos) = buf_reader.seek(SeekFrom::Start(offset)) {
+                    seek_lock.unlock(if let Ok(pos) = buf_reader.seek(SeekFrom::Start(offset)) {
                         offset == pos
                     } else {
                         false
-                    };
-                    sender.send(ret).unwrap();
+                    });
                 }
                 player::PlayerEvent::NeedData => {
                     if !input_eos {

examples/player/context.rs

@@ -10,6 +10,7 @@ pub struct PlayerContextGlutin {
     gl_api: GlApi,
 }
 
+#[allow(unused_variables)]
 impl PlayerContextGlutin {
     pub fn new(
         use_gl: bool,

examples/simple_player.rs

@@ -142,10 +142,10 @@ fn run_example(servo_media: Arc<ServoMedia>) {
                     }
                 }
             }
-            PlayerEvent::SeekData(p, sender) => {
+            PlayerEvent::SeekData(p, seek_lock) => {
                 println!("\nSeek requested to position {:?}", p);
                 seek_sender.send(p).unwrap();
-                sender.send(true).unwrap();
+                seek_lock.unlock(true);
             }
             PlayerEvent::SeekDone(p) => println!("\nSeeked to {:?}", p),
             PlayerEvent::NeedData => println!("\nNeedData"),

player/lib.rs

@@ -8,7 +8,7 @@ pub mod context;
 pub mod frame;
 pub mod metadata;
 
-use ipc_channel::ipc::IpcSender;
+use ipc_channel::ipc::{self, IpcSender};
 use servo_media_traits::Muteable;
 use std::ops::Range;
 use streams::registry::MediaStreamId;
@@ -40,6 +40,21 @@ pub enum PlayerError {
     SetStreamFailed,
 }
 
+pub type SeekLockMsg = (bool, IpcSender<()>);
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct SeekLock {
+    pub lock_channel: IpcSender<SeekLockMsg>,
+}
+
+impl SeekLock {
+    pub fn unlock(&self, result: bool) {
+        let (ack_sender, ack_recv) = ipc::channel::<()>().expect("Could not create IPC channel");
+        self.lock_channel.send((result, ack_sender)).unwrap();
+        ack_recv.recv().unwrap()
+    }
+}
+
 #[derive(Clone, Debug, Deserialize, Serialize)]
 pub enum PlayerEvent {
     EndOfStream,
@@ -54,9 +69,10 @@ pub enum PlayerEvent {
     PositionChanged(u64),
     /// The player needs the data to perform a seek to the given offset.
     /// The next push_data should get the buffers from the new offset.
-    /// The player will be blocked until the user sends, through the IPC sender,
+    /// The player will be blocked until the user unlocks it through
+    /// the given SeekLock instance.
     /// This event is only received for seekable stream types.
-    SeekData(u64, IpcSender<bool>),
+    SeekData(u64, SeekLock),
     /// The player has performed a seek to the given offset.
     SeekDone(u64),
     StateChanged(PlaybackState),

streams/lib.rs

@@ -22,5 +22,5 @@ pub trait MediaOutput: Send {
 #[derive(Clone, Copy, Debug, PartialEq)]
 pub enum MediaStreamType {
     Video,
-    Audio
+    Audio,
 }

webrtc/lib.rs

@@ -1,8 +1,8 @@
 extern crate boxfnonce;
 extern crate log;
 extern crate servo_media_streams;
-use servo_media_streams::MediaStreamType;
 use servo_media_streams::registry::MediaStreamId;
+use servo_media_streams::MediaStreamType;
 
 use std::fmt::Display;
 use std::str::FromStr;

webrtc/thread.rs

@@ -107,7 +107,10 @@ pub enum InternalEvent {
     UpdateIceConnectionState,
 }
 
-pub fn handle_rtc_event(controller: &mut dyn WebRtcControllerBackend, event: RtcThreadEvent) -> bool {
+pub fn handle_rtc_event(
+    controller: &mut dyn WebRtcControllerBackend,
+    event: RtcThreadEvent,
+) -> bool {
     let result = match event {
         RtcThreadEvent::ConfigureStun(server, policy) => controller.configure(&server, policy),
         RtcThreadEvent::SetRemoteDescription(desc, cb) => {