audio_processor: Add test for abnormally terminated worker

This proves that ManagedBlockingSeqPacketProcessor
gracefully fails (instead of hangs) if the worker process
is terminated abnormally.

BUG=b:446725446
TEST=bazel test //audio_processor:integration_tests_tests/peer
audio_processor: Add test for abnormally terminated worker

Change-Id: I1547fca70eb9471835538ed356d42522ff437fa8
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/third_party/adhd/+/7207128
Reviewed-by: Yu-Hsuan Hsu <yuhsuan@chromium.org>
Reviewed-by: Hung-Hsien Chen <hunghsienchen@chromium.org>
Commit-Queue: Li-Yu Yu <aaronyu@google.com>
Tested-by: chromeos-cop-builder@chromeos-cop.iam.gserviceaccount.com <chromeos-cop-builder@chromeos-cop.iam.gserviceaccount.com>

Author: afq984

audio_processor/c/bad_plugin.c

@@ -25,6 +25,15 @@ static enum status failing_run(struct plugin_processor* p,
   return ErrOther;
 }
 
+static enum status abort_run(struct plugin_processor* p,
+                             const struct multi_slice* in,
+                             struct multi_slice* out) {
+  (void)p;
+  (void)in;
+  (void)out;
+  abort();
+}
+
 static enum status get_output_frame_rate_48k(struct plugin_processor* p,
                                              size_t* frame_rate) {
   *frame_rate = 48000;
@@ -149,3 +158,30 @@ enum status bad_plugin_failing_get_output_frame_rate_create(
   *out = p;
   return StatusOk;
 }
+
+enum status bad_plugin_abort_create(
+    struct plugin_processor** out,
+    const struct plugin_processor_config* config) {
+  (void)out;
+  (void)config;
+  abort();
+}
+
+enum status bad_plugin_abort_run_create(
+    struct plugin_processor** out,
+    const struct plugin_processor_config* config) {
+  (void)config;
+  static const struct plugin_processor_ops ops = {
+      .run = abort_run,
+      .destroy = free_destroy,
+      .get_output_frame_rate = get_output_frame_rate_48k,
+  };
+
+  struct plugin_processor* p = calloc(1, sizeof(*p));
+  if (!p) {
+    return ErrOutOfMemory;
+  }
+  p->ops = &ops;
+  *out = p;
+  return StatusOk;
+}

audio_processor/c/bad_plugin.h

@@ -42,6 +42,14 @@ enum status bad_plugin_failing_get_output_frame_rate_create(
     struct plugin_processor** out,
     const struct plugin_processor_config* config);
 
+enum status bad_plugin_abort_create(
+    struct plugin_processor** out,
+    const struct plugin_processor_config* config);
+
+enum status bad_plugin_abort_run_create(
+    struct plugin_processor** out,
+    const struct plugin_processor_config* config);
+
 #ifdef __cplusplus
 }
 #endif

audio_processor/tests/peer.rs

@@ -80,3 +80,70 @@ fn test_peer_ffi_negate() {
     let output = processor.process(input.as_multi_slice()).unwrap();
     assert_eq!(output.into_raw(), [[-1., -2., -3., -4.]]);
 }
+
+#[test]
+#[cfg(feature = "bazel")]
+fn test_peer_ffi_abort_create() {
+    let audio_worker_executable = env!("CARGO_BIN_EXE_audio-worker");
+    let factory =
+        AudioWorkerSubprocessFactory::default().with_executable_path(audio_worker_executable);
+    let test_plugin_path = std::env::var("LIBTEST_PLUGINS_SO").expect("LIBTEST_PLUGINS_SO not set");
+
+    let result = ManagedBlockingSeqPacketProcessor::new(
+        &factory,
+        Format {
+            channels: 1,
+            block_size: 4,
+            frame_rate: 48000,
+        },
+        config::Processor::Plugin {
+            path: test_plugin_path.into(),
+            constructor: "bad_plugin_abort_create".to_string(),
+        },
+    );
+
+    let err_msg = match result {
+        Ok(_) => panic!("Expected an error, but got Ok"),
+        Err(e) => {
+            format!("{e:#}")
+        }
+    };
+    assert_eq!(
+        err_msg,
+        "BlockingSeqPacketProcessor::new: recv init message reply: empty message"
+    );
+}
+
+#[test]
+#[cfg(feature = "bazel")]
+fn test_peer_ffi_abort_run() {
+    let audio_worker_executable = env!("CARGO_BIN_EXE_audio-worker");
+    let factory =
+        AudioWorkerSubprocessFactory::default().with_executable_path(audio_worker_executable);
+    let test_plugin_path = std::env::var("LIBTEST_PLUGINS_SO").expect("LIBTEST_PLUGINS_SO not set");
+
+    let mut processor = ManagedBlockingSeqPacketProcessor::new(
+        &factory,
+        Format {
+            channels: 1,
+            block_size: 4,
+            frame_rate: 48000,
+        },
+        config::Processor::Plugin {
+            path: test_plugin_path.into(),
+            constructor: "bad_plugin_abort_run_create".to_string(),
+        },
+    )
+    .unwrap();
+
+    let mut input = MultiBuffer::from(vec![vec![1., 2., 3., 4.]]);
+    let result = processor.process(input.as_multi_slice());
+
+    let err_msg = match result {
+        Ok(_) => panic!("Expected an error, but got Ok"),
+        Err(e) => {
+            format!("{e:#}")
+        }
+    };
+    assert_eq!(err_msg, "unrecoverable error: empty message");
+}