Donate to e Foundation | Murena handsets with /e/OS | Own a part of Murena! Learn more

Commit 01df7e8c authored by Mikhail Naganov's avatar Mikhail Naganov
Browse files

audio: Fix stream cleanup sequence 

Move the cleanup of the stream worker thread from '~StreamCommonImpl'
up to concrete stream implementations. This is because when
the worker thread is stopping, it calls 'DriverInterface::shutdown'
method of the stream. At the time when '~StreamCommonImpl' is
running, the concrete stream class has already been destroyed.

The cleanup actually only happens in the case when the client
did not close the stream properly via 'IStreamCommon.close', or
when the stream creation has failed in the middle.

Bug: 355804294
Test: atest VtsHalAudioCoreTargetTest
(cherry picked from https://android-review.googlesource.com/q/commit:0413d077f76df5fe464e4f39ab1efa091df1019e)
Merged-In: Ie86f682af202976ed48d24338b2dffcfd20d9a76
Change-Id: Ie86f682af202976ed48d24338b2dffcfd20d9a76
parent 7a31a0ac

audio/aidl/default/Stream.cpp

+24 −8
Original line number Original line Diff line number Diff line
@@ -663,10 +663,14 @@ bool StreamOutWorkerLogic::write(size_t clientSize, StreamDescriptor::Reply* rep 
}

}





