File indexing completed on 2024-04-28 04:01:22

 /* -*- C++ -*-
     This file is part of ThreadWeaver.
 
     SPDX-FileCopyrightText: 2004-2013 Mirko Boehm <mirko@kde.org>
 
     SPDX-License-Identifier: LGPL-2.0-or-later
 */
 
 #include "collection_p.h"
 #include "collection.h"
 #include "debuggingaids.h"
 #include "managedjobpointer.h"
 #include "queueapi.h"
 
 namespace ThreadWeaver
 {
 namespace Private
 {
 Collection_Private::Collection_Private()
     : api(nullptr)
     , jobCounter(0)
     , selfIsExecuting(false)
 {
 }
 
 Collection_Private::~Collection_Private()
 {
 }
 
 void Collection_Private::finalCleanup(Collection *collection)
 {
     Q_ASSERT(!self.isNull());
     Q_ASSERT(!mutex.tryLock());
     if (collection->shouldAbort()) {
         collection->setStatus(Job::Status_Aborted);
     } else if (collection->status() < Job::Status_Success) {
         collection->setStatus(Job::Status_Success);
     } else {
         // At this point we either should have been running
         // in which case above would mark Success
         // or otherwise we already should be in Failed or Aborted state
         Q_ASSERT(collection->status() == Job::Status_Failed || collection->status() == Job::Status_Aborted);
     }
     freeQueuePolicyResources(self);
     api = nullptr;
 }
 
 void Collection_Private::enqueueElements()
 {
     Q_ASSERT(!mutex.tryLock());
     prepareToEnqueueElements();
     jobCounter.fetchAndStoreOrdered(elements.count() + 1); // including self
     api->enqueue(elements);
 }
 
 void Collection_Private::elementStarted(Collection *, JobPointer job, Thread *)
 {
     QMutexLocker l(&mutex);
     Q_UNUSED(l);
     Q_UNUSED(job) // except in Q_ASSERT
     Q_ASSERT(!self.isNull());
     if (jobsStarted.fetchAndAddOrdered(1) == 0) {
         // emit started() signal on beginning of first job execution
         selfExecuteWrapper.callBegin();
     }
 }
 
 namespace
 {
 struct MutexUnlocker {
     QMutexLocker<QMutex> *locker;
 
     MutexUnlocker(QMutexLocker<QMutex> *l)
         : locker(l)
     {
         locker->unlock();
     }
     ~MutexUnlocker()
     {
         locker->relock();
     }
     MutexUnlocker(const MutexUnlocker &) = delete;
     MutexUnlocker &operator=(const MutexUnlocker &) = delete;
 };
 }
 
 void Collection_Private::elementFinished(Collection *collection, JobPointer job, Thread *thread)
 {
     JobPointer saveYourSelf = self;
     Q_UNUSED(saveYourSelf);
     QMutexLocker l(&mutex);
     Q_UNUSED(l);
     Q_ASSERT(!self.isNull());
     Q_UNUSED(job) // except in Q_ASSERT
     if (selfIsExecuting) {
         // the element that is finished is the collection itself
         // the collection is always executed first
         // No need to queue elements if we were aborted
         if (!collection->shouldAbort()) {
             // queue the collection elements:
             enqueueElements();
         }
         selfIsExecuting = false;
     }
     const int started = jobsStarted.loadAcquire();
     Q_ASSERT(started >= 0);
     Q_UNUSED(started);
     processCompletedElement(collection, job, thread);
     const int remainingJobs = jobCounter.fetchAndAddOrdered(-1) - 1;
     TWDEBUG(4, "Collection_Private::elementFinished: %i\n", remainingJobs);
     if (remainingJobs <= -1) {
         // its no use to count, the elements have been dequeued, now the threads call back that have been processing jobs in the meantime
     } else {
         Q_ASSERT(remainingJobs >= 0);
         if (remainingJobs == 0) {
             // all elements can only be done if self has been executed:
             // there is a small chance that (this) has been dequeued in the
             // meantime, in this case, there is nothing left to clean up
             finalCleanup(collection);
             {
                 MutexUnlocker u(&l);
                 Q_UNUSED(u);
                 selfExecuteWrapper.callEnd();
             }
             self.clear();
         }
     }
 }
 
 void Collection_Private::prepareToEnqueueElements()
 {
     // empty in Collection
 }
 
 JobInterface::Status Collection_Private::updateStatus(Collection *collection, JobPointer job)
 {
     // Keep our collection status in running until all jobs have finished
     // but make failures sticky so on first failed job we keep it counting as failure
     auto newStatus = Job::Status_Running;
     const auto jobStatus = job->status();
     if (jobStatus != JobInterface::Status_Success) {
         newStatus = jobStatus;
     }
     collection->setStatus(newStatus);
     return newStatus;
 }
 
 void Collection_Private::processCompletedElement(Collection *collection, JobPointer job, Thread *)
 {
     updateStatus(collection, job);
 }
 
 void Collection_Private::stop(Collection *collection)
 {
     QMutexLocker l(&mutex);
     if (api != nullptr) {
         // We can't dequeue ourselves while locked because we will
         // get deadlock when our own aboutToBeDequeued will be invoked
         // which will try to acquire this same lock
         // and we need our own `api` because `finalCleanup` can be invoked in between
         auto currentApi = api;
         l.unlock();
         TWDEBUG(4, "Collection::stop: dequeueing %p.\n", collection);
         if (!currentApi->dequeue(ManagedJobPointer<Collection>(collection))) {
             l.relock();
             dequeueElements(collection, false);
         }
     }
 }
 
 void Collection_Private::dequeueElements(Collection *collection, bool queueApiIsLocked)
 {
     // dequeue everything:
     Q_ASSERT(!mutex.tryLock());
     if (api == nullptr) {
         return; // not queued
     }
 
     for (const auto &job : elements) {
         bool result;
         if (queueApiIsLocked) {
             result = api->dequeue_p(job);
         } else {
             result = api->dequeue(job);
         }
         if (result) {
             jobCounter.fetchAndAddOrdered(-1);
         }
         TWDEBUG(3,
                 "Collection::Private::dequeueElements: dequeueing %p (%s, %i jobs left).\n",
                 (void *)job.data(),
                 result ? "found" : "not found",
                 jobCounter.loadAcquire());
         elementDequeued(job);
     }
 
     if (jobCounter.loadAcquire() == 1) {
         finalCleanup(collection);
     }
 }
 
 void Collection_Private::requestAbort(Collection *collection)
 {
     stop(collection);
     QMutexLocker l(&mutex);
     for (auto job = elements.begin(); job != elements.end(); ++job) {
         if ((*job)->status() <= JobInterface::Status_Running) {
             (*job)->requestAbort();
         }
     }
 }
 
 void CollectionSelfExecuteWrapper::begin(const JobPointer &job, Thread *thread)
 {
     job_ = job;
     thread_ = thread;
 }
 
 void CollectionSelfExecuteWrapper::end(const JobPointer &job, Thread *thread)
 {
     Q_ASSERT(job_ == job && thread_ == thread);
     Q_UNUSED(job);
     Q_UNUSED(thread); // except in assert
 }
 
 void CollectionSelfExecuteWrapper::callBegin()
 {
     ExecuteWrapper::begin(job_, thread_);
 }
 
 void CollectionSelfExecuteWrapper::callEnd()
 {
     ExecuteWrapper::end(job_, thread_);
     job_.clear();
 }
 
 }
 
 }