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

net/udp/udp_socket_win.cc

Index: net/udp/udp_socket_win.cc
diff --git a/net/udp/udp_socket_win.cc b/net/udp/udp_socket_win.cc
index f67c010a8595f84832bb0dec163a53118201d377..012f97e8a11ad4cfab40a9834410f89ec11874cf 100644
--- a/net/udp/udp_socket_win.cc
+++ b/net/udp/udp_socket_win.cc
@@ -31,16 +31,131 @@ static const int kPortEnd = 65535;
 
 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();
+}
+
+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::WriteDelegate::OnObjectSignaled(HANDLE object) {
-  DCHECK_EQ(object, socket_->write_overlapped_.hEvent);
-  socket_->DidCompleteWrite();
+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,
@@ -49,17 +164,11 @@ UDPSocketWin::UDPSocketWin(DatagramSocket::BindType bind_type,
       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());
 }
@@ -80,16 +189,12 @@ void UDPSocketWin::Close() {
   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 {
@@ -155,7 +260,6 @@ int UDPSocketWin::RecvFrom(IOBuffer* buf,
   if (nread != ERR_IO_PENDING)
     return nread;
 
-  read_iobuffer_ = buf;
   read_callback_ = callback;
   recv_from_address_ = address;
   return ERR_IO_PENDING;
@@ -191,7 +295,6 @@ int UDPSocketWin::SendToOrWrite(IOBuffer* buf,
 
   if (address)
     send_to_address_.reset(new IPEndPoint(*address));
-  write_iobuffer_ = buf;
   write_callback_ = callback;
   return ERR_IO_PENDING;
 }
@@ -305,17 +408,17 @@ void UDPSocketWin::DoWriteCallback(int 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);
 }
@@ -343,14 +446,14 @@ void UDPSocketWin::LogRead(int result, const char* bytes) const {
 
 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);
 }
 
@@ -378,18 +481,20 @@ int UDPSocketWin::InternalRecvFrom(IOBuffer* buf, int buf_len,
   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) {
@@ -407,7 +512,7 @@ int UDPSocketWin::InternalRecvFrom(IOBuffer* buf, int buf_len,
       return result;
     }
   }
-  read_watcher_.StartWatching(read_overlapped_.hEvent, &read_delegate_);
+  core_->WatchForRead();
   return ERR_IO_PENDING;
 }
 
@@ -427,17 +532,19 @@ int UDPSocketWin::InternalSendTo(IOBuffer* buf, int buf_len,
     }
   }
 
-  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;
@@ -451,7 +558,7 @@ int UDPSocketWin::InternalSendTo(IOBuffer* buf, int buf_len,
     }
   }
 
-  write_watcher_.StartWatching(write_overlapped_.hEvent, &write_delegate_);
+  core_->WatchForWrite();
   return ERR_IO_PENDING;
 }