Dennis Luxen
-106
@@ -1,106 +0,0 @@
|
||||
#ifndef OSMIUM_THREAD_CHECKED_TASK_HPP
|
||||
#define OSMIUM_THREAD_CHECKED_TASK_HPP
|
||||
|
||||
/*
|
||||
|
||||
This file is part of Osmium (http://osmcode.org/libosmium).
|
||||
|
||||
Copyright 2013,2014 Jochen Topf <jochen@topf.org> and others (see README).
|
||||
|
||||
Boost Software License - Version 1.0 - August 17th, 2003
|
||||
|
||||
Permission is hereby granted, free of charge, to any person or organization
|
||||
obtaining a copy of the software and accompanying documentation covered by
|
||||
this license (the "Software") to use, reproduce, display, distribute,
|
||||
execute, and transmit the Software, and to prepare derivative works of the
|
||||
Software, and to permit third-parties to whom the Software is furnished to
|
||||
do so, all subject to the following:
|
||||
|
||||
The copyright notices in the Software and this entire statement, including
|
||||
the above license grant, this restriction and the following disclaimer,
|
||||
must be included in all copies of the Software, in whole or in part, and
|
||||
all derivative works of the Software, unless such copies or derivative
|
||||
works are solely in the form of machine-executable object code generated by
|
||||
a source language processor.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE, TITLE AND NON-INFRINGEMENT. IN NO EVENT
|
||||
SHALL THE COPYRIGHT HOLDERS OR ANYONE DISTRIBUTING THE SOFTWARE BE LIABLE
|
||||
FOR ANY DAMAGES OR OTHER LIABILITY, WHETHER IN CONTRACT, TORT OR OTHERWISE,
|
||||
ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
|
||||
DEALINGS IN THE SOFTWARE.
|
||||
|
||||
*/
|
||||
|
||||
#include <chrono>
|
||||
#include <future>
|
||||
#include <thread>
|
||||
#include <utility>
|
||||
|
||||
namespace osmium {
|
||||
|
||||
namespace thread {
|
||||
|
||||
template <class T>
|
||||
class CheckedTask {
|
||||
|
||||
std::thread m_thread {};
|
||||
std::future<void> m_future {};
|
||||
|
||||
public:
|
||||
|
||||
template <class... TArgs>
|
||||
explicit CheckedTask(TArgs&&... args) {
|
||||
std::packaged_task<void()> pack_task(T(std::forward<TArgs>(args)...));
|
||||
m_future = pack_task.get_future();
|
||||
m_thread = std::thread(std::move(pack_task));
|
||||
}
|
||||
|
||||
~CheckedTask() {
|
||||
// Make sure task is done.
|
||||
if (m_thread.joinable()) {
|
||||
m_thread.join();
|
||||
}
|
||||
}
|
||||
|
||||
CheckedTask(const CheckedTask&) = delete;
|
||||
CheckedTask& operator=(const CheckedTask&) = delete;
|
||||
|
||||
/**
|
||||
* Check task for exceptions.
|
||||
*
|
||||
* If an exception happened in the task, re-throw it in this
|
||||
* thread. This will not do anything if there was no exception.
|
||||
*/
|
||||
void check_for_exception() {
|
||||
if (m_future.valid() && m_future.wait_for(std::chrono::seconds(0)) == std::future_status::ready) {
|
||||
m_future.get();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Close the task. This will raise in this thread any exception the
|
||||
* task generated in the other thread.
|
||||
*/
|
||||
void close() {
|
||||
// If an exception happened in the task, re-throw
|
||||
// it in this thread. This will block if the task
|
||||
// isn't finished.
|
||||
if (m_future.valid()) {
|
||||
m_future.get();
|
||||
}
|
||||
|
||||
// Make sure task is done.
|
||||
if (m_thread.joinable()) {
|
||||
m_thread.join();
|
||||
}
|
||||
}
|
||||
|
||||
}; // class CheckedTask
|
||||
|
||||
} // namespace thread
|
||||
|
||||
} // namespace osmium
|
||||
|
||||
#endif // OSMIUM_THREAD_CHECKED_TASK_HPP
|
||||
-104
@@ -1,104 +0,0 @@
|
||||
#ifndef OSMIUM_THREAD_FUNCTION_WRAPPER_HPP
|
||||
#define OSMIUM_THREAD_FUNCTION_WRAPPER_HPP
|
||||
|
||||
/*
|
||||
|
||||
This file is part of Osmium (http://osmcode.org/libosmium).
|
||||
|
||||
Copyright 2013,2014 Jochen Topf <jochen@topf.org> and others (see README).
|
||||
|
||||
Boost Software License - Version 1.0 - August 17th, 2003
|
||||
|
||||
Permission is hereby granted, free of charge, to any person or organization
|
||||
obtaining a copy of the software and accompanying documentation covered by
|
||||
this license (the "Software") to use, reproduce, display, distribute,
|
||||
execute, and transmit the Software, and to prepare derivative works of the
|
||||
Software, and to permit third-parties to whom the Software is furnished to
|
||||
do so, all subject to the following:
|
||||
|
||||
The copyright notices in the Software and this entire statement, including
|
||||
the above license grant, this restriction and the following disclaimer,
|
||||
must be included in all copies of the Software, in whole or in part, and
|
||||
all derivative works of the Software, unless such copies or derivative
|
||||
works are solely in the form of machine-executable object code generated by
|
||||
a source language processor.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE, TITLE AND NON-INFRINGEMENT. IN NO EVENT
|
||||
SHALL THE COPYRIGHT HOLDERS OR ANYONE DISTRIBUTING THE SOFTWARE BE LIABLE
|
||||
FOR ANY DAMAGES OR OTHER LIABILITY, WHETHER IN CONTRACT, TORT OR OTHERWISE,
|
||||
ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
|
||||
DEALINGS IN THE SOFTWARE.
|
||||
|
||||
*/
|
||||
|
||||
#include <algorithm>
|
||||
#include <memory>
|
||||
|
||||
namespace osmium {
|
||||
|
||||
namespace thread {
|
||||
|
||||
class function_wrapper {
|
||||
|
||||
struct impl_base {
|
||||
virtual void call() = 0;
|
||||
virtual ~impl_base() {
|
||||
}
|
||||
};
|
||||
|
||||
std::unique_ptr<impl_base> impl;
|
||||
|
||||
template <typename F>
|
||||
struct impl_type : impl_base {
|
||||
F m_functor;
|
||||
|
||||
impl_type(F&& functor) :
|
||||
m_functor(std::move(functor)) {
|
||||
}
|
||||
|
||||
void call() {
|
||||
m_functor();
|
||||
}
|
||||
};
|
||||
|
||||
public:
|
||||
|
||||
// Constructor must not be "explicit" for wrapper
|
||||
// to work seemlessly.
|
||||
template <typename F>
|
||||
function_wrapper(F&& f) :
|
||||
impl(new impl_type<F>(std::move(f))) {
|
||||
}
|
||||
|
||||
void operator()() {
|
||||
impl->call();
|
||||
}
|
||||
|
||||
function_wrapper() = default;
|
||||
|
||||
function_wrapper(function_wrapper&& other) :
|
||||
impl(std::move(other.impl)) {
|
||||
}
|
||||
|
||||
function_wrapper& operator=(function_wrapper&& other) {
|
||||
impl = std::move(other.impl);
|
||||
return *this;
|
||||
}
|
||||
|
||||
function_wrapper(const function_wrapper&) = delete;
|
||||
function_wrapper(function_wrapper&) = delete;
|
||||
function_wrapper& operator=(const function_wrapper&) = delete;
|
||||
|
||||
explicit operator bool() const {
|
||||
return static_cast<bool>(impl);
|
||||
}
|
||||
|
||||
}; // class function_wrapper
|
||||
|
||||
} // namespace thread
|
||||
|
||||
} // namespace osmium
|
||||
|
||||
#endif // OSMIUM_THREAD_FUNCTION_WRAPPER_HPP
|
||||
Vendored
-61
@@ -1,61 +0,0 @@
|
||||
#ifndef OSMIUM_THREAD_NAME_HPP
|
||||
#define OSMIUM_THREAD_NAME_HPP
|
||||
|
||||
/*
|
||||
|
||||
This file is part of Osmium (http://osmcode.org/libosmium).
|
||||
|
||||
Copyright 2013,2014 Jochen Topf <jochen@topf.org> and others (see README).
|
||||
|
||||
Boost Software License - Version 1.0 - August 17th, 2003
|
||||
|
||||
Permission is hereby granted, free of charge, to any person or organization
|
||||
obtaining a copy of the software and accompanying documentation covered by
|
||||
this license (the "Software") to use, reproduce, display, distribute,
|
||||
execute, and transmit the Software, and to prepare derivative works of the
|
||||
Software, and to permit third-parties to whom the Software is furnished to
|
||||
do so, all subject to the following:
|
||||
|
||||
The copyright notices in the Software and this entire statement, including
|
||||
the above license grant, this restriction and the following disclaimer,
|
||||
must be included in all copies of the Software, in whole or in part, and
|
||||
all derivative works of the Software, unless such copies or derivative
|
||||
works are solely in the form of machine-executable object code generated by
|
||||
a source language processor.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE, TITLE AND NON-INFRINGEMENT. IN NO EVENT
|
||||
SHALL THE COPYRIGHT HOLDERS OR ANYONE DISTRIBUTING THE SOFTWARE BE LIABLE
|
||||
FOR ANY DAMAGES OR OTHER LIABILITY, WHETHER IN CONTRACT, TORT OR OTHERWISE,
|
||||
ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
|
||||
DEALINGS IN THE SOFTWARE.
|
||||
|
||||
*/
|
||||
|
||||
#ifdef __linux__
|
||||
# include <sys/prctl.h>
|
||||
#endif
|
||||
|
||||
namespace osmium {
|
||||
|
||||
namespace thread {
|
||||
|
||||
/**
|
||||
* Set name of current thread for debugging. This only works on Linux.
|
||||
*/
|
||||
#ifdef __linux__
|
||||
inline void set_thread_name(const char* name) {
|
||||
prctl(PR_SET_NAME, name, 0, 0, 0);
|
||||
}
|
||||
#else
|
||||
inline void set_thread_name(const char*) {
|
||||
// intentionally left blank
|
||||
}
|
||||
#endif
|
||||
|
||||
} // namespace thread
|
||||
|
||||
} // namespace osmium
|
||||
|
||||
#endif // OSMIUM_THREAD_NAME_HPP
|
||||
Vendored
-180
@@ -1,180 +0,0 @@
|
||||
#ifndef OSMIUM_THREAD_POOL_HPP
|
||||
#define OSMIUM_THREAD_POOL_HPP
|
||||
|
||||
/*
|
||||
|
||||
This file is part of Osmium (http://osmcode.org/libosmium).
|
||||
|
||||
Copyright 2013,2014 Jochen Topf <jochen@topf.org> and others (see README).
|
||||
|
||||
Boost Software License - Version 1.0 - August 17th, 2003
|
||||
|
||||
Permission is hereby granted, free of charge, to any person or organization
|
||||
obtaining a copy of the software and accompanying documentation covered by
|
||||
this license (the "Software") to use, reproduce, display, distribute,
|
||||
execute, and transmit the Software, and to prepare derivative works of the
|
||||
Software, and to permit third-parties to whom the Software is furnished to
|
||||
do so, all subject to the following:
|
||||
|
||||
The copyright notices in the Software and this entire statement, including
|
||||
the above license grant, this restriction and the following disclaimer,
|
||||
must be included in all copies of the Software, in whole or in part, and
|
||||
all derivative works of the Software, unless such copies or derivative
|
||||
works are solely in the form of machine-executable object code generated by
|
||||
a source language processor.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE, TITLE AND NON-INFRINGEMENT. IN NO EVENT
|
||||
SHALL THE COPYRIGHT HOLDERS OR ANYONE DISTRIBUTING THE SOFTWARE BE LIABLE
|
||||
FOR ANY DAMAGES OR OTHER LIABILITY, WHETHER IN CONTRACT, TORT OR OTHERWISE,
|
||||
ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
|
||||
DEALINGS IN THE SOFTWARE.
|
||||
|
||||
*/
|
||||
|
||||
#include <algorithm>
|
||||
#include <atomic>
|
||||
#include <cstddef>
|
||||
#include <cstdlib>
|
||||
#include <future>
|
||||
#include <thread>
|
||||
#include <type_traits>
|
||||
#include <vector>
|
||||
|
||||
#include <osmium/thread/queue.hpp>
|
||||
#include <osmium/thread/name.hpp>
|
||||
#include <osmium/thread/function_wrapper.hpp>
|
||||
|
||||
namespace osmium {
|
||||
|
||||
/**
|
||||
* @brief Threading-related low-level code
|
||||
*/
|
||||
namespace thread {
|
||||
|
||||
/**
|
||||
* Thread pool.
|
||||
*/
|
||||
class Pool {
|
||||
|
||||
// This class makes sure pool threads are joined when the pool is destructed
|
||||
class thread_joiner {
|
||||
|
||||
std::vector<std::thread>& m_threads;
|
||||
|
||||
public:
|
||||
|
||||
explicit thread_joiner(std::vector<std::thread>& threads) :
|
||||
m_threads(threads) {
|
||||
}
|
||||
|
||||
~thread_joiner() {
|
||||
for (auto& thread : m_threads) {
|
||||
if (thread.joinable()) {
|
||||
thread.join();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}; // class thread_joiner
|
||||
|
||||
std::atomic<bool> m_done;
|
||||
osmium::thread::Queue<function_wrapper> m_work_queue;
|
||||
std::vector<std::thread> m_threads;
|
||||
thread_joiner m_joiner;
|
||||
int m_num_threads;
|
||||
|
||||
void worker_thread() {
|
||||
osmium::thread::set_thread_name("_osmium_worker");
|
||||
while (!m_done) {
|
||||
function_wrapper task;
|
||||
m_work_queue.wait_and_pop_with_timeout(task);
|
||||
if (task) {
|
||||
task();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Create thread pool with the given number of threads. If
|
||||
* num_threads is 0, the number of threads is read from
|
||||
* the environment variable OSMIUM_POOL_THREADS. The default
|
||||
* value in that case is -2.
|
||||
*
|
||||
* If the number of threads is a negative number, it will be
|
||||
* set to the actual number of cores on the system plus the
|
||||
* given number, ie it will leave a number of cores unused.
|
||||
*
|
||||
* In all cases the minimum number of threads in the pool is 1.
|
||||
*/
|
||||
explicit Pool(int num_threads) :
|
||||
m_done(false),
|
||||
m_work_queue(),
|
||||
m_threads(),
|
||||
m_joiner(m_threads),
|
||||
m_num_threads(num_threads) {
|
||||
|
||||
if (m_num_threads == 0) {
|
||||
const char* env_threads = getenv("OSMIUM_POOL_THREADS");
|
||||
if (env_threads) {
|
||||
m_num_threads = std::atoi(env_threads);
|
||||
} else {
|
||||
m_num_threads = -2;
|
||||
}
|
||||
}
|
||||
|
||||
if (m_num_threads <= 0) {
|
||||
m_num_threads = std::max(1, static_cast<int>(std::thread::hardware_concurrency()) + m_num_threads);
|
||||
}
|
||||
|
||||
try {
|
||||
for (int i=0; i < m_num_threads; ++i) {
|
||||
m_threads.push_back(std::thread(&Pool::worker_thread, this));
|
||||
}
|
||||
} catch (...) {
|
||||
m_done = true;
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
public:
|
||||
|
||||
static constexpr int default_num_threads = 0;
|
||||
|
||||
static Pool& instance() {
|
||||
static Pool pool(default_num_threads);
|
||||
return pool;
|
||||
}
|
||||
|
||||
~Pool() {
|
||||
m_done = true;
|
||||
}
|
||||
|
||||
size_t queue_size() const {
|
||||
return m_work_queue.size();
|
||||
}
|
||||
|
||||
bool queue_empty() const {
|
||||
return m_work_queue.empty();
|
||||
}
|
||||
|
||||
template <typename TFunction>
|
||||
std::future<typename std::result_of<TFunction()>::type> submit(TFunction f) {
|
||||
|
||||
typedef typename std::result_of<TFunction()>::type result_type;
|
||||
|
||||
std::packaged_task<result_type()> task(std::move(f));
|
||||
std::future<result_type> future_result(task.get_future());
|
||||
m_work_queue.push(std::move(task));
|
||||
|
||||
return future_result;
|
||||
}
|
||||
|
||||
}; // class Pool
|
||||
|
||||
} // namespace thread
|
||||
|
||||
} // namespace osmium
|
||||
|
||||
#endif // OSMIUM_THREAD_POOL_HPP
|
||||
Vendored
-128
@@ -1,128 +0,0 @@
|
||||
#ifndef OSMIUM_THREAD_QUEUE_HPP
|
||||
#define OSMIUM_THREAD_QUEUE_HPP
|
||||
|
||||
/*
|
||||
|
||||
This file is part of Osmium (http://osmcode.org/libosmium).
|
||||
|
||||
Copyright 2013,2014 Jochen Topf <jochen@topf.org> and others (see README).
|
||||
|
||||
Boost Software License - Version 1.0 - August 17th, 2003
|
||||
|
||||
Permission is hereby granted, free of charge, to any person or organization
|
||||
obtaining a copy of the software and accompanying documentation covered by
|
||||
this license (the "Software") to use, reproduce, display, distribute,
|
||||
execute, and transmit the Software, and to prepare derivative works of the
|
||||
Software, and to permit third-parties to whom the Software is furnished to
|
||||
do so, all subject to the following:
|
||||
|
||||
The copyright notices in the Software and this entire statement, including
|
||||
the above license grant, this restriction and the following disclaimer,
|
||||
must be included in all copies of the Software, in whole or in part, and
|
||||
all derivative works of the Software, unless such copies or derivative
|
||||
works are solely in the form of machine-executable object code generated by
|
||||
a source language processor.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE, TITLE AND NON-INFRINGEMENT. IN NO EVENT
|
||||
SHALL THE COPYRIGHT HOLDERS OR ANYONE DISTRIBUTING THE SOFTWARE BE LIABLE
|
||||
FOR ANY DAMAGES OR OTHER LIABILITY, WHETHER IN CONTRACT, TORT OR OTHERWISE,
|
||||
ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
|
||||
DEALINGS IN THE SOFTWARE.
|
||||
|
||||
*/
|
||||
|
||||
#include <chrono>
|
||||
#include <condition_variable>
|
||||
#include <cstddef>
|
||||
#include <mutex>
|
||||
#include <queue>
|
||||
#include <utility>
|
||||
|
||||
namespace osmium {
|
||||
|
||||
namespace thread {
|
||||
|
||||
/**
|
||||
* A thread-safe queue.
|
||||
*/
|
||||
template <typename T>
|
||||
class Queue {
|
||||
|
||||
mutable std::mutex m_mutex;
|
||||
std::queue<T> m_queue;
|
||||
std::condition_variable m_data_available;
|
||||
|
||||
public:
|
||||
|
||||
Queue() :
|
||||
m_mutex(),
|
||||
m_queue(),
|
||||
m_data_available() {
|
||||
}
|
||||
|
||||
void push(T value) {
|
||||
std::lock_guard<std::mutex> lock(m_mutex);
|
||||
m_queue.push(std::move(value));
|
||||
m_data_available.notify_one();
|
||||
}
|
||||
|
||||
size_t push_and_get_size(T&& value) {
|
||||
std::lock_guard<std::mutex> lock(m_mutex);
|
||||
m_queue.push(std::forward<T>(value));
|
||||
m_data_available.notify_one();
|
||||
return m_queue.size();
|
||||
}
|
||||
|
||||
void push(T value, int) {
|
||||
push(value);
|
||||
}
|
||||
|
||||
void wait_and_pop(T& value) {
|
||||
std::unique_lock<std::mutex> lock(m_mutex);
|
||||
m_data_available.wait(lock, [this] {
|
||||
return !m_queue.empty();
|
||||
});
|
||||
value=std::move(m_queue.front());
|
||||
m_queue.pop();
|
||||
}
|
||||
|
||||
void wait_and_pop_with_timeout(T& value) {
|
||||
std::unique_lock<std::mutex> lock(m_mutex);
|
||||
if (!m_data_available.wait_for(lock, std::chrono::seconds(1), [this] {
|
||||
return !m_queue.empty();
|
||||
})) {
|
||||
return;
|
||||
}
|
||||
value=std::move(m_queue.front());
|
||||
m_queue.pop();
|
||||
}
|
||||
|
||||
bool try_pop(T& value) {
|
||||
std::lock_guard<std::mutex> lock(m_mutex);
|
||||
if (m_queue.empty()) {
|
||||
return false;
|
||||
}
|
||||
value=std::move(m_queue.front());
|
||||
m_queue.pop();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool empty() const {
|
||||
std::lock_guard<std::mutex> lock(m_mutex);
|
||||
return m_queue.empty();
|
||||
}
|
||||
|
||||
size_t size() const {
|
||||
std::lock_guard<std::mutex> lock(m_mutex);
|
||||
return m_queue.size();
|
||||
}
|
||||
|
||||
}; // class Queue
|
||||
|
||||
} // namespace thread
|
||||
|
||||
} // namespace osmium
|
||||
|
||||
#endif // OSMIUM_THREAD_QUEUE_HPP
|
||||
-159
@@ -1,159 +0,0 @@
|
||||
#ifndef OSMIUM_THREAD_SORTED_QUEUE_HPP
|
||||
#define OSMIUM_THREAD_SORTED_QUEUE_HPP
|
||||
|
||||
/*
|
||||
|
||||
This file is part of Osmium (http://osmcode.org/libosmium).
|
||||
|
||||
Copyright 2013,2014 Jochen Topf <jochen@topf.org> and others (see README).
|
||||
|
||||
Boost Software License - Version 1.0 - August 17th, 2003
|
||||
|
||||
Permission is hereby granted, free of charge, to any person or organization
|
||||
obtaining a copy of the software and accompanying documentation covered by
|
||||
this license (the "Software") to use, reproduce, display, distribute,
|
||||
execute, and transmit the Software, and to prepare derivative works of the
|
||||
Software, and to permit third-parties to whom the Software is furnished to
|
||||
do so, all subject to the following:
|
||||
|
||||
The copyright notices in the Software and this entire statement, including
|
||||
the above license grant, this restriction and the following disclaimer,
|
||||
must be included in all copies of the Software, in whole or in part, and
|
||||
all derivative works of the Software, unless such copies or derivative
|
||||
works are solely in the form of machine-executable object code generated by
|
||||
a source language processor.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE, TITLE AND NON-INFRINGEMENT. IN NO EVENT
|
||||
SHALL THE COPYRIGHT HOLDERS OR ANYONE DISTRIBUTING THE SOFTWARE BE LIABLE
|
||||
FOR ANY DAMAGES OR OTHER LIABILITY, WHETHER IN CONTRACT, TORT OR OTHERWISE,
|
||||
ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
|
||||
DEALINGS IN THE SOFTWARE.
|
||||
|
||||
*/
|
||||
|
||||
#include <condition_variable>
|
||||
#include <cstddef>
|
||||
#include <deque>
|
||||
#include <mutex>
|
||||
|
||||
namespace osmium {
|
||||
|
||||
namespace thread {
|
||||
|
||||
/**
|
||||
* This implements a sorted queue. It is a bit like a priority
|
||||
* queue. We have n worker threads pushing items into the queue
|
||||
* and one thread pulling them out again "in order". The order
|
||||
* is defined by the monotonically increasing "num" parameter
|
||||
* to the push() method. The wait_and_pop() and try_pop() methods
|
||||
* will only give out the next numbered item. This way several
|
||||
* workers can work in their own time on different pieces of
|
||||
* some incoming data, but it all gets serialized properly again
|
||||
* after the workers have done their work.
|
||||
*/
|
||||
template <typename T>
|
||||
class SortedQueue {
|
||||
|
||||
typedef typename std::deque<T>::size_type size_type;
|
||||
|
||||
mutable std::mutex m_mutex;
|
||||
std::deque<T> m_queue;
|
||||
std::condition_variable m_data_available;
|
||||
|
||||
size_type m_offset;
|
||||
|
||||
// this method expects that we already have the lock
|
||||
bool empty_intern() const {
|
||||
return m_queue.front() == T();
|
||||
}
|
||||
|
||||
public:
|
||||
|
||||
SortedQueue() :
|
||||
m_mutex(),
|
||||
m_queue(1),
|
||||
m_data_available(),
|
||||
m_offset(0) {
|
||||
}
|
||||
|
||||
/**
|
||||
* Push an item into the queue.
|
||||
*
|
||||
* @param value The item to push into the queue.
|
||||
* @param num Number to describe ordering for the items.
|
||||
* It must increase monotonically.
|
||||
*/
|
||||
void push(T value, size_type num) {
|
||||
std::lock_guard<std::mutex> lock(m_mutex);
|
||||
|
||||
num -= m_offset;
|
||||
if (m_queue.size() <= num + 1) {
|
||||
m_queue.resize(num + 2);
|
||||
}
|
||||
m_queue[num] = std::move(value);
|
||||
|
||||
m_data_available.notify_one();
|
||||
}
|
||||
|
||||
/**
|
||||
* Wait until the next item becomes available and make it
|
||||
* available through value.
|
||||
*/
|
||||
void wait_and_pop(T& value) {
|
||||
std::unique_lock<std::mutex> lock(m_mutex);
|
||||
|
||||
m_data_available.wait(lock, [this] {
|
||||
return !empty_intern();
|
||||
});
|
||||
value=std::move(m_queue.front());
|
||||
m_queue.pop_front();
|
||||
++m_offset;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get next item if it is available and return true. Or
|
||||
* return false otherwise.
|
||||
*/
|
||||
bool try_pop(T& value) {
|
||||
std::lock_guard<std::mutex> lock(m_mutex);
|
||||
|
||||
if (empty_intern()) {
|
||||
return false;
|
||||
}
|
||||
value=std::move(m_queue.front());
|
||||
m_queue.pop_front();
|
||||
++m_offset;
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* The queue is empty. This means try_pop() would fail if called.
|
||||
* It does not mean that there is nothing on the queue. Because
|
||||
* the queue is sorted, it could mean that the next item in the
|
||||
* queue is not available, but other items are.
|
||||
*/
|
||||
bool empty() const {
|
||||
std::lock_guard<std::mutex> lock(m_mutex);
|
||||
|
||||
return empty_intern();
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the number of items in the queue, regardless of whether
|
||||
* they can be accessed. If this is =0 it
|
||||
* implies empty()==true, but not the other way around.
|
||||
*/
|
||||
size_t size() const {
|
||||
std::lock_guard<std::mutex> lock(m_mutex);
|
||||
return m_queue.size();
|
||||
}
|
||||
|
||||
}; // class SortedQueue
|
||||
|
||||
} // namespace thread
|
||||
|
||||
} // namespace osmium
|
||||
|
||||
#endif // OSMIUM_THREAD_SORTED_QUEUE_HPP
|
||||
Reference in New Issue
Block a user