Unify locks in TaskQueueImpl

We no longer need to split the lock in TaskQueueImpl.

Change-Id: I05d1bc597f34fbe4a1a10ffc3a95fc451b72abc6
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/1504017
Commit-Queue: Alex Clarke <alexclarke@chromium.org>
Reviewed-by: Alexander Timin <altimin@chromium.org>
Cr-Commit-Position: refs/heads/master@{#638068}
diff --git a/base/task/sequence_manager/task_queue_impl.cc b/base/task/sequence_manager/task_queue_impl.cc
index 850f177..53e28e9 100644
--- a/base/task/sequence_manager/task_queue_impl.cc
+++ b/base/task/sequence_manager/task_queue_impl.cc
@@ -118,7 +118,7 @@
   // contains a strong reference to this TaskQueueImpl and the
   // SequenceManagerImpl destructor calls UnregisterTaskQueue on all task
   // queues.
-  DCHECK(any_thread().unregistered)
+  DCHECK(any_thread_.unregistered)
       << "UnregisterTaskQueue must be called first!";
 #endif
 }
@@ -160,20 +160,18 @@
 
   {
     AutoLock lock(any_thread_lock_);
-    AutoLock immediate_incoming_queue_lock(immediate_incoming_queue_lock_);
-
     if (main_thread_only().time_domain)
       main_thread_only().time_domain->UnregisterQueue(this);
 
-    any_thread().unregistered = true;
+    any_thread_.unregistered = true;
 
     main_thread_only().on_task_completed_handler = OnTaskCompletedHandler();
-    any_thread().time_domain = nullptr;
+    any_thread_.time_domain = nullptr;
     main_thread_only().time_domain = nullptr;
 
     main_thread_only().on_next_wake_up_changed_callback =
         OnNextWakeUpChangedCallback();
-    immediate_incoming_queue.swap(immediate_incoming_queue_);
+    immediate_incoming_queue.swap(any_thread_.immediate_incoming_queue);
 
     empty_queues_to_reload_handle_.ReleaseAtomicFlag();
   }
@@ -221,42 +219,40 @@
   // for details.
   CHECK(task.callback);
 
