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