Merge pull request #1144 from joto/refactor-dependency-tracking-step1

Refactor dependency tracking step1

Author: lonvia

src/middle-pgsql.cpp

@@ -445,8 +445,6 @@ void middle_pgsql_t::iterate_ways(middle_t::pending_processor &pf)
     while (id_tracker::is_valid(id = m_ways_pending_tracker->pop_mark())) {
         pf.enqueue_ways(id);
     }
-    // in case we had higher ones than the middle
-    pf.enqueue_ways(id_tracker::max());
 
     //let the threads work on them
     pf.process_ways();

src/osmdata.cpp

@@ -244,7 +244,10 @@ struct pending_threaded_processor : public middle_t::pending_processor
     void enqueue_ways(osmid_t id) override
     {
         for (size_t i = 0; i < outs.size(); ++i) {
-            outs[i]->enqueue_ways(queue, id, i, ids_queued);
+            if (outs[i]->need_forward_dependencies()) {
+                queue.emplace(id, i);
+                ++ids_queued;
+            }
         }
     }
 

src/output-flex.cpp

@@ -1002,49 +1002,6 @@ void output_flex_t::call_process_function(int index,
     }
 }
 
-void output_flex_t::enqueue_ways(pending_queue_t &job_queue, osmid_t id,
-                                 std::size_t output_id, std::size_t &added)
-{
-    osmid_t const prev = m_ways_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 (!m_ways_done_tracker->is_marked(id) && 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 = m_ways_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) {
-        if (!m_ways_done_tracker->is_marked(popped)) {
-            job_queue.push(pending_job_t(popped, output_id));
-            ++added;
-        }
-        popped = m_ways_pending_tracker.pop_mark();
-    }
-
-    //make sure to get this one as well and move to the next
-    if (popped > id) {
-        if (!m_ways_done_tracker->is_marked(popped) &&
-            id_tracker::is_valid(popped)) {
-            job_queue.push(pending_job_t(popped, output_id));
-            ++added;
-        }
-    }
-}
-
 void output_flex_t::pending_way(osmid_t id, int exists)
 {
     if (!m_has_process_way) {
@@ -1292,7 +1249,6 @@ output_flex_t::output_flex_t(
     std::shared_ptr<id_tracker> ways_tracker,
     std::shared_ptr<id_tracker> rels_tracker)
 : output_t(mid, o), m_tables(std::move(tables)),
-  m_ways_done_tracker(new id_tracker{}),
   m_stage2_ways_tracker(std::move(ways_tracker)),
   m_stage2_rels_tracker(std::move(rels_tracker)), m_copy_thread(copy_thread),
   m_lua_state(std::move(lua_state)), m_builder(o.projection),
@@ -1414,7 +1370,7 @@ void output_flex_t::init_lua(std::string const &filename)
 
 bool output_flex_t::has_pending() const
 {
-    return !m_ways_pending_tracker.empty() || !m_rels_pending_tracker.empty();
+    return !m_rels_pending_tracker.empty();
 }
 
 void output_flex_t::stage2_proc()

src/output-flex.hpp

@@ -61,8 +61,6 @@ class output_flex_t : public output_t
 
     void stage2_proc() override;
 
-    void enqueue_ways(pending_queue_t &job_queue, osmid_t id,
-                      std::size_t output_id, std::size_t &added) override;
     void pending_way(osmid_t id, int exists) override;
 
     void enqueue_relations(pending_queue_t &job_queue, osmid_t id,
@@ -139,9 +137,7 @@ 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_ways_pending_tracker;
     id_tracker m_rels_pending_tracker;
-    std::shared_ptr<id_tracker> m_ways_done_tracker;
 
     std::shared_ptr<id_tracker> m_stage2_ways_tracker;
     std::shared_ptr<id_tracker> m_stage2_rels_tracker;

src/output-gazetteer.hpp

@@ -46,7 +46,8 @@ class output_gazetteer_t : public output_t
     void stop(osmium::thread::Pool *) override {}
     void commit() override;
 
-    void enqueue_ways(pending_queue_t &, osmid_t, size_t, size_t &) override {}
+    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,

src/output-multi.cpp

@@ -28,7 +28,6 @@ output_multi_t::output_multi_t(
                       export_list.normal_columns(m_osm_type),
                       m_options.hstore_columns, m_processor->srid(),
                       m_options.append, m_options.hstore_mode, copy_thread}),
-  ways_done_tracker(new id_tracker{}),
   m_expire(m_options.expire_tiles_zoom, m_options.expire_tiles_max_bbox,
            m_options.projection),
   buffer(1024, osmium::memory::Buffer::auto_grow::yes),
@@ -44,10 +43,6 @@ output_multi_t::output_multi_t(
   m_processor(other->m_processor), m_proj(other->m_proj),
   m_osm_type(other->m_osm_type),
   m_table(new table_t{*other->m_table, copy_thread}),
-  // NOTE: we need to know which ways were used by relations so each thread
-  // must have a copy of the original marked done ways, its read only so its
-  // ok
-  ways_done_tracker(other->ways_done_tracker),
   m_expire(m_options.expire_tiles_zoom, m_options.expire_tiles_max_bbox,
            m_options.projection),
   buffer(1024, osmium::memory::Buffer::auto_grow::yes),
@@ -72,50 +67,7 @@ void output_multi_t::start()
 
 bool output_multi_t::has_pending() const
 {
-    return !ways_pending_tracker.empty() || !rels_pending_tracker.empty();
-}
-
-void output_multi_t::enqueue_ways(pending_queue_t &job_queue, osmid_t id,
-                                  size_t output_id, size_t &added)
-{
-    osmid_t const prev = ways_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 (!ways_done_tracker->is_marked(id) && 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 = ways_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) {
-        if (!ways_done_tracker->is_marked(popped)) {
-            job_queue.push(pending_job_t(popped, output_id));
-            ++added;
-        }
-        popped = ways_pending_tracker.pop_mark();
-    }
-
-    //make sure to get this one as well and move to the next
-    if (popped > id) {
-        if (!ways_done_tracker->is_marked(popped) &&
-            id_tracker::is_valid(popped)) {
-            job_queue.push(pending_job_t(popped, output_id));
-            ++added;
-        }
-    }
+    return !rels_pending_tracker.empty();
 }
 
 void output_multi_t::pending_way(osmid_t id, int exists)
@@ -332,15 +284,7 @@ void output_multi_t::process_way(osmium::Way *way)
         auto const geom = m_processor->process_way(*way, &m_builder);
 
         if (!geom.empty()) {
-            //if we are also interested in relations we need to mark
-            //this way pending just in case it shows up in one
-            if (m_processor->interests(geometry_processor::interest_relation)) {
-                ways_pending_tracker.mark(way->id());
-            } else {
-                // We wouldn't be interested in this as a relation, so no need to mark it pending.
-                // TODO: Does this imply anything for non-multipolygon relations?
-                copy_to_table(way->id(), geom, outtags);
-            }
+            copy_to_table(way->id(), geom, outtags);
         }
     }
 }

src/output-multi.hpp

@@ -46,8 +46,6 @@ class output_multi_t : public output_t
     void stop(osmium::thread::Pool *pool) override;
     void commit() override;
 
-    void enqueue_ways(pending_queue_t &job_queue, osmid_t id, size_t output_id,
-                      size_t &added) override;
     void pending_way(osmid_t id, int exists) override;
 
     void enqueue_relations(pending_queue_t &job_queue, osmid_t id,
@@ -87,8 +85,7 @@ class output_multi_t : public output_t
     std::shared_ptr<reprojection> m_proj;
     osmium::item_type const m_osm_type;
     std::unique_ptr<table_t> m_table;
-    id_tracker ways_pending_tracker, rels_pending_tracker;
-    std::shared_ptr<id_tracker> ways_done_tracker;
+    id_tracker rels_pending_tracker;
     expire_tiles m_expire;
     relation_helper m_relation_helper;
     osmium::memory::Buffer buffer;

src/output-null.hpp

@@ -22,9 +22,7 @@ class output_null_t : public output_t
     void commit() override {}
     void cleanup() {}
 
-    void enqueue_ways(pending_queue_t & /*job_queue*/, osmid_t /*id*/, size_t /*output_id*/,
-                      size_t & /*added*/) override
-    {}
+    bool need_forward_dependencies() const noexcept override { return false; }
 
     void pending_way(osmid_t /*id*/, int /*exists*/) override {}
 

src/output-pgsql.cpp

@@ -66,49 +66,6 @@ void output_pgsql_t::pgsql_out_way(osmium::Way const &way, taglist_t *tags,
     }
 }
 
-void output_pgsql_t::enqueue_ways(pending_queue_t &job_queue, osmid_t id,
-                                  size_t output_id, size_t &added)
-{
-    osmid_t const prev = ways_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 (!ways_done_tracker->is_marked(id) && 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 = ways_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) {
-        if (!ways_done_tracker->is_marked(popped)) {
-            job_queue.push(pending_job_t(popped, output_id));
-            ++added;
-        }
-        popped = ways_pending_tracker.pop_mark();
-    }
-
-    //make sure to get this one as well and move to the next
-    if (popped > id) {
-        if (!ways_done_tracker->is_marked(popped) &&
-            id_tracker::is_valid(popped)) {
-            job_queue.push(pending_job_t(popped, output_id));
-            ++added;
-        }
-    }
-}
-
 void output_pgsql_t::pending_way(osmid_t id, int exists)
 {
     // Try to fetch the way from the DB
@@ -436,7 +393,6 @@ output_pgsql_t::output_pgsql_t(
     std::shared_ptr<db_copy_thread_t> const &copy_thread)
 : output_t(mid, o), m_builder(o.projection),
   expire(o.expire_tiles_zoom, o.expire_tiles_max_bbox, o.projection),
-  ways_done_tracker(new id_tracker{}),
   buffer(32768, osmium::memory::Buffer::auto_grow::yes),
   rels_buffer(1024, osmium::memory::Buffer::auto_grow::yes)
 {
@@ -495,9 +451,6 @@ output_pgsql_t::output_pgsql_t(
   m_builder(m_options.projection),
   expire(m_options.expire_tiles_zoom, m_options.expire_tiles_max_bbox,
          m_options.projection),
-  //NOTE: we need to know which ways were used by relations so each thread
-  //must have a copy of the original marked done ways, its read only so its ok
-  ways_done_tracker(other->ways_done_tracker),
   buffer(1024, osmium::memory::Buffer::auto_grow::yes),
   rels_buffer(1024, osmium::memory::Buffer::auto_grow::yes)
 {
@@ -512,7 +465,7 @@ output_pgsql_t::~output_pgsql_t() = default;
 
 bool output_pgsql_t::has_pending() const
 {
-    return !ways_pending_tracker.empty() || !rels_pending_tracker.empty();
+    return !rels_pending_tracker.empty();
 }
 
 void output_pgsql_t::merge_pending_relations(output_t *other)

src/output-pgsql.hpp

@@ -46,8 +46,6 @@ class output_pgsql_t : public output_t
     void stop(osmium::thread::Pool *pool) override;
     void commit() override;
 
-    void enqueue_ways(pending_queue_t &job_queue, osmid_t id, size_t output_id,
-                      size_t &added) override;
     void pending_way(osmid_t id, int exists) override;
 
     void enqueue_relations(pending_queue_t &job_queue, osmid_t id,
@@ -88,8 +86,7 @@ class output_pgsql_t : public output_t
     geom::osmium_builder_t m_builder;
     expire_tiles expire;
 
-    id_tracker ways_pending_tracker, rels_pending_tracker;
-    std::shared_ptr<id_tracker> ways_done_tracker;
+    id_tracker rels_pending_tracker;
     osmium::memory::Buffer buffer;
     osmium::memory::Buffer rels_buffer;
 };