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;
+  };
+
+  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.
+  // In normal case without reordering, the first packet should always be the
+  // one that we're going to receive the next completion signal. However, since
+  // we can't rule out reordering happening (when system buffer is large enough
+  // on various platforms), if it happens, there could be some minor penalty
+  // looking for the right packet in the list. This list gives us a way to look
+  // for packets that we never receive its completion signal from OS.
+  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
+  // 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) {
+    // 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);
+
+    // 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 a heuristic to detect a completion signal is not returned. We can't
+  // rely on the destructor since if a user close the tab, destructor is not

[rkaplow, 2014/12/09 20:19:37]

well now that you've switched to 'a user', it should be
if a user closes.
Also, you're missing "the" destructor is not invoked
 

[guoweis_left_chromium, 2014/12/09 23:54:45]

Ah, sorry about this. Done.

+  // 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 + 100 <
+          send_metrics.packet_id) {
+    // Report an instance that a packet didn't have its completion
+    // signal returned.
+    UMA_HISTOGRAM_BOOLEAN("WebRTC.ApplicationSocketMismatchPacketsDetected_UDP",
+                          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