1 mike 1.2 //%///////////-*-c++-*-//////////////////////////////////////////////////////
2 //
3 // Copyright (c) 2000, 2001 The Open group, BMC Software, Tivoli Systems, IBM
4 //
5 // Permission is hereby granted, free of charge, to any person obtaining a copy
6 // of this software and associated documentation files (the "Software"), to
7 // deal in the Software without restriction, including without limitation the
8 // rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
9 // sell copies of the Software, and to permit persons to whom the Software is
10 // furnished to do so, subject to the following conditions:
11 //
12 // THE ABOVE COPYRIGHT NOTICE AND THIS PERMISSION NOTICE SHALL BE INCLUDED IN
13 // ALL COPIES OR SUBSTANTIAL PORTIONS OF THE SOFTWARE. THE SOFTWARE IS PROVIDED
14 // "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
15 // LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
16 // PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
17 // HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
18 // ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
19 // WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
20 //
21 //==============================================================================
22 mike 1.2 //
23 // Author: Mike Day (mdday@us.ibm.com)
24 //
25 // Modified By: Markus Mueller
26 //
27 //%/////////////////////////////////////////////////////////////////////////////
28
29
30 #ifndef Pegasus_Thread_h
31 #define Pegasus_Thread_h
32 #include <Pegasus/Common/Config.h>
|
33 mike 1.5 #include <Pegasus/Common/IPC.h>
|
34 mike 1.2 #include <Pegasus/Common/Exception.h>
35 #include <Pegasus/Common/DQueue.h>
36
37 PEGASUS_NAMESPACE_BEGIN
38
39 class PEGASUS_COMMON_LINKAGE cleanup_handler
40 {
41
42 public:
43 cleanup_handler( void (*routine)(void *), void *arg ) : _routine(routine), _arg(arg) {}
44 ~cleanup_handler() {; }
45 inline Boolean operator==(const void *key) const
46 {
47 if(key == (void *)_routine)
48 return true;
49 return false;
50 }
51 inline Boolean operator ==(const cleanup_handler & b) const
52 {
53 return(operator==((const void *)b._routine));
54 }
55 mike 1.2 private:
56 void execute(void) { _routine(_arg); }
57 cleanup_handler();
58 void (*_routine)(void *);
59
60 void *_arg;
61 PEGASUS_CLEANUP_HANDLE _cleanup_buffer;
62 friend class DQueue<class cleanup_handler>;
63 friend class Thread;
64 };
65
66 ///////////////////////////////////////////////////////////////////////////////
67
68
69 class PEGASUS_COMMON_LINKAGE thread_data
70 {
71
72 public:
73 static void default_delete(void *data);
74
|
75 kumpf 1.7 thread_data( const Sint8 *key ) : _delete_func(NULL) , _data(NULL), _size(0)
|
76 mike 1.2 {
77 PEGASUS_ASSERT(key != NULL);
78 size_t keysize = strlen(key);
79 _key = new Sint8 [keysize + 1];
80 memcpy(_key, key, keysize);
81 _key[keysize] = 0x00;
82
83 }
84
|
85 kumpf 1.7 thread_data(const Sint8 *key, size_t size) : _delete_func(default_delete), _size(size)
|
86 mike 1.2 {
87 PEGASUS_ASSERT(key != NULL);
88 size_t keysize = strlen(key);
89 _key = new Sint8 [keysize + 1];
90 memcpy(_key, key, keysize);
91 _key[keysize] = 0x00;
92 _data = ::operator new(_size) ;
93
94 }
95
|
96 kumpf 1.7 thread_data(const Sint8 *key, size_t size, void *data) : _delete_func(default_delete), _size(size)
|
97 mike 1.2 {
98 PEGASUS_ASSERT(key != NULL);
99 PEGASUS_ASSERT(data != NULL);
100 size_t keysize = strlen(key);
101
102 _key = new Sint8[keysize + 1];
103 memcpy(_key, key, keysize);
104 _key[keysize] = 0x00;
105 _data = ::operator new(_size);
106 memcpy(_data, data, size);
107 }
108
109 ~thread_data()
110 {
111 if( _data != NULL)
112 if(_delete_func != NULL)
|
113 sage 1.9 {
|
114 mike 1.2 _delete_func( _data );
|
115 sage 1.9 }
|
116 mike 1.2 if( _key != NULL )
117 delete [] _key;
118 }
119
120 void put_data(void (*del)(void *), size_t size, void *data ) throw(NullPointer)
121 {
122 if(_data != NULL)
123 if(_delete_func != NULL)
124 _delete_func(_data);
125
126 _delete_func = del;
127 _data = data;
128 _size = size;
129 return ;
130 }
131
132 size_t get_size(void) { return _size; }
133
134 void get_data(void **data, size_t *size)
135 {
136 if(data == NULL || size == NULL)
137 mike 1.2 throw NullPointer();
138
139 *data = _data;
140 *size = _size;
141 return;
142
143 }
144
145 void copy_data(void **buf, size_t *size) throw(BufferTooSmall, NullPointer)
146 {
147 if((buf == NULL) || (size == NULL))
148 throw NullPointer() ;
149 *buf = ::operator new(_size);
150 *size = _size;
151 memcpy(*buf, _data, _size);
152 return;
153 }
154
155 inline Boolean operator==(const void *key) const
156 {
157 if ( ! strcmp(_key, (Sint8 *)key))
158 mike 1.2 return(true);
159 return(false);
160 }
|
161 mday 1.8
|
162 mike 1.2 inline Boolean operator==(const thread_data& b) const
163 {
164 return(operator==((const void *)b._key));
165 }
166
167 private:
168 void (*_delete_func) (void *data) ;
169 thread_data();
170 void *_data;
171 size_t _size;
172 Sint8 *_key;
173
174 friend class DQueue<thread_data>;
175 friend class Thread;
176 };
177
178
179 ///////////////////////////////////////////////////////////////////////////
180
181 class PEGASUS_COMMON_LINKAGE ThreadPool;
182
183 mike 1.2 class PEGASUS_COMMON_LINKAGE Thread
184 {
185
186 public:
187 Thread( PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *start )(void *),
188 void *parameter, Boolean detached );
189
190 ~Thread();
191
192 void run(void);
193
194 // get the user parameter
195 inline void *get_parm(void) { return _thread_parm; }
196
197 // send the thread a signal -- may not be appropriate due to Windows
198 // void kill(int signum);
199
200 // cancellation must be deferred (not asynchronous)
201 // for user-level threads the thread itself can decide
202 // when it should die.
203 void cancel(void);
204 mike 1.2
205 // cancel if there is a pending cancellation request
206 void test_cancel(void);
207
|
208 mday 1.3 Boolean is_cancelled(void);
209
|
210 mike 1.2 // for user-level threads - put the calling thread
211 // to sleep and jump to the thread scheduler.
212 // platforms with preemptive scheduling and native threads
213 // can define this to be a no-op.
214 // platforms without preemptive scheduling like NetWare
215 // or gnu portable threads will have an existing
216 // routine that can be mapped to this method
217
218 void thread_switch(void);
219
|
220 kumpf 1.11 #if defined(PEGASUS_PLATFORM_LINUX_IX86_GNU) || defined(PEGASUS_PLATFORM_LINUX_GENERIC_GNU)
|
221 mike 1.2 // suspend this thread
222 void suspend(void) ;
223
224 // resume this thread
225 void resume(void) ;
226 #endif
227
228 static void sleep(Uint32 msec) ;
229
230 // block the calling thread until this thread terminates
231 void join( void );
232 void thread_init(void);
233
234 // thread routine needs to call this function when
235 // it is ready to exit
236 void exit_self(PEGASUS_THREAD_RETURN return_code) ;
237
238 // stack of functions to be called when thread terminates
239 // will be called last in first out (LIFO)
240 void cleanup_push( void (*routine) (void *), void *parm ) throw(IPCException);
241 void cleanup_pop(Boolean execute = true) throw(IPCException);
242 mike 1.2
243 // create and initialize a tsd
|
244 kumpf 1.7 inline void create_tsd(const Sint8 *key, int size, void *buffer) throw(IPCException)
|
245 mike 1.2 {
246 thread_data *tsd = new thread_data(key, size, buffer);
247 try { _tsd.insert_first(tsd); }
248 catch(IPCException& e) { e = e; delete tsd; throw; }
249 }
250
251 // get the buffer associated with the key
252 // NOTE: this call leaves the tsd LOCKED !!!!
|
253 kumpf 1.7 inline void *reference_tsd(const Sint8 *key) throw(IPCException)
|
254 mike 1.2 {
255 _tsd.lock();
|
256 kumpf 1.7 thread_data *tsd = _tsd.reference((const void *)key);
|
257 mike 1.2 if(tsd != NULL)
258 return( (void *)(tsd->_data) );
259 else
260 return(NULL);
261 }
262
|
263 kumpf 1.7 inline void *try_reference_tsd(const Sint8 *key) throw(IPCException)
|
264 mike 1.2 {
265 _tsd.try_lock();
|
266 kumpf 1.7 thread_data *tsd = _tsd.reference((const void *)key);
|
267 mike 1.2 if(tsd != NULL)
268 return((void *)(tsd->_data) );
269 else
270 return(NULL);
271 }
272
273
274 // release the lock held on the tsd
275 // NOTE: assumes a corresponding and prior call to reference_tsd() !!!
276 inline void dereference_tsd(void) throw(IPCException)
277 {
278 _tsd.unlock();
279 }
280
281 // delete the tsd associated with the key
|
282 kumpf 1.7 inline void delete_tsd(const Sint8 *key) throw(IPCException)
|
283 mike 1.2 {
|
284 kumpf 1.7 thread_data *tsd = _tsd.remove((const void *)key);
|
285 mike 1.2 if(tsd != NULL)
286 delete tsd;
287 }
288
|
289 kumpf 1.14 // Note: Caller must delete the thread_data object returned (if not null)
|
290 kumpf 1.7 inline void *remove_tsd(const Sint8 *key) throw(IPCException)
|
291 mike 1.2 {
|
292 kumpf 1.7 return(_tsd.remove((const void *)key));
|
293 mike 1.2 }
294
295 inline void empty_tsd(void) throw(IPCException)
296 {
|
297 kumpf 1.15 thread_data* tsd;
|
298 mday 1.16 while (0 != (tsd = _tsd.remove_first()))
|
299 kumpf 1.15 {
300 delete tsd;
301 }
302 //_tsd.empty_list();
|
303 mike 1.2 }
304
305 // create or re-initialize tsd associated with the key
|
306 kumpf 1.14 // if the tsd already exists, delete the existing buffer
307 void put_tsd(const Sint8 *key, void (*delete_func)(void *), Uint32 size, void *value)
|
308 mike 1.2 throw(IPCException)
309
310 {
311 PEGASUS_ASSERT(key != NULL);
312 thread_data *tsd ;
|
313 kumpf 1.7 tsd = _tsd.remove((const void *)key); // may throw an IPC exception
|
314 kumpf 1.14 delete tsd;
|
315 mike 1.2 thread_data *ntsd = new thread_data(key);
316 ntsd->put_data(delete_func, size, value);
317 try { _tsd.insert_first(ntsd); }
318 catch(IPCException& e) { e = e; delete ntsd; throw; }
319 }
320 inline PEGASUS_THREAD_RETURN get_exit(void) { return _exit_code; }
321 inline PEGASUS_THREAD_TYPE self(void) {return pegasus_thread_self(); }
322
323 PEGASUS_THREAD_HANDLE getThreadHandle() {return _handle;}
324
325 inline Boolean operator==(const void *key) const
326 {
327 if ( (void *)this == key)
328 return(true);
329 return(false);
330 }
331 inline Boolean operator==(const Thread & b) const
332 {
333 return(operator==((const void *)&b ));
334 }
335
336 mike 1.2 void detach(void);
337
338 private:
339 Thread();
|
340 kumpf 1.7 inline void create_tsd(const Sint8 *key ) throw(IPCException)
|
341 mike 1.2 {
342 thread_data *tsd = new thread_data(key);
343 try { _tsd.insert_first(tsd); }
344 catch(IPCException& e) { e = e; delete tsd; throw; }
345 }
346 PEGASUS_THREAD_HANDLE _handle;
347 Boolean _is_detached;
348 Boolean _cancel_enabled;
349 Boolean _cancelled;
350
351 PEGASUS_SEM_HANDLE _suspend_count;
352
353 // always pass this * as the void * parameter to the thread
354 // store the user parameter in _thread_parm
355
356 PEGASUS_THREAD_RETURN ( PEGASUS_THREAD_CDECL *_start)(void *) ;
357 DQueue<class cleanup_handler> _cleanup;
358 DQueue<class thread_data> _tsd;
359
360 void *_thread_parm;
361 PEGASUS_THREAD_RETURN _exit_code;
362 mike 1.2 static Boolean _signals_blocked;
363 friend class ThreadPool;
364 } ;
365
366
367 class PEGASUS_COMMON_LINKAGE ThreadPool
368 {
369 public:
370
371 ThreadPool(Sint16 initial_size,
|
372 kumpf 1.7 const Sint8 *key,
|
373 mike 1.2 Sint16 min,
374 Sint16 max,
375 struct timeval & alloc_wait,
376 struct timeval & dealloc_wait,
377 struct timeval & deadlock_detect);
378
379 ~ThreadPool(void);
380
|
381 mday 1.8 void allocate_and_awaken(void *parm,
|
382 mday 1.10 PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *work)(void *),
383 Semaphore *blocking = 0)
|
384 mday 1.8 throw(IPCException);
385
386
387 Uint32 kill_dead_threads( void )
|
388 mike 1.2 throw(IPCException);
389
390 void get_key(Sint8 *buf, int bufsize);
391
392 inline Boolean operator==(const void *key) const
393 {
394 if ( ! strncmp( reinterpret_cast<Sint8 *>(const_cast<void *>(key)), _key, 16 ))
395 return(true);
396 return(false);
397 }
398 inline Boolean operator==(const ThreadPool & b) const
399 {
400 return(operator==((const void *) b._key ));
401 }
402
403 inline void set_min_threads(Sint16 min)
404 {
405 _min_threads = min;
406 }
407
408 inline Sint16 get_min_threads(void) const
409 mike 1.2 {
410 return _min_threads;
411 }
412
413 inline void set_max_threads(Sint16 max)
414 {
415 _max_threads = max;
416 }
417
418 inline Sint16 get_max_threads(void) const
419 {
420 return _max_threads;
421 }
422
423 inline void set_allocate_wait(const struct timeval & alloc_wait)
424 {
425 _allocate_wait.tv_sec = alloc_wait.tv_sec;
426 _allocate_wait.tv_usec = alloc_wait.tv_usec;
427 }
428
429 inline struct timeval *get_allocate_wait(struct timeval *buffer) const
430 mike 1.2 {
431 if(buffer == 0)
432 throw NullPointer();
433 buffer->tv_sec = _allocate_wait.tv_sec;
434 buffer->tv_usec = _allocate_wait.tv_usec;
435 return buffer;
436 }
437
438 inline void set_deallocate_wait(const struct timeval & dealloc_wait)
439 {
440 _deallocate_wait.tv_sec = dealloc_wait.tv_sec;
441 _deallocate_wait.tv_usec = dealloc_wait.tv_usec;
442 }
443
444 inline struct timeval *get_deallocate_wait(struct timeval *buffer) const
445 {
446 if(buffer == 0)
447 throw NullPointer();
448 buffer->tv_sec = _deallocate_wait.tv_sec;
449 buffer->tv_usec = _deallocate_wait.tv_usec;
450 return buffer;
451 mike 1.2 }
452
453 inline void set_deadlock_detect(const struct timeval & deadlock)
454 {
455 _deadlock_detect.tv_sec = deadlock.tv_sec;
456 _deadlock_detect.tv_usec = deadlock.tv_usec;
457 }
458
459 inline struct timeval * get_deadlock_detect(struct timeval *buffer) const
460 {
461 if(buffer == 0)
462 throw NullPointer();
463 buffer->tv_sec = _deadlock_detect.tv_sec;
464 buffer->tv_usec = _deadlock_detect.tv_usec;
465 return buffer;
466 }
467
468 inline Uint32 running_count(void)
469 {
470 return _running.count();
471 }
472 mike 1.2
473 static Boolean check_time(struct timeval *start, struct timeval *interval);
474
475 private:
476 ThreadPool(void);
477 Sint16 _max_threads;
478 Sint16 _min_threads;
479 AtomicInt _current_threads;
480 struct timeval _allocate_wait;
481 struct timeval _deallocate_wait;
482 struct timeval _deadlock_detect;
483 static PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL _loop(void *);
484 Sint8 _key[17];
485 DQueue<Thread> _pool;
486 DQueue<Thread> _running;
487 DQueue<Thread> _dead;
488 AtomicInt _dying;
489
490
491 static void _sleep_sem_del(void *p);
492
493 mike 1.2 void _check_deadlock(struct timeval *start) throw(Deadlock);
494 Boolean _check_deadlock_no_throw(struct timeval *start);
495 Boolean _check_dealloc(struct timeval *start);
496 Thread *_init_thread(void) throw(IPCException);
497 void _link_pool(Thread *th) throw(IPCException);
498 static PEGASUS_THREAD_RETURN _undertaker(void *);
499
500 };
501
502
503 inline void ThreadPool::_sleep_sem_del(void *p)
504 {
505 if(p != 0)
506 {
507 delete (Semaphore *)p;
508 }
509 }
510
511 inline void ThreadPool::_check_deadlock(struct timeval *start) throw(Deadlock)
512 {
513 if (true == check_time(start, &_deadlock_detect))
514 mike 1.2 throw Deadlock(pegasus_thread_self());
515 return;
516 }
517
518
519 inline Boolean ThreadPool::_check_deadlock_no_throw(struct timeval *start)
520 {
521 return(check_time(start, &_deadlock_detect));
522 }
523
524 inline Boolean ThreadPool::_check_dealloc(struct timeval *start)
525 {
526 return(check_time(start, &_deallocate_wait));
527 }
528
529 inline Thread *ThreadPool::_init_thread(void) throw(IPCException)
530 {
|
531 mday 1.8 Thread *th = (Thread *) new Thread(_loop, this, false);
|
532 mike 1.2 // allocate a sleep semaphore and pass it in the thread context
533 // initial count is zero, loop function will sleep until
534 // we signal the semaphore
535 Semaphore *sleep_sem = (Semaphore *) new Semaphore(0);
536 th->put_tsd("sleep sem", &_sleep_sem_del, sizeof(Semaphore), (void *)sleep_sem);
|
537 mday 1.8
|
538 mike 1.2 struct timeval *dldt = (struct timeval *) ::operator new(sizeof(struct timeval));
539 th->put_tsd("deadlock timer", thread_data::default_delete, sizeof(struct timeval), (void *)dldt);
540 // thread will enter _loop(void *) and sleep on sleep_sem until we signal it
541 th->run();
542 _current_threads++;
|
543 mday 1.8 pegasus_yield();
544
|
545 mike 1.2 return th;
546 }
547
548 inline void ThreadPool::_link_pool(Thread *th) throw(IPCException)
549 {
550 if(th == 0)
551 throw NullPointer();
552 _pool.insert_first(th);
553 }
554
555
556 #if defined(PEGASUS_OS_TYPE_WINDOWS)
557 # include "ThreadWindows_inline.h"
|
558 sage 1.6 #elif defined(PEGASUS_PLATFORM_ZOS_ZSERIES_IBM)
559 # include "ThreadzOS_inline.h"
|
560 mike 1.2 #elif defined(PEGASUS_OS_TYPE_UNIX)
561 # include "ThreadUnix_inline.h"
562 #endif
563
564 PEGASUS_NAMESPACE_END
565
566 #endif // Pegasus_Thread_h
|
Return to
Thread.h
CVS log
Up to [Pegasus] / pegasus / src / Pegasus / Common