[net/udp] Create UDPSocketWin::Core which persists until all network operations complete.

net/udp/udp_socket_win.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 "net/udp/udp_socket_win.h"
 
 #include <mstcpip.h>
 
 #include "base/callback.h"
 #include "base/eintr_wrapper.h"
@@ skipping 13 matching lines @@
 namespace {
 
 static const int kBindRetries = 10;
 static const int kPortStart = 1024;
 static const int kPortEnd = 65535;
 
 }  // namespace net
 
 namespace net {
 
-void UDPSocketWin::ReadDelegate::OnObjectSignaled(HANDLE object) {
-  DCHECK_EQ(object, socket_->read_overlapped_.hEvent);
-  socket_->DidCompleteRead();
+// This class encapsulates all the state that has to be preserved as long as
+// there is a network IO operation in progress. If the owner UDPSocketWin
+// is destroyed while an operation is in progress, the Core is detached and it
+// lives until the operation completes and the OS doesn't reference any resource
+// declared on this class anymore.
+class UDPSocketWin::Core : public base::RefCounted<Core> {
+ public:
+  explicit Core(UDPSocketWin* socket);
+
+  // Start watching for the end of a read or write operation.
+  void WatchForRead();
+  void WatchForWrite();
+
+  // The UDPSocketWin is going away.
+  void Detach() { socket_ = NULL; }
+
+  // The separate OVERLAPPED variables for asynchronous operation.
+  OVERLAPPED read_overlapped_;
+  OVERLAPPED write_overlapped_;
+
+  // The buffers used in Read() and Write().
+  WSABUF read_buffer_;
+  WSABUF write_buffer_;

[wtc, 2012/09/12 01:40:29]

It turns out these two buffers don't need to be members of
Core. They can continue to be stack variables.

From the WSARecvFrom MSDN page:
http://msdn.microsoft.com/en-us/library/windows/desktop/ms741686%28v=vs.85%29...

  If this function is completed in an overlapped manner,
  it is the Winsock service provider's responsibility to
  capture the WSABUF structures before returning from this
  call. This enables applications to build stack-based
  WSABUF arrays pointed to by the lpBuffers parameter.

You don't need to make this change in this CL. You can
file a bug report and assign it to me or rvargas.

[wtc, 2012/09/12 04:00:43]

I have created a CL for TCPClientSocketWin:
http://codereview.chromium.org/10913212/

The WSASend and WSASendTo MSDN pages actually say something
contradictory, which I believe is wrong:
http://msdn.microsoft.com/en-us/library/windows/desktop/ms742203%28v=vs.85%29...

lpBuffers [in]

    A pointer to an array of WSABUF structures. Each WSABUF structure contains a
pointer to a buffer and the length, in bytes, of the buffer. For a Winsock
application, once the WSASend function is called, the system owns these buffers
and the application may not access them. This array must remain valid for the
duration of the send operation.

[szym, 2012/09/12 15:14:06]

That snippet from MSDN was the reason why I decided to put WSABUF in Core.
However, the relevant snippet about stack-based WSABUF arrays is present in both
WSARecvFrom and WSASendTo.

+  scoped_refptr<IOBuffer> read_iobuffer_;
+  scoped_refptr<IOBuffer> write_iobuffer_;
+
+ private:
+  friend class base::RefCounted<Core>;
+
+  class ReadDelegate : public base::win::ObjectWatcher::Delegate {
+   public:
+    explicit ReadDelegate(Core* core) : core_(core) {}
+    virtual ~ReadDelegate() {}
+
+    // base::ObjectWatcher::Delegate methods:
+    virtual void OnObjectSignaled(HANDLE object);
+
+   private:
+    Core* const core_;
+  };
+
+  class WriteDelegate : public base::win::ObjectWatcher::Delegate {
+   public:
+    explicit WriteDelegate(Core* core) : core_(core) {}
+    virtual ~WriteDelegate() {}
+
+    // base::ObjectWatcher::Delegate methods:
+    virtual void OnObjectSignaled(HANDLE object);
+
+   private:
+    Core* const core_;
+  };
+
+  ~Core();
+
+  // The socket that created this object.
+  UDPSocketWin* socket_;
+
+  // |reader_| handles the signals from |read_watcher_|.
+  ReadDelegate reader_;
+  // |writer_| handles the signals from |write_watcher_|.
+  WriteDelegate writer_;
+
+  // |read_watcher_| watches for events from Connect() and Read().

[wtc, 2012/09/12 01:40:29]

Delete "Connect() and"?

[szym, 2012/09/12 15:14:06]

Done.

+  base::win::ObjectWatcher read_watcher_;
+  // |write_watcher_| watches for events from Write();
+  base::win::ObjectWatcher write_watcher_;
+
+  DISALLOW_COPY_AND_ASSIGN(Core);
+};
+
+UDPSocketWin::Core::Core(UDPSocketWin* socket)
+    : socket_(socket),
+      ALLOW_THIS_IN_INITIALIZER_LIST(reader_(this)),
+      ALLOW_THIS_IN_INITIALIZER_LIST(writer_(this)) {
+  memset(&read_overlapped_, 0, sizeof(read_overlapped_));
+  memset(&write_overlapped_, 0, sizeof(write_overlapped_));
+
+  read_overlapped_.hEvent = WSACreateEvent();
+  write_overlapped_.hEvent = WSACreateEvent();
 }
 
-void UDPSocketWin::WriteDelegate::OnObjectSignaled(HANDLE object) {
-  DCHECK_EQ(object, socket_->write_overlapped_.hEvent);
-  socket_->DidCompleteWrite();
+UDPSocketWin::Core::~Core() {
+  // Make sure the message loop is not watching this object anymore.
+  read_watcher_.StopWatching();
+  write_watcher_.StopWatching();
+
+  WSACloseEvent(read_overlapped_.hEvent);
+  memset(&read_overlapped_, 0xaf, sizeof(read_overlapped_));
+  WSACloseEvent(write_overlapped_.hEvent);
+  memset(&write_overlapped_, 0xaf, sizeof(write_overlapped_));
 }
 
+void UDPSocketWin::Core::WatchForRead() {
+  // We grab an extra reference because there is an IO operation in progress.
+  // Balanced in ReadDelegate::OnObjectSignaled().
+  AddRef();
+  read_watcher_.StartWatching(read_overlapped_.hEvent, &reader_);
+}
+
+void UDPSocketWin::Core::WatchForWrite() {
+  // We grab an extra reference because there is an IO operation in progress.
+  // Balanced in WriteDelegate::OnObjectSignaled().
+  AddRef();
+  write_watcher_.StartWatching(write_overlapped_.hEvent, &writer_);
+}
+
+void UDPSocketWin::Core::ReadDelegate::OnObjectSignaled(HANDLE object) {
+  DCHECK_EQ(object, core_->read_overlapped_.hEvent);
+  if (core_->socket_)
+    core_->socket_->DidCompleteRead();
+
+  core_->Release();
+}
+
+void UDPSocketWin::Core::WriteDelegate::OnObjectSignaled(HANDLE object) {
+  DCHECK_EQ(object, core_->write_overlapped_.hEvent);
+  if (core_->socket_)
+    core_->socket_->DidCompleteWrite();
+
+  core_->Release();
+}
+
+//-----------------------------------------------------------------------------
+
 UDPSocketWin::UDPSocketWin(DatagramSocket::BindType bind_type,
                            const RandIntCallback& rand_int_cb,
                            net::NetLog* net_log,
                            const net::NetLog::Source& source)
     : socket_(INVALID_SOCKET),
       socket_options_(0),
       bind_type_(bind_type),
       rand_int_cb_(rand_int_cb),
-      ALLOW_THIS_IN_INITIALIZER_LIST(read_delegate_(this)),
-      ALLOW_THIS_IN_INITIALIZER_LIST(write_delegate_(this)),
       recv_from_address_(NULL),
       net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_UDP_SOCKET)) {
   EnsureWinsockInit();
   net_log_.BeginEvent(NetLog::TYPE_SOCKET_ALIVE,
                       source.ToEventParametersCallback());
-  memset(&read_overlapped_, 0, sizeof(read_overlapped_));
-  read_overlapped_.hEvent = WSACreateEvent();
-  memset(&write_overlapped_, 0, sizeof(write_overlapped_));
-  write_overlapped_.hEvent = WSACreateEvent();
   if (bind_type == DatagramSocket::RANDOM_BIND)
     DCHECK(!rand_int_cb.is_null());
 }
 
 UDPSocketWin::~UDPSocketWin() {
   Close();
   net_log_.EndEvent(NetLog::TYPE_SOCKET_ALIVE);
 }
 
 void UDPSocketWin::Close() {
   DCHECK(CalledOnValidThread());
 
   if (!is_connected())
     return;
 
   // Zero out any pending read/write callback state.
   read_callback_.Reset();
   recv_from_address_ = NULL;
   write_callback_.Reset();
 
-  read_watcher_.StopWatching();
-  write_watcher_.StopWatching();
-
-  WSACloseEvent(read_overlapped_.hEvent);
-  memset(&read_overlapped_, 0xaf, sizeof(read_overlapped_));
-  WSACloseEvent(write_overlapped_.hEvent);
-  memset(&write_overlapped_, 0xaf, sizeof(write_overlapped_));
-
   closesocket(socket_);
   socket_ = INVALID_SOCKET;
+
+  if (core_.get())

[wtc, 2012/09/12 01:40:29]

Is this null pointer check necessary? It is not in
TCPClientSocketWin::DoDisconnect().

I think this check at the beginning of this function
  if (!is_connected())
    return;
should provide protection.

[szym, 2012/09/12 02:23:45]

It should, I added this in response to a failing test, but I didn't realize that
the socket is guaranteed to be either Connected or Bound, so the proper place to
create the Core is either in CreateSocket or in UDPSocketWin constructor.

I'll fix that.

+    core_->Detach();
+  core_ = NULL;
 }
 
 int UDPSocketWin::GetPeerAddress(IPEndPoint* address) const {
   DCHECK(CalledOnValidThread());
   DCHECK(address);
   if (!is_connected())
     return ERR_SOCKET_NOT_CONNECTED;
 
   // TODO(szym): Simplify. http://crbug.com/126152
   if (!remote_address_.get()) {
@@ skipping 45 matching lines @@
   DCHECK_NE(INVALID_SOCKET, socket_);
   DCHECK(read_callback_.is_null());
   DCHECK(!recv_from_address_);
   DCHECK(!callback.is_null());  // Synchronous operation not supported.
   DCHECK_GT(buf_len, 0);
 
   int nread = InternalRecvFrom(buf, buf_len, address);
   if (nread != ERR_IO_PENDING)
     return nread;
 
-  read_iobuffer_ = buf;
   read_callback_ = callback;
   recv_from_address_ = address;
   return ERR_IO_PENDING;
 }
 
 int UDPSocketWin::Write(IOBuffer* buf,
                         int buf_len,
                         const CompletionCallback& callback) {
   return SendToOrWrite(buf, buf_len, NULL, callback);
 }
@@ skipping 15 matching lines @@
   DCHECK(!callback.is_null());  // Synchronous operation not supported.
   DCHECK_GT(buf_len, 0);
   DCHECK(!send_to_address_.get());
 
   int nwrite = InternalSendTo(buf, buf_len, address);
   if (nwrite != ERR_IO_PENDING)
     return nwrite;
 
   if (address)
     send_to_address_.reset(new IPEndPoint(*address));
-  write_iobuffer_ = buf;
   write_callback_ = callback;
   return ERR_IO_PENDING;
 }
 
 int UDPSocketWin::Connect(const IPEndPoint& address) {
   net_log_.BeginEvent(NetLog::TYPE_UDP_CONNECT,
                       CreateNetLogUDPConnectCallback(&address));
   int rv = InternalConnect(address);
   net_log_.EndEventWithNetErrorCode(NetLog::TYPE_UDP_CONNECT, rv);
   return rv;
@@ skipping 93 matching lines @@
   DCHECK(!write_callback_.is_null());
 
   // since Run may result in Write being called, clear write_callback_ up front.
   CompletionCallback c = write_callback_;
   write_callback_.Reset();
   c.Run(rv);
 }
 
 void UDPSocketWin::DidCompleteRead() {
   DWORD num_bytes, flags;
-  BOOL ok = WSAGetOverlappedResult(socket_, &read_overlapped_,
+  BOOL ok = WSAGetOverlappedResult(socket_, &core_->read_overlapped_,
                                    &num_bytes, FALSE, &flags);
-  WSAResetEvent(read_overlapped_.hEvent);
+  WSAResetEvent(core_->read_overlapped_.hEvent);
   int result = ok ? num_bytes : MapSystemError(WSAGetLastError());
   // Convert address.
   if (recv_from_address_ && result >= 0) {
     if (!ReceiveAddressToIPEndpoint(recv_from_address_))
       result = ERR_FAILED;
   }
-  LogRead(result, read_iobuffer_->data());
-  read_iobuffer_ = NULL;
+  LogRead(result, core_->read_iobuffer_->data());
+  core_->read_iobuffer_ = NULL;
   recv_from_address_ = NULL;
   DoReadCallback(result);
 }
 
 void UDPSocketWin::LogRead(int result, const char* bytes) const {
   if (result < 0) {
     net_log_.AddEventWithNetErrorCode(NetLog::TYPE_UDP_RECEIVE_ERROR, result);
     return;
   }
 
   if (net_log_.IsLoggingAllEvents()) {
     // Get address for logging, if |address| is NULL.
     IPEndPoint address;
     bool is_address_valid = ReceiveAddressToIPEndpoint(&address);
     net_log_.AddEvent(
         NetLog::TYPE_UDP_BYTES_RECEIVED,
         CreateNetLogUDPDataTranferCallback(
             result, bytes,
             is_address_valid ? &address : NULL));
   }
 
   base::StatsCounter read_bytes("udp.read_bytes");
   read_bytes.Add(result);
 }
 
 void UDPSocketWin::DidCompleteWrite() {
   DWORD num_bytes, flags;
-  BOOL ok = WSAGetOverlappedResult(socket_, &write_overlapped_,
+  BOOL ok = WSAGetOverlappedResult(socket_, &core_->write_overlapped_,
                                    &num_bytes, FALSE, &flags);
-  WSAResetEvent(write_overlapped_.hEvent);
+  WSAResetEvent(core_->write_overlapped_.hEvent);
   int result = ok ? num_bytes : MapSystemError(WSAGetLastError());
-  LogWrite(result, write_iobuffer_->data(), send_to_address_.get());
+  LogWrite(result, core_->write_iobuffer_->data(), send_to_address_.get());
 
   send_to_address_.reset();
-  write_iobuffer_ = NULL;
+  core_->write_iobuffer_ = NULL;
   DoWriteCallback(result);
 }
 
 void UDPSocketWin::LogWrite(int result,
                             const char* bytes,
                             const IPEndPoint* address) const {
   if (result < 0) {
     net_log_.AddEventWithNetErrorCode(NetLog::TYPE_UDP_SEND_ERROR, result);
     return;
   }
 
   if (net_log_.IsLoggingAllEvents()) {
     net_log_.AddEvent(
         NetLog::TYPE_UDP_BYTES_SENT,
         CreateNetLogUDPDataTranferCallback(result, bytes, address));
   }
 
   base::StatsCounter write_bytes("udp.write_bytes");
   write_bytes.Add(result);
 }
 
 int UDPSocketWin::InternalRecvFrom(IOBuffer* buf, int buf_len,
                                    IPEndPoint* address) {

[wtc, 2012/09/12 01:40:29]

I suggest copying
  DCHECK(!core_->read_iobuffer_);
from TCPClientSocketWin::Read().

[szym, 2012/09/12 15:14:06]

Done.

   recv_addr_len_ = sizeof(recv_addr_storage_);
   struct sockaddr* addr =
       reinterpret_cast<struct sockaddr*>(&recv_addr_storage_);
 
-  WSABUF read_buffer;
-  read_buffer.buf = buf->data();
-  read_buffer.len = buf_len;
+  if (!core_)
+    core_ = new Core(this);

[wtc, 2012/09/12 01:40:29]

I think we can simply create Core in the UDPSocketWin constructor.

rvargas: do you remember why you didn't create Core in the
TCPClientSocketWin constructor?

[szym, 2012/09/12 02:23:45]

This will definitely be moved from here, either to the constructor or to
CreateSocket. Waiting for opinion from rvargas.

[wtc, 2012/09/12 04:00:43]

I think TCPClientSocketWin creates Core lazily because
it needs to continue to use the object after Disconnect()
has been called. It is allowed to call Connect() again
after Disconnect().

If UDPSocketWin becomes "dead" after Close() is called,
then it is fine to create Core in the constructor or
CreateSocket(). Otherwise, it is best to create Core
lazily.

[szym, 2012/09/12 15:14:06]

It seems UDOSocketWin can be re-Connected or re-Bound after it's Closed. I moved
Core creation to CreateSocket.

+  core_->read_iobuffer_ = buf;

[wtc, 2012/09/12 01:40:29]

BUG: this can be a reference leak. You need to release this
reference if WSARecvFrom completes synchronously.

I suggest you copy TCPClientSocketWin::Read() and only do
  core_->read_iobuffer_ = buf;
before returning ERR_IO_PENDING.

[szym, 2012/09/12 15:14:06]

Done.

+  core_->read_buffer_.buf = buf->data();
+  core_->read_buffer_.len = buf_len;
 
   DWORD flags = 0;
   DWORD num;
   CHECK_NE(INVALID_SOCKET, socket_);
-  AssertEventNotSignaled(read_overlapped_.hEvent);
-  int rv = WSARecvFrom(socket_, &read_buffer, 1, &num, &flags, addr,
-                       &recv_addr_len_, &read_overlapped_, NULL);
+  AssertEventNotSignaled(core_->read_overlapped_.hEvent);
+  int rv = WSARecvFrom(socket_, &core_->read_buffer_, 1, &num, &flags, addr,
+                       &recv_addr_len_, &core_->read_overlapped_, NULL);
   if (rv == 0) {
-    if (ResetEventIfSignaled(read_overlapped_.hEvent)) {
+    if (ResetEventIfSignaled(core_->read_overlapped_.hEvent)) {
       int result = num;
       // Convert address.
       if (address && result >= 0) {
         if (!ReceiveAddressToIPEndpoint(address))
           result = ERR_FAILED;
       }
       LogRead(result, buf->data());
       return result;
     }
   } else {
     int os_error = WSAGetLastError();
     if (os_error != WSA_IO_PENDING) {
       int result = MapSystemError(os_error);
       LogRead(result, NULL);
       return result;
     }
   }
-  read_watcher_.StartWatching(read_overlapped_.hEvent, &read_delegate_);
+  core_->WatchForRead();
   return ERR_IO_PENDING;
 }
 
 int UDPSocketWin::InternalSendTo(IOBuffer* buf, int buf_len,

[wtc, 2012/09/12 01:40:29]

IMPORTANT: All of my comments for UDPSocketWin::InternalRecvFrom()
also apply to this method.

[szym, 2012/09/12 15:14:06]

Done.

                                  const IPEndPoint* address) {
   SockaddrStorage storage;
   struct sockaddr* addr = storage.addr;
   // Convert address.
   if (!address) {
     addr = NULL;
     storage.addr_len = 0;
   } else {
     if (!address->ToSockAddr(addr, &storage.addr_len)) {
       int result = ERR_FAILED;
       LogWrite(result, NULL, NULL);
       return result;
     }
   }
 
-  WSABUF write_buffer;
-  write_buffer.buf = buf->data();
-  write_buffer.len = buf_len;
+  if (!core_.get())
+    core_ = new Core(this);
+  core_->write_iobuffer_ = buf;
+  core_->write_buffer_.buf = buf->data();
+  core_->write_buffer_.len = buf_len;
 
   DWORD flags = 0;
   DWORD num;
-  AssertEventNotSignaled(write_overlapped_.hEvent);
-  int rv = WSASendTo(socket_, &write_buffer, 1, &num, flags,
-                     addr, storage.addr_len, &write_overlapped_, NULL);
+  AssertEventNotSignaled(core_->write_overlapped_.hEvent);
+  int rv = WSASendTo(socket_, &core_->write_buffer_, 1, &num, flags,
+                     addr, storage.addr_len, &core_->write_overlapped_, NULL);
   if (rv == 0) {
-    if (ResetEventIfSignaled(write_overlapped_.hEvent)) {
+    if (ResetEventIfSignaled(core_->write_overlapped_.hEvent)) {
       int result = num;
       LogWrite(result, buf->data(), address);
       return result;
     }
   } else {
     int os_error = WSAGetLastError();
     if (os_error != WSA_IO_PENDING) {
       int result = MapSystemError(os_error);
       LogWrite(result, NULL, NULL);
       return result;
     }
   }
 
-  write_watcher_.StartWatching(write_overlapped_.hEvent, &write_delegate_);
+  core_->WatchForWrite();
   return ERR_IO_PENDING;
 }
 
 int UDPSocketWin::SetSocketOptions() {
   BOOL true_value = 1;
   if (socket_options_ & SOCKET_OPTION_REUSE_ADDRESS) {
     int rv = setsockopt(socket_, SOL_SOCKET, SO_REUSEADDR,
                         reinterpret_cast<const char*>(&true_value),
                         sizeof(true_value));
     if (rv < 0)
@@ skipping 31 matching lines @@
   return DoBind(IPEndPoint(ip, 0));
 }
 
 bool UDPSocketWin::ReceiveAddressToIPEndpoint(IPEndPoint* address) const {
   const struct sockaddr* addr =
       reinterpret_cast<const struct sockaddr*>(&recv_addr_storage_);
   return address->FromSockAddr(addr, recv_addr_len_);
 }
 
 }  // namespace net