Merge commit '73efcc6b0ccedf8c1b6d95abdba8340cc9adf100' as 'third_party/libosmium'

2015-01-13 16:54:25 +01:00
parent d6946a8aae 73efcc6b0c
commit d69510f997
235 changed files with 53733 additions and 0 deletions
@@ -0,0 +1,110 @@
+#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 {
+
+        /**
+         * This function wrapper can collect move-only functions unlike
+         * std::function which needs copyable functions.
+         * Taken from the book "C++ Concurrency in Action".
+         */
+        class function_wrapper {
+
+            struct impl_base {
+
+                virtual ~impl_base() = default;
+                virtual void call() = 0;
+
+            }; // struct 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() override {
+                    m_functor();
+                }
+            }; // struct impl_type
+
+        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
@@ -0,0 +1,180 @@
+#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/function_wrapper.hpp>
+#include <osmium/thread/queue.hpp>
+#include <osmium/thread/util.hpp>
+#include <osmium/util/config.hpp>
+
+namespace osmium {
+
+    /**
+     * @brief Threading-related low-level code
+     */
+    namespace thread {
+
+        /**
+         *  Thread pool.
+         */
+        class Pool {
+
+            /**
+             * This class makes sure all pool threads will be 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, size_t max_queue_size) :
+                m_done(false),
+                m_work_queue(max_queue_size, "work"),
+                m_threads(),
+                m_joiner(m_threads),
+                m_num_threads(num_threads) {
+
+                if (m_num_threads == 0) {
+                    m_num_threads = osmium::config::get_pool_threads();
+                }
+
+                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 constexpr size_t max_work_queue_size = 10;
+
+            static Pool& instance() {
+                static Pool pool(default_num_threads, max_work_queue_size);
+                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
@@ -0,0 +1,178 @@
+#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 <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <cstddef>
+#include <mutex>
+#include <queue>
+#include <thread>
+#include <utility>
+
+namespace osmium {
+
+    namespace thread {
+
+        constexpr std::chrono::milliseconds full_queue_sleep_duration { 10 }; // XXX
+
+        /**
+         *  A thread-safe queue.
+         */
+        template <typename T>
+        class Queue {
+
+            /// Maximum size of this queue. If the queue is full pushing to
+            /// the queue will block.
+            const size_t m_max_size;
+
+            /// Name of this queue (for debugging only).
+            const std::string m_name;
+
+            mutable std::mutex m_mutex;
+
+            std::queue<T> m_queue;
+
+            /// Used to signal readers when data is available in the queue.
+            std::condition_variable m_data_available;
+
+#ifdef OSMIUM_DEBUG_QUEUE_SIZE
+            /// The largest size the queue has been so far.
+            size_t m_largest_size;
+
+            /// The number of times the queue was full and a thread pushing
+            /// to the queue was blocked.
+            std::atomic<int> m_full_counter;
+#endif
+
+        public:
+
+            /**
+             * Construct a multithreaded queue.
+             *
+             * @param max_size Maximum number of elements in the queue. Set to
+             *                 0 for an unlimited size.
+             * @param name Optional name for this queue. (Used for debugging.)
+             */
+            Queue(size_t max_size = 0, const std::string& name = "") :
+                m_max_size(max_size),
+                m_name(name),
+                m_mutex(),
+                m_queue(),
+                m_data_available()
+#ifdef OSMIUM_DEBUG_QUEUE_SIZE
+                ,
+                m_largest_size(0),
+                m_full_counter(0)
+#endif
+            {
+            }
+
+            ~Queue() {
+#ifdef OSMIUM_DEBUG_QUEUE_SIZE
+                std::cerr << "queue '" << m_name << "' with max_size=" << m_max_size << " had largest size " << m_largest_size << " and was full " << m_full_counter << " times\n";
+#endif
+            }
+
+            /**
+             * Push an element onto the queue. If the queue has a max size, this
+             * call will block if the queue is full.
+             */
+            void push(T value) {
+                if (m_max_size) {
+                    while (size() >= m_max_size) {
+                        std::this_thread::sleep_for(full_queue_sleep_duration);
+#ifdef OSMIUM_DEBUG_QUEUE_SIZE
+                        ++m_full_counter;
+#endif
+                    }
+                }
+                std::lock_guard<std::mutex> lock(m_mutex);
+                m_queue.push(std::move(value));
+#ifdef OSMIUM_DEBUG_QUEUE_SIZE
+                if (m_largest_size < m_queue.size()) {
+                    m_largest_size = m_queue.size();
+                }
+#endif
+                m_data_available.notify_one();
+            }
+
+            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
@@ -0,0 +1,159 @@
+#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
@@ -0,0 +1,87 @@
+#ifndef OSMIUM_THREAD_UTIL_HPP
+#define OSMIUM_THREAD_UTIL_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>
+
+#ifdef __linux__
+# include <sys/prctl.h>
+#endif
+
+namespace osmium {
+
+    namespace thread {
+
+        /**
+         * Check if the future resulted in an exception. This will re-throw
+         * the exception stored in the future if there was one. Otherwise it
+         * will just return.
+         */
+        template <class T>
+        inline void check_for_exception(std::future<T>& future) {
+            if (future.valid() && future.wait_for(std::chrono::seconds(0)) == std::future_status::ready) {
+                future.get();
+            }
+        }
+
+        /**
+         * Wait until the given future becomes ready. Will block if the future
+         * is not ready. Can be called more than once unless future.get().
+         */
+        template <class T>
+        inline void wait_until_done(std::future<T>& future) {
+            if (future.valid()) {
+                future.get();
+            }
+        }
+
+        /**
+         * 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_UTIL_HPP