Detect situation when there is no missing send completion signal in P2PSocket implementation.

content/renderer/p2p/ipc_socket_factory.cc

 // Copyright (c) 2012 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 "content/renderer/p2p/ipc_socket_factory.h"
 
 #include <algorithm>
-#include <deque>
+#include <list>
 
 #include "base/compiler_specific.h"
 #include "base/debug/trace_event.h"
 #include "base/message_loop/message_loop.h"
 #include "base/message_loop/message_loop_proxy.h"
 #include "base/metrics/field_trial.h"
 #include "base/metrics/histogram.h"
 #include "base/strings/string_number_conversions.h"
 #include "base/strings/stringprintf.h"
 #include "base/threading/non_thread_safe.h"
@@ skipping 50 matching lines @@
 
 // IpcPacketSocket implements rtc::AsyncPacketSocket interface
 // using P2PSocketClient that works over IPC-channel. It must be used
 // on the thread it was created.
 class IpcPacketSocket : public rtc::AsyncPacketSocket,
                         public P2PSocketClientDelegate {
  public:
   IpcPacketSocket();
   ~IpcPacketSocket() override;
 
+  // Struct to track information when a packet is received by this socket for
+  // send. The information tracked here will be used to match with the
+  // P2PSendPacketMetrics from the underneath system socket.
+  struct InFlightPacketRecord {
+    InFlightPacketRecord(uint64 packet_id,
+                         size_t packet_size,
+                         uint64 timeticks_received_by_socket)
+        : packet_id(packet_id),
+          packet_size(packet_size),
+          timeticks_received_by_socket(timeticks_received_by_socket) {}
+
+    uint64 packet_id;
+    size_t packet_size;
+    // The first time this packet shows up to this socket.
+    uint64 timeticks_received_by_socket;

[Sergey Ulanov, 2014/12/17 20:05:30]

use base::TimeTicks type for this. Is this used anywhere?

[guoweis_left_chromium, 2014/12/22 21:41:47]

Removed. This should been another change which tracks the time, not this one.

+  };
+
+  typedef std::list<InFlightPacketRecord> InFlightPacketList;
+
   // Always takes ownership of client even if initialization fails.
   bool Init(P2PSocketType type, P2PSocketClientImpl* client,
             const rtc::SocketAddress& local_address,
             const rtc::SocketAddress& remote_address);
 
   // rtc::AsyncPacketSocket interface.
   rtc::SocketAddress GetLocalAddress() const override;
   rtc::SocketAddress GetRemoteAddress() const override;
   int Send(const void* pv,
            size_t cb,
            const rtc::PacketOptions& options) override;
   int SendTo(const void* pv,
              size_t cb,
              const rtc::SocketAddress& addr,
              const rtc::PacketOptions& options) override;
   int Close() override;
   State GetState() const override;
   int GetOption(rtc::Socket::Option option, int* value) override;
   int SetOption(rtc::Socket::Option option, int value) override;
   int GetError() const override;
   void SetError(int error) override;
 
   // P2PSocketClientDelegate implementation.
   void OnOpen(const net::IPEndPoint& local_address,
               const net::IPEndPoint& remote_address) override;
   void OnIncomingTcpConnection(const net::IPEndPoint& address,
                                P2PSocketClient* client) override;
-  void OnSendComplete() override;
+  void OnSendComplete(const P2PSendPacketMetrics& send_metrics) override;
+
   void OnError() override;
   void OnDataReceived(const net::IPEndPoint& address,
                       const std::vector<char>& data,
                       const base::TimeTicks& timestamp) override;
 
  private:
   enum InternalState {
     IS_UNINITIALIZED,
     IS_OPENING,
     IS_OPEN,
     IS_CLOSED,
     IS_ERROR,
   };
 
   // Increment the counter for consecutive bytes discarded as socket is running
   // out of buffer.
   void IncrementDiscardCounters(size_t bytes_discarded);
 
   // Update trace of send throttling internal state. This should be called
   // immediately after any changes to |send_bytes_available_| and/or
-  // |in_flight_packet_sizes_|.
+  // |in_flight_packet_records_|.
   void TraceSendThrottlingState() const;
 
   void InitAcceptedTcp(P2PSocketClient* client,
                        const rtc::SocketAddress& local_address,
                        const rtc::SocketAddress& remote_address);
 
   int DoSetOption(P2PSocketOption option, int value);
 
   // Allow a finch experiment to control the initial value of
   // send_bytes_available_;
   void AdjustUdpSendBufferSize();
 
+  // Helper function to find the matching InFlightPacketRecord by packet_id.
+  // In normal case, it should always be the first one so the perf impact should
+  // be negligible. Return value is the iterator pointing to the matching
+  // InFlightPacketRecord. Can be InFlightPacketRecord::end() if no matching one
+  // is found.
+  InFlightPacketList::iterator FindInFlightRecord(uint64 packet_id);
+
   P2PSocketType type_;
 
   // Message loop on which this socket was created and being used.
   base::MessageLoop* message_loop_;
 
   // Corresponding P2P socket client.
   scoped_refptr<P2PSocketClient> client_;
 
   // Local address is allocated by the browser process, and the
   // renderer side doesn't know the address until it receives OnOpen()
   // event from the browser.
   rtc::SocketAddress local_address_;
 
   // Remote address for client TCP connections.
   rtc::SocketAddress remote_address_;
 
   // Current state of the object.
   InternalState state_;
 
   // Track the number of bytes allowed to be sent non-blocking. This is used to
   // throttle the sending of packets to the browser process. For each packet
   // sent, the value is decreased. As callbacks to OnSendComplete() (as IPCs
   // from the browser process) are made, the value is increased back. This
   // allows short bursts of high-rate sending without dropping packets, but
   // quickly restricts the client to a sustainable steady-state rate.
   size_t send_bytes_available_;
-  std::deque<size_t> in_flight_packet_sizes_;
+
+  // This is a sorted linked list of all InFlightPacketRecord by its packet_id.

[Sergey Ulanov, 2014/12/17 20:05:30]

You are not actually sorting them by packet_id. They are in the order in which
they were sent. It's a coincidence that it happens to be sorted by packet_id
because of the way P2PSocketClientImpl() allocates packet IDs.

[guoweis_left_chromium, 2014/12/22 21:41:47]

Removed that comment.

+  // In normal case, the first packet should always be the one that we're going
+  // to receive the next completion signal. This list gives us a way to look for
+  // packets that we never receive its completion signal from OS.

[Sergey Ulanov, 2014/12/17 20:05:30]

If I understand correctly the purpose of this CL is to be able to detect bugs in
IPC layer, i.e. when the browser process doesn't send SendComplete notification
back to the renderer process. And if that's right, then the last sentence of
this comment doesn't make much sense. 
Suggest rewording "Used to detect when browser doesn't send SendComplete message
for some packets."

[guoweis_left_chromium, 2014/12/22 21:41:47]

Done.

+  InFlightPacketList in_flight_packet_records_;
 
   // Set to true once EWOULDBLOCK was returned from Send(). Indicates that the
   // caller expects SignalWritable notification.
   bool writable_signal_expected_;
 
   // Current error code. Valid when state_ == IS_ERROR.
   int error_;
   int options_[P2P_SOCKET_OPT_MAX];
 
   // Track the maximum and current consecutive bytes discarded due to not enough
@@ skipping 61 matching lines @@
   if (total_packets_ > 0) {
     UMA_HISTOGRAM_PERCENTAGE("WebRTC.ApplicationPercentPacketsDiscarded",
                              (packets_discarded_ * 100) / total_packets_);
   }
 }
 
 void IpcPacketSocket::TraceSendThrottlingState() const {
   TRACE_COUNTER_ID1("p2p", "P2PSendBytesAvailable", local_address_.port(),
                     send_bytes_available_);
   TRACE_COUNTER_ID1("p2p", "P2PSendPacketsInFlight", local_address_.port(),
-                    in_flight_packet_sizes_.size());
+                    in_flight_packet_records_.size());
 }
 
 void IpcPacketSocket::IncrementDiscardCounters(size_t bytes_discarded) {
   current_discard_bytes_sequence_ += bytes_discarded;
   packets_discarded_++;
 
   if (current_discard_bytes_sequence_ > max_discard_bytes_sequence_) {
     max_discard_bytes_sequence_ = current_discard_bytes_sequence_;
   }
 }
