mozilla
/
gecko-dev
зеркало из https://github.com/mozilla/gecko-dev.git
1
0
Форкнуть 0
Код Задачи Пакеты Проекты Релизы Вики Активность
gecko-dev/dom/indexedDB/TransactionThreadPool.cpp

1043 строки
28 KiB
C++
Исходник Ответственный История

/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
/* vim: set ts=2 et sw=2 tw=80: */
/* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this
* file, You can obtain one at http://mozilla.org/MPL/2.0/. */
#include "TransactionThreadPool.h"
#include "IDBTransaction.h"
#include "mozilla/Monitor.h"
#include "mozilla/Move.h"
#include "mozilla/ipc/BackgroundParent.h"
#include "nsComponentManagerUtils.h"
#include "nsIEventTarget.h"
#include "nsIRunnable.h"
#include "nsISupportsPriority.h"
#include "nsIThreadPool.h"
#include "nsThreadUtils.h"
#include "nsServiceManagerUtils.h"
#include "nsXPCOMCIDInternal.h"
#include "ProfilerHelpers.h"
#include "nsThread.h"
namespace mozilla {
namespace dom {
namespace indexedDB {
using mozilla::ipc::AssertIsOnBackgroundThread;
namespace {
const uint32_t kThreadLimit = 20;
const uint32_t kIdleThreadLimit = 5;
const uint32_t kIdleThreadTimeoutMs = 30000;
#if defined(DEBUG) || defined(MOZ_ENABLE_PROFILER_SPS)
#define BUILD_THREADPOOL_LISTENER
#endif
#ifdef DEBUG
const int32_t kDEBUGThreadPriority = nsISupportsPriority::PRIORITY_NORMAL;
const uint32_t kDEBUGThreadSleepMS = 0;
#endif // DEBUG
#ifdef BUILD_THREADPOOL_LISTENER
class TransactionThreadPoolListener MOZ_FINAL
: public nsIThreadPoolListener
{
public:
TransactionThreadPoolListener()
{ }
NS_DECL_THREADSAFE_ISUPPORTS
private:
virtual ~TransactionThreadPoolListener()
{ }
NS_DECL_NSITHREADPOOLLISTENER
};
#endif // BUILD_THREADPOOL_LISTENER
} // anonymous namespace
class TransactionThreadPool::FinishTransactionRunnable MOZ_FINAL
: public nsRunnable
{
typedef TransactionThreadPool::FinishCallback FinishCallback;
nsRefPtr<TransactionThreadPool> mThreadPool;
nsRefPtr<FinishCallback> mFinishCallback;
uint64_t mTransactionId;
const nsCString mDatabaseId;
const nsTArray<nsString> mObjectStoreNames;
uint16_t mMode;
public:
FinishTransactionRunnable(already_AddRefed<TransactionThreadPool> aThreadPool,
uint64_t aTransactionId,
const nsACString& aDatabaseId,
const nsTArray<nsString>& aObjectStoreNames,
uint16_t aMode,
already_AddRefed<FinishCallback> aFinishCallback);
void
Dispatch()
{
MOZ_ALWAYS_TRUE(NS_SUCCEEDED(
mThreadPool->mOwningThread->Dispatch(this, NS_DISPATCH_NORMAL)));
}
NS_DECL_ISUPPORTS_INHERITED
private:
~FinishTransactionRunnable()
{ }
NS_DECL_NSIRUNNABLE
};
struct TransactionThreadPool::DatabaseTransactionInfo MOZ_FINAL
{
typedef nsClassHashtable<nsUint64HashKey, TransactionInfo>
TransactionHashtable;
TransactionHashtable transactions;
nsClassHashtable<nsStringHashKey, TransactionInfoPair> blockingTransactions;
DatabaseTransactionInfo()
{
MOZ_COUNT_CTOR(DatabaseTransactionInfo);
}
~DatabaseTransactionInfo()
{
MOZ_COUNT_DTOR(DatabaseTransactionInfo);
}
};
struct TransactionThreadPool::DatabasesCompleteCallback MOZ_FINAL
{
friend class nsAutoPtr<DatabasesCompleteCallback>;
nsTArray<nsCString> mDatabaseIds;
nsCOMPtr<nsIRunnable> mCallback;
DatabasesCompleteCallback()
{
MOZ_COUNT_CTOR(DatabasesCompleteCallback);
}
private:
~DatabasesCompleteCallback()
{
MOZ_COUNT_DTOR(DatabasesCompleteCallback);
}
};
class TransactionThreadPool::TransactionQueue MOZ_FINAL
: public nsRunnable
{
Monitor mMonitor;
nsRefPtr<TransactionThreadPool> mOwningThreadPool;
const uint64_t mTransactionId;
const nsID mBackgroundChildLoggingId;
const int64_t mLoggingSerialNumber;
const nsCString mDatabaseId;
const nsTArray<nsString> mObjectStoreNames;
uint16_t mMode;
nsAutoTArray<nsCOMPtr<nsIRunnable>, 10> mQueue;
nsRefPtr<FinishCallback> mFinishCallback;
bool mShouldFinish;
public:
TransactionQueue(TransactionThreadPool* aThreadPool,
uint64_t aTransactionId,
const nsACString& aDatabaseId,
const nsTArray<nsString>& aObjectStoreNames,
uint16_t aMode,
const nsID& aBackgroundChildLoggingId,
int64_t aLoggingSerialNumber);
NS_DECL_ISUPPORTS_INHERITED
void Unblock();
void Dispatch(nsIRunnable* aRunnable);
void Finish(FinishCallback* aFinishCallback);
private:
~TransactionQueue()
{ }
#ifdef MOZ_NUWA_PROCESS
nsThread* mThread;
#endif
NS_DECL_NSIRUNNABLE
};
struct TransactionThreadPool::TransactionInfo MOZ_FINAL
{
uint64_t transactionId;
nsCString databaseId;
nsRefPtr<TransactionQueue> queue;
nsTHashtable<nsPtrHashKey<TransactionInfo>> blockedOn;
nsTHashtable<nsPtrHashKey<TransactionInfo>> blocking;
TransactionInfo(TransactionThreadPool* aThreadPool,
uint64_t aTransactionId,
const nsACString& aDatabaseId,
const nsTArray<nsString>& aObjectStoreNames,
uint16_t aMode,
const nsID& aBackgroundChildLoggingId,
int64_t aLoggingSerialNumber)
: transactionId(aTransactionId), databaseId(aDatabaseId)
{
MOZ_COUNT_CTOR(TransactionInfo);
queue =
new TransactionQueue(aThreadPool,
aTransactionId,
aDatabaseId,
aObjectStoreNames,
aMode,
aBackgroundChildLoggingId,
aLoggingSerialNumber);
}
~TransactionInfo()
{
MOZ_COUNT_DTOR(TransactionInfo);
}
};
struct TransactionThreadPool::TransactionInfoPair MOZ_FINAL
{
// Multiple reading transactions can block future writes.
nsTArray<TransactionInfo*> lastBlockingWrites;
// But only a single writing transaction can block future reads.
TransactionInfo* lastBlockingReads;
TransactionInfoPair()
: lastBlockingReads(nullptr)
{
MOZ_COUNT_CTOR(TransactionInfoPair);
}
~TransactionInfoPair()
{
MOZ_COUNT_DTOR(TransactionInfoPair);
}
};
TransactionThreadPool::TransactionThreadPool()
: mOwningThread(NS_GetCurrentThread())
, mNextTransactionId(0)
, mShutdownRequested(false)
, mShutdownComplete(false)
{
AssertIsOnBackgroundThread();
MOZ_ASSERT(mOwningThread);
AssertIsOnOwningThread();
}
TransactionThreadPool::~TransactionThreadPool()
{
AssertIsOnOwningThread();
MOZ_ASSERT(mShutdownRequested);
MOZ_ASSERT(mShutdownComplete);
}
#ifdef DEBUG
void
TransactionThreadPool::AssertIsOnOwningThread() const
{
MOZ_ASSERT(mOwningThread);
bool current;
MOZ_ALWAYS_TRUE(NS_SUCCEEDED(mOwningThread->IsOnCurrentThread(&current)));
MOZ_ASSERT(current);
}
#endif // DEBUG
// static
already_AddRefed<TransactionThreadPool>
TransactionThreadPool::Create()
{
AssertIsOnBackgroundThread();
nsRefPtr<TransactionThreadPool> threadPool = new TransactionThreadPool();
threadPool->AssertIsOnOwningThread();
if (NS_WARN_IF(NS_FAILED(threadPool->Init()))) {
threadPool->CleanupAsync();
return nullptr;
}
return threadPool.forget();
}
void
TransactionThreadPool::Shutdown()
{
AssertIsOnOwningThread();
MOZ_ASSERT(!mShutdownRequested);
MOZ_ASSERT(!mShutdownComplete);
mShutdownRequested = true;
if (!mThreadPool) {
MOZ_ASSERT(!mTransactionsInProgress.Count());
MOZ_ASSERT(mCompleteCallbacks.IsEmpty());
mShutdownComplete = true;
return;
}
if (!mTransactionsInProgress.Count()) {
Cleanup();
MOZ_ASSERT(mShutdownComplete);
return;
}
nsIThread* currentThread = NS_GetCurrentThread();
MOZ_ASSERT(currentThread);
while (!mShutdownComplete) {
MOZ_ALWAYS_TRUE(NS_ProcessNextEvent(currentThread));
}
}
uint64_t
TransactionThreadPool::NextTransactionId()
{
AssertIsOnOwningThread();
MOZ_ASSERT(mNextTransactionId < UINT64_MAX);
return ++mNextTransactionId;
}
nsresult
TransactionThreadPool::Init()
{
AssertIsOnOwningThread();
nsresult rv;
mThreadPool = do_CreateInstance(NS_THREADPOOL_CONTRACTID, &rv);
if (NS_WARN_IF(NS_FAILED(rv))) {
return rv;
}
rv = mThreadPool->SetName(NS_LITERAL_CSTRING("IndexedDB Trans"));
if (NS_WARN_IF(NS_FAILED(rv))) {
return rv;
}
rv = mThreadPool->SetThreadLimit(kThreadLimit);
if (NS_WARN_IF(NS_FAILED(rv))) {
return rv;
}
rv = mThreadPool->SetIdleThreadLimit(kIdleThreadLimit);
if (NS_WARN_IF(NS_FAILED(rv))) {
return rv;
}
rv = mThreadPool->SetIdleThreadTimeout(kIdleThreadTimeoutMs);
if (NS_WARN_IF(NS_FAILED(rv))) {
return rv;
}
#ifdef BUILD_THREADPOOL_LISTENER
nsCOMPtr<nsIThreadPoolListener> listener =
new TransactionThreadPoolListener();
rv = mThreadPool->SetListener(listener);
if (NS_WARN_IF(NS_FAILED(rv))) {
return rv;
}
#endif // BUILD_THREADPOOL_LISTENER
NS_WARN_IF_FALSE(!kDEBUGThreadSleepMS,
"TransactionThreadPool thread debugging enabled, sleeping "
"after every event!");
return NS_OK;
}
void
TransactionThreadPool::Cleanup()
{
AssertIsOnOwningThread();
MOZ_ASSERT(mThreadPool);
MOZ_ASSERT(mShutdownRequested);
MOZ_ASSERT(!mShutdownComplete);
MOZ_ASSERT(!mTransactionsInProgress.Count());
MOZ_ALWAYS_TRUE(NS_SUCCEEDED(mThreadPool->Shutdown()));
if (!mCompleteCallbacks.IsEmpty()) {
// Run all callbacks manually now.
for (uint32_t count = mCompleteCallbacks.Length(), index = 0;
index < count;
index++) {
nsAutoPtr<DatabasesCompleteCallback>& completeCallback =
mCompleteCallbacks[index];
MOZ_ASSERT(completeCallback);
MOZ_ASSERT(completeCallback->mCallback);
completeCallback->mCallback->Run();
completeCallback = nullptr;
}
mCompleteCallbacks.Clear();
// And make sure they get processed.
nsIThread* currentThread = NS_GetCurrentThread();
MOZ_ASSERT(currentThread);
MOZ_ALWAYS_TRUE(NS_SUCCEEDED(NS_ProcessPendingEvents(currentThread)));
}
mShutdownComplete = true;
}
void
TransactionThreadPool::CleanupAsync()
{
AssertIsOnOwningThread();
MOZ_ASSERT(!mShutdownComplete);
MOZ_ASSERT(!mTransactionsInProgress.Count());
mShutdownRequested = true;
if (!mThreadPool) {
MOZ_ASSERT(mCompleteCallbacks.IsEmpty());
mShutdownComplete = true;
return;
}
nsCOMPtr<nsIRunnable> runnable =
NS_NewRunnableMethod(this, &TransactionThreadPool::Cleanup);
MOZ_ASSERT(runnable);
MOZ_ALWAYS_TRUE(NS_SUCCEEDED(NS_DispatchToCurrentThread(runnable)));
}
// static
PLDHashOperator
TransactionThreadPool::MaybeUnblockTransaction(
nsPtrHashKey<TransactionInfo>* aKey,
void* aUserArg)
{
AssertIsOnBackgroundThread();
TransactionInfo* maybeUnblockedInfo = aKey->GetKey();
TransactionInfo* finishedInfo = static_cast<TransactionInfo*>(aUserArg);
NS_ASSERTION(maybeUnblockedInfo->blockedOn.Contains(finishedInfo),
"Huh?");
maybeUnblockedInfo->blockedOn.RemoveEntry(finishedInfo);
if (!maybeUnblockedInfo->blockedOn.Count()) {
// Let this transaction run.
maybeUnblockedInfo->queue->Unblock();
}
return PL_DHASH_NEXT;
}
void
TransactionThreadPool::FinishTransaction(
uint64_t aTransactionId,
const nsACString& aDatabaseId,
const nsTArray<nsString>& aObjectStoreNames,
uint16_t aMode)
{
AssertIsOnOwningThread();
PROFILER_LABEL("IndexedDB",
"TransactionThreadPool::FinishTransaction",
js::ProfileEntry::Category::STORAGE);
DatabaseTransactionInfo* dbTransactionInfo;
if (!mTransactionsInProgress.Get(aDatabaseId, &dbTransactionInfo)) {
NS_ERROR("We don't know anyting about this database?!");
return;
}
DatabaseTransactionInfo::TransactionHashtable& transactionsInProgress =
dbTransactionInfo->transactions;
uint32_t transactionCount = transactionsInProgress.Count();
#ifdef DEBUG
if (aMode == IDBTransaction::VERSION_CHANGE) {
NS_ASSERTION(transactionCount == 1,
"More transactions running than should be!");
}
#endif
if (transactionCount == 1) {
#ifdef DEBUG
{
const TransactionInfo* info = transactionsInProgress.Get(aTransactionId);
NS_ASSERTION(info->transactionId == aTransactionId, "Transaction mismatch!");
}
#endif
mTransactionsInProgress.Remove(aDatabaseId);
// See if we need to fire any complete callbacks.
uint32_t index = 0;
while (index < mCompleteCallbacks.Length()) {
if (MaybeFireCallback(mCompleteCallbacks[index])) {
mCompleteCallbacks.RemoveElementAt(index);
} else {
index++;
}
}
if (mShutdownRequested) {
CleanupAsync();
}
return;
}
TransactionInfo* info = transactionsInProgress.Get(aTransactionId);
NS_ASSERTION(info, "We've never heard of this transaction?!?");
const nsTArray<nsString>& objectStoreNames = aObjectStoreNames;
for (size_t index = 0, count = objectStoreNames.Length(); index < count;
index++) {
TransactionInfoPair* blockInfo =
dbTransactionInfo->blockingTransactions.Get(objectStoreNames[index]);
NS_ASSERTION(blockInfo, "Huh?");
if (aMode == IDBTransaction::READ_WRITE &&
blockInfo->lastBlockingReads == info) {
blockInfo->lastBlockingReads = nullptr;
}
size_t i = blockInfo->lastBlockingWrites.IndexOf(info);
if (i != blockInfo->lastBlockingWrites.NoIndex) {
blockInfo->lastBlockingWrites.RemoveElementAt(i);
}
}
info->blocking.EnumerateEntries(MaybeUnblockTransaction, info);
transactionsInProgress.Remove(aTransactionId);
}
TransactionThreadPool::TransactionQueue*
TransactionThreadPool::GetQueueForTransaction(uint64_t aTransactionId,
const nsACString& aDatabaseId)
{
AssertIsOnOwningThread();
MOZ_ASSERT(aTransactionId <= mNextTransactionId);
DatabaseTransactionInfo* dbTransactionInfo;
if (mTransactionsInProgress.Get(aDatabaseId, &dbTransactionInfo)) {
DatabaseTransactionInfo::TransactionHashtable& transactionsInProgress =
dbTransactionInfo->transactions;
TransactionInfo* info = transactionsInProgress.Get(aTransactionId);
if (info) {
// We recognize this one.
return info->queue;
}
}
return nullptr;
}
TransactionThreadPool::TransactionQueue&
TransactionThreadPool::CreateQueueForTransaction(
uint64_t aTransactionId,
const nsACString& aDatabaseId,
const nsTArray<nsString>& aObjectStoreNames,
uint16_t aMode,
const nsID& aBackgroundChildLoggingId,
int64_t aLoggingSerialNumber)
{
AssertIsOnOwningThread();
MOZ_ASSERT(aTransactionId <= mNextTransactionId);
MOZ_ASSERT(!GetQueueForTransaction(aTransactionId, aDatabaseId));
// See if we can run this transaction now.
DatabaseTransactionInfo* dbTransactionInfo;
if (!mTransactionsInProgress.Get(aDatabaseId, &dbTransactionInfo)) {
// First transaction for this database.
dbTransactionInfo = new DatabaseTransactionInfo();
mTransactionsInProgress.Put(aDatabaseId, dbTransactionInfo);
}
DatabaseTransactionInfo::TransactionHashtable& transactionsInProgress =
dbTransactionInfo->transactions;
TransactionInfo* info = transactionsInProgress.Get(aTransactionId);
if (info) {
// We recognize this one.
return *info->queue;
}
TransactionInfo* transactionInfo =
new TransactionInfo(this,
aTransactionId,
aDatabaseId,
aObjectStoreNames,
aMode,
aBackgroundChildLoggingId,
aLoggingSerialNumber);
dbTransactionInfo->transactions.Put(aTransactionId, transactionInfo);;
for (uint32_t index = 0, count = aObjectStoreNames.Length(); index < count;
index++) {
TransactionInfoPair* blockInfo =
dbTransactionInfo->blockingTransactions.Get(aObjectStoreNames[index]);
if (!blockInfo) {
blockInfo = new TransactionInfoPair();
blockInfo->lastBlockingReads = nullptr;
dbTransactionInfo->blockingTransactions.Put(aObjectStoreNames[index],
blockInfo);
}
// Mark what we are blocking on.
if (blockInfo->lastBlockingReads) {
TransactionInfo* blockingInfo = blockInfo->lastBlockingReads;
transactionInfo->blockedOn.PutEntry(blockingInfo);
blockingInfo->blocking.PutEntry(transactionInfo);
}
if (aMode == IDBTransaction::READ_WRITE &&
blockInfo->lastBlockingWrites.Length()) {
for (uint32_t index = 0,
count = blockInfo->lastBlockingWrites.Length(); index < count;
index++) {
TransactionInfo* blockingInfo = blockInfo->lastBlockingWrites[index];
transactionInfo->blockedOn.PutEntry(blockingInfo);
blockingInfo->blocking.PutEntry(transactionInfo);
}
}
if (aMode == IDBTransaction::READ_WRITE) {
blockInfo->lastBlockingReads = transactionInfo;
blockInfo->lastBlockingWrites.Clear();
}
else {
blockInfo->lastBlockingWrites.AppendElement(transactionInfo);
}
}
if (!transactionInfo->blockedOn.Count()) {
transactionInfo->queue->Unblock();
}
return *transactionInfo->queue;
}
void
TransactionThreadPool::Start(uint64_t aTransactionId,
const nsACString& aDatabaseId,
const nsTArray<nsString>& aObjectStoreNames,
uint16_t aMode,
const nsID& aBackgroundChildLoggingId,
int64_t aLoggingSerialNumber,
nsIRunnable* aRunnable)
{
AssertIsOnOwningThread();
MOZ_ASSERT(aTransactionId <= mNextTransactionId);
MOZ_ASSERT(aRunnable);
MOZ_ASSERT(!mShutdownRequested);
TransactionQueue& queue = CreateQueueForTransaction(aTransactionId,
aDatabaseId,
aObjectStoreNames,
aMode,
aBackgroundChildLoggingId,
aLoggingSerialNumber);
queue.Dispatch(aRunnable);
}
void
TransactionThreadPool::Dispatch(uint64_t aTransactionId,
const nsACString& aDatabaseId,
nsIRunnable* aRunnable,
bool aFinish,
FinishCallback* aFinishCallback)
{
MOZ_ASSERT(aTransactionId <= mNextTransactionId);
TransactionQueue* queue = GetQueueForTransaction(aTransactionId, aDatabaseId);
MOZ_ASSERT(queue, "Passed an invalid transaction id!");
queue->Dispatch(aRunnable);
if (aFinish) {
queue->Finish(aFinishCallback);
}
}
void
TransactionThreadPool::WaitForDatabasesToComplete(
nsTArray<nsCString>& aDatabaseIds,
nsIRunnable* aCallback)
{
AssertIsOnOwningThread();
NS_ASSERTION(!aDatabaseIds.IsEmpty(), "No databases to wait on!");
NS_ASSERTION(aCallback, "Null pointer!");
nsAutoPtr<DatabasesCompleteCallback> callback(
new DatabasesCompleteCallback());
callback->mCallback = aCallback;
callback->mDatabaseIds.SwapElements(aDatabaseIds);
if (!MaybeFireCallback(callback)) {
mCompleteCallbacks.AppendElement(callback.forget());
}
}
// static
PLDHashOperator
TransactionThreadPool::CollectTransactions(const uint64_t& aTransactionId,
TransactionInfo* aValue,
void* aUserArg)
{
nsAutoTArray<TransactionInfo*, 50>* transactionArray =
static_cast<nsAutoTArray<TransactionInfo*, 50>*>(aUserArg);
transactionArray->AppendElement(aValue);
return PL_DHASH_NEXT;
}
struct MOZ_STACK_CLASS TransactionSearchInfo
{
explicit TransactionSearchInfo(const nsACString& aDatabaseId)
: databaseId(aDatabaseId)
, found(false)
{
}
nsCString databaseId;
bool found;
};
// static
PLDHashOperator
TransactionThreadPool::FindTransaction(const uint64_t& aTransactionId,
TransactionInfo* aValue,
void* aUserArg)
{
TransactionSearchInfo* info = static_cast<TransactionSearchInfo*>(aUserArg);
if (aValue->databaseId == info->databaseId) {
info->found = true;
return PL_DHASH_STOP;
}
return PL_DHASH_NEXT;
}
bool
TransactionThreadPool::MaybeFireCallback(DatabasesCompleteCallback* aCallback)
{
AssertIsOnOwningThread();
MOZ_ASSERT(aCallback);
MOZ_ASSERT(!aCallback->mDatabaseIds.IsEmpty());
MOZ_ASSERT(aCallback->mCallback);
PROFILER_LABEL("IndexedDB",
"TransactionThreadPool::MaybeFireCallback",
js::ProfileEntry::Category::STORAGE);
for (uint32_t count = aCallback->mDatabaseIds.Length(), index = 0;
index < count;
index++) {
const nsCString& databaseId = aCallback->mDatabaseIds[index];
MOZ_ASSERT(!databaseId.IsEmpty());
if (mTransactionsInProgress.Get(databaseId, nullptr)) {
return false;
}
}
aCallback->mCallback->Run();
return true;
}
TransactionThreadPool::
TransactionQueue::TransactionQueue(TransactionThreadPool* aThreadPool,
uint64_t aTransactionId,
const nsACString& aDatabaseId,
const nsTArray<nsString>& aObjectStoreNames,
uint16_t aMode,
const nsID& aBackgroundChildLoggingId,
int64_t aLoggingSerialNumber)
: mMonitor("TransactionQueue::mMonitor")
, mOwningThreadPool(aThreadPool)
, mTransactionId(aTransactionId)
, mBackgroundChildLoggingId(aBackgroundChildLoggingId)
, mLoggingSerialNumber(aLoggingSerialNumber)
, mDatabaseId(aDatabaseId)
, mObjectStoreNames(aObjectStoreNames)
, mMode(aMode)
, mShouldFinish(false)
#ifdef MOZ_NUWA_PROCESS
, mThread(nullptr)
#endif
{
MOZ_ASSERT(aThreadPool);
aThreadPool->AssertIsOnOwningThread();
}
void
TransactionThreadPool::TransactionQueue::Unblock()
{
MonitorAutoLock lock(mMonitor);
// NB: Finish may be called before Unblock.
MOZ_ALWAYS_TRUE(NS_SUCCEEDED(
mOwningThreadPool->mThreadPool->Dispatch(this, NS_DISPATCH_NORMAL)));
}
void
TransactionThreadPool::TransactionQueue::Dispatch(nsIRunnable* aRunnable)
{
MonitorAutoLock lock(mMonitor);
NS_ASSERTION(!mShouldFinish, "Dispatch called after Finish!");
mQueue.AppendElement(aRunnable);
#ifdef MOZ_NUWA_PROCESS
if (mThread) {
mThread->SetWorking();
}
#endif
mMonitor.Notify();
}
void
TransactionThreadPool::TransactionQueue::Finish(FinishCallback* aFinishCallback)
{
MonitorAutoLock lock(mMonitor);
NS_ASSERTION(!mShouldFinish, "Finish called more than once!");
mShouldFinish = true;
mFinishCallback = aFinishCallback;
mMonitor.Notify();
}
NS_IMPL_ISUPPORTS_INHERITED0(TransactionThreadPool::TransactionQueue,
nsRunnable)
NS_IMETHODIMP
TransactionThreadPool::TransactionQueue::Run()
{
PROFILER_LABEL("IndexedDB",
"TransactionThreadPool::TransactionQueue""Run",
js::ProfileEntry::Category::STORAGE);
IDB_LOG_MARK("IndexedDB %s: Parent Transaction[%lld]: "
"Beginning database work",
"IndexedDB %s: P T[%lld]: DB Start",
IDB_LOG_ID_STRING(mBackgroundChildLoggingId),
mLoggingSerialNumber);
nsAutoTArray<nsCOMPtr<nsIRunnable>, 10> queue;
nsRefPtr<FinishCallback> finishCallback;
bool shouldFinish = false;
#ifdef MOZ_NUWA_PROCESS
mThread = static_cast<nsThread*>(NS_GetCurrentThread());
// Set ourself as working thread. We can reset later if we found
// our queue is empty.
mThread->SetWorking();
#endif
do {
NS_ASSERTION(queue.IsEmpty(), "Should have cleared this!");
{
MonitorAutoLock lock(mMonitor);
while (!mShouldFinish && mQueue.IsEmpty()) {
#ifdef MOZ_NUWA_PROCESS
mThread->SetIdle();
#endif
if (NS_FAILED(mMonitor.Wait())) {
NS_ERROR("Failed to wait!");
}
}
mQueue.SwapElements(queue);
if (mShouldFinish) {
mFinishCallback.swap(finishCallback);
shouldFinish = true;
}
}
uint32_t count = queue.Length();
for (uint32_t index = 0; index < count; index++) {
#ifdef DEBUG
if (kDEBUGThreadSleepMS) {
MOZ_ALWAYS_TRUE(
PR_Sleep(PR_MillisecondsToInterval(kDEBUGThreadSleepMS)) ==
PR_SUCCESS);
}
#endif // DEBUG
nsCOMPtr<nsIRunnable>& runnable = queue[index];
runnable->Run();
runnable = nullptr;
}
if (count) {
queue.Clear();
}
} while (!shouldFinish);
#ifdef MOZ_NUWA_PROCESS
mThread = nullptr;
#endif
#ifdef DEBUG
if (kDEBUGThreadSleepMS) {
MOZ_ALWAYS_TRUE(
PR_Sleep(PR_MillisecondsToInterval(kDEBUGThreadSleepMS)) == PR_SUCCESS);
}
#endif // DEBUG
IDB_LOG_MARK("IndexedDB %s: Parent Transaction[%lld]: "
"Finished database work",
"IndexedDB %s: P T[%lld]: DB End",
IDB_LOG_ID_STRING(mBackgroundChildLoggingId),
mLoggingSerialNumber);
nsRefPtr<FinishTransactionRunnable> finishTransactionRunnable =
new FinishTransactionRunnable(mOwningThreadPool.forget(),
mTransactionId,
mDatabaseId,
mObjectStoreNames,
mMode,
finishCallback.forget());
finishTransactionRunnable->Dispatch();
return NS_OK;
}
TransactionThreadPool::
FinishTransactionRunnable::FinishTransactionRunnable(
already_AddRefed<TransactionThreadPool> aThreadPool,
uint64_t aTransactionId,
const nsACString& aDatabaseId,
const nsTArray<nsString>& aObjectStoreNames,
uint16_t aMode,
already_AddRefed<FinishCallback> aFinishCallback)
: mThreadPool(Move(aThreadPool)),
mFinishCallback(aFinishCallback),
mTransactionId(aTransactionId),
mDatabaseId(aDatabaseId),
mObjectStoreNames(aObjectStoreNames),
mMode(aMode)
{
NS_ASSERTION(!NS_IsMainThread(), "Wrong thread!");
}
NS_IMPL_ISUPPORTS_INHERITED0(TransactionThreadPool::FinishTransactionRunnable,
nsRunnable)
NS_IMETHODIMP
TransactionThreadPool::
FinishTransactionRunnable::Run()
{
MOZ_ASSERT(mThreadPool);
mThreadPool->AssertIsOnOwningThread();
PROFILER_LABEL("IndexedDB",
"TransactionThreadPool::FinishTransactionRunnable::Run",
js::ProfileEntry::Category::STORAGE);
nsRefPtr<TransactionThreadPool> threadPool;
mThreadPool.swap(threadPool);
nsRefPtr<FinishCallback> callback;
mFinishCallback.swap(callback);
if (callback) {
callback->TransactionFinishedBeforeUnblock();
}
threadPool->FinishTransaction(mTransactionId,
mDatabaseId,
mObjectStoreNames,
mMode);
if (callback) {
callback->TransactionFinishedAfterUnblock();
}
return NS_OK;
}
#ifdef BUILD_THREADPOOL_LISTENER
NS_IMPL_ISUPPORTS(TransactionThreadPoolListener, nsIThreadPoolListener)
NS_IMETHODIMP
TransactionThreadPoolListener::OnThreadCreated()
{
MOZ_ASSERT(!NS_IsMainThread());
#ifdef MOZ_ENABLE_PROFILER_SPS
char aLocal;
profiler_register_thread("IndexedDB Transaction", &aLocal);
#endif // MOZ_ENABLE_PROFILER_SPS
#ifdef DEBUG
if (kDEBUGThreadPriority != nsISupportsPriority::PRIORITY_NORMAL) {
NS_WARNING("TransactionThreadPool thread debugging enabled, priority has "
"been modified!");
nsCOMPtr<nsISupportsPriority> thread =
do_QueryInterface(NS_GetCurrentThread());
MOZ_ASSERT(thread);
MOZ_ALWAYS_TRUE(NS_SUCCEEDED(thread->SetPriority(kDEBUGThreadPriority)));
}
#endif // DEBUG
return NS_OK;
}
NS_IMETHODIMP
TransactionThreadPoolListener::OnThreadShuttingDown()
{
MOZ_ASSERT(!NS_IsMainThread());
#ifdef MOZ_ENABLE_PROFILER_SPS
profiler_unregister_thread();
#endif // MOZ_ENABLE_PROFILER_SPS
return NS_OK;
}
#endif // BUILD_THREADPOOL_LISTENER
} // namespace indexedDB
} // namespace dom
} // namespace mozilla
Ссылка в новой задаче Показать git blame Копировать постоянную ссылку