Dispatch IndexedDB IPC messages to worker threads

content/renderer/indexed_db_dispatcher.cc

 // 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_dispatcher.h"
 
+#include "base/lazy_instance.h"
 #include "content/common/indexed_db_messages.h"
 #include "content/renderer/render_thread_impl.h"
 #include "content/renderer/render_view_impl.h"
 #include "content/renderer/renderer_webidbcursor_impl.h"
 #include "content/renderer/renderer_webidbdatabase_impl.h"
 #include "content/renderer/renderer_webidbindex_impl.h"
 #include "content/renderer/renderer_webidbobjectstore_impl.h"
 #include "content/renderer/renderer_webidbtransaction_impl.h"
+#include "ipc/ipc_channel.h"
 #include "third_party/WebKit/Source/WebKit/chromium/public/WebFrame.h"
 #include "third_party/WebKit/Source/WebKit/chromium/public/WebIDBDatabaseCallbacks.h"
 #include "third_party/WebKit/Source/WebKit/chromium/public/WebIDBDatabaseError.h"
 #include "third_party/WebKit/Source/WebKit/chromium/public/WebIDBKeyRange.h"
-#include "third_party/WebKit/Source/WebKit/chromium/public/WebSecurityOrigin.h"
-#include "third_party/WebKit/Source/WebKit/chromium/public/platform/WebSerializedScriptValue.h"
-#include "third_party/WebKit/Source/WebKit/chromium/public/platform/WebString.h"
 
+using base::ThreadLocalPointer;
 using WebKit::WebDOMStringList;
 using WebKit::WebExceptionCode;
 using WebKit::WebFrame;
 using WebKit::WebIDBCallbacks;
 using WebKit::WebIDBKeyRange;
 using WebKit::WebIDBDatabase;
 using WebKit::WebIDBDatabaseCallbacks;
 using WebKit::WebIDBDatabaseError;
 using WebKit::WebIDBTransaction;
 using WebKit::WebIDBTransactionCallbacks;
+using webkit_glue::WorkerTaskRunner;
+
+static base::LazyInstance<ThreadLocalPointer<IndexedDBDispatcher>,
+    base::LeakyLazyInstanceTraits<ThreadLocalPointer<IndexedDBDispatcher> > >
+        g_idb_dispatcher_tls = LAZY_INSTANCE_INITIALIZER;
+
+namespace {
+
+int32 CurrentWorkerId() {
+  return WorkerTaskRunner::Instance()->CurrentWorkerId();
+}
+
+} // unnamed namespace
 
 IndexedDBDispatcher::IndexedDBDispatcher() {
 }
 
 IndexedDBDispatcher::~IndexedDBDispatcher() {
 }
 
