From f649c4fcb1e3ff1bdb74b81ed54a9a68662ae765 Mon Sep 17 00:00:00 2001 From: Eism Date: Fri, 30 Oct 2026 15:09:20 +0200 Subject: [PATCH 1/9] =?UTF-8?q?=D1=81ache=20Player=20instance=20within=20P?= =?UTF-8?q?layback?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Prevents creation of multiple Player objects for one context --- framework/audio/main/internal/playback.cpp | 10 +++++++--- framework/audio/main/internal/playback.h | 2 ++ 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/framework/audio/main/internal/playback.cpp b/framework/audio/main/internal/playback.cpp index 92847705ec..ea07f0f352 100644 --- a/framework/audio/main/internal/playback.cpp +++ b/framework/audio/main/internal/playback.cpp @@ -184,9 +184,13 @@ async::Channel Playback::initedChanged() const IPlayerPtr Playback::player() const { - std::shared_ptr p = std::make_shared(iocContext()); - p->init(); - return p; + if (!m_player) { + std::shared_ptr p = std::make_shared(iocContext()); + p->init(); + m_player = p; + } + + return m_player; } template diff --git a/framework/audio/main/internal/playback.h b/framework/audio/main/internal/playback.h index 5bfdbbbd4d..8c2fa50bb1 100644 --- a/framework/audio/main/internal/playback.h +++ b/framework/audio/main/internal/playback.h @@ -138,5 +138,7 @@ class Playback : public IPlayback, public async::Asyncable, public Contextable mutable bool m_saveSoundTrackProgressStreamInited = false; async::Channel m_saveSoundTrackProgressStream; mutable rpc::StreamId m_saveSoundTrackProgressStreamId = 0; + + mutable IPlayerPtr m_player; }; } From 9dbde7c03d956b0bcdb012c5b2085731017acec7 Mon Sep 17 00:00:00 2001 From: Eism Date: Fri, 30 Oct 2026 16:24:40 +0200 Subject: [PATCH 2/9] added unit-tests (fails) --- framework/audio/main/internal/playback.h | 2 +- framework/audio/tests/CMakeLists.txt | 2 + .../tests/mocks/startaudiocontrollermock.h | 38 +++ framework/audio/tests/playback_tests.cpp | 216 ++++++++++++++++++ 4 files changed, 257 insertions(+), 1 deletion(-) create mode 100644 framework/audio/tests/mocks/startaudiocontrollermock.h create mode 100644 framework/audio/tests/playback_tests.cpp diff --git a/framework/audio/main/internal/playback.h b/framework/audio/main/internal/playback.h index 8c2fa50bb1..b1f9385d20 100644 --- a/framework/audio/main/internal/playback.h +++ b/framework/audio/main/internal/playback.h @@ -26,7 +26,7 @@ #include "global/types/retval.h" #include "modularity/ioc.h" -#include "common/rpc/irpcchannel.h" +#include "audio/common/rpc/irpcchannel.h" #include "../istartaudiocontroller.h" #include "../iplayer.h" diff --git a/framework/audio/tests/CMakeLists.txt b/framework/audio/tests/CMakeLists.txt index dc35d14190..9bef46614e 100644 --- a/framework/audio/tests/CMakeLists.txt +++ b/framework/audio/tests/CMakeLists.txt @@ -25,10 +25,12 @@ set(MODULE_TEST_SRC ${CMAKE_CURRENT_LIST_DIR}/mocks/audiodriverfactorymock.h ${CMAKE_CURRENT_LIST_DIR}/mocks/audiodrivermock.h ${CMAKE_CURRENT_LIST_DIR}/mocks/rpcchannelmock.h + ${CMAKE_CURRENT_LIST_DIR}/mocks/startaudiocontrollermock.h ${CMAKE_CURRENT_LIST_DIR}/alignbuffer_tests.cpp ${CMAKE_CURRENT_LIST_DIR}/audioresourcetypes_tests.cpp ${CMAKE_CURRENT_LIST_DIR}/audiodrivercontroller_tests.cpp + ${CMAKE_CURRENT_LIST_DIR}/playback_tests.cpp ) set(MODULE_TEST_LINK diff --git a/framework/audio/tests/mocks/startaudiocontrollermock.h b/framework/audio/tests/mocks/startaudiocontrollermock.h new file mode 100644 index 0000000000..bca37bb39f --- /dev/null +++ b/framework/audio/tests/mocks/startaudiocontrollermock.h @@ -0,0 +1,38 @@ +/* + * SPDX-License-Identifier: GPL-3.0-only + * MuseScore-CLA-applies + * + * MuseScore + * Music Composition & Notation + * + * Copyright (C) 2026 MuseScore Limited and others + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License version 3 as + * published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ +#pragma once + +#include + +#include "audio/main/istartaudiocontroller.h" + +namespace muse::audio { +class StartAudioControllerMock : public IStartAudioController +{ +public: + MOCK_METHOD(void, startAudioProcessing, (const IApplication::RunMode&), (override)); + MOCK_METHOD(void, stopAudioProcessing, (), (override)); + + MOCK_METHOD(bool, isAudioStarted, (), (const, override)); + MOCK_METHOD(async::Channel, isAudioStartedChanged, (), (const, override)); +}; +} diff --git a/framework/audio/tests/playback_tests.cpp b/framework/audio/tests/playback_tests.cpp new file mode 100644 index 0000000000..966f00e3bf --- /dev/null +++ b/framework/audio/tests/playback_tests.cpp @@ -0,0 +1,216 @@ +/* + * SPDX-License-Identifier: GPL-3.0-only + * MuseScore-CLA-applies + * + * MuseScore + * Music Composition & Notation + * + * Copyright (C) 2026 MuseScore Limited and others + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License version 3 as + * published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ +#include +#include + +#include + +#include "async/processevents.h" + +#include "audio/common/rpc/rpcpacker.h" +#include "audio/main/internal/playback.h" + +#include "mocks/rpcchannelmock.h" +#include "mocks/startaudiocontrollermock.h" + +using ::testing::_; +using ::testing::Invoke; +using ::testing::NiceMock; +using ::testing::Return; + +using namespace muse; +using namespace muse::audio; +using namespace muse::audio::rpc; + +namespace { +const StreamId STATUS_STREAM_ID = 11; +const StreamId POSITION_STREAM_ID = 12; + +MATCHER_P(StreamIdIs, id, "") +{ + return arg && arg->streamId() == id; +} +} + +namespace muse::audio { +class Audio_PlaybackTests : public ::testing::Test +{ +protected: + struct Request { + Msg msg; + ResponseHandler onResponse; + }; + + void SetUp() override + { + m_rpcChannel = std::make_shared >(); + m_startAudioController = std::make_shared >(); + + modularity::globalIoc()->registerExport("utests", m_rpcChannel); + modularity::globalIoc()->registerExport("utests", m_startAudioController); + + ON_CALL(*m_startAudioController, isAudioStarted()).WillByDefault(Return(true)); + + ON_CALL(*m_rpcChannel, send(_, _)) + .WillByDefault(Invoke([this](const Msg& msg, const ResponseHandler& onResponse) { + m_requests.push_back({ msg, onResponse }); + })); + + m_playback = std::make_shared(m_ctx); + } + + void TearDown() override + { + m_playback.reset(); + + modularity::globalIoc()->unregister("utests"); + modularity::globalIoc()->unregister("utests"); + } + + size_t requestCount(MsgCode code) const + { + size_t count = 0; + for (const Request& r : m_requests) { + if (r.msg.code == code) { + ++count; + } + } + return count; + } + + //! NOTE Responds to the first pending request with the given code + void respond(MsgCode code, const ByteArray& data) + { + for (auto it = m_requests.begin(); it != m_requests.end(); ++it) { + if (it->msg.code == code) { + Request r = *it; + m_requests.erase(it); + ASSERT_TRUE(r.onResponse); + r.onResponse(make_response(r.msg, data)); + return; + } + } + + FAIL() << "no pending request: " << rpc::to_string(code); + } + + void initContext() + { + m_playback->init(); + async::processMessages(); + + respond(MsgCode::ContextInit, RpcPacker::pack(make_ok())); + ASSERT_TRUE(m_playback->isInited()); + } + + void respondPlayerStreams() + { + respond(MsgCode::GetPlaybackStatus, RpcPacker::pack(make_ok(), PlaybackStatus::Running, STATUS_STREAM_ID)); + respond(MsgCode::GetPlaybackPosition, RpcPacker::pack(make_ok(), secs_t(2.0), POSITION_STREAM_ID)); + } + + modularity::ContextPtr m_ctx = std::make_shared(1); + + std::shared_ptr m_rpcChannel; + std::shared_ptr m_startAudioController; + std::shared_ptr m_playback; + + std::vector m_requests; +}; + +TEST_F(Audio_PlaybackTests, Player_IsCreatedOncePerContext) +{ + // [GIVEN] The context is inited + initContext(); + + // [WHEN] Get the player twice + IPlayerPtr player1 = m_playback->player(); + IPlayerPtr player2 = m_playback->player(); + + // [THEN] The same player is returned and it requests the state only once + ASSERT_TRUE(player1); + EXPECT_EQ(player1, player2); + EXPECT_EQ(requestCount(MsgCode::GetPlaybackStatus), 1); + EXPECT_EQ(requestCount(MsgCode::GetPlaybackPosition), 1); +} + +TEST_F(Audio_PlaybackTests, Player_SubscribesToStatusAndPosition) +{ + // [GIVEN] The context is inited + initContext(); + + // [THEN] The streams of the status and the position are added + EXPECT_CALL(*m_rpcChannel, addStream(StreamIdIs(STATUS_STREAM_ID))); + EXPECT_CALL(*m_rpcChannel, addStream(StreamIdIs(POSITION_STREAM_ID))); + + // [WHEN] Get the player and the engine responds + IPlayerPtr player = m_playback->player(); + respondPlayerStreams(); + + // [THEN] The player has the initial state + EXPECT_EQ(player->playbackStatus(), PlaybackStatus::Running); + EXPECT_EQ(player->playbackPosition(), secs_t(2.0)); +} + +TEST_F(Audio_PlaybackTests, Deinit_RemovesPlayerStreams) +{ + // [GIVEN] The player is subscribed to the streams + initContext(); + IPlayerPtr player = m_playback->player(); + respondPlayerStreams(); + + // [THEN] The streams are removed + EXPECT_CALL(*m_rpcChannel, removeStream(STATUS_STREAM_ID)); + EXPECT_CALL(*m_rpcChannel, removeStream(POSITION_STREAM_ID)); + + // [WHEN] Deinit the playback + m_playback->deinit(); + EXPECT_FALSE(m_playback->isInited()); + + // [WHEN] Init it again + initContext(); + + // [THEN] A new player is created + IPlayerPtr newPlayer = m_playback->player(); + ASSERT_TRUE(newPlayer); + EXPECT_NE(newPlayer, player); +} + +TEST_F(Audio_PlaybackTests, Deinit_LateResponsesAreIgnored) +{ + // [GIVEN] The player has not received the responses yet + initContext(); + IPlayerPtr player = m_playback->player(); + + // [WHEN] Deinit the playback + m_playback->deinit(); + + // [THEN] The streams are not added by the late responses + EXPECT_CALL(*m_rpcChannel, addStream(_)).Times(0); + + // [WHEN] The engine responds after deinit + respondPlayerStreams(); + + // [THEN] The player state is not changed + EXPECT_EQ(player->playbackStatus(), PlaybackStatus::Stopped); +} +} From 773265a8ff902712be7c567754e81e160b2b6743 Mon Sep 17 00:00:00 2001 From: Eism Date: Fri, 30 Oct 2026 15:41:34 +0200 Subject: [PATCH 3/9] implement Player deinit and enhance async safety --- framework/audio/main/internal/playback.cpp | 15 ++++++++-- framework/audio/main/internal/playback.h | 4 ++- framework/audio/main/internal/player.cpp | 34 ++++++++++++++++++++-- framework/audio/main/internal/player.h | 7 +++++ 4 files changed, 54 insertions(+), 6 deletions(-) diff --git a/framework/audio/main/internal/playback.cpp b/framework/audio/main/internal/playback.cpp index ea07f0f352..b767ef0aa3 100644 --- a/framework/audio/main/internal/playback.cpp +++ b/framework/audio/main/internal/playback.cpp @@ -168,6 +168,11 @@ void Playback::deinit() m_saveSoundTrackProgressStreamInited = false; m_saveSoundTrackProgressStreamId = 0; + if (m_player) { + m_player->deinit(); + m_player.reset(); + } + channel()->send(rpc::make_request(rpc::GLOBAL_CTX_ID, MsgCode::ContextDeinit, RpcPacker::pack(ctxId()))); m_inited.set(false); } @@ -184,10 +189,14 @@ async::Channel Playback::initedChanged() const IPlayerPtr Playback::player() const { + //! NOTE The player requests the engine context, so it must exist + IF_ASSERT_FAILED(isInited()) { + return nullptr; + } + if (!m_player) { - std::shared_ptr p = std::make_shared(iocContext()); - p->init(); - m_player = p; + m_player = std::make_shared(iocContext()); + m_player->init(); } return m_player; diff --git a/framework/audio/main/internal/playback.h b/framework/audio/main/internal/playback.h index b1f9385d20..e8d7078ff3 100644 --- a/framework/audio/main/internal/playback.h +++ b/framework/audio/main/internal/playback.h @@ -32,6 +32,8 @@ #include "../iplayer.h" namespace muse::audio { +class Player; + //! NOTE It may not use contextual services, // but the playback itself is contextual, // because each context (window) has its own playback. @@ -139,6 +141,6 @@ class Playback : public IPlayback, public async::Asyncable, public Contextable async::Channel m_saveSoundTrackProgressStream; mutable rpc::StreamId m_saveSoundTrackProgressStreamId = 0; - mutable IPlayerPtr m_player; + mutable std::shared_ptr m_player; }; } diff --git a/framework/audio/main/internal/player.cpp b/framework/audio/main/internal/player.cpp index afe9ebbe6a..b9dd57bb76 100644 --- a/framework/audio/main/internal/player.cpp +++ b/framework/audio/main/internal/player.cpp @@ -48,8 +48,12 @@ void Player::init() }); Msg msg = rpc::make_request(ctxId(), MsgCode::GetPlaybackStatus); - channel()->send(msg, [this](const Msg& res) { + channel()->send(msg, [this, alive = std::weak_ptr(m_alive)](const Msg& res) { ONLY_AUDIO_MAIN_THREAD; + if (alive.expired()) { + return; + } + Ret ret; PlaybackStatus status = PlaybackStatus::Stopped; StreamId streamId = 0; @@ -58,6 +62,7 @@ void Player::init() } if (ret) { + m_playbackStatusStreamId = streamId; channel()->addReceiveStream(StreamName::PlaybackStatusStream, streamId, m_playbackStatusChanged); //! NOTE Send initial state m_playbackStatusChanged.send(status); @@ -73,8 +78,12 @@ void Player::init() }); Msg msg = rpc::make_request(ctxId(), MsgCode::GetPlaybackPosition); - channel()->send(msg, [this](const Msg& res) { + channel()->send(msg, [this, alive = std::weak_ptr(m_alive)](const Msg& res) { ONLY_AUDIO_MAIN_THREAD; + if (alive.expired()) { + return; + } + Ret ret; secs_t pos = 0.0; StreamId streamId = 0; @@ -83,6 +92,7 @@ void Player::init() } if (ret) { + m_playbackPositionStreamId = streamId; channel()->addReceiveStream(StreamName::PlaybackPositionStream, streamId, m_playbackPositionChanged); //! NOTE Send initial state m_playbackPositionChanged.send(pos); @@ -93,6 +103,26 @@ void Player::init() } } +void Player::deinit() +{ + ONLY_AUDIO_MAIN_THREAD; + + m_alive.reset(); + + if (m_playbackStatusStreamId != 0) { + channel()->removeStream(m_playbackStatusStreamId); + m_playbackStatusStreamId = 0; + } + + if (m_playbackPositionStreamId != 0) { + channel()->removeStream(m_playbackPositionStreamId); + m_playbackPositionStreamId = 0; + } + + m_playbackStatusChanged.disconnect(this); + m_playbackPositionChanged.disconnect(this); +} + async::Promise Player::prepareToPlay() { ONLY_AUDIO_MAIN_THREAD; diff --git a/framework/audio/main/internal/player.h b/framework/audio/main/internal/player.h index 0ae06a14d3..ba3726a778 100644 --- a/framework/audio/main/internal/player.h +++ b/framework/audio/main/internal/player.h @@ -40,6 +40,7 @@ class Player : public IPlayer, public Contextable, public async::Asyncable : Contextable(ctx) {} void init(); + void deinit(); async::Promise prepareToPlay() override; @@ -68,6 +69,12 @@ class Player : public IPlayer, public Contextable, public async::Asyncable secs_t m_playbackPosition = 0.0; async::Channel m_playbackPositionChanged; + + rpc::StreamId m_playbackStatusStreamId = 0; + rpc::StreamId m_playbackPositionStreamId = 0; + + //! NOTE Responses may come after deinit, they must not be handled + std::shared_ptr m_alive = std::make_shared(true); }; } From 750f14b7c6cc420491d51c9647c7cd3c52c68418 Mon Sep 17 00:00:00 2001 From: Eism Date: Mon, 2 Nov 2026 12:16:27 +0200 Subject: [PATCH 4/9] add unit tests for audio RPC stream cleanup Verify that RPC streams are correctly removed by EngineRpcController and Playback when audio contexts are deinitialized or tracks are removed. This ensures proper resource management and prevents leaks. --- framework/audio/tests/CMakeLists.txt | 3 + .../audio/tests/enginerpccontroller_tests.cpp | 207 ++++++++++++++++++ .../audio/tests/mocks/audiocontextmock.h | 103 +++++++++ framework/audio/tests/mocks/audioenginemock.h | 45 ++++ framework/audio/tests/playback_tests.cpp | 50 +++++ 5 files changed, 408 insertions(+) create mode 100644 framework/audio/tests/enginerpccontroller_tests.cpp create mode 100644 framework/audio/tests/mocks/audiocontextmock.h create mode 100644 framework/audio/tests/mocks/audioenginemock.h diff --git a/framework/audio/tests/CMakeLists.txt b/framework/audio/tests/CMakeLists.txt index 9bef46614e..55658f7b59 100644 --- a/framework/audio/tests/CMakeLists.txt +++ b/framework/audio/tests/CMakeLists.txt @@ -22,14 +22,17 @@ set(MODULE_TEST muse_audio_tests) set(MODULE_TEST_SRC ${CMAKE_CURRENT_LIST_DIR}/mocks/audioconfigurationmock.h + ${CMAKE_CURRENT_LIST_DIR}/mocks/audiocontextmock.h ${CMAKE_CURRENT_LIST_DIR}/mocks/audiodriverfactorymock.h ${CMAKE_CURRENT_LIST_DIR}/mocks/audiodrivermock.h + ${CMAKE_CURRENT_LIST_DIR}/mocks/audioenginemock.h ${CMAKE_CURRENT_LIST_DIR}/mocks/rpcchannelmock.h ${CMAKE_CURRENT_LIST_DIR}/mocks/startaudiocontrollermock.h ${CMAKE_CURRENT_LIST_DIR}/alignbuffer_tests.cpp ${CMAKE_CURRENT_LIST_DIR}/audioresourcetypes_tests.cpp ${CMAKE_CURRENT_LIST_DIR}/audiodrivercontroller_tests.cpp + ${CMAKE_CURRENT_LIST_DIR}/enginerpccontroller_tests.cpp ${CMAKE_CURRENT_LIST_DIR}/playback_tests.cpp ) diff --git a/framework/audio/tests/enginerpccontroller_tests.cpp b/framework/audio/tests/enginerpccontroller_tests.cpp new file mode 100644 index 0000000000..455d16052a --- /dev/null +++ b/framework/audio/tests/enginerpccontroller_tests.cpp @@ -0,0 +1,207 @@ +/* + * SPDX-License-Identifier: GPL-3.0-only + * MuseScore-CLA-applies + * + * MuseScore + * Music Composition & Notation + * + * Copyright (C) 2026 MuseScore Limited and others + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License version 3 as + * published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ +#include +#include + +#include +#include + +#include "audio/common/rpc/rpcpacker.h" +#include "audio/engine/internal/enginerpccontroller.h" + +#include "mocks/audiocontextmock.h" +#include "mocks/audioenginemock.h" +#include "mocks/rpcchannelmock.h" + +using ::testing::_; +using ::testing::Invoke; +using ::testing::NiceMock; +using ::testing::Return; + +using namespace muse; +using namespace muse::audio; +using namespace muse::audio::engine; +using namespace muse::audio::rpc; + +namespace { +const CtxId CTX_ID = 1; +const TrackId TRACK_ID = 2; +const TrackId OTHER_TRACK_ID = 3; + +class ExecOperationNow : public IExecOperation +{ +public: + void execOperation(OperationType, const Operation& func) override + { + func(); + } +}; +} + +namespace muse::audio::engine { +class Audio_EngineRpcControllerTests : public ::testing::Test +{ +protected: + void SetUp() override + { + m_rpcChannel = std::make_shared >(); + m_audioEngine = std::make_shared >(); + m_audioContext = std::make_shared >(); + + modularity::globalIoc()->registerExport("utests", m_rpcChannel); + modularity::globalIoc()->registerExport("utests", m_audioEngine); + + ON_CALL(*m_rpcChannel, onRequest(_, _, _)) + .WillByDefault(Invoke([this](CtxId ctxId, MsgCode code, RequestHandler h) { + m_handlers[{ ctxId, code }] = h; + })); + + ON_CALL(*m_audioEngine, addAudioContext(_)) + .WillByDefault(Return(RetVal >::make_ok(m_audioContext))); + ON_CALL(*m_audioEngine, context(CTX_ID)).WillByDefault(Return(m_audioContext)); + + ON_CALL(*m_audioContext, trackRemoved()).WillByDefault(Return(m_trackRemoved)); + ON_CALL(*m_audioContext, signalChanges(_)) + .WillByDefault(Return(RetVal::make_ok(AudioSignalChanges()))); + ON_CALL(*m_audioContext, automatedControlParamsChanges(_)) + .WillByDefault(Return(RetVal::make_ok(AutomatedControlParamsChanges()))); + ON_CALL(*m_audioContext, inputProcessingProgress(_)) + .WillByDefault(Return(RetVal::make_ok(InputProcessingProgress()))); + + m_controller = std::make_shared(&m_execOperation); + m_controller->init(); + } + + void TearDown() override + { + m_controller->deinit(); + m_controller.reset(); + + modularity::globalIoc()->unregister("utests"); + modularity::globalIoc()->unregister("utests"); + } + + Msg request(CtxId ctxId, MsgCode code, const ByteArray& data = ByteArray()) + { + auto it = m_handlers.find({ ctxId, code }); + if (it == m_handlers.end() || !it->second) { + ADD_FAILURE() << "no handler: " << rpc::to_string(code); + return Msg(); + } + + return it->second(make_request(ctxId, code, data)); + } + + //! NOTE Returns the id of the stream added by the engine for the response + StreamId requestStream(MsgCode code, const ByteArray& data = ByteArray()) + { + StreamId streamId = 0; + EXPECT_CALL(*m_rpcChannel, addStream(_)) + .WillOnce(Invoke([&streamId](std::shared_ptr s) { + streamId = s->streamId(); + })); + + request(CTX_ID, code, data); + ::testing::Mock::VerifyAndClearExpectations(m_rpcChannel.get()); + + return streamId; + } + + ExecOperationNow m_execOperation; + + std::shared_ptr m_rpcChannel; + std::shared_ptr m_audioEngine; + std::shared_ptr m_audioContext; + std::shared_ptr m_controller; + + std::map, RequestHandler> m_handlers; + async::Channel m_trackRemoved; +}; + +TEST_F(Audio_EngineRpcControllerTests, ContextDeinit_RemovesPlayerStreams) +{ + // [GIVEN] The context is inited + request(GLOBAL_CTX_ID, MsgCode::ContextInit, RpcPacker::pack(CTX_ID)); + + // [GIVEN] The player has requested the status and the position + const StreamId statusStreamId = requestStream(MsgCode::GetPlaybackStatus); + const StreamId positionStreamId = requestStream(MsgCode::GetPlaybackPosition); + ASSERT_NE(statusStreamId, 0); + ASSERT_NE(positionStreamId, 0); + + // [THEN] Both streams are removed + EXPECT_CALL(*m_rpcChannel, removeStream(statusStreamId)); + EXPECT_CALL(*m_rpcChannel, removeStream(positionStreamId)); + + // [WHEN] Deinit the context + request(GLOBAL_CTX_ID, MsgCode::ContextDeinit, RpcPacker::pack(CTX_ID)); +} + +TEST_F(Audio_EngineRpcControllerTests, ContextDeinit_StreamsAreRemovedOnlyOnce) +{ + // [GIVEN] The context is inited and the player has requested the status + request(GLOBAL_CTX_ID, MsgCode::ContextInit, RpcPacker::pack(CTX_ID)); + const StreamId statusStreamId = requestStream(MsgCode::GetPlaybackStatus); + + // [GIVEN] The context is deinited and inited again + EXPECT_CALL(*m_rpcChannel, removeStream(statusStreamId)).Times(1); + request(GLOBAL_CTX_ID, MsgCode::ContextDeinit, RpcPacker::pack(CTX_ID)); + ::testing::Mock::VerifyAndClearExpectations(m_rpcChannel.get()); + + request(GLOBAL_CTX_ID, MsgCode::ContextInit, RpcPacker::pack(CTX_ID)); + + // [THEN] The streams of the previous init are not removed again + EXPECT_CALL(*m_rpcChannel, removeStream(_)).Times(0); + + // [WHEN] Deinit the context again + request(GLOBAL_CTX_ID, MsgCode::ContextDeinit, RpcPacker::pack(CTX_ID)); +} + +TEST_F(Audio_EngineRpcControllerTests, TrackRemoved_RemovesTrackStreams) +{ + // [GIVEN] The context is inited + request(GLOBAL_CTX_ID, MsgCode::ContextInit, RpcPacker::pack(CTX_ID)); + + // [GIVEN] The streams of two tracks are requested + const StreamId signalStreamId = requestStream(MsgCode::GetSignalChanges, RpcPacker::pack(TRACK_ID)); + const StreamId automationStreamId = requestStream(MsgCode::GetAutomatedControlParamsChanges, RpcPacker::pack(TRACK_ID)); + const StreamId progressStreamId = requestStream(MsgCode::GetInputProcessingProgress, RpcPacker::pack(TRACK_ID)); + const StreamId otherTrackStreamId = requestStream(MsgCode::GetSignalChanges, RpcPacker::pack(OTHER_TRACK_ID)); + + // [THEN] Only the streams of the removed track are removed + EXPECT_CALL(*m_rpcChannel, removeStream(signalStreamId)); + EXPECT_CALL(*m_rpcChannel, removeStream(automationStreamId)); + EXPECT_CALL(*m_rpcChannel, removeStream(progressStreamId)); + EXPECT_CALL(*m_rpcChannel, removeStream(otherTrackStreamId)).Times(0); + + // [WHEN] The track is removed + m_trackRemoved.send(TRACK_ID); + ::testing::Mock::VerifyAndClearExpectations(m_rpcChannel.get()); + + // [THEN] The streams of the other track are removed on the context deinit + EXPECT_CALL(*m_rpcChannel, removeStream(otherTrackStreamId)); + EXPECT_CALL(*m_rpcChannel, removeStream(signalStreamId)).Times(0); + + // [WHEN] Deinit the context + request(GLOBAL_CTX_ID, MsgCode::ContextDeinit, RpcPacker::pack(CTX_ID)); +} +} diff --git a/framework/audio/tests/mocks/audiocontextmock.h b/framework/audio/tests/mocks/audiocontextmock.h new file mode 100644 index 0000000000..4574a8b257 --- /dev/null +++ b/framework/audio/tests/mocks/audiocontextmock.h @@ -0,0 +1,103 @@ +/* + * SPDX-License-Identifier: GPL-3.0-only + * MuseScore-CLA-applies + * + * MuseScore + * Music Composition & Notation + * + * Copyright (C) 2026 MuseScore Limited and others + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License version 3 as + * published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ +#pragma once + +#include + +#include "audio/engine/internal/iaudiocontext.h" + +namespace muse::audio::engine { +class AudioContextMock : public IAudioContext +{ +public: + MOCK_METHOD(AudioCtxId, id, (), (const, override)); + + MOCK_METHOD(Ret, init, (), (override)); + MOCK_METHOD(void, deinit, (), (override)); + + MOCK_METHOD(AudioResourceMetaList, availableInputResources, (), (const, override)); + MOCK_METHOD(SoundPresetList, availableSoundPresets, (const AudioResourceMeta&), (const, override)); + MOCK_METHOD(AudioResourceMetaList, availableOutputResources, (), (const, override)); + + using AddTrackRet = RetVal2; + MOCK_METHOD(AddTrackRet, addTrack, (const TrackName&, io::IODevice*, const TrackParams&), (override)); + MOCK_METHOD(AddTrackRet, addTrack, (const TrackName&, const mpe::PlaybackData&, const TrackParams&), (override)); + MOCK_METHOD(AddTrackRet, addAuxTrack, (const TrackName&, const TrackParams&), (override)); + + MOCK_METHOD(void, removeTrack, (const TrackId), (override)); + MOCK_METHOD(void, removeAllTracks, (), (override)); + + MOCK_METHOD(async::Channel, trackAdded, (), (const, override)); + MOCK_METHOD(async::Channel, trackRemoved, (), (const, override)); + + MOCK_METHOD(RetVal, trackIdList, (), (const, override)); + MOCK_METHOD(RetVal, trackName, (const TrackId), (const, override)); + + MOCK_METHOD(RetVal, params, (const TrackId), (const, override)); + MOCK_METHOD(void, setSourceParams, (const TrackId, const AudioSourceParams&), (override)); + MOCK_METHOD(void, setControlParams, (const TrackId, const ControlParams&), (override)); + MOCK_METHOD(void, setFxChainParams, (const TrackId, const AudioFxChain&), (override)); + MOCK_METHOD(void, setAuxSendsParams, (const TrackId, const AuxSendsParams&), (override)); + + using SourceParamsChannel = async::Channel; + using ControlParamsChannel = async::Channel; + using FxChainChannel = async::Channel; + using AuxSendsParamsChannel = async::Channel; + MOCK_METHOD(SourceParamsChannel, sourceParamsChanged, (), (const, override)); + MOCK_METHOD(ControlParamsChannel, controlParamsChanged, (), (const, override)); + MOCK_METHOD(FxChainChannel, fxChainParamsChanged, (), (const, override)); + MOCK_METHOD(AuxSendsParamsChannel, auxSendsParamsChanged, (), (const, override)); + + MOCK_METHOD(void, processInput, (const TrackId), (const, override)); + MOCK_METHOD(RetVal, inputProcessingProgress, (const TrackId), (const, override)); + + MOCK_METHOD(void, clearCache, (const TrackId), (const, override)); + MOCK_METHOD(void, clearSources, (), (override)); + MOCK_METHOD(void, clearMasterOutputParams, (), (override)); + MOCK_METHOD(void, clearAllFx, (), (override)); + + MOCK_METHOD(RetVal, signalChanges, (const TrackId), (const, override)); + + MOCK_METHOD(RetVal, automatedControlParamsChanges, (const TrackId), (const, override)); + + MOCK_METHOD(async::Promise, prepareToPlay, (), (override)); + + MOCK_METHOD(void, play, (const secs_t), (override)); + MOCK_METHOD(void, seek, (const secs_t, const bool), (override)); + MOCK_METHOD(void, stop, (), (override)); + MOCK_METHOD(void, pause, (), (override)); + MOCK_METHOD(void, resume, (const secs_t), (override)); + + MOCK_METHOD(void, setDuration, (const secs_t), (override)); + MOCK_METHOD(Ret, setLoop, (const secs_t, const secs_t), (override)); + MOCK_METHOD(void, resetLoop, (), (override)); + + MOCK_METHOD(PlaybackStatus, playbackStatus, (), (const, override)); + MOCK_METHOD(async::Channel, playbackStatusChanged, (), (const, override)); + MOCK_METHOD(secs_t, playbackPosition, (), (const, override)); + MOCK_METHOD(async::Channel, playbackPositionChanged, (), (const, override)); + + MOCK_METHOD(async::Promise, saveSoundTrack, (io::IODevice&, const SoundTrackFormat&), (override)); + MOCK_METHOD(SaveSoundTrackProgress, saveSoundTrackProgressChanged, (), (const, override)); + MOCK_METHOD(void, abortSavingAllSoundTracks, (), (override)); +}; +} diff --git a/framework/audio/tests/mocks/audioenginemock.h b/framework/audio/tests/mocks/audioenginemock.h new file mode 100644 index 0000000000..258ffaa2e7 --- /dev/null +++ b/framework/audio/tests/mocks/audioenginemock.h @@ -0,0 +1,45 @@ +/* + * SPDX-License-Identifier: GPL-3.0-only + * MuseScore-CLA-applies + * + * MuseScore + * Music Composition & Notation + * + * Copyright (C) 2026 MuseScore Limited and others + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License version 3 as + * published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ +#pragma once + +#include + +#include "audio/engine/internal/iaudioengine.h" + +namespace muse::audio::engine { +class AudioEngineMock : public IAudioEngine +{ +public: + MOCK_METHOD(Ret, init, (const OutputSpec&), (override)); + MOCK_METHOD(void, deinit, (), (override)); + + MOCK_METHOD(RetVal >, addAudioContext, (const AudioCtxId&), (override)); + MOCK_METHOD(std::shared_ptr, context, (const AudioCtxId&), (const, override)); + MOCK_METHOD(void, destroyContext, (const AudioCtxId&), (override)); + + MOCK_METHOD(void, setOutputSpec, (const OutputSpec&), (override)); + MOCK_METHOD(OutputSpec, outputSpec, (), (const, override)); + MOCK_METHOD(async::Channel, outputSpecChanged, (), (const, override)); + + MOCK_METHOD(samples_t, process, (float*, samples_t), (override)); +}; +} diff --git a/framework/audio/tests/playback_tests.cpp b/framework/audio/tests/playback_tests.cpp index 966f00e3bf..71ea51b66f 100644 --- a/framework/audio/tests/playback_tests.cpp +++ b/framework/audio/tests/playback_tests.cpp @@ -22,6 +22,7 @@ #include #include +#include #include #include "async/processevents.h" @@ -44,6 +45,11 @@ using namespace muse::audio::rpc; namespace { const StreamId STATUS_STREAM_ID = 11; const StreamId POSITION_STREAM_ID = 12; +const StreamId SIGNAL_STREAM_ID = 21; +const StreamId OTHER_SIGNAL_STREAM_ID = 22; + +const TrackId TRACK_ID = 2; +const TrackId OTHER_TRACK_ID = 3; MATCHER_P(StreamIdIs, id, "") { @@ -75,6 +81,11 @@ class Audio_PlaybackTests : public ::testing::Test m_requests.push_back({ msg, onResponse }); })); + ON_CALL(*m_rpcChannel, onNotification(_, _, _)) + .WillByDefault(Invoke([this](CtxId, MsgCode code, NotificationHandler h) { + m_notifications[code] = h; + })); + m_playback = std::make_shared(m_ctx); } @@ -122,6 +133,21 @@ class Audio_PlaybackTests : public ::testing::Test ASSERT_TRUE(m_playback->isInited()); } + void notify(MsgCode code, const ByteArray& data) + { + auto it = m_notifications.find(code); + ASSERT_TRUE(it != m_notifications.end() && it->second); + it->second(make_notification(rpc::ctxId(m_ctx), code, data)); + } + + void requestSignalChanges(TrackId trackId, StreamId streamId) + { + m_playback->signalChanges(trackId); + async::processMessages(); + + respond(MsgCode::GetSignalChanges, RpcPacker::pack(RetVal::make_ok(streamId))); + } + void respondPlayerStreams() { respond(MsgCode::GetPlaybackStatus, RpcPacker::pack(make_ok(), PlaybackStatus::Running, STATUS_STREAM_ID)); @@ -135,6 +161,7 @@ class Audio_PlaybackTests : public ::testing::Test std::shared_ptr m_playback; std::vector m_requests; + std::map m_notifications; }; TEST_F(Audio_PlaybackTests, Player_IsCreatedOncePerContext) @@ -213,4 +240,27 @@ TEST_F(Audio_PlaybackTests, Deinit_LateResponsesAreIgnored) // [THEN] The player state is not changed EXPECT_EQ(player->playbackStatus(), PlaybackStatus::Stopped); } + +TEST_F(Audio_PlaybackTests, TrackRemoved_RemovesTrackStreams) +{ + // [GIVEN] The signal changes of two tracks are requested + initContext(); + requestSignalChanges(TRACK_ID, SIGNAL_STREAM_ID); + requestSignalChanges(OTHER_TRACK_ID, OTHER_SIGNAL_STREAM_ID); + + // [THEN] Only the stream of the removed track is removed + EXPECT_CALL(*m_rpcChannel, removeStream(SIGNAL_STREAM_ID)); + EXPECT_CALL(*m_rpcChannel, removeStream(OTHER_SIGNAL_STREAM_ID)).Times(0); + + // [WHEN] The engine notifies that the track is removed + notify(MsgCode::TrackRemoved, RpcPacker::pack(TRACK_ID)); + ::testing::Mock::VerifyAndClearExpectations(m_rpcChannel.get()); + + // [THEN] The stream of the other track is removed on deinit + EXPECT_CALL(*m_rpcChannel, removeStream(OTHER_SIGNAL_STREAM_ID)); + EXPECT_CALL(*m_rpcChannel, removeStream(SIGNAL_STREAM_ID)).Times(0); + + // [WHEN] Deinit the playback + m_playback->deinit(); +} } From 88cbadd6afcbc7b554b1fa169398884c6928c518 Mon Sep 17 00:00:00 2001 From: Eism Date: Mon, 2 Nov 2026 12:31:48 +0200 Subject: [PATCH 5/9] clean up RPC streams on track removal and context deinit Track and remove RPC send/receive streams when their associated audio tracks are removed or when the audio context is deinitialized. This ensures proper resource management and prevents stream leaks. --- .../engine/internal/enginerpccontroller.cpp | 57 +++++++++++++++++++ .../engine/internal/enginerpccontroller.h | 13 +++++ framework/audio/main/internal/playback.cpp | 42 +++++++++++++- framework/audio/main/internal/playback.h | 7 +++ 4 files changed, 116 insertions(+), 3 deletions(-) diff --git a/framework/audio/engine/internal/enginerpccontroller.cpp b/framework/audio/engine/internal/enginerpccontroller.cpp index 8b6c1be61f..571aed4f68 100644 --- a/framework/audio/engine/internal/enginerpccontroller.cpp +++ b/framework/audio/engine/internal/enginerpccontroller.cpp @@ -349,6 +349,7 @@ void EngineRpcController::init() }); acontext->trackRemoved().onReceive(this, [this, ctxId](TrackId trackId) { + removeTrackStreams(ctxId, trackId); channel()->send(rpc::make_notification(ctxId, MsgCode::TrackRemoved, RpcPacker::pack(trackId))); }); @@ -506,6 +507,7 @@ void EngineRpcController::init() StreamId streamId = 0; if (ret.ret) { streamId = channel()->addSendStream(StreamName::InputProcessingProgressStream, ret.val.processedChannel); + addTrackStream(msg.ctxId, trackId, streamId); } return make_response(msg, RpcPacker::pack(ret.ret, ret.val.isStarted, streamId)); } else { @@ -571,6 +573,7 @@ void EngineRpcController::init() StreamId streamId = 0; if (ret.ret) { streamId = channel()->addSendStream(StreamName::AudioSignalStream, ret.val); + addTrackStream(msg.ctxId, trackId, streamId); } RetVal res; @@ -594,6 +597,7 @@ void EngineRpcController::init() StreamId streamId = 0; if (ret.ret) { streamId = channel()->addSendStream(StreamName::AutomatedControlParamsStream, ret.val); + addTrackStream(msg.ctxId, trackId, streamId); } RetVal res; @@ -728,6 +732,7 @@ void EngineRpcController::init() PlaybackStatus status = actx->playbackStatus(); async::Channel ch = actx->playbackStatusChanged(); StreamId streamId = channel()->addSendStream(StreamName::PlaybackStatusStream, ch); + addContextStream(msg.ctxId, streamId); return make_response(msg, RpcPacker::pack(make_ok(), status, streamId)); } else { return make_response(msg, RpcPacker::pack(make_ret(Err::InvalidContext), PlaybackStatus::Stopped, 0)); @@ -741,6 +746,7 @@ void EngineRpcController::init() secs_t pos = actx->playbackPosition(); async::Channel ch = actx->playbackPositionChanged(); StreamId streamId = channel()->addSendStream(StreamName::PlaybackPositionStream, ch); + addContextStream(msg.ctxId, streamId); return make_response(msg, RpcPacker::pack(make_ok(), pos, streamId)); } else { return make_response(msg, RpcPacker::pack(make_ret(Err::InvalidContext), 0, 0)); @@ -819,12 +825,63 @@ void EngineRpcController::init() audioContext->sourceParamsChanged().disconnect(this); audioContext->fxChainParamsChanged().disconnect(this); + removeContextStreams(ctxId); + audioEngine()->destroyContext(ctxId); return make_response_ret(msg, make_ok()); }); } +void EngineRpcController::addContextStream(rpc::CtxId ctxId, rpc::StreamId streamId) +{ + m_contextStreams[ctxId].streams.push_back(streamId); +} + +void EngineRpcController::addTrackStream(rpc::CtxId ctxId, TrackId trackId, rpc::StreamId streamId) +{ + m_contextStreams[ctxId].trackStreams[trackId].push_back(streamId); +} + +void EngineRpcController::removeTrackStreams(rpc::CtxId ctxId, TrackId trackId) +{ + auto ctxIt = m_contextStreams.find(ctxId); + if (ctxIt == m_contextStreams.end()) { + return; + } + + auto trackIt = ctxIt->second.trackStreams.find(trackId); + if (trackIt == ctxIt->second.trackStreams.end()) { + return; + } + + for (StreamId streamId : trackIt->second) { + channel()->removeStream(streamId); + } + + ctxIt->second.trackStreams.erase(trackIt); +} + +void EngineRpcController::removeContextStreams(rpc::CtxId ctxId) +{ + auto it = m_contextStreams.find(ctxId); + if (it == m_contextStreams.end()) { + return; + } + + for (StreamId streamId : it->second.streams) { + channel()->removeStream(streamId); + } + + for (const auto& pair : it->second.trackStreams) { + for (StreamId streamId : pair.second) { + channel()->removeStream(streamId); + } + } + + m_contextStreams.erase(it); +} + void EngineRpcController::onLongRequest(rpc::CtxId ctxId, rpc::MsgCode code, const RequestHandler& h) { onRequest(OperationType::LongOperation, ctxId, code, h); diff --git a/framework/audio/engine/internal/enginerpccontroller.h b/framework/audio/engine/internal/enginerpccontroller.h index c7fd71ba72..4fbef2ca3d 100644 --- a/framework/audio/engine/internal/enginerpccontroller.h +++ b/framework/audio/engine/internal/enginerpccontroller.h @@ -71,5 +71,18 @@ class EngineRpcController : public async::Asyncable async::Channel m_saveSoundTrackProgressStream; rpc::StreamId m_saveSoundTrackProgressStreamId = 0; + + //! NOTE The send streams are removed on the track removal and on the context deinit + struct ContextStreams { + std::vector streams; + std::map > trackStreams; + }; + + void addContextStream(rpc::CtxId ctxId, rpc::StreamId streamId); + void addTrackStream(rpc::CtxId ctxId, TrackId trackId, rpc::StreamId streamId); + void removeTrackStreams(rpc::CtxId ctxId, TrackId trackId); + void removeContextStreams(rpc::CtxId ctxId); + + std::map m_contextStreams; }; } diff --git a/framework/audio/main/internal/playback.cpp b/framework/audio/main/internal/playback.cpp index b767ef0aa3..5b9cf095d4 100644 --- a/framework/audio/main/internal/playback.cpp +++ b/framework/audio/main/internal/playback.cpp @@ -60,6 +60,7 @@ async::Promise Playback::init() IF_ASSERT_FAILED(RpcPacker::unpack(msg.data, trackId)) { return; } + removeTrackStreams(trackId); m_trackRemoved.send(trackId); }); @@ -173,10 +174,42 @@ void Playback::deinit() m_player.reset(); } + removeAllTrackStreams(); + channel()->send(rpc::make_request(rpc::GLOBAL_CTX_ID, MsgCode::ContextDeinit, RpcPacker::pack(ctxId()))); m_inited.set(false); } +void Playback::addTrackStream(TrackId trackId, rpc::StreamId streamId) const +{ + m_trackStreams[trackId].push_back(streamId); +} + +void Playback::removeTrackStreams(TrackId trackId) const +{ + auto it = m_trackStreams.find(trackId); + if (it == m_trackStreams.end()) { + return; + } + + for (StreamId streamId : it->second) { + channel()->removeStream(streamId); + } + + m_trackStreams.erase(it); +} + +void Playback::removeAllTrackStreams() const +{ + for (const auto& pair : m_trackStreams) { + for (StreamId streamId : pair.second) { + channel()->removeStream(streamId); + } + } + + m_trackStreams.clear(); +} + bool Playback::isInited() const { return m_inited.val; @@ -559,7 +592,7 @@ muse::async::Promise Playback::inputProcessingProgress( return async::make_promise([this, trackId](auto resolve, auto reject) { ONLY_AUDIO_MAIN_THREAD; Msg msg = rpc::make_request(ctxId(), MsgCode::GetInputProcessingProgress, RpcPacker::pack(trackId)); - channel()->send(msg, [this, resolve, reject](const Msg& res) { + channel()->send(msg, [this, trackId, resolve, reject](const Msg& res) { ONLY_AUDIO_MAIN_THREAD; Ret ret; bool isStarted = false; @@ -573,6 +606,7 @@ muse::async::Promise Playback::inputProcessingProgress( InputProcessingProgress prog; prog.isStarted = isStarted; channel()->addReceiveStream(StreamName::InputProcessingProgressStream, streamId, prog.processedChannel); + addTrackStream(trackId, streamId); (void)resolve(prog); } else { doReject(MsgCode::GetInputProcessingProgress, reject, ret); @@ -617,7 +651,7 @@ async::Promise Playback::signalChanges(const TrackId trackId return async::make_promise([this, trackId](auto resolve, auto reject) { ONLY_AUDIO_MAIN_THREAD; Msg msg = rpc::make_request(ctxId(), MsgCode::GetSignalChanges, RpcPacker::pack(trackId)); - channel()->send(msg, [this, resolve, reject](const Msg& res) { + channel()->send(msg, [this, trackId, resolve, reject](const Msg& res) { ONLY_AUDIO_MAIN_THREAD; RetVal ret; IF_ASSERT_FAILED(RpcPacker::unpack(res.data, ret)) { @@ -628,6 +662,7 @@ async::Promise Playback::signalChanges(const TrackId trackId if (ret.ret) { AudioSignalChanges ch; channel()->addReceiveStream(StreamName::AudioSignalStream, ret.val, ch); + addTrackStream(trackId, ret.val); (void)resolve(ch); } else { doReject(MsgCode::GetSignalChanges, reject, ret.ret); @@ -648,7 +683,7 @@ async::Promise Playback::automatedControlParamsCh return async::make_promise([this, trackId](auto resolve, auto reject) { ONLY_AUDIO_MAIN_THREAD; Msg msg = rpc::make_request(ctxId(), MsgCode::GetAutomatedControlParamsChanges, RpcPacker::pack(trackId)); - channel()->send(msg, [this, resolve, reject](const Msg& res) { + channel()->send(msg, [this, trackId, resolve, reject](const Msg& res) { ONLY_AUDIO_MAIN_THREAD; RetVal ret; IF_ASSERT_FAILED(RpcPacker::unpack(res.data, ret)) { @@ -659,6 +694,7 @@ async::Promise Playback::automatedControlParamsCh if (ret.ret) { AutomatedControlParamsChanges ch; channel()->addReceiveStream(StreamName::AutomatedControlParamsStream, ret.val, ch); + addTrackStream(trackId, ret.val); (void)resolve(ch); } else { doReject(MsgCode::GetAutomatedControlParamsChanges, reject, ret.ret); diff --git a/framework/audio/main/internal/playback.h b/framework/audio/main/internal/playback.h index e8d7078ff3..00e663be19 100644 --- a/framework/audio/main/internal/playback.h +++ b/framework/audio/main/internal/playback.h @@ -125,6 +125,10 @@ class Playback : public IPlayback, public async::Asyncable, public Contextable rpc::CtxId ctxId() const; + void addTrackStream(TrackId trackId, rpc::StreamId streamId) const; + void removeTrackStreams(TrackId trackId) const; + void removeAllTrackStreams() const; + ValCh m_inited; async::Channel m_trackAdded; @@ -142,5 +146,8 @@ class Playback : public IPlayback, public async::Asyncable, public Contextable mutable rpc::StreamId m_saveSoundTrackProgressStreamId = 0; mutable std::shared_ptr m_player; + + //! NOTE The receive streams are removed on the track removal and on the deinit + mutable std::map > m_trackStreams; }; } From dd3a6c208a88ba2deffd94f20a34405a4578c6a1 Mon Sep 17 00:00:00 2001 From: Eism Date: Mon, 2 Nov 2026 13:01:55 +0200 Subject: [PATCH 6/9] added cancel responce to rpc --- framework/audio/common/rpc/irpcchannel.h | 1 + .../common/rpc/platform/general/generalrpcchannel.cpp | 9 +++++++++ .../common/rpc/platform/general/generalrpcchannel.h | 1 + .../audio/common/rpc/platform/web/webrpcchannel.cpp | 5 +++++ framework/audio/common/rpc/platform/web/webrpcchannel.h | 1 + framework/audio/tests/mocks/rpcchannelmock.h | 1 + 6 files changed, 18 insertions(+) diff --git a/framework/audio/common/rpc/irpcchannel.h b/framework/audio/common/rpc/irpcchannel.h index dc392269c6..b541f71f4b 100644 --- a/framework/audio/common/rpc/irpcchannel.h +++ b/framework/audio/common/rpc/irpcchannel.h @@ -403,6 +403,7 @@ class IRpcChannel : MODULE_GLOBAL_INTERFACE, public IStreamRpcChannel virtual void process() = 0; virtual void send(const Msg& msg, const ResponseHandler& onResponse = nullptr) = 0; + virtual void cancelResponse(CallId callId) = 0; virtual void onRequest(CtxId ctxId, MsgCode code, RequestHandler h) = 0; virtual void onNotification(CtxId ctxId, MsgCode code, NotificationHandler h) = 0; diff --git a/framework/audio/common/rpc/platform/general/generalrpcchannel.cpp b/framework/audio/common/rpc/platform/general/generalrpcchannel.cpp index 09dc0b91bb..c20ab58067 100644 --- a/framework/audio/common/rpc/platform/general/generalrpcchannel.cpp +++ b/framework/audio/common/rpc/platform/general/generalrpcchannel.cpp @@ -92,6 +92,15 @@ void GeneralRpcChannel::send(const Msg& msg, const ResponseHandler& onResponse) } } +void GeneralRpcChannel::cancelResponse(CallId callId) +{ + if (s_isMainThread) { + m_mainRpcData.onResponses.erase(callId); + } else { + m_engineRpcData.onResponses.erase(callId); + } +} + void GeneralRpcChannel::receive(RpcData& to, const Msg& m) { RPCLOG() << "ctxId: " << m.ctxId diff --git a/framework/audio/common/rpc/platform/general/generalrpcchannel.h b/framework/audio/common/rpc/platform/general/generalrpcchannel.h index fd5d46ada1..7a5cc26749 100644 --- a/framework/audio/common/rpc/platform/general/generalrpcchannel.h +++ b/framework/audio/common/rpc/platform/general/generalrpcchannel.h @@ -39,6 +39,7 @@ class GeneralRpcChannel : public IRpcChannel // IRpcChannel // msgs void send(const Msg& msg, const ResponseHandler& onResponse = nullptr) override; + void cancelResponse(CallId callId) override; void onRequest(CtxId ctxId, MsgCode code, RequestHandler h) override; void onNotification(CtxId ctxId, MsgCode code, NotificationHandler h) override; diff --git a/framework/audio/common/rpc/platform/web/webrpcchannel.cpp b/framework/audio/common/rpc/platform/web/webrpcchannel.cpp index e662949b54..b449386040 100644 --- a/framework/audio/common/rpc/platform/web/webrpcchannel.cpp +++ b/framework/audio/common/rpc/platform/web/webrpcchannel.cpp @@ -123,6 +123,11 @@ void WebRpcChannel::send(const Msg& msg, const Handler& onResponse) rpcSend(&buffer[0], buffer.size()); } +void WebRpcChannel::cancelResponse(CallId callId) +{ + m_data.onResponses.erase(callId); +} + void WebRpcChannel::receive(const ByteArray& data) { IF_ASSERT_FAILED(data.size() > 0) { diff --git a/framework/audio/common/rpc/platform/web/webrpcchannel.h b/framework/audio/common/rpc/platform/web/webrpcchannel.h index f91d069235..e43a475ebe 100644 --- a/framework/audio/common/rpc/platform/web/webrpcchannel.h +++ b/framework/audio/common/rpc/platform/web/webrpcchannel.h @@ -35,6 +35,7 @@ class WebRpcChannel : public IRpcChannel void process() override; void send(const Msg& msg, const Handler& onResponse = nullptr) override; + void cancelResponse(CallId callId) override; void onRequest(MsgCode code, Handler h) override; void onNotification(MsgCode code, Handler h) override; void listenAll(Handler h) override; diff --git a/framework/audio/tests/mocks/rpcchannelmock.h b/framework/audio/tests/mocks/rpcchannelmock.h index 66d39b3855..0da058e1cc 100644 --- a/framework/audio/tests/mocks/rpcchannelmock.h +++ b/framework/audio/tests/mocks/rpcchannelmock.h @@ -35,6 +35,7 @@ class RpcChannelMock : public IRpcChannel MOCK_METHOD(void, process, (), (override)); MOCK_METHOD(void, send, (const Msg&, const ResponseHandler&), (override)); + MOCK_METHOD(void, cancelResponse, (CallId), (override)); MOCK_METHOD(void, onRequest, (CtxId, MsgCode, RequestHandler), (override)); MOCK_METHOD(void, onNotification, (CtxId, MsgCode, NotificationHandler), (override)); From 27ed2a0f86108fc490f366ce8589637f55da821a Mon Sep 17 00:00:00 2001 From: Eism Date: Mon, 2 Nov 2026 13:06:42 +0200 Subject: [PATCH 7/9] ensure RPC requests are canceled on object deinit or track removal Implement cancellation mechanisms for pending RPC requests to prevent callbacks from operating on deinitialized objects or removed resources. This prevents use-after-free issues and ensures proper asynchronous resource cleanup. --- framework/audio/main/internal/playback.cpp | 71 +++++++++++++++++- framework/audio/main/internal/playback.h | 10 +++ framework/audio/main/internal/player.cpp | 34 ++++++--- framework/audio/main/internal/player.h | 8 ++- framework/audio/tests/playback_tests.cpp | 84 +++++++++++++++++++--- 5 files changed, 183 insertions(+), 24 deletions(-) diff --git a/framework/audio/main/internal/playback.cpp b/framework/audio/main/internal/playback.cpp index 5b9cf095d4..d080911495 100644 --- a/framework/audio/main/internal/playback.cpp +++ b/framework/audio/main/internal/playback.cpp @@ -35,6 +35,11 @@ using namespace muse::audio; using namespace muse::audio::rpc; using namespace muse::async; +Playback::~Playback() +{ + cancelAllPendingTrackRequests(); +} + rpc::CtxId Playback::ctxId() const { return rpc::ctxId(iocContext()); @@ -180,6 +185,47 @@ void Playback::deinit() m_inited.set(false); } +void Playback::addPendingTrackRequest(TrackId trackId, rpc::CallId callId, const RejectFunc& reject) const +{ + m_pendingTrackRequests[trackId][callId] = reject; +} + +void Playback::removePendingTrackRequest(TrackId trackId, rpc::CallId callId) const +{ + auto it = m_pendingTrackRequests.find(trackId); + if (it == m_pendingTrackRequests.end()) { + return; + } + + it->second.erase(callId); + if (it->second.empty()) { + m_pendingTrackRequests.erase(it); + } +} + +void Playback::cancelPendingTrackRequests(TrackId trackId) const +{ + auto it = m_pendingTrackRequests.find(trackId); + if (it == m_pendingTrackRequests.end()) { + return; + } + + std::map requests = std::move(it->second); + m_pendingTrackRequests.erase(it); + + for (const auto& pair : requests) { + channel()->cancelResponse(pair.first); + pair.second(audio::make_ret(Err::InvalidTrackId)); + } +} + +void Playback::cancelAllPendingTrackRequests() const +{ + while (!m_pendingTrackRequests.empty()) { + cancelPendingTrackRequests(m_pendingTrackRequests.begin()->first); + } +} + void Playback::addTrackStream(TrackId trackId, rpc::StreamId streamId) const { m_trackStreams[trackId].push_back(streamId); @@ -187,6 +233,8 @@ void Playback::addTrackStream(TrackId trackId, rpc::StreamId streamId) const void Playback::removeTrackStreams(TrackId trackId) const { + cancelPendingTrackRequests(trackId); + auto it = m_trackStreams.find(trackId); if (it == m_trackStreams.end()) { return; @@ -208,6 +256,8 @@ void Playback::removeAllTrackStreams() const } m_trackStreams.clear(); + + cancelAllPendingTrackRequests(); } bool Playback::isInited() const @@ -592,8 +642,13 @@ muse::async::Promise Playback::inputProcessingProgress( return async::make_promise([this, trackId](auto resolve, auto reject) { ONLY_AUDIO_MAIN_THREAD; Msg msg = rpc::make_request(ctxId(), MsgCode::GetInputProcessingProgress, RpcPacker::pack(trackId)); - channel()->send(msg, [this, trackId, resolve, reject](const Msg& res) { + addPendingTrackRequest(trackId, msg.callId, [reject](const Ret& ret) { + doReject(MsgCode::GetInputProcessingProgress, reject, ret); + }); + channel()->send(msg, [this, trackId, callId = msg.callId, resolve, reject](const Msg& res) { ONLY_AUDIO_MAIN_THREAD; + removePendingTrackRequest(trackId, callId); + Ret ret; bool isStarted = false; StreamId streamId = 0; @@ -651,8 +706,13 @@ async::Promise Playback::signalChanges(const TrackId trackId return async::make_promise([this, trackId](auto resolve, auto reject) { ONLY_AUDIO_MAIN_THREAD; Msg msg = rpc::make_request(ctxId(), MsgCode::GetSignalChanges, RpcPacker::pack(trackId)); - channel()->send(msg, [this, trackId, resolve, reject](const Msg& res) { + addPendingTrackRequest(trackId, msg.callId, [reject](const Ret& ret) { + doReject(MsgCode::GetSignalChanges, reject, ret); + }); + channel()->send(msg, [this, trackId, callId = msg.callId, resolve, reject](const Msg& res) { ONLY_AUDIO_MAIN_THREAD; + removePendingTrackRequest(trackId, callId); + RetVal ret; IF_ASSERT_FAILED(RpcPacker::unpack(res.data, ret)) { doReject(MsgCode::GetSignalChanges, reject, audio::make_ret(Err::InvalidRpcData)); @@ -683,8 +743,13 @@ async::Promise Playback::automatedControlParamsCh return async::make_promise([this, trackId](auto resolve, auto reject) { ONLY_AUDIO_MAIN_THREAD; Msg msg = rpc::make_request(ctxId(), MsgCode::GetAutomatedControlParamsChanges, RpcPacker::pack(trackId)); - channel()->send(msg, [this, trackId, resolve, reject](const Msg& res) { + addPendingTrackRequest(trackId, msg.callId, [reject](const Ret& ret) { + doReject(MsgCode::GetAutomatedControlParamsChanges, reject, ret); + }); + channel()->send(msg, [this, trackId, callId = msg.callId, resolve, reject](const Msg& res) { ONLY_AUDIO_MAIN_THREAD; + removePendingTrackRequest(trackId, callId); + RetVal ret; IF_ASSERT_FAILED(RpcPacker::unpack(res.data, ret)) { doReject(MsgCode::GetAutomatedControlParamsChanges, reject, audio::make_ret(Err::InvalidRpcData)); diff --git a/framework/audio/main/internal/playback.h b/framework/audio/main/internal/playback.h index 00e663be19..222b890a5a 100644 --- a/framework/audio/main/internal/playback.h +++ b/framework/audio/main/internal/playback.h @@ -45,6 +45,7 @@ class Playback : public IPlayback, public async::Asyncable, public Contextable public: Playback(const muse::modularity::ContextPtr& ctx) : Contextable(ctx) {} + ~Playback() override; // Init async::Promise init() override; @@ -125,6 +126,12 @@ class Playback : public IPlayback, public async::Asyncable, public Contextable rpc::CtxId ctxId() const; + using RejectFunc = std::function; + void addPendingTrackRequest(TrackId trackId, rpc::CallId callId, const RejectFunc& reject) const; + void removePendingTrackRequest(TrackId trackId, rpc::CallId callId) const; + void cancelPendingTrackRequests(TrackId trackId) const; + void cancelAllPendingTrackRequests() const; + void addTrackStream(TrackId trackId, rpc::StreamId streamId) const; void removeTrackStreams(TrackId trackId) const; void removeAllTrackStreams() const; @@ -149,5 +156,8 @@ class Playback : public IPlayback, public async::Asyncable, public Contextable //! NOTE The receive streams are removed on the track removal and on the deinit mutable std::map > m_trackStreams; + + //! NOTE Responses of the pending requests are canceled on the track removal and on the deinit + mutable std::map > m_pendingTrackRequests; }; } diff --git a/framework/audio/main/internal/player.cpp b/framework/audio/main/internal/player.cpp index b9dd57bb76..a1fd8b489a 100644 --- a/framework/audio/main/internal/player.cpp +++ b/framework/audio/main/internal/player.cpp @@ -31,6 +31,11 @@ using namespace muse::async; using namespace muse::audio; using namespace muse::audio::rpc; +Player::~Player() +{ + cancelPendingResponses(); +} + rpc::CtxId Player::ctxId() const { return rpc::ctxId(iocContext()); @@ -48,11 +53,10 @@ void Player::init() }); Msg msg = rpc::make_request(ctxId(), MsgCode::GetPlaybackStatus); - channel()->send(msg, [this, alive = std::weak_ptr(m_alive)](const Msg& res) { + m_playbackStatusCallId = msg.callId; + channel()->send(msg, [this](const Msg& res) { ONLY_AUDIO_MAIN_THREAD; - if (alive.expired()) { - return; - } + m_playbackStatusCallId = 0; Ret ret; PlaybackStatus status = PlaybackStatus::Stopped; @@ -78,11 +82,10 @@ void Player::init() }); Msg msg = rpc::make_request(ctxId(), MsgCode::GetPlaybackPosition); - channel()->send(msg, [this, alive = std::weak_ptr(m_alive)](const Msg& res) { + m_playbackPositionCallId = msg.callId; + channel()->send(msg, [this](const Msg& res) { ONLY_AUDIO_MAIN_THREAD; - if (alive.expired()) { - return; - } + m_playbackPositionCallId = 0; Ret ret; secs_t pos = 0.0; @@ -107,7 +110,7 @@ void Player::deinit() { ONLY_AUDIO_MAIN_THREAD; - m_alive.reset(); + cancelPendingResponses(); if (m_playbackStatusStreamId != 0) { channel()->removeStream(m_playbackStatusStreamId); @@ -123,6 +126,19 @@ void Player::deinit() m_playbackPositionChanged.disconnect(this); } +void Player::cancelPendingResponses() +{ + if (m_playbackStatusCallId != 0) { + channel()->cancelResponse(m_playbackStatusCallId); + m_playbackStatusCallId = 0; + } + + if (m_playbackPositionCallId != 0) { + channel()->cancelResponse(m_playbackPositionCallId); + m_playbackPositionCallId = 0; + } +} + async::Promise Player::prepareToPlay() { ONLY_AUDIO_MAIN_THREAD; diff --git a/framework/audio/main/internal/player.h b/framework/audio/main/internal/player.h index ba3726a778..8683929501 100644 --- a/framework/audio/main/internal/player.h +++ b/framework/audio/main/internal/player.h @@ -38,6 +38,7 @@ class Player : public IPlayer, public Contextable, public async::Asyncable public: Player(const muse::modularity::ContextPtr& ctx) : Contextable(ctx) {} + ~Player() override; void init(); void deinit(); @@ -70,11 +71,14 @@ class Player : public IPlayer, public Contextable, public async::Asyncable secs_t m_playbackPosition = 0.0; async::Channel m_playbackPositionChanged; + void cancelPendingResponses(); + rpc::StreamId m_playbackStatusStreamId = 0; rpc::StreamId m_playbackPositionStreamId = 0; - //! NOTE Responses may come after deinit, they must not be handled - std::shared_ptr m_alive = std::make_shared(true); + //! NOTE Pending requests, their responses are dropped on deinit + rpc::CallId m_playbackStatusCallId = 0; + rpc::CallId m_playbackPositionCallId = 0; }; } diff --git a/framework/audio/tests/playback_tests.cpp b/framework/audio/tests/playback_tests.cpp index 71ea51b66f..7053b40420 100644 --- a/framework/audio/tests/playback_tests.cpp +++ b/framework/audio/tests/playback_tests.cpp @@ -25,8 +25,10 @@ #include #include +#include "async/asyncable.h" #include "async/processevents.h" +#include "audio/common/audioerrors.h" #include "audio/common/rpc/rpcpacker.h" #include "audio/main/internal/playback.h" @@ -58,7 +60,7 @@ MATCHER_P(StreamIdIs, id, "") } namespace muse::audio { -class Audio_PlaybackTests : public ::testing::Test +class Audio_PlaybackTests : public ::testing::Test, public async::Asyncable { protected: struct Request { @@ -81,6 +83,14 @@ class Audio_PlaybackTests : public ::testing::Test m_requests.push_back({ msg, onResponse }); })); + //! NOTE As the channel, the response of the canceled request is dropped + ON_CALL(*m_rpcChannel, cancelResponse(_)) + .WillByDefault(Invoke([this](CallId callId) { + std::erase_if(m_requests, [callId](const Request& r) { + return r.msg.callId == callId; + }); + })); + ON_CALL(*m_rpcChannel, onNotification(_, _, _)) .WillByDefault(Invoke([this](CtxId, MsgCode code, NotificationHandler h) { m_notifications[code] = h; @@ -222,23 +232,21 @@ TEST_F(Audio_PlaybackTests, Deinit_RemovesPlayerStreams) EXPECT_NE(newPlayer, player); } -TEST_F(Audio_PlaybackTests, Deinit_LateResponsesAreIgnored) +TEST_F(Audio_PlaybackTests, Deinit_CancelsPlayerPendingRequests) { // [GIVEN] The player has not received the responses yet initContext(); IPlayerPtr player = m_playback->player(); + // [THEN] The responses are canceled + EXPECT_CALL(*m_rpcChannel, cancelResponse(_)).Times(2); + // [WHEN] Deinit the playback m_playback->deinit(); - // [THEN] The streams are not added by the late responses - EXPECT_CALL(*m_rpcChannel, addStream(_)).Times(0); - - // [WHEN] The engine responds after deinit - respondPlayerStreams(); - - // [THEN] The player state is not changed - EXPECT_EQ(player->playbackStatus(), PlaybackStatus::Stopped); + // [THEN] No responses are expected anymore + EXPECT_EQ(requestCount(MsgCode::GetPlaybackStatus), 0); + EXPECT_EQ(requestCount(MsgCode::GetPlaybackPosition), 0); } TEST_F(Audio_PlaybackTests, TrackRemoved_RemovesTrackStreams) @@ -263,4 +271,60 @@ TEST_F(Audio_PlaybackTests, TrackRemoved_RemovesTrackStreams) // [WHEN] Deinit the playback m_playback->deinit(); } + +TEST_F(Audio_PlaybackTests, TrackRemoved_CancelsPendingRequests) +{ + // [GIVEN] The signal changes of two tracks are requested, but the engine has not responded yet + initContext(); + + bool rejected = false; + m_playback->signalChanges(TRACK_ID).onReject(this, [&rejected](int code, const std::string&) { + rejected = code == static_cast(Err::InvalidTrackId); + }); + m_playback->signalChanges(OTHER_TRACK_ID); + async::processMessages(); + + // [WHEN] The track is removed + notify(MsgCode::TrackRemoved, RpcPacker::pack(TRACK_ID)); + async::processMessages(); + + // [THEN] The request of the removed track is canceled and rejected + EXPECT_TRUE(rejected); + + // [THEN] The request of the other track is still pending + EXPECT_EQ(requestCount(MsgCode::GetSignalChanges), 1); +} + +TEST_F(Audio_PlaybackTests, Deinit_CancelsPendingTrackRequests) +{ + // [GIVEN] The signal changes are requested, but the engine has not responded yet + initContext(); + + bool rejected = false; + m_playback->signalChanges(TRACK_ID).onReject(this, [&rejected](int code, const std::string&) { + rejected = code == static_cast(Err::InvalidTrackId); + }); + async::processMessages(); + + // [WHEN] Deinit the playback + m_playback->deinit(); + async::processMessages(); + + // [THEN] The request is canceled and rejected + EXPECT_TRUE(rejected); + EXPECT_EQ(requestCount(MsgCode::GetSignalChanges), 0); +} + +TEST_F(Audio_PlaybackTests, Response_RemovesPendingRequest) +{ + // [GIVEN] The signal changes are requested + initContext(); + requestSignalChanges(TRACK_ID, SIGNAL_STREAM_ID); + + // [THEN] The completed request is not canceled + EXPECT_CALL(*m_rpcChannel, cancelResponse(_)).Times(0); + + // [WHEN] The track is removed + notify(MsgCode::TrackRemoved, RpcPacker::pack(TRACK_ID)); +} } From d24c1fb40f8cb546f11378a73764d7593b1068fa Mon Sep 17 00:00:00 2001 From: Eism Date: Mon, 2 Nov 2026 14:22:51 +0200 Subject: [PATCH 8/9] implemented context-aware RPC request and stream cancellation --- framework/audio/common/rpc/irpcchannel.h | 9 ++++--- .../platform/general/generalrpcchannel.cpp | 27 +++++++++++-------- .../rpc/platform/general/generalrpcchannel.h | 9 +++++-- .../common/rpc/platform/web/webrpcchannel.cpp | 19 ++++++++----- .../common/rpc/platform/web/webrpcchannel.h | 9 +++++-- framework/audio/tests/mocks/rpcchannelmock.h | 2 +- 6 files changed, 50 insertions(+), 25 deletions(-) diff --git a/framework/audio/common/rpc/irpcchannel.h b/framework/audio/common/rpc/irpcchannel.h index b541f71f4b..512883d5de 100644 --- a/framework/audio/common/rpc/irpcchannel.h +++ b/framework/audio/common/rpc/irpcchannel.h @@ -403,24 +403,27 @@ class IRpcChannel : MODULE_GLOBAL_INTERFACE, public IStreamRpcChannel virtual void process() = 0; virtual void send(const Msg& msg, const ResponseHandler& onResponse = nullptr) = 0; - virtual void cancelResponse(CallId callId) = 0; + virtual void cancelAllRequests(CtxId ctxId) = 0; virtual void onRequest(CtxId ctxId, MsgCode code, RequestHandler h) = 0; virtual void onNotification(CtxId ctxId, MsgCode code, NotificationHandler h) = 0; // stream (async/channel) template - StreamId addSendStream(StreamName name, const async::Channel& ch) + StreamId addSendStream(CtxId ctxId, StreamName name, const async::Channel& ch) { StreamId id = new_stream_id(); auto s = new RpcStream(this, name, id, StreamType::Send, ch, nullptr); + s->setCtxId(ctxId); addStream(std::shared_ptr(s)); return id; } template - void addReceiveStream(StreamName name, rpc::StreamId id, const async::Channel& ch, const RpcStreamExec& exec = nullptr) + void addReceiveStream(CtxId ctxId, StreamName name, rpc::StreamId id, const async::Channel& ch, + const RpcStreamExec& exec = nullptr) { auto s = new RpcStream(this, name, id, StreamType::Receive, ch, exec); + s->setCtxId(ctxId); addStream(std::shared_ptr(s)); } }; diff --git a/framework/audio/common/rpc/platform/general/generalrpcchannel.cpp b/framework/audio/common/rpc/platform/general/generalrpcchannel.cpp index c20ab58067..b0deaeb4a8 100644 --- a/framework/audio/common/rpc/platform/general/generalrpcchannel.cpp +++ b/framework/audio/common/rpc/platform/general/generalrpcchannel.cpp @@ -79,26 +79,30 @@ void GeneralRpcChannel::send(const Msg& msg, const ResponseHandler& onResponse) if (s_isMainThread) { if (onResponse) { - m_mainRpcData.onResponses[msg.callId] = onResponse; + m_mainRpcData.onResponses[msg.callId] = { msg.ctxId, onResponse }; } m_msgQueue.port1()->send(msg); } else { if (onResponse) { - m_engineRpcData.onResponses[msg.callId] = onResponse; + m_engineRpcData.onResponses[msg.callId] = { msg.ctxId, onResponse }; } m_msgQueue.port2()->send(msg); } } -void GeneralRpcChannel::cancelResponse(CallId callId) +void GeneralRpcChannel::cancelAllRequests(CtxId ctxId) { - if (s_isMainThread) { - m_mainRpcData.onResponses.erase(callId); - } else { - m_engineRpcData.onResponses.erase(callId); - } + RpcData& data = s_isMainThread ? m_mainRpcData : m_engineRpcData; + + std::erase_if(data.onResponses, [ctxId](const auto& pair) { + return pair.second.ctxId == ctxId; + }); + + std::erase_if(data.streams, [ctxId](const auto& pair) { + return pair.second->ctxId() == ctxId; + }); } void GeneralRpcChannel::receive(RpcData& to, const Msg& m) @@ -129,10 +133,11 @@ void GeneralRpcChannel::receive(RpcData& to, const Msg& m) case MsgType::Response: { auto it = to.onResponses.find(m.callId); if (it != to.onResponses.end()) { - if (it->second) { - it->second(m); - } + ResponseHandler handler = std::move(it->second.handler); to.onResponses.erase(it); + if (handler) { + handler(m); + } } } break; case MsgType::Notification: { diff --git a/framework/audio/common/rpc/platform/general/generalrpcchannel.h b/framework/audio/common/rpc/platform/general/generalrpcchannel.h index 7a5cc26749..b58d9d6c5d 100644 --- a/framework/audio/common/rpc/platform/general/generalrpcchannel.h +++ b/framework/audio/common/rpc/platform/general/generalrpcchannel.h @@ -39,7 +39,7 @@ class GeneralRpcChannel : public IRpcChannel // IRpcChannel // msgs void send(const Msg& msg, const ResponseHandler& onResponse = nullptr) override; - void cancelResponse(CallId callId) override; + void cancelAllRequests(CtxId ctxId) override; void onRequest(CtxId ctxId, MsgCode code, RequestHandler h) override; void onNotification(CtxId ctxId, MsgCode code, NotificationHandler h) override; @@ -54,7 +54,12 @@ class GeneralRpcChannel : public IRpcChannel struct RpcData { // msgs std::map onRequests; - std::map onResponses; + struct PendingResponse { + CtxId ctxId = 0; + ResponseHandler handler; + }; + + std::map onResponses; std::map onNotifications; // stream diff --git a/framework/audio/common/rpc/platform/web/webrpcchannel.cpp b/framework/audio/common/rpc/platform/web/webrpcchannel.cpp index b449386040..0579345482 100644 --- a/framework/audio/common/rpc/platform/web/webrpcchannel.cpp +++ b/framework/audio/common/rpc/platform/web/webrpcchannel.cpp @@ -117,15 +117,21 @@ void WebRpcChannel::send(const Msg& msg, const Handler& onResponse) } if (onResponse) { - m_data.onResponses[msg.callId] = onResponse; + m_data.onResponses[msg.callId] = { msg.ctxId, onResponse }; } rpcSend(&buffer[0], buffer.size()); } -void WebRpcChannel::cancelResponse(CallId callId) +void WebRpcChannel::cancelAllRequests(CtxId ctxId) { - m_data.onResponses.erase(callId); + std::erase_if(m_data.onResponses, [ctxId](const auto& pair) { + return pair.second.ctxId == ctxId; + }); + + std::erase_if(m_data.streams, [ctxId](const auto& pair) { + return pair.second->ctxId() == ctxId; + }); } void WebRpcChannel::receive(const ByteArray& data) @@ -179,10 +185,11 @@ void WebRpcChannel::receive(const Msg& msg) case MsgType::Response: { auto it = m_data.onResponses.find(msg.callId); if (it != m_data.onResponses.end()) { - if (it->second) { - it->second(msg); - } + Handler handler = std::move(it->second.handler); m_data.onResponses.erase(it); + if (handler) { + handler(msg); + } } } break; default: { diff --git a/framework/audio/common/rpc/platform/web/webrpcchannel.h b/framework/audio/common/rpc/platform/web/webrpcchannel.h index e43a475ebe..7c3720aec6 100644 --- a/framework/audio/common/rpc/platform/web/webrpcchannel.h +++ b/framework/audio/common/rpc/platform/web/webrpcchannel.h @@ -35,7 +35,7 @@ class WebRpcChannel : public IRpcChannel void process() override; void send(const Msg& msg, const Handler& onResponse = nullptr) override; - void cancelResponse(CallId callId) override; + void cancelAllRequests(CtxId ctxId) override; void onRequest(MsgCode code, Handler h) override; void onNotification(MsgCode code, Handler h) override; void listenAll(Handler h) override; @@ -55,7 +55,12 @@ class WebRpcChannel : public IRpcChannel Handler listenerAll; std::map onRequests; std::map onNotifications; - std::map onResponses; + struct PendingResponse { + CtxId ctxId = 0; + Handler handler; + }; + + std::map onResponses; // stream std::map > streams; diff --git a/framework/audio/tests/mocks/rpcchannelmock.h b/framework/audio/tests/mocks/rpcchannelmock.h index 0da058e1cc..07ed2f88d3 100644 --- a/framework/audio/tests/mocks/rpcchannelmock.h +++ b/framework/audio/tests/mocks/rpcchannelmock.h @@ -35,7 +35,7 @@ class RpcChannelMock : public IRpcChannel MOCK_METHOD(void, process, (), (override)); MOCK_METHOD(void, send, (const Msg&, const ResponseHandler&), (override)); - MOCK_METHOD(void, cancelResponse, (CallId), (override)); + MOCK_METHOD(void, cancelAllRequests, (CtxId), (override)); MOCK_METHOD(void, onRequest, (CtxId, MsgCode, RequestHandler), (override)); MOCK_METHOD(void, onNotification, (CtxId, MsgCode, NotificationHandler), (override)); From 3e051200edb85840ed848d250bf27cddd3b7192e Mon Sep 17 00:00:00 2001 From: Eism Date: Mon, 2 Nov 2026 14:26:15 +0200 Subject: [PATCH 9/9] cancel all requests on playback deinit --- .../engine/internal/enginerpccontroller.cpp | 58 ++++------- .../engine/internal/enginerpccontroller.h | 11 +-- framework/audio/main/internal/playback.cpp | 95 +++---------------- framework/audio/main/internal/playback.h | 12 +-- framework/audio/main/internal/player.cpp | 42 +------- framework/audio/main/internal/player.h | 10 -- .../audio/tests/enginerpccontroller_tests.cpp | 37 ++++---- framework/audio/tests/playback_tests.cpp | 88 ++++------------- 8 files changed, 67 insertions(+), 286 deletions(-) diff --git a/framework/audio/engine/internal/enginerpccontroller.cpp b/framework/audio/engine/internal/enginerpccontroller.cpp index 571aed4f68..875f0024b7 100644 --- a/framework/audio/engine/internal/enginerpccontroller.cpp +++ b/framework/audio/engine/internal/enginerpccontroller.cpp @@ -212,8 +212,8 @@ void EngineRpcController::init() m_execOperation->execOperation(OperationType::QuickOperation, func); }; - channel()->addReceiveStream(StreamName::PlaybackDataMainStream, mainStreamId, playbackData.mainStream, mainExec); - channel()->addReceiveStream(StreamName::PlaybackDataOffStream, offStreamId, playbackData.offStream, offExec); + channel()->addReceiveStream(msg.ctxId, StreamName::PlaybackDataMainStream, mainStreamId, playbackData.mainStream, mainExec); + channel()->addReceiveStream(msg.ctxId, StreamName::PlaybackDataOffStream, offStreamId, playbackData.offStream, offExec); auto addTrackAndSendResponse = [this](const Msg& msg, const TrackName& trackName, const mpe::PlaybackData& playbackData, const TrackParams& params) { @@ -506,7 +506,7 @@ void EngineRpcController::init() RetVal ret = actx->inputProcessingProgress(trackId); StreamId streamId = 0; if (ret.ret) { - streamId = channel()->addSendStream(StreamName::InputProcessingProgressStream, ret.val.processedChannel); + streamId = channel()->addSendStream(msg.ctxId, StreamName::InputProcessingProgressStream, ret.val.processedChannel); addTrackStream(msg.ctxId, trackId, streamId); } return make_response(msg, RpcPacker::pack(ret.ret, ret.val.isStarted, streamId)); @@ -572,7 +572,7 @@ void EngineRpcController::init() RetVal ret = audioContext(msg.ctxId)->signalChanges(trackId); StreamId streamId = 0; if (ret.ret) { - streamId = channel()->addSendStream(StreamName::AudioSignalStream, ret.val); + streamId = channel()->addSendStream(msg.ctxId, StreamName::AudioSignalStream, ret.val); addTrackStream(msg.ctxId, trackId, streamId); } @@ -596,7 +596,7 @@ void EngineRpcController::init() RetVal ret = audioContext(msg.ctxId)->automatedControlParamsChanges(trackId); StreamId streamId = 0; if (ret.ret) { - streamId = channel()->addSendStream(StreamName::AutomatedControlParamsStream, ret.val); + streamId = channel()->addSendStream(msg.ctxId, StreamName::AutomatedControlParamsStream, ret.val); addTrackStream(msg.ctxId, trackId, streamId); } @@ -731,8 +731,7 @@ void EngineRpcController::init() if (auto actx = audioContext(msg.ctxId)) { PlaybackStatus status = actx->playbackStatus(); async::Channel ch = actx->playbackStatusChanged(); - StreamId streamId = channel()->addSendStream(StreamName::PlaybackStatusStream, ch); - addContextStream(msg.ctxId, streamId); + StreamId streamId = channel()->addSendStream(msg.ctxId, StreamName::PlaybackStatusStream, ch); return make_response(msg, RpcPacker::pack(make_ok(), status, streamId)); } else { return make_response(msg, RpcPacker::pack(make_ret(Err::InvalidContext), PlaybackStatus::Stopped, 0)); @@ -745,8 +744,7 @@ void EngineRpcController::init() if (auto actx = audioContext(msg.ctxId)) { secs_t pos = actx->playbackPosition(); async::Channel ch = actx->playbackPositionChanged(); - StreamId streamId = channel()->addSendStream(StreamName::PlaybackPositionStream, ch); - addContextStream(msg.ctxId, streamId); + StreamId streamId = channel()->addSendStream(msg.ctxId, StreamName::PlaybackPositionStream, ch); return make_response(msg, RpcPacker::pack(make_ok(), pos, streamId)); } else { return make_response(msg, RpcPacker::pack(make_ret(Err::InvalidContext), 0, 0)); @@ -793,7 +791,7 @@ void EngineRpcController::init() }); if (m_saveSoundTrackProgressStreamId == 0) { - m_saveSoundTrackProgressStreamId = channel()->addSendStream(StreamName::SaveSoundTrackProgressStream, + m_saveSoundTrackProgressStreamId = channel()->addSendStream(GLOBAL_CTX_ID, StreamName::SaveSoundTrackProgressStream, m_saveSoundTrackProgressStream); } @@ -825,7 +823,8 @@ void EngineRpcController::init() audioContext->sourceParamsChanged().disconnect(this); audioContext->fxChainParamsChanged().disconnect(this); - removeContextStreams(ctxId); + channel()->cancelAllRequests(ctxId); + m_trackStreams.erase(ctxId); audioEngine()->destroyContext(ctxId); @@ -833,25 +832,20 @@ void EngineRpcController::init() }); } -void EngineRpcController::addContextStream(rpc::CtxId ctxId, rpc::StreamId streamId) -{ - m_contextStreams[ctxId].streams.push_back(streamId); -} - void EngineRpcController::addTrackStream(rpc::CtxId ctxId, TrackId trackId, rpc::StreamId streamId) { - m_contextStreams[ctxId].trackStreams[trackId].push_back(streamId); + m_trackStreams[ctxId][trackId].push_back(streamId); } void EngineRpcController::removeTrackStreams(rpc::CtxId ctxId, TrackId trackId) { - auto ctxIt = m_contextStreams.find(ctxId); - if (ctxIt == m_contextStreams.end()) { + auto ctxIt = m_trackStreams.find(ctxId); + if (ctxIt == m_trackStreams.end()) { return; } - auto trackIt = ctxIt->second.trackStreams.find(trackId); - if (trackIt == ctxIt->second.trackStreams.end()) { + auto trackIt = ctxIt->second.find(trackId); + if (trackIt == ctxIt->second.end()) { return; } @@ -859,27 +853,7 @@ void EngineRpcController::removeTrackStreams(rpc::CtxId ctxId, TrackId trackId) channel()->removeStream(streamId); } - ctxIt->second.trackStreams.erase(trackIt); -} - -void EngineRpcController::removeContextStreams(rpc::CtxId ctxId) -{ - auto it = m_contextStreams.find(ctxId); - if (it == m_contextStreams.end()) { - return; - } - - for (StreamId streamId : it->second.streams) { - channel()->removeStream(streamId); - } - - for (const auto& pair : it->second.trackStreams) { - for (StreamId streamId : pair.second) { - channel()->removeStream(streamId); - } - } - - m_contextStreams.erase(it); + ctxIt->second.erase(trackIt); } void EngineRpcController::onLongRequest(rpc::CtxId ctxId, rpc::MsgCode code, const RequestHandler& h) diff --git a/framework/audio/engine/internal/enginerpccontroller.h b/framework/audio/engine/internal/enginerpccontroller.h index 4fbef2ca3d..ae606ac6d7 100644 --- a/framework/audio/engine/internal/enginerpccontroller.h +++ b/framework/audio/engine/internal/enginerpccontroller.h @@ -72,17 +72,10 @@ class EngineRpcController : public async::Asyncable async::Channel m_saveSoundTrackProgressStream; rpc::StreamId m_saveSoundTrackProgressStreamId = 0; - //! NOTE The send streams are removed on the track removal and on the context deinit - struct ContextStreams { - std::vector streams; - std::map > trackStreams; - }; - - void addContextStream(rpc::CtxId ctxId, rpc::StreamId streamId); + //! NOTE The streams of the track are removed on the track removal, on the context deinit all streams of the context are removed by the channel void addTrackStream(rpc::CtxId ctxId, TrackId trackId, rpc::StreamId streamId); void removeTrackStreams(rpc::CtxId ctxId, TrackId trackId); - void removeContextStreams(rpc::CtxId ctxId); - std::map m_contextStreams; + std::map > > m_trackStreams; }; } diff --git a/framework/audio/main/internal/playback.cpp b/framework/audio/main/internal/playback.cpp index d080911495..658e92c7b8 100644 --- a/framework/audio/main/internal/playback.cpp +++ b/framework/audio/main/internal/playback.cpp @@ -37,7 +37,7 @@ using namespace muse::async; Playback::~Playback() { - cancelAllPendingTrackRequests(); + channel()->cancelAllRequests(ctxId()); } rpc::CtxId Playback::ctxId() const @@ -170,6 +170,8 @@ void Playback::deinit() channel()->onNotification(ctxId(), MsgCode::ControlParamsChanged, nullptr); channel()->onNotification(ctxId(), MsgCode::AuxSendsParamsChanged, nullptr); + channel()->cancelAllRequests(ctxId()); + m_saveSoundTrackProgressStream = SaveSoundTrackProgress(); m_saveSoundTrackProgressStreamInited = false; m_saveSoundTrackProgressStreamId = 0; @@ -179,53 +181,12 @@ void Playback::deinit() m_player.reset(); } - removeAllTrackStreams(); + m_trackStreams.clear(); channel()->send(rpc::make_request(rpc::GLOBAL_CTX_ID, MsgCode::ContextDeinit, RpcPacker::pack(ctxId()))); m_inited.set(false); } -void Playback::addPendingTrackRequest(TrackId trackId, rpc::CallId callId, const RejectFunc& reject) const -{ - m_pendingTrackRequests[trackId][callId] = reject; -} - -void Playback::removePendingTrackRequest(TrackId trackId, rpc::CallId callId) const -{ - auto it = m_pendingTrackRequests.find(trackId); - if (it == m_pendingTrackRequests.end()) { - return; - } - - it->second.erase(callId); - if (it->second.empty()) { - m_pendingTrackRequests.erase(it); - } -} - -void Playback::cancelPendingTrackRequests(TrackId trackId) const -{ - auto it = m_pendingTrackRequests.find(trackId); - if (it == m_pendingTrackRequests.end()) { - return; - } - - std::map requests = std::move(it->second); - m_pendingTrackRequests.erase(it); - - for (const auto& pair : requests) { - channel()->cancelResponse(pair.first); - pair.second(audio::make_ret(Err::InvalidTrackId)); - } -} - -void Playback::cancelAllPendingTrackRequests() const -{ - while (!m_pendingTrackRequests.empty()) { - cancelPendingTrackRequests(m_pendingTrackRequests.begin()->first); - } -} - void Playback::addTrackStream(TrackId trackId, rpc::StreamId streamId) const { m_trackStreams[trackId].push_back(streamId); @@ -233,8 +194,6 @@ void Playback::addTrackStream(TrackId trackId, rpc::StreamId streamId) const void Playback::removeTrackStreams(TrackId trackId) const { - cancelPendingTrackRequests(trackId); - auto it = m_trackStreams.find(trackId); if (it == m_trackStreams.end()) { return; @@ -247,19 +206,6 @@ void Playback::removeTrackStreams(TrackId trackId) const m_trackStreams.erase(it); } -void Playback::removeAllTrackStreams() const -{ - for (const auto& pair : m_trackStreams) { - for (StreamId streamId : pair.second) { - channel()->removeStream(streamId); - } - } - - m_trackStreams.clear(); - - cancelAllPendingTrackRequests(); -} - bool Playback::isInited() const { return m_inited.val; @@ -450,8 +396,8 @@ async::Promise Playback::addTrack(const TrackName& trackNa return async::make_promise([this, trackName, playbackData, params](auto resolve, auto reject) { ONLY_AUDIO_MAIN_THREAD; - rpc::StreamId mainStreamId = channel()->addSendStream(StreamName::PlaybackDataMainStream, playbackData.mainStream); - rpc::StreamId offStreamId = channel()->addSendStream(StreamName::PlaybackDataOffStream, playbackData.offStream); + rpc::StreamId mainStreamId = channel()->addSendStream(ctxId(), StreamName::PlaybackDataMainStream, playbackData.mainStream); + rpc::StreamId offStreamId = channel()->addSendStream(ctxId(), StreamName::PlaybackDataOffStream, playbackData.offStream); ByteArray data = RpcPacker::pack(trackName, playbackData, params, mainStreamId, offStreamId); @@ -642,13 +588,8 @@ muse::async::Promise Playback::inputProcessingProgress( return async::make_promise([this, trackId](auto resolve, auto reject) { ONLY_AUDIO_MAIN_THREAD; Msg msg = rpc::make_request(ctxId(), MsgCode::GetInputProcessingProgress, RpcPacker::pack(trackId)); - addPendingTrackRequest(trackId, msg.callId, [reject](const Ret& ret) { - doReject(MsgCode::GetInputProcessingProgress, reject, ret); - }); - channel()->send(msg, [this, trackId, callId = msg.callId, resolve, reject](const Msg& res) { + channel()->send(msg, [this, trackId, resolve, reject](const Msg& res) { ONLY_AUDIO_MAIN_THREAD; - removePendingTrackRequest(trackId, callId); - Ret ret; bool isStarted = false; StreamId streamId = 0; @@ -660,7 +601,7 @@ muse::async::Promise Playback::inputProcessingProgress( if (ret) { InputProcessingProgress prog; prog.isStarted = isStarted; - channel()->addReceiveStream(StreamName::InputProcessingProgressStream, streamId, prog.processedChannel); + channel()->addReceiveStream(ctxId(), StreamName::InputProcessingProgressStream, streamId, prog.processedChannel); addTrackStream(trackId, streamId); (void)resolve(prog); } else { @@ -706,13 +647,8 @@ async::Promise Playback::signalChanges(const TrackId trackId return async::make_promise([this, trackId](auto resolve, auto reject) { ONLY_AUDIO_MAIN_THREAD; Msg msg = rpc::make_request(ctxId(), MsgCode::GetSignalChanges, RpcPacker::pack(trackId)); - addPendingTrackRequest(trackId, msg.callId, [reject](const Ret& ret) { - doReject(MsgCode::GetSignalChanges, reject, ret); - }); - channel()->send(msg, [this, trackId, callId = msg.callId, resolve, reject](const Msg& res) { + channel()->send(msg, [this, trackId, resolve, reject](const Msg& res) { ONLY_AUDIO_MAIN_THREAD; - removePendingTrackRequest(trackId, callId); - RetVal ret; IF_ASSERT_FAILED(RpcPacker::unpack(res.data, ret)) { doReject(MsgCode::GetSignalChanges, reject, audio::make_ret(Err::InvalidRpcData)); @@ -721,7 +657,7 @@ async::Promise Playback::signalChanges(const TrackId trackId if (ret.ret) { AudioSignalChanges ch; - channel()->addReceiveStream(StreamName::AudioSignalStream, ret.val, ch); + channel()->addReceiveStream(ctxId(), StreamName::AudioSignalStream, ret.val, ch); addTrackStream(trackId, ret.val); (void)resolve(ch); } else { @@ -743,13 +679,8 @@ async::Promise Playback::automatedControlParamsCh return async::make_promise([this, trackId](auto resolve, auto reject) { ONLY_AUDIO_MAIN_THREAD; Msg msg = rpc::make_request(ctxId(), MsgCode::GetAutomatedControlParamsChanges, RpcPacker::pack(trackId)); - addPendingTrackRequest(trackId, msg.callId, [reject](const Ret& ret) { - doReject(MsgCode::GetAutomatedControlParamsChanges, reject, ret); - }); - channel()->send(msg, [this, trackId, callId = msg.callId, resolve, reject](const Msg& res) { + channel()->send(msg, [this, trackId, resolve, reject](const Msg& res) { ONLY_AUDIO_MAIN_THREAD; - removePendingTrackRequest(trackId, callId); - RetVal ret; IF_ASSERT_FAILED(RpcPacker::unpack(res.data, ret)) { doReject(MsgCode::GetAutomatedControlParamsChanges, reject, audio::make_ret(Err::InvalidRpcData)); @@ -758,7 +689,7 @@ async::Promise Playback::automatedControlParamsCh if (ret.ret) { AutomatedControlParamsChanges ch; - channel()->addReceiveStream(StreamName::AutomatedControlParamsStream, ret.val, ch); + channel()->addReceiveStream(ctxId(), StreamName::AutomatedControlParamsStream, ret.val, ch); addTrackStream(trackId, ret.val); (void)resolve(ch); } else { @@ -818,7 +749,7 @@ SaveSoundTrackProgress Playback::saveSoundTrackProgressChanged() const if (m_saveSoundTrackProgressStreamId == 0) { m_saveSoundTrackProgressStreamId = ret.val; - channel()->addReceiveStream(StreamName::SaveSoundTrackProgressStream, + channel()->addReceiveStream(ctxId(), StreamName::SaveSoundTrackProgressStream, m_saveSoundTrackProgressStreamId, m_saveSoundTrackProgressStream); } diff --git a/framework/audio/main/internal/playback.h b/framework/audio/main/internal/playback.h index 222b890a5a..4dbed2ef66 100644 --- a/framework/audio/main/internal/playback.h +++ b/framework/audio/main/internal/playback.h @@ -126,15 +126,8 @@ class Playback : public IPlayback, public async::Asyncable, public Contextable rpc::CtxId ctxId() const; - using RejectFunc = std::function; - void addPendingTrackRequest(TrackId trackId, rpc::CallId callId, const RejectFunc& reject) const; - void removePendingTrackRequest(TrackId trackId, rpc::CallId callId) const; - void cancelPendingTrackRequests(TrackId trackId) const; - void cancelAllPendingTrackRequests() const; - void addTrackStream(TrackId trackId, rpc::StreamId streamId) const; void removeTrackStreams(TrackId trackId) const; - void removeAllTrackStreams() const; ValCh m_inited; @@ -154,10 +147,7 @@ class Playback : public IPlayback, public async::Asyncable, public Contextable mutable std::shared_ptr m_player; - //! NOTE The receive streams are removed on the track removal and on the deinit + //! NOTE The receive streams are removed on the track removal, on the deinit all streams of the context are removed by the channel mutable std::map > m_trackStreams; - - //! NOTE Responses of the pending requests are canceled on the track removal and on the deinit - mutable std::map > m_pendingTrackRequests; }; } diff --git a/framework/audio/main/internal/player.cpp b/framework/audio/main/internal/player.cpp index a1fd8b489a..ec6752674c 100644 --- a/framework/audio/main/internal/player.cpp +++ b/framework/audio/main/internal/player.cpp @@ -31,11 +31,6 @@ using namespace muse::async; using namespace muse::audio; using namespace muse::audio::rpc; -Player::~Player() -{ - cancelPendingResponses(); -} - rpc::CtxId Player::ctxId() const { return rpc::ctxId(iocContext()); @@ -53,11 +48,8 @@ void Player::init() }); Msg msg = rpc::make_request(ctxId(), MsgCode::GetPlaybackStatus); - m_playbackStatusCallId = msg.callId; channel()->send(msg, [this](const Msg& res) { ONLY_AUDIO_MAIN_THREAD; - m_playbackStatusCallId = 0; - Ret ret; PlaybackStatus status = PlaybackStatus::Stopped; StreamId streamId = 0; @@ -66,8 +58,7 @@ void Player::init() } if (ret) { - m_playbackStatusStreamId = streamId; - channel()->addReceiveStream(StreamName::PlaybackStatusStream, streamId, m_playbackStatusChanged); + channel()->addReceiveStream(ctxId(), StreamName::PlaybackStatusStream, streamId, m_playbackStatusChanged); //! NOTE Send initial state m_playbackStatusChanged.send(status); } else { @@ -82,11 +73,8 @@ void Player::init() }); Msg msg = rpc::make_request(ctxId(), MsgCode::GetPlaybackPosition); - m_playbackPositionCallId = msg.callId; channel()->send(msg, [this](const Msg& res) { ONLY_AUDIO_MAIN_THREAD; - m_playbackPositionCallId = 0; - Ret ret; secs_t pos = 0.0; StreamId streamId = 0; @@ -95,8 +83,7 @@ void Player::init() } if (ret) { - m_playbackPositionStreamId = streamId; - channel()->addReceiveStream(StreamName::PlaybackPositionStream, streamId, m_playbackPositionChanged); + channel()->addReceiveStream(ctxId(), StreamName::PlaybackPositionStream, streamId, m_playbackPositionChanged); //! NOTE Send initial state m_playbackPositionChanged.send(pos); } else { @@ -110,35 +97,10 @@ void Player::deinit() { ONLY_AUDIO_MAIN_THREAD; - cancelPendingResponses(); - - if (m_playbackStatusStreamId != 0) { - channel()->removeStream(m_playbackStatusStreamId); - m_playbackStatusStreamId = 0; - } - - if (m_playbackPositionStreamId != 0) { - channel()->removeStream(m_playbackPositionStreamId); - m_playbackPositionStreamId = 0; - } - m_playbackStatusChanged.disconnect(this); m_playbackPositionChanged.disconnect(this); } -void Player::cancelPendingResponses() -{ - if (m_playbackStatusCallId != 0) { - channel()->cancelResponse(m_playbackStatusCallId); - m_playbackStatusCallId = 0; - } - - if (m_playbackPositionCallId != 0) { - channel()->cancelResponse(m_playbackPositionCallId); - m_playbackPositionCallId = 0; - } -} - async::Promise Player::prepareToPlay() { ONLY_AUDIO_MAIN_THREAD; diff --git a/framework/audio/main/internal/player.h b/framework/audio/main/internal/player.h index 8683929501..5a39dad632 100644 --- a/framework/audio/main/internal/player.h +++ b/framework/audio/main/internal/player.h @@ -38,7 +38,6 @@ class Player : public IPlayer, public Contextable, public async::Asyncable public: Player(const muse::modularity::ContextPtr& ctx) : Contextable(ctx) {} - ~Player() override; void init(); void deinit(); @@ -70,15 +69,6 @@ class Player : public IPlayer, public Contextable, public async::Asyncable secs_t m_playbackPosition = 0.0; async::Channel m_playbackPositionChanged; - - void cancelPendingResponses(); - - rpc::StreamId m_playbackStatusStreamId = 0; - rpc::StreamId m_playbackPositionStreamId = 0; - - //! NOTE Pending requests, their responses are dropped on deinit - rpc::CallId m_playbackStatusCallId = 0; - rpc::CallId m_playbackPositionCallId = 0; }; } diff --git a/framework/audio/tests/enginerpccontroller_tests.cpp b/framework/audio/tests/enginerpccontroller_tests.cpp index 455d16052a..16421f71aa 100644 --- a/framework/audio/tests/enginerpccontroller_tests.cpp +++ b/framework/audio/tests/enginerpccontroller_tests.cpp @@ -117,6 +117,8 @@ class Audio_EngineRpcControllerTests : public ::testing::Test StreamId streamId = 0; EXPECT_CALL(*m_rpcChannel, addStream(_)) .WillOnce(Invoke([&streamId](std::shared_ptr s) { + //! NOTE The stream must belong to the context to be removed with it + EXPECT_EQ(s->ctxId(), CTX_ID); streamId = s->streamId(); })); @@ -137,43 +139,38 @@ class Audio_EngineRpcControllerTests : public ::testing::Test async::Channel m_trackRemoved; }; -TEST_F(Audio_EngineRpcControllerTests, ContextDeinit_RemovesPlayerStreams) +TEST_F(Audio_EngineRpcControllerTests, ContextDeinit_CancelsAllRequestsOfContext) { // [GIVEN] The context is inited request(GLOBAL_CTX_ID, MsgCode::ContextInit, RpcPacker::pack(CTX_ID)); // [GIVEN] The player has requested the status and the position - const StreamId statusStreamId = requestStream(MsgCode::GetPlaybackStatus); - const StreamId positionStreamId = requestStream(MsgCode::GetPlaybackPosition); - ASSERT_NE(statusStreamId, 0); - ASSERT_NE(positionStreamId, 0); + requestStream(MsgCode::GetPlaybackStatus); + requestStream(MsgCode::GetPlaybackPosition); - // [THEN] Both streams are removed - EXPECT_CALL(*m_rpcChannel, removeStream(statusStreamId)); - EXPECT_CALL(*m_rpcChannel, removeStream(positionStreamId)); + // [THEN] The pending responses and the streams of the context are removed by the channel + EXPECT_CALL(*m_rpcChannel, cancelAllRequests(CTX_ID)); + EXPECT_CALL(*m_rpcChannel, removeStream(_)).Times(0); // [WHEN] Deinit the context request(GLOBAL_CTX_ID, MsgCode::ContextDeinit, RpcPacker::pack(CTX_ID)); } -TEST_F(Audio_EngineRpcControllerTests, ContextDeinit_StreamsAreRemovedOnlyOnce) +TEST_F(Audio_EngineRpcControllerTests, ContextDeinit_ForgetsTrackStreams) { - // [GIVEN] The context is inited and the player has requested the status + // [GIVEN] The context is inited and the stream of the track is requested request(GLOBAL_CTX_ID, MsgCode::ContextInit, RpcPacker::pack(CTX_ID)); - const StreamId statusStreamId = requestStream(MsgCode::GetPlaybackStatus); + requestStream(MsgCode::GetSignalChanges, RpcPacker::pack(TRACK_ID)); // [GIVEN] The context is deinited and inited again - EXPECT_CALL(*m_rpcChannel, removeStream(statusStreamId)).Times(1); request(GLOBAL_CTX_ID, MsgCode::ContextDeinit, RpcPacker::pack(CTX_ID)); - ::testing::Mock::VerifyAndClearExpectations(m_rpcChannel.get()); - request(GLOBAL_CTX_ID, MsgCode::ContextInit, RpcPacker::pack(CTX_ID)); - // [THEN] The streams of the previous init are not removed again + // [THEN] The stream of the previous init is not removed again EXPECT_CALL(*m_rpcChannel, removeStream(_)).Times(0); - // [WHEN] Deinit the context again - request(GLOBAL_CTX_ID, MsgCode::ContextDeinit, RpcPacker::pack(CTX_ID)); + // [WHEN] The track is removed + m_trackRemoved.send(TRACK_ID); } TEST_F(Audio_EngineRpcControllerTests, TrackRemoved_RemovesTrackStreams) @@ -197,9 +194,9 @@ TEST_F(Audio_EngineRpcControllerTests, TrackRemoved_RemovesTrackStreams) m_trackRemoved.send(TRACK_ID); ::testing::Mock::VerifyAndClearExpectations(m_rpcChannel.get()); - // [THEN] The streams of the other track are removed on the context deinit - EXPECT_CALL(*m_rpcChannel, removeStream(otherTrackStreamId)); - EXPECT_CALL(*m_rpcChannel, removeStream(signalStreamId)).Times(0); + // [THEN] The streams of the other track are removed on the context deinit by the channel with all streams of the context + EXPECT_CALL(*m_rpcChannel, cancelAllRequests(CTX_ID)); + EXPECT_CALL(*m_rpcChannel, removeStream(_)).Times(0); // [WHEN] Deinit the context request(GLOBAL_CTX_ID, MsgCode::ContextDeinit, RpcPacker::pack(CTX_ID)); diff --git a/framework/audio/tests/playback_tests.cpp b/framework/audio/tests/playback_tests.cpp index 7053b40420..3f47e34525 100644 --- a/framework/audio/tests/playback_tests.cpp +++ b/framework/audio/tests/playback_tests.cpp @@ -25,10 +25,8 @@ #include #include -#include "async/asyncable.h" #include "async/processevents.h" -#include "audio/common/audioerrors.h" #include "audio/common/rpc/rpcpacker.h" #include "audio/main/internal/playback.h" @@ -60,7 +58,7 @@ MATCHER_P(StreamIdIs, id, "") } namespace muse::audio { -class Audio_PlaybackTests : public ::testing::Test, public async::Asyncable +class Audio_PlaybackTests : public ::testing::Test { protected: struct Request { @@ -83,11 +81,11 @@ class Audio_PlaybackTests : public ::testing::Test, public async::Asyncable m_requests.push_back({ msg, onResponse }); })); - //! NOTE As the channel, the response of the canceled request is dropped - ON_CALL(*m_rpcChannel, cancelResponse(_)) - .WillByDefault(Invoke([this](CallId callId) { - std::erase_if(m_requests, [callId](const Request& r) { - return r.msg.callId == callId; + //! NOTE As the channel, the responses of the canceled requests are dropped + ON_CALL(*m_rpcChannel, cancelAllRequests(_)) + .WillByDefault(Invoke([this](CtxId ctxId) { + std::erase_if(m_requests, [ctxId](const Request& r) { + return r.msg.ctxId == ctxId; }); })); @@ -208,20 +206,20 @@ TEST_F(Audio_PlaybackTests, Player_SubscribesToStatusAndPosition) EXPECT_EQ(player->playbackPosition(), secs_t(2.0)); } -TEST_F(Audio_PlaybackTests, Deinit_RemovesPlayerStreams) +TEST_F(Audio_PlaybackTests, Deinit_CancelsAllRequestsOfContext) { // [GIVEN] The player is subscribed to the streams initContext(); IPlayerPtr player = m_playback->player(); respondPlayerStreams(); - // [THEN] The streams are removed - EXPECT_CALL(*m_rpcChannel, removeStream(STATUS_STREAM_ID)); - EXPECT_CALL(*m_rpcChannel, removeStream(POSITION_STREAM_ID)); + // [THEN] The pending responses and the streams of the context are removed by the channel + EXPECT_CALL(*m_rpcChannel, cancelAllRequests(rpc::ctxId(m_ctx))); // [WHEN] Deinit the playback m_playback->deinit(); EXPECT_FALSE(m_playback->isInited()); + ::testing::Mock::VerifyAndClearExpectations(m_rpcChannel.get()); // [WHEN] Init it again initContext(); @@ -239,10 +237,11 @@ TEST_F(Audio_PlaybackTests, Deinit_CancelsPlayerPendingRequests) IPlayerPtr player = m_playback->player(); // [THEN] The responses are canceled - EXPECT_CALL(*m_rpcChannel, cancelResponse(_)).Times(2); + EXPECT_CALL(*m_rpcChannel, cancelAllRequests(rpc::ctxId(m_ctx))); // [WHEN] Deinit the playback m_playback->deinit(); + ::testing::Mock::VerifyAndClearExpectations(m_rpcChannel.get()); // [THEN] No responses are expected anymore EXPECT_EQ(requestCount(MsgCode::GetPlaybackStatus), 0); @@ -264,67 +263,12 @@ TEST_F(Audio_PlaybackTests, TrackRemoved_RemovesTrackStreams) notify(MsgCode::TrackRemoved, RpcPacker::pack(TRACK_ID)); ::testing::Mock::VerifyAndClearExpectations(m_rpcChannel.get()); - // [THEN] The stream of the other track is removed on deinit - EXPECT_CALL(*m_rpcChannel, removeStream(OTHER_SIGNAL_STREAM_ID)); - EXPECT_CALL(*m_rpcChannel, removeStream(SIGNAL_STREAM_ID)).Times(0); + // [THEN] The stream of the other track is removed on deinit by the channel with all streams of the context + EXPECT_CALL(*m_rpcChannel, cancelAllRequests(rpc::ctxId(m_ctx))); + EXPECT_CALL(*m_rpcChannel, removeStream(_)).Times(0); // [WHEN] Deinit the playback m_playback->deinit(); -} - -TEST_F(Audio_PlaybackTests, TrackRemoved_CancelsPendingRequests) -{ - // [GIVEN] The signal changes of two tracks are requested, but the engine has not responded yet - initContext(); - - bool rejected = false; - m_playback->signalChanges(TRACK_ID).onReject(this, [&rejected](int code, const std::string&) { - rejected = code == static_cast(Err::InvalidTrackId); - }); - m_playback->signalChanges(OTHER_TRACK_ID); - async::processMessages(); - - // [WHEN] The track is removed - notify(MsgCode::TrackRemoved, RpcPacker::pack(TRACK_ID)); - async::processMessages(); - - // [THEN] The request of the removed track is canceled and rejected - EXPECT_TRUE(rejected); - - // [THEN] The request of the other track is still pending - EXPECT_EQ(requestCount(MsgCode::GetSignalChanges), 1); -} - -TEST_F(Audio_PlaybackTests, Deinit_CancelsPendingTrackRequests) -{ - // [GIVEN] The signal changes are requested, but the engine has not responded yet - initContext(); - - bool rejected = false; - m_playback->signalChanges(TRACK_ID).onReject(this, [&rejected](int code, const std::string&) { - rejected = code == static_cast(Err::InvalidTrackId); - }); - async::processMessages(); - - // [WHEN] Deinit the playback - m_playback->deinit(); - async::processMessages(); - - // [THEN] The request is canceled and rejected - EXPECT_TRUE(rejected); - EXPECT_EQ(requestCount(MsgCode::GetSignalChanges), 0); -} - -TEST_F(Audio_PlaybackTests, Response_RemovesPendingRequest) -{ - // [GIVEN] The signal changes are requested - initContext(); - requestSignalChanges(TRACK_ID, SIGNAL_STREAM_ID); - - // [THEN] The completed request is not canceled - EXPECT_CALL(*m_rpcChannel, cancelResponse(_)).Times(0); - - // [WHEN] The track is removed - notify(MsgCode::TrackRemoved, RpcPacker::pack(TRACK_ID)); + ::testing::Mock::VerifyAndClearExpectations(m_rpcChannel.get()); } }