[Try # 3] Scheduler refactoring to virtually eliminate redundant DoWorks

third_party/WebKit/Source/platform/scheduler/base/task_queue_manager.cc

 // Copyright 2014 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "platform/scheduler/base/task_queue_manager.h"
 
 #include <queue>
 #include <set>
 
 #include "base/bind.h"
@@ skipping 40 matching lines @@
 }
 
 TaskQueueManager::TaskQueueManager(
     scoped_refptr<TaskQueueManagerDelegate> delegate,
     const char* tracing_category,
     const char* disabled_by_default_tracing_category,
     const char* disabled_by_default_verbose_tracing_category)
     : real_time_domain_(new RealTimeDomain(tracing_category)),
       delegate_(delegate),
       task_was_run_on_quiescence_monitored_queue_(false),
-      other_thread_pending_wakeup_(false),
+      do_work_pending_lock_(),  // NOTE this calls the constructor!

[Sami, 2016/12/07 16:13:42]

Is this gonna work? This[1] says spinlocks can only be static globals --
presumably because they don't have initializers or even constructors to
initialize them.

[1]
https://cs.chromium.org/chromium/src/base/synchronization/spin_lock.h?rcl=148...

[alex clarke (OOO till 29th), 2016/12/08 17:38:48]

Acknowledged.

+      do_work_running_count_(0),
+      immediate_do_work_posted_count_(0),
+      is_nested_(false),
+      record_task_delay_histograms_(true),
       work_batch_size_(1),
       task_count_(0),
       tracing_category_(tracing_category),
       disabled_by_default_tracing_category_(
           disabled_by_default_tracing_category),
       disabled_by_default_verbose_tracing_category_(
           disabled_by_default_verbose_tracing_category),
       currently_executing_task_queue_(nullptr),
       observer_(nullptr),
       deletion_sentinel_(new DeletionSentinel()),
       weak_factory_(this) {
   DCHECK(delegate->RunsTasksOnCurrentThread());
   TRACE_EVENT_OBJECT_CREATED_WITH_ID(disabled_by_default_tracing_category,
                                      "TaskQueueManager", this);
   selector_.SetTaskQueueSelectorObserver(this);
 
-  from_main_thread_immediate_do_work_closure_ =
-      base::Bind(&TaskQueueManager::DoWork, weak_factory_.GetWeakPtr(),
-                 base::TimeTicks(), true);
-  from_other_thread_immediate_do_work_closure_ =
-      base::Bind(&TaskQueueManager::DoWork, weak_factory_.GetWeakPtr(),
-                 base::TimeTicks(), false);
+  delayed_do_work_closure_ =
+      base::Bind(&TaskQueueManager::DoWork, weak_factory_.GetWeakPtr(), true);
+  immediate_do_work_closure_ =
+      base::Bind(&TaskQueueManager::DoWork, weak_factory_.GetWeakPtr(), false);
 
   // TODO(alexclarke): Change this to be a parameter that's passed in.
   RegisterTimeDomain(real_time_domain_.get());
 
   delegate_->AddNestingObserver(this);
 }
 
 TaskQueueManager::~TaskQueueManager() {
   TRACE_EVENT_OBJECT_DELETED_WITH_ID(disabled_by_default_tracing_category_,
                                      "TaskQueueManager", this);
@@ skipping 44 matching lines @@
   DCHECK(main_thread_checker_.CalledOnValidThread());
   if (observer_)
     observer_->OnUnregisterTaskQueue(task_queue);
 
   // Add |task_queue| to |queues_to_delete_| so we can prevent it from being
   // freed while any of our structures hold hold a raw pointer to it.
   queues_to_delete_.insert(task_queue);
   queues_.erase(task_queue);
 
   selector_.RemoveQueue(task_queue.get());
+
+  {
+    SpinLock::Guard guard(do_work_pending_lock_);
+    has_incomming_immediate_work_.erase(task_queue.get());
+  }
 }
 
