Parallize scripting on osmium::Buffer granularity

Fixes #3447 and reduces parsing time by about 15%.

Author: TheMarex

include/extractor/scripting_environment.hpp

@@ -58,13 +58,12 @@ class ScriptingEnvironment
     virtual void ProcessTurn(ExtractionTurn &turn) = 0;
     virtual void ProcessSegment(ExtractionSegment &segment) = 0;
 
-    virtual void
-    ProcessElements(const std::vector<osmium::memory::Buffer::const_iterator> &osm_elements,
-                    const RestrictionParser &restriction_parser,
-                    tbb::concurrent_vector<std::pair<std::size_t, ExtractionNode>> &resulting_nodes,
-                    tbb::concurrent_vector<std::pair<std::size_t, ExtractionWay>> &resulting_ways,
-                    tbb::concurrent_vector<boost::optional<InputRestrictionContainer>>
-                        &resulting_restrictions) = 0;
+    virtual void ProcessElements(
+        const osmium::memory::Buffer &buffer,
+        const RestrictionParser &restriction_parser,
+        std::vector<std::pair<const osmium::Node &, ExtractionNode>> &resulting_nodes,
+        std::vector<std::pair<const osmium::Way &, ExtractionWay>> &resulting_ways,
+        std::vector<boost::optional<InputRestrictionContainer>> &resulting_restrictions) = 0;
 };
 }
 }

include/extractor/scripting_environment_lua.hpp

@@ -65,13 +65,12 @@ class Sol2ScriptingEnvironment final : public ScriptingEnvironment
     void ProcessTurn(ExtractionTurn &turn) override;
     void ProcessSegment(ExtractionSegment &segment) override;
 
-    void
-    ProcessElements(const std::vector<osmium::memory::Buffer::const_iterator> &osm_elements,
-                    const RestrictionParser &restriction_parser,
-                    tbb::concurrent_vector<std::pair<std::size_t, ExtractionNode>> &resulting_nodes,
-                    tbb::concurrent_vector<std::pair<std::size_t, ExtractionWay>> &resulting_ways,
-                    tbb::concurrent_vector<boost::optional<InputRestrictionContainer>>
-                        &resulting_restrictions) override;
+    void ProcessElements(
+        const osmium::memory::Buffer &buffer,
+        const RestrictionParser &restriction_parser,
+        std::vector<std::pair<const osmium::Node &, ExtractionNode>> &resulting_nodes,
+        std::vector<std::pair<const osmium::Way &, ExtractionWay>> &resulting_ways,
+        std::vector<boost::optional<InputRestrictionContainer>> &resulting_restrictions) override;
 
   private:
     void InitContext(LuaScriptingContext &context);

src/extractor/extractor.cpp

@@ -33,12 +33,13 @@
 
 #include <boost/filesystem.hpp>
 #include <boost/filesystem/fstream.hpp>
+#include <boost/iterator/function_input_iterator.hpp>
 #include <boost/optional/optional.hpp>
 #include <boost/scope_exit.hpp>
 
 #include <osmium/io/any_input.hpp>
 
-#include <tbb/concurrent_vector.h>
+#include <tbb/pipeline.h>
 #include <tbb/task_scheduler_init.h>
 
 #include <cstdlib>
