Method ThreadMain

dali/pipeline/util/thread_pool.cc:133–232

void OldThreadPool::ThreadMain(int thread_id, int device_id, bool set_affinity,
                            const std::string &name) {
  this_thread_idx_ = thread_id;
  SetThreadName(name.c_str());
  DeviceGuard g(device_id);
  try {
#if NVML_ENABLED
    if (set_affinity) {
      const char *env_affinity = std::getenv("DALI_AFFINITY_MASK");
      int core = -1;
      if (env_affinity) {
        const auto &vec = string_split(env_affinity, ',');
        if ((size_t)thread_id < vec.size()) {
          core = std::stoi(vec[thread_id]);
        } else {
          DALI_WARN("DALI_AFFINITY_MASK environment variable is set, "
                    "but does not have enough entries: thread_id (", thread_id,
                    ") vs #entries (", vec.size(), "). Ignoring...");
        }
      }
      nvml::SetCPUAffinity(core);
    }
#endif
  } catch (...) {
    tl_errors_[thread_id].push(std::current_exception());
  }

  while (running_) {
    // Wait for something to do
    queue_semaphore_.acquire();

    // This lock guards only the queue, not the condition - that's handled by the semaphore
    std::unique_lock lock(queue_lock_);

    if (!running_)
      break;

    // Get work from the queue.
    WorkWithThreadIdx work = std::move(work_queue_.top().second);
    work_queue_.pop();
    // Unlock the lock
    lock.unlock();

    // If an error occurs, we save it in tl_errors_. When
    // WaitForWork is called, we will check for any errors
    // in the threads and return an error if one occured.
    try {
      work(thread_id);
    } catch (...) {
      tl_errors_[thread_id].push(std::current_exception());
    }

    // The task is now complete - we can atomically decrement the number of outstanding work.
    // If it reaches zero, we must safely notify the potential threads waiting for the work
    // to complete.
    // NOTE: We don't have to acquire the mutex until the number of waiting threads reaches 0.
    if (--outstanding_work_ == 0) {
      // We don't need to guard the modification of the atomic value with a mutex -