(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.90 and 1.91

version 1.90, 2006/07/11 18:39:28 version 1.91, 2006/08/09 21:12:42
Line 43 
Line 43 
  
 #include "Thread.h" #include "Thread.h"
 #include <exception> #include <exception>
 #include <Pegasus/Common/IPC.h>  
 #include <Pegasus/Common/Tracer.h> #include <Pegasus/Common/Tracer.h>
   #include "Time.h"
 #if defined(PEGASUS_OS_TYPE_WINDOWS)  
 # include "ThreadWindows.cpp"  
 #elif defined(PEGASUS_OS_TYPE_UNIX)  
 # include "ThreadUnix.cpp"  
 #elif defined(PEGASUS_OS_TYPE_NSK)  
 # include "ThreadNsk.cpp"  
 #elif defined(PEGASUS_OS_VMS)  
 # include "ThreadVms.cpp"  
 #else  
 # error "Unsupported platform"  
 #endif  
  
 PEGASUS_USING_STD; PEGASUS_USING_STD;
   
 PEGASUS_NAMESPACE_BEGIN PEGASUS_NAMESPACE_BEGIN
  
   //==============================================================================
   //
   // POSIX Threads Implementation:
   //
   //==============================================================================
  
 void thread_data::default_delete(void * data)  #if defined(PEGASUS_HAVE_PTHREADS)
 {  
    if( data != NULL)  
       ::operator delete(data);  
 }  
  
 // l10n start  struct StartWrapperArg
 void language_delete(void * data)  
 { {
    if( data != NULL)      void *(PEGASUS_THREAD_CDECL * start) (void *);
       void *arg;
   };
   
   extern "C" void *_start_wrapper(void *arg_)
    {    {
       AutoPtr<AcceptLanguageList> al(static_cast<AcceptLanguageList *>(data));      StartWrapperArg *arg = (StartWrapperArg *) arg_;
   
       void *return_value = (*arg->start) (arg->arg);
       delete arg;
   
       return return_value;
    }    }
   
   void Thread::cancel()
   {
       _cancelled = true;
       pthread_cancel(_handle.thid.tt_handle());
 } }
 // l10n end  
  
 Boolean Thread::_signals_blocked = false;  void Thread::test_cancel()
 // l10n  {
 #ifndef PEGASUS_OS_ZOS  #if defined(PEGASUS_PLATFORM_ZOS_ZSERIES_IBM)
 PEGASUS_THREAD_KEY_TYPE Thread::_platform_thread_key = PEGASUS_THREAD_KEY_TYPE(-1);      pthread_testintr();
 #else #else
 PEGASUS_THREAD_KEY_TYPE Thread::_platform_thread_key;      pthread_testcancel();
 #endif #endif
 Boolean Thread::_key_initialized = false;  }
 Boolean Thread::_key_error = false;  
   
  
 void Thread::cleanup_push( void (*routine)(void *), void *parm)  Boolean Thread::is_cancelled(void)
 { {
     AutoPtr<cleanup_handler> cu(new cleanup_handler(routine, parm));      return _cancelled;
     _cleanup.insert_front(cu.get());  
     cu.release();  
     return;  
 } }
  
 void Thread::cleanup_pop(Boolean execute)  void Thread::thread_switch()
 {  
     AutoPtr<cleanup_handler> cu;  
     try  
     {     {
         cu.reset(_cleanup.remove_front());  #if defined(PEGASUS_PLATFORM_ZOS_ZSERIES_IBM)
       pthread_yield(NULL);
   #else
       sched_yield();
   #endif
     }     }
     catch(IPCException&)  
   /*
   ATTN: why are these missing on other platforms?
   */
   #if defined(PEGASUS_PLATFORM_LINUX_GENERIC_GNU)
   void Thread::suspend()
     {     {
         PEGASUS_ASSERT(0);      pthread_kill(_handle.thid.tt_handle(), SIGSTOP);
      }      }
     if(execute == true)  
         cu->execute();  void Thread::resume()
   {
       pthread_kill(_handle.thid.tt_handle(), SIGCONT);
 } }
   #endif
  
   void Thread::sleep(Uint32 msec)
   {
       Threads::sleep(msec);
   }
  
 //thread_data *Thread::put_tsd(const Sint8 *key, void (*delete_func)(void *), Uint32 size, void *value)  void Thread::join(void)
   {
       if (!_is_detached && Threads::id(_handle.thid) != 0)
           pthread_join(_handle.thid.tt_handle(), &_exit_code);
  
       Threads::clear(_handle.thid);
   }
  
 #ifndef PEGASUS_THREAD_EXIT_NATIVE  void Thread::thread_init(void)
 void Thread::exit_self(PEGASUS_THREAD_RETURN exit_code)  
 {  
     // execute the cleanup stack and then return  
    while( _cleanup.size() )  
    {  
        try  
        {        {
            cleanup_pop(true);  #if defined(PEGASUS_PLATFORM_ZOS_ZSERIES_IBM)
       pthread_setintr(PTHREAD_INTR_ENABLE);
       pthread_setintrtype(PTHREAD_INTR_ASYNCHRONOUS);
   #else
       pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
       pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL);
   #endif
       _cancel_enabled = true;
        }        }
        catch(IPCException&)  
   void Thread::detach(void)
        {        {
           PEGASUS_ASSERT(0);      _is_detached = true;
           break;  #if defined(PEGASUS_PLATFORM_ZOS_ZSERIES_IBM)
        }      pthread_t  thread_id=_handle.thid.tt_handle();
    }      pthread_detach(&thread_id);
    _exit_code = exit_code;  #else
    exit_thread(exit_code);      pthread_detach(_handle.thid.tt_handle());
    _handle.thid = 0;  #endif
 } }
  
   ThreadStatus Thread::run()
   {
       StartWrapperArg *arg = new StartWrapperArg;
       arg->start = _start;
       arg->arg = this;
  
 #endif      Threads::Type type = _is_detached ? Threads::DETACHED : Threads::JOINABLE;
       int rc = Threads::create(_handle.thid, type, _start_wrapper, arg);
  
 // l10n start      // On Linux distributions released prior 2005, the implementation of
 Sint8 Thread::initializeKey()      // Native POSIX Thread Library returns ENOMEM instead of EAGAIN when
 {      // there
     PEG_METHOD_ENTER(TRC_THREAD, "Thread::initializeKey");      // are no insufficient memory.  Hence we are checking for both.  See bug
     if (!Thread::_key_initialized)      // 386.
     {  
         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)      if ((rc == EAGAIN) || (rc == ENOMEM))
         {         {
             Tracer::trace(TRC_THREAD, Tracer::LEVEL4,          Threads::clear(_handle.thid);
                 "Thread: able to create a thread key");          delete arg;
             Thread::_key_initialized = true;          return PEGASUS_THREAD_INSUFFICIENT_RESOURCES;
         }         }
         else      else if (rc != 0)
         {         {
             Tracer::trace(TRC_THREAD, Tracer::LEVEL4,          Threads::clear(_handle.thid);
                 "Thread: ERROR - unable to create a thread key");          delete arg;
             Thread::_key_error = true;          return PEGASUS_THREAD_SETUP_FAILURE;
             return -1;  
         }         }
       return PEGASUS_THREAD_OK;
     }     }
  
     PEG_METHOD_EXIT();  static sigset_t *block_signal_mask(sigset_t * sig)
     return 0;  {
       sigemptyset(sig);
       // should not be used for main()
       sigaddset(sig, SIGHUP);
       sigaddset(sig, SIGINT);
       // maybe useless, since KILL can't be blocked according to POSIX
       sigaddset(sig, SIGKILL);
   
       sigaddset(sig, SIGABRT);
       sigaddset(sig, SIGALRM);
       sigaddset(sig, SIGPIPE);
   
   
   // Note: older versions of the linux pthreads library use SIGUSR1 and SIGUSR2
   // internally to stop and start threads that are blocking, the newer ones
   // implement this through the kernel's real time signals
   // since SIGSTOP/CONT can handle suspend()/resume() on Linux
   // block them
   // #if defined(PEGASUS_PLATFORM_LINUX_IX86_GNU)
   //     sigaddset(sig, SIGUSR1);
   //     sigaddset(sig, SIGUSR2);
   // #endif
   #ifndef PEGASUS_PLATFORM_ZOS_ZSERIES_IBM
       pthread_sigmask(SIG_BLOCK, sig, NULL);
   #else
       sigprocmask(SIG_BLOCK, sig, NULL);
   #endif
       return sig;
 } }
  
 Thread * Thread::getCurrent()  Thread::Thread(ThreadReturnType(PEGASUS_THREAD_CDECL * start) (void *), void *parameter, Boolean detached):_is_detached(detached),
 {  _cancel_enabled(true),
     PEG_METHOD_ENTER(TRC_THREAD, "Thread::getCurrent");  _cancelled(false),
     if (Thread::initializeKey() != 0)  _start(start), _cleanup(), _tsd(), _thread_parm(parameter), _exit_code(0)
     {     {
         return NULL;      Threads::clear(_handle.thid);
     }  
     PEG_METHOD_EXIT();  
     return (Thread *)pegasus_get_thread_specific(_platform_thread_key);  
 } }
  
 void Thread::setCurrent(Thread * thrd)  Thread::~Thread()
 {  
     PEG_METHOD_ENTER(TRC_THREAD, "Thread::setCurrent");  
     if (Thread::initializeKey() == 0)  
     {     {
         if (pegasus_set_thread_specific(      try
                Thread::_platform_thread_key, (void *) thrd) == 0)  
         {         {
             Tracer::trace(TRC_THREAD, Tracer::LEVEL4,          join();
                 "Successful set Thread * into thread specific storage");          empty_tsd();
         }         }
         else      catch(...)
         {         {
             Tracer::trace(TRC_THREAD, Tracer::LEVEL4,          // Do not allow the destructor to throw an exception
                 "ERROR: error setting Thread * into thread specific storage");  
         }  
     }     }
     PEG_METHOD_EXIT();  
 } }
  
 AcceptLanguageList * Thread::getLanguages()  #endif /* PEGASUS_HAVE_PTHREADS */
 {  
     PEG_METHOD_ENTER(TRC_THREAD, "Thread::getLanguages");  
  
     Thread * curThrd = Thread::getCurrent();  //==============================================================================
     if (curThrd == NULL)  //
         return NULL;  // Windows Threads Implementation:
     AcceptLanguageList * acceptLangs =  //
         (AcceptLanguageList *)curThrd->reference_tsd("acceptLanguages");  //==============================================================================
     curThrd->dereference_tsd();  
     PEG_METHOD_EXIT();  
     return acceptLangs;  
 }  
  
 void Thread::setLanguages(AcceptLanguageList *langs) //l10n  #if defined(PEGASUS_HAVE_WINDOWS_THREADS)
 {  
     PEG_METHOD_ENTER(TRC_THREAD, "Thread::setLanguages");  
  
     Thread* currentThrd = Thread::getCurrent();  ThreadStatus Thread::run(void)
     if (currentThrd != NULL)  
     {     {
         // deletes the old tsd and creates a new one      // Note: A Win32 thread ID is not the same thing as a pthread ID.
         currentThrd->put_tsd("acceptLanguages",      // Win32 threads have both a thread ID and a handle.  The handle
             language_delete,      // is used in the wait functions, etc.
             sizeof(AcceptLanguageList *),      // So _handle.thid is actually the thread handle.
             langs);  
     }  
  
     PEG_METHOD_EXIT();      unsigned threadid = 0;
 }  
  
 void Thread::clearLanguages() //l10n      ThreadType tt;
 {      tt.handle = (HANDLE) _beginthreadex(NULL, 0, _start, this, 0, &threadid);
     PEG_METHOD_ENTER(TRC_THREAD, "Thread::clearLanguages");      _handle.thid = tt;
  
     Thread * currentThrd = Thread::getCurrent();      if (Threads::id(_handle.thid) == 0)
     if (currentThrd != NULL)  
     {     {
         // deletes the old tsd          if (errno == EAGAIN)
         currentThrd->delete_tsd("acceptLanguages");          {
               return PEGASUS_THREAD_INSUFFICIENT_RESOURCES;
     }     }
           else
     PEG_METHOD_EXIT();          {
               return PEGASUS_THREAD_SETUP_FAILURE;
           }
       }
       return PEGASUS_THREAD_OK;
 } }
 // l10n end  
   
   
 ///////////////////////////////////////////////////////////////////////////////  
 //  
 // ThreadPool  
 //  
 ///////////////////////////////////////////////////////////////////////////////  
  
 ThreadPool::ThreadPool(  void Thread::cancel(void)
     Sint16 initialSize,  
     const char* key,  
     Sint16 minThreads,  
     Sint16 maxThreads,  
     struct timeval& deallocateWait)  
     : _maxThreads(maxThreads),  
       _minThreads(minThreads),  
       _currentThreads(0),  
       _idleThreads(),  
       _runningThreads(),  
       _dying(0)  
 { {
     _deallocateWait.tv_sec = deallocateWait.tv_sec;      _cancelled = true;
     _deallocateWait.tv_usec = deallocateWait.tv_usec;  }
  
     memset(_key, 0x00, 17);  void Thread::test_cancel(void)
     if (key != 0)  {
       if (_cancel_enabled && _cancelled)
     {     {
         strncpy(_key, key, 16);          exit_self(0);
       }
     }     }
  
     if ((_maxThreads > 0) && (_maxThreads < initialSize))  Boolean Thread::is_cancelled(void)
     {     {
         _maxThreads = initialSize;      return _cancelled;
     }     }
  
     if (_minThreads > initialSize)  void Thread::thread_switch(void)
     {     {
         _minThreads = initialSize;      Sleep(0);
     }     }
  
     for (int i = 0; i < initialSize; i++)  void Thread::sleep(Uint32 milliseconds)
     {     {
         _addToIdleThreadsQueue(_initializeThread());      Sleep(milliseconds);
     }  
 } }
  
 ThreadPool::~ThreadPool()  void Thread::join(void)
 { {
     PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::~ThreadPool");      if (Threads::id(_handle.thid) != 0)
   
     try  
     {     {
         // Set the dying flag so all thread know the destructor has been entered          if (!_is_detached)
         _dying++;  
        Tracer::trace(TRC_THREAD, Tracer::LEVEL2,  
                 "Cleaning up %d idle threads. ", _currentThreads.get());  
         while (_currentThreads.get() > 0)  
         {         {
             Thread* thread = _idleThreads.remove_front();              if (!_cancelled)
             if (thread != 0)  
             {             {
                 _cleanupThread(thread);                  // Emulate the unix join api. Caller sleeps until thread is
                 _currentThreads--;                  // done.
                   WaitForSingleObject(_handle.thid.handle, INFINITE);
             }             }
             else             else
             {             {
                 pegasus_yield();                  // Currently this is the only way to ensure this code does
                   // not
                   // hang forever.
                   if (WaitForSingleObject(_handle.thid.handle, 10000) ==
                       WAIT_TIMEOUT)
                   {
                       TerminateThread(_handle.thid.handle, 0);
             }             }
         }         }
   
               DWORD exit_code = 0;
               GetExitCodeThread(_handle.thid.handle, &exit_code);
               _exit_code = (ThreadReturnType) exit_code;
     }     }
     catch (...)  
     {          CloseHandle(_handle.thid.handle);
           Threads::clear(_handle.thid);
     }     }
 } }
  
 PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL ThreadPool::_loop(void* parm)  void Thread::thread_init(void)
 { {
     PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::_loop");      _cancel_enabled = true;
   }
     try  
     {  
         Thread* myself = (Thread *)parm;  
         PEGASUS_ASSERT(myself != 0);  
   
         // Set myself into thread specific storage  
         // This will allow code to get its own Thread  
         Thread::setCurrent(myself);  
   
         ThreadPool* pool = (ThreadPool *)myself->get_parm();  
         PEGASUS_ASSERT(pool != 0);  
   
         Semaphore* sleep_sem = 0;  
         struct timeval* lastActivityTime = 0;  
  
         try  void Thread::detach(void)
         {         {
             sleep_sem = (Semaphore *)myself->reference_tsd("sleep sem");      _is_detached = true;
             myself->dereference_tsd();  
             PEGASUS_ASSERT(sleep_sem != 0);  
   
             lastActivityTime =  
                 (struct timeval *)myself->reference_tsd("last activity time");  
             myself->dereference_tsd();  
             PEGASUS_ASSERT(lastActivityTime != 0);  
         }         }
         catch (...)  
   Thread::Thread(ThreadReturnType(PEGASUS_THREAD_CDECL * start) (void *),
                  void *parameter,
                  Boolean detached):_is_detached(detached),
   _cancel_enabled(true),
   _cancelled(false),
   _start(start), _cleanup(), _tsd(), _thread_parm(parameter), _exit_code(0)
         {         {
             Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,      Threads::clear(_handle.thid);
                 "ThreadPool::_loop: Failure getting sleep_sem or "  
                     "lastActivityTime.");  
             PEGASUS_ASSERT(false);  
             pool->_idleThreads.remove(myself);  
             pool->_currentThreads--;  
             PEG_METHOD_EXIT();  
             return((PEGASUS_THREAD_RETURN)1);  
         }         }
  
         while (1)  Thread::~Thread()
         {         {
             try             try
             {             {
                 sleep_sem->wait();          join();
           empty_tsd();
             }             }
             catch (...)             catch (...)
             {             {
                 Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,      }
                     "ThreadPool::_loop: failure on sleep_sem->wait().");  
                 PEGASUS_ASSERT(false);  
                 pool->_idleThreads.remove(myself);  
                 pool->_currentThreads--;  
                 PEG_METHOD_EXIT();  
                 return((PEGASUS_THREAD_RETURN)1);  
             }             }
  
             // When we awaken we reside on the _runningThreads queue, not the  #endif /* PEGASUS_HAVE_WINDOWS_THREADS */
             // _idleThreads queue.  
  
             PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL* work)(void *) = 0;  //==============================================================================
             void* parm = 0;  //
             Semaphore* blocking_sem = 0;  // Common implementation:
   //
   //==============================================================================
  
             try  void thread_data::default_delete(void *data)
             {  
                 work = (PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *)(void *))  
                     myself->reference_tsd("work func");  
                 myself->dereference_tsd();  
                 parm = myself->reference_tsd("work parm");  
                 myself->dereference_tsd();  
                 blocking_sem = (Semaphore *)myself->reference_tsd("blocking sem");  
                 myself->dereference_tsd();  
             }  
             catch (...)  
             {             {
                 Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,      if (data != NULL)
                     "ThreadPool::_loop: Failure accessing work func, work parm, "          ::operator  delete(data);
                         "or blocking sem.");  
                 PEGASUS_ASSERT(false);  
                 pool->_idleThreads.remove(myself);  
                 pool->_currentThreads--;  
                 PEG_METHOD_EXIT();  
                 return((PEGASUS_THREAD_RETURN)1);  
             }             }
  
             if (work == 0)  void language_delete(void *data)
             {             {
                 Tracer::trace(TRC_THREAD, Tracer::LEVEL4,      if (data != NULL)
                     "ThreadPool::_loop: work func is 0, meaning we should exit.");  
                 break;  
             }  
   
             gettimeofday(lastActivityTime, NULL);  
   
             try  
             {             {
                 PEG_TRACE_STRING(TRC_THREAD, Tracer::LEVEL4, "Work starting.");          AutoPtr < AcceptLanguageList > al(static_cast <
                 work(parm);                                            AcceptLanguageList * >(data));
                 PEG_TRACE_STRING(TRC_THREAD, Tracer::LEVEL4, "Work finished.");  
             }  
             catch (Exception & e)  
             {  
                 PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,  
                     String("Exception from work in ThreadPool::_loop: ") +  
                         e.getMessage());  
             }  
 #if !defined(PEGASUS_OS_LSB)  
             catch (const exception& e)  
             {  
                 PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,  
                     String("Exception from work in ThreadPool::_loop: ") +  
                         e.what());  
             }             }
 #endif  
             catch (...)  
             {  
                 PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,  
                     "Unknown exception from work in ThreadPool::_loop.");  
             }             }
  
             // put myself back onto the available list  Boolean Thread::_signals_blocked = false;
             try  #ifndef PEGASUS_OS_ZOS
             {  TSDKeyType Thread::_platform_thread_key = TSDKeyType(-1);
                 gettimeofday(lastActivityTime, NULL);  #else
                 if (blocking_sem != 0)  TSDKeyType Thread::_platform_thread_key;
   #endif
   Boolean Thread::_key_initialized = false;
   Boolean Thread::_key_error = false;
   
   void Thread::cleanup_push(void (*routine) (void *), void *parm)
                 {                 {
                     blocking_sem->signal();      AutoPtr < cleanup_handler > cu(new cleanup_handler(routine, parm));
       _cleanup.insert_front(cu.get());
       cu.release();
       return;
                 }                 }
  
                 pool->_runningThreads.remove(myself);  void Thread::cleanup_pop(Boolean execute)
                 pool->_idleThreads.insert_front(myself);  
             }  
             catch (...)  
             {             {
                 Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,      AutoPtr < cleanup_handler > cu;
                     "ThreadPool::_loop: Adding thread to idle pool failed.");      try
                 PEGASUS_ASSERT(false);  
                 pool->_currentThreads--;  
                 PEG_METHOD_EXIT();  
                 return((PEGASUS_THREAD_RETURN)1);  
             }  
         }  
     }  
     catch (const Exception& e)  
     {     {
         PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,          cu.reset(_cleanup.remove_front());
             "Caught exception: \"" + e.getMessage() + "\".  Exiting _loop.");  
     }     }
     catch (...)      catch(IPCException &)
     {     {
         PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,          PEGASUS_ASSERT(0);
             "Caught unrecognized exception.  Exiting _loop.");  
     }     }
       if (execute == true)
     PEG_METHOD_EXIT();          cu->execute();
     return((PEGASUS_THREAD_RETURN)0);  
 } }
  
 ThreadStatus ThreadPool::allocate_and_awaken(  
     void* parm,  
     PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL* work)(void *),  
     Semaphore* blocking)  
 {  
     PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::allocate_and_awaken");  
  
     // Allocate_and_awaken will not run if the _dying flag is set.  //thread_data *Thread::put_tsd(const Sint8 *key, void (*delete_func)(void *), Uint32 size, void *value)
     // Once the lock is acquired, ~ThreadPool will not change  
     // the value of _dying until the lock is released.  
  
     try  void Thread::exit_self(ThreadReturnType exit_code)
     {     {
         if (_dying.get())  #if defined(PEGASUS_PLATFORM_HPUX_ACC) || \
       defined(PEGASUS_PLATFORM_LINUX_GENERIC_GNU)
       // NOTE: pthread_exit exhibits unusual behavior on RHEL 3 U2, as
       // documented in Bugzilla 3836.  Where feasible, it may be advantageous
       // to avoid using this function.
       pthread_exit(exit_code);
   #else
       // execute the cleanup stack and then return
       while (_cleanup.size())
         {         {
             Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,          try
                 "ThreadPool::allocate_and_awaken: ThreadPool is dying(1).");  
             return PEGASUS_THREAD_UNAVAILABLE;  
         }  
         struct timeval start;  
         gettimeofday(&start, NULL);  
         Thread* th = 0;  
   
         th = _idleThreads.remove_front();  
   
         if (th == 0)  
         {         {
             if ((_maxThreads == 0) ||              cleanup_pop(true);
                 (_currentThreads.get() < Uint32(_maxThreads)))          }
           catch(IPCException &)
             {             {
                 th = _initializeThread();              PEGASUS_ASSERT(0);
               break;
             }             }
         }         }
       _exit_code = exit_code;
       Threads::exit(exit_code);
       Threads::clear(_handle.thid);
   #endif
   }
  
         if (th == 0)  Sint8 Thread::initializeKey()
         {         {
             // ATTN-DME-P3-20031103: This trace message should not be      PEG_METHOD_ENTER(TRC_THREAD, "Thread::initializeKey");
             // be labeled TRC_DISCARDED_DATA, because it does not      if (!Thread::_key_initialized)
             // necessarily imply that a failure has occurred.  However,      {
             // this label is being used temporarily to help isolate          if (Thread::_key_error)
             // the cause of client timeout problems.          {
             Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,              Tracer::trace(TRC_THREAD, Tracer::LEVEL4,
                 "ThreadPool::allocate_and_awaken: Insufficient resources: "                            "Thread: ERROR - thread key error");
                     " pool = %s, running threads = %d, idle threads = %d",              return -1;
                 _key, _runningThreads.size(), _idleThreads.size());  
             return PEGASUS_THREAD_INSUFFICIENT_RESOURCES;  
         }         }
  
         // initialize the thread data with the work function and parameters          if (TSDKey::create(&Thread::_platform_thread_key) == 0)
           {
         Tracer::trace(TRC_THREAD, Tracer::LEVEL4,         Tracer::trace(TRC_THREAD, Tracer::LEVEL4,
             "Initializing thread with work function and parameters: parm = %p",                            "Thread: able to create a thread key");
             parm);              Thread::_key_initialized = true;
   
         th->delete_tsd("work func");  
         th->put_tsd("work func", NULL,  
             sizeof( PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *)(void *)),  
             (void *)work);  
         th->delete_tsd("work parm");  
         th->put_tsd("work parm", NULL, sizeof(void *), parm);  
         th->delete_tsd("blocking sem");  
         if (blocking != 0)  
             th->put_tsd("blocking sem", NULL, sizeof(Semaphore *), blocking);  
   
         // put the thread on the running list  
         _runningThreads.insert_front(th);  
   
         // signal the thread's sleep semaphore to awaken it  
         Semaphore* sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");  
         PEGASUS_ASSERT(sleep_sem != 0);  
   
         Tracer::trace(TRC_THREAD, Tracer::LEVEL4, "Signal thread to awaken");  
         sleep_sem->signal();  
         th->dereference_tsd();  
     }     }
     catch (...)          else
     {     {
         Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,              Tracer::trace(TRC_THREAD, Tracer::LEVEL4,
             "ThreadPool::allocate_and_awaken: Operation Failed.");                            "Thread: ERROR - unable to create a thread key");
         PEG_METHOD_EXIT();              Thread::_key_error = true;
         // ATTN: Error result has not yet been defined              return -1;
         return PEGASUS_THREAD_SETUP_FAILURE;  
     }     }
     PEG_METHOD_EXIT();  
     return PEGASUS_THREAD_OK;  
 } }
  
 // caller is responsible for only calling this routine during slack periods      PEG_METHOD_EXIT();
 // but should call it at least once per _deallocateWait interval.      return 0;
   }
 Uint32 ThreadPool::cleanupIdleThreads()  
 {  
     PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::cleanupIdleThreads");  
   
     Uint32 numThreadsCleanedUp = 0;  
  
     Uint32 numIdleThreads = _idleThreads.size();  Thread *Thread::getCurrent()
     for (Uint32 i = 0; i < numIdleThreads; i++)  
     {     {
         // Do not dip below the minimum thread count      PEG_METHOD_ENTER(TRC_THREAD, "Thread::getCurrent");
         if (_currentThreads.get() <= (Uint32)_minThreads)      if (Thread::initializeKey() != 0)
         {         {
             break;          return NULL;
         }         }
       PEG_METHOD_EXIT();
         Thread* thread = _idleThreads.remove_back();      return (Thread *) TSDKey::get_thread_specific(_platform_thread_key);
   
         // If there are no more threads in the _idleThreads queue, we're done.  
         if (thread == 0)  
         {  
             break;  
         }         }
  
         struct timeval* lastActivityTime;  void Thread::setCurrent(Thread * thrd)
         try  
         {         {
             lastActivityTime = (struct timeval *)thread->try_reference_tsd(      PEG_METHOD_ENTER(TRC_THREAD, "Thread::setCurrent");
                 "last activity time");      if (Thread::initializeKey() == 0)
             PEGASUS_ASSERT(lastActivityTime != 0);  
         }  
         catch (...)  
         {         {
             PEGASUS_ASSERT(false);          if (TSDKey::
             _idleThreads.insert_back(thread);              set_thread_specific(Thread::_platform_thread_key,
             break;                                  (void *) thrd) == 0)
         }          {
               Tracer::trace(TRC_THREAD, Tracer::LEVEL4,
         Boolean cleanupThisThread =                  "Successful set Thread * into thread specific storage");
             _timeIntervalExpired(lastActivityTime, &_deallocateWait);  
         thread->dereference_tsd();  
   
         if (cleanupThisThread)  
         {  
             _cleanupThread(thread);  
             _currentThreads--;  
             numThreadsCleanedUp++;  
         }         }
         else         else
         {         {
             _idleThreads.insert_front(thread);              Tracer::trace(TRC_THREAD, Tracer::LEVEL4,
                   "ERROR: error setting Thread * into thread specific storage");
         }         }
     }     }
   
     PEG_METHOD_EXIT();     PEG_METHOD_EXIT();
     return numThreadsCleanedUp;  
 } }
  
 void ThreadPool::_cleanupThread(Thread* thread)  AcceptLanguageList *Thread::getLanguages()
 { {
     PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::cleanupThread");      PEG_METHOD_ENTER(TRC_THREAD, "Thread::getLanguages");
   
     // Set the "work func" and "work parm" to 0 so _loop() knows to exit.  
     thread->delete_tsd("work func");  
     thread->put_tsd(  
         "work func", 0,  
         sizeof(PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *)(void *)),  
         (void *) 0);  
     thread->delete_tsd("work parm");  
     thread->put_tsd("work parm", 0, sizeof(void *), 0);  
   
     // signal the thread's sleep semaphore to awaken it  
     Semaphore* sleep_sem = (Semaphore *)thread->reference_tsd("sleep sem");  
     PEGASUS_ASSERT(sleep_sem != 0);  
     sleep_sem->signal();  
     thread->dereference_tsd();  
   
     thread->join();  
     delete thread;  
  
       Thread *curThrd = Thread::getCurrent();
       if (curThrd == NULL)
           return NULL;
       AcceptLanguageList *acceptLangs =
           (AcceptLanguageList *) curThrd->reference_tsd("acceptLanguages");
       curThrd->dereference_tsd();
     PEG_METHOD_EXIT();     PEG_METHOD_EXIT();
       return acceptLangs;
 } }
  
 Boolean ThreadPool::_timeIntervalExpired(  void Thread::setLanguages(AcceptLanguageList * langs)   // l10n
     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;      PEG_METHOD_ENTER(TRC_THREAD, "Thread::setLanguages");
     }  
   
     struct timeval now, finish, remaining;  
     Uint32 usec;  
     pegasus_gettimeofday(&now);  
     pegasus_gettimeofday(&remaining);    // Avoid valgrind error  
   
     finish.tv_sec = start->tv_sec + interval->tv_sec;  
     usec = start->tv_usec + interval->tv_usec;  
     finish.tv_sec += (usec / 1000000);  
     usec %= 1000000;  
     finish.tv_usec = usec;  
  
     return (timeval_subtract(&remaining, &finish, &now) != 0);      Thread *currentThrd = Thread::getCurrent();
       if (currentThrd != NULL)
       {
           // deletes the old tsd and creates a new one
           currentThrd->put_tsd("acceptLanguages",
                                language_delete,
                                sizeof (AcceptLanguageList *), langs);
 } }
  
 void ThreadPool::_deleteSemaphore(void *p)      PEG_METHOD_EXIT();
 {  
     delete (Semaphore *)p;  
 } }
  
 Thread* ThreadPool::_initializeThread()  void Thread::clearLanguages()   // l10n
 { {
     PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::_initializeThread");      PEG_METHOD_ENTER(TRC_THREAD, "Thread::clearLanguages");
   
     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", &_deleteSemaphore, sizeof(Semaphore), (void *)sleep_sem);  
   
     struct timeval* lastActivityTime =  
         (struct timeval *) ::operator new(sizeof(struct timeval));  
     pegasus_gettimeofday(lastActivityTime);  
   
     th->put_tsd("last activity time", thread_data::default_delete,  
         sizeof(struct timeval), (void *)lastActivityTime);  
     // thread will enter _loop() and sleep on sleep_sem until we signal it  
  
     if (th->run() != PEGASUS_THREAD_OK)      Thread *currentThrd = Thread::getCurrent();
       if (currentThrd != NULL)
     {     {
                 Tracer::trace(TRC_THREAD, Tracer::LEVEL2,          // deletes the old tsd
                         "Could not create thread. Error code is %d.", errno);          currentThrd->delete_tsd("acceptLanguages");
         delete th;  
         return 0;  
     }     }
     _currentThreads++;  
     pegasus_yield();  
  
     PEG_METHOD_EXIT();     PEG_METHOD_EXIT();
     return th;  
 }  
   
 void ThreadPool::_addToIdleThreadsQueue(Thread* th)  
 {  
     if (th == 0)  
     {  
         Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,  
             "ThreadPool::_addToIdleThreadsQueue: Thread pointer is null.");  
         throw NullPointer();  
     }     }
  
     try  // ATTN: not sure where to put this!
     {  #ifdef PEGASUS_ZOS_SECURITY
         _idleThreads.insert_front(th);  bool isEnhancedSecurity = 99;
     }  #endif
     catch (...)  
     {  
         Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,  
             "ThreadPool::_addToIdleThreadsQueue: _idleThreads.insert_front "  
                 "failed.");  
     }  
 }  
  
 PEGASUS_NAMESPACE_END PEGASUS_NAMESPACE_END

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