zqz/platform-external-webrtc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Delete AsyncInvoker usage in DataChannelController

Browse Source 
Tasks access this via WeakPtrFactory.

Bug: webrtc:12339
Change-Id: I0aaeffd4bed59a6abfadf995286644c24c1fd716
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/212721
Reviewed-by: Harald Alvestrand <hta@webrtc.org>
Commit-Queue: Niels Moller <nisse@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#33560}
This commit is contained in:
Niels Möller
2021-03-23 09:23:10 +01:00
committed by Commit Bot
parent 4c555cca2d
commit 236e36c25f
2 changed files with 59 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
pc/data_channel_controller.cc
View File

@ -22,6 +22,7 @@
#include "rtc_base/location.h"
#include "rtc_base/logging.h"
#include "rtc_base/string_encode.h"
#include "rtc_base/task_utils/to_queued_task.h"
namespace webrtc {
@ -137,9 +138,10 @@ void DataChannelController::OnDataReceived(
cricket::ReceiveDataParams params;
params.sid = channel_id;
params.type = ToCricketDataMessageType(type);
data_channel_transport_invoker_.AsyncInvoke<void>(
RTC_FROM_HERE, signaling_thread(), [this, params, buffer] {
RTC_DCHECK_RUN_ON(signaling_thread());
signaling_thread()->PostTask(
ToQueuedTask([self = weak_factory_.GetWeakPtr(), params, buffer] {
if (self) {
RTC_DCHECK_RUN_ON(self->signaling_thread());
// TODO(bugs.webrtc.org/11547): The data being received should be
// delivered on the network thread. The way HandleOpenMessage_s works
// right now is that it's called for all types of buffers and operates
@ -148,48 +150,57 @@ void DataChannelController::OnDataReceived(
// deliver all other buffers on the network thread (change
// SignalDataChannelTransportReceivedData_s to
// SignalDataChannelTransportReceivedData_n).
if (!HandleOpenMessage_s(params, buffer)) {
SignalDataChannelTransportReceivedData_s(params, buffer);
if (!self->HandleOpenMessage_s(params, buffer)) {
self->SignalDataChannelTransportReceivedData_s(params, buffer);
}
});
}
}));
}
void DataChannelController::OnChannelClosing(int channel_id) {
RTC_DCHECK_RUN_ON(network_thread());
data_channel_transport_invoker_.AsyncInvoke<void>(
RTC_FROM_HERE, signaling_thread(), [this, channel_id] {
RTC_DCHECK_RUN_ON(signaling_thread());
SignalDataChannelTransportChannelClosing_s(channel_id);
});
signaling_thread()->PostTask(
ToQueuedTask([self = weak_factory_.GetWeakPtr(), channel_id] {
if (self) {
RTC_DCHECK_RUN_ON(self->signaling_thread());
self->SignalDataChannelTransportChannelClosing_s(channel_id);
}
}));
}
void DataChannelController::OnChannelClosed(int channel_id) {
RTC_DCHECK_RUN_ON(network_thread());
data_channel_transport_invoker_.AsyncInvoke<void>(
RTC_FROM_HERE, signaling_thread(), [this, channel_id] {
RTC_DCHECK_RUN_ON(signaling_thread());
SignalDataChannelTransportChannelClosed_s(channel_id);
});
signaling_thread()->PostTask(
ToQueuedTask([self = weak_factory_.GetWeakPtr(), channel_id] {
if (self) {
RTC_DCHECK_RUN_ON(self->signaling_thread());
self->SignalDataChannelTransportChannelClosed_s(channel_id);
}
}));
}
void DataChannelController::OnReadyToSend() {
RTC_DCHECK_RUN_ON(network_thread());
data_channel_transport_invoker_.AsyncInvoke<void>(
RTC_FROM_HERE, signaling_thread(), [this] {
RTC_DCHECK_RUN_ON(signaling_thread());
data_channel_transport_ready_to_send_ = true;
SignalDataChannelTransportWritable_s(
data_channel_transport_ready_to_send_);
});
signaling_thread()->PostTask(
ToQueuedTask([self = weak_factory_.GetWeakPtr()] {
if (self) {
RTC_DCHECK_RUN_ON(self->signaling_thread());
self->data_channel_transport_ready_to_send_ = true;
self->SignalDataChannelTransportWritable_s(
self->data_channel_transport_ready_to_send_);
}
}));
}
void DataChannelController::OnTransportClosed() {
RTC_DCHECK_RUN_ON(network_thread());
data_channel_transport_invoker_.AsyncInvoke<void>(
RTC_FROM_HERE, signaling_thread(), [this] {
RTC_DCHECK_RUN_ON(signaling_thread());
OnTransportChannelClosed();
});
signaling_thread()->PostTask(
ToQueuedTask([self = weak_factory_.GetWeakPtr()] {
if (self) {
RTC_DCHECK_RUN_ON(self->signaling_thread());
self->OnTransportChannelClosed();
}
}));
}
void DataChannelController::SetupDataChannelTransport_n() {
@ -392,12 +403,12 @@ void DataChannelController::OnSctpDataChannelClosed(SctpDataChannel* channel) {
sctp_data_channels_to_free_.push_back(*it);
sctp_data_channels_.erase(it);
signaling_thread()->PostTask(
RTC_FROM_HERE, [self = weak_factory_.GetWeakPtr()] {
ToQueuedTask([self = weak_factory_.GetWeakPtr()] {
if (self) {
RTC_DCHECK_RUN_ON(self->signaling_thread());
self->sctp_data_channels_to_free_.clear();
}
});
}));
return;
}
}
@ -598,13 +609,15 @@ bool DataChannelController::DataChannelSendData(
void DataChannelController::NotifyDataChannelsOfTransportCreated() {
RTC_DCHECK_RUN_ON(network_thread());
data_channel_transport_invoker_.AsyncInvoke<void>(
RTC_FROM_HERE, signaling_thread(), [this] {
RTC_DCHECK_RUN_ON(signaling_thread());
for (const auto& channel : sctp_data_channels_) {
signaling_thread()->PostTask(
ToQueuedTask([self = weak_factory_.GetWeakPtr()] {
if (self) {
RTC_DCHECK_RUN_ON(self->signaling_thread());
for (const auto& channel : self->sctp_data_channels_) {
channel->OnTransportChannelCreated();
}
});
}
}));
}
rtc::Thread* DataChannelController::network_thread() const {

3
pc/data_channel_controller.h
View File

@ -29,7 +29,6 @@
#include "pc/data_channel_utils.h"
#include "pc/rtp_data_channel.h"
#include "pc/sctp_data_channel.h"
#include "rtc_base/async_invoker.h"
#include "rtc_base/checks.h"
#include "rtc_base/copy_on_write_buffer.h"
#include "rtc_base/ssl_stream_adapter.h"
@ -245,6 +244,8 @@ class DataChannelController : public RtpDataChannelProviderInterface,
// Owning PeerConnection.
PeerConnection* const pc_;
// The weak pointers must be dereferenced and invalidated on the signalling
// thread only.
rtc::WeakPtrFactory<DataChannelController> weak_factory_{this};
};