Merge pull request #1486 from joto/own-thread-pool

New thread pool implementation

Author: joto

.github/workflows/ci.yml

@@ -114,7 +114,7 @@ jobs:
       LUAJIT_OPTION: OFF
       POSTGRESQL_VERSION: 10
       POSTGIS_VERSION: 3
-      CPP_VERSION: 11
+      CPP_VERSION: 14
       BUILD_TYPE: Debug
 
     steps:
@@ -133,7 +133,7 @@ jobs:
       LUAJIT_OPTION: OFF
       POSTGRESQL_VERSION: 11
       POSTGIS_VERSION: 2.5
-      CPP_VERSION: 11
+      CPP_VERSION: 14
       BUILD_TYPE: Debug
 
     steps:

CMakeLists.txt

@@ -41,7 +41,7 @@ if (NOT CMAKE_BUILD_TYPE)
 endif()
 
 if (NOT "${CMAKE_CXX_STANDARD}")
-    set(CMAKE_CXX_STANDARD 11)
+    set(CMAKE_CXX_STANDARD 14)
 endif()
 message(STATUS "Building in C++${CMAKE_CXX_STANDARD} mode")
 set(CMAKE_CXX_EXTENSIONS OFF)
@@ -53,7 +53,7 @@ if (MSVC)
 else()
     add_definitions(-Wall)
     if (CMAKE_VERSION VERSION_LESS 3.1)
-        add_definitions(-std=c++11)
+        add_definitions(-std=c++14)
     endif()
 endif()
 

README.md

@@ -61,8 +61,7 @@ It also requires access to a database server running
 [PostGIS](http://www.postgis.net/) 2.2+.
 
 Make sure you have installed the development packages for the libraries
-mentioned in the requirements section and a C++ compiler which supports C++11.
-GCC 5 and later and Clang 3.5 and later are known to work.
+mentioned in the requirements section and a C++ compiler which supports C++14.
 
 To rebuild the included man page you'll need the [pandoc](https://pandoc.org/)
 tool.

src/CMakeLists.txt

@@ -31,6 +31,7 @@ set(osm2pgsql_lib_SOURCES
   taginfo.cpp
   tagtransform-c.cpp
   tagtransform.cpp
+  thread-pool.cpp
   util.cpp
   wildcmp.cpp
 )

src/flex-table.cpp

@@ -199,8 +199,6 @@ void table_connection_t::stop(bool updateable, bool append)
         return;
     }
 
-    util::timer_t timer;
-
     if (table().has_geom_column()) {
         if (table().geom_column().needs_isvalid()) {
             drop_geom_check_trigger(m_db_connection.get(), table().schema(),
@@ -272,9 +270,6 @@ void table_connection_t::stop(bool updateable, bool append)
     log_info("Analyzing table '{}'...", table().name());
     analyze_table(*m_db_connection, table().schema(), table().name());
 
-    log_info("All postprocessing on table '{}' done in {}.", table().name(),
-             util::human_readable_duration(timer.stop()));
-
     teardown();
 }
 
@@ -319,3 +314,10 @@ void table_connection_t::delete_rows_with(osmium::item_type type, osmid_t id)
     }
     m_copy_mgr.delete_object(type_to_char(type)[0], id);
 }
+
+void table_connection_t::task_wait()
+{
+    auto const run_time = m_task_result.wait();
+    log_info("All postprocessing on table '{}' done in {}.", table().name(),
+             util::human_readable_duration(run_time));
+}

src/flex-table.hpp

@@ -14,6 +14,7 @@
 #include "flex-table-column.hpp"
 #include "osmium-builder.hpp"
 #include "pgsql.hpp"
+#include "thread-pool.hpp"
 
 #include <osmium/osm/item_type.hpp>
 
@@ -255,6 +256,13 @@ class table_connection_t
 
     geom::osmium_builder_t *get_builder() { return &m_builder; }
 
