mthreadpool.cpp

Go to the documentation of this file.

/*  -*- Mode: c++ -*-
 *
 *   Class MThreadPool
 *
 *   Copyright (C) Daniel Kristjansson 2011
 *
 *   This program is free software; you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation; either version 2 of the License, 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 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
 */

// C++ headers
#include <algorithm>
using namespace std;

// Qt headers
#include <QCoreApplication>
#include <QList>
#include <QMap>
#include <QMutex>
#include <QMutexLocker>
#include <QPair>
#include <QRunnable>
#include <QSet>
#include <QWaitCondition>
#include <utility>

// MythTV headers
#include "mthreadpool.h"
#include "mythlogging.h"
#include "mythtimer.h"
#include "logging.h"
#include "mthread.h"
#include "mythdb.h"

using MPoolEntry = QPair<QRunnable*,QString>;
using MPoolQueue = QList<MPoolEntry>;
using MPoolQueues = QMap<int, MPoolQueue>;

class MPoolThread : public MThread
{
  public:
    MPoolThread(MThreadPool &pool, int timeout) :
        MThread("PT"), m_pool(pool), m_expiryTimeout(timeout)
    {
        QMutexLocker locker(&s_lock);
        setObjectName(QString("PT%1").arg(s_thread_num));
        s_thread_num++;
    }

    void run(void) override // MThread
    {
        RunProlog();

        MythTimer t;
        t.start();
        QMutexLocker locker(&m_lock);
        while (true)
        {
            if (m_doRun && !m_runnable)
                m_wait.wait(locker.mutex(), m_expiryTimeout+1);

            if (!m_runnable)
            {
                m_doRun = false;

                locker.unlock();
                m_pool.NotifyDone(this);
                locker.relock();
                break;
            }

            if (!m_runnableName.isEmpty())
                loggingRegisterThread(m_runnableName);

            bool autodelete = m_runnable->autoDelete();
            m_runnable->run();
            if (autodelete)
                delete m_runnable;
            if (m_reserved)
                m_pool.ReleaseThread();
            m_reserved = false;
            m_runnable = nullptr;

            loggingDeregisterThread();
            loggingRegisterThread(objectName());

            GetMythDB()->GetDBManager()->PurgeIdleConnections(false);
            qApp->processEvents();
            qApp->sendPostedEvents(nullptr, QEvent::DeferredDelete);

            t.start();

            if (m_doRun)
            {
                locker.unlock();
                m_pool.NotifyAvailable(this);
                locker.relock();
            }
            else
            {
                locker.unlock();
                m_pool.NotifyDone(this);
                locker.relock();
                break;
            }
        }

        RunEpilog();
    }

    bool SetRunnable(QRunnable *runnable, QString runnableName,
                     bool reserved)
    {
        QMutexLocker locker(&m_lock);
        if (m_doRun && (m_runnable == nullptr))
        {
            m_runnable = runnable;
            m_runnableName = std::move(runnableName);
            m_reserved = reserved;
            m_wait.wakeAll();
            return true;
        }
        return false;
    }

    void Shutdown(void)
    {
        QMutexLocker locker(&m_lock);
        m_doRun = false;
        m_wait.wakeAll();
    }

    QMutex          m_lock;
    QWaitCondition  m_wait;
    MThreadPool    &m_pool;
    int             m_expiryTimeout;
    bool            m_doRun          {true};
    QString         m_runnableName;
    bool            m_reserved       {false};

    static QMutex s_lock;
    static uint s_thread_num;
};
QMutex MPoolThread::s_lock;
uint MPoolThread::s_thread_num = 0;


class MThreadPoolPrivate
{
  public:
    explicit MThreadPoolPrivate(QString name) :
        m_name(std::move(name)) {}

    int GetRealMaxThread(void)
    {
        return max(m_maxThreadCount,1) + m_reserveThread;
    }

