(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.13 and 1.72

version 1.13, 2002/03/18 15:38:18 version 1.72, 2004/10/25 18:26:02
Line 1 
Line 1 
 //%/////////////////////////////////////////////////////////////////////////////  //%2004////////////////////////////////////////////////////////////////////////
 // //
 // Copyright (c) 2000, 2001 The Open group, BMC Software, Tivoli Systems, IBM,  // Copyright (c) 2000, 2001, 2002 BMC Software; Hewlett-Packard Development
 // Compaq Computer Corporation  // 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.
   // Copyright (c) 2004 BMC Software; Hewlett-Packard Development Company, L.P.;
   // IBM Corp.; EMC Corporation; VERITAS Software 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 29 
 // //
 // 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
 // //
 //%///////////////////////////////////////////////////////////////////////////// //%/////////////////////////////////////////////////////////////////////////////
  
 #include "Thread.h" #include "Thread.h"
   #include <exception>
 #include <Pegasus/Common/IPC.h> #include <Pegasus/Common/IPC.h>
   #include <Pegasus/Common/Tracer.h>
  
 #if defined(PEGASUS_OS_TYPE_WINDOWS) #if defined(PEGASUS_OS_TYPE_WINDOWS)
 # include "ThreadWindows.cpp" # include "ThreadWindows.cpp"
Line 41 
Line 49 
 # error "Unsupported platform" # error "Unsupported platform"
 #endif #endif
  
   PEGASUS_USING_STD;
 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)
      {
         AutoPtr<AcceptLanguages> al(static_cast<AcceptLanguages *>(data));
      }
   }
   // l10n end
   
 Boolean Thread::_signals_blocked = false; Boolean Thread::_signals_blocked = false;
   // l10n
   #ifndef PEGASUS_OS_ZOS
   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_error = false;
   
  
 // for non-native implementations // for non-native implementations
 #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 81 
Line 103 
      }      }
     if(execute == true)     if(execute == true)
         cu->execute();         cu->execute();
     delete cu;  
 } }
  
 #endif #endif
Line 114 
Line 135 
  
 #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
   
   #if 0
   // 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;
   };
   #endif
   
   class try_mutex
   {
      public:
         try_mutex(Mutex* mut)
            :_mut(mut)
         {
            _mut->try_lock(pegasus_thread_self());
         }
         ~try_mutex(void)
         {
            _mut->unlock();
         }
   
         Mutex* _mut;
   };
   
   class auto_int
   {
      public:
         auto_int(AtomicInt* num)
            : _int(num)
         {
            _int->operator++();
         }
         ~auto_int(void)
         {
            _int->operator--();
         }
         AtomicInt *_int;
   };
   
   
   AtomicInt _idle_control;
   
   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 135 
Line 347 
    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 145 
