DXR is a code search and navigation tool aimed at making sense of large projects. It supports full-text and regex searches as well as structural queries.

Header

Mercurial (5350524bb654)

VCS Links

Line Code
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446
/* -*- 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 "ThrottledEventQueue.h"

#include "mozilla/Atomics.h"
#include "mozilla/ClearOnShutdown.h"
#include "mozilla/Mutex.h"
#include "mozilla/Unused.h"
#include "nsEventQueue.h"

namespace mozilla {

using mozilla::services::GetObserverService;

namespace {

static const char kShutdownTopic[] = "xpcom-shutdown";

} // anonymous namespace

// The ThrottledEventQueue is designed with inner and outer objects:
//
//       XPCOM code    nsObserverService
//            |               |
//            |               |
//            v               |
//        +-------+           |
//        | Outer |           |
//        +-------+           |
//            |               |
//            |   +-------+   |
//            +-->| Inner |<--+
//                +-------+
//
// Client code references the outer nsIEventTarget which in turn references
// an inner object.  The inner object is also held alive by the observer
// service.
//
// If the outer object is dereferenced and destroyed, it will trigger a
// shutdown operation on the inner object.  Similarly if the observer
// service notifies that the browser is shutting down, then the inner
// object also starts shutting down.
//
// Once the queue has drained we unregister from the observer service.  If
// the outer object is already gone, then the inner object is free'd at this
// point.  If the outer object still exists then calls fall back to the
// ThrottledEventQueue's base target.  We just don't queue things
// any more.  The inner is then released once the outer object is released.
//
// Note, we must keep the inner object alive and attached to the observer
// service until the TaskQueue is fully shutdown and idle.  We must delay
// xpcom shutdown if the TaskQueue is in the middle of draining.
class ThrottledEventQueue::Inner final : public nsIObserver
{
  // The runnable which is dispatched to the underlying base target.  Since
  // we only execute one event at a time we just re-use a single instance
  // of this class while there are events left in the queue.
  class Executor final : public Runnable
  {
    RefPtr<Inner> mInner;

  public:
    explicit Executor(Inner* aInner)
      : mInner(aInner)
    { }

    NS_IMETHODIMP
    Run()
    {
      mInner->ExecuteRunnable();
      return NS_OK;
    }
  };

  mutable Mutex mMutex;
  mutable CondVar mIdleCondVar;

  mozilla::CondVar mEventsAvailable;

  // any thread, protected by mutex
  nsEventQueue mEventQueue;

  // written on main thread, read on any thread
  nsCOMPtr<nsIEventTarget> mBaseTarget;

  // any thread, protected by mutex
  nsCOMPtr<nsIRunnable> mExecutor;

  // any thread, atomic
  Atomic<uint32_t> mExecutionDepth;

  // any thread, protected by mutex
  bool mShutdownStarted;

  explicit Inner(nsIEventTarget* aBaseTarget)
    : mMutex("ThrottledEventQueue")
    , mIdleCondVar(mMutex, "ThrottledEventQueue:Idle")
    , mEventsAvailable(mMutex, "[ThrottledEventQueue::Inner.mEventsAvailable]")
    , mEventQueue(mEventsAvailable, nsEventQueue::eNormalQueue)
    , mBaseTarget(aBaseTarget)
    , mExecutionDepth(0)
    , mShutdownStarted(false)
  {
  }

  ~Inner()
  {
    MOZ_ASSERT(!mExecutor);
    MOZ_ASSERT(mShutdownStarted);
  }

  void
  ExecuteRunnable()
  {
    // Any thread
    nsCOMPtr<nsIRunnable> event;
    bool shouldShutdown = false;

#ifdef DEBUG
    bool currentThread = false;
    mBaseTarget->IsOnCurrentThread(&currentThread);
    MOZ_ASSERT(currentThread);
#endif

    {
      MutexAutoLock lock(mMutex);

      // We only dispatch an executor runnable when we know there is something
      // in the queue, so this should never fail.
      MOZ_ALWAYS_TRUE(mEventQueue.GetPendingEvent(getter_AddRefs(event), lock));

      // If there are more events in the queue, then dispatch the next
      // executor.  We do this now, before running the event, because
      // the event might spin the event loop and we don't want to stall
      // the queue.
      if (mEventQueue.HasPendingEvent(lock)) {
        // Dispatch the next base target runnable to attempt to execute
        // the next throttled event.  We must do this before executing
        // the event in case the event spins the event loop.
        MOZ_ALWAYS_SUCCEEDS(
          mBaseTarget->Dispatch(mExecutor, NS_DISPATCH_NORMAL));
      }

      // Otherwise the queue is empty and we can stop dispatching the
      // executor.  We might also need to shutdown after running the
      // last event.
      else {
        shouldShutdown = mShutdownStarted;
        // Note, this breaks a ref cycle.
        mExecutor = nullptr;
        mIdleCondVar.NotifyAll();
      }
    }

    // Execute the event now that we have unlocked.
    ++mExecutionDepth;
    Unused << event->Run();
    --mExecutionDepth;

    // If shutdown was started and the queue is now empty we can now
    // finalize the shutdown.  This is performed separately at the end
    // of the method in order to wait for the event to finish running.
    if (shouldShutdown) {
      MOZ_ASSERT(IsEmpty());
      NS_DispatchToMainThread(NewRunnableMethod(this, &Inner::ShutdownComplete));
    }
  }

  void
  ShutdownComplete()
  {
    MOZ_ASSERT(NS_IsMainThread());
    MOZ_ASSERT(IsEmpty());
    nsCOMPtr<nsIObserverService> obs = GetObserverService();
    obs->RemoveObserver(this, kShutdownTopic);
  }

public:
  static already_AddRefed<Inner>
  Create(nsIEventTarget* aBaseTarget)
  {
    MOZ_ASSERT(NS_IsMainThread());

    if (ClearOnShutdown_Internal::sCurrentShutdownPhase != ShutdownPhase::NotInShutdown) {
      return nullptr;
    }

    nsCOMPtr<nsIObserverService> obs = GetObserverService();
    if (NS_WARN_IF(!obs)) {
      return nullptr;
    }

    RefPtr<Inner> ref = new Inner(aBaseTarget);

    nsresult rv = obs->AddObserver(ref, kShutdownTopic,
                                   false /* means OS will hold a strong ref */);
    if (NS_WARN_IF(NS_FAILED(rv))) {
      ref->MaybeStartShutdown();
      MOZ_ASSERT(ref->IsEmpty());
      return nullptr;
    }

    return ref.forget();
  }

  NS_IMETHOD
  Observe(nsISupports*, const char* aTopic, const char16_t*) override
  {
    MOZ_ASSERT(NS_IsMainThread());
    MOZ_ASSERT(!strcmp(aTopic, kShutdownTopic));

    MaybeStartShutdown();

    // Once shutdown begins we set the Atomic<bool> mShutdownStarted flag.
    // This prevents any new runnables from being dispatched into the
    // TaskQueue.  Therefore this loop should be finite.
    while (!IsEmpty()) {
      MOZ_ALWAYS_TRUE(NS_ProcessNextEvent());
    }

    return NS_OK;
  }

  void
  MaybeStartShutdown()
  {
    // Any thread
    MutexAutoLock lock(mMutex);

    if (mShutdownStarted) {
      return;
    }
    mShutdownStarted = true;

    // We are marked for shutdown now, but we are still processing runnables.
    // Return for now.  The shutdown will be completed once the queue is
    // drained.
    if (mExecutor) {
      return;
    }

    // The queue is empty, so we can complete immediately.
    NS_DispatchToMainThread(NewRunnableMethod(this, &Inner::ShutdownComplete));
  }

  bool
  IsEmpty() const
  {
    // Any thread
    return Length() == 0;
  }

  uint32_t
  Length() const
  {
    // Any thread
    MutexAutoLock lock(mMutex);
    return mEventQueue.Count(lock);
  }

  void
  AwaitIdle() const
  {
    // Any thread, except the main thread or our base target.  Blocking the
    // main thread is forbidden.  Blocking the base target is guaranteed to
    // produce a deadlock.
    MOZ_ASSERT(!NS_IsMainThread());
#ifdef DEBUG
    bool onBaseTarget = false;
    Unused << mBaseTarget->IsOnCurrentThread(&onBaseTarget);
    MOZ_ASSERT(!onBaseTarget);
#endif

    MutexAutoLock lock(mMutex);
    while (mExecutor) {
      mIdleCondVar.Wait();
    }
  }

  nsresult
  DispatchFromScript(nsIRunnable* aEvent, uint32_t aFlags)
  {
    // Any thread
    nsCOMPtr<nsIRunnable> r = aEvent;
    return Dispatch(r.forget(), aFlags);
  }

  nsresult
  Dispatch(already_AddRefed<nsIRunnable> aEvent, uint32_t aFlags)
  {
    MOZ_ASSERT(aFlags == NS_DISPATCH_NORMAL ||
               aFlags == NS_DISPATCH_AT_END);

    // Any thread
    MutexAutoLock lock(mMutex);

    // If we are shutting down, just fall back to our base target
    // directly.
    if (mShutdownStarted) {
      return mBaseTarget->Dispatch(Move(aEvent), aFlags);
    }

    // We are not currently processing events, so we must start
    // operating on our base target.  This is fallible, so do
    // it first.  Our lock will prevent the executor from accessing
    // the event queue before we add the event below.
    if (!mExecutor) {
      // Note, this creates a ref cycle keeping the inner alive
      // until the queue is drained.
      mExecutor = new Executor(this);
      nsresult rv = mBaseTarget->Dispatch(mExecutor, NS_DISPATCH_NORMAL);
      if (NS_WARN_IF(NS_FAILED(rv))) {
        mExecutor = nullptr;
        return rv;
      }
    }

    // Only add the event to the underlying queue if are able to
    // dispatch to our base target.
    mEventQueue.PutEvent(Move(aEvent), lock);
    return NS_OK;
  }

  nsresult
  DelayedDispatch(already_AddRefed<nsIRunnable> aEvent, uint32_t aDelay)
  {
    // The base target may implement this, but we don't.  Always fail
    // to provide consistent behavior.
    return NS_ERROR_NOT_IMPLEMENTED;
  }

  nsresult
  IsOnCurrentThread(bool* aResult)
  {
    // Any thread

    bool shutdownAndIdle = false;
    {
      MutexAutoLock lock(mMutex);
      shutdownAndIdle = mShutdownStarted && mEventQueue.Count(lock) == 0;
    }

    bool onBaseTarget = false;
    nsresult rv = mBaseTarget->IsOnCurrentThread(&onBaseTarget);
    if (NS_FAILED(rv)) {
      return rv;
    }

    // We consider the current stack on this event target if are on
    // the base target and one of the following is true
    //  1) We are currently running an event OR
    //  2) We are both shutting down and the queue is idle
    *aResult = onBaseTarget && (mExecutionDepth || shutdownAndIdle);

    return NS_OK;
  }

  NS_DECL_THREADSAFE_ISUPPORTS
};

