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

Diff for /pegasus/src/Pegasus/Common/Thread.cpp between version 1.16 and 1.44

version 1.16, 2002/05/21 04:20:03 version 1.44, 2003/09/25 12:01:22
Line 1 
Line 1 
 //%///////////////////////////////////////////////////////////////////////////// //%/////////////////////////////////////////////////////////////////////////////
 // //
 // Copyright (c) 2000, 2001 The Open group, BMC Software, Tivoli Systems, IBM,  // Copyright (c) 2000, 2001, 2002 BMC Software, Hewlett-Packard Company, IBM,
 // Compaq Computer Corporation  // The Open Group, Tivoli Systems
 // //
 // 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 44 
Line 44 
  
 PEGASUS_NAMESPACE_BEGIN PEGASUS_NAMESPACE_BEGIN
  
   
 void thread_data::default_delete(void * data) void thread_data::default_delete(void * data)
 { {
    if( data != NULL)    if( data != NULL)
       ::operator delete(data);       ::operator delete(data);
 } }
  
   // l10n start
   void language_delete(void * data)
   {
      if( data != NULL)
      {
         AcceptLanguages * al = static_cast<AcceptLanguages *>(data);
         delete al;
      }
   }
   // l10n end
   
 Boolean Thread::_signals_blocked = false; Boolean Thread::_signals_blocked = false;
   // l10n
   PEGASUS_THREAD_KEY_TYPE Thread::_platform_thread_key;
   Boolean Thread::_key_initialized = false;
   Boolean Thread::_key_error = false;
   
  
 // for non-native implementations // for non-native implementations
 #ifndef PEGASUS_THREAD_CLEANUP_NATIVE #ifndef PEGASUS_THREAD_CLEANUP_NATIVE
