Make base::Timer sequence-friendly.

base/threading/sequenced_worker_pool.cc

 // Copyright (c) 2012 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 "base/threading/sequenced_worker_pool.h"
 
 #include <stdint.h>
 
 #include <list>
 #include <map>
@@ skipping 363 matching lines @@
   };
 
   enum CleanupState {
     CLEANUP_REQUESTED,
     CLEANUP_STARTING,
     CLEANUP_RUNNING,
     CLEANUP_FINISHING,
     CLEANUP_DONE,
   };
 
+  // Clears ScheduledTasks in |delete_these_outside_lock| while ensuring that
+  // |this_worker| has the desired task info context for ~ScheduledTask() to
+  // allow RunsTasksOnCurrentThread() like checks.
+  void DeleteTheseOutsideLockHelper(

[vmpstr, 2016/12/03 01:14:04]

WDYT about just "DeleteWithoutLock" with "tasks_to_delete" as the parameter?

[gab, 2016/12/23 20:22:10]

Done on https://codereview.chromium.org/2581213002/

+      std::vector<SequencedTask>* delete_these_outside_lock,
+      Worker* this_worker);
+
   // Helper used by PostTask() to complete the work when redirection is on.
   // Returns true if the task may run at some point in the future and false if
   // it will definitely not run.
   // Coalesce upon resolution of http://crbug.com/622400.
   bool PostTaskToTaskScheduler(const SequencedTask& sequenced,
                                const TimeDelta& delay);
 
   // Returns the TaskScheduler TaskRunner for the specified |sequence_token_id|
   // and |traits|.
   scoped_refptr<TaskRunner> GetTaskSchedulerTaskRunner(
@@ skipping 17 matching lines @@
   // 3) If the return value is |GET_WORK_WAIT|, there are no tasks to run
   //    immediately, and |task| is not filled in. Likewise, |wait_time| is
   //    filled in the time to wait until the next task to run. In this case, the
   //    caller should wait the time.
   //
   // In any case, the calling code should clear the given
   // delete_these_outside_lock vector the next time the lock is released.
   // See the implementation for a more detailed description.
   GetWorkStatus GetWork(SequencedTask* task,
                         TimeDelta* wait_time,
-                        std::vector<Closure>* delete_these_outside_lock);
+                        std::vector<SequencedTask>* delete_these_outside_lock);
 
   void HandleCleanup();
 
   // Peforms init and cleanup around running the given task. WillRun...
   // returns the value from PrepareToStartAdditionalThreadIfNecessary.
   // The calling code should call FinishStartingAdditionalThread once the
   // lock is released if the return values is nonzero.
   int WillRunWorkerTask(const SequencedTask& task);
   void DidRunWorkerTask(const SequencedTask& task);
 