NS_IMPL_ISUPPORTS(ThrottledEventQueue::Inner, nsIObserver);

NS_IMPL_ISUPPORTS(ThrottledEventQueue, nsIEventTarget);

ThrottledEventQueue::ThrottledEventQueue(already_AddRefed<Inner> aInner)
  : mInner(aInner)
{
  MOZ_ASSERT(mInner);
}

ThrottledEventQueue::~ThrottledEventQueue()
{
  mInner->MaybeStartShutdown();
}

void
ThrottledEventQueue::MaybeStartShutdown()
{
  return mInner->MaybeStartShutdown();
}

already_AddRefed<ThrottledEventQueue>
ThrottledEventQueue::Create(nsIEventTarget* aBaseTarget)
{
  MOZ_ASSERT(NS_IsMainThread());
  MOZ_ASSERT(aBaseTarget);

  RefPtr<Inner> inner = Inner::Create(aBaseTarget);
  if (NS_WARN_IF(!inner)) {
    return nullptr;
  }

  RefPtr<ThrottledEventQueue> ref =
    new ThrottledEventQueue(inner.forget());
  return ref.forget();
}

bool
ThrottledEventQueue::IsEmpty() const
{
  return mInner->IsEmpty();
}

uint32_t
ThrottledEventQueue::Length() const
{
  return mInner->Length();
}

void
ThrottledEventQueue::AwaitIdle() const
{
  return mInner->AwaitIdle();
}

NS_IMETHODIMP
ThrottledEventQueue::DispatchFromScript(nsIRunnable* aEvent, uint32_t aFlags)
{
  return mInner->DispatchFromScript(aEvent, aFlags);
}

NS_IMETHODIMP
ThrottledEventQueue::Dispatch(already_AddRefed<nsIRunnable> aEvent,
                                     uint32_t aFlags)
{
  return mInner->Dispatch(Move(aEvent), aFlags);
}

NS_IMETHODIMP
ThrottledEventQueue::DelayedDispatch(already_AddRefed<nsIRunnable> aEvent,
                                            uint32_t aFlags)
{
  return mInner->DelayedDispatch(Move(aEvent), aFlags);
}

NS_IMETHODIMP
ThrottledEventQueue::IsOnCurrentThread(bool* aResult)
{
  return mInner->IsOnCurrentThread(aResult);
}

} // namespace mozilla