| 536 | } |
| 537 | |
| 538 | Status ThreadPool::DoSubmit(std::function<void()> f, ThreadPoolToken* token) { |
| 539 | DCHECK(token); |
| 540 | const MonoTime submit_time = MonoTime::Now(); |
| 541 | |
| 542 | MutexLock guard(lock_); |
| 543 | if (PREDICT_FALSE(!pool_status_.ok())) { |
| 544 | return pool_status_; |
| 545 | } |
| 546 | |
| 547 | if (PREDICT_FALSE(!token->MaySubmitNewTasks())) { |
| 548 | return Status::ServiceUnavailable("Thread pool token was shut down"); |
| 549 | } |
| 550 | |
| 551 | // Size limit check. |
| 552 | int64_t capacity_remaining = static_cast<int64_t>(max_threads_) - active_threads_ + |
| 553 | static_cast<int64_t>(max_queue_size_) - total_queued_tasks_; |
| 554 | if (capacity_remaining < 1) { |
| 555 | return Status::ServiceUnavailable( |
| 556 | Substitute("Thread pool is at capacity ($0/$1 tasks running, $2/$3 tasks queued)", |
| 557 | num_threads_ + num_threads_pending_start_, max_threads_, |
| 558 | total_queued_tasks_, max_queue_size_)); |
| 559 | } |
| 560 | |
| 561 | // Should we create another thread? |
| 562 | |
| 563 | // We assume that each current inactive thread will grab one item from the |
| 564 | // queue. If it seems like we'll need another thread, we create one. |
| 565 | // |
| 566 | // Rather than creating the thread here, while holding the lock, we defer |
| 567 | // it to down below. This is because thread creation can be rather slow |
| 568 | // (hundreds of milliseconds in some cases) and we'd like to allow the |
| 569 | // existing threads to continue to process tasks while we do so. |
| 570 | // |
| 571 | // In theory, a currently active thread could finish immediately after this |
| 572 | // calculation but before our new worker starts running. This would mean we |
| 573 | // created a thread we didn't really need. However, this race is unavoidable |
| 574 | // and harmless. |
| 575 | // |
| 576 | // Of course, we never create more than max_threads_ threads no matter what. |
| 577 | int threads_from_this_submit = |
| 578 | token->IsActive() && token->mode() == ExecutionMode::SERIAL ? 0 : 1; |
| 579 | int inactive_threads = num_threads_ + num_threads_pending_start_ - active_threads_; |
| 580 | int additional_threads = static_cast<int>(queue_.size()) |
| 581 | + threads_from_this_submit |
| 582 | - inactive_threads; |
| 583 | bool need_a_thread = false; |
| 584 | if (additional_threads > 0 && num_threads_ + num_threads_pending_start_ < max_threads_) { |
| 585 | need_a_thread = true; |
| 586 | num_threads_pending_start_++; |
| 587 | } |
| 588 | |
| 589 | Task task; |
| 590 | task.func = std::move(f); |
| 591 | task.trace = Trace::CurrentTrace(); |
| 592 | // Need to AddRef, since the thread which submitted the task may go away, |
| 593 | // and we don't want the trace to be destructed while waiting in the queue. |
| 594 | if (task.trace) { |
| 595 | task.trace->AddRef(); |
no test coverage detected