diff options
Diffstat (limited to 'chromium/content/renderer/p2p/ipc_socket_factory.cc')
| -rw-r--r-- | chromium/content/renderer/p2p/ipc_socket_factory.cc | 220 |
1 files changed, 200 insertions, 20 deletions
diff --git a/chromium/content/renderer/p2p/ipc_socket_factory.cc b/chromium/content/renderer/p2p/ipc_socket_factory.cc index 04825c35f75..43b6d163ff4 100644 --- a/chromium/content/renderer/p2p/ipc_socket_factory.cc +++ b/chromium/content/renderer/p2p/ipc_socket_factory.cc @@ -4,14 +4,18 @@ #include "content/renderer/p2p/ipc_socket_factory.h" +#include <algorithm> #include <deque> #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 "content/public/renderer/p2p_socket_client_delegate.h" +#include "base/strings/stringprintf.h" +#include "base/threading/non_thread_safe.h" +#include "content/renderer/media/webrtc_logging.h" #include "content/renderer/p2p/host_address_request.h" +#include "content/renderer/p2p/socket_client_delegate.h" #include "content/renderer/p2p/socket_client_impl.h" #include "content/renderer/p2p/socket_dispatcher.h" #include "jingle/glue/utils.h" @@ -21,6 +25,8 @@ namespace content { namespace { +const int kDefaultNonSetOptionValue = -1; + bool IsTcpClientSocket(P2PSocketType type) { return (type == P2P_SOCKET_STUN_TCP_CLIENT) || (type == P2P_SOCKET_TCP_CLIENT) || @@ -30,6 +36,30 @@ bool IsTcpClientSocket(P2PSocketType type) { (type == P2P_SOCKET_STUN_TLS_CLIENT); } +bool JingleSocketOptionToP2PSocketOption(talk_base::Socket::Option option, + P2PSocketOption* ipc_option) { + switch (option) { + case talk_base::Socket::OPT_RCVBUF: + *ipc_option = P2P_SOCKET_OPT_RCVBUF; + break; + case talk_base::Socket::OPT_SNDBUF: + *ipc_option = P2P_SOCKET_OPT_SNDBUF; + break; + case talk_base::Socket::OPT_DSCP: + *ipc_option = P2P_SOCKET_OPT_DSCP; + break; + case talk_base::Socket::OPT_DONTFRAGMENT: + case talk_base::Socket::OPT_NODELAY: + case talk_base::Socket::OPT_IPV6_V6ONLY: + case talk_base::Socket::OPT_RTP_SENDTIME_EXTN_ID: + return false; // Not supported by the chrome sockets. + default: + NOTREACHED(); + return false; + } + return true; +} + // TODO(miu): This needs tuning. http://crbug.com/237960 const size_t kMaximumInFlightBytes = 64 * 1024; // 64 KB @@ -51,14 +81,14 @@ class IpcPacketSocket : public talk_base::AsyncPacketSocket, virtual talk_base::SocketAddress GetLocalAddress() const OVERRIDE; virtual talk_base::SocketAddress GetRemoteAddress() const OVERRIDE; virtual int Send(const void *pv, size_t cb, - talk_base::DiffServCodePoint dscp) OVERRIDE; + const talk_base::PacketOptions& options) OVERRIDE; virtual int SendTo(const void *pv, size_t cb, const talk_base::SocketAddress& addr, - talk_base::DiffServCodePoint dscp) OVERRIDE; + const talk_base::PacketOptions& options) OVERRIDE; virtual int Close() OVERRIDE; virtual State GetState() const OVERRIDE; - virtual int GetOption(talk_base::Socket::Option opt, int* value) OVERRIDE; - virtual int SetOption(talk_base::Socket::Option opt, int value) OVERRIDE; + virtual int GetOption(talk_base::Socket::Option option, int* value) OVERRIDE; + virtual int SetOption(talk_base::Socket::Option option, int value) OVERRIDE; virtual int GetError() const OVERRIDE; virtual void SetError(int error) OVERRIDE; @@ -90,6 +120,9 @@ class IpcPacketSocket : public talk_base::AsyncPacketSocket, void InitAcceptedTcp(P2PSocketClient* client, const talk_base::SocketAddress& local_address, const talk_base::SocketAddress& remote_address); + + int DoSetOption(P2PSocketOption option, int value); + P2PSocketType type_; // Message loop on which this socket was created and being used. @@ -124,10 +157,37 @@ class IpcPacketSocket : public talk_base::AsyncPacketSocket, // Current error code. Valid when state_ == IS_ERROR. int error_; + int options_[P2P_SOCKET_OPT_MAX]; DISALLOW_COPY_AND_ASSIGN(IpcPacketSocket); }; +// Simple wrapper around P2PAsyncAddressResolver. The main purpose of this +// class is to send SignalDone, after OnDone callback from +// P2PAsyncAddressResolver. Libjingle sig slots are not thread safe. In case +// of MT sig slots clients must call disconnect. This class is to make sure +// we destruct from the same thread on which is created. +class AsyncAddressResolverImpl : public base::NonThreadSafe, + public talk_base::AsyncResolverInterface { + public: + AsyncAddressResolverImpl(P2PSocketDispatcher* dispatcher); + virtual ~AsyncAddressResolverImpl(); + + // talk_base::AsyncResolverInterface interface. + virtual void Start(const talk_base::SocketAddress& addr) OVERRIDE; + virtual bool GetResolvedAddress( + int family, talk_base::SocketAddress* addr) const OVERRIDE; + virtual int GetError() const OVERRIDE; + virtual void Destroy(bool wait) OVERRIDE; + + private: + virtual void OnAddressResolved(const net::IPAddressList& addresses); + + scoped_refptr<P2PAsyncAddressResolver> resolver_; + int port_; // Port number in |addr| from Start() method. + std::vector<talk_base::IPAddress> addresses_; // Resolved addresses. +}; + IpcPacketSocket::IpcPacketSocket() : type_(P2P_SOCKET_UDP), message_loop_(base::MessageLoop::current()), @@ -136,6 +196,8 @@ IpcPacketSocket::IpcPacketSocket() writable_signal_expected_(false), error_(0) { COMPILE_ASSERT(kMaximumInFlightBytes > 0, would_send_at_zero_rate); + std::fill_n(options_, static_cast<int> (P2P_SOCKET_OPT_MAX), + kDefaultNonSetOptionValue); } IpcPacketSocket::~IpcPacketSocket() { @@ -178,7 +240,12 @@ bool IpcPacketSocket::Init(P2PSocketType type, return false; } - client->Init(type, local_endpoint, remote_endpoint, this); + // We need to send both resolved and unresolved address in Init. Unresolved + // address will be used in case of TLS for certificate hostname matching. + // Certificate will be tied to domain name not to IP address. + P2PHostAndIPEndPoint remote_info(remote_address.hostname(), remote_endpoint); + + client->Init(type, local_endpoint, remote_info, this); return true; } @@ -210,14 +277,14 @@ talk_base::SocketAddress IpcPacketSocket::GetRemoteAddress() const { } int IpcPacketSocket::Send(const void *data, size_t data_size, - talk_base::DiffServCodePoint dscp) { + const talk_base::PacketOptions& options) { DCHECK_EQ(base::MessageLoop::current(), message_loop_); - return SendTo(data, data_size, remote_address_, dscp); + return SendTo(data, data_size, remote_address_, options); } int IpcPacketSocket::SendTo(const void *data, size_t data_size, const talk_base::SocketAddress& address, - talk_base::DiffServCodePoint dscp) { + const talk_base::PacketOptions& options) { DCHECK_EQ(base::MessageLoop::current(), message_loop_); switch (state_) { @@ -245,7 +312,14 @@ int IpcPacketSocket::SendTo(const void *data, size_t data_size, TRACE_EVENT_SCOPE_THREAD, "id", client_->GetSocketID()); - writable_signal_expected_ = true; + if (!writable_signal_expected_) { + WebRtcLogMessage(base::StringPrintf( + "IpcPacketSocket: sending is blocked. %d packets_in_flight.", + static_cast<int>(in_flight_packet_sizes_.size()))); + + writable_signal_expected_ = true; + } + error_ = EWOULDBLOCK; return -1; } @@ -263,8 +337,7 @@ int IpcPacketSocket::SendTo(const void *data, size_t data_size, 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, - static_cast<net::DiffServCodePoint>(dscp)); + client_->SendWithDscp(address_chrome, data_vector, options); // Fake successful send. The caller ignores result anyway. return data_size; @@ -306,14 +379,41 @@ talk_base::AsyncPacketSocket::State IpcPacketSocket::GetState() const { return STATE_CLOSED; } -int IpcPacketSocket::GetOption(talk_base::Socket::Option opt, int* value) { - // We don't support socket options for IPC sockets. - return -1; +int IpcPacketSocket::GetOption(talk_base::Socket::Option option, int* value) { + P2PSocketOption p2p_socket_option = P2P_SOCKET_OPT_MAX; + if (!JingleSocketOptionToP2PSocketOption(option, &p2p_socket_option)) { + // unsupported option. + return -1; + } + + *value = options_[p2p_socket_option]; + return 0; +} + +int IpcPacketSocket::SetOption(talk_base::Socket::Option option, int value) { + DCHECK_EQ(base::MessageLoop::current(), message_loop_); + + P2PSocketOption p2p_socket_option = P2P_SOCKET_OPT_MAX; + if (!JingleSocketOptionToP2PSocketOption(option, &p2p_socket_option)) { + // Option is not supported. + return -1; + } + + options_[p2p_socket_option] = value; + + if (state_ == IS_OPEN) { + // Options will be applied when state becomes IS_OPEN in OnOpen. + return DoSetOption(p2p_socket_option, value); + } + return 0; } -int IpcPacketSocket::SetOption(talk_base::Socket::Option opt, int value) { - // We don't support socket options for IPC sockets. - return -1; +int IpcPacketSocket::DoSetOption(P2PSocketOption option, int value) { + DCHECK_EQ(base::MessageLoop::current(), message_loop_); + DCHECK_EQ(state_, IS_OPEN); + + client_->SetOption(option, value); + return 0; } int IpcPacketSocket::GetError() const { @@ -339,6 +439,12 @@ void IpcPacketSocket::OnOpen(const net::IPEndPoint& address) { state_ = IS_OPEN; TraceSendThrottlingState(); + // Set all pending options if any. + for (int i = 0; i < P2P_SOCKET_OPT_MAX; ++i) { + if (options_[i] != kDefaultNonSetOptionValue) + DoSetOption(static_cast<P2PSocketOption> (i), options_[i]); + } + SignalAddressReady(this, local_address_); if (IsTcpClientSocket(type_)) SignalConnect(this); @@ -365,11 +471,17 @@ void IpcPacketSocket::OnSendComplete() { CHECK(!in_flight_packet_sizes_.empty()); send_bytes_available_ += in_flight_packet_sizes_.front(); + DCHECK_LE(send_bytes_available_, kMaximumInFlightBytes); + in_flight_packet_sizes_.pop_front(); 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()))); + SignalReadyToSend(this); writable_signal_expected_ = false; } @@ -377,8 +489,12 @@ void IpcPacketSocket::OnSendComplete() { void IpcPacketSocket::OnError() { DCHECK_EQ(base::MessageLoop::current(), message_loop_); + bool was_closed = (state_ == IS_ERROR || state_ == IS_CLOSED); state_ = IS_ERROR; error_ = ECONNABORTED; + if (!was_closed) { + SignalClose(this, 0); + } } void IpcPacketSocket::OnDataReceived(const net::IPEndPoint& address, @@ -399,6 +515,69 @@ void IpcPacketSocket::OnDataReceived(const net::IPEndPoint& address, packet_time); } +AsyncAddressResolverImpl::AsyncAddressResolverImpl( + P2PSocketDispatcher* dispatcher) + : resolver_(new P2PAsyncAddressResolver(dispatcher)) { +} + +AsyncAddressResolverImpl::~AsyncAddressResolverImpl() { +} + +void AsyncAddressResolverImpl::Start(const talk_base::SocketAddress& addr) { + DCHECK(CalledOnValidThread()); + // Copy port number from |addr|. |port_| must be copied + // when resolved address is returned in GetResolvedAddress. + port_ = addr.port(); + + resolver_->Start(addr, base::Bind( + &AsyncAddressResolverImpl::OnAddressResolved, + base::Unretained(this))); +} + +bool AsyncAddressResolverImpl::GetResolvedAddress( + int family, talk_base::SocketAddress* addr) const { + DCHECK(CalledOnValidThread()); + + if (addresses_.empty()) + return false; + + for (size_t i = 0; i < addresses_.size(); ++i) { + if (family == addresses_[i].family()) { + addr->SetResolvedIP(addresses_[i]); + addr->SetPort(port_); + return true; + } + } + return false; +} + +int AsyncAddressResolverImpl::GetError() const { + DCHECK(CalledOnValidThread()); + return addresses_.empty() ? -1 : 0; +} + +void AsyncAddressResolverImpl::Destroy(bool wait) { + DCHECK(CalledOnValidThread()); + resolver_->Cancel(); + // Libjingle doesn't need this object any more and it's not going to delete + // it explicitly. + delete this; +} + +void AsyncAddressResolverImpl::OnAddressResolved( + const net::IPAddressList& addresses) { + DCHECK(CalledOnValidThread()); + for (size_t i = 0; i < addresses.size(); ++i) { + talk_base::SocketAddress socket_address; + if (!jingle_glue::IPEndPointToSocketAddress( + net::IPEndPoint(addresses[i], 0), &socket_address)) { + NOTREACHED(); + } + addresses_.push_back(socket_address.ipaddr()); + } + SignalDone(this); +} + } // namespace IpcPacketSocketFactory::IpcPacketSocketFactory( @@ -469,8 +648,9 @@ talk_base::AsyncPacketSocket* IpcPacketSocketFactory::CreateClientTcpSocket( talk_base::AsyncResolverInterface* IpcPacketSocketFactory::CreateAsyncResolver() { - NOTREACHED(); - return NULL; + scoped_ptr<AsyncAddressResolverImpl> resolver( + new AsyncAddressResolverImpl(socket_dispatcher_)); + return resolver.release(); } } // namespace content |