From eb435d3e7bb0560680965df575a706912fdcadc3 Mon Sep 17 00:00:00 2001 From: arch7tect Date: Sat, 3 Oct 2026 17:16:24 +0300 Subject: [PATCH 1/2] pjsua2: Add an explicit AudioMediaPort callback fence --- pjsip/CMakeLists.txt | 1 + pjsip/build/Makefile | 2 +- pjsip/include/pjsua2/media.hpp | 25 +++ pjsip/src/pjsua2-test/audio_media_port.cpp | 172 +++++++++++++++++++++ pjsip/src/pjsua2-test/main.cpp | 4 + pjsip/src/pjsua2/media.cpp | 24 +-- 6 files changed, 215 insertions(+), 13 deletions(-) create mode 100644 pjsip/src/pjsua2-test/audio_media_port.cpp diff --git a/pjsip/CMakeLists.txt b/pjsip/CMakeLists.txt index d9a64a4da4..aaa6c49993 100644 --- a/pjsip/CMakeLists.txt +++ b/pjsip/CMakeLists.txt @@ -389,6 +389,7 @@ if(BUILD_TESTING) src/pjsua2-test/main.cpp src/pjsua2-test/instant_messaging.cpp src/pjsua2-test/auth_challenge.cpp + src/pjsua2-test/audio_media_port.cpp ) target_link_libraries(pjsua2-test diff --git a/pjsip/build/Makefile b/pjsip/build/Makefile index 2bc7ebdde6..223f7422a4 100644 --- a/pjsip/build/Makefile +++ b/pjsip/build/Makefile @@ -185,7 +185,7 @@ endif # export PJSUA2_TEST_SRCDIR = ../src/pjsua2-test export PJSUA2_TEST_OBJS += $(OS_OBJS) $(M_OBJS) $(CC_OBJS) $(HOST_OBJS) \ - main.o instant_messaging.o auth_challenge.o + main.o instant_messaging.o auth_challenge.o audio_media_port.o export PJSUA2_TEST_CFLAGS += $(_CFLAGS) $(PJ_VIDEO_CFLAGS) export PJSUA2_TEST_CXXFLAGS = $(_CXXFLAGS) $(PJSUA2_LIB_CFLAGS) $(PJ_VIDEO_CFLAGS) export PJSUA2_TEST_LDFLAGS += $(PJ_LDXXFLAGS) $(PJ_LDXXLIBS) $(LDFLAGS) diff --git a/pjsip/include/pjsua2/media.hpp b/pjsip/include/pjsua2/media.hpp index a16ccb201d..34bb64efc2 100644 --- a/pjsip/include/pjsua2/media.hpp +++ b/pjsip/include/pjsua2/media.hpp @@ -551,7 +551,32 @@ class AudioMediaPort : public AudioMedia virtual void onFrameReceived(MediaFrame &frame) { PJ_UNUSED_ARG(frame); } +protected: + /** + * Permanently stop dispatching frame callbacks to this object from the + * created media port. Repeated calls are safe. If no port has been + * created, this does nothing. + * + * This acquires the port's recursive group lock. When called from a + * thread/context which does not already own that lock, it waits for + * in-progress callbacks to finish. Calling it from a callback (or while + * already owning the lock) does not wait for that callback to return; + * it must not be used as a way to wait for oneself. + * + * Call while the object and all callback-visible state are still fully + * valid, before tearing down derived members. In particular, with + * multiple levels of inheritance, explicitly detach before destruction + * begins. This is not a general guarantee of safe concurrent destruction. + * Do not hold application locks, other ports' locks, or library locks + * needed by an in-progress callback while calling this method. + * + * This does not unregister the conference port or destroy media resources. + */ + void detachCallbacks(); + private: + /* Test access to raw frame dispatch without exposing media resources. */ + friend class AudioMediaPortTest; pj_pool_t *pool; pjmedia_port *port; }; diff --git a/pjsip/src/pjsua2-test/audio_media_port.cpp b/pjsip/src/pjsua2-test/audio_media_port.cpp new file mode 100644 index 0000000000..894401c0b5 --- /dev/null +++ b/pjsip/src/pjsua2-test/audio_media_port.cpp @@ -0,0 +1,172 @@ +/* + * Copyright (C) 2026 Teluu Inc. (http://www.teluu.com) + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version. + * + * 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, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + */ + +#include +#include +#include +#include +#include +#include +#include +#include + +/* Fail also in release builds, including stalled worker threads. */ +#define CHECK(expr) \ + do { \ + if (!(expr)) { \ + std::cerr << "AudioMediaPort: " << #expr << " at line " \ + << __LINE__ << std::endl; \ + std::abort(); \ + } \ + } while (0) + +namespace pj { + +class AudioMediaPortTest : public AudioMediaPort +{ +public: + AudioMediaPortTest() + : releaseFuture(release.get_future()), callbackFinished(false), + requested(0), received(0) + {} + + virtual void onFrameRequested(MediaFrame &frame) + { + ++requested; + holdCallback(); + frame.type = PJMEDIA_FRAME_TYPE_AUDIO; + frame.buf.assign(frame.size, 0x5a); + callbackFinished = true; + } + + virtual void onFrameReceived(MediaFrame &frame) + { + ++received; + CHECK(frame.type == PJMEDIA_FRAME_TYPE_AUDIO); + CHECK(frame.buf.size() == 16); + holdCallback(); + callbackFinished = true; + } + + static void run(Endpoint &ep, bool receive) + { + AudioMediaPortTest media; + media.detachCallbacks(); /* No port yet. */ + + MediaFormatAudio fmt; + fmt.init(PJMEDIA_FORMAT_L16, 8000, 1, 20000, 16); + media.createPort("callback-fence", fmt); + pjmedia_port *port = media.port; + + unsigned char buffer[16]; + pjmedia_frame frame; + pj_bzero(&frame, sizeof(frame)); + pj_memset(buffer, 0xa5, sizeof(buffer)); + frame.buf = buffer; + frame.type = PJMEDIA_FRAME_TYPE_AUDIO; + frame.size = sizeof(buffer); + + std::future entered = media.entered.get_future(); + std::thread callback([&]() { + ep.libRegisterThread("fence-callback"); + pj_status_t status = receive ? + pjmedia_port_put_frame(port, &frame) : + pjmedia_port_get_frame(port, &frame); + CHECK(status == PJ_SUCCESS); + }); + CHECK(entered.wait_for(std::chrono::seconds(5)) == + std::future_status::ready); + + /* Verify that the held callback really owns the fence's lock. */ + CHECK(pj_grp_lock_tryacquire(port->grp_lock) != PJ_SUCCESS); + + std::promise starting, completed; + std::future started = starting.get_future(); + std::future done = completed.get_future(); + std::thread detacher([&]() { + ep.libRegisterThread("fence-detach"); + starting.set_value(); + media.detachCallbacks(); + CHECK(media.callbackFinished.load()); + completed.set_value(); + }); + CHECK(started.wait_for(std::chrono::seconds(5)) == + std::future_status::ready); + CHECK(done.wait_for(std::chrono::milliseconds(100)) == + std::future_status::timeout); + + media.release.set_value(); + CHECK(done.wait_for(std::chrono::seconds(5)) == + std::future_status::ready); + detacher.join(); + callback.join(); + + if (!receive) { + CHECK(frame.type == PJMEDIA_FRAME_TYPE_AUDIO); + CHECK(frame.size == sizeof(buffer)); + CHECK(buffer[0] == 0x5a); + } + + media.detachCallbacks(); + /* Fencing must not unregister the port or release its resources. */ + CHECK(media.getPortInfo().portId == media.getPortId()); + for (unsigned i = 0; i < 3; ++i) { + frame.type = PJMEDIA_FRAME_TYPE_AUDIO; + frame.size = sizeof(buffer); + CHECK(pjmedia_port_get_frame(port, &frame) == PJ_SUCCESS); + + frame.type = PJMEDIA_FRAME_TYPE_AUDIO; + frame.size = sizeof(buffer); + CHECK(pjmedia_port_put_frame(port, &frame) == PJ_SUCCESS); + } + CHECK(media.requested == (receive ? 0u : 1u)); + CHECK(media.received == (receive ? 1u : 0u)); + } + +private: + void holdCallback() + { + CHECK(requested + received == 1); + entered.set_value(); + CHECK(releaseFuture.wait_for(std::chrono::seconds(5)) == + std::future_status::ready); + } + + std::promise entered, release; + std::future releaseFuture; + std::atomic callbackFinished; + unsigned requested, received; +}; + +} // namespace pj + +void audioMediaPortTest() +{ + pj::Endpoint ep; + ep.libCreate(); + pj::EpConfig cfg; + cfg.uaConfig.threadCnt = 0; + cfg.medConfig.threadCnt = 0; + cfg.logConfig.level = 2; + ep.libInit(cfg); + ep.audDevManager().setNoDev(); + pj::AudioMediaPortTest::run(ep, false); + pj::AudioMediaPortTest::run(ep, true); + ep.libDestroy(); + std::cout << "AudioMediaPort callback fence tests passed" << std::endl; +} diff --git a/pjsip/src/pjsua2-test/main.cpp b/pjsip/src/pjsua2-test/main.cpp index 67dd5d2b6d..df550c655f 100644 --- a/pjsip/src/pjsua2-test/main.cpp +++ b/pjsip/src/pjsua2-test/main.cpp @@ -22,9 +22,13 @@ using namespace pj; +void audioMediaPortTest(); + int main(int argc, char *argv[]) { try { + audioMediaPortTest(); + { InstantMessagingTests instantMessagingTests; diff --git a/pjsip/src/pjsua2/media.cpp b/pjsip/src/pjsua2/media.cpp index 441c5e2446..c05f38c0d9 100644 --- a/pjsip/src/pjsua2/media.cpp +++ b/pjsip/src/pjsua2/media.cpp @@ -295,19 +295,8 @@ AudioMediaPort::AudioMediaPort() AudioMediaPort::~AudioMediaPort() { PJSUA2_CATCH_IGNORE( unregisterMediaPort() ); + detachCallbacks(); if (port) { - struct port_data *pdata = static_cast - (port->port_data.pdata); - - /* Make sure port no longer accesses this object in its - * get/put_frame() callback. - */ - if (port->grp_lock) { - pj_grp_lock_acquire(port->grp_lock); - pdata->mport = NULL; - pj_grp_lock_release(port->grp_lock); - } - pjmedia_port_destroy(port); /* We release the pool later in port.on_destroy since * the unregistration is async and may not have completed yet. @@ -315,6 +304,17 @@ AudioMediaPort::~AudioMediaPort() } } +void AudioMediaPort::detachCallbacks() +{ + if (port && port->grp_lock) { + struct port_data *pdata = static_cast + (port->port_data.pdata); + pj_grp_lock_acquire(port->grp_lock); + pdata->mport = NULL; + pj_grp_lock_release(port->grp_lock); + } +} + static pj_status_t get_frame(pjmedia_port *port, pjmedia_frame *frame) { struct port_data *pdata = static_cast From 9b70899bf20a26268b9ceb5f1d0242693bfa5045 Mon Sep 17 00:00:00 2001 From: arch7tect Date: Sat, 3 Oct 2026 17:16:24 +0300 Subject: [PATCH 2/2] pjsua2: Return an empty frame after callback detachment --- pjsip/src/pjsua2-test/audio_media_port.cpp | 7 +++++++ pjsip/src/pjsua2/media.cpp | 5 ++++- 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/pjsip/src/pjsua2-test/audio_media_port.cpp b/pjsip/src/pjsua2-test/audio_media_port.cpp index 894401c0b5..fae6659b3e 100644 --- a/pjsip/src/pjsua2-test/audio_media_port.cpp +++ b/pjsip/src/pjsua2-test/audio_media_port.cpp @@ -126,9 +126,16 @@ class AudioMediaPortTest : public AudioMediaPort /* Fencing must not unregister the port or release its resources. */ CHECK(media.getPortInfo().portId == media.getPortId()); for (unsigned i = 0; i < 3; ++i) { + /* Seed a stale audio result; detached reads must not expose it. */ + pj_memset(buffer, 0xa5, sizeof(buffer)); frame.type = PJMEDIA_FRAME_TYPE_AUDIO; frame.size = sizeof(buffer); CHECK(pjmedia_port_get_frame(port, &frame) == PJ_SUCCESS); + CHECK(frame.type == PJMEDIA_FRAME_TYPE_NONE); + CHECK(frame.size == 0); + /* NONE/0 makes the payload invalid; it need not be overwritten. */ + for (unsigned j = 0; j < sizeof(buffer); ++j) + CHECK(buffer[j] == 0xa5); frame.type = PJMEDIA_FRAME_TYPE_AUDIO; frame.size = sizeof(buffer); diff --git a/pjsip/src/pjsua2/media.cpp b/pjsip/src/pjsua2/media.cpp index c05f38c0d9..e884be9538 100644 --- a/pjsip/src/pjsua2/media.cpp +++ b/pjsip/src/pjsua2/media.cpp @@ -323,8 +323,11 @@ static pj_status_t get_frame(pjmedia_port *port, pjmedia_frame *frame) MediaFrame frame_; pj_grp_lock_acquire(port->grp_lock); - if ((mport = pdata->mport) == NULL) + if ((mport = pdata->mport) == NULL) { + frame->type = PJMEDIA_FRAME_TYPE_NONE; + frame->size = 0; goto on_return; + } frame_.size = (unsigned)frame->size; mport->onFrameRequested(frame_);