+    void task_set(std::future<std::chrono::milliseconds> &&future)
+    {
+        m_task_result.set(std::move(future));
+    }
+
+    void task_wait();
+
 private:
     geom::osmium_builder_t m_builder;
 
@@ -271,6 +279,8 @@ class table_connection_t
     /// The connection to the database server.
     std::unique_ptr<pg_conn_t> m_db_connection;
 
+    task_result_t m_task_result;
+
     /// Has the Id index already been created?
     bool m_id_index_created = false;
 

src/logging.cpp

@@ -9,6 +9,8 @@
 
 #include "logging.hpp"
 
+thread_local unsigned int this_thread_num = 0;
+
 /// Global logger singleton
 logger the_logger{};
 

src/logging.hpp

@@ -20,6 +20,8 @@
 #include <cstdio>
 #include <utility>
 
+extern thread_local unsigned int this_thread_num;
+
 enum class log_level
 {
     debug = 1,
@@ -49,6 +51,10 @@ class logger
         std::string str = fmt::format("{:%Y-%m-%d %H:%M:%S}  ",
                                       fmt::localtime(std::time(nullptr)));
 
+        if (m_current_level == log_level::debug) {
+            str += fmt::format(ts, "[{}] ", this_thread_num);
+        }
+
         if (prefix) {
             str += fmt::format(ts, "{}: ", prefix);
         }

src/middle-pgsql.cpp

@@ -98,17 +98,12 @@ void middle_pgsql_t::table_desc::build_index(std::string const &conninfo) const
         return;
     }
 
-    util::timer_t timer;
-
     // Use a temporary connection here because we might run in a separate
     // thread context.
     pg_conn_t db_connection{conninfo};
 
     log_info("Building index on table '{}'", name());
     db_connection.exec(m_create_fw_dep_indexes);
-
-    log_info("Done postprocessing on table '{}' in {}", name(),
-             util::human_readable_duration(timer.stop()));
 }
 
 namespace {
@@ -644,13 +639,22 @@ void middle_pgsql_t::stop(thread_pool_t &pool)
     } else if (!m_options->append) {
         // Building the indexes takes time, so do it asynchronously.
         for (auto &table : m_tables) {
-            pool.submit(std::bind(&middle_pgsql_t::table_desc::build_index,
-                                  &table,
-                                  m_options->database_options.conninfo()));
+            table.task_set(pool.submit(
+                std::bind(&middle_pgsql_t::table_desc::build_index, &table,
+                          m_options->database_options.conninfo())));
         }
     }
 }
 
+void middle_pgsql_t::wait()
+{
+    for (auto &table : m_tables) {
+        auto const run_time = table.task_wait();
+        log_info("Done postprocessing on table '{}' in {}", table.name(),
+                 util::human_readable_duration(run_time));
+    }
+}
+
 static table_sql sql_for_nodes(bool create_table) noexcept
 {
     table_sql sql{};

src/middle-pgsql.hpp

@@ -74,6 +74,8 @@ struct middle_pgsql_t : public middle_t
     void start() override;
     void stop(thread_pool_t &pool) override;
 
+    void wait() override;
+
     void node(osmium::Node const &node) override;
     void way(osmium::Way const &way) override;
     void relation(osmium::Relation const &rel) override;
@@ -115,8 +117,16 @@ struct middle_pgsql_t : public middle_t
         std::string m_prepare_fw_dep_lookups;
         std::string m_create_fw_dep_indexes;
 
+        void task_set(std::future<std::chrono::milliseconds> &&future)
+        {
+            m_task_result.set(std::move(future));
+        }
+
+        std::chrono::milliseconds task_wait() { return m_task_result.wait(); }
+
     private:
         std::shared_ptr<db_target_descr_t> m_copy_target;
+        task_result_t m_task_result;
     };
 
     std::shared_ptr<middle_query_t> get_query_instance() override;