OLD | NEW |
---|---|
1 /* | 1 /* |
2 * Copyright (c) 2012 The WebRTC project authors. All Rights Reserved. | 2 * Copyright (c) 2012 The WebRTC project authors. All Rights Reserved. |
3 * | 3 * |
4 * Use of this source code is governed by a BSD-style license | 4 * Use of this source code is governed by a BSD-style license |
5 * that can be found in the LICENSE file in the root of the source | 5 * that can be found in the LICENSE file in the root of the source |
6 * tree. An additional intellectual property rights grant can be found | 6 * tree. An additional intellectual property rights grant can be found |
7 * in the file PATENTS. All contributing project authors may | 7 * in the file PATENTS. All contributing project authors may |
8 * be found in the AUTHORS file in the root of the source tree. | 8 * be found in the AUTHORS file in the root of the source tree. |
9 */ | 9 */ |
10 | 10 |
11 #include "webrtc/modules/pacing/include/paced_sender.h" | 11 #include "webrtc/modules/pacing/include/paced_sender.h" |
12 | 12 |
13 #include <assert.h> | |
14 | |
15 #include <map> | 13 #include <map> |
16 #include <queue> | 14 #include <queue> |
17 #include <set> | 15 #include <set> |
18 | 16 |
17 #include "webrtc/base/checks.h" | |
19 #include "webrtc/modules/interface/module_common_types.h" | 18 #include "webrtc/modules/interface/module_common_types.h" |
20 #include "webrtc/modules/pacing/bitrate_prober.h" | 19 #include "webrtc/modules/pacing/bitrate_prober.h" |
21 #include "webrtc/system_wrappers/interface/clock.h" | 20 #include "webrtc/system_wrappers/interface/clock.h" |
22 #include "webrtc/system_wrappers/interface/critical_section_wrapper.h" | 21 #include "webrtc/system_wrappers/interface/critical_section_wrapper.h" |
23 #include "webrtc/system_wrappers/interface/field_trial.h" | 22 #include "webrtc/system_wrappers/interface/field_trial.h" |
24 #include "webrtc/system_wrappers/interface/logging.h" | 23 #include "webrtc/system_wrappers/interface/logging.h" |
25 | 24 |
26 namespace { | 25 namespace { |
27 // Time limit in milliseconds between packet bursts. | 26 // Time limit in milliseconds between packet bursts. |
28 const int64_t kMinPacketLimitMs = 5; | 27 const int64_t kMinPacketLimitMs = 5; |
(...skipping 111 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... | |
140 dupe_map_[packet.ssrc].insert(packet.sequence_number); | 139 dupe_map_[packet.ssrc].insert(packet.sequence_number); |
141 return true; | 140 return true; |
142 } | 141 } |
143 | 142 |
144 // Insert returns a pair, where second is a bool set to true if new element. | 143 // Insert returns a pair, where second is a bool set to true if new element. |
145 return it->second.insert(packet.sequence_number).second; | 144 return it->second.insert(packet.sequence_number).second; |
146 } | 145 } |
147 | 146 |
148 void RemoveFromDupeSet(const Packet& packet) { | 147 void RemoveFromDupeSet(const Packet& packet) { |
149 SsrcSeqNoMap::iterator it = dupe_map_.find(packet.ssrc); | 148 SsrcSeqNoMap::iterator it = dupe_map_.find(packet.ssrc); |
150 assert(it != dupe_map_.end()); | 149 RTC_DCHECK(it != dupe_map_.end()); |
151 it->second.erase(packet.sequence_number); | 150 it->second.erase(packet.sequence_number); |
152 if (it->second.empty()) { | 151 if (it->second.empty()) { |
153 dupe_map_.erase(it); | 152 dupe_map_.erase(it); |
154 } | 153 } |
155 } | 154 } |
156 | 155 |
157 // List of packets, in the order the were enqueued. Since dequeueing may | 156 // List of packets, in the order the were enqueued. Since dequeueing may |
158 // occur out of order, use list instead of vector. | 157 // occur out of order, use list instead of vector. |
159 std::list<Packet> packet_list_; | 158 std::list<Packet> packet_list_; |
160 // Priority queue of the packets, sorted according to Comparator. | 159 // Priority queue of the packets, sorted according to Comparator. |
(...skipping 41 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... | |
202 int target_rate_kbps() const { return target_rate_kbps_; } | 201 int target_rate_kbps() const { return target_rate_kbps_; } |
203 | 202 |
204 private: | 203 private: |
205 static const int kWindowMs = 500; | 204 static const int kWindowMs = 500; |
206 | 205 |
207 int target_rate_kbps_; | 206 int target_rate_kbps_; |
208 int bytes_remaining_; | 207 int bytes_remaining_; |
209 }; | 208 }; |
210 } // namespace paced_sender | 209 } // namespace paced_sender |
211 | 210 |
211 const int64_t PacedSender::kMaxQueueLengthMs = 2000; | |
212 const float PacedSender::kDefaultPaceMultiplier = 2.5f; | 212 const float PacedSender::kDefaultPaceMultiplier = 2.5f; |
213 | 213 |
214 PacedSender::PacedSender(Clock* clock, | 214 PacedSender::PacedSender(Clock* clock, |
215 Callback* callback, | 215 Callback* callback, |
216 int bitrate_kbps, | 216 int bitrate_kbps, |
217 int max_bitrate_kbps, | 217 int max_bitrate_kbps, |
218 int min_bitrate_kbps) | 218 int min_bitrate_kbps) |
219 : clock_(clock), | 219 : clock_(clock), |
220 callback_(callback), | 220 callback_(callback), |
221 critsect_(CriticalSectionWrapper::CreateCriticalSection()), | 221 critsect_(CriticalSectionWrapper::CreateCriticalSection()), |
222 paused_(false), | 222 paused_(false), |
223 probing_enabled_(true), | 223 probing_enabled_(true), |
224 media_budget_(new paced_sender::IntervalBudget(max_bitrate_kbps)), | 224 media_budget_(new paced_sender::IntervalBudget(max_bitrate_kbps)), |
225 padding_budget_(new paced_sender::IntervalBudget(min_bitrate_kbps)), | 225 padding_budget_(new paced_sender::IntervalBudget(min_bitrate_kbps)), |
226 prober_(new BitrateProber()), | 226 prober_(new BitrateProber()), |
227 bitrate_bps_(1000 * bitrate_kbps), | 227 bitrate_bps_(1000 * bitrate_kbps), |
228 max_bitrate_kbps_(max_bitrate_kbps), | |
229 bitrate_boost_start_us_(-1), | |
228 time_last_update_us_(clock->TimeInMicroseconds()), | 230 time_last_update_us_(clock->TimeInMicroseconds()), |
229 packets_(new paced_sender::PacketQueue()), | 231 packets_(new paced_sender::PacketQueue()), |
230 packet_counter_(0) { | 232 packet_counter_(0) { |
231 UpdateBytesPerInterval(kMinPacketLimitMs); | 233 UpdateBytesPerInterval(kMinPacketLimitMs); |
232 } | 234 } |
233 | 235 |
234 PacedSender::~PacedSender() {} | 236 PacedSender::~PacedSender() {} |
235 | 237 |
236 void PacedSender::Pause() { | 238 void PacedSender::Pause() { |
237 CriticalSectionScoped cs(critsect_.get()); | 239 CriticalSectionScoped cs(critsect_.get()); |
238 paused_ = true; | 240 paused_ = true; |
239 } | 241 } |
240 | 242 |
241 void PacedSender::Resume() { | 243 void PacedSender::Resume() { |
242 CriticalSectionScoped cs(critsect_.get()); | 244 CriticalSectionScoped cs(critsect_.get()); |
243 paused_ = false; | 245 paused_ = false; |
244 } | 246 } |
245 | 247 |
246 void PacedSender::SetProbingEnabled(bool enabled) { | 248 void PacedSender::SetProbingEnabled(bool enabled) { |
247 assert(packet_counter_ == 0); | 249 RTC_CHECK_EQ(0u, packet_counter_); |
248 probing_enabled_ = enabled; | 250 probing_enabled_ = enabled; |
249 } | 251 } |
250 | 252 |
251 void PacedSender::UpdateBitrate(int bitrate_kbps, | 253 void PacedSender::UpdateBitrate(int bitrate_kbps, |
252 int max_bitrate_kbps, | 254 int max_bitrate_kbps, |
253 int min_bitrate_kbps) { | 255 int min_bitrate_kbps) { |
254 CriticalSectionScoped cs(critsect_.get()); | 256 CriticalSectionScoped cs(critsect_.get()); |
255 media_budget_->set_target_rate_kbps(max_bitrate_kbps); | 257 if (bitrate_boost_start_us_ == -1 || |
258 max_bitrate_kbps > media_budget_->target_rate_kbps()) { | |
259 media_budget_->set_target_rate_kbps(max_bitrate_kbps); | |
260 bitrate_boost_start_us_ = -1; | |
261 } | |
256 padding_budget_->set_target_rate_kbps(min_bitrate_kbps); | 262 padding_budget_->set_target_rate_kbps(min_bitrate_kbps); |
257 bitrate_bps_ = 1000 * bitrate_kbps; | 263 bitrate_bps_ = 1000 * bitrate_kbps; |
264 max_bitrate_kbps_ = max_bitrate_kbps; | |
258 } | 265 } |
259 | 266 |
260 void PacedSender::InsertPacket(RtpPacketSender::Priority priority, | 267 void PacedSender::InsertPacket(RtpPacketSender::Priority priority, |
261 uint32_t ssrc, | 268 uint32_t ssrc, |
262 uint16_t sequence_number, | 269 uint16_t sequence_number, |
263 int64_t capture_time_ms, | 270 int64_t capture_time_ms, |
264 size_t bytes, | 271 size_t bytes, |
265 bool retransmission) { | 272 bool retransmission) { |
266 CriticalSectionScoped cs(critsect_.get()); | 273 CriticalSectionScoped cs(critsect_.get()); |
267 | 274 |
268 if (probing_enabled_ && !prober_->IsProbing()) { | 275 if (probing_enabled_ && !prober_->IsProbing()) |
269 prober_->SetEnabled(true); | 276 prober_->SetEnabled(true); |
270 } | |
271 prober_->MaybeInitializeProbe(bitrate_bps_); | 277 prober_->MaybeInitializeProbe(bitrate_bps_); |
272 | 278 |
273 if (capture_time_ms < 0) { | 279 if (capture_time_ms < 0) |
274 capture_time_ms = clock_->TimeInMilliseconds(); | 280 capture_time_ms = clock_->TimeInMilliseconds(); |
275 } | |
276 | 281 |
277 packets_->Push(paced_sender::Packet( | 282 packets_->Push(paced_sender::Packet( |
278 priority, ssrc, sequence_number, capture_time_ms, | 283 priority, ssrc, sequence_number, capture_time_ms, |
279 clock_->TimeInMilliseconds(), bytes, retransmission, packet_counter_++)); | 284 clock_->TimeInMilliseconds(), bytes, retransmission, packet_counter_++)); |
280 } | 285 } |
281 | 286 |
282 int64_t PacedSender::ExpectedQueueTimeMs() const { | 287 int64_t PacedSender::ExpectedQueueTimeMs() const { |
283 CriticalSectionScoped cs(critsect_.get()); | 288 CriticalSectionScoped cs(critsect_.get()); |
284 int target_rate = media_budget_->target_rate_kbps(); | 289 RTC_DCHECK_GT(max_bitrate_kbps_, 0); |
285 assert(target_rate > 0); | 290 return static_cast<int64_t>(packets_->SizeInBytes() * 8 / max_bitrate_kbps_); |
286 return static_cast<int64_t>(packets_->SizeInBytes() * 8 / target_rate); | |
287 } | 291 } |
288 | 292 |
289 size_t PacedSender::QueueSizePackets() const { | 293 size_t PacedSender::QueueSizePackets() const { |
290 CriticalSectionScoped cs(critsect_.get()); | 294 CriticalSectionScoped cs(critsect_.get()); |
291 return packets_->SizeInPackets(); | 295 return packets_->SizeInPackets(); |
292 } | 296 } |
293 | 297 |
294 int64_t PacedSender::QueueInMs() const { | 298 int64_t PacedSender::QueueInMs() const { |
295 CriticalSectionScoped cs(critsect_.get()); | 299 CriticalSectionScoped cs(critsect_.get()); |
296 | 300 |
297 int64_t oldest_packet = packets_->OldestEnqueueTime(); | 301 int64_t oldest_packet = packets_->OldestEnqueueTime(); |
298 if (oldest_packet == 0) | 302 if (oldest_packet == 0) |
299 return 0; | 303 return 0; |
300 | 304 |
301 return clock_->TimeInMilliseconds() - oldest_packet; | 305 return clock_->TimeInMilliseconds() - oldest_packet; |
302 } | 306 } |
303 | 307 |
304 int64_t PacedSender::TimeUntilNextProcess() { | 308 int64_t PacedSender::TimeUntilNextProcess() { |
305 CriticalSectionScoped cs(critsect_.get()); | 309 CriticalSectionScoped cs(critsect_.get()); |
306 if (prober_->IsProbing()) { | 310 if (prober_->IsProbing()) { |
307 int64_t ret = prober_->TimeUntilNextProbe(clock_->TimeInMilliseconds()); | 311 int64_t ret = prober_->TimeUntilNextProbe(clock_->TimeInMilliseconds()); |
308 if (ret >= 0) { | 312 if (ret >= 0) |
309 return ret; | 313 return ret; |
310 } | |
311 } | 314 } |
312 int64_t elapsed_time_us = clock_->TimeInMicroseconds() - time_last_update_us_; | 315 int64_t elapsed_time_us = clock_->TimeInMicroseconds() - time_last_update_us_; |
313 int64_t elapsed_time_ms = (elapsed_time_us + 500) / 1000; | 316 int64_t elapsed_time_ms = (elapsed_time_us + 500) / 1000; |
314 return std::max<int64_t>(kMinPacketLimitMs - elapsed_time_ms, 0); | 317 return std::max<int64_t>(kMinPacketLimitMs - elapsed_time_ms, 0); |
315 } | 318 } |
316 | 319 |
317 int32_t PacedSender::Process() { | 320 int32_t PacedSender::Process() { |
318 int64_t now_us = clock_->TimeInMicroseconds(); | 321 int64_t now_us = clock_->TimeInMicroseconds(); |
319 CriticalSectionScoped cs(critsect_.get()); | 322 CriticalSectionScoped cs(critsect_.get()); |
320 int64_t elapsed_time_ms = (now_us - time_last_update_us_ + 500) / 1000; | 323 int64_t elapsed_time_ms = (now_us - time_last_update_us_ + 500) / 1000; |
321 time_last_update_us_ = now_us; | 324 time_last_update_us_ = now_us; |
322 if (paused_) | 325 if (paused_) |
323 return 0; | 326 return 0; |
324 if (elapsed_time_ms > 0) { | 327 if (elapsed_time_ms > 0) { |
328 if (bitrate_boost_start_us_ != -1 && | |
stefan-webrtc
2015/10/28 15:45:23
Is the bitrate_boost_start_us_ really needed? Can'
sprang_webrtc
2015/11/02 12:17:56
This was how I first implemented it, and it turns
stefan-webrtc
2015/11/10 14:57:02
This isn't correct I think. Those 500 bytes have b
stefan-webrtc
2015/11/10 14:59:44
And something similar could be done when a packet
sprang_webrtc
2015/11/19 10:21:48
This method won't work quite right either, since w
| |
329 (now_us - bitrate_boost_start_us_) > kMaxQueueLengthMs * 1000) { | |
330 // Boost period is up, reset to nominal bitrate. | |
331 media_budget_->set_target_rate_kbps(max_bitrate_kbps_); | |
332 bitrate_boost_start_us_ = -1; | |
333 } | |
334 | |
335 int64_t expected_queue_length_ms = | |
336 static_cast<int64_t>(packets_->SizeInBytes() * 8 / max_bitrate_kbps_); | |
stefan-webrtc
2015/10/28 15:45:23
Call ExpectedQueueTimeMs() instead.
sprang_webrtc
2015/11/02 12:17:56
Sure. Or do we care about recursively taking locks
stefan-webrtc
2015/11/10 14:57:02
I would prefer to not take them recursively. Just
| |
337 if (expected_queue_length_ms > kMaxQueueLengthMs) { | |
338 // Too much data has been put in the pacer queue, or the target bitrate | |
339 // has suddenly decreased, making the expected max pacer delay too long. | |
340 // Boost the media target bitrate so that all data should have left the | |
341 // queue within kDefaultMaxQueueLengthMs. | |
342 int required_bitrate_kbps = | |
343 expected_queue_length_ms * max_bitrate_kbps_ / kMaxQueueLengthMs; | |
344 if (required_bitrate_kbps > media_budget_->target_rate_kbps()) { | |
345 media_budget_->set_target_rate_kbps(required_bitrate_kbps); | |
346 bitrate_boost_start_us_ = now_us; | |
347 } | |
348 } | |
349 | |
325 int64_t delta_time_ms = std::min(kMaxIntervalTimeMs, elapsed_time_ms); | 350 int64_t delta_time_ms = std::min(kMaxIntervalTimeMs, elapsed_time_ms); |
326 UpdateBytesPerInterval(delta_time_ms); | 351 UpdateBytesPerInterval(delta_time_ms); |
327 } | 352 } |
328 while (!packets_->Empty()) { | 353 while (!packets_->Empty()) { |
329 if (media_budget_->bytes_remaining() == 0 && !prober_->IsProbing()) { | 354 if (media_budget_->bytes_remaining() == 0 && !prober_->IsProbing()) |
330 return 0; | 355 return 0; |
331 } | |
332 | 356 |
333 // Since we need to release the lock in order to send, we first pop the | 357 // Since we need to release the lock in order to send, we first pop the |
334 // element from the priority queue but keep it in storage, so that we can | 358 // element from the priority queue but keep it in storage, so that we can |
335 // reinsert it if send fails. | 359 // reinsert it if send fails. |
336 const paced_sender::Packet& packet = packets_->BeginPop(); | 360 const paced_sender::Packet& packet = packets_->BeginPop(); |
337 if (SendPacket(packet)) { | 361 if (SendPacket(packet)) { |
338 // Send succeeded, remove it from the queue. | 362 // Send succeeded, remove it from the queue. |
339 packets_->FinalizePop(packet); | 363 packets_->FinalizePop(packet); |
340 if (prober_->IsProbing()) { | 364 if (prober_->IsProbing()) |
341 return 0; | 365 return 0; |
342 } | |
343 } else { | 366 } else { |
344 // Send failed, put it back into the queue. | 367 // Send failed, put it back into the queue. |
345 packets_->CancelPop(packet); | 368 packets_->CancelPop(packet); |
346 return 0; | 369 return 0; |
347 } | 370 } |
348 } | 371 } |
349 | 372 |
350 if (!packets_->Empty()) | 373 if (!packets_->Empty()) |
351 return 0; | 374 return 0; |
352 | 375 |
353 size_t padding_needed; | 376 size_t padding_needed; |
354 if (prober_->IsProbing()) | 377 if (prober_->IsProbing()) { |
355 padding_needed = prober_->RecommendedPacketSize(); | 378 padding_needed = prober_->RecommendedPacketSize(); |
356 else | 379 } else { |
357 padding_needed = padding_budget_->bytes_remaining(); | 380 padding_needed = padding_budget_->bytes_remaining(); |
381 } | |
358 | 382 |
359 if (padding_needed > 0) | 383 if (padding_needed > 0) |
360 SendPadding(static_cast<size_t>(padding_needed)); | 384 SendPadding(static_cast<size_t>(padding_needed)); |
361 return 0; | 385 return 0; |
362 } | 386 } |
363 | 387 |
364 bool PacedSender::SendPacket(const paced_sender::Packet& packet) { | 388 bool PacedSender::SendPacket(const paced_sender::Packet& packet) { |
365 critsect_->Leave(); | 389 critsect_->Leave(); |
366 const bool success = callback_->TimeToSendPacket(packet.ssrc, | 390 const bool success = callback_->TimeToSendPacket(packet.ssrc, |
367 packet.sequence_number, | 391 packet.sequence_number, |
(...skipping 21 matching lines...) Expand all Loading... | |
389 media_budget_->UseBudget(bytes_sent); | 413 media_budget_->UseBudget(bytes_sent); |
390 padding_budget_->UseBudget(bytes_sent); | 414 padding_budget_->UseBudget(bytes_sent); |
391 } | 415 } |
392 } | 416 } |
393 | 417 |
394 void PacedSender::UpdateBytesPerInterval(int64_t delta_time_ms) { | 418 void PacedSender::UpdateBytesPerInterval(int64_t delta_time_ms) { |
395 media_budget_->IncreaseBudget(delta_time_ms); | 419 media_budget_->IncreaseBudget(delta_time_ms); |
396 padding_budget_->IncreaseBudget(delta_time_ms); | 420 padding_budget_->IncreaseBudget(delta_time_ms); |
397 } | 421 } |
398 } // namespace webrtc | 422 } // namespace webrtc |
OLD | NEW |