src/media/audio/audio_core/packet_queue.cc - fuchsia - Git at Google

 // 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
	// 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