(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.51 and 1.52

version 1.51, 2003/10/16 19:21:28 version 1.52, 2003/10/17 14:22:49
Line 241 
Line 241 
 } }
 // l10n end // l10n end
  
   
   
   // two special synchronization classes for ThreadPool
   //
   
   class timed_mutex
   {
      public:
         timed_mutex(Mutex* mut, int msec)
            :_mut(mut)
         {
            _mut->timed_lock(msec, pegasus_thread_self());
         }
         ~timed_mutex(void)
         {
            _mut->unlock();
         }
         Mutex* _mut;
   };
   
   
   class try_mutex
   {
      public:
         try_mutex(Mutex* mut)
            :_mut(mut)
         {
            _mut->try_lock(pegasus_thread_self());
         }
         ~try_mutex(void)
         {
            _mut->unlock();
         }
   
         Mutex* _mut;
   };
   
   
 DQueue<ThreadPool> ThreadPool::_pools(true); DQueue<ThreadPool> ThreadPool::_pools(true);
  
  
Line 305 
Line 343 
 } }
  
  
   // Note:   <<< Fri Oct 17 09:19:03 2003 mdd >>>
   // the pegasus_yield() calls that preceed each th->join() are to
   // give a thread on the running list a chance to reach a cancellation
   // point before the join
  
 ThreadPool::~ThreadPool(void) ThreadPool::~ThreadPool(void)
 { {
   
    try    try
    {    {
         // set the dying flag so all thread know the destructor has been entered
       {       {
          auto_mutex(&(this->_monitor));          auto_mutex(&(this->_monitor));
          _dying++;          _dying++;
       }       }
         // remove from the global pools list
       _pools.remove(this);       _pools.remove(this);
   
         // start with idle threads.
       Thread *th = 0;       Thread *th = 0;
       th = _pool.remove_first();       th = _pool.remove_first();
         Semaphore* sleep_sem;
   
       while(th != 0)       while(th != 0)
       {       {
          Semaphore *sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");           sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");
   
          if(sleep_sem == 0)          if(sleep_sem == 0)
          {          {
             th->dereference_tsd();             th->dereference_tsd();
Line 331 
Line 376 
  
          // 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();
   
          // Signal to get the thread past the end. See the comment          // Signal to get the thread past the end. See the comment
          // "wait to be awakend by the thread pool destructor"          // "wait to be awakend by the thread pool destructor"
          // Note: the current implementation of Thread for Windows          // Note: the current implementation of Thread for Windows
          // does not implement "pthread" cancelation points so this          // does not implement "pthread" cancelation points so this
          // is needed.          // is needed.
          sleep_sem->signal();          sleep_sem->signal();
   
          th->dereference_tsd();          th->dereference_tsd();
          // signal the thread's sleep semaphore  
          th->cancel();          th->cancel();
          th->join();          th->join();
          th->empty_tsd();  
          delete th;          delete th;
          th = _pool.remove_first();          th = _pool.remove_first();
       }       }
   
       th = _dead.remove_first();       th = _dead.remove_first();
       while(th != 0)       while(th != 0)
       {       {
          Semaphore *sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");           sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");
  
          if(sleep_sem == 0)          if(sleep_sem == 0)
          {          {
Line 358 
Line 400 
             throw NullPointer();             throw NullPointer();
          }          }
  
            // signal the thread's sleep semaphore
            sleep_sem->signal();
          sleep_sem->signal();          sleep_sem->signal();
          th->dereference_tsd();          th->dereference_tsd();
   
          // signal the thread's sleep semaphore  
          th->cancel();          th->cancel();
          th->join();          th->join();
          th->empty_tsd();  
          delete th;          delete th;
          th = _dead.remove_first();          th = _dead.remove_first();
       }       }
       {  
  
          auto_mutex(&(this->_monitor));        {
       th = _running.remove_first();       th = _running.remove_first();
       while(th != 0)       while(th != 0)
       {       {
          // signal the thread's sleep semaphore          // signal the thread's sleep semaphore
          Semaphore *sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");  
               sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");
          if(sleep_sem == 0 )          if(sleep_sem == 0 )
          {          {
             th->dereference_tsd();             th->dereference_tsd();
Line 384 
Line 424 
          }          }
  
          sleep_sem->signal();          sleep_sem->signal();
               sleep_sem->signal();
          th->dereference_tsd();          th->dereference_tsd();
   
          th->cancel();          th->cancel();
               pegasus_yield();
  
          // ensure that th->run() has a chance to execute so that the join will not  
          // block  
          th->join();          th->join();
          th->empty_tsd();  
          delete th;          delete th;
          th = _running.remove_first();          th = _running.remove_first();
       }       }
Line 402 
Line 440 
    catch(...)    catch(...)
    {    {
    }    }
   
 } }
  
 // make this static to the class // make this static to the class
Line 428 
Line 465 
       PEG_METHOD_EXIT();       PEG_METHOD_EXIT();
       throw NullPointer();       throw NullPointer();
    }    }
      if(pool->_dying.value())
      {
         PEG_METHOD_EXIT();
         return((PEGASUS_THREAD_RETURN)0);
      }
  
    Semaphore *sleep_sem = 0;    Semaphore *sleep_sem = 0;
    Semaphore *blocking_sem = 0;    Semaphore *blocking_sem = 0;
Line 441 
Line 483 
       deadlock_timer = (struct timeval *)myself->reference_tsd("deadlock timer");       deadlock_timer = (struct timeval *)myself->reference_tsd("deadlock timer");
       myself->dereference_tsd();       myself->dereference_tsd();
    }    }
    catch(IPCException &)  
    {  
       PEG_METHOD_EXIT();  
       return(0);  
    }  
    catch(...)    catch(...)
    {    {
       PEG_METHOD_EXIT();       PEG_METHOD_EXIT();
       return(0);        return((PEGASUS_THREAD_RETURN)0);
    }    }
  
    if(sleep_sem == 0 || deadlock_timer == 0)    if(sleep_sem == 0 || deadlock_timer == 0)
    {    {
       PEG_METHOD_EXIT();       PEG_METHOD_EXIT();
       throw NullPointer();        return((PEGASUS_THREAD_RETURN)0);
    }    }
  
    while(pool->_dying.value() < 1)    while(pool->_dying.value() < 1)
    {    {
         try
         {
       sleep_sem->wait();       sleep_sem->wait();
         }
         catch(IPCException& )
         {
            PEG_METHOD_EXIT();
            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;
Line 482 
Line 533 
       catch(IPCException &)       catch(IPCException &)
       {       {
          PEG_METHOD_EXIT();          PEG_METHOD_EXIT();
          return(0);           return((PEGASUS_THREAD_RETURN)0);
       }       }
  
       if(_work == 0)       if(_work == 0)
Line 501 
Line 552 
       try       try
       {       {
          {          {
             auto_mutex(&(pool->_monitor));              timed_mutex(&(pool->_monitor), 1000);
             if(pool->_dying.value())             if(pool->_dying.value())
             {             {
                break;                 _undertaker(parm);
             }             }
          }          }
          _work(parm);          _work(parm);
Line 514 
Line 565 
          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
       {       {
          auto_mutex(&(pool->_monitor));           timed_mutex(&(pool->_monitor), 1000);
          if(pool->_dying.value() == 0)          if(pool->_dying.value() == 0)
          {          {
             gettimeofday(deadlock_timer, NULL);             gettimeofday(deadlock_timer, NULL);
Line 535 
Line 584 
             return((PEGASUS_THREAD_RETURN)0);             return((PEGASUS_THREAD_RETURN)0);
          }          }
       }       }
       catch(IPCException &)  
       {  
          PEG_METHOD_EXIT();  
          return((PEGASUS_THREAD_RETURN)0);  
       }  
       catch(...)       catch(...)
       {       {
            PEG_METHOD_EXIT();
          return((PEGASUS_THREAD_RETURN)0);          return((PEGASUS_THREAD_RETURN)0);
       }       }
  
Line 554 
Line 599 
    myself->test_cancel();    myself->test_cancel();
  
    PEG_METHOD_EXIT();    PEG_METHOD_EXIT();
    myself->exit_self(0);  
    return((PEGASUS_THREAD_RETURN)0);    return((PEGASUS_THREAD_RETURN)0);
 } }
  
Line 572 
Line 616 
  
    try    try
    {    {
       auto_mutex(&(this->_monitor));        timed_mutex(&(this->_monitor), 1000);
       if(_dying.value())       if(_dying.value())
       {       {
          return;          return;
Line 600 
Line 644 
       pegasus_yield();       pegasus_yield();
       try       try
       {       {
          auto_mutex(&(this->_monitor));           timed_mutex(&(this->_monitor), 1000);
          if(_dying.value())          if(_dying.value())
          {          {
             return;             return;
Line 631 
Line 675 
          th->put_tsd("blocking sem", NULL, sizeof(Semaphore *), blocking);          th->put_tsd("blocking sem", NULL, sizeof(Semaphore *), blocking);
       try       try
       {       {
          auto_mutex(&(this->_monitor));           timed_mutex(&(this->_monitor), 1000);
          if(_dying.value())          if(_dying.value())
          {          {
             th->cancel();             th->cancel();
Line 688 
Line 732 
    // 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
    try    try
    {    {
       auto_mutex(&(this->_monitor));        timed_mutex(&(this->_monitor), 1000);
       if(_dying.value() )       if(_dying.value() )
       {       {
          return 0;          return 0;
Line 739 
Line 783 
    for( ; i < 2; i++)    for( ; i < 2; i++)
 #endif #endif
    {    {
       auto_mutex(&(this->_monitor));        try
         {
            try_mutex(&(this->_monitor));
         }
         catch(IPCException&)
         {
            return bodies;
         }
   
       q = map[i];       q = map[i];
       if(q->count() > 0 )       if(q->count() > 0 )
       {       {
Line 850 
Line 902 
                }                }
             }             }
             th = q->next(th);             th = q->next(th);
             pegasus_sleep(1);              pegasus_yield();
          }          }
          q->unlock();          q->unlock();
       }       }
Line 954 
Line 1006 
    try    try
    {    {
  
       auto_mutex(&(this->_monitor));        timed_mutex(&(this->_monitor), 1000);
       if(_dying.value())       if(_dying.value())
       {       {
          th->cancel();          th->cancel();

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