@@ skipping 538 matching lines @@
     while (true) {
 #if defined(OS_MACOSX)
       base::mac::ScopedNSAutoreleasePool autorelease_pool;
 #endif
 
       HandleCleanup();
 
       // See GetWork for what delete_these_outside_lock is doing.
       SequencedTask task;
       TimeDelta wait_time;
-      std::vector<Closure> delete_these_outside_lock;
+      std::vector<SequencedTask> delete_these_outside_lock;
       GetWorkStatus status =
           GetWork(&task, &wait_time, &delete_these_outside_lock);
       if (status == GET_WORK_FOUND) {
         TRACE_TASK_EXECUTION("SequencedWorkerPool::Inner::ThreadLoop", task);
         TRACE_EVENT_WITH_FLOW0(TRACE_DISABLED_BY_DEFAULT("toplevel.flow"),
             "SequencedWorkerPool::Inner::PostTask",
             TRACE_ID_MANGLE(GetTaskTraceID(task, static_cast<void*>(this))),
             TRACE_EVENT_FLAG_FLOW_IN);
         int new_thread_id = WillRunWorkerTask(task);
         {
           AutoUnlock unlock(lock_);
           // There may be more work available, so wake up another
           // worker thread. (Technically not required, since we
           // already get a signal for each new task, but it doesn't
           // hurt.)
           SignalHasWork();
-          delete_these_outside_lock.clear();
+          DeleteTheseOutsideLockHelper(&delete_these_outside_lock, this_worker);
 
           // Complete thread creation outside the lock if necessary.
           if (new_thread_id)
             FinishStartingAdditionalThread(new_thread_id);
 
           this_worker->set_running_task_info(
               SequenceToken(task.sequence_token_id), task.shutdown_behavior);
 
           tracked_objects::TaskStopwatch stopwatch;
           stopwatch.Start();
@@ skipping 10 matching lines @@
           // still works.
           task.task = Closure();
 
           this_worker->reset_running_task_info();
         }
         DidRunWorkerTask(task);  // Must be done inside the lock.
       } else if (cleanup_state_ == CLEANUP_RUNNING) {
         switch (status) {
           case GET_WORK_WAIT: {
               AutoUnlock unlock(lock_);
-              delete_these_outside_lock.clear();
+              DeleteTheseOutsideLockHelper(&delete_these_outside_lock,
+                                           this_worker);
             }
             break;
           case GET_WORK_NOT_FOUND:
             CHECK(delete_these_outside_lock.empty());
             cleanup_state_ = CLEANUP_FINISHING;
             cleanup_cv_.Broadcast();
             break;
           default:
             NOTREACHED();
         }
       } else {
         // When we're terminating and there's no more work, we can
         // shut down, other workers can complete any pending or new tasks.
         // We can get additional tasks posted after shutdown_called_ is set
         // but only worker threads are allowed to post tasks at that time, and
         // the workers responsible for posting those tasks will be available
         // to run them. Also, there may be some tasks stuck behind running
         // ones with the same sequence token, but additional threads won't
         // help this case.
         if (shutdown_called_ && blocking_shutdown_pending_task_count_ == 0) {
           AutoUnlock unlock(lock_);
-          delete_these_outside_lock.clear();
+          DeleteTheseOutsideLockHelper(&delete_these_outside_lock, this_worker);
           break;
         }
 
         // No work was found, but there are tasks that need deletion. The
         // deletion must happen outside of the lock.
         if (delete_these_outside_lock.size()) {
           AutoUnlock unlock(lock_);
-          delete_these_outside_lock.clear();
+          DeleteTheseOutsideLockHelper(&delete_these_outside_lock, this_worker);
 
           // Since the lock has been released, |status| may no longer be
           // accurate. It might read GET_WORK_WAIT even if there are tasks
           // ready to perform work. Jump to the top of the loop to recalculate
           // |status|.
           continue;
         }
 
         waiting_thread_count_++;
 
         switch (status) {
           case GET_WORK_NOT_FOUND:
             has_work_cv_.Wait();
             break;
           case GET_WORK_WAIT:
             has_work_cv_.TimedWait(wait_time);
             break;
           default:
             NOTREACHED();
         }
         waiting_thread_count_--;
       }
+      // |delete_these_outside_lock| should have been cleared via
+      // DeleteTheseOutsideLockHelper() above already.
+      DCHECK(delete_these_outside_lock.empty());
     }
   }  // Release lock_.
 
   // We noticed we should exit. Wake up the next worker so it knows it should
   // exit as well (because the Shutdown() code only signals once).
   SignalHasWork();
 
   // Possibly unblock shutdown.
   can_shutdown_cv_.Signal();
 }
 
