src/concurrent/qtconcurrentreducekernel.h

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