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_.IsCancelled());
+    cancel_connect_callback_.Reset(base::Bind(&CastSocket::CancelConnect,
+                                              base::Unretained(this)));

[Wez, 2014/07/30 19:14:53]

nit: Style guide would prefer wrapping this after "Reset(".

[mark a. foltz, 2014/07/31 18:33:17]

It prefers wrapping at parenthesis

+    GetTimer()->Start(FROM_HERE,
+                      connect_timeout_,
+                      cancel_connect_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 37 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_.IsCancelled());

[Wez, 2014/07/30 19:14:53]

nit: Check this on entry to DoTcpConnect()?

[Wez, 2014/07/30 19:14:53]

IIUC this is basically checking that after a PostTaskToRestartConnectLoop we
haven't somehow come around DoConnectLoop again, without having first returned
to the main loop?

[mark a. foltz, 2014/07/31 18:33:17]

Done.

[mark a. foltz, 2014/07/31 18:33:17]

There should be at most one pending callback to reenter the connect loop. 
Having multiple pending callbacks (from both net:: and the MessageLoop) would
mean the behavior of the state machine becomes indeterminate.

   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_.IsCancelled());

[Wez, 2014/07/30 19:14:53]

nit: Check this on entry to the method?

[mark a. foltz, 2014/07/31 18:33:16]

Done.

   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))));

[Wez, 2014/07/30 19:14:53]

Ick... do we really need a clean stack for SendCastMessageInternal as well as
for the completion callback...?

IIUC the problem here is basically with actions that the Delegate might perform
in error callbacks, in which case can we simplify all this code to avoid the
need for stack-popping via PostTask on operations like this by just using
PostTask to deliver the callbacks to the Delegate...?

[mark a. foltz, 2014/07/31 18:33:17]

It appears to be "not strictly necessary" but otherwise the control flow for the
two loops gets entangled.
No I think even in the normal flow we wanted to avoid mutual recursion between
DoConnectLoop and DoWriteLoop.

   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);
+  // |callback| can delete |this|

[Wez, 2014/07/30 19:14:53]

nit: Callers must inherently cope with asynchronous completion of close, so you
could just PostTask() this, and deletion wouldn't be an issue?

[mark a. foltz, 2014/07/31 18:33:16]

This comment is out of date and removed.  Deletion can happen at any time via
the ApiResourceManager if the extension is disabled or the browser shuts down.

+}
+
+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();
+  // Reset callbacks passed into us.  Or should we invoke them with an error?

[Wez, 2014/07/30 19:14:53]

nit: Suggest blank line before this comment, and before the comment below, to
visually separate this function into logical blocks.

[Wez, 2014/07/30 19:14:53]

If the socket is being closed then there shouldn't be any problem with just
dropping write callbacks, I don't think?

[mark a. foltz, 2014/07/31 18:33:16]

Done.

[mark a. foltz, 2014/07/31 18:33:17]

Actually, there may be a Promise pending on the caller side waiting for the
callback to complete, so now I fire them.  This required refactoring
cast_channel_api.cc since we cannot guarantee the API callbacks are run while
the socket still exists in the ApiResourceManager.

+  connect_callback_.Reset();
+  for (; !write_queue_.empty(); write_queue_.pop()) {
+    write_queue_.front().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();
+  cancel_connect_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