Async: support (re-)executing single Job multiple times

Storing Future and current Job progress directly in Executors means
that we cannot re-execute finished job, or even execute the same Job
multiple times in parallel. To do so, we need to make Executors stateless
and track the state elsewhere.

This change does that by moving the execution state from Executor to Execution
class. Executors now only describe the tasks to execute, while Execution holds
the current state of execution. New Execution is created every time Job::exec()
is called.

Execution holds reference to it's result (Future) and Executor which created
the Execution. This ensures that Executor is not deleted when Job (which owns
Executors) goes out of scope while the execution is still running. At the same
time Future holds reference to relevant Execution, so that the Execution is
deleted when all copies of Future referring result from the respective Execution
are deleted.

4 files changed, 228 insertions, 111 deletions

diff --git a/async/src/async.cpp b/async/src/async.cpp
index 20ba4e6..5e26bd8 100644
--- a/async/src/async.cpp
+++ b/async/src/async.cpp
@@ -26,15 +26,11 @@ using namespace Async;
 
 Private::ExecutorBase::ExecutorBase(const ExecutorBasePtr &parent)
     : mPrev(parent)
-    , mResult(0)
-    , mIsRunning(false)
-    , mIsFinished(false)
 {
 }
 
 Private::ExecutorBase::~ExecutorBase()
 {
-    delete mResult;
 }
 
 
diff --git a/async/src/async.h b/async/src/async.h
index 2741341..c6ca9e7 100644
--- a/async/src/async.h
+++ b/async/src/async.h
@@ -93,6 +93,49 @@ namespace Private
 class ExecutorBase;
 typedef QSharedPointer<ExecutorBase> ExecutorBasePtr;
 
+struct Execution {
+    Execution(const ExecutorBasePtr &executor)
+        : executor(executor)
+        , resultBase(nullptr)
+        , isRunning(false)
+        , isFinished(false)
+    {}
+
+    ~Execution()
+    {
+        if (resultBase) {
+            resultBase->releaseExecution();
+            delete resultBase;
+        }
+        prevExecution.reset();
+    }
+
+    void setFinished()
+    {
+        isFinished = true;
+        executor.clear();
+    }
+
+    template<typename T>
+    Async::Future<T>* result()
+    {
+        return static_cast<Async::Future<T>*>(resultBase);
+    }
+
+    void releaseFuture()
+    {
+        resultBase = 0;
+    }
+
+    ExecutorBasePtr executor;
+    FutureBase *resultBase;
+    bool isRunning;
+    bool isFinished;
+
+    ExecutionPtr prevExecution;
+};
+
+typedef QSharedPointer<Execution> ExecutionPtr;
 
 class ExecutorBase
 {
@@ -104,21 +147,15 @@ class ExecutorBase
 
 public:
     virtual ~ExecutorBase();
-    virtual void exec() = 0;
-
-    inline FutureBase* result() const
-    {
-        return mResult;
-    }
+    virtual ExecutionPtr exec(const ExecutorBasePtr &self) = 0;
 
 protected:
     ExecutorBase(const ExecutorBasePtr &parent);
 
-    ExecutorBasePtr mSelf;
+    template<typename T>
+    Async::Future<T>* createFuture(const ExecutionPtr &execution) const;
+
     ExecutorBasePtr mPrev;
-    FutureBase *mResult;
-    bool mIsRunning;
-    bool mIsFinished;
 };
 
 template<typename PrevOut, typename Out, typename ... In>
@@ -128,17 +165,13 @@ protected:
     Executor(ErrorHandler errorHandler, const Private::ExecutorBasePtr &parent)
         : ExecutorBase(parent)
         , mErrorFunc(errorHandler)
-        , mPrevFuture(0)
     {}
     virtual ~Executor() {}
-    inline Async::Future<PrevOut>* chainup();
-    virtual void previousFutureReady() = 0;
+    virtual void run(const ExecutionPtr &execution) = 0;
 
-    void exec();
+    ExecutionPtr exec(const ExecutorBasePtr &self);
 
