Merge pull request #1198 from joto/split-processing

Split up processing into functions for each stage

Author: lonvia

src/osmdata.cpp

@@ -367,20 +367,9 @@ class multithreaded_processor
 
 } // anonymous namespace
 
-void osmdata_t::stop() const
+void osmdata_t::process_stage1b() const
 {
-    /* Commit the transactions, so that multiple processes can
-     * access the data simultaneously to process the rest in parallel
-     * as well as see the newly created tables.
-     */
-    m_mid->commit();
-    for (auto &out : m_outs) {
-        out->sync();
-    }
-
-    // In append mode there might be dependent objects pending that we
-    // need to process.
-    if (m_append && m_dependency_manager->has_pending()) {
+    if (m_dependency_manager->has_pending()) {
         multithreaded_processor proc{m_conninfo, m_mid, m_outs,
                                      (std::size_t)m_num_procs};
 
@@ -389,36 +378,60 @@ void osmdata_t::stop() const
             m_dependency_manager->get_pending_relation_ids());
         proc.merge_expire_trees();
     }
+}
 
+void osmdata_t::process_stage2() const
+{
     for (auto &out : m_outs) {
         out->stage2_proc();
     }
+}
 
-    // Clustering, index creation, and cleanup.
-    // All the intensive parts of this are long-running PostgreSQL commands
-    {
-        osmium::thread::Pool pool{m_parallel_indexing ? m_num_procs : 1, 512};
+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};
+
+    if (m_droptemp) {
+        // When dropping middle tables, make sure they are gone before
+        // indexing starts.
+        m_mid->stop(pool);
+    }
 
-        if (m_droptemp) {
-            // When dropping middle tables, make sure they are gone before
-            // indexing starts.
-            m_mid->stop(pool);
-        }
+    for (auto &out : m_outs) {
+        out->stop(&pool);
+    }
 
-        for (auto &out : m_outs) {
-            out->stop(&pool);
-        }
+    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);
+    }
 
-        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);
-        }
+    // 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.
+}
+
+void osmdata_t::stop() const
+{
+    /* Commit the transactions, so that multiple processes can
+     * access the data simultaneously to process the rest in parallel
+     * as well as see the newly created tables.
+     */
+    m_mid->commit();
+    for (auto &out : m_outs) {
+        out->sync();
+    }
 
-        // 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.
+    if (m_append) {
+        process_stage1b();
     }
+
+    process_stage2();
+
+    process_stage3();
 }

src/osmdata.hpp

@@ -37,6 +37,23 @@ class osmdata_t
     void relation_delete(osmid_t id) const;
 
 private:
+    /**
+     * Run stage 1b processing: Process dependent objects.
+     * In append mode we need to process dependent objects that were marked
+     * earlier.
+     */
+    void process_stage1b() const;
+
+    /**
+     * Run stage 2 processing: Process objects marked in stage 1 (if any).
+     */
+    void process_stage2() const;
+
+    /**
+     * Run stage 3 processing: Clustering and index creation.
+     */
+    void process_stage3() const;
+
     slim_middle_t &slim_middle() const noexcept;
 
     std::unique_ptr<dependency_manager_t> m_dependency_manager;