-  TimeTicks now;
-  bool add_queue_time_to_tasks = sequence_manager_->GetAddQueueTimeToTasks();
-  if (delayed_fence_allowed_ || add_queue_time_to_tasks) {
-    if (current_thread == CurrentThread::kMainThread) {
-      now = main_thread_only().time_domain->Now();
-    } else {
-      AutoLock lock(any_thread_lock_);
-      now = any_thread().time_domain->Now();
-    }
-    if (add_queue_time_to_tasks) {
-      task.queue_time = now;
-    }
-  }
-
   bool should_schedule_work = false;
   {
     // TODO(alexclarke): Maybe add a main thread only immediate_incoming_queue
     // See https://crbug.com/901800
-    AutoLock lock(immediate_incoming_queue_lock_);
+    AutoLock lock(any_thread_lock_);
+    TimeTicks now;
+    bool add_queue_time_to_tasks = sequence_manager_->GetAddQueueTimeToTasks();
+    if (delayed_fence_allowed_ || add_queue_time_to_tasks) {
+      now = any_thread_.time_domain->Now();
+      if (add_queue_time_to_tasks)
+        task.queue_time = now;
+    }
+
     // The sequence number must be incremented atomically with pushing onto the
     // incoming queue. Otherwise if there are several threads posting task we
     // risk breaking the assumption that sequence numbers increase monotonically
     // within a queue.
     EnqueueOrder sequence_number = sequence_manager_->GetNextSequenceNumber();
-    bool was_immediate_incoming_queue_empty = immediate_incoming_queue_.empty();
-    immediate_incoming_queue_.push_back(
+    bool was_immediate_incoming_queue_empty =
+        any_thread_.immediate_incoming_queue.empty();
+    any_thread_.immediate_incoming_queue.push_back(
         Task(std::move(task), now, sequence_number, sequence_number));
-    sequence_manager_->WillQueueTask(&immediate_incoming_queue_.back());
+    sequence_manager_->WillQueueTask(
+        &any_thread_.immediate_incoming_queue.back());
 
     // If this queue was completely empty, then the SequenceManager needs to be
     // informed so it can reload the work queue and add us to the
     // TaskQueueSelector which can only be done from the main thread. In
     // addition it may need to schedule a DoWork if this queue isn't blocked.
-    if (was_immediate_incoming_queue_empty && immediate_work_queue_empty_) {
+    if (was_immediate_incoming_queue_empty &&
+        any_thread_.immediate_work_queue_empty) {
       empty_queues_to_reload_handle_.SetActive(true);
-      should_schedule_work = post_immediate_task_should_schedule_work_;
+      should_schedule_work =
+          any_thread_.post_immediate_task_should_schedule_work;
     }
   }
 
@@ -265,9 +261,9 @@
   // http://shortn/_ntnKNqjDQT for a discussion.
   //
   // Calling ScheduleWork outside the lock should be safe, only the main thread
-  // can mutate |post_immediate_task_should_schedule_work_|. If it transitions
-  // to false we call ScheduleWork redundantly that's harmless. If it
-  // transitions to true, the side effect of
+  // can mutate |any_thread_.post_immediate_task_should_schedule_work|. If it
+  // transitions to false we call ScheduleWork redundantly that's harmless. If
+  // it transitions to true, the side effect of
   // |empty_queues_to_reload_handle_SetActive(true)| is guaranteed to be picked
   // up by the ThreadController's call to SequenceManagerImpl::DelayTillNextTask
   // when it computes what continuation (if any) is needed.
@@ -318,7 +314,7 @@
     TimeTicks time_domain_now;
     {
       AutoLock lock(any_thread_lock_);
-      time_domain_now = any_thread().time_domain->Now();
+      time_domain_now = any_thread_.time_domain->Now();
     }
     TimeTicks time_domain_delayed_run_time = time_domain_now + task.delay;
     if (sequence_manager_->GetAddQueueTimeToTasks()) {
@@ -389,13 +385,13 @@
 }
 
 void TaskQueueImpl::TakeImmediateIncomingQueueTasks(TaskDeque* queue) {
-  AutoLock immediate_incoming_queue_lock(immediate_incoming_queue_lock_);
+  AutoLock any_thread_lock(any_thread_lock_);
   DCHECK(queue->empty());
-  queue->swap(immediate_incoming_queue_);
+  queue->swap(any_thread_.immediate_incoming_queue);
 
   // Since |immediate_incoming_queue| is empty, now is a good time to consider
   // reducing it's capacity if we're wasting memory.
-  immediate_incoming_queue_.MaybeShrinkQueue();
+  any_thread_.immediate_incoming_queue.MaybeShrinkQueue();
 
   // Activate delayed fence if necessary. This is ideologically similar to
   // ActivateDelayedFenceIfNeeded, but due to immediate tasks being posted
@@ -429,8 +425,8 @@
     return false;
   }
 
-  AutoLock lock(immediate_incoming_queue_lock_);
-  return immediate_incoming_queue_.empty();
+  AutoLock lock(any_thread_lock_);
+  return any_thread_.immediate_incoming_queue.empty();
 }
 
 size_t TaskQueueImpl::GetNumberOfPendingTasks() const {
@@ -439,8 +435,8 @@
   task_count += main_thread_only().delayed_incoming_queue.size();
   task_count += main_thread_only().immediate_work_queue->Size();
 
-  AutoLock lock(immediate_incoming_queue_lock_);
-  task_count += immediate_incoming_queue_.size();
+  AutoLock lock(any_thread_lock_);
+  task_count += any_thread_.immediate_incoming_queue.size();
   return task_count;
 }
 
@@ -460,8 +456,8 @@
   }
 
   // Finally tasks on |immediate_incoming_queue| count as immediate work.
