/*
 * Copyright (C) 2017 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
#include <utils/Log.h>

#define ATRACE_TAG ATRACE_TAG_AUDIO

#include <algorithm>
#include <chrono>
#include <thread>

#include <audio_utils/mutex.h>
#include <media/MediaMetricsItem.h>
#include <mediautils/Runnable.h>
#include <utils/SystemClock.h>
#include <utils/Trace.h>

#include "client/AudioStreamInternalPlay.h"
#include "utility/AudioClock.h"

// We do this after the #includes because if a header uses ALOG.
// it would fail on the reference to mInService.
#undef LOG_TAG
// This file is used in both client and server processes.
// This is needed to make sense of the logs more easily.
#define LOG_TAG (mInService ? "AudioStreamInternalPlay_Service" \
                            : "AudioStreamInternalPlay_Client")

using android::status_t;
using android::WrappingBuffer;

using namespace aaudio;

using android::audio_utils::TimerQueue;

AudioStreamInternalPlay::AudioStreamInternalPlay(AAudioServiceInterface  &serviceInterface,
                                                       bool inService)
        : AudioStreamInternal(serviceInterface, inService) {
}

constexpr int kRampMSec = 10; // time to apply a change in volume

aaudio_result_t AudioStreamInternalPlay::open(const AudioStreamBuilder &builder) {
    aaudio_result_t result = AudioStreamInternal::open(builder);
    const bool useVolumeRamps = (getSharingMode() == AAUDIO_SHARING_MODE_EXCLUSIVE);
    if (result == AAUDIO_OK) {
        result = mFlowGraph.configure(getFormat(),
                             getSamplesPerFrame(),
                             getSampleRate(),
                             getDeviceFormat(),
                             getDeviceSamplesPerFrame(),
                             getDeviceSampleRate(),
                             getRequireMonoBlend(),
                             useVolumeRamps,
                             getAudioBalance(),
                             aaudio::resampler::MultiChannelResampler::Quality::Medium);

        if (result != AAUDIO_OK) {
            safeReleaseClose();
        }
        // Sample rate is constrained to common values by now and should not overflow.
        int32_t numFrames = kRampMSec * getSampleRate() / AAUDIO_MILLIS_PER_SECOND;
        mFlowGraph.setRampLengthInFrames(numFrames);
    }
    if (getPerformanceMode() == AAUDIO_PERFORMANCE_MODE_POWER_SAVING_OFFLOADED &&
        !isDataCallbackSet() && mPresentationEndCallbackProc != nullptr) {
        // Client is not using data callback but has presentation end callback for offload playback,
        // initialize an executor for presentation end callback.
        std::lock_guard _l(mStreamMutex);
        mStreamEndExecutor.emplace();
    }
    // Use 1s + burst size as a safe margin to wake up the callback thread for writing more
    // data to avoid glitch.
    mOffloadSafeMarginInFrames =
            getDeviceSampleRate() * kOffloadSafeMarginMs / AAUDIO_MILLIS_PER_SECOND +
            getDeviceFramesPerBurst();
    // Use 100ms + burst size as a safe margin when calculating the safe position to flush from.
    mOffloadFlushFromSafeMarginInFrames =
            getDeviceSampleRate() * kOffloadFlushFromSafeMarginMs / AAUDIO_MILLIS_PER_SECOND +
            getDeviceFramesPerBurst();
    return result;
}

// This must be called under mStreamMutex.
aaudio_result_t AudioStreamInternalPlay::requestPause_l()
{
    aaudio_result_t result = stopCallback_l();
    if (result != AAUDIO_OK) {
        return result;
    }
    if (getServiceHandle() == AAUDIO_HANDLE_INVALID) {
        ALOGW("%s() mServiceStreamHandle invalid", __func__);
        return AAUDIO_ERROR_INVALID_STATE;
    }

    mClockModel.stop(AudioClock::getNanoseconds());
    setState(AAUDIO_STREAM_STATE_PAUSING);
    mAtomicInternalTimestamp.clear();

    // When pause is called, the service will notify the HAL so that no more data will be consumed.
    // In that case, it is no longer needed to wait for stream end.
    dropPresentationEndCallback_l();

    return mServiceInterface.pauseStream(mServiceStreamHandleInfo);
}

aaudio_result_t AudioStreamInternalPlay::requestFlush_l() {
    if (getServiceHandle() == AAUDIO_HANDLE_INVALID) {
        ALOGW("%s() mServiceStreamHandle invalid", __func__);
        return AAUDIO_ERROR_INVALID_STATE;
    }

    setState(AAUDIO_STREAM_STATE_FLUSHING);

    // When flush is called, the service will notify the HAL so that no more data will be consumed.
    // In that case, it is no longer needed to wait for stream end.
    dropPresentationEndCallback_l();

    return mServiceInterface.flushStream(mServiceStreamHandleInfo);
}

void AudioStreamInternalPlay::prepareBuffersForStart() {
    // Reset volume ramps to avoid a starting noise.
    // This was called here instead of AudioStreamInternal so that
    // it will be easier to backport.
    mFlowGraph.reset();
    // Prevent stale data from being played.
    mAudioEndpoint->eraseDataMemory();
    // All data has been erased. To avoid mixer for the shared stream use stale
    // counters, which may cause the service side thinking stream starts flowing before
    // the client actually writes data, advance the client to match server position.
    advanceClientToMatchServerPosition(0 /*serverMargin*/);
}

