[Chromecast] Complete PostProcessingPipeline changes

chromecast/media/cma/backend/alsa/stream_mixer_alsa.cc

 // Copyright 2015 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "chromecast/media/cma/backend/alsa/stream_mixer_alsa.h"
 
 #include <algorithm>
 #include <cmath>
 #include <limits>
 #include <unordered_set>
 #include <utility>
 
 #include "base/bind_helpers.h"
 #include "base/command_line.h"
 #include "base/lazy_instance.h"
 #include "base/memory/ptr_util.h"
 #include "base/memory/weak_ptr.h"
 #include "base/numerics/saturated_arithmetic.h"
 #include "base/single_thread_task_runner.h"
 #include "base/threading/platform_thread.h"
 #include "base/threading/thread_task_runner_handle.h"
+#include "base/time/time.h"
 #include "chromecast/base/chromecast_switches.h"
 #include "chromecast/media/base/audio_device_ids.h"
 #include "chromecast/media/cma/backend/alsa/alsa_wrapper.h"
 #include "chromecast/media/cma/backend/alsa/filter_group.h"
 #include "chromecast/media/cma/backend/alsa/post_processing_pipeline_parser.h"
 #include "chromecast/media/cma/backend/alsa/stream_mixer_alsa_input_impl.h"
 #include "media/audio/audio_device_description.h"
 #include "media/base/audio_bus.h"
 #include "media/base/media_switches.h"
 
@@ skipping 90 matching lines @@
 const int kUseDefaultFade = -1;
 const int kMediaDuckFadeMs = 150;
 const int kMediaUnduckFadeMs = 700;
 
 int64_t TimespecToMicroseconds(struct timespec time) {
   return static_cast<int64_t>(time.tv_sec) *
              base::Time::kMicrosecondsPerSecond +
          time.tv_nsec / 1000;
 }
 
-void VectorAccumulate(const int32_t* source, size_t size, int32_t* dest) {
-  for (size_t i = 0; i < size; ++i) {
-    dest[i] = base::SaturatedAddition(source[i], dest[i]);
-  }
+bool IsOutputDeviceId(const std::string& device) {
+  return device == ::media::AudioDeviceDescription::kDefaultDeviceId ||
+         device == ::media::AudioDeviceDescription::kCommunicationsDeviceId ||
+         device == kLocalAudioDeviceId || device == kAlarmAudioDeviceId ||
+         device == kTtsAudioDeviceId;
 }
 
 class StreamMixerAlsaInstance : public StreamMixerAlsa {
  public:
   StreamMixerAlsaInstance() {}
   ~StreamMixerAlsaInstance() override {}
 
  private:
   DISALLOW_COPY_AND_ASSIGN(StreamMixerAlsaInstance);
 };
@@ skipping 64 matching lines @@
 
   fixed_output_samples_per_second_ = fixed_samples_per_second;
 
   low_sample_rate_cutoff_ =
       chromecast::GetSwitchValueBoolean(switches::kAlsaEnableUpsampling, false)
           ? kLowSampleRateCutoff
           : 0;
 
   // Read post-processing configuration file
   PostProcessingPipelineParser pipeline_parser;
-  pipeline_parser.Initialize();
 
-  // Media filter group:
-  filter_groups_.push_back(base::MakeUnique<FilterGroup>(
-      std::unordered_set<std::string>(
-          {::media::AudioDeviceDescription::kDefaultDeviceId,
-           kLocalAudioDeviceId, "", kAlarmAudioDeviceId}),
-      AudioContentType::kMedia, kNumOutputChannels,
-      pipeline_parser.GetPipelineByDeviceId(
-          ::media::AudioDeviceDescription::kDefaultDeviceId)));
-
-  // Voice filter group:
-  filter_groups_.push_back(base::MakeUnique<FilterGroup>(
-      std::unordered_set<std::string>(
-          {kTtsAudioDeviceId,
-           ::media::AudioDeviceDescription::kCommunicationsDeviceId}),
-      AudioContentType::kCommunication, kNumOutputChannels,
-      pipeline_parser.GetPipelineByDeviceId(kTtsAudioDeviceId)));
+  CreatePostProcessors(&pipeline_parser);
 
   // TODO(bshaya): Add support for final mix AudioPostProcessor.
   DefineAlsaParameters();
 }
 
