Merge pull request #1149 from joto/refactor-dependency-tracking-step2

Refactor dependency tracking step2

Author: lonvia

src/middle-pgsql.cpp

@@ -321,14 +321,15 @@ void middle_pgsql_t::node_changed(osmid_t osm_id)
         return;
     }
 
-    //keep track of whatever ways and rels these nodes intersect
+    // Find all ways referencing this node and mark them as pending.
     auto res = exec_prepared("mark_ways_by_node", osm_id);
     for (int i = 0; i < res.num_tuples(); ++i) {
         osmid_t const marked = osmium::string_to_object_id(res.get_value(i, 0));
+        way_changed(marked);
         m_ways_pending_tracker->mark(marked);
     }
 
-    //do the rels too
+    // Find all relations referencing this node and mark them as pending.
     res = exec_prepared("mark_rels_by_node", osm_id);
     for (int i = 0; i < res.num_tuples(); ++i) {
         osmid_t const marked = osmium::string_to_object_id(res.get_value(i, 0));
@@ -453,6 +454,11 @@ void middle_pgsql_t::iterate_ways(middle_t::pending_processor &pf)
 void middle_pgsql_t::way_changed(osmid_t osm_id)
 {
     assert(m_append);
+
+    if (m_ways_pending_tracker->is_marked(osm_id)) {
+        return;
+    }
+
     //keep track of whatever rels this way intersects
     auto const res = exec_prepared("mark_rels_by_way", osm_id);
     for (int i = 0; i < res.num_tuples(); ++i) {
@@ -548,8 +554,6 @@ void middle_pgsql_t::iterate_relations(pending_processor &pf)
     while (id_tracker::is_valid(id = m_rels_pending_tracker->pop_mark())) {
         pf.enqueue_relations(id);
     }
-    // in case we had higher ones than the middle
-    pf.enqueue_relations(id_tracker::max());
 
     //let the threads work on them
     pf.process_relations();
@@ -748,8 +752,9 @@ static table_sql sql_for_relations() noexcept
                         "      AND parts[way_off+1:rel_off] && ARRAY[$1];\n";
 
     sql.prepare_mark = "PREPARE mark_rels_by_node(int8) AS"
-                       "  SELECT id FROM {prefix}_ways"
-                       "    WHERE nodes && ARRAY[$1];\n"
+                       "  SELECT id FROM {prefix}_rels"
+                       "    WHERE parts && ARRAY[$1]"
+                       "      AND parts[1:way_off] && ARRAY[$1];\n"
                        "PREPARE mark_rels_by_way(int8) AS"
                        "  SELECT id FROM {prefix}_rels"
                        "    WHERE parts && ARRAY[$1]"

src/osmdata.cpp

@@ -171,7 +171,7 @@ struct pending_threaded_processor : public middle_t::pending_processor
     using output_vec_t = std::vector<std::shared_ptr<output_t>>;
 
     static void do_jobs(output_vec_t const &outputs, pending_queue_t &queue,
-                        size_t &ids_done, std::mutex &mutex, int append,
+                        size_t &ids_done, std::mutex &mutex, bool append,
                         bool ways)
     {
         while (true) {
@@ -297,27 +297,20 @@ struct pending_threaded_processor : public middle_t::pending_processor
         ids_queued = 0;
         ids_done = 0;
 
-        //collect all the new rels that became pending from each
-        //output in each thread back to their respective main outputs
         for (auto const &clone : clones) {
-            //for each clone/original output
-            for (output_vec_t::const_iterator original_output = outs.begin(),
-                                              clone_output = clone.begin();
-                 original_output != outs.end() && clone_output != clone.end();
-                 ++original_output, ++clone_output) {
-                //done copying ways for now
-                clone_output->get()->commit();
-                //merge the pending from this threads copy of output back
-                original_output->get()->merge_pending_relations(
-                    clone_output->get());
+            for (auto const &clone_output : clone) {
+                clone_output.get()->commit();
             }
         }
     }
 
     void enqueue_relations(osmid_t id) override
     {
         for (size_t i = 0; i < outs.size(); ++i) {
-            outs[i]->enqueue_relations(queue, id, i, ids_queued);
+            if (outs[i]->need_forward_dependencies()) {
+                queue.emplace(id, i);
+                ++ids_queued;
+            }
         }
     }
 
@@ -401,21 +394,6 @@ struct pending_threaded_processor : public middle_t::pending_processor
 
 } // anonymous namespace
 
-/**
- * Is there any pending work in the middle or one of the outputs?
- */
-bool osmdata_t::has_pending() const noexcept
-{
-    if (m_mid->has_pending()) {
-        return true;
-    }
-
-    return std::any_of(m_outs.cbegin(), m_outs.cend(),
-                       [](std::shared_ptr<output_t> const &out) {
-                           return out->has_pending();
-                       });
-}
-
 void osmdata_t::stop() const
 {
     /* Commit the transactions, so that multiple processes can
@@ -432,7 +410,7 @@ void osmdata_t::stop() const
     auto const *opts = m_outs[0]->get_options();
 
     // are there any objects left pending?
-    if (has_pending()) {
+    if (m_mid->has_pending()) {
         //threaded pending processing
         pending_threaded_processor ptp(m_mid, m_outs, opts->num_procs,
                                        opts->append);

src/osmdata.hpp

@@ -34,7 +34,6 @@ class osmdata_t
 
 private:
     slim_middle_t &slim_middle() const noexcept;
-    bool has_pending() const noexcept;
 
     std::shared_ptr<middle_t> m_mid;
     std::vector<std::shared_ptr<output_t>> m_outs;

src/output-flex.cpp

@@ -1007,7 +1007,7 @@ void output_flex_t::call_process_function(int index,
     }
 }
 
-void output_flex_t::pending_way(osmid_t id, int exists)
+void output_flex_t::pending_way(osmid_t id, bool exists)
 {
     if (!m_has_process_way) {
         return;
@@ -1020,10 +1020,6 @@ void output_flex_t::pending_way(osmid_t id, int exists)
 
     if (exists) {
         way_delete(id);
-        auto const rel_ids = m_mid->relations_using_way(id);
-        for (auto const rel_id : rel_ids) {
-            m_rels_pending_tracker.mark(rel_id);
-        }
     }
 
     auto &way = m_buffer.get<osmium::Way>(0);
@@ -1035,51 +1031,7 @@ void output_flex_t::pending_way(osmid_t id, int exists)
     m_buffer.clear();
 }
 
-void output_flex_t::enqueue_relations(pending_queue_t &job_queue, osmid_t id,
-                                      std::size_t output_id, std::size_t &added)
-{
-    if (!m_has_process_relation) {
-        return;
-    }
-
-    osmid_t const prev = m_rels_pending_tracker.last_returned();
-    if (id_tracker::is_valid(prev) && prev >= id) {
-        if (prev > id) {
-            job_queue.emplace(id, output_id);
-        }
-        // already done the job
-        return;
-    }
-
-    //make sure we get the one passed in
-    if (id_tracker::is_valid(id)) {
-        job_queue.emplace(id, output_id);
-        ++added;
-    }
-
-    //grab the first one or bail if its not valid
-    osmid_t popped = m_rels_pending_tracker.pop_mark();
-    if (!id_tracker::is_valid(popped)) {
-        return;
-    }
-
-    //get all the ones up to the id that was passed in
-    while (popped < id) {
-        job_queue.emplace(popped, output_id);
-        ++added;
-        popped = m_rels_pending_tracker.pop_mark();
-    }
-
-    //make sure to get this one as well and move to the next
-    if (popped > id) {
-        if (id_tracker::is_valid(popped)) {
-            job_queue.emplace(popped, output_id);
-            ++added;
-        }
-    }
-}
-
-void output_flex_t::pending_relation(osmid_t id, int exists)
+void output_flex_t::pending_relation(osmid_t id, bool exists)
 {
     if (!m_has_process_relation) {
         return;
@@ -1373,11 +1325,6 @@ void output_flex_t::init_lua(std::string const &filename)
     lua_remove(lua_state(), 1); // global "osm2pgsql"
 }
 
-bool output_flex_t::has_pending() const
-{
-    return !m_rels_pending_tracker.empty();
-}
-
 void output_flex_t::stage2_proc()
 {
     bool const has_marked_ways = !m_stage2_ways_tracker->empty();
@@ -1447,18 +1394,6 @@ void output_flex_t::stage2_proc()
     }
 }
 
-void output_flex_t::merge_pending_relations(output_t *other)
-{
-    auto *opgsql = dynamic_cast<output_flex_t *>(other);
-    if (opgsql) {
-        osmid_t id;
-        while (id_tracker::is_valid(
-            (id = opgsql->m_rels_pending_tracker.pop_mark()))) {
-            m_rels_pending_tracker.mark(id);
-        }
-    }
-}
-
 void output_flex_t::merge_expire_trees(output_t *other)
 {
     auto *opgsql = dynamic_cast<output_flex_t *>(other);

src/output-flex.hpp

@@ -61,11 +61,8 @@ class output_flex_t : public output_t
 
     void stage2_proc() override;
 
-    void pending_way(osmid_t id, int exists) override;
-
-    void enqueue_relations(pending_queue_t &job_queue, osmid_t id,
-                           std::size_t output_id, std::size_t &added) override;
-    void pending_relation(osmid_t id, int exists) override;
+    void pending_way(osmid_t id, bool exists) override;
+    void pending_relation(osmid_t id, bool exists) override;
 
     void node_add(osmium::Node const &node) override;
     void way_add(osmium::Way *way) override;
@@ -79,9 +76,6 @@ class output_flex_t : public output_t
     void way_delete(osmid_t id) override;
     void relation_delete(osmid_t id) override;
 
-    bool has_pending() const override;
-
-    void merge_pending_relations(output_t *other) override;
     void merge_expire_trees(output_t *other) override;
 
     int app_define_table();
@@ -137,8 +131,6 @@ class output_flex_t : public output_t
     std::shared_ptr<std::vector<flex_table_t>> m_tables;
     std::vector<table_connection_t> m_table_connections;
 
-    id_tracker m_rels_pending_tracker;
-
     std::shared_ptr<id_tracker> m_stage2_ways_tracker;
     std::shared_ptr<id_tracker> m_stage2_rels_tracker;
 

src/output-gazetteer.hpp

@@ -48,12 +48,8 @@ class output_gazetteer_t : public output_t
 
     bool need_forward_dependencies() const noexcept override { return false; }
 
-    void pending_way(osmid_t, int) override {}
-
-    void enqueue_relations(pending_queue_t &, osmid_t, size_t,
-                           size_t &) override
-    {}
-    void pending_relation(osmid_t, int) override {}
+    void pending_way(osmid_t, bool) override {}
+    void pending_relation(osmid_t, bool) override {}
 
     void node_add(osmium::Node const &node) override;
     void way_add(osmium::Way *way) override;

src/output-multi.cpp

@@ -65,12 +65,7 @@ void output_multi_t::start()
                    m_options.tblsmain_data);
 }
 
-bool output_multi_t::has_pending() const
-{
-    return !rels_pending_tracker.empty();
-}
-
-void output_multi_t::pending_way(osmid_t id, int exists)
+void output_multi_t::pending_way(osmid_t id, bool exists)
 {
     // Try to fetch the way from the DB
     buffer.clear();
@@ -80,47 +75,7 @@ void output_multi_t::pending_way(osmid_t id, int exists)
     }
 }
 
-void output_multi_t::enqueue_relations(pending_queue_t &job_queue, osmid_t id,
-                                       size_t output_id, size_t &added)
-{
-    osmid_t const prev = rels_pending_tracker.last_returned();
-    if (id_tracker::is_valid(prev) && prev >= id) {
-        if (prev > id) {
-            job_queue.push(pending_job_t(id, output_id));
-        }
-        // already done the job
-        return;
-    }
-
-    //make sure we get the one passed in
-    if (id_tracker::is_valid(id)) {
-        job_queue.push(pending_job_t(id, output_id));
-        ++added;
-    }
-
-    //grab the first one or bail if its not valid
-    osmid_t popped = rels_pending_tracker.pop_mark();
-    if (!id_tracker::is_valid(popped)) {
-        return;
-    }
-
-    //get all the ones up to the id that was passed in
-    while (popped < id) {
-        job_queue.push(pending_job_t(popped, output_id));
-        ++added;
-        popped = rels_pending_tracker.pop_mark();
-    }
-
-    //make sure to get this one as well and move to the next
-    if (popped > id) {
-        if (id_tracker::is_valid(popped)) {
-            job_queue.push(pending_job_t(popped, output_id));
-            ++added;
-        }
-    }
-}
-
-void output_multi_t::pending_relation(osmid_t id, int exists)
+void output_multi_t::pending_relation(osmid_t id, bool exists)
 {
     // Try to fetch the relation from the DB
     buffer.clear();
@@ -250,10 +205,6 @@ void output_multi_t::reprocess_way(osmium::Way *way, bool exists)
     if (m_processor->interests(geometry_processor::interest_relation) &&
         exists) {
         way_delete(way->id());
-        auto const rel_ids = m_mid->relations_using_way(way->id());
-        for (auto const rel_id : rel_ids) {
-            rels_pending_tracker.mark(rel_id);
-        }
     }
 
     //check if we are keeping this way
@@ -370,19 +321,6 @@ void output_multi_t::delete_from_output(osmid_t id)
     }
 }
 
-void output_multi_t::merge_pending_relations(output_t *other)
-{
-    auto *const omulti = dynamic_cast<output_multi_t *>(other);
-
-    if (omulti) {
-        osmid_t id;
-        while (id_tracker::is_valid(
-            (id = omulti->rels_pending_tracker.pop_mark()))) {
-            rels_pending_tracker.mark(id);
-        }
-    }
-}
-
 void output_multi_t::merge_expire_trees(output_t *other)
 {
     auto *const omulti = dynamic_cast<output_multi_t *>(other);