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 |
(...skipping 14 matching lines...) Expand all Loading... | |
25 namespace { | 25 namespace { |
26 // Time limit in milliseconds between packet bursts. | 26 // Time limit in milliseconds between packet bursts. |
27 const int64_t kMinPacketLimitMs = 5; | 27 const int64_t kMinPacketLimitMs = 5; |
28 | 28 |
29 // Upper cap on process interval, in case process has not been called in a long | 29 // Upper cap on process interval, in case process has not been called in a long |
30 // time. | 30 // time. |
31 const int64_t kMaxIntervalTimeMs = 30; | 31 const int64_t kMaxIntervalTimeMs = 30; |
32 | 32 |
33 } // namespace | 33 } // namespace |
34 | 34 |
35 // TODO(sprang): Move at least PacketQueue and MediaBudget out to separate | |
36 // files, so that we can more easily test them. | |
37 | |
35 namespace webrtc { | 38 namespace webrtc { |
36 namespace paced_sender { | 39 namespace paced_sender { |
37 struct Packet { | 40 struct Packet { |
38 Packet(RtpPacketSender::Priority priority, | 41 Packet(RtpPacketSender::Priority priority, |
39 uint32_t ssrc, | 42 uint32_t ssrc, |
40 uint16_t seq_number, | 43 uint16_t seq_number, |
41 int64_t capture_time_ms, | 44 int64_t capture_time_ms, |
42 int64_t enqueue_time_ms, | 45 int64_t enqueue_time_ms, |
43 size_t length_in_bytes, | 46 size_t length_in_bytes, |
44 bool retransmission, | 47 bool retransmission, |
(...skipping 41 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... | |
86 class PacketQueue { | 89 class PacketQueue { |
87 public: | 90 public: |
88 explicit PacketQueue(Clock* clock) | 91 explicit PacketQueue(Clock* clock) |
89 : bytes_(0), | 92 : bytes_(0), |
90 clock_(clock), | 93 clock_(clock), |
91 queue_time_sum_(0), | 94 queue_time_sum_(0), |
92 time_last_updated_(clock_->TimeInMilliseconds()) {} | 95 time_last_updated_(clock_->TimeInMilliseconds()) {} |
93 virtual ~PacketQueue() {} | 96 virtual ~PacketQueue() {} |
94 | 97 |
95 void Push(const Packet& packet) { | 98 void Push(const Packet& packet) { |
96 if (!AddToDupeSet(packet)) { | 99 if (!AddToDupeSet(packet)) |
97 return; | 100 return; |
98 } | 101 |
102 UpdateQueueTime(packet.enqueue_time_ms); | |
103 | |
99 // Store packet in list, use pointers in priority queue for cheaper moves. | 104 // Store packet in list, use pointers in priority queue for cheaper moves. |
100 // Packets have a handle to its own iterator in the list, for easy removal | 105 // Packets have a handle to its own iterator in the list, for easy removal |
101 // when popping from queue. | 106 // when popping from queue. |
102 packet_list_.push_front(packet); | 107 packet_list_.push_front(packet); |
103 std::list<Packet>::iterator it = packet_list_.begin(); | 108 std::list<Packet>::iterator it = packet_list_.begin(); |
104 it->this_it = it; // Handle for direct removal from list. | 109 it->this_it = it; // Handle for direct removal from list. |
105 prio_queue_.push(&(*it)); // Pointer into list. | 110 prio_queue_.push(&(*it)); // Pointer into list. |
106 bytes_ += packet.bytes; | 111 bytes_ += packet.bytes; |
107 } | 112 } |
108 | 113 |
109 const Packet& BeginPop() { | 114 const Packet& BeginPop() { |
110 const Packet& packet = *prio_queue_.top(); | 115 const Packet& packet = *prio_queue_.top(); |
111 prio_queue_.pop(); | 116 prio_queue_.pop(); |
112 return packet; | 117 return packet; |
113 } | 118 } |
114 | 119 |
115 void CancelPop(const Packet& packet) { prio_queue_.push(&(*packet.this_it)); } | 120 void CancelPop(const Packet& packet) { prio_queue_.push(&(*packet.this_it)); } |
116 | 121 |
117 void FinalizePop(const Packet& packet) { | 122 void FinalizePop(const Packet& packet) { |
118 RemoveFromDupeSet(packet); | 123 RemoveFromDupeSet(packet); |
119 bytes_ -= packet.bytes; | 124 bytes_ -= packet.bytes; |
120 queue_time_sum_ -= (time_last_updated_ - packet.enqueue_time_ms); | 125 queue_time_sum_ -= (time_last_updated_ - packet.enqueue_time_ms); |
121 packet_list_.erase(packet.this_it); | 126 packet_list_.erase(packet.this_it); |
122 } | 127 } |
mflodman
2015/11/26 12:06:23
Can we add a DCHECK to verify queue_time_sum_ == 0
| |
123 | 128 |
124 bool Empty() const { return prio_queue_.empty(); } | 129 bool Empty() const { return prio_queue_.empty(); } |
125 | 130 |
126 size_t SizeInPackets() const { return prio_queue_.size(); } | 131 size_t SizeInPackets() const { return prio_queue_.size(); } |
127 | 132 |
128 uint64_t SizeInBytes() const { return bytes_; } | 133 uint64_t SizeInBytes() const { return bytes_; } |
129 | 134 |
130 int64_t OldestEnqueueTimeMs() const { | 135 int64_t OldestEnqueueTimeMs() const { |
131 auto it = packet_list_.rbegin(); | 136 auto it = packet_list_.rbegin(); |
132 if (it == packet_list_.rend()) | 137 if (it == packet_list_.rend()) |
133 return 0; | 138 return 0; |
134 return it->enqueue_time_ms; | 139 return it->enqueue_time_ms; |
135 } | 140 } |
136 | 141 |
137 int64_t AverageQueueTimeMs() { | 142 void UpdateQueueTime(int64_t timestamp_ms) { |
138 int64_t now = clock_->TimeInMilliseconds(); | 143 RTC_DCHECK_GE(timestamp_ms, time_last_updated_); |
139 RTC_DCHECK_GE(now, time_last_updated_); | 144 int64_t delta = timestamp_ms - time_last_updated_; |
140 int64_t delta = now - time_last_updated_; | |
141 queue_time_sum_ += delta * prio_queue_.size(); | 145 queue_time_sum_ += delta * prio_queue_.size(); |
142 time_last_updated_ = now; | 146 time_last_updated_ = timestamp_ms; |
147 } | |
148 | |
149 int64_t AverageQueueTimeMs() const { | |
150 if (prio_queue_.empty()) | |
151 return 0; | |
mflodman
2015/11/26 12:06:23
Or here if it's easier logically to check here.
| |
143 return queue_time_sum_ / prio_queue_.size(); | 152 return queue_time_sum_ / prio_queue_.size(); |
144 } | 153 } |
145 | 154 |
146 private: | 155 private: |
147 // Try to add a packet to the set of ssrc/seqno identifiers currently in the | 156 // Try to add a packet to the set of ssrc/seqno identifiers currently in the |
148 // queue. Return true if inserted, false if this is a duplicate. | 157 // queue. Return true if inserted, false if this is a duplicate. |
149 bool AddToDupeSet(const Packet& packet) { | 158 bool AddToDupeSet(const Packet& packet) { |
150 SsrcSeqNoMap::iterator it = dupe_map_.find(packet.ssrc); | 159 SsrcSeqNoMap::iterator it = dupe_map_.find(packet.ssrc); |
151 if (it == dupe_map_.end()) { | 160 if (it == dupe_map_.end()) { |
152 // First for this ssrc, just insert. | 161 // First for this ssrc, just insert. |
(...skipping 159 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... | |
312 int64_t PacedSender::QueueInMs() const { | 321 int64_t PacedSender::QueueInMs() const { |
313 CriticalSectionScoped cs(critsect_.get()); | 322 CriticalSectionScoped cs(critsect_.get()); |
314 | 323 |
315 int64_t oldest_packet = packets_->OldestEnqueueTimeMs(); | 324 int64_t oldest_packet = packets_->OldestEnqueueTimeMs(); |
316 if (oldest_packet == 0) | 325 if (oldest_packet == 0) |
317 return 0; | 326 return 0; |
318 | 327 |
319 return clock_->TimeInMilliseconds() - oldest_packet; | 328 return clock_->TimeInMilliseconds() - oldest_packet; |
320 } | 329 } |
321 | 330 |
331 int64_t PacedSender::AverageQueueTimeMs() { | |
332 CriticalSectionScoped cs(critsect_.get()); | |
333 packets_->UpdateQueueTime(clock_->TimeInMilliseconds()); | |
334 return packets_->AverageQueueTimeMs(); | |
335 } | |
336 | |
322 int64_t PacedSender::TimeUntilNextProcess() { | 337 int64_t PacedSender::TimeUntilNextProcess() { |
323 CriticalSectionScoped cs(critsect_.get()); | 338 CriticalSectionScoped cs(critsect_.get()); |
324 if (prober_->IsProbing()) { | 339 if (prober_->IsProbing()) { |
325 int64_t ret = prober_->TimeUntilNextProbe(clock_->TimeInMilliseconds()); | 340 int64_t ret = prober_->TimeUntilNextProbe(clock_->TimeInMilliseconds()); |
326 if (ret >= 0) | 341 if (ret >= 0) |
327 return ret; | 342 return ret; |
328 } | 343 } |
329 int64_t elapsed_time_us = clock_->TimeInMicroseconds() - time_last_update_us_; | 344 int64_t elapsed_time_us = clock_->TimeInMicroseconds() - time_last_update_us_; |
330 int64_t elapsed_time_ms = (elapsed_time_us + 500) / 1000; | 345 int64_t elapsed_time_ms = (elapsed_time_us + 500) / 1000; |
331 return std::max<int64_t>(kMinPacketLimitMs - elapsed_time_ms, 0); | 346 return std::max<int64_t>(kMinPacketLimitMs - elapsed_time_ms, 0); |
332 } | 347 } |
333 | 348 |
334 int32_t PacedSender::Process() { | 349 int32_t PacedSender::Process() { |
335 int64_t now_us = clock_->TimeInMicroseconds(); | 350 int64_t now_us = clock_->TimeInMicroseconds(); |
336 CriticalSectionScoped cs(critsect_.get()); | 351 CriticalSectionScoped cs(critsect_.get()); |
337 int64_t elapsed_time_ms = (now_us - time_last_update_us_ + 500) / 1000; | 352 int64_t elapsed_time_ms = (now_us - time_last_update_us_ + 500) / 1000; |
338 time_last_update_us_ = now_us; | 353 time_last_update_us_ = now_us; |
339 if (paused_) | 354 if (paused_) |
340 return 0; | 355 return 0; |
341 int target_bitrate_kbps = max_bitrate_kbps_; | 356 int target_bitrate_kbps = max_bitrate_kbps_; |
342 if (elapsed_time_ms > 0) { | 357 if (elapsed_time_ms > 0) { |
343 size_t queue_size_bytes = packets_->SizeInBytes(); | 358 size_t queue_size_bytes = packets_->SizeInBytes(); |
344 if (queue_size_bytes > 0) { | 359 if (queue_size_bytes > 0) { |
345 // Assuming equal size packets and input/output rate, the average packet | 360 // Assuming equal size packets and input/output rate, the average packet |
346 // has avg_time_left_ms left to get queue_size_bytes out of the queue, if | 361 // has avg_time_left_ms left to get queue_size_bytes out of the queue, if |
347 // time constraint shall be met. Determine bitrate needed for that. | 362 // time constraint shall be met. Determine bitrate needed for that. |
363 packets_->UpdateQueueTime(clock_->TimeInMilliseconds()); | |
348 int64_t avg_time_left_ms = std::max<int64_t>( | 364 int64_t avg_time_left_ms = std::max<int64_t>( |
349 1, kMaxQueueLengthMs - packets_->AverageQueueTimeMs()); | 365 1, kMaxQueueLengthMs - packets_->AverageQueueTimeMs()); |
350 int min_bitrate_needed_kbps = | 366 int min_bitrate_needed_kbps = |
351 static_cast<int>(queue_size_bytes * 8 / avg_time_left_ms); | 367 static_cast<int>(queue_size_bytes * 8 / avg_time_left_ms); |
352 if (min_bitrate_needed_kbps > target_bitrate_kbps) | 368 if (min_bitrate_needed_kbps > target_bitrate_kbps) |
353 target_bitrate_kbps = min_bitrate_needed_kbps; | 369 target_bitrate_kbps = min_bitrate_needed_kbps; |
354 } | 370 } |
355 | 371 |
356 media_budget_->set_target_rate_kbps(target_bitrate_kbps); | 372 media_budget_->set_target_rate_kbps(target_bitrate_kbps); |
357 | 373 |
(...skipping 63 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... | |
421 media_budget_->UseBudget(bytes_sent); | 437 media_budget_->UseBudget(bytes_sent); |
422 padding_budget_->UseBudget(bytes_sent); | 438 padding_budget_->UseBudget(bytes_sent); |
423 } | 439 } |
424 } | 440 } |
425 | 441 |
426 void PacedSender::UpdateBytesPerInterval(int64_t delta_time_ms) { | 442 void PacedSender::UpdateBytesPerInterval(int64_t delta_time_ms) { |
427 media_budget_->IncreaseBudget(delta_time_ms); | 443 media_budget_->IncreaseBudget(delta_time_ms); |
428 padding_budget_->IncreaseBudget(delta_time_ms); | 444 padding_budget_->IncreaseBudget(delta_time_ms); |
429 } | 445 } |
430 } // namespace webrtc | 446 } // namespace webrtc |
OLD | NEW |