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..87013ab037769d85ac44e09c5db6e24df4e62578
--- /dev/null
+++ b/content/renderer/indexed_db_message_filter.cc
@@ -0,0 +1,175 @@
+// 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/renderer/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"
+
+IndexedDBMessageFilter::IndexedDBMessageFilter() :
+    main_thread_loop_proxy_(base::MessageLoopProxy::current()) {
+  // This is called on the main thread.
+}
+
+IndexedDBMessageFilter::~IndexedDBMessageFilter() {
+  // This seems to never be called.
+}
+
+WorkerTaskRunner* worker_task_runner() {
+  return ChildThread::current()->worker_task_runner();
+}
+
+void IndexedDBMessageFilter::onLoopRegistered() {
+}
+
+void IndexedDBMessageFilter::onLoopUnregistered() {
+  if (idb_dispatcher_tls_.Get())
+    delete idb_dispatcher_tls_.Get();
+}
+
+int IndexedDBMessageFilter::current() {
+  return ChildThread::current()->worker_task_runner()->CurrentWorkerId();
+}
+
+int IndexedDBMessageFilter::current_worker_id() {
+  return ChildThread::current()->worker_task_runner()->CurrentWorkerId();
+}
+
+int ExtractThreadID(int callback_id) {
+  return (callback_id & 0xFF000000) >> 24;
+}
+
+int IndexedDBMessageFilter::InjectThreadID(int response_id) {
+  int32 thread_id = current();
+  DCHECK(thread_id <= 0xFF);

[michaeln, 2011/12/01 21:42:05]

(stashing this value in the high nibble :)

it looks like the classes in indexed_db_callbacks.h and
indexed_db_database_callbacks.h could work as a place on the server side to
stash the full integer value along side the 'response_id' value

[dgrogan, 2011/12/01 22:49:00]

High byte, but yeah.
I just didn't want to add the extra field to all the messages, handlers, and to
various classes on the browser side.  This is hacky, but it's probably good
enough for the matrix guys to test with (why we're rushing to get this into
m17.)  I added a TODO.

[michaeln, 2011/12/02 00:15:03]

Seems like it'd be fairly mechanical and limited to messages.h, dispatcher_host,
and the handful of callback classes mentioned earlier. The additional maps and
high-order bits and stuff are kind of gunky?

+  DCHECK(response_id <= 0x00FFFFFF);
+  return response_id | (thread_id << 24);
+}
+
+int IndexedDBMessageFilter::GetUniqueID() {
+  base::AutoLock locker(id_lock_);
+  int id = ++unique_id_;

[michaeln, 2011/12/01 21:42:05]

take a look at AtomicSequenceNumber

[dgrogan, 2011/12/01 22:49:00]

Ah nice, thanks for the pointer.

+  return InjectThreadID(id);
+}
+
+int ExtractCallbackID(const IPC::Message& msg) {
+  return IPC::MessageIterator(msg).NextInt();
+}
+
+bool IndexedDBMessageFilter::OnMessageReceived(const IPC::Message& msg) {
+  if (IPC_MESSAGE_CLASS(msg) != IndexedDBMsgStart)
+    return false;
+  int callback_id = ExtractCallbackID(msg);
+//  Forward to worker task runner
+  switch(msg.type()) {
+    case IndexedDBMsg_TransactionCallbacksAbort::ID:
+    case IndexedDBMsg_TransactionCallbacksComplete::ID:
+      ForwardToThread(msg, callback_id, transaction_id_to_run_loop_);
+      break;
+    case IndexedDBMsg_DatabaseCallbacksVersionChange::ID:
+      ForwardToThread(msg, callback_id, database_id_to_run_loop_);
+      break;
+    default:
+      ForwardToThread(msg, ExtractThreadID(callback_id));
+  }
+  return true;
+}
+
+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);
+}
+
+class MsgForwarderTask : public WebKit::WebWorkerRunLoop::Task {
+ public:
+  MsgForwarderTask(const IPC::Message& msg, IndexedDBMessageFilter* filter) :
+      msg_(msg),
+      filter_(filter) {
+  }
+  virtual ~MsgForwarderTask() { }
+  virtual void Run() {
+    IndexedDBDispatcher* dispatcher = filter_->thread_specific_idb_dispatcher();
+    dispatcher->OnMessageReceived(msg_);
+  }
+ private:
+  const IPC::Message msg_;
+  IndexedDBMessageFilter* filter_;
+};
+
+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::ForwardToThread(const IPC::Message& msg, int32 id,
+                                             const IDToRunLoopMap& map) {
+  base::AutoLock map_locker(map_lock_);
+  IDToRunLoopMap::const_iterator iter = map.find(id);
+  if (iter == map.end()) {
+    ForwardToMainThread(msg);
+    return;
+  }
+  ForwardToThread(msg, iter->second);
+}
+
+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::RemoveForType(int32 id, WebKit::WebIDBCallbacks*) {
+}
+
+void IndexedDBMessageFilter::RemoveForType(
+    int32 id,
+    WebKit::WebIDBDatabaseCallbacks*) {
+  RemoveFromMap(id, &database_id_to_run_loop_);
+}
+
+void IndexedDBMessageFilter::RemoveForType(
+    int32 id,
+    WebKit::WebIDBTransactionCallbacks*) {
+  RemoveFromMap(id, &transaction_id_to_run_loop_);
+}
+
+void IndexedDBMessageFilter::RemoveFromMap(int32 id, IDToRunLoopMap* map) {
+  // We don't store ids that are associated with the main thread.
+  if (!current())
+    return;
+  base::AutoLock map_locker(map_lock_);
+  int num_erased = map->erase(id);
+  DCHECK_EQ(num_erased, 1);
+}
+
+void IndexedDBMessageFilter::NotifyAddWithIDForType(
+    int32 id, WebKit::WebIDBTransactionCallbacks* callbacks) {
+  if (current()) {
+    base::AutoLock map_locker(map_lock_);
+    transaction_id_to_run_loop_[id] = current();
+  }
+}
+
+void IndexedDBMessageFilter::NotifyAddWithIDForType(int32 id,
+    WebKit::WebIDBDatabaseCallbacks* callbacks) {
+  if (current()) {
+    base::AutoLock map_locker(map_lock_);
+    database_id_to_run_loop_[id] = current();
+  }
+}