P2P Sockets host implementation.

content/browser/renderer_host/p2p_sockets_host.cc

Index: content/browser/renderer_host/p2p_sockets_host.cc
diff --git a/content/browser/renderer_host/p2p_sockets_host.cc b/content/browser/renderer_host/p2p_sockets_host.cc
new file mode 100644
index 0000000000000000000000000000000000000000..83771e926cb1d7e4c8aa0ab44dd1ee68f60dcc80
--- /dev/null
+++ b/content/browser/renderer_host/p2p_sockets_host.cc
@@ -0,0 +1,385 @@
+// Copyright (c) 2011 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/browser/renderer_host/p2p_sockets_host.h"
+
+#include "chrome/common/render_messages.h"
+#include "net/base/net_util.h"
+
+// Currently P2P Sockets are not implemented for Windows yet.
+// TODO(sergeyu) implement P2PSocket on Windows.
+#if !defined(OS_WIN)
+
+#include <errno.h>
+#include <net/if.h>
+#include <netinet/in.h>
+#include <sys/ioctl.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <sys/utsname.h>
+#include <unistd.h>
+
+namespace {
+
+// This method returns address of the first IPv4 enabled network
+// interface it finds ignoring the loopback interface. This address is
+// used for all sockets.
+//
+// TODO(sergeyu): This approach works only in the simplest case when
+// host has only one network connection. Instead of binding all
+// connections to this interface we must provide list of interfaces to
+// the renderer, and let the PortAllocater in the renderer process
+// choose local address.
+bool GetLocalAddress(sockaddr_in* addr) {
+  int fd;
+  if ((fd = ::socket(AF_INET, SOCK_DGRAM, 0)) < 0) {
+    LOG(ERROR) << "socket() failed: " << fd;
+    return false;
+  }
+
+  struct ifconf ifc;
+  ifc.ifc_len = 64 * sizeof(struct ifreq);
+  scoped_array<char> ifc_buffer(new char[ifc.ifc_len]);
+  ifc.ifc_buf = ifc_buffer.get();
+
+  int result = ::ioctl(fd, SIOCGIFCONF, &ifc);
+  if (result < 0) {
+    LOG(ERROR) << "GetLocalAddress: ioctl returned error:" << result;
+    ::close(fd);

[awong, 2011/03/01 19:57:53]

Do we need to explicilty specify the global scope here?  Does other code that
works with sockets in chromium do this?

[Sergey Ulanov, 2011/03/01 22:34:17]

No we don't need it, I just though it looks better this way. Removed it because
other code in src/net doesn't specify scope for sockets API.

+    return false;
+  }
+  CHECK_LT(ifc.ifc_len, static_cast<int>(64 * sizeof(struct ifreq)));
+
+  struct ifreq* ptr = reinterpret_cast<struct ifreq*>(ifc.ifc_buf);
+  struct ifreq* end =
+      reinterpret_cast<struct ifreq*>(ifc.ifc_buf + ifc.ifc_len);
+
+  bool found = false;
+  while (ptr < end) {
+    struct sockaddr_in* inaddr =
+        reinterpret_cast<struct sockaddr_in*>(&ptr->ifr_ifru.ifru_addr);
+    if (inaddr->sin_family == AF_INET &&
+        strncmp(ptr->ifr_name, "lo", 2) != 0) {
+      memcpy(addr, inaddr, sizeof(sockaddr_in));
+      found = true;
+      break;
+    }
+
+    ptr++;
+  }
+
+  ::close(fd);
+
+  return found;
+}
+
+bool SocketAddressToSockAddr(P2PSocketAddress address, sockaddr_in* addr) {
+  // TODO(sergeyu): Add IPv6 support.
+  if (address.address.size() != 4) {
+    return false;
+  }
+
+  addr->sin_family = AF_INET;
+  memcpy(&addr->sin_addr, &address.address[0], 4);
+  addr->sin_port = htons(address.port);
+  return true;
+}
+
+bool SockAddrToSocketAddress(sockaddr_in* addr,
+                             P2PSocketAddress* address) {
+  if (addr->sin_family != AF_INET) {
+    LOG(ERROR) << "SockAddrToSocketAddress: only IPv4 addresses are supported";
+    // TODO(sergeyu): Add IPv6 support.
+    return false;
+  }
+
+  address->address.resize(4);
+  memcpy(&address->address[0], &addr->sin_addr, 4);
+  address->port = ntohs(addr->sin_port);
+
+  return true;
+}
+
+}  // namespace
+
+class P2PSocketsHost::P2PSocket {

[jam, 2011/03/01 19:35:54]

i'm curious, why isn't this class defined in a separate file?  the usual style
is one class per file.  also, what about tests?

[Sergey Ulanov, 2011/03/01 22:34:17]

Done.
I will do add unittests in a separate CL.

+ public:
+  P2PSocket(P2PSocketsHost* host, int routing_id, int id);
+  ~P2PSocket();
+
+  bool Init();
+
+  void Send(P2PSocketAddress socket_address, const std::vector<char>& data);
+  void DestroyConnection(P2PSocketAddress socket_address);
+
+ private:
+  enum State {
+    STATE_UNINITIALIZED,
+    STATE_OPEN,
+    STATE_ERROR,
+  };
+
+  class ReadWatcher : public MessageLoopForIO::Watcher {
+   public:
+    explicit ReadWatcher(P2PSocketsHost::P2PSocket* socket) : socket_(socket) {}
+
+    // MessageLoopForIO::Watcher methods
+    virtual void OnFileCanReadWithoutBlocking(int /* fd */) {
+      socket_->DidCompleteRead();
+    }
+    virtual void OnFileCanWriteWithoutBlocking(int /* fd */) {}
+
+   private:
+    P2PSocketsHost::P2PSocket* socket_;
+
+    DISALLOW_COPY_AND_ASSIGN(ReadWatcher);
+  };
+
+  void DidCompleteRead();
+  void OnError();
+
+  P2PSocketsHost* host_;
+  int routing_id_;
+  int id_;
+  State state_;
+  int socket_;
+  MessageLoopForIO::FileDescriptorWatcher read_socket_watcher_;
+  ReadWatcher read_watcher_;
+
+  DISALLOW_COPY_AND_ASSIGN(P2PSocket);
+};
+
+P2PSocketsHost::P2PSocket::P2PSocket(
+    P2PSocketsHost* host, int routing_id, int id)
+    : host_(host), routing_id_(routing_id), id_(id),
+      state_(STATE_UNINITIALIZED), socket_(0), read_watcher_(this) {
+}
+
+P2PSocketsHost::P2PSocket::~P2PSocket() {
+  if (state_ == STATE_OPEN) {
+    DCHECK_NE(socket_, 0);
+    ::close(socket_);
+  }
+}
+
+bool P2PSocketsHost::P2PSocket::Init() {
+  socket_ = socket(AF_INET, SOCK_DGRAM, 0);
+  if (socket_ < 0) {
+    LOG(ERROR) << "Failed to create socket: " << socket_;
+    OnError();
+    return false;
+  }
+
+  int result = net::SetNonBlocking(socket_);
+  if (result < 0) {
+    LOG(ERROR) << "Failed to set O_NONBLOCK flag: " << result;
+    OnError();
+    return false;
+  }
+
+  sockaddr_in addr;
+  if (!GetLocalAddress(&addr)) {
+    LOG(ERROR) << "Failed to get local network address.";
+    OnError();
+    return false;
+  }
+
+  result = ::bind(socket_, reinterpret_cast<sockaddr*>(&addr), sizeof(addr));
+  if (result < 0) {
+    LOG(ERROR) << "bind() failed: " << result;
+    OnError();
+    return false;
+  }
+
+  if (!MessageLoopForIO::current()->WatchFileDescriptor(
+          socket_, true, MessageLoopForIO::WATCH_READ,
+          &read_socket_watcher_, &read_watcher_)) {
+    LOG(ERROR) << "WatchFileDescriptor failed on read, errno: " << errno;
+    OnError();
+    return false;
+  }
+
+  socklen_t addrlen = sizeof(addr);
+  result = ::getsockname(socket_, reinterpret_cast<sockaddr*>(&addr),
+                         &addrlen);
+  if (result < 0) {
+    LOG(ERROR) << "P2PSocket::Init(): unable to get local addr: "
+               << result;
+    OnError();
+    return false;
+  }
+
+  P2PSocketAddress address;
+  if (!SockAddrToSocketAddress(&addr, &address)) {
+    OnError();
+    return false;
+  }
+
+  VLOG(1) << "getsockname() returned "
+          << net::NetAddressToString(
+              reinterpret_cast<sockaddr*>(&addr), sizeof(addr))
+          << ":" << address.port;
+
+  state_ = STATE_OPEN;
+
+  host_->Send(new ViewMsg_P2P_OnSocketCreated(routing_id_, id_, address));
+
+  return true;
+}
+
+void P2PSocketsHost::P2PSocket::OnError() {
+  if (socket_ != 0) {
+    ::close(socket_);
+    socket_ = 0;
+  }
+
+  if (state_ == STATE_UNINITIALIZED || state_ == STATE_OPEN) {
+    host_->Send(new ViewMsg_P2P_OnError(routing_id_, id_));
+  }
+
+  state_ = STATE_ERROR;
+}
+
+void P2PSocketsHost::P2PSocket::DidCompleteRead() {
+  if (state_ != STATE_OPEN) {
+    return;
+  }
+
+  std::vector<char> data;
+  data.resize(4096);
+  sockaddr_in addr;
+  socklen_t addr_len = sizeof(addr);
+  int result = ::recvfrom(socket_, &data[0], data.size(), 0,
+                          reinterpret_cast<sockaddr*>(&addr), &addr_len);
+  if (result > 0) {
+    data.resize(result);
+    VLOG(2) << "received " << result << " bytes from "
+            << net::NetAddressToString(
+                reinterpret_cast<sockaddr*>(&addr), sizeof(addr))
+            << ":" << ntohs(addr.sin_port);
+    P2PSocketAddress address;
+    if (!SockAddrToSocketAddress(&addr, &address)) {
+      // Address conversion fails only if we receive a non-IPv4
+      // packet, which should never happen because the socket is IPv4.
+      NOTREACHED();
+      return;
+    }
+
+    host_->Send(new ViewMsg_P2P_OnDataReceived(routing_id_, id_,
+                                               address, data));
+  } else if (result < 0) {
+    LOG(ERROR) << "recvfrom() returned error: " << result;
+    OnError();
+  }
+}
+
+void P2PSocketsHost::P2PSocket::Send(P2PSocketAddress socket_address,
+                                     const std::vector<char>& data) {
+  sockaddr_in addr;
+  SocketAddressToSockAddr(socket_address, &addr);
+  int result = sendto(socket_, &data[0], data.size(), 0,
+                      reinterpret_cast<sockaddr*>(&addr), sizeof(addr));
+  if (result < 0) {
+    LOG(ERROR) << "Send failed.";
+  } else {
+    VLOG(2) << "Sent " << result << " bytes to "
+            << net::NetAddressToString(
+                reinterpret_cast<sockaddr*>(&addr), sizeof(addr))
+            << ":" <<  ntohs(addr.sin_port);
+  }
+}
+
+#else  // defined(OS_WIN)
+
+class P2PSocketsHost::P2PSocket {
+ public:
+  P2PSocket(P2PSocketsHost* host, int routing_id, int id)
+    : host_(host), routing_id_(routing_id), id_(id) {
+  }
+  ~P2PSocket() { }
+
+  bool Init() {
+    host_->Send(new ViewMsg_P2P_OnError(routing_id_, id_));
+    return false;
+  }
+
+  void Send(P2PSocketAddress socket_address, const std::vector<char>& data) { }
+  void DestroyConnection(P2PSocketAddress socket_address) { }
+
+ private:
+  P2PSocketsHost* host_;
+  int routing_id_;
+  int id_;
+
+  DISALLOW_COPY_AND_ASSIGN(P2PSocket);
+};
+
+#endif  // defined(OS_WIN)
+
+P2PSocketsHost::P2PSocketsHost() {
+}
+
+P2PSocketsHost::~P2PSocketsHost() {
+}
+
+void P2PSocketsHost::OnChannelClosing() {
+  BrowserMessageFilter::OnChannelClosing();
+
+  // Since the IPC channel is gone, close pending connections.
+  for (IDMap<P2PSocket>::iterator i(&sockets_); !i.IsAtEnd(); i.Advance()) {
+    sockets_.Remove(i.GetCurrentKey());
+  }
+}
+
+void P2PSocketsHost::OnDestruct() const {
+  BrowserThread::DeleteOnIOThread::Destruct(this);
+}
+
+bool P2PSocketsHost::OnMessageReceived(const IPC::Message& message,
+                                        bool* message_was_ok) {

[jam, 2011/03/01 19:35:54]

nit: spacing

[Sergey Ulanov, 2011/03/01 22:34:17]

Done.

+  bool handled = true;
+  IPC_BEGIN_MESSAGE_MAP_EX(P2PSocketsHost, message, *message_was_ok)
+    IPC_MESSAGE_HANDLER(ViewHostMsg_P2P_CreateSocket, OnCreateSocket)
+    IPC_MESSAGE_HANDLER(ViewHostMsg_P2P_Send, OnSend)
+    IPC_MESSAGE_HANDLER(ViewHostMsg_P2P_DestroySocket, OnDestroySocket)
+    IPC_MESSAGE_UNHANDLED(handled = false)
+  IPC_END_MESSAGE_MAP_EX()
+  return handled;
+}
+
+void P2PSocketsHost::OnCreateSocket(
+    const IPC::Message& msg, P2PSocketType type, int socket_id,
+    P2PSocketAddress remote_address) {
+  if (sockets_.Lookup(socket_id)) {
+    LOG(ERROR) << "Received ViewHostMsg_P2P_CreateSocket for socket "
+        "that already exists.";
+    return;
+  }
+
+  if (type != P2P_SOCKET_UDP) {
+    Send(new ViewMsg_P2P_OnError(msg.routing_id(), socket_id));
+    return;
+  }
+
+  scoped_ptr<P2PSocket> socket(
+      new P2PSocket(this, msg.routing_id(), socket_id));
+  if (socket->Init()) {
+    sockets_.AddWithID(socket.release(), socket_id);
+  }
+}
+
+void P2PSocketsHost::OnSend(const IPC::Message& msg, int socket_id,
+                             P2PSocketAddress socket_address,

[jam, 2011/03/01 19:35:54]

nit: spacing

[Sergey Ulanov, 2011/03/01 22:34:17]

Done.

+                             const std::vector<char>& data) {
+  P2PSocket* socket = sockets_.Lookup(socket_id);
+  if (!socket) {
+    LOG(ERROR) << "Received ViewHostMsg_P2P_Send for invalid socket_id.";
+    return;
+  }
+  socket->Send(socket_address, data);
+}
+
+void P2PSocketsHost::OnDestroySocket(const IPC::Message& msg, int socket_id) {
+  sockets_.Remove(socket_id);
+}