Skip to content

Cut thread hops from negotiation and getStats, share concurrent getStats calls #255

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
MarshalX merged 1 commit into from
Oct 16, 2026
Merged

Cut thread hops from negotiation and getStats, share concurrent getStats calls #255

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions python-webrtc/cpp/src/interfaces/rtc_peer_connection/create_session_description_observer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,11 +13,14 @@ namespace python_webrtc {

void CreateSessionDescriptionObserver::OnSuccess(webrtc::SessionDescriptionInterface *description) {
const std::unique_ptr<webrtc::SessionDescriptionInterface> owned(description);
if (auto peerConnection = _peerConnection.lock()) {
auto peerConnection = _peerConnection.lock();
if (peerConnection) {
peerConnection->PrepareCreatedDescription(*owned);
peerConnection->SaveCreatedDescription(RTCSessionDescriptionInit::Wrap(owned.get()));
}
auto created = RTCSessionDescription::Wrap(owned.get());
if (peerConnection) {
peerConnection->SaveCreatedDescription(created.init());
}
if (_created) {
_created(created);
} else {
Expand Down
22 changes: 17 additions & 5 deletions python-webrtc/cpp/src/interfaces/rtc_peer_connection/descriptions.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@

#include <algorithm>
#include <map>
#include <tuple>
#include <utility>

#include <api/rtp_parameters.h>
Expand Down Expand Up @@ -288,8 +289,9 @@ namespace python_webrtc {
}
// the descriptions as the operation left them, before candidates gathered later
self->_completionSnapshot = self->SnapshotDescriptions();
auto created = self->WrapTransports();
self->UpdateMaxMessageSize();
const auto sctp = self->SctpTransportsOf(self->connection());
auto created = self->WrapTransports(sctp);
self->UpdateMaxMessageSize(sctp);
self->MarkIceRolesKnown();
if (remote) {
self->RecordRemoteDescriptionCandidates();
Expand Down Expand Up @@ -597,10 +599,20 @@ namespace python_webrtc {
return 0;
}
DescriptionsSnapshot snapshot;
for (size_t kind = 0; kind < kDescriptionKinds; ++kind) {
snapshot.kinds[kind] = ReadDescription(pc, static_cast<DescriptionKind>(kind));
auto view = [&snapshot](DescriptionKind kind) -> DescriptionView & {
return snapshot.kinds[static_cast<size_t>(kind)];
};
for (auto kind : {DescriptionKind::kCurrentLocal, DescriptionKind::kCurrentRemote, DescriptionKind::kPendingLocal,
DescriptionKind::kPendingRemote}) {
view(kind) = ReadDescription(pc, kind);
snapshot.live.push_back(view(kind).description);
}
// the pending description if there's one, else the current one (webrtc-pc)
for (auto [kind, pending, current] :
{std::tuple(DescriptionKind::kLocal, DescriptionKind::kPendingLocal, DescriptionKind::kCurrentLocal),
std::tuple(DescriptionKind::kRemote, DescriptionKind::kPendingRemote, DescriptionKind::kCurrentRemote)}) {
view(kind) = view(pending).description != nullptr ? view(pending) : view(current);
}
snapshot.live = liveDescriptions(pc);
const std::scoped_lock lock(_descriptionsMutex);
auto id = ++_lastSnapshot;
_snapshots.emplace(id, std::move(snapshot));
Expand Down
111 changes: 81 additions & 30 deletions python-webrtc/cpp/src/interfaces/rtc_peer_connection/ice.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,24 +71,46 @@ namespace python_webrtc {

namespace {

// the DTLS transports of the media sections and of the data channels, some of them shared (bundled)
std::vector<webrtc::scoped_refptr<webrtc::DtlsTransportInterface>>
dtlsTransports(const webrtc::scoped_refptr<webrtc::PeerConnectionInterface> &pc) {
std::vector<webrtc::scoped_refptr<webrtc::DtlsTransportInterface>> transports;
using DtlsTransport = webrtc::scoped_refptr<webrtc::DtlsTransportInterface>;
using Transceivers = std::vector<webrtc::scoped_refptr<webrtc::RtpTransceiverInterface>>;

std::vector<DtlsTransport> dtlsTransports(const webrtc::scoped_refptr<webrtc::PeerConnectionInterface> &pc,
const DtlsTransport &sctp) {
std::vector<DtlsTransport> transports;
for (const auto &transceiver : pc->GetTransceivers()) {
transports.push_back(transceiver->sender()->dtls_transport());
transports.push_back(transceiver->receiver()->dtls_transport());
}
if (auto sctp = pc->GetSctpTransport()) {
transports.push_back(sctp->dtls_transport());
}
transports.push_back(sctp);
std::erase(transports, nullptr);
return transports;
}

DtlsTransport dtlsTransportOf(const webrtc::scoped_refptr<webrtc::PeerConnectionInterface> &pc,
const Transceivers &transceivers, const DtlsTransport &sctp,
const webrtc::ContentInfo &content) {
if (content.rejected) {
return nullptr;
}
DtlsTransport known;
if (content.type == webrtc::MediaProtocolType::kSctp) {
known = sctp;
} else if (auto transceiver = std::ranges::find_if(
transceivers, [&](const auto &candidate) { return candidate->mid() == content.mid(); });
transceiver != transceivers.end()) {
if ((*transceiver)->stopped()) {
return nullptr;
}
known = (*transceiver)->sender()->dtls_transport();
}
// no transceiver, or a rollback took its transport: the lookup waits for the network thread
return known ? known : pc->LookupDtlsTransportByMid(content.mid());
}

std::map<std::string, std::shared_ptr<RTCIceTransport>>
iceTransportsByMid(const webrtc::scoped_refptr<webrtc::PeerConnectionInterface> &pc) {
iceTransportsByMid(const webrtc::scoped_refptr<webrtc::PeerConnectionInterface> &pc, const DtlsTransport &sctp) {
std::map<std::string, std::shared_ptr<RTCIceTransport>> byMid;
auto transceivers = pc->GetTransceivers();
for (const auto *description : {pc->local_description(), pc->remote_description()}) {
if (description == nullptr) {
continue;
Expand All @@ -97,7 +119,7 @@ namespace python_webrtc {
if (byMid.contains(content.mid())) {
continue;
}
auto dtls = pc->LookupDtlsTransportByMid(content.mid());
auto dtls = dtlsTransportOf(pc, transceivers, sctp, content);
if (auto wrapper = dtls ? RTCDtlsTransport::registry().Find(dtls.get()) : nullptr) {
byMid.emplace(content.mid(), wrapper->GetIceTransport());
}
Expand All @@ -106,22 +128,47 @@ namespace python_webrtc {
return byMid;
}

std::vector<std::shared_ptr<RTCIceTransport>>
iceTransportsOf(const std::vector<std::shared_ptr<RTCDtlsTransport>> &dtlsTransports) {
std::vector<std::shared_ptr<RTCIceTransport>> iceTransports;
for (const auto &dtls : dtlsTransports) {
auto ice = dtls ? dtls->GetIceTransport() : nullptr;
if (ice && std::ranges::find(iceTransports, ice) == iceTransports.end()) {
iceTransports.push_back(std::move(ice));
}
}
return iceTransports;
}

} // namespace

RTCPeerConnection::SctpTransports
RTCPeerConnection::SctpTransportsOf(const webrtc::scoped_refptr<webrtc::PeerConnectionInterface> &pc) {
SctpTransports transports;
if (!pc) {
return transports;
}
// the connection's getter waits for the network thread too, where the transport is read
BlockingCallOn(_factory->workerThread(), [&]() {
transports.sctp = pc->GetSctpTransport();
transports.dtls = transports.sctp ? transports.sctp->dtls_transport() : nullptr;
});
return transports;
}

std::vector<std::shared_ptr<RTCIceTransport>> RTCPeerConnection::IceTransports() {
std::vector<std::shared_ptr<RTCIceTransport>> iceTransports;
auto pc = connection();
if (!pc) {
return iceTransports;
return {};
}
for (const auto &transport : dtlsTransports(pc)) {
auto dtls = RTCDtlsTransport::registry().Find(transport.get());
auto ice = dtls ? dtls->GetIceTransport() : nullptr;
if (ice && std::ranges::find(iceTransports, ice) == iceTransports.end()) {
iceTransports.push_back(ice);
}
auto sctp = SctpTransportsOf(pc).dtls;
auto transports = BlockingCallOn(_factory->signalingThread(), [&]() { return dtlsTransports(pc, sctp); });
std::vector<std::shared_ptr<RTCDtlsTransport>> wrappers;
wrappers.reserve(transports.size());
for (const auto &transport : transports) {
wrappers.push_back(RTCDtlsTransport::registry().Find(transport.get()));
}
return iceTransports;
return iceTransportsOf(wrappers);
}

std::shared_ptr<RTCIceTransport> RTCPeerConnection::IceTransportByMid(const std::string &mid) {
Expand All @@ -137,14 +184,14 @@ namespace python_webrtc {
return wrapper ? wrapper->GetIceTransport() : nullptr;
}

std::vector<std::shared_ptr<RTCIceTransport>> RTCPeerConnection::WrapTransports() {
std::vector<std::shared_ptr<RTCIceTransport>> RTCPeerConnection::WrapTransports(const SctpTransports &sctp) {
std::vector<std::shared_ptr<RTCIceTransport>> created;
auto pc = connection();
if (!pc) {
return created;
}
std::vector<std::shared_ptr<RTCDtlsTransport>> wrappers;
for (const auto &transport : dtlsTransports(pc)) {
for (const auto &transport : dtlsTransports(pc, sctp.dtls)) {
const bool existed = RTCDtlsTransport::registry().Find(transport.get()) != nullptr;
auto wrapper = RTCDtlsTransport::registry().GetOrCreate(_factory, transport);
Adopt(wrapper);
Expand All @@ -166,7 +213,7 @@ namespace python_webrtc {
}
_negotiatedTransports.assign(wrappers.begin(), wrappers.end());
}
auto byMid = iceTransportsByMid(pc);
auto byMid = iceTransportsByMid(pc, sctp.dtls);
{
const TrackedLock lock(_wrappersMutex);
std::swap(_dtlsTransports, wrappers);
Expand All @@ -193,14 +240,11 @@ namespace python_webrtc {
return;
}

std::vector<webrtc::scoped_refptr<webrtc::DtlsTransportInterface>> transports;
webrtc::scoped_refptr<webrtc::SctpTransportInterface> sctpTransport;
BlockingCallOn(_factory->signalingThread(), [&]() {
transports = dtlsTransports(pc);
sctpTransport = pc->GetSctpTransport();
});
auto sctpTransports = SctpTransportsOf(pc);
auto transports =
BlockingCallOn(_factory->signalingThread(), [&]() { return dtlsTransports(pc, sctpTransports.dtls); });

if (auto sctp = sctpTransport ? RTCSctpTransport::registry().Find(sctpTransport.get()) : nullptr) {
if (auto sctp = sctpTransports.sctp ? RTCSctpTransport::registry().Find(sctpTransports.sctp.get()) : nullptr) {
sctp->OnPeerConnectionClosed();
}
for (const auto &transport : transports) {
Expand All @@ -219,7 +263,12 @@ namespace python_webrtc {
auto state = pc->signaling_state();
if (state == SignalingState::kStable || state == SignalingState::kHaveLocalPrAnswer ||
state == SignalingState::kHaveRemotePrAnswer) {
for (const auto &iceTransport : IceTransports()) {
std::vector<std::shared_ptr<RTCDtlsTransport>> wrapped;
{
const TrackedLock lock(_wrappersMutex);
wrapped = _dtlsTransports;
}
for (const auto &iceTransport : iceTransportsOf(wrapped)) {
iceTransport->SetRoleKnown();
}
}
Expand Down Expand Up @@ -269,13 +318,15 @@ namespace python_webrtc {
if (!pc) {
return {};
}
auto sctp = SctpTransportsOf(pc).dtls;
return BlockingCallOn(_factory->signalingThread(), [&]() -> std::optional<std::pair<std::string, std::string>> {
const auto *description = local ? pc->local_description() : pc->remote_description();
if (!description) {
return std::nullopt;
}
auto transceivers = pc->GetTransceivers();
for (const auto &content : description->description()->contents()) {
auto dtls = pc->LookupDtlsTransportByMid(content.mid());
auto dtls = dtlsTransportOf(pc, transceivers, sctp, content);
const auto *info = description->description()->GetTransportInfoByName(content.mid());
if (dtls && dtls->ice_transport().get() == iceTransport && info) {
return std::make_pair(info->description.ice_ufrag, info->description.ice_pwd);
Expand Down
27 changes: 10 additions & 17 deletions python-webrtc/cpp/src/interfaces/rtc_peer_connection/rtc_peer_connection.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -463,7 +463,6 @@ namespace python_webrtc {
throw RTCException(result.error());
}

QueueNegotiationNeeded();
return Wrap(_senders, result.value());
}

Expand Down Expand Up @@ -503,7 +502,6 @@ namespace python_webrtc {
if (!error.ok()) {
throw RTCException(error);
}
QueueNegotiationNeeded();
}

std::shared_ptr<RTCRtpTransceiver>
Expand Down Expand Up @@ -532,7 +530,6 @@ namespace python_webrtc {
if (!result.ok()) {
throw RTCException(result.error());
}
QueueNegotiationNeeded();
return Wrap(_transceivers, result.value());
}

Expand Down Expand Up @@ -588,7 +585,6 @@ namespace python_webrtc {
: webrtc::RTCErrorType::UNSUPPORTED_OPERATION,
error.message());
}
QueueNegotiationNeeded();
return Wrap(_channels, result.MoveValue());
}

Expand Down Expand Up @@ -660,15 +656,13 @@ namespace python_webrtc {
});
}

void RTCPeerConnection::UpdateMaxMessageSize() {
void RTCPeerConnection::UpdateMaxMessageSize(const SctpTransports &sctp) {
auto pc = connection();
if (!pc) {
return;
}
// read by send() without waiting for the signaling thread
_maxMessageSize = BlockingCallOn(_factory->signalingThread(), [&pc]() {
return pc->GetSctpTransport() ? NegotiatedMaxMessageSize(*pc) : std::numeric_limits<double>::quiet_NaN();
});
_maxMessageSize = sctp.sctp ? NegotiatedMaxMessageSize(*pc) : std::numeric_limits<double>::quiet_NaN();
}

ConfigurationInit RTCPeerConnection::GetConfiguration() {
Expand Down Expand Up @@ -705,17 +699,16 @@ namespace python_webrtc {

void RTCPeerConnection::RestartIce() {
auto pc = connection();
// before the first local description, there are no credentials to replace: the first offer has new ones
if (pc && (pc->local_description() != nullptr)) {
pc->RestartIce();
QueueNegotiationNeeded();
if (!pc) {
return;
}
}

void RTCPeerConnection::QueueNegotiationNeeded() {
// libwebrtc posts the negotiationneeded event of a change to the signaling thread: once the thread ran it,
// the event is queued during the call
BlockingCallOn(_factory->signalingThread(), []() {});
BlockingCallOn(_factory->signalingThread(), [&pc]() {
// before the first local description, there are no credentials to replace: the first offer has new ones
if (pc->local_description() != nullptr) {
pc->RestartIce();
}
});
}

bool RTCPeerConnection::ShouldFireNegotiationNeededEvent(uint32_t eventId) {
Expand Down
20 changes: 12 additions & 8 deletions python-webrtc/cpp/src/interfaces/rtc_peer_connection/rtc_peer_connection.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -264,14 +264,18 @@ namespace python_webrtc {
template <typename T, typename U>
using Wrappers = std::unordered_map<U *, std::shared_ptr<T>>;

struct SctpTransports {
webrtc::scoped_refptr<webrtc::SctpTransportInterface> sctp;
webrtc::scoped_refptr<webrtc::DtlsTransportInterface> dtls;
};

// Python threads may call close() concurrently with other methods (the GIL is released)
webrtc::scoped_refptr<webrtc::PeerConnectionInterface> connection();

// a closed connection still has stats and transceivers
webrtc::scoped_refptr<webrtc::PeerConnectionInterface> closedConnection();

// The connection to collect the stats of, even closed, with its stats cache cleared: libwebrtc reuses a report
// for 50 ms, the stats are the current ones. None fails the request.
// the connection to collect the stats of, even closed; none fails the request
webrtc::scoped_refptr<webrtc::PeerConnectionInterface> StatsConnection(Completion &completion);

// the wrapper of a libwebrtc object of this connection, which the connection keeps while it's open
Expand Down Expand Up @@ -309,11 +313,8 @@ namespace python_webrtc {

// the SCTP transport's maxMessageSize, cached in kept while the connection still uses that transport
std::optional<double> MaxMessageSize(const webrtc::SctpTransportInterface *transport, std::optional<double> &kept);
// the spec's "update the data max message size", for a transport Python may hold
void UpdateMaxMessageSize();

// waits for the negotiationneeded event a change may have caused to be emitted
void QueueNegotiationNeeded();
// the spec's "update the data max message size", for a transport Python may hold; on the signaling thread
void UpdateMaxMessageSize(const SctpTransports &sctp);

std::shared_ptr<RTCRtpTransceiver>
AddedTransceiver(webrtc::RTCErrorOr<webrtc::scoped_refptr<webrtc::RtpTransceiverInterface>> result);
Expand Down Expand Up @@ -351,6 +352,9 @@ namespace python_webrtc {
// On the signaling thread.
uint64_t SnapshotDescriptions();

// read in one wait for the network thread, which libwebrtc's getter waits for anyway
SctpTransports SctpTransportsOf(const webrtc::scoped_refptr<webrtc::PeerConnectionInterface> &pc);

// the ICE transports of the connection that Python has wrappers of
std::vector<std::shared_ptr<RTCIceTransport>> IceTransports();

Expand All @@ -359,7 +363,7 @@ namespace python_webrtc {

// Wraps the transports a description creates, so their states are followed from the start. Returns the new
// ICE transports, which get their gathering event once the description is set. On the signaling thread.
std::vector<std::shared_ptr<RTCIceTransport>> WrapTransports();
std::vector<std::shared_ptr<RTCIceTransport>> WrapTransports(const SctpTransports &sctp);

// the transports of a closed connection fire no events
void UnbindTransports(const webrtc::scoped_refptr<webrtc::PeerConnectionInterface> &pc);
Expand Down
Loading
Loading