+void StreamMixerAlsa::CreatePostProcessors(
+    PostProcessingPipelineParser* pipeline_parser) {
+  std::unordered_set<std::string> used_streams;
+  for (auto& stream_pipeline : pipeline_parser->GetStreamPipelines()) {
+    const auto& device_ids = stream_pipeline.stream_types;
+    for (const std::string& stream_type : device_ids) {
+      CHECK(IsOutputDeviceId(stream_type))
+          << stream_type << " is not a stream type. Stream types are listed "
+          << "in chromecast/media/base/audio_device_ids.cc and "
+          << "media/audio/audio_device_description.cc";
+      CHECK(used_streams.insert(stream_type).second)
+          << "Multiple instances of stream type '" << stream_type << "' in "
+          << pipeline_parser->GetFilePath() << ".";
+    }
+    filter_groups_.push_back(base::MakeUnique<FilterGroup>(
+        kNumOutputChannels, *device_ids.begin() /* name */,
+        stream_pipeline.pipeline, device_ids,
+        std::vector<FilterGroup*>() /* mixed_inputs */));
+    if (device_ids.find(::media::AudioDeviceDescription::kDefaultDeviceId) !=
+        device_ids.end()) {
+      default_filter_ = filter_groups_.back().get();
+    }
+  }
+
+  // Always provide a default filter; OEM may only specify mix filter.
+  if (!default_filter_) {
+    std::string kDefaultDeviceId =
+        ::media::AudioDeviceDescription::kDefaultDeviceId;
+    filter_groups_.push_back(base::MakeUnique<FilterGroup>(
+        kNumOutputChannels, kDefaultDeviceId /* name */, nullptr,
+        std::unordered_set<std::string>({kDefaultDeviceId}),
+        std::vector<FilterGroup*>() /* mixed_inputs */));
+    default_filter_ = filter_groups_.back().get();
+  }
+
+  std::vector<FilterGroup*> filter_group_ptrs(filter_groups_.size());
+  std::transform(
+      filter_groups_.begin(), filter_groups_.end(), filter_group_ptrs.begin(),
+      [](const std::unique_ptr<FilterGroup>& group) { return group.get(); });
+
+  filter_groups_.push_back(base::MakeUnique<FilterGroup>(
+      kNumOutputChannels, "mix", pipeline_parser->GetMixPipeline(),
+      std::unordered_set<std::string>() /* device_ids */, filter_group_ptrs));
+  mix_filter_ = filter_groups_.back().get();
+
+  filter_groups_.push_back(base::MakeUnique<FilterGroup>(
+      kNumOutputChannels, "linearize", pipeline_parser->GetLinearizePipeline(),
+      std::unordered_set<std::string>() /* device_ids */,
+      std::vector<FilterGroup*>({mix_filter_})));
+  linearize_filter_ = filter_groups_.back().get();
+}
+
 void StreamMixerAlsa::ResetTaskRunnerForTest() {
   mixer_task_runner_ = base::ThreadTaskRunnerHandle::Get();
 }
 
