From c30e9145049c52feb2de719307ebbfee0650f01b Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Dan=20Vr=C3=A1til?= <dvratil@redhat.com>
Date: Thu, 11 Dec 2014 15:55:18 +0100
Subject: Async: move the actual task exection into Executor implementation

As of now, Job is only front interface to a chain of Executor subclasses. Each
Executor subclass specializes for given type of execution (then, each, reduce, ...),
and the chain is then executed recursively, as we did with the original Job
implementation.
---
 async/autotests/asynctest.cpp |  15 ++--
 async/src/async.cpp           |  19 +++-
 async/src/async.h             | 200 ++++++++++++++++++++++++------------------
 async/src/future.h            |   2 +
 4 files changed, 142 insertions(+), 94 deletions(-)

diff --git a/async/autotests/asynctest.cpp b/async/autotests/asynctest.cpp
index 19e40f7..403fb83 100644
--- a/async/autotests/asynctest.cpp
+++ b/async/autotests/asynctest.cpp
@@ -91,18 +91,21 @@ void AsyncTest::testAsyncPromises()
 
 void AsyncTest::testSyncEach()
 {
-  /*
     auto job = Async::start<QList<int>>(
-        []() -> Async::Future<QList<int>> {
-            Async::Future<QList<int>> future(QList<int>{ 1, 2, 3, 4 });
+        [](Async::Future<QList<int>> &future) {
+            future.setValue(QList<int>{ 1, 2, 3, 4 });
             future.setFinished();
-            return future;
         })
     .each<QList<int>, int>(
         [](const int &v, Async::Future<QList<int>> &future) {
-            setFinished();
+            future.setValue(QList<int>{ v + 1 });
+            future.setFinished();
         });
-        */
+
+    job.exec();
+    Async::Future<QList<int>> future = job.result();
+    const QList<int> expected({ 2, 3, 4, 5 });
+    QCOMPARE(future.value(), expected);
 }
 
 
diff --git a/async/src/async.cpp b/async/src/async.cpp
index 7e81f24..c4a88fd 100644
--- a/async/src/async.cpp
+++ b/async/src/async.cpp
@@ -28,13 +28,20 @@
 
 using namespace Async;
 
-JobBase::JobBase(JobBase::JobType jobType, JobBase* prev)
-    : mPrev(prev)
-    , mResult(0)
-    , mJobType(jobType)
+JobBase::JobBase(Executor *executor)
+    : mExecutor(executor)
 {
 }
 
+JobBase::~JobBase()
+{
+}
+
+void JobBase::exec()
+{
+    mExecutor->exec();
+}
+
 
 FutureBase::FutureBase()
     : mFinished(false)
@@ -48,6 +55,10 @@ FutureBase::FutureBase(const Async::FutureBase &other)
 {
 }
 
+FutureBase::~FutureBase()
+{
+}
+
 void FutureBase::setFinished()
 {
     mFinished = true;
diff --git a/async/src/async.h b/async/src/async.h
index 182e57c..0f027f5 100644
--- a/async/src/async.h
+++ b/async/src/async.h
@@ -27,6 +27,7 @@
 #include <type_traits>
 #include <iostream>
 #include <cassert>
+#include <iterator>
 
 #include "future.h"
 #include "async_impl.h"
@@ -49,97 +50,159 @@ using ReduceTask = typename detail::identity<std::function<void(In, Async::Futur
 template<typename Out, typename ... In>
 Job<Out, In ...> start(ThenTask<Out, In ...> func);
 
-namespace Private
+class Executor
 {
-    template<typename Out, typename ... In>
-    void doExec(Job<Out, In ...> *job, Async::Future<Out> &out, const In & ... args);
-}
+
+public:
+    Executor(Executor *parent)
+      : mPrev(parent)
+      , mResult(0)
+    {
+    }
+
+    virtual ~Executor()
+    {
+        delete mResult;
+    }
+
+    virtual void exec() = 0;
+
+    FutureBase* result() const
+    {
+        return mResult;
+    }
+
+    Executor *mPrev;
+    FutureBase *mResult;
+};
+
+template<typename Out, typename ... In>
+class ThenExecutor:  public Executor
+{
+
+    typedef Out OutType;
+    typedef typename std::tuple_element<0, std::tuple<In ..., void>>::type InType;
+
+
+public:
+    ThenExecutor(ThenTask<Out, In ...> then, Executor *parent = nullptr)
+        : Executor(parent)
+        , mFunc(then)
+    {
+    }
+
+    void exec()
+    {
+        Async::Future<InType> *in = 0;
+        if (mPrev) {
+            mPrev->exec();
+            in = static_cast<Async::Future<InType>*>(mPrev->result());
+            assert(in->isFinished());
+        }
+
+        auto out = new Async::Future<Out>();
+        mFunc(in ? in->value() : In() ..., *out);
+        out->waitForFinished();
+        mResult = out;
+    }
+
+private:
+    std::function<void(const In& ..., Async::Future<Out>&)> mFunc;
+};
+
+template<typename PrevOut, typename Out, typename In>
+class EachExecutor : public Executor
+{
+public:
+    EachExecutor(EachTask<Out, In> each, Executor *parent = nullptr)
+        : Executor(parent)
+        , mFunc(each)
+    {
+    }
+
+    void exec()
+    {
+        assert(mPrev);
+        mPrev->exec();
+        Async::Future<PrevOut> *in = static_cast<Async::Future<PrevOut>*>(mPrev->result());
+
+        auto *out = new Async::Future<Out>();
+        for (auto arg : in->value()) {
+            Async::Future<Out> future;
+            mFunc(arg, future);
+            future.waitForFinished();
+            out->setValue(out->value() + future.value());
+        }
+
+        mResult = out;
+    }
+
+private:
+    std::function<void(const In&, Async::Future<Out>&)> mFunc;
+};
 
 class JobBase
 {
     template<typename Out, typename ... In>
     friend class Job;
 
-protected:
-    enum JobType {
-        Then,
-        Each,
-        Reduce
-    };
-
 public:
-    JobBase(JobType jobType, JobBase *prev = nullptr);
-    virtual void exec() = 0;
+    JobBase(Executor *executor);
+    ~JobBase();
+
+    void exec();
 
 protected:
-    JobBase *mPrev;
-    void *mResult;
-    JobType mJobType;
+    Executor *mExecutor;
 };
 
 template<typename Out, typename ... In>
 class Job : public JobBase
 {
-    template<typename Out_, typename ... In_>
+    template<typename OutOther, typename ... InOther>
     friend class Job;
 
-    template<typename Out_, typename ... In_>
-    friend Job<Out_, In_ ...> start(Async::ThenTask<Out_, In_ ...> func);
-
-    typedef Out OutType;
-    typedef typename std::tuple_element<0, std::tuple<In ..., void>>::type InType;
+    template<typename OutOther, typename ... InOther>
+    friend Job<OutOther, InOther ...> start(Async::ThenTask<OutOther, InOther ...> func);
 
 public:
-    ~Job()
+    template<typename OutOther, typename ... InOther>
+    Job<OutOther, InOther ...> then(ThenTask<OutOther, InOther ...> func)
     {
-        delete reinterpret_cast<Async::Future<Out>*>(mResult);
+        Executor *exec = new ThenExecutor<OutOther, InOther ...>(func, mExecutor);
+        return Job<OutOther, InOther ...>(exec);
     }
 
-    template<typename Out_, typename ... In_>
-    Job<Out_, In_ ...> then(ThenTask<Out_, In_ ...> func)
+    template<typename OutOther, typename InOther>
+    Job<OutOther, InOther> each(EachTask<OutOther, InOther> func)
     {
-        return Job<Out_, In_ ...>::create(func, JobBase::Then, this);
-    }
-
-    template<typename Out_, typename In_>
-    Job<Out_, In_> each(EachTask<Out_, In_> func)
-    {
-        static_assert(detail::isIterable<OutType>::value,
+        static_assert(detail::isIterable<Out>::value,
                       "The 'Each' task can only be connected to a job that returns a list or array.");
-        static_assert(detail::isIterable<Out_>::value,
+        static_assert(detail::isIterable<OutOther>::value,
                       "The result type of 'Each' task must be a list or an array.");
-        return Job<Out_, In_>::create(func, JobBase::Each, this);
+        return Job<OutOther, InOther>(new EachExecutor<Out, OutOther, InOther>(func, mExecutor));
     }
 
-    template<typename Out_, typename In_>
-    Job<Out_, In_> reduce(ReduceTask<Out_, In_> func)
+    template<typename OutOther, typename InOther>
+    Job<OutOther, InOther> reduce(ReduceTask<OutOther, InOther> func)
     {
-        static_assert(Async::detail::isIterable<OutType>::value,
+        static_assert(Async::detail::isIterable<Out>::value,
                       "The result type of 'Reduce' task must be a list or an array.");
-        return Job<Out_, In_>::create(func, JobBase::Reduce, this);
+        //return Job<Out_, In_>::create(func, new ReduceEx, this);
     }
 
     Async::Future<Out> result() const
     {
-        return *reinterpret_cast<Async::Future<Out>*>(mResult);
+        return *static_cast<Async::Future<Out>*>(mExecutor->result());
     }
 
-    void exec();
-
 private:
-    Job(JobBase::JobType jobType, JobBase *parent = nullptr)
-        : JobBase(jobType, parent)
+    Job(Executor *executor)
+        : JobBase(executor)
     {
     }
-
-    template<typename F>
-    static Job<Out, In ... > create(F func, JobBase::JobType jobType, JobBase *parent = nullptr);
-
-public:
-    std::function<void(In ..., Async::Future<Out>&)> mFunc;
 };
 
-
 } // namespace Async
 
 
@@ -149,41 +212,10 @@ public:
 template<typename Out, typename ... In>
 Async::Job<Out, In ...> Async::start(ThenTask<Out, In ...> func)
 {
-    return Job<Out, In ...>::create(func, JobBase::Then);
-}
-
-template<typename Out, typename ... In>
-void Async::Private::doExec(Job<Out, In ...> *job, Async::Future<Out> &out, const In & ... args)
-{
-    job->mFunc(args ..., out);
-};
-
-template<typename Out, typename ... In>
-void Async::Job<Out, In ...>::exec()
-{
-    Async::Future<InType> *in = nullptr;
-    if (mPrev) {
-        mPrev->exec();
-        in = reinterpret_cast<Async::Future<InType>*>(mPrev->mResult);
-        assert(in->isFinished());
-    }
-
-    auto out = new Async::Future<Out>();
-    Private::doExec<Out, In ...>(this, *out, in ? in->value() : In() ...);
-    out->waitForFinished();
-    mResult = reinterpret_cast<void*>(out);
-}
-
-template<typename Out, typename ... In>
-template<typename F>
-Async::Job<Out, In ...> Async::Job<Out, In ...>::create(F func, Async::JobBase::JobType jobType, Async::JobBase* parent)
-{
-    Job<Out, In ...> job(jobType, parent);
-    job.mFunc = func;
-    return job;
+    Executor *exec = new ThenExecutor<Out, In ...>(func);
+    return Job<Out, In ...>(exec);
 }
 
-
 #endif // ASYNC_H
 
 
diff --git a/async/src/future.h b/async/src/future.h
index c53ef56..eb3de1e 100644
--- a/async/src/future.h
+++ b/async/src/future.h
@@ -31,6 +31,8 @@ class FutureBase
 public:
     FutureBase();
     FutureBase(const FutureBase &other);
+    virtual ~FutureBase();
+
     void setFinished();
     bool isFinished() const;
     void waitForFinished();
-- 
cgit v1.2.3