Hibernate the thread if there are no events in the queue. Wake it up when an event is added to the …

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.
  */
 
@@ skipping 85 matching lines @@
   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:
+  void StoreEvent(std::unique_ptr<rtclog::Event>* event);
+
   // Message queue for passing control messages to the logging thread.
   SwapQueue<RtcEventLogHelperThread::ControlMessage> message_queue_;
 
   // Message queue for passing events to the logging thread.
   SwapQueue<std::unique_ptr<rtclog::Event> > event_queue_;
 
-  rtc::Event wake_up_;
-  rtc::Event stopped_;
-
   const Clock* const clock_;
 
   RtcEventLogHelperThread helper_thread_;
   rtc::ThreadChecker thread_checker_;
 
   RTC_DISALLOW_IMPLICIT_CONSTRUCTORS(RtcEventLogImpl);
 };
 
 namespace {
 // The functions in this namespace convert enums from the runtime format
@@ skipping 33 matching lines @@
 // sent packets because they also contain received packets.
 static const int kEventsPerSecond = 1000;
 static const int kControlMessagesPerSecond = 10;
 }  // namespace
 
 // RtcEventLogImpl member functions.
 RtcEventLogImpl::RtcEventLogImpl(const Clock* clock)
     // Allocate buffers for roughly one second of history.
     : message_queue_(kControlMessagesPerSecond),
       event_queue_(kEventsPerSecond),
-      wake_up_(false, false),
-      stopped_(false, false),
       clock_(clock),
       helper_thread_(&message_queue_,
                      &event_queue_,
-                     &wake_up_,
-                     &stopped_,
                      clock),
       thread_checker_() {
   thread_checker_.DetachFromThread();
 }
 
 RtcEventLogImpl::~RtcEventLogImpl() {
   // The RtcEventLogHelperThread destructor closes the file
   // and waits for the thread to terminate.
 }
 
 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 = clock_->TimeInMicroseconds();
   message.stop_time = std::numeric_limits<int64_t>::max();
   message.file.reset(FileWrapper::Create());
   if (message.file->OpenFile(file_name.c_str(), false) != 0) {
     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.";
   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
@@ skipping 13 matching lines @@
     return false;
   }
   if (message.file->OpenFromFileHandle(file_handle, true, false) != 0) {
     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.";
   return true;
 }
 
 void RtcEventLogImpl::StopLogging() {
   RTC_DCHECK(thread_checker_.CalledOnValidThread());
   RtcEventLogHelperThread::ControlMessage message;
   message.message_type = RtcEventLogHelperThread::ControlMessage::STOP_FILE;
   message.stop_time = clock_->TimeInMicroseconds();
   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();
   }
   LOG(LS_INFO) << "Stopping WebRTC event log.";
-  wake_up_.Set();                       // Request the output thread to wake up.
-  stopped_.Wait(rtc::Event::kForever);  // Wait for the log to stop.
+  helper_thread_.WaitForFileFinished();
 }
 
 void RtcEventLogImpl::LogVideoReceiveStreamConfig(
     const VideoReceiveStream::Config& config) {
   std::unique_ptr<rtclog::Event> event(new rtclog::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();
@@ skipping 15 matching lines @@
         receiver_config->add_header_extensions();
     extension->set_name(e.uri);
     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);
   }
-  if (!event_queue_.Insert(&event)) {
-    LOG(LS_ERROR) << "Config queue full. Not logging config event.";
-  }
+  StoreEvent(&event);

[tommi, 2016/06/08 15:26:40]

This pattern looked strange to me (passing a |std:unique_pt<>r*|), so I guessed
that what you really wanted to do was to move the object (std::move), but then I
saw you're using SwapQueue for the events.  I wonder if that's what you really
want since SwapQueue has a limit and it returns a previous object - but with
StoreEvent, that always gets deleted anyhow and we always new() events that get
inserted.

It also doesn't appear that we handle the error situation when we can't add to
that queue and even in that case, we signal a new event, both of which seem like
a bug.

At a glanse, the implementation seems to be very complex for what I think
(possibly naively) should be fairly straight forward.  The
RtcEventLogHelperThread and the synchronization with it could be to a large
extent be replaced by TaskQueue, using FileWrapper looks like overkill to me
(check out the implementation), we seem to be using 3 separate event objects and
have several complex in-memory container structures.

Also, are the Start/Stop methods necessary?  Could we simply start in the
constructor and stop in the destructor?  That way it would be simple at the call
site to be sure when an object should not be eating up any resources such as
threads, when logging isn't active.

[terelius, 2016/06/08 17:52:17]