+void StreamMixerAlsa::ResetPostProcessorsForTest(
+    const std::string& pipeline_json) {
+  LOG(INFO) << __FUNCTION__ << " disregard previous PostProcessor messages.";
+  filter_groups_.clear();
+  default_filter_ = nullptr;
+  PostProcessingPipelineParser parser(pipeline_json);
+  CreatePostProcessors(&parser);
+}
+
 void StreamMixerAlsa::DefineAlsaParameters() {
   // Get the ALSA output configuration from the command line.
   alsa_buffer_size_ = GetSwitchValueNonNegativeInt(
       switches::kAlsaOutputBufferSize, kDefaultOutputBufferSizeFrames);
 
   alsa_period_size_ = GetSwitchValueNonNegativeInt(
       switches::kAlsaOutputPeriodSize, alsa_buffer_size_ / 16);
   if (alsa_period_size_ >= alsa_buffer_size_) {
     LOG(DFATAL) << "ALSA period size must be smaller than the buffer size";
     alsa_period_size_ = alsa_buffer_size_ / 2;
@@ skipping 244 matching lines @@
   }
 
   // Initialize filters
   for (auto&& filter_group : filter_groups_) {
     filter_group->Initialize(output_samples_per_second_);
   }
 
   RETURN_REPORT_ERROR(PcmPrepare, pcm_);
   RETURN_REPORT_ERROR(PcmStatusMalloc, &pcm_status_);
 
-  rendering_delay_.timestamp_microseconds = kNoTimestamp;
-  rendering_delay_.delay_microseconds = 0;
+  alsa_rendering_delay_.timestamp_microseconds = kNoTimestamp;
+  alsa_rendering_delay_.delay_microseconds = 0;
 
   state_ = kStateNormalPlayback;
 }
 
 void StreamMixerAlsa::Stop() {
   for (auto* observer : loopback_observers_) {
     observer->OnLoopbackInterrupted();
   }
 
   if (alsa_) {
@@ skipping 53 matching lines @@
 
 void StreamMixerAlsa::SetAlsaWrapperForTest(
     std::unique_ptr<AlsaWrapper> alsa_wrapper) {
   if (alsa_) {
     Close();
   }
 
   alsa_ = std::move(alsa_wrapper);
 }
 
-void StreamMixerAlsa::DisablePostProcessingForTest() {
-  for (auto& filter : filter_groups_) {
-    filter->DisablePostProcessingForTest();
-  }
-}
-
 void StreamMixerAlsa::WriteFramesForTest() {
   RUN_ON_MIXER_THREAD(&StreamMixerAlsa::WriteFramesForTest);
   WriteFrames();
 }
 
 void StreamMixerAlsa::ClearInputsForTest() {
   RUN_ON_MIXER_THREAD(&StreamMixerAlsa::ClearInputsForTest);
   inputs_.clear();
 }
 
@@ skipping 24 matching lines @@
   input->SetMuted(volume_info_[type].muted);
 
   check_close_timer_->Stop();
   switch (state_) {
     case kStateUninitialized:
       requested_output_samples_per_second_ = input->input_samples_per_second();
       Start();
     // Fallthrough intended
     case kStateNormalPlayback: {
       bool found_filter_group = false;
-      input->Initialize(rendering_delay_);
+      input->Initialize(alsa_rendering_delay_);
       for (auto&& filter_group : filter_groups_) {
         if (filter_group->CanProcessInput(input.get())) {
           found_filter_group = true;
           input->set_filter_group(filter_group.get());
+          LOG(INFO) << "Added input of type " << input->device_id() << " to "
+                    << filter_group->name();
           break;
         }
       }
-      DCHECK(found_filter_group) << "Could not find a filter group for "
-                                 << input->device_id();
+
+      // Fallback to default_filter_ if provided
+      if (!found_filter_group && default_filter_) {
+        found_filter_group = true;
+        input->set_filter_group(default_filter_);
+        LOG(INFO) << "Added input of type " << input->device_id() << " to "
+                  << default_filter_->name();
+      }
+
+      CHECK(found_filter_group)
+          << "Could not find a filter group for " << input->device_id() << "\n"
+          << "(consider adding a 'default' processor)";
       inputs_.push_back(std::move(input));
     } break;
     case kStateError:
       input->SignalError(StreamMixerAlsaInput::MixerError::kInternalError);
       ignored_inputs_.push_back(std::move(input));
       break;
     default:
       NOTREACHED();
   }
 }
@@ skipping 146 matching lines @@
     } else {
       input->OnSkipped();
     }
   }
 
   if (is_silence) {
     // No inputs have any data to provide. Push silence to prevent underrun.
     chunk_size = kPreventUnderrunChunkSize;
   }
 