-    //std::function<void(const In& ..., Async::Future<Out> &)> mFunc;
     std::function<void(int, const QString &)> mErrorFunc;
-    Async::Future<PrevOut> *mPrevFuture;
 };
 
 template<typename Out, typename ... In>
@@ -146,7 +179,7 @@ class ThenExecutor: public Executor<typename detail::prevOut<In ...>::type, Out,
 {
 public:
     ThenExecutor(ThenTask<Out, In ...> then, ErrorHandler errorHandler, const ExecutorBasePtr &parent);
-    void previousFutureReady();
+    void run(const ExecutionPtr &execution);
 private:
     ThenTask<Out, In ...> mFunc;
 };
@@ -156,7 +189,7 @@ class EachExecutor : public Executor<PrevOut, Out, In>
 {
 public:
     EachExecutor(EachTask<Out, In> each, ErrorHandler errorHandler, const ExecutorBasePtr &parent);
-    void previousFutureReady();
+    void run(const ExecutionPtr &execution);
 private:
     EachTask<Out, In> mFunc;
     QVector<Async::FutureWatcher<PrevOut>*> mFutureWatchers;
@@ -176,11 +209,11 @@ class SyncThenExecutor : public Executor<typename detail::prevOut<In ...>::type,
 {
 public:
     SyncThenExecutor(SyncThenTask<Out, In ...> then, ErrorHandler errorHandler, const ExecutorBasePtr &parent);
-    void previousFutureReady();
+    void run(const ExecutionPtr &execution);
 
 private:
-    void run(std::false_type); // !std::is_void<Out>
-    void run(std::true_type);  // std::is_void<Out>
+    void run(const ExecutionPtr &execution, std::false_type); // !std::is_void<Out>
+    void run(const ExecutionPtr &execution, std::true_type);  // std::is_void<Out>
     SyncThenTask<Out, In ...> mFunc;
 };
 
@@ -198,7 +231,7 @@ class SyncEachExecutor : public Executor<PrevOut, Out, In>
 {
 public:
     SyncEachExecutor(SyncEachTask<Out, In> each, ErrorHandler errorHandler, const ExecutorBasePtr &parent);
-    void previousFutureReady();
+    void run(const ExecutionPtr &execution);
 private:
     void run(Async::Future<Out> *future, const typename PrevOut::value_type &arg, std::false_type); // !std::is_void<Out>
     void run(Async::Future<Out> *future, const typename PrevOut::value_type &arg, std::true_type);  // std::is_void<Out>
@@ -430,18 +463,10 @@ public:
 
     Async::Future<Out> exec()
     {
-        // Have the top executor hold reference to itself during the execution.
-        // This ensures that even if the Job goes out of scope, the full Executor
-        // chain will not be destroyed.
-        // The executor will remove the self-reference once it's Future is finished.
-        mExecutor->mSelf = mExecutor;
-        mExecutor->exec();
-        return result();
-    }
+        Private::ExecutionPtr execution = mExecutor->exec(mExecutor);
+        Async::Future<Out> result = *execution->result<Out>();
 
-    Async::Future<Out> result() const
-    {
-        return *static_cast<Async::Future<Out>*>(mExecutor->result());
+        return result;
     }
 
 private:
@@ -549,74 +574,75 @@ Job<Out> error(int errorCode, const QString &errorMessage)
 
 namespace Private {
 
-template<typename PrevOut, typename Out, typename ... In>
-Future<PrevOut>* Executor<PrevOut, Out, In ...>::chainup()
+template<typename T>
+Async::Future<T>* ExecutorBase::createFuture(const ExecutionPtr &execution) const
 {
-    if (mPrev) {
-        mPrev->exec();
-        return static_cast<Async::Future<PrevOut>*>(mPrev->result());
-    } else {
-        return nullptr;
-    }
+    return new Async::Future<T>(execution);
 }
 
 template<typename PrevOut, typename Out, typename ... In>
-void Executor<PrevOut, Out, In ...>::exec()
+ExecutionPtr Executor<PrevOut, Out, In ...>::exec(const ExecutorBasePtr &self)
 {
-    // Don't chain up to job that already is running (or is finished)
-    if (mPrev && !mPrev->mIsRunning & !mPrev->mIsFinished) {
-        mPrevFuture = chainup();
+    // Passing 'self' to execution ensures that the Executor chain remains
+    // valid until the entire execution is finished
+    ExecutionPtr execution = ExecutionPtr::create(self);
+
+    // chainup
+    execution->prevExecution = mPrev ? mPrev->exec(mPrev) : ExecutionPtr();
+    /*
     } else if (mPrev && !mPrevFuture) {
         // If previous job is running or finished, just get it's future
         mPrevFuture = static_cast<Async::Future<PrevOut>*>(mPrev->result());
     }
+    */
 
-    // Initialize our future
-    mResult = new Async::Future<Out>();
+    execution->resultBase = this->createFuture<Out>(execution);
     auto fw = new Async::FutureWatcher<Out>();
     QObject::connect(fw, &Async::FutureWatcher<Out>::futureReady,
-                     [fw, this]() {
-                         mIsFinished = true;
-                         mSelf.clear();
+                     [fw, execution, this]() {
+                         execution->setFinished();
                          delete fw;
                      });
-    fw->setFuture(*static_cast<Async::Future<Out>*>(mResult));
+    fw->setFuture(*execution->result<Out>());
 
-    if (!mPrevFuture || mPrevFuture->isFinished()) {
-        if (mPrevFuture && mPrevFuture->errorCode() != 0) {
+    Async::Future<PrevOut> *prevFuture = execution->prevExecution ? execution->prevExecution->result<PrevOut>() : nullptr;
+    if (!prevFuture || prevFuture->isFinished()) {
+        if (prevFuture && prevFuture->errorCode() != 0) {
             if (mErrorFunc) {
-                mErrorFunc(mPrevFuture->errorCode(), mPrevFuture->errorMessage());
-                mResult->setFinished();
-                mIsFinished = true;
-                return;
+                mErrorFunc(prevFuture->errorCode(), prevFuture->errorMessage());
+                execution->resultBase->setFinished();
+                execution->setFinished();
+                return execution;
             } else {
                 // Propagate the error to next caller
             }
         }
-        mIsRunning = true;
-        previousFutureReady();
+        execution->isRunning = true;
+        run(execution);
     } else {
         auto futureWatcher = new Async::FutureWatcher<PrevOut>();
         QObject::connect(futureWatcher, &Async::FutureWatcher<PrevOut>::futureReady,
-                         [futureWatcher, this]() {
+                         [futureWatcher, execution, this]() {
                              auto prevFuture = futureWatcher->future();
                              assert(prevFuture.isFinished());
                              delete futureWatcher;
                              if (prevFuture.errorCode() != 0) {
                                  if (mErrorFunc) {
                                      mErrorFunc(prevFuture.errorCode(), prevFuture.errorMessage());
-                                     mResult->setFinished();
+                                     execution->resultBase->setFinished();
                                      return;
                                  } else {
                                      // Propagate the error to next caller
                                  }
                              }
-                             mIsRunning = true;
-                             previousFutureReady();
+                             execution->isRunning = true;
+                             run(execution);
                          });
 
-        futureWatcher->setFuture(*mPrevFuture);
+        futureWatcher->setFuture(*static_cast<Async::Future<PrevOut>*>(prevFuture));
     }
+
+    return execution;
 }
 
 
@@ -628,14 +654,15 @@ ThenExecutor<Out, In ...>::ThenExecutor(ThenTask<Out, In ...> then, ErrorHandler
 }
 
 template<typename Out, typename ... In>
-void ThenExecutor<Out, In ...>::previousFutureReady()
+void ThenExecutor<Out, In ...>::run(const ExecutionPtr &execution)
 {
-    if (this->mPrevFuture) {
-        assert(this->mPrevFuture->isFinished());
+    Async::Future<typename detail::prevOut<In ...>::type> *prevFuture = nullptr;
+    if (execution->prevExecution) {
+        prevFuture = execution->prevExecution->result<typename detail::prevOut<In ...>::type>();
+        assert(prevFuture->isFinished());
     }
 
-    this->mFunc(this->mPrevFuture ? this->mPrevFuture->value() : In() ...,
-                *static_cast<Async::Future<Out>*>(this->mResult));
+    this->mFunc(prevFuture ? prevFuture->value() : In() ..., *execution->result<Out>());
 }
 
 template<typename PrevOut, typename Out, typename In>
@@ -646,33 +673,37 @@ EachExecutor<PrevOut, Out, In>::EachExecutor(EachTask<Out, In> each, ErrorHandle
 }
 
 template<typename PrevOut, typename Out, typename In>
-void EachExecutor<PrevOut, Out, In>::previousFutureReady()
+void EachExecutor<PrevOut, Out, In>::run(const ExecutionPtr &execution)
 {
-    assert(this->mPrevFuture->isFinished());
-    auto out = static_cast<Async::Future<Out>*>(this->mResult);
-    if (this->mPrevFuture->value().isEmpty()) {
+    assert(execution->prevExecution);
+    auto prevFuture = execution->prevExecution->result<PrevOut>();
+    assert(prevFuture->isFinished());
+
+    auto out = execution->result<Out>();
+    if (prevFuture->value().isEmpty()) {
         out->setFinished();
         return;
     }
 
-    for (auto arg : this->mPrevFuture->value()) {
-        auto future = new Async::Future<Out>;
-        this->mFunc(arg, *future);
+    for (auto arg : prevFuture->value()) {
+        Async::Future<Out> future;
+        this->mFunc(arg, future);
         auto fw = new Async::FutureWatcher<Out>();
         mFutureWatchers.append(fw);
         QObject::connect(fw, &Async::FutureWatcher<Out>::futureReady,
-                         [out, future, fw, this]() {
-                             assert(future->isFinished());
+                         [out, fw, this]() {
+                             auto future = fw->future();
+                             assert(future.isFinished());
                              const int index = mFutureWatchers.indexOf(fw);
                              assert(index > -1);
                              mFutureWatchers.removeAt(index);
-                             out->setValue(out->value() + future->value());
-                             delete future;
+                             out->setValue(out->value() + future.value());
                              if (mFutureWatchers.isEmpty()) {
                                  out->setFinished();
                              }
+                             delete fw;
                          });
-        fw->setFuture(*future);
+        fw->setFuture(future);
     }
 }
 
@@ -690,27 +721,37 @@ SyncThenExecutor<Out, In ...>::SyncThenExecutor(SyncThenTask<Out, In ...> then,
 }
 
 template<typename Out, typename ... In>
-void SyncThenExecutor<Out, In ...>::previousFutureReady()
+void SyncThenExecutor<Out, In ...>::run(const ExecutionPtr &execution)
 {
-    if (this->mPrevFuture) {
-        assert(this->mPrevFuture->isFinished());
+    if (execution->prevExecution) {
+        assert(execution->prevExecution->resultBase->isFinished());
     }
 
-    run(std::is_void<Out>());
-    this->mResult->setFinished();
+    run(execution, std::is_void<Out>());
+    execution->resultBase->setFinished();
 }
 
 template<typename Out, typename ... In>
-void SyncThenExecutor<Out, In ...>::run(std::false_type)
-{
-    Out result = this->mFunc(this->mPrevFuture ? this->mPrevFuture->value() : In() ...);
-    static_cast<Async::Future<Out>*>(this->mResult)->setValue(result);
+void SyncThenExecutor<Out, In ...>::run(const ExecutionPtr &execution, std::false_type)
+{
+    Async::Future<typename detail::prevOut<In ...>::type> *prevFuture =
+        execution->prevExecution
+            ? execution->prevExecution->result<typename detail::prevOut<In ...>::type>()
+            : nullptr;
+    (void) prevFuture; // silence 'set but not used' warning
+    Async::Future<Out> *future = execution->result<Out>();
+    future->setValue(this->mFunc(prevFuture ? prevFuture->value() : In() ...));
 }
 
 template<typename Out, typename ... In>
-void SyncThenExecutor<Out, In ...>::run(std::true_type)
-{
-    this->mFunc(this->mPrevFuture ? this->mPrevFuture->value() : In() ...);
+void SyncThenExecutor<Out, In ...>::run(const ExecutionPtr &execution, std::true_type)
+{
+    Async::Future<typename detail::prevOut<In ...>::type> *prevFuture =
+        execution->prevExecution
+            ? execution->prevExecution->result<typename detail::prevOut<In ...>::type>()
+            : nullptr;
+    (void) prevFuture; // silence 'set but not used' warning
+    this->mFunc(prevFuture ? prevFuture->value() : In() ...);
 }
 
 template<typename PrevOut, typename Out, typename In>
@@ -721,16 +762,19 @@ SyncEachExecutor<PrevOut, Out, In>::SyncEachExecutor(SyncEachTask<Out, In> each,
 }
 
 template<typename PrevOut, typename Out, typename In>
-void SyncEachExecutor<PrevOut, Out, In>::previousFutureReady()
+void SyncEachExecutor<PrevOut, Out, In>::run(const ExecutionPtr &execution)
 {
-    assert(this->mPrevFuture->isFinished());
-    auto out = static_cast<Async::Future<Out>*>(this->mResult);
-    if (this->mPrevFuture->value().isEmpty()) {
+    assert(execution->prevExecution);
+    auto *prevFuture = execution->prevExecution->result<PrevOut>();
+    assert(prevFuture->isFinished());
+
+    auto out = execution->result<Out>();
+    if (prevFuture->value().isEmpty()) {
         out->setFinished();
         return;
     }
 
-    for (auto arg : this->mPrevFuture->value()) {
+    for (auto arg : prevFuture->value()) {
         run(out, arg, std::is_void<Out>());
     }
     out->setFinished();
@@ -743,7 +787,7 @@ void SyncEachExecutor<PrevOut, Out, In>::run(Async::Future<Out> *out, const type
 }
 
 template<typename PrevOut, typename Out, typename In>
-void SyncEachExecutor<PrevOut, Out, In>::run(Async::Future<Out> * /* unushed */, const typename PrevOut::value_type &arg, std::true_type)
+void SyncEachExecutor<PrevOut, Out, In>::run(Async::Future<Out> * /* unused */, const typename PrevOut::value_type &arg, std::true_type)
 {
     this->mFunc(arg);
 }
diff --git a/async/src/future.cpp b/async/src/future.cpp
index ab02baf..50a326a 100644
--- a/async/src/future.cpp
+++ b/async/src/future.cpp
@@ -16,6 +16,7 @@
  */
 
 #include "future.h"
+#include "async.h"
 
 using namespace Async;
 
@@ -31,6 +32,27 @@ FutureBase::~FutureBase()
 {
 }
 
+FutureBase::PrivateBase::PrivateBase(const Private::ExecutionPtr &execution)
+    : mExecution(execution)
+{
+}
+
+FutureBase::PrivateBase::~PrivateBase()
+{
+    Private::ExecutionPtr executionPtr = mExecution.toStrongRef();
+    if (executionPtr) {
+        executionPtr->releaseFuture();
+        releaseExecution();
+    }
+}
+
+void FutureBase::PrivateBase::releaseExecution()
+{
+    mExecution.clear();
+}
+
+
+
 FutureWatcherBase::FutureWatcherBase(QObject *parent)
     : QObject(parent)
 {
diff --git a/async/src/future.h b/async/src/future.h
index b580b5a..cadd96d 100644
--- a/async/src/future.h
+++ b/async/src/future.h
@@ -29,8 +29,18 @@ class QEventLoop;
 
 namespace Async {
 
+namespace Private {
+class Execution;
+class ExecutorBase;
+
+typedef QSharedPointer<Execution> ExecutionPtr;
+
+}
+
 class FutureBase
 {
+    friend class Async::Private::Execution;
+
 public:
     virtual ~FutureBase();
 
@@ -39,6 +49,20 @@ public:
     virtual void setError(int code = 1, const QString &message = QString()) = 0;
 
 protected:
+    virtual void releaseExecution() = 0;
+
+    class PrivateBase : public QSharedData
+    {
+    public:
+        PrivateBase(const Async::Private::ExecutionPtr &execution);
+        virtual ~PrivateBase();
+
+        void releaseExecution();
+
+    private:
+        QWeakPointer<Async::Private::Execution> mExecution;
+    };
+
     FutureBase();
     FutureBase(const FutureBase &other);
 };
@@ -104,9 +128,9 @@ public:
     }
 
 protected:
-    FutureGeneric()
+    FutureGeneric(const Async::Private::ExecutionPtr &execution)
         : FutureBase()
-        , d(new Private)
+        , d(new Private(execution))
     {}
 
     FutureGeneric(const FutureGeneric<T> &other)
@@ -114,10 +138,15 @@ protected:
         , d(other.d)
     {}
 
-    class Private : public QSharedData
+    class Private : public FutureBase::PrivateBase
     {
     public:
-        Private() : QSharedData(), finished(false), errorCode(0) {}
+        Private(const Async::Private::ExecutionPtr &execution)
+            : FutureBase::PrivateBase(execution)
+            , finished(false)
+            , errorCode(0)
+        {}
+
         typename std::conditional<std::is_void<T>::value, int /* dummy */, T>::type
         value;
 
@@ -129,6 +158,11 @@ protected:
 
     QExplicitlySharedDataPointer<Private> d;
 
+    void releaseExecution()
+    {
+        d->releaseExecution();
+    }
+
     void addWatcher(FutureWatcher<T> *watcher)
     {
         d->watchers.append(QPointer<FutureWatcher<T>>(watcher));
@@ -138,9 +172,14 @@ protected:
 template<typename T>
 class Future : public FutureGeneric<T>
 {
+    friend class Async::Private::ExecutorBase;
+
+    template<typename T_>
+    friend class Async::FutureWatcher;
+
 public:
     Future()
-        : FutureGeneric<T>()
+        : FutureGeneric<T>(Async::Private::ExecutionPtr())
     {}
 
     Future(const Future<T> &other)
@@ -156,19 +195,35 @@ public:
     {
         return this->d->value;
     }
+
+protected:
+    Future(const Async::Private::ExecutionPtr &execution)
+        : FutureGeneric<T>(execution)
+    {}
+
 };
 
 template<>
 class Future<void> : public FutureGeneric<void>
 {
+    friend class Async::Private::ExecutorBase;
+
+    template<typename T_>
+    friend class Async::FutureWatcher;
+
 public:
     Future()
-        : FutureGeneric<void>()
+        : FutureGeneric<void>(Async::Private::ExecutionPtr())
     {}
 
     Future(const Future<void> &other)
         : FutureGeneric<void>(other)
     {}
+
+protected:
+    Future(const Async::Private::ExecutionPtr &execution)
+        : FutureGeneric<void>(execution)
+    {}
 };
 
 
@@ -177,7 +232,7 @@ class FutureWatcherBase : public QObject
     Q_OBJECT
 
 protected:
-    FutureWatcherBase(QObject *parent = 0);
+    FutureWatcherBase(QObject *parent = nullptr);
     virtual ~FutureWatcherBase();
 
 Q_SIGNALS:
@@ -190,7 +245,7 @@ class FutureWatcher : public FutureWatcherBase
     friend class Async::FutureGeneric<T>;
 
 public:
-    FutureWatcher(QObject *parent = 0)
+    FutureWatcher(QObject *parent = nullptr)
         : FutureWatcherBase(parent)
     {}