Open thread pool earlier and make it a member of middle_t and output_t

This way we can access it from everywhere from the start on. This is not
used yet, but can be used to run processing asynchronously before we are
getting to the stop() phase.

Author: joto

src/dependency-manager.hpp

@@ -17,7 +17,7 @@
 #include <cassert>
 #include <memory>
 
-struct middle_t;
+class middle_t;
 
 /**
  * The job of the dependency manager is to keep track of the dependencies

src/middle-pgsql.cpp

@@ -622,7 +622,7 @@ void middle_pgsql_t::start()
     }
 }
 
-void middle_pgsql_t::stop(thread_pool_t &pool)
+void middle_pgsql_t::stop()
 {
     m_cache.reset();
     if (!m_options->flat_node_file.empty()) {
@@ -638,7 +638,7 @@ 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) {
-            table.task_set(pool.submit(
+            table.task_set(thread_pool().submit(
                 std::bind(&middle_pgsql_t::table_desc::build_index, &table,
                           m_options->database_options.conninfo())));
         }
@@ -779,8 +779,9 @@ static bool check_bucket_index(pg_conn_t *db_connection,
     return res.num_tuples() > 0;
 }
 
-middle_pgsql_t::middle_pgsql_t(options_t const *options)
-: m_options(options),
+middle_pgsql_t::middle_pgsql_t(std::shared_ptr<thread_pool_t> thread_pool,
+                               options_t const *options)
+: middle_t(std::move(thread_pool)), m_options(options),
   m_cache(new node_locations_t{static_cast<std::size_t>(options->cache) *
                                1024UL * 1024UL}),
   m_db_connection(m_options->database_options.conninfo()),

src/middle-pgsql.hpp

@@ -69,10 +69,11 @@ struct table_sql {
 
 struct middle_pgsql_t : public middle_t
 {
-    explicit middle_pgsql_t(options_t const *options);
+    middle_pgsql_t(std::shared_ptr<thread_pool_t> thread_pool,
+                   options_t const *options);
 
     void start() override;
-    void stop(thread_pool_t &pool) override;
+    void stop() override;
 
     void wait() override;
 

src/middle-ram.cpp

@@ -27,7 +27,9 @@
 #include <limits>
 #include <memory>
 
-middle_ram_t::middle_ram_t(options_t const *options)
+middle_ram_t::middle_ram_t(std::shared_ptr<thread_pool_t> thread_pool,
+                           options_t const *options)
+: middle_t(std::move(thread_pool))
 {
     assert(options);
 
@@ -60,7 +62,7 @@ void middle_ram_t::set_requirements(output_requirements const &requirements)
     log_debug("  relations: {}", m_store_options.relations);
 }
 
-void middle_ram_t::stop(thread_pool_t &)
+void middle_ram_t::stop()
 {
     auto const mbyte = 1024 * 1024;
 

src/middle-ram.hpp

@@ -41,12 +41,13 @@ class options_t;
 class middle_ram_t : public middle_t, public middle_query_t
 {
 public:
-    explicit middle_ram_t(options_t const *options);
+    middle_ram_t(std::shared_ptr<thread_pool_t> thread_pool,
+                 options_t const *options);
 
     ~middle_ram_t() noexcept override = default;
 
     void start() override {}
-    void stop(thread_pool_t &) override;
+    void stop() override;
 
     void node(osmium::Node const &node) override;
     void way(osmium::Way const &way) override;

src/middle.cpp

@@ -12,12 +12,15 @@
 #include "middle.hpp"
 #include "options.hpp"
 
-std::shared_ptr<middle_t> create_middle(options_t const &options)
+std::shared_ptr<middle_t>
+create_middle(std::shared_ptr<thread_pool_t> thread_pool,
+              options_t const &options)
 {
     if (options.slim) {
-        return std::make_shared<middle_pgsql_t>(&options);
+        return std::make_shared<middle_pgsql_t>(std::move(thread_pool),
+                                                &options);
     }
 
-    return std::make_shared<middle_ram_t>(&options);
+    return std::make_shared<middle_ram_t>(std::move(thread_pool), &options);
 }
 

src/middle.hpp

@@ -81,12 +81,17 @@ inline middle_query_t::~middle_query_t() = default;
  * Interface for storing "raw" OSM data in an intermediate object store and
  * getting it back.
  */
