generic-library/webrtc
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Potential dead lock in receive statistics

Browse Source 
A dead lock could occur if the following to code paths are called
concurrently:

ReceiveStatisticsImpl::IncomingPacket() ->
  StreamStatisticianImpl::IncomingPacket()

StreamStatisticianImpl::GetStatistics() ->
  ReceiveStatisticsImpl::StatisticsUpdated()

Solution is to release ReceiveStatisticsImpl lock after lookup/lazy-init of StreamStatisticianImpl. Don't need to hold it when doing the call to StreamStatisticianImpl::IncomingPacket().

BUG=2818
R=asapersson@webrtc.org, stefan@webrtc.org

Review URL: https://webrtc-codereview.appspot.com/7389004

git-svn-id: http://webrtc.googlecode.com/svn/trunk@5406 4adac7df-926f-26a2-2b94-8c16560cd09d
This commit is contained in:
sprang@webrtc.org 2014-01-21 16:33:37 +00:00
parent 32c3247418
commit 7dba27c740
2 changed files with 69 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

120
webrtc/modules/rtp_rtcp/source/receive_statistics_impl.cc
View File

@ -28,7 +28,7 @@ StreamStatisticianImpl::StreamStatisticianImpl(
Clock* clock, Clock* clock,
RtcpStatisticsCallback* rtcp_callback) RtcpStatisticsCallback* rtcp_callback)
: clock_(clock), : clock_(clock),
crit_sect_(CriticalSectionWrapper::CreateCriticalSection()), stream_lock_(CriticalSectionWrapper::CreateCriticalSection()),
incoming_bitrate_(clock, NULL), incoming_bitrate_(clock, NULL),
ssrc_(0), ssrc_(0),
max_reordering_threshold_(kDefaultMaxReorderingThreshold), max_reordering_threshold_(kDefaultMaxReorderingThreshold),
@ -55,7 +55,7 @@ StreamStatisticianImpl::StreamStatisticianImpl(
rtcp_callback_(rtcp_callback) {} rtcp_callback_(rtcp_callback) {}
void StreamStatisticianImpl::ResetStatistics() { void StreamStatisticianImpl::ResetStatistics() {
CriticalSectionScoped cs(crit_sect_.get()); CriticalSectionScoped cs(stream_lock_.get());
last_report_inorder_packets_ = 0; last_report_inorder_packets_ = 0;
last_report_old_packets_ = 0; last_report_old_packets_ = 0;
last_report_seq_max_ = 0; last_report_seq_max_ = 0;
@ -75,7 +75,7 @@ void StreamStatisticianImpl::ResetStatistics() {
void StreamStatisticianImpl::IncomingPacket(const RTPHeader& header, void StreamStatisticianImpl::IncomingPacket(const RTPHeader& header,
size_t bytes, size_t bytes,
bool retransmitted) { bool retransmitted) {
CriticalSectionScoped cs(crit_sect_.get()); CriticalSectionScoped cs(stream_lock_.get());
bool in_order = InOrderPacketInternal(header.sequenceNumber); bool in_order = InOrderPacketInternal(header.sequenceNumber);
ssrc_ = header.ssrc; ssrc_ = header.ssrc;
incoming_bitrate_.Update(bytes); incoming_bitrate_.Update(bytes);
@ -170,28 +170,39 @@ void StreamStatisticianImpl::IncomingPacket(const RTPHeader& header,
void StreamStatisticianImpl::SetMaxReorderingThreshold( void StreamStatisticianImpl::SetMaxReorderingThreshold(
int max_reordering_threshold) { int max_reordering_threshold) {
CriticalSectionScoped cs(crit_sect_.get()); CriticalSectionScoped cs(stream_lock_.get());
max_reordering_threshold_ = max_reordering_threshold; max_reordering_threshold_ = max_reordering_threshold;
} }
bool StreamStatisticianImpl::GetStatistics(RtcpStatistics* statistics, bool StreamStatisticianImpl::GetStatistics(RtcpStatistics* statistics,
bool reset) { bool reset) {
CriticalSectionScoped cs(crit_sect_.get()); {
if (received_seq_first_ == 0 && received_byte_count_ == 0) { CriticalSectionScoped cs(stream_lock_.get());
// We have not received anything. if (received_seq_first_ == 0 && received_byte_count_ == 0) {
return false; // We have not received anything.
}
if (!reset) {
if (last_report_inorder_packets_ == 0) {
// No report.
return false; return false;
} }
// Just get last report.
*statistics = last_reported_statistics_; if (!reset) {
return true; if (last_report_inorder_packets_ == 0) {
// No report.
return false;
}
// Just get last report.
*statistics = last_reported_statistics_;
return true;
}
*statistics = CalculateStatistics();
} }
rtcp_callback_->StatisticsUpdated(*statistics, ssrc_);
return true;
}
RtcpStatistics StreamStatisticianImpl::CalculateStatistics() {
RtcpStatistics stats;
if (last_report_inorder_packets_ == 0) { if (last_report_inorder_packets_ == 0) {
// First time we send a report. // First time we send a report.
last_report_seq_max_ = received_seq_first_ - 1; last_report_seq_max_ = received_seq_first_ - 1;
@ -233,32 +244,30 @@ bool StreamStatisticianImpl::GetStatistics(RtcpStatistics* statistics,
local_fraction_lost = local_fraction_lost =
static_cast<uint8_t>(255 * missing / exp_since_last); static_cast<uint8_t>(255 * missing / exp_since_last);
} }
statistics->fraction_lost = local_fraction_lost; stats.fraction_lost = local_fraction_lost;
// We need a counter for cumulative loss too. // We need a counter for cumulative loss too.
cumulative_loss_ += missing; cumulative_loss_ += missing;
statistics->cumulative_lost = cumulative_loss_; stats.cumulative_lost = cumulative_loss_;
statistics->extended_max_sequence_number = (received_seq_wraps_ << 16) + stats.extended_max_sequence_number =
received_seq_max_; (received_seq_wraps_ << 16) + received_seq_max_;
// Note: internal jitter value is in Q4 and needs to be scaled by 1/16. // Note: internal jitter value is in Q4 and needs to be scaled by 1/16.
statistics->jitter = jitter_q4_ >> 4; stats.jitter = jitter_q4_ >> 4;
if (reset) {
// Store this report.
last_reported_statistics_ = *statistics;
// Only for report blocks in RTCP SR and RR. // Store this report.
last_report_inorder_packets_ = received_inorder_packet_count_; last_reported_statistics_ = stats;
last_report_old_packets_ = received_retransmitted_packets_;
last_report_seq_max_ = received_seq_max_;
}
rtcp_callback_->StatisticsUpdated(last_reported_statistics_, ssrc_); // Only for report blocks in RTCP SR and RR.
return true; last_report_inorder_packets_ = received_inorder_packet_count_;
last_report_old_packets_ = received_retransmitted_packets_;
last_report_seq_max_ = received_seq_max_;
return stats;
} }
void StreamStatisticianImpl::GetDataCounters( void StreamStatisticianImpl::GetDataCounters(
uint32_t* bytes_received, uint32_t* packets_received) const { uint32_t* bytes_received, uint32_t* packets_received) const {
CriticalSectionScoped cs(crit_sect_.get()); CriticalSectionScoped cs(stream_lock_.get());
if (bytes_received) { if (bytes_received) {
*bytes_received = received_byte_count_; *bytes_received = received_byte_count_;
} }
@ -269,25 +278,25 @@ void StreamStatisticianImpl::GetDataCounters(
} }
uint32_t StreamStatisticianImpl::BitrateReceived() const { uint32_t StreamStatisticianImpl::BitrateReceived() const {
CriticalSectionScoped cs(crit_sect_.get()); CriticalSectionScoped cs(stream_lock_.get());
return incoming_bitrate_.BitrateNow(); return incoming_bitrate_.BitrateNow();
} }
void StreamStatisticianImpl::ProcessBitrate() { void StreamStatisticianImpl::ProcessBitrate() {
CriticalSectionScoped cs(crit_sect_.get()); CriticalSectionScoped cs(stream_lock_.get());
incoming_bitrate_.Process(); incoming_bitrate_.Process();
} }
void StreamStatisticianImpl::LastReceiveTimeNtp(uint32_t* secs, void StreamStatisticianImpl::LastReceiveTimeNtp(uint32_t* secs,
uint32_t* frac) const { uint32_t* frac) const {
CriticalSectionScoped cs(crit_sect_.get()); CriticalSectionScoped cs(stream_lock_.get());
*secs = last_receive_time_secs_; *secs = last_receive_time_secs_;
*frac = last_receive_time_frac_; *frac = last_receive_time_frac_;
} }
bool StreamStatisticianImpl::IsRetransmitOfOldPacket( bool StreamStatisticianImpl::IsRetransmitOfOldPacket(
const RTPHeader& header, int min_rtt) const { const RTPHeader& header, int min_rtt) const {
CriticalSectionScoped cs(crit_sect_.get()); CriticalSectionScoped cs(stream_lock_.get());
if (InOrderPacketInternal(header.sequenceNumber)) { if (InOrderPacketInternal(header.sequenceNumber)) {
return false; return false;
} }
@ -322,7 +331,7 @@ bool StreamStatisticianImpl::IsRetransmitOfOldPacket(
} }
bool StreamStatisticianImpl::IsPacketInOrder(uint16_t sequence_number) const { bool StreamStatisticianImpl::IsPacketInOrder(uint16_t sequence_number) const {
CriticalSectionScoped cs(crit_sect_.get()); CriticalSectionScoped cs(stream_lock_.get());
return InOrderPacketInternal(sequence_number); return InOrderPacketInternal(sequence_number);
} }
@ -347,7 +356,7 @@ ReceiveStatistics* ReceiveStatistics::Create(Clock* clock) {
ReceiveStatisticsImpl::ReceiveStatisticsImpl(Clock* clock) ReceiveStatisticsImpl::ReceiveStatisticsImpl(Clock* clock)
: clock_(clock), : clock_(clock),
crit_sect_(CriticalSectionWrapper::CreateCriticalSection()), receive_statistics_lock_(CriticalSectionWrapper::CreateCriticalSection()),
last_rate_update_ms_(0), last_rate_update_ms_(0),
rtcp_stats_callback_(NULL) {} rtcp_stats_callback_(NULL) {}
@ -360,19 +369,22 @@ ReceiveStatisticsImpl::~ReceiveStatisticsImpl() {
void ReceiveStatisticsImpl::IncomingPacket(const RTPHeader& header, void ReceiveStatisticsImpl::IncomingPacket(const RTPHeader& header,
size_t bytes, bool old_packet) { size_t bytes, bool old_packet) {
CriticalSectionScoped cs(crit_sect_.get()); StatisticianImplMap::iterator it;
StatisticianImplMap::iterator it = statisticians_.find(header.ssrc); {
if (it == statisticians_.end()) { CriticalSectionScoped cs(receive_statistics_lock_.get());
std::pair<StatisticianImplMap::iterator, uint32_t> insert_result = it = statisticians_.find(header.ssrc);
statisticians_.insert(std::make_pair( if (it == statisticians_.end()) {
header.ssrc, new StreamStatisticianImpl(clock_, this))); std::pair<StatisticianImplMap::iterator, uint32_t> insert_result =
it = insert_result.first; statisticians_.insert(std::make_pair(
header.ssrc, new StreamStatisticianImpl(clock_, this)));
it = insert_result.first;
}
} }
statisticians_[header.ssrc]->IncomingPacket(header, bytes, old_packet); it->second->IncomingPacket(header, bytes, old_packet);
} }
void ReceiveStatisticsImpl::ChangeSsrc(uint32_t from_ssrc, uint32_t to_ssrc) { void ReceiveStatisticsImpl::ChangeSsrc(uint32_t from_ssrc, uint32_t to_ssrc) {
CriticalSectionScoped cs(crit_sect_.get()); CriticalSectionScoped cs(receive_statistics_lock_.get());
StatisticianImplMap::iterator from_it = statisticians_.find(from_ssrc); StatisticianImplMap::iterator from_it = statisticians_.find(from_ssrc);
if (from_it == statisticians_.end()) if (from_it == statisticians_.end())
return; return;
@ -383,7 +395,7 @@ void ReceiveStatisticsImpl::ChangeSsrc(uint32_t from_ssrc, uint32_t to_ssrc) {
} }
StatisticianMap ReceiveStatisticsImpl::GetActiveStatisticians() const { StatisticianMap ReceiveStatisticsImpl::GetActiveStatisticians() const {
CriticalSectionScoped cs(crit_sect_.get()); CriticalSectionScoped cs(receive_statistics_lock_.get());
StatisticianMap active_statisticians; StatisticianMap active_statisticians;
for (StatisticianImplMap::const_iterator it = statisticians_.begin(); for (StatisticianImplMap::const_iterator it = statisticians_.begin();
it != statisticians_.end(); ++it) { it != statisticians_.end(); ++it) {
@ -400,7 +412,7 @@ StatisticianMap ReceiveStatisticsImpl::GetActiveStatisticians() const {
StreamStatistician* ReceiveStatisticsImpl::GetStatistician( StreamStatistician* ReceiveStatisticsImpl::GetStatistician(
uint32_t ssrc) const { uint32_t ssrc) const {
CriticalSectionScoped cs(crit_sect_.get()); CriticalSectionScoped cs(receive_statistics_lock_.get());
StatisticianImplMap::const_iterator it = statisticians_.find(ssrc); StatisticianImplMap::const_iterator it = statisticians_.find(ssrc);
if (it == statisticians_.end()) if (it == statisticians_.end())
return NULL; return NULL;
@ -409,7 +421,7 @@ StreamStatistician* ReceiveStatisticsImpl::GetStatistician(
void ReceiveStatisticsImpl::SetMaxReorderingThreshold( void ReceiveStatisticsImpl::SetMaxReorderingThreshold(
int max_reordering_threshold) { int max_reordering_threshold) {
CriticalSectionScoped cs(crit_sect_.get()); CriticalSectionScoped cs(receive_statistics_lock_.get());
for (StatisticianImplMap::iterator it = statisticians_.begin(); for (StatisticianImplMap::iterator it = statisticians_.begin();
it != statisticians_.end(); ++it) { it != statisticians_.end(); ++it) {
it->second->SetMaxReorderingThreshold(max_reordering_threshold); it->second->SetMaxReorderingThreshold(max_reordering_threshold);
@ -417,7 +429,7 @@ void ReceiveStatisticsImpl::SetMaxReorderingThreshold(
} }
int32_t ReceiveStatisticsImpl::Process() { int32_t ReceiveStatisticsImpl::Process() {
CriticalSectionScoped cs(crit_sect_.get()); CriticalSectionScoped cs(receive_statistics_lock_.get());
for (StatisticianImplMap::iterator it = statisticians_.begin(); for (StatisticianImplMap::iterator it = statisticians_.begin();
it != statisticians_.end(); ++it) { it != statisticians_.end(); ++it) {
it->second->ProcessBitrate(); it->second->ProcessBitrate();
@ -427,7 +439,7 @@ int32_t ReceiveStatisticsImpl::Process() {
} }
int32_t ReceiveStatisticsImpl::TimeUntilNextProcess() { int32_t ReceiveStatisticsImpl::TimeUntilNextProcess() {
CriticalSectionScoped cs(crit_sect_.get()); CriticalSectionScoped cs(receive_statistics_lock_.get());
int time_since_last_update = clock_->TimeInMilliseconds() - int time_since_last_update = clock_->TimeInMilliseconds() -
last_rate_update_ms_; last_rate_update_ms_;
return std::max(kStatisticsProcessIntervalMs - time_since_last_update, 0); return std::max(kStatisticsProcessIntervalMs - time_since_last_update, 0);
@ -435,7 +447,7 @@ int32_t ReceiveStatisticsImpl::TimeUntilNextProcess() {
void ReceiveStatisticsImpl::RegisterRtcpStatisticsCallback( void ReceiveStatisticsImpl::RegisterRtcpStatisticsCallback(
RtcpStatisticsCallback* callback) { RtcpStatisticsCallback* callback) {
CriticalSectionScoped cs(crit_sect_.get()); CriticalSectionScoped cs(receive_statistics_lock_.get());
if (callback != NULL) if (callback != NULL)
assert(rtcp_stats_callback_ == NULL); assert(rtcp_stats_callback_ == NULL);
rtcp_stats_callback_ = callback; rtcp_stats_callback_ = callback;
@ -443,7 +455,7 @@ void ReceiveStatisticsImpl::RegisterRtcpStatisticsCallback(
void ReceiveStatisticsImpl::StatisticsUpdated(const RtcpStatistics& statistics, void ReceiveStatisticsImpl::StatisticsUpdated(const RtcpStatistics& statistics,
uint32_t ssrc) { uint32_t ssrc) {
CriticalSectionScoped cs(crit_sect_.get()); CriticalSectionScoped cs(receive_statistics_lock_.get());
if (rtcp_stats_callback_) { if (rtcp_stats_callback_) {
rtcp_stats_callback_->StatisticsUpdated(statistics, ssrc); rtcp_stats_callback_->StatisticsUpdated(statistics, ssrc);
} }

5
webrtc/modules/rtp_rtcp/source/receive_statistics_impl.h
View File

@ -45,9 +45,10 @@ class StreamStatisticianImpl : public StreamStatistician {
private: private:
bool InOrderPacketInternal(uint16_t sequence_number) const; bool InOrderPacketInternal(uint16_t sequence_number) const;
RtcpStatistics CalculateStatistics();
Clock* clock_; Clock* clock_;
scoped_ptr<CriticalSectionWrapper> crit_sect_; scoped_ptr<CriticalSectionWrapper> stream_lock_;
Bitrate incoming_bitrate_; Bitrate incoming_bitrate_;
uint32_t ssrc_; uint32_t ssrc_;
int max_reordering_threshold_; // In number of packets or sequence numbers. int max_reordering_threshold_; // In number of packets or sequence numbers.
@ -112,7 +113,7 @@ class ReceiveStatisticsImpl : public ReceiveStatistics,
typedef std::map<uint32_t, StreamStatisticianImpl*> StatisticianImplMap; typedef std::map<uint32_t, StreamStatisticianImpl*> StatisticianImplMap;
Clock* clock_; Clock* clock_;
scoped_ptr<CriticalSectionWrapper> crit_sect_; scoped_ptr<CriticalSectionWrapper> receive_statistics_lock_;
int64_t last_rate_update_ms_; int64_t last_rate_update_ms_;
StatisticianImplMap statisticians_; StatisticianImplMap statisticians_;