Fix crash condition if caller deletes PseudoTcpAdaptor from within a callback.

jingle/glue/pseudotcp_adapter.cc

 // Copyright (c) 2011 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 "jingle/glue/pseudotcp_adapter.h"
 
 #include "base/logging.h"
 #include "base/time.h"
 #include "net/base/address_list.h"
 #include "net/base/completion_callback.h"
 #include "net/base/io_buffer.h"
 #include "net/base/net_errors.h"
 #include "net/base/net_util.h"
 
 using cricket::PseudoTcp;
 
 namespace {
 const int kReadBufferSize = 65536;  // Maximum size of a packet.
 const uint16 kDefaultMtu = 1280;
 }  // namespace
 
 namespace jingle_glue {
 
-PseudoTcpAdapter::PseudoTcpAdapter(net::Socket* socket)
-    : socket_(socket),
-      ALLOW_THIS_IN_INITIALIZER_LIST(pseudotcp_(this, 0)),
-      connect_callback_(NULL),
+class PseudoTcpAdapter::Core : public cricket::IPseudoTcpNotify,
+                               public base::RefCounted<Core> {
+ public:
+  Core(net::Socket* socket);
+  virtual ~Core();
+
+  // Functions used to implement net::StreamSocket.
+  int Read(net::IOBuffer* buffer, int buffer_size,
+           net::CompletionCallback* callback);
+  int Write(net::IOBuffer* buffer, int buffer_size,
+            net::CompletionCallback* callback);
+  int Connect(net::CompletionCallback* callback);
+
+  // cricket::IPseudoTcpNotify interface.
+  // These notifications are triggered from NotifyPacket.
+  virtual void OnTcpOpen(cricket::PseudoTcp* tcp) OVERRIDE;
+  virtual void OnTcpReadable(cricket::PseudoTcp* tcp) OVERRIDE;
+  virtual void OnTcpWriteable(cricket::PseudoTcp* tcp) OVERRIDE;
+  // This is triggered by NotifyClock or NotifyPacket.
+  virtual void OnTcpClosed(cricket::PseudoTcp* tcp, uint32 error) OVERRIDE;
+  // This is triggered by NotifyClock, NotifyPacket, Recv and Send.
+  virtual WriteResult TcpWritePacket(cricket::PseudoTcp* tcp,
+                                     const char* buffer, size_t len) OVERRIDE;
+
+  // "public" to avoid excessive boilerplate in PseudoTcpAdapter.
+  net::CompletionCallback* connect_callback_;

[Sergey Ulanov, 2011/05/23 20:09:05]

Can we make them private, and make Core class a friend of the PseudoTcpAdapter?

[Wez, 2011/05/23 21:27:02]

The only use for them was to support Disconnect() and IsConnected() (an earlier
draft had them used more widely by the PseudoTcpAdapter) so I've implemented
those as boilerplate to keep things clean.

+  net::CompletionCallback* read_callback_;
+  net::CompletionCallback* write_callback_;
+
+  cricket::PseudoTcp pseudo_tcp_;
+ private:
+  // These are invoked by the underlying Socket, and may trigger callbacks.
+  // They hold a reference to |this| while running, to protect from deletion.
+  void OnRead(int result);
+  void OnWritten(int result);
+
+  // These may trigger callbacks, so the holder must hold a reference on
+  // the stack while calling them.
+  void DoReadFromSocket();
+  void HandleReadResults(int result);
+  void HandleTcpClock();
+
+  // This re-sets |timer| without triggering callbacks.
+  void AdjustClock();
+
+  scoped_ptr<net::Socket> socket_;
+
+  scoped_refptr<net::IOBuffer> read_buffer_;
+  int read_buffer_size_;
+  scoped_refptr<net::IOBuffer> write_buffer_;
+  int write_buffer_size_;
+
+  bool socket_write_pending_;
+  scoped_refptr<net::IOBuffer> socket_read_buffer_;
+
+  net::CompletionCallbackImpl<Core> socket_read_callback_;
+  net::CompletionCallbackImpl<Core> socket_write_callback_;
+
+  base::OneShotTimer<Core> timer_;
+
+  DISALLOW_COPY_AND_ASSIGN(Core);
+};
+
+
+PseudoTcpAdapter::Core::Core(net::Socket* socket)
+    : connect_callback_(NULL),
       read_callback_(NULL),
       write_callback_(NULL),
+      ALLOW_THIS_IN_INITIALIZER_LIST(pseudo_tcp_(this, 0)),
+      socket_(socket),
       socket_write_pending_(false),
       ALLOW_THIS_IN_INITIALIZER_LIST(
-          socket_read_callback_(this, &PseudoTcpAdapter::OnRead)),
+          socket_read_callback_(this, &PseudoTcpAdapter::Core::OnRead)),
       ALLOW_THIS_IN_INITIALIZER_LIST(
-          socket_write_callback_(this, &PseudoTcpAdapter::OnWritten)) {
-  pseudotcp_.NotifyMTU(kDefaultMtu);
+          socket_write_callback_(this, &PseudoTcpAdapter::Core::OnWritten)) {
+  // Doesn't trigger callbacks.
+  pseudo_tcp_.NotifyMTU(kDefaultMtu);
 }
 
