Merge pull request #1508 from joto/thread-pool-earlier

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

Author: lonvia

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;
 };