void AudioStreamInternalPlay::prepareBuffersForStop() {
    // If this is a shared stream and the FIFO is being read by the mixer then
    // we don't have to worry about the DSP reading past the valid data. We can skip all this.
    if(!mAudioEndpoint->isFreeRunning()) {
        return;
    }
    // Sleep until the DSP has read all of the data written.
    int64_t validFramesInBuffer =
            mAudioEndpoint->getDataWriteCounter() - mAudioEndpoint->getDataReadCounter();
    if (validFramesInBuffer >= 0) {
        int64_t emptyFramesInBuffer = ((int64_t) getDeviceBufferCapacity()) - validFramesInBuffer;

        // Prevent stale data from being played if the DSP is still running.
        // Erase some of the FIFO memory in front of the DSP read cursor.
        // Subtract one burst so we do not accidentally erase data that the DSP might be using.
        int64_t framesToErase = std::max((int64_t) 0,
                                         emptyFramesInBuffer - getDeviceFramesPerBurst());
        mAudioEndpoint->eraseEmptyDataMemory(framesToErase);

        // Sleep until we are confident the DSP has consumed all of the valid data.
        // Sleep for one extra burst as a safety margin because the IsochronousClockModel
        // is not perfectly accurate.
        // The ClockModel uses the server frame position so do not use getFramesWritten().
        int64_t positionInEmptyMemory = mAudioEndpoint->getDataWriteCounter() + getDeviceFramesPerBurst();
        int64_t timeAllConsumed = mClockModel.convertPositionToTime(positionInEmptyMemory);
        int64_t durationAllConsumed = timeAllConsumed - AudioClock::getNanoseconds();
        // Prevent sleeping for too long.
        durationAllConsumed = std::min(200 * AAUDIO_NANOS_PER_MILLISECOND, durationAllConsumed);
        AudioClock::sleepForNanos(durationAllConsumed);
    }

    // Erase all of the memory in case the DSP keeps going and wraps around.
    mAudioEndpoint->eraseDataMemory();

    // Wait for the last buffer to reach the DAC.
    // This is because the expected behavior of stop() is that all data written to the stream
    // should be played before the hardware actually shuts down.
    // This is different than pause(), where we just end as soon as possible.
    // This can be important when, for example, playing car navigation and
    // you want the user to hear the complete instruction.
    if (mAtomicInternalTimestamp.isValid()) {
        // Use timestamps to calculate the latency between the DSP reading
        // a frame and when it reaches the DAC.
        // This code assumes that timestamps are accurate.
        Timestamp timestamp = mAtomicInternalTimestamp.read();
        int64_t dacPosition = timestamp.getPosition();
        int64_t hardwareReadTime = mClockModel.convertPositionToTime(dacPosition);
        int64_t hardwareLatencyNanos = timestamp.getNanoseconds() - hardwareReadTime;
        ALOGD("%s() hardwareLatencyNanos = %lld", __func__,
              (long long) hardwareLatencyNanos);
        // Prevent sleeping for too long.
        hardwareLatencyNanos = std::min(30 * AAUDIO_NANOS_PER_MILLISECOND,
                                        hardwareLatencyNanos);
        AudioClock::sleepForNanos(hardwareLatencyNanos);
    }
}

