iLab Neuromorphic Robotics Toolkit  0.1
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Groups Pages
test-BottomlessThreadPool.C
4 #include <iostream>
5 #include <chrono>
6 #include <algorithm>
7 
8 std::mutex printMtx;
9 
10 void job(int i, std::chrono::microseconds workTime)
11 {
12  std::this_thread::sleep_for(workTime);
13 }
14 
15 NRT_DECLARE_PARAMETER(numjobs, int, "The number of jobs to execute", 1000);
16 NRT_DECLARE_PARAMETER(worktime, float, "Maximum time for each thread to work (milliseconds)", 0.5);
17 NRT_DECLARE_PARAMETER(waittime, float, "Maximum time to wait between spawning each thread (milliseconds)", 0.05);
18 
19 int main(int argc, const char** argv)
20 {
21  nrt::Manager mgr(argc, argv);
22 
23  class MyParameters : public nrt::Component, public nrt::Parameter<numjobs, worktime, waittime>
24  { using nrt::Component::Component; };
25 
26  auto params = mgr.addComponent<MyParameters>("params");
27 
28  mgr.launch();
29 
30  int const numJobs = params->numjobs::get();
31  float const workTime_us = params->worktime::get() * 1000;
32  float const waitTime_us = params->waittime::get() * 1000;
33 
34  srand(time(NULL));
35 
36  std::vector<std::chrono::microseconds> worktimes(numJobs);
37  std::generate(worktimes.begin(), worktimes.end(), [workTime_us]()
38  { return std::chrono::microseconds(int(rand() / float(RAND_MAX) * workTime_us)); });
39  std::vector<std::chrono::microseconds> waittimes(numJobs);
40  std::generate(waittimes.begin(), waittimes.end(), [waitTime_us]()
41  { return std::chrono::microseconds(int(rand() / float(RAND_MAX) * waitTime_us)); });
42 
43 
44  typedef std::chrono::high_resolution_clock Clock;
45 
46  Clock::time_point threadpool_start = Clock::now();
47 
49  pool.pushJob([](){return 3;});
50 
51  std::vector<std::future<void>> futures;
52  for(int i=0; i<numJobs; ++i)
53  {
54  std::function<void()> j = std::bind(job, i, worktimes[i]);
55  futures.push_back(pool.pushJob(j));
56  std::this_thread::sleep_for(waittimes[i]);
57  }
58 
59  for(auto & f : futures)
60  f.get();
61 
62  Clock::time_point threadpool_end = Clock::now();
63 
64 
65  futures.clear();
66 
67 
68  Clock::time_point async_start = Clock::now();
69  for(int i=0; i<numJobs; ++i)
70  {
71  std::function<void()> j = std::bind(job, i, worktimes[i]);
72  futures.push_back(std::async(std::launch::async, j));
73  std::this_thread::sleep_for(waittimes[i]);
74  }
75 
76  for(auto & f : futures)
77  f.get();
78 
79  Clock::time_point async_end = Clock::now();
80 
81  ////////////////////////////////////////////////////////////////////////////////
82  nrt::UnboundedThreadPool unboundedGlobalPool(0);
83 
84  futures.clear();
85  Clock::time_point unboundedglobal_start = Clock::now();
86  for(int i=0; i<numJobs; ++i)
87  {
88  std::function<void()> j = std::bind(job, i, worktimes[i]);
89  futures.push_back(unboundedGlobalPool.pushJob(j));
90  std::this_thread::sleep_for(waittimes[i]);
91  }
92 
93  for(auto & f : futures)
94  f.get();
95 
96  Clock::time_point unboundedglobal_end = Clock::now();
97 
98  ////////////////////////////////////////////////////////////////////////////////
99 #ifndef __clang__
100  nrt::UnboundedLocalQueueThreadPool unboundedLocalPool(0);
101 
102  futures.clear();
103  Clock::time_point unboundedlocal_start = Clock::now();
104  for(int i=0; i<numJobs; ++i)
105  {
106  std::function<void()> j = std::bind(job, i, worktimes[i]);
107  futures.push_back(unboundedLocalPool.pushJob(j));
108  std::this_thread::sleep_for(waittimes[i]);
109  }
110 
111  for(auto & f : futures)
112  f.get();
113 
114  Clock::time_point unboundedlocal_end = Clock::now();
115 
116  float unboundedlocal_ms = std::chrono::duration_cast<std::chrono::milliseconds>(unboundedlocal_end - unboundedlocal_start).count();
117 #endif
118 
119  float threadpool_ms = std::chrono::duration_cast<std::chrono::milliseconds>(threadpool_end - threadpool_start).count();
120  float async_ms = std::chrono::duration_cast<std::chrono::milliseconds>(async_end - async_start).count();
121  float unboundedglobal_ms = std::chrono::duration_cast<std::chrono::milliseconds>(unboundedglobal_end - unboundedglobal_start).count();
122 
123  std::cout << "Threadpool: " << threadpool_ms << "ms Async: " << async_ms << "ms UnboundedGlobal: " << unboundedglobal_ms;
124 #ifndef __clang__
125  std::cout << "ms UnboundedLocal: " << unboundedlocal_ms << "ms" << std::endl;
126 #else
127  std::cout << "ms" << std::endl;
128 #endif
129  std::cout << "Threadpool using " << pool.numThreads() << " threads (" << numJobs << " jobs)" << std::endl;
130 
131  return 0;
132 }
133