Extract threading logic to EffectThread and other changes

CMakeLists.txt

@@ -161,6 +161,8 @@ set(FILES
         src/shim.cpp
 
         # Main
+        src/aidl/EffectThread.cpp
+
         src/viper/ViPER.cpp
         src/ViPER4Android.cpp
         src/ViperContext.cpp

src/aidl/EffectThread.cpp

@@ -0,0 +1,130 @@
+/*
+ * Copyright (C) 2022 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.
+ */
+
+#include <cstddef>
+#include <memory>
+
+#define LOG_TAG "AHAL_EffectThread"
+#include <log/log.h>
+#include <pthread.h>
+#include <sys/resource.h>
+
+#include "EffectThread.h"
+#include "EffectTypes.h"
+
+namespace aidl::android::hardware::audio::effect {
+
+    EffectThread::~EffectThread() {
+        destroyThread();
+    }
+
+    RetCode EffectThread::createThread(const std::string& name, int priority) {
+        if (mThread.joinable()) {
+            ALOGW("%s %s thread already created, no-op", mName.c_str(), __func__);
+            return RetCode::SUCCESS;
+        }
+
+        mName = name;
+        mPriority = priority;
+        {
+            std::lock_guard lg(mThreadMutex);
+            mStop = true;
+            mExit = false;
+        }
+
+        mThread = std::thread(&EffectThread::threadLoop, this);
+        ALOGV("%s %s priority %d done", mName.c_str(), __func__, mPriority);
+        return RetCode::SUCCESS;
+    }
+
+    RetCode EffectThread::destroyThread() {
+        {
+            std::lock_guard lg(mThreadMutex);
+            mStop = mExit = true;
+        }
+
+        mCv.notify_one();
+        if (mThread.joinable()) {
+            mThread.join();
+        }
+
+        ALOGV("%s", mName.c_str());
+        return RetCode::SUCCESS;
+    }
+
+    RetCode EffectThread::startThread() {
+        {
+            std::lock_guard lg(mThreadMutex);
+            if (mDraining) {
+                mDraining = false;
+            } else {
+                mStop = false;
+            }
+            mCv.notify_one();
+        }
+
+        ALOGV("%s", mName.c_str());
+        return RetCode::SUCCESS;
+    }
+
+    RetCode EffectThread::stopThread() {
+        {
+            std::lock_guard lg(mThreadMutex);
+            mStop = true;
+            mCv.notify_one();
+        }
+
+        ALOGV("%s", mName.c_str());
+        return RetCode::SUCCESS;
+    }
+
+    RetCode EffectThread::startDraining() {
+        std::lock_guard lg(mThreadMutex);
+        mDraining = true;
+        mCv.notify_one();
+
+        ALOGV("%s", mName.c_str());
+        return RetCode::SUCCESS;
+    }
+
+    RetCode EffectThread::finishDraining() {
+        std::lock_guard lg(mThreadMutex);
+        mDraining = false;
+        mStop = true;
+        mCv.notify_one();
+
+        ALOGV("%s", mName.c_str());
+        return RetCode::SUCCESS;
+    }
+
+    void EffectThread::threadLoop() {
+        pthread_setname_np(pthread_self(), mName.substr(0, kMaxTaskNameLen - 1).c_str());
+        setpriority(PRIO_PROCESS, 0, mPriority);
+        while (true) {
+            {
+                std::unique_lock l(mThreadMutex);
+                ::android::base::ScopedLockAssertion lock_assertion(mThreadMutex);
+                mCv.wait(l, [&]() REQUIRES(mThreadMutex) { return mExit || !mStop; });
+                if (mExit) {
+                    ALOGV("%s threadLoop EXIT!", mName.c_str());
+                    return;
+                }
+            }
+            process();
+        }
+    }
+
+}  // namespace aidl::android::hardware::audio::effect

src/aidl/EffectThread.h

@@ -0,0 +1,69 @@
+/*
+ * Copyright (C) 2022 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.
+ */
+
+#pragma once
+#include <atomic>
+#include <memory>
+#include <string>
+#include <thread>
+
+#include <android-base/thread_annotations.h>
+#include <fmq/EventFlag.h>
+#include <system/thread_defs.h>
+
+//#include "EffectContext.h"
+#include "EffectTypes.h"
+
+namespace aidl::android::hardware::audio::effect {
+
+    class EffectThread {
+    public:
+        virtual ~EffectThread();
+
+        // called by effect implementation
+        RetCode createThread(const std::string& name, int priority = ANDROID_PRIORITY_URGENT_AUDIO);
+        RetCode destroyThread();
+        RetCode startThread();
+        RetCode stopThread();
+        RetCode startDraining();
+        RetCode finishDraining();
+
+        // Will call process() in a loop if the thread is running.
+        void threadLoop();
+
+        /**
+         * process() call effectProcessImpl() for effect data processing, it is necessary for the
+         * processing to be called under Effect thread mutex mThreadMutex, to avoid the effect state
+         * change before/during data processing, and keep the thread and effect state consistent.
+         */
+        virtual void process() = 0;
+
+    protected:
+        bool mDraining GUARDED_BY(mThreadMutex) = false;
+
+    private:
+        static constexpr int kMaxTaskNameLen = 15;
+
+        std::mutex mThreadMutex;
+        std::condition_variable mCv;
+        bool mStop GUARDED_BY(mThreadMutex) = true;
+        bool mExit GUARDED_BY(mThreadMutex) = false;
+
+        std::thread mThread;
+        int mPriority;
+        std::string mName;
+    };
+}  // namespace aidl::android::hardware::audio::effect

src/aidl/EffectTypes.h

@@ -0,0 +1,133 @@
+/*
+ * Copyright (C) 2022 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.
+ */
+
+#pragma once
+#include <string>
+
+#include <aidl/android/hardware/audio/effect/BnEffect.h>
+#include <aidl/android/hardware/audio/effect/Range.h>
+#include <android-base/logging.h>
+//#include <system/audio_effects/aidl_effects_utils.h>
+
+typedef binder_exception_t (*EffectCreateFunctor)(
+        const ::aidl::android::media::audio::common::AudioUuid*,
+        std::shared_ptr<::aidl::android::hardware::audio::effect::IEffect>*);
+typedef binder_exception_t (*EffectDestroyFunctor)(
+        const std::shared_ptr<::aidl::android::hardware::audio::effect::IEffect>&);
+typedef binder_exception_t (*EffectQueryFunctor)(
+        const ::aidl::android::media::audio::common::AudioUuid*,
+        ::aidl::android::hardware::audio::effect::Descriptor*);
+
+struct effect_dl_interface_s {
+    EffectCreateFunctor createEffectFunc;
+    EffectDestroyFunctor destroyEffectFunc;
+    EffectQueryFunctor queryEffectFunc;
+};
+
+namespace aidl::android::hardware::audio::effect {
+
+    enum class RetCode {
+        SUCCESS,
+        ERROR_ILLEGAL_PARAMETER, /* Illegal parameter */
+        ERROR_THREAD,            /* Effect thread error */
+        ERROR_NULL_POINTER,      /* NULL pointer */
+        ERROR_ALIGNMENT_ERROR,   /* Memory alignment error */
+        ERROR_BLOCK_SIZE_EXCEED, /* Maximum block size exceeded */
+        ERROR_EFFECT_LIB_ERROR,  /* Effect implementation library error */
+        ERROR_EVENT_FLAG_ERROR   /* Error with effect event flags */
+    };
+
+    static const int INVALID_AUDIO_SESSION_ID = -1;
+
+    inline std::ostream& operator<<(std::ostream& out, const RetCode& code) {
+        switch (code) {
+            case RetCode::SUCCESS:
+                return out << "SUCCESS";
+            case RetCode::ERROR_ILLEGAL_PARAMETER:
+                return out << "ERROR_ILLEGAL_PARAMETER";
+            case RetCode::ERROR_THREAD:
+                return out << "ERROR_THREAD";
+            case RetCode::ERROR_NULL_POINTER:
+                return out << "ERROR_NULL_POINTER";
+            case RetCode::ERROR_ALIGNMENT_ERROR:
+                return out << "ERROR_ALIGNMENT_ERROR";
+            case RetCode::ERROR_BLOCK_SIZE_EXCEED:
+                return out << "ERROR_BLOCK_SIZE_EXCEED";
+            case RetCode::ERROR_EFFECT_LIB_ERROR:
+                return out << "ERROR_EFFECT_LIB_ERROR";
+            case RetCode::ERROR_EVENT_FLAG_ERROR:
+                return out << "ERROR_EVENT_FLAG_ERROR";
+        }
+
+        return out << "EnumError: " << code;
+    }
+
+#define RETURN_IF_ASTATUS_NOT_OK(status, message)                                               \
+    do {                                                                                        \
+        const ::ndk::ScopedAStatus curr_status = (status);                                      \
+        if (!curr_status.isOk()) {                                                              \
+            LOG(ERROR) << __func__ << ": line" << __LINE__                                      \
+                       << " return with status: " << curr_status.getDescription() << (message); \
+            return ndk::ScopedAStatus::fromExceptionCodeWithMessage(                            \
+                    curr_status.getExceptionCode(), (message));                                 \
+        }                                                                                       \
+    } while (0)
+
+#define RETURN_IF(expr, exception, message)                                                  \
+    do {                                                                                     \
+        if (expr) {                                                                          \
+            LOG(ERROR) << __func__ << ": line" << __LINE__ << " return with expr " << #expr; \
+            return ndk::ScopedAStatus::fromExceptionCodeWithMessage((exception), (message)); \
+        }                                                                                    \
+    } while (0)
+
+#define RETURN_OK_IF(expr)                                                                  \
+    do {                                                                                    \
+        if (expr) {                                                                         \
+            LOG(INFO) << __func__ << ": line" << __LINE__ << " return with expr " << #expr; \
+            return ndk::ScopedAStatus::ok();                                                \
+        }                                                                                   \
+    } while (0)
+
+#define RETURN_VALUE_IF(expr, ret, log)                                                       \
+    do {                                                                                      \
+        if (expr) {                                                                           \
+            LOG(ERROR) << __func__ << ": line" << __LINE__ << " return with expr \"" << #expr \
+                       << "\":" << (log);                                                     \
+            return ret;                                                                       \
+        }                                                                                     \
+    } while (0)
+
+#define RETURN_IF_BINDER_EXCEPTION(functor)                                \
+    {                                                                      \
+        binder_exception_t exception = functor;                            \
+        if (EX_NONE != exception) {                                        \
+            LOG(ERROR) << #functor << ": failed with error " << exception; \
+            return ndk::ScopedAStatus::fromExceptionCode(exception);       \
+        }                                                                  \
+    }
+
+/**
+ * Make a Range::$EffectType$Range.
+ * T: The $EffectType$, Visualizer for example.
+ * Tag: The union tag name in $EffectType$ definition, latencyMs for example.
+ * l: The value of Range::$EffectType$Range.min.
+ * r: The value of Range::$EffectType$Range.max.
+ */
+#define MAKE_RANGE(T, Tag, l, r) \
+    { .min = T::make<T::Tag>(l), .max = T::make<T::Tag>(r) }
+
+}  // namespace aidl::android::hardware::audio::effect

src/shim.cpp

@@ -1,19 +1,20 @@
 #include <string>
+#include <log/log.h>
 
 namespace android::hardware::details {
     void check(bool exp) {
-
+        ALOGE_IF(!exp, "Check failed");
     }
 
     void check(bool exp, const char* message) {
-
+        ALOGE_IF(!exp, "%s", message);
     }
 
     void logError(const std::string &message) {
-
+        ALOGE("%s", message.c_str());
     }
 
     void errorWriteLog(int tag, const char* info) {
-
+        ALOGE("%d: %s", tag, info);
     }
 } // namespace android::hardware::details

src/viper_aidl.cpp

@@ -8,6 +8,7 @@ using aidl::android::hardware::audio::effect::Descriptor;
 using aidl::android::hardware::audio::effect::Flags;
 using aidl::android::hardware::audio::effect::IEffect;
 using aidl::android::hardware::audio::effect::Parameter;
+using aidl::android::hardware::audio::effect::RetCode;
 using aidl::android::hardware::audio::effect::State;
 using aidl::android::media::audio::common::AudioUuid;
 using aidl::android::media::audio::common::PcmType;
@@ -113,6 +114,18 @@ constexpr size_t getFrameSizeInBytes(
     return 0;
 }
 
+//int ViPER4AndroidAIDL::notifyEventFlag(uint32_t flag) {
+//    if (!mEventFlag) {
+//        ALOGE("notifyEventFlag: StatusEventFlag invalid");
+//        return -1;
+//    }
+//    if (const auto ret = mEventFlag->wake(flag); ret != ::android::OK) {
+//        ALOGE("notifyEventFlag: wake failure with ret %d", ret);
+//        return -1;
+//    }
+//    return 0;
+//}
+
 ndk::ScopedAStatus ViPER4AndroidAIDL::open(const Parameter::Common &common,
                                            const std::optional<Parameter::Specific> &specific,
                                            IEffect::OpenEffectReturn *ret) {
@@ -121,7 +134,7 @@ ndk::ScopedAStatus ViPER4AndroidAIDL::open(const Parameter::Common &common,
         return ndk::ScopedAStatus::fromExceptionCode(EX_ILLEGAL_ARGUMENT);
     }
 
-    std::lock_guard lg(mMutex);
+    std::lock_guard lg(mImplMutex);
 
     if (mState != State::INIT) {
         ALOGD("open: already opened");
@@ -165,7 +178,10 @@ ndk::ScopedAStatus ViPER4AndroidAIDL::open(const Parameter::Common &common,
     ret->inputDataMQ = mInputMQ->dupeDesc();
     ret->outputDataMQ = mOutputMQ->dupeDesc();
 
-    mThread = std::thread(&ViPER4AndroidAIDL::threadLoop, this);
+    if (createThread(VIPER_NAME) != RetCode::SUCCESS) {
+        ALOGE("open: failed to create thread");
+        return ndk::ScopedAStatus::fromExceptionCode(EX_UNSUPPORTED_OPERATION);
+    }
 
     return ndk::ScopedAStatus::ok();
 }
@@ -177,23 +193,59 @@ ndk::ScopedAStatus ViPER4AndroidAIDL::close() {
 
 ndk::ScopedAStatus
 ViPER4AndroidAIDL::getDescriptor(Descriptor *descriptor) {
-    if (descriptor == nullptr) {
-        ALOGE("getDescriptor called with null descriptor");
-        return ndk::ScopedAStatus::fromExceptionCode(EX_ILLEGAL_ARGUMENT);
-    }
     ALOGD("getDescriptor: returning descriptor");
     *descriptor = kDescriptor;
     return ndk::ScopedAStatus::ok();
 }
 
-ndk::ScopedAStatus ViPER4AndroidAIDL::command(CommandId command_id) {
-    ALOGD("command called");
+ndk::ScopedAStatus ViPER4AndroidAIDL::command(CommandId id) {
+    std::lock_guard lg(mImplMutex);
+    if (mState == State::INIT) {
+        ALOGE("command: instance not open");
+        return ndk::ScopedAStatus::fromExceptionCode(EX_ILLEGAL_STATE);
+    }
+
+    switch (id) {
+        case CommandId::START: {
+            ALOGD("command: START");
+
+            if (mState == State::PROCESSING) {
+                return ndk::ScopedAStatus::ok();
+            }
+            mState = State::PROCESSING;
+
+//            if (notifyEventFlag() != 0) {
+//                ALOGE("id: failed to notify event flag");
+//                return ndk::ScopedAStatus::fromExceptionCode(EX_ILLEGAL_STATE);
+//            }
+            startThread();
+            break;
+        }
+        case CommandId::STOP: {
+            ALOGD("command: STOP");
+
+            if (mState == State::IDLE) {
+                return ndk::ScopedAStatus::ok();
+            }
+            mState = State::IDLE;
+
+            stopThread();
+            break;
+        }
+        case CommandId::RESET: {
+            ALOGD("command: RESET");
+            mState = State::IDLE;
+            stopThread();
+            break;
+        }
+    }
     return ndk::ScopedAStatus::fromExceptionCode(EX_UNSUPPORTED_OPERATION);
 }
 
 ndk::ScopedAStatus ViPER4AndroidAIDL::getState(State *state) {
-    ALOGD("getState called");
-    return ndk::ScopedAStatus::fromExceptionCode(EX_UNSUPPORTED_OPERATION);
+    ALOGD("getState: returning state");
+    *state = mState;
+    return ndk::ScopedAStatus::ok();
 }
 
 ndk::ScopedAStatus
@@ -203,64 +255,24 @@ ViPER4AndroidAIDL::setParameter(const Parameter &parameter) {
 }
 
 ndk::ScopedAStatus
-ViPER4AndroidAIDL::getParameter(const Parameter::Id &parameter_id, Parameter *parameter) {
+ViPER4AndroidAIDL::getParameter(const Parameter::Id &id, Parameter *param) {
     ALOGD("getParameter called");
     return ndk::ScopedAStatus::fromExceptionCode(EX_UNSUPPORTED_OPERATION);
 }
 
 ndk::ScopedAStatus
 ViPER4AndroidAIDL::reopen(IEffect::OpenEffectReturn *open_effect_return) {
-    ALOGD("reopen called");
-    return ndk::ScopedAStatus::fromExceptionCode(EX_UNSUPPORTED_OPERATION);
-}
-
-void ViPER4AndroidAIDL::threadLoop() {
-    ALOGD("threadLoop started");
-    while (true) {
-        {
-            std::unique_lock l(mThreadMutex);
-            ::android::base::ScopedLockAssertion lock_assertion(mThreadMutex);
-            mThreadCv.wait(l, [&]() REQUIRES(mThreadMutex) { return mThreadExit || !mThreadStop; });
-            if (mThreadExit) {
-                ALOGD("threadLoop exiting");
-                return;
-            }
-        }
-        process();
+    std::lock_guard lg(mImplMutex);
+    if (mState == State::INIT) {
+        ALOGE("reopen: already closed");
+        return ndk::ScopedAStatus::fromExceptionCode(EX_ILLEGAL_STATE);
     }
+    // TODO
+    return ndk::ScopedAStatus::ok();
 }
 
 void ViPER4AndroidAIDL::process() {
-/**
-     * wait for the EventFlag without lock, it's ok because the mEventFlag pointer will not change
-     * in the life cycle of workerThread (threadLoop).
-     */
-    uint32_t efState = 0;
-    if (!mEventFlag ||
-        mEventFlag->wait(kEventFlagDataMqNotEmpty, &efState, 0 /* no timeout */, true /* retry */) != android::OK ||
-        !(efState & kEventFlagDataMqNotEmpty)) {
-        ALOGE("process: failed to wait for event flag");
-        return;
-    }
-
-    {
-        std::lock_guard lg(mMutex);
-        if (mState != State::PROCESSING && mState != State::DRAINING) {
-            ALOGD("process: skip process in state: %d", mState);
-            return;
-        }
-
-        auto buffer = mWorkBuffer.data();
-        auto processSamples = std::min(mInputMQ->availableToRead(), mOutputMQ->availableToWrite());
-        if (processSamples) {
-            mInputMQ->read(buffer, processSamples);
-//            IEffect::Status status = effectProcessImpl(buffer, buffer, processSamples);
-            ALOGD("process: processing %zu samples", processSamples);
-            IEffect::Status status = {STATUS_OK, static_cast<int32_t>(processSamples), static_cast<int32_t>(processSamples)};
-            mOutputMQ->write(buffer, status.fmqProduced);
-            mStatusMQ->writeBlocking(&status, 1);
-        }
-    }
+    ALOGD("process called");
 }
 
 extern "C" binder_exception_t createEffect(const AudioUuid *audio_uuid, std::shared_ptr<IEffect> *instance) {

src/viper_aidl.h

@@ -5,20 +5,26 @@
 #include <aidl/android/hardware/audio/effect/BnEffect.h>
 #include <aidl/android/hardware/audio/effect/State.h>
 #include <fmq/AidlMessageQueue.h>
+#include "aidl/EffectThread.h"
 
 using aidl::android::hardware::audio::effect::BnEffect;
+using aidl::android::hardware::audio::effect::EffectThread;
 using aidl::android::hardware::audio::effect::State;
 
-class ViPER4AndroidAIDL : public BnEffect {
+class ViPER4AndroidAIDL : public BnEffect, public EffectThread {
 public:
+    // BnEffect
     ::ndk::ScopedAStatus open(const ::aidl::android::hardware::audio::effect::Parameter::Common &common, const std::optional< ::aidl::android::hardware::audio::effect::Parameter::Specific> &specific, ::aidl::android::hardware::audio::effect::IEffect::OpenEffectReturn *ret) override;
     ::ndk::ScopedAStatus close() override;
     ::ndk::ScopedAStatus getDescriptor(::aidl::android::hardware::audio::effect::Descriptor *_aidl_return) override;
-    ::ndk::ScopedAStatus command(::aidl::android::hardware::audio::effect::CommandId in_commandId) override;
+    ::ndk::ScopedAStatus command(::aidl::android::hardware::audio::effect::CommandId id) override;
     ::ndk::ScopedAStatus getState(::aidl::android::hardware::audio::effect::State *_aidl_return) override;
     ::ndk::ScopedAStatus setParameter(const ::aidl::android::hardware::audio::effect::Parameter &in_param) override;
     ::ndk::ScopedAStatus getParameter(const ::aidl::android::hardware::audio::effect::Parameter::Id &in_paramId, ::aidl::android::hardware::audio::effect::Parameter *_aidl_return) override;
     ::ndk::ScopedAStatus reopen(::aidl::android::hardware::audio::effect::IEffect::OpenEffectReturn *_aidl_return) override;
+
+    // EffectThread
+    void process() override;
 private:
     typedef ::android::AidlMessageQueue<
             IEffect::Status, ::aidl::android::hardware::common::fmq::SynchronizedReadWrite>
@@ -27,10 +33,7 @@ private:
             float, ::aidl::android::hardware::common::fmq::SynchronizedReadWrite>
             DataMQ;
 
-    void threadLoop();
-    void process();
-
-    std::mutex mMutex;
+    std::mutex mImplMutex;
     State mState = State::INIT;
 
     std::shared_ptr<StatusMQ> mStatusMQ;
@@ -38,10 +41,4 @@ private:
     std::shared_ptr<DataMQ> mOutputMQ;
     android::hardware::EventFlag *mEventFlag;
     std::vector<float> mWorkBuffer;
-
-    std::thread mThread;
-    std::mutex mThreadMutex;
-    std::condition_variable mThreadCv;
-    bool mThreadStop = true;
-    bool mThreadExit = false;
 };