Merge pull request #2186 from joto/flush-and-sync-tables

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

Author: lonvia

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() {}