-PseudoTcpAdapter::~PseudoTcpAdapter() {
+PseudoTcpAdapter::Core::~Core() {
 }
 
-int PseudoTcpAdapter::Read(net::IOBuffer* buffer, int buffer_size,
-                           net::CompletionCallback* callback) {
-  DCHECK(CalledOnValidThread());
+int PseudoTcpAdapter::Core::Read(net::IOBuffer* buffer, int buffer_size,
+                                 net::CompletionCallback* callback) {
+  DCHECK(!read_callback_);
 
-  // Verify that there is no other pending read.
-  DCHECK(read_callback_ == NULL);
+  // Reference the Core in case a callback deletes the adapter.
+  scoped_refptr<Core> core(this);
 
-  PseudoTcp::TcpState state = pseudotcp_.State();
+  // TODO(wez): This is a hack for remoting.  See JingleSession.
+  PseudoTcp::TcpState state = pseudo_tcp_.State();
   int result;
   if (state == PseudoTcp::TCP_SYN_SENT ||
       state == PseudoTcp::TCP_SYN_RECEIVED) {
     result = net::ERR_IO_PENDING;
+
   } else {
-    result = pseudotcp_.Recv(buffer->data(), buffer_size);
+    result = pseudo_tcp_.Recv(buffer->data(), buffer_size);
     if (result < 0) {
-      result = net::MapSystemError(pseudotcp_.GetError());
+      result = net::MapSystemError(pseudo_tcp_.GetError());
       DCHECK(result < 0);
     }
   }
 
   if (result == net::ERR_IO_PENDING) {
     read_buffer_ = buffer;
     read_buffer_size_ = buffer_size;
     read_callback_ = callback;
   }
 
   AdjustClock();
 
   return result;
 }
 
-int PseudoTcpAdapter::Write(net::IOBuffer* buffer, int buffer_size,
-                            net::CompletionCallback* callback) {
-  DCHECK(CalledOnValidThread());
+int PseudoTcpAdapter::Core::Write(net::IOBuffer* buffer, int buffer_size,
+                                  net::CompletionCallback* callback) {
+  DCHECK(!write_callback_);
 
-  // Verify that there is no other pending write.
-  DCHECK(write_callback_ == NULL);
+  // Reference the Core in case a callback deletes the adapter.
+  scoped_refptr<Core> core(this);
 
-  PseudoTcp::TcpState state = pseudotcp_.State();
+  // TODO(wez): This is a hack for remoting.  See JingleSession.
+  PseudoTcp::TcpState state = pseudo_tcp_.State();
   int result;
   if (state == PseudoTcp::TCP_SYN_SENT ||
       state == PseudoTcp::TCP_SYN_RECEIVED) {
     result = net::ERR_IO_PENDING;
+
   } else {
-    result = pseudotcp_.Send(buffer->data(), buffer_size);
+    result = pseudo_tcp_.Send(buffer->data(), buffer_size);
     if (result < 0) {
-      result = net::MapSystemError(pseudotcp_.GetError());
+      result = net::MapSystemError(pseudo_tcp_.GetError());
       DCHECK(result < 0);
     }
   }
 
   if (result == net::ERR_IO_PENDING) {
     write_buffer_ = buffer;
     write_buffer_size_ = buffer_size;
     write_callback_ = callback;
   }
 
   AdjustClock();
 
   return result;
 }
 
-bool PseudoTcpAdapter::SetReceiveBufferSize(int32 size) {
-  DCHECK(CalledOnValidThread());
-  // TODO(sergeyu): Implement support for adjustable buffer size and
-  // used it here.
-  return false;
-}
+int PseudoTcpAdapter::Core::Connect(net::CompletionCallback* callback) {
+  DCHECK_EQ(pseudo_tcp_.State(), cricket::PseudoTcp::TCP_LISTEN);
 
-bool PseudoTcpAdapter::SetSendBufferSize(int32 size) {
-  DCHECK(CalledOnValidThread());
-  // TODO(sergeyu): Implement support for adjustable buffer size and
-  // used it here.
-  return false;
-}
+  // Reference the Core in case a callback deletes the adapter.
+  scoped_refptr<Core> core(this);
 
-int PseudoTcpAdapter::Connect(net::CompletionCallback* callback) {
-  DCHECK(CalledOnValidThread());
-
-  // Start reading from the socket.
-  DoReadFromSocket();
-
-  int result = pseudotcp_.Connect();
+  // Start the connection attempt.
+  int result = pseudo_tcp_.Connect();
   if (result < 0)
     return net::ERR_FAILED;
 
   AdjustClock();
 
   connect_callback_ = callback;
+  DoReadFromSocket();
+
   return net::ERR_IO_PENDING;
 }
 
-void PseudoTcpAdapter::Disconnect() {
-  DCHECK(CalledOnValidThread());
-  pseudotcp_.Close(false);
-}
+void PseudoTcpAdapter::Core::OnTcpOpen(PseudoTcp* tcp) {
+  DCHECK(tcp == &pseudo_tcp_);
 
-bool PseudoTcpAdapter::IsConnected() const {
-  DCHECK(CalledOnValidThread());
-  return pseudotcp_.State() == PseudoTcp::TCP_ESTABLISHED;
-}
-
-bool PseudoTcpAdapter::IsConnectedAndIdle() const {
-  DCHECK(CalledOnValidThread());
-  NOTIMPLEMENTED();
-  return false;
-}
-
-int PseudoTcpAdapter::GetPeerAddress(net::AddressList* address) const {
-  DCHECK(CalledOnValidThread());
-
-  // We actually don't know the peer address. Returning so the upper layers
-  // won't complain.
-  net::IPAddressNumber ip_address(4);
-  *address = net::AddressList::CreateFromIPAddress(ip_address, 0);
-  return net::OK;
-}
-
-int PseudoTcpAdapter::GetLocalAddress(net::IPEndPoint* address) const {
-  DCHECK(CalledOnValidThread());
-  NOTIMPLEMENTED();
-  return net::ERR_FAILED;
-}
-
-const net::BoundNetLog& PseudoTcpAdapter::NetLog() const {
-  DCHECK(CalledOnValidThread());
-  return net_log_;
-}
-
-void PseudoTcpAdapter::SetSubresourceSpeculation() {
-  DCHECK(CalledOnValidThread());
-  NOTIMPLEMENTED();
-}
-
-void PseudoTcpAdapter::SetOmniboxSpeculation() {
-  DCHECK(CalledOnValidThread());
-  NOTIMPLEMENTED();
-}
-
-bool PseudoTcpAdapter::WasEverUsed() const {
-  DCHECK(CalledOnValidThread());
-  NOTIMPLEMENTED();
-  return true;
-}
-
-bool PseudoTcpAdapter::UsingTCPFastOpen() const {
-  DCHECK(CalledOnValidThread());
-  return false;
-}
-
-void PseudoTcpAdapter::OnTcpOpen(PseudoTcp* tcp) {
-  DCHECK(CalledOnValidThread());
-  DCHECK(connect_callback_);
-  connect_callback_->Run(net::OK);
-  connect_callback_ = NULL;
+  if (connect_callback_) {
+    net::CompletionCallback* callback = connect_callback_;
+    connect_callback_ = NULL;
+    callback->Run(net::OK);
+  }
 
   OnTcpReadable(tcp);
   OnTcpWriteable(tcp);
 }
 
-void PseudoTcpAdapter::OnTcpReadable(PseudoTcp* tcp) {
-  DCHECK(CalledOnValidThread());
-
-  if (!read_buffer_)
+void PseudoTcpAdapter::Core::OnTcpReadable(PseudoTcp* tcp) {
+  DCHECK_EQ(tcp, &pseudo_tcp_);
+  if (!read_callback_)
     return;
 
-  // Try to send the data we have pending.
-  int result = pseudotcp_.Recv(read_buffer_->data(), read_buffer_size_);
+  int result = pseudo_tcp_.Recv(read_buffer_->data(), read_buffer_size_);
   if (result < 0) {
-    result = net::MapSystemError(pseudotcp_.GetError());
+    result = net::MapSystemError(pseudo_tcp_.GetError());
     DCHECK(result < 0);
     if (result == net::ERR_IO_PENDING)
       return;
   }
 
   AdjustClock();
 
-  net::CompletionCallback* cb = read_callback_;
+  net::CompletionCallback* callback = read_callback_;
   read_callback_ = NULL;
   read_buffer_ = NULL;
-  cb->Run(result);
+  callback->Run(result);
 }
 
-void PseudoTcpAdapter::OnTcpWriteable(PseudoTcp* tcp) {
-  DCHECK(CalledOnValidThread());
-
-  if (!write_buffer_)
+void PseudoTcpAdapter::Core::OnTcpWriteable(PseudoTcp* tcp) {
+  DCHECK_EQ(tcp, &pseudo_tcp_);
+  if (!write_callback_)
     return;
 
-  // Try to send the data we have pending.
-  int result = pseudotcp_.Send(write_buffer_->data(), write_buffer_size_);
+  int result = pseudo_tcp_.Send(write_buffer_->data(), write_buffer_size_);
   if (result < 0) {
-    result = net::MapSystemError(pseudotcp_.GetError());
+    result = net::MapSystemError(pseudo_tcp_.GetError());
     DCHECK(result < 0);
     if (result == net::ERR_IO_PENDING)
       return;
   }
 
   AdjustClock();
 
-  net::CompletionCallback* cb = write_callback_;
+  net::CompletionCallback* callback = write_callback_;
   write_callback_ = NULL;
   write_buffer_ = NULL;
-  cb->Run(result);
+  callback->Run(result);
 }
 
-void PseudoTcpAdapter::OnTcpClosed(PseudoTcp* tcp, uint32 error) {
-  DCHECK(CalledOnValidThread());
+void PseudoTcpAdapter::Core::OnTcpClosed(PseudoTcp* tcp, uint32 error) {
+  DCHECK_EQ(tcp, &pseudo_tcp_);
 
   if (connect_callback_) {
-    connect_callback_->Run(net::MapSystemError(error));
+    net::CompletionCallback* callback = connect_callback_;
     connect_callback_ = NULL;
+    callback->Run(net::MapSystemError(error));
   }
 
   if (read_callback_) {
-    read_callback_->Run(net::MapSystemError(error));
+    net::CompletionCallback* callback = read_callback_;
     read_callback_ = NULL;
+    callback->Run(net::MapSystemError(error));
   }
 
   if (write_callback_) {
-    write_callback_->Run(net::MapSystemError(error));
+    net::CompletionCallback* callback = write_callback_;
     write_callback_ = NULL;
+    callback->Run(net::MapSystemError(error));
   }
 }
 
-cricket::IPseudoTcpNotify::WriteResult PseudoTcpAdapter::TcpWritePacket(
+cricket::IPseudoTcpNotify::WriteResult PseudoTcpAdapter::Core::TcpWritePacket(
     PseudoTcp* tcp,
     const char* buffer,
     size_t len) {
-  DCHECK(CalledOnValidThread());
+  DCHECK_EQ(tcp, &pseudo_tcp_);
 
   if (socket_write_pending_)
     return IPseudoTcpNotify::WR_SUCCESS;
 
   scoped_refptr<net::IOBuffer> write_buffer = new net::IOBuffer(len);
   memcpy(write_buffer->data(), buffer, len);
 
   int result = socket_->Write(write_buffer, len, &socket_write_callback_);
   if (result == net::ERR_IO_PENDING) {
     socket_write_pending_ = true;
     return IPseudoTcpNotify::WR_SUCCESS;
   } if (result == net::ERR_MSG_TOO_BIG) {
     return IPseudoTcpNotify::WR_TOO_LARGE;
   } else if (result < 0) {
     return IPseudoTcpNotify::WR_FAIL;
   } else {
     return IPseudoTcpNotify::WR_SUCCESS;
   }
 }
 
-void  PseudoTcpAdapter::DoReadFromSocket() {
-  if (!socket_read_buffer_) {
+void PseudoTcpAdapter::Core::DoReadFromSocket() {
+  if (!socket_read_buffer_)
     socket_read_buffer_ = new net::IOBuffer(kReadBufferSize);
-  }
 
   while (true) {
     int result = socket_->Read(socket_read_buffer_, kReadBufferSize,
                                &socket_read_callback_);
     if (result == net::ERR_IO_PENDING)
       break;
 
     HandleReadResults(result);
   }
 }
 
-void PseudoTcpAdapter::HandleReadResults(int result) {
+void PseudoTcpAdapter::Core::HandleReadResults(int result) {
   if (result <= 0) {
     LOG(ERROR) << "Read returned " << result;
     return;
   }
 
-  pseudotcp_.NotifyPacket(socket_read_buffer_->data(), result);
+  // TODO(wez): Disconnect on failure of NotifyPacket?
+  pseudo_tcp_.NotifyPacket(socket_read_buffer_->data(), result);
   AdjustClock();
 }
 
-void PseudoTcpAdapter::OnRead(int result) {
+void PseudoTcpAdapter::Core::OnRead(int result) {
+  // Reference the Core in case a callback deletes the adapter.
+  scoped_refptr<Core> core(this);
+
   HandleReadResults(result);
   if (result >= 0)
     DoReadFromSocket();
 }
 
-void PseudoTcpAdapter::OnWritten(int result) {
+void PseudoTcpAdapter::Core::OnWritten(int result) {
+  // Reference the Core in case a callback deletes the adapter.
+  scoped_refptr<Core> core(this);
+
   socket_write_pending_ = false;
   if (result < 0) {
     LOG(WARNING) << "Write failed. Error code: " << result;
   }
 }
 
-void PseudoTcpAdapter::AdjustClock() {
+void PseudoTcpAdapter::Core::AdjustClock() {
   long timeout = 0;
-  if (pseudotcp_.GetNextClock(PseudoTcp::Now(), timeout)) {
+  if (pseudo_tcp_.GetNextClock(PseudoTcp::Now(), timeout)) {
     timer_.Stop();
     timer_.Start(base::TimeDelta::FromMilliseconds(std::max(timeout, 0L)), this,
-                 &PseudoTcpAdapter::HandleTcpClock);
+                 &PseudoTcpAdapter::Core::HandleTcpClock);
   }
 }
 
-void PseudoTcpAdapter::HandleTcpClock() {
-  pseudotcp_.NotifyClock(PseudoTcp::Now());
+void PseudoTcpAdapter::Core::HandleTcpClock() {
+  // Reference the Core in case a callback deletes the adapter.
+  scoped_refptr<Core> core(this);
+
+  pseudo_tcp_.NotifyClock(PseudoTcp::Now());
   AdjustClock();
 }
 
+// Public interface implemention.
+
+PseudoTcpAdapter::PseudoTcpAdapter(net::Socket* socket)
+    : core_(new Core(socket)) {
+}
+
+PseudoTcpAdapter::~PseudoTcpAdapter() {
+  core_->connect_callback_ = NULL;

[Sergey Ulanov, 2011/05/23 20:09:05]

Do we also need to reset read/write buffers? Maybe add ResetCallbacks() method
in Core and use it here and in Disconnect().

[Wez, 2011/05/23 21:27:02]

I've moved clearing of the callbacks and PseudoTcp::Close() into a Disconnect()
call in the Core, which is called from the dtor and public Disconnect().

+  core_->read_callback_ = NULL;
+  core_->write_callback_ = NULL;
+}
+
+int PseudoTcpAdapter::Read(net::IOBuffer* buffer, int buffer_size,
+                           net::CompletionCallback* callback) {
+  DCHECK(CalledOnValidThread());
+  return core_->Read(buffer, buffer_size, callback);
+}
+
+int PseudoTcpAdapter::Write(net::IOBuffer* buffer, int buffer_size,
+                            net::CompletionCallback* callback) {
+  DCHECK(CalledOnValidThread());
+  return core_->Write(buffer, buffer_size, callback);
+}
+
+bool PseudoTcpAdapter::SetReceiveBufferSize(int32 size) {
+  DCHECK(CalledOnValidThread());
+  // TODO(sergeyu): Implement support for adjustable buffer size and
+  // used it here.
+  return false;
+}
+
+bool PseudoTcpAdapter::SetSendBufferSize(int32 size) {
+  DCHECK(CalledOnValidThread());
+  // TODO(sergeyu): Implement support for adjustable buffer size and
+  // used it here.
+  return false;
+}
+
+int PseudoTcpAdapter::Connect(net::CompletionCallback* callback) {
+  DCHECK(CalledOnValidThread());
+
+  // net::StreamSocket requires that Connect return OK if already connected.
+  if (IsConnected())
+    return net::OK;
+
+  return core_->Connect(callback);
+}
+
+void PseudoTcpAdapter::Disconnect() {
+  DCHECK(CalledOnValidThread());
+
+  // Don't dispatch outstanding callbacks, as mandated by net::StreamSocket.
+  core_->read_callback_ = NULL;
+  core_->write_callback_ = NULL;
+  core_->connect_callback_ = NULL;
+
+  // TODO(wez): Connect should succeed if called after Disconnect, which
+  // PseudoTcp doesn't support, so we need to teardown the internal PseudoTcp
+  // and create a new one in Connect.
+  // TODO(wez): Close sets a shutdown flag inside PseudoTcp but has no other
+  // effect.  This should be addressed in PseudoTcp, really.
+  // In the meantime we can fake OnTcpClosed notification and tear down the
+  // PseudoTcp.
+  // TODO(wez): The "force" parameter to Close() should be true, since the
+  // behaviour of disconnect is to dump any pending reads & writes.
+  core_->pseudo_tcp_.Close(false);
+}
+
+bool PseudoTcpAdapter::IsConnected() const {
+  return core_->pseudo_tcp_.State() == PseudoTcp::TCP_ESTABLISHED;
+}
+
+bool PseudoTcpAdapter::IsConnectedAndIdle() const {
+  DCHECK(CalledOnValidThread());
+  NOTIMPLEMENTED();
+  return false;
+}
+
+int PseudoTcpAdapter::GetPeerAddress(net::AddressList* address) const {
+  DCHECK(CalledOnValidThread());
+
+  // We actually don't know the peer address. Returning so the upper layers
+  // won't complain.
+  net::IPAddressNumber ip_address(4);
+  *address = net::AddressList::CreateFromIPAddress(ip_address, 0);
+  return net::OK;
+}
+
+int PseudoTcpAdapter::GetLocalAddress(net::IPEndPoint* address) const {
+  DCHECK(CalledOnValidThread());
+  NOTIMPLEMENTED();
+  return net::ERR_FAILED;
+}
+
+const net::BoundNetLog& PseudoTcpAdapter::NetLog() const {
+  DCHECK(CalledOnValidThread());
+  return net_log_;
+}
+
+void PseudoTcpAdapter::SetSubresourceSpeculation() {
+  DCHECK(CalledOnValidThread());
+  NOTIMPLEMENTED();
+}
+
+void PseudoTcpAdapter::SetOmniboxSpeculation() {
+  DCHECK(CalledOnValidThread());
+  NOTIMPLEMENTED();
+}
+
+bool PseudoTcpAdapter::WasEverUsed() const {
+  DCHECK(CalledOnValidThread());
+  NOTIMPLEMENTED();
+  return true;
+}
+
+bool PseudoTcpAdapter::UsingTCPFastOpen() const {
+  DCHECK(CalledOnValidThread());
+  return false;
+}
+
 }  // namespace jingle_glue