void AudioStreamInternalPlay::advanceClientToMatchServerPosition(int32_t serverMargin) {
    int64_t readCounter = mAudioEndpoint->getDataReadCounter() + serverMargin;
    int64_t writeCounter = mAudioEndpoint->getDataWriteCounter();

    // Bump offset so caller does not see the retrograde motion in getFramesRead().
    int64_t offset = writeCounter - readCounter;
    mFramesOffsetFromService += offset;
    ALOGV("%s() readN = %lld, writeN = %lld, offset = %lld", __func__,
          (long long)readCounter, (long long)writeCounter, (long long)mFramesOffsetFromService);

    // Force writeCounter to match readCounter.
    // This is because we cannot change the read counter in the hardware.
    mAudioEndpoint->setDataWriteCounter(readCounter);
}

void AudioStreamInternalPlay::onFlushFromServer() {
    advanceClientToMatchServerPosition(0 /*serverMargin*/);
}

// Write the data, block if needed and timeoutMillis > 0
aaudio_result_t AudioStreamInternalPlay::write(const void *buffer, int32_t numFrames,
                                               int64_t timeoutNanoseconds) {
    if (mayNeedToDrain() && !isDataCallbackSet()) {
        std::lock_guard _l(mStreamMutex);
        if (mDraining) {
            if (aaudio_result_t result = activateStream_l(); result != AAUDIO_OK) {
                return result;
            }
            mDraining = false;
        }
    }
    aaudio_result_t result = processData((void *)buffer, numFrames, timeoutNanoseconds);
    if (isDataCallbackSet() && result != numFrames) {
        // For callback case, it must always be able to write all data
        if (result >= 0) {
            // Only wrote some of the frames requested. The stream can be disconnected
            // or timed out.
            ALOGW("%s from callback thread, %d frames written, %d frames provided",
                  __func__, result, numFrames);
            processCommands();
            result = isDisconnected() ? AAUDIO_ERROR_DISCONNECTED : AAUDIO_ERROR_TIMEOUT;
        }
        maybeCallErrorCallback(result);
        return result;
    }
    if (result >= 0 && mayNeedToDrain()) {
        // If it is buffer size is big and the buffer is pretty full, sleep to drain data
        // to save battery.
        int32_t fullFrames = mAudioEndpoint->getFullFramesAvailable();
        if (fullFrames > getDeviceBufferSize() - mOffloadSafeMarginInFrames &&
            fullFrames > getDeviceSampleRate() * 1 + mOffloadSafeMarginInFrames) {
            const int64_t drainNanos = mClockModel.convertDeltaPositionToTime(
                    fullFrames - mOffloadSafeMarginInFrames);
            const int64_t wakeUpNanos = android::elapsedRealtimeNano() + drainNanos;
            android::audio_utils::unique_lock ul(mStreamMutex);
            if (aaudio_result_t ret = drainStream_l(wakeUpNanos, isDataCallbackSet());
                ret != AAUDIO_OK) {
                ALOGE("%s() failed to drain, error=%d", __func__, ret);
                return ret;
            }
            mDraining = true;
            if (isDataCallbackSet()) {
                mCallbackCV.wait_for(ul, std::chrono::nanoseconds(drainNanos),
                                     [this]() REQUIRES(mStreamMutex) {
                    return !mDraining;
                });
            }
        }
    }
    return result;
}