-void TaskQueueManager::UpdateWorkQueues(LazyNow lazy_now) {
-  TRACE_EVENT0(disabled_by_default_tracing_category_,
-               "TaskQueueManager::UpdateWorkQueues");
-
+void TaskQueueManager::UpdateWorkQueues(LazyNow* lazy_now) {
   for (TimeDomain* time_domain : time_domains_) {
-    LazyNow lazy_now_in_domain = time_domain == real_time_domain_.get()
-                                     ? lazy_now
-                                     : time_domain->CreateLazyNow();
-    time_domain->UpdateWorkQueues(lazy_now_in_domain);
+    if (time_domain == real_time_domain_.get()) {
+      time_domain->WakeupReadyDelayedQueues(lazy_now);
+      continue;
+    }
+    LazyNow time_domain_lazy_now = time_domain->CreateLazyNow();
+    time_domain->WakeupReadyDelayedQueues(&time_domain_lazy_now);
   }
 }
 
 void TaskQueueManager::OnBeginNestedMessageLoop() {
   // We just entered a nested message loop, make sure there's a DoWork posted or
   // the system will grind to a halt.
-  delegate_->PostTask(FROM_HERE, from_main_thread_immediate_do_work_closure_);
+  SpinLock::Guard guard(do_work_pending_lock_);

[Sami, 2016/12/07 16:13:42]

Seems risky holding to this during the PostTask -- could we move this below it?

[alex clarke (OOO till 29th), 2016/12/08 17:38:48]

Done.

+  delegate_->PostTask(FROM_HERE, immediate_do_work_closure_);
+  immediate_do_work_posted_count_++;
+  is_nested_ = true;
+}
+
+void TaskQueueManager::OnQueueHasImmediateWork(internal::TaskQueueImpl* queue,
+                                               bool ensure_do_work_posted) {
+  SpinLock::Guard guard(do_work_pending_lock_);
+  has_incomming_immediate_work_.insert(queue);
+  if (ensure_do_work_posted)
+    MaybeScheduleImmediateWorkLocked(FROM_HERE);
+}
+
+void TaskQueueManager::NotifyQueuesOfIncomingImmediateWorkOnMainThreadLocked() {
+  for (internal::TaskQueueImpl* queue : has_incomming_immediate_work_) {
+    queue->ReloadImmediateWorkQueueIfEmpty();
+  }
+  has_incomming_immediate_work_.clear();
 }
 
 void TaskQueueManager::MaybeScheduleImmediateWork(
     const tracked_objects::Location& from_here) {
-  bool on_main_thread = delegate_->BelongsToCurrentThread();
-  // De-duplicate DoWork posts.
-  if (on_main_thread) {
-    if (!main_thread_pending_wakeups_.insert(base::TimeTicks()).second) {
-      return;
-    }
-    delegate_->PostTask(from_here, from_main_thread_immediate_do_work_closure_);
-  } else {
-    {
-      base::AutoLock lock(other_thread_lock_);
-      if (other_thread_pending_wakeup_)
-        return;
-      other_thread_pending_wakeup_ = true;
-    }
-    delegate_->PostTask(from_here,
-                        from_other_thread_immediate_do_work_closure_);
+  SpinLock::Guard guard(do_work_pending_lock_);
+  MaybeScheduleImmediateWorkLocked(from_here);
+}
+
+void TaskQueueManager::MaybeScheduleImmediateWorkLocked(
+    const tracked_objects::Location& from_here) {
+  // Unlwss we're ensted, try to avoid posting redundant DoWorks.

[Sami, 2016/12/07 16:13:42]

typo: Unless we're nested

[alex clarke (OOO till 29th), 2016/12/08 17:38:48]

Done.

+  if (!is_nested_ &&
+      (do_work_running_count_ == 1 || immediate_do_work_posted_count_ > 0)) {
+    return;
   }
+
+  delegate_->PostTask(from_here, immediate_do_work_closure_);

[Sami, 2016/12/07 16:13:42]

Ditto about the spinlock and the PostTask.

[alex clarke (OOO till 29th), 2016/12/08 17:40:28]

Done.

+  immediate_do_work_posted_count_++;
 }
 
 void TaskQueueManager::MaybeScheduleDelayedWork(
     const tracked_objects::Location& from_here,
     base::TimeTicks now,
     base::TimeDelta delay) {
   DCHECK(main_thread_checker_.CalledOnValidThread());
+  {
+    SpinLock::Guard guard(do_work_pending_lock_);
+
+    // If there's a pending immediate DoWork then we rely on the logic in DoWork
+    // to post a continuation as needed.
+    if (immediate_do_work_posted_count_ > 0)
+      return;
+
+    // If a non-nested DoWork is running we can also rely on the logic in DoWork
+    // to post a continuation as needed.
+    if (do_work_running_count_ == 1 && !is_nested_)
+      return;
+  }
+  MaybeScheduleDelayedWorkInternal(from_here, now, delay);
+}
+
+void TaskQueueManager::MaybeScheduleDelayedWorkInternal(
+    const tracked_objects::Location& from_here,
+    base::TimeTicks now,
+    base::TimeDelta delay) {
   DCHECK_GE(delay, base::TimeDelta());
-
-  // If there's a pending immediate DoWork then we rely on
-  // TryAdvanceTimeDomains getting the TimeDomain to call
-  // MaybeScheduleDelayedWork again when the immediate DoWork is complete.
-  if (main_thread_pending_wakeups_.find(base::TimeTicks()) !=
-      main_thread_pending_wakeups_.end()) {
-    return;
-  }
   // De-duplicate DoWork posts.
   base::TimeTicks run_time = now + delay;
-  if (!main_thread_pending_wakeups_.empty() &&
-      *main_thread_pending_wakeups_.begin() <= run_time) {
+  if (next_delayed_do_work_ <= run_time && !next_delayed_do_work_.is_null())
     return;
-  }
-  main_thread_pending_wakeups_.insert(run_time);
+
+  cancelable_delayed_do_work_closure_.Reset(delayed_do_work_closure_);
+  next_delayed_do_work_ = run_time;
   delegate_->PostDelayedTask(
-      from_here, base::Bind(&TaskQueueManager::DoWork,
-                            weak_factory_.GetWeakPtr(), run_time, true),
-      delay);
+      from_here, cancelable_delayed_do_work_closure_.callback(), delay);
 }
 
-void TaskQueueManager::DoWork(base::TimeTicks run_time, bool from_main_thread) {
+void TaskQueueManager::DoWork(bool delayed) {
   DCHECK(main_thread_checker_.CalledOnValidThread());
-  TRACE_EVENT1(tracing_category_, "TaskQueueManager::DoWork",
-               "from_main_thread", from_main_thread);
+  TRACE_EVENT1("tracing_category_", "TaskQueueManager::DoWork", "delayed",
+               delayed);
+  LazyNow lazy_now(real_time_domain()->CreateLazyNow());
 
-  if (from_main_thread) {
-    main_thread_pending_wakeups_.erase(run_time);
-  } else {
-    base::AutoLock lock(other_thread_lock_);
-    other_thread_pending_wakeup_ = false;
-  }
-
-  // Posting a DoWork while a DoWork is running leads to spurious DoWorks.
-  main_thread_pending_wakeups_.insert(base::TimeTicks());
-
-  if (!delegate_->IsNested())
+  bool is_nested = delegate_->IsNested();
+  if (!is_nested)
     queues_to_delete_.clear();
 
-  LazyNow lazy_now(real_time_domain()->CreateLazyNow());
-  UpdateWorkQueues(lazy_now);
+  for (int i = 0; i < work_batch_size_; i++) {
+    {
+      SpinLock::Guard guard(do_work_pending_lock_);
 
-  for (int i = 0; i < work_batch_size_; i++) {
-    internal::WorkQueue* work_queue;
+      is_nested_ = is_nested;
+      DCHECK_EQ(is_nested_, delegate_->IsNested());
+
+      if (i == 0) {
+        do_work_running_count_++;
+
+        if (!delayed) {
+          immediate_do_work_posted_count_--;
+          DCHECK_GE(immediate_do_work_posted_count_, 0);
+        }
+      }
+
+      NotifyQueuesOfIncomingImmediateWorkOnMainThreadLocked();
+    }
+
+    UpdateWorkQueues(&lazy_now);
+
+    internal::WorkQueue* work_queue = nullptr;
     if (!SelectWorkQueueToService(&work_queue))
       break;
 
     switch (ProcessTaskFromWorkQueue(work_queue, &lazy_now)) {
       case ProcessTaskResult::DEFERRED:
         // If a task was deferred, try again with another task.
         continue;
       case ProcessTaskResult::EXECUTED:
         break;
       case ProcessTaskResult::TASK_QUEUE_MANAGER_DELETED:
         return;  // The TaskQueueManager got deleted, we must bail out.
     }
 
     work_queue = nullptr;  // The queue may have been unregistered.
 
-    UpdateWorkQueues(lazy_now);
-
     // Only run a single task per batch in nested run loops so that we can
     // properly exit the nested loop when someone calls RunLoop::Quit().
-    if (delegate_->IsNested())
+    if (is_nested)
       break;
   }
 
-  main_thread_pending_wakeups_.erase(base::TimeTicks());
-
   // TODO(alexclarke): Consider refactoring the above loop to terminate only
   // when there's no more work left to be done, rather than posting a
   // continuation task.
-  if (!selector_.EnabledWorkQueuesEmpty() || TryAdvanceTimeDomains())
-    MaybeScheduleImmediateWork(FROM_HERE);
+  if (delayed)
+    next_delayed_do_work_ = base::TimeTicks();
+
+  // If we know we're about to post an immediate do work, then there's no point
+  // calling UpdateWorkQueues since that'll get done by the next DoWork.
+  bool work_queues_might_be_empty = selector_.EnabledWorkQueuesEmpty();
+  if (work_queues_might_be_empty)
+    UpdateWorkQueues(&lazy_now);
+
+  SpinLock::Guard guard(do_work_pending_lock_);

[Sami, 2016/12/07 16:13:42]

Could you also delimit this scope explicitly? Feels like it's too easy to add
stuff at the end not realizing the spinlock is being held. Also not sure if
everything here needs to be in that scope...

[alex clarke (OOO till 29th), 2016/12/08 17:38:48]

Done.

+
+  // Likewise it's only worth calling
+  // NotifyQueuesOfIncomingImmediateWorkOnMainThreadLocked if it looks like we
+  // might have run out of immediate work to do.
+  if (work_queues_might_be_empty)
+    NotifyQueuesOfIncomingImmediateWorkOnMainThreadLocked();
+
+  base::Optional<base::TimeDelta> next_delay = DelayTillNextTask(&lazy_now);
+  DCHECK(work_queues_might_be_empty ||
+         (next_delay && next_delay.value().is_zero()));
+
+  do_work_running_count_--;
+  DCHECK_GE(do_work_running_count_, 0);
+
+  is_nested_ = is_nested;
+  DCHECK_EQ(is_nested_, delegate_->IsNested());
+
+  PostDoWorkContinuation(next_delay, &lazy_now);
 }
 
-bool TaskQueueManager::TryAdvanceTimeDomains() {
-  bool can_advance = false;
+void TaskQueueManager::PostDoWorkContinuation(

[Sami, 2016/12/07 16:13:42]

...Locked()?

[alex clarke (OOO till 29th), 2016/12/08 17:38:48]

Done.

+    base::Optional<base::TimeDelta> next_delay,
+    LazyNow* lazy_now) {
+  // If there are no tasks left then we don't need to post a continuation.
+  if (!next_delay)
+    return;
+
+  // If either an immediate DoWork or a delayed DoWork is pending then we don't
+  // need to post a continuation.
+  if (immediate_do_work_posted_count_ > 0 ||
+      (!next_delayed_do_work_.is_null() &&
+       next_delayed_do_work_ <= lazy_now->Now())) {
+    return;
+  }
+
+  // Post a continuation task based on the delay till the next task.
+  if (next_delay.value().is_zero()) {
+    delegate_->PostTask(FROM_HERE, immediate_do_work_closure_);

[Sami, 2016/12/07 16:13:42]

Ditto about calling into the delegate/selector/time domains while holding the
spinlock. Maybe I'm worrying too much about this? Feels like we should be a
little careful still.

[alex clarke (OOO till 29th), 2016/12/08 17:38:48]

Done.

+    immediate_do_work_posted_count_++;
+  } else {
+    MaybeScheduleDelayedWorkInternal(FROM_HERE, lazy_now->Now(),
+                                     next_delay.value());
+  }
+}
+
+base::Optional<base::TimeDelta> TaskQueueManager::DelayTillNextTask(
+    LazyNow* lazy_now) {
+  // If the selector has non-empty queues we trivially know there is immediate
+  // word to be done.
+  if (!selector_.EnabledWorkQueuesEmpty())
+    return base::TimeDelta();
+
+  // Otherwise we need to find the shortest delay, if any.
+  base::Optional<base::TimeDelta> next_continuation;
   for (TimeDomain* time_domain : time_domains_) {
-    can_advance |= time_domain->MaybeAdvanceTime();
+    base::Optional<base::TimeDelta> continuation =
+        time_domain->DelayTillNextTask(lazy_now);
+    if (!continuation)
+      continue;
+    if (!next_continuation || next_continuation.value() < continuation.value())

[Sami, 2016/12/07 16:13:42]

Should '<' be '>' or am I missing something?

[alex clarke (OOO till 29th), 2016/12/08 17:38:48]

Good catch, I added a test.

+      next_continuation = continuation;
   }
-  return can_advance;
+  return next_continuation;
 }
 
 bool TaskQueueManager::SelectWorkQueueToService(

[Sami, 2016/12/07 16:13:42]

...Locked()?

[alex clarke (OOO till 29th), 2016/12/08 17:38:48]

This doesn't need to be locked?

     internal::WorkQueue** out_work_queue) {
   bool should_run = selector_.SelectWorkQueueToService(out_work_queue);
   TRACE_EVENT_OBJECT_SNAPSHOT_WITH_ID(
       disabled_by_default_tracing_category_, "TaskQueueManager", this,
       AsValueWithSelectorResult(should_run, *out_work_queue));
   return should_run;
 }
 
 void TaskQueueManager::DidQueueTask(
     const internal::TaskQueueImpl::Task& pending_task) {
   task_annotator_.DidQueueTask("TaskQueueManager::PostTask", pending_task);
 }
 
 TaskQueueManager::ProcessTaskResult TaskQueueManager::ProcessTaskFromWorkQueue(
     internal::WorkQueue* work_queue,
     LazyNow* lazy_now) {
   DCHECK(main_thread_checker_.CalledOnValidThread());
   scoped_refptr<DeletionSentinel> protect(deletion_sentinel_);
   internal::TaskQueueImpl::Task pending_task =
       work_queue->TakeTaskFromWorkQueue();
 
   // It's possible the task was canceled, if so bail out.
   if (pending_task.task.IsCancelled())
     return ProcessTaskResult::EXECUTED;
 
   internal::TaskQueueImpl* queue = work_queue->task_queue();
   if (queue->GetQuiescenceMonitored())
     task_was_run_on_quiescence_monitored_queue_ = true;
 
-  if (!pending_task.nestable && delegate_->IsNested()) {
+  DCHECK_EQ(is_nested_, delegate_->IsNested());
+  if (!pending_task.nestable && is_nested_) {
     // Defer non-nestable work to the main task runner.  NOTE these tasks can be
     // arbitrarily delayed so the additional delay should not be a problem.
     // TODO(skyostil): Figure out a way to not forget which task queue the
     // task is associated with. See http://crbug.com/522843.
     // TODO(tzik): Remove base::UnsafeConvertOnceClosureToRepeating once
     // TaskRunners have migrated to OnceClosure.
     delegate_->PostNonNestableTask(
         pending_task.posted_from,
         UnsafeConvertOnceClosureToRepeating(std::move(pending_task.task)));
     return ProcessTaskResult::DEFERRED;
   }
 
-  MaybeRecordTaskDelayHistograms(pending_task, queue);
+  if (record_task_delay_histograms_)
+    MaybeRecordTaskDelayHistograms(pending_task, queue);
 
   double task_start_time = 0;
   TRACE_TASK_EXECUTION("TaskQueueManager::ProcessTaskFromWorkQueue",
                        pending_task);
   if (queue->GetShouldNotifyObservers()) {
     for (auto& observer : task_observers_)
       observer.WillProcessTask(pending_task);
     queue->NotifyWillProcessTask(pending_task);
 
     bool notify_time_observers =
@@ skipping 130 matching lines @@
   if (should_run) {
     state->SetString("selected_queue",
                      selected_work_queue->task_queue()->GetName());
     state->SetString("work_queue_name", selected_work_queue->name());
   }
 
   state->BeginArray("time_domains");
   for (auto* time_domain : time_domains_)
     time_domain->AsValueInto(state.get());
   state->EndArray();
+
+  state->SetBoolean("is_nested", is_nested_);

[Sami, 2016/12/07 16:13:42]

Looks like this should be inside the lock.

[alex clarke (OOO till 29th), 2016/12/08 17:38:48]

Done.

+
+  SpinLock::Guard guard(do_work_pending_lock_);

[Sami, 2016/12/07 16:13:42]

nit: explicit scope please.

[alex clarke (OOO till 29th), 2016/12/08 17:38:48]

Done.

+  state->SetInteger("do_work_count", do_work_running_count_);
+  state->SetInteger("immediate_do_work_posted",
+                    immediate_do_work_posted_count_);
   return std::move(state);
 }
 
 void TaskQueueManager::OnTaskQueueEnabled(internal::TaskQueueImpl* queue) {
   DCHECK(main_thread_checker_.CalledOnValidThread());
   // Only schedule DoWork if there's something to do.
   if (queue->HasPendingImmediateWork())
     MaybeScheduleImmediateWork(FROM_HERE);
 }
 
 void TaskQueueManager::OnTriedToSelectBlockedWorkQueue(
     internal::WorkQueue* work_queue) {
   DCHECK(main_thread_checker_.CalledOnValidThread());
   DCHECK(!work_queue->Empty());
   if (observer_) {
     observer_->OnTriedToExecuteBlockedTask(*work_queue->task_queue(),
                                            *work_queue->GetFrontTask());
   }
 }
 
 bool TaskQueueManager::HasImmediateWorkForTesting() const {
   return !selector_.EnabledWorkQueuesEmpty();
 }
 
+void TaskQueueManager::SetRecordTaskDelayHistograms(
+    bool record_task_delay_histograms) {
+  DCHECK(main_thread_checker_.CalledOnValidThread());
+  record_task_delay_histograms_ = record_task_delay_histograms;
+}
+
 }  // namespace scheduler
 }  // namespace blink