-struct middle_t
+class middle_t
 {
+public:
+    explicit middle_t(std::shared_ptr<thread_pool_t> thread_pool)
+    : m_thread_pool(std::move(thread_pool))
+    {}
+
     virtual ~middle_t() = 0;
 
     virtual void start() = 0;
-    virtual void stop(thread_pool_t &pool) = 0;
+    virtual void stop() = 0;
 
     virtual void wait() {}
 
@@ -115,11 +120,23 @@ struct middle_t
     virtual std::shared_ptr<middle_query_t> get_query_instance() = 0;
 
     virtual void set_requirements(output_requirements const &) {}
-};
+
+protected:
+    thread_pool_t &thread_pool() const noexcept
+    {
+        assert(m_thread_pool);
+        return *m_thread_pool;
+    }
+
+private:
+    std::shared_ptr<thread_pool_t> m_thread_pool;
+}; // class middle_t
 
 inline middle_t::~middle_t() = default;
 
 /// Factory function: Instantiate the middle based on the command line options.
-std::shared_ptr<middle_t> create_middle(options_t const &options);
+std::shared_ptr<middle_t>
+create_middle(std::shared_ptr<thread_pool_t> thread_pool,
+              options_t const &options);
 
 #endif // OSM2PGSQL_MIDDLE_HPP

src/osm2pgsql.cpp

@@ -29,11 +29,15 @@ static void run(options_t const &options)
     auto const files = prepare_input_files(
         options.input_files, options.input_format, options.append);
 
-    auto middle = create_middle(options);
+    auto thread_pool = std::make_shared<thread_pool_t>(
+        options.parallel_indexing ? options.num_procs : 1U);
+    log_debug("Started pool with {} threads.", thread_pool->num_threads());
+
+    auto middle = create_middle(thread_pool, options);
     middle->start();
 
-    auto output =
-        output_t::create_output(middle->get_query_instance(), options);
+    auto output = output_t::create_output(middle->get_query_instance(),
+                                          thread_pool, options);
 
     middle->set_requirements(output->get_requirements());
 

src/osmdata.cpp

@@ -33,7 +33,6 @@ osmdata_t::osmdata_t(std::unique_ptr<dependency_manager_t> dependency_manager,
   m_output(std::move(output)), m_conninfo(options.database_options.conninfo()),
   m_bbox(options.bbox), m_num_procs(options.num_procs),
   m_append(options.append), m_droptemp(options.droptemp),
-  m_parallel_indexing(options.parallel_indexing),
   m_with_extra_attrs(options.extra_attributes),
   m_with_forward_dependencies(options.with_forward_dependencies)
 {
@@ -373,27 +372,20 @@ void osmdata_t::reprocess_marked() const { m_output->reprocess_marked(); }
 
 void osmdata_t::postprocess_database() const
 {
-    unsigned int const num_threads = m_parallel_indexing ? m_num_procs : 1U;
-    log_debug("Starting pool with {} threads.", num_threads);
-
-    // All the intensive parts of this are long-running PostgreSQL commands.
-    // They will be run in a thread pool.
-    thread_pool_t pool{num_threads};
-
     if (m_droptemp) {
         // When dropping middle tables, make sure they are gone before
         // indexing starts.
-        m_mid->stop(pool);
+        m_mid->stop();
     }
 
-    m_output->stop(&pool);
+    m_output->stop();
 
     if (!m_droptemp) {
         // When keeping middle tables, there is quite a large index created
         // which is better done after the output tables have been copied.
         // Note that --disable-parallel-indexing needs to be used to really
         // force the order.
-        m_mid->stop(pool);
+        m_mid->stop();
     }
 
     // Waiting here for pool to execute all tasks.

src/osmdata.hpp

@@ -27,9 +27,9 @@
 #include "dependency-manager.hpp"
 #include "osmtypes.hpp"
 
+class middle_t;
 class options_t;
 class output_t;
-struct middle_t;
 
 /**
  * This class guides the processing of the OSM data through its multiple
@@ -101,7 +101,6 @@ class osmdata_t : public osmium::handler::Handler
     unsigned int m_num_procs;
     bool m_append;
     bool m_droptemp;
-    bool m_parallel_indexing;
     bool m_with_extra_attrs;
     bool m_with_forward_dependencies;
 };