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 "chrome/browser/media/webrtc/webrtc_rtp_dump_writer.h"
6
7 #include <string.h>
8
9 #include "base/big_endian.h"
10 #include "base/bind.h"
11 #include "base/files/file_util.h"
12 #include "base/logging.h"
13 #include "base/stl_util.h"
14 #include "base/task/post_task.h"
15 #include "base/task/thread_pool.h"
16 #include "content/public/browser/browser_thread.h"
17 #include "third_party/zlib/zlib.h"
18
19 namespace {
20
21 static const size_t kMinimumGzipOutputBufferSize = 256; // In bytes.
22
23 const unsigned char kRtpDumpFileHeaderFirstLine[] = "#!rtpplay1.0 0.0.0.0/0\n";
24 static const size_t kRtpDumpFileHeaderSize = 16; // In bytes.
25
26 // A helper for writing the header of the dump file.
WriteRtpDumpFileHeaderBigEndian(base::TimeTicks start,std::vector<uint8_t> * output)27 void WriteRtpDumpFileHeaderBigEndian(base::TimeTicks start,
28 std::vector<uint8_t>* output) {
29 size_t buffer_start_pos = output->size();
30 output->resize(output->size() + kRtpDumpFileHeaderSize);
31
32 char* buffer = reinterpret_cast<char*>(&(*output)[buffer_start_pos]);
33
34 base::TimeDelta delta = start - base::TimeTicks();
35 uint32_t start_sec = delta.InSeconds();
36 base::WriteBigEndian(buffer, start_sec);
37 buffer += sizeof(start_sec);
38
39 uint32_t start_usec =
40 delta.InMilliseconds() * base::Time::kMicrosecondsPerMillisecond;
41 base::WriteBigEndian(buffer, start_usec);
42 buffer += sizeof(start_usec);
43
44 // Network source, always 0.
45 base::WriteBigEndian(buffer, uint32_t(0));
46 buffer += sizeof(uint32_t);
47
48 // UDP port, always 0.
49 base::WriteBigEndian(buffer, uint16_t(0));
50 buffer += sizeof(uint16_t);
51
52 // 2 bytes padding.
53 base::WriteBigEndian(buffer, uint16_t(0));
54 }
55
56 // The header size for each packet dump.
57 static const size_t kPacketDumpHeaderSize = 8; // In bytes.
58
59 // A helper for writing the header for each packet dump.
60 // |start| is the time when the recording is started.
61 // |dump_length| is the length of the packet dump including this header.
62 // |packet_length| is the length of the RTP packet header.
WritePacketDumpHeaderBigEndian(const base::TimeTicks & start,uint16_t dump_length,uint16_t packet_length,std::vector<uint8_t> * output)63 void WritePacketDumpHeaderBigEndian(const base::TimeTicks& start,
64 uint16_t dump_length,
65 uint16_t packet_length,
66 std::vector<uint8_t>* output) {
67 size_t buffer_start_pos = output->size();
68 output->resize(output->size() + kPacketDumpHeaderSize);
69
70 char* buffer = reinterpret_cast<char*>(&(*output)[buffer_start_pos]);
71
72 base::WriteBigEndian(buffer, dump_length);
73 buffer += sizeof(dump_length);
74
75 base::WriteBigEndian(buffer, packet_length);
76 buffer += sizeof(packet_length);
77
78 uint32_t elapsed =
79 static_cast<uint32_t>((base::TimeTicks::Now() - start).InMilliseconds());
80 base::WriteBigEndian(buffer, elapsed);
81 }
82
83 // Append |src_len| bytes from |src| to |dest|.
AppendToBuffer(const uint8_t * src,size_t src_len,std::vector<uint8_t> * dest)84 void AppendToBuffer(const uint8_t* src,
85 size_t src_len,
86 std::vector<uint8_t>* dest) {
87 size_t old_dest_size = dest->size();
88 dest->resize(old_dest_size + src_len);
89 memcpy(&(*dest)[old_dest_size], src, src_len);
90 }
91
92 } // namespace
93
94 // This class runs on the backround task runner, compresses and writes the
95 // dump buffer to disk.
96 class WebRtcRtpDumpWriter::FileWorker {
97 public:
FileWorker(const base::FilePath & dump_path)98 explicit FileWorker(const base::FilePath& dump_path) : dump_path_(dump_path) {
99 DETACH_FROM_SEQUENCE(sequence_checker_);
100
101 memset(&stream_, 0, sizeof(stream_));
102 int result = deflateInit2(&stream_,
103 Z_DEFAULT_COMPRESSION,
104 Z_DEFLATED,
105 // windowBits = 15 is default, 16 is added to
106 // produce a gzip header + trailer.
107 15 + 16,
108 8, // memLevel = 8 is default.
109 Z_DEFAULT_STRATEGY);
110 DCHECK_EQ(Z_OK, result);
111 }
112
~FileWorker()113 ~FileWorker() {
114 DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
115
116 // Makes sure all allocations are freed.
117 deflateEnd(&stream_);
118 }
119
120 // Compresses the data in |buffer| and write to the dump file. If |end_stream|
121 // is true, the compression stream will be ended and the dump file cannot be
122 // written to any more.
CompressAndWriteToFileOnFileThread(std::unique_ptr<std::vector<uint8_t>> buffer,bool end_stream,FlushResult * result,size_t * bytes_written)123 void CompressAndWriteToFileOnFileThread(
124 std::unique_ptr<std::vector<uint8_t>> buffer,
125 bool end_stream,
126 FlushResult* result,
127 size_t* bytes_written) {
128 DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
129
130 // This is called either when the in-memory buffer is full or the dump
131 // should be ended.
132 DCHECK(!buffer->empty() || end_stream);
133
134 *result = FLUSH_RESULT_SUCCESS;
135 *bytes_written = 0;
136
137 // There may be nothing to compress/write if there is no RTP packet since
138 // the last flush.
139 if (!buffer->empty()) {
140 *bytes_written = CompressAndWriteBufferToFile(buffer.get(), result);
141 } else if (!base::PathExists(dump_path_)) {
142 // If the dump does not exist, it means there is no RTP packet recorded.
143 // Return FLUSH_RESULT_NO_DATA to indicate no dump file created.
144 *result = FLUSH_RESULT_NO_DATA;
145 }
146
147 if (end_stream && !EndDumpFile())
148 *result = FLUSH_RESULT_FAILURE;
149 }
150
151 private:
152 // Helper for CompressAndWriteToFileOnFileThread to compress and write one
153 // dump.
CompressAndWriteBufferToFile(std::vector<uint8_t> * buffer,FlushResult * result)154 size_t CompressAndWriteBufferToFile(std::vector<uint8_t>* buffer,
155 FlushResult* result) {
156 DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
157 DCHECK(buffer->size());
158
159 *result = FLUSH_RESULT_SUCCESS;
160
161 std::vector<uint8_t> compressed_buffer;
162 if (!Compress(buffer, &compressed_buffer)) {
163 DVLOG(2) << "Compressing buffer failed.";
164 *result = FLUSH_RESULT_FAILURE;
165 return 0;
166 }
167
168 int bytes_written = -1;
169
170 if (base::PathExists(dump_path_)) {
171 bytes_written =
172 base::AppendToFile(dump_path_, reinterpret_cast<const char*>(
173 compressed_buffer.data()),
174 compressed_buffer.size())
175 ? compressed_buffer.size()
176 : -1;
177 } else {
178 bytes_written = base::WriteFile(
179 dump_path_,
180 reinterpret_cast<const char*>(&compressed_buffer[0]),
181 compressed_buffer.size());
182 }
183
184 if (bytes_written == -1) {
185 DVLOG(2) << "Writing file failed: " << dump_path_.value();
186 *result = FLUSH_RESULT_FAILURE;
187 return 0;
188 }
189
190 DCHECK_EQ(static_cast<size_t>(bytes_written), compressed_buffer.size());
191 return bytes_written;
192 }
193
194 // Compresses |input| into |output|.
Compress(std::vector<uint8_t> * input,std::vector<uint8_t> * output)195 bool Compress(std::vector<uint8_t>* input, std::vector<uint8_t>* output) {
196 DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
197 int result = Z_OK;
198
199 output->resize(std::max(kMinimumGzipOutputBufferSize, input->size()));
200
201 stream_.next_in = &(*input)[0];
202 stream_.avail_in = input->size();
203 stream_.next_out = &(*output)[0];
204 stream_.avail_out = output->size();
205
206 result = deflate(&stream_, Z_SYNC_FLUSH);
207 DCHECK_EQ(Z_OK, result);
208 DCHECK_EQ(0U, stream_.avail_in);
209
210 output->resize(output->size() - stream_.avail_out);
211
212 stream_.next_in = NULL;
213 stream_.next_out = NULL;
214 stream_.avail_out = 0;
215 return true;
216 }
217
218 // Ends the compression stream and completes the dump file.
EndDumpFile()219 bool EndDumpFile() {
220 DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
221
222 std::vector<uint8_t> output_buffer;
223 output_buffer.resize(kMinimumGzipOutputBufferSize);
224
225 stream_.next_in = NULL;
226 stream_.avail_in = 0;
227 stream_.next_out = &output_buffer[0];
228 stream_.avail_out = output_buffer.size();
229
230 int result = deflate(&stream_, Z_FINISH);
231 DCHECK_EQ(Z_STREAM_END, result);
232
233 result = deflateEnd(&stream_);
234 DCHECK_EQ(Z_OK, result);
235
236 output_buffer.resize(output_buffer.size() - stream_.avail_out);
237
238 memset(&stream_, 0, sizeof(z_stream));
239
240 DCHECK(!output_buffer.empty());
241 return base::AppendToFile(
242 dump_path_, reinterpret_cast<const char*>(output_buffer.data()),
243 output_buffer.size());
244 }
245
246 const base::FilePath dump_path_;
247
248 z_stream stream_;
249
250 SEQUENCE_CHECKER(sequence_checker_);
251
252 DISALLOW_COPY_AND_ASSIGN(FileWorker);
253 };
254
WebRtcRtpDumpWriter(const base::FilePath & incoming_dump_path,const base::FilePath & outgoing_dump_path,size_t max_dump_size,base::RepeatingClosure max_dump_size_reached_callback)255 WebRtcRtpDumpWriter::WebRtcRtpDumpWriter(
256 const base::FilePath& incoming_dump_path,
257 const base::FilePath& outgoing_dump_path,
258 size_t max_dump_size,
259 base::RepeatingClosure max_dump_size_reached_callback)
260 : max_dump_size_(max_dump_size),
261 max_dump_size_reached_callback_(
262 std::move(max_dump_size_reached_callback)),
263 total_dump_size_on_disk_(0),
264 background_task_runner_(base::ThreadPool::CreateSequencedTaskRunner(
265 {base::MayBlock(), base::TaskPriority::BEST_EFFORT})),
266 incoming_file_thread_worker_(new FileWorker(incoming_dump_path)),
267 outgoing_file_thread_worker_(new FileWorker(outgoing_dump_path)) {}
268
~WebRtcRtpDumpWriter()269 WebRtcRtpDumpWriter::~WebRtcRtpDumpWriter() {
270 DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
271
272 bool success = background_task_runner_->DeleteSoon(
273 FROM_HERE, incoming_file_thread_worker_.release());
274 DCHECK(success);
275
276 success = background_task_runner_->DeleteSoon(
277 FROM_HERE, outgoing_file_thread_worker_.release());
278 DCHECK(success);
279 }
280
WriteRtpPacket(const uint8_t * packet_header,size_t header_length,size_t packet_length,bool incoming)281 void WebRtcRtpDumpWriter::WriteRtpPacket(const uint8_t* packet_header,
282 size_t header_length,
283 size_t packet_length,
284 bool incoming) {
285 DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
286
287 static const size_t kMaxInMemoryBufferSize = 65536;
288
289 std::vector<uint8_t>* dest_buffer =
290 incoming ? &incoming_buffer_ : &outgoing_buffer_;
291
292 // We use the capacity of the buffer to indicate if the buffer has been
293 // initialized and if the dump file header has been created.
294 if (!dest_buffer->capacity()) {
295 dest_buffer->reserve(std::min(kMaxInMemoryBufferSize, max_dump_size_));
296
297 start_time_ = base::TimeTicks::Now();
298
299 // Writes the dump file header.
300 AppendToBuffer(kRtpDumpFileHeaderFirstLine,
301 base::size(kRtpDumpFileHeaderFirstLine) - 1, dest_buffer);
302 WriteRtpDumpFileHeaderBigEndian(start_time_, dest_buffer);
303 }
304
305 size_t packet_dump_length = kPacketDumpHeaderSize + header_length;
306
307 // Flushes the buffer to disk if the buffer is full.
308 if (dest_buffer->size() + packet_dump_length > dest_buffer->capacity())
309 FlushBuffer(incoming, false, FlushDoneCallback());
310
311 WritePacketDumpHeaderBigEndian(
312 start_time_, packet_dump_length, packet_length, dest_buffer);
313
314 // Writes the actual RTP packet header.
315 AppendToBuffer(packet_header, header_length, dest_buffer);
316 }
317
EndDump(RtpDumpType type,EndDumpCallback finished_callback)318 void WebRtcRtpDumpWriter::EndDump(RtpDumpType type,
319 EndDumpCallback finished_callback) {
320 DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
321 DCHECK(type == RTP_DUMP_OUTGOING || incoming_file_thread_worker_ != nullptr);
322 DCHECK(type == RTP_DUMP_INCOMING || outgoing_file_thread_worker_ != nullptr);
323
324 bool incoming = (type == RTP_DUMP_BOTH || type == RTP_DUMP_INCOMING);
325 EndDumpContext context(type, std::move(finished_callback));
326
327 // End the incoming dump first if required. OnDumpEnded will continue to end
328 // the outgoing dump if necessary.
329 FlushBuffer(incoming, true,
330 base::BindOnce(&WebRtcRtpDumpWriter::OnDumpEnded,
331 weak_ptr_factory_.GetWeakPtr(), std::move(context),
332 incoming));
333 }
334
max_dump_size() const335 size_t WebRtcRtpDumpWriter::max_dump_size() const {
336 DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
337 return max_dump_size_;
338 }
339
EndDumpContext(RtpDumpType type,EndDumpCallback callback)340 WebRtcRtpDumpWriter::EndDumpContext::EndDumpContext(RtpDumpType type,
341 EndDumpCallback callback)
342 : type(type),
343 incoming_succeeded(false),
344 outgoing_succeeded(false),
345 callback(std::move(callback)) {}
346
347 WebRtcRtpDumpWriter::EndDumpContext::EndDumpContext(EndDumpContext&& other) =
348 default;
349
350 WebRtcRtpDumpWriter::EndDumpContext::~EndDumpContext() = default;
351
FlushBuffer(bool incoming,bool end_stream,FlushDoneCallback callback)352 void WebRtcRtpDumpWriter::FlushBuffer(bool incoming,
353 bool end_stream,
354 FlushDoneCallback callback) {
355 DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
356
357 std::unique_ptr<std::vector<uint8_t>> new_buffer(new std::vector<uint8_t>());
358
359 if (incoming) {
360 new_buffer->reserve(incoming_buffer_.capacity());
361 new_buffer->swap(incoming_buffer_);
362 } else {
363 new_buffer->reserve(outgoing_buffer_.capacity());
364 new_buffer->swap(outgoing_buffer_);
365 }
366
367 std::unique_ptr<FlushResult> result(new FlushResult(FLUSH_RESULT_FAILURE));
368
369 std::unique_ptr<size_t> bytes_written(new size_t(0));
370
371 FileWorker* worker = incoming ? incoming_file_thread_worker_.get()
372 : outgoing_file_thread_worker_.get();
373
374 // Using "Unretained(worker)" because |worker| is owner by this object and it
375 // guaranteed to be deleted on the backround task runner before this object
376 // goes away.
377 base::OnceClosure task = base::BindOnce(
378 &FileWorker::CompressAndWriteToFileOnFileThread, base::Unretained(worker),
379 std::move(new_buffer), end_stream, result.get(), bytes_written.get());
380
381 // OnFlushDone is necessary to avoid running the callback after this
382 // object is gone.
383 base::OnceClosure reply = base::BindOnce(
384 &WebRtcRtpDumpWriter::OnFlushDone, weak_ptr_factory_.GetWeakPtr(),
385 std::move(callback), std::move(result), std::move(bytes_written));
386
387 // Define the task and reply outside the method call so that getting and
388 // passing the scoped_ptr does not depend on the argument evaluation order.
389 background_task_runner_->PostTaskAndReply(FROM_HERE, std::move(task),
390 std::move(reply));
391
392 if (end_stream) {
393 bool success = background_task_runner_->DeleteSoon(
394 FROM_HERE, incoming ? incoming_file_thread_worker_.release()
395 : outgoing_file_thread_worker_.release());
396 DCHECK(success);
397 }
398 }
399
OnFlushDone(FlushDoneCallback callback,const std::unique_ptr<FlushResult> & result,const std::unique_ptr<size_t> & bytes_written)400 void WebRtcRtpDumpWriter::OnFlushDone(
401 FlushDoneCallback callback,
402 const std::unique_ptr<FlushResult>& result,
403 const std::unique_ptr<size_t>& bytes_written) {
404 DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
405
406 total_dump_size_on_disk_ += *bytes_written;
407
408 if (total_dump_size_on_disk_ >= max_dump_size_ &&
409 !max_dump_size_reached_callback_.is_null()) {
410 max_dump_size_reached_callback_.Run();
411 }
412
413 // Returns success for FLUSH_RESULT_MAX_SIZE_REACHED since the dump is still
414 // valid.
415 if (!callback.is_null()) {
416 std::move(callback).Run(*result != FLUSH_RESULT_FAILURE &&
417 *result != FLUSH_RESULT_NO_DATA);
418 }
419 }
420
OnDumpEnded(EndDumpContext context,bool incoming,bool success)421 void WebRtcRtpDumpWriter::OnDumpEnded(EndDumpContext context,
422 bool incoming,
423 bool success) {
424 DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
425
426 DVLOG(2) << "Dump ended, incoming = " << incoming
427 << ", succeeded = " << success;
428
429 if (incoming)
430 context.incoming_succeeded = success;
431 else
432 context.outgoing_succeeded = success;
433
434 // End the outgoing dump if needed.
435 if (incoming && context.type == RTP_DUMP_BOTH) {
436 FlushBuffer(false, true,
437 base::BindOnce(&WebRtcRtpDumpWriter::OnDumpEnded,
438 weak_ptr_factory_.GetWeakPtr(),
439 std::move(context), false));
440 return;
441 }
442
443 // This object might be deleted after running the callback.
444 std::move(context).callback.Run(context.incoming_succeeded,
445 context.outgoing_succeeded);
446 }
447