Merge pull request #889 from lonvia/flush-between-type-changes

middle-pgsql: Flush between type changes

Author: lonvia

middle-pgsql.cpp

@@ -14,7 +14,6 @@
 #include "node-persistent-cache.hpp"
 #include "id-tracker.hpp"
 #include <memory>
-#include <vector>
 
 struct middle_pgsql_t : public slim_middle_t {
     middle_pgsql_t();
@@ -43,6 +42,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;
 
@@ -82,6 +83,14 @@ struct middle_pgsql_t : public slim_middle_t {
     get_query_instance(std::shared_ptr<middle_t> const &mid) const override;
 
 private:
+    enum middle_tables
+    {
+        NODE_TABLE = 0,
+        WAY_TABLE,
+        REL_TABLE,
+        NUM_TABLES
+    };
+
     void pgsql_stop_one(table_desc *table);
 
     /**
@@ -92,9 +101,7 @@ struct middle_pgsql_t : public slim_middle_t {
     size_t local_nodes_get_list(osmium::WayNodeList *nodes) const;
     void local_nodes_delete(osmid_t osm_id);
 
-    std::vector<table_desc> tables;
-    int num_tables;
-    table_desc *node_table, *way_table, *rel_table;
+    table_desc tables[NUM_TABLES];
 
     bool append;
     bool mark_pending;

middle-pgsql.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-ram.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;

middle.hpp

@@ -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.cpp

@@ -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);

osmdata.hpp

@@ -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.cpp

@@ -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

parse-osmium.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  { }