13872211292

Committed 15 Mar 2025 11:00AM UTC coverage: 70.445% (+0.009%) from 70.436%

Build # 13872211292

Build Type

Pull #11951

github

Committed by

web-flow

Commit Message

Merge 643845942 into bb4e0ed67

Pull Request Pull Request #11951: Doc: Build docs using CMake

Run Details

553795 of 786140 relevant lines covered (70.44%)

221892.63 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.0

/port/cpl_worker_thread_pool.cpp

/**********************************************************************
 *
 * Project:  CPL - Common Portability Library
 * Purpose:  CPL worker thread pool
 * Author:   Even Rouault, <even dot rouault at spatialys dot com>
 *
 **********************************************************************
 * Copyright (c) 2015, Even Rouault, <even dot rouault at spatialys dot com>
 *
 * SPDX-License-Identifier: MIT
 ****************************************************************************/

#include "cpl_port.h"
#include "cpl_worker_thread_pool.h"

#include <cstddef>
#include <memory>

#include "cpl_conv.h"
#include "cpl_error.h"
#include "cpl_vsi.h"

static thread_local CPLWorkerThreadPool *threadLocalCurrentThreadPool = nullptr;

/************************************************************************/
/*                         CPLWorkerThreadPool()                        */
/************************************************************************/

/** Instantiate a new pool of worker threads.
 *
 * The pool is in an uninitialized state after this call. The Setup() method
 * must be called.
 */
CPLWorkerThreadPool::CPLWorkerThreadPool() : jobQueue{}
{
}

/** Instantiate a new pool of worker threads.
 *
 * \param nThreads  Number of threads in the pool.
 */
CPLWorkerThreadPool::CPLWorkerThreadPool(int nThreads) : jobQueue{}
{
    Setup(nThreads, nullptr, nullptr);
}

/************************************************************************/
/*                          ~CPLWorkerThreadPool()                      */
/************************************************************************/

/** Destroys a pool of worker threads.
 *
 * Any still pending job will be completed before the destructor returns.
 */
CPLWorkerThreadPool::~CPLWorkerThreadPool()
{
    WaitCompletion();

    {
        std::lock_guard<std::mutex> oGuard(m_mutex);
        eState = CPLWTS_STOP;
    }

    for (auto &wt : aWT)
    {
        {
            std::lock_guard<std::mutex> oGuard(wt->m_mutex);
            wt->m_cv.notify_one();
        }
        CPLJoinThread(wt->hThread);
    }

    CPLListDestroy(psWaitingWorkerThreadsList);
}

/************************************************************************/
/*                        GetThreadCount()                              */
/************************************************************************/

int CPLWorkerThreadPool::GetThreadCount() const
{
    std::unique_lock<std::mutex> oGuard(m_mutex);
    return m_nMaxThreads;
}

/************************************************************************/
/*                       WorkerThreadFunction()                         */
/************************************************************************/

void CPLWorkerThreadPool::WorkerThreadFunction(void *user_data)
{
    CPLWorkerThread *psWT = static_cast<CPLWorkerThread *>(user_data);
    CPLWorkerThreadPool *poTP = psWT->poTP;

    threadLocalCurrentThreadPool = poTP;

    if (psWT->pfnInitFunc)
        psWT->pfnInitFunc(psWT->pInitData);

    while (true)
    {
        std::function<void()> task = poTP->GetNextJob(psWT);
        if (!task)
            break;

        task();
#if DEBUG_VERBOSE
        CPLDebug("JOB", "%p finished a job", psWT);
#endif
        poTP->DeclareJobFinished();
    }
}

/************************************************************************/
/*                             SubmitJob()                              */
/************************************************************************/

/** Queue a new job.
 *
 * @param pfnFunc Function to run for the job.
 * @param pData User data to pass to the job function.
 * @return true in case of success.
 */
bool CPLWorkerThreadPool::SubmitJob(CPLThreadFunc pfnFunc, void *pData)
{
    return SubmitJob([=] { pfnFunc(pData); });
}

/** Queue a new job.
 *
 * @param task  Void function to execute.
 * @return true in case of success.
 */
