qtconcurrentiteratekernel.h
上传用户:detong
上传日期:2022-06-22
资源大小:20675k
文件大小:9k
源码类别:

系统编程

开发平台:

Unix_Linux

  1. /****************************************************************************
  2. **
  3. ** Copyright (C) 2008 Nokia Corporation and/or its subsidiary(-ies).
  4. ** Contact: Qt Software Information (qt-info@nokia.com)
  5. **
  6. ** This file is part of the QtCore module of the Qt Toolkit.
  7. **
  8. ** Commercial Usage
  9. ** Licensees holding valid Qt Commercial licenses may use this file in
  10. ** accordance with the Qt Commercial License Agreement provided with the
  11. ** Software or, alternatively, in accordance with the terms contained in
  12. ** a written agreement between you and Nokia.
  13. **
  14. **
  15. ** GNU General Public License Usage
  16. ** Alternatively, this file may be used under the terms of the GNU
  17. ** General Public License versions 2.0 or 3.0 as published by the Free
  18. ** Software Foundation and appearing in the file LICENSE.GPL included in
  19. ** the packaging of this file.  Please review the following information
  20. ** to ensure GNU General Public Licensing requirements will be met:
  21. ** http://www.fsf.org/licensing/licenses/info/GPLv2.html and
  22. ** http://www.gnu.org/copyleft/gpl.html.  In addition, as a special
  23. ** exception, Nokia gives you certain additional rights. These rights
  24. ** are described in the Nokia Qt GPL Exception version 1.3, included in
  25. ** the file GPL_EXCEPTION.txt in this package.
  26. **
  27. ** Qt for Windows(R) Licensees
  28. ** As a special exception, Nokia, as the sole copyright holder for Qt
  29. ** Designer, grants users of the Qt/Eclipse Integration plug-in the
  30. ** right for the Qt/Eclipse Integration to link to functionality
  31. ** provided by Qt Designer and its related libraries.
  32. **
  33. ** If you are unsure which license is appropriate for your use, please
  34. ** contact the sales department at qt-sales@nokia.com.
  35. **
  36. ****************************************************************************/
  37. #ifndef QTCONCURRENT_ITERATEKERNEL_H
  38. #define QTCONCURRENT_ITERATEKERNEL_H
  39. #include <QtCore/qglobal.h>
  40. #ifndef QT_NO_CONCURRENT
  41. #include <QtCore/qatomic.h>
  42. #include <QtCore/qtconcurrentmedian.h>
  43. #include <QtCore/qtconcurrentthreadengine.h>
  44. #include <iterator>
  45. QT_BEGIN_HEADER
  46. QT_BEGIN_NAMESPACE
  47. QT_MODULE(Core)
  48. #ifndef qdoc
  49. namespace QtConcurrent {
  50. #ifndef QT_NO_STL
  51.     using std::advance;
  52. #else
  53.     template <typename It, typename T>
  54.     void advance(It &it, T value)
  55.     {
  56.         it+=value;
  57.     }
  58. #endif
  59. /*
  60.     The BlockSizeManager class manages how many iterations a thread should
  61.     reserve and process at a time. This is done by measuring the time spent
  62.     in the user code versus the control part code, and then increasing
  63.     the block size if the ratio between them is to small. The block size
  64.     management is done on the basis of the median of several timing measuremens,
  65.     and it is done induvidualy for each thread.
  66. */
  67. class Q_CORE_EXPORT BlockSizeManager
  68. {
  69. public:
  70.     BlockSizeManager(int iterationCount);
  71.     void timeBeforeUser();
  72.     void timeAfterUser();
  73.     int blockSize();
  74. private:
  75.     inline bool blockSizeMaxed()
  76.     {
  77.         return (m_blockSize >= maxBlockSize);
  78.     }
  79.     const int maxBlockSize;
  80.     qint64 beforeUser;
  81.     qint64 afterUser;
  82.     Median<double> controlPartElapsed;
  83.     Median<double> userPartElapsed;
  84.     int m_blockSize;
  85. };
  86. template <typename T>
  87. class ResultReporter
  88. {
  89. public:
  90.     ResultReporter(ThreadEngine<T> *_threadEngine)
  91.     :threadEngine(_threadEngine)
  92.     {
  93.     }
  94.     void reserveSpace(int resultCount)
  95.     {
  96.         currentResultCount = resultCount;
  97.         vector.resize(qMax(resultCount, vector.count()));
  98.     }
  99.     void reportResults(int begin)
  100.     {
  101.         const int useVectorThreshold = 4; // Tunable parameter.
  102.         if (currentResultCount > useVectorThreshold) {
  103.             vector.resize(currentResultCount);
  104.             threadEngine->reportResults(vector, begin);
  105.         } else {
  106.             for (int i = 0; i < currentResultCount; ++i)
  107.                 threadEngine->reportResult(&vector.at(i), begin + i);
  108.         }
  109.     }
  110.     inline T * getPointer()
  111.     {
  112.         return vector.data();
  113.     }
  114.     int currentResultCount;
  115.     ThreadEngine<T> *threadEngine;
  116.     QVector<T> vector;
  117. };
  118. template <>
  119. class ResultReporter<void>
  120. {
  121. public:
  122.     inline ResultReporter(ThreadEngine<void> *) { }
  123.     inline void reserveSpace(int) { };
  124.     inline void reportResults(int) { };
  125.     inline void * getPointer() { return 0; }
  126. };
  127. inline bool selectIteration(std::bidirectional_iterator_tag)
  128. {
  129.     return false; // while
  130. }
  131. inline bool selectIteration(std::forward_iterator_tag)
  132. {
  133.     return false; // while
  134. }
  135. inline bool selectIteration(std::random_access_iterator_tag)
  136. {
  137.     return true; // for
  138. }
  139. template <typename Iterator, typename T>
  140. class IterateKernel : public ThreadEngine<T>
  141. {
  142. public:
  143.     typedef T ResultType;
  144.     IterateKernel(Iterator _begin, Iterator _end)
  145. #ifndef QT_NO_PARTIAL_TEMPLATE_SPECIALIZATION
  146.         : begin(_begin), end(_end), current(_begin), currentIndex(0),
  147.            forIteration(selectIteration(typename std::iterator_traits<Iterator>::iterator_category())), progressReportingEnabled(true)
  148. #else
  149.         : begin(_begin), end(_end), currentIndex(0),
  150.           forIteration(selectIteration(std::iterator_category(_begin))), progressReportingEnabled(true)
  151. #endif
  152.     {
  153.         iterationCount =  forIteration ? std::distance(_begin, _end) : 0;
  154.     }
  155.     virtual ~IterateKernel() { }
  156.     virtual bool runIteration(Iterator it, int index , T *result)
  157.         { Q_UNUSED(it); Q_UNUSED(index); Q_UNUSED(result); return false; }
  158.     virtual bool runIterations(Iterator _begin, int beginIndex, int endIndex, T *results)
  159.         { Q_UNUSED(_begin); Q_UNUSED(beginIndex); Q_UNUSED(endIndex); Q_UNUSED(results); return false; }
  160.     void start()
  161.     {
  162.         progressReportingEnabled = this->isProgressReportingEnabled();
  163.         if (progressReportingEnabled && iterationCount > 0)
  164.             this->setProgressRange(0, iterationCount);
  165.     }
  166.     bool shouldStartThread()
  167.     {
  168.         if (forIteration)
  169.             return (currentIndex < iterationCount) && !this->shouldThrottleThread();
  170.         else // whileIteration
  171.             return (iteratorThreads == 0);
  172.     }
  173.     ThreadFunctionResult threadFunction()
  174.     {
  175.         if (forIteration)
  176.             return this->forThreadFunction();
  177.         else // whileIteration
  178.             return this->whileThreadFunction();
  179.     }
  180.     ThreadFunctionResult forThreadFunction()
  181.     {
  182.         BlockSizeManager blockSizeManager(iterationCount);
  183.         ResultReporter<T> resultReporter(this);
  184.         for(;;) {
  185.             if (this->isCanceled())
  186.                 break;
  187.             const int currentBlockSize = blockSizeManager.blockSize();
  188.             if (currentIndex >= iterationCount)
  189.                 break;
  190.             // Atomically reserve a block of iterationCount for this thread.
  191.             const int beginIndex = currentIndex.fetchAndAddRelease(currentBlockSize);
  192.             const int endIndex = qMin(beginIndex + currentBlockSize, iterationCount);
  193.             if (beginIndex >= endIndex) {
  194.                 // No more work
  195.                 break;
  196.             }
  197.             const int finalBlockSize = endIndex - beginIndex; // block size adjusted for possible end-of-range
  198.             resultReporter.reserveSpace(finalBlockSize);
  199.             // Call user code with the current iteration range.
  200.             blockSizeManager.timeBeforeUser();
  201.             const bool resultsAvailable = this->runIterations(begin, beginIndex, endIndex, resultReporter.getPointer());
  202.             blockSizeManager.timeAfterUser();
  203.             if (resultsAvailable)
  204.                 resultReporter.reportResults(beginIndex);
  205.             // Report progress if progress reporting enabled.
  206.             if (progressReportingEnabled) {
  207.                 completed.fetchAndAddAcquire(finalBlockSize);
  208.                 this->setProgressValue(this->completed);
  209.             }
  210.             if (this->shouldThrottleThread())
  211.                 return ThrottleThread;
  212.             if (shouldStartThread())
  213.                 this->startThread();
  214.         }
  215.         return ThreadFinished;
  216.     }
  217.     ThreadFunctionResult whileThreadFunction()
  218.     {
  219.         if (iteratorThreads.testAndSetAcquire(0, 1) == false)
  220.             return ThreadFinished;
  221.         ResultReporter<T> resultReporter(this);
  222.         resultReporter.reserveSpace(1);
  223.         while (current != end) {
  224.             // The following two lines breaks support for input iterators according to
  225.             // the sgi docs: dereferencing prev after calling ++current is not allowed
  226.             // on input iterators. (prev is dereferenced inside user.runIteration())
  227.             Iterator prev = current;
  228.             ++current;
  229.             int index = currentIndex.fetchAndAddRelaxed(1);
  230.             iteratorThreads.testAndSetRelease(1, 0);
  231.             if (shouldStartThread())
  232.                 this->startThread();
  233.             const bool resultAavailable = this->runIteration(prev, index, resultReporter.getPointer());
  234.             if (resultAavailable)
  235.                 resultReporter.reportResults(index);
  236.             if (this->shouldThrottleThread())
  237.                 return ThrottleThread;
  238.             if (iteratorThreads.testAndSetAcquire(0, 1) == false)
  239.                 return ThreadFinished;
  240.         }
  241.         return ThreadFinished;
  242.     }
  243. public:
  244.     const Iterator begin;
  245.     const Iterator end;
  246.     Iterator current;
  247.     QAtomicInt currentIndex;
  248.     bool forIteration;
  249.     QAtomicInt iteratorThreads;
  250.     int iterationCount;
  251.     bool progressReportingEnabled;
  252.     QAtomicInt completed;
  253. };
  254. } // namespace QtConcurrent
  255. #endif //qdoc
  256. QT_END_NAMESPACE
  257. QT_END_HEADER
  258. #endif // QT_NO_CONCURRENT
  259. #endif