@@ skipping 104 matching lines @@
       return EWOULDBLOCK;
     case IS_CLOSED:
       return ENOTCONN;
     case IS_ERROR:
       return error_;
     case IS_OPEN:
       // Continue sending the packet.
       break;
   }
 
+  uint64 tick_received_in_socket = base::TimeTicks::Now().ToInternalValue();
+
   if (data_size == 0) {
     NOTREACHED();
     return 0;
   }
 
   total_packets_++;
 
   if (data_size > send_bytes_available_) {
     TRACE_EVENT_INSTANT1("p2p", "MaxPendingBytesWouldBlock",
                          TRACE_EVENT_SCOPE_THREAD,
                          "id",
                          client_->GetSocketID());
     if (!writable_signal_expected_) {
       WebRtcLogMessage(base::StringPrintf(
           "IpcPacketSocket: sending is blocked. %d packets_in_flight.",
-          static_cast<int>(in_flight_packet_sizes_.size())));
+          static_cast<int>(in_flight_packet_records_.size())));
 
       writable_signal_expected_ = true;
     }
 
     error_ = EWOULDBLOCK;
     IncrementDiscardCounters(data_size);
     return -1;
   } else {
     current_discard_bytes_sequence_ = 0;
   }
 
   net::IPEndPoint address_chrome;
   if (!jingle_glue::SocketAddressToIPEndPoint(address, &address_chrome)) {
     DVLOG(1) << "Failed to convert remote address to IPEndPoint: address = "
              << address.ToSensitiveString() << ", remote_address_ = "
              << remote_address_.ToSensitiveString();
     NOTREACHED();
     error_ = EINVAL;
     return -1;
   }
 
   send_bytes_available_ -= data_size;
