4851036339

Build Type

github

Committed by GitHub

Commit Message

Merge aee741046 into 87672e106

Pull Request Pull Request #515: v1.0

Run Details

5501 of 5501 new or added lines in 117 files covered. (100.0%)

4997 of 7299 relevant lines covered (68.46%)

6131966.73 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/src/thread_pool.h

/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
 * Distributed under BSD 3-Clause license.                                   *
 * Copyright by The HDF Group.                                               *
 * Copyright by the Illinois Institute of Technology.                        *
 * All rights reserved.                                                      *
 *                                                                           *
 * This file is part of Hermes. The full Hermes copyright notice, including  *
 * terms governing use, modification, and redistribution, is contained in    *
 * the COPYING file, which can be found at the top directory. If you do not  *
 * have access to the file, you may request a copy from help@hdfgroup.org.   *
 * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */

#ifndef HERMES_THREAD_POOL_H_
#define HERMES_THREAD_POOL_H_

#include <condition_variable>
#include <future>
#include <mutex>
#include <queue>
#include <thread>

namespace hermes {
/**
   A class to represent thread pool
 */
class ThreadPool {
 private:
  std::vector<std::thread> threads; /**< a vector of threads */
  /** high-priority  queue */
  mutable std::queue<std::packaged_task<void()>> queue_low;
  /** low-priority queue */
  mutable std::queue<std::packaged_task<void()>> queue_high;
  mutable std::mutex mutex;                /**< mutex lock */
  mutable std::condition_variable condvar; /**< conditional variable */

 public:
  /** construct thread pool with \a num_threads number of threads */
  explicit ThreadPool(
      unsigned num_threads = std::thread::hardware_concurrency()) {
    while (num_threads--) {
      threads.emplace_back([this] {
        while (true) {
          std::unique_lock<std::mutex> lock(mutex);
          condvar.wait(lock, [this]() {
            return !queue_high.empty() || !queue_low.empty();
          });
          bool high_priority = !queue_high.empty();
          auto task = high_priority ? std::move(queue_high.front())
                                    : std::move(queue_low.front());
          if (task.valid()) {
            if (high_priority) {
              queue_high.pop();
            } else {
              queue_low.pop();
            }
            lock.unlock();
            // run the task - this cannot throw; any exception
            // will be stored in the corresponding future
            task();
          } else {
            // an empty task is used to signal end of stream
            // don't pop it off the top; all threads need to see it
            break;
          }
        }
      });
    }
  }

  /** a template for running thread pool */
  template <typename F, typename R = std::result_of_t<F && ()>>
  std::future<R> run(F&& f, bool high_priority = false) const {
    auto task = std::packaged_task<R()>(std::forward<F>(f));
    auto future = task.get_future();
    {
      std::lock_guard<std::mutex> lock(mutex);
      // conversion to packaged_task<void()> erases the return type
      // so it can be stored in the queue. the future will still
      // contain the correct type
      if (high_priority) {
        queue_high.push(std::packaged_task<void()>(std::move(task)));
      } else {
        queue_low.push(std::packaged_task<void()>(std::move(task)));
      }
    }
    condvar.notify_one();
    return future;
  }

  ~ThreadPool() {
    // push a single empty task onto the queue and notify all threads,
    // then wait for them to terminate
    {
      std::lock_guard<std::mutex> lock(mutex);
      queue_low.push({});
    }
    condvar.notify_all();
    for (auto& thread : threads) {
      thread.join();
    }
  }
};
}  // namespace hermes
#endif  // HERMES_THREAD_POOL_H_

1	/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
2	* Distributed under BSD 3-Clause license. *
3	* Copyright by The HDF Group. *
4	* Copyright by the Illinois Institute of Technology. *
5	* All rights reserved. *
6	* *
7	* This file is part of Hermes. The full Hermes copyright notice, including *
8	* terms governing use, modification, and redistribution, is contained in *
9	* the COPYING file, which can be found at the top directory. If you do not *
10	* have access to the file, you may request a copy from help@hdfgroup.org. *
11	* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
12
13	#ifndef HERMES_THREAD_POOL_H_
14	#define HERMES_THREAD_POOL_H_
15
16	#include <condition_variable>
17	#include <future>
18	#include <mutex>
19	#include <queue>
20	#include <thread>
21
22	namespace hermes {
23	/**
24	A class to represent thread pool
25	*/
26	class ThreadPool {
27	private:
28	std::vector<std::thread> threads; /*< a vector of threads /
29	/** high-priority queue */
30	mutable std::queue<std::packaged_task<void()>> queue_low;
31	/** low-priority queue */
32	mutable std::queue<std::packaged_task<void()>> queue_high;
33	mutable std::mutex mutex; /*< mutex lock /
34	mutable std::condition_variable condvar; /*< conditional variable /
35
36	public:
37	/** construct thread pool with \a num_threads number of threads */
38	explicit ThreadPool(	×
39	unsigned num_threads = std::thread::hardware_concurrency()) {	×
40	while (num_threads--) {	×
41	threads.emplace_back([this] {	×
42	while (true) {	×
43	std::unique_lock<std::mutex> lock(mutex);	×
44	condvar.wait(lock, [this]() {	×
45	return !queue_high.empty() \|\| !queue_low.empty();	×
46	});
47	bool high_priority = !queue_high.empty();	×
48	auto task = high_priority ? std::move(queue_high.front())	×
49	: std::move(queue_low.front());	×
50	if (task.valid()) {	×
51	if (high_priority) {	×
52	queue_high.pop();	×
53	} else {
54	queue_low.pop();	×
55	}
56	lock.unlock();	×
57	// run the task - this cannot throw; any exception
58	// will be stored in the corresponding future
59	task();	×
60	} else {
61	// an empty task is used to signal end of stream
62	// don't pop it off the top; all threads need to see it
63	break;
64	}
65	}
66	});	×
67	}
68	}	×
69
70	/** a template for running thread pool */
71	template <typename F, typename R = std::result_of_t<F && ()>>
72	std::future<R> run(F&& f, bool high_priority = false) const {	×
73	auto task = std::packaged_task<R()>(std::forward<F>(f));	×
74	auto future = task.get_future();	×
75	{
76	std::lock_guard<std::mutex> lock(mutex);	×
77	// conversion to packaged_task<void()> erases the return type
78	// so it can be stored in the queue. the future will still
79	// contain the correct type
80	if (high_priority) {	×
81	queue_high.push(std::packaged_task<void()>(std::move(task)));	×
82	} else {
83	queue_low.push(std::packaged_task<void()>(std::move(task)));	×
84	}
85	}
86	condvar.notify_one();	×
87	return future;	×
88	}
89
90	~ThreadPool() {
91	// push a single empty task onto the queue and notify all threads,
92	// then wait for them to terminate
93	{
94	std::lock_guard<std::mutex> lock(mutex);
95	queue_low.push({});
96	}
97	condvar.notify_all();
98	for (auto& thread : threads) {
99	thread.join();
100	}
101	}
102	};
103	} // namespace hermes
104	#endif // HERMES_THREAD_POOL_H_

HDFGroup / hermes / 4851036339

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous