New thread pool implementation

Change the thread pool implementation giving us several new features:
* Threads are now numbered and when debug logging is used, the thread
  number is reported in each log line.
* Each task running in the thread pool is timed. This allows for
  better logging where our time is spent.
* When submitting a task we now return a future that can be used
  to access the running time of the thread. The code in the main
  thread can also wait for this future to make sure the background
  task is done.

This switches the minimum C++ version needed from 11 to 14 because we
need lambda capture by move.

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;