Merge pull request #1180 from joto/output-commit-in-thread

Move commits of outputs into threads

Author: lonvia

src/flex-table.hpp

@@ -226,7 +226,7 @@ class table_connection_t
 
     pg_result_t get_geom_by_id(osmium::item_type type, osmid_t id) const;
 
-    void commit() { m_copy_mgr.sync(); }
+    void sync() { m_copy_mgr.sync(); }
 
     void new_line() { m_copy_mgr.new_line(m_target); }
 

src/osmdata.cpp

@@ -270,6 +270,11 @@ class multithreaded_processor
                 }
             }
         }
+        for (auto const &output : outputs) {
+            if (output) {
+                output->sync();
+            }
+        }
     }
 
     /**
@@ -286,6 +291,11 @@ class multithreaded_processor
                 }
             }
         }
+        for (auto const &output : outputs) {
+            if (output) {
+                output->sync();
+            }
+        }
     }
 
     /// Runs in a worker thread: Update progress display once per second.
@@ -335,14 +345,6 @@ class multithreaded_processor
             }
         }
 
-        for (auto const &clone : m_clones) {
-            for (auto const &clone_output : clone) {
-                if (clone_output) {
-                    clone_output->commit();
-                }
-            }
-        }
-
         timer.stop();
 
         fmt::print(stderr, "\rFinished processing {} {}s in {} s\n\n",
@@ -376,8 +378,7 @@ void osmdata_t::stop() const
      */
     m_mid->commit();
     for (auto &out : m_outs) {
-        //TODO: each of the outs can be in parallel
-        out->commit();
+        out->sync();
     }
 
     // should be the same for all outputs

src/output-flex.cpp

@@ -1053,10 +1053,10 @@ void output_flex_t::pending_relation(osmid_t id)
     m_rels_buffer.clear();
 }
 
-void output_flex_t::commit()
+void output_flex_t::sync()
 {
     for (auto &table : m_table_connections) {
-        table.commit();
+        table.sync();
     }
 }
 

src/output-flex.hpp

@@ -57,7 +57,7 @@ class output_flex_t : public output_t
 
     void start() override;
     void stop(osmium::thread::Pool *pool) override;
-    void commit() override;
+    void sync() override;
 
     void stage2_proc() override;
 

src/output-gazetteer.cpp

@@ -67,7 +67,7 @@ void output_gazetteer_t::start()
     }
 }
 
-void output_gazetteer_t::commit() { m_copy.sync(); }
+void output_gazetteer_t::sync() { m_copy.sync(); }
 
 void output_gazetteer_t::node_add(osmium::Node const &node)
 {

src/output-gazetteer.hpp

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

src/output-multi.cpp

@@ -97,7 +97,7 @@ void output_multi_t::stop(osmium::thread::Pool *pool)
     }
 }
 
-void output_multi_t::commit() { m_table->commit(); }
+void output_multi_t::sync() { m_table->sync(); }
 
 void output_multi_t::node_add(osmium::Node const &node)
 {

src/output-multi.hpp

@@ -44,7 +44,7 @@ class output_multi_t : public output_t
 
     void start() override;
     void stop(osmium::thread::Pool *pool) override;
-    void commit() override;
+    void sync() override;
 
     void pending_way(osmid_t id) override;
     void pending_relation(osmid_t id) override;

src/output-null.hpp

@@ -19,7 +19,7 @@ class output_null_t : public output_t
 
     void start() override {}
     void stop(osmium::thread::Pool * /*pool*/) override {}
-    void commit() override {}
+    void sync() override {}
     void cleanup() {}
 
     bool need_forward_dependencies() const noexcept override { return false; }

src/output-pgsql.cpp

@@ -102,10 +102,10 @@ void output_pgsql_t::pending_relation(osmid_t id)
     }
 }
 
-void output_pgsql_t::commit()
+void output_pgsql_t::sync()
 {
     for (auto const &t : m_tables) {
-        t->commit();
+        t->sync();
     }
 }