(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.52 and 1.65

version 1.52, 2003/10/17 14:22:49 version 1.65, 2004/06/04 05:54:51
Line 1 
Line 1 
 //%/////////////////////////////////////////////////////////////////////////////  //%2003////////////////////////////////////////////////////////////////////////
 // //
 // Copyright (c) 2000, 2001, 2002 BMC Software, Hewlett-Packard Company, IBM,  // Copyright (c) 2000, 2001, 2002  BMC Software, Hewlett-Packard Development
 // The Open Group, Tivoli Systems  // 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 25 
Line 27 
 // //
 // Modified By: Rudy Schuet (rudy.schuet@compaq.com) 11/12/01 // Modified By: Rudy Schuet (rudy.schuet@compaq.com) 11/12/01
 //              added nsk platform support //              added nsk platform support
   //              Roger Kumpf, Hewlett-Packard Company (roger_kumpf@hp.com)
   //              Amit K Arora, IBM (amita@in.ibm.com) for PEP#101
 // //
 //%///////////////////////////////////////////////////////////////////////////// //%/////////////////////////////////////////////////////////////////////////////
  
Line 56 
Line 60 
 { {
    if( data != NULL)    if( data != NULL)
    {    {
       AcceptLanguages * al = static_cast<AcceptLanguages *>(data);        AutoPtr<AcceptLanguages> al(static_cast<AcceptLanguages *>(data));
       delete al;  
    }    }
 } }
 // l10n end // l10n end
  
 Boolean Thread::_signals_blocked = false; Boolean Thread::_signals_blocked = false;
 // l10n // l10n
   #ifndef PEGASUS_OS_ZOS
 PEGASUS_THREAD_KEY_TYPE Thread::_platform_thread_key = -1; PEGASUS_THREAD_KEY_TYPE Thread::_platform_thread_key = -1;
   #else
   PEGASUS_THREAD_KEY_TYPE Thread::_platform_thread_key;
   #endif
 Boolean Thread::_key_initialized = false; Boolean Thread::_key_initialized = false;
 Boolean Thread::_key_error = false; Boolean Thread::_key_error = false;
  
Line 73 
Line 80 
 #ifndef PEGASUS_THREAD_CLEANUP_NATIVE #ifndef PEGASUS_THREAD_CLEANUP_NATIVE
 void Thread::cleanup_push( void (*routine)(void *), void *parm) throw(IPCException) void Thread::cleanup_push( void (*routine)(void *), void *parm) throw(IPCException)
 { {
     cleanup_handler *cu = new cleanup_handler(routine, parm);      AutoPtr<cleanup_handler> cu(new cleanup_handler(routine, parm));
     try      _cleanup.insert_first(cu.get());
     {      cu.release();
         _cleanup.insert_first(cu);  
     }  
     catch(IPCException&)  
     {  
         delete cu;  
         throw;  
     }  
     return;     return;
 } }
  
 void Thread::cleanup_pop(Boolean execute) throw(IPCException) void Thread::cleanup_pop(Boolean execute) throw(IPCException)
 { {
     cleanup_handler *cu ;      AutoPtr<cleanup_handler> cu ;
     try     try
     {     {
         cu = _cleanup.remove_first() ;          cu.reset(_cleanup.remove_first());
     }     }
     catch(IPCException&)     catch(IPCException&)
     {     {
Line 99 
Line 99 
      }      }
     if(execute == true)     if(execute == true)
         cu->execute();         cu->execute();
     delete cu;  
 } }
  
 #endif #endif
Line 241 
Line 240 
 } }
 // l10n end // l10n end
  
   #if 0
   
 // two special synchronization classes for ThreadPool // two special synchronization classes for ThreadPool
 // //
  
Line 260 
Line 258 
       }       }
       Mutex* _mut;       Mutex* _mut;
 }; };
   #endif
  
 class try_mutex class try_mutex
 { {
Line 278 
Line 276 
       Mutex* _mut;       Mutex* _mut;
 }; };
  
   class auto_int
   {
      public:
         auto_int(AtomicInt* num)
            : _int(num)
         {
            _int->operator++();
         }
         ~auto_int(void)
         {
            _int->operator--();
         }
         AtomicInt *_int;
   };
   
  
 DQueue<ThreadPool> ThreadPool::_pools(true);  AtomicInt _idle_control;
  
   DQueue<ThreadPool> ThreadPool::_pools(true);
  
 void ThreadPool::kill_idle_threads(void) void ThreadPool::kill_idle_threads(void)
 { {
Line 350 
Line 364 
  
 ThreadPool::~ThreadPool(void) ThreadPool::~ThreadPool(void)
 { {
      PEG_METHOD_ENTER(TRC_THREAD, "Thread::~ThreadPool");
    try    try
    {    {
       // set the dying flag so all thread know the destructor has been entered        // Set the dying flag so all thread know the destructor has been entered
       {  
          auto_mutex(&(this->_monitor));  
          _dying++;          _dying++;
       }  
       // remove from the global pools list       // remove from the global pools list
       _pools.remove(this);       _pools.remove(this);
  
Line 368 
Line 381 
       while(th != 0)       while(th != 0)
       {       {
          sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");          sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");
            PEGASUS_ASSERT(sleep_sem != 0);
   
          if(sleep_sem == 0)          if(sleep_sem == 0)
          {          {
             th->dereference_tsd();             th->dereference_tsd();
             throw NullPointer();  
          }          }
            else
            {
          // Signal to get the thread out of the work loop.          // Signal to get the thread out of the work loop.
          sleep_sem->signal();          sleep_sem->signal();
  
Line 387 
Line 402 
          th->cancel();          th->cancel();
          th->join();          th->join();
          delete th;          delete th;
            }
          th = _pool.remove_first();          th = _pool.remove_first();
       }       }
   
         while(_idle_control.value())
            pegasus_yield();
   
       th = _dead.remove_first();       th = _dead.remove_first();
       while(th != 0)       while(th != 0)
       {       {
          sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");          sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");
            PEGASUS_ASSERT(sleep_sem != 0);
  
          if(sleep_sem == 0)          if(sleep_sem == 0)
          {          {
             th->dereference_tsd();             th->dereference_tsd();
             throw NullPointer();  
          }          }
            else
            {
               //ATTN-DME-P3-20030322: _dead queue processing in
               //ThreadPool::~ThreadPool is inconsistent with the
               //processing in kill_dead_threads.  Is this correct?
  
          // signal the thread's sleep semaphore          // signal the thread's sleep semaphore
          sleep_sem->signal();          sleep_sem->signal();
Line 407 
Line 432 
          th->cancel();          th->cancel();
          th->join();          th->join();
          delete th;          delete th;
            }
          th = _dead.remove_first();          th = _dead.remove_first();
       }       }
  
Line 417 
Line 443 
             // signal the thread's sleep semaphore             // signal the thread's sleep semaphore
  
             sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");             sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");
               PEGASUS_ASSERT(sleep_sem != 0);
   
             if(sleep_sem == 0 )             if(sleep_sem == 0 )
             {             {
                th->dereference_tsd();                th->dereference_tsd();
                throw NullPointer();  
             }             }
               else
               {
             sleep_sem->signal();             sleep_sem->signal();
             sleep_sem->signal();             sleep_sem->signal();
             th->dereference_tsd();             th->dereference_tsd();
Line 431 
Line 459 
  
             th->join();             th->join();
             delete th;             delete th;
               }
             th = _running.remove_first();             th = _running.remove_first();
          }          }
       }       }
   
    }    }
  
    catch(...)    catch(...)
Line 450 
Line 478 
    Thread *myself = (Thread *)parm;    Thread *myself = (Thread *)parm;
    if(myself == 0)    if(myself == 0)
    {    {
         Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
             "ThreadPool::_loop: Thread pointer is null");
       PEG_METHOD_EXIT();       PEG_METHOD_EXIT();
       throw NullPointer();       throw NullPointer();
    }    }
Line 462 
Line 492 
    ThreadPool *pool = (ThreadPool *)myself->get_parm();    ThreadPool *pool = (ThreadPool *)myself->get_parm();
    if(pool == 0 )    if(pool == 0 )
    {    {
         Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
             "ThreadPool::_loop: ThreadPool pointer is null");
       PEG_METHOD_EXIT();       PEG_METHOD_EXIT();
       throw NullPointer();       throw NullPointer();
    }    }
    if(pool->_dying.value())    if(pool->_dying.value())
    {    {
         Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
             "ThreadPool::_loop: ThreadPool is dying(1)");
       PEG_METHOD_EXIT();       PEG_METHOD_EXIT();
       return((PEGASUS_THREAD_RETURN)0);       return((PEGASUS_THREAD_RETURN)0);
    }    }
