diff options
author | Sebastian Dröge <sebastian@centricular.com> | 2020-12-29 18:28:19 +0300 |
---|---|---|
committer | Sebastian Dröge <sebastian@centricular.com> | 2020-12-29 18:28:19 +0300 |
commit | 640d8ef9044de8cd8eea9316b382060d13ae0397 (patch) | |
tree | 7a7f7d618dc3fc50dd1bc70e3afdf8d3d34d123c /net | |
parent | 005c62425b67af8c6396fc941c13aaf798a4578b (diff) |
rusoto/aws_transcriber: Don't hold mutex across await points
This mutex is actually only ever used from a single thread, so use
AtomicRefCell instead. It provides the guarantees of a mutex but panics
instead of blocking.
Diffstat (limited to 'net')
-rw-r--r-- | net/rusoto/Cargo.toml | 1 | ||||
-rw-r--r-- | net/rusoto/src/aws_transcriber/imp.rs | 12 |
2 files changed, 8 insertions, 5 deletions
diff --git a/net/rusoto/Cargo.toml b/net/rusoto/Cargo.toml index a9d5bf401..a961e5a84 100644 --- a/net/rusoto/Cargo.toml +++ b/net/rusoto/Cargo.toml @@ -30,6 +30,7 @@ once_cell = "1.0" serde = "1" serde_derive = "1" serde_json = "1" +atomic_refcell = "0.1" [lib] name = "gstrusoto" diff --git a/net/rusoto/src/aws_transcriber/imp.rs b/net/rusoto/src/aws_transcriber/imp.rs index f03a70c89..36724a2a4 100644 --- a/net/rusoto/src/aws_transcriber/imp.rs +++ b/net/rusoto/src/aws_transcriber/imp.rs @@ -43,6 +43,8 @@ use std::pin::Pin; use std::sync::Mutex; use std::time::Duration; +use atomic_refcell::AtomicRefCell; + use super::packet::*; use serde_derive::Deserialize; @@ -200,14 +202,14 @@ impl Default for State { } } -type WsSink = Pin<Box<dyn Sink<Message, Error = WsError> + Send>>; +type WsSink = Pin<Box<dyn Sink<Message, Error = WsError> + Send + Sync>>; pub struct Transcriber { srcpad: gst::Pad, sinkpad: gst::Pad, settings: Mutex<Settings>, state: Mutex<State>, - ws_sink: Mutex<Option<WsSink>>, + ws_sink: AtomicRefCell<Option<WsSink>>, } fn build_packet(payload: &[u8]) -> Vec<u8> { @@ -801,7 +803,7 @@ impl Transcriber { tokio::time::delay_for(Duration::from_nanos(delay.nseconds().unwrap())).await; } - if let Some(ws_sink) = self.ws_sink.lock().unwrap().as_mut() { + if let Some(ws_sink) = self.ws_sink.borrow_mut().as_mut() { if let Some(buffer) = buffer { let data = buffer.map_readable().unwrap(); for chunk in data.chunks(8192) { @@ -916,7 +918,7 @@ impl Transcriber { let (ws_sink, mut ws_stream) = ws.split(); - *self.ws_sink.lock().unwrap() = Some(Box::pin(ws_sink)); + *self.ws_sink.borrow_mut() = Some(Box::pin(ws_sink)); let element_weak = element.downgrade(); let future = async move { @@ -1047,7 +1049,7 @@ impl ObjectSubclass for Transcriber { sinkpad, settings, state: Mutex::new(State::default()), - ws_sink: Mutex::new(None), + ws_sink: AtomicRefCell::new(None), } } |