<commit_before>/* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
/*
 * This file is part of the LibreOffice project.
 *
 * This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/.
 */

#include <comphelper/threadpool.hxx>

#include <rtl/instance.hxx>
#include <boost/shared_ptr.hpp>
#include <thread>
#include <algorithm>

namespace comphelper {

class ThreadPool::ThreadWorker : public salhelper::Thread
{
    ThreadPool    *mpPool;
    osl::Condition maNewWork;
    bool           mbWorking;
public:

    ThreadWorker( ThreadPool *pPool ) :
        salhelper::Thread("thread-pool"),
        mpPool( pPool ),
        mbWorking( false )
    {
    }

    virtual void execute() SAL_OVERRIDE
    {
        ThreadTask *pTask;
        while ( ( pTask = waitForWork() ) )
        {
            pTask->doWork();
            delete pTask;
        }
    }

    ThreadTask *waitForWork()
    {
        ThreadTask *pRet = NULL;

        osl::ResettableMutexGuard aGuard( mpPool->maGuard );

        pRet = mpPool->popWork();

        while( !pRet )
        {
            if (mbWorking)
                mpPool->stopWork();
            mbWorking = false;
            maNewWork.reset();

            if( mpPool->mbTerminate )
                break;

            aGuard.clear(); // unlock

            maNewWork.wait();

            aGuard.reset(); // lock

            pRet = mpPool->popWork();
        }

        if (pRet)
        {
            if (!mbWorking)
                mpPool->startWork();
            mbWorking = true;
        }

        return pRet;
    }

    // Why a condition per worker thread - you may ask.
    //
    // Unfortunately the Windows synchronisation API that we wrap
    // is horribly inadequate cf.
    //    http://www.cs.wustl.edu/~schmidt/win32-cv-1.html
    // The existing osl::Condition API should only ever be used
    // between one producer and one consumer thread to avoid the
    // lost wakeup problem.

    void signalNewWork()
    {
        maNewWork.set();
    }
};

ThreadPool::ThreadPool( sal_Int32 nWorkers ) :
    mnThreadsWorking( 0 ),
    mbTerminate( false )
{
    for( sal_Int32 i = 0; i < nWorkers; i++ )
        maWorkers.push_back( new ThreadWorker( this ) );

    maTasksComplete.reset();

    osl::MutexGuard aGuard( maGuard );
    for( size_t i = 0; i < maWorkers.size(); i++ )
        maWorkers[ i ]->launch();
}

ThreadPool::~ThreadPool()
{
    waitAndCleanupWorkers();
}

struct ThreadPoolStatic : public rtl::StaticWithInit< boost::shared_ptr< ThreadPool >,
                                                      ThreadPoolStatic >
{
    boost::shared_ptr< ThreadPool > operator () () {
        sal_Int32 nThreads = std::max( std::thread::hardware_concurrency(), 1U );
        return boost::shared_ptr< ThreadPool >( new ThreadPool( nThreads ) );
    };
};

ThreadPool& ThreadPool::getSharedOptimalPool()
{
    return *ThreadPoolStatic::get().get();
}

void ThreadPool::waitAndCleanupWorkers()
{
    waitUntilEmpty();

    osl::ResettableMutexGuard aGuard( maGuard );
    mbTerminate = true;<commit_msg>thread-pool: Initialize empty pools to start complete.<commit_after>/* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
/*
 * This file is part of the LibreOffice project.
 *
 * This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/.
 */

#include <comphelper/threadpool.hxx>

#include <rtl/instance.hxx>
#include <boost/shared_ptr.hpp>
#include <thread>
#include <algorithm>

namespace comphelper {

class ThreadPool::ThreadWorker : public salhelper::Thread
{
    ThreadPool    *mpPool;
    osl::Condition maNewWork;
    bool           mbWorking;
public:

    ThreadWorker( ThreadPool *pPool ) :
        salhelper::Thread("thread-pool"),
        mpPool( pPool ),
        mbWorking( false )
    {
    }

    virtual void execute() SAL_OVERRIDE
    {
        ThreadTask *pTask;
        while ( ( pTask = waitForWork() ) )
        {
            pTask->doWork();
            delete pTask;
        }
    }

    ThreadTask *waitForWork()
    {
        ThreadTask *pRet = NULL;

        osl::ResettableMutexGuard aGuard( mpPool->maGuard );

        pRet = mpPool->popWork();

        while( !pRet )
        {
            if (mbWorking)
                mpPool->stopWork();
            mbWorking = false;
            maNewWork.reset();

            if( mpPool->mbTerminate )
                break;

            aGuard.clear(); // unlock

            maNewWork.wait();

            aGuard.reset(); // lock

            pRet = mpPool->popWork();
        }

        if (pRet)
        {
            if (!mbWorking)
                mpPool->startWork();
            mbWorking = true;
        }

        return pRet;
    }

    // Why a condition per worker thread - you may ask.
    //
    // Unfortunately the Windows synchronisation API that we wrap
    // is horribly inadequate cf.
    //    http://www.cs.wustl.edu/~schmidt/win32-cv-1.html
    // The existing osl::Condition API should only ever be used
    // between one producer and one consumer thread to avoid the
    // lost wakeup problem.

    void signalNewWork()
    {
        maNewWork.set();
    }
};

ThreadPool::ThreadPool( sal_Int32 nWorkers ) :
    mnThreadsWorking( 0 ),
    mbTerminate( false )
{
    for( sal_Int32 i = 0; i < nWorkers; i++ )
        maWorkers.push_back( new ThreadWorker( this ) );

    maTasksComplete.set();

    osl::MutexGuard aGuard( maGuard );
    for( size_t i = 0; i < maWorkers.size(); i++ )
        maWorkers[ i ]->launch();
}

ThreadPool::~ThreadPool()
{
    waitAndCleanupWorkers();
}

struct ThreadPoolStatic : public rtl::StaticWithInit< boost::shared_ptr< ThreadPool >,
                                                      ThreadPoolStatic >
{
    boost::shared_ptr< ThreadPool > operator () () {
        sal_Int32 nThreads = std::max( std::thread::hardware_concurrency(), 1U );
        return boost::shared_ptr< ThreadPool >( new ThreadPool( nThreads ) );
    };
};

ThreadPool& ThreadPool::getSharedOptimalPool()
{
    return *ThreadPoolStatic::get().get();
}

void ThreadPool::waitAndCleanupWorkers()
{
    waitUntilEmpty();

    osl::ResettableMutexGuard aGuard( maGuard );
    mbTerminate = true;
{'fraction_non_alphanumeric': 0.1209366815590818, 'fraction_numerical': 0.005854259744261285, 'mean_word_length': 2.1226551226551225, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 37, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4922544', 'n_tokens_mistral': 2031, 'n_tokens_neox': 1934, 'n_words': 495}