Merge "audio: Fix stream cleanup sequence" into main

audio/aidl/default/Stream.cpp

@@ -663,10 +663,14 @@ bool StreamOutWorkerLogic::write(size_t clientSize, StreamDescriptor::Reply* rep
 }
 
 StreamCommonImpl::~StreamCommonImpl() {
-    if (!isClosed()) {
-        LOG(ERROR) << __func__ << ": stream was not closed prior to destruction, resource leak";
-        stopWorker();
-        // The worker and the context should clean up by themselves via destructors.
+    // It is responsibility of the class that implements 'DriverInterface' to call 'cleanupWorker'
+    // in the destructor. Note that 'cleanupWorker' can not be properly called from this destructor
+    // because any subclasses have already been destroyed and thus the 'DriverInterface'
+    // 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() {
     LOG(DEBUG) << __func__;
     if (!isClosed()) {
-        stopWorker();
-        LOG(DEBUG) << __func__ << ": joining the worker thread...";
-        mWorker->join();
-        LOG(DEBUG) << __func__ << ": worker thread joined";
+        stopAndJoinWorker();
         onClose(mWorker->setClosed());
         return ndk::ScopedAStatus::ok();
     } else {
@@ -791,6 +792,20 @@ ndk::ScopedAStatus StreamCommonImpl::prepareToClose() {
     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() {
     if (auto commandMQ = mContext.getCommandMQ(); commandMQ != nullptr) {
         LOG(DEBUG) << __func__ << ": asking the worker to exit...";
@@ -805,6 +820,7 @@ void StreamCommonImpl::stopWorker() {
         }
         LOG(DEBUG) << __func__ << ": done";
     }
+    mWorkerStopIssued = true;
 }
 
 ndk::ScopedAStatus StreamCommonImpl::updateMetadataCommon(const Metadata& metadata) {

audio/aidl/default/alsa/StreamAlsa.cpp

@@ -37,6 +37,10 @@ StreamAlsa::StreamAlsa(StreamContext* context, const Metadata& metadata, int rea
       mConfig(alsa::getPcmConfig(getContext(), mIsInput)),
       mReadWriteRetries(readWriteRetries) {}
 
+StreamAlsa::~StreamAlsa() {
+    cleanupWorker();
+}
+
 ::android::status_t StreamAlsa::init() {
     return mConfig.has_value() ? ::android::OK : ::android::NO_INIT;
 }

audio/aidl/default/bluetooth/StreamBluetooth.cpp

@@ -66,6 +66,10 @@ StreamBluetooth::StreamBluetooth(StreamContext* context, const Metadata& metadat
                                                  1000),
       mBtDeviceProxy(btDeviceProxy) {}
 
+StreamBluetooth::~StreamBluetooth() {
+    cleanupWorker();
+}
+
 ::android::status_t StreamBluetooth::init() {
     std::lock_guard guard(mLock);
     if (mBtDeviceProxy == nullptr) {

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

@@ -457,6 +457,11 @@ class StreamCommonImpl : virtual public StreamCommonInterface, virtual public Dr
     }
 
     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();
 
     const StreamContext& mContext;
@@ -464,6 +469,9 @@ class StreamCommonImpl : virtual public StreamCommonInterface, virtual public Dr
     std::unique_ptr<StreamWorkerInterface> mWorker;
     ChildInterface<StreamCommonDelegator> mCommon;
     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

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

@@ -32,6 +32,8 @@ namespace aidl::android::hardware::audio::core {
 class StreamAlsa : public StreamCommonImpl {
   public:
     StreamAlsa(StreamContext* context, const Metadata& metadata, int readWriteRetries);
+    ~StreamAlsa();
+
     // Methods of 'DriverInterface'.
     ::android::status_t init() override;
     ::android::status_t drain(StreamDescriptor::DrainMode) override;

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

@@ -41,6 +41,8 @@ class StreamBluetooth : public StreamCommonImpl {
             const std::shared_ptr<::android::bluetooth::audio::aidl::BluetoothAudioPortAidl>&
                     btDeviceProxy,
             const ::aidl::android::hardware::bluetooth::audio::PcmConfiguration& pcmConfig);
+    ~StreamBluetooth();
+
     // Methods of 'DriverInterface'.
     ::android::status_t init() override;
     ::android::status_t drain(StreamDescriptor::DrainMode) override;

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

@@ -29,7 +29,9 @@ class StreamRemoteSubmix : public StreamCommonImpl {
     StreamRemoteSubmix(
             StreamContext* context, const Metadata& metadata,
             const ::aidl::android::media::audio::common::AudioDeviceAddress& deviceAddress);
+    ~StreamRemoteSubmix();
 
+    // Methods of 'DriverInterface'.
     ::android::status_t init() override;
     ::android::status_t drain(StreamDescriptor::DrainMode) override;
     ::android::status_t flush() override;

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

@@ -23,6 +23,8 @@ namespace aidl::android::hardware::audio::core {
 class StreamStub : public StreamCommonImpl {
   public:
     StreamStub(StreamContext* context, const Metadata& metadata);
+    ~StreamStub();
+
     // Methods of 'DriverInterface'.
     ::android::status_t init() override;
     ::android::status_t drain(StreamDescriptor::DrainMode) override;

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

@@ -29,6 +29,7 @@ namespace aidl::android::hardware::audio::core {
 class StreamUsb : public StreamAlsa {
   public:
     StreamUsb(StreamContext* context, const Metadata& metadata);
+
     // Methods of 'DriverInterface'.
     ::android::status_t transfer(void* buffer, size_t frameCount, size_t* actualFrameCount,
                                  int32_t* latencyMs) override;

audio/aidl/default/r_submix/StreamRemoteSubmix.cpp

@@ -43,6 +43,10 @@ StreamRemoteSubmix::StreamRemoteSubmix(StreamContext* context, const Metadata& m
     mStreamConfig.sampleRate = context->getSampleRate();
 }
 
+StreamRemoteSubmix::~StreamRemoteSubmix() {
+    cleanupWorker();
+}
+
 ::android::status_t StreamRemoteSubmix::init() {
     mCurrentRoute = SubmixRoute::findOrCreateRoute(mDeviceAddress, mStreamConfig);
     if (mCurrentRoute == nullptr) {

audio/aidl/default/stub/StreamStub.cpp

@@ -39,6 +39,10 @@ StreamStub::StreamStub(StreamContext* context, const Metadata& metadata)
       mIsAsynchronous(!!getContext().getAsyncCallback()),
       mIsInput(isInput(metadata)) {}
 
+StreamStub::~StreamStub() {
+    cleanupWorker();
+}
+
 ::android::status_t StreamStub::init() {
     mIsInitialized = true;
     return ::android::OK;