xpcom/threads/TaskQueue.cpp

   1 /* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
   2 /* vim: set ts=8 sts=2 et sw=2 tw=80: */
   3 /* This Source Code Form is subject to the terms of the Mozilla Public
   4  * License, v. 2.0. If a copy of the MPL was not distributed with this
   5  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
   6
   7 #include "mozilla/TaskQueue.h"
   8
   9 #include "mozilla/DelayedRunnable.h"
  10 #include "mozilla/ProfilerRunnable.h"
  11 #include "nsIEventTarget.h"
  12 #include "nsITargetShutdownTask.h"
  13 #include "nsThreadUtils.h"
  14 #include "nsQueryObject.h"
  15
  16 namespace mozilla {
  17
  18 // Handle for a TaskQueue being tracked by a TaskQueueTracker. When created,
  19 // it is registered with the TaskQueueTracker, and when destroyed it is
  20 // unregistered. Holds a threadsafe weak reference to the TaskQueue.
  21 class TaskQueueTrackerEntry final
  22     : private LinkedListElement<TaskQueueTrackerEntry> {
  23  public:
  24   TaskQueueTrackerEntry(TaskQueueTracker* aTracker,
  25                         const RefPtr<TaskQueue>& aQueue)
  26       : mTracker(aTracker), mQueue(aQueue) {
  27     MutexAutoLock lock(mTracker->mMutex);
  28     mTracker->mEntries.insertFront(this);
  29   }
  30   ~TaskQueueTrackerEntry() {
  31     MutexAutoLock lock(mTracker->mMutex);
  32     removeFrom(mTracker->mEntries);
  33   }
  34
  35   TaskQueueTrackerEntry(const TaskQueueTrackerEntry&) = delete;
  36   TaskQueueTrackerEntry(TaskQueueTrackerEntry&&) = delete;
  37   TaskQueueTrackerEntry& operator=(const TaskQueueTrackerEntry&) = delete;
  38   TaskQueueTrackerEntry& operator=(TaskQueueTrackerEntry&&) = delete;
  39
  40   RefPtr<TaskQueue> GetQueue() const { return RefPtr<TaskQueue>(mQueue); }
  41
  42  private:
  43   friend class LinkedList<TaskQueueTrackerEntry>;
  44   friend class LinkedListElement<TaskQueueTrackerEntry>;
  45
  46   const RefPtr<TaskQueueTracker> mTracker;
  47   const ThreadSafeWeakPtr<TaskQueue> mQueue;
  48 };
  49
  50 RefPtr<TaskQueue> TaskQueue::Create(already_AddRefed<nsIEventTarget> aTarget,
  51                                     const char* aName,
  52                                     bool aSupportsTailDispatch) {
  53   nsCOMPtr<nsIEventTarget> target(std::move(aTarget));
  54   RefPtr<TaskQueue> queue =
  55       new TaskQueue(do_AddRef(target), aName, aSupportsTailDispatch);
  56
  57   // If |target| is a TaskQueueTracker, register this TaskQueue with it. It will
  58   // be unregistered when the TaskQueue is destroyed or shut down.
  59   if (RefPtr<TaskQueueTracker> tracker = do_QueryObject(target)) {
  60     MonitorAutoLock lock(queue->mQueueMonitor);
  61     queue->mTrackerEntry = MakeUnique<TaskQueueTrackerEntry>(tracker, queue);
  62   }
  63
  64   return queue;
  65 }
  66
  67 TaskQueue::TaskQueue(already_AddRefed<nsIEventTarget> aTarget,
  68                      const char* aName, bool aSupportsTailDispatch)
  69     : AbstractThread(aSupportsTailDispatch),
  70       mTarget(aTarget),
  71       mQueueMonitor("TaskQueue::Queue"),
  72       mTailDispatcher(nullptr),
  73       mIsRunning(false),
  74       mIsShutdown(false),
  75       mName(aName) {}
  76
  77 TaskQueue::~TaskQueue() {
  78   // We should never free the TaskQueue if it was destroyed abnormally, meaning
  79   // that all cleanup tasks should be complete if we do.
  80   MOZ_ASSERT(mShutdownTasks.IsEmpty());
  81 }
  82
  83 NS_IMPL_ADDREF_INHERITED(TaskQueue, SupportsThreadSafeWeakPtr<TaskQueue>)
  84 NS_IMPL_RELEASE_INHERITED(TaskQueue, SupportsThreadSafeWeakPtr<TaskQueue>)
  85 NS_IMPL_QUERY_INTERFACE(TaskQueue, nsIDirectTaskDispatcher,
  86                         nsISerialEventTarget, nsIEventTarget)
  87
  88 TaskDispatcher& TaskQueue::TailDispatcher() {
  89   MOZ_ASSERT(IsCurrentThreadIn());
  90   MOZ_ASSERT(mTailDispatcher);
  91   return *mTailDispatcher;
  92 }
  93
  94 // Note aRunnable is passed by ref to support conditional ownership transfer.
  95 // See Dispatch() in TaskQueue.h for more details.
  96 nsresult TaskQueue::DispatchLocked(nsCOMPtr<nsIRunnable>& aRunnable,
  97                                    uint32_t aFlags, DispatchReason aReason) {
  98   mQueueMonitor.AssertCurrentThreadOwns();
  99
 100   // Continue to allow dispatches after shutdown until the last message has been
 101   // processed, at which point no more messages will be accepted.
 102   if (mIsShutdown && !mIsRunning) {
 103     return NS_ERROR_UNEXPECTED;
 104   }
 105
 106   AbstractThread* currentThread;
 107   if (aReason != TailDispatch && (currentThread = GetCurrent()) &&
 108       RequiresTailDispatch(currentThread) &&
 109       currentThread->IsTailDispatcherAvailable()) {
 110     MOZ_ASSERT(aFlags == NS_DISPATCH_NORMAL,
 111                "Tail dispatch doesn't support flags");
 112     return currentThread->TailDispatcher().AddTask(this, aRunnable.forget());
 113   }
 114
 115   LogRunnable::LogDispatch(aRunnable);
 116   mTasks.Push({std::move(aRunnable), aFlags});
 117
 118   if (mIsRunning) {
 119     return NS_OK;
 120   }
 121   RefPtr<nsIRunnable> runner(new Runner(this));
 122   nsresult rv = mTarget->Dispatch(runner.forget(), aFlags);
 123   if (NS_FAILED(rv)) {
 124     NS_WARNING("Failed to dispatch runnable to run TaskQueue");
 125     return rv;
 126   }
 127   mIsRunning = true;
 128
 129   return NS_OK;
 130 }
 131
 132 nsresult TaskQueue::RegisterShutdownTask(nsITargetShutdownTask* aTask) {
 133   NS_ENSURE_ARG(aTask);
 134
 135   MonitorAutoLock mon(mQueueMonitor);
 136   if (mIsShutdown) {
 137     return NS_ERROR_UNEXPECTED;
 138   }
 139
 140   MOZ_ASSERT(!mShutdownTasks.Contains(aTask));
 141   mShutdownTasks.AppendElement(aTask);
 142   return NS_OK;
 143 }
 144
 145 nsresult TaskQueue::UnregisterShutdownTask(nsITargetShutdownTask* aTask) {
 146   NS_ENSURE_ARG(aTask);
 147
 148   MonitorAutoLock mon(mQueueMonitor);
 149   if (mIsShutdown) {
 150     return NS_ERROR_UNEXPECTED;
 151   }
 152
 153   return mShutdownTasks.RemoveElement(aTask) ? NS_OK : NS_ERROR_UNEXPECTED;
 154 }
 155
 156 void TaskQueue::AwaitIdle() {
 157   MonitorAutoLock mon(mQueueMonitor);
 158   AwaitIdleLocked();
 159 }
 160
 161 void TaskQueue::AwaitIdleLocked() {
 162   // Make sure there are no tasks for this queue waiting in the caller's tail
 163   // dispatcher.
 164   MOZ_ASSERT_IF(AbstractThread::GetCurrent(),
 165                 !AbstractThread::GetCurrent()->HasTailTasksFor(this));
 166
 167   mQueueMonitor.AssertCurrentThreadOwns();
 168   MOZ_ASSERT(mIsRunning || mTasks.IsEmpty());
 169   while (mIsRunning) {
 170     mQueueMonitor.Wait();
 171   }
 172 }
 173
 174 void TaskQueue::AwaitShutdownAndIdle() {
 175   MOZ_ASSERT(!IsCurrentThreadIn());
 176   // Make sure there are no tasks for this queue waiting in the caller's tail
 177   // dispatcher.
 178   MOZ_ASSERT_IF(AbstractThread::GetCurrent(),
 179                 !AbstractThread::GetCurrent()->HasTailTasksFor(this));
 180
 181   MonitorAutoLock mon(mQueueMonitor);
 182   while (!mIsShutdown) {
 183     mQueueMonitor.Wait();
 184   }
 185   AwaitIdleLocked();
 186 }
 187 RefPtr<ShutdownPromise> TaskQueue::BeginShutdown() {
 188   // Dispatch any tasks for this queue waiting in the caller's tail dispatcher,
 189   // since this is the last opportunity to do so.
 190   if (AbstractThread* currentThread = AbstractThread::GetCurrent()) {
 191     currentThread->TailDispatchTasksFor(this);
 192   }
 193
 194   MonitorAutoLock mon(mQueueMonitor);
 195   // Dispatch any cleanup tasks to the queue before we put it into full
 196   // shutdown.
 197   for (auto& task : mShutdownTasks) {
 198     nsCOMPtr runnable{task->AsRunnable()};
 199     MOZ_ALWAYS_SUCCEEDS(
 200         DispatchLocked(runnable, NS_DISPATCH_NORMAL, TailDispatch));
 201   }
 202   mShutdownTasks.Clear();
 203   mIsShutdown = true;
 204
 205   RefPtr<ShutdownPromise> p = mShutdownPromise.Ensure(__func__);
 206   MaybeResolveShutdown();
 207   mon.NotifyAll();
 208   return p;
 209 }
 210
 211 void TaskQueue::MaybeResolveShutdown() {
 212   mQueueMonitor.AssertCurrentThreadOwns();
 213   if (mIsShutdown && !mIsRunning) {
 214     mShutdownPromise.ResolveIfExists(true, __func__);
 215     // Disconnect from our target as we won't try to dispatch any more events.
 216     mTrackerEntry = nullptr;
 217     mTarget = nullptr;
 218   }
 219 }
 220
 221 bool TaskQueue::IsEmpty() {
 222   MonitorAutoLock mon(mQueueMonitor);
 223   return mTasks.IsEmpty();
 224 }
 225
 226 bool TaskQueue::IsCurrentThreadIn() const {
 227   bool in = mRunningThread == PR_GetCurrentThread();
 228   return in;
 229 }
 230
 231 nsresult TaskQueue::Runner::Run() {
 232   TaskStruct event;
 233   {
 234     MonitorAutoLock mon(mQueue->mQueueMonitor);
 235     MOZ_ASSERT(mQueue->mIsRunning);
 236     if (mQueue->mTasks.IsEmpty()) {
 237       mQueue->mIsRunning = false;
 238       mQueue->MaybeResolveShutdown();
 239       mon.NotifyAll();
 240       return NS_OK;
 241     }
 242     event = std::move(mQueue->mTasks.FirstElement());
 243     mQueue->mTasks.Pop();
 244   }
 245   MOZ_ASSERT(event.event);
 246
 247   // Note that dropping the queue monitor before running the task, and
 248   // taking the monitor again after the task has run ensures we have memory
 249   // fences enforced. This means that if the object we're calling wasn't
 250   // designed to be threadsafe, it will be, provided we're only calling it
 251   // in this task queue.
 252   {
 253     AutoTaskGuard g(mQueue);
 254     SerialEventTargetGuard tg(mQueue);
 255     {
 256       LogRunnable::Run log(event.event);
 257
 258       AUTO_PROFILE_FOLLOWING_RUNNABLE(event.event);
 259       event.event->Run();
 260
 261       // Drop the reference to event. The event will hold a reference to the
 262       // object it's calling, and we don't want to keep it alive, it may be
 263       // making assumptions what holds references to it. This is especially
 264       // the case if the object is waiting for us to shutdown, so that it
 265       // can shutdown (like in the MediaDecoderStateMachine's SHUTDOWN case).
 266       event.event = nullptr;
 267     }
 268   }
 269
 270   {
 271     MonitorAutoLock mon(mQueue->mQueueMonitor);
 272     if (mQueue->mTasks.IsEmpty()) {
 273       // No more events to run. Exit the task runner.
 274       mQueue->mIsRunning = false;
 275       mQueue->MaybeResolveShutdown();
 276       mon.NotifyAll();
 277       return NS_OK;
 278     }
 279   }
 280
 281   // There's at least one more event that we can run. Dispatch this Runner
 282   // to the target again to ensure it runs again. Note that we don't just
 283   // run in a loop here so that we don't hog the target. This means we may
 284   // run on another thread next time, but we rely on the memory fences from
 285   // mQueueMonitor for thread safety of non-threadsafe tasks.
 286   nsresult rv;
 287   {
 288     MonitorAutoLock mon(mQueue->mQueueMonitor);
 289     rv = mQueue->mTarget->Dispatch(
 290         this, mQueue->mTasks.FirstElement().flags | NS_DISPATCH_AT_END);
 291   }
 292   if (NS_FAILED(rv)) {
 293     // Failed to dispatch, shutdown!
 294     MonitorAutoLock mon(mQueue->mQueueMonitor);
 295     mQueue->mIsRunning = false;
 296     mQueue->mIsShutdown = true;
 297     mQueue->MaybeResolveShutdown();
 298     mon.NotifyAll();
 299   }
 300
 301   return NS_OK;
 302 }
 303
 304 //-----------------------------------------------------------------------------
 305 // nsIDirectTaskDispatcher
 306 //-----------------------------------------------------------------------------
 307
 308 NS_IMETHODIMP
 309 TaskQueue::DispatchDirectTask(already_AddRefed<nsIRunnable> aEvent) {
 310   if (!IsCurrentThreadIn()) {
 311     return NS_ERROR_FAILURE;
 312   }
 313   mDirectTasks.AddTask(std::move(aEvent));
 314   return NS_OK;
 315 }
 316
 317 NS_IMETHODIMP TaskQueue::DrainDirectTasks() {
 318   if (!IsCurrentThreadIn()) {
 319     return NS_ERROR_FAILURE;
 320   }
 321   mDirectTasks.DrainTasks();
 322   return NS_OK;
 323 }
 324
 325 NS_IMETHODIMP TaskQueue::HaveDirectTasks(bool* aValue) {
 326   if (!IsCurrentThreadIn()) {
 327     return NS_ERROR_FAILURE;
 328   }
 329
 330   *aValue = mDirectTasks.HaveTasks();
 331   return NS_OK;
 332 }
 333
 334 nsTArray<RefPtr<TaskQueue>> TaskQueueTracker::GetAllTrackedTaskQueues() {
 335   MutexAutoLock lock(mMutex);
 336   nsTArray<RefPtr<TaskQueue>> queues;
 337   for (auto* entry : mEntries) {
 338     if (auto queue = entry->GetQueue()) {
 339       queues.AppendElement(queue);
 340     }
 341   }
 342   return queues;
 343 }
 344
 345 TaskQueueTracker::~TaskQueueTracker() = default;
 346
 347 }  // namespace mozilla