Introduce flush when type of import object changes

This replaces the endCopy operation in the middle_t
read functions. These functions are const and never
should have called a function that changes state.
Instead of falling out of copy then, all copy operations
are finished, when processing of an OSM type is done.
Read functions should never be called when actually
processing the same table, so the one-time end-copy
should be sufficient.

Author: lonvia

middle-pgsql.cpp

@@ -311,7 +311,8 @@ size_t middle_pgsql_t::local_nodes_get_list(osmium::WayNodeList *nodes) const
     // get any remaining nodes from the DB
     buffer[buffer.size() - 1] = '}';
 
-    pgsql_endCopy(node_table);
+    // Nodes must have been written back at this point.
+    assert(node_table->copyMode == 0);
 
     PGconn *sql_conn = node_table->sql_conn;
 
@@ -462,8 +463,8 @@ bool middle_pgsql_t::ways_get(osmid_t id, osmium::memory::Buffer &buffer) const
     char const *paramValues[1];
     PGconn *sql_conn = way_table->sql_conn;
 
-    // Make sure we're out of copy mode */
-    pgsql_endCopy(way_table);
+    // Make sure we're out of copy mode
+    assert(way_table->copyMode == 0);
 
     char tmp[16];
     snprintf(tmp, sizeof(tmp), "%" PRIdOSMID, id);
@@ -511,7 +512,8 @@ size_t middle_pgsql_t::rel_way_members_get(osmium::Relation const &rel,
     // replace last , with } to complete list of ids
     tmp2[tmp2.length() - 1] = '}'; 
 
-    pgsql_endCopy(way_table);
+    // Make sures all ways have been written back.
+    assert(way_table->copyMode == 0);
 
     PGconn *sql_conn = way_table->sql_conn;
 
@@ -696,8 +698,8 @@ bool middle_pgsql_t::relations_get(osmid_t id, osmium::memory::Buffer &buffer) c
     PGconn *sql_conn = rel_table->sql_conn;
     taglist_t member_temp;
 
-    // Make sure we're out of copy mode */
-    pgsql_endCopy(rel_table);
+    // Make sure we're out of copy mode
+    assert(rel_table->copyMode == 0);
 
     snprintf(tmp, sizeof(tmp), "%" PRIdOSMID, id);
     paramValues[0] = tmp;
@@ -791,7 +793,7 @@ idlist_t middle_pgsql_t::relations_using_way(osmid_t way_id) const
     char const *paramValues[1];
     char buffer[64];
     // Make sure we're out of copy mode */
-    pgsql_endCopy( rel_table );
+    assert(rel_table->copyMode == 0);
 
     sprintf(buffer, "%" PRIdOSMID, way_id);
     paramValues[0] = buffer;
@@ -1029,6 +1031,13 @@ void middle_pgsql_t::commit(void) {
     }
 }
 
+void middle_pgsql_t::flush()
+{
+    for (auto &table : tables) {
+        pgsql_endCopy(&table);
+    }
+}
+
 void middle_pgsql_t::pgsql_stop_one(table_desc *table)
 {
     time_t start, end;

middle-pgsql.hpp

@@ -43,6 +43,8 @@ struct middle_pgsql_t : public slim_middle_t {
     void relations_delete(osmid_t id) override;
     void relation_changed(osmid_t id) override;
 
+    void flush() override;
+
     void iterate_ways(middle_t::pending_processor& pf) override;
     void iterate_relations(pending_processor& pf) override;
 

middle-ram.hpp

@@ -107,6 +107,8 @@ struct middle_ram_t : public middle_t {
     int relations_delete(osmid_t id);
     int relation_changed(osmid_t id);
 
+    void flush() override {}
+
     idlist_t relations_using_way(osmid_t way_id) const override;
 
     void iterate_ways(middle_t::pending_processor& pf) override;

middle.hpp

@@ -92,6 +92,9 @@ struct middle_t : public middle_query_t {
     virtual void ways_set(osmium::Way const &way) = 0;
     virtual void relations_set(osmium::Relation const &rel) = 0;
 
+    /// Write all pending data to permanent storage.
+    virtual void flush() = 0;
+
     struct pending_processor {
         virtual ~pending_processor() {}
         virtual void enqueue_ways(osmid_t id) = 0;

osmdata.cpp

@@ -180,6 +180,11 @@ void osmdata_t::start() {
     mid->start(outs[0]->get_options());
 }
 
+void osmdata_t::type_changed()
+{
+    mid->flush();
+}
+
 namespace {
 
 //TODO: have the main thread using the main middle to query the middle for batches of ways (configurable number)

osmdata.hpp

@@ -14,7 +14,8 @@ class output_t;
 struct middle_t;
 class reprojection;
 
-class osmdata_t {
+class osmdata_t
+{
 public:
     osmdata_t(std::shared_ptr<middle_t> mid_,
               std::shared_ptr<output_t> const &out_,
@@ -25,6 +26,7 @@ class osmdata_t {
     ~osmdata_t();
 
     void start();
+    void type_changed();
     void stop();
 
     int node_add(osmium::Node const &node);

parse-osmium.cpp

@@ -124,13 +124,19 @@ void parse_osmium_t::stream_file(const std::string &filename, const std::string
 
     fprintf(stderr, "Using %s parser.\n", osmium::io::as_string(infile.format()));
 
+    m_type = osmium::item_type::node;
     osmium::io::Reader reader(infile);
     osmium::apply(reader, *this);
     reader.close();
 }
 
 void parse_osmium_t::node(osmium::Node const &node)
 {
+    if (m_type != osmium::item_type::node) {
+        m_type = osmium::item_type::node;
+        m_data->type_changed();
+    }
+
     if (node.deleted()) {
         m_data->node_delete(node.id());
     } else {
@@ -159,6 +165,11 @@ void parse_osmium_t::node(osmium::Node const &node)
 
 void parse_osmium_t::way(osmium::Way& way)
 {
+    if (m_type != osmium::item_type::way) {
+        m_type = osmium::item_type::way;
+        m_data->type_changed();
+    }
+
     if (way.deleted()) {
         m_data->way_delete(way.id());
     } else {
@@ -173,6 +184,11 @@ void parse_osmium_t::way(osmium::Way& way)
 
 void parse_osmium_t::relation(osmium::Relation const &rel)
 {
+    if (m_type != osmium::item_type::relation) {
+        m_type = osmium::item_type::relation;
+        m_data->type_changed();
+    }
+
     if (rel.deleted()) {
         m_data->relation_delete(rel.id());
     } else {

parse-osmium.hpp

@@ -129,6 +129,8 @@ class parse_osmium_t: public osmium::handler::Handler
     bool m_append;
     boost::optional<osmium::Box> m_bbox;
     parse_stats_t m_stats;
+    // Current type being parsed.
+    osmium::item_type m_type;
 };
 
 #endif

tests/mockups.hpp

@@ -9,6 +9,7 @@ struct dummy_middle_t : public middle_t {
 
     void start(const options_t *) override { }
     void stop(osmium::thread::Pool &) override {}
+    void flush() override {}
     void cleanup(void) { }
     void analyze(void) override  { }
     void commit(void) override  { }
@@ -46,6 +47,7 @@ struct dummy_slim_middle_t : public slim_middle_t {
 
     void start(const options_t *) override  { }
     void stop(osmium::thread::Pool &) override {}
+    void flush() override {}
     void cleanup(void) { }
     void analyze(void) override  { }
     void commit(void) override  { }