bool CPLWorkerThreadPool::SubmitJob(std::function<void()> task)
{
#ifdef DEBUG
    {
        std::unique_lock<std::mutex> oGuard(m_mutex);
        CPLAssert(m_nMaxThreads > 0);
    }
#endif

    bool bMustIncrementWaitingWorkerThreadsAfterSubmission = false;
    if (threadLocalCurrentThreadPool == this)
    {
        // If there are waiting threads or we have not started all allowed
        // threads, we can submit this job asynchronously
        {
            std::unique_lock<std::mutex> oGuard(m_mutex);
            if (nWaitingWorkerThreads > 0 ||
                static_cast<int>(aWT.size()) < m_nMaxThreads)
            {
                bMustIncrementWaitingWorkerThreadsAfterSubmission = true;
                nWaitingWorkerThreads--;
            }
        }
        if (!bMustIncrementWaitingWorkerThreadsAfterSubmission)
        {
            // otherwise there is a risk of deadlock, so execute synchronously.
            task();
            return true;
        }
    }

    std::unique_lock<std::mutex> oGuard(m_mutex);

    if (bMustIncrementWaitingWorkerThreadsAfterSubmission)
        nWaitingWorkerThreads++;

    if (static_cast<int>(aWT.size()) < m_nMaxThreads)
    {
        // CPLDebug("CPL", "Starting new thread...");
        auto wt = std::make_unique<CPLWorkerThread>();
        wt->poTP = this;
        //ABELL - Why should this fail? And this is a *pool* thread, not necessarily
        //  tied to the submitted job. The submitted job still needs to run, even if
        //  this fails. If we can't create a thread, should the entire pool become invalid?
        wt->hThread = CPLCreateJoinableThread(WorkerThreadFunction, wt.get());
        /**
        if (!wt->hThread)
        {
            VSIFree(psJob);
            VSIFree(psItem);
            return false;
        }
        **/
        if (wt->hThread)
            aWT.emplace_back(std::move(wt));
    }

    jobQueue.emplace(task);
    nPendingJobs++;

    if (psWaitingWorkerThreadsList)
    {
        CPLWorkerThread *psWorkerThread =
            static_cast<CPLWorkerThread *>(psWaitingWorkerThreadsList->pData);

        CPLAssert(psWorkerThread->bMarkedAsWaiting);
        psWorkerThread->bMarkedAsWaiting = false;

        CPLList *psNext = psWaitingWorkerThreadsList->psNext;
        CPLList *psToFree = psWaitingWorkerThreadsList;
        psWaitingWorkerThreadsList = psNext;
        nWaitingWorkerThreads--;

#if DEBUG_VERBOSE
        CPLDebug("JOB", "Waking up %p", psWorkerThread);
#endif

        {
            std::lock_guard<std::mutex> oGuardWT(psWorkerThread->m_mutex);
            // coverity[ uninit_use_in_call]
            oGuard.unlock();
            psWorkerThread->m_cv.notify_one();
        }

        CPLFree(psToFree);
    }

    return true;
}

/************************************************************************/
/*                             SubmitJobs()                              */
/************************************************************************/

/** Queue several jobs
 *
 * @param pfnFunc Function to run for the job.
 * @param apData User data instances to pass to the job function.
 * @return true in case of success.
 */
bool CPLWorkerThreadPool::SubmitJobs(CPLThreadFunc pfnFunc,
                                     const std::vector<void *> &apData)
{
    if (apData.empty())
        return false;

#ifdef DEBUG
    {
        std::unique_lock<std::mutex> oGuard(m_mutex);
        CPLAssert(m_nMaxThreads > 0);
    }
#endif

    if (threadLocalCurrentThreadPool == this)
    {
        // If SubmitJob() is called from a worker thread of this queue,
        // then synchronously run the task to avoid deadlock.
        for (void *pData : apData)
            pfnFunc(pData);
        return true;
    }

    std::unique_lock<std::mutex> oGuard(m_mutex);

    for (void *pData : apData)
    {
        if (static_cast<int>(aWT.size()) < m_nMaxThreads)
        {
            std::unique_ptr<CPLWorkerThread> wt(new CPLWorkerThread);
            wt->poTP = this;
            wt->hThread =
                CPLCreateJoinableThread(WorkerThreadFunction, wt.get());
            if (wt->hThread == nullptr)
            {
                if (aWT.empty())
                    return false;
            }
            else
            {
                aWT.emplace_back(std::move(wt));
            }
        }

        jobQueue.emplace([=] { pfnFunc(pData); });
        nPendingJobs++;
    }

    for (size_t i = 0; i < apData.size(); i++)
    {
        if (psWaitingWorkerThreadsList)
        {
            CPLWorkerThread *psWorkerThread;

            psWorkerThread = static_cast<CPLWorkerThread *>(
                psWaitingWorkerThreadsList->pData);

            CPLAssert(psWorkerThread->bMarkedAsWaiting);
            psWorkerThread->bMarkedAsWaiting = false;

            CPLList *psNext = psWaitingWorkerThreadsList->psNext;
            CPLList *psToFree = psWaitingWorkerThreadsList;
            psWaitingWorkerThreadsList = psNext;
            nWaitingWorkerThreads--;

#if DEBUG_VERBOSE
            CPLDebug("JOB", "Waking up %p", psWorkerThread);
#endif
            {
                std::lock_guard<std::mutex> oGuardWT(psWorkerThread->m_mutex);
                // coverity[ uninit_use_in_call]
                oGuard.unlock();
                psWorkerThread->m_cv.notify_one();
            }

            CPLFree(psToFree);
            oGuard.lock();
        }
        else
        {
            break;
        }
    }

    return true;
}

/************************************************************************/
/*                            WaitCompletion()                          */
/************************************************************************/

/** Wait for completion of part or whole jobs.
 *
 * @param nMaxRemainingJobs Maximum number of pendings jobs that are allowed
 *                          in the queue after this method has completed. Might
 * be 0 to wait for all jobs.
 */
void CPLWorkerThreadPool::WaitCompletion(int nMaxRemainingJobs)
{
    if (nMaxRemainingJobs < 0)
        nMaxRemainingJobs = 0;
    std::unique_lock<std::mutex> oGuard(m_mutex);
    m_cv.wait(oGuard, [this, nMaxRemainingJobs]
              { return nPendingJobs <= nMaxRemainingJobs; });
}

/************************************************************************/
/*                            WaitEvent()                               */
/************************************************************************/

/** Wait for completion of at least one job, if there are any remaining
 */
