Implement PepperPacketSocketFactory.

remoting/client/plugin/pepper_packet_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 "remoting/client/plugin/pepper_packet_socket_factory.h"
+
+#include "base/bind.h"
+#include "base/logging.h"
+#include "net/base/io_buffer.h"
+#include "ppapi/cpp/private/net_address_private.h"
+#include "ppapi/cpp/private/udp_socket_private.h"
+#include "remoting/client/plugin/pepper_util.h"
+#include "third_party/libjingle/source/talk/base/asyncpacketsocket.h"
+
+namespace remoting {
+
+namespace {
+
+// Size of the buffer to allocate for RecvFrom().
+const int kReceiveBufferSize = 65536;
+
+// Maximum amount of data in the send buffers.
+const int kMaxSendBufferSize = 256 * 1024;

[Wez, 2012/04/20 21:37:07]

This is a limit on the amount of user data; the limit on the amount of memory
consumed by packet buffers is much higher if the caller sends lots of small
packets, and similarly the amount of data actually sent on the wire. :(

[Sergey Ulanov, 2012/04/23 22:16:51]

Right. I don't think it matters though. Normally that limit will never be
reached, it's just here to prevent from OOM in case there is a bug in the lower
layer and it stops sending data for some reason.

[Wez, 2012/04/25 00:01:59]

Right, so best to make it clear in the comment what the maximum is used for.

[Sergey Ulanov, 2012/04/25 00:25:01]

Done.

+
+class UdpPacketSocket : public talk_base::AsyncPacketSocket {
+ public:
+  explicit UdpPacketSocket(const pp::InstanceHandle& instance);
+  virtual ~UdpPacketSocket();
+
+  // Always takes ownership of client even if initialization fails.
+  bool Init(const talk_base::SocketAddress& local_address,
+            int min_port,
+            int max_port);
+
+  // talk_base::AsyncPacketSocket interface.
+  virtual talk_base::SocketAddress GetLocalAddress() const;
+  virtual talk_base::SocketAddress GetRemoteAddress() const;
+  virtual int Send(const void* data, size_t data_size);
+  virtual int SendTo(const void* data,
+                     size_t data_size,
+                     const talk_base::SocketAddress& address);
+  virtual int Close();
+  virtual State GetState() const;
+  virtual int GetOption(talk_base::Socket::Option opt, int* value);
+  virtual int SetOption(talk_base::Socket::Option opt, int value);
+  virtual int GetError() const;
+  virtual void SetError(int error);
+
+ private:
+  struct PendingPacket {
+    PendingPacket(const void* buffer,
+                  int buffer_size,
+                  const PP_NetAddress_Private& address);
+
+    scoped_refptr<net::IOBufferWithSize> data;
+    PP_NetAddress_Private address;
+  };
+
+  void OnBindCompleted(int error);
+
+  void DoSend();
+  void OnSendCompleted(int result);
+
+  void DoRead();
+  void OnReadCompleted(int result);
+  void HandleReadResult(int result);
+
+  pp::UDPSocketPrivate socket;

[Wez, 2012/04/20 21:37:07]

socket -> socket_

[Sergey Ulanov, 2012/04/23 22:16:51]

Done.

+
+  State state_;
+  int error_;
+
+  talk_base::SocketAddress local_address_;

[Wez, 2012/04/20 21:37:07]

Add a comment indicating that this is contains the address to bind to, or "any",
until we're bound after which point it contains the bound address & port.

[Sergey Ulanov, 2012/04/23 22:16:51]

Done.

+
+  // Used when the caller specified min_port_ and max_port_.

[Wez, 2012/04/20 21:37:07]

Doesn't the caller always specify those? The key thing to get across is that
these are only used during the Bind() process, when we're scanning for ports in
the caller-specified range.

[Sergey Ulanov, 2012/04/23 22:16:51]

Done.

+  uint16_t current_port_;

[Wez, 2012/04/20 21:37:07]

nit: |current_port_| sounds like it's the port to which we're bound. You could
instead just call in |min_port_|, unless you're worried that the fact it gets
incremented as we scan would be confusing.  Or |next_port_| (and increment
earlier)?

[Sergey Ulanov, 2012/04/23 22:16:51]

Done.

+  uint16_t max_port_;
+
+  std::vector<char> receive_buffer_;
+
+  bool send_pending_;
+  std::list<PendingPacket> send_queue_;
+  int send_queue_size_;
+
+  DISALLOW_COPY_AND_ASSIGN(UdpPacketSocket);
+};
+
+UdpPacketSocket::PendingPacket::PendingPacket(
+    const void* buffer,
+    int buffer_size,
+    const PP_NetAddress_Private& address)
+    : data(new net::IOBufferWithSize(buffer_size)),
+      address(address) {
+  memcpy(data->data(), buffer, buffer_size);
+}
+
+UdpPacketSocket::UdpPacketSocket(const pp::InstanceHandle& instance)
+    : socket(instance),
+      state_(STATE_CLOSED),
+      error_(0),
+      current_port_(0),
+      max_port_(0),
+      send_pending_(false),
+      send_queue_size_(0) {
+}
+
+UdpPacketSocket::~UdpPacketSocket() {
+}
+
+bool SocketAddressToPpAddressWithPort(const talk_base::SocketAddress& address,
+                                      PP_NetAddress_Private* pp_address,
+                                      uint16_t port) {
+  bool result;
+  switch(address.ipaddr().family()) {
+    case AF_INET: {
+      in_addr addr = address.ipaddr().ipv4_address();
+      result = pp::NetAddressPrivate::CreateFromIPv4Address(
+          reinterpret_cast<uint8_t*>(&addr), port, pp_address);
+      break;
+    }
+    case AF_INET6: {
+      in6_addr addr = address.ipaddr().ipv6_address();
+      result = pp::NetAddressPrivate::CreateFromIPv6Address(
+          addr.s6_addr, 0, port, pp_address);
+      break;
+    }
+    default: {
+      LOG(WARNING) << "Unknown address family: " << address.ipaddr().family();
+    }
+  }
+  if (!result) {
+    LOG(WARNING) << "Failed to convert address: " << address.ToString();
+  }
+  return result;
+}
+
+bool SocketAddressToPpAddress(const talk_base::SocketAddress& address,
+                              PP_NetAddress_Private* pp_address) {
+  return SocketAddressToPpAddressWithPort(address, pp_address, address.port());
+}
+
+bool PpAddressToSocketAddress(const PP_NetAddress_Private& pp_address,
+                              talk_base::SocketAddress* address) {
+  uint8_t addr_storage[16];
+  bool result = pp::NetAddressPrivate::GetAddress(
+      pp_address, &addr_storage, sizeof(addr_storage));
+
+  if (result) {
+    switch (pp::NetAddressPrivate::GetFamily(pp_address)) {
+      case PP_NETADDRESSFAMILY_IPV4:
+        address->SetIP(talk_base::IPAddress(
+            *reinterpret_cast<in_addr*>(addr_storage)));
+        break;
+      case PP_NETADDRESSFAMILY_IPV6:
+        address->SetIP(talk_base::IPAddress(
+            *reinterpret_cast<in6_addr*>(addr_storage)));
+        break;
+      default:
+        result = false;
+    }
+  }
+
+  if (!result) {
+    LOG(WARNING) << "Failed to convert address: "
+                 << pp::NetAddressPrivate::Describe(pp_address, true);
+  } else {
+    address->SetPort(pp::NetAddressPrivate::GetPort(pp_address));
+  }
+  return result;
+}
+
+bool UdpPacketSocket::Init(const talk_base::SocketAddress& local_address,
+                           int min_port,
+                           int max_port) {
+  if (socket.is_null()) {
+    return false;
+  }
+
+  local_address_ = local_address;
+  max_port_ = max_port;
+  current_port_ = min_port;
+
+  PP_NetAddress_Private pp_local_address;
+  if (!SocketAddressToPpAddressWithPort(local_address_, &pp_local_address,
+                                        current_port_)) {
+    return false;
+  }
+
+  int result = socket.Bind(&pp_local_address, PpCompletionCallback(
+      base::Bind(&UdpPacketSocket::OnBindCompleted, base::Unretained(this))));
+  DCHECK_EQ(result, PP_OK_COMPLETIONPENDING);
+  state_ = STATE_BINDING;
+
+  return true;
+}
+
+void UdpPacketSocket::OnBindCompleted(int result) {
+  DCHECK_EQ(state_, STATE_BINDING);
+
+  if (result == PP_OK) {
+    PP_NetAddress_Private address;
+    if (socket.GetBoundAddress(&address)) {
+      PpAddressToSocketAddress(address, &local_address_);
+    } else {
+      LOG(ERROR) << "Failed to get bind address for bound socket?";
+      error_ = EINVAL;
+      return;
+    }
+    state_ = STATE_BOUND;
+    SignalAddressReady(this, local_address_);
+    DoRead();
+    return;
+  }
+
+  if (current_port_ < max_port_) {
+    // Try to bind to the next available port.
+    ++current_port_;
+    PP_NetAddress_Private pp_local_address;
+    if (SocketAddressToPpAddressWithPort(local_address_, &pp_local_address,
+                                         current_port_)) {
+      int result = socket.Bind(&pp_local_address, PpCompletionCallback(
+          base::Bind(&UdpPacketSocket::OnBindCompleted,
+                     base::Unretained(this))));

[Wez, 2012/04/20 21:37:07]

Hmmm, I think we'll leak a base::Callback from PpCompletionCallback in the case
where |socket| gets torn down before Bind() has completed, and similarly for the
other uses of PpCompletionCallback here?

[Sergey Ulanov, 2012/04/23 22:16:51]

The callbacks must be invoked with PP_ERROR_ABORTED when we destroy the socket.
I added   Close() call in the destructor to make sure this happens while the
object is still alive.

[Wez, 2012/04/25 00:01:59]

If the caller deletes this object while the socket.Bind() is pending then won't
the PP_ERROR_ABORTED from the Close() call in the dtor cause a new socket.Bind()
to be dispatched, to the next port number, though?

[Sergey Ulanov, 2012/04/25 00:25:01]

More recent version of OnBindCompleted() handles this case properly.

+      DCHECK_EQ(result, PP_OK_COMPLETIONPENDING);
+    }
+  } else {
+    LOG(ERROR) << "Failed to bind UDP socket: " << result;
+  }
+}
+
+talk_base::SocketAddress UdpPacketSocket::GetLocalAddress() const {
+  return local_address_;
+}
+
+talk_base::SocketAddress UdpPacketSocket::GetRemoteAddress() const {
+  // UDP sockets are not connected - this method should never be called.
+  NOTREACHED();
+  return talk_base::SocketAddress();
+}
+
+int UdpPacketSocket::Send(const void* data, size_t data_size) {
+  // UDP sockets are not connected - this method should never be called.

[Wez, 2012/04/20 21:37:07]

They _can_ be connected, in which case this and GetRemoveAddress() are both
valid.

[Sergey Ulanov, 2012/04/23 22:16:51]

UDP instances of AsyncPacketSocket are never connected, so Send() method should
never be called for them.

+  NOTREACHED();
+  return EWOULDBLOCK;
+}
+
+int UdpPacketSocket::SendTo(const void* data,
+                            size_t data_size,
+                            const talk_base::SocketAddress& address) {
+  if (error_ != 0) {
+    return error_;
+  }
+
+  PP_NetAddress_Private pp_address;
+  if (!SocketAddressToPpAddress(address, &pp_address)) {
+    return EINVAL;
+  }
+
+  if (send_queue_size_ >= kMaxSendBufferSize) {
+    return EWOULDBLOCK;
+  }
+
+  send_queue_.push_back(PendingPacket(data, data_size, pp_address));
+  send_queue_size_ += data_size;
+  DoSend();
+  return data_size;
+}
+
+int UdpPacketSocket::Close() {
+  socket.Close();
+  state_ = STATE_CLOSED;
+  return 0;
+}
+
+talk_base::AsyncPacketSocket::State UdpPacketSocket::GetState() const {
+  return state_;
+}
+
+int UdpPacketSocket::GetOption(talk_base::Socket::Option opt, int* value) {
+  return -1;

[Wez, 2012/04/20 21:37:07]

nit: Add a // comment explaining this return code.

[Sergey Ulanov, 2012/04/23 22:16:51]

Done.

+}
+
+int UdpPacketSocket::SetOption(talk_base::Socket::Option opt, int value) {
+  return -1;
+}
+
+int UdpPacketSocket::GetError() const {
+  return error_;
+}
+
+void UdpPacketSocket::SetError(int error) {
+  error_ = error;
+}
+
+void UdpPacketSocket::DoSend() {
+  if (send_pending_ || send_queue_.empty())
+    return;
+
+  int result = socket.SendTo(
+      send_queue_.front().data->data(), send_queue_.front().data->size(),
+      &send_queue_.front().address,
+      PpCompletionCallback(base::Bind(&UdpPacketSocket::OnSendCompleted,
+                                      base::Unretained(this))));

[Wez, 2012/04/20 21:37:07]

Is the SendTo() interface actually defined to not be callable while a previous
SendTo() is still pending? Would it make sense to call SendTo() and handle the
in-progress return code rather than keeping the |send_pending_| flag?

[Sergey Ulanov, 2012/04/23 22:16:51]

Yes, SendTo() would return an error when another send is pending. Problem is
that with the current implementation it would require additional IPC roundtrip
to the browser and we cannot distinguish that error from other errors. Also this
callback is not optional, so we get PP_OK_COMPLETIONPENDING in either case, so
OnSendCompleted() would need to know what packet it's being called for.

[Wez, 2012/04/25 00:01:59]

Shouldn't you get PP_ERROR_IN_PROGRESS to the completion callback in the case
where the previous SendTo() hasn't completed?  So you'd need to base::Bind() the
packet to the completion callback and push it on to send_queue_ if you saw
PP_ERROR_IN_PROGRESS returned. Agreed that the extra IPC round-trip seems
undesirable.

[Sergey Ulanov, 2012/04/25 00:25:01]

I take this back - looks like the current UDP sockets implementation should
correctly return PP_ERROR_INPROGRESS.
In any case I still don't think that using it would make this code any cleaner.
Currently socket.SendTo() is called in only one place. With the approach you
describe it would have to be called in two places.
And we need to copy the data being sent anyway because the send buffer must be
alive while the SendTo() call is pending.

+  DCHECK_EQ(result, PP_OK_COMPLETIONPENDING);
+  send_pending_ = true;
+}
+
+void UdpPacketSocket::OnSendCompleted(int result) {
+  if (result < 0) {
+    LOG(ERROR) << "Send failed on a UDP socket: " << result;
+    error_ = EINVAL;
+    return;
+  }
+
+  send_pending_ = false;
+  send_queue_size_ -= send_queue_.front().data->size();
+  send_queue_.pop_front();
+  DoSend();
+}
+
+void UdpPacketSocket::DoRead() {
+  receive_buffer_.resize(kReceiveBufferSize);
+  int result = socket.RecvFrom(
+      &receive_buffer_[0], receive_buffer_.size(),
+      PpCompletionCallback(base::Bind(&UdpPacketSocket::OnReadCompleted,
+                                      base::Unretained(this))));
+  DCHECK_EQ(result, PP_OK_COMPLETIONPENDING);
+}
+
+void UdpPacketSocket::OnReadCompleted(int result) {
+  HandleReadResult(result);
+  if (result > 0) {
+    DoRead();
+  }
+}
+
+void UdpPacketSocket::HandleReadResult(int result) {
+  if (result > 0) {
+    PP_NetAddress_Private pp_address;
+    if (!socket.GetRecvFromAddress(&pp_address)) {
+      LOG(ERROR) << "GetRecvFromAddress() failed after successfull RecvFrom().";
+      return;
+    }
+    talk_base::SocketAddress address;
+    if (!PpAddressToSocketAddress(pp_address, &address)) {
+      LOG(ERROR) << "Failed to covert address received from RecvFrom().";
+      return;
+    }
+    SignalReadPacket(this, &receive_buffer_[0], result, address);

[Wez, 2012/04/20 21:37:07]

SignalReadPacket() could trigger the caller to tear us down, so do we need to
dispatch it asynchronously?

[Sergey Ulanov, 2012/04/23 22:16:51]

It should be safe to delete the socket from a callback.

[Wez, 2012/04/25 00:01:59]

Right.  So if calling code deletes us from within SignalReadPacket() in
HandleReadResult(), which in turn came from OnReadCompleted(), then we'll go on
to try to DoRead(), and crash?

[Sergey Ulanov, 2012/04/25 00:25:01]

ah, I see what you are talking about now. Actually signal<> handlers are not
allowed to delete the object that contain the fired signal<>. That's the reason
we always have to delete libjingle objects asynchronously. So we are safe here
:)

+  }
+}
+
+}  // namespace
+
+PepperPacketSocketFactory::PepperPacketSocketFactory(
+    const pp::InstanceHandle& instance)
+    : pp_instance_(instance) {
+}
+
+PepperPacketSocketFactory::~PepperPacketSocketFactory() {
+}
+
+talk_base::AsyncPacketSocket* PepperPacketSocketFactory::CreateUdpSocket(
+      const talk_base::SocketAddress& local_address,
+      int min_port,
+      int max_port) {
+  scoped_ptr<UdpPacketSocket> result(new UdpPacketSocket(pp_instance_));
+  if (!result->Init(local_address, min_port, max_port))
+    return NULL;
+  return result.release();
+}
+
+talk_base::AsyncPacketSocket* PepperPacketSocketFactory::CreateServerTcpSocket(
+    const talk_base::SocketAddress& local_address,
+    int min_port,
+    int max_port,
+    bool ssl) {
+  // Don't use TCP sockets for remoting connections.

[Wez, 2012/04/20 21:37:07]

NOTREACHED()?

[Sergey Ulanov, 2012/04/23 22:16:51]

Done.

+  return NULL;
+}
+
+talk_base::AsyncPacketSocket* PepperPacketSocketFactory::CreateClientTcpSocket(
+      const talk_base::SocketAddress& local_address,
+      const talk_base::SocketAddress& remote_address,
+      const talk_base::ProxyInfo& proxy_info,
+      const std::string& user_agent,
+      bool ssl) {
+  // Don't use TCP sockets for remoting connections.
+  return NULL;
+}
+
+}  // namespace remoting