    mutable QMutex m_lock;
    QString m_name;
    QWaitCondition m_wait;
    bool m_running          {true};
    int  m_expiryTimeout    {120 * 1000};
    int  m_maxThreadCount   {QThread::idealThreadCount()};
    int  m_reserveThread    {0};

    MPoolQueues         m_runQueues;
    QSet<MPoolThread*>  m_availThreads;
    QSet<MPoolThread*>  m_runningThreads;
    QList<MPoolThread*> m_deleteThreads;

    static QMutex s_pool_lock;
    static MThreadPool *s_pool;
    static QList<MThreadPool*> s_all_pools;
};

QMutex MThreadPoolPrivate::s_pool_lock(QMutex::Recursive);
MThreadPool *MThreadPoolPrivate::s_pool = nullptr;
QList<MThreadPool*> MThreadPoolPrivate::s_all_pools;


MThreadPool::MThreadPool(const QString &name) :
    m_priv(new MThreadPoolPrivate(name))
{
    QMutexLocker locker(&MThreadPoolPrivate::s_pool_lock);
    MThreadPoolPrivate::s_all_pools.push_back(this);
}

MThreadPool::~MThreadPool()
{
    Stop();
    DeletePoolThreads();
    {
        QMutexLocker locker(&MThreadPoolPrivate::s_pool_lock);
        MThreadPoolPrivate::s_all_pools.removeAll(this);
    }
    delete m_priv;
    m_priv = nullptr;
}

void MThreadPool::Stop(void)
{
    QMutexLocker locker(&m_priv->m_lock);
    m_priv->m_running = false;
    QSet<MPoolThread*>::iterator it = m_priv->m_availThreads.begin();
    for (; it != m_priv->m_availThreads.end(); ++it)
        (*it)->Shutdown();
    it = m_priv->m_runningThreads.begin();
    for (; it != m_priv->m_runningThreads.end(); ++it)
        (*it)->Shutdown();
    m_priv->m_wait.wakeAll();
}

void MThreadPool::DeletePoolThreads(void)
{
    waitForDone();

    QMutexLocker locker(&m_priv->m_lock);
    foreach (auto thread, m_priv->m_availThreads)
    {
        m_priv->m_deleteThreads.push_front(thread);
    }
    m_priv->m_availThreads.clear();

    while (!m_priv->m_deleteThreads.empty())
    {
        MPoolThread *thread = m_priv->m_deleteThreads.back();
        locker.unlock();

        thread->wait();

        locker.relock();
        delete thread;
        if (m_priv->m_deleteThreads.back() == thread)
            m_priv->m_deleteThreads.pop_back();
        else
            m_priv->m_deleteThreads.removeAll(thread);
    }
}

MThreadPool *MThreadPool::globalInstance(void)
{
    QMutexLocker locker(&MThreadPoolPrivate::s_pool_lock);
    if (!MThreadPoolPrivate::s_pool)
        MThreadPoolPrivate::s_pool = new MThreadPool("GlobalPool");
    return MThreadPoolPrivate::s_pool;
}

void MThreadPool::StopAllPools(void)
{
    QMutexLocker locker(&MThreadPoolPrivate::s_pool_lock);
    QList<MThreadPool*>::iterator it;
    for (it = MThreadPoolPrivate::s_all_pools.begin();
         it != MThreadPoolPrivate::s_all_pools.end(); ++it)
    {
        (*it)->Stop();
    }
}

void MThreadPool::ShutdownAllPools(void)
{
    QMutexLocker locker(&MThreadPoolPrivate::s_pool_lock);
    QList<MThreadPool*>::iterator it;
    for (it = MThreadPoolPrivate::s_all_pools.begin();
         it != MThreadPoolPrivate::s_all_pools.end(); ++it)
    {
        (*it)->Stop();
    }
    for (it = MThreadPoolPrivate::s_all_pools.begin();
         it != MThreadPoolPrivate::s_all_pools.end(); ++it)
    {
        (*it)->DeletePoolThreads();
    }
}

