JobScheduler.cpp

/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
/* vim: set ts=8 sts=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 "JobScheduler.h"
#include "Logging.h"

namespace mozilla {
namespace gfx {

JobScheduler* JobScheduler::sSingleton = nullptr;

bool JobScheduler::Init(uint32_t aNumThreads, uint32_t aNumQueues) {
  MOZ_ASSERT(!sSingleton);
  MOZ_ASSERT(aNumThreads >= aNumQueues);

  sSingleton = new JobScheduler();
  sSingleton->mNextQueue = 0;

  for (uint32_t i = 0; i < aNumQueues; ++i) {
    sSingleton->mDrawingQueues.push_back(new MultiThreadedJobQueue());
  }

  for (uint32_t i = 0; i < aNumThreads; ++i) {
    sSingleton->mWorkerThreads.push_back(
        WorkerThread::Create(sSingleton->mDrawingQueues[i % aNumQueues]));
  }
  return true;
}

void JobScheduler::ShutDown() {
  MOZ_ASSERT(IsEnabled());
  if (!IsEnabled()) {
    return;
  }

  for (auto queue : sSingleton->mDrawingQueues) {
    queue->ShutDown();
    delete queue;
  }

  for (WorkerThread* thread : sSingleton->mWorkerThreads) {
    // this will block until the thread is joined.
    delete thread;
  }

  sSingleton->mWorkerThreads.clear();
  delete sSingleton;
  sSingleton = nullptr;
}

JobStatus JobScheduler::ProcessJob(Job* aJob) {
  MOZ_ASSERT(aJob);
  auto status = aJob->Run();
  if (status == JobStatus::Error || status == JobStatus::Complete) {
    delete aJob;
  }
  return status;
}

void JobScheduler::SubmitJob(Job* aJob) {
  MOZ_ASSERT(aJob);
  RefPtr<SyncObject> start = aJob->GetStartSync();
  if (start && start->Register(aJob)) {
    // The Job buffer starts with a non-signaled sync object, it
    // is now registered in the list of task buffers waiting on the
    // sync object, so we should not place it in the queue.
    return;
  }

  GetQueueForJob(aJob)->SubmitJob(aJob);
}

void JobScheduler::Join(SyncObject* aCompletion) {
  RefPtr<EventObject> waitForCompletion = new EventObject();
  JobScheduler::SubmitJob(new SetEventJob(waitForCompletion, aCompletion));
  waitForCompletion->Wait();
}

MultiThreadedJobQueue* JobScheduler::GetQueueForJob(Job* aJob) {
  return aJob->IsPinnedToAThread() ? aJob->GetWorkerThread()->GetJobQueue()
                                   : GetDrawingQueue();
}

Job::Job(SyncObject* aStart, SyncObject* aCompletion, WorkerThread* aThread)
    : mNextWaitingJob(nullptr),
      mStartSync(aStart),
      mCompletionSync(aCompletion),
      mPinToThread(aThread) {
  if (mStartSync) {
    mStartSync->AddSubsequent(this);
  }
  if (mCompletionSync) {
    mCompletionSync->AddPrerequisite(this);
  }
}

Job::~Job() {
  if (mCompletionSync) {
    // printf(" -- Job %p dtor completion %p\n", this, mCompletionSync);
    mCompletionSync->Signal();
    mCompletionSync = nullptr;
  }
}

JobStatus SetEventJob::Run() {
  mEvent->Set();
  return JobStatus::Complete;
}

SetEventJob::SetEventJob(EventObject* aEvent, SyncObject* aStart,
                         SyncObject* aCompletion, WorkerThread* aWorker)
    : Job(aStart, aCompletion, aWorker), mEvent(aEvent) {}

SetEventJob::~SetEventJob() {}

SyncObject::SyncObject(uint32_t aNumPrerequisites)
    : mSignals(aNumPrerequisites),
      mFirstWaitingJob(nullptr)
#ifdef DEBUG
      ,
      mNumPrerequisites(aNumPrerequisites),
      mAddedPrerequisites(0)
#endif
{
}

SyncObject::~SyncObject() { MOZ_ASSERT(mFirstWaitingJob == nullptr); }

bool SyncObject::Register(Job* aJob) {
  MOZ_ASSERT(aJob);

  // For now, ensure that when we schedule the first subsequent, we have already
  // created all of the prerequisites. This is an arbitrary restriction because
  // we specify the number of prerequisites in the constructor, but in the
  // typical scenario, if the assertion FreezePrerequisite blows up here it
  // probably means we got the initial nmber of prerequisites wrong. We can
  // decide to remove this restriction if needed.
  FreezePrerequisites();

  int32_t signals = mSignals;

  if (signals > 0) {
    AddWaitingJob(aJob);
    // Since Register and Signal can be called concurrently, it can happen that
    // reading mSignals in Register happens before decrementing mSignals in
    // Signal, but SubmitWaitingJobs happens before AddWaitingJob. This ordering
    // means the SyncObject ends up in the signaled state with a task sitting in
    // the waiting list. To prevent that we check mSignals a second time and
    // submit again if signals reached zero in the mean time. We do this instead
    // of holding a mutex around mSignals+mJobs to reduce lock contention.
    int32_t signals2 = mSignals;
    if (signals2 == 0) {
      SubmitWaitingJobs();
    }
    return true;
  }

  return false;
}

void SyncObject::Signal() {
  int32_t signals = --mSignals;
  MOZ_ASSERT(signals >= 0);

  if (signals == 0) {
    SubmitWaitingJobs();
  }
}

void SyncObject::AddWaitingJob(Job* aJob) {
  // Push (using atomics) the task into the list of waiting tasks.
  for (;;) {
    Job* first = mFirstWaitingJob;
    aJob->mNextWaitingJob = first;
    if (mFirstWaitingJob.compareExchange(first, aJob)) {
      break;
    }
  }
}

void SyncObject::SubmitWaitingJobs() {
  // Scheduling the tasks can cause code that modifies <this>'s reference
  // count to run concurrently, and cause the caller of this function to
  // be owned by another thread. We need to make sure the reference count
  // does not reach 0 on another thread before the end of this method, so
  // hold a strong ref to prevent that!
  RefPtr<SyncObject> kungFuDeathGrip(this);

  // First atomically swap mFirstWaitingJob and waitingJobs...
  Job* waitingJobs = nullptr;
  for (;;) {
    waitingJobs = mFirstWaitingJob;
    if (mFirstWaitingJob.compareExchange(waitingJobs, nullptr)) {
      break;
    }
  }

  // ... and submit all of the waiting tasks in waitingJob now that they belong
  // to this thread.
  while (waitingJobs) {
    Job* next = waitingJobs->mNextWaitingJob;
    waitingJobs->mNextWaitingJob = nullptr;
    JobScheduler::GetQueueForJob(waitingJobs)->SubmitJob(waitingJobs);
    waitingJobs = next;
  }
}

bool SyncObject::IsSignaled() { return mSignals == 0; }

void SyncObject::FreezePrerequisites() {
  MOZ_ASSERT(mAddedPrerequisites == mNumPrerequisites);
}

void SyncObject::AddPrerequisite(Job* aJob) {
  MOZ_ASSERT(++mAddedPrerequisites <= mNumPrerequisites);
}

void SyncObject::AddSubsequent(Job* aJob) {}

WorkerThread::WorkerThread(MultiThreadedJobQueue* aJobQueue)
    : mQueue(aJobQueue) {
  aJobQueue->RegisterThread();
}

void WorkerThread::Run() {
  SetName("gfx worker");

  for (;;) {
    Job* commands = nullptr;
    if (!mQueue->WaitForJob(commands)) {
      mQueue->UnregisterThread();
      return;
    }

    JobStatus status = JobScheduler::ProcessJob(commands);

    if (status == JobStatus::Error) {
      // Don't try to handle errors for now, but that's open to discussions.
      // I expect errors to be mostly OOM issues.
      gfxDevCrash(LogReason::JobStatusError)
          << "Invalid job status " << (int)status;
    }
  }
}

}  // namespace gfx
}  // namespace mozilla