void CPLWorkerThreadPool::WaitEvent()
{
    // NOTE - This isn't quite right. After nPendingJobsBefore is set but before
    // a notification occurs, jobs could be submitted which would increase
    // nPendingJobs, so a job completion may looks like a spurious wakeup.
    std::unique_lock<std::mutex> oGuard(m_mutex);
    if (nPendingJobs == 0)
        return;
    const int nPendingJobsBefore = nPendingJobs;
    m_cv.wait(oGuard, [this, nPendingJobsBefore]
              { return nPendingJobs < nPendingJobsBefore; });
}

/************************************************************************/
/*                                Setup()                               */
/************************************************************************/

/** Setup the pool.
 *
 * @param nThreads Number of threads to launch
 * @param pfnInitFunc Initialization function to run in each thread. May be NULL
 * @param pasInitData Array of initialization data. Its length must be nThreads,
 *                    or it should be NULL.
 * @return true if initialization was successful.
 */
bool CPLWorkerThreadPool::Setup(int nThreads, CPLThreadFunc pfnInitFunc,
                                void **pasInitData)
{
    return Setup(nThreads, pfnInitFunc, pasInitData, true);
}

/** Setup the pool.
 *
 * @param nThreads Number of threads to launch
 * @param pfnInitFunc Initialization function to run in each thread. May be NULL
 * @param pasInitData Array of initialization data. Its length must be nThreads,
 *                    or it should be NULL.
 * @param bWaitallStarted Whether to wait for all threads to be fully started.
 * @return true if initialization was successful.
 */
bool CPLWorkerThreadPool::Setup(int nThreads, CPLThreadFunc pfnInitFunc,
                                void **pasInitData, bool bWaitallStarted)
{
    CPLAssert(nThreads > 0);

    if (nThreads > static_cast<int>(aWT.size()) && pfnInitFunc == nullptr &&
        pasInitData == nullptr && !bWaitallStarted)
    {
        std::lock_guard<std::mutex> oGuard(m_mutex);
        if (nThreads > m_nMaxThreads)
            m_nMaxThreads = nThreads;
        return true;
    }

    bool bRet = true;
    for (int i = static_cast<int>(aWT.size()); i < nThreads; i++)
    {
        auto wt = std::make_unique<CPLWorkerThread>();
        wt->pfnInitFunc = pfnInitFunc;
        wt->pInitData = pasInitData ? pasInitData[i] : nullptr;
        wt->poTP = this;
        wt->hThread = CPLCreateJoinableThread(WorkerThreadFunction, wt.get());
        if (wt->hThread == nullptr)
        {
            nThreads = i;
            bRet = false;
            break;
        }
        aWT.emplace_back(std::move(wt));
    }

    {
        std::lock_guard<std::mutex> oGuard(m_mutex);
        if (nThreads > m_nMaxThreads)
            m_nMaxThreads = nThreads;
    }

    if (bWaitallStarted)
    {
        // Wait all threads to be started
        std::unique_lock<std::mutex> oGuard(m_mutex);
        while (nWaitingWorkerThreads < nThreads)
        {
            m_cv.wait(oGuard);
        }
    }

    if (eState == CPLWTS_ERROR)
        bRet = false;

    return bRet;
}

/************************************************************************/
/*                          DeclareJobFinished()                        */
/************************************************************************/

void CPLWorkerThreadPool::DeclareJobFinished()
{
    std::lock_guard<std::mutex> oGuard(m_mutex);
    nPendingJobs--;
    m_cv.notify_one();
}

/************************************************************************/
/*                             GetNextJob()                             */
/************************************************************************/

std::function<void()>
CPLWorkerThreadPool::GetNextJob(CPLWorkerThread *psWorkerThread)
{
    while (true)
    {
        std::unique_lock<std::mutex> oGuard(m_mutex);
        if (eState == CPLWTS_STOP)
            return std::function<void()>();

        if (jobQueue.size())
        {
#if DEBUG_VERBOSE
            CPLDebug("JOB", "%p got a job", psWorkerThread);
#endif
            auto task = std::move(jobQueue.front());
            jobQueue.pop();
            return task;
        }

        if (!psWorkerThread->bMarkedAsWaiting)
        {
            psWorkerThread->bMarkedAsWaiting = true;
            nWaitingWorkerThreads++;

            CPLList *psItem =
                static_cast<CPLList *>(VSI_MALLOC_VERBOSE(sizeof(CPLList)));
            if (psItem == nullptr)
            {
                eState = CPLWTS_ERROR;
                m_cv.notify_one();

                return nullptr;
            }

            psItem->pData = psWorkerThread;
            psItem->psNext = psWaitingWorkerThreadsList;
            psWaitingWorkerThreadsList = psItem;

#if DEBUG_VERBOSE
            CPLAssert(CPLListCount(psWaitingWorkerThreadsList) ==
                      nWaitingWorkerThreads);
#endif
        }

        m_cv.notify_one();

#if DEBUG_VERBOSE
        CPLDebug("JOB", "%p sleeping", psWorkerThread);
#endif

        std::unique_lock<std::mutex> oGuardThisThread(psWorkerThread->m_mutex);
        // coverity[uninit_use_in_call]
        oGuard.unlock();
        // coverity[wait_not_in_locked_loop]
        psWorkerThread->m_cv.wait(oGuardThisThread);
    }
}

/************************************************************************/
/*                         CreateJobQueue()                             */
/************************************************************************/

/** Create a new job queue based on this worker thread pool.
 *
 * The worker thread pool must remain alive while the returned object is
 * itself alive.
 *
 * @since GDAL 3.2
 */
