Merge pull request #750 from neutronimaging/issue746_transform

Issue746 transform

core/kipl/UnitTests/tThreadPool/tst_ThreadPool.cpp

@@ -24,11 +24,12 @@ private slots:
     void test_FillTaskList();
     void test_Processor();
     void test_ProcessorSingle();
+    void test_Transform();
 
 
 private:
     std::string dataPath;
-
+    size_t Ndata;
 };
 
 TestThreadPool::TestThreadPool()
@@ -37,6 +38,7 @@ TestThreadPool::TestThreadPool()
     dataPath = dataPath + "/../../../../../TestData/";
     // kipl::strings::filenames::CheckPathSlashes(dataPath,true);
 
+    Ndata = 100000;
 
 }
 
@@ -89,7 +91,7 @@ void TestThreadPool::test_Processor()
     size_t N=std::thread::hardware_concurrency();
     DummyProcessor processor(N);
 
-    std::vector<float> data(1000000*N);
+    std::vector<float> data(Ndata*N);
     std::vector<float> result;
     std::iota(data.begin(),data.end(),0);
 
@@ -108,7 +110,7 @@ void TestThreadPool::test_ProcessorSingle()
     size_t N=std::thread::hardware_concurrency();
     DummyProcessor processor(1);
 
-    std::vector<float> data(1000000*N);
+    std::vector<float> data(Ndata*N);
     std::vector<float> result;
     std::iota(data.begin(),data.end(),0);
 
@@ -122,6 +124,27 @@ void TestThreadPool::test_ProcessorSingle()
         QCOMPARE(result[i],std::floor(sqrt(data[i])*1000.0f));
 }
 
+void TestThreadPool::test_Transform()
+{
+    size_t N=std::thread::hardware_concurrency();
+    kipl::utilities::ThreadPool pool(N);
+
+    std::vector<float> data(Ndata*N);
+
+    QBENCHMARK {
+        std::iota(data.begin(),data.end(),0);
+
+        pool.transform(data.data(),data.size(),[](float &val){
+            val = std::floor(sqrt(val)*1000.0f);
+        },8192UL);
+    }
+
+    for (size_t i=0; i<data.size(); ++i) {
+        std::string errorMsg = "Error in data[" + std::to_string(i) + "]=" + std::to_string(data[i]) + " != " + std::to_string(std::floor(sqrt(static_cast<float>(i)) * 1000.0f));
+        QVERIFY2(data[i] == std::floor(sqrt(static_cast<float>(i)) * 1000.0f), errorMsg.c_str());
+    }
+}   
+
 QTEST_APPLESS_MAIN(TestThreadPool)
 
 #include "tst_ThreadPool.moc"

core/kipl/kipl/include/utilities/threadpool.h

@@ -59,6 +59,29 @@ class KIPLSHARED_EXPORT ThreadPool
     /// @return the number of submitted tasks
     size_t tasks_submitted();
 
+
+    /// @brief Transforms the data in parallel
+    /// @tparam T data type
+    /// @tparam F lambda function template
+    /// @param data The data to be transformed
+    /// @param size The size of the data
+    /// @param f The lambda containing the transformation
+    /// @param block_size The size of the block to be processed in one go
+    template<class T, class F>
+    void transform(T* data, size_t size, F &&f, size_t block_size = 1024) {
+        for (size_t i = 0; i < size; i += block_size) {
+            // std::cout << "Processed block " << i << " to " << std::min(i + block_size, size) << std::endl;
+            this->enqueue([&, i] {
+                // std::cout << "Task processed block " << i << " to " << std::min(i + block_size, size) << std::endl;
+                for (size_t j = i; j < std::min(i + block_size, size); ++j) {
+                    auto val = data[j];
+                    f(data[j]);
+                }
+            });
+        }
+        this->barrier();
+    }   
+
     ~ThreadPool();
 
 private: