File indexing completed on 2024-05-05 04:42:57

 /*
     SPDX-FileCopyrightText: 2020 Volker Krause <vkrause@kde.org>
 
     SPDX-License-Identifier: LGPL-2.0-or-later
 */
 
 #include "overpassquerymanager.h"
 #include "overpassquery.h"
 
 #include <QDateTime>
 #include <QNetworkAccessManager>
 #include <QNetworkDiskCache>
 #include <QNetworkReply>
 #include <QStandardPaths>
 #include <QTimer>
 #include <QUrl>
 #include <QUrlQuery>
 
 #include <chrono>
 #include <deque>
 
 using namespace OSM;
 
 namespace OSM {
 struct OverpassQueryTask {
     OverpassQuery *query = nullptr;
     QRectF bbox;
     bool forceReload = false;
 };
 
 struct OverpassQueryExecutor {
     QUrl endpoint;
     std::chrono::seconds cooldownTime = std::chrono::seconds(3);
     QDateTime nextSlot;
     std::unique_ptr<OverpassQueryTask> task;
 };
 
 class OverpassQueryManagerPrivate {
 public:
     void executeTasks();
     void taskFinished(OverpassQueryExecutor *executor, QNetworkReply *reply);
     void checkQueryFinished(OverpassQuery *query) const;
     void cancelQuery(OverpassQuery *query);
 
     OverpassQueryManager *q;
     QNetworkAccessManager *m_nam;
     QTimer *m_nextTaskTimer;
     std::vector<OverpassQueryExecutor> m_executors;
     std::deque<std::unique_ptr<OverpassQueryTask>> m_tasks;
 };
 }
 
 static const char* executor_configs[] = {
     "https://overpass-api.de/api/interpreter",
     "https://1.overpass.kumi.systems/api/interpreter",
     "https://2.overpass.kumi.systems/api/interpreter",
     "https://3.overpass.kumi.systems/api/interpreter",
     "https://4.overpass.kumi.systems/api/interpreter",
 };
 
 OverpassQueryManager::OverpassQueryManager(QObject *parent)
     : QObject(parent)
     , d(new OverpassQueryManagerPrivate)
 {
     d->q = this;
     d->m_nam = new QNetworkAccessManager(this);
     d->m_nam->setRedirectPolicy(QNetworkRequest::NoLessSafeRedirectPolicy);
     d->m_nam->setStrictTransportSecurityEnabled(true);
     d->m_nam->enableStrictTransportSecurityStore(true, QStandardPaths::writableLocation(QStandardPaths::GenericCacheLocation) + QLatin1String("/org.kde.osm/hsts/"));
 
     auto diskCache = new QNetworkDiskCache;
     diskCache->setCacheDirectory(QStandardPaths::writableLocation(QStandardPaths::GenericCacheLocation) + QLatin1String("/org.kde.osm/overpass-cache/"));
     diskCache->setMaximumCacheSize(1'000'000'000); // 1GB
     d->m_nam->setCache(diskCache);
 
     d->m_nextTaskTimer = new QTimer(this);
     d->m_nextTaskTimer->setSingleShot(true);
     connect(d->m_nextTaskTimer, &QTimer::timeout, this, [this]() { d->executeTasks(); });
 
     for (const auto &config : executor_configs) {
         OverpassQueryExecutor executor;
         executor.endpoint = QUrl(QString::fromUtf8(config));
         d->m_executors.push_back(std::move(executor));
     }
 }
 
 OverpassQueryManager::~OverpassQueryManager() = default;
 
 void OverpassQueryManager::execute(OverpassQuery *query)
 {
     // validate input
     if (query->query().isEmpty() || query->boundingBox().isNull() || !query->boundingBox().isValid() || query->tileSize().isNull() || !query->tileSize().isValid()) {
         query->m_error = OverpassQuery::QueryError;
         QTimer::singleShot(0, query, &OverpassQuery::finished);
         return;
     }
 
     // generate tasks for the query
     const auto xTileCount = std::max<int>(1, query->boundingBox().width() / query->tileSize().width());
     const auto yTileCount = std::max<int>(1, query->boundingBox().height() / query->tileSize().height());
     const auto xTileSize = query->boundingBox().width() / xTileCount;
     const auto yTileSize = query->boundingBox().height() / yTileCount;
     qDebug() << "Creating" << xTileCount * yTileCount << "tasks with tile size" << xTileSize << "x" << yTileSize;
     for (auto x = 0; x < xTileCount; ++x) {
         for (auto y = 0; y < yTileCount; ++y) {
             auto task = std::make_unique<OverpassQueryTask>();
             task->query = query;
             task->bbox = { query->boundingBox().x() + x * xTileSize, query->boundingBox().y() + y * yTileSize, xTileSize, yTileSize };
             d->m_tasks.push_back(std::move(task));
         }
     }
 
     d->executeTasks();
 }
 
 void OverpassQueryManagerPrivate::executeTasks()
 {
     const auto now = QDateTime::currentDateTimeUtc();
     std::chrono::seconds nextSlot = std::chrono::hours(1);
 
     for (auto &executor : m_executors) {
         if (m_tasks.empty()) { // nothing to do
             return;
         }
 
         if (executor.task) { // executor is busy already
             continue;
         }
 
         if (executor.nextSlot > now) { // executor is still in rate limit cooldown
             nextSlot = std::min(std::chrono::seconds(now.secsTo(executor.nextSlot)), nextSlot);
             nextSlot += std::chrono::seconds(1); // for msec rounding errors that would other wise give us a busy loop
             if (m_tasks.front()->forceReload) {
                 continue;
             }
         }
 
         executor.task = std::move(m_tasks.front());
         m_tasks.pop_front();
 
         // actually execute query
         auto url = executor.endpoint;
         QUrlQuery params;
         params.addQueryItem(QStringLiteral("data"), executor.task->query->query(executor.task->bbox));
         url.setQuery(params);
         QNetworkRequest req(url);
         req.setAttribute(QNetworkRequest::CacheLoadControlAttribute, executor.task->forceReload ? QNetworkRequest::PreferNetwork : QNetworkRequest::AlwaysCache);
         auto reply = m_nam->get(req);
         // TODO enable stream parsing for XML replies by connecting to QNetworkReply::readyRead
         QObject::connect(reply, &QNetworkReply::finished, q, [this, &executor, reply]() {
             taskFinished(&executor, reply);
             reply->deleteLater();
         });
     }
 
     m_nextTaskTimer->start(nextSlot);
 }
 
 void OverpassQueryManagerPrivate::taskFinished(OverpassQueryExecutor *executor, QNetworkReply *reply)
 {
     auto query = executor->task->query;
     if (reply->error() == QNetworkReply::UnknownContentError && reply->attribute(QNetworkRequest::HttpStatusCodeAttribute).toInt() == 429) {
         // rate limiting error
         executor->cooldownTime *= 2;
         qDebug() << "rate limit error, increasing cooldown time to" << executor->cooldownTime.count() << "seconds";
         m_tasks.push_back(std::move(executor->task));
     } else if (reply->error() == QNetworkReply::ContentNotFoundError && !executor->task->forceReload) {
         // cache miss, retry from network
         executor->task->forceReload = true;
         m_tasks.push_back(std::move(executor->task));
     } else if (reply->error() != QNetworkReply::NoError) {
         // TODO disable affected executors here and reschedule the failed task, rather than cancelling entirely
         qDebug() << reply->error() << reply->errorString() << reply->attribute(QNetworkRequest::HttpStatusCodeAttribute) << reply->readAll();
         query->m_error = OverpassQuery::NetworkError;
         cancelQuery(query);
     } else {
         const auto queryError = query->processReply(reply);
         // on query timeout, break up the task in 4 sub-tasks, if we are allowed to
         if (queryError == OverpassQuery::QueryTimeout
             && executor->task->bbox.width() > query->minimumTileSize().width()
             && executor->task->bbox.height() > query->minimumTileSize().height())
         {
             qDebug() << "Splitting task due to query timeout:" << executor->task->bbox;
             const auto xTileSize = executor->task->bbox.width() / 2.0;
             const auto yTileSize = executor->task->bbox.height() / 2.0;
             for (auto x = 0; x < 2; ++x) {
                 for (auto y = 0; y < 2; ++y) {
                     auto task = std::make_unique<OverpassQueryTask>();
                     task->query = query;
                     task->bbox = { executor->task->bbox.x() + x * xTileSize, executor->task->bbox.y() + y * yTileSize, xTileSize, yTileSize };
                     m_tasks.push_back(std::move(task));
                 }
             }
         }
         else if (queryError != OverpassQuery::NoError) {
             if (executor->task->forceReload) {
                 query->m_error = queryError;
                 cancelQuery(query);
             } else {
                 // query error in cached result, retry
                 executor->task->forceReload = true;
                 m_tasks.push_back(std::move(executor->task));
             }
         }
     }
 
     // free the executor for the next query
     executor->task.reset();
     executor->nextSlot = QDateTime::currentDateTimeUtc().addSecs(executor->cooldownTime.count());
 
     checkQueryFinished(query);
     executeTasks();
 }
 
 void OverpassQueryManagerPrivate::checkQueryFinished(OverpassQuery *query) const
 {
     if (std::any_of(m_executors.begin(), m_executors.end(), [query](const auto &executor) { return executor.task && executor.task->query == query; })
         || std::any_of(m_tasks.begin(), m_tasks.end(), [query](const auto &task) { return task->query == query; }))
         return;
     Q_EMIT query->finished();
 }
 
 void OverpassQueryManagerPrivate::cancelQuery(OverpassQuery *query)
 {
     qDebug() << "cancelling query...";
     m_tasks.erase(std::remove_if(m_tasks.begin(), m_tasks.end(), [query](const auto &task) { return task->query == query; }), m_tasks.end());
     checkQueryFinished(query);
 }