| 469 | } |
| 470 | |
| 471 | void RunHandlerThreadPool::WaitForWork(bool is_blocking, int thread_id, |
| 472 | int32 max_blocking_inflight) { |
| 473 | const int kMaxSleepMicros = 250; |
| 474 | |
| 475 | // The non-blocking thread will just sleep. |
| 476 | if (!is_blocking) { |
| 477 | Env::Default()->SleepForMicroseconds(kMaxSleepMicros); |
| 478 | return; |
| 479 | } |
| 480 | |
| 481 | ThreadWorkSource* tws = nullptr; |
| 482 | { |
| 483 | Eigen::MaxSizeVector<ThreadWorkSource*>* thread_work_sources = |
| 484 | &thread_data_[thread_id].thread_work_sources; |
| 485 | mutex_lock l(thread_data_[thread_id].mu); |
| 486 | while (!cancelled_ && thread_work_sources->empty()) { |
| 487 | // Wait until there is new request |
| 488 | thread_data_[thread_id].sources_not_empty.wait(l); |
| 489 | } |
| 490 | if (cancelled_) { |
| 491 | return; |
| 492 | } |
| 493 | tws = (*thread_work_sources)[0]; |
| 494 | } |
| 495 | |
| 496 | if (tws->GetInflightTaskCount(true) >= max_blocking_inflight) { |
| 497 | // Sleep to reduce contention in PropagateOutputs |
| 498 | Env::Default()->SleepForMicroseconds(kMaxSleepMicros); |
| 499 | } |
| 500 | tws->WaitForWork(kMaxSleepMicros); |
| 501 | } |
| 502 | |
| 503 | } // namespace |
| 504 |
nothing calls this directly
no test coverage detected