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_t packet_id, size_t packet_size)
+        : packet_id(packet_id), packet_size(packet_size) {}
+
+    uint64_t packet_id;
+    size_t packet_size;
+  };
+
+  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;
+

[Sergey Ulanov, 2014/12/23 01:11:02]

nit: Don't need this empty line.

[guoweis_left_chromium, 2014/12/23 18:53:16]

Done.

   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_;
@@ skipping 18 matching lines @@
   // 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_;
+
+  // Used to detect when browser doesn't send SendComplete message for some
+  // packets. In normal case, the first packet should be the one that we're
+  // going to receive the next completion signal.
+  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 119 matching lines @@
   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_t 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);
+
+  in_flight_packet_records_.push_back(
+      InFlightPacketRecord(packet_id, data_size));
+  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() {
+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 = in_flight_packet_records_.begin();

[Sergey Ulanov, 2014/12/23 01:11:02]

nit: this can be:
const InFlightPacketRecord& packet_record = in_flight_packet_records_.front();

[guoweis_left_chromium, 2014/12/23 18:53:16]

Done.

+
+  // Tracking is not turned on for TCP so it's always 0. For UDP, this will
+  // cause a crash when the packet ids don't match.
+  CHECK(send_metrics.packet_id == 0 || it->packet_id == send_metrics.packet_id);
+
+  send_bytes_available_ += it->packet_size;
 
   DCHECK_LE(send_bytes_available_, kMaximumInFlightBytes);
 
-  in_flight_packet_sizes_.pop_front();
+  in_flight_packet_records_.erase(it);

[Sergey Ulanov, 2014/12/23 01:11:02]

pop_front()?

[guoweis_left_chromium, 2014/12/23 18:53:16]

Done.

   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