(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.1.2.13 and 1.9

version 1.1.2.13, 2001/10/03 16:55:03 version 1.9, 2002/03/18 13:36:27
Line 29 
Line 29 
  
 #ifndef Pegasus_Thread_h #ifndef Pegasus_Thread_h
 #define Pegasus_Thread_h #define Pegasus_Thread_h
 #include <Pegasus/Common/IPC.h>  
 #include <Pegasus/Common/Config.h> #include <Pegasus/Common/Config.h>
   #include <Pegasus/Common/IPC.h>
 #include <Pegasus/Common/Exception.h> #include <Pegasus/Common/Exception.h>
 #include <Pegasus/Common/DQueue.h> #include <Pegasus/Common/DQueue.h>
  
   // REVIEW: Spend time getting to know this.
   
 PEGASUS_NAMESPACE_BEGIN PEGASUS_NAMESPACE_BEGIN
   PEGASUS_USING_STD;
  
 class PEGASUS_EXPORT cleanup_handler  class PEGASUS_COMMON_LINKAGE cleanup_handler
 { {
  
    public:    public:
       cleanup_handler( void (*routine)(void *), void *arg  ) : _routine(routine), _arg(arg)  {}       cleanup_handler( void (*routine)(void *), void *arg  ) : _routine(routine), _arg(arg)  {}
       ~cleanup_handler()  {; }       ~cleanup_handler()  {; }
   
    private:  
       void execute(void) { _routine(_arg); }  
       cleanup_handler();  
       void (*_routine)(void *);  
       inline Boolean operator==(const void *key) const       inline Boolean operator==(const void *key) const
       {       {
          if(key == (void *)_routine)          if(key == (void *)_routine)
Line 57 
Line 55 
       {       {
          return(operator==((const void *)b._routine));          return(operator==((const void *)b._routine));
       }       }
      private:
         void execute(void) { _routine(_arg); }
         cleanup_handler();
         void (*_routine)(void *);
   
       void *_arg;       void *_arg;
       PEGASUS_CLEANUP_HANDLE _cleanup_buffer;       PEGASUS_CLEANUP_HANDLE _cleanup_buffer;
       friend class DQueue<class cleanup_handler>;       friend class DQueue<class cleanup_handler>;
Line 65 
Line 68 
  
 /////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////
  
 class PEGASUS_EXPORT SimpleThread  
 {  
   
    public:  
       SimpleThread( PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *start )(void *),  
                     void *parameter, Boolean detached );  
   
       ~SimpleThread();  
   
       void run(void);  
   
       Uint32 threadId(void);  
   
       // get the user parameter  
       void *get_parm(void);  
   
       // cancellation must be deferred (not asynchronous)  
       // for user-level threads the thread itself can decide  
       // when it should die.  
       void cancel(void);  
   
       void kill(int signum);  
   
       // cancel if there is a pending cancellation request  
       void test_cancel(void);  
   
       // for user-level threads  - put the calling thread  
       // to sleep and jump to the thread scheduler.  
       // platforms with preemptive scheduling and native threads  
       // can define this to be a no-op.  
       // platforms without preemptive scheduling like NetWare  
       // or gnu portable threads will have an existing  
       // routine that can be mapped to this method  
   
       void thread_switch(void);  
   
       // suspend this thread  
       void suspend(void) ;  
   
       // resume this thread  
       void resume(void) ;  
   
       void sleep(Uint32 msec) ;  
   
       // block the calling thread until this thread terminates  
       void join( PEGASUS_THREAD_RETURN *ret_val);  
   
   
       // stack of functions to be called when thread terminates  
       // will be called last in first out (LIFO)  
       void cleanup_push( void (*routine) (void *), void *parm );  
       void cleanup_pop(Boolean execute) ;  
   
       PEGASUS_THREAD_TYPE self(void) ;  
   
    private:  
       SimpleThread();  
   
       PEGASUS_THREAD_HANDLE _handle;  
       Boolean _is_detached;  
       Boolean _cancel_enabled;  
       Boolean _cancelled;  
   
       //PEGASUS_SEM_HANDLE _suspend_count;  
       Semaphore _suspend;  
   
       // always pass this * as the void * parameter to the thread  
       // store the user parameter in _thread_parm  
   
       PEGASUS_THREAD_RETURN  ( PEGASUS_THREAD_CDECL *_start)(void *) ;  
   
       void *_thread_parm;  
 } ;  
   
 ///////////////////////////////////////////////////////////////////////////////  
   
  
   class  PEGASUS_COMMON_LINKAGE thread_data
 class  PEGASUS_EXPORT thread_data  
 { {
  
    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 159 
Line 85 
  
       }       }
  
       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 170 
Line 96 
  
       }       }
  
       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 187 
Line 113 
       {       {
          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 233 
Line 161 
             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 244 
Line 173 
       void *_data;       void *_data;
       size_t _size;       size_t _size;
       Sint8 *_key;       Sint8 *_key;
   
       friend class DQueue<thread_data>;       friend class DQueue<thread_data>;
       friend class Thread;       friend class Thread;
 }; };
Line 251 
Line 181 
  
 /////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////
  
 class PEGASUS_EXPORT Thread  class PEGASUS_COMMON_LINKAGE ThreadPool;
   
   class PEGASUS_COMMON_LINKAGE Thread
 { {
  
    public:    public:
Line 276 
Line 208 
       // cancel if there is a pending cancellation request       // cancel if there is a pending cancellation request
       void test_cancel(void);       void test_cancel(void);
  
         Boolean is_cancelled(void);
   
       // for user-level threads  - put the calling thread       // for user-level threads  - put the calling thread
       // to sleep and jump to the thread scheduler.       // to sleep and jump to the thread scheduler.
       // platforms with preemptive scheduling and native threads       // platforms with preemptive scheduling and native threads
Line 294 
Line 228 
       void resume(void) ;       void resume(void) ;
 #endif #endif
  
       void sleep(Uint32 msec) ;        static void sleep(Uint32 msec) ;
  
       // block the calling thread until this thread terminates       // block the calling thread until this thread terminates
       void join( void );       void join( void );
Line 310 
Line 244 
       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 319 
Line 253 
  
       // 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(const Sint8 *key) throw(IPCException)
         {
            _tsd.try_lock();
            thread_data *tsd = _tsd.reference((const void *)key);
            if(tsd != NULL)
               return((void *)(tsd->_data) );
            else
               return(NULL);
         }
   
   
       // release the lock held on the tsd       // release the lock held on the tsd
       // NOTE: assumes a corresponding and prior call to reference_tsd() !!!       // NOTE: assumes a corresponding and prior call to reference_tsd() !!!
       inline void dereference_tsd(void) throw(IPCException)       inline void dereference_tsd(void) throw(IPCException)
Line 337 
Line 282 
       }       }
  
       // 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(const Sint8 *key) throw(IPCException)
         {
            return(_tsd.remove((const void *)key));
         }
   
         inline void empty_tsd(void) throw(IPCException)
         {
            _tsd.empty_list();
         }
   
       // 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, return the existing buffer
       thread_data *put_tsd(Sint8 *key, void (*delete_func)(void *), Uint32 size, void *value)        thread_data *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);
          PEGASUS_ASSERT(delete_func != 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
          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); }
Line 365 
Line 319 
  
       PEGASUS_THREAD_HANDLE getThreadHandle() {return _handle;}       PEGASUS_THREAD_HANDLE getThreadHandle() {return _handle;}
  
         inline Boolean operator==(const void *key) const
         {
            if ( (void *)this == key)
               return(true);
            return(false);
         }
         inline Boolean operator==(const Thread & b) const
         {
            return(operator==((const void *)&b ));
         }
   
         void detach(void);
   
    private:    private:
       Thread();       Thread();
       inline void create_tsd(Sint8 *key ) throw(IPCException)        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 390 
Line 357 
       void *_thread_parm;       void *_thread_parm;
       PEGASUS_THREAD_RETURN _exit_code;       PEGASUS_THREAD_RETURN _exit_code;
       static Boolean _signals_blocked;       static Boolean _signals_blocked;
         friend class ThreadPool;
 } ; } ;
  
  
 #if 0  class PEGASUS_COMMON_LINKAGE ThreadPool
 class PEGASUS_EXPORT Aggregator {  {
   
    public:    public:
  
       Aggregator();        ThreadPool(Sint16 initial_size,
       ~Aggregator();                   const Sint8 *key,
                    Sint16 min,
                    Sint16 max,
                    struct timeval & alloc_wait,
                    struct timeval & dealloc_wait,
                    struct timeval & deadlock_detect);
   
         ~ThreadPool(void);
         void allocate_and_awaken(void *parm,
                                  PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *work)(void *))
            throw(IPCException);
   
         void allocate_and_awaken(void *parm,
                                  PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *work)(void *),
                                  Semaphore *blocking_sem)
            throw(IPCException);
   
   
         Uint32 kill_dead_threads( void )
            throw(IPCException);
   
         void get_key(Sint8 *buf, int bufsize);
   
         inline Boolean operator==(const void *key) const
         {
            if ( ! strncmp( reinterpret_cast<Sint8 *>(const_cast<void *>(key)), _key, 16  ))
               return(true);
            return(false);
         }
         inline Boolean operator==(const ThreadPool & b) const
         {
            return(operator==((const void *) b._key ));
         }
   
         inline void set_min_threads(Sint16 min)
         {
            _min_threads = min;
         }
   
         inline Sint16 get_min_threads(void) const
         {
            return _min_threads;
         }
  
       void started(void);        inline void set_max_threads(Sint16 max)
       void completed(void);        {
       void remaining(int operations);           _max_threads = max;
       void put_result(CIMReference *ref);        }
   
         inline Sint16 get_max_threads(void) const
         {
            return _max_threads;
         }
   
         inline void set_allocate_wait(const struct timeval & alloc_wait)
         {
            _allocate_wait.tv_sec = alloc_wait.tv_sec;
            _allocate_wait.tv_usec = alloc_wait.tv_usec;
         }
   
         inline struct timeval *get_allocate_wait(struct timeval *buffer) const
         {
            if(buffer == 0)
               throw NullPointer();
            buffer->tv_sec = _allocate_wait.tv_sec;
            buffer->tv_usec = _allocate_wait.tv_usec;
            return buffer;
         }
   
         inline void set_deallocate_wait(const struct timeval & dealloc_wait)
         {
            _deallocate_wait.tv_sec = dealloc_wait.tv_sec;
            _deallocate_wait.tv_usec = dealloc_wait.tv_usec;
         }
   
         inline struct timeval *get_deallocate_wait(struct timeval *buffer) const
         {
            if(buffer == 0)
               throw NullPointer();
            buffer->tv_sec = _deallocate_wait.tv_sec;
            buffer->tv_usec = _deallocate_wait.tv_usec;
            return buffer;
         }
   
         inline void set_deadlock_detect(const struct timeval & deadlock)
         {
            _deadlock_detect.tv_sec = deadlock.tv_sec;
            _deadlock_detect.tv_usec = deadlock.tv_usec;
         }
   
         inline struct timeval * get_deadlock_detect(struct timeval *buffer) const
         {
            if(buffer == 0)
               throw NullPointer();
            buffer->tv_sec = _deadlock_detect.tv_sec;
            buffer->tv_usec = _deadlock_detect.tv_usec;
            return buffer;
         }
   
         inline Uint32 running_count(void)
         {
            return _running.count();
         }
   
         static Boolean check_time(struct timeval *start, struct timeval *interval);
  
    private:    private:
       int _reference_count;        ThreadPool(void);
         Sint16 _max_threads;
         Sint16 _min_threads;
         AtomicInt _current_threads;
         struct timeval _allocate_wait;
         struct timeval _deallocate_wait;
         struct timeval _deadlock_detect;
         static PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL _loop(void *);
         Sint8 _key[17];
         DQueue<Thread> _pool;
         DQueue<Thread> _running;
         DQueue<Thread> _dead;
         AtomicInt _dying;
   
   
         static void _sleep_sem_del(void *p);
   
         void _check_deadlock(struct timeval *start) throw(Deadlock);
         Boolean _check_deadlock_no_throw(struct timeval *start);
         Boolean _check_dealloc(struct timeval *start);
         Thread *_init_thread(void) throw(IPCException);
         void _link_pool(Thread *th) throw(IPCException);
         static PEGASUS_THREAD_RETURN  _undertaker(void *);
  
       // keep track of the thread running this operation so we can kill  
       // it if necessary  
       Thread _owner;  
   
       // this is a phased aggregate. when it is complete is will  
       // be streamed to the client regardless of the state of  
       // siblings  
       Boolean _is_phased;  
   
       int _total_values;  
       int _completed_values;  
       int _total_child_values;  
       int _completed_child_values;  
       int _completion_state;  
       struct timeval _last_update;  
       time_t lifetime;  
       Aggregator *_parent;  
       // children may be phased or not phased  
       DQueue _children;  
       // empty results that are filled by provider  
       DQueue _results;  
       // array of predicates for events and  
       // stored queries (cursors)  
       Array _filter;  
 } ; } ;
   
   
   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++;
      pegasus_yield();
   
      return th;
   }
   
   inline void ThreadPool::_link_pool(Thread *th) throw(IPCException)
   {
      if(th == 0)
         throw NullPointer();
      _pool.insert_first(th);
   }
   
   
   #if defined(PEGASUS_OS_TYPE_WINDOWS)
   # include "ThreadWindows_inline.h"
   #elif defined(PEGASUS_PLATFORM_ZOS_ZSERIES_IBM)
   # include "ThreadzOS_inline.h"
   #elif defined(PEGASUS_OS_TYPE_UNIX)
   # include "ThreadUnix_inline.h"
 #endif #endif
  
 PEGASUS_NAMESPACE_END PEGASUS_NAMESPACE_END

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