chrome/browser/media/webrtc_rtp_dump_writer.cc

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