Merge pull request #898 from lonvia/shared-pointer-for-middle

 Hand outputs a shared pointer to the middle query object

Author: lonvia

osm2pgsql.cpp

@@ -60,11 +60,13 @@ int main(int argc, char *argv[])
 
         if (options.slim) {
             auto mid = std::make_shared<middle_pgsql_t>();
-            outputs = output_t::create_outputs(mid.get(), options);
+            outputs = output_t::create_outputs(
+                std::static_pointer_cast<middle_query_t>(mid), options);
             middle = std::static_pointer_cast<middle_t>(mid);
         } else {
             auto mid = std::make_shared<middle_ram_t>();
-            outputs = output_t::create_outputs(mid.get(), options);
+            outputs = output_t::create_outputs(
+                std::static_pointer_cast<middle_query_t>(mid), options);
             middle = std::static_pointer_cast<middle_t>(mid);
         }
 

osmdata.cpp

@@ -186,8 +186,7 @@ namespace {
 //since the fetching from middle should be faster than the processing in each backend.
 
 struct pending_threaded_processor : public middle_t::pending_processor {
-    typedef std::vector<std::shared_ptr<output_t>> output_vec_t;
-    typedef std::pair<std::shared_ptr<middle_query_t>, output_vec_t> clone_t;
+    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, bool ways) {
         while (true) {
@@ -253,11 +252,11 @@ struct pending_threaded_processor : public middle_t::pending_processor {
             //clone the outs
             output_vec_t out_clones;
             for (const auto& out: outs) {
-                out_clones.push_back(out->clone(mid_clone.get()));
+                out_clones.push_back(out->clone(mid_clone));
             }
 
             //keep the clones for a specific thread to use
-            clones.push_back(clone_t(mid_clone, out_clones));
+            clones.push_back(out_clones);
         }
     }
 
@@ -282,11 +281,10 @@ struct pending_threaded_processor : public middle_t::pending_processor {
 
         //make the threads and start them
         std::vector<std::future<void>> workers;
-        for (size_t i = 0; i < clones.size(); ++i) {
-            workers.push_back(std::async(std::launch::async,
-                                         do_jobs, std::cref(clones[i].second),
-                                         std::ref(queue), std::ref(ids_done),
-                                         std::ref(mutex), append, true));
+        for (auto const &clone : clones) {
+            workers.push_back(std::async(
+                std::launch::async, do_jobs, std::cref(clone), std::ref(queue),
+                std::ref(ids_done), std::ref(mutex), append, true));
         }
         workers.push_back(std::async(std::launch::async, print_stats,
                                      std::ref(queue), std::ref(mutex)));
@@ -315,10 +313,12 @@ struct pending_threaded_processor : public middle_t::pending_processor {
 
         //collect all the new rels that became pending from each
         //output in each thread back to their respective main outputs
-        for (const auto& clone: clones) {
+        for (auto const &clone : clones) {
             //for each clone/original output
-            for(output_vec_t::const_iterator original_output = outs.begin(), clone_output = clone.second.begin();
-                original_output != outs.end() && clone_output != clone.second.end(); ++original_output, ++clone_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
@@ -344,11 +344,10 @@ struct pending_threaded_processor : public middle_t::pending_processor {
 
         //make the threads and start them
         std::vector<std::future<void>> workers;
-        for (size_t i = 0; i < clones.size(); ++i) {
-            workers.push_back(std::async(std::launch::async,
-                                         do_jobs, std::cref(clones[i].second),
-                                         std::ref(queue), std::ref(ids_done),
-                                         std::ref(mutex), append, false));
+        for (auto const &clone : clones) {
+            workers.push_back(std::async(
+                std::launch::async, do_jobs, std::cref(clone), std::ref(queue),
+                std::ref(ids_done), std::ref(mutex), append, false));
         }
         workers.push_back(std::async(std::launch::async, print_stats,
                                      std::ref(queue), std::ref(mutex)));
@@ -376,10 +375,12 @@ struct pending_threaded_processor : public middle_t::pending_processor {
         ids_done = 0;
 
         //collect all expiry tree informations together into one
-        for (const auto& clone: clones) {
+        for (auto const &clone : clones) {
             //for each clone/original output
-            for(output_vec_t::const_iterator original_output = outs.begin(), clone_output = clone.second.begin();
-                original_output != outs.end() && clone_output != clone.second.end(); ++original_output, ++clone_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 rels for now
                 clone_output->get()->commit();
                 //merge the expire tree from this threads copy of output back
@@ -389,8 +390,8 @@ struct pending_threaded_processor : public middle_t::pending_processor {
     }
 
 private:
-    //middle and output copies
-    std::vector<clone_t> clones;
+    // output copies, one vector per thread
+    std::vector<output_vec_t> clones;
     output_vec_t outs; //would like to move ownership of outs to osmdata_t and middle passed to output_t instead of owned by it
     //how many jobs do we have in the queue to start with
     size_t ids_queued;

output-gazetteer.hpp

@@ -14,36 +14,38 @@
 #include "pgsql.hpp"
 #include "util.hpp"
 
-
-class output_gazetteer_t : public output_t {
-public:
-    output_gazetteer_t(const middle_query_t *mid_, const options_t &options_)
-    : output_t(mid_, options_), Connection(NULL), ConnectionDelete(NULL),
-      ConnectionError(NULL), copy_active(false),
-      m_builder(options_.projection, true), single_fmt("%1%"),
+class output_gazetteer_t : public output_t
+{
+    output_gazetteer_t(output_gazetteer_t const *other,
+                       std::shared_ptr<middle_query_t> const &cloned_mid)
+    : output_t(cloned_mid, other->m_options), Connection(NULL),
+      ConnectionDelete(NULL), ConnectionError(NULL), copy_active(false),
+      m_builder(other->m_options.projection, true),
+      single_fmt(other->single_fmt),
       osmium_buffer(PLACE_BUFFER_SIZE, osmium::memory::Buffer::auto_grow::yes)
     {
         buffer.reserve(PLACE_BUFFER_SIZE);
-        m_style.load_style(options_.style);
+        connect();
     }
 
-    output_gazetteer_t(const output_gazetteer_t &other)
-    : output_t(other.m_mid, other.m_options), Connection(NULL),
-      ConnectionDelete(NULL), ConnectionError(NULL), copy_active(false),
-      m_builder(other.m_options.projection, true), single_fmt(other.single_fmt),
+public:
+    output_gazetteer_t(std::shared_ptr<middle_query_t> const &mid,
+                       options_t const &options)
+    : output_t(mid, options), Connection(NULL), ConnectionDelete(NULL),
+      ConnectionError(NULL), copy_active(false),
+      m_builder(options.projection, true), single_fmt("%1%"),
       osmium_buffer(PLACE_BUFFER_SIZE, osmium::memory::Buffer::auto_grow::yes)
     {
         buffer.reserve(PLACE_BUFFER_SIZE);
-        connect();
+        m_style.load_style(options.style);
     }
 
     virtual ~output_gazetteer_t() {}
 
-    std::shared_ptr<output_t> clone(const middle_query_t* cloned_middle) const override
+    std::shared_ptr<output_t>
+    clone(std::shared_ptr<middle_query_t> const &mid) const override
     {
-        output_gazetteer_t *clone = new output_gazetteer_t(*this);
-        clone->m_mid = cloned_middle;
-        return std::shared_ptr<output_t>(clone);
+        return std::shared_ptr<output_t>(new output_gazetteer_t(this, mid));
     }
 
     int start() override;

output-multi.cpp

@@ -11,12 +11,12 @@
 #include <boost/algorithm/string/predicate.hpp>
 #include <vector>
 
-output_multi_t::output_multi_t(const std::string &name,
+output_multi_t::output_multi_t(std::string const &name,
                                std::shared_ptr<geometry_processor> processor_,
-                               const struct export_list &export_list_,
-                               const middle_query_t *mid_,
-                               const options_t &options_)
-: output_t(mid_, options_),
+                               export_list const &export_list_,
+                               std::shared_ptr<middle_query_t> const &mid,
+                               options_t const &options)
+: output_t(mid, options),
   m_tagtransform(tagtransform_t::make_tagtransform(&m_options)),
   m_export_list(new export_list(export_list_)), m_processor(processor_),
   m_proj(m_options.projection),
@@ -40,31 +40,31 @@ output_multi_t::output_multi_t(const std::string &name,
 {
 }
 
-output_multi_t::output_multi_t(const output_multi_t &other)
-: output_t(other.m_mid, other.m_options),
+output_multi_t::output_multi_t(output_multi_t const *other,
+                               std::shared_ptr<middle_query_t> const &mid)
+: output_t(mid, other->m_options),
   m_tagtransform(tagtransform_t::make_tagtransform(&m_options)),
-  m_export_list(new export_list(*other.m_export_list)),
-  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)),
+  m_export_list(new export_list(*other->m_export_list)),
+  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)),
   // 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),
+  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),
   m_builder(m_options.projection, m_options.enable_multi),
-  m_way_area(other.m_way_area)
+  m_way_area(other->m_way_area)
 {
 }
 
 output_multi_t::~output_multi_t() = default;
 
-std::shared_ptr<output_t> output_multi_t::clone(const middle_query_t* cloned_middle) const
+std::shared_ptr<output_t>
+output_multi_t::clone(std::shared_ptr<middle_query_t> const &mid) const
 {
-    auto *clone = new output_multi_t(*this);
-    clone->m_mid = cloned_middle;
-    return std::shared_ptr<output_t>(clone);
+    return std::shared_ptr<output_t>(new output_multi_t(this, mid));
 }
 
 int output_multi_t::start() {
@@ -365,7 +365,7 @@ int output_multi_t::process_relation(osmium::Relation const &rel,
     if (!filter) {
         //TODO: move this into geometry processor, figure a way to come back for tag transform
         //grab ways/nodes of the members in the relation, bail if none were used
-        if (m_relation_helper.set(rel, m_mid) < 1)
+        if (m_relation_helper.set(rel, m_mid.get()) < 1)
             return 0;
 
         //NOTE: make_polygon is preset here this is to force the tag matching
@@ -383,7 +383,7 @@ int output_multi_t::process_relation(osmium::Relation const &rel,
             outtags, true);
         if (!filter)
         {
-            m_relation_helper.add_way_locations(m_mid);
+            m_relation_helper.add_way_locations(m_mid.get());
             auto geoms = m_processor->process_relation(
                 rel, m_relation_helper.data, &m_builder);
             for (const auto geom : geoms) {

output-multi.hpp

@@ -23,16 +23,21 @@ struct export_list;
 struct middle_query_t;
 struct options_t;
 
-class output_multi_t : public output_t {
+class output_multi_t : public output_t
+{
+    output_multi_t(output_multi_t const *other,
+                   std::shared_ptr<middle_query_t> const &mid);
+
 public:
-    output_multi_t(const std::string &name,
+    output_multi_t(std::string const &name,
                    std::shared_ptr<geometry_processor> processor_,
-                   const export_list &export_list_,
-                   const middle_query_t* mid_, const options_t &options_);
-    output_multi_t(const output_multi_t& other);
+                   export_list const &export_list_,
+                   std::shared_ptr<middle_query_t> const &mid,
+                   options_t const &options);
     virtual ~output_multi_t();
 
-    std::shared_ptr<output_t> clone(const middle_query_t* cloned_middle) const override;
+    std::shared_ptr<output_t>
+    clone(std::shared_ptr<middle_query_t> const &mid) const override;
 
     int start() override;
     void stop(osmium::thread::Pool *pool) override;

output-null.cpp

@@ -62,17 +62,22 @@ int output_null_t::relation_modify(osmium::Relation const &) {
   return 0;
 }
 
-std::shared_ptr<output_t> output_null_t::clone(const middle_query_t* cloned_middle) const {
-    output_null_t *clone = new output_null_t(*this);
-    clone->m_mid = cloned_middle;
-    return std::shared_ptr<output_t>(clone);
+std::shared_ptr<output_t>
+output_null_t::clone(std::shared_ptr<middle_query_t> const &mid) const
+{
+    return std::shared_ptr<output_t>(new output_null_t(this, mid));
 }
 
-output_null_t::output_null_t(const middle_query_t* mid_, const options_t &options_): output_t(mid_, options_) {
+output_null_t::output_null_t(std::shared_ptr<middle_query_t> const &mid,
+                             options_t const &options)
+: output_t(mid, options)
+{
 }
 
-output_null_t::output_null_t(const output_null_t& other): output_t(other.m_mid, other.m_options) {
+output_null_t::output_null_t(output_null_t const *other,
+                             std::shared_ptr<middle_query_t> const &mid)
+: output_t(mid, other->m_options)
+{
 }
 
-output_null_t::~output_null_t() {
-}
+output_null_t::~output_null_t() = default;

output-null.hpp

@@ -7,12 +7,16 @@
 #include "output.hpp"
 
 class output_null_t : public output_t {
+    output_null_t(output_null_t const *other,
+                  std::shared_ptr<middle_query_t> const &mid);
+
 public:
-    output_null_t(const middle_query_t* mid_, const options_t &options);
-    output_null_t(const output_null_t& other);
+    output_null_t(std::shared_ptr<middle_query_t> const &mid,
+                  options_t const &options);
     virtual ~output_null_t();
 
-    std::shared_ptr<output_t> clone(const middle_query_t* cloned_middle) const override;
+    std::shared_ptr<output_t>
+    clone(std::shared_ptr<middle_query_t> const &mid) const override;
 
     int start() override;
     void stop(osmium::thread::Pool *pool) override;

output-pgsql.cpp

@@ -502,14 +502,14 @@ int output_pgsql_t::start()
     return 0;
 }
 
-std::shared_ptr<output_t> output_pgsql_t::clone(const middle_query_t* cloned_middle) const
+std::shared_ptr<output_t>
+output_pgsql_t::clone(std::shared_ptr<middle_query_t> const &mid) const
 {
-    auto *clone = new output_pgsql_t(*this);
-    clone->m_mid = cloned_middle;
-    return std::shared_ptr<output_t>(clone);
+    return std::shared_ptr<output_t>(new output_pgsql_t(this, mid));
 }
 
-output_pgsql_t::output_pgsql_t(const middle_query_t *mid, const options_t &o)
+output_pgsql_t::output_pgsql_t(std::shared_ptr<middle_query_t> const &mid,
+                               options_t const &o)
 : output_t(mid, o), m_builder(o.projection, o.enable_multi),
   expire(o.expire_tiles_zoom, o.expire_tiles_max_bbox, o.projection),
   ways_done_tracker(new id_tracker()),
@@ -576,23 +576,24 @@ output_pgsql_t::output_pgsql_t(const middle_query_t *mid, const options_t &o)
     }
 }
 
-output_pgsql_t::output_pgsql_t(const output_pgsql_t &other)
-: output_t(other.m_mid, other.m_options),
+output_pgsql_t::output_pgsql_t(output_pgsql_t const *other,
+                               std::shared_ptr<middle_query_t> const &mid)
+: output_t(mid, other->m_options),
   m_tagtransform(tagtransform_t::make_tagtransform(&m_options)),
-  m_enable_way_area(other.m_enable_way_area),
-  m_export_list(new export_list(*other.m_export_list)),
-  m_builder(m_options.projection, other.m_options.enable_multi),
+  m_enable_way_area(other->m_enable_way_area),
+  m_export_list(new export_list(*other->m_export_list)),
+  m_builder(m_options.projection, other->m_options.enable_multi),
   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),
+  ways_done_tracker(other->ways_done_tracker),
   buffer(1024, osmium::memory::Buffer::auto_grow::yes),
   rels_buffer(1024, osmium::memory::Buffer::auto_grow::yes)
 {
-    for(std::vector<std::shared_ptr<table_t> >::const_iterator t = other.m_tables.begin(); t != other.m_tables.end(); ++t) {
+    for (auto const &t : other->m_tables) {
         //copy constructor will just connect to the already there table
-        m_tables.push_back(std::shared_ptr<table_t>(new table_t(**t)));
+        m_tables.push_back(std::shared_ptr<table_t>(new table_t(*t.get())));
     }
 }