1 /*
2 * Copyright (c) 2015 The WebRTC project authors. All Rights Reserved.
3 *
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
6 * tree. An additional intellectual property rights grant can be found
7 * in the file PATENTS. All contributing project authors may
8 * be found in the AUTHORS file in the root of the source tree.
9 */
10
11 #include "logging/rtc_event_log/rtc_event_log_impl.h"
12
13 #include <functional>
14 #include <limits>
15 #include <memory>
16 #include <utility>
17 #include <vector>
18
19 #include "absl/strings/string_view.h"
20 #include "absl/types/optional.h"
21 #include "api/task_queue/task_queue_base.h"
22 #include "api/units/time_delta.h"
23 #include "logging/rtc_event_log/encoder/rtc_event_log_encoder_legacy.h"
24 #include "logging/rtc_event_log/encoder/rtc_event_log_encoder_new_format.h"
25 #include "rtc_base/checks.h"
26 #include "rtc_base/event.h"
27 #include "rtc_base/logging.h"
28 #include "rtc_base/numerics/safe_conversions.h"
29 #include "rtc_base/numerics/safe_minmax.h"
30 #include "rtc_base/time_utils.h"
31
32 namespace webrtc {
33 namespace {
34 constexpr size_t kMaxEventsInHistory = 10000;
35 // The config-history is supposed to be unbounded, but needs to have some bound
36 // to prevent an attack via unreasonable memory use.
37 constexpr size_t kMaxEventsInConfigHistory = 1000;
38
CreateEncoder(RtcEventLog::EncodingType type)39 std::unique_ptr<RtcEventLogEncoder> CreateEncoder(
40 RtcEventLog::EncodingType type) {
41 switch (type) {
42 case RtcEventLog::EncodingType::Legacy:
43 RTC_DLOG(LS_INFO) << "Creating legacy encoder for RTC event log.";
44 return std::make_unique<RtcEventLogEncoderLegacy>();
45 case RtcEventLog::EncodingType::NewFormat:
46 RTC_DLOG(LS_INFO) << "Creating new format encoder for RTC event log.";
47 return std::make_unique<RtcEventLogEncoderNewFormat>();
48 default:
49 RTC_LOG(LS_ERROR) << "Unknown RtcEventLog encoder type (" << int(type)
50 << ")";
51 RTC_DCHECK_NOTREACHED();
52 return std::unique_ptr<RtcEventLogEncoder>(nullptr);
53 }
54 }
55 } // namespace
56
RtcEventLogImpl(RtcEventLog::EncodingType encoding_type,TaskQueueFactory * task_queue_factory)57 RtcEventLogImpl::RtcEventLogImpl(RtcEventLog::EncodingType encoding_type,
58 TaskQueueFactory* task_queue_factory)
59 : event_encoder_(CreateEncoder(encoding_type)),
60 num_config_events_written_(0),
61 last_output_ms_(rtc::TimeMillis()),
62 output_scheduled_(false),
63 logging_state_started_(false),
64 task_queue_(
65 std::make_unique<rtc::TaskQueue>(task_queue_factory->CreateTaskQueue(
66 "rtc_event_log",
67 TaskQueueFactory::Priority::NORMAL))) {}
68
~RtcEventLogImpl()69 RtcEventLogImpl::~RtcEventLogImpl() {
70 // If we're logging to the output, this will stop that. Blocking function.
71 if (logging_state_started_) {
72 logging_state_checker_.Detach();
73 StopLogging();
74 }
75
76 // We want to block on any executing task by invoking ~TaskQueue() before
77 // we set unique_ptr's internal pointer to null.
78 rtc::TaskQueue* tq = task_queue_.get();
79 delete tq;
80 task_queue_.release();
81 }
82
StartLogging(std::unique_ptr<RtcEventLogOutput> output,int64_t output_period_ms)83 bool RtcEventLogImpl::StartLogging(std::unique_ptr<RtcEventLogOutput> output,
84 int64_t output_period_ms) {
85 RTC_DCHECK(output_period_ms == kImmediateOutput || output_period_ms > 0);
86
87 if (!output->IsActive()) {
88 // TODO(eladalon): We may want to remove the IsActive method. Otherwise
89 // we probably want to be consistent and terminate any existing output.
90 return false;
91 }
92
93 const int64_t timestamp_us = rtc::TimeMillis() * 1000;
94 const int64_t utc_time_us = rtc::TimeUTCMillis() * 1000;
95 RTC_LOG(LS_INFO) << "Starting WebRTC event log. (Timestamp, UTC) = ("
96 << timestamp_us << ", " << utc_time_us << ").";
97
98 RTC_DCHECK_RUN_ON(&logging_state_checker_);
99 logging_state_started_ = true;
100 // Binding to `this` is safe because `this` outlives the `task_queue_`.
101 task_queue_->PostTask([this, output_period_ms, timestamp_us, utc_time_us,
102 output = std::move(output)]() mutable {
103 RTC_DCHECK_RUN_ON(task_queue_.get());
104 RTC_DCHECK(output->IsActive());
105 output_period_ms_ = output_period_ms;
106 event_output_ = std::move(output);
107 num_config_events_written_ = 0;
108 WriteToOutput(event_encoder_->EncodeLogStart(timestamp_us, utc_time_us));
109 LogEventsFromMemoryToOutput();
110 });
111
112 return true;
113 }
114
StopLogging()115 void RtcEventLogImpl::StopLogging() {
116 RTC_DLOG(LS_INFO) << "Stopping WebRTC event log.";
117 // TODO(danilchap): Do not block current thread waiting on the task queue.
118 // It might work for now, for current callers, but disallows caller to share
119 // threads with the `task_queue_`.
120 rtc::Event output_stopped;
121 StopLogging([&output_stopped]() { output_stopped.Set(); });
122 output_stopped.Wait(rtc::Event::kForever);
123
124 RTC_DLOG(LS_INFO) << "WebRTC event log successfully stopped.";
125 }
126
StopLogging(std::function<void ()> callback)127 void RtcEventLogImpl::StopLogging(std::function<void()> callback) {
128 RTC_DCHECK_RUN_ON(&logging_state_checker_);
129 logging_state_started_ = false;
130 task_queue_->PostTask([this, callback] {
131 RTC_DCHECK_RUN_ON(task_queue_.get());
132 if (event_output_) {
133 RTC_DCHECK(event_output_->IsActive());
134 LogEventsFromMemoryToOutput();
135 }
136 StopLoggingInternal();
137 callback();
138 });
139 }
140
Log(std::unique_ptr<RtcEvent> event)141 void RtcEventLogImpl::Log(std::unique_ptr<RtcEvent> event) {
142 RTC_CHECK(event);
143
144 // Binding to `this` is safe because `this` outlives the `task_queue_`.
145 task_queue_->PostTask([this, event = std::move(event)]() mutable {
146 RTC_DCHECK_RUN_ON(task_queue_.get());
147 LogToMemory(std::move(event));
148 if (event_output_)
149 ScheduleOutput();
150 });
151 }
152
ScheduleOutput()153 void RtcEventLogImpl::ScheduleOutput() {
154 RTC_DCHECK(event_output_ && event_output_->IsActive());
155 if (history_.size() >= kMaxEventsInHistory) {
156 // We have to emergency drain the buffer. We can't wait for the scheduled
157 // output task because there might be other event incoming before that.
158 LogEventsFromMemoryToOutput();
159 return;
160 }
161
162 RTC_DCHECK(output_period_ms_.has_value());
163 if (*output_period_ms_ == kImmediateOutput) {
164 // We are already on the `task_queue_` so there is no reason to post a task
165 // if we want to output immediately.
166 LogEventsFromMemoryToOutput();
167 return;
168 }
169
170 if (!output_scheduled_) {
171 output_scheduled_ = true;
172 // Binding to `this` is safe because `this` outlives the `task_queue_`.
173 auto output_task = [this]() {
174 RTC_DCHECK_RUN_ON(task_queue_.get());
175 if (event_output_) {
176 RTC_DCHECK(event_output_->IsActive());
177 LogEventsFromMemoryToOutput();
178 }
179 output_scheduled_ = false;
180 };
181 const int64_t now_ms = rtc::TimeMillis();
182 const int64_t time_since_output_ms = now_ms - last_output_ms_;
183 const uint32_t delay = rtc::SafeClamp(
184 *output_period_ms_ - time_since_output_ms, 0, *output_period_ms_);
185 task_queue_->PostDelayedTask(std::move(output_task),
186 TimeDelta::Millis(delay));
187 }
188 }
189
LogToMemory(std::unique_ptr<RtcEvent> event)190 void RtcEventLogImpl::LogToMemory(std::unique_ptr<RtcEvent> event) {
191 std::deque<std::unique_ptr<RtcEvent>>& container =
192 event->IsConfigEvent() ? config_history_ : history_;
193 const size_t container_max_size =
194 event->IsConfigEvent() ? kMaxEventsInConfigHistory : kMaxEventsInHistory;
195
196 if (container.size() >= container_max_size) {
197 RTC_DCHECK(!event_output_); // Shouldn't lose events if we have an output.
198 container.pop_front();
199 }
200 container.push_back(std::move(event));
201 }
202
LogEventsFromMemoryToOutput()203 void RtcEventLogImpl::LogEventsFromMemoryToOutput() {
204 RTC_DCHECK(event_output_ && event_output_->IsActive());
205 last_output_ms_ = rtc::TimeMillis();
206
207 // Serialize all stream configurations that haven't already been written to
208 // this output. `num_config_events_written_` is used to track which configs we
209 // have already written. (Note that the config may have been written to
210 // previous outputs; configs are not discarded.)
211 std::string encoded_configs;
212 RTC_DCHECK_LE(num_config_events_written_, config_history_.size());
213 if (num_config_events_written_ < config_history_.size()) {
214 const auto begin = config_history_.begin() + num_config_events_written_;
215 const auto end = config_history_.end();
216 encoded_configs = event_encoder_->EncodeBatch(begin, end);
217 num_config_events_written_ = config_history_.size();
218 }
219
220 // Serialize the events in the event queue. Note that the write may fail,
221 // for example if we are writing to a file and have reached the maximum limit.
222 // We don't get any feedback if this happens, so we still remove the events
223 // from the event log history. This is normally not a problem, but if another
224 // log is started immediately after the first one becomes full, then one
225 // cannot rely on the second log to contain everything that isn't in the first
226 // log; one batch of events might be missing.
227 std::string encoded_history =
228 event_encoder_->EncodeBatch(history_.begin(), history_.end());
229 history_.clear();
230
231 WriteConfigsAndHistoryToOutput(encoded_configs, encoded_history);
232 }
233
WriteConfigsAndHistoryToOutput(absl::string_view encoded_configs,absl::string_view encoded_history)234 void RtcEventLogImpl::WriteConfigsAndHistoryToOutput(
235 absl::string_view encoded_configs,
236 absl::string_view encoded_history) {
237 // This function is used to merge the strings instead of calling the output
238 // object twice with small strings. The function also avoids copying any
239 // strings in the typical case where there are no config events.
240 if (encoded_configs.empty()) {
241 WriteToOutput(encoded_history); // Typical case.
242 } else if (encoded_history.empty()) {
243 WriteToOutput(encoded_configs); // Very unusual case.
244 } else {
245 std::string s;
246 s.reserve(encoded_configs.size() + encoded_history.size());
247 s.append(encoded_configs.data(), encoded_configs.size());
248 s.append(encoded_history.data(), encoded_history.size());
249 WriteToOutput(s);
250 }
251 }
252
StopOutput()253 void RtcEventLogImpl::StopOutput() {
254 event_output_.reset();
255 }
256
StopLoggingInternal()257 void RtcEventLogImpl::StopLoggingInternal() {
258 if (event_output_) {
259 RTC_DCHECK(event_output_->IsActive());
260 const int64_t timestamp_us = rtc::TimeMillis() * 1000;
261 event_output_->Write(event_encoder_->EncodeLogEnd(timestamp_us));
262 }
263 StopOutput();
264 }
265
WriteToOutput(absl::string_view output_string)266 void RtcEventLogImpl::WriteToOutput(absl::string_view output_string) {
267 RTC_DCHECK(event_output_ && event_output_->IsActive());
268 if (!event_output_->Write(output_string)) {
269 RTC_LOG(LS_ERROR) << "Failed to write RTC event to output.";
270 // The first failure closes the output.
271 RTC_DCHECK(!event_output_->IsActive());
272 StopOutput(); // Clean-up.
273 return;
274 }
275 }
276
277 } // namespace webrtc
278