(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.41 and 1.49

version 1.41, 2003/08/04 20:18:51 version 1.49, 2003/10/16 03:11:44
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 // l10n
 PEGASUS_THREAD_KEY_TYPE Thread::_platform_thread_key;  PEGASUS_THREAD_KEY_TYPE Thread::_platform_thread_key = -1;
 Boolean Thread::_key_initialized = false; Boolean Thread::_key_initialized = false;
 Boolean Thread::_key_error = false; Boolean Thread::_key_error = false;
  
Line 206 
Line 218 
    {    {
                 // deletes the old tsd and creates a new one                 // deletes the old tsd and creates a new one
                 currentThrd->put_tsd("acceptLanguages",                 currentThrd->put_tsd("acceptLanguages",
                         thread_data::default_delete,                          language_delete,
                         sizeof(AcceptLanguages *),                         sizeof(AcceptLanguages *),
                         langs);                         langs);
    }    }
Line 296 
Line 308 
  
 ThreadPool::~ThreadPool(void) ThreadPool::~ThreadPool(void)
 { {
   
    try    try
    {    {
       _pools.remove(this);        {
            auto_mutex(&(this->_monitor));
       _dying++;       _dying++;
         }
   
         _pools.remove(this);
       Thread *th = 0;       Thread *th = 0;
       th = _pool.remove_first();       th = _pool.remove_first();
       while(th != 0)       while(th != 0)
Line 312 
Line 329 
             throw NullPointer();             throw NullPointer();
          }          }
  
            // 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
            // "wait to be awakend by the thread pool destructor"
            // Note: the current implementation of Thread for Windows
            // does not implement "pthread" cancelation points so this
            // is needed.
          sleep_sem->signal();          sleep_sem->signal();
   
          th->dereference_tsd();          th->dereference_tsd();
          // signal the thread's sleep semaphore          // signal the thread's sleep semaphore
          th->cancel();          th->cancel();
Line 323 
Line 347 
          th = _pool.remove_first();          th = _pool.remove_first();
       }       }
  
       th = _running.remove_first();        th = _dead.remove_first();
       while(th != 0)       while(th != 0)
       {       {
            Semaphore *sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");
   
            if(sleep_sem == 0)
            {
               th->dereference_tsd();
               throw NullPointer();
            }
   
   
            sleep_sem->signal();
            th->dereference_tsd();
   
          // signal the thread's sleep semaphore          // signal the thread's sleep semaphore
          th->cancel();          th->cancel();
          th->join();          th->join();
          th->empty_tsd();          th->empty_tsd();
          delete th;          delete th;
          th = _running.remove_first();           th = _dead.remove_first();
       }       }
         {
  
       th = _dead.remove_first();           auto_mutex(&(this->_monitor));
         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");
            if(sleep_sem == 0 )
            {
               th->dereference_tsd();
               throw NullPointer();
            }
   
            sleep_sem->signal();
            th->dereference_tsd();
   
          th->cancel();          th->cancel();
   
            // ensure that th->run() has a chance to execute so that the join will not
            // block
          th->join();          th->join();
          th->empty_tsd();          th->empty_tsd();
          delete th;          delete th;
          th = _dead.remove_first();           th = _running.remove_first();
       }       }
    }    }
   
      }
   
    catch(...)    catch(...)
    {    {
    }    }
   
 } }
  
 // make this static to the class // make this static to the class
Line 373 
Line 428 
       PEG_METHOD_EXIT();       PEG_METHOD_EXIT();
       throw NullPointer();       throw NullPointer();
    }    }
   
    Semaphore *sleep_sem = 0;    Semaphore *sleep_sem = 0;
    Semaphore *blocking_sem = 0;    Semaphore *blocking_sem = 0;
  
Line 388 
Line 444 
    catch(IPCException &)    catch(IPCException &)
    {    {
       PEG_METHOD_EXIT();       PEG_METHOD_EXIT();
       myself->exit_self(0);        return(0);
    }    }
    catch(...)    catch(...)
    {    {
       PEG_METHOD_EXIT();       PEG_METHOD_EXIT();
       myself->exit_self(0);        return(0);
    }    }
  
    if(sleep_sem == 0 || deadlock_timer == 0)    if(sleep_sem == 0 || deadlock_timer == 0)
