michael@0: /* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
michael@0: /* vim: set ts=2 et sw=2 tw=80: */
michael@0: /* This Source Code Form is subject to the terms of the Mozilla Public
michael@0:  * License, v. 2.0. If a copy of the MPL was not distributed with this
michael@0:  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
michael@0: 
michael@0: #include "TransactionThreadPool.h"
michael@0: 
michael@0: #include "nsIObserverService.h"
michael@0: #include "nsIThreadPool.h"
michael@0: 
michael@0: #include "nsComponentManagerUtils.h"
michael@0: #include "nsThreadUtils.h"
michael@0: #include "nsServiceManagerUtils.h"
michael@0: #include "nsXPCOMCIDInternal.h"
michael@0: 
michael@0: #include "ProfilerHelpers.h"
michael@0: 
michael@0: using mozilla::MonitorAutoLock;
michael@0: 
michael@0: USING_INDEXEDDB_NAMESPACE
michael@0: 
michael@0: namespace {
michael@0: 
michael@0: const uint32_t kThreadLimit = 20;
michael@0: const uint32_t kIdleThreadLimit = 5;
michael@0: const uint32_t kIdleThreadTimeoutMs = 30000;
michael@0: 
michael@0: TransactionThreadPool* gThreadPool = nullptr;
michael@0: bool gShutdown = false;
michael@0: 
michael@0: #ifdef MOZ_ENABLE_PROFILER_SPS
michael@0: 
michael@0: class TransactionThreadPoolListener : public nsIThreadPoolListener
michael@0: {
michael@0: public:
michael@0:   NS_DECL_THREADSAFE_ISUPPORTS
michael@0:   NS_DECL_NSITHREADPOOLLISTENER
michael@0: 
michael@0: private:
michael@0:   virtual ~TransactionThreadPoolListener()
michael@0:   { }
michael@0: };
michael@0: 
michael@0: #endif // MOZ_ENABLE_PROFILER_SPS
michael@0: 
michael@0: } // anonymous namespace
michael@0: 
michael@0: BEGIN_INDEXEDDB_NAMESPACE
michael@0: 
michael@0: class FinishTransactionRunnable MOZ_FINAL : public nsIRunnable
michael@0: {
michael@0: public:
michael@0:   NS_DECL_THREADSAFE_ISUPPORTS
michael@0:   NS_DECL_NSIRUNNABLE
michael@0: 
michael@0:   inline FinishTransactionRunnable(IDBTransaction* aTransaction,
michael@0:                                    nsCOMPtr<nsIRunnable>& aFinishRunnable);
michael@0: 
michael@0: private:
michael@0:   IDBTransaction* mTransaction;
michael@0:   nsCOMPtr<nsIRunnable> mFinishRunnable;
michael@0: };
michael@0: 
michael@0: END_INDEXEDDB_NAMESPACE
michael@0: 
michael@0: TransactionThreadPool::TransactionThreadPool()
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Wrong thread!");
michael@0:   NS_ASSERTION(!gThreadPool, "More than one instance!");
michael@0: }
michael@0: 
michael@0: TransactionThreadPool::~TransactionThreadPool()
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Wrong thread!");
michael@0:   NS_ASSERTION(gThreadPool == this, "Different instances!");
michael@0:   gThreadPool = nullptr;
michael@0: }
michael@0: 
michael@0: // static
michael@0: TransactionThreadPool*
michael@0: TransactionThreadPool::GetOrCreate()
michael@0: {
michael@0:   if (!gThreadPool && !gShutdown) {
michael@0:     NS_ASSERTION(NS_IsMainThread(), "Wrong thread!");
michael@0:     nsAutoPtr<TransactionThreadPool> pool(new TransactionThreadPool());
michael@0: 
michael@0:     nsresult rv = pool->Init();
michael@0:     NS_ENSURE_SUCCESS(rv, nullptr);
michael@0: 
michael@0:     gThreadPool = pool.forget();
michael@0:   }
michael@0:   return gThreadPool;
michael@0: }
michael@0: 
michael@0: // static
michael@0: TransactionThreadPool*
michael@0: TransactionThreadPool::Get()
michael@0: {
michael@0:   return gThreadPool;
michael@0: }
michael@0: 
michael@0: // static
michael@0: void
michael@0: TransactionThreadPool::Shutdown()
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Wrong thread!");
michael@0: 
michael@0:   gShutdown = true;
michael@0: 
michael@0:   if (gThreadPool) {
michael@0:     if (NS_FAILED(gThreadPool->Cleanup())) {
michael@0:       NS_WARNING("Failed to shutdown thread pool!");
michael@0:     }
michael@0:     delete gThreadPool;
michael@0:     gThreadPool = nullptr;
michael@0:   }
michael@0: }
michael@0: 
michael@0: nsresult
michael@0: TransactionThreadPool::Init()
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Wrong thread!");
michael@0: 
michael@0:   nsresult rv;
michael@0:   mThreadPool = do_CreateInstance(NS_THREADPOOL_CONTRACTID, &rv);
michael@0:   NS_ENSURE_SUCCESS(rv, rv);
michael@0: 
michael@0:   rv = mThreadPool->SetName(NS_LITERAL_CSTRING("IndexedDB Trans"));
michael@0:   NS_ENSURE_SUCCESS(rv, rv);
michael@0: 
michael@0:   rv = mThreadPool->SetThreadLimit(kThreadLimit);
michael@0:   NS_ENSURE_SUCCESS(rv, rv);
michael@0: 
michael@0:   rv = mThreadPool->SetIdleThreadLimit(kIdleThreadLimit);
michael@0:   NS_ENSURE_SUCCESS(rv, rv);
michael@0: 
michael@0:   rv = mThreadPool->SetIdleThreadTimeout(kIdleThreadTimeoutMs);
michael@0:   NS_ENSURE_SUCCESS(rv, rv);
michael@0: 
michael@0: #ifdef MOZ_ENABLE_PROFILER_SPS
michael@0:   nsCOMPtr<nsIThreadPoolListener> listener =
michael@0:     new TransactionThreadPoolListener();
michael@0: 
michael@0:   rv = mThreadPool->SetListener(listener);
michael@0:   NS_ENSURE_SUCCESS(rv, rv);
michael@0: #endif
michael@0: 
michael@0:   return NS_OK;
michael@0: }
michael@0: 
michael@0: nsresult
michael@0: TransactionThreadPool::Cleanup()
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Wrong thread!");
michael@0: 
michael@0:   PROFILER_MAIN_THREAD_LABEL("IndexedDB", "TransactionThreadPool::Cleanup");
michael@0: 
michael@0:   nsresult rv = mThreadPool->Shutdown();
michael@0:   NS_ENSURE_SUCCESS(rv, rv);
michael@0: 
michael@0:   // Make sure the pool is still accessible while any callbacks generated from
michael@0:   // the other threads are processed.
michael@0:   rv = NS_ProcessPendingEvents(nullptr);
michael@0:   NS_ENSURE_SUCCESS(rv, rv);
michael@0: 
michael@0:   if (!mCompleteCallbacks.IsEmpty()) {
michael@0:     // Run all callbacks manually now.
michael@0:     for (uint32_t index = 0; index < mCompleteCallbacks.Length(); index++) {
michael@0:       mCompleteCallbacks[index].mCallback->Run();
michael@0:     }
michael@0:     mCompleteCallbacks.Clear();
michael@0: 
michael@0:     // And make sure they get processed.
michael@0:     rv = NS_ProcessPendingEvents(nullptr);
michael@0:     NS_ENSURE_SUCCESS(rv, rv);
michael@0:   }
michael@0: 
michael@0:   return NS_OK;
michael@0: }
michael@0: 
michael@0: // static
michael@0: PLDHashOperator
michael@0: TransactionThreadPool::MaybeUnblockTransaction(nsPtrHashKey<TransactionInfo>* aKey,
michael@0:                                                void* aUserArg)
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Wrong thread!");
michael@0: 
michael@0:   TransactionInfo* maybeUnblockedInfo = aKey->GetKey();
michael@0:   TransactionInfo* finishedInfo = static_cast<TransactionInfo*>(aUserArg);
michael@0: 
michael@0:   NS_ASSERTION(maybeUnblockedInfo->blockedOn.Contains(finishedInfo),
michael@0:                "Huh?");
michael@0:   maybeUnblockedInfo->blockedOn.RemoveEntry(finishedInfo);
michael@0:   if (!maybeUnblockedInfo->blockedOn.Count()) {
michael@0:     // Let this transaction run.
michael@0:     maybeUnblockedInfo->queue->Unblock();
michael@0:   }
michael@0: 
michael@0:   return PL_DHASH_NEXT;
michael@0: }
michael@0: 
michael@0: void
michael@0: TransactionThreadPool::FinishTransaction(IDBTransaction* aTransaction)
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Wrong thread!");
michael@0:   NS_ASSERTION(aTransaction, "Null pointer!");
michael@0: 
michael@0:   PROFILER_MAIN_THREAD_LABEL("IndexedDB",
michael@0:                              "TransactionThreadPool::FinishTransaction");
michael@0: 
michael@0:   // AddRef here because removing from the hash will call Release.
michael@0:   nsRefPtr<IDBTransaction> transaction(aTransaction);
michael@0: 
michael@0:   const nsACString& databaseId = aTransaction->mDatabase->Id();
michael@0: 
michael@0:   DatabaseTransactionInfo* dbTransactionInfo;
michael@0:   if (!mTransactionsInProgress.Get(databaseId, &dbTransactionInfo)) {
michael@0:     NS_ERROR("We don't know anyting about this database?!");
michael@0:     return;
michael@0:   }
michael@0: 
michael@0:   DatabaseTransactionInfo::TransactionHashtable& transactionsInProgress =
michael@0:     dbTransactionInfo->transactions;
michael@0: 
michael@0:   uint32_t transactionCount = transactionsInProgress.Count();
michael@0: 
michael@0: #ifdef DEBUG
michael@0:   if (aTransaction->mMode == IDBTransaction::VERSION_CHANGE) {
michael@0:     NS_ASSERTION(transactionCount == 1,
michael@0:                  "More transactions running than should be!");
michael@0:   }
michael@0: #endif
michael@0: 
michael@0:   if (transactionCount == 1) {
michael@0: #ifdef DEBUG
michael@0:     {
michael@0:       const TransactionInfo* info = transactionsInProgress.Get(aTransaction);
michael@0:       NS_ASSERTION(info->transaction == aTransaction, "Transaction mismatch!");
michael@0:     }
michael@0: #endif
michael@0:     mTransactionsInProgress.Remove(databaseId);
michael@0: 
michael@0:     // See if we need to fire any complete callbacks.
michael@0:     uint32_t index = 0;
michael@0:     while (index < mCompleteCallbacks.Length()) {
michael@0:       if (MaybeFireCallback(mCompleteCallbacks[index])) {
michael@0:         mCompleteCallbacks.RemoveElementAt(index);
michael@0:       }
michael@0:       else {
michael@0:         index++;
michael@0:       }
michael@0:     }
michael@0: 
michael@0:     return;
michael@0:   }
michael@0:   TransactionInfo* info = transactionsInProgress.Get(aTransaction);
michael@0:   NS_ASSERTION(info, "We've never heard of this transaction?!?");
michael@0: 
michael@0:   const nsTArray<nsString>& objectStoreNames = aTransaction->mObjectStoreNames;
michael@0:   for (uint32_t index = 0, count = objectStoreNames.Length(); index < count;
michael@0:        index++) {
michael@0:     TransactionInfoPair* blockInfo =
michael@0:       dbTransactionInfo->blockingTransactions.Get(objectStoreNames[index]);
michael@0:     NS_ASSERTION(blockInfo, "Huh?");
michael@0: 
michael@0:     if (aTransaction->mMode == IDBTransaction::READ_WRITE &&
michael@0:         blockInfo->lastBlockingReads == info) {
michael@0:       blockInfo->lastBlockingReads = nullptr;
michael@0:     }
michael@0: 
michael@0:     uint32_t i = blockInfo->lastBlockingWrites.IndexOf(info);
michael@0:     if (i != blockInfo->lastBlockingWrites.NoIndex) {
michael@0:       blockInfo->lastBlockingWrites.RemoveElementAt(i);
michael@0:     }
michael@0:   }
michael@0: 
michael@0:   info->blocking.EnumerateEntries(MaybeUnblockTransaction, info);
michael@0: 
michael@0:   transactionsInProgress.Remove(aTransaction);
michael@0: }
michael@0: 
michael@0: TransactionThreadPool::TransactionQueue&
michael@0: TransactionThreadPool::GetQueueForTransaction(IDBTransaction* aTransaction)
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Wrong thread!");
michael@0:   NS_ASSERTION(aTransaction, "Null pointer!");
michael@0: 
michael@0:   const nsACString& databaseId = aTransaction->mDatabase->Id();
michael@0: 
michael@0:   const nsTArray<nsString>& objectStoreNames = aTransaction->mObjectStoreNames;
michael@0:   const uint16_t mode = aTransaction->mMode;
michael@0: 
michael@0:   // See if we can run this transaction now.
michael@0:   DatabaseTransactionInfo* dbTransactionInfo;
michael@0:   if (!mTransactionsInProgress.Get(databaseId, &dbTransactionInfo)) {
michael@0:     // First transaction for this database.
michael@0:     dbTransactionInfo = new DatabaseTransactionInfo();
michael@0:     mTransactionsInProgress.Put(databaseId, dbTransactionInfo);
michael@0:   }
michael@0: 
michael@0:   DatabaseTransactionInfo::TransactionHashtable& transactionsInProgress =
michael@0:     dbTransactionInfo->transactions;
michael@0:   TransactionInfo* info = transactionsInProgress.Get(aTransaction);
michael@0:   if (info) {
michael@0:     // We recognize this one.
michael@0:     return *info->queue;
michael@0:   }
michael@0: 
michael@0:   TransactionInfo* transactionInfo = new TransactionInfo(aTransaction);
michael@0: 
michael@0:   dbTransactionInfo->transactions.Put(aTransaction, transactionInfo);;
michael@0: 
michael@0:   for (uint32_t index = 0, count = objectStoreNames.Length(); index < count;
michael@0:        index++) {
michael@0:     TransactionInfoPair* blockInfo =
michael@0:       dbTransactionInfo->blockingTransactions.Get(objectStoreNames[index]);
michael@0:     if (!blockInfo) {
michael@0:       blockInfo = new TransactionInfoPair();
michael@0:       blockInfo->lastBlockingReads = nullptr;
michael@0:       dbTransactionInfo->blockingTransactions.Put(objectStoreNames[index],
michael@0:                                                   blockInfo);
michael@0:     }
michael@0: 
michael@0:     // Mark what we are blocking on.
michael@0:     if (blockInfo->lastBlockingReads) {
michael@0:       TransactionInfo* blockingInfo = blockInfo->lastBlockingReads;
michael@0:       transactionInfo->blockedOn.PutEntry(blockingInfo);
michael@0:       blockingInfo->blocking.PutEntry(transactionInfo);
michael@0:     }
michael@0: 
michael@0:     if (mode == IDBTransaction::READ_WRITE &&
michael@0:         blockInfo->lastBlockingWrites.Length()) {
michael@0:       for (uint32_t index = 0,
michael@0:            count = blockInfo->lastBlockingWrites.Length(); index < count;
michael@0:            index++) {
michael@0:         TransactionInfo* blockingInfo = blockInfo->lastBlockingWrites[index];
michael@0:         transactionInfo->blockedOn.PutEntry(blockingInfo);
michael@0:         blockingInfo->blocking.PutEntry(transactionInfo);
michael@0:       }
michael@0:     }
michael@0: 
michael@0:     if (mode == IDBTransaction::READ_WRITE) {
michael@0:       blockInfo->lastBlockingReads = transactionInfo;
michael@0:       blockInfo->lastBlockingWrites.Clear();
michael@0:     }
michael@0:     else {
michael@0:       blockInfo->lastBlockingWrites.AppendElement(transactionInfo);
michael@0:     }
michael@0:   }
michael@0: 
michael@0:   if (!transactionInfo->blockedOn.Count()) {
michael@0:     transactionInfo->queue->Unblock();
michael@0:   }
michael@0: 
michael@0:   return *transactionInfo->queue;
michael@0: }
michael@0: 
michael@0: nsresult
michael@0: TransactionThreadPool::Dispatch(IDBTransaction* aTransaction,
michael@0:                                 nsIRunnable* aRunnable,
michael@0:                                 bool aFinish,
michael@0:                                 nsIRunnable* aFinishRunnable)
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Wrong thread!");
michael@0:   NS_ASSERTION(aTransaction, "Null pointer!");
michael@0:   NS_ASSERTION(aRunnable, "Null pointer!");
michael@0: 
michael@0:   if (aTransaction->mDatabase->IsInvalidated() && !aFinish) {
michael@0:     return NS_ERROR_NOT_AVAILABLE;
michael@0:   }
michael@0: 
michael@0:   TransactionQueue& queue = GetQueueForTransaction(aTransaction);
michael@0: 
michael@0:   queue.Dispatch(aRunnable);
michael@0:   if (aFinish) {
michael@0:     queue.Finish(aFinishRunnable);
michael@0:   }
michael@0:   return NS_OK;
michael@0: }
michael@0: 
michael@0: void
michael@0: TransactionThreadPool::WaitForDatabasesToComplete(
michael@0:                                    nsTArray<nsRefPtr<IDBDatabase> >& aDatabases,
michael@0:                                    nsIRunnable* aCallback)
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Wrong thread!");
michael@0:   NS_ASSERTION(!aDatabases.IsEmpty(), "No databases to wait on!");
michael@0:   NS_ASSERTION(aCallback, "Null pointer!");
michael@0: 
michael@0:   DatabasesCompleteCallback* callback = mCompleteCallbacks.AppendElement();
michael@0: 
michael@0:   callback->mCallback = aCallback;
michael@0:   callback->mDatabases.SwapElements(aDatabases);
michael@0: 
michael@0:   if (MaybeFireCallback(*callback)) {
michael@0:     mCompleteCallbacks.RemoveElementAt(mCompleteCallbacks.Length() - 1);
michael@0:   }
michael@0: }
michael@0: 
michael@0: // static
michael@0: PLDHashOperator
michael@0: TransactionThreadPool::CollectTransactions(IDBTransaction* aKey,
michael@0:                                            TransactionInfo* aValue,
michael@0:                                            void* aUserArg)
michael@0: {
michael@0:   nsAutoTArray<nsRefPtr<IDBTransaction>, 50>* transactionArray =
michael@0:     static_cast<nsAutoTArray<nsRefPtr<IDBTransaction>, 50>*>(aUserArg);
michael@0:   transactionArray->AppendElement(aKey);
michael@0: 
michael@0:   return PL_DHASH_NEXT;
michael@0: }
michael@0: 
michael@0: void
michael@0: TransactionThreadPool::AbortTransactionsForDatabase(IDBDatabase* aDatabase)
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Wrong thread!");
michael@0:   NS_ASSERTION(aDatabase, "Null pointer!");
michael@0: 
michael@0:   PROFILER_MAIN_THREAD_LABEL("IndexedDB",
michael@0:                              "TransactionThreadPool::"
michael@0:                              "AbortTransactionsForDatabase");
michael@0: 
michael@0:   // Get list of transactions for this database id
michael@0:   DatabaseTransactionInfo* dbTransactionInfo;
michael@0:   if (!mTransactionsInProgress.Get(aDatabase->Id(), &dbTransactionInfo)) {
michael@0:     // If there are no transactions, we're done.
michael@0:     return;
michael@0:   }
michael@0: 
michael@0:   // Collect any running transactions
michael@0:   DatabaseTransactionInfo::TransactionHashtable& transactionsInProgress =
michael@0:     dbTransactionInfo->transactions;
michael@0: 
michael@0:   NS_ASSERTION(transactionsInProgress.Count(), "Should never be 0!");
michael@0: 
michael@0:   nsAutoTArray<nsRefPtr<IDBTransaction>, 50> transactions;
michael@0:   transactionsInProgress.EnumerateRead(CollectTransactions, &transactions);
michael@0: 
michael@0:   // Abort transactions. Do this after collecting the transactions in case
michael@0:   // calling Abort() modifies the data structures we're iterating above.
michael@0:   for (uint32_t index = 0; index < transactions.Length(); index++) {
michael@0:     if (transactions[index]->Database() != aDatabase) {
michael@0:       continue;
michael@0:     }
michael@0: 
michael@0:     // This can fail, for example if the transaction is in the process of
michael@0:     // being comitted. That is expected and fine, so we ignore any returned
michael@0:     // errors.
michael@0:     ErrorResult rv;
michael@0:     transactions[index]->Abort(rv);
michael@0:   }
michael@0: }
michael@0: 
michael@0: struct MOZ_STACK_CLASS TransactionSearchInfo
michael@0: {
michael@0:   TransactionSearchInfo(nsIOfflineStorage* aDatabase)
michael@0:     : db(aDatabase), found(false)
michael@0:   {
michael@0:   }
michael@0: 
michael@0:   nsIOfflineStorage* db;
michael@0:   bool found;
michael@0: };
michael@0: 
michael@0: // static
michael@0: PLDHashOperator
michael@0: TransactionThreadPool::FindTransaction(IDBTransaction* aKey,
michael@0:                                        TransactionInfo* aValue,
michael@0:                                        void* aUserArg)
michael@0: {
michael@0:   TransactionSearchInfo* info = static_cast<TransactionSearchInfo*>(aUserArg);
michael@0: 
michael@0:   if (aKey->Database() == info->db) {
michael@0:     info->found = true;
michael@0:     return PL_DHASH_STOP;
michael@0:   }
michael@0: 
michael@0:   return PL_DHASH_NEXT;
michael@0: }
michael@0: bool
michael@0: TransactionThreadPool::HasTransactionsForDatabase(IDBDatabase* aDatabase)
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Wrong thread!");
michael@0:   NS_ASSERTION(aDatabase, "Null pointer!");
michael@0: 
michael@0:   DatabaseTransactionInfo* dbTransactionInfo = nullptr;
michael@0:   dbTransactionInfo = mTransactionsInProgress.Get(aDatabase->Id());
michael@0:   if (!dbTransactionInfo) {
michael@0:     return false;
michael@0:   }
michael@0: 
michael@0:   TransactionSearchInfo info(aDatabase);
michael@0:   dbTransactionInfo->transactions.EnumerateRead(FindTransaction, &info);
michael@0: 
michael@0:   return info.found;
michael@0: }
michael@0: 
michael@0: bool
michael@0: TransactionThreadPool::MaybeFireCallback(DatabasesCompleteCallback aCallback)
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Wrong thread!");
michael@0: 
michael@0:   PROFILER_MAIN_THREAD_LABEL("IndexedDB",
michael@0:                              "TransactionThreadPool::MaybeFireCallback");
michael@0: 
michael@0:   for (uint32_t index = 0; index < aCallback.mDatabases.Length(); index++) {
michael@0:     IDBDatabase* database = aCallback.mDatabases[index];
michael@0:     if (!database) {
michael@0:       MOZ_CRASH();
michael@0:     }
michael@0: 
michael@0:     if (mTransactionsInProgress.Get(database->Id(), nullptr)) {
michael@0:       return false;
michael@0:     }
michael@0:   }
michael@0: 
michael@0:   aCallback.mCallback->Run();
michael@0:   return true;
michael@0: }
michael@0: 
michael@0: TransactionThreadPool::
michael@0: TransactionQueue::TransactionQueue(IDBTransaction* aTransaction)
michael@0: : mMonitor("TransactionQueue::mMonitor"),
michael@0:   mTransaction(aTransaction),
michael@0:   mShouldFinish(false)
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Wrong thread!");
michael@0:   NS_ASSERTION(aTransaction, "Null pointer!");
michael@0: }
michael@0: 
michael@0: void
michael@0: TransactionThreadPool::TransactionQueue::Unblock()
michael@0: {
michael@0:   MonitorAutoLock lock(mMonitor);
michael@0: 
michael@0:   // NB: Finish may be called before Unblock.
michael@0: 
michael@0:   TransactionThreadPool::Get()->mThreadPool->
michael@0:     Dispatch(this, NS_DISPATCH_NORMAL);
michael@0: }
michael@0: 
michael@0: void
michael@0: TransactionThreadPool::TransactionQueue::Dispatch(nsIRunnable* aRunnable)
michael@0: {
michael@0:   MonitorAutoLock lock(mMonitor);
michael@0: 
michael@0:   NS_ASSERTION(!mShouldFinish, "Dispatch called after Finish!");
michael@0: 
michael@0:   mQueue.AppendElement(aRunnable);
michael@0: 
michael@0:   mMonitor.Notify();
michael@0: }
michael@0: 
michael@0: void
michael@0: TransactionThreadPool::TransactionQueue::Finish(nsIRunnable* aFinishRunnable)
michael@0: {
michael@0:   MonitorAutoLock lock(mMonitor);
michael@0: 
michael@0:   NS_ASSERTION(!mShouldFinish, "Finish called more than once!");
michael@0: 
michael@0:   mShouldFinish = true;
michael@0:   mFinishRunnable = aFinishRunnable;
michael@0: 
michael@0:   mMonitor.Notify();
michael@0: }
michael@0: 
michael@0: NS_IMPL_ISUPPORTS(TransactionThreadPool::TransactionQueue, nsIRunnable)
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: TransactionThreadPool::TransactionQueue::Run()
michael@0: {
michael@0:   NS_ASSERTION(!NS_IsMainThread(), "Wrong thread!");
michael@0:   NS_ASSERTION(IndexedDatabaseManager::IsMainProcess(), "Wrong process!");
michael@0: 
michael@0:   PROFILER_LABEL("IndexedDB", "TransactionQueue::Run");
michael@0: 
michael@0:   IDB_PROFILER_MARK("IndexedDB Transaction %llu: Beginning database work",
michael@0:                     "IDBTransaction[%llu] DT Start",
michael@0:                     mTransaction->GetSerialNumber());
michael@0: 
michael@0:   nsAutoTArray<nsCOMPtr<nsIRunnable>, 10> queue;
michael@0:   nsCOMPtr<nsIRunnable> finishRunnable;
michael@0:   bool shouldFinish = false;
michael@0: 
michael@0:   do {
michael@0:     NS_ASSERTION(queue.IsEmpty(), "Should have cleared this!");
michael@0: 
michael@0:     {
michael@0:       MonitorAutoLock lock(mMonitor);
michael@0:       while (!mShouldFinish && mQueue.IsEmpty()) {
michael@0:         if (NS_FAILED(mMonitor.Wait())) {
michael@0:           NS_ERROR("Failed to wait!");
michael@0:         }
michael@0:       }
michael@0: 
michael@0:       mQueue.SwapElements(queue);
michael@0:       if (mShouldFinish) {
michael@0:         mFinishRunnable.swap(finishRunnable);
michael@0:         shouldFinish = true;
michael@0:       }
michael@0:     }
michael@0: 
michael@0:     uint32_t count = queue.Length();
michael@0:     for (uint32_t index = 0; index < count; index++) {
michael@0:       nsCOMPtr<nsIRunnable>& runnable = queue[index];
michael@0:       runnable->Run();
michael@0:       runnable = nullptr;
michael@0:     }
michael@0: 
michael@0:     if (count) {
michael@0:       queue.Clear();
michael@0:     }
michael@0:   } while (!shouldFinish);
michael@0: 
michael@0:   IDB_PROFILER_MARK("IndexedDB Transaction %llu: Finished database work",
michael@0:                     "IDBTransaction[%llu] DT Done",
michael@0:                     mTransaction->GetSerialNumber());
michael@0: 
michael@0:   nsCOMPtr<nsIRunnable> finishTransactionRunnable =
michael@0:     new FinishTransactionRunnable(mTransaction, finishRunnable);
michael@0:   if (NS_FAILED(NS_DispatchToMainThread(finishTransactionRunnable,
michael@0:                                         NS_DISPATCH_NORMAL))) {
michael@0:     NS_WARNING("Failed to dispatch finishTransactionRunnable!");
michael@0:   }
michael@0: 
michael@0:   return NS_OK;
michael@0: }
michael@0: 
michael@0: FinishTransactionRunnable::FinishTransactionRunnable(
michael@0:                                          IDBTransaction* aTransaction,
michael@0:                                          nsCOMPtr<nsIRunnable>& aFinishRunnable)
michael@0: : mTransaction(aTransaction)
michael@0: {
michael@0:   NS_ASSERTION(!NS_IsMainThread(), "Wrong thread!");
michael@0:   NS_ASSERTION(aTransaction, "Null pointer!");
michael@0:   mFinishRunnable.swap(aFinishRunnable);
michael@0: }
michael@0: 
michael@0: NS_IMPL_ISUPPORTS(FinishTransactionRunnable, nsIRunnable)
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: FinishTransactionRunnable::Run()
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Wrong thread!");
michael@0: 
michael@0:   PROFILER_MAIN_THREAD_LABEL("IndexedDB", "FinishTransactionRunnable::Run");
michael@0: 
michael@0:   if (!gThreadPool) {
michael@0:     NS_ERROR("Running after shutdown!");
michael@0:     return NS_ERROR_FAILURE;
michael@0:   }
michael@0: 
michael@0:   gThreadPool->FinishTransaction(mTransaction);
michael@0: 
michael@0:   if (mFinishRunnable) {
michael@0:     mFinishRunnable->Run();
michael@0:     mFinishRunnable = nullptr;
michael@0:   }
michael@0: 
michael@0:   return NS_OK;
michael@0: }
michael@0: 
michael@0: #ifdef MOZ_ENABLE_PROFILER_SPS
michael@0: 
michael@0: NS_IMPL_ISUPPORTS(TransactionThreadPoolListener, nsIThreadPoolListener)
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: TransactionThreadPoolListener::OnThreadCreated()
michael@0: {
michael@0:   MOZ_ASSERT(!NS_IsMainThread());
michael@0:   char aLocal;
michael@0:   profiler_register_thread("IndexedDB Transaction", &aLocal);
michael@0:   return NS_OK;
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: TransactionThreadPoolListener::OnThreadShuttingDown()
michael@0: {
michael@0:   MOZ_ASSERT(!NS_IsMainThread());
michael@0:   profiler_unregister_thread();
michael@0:   return NS_OK;
michael@0: }
michael@0: 
michael@0: #endif // MOZ_ENABLE_PROFILER_SPS