Cancel loading when body stream reader is detached.

content/child/shared_memory_data_consumer_handle.cc

 // Copyright 2015 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 "content/child/shared_memory_data_consumer_handle.h"
 
 #include <algorithm>
 #include <deque>
 
 #include "base/bind.h"
 #include "base/message_loop/message_loop.h"
 #include "base/single_thread_task_runner.h"
 #include "base/synchronization/lock.h"
+#include "base/thread_task_runner_handle.h"
 #include "content/public/child/fixed_received_data.h"
 
 namespace content {
 
 namespace {
 
 class DelegateThreadSafeReceivedData final
     : public RequestPeer::ThreadSafeReceivedData {
  public:
   explicit DelegateThreadSafeReceivedData(
       scoped_ptr<RequestPeer::ReceivedData> data)
-      : data_(data.Pass()),
-        task_runner_(base::MessageLoop::current()->task_runner()) {}
+      : data_(data.Pass()), task_runner_(base::ThreadTaskRunnerHandle::Get()) {}
   ~DelegateThreadSafeReceivedData() override {
     if (!task_runner_->BelongsToCurrentThread()) {
       // Delete the data on the original thread.
       task_runner_->DeleteSoon(FROM_HERE, data_.release());
     }
   }
 
   const char* payload() const override { return data_->payload(); }
   int length() const override { return data_->length(); }
   int encoded_length() const override { return data_->encoded_length(); }
 
  private:
   scoped_ptr<RequestPeer::ReceivedData> data_;
   scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
 
   DISALLOW_COPY_AND_ASSIGN(DelegateThreadSafeReceivedData);
 };
 
 }  // namespace
 
 using Result = blink::WebDataConsumerHandle::Result;
 
 class SharedMemoryDataConsumerHandle::Context final
     : public base::RefCountedThreadSafe<Context> {
  public:
-  Context()
+  explicit Context(const base::Closure& on_reader_detached)
       : result_(Ok),
         first_offset_(0),
         client_(nullptr),
+        writer_task_runner_(base::ThreadTaskRunnerHandle::Get()),
+        on_reader_detached_(on_reader_detached),
+        is_on_reader_detached_valid_(!on_reader_detached_.is_null()),
         is_handle_active_(true),
         is_two_phase_read_in_progress_(false) {}
 
   bool IsEmpty() const { return queue_.empty(); }
   void ClearIfNecessary() {
     if (!is_handle_locked() && !is_handle_active()) {
       // No one is interested in the contents.
+      if (is_on_reader_detached_valid_) {
+        // We post a task even in the writer thread in order to avoid a
+        // reentrance problem as calling |on_reader_detached_| may manipulate
+        // the context synchronously.
+        writer_task_runner_->PostTask(FROM_HERE, on_reader_detached_);
+      }
       Clear();
     }
   }
   void ClearQueue() {
     for (auto& data : queue_) {
       delete data;
     }
     queue_.clear();
     first_offset_ = 0;
   }
   RequestPeer::ThreadSafeReceivedData* Top() { return queue_.front(); }
   void Push(scoped_ptr<RequestPeer::ThreadSafeReceivedData> data) {
     queue_.push_back(data.release());
   }
   size_t first_offset() const { return first_offset_; }
   Result result() const { return result_; }
   void set_result(Result r) { result_ = r; }
   void AcquireReaderLock(Client* client) {
     DCHECK(!notification_task_runner_);
     DCHECK(!client_);
-    notification_task_runner_ = base::MessageLoop::current()->task_runner();
+    notification_task_runner_ = base::ThreadTaskRunnerHandle::Get();
     client_ = client;
     if (client && !(IsEmpty() && result() == Ok)) {
       // We cannot notify synchronously because the user doesn't have the reader
       // yet.
       notification_task_runner_->PostTask(
           FROM_HERE, base::Bind(&Context::NotifyInternal, this, false));
     }
   }
   void ReleaseReaderLock() {
     DCHECK(notification_task_runner_);
     notification_task_runner_ = nullptr;
     client_ = nullptr;
   }
   void PostNotify() {
     auto runner = notification_task_runner_;
     if (!runner)
       return;
     // We don't re-post the task when the runner changes while waiting for
     // this task because in this case a new reader is obtained and
     // notification is already done at the reader creation time if necessary.
     runner->PostTask(FROM_HERE,
                      base::Bind(&Context::NotifyInternal, this, false));
   }
   void Notify() { NotifyInternal(true); }
+  // This function doesn't work in the destructor if |on_reader_detached_| is
+  // not null.
+  void ResetOnReaderDetached() {
+    if (on_reader_detached_.is_null()) {
+      DCHECK(!is_on_reader_detached_valid_);
+      return;
+    }
+    is_on_reader_detached_valid_ = false;
+    if (writer_task_runner_->BelongsToCurrentThread()) {
+      // We can reset the closure immediately.
+      on_reader_detached_.Reset();
+    } else {
+      // We need to reset |on_reader_detached_| on the right thread because it
+      // might lead to the object destruction.
+      writer_task_runner_->PostTask(
+          FROM_HERE, base::Bind(&Context::ResetOnReaderDetachedWithLock, this));
+    }
+  }
   bool is_handle_locked() const { return notification_task_runner_; }
   bool IsReaderBoundToCurrentThread() const {
     return notification_task_runner_ &&
            notification_task_runner_->BelongsToCurrentThread();
   }
   bool is_handle_active() const { return is_handle_active_; }
   void set_is_handle_active(bool b) { is_handle_active_ = b; }
   void Consume(size_t s) {
     first_offset_ += s;
     auto top = Top();
@@ skipping 34 matching lines @@
                        base::Bind(&Context::NotifyInternal, this, false));
     }
   }
   void Clear() {
     for (auto& data : queue_) {
       delete data;
     }
     queue_.clear();
     first_offset_ = 0;
     client_ = nullptr;
+    // Note this doesn't work in the destructor if |on_reader_detached_| is not
+    // null. We have an assert in the destructor.
+    ResetOnReaderDetached();
+  }
+  void ResetOnReaderDetachedWithLock() {
+    base::AutoLock lock(lock_);
+    ResetOnReaderDetached();
   }
 
   friend class base::RefCountedThreadSafe<Context>;
   ~Context() {
+    DCHECK(on_reader_detached_.is_null());
+
     // This is necessary because the queue stores raw pointers.
     Clear();
   }
 
   base::Lock lock_;
   // |result_| stores the ultimate state of this handle if it has. Otherwise,
   // |Ok| is set.
   Result result_;
   // TODO(yhirano): Use std::deque<scoped_ptr<ThreadSafeReceivedData>>
   // once it is allowed.
   std::deque<RequestPeer::ThreadSafeReceivedData*> queue_;
   size_t first_offset_;
   Client* client_;
   scoped_refptr<base::SingleThreadTaskRunner> notification_task_runner_;