-  // Mix and filter each group.
-  std::vector<uint8_t>* interleaved = nullptr;
-  for (auto&& filter_group : filter_groups_) {
-    if (filter_group->MixAndFilter(chunk_size)) {
-      if (!interleaved) {
-        interleaved = filter_group->GetInterleaved();
-      } else {
-        DCHECK_EQ(4, BytesPerOutputFormatSample());
-        VectorAccumulate(
-            reinterpret_cast<int32_t*>(filter_group->GetInterleaved()->data()),
-            chunk_size * kNumOutputChannels,
-            reinterpret_cast<int32_t*>(interleaved->data()));
-      }
-    }
-  }
+  // Recursively mix and filter each group.
+  linearize_filter_->MixAndFilter(chunk_size);
 
-  if (!interleaved) {
-    // No group has any data, write empty buffer.
-    filter_groups_[0]->ClearInterleaved(chunk_size);
-    interleaved = filter_groups_[0]->GetInterleaved();
-  }
-
-  WriteMixedPcm(interleaved, chunk_size);
+  WriteMixedPcm(chunk_size);
   return true;
 }
 
 size_t StreamMixerAlsa::InterleavedSize(int frames) {
   return BytesPerOutputFormatSample() *
          static_cast<size_t>(frames * kNumOutputChannels);
 }
 
 ssize_t StreamMixerAlsa::BytesPerOutputFormatSample() {
   return alsa_->PcmFormatSize(pcm_format_, 1);
 }
 