// Write as much data as we can without blocking.
aaudio_result_t AudioStreamInternalPlay::processDataNow(void *buffer, int32_t numFrames,
                                              int64_t currentNanoTime, int64_t *wakeTimePtr) {
    aaudio_result_t result = processCommands();
    if (result != AAUDIO_OK) {
        return result;
    }

    const char *traceName = "aaWrNow";
    ATRACE_BEGIN(traceName);

    if (mClockModel.isStarting()) {
        // Still haven't got any timestamps from server.
        // Keep waiting until we get some valid timestamps then start writing to the
        // current buffer position.
        ALOGV("%s() wait for valid timestamps", __func__);
        // Sleep very briefly and hope we get a timestamp soon.
        *wakeTimePtr = currentNanoTime + (2000 * AAUDIO_NANOS_PER_MICROSECOND);
        ATRACE_END();
        return 0;
    }
    // If we have gotten this far then we have at least one timestamp from server.

    // If a DMA channel or DSP is reading the other end then we have to update the readCounter.
    if (mAudioEndpoint->isFreeRunning()) {
        // Update data queue based on the timing model.
        int64_t estimatedReadCounter = mClockModel.convertTimeToPosition(currentNanoTime);
        // ALOGD("AudioStreamInternal::processDataNow() - estimatedReadCounter = %d", (int)estimatedReadCounter);
        mAudioEndpoint->setDataReadCounter(estimatedReadCounter);
    }

    if (mNeedCatchUp.isRequested()) {
        // Catch an MMAP pointer that is already advancing.
        // This will avoid initial underruns caused by a slow cold start.
        // We add a one burst margin in case the DSP advances before we can write the data.
        // This can help prevent the beginning of the stream from being skipped.
        advanceClientToMatchServerPosition(getDeviceFramesPerBurst());
        mNeedCatchUp.acknowledge();
    }

    // If the read index passed the write index then consider it an underrun.
    // For shared streams, the xRunCount is passed up from the service.
    if (mAudioEndpoint->isFreeRunning() && mAudioEndpoint->getFullFramesAvailable() < 0) {
        mXRunCount++;
        if (ATRACE_ENABLED()) {
            ATRACE_INT("aaUnderRuns", mXRunCount);
        }
    }

    // Write some data to the buffer.
    //ALOGD("AudioStreamInternal::processDataNow() - writeNowWithConversion(%d)", numFrames);
    int32_t framesWritten = writeNowWithConversion(buffer, numFrames);
    //ALOGD("AudioStreamInternal::processDataNow() - tried to write %d frames, wrote %d",
    //    numFrames, framesWritten);
    if (ATRACE_ENABLED()) {
        ATRACE_INT("aaWrote", framesWritten);
    }

    // Sleep if there is too much data in the buffer.
    // Calculate an ideal time to wake up.
    if (wakeTimePtr != nullptr
            && (mAudioEndpoint->getFullFramesAvailable() >= getDeviceBufferSize())) {
        // By default wake up a few milliseconds from now.  // TODO review
        int64_t wakeTime = currentNanoTime + (1 * AAUDIO_NANOS_PER_MILLISECOND);
        aaudio_stream_state_t state = getState();
        //ALOGD("AudioStreamInternal::processDataNow() - wakeTime based on %s",
        //      AAudio_convertStreamStateToText(state));
        switch (state) {
            case AAUDIO_STREAM_STATE_OPEN:
            case AAUDIO_STREAM_STATE_STARTING:
                if (framesWritten != 0) {
                    // Don't wait to write more data. Just prime the buffer.
                    wakeTime = currentNanoTime;
                }
                break;
            case AAUDIO_STREAM_STATE_STARTED:
            {
                // Calculate when there will be room available to write to the buffer.
                // If the appBufferSize is smaller than the endpointBufferSize then
                // we will have room to write data beyond the appBufferSize.
                // That is a technique used to reduce glitches without adding latency.
                const int64_t appBufferSize = getDeviceBufferSize();
                // The endpoint buffer size is set to the maximum that can be written.
                // If we use it then we must carve out some room to write data when we wake up.
                const int64_t endBufferSize = mAudioEndpoint->getBufferSizeInFrames()
                        - getDeviceFramesPerBurst();
                const int64_t bestBufferSize = std::min(appBufferSize, endBufferSize);
                int64_t targetReadPosition = mAudioEndpoint->getDataWriteCounter() - bestBufferSize;
                wakeTime = mClockModel.convertPositionToTime(targetReadPosition);
            }
                break;
            default:
                break;
        }
        *wakeTimePtr = wakeTime;

    }

    ATRACE_END();
    return framesWritten;
}


