services/audio/input_stream.cc

// Copyright 2018 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include "services/audio/input_stream.h"

#include <inttypes.h>

#include <string>
#include <utility>

#include "base/functional/bind.h"
#include "base/functional/callback_helpers.h"
#include "base/memory/raw_ptr.h"
#include "base/strings/strcat.h"
#include "base/strings/stringprintf.h"
#include "base/strings/to_string.h"
#include "base/task/sequenced_task_runner.h"
#include "base/time/time.h"
#include "base/trace_event/trace_event.h"
#include "media/audio/audio_manager.h"
#include "media/base/audio_parameters.h"
#include "media/mojo/mojom/audio_processing.mojom.h"
#include "mojo/public/cpp/system/buffer.h"
#include "mojo/public/cpp/system/handle.h"
#include "mojo/public/cpp/system/platform_handle.h"
#include "services/audio/input_sync_writer.h"
#include "services/audio/reference_signal_provider.h"
#include "third_party/perfetto/include/perfetto/tracing/track.h"

namespace audio {

namespace {
const int kMaxInputChannels = 3;

using InputStreamErrorCode = media::mojom::InputStreamErrorCode;
using DisconnectReason =
    media::mojom::AudioInputStreamObserver::DisconnectReason;

const char* ErrorCodeToString(InputController::ErrorCode error) {
  switch (error) {
    case (InputController::STREAM_CREATE_ERROR):
      return "STREAM_CREATE_ERROR";
    case (InputController::STREAM_OPEN_ERROR):
      return "STREAM_OPEN_ERROR";
    case (InputController::STREAM_ERROR):
      return "STREAM_ERROR";
    case (InputController::STREAM_OPEN_SYSTEM_PERMISSIONS_ERROR):
      return "STREAM_OPEN_SYSTEM_PERMISSIONS_ERROR";
    case (InputController::STREAM_OPEN_DEVICE_IN_USE_ERROR):
      return "STREAM_OPEN_DEVICE_IN_USE_ERROR";
    case (InputController::REFERENCE_STREAM_ERROR):
      return "REFERENCE_STREAM_ERROR";
    case (InputController::REFERENCE_STREAM_CREATE_ERROR):
      return "REFERENCE_STREAM_CREATE_ERROR";
    case (InputController::REFERENCE_STREAM_OPEN_ERROR):
      return "REFERENCE_STREAM_OPEN_ERROR";
    case (InputController::REFERENCE_STREAM_OPEN_SYSTEM_PERMISSIONS_ERROR):
      return "REFERENCE_STREAM_OPEN_SYSTEM_PERMISSIONS_ERROR";
    case (InputController::REFERENCE_STREAM_OPEN_DEVICE_IN_USE_ERROR):
      return "REFERENCE_STREAM_OPEN_DEVICE_IN_USE_ERROR";
    default:
      NOTREACHED();
  }
}

std::string GetCtorLogString(const std::string& device_id,
                             const media::AudioParameters& params,
                             bool enable_agc) {
  std::string str = base::StringPrintf("Ctor(");
  base::StringAppendF(&str, "{device_id=%s}, ", device_id.c_str());
  base::StringAppendF(&str, "{params=[%s]}, ",
                      params.AsHumanReadableString().c_str());
  base::StringAppendF(&str, "{enable_agc=%d})", enable_agc);
  return str;
}

}  // namespace

InputStream::InputStream(
    CreatedCallback created_callback,
    DeleteCallback delete_callback,
    mojo::PendingReceiver<media::mojom::AudioInputStream> receiver,
    mojo::PendingRemote<media::mojom::AudioInputStreamClient> client,
    mojo::PendingRemote<media::mojom::AudioInputStreamObserver> observer,
    mojo::SharedRemote<media::mojom::AudioLog> log,
    media::AudioManager* audio_manager,
    media::AecdumpRecordingManager* aecdump_recording_manager,
    raw_ptr<MlModelManager> ml_model_manager,
    std::unique_ptr<ReferenceSignalProvider> reference_signal_provider,
    media::mojom::AudioProcessingConfigPtr processing_config,
    LoopbackMixin::MaybeCreateCallback maybe_create_loopback_mixin_cb,
    const std::string& device_id,
    const media::AudioParameters& params,
    uint32_t shared_memory_count,
    bool enable_agc)
    : id_(base::UnguessableToken::Create()),
      receiver_(this, std::move(receiver)),
      client_(std::move(client)),
      observer_(std::move(observer)),
      log_(std::move(log)),
      created_callback_(std::move(created_callback)),
      delete_callback_(std::move(delete_callback)),
      foreign_socket_(),
      writer_(InputSyncWriter::Create(
          log_ ? base::BindRepeating(&media::mojom::AudioLog::OnLogMessage,
                                     base::Unretained(log_.get()))
               : base::DoNothing(),
          shared_memory_count,
          params,
          &foreign_socket_)) {
  DCHECK(audio_manager);
  DCHECK(receiver_.is_bound());
  DCHECK(client_);
  DCHECK(created_callback_);
  DCHECK(delete_callback_);
  DCHECK(params.IsValid());
  const base::TimeTicks start_time = base::TimeTicks::Now();
  TRACE_EVENT_BEGIN("audio", "audio::InputStream",
                    perfetto::Track::FromPointer(this));
  TRACE_EVENT_BEGIN("audio", "InputStream", perfetto::Track::FromPointer(this),
                    "device id", device_id, "params",
                    params.AsHumanReadableString());
  SendLogMessage(GetCtorLogString(device_id, params, enable_agc));

  // |this| owns these objects, so unretained is safe.
  base::RepeatingClosure error_handler =
      base::BindRepeating(&InputStream::OnStreamError, base::Unretained(this),
                          std::optional<DisconnectReason>());
  receiver_.set_disconnect_handler(error_handler);
  client_.set_disconnect_handler(error_handler);

  if (observer_)
    observer_.set_disconnect_handler(std::move(error_handler));

  if (log_)
    log_->OnCreated(params, device_id);

  // Only MONO, STEREO and STEREO_AND_KEYBOARD_MIC channel layouts are expected,
  // see AudioManagerBase::MakeAudioInputStream().
  if (params.channels() > kMaxInputChannels) {
    OnStreamPlatformError();
    return;
  }

  if (!writer_) {
    OnStreamPlatformError();
    return;
  }

  controller_ = InputController::Create(
      audio_manager, this, writer_.get(), std::move(reference_signal_provider),
      aecdump_recording_manager, ml_model_manager, std::move(processing_config),
      std::move(maybe_create_loopback_mixin_cb), params, device_id, enable_agc);
  SendLogMessage(base::StringPrintf(
      "Create => (duration=%" PRId64 " ms)",
      (base::TimeTicks::Now() - start_time).InMilliseconds()));
}

InputStream::~InputStream() {
  DCHECK_CALLED_ON_VALID_SEQUENCE(owning_sequence_);
  const base::TimeTicks start_time = base::TimeTicks::Now();

  if (log_)
    log_->OnClosed();

  if (observer_) {
    observer_.ResetWithReason(
        static_cast<uint32_t>(DisconnectReason::kTerminatedByClient),
        std::string());
  }

  if (created_callback_) {
    // Didn't manage to create the stream. Call the callback anyways as mandated
    // by mojo.
    std::move(created_callback_).Run(nullptr, false, std::nullopt);
  }

  if (controller_) {
    // TODO(crbug.com/40558532): remove InputController::Close() after
    // content/ streams are removed, destructor should suffice.
    controller_->Close();
  }

  TRACE_EVENT_END("audio",
                  /* InputStream */ perfetto::Track::FromPointer(this));
  TRACE_EVENT_END("audio",
                  /* audio::InputStream */ perfetto::Track::FromPointer(this));

  SendLogMessage(base::StringPrintf(
      "Dtor => (completed, duration=%" PRId64 " ms)",
      (base::TimeTicks::Now() - start_time).InMilliseconds()));
}

void InputStream::SetOutputDeviceForAec(const std::string& output_device_id) {
  DCHECK_CALLED_ON_VALID_SEQUENCE(owning_sequence_);
  DCHECK(controller_);
  controller_->SetOutputDeviceForAec(output_device_id);
  SendLogMessage(base::StringPrintf("%s({output_device_id=%s})", __func__,
                                    output_device_id.c_str()));
}

void InputStream::Record() {
  DCHECK_CALLED_ON_VALID_SEQUENCE(owning_sequence_);
  DCHECK(controller_);
  TRACE_EVENT_INSTANT("audio", "Record", perfetto::Track::FromPointer(this));
  const base::TimeTicks start_time = base::TimeTicks::Now();
  SendLogMessage(base::StringPrintf("%s()", __func__));
  controller_->Record();
  if (observer_)
    observer_->DidStartRecording();
  if (log_)
    log_->OnStarted();
  SendLogMessage(base::StringPrintf(
      "%s() => (duration=%" PRId64 " ms)", __func__,
      (base::TimeTicks::Now() - start_time).InMilliseconds()));
}

void InputStream::SetVolume(double volume) {
  DCHECK_CALLED_ON_VALID_SEQUENCE(owning_sequence_);
  DCHECK(controller_);
  TRACE_EVENT_INSTANT("audio", "SetVolume", perfetto::Track::FromPointer(this),
                      "volume", volume);
  const base::TimeTicks start_time = base::TimeTicks::Now();
  SendLogMessage(base::StringPrintf("%s({volume=%.2f})", __func__, volume));

  if (volume < 0 || volume > 1) {
    receiver_.ReportBadMessage("Invalid volume");
    OnStreamPlatformError();
    return;
  }

  controller_->SetVolume(volume);
  if (log_)
    log_->OnSetVolume(volume);
  SendLogMessage(base::StringPrintf(
      "%s({volume=%.2f}) => (duration=%" PRId64 " ms)", __func__, volume,
      (base::TimeTicks::Now() - start_time).InMilliseconds()));
}

void InputStream::OnCreated(bool initially_muted) {
  TRACE_EVENT_INSTANT("audio", "Created", perfetto::Track::FromPointer(this),
                      "initially muted", initially_muted);
  DCHECK_CALLED_ON_VALID_SEQUENCE(owning_sequence_);
  const base::TimeTicks start_time = base::TimeTicks::Now();
  SendLogMessage(base::StringPrintf("%s({muted=%s})", __func__,
                                    base::ToString(initially_muted).c_str()));

  base::UnsafeSharedMemoryRegion shared_memory_region =
      writer_->TakeSharedMemoryRegion();
  if (!shared_memory_region.IsValid()) {
    OnStreamPlatformError();
    return;
  }

  mojo::PlatformHandle socket_handle(foreign_socket_.Take());
  DCHECK(socket_handle.is_valid());

  std::move(created_callback_)
      .Run({std::in_place, std::move(shared_memory_region),
            std::move(socket_handle)},
           initially_muted, id_);
  SendLogMessage(base::StringPrintf(
      "%s({muted=%s}) => (completed, duration=%" PRId64 " ms)", __func__,
      base::ToString(initially_muted).c_str(),
      (base::TimeTicks::Now() - start_time).InMilliseconds()));
}

DisconnectReason InputErrorToDisconnectReason(InputController::ErrorCode code) {
  switch (code) {
    case InputController::STREAM_OPEN_SYSTEM_PERMISSIONS_ERROR:
      return DisconnectReason::kSystemPermissions;
    case InputController::STREAM_OPEN_DEVICE_IN_USE_ERROR:
      return DisconnectReason::kDeviceInUse;
    default:
      break;
  }
  return DisconnectReason::kPlatformError;
}

InputStreamErrorCode InputControllerErrorToStreamError(
    InputController::ErrorCode code) {
  switch (code) {
    case InputController::STREAM_OPEN_SYSTEM_PERMISSIONS_ERROR:
      return InputStreamErrorCode::kSystemPermissions;
    case InputController::STREAM_OPEN_DEVICE_IN_USE_ERROR:
      return InputStreamErrorCode::kDeviceInUse;
    default:
      break;
  }
  return InputStreamErrorCode::kUnknown;
}

void InputStream::OnError(InputController::ErrorCode error_code) {
  DCHECK_CALLED_ON_VALID_SEQUENCE(owning_sequence_);
  TRACE_EVENT_INSTANT("audio", "Error", perfetto::Track::FromPointer(this));

  client_->OnError(InputControllerErrorToStreamError(error_code));
  if (log_)
    log_->OnError();
  SendLogMessage(base::StringPrintf("%s({error_code=%s})", __func__,
                                    ErrorCodeToString(error_code)));
  OnStreamError(InputErrorToDisconnectReason(error_code));
}

void InputStream::OnLog(std::string_view message) {
  // No sequence check: |log_| is thread-safe and id_ is const.
  if (log_)
    log_->OnLogMessage(std::string(message) + " [id=" + id_.ToString() + "]");
}

void InputStream::OnMuted(bool is_muted) {
  DCHECK_CALLED_ON_VALID_SEQUENCE(owning_sequence_);
  client_->OnMutedStateChanged(is_muted);
}

void InputStream::OnStreamPlatformError() {
  OnStreamError(DisconnectReason::kPlatformError);
}

void InputStream::OnStreamError(
    std::optional<DisconnectReason> reason_to_report) {
  DCHECK_CALLED_ON_VALID_SEQUENCE(owning_sequence_);
  TRACE_EVENT_INSTANT("audio", "OnStreamError",
                      perfetto::Track::FromPointer(this));

  if (reason_to_report.has_value()) {
    if (observer_) {
      observer_.ResetWithReason(static_cast<uint32_t>(reason_to_report.value()),
                                std::string());
    }
    SendLogMessage(base::StringPrintf("%s()", __func__));
  }

  // Defer callback so we're not destructed while in the constructor.
  base::SequencedTaskRunner::GetCurrentDefault()->PostTask(
      FROM_HERE,
      base::BindOnce(&InputStream::CallDeleter, weak_factory_.GetWeakPtr()));
  receiver_.reset();
}

void InputStream::CallDeleter() {
  DCHECK_CALLED_ON_VALID_SEQUENCE(owning_sequence_);

  std::move(delete_callback_).Run(this);
}

void InputStream::SendLogMessage(const std::string& message) {
  DCHECK_CALLED_ON_VALID_SEQUENCE(owning_sequence_);
  if (!log_) {
    return;
  }
  log_->OnLogMessage("audio::IS::" + message +
                     base::StringPrintf(" [id=%s]", id_.ToString().c_str()));
}

}  // namespace audio