Dispatch IndexedDB IPC messages to worker threads

content/renderer/indexed_db_message_filter.cc

Index: content/renderer/indexed_db_message_filter.cc
diff --git a/content/renderer/indexed_db_message_filter.cc b/content/renderer/indexed_db_message_filter.cc
new file mode 100644
index 0000000000000000000000000000000000000000..20538607cd39e845059c598ac1bd367809fdb93e
--- /dev/null
+++ b/content/renderer/indexed_db_message_filter.cc
@@ -0,0 +1,145 @@
+// 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 "content/common/indexed_db_message_filter.h"
+
+#include "base/bind.h"
+#include "base/location.h"
+#include "content/common/indexed_db_messages.h"
+#include "content/common/worker_task_runner.h"
+#include "content/renderer/indexed_db_dispatcher.h"
+#include "content/renderer/render_thread_impl.h"
+
+namespace {
+
+WorkerTaskRunner* worker_task_runner() {
+  return ChildThread::current()->worker_task_runner();
+}
+
+int current_worker_id() {
+  return ChildThread::current()->worker_task_runner()->CurrentWorkerId();
+}
+
+}
+
+class IndexedDBMessageFilter::MsgForwarderTask :
+    public WebKit::WebWorkerRunLoop::Task {
+ public:
+  MsgForwarderTask(const IPC::Message& msg, IndexedDBMessageFilter* filter) :
+      msg_(msg),
+      filter_(filter) {
+  }
+  virtual ~MsgForwarderTask() { }
+  virtual void Run() {
+    filter_->DispatchMessage(msg_);
+  }
+ private:
+  const IPC::Message msg_;
+  IndexedDBMessageFilter* filter_;
+};
+
+IndexedDBMessageFilter::IndexedDBMessageFilter() :
+    main_thread_loop_proxy_(base::MessageLoopProxy::current()) {
+  // This is called on the main thread.
+  // We want our ids to start at 1, 0 is used to indicate the main thread.
+  DCHECK(!unique_id_.GetNext());
+}
+
+IndexedDBMessageFilter::~IndexedDBMessageFilter() {
+  // This seems to never be called.
+}
+
+bool IndexedDBMessageFilter::OnMessageReceived(const IPC::Message& msg) {
+  if (IPC_MESSAGE_CLASS(msg) != IndexedDBMsgStart)
+    return false;
+  int callback_id = ExtractCallbackID(msg);
+  switch(msg.type()) {
+    case IndexedDBMsg_TransactionCallbacksAbort::ID:
+    case IndexedDBMsg_TransactionCallbacksComplete::ID:
+      LookupAndForwardToThread(msg, callback_id, transaction_id_to_run_loop_);
+      break;
+    default:
+      ForwardToThread(msg, ExtractThreadID(callback_id));
+  }
+  return true;
+}
+
+IndexedDBDispatcher* IndexedDBMessageFilter::thread_specific_idb_dispatcher() {
+  if (idb_dispatcher_tls_.Get())
+    return idb_dispatcher_tls_.Get();
+
+  IndexedDBDispatcher* dispatcher = new IndexedDBDispatcher(this);
+  idb_dispatcher_tls_.Set(dispatcher);
+  return dispatcher;
+}
+
+void IndexedDBMessageFilter::onLoopRegistered() {
+}
+
+void IndexedDBMessageFilter::onLoopUnregistered() {
+  if (idb_dispatcher_tls_.Get())
+    delete idb_dispatcher_tls_.Get();
+}
+
+int IndexedDBMessageFilter::GetUniqueCallbackIDWithThreadID() {
+  int id = unique_id_.GetNext();
+  return InjectThreadID(id, current_worker_id());
+}
+
+void IndexedDBMessageFilter::DidRemoveTransactionID(int32 id) {
+  base::AutoLock map_locker(map_lock_);
+  int num_erased = transaction_id_to_run_loop_.erase(id);
+  DCHECK_EQ(num_erased, 1);
+}
+
+void IndexedDBMessageFilter::DidAddTransactionID(int32 id) {
+  base::AutoLock map_locker(map_lock_);
+  transaction_id_to_run_loop_[id] = current_worker_id();
+}
+
+void IndexedDBMessageFilter::LookupAndForwardToThread(
+    const IPC::Message& msg,
+    int32 callback_id,
+    const CallbackIDToRunLoopMap& map) {
+  base::AutoLock map_locker(map_lock_);
+  CallbackIDToRunLoopMap::const_iterator iter = map.find(callback_id);
+
+  DCHECK (iter != map.end());
+  ForwardToThread(msg, iter->second);
+}
+
+void IndexedDBMessageFilter::ForwardToThread(const IPC::Message& msg,
+                                             int thread_id) {
+  if (!thread_id) {
+    ForwardToMainThread(msg);
+    return;
+  }
+  MsgForwarderTask* task = new MsgForwarderTask(msg, this);
+  worker_task_runner()->PostTask(thread_id, task);
+}
+
+void IndexedDBMessageFilter::ForwardToMainThread(const IPC::Message& msg) {
+  main_thread_loop_proxy_->PostTask(FROM_HERE,
+      base::Bind(&IndexedDBMessageFilter::DispatchMessage, this, msg));
+}
+
+void IndexedDBMessageFilter::DispatchMessage(const IPC::Message& msg) {
+  thread_specific_idb_dispatcher()->OnMessageReceived(msg);
+}
+
+int IndexedDBMessageFilter::ExtractCallbackID(const IPC::Message& msg) {
+  return IPC::MessageIterator(msg).NextInt();
+}
+
+// TODO(dgrogan): Get rid of this bit-twiddling and instead put the entire
+// thread id in the outgoing and incoming IPCs.
+int IndexedDBMessageFilter::ExtractThreadID(int callback_id) {
+  return (callback_id & 0xFF000000) >> 24;
+}
+
+int IndexedDBMessageFilter::InjectThreadID(int callback_id, int thread_id) {
+  DCHECK(thread_id <= 0xFF);
+  DCHECK(callback_id <= 0x00FFFFFF);
+  return callback_id | (thread_id << 24);
+}