From 8070e0c84b38fdd2ec14d6364c581d5e948b72e5 Mon Sep 17 00:00:00 2001
From: Maksim Andrianov <maksimandrianov1@gmail.com>
Date: Tue, 13 Aug 2019 18:10:09 +0300
Subject: [PATCH] [generator] Added new parser interface.

---
 coding/parse_xml.hpp         |  42 ++++++++
 generator/osm_o5m_source.hpp |   6 +-
 generator/osm_source.cpp     | 190 ++++++++++++++++++++---------------
 generator/osm_source.hpp     |  51 +++++++++-
 4 files changed, 206 insertions(+), 83 deletions(-)

diff --git a/coding/parse_xml.hpp b/coding/parse_xml.hpp
index db00519b03..5a129fbe02 100644
--- a/coding/parse_xml.hpp
+++ b/coding/parse_xml.hpp
@@ -40,6 +40,48 @@ uint64_t ParseXMLSequence(SequenceT & source, XMLDispatcherT & dispatcher, bool
   return res;
 }
 
+template <typename SequenceT, typename XMLDispatcherT>
+class ParserXMLSequence
+{
+public:
+  ParserXMLSequence(SequenceT & source, XMLDispatcherT & dispatcher)
+    : m_res(0)
+    , m_readed(0)
+    , m_source(source)
+    , m_parser(dispatcher, false /* useCharData */)
+  {
+    CHECK(m_parser.Create(), ());
+
+  }
+
+  bool Read()
+  {
+    char * buffer = static_cast<char *>(m_parser.GetBuffer(kBufferSize));
+    ASSERT(buffer, ());
+
+    m_readed = m_source.Read(buffer, kBufferSize);
+    if (m_readed == 0)
+      return false;
+
+    if (!m_parser.ParseBuffer(static_cast<uint32_t>(m_readed), false))
+    {
+      m_parser.PrintError();
+      return false;
+    }
+
+    m_res += m_readed;
+    return m_readed == kBufferSize;
+  }
+
+private:
+  uint32_t static const kBufferSize = 16 * 1024;
+
+  uint64_t m_res;
+  uint64_t m_readed;
+  SequenceT & m_source;
+  XmlParser<XMLDispatcherT> m_parser;
+};
+
 namespace
 {
 
diff --git a/generator/osm_o5m_source.hpp b/generator/osm_o5m_source.hpp
index 78c3b2b731..2e250232b0 100644
--- a/generator/osm_o5m_source.hpp
+++ b/generator/osm_o5m_source.hpp
@@ -215,7 +215,8 @@ public:
         NextValue();
       }
 
-      bool operator!=(Iterator const & iter) { return m_reader != iter.m_reader; }
+      bool operator==(Iterator const & iter) { return m_reader == iter.m_reader; }
+      bool operator!=(Iterator const & iter) { return !(*this == iter); }
 
       Iterator & operator++() { NextValue(); return *this; }
 
@@ -567,7 +568,8 @@ public:
       NextValue();
     }
 
-    bool operator!=(Iterator const & iter) { return m_reader != iter.m_reader; }
+    bool operator==(Iterator const & iter) { return m_reader == iter.m_reader; }
+    bool operator!=(Iterator const & iter) { return !(*this == iter); }
     Iterator & operator++() { NextValue(); return *this; }
 
     void NextValue() { m_reader = m_reader->ReadEntity(&m_entity) ? m_reader : nullptr; }
diff --git a/generator/osm_source.cpp b/generator/osm_source.cpp
index 4f7488c8e7..07a3601970 100644
--- a/generator/osm_source.cpp
+++ b/generator/osm_source.cpp
@@ -1,18 +1,16 @@
 #include "generator/osm_source.hpp"
 
+#include "generator/intermediate_data.hpp"
 #include "generator/intermediate_elements.hpp"
 #include "generator/osm_element.hpp"
-#include "generator/osm_o5m_source.hpp"
-#include "generator/osm_xml_source.hpp"
 #include "generator/towns_dumper.hpp"
+#include "generator/translator_factory.hpp"
 
 #include "platform/platform.hpp"
 
 #include "geometry/mercator.hpp"
 #include "geometry/tree4d.hpp"
 
-#include "coding/parse_xml.hpp"
-
 #include "base/assert.hpp"
 #include "base/stl_helpers.hpp"
 #include "base/file_name_utils.hpp"
@@ -24,6 +22,7 @@
 #include "defines.hpp"
 
 using namespace std;
