sousa-gecko/xpcom/threads/TaskQueue.cpp

/* 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 "mozilla/TaskQueue.h"

#include "mozilla/FlowMarkers.h"
#include "mozilla/ProfilerRunnable.h"
#include "nsIEventTarget.h"
#include "nsITargetShutdownTask.h"
#include "nsQueryObject.h"
#include "nsThreadUtils.h"

namespace mozilla {

static LazyLogModule sTaskQueueLog("TaskQueue");

#define LOG_TQ(level, msg, ...) \
  MOZ_LOG(sTaskQueueLog, level, (msg, ##__VA_ARGS__))

RefPtr<TaskQueue> TaskQueue::Create(
    already_AddRefed<nsIEventTarget> aTarget, StaticString aName,
    enum TailDispatchPolicy aTailDispatchPolicy) {
  nsCOMPtr<nsIEventTarget> target(std::move(aTarget));
  LOG_TQ(LogLevel::Debug,
         "Creating TaskQueue '%s' on target %p (tail-dispatch policy %s)",
         aName.get(), target.get(), EnumValueToString(aTailDispatchPolicy));

  RefPtr<TaskQueue> queue =
      new TaskQueue(do_AddRef(target), aName, aTailDispatchPolicy);

  return queue;
}

TaskQueue::TaskQueue(already_AddRefed<nsIEventTarget> aTarget,
                     const char* aName, enum TailDispatchPolicy aPolicy)
    : AbstractThread(aPolicy),
      mTarget(aTarget),
      mQueueMonitor("TaskQueue::Queue"),
      mTailDispatcher(nullptr),
      mIsTargetShutdownTaskRegistered(false),
      mIsRunning(false),
      mIsShutdown(false),
      mName(aName) {}

TaskQueue::~TaskQueue() {
  LOG_TQ(LogLevel::Debug, "Destroying TaskQueue '%s'", mName);
  // A TaskQueue with shutdown tasks deserves a regular shutdown.
  // Note that if the target SUPPORTS_SHUTDOWN_TASK_DISPATCH the TaskQueue will
  // be kept alive until explicit (or target) shutdown, anyways.
  MOZ_ASSERT(mIsShutdown || mShutdownTasks.IsEmpty());
}

NS_IMPL_ADDREF(TaskQueue)
NS_IMPL_RELEASE(TaskQueue)

NS_INTERFACE_MAP_BEGIN(TaskQueue)
  NS_INTERFACE_MAP_ENTRY(nsIDirectTaskDispatcher)
  NS_INTERFACE_MAP_ENTRY(nsISerialEventTarget)
  NS_INTERFACE_MAP_ENTRY(nsIEventTarget)
  NS_INTERFACE_MAP_ENTRY(nsITargetShutdownTask)
  NS_INTERFACE_MAP_ENTRY_CONCRETE(TaskQueue)
NS_INTERFACE_MAP_END

TaskDispatcher& TaskQueue::TailDispatcher() {
  MOZ_ASSERT(IsCurrentThreadIn());
  MOZ_ASSERT(mTailDispatcher);
  return *mTailDispatcher;
}

void TaskQueue::TargetShutdown() {
  // Nobody needs to wait for the promise as the Runner will ensure all
  // dispatched tasks are completed before the TaskQueue is destroyed
  // given the target SUPPORTS_SHUTDOWN_TASK_DISPATCH.
  LOG_TQ(LogLevel::Debug, "TaskQueue::TargetShutdown '%s'", mName);
  BeginShutdown();
}

void TaskQueue::MaybeUnregisterTargetShutdownTask() {
  if (mIsTargetShutdownTaskRegistered) {
    mTarget->UnregisterShutdownTask(this);
    // We cannot always expect success here because the target might shut
    // down already and this call might be an indirect consequence through
    // some other target shutdown task running first.
    mIsTargetShutdownTaskRegistered = false;
  }
}

// Note aRunnable is passed by ref to support conditional ownership transfer.
// See Dispatch() in TaskQueue.h for more details.
nsresult TaskQueue::DispatchLocked(nsCOMPtr<nsIRunnable>& aRunnable,
                                   DispatchFlags aFlags,
                                   DispatchReason aReason) {
  mQueueMonitor.AssertCurrentThreadOwns();

  // Continue to allow dispatches after shutdown until the last runnable has
  // been processed, at which point no more runnables will be accepted.
  if (mIsShutdown) {
    LOG_TQ(LogLevel::Debug,
           "TaskQueue::DispatchLocked '%s' %s dispatch during shutdown", mName,
           mIsRunning ? "accepting" : "rejecting");
    if (!mIsRunning) {
      return NS_ERROR_ILLEGAL_DURING_SHUTDOWN;
    }
  }

  AbstractThread* currentThread;
  if (aReason != TailDispatch && (currentThread = GetCurrent()) &&
      RequiresTailDispatch(currentThread) &&
      currentThread->IsTailDispatcherAvailable()) {
    return currentThread->TailDispatcher().AddTask(this, aRunnable.forget());
  }

  PROFILER_MARKER("TaskQueue::DispatchLocked", OTHER,
                  {MarkerStack::MaybeCapture(
                      profiler_feature_active(ProfilerFeature::Flows))},
                  FlowMarker, Flow::FromPointer(aRunnable.get()));
  LogRunnable::LogDispatch(aRunnable);
  mTasks.EmplaceBack(TaskStruct{std::move(aRunnable), aFlags});

  if (mIsRunning) {
    return NS_OK;
  }
  auto runner = MakeRefPtr<Runner>(this, mTarget, mObserver, std::move(mTasks));
  nsresult rv =
      mTarget->Dispatch(runner.forget(), aFlags | NS_DISPATCH_FALLIBLE);
  if (NS_FAILED(rv)) {
    NS_WARNING("Failed to dispatch runnable to run TaskQueue");
    return rv;
  }
  mIsRunning = true;

  return NS_OK;
}

nsresult TaskQueue::RegisterShutdownTask(nsITargetShutdownTask* aTask) {
  NS_ENSURE_ARG(aTask);

  LOG_TQ(LogLevel::Debug,
         "TaskQueue::RegisterShutdownTask '%s' registering shutdown task %p",
         mName, aTask);
  MonitorAutoLock mon(mQueueMonitor);
  if (mIsShutdown) {
    return NS_ERROR_UNEXPECTED;
  }
  if (!mIsTargetShutdownTaskRegistered && mShutdownTasks.IsEmpty()) {
    FeatureFlags f = mTarget->GetFeatures();
    if ((f & SUPPORTS_SHUTDOWN_TASKS) &&
        (f & SUPPORTS_SHUTDOWN_TASK_DISPATCH)) {
      MOZ_TRY(mTarget->RegisterShutdownTask(this));
      mIsTargetShutdownTaskRegistered = true;
    }
  }
  return mShutdownTasks.AddTask(aTask);
}

nsresult TaskQueue::UnregisterShutdownTask(nsITargetShutdownTask* aTask) {
  NS_ENSURE_ARG(aTask);

  LOG_TQ(
      LogLevel::Debug,
      "TaskQueue::UnregisterShutdownTask '%s' unregistering shutdown task %p",
      mName, aTask);
  MonitorAutoLock mon(mQueueMonitor);
  nsresult rv = mShutdownTasks.RemoveTask(aTask);
  if (mShutdownTasks.IsEmpty()) {
    MaybeUnregisterTargetShutdownTask();
  }
  return rv;
}

nsIEventTarget::FeatureFlags TaskQueue::GetFeatures() {
  FeatureFlags supports = SUPPORTS_BASE;
  nsCOMPtr<nsIEventTarget> target;
  {
    MonitorAutoLock mon(mQueueMonitor);
    target = mTarget;
  }
  if (target) {
    supports = target->GetFeatures();
  }
  // If the target does not SUPPORTS_SHUTDOWN_TASKS/_SHUTDOWN_TASK_DISPATCH, we
  // still support SHUTDOWN_TASKS but we cannot guarantee they're executed on
  // target shutdown. See bug 2011046 where we might want to change this.
  return supports | SUPPORTS_SHUTDOWN_TASKS;
}

void TaskQueue::AwaitIdle() {
  MonitorAutoLock mon(mQueueMonitor);
  AwaitIdleLocked();
}

void TaskQueue::AwaitIdleLocked() {
  // Make sure there are no tasks for this queue waiting in the caller's tail
  // dispatcher.
  MOZ_ASSERT_IF(AbstractThread::GetCurrent(),
                !AbstractThread::GetCurrent()->HasTailTasksFor(this));

  mQueueMonitor.AssertCurrentThreadOwns();
  MOZ_ASSERT(mIsRunning || mTasks.IsEmpty());
  while (mIsRunning) {
    mQueueMonitor.Wait();
  }
  LOG_TQ(LogLevel::Debug, "TaskQueue::AwaitIdleLocked '%s' is now idle", mName);
}

void TaskQueue::AwaitShutdownAndIdle() {
  MOZ_ASSERT(!IsCurrentThreadIn());
  // Make sure there are no tasks for this queue waiting in the caller's tail
  // dispatcher.
  MOZ_ASSERT_IF(AbstractThread::GetCurrent(),
                !AbstractThread::GetCurrent()->HasTailTasksFor(this));

  MonitorAutoLock mon(mQueueMonitor);
  while (!mIsShutdown) {
    mQueueMonitor.Wait();
  }
  AwaitIdleLocked();
}

RefPtr<ShutdownPromise> TaskQueue::BeginShutdown() {
  LOG_TQ(LogLevel::Debug, "TaskQueue::BeginShutdown '%s'", mName);
  // Dispatch any tasks for this queue waiting in the caller's tail dispatcher,
  // since this is the last opportunity to do so.
  if (AbstractThread* currentThread = AbstractThread::GetCurrent()) {
    currentThread->TailDispatchTasksFor(this);
  }

  MonitorAutoLock mon(mQueueMonitor);
  if (!mIsShutdown) {
    MaybeUnregisterTargetShutdownTask();
    // Dispatch all cleanup tasks to the queue before we put it into full
    // shutdown.
    TargetShutdownTaskSet::TasksArray tasks = mShutdownTasks.Extract();
    for (auto& task : tasks) {
      LOG_TQ(LogLevel::Debug,
             "TaskQueue::BeginShutdown '%s' dispatching shutdown task %p",
             mName, task.get());
      nsCOMPtr runnable{task->AsRunnable()};
      MOZ_ALWAYS_SUCCEEDS(
          DispatchLocked(runnable, NS_DISPATCH_NORMAL, TailDispatch));
    }
    mIsShutdown = true;
  }

  RefPtr<ShutdownPromise> p = mShutdownPromise.Ensure(__func__);
  MaybeResolveShutdown();
  mon.NotifyAll();
  return p;
}

void TaskQueue::MaybeResolveShutdown() {
  mQueueMonitor.AssertCurrentThreadOwns();
  if (mIsShutdown && !mIsRunning) {
    LOG_TQ(LogLevel::Debug, "TaskQueue::MaybeResolveShutdown '%s' resolve",
           mName);
    MOZ_ASSERT(!mIsTargetShutdownTaskRegistered);
    mShutdownPromise.ResolveIfExists(true, __func__);
    // Disconnect from our target as we won't try to dispatch any more events.
    mTarget = nullptr;
    mObserver = nullptr;
  }
}

bool TaskQueue::IsEmpty() {
  MonitorAutoLock mon(mQueueMonitor);
  // NOTE: If `mTasks` is empty, we may still have queued tasks which are owned
  // by the active runner. As we will always start running as soon as a task has
  // been added, we can skip checking `mTasks` and can just check `mIsRunning`.
  return !mIsRunning;
}

bool TaskQueue::IsCurrentThreadIn() const {
  bool in = mRunningThread == PR_GetCurrentThread();
  return in;
}

void TaskQueue::SetObserver(Observer* aObserver) {
  MonitorAutoLock mon(mQueueMonitor);
  MOZ_ASSERT_IF(aObserver, !mObserver);
  mObserver = std::move(aObserver);
}

nsresult TaskQueue::Runner::Run() {
  MOZ_ASSERT(mNextTask < mTasks.Length(), "No tasks to do?");

  // Process mTasks[mNextTask] with an AutoTaskGuard on the stack.
  {
    AutoTaskGuard g(mQueue, mObserver);

    TaskStruct& task = mTasks[mNextTask++];
    MOZ_ASSERT(task.event);

    LogRunnable::Run log(task.event);

    AUTO_PROFILE_FOLLOWING_RUNNABLE(task.event);
    task.event->Run();

    // Drop the reference to task.event before AutoTaskGuard is destroyed. The
    // event will hold a reference to the object it's calling, and we don't want
    // to keep it alive, it may be making assumptions what holds references to
    // it. This is especially the case if the object is waiting for us to
    // shutdown, so that it can shutdown (like in the MediaDecoderStateMachine's
    // SHUTDOWN case).
    task.event = nullptr;
  }

  // If mTasks is exhausted, check to see if new tasks have been dispatched.
  if (mNextTask >= mTasks.Length()) {
    MonitorAutoLock mon(mQueue->mQueueMonitor);
    MOZ_ASSERT(mQueue->mIsRunning);

    // If mQueue->mTasks is empty, we're done for now, and can stop running.
    if (mQueue->mTasks.IsEmpty()) {
      mQueue->mIsRunning = false;
      mQueue->MaybeResolveShutdown();
      mon.NotifyAll();
      return NS_OK;
    }

    // Otherwise, reload from mQueue and continue running.
    mTarget = mQueue->mTarget;
    mObserver = mQueue->mObserver;
    mTasks = std::move(mQueue->mTasks);
    mNextTask = 0;
  }

  // We still have tasks to execute. Dispatch this Runner to the target to
  // ensure it runs again. Note that we don't just run in a loop here. This
  // keeps the ratio of Runnables dispatched to the TaskQueue and the underlying
  // EventTarget as 1:1, hopefully avoiding flooding.
  MOZ_ASSERT(mNextTask < mTasks.Length());
  nsresult rv =
      mTarget->Dispatch(this, mTasks[mNextTask].flags | NS_DISPATCH_AT_END |
                                  NS_DISPATCH_FALLIBLE);
  if (NS_FAILED(rv)) {
    NS_WARNING("Underlying EventTarget for TaskQueue not accepting new tasks");

    // The Dispatch to mTarget failed. Unfortunately we cannot recover at this
    // point, so shut down the TaskQueue, as we can no longer accept new events.
    MonitorAutoLock mon(mQueue->mQueueMonitor);
    mQueue->mIsRunning = false;
    mQueue->mIsShutdown = true;
    mQueue->MaybeUnregisterTargetShutdownTask();
    mQueue->MaybeResolveShutdown();
    mon.NotifyAll();
  }

  return NS_OK;
}

//-----------------------------------------------------------------------------
// nsIDirectTaskDispatcher
//-----------------------------------------------------------------------------

NS_IMETHODIMP
TaskQueue::DispatchDirectTask(already_AddRefed<nsIRunnable> aEvent) {
  if (!IsCurrentThreadIn()) {
    return NS_ERROR_FAILURE;
  }
  mDirectTasks.AddTask(std::move(aEvent));
  return NS_OK;
}

NS_IMETHODIMP TaskQueue::DrainDirectTasks() {
  if (!IsCurrentThreadIn()) {
    return NS_ERROR_FAILURE;
  }
  mDirectTasks.DrainTasks();
  return NS_OK;
}

NS_IMETHODIMP TaskQueue::HaveDirectTasks(bool* aValue) {
  if (!IsCurrentThreadIn()) {
    return NS_ERROR_FAILURE;
  }

  *aValue = mDirectTasks.HaveTasks();
  return NS_OK;
}

#undef LOG_TQ

}  // namespace mozilla