| 37 | } |
| 38 | |
| 39 | utils::TaskRescheduleInfo EventDrivenSchedulingAgent::run(const std::shared_ptr<core::Processor> &processor, const std::shared_ptr<core::ProcessContext> &processContext, |
| 40 | const std::shared_ptr<core::ProcessSessionFactory> &sessionFactory) { |
| 41 | if (this->running_) { |
| 42 | auto start_time = std::chrono::steady_clock::now(); |
| 43 | // trigger processor until it has work to do, but no more than half a sec |
| 44 | while (processor->isRunning() && (std::chrono::steady_clock::now() - start_time < time_slice_)) { |
| 45 | bool shouldYield = this->onTrigger(processor, processContext, sessionFactory); |
| 46 | if (processor->isYield()) { |
| 47 | // Honor the yield |
| 48 | return utils::TaskRescheduleInfo::RetryIn(std::chrono::milliseconds(processor->getYieldTime())); |
| 49 | } else if (shouldYield) { |
| 50 | // No work to do or need to apply back pressure |
| 51 | return utils::TaskRescheduleInfo::RetryIn( |
| 52 | std::chrono::milliseconds((this->bored_yield_duration_ > 0) ? this->bored_yield_duration_ : 10)); // No work left to do, stand by |
| 53 | } |
| 54 | } |
| 55 | return utils::TaskRescheduleInfo::RetryImmediately(); // Let's continue work as soon as a thread is available |
| 56 | } |
| 57 | return utils::TaskRescheduleInfo::Done(); |
| 58 | } |
| 59 | |
| 60 | } // namespace minifi |
| 61 | } // namespace nifi |
nothing calls this directly
no test coverage detected