-void StreamMixerAlsa::WriteMixedPcm(std::vector<uint8_t>* interleaved,
-                                    int frames) {
+void StreamMixerAlsa::WriteMixedPcm(int frames) {
   DCHECK(mixer_task_runner_->BelongsToCurrentThread());
   CHECK_PCM_INITIALIZED();
-  DCHECK(interleaved);
-  DCHECK_GE(interleaved->size(), InterleavedSize(frames));
+
+  // Resize interleaved if necessary.
+  size_t interleaved_size = static_cast<size_t>(frames) * kNumOutputChannels *
+                            BytesPerOutputFormatSample();
+  if (interleaved_.size() < interleaved_size) {
+    interleaved_.resize(interleaved_size);
+  }
+
+  // Get data for loopback.
+  mix_filter_->data()->ToInterleaved(frames, BytesPerOutputFormatSample(),
+                                     interleaved_.data());
 
   int64_t expected_playback_time;
-  if (rendering_delay_.timestamp_microseconds == kNoTimestamp) {
+  if (alsa_rendering_delay_.timestamp_microseconds == kNoTimestamp) {
     expected_playback_time = kNoTimestamp;
   } else {
-    expected_playback_time = rendering_delay_.timestamp_microseconds +
-                             rendering_delay_.delay_microseconds;
+    expected_playback_time = alsa_rendering_delay_.timestamp_microseconds +
+                             alsa_rendering_delay_.delay_microseconds +
+                             linearize_filter_->GetRenderingDelayMicroseconds();
   }
 
   for (CastMediaShlib::LoopbackAudioObserver* observer : loopback_observers_) {
     observer->OnLoopbackAudio(expected_playback_time, kSampleFormatS32,
                               output_samples_per_second_, kNumOutputChannels,
-                              interleaved->data(), InterleavedSize(frames));
+                              interleaved_.data(), InterleavedSize(frames));
   }
 
+  // Get data for playout.
+  linearize_filter_->data()->ToInterleaved(frames, BytesPerOutputFormatSample(),
+                                           interleaved_.data());
+
   // If the PCM has been drained it will be in SND_PCM_STATE_SETUP and need
   // to be prepared in order for playback to work.
   if (alsa_->PcmState(pcm_) == SND_PCM_STATE_SETUP) {
     RETURN_REPORT_ERROR(PcmPrepare, pcm_);
   }
 
   int frames_left = frames;
-  uint8_t* data = interleaved->data();
+  uint8_t* data = interleaved_.data();
   while (frames_left) {
     int frames_or_error;
     while ((frames_or_error = alsa_->PcmWritei(pcm_, data, frames_left)) < 0) {
       for (auto* observer : loopback_observers_) {
         observer->OnLoopbackInterrupted();
       }
       RETURN_REPORT_ERROR(PcmRecover, pcm_, frames_or_error,
                           kPcmRecoverIsSilent);
     }
     frames_left -= frames_or_error;
     DCHECK_GE(frames_left, 0);
     data += frames_or_error * kNumOutputChannels * BytesPerOutputFormatSample();
   }
   UpdateRenderingDelay(frames);
-  for (auto&& input : inputs_)
-    input->AfterWriteFrames(rendering_delay_);
+  MediaPipelineBackendAlsa::RenderingDelay common_rendering_delay =
+      alsa_rendering_delay_;
+  common_rendering_delay.delay_microseconds +=
+      linearize_filter_->GetRenderingDelayMicroseconds() +
+      mix_filter_->GetRenderingDelayMicroseconds();
+  for (auto&& input : inputs_) {
+    MediaPipelineBackendAlsa::RenderingDelay stream_rendering_delay =
+        common_rendering_delay;
+    stream_rendering_delay.delay_microseconds +=
+        input->filter_group()->GetRenderingDelayMicroseconds();
+    input->AfterWriteFrames(stream_rendering_delay);
+  }
 }
 
 void StreamMixerAlsa::UpdateRenderingDelay(int newly_pushed_frames) {
   DCHECK(mixer_task_runner_->BelongsToCurrentThread());
   CHECK_PCM_INITIALIZED();
 
   // TODO(bshaya): Add rendering delay from post-processors.
   if (alsa_->PcmStatus(pcm_, pcm_status_) != 0 ||
       alsa_->PcmStatusGetState(pcm_status_) != SND_PCM_STATE_RUNNING) {
-    rendering_delay_.timestamp_microseconds = kNoTimestamp;
-    rendering_delay_.delay_microseconds = 0;
+    alsa_rendering_delay_.timestamp_microseconds = kNoTimestamp;
+    alsa_rendering_delay_.delay_microseconds = 0;
     return;
   }
 
   snd_htimestamp_t status_timestamp = {};
   alsa_->PcmStatusGetHtstamp(pcm_status_, &status_timestamp);
-  rendering_delay_.timestamp_microseconds =
+  alsa_rendering_delay_.timestamp_microseconds =
       TimespecToMicroseconds(status_timestamp);
   snd_pcm_sframes_t delay_frames = alsa_->PcmStatusGetDelay(pcm_status_);
-  rendering_delay_.delay_microseconds = static_cast<int64_t>(delay_frames) *
-                                        base::Time::kMicrosecondsPerSecond /
-                                        output_samples_per_second_;
+  alsa_rendering_delay_.delay_microseconds =
+      static_cast<int64_t>(delay_frames) * base::Time::kMicrosecondsPerSecond /
+      output_samples_per_second_;
 }
 
 void StreamMixerAlsa::AddLoopbackAudioObserver(
     CastMediaShlib::LoopbackAudioObserver* observer) {
   RUN_ON_MIXER_THREAD(&StreamMixerAlsa::AddLoopbackAudioObserver, observer);
   DCHECK(observer);
   DCHECK(std::find(loopback_observers_.begin(), loopback_observers_.end(),
                    observer) == loopback_observers_.end());
   loopback_observers_.push_back(observer);
 }
@@ skipping 52 matching lines @@
   for (auto&& input : inputs_) {
     // Volume limits don't apply to effects streams.
     if (input->primary() && input->content_type() == type) {
       input->SetContentTypeVolume(effective_volume, fade_ms);
     }
   }
 }
 
 }  // namespace media
 }  // namespace chromecast