10#include <QtCore/qpointer.h>
16using namespace std::chrono_literals;
20using namespace Qt::StringLiterals;
23struct QThreadGlobalPool
25 QPointer<QThreadPool> theInstance;
29Q_GLOBAL_STATIC(QThreadGlobalPool, globalPool)
32
33
48
49
53
54
56 :
manager(manager), runnable(
nullptr)
58 setStackSize(manager->stackSize);
62
63
66 QMutexLocker locker(&
manager->mutex);
68 QRunnable *r = runnable;
74 const bool del = r->autoDelete();
78#ifndef QT_NO_EXCEPTIONS
82#ifndef QT_NO_EXCEPTIONS
84 qWarning(
"Qt Concurrent has caught an exception thrown from a worker thread.\n"
85 "This is not supported, exceptions thrown in worker threads must be\n"
86 "caught before control returns to Qt Concurrent.");
115 if (!
manager->allThreads.contains(
this)) {
122 manager->expiredThreads.enqueue(
this);
126 manager->waitingThreads.enqueue(
this);
129 runnableReady.wait(locker.mutex(), QDeadlineTimer(manager->expiryTimeout));
131 if (!
manager->allThreads.contains(
this)) {
132 Q_ASSERT(
manager->queue.isEmpty());
135 if (
manager->waitingThreads.removeOne(
this)) {
136 manager->expiredThreads.enqueue(
this);
146 manager->noActiveThreads.wakeAll();
151
152
158 if (!globalPool.exists())
161 QThreadPool *globalThreadPool = []{
162 const QMutexLocker locker(&theMutex);
163 QThreadPool *p = globalPool->theInstance.get();
164 globalPool->theInstance =
nullptr;
167 if (globalThreadPool)
168 delete globalThreadPool;
173 Q_ASSERT(task !=
nullptr);
174 if (allThreads.isEmpty()) {
184 if (!waitingThreads.isEmpty()) {
187 waitingThreads.takeFirst()->runnableReady.wakeOne();
191 if (!expiredThreads.isEmpty()) {
194 Q_ASSERT(thread->runnable ==
nullptr);
198 thread->runnable = task;
203 Q_ASSERT(thread->isFinished());
204 thread->start(threadPriority);
220 Q_ASSERT(runnable !=
nullptr);
221 for (QueuePage *page : std::as_const(queue)) {
222 if (page->priority() == priority && !page->isFull()) {
223 page->push(runnable);
227 auto it =
std::upper_bound(queue.constBegin(), queue.constEnd(), priority,
comparePriority);
228 queue.insert(std::distance(queue.constBegin(), it),
new QueuePage(runnable, priority));
233 return (allThreads.size()
234 - expiredThreads.size()
235 - waitingThreads.size()
242 while (!queue.isEmpty()) {
244 if (!tryStart(page->first()))
269
270
273 Q_ASSERT(runnable !=
nullptr);
275 if (objectName.isEmpty())
276 objectName = u"Thread (pooled)"_s;
277 thread->setObjectName(objectName);
278 thread->setServiceLevel(serviceLevel);
279 Q_ASSERT(!allThreads.contains(thread.get()));
280 allThreads.insert(thread.get());
283 thread->runnable = runnable;
284 thread.release()->start(threadPriority);
288
289
290
291
292
293
297 auto allThreadsCopy = std::exchange(allThreads, {});
298 expiredThreads.clear();
299 waitingThreads.clear();
303 for (QThreadPoolThread *thread : std::as_const(allThreadsCopy)) {
304 if (thread->isRunning()) {
305 thread->runnableReady.wakeAll();
315
316
317
318
321 QMutexLocker locker(&mutex);
322 while (!(queue.isEmpty() && activeThreads == 0) && !timer.hasExpired())
323 noActiveThreads.wait(&mutex, timer);
325 if (!queue.isEmpty() || activeThreads)
337 QMutexLocker locker(&mutex);
338 while (!queue.isEmpty()) {
339 auto *page = queue.takeLast();
340 while (!page->isFinished()) {
341 QRunnable *r = page->pop();
342 if (r && r->autoDelete()) {
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369bool QThreadPool::tryTake(QRunnable *runnable)
373 if (runnable ==
nullptr)
376 QMutexLocker locker(&d->mutex);
377 for (QueuePage *page : std::as_const(d->queue)) {
378 if (page->tryTake(runnable)) {
379 if (page->isFinished()) {
380 d->queue.removeOne(page);
391
392
393
394
395
399 if (!q->tryTake(runnable))
402 const bool del = runnable->autoDelete();
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
464
465
466QThreadPool::QThreadPool(QObject *parent)
467 : QObject(*
new QThreadPoolPrivate, parent)
470 connect(
this, &QObject::objectNameChanged,
this, [d](
const QString &newName) {
472 QMutexLocker locker(&d->mutex);
473 d->objectName = newName;
478
479
480
481QThreadPool::~QThreadPool()
485 Q_ASSERT(d->queue.isEmpty());
486 Q_ASSERT(d->allThreads.isEmpty());
490
491
492QThreadPool *QThreadPool::globalInstance()
494 if (!globalPool.exists() && QCoreApplication::closingDown())
497 QPointer<QThreadPool> &theInstance = globalPool->theInstance;
498 const QMutexLocker locker(&theMutex);
500 theInstance =
new QThreadPool();
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519void QThreadPool::start(QRunnable *runnable,
int priority)
525 QMutexLocker locker(&d->mutex);
527 if (!d->tryStart(runnable))
528 d->enqueueTask(runnable, priority);
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564bool QThreadPool::tryStart(QRunnable *runnable)
570 QMutexLocker locker(&d->mutex);
571 if (d->tryStart(runnable))
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
595
596
597
598
599
600
601
602
603
604
605
606
607
609int QThreadPool::expiryTimeout()
const
611 using namespace std::chrono;
612 Q_D(
const QThreadPool);
613 QMutexLocker locker(&d->mutex);
614 if (d->expiryTimeout ==
decltype(d->expiryTimeout)::max())
616 return duration_cast<milliseconds>(d->expiryTimeout).count();
619void QThreadPool::setExpiryTimeout(
int expiryTimeout)
622 QMutexLocker locker(&d->mutex);
623 if (expiryTimeout < 0)
624 d->expiryTimeout =
decltype(d->expiryTimeout)::max();
626 d->expiryTimeout = expiryTimeout * 1ms;
630
631
632
633
634
635
636
637
638
639
641int QThreadPool::maxThreadCount()
const
643 Q_D(
const QThreadPool);
644 QMutexLocker locker(&d->mutex);
645 return d->requestedMaxThreadCount;
648void QThreadPool::setMaxThreadCount(
int maxThreadCount)
651 QMutexLocker locker(&d->mutex);
653 if (maxThreadCount == d->requestedMaxThreadCount)
656 d->requestedMaxThreadCount = maxThreadCount;
657 d->tryToStartMoreThreads();
661
662
663
664
665
666
667
668
670int QThreadPool::activeThreadCount()
const
672 Q_D(
const QThreadPool);
673 QMutexLocker locker(&d->mutex);
674 return d->activeThreadCount();
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692void QThreadPool::reserveThread()
695 QMutexLocker locker(&d->mutex);
696 ++d->reservedThreads;
700
701
702
703
704
705
706
707
708
709
710
711void QThreadPool::setStackSize(uint stackSize)
714 QMutexLocker locker(&d->mutex);
715 d->stackSize = stackSize;
718uint QThreadPool::stackSize()
const
720 Q_D(
const QThreadPool);
721 QMutexLocker locker(&d->mutex);
726
727
728
729
730
731
732
733
734
735
736
737
739void QThreadPool::setThreadPriority(QThread::Priority priority)
742 QMutexLocker locker(&d->mutex);
743 d->threadPriority = priority;
746QThread::Priority QThreadPool::threadPriority()
const
748 Q_D(
const QThreadPool);
749 QMutexLocker locker(&d->mutex);
750 return d->threadPriority;
754
755
756
757
758
759
760
761
762
763
764
765void QThreadPool::releaseThread()
768 QMutexLocker locker(&d->mutex);
769 --d->reservedThreads;
770 d->tryToStartMoreThreads();
774
775
776
777
778
779
780
781
782
783
784void QThreadPool::setServiceLevel(QThread::QualityOfService serviceLevel)
787 QMutexLocker locker(&d->mutex);
788 d->serviceLevel = serviceLevel;
792
793
794
795
796
797
798QThread::QualityOfService QThreadPool::serviceLevel()
const
800 Q_D(
const QThreadPool);
801 QMutexLocker locker(&d->mutex);
802 return d->serviceLevel;
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824void QThreadPool::startOnReservedThread(QRunnable *runnable)
827 return releaseThread();
830 QMutexLocker locker(&d->mutex);
831 Q_ASSERT(d->reservedThreads > 0);
832 --d->reservedThreads;
834 if (!d->tryStart(runnable)) {
837 d->enqueueTask(runnable, INT_MAX);
842
843
844
845
846
847
848
849
850
851
852
853
854
857
858
859
860
861
862
865
866
867
868
869
870
871bool QThreadPool::waitForDone(QDeadlineTimer deadline)
874 return d->waitForDone(deadline);
878
879
880
881
882
883
884
885
886void QThreadPool::clear()
893
894
895
896
897bool QThreadPool::contains(
const QThread *thread)
const
899 Q_D(
const QThreadPool);
900 const QThreadPoolThread *poolThread = qobject_cast<
const QThreadPoolThread *>(thread);
903 QMutexLocker locker(&d->mutex);
904 return d->allThreads.contains(
const_cast<QThreadPoolThread *>(poolThread));
909#include "moc_qthreadpool.cpp"
910#include "qthreadpool.moc"
bool areAllThreadsActive() const
bool tryStart(QRunnable *task)
void enqueueTask(QRunnable *task, int priority=0)
void tryToStartMoreThreads()
static void destroyGlobalThreadPool()
int activeThreadCount() const
int maxThreadCount() const
void startThread(QRunnable *runnable=nullptr)
void stealAndRunRunnable(QRunnable *runnable)
bool waitForDone(const QDeadlineTimer &timer)
bool tooManyThreadsActive() const
QThreadPoolPrivate * manager
void registerThreadInactive()
QWaitCondition runnableReady
bool comparePriority(int priority, const QueuePage *p)
QThreadPoolPrivate::QueuePage QueuePage