Line 115 
Line 132 
  
 #endif #endif
  
   // l10n start
   Sint8 Thread::initializeKey()
   {
      PEG_METHOD_ENTER(TRC_THREAD, "Thread::initializeKey");
      if (!Thread::_key_initialized)
      {
           if (Thread::_key_error)
           {
                   Tracer::trace(TRC_THREAD, Tracer::LEVEL4,
                             "Thread: ERROR - thread key error");
                   return -1;
           }
   
           if (pegasus_key_create(&Thread::_platform_thread_key) == 0)
           {
                   Tracer::trace(TRC_THREAD, Tracer::LEVEL4,
                             "Thread: able to create a thread key");
                   Thread::_key_initialized = true;
           }
           else
           {
                   Tracer::trace(TRC_THREAD, Tracer::LEVEL4,
                             "Thread: ERROR - unable to create a thread key");
                   Thread::_key_error = true;
                   return -1;
           }
      }
   
      PEG_METHOD_EXIT();
      return 0;
   }
   
   Thread * Thread::getCurrent()
   {
       PEG_METHOD_ENTER(TRC_THREAD, "Thread::getCurrent");
       if (Thread::initializeKey() != 0)
       {
           return NULL;
       }
       PEG_METHOD_EXIT();
       return (Thread *)pegasus_get_thread_specific(_platform_thread_key);
   }
   
   void Thread::setCurrent(Thread * thrd)
   {
      PEG_METHOD_ENTER(TRC_THREAD, "Thread::setCurrent");
      if (Thread::initializeKey() == 0)
      {
           if (pegasus_set_thread_specific(Thread::_platform_thread_key,
                                                                    (void *) thrd) == 0)
           {
                   Tracer::trace(TRC_THREAD, Tracer::LEVEL4,
                             "Successful set Thread * into thread specific storage");
           }
           else
           {
                   Tracer::trace(TRC_THREAD, Tracer::LEVEL4,
                             "ERROR: got error setting Thread * into thread specific storage");
           }
      }
      PEG_METHOD_EXIT();
   }
   
   AcceptLanguages * Thread::getLanguages()
   {
       PEG_METHOD_ENTER(TRC_THREAD, "Thread::getLanguages");
   
           Thread * curThrd = Thread::getCurrent();
           if (curThrd == NULL)
                   return NULL;
           AcceptLanguages * acceptLangs =
                    (AcceptLanguages *)curThrd->reference_tsd("acceptLanguages");
           curThrd->dereference_tsd();
       PEG_METHOD_EXIT();
           return acceptLangs;
   }
   
   void Thread::setLanguages(AcceptLanguages *langs) //l10n
   {
      PEG_METHOD_ENTER(TRC_THREAD, "Thread::setLanguages");
   
      Thread * currentThrd = Thread::getCurrent();
      if (currentThrd != NULL)
      {
                   // deletes the old tsd and creates a new one
                   currentThrd->put_tsd("acceptLanguages",
                           language_delete,
                           sizeof(AcceptLanguages *),
                           langs);
      }
   
      PEG_METHOD_EXIT();
   }
   
   void Thread::clearLanguages() //l10n
   {
      PEG_METHOD_ENTER(TRC_THREAD, "Thread::clearLanguages");
   
      Thread * currentThrd = Thread::getCurrent();
      if (currentThrd != NULL)
      {
                   // deletes the old tsd
                   currentThrd->delete_tsd("acceptLanguages");
      }
   
      PEG_METHOD_EXIT();
   }
   // l10n end
   
   DQueue<ThreadPool> ThreadPool::_pools(true);
   
   
   void ThreadPool::kill_idle_threads(void)
   {
      static struct timeval now, last = {0, 0};
   
      pegasus_gettimeofday(&now);
      if(now.tv_sec - last.tv_sec > 5)
      {
         _pools.lock();
         ThreadPool *p = _pools.next(0);
         while(p != 0)
         {
            try
            {
               p->kill_dead_threads();
            }
            catch(...)
            {
            }
            p = _pools.next(p);
         }
         _pools.unlock();
         pegasus_gettimeofday(&last);
      }
   }
   
   
 ThreadPool::ThreadPool(Sint16 initial_size, ThreadPool::ThreadPool(Sint16 initial_size,
                        const Sint8 *key,                        const Sint8 *key,
                        Sint16 min,                        Sint16 min,
Line 136 
Line 291 
    memset(_key, 0x00, 17);    memset(_key, 0x00, 17);
    if(key != 0)    if(key != 0)
       strncpy(_key, key, 16);       strncpy(_key, key, 16);
    if(_max_threads < initial_size)     if(_max_threads > 0 && _max_threads < initial_size)
       _max_threads = initial_size;       _max_threads = initial_size;
    if(_min_threads > initial_size)    if(_min_threads > initial_size)
       _min_threads = initial_size;       _min_threads = initial_size;
Line 146 
Line 301 
    {    {
       _link_pool(_init_thread());       _link_pool(_init_thread());
    }    }
      _pools.insert_last(this);
 } }
  
  
  
 ThreadPool::~ThreadPool(void) ThreadPool::~ThreadPool(void)
 { {
      try
      {
         _pools.remove(this);
    _dying++;    _dying++;
    Thread *th = 0;    Thread *th = 0;
    th = _pool.remove_first();    th = _pool.remove_first();
Line 198 
Line 357 
       th = _dead.remove_first();       th = _dead.remove_first();
    }    }
 } }
      catch(...)
      {
      }
   }
  
 // make this static to the class // make this static to the class
 PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL ThreadPool::_loop(void *parm) PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL ThreadPool::_loop(void *parm)
Line 210 
Line 373 
       PEG_METHOD_EXIT();       PEG_METHOD_EXIT();
       throw NullPointer();       throw NullPointer();
    }    }
   
   // l10n
      // Set myself into thread specific storage
      // This will allow code to get its own Thread
      Thread::setCurrent(myself);
   
    ThreadPool *pool = (ThreadPool *)myself->get_parm();    ThreadPool *pool = (ThreadPool *)myself->get_parm();
    if(pool == 0 )    if(pool == 0 )
    {    {
Line 233 
Line 402 
       PEG_METHOD_EXIT();       PEG_METHOD_EXIT();
       myself->exit_self(0);       myself->exit_self(0);
    }    }
      catch(...)
      {
         PEG_METHOD_EXIT();
         myself->exit_self(0);
      }
   
    if(sleep_sem == 0 || deadlock_timer == 0)    if(sleep_sem == 0 || deadlock_timer == 0)
    {    {
       PEG_METHOD_EXIT();       PEG_METHOD_EXIT();
Line 272 
Line 447 
          PEG_METHOD_EXIT();          PEG_METHOD_EXIT();
          throw NullPointer();          throw NullPointer();
       }       }
   
         if(_work ==
            (PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *)(void *)) &_undertaker)
         {
            _work(parm);
         }
   
       gettimeofday(deadlock_timer, NULL);       gettimeofday(deadlock_timer, NULL);
         try
         {
       _work(parm);       _work(parm);
         }
         catch(...)
         {
            gettimeofday(deadlock_timer, NULL);
         }
  
         gettimeofday(deadlock_timer, NULL);
       if( blocking_sem != 0 )       if( blocking_sem != 0 )
          blocking_sem->signal();          blocking_sem->signal();
  