+void SequencedWorkerPool::Inner::DeleteTheseOutsideLockHelper(
+    std::vector<SequencedTask>* delete_these_outside_lock,
+    Worker* this_worker) {
+  while (!delete_these_outside_lock->empty()) {
+    const SequencedTask& deleted_task = delete_these_outside_lock->back();
+    this_worker->set_running_task_info(
+        SequenceToken(deleted_task.sequence_token_id),
+        deleted_task.shutdown_behavior);
+    delete_these_outside_lock->pop_back();
+    this_worker->reset_running_task_info();

[vmpstr, 2016/12/03 01:14:04]

Does this need to happen after every iteration, or can it just happen once
before the function exits?

[gab, 2016/12/23 20:22:10]

Done on https://codereview.chromium.org/2581213002/

+  }
+}
+
 void SequencedWorkerPool::Inner::HandleCleanup() {
   DCHECK_EQ(AllPoolsState::USE_WORKER_POOL, g_all_pools_state);
 
   lock_.AssertAcquired();
   if (cleanup_state_ == CLEANUP_DONE)
     return;
   if (cleanup_state_ == CLEANUP_REQUESTED) {
     // We win, we get to do the cleanup as soon as the others wise up and idle.
     cleanup_state_ = CLEANUP_STARTING;
     while (thread_being_created_ ||
@@ skipping 47 matching lines @@
 
 int64_t SequencedWorkerPool::Inner::LockedGetNextSequenceTaskNumber() {
   lock_.AssertAcquired();
   // We assume that we never create enough tasks to wrap around.
   return next_sequence_task_number_++;
 }
 
 SequencedWorkerPool::Inner::GetWorkStatus SequencedWorkerPool::Inner::GetWork(
     SequencedTask* task,
     TimeDelta* wait_time,
-    std::vector<Closure>* delete_these_outside_lock) {
+    std::vector<SequencedTask>* delete_these_outside_lock) {
   DCHECK_EQ(AllPoolsState::USE_WORKER_POOL, g_all_pools_state);
 
   lock_.AssertAcquired();
 
   // Find the next task with a sequence token that's not currently in use.
   // If the token is in use, that means another thread is running something
   // in that sequence, and we can't run it without going out-of-order.
   //
   // This algorithm is simple and fair, but inefficient in some cases. For
   // example, say somebody schedules 1000 slow tasks with the same sequence
@@ skipping 25 matching lines @@
       unrunnable_tasks++;
       ++i;
       continue;
     }
 
     if (shutdown_called_ && i->shutdown_behavior != BLOCK_SHUTDOWN) {
       // We're shutting down and the task we just found isn't blocking
       // shutdown. Delete it and get more work.
       //
       // Note that we do not want to delete unrunnable tasks. Deleting a task
-      // can have side effects (like freeing some objects) and deleting a
-      // task that's supposed to run after one that's currently running could
-      // cause an obscure crash.
+      // can have side effects (like freeing some objects) and deleting a task
+      // that's supposed to run after one that's currently running could cause
+      // an obscure crash.
       //
       // We really want to delete these tasks outside the lock in case the
-      // closures are holding refs to objects that want to post work from
-      // their destructorss (which would deadlock). The closures are
-      // internally refcounted, so we just need to keep a copy of them alive
-      // until the lock is exited. The calling code can just clear() the
-      // vector they passed to us once the lock is exited to make this
-      // happen.
-      delete_these_outside_lock->push_back(i->task);
+      // closures are holding refs to objects that want to post work from their
+      // destructors (which would deadlock). The closures are internally
+      // refcounted, so we just need to keep a copy of them alive until the lock
+      // is exited. The calling code can just clear() the vector they passed to
+      // us once the lock is exited to make this happen.
+      delete_these_outside_lock->push_back(*i);
       pending_tasks_.erase(i++);
       continue;
     }
 
     if (i->time_to_run > current_time) {
       // The time to run has not come yet.
       *wait_time = i->time_to_run - current_time;
       status = GET_WORK_WAIT;
       if (cleanup_state_ == CLEANUP_RUNNING) {
         // Deferred tasks are deleted when cleaning up, see Inner::ThreadLoop.
-        delete_these_outside_lock->push_back(i->task);
+        delete_these_outside_lock->push_back(*i);
         pending_tasks_.erase(i);
       }
       break;
     }
 
     // Found a runnable task.
     *task = *i;
     pending_tasks_.erase(i);
     if (task->shutdown_behavior == BLOCK_SHUTDOWN) {
       blocking_shutdown_pending_task_count_--;
@@ skipping 358 matching lines @@
 bool SequencedWorkerPool::IsShutdownInProgress() {
   return inner_->IsShutdownInProgress();
 }
 
 bool SequencedWorkerPool::IsRunningSequenceOnCurrentThread(
     SequenceToken sequence_token) const {
   return inner_->IsRunningSequenceOnCurrentThread(sequence_token);
 }
 
 }  // namespace base