Add flag to jobs to indicate if they need to run or not

Once aspect jobs are inserted in the queue, they will always run, but
some of them don't actually have anything to do at every frame.

Ideally, when building a graph of jobs, you would include only jobs
that are required. However this makes building the graph more
complicated.

This introduces a method on the private job pimple which indicated if
the job actually has something to do or not.

When the scheduler is ready to run a job, it will call that method. If
there's nothing to do, then the job will be skipped and the dependent
jobs (if any) run instead.

Change-Id: Iaa8d3d064f03d8c0bab407879a9acea292a06784
Reviewed-by: Paul Lemire <paul.lemire@kdab.com>

10 files changed, 67 insertions, 21 deletions

diff --git a/src/core/jobs/qabstractaspectjobmanager_p.h b/src/core/jobs/qabstractaspectjobmanager_p.h
index b3e76b86d..c7f8cf721 100644
--- a/src/core/jobs/qabstractaspectjobmanager_p.h
+++ b/src/core/jobs/qabstractaspectjobmanager_p.h
@@ -66,7 +66,7 @@ public:
 
     virtual void initialize() {}
     virtual void enqueueJobs(const QVector<QAspectJobPtr> &jobQueue) = 0;
-    virtual void waitForAllJobs() = 0;
+    virtual int waitForAllJobs() = 0;
 
     // Callback signature for running SynchronizedJobs
     typedef void (*JobFunction)(void *);
diff --git a/src/core/jobs/qaspectjob.cpp b/src/core/jobs/qaspectjob.cpp
index 00ad1ace8..44efc2c0b 100644
--- a/src/core/jobs/qaspectjob.cpp
+++ b/src/core/jobs/qaspectjob.cpp
@@ -66,6 +66,11 @@ QAspectJobPrivate *QAspectJobPrivate::get(QAspectJob *job)
     return job->d_func();
 }
 
+bool QAspectJobPrivate::isRequired()
+{
+    return true;
+}
+
 void QAspectJobPrivate::postFrame(QAspectManager *aspectManager)
 {
     Q_UNUSED(aspectManager)
diff --git a/src/core/jobs/qaspectjob_p.h b/src/core/jobs/qaspectjob_p.h
index ddad09c86..5b72e64ee 100644
--- a/src/core/jobs/qaspectjob_p.h
+++ b/src/core/jobs/qaspectjob_p.h
@@ -72,6 +72,7 @@ public:
 
     static QAspectJobPrivate *get(QAspectJob *job);
 
+    virtual bool isRequired();
     virtual void postFrame(QAspectManager *aspectManager);
 
     QVector<QWeakPointer<QAspectJob> > m_dependencies;
diff --git a/src/core/jobs/qaspectjobmanager.cpp b/src/core/jobs/qaspectjobmanager.cpp
index be7942359..7103f5100 100644
--- a/src/core/jobs/qaspectjobmanager.cpp
+++ b/src/core/jobs/qaspectjobmanager.cpp
@@ -86,7 +86,7 @@ void QAspectJobManager::enqueueJobs(const QVector<QAspectJobPtr> &jobQueue)
         taskList << task;
     }
 