-  in_flight_packet_sizes_.push_back(data_size);
-  TraceSendThrottlingState();
 
   const char* data_char = reinterpret_cast<const char*>(data);
   std::vector<char> data_vector(data_char, data_char + data_size);
-  client_->SendWithDscp(address_chrome, data_vector, options);
+  uint64 packet_id =
+      client_->SendWithDscp(address_chrome, data_vector, options);
+
+  // Ensure packet_id is not 0. It can't be the case according to
+  // P2PSocketClientImpl::SendWithDscp().
+  DCHECK_NE(packet_id, 0uL);
+
+  // Since OnSendComplete happens on the same thread, there is no chance that

[Sergey Ulanov, 2014/12/17 20:05:30]

nit: I don't think you really need this comment. It should be obvious.

[guoweis_left_chromium, 2014/12/22 21:41:47]

Done.

+  // for the same packet, OnSendComplete is called before SendWithDscp is
+  // finished with |packet_id| back.
+  in_flight_packet_records_.push_back(
+      InFlightPacketRecord(packet_id, data_size, tick_received_in_socket));
+  TraceSendThrottlingState();
 
   // Fake successful send. The caller ignores result anyway.
   return data_size;
 }
 
 int IpcPacketSocket::Close() {
   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 
   client_->Close();
   state_ = IS_CLOSED;
@@ skipping 125 matching lines @@
 
   rtc::SocketAddress remote_address;
   if (!jingle_glue::IPEndPointToSocketAddress(address, &remote_address)) {
     // Always expect correct IPv4 address to be allocated.
     NOTREACHED();
   }
   socket->InitAcceptedTcp(client, local_address_, remote_address);
   SignalNewConnection(this, socket.release());
 }
 
