request_manager.cpp

Go to the documentation of this file.

#include "server/request_manager.hpp"
#include "server/http_request.hpp"
#include "server/meta_identifier.hpp"
#include "server/tile_identifier.hpp"
#include "server/renderer/render_canvas.hpp"
#include "server/job.hpp"
#include "general/configuration.hpp"


class RequestManager::RunningQueue
{
private:
    boost::mutex runningMutex;
    std::list<Job*> jobs;

public:
    bool start(Job* job, const shared_ptr<MetaIdentifier>& mid)
    {
        boost::mutex::scoped_lock lock(runningMutex);
        for (auto j : jobs) {
            if (j->getIdentifier() == mid)
                return false;
        }

        jobs.push_back(job);
        return true;
    }

    bool start(Job* job, const shared_ptr<TileIdentifier>& ti, const shared_ptr<HttpRequest>& r)
    {
        boost::mutex::scoped_lock lock(runningMutex);
        for (auto j : jobs) {
            if (j->getIdentifier()->contains(ti)) {
                j->addRequest(r, ti);
                return false;
            }
        }

        jobs.push_back(job);
        job->addRequest(r, ti);
        return true;
    }

    void finished(Job* job)
    {
        boost::mutex::scoped_lock lock(runningMutex);
        jobs.remove(job);
    }
};

RequestManager::RequestManager( const shared_ptr<Configuration>& config,
                                const shared_ptr<Geodata>& data,
                                const shared_ptr<Renderer>& renderer,
                                const shared_ptr<Cache>& cache,
                                const shared_ptr<StylesheetManager>& ssm )
    : config(config)
    , data(data)
    , renderer(renderer)
    , cache(cache)
    , ssm(ssm)
    , preventStop(jobPool)
    , running(new RunningQueue())
{
    int threads = config->get<int>(opt::server::num_threads);
    currentPrerenderingThreads = 0;
    for (int i = 0; i < threads; ++i) {
        auto worker = boost::make_shared<boost::thread>(
            boost::bind(&boost::asio::io_service::run, &jobPool)
        );
        workers.push_back(worker);
        factories.push(boost::make_shared<RenderCanvasFactory>());
    }

    // start prerender timing
        prerender_start = std::chrono::system_clock::now();
}

RequestManager::~RequestManager() {
    stop();
    LOG_SEV(request_log, debug) << "RequestManager destructed";
}

void RequestManager::stop()
{
    jobPool.stop();
    for (auto worker : workers) {
        worker->join();
    }
    workers.clear();
}


void RequestManager::enqueue(const shared_ptr<HttpRequest>& r)
{
    boost::mutex::scoped_lock userLock(userRJMutex);
    if (userRequests.size() >= config->get<int>(opt::server::max_queue_size)) {
        r->answer(HttpRequest::Reply::service_unavailable);
        return;
    }
    userRequests.push(r);
    userLock.unlock();

    jobPool.post( boost::bind(&RequestManager::processNextRequest, shared_from_this()) );
}


void RequestManager::enqueue(const shared_ptr<MetaIdentifier>& ti)
{
    preRJMutex.lock();
    preRenderRequests.push(ti);
    preRJMutex.unlock();

    jobPool.post( boost::bind(&RequestManager::processNextRequest, shared_from_this()) );
}

void RequestManager::processNextRequest()
{
    if (!nextUserRequest())
        if (!nextPreRenderRequest())
            LOG_SEV(request_log, error) << "Trying to run a job, but there is none.";
}

bool RequestManager::nextUserRequest()
{
    boost::mutex::scoped_lock userLock(userRJMutex);
    if (userRequests.empty())
        return false;

    shared_ptr<HttpRequest> req = userRequests.front();
    userRequests.pop();
    userLock.unlock();

    shared_ptr<TileIdentifier> ti;
    try
    {
        ti = TileIdentifier::Create(req->getURL(), this->ssm, config);
    }
    catch (excp::MalformedURLException &e)
    {
        LOG_SEV(request_log, info) << "MalformedURLException: "
            << e.what()  << " Url: " << req->getURL();

        req->answer(HttpRequest::Reply::forbidden);
    }
    catch (excp::UnknownImageFormatException &e)
    {
        LOG_SEV(request_log, info) << "UnknownImageFormatException: "
            << e.what()  << " Url: " << req->getURL();

        req->answer(HttpRequest::Reply::not_implemented);
    }

    // identifier could not be parsed
    if (!ti)
        return true;

    shared_ptr<MetaIdentifier> mid = MetaIdentifier::Create(ti);

    factoriesMutex.lock();
    shared_ptr<RenderCanvasFactory> factory = factories.front();
    factories.pop();
    factoriesMutex.unlock();

    shared_ptr<RenderCanvas> canvas = factory->getCanvas(ti->getImageFormat());

    Job job(mid, config, shared_from_this(), canvas);

    // check if tiles are already in progress
    if (running->start(&job, ti, req))
    {
        job.process();

        running->finished(&job);

        job.deliver();
    }

    factoriesMutex.lock();
    factories.push(factory);
    factoriesMutex.unlock();

    return true;
}

bool RequestManager::nextPreRenderRequest()
{
    boost::mutex::scoped_lock preLock(preRJMutex);
    if (preRenderRequests.empty())
        return false;

    shared_ptr<MetaIdentifier> mid = preRenderRequests.front();
    preRenderRequests.pop();
    currentPrerenderingThreads++;
    preLock.unlock();

    factoriesMutex.lock();
    shared_ptr<RenderCanvasFactory> factory = factories.front();
    factories.pop();
    factoriesMutex.unlock();

    shared_ptr<RenderCanvas> canvas = factory->getCanvas(mid->getImageFormat());

    Job job(mid, config, shared_from_this(), canvas);

    // check if tiles are already in progress
    if (running->start(&job, mid))
    {
        job.process();

        running->finished(&job);

        job.deliver();
    }

    factoriesMutex.lock();
    factories.push(factory);
    factoriesMutex.unlock();

    if (!job.isEmpty() && mid->getZoom() < config->get<int>(opt::server::prerender_level)) {
        std::vector<shared_ptr<MetaIdentifier>> children;
        mid->getSubIdentifiers(children);
        for (auto& c : children)
            enqueue(c);
    }

    preLock.lock();
    currentPrerenderingThreads--;
    preLock.unlock();

    if (currentPrerenderingThreads == 0 && preRenderRequests.size() == 0) {
        prerender_stop = std::chrono::system_clock::now();
        int elapsed_seconds = std::chrono::duration_cast<std::chrono::seconds>
                     (prerender_stop-prerender_start).count();
        LOG_SEV(request_log, info) << "Prerendering finished in "
            << std::setfill('0') << std::setw(2) << elapsed_seconds / 60 << ":" << elapsed_seconds % 60;
    }

    return true;
}

shared_ptr<Geodata> RequestManager::getGeodata() const
{
    return data;
}


shared_ptr<StylesheetManager> RequestManager::getStylesheetManager() const
{
    return ssm;
}


shared_ptr<Cache> RequestManager::getCache() const
{
    return cache;
}


shared_ptr<Renderer> RequestManager::getRenderer() const
{
    return renderer;
}