plasma-framework/runnermanager.cpp

/*
 *   Copyright (C) 2006-2007 Ryan P. Bitanga <ryan.bitanga@gmail.com> 
 *   Copyright (C) 2006 Aaron Seigo <aseigo@kde.org> 
 *   Copyright 2008 Jordi Polo <mumismo@gmail.com>
 *
 *   This program is free software; you can redistribute it and/or modify
 *   it under the terms of the GNU Library General Public License as
 *   published by the Free Software Foundation; either version 2, or
 *   (at your option) any later version.
 *
 *   This program is distributed in the hope that it will be useful,
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *   GNU General Public License for more details
 *
 *   You should have received a copy of the GNU Library General Public
 *   License along with this program; if not, write to the
 *   Free Software Foundation, Inc.,
 *   51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 */

#include "runnermanager.h"

#include <KServiceTypeTrader>
#include <KPluginInfo>
#include <KDebug>

#include <Solid/Device>
#include <Solid/DeviceInterface>

#include <threadweaver/DebuggingAids.h>
#include <ThreadWeaver/Thread>
#include <ThreadWeaver/Job>
#include <ThreadWeaver/State>
#include <ThreadWeaver/QueuePolicy>
#include <ThreadWeaver/Weaver>
#include <ThreadWeaver/WeaverObserver>
#include <QMutex>
#include <QTimer>
#include <QCoreApplication>

#include "querymatch.h"

using ThreadWeaver::Weaver;
using ThreadWeaver::Job;


namespace Plasma
{


/*****************************************************
*  RunnerRestrictionPolicy class
* Restricts simultaneous jobs of the same type
* Similar to ResourceRestrictionPolicy but check the object type first
******************************************************/
class RunnerRestrictionPolicy : public ThreadWeaver::QueuePolicy
{
public:
    ~RunnerRestrictionPolicy();

    static RunnerRestrictionPolicy& instance();

    void setCap(int cap)
    {
        m_cap = cap;
    }
    int cap() const
    {
        return m_cap;
    }

    bool canRun(Job* job);
    void free(Job* job);
    void release(Job* job);
    void destructed(Job* job);
private:
    RunnerRestrictionPolicy();

//     QHash<QString, int> m_runCounts;
    int m_count;
    int m_cap;
    QMutex m_mutex;
};

RunnerRestrictionPolicy::RunnerRestrictionPolicy()
    : QueuePolicy(),
      m_cap(2)
{
}

RunnerRestrictionPolicy::~RunnerRestrictionPolicy()
{
}

RunnerRestrictionPolicy& RunnerRestrictionPolicy::instance()
{
    static RunnerRestrictionPolicy policy;
    return policy;
}

bool RunnerRestrictionPolicy::canRun(Job* job)
{
    Q_UNUSED(job)
    QMutexLocker l(&m_mutex);
//     QString type = job->objectName();
    if (m_count/*m_runCounts.value(type)*/ > m_cap) {
//         kDebug() << "Denying job " << type << " because of " << m_count/*m_runCounts[type]*/ << " current jobs" << endl;
        return false;
    } else {
//         m_runCounts[type]++;
        ++m_count;
        return true;
    }
}

void RunnerRestrictionPolicy::free(Job* job)
{
    Q_UNUSED(job)
    QMutexLocker l(&m_mutex);
//     QString type = job->objectName();
    --m_count;
//     if (m_runCounts.value(type)) {
//         m_runCounts[type]--;
//     }
}

void RunnerRestrictionPolicy::release(Job* job)
{
    free(job);
}

void RunnerRestrictionPolicy::destructed(Job* job)
{
    Q_UNUSED(job)
}


/*****************************************************
*  FindMatchesJob class
* Class to run queries in different threads
******************************************************/
class FindMatchesJob : public Job
{
public:
    FindMatchesJob(Plasma::AbstractRunner* runner, Plasma::RunnerContext* context, QObject* parent = 0);

    int priority() const;
    Plasma::AbstractRunner* runner() const;

protected:
    void run();
private:
    Plasma::RunnerContext* m_context;
    Plasma::AbstractRunner* m_runner;
};

FindMatchesJob::FindMatchesJob(Plasma::AbstractRunner* runner,
                               Plasma::RunnerContext* context, QObject* parent)
    : ThreadWeaver::Job(parent),
      m_context(context),
      m_runner(runner)
{
    if (runner->speed() == Plasma::AbstractRunner::SlowSpeed) {
        assignQueuePolicy(&RunnerRestrictionPolicy::instance());
    }
}

void FindMatchesJob::run()
{
//     kDebug() << "Running match for " << m_runner->objectName() << " in Thread " << thread()->id() << endl;
    m_runner->performMatch(*m_context);
}

int FindMatchesJob::priority() const
{
    return m_runner->priority();
}

Plasma::AbstractRunner* FindMatchesJob::runner() const
{
    return m_runner;
}

/*****************************************************
*  RunnerManager::Private class
*
*****************************************************/
class RunnerManager::Private
{
public:

