/**************************************************************************** ** ** Copyright (C) 2016 The Qt Company Ltd. ** Contact: https://www.qt.io/licensing/ ** ** This file is part of Qt Creator. ** ** Commercial License Usage ** Licensees holding valid commercial Qt licenses may use this file in ** accordance with the commercial license agreement provided with the ** Software or, alternatively, in accordance with the terms contained in ** a written agreement between you and The Qt Company. For licensing terms ** and conditions see https://www.qt.io/terms-conditions. For further ** information use the contact form at https://www.qt.io/contact-us. ** ** GNU General Public License Usage ** Alternatively, this file may be used under the terms of the GNU ** General Public License version 3 as published by the Free Software ** Foundation with exceptions as appearing in the file LICENSE.GPL3-EXCEPT ** included in the packaging of this file. Please review the following ** information to ensure the GNU General Public License requirements will ** be met: https://www.gnu.org/licenses/gpl-3.0.html. ** ****************************************************************************/ #pragma once #include "utils_global.h" #include "algorithm.h" #include "runextensions.h" #include namespace Utils { enum class MapReduceOption { Ordered, Unordered }; namespace Internal { class QTCREATOR_UTILS_EXPORT MapReduceObject : public QObject { Q_OBJECT }; template class MapReduceBase : public MapReduceObject { protected: static const int MAX_PROGRESS = 1000000; // either const or non-const reference wrapper for items from the iterator using ItemReferenceWrapper = std::reference_wrapper>; public: MapReduceBase(QFutureInterface futureInterface, ForwardIterator begin, ForwardIterator end, MapFunction &&map, State &state, ReduceFunction &&reduce, MapReduceOption option, QThreadPool *pool, int size) : m_futureInterface(futureInterface), m_iterator(begin), m_end(end), m_map(std::forward(map)), m_state(state), m_reduce(std::forward(reduce)), m_threadPool(pool), m_handleProgress(size >= 0), m_size(size), m_option(option) { if (!m_threadPool) m_threadPool = new QThreadPool(this); if (m_handleProgress) // progress is handled by us m_futureInterface.setProgressRange(0, MAX_PROGRESS); connect(&m_selfWatcher, &QFutureWatcher::canceled, this, &MapReduceBase::cancelAll); m_selfWatcher.setFuture(futureInterface.future()); } void exec() { // do not enter event loop for empty containers or if already canceled if (!m_futureInterface.isCanceled() && schedule()) m_loop.exec(); } protected: virtual void reduce(QFutureWatcher *watcher, int index) = 0; bool schedule() { bool didSchedule = false; while (m_iterator != m_end && m_mapWatcher.size() < std::max(m_threadPool->maxThreadCount(), 1)) { didSchedule = true; auto watcher = new QFutureWatcher(); connect(watcher, &QFutureWatcher::finished, this, [this, watcher]() { mapFinished(watcher); }); if (m_handleProgress) { connect(watcher, &QFutureWatcher::progressValueChanged, this, &MapReduceBase::updateProgress); connect(watcher, &QFutureWatcher::progressRangeChanged, this, &MapReduceBase::updateProgress); } m_mapWatcher.append(watcher); m_watcherIndex.append(m_currentIndex); ++m_currentIndex; watcher->setFuture(runAsync(m_threadPool, std::cref(m_map), ItemReferenceWrapper(*m_iterator))); ++m_iterator; } return didSchedule; } void mapFinished(QFutureWatcher *watcher) { int index = m_mapWatcher.indexOf(watcher); int watcherIndex = m_watcherIndex.at(index); m_mapWatcher.removeAt(index); // remove so we can schedule next one m_watcherIndex.removeAt(index); bool didSchedule = false; if (!m_futureInterface.isCanceled()) { // first schedule the next map... didSchedule = schedule(); ++m_successfullyFinishedMapCount; updateProgress(); // ...then reduce reduce(watcher, watcherIndex); } delete watcher; if (!didSchedule && m_mapWatcher.isEmpty()) m_loop.quit(); } void updateProgress() { if (!m_handleProgress) // cannot compute progress return; if (m_size == 0 || m_successfullyFinishedMapCount == m_size) { m_futureInterface.setProgressValue(MAX_PROGRESS); return; } if (!m_futureInterface.isProgressUpdateNeeded()) return; const double progressPerMap = MAX_PROGRESS / double(m_size); double progress = m_successfullyFinishedMapCount * progressPerMap; foreach (const QFutureWatcher *watcher, m_mapWatcher) { if (watcher->progressMinimum() != watcher->progressMaximum()) { const double range = watcher->progressMaximum() - watcher->progressMinimum(); progress += (watcher->progressValue() - watcher->progressMinimum()) / range * progressPerMap; } } m_futureInterface.setProgressValue(int(progress)); } void cancelAll() { foreach (QFutureWatcher *watcher, m_mapWatcher) watcher->cancel(); } QFutureWatcher m_selfWatcher; QFutureInterface m_futureInterface; ForwardIterator m_iterator; const ForwardIterator m_end; MapFunction m_map; State &m_state; ReduceFunction m_reduce; QEventLoop m_loop; QThreadPool *m_threadPool; // for reusing threads QList *> m_mapWatcher; QList m_watcherIndex; int m_currentIndex = 0; const bool m_handleProgress; const int m_size; int m_successfullyFinishedMapCount = 0; MapReduceOption m_option; }; // non-void result of map function. template class MapReduce : public MapReduceBase { using BaseType = MapReduceBase; public: MapReduce(QFutureInterface futureInterface, ForwardIterator begin, ForwardIterator end, MapFunction &&map, State &state, ReduceFunction &&reduce, MapReduceOption option, QThreadPool *pool, int size) : BaseType(futureInterface, begin, end, std::forward(map), state, std::forward(reduce), option, pool, size) { } protected: void reduce(QFutureWatcher *watcher, int index) override { if (BaseType::m_option == MapReduceOption::Unordered) { reduceOne(watcher->future().results()); } else { if (m_nextIndex == index) { // handle this result and all directly following reduceOne(watcher->future().results()); ++m_nextIndex; while (!m_pendingResults.isEmpty() && m_pendingResults.firstKey() == m_nextIndex) { reduceOne(m_pendingResults.take(m_nextIndex)); ++m_nextIndex; } } else { // add result to pending results m_pendingResults.insert(index, watcher->future().results()); } } } private: void reduceOne(const QList &results) { const int resultCount = results.size(); for (int i = 0; i < resultCount; ++i) { Internal::runAsyncImpl(BaseType::m_futureInterface, BaseType::m_reduce, BaseType::m_state, results.at(i)); } } QMap> m_pendingResults; int m_nextIndex = 0; }; // specialization for void result of map function. Reducing is a no-op. template class MapReduce : public MapReduceBase { using BaseType = MapReduceBase; public: MapReduce(QFutureInterface futureInterface, ForwardIterator begin, ForwardIterator end, MapFunction &&map, State &state, ReduceFunction &&reduce, MapReduceOption option, QThreadPool *pool, int size) : BaseType(futureInterface, begin, end, std::forward(map), state, std::forward(reduce), option, pool, size) { } protected: void reduce(QFutureWatcher *, int) override { } }; template functionResult_t callWithMaybeFutureInterfaceDispatch(std::false_type, QFutureInterface &, Function &&function, Args&&... args) { return function(std::forward(args)...); } template functionResult_t callWithMaybeFutureInterfaceDispatch(std::true_type, QFutureInterface &futureInterface, Function &&function, Args&&... args) { return function(futureInterface, std::forward(args)...); } template functionResult_t callWithMaybeFutureInterface(QFutureInterface &futureInterface, Function &&function, Args&&... args) { return callWithMaybeFutureInterfaceDispatch( functionTakesArgument&>(), futureInterface, std::forward(function), std::forward(args)...); } template void blockingIteratorMapReduce(QFutureInterface &futureInterface, ForwardIterator begin, ForwardIterator end, InitFunction &&init, MapFunction &&map, ReduceFunction &&reduce, CleanUpFunction &&cleanup, MapReduceOption option, QThreadPool *pool, int size) { auto state = callWithMaybeFutureInterface (futureInterface, std::forward(init)); MapReduce::type, MapFunction, decltype(state), ReduceResult, ReduceFunction> mr(futureInterface, begin, end, std::forward(map), state, std::forward(reduce), option, pool, size); mr.exec(); callWithMaybeFutureInterface&> (futureInterface, std::forward(cleanup), state); } template void blockingContainerMapReduce(QFutureInterface &futureInterface, Container &&container, InitFunction &&init, MapFunction &&map, ReduceFunction &&reduce, CleanUpFunction &&cleanup, MapReduceOption option, QThreadPool *pool) { blockingIteratorMapReduce(futureInterface, std::begin(container), std::end(container), std::forward(init), std::forward(map), std::forward(reduce), std::forward(cleanup), option, pool, static_cast(container.size())); } template void blockingContainerRefMapReduce(QFutureInterface &futureInterface, std::reference_wrapper containerWrapper, InitFunction &&init, MapFunction &&map, ReduceFunction &&reduce, CleanUpFunction &&cleanup, MapReduceOption option, QThreadPool *pool) { blockingContainerMapReduce(futureInterface, containerWrapper.get(), std::forward(init), std::forward(map), std::forward(reduce), std::forward(cleanup), option, pool); } template static void *dummyInit() { return nullptr; } // copies or moves state to member, and then moves it to the result of the call operator template struct StateWrapper { using StateResult = std::decay_t; // State is const& or & for lvalues StateWrapper(State &&state) : m_state(std::forward(state)) { } StateResult operator()() { return std::move(m_state); // invalidates m_state } StateResult m_state; }; // copies or moves reduce function to member, calls the reduce function with state and mapped value template struct ReduceWrapper { using Reduce = std::decay_t; ReduceWrapper(ReduceFunction &&reduce) : m_reduce(std::forward(reduce)) { } void operator()(QFutureInterface &, StateResult &state, const MapResult &mapResult) { m_reduce(state, mapResult); } Reduce m_reduce; }; template struct DummyReduce { MapResult operator()(void *, const MapResult &result) const { return result; } }; template <> struct DummyReduce { void operator()() const { } // needed for resultType with MSVC2013 }; template static void dummyCleanup(void *) { } template static void cleanupReportingState(QFutureInterface &fi, StateResult &state) { fi.reportResult(state); } } // Internal template ::type> QFuture mapReduce(ForwardIterator begin, ForwardIterator end, InitFunction &&init, MapFunction &&map, ReduceFunction &&reduce, CleanUpFunction &&cleanup, MapReduceOption option = MapReduceOption::Unordered, QThreadPool *pool = nullptr, QThread::Priority priority = QThread::InheritPriority, int size = -1) { return runAsync(priority, Internal::blockingIteratorMapReduce< ForwardIterator, std::decay_t, std::decay_t, std::decay_t, std::decay_t, std::decay_t>, begin, end, std::forward(init), std::forward(map), std::forward(reduce), std::forward(cleanup), option, pool, size); } /*! Calls the map function on all items in \a container in parallel through Utils::runAsync. The reduce function is called in the mapReduce thread with each of the reported results from the map function, in arbitrary order, but never in parallel. It gets passed a reference to a user defined state object, and a result from the map function. If it takes a QFutureInterface reference as its first argument, it can report results for the mapReduce operation through that. Otherwise, any values returned by the reduce function are reported as results of the mapReduce operation. The init function is called in the mapReduce thread before the actual mapping starts, and must return the initial state object for the reduce function. It gets the QFutureInterface of the mapReduce operation passed as an argument. The cleanup function is called in the mapReduce thread after all map and reduce calls have finished, with the QFutureInterface of the mapReduce operation and the final state object as arguments, and can be used to clean up any resources, or report a final result of the mapReduce. Container StateType InitFunction(QFutureInterface&) or StateType InitFunction() void MapFunction(QFutureInterface&, const ItemType&) or MapResultType MapFunction(const ItempType&) void ReduceFunction(QFutureInterface&, StateType&, const MapResultType&) or ReduceResultType ReduceFunction(StateType&, const MapResultType&) void CleanUpFunction(QFutureInterface&, StateType&) or void CleanUpFunction(StateType&) Notes: \list \li Container can be a move-only type or a temporary. If it is a lvalue reference, it will be copied to the mapReduce thread. You can avoid that by using the version that takes iterators, or by using std::ref/cref to pass a reference_wrapper. \li ItemType can be a move-only type, if the map function takes (const) references to ItemType. \li StateType can be a move-only type. \li The init, map, reduce and cleanup functions can be move-only types and are moved to the mapReduce thread if they are rvalues. \endlist */ template ::type> QFuture mapReduce(Container &&container, InitFunction &&init, MapFunction &&map, ReduceFunction &&reduce, CleanUpFunction &&cleanup, MapReduceOption option = MapReduceOption::Unordered, QThreadPool *pool = nullptr, QThread::Priority priority = QThread::InheritPriority) { return runAsync(priority, Internal::blockingContainerMapReduce< std::decay_t, std::decay_t, std::decay_t, std::decay_t, std::decay_t, std::decay_t>, std::forward(container), std::forward(init), std::forward(map), std::forward(reduce), std::forward(cleanup), option, pool); } template ::type> QFuture mapReduce(std::reference_wrapper containerWrapper, InitFunction &&init, MapFunction &&map, ReduceFunction &&reduce, CleanUpFunction &&cleanup, MapReduceOption option = MapReduceOption::Unordered, QThreadPool *pool = nullptr, QThread::Priority priority = QThread::InheritPriority) { return runAsync(priority, Internal::blockingContainerRefMapReduce< Container, std::decay_t, std::decay_t, std::decay_t, std::decay_t, std::decay_t>, containerWrapper, std::forward(init), std::forward(map), std::forward(reduce), std::forward(cleanup), option, pool); } template , // State = T& or const T& for lvalues, so decay that away typename MapResult = typename Internal::resultType::type> QFuture mapReduce(ForwardIterator begin, ForwardIterator end, MapFunction &&map, State &&initialState, ReduceFunction &&reduce, MapReduceOption option = MapReduceOption::Unordered, QThreadPool *pool = nullptr, QThread::Priority priority = QThread::InheritPriority, int size = -1) { return mapReduce(begin, end, Internal::StateWrapper(std::forward(initialState)), std::forward(map), Internal::ReduceWrapper(std::forward(reduce)), &Internal::cleanupReportingState, option, pool, priority, size); } template , // State = T& or const T& for lvalues, so decay that away typename MapResult = typename Internal::resultType::type> QFuture mapReduce(Container &&container, MapFunction &&map, State &&initialState, ReduceFunction &&reduce, MapReduceOption option = MapReduceOption::Unordered, QThreadPool *pool = nullptr, QThread::Priority priority = QThread::InheritPriority) { return mapReduce(std::forward(container), Internal::StateWrapper(std::forward(initialState)), std::forward(map), Internal::ReduceWrapper(std::forward(reduce)), &Internal::cleanupReportingState, option, pool, priority); } template , // State = T& or const T& for lvalues, so decay that away typename MapResult = typename Internal::resultType::type> Q_REQUIRED_RESULT StateResult mappedReduced(ForwardIterator begin, ForwardIterator end, MapFunction &&map, State &&initialState, ReduceFunction &&reduce, MapReduceOption option = MapReduceOption::Unordered, QThreadPool *pool = nullptr, QThread::Priority priority = QThread::InheritPriority, int size = -1) { return mapReduce(begin, end, std::forward(map), std::forward(initialState), std::forward(reduce), option, pool, priority, size).result(); } template , // State = T& or const T& for lvalues, so decay that away typename MapResult = typename Internal::resultType::type> Q_REQUIRED_RESULT StateResult mappedReduced(Container &&container, MapFunction &&map, State &&initialState, ReduceFunction &&reduce, MapReduceOption option = MapReduceOption::Unordered, QThreadPool *pool = nullptr, QThread::Priority priority = QThread::InheritPriority) { return mapReduce(std::forward(container), std::forward(map), std::forward(initialState), std::forward(reduce), option, pool, priority).result(); } template ::type> QFuture map(ForwardIterator begin, ForwardIterator end, MapFunction &&map, MapReduceOption option = MapReduceOption::Ordered, QThreadPool *pool = nullptr, QThread::Priority priority = QThread::InheritPriority, int size = -1) { return mapReduce(begin, end, &Internal::dummyInit, std::forward(map), Internal::DummyReduce(), &Internal::dummyCleanup, option, pool, priority, size); } template ::type> QFuture map(Container &&container, MapFunction &&map, MapReduceOption option = MapReduceOption::Ordered, QThreadPool *pool = nullptr, QThread::Priority priority = QThread::InheritPriority) { return mapReduce(std::forward(container), Internal::dummyInit, std::forward(map), Internal::DummyReduce(), Internal::dummyCleanup, option, pool, priority); } template class ResultContainer, typename ForwardIterator, typename MapFunction, typename MapResult = typename Internal::resultType::type> Q_REQUIRED_RESULT ResultContainer mapped(ForwardIterator begin, ForwardIterator end, MapFunction &&mapFun, MapReduceOption option = MapReduceOption::Ordered, QThreadPool *pool = nullptr, QThread::Priority priority = QThread::InheritPriority, int size = -1) { return Utils::transform(map(begin, end, std::forward(mapFun), option, pool, priority, size).results(), [](const MapResult &r) { return r; }); } template class ResultContainer, typename Container, typename MapFunction, typename MapResult = typename Internal::resultType::type> Q_REQUIRED_RESULT ResultContainer mapped(Container &&container, MapFunction &&mapFun, MapReduceOption option = MapReduceOption::Ordered, QThreadPool *pool = nullptr, QThread::Priority priority = QThread::InheritPriority) { return Utils::transform(map(container, std::forward(mapFun), option, pool, priority).results(), [](const MapResult &r) { return r; }); } } // Utils