audio-worker: Set priority when called from cras_processor

BUG=b:233174528
TEST=cras_tests capture /dev/null --effects=0x300
TEST=ps -o rtprio,comm -T -u cras

Change-Id: I8d74f16fb787cbe544374c53eb8a03f84945f9a9
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/third_party/adhd/+/5665517
Tested-by: [email protected] <[email protected]>
Reviewed-by: Chih-Yang Hsia <[email protected]>
Commit-Queue: Li-Yu Yu <[email protected]>

Author: afq984

audio_processor/src/bin/offline-pipeline.rs

@@ -169,7 +169,7 @@ fn run(command: Command) {
     let (profile_sender, profile_receiver) = channel();
     let mut pipeline = PipelineBuilder::new(check_shape.get_output_format())
         .with_profile_sender(profile_sender)
-        .with_worker_factory(AudioWorkerSubprocessFactory)
+        .with_worker_factory(AudioWorkerSubprocessFactory::default())
         .build(Processor::Pipeline {
             processors: pipeline_decl,
         })

audio_processor/src/processors/peer/managed.rs

@@ -10,6 +10,7 @@ use command_fds::CommandFdExt;
 use command_fds::FdMapping;
 
 use super::create_socketpair;
+use super::worker::AUDIO_WORKER_SET_THREAD_PRIORITY;
 use super::BlockingSeqPacketProcessor;
 use super::Worker;
 use crate::config::Processor;
@@ -90,18 +91,31 @@ impl Drop for ThreadedWorkerHandle {
     }
 }
 
-pub struct AudioWorkerSubprocessFactory;
+#[derive(Default)]
+pub struct AudioWorkerSubprocessFactory {
+    set_thread_priority: bool,
+}
+
+impl AudioWorkerSubprocessFactory {
+    pub fn with_set_thread_priority(mut self) -> Self {
+        self.set_thread_priority = true;
+        self
+    }
+}
 
 impl WorkerFactory for AudioWorkerSubprocessFactory {
     fn create(&self, worker_fd: OwnedFd) -> anyhow::Result<Box<dyn WorkerHandle>> {
-        let child = std::process::Command::new("audio-worker")
+        let mut command = std::process::Command::new("audio-worker");
+        command
             .fd_mappings(vec![FdMapping {
                 parent_fd: worker_fd,
                 child_fd: 3,
             }])
-            .context("fd_mappings")?
-            .spawn()
-            .context("spawn audio-worker")?;
+            .context("fd_mappings")?;
+        if self.set_thread_priority {
+            command.env(AUDIO_WORKER_SET_THREAD_PRIORITY, "1");
+        }
+        let child = command.spawn().context("spawn audio-worker")?;
         Ok(Box::new(AudioWorkerSubprocessHandle { child }))
     }
 }

audio_processor/src/processors/peer/worker.rs

@@ -19,11 +19,15 @@ use crate::processors::peer::messages::recv;
 use crate::processors::peer::messages::Init;
 use crate::processors::peer::messages::RequestOp;
 use crate::processors::peer::messages::INIT_MAX_SIZE;
+use crate::util::set_thread_priority;
 use crate::AudioProcessor;
 use crate::MultiBuffer;
 use crate::MultiSlice;
 use crate::ProcessorVec;
 
+pub(super) const AUDIO_WORKER_SET_THREAD_PRIORITY: &'static str =
+    "AUDIO_WORKER_SET_THREAD_PRIORITY";
+
 pub struct Worker<'a> {
     fd: BorrowedFd<'a>,
     pipeline: ProcessorVec,
@@ -37,6 +41,10 @@ enum Response<'a> {
 
 impl<'a> Worker<'a> {
     fn new(fd: BorrowedFd<'a>) -> anyhow::Result<Self> {
+        if std::env::var(AUDIO_WORKER_SET_THREAD_PRIORITY).is_ok() {
+            set_thread_priority().context("set_thread_priority")?;
+        }
+
         let mut buf = vec![0u8; INIT_MAX_SIZE];
         let (request_op, payload_len) = recv::<RequestOp>(fd.as_fd(), &mut buf).context("init")?;
         if request_op != RequestOp::Init {

audio_processor/src/processors/thread.rs

@@ -6,35 +6,15 @@ use std::marker::PhantomData;
 use std::sync::mpsc::Receiver;
 use std::sync::mpsc::Sender;
 
-use anyhow::ensure;
 use anyhow::Context;
-use nix::sys::resource::setrlimit;
-use nix::sys::resource::Resource;
 
+use crate::util::set_thread_priority;
 use crate::AudioProcessor;
 use crate::Format;
 use crate::MultiBuffer;
 use crate::Result;
 use crate::Shape;
 
-// TODO(b/268271100): Call the C version when we can build C code before Rust.
-fn set_thread_priority() -> anyhow::Result<()> {
-    // CRAS_SERVER_RT_THREAD_PRIORITY 12
-    let p = 12;
-    setrlimit(Resource::RLIMIT_RTPRIO, p, p).context("setrlimit")?;
-
-    // SAFETY: sched_param is properly initialized.
-    unsafe {
-        let sched_param = libc::sched_param {
-            sched_priority: p as i32,
-        };
-        let rc = libc::pthread_setschedparam(libc::pthread_self(), libc::SCHED_RR, &sched_param);
-        ensure!(rc == 0, "pthread_setschedparam returned {rc}");
-    }
-
-    Ok(())
-}
-
 pub struct ThreadedProcessor<T: AudioProcessor> {
     phantom: PhantomData<T>,
     join_handle: Option<std::thread::JoinHandle<()>>,

audio_processor/src/util.rs

@@ -4,8 +4,11 @@
 
 use std::path::Path;
 
+use anyhow::ensure;
 use anyhow::Context;
 use hound::WavReader;
+use nix::sys::resource::setrlimit;
+use nix::sys::resource::Resource;
 
 use crate::MultiBuffer;
 use crate::Sample;
@@ -25,3 +28,21 @@ pub fn read_wav<T: Sample + hound::Sample>(
 
     Ok((spec, out))
 }
+
+// TODO(b/268271100): Call the C version when we can build C code before Rust.
+pub fn set_thread_priority() -> anyhow::Result<()> {
+    // CRAS_SERVER_RT_THREAD_PRIORITY 12
+    let p = 12;
+    setrlimit(Resource::RLIMIT_RTPRIO, p, p).context("setrlimit")?;
+
+    // SAFETY: sched_param is properly initialized.
+    unsafe {
+        let sched_param = libc::sched_param {
+            sched_priority: p as i32,
+        };
+        let rc = libc::pthread_setschedparam(libc::pthread_self(), libc::SCHED_RR, &sched_param);
+        ensure!(rc == 0, "pthread_setschedparam returned {rc}");
+    }
+
+    Ok(())
+}

cras/src/server/rust/src/cras_processor.rs

@@ -256,7 +256,7 @@ impl CrasProcessor {
         let decl_debug = format!("{decl:?}");
         let pipeline = PipelineBuilder::new(config.format())
             // TODO(b/349784210): Use a hardened worker factory.
-            .with_worker_factory(AudioWorkerSubprocessFactory)
+            .with_worker_factory(AudioWorkerSubprocessFactory::default().with_set_thread_priority())
             .build(Processor::Pipeline { processors: decl })
             .context("failed to build pipeline")?;