media/libnblog/Merger.cpp - platform/frameworks/av - Git at Google

 /*
  * Copyright (C) 2018 The Android Open Source Project
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 #define LOG_TAG "NBLog"
 //#define LOG_NDEBUG 0

 #include <memory>
 #include <queue>
 #include <stddef.h>
 #include <stdint.h>
 #include <vector>

 #include <audio_utils/fifo.h>
 #include <json/json.h>
 #include <media/nblog/Merger.h>
 #include <media/nblog/PerformanceAnalysis.h>
 #include <media/nblog/ReportPerformance.h>
 #include <media/nblog/Reader.h>
 #include <media/nblog/Timeline.h>
 #include <utils/Condition.h>
 #include <utils/Log.h>
 #include <utils/Mutex.h>
 #include <utils/RefBase.h>
 #include <utils/String16.h>
 #include <utils/String8.h>
 #include <utils/Thread.h>
 #include <utils/Timers.h>
 #include <utils/Vector.h>

 namespace android {
 namespace NBLog {

 Merger::Merger(const void *shared, size_t size):
       mShared((Shared *) shared),
       mFifo(mShared != NULL ?
         new audio_utils_fifo(size, sizeof(uint8_t),
             mShared->mBuffer, mShared->mRear, NULL /*throttlesFront*/) : NULL),
       mFifoWriter(mFifo != NULL ? new audio_utils_fifo_writer(*mFifo) : NULL)
 {
 }

 void Merger::addReader(const sp<Reader> &reader)
 {
     // FIXME This is called by binder thread in MediaLogService::registerWriter
     //       but the access to shared variable mReaders is not yet protected by a lock.
     mReaders.push_back(reader);
 }

 // items placed in priority queue during merge
 // composed by a timestamp and the index of the snapshot where the timestamp came from
 struct MergeItem
 {
     int64_t ts;
     int index;
     MergeItem(int64_t ts, int index): ts(ts), index(index) {}
 };

 bool operator>(const struct MergeItem &i1, const struct MergeItem &i2)
 {
     return i1.ts > i2.ts || (i1.ts == i2.ts && i1.index > i2.index);
 }

 // Merge registered readers, sorted by timestamp, and write data to a single FIFO in local memory
 void Merger::merge()
 {
     if (true) return; // Merging is not necessary at the moment, so this is to disable it
                       // and bypass compiler warnings about member variables not being used.
     const int nLogs = mReaders.size();
     std::vector<std::unique_ptr<Snapshot>> snapshots(nLogs);
     std::vector<EntryIterator> offsets;
     offsets.reserve(nLogs);
     for (int i = 0; i < nLogs; ++i) {
         snapshots[i] = mReaders[i]->getSnapshot();
         offsets.push_back(snapshots[i]->begin());
     }
     // initialize offsets
     // TODO custom heap implementation could allow to update top, improving performance
     // for bursty buffers
     std::priority_queue<MergeItem, std::vector<MergeItem>, std::greater<MergeItem>> timestamps;
     for (int i = 0; i < nLogs; ++i)
     {
         if (offsets[i] != snapshots[i]->end()) {
             std::unique_ptr<AbstractEntry> abstractEntry = AbstractEntry::buildEntry(offsets[i]);
             if (abstractEntry == nullptr) {
                 continue;
             }
             timestamps.emplace(abstractEntry->timestamp(), i);
         }
     }

     while (!timestamps.empty()) {
         int index = timestamps.top().index;     // find minimum timestamp
         // copy it to the log, increasing offset
         offsets[index] = AbstractEntry::buildEntry(offsets[index])->
             copyWithAuthor(mFifoWriter, index);
         // update data structures
         timestamps.pop();
         if (offsets[index] != snapshots[index]->end()) {
             int64_t ts = AbstractEntry::buildEntry(offsets[index])->timestamp();
             timestamps.emplace(ts, index);
         }
     }
 }

 const std::vector<sp<Reader>>& Merger::getReaders() const
 {
     //AutoMutex _l(mLock);
     return mReaders;
 }

 // ---------------------------------------------------------------------------

 MergeReader::MergeReader(const void *shared, size_t size, Merger &merger)
     : Reader(shared, size, "MergeReader"), mReaders(merger.getReaders())
 {
 }

 // Takes raw content of the local merger FIFO, processes log entries, and
 // writes the data to a map of class PerformanceAnalysis, based on their thread ID.
 void MergeReader::processSnapshot(Snapshot &snapshot, int author)
 {
     ReportPerformance::PerformanceData& data = mThreadPerformanceData[author];
     // We don't do "auto it" because it reduces readability in this case.
     for (EntryIterator it = snapshot.begin(); it != snapshot.end(); ++it) {
         switch (it->type) {
         case EVENT_HISTOGRAM_ENTRY_TS: {
             const HistTsEntry payload = it.payload<HistTsEntry>();
             // TODO: hash for histogram ts and audio state need to match
             // and correspond to audio production source file location
             mThreadPerformanceAnalysis[author][0 /*hash*/].logTsEntry(payload.ts);
         } break;
         case EVENT_AUDIO_STATE: {
             mThreadPerformanceAnalysis[author][0 /*hash*/].handleStateChange();
         } break;
         case EVENT_THREAD_INFO: {
             const thread_info_t info = it.payload<thread_info_t>();
             data.threadInfo = info;
         } break;
         case EVENT_THREAD_PARAMS: {
             const thread_params_t params = it.payload<thread_params_t>();
             data.threadParams = params;
         } break;
         case EVENT_LATENCY: {
             const double latencyMs = it.payload<double>();
             data.latencyHist.add(latencyMs);
         } break;
         case EVENT_WORK_TIME: {
             const int64_t monotonicNs = it.payload<int64_t>();
             const double monotonicMs = monotonicNs * 1e-6;
             data.workHist.add(monotonicMs);
             data.active += monotonicNs;
         } break;
         case EVENT_WARMUP_TIME: {
             const double timeMs = it.payload<double>();
             data.warmupHist.add(timeMs);
         } break;
         case EVENT_UNDERRUN: {
             const int64_t ts = it.payload<int64_t>();
             data.underruns++;
             data.snapshots.emplace_front(EVENT_UNDERRUN, ts);
             // TODO have a data structure to automatically handle resizing
             if (data.snapshots.size() > ReportPerformance::PerformanceData::kMaxSnapshotsToStore) {
                 data.snapshots.pop_back();
             }
         } break;
         case EVENT_OVERRUN: {
             const int64_t ts = it.payload<int64_t>();
             data.overruns++;
             data.snapshots.emplace_front(EVENT_UNDERRUN, ts);
             // TODO have a data structure to automatically handle resizing
             if (data.snapshots.size() > ReportPerformance::PerformanceData::kMaxSnapshotsToStore) {
                 data.snapshots.pop_back();
             }
         } break;
         case EVENT_RESERVED:
         case EVENT_UPPER_BOUND:
             ALOGW("warning: unexpected event %d", it->type);
             break;
         default:
             break;
         }
     }
 }

 void MergeReader::getAndProcessSnapshot()
 {
     // get a snapshot of each reader and process them
     // TODO insert lock here
     const size_t nLogs = mReaders.size();
     std::vector<std::unique_ptr<Snapshot>> snapshots(nLogs);
     for (size_t i = 0; i < nLogs; i++) {
         snapshots[i] = mReaders[i]->getSnapshot();
     }
     // TODO unlock lock here
     for (size_t i = 0; i < nLogs; i++) {
         if (snapshots[i] != nullptr) {
             processSnapshot(*(snapshots[i]), i);
         }
     }
     checkPushToMediaMetrics();
 }

 void MergeReader::checkPushToMediaMetrics()
 {
     const nsecs_t now = systemTime();
     for (auto& item : mThreadPerformanceData) {
         ReportPerformance::PerformanceData& data = item.second;
         if (now - data.start >= kPeriodicMediaMetricsPush) {
             (void)ReportPerformance::sendToMediaMetrics(data);
             data.reset();   // data is persistent per thread
         }
     }
 }

 void MergeReader::dump(int fd, const Vector<String16>& args)
 {
     // TODO: add a mutex around media.log dump
     // Options for dumpsys
     bool pa = false, json = false, plots = false, retro = false;
     for (const auto &arg : args) {
         if (arg == String16("--pa")) {
             pa = true;
         } else if (arg == String16("--json")) {
             json = true;
         } else if (arg == String16("--plots")) {
             plots = true;
         } else if (arg == String16("--retro")) {
             retro = true;
         }
     }
     if (pa) {
         ReportPerformance::dump(fd, 0 /*indent*/, mThreadPerformanceAnalysis);
     }
     if (json) {
         ReportPerformance::dumpJson(fd, mThreadPerformanceData);
     }
     if (plots) {
         ReportPerformance::dumpPlots(fd, mThreadPerformanceData);
     }
     if (retro) {
         ReportPerformance::dumpRetro(fd, mThreadPerformanceData);
     }
 }

 void MergeReader::handleAuthor(const AbstractEntry &entry, String8 *body)
 {
     int author = entry.author();
     if (author == -1) {
         return;
     }
     // FIXME Needs a lock
     const char* name = mReaders[author]->name().c_str();
     body->appendFormat("%s: ", name);
 }

 // ---------------------------------------------------------------------------

 MergeThread::MergeThread(Merger &merger, MergeReader &mergeReader)
     : mMerger(merger),
       mMergeReader(mergeReader),
       mTimeoutUs(0)
 {
 }

 MergeThread::~MergeThread()
 {
     // set exit flag, set timeout to 0 to force threadLoop to exit and wait for the thread to join
     requestExit();
     setTimeoutUs(0);
     join();
 }

 bool MergeThread::threadLoop()
 {
     bool doMerge;
     {
         AutoMutex _l(mMutex);
         // If mTimeoutUs is negative, wait on the condition variable until it's positive.
         // If it's positive, merge. The minimum period between waking the condition variable
         // is handled in AudioFlinger::MediaLogNotifier::threadLoop().
         mCond.wait(mMutex);
         doMerge = mTimeoutUs > 0;
         mTimeoutUs -= kThreadSleepPeriodUs;
     }
     if (doMerge) {
         // Merge data from all the readers
         mMerger.merge();
         // Process the data collected by mMerger and write it to PerformanceAnalysis
         // FIXME: decide whether to call getAndProcessSnapshot every time
         // or whether to have a separate thread that calls it with a lower frequency
         mMergeReader.getAndProcessSnapshot();
     }
     return true;
 }

 void MergeThread::wakeup()
 {
     setTimeoutUs(kThreadWakeupPeriodUs);
 }

 void MergeThread::setTimeoutUs(int time)
 {
     AutoMutex _l(mMutex);
     mTimeoutUs = time;
     mCond.signal();
 }

 }   // namespace NBLog
 }   // namespace android
	/*
	* Copyright (C) 2018 The Android Open Source Project
	*
	* Licensed under the Apache License, Version 2.0 (the "License");
	* you may not use this file except in compliance with the License.
	* You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	#define LOG_TAG "NBLog"
	//#define LOG_NDEBUG 0

	#include <memory>
	#include <queue>
	#include <stddef.h>
	#include <stdint.h>
	#include <vector>

	#include <audio_utils/fifo.h>
	#include <json/json.h>
	#include <media/nblog/Merger.h>
	#include <media/nblog/PerformanceAnalysis.h>
	#include <media/nblog/ReportPerformance.h>
	#include <media/nblog/Reader.h>
	#include <media/nblog/Timeline.h>
	#include <utils/Condition.h>
	#include <utils/Log.h>
	#include <utils/Mutex.h>
	#include <utils/RefBase.h>
	#include <utils/String16.h>
	#include <utils/String8.h>
	#include <utils/Thread.h>
	#include <utils/Timers.h>
	#include <utils/Vector.h>

	namespace android {
	namespace NBLog {

	Merger::Merger(const void *shared, size_t size):
	mShared((Shared *) shared),
	mFifo(mShared != NULL ?
	new audio_utils_fifo(size, sizeof(uint8_t),
	mShared->mBuffer, mShared->mRear, NULL /throttlesFront/) : NULL),
	mFifoWriter(mFifo != NULL ? new audio_utils_fifo_writer(*mFifo) : NULL)
	{
	}

	void Merger::addReader(const sp<Reader> &reader)
	{
	// FIXME This is called by binder thread in MediaLogService::registerWriter
	// but the access to shared variable mReaders is not yet protected by a lock.
	mReaders.push_back(reader);
	}

	// items placed in priority queue during merge
	// composed by a timestamp and the index of the snapshot where the timestamp came from
	struct MergeItem
	{
	int64_t ts;
	int index;
	MergeItem(int64_t ts, int index): ts(ts), index(index) {}
	};

	bool operator>(const struct MergeItem &i1, const struct MergeItem &i2)
	{
	return i1.ts > i2.ts \|\| (i1.ts == i2.ts && i1.index > i2.index);
	}

	// Merge registered readers, sorted by timestamp, and write data to a single FIFO in local memory
	void Merger::merge()
	{
	if (true) return; // Merging is not necessary at the moment, so this is to disable it
	// and bypass compiler warnings about member variables not being used.
	const int nLogs = mReaders.size();
	std::vector<std::unique_ptr<Snapshot>> snapshots(nLogs);
	std::vector<EntryIterator> offsets;
	offsets.reserve(nLogs);
	for (int i = 0; i < nLogs; ++i) {
	snapshots[i] = mReaders[i]->getSnapshot();
	offsets.push_back(snapshots[i]->begin());
	}
	// initialize offsets
	// TODO custom heap implementation could allow to update top, improving performance
	// for bursty buffers
	std::priority_queue<MergeItem, std::vector<MergeItem>, std::greater<MergeItem>> timestamps;
	for (int i = 0; i < nLogs; ++i)
	{
	if (offsets[i] != snapshots[i]->end()) {
	std::unique_ptr<AbstractEntry> abstractEntry = AbstractEntry::buildEntry(offsets[i]);
	if (abstractEntry == nullptr) {
	continue;
	}
	timestamps.emplace(abstractEntry->timestamp(), i);
	}
	}

	while (!timestamps.empty()) {
	int index = timestamps.top().index; // find minimum timestamp
	// copy it to the log, increasing offset
	offsets[index] = AbstractEntry::buildEntry(offsets[index])->
	copyWithAuthor(mFifoWriter, index);
	// update data structures
	timestamps.pop();
	if (offsets[index] != snapshots[index]->end()) {
	int64_t ts = AbstractEntry::buildEntry(offsets[index])->timestamp();
	timestamps.emplace(ts, index);
	}
	}
	}

	const std::vector<sp<Reader>>& Merger::getReaders() const
	{
	//AutoMutex _l(mLock);
	return mReaders;
	}

	// ---------------------------------------------------------------------------

	MergeReader::MergeReader(const void *shared, size_t size, Merger &merger)
	: Reader(shared, size, "MergeReader"), mReaders(merger.getReaders())
	{
	}

	// Takes raw content of the local merger FIFO, processes log entries, and
	// writes the data to a map of class PerformanceAnalysis, based on their thread ID.
	void MergeReader::processSnapshot(Snapshot &snapshot, int author)
	{
	ReportPerformance::PerformanceData& data = mThreadPerformanceData[author];
	// We don't do "auto it" because it reduces readability in this case.
	for (EntryIterator it = snapshot.begin(); it != snapshot.end(); ++it) {
	switch (it->type) {
	case EVENT_HISTOGRAM_ENTRY_TS: {
	const HistTsEntry payload = it.payload<HistTsEntry>();
	// TODO: hash for histogram ts and audio state need to match
	// and correspond to audio production source file location
	mThreadPerformanceAnalysis[author][0 /hash/].logTsEntry(payload.ts);
	} break;
	case EVENT_AUDIO_STATE: {
	mThreadPerformanceAnalysis[author][0 /hash/].handleStateChange();
	} break;
	case EVENT_THREAD_INFO: {
	const thread_info_t info = it.payload<thread_info_t>();
	data.threadInfo = info;
	} break;
	case EVENT_THREAD_PARAMS: {
	const thread_params_t params = it.payload<thread_params_t>();
	data.threadParams = params;
	} break;
	case EVENT_LATENCY: {
	const double latencyMs = it.payload<double>();
	data.latencyHist.add(latencyMs);
	} break;
	case EVENT_WORK_TIME: {
	const int64_t monotonicNs = it.payload<int64_t>();
	const double monotonicMs = monotonicNs * 1e-6;
	data.workHist.add(monotonicMs);
	data.active += monotonicNs;
	} break;
	case EVENT_WARMUP_TIME: {
	const double timeMs = it.payload<double>();
	data.warmupHist.add(timeMs);
	} break;
	case EVENT_UNDERRUN: {
	const int64_t ts = it.payload<int64_t>();
	data.underruns++;
	data.snapshots.emplace_front(EVENT_UNDERRUN, ts);
	// TODO have a data structure to automatically handle resizing
	if (data.snapshots.size() > ReportPerformance::PerformanceData::kMaxSnapshotsToStore) {
	data.snapshots.pop_back();
	}
	} break;
	case EVENT_OVERRUN: {
	const int64_t ts = it.payload<int64_t>();
	data.overruns++;
	data.snapshots.emplace_front(EVENT_UNDERRUN, ts);
	// TODO have a data structure to automatically handle resizing
	if (data.snapshots.size() > ReportPerformance::PerformanceData::kMaxSnapshotsToStore) {
	data.snapshots.pop_back();
	}
	} break;
	case EVENT_RESERVED:
	case EVENT_UPPER_BOUND:
	ALOGW("warning: unexpected event %d", it->type);
	break;
	default:
	break;
	}
	}
	}

	void MergeReader::getAndProcessSnapshot()
	{
	// get a snapshot of each reader and process them
	// TODO insert lock here
	const size_t nLogs = mReaders.size();
	std::vector<std::unique_ptr<Snapshot>> snapshots(nLogs);
	for (size_t i = 0; i < nLogs; i++) {
	snapshots[i] = mReaders[i]->getSnapshot();
	}
	// TODO unlock lock here
	for (size_t i = 0; i < nLogs; i++) {
	if (snapshots[i] != nullptr) {
	processSnapshot(*(snapshots[i]), i);
	}
	}
	checkPushToMediaMetrics();
	}

	void MergeReader::checkPushToMediaMetrics()
	{
	const nsecs_t now = systemTime();
	for (auto& item : mThreadPerformanceData) {
	ReportPerformance::PerformanceData& data = item.second;
	if (now - data.start >= kPeriodicMediaMetricsPush) {
	(void)ReportPerformance::sendToMediaMetrics(data);
	data.reset(); // data is persistent per thread
	}
	}
	}

	void MergeReader::dump(int fd, const Vector<String16>& args)
	{
	// TODO: add a mutex around media.log dump
	// Options for dumpsys
	bool pa = false, json = false, plots = false, retro = false;
	for (const auto &arg : args) {
	if (arg == String16("--pa")) {
	pa = true;
	} else if (arg == String16("--json")) {
	json = true;
	} else if (arg == String16("--plots")) {
	plots = true;
	} else if (arg == String16("--retro")) {
	retro = true;
	}
	}
	if (pa) {
	ReportPerformance::dump(fd, 0 /indent/, mThreadPerformanceAnalysis);
	}
	if (json) {
	ReportPerformance::dumpJson(fd, mThreadPerformanceData);
	}
	if (plots) {
	ReportPerformance::dumpPlots(fd, mThreadPerformanceData);
	}
	if (retro) {
	ReportPerformance::dumpRetro(fd, mThreadPerformanceData);
	}
	}

	void MergeReader::handleAuthor(const AbstractEntry &entry, String8 *body)
	{
	int author = entry.author();
	if (author == -1) {
	return;
	}
	// FIXME Needs a lock
	const char* name = mReaders[author]->name().c_str();
	body->appendFormat("%s: ", name);
	}

	// ---------------------------------------------------------------------------

	MergeThread::MergeThread(Merger &merger, MergeReader &mergeReader)
	: mMerger(merger),
	mMergeReader(mergeReader),
	mTimeoutUs(0)
	{
	}

	MergeThread::~MergeThread()
	{
	// set exit flag, set timeout to 0 to force threadLoop to exit and wait for the thread to join
	requestExit();
	setTimeoutUs(0);
	join();
	}

	bool MergeThread::threadLoop()
	{
	bool doMerge;
	{
	AutoMutex _l(mMutex);
	// If mTimeoutUs is negative, wait on the condition variable until it's positive.
	// If it's positive, merge. The minimum period between waking the condition variable
	// is handled in AudioFlinger::MediaLogNotifier::threadLoop().
	mCond.wait(mMutex);
	doMerge = mTimeoutUs > 0;
	mTimeoutUs -= kThreadSleepPeriodUs;
	}
	if (doMerge) {
	// Merge data from all the readers
	mMerger.merge();
	// Process the data collected by mMerger and write it to PerformanceAnalysis
	// FIXME: decide whether to call getAndProcessSnapshot every time
	// or whether to have a separate thread that calls it with a lower frequency
	mMergeReader.getAndProcessSnapshot();
	}
	return true;
	}

	void MergeThread::wakeup()
	{
	setTimeoutUs(kThreadWakeupPeriodUs);
	}

	void MergeThread::setTimeoutUs(int time)
	{
	AutoMutex _l(mMutex);
	mTimeoutUs = time;
	mCond.signal();
	}

	} // namespace NBLog
	} // namespace android