Added llvm into exceptions as we can't add README.chromium into 3rd party repository
[chromium-blink-merge.git] / base / threading / sequenced_worker_pool.cc
blob89224f74aa4d7b5cafa4398e5b6d2a3c6121c5bb
1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
2 // Use of this source code is governed by a BSD-style license that can be
3 // found in the LICENSE file.
5 #include "base/threading/sequenced_worker_pool.h"
7 #include <list>
8 #include <map>
9 #include <set>
10 #include <utility>
11 #include <vector>
13 #include "base/atomic_sequence_num.h"
14 #include "base/callback.h"
15 #include "base/compiler_specific.h"
16 #include "base/critical_closure.h"
17 #include "base/lazy_instance.h"
18 #include "base/logging.h"
19 #include "base/memory/linked_ptr.h"
20 #include "base/message_loop/message_loop_proxy.h"
21 #include "base/stl_util.h"
22 #include "base/strings/stringprintf.h"
23 #include "base/synchronization/condition_variable.h"
24 #include "base/synchronization/lock.h"
25 #include "base/threading/platform_thread.h"
26 #include "base/threading/simple_thread.h"
27 #include "base/threading/thread_local.h"
28 #include "base/threading/thread_restrictions.h"
29 #include "base/time/time.h"
30 #include "base/trace_event/trace_event.h"
31 #include "base/tracked_objects.h"
33 #if defined(OS_MACOSX)
34 #include "base/mac/scoped_nsautorelease_pool.h"
35 #elif defined(OS_WIN)
36 #include "base/win/scoped_com_initializer.h"
37 #endif
39 #if !defined(OS_NACL)
40 #include "base/metrics/histogram.h"
41 #endif
43 namespace base {
45 namespace {
47 struct SequencedTask : public TrackingInfo {
48 SequencedTask()
49 : sequence_token_id(0),
50 trace_id(0),
51 sequence_task_number(0),
52 shutdown_behavior(SequencedWorkerPool::BLOCK_SHUTDOWN) {}
54 explicit SequencedTask(const tracked_objects::Location& from_here)
55 : base::TrackingInfo(from_here, TimeTicks()),
56 sequence_token_id(0),
57 trace_id(0),
58 sequence_task_number(0),
59 shutdown_behavior(SequencedWorkerPool::BLOCK_SHUTDOWN) {}
61 ~SequencedTask() {}
63 int sequence_token_id;
64 int trace_id;
65 int64 sequence_task_number;
66 SequencedWorkerPool::WorkerShutdown shutdown_behavior;
67 tracked_objects::Location posted_from;
68 Closure task;
70 // Non-delayed tasks and delayed tasks are managed together by time-to-run
71 // order. We calculate the time by adding the posted time and the given delay.
72 TimeTicks time_to_run;
75 struct SequencedTaskLessThan {
76 public:
77 bool operator()(const SequencedTask& lhs, const SequencedTask& rhs) const {
78 if (lhs.time_to_run < rhs.time_to_run)
79 return true;
81 if (lhs.time_to_run > rhs.time_to_run)
82 return false;
84 // If the time happen to match, then we use the sequence number to decide.
85 return lhs.sequence_task_number < rhs.sequence_task_number;
89 // SequencedWorkerPoolTaskRunner ---------------------------------------------
90 // A TaskRunner which posts tasks to a SequencedWorkerPool with a
91 // fixed ShutdownBehavior.
93 // Note that this class is RefCountedThreadSafe (inherited from TaskRunner).
94 class SequencedWorkerPoolTaskRunner : public TaskRunner {
95 public:
96 SequencedWorkerPoolTaskRunner(
97 const scoped_refptr<SequencedWorkerPool>& pool,
98 SequencedWorkerPool::WorkerShutdown shutdown_behavior);
100 // TaskRunner implementation
101 bool PostDelayedTask(const tracked_objects::Location& from_here,
102 const Closure& task,
103 TimeDelta delay) override;
104 bool RunsTasksOnCurrentThread() const override;
106 private:
107 ~SequencedWorkerPoolTaskRunner() override;
109 const scoped_refptr<SequencedWorkerPool> pool_;
111 const SequencedWorkerPool::WorkerShutdown shutdown_behavior_;
113 DISALLOW_COPY_AND_ASSIGN(SequencedWorkerPoolTaskRunner);
116 SequencedWorkerPoolTaskRunner::SequencedWorkerPoolTaskRunner(
117 const scoped_refptr<SequencedWorkerPool>& pool,
118 SequencedWorkerPool::WorkerShutdown shutdown_behavior)
119 : pool_(pool),
120 shutdown_behavior_(shutdown_behavior) {
123 SequencedWorkerPoolTaskRunner::~SequencedWorkerPoolTaskRunner() {
126 bool SequencedWorkerPoolTaskRunner::PostDelayedTask(
127 const tracked_objects::Location& from_here,
128 const Closure& task,
129 TimeDelta delay) {
130 if (delay == TimeDelta()) {
131 return pool_->PostWorkerTaskWithShutdownBehavior(
132 from_here, task, shutdown_behavior_);
134 return pool_->PostDelayedWorkerTask(from_here, task, delay);
137 bool SequencedWorkerPoolTaskRunner::RunsTasksOnCurrentThread() const {
138 return pool_->RunsTasksOnCurrentThread();
141 // SequencedWorkerPoolSequencedTaskRunner ------------------------------------
142 // A SequencedTaskRunner which posts tasks to a SequencedWorkerPool with a
143 // fixed sequence token.
145 // Note that this class is RefCountedThreadSafe (inherited from TaskRunner).
146 class SequencedWorkerPoolSequencedTaskRunner : public SequencedTaskRunner {
147 public:
148 SequencedWorkerPoolSequencedTaskRunner(
149 const scoped_refptr<SequencedWorkerPool>& pool,
150 SequencedWorkerPool::SequenceToken token,
151 SequencedWorkerPool::WorkerShutdown shutdown_behavior);
153 // TaskRunner implementation
154 bool PostDelayedTask(const tracked_objects::Location& from_here,
155 const Closure& task,
156 TimeDelta delay) override;
157 bool RunsTasksOnCurrentThread() const override;
159 // SequencedTaskRunner implementation
160 bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here,
161 const Closure& task,
162 TimeDelta delay) override;
164 private:
165 ~SequencedWorkerPoolSequencedTaskRunner() override;
167 const scoped_refptr<SequencedWorkerPool> pool_;
169 const SequencedWorkerPool::SequenceToken token_;
171 const SequencedWorkerPool::WorkerShutdown shutdown_behavior_;
173 DISALLOW_COPY_AND_ASSIGN(SequencedWorkerPoolSequencedTaskRunner);
176 SequencedWorkerPoolSequencedTaskRunner::SequencedWorkerPoolSequencedTaskRunner(
177 const scoped_refptr<SequencedWorkerPool>& pool,
178 SequencedWorkerPool::SequenceToken token,
179 SequencedWorkerPool::WorkerShutdown shutdown_behavior)
180 : pool_(pool),
181 token_(token),
182 shutdown_behavior_(shutdown_behavior) {
185 SequencedWorkerPoolSequencedTaskRunner::
186 ~SequencedWorkerPoolSequencedTaskRunner() {
189 bool SequencedWorkerPoolSequencedTaskRunner::PostDelayedTask(
190 const tracked_objects::Location& from_here,
191 const Closure& task,
192 TimeDelta delay) {
193 if (delay == TimeDelta()) {
194 return pool_->PostSequencedWorkerTaskWithShutdownBehavior(
195 token_, from_here, task, shutdown_behavior_);
197 return pool_->PostDelayedSequencedWorkerTask(token_, from_here, task, delay);
200 bool SequencedWorkerPoolSequencedTaskRunner::RunsTasksOnCurrentThread() const {
201 return pool_->IsRunningSequenceOnCurrentThread(token_);
204 bool SequencedWorkerPoolSequencedTaskRunner::PostNonNestableDelayedTask(
205 const tracked_objects::Location& from_here,
206 const Closure& task,
207 TimeDelta delay) {
208 // There's no way to run nested tasks, so simply forward to
209 // PostDelayedTask.
210 return PostDelayedTask(from_here, task, delay);
213 // Create a process-wide unique ID to represent this task in trace events. This
214 // will be mangled with a Process ID hash to reduce the likelyhood of colliding
215 // with MessageLoop pointers on other processes.
216 uint64 GetTaskTraceID(const SequencedTask& task,
217 void* pool) {
218 return (static_cast<uint64>(task.trace_id) << 32) |
219 static_cast<uint64>(reinterpret_cast<intptr_t>(pool));
222 base::LazyInstance<base::ThreadLocalPointer<
223 SequencedWorkerPool::SequenceToken> >::Leaky g_lazy_tls_ptr =
224 LAZY_INSTANCE_INITIALIZER;
226 } // namespace
228 // Worker ---------------------------------------------------------------------
230 class SequencedWorkerPool::Worker : public SimpleThread {
231 public:
232 // Hold a (cyclic) ref to |worker_pool|, since we want to keep it
233 // around as long as we are running.
234 Worker(const scoped_refptr<SequencedWorkerPool>& worker_pool,
235 int thread_number,
236 const std::string& thread_name_prefix);
237 ~Worker() override;
239 // SimpleThread implementation. This actually runs the background thread.
240 void Run() override;
242 void set_running_task_info(SequenceToken token,
243 WorkerShutdown shutdown_behavior) {
244 running_sequence_ = token;
245 running_shutdown_behavior_ = shutdown_behavior;
248 SequenceToken running_sequence() const {
249 return running_sequence_;
252 WorkerShutdown running_shutdown_behavior() const {
253 return running_shutdown_behavior_;
256 private:
257 scoped_refptr<SequencedWorkerPool> worker_pool_;
258 SequenceToken running_sequence_;
259 WorkerShutdown running_shutdown_behavior_;
261 DISALLOW_COPY_AND_ASSIGN(Worker);
264 // Inner ----------------------------------------------------------------------
266 class SequencedWorkerPool::Inner {
267 public:
268 // Take a raw pointer to |worker| to avoid cycles (since we're owned
269 // by it).
270 Inner(SequencedWorkerPool* worker_pool, size_t max_threads,
271 const std::string& thread_name_prefix,
272 TestingObserver* observer);
274 ~Inner();
276 SequenceToken GetSequenceToken();
278 SequenceToken GetNamedSequenceToken(const std::string& name);
280 // This function accepts a name and an ID. If the name is null, the
281 // token ID is used. This allows us to implement the optional name lookup
282 // from a single function without having to enter the lock a separate time.
283 bool PostTask(const std::string* optional_token_name,
284 SequenceToken sequence_token,
285 WorkerShutdown shutdown_behavior,
286 const tracked_objects::Location& from_here,
287 const Closure& task,
288 TimeDelta delay);
290 bool RunsTasksOnCurrentThread() const;
292 bool IsRunningSequenceOnCurrentThread(SequenceToken sequence_token) const;
294 void CleanupForTesting();
296 void SignalHasWorkForTesting();
298 int GetWorkSignalCountForTesting() const;
300 void Shutdown(int max_blocking_tasks_after_shutdown);
302 bool IsShutdownInProgress();
304 // Runs the worker loop on the background thread.
305 void ThreadLoop(Worker* this_worker);
307 private:
308 enum GetWorkStatus {
309 GET_WORK_FOUND,
310 GET_WORK_NOT_FOUND,
311 GET_WORK_WAIT,
314 enum CleanupState {
315 CLEANUP_REQUESTED,
316 CLEANUP_STARTING,
317 CLEANUP_RUNNING,
318 CLEANUP_FINISHING,
319 CLEANUP_DONE,
322 // Called from within the lock, this converts the given token name into a
323 // token ID, creating a new one if necessary.
324 int LockedGetNamedTokenID(const std::string& name);
326 // Called from within the lock, this returns the next sequence task number.
327 int64 LockedGetNextSequenceTaskNumber();
329 // Called from within the lock, returns the shutdown behavior of the task
330 // running on the currently executing worker thread. If invoked from a thread
331 // that is not one of the workers, returns CONTINUE_ON_SHUTDOWN.
332 WorkerShutdown LockedCurrentThreadShutdownBehavior() const;
334 // Gets new task. There are 3 cases depending on the return value:
336 // 1) If the return value is |GET_WORK_FOUND|, |task| is filled in and should
337 // be run immediately.
338 // 2) If the return value is |GET_WORK_NOT_FOUND|, there are no tasks to run,
339 // and |task| is not filled in. In this case, the caller should wait until
340 // a task is posted.
341 // 3) If the return value is |GET_WORK_WAIT|, there are no tasks to run
342 // immediately, and |task| is not filled in. Likewise, |wait_time| is
343 // filled in the time to wait until the next task to run. In this case, the
344 // caller should wait the time.
346 // In any case, the calling code should clear the given
347 // delete_these_outside_lock vector the next time the lock is released.
348 // See the implementation for a more detailed description.
349 GetWorkStatus GetWork(SequencedTask* task,
350 TimeDelta* wait_time,
351 std::vector<Closure>* delete_these_outside_lock);
353 void HandleCleanup();
355 // Peforms init and cleanup around running the given task. WillRun...
356 // returns the value from PrepareToStartAdditionalThreadIfNecessary.
357 // The calling code should call FinishStartingAdditionalThread once the
358 // lock is released if the return values is nonzero.
359 int WillRunWorkerTask(const SequencedTask& task);
360 void DidRunWorkerTask(const SequencedTask& task);
362 // Returns true if there are no threads currently running the given
363 // sequence token.
364 bool IsSequenceTokenRunnable(int sequence_token_id) const;
366 // Checks if all threads are busy and the addition of one more could run an
367 // additional task waiting in the queue. This must be called from within
368 // the lock.
370 // If another thread is helpful, this will mark the thread as being in the
371 // process of starting and returns the index of the new thread which will be
372 // 0 or more. The caller should then call FinishStartingAdditionalThread to
373 // complete initialization once the lock is released.
375 // If another thread is not necessary, returne 0;
377 // See the implementedion for more.
378 int PrepareToStartAdditionalThreadIfHelpful();
380 // The second part of thread creation after
381 // PrepareToStartAdditionalThreadIfHelpful with the thread number it
382 // generated. This actually creates the thread and should be called outside
383 // the lock to avoid blocking important work starting a thread in the lock.
384 void FinishStartingAdditionalThread(int thread_number);
386 // Signal |has_work_| and increment |has_work_signal_count_|.
387 void SignalHasWork();
389 // Checks whether there is work left that's blocking shutdown. Must be
390 // called inside the lock.
391 bool CanShutdown() const;
393 SequencedWorkerPool* const worker_pool_;
395 // The last sequence number used. Managed by GetSequenceToken, since this
396 // only does threadsafe increment operations, you do not need to hold the
397 // lock. This is class-static to make SequenceTokens issued by
398 // GetSequenceToken unique across SequencedWorkerPool instances.
399 static base::StaticAtomicSequenceNumber g_last_sequence_number_;
401 // This lock protects |everything in this class|. Do not read or modify
402 // anything without holding this lock. Do not block while holding this
403 // lock.
404 mutable Lock lock_;
406 // Condition variable that is waited on by worker threads until new
407 // tasks are posted or shutdown starts.
408 ConditionVariable has_work_cv_;
410 // Condition variable that is waited on by non-worker threads (in
411 // Shutdown()) until CanShutdown() goes to true.
412 ConditionVariable can_shutdown_cv_;
414 // The maximum number of worker threads we'll create.
415 const size_t max_threads_;
417 const std::string thread_name_prefix_;
419 // Associates all known sequence token names with their IDs.
420 std::map<std::string, int> named_sequence_tokens_;
422 // Owning pointers to all threads we've created so far, indexed by
423 // ID. Since we lazily create threads, this may be less than
424 // max_threads_ and will be initially empty.
425 typedef std::map<PlatformThreadId, linked_ptr<Worker> > ThreadMap;
426 ThreadMap threads_;
428 // Set to true when we're in the process of creating another thread.
429 // See PrepareToStartAdditionalThreadIfHelpful for more.
430 bool thread_being_created_;
432 // Number of threads currently waiting for work.
433 size_t waiting_thread_count_;
435 // Number of threads currently running tasks that have the BLOCK_SHUTDOWN
436 // or SKIP_ON_SHUTDOWN flag set.
437 size_t blocking_shutdown_thread_count_;
439 // A set of all pending tasks in time-to-run order. These are tasks that are
440 // either waiting for a thread to run on, waiting for their time to run,
441 // or blocked on a previous task in their sequence. We have to iterate over
442 // the tasks by time-to-run order, so we use the set instead of the
443 // traditional priority_queue.
444 typedef std::set<SequencedTask, SequencedTaskLessThan> PendingTaskSet;
445 PendingTaskSet pending_tasks_;
447 // The next sequence number for a new sequenced task.
448 int64 next_sequence_task_number_;
450 // Number of tasks in the pending_tasks_ list that are marked as blocking
451 // shutdown.
452 size_t blocking_shutdown_pending_task_count_;
454 // Lists all sequence tokens currently executing.
455 std::set<int> current_sequences_;
457 // An ID for each posted task to distinguish the task from others in traces.
458 int trace_id_;
460 // Set when Shutdown is called and no further tasks should be
461 // allowed, though we may still be running existing tasks.
462 bool shutdown_called_;
464 // The number of new BLOCK_SHUTDOWN tasks that may be posted after Shudown()
465 // has been called.
466 int max_blocking_tasks_after_shutdown_;
468 // State used to cleanup for testing, all guarded by lock_.
469 CleanupState cleanup_state_;
470 size_t cleanup_idlers_;
471 ConditionVariable cleanup_cv_;
473 TestingObserver* const testing_observer_;
475 DISALLOW_COPY_AND_ASSIGN(Inner);
478 // Worker definitions ---------------------------------------------------------
480 SequencedWorkerPool::Worker::Worker(
481 const scoped_refptr<SequencedWorkerPool>& worker_pool,
482 int thread_number,
483 const std::string& prefix)
484 : SimpleThread(prefix + StringPrintf("Worker%d", thread_number)),
485 worker_pool_(worker_pool),
486 running_shutdown_behavior_(CONTINUE_ON_SHUTDOWN) {
487 Start();
490 SequencedWorkerPool::Worker::~Worker() {
493 void SequencedWorkerPool::Worker::Run() {
494 #if defined(OS_WIN)
495 win::ScopedCOMInitializer com_initializer;
496 #endif
498 // Store a pointer to the running sequence in thread local storage for
499 // static function access.
500 g_lazy_tls_ptr.Get().Set(&running_sequence_);
502 // Just jump back to the Inner object to run the thread, since it has all the
503 // tracking information and queues. It might be more natural to implement
504 // using DelegateSimpleThread and have Inner implement the Delegate to avoid
505 // having these worker objects at all, but that method lacks the ability to
506 // send thread-specific information easily to the thread loop.
507 worker_pool_->inner_->ThreadLoop(this);
508 // Release our cyclic reference once we're done.
509 worker_pool_ = NULL;
512 // Inner definitions ---------------------------------------------------------
514 SequencedWorkerPool::Inner::Inner(
515 SequencedWorkerPool* worker_pool,
516 size_t max_threads,
517 const std::string& thread_name_prefix,
518 TestingObserver* observer)
519 : worker_pool_(worker_pool),
520 lock_(),
521 has_work_cv_(&lock_),
522 can_shutdown_cv_(&lock_),
523 max_threads_(max_threads),
524 thread_name_prefix_(thread_name_prefix),
525 thread_being_created_(false),
526 waiting_thread_count_(0),
527 blocking_shutdown_thread_count_(0),
528 next_sequence_task_number_(0),
529 blocking_shutdown_pending_task_count_(0),
530 trace_id_(0),
531 shutdown_called_(false),
532 max_blocking_tasks_after_shutdown_(0),
533 cleanup_state_(CLEANUP_DONE),
534 cleanup_idlers_(0),
535 cleanup_cv_(&lock_),
536 testing_observer_(observer) {}
538 SequencedWorkerPool::Inner::~Inner() {
539 // You must call Shutdown() before destroying the pool.
540 DCHECK(shutdown_called_);
542 // Need to explicitly join with the threads before they're destroyed or else
543 // they will be running when our object is half torn down.
544 for (ThreadMap::iterator it = threads_.begin(); it != threads_.end(); ++it)
545 it->second->Join();
546 threads_.clear();
548 if (testing_observer_)
549 testing_observer_->OnDestruct();
552 SequencedWorkerPool::SequenceToken
553 SequencedWorkerPool::Inner::GetSequenceToken() {
554 // Need to add one because StaticAtomicSequenceNumber starts at zero, which
555 // is used as a sentinel value in SequenceTokens.
556 return SequenceToken(g_last_sequence_number_.GetNext() + 1);
559 SequencedWorkerPool::SequenceToken
560 SequencedWorkerPool::Inner::GetNamedSequenceToken(const std::string& name) {
561 AutoLock lock(lock_);
562 return SequenceToken(LockedGetNamedTokenID(name));
565 bool SequencedWorkerPool::Inner::PostTask(
566 const std::string* optional_token_name,
567 SequenceToken sequence_token,
568 WorkerShutdown shutdown_behavior,
569 const tracked_objects::Location& from_here,
570 const Closure& task,
571 TimeDelta delay) {
572 DCHECK(delay == TimeDelta() || shutdown_behavior == SKIP_ON_SHUTDOWN);
573 SequencedTask sequenced(from_here);
574 sequenced.sequence_token_id = sequence_token.id_;
575 sequenced.shutdown_behavior = shutdown_behavior;
576 sequenced.posted_from = from_here;
577 sequenced.task =
578 shutdown_behavior == BLOCK_SHUTDOWN ?
579 base::MakeCriticalClosure(task) : task;
580 sequenced.time_to_run = TimeTicks::Now() + delay;
582 int create_thread_id = 0;
584 AutoLock lock(lock_);
585 if (shutdown_called_) {
586 if (shutdown_behavior != BLOCK_SHUTDOWN ||
587 LockedCurrentThreadShutdownBehavior() == CONTINUE_ON_SHUTDOWN) {
588 return false;
590 if (max_blocking_tasks_after_shutdown_ <= 0) {
591 DLOG(WARNING) << "BLOCK_SHUTDOWN task disallowed";
592 return false;
594 max_blocking_tasks_after_shutdown_ -= 1;
597 // The trace_id is used for identifying the task in about:tracing.
598 sequenced.trace_id = trace_id_++;
600 TRACE_EVENT_FLOW_BEGIN0(TRACE_DISABLED_BY_DEFAULT("toplevel.flow"),
601 "SequencedWorkerPool::PostTask",
602 TRACE_ID_MANGLE(GetTaskTraceID(sequenced, static_cast<void*>(this))));
604 sequenced.sequence_task_number = LockedGetNextSequenceTaskNumber();
606 // Now that we have the lock, apply the named token rules.
607 if (optional_token_name)
608 sequenced.sequence_token_id = LockedGetNamedTokenID(*optional_token_name);
610 pending_tasks_.insert(sequenced);
611 if (shutdown_behavior == BLOCK_SHUTDOWN)
612 blocking_shutdown_pending_task_count_++;
614 create_thread_id = PrepareToStartAdditionalThreadIfHelpful();
617 // Actually start the additional thread or signal an existing one now that
618 // we're outside the lock.
619 if (create_thread_id)
620 FinishStartingAdditionalThread(create_thread_id);
621 else
622 SignalHasWork();
624 return true;
627 bool SequencedWorkerPool::Inner::RunsTasksOnCurrentThread() const {
628 AutoLock lock(lock_);
629 return ContainsKey(threads_, PlatformThread::CurrentId());
632 bool SequencedWorkerPool::Inner::IsRunningSequenceOnCurrentThread(
633 SequenceToken sequence_token) const {
634 AutoLock lock(lock_);
635 ThreadMap::const_iterator found = threads_.find(PlatformThread::CurrentId());
636 if (found == threads_.end())
637 return false;
638 return sequence_token.Equals(found->second->running_sequence());
641 // See https://code.google.com/p/chromium/issues/detail?id=168415
642 void SequencedWorkerPool::Inner::CleanupForTesting() {
643 DCHECK(!RunsTasksOnCurrentThread());
644 base::ThreadRestrictions::ScopedAllowWait allow_wait;
645 AutoLock lock(lock_);
646 CHECK_EQ(CLEANUP_DONE, cleanup_state_);
647 if (shutdown_called_)
648 return;
649 if (pending_tasks_.empty() && waiting_thread_count_ == threads_.size())
650 return;
651 cleanup_state_ = CLEANUP_REQUESTED;
652 cleanup_idlers_ = 0;
653 has_work_cv_.Signal();
654 while (cleanup_state_ != CLEANUP_DONE)
655 cleanup_cv_.Wait();
658 void SequencedWorkerPool::Inner::SignalHasWorkForTesting() {
659 SignalHasWork();
662 void SequencedWorkerPool::Inner::Shutdown(
663 int max_new_blocking_tasks_after_shutdown) {
664 DCHECK_GE(max_new_blocking_tasks_after_shutdown, 0);
666 AutoLock lock(lock_);
667 // Cleanup and Shutdown should not be called concurrently.
668 CHECK_EQ(CLEANUP_DONE, cleanup_state_);
669 if (shutdown_called_)
670 return;
671 shutdown_called_ = true;
672 max_blocking_tasks_after_shutdown_ = max_new_blocking_tasks_after_shutdown;
674 // Tickle the threads. This will wake up a waiting one so it will know that
675 // it can exit, which in turn will wake up any other waiting ones.
676 SignalHasWork();
678 // There are no pending or running tasks blocking shutdown, we're done.
679 if (CanShutdown())
680 return;
683 // If we're here, then something is blocking shutdown. So wait for
684 // CanShutdown() to go to true.
686 if (testing_observer_)
687 testing_observer_->WillWaitForShutdown();
689 #if !defined(OS_NACL)
690 TimeTicks shutdown_wait_begin = TimeTicks::Now();
691 #endif
694 base::ThreadRestrictions::ScopedAllowWait allow_wait;
695 AutoLock lock(lock_);
696 while (!CanShutdown())
697 can_shutdown_cv_.Wait();
699 #if !defined(OS_NACL)
700 UMA_HISTOGRAM_TIMES("SequencedWorkerPool.ShutdownDelayTime",
701 TimeTicks::Now() - shutdown_wait_begin);
702 #endif
705 bool SequencedWorkerPool::Inner::IsShutdownInProgress() {
706 AutoLock lock(lock_);
707 return shutdown_called_;
710 void SequencedWorkerPool::Inner::ThreadLoop(Worker* this_worker) {
712 AutoLock lock(lock_);
713 DCHECK(thread_being_created_);
714 thread_being_created_ = false;
715 std::pair<ThreadMap::iterator, bool> result =
716 threads_.insert(
717 std::make_pair(this_worker->tid(), make_linked_ptr(this_worker)));
718 DCHECK(result.second);
720 while (true) {
721 #if defined(OS_MACOSX)
722 base::mac::ScopedNSAutoreleasePool autorelease_pool;
723 #endif
725 HandleCleanup();
727 // See GetWork for what delete_these_outside_lock is doing.
728 SequencedTask task;
729 TimeDelta wait_time;
730 std::vector<Closure> delete_these_outside_lock;
731 GetWorkStatus status =
732 GetWork(&task, &wait_time, &delete_these_outside_lock);
733 if (status == GET_WORK_FOUND) {
734 TRACE_EVENT_FLOW_END0(TRACE_DISABLED_BY_DEFAULT("toplevel.flow"),
735 "SequencedWorkerPool::PostTask",
736 TRACE_ID_MANGLE(GetTaskTraceID(task, static_cast<void*>(this))));
737 TRACE_EVENT2("toplevel", "SequencedWorkerPool::ThreadLoop",
738 "src_file", task.posted_from.file_name(),
739 "src_func", task.posted_from.function_name());
740 int new_thread_id = WillRunWorkerTask(task);
742 AutoUnlock unlock(lock_);
743 // There may be more work available, so wake up another
744 // worker thread. (Technically not required, since we
745 // already get a signal for each new task, but it doesn't
746 // hurt.)
747 SignalHasWork();
748 delete_these_outside_lock.clear();
750 // Complete thread creation outside the lock if necessary.
751 if (new_thread_id)
752 FinishStartingAdditionalThread(new_thread_id);
754 this_worker->set_running_task_info(
755 SequenceToken(task.sequence_token_id), task.shutdown_behavior);
757 tracked_objects::ThreadData::PrepareForStartOfRun(task.birth_tally);
758 tracked_objects::TaskStopwatch stopwatch;
759 stopwatch.Start();
760 task.task.Run();
761 stopwatch.Stop();
763 tracked_objects::ThreadData::TallyRunOnNamedThreadIfTracking(
764 task, stopwatch);
766 // Make sure our task is erased outside the lock for the
767 // same reason we do this with delete_these_oustide_lock.
768 // Also, do it before calling set_running_task_info() so
769 // that sequence-checking from within the task's destructor
770 // still works.
771 task.task = Closure();
773 this_worker->set_running_task_info(
774 SequenceToken(), CONTINUE_ON_SHUTDOWN);
776 DidRunWorkerTask(task); // Must be done inside the lock.
777 } else if (cleanup_state_ == CLEANUP_RUNNING) {
778 switch (status) {
779 case GET_WORK_WAIT: {
780 AutoUnlock unlock(lock_);
781 delete_these_outside_lock.clear();
783 break;
784 case GET_WORK_NOT_FOUND:
785 CHECK(delete_these_outside_lock.empty());
786 cleanup_state_ = CLEANUP_FINISHING;
787 cleanup_cv_.Broadcast();
788 break;
789 default:
790 NOTREACHED();
792 } else {
793 // When we're terminating and there's no more work, we can
794 // shut down, other workers can complete any pending or new tasks.
795 // We can get additional tasks posted after shutdown_called_ is set
796 // but only worker threads are allowed to post tasks at that time, and
797 // the workers responsible for posting those tasks will be available
798 // to run them. Also, there may be some tasks stuck behind running
799 // ones with the same sequence token, but additional threads won't
800 // help this case.
801 if (shutdown_called_ &&
802 blocking_shutdown_pending_task_count_ == 0)
803 break;
804 waiting_thread_count_++;
806 switch (status) {
807 case GET_WORK_NOT_FOUND:
808 has_work_cv_.Wait();
809 break;
810 case GET_WORK_WAIT:
811 has_work_cv_.TimedWait(wait_time);
812 break;
813 default:
814 NOTREACHED();
816 waiting_thread_count_--;
819 } // Release lock_.
821 // We noticed we should exit. Wake up the next worker so it knows it should
822 // exit as well (because the Shutdown() code only signals once).
823 SignalHasWork();
825 // Possibly unblock shutdown.
826 can_shutdown_cv_.Signal();
829 void SequencedWorkerPool::Inner::HandleCleanup() {
830 lock_.AssertAcquired();
831 if (cleanup_state_ == CLEANUP_DONE)
832 return;
833 if (cleanup_state_ == CLEANUP_REQUESTED) {
834 // We win, we get to do the cleanup as soon as the others wise up and idle.
835 cleanup_state_ = CLEANUP_STARTING;
836 while (thread_being_created_ ||
837 cleanup_idlers_ != threads_.size() - 1) {
838 has_work_cv_.Signal();
839 cleanup_cv_.Wait();
841 cleanup_state_ = CLEANUP_RUNNING;
842 return;
844 if (cleanup_state_ == CLEANUP_STARTING) {
845 // Another worker thread is cleaning up, we idle here until thats done.
846 ++cleanup_idlers_;
847 cleanup_cv_.Broadcast();
848 while (cleanup_state_ != CLEANUP_FINISHING) {
849 cleanup_cv_.Wait();
851 --cleanup_idlers_;
852 cleanup_cv_.Broadcast();
853 return;
855 if (cleanup_state_ == CLEANUP_FINISHING) {
856 // We wait for all idlers to wake up prior to being DONE.
857 while (cleanup_idlers_ != 0) {
858 cleanup_cv_.Broadcast();
859 cleanup_cv_.Wait();
861 if (cleanup_state_ == CLEANUP_FINISHING) {
862 cleanup_state_ = CLEANUP_DONE;
863 cleanup_cv_.Signal();
865 return;
869 int SequencedWorkerPool::Inner::LockedGetNamedTokenID(
870 const std::string& name) {
871 lock_.AssertAcquired();
872 DCHECK(!name.empty());
874 std::map<std::string, int>::const_iterator found =
875 named_sequence_tokens_.find(name);
876 if (found != named_sequence_tokens_.end())
877 return found->second; // Got an existing one.
879 // Create a new one for this name.
880 SequenceToken result = GetSequenceToken();
881 named_sequence_tokens_.insert(std::make_pair(name, result.id_));
882 return result.id_;
885 int64 SequencedWorkerPool::Inner::LockedGetNextSequenceTaskNumber() {
886 lock_.AssertAcquired();
887 // We assume that we never create enough tasks to wrap around.
888 return next_sequence_task_number_++;
891 SequencedWorkerPool::WorkerShutdown
892 SequencedWorkerPool::Inner::LockedCurrentThreadShutdownBehavior() const {
893 lock_.AssertAcquired();
894 ThreadMap::const_iterator found = threads_.find(PlatformThread::CurrentId());
895 if (found == threads_.end())
896 return CONTINUE_ON_SHUTDOWN;
897 return found->second->running_shutdown_behavior();
900 SequencedWorkerPool::Inner::GetWorkStatus SequencedWorkerPool::Inner::GetWork(
901 SequencedTask* task,
902 TimeDelta* wait_time,
903 std::vector<Closure>* delete_these_outside_lock) {
904 lock_.AssertAcquired();
906 // Find the next task with a sequence token that's not currently in use.
907 // If the token is in use, that means another thread is running something
908 // in that sequence, and we can't run it without going out-of-order.
910 // This algorithm is simple and fair, but inefficient in some cases. For
911 // example, say somebody schedules 1000 slow tasks with the same sequence
912 // number. We'll have to go through all those tasks each time we feel like
913 // there might be work to schedule. If this proves to be a problem, we
914 // should make this more efficient.
916 // One possible enhancement would be to keep a map from sequence ID to a
917 // list of pending but currently blocked SequencedTasks for that ID.
918 // When a worker finishes a task of one sequence token, it can pick up the
919 // next one from that token right away.
921 // This may lead to starvation if there are sufficient numbers of sequences
922 // in use. To alleviate this, we could add an incrementing priority counter
923 // to each SequencedTask. Then maintain a priority_queue of all runnable
924 // tasks, sorted by priority counter. When a sequenced task is completed
925 // we would pop the head element off of that tasks pending list and add it
926 // to the priority queue. Then we would run the first item in the priority
927 // queue.
929 GetWorkStatus status = GET_WORK_NOT_FOUND;
930 int unrunnable_tasks = 0;
931 PendingTaskSet::iterator i = pending_tasks_.begin();
932 // We assume that the loop below doesn't take too long and so we can just do
933 // a single call to TimeTicks::Now().
934 const TimeTicks current_time = TimeTicks::Now();
935 while (i != pending_tasks_.end()) {
936 if (!IsSequenceTokenRunnable(i->sequence_token_id)) {
937 unrunnable_tasks++;
938 ++i;
939 continue;
942 if (shutdown_called_ && i->shutdown_behavior != BLOCK_SHUTDOWN) {
943 // We're shutting down and the task we just found isn't blocking
944 // shutdown. Delete it and get more work.
946 // Note that we do not want to delete unrunnable tasks. Deleting a task
947 // can have side effects (like freeing some objects) and deleting a
948 // task that's supposed to run after one that's currently running could
949 // cause an obscure crash.
951 // We really want to delete these tasks outside the lock in case the
952 // closures are holding refs to objects that want to post work from
953 // their destructorss (which would deadlock). The closures are
954 // internally refcounted, so we just need to keep a copy of them alive
955 // until the lock is exited. The calling code can just clear() the
956 // vector they passed to us once the lock is exited to make this
957 // happen.
958 delete_these_outside_lock->push_back(i->task);
959 pending_tasks_.erase(i++);
960 continue;
963 if (i->time_to_run > current_time) {
964 // The time to run has not come yet.
965 *wait_time = i->time_to_run - current_time;
966 status = GET_WORK_WAIT;
967 if (cleanup_state_ == CLEANUP_RUNNING) {
968 // Deferred tasks are deleted when cleaning up, see Inner::ThreadLoop.
969 delete_these_outside_lock->push_back(i->task);
970 pending_tasks_.erase(i);
972 break;
975 // Found a runnable task.
976 *task = *i;
977 pending_tasks_.erase(i);
978 if (task->shutdown_behavior == BLOCK_SHUTDOWN) {
979 blocking_shutdown_pending_task_count_--;
982 status = GET_WORK_FOUND;
983 break;
986 return status;
989 int SequencedWorkerPool::Inner::WillRunWorkerTask(const SequencedTask& task) {
990 lock_.AssertAcquired();
992 // Mark the task's sequence number as in use.
993 if (task.sequence_token_id)
994 current_sequences_.insert(task.sequence_token_id);
996 // Ensure that threads running tasks posted with either SKIP_ON_SHUTDOWN
997 // or BLOCK_SHUTDOWN will prevent shutdown until that task or thread
998 // completes.
999 if (task.shutdown_behavior != CONTINUE_ON_SHUTDOWN)
1000 blocking_shutdown_thread_count_++;
1002 // We just picked up a task. Since StartAdditionalThreadIfHelpful only
1003 // creates a new thread if there is no free one, there is a race when posting
1004 // tasks that many tasks could have been posted before a thread started
1005 // running them, so only one thread would have been created. So we also check
1006 // whether we should create more threads after removing our task from the
1007 // queue, which also has the nice side effect of creating the workers from
1008 // background threads rather than the main thread of the app.
1010 // If another thread wasn't created, we want to wake up an existing thread
1011 // if there is one waiting to pick up the next task.
1013 // Note that we really need to do this *before* running the task, not
1014 // after. Otherwise, if more than one task is posted, the creation of the
1015 // second thread (since we only create one at a time) will be blocked by
1016 // the execution of the first task, which could be arbitrarily long.
1017 return PrepareToStartAdditionalThreadIfHelpful();
1020 void SequencedWorkerPool::Inner::DidRunWorkerTask(const SequencedTask& task) {
1021 lock_.AssertAcquired();
1023 if (task.shutdown_behavior != CONTINUE_ON_SHUTDOWN) {
1024 DCHECK_GT(blocking_shutdown_thread_count_, 0u);
1025 blocking_shutdown_thread_count_--;
1028 if (task.sequence_token_id)
1029 current_sequences_.erase(task.sequence_token_id);
1032 bool SequencedWorkerPool::Inner::IsSequenceTokenRunnable(
1033 int sequence_token_id) const {
1034 lock_.AssertAcquired();
1035 return !sequence_token_id ||
1036 current_sequences_.find(sequence_token_id) ==
1037 current_sequences_.end();
1040 int SequencedWorkerPool::Inner::PrepareToStartAdditionalThreadIfHelpful() {
1041 lock_.AssertAcquired();
1042 // How thread creation works:
1044 // We'de like to avoid creating threads with the lock held. However, we
1045 // need to be sure that we have an accurate accounting of the threads for
1046 // proper Joining and deltion on shutdown.
1048 // We need to figure out if we need another thread with the lock held, which
1049 // is what this function does. It then marks us as in the process of creating
1050 // a thread. When we do shutdown, we wait until the thread_being_created_
1051 // flag is cleared, which ensures that the new thread is properly added to
1052 // all the data structures and we can't leak it. Once shutdown starts, we'll
1053 // refuse to create more threads or they would be leaked.
1055 // Note that this creates a mostly benign race condition on shutdown that
1056 // will cause fewer workers to be created than one would expect. It isn't
1057 // much of an issue in real life, but affects some tests. Since we only spawn
1058 // one worker at a time, the following sequence of events can happen:
1060 // 1. Main thread posts a bunch of unrelated tasks that would normally be
1061 // run on separate threads.
1062 // 2. The first task post causes us to start a worker. Other tasks do not
1063 // cause a worker to start since one is pending.
1064 // 3. Main thread initiates shutdown.
1065 // 4. No more threads are created since the shutdown_called_ flag is set.
1067 // The result is that one may expect that max_threads_ workers to be created
1068 // given the workload, but in reality fewer may be created because the
1069 // sequence of thread creation on the background threads is racing with the
1070 // shutdown call.
1071 if (!shutdown_called_ &&
1072 !thread_being_created_ &&
1073 cleanup_state_ == CLEANUP_DONE &&
1074 threads_.size() < max_threads_ &&
1075 waiting_thread_count_ == 0) {
1076 // We could use an additional thread if there's work to be done.
1077 for (PendingTaskSet::const_iterator i = pending_tasks_.begin();
1078 i != pending_tasks_.end(); ++i) {
1079 if (IsSequenceTokenRunnable(i->sequence_token_id)) {
1080 // Found a runnable task, mark the thread as being started.
1081 thread_being_created_ = true;
1082 return static_cast<int>(threads_.size() + 1);
1086 return 0;
1089 void SequencedWorkerPool::Inner::FinishStartingAdditionalThread(
1090 int thread_number) {
1091 // Called outside of the lock.
1092 DCHECK(thread_number > 0);
1094 // The worker is assigned to the list when the thread actually starts, which
1095 // will manage the memory of the pointer.
1096 new Worker(worker_pool_, thread_number, thread_name_prefix_);
1099 void SequencedWorkerPool::Inner::SignalHasWork() {
1100 has_work_cv_.Signal();
1101 if (testing_observer_) {
1102 testing_observer_->OnHasWork();
1106 bool SequencedWorkerPool::Inner::CanShutdown() const {
1107 lock_.AssertAcquired();
1108 // See PrepareToStartAdditionalThreadIfHelpful for how thread creation works.
1109 return !thread_being_created_ &&
1110 blocking_shutdown_thread_count_ == 0 &&
1111 blocking_shutdown_pending_task_count_ == 0;
1114 base::StaticAtomicSequenceNumber
1115 SequencedWorkerPool::Inner::g_last_sequence_number_;
1117 // SequencedWorkerPool --------------------------------------------------------
1119 // static
1120 SequencedWorkerPool::SequenceToken
1121 SequencedWorkerPool::GetSequenceTokenForCurrentThread() {
1122 // Don't construct lazy instance on check.
1123 if (g_lazy_tls_ptr == NULL)
1124 return SequenceToken();
1126 SequencedWorkerPool::SequenceToken* token = g_lazy_tls_ptr.Get().Get();
1127 if (!token)
1128 return SequenceToken();
1129 return *token;
1132 SequencedWorkerPool::SequencedWorkerPool(
1133 size_t max_threads,
1134 const std::string& thread_name_prefix)
1135 : constructor_message_loop_(MessageLoopProxy::current()),
1136 inner_(new Inner(this, max_threads, thread_name_prefix, NULL)) {
1139 SequencedWorkerPool::SequencedWorkerPool(
1140 size_t max_threads,
1141 const std::string& thread_name_prefix,
1142 TestingObserver* observer)
1143 : constructor_message_loop_(MessageLoopProxy::current()),
1144 inner_(new Inner(this, max_threads, thread_name_prefix, observer)) {
1147 SequencedWorkerPool::~SequencedWorkerPool() {}
1149 void SequencedWorkerPool::OnDestruct() const {
1150 DCHECK(constructor_message_loop_.get());
1151 // Avoid deleting ourselves on a worker thread (which would
1152 // deadlock).
1153 if (RunsTasksOnCurrentThread()) {
1154 constructor_message_loop_->DeleteSoon(FROM_HERE, this);
1155 } else {
1156 delete this;
1160 SequencedWorkerPool::SequenceToken SequencedWorkerPool::GetSequenceToken() {
1161 return inner_->GetSequenceToken();
1164 SequencedWorkerPool::SequenceToken SequencedWorkerPool::GetNamedSequenceToken(
1165 const std::string& name) {
1166 return inner_->GetNamedSequenceToken(name);
1169 scoped_refptr<SequencedTaskRunner> SequencedWorkerPool::GetSequencedTaskRunner(
1170 SequenceToken token) {
1171 return GetSequencedTaskRunnerWithShutdownBehavior(token, BLOCK_SHUTDOWN);
1174 scoped_refptr<SequencedTaskRunner>
1175 SequencedWorkerPool::GetSequencedTaskRunnerWithShutdownBehavior(
1176 SequenceToken token, WorkerShutdown shutdown_behavior) {
1177 return new SequencedWorkerPoolSequencedTaskRunner(
1178 this, token, shutdown_behavior);
1181 scoped_refptr<TaskRunner>
1182 SequencedWorkerPool::GetTaskRunnerWithShutdownBehavior(
1183 WorkerShutdown shutdown_behavior) {
1184 return new SequencedWorkerPoolTaskRunner(this, shutdown_behavior);
1187 bool SequencedWorkerPool::PostWorkerTask(
1188 const tracked_objects::Location& from_here,
1189 const Closure& task) {
1190 return inner_->PostTask(NULL, SequenceToken(), BLOCK_SHUTDOWN,
1191 from_here, task, TimeDelta());
1194 bool SequencedWorkerPool::PostDelayedWorkerTask(
1195 const tracked_objects::Location& from_here,
1196 const Closure& task,
1197 TimeDelta delay) {
1198 WorkerShutdown shutdown_behavior =
1199 delay == TimeDelta() ? BLOCK_SHUTDOWN : SKIP_ON_SHUTDOWN;
1200 return inner_->PostTask(NULL, SequenceToken(), shutdown_behavior,
1201 from_here, task, delay);
1204 bool SequencedWorkerPool::PostWorkerTaskWithShutdownBehavior(
1205 const tracked_objects::Location& from_here,
1206 const Closure& task,
1207 WorkerShutdown shutdown_behavior) {
1208 return inner_->PostTask(NULL, SequenceToken(), shutdown_behavior,
1209 from_here, task, TimeDelta());
1212 bool SequencedWorkerPool::PostSequencedWorkerTask(
1213 SequenceToken sequence_token,
1214 const tracked_objects::Location& from_here,
1215 const Closure& task) {
1216 return inner_->PostTask(NULL, sequence_token, BLOCK_SHUTDOWN,
1217 from_here, task, TimeDelta());
1220 bool SequencedWorkerPool::PostDelayedSequencedWorkerTask(
1221 SequenceToken sequence_token,
1222 const tracked_objects::Location& from_here,
1223 const Closure& task,
1224 TimeDelta delay) {
1225 WorkerShutdown shutdown_behavior =
1226 delay == TimeDelta() ? BLOCK_SHUTDOWN : SKIP_ON_SHUTDOWN;
1227 return inner_->PostTask(NULL, sequence_token, shutdown_behavior,
1228 from_here, task, delay);
1231 bool SequencedWorkerPool::PostNamedSequencedWorkerTask(
1232 const std::string& token_name,
1233 const tracked_objects::Location& from_here,
1234 const Closure& task) {
1235 DCHECK(!token_name.empty());
1236 return inner_->PostTask(&token_name, SequenceToken(), BLOCK_SHUTDOWN,
1237 from_here, task, TimeDelta());
1240 bool SequencedWorkerPool::PostSequencedWorkerTaskWithShutdownBehavior(
1241 SequenceToken sequence_token,
1242 const tracked_objects::Location& from_here,
1243 const Closure& task,
1244 WorkerShutdown shutdown_behavior) {
1245 return inner_->PostTask(NULL, sequence_token, shutdown_behavior,
1246 from_here, task, TimeDelta());
1249 bool SequencedWorkerPool::PostDelayedTask(
1250 const tracked_objects::Location& from_here,
1251 const Closure& task,
1252 TimeDelta delay) {
1253 return PostDelayedWorkerTask(from_here, task, delay);
1256 bool SequencedWorkerPool::RunsTasksOnCurrentThread() const {
1257 return inner_->RunsTasksOnCurrentThread();
1260 bool SequencedWorkerPool::IsRunningSequenceOnCurrentThread(
1261 SequenceToken sequence_token) const {
1262 return inner_->IsRunningSequenceOnCurrentThread(sequence_token);
1265 void SequencedWorkerPool::FlushForTesting() {
1266 inner_->CleanupForTesting();
1269 void SequencedWorkerPool::SignalHasWorkForTesting() {
1270 inner_->SignalHasWorkForTesting();
1273 void SequencedWorkerPool::Shutdown(int max_new_blocking_tasks_after_shutdown) {
1274 DCHECK(constructor_message_loop_->BelongsToCurrentThread());
1275 inner_->Shutdown(max_new_blocking_tasks_after_shutdown);
1278 bool SequencedWorkerPool::IsShutdownInProgress() {
1279 return inner_->IsShutdownInProgress();
1282 } // namespace base