Line 357 
    {    {
       _link_pool(_init_thread());       _link_pool(_init_thread());
    }    }
      _pools.insert_last(this);
 } }
  
  
   // 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)
 { {
      PEG_METHOD_ENTER(TRC_THREAD, "Thread::~ThreadPool");
      try
      {
         // Set the dying flag so all thread know the destructor has been entered
    _dying++;    _dying++;
   
         // remove from the global pools list
         _pools.remove(this);
   
         // start with idle threads.
    Thread *th = 0;    Thread *th = 0;
         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");
            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.
       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  
       th->cancel();  
       th->join();       th->join();
       th->empty_tsd();  
       delete th;       delete th;
            }
       th = _pool.remove_first();       th = _pool.remove_first();
    }    }
  
    th = _running.remove_first();        while(_idle_control.value())
            pegasus_yield();
   
         th = _dead.remove_first();
    while(th != 0)    while(th != 0)
    {    {
            sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");
            PEGASUS_ASSERT(sleep_sem != 0);
   
            if(sleep_sem == 0)
            {
               th->dereference_tsd();
            }
            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
       th->cancel();              sleep_sem->signal();
               sleep_sem->signal();
               th->dereference_tsd();
       th->join();       th->join();
       th->empty_tsd();  
       delete th;       delete th;
       th = _running.remove_first();  
    }    }
   
    th = _dead.remove_first();    th = _dead.remove_first();
         }
   
         {
            th = _running.remove_first();
    while(th != 0)    while(th != 0)
    {    {
       // signal the thread's sleep semaphore       // signal the thread's sleep semaphore
       th->cancel();  
               sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");
               PEGASUS_ASSERT(sleep_sem != 0);
   
               if(sleep_sem == 0 )
               {
                  th->dereference_tsd();
               }
               else
               {
                  sleep_sem->signal();
                  sleep_sem->signal();
                  th->dereference_tsd();
                  //th->cancel();
                  pegasus_yield();
   
       th->join();       th->join();
       th->empty_tsd();  
       delete th;       delete th;
       th = _dead.remove_first();              }
               th = _running.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)
 { {
      PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::_loop");
   
    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();
       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 )
      {
         Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
             "ThreadPool::_loop: ThreadPool pointer is null");
         PEG_METHOD_EXIT();
       throw NullPointer();       throw NullPointer();
      }
      if(pool->_dying.value())
      {
         Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
             "ThreadPool::_loop: ThreadPool is dying(1)");
         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 218 
Line 519 
       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 &)  
    {  
  
       myself->exit_self(0);     catch(...)
      {
         Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
                       "ThreadPool::_loop: Failure getting sleep_sem or deadlock_timer.");
         _graveyard(myself);
         PEG_METHOD_EXIT();
         return((PEGASUS_THREAD_RETURN)0);
    }    }
   
    if(sleep_sem == 0 || deadlock_timer == 0)    if(sleep_sem == 0 || deadlock_timer == 0)
       throw NullPointer();     {
         Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
             "ThreadPool::_loop: sleep_sem or deadlock_timer are null.");
         _graveyard(myself);
         PEG_METHOD_EXIT();
         return((PEGASUS_THREAD_RETURN)0);
      }
   
      while(1)
      {
         if(pool->_dying.value())
            break;
   
         try
         {
                                   Boolean ignoreInterrupt = false;
                                   sleep_sem->wait(ignoreInterrupt);
         }
         catch (WaitInterrupted &e)
         {
           /* From the sem_wait manpage:
    The sem_trywait() and sem_wait() functions may fail if:
  
    while(pool->_dying < 1)         EINTR  A signal interrupted this function.
           */
               PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,
                   "Sleep semaphore wait failed. Doing a continue");
               continue;
         }
         catch(IPCException& )
    {    {
       sleep_sem->wait();           Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
              "ThreadPool::_loop: failure on sleep_sem->wait().");
            _graveyard(myself);
            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 > 0)        /* Hence no need to move the thread to the _dead queue, as the _running
          break;         * queue is only dused by kill_dead_threads which makes sure that the
          * the threads are cleaned up (unlocking any locked lists in the TSD, etc)
          * before killing it.
          */
  
       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 250 