Line 402 
Line 458 
       throw NullPointer();       throw NullPointer();
    }    }
  
    while(pool->_dying < 1)     while(pool->_dying.value() < 1)
    {    {
       sleep_sem->wait();       sleep_sem->wait();
  
       // 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 > 0)  
          break;  
  
       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 427 
Line 482 
       catch(IPCException &)       catch(IPCException &)
       {       {
          PEG_METHOD_EXIT();          PEG_METHOD_EXIT();
          myself->exit_self(0);           return(0);
       }       }
  
       if(_work == 0)       if(_work == 0)
Line 445 
Line 500 
       gettimeofday(deadlock_timer, NULL);       gettimeofday(deadlock_timer, NULL);
       try       try
       {       {
            {
               auto_mutex(&(pool->_monitor));
               if(pool->_dying.value())
               {
                  break;
               }
            }
          _work(parm);          _work(parm);
       }       }
       catch(...)       catch(...)
       {       {
          gettimeofday(deadlock_timer, NULL);           return((PEGASUS_THREAD_RETURN)0);
       }       }
  
       gettimeofday(deadlock_timer, NULL);  
       if( blocking_sem != 0 )  
          blocking_sem->signal();  
  
       // put myself back onto the available list       // put myself back onto the available list
       try       try
       {       {
            auto_mutex(&(pool->_monitor));
            if(pool->_dying.value() == 0)
            {
               gettimeofday(deadlock_timer, NULL);
               if( blocking_sem != 0 )
                  blocking_sem->signal();
   
          pool->_running.remove((void *)myself);          pool->_running.remove((void *)myself);
          pool->_link_pool(myself);              pool->_pool.insert_first(myself);
            }
            else
            {
               PEG_METHOD_EXIT();
               return((PEGASUS_THREAD_RETURN)0);
            }
       }       }
       catch(IPCException &)       catch(IPCException &)
       {       {
          PEG_METHOD_EXIT();          PEG_METHOD_EXIT();
          myself->exit_self(0);           return((PEGASUS_THREAD_RETURN)0);
       }       }
    }    }
   
      // TODO: Why is this needed? Why not just continue?
    // wait to be awakend by the thread pool destructor    // wait to be awakend by the thread pool destructor
    sleep_sem->wait();    sleep_sem->wait();
   
    myself->test_cancel();    myself->test_cancel();
  
    PEG_METHOD_EXIT();    PEG_METHOD_EXIT();
Line 487 
Line 563 
    PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::allocate_and_awaken");    PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::allocate_and_awaken");
    struct timeval start;    struct timeval start;
    gettimeofday(&start, NULL);    gettimeofday(&start, NULL);
      Thread *th = 0;
   
      try
      {
         auto_mutex(&(this->_monitor));
         if(_dying.value())
         {
            return;
         }
         th = _pool.remove_first();
      }
      catch(...)
      {
         return;
   
      }
  
    Thread *th = _pool.remove_first();  
  
    // wait for the right interval and try again    // wait for the right interval and try again
    while (th == 0 && _dying < 1)     while (th == 0 && _dying.value() < 1)
    {    {
         // will throw an IPCException&
       _check_deadlock(&start) ;       _check_deadlock(&start) ;
  
       if(_max_threads == 0 || _current_threads < _max_threads)       if(_max_threads == 0 || _current_threads < _max_threads)
Line 501 
Line 593 
          continue;          continue;
       }       }
       pegasus_yield();       pegasus_yield();
         try
         {
            auto_mutex(&(this->_monitor));
            if(_dying.value())
            {
               return;
            }
       th = _pool.remove_first();       th = _pool.remove_first();
    }    }
         catch(...)
         {
            return ;
         }
      }
  
      if(_dying.value() < 1)
    if(_dying < 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,
Line 521 
Line 624 
       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
         {
            auto_mutex(&(this->_monitor));
            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");
  
Line 538 
Line 652 
       sleep_sem->signal();       sleep_sem->signal();
       th->dereference_tsd();       th->dereference_tsd();
    }    }
         catch(...)
         {
            PEG_METHOD_EXIT();
            return;
         }
   
      }
    else    else
       _pool.insert_first(th);     {
         th->cancel();
         th->join();
         delete th;
      }
  
    PEG_METHOD_EXIT();    PEG_METHOD_EXIT();
 } }
Line 556 
Line 681 
    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)     try
      {
         auto_mutex(&(this->_monitor));
         if(_dying.value() )
         {
            return 0;
         }
   
         while(_dead.count() > 0 && _dying.value() == 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 == 0)
          throw NullPointer();          throw NullPointer();
       dead->join();       dead->join();
       delete dead;       delete dead;
    }    }
      }
      catch(...)
      {
      }
   
  
    DQueue<Thread> * map[2] =    DQueue<Thread> * map[2] =
       {       {
Line 595 
Line 734 
    for( ; i < 2; i++)    for( ; i < 2; i++)
 #endif #endif
    {    {
         auto_mutex(&(this->_monitor));
       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 703 
Line 848 
             pegasus_sleep(1);             pegasus_sleep(1);
          }          }
          q->unlock();          q->unlock();
          while (needed.value() > 0)        }
          {     }
      if(_dying.value() )
         return bodies;
   
      while (needed.value() > 0)   {
             _link_pool(_init_thread());             _link_pool(_init_thread());
             needed--;             needed--;
             pegasus_sleep(0);             pegasus_sleep(0);
          }          }
       }  
    }  
     return bodies;     return bodies;
 } }
  
Line 799 
Line 946 
 { {
    if(th == 0)    if(th == 0)
       throw NullPointer();       throw NullPointer();
      try
      {
   
         auto_mutex(&(this->_monitor));
         if(_dying.value())
         {
            th->cancel();
            th->join();
            delete th;
         }
   
    _pool.insert_first(th);    _pool.insert_first(th);
   
      }
      catch(...)
      {
      }
 } }
  
  

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