diff --git a/libs/image/kis_simple_update_queue.cpp b/libs/image/kis_simple_update_queue.cpp index 0eb1490122..7c1ade7e1e 100644 --- a/libs/image/kis_simple_update_queue.cpp +++ b/libs/image/kis_simple_update_queue.cpp @@ -1,424 +1,431 @@ /* * Copyright (c) 2010 Dmitry Kazakov * * This program is free software; you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation; either version 2 of the License, or * (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * * You should have received a copy of the GNU General Public License * along with this program; if not, write to the Free Software * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. */ #include "kis_simple_update_queue.h" #include #include #include "kis_image_config.h" #include "kis_full_refresh_walker.h" #include "kis_spontaneous_job.h" //#define ENABLE_DEBUG_JOIN //#define ENABLE_ACCUMULATOR #ifdef ENABLE_DEBUG_JOIN #define DEBUG_JOIN(baseRect, newRect, alpha) \ dbgKrita << "Two rects were joined:\t" \ << (baseRect) << "+" << (newRect) << "->" \ << ((baseRect) | (newRect)) << "(" << alpha << ")" #else #define DEBUG_JOIN(baseRect, newRect, alpha) #endif /* ENABLE_DEBUG_JOIN */ #ifdef ENABLE_ACCUMULATOR #define DECLARE_ACCUMULATOR() static qreal _baseAmount=0, _newAmount=0 #define ACCUMULATOR_ADD(baseAmount, newAmount) \ do {_baseAmount += baseAmount; _newAmount += newAmount;} while (0) #define ACCUMULATOR_DEBUG() \ dbgKrita << "Accumulated alpha:" << _newAmount / _baseAmount #else #define DECLARE_ACCUMULATOR() #define ACCUMULATOR_ADD(baseAmount, newAmount) #define ACCUMULATOR_DEBUG() #endif /* ENABLE_ACCUMULATOR */ KisSimpleUpdateQueue::KisSimpleUpdateQueue() : m_overrideLevelOfDetail(-1) { updateSettings(); } KisSimpleUpdateQueue::~KisSimpleUpdateQueue() { // QMutexLocker locker(&m_lock); QWriteLocker l(&m_rwLock); while (!m_spontaneousJobsList.isEmpty()) { delete m_spontaneousJobsList.takeLast(); } } void KisSimpleUpdateQueue::updateSettings() { // QMutexLocker locker(&m_lock); QWriteLocker l(&m_rwLock); KisImageConfig config(true); m_patchWidth = config.updatePatchWidth(); m_patchHeight = config.updatePatchHeight(); m_maxCollectAlpha = config.maxCollectAlpha(); m_maxMergeAlpha = config.maxMergeAlpha(); m_maxMergeCollectAlpha = config.maxMergeCollectAlpha(); } int KisSimpleUpdateQueue::overrideLevelOfDetail() const { return m_overrideLevelOfDetail; } void KisSimpleUpdateQueue::processQueue(KisUpdaterContext &updaterContext) { updaterContext.lock(); m_rwLock.lockForRead(); // QMutexLocker locker(&m_lock); while(updaterContext.hasSpareThread() && processOneJob(updaterContext)); // locker.unlock(); m_rwLock.unlock(); updaterContext.unlock(); } bool KisSimpleUpdateQueue::processOneJob(KisUpdaterContext &updaterContext) { // QMutexLocker locker(&m_lock); KisBaseRectsWalkerSP item; KisMutableWalkersListIterator iter(m_updatesList); + QVector walkers; bool jobAdded = false; int currentLevelOfDetail = updaterContext.currentLevelOfDetail(); while(iter.hasNext()) { item = iter.next(); if ((currentLevelOfDetail < 0 || currentLevelOfDetail == item->levelOfDetail()) && !item->checksumValid()) { m_rwLock.unlock(); m_rwLock.lockForWrite(); m_overrideLevelOfDetail = item->levelOfDetail(); item->recalculate(item->requestedRect()); m_overrideLevelOfDetail = -1; m_rwLock.unlock(); m_rwLock.lockForRead(); } if ((currentLevelOfDetail < 0 || currentLevelOfDetail == item->levelOfDetail()) && updaterContext.isJobAllowed(item)) { - updaterContext.addMergeJob(item); + walkers.append(item); +// updaterContext.addMergeJob(item); m_rwLock.unlock(); m_rwLock.lockForWrite(); iter.remove(); m_rwLock.unlock(); m_rwLock.lockForRead(); - jobAdded = true; - break; +// jobAdded = true; +// break; } } + if (!walkers.isEmpty()) { + updaterContext.addMergeJobs(walkers); + jobAdded = true; + } + if (jobAdded) return true; if (!m_spontaneousJobsList.isEmpty()) { /** * WARNING: Please note that this still doesn't guarantee that * the spontaneous jobs are exclusive, since updates and/or * strokes can be added after them. The only thing it * guarantees that two spontaneous jobs will not be executed * in parallel. * * Right now it works as it is. Probably will need to be fixed * in the future. */ qint32 numMergeJobs; qint32 numStrokeJobs; updaterContext.getJobsSnapshot(numMergeJobs, numStrokeJobs); if (!numMergeJobs && !numStrokeJobs) { KisSpontaneousJob *job = m_spontaneousJobsList.takeFirst(); updaterContext.addSpontaneousJob(job); jobAdded = true; } } return jobAdded; } void KisSimpleUpdateQueue::addUpdateJob(KisNodeSP node, const QVector &rects, const QRect& cropRect, int levelOfDetail) { addJob(node, rects, cropRect, levelOfDetail, KisBaseRectsWalker::UPDATE); } void KisSimpleUpdateQueue::addUpdateJob(KisNodeSP node, const QRect &rc, const QRect& cropRect, int levelOfDetail) { addJob(node, {rc}, cropRect, levelOfDetail, KisBaseRectsWalker::UPDATE); } void KisSimpleUpdateQueue::addUpdateNoFilthyJob(KisNodeSP node, const QRect& rc, const QRect& cropRect, int levelOfDetail) { addJob(node, {rc}, cropRect, levelOfDetail, KisBaseRectsWalker::UPDATE_NO_FILTHY); } void KisSimpleUpdateQueue::addFullRefreshJob(KisNodeSP node, const QRect& rc, const QRect& cropRect, int levelOfDetail) { addJob(node, {rc}, cropRect, levelOfDetail, KisBaseRectsWalker::FULL_REFRESH); } void KisSimpleUpdateQueue::addJob(KisNodeSP node, const QVector &rects, const QRect& cropRect, int levelOfDetail, KisBaseRectsWalker::UpdateType type) { QList walkers; Q_FOREACH (const QRect &rc, rects) { if (rc.isEmpty()) continue; KisBaseRectsWalkerSP walker; if(trySplitJob(node, rc, cropRect, levelOfDetail, type)) continue; if(tryMergeJob(node, rc, cropRect, levelOfDetail, type)) continue; if (type == KisBaseRectsWalker::UPDATE) { walker = new KisMergeWalker(cropRect, KisMergeWalker::DEFAULT); } else if (type == KisBaseRectsWalker::FULL_REFRESH) { walker = new KisFullRefreshWalker(cropRect); } else if (type == KisBaseRectsWalker::UPDATE_NO_FILTHY) { walker = new KisMergeWalker(cropRect, KisMergeWalker::NO_FILTHY); } /* else if(type == KisBaseRectsWalker::UNSUPPORTED) fatalKrita; */ walker->collectRects(node, rc); walkers.append(walker); } if (!walkers.isEmpty()) { // m_lock.lock(); QWriteLocker l(&m_rwLock); m_updatesList.append(walkers); // m_lock.unlock(); } } void KisSimpleUpdateQueue::addSpontaneousJob(KisSpontaneousJob *spontaneousJob) { // QMutexLocker locker(&m_lock); QWriteLocker l(&m_rwLock); KisSpontaneousJob *item; KisMutableSpontaneousJobsListIterator iter(m_spontaneousJobsList); iter.toBack(); while(iter.hasPrevious()) { item = iter.previous(); if (spontaneousJob->overrides(item)) { iter.remove(); delete item; } } m_spontaneousJobsList.append(spontaneousJob); } bool KisSimpleUpdateQueue::isEmpty() const { // QMutexLocker locker(&m_lock); QReadLocker l(&m_rwLock); return m_updatesList.isEmpty() && m_spontaneousJobsList.isEmpty(); } qint32 KisSimpleUpdateQueue::sizeMetric() const { // QMutexLocker locker(&m_lock); QReadLocker l(&m_rwLock); return m_updatesList.size() + m_spontaneousJobsList.size(); } bool KisSimpleUpdateQueue::trySplitJob(KisNodeSP node, const QRect& rc, const QRect& cropRect, int levelOfDetail, KisBaseRectsWalker::UpdateType type) { if(rc.width() <= m_patchWidth || rc.height() <= m_patchHeight) return false; // a bit of recursive splitting... qint32 firstCol = rc.x() / m_patchWidth; qint32 firstRow = rc.y() / m_patchHeight; qint32 lastCol = (rc.x() + rc.width()) / m_patchWidth; qint32 lastRow = (rc.y() + rc.height()) / m_patchHeight; QVector splitRects; for(qint32 i = firstRow; i <= lastRow; i++) { for(qint32 j = firstCol; j <= lastCol; j++) { QRect maxPatchRect(j * m_patchWidth, i * m_patchHeight, m_patchWidth, m_patchHeight); QRect patchRect = rc & maxPatchRect; splitRects.append(patchRect); } } KIS_SAFE_ASSERT_RECOVER_NOOP(!splitRects.isEmpty()); addJob(node, splitRects, cropRect, levelOfDetail, type); return true; } bool KisSimpleUpdateQueue::tryMergeJob(KisNodeSP node, const QRect& rc, const QRect& cropRect, int levelOfDetail, KisBaseRectsWalker::UpdateType type) { // QMutexLocker locker(&m_lock); QWriteLocker l(&m_rwLock); QRect baseRect = rc; KisBaseRectsWalkerSP goodCandidate; KisBaseRectsWalkerSP item; KisWalkersListIterator iter(m_updatesList); /** * We add new jobs to the tail of the list, * so it's more probable to find a good candidate here. */ iter.toBack(); while(iter.hasPrevious()) { item = iter.previous(); if(item->startNode() != node) continue; if(item->type() != type) continue; if(item->cropRect() != cropRect) continue; if(item->levelOfDetail() != levelOfDetail) continue; if(joinRects(baseRect, item->requestedRect(), m_maxMergeAlpha)) { goodCandidate = item; break; } } if(goodCandidate) collectJobs(goodCandidate, baseRect, m_maxMergeCollectAlpha); return (bool)goodCandidate; } void KisSimpleUpdateQueue::optimize() { // QMutexLocker locker(&m_lock); QWriteLocker l(&m_rwLock); if(m_updatesList.size() <= 1) { return; } KisBaseRectsWalkerSP baseWalker = m_updatesList.first(); QRect baseRect = baseWalker->requestedRect(); collectJobs(baseWalker, baseRect, m_maxCollectAlpha); } void KisSimpleUpdateQueue::collectJobs(KisBaseRectsWalkerSP &baseWalker, QRect baseRect, const qreal maxAlpha) { KisBaseRectsWalkerSP item; KisMutableWalkersListIterator iter(m_updatesList); while(iter.hasNext()) { item = iter.next(); if(item == baseWalker) continue; if(item->type() != baseWalker->type()) continue; if(item->startNode() != baseWalker->startNode()) continue; if(item->cropRect() != baseWalker->cropRect()) continue; if(item->levelOfDetail() != baseWalker->levelOfDetail()) continue; if(joinRects(baseRect, item->requestedRect(), maxAlpha)) { iter.remove(); } } if(baseWalker->requestedRect() != baseRect) { baseWalker->collectRects(baseWalker->startNode(), baseRect); } } bool KisSimpleUpdateQueue::joinRects(QRect& baseRect, const QRect& newRect, qreal maxAlpha) { QRect unitedRect = baseRect | newRect; if(unitedRect.width() > m_patchWidth || unitedRect.height() > m_patchHeight) return false; bool result = false; qint64 baseWork = baseRect.width() * baseRect.height() + newRect.width() * newRect.height(); qint64 newWork = unitedRect.width() * unitedRect.height(); qreal alpha = qreal(newWork) / baseWork; if(alpha < maxAlpha) { DEBUG_JOIN(baseRect, newRect, alpha); DECLARE_ACCUMULATOR(); ACCUMULATOR_ADD(baseWork, newWork); ACCUMULATOR_DEBUG(); baseRect = unitedRect; result = true; } return result; } KisWalkersList& KisTestableSimpleUpdateQueue::getWalkersList() { return m_updatesList; } KisSpontaneousJobsList& KisTestableSimpleUpdateQueue::getSpontaneousJobsList() { return m_spontaneousJobsList; } diff --git a/libs/image/kis_update_job_item.h b/libs/image/kis_update_job_item.h index 38c67a3b1f..422ff9875f 100644 --- a/libs/image/kis_update_job_item.h +++ b/libs/image/kis_update_job_item.h @@ -1,258 +1,287 @@ /* * Copyright (c) 2011 Dmitry Kazakov * * This program is free software; you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation; either version 2 of the License, or * (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * * You should have received a copy of the GNU General Public License * along with this program; if not, write to the Free Software * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. */ #ifndef __KIS_UPDATE_JOB_ITEM_H #define __KIS_UPDATE_JOB_ITEM_H #include #include #include #include "kis_stroke_job.h" #include "kis_spontaneous_job.h" #include "kis_base_rects_walker.h" #include "kis_async_merger.h" +#include "kis_debug.h" class KisUpdateJobItem : public QObject, public QRunnable { Q_OBJECT public: enum class Type : int { EMPTY = 0, WAITING, MERGE, STROKE, SPONTANEOUS }; public: KisUpdateJobItem(QReadWriteLock *exclusiveJobLock) : m_exclusiveJobLock(exclusiveJobLock), m_atomicType(Type::EMPTY), m_runnableJob(0) { setAutoDelete(false); KIS_SAFE_ASSERT_RECOVER_NOOP(m_atomicType.is_lock_free()); } ~KisUpdateJobItem() override { delete m_runnableJob; } void run() override { if (!isRunning()) return; /** * Here we break the idea of QThreadPool a bit. Ideally, we should split the * jobs into distinct QRunnable objects and pass all of them to QThreadPool. * That is a nice idea, but it doesn't work well when the jobs are small enough * and the number of available cores is high (>4 cores). It this case the * threads just tend to execute the job very quickly and go to sleep, which is * an expensive operation. * * To overcome this problem we try to bulk-process the jobs. In sigJobFinished() * signal (which is DirectConnection), the context may add the job to ourselves(!!!), * so we switch from "done" state into "running" again. */ while (1) { KIS_SAFE_ASSERT_RECOVER_RETURN(isRunning()); if(m_exclusive) { m_exclusiveJobLock->lockForWrite(); } else { m_exclusiveJobLock->lockForRead(); } if(m_atomicType == Type::MERGE) { runMergeJob(); } else { KIS_ASSERT(m_atomicType == Type::STROKE || m_atomicType == Type::SPONTANEOUS); m_runnableJob->run(); } setDone(); // emit sigDoSomeUsefulWork(); // may flip the current state from Waiting -> Running again emit sigJobFinished(); m_exclusiveJobLock->unlock(); // try to exit the loop. Please note, that no one can flip the state from // WAITING to EMPTY except ourselves! Type expectedValue = Type::WAITING; if (m_atomicType.compare_exchange_strong(expectedValue, Type::EMPTY)) { break; } } } inline void runMergeJob() { KIS_SAFE_ASSERT_RECOVER_RETURN(m_atomicType == Type::MERGE); - KIS_SAFE_ASSERT_RECOVER_RETURN(m_walker); +// KIS_SAFE_ASSERT_RECOVER_RETURN(m_walker); // dbgKrita << "Executing merge job" << m_walker->changeRect() // << "on thread" << QThread::currentThreadId(); - m_merger.startMerge(*m_walker); + QRect changeRect; + + for (auto walker : m_walkers) { + m_merger.startMerge(*walker); + changeRect |= walker->changeRect(); + } - QRect changeRect = m_walker->changeRect(); emit sigContinueUpdate(changeRect); } // return true if the thread should actually be started inline bool setWalker(KisBaseRectsWalkerSP walker) { KIS_ASSERT(m_atomicType <= Type::WAITING); m_accessRect = walker->accessRect(); m_changeRect = walker->changeRect(); m_walker = walker; m_exclusive = false; m_runnableJob = 0; const Type oldState = m_atomicType.exchange(Type::MERGE); return oldState == Type::EMPTY; } + inline bool setWalkers(QVector &walkers) { + KIS_ASSERT(m_atomicType <= Type::WAITING); + + m_accessRect = QRect(); + m_changeRect = QRect(); + + m_walkers.swap(walkers); + + m_exclusive = false; + m_runnableJob = 0; + + for (auto walker : m_walkers) { + m_accessRect |= walker->accessRect(); + m_changeRect |= walker->changeRect(); + } + + const Type oldState = m_atomicType.exchange(Type::MERGE); + return oldState == Type::EMPTY; + } + // return true if the thread should actually be started inline bool setStrokeJob(KisStrokeJob *strokeJob) { KIS_ASSERT(m_atomicType <= Type::WAITING); m_runnableJob = strokeJob; m_strokeJobSequentiality = strokeJob->sequentiality(); m_exclusive = strokeJob->isExclusive(); + m_walkers.clear(); m_walker = 0; m_accessRect = m_changeRect = QRect(); const Type oldState = m_atomicType.exchange(Type::STROKE); return oldState == Type::EMPTY; } // return true if the thread should actually be started inline bool setSpontaneousJob(KisSpontaneousJob *spontaneousJob) { KIS_ASSERT(m_atomicType <= Type::WAITING); m_runnableJob = spontaneousJob; m_exclusive = spontaneousJob->isExclusive(); + m_walkers.clear(); m_walker = 0; m_accessRect = m_changeRect = QRect(); const Type oldState = m_atomicType.exchange(Type::SPONTANEOUS); return oldState == Type::EMPTY; } inline void setDone() { + m_walkers.clear(); m_walker = 0; delete m_runnableJob; m_runnableJob = 0; m_atomicType = Type::WAITING; } inline bool isRunning() const { return m_atomicType >= Type::MERGE; } inline Type type() const { return m_atomicType; } inline const QRect& accessRect() const { return m_accessRect; } inline const QRect& changeRect() const { return m_changeRect; } inline KisStrokeJobData::Sequentiality strokeJobSequentiality() const { return m_strokeJobSequentiality; } Q_SIGNALS: void sigContinueUpdate(const QRect& rc); void sigDoSomeUsefulWork(); void sigJobFinished(); private: /** * Open walker and stroke job for the testing suite. * Please, do not use it in production code. */ friend class KisSimpleUpdateQueueTest; friend class KisStrokesQueueTest; friend class KisUpdateSchedulerTest; friend class KisTestableUpdaterContext; inline KisBaseRectsWalkerSP walker() const { return m_walker; } inline KisStrokeJob* strokeJob() const { KisStrokeJob *job = dynamic_cast(m_runnableJob); Q_ASSERT(job); return job; } inline void testingSetDone() { setDone(); } private: /** * \see KisUpdaterContext::m_exclusiveJobLock */ QReadWriteLock *m_exclusiveJobLock; bool m_exclusive; std::atomic m_atomicType; volatile KisStrokeJobData::Sequentiality m_strokeJobSequentiality; /** * Runnable jobs part * The job is owned by the context and deleted after completion */ KisRunnable *m_runnableJob; /** * Merge jobs part */ KisBaseRectsWalkerSP m_walker; + QVector m_walkers; KisAsyncMerger m_merger; /** * These rects cache actual values from the walker * to eliminate concurrent access to a walker structure */ QRect m_accessRect; QRect m_changeRect; }; #endif /* __KIS_UPDATE_JOB_ITEM_H */ diff --git a/libs/image/kis_updater_context.cpp b/libs/image/kis_updater_context.cpp index 222f95d68c..e841821c7f 100644 --- a/libs/image/kis_updater_context.cpp +++ b/libs/image/kis_updater_context.cpp @@ -1,322 +1,337 @@ /* * Copyright (c) 2010 Dmitry Kazakov * * This program is free software; you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation; either version 2 of the License, or * (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * * You should have received a copy of the GNU General Public License * along with this program; if not, write to the Free Software * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. */ #include "kis_updater_context.h" #include #include #include "kis_update_job_item.h" #include "kis_stroke_job.h" const int KisUpdaterContext::useIdealThreadCountTag = -1; KisUpdaterContext::KisUpdaterContext(qint32 threadCount, QObject *parent) : QObject(parent) { if(threadCount <= 0) { threadCount = QThread::idealThreadCount(); threadCount = threadCount > 0 ? threadCount : 1; } setThreadsLimit(threadCount); } KisUpdaterContext::~KisUpdaterContext() { m_threadPool.waitForDone(); for(qint32 i = 0; i < m_jobs.size(); i++) delete m_jobs[i]; } void KisUpdaterContext::getJobsSnapshot(qint32 &numMergeJobs, qint32 &numStrokeJobs) { numMergeJobs = 0; numStrokeJobs = 0; Q_FOREACH (const KisUpdateJobItem *item, m_jobs) { if(item->type() == KisUpdateJobItem::Type::MERGE || item->type() == KisUpdateJobItem::Type::SPONTANEOUS) { numMergeJobs++; } else if(item->type() == KisUpdateJobItem::Type::STROKE) { numStrokeJobs++; } } } KisUpdaterContextSnapshotEx KisUpdaterContext::getContextSnapshotEx() const { KisUpdaterContextSnapshotEx state = ContextEmpty; Q_FOREACH (const KisUpdateJobItem *item, m_jobs) { if (item->type() == KisUpdateJobItem::Type::MERGE || item->type() == KisUpdateJobItem::Type::SPONTANEOUS) { state |= HasMergeJob; } else if(item->type() == KisUpdateJobItem::Type::STROKE) { switch (item->strokeJobSequentiality()) { case KisStrokeJobData::SEQUENTIAL: state |= HasSequentialJob; break; case KisStrokeJobData::CONCURRENT: state |= HasConcurrentJob; break; case KisStrokeJobData::BARRIER: state |= HasBarrierJob; break; case KisStrokeJobData::UNIQUELY_CONCURRENT: state |= HasUniquelyConcurrentJob; break; } } } return state; } int KisUpdaterContext::currentLevelOfDetail() const { return m_lodCounter.readLod(); } bool KisUpdaterContext::hasSpareThread() { bool found = false; Q_FOREACH (const KisUpdateJobItem *item, m_jobs) { if(!item->isRunning()) { found = true; break; } } return found; } bool KisUpdaterContext::isJobAllowed(KisBaseRectsWalkerSP walker) { int lod = this->currentLevelOfDetail(); if (lod >= 0 && walker->levelOfDetail() != lod) return false; bool intersects = false; Q_FOREACH (const KisUpdateJobItem *item, m_jobs) { if(item->isRunning() && walkerIntersectsJob(walker, item)) { intersects = true; break; } } return !intersects; } /** * NOTE: In theory, isJobAllowed() and addMergeJob() should be merged into * one atomic method like `bool push()`, because this implementation * of KisUpdaterContext will not work in case of multiple * producers. But currently we have only one producer (one thread * in a time), that is guaranteed by the lock()/unlock() pair in * KisAbstractUpdateQueue::processQueue. */ void KisUpdaterContext::addMergeJob(KisBaseRectsWalkerSP walker) { m_lodCounter.addLod(walker->levelOfDetail()); qint32 jobIndex = findSpareThread(); Q_ASSERT(jobIndex >= 0); const bool shouldStartThread = m_jobs[jobIndex]->setWalker(walker); // it might happen that we call this function from within // the thread itself, right when it finished its work if (shouldStartThread) { m_threadPool.start(m_jobs[jobIndex]); } } +void KisUpdaterContext::addMergeJobs(QVector &walkers) +{ + m_lodCounter.addLod(walkers[0]->levelOfDetail()); + qint32 jobIndex = findSpareThread(); + Q_ASSERT(jobIndex >= 0); + + const bool shouldStartThread = m_jobs[jobIndex]->setWalkers(walkers); + + // it might happen that we call this function from within + // the thread itself, right when it finished its work + if (shouldStartThread) { + m_threadPool.start(m_jobs[jobIndex]); + } +} + /** * This variant is for use in a testing suite only */ void KisTestableUpdaterContext::addMergeJob(KisBaseRectsWalkerSP walker) { m_lodCounter.addLod(walker->levelOfDetail()); qint32 jobIndex = findSpareThread(); Q_ASSERT(jobIndex >= 0); const bool shouldStartThread = m_jobs[jobIndex]->setWalker(walker); // HINT: Not calling start() here Q_UNUSED(shouldStartThread); } void KisUpdaterContext::addStrokeJob(KisStrokeJob *strokeJob) { m_lodCounter.addLod(strokeJob->levelOfDetail()); qint32 jobIndex = findSpareThread(); Q_ASSERT(jobIndex >= 0); const bool shouldStartThread = m_jobs[jobIndex]->setStrokeJob(strokeJob); // it might happen that we call this function from within // the thread itself, right when it finished its work if (shouldStartThread) { m_threadPool.start(m_jobs[jobIndex]); } } /** * This variant is for use in a testing suite only */ void KisTestableUpdaterContext::addStrokeJob(KisStrokeJob *strokeJob) { m_lodCounter.addLod(strokeJob->levelOfDetail()); qint32 jobIndex = findSpareThread(); Q_ASSERT(jobIndex >= 0); const bool shouldStartThread = m_jobs[jobIndex]->setStrokeJob(strokeJob); // HINT: Not calling start() here Q_UNUSED(shouldStartThread); } void KisUpdaterContext::addSpontaneousJob(KisSpontaneousJob *spontaneousJob) { m_lodCounter.addLod(spontaneousJob->levelOfDetail()); qint32 jobIndex = findSpareThread(); Q_ASSERT(jobIndex >= 0); const bool shouldStartThread = m_jobs[jobIndex]->setSpontaneousJob(spontaneousJob); // it might happen that we call this function from within // the thread itself, right when it finished its work if (shouldStartThread) { m_threadPool.start(m_jobs[jobIndex]); } } /** * This variant is for use in a testing suite only */ void KisTestableUpdaterContext::addSpontaneousJob(KisSpontaneousJob *spontaneousJob) { m_lodCounter.addLod(spontaneousJob->levelOfDetail()); qint32 jobIndex = findSpareThread(); Q_ASSERT(jobIndex >= 0); const bool shouldStartThread = m_jobs[jobIndex]->setSpontaneousJob(spontaneousJob); // HINT: Not calling start() here Q_UNUSED(shouldStartThread); } void KisUpdaterContext::waitForDone() { m_threadPool.waitForDone(); } bool KisUpdaterContext::walkerIntersectsJob(KisBaseRectsWalkerSP walker, const KisUpdateJobItem* job) { return (walker->accessRect().intersects(job->changeRect())) || (job->accessRect().intersects(walker->changeRect())); } qint32 KisUpdaterContext::findSpareThread() { for(qint32 i=0; i < m_jobs.size(); i++) if(!m_jobs[i]->isRunning()) return i; return -1; } void KisUpdaterContext::slotJobFinished() { m_lodCounter.removeLod(); // Be careful. This slot can be called asynchronously without locks. emit sigSpareThreadAppeared(); } void KisUpdaterContext::lock() { m_lock.lock(); } void KisUpdaterContext::unlock() { m_lock.unlock(); } void KisUpdaterContext::setThreadsLimit(int value) { m_threadPool.setMaxThreadCount(value); for (int i = 0; i < m_jobs.size(); i++) { KIS_SAFE_ASSERT_RECOVER_RETURN(!m_jobs[i]->isRunning()); // don't delete the jobs until all of them are checked! } for (int i = 0; i < m_jobs.size(); i++) { delete m_jobs[i]; } m_jobs.resize(value); for(qint32 i = 0; i < m_jobs.size(); i++) { m_jobs[i] = new KisUpdateJobItem(&m_exclusiveJobLock); connect(m_jobs[i], SIGNAL(sigContinueUpdate(const QRect&)), SIGNAL(sigContinueUpdate(const QRect&)), Qt::DirectConnection); connect(m_jobs[i], SIGNAL(sigDoSomeUsefulWork()), SIGNAL(sigDoSomeUsefulWork()), Qt::DirectConnection); connect(m_jobs[i], SIGNAL(sigJobFinished()), SLOT(slotJobFinished()), Qt::DirectConnection); } } int KisUpdaterContext::threadsLimit() const { KIS_SAFE_ASSERT_RECOVER_NOOP(m_jobs.size() == m_threadPool.maxThreadCount()); return m_jobs.size(); } KisTestableUpdaterContext::KisTestableUpdaterContext(qint32 threadCount) : KisUpdaterContext(threadCount) { } KisTestableUpdaterContext::~KisTestableUpdaterContext() { clear(); } const QVector KisTestableUpdaterContext::getJobs() { return m_jobs; } void KisTestableUpdaterContext::clear() { Q_FOREACH (KisUpdateJobItem *item, m_jobs) { item->testingSetDone(); } m_lodCounter.testingClear(); } diff --git a/libs/image/kis_updater_context.h b/libs/image/kis_updater_context.h index 15421d6516..30257d9548 100644 --- a/libs/image/kis_updater_context.h +++ b/libs/image/kis_updater_context.h @@ -1,195 +1,196 @@ /* * Copyright (c) 2010 Dmitry Kazakov * * This program is free software; you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation; either version 2 of the License, or * (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * * You should have received a copy of the GNU General Public License * along with this program; if not, write to the Free Software * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. */ #ifndef __KIS_UPDATER_CONTEXT_H #define __KIS_UPDATER_CONTEXT_H #include #include #include #include #include "kis_base_rects_walker.h" #include "kis_async_merger.h" #include "kis_lock_free_lod_counter.h" #include "KisUpdaterContextSnapshotEx.h" class KisUpdateJobItem; class KisSpontaneousJob; class KisStrokeJob; class KRITAIMAGE_EXPORT KisUpdaterContext : public QObject { Q_OBJECT public: static const int useIdealThreadCountTag; public: KisUpdaterContext(qint32 threadCount = useIdealThreadCountTag, QObject *parent = 0); ~KisUpdaterContext() override; /** * Returns the number of currently running jobs of each type. * To use this information you should lock the context beforehand. * * \see lock() */ void getJobsSnapshot(qint32 &numMergeJobs, qint32 &numStrokeJobs); KisUpdaterContextSnapshotEx getContextSnapshotEx() const; /** * Returns the current level of detail of all the running jobs in the * context. If there are no jobs, returns -1. */ int currentLevelOfDetail() const; /** * Check whether there is a spare thread for running * one more job */ bool hasSpareThread(); /** * Checks whether the walker intersects with any * of currently executing walkers. If it does, * it is not allowed to go in. It should be called * with the lock held. * * \see lock() */ bool isJobAllowed(KisBaseRectsWalkerSP walker); /** * Registers the job and starts executing it. * The caller must ensure that the context is locked * with lock(), job is allowed with isWalkerAllowed() and * there is a spare thread for running it with hasSpareThread() * * \see lock() * \see isWalkerAllowed() * \see hasSpareThread() */ virtual void addMergeJob(KisBaseRectsWalkerSP walker); + void addMergeJobs(QVector &walkers); /** * Adds a stroke job to the context. The prerequisites are * the same as for addMergeJob() * \see addMergeJob() */ virtual void addStrokeJob(KisStrokeJob *strokeJob); /** * Adds a spontaneous job to the context. The prerequisites are * the same as for addMergeJob() * \see addMergeJob() */ virtual void addSpontaneousJob(KisSpontaneousJob *spontaneousJob); /** * Block execution of the caller until all the jobs are finished */ void waitForDone(); /** * Locks the context to guarantee an exclusive access * to the context */ void lock(); /** * Unlocks the context * * \see lock() */ void unlock(); /** * Set the number of threads available for this updater context * WARNING: one cannot change the number of threads if there is * at least one job running in the context! So before * calling this method make sure you do two things: * 1) barrierLock() the update scheduler * 2) lock() the context */ void setThreadsLimit(int value); /** * Return the number of available threads in the context. Make sure you * lock the context before calling this function! */ int threadsLimit() const; Q_SIGNALS: void sigContinueUpdate(const QRect& rc); void sigDoSomeUsefulWork(); void sigSpareThreadAppeared(); protected Q_SLOTS: void slotJobFinished(); protected: static bool walkerIntersectsJob(KisBaseRectsWalkerSP walker, const KisUpdateJobItem* job); qint32 findSpareThread(); protected: /** * The lock is shared by all the child update job items. * When an item wants to run a usual (non-exclusive) job, * it locks the lock for read access. When an exclusive * access is requested, it locks it for write */ QReadWriteLock m_exclusiveJobLock; QMutex m_lock; QVector m_jobs; QThreadPool m_threadPool; KisLockFreeLodCounter m_lodCounter; }; class KRITAIMAGE_EXPORT KisTestableUpdaterContext : public KisUpdaterContext { public: /** * Creates an explicit number of threads */ KisTestableUpdaterContext(qint32 threadCount); ~KisTestableUpdaterContext() override; /** * The only difference - it doesn't start execution * of the job */ void addMergeJob(KisBaseRectsWalkerSP walker) override; void addStrokeJob(KisStrokeJob *strokeJob) override; void addSpontaneousJob(KisSpontaneousJob *spontaneousJob) override; const QVector getJobs(); void clear(); }; #endif /* __KIS_UPDATER_CONTEXT_H */