-bool IndexedDBDispatcher::OnMessageReceived(const IPC::Message& msg) {
+IndexedDBDispatcher* IndexedDBDispatcher::ThreadSpecificInstance() {
+  if (g_idb_dispatcher_tls.Pointer()->Get())
+    return g_idb_dispatcher_tls.Pointer()->Get();
+
+  IndexedDBDispatcher* dispatcher = new IndexedDBDispatcher;
+  g_idb_dispatcher_tls.Pointer()->Set(dispatcher);
+  if (WorkerTaskRunner::Instance()->CurrentWorkerId())
+    webkit_glue::WorkerTaskRunner::Instance()->AddStopObserver(dispatcher);
+  return dispatcher;
+}
+
+void IndexedDBDispatcher::OnWorkerRunLoopStopped() {
+  g_idb_dispatcher_tls.Pointer()->Set(NULL);

[michaeln, 2011/12/15 02:20:48]

consider putting this line in the dtor so when the main thread's instance is
deleted, the tls ptr also gets reset for that instance (as it is now, the tls
ptr for that guy is left dangling).

and if the reset goes in the dtor, consider putting the Set(tosomething) call in
the ctor for the sake of symmetry

+  delete this;
+}
+
+void IndexedDBDispatcher::OnMessageReceived(const IPC::Message& msg) {
   bool handled = true;
   IPC_BEGIN_MESSAGE_MAP(IndexedDBDispatcher, msg)
     IPC_MESSAGE_HANDLER(IndexedDBMsg_CallbacksSuccessIDBCursor,
                         OnSuccessOpenCursor)
     IPC_MESSAGE_HANDLER(IndexedDBMsg_CallbacksSuccessCursorContinue,
                         OnSuccessCursorContinue)
     IPC_MESSAGE_HANDLER(IndexedDBMsg_CallbacksSuccessCursorPrefetch,
                         OnSuccessCursorPrefetch)
     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;
+  // If a message gets here, IndexedDBMessageFilter already determined that it
+  // is an IndexedDB message.
+  DCHECK(handled);
 }
 
 void IndexedDBDispatcher::Send(IPC::Message* msg) {
   ChildThread::current()->Send(msg);
 }
 
 void IndexedDBDispatcher::RequestIDBCursorUpdate(
     const content::SerializedScriptValue& value,
     WebIDBCallbacks* callbacks_ptr,
     int32 idb_cursor_id,
     WebExceptionCode* ec) {
   ResetCursorPrefetchCaches();
   scoped_ptr<WebIDBCallbacks> callbacks(callbacks_ptr);
 
   int32 response_id = pending_callbacks_.Add(callbacks.release());
   Send(
-      new IndexedDBHostMsg_CursorUpdate(idb_cursor_id, response_id, value, ec));
+      new IndexedDBHostMsg_CursorUpdate(idb_cursor_id, CurrentWorkerId(),
+                                        response_id, value, ec));
   if (*ec)
     pending_callbacks_.Remove(response_id);
 }
 
 void IndexedDBDispatcher::RequestIDBCursorContinue(
     const IndexedDBKey& key,
     WebIDBCallbacks* callbacks_ptr,
     int32 idb_cursor_id,
     WebExceptionCode* ec) {
   // Reset all cursor prefetch caches except for this cursor.
   ResetCursorPrefetchCaches(idb_cursor_id);
 
   scoped_ptr<WebIDBCallbacks> callbacks(callbacks_ptr);
 
   int32 response_id = pending_callbacks_.Add(callbacks.release());
   Send(
-      new IndexedDBHostMsg_CursorContinue(idb_cursor_id, response_id, key, ec));
+      new IndexedDBHostMsg_CursorContinue(idb_cursor_id, CurrentWorkerId(),
+                                          response_id, key, ec));
   if (*ec)
     pending_callbacks_.Remove(response_id);
 }
 
 void IndexedDBDispatcher::RequestIDBCursorPrefetch(
     int n,
     WebIDBCallbacks* callbacks_ptr,
     int32 idb_cursor_id,
     WebExceptionCode* ec) {
   scoped_ptr<WebIDBCallbacks> callbacks(callbacks_ptr);
 
   int32 response_id = pending_callbacks_.Add(callbacks.release());
-  RenderThreadImpl::current()->Send(
-      new IndexedDBHostMsg_CursorPrefetch(idb_cursor_id, response_id, n, ec));
+  Send(new IndexedDBHostMsg_CursorPrefetch(idb_cursor_id, CurrentWorkerId(),
+                                           response_id, n, ec));
   if (*ec)
     pending_callbacks_.Remove(response_id);
 }
 
 void IndexedDBDispatcher::RequestIDBCursorPrefetchReset(
     int used_prefetches, int unused_prefetches, int32 idb_cursor_id) {
-  RenderThreadImpl::current()->Send(
-      new IndexedDBHostMsg_CursorPrefetchReset(idb_cursor_id, used_prefetches,
-                                               unused_prefetches));
+  Send(new IndexedDBHostMsg_CursorPrefetchReset(idb_cursor_id,
+                                                used_prefetches,
+                                                unused_prefetches));
 }
 
 void IndexedDBDispatcher::RequestIDBCursorDelete(
     WebIDBCallbacks* callbacks_ptr,
     int32 idb_cursor_id,
     WebExceptionCode* ec) {
   ResetCursorPrefetchCaches();
   scoped_ptr<WebIDBCallbacks> callbacks(callbacks_ptr);
 
   int32 response_id = pending_callbacks_.Add(callbacks.release());
-  Send(new IndexedDBHostMsg_CursorDelete(idb_cursor_id, response_id, ec));
+  Send(new IndexedDBHostMsg_CursorDelete(idb_cursor_id, CurrentWorkerId(),
+                                         response_id, ec));
   if (*ec)
     pending_callbacks_.Remove(response_id);
 }
 
 void IndexedDBDispatcher::RequestIDBFactoryOpen(
     const string16& name,
     WebIDBCallbacks* callbacks_ptr,
     const string16& origin,
     WebFrame* web_frame) {
   ResetCursorPrefetchCaches();
   scoped_ptr<WebIDBCallbacks> callbacks(callbacks_ptr);
 
   if (!web_frame)
     return; // We must be shutting down.
   RenderViewImpl* render_view = RenderViewImpl::FromWebView(web_frame->view());
   if (!render_view)
     return; // We must be shutting down.
 
   IndexedDBHostMsg_FactoryOpen_Params params;
+  params.thread_id = CurrentWorkerId();
   params.response_id = pending_callbacks_.Add(callbacks.release());
   params.origin = origin;
   params.name = name;
   Send(new IndexedDBHostMsg_FactoryOpen(params));
 }
 
 void IndexedDBDispatcher::RequestIDBFactoryGetDatabaseNames(
     WebIDBCallbacks* callbacks_ptr,
     const string16& origin,
     WebFrame* web_frame) {
   ResetCursorPrefetchCaches();
   scoped_ptr<WebIDBCallbacks> callbacks(callbacks_ptr);
 
   if (!web_frame)
     return; // We must be shutting down.
   RenderViewImpl* render_view = RenderViewImpl::FromWebView(web_frame->view());
   if (!render_view)
     return; // We must be shutting down.
 
   IndexedDBHostMsg_FactoryGetDatabaseNames_Params params;
+  params.thread_id = CurrentWorkerId();
   params.response_id = pending_callbacks_.Add(callbacks.release());
   params.origin = origin;
   Send(new IndexedDBHostMsg_FactoryGetDatabaseNames(params));
 }
 
 void IndexedDBDispatcher::RequestIDBFactoryDeleteDatabase(
     const string16& name,
     WebIDBCallbacks* callbacks_ptr,
     const string16& origin,
     WebFrame* web_frame) {
   ResetCursorPrefetchCaches();
   scoped_ptr<WebIDBCallbacks> callbacks(callbacks_ptr);
 
   if (!web_frame)
     return; // We must be shutting down.
   RenderViewImpl* render_view = RenderViewImpl::FromWebView(web_frame->view());
   if (!render_view)
     return; // We must be shutting down.
 
   IndexedDBHostMsg_FactoryDeleteDatabase_Params params;
+  params.thread_id = CurrentWorkerId();
   params.response_id = pending_callbacks_.Add(callbacks.release());
   params.origin = origin;
   params.name = name;
   Send(new IndexedDBHostMsg_FactoryDeleteDatabase(params));
 }
 
 void IndexedDBDispatcher::RequestIDBDatabaseClose(int32 idb_database_id) {
   ResetCursorPrefetchCaches();
   Send(new IndexedDBHostMsg_DatabaseClose(idb_database_id));
   pending_database_callbacks_.Remove(idb_database_id);
 }
 
 void IndexedDBDispatcher::RequestIDBDatabaseOpen(
       WebIDBDatabaseCallbacks* callbacks_ptr,
       int32 idb_database_id) {
   ResetCursorPrefetchCaches();
   scoped_ptr<WebIDBDatabaseCallbacks> callbacks(callbacks_ptr);
 
-  int32 response_id = pending_database_callbacks_.Add(callbacks.release());
-  Send(new IndexedDBHostMsg_DatabaseOpen(response_id, idb_database_id));
+  DCHECK(!pending_database_callbacks_.Lookup(idb_database_id));
+  pending_database_callbacks_.AddWithID(callbacks.release(), idb_database_id);
+  Send(new IndexedDBHostMsg_DatabaseOpen(idb_database_id, CurrentWorkerId(),
+                                         idb_database_id));
 }
 
 void IndexedDBDispatcher::RequestIDBDatabaseSetVersion(
     const string16& version,
     WebIDBCallbacks* callbacks_ptr,
     int32 idb_database_id,
     WebExceptionCode* ec) {
   ResetCursorPrefetchCaches();
   scoped_ptr<WebIDBCallbacks> callbacks(callbacks_ptr);
 
   int32 response_id = pending_callbacks_.Add(callbacks.release());
-  Send(new IndexedDBHostMsg_DatabaseSetVersion(idb_database_id, response_id,
-                                               version, ec));
+  Send(new IndexedDBHostMsg_DatabaseSetVersion(idb_database_id,
+                                               CurrentWorkerId(),
+                                               response_id, version, ec));
   if (*ec)
     pending_callbacks_.Remove(response_id);
 }
 
 void IndexedDBDispatcher::RequestIDBIndexOpenObjectCursor(
     const WebIDBKeyRange& idb_key_range,
     unsigned short direction,
     WebIDBCallbacks* callbacks_ptr,
     int32 idb_index_id,
     const WebIDBTransaction& transaction,
     WebExceptionCode* ec) {
   ResetCursorPrefetchCaches();
   scoped_ptr<WebIDBCallbacks> callbacks(callbacks_ptr);
   IndexedDBHostMsg_IndexOpenCursor_Params params;
+  params.thread_id = CurrentWorkerId();
   params.response_id = pending_callbacks_.Add(callbacks.release());
   params.lower_key.Set(idb_key_range.lower());
   params.upper_key.Set(idb_key_range.upper());
   params.lower_open = idb_key_range.lowerOpen();
   params.upper_open = idb_key_range.upperOpen();
   params.direction = direction;
   params.idb_index_id = idb_index_id;
   params.transaction_id = TransactionId(transaction);
   Send(new IndexedDBHostMsg_IndexOpenObjectCursor(params, ec));
   if (*ec)
     pending_callbacks_.Remove(params.response_id);
 }
 
 void IndexedDBDispatcher::RequestIDBIndexOpenKeyCursor(
     const WebIDBKeyRange& idb_key_range,
     unsigned short direction,
     WebIDBCallbacks* callbacks_ptr,
     int32 idb_index_id,
     const WebIDBTransaction& transaction,
     WebExceptionCode* ec) {
   ResetCursorPrefetchCaches();
   scoped_ptr<WebIDBCallbacks> callbacks(callbacks_ptr);
   IndexedDBHostMsg_IndexOpenCursor_Params params;
+  params.thread_id = CurrentWorkerId();
   params.response_id = pending_callbacks_.Add(callbacks.release());
   // TODO(jorlow): We really should just create a Chromium abstraction for
   //               KeyRange rather than doing it ad-hoc like this.
   params.lower_key.Set(idb_key_range.lower());
   params.upper_key.Set(idb_key_range.upper());
   params.lower_open = idb_key_range.lowerOpen();
   params.upper_open = idb_key_range.upperOpen();
   params.direction = direction;
   params.idb_index_id = idb_index_id;
   params.transaction_id = TransactionId(transaction);
   Send(new IndexedDBHostMsg_IndexOpenKeyCursor(params, ec));
   if (*ec)
     pending_callbacks_.Remove(params.response_id);
 }
 
 void IndexedDBDispatcher::RequestIDBIndexGetObject(
     const IndexedDBKey& key,
     WebIDBCallbacks* callbacks_ptr,
     int32 idb_index_id,
     const WebIDBTransaction& transaction,
     WebExceptionCode* ec) {
   ResetCursorPrefetchCaches();
   scoped_ptr<WebIDBCallbacks> callbacks(callbacks_ptr);
   int32 response_id = pending_callbacks_.Add(callbacks.release());
-  Send(new IndexedDBHostMsg_IndexGetObject(idb_index_id, response_id, key,
+  Send(new IndexedDBHostMsg_IndexGetObject(idb_index_id, CurrentWorkerId(),
+                                           response_id, key,
                                            TransactionId(transaction), ec));
   if (*ec)
     pending_callbacks_.Remove(response_id);
 }
 
 void IndexedDBDispatcher::RequestIDBIndexGetKey(
     const IndexedDBKey& key,
     WebIDBCallbacks* callbacks_ptr,
     int32 idb_index_id,
     const WebIDBTransaction& transaction,
     WebExceptionCode* ec) {
   ResetCursorPrefetchCaches();
   scoped_ptr<WebIDBCallbacks> callbacks(callbacks_ptr);
   int32 response_id = pending_callbacks_.Add(callbacks.release());
   Send(new IndexedDBHostMsg_IndexGetKey(
-           idb_index_id, response_id, key,
+           idb_index_id, CurrentWorkerId(), response_id, key,
            TransactionId(transaction), ec));
   if (*ec)
     pending_callbacks_.Remove(response_id);
 }
 
 void IndexedDBDispatcher::RequestIDBObjectStoreGet(
     const IndexedDBKey& key,
     WebIDBCallbacks* callbacks_ptr,
     int32 idb_object_store_id,
     const WebIDBTransaction& transaction,
     WebExceptionCode* ec) {
   ResetCursorPrefetchCaches();
   scoped_ptr<WebIDBCallbacks> callbacks(callbacks_ptr);
 
   int32 response_id = pending_callbacks_.Add(callbacks.release());
   Send(new IndexedDBHostMsg_ObjectStoreGet(
-           idb_object_store_id, response_id,
+           idb_object_store_id, CurrentWorkerId(), response_id,
            key, TransactionId(transaction), ec));
   if (*ec)
     pending_callbacks_.Remove(response_id);
 }
 
 void IndexedDBDispatcher::RequestIDBObjectStorePut(
     const content::SerializedScriptValue& value,
     const IndexedDBKey& key,
     WebKit::WebIDBObjectStore::PutMode put_mode,
     WebIDBCallbacks* callbacks_ptr,
     int32 idb_object_store_id,
     const WebIDBTransaction& transaction,
     WebExceptionCode* ec) {
   ResetCursorPrefetchCaches();
   scoped_ptr<WebIDBCallbacks> callbacks(callbacks_ptr);
   IndexedDBHostMsg_ObjectStorePut_Params params;
+  params.thread_id = CurrentWorkerId();
   params.idb_object_store_id = idb_object_store_id;
   params.response_id = pending_callbacks_.Add(callbacks.release());
   params.serialized_value = value;
   params.key = key;
   params.put_mode = put_mode;
   params.transaction_id = TransactionId(transaction);
   Send(new IndexedDBHostMsg_ObjectStorePut(params, ec));
   if (*ec)
     pending_callbacks_.Remove(params.response_id);
 }
 
 void IndexedDBDispatcher::RequestIDBObjectStoreDelete(
     const IndexedDBKey& key,
     WebIDBCallbacks* callbacks_ptr,
     int32 idb_object_store_id,
     const WebIDBTransaction& transaction,
     WebExceptionCode* ec) {
   ResetCursorPrefetchCaches();
   scoped_ptr<WebIDBCallbacks> callbacks(callbacks_ptr);
 
   int32 response_id = pending_callbacks_.Add(callbacks.release());
   Send(new IndexedDBHostMsg_ObjectStoreDelete(
-      idb_object_store_id, response_id, key, TransactionId(transaction), ec));
+      idb_object_store_id, CurrentWorkerId(), response_id, key,
+      TransactionId(transaction), ec));
   if (*ec)
     pending_callbacks_.Remove(response_id);
 }
 
 void IndexedDBDispatcher::RequestIDBObjectStoreClear(
     WebIDBCallbacks* callbacks_ptr,
     int32 idb_object_store_id,
     const WebIDBTransaction& transaction,
     WebExceptionCode* ec) {
   ResetCursorPrefetchCaches();
   scoped_ptr<WebIDBCallbacks> callbacks(callbacks_ptr);
 
   int32 response_id = pending_callbacks_.Add(callbacks.release());
   Send(new IndexedDBHostMsg_ObjectStoreClear(
-      idb_object_store_id, response_id, TransactionId(transaction), ec));
+      idb_object_store_id, CurrentWorkerId(), response_id,
+      TransactionId(transaction), ec));
   if (*ec)
     pending_callbacks_.Remove(response_id);
 }
 
 void IndexedDBDispatcher::RequestIDBObjectStoreOpenCursor(
     const WebIDBKeyRange& idb_key_range,
     unsigned short direction,
     WebIDBCallbacks* callbacks_ptr,
     int32 idb_object_store_id,
     const WebIDBTransaction& transaction,
     WebExceptionCode* ec) {
   ResetCursorPrefetchCaches();
   scoped_ptr<WebIDBCallbacks> callbacks(callbacks_ptr);
   IndexedDBHostMsg_ObjectStoreOpenCursor_Params params;
+  params.thread_id = CurrentWorkerId();
   params.response_id = pending_callbacks_.Add(callbacks.release());
   params.lower_key.Set(idb_key_range.lower());
   params.upper_key.Set(idb_key_range.upper());
   params.lower_open = idb_key_range.lowerOpen();
   params.upper_open = idb_key_range.upperOpen();
   params.direction = direction;
   params.idb_object_store_id = idb_object_store_id;
   params.transaction_id = TransactionId(transaction);
   Send(new IndexedDBHostMsg_ObjectStoreOpenCursor(params, ec));
   if (*ec)
@@ skipping 10 matching lines @@
   cursors_.erase(cursor_id);
 }
 
 int32 IndexedDBDispatcher::TransactionId(
     const WebIDBTransaction& transaction) {
   const RendererWebIDBTransactionImpl* impl =
       static_cast<const RendererWebIDBTransactionImpl*>(&transaction);
   return impl->id();
 }
 
