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..c2bf218aac021548afcaa894d29a1f2ad9947885
--- /dev/null
+++ b/content/renderer/indexed_db_message_filter.cc
@@ -0,0 +1,262 @@
+// 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/renderer/indexed_db_dispatcher.h"
+#include "content/renderer/render_thread_impl.h"
+#include "third_party/WebKit/Source/WebKit/chromium/public/WebWorkerRunLoop.h"
+
+using WebKit::WebWorkerRunLoop;
+
+IndexedDBMessageFilter::IndexedDBMessageFilter() :
+    WebCoreWorkerTracker(),
+    main_thread_loop_proxy_(base::MessageLoopProxy::current()),
+    main_thread_dispatcher_(new IndexedDBDispatcher(this)) {
+}
+
+void IndexedDBMessageFilter::DidStartWorkerRunLoop(
+    const WebWorkerRunLoop& run_loop) {
+  DCHECK(!tls_worker_run_loop_.Get());
+  base::AutoLock locker(run_loops_lock_);
+  std::pair<LoopSet::iterator, bool> result = run_loops_.insert(run_loop);
+  DCHECK(result.second);
+  // We're storing WebWorkerRunLoop in the set, which doesn't allow
+  // modification through its iterators so as not to invalidate the uniqueness
+  // constraint on the set.  But we've overloaded the comparison operators on
+  // WebWorkerRunLoop to ignore the non-const state.
+  WebWorkerRunLoop& a = const_cast<WebWorkerRunLoop&>(*result.first);
+  tls_worker_run_loop_.Set(&a);
+}
+
+void IndexedDBMessageFilter::DidStopWorkerRunLoop(
+    const WebWorkerRunLoop& run_loop) {
+  DCHECK(Contains(run_loop));
+  base::AutoLock locker(run_loops_lock_);
+  run_loops_.erase(run_loop);
+
+  DCHECK(tls_worker_run_loop_.Get());
+  tls_worker_run_loop_.Set(NULL);

[michaeln, 2011/11/30 21:48:14]

Looks like there can tbe other instances of WebWorkerRunLoop stashed in the
various id_maps that are not safe to use upon return from this method. What
prevents their use?

+}
+
+WebKit::WebWorkerRunLoop* IndexedDBMessageFilter::current() {
+  return tls_worker_run_loop_.Get();
+}
+
+bool IndexedDBMessageFilter::Contains(const WebWorkerRunLoop& run_loop) {
+  base::AutoLock locker(run_loops_lock_);
+  LoopSet::iterator iter = run_loops_.find(run_loop);
+  return iter != run_loops_.end();
+}
+
+int IndexedDBMessageFilter::GetUniqueID() {
+  base::AutoLock locker(id_lock_);
+  int id = ++unique_id_;
+  if (current()) {
+    base::AutoLock map_locker(map_lock_);
+    id_to_run_loop_[id] = current();
+  }
+  return id;
+}
+
+IndexedDBMessageFilter::~IndexedDBMessageFilter() {
+  // This is never called.
+  DCHECK_EQ(run_loops_.size(), 0UL);
+}
+
+bool IndexedDBMessageFilter::OnMessageReceived(const IPC::Message& msg) {
+  bool handled = true;
+  IPC_BEGIN_MESSAGE_MAP(IndexedDBMessageFilter, msg)
+    IPC_MESSAGE_HANDLER(IndexedDBMsg_CallbacksSuccessIDBCursor,
+                        OnSuccessOpenCursor)
+    IPC_MESSAGE_HANDLER(IndexedDBMsg_CallbacksSuccessCursorContinue,
+                        OnSuccessCursorContinue)
+    IPC_MESSAGE_HANDLER(IndexedDBMsg_CallbacksSuccessIDBDatabase,
+                        OnSuccessIDBDatabase)
+    IPC_MESSAGE_HANDLER(IndexedDBMsg_CallbacksSuccessIndexedDBKey,
+                        OnSuccessIndexedDBKey)
+    IPC_MESSAGE_HANDLER(IndexedDBMsg_CallbacksSuccessIDBTransaction,
+                        OnSuccessIDBTransaction)
+    IPC_MESSAGE_HANDLER(IndexedDBMsg_CallbacksSuccessStringList,
+                        OnSuccessStringList)
+    IPC_MESSAGE_HANDLER(IndexedDBMsg_CallbacksSuccessSerializedScriptValue,
+                        OnSuccessSerializedScriptValue)
+    IPC_MESSAGE_HANDLER(IndexedDBMsg_CallbacksError, OnError)
+    IPC_MESSAGE_HANDLER(IndexedDBMsg_CallbacksBlocked, OnBlocked)
+    IPC_MESSAGE_HANDLER(IndexedDBMsg_TransactionCallbacksAbort, OnAbort)
+    IPC_MESSAGE_HANDLER(IndexedDBMsg_TransactionCallbacksComplete, OnComplete)
+    IPC_MESSAGE_HANDLER(IndexedDBMsg_DatabaseCallbacksVersionChange,
+                        OnVersionChange)
+    IPC_MESSAGE_UNHANDLED(handled = false)
+  IPC_END_MESSAGE_MAP()
+  return handled;
+}
+
+void IndexedDBMessageFilter::ForwardToMainThread(const IPC::Message& msg) {
+  main_thread_loop_proxy_->PostTask(FROM_HERE,
+      base::Bind(&IndexedDBDispatcher::OnMessageReceived,
+                 main_thread_dispatcher_, 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, int32 id,
+                                             const IDToRunLoop& map) {
+  IDToRunLoop::const_iterator iter = map.find(id);
+  if (iter == map.end()) {
+    ForwardToMainThread(msg);

[michaeln, 2011/11/30 21:48:14]

Do we need a stronger test for whether or not the msg should be delivered to a
worker? what if the worker exited prior to a callback, it wouldn't be found in
the map, and it shouldn't be delivered to the main thread.

[dgrogan, 2011/11/30 23:25:18]

As it stands now these maps aren't cleaned up, so in the case you describe the
message would be forwarded to a dead runloop.

But assuming that these maps are cleaned up when a worker shuts down, you're
right, this would be bad.  Tracking main-thread object ids in a separate data
structure is one option.  Another would be to store the id of the run loop, and
push the task over to WorkerTaskRunner.  The main thread could have its own
magic constant.

+    return;
+  }
+  MsgForwarderTask* task = new MsgForwarderTask(msg, this);
+  iter->second->postTask(task);

[michaeln, 2011/11/30 21:48:14]

Should this be under the run_loops_lock_? What guarantee is there that the loop
isn't exiting?

[dgrogan, 2011/11/30 23:25:18]

map_lock_, but yes.  Fixed.
I don't think there is one.  If the runloop is exiting I don't think it matters
if this message goes undelivered.  But after looking at WorkerRunLoop and
WTF::MessageQueue, I'm not sure the tasks's memory gets reclaimed.  I should
probably deal with that in webkit.

+}
+
+void IndexedDBMessageFilter::OnSuccessIDBDatabase(const IPC::Message& msg,
+                                                  int32 response_id,
+                                                  int32 object_id) {
+  ForwardToThread(msg, response_id, id_to_run_loop_);
+}
+
+void IndexedDBMessageFilter::OnSuccessIndexedDBKey(const IPC::Message& msg,
+                                                   int32 response_id,
+                                                   const IndexedDBKey& key) {
+  ForwardToThread(msg, response_id, id_to_run_loop_);
+}
+
+void IndexedDBMessageFilter::OnSuccessIDBTransaction(const IPC::Message& msg,
+                                                     int32 response_id,
+                                                     int32 object_id) {
+  ForwardToThread(msg, response_id, id_to_run_loop_);
+}
+
+void IndexedDBMessageFilter::OnSuccessStringList(const IPC::Message& msg,
+    int32 response_id, const std::vector<string16>& value) {
+  ForwardToThread(msg, response_id, id_to_run_loop_);
+}
+
+void IndexedDBMessageFilter::OnSuccessSerializedScriptValue(
+    const IPC::Message& msg, int32 response_id,
+    const content::SerializedScriptValue& value) {
+  ForwardToThread(msg, response_id, id_to_run_loop_);
+}
+
+void IndexedDBMessageFilter::OnSuccessOpenCursor(const IPC::Message& msg,
+    int32 response_id, int32 object_id, const IndexedDBKey& key,
+    const IndexedDBKey& primaryKey,
+    const content::SerializedScriptValue& value) {
+  ForwardToThread(msg, response_id, id_to_run_loop_);
+}
+
+void IndexedDBMessageFilter::OnSuccessCursorContinue(const IPC::Message& msg,
+    int32 response_id,
+                             int32 cursor_id,
+                             const IndexedDBKey& key,
+                             const IndexedDBKey& primary_key,
+                             const content::SerializedScriptValue& value) {
+  ForwardToThread(msg, response_id, id_to_run_loop_);
+}
+
+void IndexedDBMessageFilter::OnBlocked(const IPC::Message& msg,
+                                       int32 response_id) {
+  ForwardToThread(msg, response_id, id_to_run_loop_);
+}
+
+void IndexedDBMessageFilter::OnError(const IPC::Message& msg,
+                                     int32 response_id,
+                                     int code,
+                                     const string16& message) {
+  ForwardToThread(msg, response_id, id_to_run_loop_);
+}
+
+void IndexedDBMessageFilter::OnAbort(const IPC::Message& msg,
+                                     int32 transaction_id) {
+  ForwardToThread(msg, transaction_id, transaction_id_to_run_loop_);
+}
+
+void IndexedDBMessageFilter::OnComplete(const IPC::Message& msg,
+                                        int32 transaction_id) {
+  ForwardToThread(msg, transaction_id, transaction_id_to_run_loop_);
+}
+
+void IndexedDBMessageFilter::OnVersionChange(const IPC::Message& msg,
+                                             int32 database_id,
+                                             const string16& newVersion) {
+  ForwardToThread(msg, database_id, database_id_to_run_loop_);
+}
+
+IndexedDBDispatcher* IndexedDBMessageFilter::thread_specific_idb_dispatcher() {
+  WebWorkerRunLoop* current_loop = current();
+
+  if (current_loop == NULL) {
+    // If we're not on a worker thread we should be on the main thread.
+    DCHECK(RenderThreadImpl::current());
+    DCHECK(main_thread_dispatcher_.get());
+    return main_thread_dispatcher_.get();
+  }
+  DCHECK(Contains(*current_loop));
+  base::AutoLock lock(dispatchers_lock_);
+  LoopToDispatcherMap::iterator iter = dispatchers_.find(current_loop);
+  if (iter == dispatchers_.end()) {
+    dispatchers_[current_loop] = new IndexedDBDispatcher(this);

[michaeln, 2011/11/30 21:48:14]

where is this map cleaned up?

[dgrogan, 2011/11/30 23:25:18]

It appears that it isn't.  I had intended to clean it up in
DidStopWorkerRunLoop, which will be complicated by the move of those methods to
WorkerTaskRunner.  Maybe WebKitPlatformSupport will notify an observer list when
worker runloops are started and stopped.

It looks like I can get rid of dispatchers_ and dispatchers_lock_ in favor of a
TLS holding the current IDBDispatcher.  This class would still have to listen
for RunLoops stopping to deallocate a stopping-worker's IndexedDBDispatcher
object.

+  }
+  return dispatchers_[current_loop];
+}
+
+void IndexedDBMessageFilter::RemoveForType(int32 id, WebKit::WebIDBCallbacks*) {
+  RemoveFromMap(id, &id_to_run_loop_);
+}
+
+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, IDToRunLoop* 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();
+  }
+}