Line 486 
Line 520 
  
    catch(...)    catch(...)
    {    {
         Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
             "ThreadPool::_loop: Failure getting sleep_sem or deadlock_timer");
       PEG_METHOD_EXIT();       PEG_METHOD_EXIT();
       return((PEGASUS_THREAD_RETURN)0);       return((PEGASUS_THREAD_RETURN)0);
    }    }
  
    if(sleep_sem == 0 || deadlock_timer == 0)    if(sleep_sem == 0 || deadlock_timer == 0)
    {    {
         Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
             "ThreadPool::_loop: sleep_sem or deadlock_timer are null.");
       PEG_METHOD_EXIT();       PEG_METHOD_EXIT();
       return((PEGASUS_THREAD_RETURN)0);       return((PEGASUS_THREAD_RETURN)0);
    }    }
  
    while(pool->_dying.value() < 1)     while(1)
    {    {
         if(pool->_dying.value())
            break;
   
       try       try
       {       {
          sleep_sem->wait();          sleep_sem->wait();
       }       }
       catch(IPCException& )       catch(IPCException& )
       {       {
            Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
             "ThreadPool::_loop: failure on sleep_sem->wait().");
          PEG_METHOD_EXIT();          PEG_METHOD_EXIT();
          return((PEGASUS_THREAD_RETURN)0);          return((PEGASUS_THREAD_RETURN)0);
       }       }
  
       // when we awaken we reside on the running queue, not the pool queue       // when we awaken we reside on the running queue, not the pool queue
       if(pool->_dying.value())  
       {  
          PEG_METHOD_EXIT();  
          return((PEGASUS_THREAD_RETURN)0);  
       }  
   
  
       PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *_work)(void *) = 0;       PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *_work)(void *) = 0;
       void *parm = 0;       void *parm = 0;
Line 532 
Line 569 
       }       }
       catch(IPCException &)       catch(IPCException &)
       {       {
            Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
              "ThreadPool::_loop: Failure accessing work func, work parm, or blocking sem.");
          PEG_METHOD_EXIT();          PEG_METHOD_EXIT();
          return((PEGASUS_THREAD_RETURN)0);          return((PEGASUS_THREAD_RETURN)0);
       }       }
  
       if(_work == 0)       if(_work == 0)
       {       {
            Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
              "ThreadPool::_loop: work func is null.");
          PEG_METHOD_EXIT();          PEG_METHOD_EXIT();
          throw NullPointer();           return((PEGASUS_THREAD_RETURN)0);
       }       }
  
       if(_work ==       if(_work ==
          (PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *)(void *)) &_undertaker)          (PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *)(void *)) &_undertaker)
       {       {
            PEG_METHOD_EXIT();
          _work(parm);          _work(parm);
       }       }
  
       gettimeofday(deadlock_timer, NULL);       gettimeofday(deadlock_timer, NULL);
       try  
       {        if (pool->_dying.value() == 0)
          {          {
             timed_mutex(&(pool->_monitor), 1000);           try
             if(pool->_dying.value())  
             {             {
                _undertaker(parm);  
             }  
          }  
          _work(parm);          _work(parm);
       }       }
            catch(Exception & e)
            {
               PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,
                  String("Exception from _work in ThreadPool::_loop: ") +
                     e.getMessage());
               PEG_METHOD_EXIT();
               return((PEGASUS_THREAD_RETURN)0);
            }
       catch(...)       catch(...)
       {       {
               Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
                 "ThreadPool::_loop: execution of _work failed.");
               PEG_METHOD_EXIT();
          return((PEGASUS_THREAD_RETURN)0);          return((PEGASUS_THREAD_RETURN)0);
       }       }
          }
  
       // put myself back onto the available list       // put myself back onto the available list
       try       try
       {       {
          timed_mutex(&(pool->_monitor), 1000);  
          if(pool->_dying.value() == 0)          if(pool->_dying.value() == 0)
          {          {
             gettimeofday(deadlock_timer, NULL);             gettimeofday(deadlock_timer, NULL);
             if( blocking_sem != 0 )             if( blocking_sem != 0 )
                blocking_sem->signal();                blocking_sem->signal();
  
             pool->_running.remove((void *)myself);              // If we are not on _running then ~ThreadPool has removed
               // us and now "owns" our pointer.
               if ( pool->_running.remove((void *)myself) != 0 )
               {
             pool->_pool.insert_first(myself);             pool->_pool.insert_first(myself);
          }          }
          else          else
          {          {
                  Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
                     "ThreadPool::_loop: Failed to remove thread from running queue.");
                  PEG_METHOD_EXIT();
                  return((PEGASUS_THREAD_RETURN)0);
               }
            }
            else
            {
             PEG_METHOD_EXIT();             PEG_METHOD_EXIT();
             return((PEGASUS_THREAD_RETURN)0);             return((PEGASUS_THREAD_RETURN)0);
          }          }
       }       }
       catch(...)       catch(...)
       {       {
           Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
                "ThreadPool::_loop: Adding thread to idle pool failed.");
          PEG_METHOD_EXIT();          PEG_METHOD_EXIT();
          return((PEGASUS_THREAD_RETURN)0);          return((PEGASUS_THREAD_RETURN)0);
       }       }
