Flush and close COPYs after nodes, ways, and relations in flex output

Most output tables are only filled by either nodes or ways or relations.
So it makes sense to close COPY operations after the last
node/way/relation was found. Before this change active COPY operations
could sometimes be open for a long time.

Author: joto

src/db-copy-mgr.hpp

@@ -262,10 +262,12 @@ class db_copy_mgr_t
 
     void flush()
     {
-        // finish any ongoing copy operations
+        // flush current buffer if there is one
         if (m_current) {
             m_processor->add_buffer(std::move(m_current));
         }
+        // close any ongoing copy operations
+        m_processor->end_copy();
     }
 
     /**

src/db-copy.cpp

@@ -105,6 +105,11 @@ void db_copy_thread_t::add_buffer(std::unique_ptr<db_cmd_t> &&buffer)
     m_shared.queue_cond.notify_one();
 }
 
+void db_copy_thread_t::end_copy()
+{
+    add_buffer(std::make_unique<db_cmd_end_copy_t>());
+}
+
 void db_copy_thread_t::sync_and_wait()
 {
     std::promise<void> barrier;
@@ -157,6 +162,9 @@ void db_copy_thread_t::thread_t::operator()()
             case db_cmd_t::Cmd_copy:
                 write_to_db(static_cast<db_cmd_copy_t *>(item.get()));
                 break;
+            case db_cmd_t::Cmd_end_copy:
+                finish_copy();
+                break;
             case db_cmd_t::Cmd_sync:
                 finish_copy();
                 static_cast<db_cmd_sync_t *>(item.get())->barrier.set_value();

src/db-copy.hpp

@@ -144,8 +144,9 @@ class db_cmd_t
 public:
     enum cmd_t
     {
-        Cmd_copy, ///< Copy buffer content into given target.
-        Cmd_sync, ///< Synchronize with parent.
+        Cmd_copy,     ///< Copy buffer content into given target.
+        Cmd_end_copy, ///< End COPY command.
+        Cmd_sync,     ///< Synchronize with parent.
         Cmd_finish
     };
 
@@ -229,6 +230,11 @@ class db_cmd_copy_delete_t : public db_cmd_copy_t
     DELETER m_deleter;
 };
 
+struct db_cmd_end_copy_t : public db_cmd_t
+{
+    db_cmd_end_copy_t() : db_cmd_t(db_cmd_t::Cmd_end_copy) {}
+};
+
 struct db_cmd_sync_t : public db_cmd_t
 {
     std::promise<void> barrier;
@@ -264,6 +270,9 @@ class db_copy_thread_t
      */
     void add_buffer(std::unique_ptr<db_cmd_t> &&buffer);
 
+    /// Close COPY if one is open
+    void end_copy();
+
     /**
      * Send sync command and wait for the notification.
      */

src/osmdata.cpp

@@ -191,6 +191,7 @@ void osmdata_t::relation(osmium::Relation const &rel)
 void osmdata_t::after_relations()
 {
     m_mid->after_relations();
+    m_output->after_relations();
 
     if (m_append) {
         // Remove ids from changed relations in the input data from

src/output-flex.cpp

@@ -931,19 +931,18 @@ void output_flex_t::sync()
     }
 }
 
-void output_flex_t::after_nodes()
+static void flush_tables(std::vector<table_connection_t> &table_connections)
 {
-    for (auto &table : m_table_connections) {
+    for (auto &table : table_connections) {
         table.flush();
     }
 }
 
-void output_flex_t::after_ways()
-{
-    for (auto &table : m_table_connections) {
-        table.flush();
-    }
-}
+void output_flex_t::after_nodes() { flush_tables(m_table_connections); }
+
+void output_flex_t::after_ways() { flush_tables(m_table_connections); }
+
+void output_flex_t::after_relations() { flush_tables(m_table_connections); }
 
 void output_flex_t::stop()
 {

src/output-flex.hpp

@@ -123,6 +123,7 @@ class output_flex_t : public output_t
 
     void after_nodes() override;
     void after_ways() override;
+    void after_relations() override;
 
     void wait() override;
 

src/output.hpp

@@ -65,6 +65,7 @@ class output_t
 
     virtual void after_nodes() {}
     virtual void after_ways() {}
+    virtual void after_relations() {}
 
     virtual void wait() {}