aaudio_result_t AudioStreamInternalPlay::writeNowWithConversion(const void *buffer,
                                                            int32_t numFrames) {
    WrappingBuffer wrappingBuffer;
    uint8_t *byteBuffer = (uint8_t *) buffer;
    int32_t framesLeftInByteBuffer = numFrames;

    mAudioEndpoint->getEmptyFramesAvailable(&wrappingBuffer);

    // Write data in one or two parts.
    int partIndex = 0;
    int framesWrittenToAudioEndpoint = 0;
    while (framesLeftInByteBuffer > 0 && partIndex < WrappingBuffer::SIZE) {
        int32_t framesAvailableInWrappingBuffer = wrappingBuffer.numFrames[partIndex];
        uint8_t *currentWrappingBuffer = (uint8_t *) wrappingBuffer.data[partIndex];

        if (framesAvailableInWrappingBuffer > 0) {
            // Pull data from the flowgraph in case there is residual data.
            const int32_t framesActuallyWrittenToWrappingBuffer = mFlowGraph.pull(
                (void*) currentWrappingBuffer,
                framesAvailableInWrappingBuffer);

            const int32_t numBytesActuallyWrittenToWrappingBuffer =
                framesActuallyWrittenToWrappingBuffer * getBytesPerDeviceFrame();
            currentWrappingBuffer += numBytesActuallyWrittenToWrappingBuffer;
            framesAvailableInWrappingBuffer -= framesActuallyWrittenToWrappingBuffer;
            framesWrittenToAudioEndpoint += framesActuallyWrittenToWrappingBuffer;
        } else {
            break;
        }

        // Put data from byteBuffer into the flowgraph one buffer (8 frames) at a time.
        // Continuously pull as much data as possible from the flowgraph into the wrapping buffer.
        // The return value of mFlowGraph.process is the number of frames actually pulled.
        while (framesAvailableInWrappingBuffer > 0 && framesLeftInByteBuffer > 0) {
            int32_t framesToWriteFromByteBuffer = std::min(flowgraph::kDefaultBufferSize,
                    framesLeftInByteBuffer);
            // If the wrapping buffer is running low, write one frame at a time.
            if (framesAvailableInWrappingBuffer < flowgraph::kDefaultBufferSize) {
                framesToWriteFromByteBuffer = 1;
            }

            const int32_t numBytesToWriteFromByteBuffer = getBytesPerFrame() *
                    framesToWriteFromByteBuffer;

            //ALOGD("%s() framesLeftInByteBuffer %d, framesAvailableInWrappingBuffer %d"
            //      "framesToWriteFromByteBuffer %d, numBytesToWriteFromByteBuffer %d"
            //      , __func__, framesLeftInByteBuffer, framesAvailableInWrappingBuffer,
            //      framesToWriteFromByteBuffer, numBytesToWriteFromByteBuffer);

            const int32_t framesActuallyWrittenToWrappingBuffer = mFlowGraph.process(
                    (void *)byteBuffer,
                    framesToWriteFromByteBuffer,
                    (void *)currentWrappingBuffer,
                    framesAvailableInWrappingBuffer);

            byteBuffer += numBytesToWriteFromByteBuffer;
            framesLeftInByteBuffer -= framesToWriteFromByteBuffer;
            const int32_t numBytesActuallyWrittenToWrappingBuffer =
                    framesActuallyWrittenToWrappingBuffer * getBytesPerDeviceFrame();
            currentWrappingBuffer += numBytesActuallyWrittenToWrappingBuffer;
            framesAvailableInWrappingBuffer -= framesActuallyWrittenToWrappingBuffer;
            framesWrittenToAudioEndpoint += framesActuallyWrittenToWrappingBuffer;

            //ALOGD("%s() numBytesActuallyWrittenToWrappingBuffer %d, framesLeftInByteBuffer %d"
            //      "framesActuallyWrittenToWrappingBuffer %d, numBytesToWriteFromByteBuffer %d"
            //      "framesWrittenToAudioEndpoint %d"
            //      , __func__, numBytesActuallyWrittenToWrappingBuffer, framesLeftInByteBuffer,
            //      framesActuallyWrittenToWrappingBuffer, numBytesToWriteFromByteBuffer,
            //      framesWrittenToAudioEndpoint);
        }
        partIndex++;
    }
    //ALOGD("%s() framesWrittenToAudioEndpoint %d, numFrames %d"
    //              "framesLeftInByteBuffer %d"
    //              , __func__, framesWrittenToAudioEndpoint, numFrames,
    //              framesLeftInByteBuffer);

    // The audio endpoint should reference the number of frames written to the wrapping buffer.
    mAudioEndpoint->advanceWriteIndex(framesWrittenToAudioEndpoint);

    // The internal code should use the number of frames read from the app.
    return numFrames - framesLeftInByteBuffer;
}

int64_t AudioStreamInternalPlay::getFramesRead() {
    if (mAudioEndpoint) {
        const int64_t framesReadHardware = isClockModelInControl()
                ? mClockModel.convertTimeToPosition(AudioClock::getNanoseconds())
                : mAudioEndpoint->getDataReadCounter();
        // Add service offset and prevent retrograde motion.
        mLastFramesRead = std::max(mLastFramesRead, framesReadHardware + mFramesOffsetFromService);
    }
    return mLastFramesRead;
}

int64_t AudioStreamInternalPlay::getFramesWritten() {
    if (mAudioEndpoint) {
        mLastFramesWritten = std::max(
                mLastFramesWritten,
                mAudioEndpoint->getDataWriteCounter() + mFramesOffsetFromService);
    }
    return mLastFramesWritten;
}

