audiohal: Make sure audio data transfer related commands go via FMQ

When outputting audio, the framework issues several HAL calls
from the same thread that writes into data FMQ. These calls
also need to be served on the same thread that writes audio data
to HAL. The same thing happens when audio input is commenced.

Add a command FMQ for passing different commands to the HAL thread.
This way, depending on the calling thread, the same call may go
either via hwbinder or via the command queue.

This dramatically reduces jitter in RTT measurements (although
doesn't improve the latency).

Bug: 30222631
Test: scripted RTT app
Change-Id: I04c826e2479d8210fd9c99756241156cda3143b6

1 files changed, 80 insertions, 34 deletions

diff --git a/audio/2.0/default/StreamIn.cpp b/audio/2.0/default/StreamIn.cpp
index ad18986401..1cde4ac1b9 100644
--- a/audio/2.0/default/StreamIn.cpp
+++ b/audio/2.0/default/StreamIn.cpp
@@ -38,6 +38,7 @@ class ReadThread : public Thread {
     // ReadThread's lifespan never exceeds StreamIn's lifespan.
     ReadThread(std::atomic<bool>* stop,
             audio_stream_in_t* stream,
+            StreamIn::CommandMQ* commandMQ,
             StreamIn::DataMQ* dataMQ,
             StreamIn::StatusMQ* statusMQ,
             EventFlag* efGroup,
@@ -45,6 +46,7 @@ class ReadThread : public Thread {
             : Thread(false /*canCallJava*/),
               mStop(stop),
               mStream(stream),
+              mCommandMQ(commandMQ),
               mDataMQ(dataMQ),
               mStatusMQ(statusMQ),
               mEfGroup(efGroup),
@@ -58,13 +60,19 @@ class ReadThread : public Thread {
   private:
     std::atomic<bool>* mStop;
     audio_stream_in_t* mStream;
+    StreamIn::CommandMQ* mCommandMQ;
     StreamIn::DataMQ* mDataMQ;
     StreamIn::StatusMQ* mStatusMQ;
     EventFlag* mEfGroup;
     ThreadPriority mThreadPriority;
     std::unique_ptr<uint8_t[]> mBuffer;
+    IStreamIn::ReadParameters mParameters;
+    IStreamIn::ReadStatus mStatus;
 
     bool threadLoop() override;
+
+    void doGetCapturePosition();
+    void doRead();
 };
 
 status_t ReadThread::readyToRun() {
@@ -77,6 +85,32 @@ status_t ReadThread::readyToRun() {
     return OK;
 }
 
+void ReadThread::doRead() {
+    size_t availableToWrite = mDataMQ->availableToWrite();
+    size_t requestedToRead = mParameters.params.read;
+    if (requestedToRead > availableToWrite) {
+        ALOGW("truncating read data from %d to %d due to insufficient data queue space",
+                (int32_t)requestedToRead, (int32_t)availableToWrite);
+        requestedToRead = availableToWrite;
+    }
+    ssize_t readResult = mStream->read(mStream, &mBuffer[0], requestedToRead);
+    mStatus.retval = Result::OK;
+    uint64_t read = 0;
+    if (readResult >= 0) {
+        mStatus.reply.read = readResult;
+        if (!mDataMQ->write(&mBuffer[0], readResult)) {
+            ALOGW("data message queue write failed");
+        }
+    } else {
+        mStatus.retval = Stream::analyzeStatus("read", readResult);
+    }
+}
+
+void ReadThread::doGetCapturePosition() {
+    mStatus.retval = StreamIn::getCapturePositionImpl(
+            mStream, &mStatus.reply.capturePosition.frames, &mStatus.reply.capturePosition.time);
+}
+
 bool ReadThread::threadLoop() {
     // This implementation doesn't return control back to the Thread until it decides to stop,
     // as the Thread uses mutexes, and this can lead to priority inversion.
@@ -87,21 +121,23 @@ bool ReadThread::threadLoop() {
         if (!(efState & static_cast<uint32_t>(MessageQueueFlagBits::NOT_FULL))) {
             continue;  // Nothing to do.
         }
-
-        const size_t availToWrite = mDataMQ->availableToWrite();
-        ssize_t readResult = mStream->read(mStream, &mBuffer[0], availToWrite);
-        Result retval = Result::OK;
-        uint64_t read = 0;
-        if (readResult >= 0) {
-            read = readResult;
-            if (!mDataMQ->write(&mBuffer[0], readResult)) {
-                ALOGW("data message queue write failed");
-            }
-        } else {
-            retval = Stream::analyzeStatus("read", readResult);
+        if (!mCommandMQ->read(&mParameters)) {
+            continue;  // Nothing to do.
         }
-        IStreamIn::ReadStatus status = { retval, read };
-        if (!mStatusMQ->write(&status)) {
+        mStatus.replyTo = mParameters.command;
+        switch (mParameters.command) {
+            case IStreamIn::ReadCommand::READ:
+                doRead();
+                break;
+            case IStreamIn::ReadCommand::GET_CAPTURE_POSITION:
+                doGetCapturePosition();
+                break;
+            default:
+                ALOGE("Unknown read thread command code %d", mParameters.command);
+                mStatus.retval = Result::NOT_SUPPORTED;
+                break;
+        }
+        if (!mStatusMQ->write(&mStatus)) {
             ALOGW("status message queue write failed");
         }
         mEfGroup->wake(static_cast<uint32_t>(MessageQueueFlagBits::NOT_EMPTY));
@@ -275,17 +311,19 @@ Return<void> StreamIn::prepareForReading(
     if (mDataMQ) {
         ALOGE("the client attempts to call prepareForReading twice");
         _hidl_cb(Result::INVALID_STATE,
-                DataMQ::Descriptor(), StatusMQ::Descriptor());
+                CommandMQ::Descriptor(), DataMQ::Descriptor(), StatusMQ::Descriptor());
         return Void();
     }
+    std::unique_ptr<CommandMQ> tempCommandMQ(new CommandMQ(1));
     std::unique_ptr<DataMQ> tempDataMQ(
             new DataMQ(frameSize * framesCount, true /* EventFlag */));
     std::unique_ptr<StatusMQ> tempStatusMQ(new StatusMQ(1));
-    if (!tempDataMQ->isValid() || !tempStatusMQ->isValid()) {
+    if (!tempCommandMQ->isValid() || !tempDataMQ->isValid() || !tempStatusMQ->isValid()) {
+        ALOGE_IF(!tempCommandMQ->isValid(), "command MQ is invalid");
         ALOGE_IF(!tempDataMQ->isValid(), "data MQ is invalid");
         ALOGE_IF(!tempStatusMQ->isValid(), "status MQ is invalid");
         _hidl_cb(Result::INVALID_ARGUMENTS,
-                DataMQ::Descriptor(), StatusMQ::Descriptor());
+                CommandMQ::Descriptor(), DataMQ::Descriptor(), StatusMQ::Descriptor());
         return Void();
     }
     // TODO: Remove event flag management once blocking MQ is implemented. b/33815422
@@ -293,7 +331,7 @@ Return<void> StreamIn::prepareForReading(
     if (status != OK || !mEfGroup) {
         ALOGE("failed creating event flag for data MQ: %s", strerror(-status));
         _hidl_cb(Result::INVALID_ARGUMENTS,
-                DataMQ::Descriptor(), StatusMQ::Descriptor());
+                CommandMQ::Descriptor(), DataMQ::Descriptor(), StatusMQ::Descriptor());
         return Void();
     }
 
@@ -301,6 +339,7 @@ Return<void> StreamIn::prepareForReading(
     mReadThread = new ReadThread(
             &mStopReadThread,
             mStream,
+            tempCommandMQ.get(),
             tempDataMQ.get(),
             tempStatusMQ.get(),
             mEfGroup,
@@ -309,13 +348,14 @@ Return<void> StreamIn::prepareForReading(
     if (status != OK) {
         ALOGW("failed to start reader thread: %s", strerror(-status));
         _hidl_cb(Result::INVALID_ARGUMENTS,
-                DataMQ::Descriptor(), StatusMQ::Descriptor());
+                CommandMQ::Descriptor(), DataMQ::Descriptor(), StatusMQ::Descriptor());
         return Void();
     }
 
+    mCommandMQ = std::move(tempCommandMQ);
     mDataMQ = std::move(tempDataMQ);
     mStatusMQ = std::move(tempStatusMQ);
-    _hidl_cb(Result::OK, *mDataMQ->getDesc(), *mStatusMQ->getDesc());
+    _hidl_cb(Result::OK, *mCommandMQ->getDesc(), *mDataMQ->getDesc(), *mStatusMQ->getDesc());
     return Void();
 }
 
@@ -323,22 +363,28 @@ Return<uint32_t> StreamIn::getInputFramesLost()  {
     return mStream->get_input_frames_lost(mStream);
 }
 
-Return<void> StreamIn::getCapturePosition(getCapturePosition_cb _hidl_cb)  {
+// static
+Result StreamIn::getCapturePositionImpl(
+        audio_stream_in_t *stream, uint64_t *frames, uint64_t *time) {
     Result retval(Result::NOT_SUPPORTED);
-    uint64_t frames = 0, time = 0;
-    if (mStream->get_capture_position != NULL) {
-        int64_t halFrames, halTime;
-        retval = Stream::analyzeStatus(
-                "get_capture_position",
-                mStream->get_capture_position(mStream, &halFrames, &halTime),
-                // HAL may have a stub function, always returning ENOSYS, don't
-                // spam the log in this case.
-                ENOSYS);
-        if (retval == Result::OK) {
-            frames = halFrames;
-            time = halTime;
-        }
+    if (stream->get_capture_position != NULL) return retval;
+    int64_t halFrames, halTime;
+    retval = Stream::analyzeStatus(
+            "get_capture_position",
+            stream->get_capture_position(stream, &halFrames, &halTime),
+            // HAL may have a stub function, always returning ENOSYS, don't
+            // spam the log in this case.
+            ENOSYS);
+    if (retval == Result::OK) {
+        *frames = halFrames;
+        *time = halTime;
     }
+    return retval;
+};
+
+Return<void> StreamIn::getCapturePosition(getCapturePosition_cb _hidl_cb)  {
+    uint64_t frames = 0, time = 0;
+    Result retval = getCapturePositionImpl(mStream, &frames, &time);
     _hidl_cb(retval, frames, time);
     return Void();
 }