    Private(RunnerManager *parent)
      : q(parent),
        deferredRun(0)
    {
        matchChangeTimer.setSingleShot(true);
        connect(&matchChangeTimer, SIGNAL(timeout()), q, SLOT(scheduleMatchesChanged()));
        connect(&context, SIGNAL(matchesChanged()), q, SLOT(matchesChanged()));
    }

    void scheduleMatchesChanged()
    {
        matchChangeTimer.start(100);
    }

    void matchesChanged()
    {
        emit q->matchesChanged(context.matches());
    }

    void loadConfiguration(KConfigGroup& conf)
    {
        config = conf;

        //The number of threads used scales with the number of processors.     
        const int numProcs = qMax(Solid::Device::listFromType(Solid::DeviceInterface::Processor).count(), 1);
        //This entry allows to define a hard upper limit independent of the number of processors.
        const int maxThreads = config.readEntry("maxThreads",16);
        const int numThreads = qMin(maxThreads, 2 + ((numProcs - 1) * 2));
//        kDebug() << "setting up" << numThreads << "threads for" << numProcs << "processors";
        Weaver::instance()->setMaximumNumberOfThreads(numThreads);

        //Preferred order of execution of runners
        //prioritylist = config.readEntry("priority", QStringList());

        //If set, this list defines which runners won't be used at runtime
        blacklist = config.readEntry("blacklist", QStringList());
    }

    void loadAll()
    {
        AbstractRunner::List firstRunners;
        AbstractRunner::List normalRunners;
        AbstractRunner::List lastRunners;

        KService::List offers = KServiceTypeTrader::self()->query("Plasma/Runner");
        QString error;
        foreach (const KService::Ptr &service, offers) {
            //kDebug() << "Loading runner: " << service->name() << service->storageId();
            KPluginInfo description(service);
            QString runnerName = description.pluginName();
            if (blacklist.contains(runnerName)) {
                kDebug() << "The plugin" << service->name() << "was blackListed and will not load";
                continue;
            }
            QString api = service->property("X-Plasma-API").toString();
            AbstractRunner* runner = 0;

            if (api.isEmpty()) {
                QVariantList args;
                args << service->storageId();
                runner = service->createInstance<AbstractRunner>(q, args, &error);
            } else {
                //kDebug() << "got a script runner known as" << api;
                runner = new AbstractRunner(q, service->storageId());
            }

            if (runner) {
                //kDebug() << "loading runner: " << service->name();
                QString phase = service->property("X-Plasma-RunnerPhase").toString();
                if (phase == "last") {
                    lastRunners.append(runner);
                } else if (phase == "first") {
                    firstRunners.append(runner);
                } else {
                    normalRunners.append(runner);
                }
            } else {
                kDebug() << "failed to load runner : " << service->name() << ". error reported: " << error;
            }
        }

        firstRunners << normalRunners << lastRunners;
        runners.clear();
        runners = firstRunners;
        //kDebug() << "All runners loaded, total:" << runners.count();
    }

    void jobDone(ThreadWeaver::Job* job)
    {
        FindMatchesJob *runJob = static_cast<FindMatchesJob*>(job);
        if (deferredRun.isEnabled() && runJob->runner() == deferredRun.runner()) {
            //kDebug() << "job actually done, running now **************";
            deferredRun.run(context);
            deferredRun = QueryMatch(0);
        }
        searchJobs.removeAll(runJob);
        delete runJob;
    }

