Class ThreadPool

be/src/util/thread-pool.h:34–218

/// blocking queue by Offer(). Each item is processed by a single user-supplied method.
template <typename T>
class ThreadPool : public CacheLineAligned {
 public:
  /// Signature of a work-processing function. Takes the integer id of the thread which is
  /// calling it (ids run from 0 to num_threads - 1) and a reference to the item to
  /// process.
  typedef boost::function<void (int thread_id, const T& workitem)> WorkFunction;

  /// Creates a new thread pool without starting any threads. Code must call
  /// Init() on this thread pool before any calls to Offer().
  ///  -- num_threads: how many threads are part of this pool
  ///  -- queue_size: the maximum size of the queue on which work items are offered. If the
  ///     queue exceeds this size, subsequent calls to Offer will block until there is
  ///     capacity available.
  ///  -- work_function: the function to run every time an item is consumed from the queue
  ///  -- fault_injection_eligible - If set to true, allow fault injection at this
  ///     callsite (see thread_creation_fault_injection). If set to false, fault
  ///     injection is diabled at this callsite. Thread creation sites that crash
  ///     Impala or abort startup must have this set to false.
  ThreadPool(const std::string& group, const std::string& thread_prefix,
      uint32_t num_threads, uint32_t queue_size, const WorkFunction& work_function,
      bool fault_injection_eligible = false)
    : group_(group), thread_prefix_(thread_prefix), num_threads_(num_threads),
      work_function_(work_function), work_queue_(queue_size),
      fault_injection_eligible_(fault_injection_eligible) {}

  /// Destructor ensures that all threads are terminated before this object is freed
  /// (otherwise they may continue to run and reference member variables)
  virtual ~ThreadPool() {
    Shutdown();
    Join();
  }

  /// Create the threads needed for this ThreadPool. Returns an error on any
  /// error spawning the threads.
  Status Init() {
    for (int i = 0; i < num_threads_; ++i) {
      std::stringstream threadname;
      threadname << thread_prefix_ << "(" << i + 1 << ":" << num_threads_ << ")";
      std::unique_ptr<Thread> t;
      Status status = Thread::Create(group_, threadname.str(),
          boost::bind<void>(boost::mem_fn(&ThreadPool<T>::WorkerThread), this, i), &t,
          fault_injection_eligible_);
      if (!status.ok()) {
        // The thread pool initialization failed. Shutdown any threads that were
        // spawned. Note: Shutdown() and Join() are safe to call multiple times.
        Shutdown();
        Join();
        return status;
      }
      threads_.AddThread(std::move(t));
    }
    initialized_ = true;
    return Status::OK();
  }

  /// Blocking operation that puts a work item on the queue. If the queue is full, blocks
  /// until there is capacity available. The ThreadPool must be initialized before
  /// calling this method.