+using namespace base::thread_pool::computational;
 
 namespace generator
 {
@@ -121,8 +120,13 @@ void BuildIntermediateDataFromXML(SourceReader & stream, cache::IntermediateData
 
 void ProcessOsmElementsFromXML(SourceReader & stream, function<void(OsmElement *)> processor)
 {
-  XMLSource parser([&](OsmElement * element) { processor(element); });
-  ParseXMLSequence(stream, parser);
+  ProcessorXmlElementsFromXml processorXmlElementsFromXml(stream);
+  OsmElement element;
+  while (processorXmlElementsFromXml.TryRead(element))
+  {
+    processor(&element);
+    element = {};
+  }
 }
 
 void BuildIntermediateDataFromO5M(SourceReader & stream, cache::IntermediateDataWriter & cache,
@@ -140,14 +144,29 @@ void BuildIntermediateDataFromO5M(SourceReader & stream, cache::IntermediateData
 
 void ProcessOsmElementsFromO5M(SourceReader & stream, function<void(OsmElement *)> processor)
 {
-  using Type = osm::O5MSource::EntityType;
-
-  osm::O5MSource dataset([&stream](uint8_t * buffer, size_t size)
+  ProcessorOsmElementsFromO5M processorOsmElementsFromO5M(stream);
+  OsmElement element;
+  while (processorOsmElementsFromO5M.TryRead(element))
   {
-    return stream.Read(reinterpret_cast<char *>(buffer), size);
-  });
+    processor(&element);
+    element = {};
+  }
+}
 
-  auto translate = [](Type t) -> OsmElement::EntityType {
+ProcessorOsmElementsFromO5M::ProcessorOsmElementsFromO5M(SourceReader & stream)
+  : m_stream(stream)
+  , m_dataset([&, this](uint8_t * buffer, size_t size) { return m_stream.Read(reinterpret_cast<char *>(buffer), size); })
+  , m_pos(m_dataset.begin())
+{
+}
+
+bool ProcessorOsmElementsFromO5M::TryRead(OsmElement & element)
+{
+  if (m_pos == m_dataset.end())
+    return false;
+
+  using Type = osm::O5MSource::EntityType;
+  auto const translate = [](Type t) -> OsmElement::EntityType {
     switch (t)
     {
     case Type::Node: return OsmElement::EntityType::Node;
@@ -157,48 +176,69 @@ void ProcessOsmElementsFromO5M(SourceReader & stream, function<void(OsmElement *
     }
   };
 
-  // Be careful, we could call Nodes(), Members(), Tags() from O5MSource::Entity
-  // only once (!). Because these functions read data from file simultaneously with
-  // iterating in loop. Furthermore, into Tags() method calls Nodes.Skip() and Members.Skip(),
-  // thus first call of Nodes (Members) after Tags() will not return any results.
-  // So don not reorder the "for" loops (!).
-
-  for (auto const & entity : dataset)
+  auto entity = *m_pos;
+  element.m_id = entity.id;
+  switch (entity.type)
   {
-    OsmElement p;
-    p.m_id = entity.id;
-
-    switch (entity.type)
-    {
-    case Type::Node:
-    {
-      p.m_type = OsmElement::EntityType::Node;
-      p.m_lat = entity.lat;
-      p.m_lon = entity.lon;
-      break;
-    }
-    case Type::Way:
-    {
-      p.m_type = OsmElement::EntityType::Way;
-      for (uint64_t nd : entity.Nodes())
-        p.AddNd(nd);
-      break;
-    }
-    case Type::Relation:
-    {
-      p.m_type = OsmElement::EntityType::Relation;
-      for (auto const & member : entity.Members())
-        p.AddMember(member.ref, translate(member.type), member.role);
-      break;
-    }
-    default: break;
-    }
-
-    for (auto const & tag : entity.Tags())
-      p.AddTag(tag.key, tag.value);
-
-    processor(&p);
+  case Type::Node:
+  {
+    element.m_type = OsmElement::EntityType::Node;
+    element.m_lat = entity.lat;
+    element.m_lon = entity.lon;
+    break;
   }
+  case Type::Way:
+  {
+    element.m_type = OsmElement::EntityType::Way;
+    for (uint64_t nd : entity.Nodes())
+      element.AddNd(nd);
+    break;
+  }
+  case Type::Relation:
+  {
+    element.m_type = OsmElement::EntityType::Relation;
+    for (auto const & member : entity.Members())
+      element.AddMember(member.ref, translate(member.type), member.role);
+    break;
+  }
+  default: break;
+  }
+
+  for (auto const & tag : entity.Tags())
+    element.AddTag(tag.key, tag.value);
+
+  ++m_pos;
+  return true;
+}
+
+ProcessorXmlElementsFromXml::ProcessorXmlElementsFromXml(SourceReader & stream)
+  : m_xmlSource([&, this](auto * element) { m_queue.emplace(*element); })
+  , m_parser(stream, m_xmlSource)
+{
+}
+
+bool ProcessorXmlElementsFromXml::TryReadFromQueue(OsmElement & element)
+{
+  if (m_queue.empty())
+    return false;
+
+  element = m_queue.front();
+  m_queue.pop();
+  return true;
+}
+
+bool ProcessorXmlElementsFromXml::TryRead(OsmElement & element)
+{
+  if (TryReadFromQueue(element))
+    return true;
+
+  while (m_parser.Read())
+  {
+    if (TryReadFromQueue(element))
+      return true;
+  }
+
+  return TryReadFromQueue(element);
 }
 
 ///////////////////////////////////////////////////////////////////////////////////////////////////
@@ -251,35 +291,27 @@ cache::IntermediateDataReader & CacheLoader::GetCache()
 
 bool GenerateIntermediateData(feature::GenerateInfo & info)
 {
-  try
+  auto nodes = cache::CreatePointStorageWriter(info.m_nodeStorageType,
+                                               info.GetIntermediateFileName(NODES_FILE));
+  cache::IntermediateDataWriter cache(*nodes, info);
+  TownsDumper towns;
+  SourceReader reader = info.m_osmFileName.empty() ? SourceReader() : SourceReader(info.m_osmFileName);
+
+  LOG(LINFO, ("Data source:", info.m_osmFileName));
+
+  switch (info.m_osmFileType)
   {
-    auto nodes = cache::CreatePointStorageWriter(info.m_nodeStorageType,
-                                                 info.GetIntermediateFileName(NODES_FILE));
-    cache::IntermediateDataWriter cache(*nodes, info);
-    TownsDumper towns;
-    SourceReader reader = info.m_osmFileName.empty() ? SourceReader() : SourceReader(info.m_osmFileName);
-
-    LOG(LINFO, ("Data source:", info.m_osmFileName));
-
-    switch (info.m_osmFileType)
-    {
-    case feature::GenerateInfo::OsmSourceType::XML:
-      BuildIntermediateDataFromXML(reader, cache, towns);
-      break;
-    case feature::GenerateInfo::OsmSourceType::O5M:
-      BuildIntermediateDataFromO5M(reader, cache, towns);
-      break;
-    }
-
-    cache.SaveIndex();
-    towns.Dump(info.GetIntermediateFileName(TOWNS_FILE));
-    LOG(LINFO, ("Added points count =", nodes->GetNumProcessedPoints()));
-  }
-  catch (Writer::Exception const & e)
-  {
-    LOG(LCRITICAL, ("Error with file:", e.what()));
+  case feature::GenerateInfo::OsmSourceType::XML:
+    BuildIntermediateDataFromXML(reader, cache, towns);
+    break;
+  case feature::GenerateInfo::OsmSourceType::O5M:
+    BuildIntermediateDataFromO5M(reader, cache, towns);
+    break;
   }
+
+  cache.SaveIndex();
+  towns.Dump(info.GetIntermediateFileName(TOWNS_FILE));
+  LOG(LINFO, ("Added points count =", nodes->GetNumProcessedPoints()));
   return true;
 }
-
 }  // namespace generator
diff --git a/generator/osm_source.hpp b/generator/osm_source.hpp
index 913b287367..70e012d917 100644
--- a/generator/osm_source.hpp
+++ b/generator/osm_source.hpp
@@ -1,13 +1,24 @@
 #pragma once
 
-#include "generator/emitter_interface.hpp"
+#include "generator/feature_generator.hpp"
+#include "generator/final_processor_intermediate_mwm.hpp"
+#include "generator/processor_interface.hpp"
 #include "generator/generate_info.hpp"
 #include "generator/intermediate_data.hpp"
+#include "generator/osm_o5m_source.hpp"
+#include "generator/osm_xml_source.hpp"
+#include "generator/translator_collection.hpp"
 #include "generator/translator_interface.hpp"
 
+#include "coding/parse_xml.hpp"
+
+#include "base/thread_pool_computational.hpp"
+#include "base/thread_safe_queue.hpp"
+
 #include <functional>
 #include <iostream>
 #include <memory>
+#include <queue>
 #include <sstream>
 #include <string>
 #include <vector>
@@ -62,7 +73,6 @@ private:
 
   DISALLOW_COPY(CacheLoader);
 };
-
 // This function is needed to generate intermediate data from OSM elements.
 // The translators collection contains translators that translate the OSM element into
 // some intermediate representation.
@@ -116,4 +126,41 @@ bool GenerateIntermediateData(feature::GenerateInfo & info);
 
 void ProcessOsmElementsFromO5M(SourceReader & stream, std::function<void(OsmElement *)> processor);
 void ProcessOsmElementsFromXML(SourceReader & stream, std::function<void(OsmElement *)> processor);
+
+class ProcessorOsmElementsInterface
+{
+public:
+  virtual ~ProcessorOsmElementsInterface() = default;
+
+  virtual bool TryRead(OsmElement & element) = 0;
+};
+
+class ProcessorOsmElementsFromO5M : public ProcessorOsmElementsInterface
+{
+public:
+  explicit ProcessorOsmElementsFromO5M(SourceReader & stream);
+
+  // ProcessorOsmElementsInterface overrides:
+  bool TryRead(OsmElement & element) override;
+
+private:
+  SourceReader & m_stream;
+  osm::O5MSource m_dataset;
+  osm::O5MSource::Iterator m_pos;
+};
+
+class ProcessorXmlElementsFromXml : public ProcessorOsmElementsInterface
+{
+public:
+  explicit ProcessorXmlElementsFromXml(SourceReader & stream);
+  // ProcessorOsmElementsInterface overrides:
+  bool TryRead(OsmElement & element) override;
+
+private:
+  bool TryReadFromQueue(OsmElement & element);
+
+  XMLSource m_xmlSource;
+  ParserXMLSequence<SourceReader, XMLSource> m_parser;
+  std::queue<OsmElement> m_queue;
+};
 }  // namespace generator