Method ScheduleLoop

oneflow/core/vm/virtual_machine.cpp:330–368

}  // namespace

void VirtualMachine::ScheduleLoop(const std::function<void()>& Initializer) {
  SyncVmModeGuard guard(SyncVmMode::kEnable);
  Initializer();
  MultiThreadScheduleCtx schedule_ctx{};
  while (pending_notifier_.WaitAndClearNotifiedCnt() == kNotifierStatusSuccess) {
    OF_PROFILER_RANGE_GUARD("VirtualMachine::ScheduleLoop");
    auto start = std::chrono::steady_clock::now();
    static constexpr int kWorkingMicroseconds = 1000;
    // Every time this thread wakes up, engine_ is scheduled for about `kWorkingMicroseconds`.
    // The cost of os thread switching is about 5-10 microseconds. Doing more scheduling in
    // a single waiting up can reach higher performance.
    do {
      // Use SchedulerThreadUnsafeEmpty to avoid acquiring mutex lock.
      // It's safe to use SchedulerThreadUnsafeEmpty here. pending_notifier_.notified_cnt_ will be
      // greater than zero when inconsistency between
      // engine_->pending_instruction_list.list_head_.list_head_.container_ and
      // engine_->pending_instruction_list.list_head_.list_head_.size_ occured. hence the pending
      // instructions
      // will get handled in the next iteration.
      //  VirtualMachine::Receive may be less effiencient if the thread safe version
      //  `engine_->SchedulerEmpty()`
      // used
      //  here, because VirtualMachine::ScheduleLoop is more likely to get the mutex lock.
      do {
        const size_t total_inserted = engine_->total_inserted_instruction_cnt();
        const size_t total_erased = engine_->total_erased_instruction_cnt();
        engine_->Schedule(schedule_ctx);
        if (ThreadLocalEnvBool<ONEFLOW_VM_ENABLE_SCHEDULE_YIELD>()
            && total_inserted == engine_->total_inserted_instruction_cnt()
            && total_erased == engine_->total_erased_instruction_cnt()) {  // nothing handled.
          std::this_thread::yield();
        }
      } while (!engine_->SchedulerThreadUnsafeEmpty());
    } while (MicrosecondsFrom(start) < kWorkingMicroseconds);
  }
  ScheduleUntilVMEmpty(engine_.Mutable(), schedule_ctx);
  CHECK_JUST(CloseWorkerThreads());
  scheduler_stopped_ = true;
}

intrusive::shared_ptr<vm::Dependence> VirtualMachine::FindOrCreateScheduleDependence(
    Symbol<Stream> stream) {