    RunnerManager *q;
    QueryMatch deferredRun;
    RunnerContext context;
    QTimer matchChangeTimer;
    AbstractRunner::List runners;
    QList<FindMatchesJob*> searchJobs;
    QStringList prioritylist;
    QStringList blacklist;
    KConfigGroup config;
};



/*****************************************************
*  RunnerManager::Public class
*
*****************************************************/
RunnerManager::RunnerManager(QObject *parent)
    : QObject(parent), 
      d(new Private(this))
{
    KConfigGroup config(KGlobal::config(), "PlasmaRunnerManager");
    d->loadConfiguration(config);
    //ThreadWeaver::setDebugLevel(true, 4);
}


RunnerManager::RunnerManager(KConfigGroup& config, QObject *parent)
    : QObject(parent), 
      d(new Private(this))
{
    d->loadConfiguration(config);
    //ThreadWeaver::setDebugLevel(true, 4);
}

RunnerManager::~RunnerManager()
{
    delete d;
}

AbstractRunner* RunnerManager::runner(const QString &name) const
{
    if (d->runners.isEmpty()) {
        d->loadAll();
    }

   //TODO: using a list of runners, if this gets slow, switch to hash
    foreach (Plasma::AbstractRunner* runner, d->runners) {
        //kDebug() << "comparing" << name << "with" << runner->id() << runner->name();
        if (name == runner->id()) {
            return runner;
        }
    }
    return 0;
}


RunnerContext* RunnerManager::searchContext() const
{
    return &d->context;
}

//Reordering is here so data is not reordered till strictly needed
QList<QueryMatch> RunnerManager::matches() const
{
    return d->context.matches();
}

void RunnerManager::run(const QString &id)
{
    run(d->context.match(id));
}

void RunnerManager::run(const QueryMatch &match)
{
    if (!match.isEnabled()) {
        return;
    }

    //TODO: this function is not const as it may be used for learning
    AbstractRunner *runner = match.runner();

    foreach (FindMatchesJob *job, d->searchJobs) {
        if (job->runner() == runner && !job->isFinished()) {
            //kDebug() << "!!!!!!!!!!!!!!!!!!! uh oh!";
            d->deferredRun = match;
            return;
        }
    }

    match.run(d->context);

    if (d->deferredRun.isValid()) {
        d->deferredRun = QueryMatch(0);
    }
}

void RunnerManager::launchQuery(const QString &term)
{
    launchQuery(term, QString());
}

void RunnerManager::launchQuery(const QString &term, const QString &runnerName)
{
    if (d->runners.isEmpty()) {
        d->loadAll();
    }

    if (term.isEmpty()) {
        reset();
        return;
    }

    if (d->context.query() == term) {
        // we already are searching for this!
        return;
    }

    reset();
//    kDebug() << "runners searching for" << term << "on" << runnerName;
    d->context.setQuery(term);

    AbstractRunner::List runable;

    //if the name is not empty we will launch only the specified runner
    if (!runnerName.isEmpty()) {
        runable.append(runner(runnerName));
    } else {
        runable = d->runners; 
    }

    foreach (Plasma::AbstractRunner* r, runable) {
        if ((r->ignoredTypes() & d->context.type()) == 0) {
//            kDebug() << "launching" << r->name();
            FindMatchesJob *job = new FindMatchesJob(r, &d->context, this);
            connect(job, SIGNAL(done(ThreadWeaver::Job*)), this, SLOT(jobDone(ThreadWeaver::Job*)));
            Weaver::instance()->enqueue(job);
            d->searchJobs.append(job);
        }
    }
}

bool RunnerManager::execQuery(const QString &term)
{
    return execQuery(term, QString());
}

bool RunnerManager::execQuery(const QString &term, const QString &runnerName)
{
    if (d->runners.isEmpty()) {
        d->loadAll();
    }

    if (term.isEmpty()) {
        reset();
        return false;
    }

    if (d->context.query() == term) {
        // we already are searching for this!
        emit matchesChanged(d->context.matches());
        return false;
    }

    reset();
    //kDebug() << "executing query about " << term << "on" << runnerName;
    d->context.setQuery(term);
    AbstractRunner *r = runner(runnerName);

    if (!r) {
        //kDebug() << "failed to find the runner";
        return false;
    }

    if ((r->ignoredTypes() & d->context.type()) != 0) {
        //kDebug() << "ignored!";
        return false;
    }

    r->performMatch(d->context);
    //kDebug() << "succeeded with" << d->context.matches().count() << "results";
    emit matchesChanged(d->context.matches());
    return true;
}

QString RunnerManager::query() const
{
    return d->context.query();
}

void RunnerManager::reset()
{
    // If ThreadWeaver is idle, it is safe to clear previous jobs 
    if (Weaver::instance()->isIdle()) {
        qDeleteAll(d->searchJobs);
        d->searchJobs.clear();
    } else {
        Weaver::instance()->dequeue();
    }

    if (d->deferredRun.isEnabled()) {
        //kDebug() << "job actually done, running now **************";
        d->deferredRun.run(d->context);
        d->deferredRun = QueryMatch(0);
    }

    d->context.reset();

   //FIXME: if the weaver is not idle we have a number of jobs that 
   //      will eventually return their matches
   //      we need a way to stop them (Weaver::requestAbort() Job::requestAbort ) 
   //      or discard results 
}

} // Plasma namespace

#include "runnermanager.moc"