TBBThreadPool.h

Go to the documentation of this file.

/*
 * Custom TensorFlow thread pool implementation that schedules tasks in TBB.
 * Based on TensorFlow 2.1.
 * For more info, see https://gitlab.cern.ch/mrieger/CMSSW-DNN.
 *
 * Author: Marcel Rieger
 */
 
#ifndef PHYSICSTOOLS_TENSORFLOW_TBBTHREADPOOL_H
#define PHYSICSTOOLS_TENSORFLOW_TBBTHREADPOOL_H
 
#include "FWCore/Utilities/interface/thread_safety_macros.h"
 
#include "tensorflow/core/lib/core/threadpool.h"
 
#include "tbb/task_arena.h"
#include "tbb/task_group.h"
#include "tbb/global_control.h"
 
namespace tensorflow {
 
  class TBBThreadPool : public tensorflow::thread::ThreadPoolInterface {
  public:
    static TBBThreadPool& instance(int nThreads = -1) {
      CMS_THREAD_SAFE static TBBThreadPool pool(nThreads);
      return pool;
    }
 
    explicit TBBThreadPool(int nThreads = -1)
        : nThreads_(nThreads > 0 ? nThreads
                                 : tbb::global_control::active_value(tbb::global_control::max_allowed_parallelism)),
          numScheduleCalled_(0) {
      // when nThreads is zero or smaller, use the default value determined by tbb
    }
 
    void Schedule(std::function<void()> fn) override {
      numScheduleCalled_ += 1;
 
      // use a task arena to avoid having unrelated tasks start
      // running on this thread, which could potentially start deadlocks
      tbb::task_arena taskArena;
      tbb::task_group taskGroup;
 
      // we are required to always call wait before destructor
      auto doneWithTaskGroup = [&taskArena, &taskGroup](void*) {
        taskArena.execute([&taskGroup]() { taskGroup.wait(); });
      };
      std::unique_ptr<tbb::task_group, decltype(doneWithTaskGroup)> taskGuard(&taskGroup, doneWithTaskGroup);
 
      // schedule the task
      taskArena.execute([&taskGroup, &fn] { taskGroup.run(fn); });
 
      // reset the task guard which will call wait
      taskGuard.reset();
    }
 
    void ScheduleWithHint(std::function<void()> fn, int start, int end) override { Schedule(fn); }
 
    void Cancel() override {}
 
    int NumThreads() const override { return nThreads_; }
 
    int CurrentThreadId() const override {
      static std::atomic<int> idCounter{0};
      thread_local const int id = idCounter++;
      return id;
    }
 
    int GetNumScheduleCalled() { return numScheduleCalled_; }
 
  private:
    const int nThreads_;
    std::atomic<int> numScheduleCalled_;
  };
 
}  // namespace tensorflow
 
#endif  // PHYSICSTOOLS_TENSORFLOW_TBBTHREADPOOL_H