aaudio_result_t AudioStreamInternalPlay::setOffloadEndOfStream() {
    if (getPerformanceMode() != AAUDIO_PERFORMANCE_MODE_POWER_SAVING_OFFLOADED ||
        getSharingMode() != AAUDIO_SHARING_MODE_EXCLUSIVE) {
        // Offload end of stream callback is only available for offload playback.
        // Offload playback must be exclusive mode.
        return AAUDIO_ERROR_UNIMPLEMENTED;
    }
    std::lock_guard<std::mutex> lock(mStreamMutex);
    if (getState() != AAUDIO_STREAM_STATE_STARTED || mClockModel.isStarting()) {
        // If the stream is not running or there is not timestamp from the service side,
        // it is not possible to set offload end of stream.
        return AAUDIO_ERROR_INVALID_STATE;
    }
    mOffloadEosPending = true;
    if (!isDataCallbackSet()) {
        const int64_t streamEndNanos = mClockModel.convertDeltaPositionToTime(
                std::max(0, mAudioEndpoint->getFullFramesAvailable() - getDeviceFramesPerBurst()));
        auto streamPtr = getPtr();
        mStreamEndExecutor->enqueue(android::mediautils::Runnable{
            [streamPtr, streamEndNanos]() {
                {
                    android::audio_utils::unique_lock ul(streamPtr->mStreamMutex);
                    streamPtr->mStreamEndCV.wait_for(
                            ul, std::chrono::nanoseconds(streamEndNanos),
                            [streamPtr]() REQUIRES(streamPtr->mStreamMutex) {
                                return !streamPtr->mOffloadEosPending;
                            });
                    if (!streamPtr->mOffloadEosPending) return;
                    streamPtr->mOffloadEosPending = false;
                }
                streamPtr->maybeCallPresentationEndCallback();
            }});
    }
    return AAUDIO_OK;
}

bool AudioStreamInternalPlay::shouldStopStream() {
    if (getPerformanceMode() != AAUDIO_PERFORMANCE_MODE_POWER_SAVING_OFFLOADED) {
        return true;
    }
    std::lock_guard _l(mStreamMutex);
    return !mOffloadEosPending;
}

void AudioStreamInternalPlay::maybeCallPresentationEndCallback() {
    if (mPresentationEndCallbackProc != nullptr) {
        pid_t expected = CALLBACK_THREAD_NONE;
        if (mPresentationEndCallbackThread.compare_exchange_strong(expected, gettid())) {
            (*mPresentationEndCallbackProc)(
                    (AAudioStream *) this, mPresentationEndCallbackUserData);
            mPresentationEndCallbackThread.store(CALLBACK_THREAD_NONE);
        } else {
            ALOGW("%s() presentation end callback already running!", __func__);
        }
    }
}

void AudioStreamInternalPlay::dropPresentationEndCallback_l() {
    mOffloadEosPending = false;
    mStreamEndCV.notify_one();
}

aaudio_result_t AudioStreamInternalPlay::requestStop_l() {
    // When stop is called, the service will notify the HAL so that no more data will be consumed.
    // In that case, it is no longer needed to wait for stream end.
    dropPresentationEndCallback_l();
    return AudioStreamInternal::requestStop_l();
}

void AudioStreamInternalPlay::wakeupCallbackThread_l() {
    if (!isDataCallbackSet()) {
        return;
    }
    mOffloadEosPending = false;
    mDraining = false;
    mStreamEndCV.notify_one();
    mCallbackCV.notify_one();
}

aaudio_result_t AudioStreamInternalPlay::flushFromFrame_l(
        AAudio_FlushFromAccuracy accuracy, int64_t* position) {
    if (getServiceHandle() == AAUDIO_HANDLE_INVALID) {
        ALOGD("%s() mServiceStreamHandle invalid", __func__);
        return AAUDIO_ERROR_DISCONNECTED;
    }
    if (isDisconnected()) {
        ALOGD("%s() but DISCONNECTED", __func__);
        return AAUDIO_ERROR_DISCONNECTED;
    }

    aaudio_result_t result = AAUDIO_OK;
    {
        std::lock_guard _endpointLock(mEndpointMutex);
        int64_t framesWritten = getFramesWritten();
        if (framesWritten < *position) {
            ALOGE("%s(), the requested position is not yet written", __func__);
            result = AAUDIO_ERROR_OUT_OF_RANGE;
        }

        // The position is updated from the server, it may not be very accurate if the stream has
        // been active for a while. In that case, updates the latest timestamp and then get the
        // actual rewind position again.
        if (aaudio_result_t res = mServiceInterface.updateTimestamp(mServiceStreamHandleInfo);
                res != AAUDIO_OK) {
            ALOGE("%s() failed to update timestamp, error=%d", __func__, res);
            return res;
        }
        processCommands();
        const int64_t safePosition = getFramesRead() + mOffloadFlushFromSafeMarginInFrames;
        if (safePosition > framesWritten) {
            ALOGE("%s() do not have enough data, safePosition=%jd, frameWritten=%jd",
                  __func__, safePosition, framesWritten);
            return AAUDIO_ERROR_OUT_OF_RANGE;
        }
        int64_t actualPosition = std::max(safePosition, *position);
        if (accuracy == AAUDIO_FLUSH_FROM_FRAME_ACCURATE && actualPosition != *position) {
            result = AAUDIO_ERROR_OUT_OF_RANGE;
        }
        *position = actualPosition;
        if (result != AAUDIO_OK) {
            return result;
        }

        // Rewind successfully, update the written position as the rewound position.
        mLastFramesWritten = actualPosition;
        mAudioEndpoint->setDataWriteCounter(actualPosition - mFramesOffsetFromService);
    }
    wakeupCallbackThread_l();
    return result;
}