Line 602 
Line 664 
    return((PEGASUS_THREAD_RETURN)0);    return((PEGASUS_THREAD_RETURN)0);
 } }
  
 void ThreadPool::allocate_and_awaken(void *parm,  Boolean ThreadPool::allocate_and_awaken(void *parm,
                                      PEGASUS_THREAD_RETURN \                                      PEGASUS_THREAD_RETURN \
                                      (PEGASUS_THREAD_CDECL *work)(void *),                                      (PEGASUS_THREAD_CDECL *work)(void *),
                                      Semaphore *blocking)                                      Semaphore *blocking)
   
    throw(IPCException)    throw(IPCException)
 { {
    PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::allocate_and_awaken");    PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::allocate_and_awaken");
    struct timeval start;  
    gettimeofday(&start, NULL);     // Allocate_and_awaken will not run if the _dying flag is set.
    Thread *th = 0;     // Once the lock is acquired, ~ThreadPool will not change
      // the value of _dying until the lock is released.
  
    try    try
    {    {
       timed_mutex(&(this->_monitor), 1000);  
       if(_dying.value())       if(_dying.value())
       {       {
          return;           Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
       }            "ThreadPool::allocate_and_awaken: ThreadPool is dying(1).");
       th = _pool.remove_first();           // ATTN: Error result has not yet been defined
    }           return true;
    catch(...)  
    {  
       return;  
   
    }    }
         struct timeval now;
         struct timeval start;
         gettimeofday(&start, NULL);
         Thread *th = 0;
  
         th = _pool.remove_first();
  
    // wait for the right interval and try again        if (th == 0)
    while (th == 0 && _dying.value() < 1)  
    {    {
       // will throw an IPCException&       // will throw an IPCException&
       _check_deadlock(&start) ;       _check_deadlock(&start) ;
Line 639 
Line 700 
       if(_max_threads == 0 || _current_threads < _max_threads)       if(_max_threads == 0 || _current_threads < _max_threads)
       {       {
          th = _init_thread();          th = _init_thread();
          continue;  
       }       }
       pegasus_yield();  
       try  
       {  
          timed_mutex(&(this->_monitor), 1000);  
          if(_dying.value())  
          {  
             return;  
          }  
          th = _pool.remove_first();  
       }       }
       catch(...)  
         if (th == 0)
       {       {
          return ;          // ATTN-DME-P3-20031103: This trace message should not be
       }          // be labeled TRC_DISCARDED_DATA, because it does not
           // necessarily imply that a failure has occurred.  However,
           // this label is being used temporarily to help isolate
           // the cause of client timeout problems.
   
           Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
              "ThreadPool::allocate_and_awaken: Insufficient resources: "
              " pool = %s, running threads = %d, idle threads = %d, dead threads = %d ",
              _key, _running.count(), _pool.count(), _dead.count());
            return false;
    }    }
  
    if(_dying.value() < 1)  
    {  
       // initialize the thread data with the work function and parameters       // initialize the thread data with the work function and parameters
       Tracer::trace(TRC_THREAD, Tracer::LEVEL4,       Tracer::trace(TRC_THREAD, Tracer::LEVEL4,
           "Initializing thread with work function and parameters: parm = %p",           "Initializing thread with work function and parameters: parm = %p",
Line 673 
Line 732 
       th->delete_tsd("blocking sem");       th->delete_tsd("blocking sem");
       if(blocking != 0 )       if(blocking != 0 )
          th->put_tsd("blocking sem", NULL, sizeof(Semaphore *), blocking);          th->put_tsd("blocking sem", NULL, sizeof(Semaphore *), blocking);
       try  
       {  
          timed_mutex(&(this->_monitor), 1000);  
          if(_dying.value())  
          {  
             th->cancel();  
             th->join();  
             delete th;  
             return;  
          }  
  
          // put the thread on the running list          // put the thread on the running list
   
   
          _running.insert_first(th);          _running.insert_first(th);
   
       // signal the thread's sleep semaphore to awaken it       // signal the thread's sleep semaphore to awaken it
          Semaphore *sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");          Semaphore *sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");
  
          if(sleep_sem == 0)          if(sleep_sem == 0)
          {          {
             th->dereference_tsd();             th->dereference_tsd();
            Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
              "ThreadPool::allocate_and_awaken: thread data is corrupted.");
             PEG_METHOD_EXIT();             PEG_METHOD_EXIT();
             throw NullPointer();             throw NullPointer();
          }          }
Line 703 
Line 753 
       }       }
       catch(...)       catch(...)
       {       {
         Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
             "ThreadPool::allocate_and_awaken: Operation Failed.");
          PEG_METHOD_EXIT();          PEG_METHOD_EXIT();
          return;        // ATTN: Error result has not yet been defined
       }        return true;
   
    }  
    else  
    {  
       th->cancel();  
       th->join();  
       delete th;  
    }    }
   
    PEG_METHOD_EXIT();    PEG_METHOD_EXIT();
      return true;
 } }
  
 // caller is responsible for only calling this routine during slack periods // caller is responsible for only calling this routine during slack periods
