Repository: SparseLinearAlgebra/spla (language: python)

Issue:
Expression future
- [x] Check when exression evaluated
- [x] Block until completed
- [x] Get error
- [x] Expression dependencies


Discussion / hints:

Fix patch:
diff --git a/include/spla-cpp/SplaExpression.hpp b/include/spla-cpp/SplaExpression.hpp
index 51078a2b4..08d24e079 100644
--- a/include/spla-cpp/SplaExpression.hpp
+++ b/include/spla-cpp/SplaExpression.hpp
@@ -47,11 +47,22 @@ namespace spla {
 
     /**
      * @class Expression
+     * @brief Computational expression for submission.
      *
+     * Represents single self-sufficient set of computational nodes with dependencies,
+     * which can be submitted for the execution at once and then asynchronously
+     * checked state of computation of waited (blocked) until completion.
+     *
+     * @warning Expression cannot be modified after submission.
+     * @warning Expression::Wait() is valid call only for submitted expressions.
+     *
+     * @note Call Expression::Wait() to ensure completion.
+     * @note Call Expression::GetState() to get current state to ensure, that expression is not running.
+     * @note If expression is released, it implicitly calls Expression::Wait() internally before destruction.
      */
     class SPLA_API Expression final : public Object {
     public:
-        ~Expression() override = default;
+        ~Expression() override;
 
         /** Current expression state */
         enum class State {
@@ -61,10 +72,21 @@ namespace spla {
             Submitted,
             /** Expression successfully evaluated after submission */
             Evaluated,
-            /** Expression evaluation aborted due error/exception */
+            /** Expression evaluation aborted due some error/exception */
             Aborted
         };
 
+        /**
+         * Wait for expression until it is computation is completed.
+         *
+         * @note Blocks current thread until evaluated or aborted.
+         * @note Must be called only after expression is submitted.
+         * @note Check expression state after completion to ensure correctness.
+         *
+         * @see Expression::GetState()
+         */
+        void Wait();
+
         /**
          * Makes dependency between provided expression nodes.
          * Next `succ` node will be evaluated only after `pred` node evaluation is finished.
@@ -171,9 +193,13 @@ namespace spla {
                                         const RefPtr<Descriptor> &desc);
 
         void SetState(State state);
+        void SetFuture(std::unique_ptr<class ExpressionFuture> &&future);
+        void SetTasks(std::unique_ptr<class ExpressionTasks> &&tasks);
 
         std::vector<RefPtr<class ExpressionNode>> mNodes;
-        std::atomic_long mState;
+        std::unique_ptr<class ExpressionFuture> mFuture;
+        std::unique_ptr<class ExpressionTasks> mTasks;
+        std::atomic<State> mState;
     };
 
     /**
diff --git a/include/spla-cpp/SplaRefCnt.hpp b/include/spla-cpp/SplaRefCnt.hpp
index cb1b77969..fea08b02c 100644
--- a/include/spla-cpp/SplaRefCnt.hpp
+++ b/include/spla-cpp/SplaRefCnt.hpp
@@ -65,7 +65,7 @@ namespace spla {
         }
 
         int32_t AddRef() const {
-            assert(GetRefs() > 0);
+            assert(GetRefs() >= 0);
             return mRefs.fetch_add(1);
         }
 
@@ -83,8 +83,8 @@ namespace spla {
         }
 
     private:
-        // This type of object after creation always has at least 1 reference
-        mutable std::atomic_int32_t mRefs{1};
+        // This type of object after creation always has no reference
+        mutable std::atomic_int32_t mRefs{0};
     };
 
     template<typename T>
diff --git a/sources/CMakeLists.txt b/sources/CMakeLists.txt
index e5a267f80..283c988f5 100644
--- a/sources/CMakeLists.txt
+++ b/sources/CMakeLists.txt
@@ -28,10 +28,10 @@ set(SPLA_EXPRESSION_SOURCES
         sources/expression/matrix/SplaMatrixDataWrite.hpp
         sources/expression/vector/SplaVectorDataWrite.cpp
         sources/expression/vector/SplaVectorDataWrite.hpp
-        sources/expression/SplaExpressionContext.cpp
-        sources/expression/SplaExpressionContext.hpp
+        sources/expression/SplaExpressionFuture.hpp
         sources/expression/SplaExpressionManager.cpp
         sources/expression/SplaExpressionManager.hpp
+        sources/expression/SplaExpressionTasks.hpp
         sources/expression/SplaNodeProcessor.hpp)
 
 set(SPLA_STORAGE_SOURCES
diff --git a/sources/SplaExpression.cpp b/sources/SplaExpression.cpp
index 60e258e5c..485bfd6e4 100644
--- a/sources/SplaExpression.cpp
+++ b/sources/SplaExpression.cpp
@@ -27,7 +27,18 @@
 
 #include <detail/SplaError.hpp>
 #include <detail/SplaLibraryPrivate.hpp>
-#include <spla-cpp/SplaExpression.hpp>
+#include <expression/SplaExpressionFuture.hpp>
+#include <expression/SplaExpressionTasks.hpp>
+
+spla::Expression::~Expression() {
+    // Before destruction we must wait until it is executed
+    // NOTE: if was not submitted, nothing to do
+    if (GetState() != State::Default)
+        Wait();
+
+    mFuture.reset();
+    mTasks.reset();
+}
 
 spla::RefPtr<spla::Expression> spla::Expression::Make(spla::Library &library) {
     return spla::RefPtr<spla::Expression>(new Expression(library));
@@ -37,8 +48,18 @@ spla::Expression::Expression(spla::Library &library) : Object(Object::TypeName::
     SetState(State::Default);
 }
 
+void spla::Expression::Wait() {
+    CHECK_RAISE_ERROR(GetState() != State::Default, InvalidState, "Expression must be submitted for the execution");
+
+    // NOTE: Wait only if submitted
+    if (GetState() == State::Submitted)
+        mFuture->Get().wait();
+}
+
 void spla::Expression::Dependency(const spla::RefPtr<spla::ExpressionNode> &pred,
                                   const spla::RefPtr<spla::ExpressionNode> &succ) {
+    CHECK_RAISE_ERROR(GetState() == State::Default, InvalidState, "Expression must be in default state");
+
     CHECK_RAISE_ERROR(pred.IsNotNull(), InvalidArgument, "Passed null arg");
     CHECK_RAISE_ERROR(succ.IsNotNull(), InvalidArgument, "Passed null arg");
 
@@ -65,6 +86,8 @@ spla::RefPtr<spla::ExpressionNode>
 spla::Expression::MakeNode(spla::ExpressionNode::Operation op,
                            std::vector<RefPtr<Object>> &&args,
                            const spla::RefPtr<spla::Descriptor> &desc) {
+    CHECK_RAISE_ERROR(GetState() == State::Default, InvalidState, "Expression must be in default state");
+
     for (const auto &arg : args)
         CHECK_RAISE_ERROR(arg.IsNotNull(), InvalidArgument, "Passed null argument to op=" << ExpressionNodeOpToStr(op));
 
@@ -129,5 +152,13 @@ spla::Expression::MakeDataRead(const spla::RefPtr<spla::Vector> &vector,
 }
 
 void spla::Expression::SetState(State state) {
-    mState.store(static_cast<long>(state));
+    mState.store(state);
+}
+
+void spla::Expression::SetFuture(std::unique_ptr<class ExpressionFuture> &&future) {
+    mFuture = std::move(future);
+}
+
+void spla::Expression::SetTasks(std::unique_ptr<class ExpressionTasks> &&tasks) {
+    mTasks = std::move(tasks);
 }
diff --git a/sources/detail/SplaLibraryPrivate.cpp b/sources/detail/SplaLibraryPrivate.cpp
index 39baf0b63..00cd4465c 100644
--- a/sources/detail/SplaLibraryPrivate.cpp
+++ b/sources/detail/SplaLibraryPrivate.cpp
@@ -72,7 +72,7 @@ namespace {
             sinks.push_back(fileSink);
         }
 
-        auto logger = std::make_shared<spdlog::logger>("spla-logger", sinks.begin(), sinks.end());
+        auto logger = std::make_shared<spdlog::logger>("spla", sinks.begin(), sinks.end());
         logger->set_level(static_cast<spdlog::level::level_enum>(SPDLOG_ACTIVE_LEVEL));
 
         return logger;
diff --git a/sources/expression/SplaExpressionContext.hpp b/sources/expression/SplaExpressionFuture.hpp
similarity index 75%
rename from sources/expression/SplaExpressionContext.hpp
rename to sources/expression/SplaExpressionFuture.hpp
index 8f0f73d9d..45a261851 100644
--- a/sources/expression/SplaExpressionContext.hpp
+++ b/sources/expression/SplaExpressionFuture.hpp
@@ -25,32 +25,35 @@
 /* SOFTWARE.                                                                      */
 /**********************************************************************************/
 
-#ifndef SPLA_SPLAEXPRESSIONCONTEXT_HPP
-#define SPLA_SPLAEXPRESSIONCONTEXT_HPP
+#ifndef SPLA_SPLAEXPRESSIONFUTURE_HPP
+#define SPLA_SPLAEXPRESSIONFUTURE_HPP
 
-#include <spla-cpp/SplaExpression.hpp>
-#include <spla-cpp/SplaExpressionNode.hpp>
+#include <spla-cpp/SplaRefCnt.hpp>
 #include <taskflow/taskflow.hpp>
-#include <vector>
 
 namespace spla {
 
-    class ExpressionContext {
+    /**
+     * @addtogroup Internal
+     * @{
+     */
+
+    /**
+     * @class ExpressionFuture
+     * @brief Wrapper for taskflow future.
+     */
+    class ExpressionFuture {
     public:
-        /** Associated expression */
-        RefPtr<Expression> expression;
-        /** Nodes to start evaluation */
-        std::vector<size_t> startNodes;
-        /** Final nodes */
-        std::vector<size_t> endNodes;
-        /** Order of the expression traversal to submit computation */
-        std::vector<size_t> traversal;
-        /** Expression taskflow graph */
-        tf::Taskflow taskflow;
-        /** Taskflows composed for each node */
-        std::vector<tf::Taskflow> nodesTaskflow;
+        explicit ExpressionFuture(tf::Future<void> future) : mFuture(std::move(future)) {}
+        tf::Future<void> &Get() { return mFuture; }
+    private:
+        tf::Future<void> mFuture;
     };
 
+    /**
+     * @}
+     */
+
 }// namespace spla
 
-#endif//SPLA_SPLAEXPRESSIONCONTEXT_HPP
\ No newline at end of file
+#endif//SPLA_SPLAEXPRESSIONFUTURE_HPP
diff --git a/sources/expression/SplaExpressionManager.cpp b/sources/expression/SplaExpressionManager.cpp
index fe7201fe6..f0a74601e 100644
--- a/sources/expression/SplaExpressionManager.cpp
+++ b/sources/expression/SplaExpressionManager.cpp
@@ -27,7 +27,9 @@
 
 #include <detail/SplaError.hpp>
 #include <detail/SplaLibraryPrivate.hpp>
+#include <expression/SplaExpressionFuture.hpp>
 #include <expression/SplaExpressionManager.hpp>
+#include <expression/SplaExpressionTasks.hpp>
 #include <numeric>
 #include <vector>
 
@@ -57,7 +59,7 @@ void spla::ExpressionManager::Submit(const spla::RefPtr<spla::Expression> &expre
         return;
     }
 
-    ExpressionContext context;
+    TraversalInfo context;
     context.expression = expression;
 
     FindStartNodes(context);
@@ -74,17 +76,19 @@ void spla::ExpressionManager::Submit(const spla::RefPtr<spla::Expression> &expre
     DefineTraversalPath(context);
 
     auto &traversal = context.traversal;
-    auto &taskflow = context.taskflow;
-    auto &nodesTaskflow = context.nodesTaskflow;
     auto &nodes = expression->GetNodes();
 
+    auto expressionTasks = std::make_unique<ExpressionTasks>();
+    auto &taskflow = expressionTasks->taskflow;
+    auto &nodesTaskflow = expressionTasks->nodesTaskflow;
+
     nodesTaskflow.resize(nodes.size());
 
     for (auto idx : traversal) {
         // Select processor for node
-        auto processor = SelectProcessor(idx, context);
+        auto processor = SelectProcessor(idx, *expression);
         // Actually process node
-        processor->Process(idx, context);
+        processor->Process(idx, *expression, nodesTaskflow[idx]);
     }
 
     std::vector<tf::Task> modules;
@@ -108,12 +112,20 @@ void spla::ExpressionManager::Submit(const spla::RefPtr<spla::Expression> &expre
         }
     }
 
+    // Dummy task to notify expression state
+    auto notification = taskflow.emplace([expression = expression.Get()]() {
+                                    expression->SetState(Expression::State::Evaluated);
+                                })
+                                .name("Notification");
+
+    for (auto &module : modules)
+        module.precede(notification);
+
     auto &executor = mLibrary.GetPrivate().GetTaskFlowExecutor();
-    auto future = executor.run(taskflow);
+    auto expressionFuture = std::make_unique<ExpressionFuture>(executor.run(taskflow));
 
-    // todo: check async completion
-    future.wait();
-    expression->SetState(Expression::State::Evaluated);
+    expression->SetTasks(std::move(expressionTasks));
+    expression->SetFuture(std::move(expressionFuture));
 }
 
 void spla::ExpressionManager::Register(const spla::RefPtr<spla::NodeProcessor> &processor) {
@@ -128,7 +140,7 @@ void spla::ExpressionManager::Register(const spla::RefPtr<spla::NodeProcessor> &
     list->second.push_back(processor);
 }
 
-void spla::ExpressionManager::FindStartNodes(spla::ExpressionContext &context) {
+void spla::ExpressionManager::FindStartNodes(spla::ExpressionManager::TraversalInfo &context) {
     auto &expression = context.expression;
     auto &nodes = expression->GetNodes();
     auto &start = context.startNodes;
@@ -140,7 +152,7 @@ void spla::ExpressionManager::FindStartNodes(spla::ExpressionContext &context) {
     }
 }
 
-void spla::ExpressionManager::FindEndNodes(spla::ExpressionContext &context) {
+void spla::ExpressionManager::FindEndNodes(spla::ExpressionManager::TraversalInfo &context) {
     auto &expression = context.expression;
     auto &nodes = expression->GetNodes();
     auto &end = context.endNodes;
@@ -152,7 +164,7 @@ void spla::ExpressionManager::FindEndNodes(spla::ExpressionContext &context) {
     }
 }
 
-void spla::ExpressionManager::CheckCycles(spla::ExpressionContext &context) {
+void spla::ExpressionManager::CheckCycles(spla::ExpressionManager::TraversalInfo &context) {
     auto &expression = context.expression;
     auto &nodes = expression->GetNodes();
     auto &start = context.startNodes;
@@ -181,7 +193,7 @@ bool spla::ExpressionManager::CheckCyclesImpl(size_t idx, std::vector<int> &visi
     return false;
 }
 
-void spla::ExpressionManager::DefineTraversalPath(spla::ExpressionContext &context) {
+void spla::ExpressionManager::DefineTraversalPath(spla::ExpressionManager::TraversalInfo &context) {
     auto &expression = context.expression;
     auto &nodes = expression->GetNodes();
     auto &start = context.startNodes;
@@ -229,8 +241,8 @@ void spla::ExpressionManager::DefineTraversalPathImpl(size_t idx, size_t &t, std
 }
 
 spla::RefPtr<spla::NodeProcessor>
-spla::ExpressionManager::SelectProcessor(std::size_t nodeIdx, spla::ExpressionContext &context) {
-    auto &nodes = context.expression->GetNodes();
+spla::ExpressionManager::SelectProcessor(std::size_t nodeIdx, const spla::Expression &expression) {
+    auto &nodes = expression.GetNodes();
     auto &node = nodes[nodeIdx];
     ExpressionNode::Operation op = node->GetNodeOp();
 
@@ -244,9 +256,9 @@ spla::ExpressionManager::SelectProcessor(std::size_t nodeIdx, spla::ExpressionCo
     // NOTE: Iterate through all processors for this operation and
     // select the first one, which meets requirements
     for (auto &processor : processors)
-        if (processor->Select(nodeIdx, context))
+        if (processor->Select(nodeIdx, expression))
             return processor;
 
     RAISE_ERROR(InvalidState,
-                L"Failed to find suitable processor for the node op=" << ExpressionNodeOpToStr(op));
-}
+                "Failed to find suitable processor for the node op=" << ExpressionNodeOpToStr(op));
+}
\ No newline at end of file
diff --git a/sources/expression/SplaExpressionManager.hpp b/sources/expression/SplaExpressionManager.hpp
index 820d3938a..1394716ce 100644
--- a/sources/expression/SplaExpressionManager.hpp
+++ b/sources/expression/SplaExpressionManager.hpp
@@ -28,7 +28,6 @@
 #ifndef SPLA_SPLAEXPRESSIONMANAGER_HPP
 #define SPLA_SPLAEXPRESSIONMANAGER_HPP
 
-#include <expression/SplaExpressionContext.hpp>
 #include <expression/SplaNodeProcessor.hpp>
 #include <spla-cpp/SplaExpression.hpp>
 #include <spla-cpp/SplaExpressionNode.hpp>
@@ -37,6 +36,11 @@
 
 namespace spla {
 
+    /**
+     * @addtogroup Internal
+     * @{
+     */
+
     /**
      * @brief ExpressionManager
      *
@@ -53,6 +57,18 @@ namespace spla {
      */
     class ExpressionManager final : public RefCnt {
     public:
+        /** Expression DAG structure */
+        struct TraversalInfo {
+            /** Expression */
+            RefPtr<Expression> expression;
+            /** Nodes to start evaluation */
+            std::vector<size_t> startNodes;
+            /** Final nodes */
+            std::vector<size_t> endNodes;
+            /** Order of the expression traversal to submit computation */
+            std::vector<size_t> traversal;
+        };
+
         explicit ExpressionManager(Library &library);
         ~ExpressionManager() override = default;
 
@@ -60,23 +76,27 @@ namespace spla {
         void Register(const RefPtr<NodeProcessor> &processor);
 
     private:
-        void FindStartNodes(ExpressionContext &context);
-        void FindEndNodes(ExpressionContext &context);
-        void CheckCycles(ExpressionContext &context);
+        void FindStartNodes(TraversalInfo &context);
+        void FindEndNodes(TraversalInfo &context);
+        void CheckCycles(TraversalInfo &context);
         bool CheckCyclesImpl(size_t idx, std::vector<int> &visited, const std::vector<RefPtr<ExpressionNode>> &nodes);
-        void DefineTraversalPath(ExpressionContext &context);
+        void DefineTraversalPath(TraversalInfo &context);
         void DefineTraversalPathImpl(size_t idx, size_t &t, std::vector<size_t> &out, const std::vector<RefPtr<ExpressionNode>> &nodes);
 
-        RefPtr<NodeProcessor> SelectProcessor(std::size_t nodeIdx, ExpressionContext &context);
+        RefPtr<NodeProcessor> SelectProcessor(std::size_t nodeIdx, const Expression &expression);
 
     private:
         using ProcessorList = std::vector<RefPtr<NodeProcessor>>;
         using ProcessorMap = std::unordered_map<ExpressionNode::Operation, ProcessorList>;
-        ProcessorMap mProcessors;
 
+        ProcessorMap mProcessors;
         Library &mLibrary;
     };
 
+    /**
+     * @}
+     */
+
 }// namespace spla
 
 #endif//SPLA_SPLAEXPRESSIONMANAGER_HPP
diff --git a/sources/expression/SplaExpressionContext.cpp b/sources/expression/SplaExpressionTasks.hpp
similarity index 84%
rename from sources/expression/SplaExpressionContext.cpp
rename to sources/expression/SplaExpressionTasks.hpp
index 3354145f3..ac94a5075 100644
--- a/sources/expression/SplaExpressionContext.cpp
+++ b/sources/expression/SplaExpressionTasks.hpp
@@ -25,4 +25,22 @@
 /* SOFTWARE.                                                                      */
 /**********************************************************************************/
 
-#include "SplaExpressionContext.hpp"
+#ifndef SPLA_SPLAEXPRESSIONTASKS_HPP
+#define SPLA_SPLAEXPRESSIONTASKS_HPP
+
+#include <taskflow/taskflow.hpp>
+#include <vector>
+
+namespace spla {
+
+    class ExpressionTasks {
+    public:
+        /** Expression taskflow graph */
+        tf::Taskflow taskflow;
+        /** Taskflows composed for each node */
+        std::vector<tf::Taskflow> nodesTaskflow;
+    };
+
+}
+
+#endif//SPLA_SPLAEXPRESSIONTASKS_HPP
diff --git a/sources/expression/SplaNodeProcessor.hpp b/sources/expression/SplaNodeProcessor.hpp
index abc52e281..f6351bbfb 100644
--- a/sources/expression/SplaNodeProcessor.hpp
+++ b/sources/expression/SplaNodeProcessor.hpp
@@ -28,9 +28,10 @@
 #ifndef SPLA_SPLANODEPROCESSOR_HPP
 #define SPLA_SPLANODEPROCESSOR_HPP
 
-#include <expression/SplaExpressionContext.hpp>
+#include <spla-cpp/SplaExpression.hpp>
 #include <spla-cpp/SplaExpressionNode.hpp>
 #include <spla-cpp/SplaRefCnt.hpp>
+#include <taskflow/taskflow.hpp>
 
 namespace spla {
 
@@ -59,20 +60,19 @@ namespace spla {
          * be used for processing this node (if more then one is available).
          *
          * @note Node passed as index in the expression.
-         * @note Expression related info is stored in the context.
          *
          * @param nodeIdx Index of the node in expression
-         * @param context Expression context
+         * @param expression Expression being processed
          *
          * @return True if this node can process this node
          */
-        virtual bool Select(std::size_t nodeIdx, ExpressionContext &context) = 0;
+        virtual bool Select(std::size_t nodeIdx, const Expression &expression) = 0;
 
         /**
          * @brief Process specified node in the expression.
          *
          * Node processor is supposed to construct execution task graph and store it
-         * in the context at specified node index. No actual computation must happen
+         * in the taskflow of the node. No actual computation must happen
          * or be triggered within Process method call. Only when all nodes are processed,
          * result task graph is submitted for the execution.
          *
@@ -80,9 +80,10 @@ namespace spla {
          * @note Expression related info is stored in the context.
          *
          * @param nodeIdx Index of the node in expression
-         * @param context Expression context
+         * @param expression Expression being processed
+         * @param taskflow Taskflow graph of node nodeIdx
          */
-        virtual void Process(std::size_t nodeIdx, ExpressionContext &context) = 0;
+        virtual void Process(std::size_t nodeIdx, const Expression &expression, tf::Taskflow &taskflow) = 0;
 
         /** @return Type of the expression node operation, handled by this processor */
         virtual ExpressionNode::Operation GetOperationType() const = 0;
diff --git a/sources/expression/matrix/SplaMatrixDataRead.cpp b/sources/expression/matrix/SplaMatrixDataRead.cpp
index 9b0bfb747..15cbcfd14 100644
--- a/sources/expression/matrix/SplaMatrixDataRead.cpp
+++ b/sources/expression/matrix/SplaMatrixDataRead.cpp
@@ -46,15 +46,14 @@ namespace spla {
     }// namespace
 }// namespace spla
 
-bool spla::MatrixDataRead::Select(std::size_t nodeIdx, spla::ExpressionContext &context) {
+bool spla::MatrixDataRead::Select(std::size_t nodeIdx, const spla::Expression &expression) {
     return true;
 }
 
-void spla::MatrixDataRead::Process(std::size_t nodeIdx, spla::ExpressionContext &context) {
-    auto &taskflow = context.nodesTaskflow[nodeIdx];
-    auto &nodes = context.expression->GetNodes();
+void spla::MatrixDataRead::Process(std::size_t nodeIdx, const spla::Expression &expression, tf::Taskflow &taskflow) {
+    auto &nodes = expression.GetNodes();
     auto node = nodes[nodeIdx];
-    auto library = context.expression->GetLibrary().GetPrivatePtr();
+    auto library = expression.GetLibrary().GetPrivatePtr();
     auto logger = library->GetLogger();
 
     auto matrix = node->GetArg(0).Cast<Matrix>();
diff --git a/sources/expression/matrix/SplaMatrixDataRead.hpp b/sources/expression/matrix/SplaMatrixDataRead.hpp
index 20ad45ffe..61c5e1466 100644
--- a/sources/expression/matrix/SplaMatrixDataRead.hpp
+++ b/sources/expression/matrix/SplaMatrixDataRead.hpp
@@ -35,8 +35,8 @@ namespace spla {
     class MatrixDataRead final : public NodeProcessor {
     public:
         ~MatrixDataRead() override = default;
-        bool Select(std::size_t nodeIdx, ExpressionContext &context) override;
-        void Process(std::size_t nodeIdx, ExpressionContext &context) override;
+        bool Select(std::size_t nodeIdx, const Expression &expression) override;
+        void Process(std::size_t nodeIdx, const Expression &expression, tf::Taskflow &taskflow) override;
         ExpressionNode::Operation GetOperationType() const override;
     };
 
diff --git a/sources/expression/matrix/SplaMatrixDataWrite.cpp b/sources/expression/matrix/SplaMatrixDataWrite.cpp
index 176be6ceb..e1c2f7057 100644
--- a/sources/expression/matrix/SplaMatrixDataWrite.cpp
+++ b/sources/expression/matrix/SplaMatrixDataWrite.cpp
@@ -34,15 +34,14 @@
 #include <storage/block/SplaMatrixCOO.hpp>
 #include <vector>
 
-bool spla::MatrixDataWrite::Select(std::size_t nodeIdx, spla::ExpressionContext &context) {
+bool spla::MatrixDataWrite::Select(std::size_t nodeIdx, const spla::Expression &expression) {
     return true;
 }
 
-void spla::MatrixDataWrite::Process(std::size_t nodeIdx, spla::ExpressionContext &context) {
-    auto &taskflow = context.nodesTaskflow[nodeIdx];
-    auto &nodes = context.expression->GetNodes();
+void spla::MatrixDataWrite::Process(std::size_t nodeIdx, const spla::Expression &expression, tf::Taskflow &taskflow) {
+    auto &nodes = expression.GetNodes();
     auto node = nodes[nodeIdx];
-    auto library = context.expression->GetLibrary().GetPrivatePtr();
+    auto library = expression.GetLibrary().GetPrivatePtr();
     auto logger = library->GetLogger();
 
     auto matrix = node->GetArg(0).Cast<Matrix>();
diff --git a/sources/expression/matrix/SplaMatrixDataWrite.hpp b/sources/expression/matrix/SplaMatrixDataWrite.hpp
index 8a6a06abe..938c60f61 100644
--- a/sources/expression/matrix/SplaMatrixDataWrite.hpp
+++ b/sources/expression/matrix/SplaMatrixDataWrite.hpp
@@ -35,8 +35,8 @@ namespace spla {
     class MatrixDataWrite final : public NodeProcessor {
     public:
         ~MatrixDataWrite() override = default;
-        bool Select(std::size_t nodeIdx, ExpressionContext &context) override;
-        void Process(std::size_t nodeIdx, ExpressionContext &context) override;
+        bool Select(std::size_t nodeIdx, const Expression &expression) override;
+        void Process(std::size_t nodeIdx, const Expression &expression, tf::Taskflow &taskflow) override;
         ExpressionNode::Operation GetOperationType() const override;
     };
 
diff --git a/sources/expression/vector/SplaVectorDataWrite.cpp b/sources/expression/vector/SplaVectorDataWrite.cpp
index 8897ee7a6..bf3ac7de3 100644
--- a/sources/expression/vector/SplaVectorDataWrite.cpp
+++ b/sources/expression/vector/SplaVectorDataWrite.cpp
@@ -28,14 +28,13 @@
 #include <detail/SplaLibraryPrivate.hpp>
 #include <expression/vector/SplaVectorDataWrite.hpp>
 
-bool spla::VectorDataWrite::Select(std::size_t nodeIdx, spla::ExpressionContext &context) {
+bool spla::VectorDataWrite::Select(std::size_t nodeIdx, const spla::Expression &expression) {
     return true;
 }
 
-void spla::VectorDataWrite::Process(std::size_t nodeIdx, spla::ExpressionContext &context) {
+void spla::VectorDataWrite::Process(std::size_t nodeIdx, const spla::Expression &expression, tf::Taskflow &taskflow) {
     // todo: impl me!
-    auto &taskflow = context.nodesTaskflow[nodeIdx];
-    auto logger = context.expression->GetLibrary().GetPrivate().GetLogger();
+    auto logger = expression.GetLibrary().GetPrivate().GetLogger();
 
     // todo: remove it!
     taskflow.emplace([=]() {
diff --git a/sources/expression/vector/SplaVectorDataWrite.hpp b/sources/expression/vector/SplaVectorDataWrite.hpp
index 8d89d7310..6fb13b245 100644
--- a/sources/expression/vector/SplaVectorDataWrite.hpp
+++ b/sources/expression/vector/SplaVectorDataWrite.hpp
@@ -35,8 +35,8 @@ namespace spla {
     class VectorDataWrite final : public NodeProcessor {
     public:
         ~VectorDataWrite() override = default;
-        bool Select(std::size_t nodeIdx, ExpressionContext &context) override;
-        void Process(std::size_t nodeIdx, ExpressionContext &context) override;
+        bool Select(std::size_t nodeIdx, const Expression &expression) override;
+        void Process(std::size_t nodeIdx, const Expression &expression, tf::Taskflow &taskflow) override;
         ExpressionNode::Operation GetOperationType() const override;
     };
 


Test patch:
diff --git a/tests/TestDataMatrix.cpp b/tests/TestDataMatrix.cpp
index ba7ce230a..42b570aab 100644
--- a/tests/TestDataMatrix.cpp
+++ b/tests/TestDataMatrix.cpp
@@ -45,6 +45,7 @@ static void testCommon(spla::Library &library, size_t M, size_t N, size_t nvals,
     auto spWrite = spExprWrite->MakeDataWrite(spM, spDataSrc, spDesc);
 
     library.Submit(spExprWrite);
+    spExprWrite->Wait();
     EXPECT_EQ(spExprWrite->GetState(), spla::Expression::State::Evaluated);
 
     utils::Matrix<float> expected = source.SortReduceDuplicates();
@@ -73,6 +74,7 @@ static void testSortedNoDuplicates(spla::Library &library, size_t M, size_t N, s
     auto spWrite = spExprWrite->MakeDataWrite(spM, spDataSrc, spDesc);
 
     library.Submit(spExprWrite);
+    spExprWrite->Wait();
     EXPECT_EQ(spExprWrite->GetState(), spla::Expression::State::Evaluated);
 
     utils::Matrix<float> &expected = source;
diff --git a/tests/TestDataVector.cpp b/tests/TestDataVector.cpp
index 6709354ab..986e8dc4f 100644
--- a/tests/TestDataVector.cpp
+++ b/tests/TestDataVector.cpp
@@ -44,6 +44,7 @@ static void testCommon(spla::Library &library, size_t M, size_t nvals, size_t se
     auto spWrite = spExprWrite->MakeDataWrite(spV, spDataSrc, spDesc);
 
     library.Submit(spExprWrite);
+    spExprWrite->Wait();
     EXPECT_EQ(spExprWrite->GetState(), spla::Expression::State::Evaluated);
 
     utils::Vector expected = source.SortReduceDuplicates();
@@ -72,6 +73,7 @@ static void testSortedNoDuplicates(spla::Library &library, size_t M, size_t nval
     auto spWrite = spExprWrite->MakeDataWrite(spV, spDataSrc, spDesc);
 
     library.Submit(spExprWrite);
+    spExprWrite->Wait();
     EXPECT_EQ(spExprWrite->GetState(), spla::Expression::State::Evaluated);
 
     utils::Vector<float> &expected = source;
diff --git a/tests/utils/Matrix.hpp b/tests/utils/Matrix.hpp
index 0f8e33284..2dc88f5bb 100644
--- a/tests/utils/Matrix.hpp
+++ b/tests/utils/Matrix.hpp
@@ -122,6 +122,8 @@ namespace utils {
             readExpr->MakeDataRead(m, data);
             library.Submit(readExpr);
 
+            readExpr->Wait();
+
             if (readExpr->GetState() != spla::Expression::State::Evaluated)
                 return false;
 
diff --git a/tests/utils/Vector.hpp b/tests/utils/Vector.hpp
index 1ceaef8ff..079714aac 100644
--- a/tests/utils/Vector.hpp
+++ b/tests/utils/Vector.hpp
@@ -105,6 +105,8 @@ namespace utils {
             readExpr->MakeDataRead(v, data);
             library.Submit(readExpr);
 
+            readExpr->Wait();
+
             if (readExpr->GetState() != spla::Expression::State::Evaluated)
                 return false;