Set error on WebDataConsumerHandle while loading body.

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"
@@ skipping 35 matching lines @@
 
 using Result = blink::WebDataConsumerHandle::Result;
 
 class SharedMemoryDataConsumerHandle::Context final
     : public base::RefCountedThreadSafe<Context> {
  public:
   Context()
       : result_(Ok),
         first_offset_(0),
         client_(nullptr),
-        is_handle_active_(true) {}
+        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.
       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_);
@@ skipping 21 matching lines @@
   void set_is_handle_active(bool b) { is_handle_active_ = b; }
   void Consume(size_t s) {
     first_offset_ += s;
     auto top = Top();
     if (static_cast<size_t>(top->length()) <= first_offset_) {
       delete top;
       queue_.pop_front();
       first_offset_ = 0;
     }
   }
+  bool is_two_phase_read_in_progress() const {
+    return is_two_phase_read_in_progress_;
+  }
+  void set_is_two_phase_read_in_progress(bool b) {
+    is_two_phase_read_in_progress_ = b;
+  }
   base::Lock& lock() { return lock_; }
 
  private:
   void NotifyInternal(bool repost) {
     // Note that this function is not protected by |lock_|.
 
     auto runner = notification_task_runner_;
     if (!runner)
       return;
 
@@ skipping 31 matching lines @@
   // |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_;
   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) {
 }
 
@@ skipping 38 matching lines @@
     base::AutoLock lock(context_->lock());
     if (context_->result() == Ok) {
       context_->set_result(Done);
       needs_notification = context_->IsEmpty();
     }
   }
   if (needs_notification)
     context_->Notify();
 }
 
+void SharedMemoryDataConsumerHandle::Writer::Error() {
+  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.
+        // Otherwise, clear the data.

[hiroshige, 2015/06/18 07:01:20]

Please mention in the comment when/how ClearQueue() is called when
is_two_phase_read_in_progress() is true.
(In my understandings, if two-phase read is ongoing then ClearQueue() is called
in endRead() because we set_result(UnexpectedError) here.)

[yhirano, 2015/06/18 08:13:33]

Done.

+        context_->ClearQueue();
+      }
+
+      needs_notification = true;
+    }
+  }
+  if (needs_notification)
+    context_->Notify();
+}
+
 SharedMemoryDataConsumerHandle::ReaderImpl::ReaderImpl(
     scoped_refptr<Context> context,
     Client* client)
     : context_(context) {
   base::AutoLock lock(context_->lock());
   DCHECK(!context_->is_handle_locked());
   context_->AcquireReaderLock(client);
 }
 
 SharedMemoryDataConsumerHandle::ReaderImpl::~ReaderImpl() {
   base::AutoLock lock(context_->lock());
   context_->ReleaseReaderLock();
   context_->ClearIfNecessary();
 }
 
 Result SharedMemoryDataConsumerHandle::ReaderImpl::read(
     void* data,
     size_t size,
     Flags flags,
     size_t* read_size_to_return) {
   base::AutoLock lock(context_->lock());
 
   size_t total_read_size = 0;
   *read_size_to_return = 0;
+
+  if (context_->is_two_phase_read_in_progress())
+    context_->set_result(UnexpectedError);

[hiroshige, 2015/06/18 07:01:21]

Is this change for setting errors on read()/beginRead() during two-phase read
(which was previously unchecked) and thus independent from error notification to
WebDataConsumerHandle?
If so, please mention this in the CL description.

[yhirano, 2015/06/18 08:13:33]

Done. By the way, I added a condition to stabilize the result.

+
   if (context_->result() != Ok && context_->result() != Done)
     return context_->result();
 
   while (!context_->IsEmpty() && total_read_size < size) {
     const auto& top = context_->Top();
     size_t readable = top->length() - context_->first_offset();
     size_t writable = size - total_read_size;
     size_t read_size = std::min(readable, writable);
     const char* begin = top->payload() + context_->first_offset();
     std::copy(begin, begin + read_size,
               static_cast<char*>(data) + total_read_size);
     total_read_size += read_size;
     context_->Consume(read_size);
   }
   *read_size_to_return = total_read_size;
   return total_read_size ? Ok : context_->result() == Done ? Done : ShouldWait;
 }
 
 Result SharedMemoryDataConsumerHandle::ReaderImpl::beginRead(
     const void** buffer,
     Flags flags,
     size_t* available) {
   *buffer = nullptr;
   *available = 0;
 
   base::AutoLock lock(context_->lock());
 
+  if (context_->is_two_phase_read_in_progress())
+    context_->set_result(UnexpectedError);

[hiroshige, 2015/06/18 07:01:21]

ditto.

[yhirano, 2015/06/18 08:13:33]

Done.

+
   if (context_->result() != Ok && context_->result() != Done)
     return context_->result();
 
   if (context_->IsEmpty())
     return context_->result() == Done ? Done : ShouldWait;
 
+  context_->set_is_two_phase_read_in_progress(true);
   const auto& top = context_->Top();
   *buffer = top->payload() + context_->first_offset();
   *available = top->length() - context_->first_offset();
 
   return Ok;
 }
 
 Result SharedMemoryDataConsumerHandle::ReaderImpl::endRead(size_t read_size) {
   base::AutoLock lock(context_->lock());
 
-  if (context_->IsEmpty())
+  if (!context_->is_two_phase_read_in_progress())
     return UnexpectedError;

[hiroshige, 2015/06/18 07:01:20]

Can/should we call ClearQueue() in this then clause? (not sure)

[yhirano, 2015/06/18 08:13:33]

I do not want to do so, because it may complicate the code. Clearing the queue
is not mandatory in general: it should be cleared in the destruction process. I
added early-clear when closed / errored for performance. In L328 case, a user
does something wrong. I don't want to implement such optimization for bad users.

[yhirano, 2015/06/18 08:14:51]

Correction: I added early-clear when errored

[hiroshige, 2015/06/18 08:35:11]

OK.

 
-  context_->Consume(read_size);
+  context_->set_is_two_phase_read_in_progress(false);
+  if (context_->result() != Ok && context_->result() != Done) {
+    // We have an error, so we can discard the stored data.
+    context_->ClearQueue();
+  } else {
+    context_->Consume(read_size);
+  }
+
   return Ok;
 }
 
 SharedMemoryDataConsumerHandle::SharedMemoryDataConsumerHandle(
     BackpressureMode mode,
     scoped_ptr<Writer>* writer)
     : context_(new Context) {
   writer->reset(new Writer(context_, mode));
 }
 
@@ skipping 63 matching lines @@
       DCHECK(context_->IsReaderBoundToCurrentThread());
       needs_unlock = true;
     }
   }
   if (needs_unlock) {
     reader_.reset();
   }
 }
 
 }  // namespace content