content/renderer/p2p/ipc_socket_factory.cc

   1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "content/renderer/p2p/ipc_socket_factory.h"
   6
   7 #include <algorithm>
   8 #include <list>
   9
  10 #include "base/compiler_specific.h"
  11 #include "base/message_loop/message_loop.h"
  12 #include "base/message_loop/message_loop_proxy.h"
  13 #include "base/metrics/field_trial.h"
  14 #include "base/metrics/histogram.h"
  15 #include "base/strings/string_number_conversions.h"
  16 #include "base/strings/stringprintf.h"
  17 #include "base/threading/non_thread_safe.h"
  18 #include "base/trace_event/trace_event.h"
  19 #include "content/renderer/media/webrtc_logging.h"
  20 #include "content/renderer/p2p/host_address_request.h"
  21 #include "content/renderer/p2p/socket_client_delegate.h"
  22 #include "content/renderer/p2p/socket_client_impl.h"
  23 #include "content/renderer/p2p/socket_dispatcher.h"
  24 #include "jingle/glue/utils.h"
  25 #include "third_party/webrtc/base/asyncpacketsocket.h"
  26
  27 namespace content {
  28
  29 namespace {
  30
  31 const int kDefaultNonSetOptionValue = -1;
  32
  33 bool IsTcpClientSocket(P2PSocketType type) {
  34   return (type == P2P_SOCKET_STUN_TCP_CLIENT) ||
  35          (type == P2P_SOCKET_TCP_CLIENT) ||
  36          (type == P2P_SOCKET_STUN_SSLTCP_CLIENT) ||
  37          (type == P2P_SOCKET_SSLTCP_CLIENT) ||
  38          (type == P2P_SOCKET_TLS_CLIENT) ||
  39          (type == P2P_SOCKET_STUN_TLS_CLIENT);
  40 }
  41
  42 bool JingleSocketOptionToP2PSocketOption(rtc::Socket::Option option,
  43                                          P2PSocketOption* ipc_option) {
  44   switch (option) {
  45     case rtc::Socket::OPT_RCVBUF:
  46       *ipc_option = P2P_SOCKET_OPT_RCVBUF;
  47       break;
  48     case rtc::Socket::OPT_SNDBUF:
  49       *ipc_option = P2P_SOCKET_OPT_SNDBUF;
  50       break;
  51     case rtc::Socket::OPT_DSCP:
  52       *ipc_option = P2P_SOCKET_OPT_DSCP;
  53       break;
  54     case rtc::Socket::OPT_DONTFRAGMENT:
  55     case rtc::Socket::OPT_NODELAY:
  56     case rtc::Socket::OPT_IPV6_V6ONLY:
  57     case rtc::Socket::OPT_RTP_SENDTIME_EXTN_ID:
  58       return false;  // Not supported by the chrome sockets.
  59     default:
  60       NOTREACHED();
  61       return false;
  62   }
  63   return true;
  64 }
  65
  66 // TODO(miu): This needs tuning.  http://crbug.com/237960
  67 // http://crbug.com/427555
  68 const size_t kMaximumInFlightBytes = 64 * 1024;  // 64 KB
  69
  70 // IpcPacketSocket implements rtc::AsyncPacketSocket interface
  71 // using P2PSocketClient that works over IPC-channel. It must be used
  72 // on the thread it was created.
  73 class IpcPacketSocket : public rtc::AsyncPacketSocket,
  74                         public P2PSocketClientDelegate {
  75  public:
  76   IpcPacketSocket();
  77   ~IpcPacketSocket() override;
  78
  79   // Struct to track information when a packet is received by this socket for
  80   // send. The information tracked here will be used to match with the
  81   // P2PSendPacketMetrics from the underneath system socket.
  82   struct InFlightPacketRecord {
  83     InFlightPacketRecord(uint64_t packet_id, size_t packet_size)
  84         : packet_id(packet_id), packet_size(packet_size) {}
  85
  86     uint64_t packet_id;
  87     size_t packet_size;
  88   };
  89
  90   typedef std::list<InFlightPacketRecord> InFlightPacketList;
  91
  92   // Always takes ownership of client even if initialization fails.
  93   bool Init(P2PSocketType type, P2PSocketClientImpl* client,
  94             const rtc::SocketAddress& local_address,
  95             const rtc::SocketAddress& remote_address);
  96
  97   // rtc::AsyncPacketSocket interface.
  98   rtc::SocketAddress GetLocalAddress() const override;
  99   rtc::SocketAddress GetRemoteAddress() const override;
 100   int Send(const void* pv,
 101            size_t cb,
 102            const rtc::PacketOptions& options) override;
 103   int SendTo(const void* pv,
 104              size_t cb,
 105              const rtc::SocketAddress& addr,
 106              const rtc::PacketOptions& options) override;
 107   int Close() override;
 108   State GetState() const override;
 109   int GetOption(rtc::Socket::Option option, int* value) override;
 110   int SetOption(rtc::Socket::Option option, int value) override;
 111   int GetError() const override;
 112   void SetError(int error) override;
 113
 114   // P2PSocketClientDelegate implementation.
 115   void OnOpen(const net::IPEndPoint& local_address,
 116               const net::IPEndPoint& remote_address) override;
 117   void OnIncomingTcpConnection(const net::IPEndPoint& address,
 118                                P2PSocketClient* client) override;
 119   void OnSendComplete(const P2PSendPacketMetrics& send_metrics) override;
 120   void OnError() override;
 121   void OnDataReceived(const net::IPEndPoint& address,
 122                       const std::vector<char>& data,
 123                       const base::TimeTicks& timestamp) override;
 124
 125  private:
 126   enum InternalState {
 127     IS_UNINITIALIZED,
 128     IS_OPENING,
 129     IS_OPEN,
 130     IS_CLOSED,
 131     IS_ERROR,
 132   };
 133
 134   // Increment the counter for consecutive bytes discarded as socket is running
 135   // out of buffer.
 136   void IncrementDiscardCounters(size_t bytes_discarded);
 137
 138   // Update trace of send throttling internal state. This should be called
 139   // immediately after any changes to |send_bytes_available_| and/or
 140   // |in_flight_packet_records_|.
 141   void TraceSendThrottlingState() const;
 142
 143   void InitAcceptedTcp(P2PSocketClient* client,
 144                        const rtc::SocketAddress& local_address,
 145                        const rtc::SocketAddress& remote_address);
 146
 147   int DoSetOption(P2PSocketOption option, int value);
 148
 149   // Allow a finch experiment to control the initial value of
 150   // send_bytes_available_;
 151   void AdjustUdpSendBufferSize();
 152
 153   P2PSocketType type_;
 154
 155   // Message loop on which this socket was created and being used.
 156   base::MessageLoop* message_loop_;
 157
 158   // Corresponding P2P socket client.
 159   scoped_refptr<P2PSocketClient> client_;
 160
 161   // Local address is allocated by the browser process, and the
 162   // renderer side doesn't know the address until it receives OnOpen()
 163   // event from the browser.
 164   rtc::SocketAddress local_address_;
 165
 166   // Remote address for client TCP connections.
 167   rtc::SocketAddress remote_address_;
 168
 169   // Current state of the object.
 170   InternalState state_;
 171
 172   // Track the number of bytes allowed to be sent non-blocking. This is used to
 173   // throttle the sending of packets to the browser process. For each packet
 174   // sent, the value is decreased. As callbacks to OnSendComplete() (as IPCs
 175   // from the browser process) are made, the value is increased back. This
 176   // allows short bursts of high-rate sending without dropping packets, but
 177   // quickly restricts the client to a sustainable steady-state rate.
 178   size_t send_bytes_available_;
 179
 180   // Used to detect when browser doesn't send SendComplete message for some
 181   // packets. In normal case, the first packet should be the one that we're
 182   // going to receive the next completion signal.
 183   InFlightPacketList in_flight_packet_records_;
 184
 185   // Set to true once EWOULDBLOCK was returned from Send(). Indicates that the
 186   // caller expects SignalWritable notification.
 187   bool writable_signal_expected_;
 188
 189   // Current error code. Valid when state_ == IS_ERROR.
 190   int error_;
 191   int options_[P2P_SOCKET_OPT_MAX];
 192
 193   // Track the maximum and current consecutive bytes discarded due to not enough
 194   // send_bytes_available_.
 195   size_t max_discard_bytes_sequence_;
 196   size_t current_discard_bytes_sequence_;
 197
 198   // Track the total number of packets and the number of packets discarded.
 199   size_t packets_discarded_;
 200   size_t total_packets_;
 201
 202   DISALLOW_COPY_AND_ASSIGN(IpcPacketSocket);
 203 };
 204
 205 // Simple wrapper around P2PAsyncAddressResolver. The main purpose of this
 206 // class is to send SignalDone, after OnDone callback from
 207 // P2PAsyncAddressResolver. Libjingle sig slots are not thread safe. In case
 208 // of MT sig slots clients must call disconnect. This class is to make sure
 209 // we destruct from the same thread on which is created.
 210 class AsyncAddressResolverImpl :  public base::NonThreadSafe,
 211                                   public rtc::AsyncResolverInterface {
 212  public:
 213   AsyncAddressResolverImpl(P2PSocketDispatcher* dispatcher);
 214   ~AsyncAddressResolverImpl() override;
 215
 216   // rtc::AsyncResolverInterface interface.
 217   void Start(const rtc::SocketAddress& addr) override;
 218   bool GetResolvedAddress(int family, rtc::SocketAddress* addr) const override;
 219   int GetError() const override;
 220   void Destroy(bool wait) override;
 221
 222  private:
 223   virtual void OnAddressResolved(const net::IPAddressList& addresses);
 224
 225   scoped_refptr<P2PAsyncAddressResolver> resolver_;
 226   int port_;   // Port number in |addr| from Start() method.
 227   std::vector<rtc::IPAddress> addresses_;  // Resolved addresses.
 228 };
 229
 230 IpcPacketSocket::IpcPacketSocket()
 231     : type_(P2P_SOCKET_UDP),
 232       message_loop_(base::MessageLoop::current()),
 233       state_(IS_UNINITIALIZED),
 234       send_bytes_available_(kMaximumInFlightBytes),
 235       writable_signal_expected_(false),
 236       error_(0),
 237       max_discard_bytes_sequence_(0),
 238       current_discard_bytes_sequence_(0),
 239       packets_discarded_(0),
 240       total_packets_(0) {
 241   static_assert(kMaximumInFlightBytes > 0, "would send at zero rate");
 242   std::fill_n(options_, static_cast<int> (P2P_SOCKET_OPT_MAX),
 243               kDefaultNonSetOptionValue);
 244 }
 245
 246 IpcPacketSocket::~IpcPacketSocket() {
 247   if (state_ == IS_OPENING || state_ == IS_OPEN ||
 248       state_ == IS_ERROR) {
 249     Close();
 250   }
 251
 252   UMA_HISTOGRAM_CUSTOM_COUNTS("WebRTC.ApplicationMaxConsecutiveBytesDiscard.v2",
 253                               max_discard_bytes_sequence_, 1, 1000000, 200);
 254   if (total_packets_ > 0) {
 255     UMA_HISTOGRAM_PERCENTAGE("WebRTC.ApplicationPercentPacketsDiscarded",
 256                              (packets_discarded_ * 100) / total_packets_);
 257   }
 258 }
 259
 260 void IpcPacketSocket::TraceSendThrottlingState() const {
 261   TRACE_COUNTER_ID1("p2p", "P2PSendBytesAvailable", local_address_.port(),
 262                     send_bytes_available_);
 263   TRACE_COUNTER_ID1("p2p", "P2PSendPacketsInFlight", local_address_.port(),
 264                     in_flight_packet_records_.size());
 265 }
 266
 267 void IpcPacketSocket::IncrementDiscardCounters(size_t bytes_discarded) {
 268   current_discard_bytes_sequence_ += bytes_discarded;
 269   packets_discarded_++;
 270
 271   if (current_discard_bytes_sequence_ > max_discard_bytes_sequence_) {
 272     max_discard_bytes_sequence_ = current_discard_bytes_sequence_;
 273   }
 274 }
 275
 276 void IpcPacketSocket::AdjustUdpSendBufferSize() {
 277   DCHECK_EQ(type_, P2P_SOCKET_UDP);
 278   unsigned int send_buffer_size = 0;
 279
 280   base::StringToUint(
 281       base::FieldTrialList::FindFullName("WebRTC-ApplicationUDPSendSocketSize"),
 282       &send_buffer_size);
 283
 284   if (send_buffer_size > 0) {
 285     send_bytes_available_ = send_buffer_size;
 286   }
 287 }
 288
 289 bool IpcPacketSocket::Init(P2PSocketType type,
 290                            P2PSocketClientImpl* client,
 291                            const rtc::SocketAddress& local_address,
 292                            const rtc::SocketAddress& remote_address) {
 293   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 294   DCHECK_EQ(state_, IS_UNINITIALIZED);
 295
 296   type_ = type;
 297   client_ = client;
 298   local_address_ = local_address;
 299   remote_address_ = remote_address;
 300   state_ = IS_OPENING;
 301
 302   net::IPEndPoint local_endpoint;
 303   if (!jingle_glue::SocketAddressToIPEndPoint(
 304           local_address, &local_endpoint)) {
 305     return false;
 306   }
 307
 308   if (type_ == P2P_SOCKET_UDP) {
 309     AdjustUdpSendBufferSize();
 310   }
 311
 312   net::IPEndPoint remote_endpoint;
 313   if (!remote_address.IsNil()) {
 314     DCHECK(IsTcpClientSocket(type_));
 315
 316     if (remote_address.IsUnresolvedIP()) {
 317       remote_endpoint =
 318           net::IPEndPoint(net::IPAddressNumber(), remote_address.port());
 319     } else {
 320       if (!jingle_glue::SocketAddressToIPEndPoint(remote_address,
 321                                                   &remote_endpoint)) {
 322         return false;
 323       }
 324     }
 325   }
 326
 327   // We need to send both resolved and unresolved address in Init. Unresolved
 328   // address will be used in case of TLS for certificate hostname matching.
 329   // Certificate will be tied to domain name not to IP address.
 330   P2PHostAndIPEndPoint remote_info(remote_address.hostname(), remote_endpoint);
 331
 332   client->Init(type, local_endpoint, remote_info, this);
 333
 334   return true;
 335 }
 336
 337 void IpcPacketSocket::InitAcceptedTcp(
 338     P2PSocketClient* client,
 339     const rtc::SocketAddress& local_address,
 340     const rtc::SocketAddress& remote_address) {
 341   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 342   DCHECK_EQ(state_, IS_UNINITIALIZED);
 343
 344   client_ = client;
 345   local_address_ = local_address;
 346   remote_address_ = remote_address;
 347   state_ = IS_OPEN;
 348   TraceSendThrottlingState();
 349   client_->SetDelegate(this);
 350 }
 351
 352 // rtc::AsyncPacketSocket interface.
 353 rtc::SocketAddress IpcPacketSocket::GetLocalAddress() const {
 354   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 355   return local_address_;
 356 }
 357
 358 rtc::SocketAddress IpcPacketSocket::GetRemoteAddress() const {
 359   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 360   return remote_address_;
 361 }
 362
 363 int IpcPacketSocket::Send(const void *data, size_t data_size,
 364                           const rtc::PacketOptions& options) {
 365   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 366   return SendTo(data, data_size, remote_address_, options);
 367 }
 368
 369 int IpcPacketSocket::SendTo(const void *data, size_t data_size,
 370                             const rtc::SocketAddress& address,
 371                             const rtc::PacketOptions& options) {
 372   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 373
 374   switch (state_) {
 375     case IS_UNINITIALIZED:
 376       NOTREACHED();
 377       return EWOULDBLOCK;
 378     case IS_OPENING:
 379       return EWOULDBLOCK;
 380     case IS_CLOSED:
 381       return ENOTCONN;
 382     case IS_ERROR:
 383       return error_;
 384     case IS_OPEN:
 385       // Continue sending the packet.
 386       break;
 387   }
 388
 389   if (data_size == 0) {
 390     NOTREACHED();
 391     return 0;
 392   }
 393
 394   total_packets_++;
 395
 396   if (data_size > send_bytes_available_) {
 397     TRACE_EVENT_INSTANT1("p2p", "MaxPendingBytesWouldBlock",
 398                          TRACE_EVENT_SCOPE_THREAD,
 399                          "id",
 400                          client_->GetSocketID());
 401     if (!writable_signal_expected_) {
 402       WebRtcLogMessage(base::StringPrintf(
 403           "IpcPacketSocket: sending is blocked. %d packets_in_flight.",
 404           static_cast<int>(in_flight_packet_records_.size())));
 405
 406       writable_signal_expected_ = true;
 407     }
 408
 409     error_ = EWOULDBLOCK;
 410     IncrementDiscardCounters(data_size);
 411     return -1;
 412   } else {
 413     current_discard_bytes_sequence_ = 0;
 414   }
 415
 416   net::IPEndPoint address_chrome;
 417   if (address.IsUnresolvedIP()) {
 418     address_chrome = net::IPEndPoint(net::IPAddressNumber(), address.port());
 419   } else {
 420     if (!jingle_glue::SocketAddressToIPEndPoint(address, &address_chrome)) {
 421       LOG(WARNING) << "Failed to convert remote address to IPEndPoint: address="
 422                    << address.ipaddr().ToSensitiveString()
 423                    << ", remote_address_="
 424                    << remote_address_.ipaddr().ToSensitiveString();
 425       NOTREACHED();
 426       error_ = EINVAL;
 427       return -1;
 428     }
 429   }
 430
 431   send_bytes_available_ -= data_size;
 432
 433   const char* data_char = reinterpret_cast<const char*>(data);
 434   std::vector<char> data_vector(data_char, data_char + data_size);
 435   uint64_t packet_id = client_->Send(address_chrome, data_vector, options);
 436
 437   // Ensure packet_id is not 0. It can't be the case according to
 438   // P2PSocketClientImpl::Send().
 439   DCHECK_NE(packet_id, 0uL);
 440
 441   in_flight_packet_records_.push_back(
 442       InFlightPacketRecord(packet_id, data_size));
 443   TraceSendThrottlingState();
 444
 445   // Fake successful send. The caller ignores result anyway.
 446   return data_size;
 447 }
 448
 449 int IpcPacketSocket::Close() {
 450   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 451
 452   client_->Close();
 453   state_ = IS_CLOSED;
 454
 455   return 0;
 456 }
 457
 458 rtc::AsyncPacketSocket::State IpcPacketSocket::GetState() const {
 459   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 460
 461   switch (state_) {
 462     case IS_UNINITIALIZED:
 463       NOTREACHED();
 464       return STATE_CLOSED;
 465
 466     case IS_OPENING:
 467       return STATE_BINDING;
 468
 469     case IS_OPEN:
 470       if (IsTcpClientSocket(type_)) {
 471         return STATE_CONNECTED;
 472       } else {
 473         return STATE_BOUND;
 474       }
 475
 476     case IS_CLOSED:
 477     case IS_ERROR:
 478       return STATE_CLOSED;
 479   }
 480
 481   NOTREACHED();
 482   return STATE_CLOSED;
 483 }
 484
 485 int IpcPacketSocket::GetOption(rtc::Socket::Option option, int* value) {
 486   P2PSocketOption p2p_socket_option = P2P_SOCKET_OPT_MAX;
 487   if (!JingleSocketOptionToP2PSocketOption(option, &p2p_socket_option)) {
 488     // unsupported option.
 489     return -1;
 490   }
 491
 492   *value = options_[p2p_socket_option];
 493   return 0;
 494 }
 495
 496 int IpcPacketSocket::SetOption(rtc::Socket::Option option, int value) {
 497   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 498
 499   P2PSocketOption p2p_socket_option = P2P_SOCKET_OPT_MAX;
 500   if (!JingleSocketOptionToP2PSocketOption(option, &p2p_socket_option)) {
 501     // Option is not supported.
 502     return -1;
 503   }
 504
 505   options_[p2p_socket_option] = value;
 506
 507   if (state_ == IS_OPEN) {
 508     // Options will be applied when state becomes IS_OPEN in OnOpen.
 509     return DoSetOption(p2p_socket_option, value);
 510   }
 511   return 0;
 512 }
 513
 514 int IpcPacketSocket::DoSetOption(P2PSocketOption option, int value) {
 515   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 516   DCHECK_EQ(state_, IS_OPEN);
 517
 518   client_->SetOption(option, value);
 519   return 0;
 520 }
 521
 522 int IpcPacketSocket::GetError() const {
 523   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 524   return error_;
 525 }
 526
 527 void IpcPacketSocket::SetError(int error) {
 528   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 529   error_ = error;
 530 }
 531
 532 void IpcPacketSocket::OnOpen(const net::IPEndPoint& local_address,
 533                              const net::IPEndPoint& remote_address) {
 534   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 535
 536   if (!jingle_glue::IPEndPointToSocketAddress(local_address, &local_address_)) {
 537     // Always expect correct IPv4 address to be allocated.
 538     NOTREACHED();
 539     OnError();
 540     return;
 541   }
 542
 543   state_ = IS_OPEN;
 544   TraceSendThrottlingState();
 545
 546   // Set all pending options if any.
 547   for (int i = 0; i < P2P_SOCKET_OPT_MAX; ++i) {
 548     if (options_[i] != kDefaultNonSetOptionValue)
 549       DoSetOption(static_cast<P2PSocketOption> (i), options_[i]);
 550   }
 551
 552   SignalAddressReady(this, local_address_);
 553   if (IsTcpClientSocket(type_)) {
 554     // If remote address is unresolved, set resolved remote IP address received
 555     // in the callback. This address will be used while sending the packets
 556     // over the network.
 557     if (remote_address_.IsUnresolvedIP()) {
 558       rtc::SocketAddress jingle_socket_address;
 559       // |remote_address| could be unresolved if the connection is behind a
 560       // proxy.
 561       if (!remote_address.address().empty() &&
 562           jingle_glue::IPEndPointToSocketAddress(
 563               remote_address, &jingle_socket_address)) {
 564         // Set only the IP address.
 565         remote_address_.SetResolvedIP(jingle_socket_address.ipaddr());
 566       }
 567     }
 568
 569     // SignalConnect after updating the |remote_address_| so that the listener
 570     // can get the resolved remote address.
 571     SignalConnect(this);
 572   }
 573 }
 574
 575 void IpcPacketSocket::OnIncomingTcpConnection(
 576     const net::IPEndPoint& address,
 577     P2PSocketClient* client) {
 578   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 579
 580   scoped_ptr<IpcPacketSocket> socket(new IpcPacketSocket());
 581
 582   rtc::SocketAddress remote_address;
 583   if (!jingle_glue::IPEndPointToSocketAddress(address, &remote_address)) {
 584     // Always expect correct IPv4 address to be allocated.
 585     NOTREACHED();
 586   }
 587   socket->InitAcceptedTcp(client, local_address_, remote_address);
 588   SignalNewConnection(this, socket.release());
 589 }
 590
 591 void IpcPacketSocket::OnSendComplete(const P2PSendPacketMetrics& send_metrics) {
 592   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 593
 594   CHECK(!in_flight_packet_records_.empty());
 595
 596   const InFlightPacketRecord& record = in_flight_packet_records_.front();
 597
 598   // Tracking is not turned on for TCP so it's always 0. For UDP, this will
 599   // cause a crash when the packet ids don't match.
 600   CHECK(send_metrics.packet_id == 0 ||
 601         record.packet_id == send_metrics.packet_id);
 602
 603   send_bytes_available_ += record.packet_size;
 604
 605   DCHECK_LE(send_bytes_available_, kMaximumInFlightBytes);
 606
 607   in_flight_packet_records_.pop_front();
 608   TraceSendThrottlingState();
 609
 610   if (writable_signal_expected_ && send_bytes_available_ > 0) {
 611     WebRtcLogMessage(base::StringPrintf(
 612         "IpcPacketSocket: sending is unblocked. %d packets in flight.",
 613         static_cast<int>(in_flight_packet_records_.size())));
 614
 615     SignalReadyToSend(this);
 616     writable_signal_expected_ = false;
 617   }
 618 }
 619
 620 void IpcPacketSocket::OnError() {
 621   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 622   bool was_closed = (state_ == IS_ERROR || state_ == IS_CLOSED);
 623   state_ = IS_ERROR;
 624   error_ = ECONNABORTED;
 625   if (!was_closed) {
 626     SignalClose(this, 0);
 627   }
 628 }
 629
 630 void IpcPacketSocket::OnDataReceived(const net::IPEndPoint& address,
 631                                      const std::vector<char>& data,
 632                                      const base::TimeTicks& timestamp) {
 633   DCHECK_EQ(base::MessageLoop::current(), message_loop_);
 634
 635   rtc::SocketAddress address_lj;
 636
 637   if (address.address().empty()) {
 638     DCHECK(IsTcpClientSocket(type_));
 639     // |address| could be empty for TCP connections behind a proxy.
 640     address_lj = remote_address_;
 641   } else {
 642     if (!jingle_glue::IPEndPointToSocketAddress(address, &address_lj)) {
 643       // We should always be able to convert address here because we
 644       // don't expect IPv6 address on IPv4 connections.
 645       NOTREACHED();
 646       return;
 647     }
 648   }
 649
 650   rtc::PacketTime packet_time(timestamp.ToInternalValue(), 0);
 651   SignalReadPacket(this, &data[0], data.size(), address_lj,
 652                    packet_time);
 653 }
 654
 655 AsyncAddressResolverImpl::AsyncAddressResolverImpl(
 656     P2PSocketDispatcher* dispatcher)
 657     : resolver_(new P2PAsyncAddressResolver(dispatcher)) {
 658 }
 659
 660 AsyncAddressResolverImpl::~AsyncAddressResolverImpl() {
 661 }
 662
 663 void AsyncAddressResolverImpl::Start(const rtc::SocketAddress& addr) {
 664   DCHECK(CalledOnValidThread());
 665   // Copy port number from |addr|. |port_| must be copied
 666   // when resolved address is returned in GetResolvedAddress.
 667   port_ = addr.port();
 668
 669   resolver_->Start(addr, base::Bind(
 670       &AsyncAddressResolverImpl::OnAddressResolved,
 671       base::Unretained(this)));
 672 }
 673
 674 bool AsyncAddressResolverImpl::GetResolvedAddress(
 675     int family, rtc::SocketAddress* addr) const {
 676   DCHECK(CalledOnValidThread());
 677
 678   if (addresses_.empty())
 679    return false;
 680
 681   for (size_t i = 0; i < addresses_.size(); ++i) {
 682     if (family == addresses_[i].family()) {
 683       addr->SetResolvedIP(addresses_[i]);
 684       addr->SetPort(port_);
 685       return true;
 686     }
 687   }
 688   return false;
 689 }
 690
 691 int AsyncAddressResolverImpl::GetError() const {
 692   DCHECK(CalledOnValidThread());
 693   return addresses_.empty() ? -1 : 0;
 694 }
 695
 696 void AsyncAddressResolverImpl::Destroy(bool wait) {
 697   DCHECK(CalledOnValidThread());
 698   resolver_->Cancel();
 699   // Libjingle doesn't need this object any more and it's not going to delete
 700   // it explicitly.
 701   delete this;
 702 }
 703
 704 void AsyncAddressResolverImpl::OnAddressResolved(
 705     const net::IPAddressList& addresses) {
 706   DCHECK(CalledOnValidThread());
 707   for (size_t i = 0; i < addresses.size(); ++i) {
 708     rtc::SocketAddress socket_address;
 709     if (!jingle_glue::IPEndPointToSocketAddress(
 710             net::IPEndPoint(addresses[i], 0), &socket_address)) {
 711       NOTREACHED();
 712     }
 713     addresses_.push_back(socket_address.ipaddr());
 714   }
 715   SignalDone(this);
 716 }
 717
 718 }  // namespace
 719
 720 IpcPacketSocketFactory::IpcPacketSocketFactory(
 721     P2PSocketDispatcher* socket_dispatcher)
 722     : socket_dispatcher_(socket_dispatcher) {
 723 }
 724
 725 IpcPacketSocketFactory::~IpcPacketSocketFactory() {
 726 }
 727
 728 rtc::AsyncPacketSocket* IpcPacketSocketFactory::CreateUdpSocket(
 729     const rtc::SocketAddress& local_address, uint16 min_port, uint16 max_port) {
 730   rtc::SocketAddress crome_address;
 731   P2PSocketClientImpl* socket_client =
 732       new P2PSocketClientImpl(socket_dispatcher_);
 733   scoped_ptr<IpcPacketSocket> socket(new IpcPacketSocket());
 734   // TODO(sergeyu): Respect local_address and port limits here (need
 735   // to pass them over IPC channel to the browser).
 736   if (!socket->Init(P2P_SOCKET_UDP, socket_client,
 737                     local_address, rtc::SocketAddress())) {
 738     return NULL;
 739   }
 740   return socket.release();
 741 }
 742
 743 rtc::AsyncPacketSocket* IpcPacketSocketFactory::CreateServerTcpSocket(
 744     const rtc::SocketAddress& local_address, uint16 min_port, uint16 max_port,
 745     int opts) {
 746   // TODO(sergeyu): Implement SSL support.
 747   if (opts & rtc::PacketSocketFactory::OPT_SSLTCP)
 748     return NULL;
 749
 750   P2PSocketType type = (opts & rtc::PacketSocketFactory::OPT_STUN) ?
 751       P2P_SOCKET_STUN_TCP_SERVER : P2P_SOCKET_TCP_SERVER;
 752   P2PSocketClientImpl* socket_client =
 753       new P2PSocketClientImpl(socket_dispatcher_);
 754   scoped_ptr<IpcPacketSocket> socket(new IpcPacketSocket());
 755   if (!socket->Init(type, socket_client, local_address,
 756                     rtc::SocketAddress())) {
 757     return NULL;
 758   }
 759   return socket.release();
 760 }
 761
 762 rtc::AsyncPacketSocket* IpcPacketSocketFactory::CreateClientTcpSocket(
 763     const rtc::SocketAddress& local_address,
 764     const rtc::SocketAddress& remote_address,
 765     const rtc::ProxyInfo& proxy_info,
 766     const std::string& user_agent, int opts) {
 767   P2PSocketType type;
 768   if (opts & rtc::PacketSocketFactory::OPT_SSLTCP) {
 769     type = (opts & rtc::PacketSocketFactory::OPT_STUN) ?
 770         P2P_SOCKET_STUN_SSLTCP_CLIENT : P2P_SOCKET_SSLTCP_CLIENT;
 771   } else if (opts & rtc::PacketSocketFactory::OPT_TLS) {
 772     type = (opts & rtc::PacketSocketFactory::OPT_STUN) ?
 773         P2P_SOCKET_STUN_TLS_CLIENT : P2P_SOCKET_TLS_CLIENT;
 774   } else {
 775     type = (opts & rtc::PacketSocketFactory::OPT_STUN) ?
 776         P2P_SOCKET_STUN_TCP_CLIENT : P2P_SOCKET_TCP_CLIENT;
 777   }
 778   P2PSocketClientImpl* socket_client =
 779       new P2PSocketClientImpl(socket_dispatcher_);
 780   scoped_ptr<IpcPacketSocket> socket(new IpcPacketSocket());
 781   if (!socket->Init(type, socket_client, local_address, remote_address))
 782     return NULL;
 783   return socket.release();
 784 }
 785
 786 rtc::AsyncResolverInterface*
 787 IpcPacketSocketFactory::CreateAsyncResolver() {
 788   scoped_ptr<AsyncAddressResolverImpl> resolver(
 789     new AsyncAddressResolverImpl(socket_dispatcher_));
 790   return resolver.release();
 791 }
 792
 793 }  // namespace content