Revert 214172 "Revert 214162 "Make Platform::queryStorageUsageAn..."

> Revert 214162 "Make Platform::queryStorageUsageAndQuota work fro..."
> 
> > Make Platform::queryStorageUsageAndQuota work from worker threads
> > 
> > For renderer/worker to browser side:
> > - Always use thread-local QuotaDispatcher
> > - Make QuotaDispatcher use ThreadSafeSender to send IPC msgs
> > - Keep ipc_thread_id:request_id mapping in QuotaMessageFilter
> > 
> > For browser to renderer/worker side:
> > - Add QuotaMessageFilter to receive quota IPC msgs and
> > to relay the msgs to the correct thread's QuotaDispatcher
> > 
> > BUG=259660
> > TEST=no behavioral changes yet (fast/workers/*storagequota* after blink side changes)
> > R=jam@chromium.org, michaeln@chromium.org
> > 
> > Review URL: https://codereview.chromium.org/20015002
> 
> TBR=kinuko@chromium.org
> 
> Review URL: https://codereview.chromium.org/21042002

TBR=kinuko@chromium.org

Review URL: https://codereview.chromium.org/21039004

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@214173 0039d316-1c4b-4281-b951-d872f2087c98

6 files changed, 204 insertions, 19 deletions

diff --git a/content/child/child_thread.cc b/content/child/child_thread.cc
index eefa4c5..d41c5ae 100644
--- a/content/child/child_thread.cc
+++ b/content/child/child_thread.cc
@@ -20,6 +20,7 @@
 #include "content/child/child_resource_message_filter.h"
 #include "content/child/fileapi/file_system_dispatcher.h"
 #include "content/child/quota_dispatcher.h"
+#include "content/child/quota_message_filter.h"
 #include "content/child/resource_dispatcher.h"
 #include "content/child/socket_stream_dispatcher.h"
 #include "content/child/thread_safe_sender.h"
@@ -129,24 +130,30 @@ void ChildThread::Init() {
   IPC::Logging::GetInstance()->SetIPCSender(this);
 #endif
 
-  resource_dispatcher_.reset(new ResourceDispatcher(this));
-  socket_stream_dispatcher_.reset(new SocketStreamDispatcher());
-  file_system_dispatcher_.reset(new FileSystemDispatcher());
-  quota_dispatcher_.reset(new QuotaDispatcher());
-
   sync_message_filter_ =
       new IPC::SyncMessageFilter(ChildProcess::current()->GetShutDownEvent());
   thread_safe_sender_ = new ThreadSafeSender(
       base::MessageLoopProxy::current().get(), sync_message_filter_.get());
+
+  resource_dispatcher_.reset(new ResourceDispatcher(this));
+  socket_stream_dispatcher_.reset(new SocketStreamDispatcher());
+  file_system_dispatcher_.reset(new FileSystemDispatcher());
+
   histogram_message_filter_ = new ChildHistogramMessageFilter();
   resource_message_filter_ =
       new ChildResourceMessageFilter(resource_dispatcher());
 
+  quota_message_filter_ =
+      new QuotaMessageFilter(thread_safe_sender_.get());
+  quota_dispatcher_.reset(new QuotaDispatcher(thread_safe_sender_.get(),
+                                              quota_message_filter_.get()));
+
   channel_->AddFilter(histogram_message_filter_.get());
   channel_->AddFilter(sync_message_filter_.get());
   channel_->AddFilter(new tracing::ChildTraceMessageFilter(
       ChildProcess::current()->io_message_loop_proxy()));
   channel_->AddFilter(resource_message_filter_.get());
+  channel_->AddFilter(quota_message_filter_.get());
 
 #if defined(OS_POSIX)
   // Check that --process-type is specified so we don't do this in unit tests
@@ -192,6 +199,7 @@ ChildThread::~ChildThread() {
   IPC::Logging::GetInstance()->SetIPCSender(NULL);
 #endif
 
+  channel_->RemoveFilter(quota_message_filter_.get());
   channel_->RemoveFilter(histogram_message_filter_.get());
   channel_->RemoveFilter(sync_message_filter_.get());
 
@@ -296,8 +304,6 @@ bool ChildThread::OnMessageReceived(const IPC::Message& msg) {
     return true;
   if (file_system_dispatcher_->OnMessageReceived(msg))
     return true;
-  if (quota_dispatcher_->OnMessageReceived(msg))
-    return true;
 
   bool handled = true;
   IPC_BEGIN_MESSAGE_MAP(ChildThread, msg)
diff --git a/content/child/child_thread.h b/content/child/child_thread.h
index 3b37e87..173fa2f 100644
--- a/content/child/child_thread.h
+++ b/content/child/child_thread.h
@@ -37,6 +37,7 @@ class ChildHistogramMessageFilter;
 class ChildResourceMessageFilter;
 class FileSystemDispatcher;
 class QuotaDispatcher;
+class QuotaMessageFilter;
 class ResourceDispatcher;
 class SocketStreamDispatcher;
 class ThreadSafeSender;
@@ -112,6 +113,10 @@ class CONTENT_EXPORT ChildThread : public IPC::Listener, public IPC::Sender {
     return histogram_message_filter_.get();
   }
 
+  QuotaMessageFilter* quota_message_filter() const {
+    return quota_message_filter_.get();
+  }
+
   base::MessageLoop* message_loop() const { return message_loop_; }
 
   // Returns the one child thread. Can only be called on the main thread.
@@ -188,6 +193,8 @@ class CONTENT_EXPORT ChildThread : public IPC::Listener, public IPC::Sender {
 
   scoped_refptr<ChildResourceMessageFilter> resource_message_filter_;
 
+  scoped_refptr<QuotaMessageFilter> quota_message_filter_;
+
   base::WeakPtrFactory<ChildThread> channel_connected_factory_;
 
   // Observes the trace event system. When tracing is enabled, optionally
diff --git a/content/child/quota_dispatcher.cc b/content/child/quota_dispatcher.cc
index 69bec35..2db0294 100644
--- a/content/child/quota_dispatcher.cc
+++ b/content/child/quota_dispatcher.cc
@@ -5,7 +5,11 @@
 #include "content/child/quota_dispatcher.h"
 
 #include "base/basictypes.h"
+#include "base/lazy_instance.h"
+#include "base/threading/thread_local.h"
 #include "content/child/child_thread.h"
+#include "content/child/quota_message_filter.h"
+#include "content/child/thread_safe_sender.h"
 #include "content/common/quota_messages.h"
 #include "third_party/WebKit/public/web/WebStorageQuotaCallbacks.h"
 #include "third_party/WebKit/public/web/WebStorageQuotaType.h"
@@ -18,7 +22,13 @@ using WebKit::WebStorageQuotaCallbacks;
 using WebKit::WebStorageQuotaError;
 using WebKit::WebStorageQuotaType;
 
+using webkit_glue::WorkerTaskRunner;
+
 namespace content {
+
+static base::LazyInstance<base::ThreadLocalPointer<QuotaDispatcher> >::Leaky
+    g_quota_dispatcher_tls = LAZY_INSTANCE_INITIALIZER;
+
 namespace {
 
 // QuotaDispatcher::Callback implementation for WebStorageQuotaCallbacks.
@@ -44,9 +54,17 @@ class WebStorageQuotaDispatcherCallback : public QuotaDispatcher::Callback {
   WebKit::WebStorageQuotaCallbacks* callbacks_;
 };
 
+int CurrentWorkerId() {
+  return WorkerTaskRunner::Instance()->CurrentWorkerId();
+}
+
 }  // namespace
 
-QuotaDispatcher::QuotaDispatcher() {
+QuotaDispatcher::QuotaDispatcher(ThreadSafeSender* thread_safe_sender,
+                                 QuotaMessageFilter* quota_message_filter)
+    : thread_safe_sender_(thread_safe_sender),
+      quota_message_filter_(quota_message_filter) {
+  g_quota_dispatcher_tls.Pointer()->Set(this);
 }
 
 QuotaDispatcher::~QuotaDispatcher() {
@@ -55,9 +73,28 @@ QuotaDispatcher::~QuotaDispatcher() {
     iter.GetCurrentValue()->DidFail(quota::kQuotaErrorAbort);
     iter.Advance();
   }
+
+  g_quota_dispatcher_tls.Pointer()->Set(NULL);
+}
+
+QuotaDispatcher* QuotaDispatcher::ThreadSpecificInstance(
+    ThreadSafeSender* thread_safe_sender,
+    QuotaMessageFilter* quota_message_filter) {
+  if (g_quota_dispatcher_tls.Pointer()->Get())
+    return g_quota_dispatcher_tls.Pointer()->Get();
+
+  QuotaDispatcher* dispatcher = new QuotaDispatcher(
+      thread_safe_sender, quota_message_filter);
+  if (WorkerTaskRunner::Instance()->CurrentWorkerId())
+    WorkerTaskRunner::Instance()->AddStopObserver(dispatcher);
+  return dispatcher;
+}
+
+void QuotaDispatcher::OnWorkerRunLoopStopped() {
+  delete this;
 }
 
-bool QuotaDispatcher::OnMessageReceived(const IPC::Message& msg) {
+void QuotaDispatcher::OnMessageReceived(const IPC::Message& msg) {
   bool handled = true;
   IPC_BEGIN_MESSAGE_MAP(QuotaDispatcher, msg)
     IPC_MESSAGE_HANDLER(QuotaMsg_DidGrantStorageQuota,
@@ -67,7 +104,7 @@ bool QuotaDispatcher::OnMessageReceived(const IPC::Message& msg) {
     IPC_MESSAGE_HANDLER(QuotaMsg_DidFail, DidFail);
     IPC_MESSAGE_UNHANDLED(handled = false)
   IPC_END_MESSAGE_MAP()
-  return handled;
+  DCHECK(handled) << "Unhandled message:" << msg.type();
 }
 
 void QuotaDispatcher::QueryStorageUsageAndQuota(
@@ -76,7 +113,8 @@ void QuotaDispatcher::QueryStorageUsageAndQuota(
     Callback* callback) {
   DCHECK(callback);
   int request_id = pending_quota_callbacks_.Add(callback);
-  ChildThread::current()->Send(new QuotaHostMsg_QueryStorageUsageAndQuota(
+  quota_message_filter_->RegisterRequestID(request_id, CurrentWorkerId());
+  thread_safe_sender_->Send(new QuotaHostMsg_QueryStorageUsageAndQuota(
       request_id, origin_url, type));
 }
 
@@ -87,8 +125,10 @@ void QuotaDispatcher::RequestStorageQuota(
     int64 requested_size,
     Callback* callback) {
   DCHECK(callback);
+  DCHECK(CurrentWorkerId() == 0);
   int request_id = pending_quota_callbacks_.Add(callback);
-  ChildThread::current()->Send(new QuotaHostMsg_RequestStorageQuota(
+  quota_message_filter_->RegisterRequestID(request_id, CurrentWorkerId());
+  thread_safe_sender_->Send(new QuotaHostMsg_RequestStorageQuota(
       render_view_id, request_id, origin_url, type, requested_size));
 }
 
diff --git a/content/child/quota_dispatcher.h b/content/child/quota_dispatcher.h
index a2fbfcc..8a8754f 100644
--- a/content/child/quota_dispatcher.h
+++ b/content/child/quota_dispatcher.h
@@ -10,7 +10,8 @@
 
 #include "base/basictypes.h"
 #include "base/id_map.h"
-#include "ipc/ipc_listener.h"
+#include "base/memory/ref_counted.h"
+#include "webkit/child/worker_task_runner.h"
 #include "webkit/common/quota/quota_types.h"
 
 class GURL;
@@ -25,10 +26,14 @@ class WebStorageQuotaCallbacks;
 
 namespace content {
 
+class ThreadSafeSender;
+class QuotaMessageFilter;
+
 // Dispatches and sends quota related messages sent to/from a child
 // process from/to the main browser process.  There is one instance
-// per child process.  Messages are dispatched on the main child thread.
-class QuotaDispatcher : public IPC::Listener {
+// per each thread.  Thread-specific instance can be obtained by
+// ThreadSpecificInstance().
+class QuotaDispatcher : public webkit_glue::WorkerTaskRunner::Observer {
  public:
   class Callback {
    public:
@@ -38,11 +43,20 @@ class QuotaDispatcher : public IPC::Listener {
     virtual void DidFail(quota::QuotaStatusCode status) = 0;
   };
 
-  QuotaDispatcher();
+  QuotaDispatcher(ThreadSafeSender* thread_safe_sender,
+                  QuotaMessageFilter* quota_message_filter);
   virtual ~QuotaDispatcher();
 
-  // IPC::Listener implementation.
-  virtual bool OnMessageReceived(const IPC::Message& msg) OVERRIDE;
+  // |thread_safe_sender| and |quota_message_filter| are used if
+  // calling this leads to construction.
+  static QuotaDispatcher* ThreadSpecificInstance(
+      ThreadSafeSender* thread_safe_sender,
+      QuotaMessageFilter* quota_message_filter);
+
+  // webkit_glue::WorkerTaskRunner::Observer implementation.
+  virtual void OnWorkerRunLoopStopped() OVERRIDE;
+
+  void OnMessageReceived(const IPC::Message& msg);
 
   void QueryStorageUsageAndQuota(const GURL& gurl,
                                  quota::StorageType type,
@@ -64,10 +78,14 @@ class QuotaDispatcher : public IPC::Listener {
                                     int64 current_quota);
   void DidGrantStorageQuota(int request_id,
                             int64 granted_quota);
-  void DidFail(int request_id, quota::QuotaStatusCode error);
+  void DidFail(int request_id,
+               quota::QuotaStatusCode error);
 
   IDMap<Callback, IDMapOwnPointer> pending_quota_callbacks_;
 
+  scoped_refptr<ThreadSafeSender> thread_safe_sender_;
+  scoped_refptr<QuotaMessageFilter> quota_message_filter_;
+
   DISALLOW_COPY_AND_ASSIGN(QuotaDispatcher);
 };
 
diff --git a/content/child/quota_message_filter.cc b/content/child/quota_message_filter.cc
new file mode 100644
index 0000000..db51239
--- /dev/null
+++ b/content/child/quota_message_filter.cc
@@ -0,0 +1,63 @@
+// Copyright 2013 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/quota_message_filter.h"
+
+#include "base/bind.h"
+#include "base/location.h"
+#include "base/message_loop/message_loop_proxy.h"
+#include "base/pickle.h"
+#include "content/child/quota_dispatcher.h"
+#include "content/child/thread_safe_sender.h"
+#include "content/common/quota_messages.h"
+#include "webkit/child/worker_task_runner.h"
+
+using webkit_glue::WorkerTaskRunner;
+
+namespace content {
+
+QuotaMessageFilter::QuotaMessageFilter(
+    ThreadSafeSender* thread_safe_sender)
+    : main_thread_loop_proxy_(base::MessageLoopProxy::current()),
+      thread_safe_sender_(thread_safe_sender) {
+}
+
+bool QuotaMessageFilter::OnMessageReceived(const IPC::Message& msg) {
+  if (IPC_MESSAGE_CLASS(msg) != QuotaMsgStart)
+    return false;
+  int request_id = -1;
+  bool result = PickleIterator(msg).ReadInt(&request_id);
+  DCHECK(result);
+  base::Closure closure = base::Bind(
+      &QuotaMessageFilter::DispatchMessage, this, msg);
+  int thread_id = 0;
+  {
+    base::AutoLock lock(request_id_map_lock_);
+    RequestIdToThreadId::iterator found = request_id_map_.find(request_id);
+    if (found != request_id_map_.end()) {
+      thread_id = found->second;
+      request_id_map_.erase(found);
+    }
+  }
+  if (!thread_id) {
+    main_thread_loop_proxy_->PostTask(FROM_HERE, closure);
+    return true;
+  }
+  WorkerTaskRunner::Instance()->PostTask(thread_id, closure);
+  return true;
+}
+
+void QuotaMessageFilter::RegisterRequestID(int request_id, int thread_id) {
+  base::AutoLock lock(request_id_map_lock_);
+  request_id_map_[request_id] = thread_id;
+}
+
+QuotaMessageFilter::~QuotaMessageFilter() {}
+
+void QuotaMessageFilter::DispatchMessage(const IPC::Message& msg) {
+  QuotaDispatcher::ThreadSpecificInstance(thread_safe_sender_.get(), this)
+      ->OnMessageReceived(msg);
+}
+
+}  // namespace content
diff --git a/content/child/quota_message_filter.h b/content/child/quota_message_filter.h
new file mode 100644
index 0000000..5e90c6c
--- /dev/null
+++ b/content/child/quota_message_filter.h
@@ -0,0 +1,51 @@
+// Copyright 2013 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.
+
+#ifndef CONTENT_CHILD_QUOTA_MESSAGE_FILTER_H_
+#define CONTENT_CHILD_QUOTA_MESSAGE_FILTER_H_
+
+#include <map>
+
+#include "base/synchronization/lock.h"
+#include "ipc/ipc_channel_proxy.h"
+
+namespace base {
+class MessageLoopProxy;
+}
+
+namespace content {
+
+class ThreadSafeSender;
+
+class QuotaMessageFilter : public IPC::ChannelProxy::MessageFilter {
+ public:
+  explicit QuotaMessageFilter(ThreadSafeSender* thread_safe_sender);
+
+  // IPC::Listener implementation.
+  virtual bool OnMessageReceived(const IPC::Message& msg) OVERRIDE;
+
+  // Registers { request_id, thread_id } map to the message filter.
+  // This method can be called on any thread.
+  void RegisterRequestID(int request_id, int thread_id);
+
+ protected:
+  virtual ~QuotaMessageFilter();
+
+ private:
+  typedef std::map<int, int> RequestIdToThreadId;
+
+  void DispatchMessage(const IPC::Message& msg);
+
+  scoped_refptr<base::MessageLoopProxy> main_thread_loop_proxy_;
+  scoped_refptr<ThreadSafeSender> thread_safe_sender_;
+
+  base::Lock request_id_map_lock_;
+  RequestIdToThreadId request_id_map_;
+
+  DISALLOW_COPY_AND_ASSIGN(QuotaMessageFilter);
+};
+
+}  // namespace content
+
+#endif  // CONTENT_CHILD_QUOTA_MESSAGE_FILTER_H_