mariadb-columnstore-engine/utils/threadpool/prioritythreadpool.cpp

/* Copyright (C) 2014 InfiniDB, Inc.

   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; version 2 of
   the License.

   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. */

/***********************************************************************
*   $Id: threadpool.cpp 553 2008-02-27 17:51:16Z rdempsey $
*
*
***********************************************************************/

#include <stdexcept>
#include <unistd.h>
#include <exception>
using namespace std;

#include "messageobj.h"
#include "messagelog.h"
using namespace logging;

#include "prioritythreadpool.h"
using namespace boost;

#include "dbcon/joblist/primitivemsg.h"

namespace threadpool
{

PriorityThreadPool::PriorityThreadPool(uint targetWeightPerRun, uint highThreads,
                                       uint midThreads, uint lowThreads, uint ID) :
    _stop(false), weightPerRun(targetWeightPerRun), id(ID)
{
    boost::thread* newThread;
    for (uint32_t i = 0; i < highThreads; i++)
    {
		newThread = threads.create_thread(ThreadHelper(this, HIGH));
        newThread->detach();
    }
    for (uint32_t i = 0; i < midThreads; i++)
    {
		newThread = threads.create_thread(ThreadHelper(this, MEDIUM));
        newThread->detach();
    }
    for (uint32_t i = 0; i < lowThreads; i++)
    {
		newThread = threads.create_thread(ThreadHelper(this, LOW));
        newThread->detach();
    }
    cout << "started " << highThreads << " high, " << midThreads << " med, " << lowThreads
         << " low.\n";
    defaultThreadCounts[HIGH] = threadCounts[HIGH] = highThreads;
    defaultThreadCounts[MEDIUM] = threadCounts[MEDIUM] = midThreads;
    defaultThreadCounts[LOW] = threadCounts[LOW] = lowThreads;
}

PriorityThreadPool::~PriorityThreadPool()
{
    stop();
}

void PriorityThreadPool::addJob(const Job& job, bool useLock)
{
    boost::thread* newThread;
    mutex::scoped_lock lk(mutex, defer_lock_t());

    if (useLock)
        lk.lock();

    // Create any missing threads
    if (defaultThreadCounts[HIGH] != threadCounts[HIGH])
    {
        newThread = threads.create_thread(ThreadHelper(this, HIGH));
        newThread->detach();
        threadCounts[HIGH]++;
    }

    if (defaultThreadCounts[MEDIUM] != threadCounts[MEDIUM])
    {
        newThread = threads.create_thread(ThreadHelper(this, MEDIUM));
        newThread->detach();
        threadCounts[MEDIUM]++;
    }

    if (defaultThreadCounts[LOW] != threadCounts[LOW])
    {
        newThread = threads.create_thread(ThreadHelper(this, LOW));
        newThread->detach();
        threadCounts[LOW]++;
    }

    if (job.priority > 66)
        jobQueues[HIGH].push_back(job);
    else if (job.priority > 33)
        jobQueues[MEDIUM].push_back(job);
    else
        jobQueues[LOW].push_back(job);

    if (useLock)
        newJob.notify_one();
}

void PriorityThreadPool::removeJobs(uint32_t id)
{
    list<Job>::iterator it;

    mutex::scoped_lock lk(mutex);

    for (uint32_t i = 0; i < _COUNT; i++)
        for (it = jobQueues[i].begin(); it != jobQueues[i].end();)
            if (it->id == id)
                it = jobQueues[i].erase(it);
            else
                ++it;
}

PriorityThreadPool::Priority PriorityThreadPool::pickAQueue(Priority preference)
{
    if (!jobQueues[preference].empty())
        return preference;
    else if (!jobQueues[HIGH].empty())
        return HIGH;
    else if (!jobQueues[MEDIUM].empty())
        return MEDIUM;
    else
        return LOW;
}

void PriorityThreadPool::threadFcn(const Priority preferredQueue) throw()
{
    Priority queue;
    uint32_t weight, i;
    vector<Job> runList;
    vector<bool> reschedule;
    uint32_t rescheduleCount;
    uint32_t queueSize;
    bool running = false;

    try
    {
        while (!_stop)
        {

            mutex::scoped_lock lk(mutex);

            queue = pickAQueue(preferredQueue);

            if (jobQueues[queue].empty())
            {
                newJob.wait(lk);
                continue;
            }

            queueSize = jobQueues[queue].size();
            weight = 0;
            // 3 conditions stop this thread from grabbing all jobs in the queue
            //
            // 1: The weight limit has been exceeded
            // 2: The queue is empty
            // 3: It has grabbed more than half of the jobs available &
            //     should leave some to the other threads

            while ((weight < weightPerRun) && (!jobQueues[queue].empty())
                    && (runList.size() <= queueSize / 2))
            {
                runList.push_back(jobQueues[queue].front());
                jobQueues[queue].pop_front();
                weight += runList.back().weight;
            }

            lk.unlock();

            reschedule.resize(runList.size());
            rescheduleCount = 0;

            for (i = 0; i < runList.size() && !_stop; i++)
            {
                reschedule[i] = false;
                running = true;
                reschedule[i] = (*(runList[i].functor))();
                running = false;

                if (reschedule[i])
                    rescheduleCount++;
            }

            // no real work was done, prevent intensive busy waiting
            if (rescheduleCount == runList.size())
                usleep(1000);

            if (rescheduleCount > 0)
            {
                lk.lock();

                for (i = 0; i < runList.size(); i++)
                    if (reschedule[i])
                        addJob(runList[i], false);

                if (rescheduleCount > 1)
                    newJob.notify_all();
                else
                    newJob.notify_one();

                lk.unlock();
            }

            runList.clear();
        }
    }
    catch (std::exception& ex)
    {
        // Log the exception and exit this thread
        try
        {
            threadCounts[queue]--;
#ifndef NOLOGGING
            logging::Message::Args args;
            logging::Message message(5);
            args.add("threadFcn: Caught exception: ");
            args.add(ex.what());

            message.format( args );

            logging::LoggingID lid(22);
            logging::MessageLog ml(lid);

            ml.logErrorMessage( message );
#endif

            if (running)
                sendErrorMsg(runList[i].uniqueID, runList[i].stepID, runList[i].sock);
        }
        catch (...)
        {
        }
    }
    catch (...)
    {

        // Log the exception and exit this thread
        try
        {
            threadCounts[queue]--;
#ifndef NOLOGGING
            logging::Message::Args args;
            logging::Message message(6);
            args.add("threadFcn: Caught unknown exception!");

            message.format( args );

            logging::LoggingID lid(22);
            logging::MessageLog ml(lid);

            ml.logErrorMessage( message );
#endif

            if (running)
                sendErrorMsg(runList[i].uniqueID, runList[i].stepID, runList[i].sock);
        }
        catch (...)
        {
        }
    }
}

void PriorityThreadPool::sendErrorMsg(uint32_t id, uint32_t step, primitiveprocessor::SP_UM_IOSOCK sock)
{
    ISMPacketHeader ism;
    PrimitiveHeader ph = {0};

    ism.Status =  logging::primitiveServerErr;
    ph.UniqueID = id;
    ph.StepID = step;
    ByteStream msg(sizeof(ISMPacketHeader) + sizeof(PrimitiveHeader));
    msg.append((uint8_t*) &ism, sizeof(ism));
    msg.append((uint8_t*) &ph, sizeof(ph));

    sock->write(msg);
}

void PriorityThreadPool::stop()
{
    _stop = true;
}

} // namespace threadpool
// vim:ts=4 sw=4: