| 681 | } |
| 682 | |
| 683 | void ThreadPool::DispatchThread() { |
| 684 | MutexLock unique_lock(lock_); |
| 685 | InsertOrDie(&threads_, Thread::current_thread()); |
| 686 | DCHECK_GT(num_threads_pending_start_, 0); |
| 687 | num_threads_++; |
| 688 | num_threads_pending_start_--; |
| 689 | // If we are one of the first 'min_threads_' to start, we must be |
| 690 | // a "permanent" thread. |
| 691 | bool permanent = num_threads_ <= min_threads_; |
| 692 | |
| 693 | // Owned by this worker thread and added/removed from idle_threads_ as needed. |
| 694 | IdleThread me(&lock_); |
| 695 | |
| 696 | while (true) { |
| 697 | // Note: Status::Aborted() is used to indicate normal shutdown. |
| 698 | if (!pool_status_.ok()) { |
| 699 | VLOG(2) << "DispatchThread exiting: " << pool_status_.ToString(); |
| 700 | break; |
| 701 | } |
| 702 | |
| 703 | if (queue_.empty()) { |
| 704 | // There's no work to do, let's go idle. |
| 705 | // |
| 706 | // Note: if FIFO behavior is desired, it's as simple as changing this to push_back(). |
| 707 | idle_threads_.push_front(me); |
| 708 | NotifyLoadMeterUnlocked(); |
| 709 | SCOPED_CLEANUP({ |
| 710 | // For some wake ups (i.e. Shutdown or DoSubmit) this thread is |
| 711 | // guaranteed to be unlinked after being awakened. In others (i.e. |
| 712 | // spurious wake-up or Wait timeout), it'll still be linked. |
| 713 | if (me.is_linked()) { |
| 714 | idle_threads_.erase(idle_threads_.iterator_to(me)); |
| 715 | } |
| 716 | }); |
| 717 | if (permanent) { |
| 718 | me.not_empty.Wait(); |
| 719 | } else { |
| 720 | if (!me.not_empty.WaitFor(idle_timeout_)) { |
| 721 | // After much investigation, it appears that pthread condition variables have |
| 722 | // a weird behavior in which they can return ETIMEDOUT from timed_wait even if |
| 723 | // another thread did in fact signal. Apparently after a timeout there is some |
| 724 | // brief period during which another thread may actually grab the internal mutex |
| 725 | // protecting the state, signal, and release again before we get the mutex. So, |
| 726 | // we'll recheck the empty queue case regardless. |
| 727 | if (queue_.empty()) { |
| 728 | VLOG(3) << "Releasing worker thread from pool " << name_ << " after " |
| 729 | << idle_timeout_.ToMilliseconds() << "ms of idle time."; |
| 730 | break; |
| 731 | } |
| 732 | } |
| 733 | } |
| 734 | continue; |
| 735 | } |
| 736 | |
| 737 | // Get the next token and task to execute. |
| 738 | ThreadPoolToken* token = queue_.front(); |
| 739 | queue_.pop_front(); |
| 740 | DCHECK_EQ(ThreadPoolToken::State::RUNNING, token->state()); |
no test coverage detected