-void IpcPacketSocket::OnSendComplete() {
+// For most of the case, the completion signal should be returned for the packet
+// at the beginning of |in_flight_packet_records_| so the perf impact should be
+// negligible.
+IpcPacketSocket::InFlightPacketList::iterator
+IpcPacketSocket::FindInFlightRecord(uint64 packet_id) {
+  CHECK(!in_flight_packet_records_.empty());
+  InFlightPacketList::iterator it = in_flight_packet_records_.begin();
+  for (; it != in_flight_packet_records_.end(); ++it) {
+    if (it->packet_id == packet_id) {
+      break;
+    }
+  }
+
+  // We should always be able to find matching packet_id in the
+  // in_flight_packet_records_.
+  DCHECK(it != in_flight_packet_records_.end());
+  return it;
+}
+
+void IpcPacketSocket::OnSendComplete(const P2PSendPacketMetrics& send_metrics) {
   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 
-  CHECK(!in_flight_packet_sizes_.empty());
-  send_bytes_available_ += in_flight_packet_sizes_.front();
+  CHECK(!in_flight_packet_records_.empty());
+
+  InFlightPacketList::iterator it;
+  if (send_metrics.packet_id == 0) {

[Sergey Ulanov, 2014/12/17 20:05:30]

When does this happen? Why do you need this?

[guoweis_left_chromium, 2014/12/22 21:41:47]

TCP. We're not tracking tcp's packet id so this is needed. Added a comment for
that.

+    // If send_metrics doesn't carry valid packet_id, it means that we should
+    // not try to detect mismatch packets.
+    it = in_flight_packet_records_.begin();
+  } else {
+    it = FindInFlightRecord(send_metrics.packet_id);

[Sergey Ulanov, 2014/12/17 20:05:30]

I don't think you need this because we never expect out-of-order delivery of IPC
messages. Just get the first packet and DCHECK() that packet_id is the same.

[guoweis_left_chromium, 2014/12/22 21:41:47]

Done.

+
+    // If we can't find the record, something has gone very wrong at this point.
+    if (it == in_flight_packet_records_.end()) {
+      VLOG(1) << "Failed to find in-flight record with packet_id = "
+              << send_metrics.packet_id;
+      NOTREACHED();
+      return;
+    }
+  }
+
+  send_bytes_available_ += it->packet_size;
+
+  // Here is to detect a completion signal is not returned. We can't rely on the
+  // destructor since if a user closes the tab, the destructor is not
+  // invoked. Or if a user inputs another url, the destructor will be called
+  // before all completion signals return.
+  if (send_metrics.packet_id != 0 &&
+      in_flight_packet_records_.front().packet_id != send_metrics.packet_id) {
+    // Report an instance that a packet didn't have its completion
+    // signal returned.
+    UMA_HISTOGRAM_BOOLEAN("WebRTC.ApplicationSocketMismatchPacketsDetected_UDP",

[Sergey Ulanov, 2014/12/17 20:05:30]

I don't think you need UMA metric for this. Just CHECK() here and see if we get
crash reports.

[guoweis_left_chromium, 2014/12/22 21:41:47]

Done.

+                          true);
+  }
 
   DCHECK_LE(send_bytes_available_, kMaximumInFlightBytes);
 
-  in_flight_packet_sizes_.pop_front();
+  in_flight_packet_records_.erase(it);
   TraceSendThrottlingState();
 
   if (writable_signal_expected_ && send_bytes_available_ > 0) {
     WebRtcLogMessage(base::StringPrintf(
         "IpcPacketSocket: sending is unblocked. %d packets in flight.",
-        static_cast<int>(in_flight_packet_sizes_.size())));
+        static_cast<int>(in_flight_packet_records_.size())));
 
     SignalReadyToSend(this);
     writable_signal_expected_ = false;
   }
 }
 
 void IpcPacketSocket::OnError() {
   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
   bool was_closed = (state_ == IS_ERROR || state_ == IS_CLOSED);
   state_ = IS_ERROR;
@@ skipping 153 matching lines @@
 }
 
 rtc::AsyncResolverInterface*
 IpcPacketSocketFactory::CreateAsyncResolver() {
   scoped_ptr<AsyncAddressResolverImpl> resolver(
     new AsyncAddressResolverImpl(socket_dispatcher_));
   return resolver.release();
 }
 
 }  // namespace content