Repository: SparseLinearAlgebra/spla (language: python)

Issue:
Task builder
Custom task builder (based on taskflow) for expression nodes processing task graph composition:

## Motivation

- Abstact task creation
- Implicit error/exception handling
- Notify expresion owner about error
- Cancel graph if error occurs
- Use c++ exceptions inside tasks (safely)

## What to do

- [ ] Task builder class
- [ ] Emplace task method
- [ ] Get final task graph (taskflow)


Discussion / hints:
Hi @EgorOrachyov , I noticed you are using taskflow to SPLA - which is something I have been thinking about in the past year. Please feel free to [contact me](tsung-wei.huang@utah.edu) with any questions or collaboration :)
@tsung-wei-huang Hi, thank you for your response. Yeah, we've decided to use `taskflow` to build task graphs from math expressions. It seems, that `taskflow` can fulfill all our needs!

Fix patch:
diff --git a/include/spla-cpp/SplaExpression.hpp b/include/spla-cpp/SplaExpression.hpp
index 08d24e079..15f7b8b5a 100644
--- a/include/spla-cpp/SplaExpression.hpp
+++ b/include/spla-cpp/SplaExpression.hpp
@@ -186,6 +186,7 @@ namespace spla {
 
     private:
         friend class ExpressionManager;
+        friend class TaskBuilder;
         explicit Expression(class Library &library);
 
         RefPtr<ExpressionNode> MakeNode(ExpressionNode::Operation op,
diff --git a/sources/CMakeLists.txt b/sources/CMakeLists.txt
index 78c5ec43f..ac3dba6ec 100644
--- a/sources/CMakeLists.txt
+++ b/sources/CMakeLists.txt
@@ -28,7 +28,9 @@ set(SPLA_CORE_SOURCES
         sources/core/SplaHash.hpp
         sources/core/SplaLibraryPrivate.cpp
         sources/core/SplaLibraryPrivate.hpp
-        sources/core/SplaMath.hpp)
+        sources/core/SplaMath.hpp
+        sources/core/SplaTaskBuilder.cpp
+        sources/core/SplaTaskBuilder.hpp)
 
 set(SPLA_EXPRESSION_SOURCES
         sources/expression/matrix/SplaMatrixDataRead.cpp
diff --git a/sources/algo/SplaAlgorithmManager.cpp b/sources/algo/SplaAlgorithmManager.cpp
index beaa6548e..f6040ce81 100644
--- a/sources/algo/SplaAlgorithmManager.cpp
+++ b/sources/algo/SplaAlgorithmManager.cpp
@@ -50,10 +50,10 @@ void spla::AlgorithmManager::Dispatch(spla::Algorithm::Type type, const spla::Re
     algorithm->Process(*params);
 }
 
-tf::Task spla::AlgorithmManager::Dispatch(spla::Algorithm::Type type, const spla::RefPtr<spla::AlgorithmParams> &params, tf::Taskflow &taskflow) {
+tf::Task spla::AlgorithmManager::Dispatch(spla::Algorithm::Type type, const spla::RefPtr<spla::AlgorithmParams> &params, TaskBuilder &builder) {
     assert(params.IsNotNull());
     auto algorithm = SelectAlgorithm(type, params);
-    return taskflow.emplace([=]() {
+    return builder.Emplace([=]() {
         algorithm->Process(*params);
     });
 }
diff --git a/sources/algo/SplaAlgorithmManager.hpp b/sources/algo/SplaAlgorithmManager.hpp
index 311863392..dd612410b 100644
--- a/sources/algo/SplaAlgorithmManager.hpp
+++ b/sources/algo/SplaAlgorithmManager.hpp
@@ -29,8 +29,8 @@
 #define SPLA_SPLAALGORITHMMANAGER_HPP
 
 #include <algo/SplaAlgorithm.hpp>
+#include <core/SplaTaskBuilder.hpp>
 #include <spla-cpp/SplaRefCnt.hpp>
-#include <taskflow/taskflow.hpp>
 #include <unordered_map>
 #include <vector>
 
@@ -59,7 +59,7 @@ namespace spla {
 
         void Register(const RefPtr<Algorithm> &algo);
         void Dispatch(Algorithm::Type type, const RefPtr<AlgorithmParams> &params);
-        tf::Task Dispatch(Algorithm::Type type, const RefPtr<AlgorithmParams> &params, tf::Taskflow &taskflow);
+        tf::Task Dispatch(Algorithm::Type type, const RefPtr<AlgorithmParams> &params, TaskBuilder &builder);
 
     private:
         RefPtr<Algorithm> SelectAlgorithm(Algorithm::Type type, const RefPtr<AlgorithmParams> &params);
diff --git a/sources/core/SplaError.hpp b/sources/core/SplaError.hpp
index 018d03346..38fc9b184 100644
--- a/sources/core/SplaError.hpp
+++ b/sources/core/SplaError.hpp
@@ -34,6 +34,7 @@
 
 namespace spla {
 
+    /** Possible error status */
     enum class Status {
         Error,
         DeviceError,
@@ -50,7 +51,7 @@ namespace spla {
      */
     class Exception : public std::exception {
     public:
-        Exception(std::string message, std::string function, std::string file, size_t line, Status status, bool critical)
+        Exception(std::string message, std::string function, std::string file, std::size_t line, Status status, bool critical)
             : std::exception(),
               mMessage(std::move(message)),
               mFunction(std::move(function)),
@@ -69,9 +70,10 @@ namespace spla {
                 mCached = true;
 
                 std::stringstream s;
-                s << "Exception in" << GetFile()
-                  << ": line: " << GetLine()
-                  << " function: " << GetFunction();
+                s << mMessage << " "
+                  << "in: " << GetFile() << " "
+                  << "line: " << GetLine() << " "
+                  << "function: " << GetFunction();
 
                 mWhatCached = s.str();
             }
@@ -91,7 +93,7 @@ namespace spla {
             return mFile;
         }
 
-        size_t GetLine() const {
+        std::size_t GetLine() const {
             return mLine;
         }
 
@@ -107,7 +109,7 @@ namespace spla {
         std::string mMessage;
         std::string mFunction;
         std::string mFile;
-        size_t mLine;
+        std::size_t mLine;
         Status mStatus;
         bool mCritical;
 
@@ -122,7 +124,7 @@ namespace spla {
     template<Status status>
     class TException : public Exception {
     public:
-        TException(std::string message, std::string &&function, std::string &&file, size_t line, bool critical)
+        TException(std::string message, std::string &&function, std::string &&file, std::size_t line, bool critical)
             : Exception(std::move(message), std::move(function), std::move(file), line, status, critical) {
         }
 
diff --git a/sources/core/SplaMath.hpp b/sources/core/SplaMath.hpp
index 6973d6125..ae7283c76 100644
--- a/sources/core/SplaMath.hpp
+++ b/sources/core/SplaMath.hpp
@@ -30,23 +30,19 @@
 
 #include <cstddef>
 
-namespace spla {
-
-    namespace math {
-
-        static inline size_t GetBlocksCount(size_t dim, size_t blockSize) {
+namespace spla::math {
+    namespace {
+        inline std::size_t GetBlocksCount(std::size_t dim, std::size_t blockSize) {
             auto rest = dim % blockSize;
             return dim / blockSize + (rest ? 1 : 0);
         }
 
-        static inline size_t GetBlockActualSize(size_t blockIdx, size_t dim, size_t blockSize) {
+        inline std::size_t GetBlockActualSize(std::size_t blockIdx, std::size_t dim, std::size_t blockSize) {
             auto first = blockIdx * blockSize;
             auto size = dim - first;
             return size < blockSize ? size : blockSize;
         }
-
-    }// namespace math
-
-}// namespace spla
+    }// namespace
+}// namespace spla::math
 
 #endif//SPLA_SPLAMATH_HPP
diff --git a/sources/core/SplaTaskBuilder.cpp b/sources/core/SplaTaskBuilder.cpp
new file mode 100644
index 000000000..d246c4968
--- /dev/null
+++ b/sources/core/SplaTaskBuilder.cpp
@@ -0,0 +1,53 @@
+/**********************************************************************************/
+/* This file is part of spla project                                              */
+/* https://github.com/JetBrains-Research/spla                                     */
+/**********************************************************************************/
+/* MIT License                                                                    */
+/*                                                                                */
+/* Copyright (c) 2021 JetBrains-Research                                          */
+/*                                                                                */
+/* Permission is hereby granted, free of charge, to any person obtaining a copy   */
+/* of this software and associated documentation files (the "Software"), to deal  */
+/* in the Software without restriction, including without limitation the rights   */
+/* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell      */
+/* copies of the Software, and to permit persons to whom the Software is          */
+/* furnished to do so, subject to the following conditions:                       */
+/*                                                                                */
+/* The above copyright notice and this permission notice shall be included in all */
+/* copies or substantial portions of the Software.                                */
+/*                                                                                */
+/* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR     */
+/* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,       */
+/* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE    */
+/* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER         */
+/* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,  */
+/* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE  */
+/* SOFTWARE.                                                                      */
+/**********************************************************************************/
+
+#include <core/SplaError.hpp>
+#include <core/SplaLibraryPrivate.hpp>
+#include <core/SplaTaskBuilder.hpp>
+#include <spdlog/spdlog.h>
+
+tf::Task spla::TaskBuilder::Emplace(std::function<void()> work) {
+    auto task = [expression = mExpression, work = std::move(work)]() {
+        // If some error occurred earlier, no sense to continue
+        if (expression->GetState() == Expression::State::Aborted)
+            return;
+
+        try {
+            work();
+        } catch (std::exception &ex) {
+            expression->SetState(Expression::State::Aborted);
+            auto logger = expression->GetLibrary().GetPrivate().GetLogger();
+            SPDLOG_LOGGER_ERROR(logger, "Error inside expression node task. {}", ex.what());
+        }
+    };
+
+    return mSubflow.emplace(std::move(task));
+}
+
+spla::TaskBuilder::TaskBuilder(spla::Expression *expression, tf::Subflow &subflow)
+    : mExpression(expression), mSubflow(subflow) {
+}
\ No newline at end of file
diff --git a/sources/core/SplaTaskBuilder.hpp b/sources/core/SplaTaskBuilder.hpp
new file mode 100644
index 000000000..5f5eb55d2
--- /dev/null
+++ b/sources/core/SplaTaskBuilder.hpp
@@ -0,0 +1,75 @@
+/**********************************************************************************/
+/* This file is part of spla project                                              */
+/* https://github.com/JetBrains-Research/spla                                     */
+/**********************************************************************************/
+/* MIT License                                                                    */
+/*                                                                                */
+/* Copyright (c) 2021 JetBrains-Research                                          */
+/*                                                                                */
+/* Permission is hereby granted, free of charge, to any person obtaining a copy   */
+/* of this software and associated documentation files (the "Software"), to deal  */
+/* in the Software without restriction, including without limitation the rights   */
+/* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell      */
+/* copies of the Software, and to permit persons to whom the Software is          */
+/* furnished to do so, subject to the following conditions:                       */
+/*                                                                                */
+/* The above copyright notice and this permission notice shall be included in all */
+/* copies or substantial portions of the Software.                                */
+/*                                                                                */
+/* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR     */
+/* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,       */
+/* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE    */
+/* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER         */
+/* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,  */
+/* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE  */
+/* SOFTWARE.                                                                      */
+/**********************************************************************************/
+
+#ifndef SPLA_SPLATASKBUILDER_HPP
+#define SPLA_SPLATASKBUILDER_HPP
+
+#include <functional>
+#include <spla-cpp/SplaExpression.hpp>
+#include <spla-cpp/SplaExpressionNode.hpp>
+#include <taskflow/taskflow.hpp>
+
+namespace spla {
+
+    /**
+     * @addtogroup Internal
+     * @{
+     */
+
+    /**
+     * @class TaskBuilder
+     * @brief Task builder for expression nodes.
+     *
+     * Allows to compose dynamically subflows inside expression node
+     * task flow graph. Automates exception handling, expression owner notification
+     * about errors, and cancellation of the computation if error occurs.
+     */
+    class TaskBuilder {
+    public:
+        /**
+         * Emplace work to the subflow.
+         *
+         * @param work Function to execute as work inside task.
+         * @return Taskflow task handle.
+         */
+        tf::Task Emplace(std::function<void()> work);
+
+    private:
+        friend class ExpressionManager;
+        TaskBuilder(Expression *expression, tf::Subflow &subflow);
+
+        Expression *mExpression;
+        tf::Subflow &mSubflow;
+    };
+
+    /**
+     * @}
+     */
+
+}// namespace spla
+
+#endif//SPLA_SPLATASKBUILDER_HPP
diff --git a/sources/expression/SplaExpressionManager.cpp b/sources/expression/SplaExpressionManager.cpp
index 82fbb82f9..27ebc729a 100644
--- a/sources/expression/SplaExpressionManager.cpp
+++ b/sources/expression/SplaExpressionManager.cpp
@@ -27,6 +27,7 @@
 
 #include <core/SplaError.hpp>
 #include <core/SplaLibraryPrivate.hpp>
+#include <core/SplaTaskBuilder.hpp>
 #include <expression/SplaExpressionFuture.hpp>
 #include <expression/SplaExpressionManager.hpp>
 #include <expression/SplaExpressionTasks.hpp>
@@ -82,26 +83,35 @@ void spla::ExpressionManager::Submit(const spla::RefPtr<spla::Expression> &expre
 
     auto expressionTasks = std::make_unique<ExpressionTasks>();
     auto &taskflow = expressionTasks->taskflow;
-    auto &nodesTaskflow = expressionTasks->nodesTaskflow;
 
-    nodesTaskflow.resize(nodes.size());
+    std::vector<tf::Task> modules;
+    modules.reserve(nodes.size());
 
     for (auto idx : traversal) {
         // Select processor for node
         auto processor = SelectProcessor(idx, *expression);
-        // Actually process node
-        processor->Process(idx, *expression, nodesTaskflow[idx]);
-    }
-
-    std::vector<tf::Task> modules;
-    modules.reserve(nodes.size());
-
-    for (auto &nodeTaskflow : nodesTaskflow) {
-        // Build temporary modules to sync sub-flows
-        modules.push_back(taskflow.composed_of(nodeTaskflow));
+        // Wrap processor into task to handle dynamic changes of expression nodes params
+        auto task = [idx, expression = expression.Get(), processor = processor.Get()](tf::Subflow &subflow) {
+            // If aborted in previous tasks, candle run
+            if (expression->GetState() == Expression::State::Aborted)
+                return;
+
+            // Task build wraps error handling and abortion
+            TaskBuilder taskBuilder(expression, subflow);
+
+            try {
+                // Actual task graph composition
+                processor->Process(idx, *expression, taskBuilder);
+            } catch (std::exception &ex) {
+                expression->SetState(Expression::State::Aborted);
+                auto logger = expression->GetLibrary().GetPrivate().GetLogger();
+                SPDLOG_LOGGER_ERROR(logger, "Error inside expression node. {}", ex.what());
+            }
+        };
+        modules.push_back(taskflow.emplace(std::move(task)));
     }
 
-    for (size_t idx : traversal) {
+    for (std::size_t idx : traversal) {
         // Compose final taskflow graph
         auto &node = nodes[idx];
         auto &next = node->GetNext();
@@ -116,12 +126,14 @@ 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);
+                                    if (expression->GetState() != Expression::State::Aborted)
+                                        expression->SetState(Expression::State::Evaluated);
                                 })
                                 .name("Notification");
 
-    for (auto &module : modules)
-        module.precede(notification);
+    // Add dependency for end nodes
+    for (auto end : context.endNodes)
+        modules[end].precede(notification);
 
     auto &executor = mLibrary.GetPrivate().GetTaskFlowExecutor();
     auto expressionFuture = std::make_unique<ExpressionFuture>(executor.run(taskflow));
diff --git a/sources/expression/SplaExpressionTasks.hpp b/sources/expression/SplaExpressionTasks.hpp
index 672528df6..cf22de677 100644
--- a/sources/expression/SplaExpressionTasks.hpp
+++ b/sources/expression/SplaExpressionTasks.hpp
@@ -42,8 +42,6 @@ namespace spla {
     public:
         /** Expression taskflow graph */
         tf::Taskflow taskflow;
-        /** Taskflows composed for each node */
-        std::vector<tf::Taskflow> nodesTaskflow;
     };
 
     /**
diff --git a/sources/expression/SplaNodeProcessor.hpp b/sources/expression/SplaNodeProcessor.hpp
index f6351bbfb..6498fba0e 100644
--- a/sources/expression/SplaNodeProcessor.hpp
+++ b/sources/expression/SplaNodeProcessor.hpp
@@ -28,10 +28,10 @@
 #ifndef SPLA_SPLANODEPROCESSOR_HPP
 #define SPLA_SPLANODEPROCESSOR_HPP
 
+#include <core/SplaTaskBuilder.hpp>
 #include <spla-cpp/SplaExpression.hpp>
 #include <spla-cpp/SplaExpressionNode.hpp>
 #include <spla-cpp/SplaRefCnt.hpp>
-#include <taskflow/taskflow.hpp>
 
 namespace spla {
 
@@ -81,9 +81,9 @@ namespace spla {
          *
          * @param nodeIdx Index of the node in expression
          * @param expression Expression being processed
-         * @param taskflow Taskflow graph of node nodeIdx
+         * @param builder TaskBuilder for node tasks
          */
-        virtual void Process(std::size_t nodeIdx, const Expression &expression, tf::Taskflow &taskflow) = 0;
+        virtual void Process(std::size_t nodeIdx, const Expression &expression, TaskBuilder &builder) = 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 d58418abe..7cb458c93 100644
--- a/sources/expression/matrix/SplaMatrixDataRead.cpp
+++ b/sources/expression/matrix/SplaMatrixDataRead.cpp
@@ -50,7 +50,7 @@ bool spla::MatrixDataRead::Select(std::size_t nodeIdx, const spla::Expression &e
     return true;
 }
 
-void spla::MatrixDataRead::Process(std::size_t nodeIdx, const spla::Expression &expression, tf::Taskflow &taskflow) {
+void spla::MatrixDataRead::Process(std::size_t nodeIdx, const spla::Expression &expression, spla::TaskBuilder &builder) {
     auto &nodes = expression.GetNodes();
     auto node = nodes[nodeIdx];
     auto library = expression.GetLibrary().GetPrivatePtr();
@@ -77,7 +77,7 @@ void spla::MatrixDataRead::Process(std::size_t nodeIdx, const spla::Expression &
                           "Supported only COO matrix block format");
     }
 
-    auto collectNnz = taskflow.emplace([=]() {
+    auto collectNnz = builder.Emplace([=]() {
         auto &blockRowsNvals = shared->blockRowsNvals;
         auto &blockRowsOffsets = shared->blockRowsOffsets;
 
@@ -97,7 +97,7 @@ void spla::MatrixDataRead::Process(std::size_t nodeIdx, const spla::Expression &
     });
 
     for (std::size_t i = 0; i < storage->GetNblockRows(); i++) {
-        auto copyBlocksInRow = taskflow.emplace([=]() {
+        auto copyBlocksInRow = builder.Emplace([=]() {
             using namespace boost;
 
             // todo: gpu and device queue management
diff --git a/sources/expression/matrix/SplaMatrixDataRead.hpp b/sources/expression/matrix/SplaMatrixDataRead.hpp
index 61c5e1466..d3327df9d 100644
--- a/sources/expression/matrix/SplaMatrixDataRead.hpp
+++ b/sources/expression/matrix/SplaMatrixDataRead.hpp
@@ -36,7 +36,7 @@ namespace spla {
     public:
         ~MatrixDataRead() override = default;
         bool Select(std::size_t nodeIdx, const Expression &expression) override;
-        void Process(std::size_t nodeIdx, const Expression &expression, tf::Taskflow &taskflow) override;
+        void Process(std::size_t nodeIdx, const Expression &expression, TaskBuilder &builder) override;
         ExpressionNode::Operation GetOperationType() const override;
     };
 
diff --git a/sources/expression/matrix/SplaMatrixDataWrite.cpp b/sources/expression/matrix/SplaMatrixDataWrite.cpp
index ff6ec517f..8a6a58610 100644
--- a/sources/expression/matrix/SplaMatrixDataWrite.cpp
+++ b/sources/expression/matrix/SplaMatrixDataWrite.cpp
@@ -38,7 +38,7 @@ bool spla::MatrixDataWrite::Select(std::size_t nodeIdx, const spla::Expression &
     return true;
 }
 
-void spla::MatrixDataWrite::Process(std::size_t nodeIdx, const spla::Expression &expression, tf::Taskflow &taskflow) {
+void spla::MatrixDataWrite::Process(std::size_t nodeIdx, const spla::Expression &expression, spla::TaskBuilder &builder) {
     auto &nodes = expression.GetNodes();
     auto node = nodes[nodeIdx];
     auto library = expression.GetLibrary().GetPrivatePtr();
@@ -64,7 +64,7 @@ void spla::MatrixDataWrite::Process(std::size_t nodeIdx, const spla::Expression
     for (size_t i = 0; i < blocksCountInRow; i++) {
         for (size_t j = 0; j < blocksCountInCol; j++) {
             auto deviceId = devicesIds[i * blocksCountInCol + j];
-            taskflow.emplace([=]() {
+            builder.Emplace([=]() {
                 using namespace boost;
 
                 compute::context ctx = library->GetContext();
diff --git a/sources/expression/matrix/SplaMatrixDataWrite.hpp b/sources/expression/matrix/SplaMatrixDataWrite.hpp
index 938c60f61..3731cd226 100644
--- a/sources/expression/matrix/SplaMatrixDataWrite.hpp
+++ b/sources/expression/matrix/SplaMatrixDataWrite.hpp
@@ -36,7 +36,7 @@ namespace spla {
     public:
         ~MatrixDataWrite() override = default;
         bool Select(std::size_t nodeIdx, const Expression &expression) override;
-        void Process(std::size_t nodeIdx, const Expression &expression, tf::Taskflow &taskflow) override;
+        void Process(std::size_t nodeIdx, const Expression &expression, TaskBuilder &builder) override;
         ExpressionNode::Operation GetOperationType() const override;
     };
 
diff --git a/sources/expression/vector/SplaVectorDataRead.cpp b/sources/expression/vector/SplaVectorDataRead.cpp
index 3ef68179e..45d114d55 100644
--- a/sources/expression/vector/SplaVectorDataRead.cpp
+++ b/sources/expression/vector/SplaVectorDataRead.cpp
@@ -46,11 +46,11 @@ namespace spla {
     }// namespace
 }// namespace spla
 
-bool spla::VectorDataRead::Select(std::size_t nodeIdx, const Expression &expression) {
+bool spla::VectorDataRead::Select(std::size_t nodeIdx, const spla::Expression &expression) {
     return true;
 }
 
-void spla::VectorDataRead::Process(std::size_t nodeIdx, const Expression &expression, tf::Taskflow &taskflow) {
+void spla::VectorDataRead::Process(std::size_t nodeIdx, const spla::Expression &expression, spla::TaskBuilder &builder) {
     auto &nodes = expression.GetNodes();
     auto node = nodes[nodeIdx];
     auto library = expression.GetLibrary().GetPrivatePtr();
@@ -77,7 +77,7 @@ void spla::VectorDataRead::Process(std::size_t nodeIdx, const Expression &expres
                           "Supported only COO vector block format");
     }
 
-    auto collectNnz = taskflow.emplace([=]() {
+    auto collectNnz = builder.Emplace([=]() {
         auto &blockRowsNvals = shared->blockRowsNvals;
         auto &blockRowsOffsets = shared->blockRowsOffsets;
 
@@ -98,7 +98,7 @@ void spla::VectorDataRead::Process(std::size_t nodeIdx, const Expression &expres
 
     for (std::size_t i = 0; i < storage->GetNblockRows(); i++) {
         tf::Task copyBlocksInRow;
-        copyBlocksInRow = taskflow.emplace([=]() {
+        copyBlocksInRow = builder.Emplace([=]() {
             using namespace boost;
 
             // todo: gpu and device queue management
diff --git a/sources/expression/vector/SplaVectorDataRead.hpp b/sources/expression/vector/SplaVectorDataRead.hpp
index f834e567d..eadf59fda 100644
--- a/sources/expression/vector/SplaVectorDataRead.hpp
+++ b/sources/expression/vector/SplaVectorDataRead.hpp
@@ -36,7 +36,7 @@ namespace spla {
     public:
         ~VectorDataRead() override = default;
         bool Select(std::size_t nodeIdx, const Expression &expression) override;
-        void Process(std::size_t nodeIdx, const Expression &expression, tf::Taskflow &taskflow) override;
+        void Process(std::size_t nodeIdx, const Expression &expression, TaskBuilder &builder) override;
         ExpressionNode::Operation GetOperationType() const override;
     };
 
diff --git a/sources/expression/vector/SplaVectorDataWrite.cpp b/sources/expression/vector/SplaVectorDataWrite.cpp
index 420aaf1fe..d6a42d7f0 100644
--- a/sources/expression/vector/SplaVectorDataWrite.cpp
+++ b/sources/expression/vector/SplaVectorDataWrite.cpp
@@ -39,7 +39,7 @@ bool spla::VectorDataWrite::Select(std::size_t nodeIdx, const spla::Expression &
     return true;
 }
 
-void spla::VectorDataWrite::Process(std::size_t nodeIdx, const spla::Expression &expression, tf::Taskflow &taskflow) {
+void spla::VectorDataWrite::Process(std::size_t nodeIdx, const spla::Expression &expression, spla::TaskBuilder &builder) {
     auto &nodes = expression.GetNodes();
     auto node = nodes[nodeIdx];
     auto library = expression.GetLibrary().GetPrivatePtr();
@@ -57,7 +57,7 @@ void spla::VectorDataWrite::Process(std::size_t nodeIdx, const spla::Expression
     auto blockSize = library->GetBlockSize();
 
     for (std::size_t i = 0; i < math::GetBlocksCount(nrows, blockSize); i++) {
-        taskflow.emplace([=]() {
+        builder.Emplace([=]() {
             using namespace boost;
 
             // todo: gpu and device queue management
diff --git a/sources/expression/vector/SplaVectorDataWrite.hpp b/sources/expression/vector/SplaVectorDataWrite.hpp
index 6fb13b245..90bb774c6 100644
--- a/sources/expression/vector/SplaVectorDataWrite.hpp
+++ b/sources/expression/vector/SplaVectorDataWrite.hpp
@@ -36,7 +36,7 @@ namespace spla {
     public:
         ~VectorDataWrite() override = default;
         bool Select(std::size_t nodeIdx, const Expression &expression) override;
-        void Process(std::size_t nodeIdx, const Expression &expression, tf::Taskflow &taskflow) override;
+        void Process(std::size_t nodeIdx, const Expression &expression, TaskBuilder &builder) override;
         ExpressionNode::Operation GetOperationType() const override;
     };
 


Test patch:
diff --git a/tests/TestDataMatrix.cpp b/tests/TestDataMatrix.cpp
index 854a47111..211c7d79f 100644
--- a/tests/TestDataMatrix.cpp
+++ b/tests/TestDataMatrix.cpp
@@ -46,10 +46,10 @@ static void testCommon(spla::Library &library, size_t M, size_t N, size_t nvals,
 
     library.Submit(spExprWrite);
     spExprWrite->Wait();
-    EXPECT_EQ(spExprWrite->GetState(), spla::Expression::State::Evaluated);
+    ASSERT_EQ(spExprWrite->GetState(), spla::Expression::State::Evaluated);
 
     utils::Matrix<float> expected = source.SortReduceDuplicates();
-    EXPECT_TRUE(expected.Equals(spM));
+    ASSERT_TRUE(expected.Equals(spM));
 }
 
 static void testSortedNoDuplicates(spla::Library &library, size_t M, size_t N, size_t nvals, size_t seed = 0) {
@@ -75,10 +75,10 @@ static void testSortedNoDuplicates(spla::Library &library, size_t M, size_t N, s
 
     library.Submit(spExprWrite);
     spExprWrite->Wait();
-    EXPECT_EQ(spExprWrite->GetState(), spla::Expression::State::Evaluated);
+    ASSERT_EQ(spExprWrite->GetState(), spla::Expression::State::Evaluated);
 
     utils::Matrix<float> &expected = source;
-    EXPECT_TRUE(expected.Equals(spM));
+    ASSERT_TRUE(expected.Equals(spM));
 }
 
 static void test(size_t M, size_t N, size_t base, size_t step, size_t iter) {
diff --git a/tests/TestDataVector.cpp b/tests/TestDataVector.cpp
index e5f6db8fe..32d2e24ed 100644
--- a/tests/TestDataVector.cpp
+++ b/tests/TestDataVector.cpp
@@ -45,11 +45,11 @@ static void testCommon(spla::Library &library, size_t M, size_t nvals, size_t se
 
     library.Submit(spExprWrite);
     spExprWrite->Wait();
-    EXPECT_EQ(spExprWrite->GetState(), spla::Expression::State::Evaluated);
+    ASSERT_EQ(spExprWrite->GetState(), spla::Expression::State::Evaluated);
 
     utils::Vector expected = source.SortReduceDuplicates();
 
-    EXPECT_TRUE(expected.Equals(spV));
+    ASSERT_TRUE(expected.Equals(spV));
 }
 
 static void testSortedNoDuplicates(spla::Library &library, size_t M, size_t nvals, size_t seed = 0) {
@@ -74,10 +74,10 @@ static void testSortedNoDuplicates(spla::Library &library, size_t M, size_t nval
 
     library.Submit(spExprWrite);
     spExprWrite->Wait();
-    EXPECT_EQ(spExprWrite->GetState(), spla::Expression::State::Evaluated);
+    ASSERT_EQ(spExprWrite->GetState(), spla::Expression::State::Evaluated);
 
     utils::Vector<float> expected = source;
-    EXPECT_TRUE(expected.Equals(spV));
+    ASSERT_TRUE(expected.Equals(spV));
 }
 
 static void test(size_t M, size_t base, size_t step, size_t iter) {