@@ -252,58 +253,77 @@ std::vector<TurnRestriction> Extractor::ParseOSMData(ScriptingEnvironment &scrip
 
     timestamp_file.WriteFrom(timestamp.c_str(), timestamp.length());
 
-    // initialize vectors holding parsed objects
-    tbb::concurrent_vector<std::pair<std::size_t, ExtractionNode>> resulting_nodes;
-    tbb::concurrent_vector<std::pair<std::size_t, ExtractionWay>> resulting_ways;
-    tbb::concurrent_vector<boost::optional<InputRestrictionContainer>> resulting_restrictions;
-
     std::vector<std::string> restrictions = scripting_environment.GetRestrictions();
     // setup restriction parser
     const RestrictionParser restriction_parser(
         scripting_environment.GetProfileProperties().use_turn_restrictions,
         config.parse_conditionals,
         restrictions);
 
-    // create a vector of iterators into the buffer
-    for (std::vector<osmium::memory::Buffer::const_iterator> osm_elements;
-         const osmium::memory::Buffer buffer = reader.read();
-         osm_elements.clear())
-    {
-        for (auto iter = std::begin(buffer), end = std::end(buffer); iter != end; ++iter)
-        {
-            osm_elements.push_back(iter);
-        }
+    std::mutex process_mutex;
 
-        // clear resulting vectors
-        resulting_nodes.clear();
-        resulting_ways.clear();
-        resulting_restrictions.clear();
+    using SharedBuffer = std::shared_ptr<const osmium::memory::Buffer>;
+    struct ParsedBuffer
+    {
+        SharedBuffer buffer;
+        std::vector<std::pair<const osmium::Node &, ExtractionNode>> resulting_nodes;
+        std::vector<std::pair<const osmium::Way &, ExtractionWay>> resulting_ways;
+        std::vector<boost::optional<InputRestrictionContainer>> resulting_restrictions;
+    };
 
-        scripting_environment.ProcessElements(osm_elements,
-                                              restriction_parser,
-                                              resulting_nodes,
-                                              resulting_ways,
-                                              resulting_restrictions);
+    tbb::filter_t<void, SharedBuffer> buffer_reader(
+        tbb::filter::serial_in_order, [&](tbb::flow_control &fc) {
+            if (auto buffer = reader.read())
+            {
+                return std::make_shared<const osmium::memory::Buffer>(std::move(buffer));
+            }
+            else
+            {
+                fc.stop();
+                return SharedBuffer{};
+            }
+        });
+    tbb::filter_t<SharedBuffer, std::shared_ptr<ParsedBuffer>> buffer_transform(
+        tbb::filter::parallel, [&](const SharedBuffer buffer) {
+            if (!buffer)
+                return std::shared_ptr<ParsedBuffer>{};
+
+            auto parsed_buffer = std::make_shared<ParsedBuffer>();
+            parsed_buffer->buffer = buffer;
+            scripting_environment.ProcessElements(*buffer,
+                                                  restriction_parser,
+                                                  parsed_buffer->resulting_nodes,
+                                                  parsed_buffer->resulting_ways,
+                                                  parsed_buffer->resulting_restrictions);
+            return parsed_buffer;
+        });
+    tbb::filter_t<std::shared_ptr<ParsedBuffer>, void> buffer_storage(
+        tbb::filter::serial_in_order, [&](const std::shared_ptr<ParsedBuffer> parsed_buffer) {
+            if (!parsed_buffer)
+                return;
+
+            number_of_nodes += parsed_buffer->resulting_nodes.size();
+            // put parsed objects thru extractor callbacks
+            for (const auto &result : parsed_buffer->resulting_nodes)
+            {
+                extractor_callbacks->ProcessNode(result.first, result.second);
+            }
+            number_of_ways += parsed_buffer->resulting_ways.size();
+            for (const auto &result : parsed_buffer->resulting_ways)
+            {
+                extractor_callbacks->ProcessWay(result.first, result.second);
+            }
+            number_of_relations += parsed_buffer->resulting_restrictions.size();
+            for (const auto &result : parsed_buffer->resulting_restrictions)
+            {
+                extractor_callbacks->ProcessRestriction(result);
+            }
+        });
+
+    // Number of pipeline tokens that yielded the best speedup was about 1.5 * num_cores
+    tbb::parallel_pipeline(tbb::task_scheduler_init::default_num_threads() * 1.5,
+                           buffer_reader & buffer_transform & buffer_storage);
 
-        number_of_nodes += resulting_nodes.size();
-        // put parsed objects thru extractor callbacks
-        for (const auto &result : resulting_nodes)
-        {
-            extractor_callbacks->ProcessNode(
-                static_cast<const osmium::Node &>(*(osm_elements[result.first])), result.second);
-        }
-        number_of_ways += resulting_ways.size();
-        for (const auto &result : resulting_ways)
-        {
-            extractor_callbacks->ProcessWay(
-                static_cast<const osmium::Way &>(*(osm_elements[result.first])), result.second);
-        }
-        number_of_relations += resulting_restrictions.size();
-        for (const auto &result : resulting_restrictions)
-        {
-            extractor_callbacks->ProcessRestriction(result);
-        }
-    }
     TIMER_STOP(parsing);
     util::Log() << "Parsing finished after " << TIMER_SEC(parsing) << " seconds";
 

src/extractor/scripting_environment_lua.cpp

@@ -489,61 +489,54 @@ LuaScriptingContext &Sol2ScriptingEnvironment::GetSol2Context()
 }
 
 void Sol2ScriptingEnvironment::ProcessElements(
-    const std::vector<osmium::memory::Buffer::const_iterator> &osm_elements,
+    const osmium::memory::Buffer &buffer,
     const RestrictionParser &restriction_parser,
-    tbb::concurrent_vector<std::pair<std::size_t, ExtractionNode>> &resulting_nodes,
-    tbb::concurrent_vector<std::pair<std::size_t, ExtractionWay>> &resulting_ways,
-    tbb::concurrent_vector<boost::optional<InputRestrictionContainer>> &resulting_restrictions)
+    std::vector<std::pair<const osmium::Node &, ExtractionNode>> &resulting_nodes,
+    std::vector<std::pair<const osmium::Way &, ExtractionWay>> &resulting_ways,
+    std::vector<boost::optional<InputRestrictionContainer>> &resulting_restrictions)
 {
-    // parse OSM entities in parallel, store in resulting vectors
-    tbb::parallel_for(
-        tbb::blocked_range<std::size_t>(0, osm_elements.size()),
-        [&](const tbb::blocked_range<std::size_t> &range) {
-            ExtractionNode result_node;
-            ExtractionWay result_way;
-            std::vector<InputRestrictionContainer> result_res;
-            auto &local_context = this->GetSol2Context();
-
-            for (auto x = range.begin(), end = range.end(); x != end; ++x)
+    ExtractionNode result_node;
+    ExtractionWay result_way;
+    std::vector<InputRestrictionContainer> result_res;
+    auto &local_context = this->GetSol2Context();
+
+    for (auto entity = buffer.cbegin(), end = buffer.cend(); entity != end; ++entity)
+    {
+        switch (entity->type())
+        {
+        case osmium::item_type::node:
+            result_node.clear();
+            if (local_context.has_node_function &&
+                (!static_cast<const osmium::Node &>(*entity).tags().empty() ||
+                 local_context.properties.call_tagless_node_function))
             {
-                const auto entity = osm_elements[x];
-
-                switch (entity->type())
-                {
-                case osmium::item_type::node:
-                    result_node.clear();
-                    if (local_context.has_node_function &&
-                        (!static_cast<const osmium::Node &>(*entity).tags().empty() ||
-                         local_context.properties.call_tagless_node_function))
-                    {
-                        local_context.ProcessNode(static_cast<const osmium::Node &>(*entity),
-                                                  result_node);
-                    }
-                    resulting_nodes.push_back(std::make_pair(x, std::move(result_node)));
-                    break;
-                case osmium::item_type::way:
-                    result_way.clear();
-                    if (local_context.has_way_function)
-                    {
-                        local_context.ProcessWay(static_cast<const osmium::Way &>(*entity),
-                                                 result_way);
-                    }
-                    resulting_ways.push_back(std::make_pair(x, std::move(result_way)));
-                    break;
-                case osmium::item_type::relation:
-                    result_res.clear();
-                    result_res =
-                        restriction_parser.TryParse(static_cast<const osmium::Relation &>(*entity));
-                    for (const InputRestrictionContainer &r : result_res)
-                    {
-                        resulting_restrictions.push_back(r);
-                    }
-                    break;
-                default:
-                    break;
-                }
+                local_context.ProcessNode(static_cast<const osmium::Node &>(*entity), result_node);
             }
-        });
+            resulting_nodes.push_back(std::pair<const osmium::Node &, ExtractionNode>(
+                static_cast<const osmium::Node &>(*entity), std::move(result_node)));
+            break;
+        case osmium::item_type::way:
+            result_way.clear();
+            if (local_context.has_way_function)
+            {
+                local_context.ProcessWay(static_cast<const osmium::Way &>(*entity), result_way);
+            }
+            resulting_ways.push_back(std::pair<const osmium::Way &, ExtractionWay>(
+                static_cast<const osmium::Way &>(*entity), std::move(result_way)));
+            break;
+        case osmium::item_type::relation:
+            result_res.clear();
+            result_res =
+                restriction_parser.TryParse(static_cast<const osmium::Relation &>(*entity));
+            for (const InputRestrictionContainer &r : result_res)
+            {
+                resulting_restrictions.push_back(r);
+            }
+            break;
+        default:
+            break;
+        }
+    }
 }
 
 std::vector<std::string> Sol2ScriptingEnvironment::GetNameSuffixList()