std::unique_ptr<CPLJobQueue> CPLWorkerThreadPool::CreateJobQueue()
{
    return std::unique_ptr<CPLJobQueue>(new CPLJobQueue(this));
}

/************************************************************************/
/*                            CPLJobQueue()                             */
/************************************************************************/

//! @cond Doxygen_Suppress
CPLJobQueue::CPLJobQueue(CPLWorkerThreadPool *poPool) : m_poPool(poPool)
{
}

//! @endcond

/************************************************************************/
/*                           ~CPLJobQueue()                             */
/************************************************************************/

CPLJobQueue::~CPLJobQueue()
{
    WaitCompletion();
}

/************************************************************************/
/*                          DeclareJobFinished()                        */
/************************************************************************/

void CPLJobQueue::DeclareJobFinished()
{
    std::lock_guard<std::mutex> oGuard(m_mutex);
    m_nPendingJobs--;
    m_cv.notify_one();
}

/************************************************************************/
/*                             SubmitJob()                              */
/************************************************************************/

/** Queue a new job.
 *
 * @param pfnFunc Function to run for the job.
 * @param pData User data to pass to the job function.
 * @return true in case of success.
 */
bool CPLJobQueue::SubmitJob(CPLThreadFunc pfnFunc, void *pData)
{
    return SubmitJob([=] { pfnFunc(pData); });
}

/** Queue a new job.
 *
 * @param task  Task to execute.
 * @return true in case of success.
 */
bool CPLJobQueue::SubmitJob(std::function<void()> task)
{
    {
        std::lock_guard<std::mutex> oGuard(m_mutex);
        m_nPendingJobs++;
    }

    // coverity[uninit_member,copy_constructor_call]
    const auto lambda = [this, task]
    {
        task();
        DeclareJobFinished();
    };
    // cppcheck-suppress knownConditionTrueFalse
    return m_poPool->SubmitJob(lambda);
}

/************************************************************************/
/*                            WaitCompletion()                          */
/************************************************************************/

/** Wait for completion of part or whole jobs.
 *
 * @param nMaxRemainingJobs Maximum number of pendings jobs that are allowed
 *                          in the queue after this method has completed. Might
 * be 0 to wait for all jobs.
 */
void CPLJobQueue::WaitCompletion(int nMaxRemainingJobs)
{
    std::unique_lock<std::mutex> oGuard(m_mutex);
    m_cv.wait(oGuard, [this, nMaxRemainingJobs]
              { return m_nPendingJobs <= nMaxRemainingJobs; });
}

/************************************************************************/
/*                             WaitEvent()                              */
/************************************************************************/

/** Wait for completion for at least one job.
 *
 * @return true if there are remaining jobs.
 */
bool CPLJobQueue::WaitEvent()
{
    // NOTE - This isn't quite right. After nPendingJobsBefore is set but before
    // a notification occurs, jobs could be submitted which would increase
    // nPendingJobs, so a job completion may looks like a spurious wakeup.
    std::unique_lock<std::mutex> oGuard(m_mutex);
    if (m_nPendingJobs == 0)
        return false;

    const int nPendingJobsBefore = m_nPendingJobs;
    m_cv.wait(oGuard, [this, nPendingJobsBefore]
              { return m_nPendingJobs < nPendingJobsBefore; });
    return m_nPendingJobs > 0;
}