Line 317 
Line 507 
    {    {
       _check_deadlock(&start) ;       _check_deadlock(&start) ;
  
       if(_current_threads < _max_threads)        if(_max_threads == 0 || _current_threads < _max_threads)
       {       {
          th = _init_thread();          th = _init_thread();
          continue;          continue;
Line 377 
Line 567 
    gettimeofday(&now, NULL);    gettimeofday(&now, NULL);
    Uint32 bodies = 0;    Uint32 bodies = 0;
  
   
    // first go thread the dead q and clean it up as much as possible    // first go thread the dead q and clean it up as much as possible
    while(_dead.count() > 0)    while(_dead.count() > 0)
    {    {
         Tracer::trace(TRC_THREAD, Tracer::LEVEL4, "ThreadPool:: removing and joining dead thread");
       Thread *dead = _dead.remove_first();       Thread *dead = _dead.remove_first();
       if(dead == 0)       if(dead == 0)
          throw NullPointer();          throw NullPointer();
       if(dead->_handle.thid != 0)        dead->join();
       {  
          dead->detach();  
          destroy_thread(dead->_handle.thid, 0);  
          dead->_handle.thid = 0;  
          while(dead->_cleanup.count() )  
          {  
             // this may throw a permission exception,  
             // which I will remove from the code prior to stabilizing  
             dead->cleanup_pop(true);  
          }  
       }  
       delete dead;       delete dead;
    }    }
  
Line 409 
Line 588 
    int i = 0;    int i = 0;
    AtomicInt needed(0);    AtomicInt needed(0);
  
    for( q = map[i] ; i < 2; i++, q = map[i])  #ifdef PEGASUS_DISABLE_KILLING_HUNG_THREADS
      // This change prevents the thread pool from killing "hung" threads.
      // The definition of a "hung" thread is one that has been on the run queue
      // for longer than the time interval set when the thread pool was created.
      // Cancelling "hung" threads has proven to be problematic.
   
      // With this change the thread pool will not cancel "hung" threads.  This
      // may prevent a crash depending upon the state of the "hung" thread.  In
      // the case that the thread is actually hung, this change causes the
      // thread resources not to be reclaimed.
   
      // Idle threads, those that have not executed a routine for a time
      // interval, continue to be destroyed.  This is normal and should not
      // cause any problems.
      for( ; i < 1; i++)
   #else
      for( ; i < 2; i++)
   #endif
    {    {
         q = map[i];
       if(q->count() > 0 )       if(q->count() > 0 )
       {       {
          try          try
          {          {
             q->try_lock();             q->try_lock();
          }          }
          catch(AlreadyLocked &)           catch(...)
          {          {
             q++;              return bodies;
             continue;  
          }          }
  
          struct timeval dt = { 0, 0 };          struct timeval dt = { 0, 0 };
Line 433 
Line 629 
             {             {
                dtp = (struct timeval *)th->try_reference_tsd("deadlock timer");                dtp = (struct timeval *)th->try_reference_tsd("deadlock timer");
             }             }
             catch(AlreadyLocked &)              catch(...)
             {             {
                th = q->next(th);                 q->unlock();
                continue;                 return bodies;
             }             }
  
             if(dtp != 0)             if(dtp != 0)
             {             {
                memcpy(&dt, dtp, sizeof(struct timeval));                memcpy(&dt, dtp, sizeof(struct timeval));
   
             }             }
             th->dereference_tsd();             th->dereference_tsd();
             struct timeval deadlock_timeout;             struct timeval deadlock_timeout;
             if( true == check_time(&dt, get_deadlock_detect(&deadlock_timeout) ))              Boolean too_long;
               if( i == 0)
               {
                  too_long = check_time(&dt, get_deallocate_wait(&deadlock_timeout));
               }
               else
               {
                  too_long = check_time(&dt, get_deadlock_detect(&deadlock_timeout));
               }
   
               if( true == too_long)
             {             {
                // if we are deallocating from the pool, escape if we are                // if we are deallocating from the pool, escape if we are
                // down to the minimum thread count                // down to the minimum thread count
                _current_threads--;                _current_threads--;
                if( _current_threads.value() <= (Uint32)_min_threads )                 if( _current_threads.value() < (Uint32)_min_threads )
                {                {
                   if( i == 0)                   if( i == 0)
                   {                   {
Line 472 
Line 677 
  
                if(th != 0)                if(th != 0)
                {                {
                     if( i == 0 )
                     {
                   th->delete_tsd("work func");                   th->delete_tsd("work func");
                   th->put_tsd("work func", NULL,                   th->put_tsd("work func", NULL,
                               sizeof( PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *)(void *)),                               sizeof( PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *)(void *)),
Line 484 
Line 691 
  
                   if(sleep_sem == 0)                   if(sleep_sem == 0)
                   {                   {
                           q->unlock();
                      th->dereference_tsd();                      th->dereference_tsd();
                      throw NullPointer();                      throw NullPointer();
                   }                   }
                   // put the thread on the dead  list  
                   _dead.insert_first(th);  
                   bodies++;  
                   sleep_sem->signal();  
  
                        bodies++;
                   th->dereference_tsd();                   th->dereference_tsd();
                        _dead.insert_first(th);
                        sleep_sem->signal();
                   th = 0;                   th = 0;
                }                }
                     else
                     {
                        // deadlocked threads
                        Tracer::trace(TRC_THREAD, Tracer::LEVEL4, "Killing a deadlocked thread");
                        th->cancel();
                        delete th;
                     }
                  }
             }             }
             th = q->next(th);             th = q->next(th);
               pegasus_sleep(1);
          }          }
          q->unlock();          q->unlock();
          while (needed.value() > 0)          while (needed.value() > 0)
          {          {
             _link_pool(_init_thread());             _link_pool(_init_thread());
             needed--;             needed--;
               pegasus_sleep(0);
          }          }
       }       }
    }    }
   
   
    return bodies;    return bodies;
 } }
  
  
 Boolean ThreadPool::check_time(struct timeval *start, struct timeval *interval) Boolean ThreadPool::check_time(struct timeval *start, struct timeval *interval)
 { {
      // never time out if the interval is zero
      if(interval && interval->tv_sec == 0 && interval->tv_usec == 0)
         return false;
   
    struct timeval now, finish, remaining;    struct timeval now, finish, remaining;
    Uint32 usec;    Uint32 usec;
    gettimeofday(&now, NULL);     pegasus_gettimeofday(&now);
      /* remove valgrind error */
      pegasus_gettimeofday(&remaining);
   
  
    finish.tv_sec = start->tv_sec + interval->tv_sec;    finish.tv_sec = start->tv_sec + interval->tv_sec;
    usec = start->tv_usec + interval->tv_usec;    usec = start->tv_usec + interval->tv_usec;
Line 530 
Line 752 
       return false;       return false;
 } }
  
   
 PEGASUS_THREAD_RETURN ThreadPool::_undertaker( void *parm ) PEGASUS_THREAD_RETURN ThreadPool::_undertaker( void *parm )
 { {
    Thread *myself = reinterpret_cast<Thread *>(parm);     exit_thread((PEGASUS_THREAD_RETURN)1);
    if(myself != 0)     return (PEGASUS_THREAD_RETURN)1;
   }
   
   
    void ThreadPool::_sleep_sem_del(void *p)
    {    {
       myself->detach();     if(p != 0)
       myself->_handle.thid = 0;     {
       myself->cancel();        delete (Semaphore *)p;
       myself->test_cancel();  
       myself->exit_self(0);  
    }    }
    return((PEGASUS_THREAD_RETURN)0);  }
   
    void ThreadPool::_check_deadlock(struct timeval *start) throw(Deadlock)
   {
      if (true == check_time(start, &_deadlock_detect))
         throw Deadlock(pegasus_thread_self());
      return;
   }
   
   
    Boolean ThreadPool::_check_deadlock_no_throw(struct timeval *start)
   {
      return(check_time(start, &_deadlock_detect));
   }
   
    Boolean ThreadPool::_check_dealloc(struct timeval *start)
   {
      return(check_time(start, &_deallocate_wait));
   }
   
    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));
      pegasus_gettimeofday(dldt);
   
      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;
   }
   
    void ThreadPool::_link_pool(Thread *th) throw(IPCException)
   {
      if(th == 0)
         throw NullPointer();
      _pool.insert_first(th);
 } }
  
  

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