Line 725 
Line 770 
 Uint32 ThreadPool::kill_dead_threads(void) Uint32 ThreadPool::kill_dead_threads(void)
          throw(IPCException)          throw(IPCException)
 { {
    struct timeval now;     // Since the kill_dead_threads, ThreadPool or allocate_and_awaken
    gettimeofday(&now, NULL);     // manipulate the threads on the ThreadPool queues, they should never
    Uint32 bodies = 0;     // be allowed to run at the same time.
   
      // << Thu Oct 23 14:41:02 2003 mdd >>
      // not true, the queues are thread safe. they are syncrhonized.
   
      auto_int do_not_destruct(&_idle_control);
  
    // first go thread the dead q and clean it up as much as possible  
    try    try
    {    {
       timed_mutex(&(this->_monitor), 1000);  
       if(_dying.value() )       if(_dying.value() )
       {       {
          return 0;          return 0;
       }       }
  
       while(_dead.count() > 0 && _dying.value() == 0 )        struct timeval now;
         gettimeofday(&now, NULL);
         Uint32 bodies = 0;
   
         // first go thread the dead q and clean it up as much as possible
         try
         {
            while(_dying.value() == 0 && _dead.count() > 0)
       {       {
          Tracer::trace(TRC_THREAD, Tracer::LEVEL4, "ThreadPool:: removing and joining dead thread");          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 )
             throw NullPointer();              {
          dead->join();          dead->join();
          delete dead;          delete dead;
       }       }
    }    }
         }
    catch(...)    catch(...)
    {    {
    }    }
  
         if (_dying.value())
         {
            return 0;
         }
  
    DQueue<Thread> * map[2] =    DQueue<Thread> * map[2] =
       {       {
Line 763 
Line 823 
    DQueue<Thread> *q = 0;    DQueue<Thread> *q = 0;
    int i = 0;    int i = 0;
    AtomicInt needed(0);    AtomicInt needed(0);
         Thread *th = 0;
         internal_dq idq;
  
 #ifdef PEGASUS_DISABLE_KILLING_HUNG_THREADS  #ifdef PEGASUS_KILL_LONG_RUNNING_THREADS
    // This change prevents the thread pool from killing "hung" threads.        // Defining PEGASUS_KILL_LONG_RUNNING_THREADS causes the thread pool
    // The definition of a "hung" thread is one that has been on the run queue        // to kill threads that are on the _running queue longer than the
    // for longer than the time interval set when the thread pool was created.        // _deadlock_detect time interval specified for the thread pool.
    // Cancelling "hung" threads has proven to be problematic.        // Cancelling long-running threads has proven to be problematic and
         // may cause a crash depending on the state of the thread when it is
    // With this change the thread pool will not cancel "hung" threads.  This        // killed.  Use this option with care.
    // 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++)    for( ; i < 2; i++)
   #else
         for( ; i < 1; i++)
 #endif #endif
    {    {
       try  
       {  
          try_mutex(&(this->_monitor));  
       }  
       catch(IPCException&)  
       {  
          return bodies;  
       }  
   
       q = map[i];       q = map[i];
       if(q->count() > 0 )       if(q->count() > 0 )
       {       {
          try          try
          {          {
             if(_dying.value())  
             {  
                return bodies;  
             }  
   
             q->try_lock();             q->try_lock();
          }          }
          catch(...)          catch(...)
Line 811 
Line 852 
  
          struct timeval dt = { 0, 0 };          struct timeval dt = { 0, 0 };
          struct timeval *dtp;          struct timeval *dtp;
          Thread *th = 0;  
          th = q->next(th);          th = q->next(th);
          while (th != 0 )          while (th != 0 )
          {          {
Line 864 
Line 905 
                }                }
  
                th = q->remove_no_lock((void *)th);                th = q->remove_no_lock((void *)th);
                     idq.insert_first((void*)th);
                  }
                  th = q->next(th);
               }
               q->unlock();
            }
  
                if(th != 0)           th = (Thread*)idq.remove_last();
            while(th != 0)
                {                {
                   if( i == 0 )                   if( i == 0 )
                   {                   {
Line 878 
Line 926 
  
                      // signal the thread's sleep semaphore to awaken it                      // signal the thread's sleep semaphore to awaken it
                      Semaphore *sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");                      Semaphore *sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");
                  PEGASUS_ASSERT(sleep_sem != 0);
                      if(sleep_sem == 0)  
                      {  
                         q->unlock();  
                         th->dereference_tsd();  
                         throw NullPointer();  
                      }  
  
                      bodies++;                      bodies++;
                      th->dereference_tsd();                      th->dereference_tsd();
                      _dead.insert_first(th);                 // Putting thread on _dead queue delays availability to others
                  //_dead.insert_first(th);
                      sleep_sem->signal();                      sleep_sem->signal();
                  th->join();  // Note: Clean up the thread here rather than
                  delete th;   // leave it sitting unused on the _dead queue
                      th = 0;                      th = 0;
                   }                   }
                   else                   else
                   {                   {
                      // deadlocked threads                      // deadlocked threads
                      Tracer::trace(TRC_THREAD, Tracer::LEVEL4, "Killing a deadlocked thread");                 struct timeval deadlock_timeout;
                  Tracer::trace(TRC_THREAD, Tracer::LEVEL2,
                                "A thread has run longer than %u seconds and "
                                    "will be cancelled.",
                                Uint32(_deadlock_detect.tv_sec));
                  Logger::put_l(Logger::ERROR_LOG, System::CIMSERVER,
                                Logger::SEVERE,
                                "Common.Thread.CANCEL_LONG_RUNNING_THREAD",
                                "A thread has run longer than {0} seconds and "
                                    "will be cancelled.",
                                Uint32(_deadlock_detect.tv_sec));
                      th->cancel();                      th->cancel();
                      delete th;                      delete th;
                   }                   }
               th = (Thread*)idq.remove_last();
                }                }
             }             }
             th = q->next(th);  
             pegasus_yield();  
          }  
          q->unlock();  
       }  
    }  
    if(_dying.value() )  
       return bodies;  
  
    while (needed.value() > 0)   {    while (needed.value() > 0)   {
       _link_pool(_init_thread());       _link_pool(_init_thread());
Line 917 
Line 965 
    }    }
     return bodies;     return bodies;
 } }
       catch (...)
       {
       }
       return 0;
   }
  
  
 Boolean ThreadPool::check_time(struct timeval *start, struct timeval *interval) Boolean ThreadPool::check_time(struct timeval *start, struct timeval *interval)
Line 992 
Line 1045 
    th->put_tsd("deadlock timer", thread_data::default_delete, sizeof(struct timeval), (void *)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    // thread will enter _loop(void *) and sleep on sleep_sem until we signal it
  
    th->run();     if (!th->run())
      {
         delete th;
         return 0;
      }
    _current_threads++;    _current_threads++;
    pegasus_yield();    pegasus_yield();
  
Line 1002 
Line 1059 
  void ThreadPool::_link_pool(Thread *th) throw(IPCException)  void ThreadPool::_link_pool(Thread *th) throw(IPCException)
 { {
    if(th == 0)    if(th == 0)
      {
         Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
             "ThreadPool::_link_pool: Thread pointer is null.");
       throw NullPointer();       throw NullPointer();
      }
    try    try
    {    {
   
       timed_mutex(&(this->_monitor), 1000);  
       if(_dying.value())  
       {  
          th->cancel();  
          th->join();  
          delete th;  
       }  
   
       _pool.insert_first(th);       _pool.insert_first(th);
   
    }    }
    catch(...)    catch(...)
    {    {
         Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
             "ThreadPool::_link_pool: _pool.insert_first failed.");
    }    }
 } }
  

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