Reorder parts of the pending_threaded_processor class

Author: joto

src/osmdata.cpp

@@ -162,12 +162,64 @@ void osmdata_t::flush() const
 
 namespace {
 
-struct pending_threaded_processor
+class pending_threaded_processor
 {
+public:
     using output_vec_t = std::vector<std::shared_ptr<output_t>>;
-    using pending_queue_t = idlist_t;
 
-    static osmid_t pop_id(pending_queue_t &queue, std::mutex &mutex)
+    pending_threaded_processor(std::shared_ptr<middle_t> mid,
+                               output_vec_t const &outs, size_t thread_count)
+    : m_outputs(outs)
+    {
+        assert(!outs.empty());
+
+        // The database connection info should be the same for all outputs,
+        // we take it arbitrarily from the first.
+        std::string const &conninfo =
+            m_outputs[0]->get_options()->database_options.conninfo();
+
+        m_clones.resize(thread_count);
+        for (size_t i = 0; i < thread_count; ++i) {
+            auto const midq = mid->get_query_instance();
+            auto copy_thread = std::make_shared<db_copy_thread_t>(conninfo);
+
+            for (auto const &out : m_outputs) {
+                if (out->need_forward_dependencies()) {
+                    m_clones[i].push_back(out->clone(midq, copy_thread));
+                } else {
+                    m_clones[i].emplace_back(nullptr);
+                }
+            }
+        }
+    }
+
+    void process_ways(idlist_t &&list)
+    {
+        m_queue = std::move(list);
+        process_queue("way", do_ways);
+    }
+
+    void process_relations(idlist_t &&list)
+    {
+        m_queue = std::move(list);
+        process_queue("relation", do_rels);
+
+        // Collect expiry tree information from all clones and merge it back
+        // into the original outputs.
+        for (auto const &clone : m_clones) {
+            auto it = clone.begin();
+            for (auto const &output : m_outputs) {
+                assert(it != clone.end());
+                if (*it) {
+                    output->merge_expire_trees(it->get());
+                }
+                ++it;
+            }
+        }
+    }
+
+private:
+    static osmid_t pop_id(idlist_t &queue, std::mutex &mutex)
     {
         osmid_t id = 0;
 
@@ -180,7 +232,7 @@ struct pending_threaded_processor
         return id;
     }
 
-    static void do_ways(output_vec_t const &outputs, pending_queue_t &queue,
+    static void do_ways(output_vec_t const &outputs, idlist_t &queue,
                         std::mutex &mutex)
     {
         while (osmid_t const id = pop_id(queue, mutex)) {
@@ -192,7 +244,7 @@ struct pending_threaded_processor
         }
     }
 
-    static void do_rels(output_vec_t const &outputs, pending_queue_t &queue,
+    static void do_rels(output_vec_t const &outputs, idlist_t &queue,
                         std::mutex &mutex)
     {
         while (osmid_t const id = pop_id(queue, mutex)) {
@@ -204,7 +256,7 @@ struct pending_threaded_processor
         }
     }
 
-    static void print_stats(pending_queue_t &queue, std::mutex &mutex)
+    static void print_stats(idlist_t &queue, std::mutex &mutex)
     {
         size_t queue_size;
         do {
@@ -218,32 +270,6 @@ struct pending_threaded_processor
         } while (queue_size > 0);
     }
 
-    pending_threaded_processor(std::shared_ptr<middle_t> mid,
-                               output_vec_t const &outs, size_t thread_count)
-    : m_outputs(outs)
-    {
-        assert(!outs.empty());
-
-        // The database connection info should be the same for all outputs,
-        // we take it arbitrarily from the first.
-        std::string const &conninfo =
-            m_outputs[0]->get_options()->database_options.conninfo();
-
-        m_clones.resize(thread_count);
-        for (size_t i = 0; i < thread_count; ++i) {
-            auto const midq = mid->get_query_instance();
-            auto copy_thread = std::make_shared<db_copy_thread_t>(conninfo);
-
-            for (auto const &out : m_outputs) {
-                if (out->need_forward_dependencies()) {
-                    m_clones[i].push_back(out->clone(midq, copy_thread));
-                } else {
-                    m_clones[i].emplace_back(nullptr);
-                }
-            }
-        }
-    }
-
     template <typename FUNCTION>
     void process_queue(char const *type, FUNCTION &&function)
     {
@@ -297,40 +323,14 @@ struct pending_threaded_processor
         }
     }
 
-    void process_ways(idlist_t &&list)
-    {
-        m_queue = std::move(list);
-        process_queue("way", do_ways);
-    }
-
-    void process_relations(idlist_t &&list)
-    {
-        m_queue = std::move(list);
-        process_queue("relation", do_rels);
-
-        // Collect expiry tree information from all clones and merge it back
-        // into the original outputs.
-        for (auto const &clone : m_clones) {
-            auto it = clone.begin();
-            for (auto const &output : m_outputs) {
-                assert(it != clone.end());
-                if (*it) {
-                    output->merge_expire_trees(it->get());
-                }
-                ++it;
-            }
-        }
-    }
-
-private:
     /// Clones of all outputs, one vector of clones per thread.
     std::vector<output_vec_t> m_clones;
 
     /// All outputs.
     output_vec_t m_outputs;
 
     /// The queue with ids that the worker threads work on.
-    pending_queue_t m_queue;
+    idlist_t m_queue;
 
     /// Mutex to make sure worker threads coordinate access to queue.
     std::mutex m_mutex;