aaudio_result_t AudioStreamInternalPlay::drainStream_l(int64_t wakeUpNanos,
                                                       bool allowSoftWakeUp) {
    aaudio_result_t result = mServiceInterface.drainStream(
            mServiceStreamHandleInfo, wakeUpNanos, allowSoftWakeUp, &mWakeUpHandle);
    if (result == AAUDIO_OK) {
        return result;
    }
    ALOGE("%s() failed, error=%d", __func__, result);
    processCommands();
    result = isDisconnected() ? AAUDIO_ERROR_DISCONNECTED : result;
    maybeCallErrorCallback(result);
    return result;
}

aaudio_result_t AudioStreamInternalPlay::activateStream_l() {
    aaudio_result_t result = mServiceInterface.activateStream(
            mServiceStreamHandleInfo, mWakeUpHandle);
    if (result == AAUDIO_OK) {
        mWakeUpHandle = TimerQueue::INVALID_HANDLE;
        return result;
    }
    ALOGE("%s() failed, error=%d", __func__, result);
    processCommands();
    result = isDisconnected() ? AAUDIO_ERROR_DISCONNECTED : result;
    maybeCallErrorCallback(result);
    return result;
}

aaudio_result_t AudioStreamInternalPlay::setPlaybackParameters_l(
        const AAudioPlaybackParameters* parameters) {
    if (getPerformanceMode() != AAUDIO_PERFORMANCE_MODE_POWER_SAVING_OFFLOADED) {
        // Setting playback parameters is not supported for offload stream.
        return AAUDIO_ERROR_UNIMPLEMENTED;
    }
    if (isAAudioPlaybackParametersEqual(*parameters, mPlaybackParameters)) {
        return AAUDIO_OK;
    }
    android::AudioPlaybackRate rate = android::AUDIO_PLAYBACK_RATE_DEFAULT;
    if (aaudio_result_t result =
                AAudioConvert_aaudioToAndroidPlaybackParameters(*parameters, &rate);
            result != AAUDIO_OK) {
        ALOGE("%s failed to convert to android playback parameters", __func__);
        return result;
    }
    if (!android::isAudioPlaybackRateValid(rate)) {
        ALOGE("%s failed, the playback parameters are not valid", __func__);
        return AAUDIO_ERROR_ILLEGAL_ARGUMENT;
    }

    const aaudio_result_t result = mServiceInterface.setPlaybackParameters(
            mServiceStreamHandleInfo, rate);
    if (result == AAUDIO_OK) {
        mPlaybackParameters = *parameters;
        // The playback speed is guaranteed to be greater than 0 by `isAudioPlaybackRateValid`.
        mClockModel.setPlaybackSpeed(mPlaybackParameters.speed);
    } else {
        ALOGE("%s failed, error=%d", __func__, result);
    }
    return result;
}

aaudio_result_t AudioStreamInternalPlay::getPlaybackParameters_l(
        AAudioPlaybackParameters* parameters) {
    if (getPerformanceMode() != AAUDIO_PERFORMANCE_MODE_POWER_SAVING_OFFLOADED) {
        // Setting playback parameters is not supported for offload stream.
        return AAUDIO_ERROR_UNIMPLEMENTED;
    }

    android::AudioPlaybackRate rate;
    aaudio_result_t result = mServiceInterface.getPlaybackParameters(
            mServiceStreamHandleInfo, &rate);
    if (result != AAUDIO_OK) {
        ALOGE("%s failed to query from service", __func__);
        return result;
    }
    AAudioPlaybackParameters tempParam;
    if (result = AAudioConvert_androidToAAudioPlaybackParameters(rate, &tempParam);
        result != AAUDIO_OK) {
        ALOGE("%s failed to convert to aaudio playback parameters", __func__);
        return result;
    }
    mPlaybackParameters = tempParam;
    mClockModel.setPlaybackSpeed(mPlaybackParameters.speed);
    *parameters = tempParam;
    return AAUDIO_OK;
}

