From dd6c82b89fa01f14003f442fd8f67fe6f600105f Mon Sep 17 00:00:00 2001 From: ruccho Date: Wed, 28 Oct 2026 15:07:55 +0900 Subject: [PATCH 1/3] Reduce the CPU usage of the Android encoder by about 50 ms per second while recording Measured on a Galaxy S22 with the default settings, the CPU time added by recording fell from about 307 ms/s to about 255 ms/s. - The outputs of the video and audio encoders are drained on a dedicated thread each, and audio input is fed on a dedicated thread, instead of handing every dequeue to the blocking pool of the executor. Encoded data is copied directly from the direct buffers of MediaCodec rather than through Java arrays. - The Vulkan image, memory, view and framebuffer imported from each hardware buffer of the ImageWriter are cached and reused, instead of being created and destroyed for every frame. - The bindings generated by java_api! cache the class and the method and field IDs and call them without looking them up again. --- .../crates/java_api_macros/src/descriptor.rs | 19 ++ .../unienc/crates/java_api_macros/src/lib.rs | 87 +++++- .../crates/unienc_android_mc/src/audio/mod.rs | 295 +++++++++++------- .../crates/unienc_android_mc/src/common.rs | 200 +++++++----- .../crates/unienc_android_mc/src/error.rs | 12 + .../crates/unienc_android_mc/src/video/mod.rs | 17 +- .../src/vulkan/hardware_buffer.rs | 5 + .../src/vulkan/hardware_buffer_surface.rs | 69 +++- 8 files changed, 482 insertions(+), 222 deletions(-) diff --git a/InstantReplay.Externals/unienc/crates/java_api_macros/src/descriptor.rs b/InstantReplay.Externals/unienc/crates/java_api_macros/src/descriptor.rs index 2211a3eb..69a7f071 100644 --- a/InstantReplay.Externals/unienc/crates/java_api_macros/src/descriptor.rs +++ b/InstantReplay.Externals/unienc/crates/java_api_macros/src/descriptor.rs @@ -65,6 +65,25 @@ impl JType { } } + /// `jni::signature::ReturnType` of a method or field of this type. + pub fn return_signature(self) -> TokenStream { + let primitive = |name: TokenStream| { + quote!(::jni::signature::ReturnType::Primitive(::jni::signature::Primitive::#name)) + }; + match self { + JType::Void => primitive(quote!(Void)), + JType::Bool => primitive(quote!(Boolean)), + JType::Byte => primitive(quote!(Byte)), + JType::Char => primitive(quote!(Char)), + JType::Short => primitive(quote!(Short)), + JType::Int => primitive(quote!(Int)), + JType::Long => primitive(quote!(Long)), + JType::Float => primitive(quote!(Float)), + JType::Double => primitive(quote!(Double)), + JType::String | JType::Object => quote!(::jni::signature::ReturnType::Object), + } + } + /// Expression extracting this type out of `#value`, a `JValueOwned`. pub fn from_jvalue(self, value: TokenStream) -> TokenStream { match self { diff --git a/InstantReplay.Externals/unienc/crates/java_api_macros/src/lib.rs b/InstantReplay.Externals/unienc/crates/java_api_macros/src/lib.rs index 72dee7bd..6cca6a56 100644 --- a/InstantReplay.Externals/unienc/crates/java_api_macros/src/lib.rs +++ b/InstantReplay.Externals/unienc/crates/java_api_macros/src/lib.rs @@ -277,6 +277,7 @@ fn expand_class(declaration: &JavaApi, metadata: &Metadata, class: &ClassDecl) - ); } + let runtime = &declaration.runtime; let class_since = metadata.classes[&jni_name].since; let doc = format!( "Wrappers for `{}`.\n\nIntroduced in API level {class_since}.", @@ -300,6 +301,23 @@ fn expand_class(declaration: &JavaApi, metadata: &Metadata, class: &ClassDecl) - /// JNI name of the class, as accepted by `JNIEnv::find_class`. pub const CLASS: &'static str = #jni_name; + /// Global reference to the class, looked up on first use. It also keeps the class from + /// being unloaded, which keeps the member IDs cached by the wrappers valid. + pub fn class( + env: &mut ::jni::JNIEnv<'_>, + ) -> ::jni::errors::Result<&'static ::jni::objects::GlobalRef> { + static CLASS: ::std::sync::OnceLock<::jni::objects::GlobalRef> = + ::std::sync::OnceLock::new(); + if let Some(class) = CLASS.get() { + return Ok(class); + } + let __result = env.find_class(#jni_name); + let local = #runtime::checked(env, __result)?; + let global = env.new_global_ref(&local)?; + env.delete_local_ref(local)?; + Ok(CLASS.get_or_init(|| global)) + } + #members } } @@ -515,9 +533,30 @@ fn generate( let name = &member.name; let return_ty = return_type.return_type(); + let return_signature = return_type.return_signature(); + let class_ty = &class.name; + + // Expression yielding the member ID, which is looked up once and cached. Looking a member up by + // name and parsing its descriptor costs far more than the call itself, and the wrappers are + // called for every frame. + let cached_id = |id_ty: TokenStream, lookup: TokenStream| { + quote! {{ + static ID: ::std::sync::OnceLock<#id_ty> = ::std::sync::OnceLock::new(); + match ID.get() { + Some(id) => *id, + None => { + let class = #class_ty::class(env)?; + let __result = env.#lookup(class, #java_name, #descriptor); + let id = #runtime::checked(env, __result)?; + *ID.get_or_init(|| id) + } + } + }} + }; match member.kind { MemberKind::Method if is_constructor => { + let method_id = cached_id(quote!(::jni::objects::JMethodID), quote!(get_method_id)); quote! { #(#attrs)* #[doc = #doc] @@ -527,16 +566,43 @@ fn generate( #api_param #(#params),* ) -> ::jni::errors::Result<::jni::objects::JObject<'local>> { - let __result = env.new_object(#class_name, #descriptor, &[#(#args),*]); + let __id = #method_id; + let class = #class_ty::class(env)?; + // SAFETY: The ID was looked up with the descriptor the arguments are built from. + let __result = + unsafe { env.new_object_unchecked(class, __id, &[#(#args.as_jni()),*]) }; #check(env, __result) } } } MemberKind::Method => { + // SAFETY (of the generated calls): The ID was looked up with the descriptor that the + // argument values and the return type are derived from. let call = if member.is_static { - quote!(env.call_static_method(#class_name, #java_name, #descriptor, &[#(#args),*])) + let method_id = cached_id( + quote!(::jni::objects::JStaticMethodID), + quote!(get_static_method_id), + ); + quote! {{ + let __id = #method_id; + let class = #class_ty::class(env)?; + unsafe { + env.call_static_method_unchecked( + class, + __id, + #return_signature, + &[#(#args.as_jni()),*], + ) + } + }} } else { - quote!(env.call_method(this, #java_name, #descriptor, &[#(#args),*])) + let method_id = cached_id(quote!(::jni::objects::JMethodID), quote!(get_method_id)); + quote! {{ + let __id = #method_id; + unsafe { + env.call_method_unchecked(this, __id, #return_signature, &[#(#args.as_jni()),*]) + } + }} }; let extract = return_type.from_jvalue(quote!(__value)); quote! { @@ -558,10 +624,15 @@ fn generate( MemberKind::Field => { let setter = format_ident!("set_{}", name); let extract = return_type.from_jvalue(quote!(__value)); + let field_id = cached_id(quote!(::jni::objects::JFieldID), quote!(get_field_id)); let get = if member.is_static { + // Static fields are read once and cached by their callers, so their ID is not. quote!(env.get_static_field(#class_name, #java_name, #descriptor)) } else { - quote!(env.get_field(this, #java_name, #descriptor)) + quote! {{ + let __id = #field_id; + env.get_field_unchecked(this, __id, #return_signature) + }} }; let set = if member.is_static { // Static field writes are not generated: nothing in this crate needs them, and the @@ -580,7 +651,8 @@ fn generate( this: &::jni::objects::JObject<'_>, #value_arg: #value_ty, ) -> ::jni::errors::Result<()> { - let __result = env.set_field(this, #java_name, #descriptor, #value_jvalue); + let __id = #field_id; + let __result = env.set_field_unchecked(this, __id, #value_jvalue); #check(env, __result) } }) @@ -674,10 +746,7 @@ C java/lang/Object 1 generated.contains(r#""getKeys" , "()Ljava/util/Set;""#), "{generated}" ); - assert!( - generated.contains(r#"new_object ("android/media/MediaFormat" , "()V""#), - "{generated}" - ); + assert!(generated.contains(r#""" , "()V""#), "{generated}"); assert!(generated.contains(r#""flags" , "I""#), "{generated}"); // The declaration string itself must never reach a JNI call. assert!( diff --git a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/audio/mod.rs b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/audio/mod.rs index 7757e980..12a7e9a3 100644 --- a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/audio/mod.rs +++ b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/audio/mod.rs @@ -1,5 +1,6 @@ use jni::{JNIEnv, sys::jint}; -use std::time::Duration; +use std::marker::PhantomData; +use std::sync::{Arc, Mutex}; use unienc_common::{AudioSample, Encoder, EncoderInput, EncoderOutput}; use crate::error::{AndroidError, Result}; @@ -16,29 +17,39 @@ pub struct MediaCodecAudioEncoder { } pub struct MediaCodecAudioEncoderInput { - codec: MediaCodec, - sample_rate: u32, channels: u32, - last_timestamp: i64, - /// Running presentation position (in samples) of the next input chunk. Anchored to the first input - /// buffer's timestamp and advanced by the number of frames actually queued, so that a single large - /// (batched) input buffer split across multiple MediaCodec input buffers yields strictly increasing, - /// contiguous timestamps instead of repeating the same input timestamp. - position_in_samples: Option, /// Expected input timestamp (in samples) of the next push, i.e. the previous push's timestamp plus /// the number of frames it delivered. Used to detect discontinuities in the input timeline. next_input_position: Option, - runtime: R, + /// Hands PCM to the input thread. Dropping it lets the thread signal the end of the stream. + tx: tokio::sync::mpsc::Sender, + /// The error that stopped the input thread, if any. + error: Arc>>, + _runtime: PhantomData, } unsafe impl Send for MediaCodecAudioEncoderInput {} pub struct MediaCodecAudioEncoderOutput { - codec: MediaCodec, - end_of_stream: bool, - runtime: R, + drainer: OutputDrainer, + _runtime: PhantomData, +} + +/// PCM handed from `push` to the input thread. +struct AudioInputChunk { + /// Input timestamp (in samples) of the first frame of `bytes`. Only the first chunk's is used; + /// the thread derives the rest from the frames it has queued. + timestamp_in_samples: u64, + /// Signed 16-bit little-endian PCM, including any leading silence that fills a gap. + bytes: Vec, } +/// Chunks `push` may hand over before it waits for the input thread. +const INPUT_QUEUE_CAPACITY: usize = 2; + +/// Timed-out dequeues tolerated while waiting for the input buffer that carries the end of stream. +const EOS_DEQUEUE_ATTEMPTS: u32 = 20; + impl Encoder for MediaCodecAudioEncoder { type InputType = MediaCodecAudioEncoderInput; type OutputType = MediaCodecAudioEncoderOutput; @@ -49,7 +60,7 @@ impl Encoder for MediaCodecAudioEncoder } impl MediaCodecAudioEncoder { - pub fn new(options: &A, runtime: R) -> Result { + pub fn new(options: &A, _runtime: R) -> Result { let env = &mut attach_current_thread()?; // Create MediaFormat @@ -64,63 +75,41 @@ impl MediaCodecAudioEncoder { // Start encoder codec.start()?; - // Clone for both input and output - let codec_input = codec.clone(); - let codec_output = codec; + let drainer = OutputDrainer::start(&codec, "unienc-aud-out")?; + + let (tx, rx) = tokio::sync::mpsc::channel(INPUT_QUEUE_CAPACITY); + let error = Arc::new(Mutex::new(None)); + let input_thread = AudioInputThread { + codec, + sample_rate: options.sample_rate(), + channels: options.channels(), + last_timestamp: 0, + position_in_samples: None, + }; + { + let error = error.clone(); + std::thread::Builder::new() + .name("unienc-aud-in".to_owned()) + .spawn(move || input_thread.run(rx, error)) + .map_err(AndroidError::ThreadSpawnFailed)?; + } Ok(Self { input: MediaCodecAudioEncoderInput { - codec: codec_input, - sample_rate: options.sample_rate(), channels: options.channels(), - last_timestamp: 0, - position_in_samples: None, next_input_position: None, - runtime: runtime.clone(), + tx, + error, + _runtime: PhantomData, }, output: MediaCodecAudioEncoderOutput { - codec: codec_output, - end_of_stream: false, - runtime, + drainer, + _runtime: PhantomData, }, }) } } -impl Drop for MediaCodecAudioEncoderInput { - fn drop(&mut self) { - // notify end of stream - || -> Result<()> { - // `Drop` cannot await, so this one wait stays on the calling - // thread. It is bounded and happens once per stream, unlike the - // per-buffer waits in `push`, but it is the same shape: if the - // codec has no free input buffer it needs the pull task to - // release an output buffer, and that task needs a worker. - loop { - let buffer_index = self - .codec - .dequeue_input_buffer(Duration::from_millis(100))?; - if buffer_index >= 0 { - self.codec.queue_input_buffer( - buffer_index, - 0, - 0, - self.last_timestamp, - BUFFER_FLAG_END_OF_STREAM, - )?; - return Ok(()); - } - if buffer_index == media_codec_errors::INFO_TRY_AGAIN_LATER { - std::thread::sleep(Duration::from_millis(10)); - } else { - return Err(AndroidError::NoInputBuffer); - } - } - }() - .unwrap(); - } -} - impl EncoderInput for MediaCodecAudioEncoderInput { type Data = AudioSample; @@ -140,8 +129,6 @@ async fn push_impl( // are ignored by `forward_audio_discontinuity`, so the timeline never regresses. let channels = (this.channels as u64).max(1); let frames_in_push = data.data.len() as u64 / channels; - this.position_in_samples - .get_or_insert(data.timestamp_in_samples); let gap = unienc_common::forward_audio_discontinuity( this.next_input_position, data.timestamp_in_samples, @@ -151,63 +138,21 @@ async fn push_impl( // Convert i16 samples to signed 16-bit little-endian PCM bytes, prefixed with `gap` frames of silence // when the input timeline jumped forward (2 bytes per i16 sample, `channels` per frame). let byte_data = data.data_as_s16le_bytes(); - let byte_data = if gap > 0 { + let bytes = if gap > 0 { let mut padded = vec![0u8; (gap * channels) as usize * 2]; padded.extend_from_slice(&byte_data); padded } else { byte_data }; - let mut byte_data = byte_data.as_slice(); - - while !byte_data.is_empty() { - // Get input buffer. The wait happens on the blocking pool so that this - // task does not hold an executor worker while the codec has nothing to - // give; see `dequeue_input_buffer_off_executor`. - let buffer_index = - dequeue_input_buffer_off_executor(&this.runtime, &this.codec, DEQUEUE_TIMEOUT).await?; - if buffer_index >= 0 { - let input_buffer = this.codec.get_input_buffer(buffer_index)?; - { - let env = &mut attach_current_thread()?; - let (_base_ptr, capacity, position) = - get_direct_buffer_info(env, input_buffer.as_obj())?; - - let bytes_to_write = std::cmp::min(byte_data.len(), capacity - position); - crate::common::write_to_buffer(env, &input_buffer, &byte_data[..bytes_to_write])?; - byte_data = &byte_data[bytes_to_write..]; - - // A single push can be split across multiple MediaCodec input buffers (e.g. when a large - // batched buffer is pushed while the main thread is stalled by framerate jitter). Assigning - // every chunk the same input timestamp would make MediaCodec emit output packets with - // duplicate / non-monotonic PTS that the muxer rejects. Instead, derive a contiguous - // timeline that advances by the number of frames queued. The running position is seeded to - // the first input timestamp at the start of `push`; forward discontinuities are materialized - // there as leading silence, so the position simply advances by the frames queued here. - let position_in_samples = this - .position_in_samples - .expect("position_in_samples is initialized at the start of push"); - - // Calculate timestamp in microseconds - let timestamp_us = - (position_in_samples as f64 / this.sample_rate as f64 * 1_000_000.0) as i64; - - this.last_timestamp = timestamp_us; - - // Queue input buffer - this.codec - .queue_input_buffer(buffer_index, 0, bytes_to_write, timestamp_us, 0)?; - - // Advance by the number of whole frames written (2 bytes per i16 sample, `channels` per frame). - let frames_written = (bytes_to_write / (2 * this.channels as usize)) as u64; - this.position_in_samples = Some(position_in_samples + frames_written); - } - } else if buffer_index == media_codec_errors::INFO_TRY_AGAIN_LATER { - // The dequeue above already waited, so retry straight away. - continue; - } else { - return Err(AndroidError::NoInputBuffer); - } + + let chunk = AudioInputChunk { + timestamp_in_samples: data.timestamp_in_samples, + bytes, + }; + if this.tx.send(chunk).await.is_err() { + let error = this.error.lock().map_err(|_| AndroidError::MutexPoisoned)?.take(); + return Err(error.unwrap_or(AndroidError::EncoderInputThreadStopped)); } Ok(()) @@ -217,9 +162,129 @@ impl EncoderOutput for MediaCodecAudioEncod type Data = CommonEncodedData; async fn pull(&mut self) -> unienc_common::Result> { - pull_encoded_data_with_codec(self.runtime.clone(), &self.codec, &mut self.end_of_stream) - .await - .map_err(Into::into) + self.drainer.pull().await.map_err(Into::into) + } +} + +/// Feeds PCM to the codec on a dedicated thread, which waits in `dequeueInputBuffer` itself rather +/// than handing every wait to the blocking pool (see [`OutputDrainer`] for the rationale). Once the +/// input is dropped and the queued PCM has been fed, it signals the end of the stream. +struct AudioInputThread { + codec: MediaCodec, + sample_rate: u32, + channels: u32, + last_timestamp: i64, + /// Running presentation position (in samples) of the next input chunk. Anchored to the first input + /// buffer's timestamp and advanced by the number of frames actually queued, so that a single large + /// (batched) input buffer split across multiple MediaCodec input buffers yields strictly increasing, + /// contiguous timestamps instead of repeating the same input timestamp. + position_in_samples: Option, +} + +impl AudioInputThread { + fn run( + mut self, + mut rx: tokio::sync::mpsc::Receiver, + error: Arc>>, + ) { + let result = (|| -> Result<()> { + while let Some(chunk) = rx.blocking_recv() { + self.queue(chunk)?; + } + self.queue_end_of_stream() + })(); + + if let Err(e) = result { + log::error!("unienc: audio encoder input failed: {e}"); + if let Ok(mut error) = error.lock() { + *error = Some(e); + } + } + } + + /// Waits for a free input buffer, giving up after `max_attempts` timed-out dequeues if given. + fn dequeue_input_buffer(&self, max_attempts: Option) -> Result { + let mut attempts = 0; + loop { + let buffer_index = self.codec.dequeue_input_buffer(DEQUEUE_TIMEOUT)?; + if buffer_index >= 0 { + return Ok(buffer_index); + } + // The dequeue above already waited, so retry straight away. + if buffer_index != media_codec_errors::INFO_TRY_AGAIN_LATER { + return Err(AndroidError::NoInputBuffer); + } + attempts += 1; + if max_attempts.is_some_and(|max| attempts >= max) { + return Err(AndroidError::NoInputBuffer); + } + } + } + + fn queue(&mut self, chunk: AudioInputChunk) -> Result<()> { + self.position_in_samples + .get_or_insert(chunk.timestamp_in_samples); + let mut byte_data = chunk.bytes.as_slice(); + + while !byte_data.is_empty() { + let buffer_index = self.dequeue_input_buffer(None)?; + let input_buffer = self.codec.get_input_buffer(buffer_index)?; + let env = &mut attach_current_thread()?; + let (base_ptr, capacity, position) = + get_direct_buffer_info(env, input_buffer.as_obj())?; + + let bytes_to_write = std::cmp::min(byte_data.len(), capacity - position); + // SAFETY: The input buffer is direct and has `capacity - position` bytes left from + // `position`, and the codec does not read it until it is queued. + unsafe { + std::ptr::copy_nonoverlapping( + byte_data.as_ptr(), + base_ptr.add(position), + bytes_to_write, + ); + } + byte_data = &byte_data[bytes_to_write..]; + + // A single push can be split across multiple MediaCodec input buffers (e.g. when a large + // batched buffer is pushed while the main thread is stalled by framerate jitter). Assigning + // every chunk the same input timestamp would make MediaCodec emit output packets with + // duplicate / non-monotonic PTS that the muxer rejects. Instead, derive a contiguous + // timeline that advances by the number of frames queued. The running position is seeded to + // the first input timestamp; forward discontinuities are materialized by `push` as leading + // silence, so the position simply advances by the frames queued here. + let position_in_samples = self + .position_in_samples + .expect("position_in_samples is initialized above"); + + // Calculate timestamp in microseconds + let timestamp_us = + (position_in_samples as f64 / self.sample_rate as f64 * 1_000_000.0) as i64; + + self.last_timestamp = timestamp_us; + + // Queue input buffer + self.codec + .queue_input_buffer(buffer_index, position, bytes_to_write, timestamp_us, 0)?; + + // Advance by the number of whole frames written (2 bytes per i16 sample, `channels` per frame). + let frames_written = (bytes_to_write / (2 * self.channels as usize)) as u64; + self.position_in_samples = Some(position_in_samples + frames_written); + } + + Ok(()) + } + + fn queue_end_of_stream(&self) -> Result<()> { + // A free input buffer appears only while the output is drained. If the output has already been + // dropped, none will, so give up rather than keep the thread waiting forever. + let buffer_index = self.dequeue_input_buffer(Some(EOS_DEQUEUE_ATTEMPTS))?; + self.codec.queue_input_buffer( + buffer_index, + 0, + 0, + self.last_timestamp, + BUFFER_FLAG_END_OF_STREAM, + ) } } diff --git a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/common.rs b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/common.rs index 751c3bcc..eb69f393 100644 --- a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/common.rs +++ b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/common.rs @@ -447,24 +447,28 @@ pub fn write_to_buffer(env: &mut JNIEnv, buffer: &SafeGlobalRef, data: &[u8]) -> } /// Read data from ByteBuffer +/// +/// The buffers MediaCodec hands out are direct, so the bytes are copied straight out of the +/// buffer's memory instead of through a Java byte array. pub fn read_from_buffer( env: &mut JNIEnv, buffer: &SafeGlobalRef, offset: usize, size: usize, ) -> Result> { - bindings::ByteBuffer::set_position(env, buffer.as_obj(), offset as jint)?; - - let byte_array = env.new_byte_array(size as jint)?; - bindings::ByteBuffer::get(env, buffer.as_obj(), &byte_array, 0, size as jint)?; - - // Convert to Vec - let mut result = vec![0u8; size]; - env.get_byte_array_region(&byte_array, 0, unsafe { - std::slice::from_raw_parts_mut(result.as_mut_ptr() as *mut i8, size) - })?; + let byte_buffer: &jni::objects::JByteBuffer = buffer.as_obj().into(); + let base_address = env.get_direct_buffer_address(byte_buffer)?; + if base_address.is_null() { + return Err(AndroidError::NotDirectBuffer); + } + let capacity = env.get_direct_buffer_capacity(byte_buffer)?; + if offset.checked_add(size).is_none_or(|end| end > capacity) { + return Err(AndroidError::BufferOverflow); + } - Ok(result) + // SAFETY: The range lies within the buffer, which the codec does not touch until the output + // buffer is released. + Ok(unsafe { std::slice::from_raw_parts(base_address.add(offset), size) }.to_vec()) } /// Read data from ByteBuffer @@ -575,89 +579,125 @@ pub(crate) fn dequeue_input_buffer_off_executor( runtime.spawn_blocking(move || codec.dequeue_input_buffer(timeout)) } -/// The output-side counterpart of [`dequeue_input_buffer_off_executor`], for the -/// same reason. -pub(crate) fn dequeue_output_buffer_off_executor( - runtime: &R, - codec: &MediaCodec, - buffer_info: &SafeGlobalRef, - timeout: Duration, -) -> Pin> + Send + 'static>> { - let codec = codec.clone(); - let buffer_info = buffer_info.clone(); - let timeout_us = timeout.as_micros() as i64; - runtime.spawn_blocking(move || codec.dequeue_output_buffer(&buffer_info, timeout_us)) +/// Encoded outputs that the drain thread may hold before it waits for the consumer. +const OUTPUT_QUEUE_CAPACITY: usize = 8; + +/// Drains the output of a codec on a dedicated thread. +/// +/// The thread waits in `dequeueOutputBuffer` itself, copies each output out of the codec and +/// releases the buffer, and hands the result over through a bounded channel. Handing every dequeue +/// to the blocking pool instead costs a task, a wake-up and a park per call, which amounts to more +/// than the JNI calls themselves at the rate audio and video produce outputs. The thread stays +/// attached to the JVM for its lifetime. +/// +/// The thread ends after the end-of-stream output, after an error, or once the receiver is dropped, +/// which it notices within [`DEQUEUE_TIMEOUT`]. It holds a reference to the codec until then. +pub(crate) struct OutputDrainer { + rx: tokio::sync::mpsc::Receiver>, } -/// Takes the codec's runtime by value rather than by reference because the -/// future has to be `Send` and `Runtime` is not `Sync`. -pub(crate) async fn pull_encoded_data_with_codec( - runtime: R, - codec: &MediaCodec, - end_of_stream: &mut bool, -) -> Result> { - if *end_of_stream { - return Ok(None); +impl OutputDrainer { + /// Starts draining. The codec must have been started. + pub fn start(codec: &MediaCodec, thread_name: &str) -> Result { + let (tx, rx) = tokio::sync::mpsc::channel(OUTPUT_QUEUE_CAPACITY); + let codec = codec.clone(); + std::thread::Builder::new() + .name(thread_name.to_owned()) + .spawn(move || drain_output(codec, tx)) + .map_err(AndroidError::ThreadSpawnFailed)?; + Ok(Self { rx }) } - // One `BufferInfo` serves every iteration: `dequeueOutputBuffer` overwrites - // its fields on each successful call. - let buffer_info = { - let env = &mut attach_current_thread()?; - create_buffer_info(env)? - }; - - loop { - let buffer_index = - dequeue_output_buffer_off_executor(&runtime, codec, &buffer_info, DEQUEUE_TIMEOUT) - .await?; + /// Returns the next output, or `None` once the end-of-stream output has been returned. + pub async fn pull(&mut self) -> Result> { + match self.rx.recv().await { + Some(result) => result.map(Some), + None => Ok(None), + } + } +} - if buffer_index >= 0 { +fn drain_output(codec: MediaCodec, tx: tokio::sync::mpsc::Sender>) { + let result = (|| -> Result<()> { + // One `BufferInfo` serves every iteration: `dequeueOutputBuffer` overwrites its fields on + // each successful call. + let buffer_info = { let env = &mut attach_current_thread()?; - let output_buffer = codec.get_output_buffer(buffer_index)?; - let (offset, size, flags, timestamp) = read_buffer_info_common(env, &buffer_info)?; - - // Read encoded data - let encoded_data = read_from_buffer(env, &output_buffer, offset, size)?; - - log::trace!( - "new frame data: flags: {:?}, length: {}, timestamp: {}, offset: {}, {:?}", - flags, - encoded_data.len(), - timestamp, - offset, - encoded_data.iter().take(32).collect::>() - ); + create_buffer_info(env)? + }; - let video_data = CommonEncodedData { - content: CommonEncodedDataContent::Buffer { - data: encoded_data, - buffer_flag: flags, - }, - timestamp: timestamp as f64 / 1_000_000.0, // Convert from microseconds + loop { + if tx.is_closed() { + return Ok(()); + } + let Some((data, end_of_stream)) = + dequeue_encoded_data(&codec, &buffer_info, DEQUEUE_TIMEOUT)? + else { + continue; }; - - codec.release_output_buffer(buffer_index, false)?; - - if (flags & media_codec_buffer_flag::BUFFER_FLAG_END_OF_STREAM) != 0 { - *end_of_stream = true; + if tx.blocking_send(Ok(data)).is_err() || end_of_stream { + return Ok(()); } - return Ok(Some(video_data)); } + })(); - if buffer_index == media_codec_errors::INFO_OUTPUT_FORMAT_CHANGED { - let map = codec.get_output_format()?; + if let Err(error) = result { + _ = tx.blocking_send(Err(error)); + } +} - let metadata = CommonEncodedData { - content: CommonEncodedDataContent::FormatInfo(map), - timestamp: 0.0, - }; - return Ok(Some(metadata)); - } +/// Waits up to `timeout` for an output of the codec. Returns `None` when there is nothing to take +/// yet, and otherwise the output together with whether it ends the stream. +fn dequeue_encoded_data( + codec: &MediaCodec, + buffer_info: &SafeGlobalRef, + timeout: Duration, +) -> Result> { + let buffer_index = codec.dequeue_output_buffer(buffer_info, timeout.as_micros() as i64)?; - // Anything else, `INFO_TRY_AGAIN_LATER` included, means there is nothing - // to take yet. The dequeue above already waited, so retry straight away. + if buffer_index >= 0 { + let env = &mut attach_current_thread()?; + let output_buffer = codec.get_output_buffer(buffer_index)?; + let (offset, size, flags, timestamp) = read_buffer_info_common(env, buffer_info)?; + + // Read encoded data + let encoded_data = read_from_buffer(env, &output_buffer, offset, size)?; + + log::trace!( + "new frame data: flags: {:?}, length: {}, timestamp: {}, offset: {}, {:?}", + flags, + encoded_data.len(), + timestamp, + offset, + encoded_data.iter().take(32).collect::>() + ); + + let video_data = CommonEncodedData { + content: CommonEncodedDataContent::Buffer { + data: encoded_data, + buffer_flag: flags, + }, + timestamp: timestamp as f64 / 1_000_000.0, // Convert from microseconds + }; + + codec.release_output_buffer(buffer_index, false)?; + + let end_of_stream = (flags & media_codec_buffer_flag::BUFFER_FLAG_END_OF_STREAM) != 0; + return Ok(Some((video_data, end_of_stream))); } + + if buffer_index == media_codec_errors::INFO_OUTPUT_FORMAT_CHANGED { + let map = codec.get_output_format()?; + + let metadata = CommonEncodedData { + content: CommonEncodedDataContent::FormatInfo(map), + timestamp: 0.0, + }; + return Ok(Some((metadata, false))); + } + + // Anything else, `INFO_TRY_AGAIN_LATER` included, means there is nothing to take yet. + Ok(None) } pub(crate) fn format_to_map( diff --git a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/error.rs b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/error.rs index 0064d115..b9711015 100644 --- a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/error.rs +++ b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/error.rs @@ -20,6 +20,15 @@ pub enum AndroidError { #[error("Buffer is not a direct buffer")] NotDirectBuffer, + #[error("Range exceeds the capacity of the buffer")] + BufferOverflow, + + #[error("Failed to spawn a thread: {0}")] + ThreadSpawnFailed(std::io::Error), + + #[error("Encoder input thread has stopped")] + EncoderInputThreadStopped, + // MediaCodec related errors #[error("Image is null")] ImageNull, @@ -167,6 +176,9 @@ impl CategorizedError for AndroidError { // Resource allocation errors AndroidError::NotDirectBuffer => ErrorCategory::ResourceAllocation, + AndroidError::BufferOverflow => ErrorCategory::ResourceAllocation, + AndroidError::ThreadSpawnFailed(_) => ErrorCategory::ResourceAllocation, + AndroidError::EncoderInputThreadStopped => ErrorCategory::Communication, AndroidError::ImageNull => ErrorCategory::ResourceAllocation, AndroidError::NoInputBuffer => ErrorCategory::ResourceAllocation, AndroidError::ByteBufferCreationFailed => ErrorCategory::ResourceAllocation, diff --git a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/video/mod.rs b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/video/mod.rs index b598ee2b..564b1622 100644 --- a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/video/mod.rs +++ b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/video/mod.rs @@ -65,9 +65,10 @@ unsafe impl Send for MediaCodecVideoEncoder pub struct MediaCodecVideoEncoderOutput { codec: MediaCodec, - end_of_stream: bool, initialization: Option>, - runtime: R, + /// Started on the first pull after the codec has been started. + drainer: Option, + _runtime: std::marker::PhantomData, } impl Encoder for MediaCodecVideoEncoder { @@ -162,9 +163,9 @@ impl MediaCodecVideoEncoder { }, output: MediaCodecVideoEncoderOutput { codec: codec_output, - end_of_stream: false, initialization: rx.into(), - runtime, + drainer: None, + _runtime: std::marker::PhantomData, }, }) } @@ -399,7 +400,13 @@ async fn pull_video_output_impl( this.initialization = None; } - pull_encoded_data_with_codec(this.runtime.clone(), &this.codec, &mut this.end_of_stream).await + let drainer = match &mut this.drainer { + Some(drainer) => drainer, + None => this + .drainer + .insert(OutputDrainer::start(&this.codec, "unienc-vid-out")?), + }; + drainer.pull().await } // Helper functions for JNI MediaCodec calls diff --git a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/vulkan/hardware_buffer.rs b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/vulkan/hardware_buffer.rs index 9d97758e..c3b416de 100644 --- a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/vulkan/hardware_buffer.rs +++ b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/vulkan/hardware_buffer.rs @@ -169,6 +169,11 @@ impl HardwareBufferImage { pub fn vk_image_view(&self) -> vk::ImageView { *self.view } + + /// The hardware buffer this image was imported from. + pub fn ahb(&self) -> *mut ndk_sys::AHardwareBuffer { + self.ahb + } } impl Drop for HardwareBufferImage { diff --git a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/vulkan/hardware_buffer_surface.rs b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/vulkan/hardware_buffer_surface.rs index cbfffa9f..c42e2b64 100644 --- a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/vulkan/hardware_buffer_surface.rs +++ b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/vulkan/hardware_buffer_surface.rs @@ -5,6 +5,8 @@ use crate::vulkan::CONTEXT; use crate::vulkan::hardware_buffer::HardwareBufferImage; use crate::vulkan::types::VulkanFramebufferHandle; use ash::vk; +use std::collections::VecDeque; +use std::sync::{Arc, Mutex}; /// A surface backed by ImageWriter and HardwareBuffer /// This provides explicit control over HardwareBuffer usage flags (VIDEO_ENCODE) @@ -12,6 +14,21 @@ pub struct HardwareBufferSurface { image_writer: ImageWriter, width: u32, height: u32, + /// Vulkan resources of the hardware buffers seen so far, most recently imported last. + targets: Mutex>>, + max_targets: usize, +} + +/// A hardware buffer imported into Vulkan, together with the framebuffer that renders into it. +/// +/// The ImageWriter cycles through a fixed set of hardware buffers, so the import, which allocates +/// device memory and creates an image, a view and a framebuffer, is done once per buffer rather +/// than once per frame. The imported image holds a reference to its hardware buffer, so the buffer +/// cannot be freed, and its address cannot be reused by another buffer, while it is cached. +struct HardwareBufferTarget { + // Dropped before the image it is attached to. + framebuffer: VulkanFramebufferHandle, + vk_image: HardwareBufferImage, } impl HardwareBufferSurface { @@ -29,6 +46,10 @@ impl HardwareBufferSurface { image_writer, width, height, + targets: Mutex::new(VecDeque::new()), + // The ImageWriter is expected to keep `max_images` buffers, but it may replace them. + // Buffers that are no longer in use are evicted once more than this many are cached. + max_targets: max_images.max(1) as usize * 2, }) } @@ -39,6 +60,24 @@ impl HardwareBufferSurface { // Get the hardware buffer let ahb = image.get_hardware_buffer()?; + let target = self.get_or_import_target(ahb)?; + + Ok(HardwareBufferFrame { + image, + target, + width: self.width, + height: self.height, + }) + } + + fn get_or_import_target( + &self, + ahb: *mut ndk_sys::AHardwareBuffer, + ) -> Result> { + let mut targets = self.targets.lock().map_err(|_| AndroidError::MutexPoisoned)?; + if let Some(target) = targets.iter().find(|target| target.vk_image.ahb() == ahb) { + return Ok(target.clone()); + } // Get the Vulkan context let cx = CONTEXT @@ -67,20 +106,25 @@ impl HardwareBufferSurface { cx.device.clone(), ); - Ok(HardwareBufferFrame { - image, - vk_image, + let target = Arc::new(HardwareBufferTarget { framebuffer, - width: self.width, - height: self.height, - }) + vk_image, + }); + + // A frame in flight holds its own reference, so an evicted target is destroyed only after + // the frame has been queued. + if targets.len() >= self.max_targets { + targets.pop_front(); + } + targets.push_back(target.clone()); + Ok(target) } /// Queue a frame back to the encoder with timestamp pub fn queue_frame(&self, frame: HardwareBufferFrame, timestamp_ns: i64) -> Result<()> { - // Drop Vulkan resources first (framebuffer, vk_image) - drop(frame.framebuffer); - drop(frame.vk_image); + // Release this frame's reference to the Vulkan resources first. They stay cached for the + // next time the ImageWriter hands out the same hardware buffer. + drop(frame.target); // Then queue the image to ImageWriter self.image_writer @@ -94,18 +138,17 @@ impl HardwareBufferSurface { /// Contains both the ImageWriter image and the imported Vulkan resources pub struct HardwareBufferFrame { image: ImageWriterImage, - pub vk_image: HardwareBufferImage, - pub framebuffer: VulkanFramebufferHandle, + target: Arc, pub width: u32, pub height: u32, } impl HardwareBufferFrame { pub fn vk_image_handle(&self) -> vk::Image { - self.vk_image.vk_image() + self.target.vk_image.vk_image() } pub fn vk_framebuffer(&self) -> vk::Framebuffer { - *self.framebuffer + *self.target.framebuffer } } From 273948b7143c0a088af59cb7571b1e5d4c61e04e Mon Sep 17 00:00:00 2001 From: ruccho Date: Wed, 28 Oct 2026 15:19:31 +0900 Subject: [PATCH 2/3] Apply cargo fmt --- .../crates/java_api_macros/src/descriptor.rs | 4 +--- .../crates/unienc_android_mc/src/audio/mod.rs | 15 ++++++++++++--- .../src/vulkan/hardware_buffer_surface.rs | 5 ++++- 3 files changed, 17 insertions(+), 7 deletions(-) diff --git a/InstantReplay.Externals/unienc/crates/java_api_macros/src/descriptor.rs b/InstantReplay.Externals/unienc/crates/java_api_macros/src/descriptor.rs index 69a7f071..53800520 100644 --- a/InstantReplay.Externals/unienc/crates/java_api_macros/src/descriptor.rs +++ b/InstantReplay.Externals/unienc/crates/java_api_macros/src/descriptor.rs @@ -67,9 +67,7 @@ impl JType { /// `jni::signature::ReturnType` of a method or field of this type. pub fn return_signature(self) -> TokenStream { - let primitive = |name: TokenStream| { - quote!(::jni::signature::ReturnType::Primitive(::jni::signature::Primitive::#name)) - }; + let primitive = |name: TokenStream| quote!(::jni::signature::ReturnType::Primitive(::jni::signature::Primitive::#name)); match self { JType::Void => primitive(quote!(Void)), JType::Bool => primitive(quote!(Boolean)), diff --git a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/audio/mod.rs b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/audio/mod.rs index 12a7e9a3..d40b4849 100644 --- a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/audio/mod.rs +++ b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/audio/mod.rs @@ -151,7 +151,11 @@ async fn push_impl( bytes, }; if this.tx.send(chunk).await.is_err() { - let error = this.error.lock().map_err(|_| AndroidError::MutexPoisoned)?.take(); + let error = this + .error + .lock() + .map_err(|_| AndroidError::MutexPoisoned)? + .take(); return Err(error.unwrap_or(AndroidError::EncoderInputThreadStopped)); } @@ -263,8 +267,13 @@ impl AudioInputThread { self.last_timestamp = timestamp_us; // Queue input buffer - self.codec - .queue_input_buffer(buffer_index, position, bytes_to_write, timestamp_us, 0)?; + self.codec.queue_input_buffer( + buffer_index, + position, + bytes_to_write, + timestamp_us, + 0, + )?; // Advance by the number of whole frames written (2 bytes per i16 sample, `channels` per frame). let frames_written = (bytes_to_write / (2 * self.channels as usize)) as u64; diff --git a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/vulkan/hardware_buffer_surface.rs b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/vulkan/hardware_buffer_surface.rs index c42e2b64..58270e03 100644 --- a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/vulkan/hardware_buffer_surface.rs +++ b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/vulkan/hardware_buffer_surface.rs @@ -74,7 +74,10 @@ impl HardwareBufferSurface { &self, ahb: *mut ndk_sys::AHardwareBuffer, ) -> Result> { - let mut targets = self.targets.lock().map_err(|_| AndroidError::MutexPoisoned)?; + let mut targets = self + .targets + .lock() + .map_err(|_| AndroidError::MutexPoisoned)?; if let Some(target) = targets.iter().find(|target| target.vk_image.ahb() == ahb) { return Ok(target.clone()); } From c4bb13d7c78c3c56966d1c3f24a72d8aab118998 Mon Sep 17 00:00:00 2001 From: ruccho Date: Wed, 28 Oct 2026 18:29:40 +0900 Subject: [PATCH 3/3] Run the encoder loops on the blocking pool instead of dedicated threads Each loop is handed to the blocking pool once, so it keeps waiting in MediaCodec itself without a hand-off per dequeue, and its thread returns to the pool when the encoder ends. --- .../crates/unienc_android_mc/src/audio/mod.rs | 34 +++++++++--------- .../crates/unienc_android_mc/src/common.rs | 36 +++++++++++-------- .../crates/unienc_android_mc/src/error.rs | 10 ++---- .../crates/unienc_android_mc/src/video/mod.rs | 6 ++-- 4 files changed, 43 insertions(+), 43 deletions(-) diff --git a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/audio/mod.rs b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/audio/mod.rs index d40b4849..03d1a415 100644 --- a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/audio/mod.rs +++ b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/audio/mod.rs @@ -21,9 +21,9 @@ pub struct MediaCodecAudioEncoderInput { /// Expected input timestamp (in samples) of the next push, i.e. the previous push's timestamp plus /// the number of frames it delivered. Used to detect discontinuities in the input timeline. next_input_position: Option, - /// Hands PCM to the input thread. Dropping it lets the thread signal the end of the stream. + /// Hands PCM to the input loop. Dropping it lets the loop signal the end of the stream. tx: tokio::sync::mpsc::Sender, - /// The error that stopped the input thread, if any. + /// The error that stopped the input loop, if any. error: Arc>>, _runtime: PhantomData, } @@ -35,16 +35,16 @@ pub struct MediaCodecAudioEncoderOutput { _runtime: PhantomData, } -/// PCM handed from `push` to the input thread. +/// PCM handed from `push` to the input loop. struct AudioInputChunk { /// Input timestamp (in samples) of the first frame of `bytes`. Only the first chunk's is used; - /// the thread derives the rest from the frames it has queued. + /// the loop derives the rest from the frames it has queued. timestamp_in_samples: u64, /// Signed 16-bit little-endian PCM, including any leading silence that fills a gap. bytes: Vec, } -/// Chunks `push` may hand over before it waits for the input thread. +/// Chunks `push` may hand over before it waits for the input loop. const INPUT_QUEUE_CAPACITY: usize = 2; /// Timed-out dequeues tolerated while waiting for the input buffer that carries the end of stream. @@ -60,7 +60,7 @@ impl Encoder for MediaCodecAudioEncoder } impl MediaCodecAudioEncoder { - pub fn new(options: &A, _runtime: R) -> Result { + pub fn new(options: &A, runtime: R) -> Result { let env = &mut attach_current_thread()?; // Create MediaFormat @@ -75,11 +75,11 @@ impl MediaCodecAudioEncoder { // Start encoder codec.start()?; - let drainer = OutputDrainer::start(&codec, "unienc-aud-out")?; + let drainer = OutputDrainer::start(&runtime, &codec); let (tx, rx) = tokio::sync::mpsc::channel(INPUT_QUEUE_CAPACITY); let error = Arc::new(Mutex::new(None)); - let input_thread = AudioInputThread { + let input_loop = AudioInputLoop { codec, sample_rate: options.sample_rate(), channels: options.channels(), @@ -88,10 +88,7 @@ impl MediaCodecAudioEncoder { }; { let error = error.clone(); - std::thread::Builder::new() - .name("unienc-aud-in".to_owned()) - .spawn(move || input_thread.run(rx, error)) - .map_err(AndroidError::ThreadSpawnFailed)?; + run_on_blocking_pool(&runtime, move || input_loop.run(rx, error)); } Ok(Self { @@ -156,7 +153,7 @@ async fn push_impl( .lock() .map_err(|_| AndroidError::MutexPoisoned)? .take(); - return Err(error.unwrap_or(AndroidError::EncoderInputThreadStopped)); + return Err(error.unwrap_or(AndroidError::EncoderInputStopped)); } Ok(()) @@ -170,10 +167,11 @@ impl EncoderOutput for MediaCodecAudioEncod } } -/// Feeds PCM to the codec on a dedicated thread, which waits in `dequeueInputBuffer` itself rather -/// than handing every wait to the blocking pool (see [`OutputDrainer`] for the rationale). Once the +/// Feeds PCM to the codec in a loop that runs on the blocking pool and waits in `dequeueInputBuffer` +/// itself, rather than handing every wait to the blocking pool as a separate call (see +/// [`OutputDrainer`] for the rationale). Once the /// input is dropped and the queued PCM has been fed, it signals the end of the stream. -struct AudioInputThread { +struct AudioInputLoop { codec: MediaCodec, sample_rate: u32, channels: u32, @@ -185,7 +183,7 @@ struct AudioInputThread { position_in_samples: Option, } -impl AudioInputThread { +impl AudioInputLoop { fn run( mut self, mut rx: tokio::sync::mpsc::Receiver, @@ -285,7 +283,7 @@ impl AudioInputThread { fn queue_end_of_stream(&self) -> Result<()> { // A free input buffer appears only while the output is drained. If the output has already been - // dropped, none will, so give up rather than keep the thread waiting forever. + // dropped, none will, so give up rather than keep the loop waiting forever. let buffer_index = self.dequeue_input_buffer(Some(EOS_DEQUEUE_ATTEMPTS))?; self.codec.queue_input_buffer( buffer_index, diff --git a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/common.rs b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/common.rs index eb69f393..f63df276 100644 --- a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/common.rs +++ b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/common.rs @@ -7,7 +7,7 @@ use jni::{ use std::future::Future; use std::pin::Pin; use std::{collections::HashMap, fmt::Display, sync::Arc, time::Duration}; -use unienc_common::{EncodedData, SpawnBlocking, UniencSampleKind, VideoFrameBgra32}; +use unienc_common::{EncodedData, Runtime, SpawnBlocking, UniencSampleKind, VideoFrameBgra32}; use crate::bindings; use crate::error::{AndroidError, Result}; @@ -579,18 +579,27 @@ pub(crate) fn dequeue_input_buffer_off_executor( runtime.spawn_blocking(move || codec.dequeue_input_buffer(timeout)) } -/// Encoded outputs that the drain thread may hold before it waits for the consumer. +/// Runs a long-running blocking loop on the blocking pool of the runtime. +/// +/// The future of the blocking call is awaited by a task on the runtime rather than dropped, because +/// dropping it may cancel the call before it starts. +pub(crate) fn run_on_blocking_pool(runtime: &R, f: impl FnOnce() + Send + 'static) { + runtime.spawn(runtime.spawn_blocking(f)); +} + +/// Encoded outputs that the drain loop may hold before it waits for the consumer. const OUTPUT_QUEUE_CAPACITY: usize = 8; -/// Drains the output of a codec on a dedicated thread. +/// Drains the output of a codec in a loop that runs on the blocking pool. /// -/// The thread waits in `dequeueOutputBuffer` itself, copies each output out of the codec and -/// releases the buffer, and hands the result over through a bounded channel. Handing every dequeue -/// to the blocking pool instead costs a task, a wake-up and a park per call, which amounts to more -/// than the JNI calls themselves at the rate audio and video produce outputs. The thread stays -/// attached to the JVM for its lifetime. +/// The loop waits in `dequeueOutputBuffer` itself, copies each output out of the codec and releases +/// the buffer, and hands the result over through a bounded channel. Handing every dequeue to the +/// blocking pool as a separate call instead costs a task, a wake-up and a park per call, which +/// amounts to more than the JNI calls themselves at the rate audio and video produce outputs. The +/// whole loop is handed over once instead, so it occupies one thread of the pool while it runs, and +/// the thread returns to the pool afterwards. /// -/// The thread ends after the end-of-stream output, after an error, or once the receiver is dropped, +/// The loop ends after the end-of-stream output, after an error, or once the receiver is dropped, /// which it notices within [`DEQUEUE_TIMEOUT`]. It holds a reference to the codec until then. pub(crate) struct OutputDrainer { rx: tokio::sync::mpsc::Receiver>, @@ -598,14 +607,11 @@ pub(crate) struct OutputDrainer { impl OutputDrainer { /// Starts draining. The codec must have been started. - pub fn start(codec: &MediaCodec, thread_name: &str) -> Result { + pub fn start(runtime: &R, codec: &MediaCodec) -> Self { let (tx, rx) = tokio::sync::mpsc::channel(OUTPUT_QUEUE_CAPACITY); let codec = codec.clone(); - std::thread::Builder::new() - .name(thread_name.to_owned()) - .spawn(move || drain_output(codec, tx)) - .map_err(AndroidError::ThreadSpawnFailed)?; - Ok(Self { rx }) + run_on_blocking_pool(runtime, move || drain_output(codec, tx)); + Self { rx } } /// Returns the next output, or `None` once the end-of-stream output has been returned. diff --git a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/error.rs b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/error.rs index b9711015..8cae5ec5 100644 --- a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/error.rs +++ b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/error.rs @@ -23,11 +23,8 @@ pub enum AndroidError { #[error("Range exceeds the capacity of the buffer")] BufferOverflow, - #[error("Failed to spawn a thread: {0}")] - ThreadSpawnFailed(std::io::Error), - - #[error("Encoder input thread has stopped")] - EncoderInputThreadStopped, + #[error("Encoder input loop has stopped")] + EncoderInputStopped, // MediaCodec related errors #[error("Image is null")] @@ -177,8 +174,7 @@ impl CategorizedError for AndroidError { // Resource allocation errors AndroidError::NotDirectBuffer => ErrorCategory::ResourceAllocation, AndroidError::BufferOverflow => ErrorCategory::ResourceAllocation, - AndroidError::ThreadSpawnFailed(_) => ErrorCategory::ResourceAllocation, - AndroidError::EncoderInputThreadStopped => ErrorCategory::Communication, + AndroidError::EncoderInputStopped => ErrorCategory::Communication, AndroidError::ImageNull => ErrorCategory::ResourceAllocation, AndroidError::NoInputBuffer => ErrorCategory::ResourceAllocation, AndroidError::ByteBufferCreationFailed => ErrorCategory::ResourceAllocation, diff --git a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/video/mod.rs b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/video/mod.rs index 564b1622..0c8f8068 100644 --- a/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/video/mod.rs +++ b/InstantReplay.Externals/unienc/crates/unienc_android_mc/src/video/mod.rs @@ -68,7 +68,7 @@ pub struct MediaCodecVideoEncoderOutput { initialization: Option>, /// Started on the first pull after the codec has been started. drainer: Option, - _runtime: std::marker::PhantomData, + runtime: R, } impl Encoder for MediaCodecVideoEncoder { @@ -165,7 +165,7 @@ impl MediaCodecVideoEncoder { codec: codec_output, initialization: rx.into(), drainer: None, - _runtime: std::marker::PhantomData, + runtime, }, }) } @@ -404,7 +404,7 @@ async fn pull_video_output_impl( Some(drainer) => drainer, None => this .drainer - .insert(OutputDrainer::start(&this.codec, "unienc-vid-out")?), + .insert(OutputDrainer::start(&this.runtime, &this.codec)), }; drainer.pull().await }