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

[Sergey Ulanov, 2014/12/16 00:21:43]

I do not understand how reordering might happen. Shouldn't browser always send
OnSendComplete() in the same order it received data from the renderer?

And if it doesn't reorder then you don't need to pass packet_id back to the
renderer.

[guoweis_left_chromium, 2014/12/16 23:12:33]

Yes, reordering probably shouldn't happen. I was concerned about on windows,
when doing overlapped IO, I'm not 100% sure whether the call back will be in
sequence or not. However, with more understanding, I realized that can't happen.

I'm not sure I understand your commetn about not passing packet_id, even if
reordering is not possible. Without the packet_id, how do we detect a callback
is skipped?

+  // 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 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 + 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