1	/**********************************************************************
2	*
3	* Project: CPL - Common Portability Library
4	* Purpose: CPL worker thread pool
5	* Author: Even Rouault, <even dot rouault at spatialys dot com>
6	*
7	**********************************************************************
8	* Copyright (c) 2015, Even Rouault, <even dot rouault at spatialys dot com>
9	*
10	* SPDX-License-Identifier: MIT
11	****************************************************************************/
12
13	#include "cpl_port.h"
14	#include "cpl_worker_thread_pool.h"
15
16	#include <cstddef>
17	#include <memory>
18
19	#include "cpl_conv.h"
20	#include "cpl_error.h"
21	#include "cpl_vsi.h"
22
23	static thread_local CPLWorkerThreadPool *threadLocalCurrentThreadPool = nullptr;
24
25	/************************************************************************/
26	/* CPLWorkerThreadPool() */
27	/************************************************************************/
28
29	/** Instantiate a new pool of worker threads.
30	*
31	* The pool is in an uninitialized state after this call. The Setup() method
32	* must be called.
33	*/
34	CPLWorkerThreadPool::CPLWorkerThreadPool() : jobQueue{}	610✔
35	{
36	}	610✔
37
38	/** Instantiate a new pool of worker threads.
39	*
40	* \param nThreads Number of threads in the pool.
41	*/
42	CPLWorkerThreadPool::CPLWorkerThreadPool(int nThreads) : jobQueue{}	625✔
43	{
44	Setup(nThreads, nullptr, nullptr);	625✔
45	}	624✔
46
47	/************************************************************************/
48	/* ~CPLWorkerThreadPool() */
49	/************************************************************************/
50
51	/** Destroys a pool of worker threads.
52	*
53	* Any still pending job will be completed before the destructor returns.
54	*/
55	CPLWorkerThreadPool::~CPLWorkerThreadPool()	1,230✔
56	{
57	WaitCompletion();	1,230✔
58
59	{
60	std::lock_guard<std::mutex> oGuard(m_mutex);	1,230✔
61	eState = CPLWTS_STOP;	1,230✔
62	}
63
64	for (auto &wt : aWT)	4,894✔
65	{
66	{
67	std::lock_guard<std::mutex> oGuard(wt->m_mutex);	7,328✔
68	wt->m_cv.notify_one();	3,664✔
69	}
70	CPLJoinThread(wt->hThread);	3,664✔
71	}
72
73	CPLListDestroy(psWaitingWorkerThreadsList);	1,230✔
74	}	1,230✔
75
76	/************************************************************************/
77	/* GetThreadCount() */
78	/************************************************************************/
79
80	int CPLWorkerThreadPool::GetThreadCount() const	1,140✔
81	{
82	std::unique_lock<std::mutex> oGuard(m_mutex);	1,140✔
83	return m_nMaxThreads;	2,280✔
84	}
85
86	/************************************************************************/
87	/* WorkerThreadFunction() */
88	/************************************************************************/
89
90	void CPLWorkerThreadPool::WorkerThreadFunction(void *user_data)	4,700✔
91	{
92	CPLWorkerThread psWT = static_cast<CPLWorkerThread >(user_data);	4,700✔
93	CPLWorkerThreadPool *poTP = psWT->poTP;	4,700✔
94
95	threadLocalCurrentThreadPool = poTP;	4,700✔
96
97	if (psWT->pfnInitFunc)	4,700✔
98	psWT->pfnInitFunc(psWT->pInitData);	×
99
100	while (true)
101	{
102	std::function<void()> task = poTP->GetNextJob(psWT);	154,164✔
103	if (!task)	153,078✔
104	break;	3,664✔
105
106	task();	149,367✔
107	#if DEBUG_VERBOSE
108	CPLDebug("JOB", "%p finished a job", psWT);
109	#endif
110	poTP->DeclareJobFinished();	149,429✔
111	}	149,464✔
112	}	3,684✔
113
114	/************************************************************************/
115	/* SubmitJob() */
116	/************************************************************************/
117
118	/** Queue a new job.
119	*
120	* @param pfnFunc Function to run for the job.
121	* @param pData User data to pass to the job function.
122	* @return true in case of success.
123	*/
124	bool CPLWorkerThreadPool::SubmitJob(CPLThreadFunc pfnFunc, void *pData)	6,909✔
125	{
126	return SubmitJob([=] { pfnFunc(pData); });	13,804✔
127	}
128
129	/** Queue a new job.
130	*
131	* @param task Void function to execute.
132	* @return true in case of success.
133	*/
134	bool CPLWorkerThreadPool::SubmitJob(std::function<void()> task)	191,237✔
135	{
136	#ifdef DEBUG
137	{
138	std::unique_lock<std::mutex> oGuard(m_mutex);	382,318✔
139	CPLAssert(m_nMaxThreads > 0);	191,081✔
140	}
141	#endif
142
143	bool bMustIncrementWaitingWorkerThreadsAfterSubmission = false;	191,024✔
144	if (threadLocalCurrentThreadPool == this)	191,024✔
145	{
146	// If there are waiting threads or we have not started all allowed
147	// threads, we can submit this job asynchronously
148	{
149	std::unique_lock<std::mutex> oGuard(m_mutex);	331,884✔
150	if (nWaitingWorkerThreads > 0 \|\|	209,137✔
151	static_cast<int>(aWT.size()) < m_nMaxThreads)	43,220✔
152	{
153	bMustIncrementWaitingWorkerThreadsAfterSubmission = true;	122,767✔
154	nWaitingWorkerThreads--;	122,767✔
155	}
156	}
157	if (!bMustIncrementWaitingWorkerThreadsAfterSubmission)	165,754✔
158	{
159	// otherwise there is a risk of deadlock, so execute synchronously.
160	task();	43,208✔
161	return true;	43,197✔
162	}
163	}
164
165	std::unique_lock<std::mutex> oGuard(m_mutex);	147,603✔
166
167	if (bMustIncrementWaitingWorkerThreadsAfterSubmission)	148,004✔
168	nWaitingWorkerThreads++;	122,794✔
169
170	if (static_cast<int>(aWT.size()) < m_nMaxThreads)	148,004✔
171	{
172	// CPLDebug("CPL", "Starting new thread...");
173	auto wt = std::make_unique<CPLWorkerThread>();	2,176✔
174	wt->poTP = this;	1,088✔
175	//ABELL - Why should this fail? And this is a pool thread, not necessarily
176	// tied to the submitted job. The submitted job still needs to run, even if
177	// this fails. If we can't create a thread, should the entire pool become invalid?
178	wt->hThread = CPLCreateJoinableThread(WorkerThreadFunction, wt.get());	1,088✔
179	/**
180	if (!wt->hThread)
181	{
182	VSIFree(psJob);
183	VSIFree(psItem);
184	return false;
185	}
186	**/
187	if (wt->hThread)	1,088✔
188	aWT.emplace_back(std::move(wt));	1,088✔
189	}
190
191	jobQueue.emplace(task);	147,854✔
192	nPendingJobs++;	147,935✔
193
194	if (psWaitingWorkerThreadsList)	147,935✔
195	{
196	CPLWorkerThread *psWorkerThread =	141,785✔
197	static_cast<CPLWorkerThread *>(psWaitingWorkerThreadsList->pData);	141,785✔
198
199	CPLAssert(psWorkerThread->bMarkedAsWaiting);	141,785✔
200	psWorkerThread->bMarkedAsWaiting = false;	141,785✔
201
202	CPLList *psNext = psWaitingWorkerThreadsList->psNext;	141,785✔
203	CPLList *psToFree = psWaitingWorkerThreadsList;	141,785✔
204	psWaitingWorkerThreadsList = psNext;	141,785✔
205	nWaitingWorkerThreads--;	141,785✔
206
207	#if DEBUG_VERBOSE
208	CPLDebug("JOB", "Waking up %p", psWorkerThread);
209	#endif
210
211	{
212	std::lock_guard<std::mutex> oGuardWT(psWorkerThread->m_mutex);	283,726✔
213	// coverity[ uninit_use_in_call]
214	oGuard.unlock();	141,915✔
215	psWorkerThread->m_cv.notify_one();	141,903✔
216	}
217
218	CPLFree(psToFree);	142,007✔
219	}
220
221	return true;	148,128✔
222	}
223
224	/************************************************************************/
225	/* SubmitJobs() */
226	/************************************************************************/
227
228	/** Queue several jobs
229	*
230	* @param pfnFunc Function to run for the job.
231	* @param apData User data instances to pass to the job function.
232	* @return true in case of success.
233	*/
234	bool CPLWorkerThreadPool::SubmitJobs(CPLThreadFunc pfnFunc,	156✔
235	const std::vector<void *> &apData)
236	{
237	if (apData.empty())	156✔
238	return false;	×
239
240	#ifdef DEBUG
241	{
242	std::unique_lock<std::mutex> oGuard(m_mutex);	312✔
243	CPLAssert(m_nMaxThreads > 0);	156✔
244	}
245	#endif
246
247	if (threadLocalCurrentThreadPool == this)	156✔
248	{
249	// If SubmitJob() is called from a worker thread of this queue,
250	// then synchronously run the task to avoid deadlock.
251	for (void *pData : apData)	×
252	pfnFunc(pData);	×
253	return true;	×
254	}
255
256	std::unique_lock<std::mutex> oGuard(m_mutex);	312✔
257
258	for (void *pData : apData)	1,534✔
259	{
260	if (static_cast<int>(aWT.size()) < m_nMaxThreads)	1,378✔
261	{
262	std::unique_ptr<CPLWorkerThread> wt(new CPLWorkerThread);	×
263	wt->poTP = this;	×
264	wt->hThread =	×
265	CPLCreateJoinableThread(WorkerThreadFunction, wt.get());	×
266	if (wt->hThread == nullptr)	×
267	{
268	if (aWT.empty())	×
269	return false;	×
270	}
271	else
272	{
273	aWT.emplace_back(std::move(wt));	×
274	}
275	}
276
277	jobQueue.emplace([=] { pfnFunc(pData); });	2,754✔
278	nPendingJobs++;	1,378✔
279	}
280
281	for (size_t i = 0; i < apData.size(); i++)	535✔
282	{
283	if (psWaitingWorkerThreadsList)	381✔
284	{
285	CPLWorkerThread *psWorkerThread;
286
287	psWorkerThread = static_cast<CPLWorkerThread *>(	379✔
288	psWaitingWorkerThreadsList->pData);	379✔
289
290	CPLAssert(psWorkerThread->bMarkedAsWaiting);	379✔
291	psWorkerThread->bMarkedAsWaiting = false;	379✔
292
293	CPLList *psNext = psWaitingWorkerThreadsList->psNext;	379✔
294	CPLList *psToFree = psWaitingWorkerThreadsList;	379✔
295	psWaitingWorkerThreadsList = psNext;	379✔
296	nWaitingWorkerThreads--;	379✔
297
298	#if DEBUG_VERBOSE
299	CPLDebug("JOB", "Waking up %p", psWorkerThread);
300	#endif
301	{
302	std::lock_guard<std::mutex> oGuardWT(psWorkerThread->m_mutex);	758✔
303	// coverity[ uninit_use_in_call]
304	oGuard.unlock();	379✔
305	psWorkerThread->m_cv.notify_one();	379✔
306	}
307
308	CPLFree(psToFree);	379✔
309	oGuard.lock();	379✔
310	}
311	else
312	{
313	break;	2✔
314	}
315	}
316
317	return true;	156✔
318	}
319
320	/************************************************************************/
321	/* WaitCompletion() */
322	/************************************************************************/
323
324	/** Wait for completion of part or whole jobs.
325	*
326	* @param nMaxRemainingJobs Maximum number of pendings jobs that are allowed
327	* in the queue after this method has completed. Might
328	* be 0 to wait for all jobs.
329	*/
330	void CPLWorkerThreadPool::WaitCompletion(int nMaxRemainingJobs)	5,841✔
331	{
332	if (nMaxRemainingJobs < 0)	5,841✔
333	nMaxRemainingJobs = 0;	×
334	std::unique_lock<std::mutex> oGuard(m_mutex);	11,682✔
335	m_cv.wait(oGuard, [this, nMaxRemainingJobs]	5,841✔
336	{ return nPendingJobs <= nMaxRemainingJobs; });	7,506✔
337	}	5,841✔
338
339	/************************************************************************/
340	/* WaitEvent() */
341	/************************************************************************/
342
343	/** Wait for completion of at least one job, if there are any remaining
344	*/
345	void CPLWorkerThreadPool::WaitEvent()	1,099✔
346	{
347	// NOTE - This isn't quite right. After nPendingJobsBefore is set but before
348	// a notification occurs, jobs could be submitted which would increase
349	// nPendingJobs, so a job completion may looks like a spurious wakeup.
350	std::unique_lock<std::mutex> oGuard(m_mutex);	1,099✔
351	if (nPendingJobs == 0)	1,099✔
352	return;	45✔
353	const int nPendingJobsBefore = nPendingJobs;	1,054✔
354	m_cv.wait(oGuard, [this, nPendingJobsBefore]	1,054✔
355	{ return nPendingJobs < nPendingJobsBefore; });	2,362✔
356	}
357
358	/************************************************************************/
359	/* Setup() */
360	/************************************************************************/
361
362	/** Setup the pool.
363	*
364	* @param nThreads Number of threads to launch
365	* @param pfnInitFunc Initialization function to run in each thread. May be NULL
366	* @param pasInitData Array of initialization data. Its length must be nThreads,
367	* or it should be NULL.
368	* @return true if initialization was successful.
369	*/
370	bool CPLWorkerThreadPool::Setup(int nThreads, CPLThreadFunc pfnInitFunc,	911✔
371	void **pasInitData)
372	{
373	return Setup(nThreads, pfnInitFunc, pasInitData, true);	911✔
374	}
375
376	/** Setup the pool.
377	*
378	* @param nThreads Number of threads to launch
379	* @param pfnInitFunc Initialization function to run in each thread. May be NULL
380	* @param pasInitData Array of initialization data. Its length must be nThreads,
381	* or it should be NULL.
382	* @param bWaitallStarted Whether to wait for all threads to be fully started.
383	* @return true if initialization was successful.
384	*/
385	bool CPLWorkerThreadPool::Setup(int nThreads, CPLThreadFunc pfnInitFunc,	933✔
386	void **pasInitData, bool bWaitallStarted)
387	{
388	CPLAssert(nThreads > 0);	933✔
389
390	if (nThreads > static_cast<int>(aWT.size()) && pfnInitFunc == nullptr &&	1,866✔
391	pasInitData == nullptr && !bWaitallStarted)	1,866✔
392	{
393	std::lock_guard<std::mutex> oGuard(m_mutex);	21✔
394	if (nThreads > m_nMaxThreads)	21✔
395	m_nMaxThreads = nThreads;	21✔
396	return true;	21✔
397	}
398
399	bool bRet = true;	912✔
400	for (int i = static_cast<int>(aWT.size()); i < nThreads; i++)	4,526✔
401	{
402	auto wt = std::make_unique<CPLWorkerThread>();	3,614✔
403	wt->pfnInitFunc = pfnInitFunc;	3,614✔
404	wt->pInitData = pasInitData ? pasInitData[i] : nullptr;	3,614✔
405	wt->poTP = this;	3,614✔
406	wt->hThread = CPLCreateJoinableThread(WorkerThreadFunction, wt.get());	3,614✔
407	if (wt->hThread == nullptr)	3,614✔
408	{
409	nThreads = i;	×
410	bRet = false;	×
411	break;	×
412	}
413	aWT.emplace_back(std::move(wt));	3,614✔
414	}
415
416	{
417	std::lock_guard<std::mutex> oGuard(m_mutex);	1,824✔
418	if (nThreads > m_nMaxThreads)	912✔
419	m_nMaxThreads = nThreads;	912✔
420	}
421
422	if (bWaitallStarted)	912✔
423	{
424	// Wait all threads to be started
425	std::unique_lock<std::mutex> oGuard(m_mutex);	1,824✔
426	while (nWaitingWorkerThreads < nThreads)	2,914✔
427	{
428	m_cv.wait(oGuard);	2,002✔
429	}
430	}
431
432	if (eState == CPLWTS_ERROR)	911✔
433	bRet = false;	×
434
435	return bRet;	911✔
436	}
437
438	/************************************************************************/
439	/* DeclareJobFinished() */
440	/************************************************************************/
441
442	void CPLWorkerThreadPool::DeclareJobFinished()	149,383✔
443	{
444	std::lock_guard<std::mutex> oGuard(m_mutex);	298,867✔
445	nPendingJobs--;	149,486✔
446	m_cv.notify_one();	149,486✔
447	}	149,464✔
448
449	/************************************************************************/
450	/* GetNextJob() */
451	/************************************************************************/
452
453	std::function<void()>
454	CPLWorkerThreadPool::GetNextJob(CPLWorkerThread *psWorkerThread)	299,665✔
455	{
456	while (true)
457	{
458	std::unique_lock<std::mutex> oGuard(m_mutex);	299,665✔
459	if (eState == CPLWTS_STOP)	299,717✔
460	return std::function<void()>();	3,664✔
461
462	if (jobQueue.size())	296,053✔
463	{
464	#if DEBUG_VERBOSE
465	CPLDebug("JOB", "%p got a job", psWorkerThread);
466	#endif
467	auto task = std::move(jobQueue.front());	298,931✔
468	jobQueue.pop();	149,405✔
469	return task;	149,365✔
470	}
471
472	if (!psWorkerThread->bMarkedAsWaiting)	146,659✔
473	{
474	psWorkerThread->bMarkedAsWaiting = true;	146,701✔
475	nWaitingWorkerThreads++;	146,701✔
476
477	CPLList *psItem =
478	static_cast<CPLList *>(VSI_MALLOC_VERBOSE(sizeof(CPLList)));	146,701✔
479	if (psItem == nullptr)	146,740✔
480	{
481	eState = CPLWTS_ERROR;	×
482	m_cv.notify_one();	×
483
484	return nullptr;	×
485	}
486
487	psItem->pData = psWorkerThread;	146,740✔
488	psItem->psNext = psWaitingWorkerThreadsList;	146,740✔
489	psWaitingWorkerThreadsList = psItem;	146,740✔
490
491	#if DEBUG_VERBOSE
492	CPLAssert(CPLListCount(psWaitingWorkerThreadsList) ==
493	nWaitingWorkerThreads);
494	#endif
495	}
496
497	m_cv.notify_one();	146,698✔
498
499	#if DEBUG_VERBOSE
500	CPLDebug("JOB", "%p sleeping", psWorkerThread);
501	#endif
502
503	std::unique_lock<std::mutex> oGuardThisThread(psWorkerThread->m_mutex);	292,393✔
504	// coverity[uninit_use_in_call]
505	oGuard.unlock();	146,721✔
506	// coverity[wait_not_in_locked_loop]
507	psWorkerThread->m_cv.wait(oGuardThisThread);	146,698✔
508	}	145,539✔
509	}
510
511	/************************************************************************/
512	/* CreateJobQueue() */
513	/************************************************************************/
514
515	/** Create a new job queue based on this worker thread pool.
516	*
517	* The worker thread pool must remain alive while the returned object is
518	* itself alive.
519	*
520	* @since GDAL 3.2
521	*/
522	std::unique_ptr<CPLJobQueue> CPLWorkerThreadPool::CreateJobQueue()	84,796✔
523	{
524	return std::unique_ptr<CPLJobQueue>(new CPLJobQueue(this));	84,796✔
525	}
526
527	/************************************************************************/
528	/* CPLJobQueue() */
529	/************************************************************************/
530
531	//! @cond Doxygen_Suppress
532	CPLJobQueue::CPLJobQueue(CPLWorkerThreadPool *poPool) : m_poPool(poPool)	84,886✔
533	{
534	}	84,920✔
535
536	//! @endcond
537
538	/************************************************************************/
539	/* ~CPLJobQueue() */
540	/************************************************************************/
541
542	CPLJobQueue::~CPLJobQueue()	84,975✔
543	{
544	WaitCompletion();	84,975✔
545	}	84,976✔
546
547	/************************************************************************/
548	/* DeclareJobFinished() */
549	/************************************************************************/
550
551	void CPLJobQueue::DeclareJobFinished()	184,263✔
552	{
553	std::lock_guard<std::mutex> oGuard(m_mutex);	368,705✔
554	m_nPendingJobs--;	184,307✔
555	m_cv.notify_one();	184,307✔
556	}	184,166✔
557
558	/************************************************************************/
559	/* SubmitJob() */
560	/************************************************************************/
561
562	/** Queue a new job.
563	*
564	* @param pfnFunc Function to run for the job.
565	* @param pData User data to pass to the job function.
566	* @return true in case of success.
567	*/
568	bool CPLJobQueue::SubmitJob(CPLThreadFunc pfnFunc, void *pData)	14,778✔
569	{
570	return SubmitJob([=] { pfnFunc(pData); });	29,550✔
571	}
572
573	/** Queue a new job.
574	*
575	* @param task Task to execute.
576	* @return true in case of success.
577	*/
578	bool CPLJobQueue::SubmitJob(std::function<void()> task)	184,305✔
579	{
580	{
581	std::lock_guard<std::mutex> oGuard(m_mutex);	184,305✔
582	m_nPendingJobs++;	184,193✔
583	}
584
585	// coverity[uninit_member,copy_constructor_call]
586	const auto lambda = [this, task]	368,493✔
587	{
588	task();	184,209✔
589	DeclareJobFinished();	184,284✔
590	};	184,138✔
591	// cppcheck-suppress knownConditionTrueFalse
592	return m_poPool->SubmitJob(lambda);	368,480✔
593	}
594
595	/************************************************************************/
596	/* WaitCompletion() */
597	/************************************************************************/
598
599	/** Wait for completion of part or whole jobs.
600	*
601	* @param nMaxRemainingJobs Maximum number of pendings jobs that are allowed
602	* in the queue after this method has completed. Might
603	* be 0 to wait for all jobs.
604	*/
605	void CPLJobQueue::WaitCompletion(int nMaxRemainingJobs)	169,335✔
606	{
607	std::unique_lock<std::mutex> oGuard(m_mutex);	338,648✔
608	m_cv.wait(oGuard, [this, nMaxRemainingJobs]	169,327✔
609	{ return m_nPendingJobs <= nMaxRemainingJobs; });	219,830✔
610	}	169,349✔
611
612	/************************************************************************/
613	/* WaitEvent() */
614	/************************************************************************/
615
616	/** Wait for completion for at least one job.
617	*
618	* @return true if there are remaining jobs.
619	*/
620	bool CPLJobQueue::WaitEvent()	243✔
621	{
622	// NOTE - This isn't quite right. After nPendingJobsBefore is set but before
623	// a notification occurs, jobs could be submitted which would increase
624	// nPendingJobs, so a job completion may looks like a spurious wakeup.
625	std::unique_lock<std::mutex> oGuard(m_mutex);	486✔
626	if (m_nPendingJobs == 0)	243✔
627	return false;	×
628
629	const int nPendingJobsBefore = m_nPendingJobs;	243✔
630	m_cv.wait(oGuard, [this, nPendingJobsBefore]	243✔
631	{ return m_nPendingJobs < nPendingJobsBefore; });	486✔
632	return m_nPendingJobs > 0;	243✔
633	}

OSGeo / gdal / 13872211292

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous