(file) Return to Thread.h CVS log (file) (dir) Up to [Pegasus] / pegasus / src / Pegasus / Common

Diff for /pegasus/src/Pegasus/Common/Thread.h between version 1.5 and 1.35

version 1.5, 2001/12/25 08:26:09 version 1.35, 2003/10/23 19:29:53
Line 1 
Line 1 
 //%///////////-*-c++-*-//////////////////////////////////////////////////////  //%2003////////////////////////////////////////////////////////////////////////
 // //
 // Copyright (c) 2000, 2001 The Open group, BMC Software, Tivoli Systems, IBM  // Copyright (c) 2000, 2001, 2002  BMC Software, Hewlett-Packard Development
   // Company, L. P., IBM Corp., The Open Group, Tivoli Systems.
   // Copyright (c) 2003 BMC Software; Hewlett-Packard Development Company, L. P.;
   // IBM Corp.; EMC Corporation, The Open Group.
 // //
 // Permission is hereby granted, free of charge, to any person obtaining a copy // Permission is hereby granted, free of charge, to any person obtaining a copy
 // of this software and associated documentation files (the "Software"), to // of this software and associated documentation files (the "Software"), to
Line 26 
Line 29 
 // //
 //%///////////////////////////////////////////////////////////////////////////// //%/////////////////////////////////////////////////////////////////////////////
  
   
 #ifndef Pegasus_Thread_h #ifndef Pegasus_Thread_h
 #define Pegasus_Thread_h #define Pegasus_Thread_h
   
   #include <cstring>
 #include <Pegasus/Common/Config.h> #include <Pegasus/Common/Config.h>
 #include <Pegasus/Common/IPC.h> #include <Pegasus/Common/IPC.h>
 #include <Pegasus/Common/Exception.h>  #include <Pegasus/Common/InternalException.h>
 #include <Pegasus/Common/DQueue.h> #include <Pegasus/Common/DQueue.h>
   #include <Pegasus/Common/AcceptLanguages.h>  // l10n
 // REVIEW: Spend time getting to know this.  #include <Pegasus/Common/Linkage.h>
  
 PEGASUS_NAMESPACE_BEGIN PEGASUS_NAMESPACE_BEGIN
  
Line 74 
Line 78 
    public:    public:
       static void default_delete(void *data);       static void default_delete(void *data);
  
       thread_data( Sint8 *key ) : _delete_func(NULL) , _data(NULL), _size(0)        thread_data( const Sint8 *key ) : _delete_func(NULL) , _data(NULL), _size(0)
       {       {
          PEGASUS_ASSERT(key != NULL);          PEGASUS_ASSERT(key != NULL);
          size_t keysize = strlen(key);          size_t keysize = strlen(key);
Line 84 
Line 88 
  
       }       }
  
       thread_data(Sint8 *key, size_t size) : _delete_func(default_delete), _size(size)        thread_data(const Sint8 *key, size_t size) : _delete_func(default_delete), _size(size)
       {       {
          PEGASUS_ASSERT(key != NULL);          PEGASUS_ASSERT(key != NULL);
          size_t keysize = strlen(key);          size_t keysize = strlen(key);
Line 95 
Line 99 
  
       }       }
  
       thread_data(Sint8 *key, size_t size, void *data) : _delete_func(default_delete), _size(size)        thread_data(const Sint8 *key, size_t size, void *data) : _delete_func(default_delete), _size(size)
       {       {
          PEGASUS_ASSERT(key != NULL);          PEGASUS_ASSERT(key != NULL);
          PEGASUS_ASSERT(data != NULL);          PEGASUS_ASSERT(data != NULL);
Line 112 
Line 116 
       {       {
          if( _data != NULL)          if( _data != NULL)
             if(_delete_func != NULL)             if(_delete_func != NULL)
               {
                _delete_func( _data );                _delete_func( _data );
               }
          if( _key != NULL )          if( _key != NULL )
             delete [] _key;             delete [] _key;
       }       }
Line 142 
Line 148 
  
       }       }
  
       void copy_data(void **buf, size_t *size) throw(BufferTooSmall, NullPointer)        void copy_data(void **buf, size_t *size) throw(NullPointer)
       {       {
          if((buf == NULL) || (size == NULL))          if((buf == NULL) || (size == NULL))
             throw NullPointer() ;             throw NullPointer() ;
Line 158 
Line 164 
             return(true);             return(true);
          return(false);          return(false);
       }       }
   
       inline Boolean operator==(const thread_data& b) const       inline Boolean operator==(const thread_data& b) const
       {       {
          return(operator==((const void *)b._key));          return(operator==((const void *)b._key));
Line 216 
Line 223 
  
       void thread_switch(void);       void thread_switch(void);
  
 #ifdef PEGASUS_PLATFORM_LINUX_IX86_GNU  #if defined(PEGASUS_PLATFORM_LINUX_GENERIC_GNU)
       // suspend this thread       // suspend this thread
       void suspend(void) ;       void suspend(void) ;
  
Line 240 
Line 247 
       void cleanup_pop(Boolean execute = true) throw(IPCException);       void cleanup_pop(Boolean execute = true) throw(IPCException);
  
       // create and initialize a tsd       // create and initialize a tsd
       inline void create_tsd(Sint8 *key, int size, void *buffer) throw(IPCException)        inline void create_tsd(const Sint8 *key, int size, void *buffer) throw(IPCException)
       {       {
          thread_data *tsd = new thread_data(key, size, buffer);          thread_data *tsd = new thread_data(key, size, buffer);
          try { _tsd.insert_first(tsd); }          try { _tsd.insert_first(tsd); }
Line 249 
Line 256 
  
       // get the buffer associated with the key       // get the buffer associated with the key
       // NOTE: this call leaves the tsd LOCKED !!!!       // NOTE: this call leaves the tsd LOCKED !!!!
       inline void *reference_tsd(Sint8 *key) throw(IPCException)        inline void *reference_tsd(const Sint8 *key) throw(IPCException)
       {       {
          _tsd.lock();          _tsd.lock();
          thread_data *tsd = _tsd.reference((void *)key);           thread_data *tsd = _tsd.reference((const void *)key);
          if(tsd != NULL)          if(tsd != NULL)
             return( (void *)(tsd->_data) );             return( (void *)(tsd->_data) );
          else          else
             return(NULL);             return(NULL);
       }       }
  
       inline void *try_reference_tsd(Sint8 *key) throw(IPCException)        inline void *try_reference_tsd(const Sint8 *key) throw(IPCException)
       {       {
          _tsd.try_lock();          _tsd.try_lock();
          thread_data *tsd = _tsd.reference((void *)key);           thread_data *tsd = _tsd.reference((const void *)key);
          if(tsd != NULL)          if(tsd != NULL)
             return((void *)(tsd->_data) );             return((void *)(tsd->_data) );
          else          else
Line 278 
Line 285 
       }       }
  
       // delete the tsd associated with the key       // delete the tsd associated with the key
       inline void delete_tsd(Sint8 *key) throw(IPCException)        inline void delete_tsd(const Sint8 *key) throw(IPCException)
       {       {
          thread_data *tsd = _tsd.remove((void *)key);           thread_data *tsd = _tsd.remove((const void *)key);
          if(tsd != NULL)          if(tsd != NULL)
             delete tsd;             delete tsd;
       }       }
  
       inline void *remove_tsd(Sint8 *key) throw(IPCException)        // Note: Caller must delete the thread_data object returned (if not null)
         inline void *remove_tsd(const Sint8 *key) throw(IPCException)
       {       {
          return(_tsd.remove((void *)key));           return(_tsd.remove((const void *)key));
       }       }
  
       inline void empty_tsd(void) throw(IPCException)       inline void empty_tsd(void) throw(IPCException)
       {       {
          _tsd.empty_list();  
            try
            {
   
               _tsd.try_lock();
            }
            catch(IPCException&)
            {
               return;
            }
   
            thread_data* tsd = _tsd.next(0);
            while(tsd)
            {
               _tsd.remove_no_lock(tsd);
               delete tsd;
               tsd = _tsd.next(0);
            }
            _tsd.unlock();
       }       }
  
       // create or re-initialize tsd associated with the key       // create or re-initialize tsd associated with the key
       // if the tsd already exists, return the existing buffer        // if the tsd already exists, delete the existing buffer
       thread_data *put_tsd(Sint8 *key, void (*delete_func)(void *), Uint32 size, void *value)        void put_tsd(const Sint8 *key, void (*delete_func)(void *), Uint32 size, void *value)
          throw(IPCException)          throw(IPCException)
  
       {       {
          PEGASUS_ASSERT(key != NULL);          PEGASUS_ASSERT(key != NULL);
          thread_data *tsd ;          thread_data *tsd ;
          tsd = _tsd.remove((void *)key);  // may throw an IPC exception           tsd = _tsd.remove((const void *)key);  // may throw an IPC exception
            delete tsd;
          thread_data *ntsd = new thread_data(key);          thread_data *ntsd = new thread_data(key);
          ntsd->put_data(delete_func, size, value);          ntsd->put_data(delete_func, size, value);
          try { _tsd.insert_first(ntsd); }          try { _tsd.insert_first(ntsd); }
          catch(IPCException& e) { e = e; delete ntsd; throw; }          catch(IPCException& e) { e = e; delete ntsd; throw; }
          return(tsd);  
       }       }
       inline PEGASUS_THREAD_RETURN get_exit(void) { return _exit_code; }       inline PEGASUS_THREAD_RETURN get_exit(void) { return _exit_code; }
       inline PEGASUS_THREAD_TYPE self(void) {return pegasus_thread_self(); }       inline PEGASUS_THREAD_TYPE self(void) {return pegasus_thread_self(); }
Line 328 
Line 354 
  
       void detach(void);       void detach(void);
  
         //
         //  Gets the Thread object associated with the caller's thread.
         //  Note: this may return NULL if no Thread object is associated
         //  with the caller's thread.
         //
         static Thread * getCurrent();  // l10n
   
         //
         //  Sets the Thread object associated with the caller's thread.
         //  Note: the Thread object must be placed on the heap.
         //
         static void setCurrent(Thread * thrd); // l10n
   
         //
         //  Gets the AcceptLanguages object associated with the caller's
         //  Thread.
         //  Note: this may return NULL if no Thread object, or no
         //  AcceptLanguages object, is associated with the caller's thread.
         //
         static AcceptLanguages * getLanguages(); //l10n
   
         //
         //  Sets the AcceptLanguages object associated with the caller's
         //  Thread.
         //  Note: a Thread object must have been previously associated with
         //  the caller's thread.
         //  Note: the AcceptLanguages object must be placed on the heap.
         //
         static void setLanguages(AcceptLanguages *langs); //l10n
   
         //
         //  Removes the AcceptLanguages object associated with the caller's
         //  Thread.
         //
         static void clearLanguages(); //l10n
   
    private:    private:
       Thread();       Thread();
       inline void create_tsd(Sint8 *key ) throw(IPCException)  
         static Sint8 initializeKey();  // l10n
   
         inline void create_tsd(const Sint8 *key ) throw(IPCException)
       {       {
          thread_data *tsd = new thread_data(key);          thread_data *tsd = new thread_data(key);
          try { _tsd.insert_first(tsd); }          try { _tsd.insert_first(tsd); }
Line 353 
Line 418 
       void *_thread_parm;       void *_thread_parm;
       PEGASUS_THREAD_RETURN _exit_code;       PEGASUS_THREAD_RETURN _exit_code;
       static Boolean _signals_blocked;       static Boolean _signals_blocked;
         static PEGASUS_THREAD_KEY_TYPE _platform_thread_key;  //l10n
         static Boolean _key_initialized; // l10n
         static Boolean _key_error; // l10n
       friend class ThreadPool;       friend class ThreadPool;
 } ; } ;
  
Line 362 
Line 430 
    public:    public:
  
       ThreadPool(Sint16 initial_size,       ThreadPool(Sint16 initial_size,
                  Sint8 *key,                   const Sint8 *key,
                  Sint16 min,                  Sint16 min,
                  Sint16 max,                  Sint16 max,
                  struct timeval & alloc_wait,                  struct timeval & alloc_wait,
Line 370 
Line 438 
                  struct timeval & deadlock_detect);                  struct timeval & deadlock_detect);
  
       ~ThreadPool(void);       ~ThreadPool(void);
   
       void allocate_and_awaken(void *parm,       void allocate_and_awaken(void *parm,
                                PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *work)(void *))                                 PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *work)(void *),
                                  Semaphore *blocking = 0)
          throw(IPCException);          throw(IPCException);
  
       void kill_dead_threads( void )  
         Uint32 kill_dead_threads( void )
          throw(IPCException);          throw(IPCException);
  
       void get_key(Sint8 *buf, int bufsize);       void get_key(Sint8 *buf, int bufsize);
Line 460 
Line 531 
          return _running.count();          return _running.count();
       }       }
  
         inline Uint32 pool_count(void)
           {
             return _pool.count();
           }
         inline Uint32 dead_count(void)
           {
             return _dead.count();
           }
   
   
       static Boolean check_time(struct timeval *start, struct timeval *interval);       static Boolean check_time(struct timeval *start, struct timeval *interval);
  
         Boolean operator ==(const ThreadPool & p)
         {
            return operator==((const void *)&p);
         }
   
         Boolean operator ==(const void *p)
         {
            if((void *)this == p)
               return true;
            return false;
         }
   
         static void kill_idle_threads(void);
   
    private:    private:
       ThreadPool(void);       ThreadPool(void);
       Sint16 _max_threads;       Sint16 _max_threads;