StreamCommonImpl::~StreamCommonImpl() {

StreamCommonImpl::~StreamCommonImpl() {

    if (!isClosed()) {

    // It is responsibility of the class that implements 'DriverInterface' to call 'cleanupWorker'

        LOG(ERROR) << __func__ << ": stream was not closed prior to destruction, resource leak";

    // in the destructor. Note that 'cleanupWorker' can not be properly called from this destructor

        stopWorker();

    // because any subclasses have already been destroyed and thus the 'DriverInterface'

        // The worker and the context should clean up by themselves via destructors.

    // implementation is not valid. Thus, here it can only be asserted whether the subclass has done

    // its job.

    if (!mWorkerStopIssued && !isClosed()) {

        LOG(FATAL) << __func__ << ": the stream implementation must call 'cleanupWorker' "

                   << "in order to clean up the worker thread.";

    }

    }

}

}
@@ -770,10 +774,7 @@ ndk::ScopedAStatus StreamCommonImpl::removeEffect( 
ndk::ScopedAStatus StreamCommonImpl::close() {

ndk::ScopedAStatus StreamCommonImpl::close() {

    LOG(DEBUG) << __func__;

    LOG(DEBUG) << __func__;

    if (!isClosed()) {

    if (!isClosed()) {

        stopWorker();

        stopAndJoinWorker();

        LOG(DEBUG) << __func__ << ": joining the worker thread...";

        mWorker->join();

        LOG(DEBUG) << __func__ << ": worker thread joined";

        onClose(mWorker->setClosed());

        onClose(mWorker->setClosed());

        return ndk::ScopedAStatus::ok();

        return ndk::ScopedAStatus::ok();

    } else {

    } else {
@@ -791,6 +792,20 @@ ndk::ScopedAStatus StreamCommonImpl::prepareToClose() { 
    return ndk::ScopedAStatus::fromExceptionCode(EX_ILLEGAL_STATE);

    return ndk::ScopedAStatus::fromExceptionCode(EX_ILLEGAL_STATE);

}

}





void StreamCommonImpl::cleanupWorker() {

    if (!isClosed()) {

        LOG(ERROR) << __func__ << ": stream was not closed prior to destruction, resource leak";

        stopAndJoinWorker();

    }

}



void StreamCommonImpl::stopAndJoinWorker() {

    stopWorker();

    LOG(DEBUG) << __func__ << ": joining the worker thread...";

    mWorker->join();

    LOG(DEBUG) << __func__ << ": worker thread joined";

}



void StreamCommonImpl::stopWorker() {

void StreamCommonImpl::stopWorker() {

    if (auto commandMQ = mContext.getCommandMQ(); commandMQ != nullptr) {

    if (auto commandMQ = mContext.getCommandMQ(); commandMQ != nullptr) {

        LOG(DEBUG) << __func__ << ": asking the worker to exit...";

        LOG(DEBUG) << __func__ << ": asking the worker to exit...";
@@ -805,6 +820,7 @@ void StreamCommonImpl::stopWorker() { 
        }

        }

        LOG(DEBUG) << __func__ << ": done";

        LOG(DEBUG) << __func__ << ": done";

    }

    }

    mWorkerStopIssued = true;

}

}





ndk::ScopedAStatus StreamCommonImpl::updateMetadataCommon(const Metadata& metadata) {

ndk::ScopedAStatus StreamCommonImpl::updateMetadataCommon(const Metadata& metadata) {

audio/aidl/default/alsa/StreamAlsa.cpp

+4 −0
Original line number Original line Diff line number Diff line
@@ -37,6 +37,10 @@ StreamAlsa::StreamAlsa(StreamContext* context, const Metadata& metadata, int rea 
      mConfig(alsa::getPcmConfig(getContext(), mIsInput)),

      mConfig(alsa::getPcmConfig(getContext(), mIsInput)),

      mReadWriteRetries(readWriteRetries) {}

      mReadWriteRetries(readWriteRetries) {}





StreamAlsa::~StreamAlsa() {

    cleanupWorker();

}



::android::status_t StreamAlsa::init() {

::android::status_t StreamAlsa::init() {

    return mConfig.has_value() ? ::android::OK : ::android::NO_INIT;

    return mConfig.has_value() ? ::android::OK : ::android::NO_INIT;

}

}

audio/aidl/default/bluetooth/StreamBluetooth.cpp

+4 −0
Original line number Original line Diff line number Diff line
@@ -66,6 +66,10 @@ StreamBluetooth::StreamBluetooth(StreamContext* context, const Metadata& metadat 
                                                 1000),

                                                 1000),

      mBtDeviceProxy(btDeviceProxy) {}

      mBtDeviceProxy(btDeviceProxy) {}





StreamBluetooth::~StreamBluetooth() {

    cleanupWorker();

}



::android::status_t StreamBluetooth::init() {

::android::status_t StreamBluetooth::init() {

    std::lock_guard guard(mLock);

    std::lock_guard guard(mLock);

    if (mBtDeviceProxy == nullptr) {

    if (mBtDeviceProxy == nullptr) {

audio/aidl/default/include/core-impl/Stream.h

+8 −0
Original line number Original line Diff line number Diff line
@@ -457,6 +457,11 @@ class StreamCommonImpl : virtual public StreamCommonInterface, virtual public Dr 
    }

    }





    virtual void onClose(StreamDescriptor::State statePriorToClosing) = 0;

    virtual void onClose(StreamDescriptor::State statePriorToClosing) = 0;

    // Any stream class implementing 'DriverInterface::shutdown' must call 'cleanupWorker' in

    // the destructor in order to stop and join the worker thread in the case when the client

    // has not called 'IStreamCommon::close' method.

    void cleanupWorker();

    void stopAndJoinWorker();

    void stopWorker();

    void stopWorker();





    const StreamContext& mContext;

    const StreamContext& mContext;
@@ -464,6 +469,9 @@ class StreamCommonImpl : virtual public StreamCommonInterface, virtual public Dr 
    std::unique_ptr<StreamWorkerInterface> mWorker;

    std::unique_ptr<StreamWorkerInterface> mWorker;

    ChildInterface<StreamCommonDelegator> mCommon;

    ChildInterface<StreamCommonDelegator> mCommon;

    ConnectedDevices mConnectedDevices;

    ConnectedDevices mConnectedDevices;



  private:

    std::atomic<bool> mWorkerStopIssued = false;

};

};





// Note: 'StreamIn/Out' can not be used on their own. Instead, they must be used for defining

// Note: 'StreamIn/Out' can not be used on their own. Instead, they must be used for defining

audio/aidl/default/include/core-impl/StreamAlsa.h

+2 −0
Original line number Original line Diff line number Diff line
@@ -32,6 +32,8 @@ namespace aidl::android::hardware::audio::core { 
class StreamAlsa : public StreamCommonImpl {

class StreamAlsa : public StreamCommonImpl {

  public:

  public:

    StreamAlsa(StreamContext* context, const Metadata& metadata, int readWriteRetries);

    StreamAlsa(StreamContext* context, const Metadata& metadata, int readWriteRetries);

    ~StreamAlsa();



    // Methods of 'DriverInterface'.

    // Methods of 'DriverInterface'.

    ::android::status_t init() override;

    ::android::status_t init() override;

    ::android::status_t drain(StreamDescriptor::DrainMode) override;

    ::android::status_t drain(StreamDescriptor::DrainMode) override;