IBR-DTNSuite  0.12
Thread.cpp
Go to the documentation of this file.
1 /*
2  * Thread.cpp
3  *
4  * Copyright (C) 2011 IBR, TU Braunschweig
5  *
6  * Written-by: Johannes Morgenroth <morgenroth@ibr.cs.tu-bs.de>
7  *
8  * Licensed under the Apache License, Version 2.0 (the "License");
9  * you may not use this file except in compliance with the License.
10  * You may obtain a copy of the License at
11  *
12  * http://www.apache.org/licenses/LICENSE-2.0
13  *
14  * Unless required by applicable law or agreed to in writing, software
15  * distributed under the License is distributed on an "AS IS" BASIS,
16  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17  * See the License for the specific language governing permissions and
18  * limitations under the License.
19  *
20  */
21 
22 #include "ibrcommon/config.h"
23 #include "ibrcommon/thread/Thread.h"
24 #include "ibrcommon/thread/MutexLock.h"
25 #include "ibrcommon/Logger.h"
26 #include <stdexcept>
27 #include <pthread.h>
28 #include <stdio.h>
29 #include <signal.h>
30 
31 #ifdef __WIN32__
32 #include <windows.h>
33 #include <unistd.h>
34 #elif MACOS
35 #include <sys/param.h>
36 #include <sys/sysctl.h>
37 #else
38 #include <unistd.h>
39 #endif
40 
41 #ifdef __DEVELOPMENT_ASSERTIONS__
42 #include <cassert>
43 #endif
44 
45 namespace ibrcommon
46 {
47  size_t Thread::getNumberOfProcessors()
48  {
49 #ifdef __WIN32__
50  SYSTEM_INFO sysinfo;
51  GetSystemInfo(&sysinfo);
52  return sysinfo.dwNumberOfProcessors;
53 #elif MACOS
54  int nm[2];
55  size_t len = 4;
56  uint32_t count;
57 
58  nm[0] = CTL_HW; nm[1] = HW_AVAILCPU;
59  sysctl(nm, 2, &count, &len, NULL, 0);
60 
61  if(count < 1) {
62  nm[1] = HW_NCPU;
63  sysctl(nm, 2, &count, &len, NULL, 0);
64  if(count < 1) { count = 1; }
65  }
66  return count;
67 #else
68  return sysconf(_SC_NPROCESSORS_ONLN);
69 #endif
70  }
71 
72  void* Thread::__execute__(void *obj) throw ()
73  {
74 #ifdef __DEVELOPMENT_ASSERTIONS__
75  // the given object should never be null
76  assert(obj != NULL);
77 #endif
78  // cast the thread object
79  Thread *th = static_cast<Thread *>(obj);
80 
81  // set the state to running
82  th->_state = THREAD_RUNNING;
83 
84  // run threads run routine
85  th->run();
86 
87  // set the state to finalizing, blocking all threads until this is done
88  th->_state = THREAD_FINALIZING;
89 
90  // call the finally method
91  th->finally();
92 
93  // delete the thread-object is requested
94  if (th->_detached) {
95  th->_state = THREAD_FINALIZED;
96  delete th;
97  } else {
98  // set the state to JOINABLE
99  th->_state = THREAD_JOINABLE;
100  }
101 
102  // exit the thread
103  return NULL;
104  }
105 
106  Thread::Thread(size_t size)
107 #ifdef __WIN32__
108  : _state(THREAD_CREATED, THREAD_FINALIZED), tid(), stack(size), priority(0), _detached(false)
109 #else
110  : _state(THREAD_CREATED, THREAD_FINALIZED), tid(0), stack(size), priority(0), _detached(false)
111 #endif
112  {
113  pthread_attr_init(&attr);
114  }
115 
116  void Thread::yield(void)
117  {
118 #if defined(HAVE_PTHREAD_YIELD_NP)
119  pthread_yield_np();
120 #elif defined(HAVE_PTHREAD_YIELD)
121  pthread_yield();
122 #else
123  sched_yield();
124 #endif
125  }
126 
127  void Thread::concurrency(int level)
128  {
129  #if defined(HAVE_PTHREAD_SETCONCURRENCY)
130  pthread_setconcurrency(level);
131  #endif
132  }
133 
134  void Thread::sleep(time_t timeout)
135  {
136  #if defined(HAVE_PTHREAD_DELAY)
137  timespec ts;
138  ts.tv_sec = timeout / 1000l;
139  ts.tv_nsec = (timeout % 1000l) * 1000000l;
140  pthread_delay(&ts);
141  #elif defined(HAVE_PTHREAD_DELAY_NP)
142  timespec ts;
143  ts.tv_sec = timeout / 1000l;
144  ts.tv_nsec = (timeout % 1000l) * 1000000l;
145  pthread_delay_np(&ts);
146  #else
147  ::usleep(static_cast<useconds_t>(timeout) * 1000);
148  #endif
149  }
150 
151  Thread::~Thread()
152  {
153  pthread_attr_destroy(&attr);
154  }
155 
156  void Thread::reset() throw (ThreadException)
157  {
158  if ( _state != THREAD_FINALIZED )
159  throw ThreadException(0, "invalid state for reset");
160 
161  pthread_attr_destroy(&attr);
162 
163  _state.reset(THREAD_CREATED);
164 #if __WIN32__
165  tid = pthread_t();
166 #else
167  tid = 0;
168 #endif
169  _detached = false;
170 
171  pthread_attr_init(&attr);
172  }
173 
174  int Thread::kill(int sig)
175  {
176  if (pthread_equal(tid, pthread_t())) return -1;
177 
178  return pthread_kill(tid, sig);
179  }
180 
181  void Thread::cancel() throw ()
182  {
183  // block multiple cancel calls
184  {
185  ibrcommon::ThreadsafeState<THREAD_STATE>::Locked ls = _state.lock();
186 
187  // if the thread never has been started: exit and deny any further startup
188  if (ls == THREAD_CREATED)
189  {
190  ls = THREAD_FINALIZED;
191  return;
192  }
193 
194  if ((ls == THREAD_CANCELLED) || (ls == THREAD_FINALIZED) || (ls == THREAD_JOINABLE) || (ls == THREAD_FINALIZING)) return;
195 
196  // wait until a state is reached where cancellation is possible
197  ls.wait(THREAD_RUNNING | THREAD_JOINABLE | THREAD_FINALIZING | THREAD_FINALIZED);
198 
199  // exit if the thread is not running
200  if (ls != THREAD_RUNNING) return;
201 
202  // set state to cancelled
203  ls = THREAD_CANCELLED;
204  }
205 
206  // run custom cancellation
207  __cancellation();
208  }
209 
210  bool Thread::equal(pthread_t t1, pthread_t t2)
211  {
212  return (pthread_equal(t1, t2) != 0);
213  }
214 
215  bool Thread::isFinalized() throw ()
216  {
217  return _state == THREAD_FINALIZED;
218  }
219 
220  JoinableThread::JoinableThread(size_t size)
221  : Thread(size)
222  {
223  }
224 
225  JoinableThread::~JoinableThread()
226  {
227 #ifdef __DEVELOPMENT_ASSERTIONS__
228  // every thread should be joined, when the destructor is called.
229  if ( _state != THREAD_FINALIZED )
230  {
231  std::cerr << "FAILURE: Thread not finalized before! Current state:" << _state.get() << std::endl;
232  assert( _state != THREAD_FINALIZED );
233  }
234 #endif
235  join();
236  }
237 
238  void JoinableThread::start(int adj) throw (ThreadException)
239  {
240  int ret;
241 
242  // switch to STARTED only if this is a fresh thread
243  {
244  ibrcommon::ThreadsafeState<THREAD_STATE>::Locked ls = _state.lock();
245 
246  // only start once
247  if (ls != THREAD_CREATED) return;
248 
249  // set the thread state to STARTED
250  ls = THREAD_STARTED;
251  }
252 
253  // set priority
254  priority = adj;
255 
256  // call the setup method
257  setup();
258 
259  // set the state to running
260  _state = THREAD_INITIALIZED;
261 
262 #ifndef __PTH__
263  // modify the threads attributes - set as joinable thread
264  pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
265 
266 #ifndef ANDROID
267  // ignore scheduling policy and use the same as the parent
268  pthread_attr_setinheritsched(&attr, PTHREAD_INHERIT_SCHED);
269 #endif
270 #endif
271  // we typically use "stack 1" for min stack...
272 #ifdef PTHREAD_STACK_MIN
273  // if the given stack size is too small...
274  if(stack && stack < PTHREAD_STACK_MIN)
275  {
276  // set it to the min stack size
277  stack = PTHREAD_STACK_MIN;
278  }
279 #else
280  // if stack size if larger than zero and smaller than two...
281  if (stack && stack < 2)
282  {
283  // set it to zero (we will not set the stack size)
284  stack = 0;
285  }
286 #endif
287 
288 #ifdef __PTH__
289  // spawn a new thread
290  tid = pth_spawn(PTH_ATTR_DEFAULT, &Thread::__execute__, this);
291 #else
292  // if the stack size is specified (> 0)
293  if (stack)
294  {
295  // set the stack size attribute
296  pthread_attr_setstacksize(&attr, stack);
297  }
298 
299  // spawn a new thread
300  ret = pthread_create(&tid, &attr, &Thread::__execute__, this);
301 
302  switch (ret)
303  {
304  case EAGAIN:
305  _state = THREAD_FINALIZED;
306  throw ThreadException(ret, "The system lacked the necessary resources to create another thread, or the system-imposed limit on the total number of threads in a process PTHREAD_THREADS_MAX would be exceeded.");
307  case EINVAL:
308  _state = THREAD_FINALIZED;
309  throw ThreadException(ret, "The value specified by attr is invalid.");
310  case EPERM:
311  _state = THREAD_FINALIZED;
312  throw ThreadException(ret, "The caller does not have appropriate permission to set the required scheduling parameters or scheduling policy.");
313  }
314 #endif
315  }
316 
317  void JoinableThread::stop() throw ()
318  {
319  Thread::cancel();
320  }
321 
322  void JoinableThread::join(void) throw (ThreadException)
323  {
324  ibrcommon::ThreadsafeState<THREAD_STATE>::Locked ls = _state.lock();
325 
326  // if the thread never has been started: exit and deny any further startup
327  if (ls == THREAD_CREATED)
328  {
329  ls = THREAD_FINALIZED;
330  return;
331  }
332 
333  // wait until the finalized state is reached
334  ls.wait(THREAD_JOINABLE | THREAD_FINALIZED);
335 
336  // if the thread has been joined already: exit
337  if (ls == THREAD_FINALIZED) return;
338 
339  // get the thread-id of the calling thread
340  pthread_t self = pthread_self();
341 
342 #ifdef __DEVELOPMENT_ASSERTIONS__
343  // never try to join our own thread, check this here
344  assert( !equal(tid, self) );
345 #endif
346 
347  // if the thread has been started, do join
348  int ret = 0;
349  if ((ret = pthread_join(tid, NULL)) == 0)
350  {
351  // set the state to joined
352  ls = THREAD_FINALIZED;
353  }
354  else
355  {
356  throw ThreadException(ret, "Join on a thread failed");
357  }
358  }
359 
360  DetachedThread::DetachedThread(size_t size) : Thread(size)
361  {
362  }
363 
364  DetachedThread::~DetachedThread()
365  {
366  }
367 
368  void DetachedThread::start(int adj) throw (ThreadException)
369  {
370  int ret = 0;
371 
372  // switch to STARTED only if this is a fresh thread
373  {
374  ibrcommon::ThreadsafeState<THREAD_STATE>::Locked ls = _state.lock();
375 
376  // only start once
377  if (ls != THREAD_CREATED) return;
378 
379  // set the thread state to STARTED
380  ls = THREAD_STARTED;
381  }
382 
383  // set the priority
384  priority = adj;
385 
386  // call the setup method
387  setup();
388 
389  // set the state to running
390  _state = THREAD_INITIALIZED;
391 
392 #ifndef __PTH__
393  // modify the threads attributes - set as detached thread
394  pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
395 
396 #ifndef ANDROID
397  // ignore scheduling policy and use the same as the parent
398  pthread_attr_setinheritsched(&attr, PTHREAD_INHERIT_SCHED);
399 #endif
400 #endif
401  // we typically use "stack 1" for min stack...
402 #ifdef PTHREAD_STACK_MIN
403  // if the given stack size is too small...
404  if(stack && stack < PTHREAD_STACK_MIN)
405  {
406  // set it to the min stack size
407  stack = PTHREAD_STACK_MIN;
408  }
409 #else
410  // if stack size if larger than zero and smaller than two...
411  if (stack && stack < 2)
412  {
413  // set it to zero (we will not set the stack size)
414  stack = 0;
415  }
416 #endif
417 
418 #ifdef __PTH__
419  // spawn a new thread
420  tid = pth_spawn(PTH_ATTR_DEFAULT, &Thread::__execute__, this);
421 #else
422  // if the stack size is specified (> 0)
423  if (stack)
424  {
425  // set the stack size attribute
426  pthread_attr_setstacksize(&attr, stack);
427  }
428 
429  // set this thread as detached
430  _detached = true;
431 
432  // spawn a new thread
433  ret = pthread_create(&tid, &attr, &Thread::__execute__, this);
434 
435  // check for errors
436  switch (ret)
437  {
438  case EAGAIN:
439  _state = THREAD_FINALIZED;
440  throw ThreadException(ret, "The system lacked the necessary resources to create another thread, or the system-imposed limit on the total number of threads in a process PTHREAD_THREADS_MAX would be exceeded.");
441  case EINVAL:
442  _state = THREAD_FINALIZED;
443  throw ThreadException(ret, "The value specified by attr is invalid.");
444  case EPERM:
445  _state = THREAD_FINALIZED;
446  throw ThreadException(ret, "The caller does not have appropriate permission to set the required scheduling parameters or scheduling policy.");
447  }
448 #endif
449  }
450 
451  void DetachedThread::stop() throw (ThreadException)
452  {
453  Thread::cancel();
454  }
455 }