-void IndexedDBDispatcher::OnSuccessIDBDatabase(int32 response_id,
+void IndexedDBDispatcher::OnSuccessIDBDatabase(int32 thread_id,
+                                               int32 response_id,
                                                int32 object_id) {
+  DCHECK_EQ(thread_id, CurrentWorkerId());
   WebIDBCallbacks* callbacks = pending_callbacks_.Lookup(response_id);
   if (!callbacks)
     return;
   callbacks->onSuccess(new RendererWebIDBDatabaseImpl(object_id));
   pending_callbacks_.Remove(response_id);
 }
 
-void IndexedDBDispatcher::OnSuccessIndexedDBKey(int32 response_id,
+void IndexedDBDispatcher::OnSuccessIndexedDBKey(int32 thread_id,
+                                                int32 response_id,
                                                 const IndexedDBKey& key) {
+  DCHECK_EQ(thread_id, CurrentWorkerId());
   WebIDBCallbacks* callbacks = pending_callbacks_.Lookup(response_id);
   if (!callbacks)
     return;
   callbacks->onSuccess(key);
   pending_callbacks_.Remove(response_id);
 }
 
-void IndexedDBDispatcher::OnSuccessIDBTransaction(int32 response_id,
+void IndexedDBDispatcher::OnSuccessIDBTransaction(int32 thread_id,
+                                                  int32 response_id,
                                                   int32 object_id) {
+  DCHECK_EQ(thread_id, CurrentWorkerId());
   WebIDBCallbacks* callbacks = pending_callbacks_.Lookup(response_id);
   if (!callbacks)
     return;
   callbacks->onSuccess(new RendererWebIDBTransactionImpl(object_id));
   pending_callbacks_.Remove(response_id);
 }
 
 void IndexedDBDispatcher::OnSuccessStringList(
-    int32 response_id, const std::vector<string16>& value) {
+    int32 thread_id, int32 response_id, const std::vector<string16>& value) {
+  DCHECK_EQ(thread_id, CurrentWorkerId());
   WebIDBCallbacks* callbacks = pending_callbacks_.Lookup(response_id);
   if (!callbacks)
     return;
   WebDOMStringList string_list;
   for (std::vector<string16>::const_iterator it = value.begin();
        it != value.end(); ++it)
       string_list.append(*it);
   callbacks->onSuccess(string_list);
   pending_callbacks_.Remove(response_id);
 }
 
 void IndexedDBDispatcher::OnSuccessSerializedScriptValue(
-    int32 response_id, const content::SerializedScriptValue& value) {
+    int32 thread_id, int32 response_id,
+    const content::SerializedScriptValue& value) {
+  DCHECK_EQ(thread_id, CurrentWorkerId());
   WebIDBCallbacks* callbacks = pending_callbacks_.Lookup(response_id);
   if (!callbacks)
     return;
   callbacks->onSuccess(value);
   pending_callbacks_.Remove(response_id);
 }
 
-void IndexedDBDispatcher::OnSuccessOpenCursor(int32 repsonse_id,
-    int32 object_id, const IndexedDBKey& key, const IndexedDBKey& primary_key,
-    const content::SerializedScriptValue& value) {
+void IndexedDBDispatcher::OnSuccessOpenCursor(
+    const IndexedDBMsg_CallbacksSuccessIDBCursor_Params& p) {
+  DCHECK_EQ(p.thread_id, CurrentWorkerId());
+  int32 response_id = p.response_id;
+  int32 object_id = p.cursor_id;
+  const IndexedDBKey& key = p.key;
+  const IndexedDBKey& primary_key = p.primary_key;
+  const content::SerializedScriptValue& value = p.serialized_value;
+
   WebIDBCallbacks* callbacks =
-      pending_callbacks_.Lookup(repsonse_id);
+      pending_callbacks_.Lookup(response_id);
   if (!callbacks)
     return;
 
   RendererWebIDBCursorImpl* cursor = new RendererWebIDBCursorImpl(object_id);
   cursors_[object_id] = cursor;
   cursor->SetKeyAndValue(key, primary_key, value);
   callbacks->onSuccess(cursor);
 
-  pending_callbacks_.Remove(repsonse_id);
+  pending_callbacks_.Remove(response_id);
 }
 
 void IndexedDBDispatcher::OnSuccessCursorContinue(
-    int32 response_id,
-    int32 cursor_id,
-    const IndexedDBKey& key,
-    const IndexedDBKey& primary_key,
-    const content::SerializedScriptValue& value) {
+    const IndexedDBMsg_CallbacksSuccessCursorContinue_Params& p) {
+  DCHECK_EQ(p.thread_id, CurrentWorkerId());
+  int32 response_id = p.response_id;
+  int32 cursor_id = p.cursor_id;
+  const IndexedDBKey& key = p.key;
+  const IndexedDBKey& primary_key = p.primary_key;
+  const content::SerializedScriptValue& value = p.serialized_value;
+
   RendererWebIDBCursorImpl* cursor = cursors_[cursor_id];
   DCHECK(cursor);
   cursor->SetKeyAndValue(key, primary_key, value);
 
   WebIDBCallbacks* callbacks = pending_callbacks_.Lookup(response_id);
   if (!callbacks)
     return;
   callbacks->onSuccessWithContinuation();
 
   pending_callbacks_.Remove(response_id);
 }
 
 void IndexedDBDispatcher::OnSuccessCursorPrefetch(
-      int32 response_id,
-      int32 cursor_id,
-      const std::vector<IndexedDBKey>& keys,
-      const std::vector<IndexedDBKey>& primary_keys,
-      const std::vector<content::SerializedScriptValue>& values) {
+    const IndexedDBMsg_CallbacksSuccessCursorPrefetch_Params& p) {
+  DCHECK_EQ(p.thread_id, CurrentWorkerId());
+  int32 response_id = p.response_id;
+  int32 cursor_id = p.cursor_id;
+  const std::vector<IndexedDBKey>& keys = p.keys;
+  const std::vector<IndexedDBKey>& primary_keys = p.primary_keys;
+  const std::vector<content::SerializedScriptValue>& values = p.values;
   RendererWebIDBCursorImpl* cursor = cursors_[cursor_id];
   DCHECK(cursor);
   cursor->SetPrefetchData(keys, primary_keys, values);
 
   WebIDBCallbacks* callbacks = pending_callbacks_.Lookup(response_id);
   cursor->CachedContinue(callbacks);
   pending_callbacks_.Remove(response_id);
 }
 
-void IndexedDBDispatcher::OnBlocked(int32 response_id) {
+void IndexedDBDispatcher::OnBlocked(int32 thread_id, int32 response_id) {
+  DCHECK_EQ(thread_id, CurrentWorkerId());
   WebIDBCallbacks* callbacks = pending_callbacks_.Lookup(response_id);
   callbacks->onBlocked();
 }
 
-void IndexedDBDispatcher::OnError(int32 response_id, int code,
+void IndexedDBDispatcher::OnError(int32 thread_id, int32 response_id, int code,
                                   const string16& message) {
+  DCHECK_EQ(thread_id, CurrentWorkerId());
   WebIDBCallbacks* callbacks = pending_callbacks_.Lookup(response_id);
   if (!callbacks)
     return;
   callbacks->onError(WebIDBDatabaseError(code, message));
   pending_callbacks_.Remove(response_id);
 }
 
-void IndexedDBDispatcher::OnAbort(int32 transaction_id) {
+void IndexedDBDispatcher::OnAbort(int32 thread_id, int32 transaction_id) {
+  DCHECK_EQ(thread_id, CurrentWorkerId());
   WebIDBTransactionCallbacks* callbacks =
       pending_transaction_callbacks_.Lookup(transaction_id);
   if (!callbacks)
     return;
   callbacks->onAbort();
   pending_transaction_callbacks_.Remove(transaction_id);
 }
 
-void IndexedDBDispatcher::OnComplete(int32 transaction_id) {
+void IndexedDBDispatcher::OnComplete(int32 thread_id, int32 transaction_id) {
+  DCHECK_EQ(thread_id, CurrentWorkerId());
   WebIDBTransactionCallbacks* callbacks =
       pending_transaction_callbacks_.Lookup(transaction_id);
   if (!callbacks)
     return;
   callbacks->onComplete();
   pending_transaction_callbacks_.Remove(transaction_id);
 }
 
-void IndexedDBDispatcher::OnVersionChange(int32 database_id,
+void IndexedDBDispatcher::OnVersionChange(int32 thread_id,
+                                          int32 database_id,
                                           const string16& newVersion) {
+  DCHECK_EQ(thread_id, CurrentWorkerId());
   WebIDBDatabaseCallbacks* callbacks =
       pending_database_callbacks_.Lookup(database_id);
   // callbacks would be NULL if a versionchange event is received after close
   // has been called.
   if (!callbacks)
     return;
   callbacks->onVersionChange(newVersion);
 }
 
 void IndexedDBDispatcher::ResetCursorPrefetchCaches(int32 exception_cursor_id) {
   typedef std::map<int32, RendererWebIDBCursorImpl*>::iterator Iterator;
   for (Iterator i = cursors_.begin(); i != cursors_.end(); ++i) {
     if (i->first == exception_cursor_id)
       continue;
     i->second->ResetPrefetchCache();
   }
 }