Make RtcEventLogImpl to use a TaskQueue instead of a helper-thread

webrtc/logging/rtc_event_log/rtc_event_log.cc

 /*
  *  Copyright (c) 2015 The WebRTC project authors. All Rights Reserved.
  *
  *  Use of this source code is governed by a BSD-style license
  *  that can be found in the LICENSE file in the root of the source
  *  tree. An additional intellectual property rights grant can be found
  *  in the file PATENTS.  All contributing project authors may
  *  be found in the AUTHORS file in the root of the source tree.
  */
 
 #include "webrtc/logging/rtc_event_log/rtc_event_log.h"
 
+#include <atomic>
+#include <deque>
+#include <functional>
 #include <limits>
+#include <memory>
 #include <utility>
 #include <vector>
 
-#include "webrtc/logging/rtc_event_log/rtc_event_log_helper_thread.h"
 #include "webrtc/modules/audio_coding/audio_network_adaptor/include/audio_network_adaptor.h"
 #include "webrtc/modules/remote_bitrate_estimator/include/bwe_defines.h"
 #include "webrtc/modules/rtp_rtcp/include/rtp_rtcp_defines.h"
 #include "webrtc/modules/rtp_rtcp/source/byte_io.h"
 #include "webrtc/modules/rtp_rtcp/source/rtcp_packet/app.h"
 #include "webrtc/modules/rtp_rtcp/source/rtcp_packet/bye.h"
 #include "webrtc/modules/rtp_rtcp/source/rtcp_packet/common_header.h"
 #include "webrtc/modules/rtp_rtcp/source/rtcp_packet/extended_jitter_report.h"
 #include "webrtc/modules/rtp_rtcp/source/rtcp_packet/extended_reports.h"
 #include "webrtc/modules/rtp_rtcp/source/rtcp_packet/psfb.h"
 #include "webrtc/modules/rtp_rtcp/source/rtcp_packet/receiver_report.h"
 #include "webrtc/modules/rtp_rtcp/source/rtcp_packet/rtpfb.h"
 #include "webrtc/modules/rtp_rtcp/source/rtcp_packet/sdes.h"
 #include "webrtc/modules/rtp_rtcp/source/rtcp_packet/sender_report.h"
-#include "webrtc/rtc_base/atomicops.h"
 #include "webrtc/rtc_base/checks.h"
 #include "webrtc/rtc_base/constructormagic.h"
 #include "webrtc/rtc_base/event.h"
+#include "webrtc/rtc_base/ignore_wundef.h"
 #include "webrtc/rtc_base/logging.h"
 #include "webrtc/rtc_base/protobuf_utils.h"
-#include "webrtc/rtc_base/swap_queue.h"
-#include "webrtc/rtc_base/thread_checker.h"
+#include "webrtc/rtc_base/ptr_util.h"
+#include "webrtc/rtc_base/sequenced_task_checker.h"
+#include "webrtc/rtc_base/task_queue.h"
+#include "webrtc/rtc_base/thread_annotations.h"
 #include "webrtc/rtc_base/timeutils.h"
 #include "webrtc/system_wrappers/include/file_wrapper.h"
+#include "webrtc/typedefs.h"
 
 #ifdef ENABLE_RTC_EVENT_LOG
-// *.pb.h files are generated at build-time by the protobuf compiler.
+// *.ph.h files are generated at build-time by the protobuf compiler.
+RTC_PUSH_IGNORING_WUNDEF()
 #ifdef WEBRTC_ANDROID_PLATFORM_BUILD
 #include "external/webrtc/webrtc/logging/rtc_event_log/rtc_event_log.pb.h"
 #else
 #include "webrtc/logging/rtc_event_log/rtc_event_log.pb.h"
 #endif
