Remove weak pointers from CastSocket.

extensions/browser/api/cast_channel/cast_socket.cc

 // Copyright 2014 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 "extensions/browser/api/cast_channel/cast_socket.h"
 
 #include <stdlib.h>
 #include <string.h>
 
 #include "base/bind.h"
@@ skipping 81 matching lines @@
   net_log_source_.id = net_log_->NextID();
 
   // Reuse these buffers for each message.
   header_read_buffer_ = new net::GrowableIOBuffer();
   header_read_buffer_->SetCapacity(MessageHeader::header_size());
   body_read_buffer_ = new net::GrowableIOBuffer();
   body_read_buffer_->SetCapacity(MessageHeader::max_message_size());
   current_read_buffer_ = header_read_buffer_;
 }
 
-CastSocket::~CastSocket() { }
+CastSocket::~CastSocket() {
+  CloseInternal();
+}
 
 ReadyState CastSocket::ready_state() const {
   return ready_state_;
 }
 
 ChannelError CastSocket::error_state() const {
   return error_state_;
 }
 
 scoped_ptr<net::TCPClientSocket> CastSocket::CreateTcpSocket() {
@@ skipping 56 matching lines @@
   DCHECK(CalledOnValidThread());
   VLOG_WITH_CONNECTION(1) << "Connect readyState = " << ready_state_;
   if (ready_state_ != READY_STATE_NONE) {
     callback.Run(net::ERR_CONNECTION_FAILED);
     return;
   }
   ready_state_ = READY_STATE_CONNECTING;
   connect_callback_ = callback;
   connect_state_ = CONN_STATE_TCP_CONNECT;
   if (connect_timeout_.InMicroseconds() > 0) {
-    GetTimer()->Start(
-        FROM_HERE,
-        connect_timeout_,
-        base::Bind(&CastSocket::CancelConnect, AsWeakPtr()));
+    DCHECK(cancel_connect_callback_.is_null());
+    cancel_connect_callback_ = base::Bind(&CastSocket::CancelConnect,
+                                          base::Unretained(this));
+    GetTimer()->Start(FROM_HERE, connect_timeout_, cancel_connect_callback_);
   }
   DoConnectLoop(net::OK);
 }
 
 void CastSocket::PostTaskToStartConnectLoop(int result) {
   DCHECK(CalledOnValidThread());
-  base::MessageLoop::current()->PostTask(
-      FROM_HERE,
-      base::Bind(&CastSocket::DoConnectLoop, AsWeakPtr(), result));
+  DCHECK(connect_loop_callback_.is_null());
+  connect_loop_callback_ = base::Bind(&CastSocket::DoConnectLoop,
+                                      base::Unretained(this),
+                                      result);
+  base::MessageLoop::current()->PostTask(FROM_HERE, connect_loop_callback_);

[Kevin M, 2014/07/28 22:08:16]

Should this be turned into PostNonNestableTask now, or should that be in a
different CL?

[Wez, 2014/07/28 23:14:56]

PostNonNestableTask() isn't required here, since we don't nest
MessageLoop::Run() in production code - it should always be safe to call
PostTask(), w/out risking re-entrancy within the PostTask itself.

[Wez, 2014/07/28 23:14:56]

This will blow up if PostTaskToStartConnectLoop() is called and something
deletes the CastSocket before the task is processed.

[mark a. foltz, 2014/07/29 00:16:24]

So Reset()-ing the callback in the dtor is a no-op?

Is there any way to cancel a pending task once submitted to the MessageLoop or
must we rely on indirect techniques like WeakPtr?

[mark a. foltz, 2014/07/29 00:16:25]

Acknowledged.

[Wez, 2014/07/29 00:41:25]

base::Closure is ref-counted, so when you message_loop->PostTask(..., a) and
then a.Reset(), the message_loop's queue still holds a reference to the task,
even though |a| itself is cleared.
There is base::CancelableCallback. Internally, though, that just uses
WeakPtrFactory. MessageQueues don't support cancellation, AFAIK, presumably
because that would complicate them and impact performance of the common case of
non-cancelled tasks.

 }
 
 void CastSocket::CancelConnect() {
   DCHECK(CalledOnValidThread());
   // Stop all pending connection setup tasks and report back to the client.
   is_canceled_ = true;
   VLOG_WITH_CONNECTION(1) << "Timeout while establishing a connection.";
   DoConnectCallback(net::ERR_TIMED_OUT);
 }
 
