Move has_incoming_immediate_work to the TaskQueueManager

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 129 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());
+
+  {
+    base::AutoLock lock(any_thread_lock_);
+    any_thread().has_incoming_immediate_work.erase(task_queue.get());
+  }
 }
 
-void TaskQueueManager::UpdateWorkQueues(LazyNow* lazy_now) {
+void TaskQueueManager::ReloadEmptyWorkQueues(
+    const std::set<internal::TaskQueueImpl*>& queues_to_reload) const {
+  // There are two cases where a queue needs reloading.  First, it might be
+  // completely empty and we've just posted a task (this method handles that
+  // case). Secondly

[Sami, 2017/01/24 14:52:40]

Secondly...? :)

[alex clarke (OOO till 29th), 2017/01/24 15:28:14]

Done.

+  for (internal::TaskQueueImpl* queue : queues_to_reload) {
+    queue->ReloadImmediateWorkQueueIfEmpty();
+  }
+}
+
+void TaskQueueManager::WakeupReadyDelayedQueues(LazyNow* lazy_now) {
   TRACE_EVENT0(disabled_by_default_tracing_category_,
-               "TaskQueueManager::UpdateWorkQueues");
+               "TaskQueueManager::WakeupReadyDelayedQueues");
 
   for (TimeDomain* time_domain : time_domains_) {
     if (time_domain == real_time_domain_.get()) {
-      time_domain->UpdateWorkQueues(lazy_now);
+      time_domain->WakeupReadyDelayedQueues(lazy_now);
     } else {
       LazyNow time_domain_lazy_now = time_domain->CreateLazyNow();
-      time_domain->UpdateWorkQueues(&time_domain_lazy_now);
+      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_);
 }
 
+void TaskQueueManager::OnQueueHasIncomingImmediateWork(
+    internal::TaskQueueImpl* queue,
+    bool ensure_do_work_posted) {
+  bool on_main_thread = delegate_->BelongsToCurrentThread();
+
+  {
+    base::AutoLock lock(any_thread_lock_);
+    any_thread().has_incoming_immediate_work.insert(queue);
+
+    if (!ensure_do_work_posted)
+      return;
+
+    // De-duplicate DoWork posts.
+    if (on_main_thread) {
+      if (!main_thread_pending_wakeups_.insert(base::TimeTicks()).second)
+        return;
+    } else {
+      if (any_thread().other_thread_pending_wakeup)
+        return;
+      any_thread().other_thread_pending_wakeup = true;
+    }
+  }
+
+  if (on_main_thread) {
+    delegate_->PostTask(FROM_HERE, from_main_thread_immediate_do_work_closure_);
+  } else {
+    delegate_->PostTask(FROM_HERE,
+                        from_other_thread_immediate_do_work_closure_);
+  }
+}
+
 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 {
@@ skipping 48 matching lines @@
   }
 
   // Posting a DoWork while a DoWork is running leads to spurious DoWorks.
   main_thread_pending_wakeups_.insert(base::TimeTicks());
 
   bool is_nested = delegate_->IsNested();
   if (!is_nested)
     queues_to_delete_.clear();
 
   LazyNow lazy_now(real_time_domain()->CreateLazyNow());
