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