@@ skipping 52 matching lines @@
   if (rv != net::ERR_IO_PENDING) {
     GetTimer()->Stop();
     DoConnectCallback(rv);
   }
 }
 
 int CastSocket::DoTcpConnect() {
   VLOG_WITH_CONNECTION(1) << "DoTcpConnect";
   connect_state_ = CONN_STATE_TCP_CONNECT_COMPLETE;
   tcp_socket_ = CreateTcpSocket();
+  DCHECK(connect_loop_callback_.is_null());
   return tcp_socket_->Connect(
-      base::Bind(&CastSocket::DoConnectLoop, AsWeakPtr()));
+      base::Bind(&CastSocket::DoConnectLoop, base::Unretained(this)));
 }
 
 int CastSocket::DoTcpConnectComplete(int result) {
   VLOG_WITH_CONNECTION(1) << "DoTcpConnectComplete: " << result;
   if (result == net::OK) {
     // Enable TCP protocol-level keep-alive.
     bool result = tcp_socket_->SetKeepAlive(true, kTcpKeepAliveDelaySecs);
     LOG_IF(WARNING, !result) << "Failed to SetKeepAlive.";
     connect_state_ = CONN_STATE_SSL_CONNECT;
   }
   return result;
 }
 
 int CastSocket::DoSslConnect() {
   VLOG_WITH_CONNECTION(1) << "DoSslConnect";
   connect_state_ = CONN_STATE_SSL_CONNECT_COMPLETE;
   socket_ = CreateSslSocket(tcp_socket_.PassAs<net::StreamSocket>());
+  DCHECK(connect_loop_callback_.is_null());
   return socket_->Connect(
-      base::Bind(&CastSocket::DoConnectLoop, AsWeakPtr()));
+      base::Bind(&CastSocket::DoConnectLoop, base::Unretained(this)));
 }
 
 int CastSocket::DoSslConnectComplete(int result) {
   VLOG_WITH_CONNECTION(1) << "DoSslConnectComplete: " << result;
   if (result == net::ERR_CERT_AUTHORITY_INVALID &&
       peer_cert_.empty() && ExtractPeerCert(&peer_cert_)) {
     connect_state_ = CONN_STATE_TCP_CONNECT;
   } else if (result == net::OK &&
              channel_auth_ == CHANNEL_AUTH_TYPE_SSL_VERIFIED) {
     connect_state_ = CONN_STATE_AUTH_CHALLENGE_SEND;
   }
   return result;
 }
 
 int CastSocket::DoAuthChallengeSend() {
   VLOG_WITH_CONNECTION(1) << "DoAuthChallengeSend";
   connect_state_ = CONN_STATE_AUTH_CHALLENGE_SEND_COMPLETE;
   CastMessage challenge_message;
   CreateAuthChallengeMessage(&challenge_message);
   VLOG_WITH_CONNECTION(1) << "Sending challenge: "
                           << CastMessageToString(challenge_message);
   // Post a task to send auth challenge so that DoWriteLoop is not nested inside
   // DoConnectLoop. This is not strictly necessary but keeps the write loop
-  // code decoupled from connect loop code.
-  base::MessageLoop::current()->PostTask(
-      FROM_HERE,
-      base::Bind(&CastSocket::SendCastMessageInternal,
-                 AsWeakPtr(),
-                 challenge_message,
-                 base::Bind(&CastSocket::DoConnectLoop, AsWeakPtr())));
+  // code decoupleld from connect loop code.

[Kevin M, 2014/07/28 22:08:16]

typo

[mark a. foltz, 2014/07/29 00:16:24]

Done.

+  DCHECK(connect_loop_callback_.is_null());
+  DCHECK(send_auth_challenge_callback_.is_null());
+  send_auth_challenge_callback_ =
+    base::Bind(&CastSocket::SendCastMessageInternal,

[Kevin M, 2014/07/28 22:08:16]

4 space indentation

[mark a. foltz, 2014/07/29 00:16:25]

Done.

+               base::Unretained(this),
+               challenge_message,
+               base::Bind(&CastSocket::DoConnectLoop,
+                          base::Unretained(this)));
+  base::MessageLoop::current()->PostTask(FROM_HERE,
+                                         send_auth_challenge_callback_);

[Wez, 2014/07/28 23:14:56]

This will blow up, as above.

[mark a. foltz, 2014/07/29 00:16:25]

Acknowledged.

   // Always return IO_PENDING since the result is always asynchronous.
   return net::ERR_IO_PENDING;
 }
 
 int CastSocket::DoAuthChallengeSendComplete(int result) {
   VLOG_WITH_CONNECTION(1) << "DoAuthChallengeSendComplete: " << result;
   if (result < 0)
     return result;
   connect_state_ = CONN_STATE_AUTH_CHALLENGE_REPLY_COMPLETE;
   // Post a task to start read loop so that DoReadLoop is not nested inside
@@ skipping 22 matching lines @@
   } else if (result == net::ERR_TIMED_OUT) {
     error_state_ = CHANNEL_ERROR_CONNECT_TIMEOUT;
   } else {
     error_state_ = CHANNEL_ERROR_CONNECT_ERROR;
   }
   VLOG_WITH_CONNECTION(1) << "Calling Connect_Callback";
   base::ResetAndReturn(&connect_callback_).Run(result);
 }
 
 void CastSocket::Close(const net::CompletionCallback& callback) {
+  CloseInternal();
+  callback.Run(net::OK);
+  // |callback| can delete |this|
+}
+
+void CastSocket::CloseInternal() {
   DCHECK(CalledOnValidThread());
+  if (ready_state_ == READY_STATE_CLOSED) {
+    return;
+  }
   VLOG_WITH_CONNECTION(1) << "Close ReadyState = " << ready_state_;
   tcp_socket_.reset();
   socket_.reset();
   cert_verifier_.reset();
   transport_security_state_.reset();
+  // Reset callbacks passed into us.  Or should we invoke them with an error?

[Kevin M, 2014/07/28 22:08:16]

Stop the timer?

[mark a. foltz, 2014/07/29 00:16:24]

Done.

+  connect_callback_.Reset();

[Kevin M, 2014/07/28 22:08:16]

Question: does resetting the callbacks also dequeue them from the message loop?

[mark a. foltz, 2014/07/29 00:16:24]

Good question.

+  for (; !write_queue_.empty(); write_queue_.pop()) {
+    write_queue_.front().callback.Reset();
+  }
+  // Reset callbacks that we queued ourselves.
+  connect_loop_callback_.Reset();
+  read_loop_callback_.Reset();
+  cancel_connect_callback_.Reset();
+  send_auth_challenge_callback_.Reset();
   ready_state_ = READY_STATE_CLOSED;
-  callback.Run(net::OK);
-  // |callback| can delete |this|
 }
 
 void CastSocket::SendMessage(const MessageInfo& message,
                              const net::CompletionCallback& callback) {
   DCHECK(CalledOnValidThread());
   if (ready_state_ != READY_STATE_OPEN) {
     callback.Run(net::ERR_FAILED);
     return;
   }
   CastMessage message_proto;
@@ skipping 71 matching lines @@
 
 int CastSocket::DoWrite() {
   DCHECK(!write_queue_.empty());
   WriteRequest& request = write_queue_.front();
 
   VLOG_WITH_CONNECTION(2) << "WriteData byte_count = "
                           << request.io_buffer->size() << " bytes_written "
                           << request.io_buffer->BytesConsumed();
 
   write_state_ = WRITE_STATE_WRITE_COMPLETE;
-
   return socket_->Write(
       request.io_buffer.get(),
       request.io_buffer->BytesRemaining(),
-      base::Bind(&CastSocket::DoWriteLoop, AsWeakPtr()));
+      base::Bind(&CastSocket::DoWriteLoop, base::Unretained(this)));
 }
 
 int CastSocket::DoWriteComplete(int result) {
   DCHECK(!write_queue_.empty());
   if (result <= 0) {  // NOTE that 0 also indicates an error
     error_state_ = CHANNEL_ERROR_SOCKET_ERROR;
     write_state_ = WRITE_STATE_ERROR;
     return result == 0 ? net::ERR_FAILED : result;
   }
 
@@ skipping 47 matching lines @@
   while (!write_queue_.empty()) {
     WriteRequest& request = write_queue_.front();
     request.callback.Run(result);
     write_queue_.pop();
   }
   return net::ERR_FAILED;
 }
 
 void CastSocket::PostTaskToStartReadLoop() {
   DCHECK(CalledOnValidThread());
-  base::MessageLoop::current()->PostTask(
-      FROM_HERE,
-      base::Bind(&CastSocket::StartReadLoop, AsWeakPtr()));
+  DCHECK(read_loop_callback_.is_null());
+  read_loop_callback_ = base::Bind(&CastSocket::StartReadLoop,
+                                   base::Unretained(this));
+  base::MessageLoop::current()->PostTask(FROM_HERE, read_loop_callback_);

[Wez, 2014/07/28 23:14:56]

Boom!

[mark a. foltz, 2014/07/29 00:16:25]

Acknowledged.

 }
 
 void CastSocket::StartReadLoop() {
   // Read loop would have already been started if read state is not NONE
   if (read_state_ == READ_STATE_NONE) {
     read_state_ = READ_STATE_READ;
     DoReadLoop(net::OK);
   }
 }
 
