/**

* SPDX-License-Identifier: GPL-2.0-or-later

*

* This file is part of osm2pgsql (https://osm2pgsql.org/).

*

* Copyright (C) 2006-2026 by the osm2pgsql developer community.

* For a full list of authors see the git log.

*/

#include <cassert>

#include <chrono>

#include <functional>

#include <future>

#include <memory>

#include <mutex>

#include <utility>

#include <vector>

#include "db-copy.hpp"

#include "format.hpp"

#include "logging.hpp"

#include "middle.hpp"

#include "options.hpp"

#include "osmdata.hpp"

#include "output.hpp"

#include "util.hpp"

osmdata_t::osmdata_t(std::shared_ptr<middle_t> mid,

std::shared_ptr<output_t> output, options_t const &options)

: m_mid(std::move(mid)), m_output(std::move(output)),

m_connection_params(options.connection_params), m_bbox(options.bbox),

m_num_procs(options.num_procs), m_append(options.append),

m_droptemp(options.droptemp)

{

assert(m_mid);

assert(m_output);

m_output->start();

}

void osmdata_t::node(osmium::Node const &node)

{

if (node.visible()) {

if (!node.location().valid()) {

log_warn("Ignored node {} (version {}) with invalid location.",

node.id(), node.version());

return;

}

if (m_bbox.valid() && !m_bbox.contains(node.location())) {

return;

}

}

m_mid->node(node);

if (node.deleted()) {

m_output->node_delete(node);

return;

}

if (m_append) {

m_output->node_modify(node);

// Version 1 means this is a new node, so there can't be an existing

// way or relation referencing it, so we don't have to add that node

// to the list of changed nodes. If the input data doesn't contain

// object versions this will still work, because then the version is 0.

if (node.version() != 1) {

m_changed_nodes.push_back(node.id());

}

} else {

m_output->node_add(node);

}

}

void osmdata_t::after_nodes()

{

m_mid->after_nodes();

m_output->after_nodes();

if (!m_append) {

return;

}

if (!m_changed_nodes.empty()) {

m_mid->get_node_parents(m_changed_nodes, &m_ways_pending_tracker,

&m_rels_pending_tracker);

m_changed_nodes.clear();

}

}

void osmdata_t::way(osmium::Way &way)

{

m_mid->way(way);

if (way.deleted()) {

m_output->way_delete(&way);

return;

}

if (m_append) {

m_output->way_modify(&way);

// Version 1 means this is a new way, so there can't be an existing

// relation referencing it, so we don't have to add that way to the

// list of changed ways. If the input data doesn't contain object

// versions this will still work, because then the version is 0.

if (way.version() != 1) {

m_changed_ways.push_back(way.id());

}

} else {

m_output->way_add(&way);

}

}

void osmdata_t::after_ways()

{

m_mid->after_ways();

m_output->after_ways();

if (!m_append) {

return;

}

if (!m_changed_ways.empty()) {

if (!m_ways_pending_tracker.empty()) {

// Remove ids from changed ways in the input data from

// m_ways_pending_tracker, because they have already been

// processed.

m_ways_pending_tracker.remove_ids_if_in(m_changed_ways);

// Add the list of pending way ids to the list of changed ways,

// because we need the parents for them, too.

m_changed_ways.merge_sorted(m_ways_pending_tracker);

}

m_mid->get_way_parents(m_changed_ways, &m_rels_pending_tracker);

m_changed_ways.clear();

return;

}

if (!m_ways_pending_tracker.empty()) {

m_mid->get_way_parents(m_ways_pending_tracker, &m_rels_pending_tracker);

}

}

void osmdata_t::relation(osmium::Relation const &rel)

{

if (rel.members().size() > 32767) {

log_warn(

"Relation id {} ignored, because it has more than 32767 members",

rel.id());

return;

}

if (m_append && !rel.deleted()) {

m_output->select_relation_members(rel.id());

}

m_mid->relation(rel);

if (rel.deleted()) {

m_output->relation_delete(rel);

return;

}

if (m_append) {

m_output->relation_modify(rel);

m_changed_relations.push_back(rel.id());

} else {

m_output->relation_add(rel);

}

}

void osmdata_t::after_relations()

{

m_mid->after_relations();

m_output->after_relations();

if (m_append) {

// Remove ids from changed relations in the input data from

// m_rels_pending_tracker, because they have already been processed.

m_rels_pending_tracker.remove_ids_if_in(m_changed_relations);

m_changed_relations.clear();

}

m_output->sync();

}

