By gop parallelism

Cargo.toml

@@ -23,7 +23,7 @@ default-run = "rav1e"
 
 [features]
 unstable = []
-channel-api = ["crossbeam"]
+channel-api = []
 decode_test = ["aom-sys"]
 decode_test_dav1d = ["dav1d-sys"]
 binaries = [
@@ -76,7 +76,7 @@ ivf = { version = "0.1", path = "ivf/", optional = true }
 v_frame = { version = "0.2.1", path = "v_frame/" }
 av-metrics = { version = "0.6.2", optional = true, default-features = false }
 rayon = "1.0"
-crossbeam = { version = "0.8", optional = true }
+crossbeam = "0.8"
 toml = { version = "0.5", optional = true }
 arrayvec = "0.7"
 thiserror = "1.0"

src/api/channel/data.rs

@@ -33,24 +33,48 @@ impl RcDataSender {
   pub(crate) fn new(limit: u64, sender: Sender<RcData>) -> RcDataSender {
     Self { sender, limit, count: 0 }
   }
+  fn split_try_send(
+    &mut self, data: RcData,
+  ) -> Result<(), TrySendError<RcData>> {
+    match data {
+      RcData::Summary(_) => self.sender.try_send(data)?,
+      RcData::Frame(data) => {
+        for f in data.chunks(crate::rate::TWOPASS_PACKET_SZ) {
+          self.sender.try_send(RcData::Frame(f.to_vec().into_boxed_slice()))?
+        }
+      }
+    }
+    Ok(())
+  }
   pub fn try_send(
     &mut self, data: RcData,
   ) -> Result<(), TrySendError<RcData>> {
     if self.limit <= self.count {
-      Err(TrySendError::Full(data))
+      Err(TrySendError::Disconnected(data))
     } else {
-      let r = self.sender.try_send(data);
+      let r = self.split_try_send(data);
       if r.is_ok() {
         self.count += 1;
       }
       r
     }
   }
+  fn split_send(&mut self, data: RcData) -> Result<(), SendError<RcData>> {
+    match data {
+      RcData::Summary(_) => self.sender.send(data)?,
+      RcData::Frame(data) => {
+        for f in data.chunks(crate::rate::TWOPASS_PACKET_SZ) {
+          self.sender.send(RcData::Frame(f.to_vec().into_boxed_slice()))?
+        }
+      }
+    }
+    Ok(())
+  }
   pub fn send(&mut self, data: RcData) -> Result<(), SendError<RcData>> {
     if self.limit <= self.count {
       Err(SendError(data))
     } else {
-      let r = self.sender.send(data);
+      let r = self.split_send(data);
       if r.is_ok() {
         self.count += 1;
       }
@@ -63,6 +87,9 @@ impl RcDataSender {
   pub fn is_empty(&self) -> bool {
     self.sender.is_empty()
   }
+  pub fn capacity(&self) -> Option<usize> {
+    self.sender.capacity()
+  }
 
   // TODO: proxy more methods
 }
@@ -117,7 +144,7 @@ impl<T: Pixel> FrameSender<T> {
     &mut self, frame: F,
   ) -> Result<(), TrySendError<FrameInput<T>>> {
     if self.limit <= self.count {
-      Err(TrySendError::Full(frame.into()))
+      Err(TrySendError::Disconnected(frame.into()))
     } else {
       let r = self.sender.try_send(frame.into());
       if r.is_ok() {

src/api/channel/mod.rs

@@ -50,7 +50,7 @@ impl Config {
       return Err(InvalidConfig::RateControlConfigurationMismatch);
     }
 
-    let (v, _) = self.new_channel_internal()?;
+    let (v, _) = self.new_channel_internal(true)?;
 
     Ok(v)
   }
@@ -70,7 +70,7 @@ impl Config {
     if !rc.emit_pass_data {
       return Err(InvalidConfig::RateControlConfigurationMismatch);
     }
-    let (v, (_, r)) = self.new_channel_internal()?;
+    let (v, (_, r)) = self.new_channel_internal(true)?;
 
     Ok((v, r.unwrap()))
   }
@@ -88,7 +88,7 @@ impl Config {
       return Err(InvalidConfig::RateControlConfigurationMismatch);
     }
 
-    let (v, (s, _)) = self.new_channel_internal()?;
+    let (v, (s, _)) = self.new_channel_internal(true)?;
 
     Ok((v, s.unwrap()))
   }
@@ -109,14 +109,15 @@ impl Config {
       return Err(InvalidConfig::RateControlConfigurationMismatch);
     }
 
-    let (v, (s, r)) = self.new_channel_internal()?;
+    let (v, (s, r)) = self.new_channel_internal(true)?;
 
     Ok((v, (s.unwrap(), r.unwrap())))
   }
 }
 
 trait RcFirstPass {
-  fn send_pass_data(&mut self, rc_state: &mut RCState);
+  fn store_pass_data(&mut self, rc_state: &mut RCState);
+  fn send_pass_data(&mut self);
   fn send_pass_summary(&mut self, rc_state: &mut RCState);
 }
 
@@ -126,29 +127,46 @@ trait RcSecondPass {
   ) -> Result<(), ()>;
 }
 
-impl RcFirstPass for Sender<RcData> {
-  fn send_pass_data(&mut self, rc_state: &mut RCState) {
+struct PassDataSender {
+  sender: Sender<RcData>,
+  data: Vec<u8>,
+}
+
+impl RcFirstPass for PassDataSender {
+  fn store_pass_data(&mut self, rc_state: &mut RCState) {
     if let Some(data) = rc_state.emit_frame_data() {
-      let data = data.to_vec().into_boxed_slice();
-      self.send(RcData::Frame(data)).unwrap();
+      self.data.extend(data.iter());
     } else {
       unreachable!(
         "The encoder received more frames than its internal
               limit allows"
       );
     }
   }
+
+  fn send_pass_data(&mut self) {
+    let data = self.data.to_vec().into_boxed_slice();
+    self.data.clear();
+    self.sender.send(RcData::Frame(data)).unwrap();
+  }
   fn send_pass_summary(&mut self, rc_state: &mut RCState) {
     let data = rc_state.emit_summary();
     let data = data.to_vec().into_boxed_slice();
-    self.send(RcData::Summary(data)).unwrap();
+    self.sender.send(RcData::Summary(data)).unwrap();
   }
 }
 
-impl RcFirstPass for Option<Sender<RcData>> {
-  fn send_pass_data(&mut self, rc_state: &mut RCState) {
+impl RcFirstPass for Option<PassDataSender> {
+  fn store_pass_data(&mut self, rc_state: &mut RCState) {
+    match self.as_mut() {
+      Some(s) => s.store_pass_data(rc_state),
+      None => {}
+    }
+  }
+
+  fn send_pass_data(&mut self) {
     match self.as_mut() {
-      Some(s) => s.send_pass_data(rc_state),
+      Some(s) => s.send_pass_data(),
       None => {}
     }
   }
@@ -164,16 +182,18 @@ impl RcSecondPass for Receiver<RcData> {
   fn feed_pass_data<T: Pixel>(
     &mut self, inner: &mut ContextInner<T>,
   ) -> Result<(), ()> {
-    while inner.rc_state.twopass_in_frames_needed() > 0
-      && !inner.done_processing()
-    {
+    if inner.done_processing() {
+      return Ok(());
+    }
+
+    while inner.rc_state.twopass_in_frames_needed() > 0 {
       if let Ok(RcData::Frame(data)) = self.recv() {
         inner
           .rc_state
           .parse_frame_data_packet(data.as_ref())
           .unwrap_or_else(|_| todo!("Error reporting"));
       } else {
-        todo!("Error reporting");
+        todo!("No data reporting");
       }
     }
 
@@ -193,8 +213,8 @@ impl RcSecondPass for Option<Receiver<RcData>> {
 }
 
 impl Config {
-  fn new_channel_internal<T: Pixel>(
-    &self,
+  pub(crate) fn new_channel_internal<T: Pixel>(
+    &self, serial_api_compat: bool,
   ) -> Result<
     (VideoDataChannel<T>, (Option<RcDataSender>, Option<RcDataReceiver>)),
     InvalidConfig,
@@ -205,14 +225,18 @@ impl Config {
     // TODO: make it user-settable
     let input_len = self.enc.rdo_lookahead_frames as usize * 2;
 
-    let (send_frame, receive_frame) = bounded(input_len);
+    let (send_frame, receive_frame) =
+      if serial_api_compat { unbounded() } else { bounded(input_len) };
     let (send_packet, receive_packet) = unbounded();
 
     let rc = &self.rate_control;
 
     let (mut send_rc_pass1, rc_data_receiver) = if rc.emit_pass_data {
       let (send_rc_pass1, receive_rc_pass1) = unbounded();
-      (Some(send_rc_pass1), Some(RcDataReceiver(receive_rc_pass1)))
+      (
+        Some(PassDataSender { sender: send_rc_pass1, data: Vec::new() }),
+        Some(RcDataReceiver(receive_rc_pass1)),
+      )
     } else {
       (None, None)
     };
@@ -221,15 +245,21 @@ impl Config {
       .summary
       .is_some()
     {
-      let (frame_limit, pass_limit) =
+      // the pass data packets are now lumped together to match the packets emitted.
+      let (frame_limit, _pass_limit) =
         rc.summary.as_ref().map(|s| (s.ntus as u64, s.total as u64)).unwrap();
 
       inner.limit = Some(frame_limit);
 
-      let (send_rc_pass2, receive_rc_pass2) = unbounded();
+      let (send_rc_pass2, receive_rc_pass2) = if serial_api_compat {
+        // TODO: derive it from the Encoder configuration
+        bounded(input_len)
+      } else {
+        unbounded()
+      };
 
       (
-        Some(RcDataSender::new(pass_limit, send_rc_pass2)),
+        Some(RcDataSender::new(frame_limit, send_rc_pass2)),
         Some(receive_rc_pass2),
         frame_limit,
       )
@@ -255,17 +285,21 @@ impl Config {
           // already.
           //
           // this call should return either Ok or Err(Encoded)
-          let has_pass_data = match inner.receive_packet() {
+          let pass_data_ready = match inner.receive_packet() {
             Ok(p) => {
               send_packet.send(p).unwrap();
+              send_rc_pass1.store_pass_data(&mut inner.rc_state);
               true
             }
-            Err(EncoderStatus::Encoded) => true,
+            Err(EncoderStatus::Encoded) => {
+              send_rc_pass1.store_pass_data(&mut inner.rc_state);
+              false
+            }
             Err(EncoderStatus::NotReady) => todo!("Error reporting"),
             _ => unreachable!(),
           };
-          if has_pass_data {
-            send_rc_pass1.send_pass_data(&mut inner.rc_state);
+          if pass_data_ready {
+            send_rc_pass1.send_pass_data();
           }
         }
 
@@ -279,20 +313,23 @@ impl Config {
       loop {
         receive_rc_pass2.feed_pass_data(&mut inner).unwrap();
         let r = inner.receive_packet();
-        let has_pass_data = match r {
+        let pass_data_ready = match r {
           Ok(p) => {
-            // warn!("Sending out {}", p.input_frameno);
             send_packet.send(p).unwrap();
+            send_rc_pass1.store_pass_data(&mut inner.rc_state);
             true
           }
           Err(EncoderStatus::LimitReached) => break,
-          Err(EncoderStatus::Encoded) => true,
+          Err(EncoderStatus::Encoded) => {
+            send_rc_pass1.store_pass_data(&mut inner.rc_state);
+            false
+          }
           Err(EncoderStatus::NotReady) => todo!("Error reporting"),
           _ => unreachable!(),
         };
 
-        if has_pass_data {
-          send_rc_pass1.send_pass_data(&mut inner.rc_state);
+        if pass_data_ready {
+          send_rc_pass1.send_pass_data();
         }
       }
 

src/api/config/mod.rs

@@ -11,7 +11,7 @@ use thiserror::Error;
 
 use std::sync::Arc;
 
-use crate::api::{ChromaSampling, Context, ContextInner, PixelRange};
+use crate::api::{ChromaSampling, ContextInner, PixelRange};
 use crate::rayon::{ThreadPool, ThreadPoolBuilder};
 use crate::util::Pixel;
 
@@ -217,8 +217,8 @@ impl Config {
     let mut inner = ContextInner::new(&config);
 
     if self.rate_control.emit_pass_data {
-      let params = inner.rc_state.get_twopass_out_params(&inner, 0);
-      inner.rc_state.init_first_pass(params.pass1_log_base_q);
+      let pass1_log_base_q = inner.rc_state.get_twopass_out_params(&inner, 0);
+      inner.rc_state.init_first_pass(pass1_log_base_q);
     }
 
     if let Some(ref s) = self.rate_control.summary {
@@ -242,30 +242,6 @@ impl Config {
       None
     }
   }
-
-  /// Creates a [`Context`] with this configuration.
-  ///
-  /// # Examples
-  ///
-  /// ```
-  /// use rav1e::prelude::*;
-  ///
-  /// # fn main() -> Result<(), InvalidConfig> {
-  /// let cfg = Config::default();
-  /// let ctx: Context<u8> = cfg.new_context()?;
-  /// # Ok(())
-  /// # }
-  /// ```
-  ///
-  /// [`Context`]: struct.Context.html
-  pub fn new_context<T: Pixel>(&self) -> Result<Context<T>, InvalidConfig> {
-    let inner = self.new_inner()?;
-    let config = *inner.config;
-    let pool = self.new_thread_pool();
-
-    Ok(Context { is_flushing: false, inner, pool, config })
-  }
-
   /// Validates the configuration.
   pub fn validate(&self) -> Result<(), InvalidConfig> {
     use InvalidConfig::*;