diff options
Diffstat (limited to 'chromium/chrome/browser/media/webrtc/webrtc_rtp_dump_writer.cc')
| -rw-r--r-- | chromium/chrome/browser/media/webrtc/webrtc_rtp_dump_writer.cc | 451 |
1 files changed, 451 insertions, 0 deletions
diff --git a/chromium/chrome/browser/media/webrtc/webrtc_rtp_dump_writer.cc b/chromium/chrome/browser/media/webrtc/webrtc_rtp_dump_writer.cc new file mode 100644 index 00000000000..d676f226252 --- /dev/null +++ b/chromium/chrome/browser/media/webrtc/webrtc_rtp_dump_writer.cc @@ -0,0 +1,451 @@ +// Copyright 2014 The Chromium Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. + +#include "chrome/browser/media/webrtc/webrtc_rtp_dump_writer.h" + +#include <string.h> + +#include "base/big_endian.h" +#include "base/bind.h" +#include "base/files/file_util.h" +#include "base/logging.h" +#include "base/stl_util.h" +#include "base/task/post_task.h" +#include "content/public/browser/browser_thread.h" +#include "third_party/zlib/zlib.h" + +namespace { + +static const size_t kMinimumGzipOutputBufferSize = 256; // In bytes. + +const unsigned char kRtpDumpFileHeaderFirstLine[] = "#!rtpplay1.0 0.0.0.0/0\n"; +static const size_t kRtpDumpFileHeaderSize = 16; // In bytes. + +// A helper for writing the header of the dump file. +void WriteRtpDumpFileHeaderBigEndian(base::TimeTicks start, + std::vector<uint8_t>* output) { + size_t buffer_start_pos = output->size(); + output->resize(output->size() + kRtpDumpFileHeaderSize); + + char* buffer = reinterpret_cast<char*>(&(*output)[buffer_start_pos]); + + base::TimeDelta delta = start - base::TimeTicks(); + uint32_t start_sec = delta.InSeconds(); + base::WriteBigEndian(buffer, start_sec); + buffer += sizeof(start_sec); + + uint32_t start_usec = + delta.InMilliseconds() * base::Time::kMicrosecondsPerMillisecond; + base::WriteBigEndian(buffer, start_usec); + buffer += sizeof(start_usec); + + // Network source, always 0. + base::WriteBigEndian(buffer, uint32_t(0)); + buffer += sizeof(uint32_t); + + // UDP port, always 0. + base::WriteBigEndian(buffer, uint16_t(0)); + buffer += sizeof(uint16_t); + + // 2 bytes padding. + base::WriteBigEndian(buffer, uint16_t(0)); +} + +// The header size for each packet dump. +static const size_t kPacketDumpHeaderSize = 8; // In bytes. + +// A helper for writing the header for each packet dump. +// |start| is the time when the recording is started. +// |dump_length| is the length of the packet dump including this header. +// |packet_length| is the length of the RTP packet header. +void WritePacketDumpHeaderBigEndian(const base::TimeTicks& start, + uint16_t dump_length, + uint16_t packet_length, + std::vector<uint8_t>* output) { + size_t buffer_start_pos = output->size(); + output->resize(output->size() + kPacketDumpHeaderSize); + + char* buffer = reinterpret_cast<char*>(&(*output)[buffer_start_pos]); + + base::WriteBigEndian(buffer, dump_length); + buffer += sizeof(dump_length); + + base::WriteBigEndian(buffer, packet_length); + buffer += sizeof(packet_length); + + uint32_t elapsed = + static_cast<uint32_t>((base::TimeTicks::Now() - start).InMilliseconds()); + base::WriteBigEndian(buffer, elapsed); +} + +// Append |src_len| bytes from |src| to |dest|. +void AppendToBuffer(const uint8_t* src, + size_t src_len, + std::vector<uint8_t>* dest) { + size_t old_dest_size = dest->size(); + dest->resize(old_dest_size + src_len); + memcpy(&(*dest)[old_dest_size], src, src_len); +} + +} // namespace + +// This class runs on the backround task runner, compresses and writes the +// dump buffer to disk. +class WebRtcRtpDumpWriter::FileWorker { + public: + explicit FileWorker(const base::FilePath& dump_path) : dump_path_(dump_path) { + DETACH_FROM_SEQUENCE(sequence_checker_); + + memset(&stream_, 0, sizeof(stream_)); + int result = deflateInit2(&stream_, + Z_DEFAULT_COMPRESSION, + Z_DEFLATED, + // windowBits = 15 is default, 16 is added to + // produce a gzip header + trailer. + 15 + 16, + 8, // memLevel = 8 is default. + Z_DEFAULT_STRATEGY); + DCHECK_EQ(Z_OK, result); + } + + ~FileWorker() { + DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); + + // Makes sure all allocations are freed. + deflateEnd(&stream_); + } + + // Compresses the data in |buffer| and write to the dump file. If |end_stream| + // is true, the compression stream will be ended and the dump file cannot be + // written to any more. + void CompressAndWriteToFileOnFileThread( + std::unique_ptr<std::vector<uint8_t>> buffer, + bool end_stream, + FlushResult* result, + size_t* bytes_written) { + DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); + + // This is called either when the in-memory buffer is full or the dump + // should be ended. + DCHECK(!buffer->empty() || end_stream); + + *result = FLUSH_RESULT_SUCCESS; + *bytes_written = 0; + + // There may be nothing to compress/write if there is no RTP packet since + // the last flush. + if (!buffer->empty()) { + *bytes_written = CompressAndWriteBufferToFile(buffer.get(), result); + } else if (!base::PathExists(dump_path_)) { + // If the dump does not exist, it means there is no RTP packet recorded. + // Return FLUSH_RESULT_NO_DATA to indicate no dump file created. + *result = FLUSH_RESULT_NO_DATA; + } + + if (end_stream && !EndDumpFile()) + *result = FLUSH_RESULT_FAILURE; + } + + private: + // Helper for CompressAndWriteToFileOnFileThread to compress and write one + // dump. + size_t CompressAndWriteBufferToFile(std::vector<uint8_t>* buffer, + FlushResult* result) { + DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); + DCHECK(buffer->size()); + + *result = FLUSH_RESULT_SUCCESS; + + std::vector<uint8_t> compressed_buffer; + if (!Compress(buffer, &compressed_buffer)) { + DVLOG(2) << "Compressing buffer failed."; + *result = FLUSH_RESULT_FAILURE; + return 0; + } + + int bytes_written = -1; + + if (base::PathExists(dump_path_)) { + bytes_written = + base::AppendToFile(dump_path_, reinterpret_cast<const char*>( + compressed_buffer.data()), + compressed_buffer.size()) + ? compressed_buffer.size() + : -1; + } else { + bytes_written = base::WriteFile( + dump_path_, + reinterpret_cast<const char*>(&compressed_buffer[0]), + compressed_buffer.size()); + } + + if (bytes_written == -1) { + DVLOG(2) << "Writing file failed: " << dump_path_.value(); + *result = FLUSH_RESULT_FAILURE; + return 0; + } + + DCHECK_EQ(static_cast<size_t>(bytes_written), compressed_buffer.size()); + return bytes_written; + } + + // Compresses |input| into |output|. + bool Compress(std::vector<uint8_t>* input, std::vector<uint8_t>* output) { + DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); + int result = Z_OK; + + output->resize(std::max(kMinimumGzipOutputBufferSize, input->size())); + + stream_.next_in = &(*input)[0]; + stream_.avail_in = input->size(); + stream_.next_out = &(*output)[0]; + stream_.avail_out = output->size(); + + result = deflate(&stream_, Z_SYNC_FLUSH); + DCHECK_EQ(Z_OK, result); + DCHECK_EQ(0U, stream_.avail_in); + + output->resize(output->size() - stream_.avail_out); + + stream_.next_in = NULL; + stream_.next_out = NULL; + stream_.avail_out = 0; + return true; + } + + // Ends the compression stream and completes the dump file. + bool EndDumpFile() { + DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); + + std::vector<uint8_t> output_buffer; + output_buffer.resize(kMinimumGzipOutputBufferSize); + + stream_.next_in = NULL; + stream_.avail_in = 0; + stream_.next_out = &output_buffer[0]; + stream_.avail_out = output_buffer.size(); + + int result = deflate(&stream_, Z_FINISH); + DCHECK_EQ(Z_STREAM_END, result); + + result = deflateEnd(&stream_); + DCHECK_EQ(Z_OK, result); + + output_buffer.resize(output_buffer.size() - stream_.avail_out); + + memset(&stream_, 0, sizeof(z_stream)); + + DCHECK(!output_buffer.empty()); + return base::AppendToFile( + dump_path_, reinterpret_cast<const char*>(output_buffer.data()), + output_buffer.size()); + } + + const base::FilePath dump_path_; + + z_stream stream_; + + SEQUENCE_CHECKER(sequence_checker_); + + DISALLOW_COPY_AND_ASSIGN(FileWorker); +}; + +WebRtcRtpDumpWriter::WebRtcRtpDumpWriter( + const base::FilePath& incoming_dump_path, + const base::FilePath& outgoing_dump_path, + size_t max_dump_size, + const base::Closure& max_dump_size_reached_callback) + : max_dump_size_(max_dump_size), + max_dump_size_reached_callback_(max_dump_size_reached_callback), + total_dump_size_on_disk_(0), + background_task_runner_( + base::CreateSequencedTaskRunner({base::ThreadPool(), base::MayBlock(), + base::TaskPriority::BEST_EFFORT})), + incoming_file_thread_worker_(new FileWorker(incoming_dump_path)), + outgoing_file_thread_worker_(new FileWorker(outgoing_dump_path)) {} + +WebRtcRtpDumpWriter::~WebRtcRtpDumpWriter() { + DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); + + bool success = background_task_runner_->DeleteSoon( + FROM_HERE, incoming_file_thread_worker_.release()); + DCHECK(success); + + success = background_task_runner_->DeleteSoon( + FROM_HERE, outgoing_file_thread_worker_.release()); + DCHECK(success); +} + +void WebRtcRtpDumpWriter::WriteRtpPacket(const uint8_t* packet_header, + size_t header_length, + size_t packet_length, + bool incoming) { + DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); + + static const size_t kMaxInMemoryBufferSize = 65536; + + std::vector<uint8_t>* dest_buffer = + incoming ? &incoming_buffer_ : &outgoing_buffer_; + + // We use the capacity of the buffer to indicate if the buffer has been + // initialized and if the dump file header has been created. + if (!dest_buffer->capacity()) { + dest_buffer->reserve(std::min(kMaxInMemoryBufferSize, max_dump_size_)); + + start_time_ = base::TimeTicks::Now(); + + // Writes the dump file header. + AppendToBuffer(kRtpDumpFileHeaderFirstLine, + base::size(kRtpDumpFileHeaderFirstLine) - 1, dest_buffer); + WriteRtpDumpFileHeaderBigEndian(start_time_, dest_buffer); + } + + size_t packet_dump_length = kPacketDumpHeaderSize + header_length; + + // Flushes the buffer to disk if the buffer is full. + if (dest_buffer->size() + packet_dump_length > dest_buffer->capacity()) + FlushBuffer(incoming, false, FlushDoneCallback()); + + WritePacketDumpHeaderBigEndian( + start_time_, packet_dump_length, packet_length, dest_buffer); + + // Writes the actual RTP packet header. + AppendToBuffer(packet_header, header_length, dest_buffer); +} + +void WebRtcRtpDumpWriter::EndDump(RtpDumpType type, + const EndDumpCallback& finished_callback) { + DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); + DCHECK(type == RTP_DUMP_OUTGOING || incoming_file_thread_worker_ != NULL); + DCHECK(type == RTP_DUMP_INCOMING || outgoing_file_thread_worker_ != NULL); + + bool incoming = (type == RTP_DUMP_BOTH || type == RTP_DUMP_INCOMING); + EndDumpContext context(type, finished_callback); + + // End the incoming dump first if required. OnDumpEnded will continue to end + // the outgoing dump if necessary. + FlushBuffer(incoming, + true, + base::Bind(&WebRtcRtpDumpWriter::OnDumpEnded, + weak_ptr_factory_.GetWeakPtr(), + context, + incoming)); +} + +size_t WebRtcRtpDumpWriter::max_dump_size() const { + DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); + return max_dump_size_; +} + +WebRtcRtpDumpWriter::EndDumpContext::EndDumpContext( + RtpDumpType type, + const EndDumpCallback& callback) + : type(type), + incoming_succeeded(false), + outgoing_succeeded(false), + callback(callback) { +} + +WebRtcRtpDumpWriter::EndDumpContext::EndDumpContext( + const EndDumpContext& other) = default; + +WebRtcRtpDumpWriter::EndDumpContext::~EndDumpContext() { +} + +void WebRtcRtpDumpWriter::FlushBuffer(bool incoming, + bool end_stream, + const FlushDoneCallback& callback) { + DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); + + std::unique_ptr<std::vector<uint8_t>> new_buffer(new std::vector<uint8_t>()); + + if (incoming) { + new_buffer->reserve(incoming_buffer_.capacity()); + new_buffer->swap(incoming_buffer_); + } else { + new_buffer->reserve(outgoing_buffer_.capacity()); + new_buffer->swap(outgoing_buffer_); + } + + std::unique_ptr<FlushResult> result(new FlushResult(FLUSH_RESULT_FAILURE)); + + std::unique_ptr<size_t> bytes_written(new size_t(0)); + + FileWorker* worker = incoming ? incoming_file_thread_worker_.get() + : outgoing_file_thread_worker_.get(); + + // Using "Unretained(worker)" because |worker| is owner by this object and it + // guaranteed to be deleted on the backround task runner before this object + // goes away. + base::OnceClosure task = base::BindOnce( + &FileWorker::CompressAndWriteToFileOnFileThread, base::Unretained(worker), + std::move(new_buffer), end_stream, result.get(), bytes_written.get()); + + // OnFlushDone is necessary to avoid running the callback after this + // object is gone. + base::OnceClosure reply = base::BindOnce( + &WebRtcRtpDumpWriter::OnFlushDone, weak_ptr_factory_.GetWeakPtr(), + callback, std::move(result), std::move(bytes_written)); + + // Define the task and reply outside the method call so that getting and + // passing the scoped_ptr does not depend on the argument evaluation order. + background_task_runner_->PostTaskAndReply(FROM_HERE, std::move(task), + std::move(reply)); + + if (end_stream) { + bool success = background_task_runner_->DeleteSoon( + FROM_HERE, incoming ? incoming_file_thread_worker_.release() + : outgoing_file_thread_worker_.release()); + DCHECK(success); + } +} + +void WebRtcRtpDumpWriter::OnFlushDone( + const FlushDoneCallback& callback, + const std::unique_ptr<FlushResult>& result, + const std::unique_ptr<size_t>& bytes_written) { + DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); + + total_dump_size_on_disk_ += *bytes_written; + + if (total_dump_size_on_disk_ >= max_dump_size_ && + !max_dump_size_reached_callback_.is_null()) { + max_dump_size_reached_callback_.Run(); + } + + // Returns success for FLUSH_RESULT_MAX_SIZE_REACHED since the dump is still + // valid. + if (!callback.is_null()) { + callback.Run(*result != FLUSH_RESULT_FAILURE && + *result != FLUSH_RESULT_NO_DATA); + } +} + +void WebRtcRtpDumpWriter::OnDumpEnded(EndDumpContext context, + bool incoming, + bool success) { + DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_); + + DVLOG(2) << "Dump ended, incoming = " << incoming + << ", succeeded = " << success; + + if (incoming) + context.incoming_succeeded = success; + else + context.outgoing_succeeded = success; + + // End the outgoing dump if needed. + if (incoming && context.type == RTP_DUMP_BOTH) { + FlushBuffer(false, + true, + base::Bind(&WebRtcRtpDumpWriter::OnDumpEnded, + weak_ptr_factory_.GetWeakPtr(), + context, + false)); + return; + } + + // This object might be deleted after running the callback. + context.callback.Run(context.incoming_succeeded, context.outgoing_succeeded); +} |