// Copyright 2016 The Fuchsia 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 "src/media/audio/audio_core/packet_queue.h"

#include <iomanip>

#include <trace/event.h>

#include "src/lib/syslog/cpp/logger.h"
#include "src/media/audio/audio_core/audio_object.h"
#include "src/media/audio/audio_core/format.h"
#include "src/media/audio/lib/logging/logging.h"

namespace media::audio {
namespace {

// To what extent should client-side underflows be logged? (A "client-side underflow" refers to when
// all or part of a packet's data is discarded because its start timestamp has already passed.)
// For each packet queue, we will log the first underflow. For subsequent occurrences, depending on
// audio_core's logging level, we throttle how frequently these are displayed. If log_level is set
// to TRACE or SPEW, all client-side underflows are logged -- at log_level -1: VLOG TRACE -- as
// specified by kUnderflowTraceInterval. If set to INFO, we log less often, at log_level 1: INFO,
// throttling by the factor kUnderflowInfoInterval. If set to WARNING or higher, we throttle these
// even more, specified by kUnderflowErrorInterval. Note: by default we set NDEBUG builds to WARNING
// and DEBUG builds to INFO. To disable all logging of client-side underflows, set kLogUnderflow to
// false.
static constexpr bool kLogUnderflow = true;
static constexpr uint16_t kUnderflowTraceInterval = 1;
static constexpr uint16_t kUnderflowInfoInterval = 10;
static constexpr uint16_t kUnderflowErrorInterval = 100;

}  // namespace

PacketQueue::PacketQueue(Format format) : PacketQueue(format, nullptr) {}

PacketQueue::PacketQueue(Format format, fbl::RefPtr<VersionedTimelineFunction> timeline_function)
    : ReadableStream(std::move(format)), timeline_function_(std::move(timeline_function)) {}

PacketQueue::~PacketQueue() {
  pending_flush_packet_queue_.clear();
  pending_packet_queue_.clear();
  pending_flush_token_queue_.clear();
}

void PacketQueue::PushPacket(const fbl::RefPtr<Packet>& packet) {
  TRACE_DURATION("audio", "PacketQueue::PushPacket");
  std::lock_guard<std::mutex> locker(pending_mutex_);
  pending_packet_queue_.emplace_back(std::move(packet));
}

void PacketQueue::Flush(const fbl::RefPtr<PendingFlushToken>& flush_token) {
  TRACE_DURATION("audio", "PacketQueue::Flush");
  std::deque<fbl::RefPtr<Packet>> flushed_packets;

  {
    std::lock_guard<std::mutex> locker(pending_mutex_);

    flushed_ = true;

    if (processing_in_progress_) {
      // Is the sink currently mixing?  If so, the flush cannot complete until the mix operation has
      // finished.  Move the 'waiting to be rendered' packets to the back of the 'waiting to be
      // flushed queue', and append our flush token (if any) to the pending flush token queue.  The
      // sink's thread will take are of releasing these objects back to the service thread for
      // cleanup when it has finished it's current job.
      for (auto& packet : pending_packet_queue_) {
        pending_flush_packet_queue_.emplace_back(std::move(packet));
      }
      pending_packet_queue_.clear();

      if (flush_token != nullptr) {
        pending_flush_token_queue_.emplace_back(std::move(flush_token));
      }

      return;
    } else {
      // If the sink is not currently mixing, then we just swap the contents the pending packet
      // queues with out local queue and release the packets in the proper order once we have left
      // the pending mutex lock.
      FX_DCHECK(pending_flush_packet_queue_.empty());
      FX_DCHECK(pending_flush_token_queue_.empty());
      flushed_packets.swap(pending_packet_queue_);
    }
  }

  // Release the packets, front to back.
  for (auto& ptr : flushed_packets) {
    ptr = nullptr;
  }
}

std::optional<ReadableStream::Buffer> PacketQueue::ReadLock(zx::time now, int64_t frame,
                                                            uint32_t frame_count) {
  TRACE_DURATION("audio", "PacketQueue::ReadLock");
  std::lock_guard<std::mutex> locker(pending_mutex_);

  FX_DCHECK(!processing_in_progress_);
  if (!pending_packet_queue_.size()) {
    return std::nullopt;
  }

  processing_in_progress_ = true;
  auto& packet = pending_packet_queue_.front();
  bool is_continuous = !flushed_;
  flushed_ = false;
  return std::make_optional<ReadableStream::Buffer>(
      packet->start(), packet->length(), packet->payload(), is_continuous,
      [this](bool fully_consumed) { this->ReadUnlock(fully_consumed); });
}

void PacketQueue::ReadUnlock(bool fully_consumed) {
  TRACE_DURATION("audio", "PacketQueue::ReadUnlock");
  std::lock_guard<std::mutex> locker(pending_mutex_);

  FX_DCHECK(processing_in_progress_);
  processing_in_progress_ = false;

  // Did a flush take place while we were working?  If so release each of the packets waiting to
  // be flushed back to the service thread, then release each of the flush tokens.
  if (!pending_flush_packet_queue_.empty() || !pending_flush_token_queue_.empty()) {
    for (auto& ptr : pending_flush_packet_queue_) {
      ptr = nullptr;
    }

    for (auto& ptr : pending_flush_token_queue_) {
      ptr = nullptr;
    }

    pending_flush_packet_queue_.clear();
    pending_flush_token_queue_.clear();
    return;
  }

  // If the buffer was fully consumed, release the first packet. The queue must not be empty,
  // unless the queue was flushed between ReadLock and ReadUnlock, but that case is handled above.
  if (fully_consumed) {
    FX_DCHECK(!pending_packet_queue_.empty());
    pending_packet_queue_.pop_front();
  }
}

void PacketQueue::Trim(zx::time ref_time) {
  TRACE_DURATION("audio", "PacketQueue::Trim");
  int64_t local_now_ticks = (ref_time - zx::time(0)).to_nsecs();
  auto trim_threshold =
      FractionalFrames<int64_t>::FromRaw(timeline_function_->get().first(local_now_ticks));

  std::lock_guard<std::mutex> locker(pending_mutex_);
  while (!pending_packet_queue_.empty()) {
    auto packet = pending_packet_queue_.front();

    if (packet->end() > trim_threshold) {
      return;
    }
    pending_packet_queue_.pop_front();
  }
}

BaseStream::TimelineFunctionSnapshot PacketQueue::ReferenceClockToFractionalFrames() const {
  if (!timeline_function_) {
    return {
        .timeline_function = TimelineFunction(),
        .generation = kInvalidGenerationId,
    };
  }
  auto [timeline_function, generation] = timeline_function_->get();
  return {
      .timeline_function = timeline_function,
      .generation = generation,
  };
}

void PacketQueue::ReportUnderflow(FractionalFrames<int64_t> frac_source_start,
                                  FractionalFrames<int64_t> frac_source_mix_point,
                                  zx::duration underflow_duration) {
  TRACE_INSTANT("audio", "PacketQueue::ReportUnderflow", TRACE_SCOPE_PROCESS);
  uint16_t underflow_count = std::atomic_fetch_add<uint16_t>(&underflow_count_, 1u);

  if (underflow_reporter_) {
    underflow_reporter_(underflow_duration);
  }

  if constexpr (kLogUnderflow) {
    auto underflow_msec = static_cast<double>(underflow_duration.to_nsecs()) / ZX_MSEC(1);
    if ((kUnderflowErrorInterval > 0) && (underflow_count % kUnderflowErrorInterval == 0)) {
      FX_LOGS(ERROR) << "PACKET QUEUE UNDERFLOW #" << underflow_count + 1 << " (1/"
                     << kUnderflowErrorInterval << "): source-start 0x" << std::hex
                     << frac_source_start.raw_value() << " missed mix-point 0x"
                     << frac_source_mix_point.raw_value() << " by " << std::setprecision(4)
                     << underflow_msec << " ms";

    } else if ((kUnderflowInfoInterval > 0) && (underflow_count % kUnderflowInfoInterval == 0)) {
      FX_LOGS(INFO) << "PACKET QUEUE UNDERFLOW #" << underflow_count + 1 << " (1/"
                    << kUnderflowErrorInterval << "): source-start 0x" << std::hex
                    << frac_source_start.raw_value() << " missed mix-point 0x"
                    << frac_source_mix_point.raw_value() << " by " << std::setprecision(4)
                    << underflow_msec << " ms";

    } else if ((kUnderflowTraceInterval > 0) && (underflow_count % kUnderflowTraceInterval == 0)) {
      FX_VLOGS(TRACE) << "PACKET QUEUE UNDERFLOW #" << underflow_count + 1 << " (1/"
                      << kUnderflowErrorInterval << "): source-start 0x" << std::hex
                      << frac_source_start.raw_value() << " missed mix-point 0x"
                      << frac_source_mix_point.raw_value() << " by " << std::setprecision(4)
                      << underflow_msec << " ms";
    }
  }
}

void PacketQueue::ReportPartialUnderflow(FractionalFrames<int64_t> frac_source_offset,
                                         int64_t dest_mix_offset) {
  TRACE_INSTANT("audio", "PacketQueue::ReportPartialUnderflow", TRACE_SCOPE_PROCESS);

  // Shifts by less than four source frames do not necessarily indicate underflow. A shift of this
  // duration can be caused by the round-to-nearest-dest-frame step, when our rate-conversion ratio
  // is sufficiently large (it can be as large as 4:1).
  if (frac_source_offset >= 4) {
    auto partial_underflow_count = std::atomic_fetch_add<uint16_t>(&partial_underflow_count_, 1u);
    if constexpr (kLogUnderflow) {
      if ((kUnderflowErrorInterval > 0) &&
          (partial_underflow_count % kUnderflowErrorInterval == 0)) {
        FX_LOGS(WARNING) << "PACKET QUEUE SHIFT #" << partial_underflow_count + 1 << " (1/"
                         << kUnderflowErrorInterval << "): shifted by "
                         << (frac_source_offset < 0 ? "-0x" : "0x") << std::hex
                         << abs(frac_source_offset.raw_value()) << " source subframes and "
                         << std::dec << dest_mix_offset << " mix (output) frames";
      } else if ((kUnderflowInfoInterval > 0) &&
                 (partial_underflow_count % kUnderflowInfoInterval == 0)) {
        FX_LOGS(INFO) << "PACKET QUEUE SHIFT #" << partial_underflow_count + 1 << " (1/"
                      << kUnderflowInfoInterval << "): shifted by "
                      << (frac_source_offset < 0 ? "-0x" : "0x") << std::hex
                      << abs(frac_source_offset.raw_value()) << " source subframes and " << std::dec
                      << dest_mix_offset << " mix (output) frames";
      } else if ((kUnderflowTraceInterval > 0) &&
                 (partial_underflow_count % kUnderflowTraceInterval == 0)) {
        FX_VLOGS(TRACE) << "PACKET QUEUE SHIFT #" << partial_underflow_count + 1 << " (1/"
                        << kUnderflowTraceInterval << "): shifted by "
                        << (frac_source_offset < 0 ? "-0x" : "0x") << std::hex
                        << abs(frac_source_offset.raw_value()) << " source subframes and "
                        << std::dec << dest_mix_offset << " mix (output) frames";
      }
    }
  } else {
    if constexpr (kLogUnderflow) {
      FX_VLOGS(TRACE) << "shifted " << dest_mix_offset
                      << " mix (output) frames to align with source packet";
    }
  }
}

}  // namespace media::audio