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

   1 /*
   2  * Copyright (c) 2017 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 <iostream>
  19 #include <stdlib.h>
  20 #include <unistd.h>
  21 #include <type_traits>
  22 #include <dali-test-suite-utils.h>
  23 #include <dali/devel-api/threading/thread-pool.h>
  24
  25 namespace
  26 {
  27 Dali::ThreadPool gThreadPool;
  28
  29 // Helper function dividing workload into N batches
  30 // the loop lambda contains
  31 Dali::UniqueFutureGroup ForEachMT( Dali::ThreadPool* pThreadPool,
  32                                    uint32_t first,
  33                                    uint32_t size,
  34                                    std::function<void( uint32_t, uint32_t, uint32_t )> task )
  35 {
  36   uint32_t i = 0;
  37   uint32_t j = 0;
  38   const auto workerCount = uint32_t(pThreadPool->GetWorkerCount());
  39   const auto step = size / workerCount;
  40   j = workerCount + step;
  41
  42   std::vector<Dali::Task> tasks;
  43   tasks.reserve( workerCount );
  44
  45   for( auto threadIndex = 0u; threadIndex < workerCount; ++threadIndex )
  46   {
  47     Dali::Task lambda = [task, i, j]( int workerIndex )
  48     {
  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 }
  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   {
  89     for( auto i = begin; i < end; ++i )
  90     {
  91       outputs[workerIndex] += inputs[i];
  92     }
  93   });
  94
  95   future->Wait();
  96
  97   // check outputs
  98   int checksum2 = 0;
  99   for( auto output : outputs )
 100   {
 101     checksum2 += output;
 102   }
 103
 104   printf("sum: %d, sum2: %d\n", checksum, checksum2);
 105
 106
 107   DALI_TEST_EQUALS( checksum, checksum2, TEST_LOCATION );
 108
 109   END_TEST;
 110 }
 111
 112 int UtcDaliThreadPoolSingleTask(void)
 113 {
 114   // initialise global thread pool
 115   if( !gThreadPool.GetWorkerCount() )
 116   {
 117     gThreadPool.Initialize( 0u );
 118   }
 119
 120   // some long lasting task
 121   int counter = 0;
 122   auto task = [&counter]( int workerIndex ){
 123     for( int i = 0; i < 10; ++i )
 124     {
 125       counter++;
 126       usleep( 16 * 1000 );
 127     }
 128   };
 129
 130   auto future = gThreadPool.SubmitTask( 0, task );
 131   future->Wait();
 132   DALI_TEST_EQUALS( counter, 10, TEST_LOCATION );
 133
 134   END_TEST;
 135 }
 136
 137 int UtcDaliThreadPoolSubmitTasksCopyArray(void)
 138 {
 139   // initialise global thread pool
 140   if( !gThreadPool.GetWorkerCount() )
 141   {
 142     gThreadPool.Initialize( 0u );
 143   }
 144
 145   std::array<uint8_t, 1024*1024> dataSrc;
 146   for( auto i = 0; i < decltype(i)(dataSrc.size()); ++i)
 147   {
 148     dataSrc[i] = (std::rand() % 0xff);
 149   }
 150
 151   std::array<uint8_t, 1024*1024> dataDst;
 152
 153   // each task copies 1kb od data
 154   std::vector<Dali::Task> tasks;
 155   for( int i = 0; i < 1024; ++i )
 156   {
 157     auto task = [&dataSrc, &dataDst, i ]( int workerIndex )
 158     {
 159       for( int k = 0; k < 1024; ++k )
 160       {
 161         dataDst[i*1024+k] = dataSrc[i*1024+k];
 162       }
 163     };
 164     tasks.push_back( task );
 165   }
 166
 167   DALI_TEST_EQUALS( 1024, tasks.size(), TEST_LOCATION );
 168
 169   gThreadPool.SubmitTasks( tasks );
 170
 171   // wait for pool to finish
 172   gThreadPool.Wait();
 173
 174   // compare arrays
 175   for( auto i = 0; i < decltype(i)(dataSrc.size()); ++i )
 176   {
 177     DALI_TEST_EQUALS( dataSrc[i], dataDst[i], TEST_LOCATION );
 178     if( dataSrc[i] != dataDst[i]  )
 179     {
 180       break;
 181     }
 182   }
 183
 184   END_TEST;
 185 }