src/corelib/concurrent/qtconcurrentreducekernel.h

   1 /****************************************************************************
   2 **
   3 ** Copyright (C) 2009 Nokia Corporation and/or its subsidiary(-ies).
   4 ** All rights reserved.
   5 ** Contact: Nokia Corporation (qt-info@nokia.com)
   6 **
   7 ** This file is part of the QtCore module of the Qt Toolkit.
   8 **
   9 ** $QT_BEGIN_LICENSE:LGPL$
  10 ** No Commercial Usage
  11 ** This file contains pre-release code and may not be distributed.
  12 ** You may use this file in accordance with the terms and conditions
  13 ** contained in the Technology Preview License Agreement accompanying
  14 ** this package.
  15 **
  16 ** GNU Lesser General Public License Usage
  17 ** Alternatively, this file may be used under the terms of the GNU Lesser
  18 ** General Public License version 2.1 as published by the Free Software
  19 ** Foundation and appearing in the file LICENSE.LGPL included in the
  20 ** packaging of this file.  Please review the following information to
  21 ** ensure the GNU Lesser General Public License version 2.1 requirements
  22 ** will be met: http://www.gnu.org/licenses/old-licenses/lgpl-2.1.html.
  23 **
  24 ** In addition, as a special exception, Nokia gives you certain additional
  25 ** rights.  These rights are described in the Nokia Qt LGPL Exception
  26 ** version 1.1, included in the file LGPL_EXCEPTION.txt in this package.
  27 **
  28 ** If you have questions regarding the use of this file, please contact
  29 ** Nokia at qt-info@nokia.com.
  30 **
  31 **
  32 **
  33 **
  34 **
  35 **
  36 **
  37 **
  38 ** $QT_END_LICENSE$
  39 **
  40 ****************************************************************************/
  41
  42 #ifndef QTCONCURRENT_REDUCEKERNEL_H
  43 #define QTCONCURRENT_REDUCEKERNEL_H
  44
  45 #include <QtCore/qglobal.h>
  46
  47 #ifndef QT_NO_CONCURRENT
  48
  49 #include <QtCore/qatomic.h>
  50 #include <QtCore/qlist.h>
  51 #include <QtCore/qmap.h>
  52 #include <QtCore/qmutex.h>
  53 #include <QtCore/qthread.h>
  54 #include <QtCore/qthreadpool.h>
  55 #include <QtCore/qvector.h>
  56
  57 QT_BEGIN_HEADER
  58 QT_BEGIN_NAMESPACE
  59
  60 QT_MODULE(Core)
  61
  62 namespace QtConcurrent {
  63
  64 #ifndef qdoc
  65
  66 /*
  67     The ReduceQueueStartLimit and ReduceQueueThrottleLimit constants
  68     limit the reduce queue size for MapReduce. When the number of
  69     reduce blocks in the queue exceeds ReduceQueueStartLimit,
  70     MapReduce won't start any new threads, and when it exceeds
  71     ReduceQueueThrottleLimit running threads will be stopped.
  72 */
  73 enum {
  74     ReduceQueueStartLimit = 20,
  75     ReduceQueueThrottleLimit = 30
  76 };
  77
  78 // IntermediateResults holds a block of intermediate results from a
  79 // map or filter functor. The begin/end offsets indicates the origin
  80 // and range of the block.
  81 template <typename T>
  82 class IntermediateResults
  83 {
  84 public:
  85     int begin, end;
  86     QVector<T> vector;
  87 };
  88
  89 #endif // qdoc
  90
  91 enum ReduceOption {
  92     UnorderedReduce = 0x1,
  93     OrderedReduce = 0x2,
  94     SequentialReduce = 0x4
  95     // ParallelReduce = 0x8
  96 };
  97 Q_DECLARE_FLAGS(ReduceOptions, ReduceOption)
  98 Q_DECLARE_OPERATORS_FOR_FLAGS(ReduceOptions)
  99
 100 #ifndef qdoc
 101
 102 // supports both ordered and out-of-order reduction
 103 template <typename ReduceFunctor, typename ReduceResultType, typename T>
 104 class ReduceKernel
 105 {
 106     typedef QMap<int, IntermediateResults<T> > ResultsMap;
 107
 108     const ReduceOptions reduceOptions;
 109
 110     QMutex mutex;
 111     int progress, resultsMapSize, threadCount;
 112     ResultsMap resultsMap;
 113
 114     bool canReduce(int begin) const
 115     {
 116         return (((reduceOptions & UnorderedReduce)
 117                  && progress == 0)
 118                 || ((reduceOptions & OrderedReduce)
 119                     && progress == begin));
 120     }
 121
 122     void reduceResult(ReduceFunctor &reduce,
 123                       ReduceResultType &r,
 124                       const IntermediateResults<T> &result)
 125     {
 126         for (int i = 0; i < result.vector.size(); ++i) {
 127             reduce(r, result.vector.at(i));
 128         }
 129     }
 130
 131     void reduceResults(ReduceFunctor &reduce,
 132                        ReduceResultType &r,
 133                        ResultsMap &map)
 134     {
 135         typename ResultsMap::iterator it = map.begin();
 136         while (it != map.end()) {
 137             reduceResult(reduce, r, it.value());
 138             ++it;
 139         }
 140     }
 141
 142 public:
 143     ReduceKernel(ReduceOptions _reduceOptions)
 144         : reduceOptions(_reduceOptions), progress(0), resultsMapSize(0),
 145           threadCount(QThreadPool::globalInstance()->maxThreadCount())
 146     { }
 147
 148     void runReduce(ReduceFunctor &reduce,
 149                    ReduceResultType &r,
 150                    const IntermediateResults<T> &result)
 151     {
 152         QMutexLocker locker(&mutex);
 153         if (!canReduce(result.begin)) {
 154             ++resultsMapSize;
 155             resultsMap.insert(result.begin, result);
 156             return;
 157         }
 158
 159         if (reduceOptions & UnorderedReduce) {
 160             // UnorderedReduce
 161             progress = -1;
 162
 163             // reduce this result
 164             locker.unlock();
 165             reduceResult(reduce, r, result);
 166             locker.relock();
 167
 168             // reduce all stored results as well
 169             while (!resultsMap.isEmpty()) {
 170                 ResultsMap resultsMapCopy = resultsMap;
 171                 resultsMap.clear();
 172
 173                 locker.unlock();
 174                 reduceResults(reduce, r, resultsMapCopy);
 175                 locker.relock();
 176
 177                 resultsMapSize -= resultsMapCopy.size();
 178             }
 179
 180             progress = 0;
 181         } else {
 182             // reduce this result
 183             locker.unlock();
 184             reduceResult(reduce, r, result);
 185             locker.relock();
 186
 187             // OrderedReduce
 188             progress += result.end - result.begin;
 189
 190             // reduce as many other results as possible
 191             typename ResultsMap::iterator it = resultsMap.begin();
 192             while (it != resultsMap.end()) {
 193                 if (it.value().begin != progress)
 194                     break;
 195
 196                 locker.unlock();
 197                 reduceResult(reduce, r, it.value());
 198                 locker.relock();
 199
 200                 --resultsMapSize;
 201                 progress += it.value().end - it.value().begin;
 202                 it = resultsMap.erase(it);
 203             }
 204         }
 205     }
 206
 207     // final reduction
 208     void finish(ReduceFunctor &reduce, ReduceResultType &r)
 209     {
 210         reduceResults(reduce, r, resultsMap);
 211     }
 212
 213     inline bool shouldThrottle()
 214     {
 215         return (resultsMapSize > (ReduceQueueThrottleLimit * threadCount));
 216     }
 217
 218     inline bool shouldStartThread()
 219     {
 220         return (resultsMapSize <= (ReduceQueueStartLimit * threadCount));
 221     }
 222 };
 223
 224 template <typename Sequence, typename Base, typename Functor1, typename Functor2>
 225 struct SequenceHolder2 : public Base
 226 {
 227     SequenceHolder2(const Sequence &_sequence,
 228                     Functor1 functor1,
 229                     Functor2 functor2,
 230                     ReduceOptions reduceOptions)
 231         : Base(_sequence.begin(), _sequence.end(), functor1, functor2, reduceOptions),
 232           sequence(_sequence)
 233     { }
 234
 235     Sequence sequence;
 236
 237     void finish()
 238     {
 239         Base::finish();
 240         // Clear the sequence to make sure all temporaries are destroyed
 241         // before finished is signaled.
 242         sequence = Sequence();
 243     }
 244 };
 245
 246 #endif //qdoc
 247
 248 } // namespace QtConcurrent
 249
 250 QT_END_NAMESPACE
 251 QT_END_HEADER
 252
 253 #endif // QT_NO_CONCURRENT
 254
 255 #endif