Method EnqueueTask

tensorflow/core/framework/run_handler.cc:121–157

  }

  Task EnqueueTask(Task t, bool is_blocking) {
    mutex* mu = nullptr;
    Queue* task_queue = nullptr;
    thread_local int64 closure_counter = 0;

    if (!is_blocking) {
      int queue_index = ++closure_counter % non_blocking_work_sharding_factor_;
      task_queue = &(non_blocking_work_queues_[queue_index]->queue);
      mu = &non_blocking_work_queues_[queue_index]->queue_op_mu;
    } else {
      task_queue = &blocking_work_queue_;
      mu = &blocking_queue_op_mu_;
    }

    {
      mutex_lock l(*mu);
      // For a given queue, only one thread can call PushFront.
      t = task_queue->PushFront(std::move(t));
    }

    // Only wake up the thread that can take tasks from both blocking and
    // non-blocking queues. The rational is that we don't want to wake up more
    // threads than the available physical cores for them to compete for
    // resource. The non-blocking threads are used only to compensate for
    // threads that may be blocked on some tasks. There is less need to
    // proactively wake up those threads.
    static int max_rank_to_wakeup = static_cast<int>(ParamFromEnvWithDefault(
        "TF_RUN_HANDLER_MAX_RANK_TO_WAKE_UP", kMaxConcurrentHandlers));
    if (max_rank_to_wakeup > 0 &&
        rank_.load(std::memory_order_relaxed) <= max_rank_to_wakeup) {
      mutex_lock l(waiters_mu_);
      queue_waiters_.next->cv.notify_one();
    }
    VLOG(3) << "Added " << (is_blocking ? "inter" : "intra") << " work from "
            << traceme_id_.load(std::memory_order_relaxed);
    return t;
  }

  Task PopBlockingTask() { return blocking_work_queue_.PopBack(); }