void MThreadPool::start(QRunnable *runnable, const QString& debugName, int priority)
{
    QMutexLocker locker(&m_priv->m_lock);
    if (TryStartInternal(runnable, debugName, false))
        return;

    MPoolQueues::iterator it = m_priv->m_runQueues.find(priority);
    if (it != m_priv->m_runQueues.end())
    {
        (*it).push_back(MPoolEntry(runnable,debugName));
    }
    else
    {
        MPoolQueue list;
        list.push_back(MPoolEntry(runnable,debugName));
        m_priv->m_runQueues[priority] = list;
    }
}

void MThreadPool::startReserved(
    QRunnable *runnable, const QString& debugName, int waitForAvailMS)
{
    QMutexLocker locker(&m_priv->m_lock);
    if (waitForAvailMS > 0 && m_priv->m_availThreads.empty() &&
        m_priv->m_runningThreads.size() >= m_priv->m_maxThreadCount)
    {
        MythTimer t;
        t.start();
        int left = waitForAvailMS - t.elapsed();
        while (left > 0 && m_priv->m_availThreads.empty() &&
               m_priv->m_runningThreads.size() >= m_priv->m_maxThreadCount)
        {
            m_priv->m_wait.wait(locker.mutex(), left);
            left = waitForAvailMS - t.elapsed();
        }
    }
    TryStartInternal(runnable, debugName, true);
}


bool MThreadPool::tryStart(QRunnable *runnable, const QString& debugName)
{
    QMutexLocker locker(&m_priv->m_lock);
    return TryStartInternal(runnable, debugName, false);
}

bool MThreadPool::TryStartInternal(
    QRunnable *runnable, const QString& debugName, bool reserved)
{
    if (!m_priv->m_running)
        return false;

    while (!m_priv->m_deleteThreads.empty())
    {
        m_priv->m_deleteThreads.back()->wait();
        delete m_priv->m_deleteThreads.back();
        m_priv->m_deleteThreads.pop_back();
    }

    while (m_priv->m_availThreads.begin() != m_priv->m_availThreads.end())
    {
        MPoolThread *thread = *m_priv->m_availThreads.begin();
        m_priv->m_availThreads.erase(m_priv->m_availThreads.begin());
        m_priv->m_runningThreads.insert(thread);
        if (reserved)
            m_priv->m_reserveThread++;
        if (thread->SetRunnable(runnable, debugName, reserved))
        {
            return true;
        }

        if (reserved)
            m_priv->m_reserveThread--;
        thread->Shutdown();
        m_priv->m_runningThreads.remove(thread);
        m_priv->m_deleteThreads.push_front(thread);
    }

    if (reserved ||
        m_priv->m_runningThreads.size() < m_priv->GetRealMaxThread())
    {
        if (reserved)
            m_priv->m_reserveThread++;
        auto *thread = new MPoolThread(*this, m_priv->m_expiryTimeout);
        m_priv->m_runningThreads.insert(thread);
        thread->SetRunnable(runnable, debugName, reserved);
        thread->start();
        if (thread->isRunning())
        {
            return true;
        }

        // Thread failed to run, OOM?
        // QThread will print an error, so we don't have to
        if (reserved)
            m_priv->m_reserveThread--;
        thread->Shutdown();
        m_priv->m_runningThreads.remove(thread);
        m_priv->m_deleteThreads.push_front(thread);
    }

    return false;
}

