[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!
+      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_);
+  delegate_->PostTask(FROM_HERE, immediate_do_work_closure_);
+  immediate_do_work_posted_count_++;
+  is_nested_ = true;
+}
+
+void TaskQueueManager::OnQueueHasImmediateWork(internal::TaskQueueImpl* queue) {
+  SpinLock::Guard guard(do_work_pending_lock_);
+  has_incomming_immediate_work_.insert(queue);
+
+  // There's no point posting a DoWork for a blocked or disabled queue, although
+  // we can only determine that on the main thread.
+  if (queue->RunsTasksOnCurrentThread() &&
+      (!queue->IsQueueEnabled() || queue->BlockedByFenceLocked())) {

[Sami, 2016/12/05 17:52:54]

Hmm, we're not holding the any thread lock here, are we?

[alex clarke (OOO till 29th), 2016/12/06 17:37:54]

Actually we are but that's perhaps a little fragile WRT subsequent patches. 
Call stack is

TaskQueueImpl::BlockedByFenceLocked 
TaskQueueManager::OnQueueHasImmediateWork
TaskQueueImpl::PushOntoImmediateIncomingQueueLocked
...

Note TaskQueueImpl::BlockedByFenceLocked has any_thread_lock_.AssertAcquired();

Perhaps the cleanest thing to do is to pull the logic checking if we should call
MaybeScheduleImmediateWorkLocked back out of the function?

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

Yeah, something like that would be better. The coupling between TaskQueueImpl
and TaskQueueManager is already pretty tight and I wouldn't want them to start
making assumptions about each other's critical sections.

+    return;
+  }
+
+  MaybeScheduleImmediateWorkLocked(FROM_HERE);
+}
+
+void TaskQueueManager::NotifyQueuesOfIncomingImmediateWorkOnMainThreadLocked() {
+  for (internal::TaskQueueImpl* queue : has_incomming_immediate_work_) {
+    queue->OnIncomingImmediateTaskAvailable();
+  }
+  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) {

[Sami, 2016/12/05 17:52:54]

Can we check that a spinlock is held? Not sure...

We could add a debug-only flag if necessary.

[alex clarke (OOO till 29th), 2016/12/06 17:37:54]

Currently no it's not possible to do so in a good way.

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

I was thinking something like a checked spinlock that also uses a regular lock
in debug builds. Can be left for later too...

+  // Unlwss we're ensted, try to avoid posting redundant DoWorks.
+  if (!is_nested_ &&
+      (do_work_running_count_ == 1 || immediate_do_work_posted_count_ > 0)) {
+    return;
   }
+
+  delegate_->PostTask(from_here, immediate_do_work_closure_);
+  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) {

[Sami, 2016/12/05 17:52:54]

meta-comment: This function is getting long -- maybe we should split it into
more helpers?

[alex clarke (OOO till 29th), 2016/12/06 17:37:54]

Done.

   DCHECK(main_thread_checker_.CalledOnValidThread());
-  TRACE_EVENT1(tracing_category_, "TaskQueueManager::DoWork",
-               "from_main_thread", from_main_thread);
-
-  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())
-    queues_to_delete_.clear();
-
+  TRACE_EVENT1("tracing_category_", "TaskQueueManager::DoWork", "delayed",
+               delayed ? "false" : "true");

[Sami, 2016/12/05 17:52:54]

nit: you can just pass |delayed| here and tracing will know to turn it into a
boolean.

[alex clarke (OOO till 29th), 2016/12/06 17:37:54]

Done.

   LazyNow lazy_now(real_time_domain()->CreateLazyNow());
   base::TimeTicks task_start_time;
 
-  if (!delegate_->IsNested() && task_time_observers_.might_have_observers())
-    task_start_time = lazy_now.Now();
+  bool is_nested = delegate_->IsNested();
+  if (!delegate_->IsNested()) {
+    if (task_time_observers_.might_have_observers())
+      task_start_time = lazy_now.Now();
 
-  UpdateWorkQueues(lazy_now);
+    queues_to_delete_.clear();
+  }
 
   for (int i = 0; i < work_batch_size_; i++) {
-    internal::WorkQueue* work_queue;
+    {
+      SpinLock::Guard guard(do_work_pending_lock_);
+
+      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;
 
     // TaskQueueManager guarantees that task queue will not be deleted
     // when we are in DoWork (but WorkQueue may be deleted).
     internal::TaskQueueImpl* task_queue = work_queue->task_queue();
 
     switch (ProcessTaskFromWorkQueue(work_queue)) {
       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.
     }
 
     lazy_now = real_time_domain()->CreateLazyNow();
-    if (!delegate_->IsNested() && task_start_time != base::TimeTicks()) {
+    if (!is_nested && task_start_time != base::TimeTicks()) {
       // Only report top level task durations.
       base::TimeTicks task_end_time = lazy_now.Now();
       for (auto& observer : task_time_observers_) {
         observer.ReportTaskTime(task_queue,
                                 MonotonicTimeInSeconds(task_start_time),
                                 MonotonicTimeInSeconds(task_end_time));
       }
       task_start_time = task_end_time;
     }
 
     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();
+
+  LazyNow continuation_lazy_now(real_time_domain()->CreateLazyNow());
+  UpdateWorkQueues(&continuation_lazy_now);
+
+  SpinLock::Guard guard(do_work_pending_lock_);
+  NotifyQueuesOfIncomingImmediateWorkOnMainThreadLocked();
+  base::Optional<base::TimeDelta> next_delay =
+      DelayTillNextTask(&continuation_lazy_now);
+
+  do_work_running_count_--;
+  DCHECK_GE(do_work_running_count_, 0);
+
+  is_nested_ = is_nested;
+  DCHECK_EQ(is_nested_, delegate_->IsNested());
+
+  // 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_ < continuation_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_);
+    immediate_do_work_posted_count_++;
+  } else {
+    MaybeScheduleDelayedWorkInternal(FROM_HERE, lazy_now.Now(),
+                                     next_delay.value());
+  }
 }
 
-bool TaskQueueManager::TryAdvanceTimeDomains() {
-  bool can_advance = false;
+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())
+      next_continuation = continuation;
   }
-  return can_advance;
+  return next_continuation;
 }
 
 bool TaskQueueManager::SelectWorkQueueToService(
     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;
 }
@@ skipping 11 matching lines @@
       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);
 
   TRACE_TASK_EXECUTION("TaskQueueManager::ProcessTaskFromWorkQueue",
                        pending_task);
   if (queue->GetShouldNotifyObservers()) {
     for (auto& observer : task_observers_)
       observer.WillProcessTask(pending_task);
     queue->NotifyWillProcessTask(pending_task);
   }
   TRACE_EVENT1(tracing_category_, "TaskQueueManager::RunTask", "queue",
                queue->GetName());
@@ skipping 112 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_);
+
+  SpinLock::Guard guard(do_work_pending_lock_);
+  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