-  UpdateWorkQueues(&lazy_now);
+  WakeupReadyDelayedQueues(&lazy_now);
 
   for (int i = 0; i < work_batch_size_; i++) {
+    std::set<internal::TaskQueueImpl*> queues_to_reload;

[Sami, 2017/01/24 14:52:40]

Probably doesn't make a huge difference, but unordered_set?

[alex clarke (OOO till 29th), 2017/01/24 15:28:14]

I guess.

+
+    {
+      base::AutoLock lock(any_thread_lock_);
+      std::swap(queues_to_reload, any_thread().has_incoming_immediate_work);
+    }
+
+    // It's important we call ReloadEmptyWorkQueues out side of the lock to
+    // avoid a lock order inversion.
+    ReloadEmptyWorkQueues(queues_to_reload);
+
     internal::WorkQueue* work_queue;
     if (!SelectWorkQueueToService(&work_queue))
       break;
 
     base::TimeTicks time_after_task;
     switch (ProcessTaskFromWorkQueue(work_queue, is_nested, lazy_now,
                                      &time_after_task)) {
       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.
 
     lazy_now = time_after_task.is_null() ? real_time_domain()->CreateLazyNow()
                                          : LazyNow(time_after_task);
-    UpdateWorkQueues(&lazy_now);
+    WakeupReadyDelayedQueues(&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 (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.
   base::Optional<base::TimeDelta> next_delay =
       ComputeDelayTillNextTask(&lazy_now);
 
   if (!next_delay)
     return;
 
   base::TimeDelta delay = next_delay.value();
   if (delay.is_zero()) {
     MaybeScheduleImmediateWork(FROM_HERE);
   } else {
     MaybeScheduleDelayedWork(FROM_HERE, lazy_now.Now(), delay);
   }
 }
 
 base::Optional<base::TimeDelta> TaskQueueManager::ComputeDelayTillNextTask(
     LazyNow* lazy_now) {
+  DCHECK(main_thread_checker_.CalledOnValidThread());
+
   // If the selector has non-empty queues we trivially know there is immediate
   // work to be done.
   if (!selector_.EnabledWorkQueuesEmpty())
     return base::TimeDelta();
 
+  {
+    base::AutoLock lock(any_thread_lock_);

[Sami, 2017/01/24 14:52:40]

Any lock inversion issues here? Basically I think we need to make sure
any_thread_lock_ and the per-queue-locks aren't held at the same time if
possible.

[alex clarke (OOO till 29th), 2017/01/24 15:28:14]

Yeah actually that could be one.  Can probably get the majority of the benefit
of this by asserting it doesn't have a fence and it's not disabled. Neither of
which need a lock.

[alex clarke (OOO till 29th), 2017/01/24 15:29:11]

Actually on second thoughts I think I'm trying to be too clever here.  I'll call
ReloadEmptyWorkQueues. 

+    // If we have incoming immediate work for any queue able to run, we know
+    // there is immediate work to be done.
+    for (internal::TaskQueueImpl* queue :
+         any_thread().has_incoming_immediate_work) {
+      // Note when posted ImmediateTaskCouldRun would have returned true but
+      // conditions may have changed since then.
+      if (queue->ImmediateTaskCouldRun())
+        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_) {
     base::Optional<base::TimeDelta> continuation =
         time_domain->DelayTillNextTask(lazy_now);
     if (!continuation)
       continue;
     if (!next_continuation || next_continuation.value() > continuation.value())
       next_continuation = continuation;
   }
@@ skipping 191 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();
+  {
+    base::AutoLock lock(any_thread_lock_);
+    state->BeginArray("has_incoming_immediate_work");
+    for (internal::TaskQueueImpl* task_queue :
+         any_thread().has_incoming_immediate_work) {
+      state->AppendString(task_queue->GetName());
+    }
+    state->EndArray();
+  }
   return std::move(state);
 }
 
 void TaskQueueManager::OnTaskQueueEnabled(internal::TaskQueueImpl* queue) {
   DCHECK(main_thread_checker_.CalledOnValidThread());
   DCHECK(queue->IsQueueEnabled());
   // Only schedule DoWork if there's something to do.
   if (queue->HasPendingImmediateWork() && !queue->BlockedByFence())
     MaybeScheduleImmediateWork(FROM_HERE);
 }
@@ skipping 23 matching lines @@
   for (const scoped_refptr<internal::TaskQueueImpl>& queue : queues_) {
     TimeDomain* time_domain = queue->GetTimeDomain();
     if (time_domain_now.find(time_domain) == time_domain_now.end())
       time_domain_now.insert(std::make_pair(time_domain, time_domain->Now()));
     queue->SweepCanceledDelayedTasks(time_domain_now[time_domain]);
   }
 }
 
 }  // namespace scheduler
 }  // namespace blink