You're right. I thought about the argument type too, but decided to pass a
unique_ptr<>* anyway, because that's what the SwapQueue uses.

As we discussed before, there is no good way to reuse the old events without
having one queue per event type. The reason is that different events allocate
memory for protobuf objects of different types. When you want to reuse an event,
there is a fairly high chance that the new event type doesn't match the event
type of the old event you have available.

One way around this would be to serialize the event and store the string
representation in the ringbuffer. This would avoid extra allocations / memory
fragmentation and would be significantly more compact in terms of memory usage.
The downside is that we would need a slightly more complicated ringbuffer, and
we would be doing the serialization on the logging thread.
The queue size is intended to be significantly larger than the expected number
of events between each Process() cycle, so we shouldn't normally fail to insert.
However, if we do, we print an error message and we drop the event. I think this
is the best we can do with a fixed size queue.

If the queue is full, then it is definitely not a bug to signal that there are
new (i.e. unprocessed) events. Perhaps the function name could be chosen better,
but I wanted a name that makes it clear that the function should be called after
every inserted event.
Much of the complexity comes from keeping a history of old events in memory.
This was decided in the one of the first design meetings as a way to let users
start logging when they experience problems but at the same time make the log
contain the cause of the problem. It would be significantly easier if we didn't
have to do that extra buffering.

I agree about the FileWrapper. TaskQueue might be useful, but it did not exist
at the time.
The Start and Stop methods are used to control whether we log to file. In
chromium we get the filename from the user, so we can't easily start logging
automatically. Note that calling Stop in the destructor wouldn't help at all;
the problem isn't that we somehow "forget" to stop, the problem is that the
destructors never get called in the first place.

 }
 
 void RtcEventLogImpl::LogVideoSendStreamConfig(
     const VideoSendStream::Config& config) {
   std::unique_ptr<rtclog::Event> event(new rtclog::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.uri);
     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);
-  if (!event_queue_.Insert(&event)) {
-    LOG(LS_ERROR) << "Config queue full. Not logging config event.";
-  }
+  StoreEvent(&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;
   }
 
   std::unique_ptr<rtclog::Event> rtp_event(new rtclog::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);
-  if (!event_queue_.Insert(&rtp_event)) {
-    LOG(LS_ERROR) << "RTP queue full. Not logging RTP packet.";
-  }
+  StoreEvent(&rtp_event);
 }
 
 void RtcEventLogImpl::LogRtcpPacket(PacketDirection direction,
                                     MediaType media_type,
                                     const uint8_t* packet,
                                     size_t length) {
   std::unique_ptr<rtclog::Event> rtcp_event(new rtclog::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);
@@ skipping 37 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);
-  if (!event_queue_.Insert(&rtcp_event)) {
-    LOG(LS_ERROR) << "RTCP queue full. Not logging RTCP packet.";
-  }
+  StoreEvent(&rtcp_event);
 }
 
 void RtcEventLogImpl::LogAudioPlayout(uint32_t ssrc) {
   std::unique_ptr<rtclog::Event> event(new rtclog::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);
-  if (!event_queue_.Insert(&event)) {
-    LOG(LS_ERROR) << "Playout queue full. Not logging ACM playout.";
-  }
+  StoreEvent(&event);
 }
 
 void RtcEventLogImpl::LogBwePacketLossEvent(int32_t bitrate,
                                             uint8_t fraction_loss,
                                             int32_t total_packets) {
   std::unique_ptr<rtclog::Event> event(new rtclog::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);
-  if (!event_queue_.Insert(&event)) {
-    LOG(LS_ERROR) << "BWE loss queue full. Not logging BWE update.";
+  StoreEvent(&event);
+}
+
+void RtcEventLogImpl::StoreEvent(std::unique_ptr<rtclog::Event>* event) {
+  if (!event_queue_.Insert(event)) {
+    LOG(LS_ERROR) << "WebRTC event log queue full. Dropping event.";
   }
+  helper_thread_.SignalNewEvent();
 }
 
 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) != 0) {
     return false;
   }
@@ skipping 10 matching lines @@
 // RtcEventLog member functions.
 std::unique_ptr<RtcEventLog> RtcEventLog::Create(const Clock* clock) {
 #ifdef ENABLE_RTC_EVENT_LOG
   return std::unique_ptr<RtcEventLog>(new RtcEventLogImpl(clock));
 #else
   return std::unique_ptr<RtcEventLog>(new RtcEventLogNullImpl());
 #endif  // ENABLE_RTC_EVENT_LOG
 }
 
 }  // namespace webrtc