Mediapipe의 ThereadPool의 실복을 읽었어요.

Mediapipe는 Google이 개발한 오픈소스 컴퓨터 비주얼 파이프라인 프레임워크로 방향 그래프를 기술하여 추적 등 처리를 할 수 있습니다.
또한 몇 개의 처리용 도표를 준비했는데 이 도표만 사용하면 대체적으로 많은 처리를 수행할 수 있다.
그 도표는 처리에 가까운 입력 순서에서 라인 수영장 임무를 병행 처리하여 어느 정도 병행 처리한다.
우선 그 처리된 ThreadPool 부분의 실복을 읽어 보았다.
Mediapipe의 ThereadPool은 다음과 같이 사용됩니다.
매우 간단한 API입니다.설치도 200줄로 매우 가볍다.
onxruntime의 ThereadPool은 기능이 풍부하고 매우 길다.
thread 설치는 리눅스 시스템의 설치와 이외의 설치를 포함하여 리눅스의 처리와 읽기에 사용됩니다.

  ThreadPool pool("testpool", num_workers);
  pool.StartWorkers();
  for (int i = 0; i < N; ++i) {
    pool.Schedule([i]() { DoWork(i); });
  }

API에 사용되는 구성원 변수는 다음과 같습니다.
tasks는 작업 대기열로, FIFO로 사용됩니다.
CondVar는 다음과 같은 조건 변수입니다.

  absl::Mutex mutex_;
  absl::CondVar condition_;
  bool stopped_ ABSL_GUARDED_BY(mutex_) = false;
  std::deque<std::function<void()>> tasks_ ABSL_GUARDED_BY(mutex_);

StartWorkers에서 작업 스레드를 생성합니다.이 작업 스레드를 만들면 스레드가 생성됩니다. 이 스레드는 온라인 스레드 탱크에서 순환을 실행하고 단서를 기다립니다.내부는 pthread로 라인의 생성과 설정을 하는 것 같습니다.왜 사용하지 않았는지 모르겠다std::thread.리눅스 계열 이외의 설치를 위해 저쪽에서 사용한 것 같다std::thread.
등 스레드 대기열의 순환은 RunWorker 방법으로 다음과 같이 잠금mutex, 순환, 함수를 얻은 후 잠금을 해제하고 임무를 수행합니다.작업이 없을 때, 모든 라인은 condition입니다.Wait에서 추가 작업을 기다리는 중입니다.Schedule 방법으로 작업을 Push한 후 condinSignal을 통해 Wait 스레드 중 하나의 스레드의 블록을 구합니다.이렇게 하면 그 라인은 작업을 수행할 수 있으며, 작업 대기열에서 작업Unlock mutex를 꺼내서 작업을 수행할 수 있습니다.

void ThreadPool::StartWorkers() {
  for (int i = 0; i < num_threads_; ++i) {
    threads_.push_back(new WorkerThread(this, name_prefix_));
  }
}

void ThreadPool::Schedule(std::function<void()> callback) {
  mutex_.Lock();
  tasks_.push_back(std::move(callback));
  condition_.Signal();
  mutex_.Unlock();
}
...

void ThreadPool::RunWorker() {
  mutex_.Lock();
  while (true) {
    if (!tasks_.empty()) {
      std::function<void()> task = std::move(tasks_.front());
      tasks_.pop_front();
      mutex_.Unlock();
      task();
      mutex_.Lock();
    } else {
      if (stopped_) {
        break;
      } else {
        condition_.Wait(&mutex_);
      }
    }
  }
  mutex_.Unlock();
}

RunWorker에 대해 다음과 같이 매번tasks를 검사할 수 있다고 생각합니다.
그렇다면 다음과 같은 두 가지 결점이 있다고 생각합니다.

mutex.Lock 온라인 스레드 풀 내의 모든 스레드가 순환하기 때문에 Schedule 방법의 실행이 지연될 수 있습니다.

mutex.모든 라인에 Lock 가져오기와 검사 작업 대기열이 나타나기 때문에 모든 라인을 condition합니다.Wait로 대기하기보다는 CPU 점유율이 조금 높지 않나

void ThreadPool::RunWorker() { 
  while (true) {
    mutex_.Lock();
    if (!tasks_.empty()) {
      std::function<void()> task = std::move(tasks_.front());
      tasks_.pop_front();
      mutex_.Unlock();
      
      task();
      
    } else {
      mutex_.Unlock();
      if (stopped_) {
        break;
      }
    }
  }
}