remoting/jingle_glue/chromium_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 "remoting/jingle_glue/chromium_socket_factory.h"
   6
   7 #include "base/bind.h"
   8 #include "base/logging.h"
   9 #include "base/memory/scoped_ptr.h"
  10 #include "jingle/glue/utils.h"
  11 #include "net/base/io_buffer.h"
  12 #include "net/base/ip_endpoint.h"
  13 #include "net/base/net_errors.h"
  14 #include "net/udp/udp_server_socket.h"
  15 #include "third_party/libjingle/source/talk/base/asyncpacketsocket.h"
  16 #include "third_party/libjingle/source/talk/base/asyncresolverinterface.h"
  17
  18 namespace remoting {
  19
  20 namespace {
  21
  22 // Size of the buffer to allocate for RecvFrom().
  23 const int kReceiveBufferSize = 65536;
  24
  25 // Maximum amount of data in the send buffers. This is necessary to
  26 // prevent out-of-memory crashes if the caller sends data faster than
  27 // Pepper's UDP API can handle it. This maximum should never be
  28 // reached under normal conditions.
  29 const int kMaxSendBufferSize = 256 * 1024;
  30
  31 // Defines set of transient errors. These errors are ignored when we get them
  32 // from sendto() calls.
  33 bool IsTransientError(int error) {
  34   return error == net::ERR_ADDRESS_UNREACHABLE ||
  35       error == net::ERR_ADDRESS_INVALID;
  36 }
  37
  38 // TODO(lambroslambrou): Move STUN/relay address resolution from
  39 // PepperPortAllocator to this class.
  40 class DummyAsyncResolver : public talk_base::AsyncResolverInterface {
  41  public:
  42   DummyAsyncResolver() {}
  43   virtual ~DummyAsyncResolver() {}
  44   virtual void Start(const talk_base::SocketAddress& addr) OVERRIDE {}
  45   virtual bool GetResolvedAddress(
  46       int family,
  47       talk_base::SocketAddress* addr) const OVERRIDE {
  48     return false;
  49   }
  50   virtual int GetError() const OVERRIDE {
  51     return 0;
  52   }
  53   virtual void Destroy(bool wait) OVERRIDE {
  54     delete this;
  55   }
  56
  57  private:
  58   DISALLOW_COPY_AND_ASSIGN(DummyAsyncResolver);
  59 };
  60
  61 class UdpPacketSocket : public talk_base::AsyncPacketSocket {
  62  public:
  63   UdpPacketSocket();
  64   virtual ~UdpPacketSocket();
  65
  66   bool Init(const talk_base::SocketAddress& local_address,
  67             int min_port, int max_port);
  68
  69   // talk_base::AsyncPacketSocket interface.
  70   virtual talk_base::SocketAddress GetLocalAddress() const OVERRIDE;
  71   virtual talk_base::SocketAddress GetRemoteAddress() const OVERRIDE;
  72   virtual int Send(const void* data, size_t data_size,
  73                    talk_base::DiffServCodePoint dscp) OVERRIDE;
  74   virtual int SendTo(const void* data, size_t data_size,
  75                      const talk_base::SocketAddress& address,
  76                      talk_base::DiffServCodePoint dscp) OVERRIDE;
  77   virtual int Close() OVERRIDE;
  78   virtual State GetState() const OVERRIDE;
  79   virtual int GetOption(talk_base::Socket::Option option, int* value) OVERRIDE;
  80   virtual int SetOption(talk_base::Socket::Option option, int value) OVERRIDE;
  81   virtual int GetError() const OVERRIDE;
  82   virtual void SetError(int error) OVERRIDE;
  83
  84  private:
  85   struct PendingPacket {
  86     PendingPacket(const void* buffer,
  87                   int buffer_size,
  88                   const net::IPEndPoint& address);
  89
  90     scoped_refptr<net::IOBufferWithSize> data;
  91     net::IPEndPoint address;
  92   };
  93
  94   void OnBindCompleted(int error);
  95
  96   void DoSend();
  97   void OnSendCompleted(int result);
  98
  99   void DoRead();
 100   void OnReadCompleted(int result);
 101   void HandleReadResult(int result);
 102
 103   scoped_ptr<net::UDPServerSocket> socket_;
 104
 105   State state_;
 106   int error_;
 107
 108   talk_base::SocketAddress local_address_;
 109
 110   // Receive buffer and address are populated by asynchronous reads.
 111   scoped_refptr<net::IOBuffer> receive_buffer_;
 112   net::IPEndPoint receive_address_;
 113
 114   bool send_pending_;
 115   std::list<PendingPacket> send_queue_;
 116   int send_queue_size_;
 117
 118   DISALLOW_COPY_AND_ASSIGN(UdpPacketSocket);
 119 };
 120
 121 UdpPacketSocket::PendingPacket::PendingPacket(
 122     const void* buffer,
 123     int buffer_size,
 124     const net::IPEndPoint& address)
 125     : data(new net::IOBufferWithSize(buffer_size)),
 126       address(address) {
 127   memcpy(data->data(), buffer, buffer_size);
 128 }
 129
 130 UdpPacketSocket::UdpPacketSocket()
 131     : state_(STATE_CLOSED),
 132       error_(0),
 133       send_pending_(false),
 134       send_queue_size_(0) {
 135 }
 136
 137 UdpPacketSocket::~UdpPacketSocket() {
 138   Close();
 139 }
 140
 141 bool UdpPacketSocket::Init(const talk_base::SocketAddress& local_address,
 142                            int min_port, int max_port) {
 143   net::IPEndPoint local_endpoint;
 144   if (!jingle_glue::SocketAddressToIPEndPoint(
 145           local_address, &local_endpoint)) {
 146     return false;
 147   }
 148
 149   for (int port = min_port; port <= max_port; ++port) {
 150     socket_.reset(new net::UDPServerSocket(NULL, net::NetLog::Source()));
 151     int result = socket_->Listen(
 152         net::IPEndPoint(local_endpoint.address(), port));
 153     if (result == net::OK) {
 154       break;
 155     } else {
 156       socket_.reset();
 157     }
 158   }
 159
 160   if (!socket_.get()) {
 161     // Failed to bind the socket.
 162     return false;
 163   }
 164
 165   if (socket_->GetLocalAddress(&local_endpoint) != net::OK ||
 166       !jingle_glue::IPEndPointToSocketAddress(local_endpoint,
 167                                               &local_address_)) {
 168     return false;
 169   }
 170
 171   state_ = STATE_BOUND;
 172   DoRead();
 173
 174   return true;
 175 }
 176
 177 talk_base::SocketAddress UdpPacketSocket::GetLocalAddress() const {
 178   DCHECK_EQ(state_, STATE_BOUND);
 179   return local_address_;
 180 }
 181
 182 talk_base::SocketAddress UdpPacketSocket::GetRemoteAddress() const {
 183   // UDP sockets are not connected - this method should never be called.
 184   NOTREACHED();
 185   return talk_base::SocketAddress();
 186 }
 187
 188 int UdpPacketSocket::Send(const void* data, size_t data_size,
 189                           talk_base::DiffServCodePoint dscp) {
 190   // UDP sockets are not connected - this method should never be called.
 191   NOTREACHED();
 192   return EWOULDBLOCK;
 193 }
 194
 195 int UdpPacketSocket::SendTo(const void* data, size_t data_size,
 196                             const talk_base::SocketAddress& address,
 197                             talk_base::DiffServCodePoint dscp) {
 198   if (state_ != STATE_BOUND) {
 199     NOTREACHED();
 200     return EINVAL;
 201   }
 202
 203   if (error_ != 0) {
 204     return error_;
 205   }
 206
 207   net::IPEndPoint endpoint;
 208   if (!jingle_glue::SocketAddressToIPEndPoint(address, &endpoint)) {
 209     return EINVAL;
 210   }
 211
 212   if (send_queue_size_ >= kMaxSendBufferSize) {
 213     return EWOULDBLOCK;
 214   }
 215
 216   send_queue_.push_back(PendingPacket(data, data_size, endpoint));
 217   send_queue_size_ += data_size;
 218
 219   DoSend();
 220   return data_size;
 221 }
 222
 223 int UdpPacketSocket::Close() {
 224   state_ = STATE_CLOSED;
 225   socket_.reset();
 226   return 0;
 227 }
 228
 229 talk_base::AsyncPacketSocket::State UdpPacketSocket::GetState() const {
 230   return state_;
 231 }
 232
 233 int UdpPacketSocket::GetOption(talk_base::Socket::Option option, int* value) {
 234   // This method is never called by libjingle.
 235   NOTIMPLEMENTED();
 236   return -1;
 237 }
 238
 239 int UdpPacketSocket::SetOption(talk_base::Socket::Option option, int value) {
 240   if (state_ != STATE_BOUND) {
 241     NOTREACHED();
 242     return EINVAL;
 243   }
 244
 245   switch (option) {
 246     case talk_base::Socket::OPT_DONTFRAGMENT:
 247       NOTIMPLEMENTED();
 248       return -1;
 249
 250     case talk_base::Socket::OPT_RCVBUF: {
 251       bool success = socket_->SetReceiveBufferSize(value);
 252       return success ? 0 : -1;
 253     }
 254
 255     case talk_base::Socket::OPT_SNDBUF: {
 256       bool success = socket_->SetSendBufferSize(value);
 257       return success ? 0 : -1;
 258     }
 259
 260     case talk_base::Socket::OPT_NODELAY:
 261       // OPT_NODELAY is only for TCP sockets.
 262       NOTREACHED();
 263       return -1;
 264
 265     case talk_base::Socket::OPT_IPV6_V6ONLY:
 266       NOTIMPLEMENTED();
 267       return -1;
 268
 269     case talk_base::Socket::OPT_DSCP:
 270       NOTIMPLEMENTED();
 271       return -1;
 272   }
 273
 274   NOTREACHED();
 275   return -1;
 276 }
 277
 278 int UdpPacketSocket::GetError() const {
 279   return error_;
 280 }
 281
 282 void UdpPacketSocket::SetError(int error) {
 283   error_ = error;
 284 }
 285
 286 void UdpPacketSocket::DoSend() {
 287   if (send_pending_ || send_queue_.empty())
 288     return;
 289
 290   PendingPacket& packet = send_queue_.front();
 291   int result = socket_->SendTo(
 292       packet.data.get(),
 293       packet.data->size(),
 294       packet.address,
 295       base::Bind(&UdpPacketSocket::OnSendCompleted, base::Unretained(this)));
 296   if (result == net::ERR_IO_PENDING) {
 297     send_pending_ = true;
 298   } else {
 299     OnSendCompleted(result);
 300   }
 301 }
 302
 303 void UdpPacketSocket::OnSendCompleted(int result) {
 304   send_pending_ = false;
 305
 306   if (result < 0) {
 307     if (!IsTransientError(result)) {
 308       LOG(ERROR) << "Send failed on a UDP socket: " << result;
 309       error_ = EINVAL;
 310       return;
 311     }
 312   }
 313
 314   // Don't need to worry about partial sends because this is a datagram
 315   // socket.
 316   send_queue_size_ -= send_queue_.front().data->size();
 317   send_queue_.pop_front();
 318   DoSend();
 319 }
 320
 321 void UdpPacketSocket::DoRead() {
 322   int result = 0;
 323   while (result >= 0) {
 324     receive_buffer_ = new net::IOBuffer(kReceiveBufferSize);
 325     result = socket_->RecvFrom(
 326         receive_buffer_.get(),
 327         kReceiveBufferSize,
 328         &receive_address_,
 329         base::Bind(&UdpPacketSocket::OnReadCompleted, base::Unretained(this)));
 330     HandleReadResult(result);
 331   }
 332 }
 333
 334 void UdpPacketSocket::OnReadCompleted(int result) {
 335   HandleReadResult(result);
 336   if (result >= 0) {
 337     DoRead();
 338   }
 339 }
 340
 341 void UdpPacketSocket::HandleReadResult(int result) {
 342   if (result == net::ERR_IO_PENDING) {
 343     return;
 344   }
 345
 346   if (result > 0) {
 347     talk_base::SocketAddress address;
 348     if (!jingle_glue::IPEndPointToSocketAddress(receive_address_, &address)) {
 349       NOTREACHED();
 350       LOG(ERROR) << "Failed to convert address received from RecvFrom().";
 351       return;
 352     }
 353     SignalReadPacket(this, receive_buffer_->data(), result, address);
 354   } else {
 355     LOG(ERROR) << "Received error when reading from UDP socket: " << result;
 356   }
 357 }
 358
 359 }  // namespace
 360
 361 ChromiumPacketSocketFactory::ChromiumPacketSocketFactory() {
 362 }
 363
 364 ChromiumPacketSocketFactory::~ChromiumPacketSocketFactory() {
 365 }
 366
 367 talk_base::AsyncPacketSocket* ChromiumPacketSocketFactory::CreateUdpSocket(
 368       const talk_base::SocketAddress& local_address,
 369       int min_port, int max_port) {
 370   scoped_ptr<UdpPacketSocket> result(new UdpPacketSocket());
 371   if (!result->Init(local_address, min_port, max_port))
 372     return NULL;
 373   return result.release();
 374 }
 375
 376 talk_base::AsyncPacketSocket*
 377 ChromiumPacketSocketFactory::CreateServerTcpSocket(
 378     const talk_base::SocketAddress& local_address,
 379     int min_port, int max_port,
 380     int opts) {
 381   // We don't use TCP sockets for remoting connections.
 382   NOTREACHED();
 383   return NULL;
 384 }
 385
 386 talk_base::AsyncPacketSocket*
 387 ChromiumPacketSocketFactory::CreateClientTcpSocket(
 388       const talk_base::SocketAddress& local_address,
 389       const talk_base::SocketAddress& remote_address,
 390       const talk_base::ProxyInfo& proxy_info,
 391       const std::string& user_agent,
 392       int opts) {
 393   // We don't use TCP sockets for remoting connections.
 394   NOTREACHED();
 395   return NULL;
 396 }
 397
 398 talk_base::AsyncResolverInterface*
 399 ChromiumPacketSocketFactory::CreateAsyncResolver() {
 400   return new DummyAsyncResolver();
 401 }
 402
 403 }  // namespace remoting