namespace {

/**

* After all objects in a change file have been processed, all objects

* depending on the changed objects must also be processed. This class

* handles this extra processing by starting a number of threads and doing

* the processing in them.

*/

class multithreaded_processor_t

{

public:

multithreaded_processor_t(connection_params_t const &connection_params,

std::shared_ptr<middle_t> const &mid,

std::shared_ptr<output_t> output,

std::size_t thread_count)

: m_output(std::move(output))

{

assert(mid);

assert(m_output);

// For each thread we create a clone of the output.

for (std::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>(connection_params);

m_clones.push_back(m_output->clone(midq, copy_thread));

}

}

/**

* Process all ways in the list.

*

* \param list List of way ids to work on. The list is moved into the

* function.

*/

void process_ways(idlist_t &&list)

{

process_queue("way", std::move(list), &output_t::pending_way);

}

/**

* Process all relations in the list.

*

* \param list List of relation ids to work on. The list is moved into the

* function.

*/

void process_relations(idlist_t &&list)

{

process_queue("relation", std::move(list), &output_t::pending_relation);

}

/**

* Process all relations in the list in stage1c.

*

* \param list List of relation ids to work on. The list is moved into the

* function.

*/

void process_relations_stage1c(idlist_t &&list)

{

process_queue("relation", std::move(list),

&output_t::pending_relation_stage1c);

}

private:

/// Get the next id from the queue.

static osmid_t pop_id(idlist_t *queue, std::mutex *mutex)

{

osmid_t id = 0;

std::lock_guard<std::mutex> const lock{*mutex};

if (!queue->empty()) {

id = queue->pop_id();

}

return id;

}

// Pointer to a member function of output_t taking an osm_id

using output_member_fn_ptr = void (output_t::*)(osmid_t);

/**

* Runs in the worker threads: As long as there are any, get ids from

* the queue and let the output process it by calling "func".

*/

static void run(std::shared_ptr<output_t> const &output, idlist_t *queue,

std::mutex *mutex, output_member_fn_ptr func)

{

while (osmid_t const id = pop_id(queue, mutex)) {

(output.get()->*func)(id);

}

output->sync();

}

/// Runs in a worker thread: Update progress display once per second.

static void print_stats(idlist_t *queue, std::mutex *mutex)

{

std::size_t queue_size = 0;

do {

mutex->lock();

queue_size = queue->size();

mutex->unlock();

if (get_logger().show_progress()) {

fmt::print(stderr, "\rLeft to process: {}...", queue_size);

}

std::this_thread::sleep_for(std::chrono::seconds{1});

} while (queue_size > 0);

}

void process_queue(char const *type, idlist_t list,

output_member_fn_ptr function)

{

auto const ids_queued = list.size();

util::timer_t timer;

if (ids_queued < 100) {

// Worker startup is quite expensive. Run the processing directly

// when only few items need to be processed.

log_info("Going over {} pending {}s", ids_queued, type);

for (auto const oid : list) {

(m_clones[0].get()->*function)(oid);

}

m_clones[0]->sync();

} else {

log_info("Going over {} pending {}s (using {} threads)", ids_queued,

type, m_clones.size());

std::vector<std::future<void>> workers;

workers.reserve(m_clones.size() + 1);

for (auto const &clone : m_clones) {

workers.push_back(std::async(std::launch::async, run,

std::cref(clone), &list, &m_mutex,

function));

}

workers.push_back(

std::async(std::launch::async, print_stats, &list, &m_mutex));

for (auto &worker : workers) {

try {

worker.get();

} catch (...) {

// Drain the queue, so that the other workers finish early.

m_mutex.lock();

list.clear();

m_mutex.unlock();

throw;

}

}

if (get_logger().show_progress()) {

fmt::print(stderr, "\rLeft to process: 0.\n");

}

}

timer.stop();

log_info("Processing {} pending {}s took {} at a rate of {:.2f}/s",

ids_queued, type,

util::human_readable_duration(timer.elapsed()),

timer.per_second(ids_queued));

}

/// Clones of output, one clone per thread.

std::vector<std::shared_ptr<output_t>> m_clones;

/// The output.

std::shared_ptr<output_t> m_output;

/// Mutex to make sure worker threads coordinate access to queue.

std::mutex m_mutex;

};

} // anonymous namespace

void osmdata_t::process_dependents()

{

multithreaded_processor_t proc{m_connection_params, m_mid, m_output,

m_num_procs};

// stage 1b processing: process parents of changed objects

if (!m_ways_pending_tracker.empty() || !m_rels_pending_tracker.empty()) {

if (!m_ways_pending_tracker.empty()) {

m_ways_pending_tracker.sort_unique();

proc.process_ways(std::move(m_ways_pending_tracker));

}

if (!m_rels_pending_tracker.empty()) {

m_rels_pending_tracker.sort_unique();

proc.process_relations(std::move(m_rels_pending_tracker));

}

}

// stage 1c processing: mark parent relations of marked objects as changed

auto const &marked_nodes = m_output->get_marked_node_ids();

auto const &marked_ways = m_output->get_marked_way_ids();

if (marked_nodes.empty() && marked_ways.empty()) {

return;

}

// process parent relations of marked nodes and ways

idlist_t rels_pending_tracker{};

m_mid->get_node_parents(marked_nodes, nullptr, &rels_pending_tracker);

m_mid->get_way_parents(marked_ways, &rels_pending_tracker);

if (rels_pending_tracker.empty()) {

return;

}

rels_pending_tracker.sort_unique();

proc.process_relations_stage1c(std::move(rels_pending_tracker));

}

void osmdata_t::stop()

{

if (m_append) {

process_dependents();

}

// Run stage 2 processing: Reprocess objects marked in stage 1 (if any).

m_output->reprocess_marked();

// Run postprocessing on database: Clustering and index creation.

m_output->free_middle_references();

if (m_droptemp) {

// When dropping middle tables, make sure they are gone before

// indexing starts.

m_mid->stop();

}

m_output->stop();

if (!m_droptemp) {

// When keeping middle tables, there is quite a large index created

// which is better done after the output tables have been copied.

// Note that --disable-parallel-indexing needs to be used to really

// force the order.

m_mid->stop();

}

// Waiting here for pool to execute all tasks.

m_mid->wait();

m_output->wait();

}