void MThreadPool::NotifyAvailable(MPoolThread *thread)
{
    QMutexLocker locker(&m_priv->m_lock);

    if (!m_priv->m_running)
    {
        m_priv->m_runningThreads.remove(thread);
        thread->Shutdown();
        m_priv->m_deleteThreads.push_front(thread);
        m_priv->m_wait.wakeAll();
        return;
    }

    MPoolQueues::iterator it = m_priv->m_runQueues.begin();
    if (it == m_priv->m_runQueues.end())
    {
        m_priv->m_runningThreads.remove(thread);
        m_priv->m_availThreads.insert(thread);
        m_priv->m_wait.wakeAll();
        return;
    }

    MPoolEntry e = (*it).front();
    if (!thread->SetRunnable(e.first, e.second, false))
    {
        m_priv->m_runningThreads.remove(thread);
        m_priv->m_wait.wakeAll();
        if (!TryStartInternal(e.first, e.second, false))
        {
            thread->Shutdown();
            m_priv->m_deleteThreads.push_front(thread);
            return;
        }
        thread->Shutdown();
        m_priv->m_deleteThreads.push_front(thread);
    }

    (*it).pop_front();
    if ((*it).empty())
        m_priv->m_runQueues.erase(it);
}

void MThreadPool::NotifyDone(MPoolThread *thread)
{
    QMutexLocker locker(&m_priv->m_lock);
    m_priv->m_runningThreads.remove(thread);
    m_priv->m_availThreads.remove(thread);
    if (!m_priv->m_deleteThreads.contains(thread))
        m_priv->m_deleteThreads.push_front(thread);
    m_priv->m_wait.wakeAll();
}

int MThreadPool::expiryTimeout(void) const
{
    QMutexLocker locker(&m_priv->m_lock);
    return m_priv->m_expiryTimeout;
}

void MThreadPool::setExpiryTimeout(int expiryTimeout)
{
    QMutexLocker locker(&m_priv->m_lock);
    m_priv->m_expiryTimeout = expiryTimeout;
}

int MThreadPool::maxThreadCount(void) const
{
    QMutexLocker locker(&m_priv->m_lock);
    return m_priv->m_maxThreadCount;
}

void MThreadPool::setMaxThreadCount(int maxThreadCount)
{
    QMutexLocker locker(&m_priv->m_lock);
    m_priv->m_maxThreadCount = maxThreadCount;
}

int MThreadPool::activeThreadCount(void) const
{
    QMutexLocker locker(&m_priv->m_lock);
    return m_priv->m_availThreads.size() + m_priv->m_runningThreads.size();
}

/*
void MThreadPool::reserveThread(void)
{
    QMutexLocker locker(&m_priv->m_lock);
    m_priv->m_reserveThread++;
}

void MThreadPool::releaseThread(void)
{
    QMutexLocker locker(&m_priv->m_lock);
    if (m_priv->m_reserveThread > 0)
        m_priv->m_reserveThread--;
}
*/

void MThreadPool::ReleaseThread(void)
{
    QMutexLocker locker(&m_priv->m_lock);
    if (m_priv->m_reserveThread > 0)
        m_priv->m_reserveThread--;
}

#if 0
static void print_set(QString title, QSet<MPoolThread*> set)
{
    LOG(VB_GENERAL, LOG_INFO, title);
    foreach (auto item, set)
    {
        LOG(VB_GENERAL, LOG_INFO, QString(" : 0x%1")
            .arg((quint64)item,0,16));
    }
    LOG(VB_GENERAL, LOG_INFO, "");
}
#endif

void MThreadPool::waitForDone(void)
{
    QMutexLocker locker(&m_priv->m_lock);
    while (true)
    {
        while (!m_priv->m_deleteThreads.empty())
        {
            m_priv->m_deleteThreads.back()->wait();
            delete m_priv->m_deleteThreads.back();
            m_priv->m_deleteThreads.pop_back();
        }

        if (m_priv->m_running && !m_priv->m_runQueues.empty())
        {
            m_priv->m_wait.wait(locker.mutex());
            continue;
        }

        QSet<MPoolThread*> working = m_priv->m_runningThreads;
        working = working.subtract(m_priv->m_availThreads);
        if (working.empty())
            break;
        m_priv->m_wait.wait(locker.mutex());
    }
}

/* vim: set expandtab tabstop=4 shiftwidth=4: */