automated-tests/src/dali/utc-Dali-ThreadPool.cpp

   1 /*
   2  * Copyright (c) 2020 Samsung Electronics Co., Ltd.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  * http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  *
  16  */
  17
  18 #include <dali-test-suite-utils.h>
  19 #include <dali/devel-api/threading/thread-pool.h>
  20 #include <stdlib.h>
  21 #include <unistd.h>
  22
  23 #include <iostream>
  24 #include <type_traits>
  25
  26 namespace
  27 {
  28 Dali::ThreadPool gThreadPool;
  29
  30 // Helper function dividing workload into N batches
  31 // the loop lambda contains
  32 Dali::UniqueFutureGroup ForEachMT(Dali::ThreadPool*                                 pThreadPool,
  33                                   uint32_t                                          first,
  34                                   uint32_t                                          size,
  35                                   std::function<void(uint32_t, uint32_t, uint32_t)> task)
  36 {
  37   uint32_t   i           = 0;
  38   uint32_t   j           = 0;
  39   const auto workerCount = uint32_t(pThreadPool->GetWorkerCount());
  40   const auto step        = size / workerCount;
  41   j                      = workerCount + step;
  42
  43   std::vector<Dali::Task> tasks;
  44   tasks.reserve(workerCount);
  45
  46   for(auto threadIndex = 0u; threadIndex < workerCount; ++threadIndex)
  47   {
  48     Dali::Task lambda = [task, i, j](int workerIndex) {
  49       task(uint32_t(workerIndex), i, j);
  50     };
  51     tasks.emplace_back(lambda);
  52     i = j;
  53     j = i + step;
  54     if(j > size)
  55       j = size;
  56   }
  57   return pThreadPool->SubmitTasks(tasks, workerCount);
  58 }
  59
  60 } // namespace
  61
  62 int UtcDaliThreadPoolMultipleTasks(void)
  63 {
  64   // initialise global thread pool
  65   if(!gThreadPool.GetWorkerCount())
  66   {
  67     gThreadPool.Initialize(0u);
  68   }
  69
  70   // populate inputs
  71   std::array<int, 8192> inputs;
  72   int                   checksum = 0;
  73   for(auto i = 0; i < decltype(i)(inputs.size()); ++i)
  74   {
  75     inputs[i] = i;
  76     checksum += i;
  77   }
  78
  79   // allocate outputs ( number of outputs equals number of worker threads
  80   auto workerCount = gThreadPool.GetWorkerCount();
  81
  82   std::vector<int> outputs;
  83   outputs.resize(workerCount);
  84   std::fill(outputs.begin(), outputs.end(), 0);
  85
  86   // submit
  87   auto future = ForEachMT(&gThreadPool, 0, inputs.size(), [&inputs, &outputs](uint32_t workerIndex, uint32_t begin, uint32_t end) {
  88     for(auto i = begin; i < end; ++i)
  89     {
  90       outputs[workerIndex] += inputs[i];
  91     }
  92   });
  93
  94   future->Wait();
  95
  96   // check outputs
  97   int checksum2 = 0;
  98   for(auto output : outputs)
  99   {
 100     checksum2 += output;
 101   }
 102
 103   printf("sum: %d, sum2: %d\n", checksum, checksum2);
 104
 105   DALI_TEST_EQUALS(checksum, checksum2, TEST_LOCATION);
 106
 107   END_TEST;
 108 }
 109
 110 int UtcDaliThreadPoolSingleTask(void)
 111 {
 112   // initialise global thread pool
 113   if(!gThreadPool.GetWorkerCount())
 114   {
 115     gThreadPool.Initialize(0u);
 116   }
 117
 118   // some long lasting task
 119   int  counter = 0;
 120   auto task    = [&counter](int workerIndex) {
 121     for(int i = 0; i < 10; ++i)
 122     {
 123       counter++;
 124       usleep(16 * 1000);
 125     }
 126   };
 127
 128   auto future = gThreadPool.SubmitTask(0, task);
 129   future->Wait();
 130   DALI_TEST_EQUALS(counter, 10, TEST_LOCATION);
 131
 132   END_TEST;
 133 }
 134
 135 int UtcDaliThreadPoolSubmitTasksCopyArray(void)
 136 {
 137   // initialise global thread pool
 138   if(!gThreadPool.GetWorkerCount())
 139   {
 140     gThreadPool.Initialize(0u);
 141   }
 142
 143   std::array<uint8_t, 1024 * 1024> dataSrc;
 144   for(auto i = 0; i < decltype(i)(dataSrc.size()); ++i)
 145   {
 146     dataSrc[i] = (std::rand() % 0xff);
 147   }
 148
 149   std::array<uint8_t, 1024 * 1024> dataDst;
 150
 151   // each task copies 1kb od data
 152   std::vector<Dali::Task> tasks;
 153   for(int i = 0; i < 1024; ++i)
 154   {
 155     auto task = [&dataSrc, &dataDst, i](int workerIndex) {
 156       for(int k = 0; k < 1024; ++k)
 157       {
 158         dataDst[i * 1024 + k] = dataSrc[i * 1024 + k];
 159       }
 160     };
 161     tasks.push_back(task);
 162   }
 163
 164   DALI_TEST_EQUALS(1024, tasks.size(), TEST_LOCATION);
 165
 166   gThreadPool.SubmitTasks(tasks);
 167
 168   // wait for pool to finish
 169   gThreadPool.Wait();
 170
 171   // compare arrays
 172   for(auto i = 0; i < decltype(i)(dataSrc.size()); ++i)
 173   {
 174     DALI_TEST_EQUALS(dataSrc[i], dataDst[i], TEST_LOCATION);
 175     if(dataSrc[i] != dataDst[i])
 176     {
 177       break;
 178     }
 179   }
 180
 181   END_TEST;
 182 }