Line 591 
       }       }
       catch(IPCException &)       catch(IPCException &)
       {       {
            Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
          myself->exit_self(0);             "ThreadPool::_loop: Failure accessing work func, work parm, or blocking sem.");
           /*
            * We cannot move ourselves to the dead queue b/c the TSD might be still
            * locked and _graveyard is not equipped to de-lock (dereference_tsd) the TSD.
            * Only the kill_dead_threads has enough logic to handle such situations.
            _graveyard( myself);
           */
            PEG_METHOD_EXIT();
            return((PEGASUS_THREAD_RETURN)0);
       }       }
  
       if(_work == 0)       if(_work == 0)
          throw NullPointer();        {
            Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
              "ThreadPool::_loop: work func is null.");
            PEG_METHOD_EXIT();
            return((PEGASUS_THREAD_RETURN)0);
         }
   
         if(_work ==
            (PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *)(void *)) &_undertaker)
         {
           /*
           * The undertaker is set by  ThreadPool::kill_dead_threads which awakens this thread,
           *  joins it and then removes it from the queue. Hence no reason to go to the
           _graveyard( myself);
           */
            PEG_METHOD_EXIT();
            _work(parm);
         }
   
       gettimeofday(deadlock_timer, NULL);       gettimeofday(deadlock_timer, NULL);
   
         if (pool->_dying.value() == 0)
         {
            try
            {
               PEG_TRACE_STRING(TRC_THREAD, Tracer::LEVEL4,
                   "Worker started");
       _work(parm);       _work(parm);
               PEG_TRACE_STRING(TRC_THREAD, Tracer::LEVEL4,
                   "Worker finished");
            }
            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);
            }
   #if !defined(PEGASUS_OS_LSB)
            catch (exception& e)
            {
               PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,
                  String("Exception from _work in ThreadPool::_loop: ") +
                     e.what());
               PEG_METHOD_EXIT();
               return((PEGASUS_THREAD_RETURN)0);
            }
   #endif
            catch(...)
            {
               Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
                 "ThreadPool::_loop: execution of _work failed.");
               PEG_METHOD_EXIT();
               return((PEGASUS_THREAD_RETURN)0);
            }
          }
  
         // put myself back onto the available list
         try
         {
            if(pool->_dying.value() == 0)
            {
               gettimeofday(deadlock_timer, NULL);
       if( blocking_sem != 0 )       if( blocking_sem != 0 )
          blocking_sem->signal();          blocking_sem->signal();
  
       // put myself back onto the available list              // If we are not on _running then ~ThreadPool has removed
       try              // us and now "owns" our pointer.
               if ( pool->_running.remove((void *)myself) != 0 )
       {       {
          pool->_running.remove((void *)myself);                 pool->_pool.insert_first(myself);
          pool->_link_pool(myself);  
       }       }
       catch(IPCException &)              else
       {       {
                  Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
          myself->exit_self(0);                    "ThreadPool::_loop: Failed to remove thread from running queue.");
                  PEG_METHOD_EXIT();
                  return((PEGASUS_THREAD_RETURN)0);
               }
            }
            else
            {
               PEG_METHOD_EXIT();
               return((PEGASUS_THREAD_RETURN)0);
            }
         }
         catch(...)
         {
           Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
                "ThreadPool::_loop: Adding thread to idle pool failed.");
            PEG_METHOD_EXIT();
            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();
    myself->exit_self(0);  
      PEG_METHOD_EXIT();
    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");
   
      // Allocate_and_awaken will not run if the _dying flag is set.
      // Once the lock is acquired, ~ThreadPool will not change
      // the value of _dying until the lock is released.
   
      try
      {
         if (_dying.value())
         {
            Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
             "ThreadPool::allocate_and_awaken: ThreadPool is dying(1).");
            // ATTN: Error result has not yet been defined
            return true;
         }
    struct timeval start;    struct timeval start;
    gettimeofday(&start, NULL);    gettimeofday(&start, NULL);
         Thread *th = 0;
  
    Thread *th = _pool.remove_first();        th = _pool.remove_first();
  
    // wait for the right interval and try again        if (th == 0)
    while(th == 0 && _dying < 1)  
    {    {
            // will throw an IPCException&
       _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;  
       }       }
       pegasus_yield();  
       th = _pool.remove_first();  
    }    }
  
         if (th == 0)
    if(_dying < 1)  
    {    {
           // 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;
         }
   
       // initialize the thread data with the work function and parameters       // initialize the thread data with the work function and parameters
       th->remove_tsd("work func");        Tracer::trace(TRC_THREAD, Tracer::LEVEL4,
            "Initializing thread with work function and parameters: parm = %p",
             parm);
   
         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 *)),
                   (void *)work);                   (void *)work);
       th->remove_tsd("work parm");        th->delete_tsd("work parm");
       th->put_tsd("work parm", NULL, sizeof(void *), parm);       th->put_tsd("work parm", NULL, sizeof(void *), parm);
       th->remove_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);
  
Line 330 
Line 788 
       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();
          throw NullPointer();          throw NullPointer();
       }       }
         Tracer::trace(TRC_THREAD, Tracer::LEVEL4, "Signal thread to awaken");
       sleep_sem->signal();       sleep_sem->signal();
       th->dereference_tsd();       th->dereference_tsd();
    }    }
    else     catch (...)
       _pool.insert_first(th);     {
         Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
             "ThreadPool::allocate_and_awaken: Operation Failed.");
         PEG_METHOD_EXIT();
         // ATTN: Error result has not yet been defined
         return true;
      }
      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 346 
Line 816 
 Uint32 ThreadPool::kill_dead_threads(void) Uint32 ThreadPool::kill_dead_threads(void)
          throw(IPCException)          throw(IPCException)
 { {
      // Since the kill_dead_threads, ThreadPool or allocate_and_awaken
      // manipulate the threads on the ThreadPool queues, they should never
      // be allowed to run at the same time.
   
      PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::kill_dead_threads");
      // << 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);
   
      try
      {
         if (_dying.value())
         {
            return 0;
         }
   
    struct timeval now;    struct timeval now;
    gettimeofday(&now, NULL);    gettimeofday(&now, NULL);
    Uint32 bodies = 0;    Uint32 bodies = 0;
         AtomicInt needed(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
    {    {
            while(_dying.value() == 0 && _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)  
          throw NullPointer();              if(dead )
       if(dead->_handle.thid != 0)  
       {       {
          dead->detach();                 dead->join();
          destroy_thread(dead->_handle.thid, 0);                 delete dead;
          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;  
    }    }
         catch(...)
    DQueue<Thread> * map[2] =  
       {       {
          &_pool, &_running              Tracer::trace(TRC_THREAD, Tracer::LEVEL4, "Exception when deleting dead");
       };        }
  
         if (_dying.value())
         {
            return 0;
         }
  
    DQueue<Thread> *q = 0;        Thread *th = 0;
    int i = 0;        internal_dq idq;
    AtomicInt needed(0);  
  
    for( q = map[i] ; i < 2; i++, q = map[i])        if(_pool.count() > 0 )
    {  
       if(q->count() > 0 )  
       {       {
          try          try
          {          {
             q->try_lock();              _pool.try_lock();
          }          }
          catch(AlreadyLocked &)           catch(...)
          {          {
             q++;              return bodies;
             continue;  
          }          }
  
          struct timeval dt = { 0, 0 };          struct timeval dt = { 0, 0 };
          struct timeval *dtp;          struct timeval *dtp;
          Thread *th = 0;  
          th = q->next(th);           th = _pool.next(th);
          while (th != 0 )          while (th != 0 )
          {          {
             try             try
             {             {
                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);                 _pool.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;
               too_long = check_time(&dt, get_deallocate_wait(&deadlock_timeout));
   
               if( true == too_long)
             {             {
                // if we are deallocating from the pool, escape if we are                 // 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)  
                   {                   {
                      _current_threads++;                      _current_threads++;
                      th = q->next(th);                    th = _pool.next(th);
                      continue;                      continue;
                   }                   }
                   else  
                   {                 th = _pool.remove_no_lock((void *)th);
                      // we are killing a hung thread and we will drop below the                 idq.insert_first((void*)th);
                      // minimum. create another thread to make up for the one  
                      // we are about to kill  
                      needed++;  
                   }                   }
               th = _pool.next(th);
            }
            _pool.unlock();
                }                }
  
                th = q->remove_no_lock((void *)th);        th = (Thread*)idq.remove_last();
         while(th != 0)
                if(th != 0)  
                {                {
                   th->remove_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 *)),
                               (void *)&_undertaker);                               (void *)&_undertaker);
                   th->remove_tsd("work parm");           th->delete_tsd("work parm");
                   th->put_tsd("work parm", NULL, sizeof(void *), th);                   th->put_tsd("work parm", NULL, sizeof(void *), 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");
            PEGASUS_ASSERT(sleep_sem != 0);
  
                   if(sleep_sem == 0)  
                   {  
                      th->dereference_tsd();  
                      throw NullPointer();  
                   }  
                   // put the thread on the dead  list  
                   _dead.insert_first(th);  
                   bodies++;                   bodies++;
                   sleep_sem->signal();  
   
                   th->dereference_tsd();                   th->dereference_tsd();
                   th = 0;           sleep_sem->signal();
                }           th->join();  // Note: Clean up the thread here rather than
             }           delete th;   // leave it sitting unused on the _dead queue
             th = q->next(th);           th = (Thread*)idq.remove_last();
          }          }
          q->unlock();  
          while (needed.value() > 0)       Tracer::trace(TRC_THREAD, Tracer::LEVEL2,
          {                  "We need %u new threads", needed.value());
         while (needed.value() > 0)   {
             _link_pool(_init_thread());             _link_pool(_init_thread());
             needed--;             needed--;
            pegasus_sleep(0);
          }          }
          return bodies;
       }       }
       catch (...)
       {
    }    }
      PEG_METHOD_EXIT();
       return 0;
    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 503 
Line 985 
       return false;       return false;
 } }
  
   
 PEGASUS_THREAD_RETURN ThreadPool::_undertaker( void *parm ) PEGASUS_THREAD_RETURN ThreadPool::_undertaker( void *parm )
 { {
    Thread *myself = reinterpret_cast<Thread *>(parm);  
    if(myself != 0)     PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::_undertaker");
      exit_thread((PEGASUS_THREAD_RETURN)1);
      PEG_METHOD_EXIT();
      return (PEGASUS_THREAD_RETURN)1;
   }
   
   PEGASUS_THREAD_RETURN ThreadPool::_graveyard(Thread *t)
    {    {
       myself->detach();    PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::_graveyard");
       myself->_handle.thid = 0;    ThreadPool *pool = (ThreadPool *)t->get_parm();
       myself->cancel();    if(pool == 0 ) {
       myself->test_cancel();      Tracer::trace(TRC_THREAD, Tracer::LEVEL2,
       myself->exit_self(0);                    "Could not obtain the pool information from the Thread.", t);
   
         return (PEGASUS_THREAD_RETURN)1;
     }
     if (pool->_pool.exists(t))
       {
         if (pool->_pool.remove( (void *) t) != 0)
           {
           Tracer::trace(TRC_THREAD, Tracer::LEVEL4,
                   "Moving thread %p", t);
           /* We are moving the thread to the _running queue b/c
           _only_ kill_dead_threads has enough logic to take care
           of cleaning up the threads.*/
   
             pool->_running.insert_first( t );
           }
         else
           {
             Tracer::trace(TRC_THREAD, Tracer::LEVEL4,
                           "Could not move Thread %p from _pool to _runing queue.", t);
             return (PEGASUS_THREAD_RETURN)1;
           }
       }
   
     else if (pool->_running.exists(t))
       {
            Tracer::trace(TRC_THREAD, Tracer::LEVEL4,
                           "Thread %p is on _running queue. Letting kill_dead_threads take care of the problem.", t);
             return (PEGASUS_THREAD_RETURN)1;
       }
     if (!pool->_dead.exists(t))
       {
         Tracer::trace(TRC_THREAD, Tracer::LEVEL2,
                       "Thread is not on any queue! Moving it to the running queue.");
         pool->_running.insert_first( t );
       }
     PEG_METHOD_EXIT();
     return (PEGASUS_THREAD_RETURN)0;
   }
   
    void ThreadPool::_sleep_sem_del(void *p)
   {
      if(p != 0)
      {
         delete (Semaphore *)p;
      }
   }
   
    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)
   {
     PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::_init_thread");
      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
   
      if (!th->run())
      {
         delete th;
         return 0;
      }
      _current_threads++;
      pegasus_yield();
     PEG_METHOD_EXIT();
   
      return th;
   }
   
    void ThreadPool::_link_pool(Thread *th) throw(IPCException)
   {
      if(th == 0)
      {
         Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
             "ThreadPool::_link_pool: Thread pointer is null.");
         throw NullPointer();
      }
      try
      {
         _pool.insert_first(th);
      }
      catch(...)
      {
         Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
             "ThreadPool::_link_pool: _pool.insert_first failed.");
    }    }
    return((PEGASUS_THREAD_RETURN)0);  
 } }
  
  

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