+RTC_POP_IGNORING_WUNDEF()
 #endif
 
 namespace webrtc {
 
 #ifdef ENABLE_RTC_EVENT_LOG
 
+namespace {
+const int kEventsInHistory = 10000;
+
+bool IsConfigEvent(const rtclog::Event& event) {
+  rtclog::Event_EventType event_type = event.type();
+  return event_type == rtclog::Event::VIDEO_RECEIVER_CONFIG_EVENT ||
+         event_type == rtclog::Event::VIDEO_SENDER_CONFIG_EVENT ||
+         event_type == rtclog::Event::AUDIO_RECEIVER_CONFIG_EVENT ||
+         event_type == rtclog::Event::AUDIO_SENDER_CONFIG_EVENT;
+}
+
+// This class exists because C++11 doesn't allow transferring a unique_ptr to
+// a lambda (a copy constructor is required).
+class RtcEvent final : public rtc::QueuedTask {

[terelius, 2017/08/28 13:35:55]

Maybe RtcEventTask or RtcEventLogTask to make the name less similar to
rtc::Event?

[eladalon, 2017/08/29 13:36:55]

Done.

+ public:
+  RtcEvent(std::unique_ptr<rtclog::Event> event,
+           std::function<void(std::unique_ptr<rtclog::Event>)> handler)
+      : event_(std::move(event)), handler_(handler) {}
+
+  bool Run() override {
+    handler_(std::move(event_));
+    return true;
+  }
+
+ private:
+  std::unique_ptr<rtclog::Event> event_;
+  std::function<void(std::unique_ptr<rtclog::Event>)> handler_;
+};
+}  // namespace
+
 class RtcEventLogImpl final : public RtcEventLog {
   friend std::unique_ptr<RtcEventLog> RtcEventLog::Create();
 
  public:
   ~RtcEventLogImpl() override;
 
   bool StartLogging(const std::string& file_name,
                     int64_t max_size_bytes) override;
   bool StartLogging(rtc::PlatformFile platform_file,
                     int64_t max_size_bytes) override;
@@ skipping 22 matching lines @@
       const AudioEncoderRuntimeConfig& config) override;
   void LogProbeClusterCreated(int id,
                               int bitrate_bps,
                               int min_probes,
                               int min_bytes) override;
   void LogProbeResultSuccess(int id, int bitrate_bps) override;
   void LogProbeResultFailure(int id,
                              ProbeFailureReason failure_reason) override;
 
  private:
-  // Private constructor to ensure that creation is done by RtcEventLog::Create.
-  RtcEventLogImpl();
+  void StartLoggingInternal(std::unique_ptr<FileWrapper> file,
+                            int64_t max_size_bytes);
+
+  RtcEventLogImpl();  // Creation is done by RtcEventLog::Create.
 
   void StoreEvent(std::unique_ptr<rtclog::Event> event);
+  void LogEvent(std::unique_ptr<rtclog::Event> rtclog_event);
   void LogProbeResult(int id,
                       rtclog::BweProbeResult::ResultType result,
                       int bitrate_bps);
 
-  static volatile int log_count_;
+  // Attempts to append an event to the output protobuf string, while observing
+  // the (potential) limitation on file size. (False is returned when observing
+  // the limitation disallows appending the event.)
+  bool AppendEventToString(rtclog::Event* event,
+                           ProtoString* output_string) RTC_WARN_UNUSED_RESULT;
 
-  // Message queue for passing control messages to the logging thread.
-  SwapQueue<RtcEventLogHelperThread::ControlMessage> message_queue_;
+  void LogToMemory(std::unique_ptr<rtclog::Event> event);
 
-  // Message queue for passing events to the logging thread.
-  SwapQueue<std::unique_ptr<rtclog::Event> > event_queue_;
+  void StartLogFile(int64_t start_time);
+  void LogToFile(std::unique_ptr<rtclog::Event> event);
+  void StopLogFile(int64_t stop_time);
 
-  RtcEventLogHelperThread helper_thread_;
-  rtc::ThreadChecker thread_checker_;
+  // TODO(eladalon): Known issue - there's a race over |log_count_|.
+  static std::atomic<int> log_count_;
+
+  rtc::SequencedTaskChecker worker_sequence_checker_;
+  rtc::SequencedTaskChecker owner_sequence_checker_;
+
+  // History containing all past configuration events.
+  std::vector<std::unique_ptr<rtclog::Event>> config_history_
+      GUARDED_BY(worker_sequence_checker_);
+
+  // History containing the most recent (non-configuration) events (~ 10 s).
+  std::deque<std::unique_ptr<rtclog::Event>> history_
+      GUARDED_BY(worker_sequence_checker_);
+
+  std::unique_ptr<FileWrapper> file_ GUARDED_BY(worker_sequence_checker_);
+
+  size_t max_size_bytes_ GUARDED_BY(worker_sequence_checker_);
+  size_t written_bytes_ GUARDED_BY(worker_sequence_checker_);
+
+  // Keep this last to ensure it destructs first, or else tasks living on the
+  // queue might access other members after they've been torn down.
+  rtc::TaskQueue task_queue_;
 
   RTC_DISALLOW_COPY_AND_ASSIGN(RtcEventLogImpl);
 };
 
 namespace {
 // The functions in this namespace convert enums from the runtime format
 // that the rest of the WebRtc project can use, to the corresponding
 // serialized enum which is defined by the protobuf.
 
 rtclog::VideoReceiveConfig_RtcpMode ConvertRtcpMode(RtcpMode rtcp_mode) {
@@ skipping 31 matching lines @@
       return rtclog::BweProbeResult::INVALID_SEND_RECEIVE_INTERVAL;
     case kInvalidSendReceiveRatio:
       return rtclog::BweProbeResult::INVALID_SEND_RECEIVE_RATIO;
     case kTimeout:
       return rtclog::BweProbeResult::TIMEOUT;
   }
   RTC_NOTREACHED();
   return rtclog::BweProbeResult::SUCCESS;
 }
 