@@ skipping 54 matching lines @@
     num_bytes_to_read = current_message_size_ - body_read_buffer_->offset();
     current_read_buffer_ = body_read_buffer_;
     CHECK_LE(num_bytes_to_read, MessageHeader::max_message_size());
   }
   CHECK_GT(num_bytes_to_read, 0U);
 
   // Read up to num_bytes_to_read into |current_read_buffer_|.
   return socket_->Read(
       current_read_buffer_.get(),
       num_bytes_to_read,
-      base::Bind(&CastSocket::DoReadLoop, AsWeakPtr()));
+      base::Bind(&CastSocket::DoReadLoop, base::Unretained(this)));
 }
 
 int CastSocket::DoReadComplete(int result) {
   VLOG_WITH_CONNECTION(2) << "DoReadComplete result = " << result
                           << " header offset = "
                           << header_read_buffer_->offset()
                           << " body offset = " << body_read_buffer_->offset();
   if (result <= 0) {  // 0 means EOF: the peer closed the socket
     VLOG_WITH_CONNECTION(1) << "Read error, peer closed the socket";
     error_state_ = CHANNEL_ERROR_SOCKET_ERROR;
@@ skipping 99 matching lines @@
     return false;
   }
   CastSocket::MessageHeader header;
   header.SetMessageSize(message_size);
   header.PrependToString(message_data);
   return true;
 }
 
 void CastSocket::CloseWithError(ChannelError error) {
   DCHECK(CalledOnValidThread());
-  socket_.reset(NULL);
-  ready_state_ = READY_STATE_CLOSED;
+  CloseInternal();
   error_state_ = error;
   if (delegate_)
     delegate_->OnError(this, error);
 }
 
 std::string CastSocket::CastUrl() const {
   return ((channel_auth_ == CHANNEL_AUTH_TYPE_SSL_VERIFIED) ?
           "casts://" : "cast://") + ip_endpoint_.ToString();
 }
 
