Libosmium  2.12.2
Fast and flexible C++ library for working with OpenStreetMap data
reader.hpp
Go to the documentation of this file.
1 #ifndef OSMIUM_IO_READER_HPP
2 #define OSMIUM_IO_READER_HPP
3 
4 /*
5 
6 This file is part of Osmium (http://osmcode.org/libosmium).
7 
8 Copyright 2013-2017 Jochen Topf <jochen@topf.org> and others (see README).
9 
10 Boost Software License - Version 1.0 - August 17th, 2003
11 
12 Permission is hereby granted, free of charge, to any person or organization
13 obtaining a copy of the software and accompanying documentation covered by
14 this license (the "Software") to use, reproduce, display, distribute,
15 execute, and transmit the Software, and to prepare derivative works of the
16 Software, and to permit third-parties to whom the Software is furnished to
17 do so, all subject to the following:
18 
19 The copyright notices in the Software and this entire statement, including
20 the above license grant, this restriction and the following disclaimer,
21 must be included in all copies of the Software, in whole or in part, and
22 all derivative works of the Software, unless such copies or derivative
23 works are solely in the form of machine-executable object code generated by
24 a source language processor.
25 
26 THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
27 IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
28 FITNESS FOR A PARTICULAR PURPOSE, TITLE AND NON-INFRINGEMENT. IN NO EVENT
29 SHALL THE COPYRIGHT HOLDERS OR ANYONE DISTRIBUTING THE SOFTWARE BE LIABLE
30 FOR ANY DAMAGES OR OTHER LIABILITY, WHETHER IN CONTRACT, TORT OR OTHERWISE,
31 ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
32 DEALINGS IN THE SOFTWARE.
33 
34 */
35 
36 #include <cerrno>
37 #include <cstdlib>
38 #include <fcntl.h>
39 #include <future>
40 #include <memory>
41 #include <string>
42 #include <system_error>
43 #include <thread>
44 #include <utility>
45 
46 #ifndef _WIN32
47 # include <sys/wait.h>
48 #endif
49 
50 #ifndef _MSC_VER
51 # include <unistd.h>
52 #endif
53 
55 #include <osmium/io/detail/input_format.hpp>
56 #include <osmium/io/detail/read_thread.hpp>
57 #include <osmium/io/detail/read_write.hpp>
58 #include <osmium/io/detail/queue_util.hpp>
59 #include <osmium/io/error.hpp>
60 #include <osmium/io/file.hpp>
61 #include <osmium/io/header.hpp>
62 #include <osmium/memory/buffer.hpp>
64 #include <osmium/thread/util.hpp>
65 #include <osmium/util/config.hpp>
66 
67 namespace osmium {
68 
69  namespace io {
70 
71  namespace detail {
72 
73  inline size_t get_input_queue_size() noexcept {
74  const size_t n = osmium::config::get_max_queue_size("INPUT", 20);
75  return n > 2 ? n : 2;
76  }
77 
78  inline size_t get_osmdata_queue_size() noexcept {
79  const size_t n = osmium::config::get_max_queue_size("OSMDATA", 20);
80  return n > 2 ? n : 2;
81  }
82 
83  } // namespace detail
84 
91  class Reader {
92 
94 
95  detail::ParserFactory::create_parser_type m_creator;
96 
97  enum class status {
98  okay = 0, // normal reading
99  error = 1, // some error occurred while reading
100  closed = 2, // close() called
101  eof = 3 // eof of file was reached without error
102  } m_status;
103 
105 
106  detail::future_string_queue_type m_input_queue;
107 
108  std::unique_ptr<osmium::io::Decompressor> m_decompressor;
109 
110  osmium::io::detail::ReadThreadManager m_read_thread_manager;
111 
112  detail::future_buffer_queue_type m_osmdata_queue;
113  detail::queue_wrapper<osmium::memory::Buffer> m_osmdata_queue_wrapper;
114 
115  std::future<osmium::io::Header> m_header_future;
117 
119 
120  size_t m_file_size;
121 
124 
126  m_read_which_entities = value;
127  }
128 
129  void set_option(osmium::io::read_meta value) noexcept {
130  m_read_metadata = value;
131  }
132 
133  // This function will run in a separate thread.
134  static void parser_thread(const detail::ParserFactory::create_parser_type& creator,
135  detail::future_string_queue_type& input_queue,
136  detail::future_buffer_queue_type& osmdata_queue,
137  std::promise<osmium::io::Header>&& header_promise,
138  osmium::osm_entity_bits::type read_which_entities,
139  osmium::io::read_meta read_metadata) {
140  std::promise<osmium::io::Header> promise = std::move(header_promise);
141  osmium::io::detail::parser_arguments args = {
142  input_queue,
143  osmdata_queue,
144  promise,
145  read_which_entities,
146  read_metadata
147  };
148  creator(args)->parse();
149  }
150 
151 #ifndef _WIN32
152 
163  static int execute(const std::string& command, const std::string& filename, int* childpid) {
164  int pipefd[2];
165  if (pipe(pipefd) < 0) {
166  throw std::system_error{errno, std::system_category(), "opening pipe failed"};
167  }
168  const pid_t pid = fork();
169  if (pid < 0) {
170  throw std::system_error{errno, std::system_category(), "fork failed"};
171  }
172  if (pid == 0) { // child
173  // close all file descriptors except one end of the pipe
174  for (int i = 0; i < 32; ++i) {
175  if (i != pipefd[1]) {
176  ::close(i);
177  }
178  }
179  if (dup2(pipefd[1], 1) < 0) { // put end of pipe as stdout/stdin
180  exit(1);
181  }
182 
183  ::open("/dev/null", O_RDONLY); // stdin
184  ::open("/dev/null", O_WRONLY); // stderr
185  // hack: -g switches off globbing in curl which allows [] to be used in file names
186  // this is important for XAPI URLs
187  // in theory this execute() function could be used for other commands, but it is
188  // only used for curl at the moment, so this is okay.
189  if (::execlp(command.c_str(), command.c_str(), "-g", filename.c_str(), nullptr) < 0) {
190  exit(1);
191  }
192  }
193  // parent
194  *childpid = pid;
195  ::close(pipefd[1]);
196  return pipefd[0];
197  }
198 #endif
199 
208  static int open_input_file_or_url(const std::string& filename, int* childpid) {
209  std::string protocol = filename.substr(0, filename.find_first_of(':'));
210  if (protocol == "http" || protocol == "https" || protocol == "ftp" || protocol == "file") {
211 #ifndef _WIN32
212  return execute("curl", filename, childpid);
213 #else
214  throw io_error{"Reading OSM files from the network currently not supported on Windows."};
215 #endif
216  } else {
217  return osmium::io::detail::open_for_reading(filename);
218  }
219  }
220 
221  public:
222 
245  template <typename... TArgs>
246  explicit Reader(const osmium::io::File& file, TArgs&&... args) :
247  m_file(file.check()),
248  m_creator(detail::ParserFactory::instance().get_creator_function(m_file)),
249  m_status(status::okay),
250  m_childpid(0),
251  m_input_queue(detail::get_input_queue_size(), "raw_input"),
252  m_decompressor(m_file.buffer() ?
253  osmium::io::CompressionFactory::instance().create_decompressor(file.compression(), m_file.buffer(), m_file.buffer_size()) :
254  osmium::io::CompressionFactory::instance().create_decompressor(file.compression(), open_input_file_or_url(m_file.filename(), &m_childpid))),
255  m_read_thread_manager(*m_decompressor, m_input_queue),
256  m_osmdata_queue(detail::get_osmdata_queue_size(), "parser_results"),
257  m_osmdata_queue_wrapper(m_osmdata_queue),
258  m_header_future(),
259  m_header(),
260  m_thread(),
261  m_file_size(m_decompressor->file_size()) {
262 
263  (void)std::initializer_list<int>{
264  (set_option(args), 0)...
265  };
266 
267  std::promise<osmium::io::Header> header_promise;
268  m_header_future = header_promise.get_future();
269  m_thread = osmium::thread::thread_handler{parser_thread, std::ref(m_creator), std::ref(m_input_queue), std::ref(m_osmdata_queue), std::move(header_promise), m_read_which_entities, m_read_metadata};
270  }
271 
272  template <typename... TArgs>
273  explicit Reader(const std::string& filename, TArgs&&... args) :
274  Reader(osmium::io::File(filename), std::forward<TArgs>(args)...) {
275  }
276 
277  template <typename... TArgs>
278  explicit Reader(const char* filename, TArgs&&... args) :
279  Reader(osmium::io::File(filename), std::forward<TArgs>(args)...) {
280  }
281 
282  Reader(const Reader&) = delete;
283  Reader& operator=(const Reader&) = delete;
284 
285  Reader(Reader&&) = default;
286  Reader& operator=(Reader&&) = default;
287 
288  ~Reader() noexcept {
289  try {
290  close();
291  } catch (...) {
292  // Ignore any exceptions because destructor must not throw.
293  }
294  }
295 
304  void close() {
305  m_status = status::closed;
306 
307  m_read_thread_manager.stop();
308 
309  m_osmdata_queue_wrapper.drain();
310 
311  try {
312  m_read_thread_manager.close();
313  } catch (...) {
314  // Ignore any exceptions.
315  }
316 
317 #ifndef _WIN32
318  if (m_childpid) {
319  int status;
320  const pid_t pid = ::waitpid(m_childpid, &status, 0);
321 #pragma GCC diagnostic push
322 #pragma GCC diagnostic ignored "-Wold-style-cast"
323  if (pid < 0 || !WIFEXITED(status) || WEXITSTATUS(status) != 0) {
324  throw std::system_error{errno, std::system_category(), "subprocess returned error"};
325  }
326 #pragma GCC diagnostic pop
327  m_childpid = 0;
328  }
329 #endif
330  }
331 
339  if (m_status == status::error) {
340  throw io_error{"Can not get header from reader when in status 'error'"};
341  }
342 
343  try {
344  if (m_header_future.valid()) {
345  m_header = m_header_future.get();
346  }
347  } catch (...) {
348  close();
349  m_status = status::error;
350  throw;
351  }
352 
353  return m_header;
354  }
355 
365  osmium::memory::Buffer buffer;
366 
367  if (m_status != status::okay) {
368  throw io_error{"Can not read from reader when in status 'closed', 'eof', or 'error'"};
369  }
370 
371  if (m_read_which_entities == osmium::osm_entity_bits::nothing) {
372  m_status = status::eof;
373  return buffer;
374  }
375 
376  try {
377  // m_input_format.read() can return an invalid buffer to signal EOF,
378  // or a valid buffer with or without data. A valid buffer
379  // without data is not an error, it just means we have to
380  // keep getting the next buffer until there is one with data.
381  while (true) {
382  buffer = m_osmdata_queue_wrapper.pop();
383  if (detail::at_end_of_data(buffer)) {
384  m_status = status::eof;
385  m_read_thread_manager.close();
386  return buffer;
387  }
388  if (buffer.committed() > 0) {
389  return buffer;
390  }
391  }
392  } catch (...) {
393  close();
394  m_status = status::error;
395  throw;
396  }
397  }
398 
403  bool eof() const {
404  return m_status == status::eof || m_status == status::closed;
405  }
406 
411  size_t file_size() const noexcept {
412  return m_file_size;
413  }
414 
429  size_t offset() const noexcept {
430  return m_decompressor->offset();
431  }
432 
433  }; // class Reader
434 
443  template <typename... TArgs>
446 
447  Reader reader(std::forward<TArgs>(args)...);
448  while (osmium::memory::Buffer read_buffer = reader.read()) {
449  buffer.add_buffer(read_buffer);
450  buffer.commit();
451  }
452 
453  return buffer;
454  }
455 
456  } // namespace io
457 
458 } // namespace osmium
459 
460 #endif // OSMIUM_IO_READER_HPP
detail::queue_wrapper< osmium::memory::Buffer > m_osmdata_queue_wrapper
Definition: reader.hpp:113
status
Definition: reader.hpp:97
osmium::memory::Buffer read()
Definition: reader.hpp:364
Reader(const osmium::io::File &file, TArgs &&... args)
Definition: reader.hpp:246
type
Definition: entity_bits.hpp:63
size_t file_size(int fd)
Definition: file.hpp:70
bool eof() const
Definition: reader.hpp:403
int m_childpid
Definition: reader.hpp:104
Reader(const char *filename, TArgs &&... args)
Definition: reader.hpp:278
std::future< osmium::io::Header > m_header_future
Definition: reader.hpp:115
Definition: reader_iterator.hpp:39
std::unique_ptr< osmium::io::Decompressor > m_decompressor
Definition: reader.hpp:108
object or changeset
Definition: entity_bits.hpp:76
detail::future_string_queue_type m_input_queue
Definition: reader.hpp:106
osmium::memory::Buffer read_file(TArgs &&... args)
Definition: reader.hpp:444
osmium::io::File m_file
Definition: reader.hpp:93
Definition: file.hpp:72
Namespace for everything in the Osmium library.
Definition: assembler.hpp:63
void set_option(osmium::osm_entity_bits::type value) noexcept
Definition: reader.hpp:125
Definition: attr.hpp:333
void add_buffer(const Buffer &buffer)
Definition: buffer.hpp:504
osmium::io::detail::ReadThreadManager m_read_thread_manager
Definition: reader.hpp:110
size_t file_size() const noexcept
Definition: reader.hpp:411
static int execute(const std::string &command, const std::string &filename, int *childpid)
Definition: reader.hpp:163
Reader(const std::string &filename, TArgs &&... args)
Definition: reader.hpp:273
Definition: reader.hpp:91
~Reader() noexcept
Definition: reader.hpp:288
Definition: error.hpp:44
osmium::io::Header header()
Definition: reader.hpp:338
void set_option(osmium::io::read_meta value) noexcept
Definition: reader.hpp:129
size_t committed() const noexcept
Definition: buffer.hpp:263
size_t get_max_queue_size(const char *queue_name, size_t default_value) noexcept
Definition: config.hpp:69
Definition: buffer.hpp:97
size_t m_file_size
Definition: reader.hpp:120
osmium::thread::thread_handler m_thread
Definition: reader.hpp:118
static int open_input_file_or_url(const std::string &filename, int *childpid)
Definition: reader.hpp:208
detail::future_buffer_queue_type m_osmdata_queue
Definition: reader.hpp:112
osmium::io::Header m_header
Definition: reader.hpp:116
read_meta
Definition: file_format.hpp:52
Definition: entity_bits.hpp:67
static void parser_thread(const detail::ParserFactory::create_parser_type &creator, detail::future_string_queue_type &input_queue, detail::future_buffer_queue_type &osmdata_queue, std::promise< osmium::io::Header > &&header_promise, osmium::osm_entity_bits::type read_which_entities, osmium::io::read_meta read_metadata)
Definition: reader.hpp:134
Definition: compression.hpp:138
size_t offset() const noexcept
Definition: reader.hpp:429
std::string get(const std::string &key, const std::string &default_value="") const noexcept
Definition: options.hpp:124
detail::ParserFactory::create_parser_type m_creator
Definition: reader.hpp:95
Definition: header.hpp:68
void close()
Definition: reader.hpp:304
size_t commit()
Definition: buffer.hpp:358
Definition: util.hpp:85