media/jni/soundpool/StreamManager.cpp - platform/frameworks/base - Git at Google

 /*
  * Copyright (C) 2019 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_NDEBUG 0
 #define LOG_TAG "SoundPool::StreamManager"
 #include <utils/Log.h>

 #include "StreamManager.h"

 #include <audio_utils/clock.h>
 #include <audio_utils/roundup.h>

 namespace android::soundpool {

 // kMaxStreams is number that should be less than the current AudioTrack max per UID of 40.
 // It is the maximum number of AudioTrack resources allowed in the SoundPool.
 // We suggest a value at least 4 or greater to allow CTS tests to pass.
 static constexpr int32_t kMaxStreams = 32;

 // kStealActiveStream_OldestFirst = false historically (Q and earlier)
 // Changing to true could break app expectations but could change behavior beneficially.
 // In R, we change this to true, as it is the correct way per SoundPool documentation.
 static constexpr bool kStealActiveStream_OldestFirst = true;

 // kPlayOnCallingThread = true prior to R.
 // Changing to false means calls to play() are almost instantaneous instead of taking around
 // ~10ms to launch the AudioTrack. It is perhaps 100x faster.
 static constexpr bool kPlayOnCallingThread = true;

 // Amount of time for a StreamManager thread to wait before closing.
 static constexpr int64_t kWaitTimeBeforeCloseNs = 9 * NANOS_PER_SECOND;

 ////////////

 StreamMap::StreamMap(int32_t streams) {
     ALOGV("%s(%d)", __func__, streams);
     if (streams > kMaxStreams) {
         ALOGW("%s: requested %d streams, clamping to %d", __func__, streams, kMaxStreams);
         streams = kMaxStreams;
     } else if (streams < 1) {
         ALOGW("%s: requested %d streams, clamping to 1", __func__, streams);
         streams = 1;
     }
     mStreamPoolSize = streams * 2;
     mStreamPool = std::make_unique<Stream[]>(mStreamPoolSize); // create array of streams.
     // we use a perfect hash table with 2x size to map StreamIDs to Stream pointers.
     mPerfectHash = std::make_unique<PerfectHash<int32_t, Stream *>>(roundup(mStreamPoolSize * 2));
 }

 Stream* StreamMap::findStream(int32_t streamID) const
 {
     Stream *stream = lookupStreamFromId(streamID);
     return stream != nullptr && stream->getStreamID() == streamID ? stream : nullptr;
 }

 size_t StreamMap::streamPosition(const Stream* stream) const
 {
     ptrdiff_t index = stream - mStreamPool.get();
     LOG_ALWAYS_FATAL_IF(index < 0 || (size_t)index >= mStreamPoolSize,
             "%s: stream position out of range: %td", __func__, index);
     return (size_t)index;
 }

 Stream* StreamMap::lookupStreamFromId(int32_t streamID) const
 {
     return streamID > 0 ? mPerfectHash->getValue(streamID).load() : nullptr;
 }

 int32_t StreamMap::getNextIdForStream(Stream* stream) const {
     // even though it is const, it mutates the internal hash table.
     const int32_t id = mPerfectHash->generateKey(
         stream,
         [] (Stream *stream) {
             return stream == nullptr ? 0 : stream->getStreamID();
         }, /* getKforV() */
         stream->getStreamID() /* oldID */);
     return id;
 }

 ////////////

 // Thread safety analysis is supposed to be disabled for constructors and destructors
 // but clang in R seems to have a bug.  We use pragma to disable.
 #pragma clang diagnostic push
 #pragma clang diagnostic ignored "-Wthread-safety-analysis"

 StreamManager::StreamManager(
         int32_t streams, size_t threads, const audio_attributes_t* attributes,
         std::string opPackageName)
     : StreamMap(streams)
     , mAttributes(*attributes)
     , mOpPackageName(std::move(opPackageName))
 {
     ALOGV("%s(%d, %zu, ...)", __func__, streams, threads);
     forEach([this](Stream *stream) {
         stream->setStreamManager(this);
         if ((streamPosition(stream) & 1) == 0) { // put the first stream of pair as available.
             mAvailableStreams.insert(stream);
         }
     });

     mThreadPool = std::make_unique<ThreadPool>(
             std::min(threads, (size_t)std::thread::hardware_concurrency()),
             "SoundPool_");
 }

 #pragma clang diagnostic pop

 StreamManager::~StreamManager()
 {
     ALOGV("%s", __func__);
     {
         std::unique_lock lock(mStreamManagerLock);
         mQuit = true;
         mStreamManagerCondition.notify_all();
     }
     mThreadPool->quit();

     // call stop on the stream pool
     forEach([](Stream *stream) { stream->stop(); });

     // This invokes the destructor on the AudioTracks -
     // we do it here to ensure that AudioTrack callbacks will not occur
     // afterwards.
     forEach([](Stream *stream) { stream->clearAudioTrack(); });
 }


 int32_t StreamManager::queueForPlay(const std::shared_ptr<Sound> &sound,
         int32_t soundID, float leftVolume, float rightVolume,
         int32_t priority, int32_t loop, float rate)
 {
     ALOGV("%s(sound=%p, soundID=%d, leftVolume=%f, rightVolume=%f, priority=%d, loop=%d, rate=%f)",
             __func__, sound.get(), soundID, leftVolume, rightVolume, priority, loop, rate);
     bool launchThread = false;
     int32_t streamID = 0;

     { // for lock
         std::unique_lock lock(mStreamManagerLock);
         Stream *newStream = nullptr;
         bool fromAvailableQueue = false;
         ALOGV("%s: mStreamManagerLock lock acquired", __func__);

         sanityCheckQueue_l();
         // find an available stream, prefer one that has matching sound id.
         if (mAvailableStreams.size() > 0) {
             for (auto stream : mAvailableStreams) {
                 if (stream->getSoundID() == soundID) {
                     newStream = stream;
                     ALOGV("%s: found soundID %d in available queue", __func__, soundID);
                     break;
                 }
             }
             if (newStream == nullptr) {
                 ALOGV("%s: found stream in available queue", __func__);
                 newStream = *mAvailableStreams.begin();
             }
             newStream->setStopTimeNs(systemTime());
             fromAvailableQueue = true;
         }

         // also look in the streams restarting (if the paired stream doesn't have a pending play)
         if (newStream == nullptr || newStream->getSoundID() != soundID) {
             for (auto [unused , stream] : mRestartStreams) {
                 if (!stream->getPairStream()->hasSound()) {
                     if (stream->getSoundID() == soundID) {
                         ALOGV("%s: found soundID %d in restart queue", __func__, soundID);
                         newStream = stream;
                         fromAvailableQueue = false;
                         break;
                     } else if (newStream == nullptr) {
                         ALOGV("%s: found stream in restart queue", __func__);
                         newStream = stream;
                     }
                 }
             }
         }

         // no available streams, look for one to steal from the active list
         if (newStream == nullptr) {
             for (auto stream : mActiveStreams) {
                 if (stream->getPriority() <= priority) {
                     if (newStream == nullptr
                             || newStream->getPriority() > stream->getPriority()) {
                         newStream = stream;
                         ALOGV("%s: found stream in active queue", __func__);
                     }
                 }
             }
             if (newStream != nullptr) { // we need to mute as it is still playing.
                 (void)newStream->requestStop(newStream->getStreamID());
             }
         }

         // none found, look for a stream that is restarting, evict one.
         if (newStream == nullptr) {
             for (auto [unused, stream] : mRestartStreams) {
                 if (stream->getPairPriority() <= priority) {
                     ALOGV("%s: evict stream from restart queue", __func__);
                     newStream = stream;
                     break;
                 }
             }
         }

         // DO NOT LOOK into mProcessingStreams as those are held by the StreamManager threads.

         if (newStream == nullptr) {
             ALOGD("%s: unable to find stream, returning 0", __func__);
             return 0; // unable to find available stream
         }

         Stream *pairStream = newStream->getPairStream();
         streamID = getNextIdForStream(pairStream);
         ALOGV("%s: newStream:%p  pairStream:%p, streamID:%d",
                 __func__, newStream, pairStream, streamID);
         pairStream->setPlay(
                 streamID, sound, soundID, leftVolume, rightVolume, priority, loop, rate);
         if (fromAvailableQueue && kPlayOnCallingThread) {
             removeFromQueues_l(newStream);
             mProcessingStreams.emplace(newStream);
             lock.unlock();
             if (Stream* nextStream = newStream->playPairStream()) {
                 lock.lock();
                 ALOGV("%s: starting streamID:%d", __func__, nextStream->getStreamID());
                 addToActiveQueue_l(nextStream);
             } else {
                 lock.lock();
                 mAvailableStreams.insert(newStream);
                 streamID = 0;
             }
             mProcessingStreams.erase(newStream);
         } else {
             launchThread = moveToRestartQueue_l(newStream) && needMoreThreads_l();
         }
         sanityCheckQueue_l();
         ALOGV("%s: mStreamManagerLock released", __func__);
     } // lock

     if (launchThread) {
         const int32_t id = mThreadPool->launch([this](int32_t id) { run(id); });
         (void)id; // avoid clang warning -Wunused-variable -Wused-but-marked-unused
         ALOGV_IF(id != 0, "%s: launched thread %d", __func__, id);
     }
     ALOGV("%s: returning %d", __func__, streamID);
     return streamID;
 }

 void StreamManager::moveToRestartQueue(
         Stream* stream, int32_t activeStreamIDToMatch)
 {
     ALOGV("%s(stream(ID)=%d, activeStreamIDToMatch=%d)",
             __func__, stream->getStreamID(), activeStreamIDToMatch);
     bool restart;
     {
         std::lock_guard lock(mStreamManagerLock);
         sanityCheckQueue_l();
         if (mProcessingStreams.count(stream) > 0 ||
                 mProcessingStreams.count(stream->getPairStream()) > 0) {
             ALOGD("%s: attempting to restart processing stream(%d)",
                     __func__, stream->getStreamID());
             restart = false;
         } else {
             moveToRestartQueue_l(stream, activeStreamIDToMatch);
             restart = needMoreThreads_l();
         }
         sanityCheckQueue_l();
     }
     if (restart) {
         const int32_t id = mThreadPool->launch([this](int32_t id) { run(id); });
         (void)id; // avoid clang warning -Wunused-variable -Wused-but-marked-unused
         ALOGV_IF(id != 0, "%s: launched thread %d", __func__, id);
     }
 }

 bool StreamManager::moveToRestartQueue_l(
         Stream* stream, int32_t activeStreamIDToMatch)
 {
     ALOGV("%s(stream(ID)=%d, activeStreamIDToMatch=%d)",
             __func__, stream->getStreamID(), activeStreamIDToMatch);
     if (activeStreamIDToMatch > 0 && stream->getStreamID() != activeStreamIDToMatch) {
         return false;
     }
     const ssize_t found = removeFromQueues_l(stream, activeStreamIDToMatch);
     if (found < 0) return false;

     LOG_ALWAYS_FATAL_IF(found > 1, "stream on %zd > 1 stream lists", found);

     addToRestartQueue_l(stream);
     mStreamManagerCondition.notify_one();
     return true;
 }

 ssize_t StreamManager::removeFromQueues_l(
         Stream* stream, int32_t activeStreamIDToMatch) {
     size_t found = 0;
     for (auto it = mActiveStreams.begin(); it != mActiveStreams.end(); ++it) {
         if (*it == stream) {
             mActiveStreams.erase(it); // we erase the iterator and break (otherwise it not safe).
             ++found;
             break;
         }
     }
     // activeStreamIDToMatch is nonzero indicates we proceed only if found.
     if (found == 0 && activeStreamIDToMatch > 0) {
         return -1;  // special code: not present on active streams, ignore restart request
     }

     for (auto it = mRestartStreams.begin(); it != mRestartStreams.end(); ++it) {
         if (it->second == stream) {
             mRestartStreams.erase(it);
             ++found;
             break;
         }
     }
     found += mAvailableStreams.erase(stream);

     // streams on mProcessingStreams are undergoing processing by the StreamManager thread
     // and do not participate in normal stream migration.
     return found;
 }

 void StreamManager::addToRestartQueue_l(Stream *stream) {
     mRestartStreams.emplace(stream->getStopTimeNs(), stream);
 }

 void StreamManager::addToActiveQueue_l(Stream *stream) {
     if (kStealActiveStream_OldestFirst) {
         mActiveStreams.push_back(stream);  // oldest to newest
     } else {
         mActiveStreams.push_front(stream); // newest to oldest
     }
 }

 void StreamManager::run(int32_t id)
 {
     ALOGV("%s(%d) entering", __func__, id);
     int64_t waitTimeNs = kWaitTimeBeforeCloseNs;
     std::unique_lock lock(mStreamManagerLock);
     while (!mQuit) {
         if (mRestartStreams.empty()) { // on thread start, mRestartStreams can be non-empty.
             mStreamManagerCondition.wait_for(
                     lock, std::chrono::duration<int64_t, std::nano>(waitTimeNs));
         }
         ALOGV("%s(%d) awake", __func__, id);

         sanityCheckQueue_l();

         if (mQuit || (mRestartStreams.empty() && waitTimeNs == kWaitTimeBeforeCloseNs)) {
             break;  // end the thread
         }

         waitTimeNs = kWaitTimeBeforeCloseNs;
         while (!mQuit && !mRestartStreams.empty()) {
             const nsecs_t nowNs = systemTime();
             auto it = mRestartStreams.begin();
             Stream* const stream = it->second;
             const int64_t diffNs = stream->getStopTimeNs() - nowNs;
             if (diffNs > 0) {
                 waitTimeNs = std::min(waitTimeNs, diffNs);
                 break;
             }
             mRestartStreams.erase(it);
             mProcessingStreams.emplace(stream);
             lock.unlock();
             stream->stop();
             ALOGV("%s(%d) stopping streamID:%d", __func__, id, stream->getStreamID());
             if (Stream* nextStream = stream->playPairStream()) {
                 ALOGV("%s(%d) starting streamID:%d", __func__, id, nextStream->getStreamID());
                 lock.lock();
                 if (nextStream->getStopTimeNs() > 0) {
                     // the next stream was stopped before we can move it to the active queue.
                     ALOGV("%s(%d) stopping started streamID:%d",
                             __func__, id, nextStream->getStreamID());
                     moveToRestartQueue_l(nextStream);
                 } else {
                     addToActiveQueue_l(nextStream);
                 }
             } else {
                 lock.lock();
                 mAvailableStreams.insert(stream);
             }
             mProcessingStreams.erase(stream);
             sanityCheckQueue_l();
         }
     }
     ALOGV("%s(%d) exiting", __func__, id);
 }

 void StreamManager::dump() const
 {
     forEach([](const Stream *stream) { stream->dump(); });
 }

 void StreamManager::sanityCheckQueue_l() const
 {
     // We want to preserve the invariant that each stream pair is exactly on one of the queues.
     const size_t availableStreams = mAvailableStreams.size();
     const size_t restartStreams = mRestartStreams.size();
     const size_t activeStreams = mActiveStreams.size();
     const size_t processingStreams = mProcessingStreams.size();
     const size_t managedStreams = availableStreams + restartStreams + activeStreams
                 + processingStreams;
     const size_t totalStreams = getStreamMapSize() >> 1;
     LOG_ALWAYS_FATAL_IF(managedStreams != totalStreams,
             "%s: mAvailableStreams:%zu + mRestartStreams:%zu + "
             "mActiveStreams:%zu + mProcessingStreams:%zu = %zu != total streams %zu",
             __func__, availableStreams, restartStreams, activeStreams, processingStreams,
             managedStreams, totalStreams);
     ALOGV("%s: mAvailableStreams:%zu + mRestartStreams:%zu + "
             "mActiveStreams:%zu + mProcessingStreams:%zu = %zu (total streams: %zu)",
             __func__, availableStreams, restartStreams, activeStreams, processingStreams,
             managedStreams, totalStreams);
 }

 } // namespace android::soundpool
	/*
	* Copyright (C) 2019 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_NDEBUG 0
	#define LOG_TAG "SoundPool::StreamManager"
	#include <utils/Log.h>

	#include "StreamManager.h"

	#include <audio_utils/clock.h>
	#include <audio_utils/roundup.h>

	namespace android::soundpool {

	// kMaxStreams is number that should be less than the current AudioTrack max per UID of 40.
	// It is the maximum number of AudioTrack resources allowed in the SoundPool.
	// We suggest a value at least 4 or greater to allow CTS tests to pass.
	static constexpr int32_t kMaxStreams = 32;

	// kStealActiveStream_OldestFirst = false historically (Q and earlier)
	// Changing to true could break app expectations but could change behavior beneficially.
	// In R, we change this to true, as it is the correct way per SoundPool documentation.
	static constexpr bool kStealActiveStream_OldestFirst = true;

	// kPlayOnCallingThread = true prior to R.
	// Changing to false means calls to play() are almost instantaneous instead of taking around
	// ~10ms to launch the AudioTrack. It is perhaps 100x faster.
	static constexpr bool kPlayOnCallingThread = true;

	// Amount of time for a StreamManager thread to wait before closing.
	static constexpr int64_t kWaitTimeBeforeCloseNs = 9 * NANOS_PER_SECOND;

	////////////

	StreamMap::StreamMap(int32_t streams) {
	ALOGV("%s(%d)", __func__, streams);
	if (streams > kMaxStreams) {
	ALOGW("%s: requested %d streams, clamping to %d", __func__, streams, kMaxStreams);
	streams = kMaxStreams;
	} else if (streams < 1) {
	ALOGW("%s: requested %d streams, clamping to 1", __func__, streams);
	streams = 1;
	}
	mStreamPoolSize = streams * 2;
	mStreamPool = std::make_unique<Stream[]>(mStreamPoolSize); // create array of streams.
	// we use a perfect hash table with 2x size to map StreamIDs to Stream pointers.
	mPerfectHash = std::make_unique<PerfectHash<int32_t, Stream >>(roundup(mStreamPoolSize 2));
	}

	Stream* StreamMap::findStream(int32_t streamID) const
	{
	Stream *stream = lookupStreamFromId(streamID);
	return stream != nullptr && stream->getStreamID() == streamID ? stream : nullptr;
	}

	size_t StreamMap::streamPosition(const Stream* stream) const
	{
	ptrdiff_t index = stream - mStreamPool.get();
	LOG_ALWAYS_FATAL_IF(index < 0 \|\| (size_t)index >= mStreamPoolSize,
	"%s: stream position out of range: %td", __func__, index);
	return (size_t)index;
	}

	Stream* StreamMap::lookupStreamFromId(int32_t streamID) const
	{
	return streamID > 0 ? mPerfectHash->getValue(streamID).load() : nullptr;
	}

	int32_t StreamMap::getNextIdForStream(Stream* stream) const {
	// even though it is const, it mutates the internal hash table.
	const int32_t id = mPerfectHash->generateKey(
	stream,
	[] (Stream *stream) {
	return stream == nullptr ? 0 : stream->getStreamID();
	}, /* getKforV() */
	stream->getStreamID() /* oldID */);
	return id;
	}

	////////////

	// Thread safety analysis is supposed to be disabled for constructors and destructors
	// but clang in R seems to have a bug. We use pragma to disable.
	#pragma clang diagnostic push
	#pragma clang diagnostic ignored "-Wthread-safety-analysis"

	StreamManager::StreamManager(
	int32_t streams, size_t threads, const audio_attributes_t* attributes,
	std::string opPackageName)
	: StreamMap(streams)
	, mAttributes(*attributes)
	, mOpPackageName(std::move(opPackageName))
	{
	ALOGV("%s(%d, %zu, ...)", __func__, streams, threads);
	forEach([this](Stream *stream) {
	stream->setStreamManager(this);
	if ((streamPosition(stream) & 1) == 0) { // put the first stream of pair as available.
	mAvailableStreams.insert(stream);
	}
	});

	mThreadPool = std::make_unique<ThreadPool>(
	std::min(threads, (size_t)std::thread::hardware_concurrency()),
	"SoundPool_");
	}

	#pragma clang diagnostic pop

	StreamManager::~StreamManager()
	{
	ALOGV("%s", __func__);
	{
	std::unique_lock lock(mStreamManagerLock);
	mQuit = true;
	mStreamManagerCondition.notify_all();
	}
	mThreadPool->quit();

	// call stop on the stream pool
	forEach([](Stream *stream) { stream->stop(); });

	// This invokes the destructor on the AudioTracks -
	// we do it here to ensure that AudioTrack callbacks will not occur
	// afterwards.
	forEach([](Stream *stream) { stream->clearAudioTrack(); });
	}


	int32_t StreamManager::queueForPlay(const std::shared_ptr<Sound> &sound,
	int32_t soundID, float leftVolume, float rightVolume,
	int32_t priority, int32_t loop, float rate)
	{
	ALOGV("%s(sound=%p, soundID=%d, leftVolume=%f, rightVolume=%f, priority=%d, loop=%d, rate=%f)",
	__func__, sound.get(), soundID, leftVolume, rightVolume, priority, loop, rate);
	bool launchThread = false;
	int32_t streamID = 0;

	{ // for lock
	std::unique_lock lock(mStreamManagerLock);
	Stream *newStream = nullptr;
	bool fromAvailableQueue = false;
	ALOGV("%s: mStreamManagerLock lock acquired", __func__);

	sanityCheckQueue_l();
	// find an available stream, prefer one that has matching sound id.
	if (mAvailableStreams.size() > 0) {
	for (auto stream : mAvailableStreams) {
	if (stream->getSoundID() == soundID) {
	newStream = stream;
	ALOGV("%s: found soundID %d in available queue", __func__, soundID);
	break;
	}
	}
	if (newStream == nullptr) {
	ALOGV("%s: found stream in available queue", __func__);
	newStream = *mAvailableStreams.begin();
	}
	newStream->setStopTimeNs(systemTime());
	fromAvailableQueue = true;
	}

	// also look in the streams restarting (if the paired stream doesn't have a pending play)
	if (newStream == nullptr \|\| newStream->getSoundID() != soundID) {
	for (auto [unused , stream] : mRestartStreams) {
	if (!stream->getPairStream()->hasSound()) {
	if (stream->getSoundID() == soundID) {
	ALOGV("%s: found soundID %d in restart queue", __func__, soundID);
	newStream = stream;
	fromAvailableQueue = false;
	break;
	} else if (newStream == nullptr) {
	ALOGV("%s: found stream in restart queue", __func__);
	newStream = stream;
	}
	}
	}
	}

	// no available streams, look for one to steal from the active list
	if (newStream == nullptr) {
	for (auto stream : mActiveStreams) {
	if (stream->getPriority() <= priority) {
	if (newStream == nullptr
	\|\| newStream->getPriority() > stream->getPriority()) {
	newStream = stream;
	ALOGV("%s: found stream in active queue", __func__);
	}
	}
	}
	if (newStream != nullptr) { // we need to mute as it is still playing.
	(void)newStream->requestStop(newStream->getStreamID());
	}
	}

	// none found, look for a stream that is restarting, evict one.
	if (newStream == nullptr) {
	for (auto [unused, stream] : mRestartStreams) {
	if (stream->getPairPriority() <= priority) {
	ALOGV("%s: evict stream from restart queue", __func__);
	newStream = stream;
	break;
	}
	}
	}

	// DO NOT LOOK into mProcessingStreams as those are held by the StreamManager threads.

	if (newStream == nullptr) {
	ALOGD("%s: unable to find stream, returning 0", __func__);
	return 0; // unable to find available stream
	}

	Stream *pairStream = newStream->getPairStream();
	streamID = getNextIdForStream(pairStream);
	ALOGV("%s: newStream:%p pairStream:%p, streamID:%d",
	__func__, newStream, pairStream, streamID);
	pairStream->setPlay(
	streamID, sound, soundID, leftVolume, rightVolume, priority, loop, rate);
	if (fromAvailableQueue && kPlayOnCallingThread) {
	removeFromQueues_l(newStream);
	mProcessingStreams.emplace(newStream);
	lock.unlock();
	if (Stream* nextStream = newStream->playPairStream()) {
	lock.lock();
	ALOGV("%s: starting streamID:%d", __func__, nextStream->getStreamID());
	addToActiveQueue_l(nextStream);
	} else {
	lock.lock();
	mAvailableStreams.insert(newStream);
	streamID = 0;
	}
	mProcessingStreams.erase(newStream);
	} else {
	launchThread = moveToRestartQueue_l(newStream) && needMoreThreads_l();
	}
	sanityCheckQueue_l();
	ALOGV("%s: mStreamManagerLock released", __func__);
	} // lock

	if (launchThread) {
	const int32_t id = mThreadPool->launch([this](int32_t id) { run(id); });
	(void)id; // avoid clang warning -Wunused-variable -Wused-but-marked-unused
	ALOGV_IF(id != 0, "%s: launched thread %d", __func__, id);
	}
	ALOGV("%s: returning %d", __func__, streamID);
	return streamID;
	}

	void StreamManager::moveToRestartQueue(
	Stream* stream, int32_t activeStreamIDToMatch)
	{
	ALOGV("%s(stream(ID)=%d, activeStreamIDToMatch=%d)",
	__func__, stream->getStreamID(), activeStreamIDToMatch);
	bool restart;
	{
	std::lock_guard lock(mStreamManagerLock);
	sanityCheckQueue_l();
	if (mProcessingStreams.count(stream) > 0 \|\|
	mProcessingStreams.count(stream->getPairStream()) > 0) {
	ALOGD("%s: attempting to restart processing stream(%d)",
	__func__, stream->getStreamID());
	restart = false;
	} else {
	moveToRestartQueue_l(stream, activeStreamIDToMatch);
	restart = needMoreThreads_l();
	}
	sanityCheckQueue_l();
	}
	if (restart) {
	const int32_t id = mThreadPool->launch([this](int32_t id) { run(id); });
	(void)id; // avoid clang warning -Wunused-variable -Wused-but-marked-unused
	ALOGV_IF(id != 0, "%s: launched thread %d", __func__, id);
	}
	}

	bool StreamManager::moveToRestartQueue_l(
	Stream* stream, int32_t activeStreamIDToMatch)
	{
	ALOGV("%s(stream(ID)=%d, activeStreamIDToMatch=%d)",
	__func__, stream->getStreamID(), activeStreamIDToMatch);
	if (activeStreamIDToMatch > 0 && stream->getStreamID() != activeStreamIDToMatch) {
	return false;
	}
	const ssize_t found = removeFromQueues_l(stream, activeStreamIDToMatch);
	if (found < 0) return false;

	LOG_ALWAYS_FATAL_IF(found > 1, "stream on %zd > 1 stream lists", found);

	addToRestartQueue_l(stream);
	mStreamManagerCondition.notify_one();
	return true;
	}

	ssize_t StreamManager::removeFromQueues_l(
	Stream* stream, int32_t activeStreamIDToMatch) {
	size_t found = 0;
	for (auto it = mActiveStreams.begin(); it != mActiveStreams.end(); ++it) {
	if (*it == stream) {
	mActiveStreams.erase(it); // we erase the iterator and break (otherwise it not safe).
	++found;
	break;
	}
	}
	// activeStreamIDToMatch is nonzero indicates we proceed only if found.
	if (found == 0 && activeStreamIDToMatch > 0) {
	return -1; // special code: not present on active streams, ignore restart request
	}

	for (auto it = mRestartStreams.begin(); it != mRestartStreams.end(); ++it) {
	if (it->second == stream) {
	mRestartStreams.erase(it);
	++found;
	break;
	}
	}
	found += mAvailableStreams.erase(stream);

	// streams on mProcessingStreams are undergoing processing by the StreamManager thread
	// and do not participate in normal stream migration.
	return found;
	}

	void StreamManager::addToRestartQueue_l(Stream *stream) {
	mRestartStreams.emplace(stream->getStopTimeNs(), stream);
	}

	void StreamManager::addToActiveQueue_l(Stream *stream) {
	if (kStealActiveStream_OldestFirst) {
	mActiveStreams.push_back(stream); // oldest to newest
	} else {
	mActiveStreams.push_front(stream); // newest to oldest
	}
	}

	void StreamManager::run(int32_t id)
	{
	ALOGV("%s(%d) entering", __func__, id);
	int64_t waitTimeNs = kWaitTimeBeforeCloseNs;
	std::unique_lock lock(mStreamManagerLock);
	while (!mQuit) {
	if (mRestartStreams.empty()) { // on thread start, mRestartStreams can be non-empty.
	mStreamManagerCondition.wait_for(
	lock, std::chrono::duration<int64_t, std::nano>(waitTimeNs));
	}
	ALOGV("%s(%d) awake", __func__, id);

	sanityCheckQueue_l();

	if (mQuit \|\| (mRestartStreams.empty() && waitTimeNs == kWaitTimeBeforeCloseNs)) {
	break; // end the thread
	}

	waitTimeNs = kWaitTimeBeforeCloseNs;
	while (!mQuit && !mRestartStreams.empty()) {
	const nsecs_t nowNs = systemTime();
	auto it = mRestartStreams.begin();
	Stream* const stream = it->second;
	const int64_t diffNs = stream->getStopTimeNs() - nowNs;
	if (diffNs > 0) {
	waitTimeNs = std::min(waitTimeNs, diffNs);
	break;
	}
	mRestartStreams.erase(it);
	mProcessingStreams.emplace(stream);
	lock.unlock();
	stream->stop();
	ALOGV("%s(%d) stopping streamID:%d", __func__, id, stream->getStreamID());
	if (Stream* nextStream = stream->playPairStream()) {
	ALOGV("%s(%d) starting streamID:%d", __func__, id, nextStream->getStreamID());
	lock.lock();
	if (nextStream->getStopTimeNs() > 0) {
	// the next stream was stopped before we can move it to the active queue.
	ALOGV("%s(%d) stopping started streamID:%d",
	__func__, id, nextStream->getStreamID());
	moveToRestartQueue_l(nextStream);
	} else {
	addToActiveQueue_l(nextStream);
	}
	} else {
	lock.lock();
	mAvailableStreams.insert(stream);
	}
	mProcessingStreams.erase(stream);
	sanityCheckQueue_l();
	}
	}
	ALOGV("%s(%d) exiting", __func__, id);
	}

	void StreamManager::dump() const
	{
	forEach([](const Stream *stream) { stream->dump(); });
	}

	void StreamManager::sanityCheckQueue_l() const
	{
	// We want to preserve the invariant that each stream pair is exactly on one of the queues.
	const size_t availableStreams = mAvailableStreams.size();
	const size_t restartStreams = mRestartStreams.size();
	const size_t activeStreams = mActiveStreams.size();
	const size_t processingStreams = mProcessingStreams.size();
	const size_t managedStreams = availableStreams + restartStreams + activeStreams
	+ processingStreams;
	const size_t totalStreams = getStreamMapSize() >> 1;
	LOG_ALWAYS_FATAL_IF(managedStreams != totalStreams,
	"%s: mAvailableStreams:%zu + mRestartStreams:%zu + "
	"mActiveStreams:%zu + mProcessingStreams:%zu = %zu != total streams %zu",
	__func__, availableStreams, restartStreams, activeStreams, processingStreams,
	managedStreams, totalStreams);
	ALOGV("%s: mAvailableStreams:%zu + mRestartStreams:%zu + "
	"mActiveStreams:%zu + mProcessingStreams:%zu = %zu (total streams: %zu)",
	__func__, availableStreams, restartStreams, activeStreams, processingStreams,
	managedStreams, totalStreams);
	}

	} // namespace android::soundpool