Add checks for double task execution

On ThreadPooler done tasks are also removed from the dependency pair
list. Also added reserved flag for tasks. These prevent tasks getting
into execution more than once.

Change-Id: Ia9e5377c883ad08079a9af32c4a2a294db425eb2
Reviewed-by: Sean Harmer <sean.harmer@kdab.com>
Reviewed-by: Paul Lemire <paul.lemire@kdab.com>

6 files changed, 48 insertions, 18 deletions

diff --git a/src/core/jobs/dependencyhandler.cpp b/src/core/jobs/dependencyhandler.cpp
index 118602a61..a805638fe 100644
--- a/src/core/jobs/dependencyhandler.cpp
+++ b/src/core/jobs/dependencyhandler.cpp
@@ -69,13 +69,25 @@ namespace {
         bool operator()(const Dependency &candidate) const
         {
             if (dependee == candidate.dependee) {
-                freedList->append(candidate.depender);
+                if (!candidate.depender->reserved())
+                    freedList->append(candidate.depender);
                 return true;
             }
             return false;
         }
     };
 
+    struct DependerEquals : std::unary_function<Dependency, bool>
+    {
+        const RunnableInterface *depender;
+        explicit DependerEquals(const RunnableInterface *depender)
+            : depender(qMove(depender)) {}
+        bool operator()(const Dependency &candidate) const
+        {
+            return depender == candidate.depender;
+        }
+    };
+
     struct ByDependerThenDependee : std::binary_function<Dependency, Dependency, bool>
     {
         // Defines a lexicographical order (depender first).
@@ -118,14 +130,19 @@ bool DependencyHandler::hasDependency(const RunnableInterface *depender)
  * Removes all the entries on the m_dependencyMap that have given task as a dependee,
  * i.e. entries where the dependency is on the given task.
  */
-QVector<RunnableInterface *> DependencyHandler::freeDependencies(const RunnableInterface *dependee)
+QVector<RunnableInterface *> DependencyHandler::freeDependencies(const RunnableInterface *task)
 {
-    const QMutexLocker locker(m_mutex);
+    // The caller has to set the mutex, which is QThreadPooler::taskFinished
+
+    m_dependencyMap.erase(std::remove_if(m_dependencyMap.begin(),
+                                         m_dependencyMap.end(),
+                                         DependerEquals(task)),
+                          m_dependencyMap.end());
 
     QVector<RunnableInterface *> freedList;
     m_dependencyMap.erase(std::remove_if(m_dependencyMap.begin(),
                                          m_dependencyMap.end(),
-                                         DependeeEquals(dependee, &freedList)),
+                                         DependeeEquals(task, &freedList)),
                           m_dependencyMap.end());
 
     return freedList;
diff --git a/src/core/jobs/dependencyhandler_p.h b/src/core/jobs/dependencyhandler_p.h
index e48113b80..288ee20b3 100644
--- a/src/core/jobs/dependencyhandler_p.h
+++ b/src/core/jobs/dependencyhandler_p.h
@@ -81,7 +81,7 @@ public:
 
     void addDependencies(QVector<Dependency> dependencies);
     bool hasDependency(const RunnableInterface *depender);
-    QVector<RunnableInterface *> freeDependencies(const RunnableInterface *dependee);
+    QVector<RunnableInterface *> freeDependencies(const RunnableInterface *task);
     void setMutex(QMutex *mutex) { m_mutex = mutex; }
 
 private:
diff --git a/src/core/jobs/qthreadpooler.cpp b/src/core/jobs/qthreadpooler.cpp
index 890a8942a..94d5b29c2 100644
--- a/src/core/jobs/qthreadpooler.cpp
+++ b/src/core/jobs/qthreadpooler.cpp
@@ -73,21 +73,25 @@ void QThreadPooler::enqueueTasks(QVector<RunnableInterface *> &tasks)
 
     for (QVector<RunnableInterface *>::iterator it = tasks.begin();
          it != tasks.end(); it++) {
-        if (!m_dependencyHandler->hasDependency((*it))) {
+        if (!m_dependencyHandler->hasDependency((*it)) && !(*it)->reserved()) {
+            (*it)->setReserved(true);
             (*it)->setPooler(this);
             QThreadPool::globalInstance()->start((*it));
         }
     }
 }
 
-void QThreadPooler::taskFinished(QVector<RunnableInterface *> tasks)
+void QThreadPooler::taskFinished(RunnableInterface *task)
 {
     const QMutexLocker locker(m_mutex);
 
     release();
 
-    if (tasks.size())
-        enqueueTasks(tasks);
+    QVector<RunnableInterface *> freedTasks;
+    if (task->dependencyHandler())
+        freedTasks = m_dependencyHandler->freeDependencies(task);
+    if (freedTasks.size())
+        enqueueTasks(freedTasks);
 
     if (currentCount() == 0) {
         if (m_futureInterface) {
diff --git a/src/core/jobs/qthreadpooler_p.h b/src/core/jobs/qthreadpooler_p.h
index 9abbdc8d0..4a0a189a6 100644
--- a/src/core/jobs/qthreadpooler_p.h
+++ b/src/core/jobs/qthreadpooler_p.h
@@ -58,7 +58,7 @@ public:
     ~QThreadPooler();
 
     QFuture<void> mapDependables(QVector<RunnableInterface *> &taskQueue);
-    void taskFinished(QVector<RunnableInterface *> tasks);
+    void taskFinished(RunnableInterface *task);
     QFuture<void> future();
 
     void setDependencyHandler(DependencyHandler *handler);
diff --git a/src/core/jobs/task.cpp b/src/core/jobs/task.cpp
index 55f4bacb6..8cfc365fa 100644
--- a/src/core/jobs/task.cpp
+++ b/src/core/jobs/task.cpp
@@ -53,7 +53,8 @@ RunnableInterface::~RunnableInterface()
 // Aspect task
 
 AspectTaskRunnable::AspectTaskRunnable()
-    : m_dependencyHandler(0)
+    : m_dependencyHandler(0),
+      m_reserved(false)
 {
 }
 
@@ -66,12 +67,8 @@ void AspectTaskRunnable::run()
     if (m_job)
         m_job->run();
 
-    QVector<RunnableInterface *> freedTasks;
-    if (m_dependencyHandler)
-        freedTasks = m_dependencyHandler->freeDependencies(this);
-
     if (m_pooler)
-        m_pooler->taskFinished(freedTasks);
+        m_pooler->taskFinished(this);
 }
 
 void AspectTaskRunnable::setDependencyHandler(DependencyHandler *handler)
@@ -91,7 +88,8 @@ SyncTaskRunnable::SyncTaskRunnable(QAbstractAspectJobManager::JobFunction func,
     : m_func(func),
       m_arg(arg),
       m_atomicCount(atomicCount),
-      m_pooler(Q_NULLPTR)
+      m_pooler(Q_NULLPTR),
+      m_reserved(false)
 {
 }
 
@@ -112,7 +110,7 @@ void SyncTaskRunnable::run()
         QThread::currentThread()->yieldCurrentThread();
 
     if (m_pooler)
-        m_pooler->taskFinished(QVector<RunnableInterface *>());
+        m_pooler->taskFinished(this);
 }
 
 void SyncTaskRunnable::setDependencyHandler(DependencyHandler *handler)
diff --git a/src/core/jobs/task_p.h b/src/core/jobs/task_p.h
index 9b5c6a0f2..ea8547303 100644
--- a/src/core/jobs/task_p.h
+++ b/src/core/jobs/task_p.h
@@ -66,6 +66,9 @@ public:
     virtual int id() = 0;
     virtual void setId(int id) = 0;
 
+    virtual void setReserved(bool reserved) = 0;
+    virtual bool reserved() = 0;
+
     virtual void setPooler(QThreadPooler *pooler) = 0;
 };
 
@@ -82,6 +85,9 @@ public:
 
     void setPooler(QThreadPooler *pooler) Q_DECL_OVERRIDE { m_pooler = pooler; }
 
+    void setReserved(bool reserved) Q_DECL_OVERRIDE { m_reserved = reserved; }
+    bool reserved() Q_DECL_OVERRIDE { return m_reserved; }
+
     int id() Q_DECL_OVERRIDE { return m_id; }
     void setId(int id) Q_DECL_OVERRIDE { m_id = id; }
 
@@ -91,6 +97,7 @@ public:
 private:
     DependencyHandler *m_dependencyHandler;
     QThreadPooler *m_pooler;
+    bool m_reserved;
 
     int m_id; // For testing purposes for now
 };
@@ -109,6 +116,9 @@ public:
 
     void setPooler(QThreadPooler *pooler) Q_DECL_OVERRIDE { m_pooler = pooler; }
 
+    void setReserved(bool reserved) Q_DECL_OVERRIDE { m_reserved = reserved; }
+    bool reserved() Q_DECL_OVERRIDE { return m_reserved; }
+
     int id() Q_DECL_OVERRIDE { return m_id; }
     void setId(int id) Q_DECL_OVERRIDE { m_id = id; }
 
@@ -118,6 +128,7 @@ private:
     QAtomicInt *m_atomicCount;
 
     QThreadPooler *m_pooler;
+    bool m_reserved;
 
     int m_id;
 };