-  AutoLock lock(immediate_incoming_queue_lock_);
-  return !immediate_incoming_queue_.empty();
+  AutoLock lock(any_thread_lock_);
+  return !any_thread_.immediate_incoming_queue.empty();
 }
 
 Optional<DelayedWakeUp> TaskQueueImpl::GetNextScheduledWakeUpImpl() {
@@ -520,9 +516,9 @@
   if (!associated_thread_->IsBoundToCurrentThread())
     return;
 
-  AutoLock lock(immediate_incoming_queue_lock_);
+  AutoLock lock(any_thread_lock_);
   TRACE_COUNTER1(TRACE_DISABLED_BY_DEFAULT("sequence_manager"), GetName(),
-                 immediate_incoming_queue_.size() +
+                 any_thread_.immediate_incoming_queue.size() +
                      main_thread_only().immediate_work_queue->Size() +
                      main_thread_only().delayed_work_queue->Size() +
                      main_thread_only().delayed_incoming_queue.size());
@@ -545,10 +541,9 @@
                                 trace_event::TracedValue* state,
                                 bool force_verbose) const {
   AutoLock lock(any_thread_lock_);
-  AutoLock immediate_incoming_queue_lock(immediate_incoming_queue_lock_);
   state->BeginDictionary();
   state->SetString("name", GetName());
-  if (any_thread().unregistered) {
+  if (any_thread_.unregistered) {
     state->SetBoolean("unregistered", true);
     state->EndDictionary();
     return;
@@ -564,8 +559,8 @@
   state->SetBoolean("enabled", IsQueueEnabled());
   state->SetString("time_domain_name",
                    main_thread_only().time_domain->GetName());
-  state->SetInteger("immediate_incoming_queue_size",
-                    immediate_incoming_queue_.size());
+  state->SetInteger("any_thread_.immediate_incoming_queuesize",
+                    any_thread_.immediate_incoming_queue.size());
   state->SetInteger("delayed_incoming_queue_size",
                     main_thread_only().delayed_incoming_queue.size());
   state->SetInteger("immediate_work_queue_size",
@@ -573,8 +568,8 @@
   state->SetInteger("delayed_work_queue_size",
                     main_thread_only().delayed_work_queue->Size());
 
-  state->SetInteger("immediate_incoming_queue_capacity",
-                    immediate_incoming_queue_.capacity());
+  state->SetInteger("any_thread_.immediate_incoming_queuecapacity",
+                    any_thread_.immediate_incoming_queue.capacity());
   state->SetInteger("immediate_work_queue_capacity",
                     immediate_work_queue()->Capacity());
   state->SetInteger("delayed_work_queue_capacity",
@@ -602,7 +597,7 @@
 
   if (verbose || force_verbose) {
     state->BeginArray("immediate_incoming_queue");
-    QueueAsValueInto(immediate_incoming_queue_, now, state);
+    QueueAsValueInto(any_thread_.immediate_incoming_queue, now, state);
     state->EndArray();
     state->BeginArray("delayed_work_queue");
     main_thread_only().delayed_work_queue->AsValueInto(now, state);
@@ -647,14 +642,14 @@
   {
     AutoLock lock(any_thread_lock_);
     DCHECK(time_domain);
-    DCHECK(!any_thread().unregistered);
-    if (any_thread().unregistered)
+    DCHECK(!any_thread_.unregistered);
+    if (any_thread_.unregistered)
       return;
     DCHECK_CALLED_ON_VALID_THREAD(associated_thread_->thread_checker);
     if (time_domain == main_thread_only().time_domain)
       return;
 
-    any_thread().time_domain = time_domain;
+    any_thread_.time_domain = time_domain;
   }
 
   main_thread_only().time_domain->UnregisterQueue(this);
@@ -697,11 +692,13 @@
       main_thread_only().delayed_work_queue->InsertFence(current_fence);
 
   {
-    AutoLock lock(immediate_incoming_queue_lock_);
+    AutoLock lock(any_thread_lock_);
     if (!task_unblocked && previous_fence && previous_fence < current_fence) {
-      if (!immediate_incoming_queue_.empty() &&
-          immediate_incoming_queue_.front().enqueue_order() > previous_fence &&
-          immediate_incoming_queue_.front().enqueue_order() < current_fence) {
+      if (!any_thread_.immediate_incoming_queue.empty() &&
+          any_thread_.immediate_incoming_queue.front().enqueue_order() >
+              previous_fence &&
+          any_thread_.immediate_incoming_queue.front().enqueue_order() <
+              current_fence) {
         task_unblocked = true;
       }
     }
@@ -732,10 +729,11 @@
   task_unblocked |= main_thread_only().delayed_work_queue->RemoveFence();
 
   {
-    AutoLock lock(immediate_incoming_queue_lock_);
+    AutoLock lock(any_thread_lock_);
     if (!task_unblocked && previous_fence) {
-      if (!immediate_incoming_queue_.empty() &&
-          immediate_incoming_queue_.front().enqueue_order() > previous_fence) {
+      if (!any_thread_.immediate_incoming_queue.empty() &&
+          any_thread_.immediate_incoming_queue.front().enqueue_order() >
+              previous_fence) {
         task_unblocked = true;
       }
     }
@@ -756,11 +754,11 @@
     return false;
   }
 
-  AutoLock lock(immediate_incoming_queue_lock_);
-  if (immediate_incoming_queue_.empty())
+  AutoLock lock(any_thread_lock_);
+  if (any_thread_.immediate_incoming_queue.empty())
     return true;
 
-  return immediate_incoming_queue_.front().enqueue_order() >
+  return any_thread_.immediate_incoming_queue.front().enqueue_order() >
          main_thread_only().current_fence;
 }
 
@@ -833,7 +831,7 @@
   bool has_pending_immediate_work;
 
   {
-    AutoLock lock(immediate_incoming_queue_lock_);
+    AutoLock lock(any_thread_lock_);
     UpdateCrossThreadQueueStateLocked();
     has_pending_immediate_work = HasPendingImmediateWorkLocked();
   }
@@ -854,18 +852,18 @@
 }
 
 void TaskQueueImpl::UpdateCrossThreadQueueStateLocked() {
-  immediate_work_queue_empty_ =
+  any_thread_.immediate_work_queue_empty =
       main_thread_only().immediate_work_queue->Empty();
 
   if (main_thread_only().on_next_wake_up_changed_callback) {
     // If there's a callback we need a DoWork for the callback to be issued by
     // ReloadEmptyImmediateWorkQueue. The callback isn't
     // sent for disabled queues.
-    post_immediate_task_should_schedule_work_ = IsQueueEnabled();
+    any_thread_.post_immediate_task_should_schedule_work = IsQueueEnabled();
   } else {
     // Otherwise we need PostImmediateTaskImpl to ScheduleWork unless the queue
     // is blocked or disabled.
-    post_immediate_task_should_schedule_work_ =
+    any_thread_.post_immediate_task_should_schedule_work =
         IsQueueEnabled() && !main_thread_only().current_fence;
   }
 }
@@ -880,8 +878,8 @@
   main_thread_only().immediate_work_queue->MaybeShrinkQueue();
 
   {
-    AutoLock lock(immediate_incoming_queue_lock_);
-    immediate_incoming_queue_.MaybeShrinkQueue();
+    AutoLock lock(any_thread_lock_);
+    any_thread_.immediate_incoming_queue.MaybeShrinkQueue();
   }
 
   LazyNow lazy_now(now);
@@ -889,8 +887,8 @@
 }
 
 void TaskQueueImpl::PushImmediateIncomingTaskForTest(Task&& task) {
-  AutoLock lock(immediate_incoming_queue_lock_);
-  immediate_incoming_queue_.push_back(std::move(task));
+  AutoLock lock(any_thread_lock_);
+  any_thread_.immediate_incoming_queue.push_back(std::move(task));
 }
 
 void TaskQueueImpl::RequeueDeferredNonNestableTask(
@@ -906,15 +904,16 @@
   } else {
     // We're about to push |task| onto an empty |immediate_work_queue|. This
     // may mean we'd be violating the contract of AtomicFlagSet, it's
-    // only supposed to contain queues where |immediate_incoming_queue_| is
-    // non empty but |immediate_work_queue| is empty. We remedy that by removing
-    // ourselves from that list (a NOP if we're not in the list).
+    // only supposed to contain queues where
+    // |any_thread_.immediate_incoming_queue| is non empty but
+    // |immediate_work_queue| is empty. We remedy that by removing ourselves
+    // from that list (a NOP if we're not in the list).
     if (main_thread_only().immediate_work_queue->Empty()) {
       empty_queues_to_reload_handle_.SetActive(false);
 
       {
-        AutoLock lock(immediate_incoming_queue_lock_);
-        immediate_work_queue_empty_ = false;
+        AutoLock lock(any_thread_lock_);
+        any_thread_.immediate_work_queue_empty = false;
         main_thread_only().immediate_work_queue->PushNonNestableTaskToFront(
             std::move(task.task));
       }
@@ -974,15 +973,14 @@
   }
 
   // Finally tasks on |immediate_incoming_queue| count as immediate work.
-  AutoLock lock(immediate_incoming_queue_lock_);
-  return !immediate_incoming_queue_.empty();
+  AutoLock lock(any_thread_lock_);
+  return !any_thread_.immediate_incoming_queue.empty();
 }
 
 bool TaskQueueImpl::HasPendingImmediateWorkLocked() {
-  immediate_incoming_queue_lock_.AssertAcquired();
   return !main_thread_only().delayed_work_queue->Empty() ||
          !main_thread_only().immediate_work_queue->Empty() ||
-         !immediate_incoming_queue_.empty();
+         !any_thread_.immediate_incoming_queue.empty();
 }
 
 void TaskQueueImpl::SetOnTaskStartedHandler(
@@ -1016,7 +1014,7 @@
 
 bool TaskQueueImpl::IsUnregistered() const {
   AutoLock lock(any_thread_lock_);
-  return any_thread().unregistered;
+  return any_thread_.unregistered;
 }
 
 WeakPtr<SequenceManagerImpl> TaskQueueImpl::GetSequenceManagerWeakPtr() {
@@ -1044,9 +1042,9 @@
   {
     // Limit the scope of the lock to ensure that the deque is destroyed
     // outside of the lock to allow it to post tasks.
-    base::AutoLock lock(immediate_incoming_queue_lock_);
-    deque.swap(immediate_incoming_queue_);
-    immediate_work_queue_empty_ = true;
+    base::AutoLock lock(any_thread_lock_);
+    deque.swap(any_thread_.immediate_incoming_queue);
+    any_thread_.immediate_work_queue_empty = true;
   }
 
   LazyNow lazy_now = main_thread_only().time_domain->CreateLazyNow();
@@ -1061,8 +1059,8 @@
   if (!main_thread_only().delayed_incoming_queue.empty())
     return true;
 
-  base::AutoLock lock(immediate_incoming_queue_lock_);
-  if (!immediate_incoming_queue_.empty())
+  base::AutoLock lock(any_thread_lock_);
+  if (!any_thread_.immediate_incoming_queue.empty())
     return true;
 
   return false;
diff --git a/base/task/sequence_manager/task_queue_impl.h b/base/task/sequence_manager/task_queue_impl.h
index 24af883..f06f599 100644
--- a/base/task/sequence_manager/task_queue_impl.h
+++ b/base/task/sequence_manager/task_queue_impl.h
@@ -146,7 +146,7 @@
   // Used to check if we need to generate notifications about delayed work.
   bool HasPendingImmediateWork();
   bool HasPendingImmediateWorkLocked()
-      EXCLUSIVE_LOCKS_REQUIRED(immediate_incoming_queue_lock_);
+      EXCLUSIVE_LOCKS_REQUIRED(any_thread_lock_);
 
   bool has_pending_high_resolution_tasks() const {
     return main_thread_only()
@@ -282,18 +282,6 @@
     const int task_type_;
   };
 
-  struct AnyThread {
-    explicit AnyThread(TimeDomain* time_domain);
-    ~AnyThread();
-
-    // TimeDomain is maintained in two copies: inside AnyThread and inside
-    // MainThreadOnly. It can be changed only from main thread, so it should be
-    // locked before accessing from other threads.
-    TimeDomain* time_domain;
-
-    bool unregistered = false;
-  };
-
   // A queue for holding delayed tasks before their delay has expired.
   struct DelayedIncomingQueue {
    public:
@@ -376,7 +364,8 @@
 
   void ScheduleDelayedWorkTask(Task pending_task);
 
-  void MoveReadyImmediateTasksToImmediateWorkQueueLocked();
+  void MoveReadyImmediateTasksToImmediateWorkQueueLocked()
+      EXCLUSIVE_LOCKS_REQUIRED(any_thread_lock_);
 
   // LazilyDeallocatedDeque use TimeTicks to figure out when to resize.  We
   // should use real time here always.
@@ -409,9 +398,9 @@
   // Activate a delayed fence if a time has come.
   void ActivateDelayedFenceIfNeeded(TimeTicks now);
 
-  // Updates state protected by immediate_incoming_queue_lock_.
+  // Updates state protected by any_thread_lock_.
   void UpdateCrossThreadQueueStateLocked()
-      EXCLUSIVE_LOCKS_REQUIRED(immediate_incoming_queue_lock_);
+      EXCLUSIVE_LOCKS_REQUIRED(any_thread_lock_);
 
   const char* name_;
   SequenceManagerImpl* const sequence_manager_;
@@ -421,15 +410,27 @@
   const scoped_refptr<GuardedTaskPoster> task_poster_;
 
   mutable Lock any_thread_lock_;
-  AnyThread any_thread_;
-  struct AnyThread& any_thread() {
-    any_thread_lock_.AssertAcquired();
-    return any_thread_;
-  }
-  const struct AnyThread& any_thread() const {
-    any_thread_lock_.AssertAcquired();
-    return any_thread_;
-  }
+
+  struct AnyThread {
+    explicit AnyThread(TimeDomain* time_domain);
+    ~AnyThread();
+
+    // TimeDomain is maintained in two copies: inside AnyThread and inside
+    // MainThreadOnly. It can be changed only from main thread, so it should be
+    // locked before accessing from other threads.
+    TimeDomain* time_domain;
+
+    TaskDeque immediate_incoming_queue;
+
+    // True if main_thread_only().immediate_work_queue is empty.
+    bool immediate_work_queue_empty = true;
+
+    bool post_immediate_task_should_schedule_work = true;
+
+    bool unregistered = false;
+  };
+
+  AnyThread any_thread_ GUARDED_BY(any_thread_lock_);
 
   MainThreadOnly main_thread_only_;
   MainThreadOnly& main_thread_only() {
@@ -441,15 +442,6 @@
     return main_thread_only_;
   }
 
-  mutable Lock immediate_incoming_queue_lock_;
-  TaskDeque immediate_incoming_queue_
-      GUARDED_BY(immediate_incoming_queue_lock_);
-  // True if main_thread_only().immediate_work_queue is empty.
-  bool immediate_work_queue_empty_ GUARDED_BY(immediate_incoming_queue_lock_) =
-      true;
-  bool post_immediate_task_should_schedule_work_
-      GUARDED_BY(immediate_incoming_queue_lock_) = true;
-
   // Handle to our entry within the SequenceManagers |empty_queues_to_reload_|
   // atomic flag set. Used to signal that this queue needs to be reloaded.
   AtomicFlagSet::AtomicFlag empty_queues_to_reload_handle_;