+  scoped_refptr<base::SingleThreadTaskRunner> writer_task_runner_;
+  base::Closure on_reader_detached_;
+  // We need this boolean variable to remember if |on_reader_detached_| is
+  // callable because we need to reset |on_reader_detached_| only on the writer
+  // thread and hence |on_reader_detached_.is_null()| is untrustworthy on
+  // other threads.
+  bool is_on_reader_detached_valid_;
   bool is_handle_active_;
   bool is_two_phase_read_in_progress_;
 
   DISALLOW_COPY_AND_ASSIGN(Context);
 };
 
 SharedMemoryDataConsumerHandle::Writer::Writer(
     const scoped_refptr<Context>& context,
     BackpressureMode mode)
     : context_(context), mode_(mode) {
 }
 
 SharedMemoryDataConsumerHandle::Writer::~Writer() {
   Close();
+  base::AutoLock lock(context_->lock());
+  context_->ResetOnReaderDetached();
 }
 
 void SharedMemoryDataConsumerHandle::Writer::AddData(
     scoped_ptr<RequestPeer::ReceivedData> data) {
   if (!data->length()) {
     // We omit empty data.
     return;
   }
 
   bool needs_notification = false;
@@ skipping 23 matching lines @@
   }
 }
 
 void SharedMemoryDataConsumerHandle::Writer::Close() {
   bool needs_notification = false;
 
   {
     base::AutoLock lock(context_->lock());
     if (context_->result() == Ok) {
       context_->set_result(Done);
+      context_->ResetOnReaderDetached();
       needs_notification = context_->IsEmpty();
     }
   }
   if (needs_notification) {
     // We cannot issue the notification synchronously because this function can
     // be called in the client's callback.
     context_->PostNotify();
   }
 }
 
 void SharedMemoryDataConsumerHandle::Writer::Fail() {
   bool needs_notification = false;
   {
     base::AutoLock lock(context_->lock());
     if (context_->result() == Ok) {
       // TODO(yhirano): Use an appropriate error code other than
       // UnexpectedError.
       context_->set_result(UnexpectedError);
 
       if (context_->is_two_phase_read_in_progress()) {
         // If we are in two-phase read session, we cannot discard the data. We
         // will clear the queue at the end of the session.
       } else {
         context_->ClearQueue();
       }
 
+      context_->ResetOnReaderDetached();
       needs_notification = true;
     }
   }
   if (needs_notification) {
     // We cannot issue the notification synchronously because this function can
     // be called in the client's callback.
     context_->PostNotify();
   }
 }
 
@@ skipping 82 matching lines @@
   } else {
     context_->Consume(read_size);
   }
 
   return Ok;
 }
 
 SharedMemoryDataConsumerHandle::SharedMemoryDataConsumerHandle(
     BackpressureMode mode,
     scoped_ptr<Writer>* writer)
-    : context_(new Context) {
+    : SharedMemoryDataConsumerHandle(mode, base::Closure(), writer) {
+}
+
+SharedMemoryDataConsumerHandle::SharedMemoryDataConsumerHandle(
+    BackpressureMode mode,
+    const base::Closure& on_reader_detached,
+    scoped_ptr<Writer>* writer)
+    : context_(new Context(on_reader_detached)) {
   writer->reset(new Writer(context_, mode));
 }
 
 SharedMemoryDataConsumerHandle::~SharedMemoryDataConsumerHandle() {
   base::AutoLock lock(context_->lock());
   context_->set_is_handle_active(false);
   context_->ClearIfNecessary();
 }
 
 scoped_ptr<blink::WebDataConsumerHandle::Reader>
 SharedMemoryDataConsumerHandle::ObtainReader(Client* client) {
   return make_scoped_ptr(obtainReaderInternal(client));
 }
 
 SharedMemoryDataConsumerHandle::ReaderImpl*
 SharedMemoryDataConsumerHandle::obtainReaderInternal(Client* client) {
   return new ReaderImpl(context_, client);
 }
 
 const char* SharedMemoryDataConsumerHandle::debugName() const {
   return "SharedMemoryDataConsumerHandle";
 }
 
 }  // namespace content