1 // Copyright 2014 The Chromium Authors. All rights reserved.
2 // Use of this source code is governed by a BSD-style license that can be
3 // found in the LICENSE file.
4
5 #include "media/cast/logging/encoding_event_subscriber.h"
6
7 #include <cstring>
8 #include <utility>
9
10 #include "base/logging.h"
11 #include "media/cast/logging/proto/proto_utils.h"
12
13 using google::protobuf::RepeatedPtrField;
14 using media::cast::proto::AggregatedFrameEvent;
15 using media::cast::proto::AggregatedPacketEvent;
16 using media::cast::proto::BasePacketEvent;
17 using media::cast::proto::LogMetadata;
18
19 namespace {
20
21 // A size limit on maps to keep lookups fast.
22 const size_t kMaxMapSize = 200;
23
24 // The smallest (oredered by RTP timestamp) |kNumMapEntriesToTransfer| entries
25 // will be moved when the map size reaches |kMaxMapSize|.
26 // Must be smaller than |kMaxMapSize|.
27 const size_t kNumMapEntriesToTransfer = 100;
28
29 template <typename ProtoPtr>
IsRtpTimestampLessThan(const ProtoPtr & lhs,const ProtoPtr & rhs)30 bool IsRtpTimestampLessThan(const ProtoPtr& lhs, const ProtoPtr& rhs) {
31 return lhs->relative_rtp_timestamp() < rhs->relative_rtp_timestamp();
32 }
33
GetNewBasePacketEvent(AggregatedPacketEvent * event_proto,int packet_id,int size)34 BasePacketEvent* GetNewBasePacketEvent(AggregatedPacketEvent* event_proto,
35 int packet_id, int size) {
36 BasePacketEvent* base = event_proto->add_base_packet_event();
37 base->set_packet_id(packet_id);
38 base->set_size(size);
39 return base;
40 }
41
42 }
43
44 namespace media {
45 namespace cast {
46
EncodingEventSubscriber(EventMediaType event_media_type,size_t max_frames)47 EncodingEventSubscriber::EncodingEventSubscriber(
48 EventMediaType event_media_type,
49 size_t max_frames)
50 : event_media_type_(event_media_type),
51 max_frames_(max_frames),
52 frame_event_storage_index_(0),
53 packet_event_storage_index_(0),
54 seen_first_rtp_timestamp_(false),
55 first_rtp_timestamp_(0u) {}
56
~EncodingEventSubscriber()57 EncodingEventSubscriber::~EncodingEventSubscriber() {
58 DCHECK(thread_checker_.CalledOnValidThread());
59 }
60
OnReceiveFrameEvent(const FrameEvent & frame_event)61 void EncodingEventSubscriber::OnReceiveFrameEvent(
62 const FrameEvent& frame_event) {
63 DCHECK(thread_checker_.CalledOnValidThread());
64
65 if (event_media_type_ != frame_event.media_type)
66 return;
67
68 RtpTimestamp relative_rtp_timestamp =
69 GetRelativeRtpTimestamp(frame_event.rtp_timestamp);
70 FrameEventMap::iterator it = frame_event_map_.find(relative_rtp_timestamp);
71 linked_ptr<AggregatedFrameEvent> event_proto;
72
73 // Look up existing entry. If not found, create a new entry and add to map.
74 if (it == frame_event_map_.end()) {
75 event_proto.reset(new AggregatedFrameEvent);
76 event_proto->set_relative_rtp_timestamp(relative_rtp_timestamp);
77 frame_event_map_.insert(
78 std::make_pair(relative_rtp_timestamp, event_proto));
79 } else {
80 event_proto = it->second;
81 if (event_proto->event_type_size() >= kMaxEventsPerProto) {
82 DVLOG(2) << "Too many events in frame " << frame_event.rtp_timestamp
83 << ". Using new frame event proto.";
84 AddFrameEventToStorage(event_proto);
85 event_proto.reset(new AggregatedFrameEvent);
86 event_proto->set_relative_rtp_timestamp(relative_rtp_timestamp);
87 it->second = event_proto;
88 }
89 }
90
91 event_proto->add_event_type(ToProtoEventType(frame_event.type));
92 event_proto->add_event_timestamp_ms(
93 (frame_event.timestamp - base::TimeTicks()).InMilliseconds());
94
95 if (frame_event.type == FRAME_ENCODED) {
96 event_proto->set_encoded_frame_size(frame_event.size);
97 if (frame_event.media_type == VIDEO_EVENT) {
98 event_proto->set_encoded_frame_size(frame_event.size);
99 event_proto->set_key_frame(frame_event.key_frame);
100 event_proto->set_target_bitrate(frame_event.target_bitrate);
101 }
102 } else if (frame_event.type == FRAME_PLAYOUT) {
103 event_proto->set_delay_millis(frame_event.delay_delta.InMilliseconds());
104 }
105
106 if (frame_event_map_.size() > kMaxMapSize)
107 TransferFrameEvents(kNumMapEntriesToTransfer);
108
109 DCHECK(frame_event_map_.size() <= kMaxMapSize);
110 DCHECK(frame_event_storage_.size() <= max_frames_);
111 }
112
OnReceivePacketEvent(const PacketEvent & packet_event)113 void EncodingEventSubscriber::OnReceivePacketEvent(
114 const PacketEvent& packet_event) {
115 DCHECK(thread_checker_.CalledOnValidThread());
116
117 if (event_media_type_ != packet_event.media_type)
118 return;
119
120 RtpTimestamp relative_rtp_timestamp =
121 GetRelativeRtpTimestamp(packet_event.rtp_timestamp);
122 PacketEventMap::iterator it =
123 packet_event_map_.find(relative_rtp_timestamp);
124 linked_ptr<AggregatedPacketEvent> event_proto;
125 BasePacketEvent* base_packet_event_proto = NULL;
126
127 // Look up existing entry. If not found, create a new entry and add to map.
128 if (it == packet_event_map_.end()) {
129 event_proto.reset(new AggregatedPacketEvent);
130 event_proto->set_relative_rtp_timestamp(relative_rtp_timestamp);
131 packet_event_map_.insert(
132 std::make_pair(relative_rtp_timestamp, event_proto));
133 base_packet_event_proto = GetNewBasePacketEvent(
134 event_proto.get(), packet_event.packet_id, packet_event.size);
135 } else {
136 // Found existing entry, now look up existing BasePacketEvent using packet
137 // ID. If not found, create a new entry and add to proto.
138 event_proto = it->second;
139 RepeatedPtrField<BasePacketEvent>* field =
140 event_proto->mutable_base_packet_event();
141 for (RepeatedPtrField<BasePacketEvent>::pointer_iterator base_it =
142 field->pointer_begin();
143 base_it != field->pointer_end();
144 ++base_it) {
145 if ((*base_it)->packet_id() == packet_event.packet_id) {
146 base_packet_event_proto = *base_it;
147 break;
148 }
149 }
150 if (!base_packet_event_proto) {
151 if (event_proto->base_packet_event_size() >= kMaxPacketsPerFrame) {
152 DVLOG(3) << "Too many packets in AggregatedPacketEvent "
153 << packet_event.rtp_timestamp << ". "
154 << "Using new packet event proto.";
155 AddPacketEventToStorage(event_proto);
156 event_proto.reset(new AggregatedPacketEvent);
157 event_proto->set_relative_rtp_timestamp(relative_rtp_timestamp);
158 it->second = event_proto;
159 }
160
161 base_packet_event_proto = GetNewBasePacketEvent(
162 event_proto.get(), packet_event.packet_id, packet_event.size);
163 } else if (base_packet_event_proto->event_type_size() >=
164 kMaxEventsPerProto) {
165 DVLOG(3) << "Too many events in packet "
166 << packet_event.rtp_timestamp << ", "
167 << packet_event.packet_id << ". Using new packet event proto.";
168 AddPacketEventToStorage(event_proto);
169 event_proto.reset(new AggregatedPacketEvent);
170 event_proto->set_relative_rtp_timestamp(relative_rtp_timestamp);
171 it->second = event_proto;
172 base_packet_event_proto = GetNewBasePacketEvent(
173 event_proto.get(), packet_event.packet_id, packet_event.size);
174 }
175 }
176
177 base_packet_event_proto->add_event_type(
178 ToProtoEventType(packet_event.type));
179 base_packet_event_proto->add_event_timestamp_ms(
180 (packet_event.timestamp - base::TimeTicks()).InMilliseconds());
181
182 // |base_packet_event_proto| could have been created with a receiver event
183 // which does not have the packet size and we would need to overwrite it when
184 // we see a sender event, which does have the packet size.
185 if (packet_event.size > 0) {
186 base_packet_event_proto->set_size(packet_event.size);
187 }
188
189 if (packet_event_map_.size() > kMaxMapSize)
190 TransferPacketEvents(kNumMapEntriesToTransfer);
191
192 DCHECK(packet_event_map_.size() <= kMaxMapSize);
193 DCHECK(packet_event_storage_.size() <= max_frames_);
194 }
195
GetEventsAndReset(LogMetadata * metadata,FrameEventList * frame_events,PacketEventList * packet_events)196 void EncodingEventSubscriber::GetEventsAndReset(LogMetadata* metadata,
197 FrameEventList* frame_events, PacketEventList* packet_events) {
198 DCHECK(thread_checker_.CalledOnValidThread());
199
200 // Flush all events.
201 TransferFrameEvents(frame_event_map_.size());
202 TransferPacketEvents(packet_event_map_.size());
203 std::sort(frame_event_storage_.begin(), frame_event_storage_.end(),
204 &IsRtpTimestampLessThan<linked_ptr<AggregatedFrameEvent> >);
205 std::sort(packet_event_storage_.begin(), packet_event_storage_.end(),
206 &IsRtpTimestampLessThan<linked_ptr<AggregatedPacketEvent> >);
207
208 metadata->set_is_audio(event_media_type_ == AUDIO_EVENT);
209 metadata->set_first_rtp_timestamp(first_rtp_timestamp_);
210 metadata->set_num_frame_events(frame_event_storage_.size());
211 metadata->set_num_packet_events(packet_event_storage_.size());
212 metadata->set_reference_timestamp_ms_at_unix_epoch(
213 (base::TimeTicks::UnixEpoch() - base::TimeTicks()).InMilliseconds());
214 frame_events->swap(frame_event_storage_);
215 packet_events->swap(packet_event_storage_);
216 Reset();
217 }
218
TransferFrameEvents(size_t max_num_entries)219 void EncodingEventSubscriber::TransferFrameEvents(size_t max_num_entries) {
220 DCHECK(frame_event_map_.size() >= max_num_entries);
221
222 FrameEventMap::iterator it = frame_event_map_.begin();
223 for (size_t i = 0;
224 i < max_num_entries && it != frame_event_map_.end();
225 i++, ++it) {
226 AddFrameEventToStorage(it->second);
227 }
228
229 frame_event_map_.erase(frame_event_map_.begin(), it);
230 }
231
TransferPacketEvents(size_t max_num_entries)232 void EncodingEventSubscriber::TransferPacketEvents(size_t max_num_entries) {
233 PacketEventMap::iterator it = packet_event_map_.begin();
234 for (size_t i = 0;
235 i < max_num_entries && it != packet_event_map_.end();
236 i++, ++it) {
237 AddPacketEventToStorage(it->second);
238 }
239
240 packet_event_map_.erase(packet_event_map_.begin(), it);
241 }
242
AddFrameEventToStorage(const linked_ptr<AggregatedFrameEvent> & frame_event_proto)243 void EncodingEventSubscriber::AddFrameEventToStorage(
244 const linked_ptr<AggregatedFrameEvent>& frame_event_proto) {
245 if (frame_event_storage_.size() >= max_frames_) {
246 frame_event_storage_[frame_event_storage_index_] = frame_event_proto;
247 } else {
248 frame_event_storage_.push_back(frame_event_proto);
249 }
250
251 frame_event_storage_index_ = (frame_event_storage_index_ + 1) % max_frames_;
252 }
253
AddPacketEventToStorage(const linked_ptr<AggregatedPacketEvent> & packet_event_proto)254 void EncodingEventSubscriber::AddPacketEventToStorage(
255 const linked_ptr<AggregatedPacketEvent>& packet_event_proto) {
256 if (packet_event_storage_.size() >= max_frames_)
257 packet_event_storage_[packet_event_storage_index_] = packet_event_proto;
258 else
259 packet_event_storage_.push_back(packet_event_proto);
260
261 packet_event_storage_index_ = (packet_event_storage_index_ + 1) % max_frames_;
262 }
263
GetRelativeRtpTimestamp(RtpTimestamp rtp_timestamp)264 RtpTimestamp EncodingEventSubscriber::GetRelativeRtpTimestamp(
265 RtpTimestamp rtp_timestamp) {
266 if (!seen_first_rtp_timestamp_) {
267 seen_first_rtp_timestamp_ = true;
268 first_rtp_timestamp_ = rtp_timestamp;
269 }
270
271 return rtp_timestamp - first_rtp_timestamp_;
272 }
273
Reset()274 void EncodingEventSubscriber::Reset() {
275 frame_event_map_.clear();
276 frame_event_storage_.clear();
277 frame_event_storage_index_ = 0;
278 packet_event_map_.clear();
279 packet_event_storage_.clear();
280 packet_event_storage_index_ = 0;
281 seen_first_rtp_timestamp_ = false;
282 first_rtp_timestamp_ = 0u;
283 }
284
285 } // namespace cast
286 } // namespace media
287