-    for (const QSharedPointer<QAspectJob> &job : jobQueue) {
+    for (const QAspectJobPtr &job : jobQueue) {
         const QVector<QWeakPointer<QAspectJob> > &deps = job->dependencies();
         AspectTaskRunnable *taskDepender = tasksMap.value(job.data());
 
@@ -108,9 +108,9 @@ void QAspectJobManager::enqueueJobs(const QVector<QAspectJobPtr> &jobQueue)
 }
 
 // Wait for all aspects jobs to be completed
-void QAspectJobManager::waitForAllJobs()
+int QAspectJobManager::waitForAllJobs()
 {
-    m_threadPooler->future().waitForFinished();
+    return m_threadPooler->waitForAllJobs();
 }
 
 void QAspectJobManager::waitForPerThreadFunction(JobFunction func, void *arg)
diff --git a/src/core/jobs/qaspectjobmanager_p.h b/src/core/jobs/qaspectjobmanager_p.h
index acbd0263e..db0075443 100644
--- a/src/core/jobs/qaspectjobmanager_p.h
+++ b/src/core/jobs/qaspectjobmanager_p.h
@@ -76,7 +76,7 @@ public:
 
     void enqueueJobs(const QVector<QAspectJobPtr> &jobQueue) override;
 
-    void waitForAllJobs() override;
+    int waitForAllJobs() override;
 
     void waitForPerThreadFunction(JobFunction func, void *arg) override;
 
diff --git a/src/core/jobs/qthreadpooler.cpp b/src/core/jobs/qthreadpooler.cpp
index f5c50062a..621881597 100644
--- a/src/core/jobs/qthreadpooler.cpp
+++ b/src/core/jobs/qthreadpooler.cpp
@@ -50,6 +50,7 @@ QThreadPooler::QThreadPooler(QObject *parent)
     , m_mutex()
     , m_taskCount(0)
     , m_threadPool(QThreadPool::globalInstance())
+    , m_totalRunJobs(0)
 {
     const QByteArray maxThreadCount = qgetenv("QT3D_MAX_THREAD_COUNT");
     if (!maxThreadCount.isEmpty()) {
@@ -75,6 +76,7 @@ void QThreadPooler::enqueueTasks(const QVector<RunnableInterface *> &tasks)
     // The caller have to set the mutex
     const QVector<RunnableInterface *>::const_iterator end = tasks.cend();
 
+    m_totalRunJobs = 0;
     for (QVector<RunnableInterface *>::const_iterator it = tasks.cbegin();
          it != end; ++it) {
 
@@ -86,32 +88,48 @@ void QThreadPooler::enqueueTasks(const QVector<RunnableInterface *> &tasks)
 
         if (!hasDependencies(*it) && !(*it)->reserved()) {
             (*it)->setReserved(true);
-            (*it)->setPooler(this);
-            m_threadPool->start((*it));
+            if ((*it)->isRequired()) {
+                (*it)->setPooler(this);
+                m_threadPool->start((*it));
+            } else {
+                release();
+                enqueueDepencies(*it);
+            }
         }
     }
 }
 
-void QThreadPooler::taskFinished(RunnableInterface *task)
+void QThreadPooler::enqueueDepencies(RunnableInterface *task)
 {
-    const QMutexLocker locker(&m_mutex);
-
-    release();
-
     if (task->type() == RunnableInterface::RunnableType::AspectTask) {
         AspectTaskRunnable *aspectTask = static_cast<AspectTaskRunnable *>(task);
         const auto &dependers = aspectTask->m_dependers;
         for (auto it = dependers.begin(); it != dependers.end(); ++it) {
-            aspectTask = static_cast<AspectTaskRunnable *>(*it);
-            if (--aspectTask->m_dependerCount == 0) {
-                if (!aspectTask->reserved()) {
-                    aspectTask->setReserved(true);
-                    aspectTask->setPooler(this);
-                    m_threadPool->start(aspectTask);
+            AspectTaskRunnable *dependerTask = static_cast<AspectTaskRunnable *>(*it);
+            if (--dependerTask->m_dependerCount == 0) {
+                if (!dependerTask->reserved()) {
+                    dependerTask->setReserved(true);
+                    if ((*it)->isRequired()) {
+                        dependerTask->setPooler(this);
+                        m_threadPool->start(dependerTask);
+                    } else {
+                        release();
+                        enqueueDepencies(*it);
+                    }
                 }
             }
         }
     }
+}
+
+void QThreadPooler::taskFinished(RunnableInterface *task)
+{
+    const QMutexLocker locker(&m_mutex);
+
+    release();
+    m_totalRunJobs++;
+
+    enqueueDepencies(task);
 
     if (currentCount() == 0) {
         if (m_futureInterface) {
@@ -137,6 +155,12 @@ QFuture<void> QThreadPooler::mapDependables(QVector<RunnableInterface *> &taskQu
     return QFuture<void>(m_futureInterface);
 }
 
+int QThreadPooler::waitForAllJobs()
+{
+    future().waitForFinished();
+    return m_totalRunJobs;
+}
+
 QFuture<void> QThreadPooler::future()
 {
     const QMutexLocker locker(&m_mutex);
diff --git a/src/core/jobs/qthreadpooler_p.h b/src/core/jobs/qthreadpooler_p.h
index 1970641b8..fafd549b2 100644
--- a/src/core/jobs/qthreadpooler_p.h
+++ b/src/core/jobs/qthreadpooler_p.h
@@ -69,10 +69,11 @@ class Q_3DCORE_PRIVATE_EXPORT QThreadPooler : public QObject
     Q_OBJECT
 
 public:
-    explicit QThreadPooler(QObject *parent = 0);
+    explicit QThreadPooler(QObject *parent = nullptr);
     ~QThreadPooler();
 
     QFuture<void> mapDependables(QVector<RunnableInterface *> &taskQueue);
+    int waitForAllJobs();
     void taskFinished(RunnableInterface *task);
     QFuture<void> future();
 
@@ -80,6 +81,7 @@ public:
 
 private:
     void enqueueTasks(const QVector<RunnableInterface *> &tasks);
+    void enqueueDepencies(RunnableInterface *task);
     void acquire(int add);
     void release();
     int currentCount() const;
@@ -89,6 +91,7 @@ private:
     QMutex m_mutex;
     QAtomicInt m_taskCount;
     QThreadPool *m_threadPool;
+    int m_totalRunJobs;
 };
 
 } // namespace Qt3DCore
diff --git a/src/core/jobs/task.cpp b/src/core/jobs/task.cpp
index 1dd5712c9..f5bfae014 100644
--- a/src/core/jobs/task.cpp
+++ b/src/core/jobs/task.cpp
@@ -68,6 +68,11 @@ AspectTaskRunnable::~AspectTaskRunnable()
 {
 }
 
+bool AspectTaskRunnable::isRequired()
+{
+    return m_job ? QAspectJobPrivate::get(m_job.data())->isRequired() : false;
+}
+
 void AspectTaskRunnable::run()
 {
     if (m_job) {
@@ -100,6 +105,11 @@ SyncTaskRunnable::~SyncTaskRunnable()
 {
 }
 
+bool SyncTaskRunnable::isRequired()
+{
+    return true;
+}
+
 void SyncTaskRunnable::run()
 {
     // Call the function
diff --git a/src/core/jobs/task_p.h b/src/core/jobs/task_p.h
index ff411539f..90d0674b4 100644
--- a/src/core/jobs/task_p.h
+++ b/src/core/jobs/task_p.h
@@ -77,6 +77,7 @@ public:
 
     virtual ~RunnableInterface();
 
+    virtual bool isRequired() = 0;
     virtual void run() = 0;
 
     virtual int id() = 0;
@@ -96,6 +97,7 @@ public:
     AspectTaskRunnable(QSystemInformationService *service);
     ~AspectTaskRunnable();
 
+    bool isRequired() override;
     void run() override;
 
     void setPooler(QThreadPooler *pooler) override { m_pooler = pooler; }
@@ -127,6 +129,7 @@ public:
                               QAtomicInt *atomicCount);
     ~SyncTaskRunnable();
 
+    bool isRequired() override;
     void run() override;
 
     void setPooler(QThreadPooler *pooler) override { m_pooler = pooler; }
diff --git a/src/core/qscheduler.cpp b/src/core/qscheduler.cpp
index 17ca1fbfd..9ff8a54f3 100644
--- a/src/core/qscheduler.cpp
+++ b/src/core/qscheduler.cpp
@@ -91,7 +91,7 @@ int QScheduler::scheduleAndWaitForFrameAspectJobs(qint64 time)
     // Do any other work here that the aspect thread can usefully be doing
     // whilst the threadpool works its way through the jobs
 
-    m_aspectManager->jobManager()->waitForAllJobs();
+    int totalJobs = m_aspectManager->jobManager()->waitForAllJobs();
 
     {
         QTaskLogger logger(m_aspectManager->serviceLocator()->systemInformation(), 4097, 0, QTaskLogger::AspectJob);
@@ -103,7 +103,7 @@ int QScheduler::scheduleAndWaitForFrameAspectJobs(qint64 time)
             QAbstractAspectPrivate::get(aspect)->jobsDone();
     }
 
-    return jobQueue.size();
+    return totalJobs;
 }
 
 } // namespace Qt3DCore