-// The RTP and RTCP buffers reserve space for twice the expected number of
-// sent packets because they also contain received packets.
-static const int kEventsPerSecond = 1000;
-static const int kControlMessagesPerSecond = 10;
 }  // namespace
 
-volatile int RtcEventLogImpl::log_count_ = 0;
+std::atomic<int> RtcEventLogImpl::log_count_(0);
 
-// RtcEventLogImpl member functions.
 RtcEventLogImpl::RtcEventLogImpl()
-    // Allocate buffers for roughly one second of history.
-    : message_queue_(kControlMessagesPerSecond),
-      event_queue_(kEventsPerSecond),
-      helper_thread_(&message_queue_, &event_queue_),
-      thread_checker_() {
-  thread_checker_.DetachFromThread();
+    : file_(FileWrapper::Create()),
+      max_size_bytes_(std::numeric_limits<decltype(max_size_bytes_)>::max()),
+      written_bytes_(0),
+      task_queue_("rtc_event_log") {
+  worker_sequence_checker_.Detach();
 }
 
 RtcEventLogImpl::~RtcEventLogImpl() {
-  // The RtcEventLogHelperThread destructor closes the file
-  // and waits for the thread to terminate.
-  int count = rtc::AtomicOps::Decrement(&RtcEventLogImpl::log_count_);
+  RTC_DCHECK_CALLED_SEQUENTIALLY(&owner_sequence_checker_);
+
+  // If we're logging to the file, this will stop that. Blocking function.
+  StopLogging();
+
+  int count = std::atomic_fetch_sub(&RtcEventLogImpl::log_count_, 1) - 1;
   RTC_DCHECK_GE(count, 0);
 }
 
 bool RtcEventLogImpl::StartLogging(const std::string& file_name,
                                    int64_t max_size_bytes) {
-  RTC_DCHECK(thread_checker_.CalledOnValidThread());
-  RtcEventLogHelperThread::ControlMessage message;
-  message.message_type = RtcEventLogHelperThread::ControlMessage::START_FILE;
-  message.max_size_bytes = max_size_bytes <= 0
-                               ? std::numeric_limits<int64_t>::max()
-                               : max_size_bytes;
-  message.start_time = rtc::TimeMicros();
-  message.stop_time = std::numeric_limits<int64_t>::max();
-  message.file.reset(FileWrapper::Create());
-  if (!message.file->OpenFile(file_name.c_str(), false)) {
+  RTC_DCHECK_CALLED_SEQUENTIALLY(&owner_sequence_checker_);
+
+  auto file = rtc::WrapUnique<FileWrapper>(FileWrapper::Create());
+  if (!file->OpenFile(file_name.c_str(), false)) {
     LOG(LS_ERROR) << "Can't open file. WebRTC event log not started.";
     return false;
   }
-  if (!message_queue_.Insert(&message)) {
-    LOG(LS_ERROR) << "Message queue full. Can't start logging.";
-    return false;
-  }
-  helper_thread_.SignalNewEvent();
-  LOG(LS_INFO) << "Starting WebRTC event log.";
+
+  StartLoggingInternal(std::move(file), max_size_bytes);
+
   return true;
 }
 
 bool RtcEventLogImpl::StartLogging(rtc::PlatformFile platform_file,
                                    int64_t max_size_bytes) {
-  RTC_DCHECK(thread_checker_.CalledOnValidThread());
-  RtcEventLogHelperThread::ControlMessage message;
-  message.message_type = RtcEventLogHelperThread::ControlMessage::START_FILE;
-  message.max_size_bytes = max_size_bytes <= 0
-                               ? std::numeric_limits<int64_t>::max()
-                               : max_size_bytes;
-  message.start_time = rtc::TimeMicros();
-  message.stop_time = std::numeric_limits<int64_t>::max();
-  message.file.reset(FileWrapper::Create());
+  RTC_DCHECK_CALLED_SEQUENTIALLY(&owner_sequence_checker_);
+
+  auto file = rtc::WrapUnique<FileWrapper>(FileWrapper::Create());
   FILE* file_handle = rtc::FdopenPlatformFileForWriting(platform_file);
   if (!file_handle) {
     LOG(LS_ERROR) << "Can't open file. WebRTC event log not started.";
     // Even though we failed to open a FILE*, the platform_file is still open
     // and needs to be closed.
     if (!rtc::ClosePlatformFile(platform_file)) {
       LOG(LS_ERROR) << "Can't close file.";
     }
     return false;
   }
-  if (!message.file->OpenFromFileHandle(file_handle)) {
+  if (!file->OpenFromFileHandle(file_handle)) {
     LOG(LS_ERROR) << "Can't open file. WebRTC event log not started.";
     return false;
   }
-  if (!message_queue_.Insert(&message)) {
-    LOG(LS_ERROR) << "Message queue full. Can't start logging.";
-    return false;
-  }
-  helper_thread_.SignalNewEvent();
-  LOG(LS_INFO) << "Starting WebRTC event log.";
+
+  StartLoggingInternal(std::move(file), max_size_bytes);
+
   return true;
 }
 
 void RtcEventLogImpl::StopLogging() {
-  RTC_DCHECK(thread_checker_.CalledOnValidThread());
-  RtcEventLogHelperThread::ControlMessage message;
-  message.message_type = RtcEventLogHelperThread::ControlMessage::STOP_FILE;
-  message.stop_time = rtc::TimeMicros();
-  while (!message_queue_.Insert(&message)) {
-    // TODO(terelius): We would like to have a blocking Insert function in the
-    // SwapQueue, but for the time being we will just clear any previous
-    // messages.
-    // Since StopLogging waits for the thread, it is essential that we don't
-    // clear any STOP_FILE messages. To ensure that there is only one call at a
-    // time, we require that all calls to StopLogging are made on the same
-    // thread.
-    LOG(LS_ERROR) << "Message queue full. Clearing queue to stop logging.";
-    message_queue_.Clear();
-  }
+  RTC_DCHECK_CALLED_SEQUENTIALLY(&owner_sequence_checker_);
+
   LOG(LS_INFO) << "Stopping WebRTC event log.";
-  helper_thread_.WaitForFileFinished();
+
+  const int64_t stop_time = rtc::TimeMicros();
+
+  task_queue_.PostTask([this, stop_time]() {
+    RTC_DCHECK_CALLED_SEQUENTIALLY(&worker_sequence_checker_);
+    if (file_->is_open()) {
+      StopLogFile(stop_time);
+    }
+  });
+
+  // Wait for an indication that the previous task, which stops the log file,
+  // has been executed to completion.
+  rtc::Event file_finished(true, false);
+  task_queue_.PostTask([this, &file_finished]() {
+    RTC_DCHECK_CALLED_SEQUENTIALLY(&worker_sequence_checker_);
+    file_finished.Set();
+  });
+  file_finished.Wait(rtc::Event::kForever);
+
+  LOG(LS_INFO) << "WebRTC event log successfully stopped.";
 }
 
 void RtcEventLogImpl::LogVideoReceiveStreamConfig(
     const rtclog::StreamConfig& config) {
   std::unique_ptr<rtclog::Event> event(new rtclog::Event());
   event->set_timestamp_us(rtc::TimeMicros());
   event->set_type(rtclog::Event::VIDEO_RECEIVER_CONFIG_EVENT);
 
   rtclog::VideoReceiveConfig* receiver_config =
       event->mutable_video_receiver_config();
@@ skipping 280 matching lines @@
   event->set_type(rtclog::Event::BWE_PROBE_RESULT_EVENT);
 
   auto probe_result = event->mutable_probe_result();
   probe_result->set_id(id);
   probe_result->set_result(result);
   if (result == rtclog::BweProbeResult::SUCCESS)
     probe_result->set_bitrate_bps(bitrate_bps);
   StoreEvent(std::move(event));
 }
 
+void RtcEventLogImpl::StartLoggingInternal(std::unique_ptr<FileWrapper> file,
+                                           int64_t max_size_bytes) {
+  LOG(LS_INFO) << "Starting WebRTC event log.";
+
+  int64_t start_time = rtc::TimeMicros();
+  max_size_bytes = (max_size_bytes <= 0) ? std::numeric_limits<int64_t>::max()
+                                         : max_size_bytes;
+
+  // Kept as a raw pointer due to limitations C++11 imposes on lambdas; namely,
+  // the necessity of a copy-constructor.
+  FileWrapper* file_raw = file.release();
+
+  task_queue_.PostTask([this, start_time, max_size_bytes, file_raw]() {

[terelius, 2017/08/28 13:35:55]

Are tasks guaranteed to be executed or is there any chance that we leak the
FileWrapper memory and an open file descriptor?? What happens if we delete the
event log and/or taskqueue?

[eladalon, 2017/08/29 13:36:55]

You're right in theory (thanks for catching!), because TaskQueue's
implementation says pending tasks get deleted when the destructor is invoked. So
I'll handle this in the way I'd handled the events - create a class for it.

In practice, at least on my Mac, the destructor seems to block until the task
queue becomes empty (which means the raw-pointer can never leak). This, however,
seems like a bug that we should fix. Please see this CL for the demonstration of
the bug in action: https://codereview.webrtc.org/3009763002

+    RTC_DCHECK_CALLED_SEQUENTIALLY(&worker_sequence_checker_);
+    if (!file_->is_open()) {
+      max_size_bytes_ = max_size_bytes;
+      file_.reset(file_raw);
+      StartLogFile(start_time);
+    } else {
+      // Already started. Ignore message and close file handle.
+      file_raw->CloseFile();
+      delete file_raw;
+    }
+  });
+}
+
 void RtcEventLogImpl::StoreEvent(std::unique_ptr<rtclog::Event> event) {
-  RTC_DCHECK(event.get() != nullptr);
-  if (!event_queue_.Insert(&event)) {
-    LOG(LS_ERROR) << "WebRTC event log queue full. Dropping event.";
+  RTC_DCHECK(event);
+  // LogEvent is necessary because lambdas in C++11 need their captured
+  // arguments to have a copy-constructor.
+  auto event_handler = [this](std::unique_ptr<rtclog::Event> rtclog_event) {
+    LogEvent(std::move(rtclog_event));
+  };
+  auto task = rtc::WrapUnique<rtc::QueuedTask>(
+      new RtcEvent(std::move(event), event_handler));
+  task_queue_.PostTask(std::move(task));
+}
+
+void RtcEventLogImpl::LogEvent(std::unique_ptr<rtclog::Event> event) {
+  RTC_DCHECK_CALLED_SEQUENTIALLY(&worker_sequence_checker_);
+  if (file_->is_open()) {
+    LogToFile(std::move(event));
+  } else {
+    LogToMemory(std::move(event));
   }
-  helper_thread_.SignalNewEvent();
+}
+
+bool RtcEventLogImpl::AppendEventToString(rtclog::Event* event,
+                                          ProtoString* output_string) {
+  RTC_DCHECK_CALLED_SEQUENTIALLY(&worker_sequence_checker_);
+  rtclog::EventStream event_stream;
+  event_stream.add_stream();
+  event_stream.mutable_stream(0)->Swap(event);
+  // We create a new event stream per event but because of the way protobufs
+  // are encoded, events can be merged by concatenating them. Therefore,
+  // it will look like a single stream when we read it back from file.
+  bool appended;
+  size_t potential_new_size =
+      written_bytes_ + output_string->size() + event_stream.ByteSize();
+  if (potential_new_size <= max_size_bytes_) {
+    event_stream.AppendToString(output_string);
+    appended = true;
+  } else {
+    appended = false;
+  }
+  // Swap the event back so that we don't mix event types in the queues.

[terelius, 2017/08/28 13:35:55]

I still think we should do the Swap(), but this comment is outdated. (At some
point we had separate queues for each event type.)

[eladalon, 2017/08/29 13:36:55]

I've freshened up the comments.

+  event_stream.mutable_stream(0)->Swap(event);
+  return appended;
+}
+
+void RtcEventLogImpl::LogToMemory(std::unique_ptr<rtclog::Event> event) {
+  RTC_DCHECK_CALLED_SEQUENTIALLY(&worker_sequence_checker_);
+  RTC_DCHECK(!file_->is_open());
+
+  if (IsConfigEvent(*event.get())) {
+    config_history_.push_back(std::move(event));
+  } else {
+    history_.push_back(std::move(event));
+    if (history_.size() > kEventsInHistory) {
+      history_.pop_front();
+    }
+  }
+}
+
+void RtcEventLogImpl::StartLogFile(int64_t start_time) {
+  RTC_DCHECK_CALLED_SEQUENTIALLY(&worker_sequence_checker_);
+  RTC_DCHECK(file_->is_open());
+
+  ProtoString output_string;
+
+  // Create and serialize the LOG_START event.
+  rtclog::Event start_event;
+  // TODO(eladalon): !!! This time precedes the timestamps in history_. Is that

[eladalon, 2017/08/24 09:23:25]

Björn, what do you think?

[terelius, 2017/08/28 13:35:55]

You mean that it is written to file before serializing the older events in
history_? Encapsulating each log segment in a LOG_START ... LOG_END pair
actually simplifies parsing in the cases where we only want to look at one
segment. The next time we change the log format we should consider adding some
extra information to LOG_START, to avoid this implicit dependency on the order
of events in the file.

Please keep this as-is for now.

[eladalon, 2017/08/29 13:36:55]

For posterity - we've discussed this some more offline. For a while, it sounded
as if taking the oldest event's timestamp would be the way to go, but ended up
deciding that would 1. not be very meaningful, 2. it wouldn't be clear if we
should take the oldest event or the oldest config-event, and 3. the log's
timestamps are not monotonic anyway.

+  // a problem for us?
+  start_event.set_timestamp_us(start_time);
+  start_event.set_type(rtclog::Event::LOG_START);
+  bool appended = AppendEventToString(&start_event, &output_string);
+
+  // Serialize the config information for all old streams, including streams
+  // which were already logged to previous files.
+  for (auto& event : config_history_) {
+    if (!appended) {
+      break;
+    }
+    appended = AppendEventToString(event.get(), &output_string);
+  }
+
+  // Serialize the events in the event queue.
+  while (appended && !history_.empty()) {
+    appended = AppendEventToString(history_.front().get(), &output_string);
+    if (appended) {
+      // TODO(eladalon): !!! Do we really want to pop before the write to

[eladalon, 2017/08/24 09:23:26]

Björn, what do you think?

[terelius, 2017/08/28 13:35:54]

I don't think this matters one way or the other. The only way this will have any
measurable effect is if the write fails despite not reaching the file limit, and
we start logging again within ~10 seconds without having received any
notification that the old log has stopped, and the new log actually can write to
the file. I'm not sure under what circumstances this would actually happen.

Feel free to change this behavior if you see a clean way to do so. However, I'm
more worried that future maintenance and refactorings will cause duplicated
events in the log if we separate processing the events in history and removing
the processed events.

[eladalon, 2017/08/29 13:36:55]

I can document it as a known-potential-unlikely-issue. I think I'll do that,
because writing code to handle this edge-case would require unit-tests
simulating this edge-case, and this seems like a poor investment of our time.

+      // the file proves successful?
+      history_.pop_front();
+    }
+  }
+
+  // Write to file.
+  if (!file_->Write(output_string.data(), output_string.size())) {
+    LOG(LS_ERROR) << "FileWrapper failed to write WebRtcEventLog file.";
+    // The current FileWrapper implementation closes the file on error.
+    RTC_DCHECK(!file_->is_open());
+    return;
+  }
+  written_bytes_ += output_string.size();
+
+  if (!appended) {
+    RTC_DCHECK(file_->is_open());
+    StopLogFile(rtc::TimeMicros());
+  }
+}
+
+void RtcEventLogImpl::LogToFile(std::unique_ptr<rtclog::Event> event) {
+  RTC_DCHECK_CALLED_SEQUENTIALLY(&worker_sequence_checker_);
+  RTC_DCHECK(file_->is_open());
+
+  ProtoString output_string;
+
+  bool appended = AppendEventToString(event.get(), &output_string);
+
+  if (IsConfigEvent(*event.get())) {
+    config_history_.push_back(std::move(event));
+  }
+
+  if (!appended) {
+    RTC_DCHECK(file_->is_open());
+    StopLogFile(rtc::TimeMicros());
+    return;
+  }
+
+  // Write string to file.
+  if (file_->Write(output_string.data(), output_string.size())) {
+    written_bytes_ += output_string.size();
+  } else {
+    LOG(LS_ERROR) << "FileWrapper failed to write WebRtcEventLog file.";
+    // The current FileWrapper implementation closes the file on error.
+    RTC_DCHECK(!file_->is_open());
+  }
+}
+
+void RtcEventLogImpl::StopLogFile(int64_t stop_time) {
+  RTC_DCHECK_CALLED_SEQUENTIALLY(&worker_sequence_checker_);
+  RTC_DCHECK(file_->is_open());
+
+  ProtoString output_string;
+
+  rtclog::Event end_event;
+  end_event.set_timestamp_us(stop_time);
+  end_event.set_type(rtclog::Event::LOG_END);
+  bool appended = AppendEventToString(&end_event, &output_string);
+
+  if (appended) {
+    if (!file_->Write(output_string.data(), output_string.size())) {
+      LOG(LS_ERROR) << "FileWrapper failed to write WebRtcEventLog file.";
+      // The current FileWrapper implementation closes the file on error.
+      RTC_DCHECK(!file_->is_open());
+    }
+    written_bytes_ += output_string.size();
+  }
+
+  max_size_bytes_ = std::numeric_limits<int64_t>::max();
+  written_bytes_ = 0;
+
+  file_->CloseFile();
+  RTC_DCHECK(!file_->is_open());
 }
 
 bool RtcEventLog::ParseRtcEventLog(const std::string& file_name,
                                    rtclog::EventStream* result) {
   char tmp_buffer[1024];
   int bytes_read = 0;
   std::unique_ptr<FileWrapper> dump_file(FileWrapper::Create());
   if (!dump_file->OpenFile(file_name.c_str(), true)) {
     return false;
   }
   ProtoString dump_buffer;
   while ((bytes_read = dump_file->Read(tmp_buffer, sizeof(tmp_buffer))) > 0) {
     dump_buffer.append(tmp_buffer, bytes_read);
   }
   dump_file->CloseFile();
   return result->ParseFromString(dump_buffer);
 }
 
 #endif  // ENABLE_RTC_EVENT_LOG
 
 // RtcEventLog member functions.
 std::unique_ptr<RtcEventLog> RtcEventLog::Create() {
 #ifdef ENABLE_RTC_EVENT_LOG
+  // TODO(eladalon): Known issue - there's a race over |log_count_| here.
   constexpr int kMaxLogCount = 5;
-  int count = rtc::AtomicOps::Increment(&RtcEventLogImpl::log_count_);
+  int count = 1 + std::atomic_fetch_add(&RtcEventLogImpl::log_count_, 1);
   if (count > kMaxLogCount) {
     LOG(LS_WARNING) << "Denied creation of additional WebRTC event logs. "
                     << count - 1 << " logs open already.";
-    rtc::AtomicOps::Decrement(&RtcEventLogImpl::log_count_);
+    std::atomic_fetch_sub(&RtcEventLogImpl::log_count_, 1);
     return std::unique_ptr<RtcEventLog>(new RtcEventLogNullImpl());
   }
   return std::unique_ptr<RtcEventLog>(new RtcEventLogImpl());
 #else
-  return std::unique_ptr<RtcEventLog>(new RtcEventLogNullImpl());
+  return CreateNull();
 #endif  // ENABLE_RTC_EVENT_LOG
 }
 
 std::unique_ptr<RtcEventLog> RtcEventLog::CreateNull() {
   return std::unique_ptr<RtcEventLog>(new RtcEventLogNullImpl());
 }
 
 }  // namespace webrtc