Refactored CL for moving the output to a separate thread.

webrtc/call/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/call/rtc_event_log.h"
 
-#include <deque>
 #include <vector>
+#include <limits>
 
 #include "webrtc/base/checks.h"
+#include "webrtc/base/constructormagic.h"
 #include "webrtc/base/criticalsection.h"
+#include "webrtc/base/event.h"
+#include "webrtc/base/platform_thread.h"
 #include "webrtc/base/thread_annotations.h"
 #include "webrtc/call.h"
+#include "webrtc/call/ringbuffer.h"
+#include "webrtc/common_audio/swap_queue.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_utility.h"
 #include "webrtc/system_wrappers/include/clock.h"
 #include "webrtc/system_wrappers/include/file_wrapper.h"
+#include "webrtc/system_wrappers/include/logging.h"
 
 #ifdef ENABLE_RTC_EVENT_LOG
 // Files generated at build-time by the protobuf compiler.
 #ifdef WEBRTC_ANDROID_PLATFORM_BUILD
 #include "external/webrtc/webrtc/call/rtc_event_log.pb.h"
 #else
 #include "webrtc/call/rtc_event_log.pb.h"
 #endif
 #endif
 
 namespace webrtc {
 
 #ifndef ENABLE_RTC_EVENT_LOG
 
 // No-op implementation if flag is not set.
-class RtcEventLogImpl final : public RtcEventLog {
+class RtcEventLogNullImpl final : public RtcEventLog {
  public:
-  void SetBufferDuration(int64_t buffer_duration_us) override {}
-  void StartLogging(const std::string& file_name, int duration_ms) override {}
-  bool StartLogging(rtc::PlatformFile log_file) override { return false; }
-  void StopLogging(void) override {}
+  bool StartLogging(const std::string& file_name,
+                    int64_t max_size_bytes) override {
+    return false;
+  }
+  bool StartLogging(rtc::PlatformFile platform_file,
+                    int64_t max_size_bytes) override {
+    return false;
+  }
+  void StopLogging() override {}
   void LogVideoReceiveStreamConfig(
       const VideoReceiveStream::Config& config) override {}
   void LogVideoSendStreamConfig(
       const VideoSendStream::Config& config) override {}
   void LogRtpHeader(PacketDirection direction,
                     MediaType media_type,
                     const uint8_t* header,
                     size_t packet_length) override {}
   void LogRtcpPacket(PacketDirection direction,
                      MediaType media_type,
                      const uint8_t* packet,
                      size_t length) override {}
   void LogAudioPlayout(uint32_t ssrc) override {}
   void LogBwePacketLossEvent(int32_t bitrate,
                              uint8_t fraction_loss,
                              int32_t total_packets) override {}
 };
 
 #else  // ENABLE_RTC_EVENT_LOG is defined
 
 class RtcEventLogImpl final : public RtcEventLog {
  public:
-  RtcEventLogImpl();
+  explicit RtcEventLogImpl(const Clock* clock);
+  ~RtcEventLogImpl() override;
 
-  void SetBufferDuration(int64_t buffer_duration_us) override;
-  void StartLogging(const std::string& file_name, int duration_ms) override;
-  bool StartLogging(rtc::PlatformFile log_file) 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;
   void StopLogging() override;
   void LogVideoReceiveStreamConfig(
       const VideoReceiveStream::Config& config) override;
   void LogVideoSendStreamConfig(const VideoSendStream::Config& config) override;
   void LogRtpHeader(PacketDirection direction,
                     MediaType media_type,
                     const uint8_t* header,
                     size_t packet_length) override;
   void LogRtcpPacket(PacketDirection direction,
                      MediaType media_type,
                      const uint8_t* packet,
                      size_t length) override;
   void LogAudioPlayout(uint32_t ssrc) override;
   void LogBwePacketLossEvent(int32_t bitrate,
                              uint8_t fraction_loss,
                              int32_t total_packets) override;
 
  private:
-  // Starts logging. This function assumes the file_ has been opened succesfully
-  // and that the start_time_us_ and _duration_us_ have been set.
-  void StartLoggingLocked() EXCLUSIVE_LOCKS_REQUIRED(crit_);
-  // Stops logging and clears the stored data and buffers.
-  void StopLoggingLocked() EXCLUSIVE_LOCKS_REQUIRED(crit_);
-  // Adds a new event to the logfile if logging is active, or adds it to the
-  // list of recent log events otherwise.
-  void HandleEvent(rtclog::Event* event) EXCLUSIVE_LOCKS_REQUIRED(crit_);
-  // Writes the event to the file. Note that this will destroy the state of the
-  // input argument.
-  void StoreToFile(rtclog::Event* event) EXCLUSIVE_LOCKS_REQUIRED(crit_);
-  // Adds the event to the list of recent events, and removes any events that
-  // are too old and no longer fall in the time window.
-  void AddRecentEvent(const rtclog::Event& event)
+  // The normal state progression is
+  // TO_MEMORY -> STARTING_FILE -> TO_FILE -> STOPPING_FILE -> TO_MEMORY
+  // but it can also go directly from STARTING_FILE to STOPPING_FILE.
+  enum LogState {
+    TO_MEMORY,
+    STARTING_FILE,
+    TO_FILE,
+    STOPPING_FILE,
+    TERMINATE_THREAD,
+  };
+
+  static bool ThreadOutputFunction(void* obj);
+  void TerminateThread();
+
+  bool AppendEventToString(rtclog::Event* event)
       EXCLUSIVE_LOCKS_REQUIRED(crit_);
+  void AppendEventToHistory(const rtclog::Event& event)
+      EXCLUSIVE_LOCKS_REQUIRED(crit_);
+  bool ProcessInOrder(bool memory, int64_t current_time)
+      EXCLUSIVE_LOCKS_REQUIRED(crit_);
+  void LogToMemory() EXCLUSIVE_LOCKS_REQUIRED(crit_);

[the sun, 2016/02/10 10:47:23]

To make things even clearer, the parts of this class that are only accessed from
the logging thread, could be pulled out into a separate class, in its separate
file.

class LoggerThread {
 public:
  LoggerThread(SwapQueue<rtclog::Event>* config_queue, SwapQueue<rtclog::Event>*
config_queue, ..., rtc::Event* file_finished_event);
...
then many fields could be tucked away in this class.

You could also make the state machine simpler by adding a
SwapQueue<MyMessageStruct> to the ctor, and pass state transitions in the queue
instead of synchronizing access to the state_ variable. The message struct could
be a proto buffer or just a simple struct/union (it would then be defined in
rtc_event_logger.h together with the state enum).

[terelius, 2016/02/11 21:27:28]

One side effect of moving the state to a separate class is that we would not be
able to know whether a call (e.g. StartLogging) has succeeded or not. We would
probably have to duplicate at least parts of the state in both the event log and
the new class.

[the sun, 2016/02/12 09:37:32]

For StartLogging() it would likely make sense to open the output file in
RtcEventLogImpl and pass the file pointer to the thread class in the START_FILE
message. For StopLogging() my suggestion was to keep the event as now, to be
able to block until the whole buffer has been purged and the file closed (by the
logging thread). If you want more status information at that point, you could
use an event queue in the thread->rtceventlogimpl direction too.

[terelius, 2016/02/12 14:13:38]

But even if StartLogging opens the file we wouldn't know whether we are already
logging. We could add a bool in the RtcEventLog, set it in   StartLogging and
clear it in StopLogging, but that would not detect whether the log has stopped
on it's own due to file size or time limitations.

Another problem is how we terminate the thread. Presumably, the three main types
of messages would be START_FILE, STOP_FILE and TERMINATE_THREAD. The
TERMINATE_THREAD message would be sent by the RtcEventLog destructor.  If we
process all events (including messages) in time order, then the destructor would
wait for any remaining events to be copied (to file ot memory). This is clearly
not the ideal behavior if the user is trying to close a malfunctioning browser
tab. Even if we treated the messages separately from other events we could end
up with the same problem if there is a STOP_FILE message before the
TERMINATE_THREAD message.

[the sun, 2016/02/12 15:37:27]

You probably want to process the control messages first in the thread. If you
want everything in a strict order, you should likely use the same queue. But
really, if there are pending messages in the event queues when a stop message is
forwarded, you can purge those queues, or just drop the pending messages. I
don't think it matters a lot.

If the time that the destructor may take is a problem (it sounds rather unlikely
IMO) then you could allow an interface to register a Future to signal that the
file has been written, instead of using the event to signal. That is one option,
but I doubt we need that.

[terelius, 2016/02/12 16:22:24]

Though if we process the control messages before the events, what would happen
if we are logging to file and the message queue contains STOP_FILE followed by
START_FILE? We'd first pop the STOP_FILE message and set the new time limit.
Then we'd look at the START_FILE and presumably discard it because we are
already logging, right? If we process all control messages first, it seems we
wouldn't be using a message queue but rather the same type of state signalling
that is in place right now.

If we process the control messages first, then I agree that the destructor won't
be a problem. What I am worried about is if we process all events and messages
strictly in time order so we won't see the TERMINATE_THREAD message until we
have written all RTP events. I don't know whether it'll be significant, but in
any case I think it is bad practice to do extensive computations or unnecessary
I/O in the destructor.

[the sun, 2016/02/16 13:29:15]

I thought a START_FILE message would contain the file resource to log to. Then
in this case STOP_FILE would purge current buffers and close the current file,
and START_FILE starts logging to a new file, but writes nothing to it, yet.
a) If processing control messages first, you can choose what to do when you see
the TERMINATE_THREAD: purge buffers or close file+terminate. I don't think it
matters much if you have a few messages pending in the queues (the 10s backlog
should have been written out at START_FILE).
b) Destructors are not different from other code. Compare:

delete myResource;    // d-tor purges.

to 

myResource.Purge();
delete myResource;    // d-tor doesn't purge.

They run on the same thread, same order. It's not like the d-tor runs under any
real time requirements.

The biggest "downside" with doing work in the d-tor is that calls to them are
implicit, so they may be called in a context you don't expect (that is, if you
do not take proper care managing your object life times). A less surprising
attribute is that you can't return an error code from the d-tor, or (in C++)
throw an exception.

[terelius, 2016/02/16 14:36:58]

Oh, so you are saying that you would always process one "batch" of events in
between each control message?

Yeah, START_FILE would contain the file resource, but my view was that we would
process all control messages before the regular events. Since we can't close the
file until we have processed all events earlier than the  STOP message, we
wouldn't actually have stopped when we get the second START event.
Assuming we're processing control messages first, then we're in agreement
regarding the destructor.

That said, guaranteeing exception-freeness is quite difficult imo. In
particular, new can throw (unless we're using some special compiler flags?), so
I tend to assume that any computation I don't fully control can throw. The
second reason is more of a UI thing; people might be ok with a loading screen
because initialization takes a while, but program termination is still expected
to be near instantaneous. If we were processing regular events before control
messages, then we would copy all regular events to the history buffer and after
that read the TERMINATE_THREAD message and throw away the history buffer. My
point was just that I want to process control messages before regular events to
avoid this unnecessary buffer copying (or similar if logging to file).

[the sun, 2016/02/17 09:44:20]

No, I would handle all the control messages first, so that all updates to the
thread state go first, then handle the event queues. Yes, that would mean giving
up on precise timing between the control messages and log events, but I don't
think that would matter too much. The important bit is that events in the log
are in order.

That said, the devil is in the details, so I can't tell you exactly *how* the
solution would end up if I were to do it. But I would definitely break out the
state that belongs to the logging thread into its own class, so that we can
avoid explicit use of locks. 
We do not use exceptions in Chromium/WebRTC C++ and I believe the code is
compiled without exception support/RTTI (otherwise it would unnecessarily bloat
the binary and invite use of exceptions). In Chrome, set_new_handler() is used
in the render process to output log messages in OOM situations, though curiously
the process isn't terminated so we will continue, likely causing an access
violation instead.

Likewise, you can assume that no libraries that Chrome uses will throw
exceptions, and certainly not the FileWrapper you use to write the file here.

+  void StartLogFile() EXCLUSIVE_LOCKS_REQUIRED(crit_);
+  void LogToFile() EXCLUSIVE_LOCKS_REQUIRED(crit_);
+  void StopLogFile() EXCLUSIVE_LOCKS_REQUIRED(crit_);
+  void WriteLog();
+
+  // Message queues for passing events to the logging thread.
+  SwapQueue<rtclog::Event> config_buffer_;

[the sun, 2016/02/10 10:47:23]

I think it is better to call these config_queue_ etc.

[terelius, 2016/02/11 21:27:28]

Done.

+  SwapQueue<rtclog::Event> rtp_buffer_;
+  SwapQueue<rtclog::Event> rtcp_buffer_;
+  SwapQueue<rtclog::Event> acm_playout_buffer_;
+  SwapQueue<rtclog::Event> bwe_loss_buffer_;
 
   rtc::CriticalSection crit_;
-  rtc::scoped_ptr<FileWrapper> file_ GUARDED_BY(crit_) =
-      rtc::scoped_ptr<FileWrapper>(FileWrapper::Create());
-  rtc::PlatformFile platform_file_ GUARDED_BY(crit_) =
-      rtc::kInvalidPlatformFileValue;
-  rtclog::EventStream stream_ GUARDED_BY(crit_);
-  std::deque<rtclog::Event> recent_log_events_ GUARDED_BY(crit_);
-  std::vector<rtclog::Event> config_events_ GUARDED_BY(crit_);
 
-  // Microseconds to record log events, before starting the actual log.
-  int64_t buffer_duration_us_ GUARDED_BY(crit_);
-  bool currently_logging_ GUARDED_BY(crit_);
-  int64_t start_time_us_ GUARDED_BY(crit_);
-  int64_t duration_us_ GUARDED_BY(crit_);
+  // History containing the most recent events (~ 10 s).
+  RingBuffer<rtclog::Event> history_ GUARDED_BY(crit_);
+
+  // History containing all past configuration events.
+  std::vector<rtclog::Event> config_history_ GUARDED_BY(crit_);
+
+  LogState state_ GUARDED_BY(crit_);
+  rtc::scoped_ptr<FileWrapper> file_ GUARDED_BY(crit_);
+  rtc::PlatformThread thread_ GUARDED_BY(crit_);
+
+  int64_t max_size_bytes_ GUARDED_BY(crit_);
+  int64_t written_bytes_ GUARDED_BY(crit_);
+  int64_t start_time_ GUARDED_BY(crit_);
+  int64_t stop_time_ GUARDED_BY(crit_);
+
+  // We want to extract the elements from the buffers in timestamp order
+  // and, because we can't peek into the SwapQueue, we need to keep the first
+  // elements outside.
+  rtclog::Event config_event_;
+  rtclog::Event rtp_event_;
+  rtclog::Event rtcp_event_;
+  rtclog::Event playout_event_;
+  rtclog::Event loss_event_;
+  bool valid_config_event_;

[the sun, 2016/02/10 10:47:23]

You may be able to avoid these by using rtc::Optional

[terelius, 2016/02/11 21:27:28]

There are several reasons why Optional won't work. To begin with, although our
Optional always contains a value, the std::optional does not, so the code would
break if we move to std::optional in the future. Secondly, the normal access
functions actually check that has_value is set when you try to access the value.
In other words, you can't obtain a pointer to pass to the SwapQueue::Remove()
unless the has_value bit is set. Finally, there is no clean way to set or clear
the has_value bit, so we can't get the return value from Insert and Remove into
the Optional..

+  bool valid_rtp_event_;
+  bool valid_rtcp_event_;
+  bool valid_playout_event_;
+  bool valid_loss_event_;
+
+  // Temporary space for serializing profobuf data.
+  std::string output_string_ GUARDED_BY(crit_);
+
+  rtc::Event wake_up_;
+  rtc::Event file_finished_;
+
   const Clock* const clock_;
+
+  RTC_DISALLOW_IMPLICIT_CONSTRUCTORS(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.
 
-// Do not add default return values to the conversion functions in this
-// unnamed namespace. The intention is to make the compiler warn if anyone
-// adds unhandled new events/modes/etc.
-
 rtclog::VideoReceiveConfig_RtcpMode ConvertRtcpMode(RtcpMode rtcp_mode) {
   switch (rtcp_mode) {
     case RtcpMode::kCompound:
       return rtclog::VideoReceiveConfig::RTCP_COMPOUND;
     case RtcpMode::kReducedSize:
       return rtclog::VideoReceiveConfig::RTCP_REDUCEDSIZE;
     case RtcpMode::kOff:
       RTC_NOTREACHED();
       return rtclog::VideoReceiveConfig::RTCP_COMPOUND;
   }
   RTC_NOTREACHED();
   return rtclog::VideoReceiveConfig::RTCP_COMPOUND;
 }
 
 rtclog::MediaType ConvertMediaType(MediaType media_type) {
   switch (media_type) {
     case MediaType::ANY:
       return rtclog::MediaType::ANY;
     case MediaType::AUDIO:
       return rtclog::MediaType::AUDIO;
     case MediaType::VIDEO:
       return rtclog::MediaType::VIDEO;
     case MediaType::DATA:
       return rtclog::MediaType::DATA;
   }
   RTC_NOTREACHED();
   return rtclog::ANY;
 }
 
+// The RTP and RTCP buffers reserve space for twice the expected number of
+// sent packets because they also contain received packets.
+const int kStreamConfigsPerSecond = 64;  // 16 clients w. 4 streams each.
+const int kRtpPacketsPerSecond = 500;  // 125 sent video packets/s @ 1 Mbps.
+const int kRtcpPacketsPerSecond = 40;  // Assume RTCP sent 20 times/s.
+const int kPlayoutsPerSecond = 100;    // Playout called every 10 ms.
+const int kBweUpdatesPerSecond = 20;   // One BWE update per RTCP packet.
+const int kEventsInHistory = 10000;
+
 }  // namespace
 
-namespace {
-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;
-}
-}  // namespace
 
 // RtcEventLogImpl member functions.
-RtcEventLogImpl::RtcEventLogImpl()
-    : file_(FileWrapper::Create()),
-      stream_(),
-      buffer_duration_us_(10000000),
-      currently_logging_(false),
-      start_time_us_(0),
-      duration_us_(0),
-      clock_(Clock::GetRealTimeClock()) {
+RtcEventLogImpl::RtcEventLogImpl(const Clock* clock)
+    // Allocate buffers for roughly one second of history.
+    : config_buffer_(kStreamConfigsPerSecond),
+      rtp_buffer_(kRtpPacketsPerSecond),
+      rtcp_buffer_(kRtcpPacketsPerSecond),
+      acm_playout_buffer_(kPlayoutsPerSecond),
+      bwe_loss_buffer_(kBweUpdatesPerSecond),
+      crit_(),
+      history_(kEventsInHistory),
+      config_history_(),
+      state_(TO_MEMORY),
+      file_(FileWrapper::Create()),
+      thread_(&ThreadOutputFunction, this, "RtcEventLog thread"),
+      max_size_bytes_(std::numeric_limits<int64_t>::max()),
+      written_bytes_(0),
+      start_time_(0),
+      stop_time_(std::numeric_limits<int64_t>::max()),
+      config_event_(),
+      rtp_event_(),
+      rtcp_event_(),
+      playout_event_(),
+      loss_event_(),
+      valid_config_event_(false),
+      valid_rtp_event_(false),
+      valid_rtcp_event_(false),
+      valid_playout_event_(false),
+      valid_loss_event_(false),
+      output_string_(),
+      wake_up_(false, false),
+      file_finished_(false, false),
+      clock_(clock) {
+  thread_.Start();
 }
 
-void RtcEventLogImpl::SetBufferDuration(int64_t buffer_duration_us) {
-  rtc::CritScope lock(&crit_);
-  buffer_duration_us_ = buffer_duration_us;
+RtcEventLogImpl::~RtcEventLogImpl() {
+  TerminateThread();
+  thread_.Stop();
 }
 
-void RtcEventLogImpl::StartLogging(const std::string& file_name,
-                                   int duration_ms) {
+bool RtcEventLogImpl::StartLogging(const std::string& file_name,
+                                   int64_t max_size_bytes) {
   rtc::CritScope lock(&crit_);
-  if (currently_logging_) {
-    StopLoggingLocked();
+  if (state_ != TO_MEMORY) {
+    return false;
   }
+  file_.reset(FileWrapper::Create());
   if (file_->OpenFile(file_name.c_str(), false) != 0) {
-    return;
-  }
-  start_time_us_ = clock_->TimeInMicroseconds();
-  duration_us_ = static_cast<int64_t>(duration_ms) * 1000;
-  StartLoggingLocked();
-}
-
-bool RtcEventLogImpl::StartLogging(rtc::PlatformFile log_file) {
-  rtc::CritScope lock(&crit_);
-
-  if (currently_logging_) {
-    StopLoggingLocked();
-  }
-  RTC_DCHECK(platform_file_ == rtc::kInvalidPlatformFileValue);
-
-  FILE* file_stream = rtc::FdopenPlatformFileForWriting(log_file);
-  if (!file_stream) {
-    rtc::ClosePlatformFile(log_file);
     return false;
   }
 
-  if (file_->OpenFromFileHandle(file_stream, true, false) != 0) {
-    rtc::ClosePlatformFile(log_file);
-    return false;
-  }
-  platform_file_ = log_file;
-  // Set the start time and duration to keep logging for 10 minutes.
-  start_time_us_ = clock_->TimeInMicroseconds();
-  duration_us_ = 10 * 60 * 1000000;
-  StartLoggingLocked();
+  max_size_bytes_ = max_size_bytes;
+  start_time_ = clock_->TimeInMicroseconds();
+  stop_time_ = std::numeric_limits<int64_t>::max();
+  state_ = STARTING_FILE;
   return true;
 }
 
-void RtcEventLogImpl::StartLoggingLocked() {
-  currently_logging_ = true;
+bool RtcEventLogImpl::StartLogging(rtc::PlatformFile platform_file,
+                                   int64_t max_size_bytes) {
+  rtc::CritScope lock(&crit_);
+  if (state_ != TO_MEMORY) {
+    return false;
+  }
 
-  // Write all old configuration events to the log file.
-  for (auto& event : config_events_) {
-    StoreToFile(&event);
+  FILE* file_handle = rtc::FdopenPlatformFileForWriting(platform_file);
+  if (!file_handle) {
+    return false;
   }
-  // Write all recent configuration events to the log file, and
-  // write all other recent events to the log file, ignoring any old events.
-  for (auto& event : recent_log_events_) {
-    if (IsConfigEvent(event)) {
-      StoreToFile(&event);
-      config_events_.push_back(event);
-    } else if (event.timestamp_us() >= start_time_us_ - buffer_duration_us_) {
-      StoreToFile(&event);
-    }
+  file_.reset(FileWrapper::Create());
+  if (file_->OpenFromFileHandle(file_handle, true, false) != 0) {
+    return false;
   }
-  recent_log_events_.clear();
-  // Write a LOG_START event to the file.
-  rtclog::Event start_event;
-  start_event.set_timestamp_us(start_time_us_);
-  start_event.set_type(rtclog::Event::LOG_START);
-  StoreToFile(&start_event);
+
+  max_size_bytes_ = max_size_bytes;
+  start_time_ = clock_->TimeInMicroseconds();
+  stop_time_ = std::numeric_limits<int64_t>::max();
+  state_ = STARTING_FILE;
+  return true;
 }
 
 void RtcEventLogImpl::StopLogging() {
-  rtc::CritScope lock(&crit_);
-  StopLoggingLocked();
+  {
+    rtc::CritScope lock(&crit_);
+    if (state_ == TO_FILE || state_ == STARTING_FILE) {
+      stop_time_ = clock_->TimeInMicroseconds();
+      wake_up_.Set();  // Request the output thread to wake up.
+    } else {
+      return;
+    }
+  }
+  file_finished_.Wait(
+      rtc::Event::kForever);  // Wait for the log file to be completed.
 }
 
 void RtcEventLogImpl::LogVideoReceiveStreamConfig(
     const VideoReceiveStream::Config& config) {
-  rtc::CritScope lock(&crit_);
-
   rtclog::Event event;
   event.set_timestamp_us(clock_->TimeInMicroseconds());
   event.set_type(rtclog::Event::VIDEO_RECEIVER_CONFIG_EVENT);
 
   rtclog::VideoReceiveConfig* receiver_config =
       event.mutable_video_receiver_config();
   receiver_config->set_remote_ssrc(config.rtp.remote_ssrc);
   receiver_config->set_local_ssrc(config.rtp.local_ssrc);
 
   receiver_config->set_rtcp_mode(ConvertRtcpMode(config.rtp.rtcp_mode));
@@ skipping 11 matching lines @@
         receiver_config->add_header_extensions();
     extension->set_name(e.name);
     extension->set_id(e.id);
   }
 
   for (const auto& d : config.decoders) {
     rtclog::DecoderConfig* decoder = receiver_config->add_decoders();
     decoder->set_name(d.payload_name);
     decoder->set_payload_type(d.payload_type);
   }
-  HandleEvent(&event);
+  if (!config_buffer_.Insert(&event)) {
+    LOG(LS_WARNING) << "Config buffer full. Not logging config event.";
+  }
 }
 
 void RtcEventLogImpl::LogVideoSendStreamConfig(
     const VideoSendStream::Config& config) {
-  rtc::CritScope lock(&crit_);
-
   rtclog::Event event;
   event.set_timestamp_us(clock_->TimeInMicroseconds());
   event.set_type(rtclog::Event::VIDEO_SENDER_CONFIG_EVENT);
 
   rtclog::VideoSendConfig* sender_config = event.mutable_video_sender_config();
 
   for (const auto& ssrc : config.rtp.ssrcs) {
     sender_config->add_ssrcs(ssrc);
   }
 
   for (const auto& e : config.rtp.extensions) {
     rtclog::RtpHeaderExtension* extension =
         sender_config->add_header_extensions();
     extension->set_name(e.name);
     extension->set_id(e.id);
   }
 
   for (const auto& rtx_ssrc : config.rtp.rtx.ssrcs) {
     sender_config->add_rtx_ssrcs(rtx_ssrc);
   }
   sender_config->set_rtx_payload_type(config.rtp.rtx.payload_type);
 
   rtclog::EncoderConfig* encoder = sender_config->mutable_encoder();
   encoder->set_name(config.encoder_settings.payload_name);
   encoder->set_payload_type(config.encoder_settings.payload_type);
-  HandleEvent(&event);
+  if (!config_buffer_.Insert(&event)) {
+    LOG(LS_WARNING) << "Config buffer full. Not logging config event.";
+  }
 }
 
 void RtcEventLogImpl::LogRtpHeader(PacketDirection direction,
                                    MediaType media_type,
                                    const uint8_t* header,
                                    size_t packet_length) {
   // Read header length (in bytes) from packet data.
   if (packet_length < 12u) {
     return;  // Don't read outside the packet.
   }
   const bool x = (header[0] & 0x10) != 0;
   const uint8_t cc = header[0] & 0x0f;
   size_t header_length = 12u + cc * 4u;
 
   if (x) {
     if (packet_length < 12u + cc * 4u + 4u) {
       return;  // Don't read outside the packet.
     }
     size_t x_len = ByteReader<uint16_t>::ReadBigEndian(header + 14 + cc * 4);
     header_length += (x_len + 1) * 4;
   }
 
-  rtc::CritScope lock(&crit_);
   rtclog::Event rtp_event;
   rtp_event.set_timestamp_us(clock_->TimeInMicroseconds());
   rtp_event.set_type(rtclog::Event::RTP_EVENT);
   rtp_event.mutable_rtp_packet()->set_incoming(direction == kIncomingPacket);
   rtp_event.mutable_rtp_packet()->set_type(ConvertMediaType(media_type));
   rtp_event.mutable_rtp_packet()->set_packet_length(packet_length);
   rtp_event.mutable_rtp_packet()->set_header(header, header_length);
-  HandleEvent(&rtp_event);
+  if (!rtp_buffer_.Insert(&rtp_event)) {
+    LOG(LS_WARNING) << "RTP buffer full. Not logging RTP packet.";
+  }
 }
 
 void RtcEventLogImpl::LogRtcpPacket(PacketDirection direction,
                                     MediaType media_type,
                                     const uint8_t* packet,
                                     size_t length) {
-  rtc::CritScope lock(&crit_);
   rtclog::Event rtcp_event;
   rtcp_event.set_timestamp_us(clock_->TimeInMicroseconds());
   rtcp_event.set_type(rtclog::Event::RTCP_EVENT);
   rtcp_event.mutable_rtcp_packet()->set_incoming(direction == kIncomingPacket);
   rtcp_event.mutable_rtcp_packet()->set_type(ConvertMediaType(media_type));
 
   RTCPUtility::RtcpCommonHeader header;
   const uint8_t* block_begin = packet;
   const uint8_t* packet_end = packet + length;
   RTC_DCHECK(length <= IP_PACKET_SIZE);
@@ skipping 31 matching lines @@
         FALLTHROUGH();
       default:
         // We don't log sender descriptions, application defined messages
         // or message blocks of unknown type.
         break;
     }
 
     block_begin += block_size;
   }
   rtcp_event.mutable_rtcp_packet()->set_packet_data(buffer, buffer_length);
-  HandleEvent(&rtcp_event);
+  if (!rtcp_buffer_.Insert(&rtcp_event)) {
+    LOG(LS_WARNING) << "RTCP buffer full. Not logging RTCP packet.";
+  }
 }
 
 void RtcEventLogImpl::LogAudioPlayout(uint32_t ssrc) {
-  rtc::CritScope lock(&crit_);
   rtclog::Event event;
   event.set_timestamp_us(clock_->TimeInMicroseconds());
   event.set_type(rtclog::Event::AUDIO_PLAYOUT_EVENT);
   auto playout_event = event.mutable_audio_playout_event();
   playout_event->set_local_ssrc(ssrc);
-  HandleEvent(&event);
+  if (!acm_playout_buffer_.Insert(&event)) {
+    LOG(LS_WARNING) << "Playout buffer full. Not logging ACM playout.";
+  }
 }
 
 void RtcEventLogImpl::LogBwePacketLossEvent(int32_t bitrate,
                                             uint8_t fraction_loss,
                                             int32_t total_packets) {
-  rtc::CritScope lock(&crit_);
   rtclog::Event event;
   event.set_timestamp_us(clock_->TimeInMicroseconds());
   event.set_type(rtclog::Event::BWE_PACKET_LOSS_EVENT);
   auto bwe_event = event.mutable_bwe_packet_loss_event();
   bwe_event->set_bitrate(bitrate);
   bwe_event->set_fraction_loss(fraction_loss);
   bwe_event->set_total_packets(total_packets);
-  HandleEvent(&event);
-}
-
-void RtcEventLogImpl::StopLoggingLocked() {
-  if (currently_logging_) {
-    currently_logging_ = false;
-    // Create a LogEnd event
-    rtclog::Event event;
-    event.set_timestamp_us(clock_->TimeInMicroseconds());
-    event.set_type(rtclog::Event::LOG_END);
-    // Store the event and close the file
-    RTC_DCHECK(file_->Open());
-    StoreToFile(&event);
-    file_->CloseFile();
-    if (platform_file_ != rtc::kInvalidPlatformFileValue) {
-      rtc::ClosePlatformFile(platform_file_);
-      platform_file_ = rtc::kInvalidPlatformFileValue;
-    }
-  }
-  RTC_DCHECK(!file_->Open());
-  stream_.Clear();
-}
-
-void RtcEventLogImpl::HandleEvent(rtclog::Event* event) {
-  if (currently_logging_) {
-    if (clock_->TimeInMicroseconds() < start_time_us_ + duration_us_) {
-      StoreToFile(event);
-      return;
-    }
-    StopLoggingLocked();
-  }
-  AddRecentEvent(*event);
-}
-
-void RtcEventLogImpl::StoreToFile(rtclog::Event* event) {
-  // Reuse the same object at every log event.
-  if (stream_.stream_size() < 1) {
-    stream_.add_stream();
-  }
-  RTC_DCHECK_EQ(stream_.stream_size(), 1);
-  stream_.mutable_stream(0)->Swap(event);
-  // TODO(terelius): Doesn't this create a new EventStream per event?
+  if (!bwe_loss_buffer_.Insert(&event)) {
+    LOG(LS_WARNING) << "BWE loss buffer full. Not logging BWE update.";
+  }
+}
+
+bool RtcEventLogImpl::AppendEventToString(rtclog::Event* event) {
+  rtclog::EventStream event_stream;
+  event_stream.add_stream();
+  event_stream.mutable_stream(0)->Swap(event);
+  // TODO(terelius): We create a new event strem per event, but it will look
+  // like a single stream when we read it back from file.
   // Is this guaranteed to work e.g. in future versions of protobuf?
-  std::string dump_buffer;
-  stream_.SerializeToString(&dump_buffer);
-  file_->Write(dump_buffer.data(), dump_buffer.size());
-}
-
-void RtcEventLogImpl::AddRecentEvent(const rtclog::Event& event) {
-  recent_log_events_.push_back(event);
-  while (recent_log_events_.front().timestamp_us() <
-         event.timestamp_us() - buffer_duration_us_) {
-    if (IsConfigEvent(recent_log_events_.front())) {
-      config_events_.push_back(recent_log_events_.front());
-    }
-    recent_log_events_.pop_front();
-  }
-}
-
+  bool stop = true;
+  if (written_bytes_ + static_cast<int64_t>(output_string_.size()) +
+          event_stream.ByteSize() <=
+      max_size_bytes_) {
+    event_stream.AppendToString(&output_string_);
+    stop = false;
+  }
+  // Swap the event back so that we don't mix event types in the queues.
+  event_stream.mutable_stream(0)->Swap(event);
+  return stop;
+}
+
+void RtcEventLogImpl::AppendEventToHistory(const rtclog::Event& event) {
+  history_.push_back(event);
+}
+
+// Traverses the SwapQueues in timestamp order and copies all events earlier
+// than |current_time| either to the history or to a string that will be
+// written to disc.
+bool RtcEventLogImpl::ProcessInOrder(bool memory, int64_t current_time) {
+  bool stop = false;
+  enum EventType {
+    CONFIG_EVENT,
+    RTP_EVENT,
+    RTCP_EVENT,
+    PLAYOUT_EVENT,
+    LOSS_EVENT
+  };
+
+  // Extract the head of each queue.
+  if (!valid_config_event_) {
+    valid_config_event_ = config_buffer_.Remove(&config_event_);
+  }
+
+  if (!valid_rtp_event_) {
+    valid_rtp_event_ = rtp_buffer_.Remove(&rtp_event_);
+  }
+
+  if (!valid_rtcp_event_) {
+    valid_rtcp_event_ = rtcp_buffer_.Remove(&rtcp_event_);
+  }
+
+  if (!valid_playout_event_) {
+    valid_playout_event_ = acm_playout_buffer_.Remove(&playout_event_);
+  }
+
+  if (!valid_loss_event_) {
+    valid_loss_event_ = bwe_loss_buffer_.Remove(&loss_event_);
+  }
+
+  while ((valid_config_event_ || valid_rtp_event_ || valid_rtcp_event_ ||
+          valid_playout_event_ || valid_loss_event_) &&
+         !stop) {
+    // Find the earliest event (in timestamp order).
+    EventType type = CONFIG_EVENT;
+    int64_t first_timestamp =
+        (valid_config_event_ ? config_event_.timestamp_us()
+                             : std::numeric_limits<int64_t>::max());
+    if (valid_rtp_event_ && (rtp_event_.timestamp_us() < first_timestamp)) {
+      first_timestamp = rtp_event_.timestamp_us();
+      type = RTP_EVENT;
+    }
+    if (valid_rtcp_event_ && (rtcp_event_.timestamp_us() < first_timestamp)) {
+      first_timestamp = rtcp_event_.timestamp_us();
+      type = RTCP_EVENT;
+    }
+    if (valid_playout_event_ &&
+        playout_event_.timestamp_us() < first_timestamp) {
+      first_timestamp = playout_event_.timestamp_us();
+      type = PLAYOUT_EVENT;
+    }
+    if (valid_loss_event_ && loss_event_.timestamp_us() < first_timestamp) {
+      first_timestamp = loss_event_.timestamp_us();
+      type = LOSS_EVENT;
+    }
+
+    if (first_timestamp > current_time) {
+      // We have handled all events earlier than current_time.
+      break;
+    }
+    // Serialize the event and fetch the next event of that type.
+    switch (type) {
+      case CONFIG_EVENT:
+        if (!memory) {
+          stop = AppendEventToString(&config_event_);
+        }
+        config_history_.push_back(config_event_);
+        if (!stop) {
+          valid_config_event_ = config_buffer_.Remove(&config_event_);
+        }
+        break;
+      case RTP_EVENT:
+        if (memory) {
+          AppendEventToHistory(rtp_event_);
+        } else {
+          stop = AppendEventToString(&rtp_event_);
+        }
+        if (!stop) {
+          valid_rtp_event_ = rtp_buffer_.Remove(&rtp_event_);
+        }
+        break;
+      case RTCP_EVENT:
+        if (memory) {
+          AppendEventToHistory(rtcp_event_);
+        } else {
+          stop = AppendEventToString(&rtcp_event_);
+        }
+        if (!stop) {
+          valid_rtcp_event_ = rtcp_buffer_.Remove(&rtcp_event_);
+        }
+        break;
+      case PLAYOUT_EVENT:
+        if (memory) {
+          AppendEventToHistory(playout_event_);
+        } else {
+          stop = AppendEventToString(&playout_event_);
+        }
+        if (!stop) {
+          valid_playout_event_ = acm_playout_buffer_.Remove(&playout_event_);
+        }
+        break;
+      case LOSS_EVENT:
+        if (memory) {
+          AppendEventToHistory(loss_event_);
+        } else {
+          stop = AppendEventToString(&loss_event_);
+        }
+        if (!stop) {
+          valid_loss_event_ = bwe_loss_buffer_.Remove(&loss_event_);
+        }
+        break;
+    }
+  }
+  // We want to stop logging either if we have struck the file size limit
+  // or if we have logged all events older than |stop_time_|
+  return stop || (current_time > stop_time_);
+}
+
+void RtcEventLogImpl::LogToMemory() {
+  // Process each event in order and append it to the appropriate history_.
+  int64_t current_time = clock_->TimeInMicroseconds();
+  ProcessInOrder(true, current_time);
+}
+
+void RtcEventLogImpl::StartLogFile() {
+  bool stop = false;
+  output_string_.clear();
+
+  // Create and serialize the LOG_START event.
+  rtclog::Event start_event;
+  start_event.set_timestamp_us(start_time_);
+  start_event.set_type(rtclog::Event::LOG_START);
+  AppendEventToString(&start_event);
+
+  // Serialize the config information for all old streams.
+  for (rtclog::Event& event : config_history_) {
+    AppendEventToString(&event);
+  }
+
+  // Serialize the events in the event queue.
+  for (int i = 0; !history_.empty() && !stop; i++) {
+    stop = AppendEventToString(&history_.front());
+    if (!stop) {
+      history_.pop_front();
+    }
+  }
+
+  // Write to file.
+  file_->Write(output_string_.data(), output_string_.size());
+  written_bytes_ += output_string_.size();
+  state_ = (stop ? STOPPING_FILE : TO_FILE);
+}
+
+void RtcEventLogImpl::LogToFile() {
+  output_string_.clear();
+
+  // Process each event in order and append it to the output_string_.
+  int64_t current_time = clock_->TimeInMicroseconds();
+  bool stop = ProcessInOrder(false, current_time);
+  if (stop || stop_time_ <= current_time) {
+    state_ = STOPPING_FILE;
+  }
+
+  // Write string to file.
+  file_->Write(output_string_.data(), output_string_.size());
+  written_bytes_ += output_string_.size();
+}
+
+void RtcEventLogImpl::StopLogFile() {
+  output_string_.clear();
+
+  rtclog::Event end_event;
+  end_event.set_timestamp_us(stop_time_);
+  end_event.set_type(rtclog::Event::LOG_END);
+  AppendEventToString(&end_event);
+
+  if (written_bytes_ + static_cast<int64_t>(output_string_.size()) <=
+      max_size_bytes_) {
+    file_->Write(output_string_.data(), output_string_.size());
+    written_bytes_ += output_string_.size();
+  }
+
+  state_ = TO_MEMORY;
+  max_size_bytes_ = std::numeric_limits<int64_t>::max();
+  written_bytes_ = 0;
+  start_time_ = 0;
+  stop_time_ = std::numeric_limits<int64_t>::max();
+  output_string_.clear();
+  file_->CloseFile();
+  file_finished_.Set();
+}
+
+void RtcEventLogImpl::WriteLog() {
+  bool sleep;
+
+  while (true) {
+    {
+      rtc::CritScope lock(&crit_);
+      switch (state_) {

[the sun, 2016/02/10 10:47:23]

Pull this switch out into a separate function of the suggested LoggerThread
class, and you can add unit tests for the logic!

[terelius, 2016/02/11 21:27:28]

I don't understand what logic you want to test. WriteLog() is just a simple
message loop that dispatches to the functions that do the actual work based on
the state of the log.

[the sun, 2016/02/12 09:37:32]

I was thinking mostly about the stuff in the functions called from here, in
particular the transitive calls to ProcessInOrder().

More importantly though, by pulling the switch out (and using a message queue)
you can easily tailor test cases for different scenarios. 
1. log to memory, terminate thread
2. log to memory, log to file, terminate thread
3. log to memory, log to file, log to memory, terminate thread
...

The key is to make the logic testable in a reproducible and consistent manner
*without* threads or locks interfering.

[terelius, 2016/02/12 14:13:38]

I still don't understand what you want to pull out and how it would help
testing. Do you mean that you want to have
while (1) {
  DoSwitch()
}
or do you mean that we shouldn't have any switch statement because we wouldn't
need any state enum if we had a message queue.

[the sun, 2016/02/12 15:37:27]

Yes, I think you could pretty pretty much write the thread function as you
suggest (add the sleep section as well).

Your test(s) could then be:
1. put event on nn queue
2. put event on oo queue
3. DoSwitch()
4. Verify contents of file+memory
5. put TO_FILE on command queue
6. put event on nn queue
7. DoSwitch()
8. Verify contents of file+memory
...

+        case TO_MEMORY:
+          LogToMemory();
+          break;
+        case STARTING_FILE:
+          StartLogFile();
+          break;
+        case TO_FILE:
+          LogToFile();
+          break;
+        case STOPPING_FILE:
+          StopLogFile();
+          break;
+        case TERMINATE_THREAD:
+          if (file_->Open()) {
+            RTC_DCHECK(stop_time_ < std::numeric_limits<int64_t>::max());
+            StopLogFile();
+            RTC_DCHECK(stop_time_ == std::numeric_limits<int64_t>::max());
+            RTC_DCHECK(!file_->Open());
+            state_ = TERMINATE_THREAD;
+          }
+          return;
+      }
+      int64_t current_time = clock_->TimeInMicroseconds();
+      sleep = (state_ != STOPPING_FILE && state_ != TERMINATE_THREAD &&
+               current_time <= stop_time_);
+    }
+
+    // Accumulate a new batch of events instead of processing them one at a
+    // time. By waiting, we also release the lock so that the main thread can
+    // call StopLogging and StartLogging.
+    if (sleep) {
+      wake_up_.Wait(50);
+    }
+  }
+}
+
+bool RtcEventLogImpl::ThreadOutputFunction(void* obj) {
+  RtcEventLogImpl* event_log = static_cast<RtcEventLogImpl*>(obj);
+  event_log->WriteLog();
+  return false;
+}
+
+void RtcEventLogImpl::TerminateThread() {
+  {
+    rtc::CritScope lock(&crit_);
+    stop_time_ = clock_->TimeInMicroseconds();
+    state_ = TERMINATE_THREAD;
+    wake_up_.Set();  // Wake up the output thread.
+  }
+}
+
 bool RtcEventLog::ParseRtcEventLog(const std::string& file_name,
                                    rtclog::EventStream* result) {
   char tmp_buffer[1024];
   int bytes_read = 0;
   rtc::scoped_ptr<FileWrapper> dump_file(FileWrapper::Create());
   if (dump_file->OpenFile(file_name.c_str(), true) != 0) {
     return false;
   }
   std::string 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.
-rtc::scoped_ptr<RtcEventLog> RtcEventLog::Create() {
-  return rtc::scoped_ptr<RtcEventLog>(new RtcEventLogImpl());
+rtc::scoped_ptr<RtcEventLog> RtcEventLog::Create(const Clock* clock) {
+#ifdef ENABLE_RTC_EVENT_LOG
+  return rtc::scoped_ptr<RtcEventLog>(new RtcEventLogImpl(clock));
+#else
+  return rtc::scoped_ptr<RtcEventLog>(new RtcEventLogNullImpl());
+#endif  // ENABLE_RTC_EVENT_LOG
 }
 
 }  // namespace webrtc