Fix: Results of tasks run in thread pool were not checked

If a function run in the thread pool throws an exception, this exception
was never "collected", it was silently ignored. This commit fixes this
by wrapping the osmium::thread::Pool class in a class of our own that
keeps track of the futures returned by submit() and allows checking
them.

Author: joto

src/CMakeLists.txt

@@ -66,6 +66,7 @@ set(osm2pgsql_lib_SOURCES
   taginfo-impl.hpp
   taginfo.hpp
   tagtransform.hpp
+  thread-pool.hpp
   util.hpp
   wildcmp.hpp
   wkb.hpp

src/middle-pgsql.cpp

@@ -586,7 +586,7 @@ void middle_pgsql_t::commit()
 
 void middle_pgsql_t::flush() { m_db_copy.sync(); }
 
-void middle_pgsql_t::stop(osmium::thread::Pool &pool)
+void middle_pgsql_t::stop(thread_pool_t &pool)
 {
     m_cache.reset();
     if (m_out_options->flat_node_cache_enabled) {

src/middle-pgsql.hpp

@@ -58,7 +58,7 @@ struct middle_pgsql_t : public slim_middle_t
     middle_pgsql_t(options_t const *options);
 
     void start() override;
-    void stop(osmium::thread::Pool &pool) override;
+    void stop(thread_pool_t &pool) override;
     void analyze() override;
     void commit() override;
 

src/middle-ram.cpp

@@ -110,7 +110,7 @@ bool middle_ram_t::relation_get(osmid_t id,
     return true;
 }
 
-void middle_ram_t::stop(osmium::thread::Pool &)
+void middle_ram_t::stop(thread_pool_t &)
 {
     m_cache.reset();
     m_ways.clear();

src/middle-ram.hpp

@@ -95,7 +95,7 @@ struct middle_ram_t : public middle_t, public middle_query_t
     virtual ~middle_ram_t() noexcept = default;
 
     void start() override {}
-    void stop(osmium::thread::Pool &pool) override;
+    void stop(thread_pool_t &pool) override;
     void analyze() override {}
     void commit() override {}
 

src/middle.hpp

@@ -12,10 +12,9 @@
 #include <cstddef>
 #include <memory>
 
-#include <osmium/thread/pool.hpp>
-
 #include "osmtypes.hpp"
 #include "reprojection.hpp"
+#include "thread-pool.hpp"
 
 /**
  * Interface for returning information about raw OSM input data from a cache.
@@ -88,7 +87,7 @@ struct middle_t
     virtual ~middle_t() = 0;
 
     virtual void start() = 0;
-    virtual void stop(osmium::thread::Pool &pool) = 0;
+    virtual void stop(thread_pool_t &pool) = 0;
     virtual void analyze(void) = 0;
     virtual void commit(void) = 0;
 

src/osmdata.cpp

@@ -9,7 +9,6 @@
 #include <vector>
 
 #include <osmium/io/any_input.hpp>
-#include <osmium/thread/pool.hpp>
 #include <osmium/visitor.hpp>
 
 #include "db-copy.hpp"
@@ -19,6 +18,7 @@
 #include "options.hpp"
 #include "osmdata.hpp"
 #include "output.hpp"
+#include "thread-pool.hpp"
 #include "util.hpp"
 
 osmdata_t::osmdata_t(std::unique_ptr<dependency_manager_t> dependency_manager,
@@ -395,7 +395,7 @@ void osmdata_t::process_stage3() const
 {
     // All the intensive parts of this are long-running PostgreSQL commands.
     // They will be run in a thread pool.
-    osmium::thread::Pool pool{m_parallel_indexing ? m_num_procs : 1, 512};
+    thread_pool_t pool{m_parallel_indexing ? m_num_procs : 1};
 
     if (m_droptemp) {
         // When dropping middle tables, make sure they are gone before
@@ -415,9 +415,9 @@ void osmdata_t::process_stage3() const
         m_mid->stop(pool);
     }
 
-    // Waiting here for pool to execute all tasks.
-    // XXX If one of them has an error, all other will finish first,
-    //     which may take a long time.
+    // Waiting here for pool to execute all tasks. If one of them throws an
+    // exception, this will throw.
+    pool.check_for_exceptions();
 }
 
 void osmdata_t::stop() const

src/output-flex.cpp

@@ -1092,7 +1092,7 @@ void output_flex_t::sync()
     }
 }
 
-void output_flex_t::stop(osmium::thread::Pool *pool)
+void output_flex_t::stop(thread_pool_t *pool)
 {
     for (auto &table : m_table_connections) {
         pool->submit([&]() {

src/output-flex.hpp

@@ -112,7 +112,7 @@ class output_flex_t : public output_t
           std::shared_ptr<db_copy_thread_t> const &copy_thread) const override;
 
     void start() override;
-    void stop(osmium::thread::Pool *pool) override;
+    void stop(thread_pool_t *pool) override;
     void sync() override;
 
     void stage2_proc() override;

src/output-gazetteer.hpp

@@ -42,7 +42,7 @@ class output_gazetteer_t : public output_t
     }
 
     void start() override;
-    void stop(osmium::thread::Pool *) noexcept override {}
+    void stop(thread_pool_t *) noexcept override {}
     void sync() override;
 
     bool need_forward_dependencies() const noexcept override { return false; }