@@ skipping 11 matching lines @@
   DCHECK_LT(size, static_cast<size_t>(kuint32max));
   DCHECK_GT(size, 0U);
   message_size = size;
 }
 
 // TODO(mfoltz): Investigate replacing header serialization with base::Pickle,
 // if bit-for-bit compatible.
 void CastSocket::MessageHeader::PrependToString(std::string* str) {
   MessageHeader output = *this;
   output.message_size = base::HostToNet32(message_size);
-  size_t header_size = base::checked_cast<size_t,uint32>(
+  size_t header_size = base::checked_cast<size_t, uint32>(
       MessageHeader::header_size());
   scoped_ptr<char, base::FreeDeleter> char_array(
       static_cast<char*>(malloc(header_size)));
   memcpy(char_array.get(), &output, header_size);
   str->insert(0, char_array.get(), header_size);
 }
 
 // TODO(mfoltz): Investigate replacing header deserialization with base::Pickle,
 // if bit-for-bit compatible.
 void CastSocket::MessageHeader::ReadFromIOBuffer(
     net::GrowableIOBuffer* buffer, MessageHeader* header) {
   uint32 message_size;
-  size_t header_size = base::checked_cast<size_t,uint32>(
+  size_t header_size = base::checked_cast<size_t, uint32>(
       MessageHeader::header_size());
   memcpy(&message_size, buffer->StartOfBuffer(), header_size);
   header->message_size = base::NetToHost32(message_size);
 }
 
 std::string CastSocket::MessageHeader::ToString() {
   return "{message_size: " + base::UintToString(message_size) + "}";
 }
 
 CastSocket::WriteRequest::WriteRequest(const net::CompletionCallback& callback)
   : callback(callback) { }
 
 bool CastSocket::WriteRequest::SetContent(const CastMessage& message_proto) {
   DCHECK(!io_buffer.get());
   std::string message_data;
   if (!Serialize(message_proto, &message_data))
     return false;
   io_buffer = new net::DrainableIOBuffer(new net::StringIOBuffer(message_data),
                                          message_data.size());
   return true;
 }
 
 CastSocket::WriteRequest::~WriteRequest() { }
 
 }  // namespace cast_channel
 }  // namespace core_api
 }  // namespace extensions
 
 #undef VLOG_WITH_CONNECTION