OpenWalnut  1.2.5
 All Classes Namespaces Functions Variables Typedefs Enumerations Enumerator Friends Groups Pages
WThreadedJobs.h
1 //---------------------------------------------------------------------------
2 //
3 // Project: OpenWalnut ( http://www.openwalnut.org )
4 //
5 // Copyright 2009 OpenWalnut Community, BSV@Uni-Leipzig and CNCF@MPI-CBS
6 // For more information see http://www.openwalnut.org/copying
7 //
8 // This file is part of OpenWalnut.
9 //
10 // OpenWalnut is free software: you can redistribute it and/or modify
11 // it under the terms of the GNU Lesser General Public License as published by
12 // the Free Software Foundation, either version 3 of the License, or
13 // (at your option) any later version.
14 //
15 // OpenWalnut is distributed in the hope that it will be useful,
16 // but WITHOUT ANY WARRANTY; without even the implied warranty of
17 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18 // GNU Lesser General Public License for more details.
19 //
20 // You should have received a copy of the GNU Lesser General Public License
21 // along with OpenWalnut. If not, see <http://www.gnu.org/licenses/>.
22 //
23 //---------------------------------------------------------------------------
24 
25 #ifndef WTHREADEDJOBS_H
26 #define WTHREADEDJOBS_H
27 
28 #include <iostream>
29 #include <string>
30 
31 #include <boost/shared_ptr.hpp>
32 
33 #include "WException.h"
34 #include "WFlag.h"
35 #include "WLogger.h"
36 
37 /**
38  * \class WThreadedJobs
39  *
40  * A threaded functor base class for producer-consumer-style multithreaded computation.
41  *
42  * A job generator function produces jobs that are then distributed to the threads in
43  * a first come first serve manner. The first template parameter is the type of the input data,
44  * for example a WDataSetScalar. The second template parameter is the type of object that
45  * represents the jobs.
46  *
47  * Both the getJob() and the compute() functions need to be implemented.
48  *
49  * \ingroup common
50  */
51 template< class Input_T, class Job_T >
53 {
54 public:
55  //! the input type
56  typedef Input_T InputType;
57 
58  //! the job type
59  typedef Job_T JobType;
60 
61  /**
62  * Constructor.
63  *
64  * \param input The input.
65  */
66  WThreadedJobs( boost::shared_ptr< InputType const > input ); // NOLINT
67 
68  /**
69  * Destructor.
70  */
71  virtual ~WThreadedJobs();
72 
73  /**
74  * The threaded function operation. Pulls jobs and executes the \see compute()
75  * function.
76  *
77  * \param id The thread's ID.
78  * \param numThreads How many threads are working on the jobs.
79  * \param shutdown A shared flag indicating the thread should be stopped.
80  */
81  void operator() ( std::size_t id, std::size_t numThreads, WBoolFlag const& shutdown );
82 
83  /**
84  * Abstract function for the job aquisition.
85  *
86  * \param job The job (output).
87  * \return false, iff no more jobs need to be processed.
88  */
89  virtual bool getJob( JobType& job ) = 0; // NOLINT
90 
91  /**
92  * Abstract function that performs the actual computation per job.
93  *
94  * \param input The input data.
95  * \param job The current job.
96  */
97  virtual void compute( boost::shared_ptr< InputType const > input, JobType const& job ) = 0;
98 
99 protected:
100 
101  //! the input
102  boost::shared_ptr< InputType const > m_input;
103 private:
104 };
105 
106 template< class Input_T, class Job_T >
107 WThreadedJobs< Input_T, Job_T >::WThreadedJobs( boost::shared_ptr< InputType const > input )
108  : m_input( input )
109 {
110  if( !m_input )
111  {
112  throw WException( std::string( "Invalid input." ) );
113  }
114 }
115 
116 template< class Input_T, class Job_T >
118 {
119 }
120 
121 template< class Input_T, class Job_T >
122 void WThreadedJobs< Input_T, Job_T >::operator() ( std::size_t /* id */, std::size_t /* numThreads */, WBoolFlag const& shutdown )
123 {
124  JobType job;
125  while( getJob( job ) && !shutdown() )
126  {
127  compute( m_input, job );
128  }
129 }
130 
131 /**
132  * Nearly the same class as WThreadedJobs, but this class is intended to be used for multithreaded operations on voxels and therefore it
133  * uses Striping to partition the data. This is necessarry since if the threads are not operating on blocks, they slow down!
134  */
135 template< class Input_T, class Job_T >
137 {
138 public:
139  //! the input type
140  typedef Input_T InputType;
141 
142  //! the job type
143  typedef Job_T JobType;
144 
145  /**
146  * Constructor.
147  *
148  * \param input The input.
149  */
150  WThreadedStripingJobs( boost::shared_ptr< InputType const > input ); // NOLINT
151 
152  /**
153  * Destructor.
154  */
155  virtual ~WThreadedStripingJobs();
156 
157  /**
158  * The threaded function operation. Pulls jobs and executes the \see compute()
159  * function.
160  *
161  * \param id The thread's ID.
162  * \param numThreads How many threads are working on the jobs.
163  * \param shutdown A shared flag indicating the thread should be stopped.
164  */
165  void operator() ( std::size_t id, std::size_t numThreads, WBoolFlag const& shutdown );
166 
167  /**
168  * Abstract function that performs the actual computation per voxel.
169  *
170  * \param input The input data.
171  * \param voxelNum The voxel number to operate on.
172  */
173  virtual void compute( boost::shared_ptr< InputType const > input, std::size_t voxelNum ) = 0;
174 
175 protected:
176 
177  //! the input
178  boost::shared_ptr< InputType const > m_input;
179 private:
180 };
181 
182 template< class Input_T, class Job_T >
183 WThreadedStripingJobs< Input_T, Job_T >::WThreadedStripingJobs( boost::shared_ptr< InputType const > input )
184  : m_input( input )
185 {
186  if( !m_input )
187  {
188  throw WException( std::string( "Invalid input." ) );
189  }
190 }
191 
192 template< class Input_T, class Job_T >
194 {
195 }
196 
197 template< class Input_T, class Job_T >
198 void WThreadedStripingJobs< Input_T, Job_T >::operator() ( std::size_t id, std::size_t numThreads, WBoolFlag const& shutdown )
199 {
200  WAssert( m_input, "Bug: operations of an invalid input requested." );
201  size_t numElements = m_input->size();
202 
203  // partition the voxels via simple striping
204  size_t start = numElements / numThreads * id;
205  size_t end = ( id + 1 ) * ( numElements / numThreads );
206  if( id == numThreads - 1 ) // last thread may have less elements to take care.
207  {
208  end = numElements;
209  }
210 
211  for( size_t voxelNum = start; ( voxelNum < end ) && !shutdown(); ++voxelNum )
212  {
213  compute( m_input, voxelNum );
214  }
215 }
216 
217 #endif // WTHREADEDJOBS_H