Test submission of complete AEC-dump refactoring.

webrtc/modules/audio_processing/aec_dumper/aec_dumper_impl.cc

+/*
+ *  Copyright (c) 2017 The WebRTC project authors. All Rights Reserved.
+ *
+ *  Use of this source code is governed by a BSD-style license
+ *  that can be found in the LICENSE file in the root of the source
+ *  tree. An additional intellectual property rights grant can be found
+ *  in the file PATENTS.  All contributing project authors may
+ *  be found in the AUTHORS file in the root of the source tree.
+ */
+
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "webrtc/modules/audio_processing/aec_dumper/aec_dumper.h"
+
+#include "webrtc/base/checks.h"
+#include "webrtc/base/event.h"
+#include "webrtc/base/ignore_wundef.h"
+#include "webrtc/base/task_queue.h"
+#include "webrtc/base/thread_checker.h"
+#include "webrtc/modules/audio_processing/aec_dumper/capture_stream_info_impl.h"
+#include "webrtc/modules/include/module_common_types.h"
+#include "webrtc/system_wrappers/include/file_wrapper.h"
+
+// Files generated at build-time by the protobuf compiler.
+RTC_PUSH_IGNORING_WUNDEF()
+#ifdef WEBRTC_ANDROID_PLATFORM_BUILD
+#include "external/webrtc/webrtc/modules/audio_processing/debug.pb.h"
+#else
+#include "webrtc/modules/audio_processing/debug.pb.h"
+#endif
+RTC_POP_IGNORING_WUNDEF()
+
+namespace webrtc {
+
+namespace {
+
+// Task-queue based implementation of AecDumper. It is thread safe by
+// relying on locks in TaskQueue.
+class AecDumperImpl : public AecDumper {
+ public:
+  AecDumperImpl(std::string file_name,
+                int64_t max_log_size_bytes,
+                rtc::TaskQueue* worker_queue);
+  AecDumperImpl(FILE* handle,
+                int64_t max_log_size_bytes,
+                rtc::TaskQueue* worker_queue);
+  ~AecDumperImpl() override;
+
+  std::unique_ptr<CaptureStreamInfo> GetCaptureStreamInfo() override;
+
+  void WriteInitMessage(const ProcessingConfig& api_format) override;
+  void WriteReverseStreamMessage(const AudioFrame& frame) override;
+  void WriteReverseStreamMessage(
+      const std::vector<rtc::ArrayView<const float>> src) override;
+  void WriteCaptureStreamMessage(
+      std::unique_ptr<CaptureStreamInfo> capture_stream_info) override;
+  void WriteConfig(InternalAPMConfig config, bool forced) override;
+
+ private:
+  void PostTask(std::unique_ptr<audioproc::Event> event);
+
+  // Implementation detail of WriteConfig: If not |forced|, only
+  // writes the current config if it is different from the last saved
+  // one; if |forced|, writes the config regardless of the last saved.
+  std::string last_serialized_capture_config_ GUARDED_BY(config_string_lock_) =
+      "";
+  std::unique_ptr<FileWrapper> debug_file_;
+  int64_t num_bytes_left_for_log_ = 0;
+
+  rtc::TaskQueue* worker_queue_;
+  rtc::CriticalSection config_string_lock_;
+};
+
+class WriteToFileTask : public rtc::QueuedTask {
+ public:
+  WriteToFileTask(webrtc::FileWrapper* debug_file,
+                  std::unique_ptr<audioproc::Event> event,
+                  int64_t* num_bytes_left_for_log)
+      : debug_file_(debug_file),
+        event_(std::move(event)),
+        num_bytes_left_for_log_(num_bytes_left_for_log) {}
+
+ private:
+  bool IsRoomForNextEvent(size_t event_byte_size) const {
+    int64_t next_message_size = event_byte_size + sizeof(int32_t);
+    return (*num_bytes_left_for_log_ < 0) ||
+           (*num_bytes_left_for_log_ >= next_message_size);
+  }
+
+  void UpdateBytesLeft(size_t event_byte_size) {
+    RTC_DCHECK(IsRoomForNextEvent(event_byte_size));
+    if (*num_bytes_left_for_log_ >= 0) {
+      *num_bytes_left_for_log_ -= (sizeof(int32_t) + event_byte_size);
+    }
+  }
+
+  bool Run() override {
+    if (!debug_file_->is_open()) {
+      return true;
+    }
+
+    std::string event_string;
+    event_->SerializeToString(&event_string);
+
+    const size_t event_byte_size = event_->ByteSize();
+
+    if (!IsRoomForNextEvent(event_byte_size)) {
+      debug_file_->CloseFile();
+      return true;
+    }
+
+    UpdateBytesLeft(event_byte_size);
+
+    // Write message preceded by its size.
+    if (!debug_file_->Write(&event_byte_size, sizeof(int32_t))) {
+      RTC_NOTREACHED();
+    }
+    if (!debug_file_->Write(event_string.data(), event_string.length())) {
+      RTC_NOTREACHED();
+    }
+    return true;  // Delete task from queue at once. TODO(aleloi):
+                  // instead consider a 'mega-task' that returns
+                  // 'false', checks if there is something in a
+                  // swap-queue and reposts itself periodically.
+  }
+
+  webrtc::FileWrapper* debug_file_;
+  std::unique_ptr<audioproc::Event> event_;
+  int64_t* num_bytes_left_for_log_;
+};
+
+AecDumperImpl::AecDumperImpl(std::string file_name,
+                             int64_t max_log_size_bytes,
+                             rtc::TaskQueue* worker_queue)
+    : debug_file_(FileWrapper::Create()), worker_queue_(worker_queue) {
+  RTC_DCHECK(debug_file_);
+  worker_queue_->PostTask([this, file_name, max_log_size_bytes]() {
+    num_bytes_left_for_log_ = max_log_size_bytes;
+    debug_file_->OpenFile(file_name.c_str(), false);
+  });
+}
+
+AecDumperImpl::AecDumperImpl(FILE* handle,
+                             int64_t max_log_size_bytes,
+                             rtc::TaskQueue* worker_queue)
+    : debug_file_(FileWrapper::Create()), worker_queue_(worker_queue) {
+  RTC_DCHECK(debug_file_);
+  worker_queue_->PostTask([this, handle, max_log_size_bytes]() {
+    num_bytes_left_for_log_ = max_log_size_bytes;
+    debug_file_->OpenFromFileHandle(handle);
+  });
+}
+
+AecDumperImpl::~AecDumperImpl() {
+  // Block until all tasks have finished running.
+  rtc::Event thread_sync_event(false /* manual_reset */, false);
+  worker_queue_->PostTask([&thread_sync_event] { thread_sync_event.Set(); });
+  thread_sync_event.Wait(rtc::Event::kForever);
+}
+
+std::unique_ptr<AecDumper::CaptureStreamInfo>
+AecDumperImpl::GetCaptureStreamInfo() {
+  return std::unique_ptr<CaptureStreamInfoImpl>(new CaptureStreamInfoImpl(
+      std::unique_ptr<audioproc::Event>(new audioproc::Event())));
+}
+
+void AecDumperImpl::WriteInitMessage(const ProcessingConfig& api_format) {
+  auto event = std::unique_ptr<audioproc::Event>(new audioproc::Event());
+  event->set_type(audioproc::Event::INIT);
+  audioproc::Init* msg = event->mutable_init();
+
+  msg->set_sample_rate(api_format.input_stream().sample_rate_hz());
+  msg->set_num_input_channels(static_cast<google::protobuf::int32>(
+      api_format.input_stream().num_channels()));
+  msg->set_num_output_channels(static_cast<google::protobuf::int32>(
+      api_format.output_stream().num_channels()));
+  msg->set_num_reverse_channels(static_cast<google::protobuf::int32>(
+      api_format.reverse_input_stream().num_channels()));
+  msg->set_reverse_sample_rate(
+      api_format.reverse_input_stream().sample_rate_hz());
+  msg->set_output_sample_rate(api_format.output_stream().sample_rate_hz());
+  msg->set_reverse_output_sample_rate(
+      api_format.reverse_output_stream().sample_rate_hz());
+  msg->set_num_reverse_output_channels(
+      api_format.reverse_output_stream().num_channels());
+
+  PostTask(std::move(event));
+}
+
+void AecDumperImpl::WriteReverseStreamMessage(const AudioFrame& frame) {
+  auto event = std::unique_ptr<audioproc::Event>(new audioproc::Event());
+
+  event->set_type(audioproc::Event::REVERSE_STREAM);
+  audioproc::ReverseStream* msg = event->mutable_reverse_stream();
+  const size_t data_size =
+      sizeof(int16_t) * frame.samples_per_channel_ * frame.num_channels_;
+  msg->set_data(frame.data_, data_size);
+
+  PostTask(std::move(event));
+}
+
+void AecDumperImpl::WriteReverseStreamMessage(
+    const std::vector<rtc::ArrayView<const float>> src) {
+  auto event = std::unique_ptr<audioproc::Event>(new audioproc::Event());
+  event->set_type(audioproc::Event::REVERSE_STREAM);
+
+  audioproc::ReverseStream* msg = event->mutable_reverse_stream();
+
+  for (const auto& frame : src) {
+    msg->add_channel(frame.begin(), frame.size());
+  }
+
+  PostTask(std::move(event));
+}
+
+void AecDumperImpl::WriteCaptureStreamMessage(
+    std::unique_ptr<CaptureStreamInfo> capture_stream_info) {
+  // Really ugly, how is it done better?
+  auto event_ptr =
+      static_cast<CaptureStreamInfoImpl*>(capture_stream_info.get())
+          ->GetEventMsg();
+  if (event_ptr) {
+    PostTask(std::move(event_ptr));
+  }
+}
+
+void CopyFromConfigToEvent(webrtc::InternalAPMConfig config,
+                           webrtc::audioproc::Config* pb_cfg) {
+  pb_cfg->set_aec_enabled(config.aec_enabled);
+  pb_cfg->set_aec_delay_agnostic_enabled(config.aec_delay_agnostic_enabled);
+  pb_cfg->set_aec_drift_compensation_enabled(
+      config.aec_drift_compensation_enabled);
+  pb_cfg->set_aec_extended_filter_enabled(config.aec_extended_filter_enabled);
+  pb_cfg->set_aec_suppression_level(config.aec_suppression_level);
+
+  pb_cfg->set_aecm_enabled(config.aecm_enabled);
+  pb_cfg->set_aecm_comfort_noise_enabled(config.aecm_comfort_noise_enabled);
+  pb_cfg->set_aecm_routing_mode(config.aecm_routing_mode);
+
+  pb_cfg->set_agc_enabled(config.agc_enabled);
+  pb_cfg->set_agc_mode(config.agc_mode);
+  pb_cfg->set_agc_limiter_enabled(config.agc_limiter_enabled);
+  pb_cfg->set_noise_robust_agc_enabled(config.noise_robust_agc_enabled);
+
+  pb_cfg->set_hpf_enabled(config.hpf_enabled);
+
+  pb_cfg->set_ns_enabled(config.ns_enabled);
+  pb_cfg->set_ns_level(config.ns_level);
+
+  pb_cfg->set_transient_suppression_enabled(
+      config.transient_suppression_enabled);
+  pb_cfg->set_intelligibility_enhancer_enabled(
+      config.intelligibility_enhancer_enabled);
+
+  pb_cfg->set_experiments_description(config.experiments_description);
+}
+
+void AecDumperImpl::WriteConfig(InternalAPMConfig config, bool forced) {
+  auto event = std::unique_ptr<audioproc::Event>(new audioproc::Event());
+  event->set_type(audioproc::Event::CONFIG);
+  CopyFromConfigToEvent(config, event->mutable_config());
+
+  std::string serialized_config = event->mutable_config()->SerializeAsString();
+  {
+    rtc::CritScope cs(&config_string_lock_);
+    if (!forced && serialized_config == last_serialized_capture_config_) {
+      return;
+    }
+    last_serialized_capture_config_ = serialized_config;
+  }
+
+  PostTask(std::move(event));
+}
+
+void AecDumperImpl::PostTask(std::unique_ptr<audioproc::Event> event) {
+  RTC_DCHECK(event);
+  worker_queue_->PostTask(std::unique_ptr<rtc::QueuedTask>(new WriteToFileTask(
+      debug_file_.get(), std::move(event), &num_bytes_left_for_log_)));
+}
+}  // namespace
+
+std::unique_ptr<AecDumper> AecDumper::Create(std::string file_name,
+                                             int64_t max_log_size_bytes,
+                                             rtc::TaskQueue* worker_queue) {
+  return std::unique_ptr<AecDumperImpl>(
+      new AecDumperImpl(file_name, max_log_size_bytes, worker_queue));
+}
+
+std::unique_ptr<AecDumper> AecDumper::Create(FILE* handle,
+                                             int64_t max_log_size_bytes,
+                                             rtc::TaskQueue* worker_queue) {
+  return std::unique_ptr<AecDumperImpl>(
+      new AecDumperImpl(handle, max_log_size_bytes, worker_queue));
+}
+}  // namespace webrtc