Line 471 
Line 566 
       struct timeval _deallocate_wait;       struct timeval _deallocate_wait;
       struct timeval _deadlock_detect;       struct timeval _deadlock_detect;
       static PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL _loop(void *);       static PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL _loop(void *);
       Semaphore _waiters;  
       Sint8 _key[17];       Sint8 _key[17];
       Semaphore _pool_sem;  
       DQueue<Thread> _pool;       DQueue<Thread> _pool;
       DQueue<Thread> _running;       DQueue<Thread> _running;
       DQueue<Thread> _dead;       DQueue<Thread> _dead;
       AtomicInt _dying;       AtomicInt _dying;
  
   
       static void _sleep_sem_del(void *p);       static void _sleep_sem_del(void *p);
  
       void _check_deadlock(struct timeval *start) throw(Deadlock);       void _check_deadlock(struct timeval *start) throw(Deadlock);
Line 488 
Line 580 
       Thread *_init_thread(void) throw(IPCException);       Thread *_init_thread(void) throw(IPCException);
       void _link_pool(Thread *th) throw(IPCException);       void _link_pool(Thread *th) throw(IPCException);
       static PEGASUS_THREAD_RETURN  _undertaker(void *);       static PEGASUS_THREAD_RETURN  _undertaker(void *);
         static DQueue<ThreadPool> _pools;
  };  };
  
  
 inline void ThreadPool::_sleep_sem_del(void *p)  
 {  
    if(p != 0)  
    {  
       delete (Semaphore *)p;  
    }  
 }  
   
 inline void ThreadPool::_check_deadlock(struct timeval *start) throw(Deadlock)  
 {  
    if (true == check_time(start, &_deadlock_detect))  
       throw Deadlock(pegasus_thread_self());  
    return;  
 }  
   
   
 inline Boolean ThreadPool::_check_deadlock_no_throw(struct timeval *start)  
 {  
    return(check_time(start, &_deadlock_detect));  
 }  
   
 inline Boolean ThreadPool::_check_dealloc(struct timeval *start)  
 {  
    return(check_time(start, &_deallocate_wait));  
 }  
   
 inline Thread *ThreadPool::_init_thread(void) throw(IPCException)  
 {  
    Thread *th = (Thread *) new Thread(&_loop, this, false);  
    // allocate a sleep semaphore and pass it in the thread context  
    // initial count is zero, loop function will sleep until  
    // we signal the semaphore  
    Semaphore *sleep_sem = (Semaphore *) new Semaphore(0);  
    th->put_tsd("sleep sem", &_sleep_sem_del, sizeof(Semaphore), (void *)sleep_sem);  
    struct timeval *dldt = (struct timeval *) ::operator new(sizeof(struct timeval));  
    th->put_tsd("deadlock timer", thread_data::default_delete, sizeof(struct timeval), (void *)dldt);  
    // thread will enter _loop(void *) and sleep on sleep_sem until we signal it  
    th->run();  
    _current_threads++;  
    return th;  
 }  
   
 inline void ThreadPool::_link_pool(Thread *th) throw(IPCException)  
 {  
    if(th == 0)  
       throw NullPointer();  
    _pool.insert_first(th);  
    _pool_sem.signal();  
 }  
  
  
 #if defined(PEGASUS_OS_TYPE_WINDOWS) #if defined(PEGASUS_OS_TYPE_WINDOWS)
 # include "ThreadWindows_inline.h" # include "ThreadWindows_inline.h"
   #elif defined(PEGASUS_PLATFORM_ZOS_ZSERIES_IBM)
   # include "ThreadzOS_inline.h"
 #elif defined(PEGASUS_OS_TYPE_UNIX) #elif defined(PEGASUS_OS_TYPE_UNIX)
 # include "ThreadUnix_inline.h" # include "ThreadUnix_inline.h"
 #endif #endif

Legend:
Removed from v.1.5  
changed lines
  Added in v.1.35
No CVS admin address has been configured
 Powered by
ViewCVS 0.9.2