// Render audio in the application callback and then write the data to the stream.
void *AudioStreamInternalPlay::callbackLoop() {
    ALOGD("%s() entering >>>>>>>>>>>>>>>", __func__);
    aaudio_result_t result = AAUDIO_OK;
    int32_t callbackResult = 0;
    if (!isDataCallbackSet()) return nullptr;
    int64_t timeoutNanos = calculateReasonableTimeout(mCallbackFrames);

    // result might be a frame count
    while (mCallbackEnabled.load() && isActive() && (result >= 0)) {
        processCommands();
        if (getPerformanceMode() == AAUDIO_PERFORMANCE_MODE_POWER_SAVING_OFFLOADED) {
            android::audio_utils::unique_lock ul(mStreamMutex);
            if (mOffloadEosPending) {
                const int64_t streamEndNanos = mClockModel.convertDeltaPositionToTime(std::max(0,
                        mAudioEndpoint->getFullFramesAvailable() - getDeviceFramesPerBurst()));
                const int64_t wakeUpNanos = android::elapsedRealtimeNano() + streamEndNanos;
                if (result = drainStream_l(wakeUpNanos, false /*allowSoftWakeUp*/);
                    result != AAUDIO_OK) {
                    ALOGE("%s() failed to drain, error=%d", __func__, result);
                    break;
                }
                mStreamEndCV.wait_for(ul, std::chrono::nanoseconds(streamEndNanos),
                                      [this]() REQUIRES(mStreamMutex) {
                    return !mOffloadEosPending;
                });
                if (mOffloadEosPending || android::elapsedRealtimeNano() >= wakeUpNanos) {
                    maybeCallPresentationEndCallback();
                    mOffloadEosPending = false;
                }
            }
        }
        {
            std::lock_guard _endpointLock(mEndpointMutex);
            // Call application using the AAudio callback interface.
            callbackResult = maybeCallDataCallback(mCallbackBuffer.get(), mCallbackFrames);
        }

        if (callbackResult < 0) {
            if (!shouldStopStream()) {
                ALOGD("%s(): callback request to stop but should not as it may be pending for"
                      "stream end", __func__);
                continue;
            }
            ALOGD("%s(): callback request to stop", __func__);
            result = systemStopInternal();
            break;
        }

        // Write audio data to stream. This is a BLOCKING WRITE!
        // Write data regardless of the callbackResult because we assume the data
        // is valid even when the callback returns AAUDIO_CALLBACK_RESULT_STOP.
        // Imagine a callback that is playing a large sound in menory.
        // When it gets to the end of the sound it can partially fill
        // the last buffer with the end of the sound, then zero pad the buffer, then return STOP.
        // If the callback has no valid data then it should zero-fill the entire buffer.
        result = write(mCallbackBuffer.get(), callbackResult, timeoutNanos);
        if (result != callbackResult) {
            break;
        }
    }

    ALOGD("%s() exiting, result = %d, isActive() = %d <<<<<<<<<<<<<<",
          __func__, result, (int) isActive());
    return nullptr;
}

//------------------------------------------------------------------------------
// Implementation of PlayerBase
status_t AudioStreamInternalPlay::doSetVolume() {
    float combinedVolume = mStreamVolume * getDuckAndMuteVolume();
    ALOGD("%s() mStreamVolume * duckAndMuteVolume = %f * %f = %f",
          __func__, mStreamVolume, getDuckAndMuteVolume(), combinedVolume);
    mFlowGraph.setTargetVolume(combinedVolume);
    return android::NO_ERROR;
}

//------------------------------------------------------------------------------
// Implementation of AAudioClientCallback
void AudioStreamInternalPlay::onWakeUp_l(android::audio_utils::TimerQueue::handle_t handle) {
    if (handle != mWakeUpHandle) {
        ALOGW("%s the wake up handle does not match %jd %jd", __func__, handle, mWakeUpHandle);
    }
    mWakeUpHandle = TimerQueue::INVALID_HANDLE;
    wakeupCallbackThread_l();
}