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(connect_timeout_callback_.IsCancelled());
+    connect_timeout_callback_.Reset(base::Bind(&CastSocket::CancelConnect,
+                                               base::Unretained(this)));
+    GetTimer()->Start(FROM_HERE,
+                      connect_timeout_,
+                      connect_timeout_callback_.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_.IsCancelled());
+  connect_loop_callback_.Reset(base::Bind(&CastSocket::DoConnectLoop,
+                                          base::Unretained(this),
+                                          result));
+  base::MessageLoop::current()->PostTask(FROM_HERE,
+                                         connect_loop_callback_.callback());
 }
 
 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);
 }
 
 // This method performs the state machine transitions for connection flow.
 // There are two entry points to this method:
 // 1. Connect method: this starts the flow
 // 2. Callback from network operations that finish asynchronously
 void CastSocket::DoConnectLoop(int result) {
+  connect_loop_callback_.Cancel();
   if (is_canceled_) {
     LOG(ERROR) << "CANCELLED - Aborting DoConnectLoop.";
     return;
   }
   // Network operations can either finish synchronously or asynchronously.
   // This method executes the state machine transitions in a loop so that
   // correct state transitions happen even when network operations finish
   // synchronously.
   int rv = result;
   do {
@@ skipping 34 matching lines @@
   // b. The Do* method called did not change state
 
   // Connect loop is finished: if there is no pending IO invoke the callback.
   if (rv != net::ERR_IO_PENDING) {
     GetTimer()->Stop();
     DoConnectCallback(rv);
   }
 }
 
 int CastSocket::DoTcpConnect() {
+  DCHECK(connect_loop_callback_.IsCancelled());
   VLOG_WITH_CONNECTION(1) << "DoTcpConnect";
   connect_state_ = CONN_STATE_TCP_CONNECT_COMPLETE;
   tcp_socket_ = CreateTcpSocket();
   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() {
+  DCHECK(connect_loop_callback_.IsCancelled());
   VLOG_WITH_CONNECTION(1) << "DoSslConnect";
   connect_state_ = CONN_STATE_SSL_CONNECT_COMPLETE;
   socket_ = CreateSslSocket(tcp_socket_.PassAs<net::StreamSocket>());
   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.
+  DCHECK(send_auth_challenge_callback_.IsCancelled());
+  send_auth_challenge_callback_.Reset(
+      base::Bind(&CastSocket::SendCastMessageInternal,
+                 base::Unretained(this),
+                 challenge_message,
+                 base::Bind(&CastSocket::DoAuthChallengeSendWriteComplete,
+                            base::Unretained(this))));
   base::MessageLoop::current()->PostTask(
       FROM_HERE,
-      base::Bind(&CastSocket::SendCastMessageInternal,
-                 AsWeakPtr(),
-                 challenge_message,
-                 base::Bind(&CastSocket::DoConnectLoop, AsWeakPtr())));
+      send_auth_challenge_callback_.callback());
   // Always return IO_PENDING since the result is always asynchronous.
   return net::ERR_IO_PENDING;
 }
 
+void CastSocket::DoAuthChallengeSendWriteComplete(int result) {
+  send_auth_challenge_callback_.Cancel();
+  VLOG_WITH_CONNECTION(2) << "DoAuthChallengeSendWriteComplete: " << result;
+  DCHECK_GT(result, 0);
+  DCHECK_EQ(write_queue_.size(), 1UL);
+  PostTaskToStartConnectLoop(result);
+}
+
 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
   // DoConnectLoop. This is not strictly necessary but keeps the read loop
   // code decoupled from connect loop code.
   PostTaskToStartReadLoop();
   // Always return IO_PENDING since the result is always asynchronous.
@@ skipping 18 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) {
-  DCHECK(CalledOnValidThread());
+  CloseInternal();
+  callback.Run(net::OK);
+}
+
+void CastSocket::CloseInternal() {
+  // TODO(mfoltz): Enforce this when CastChannelAPITest is rewritten to create
+  // and free sockets on the same thread.  crbug.com/398242
+  // 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();
+  GetTimer()->Stop();
+
+  // Fire any remaining callbacks passed into us.
+  if (!connect_callback_.is_null()) {
+    connect_callback_.Run(net::ERR_CONNECTION_FAILED);

[Wez, 2014/07/31 22:57:53]

So there is no possibility of this callback deleting this object?

[mark a. foltz, 2014/08/01 23:21:48]

Yes this is possible and that would be bad.  The only callback it is safe to
delete the object should be the one for Close() since it is invoked last.

The CastChannelAPI owns this object as well as the callbacks, and should know
better, but do you suggest a boolean guard against this?

+    connect_callback_.Reset();
+  }
+  for (; !write_queue_.empty(); write_queue_.pop()) {
+    net::CompletionCallback& callback = write_queue_.front().callback;
+    callback.Run(net::ERR_FAILED);
+    callback.Reset();
+  }
+
+  // Cancel callbacks that we queued ourselves to re-enter the connect or read
+  // loops.
+  connect_loop_callback_.Cancel();
+  send_auth_challenge_callback_.Cancel();
+  read_loop_callback_.Cancel();
+  connect_timeout_callback_.Cancel();
   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;
   if (!MessageInfoToCastMessage(message, &message_proto)) {
     callback.Run(net::ERR_FAILED);
     return;
   }
-
   SendCastMessageInternal(message_proto, callback);
 }
 
 void CastSocket::SendCastMessageInternal(
     const CastMessage& message,
     const net::CompletionCallback& callback) {
   WriteRequest write_request(callback);
   if (!write_request.SetContent(message)) {
     callback.Run(net::ERR_FAILED);
     return;
@@ skipping 56 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;
   }
 
   // Some bytes were successfully written
   WriteRequest& request = write_queue_.front();
   scoped_refptr<net::DrainableIOBuffer> io_buffer = request.io_buffer;
   io_buffer->DidConsume(result);
   if (io_buffer->BytesRemaining() == 0)  // Message fully sent
     write_state_ = WRITE_STATE_DO_CALLBACK;
   else
     write_state_ = WRITE_STATE_WRITE;
 
   return net::OK;
 }
 
 int CastSocket::DoWriteCallback() {
   DCHECK(!write_queue_.empty());
+  write_state_ = WRITE_STATE_WRITE;
   WriteRequest& request = write_queue_.front();
   int bytes_consumed = request.io_buffer->BytesConsumed();
-
-  // If inside connection flow, then there should be exaclty one item in
-  // the write queue.
-  if (ready_state_ == READY_STATE_CONNECTING) {
-    write_queue_.pop();
-    DCHECK(write_queue_.empty());
-    PostTaskToStartConnectLoop(bytes_consumed);
-  } else {
-    WriteRequest& request = write_queue_.front();
-    request.callback.Run(bytes_consumed);
-    write_queue_.pop();
-  }
-  write_state_ = WRITE_STATE_WRITE;
+  request.callback.Run(bytes_consumed);
+  write_queue_.pop();
   return net::OK;
 }
 
 int CastSocket::DoWriteError(int result) {
   DCHECK(!write_queue_.empty());
   DCHECK_LT(result, 0);
 
   // If inside connection flow, then there should be exactly one item in
   // the write queue.
   if (ready_state_ == READY_STATE_CONNECTING) {
     write_queue_.pop();
     DCHECK(write_queue_.empty());
     PostTaskToStartConnectLoop(result);
     // Connect loop will handle the error. Return net::OK so that write flow
     // does not try to report error also.
     return net::OK;
   }
 
   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_.IsCancelled());
+  read_loop_callback_.Reset(base::Bind(&CastSocket::StartReadLoop,
+                                       base::Unretained(this)));
+  base::MessageLoop::current()->PostTask(FROM_HERE,
+                                         read_loop_callback_.callback());
 }
 
 void CastSocket::StartReadLoop() {
+  read_loop_callback_.Cancel();
   // 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);
   }
 }
 
 void CastSocket::DoReadLoop(int result) {
   DCHECK(CalledOnValidThread());
   // Network operations can either finish synchronously or asynchronously.
@@ skipping 51 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