xpcom/threads/TaskDispatcher.h

   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 #ifndef XPCOM_THREADS_TASKDISPATCHER_H_
   8 #define XPCOM_THREADS_TASKDISPATCHER_H_
   9
  10 #include <queue>
  11
  12 #include "mozilla/AbstractThread.h"
  13 #include "mozilla/Maybe.h"
  14 #include "mozilla/ProfilerRunnable.h"
  15 #include "mozilla/UniquePtr.h"
  16 #include "nsIDirectTaskDispatcher.h"
  17 #include "nsISupportsImpl.h"
  18 #include "nsTArray.h"
  19 #include "nsThreadUtils.h"
  20
  21 namespace mozilla {
  22
  23 class SimpleTaskQueue {
  24  public:
  25   SimpleTaskQueue() = default;
  26   virtual ~SimpleTaskQueue() = default;
  27
  28   void AddTask(already_AddRefed<nsIRunnable> aRunnable) {
  29     if (!mTasks) {
  30       mTasks.emplace();
  31     }
  32     mTasks->push(std::move(aRunnable));
  33   }
  34
  35   void DrainTasks() {
  36     if (!mTasks) {
  37       return;
  38     }
  39     auto& queue = mTasks.ref();
  40     while (!queue.empty()) {
  41       nsCOMPtr<nsIRunnable> r = std::move(queue.front());
  42       queue.pop();
  43       AUTO_PROFILE_FOLLOWING_RUNNABLE(r);
  44       r->Run();
  45     }
  46   }
  47
  48   bool HaveTasks() const { return mTasks && !mTasks->empty(); }
  49
  50  private:
  51   // We use a Maybe<> because (a) when used for DirectTasks it often doesn't get
  52   // anything put into it, and (b) the std::queue implementation in GNU
  53   // libstdc++ does two largish heap allocations when creating a new std::queue.
  54   Maybe<std::queue<nsCOMPtr<nsIRunnable>>> mTasks;
  55 };
  56
  57 /*
  58  * A classic approach to cross-thread communication is to dispatch asynchronous
  59  * runnables to perform updates on other threads. This generally works well, but
  60  * there are sometimes reasons why we might want to delay the actual dispatch of
  61  * these tasks until a specified moment. At present, this is primarily useful to
  62  * ensure that mirrored state gets updated atomically - but there may be other
  63  * applications as well.
  64  *
  65  * TaskDispatcher is a general abstract class that accepts tasks and dispatches
  66  * them at some later point. These groups of tasks are per-target-thread, and
  67  * contain separate queues for several kinds of tasks (see comments  below). -
  68  * "state change tasks" (which run first, and are intended to be used to update
  69  * the value held by mirrors), and regular tasks, which are other arbitrary
  70  * operations that the are gated to run after all the state changes have
  71  * completed.
  72  */
  73 class TaskDispatcher {
  74  public:
  75   TaskDispatcher() = default;
  76   virtual ~TaskDispatcher() = default;
  77
  78   // Direct tasks are run directly (rather than dispatched asynchronously) when
  79   // the tail dispatcher fires. A direct task may cause other tasks to be added
  80   // to the tail dispatcher.
  81   virtual void AddDirectTask(already_AddRefed<nsIRunnable> aRunnable) = 0;
  82
  83   // State change tasks are dispatched asynchronously always run before regular
  84   // tasks. They are intended to be used to update the value held by mirrors
  85   // before any other dispatched tasks are run on the target thread.
  86   virtual void AddStateChangeTask(AbstractThread* aThread,
  87                                   already_AddRefed<nsIRunnable> aRunnable) = 0;
  88
  89   // Regular tasks are dispatched asynchronously, and run after state change
  90   // tasks.
  91   virtual nsresult AddTask(AbstractThread* aThread,
  92                            already_AddRefed<nsIRunnable> aRunnable) = 0;
  93
  94   virtual nsresult DispatchTasksFor(AbstractThread* aThread) = 0;
  95   virtual bool HasTasksFor(AbstractThread* aThread) = 0;
  96   virtual void DrainDirectTasks() = 0;
  97 };
  98
  99 /*
 100  * AutoTaskDispatcher is a stack-scoped TaskDispatcher implementation that fires
 101  * its queued tasks when it is popped off the stack.
 102  */
 103 class AutoTaskDispatcher : public TaskDispatcher {
 104  public:
 105   explicit AutoTaskDispatcher(nsIDirectTaskDispatcher* aDirectTaskDispatcher,
 106                               bool aIsTailDispatcher = false)
 107       : mDirectTaskDispatcher(aDirectTaskDispatcher),
 108         mIsTailDispatcher(aIsTailDispatcher) {}
 109
 110   ~AutoTaskDispatcher() {
 111     // Given that direct tasks may trigger other code that uses the tail
 112     // dispatcher, it's better to avoid processing them in the tail dispatcher's
 113     // destructor. So we require TailDispatchers to manually invoke
 114     // DrainDirectTasks before the AutoTaskDispatcher gets destroyed. In truth,
 115     // this is only necessary in the case where this AutoTaskDispatcher can be
 116     // accessed by the direct tasks it dispatches (true for TailDispatchers, but
 117     // potentially not true for other hypothetical AutoTaskDispatchers). Feel
 118     // free to loosen this restriction to apply only to mIsTailDispatcher if a
 119     // use-case requires it.
 120     MOZ_ASSERT(!HaveDirectTasks());
 121
 122     for (size_t i = 0; i < mTaskGroups.Length(); ++i) {
 123       DispatchTaskGroup(std::move(mTaskGroups[i]));
 124     }
 125   }
 126
 127   bool HaveDirectTasks() {
 128     return mDirectTaskDispatcher && mDirectTaskDispatcher->HaveDirectTasks();
 129   }
 130
 131   void DrainDirectTasks() override {
 132     if (mDirectTaskDispatcher) {
 133       mDirectTaskDispatcher->DrainDirectTasks();
 134     }
 135   }
 136
 137   void AddDirectTask(already_AddRefed<nsIRunnable> aRunnable) override {
 138     MOZ_ASSERT(mDirectTaskDispatcher);
 139     mDirectTaskDispatcher->DispatchDirectTask(std::move(aRunnable));
 140   }
 141
 142   void AddStateChangeTask(AbstractThread* aThread,
 143                           already_AddRefed<nsIRunnable> aRunnable) override {
 144     nsCOMPtr<nsIRunnable> r = aRunnable;
 145     MOZ_RELEASE_ASSERT(r);
 146     EnsureTaskGroup(aThread).mStateChangeTasks.AppendElement(r.forget());
 147   }
 148
 149   nsresult AddTask(AbstractThread* aThread,
 150                    already_AddRefed<nsIRunnable> aRunnable) override {
 151     nsCOMPtr<nsIRunnable> r = aRunnable;
 152     MOZ_RELEASE_ASSERT(r);
 153     // To preserve the event order, we need to append a new group if the last
 154     // group is not targeted for |aThread|.
 155     // See https://bugzilla.mozilla.org/show_bug.cgi?id=1318226&mark=0-3#c0
 156     // for the details of the issue.
 157     if (mTaskGroups.Length() == 0 ||
 158         mTaskGroups.LastElement()->mThread != aThread) {
 159       mTaskGroups.AppendElement(new PerThreadTaskGroup(aThread));
 160     }
 161
 162     PerThreadTaskGroup& group = *mTaskGroups.LastElement();
 163     group.mRegularTasks.AppendElement(r.forget());
 164
 165     return NS_OK;
 166   }
 167
 168   bool HasTasksFor(AbstractThread* aThread) override {
 169     return !!GetTaskGroup(aThread) ||
 170            (aThread == AbstractThread::GetCurrent() && HaveDirectTasks());
 171   }
 172
 173   nsresult DispatchTasksFor(AbstractThread* aThread) override {
 174     nsresult rv = NS_OK;
 175
 176     // Dispatch all groups that match |aThread|.
 177     for (size_t i = 0; i < mTaskGroups.Length(); ++i) {
 178       if (mTaskGroups[i]->mThread == aThread) {
 179         nsresult rv2 = DispatchTaskGroup(std::move(mTaskGroups[i]));
 180
 181         if (NS_WARN_IF(NS_FAILED(rv2)) && NS_SUCCEEDED(rv)) {
 182           // We should try our best to call DispatchTaskGroup() as much as
 183           // possible and return an error if any of DispatchTaskGroup() calls
 184           // failed.
 185           rv = rv2;
 186         }
 187
 188         mTaskGroups.RemoveElementAt(i--);
 189       }
 190     }
 191
 192     return rv;
 193   }
 194
 195  private:
 196   struct PerThreadTaskGroup {
 197    public:
 198     explicit PerThreadTaskGroup(AbstractThread* aThread) : mThread(aThread) {
 199       MOZ_COUNT_CTOR(PerThreadTaskGroup);
 200     }
 201
 202     MOZ_COUNTED_DTOR(PerThreadTaskGroup)
 203
 204     RefPtr<AbstractThread> mThread;
 205     nsTArray<nsCOMPtr<nsIRunnable>> mStateChangeTasks;
 206     nsTArray<nsCOMPtr<nsIRunnable>> mRegularTasks;
 207   };
 208
 209   class TaskGroupRunnable : public Runnable {
 210    public:
 211     explicit TaskGroupRunnable(UniquePtr<PerThreadTaskGroup>&& aTasks)
 212         : Runnable("AutoTaskDispatcher::TaskGroupRunnable"),
 213           mTasks(std::move(aTasks)) {}
 214
 215     NS_IMETHOD Run() override {
 216       // State change tasks get run all together before any code is run, so
 217       // that all state changes are made in an atomic unit.
 218       for (size_t i = 0; i < mTasks->mStateChangeTasks.Length(); ++i) {
 219         mTasks->mStateChangeTasks[i]->Run();
 220       }
 221
 222       // Once the state changes have completed, drain any direct tasks
 223       // generated by those state changes (i.e. watcher notification tasks).
 224       // This needs to be outside the loop because we don't want to run code
 225       // that might observe intermediate states.
 226       MaybeDrainDirectTasks();
 227
 228       for (size_t i = 0; i < mTasks->mRegularTasks.Length(); ++i) {
 229         AUTO_PROFILE_FOLLOWING_RUNNABLE(mTasks->mRegularTasks[i]);
 230         mTasks->mRegularTasks[i]->Run();
 231
 232         // Scope direct tasks tightly to the task that generated them.
 233         MaybeDrainDirectTasks();
 234       }
 235
 236       return NS_OK;
 237     }
 238
 239    private:
 240     void MaybeDrainDirectTasks() {
 241       AbstractThread* currentThread = AbstractThread::GetCurrent();
 242       if (currentThread && currentThread->MightHaveTailTasks()) {
 243         currentThread->TailDispatcher().DrainDirectTasks();
 244       }
 245     }
 246
 247     UniquePtr<PerThreadTaskGroup> mTasks;
 248   };
 249
 250   PerThreadTaskGroup& EnsureTaskGroup(AbstractThread* aThread) {
 251     PerThreadTaskGroup* existing = GetTaskGroup(aThread);
 252     if (existing) {
 253       return *existing;
 254     }
 255
 256     mTaskGroups.AppendElement(new PerThreadTaskGroup(aThread));
 257     return *mTaskGroups.LastElement();
 258   }
 259
 260   PerThreadTaskGroup* GetTaskGroup(AbstractThread* aThread) {
 261     for (size_t i = 0; i < mTaskGroups.Length(); ++i) {
 262       if (mTaskGroups[i]->mThread == aThread) {
 263         return mTaskGroups[i].get();
 264       }
 265     }
 266
 267     // Not found.
 268     return nullptr;
 269   }
 270
 271   nsresult DispatchTaskGroup(UniquePtr<PerThreadTaskGroup> aGroup) {
 272     RefPtr<AbstractThread> thread = aGroup->mThread;
 273
 274     AbstractThread::DispatchReason reason =
 275         mIsTailDispatcher ? AbstractThread::TailDispatch
 276                           : AbstractThread::NormalDispatch;
 277     nsCOMPtr<nsIRunnable> r = new TaskGroupRunnable(std::move(aGroup));
 278     return thread->Dispatch(r.forget(), reason);
 279   }
 280
 281   // Task groups, organized by thread.
 282   nsTArray<UniquePtr<PerThreadTaskGroup>> mTaskGroups;
 283
 284   nsCOMPtr<nsIDirectTaskDispatcher> mDirectTaskDispatcher;
 285   // True if this TaskDispatcher represents the tail dispatcher for the thread
 286   // upon which it runs.
 287   const bool mIsTailDispatcher;
 288 };
 289
 290 // Little utility class to allow declaring AutoTaskDispatcher as a default
 291 // parameter for methods that take a TaskDispatcher&.
 292 template <typename T>
 293 class PassByRef {
 294  public:
 295   PassByRef() = default;
 296   operator T&() { return mVal; }
 297
 298  private:
 299   T mVal;
 300 };
 301
 302 }  // namespace mozilla
 303
 304 #endif