(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.2.7 and 1.100.2.6

version 1.90.2.7, 2006/07/29 01:08:49 version 1.100.2.6, 2008/10/23 18:59:32
Line 29 
Line 29 
 // //
 //============================================================================== //==============================================================================
 // //
 // Author: Mike Day (mdday@us.ibm.com)  
 //  
 // Modified By: Rudy Schuet (rudy.schuet@compaq.com) 11/12/01  
 //              added nsk platform support  
 //              Roger Kumpf, Hewlett-Packard Company (roger_kumpf@hp.com)  
 //              Amit K Arora, IBM (amita@in.ibm.com) for PEP#101  
 //              Sean Keenan, Hewlett-Packard Company (sean.keenan@hp.com)  
 //              David Dillard, VERITAS Software Corp.  
 //                  (david.dillard@veritas.com)  
 //  
 //%///////////////////////////////////////////////////////////////////////////// //%/////////////////////////////////////////////////////////////////////////////
  
 #include "Thread.h" #include "Thread.h"
   #include <errno.h>
 #include <exception> #include <exception>
 #include <Pegasus/Common/Tracer.h> #include <Pegasus/Common/Tracer.h>
 #include "Time.h" #include "Time.h"
Line 58 
Line 49 
  
 #if defined(PEGASUS_HAVE_PTHREADS) #if defined(PEGASUS_HAVE_PTHREADS)
  
   struct StartWrapperArg
   {
       void *(PEGASUS_THREAD_CDECL * start) (void *);
       void *arg;
   };
   
 extern "C" void *_start_wrapper(void *arg_) extern "C" void *_start_wrapper(void *arg_)
 { {
     StartWrapperArg *arg = (StartWrapperArg *) arg_;      // Clean up dynamic memory now to prevent a leak if the thread is canceled.
       StartWrapperArg arg;
       arg.start = ((StartWrapperArg *) arg_)->start;
       arg.arg = ((StartWrapperArg *) arg_)->arg;
       delete (StartWrapperArg *) arg_;
   
       // establish cancelability of the thread
       pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
       pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL);
  
     void *return_value = (*arg->start) (arg->arg);      void *return_value = (*arg.start) (arg.arg);
     delete arg;  
  
     return return_value;     return return_value;
 } }
  
   void Thread::cancel()
   {
       _cancelled = true;
       pthread_cancel(_handle.thid.thread);
   }
   
   void Thread::test_cancel()
   {
   #if defined(PEGASUS_PLATFORM_ZOS_ZSERIES_IBM)
       pthread_testintr();
   #else
       pthread_testcancel();
   #endif
   }
   
   Boolean Thread::is_cancelled()
   {
       return _cancelled;
   }
   
   void Thread::thread_switch()
   {
   #if defined(PEGASUS_PLATFORM_ZOS_ZSERIES_IBM)
       pthread_yield(NULL);
   #else
       sched_yield();
   #endif
   }
   
   /*
   ATTN: why are these missing on other platforms?
   */
   #if defined(PEGASUS_PLATFORM_LINUX_GENERIC_GNU)
   void Thread::suspend()
   {
       pthread_kill(_handle.thid.thread, SIGSTOP);
   }
   
   void Thread::resume()
   {
       pthread_kill(_handle.thid.thread, SIGCONT);
   }
   #endif
   
   void Thread::sleep(Uint32 msec)
   {
       Threads::sleep(msec);
   }
   
   void Thread::join()
   {
       if (!_is_detached && !Threads::null(_handle.thid))
           pthread_join(_handle.thid.thread, &_exit_code);
   
       Threads::clear(_handle.thid);
   }
   
   void Thread::thread_init()
   {
   #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;
   }
   
   void Thread::detach()
   {
       _is_detached = true;
   #if defined(PEGASUS_PLATFORM_ZOS_ZSERIES_IBM)
       pthread_t  thread_id=_handle.thid.thread;
       pthread_detach(&thread_id);
   #else
       pthread_detach(_handle.thid.thread);
   #endif
   }
   
   ThreadStatus Thread::run()
   {
       StartWrapperArg *arg = new StartWrapperArg;
       arg->start = _start;
       arg->arg = this;
   
       Threads::Type type =
           _is_detached ? Threads::THREADS_DETACHED : Threads::THREADS_JOINABLE;
       int rc = Threads::create(_handle.thid, type, _start_wrapper, arg);
   
       // On Linux distributions released prior 2005, the implementation of
       // Native POSIX Thread Library returns ENOMEM instead of EAGAIN when
       // there
       // are no insufficient memory.  Hence we are checking for both.  See bug
       // 386.
   
       if (rc == -1)
           rc = errno;
       if ((rc == EAGAIN) || (rc == ENOMEM))
       {
           Threads::clear(_handle.thid);
           delete arg;
           return PEGASUS_THREAD_INSUFFICIENT_RESOURCES;
       }
       else if (rc != 0)
       {
           Threads::clear(_handle.thid);
           delete arg;
           return PEGASUS_THREAD_SETUP_FAILURE;
       }
       return PEGASUS_THREAD_OK;
   }
   
 static sigset_t *block_signal_mask(sigset_t * sig) static sigset_t *block_signal_mask(sigset_t * sig)
 { {
     sigemptyset(sig);     sigemptyset(sig);
Line 91 
Line 208 
 //     sigaddset(sig, SIGUSR1); //     sigaddset(sig, SIGUSR1);
 //     sigaddset(sig, SIGUSR2); //     sigaddset(sig, SIGUSR2);
 // #endif // #endif
 #ifndef PEGASUS_PLATFORM_ZOS_ZSERIES_IBM  #if defined (PEGASUS_PLATFORM_ZOS_ZSERIES_IBM) || defined (PEGASUS_OS_VMS)
     pthread_sigmask(SIG_BLOCK, sig, NULL);  
 #else  
     sigprocmask(SIG_BLOCK, sig, NULL);     sigprocmask(SIG_BLOCK, sig, NULL);
   #else
       pthread_sigmask(SIG_BLOCK, sig, NULL);
 #endif #endif
     return sig;     return sig;
 } }
  
 Thread::Thread(ThreadReturnType(PEGASUS_THREAD_CDECL * start) (void *), void *parameter, Boolean detached):_is_detached(detached),  /*
   ATTN: remove this!
   */
   #if 1
   static Uint32 _num_threads = 0;
   static Mutex _num_threads_mutex;
   #endif
   
   Thread::Thread(
       ThreadReturnType(PEGASUS_THREAD_CDECL* start) (void*),
       void* parameter,
       Boolean detached)
       : _is_detached(detached),
 _cancel_enabled(true), _cancel_enabled(true),
 _cancelled(false), _cancelled(false),
 _start(start), _cleanup(), _tsd(), _thread_parm(parameter), _exit_code(0)        _start(start),
 {        _cleanup(),
         _tsd(),
         _thread_parm(parameter),
         _exit_code(0)
   {
   /*
   ATTN: remove this!
   */
   #if 1
       Uint32 num_threads;
       _num_threads_mutex.lock();
       _num_threads++;
       num_threads = _num_threads;
       printf("Thread::Thread(): num_threads=%u\n", num_threads);
       _num_threads_mutex.unlock();
   #endif
   
     Threads::clear(_handle.thid);     Threads::clear(_handle.thid);
 } }
  
 Thread::~Thread() Thread::~Thread()
 { {
   /*
   ATTN: remove this!
   */
   #if 1
       Uint32 num_threads;
       _num_threads_mutex.lock();
       _num_threads--;
       num_threads = _num_threads;
       printf("Thread::~Thread(): num_threads=%u\n", num_threads);
       _num_threads_mutex.unlock();
   #endif
   
     try     try
     {     {
         join();         join();
Line 130 
Line 287 
  
 #if defined(PEGASUS_HAVE_WINDOWS_THREADS) #if defined(PEGASUS_HAVE_WINDOWS_THREADS)
  
   ThreadStatus Thread::run()
   {
       // Note: A Win32 thread ID is not the same thing as a pthread ID.
       // Win32 threads have both a thread ID and a handle.  The handle
       // is used in the wait functions, etc.
       // So _handle.thid is actually the thread handle.
   
       unsigned threadid = 0;
   
       ThreadType tt;
       tt.handle = (HANDLE) _beginthreadex(NULL, 0, _start, this, 0, &threadid);
       _handle.thid = tt;
   
       if (Threads::null(_handle.thid))
       {
           if (errno == EAGAIN)
           {
               return PEGASUS_THREAD_INSUFFICIENT_RESOURCES;
           }
           else
           {
               return PEGASUS_THREAD_SETUP_FAILURE;
           }
       }
       return PEGASUS_THREAD_OK;
   }
   
   void Thread::cancel()
   {
       _cancelled = true;
   }
   
   void Thread::test_cancel()
   {
       if (_cancel_enabled && _cancelled)
       {
           exit_self(0);
       }
   }
   
   Boolean Thread::is_cancelled()
   {
       return _cancelled;
   }
   
   void Thread::thread_switch()
   {
       Sleep(0);
   }
   
   void Thread::sleep(Uint32 milliseconds)
   {
       Sleep(milliseconds);
   }
   
   void Thread::join()
   {
       if (!Threads::null(_handle.thid))
       {
           if (!_is_detached)
           {
               if (!_cancelled)
               {
                   // Emulate the unix join api. Caller sleeps until thread is
                   // done.
                   WaitForSingleObject(_handle.thid.handle, INFINITE);
               }
               else
               {
                   // 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;
           }
   
           CloseHandle(_handle.thid.handle);
           Threads::clear(_handle.thid);
       }
   }
   
   void Thread::thread_init()
   {
       _cancel_enabled = true;
   }
   
   void Thread::detach()
   {
       _is_detached = true;
   }
   
 Thread::Thread(ThreadReturnType(PEGASUS_THREAD_CDECL * start) (void *), Thread::Thread(ThreadReturnType(PEGASUS_THREAD_CDECL * start) (void *),
                void *parameter,                void *parameter,
                Boolean detached):_is_detached(detached),                Boolean detached):_is_detached(detached),
Line 208 
Line 464 
 } }
  
  
 //thread_data *Thread::put_tsd(const Sint8 *key, void (*delete_func)(void *), Uint32 size, void *value)  
   
   
 void Thread::exit_self(ThreadReturnType exit_code) void Thread::exit_self(ThreadReturnType exit_code)
 { {
 #if defined(PEGASUS_PLATFORM_HPUX_ACC) || \  #if !defined(PEGASUS_PLATFORM_AIX_RS_IBMCXX) \
     defined(PEGASUS_PLATFORM_LINUX_GENERIC_GNU)      && !defined(PEGASUS_PLATFORM_PASE_ISERIES_IBMCXX)
     // NOTE: pthread_exit exhibits unusual behavior on RHEL 3 U2, as      Threads::exit(exit_code);
     // documented in Bugzilla 3836.  Where feasible, it may be advantageous  
     // to avoid using this function.  
     pthread_exit(exit_code);  
 #else #else
     // execute the cleanup stack and then return     // execute the cleanup stack and then return
     while (_cleanup.size())     while (_cleanup.size())
Line 246 
Line 496 
     {     {
         if (Thread::_key_error)         if (Thread::_key_error)
         {         {
             Tracer::trace(TRC_THREAD, Tracer::LEVEL4,              PEG_TRACE_CSTRING(TRC_THREAD, Tracer::LEVEL4,
                           "Thread: ERROR - thread key error");                           "Thread: ERROR - thread key error");
             return -1;             return -1;
         }         }
  
         if (TSDKey::create(&Thread::_platform_thread_key) == 0)         if (TSDKey::create(&Thread::_platform_thread_key) == 0)
         {         {
             Tracer::trace(TRC_THREAD, Tracer::LEVEL4,              PEG_TRACE_CSTRING(TRC_THREAD, Tracer::LEVEL4,
                           "Thread: able to create a thread key");                           "Thread: able to create a thread key");
             Thread::_key_initialized = true;             Thread::_key_initialized = true;
         }         }
         else         else
         {         {
             Tracer::trace(TRC_THREAD, Tracer::LEVEL4,              PEG_TRACE_CSTRING(TRC_THREAD, Tracer::LEVEL4,
                           "Thread: ERROR - unable to create a thread key");                           "Thread: ERROR - unable to create a thread key");
             Thread::_key_error = true;             Thread::_key_error = true;
             return -1;             return -1;
Line 290 
Line 540 
             set_thread_specific(Thread::_platform_thread_key,             set_thread_specific(Thread::_platform_thread_key,
                                 (void *) thrd) == 0)                                 (void *) thrd) == 0)
         {         {
             Tracer::trace(TRC_THREAD, Tracer::LEVEL4,              PEG_TRACE_CSTRING(TRC_THREAD, Tracer::LEVEL4,
                 "Successful set Thread * into thread specific storage");                 "Successful set Thread * into thread specific storage");
         }         }
         else         else
         {         {
             Tracer::trace(TRC_THREAD, Tracer::LEVEL4,              PEG_TRACE_CSTRING(TRC_THREAD, Tracer::LEVEL4,
                 "ERROR: error setting Thread * into thread specific storage");                 "ERROR: error setting Thread * into thread specific storage");
         }         }
     }     }
Line 316 
Line 566 
     return acceptLangs;     return acceptLangs;
 } }
  
 void Thread::setLanguages(AcceptLanguageList * langs)   // l10n  void Thread::setLanguages(AcceptLanguageList * langs)
 { {
     PEG_METHOD_ENTER(TRC_THREAD, "Thread::setLanguages");     PEG_METHOD_ENTER(TRC_THREAD, "Thread::setLanguages");
  
Line 332 
Line 582 
     PEG_METHOD_EXIT();     PEG_METHOD_EXIT();
 } }
  
 void Thread::clearLanguages()   // l10n  void Thread::clearLanguages()
 { {
     PEG_METHOD_ENTER(TRC_THREAD, "Thread::clearLanguages");     PEG_METHOD_ENTER(TRC_THREAD, "Thread::clearLanguages");
  
Line 346 
Line 596 
     PEG_METHOD_EXIT();     PEG_METHOD_EXIT();
 } }
  
 ///////////////////////////////////////////////////////////////////////////////  
 //  
 // ThreadPool  
 //  
 ///////////////////////////////////////////////////////////////////////////////  
   
 ThreadPool::ThreadPool(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;  
     _deallocateWait.tv_usec = deallocateWait.tv_usec;  
   
     memset(_key, 0x00, 17);  
     if (key != 0)  
     {  
         strncpy(_key, key, 16);  
     }  
   
     if ((_maxThreads > 0) && (_maxThreads < initialSize))  
     {  
         _maxThreads = initialSize;  
     }  
   
     if (_minThreads > initialSize)  
     {  
         _minThreads = initialSize;  
     }  
   
     for (int i = 0; i < initialSize; i++)  
     {  
         _addToIdleThreadsQueue(_initializeThread());  
     }  
 }  
   
 ThreadPool::~ThreadPool()  
 {  
     PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::~ThreadPool");  
   
     try  
     {  
         // Set the dying flag so all thread know the destructor has been  
         // entered  
         _dying++;  
         Tracer::trace(TRC_THREAD, Tracer::LEVEL2,  
                       "Cleaning up %d idle threads. ", _currentThreads.get());  
   
         while (_currentThreads.get() > 0)  
         {  
             Thread *thread = _idleThreads.remove_front();  
             if (thread != 0)  
             {  
                 _cleanupThread(thread);  
                 _currentThreads--;  
             }  
             else  
             {  
                 Threads::yield();  
             }  
         }  
     }  
     catch(...)  
     {  
     }  
 }  
   
 ThreadReturnType PEGASUS_THREAD_CDECL ThreadPool::_loop(void *parm)  
 {  
     PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::_loop");  
   
     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  
         {  
             sleep_sem = (Semaphore *) myself->reference_tsd("sleep sem");  
             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(...)  
         {  
             Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,  
                           "ThreadPool::_loop: Failure getting sleep_sem or "  
                           "lastActivityTime.");  
             PEGASUS_ASSERT(false);  
             pool->_idleThreads.remove(myself);  
             pool->_currentThreads--;  
             PEG_METHOD_EXIT();  
             return ((ThreadReturnType) 1);  
         }  
   
         while (1)  
         {  
             try  
             {  
                 sleep_sem->wait();  
             }  
             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 ((ThreadReturnType) 1);  
             }  
   
             // When we awaken we reside on the _runningThreads queue, not the  
             // _idleThreads queue.  
   
             ThreadReturnType(PEGASUS_THREAD_CDECL * work) (void *) = 0;  
             void *parm = 0;  
             Semaphore *blocking_sem = 0;  
   
             try  
             {  
                 work = (ThreadReturnType(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,  
                               "ThreadPool::_loop: Failure accessing work func, work parm, "  
                               "or blocking sem.");  
                 PEGASUS_ASSERT(false);  
                 pool->_idleThreads.remove(myself);  
                 pool->_currentThreads--;  
                 PEG_METHOD_EXIT();  
                 return ((ThreadReturnType) 1);  
             }  
   
             if (work == 0)  
             {  
                 Tracer::trace(TRC_THREAD, Tracer::LEVEL4,  
                               "ThreadPool::_loop: work func is 0, meaning we should exit.");  
                 break;  
             }  
   
             Time::gettimeofday(lastActivityTime);  
   
             try  
             {  
                 PEG_TRACE_STRING(TRC_THREAD, Tracer::LEVEL4,  
                                  "Work starting.");  
                 work(parm);  
                 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  
             try  
             {  
                 Time::gettimeofday(lastActivityTime);  
                 if (blocking_sem != 0)  
                 {  
                     blocking_sem->signal();  
                 }  
   
                 pool->_runningThreads.remove(myself);  
                 pool->_idleThreads.insert_front(myself);  
             }  
             catch(...)  
             {  
                 Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,  
                               "ThreadPool::_loop: Adding thread to idle pool failed.");  
                 PEGASUS_ASSERT(false);  
                 pool->_currentThreads--;  
                 PEG_METHOD_EXIT();  
                 return ((ThreadReturnType) 1);  
             }  
         }  
     }  
     catch(const Exception & e)  
     {  
         PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,  
                          "Caught exception: \"" + e.getMessage() +  
                          "\".  Exiting _loop.");  
     }  
     catch(...)  
     {  
         PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,  
                          "Caught unrecognized exception.  Exiting _loop.");  
     }  
   
     PEG_METHOD_EXIT();  
     return ((ThreadReturnType) 0);  
 }  
   
 ThreadStatus ThreadPool::allocate_and_awaken(void *parm,  
                                              ThreadReturnType  
                                              (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.  
     // Once the lock is acquired, ~ThreadPool will not change  
     // the value of _dying until the lock is released.  
   
     try  
     {  
         if (_dying.get())  
         {  
             Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,  
                           "ThreadPool::allocate_and_awaken: ThreadPool is dying(1).");  
             return PEGASUS_THREAD_UNAVAILABLE;  
         }  
         struct timeval start;  
         Time::gettimeofday(&start);  
         Thread *th = 0;  
   
         th = _idleThreads.remove_front();  
   
         if (th == 0)  
         {  
             if ((_maxThreads == 0) ||  
                 (_currentThreads.get() < Uint32(_maxThreads)))  
             {  
                 th = _initializeThread();  
             }  
         }  
   
         if (th == 0)  
         {  
             // 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",  
                           _key, _runningThreads.size(), _idleThreads.size());  
             return PEGASUS_THREAD_INSUFFICIENT_RESOURCES;  
         }  
   
         // initialize the thread data with the work function and parameters  
         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,  
                     sizeof (ThreadReturnType(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(...)  
     {  
         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 PEGASUS_THREAD_SETUP_FAILURE;  
     }  
     PEG_METHOD_EXIT();  
     return PEGASUS_THREAD_OK;  
 }  
   
 // caller is responsible for only calling this routine during slack periods  
 // but should call it at least once per _deallocateWait interval.  
   
 Uint32 ThreadPool::cleanupIdleThreads()  
 {  
     PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::cleanupIdleThreads");  
   
     Uint32 numThreadsCleanedUp = 0;  
   
     Uint32 numIdleThreads = _idleThreads.size();  
     for (Uint32 i = 0; i < numIdleThreads; i++)  
     {  
         // Do not dip below the minimum thread count  
         if (_currentThreads.get() <= (Uint32) _minThreads)  
         {  
             break;  
         }  
   
         Thread *thread = _idleThreads.remove_back();  
   
         // If there are no more threads in the _idleThreads queue, we're  
         // done.  
         if (thread == 0)  
         {  
             break;  
         }  
   
         struct timeval *lastActivityTime;  
         try  
         {  
             lastActivityTime =  
                 (struct timeval *) thread->  
                 try_reference_tsd("last activity time");  
             PEGASUS_ASSERT(lastActivityTime != 0);  
         }  
         catch(...)  
         {  
             PEGASUS_ASSERT(false);  
             _idleThreads.insert_back(thread);  
             break;  
         }  
   
         Boolean cleanupThisThread =  
             _timeIntervalExpired(lastActivityTime, &_deallocateWait);  
         thread->dereference_tsd();  
   
         if (cleanupThisThread)  
         {  
             _cleanupThread(thread);  
             _currentThreads--;  
             numThreadsCleanedUp++;  
         }  
         else  
         {  
             _idleThreads.insert_front(thread);  
         }  
     }  
   
     PEG_METHOD_EXIT();  
     return numThreadsCleanedUp;  
 }  
   
 void ThreadPool::_cleanupThread(Thread * thread)  
 {  
     PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::cleanupThread");  
   
     // 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 (ThreadReturnType(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;  
   
     PEG_METHOD_EXIT();  
 }  
   
 Boolean ThreadPool::_timeIntervalExpired(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;  
     Uint32 usec;  
     Time::gettimeofday(&now);  
     Time::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 (Time::subtract(&remaining, &finish, &now) != 0);  
 }  
   
 void ThreadPool::_deleteSemaphore(void *p)  
 {  
     delete(Semaphore *) p;  
 }  
   
 Thread *ThreadPool::_initializeThread()  
 {  
     PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::_initializeThread");  
   
     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));  
     Time::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)  
     {  
         Tracer::trace(TRC_THREAD, Tracer::LEVEL2,  
                       "Could not create thread. Error code is %d.", errno);  
         delete th;  
         return 0;  
     }  
     _currentThreads++;  
     Threads::yield();  
   
     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  
     {  
         _idleThreads.insert_front(th);  
     }  
     catch(...)  
     {  
         Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,  
                       "ThreadPool::_addToIdleThreadsQueue: _idleThreads.insert_front "  
                       "failed.");  
     }  
 }  
   
 // ATTN: not sure where to put this! // ATTN: not sure where to put this!
 #ifdef PEGASUS_ZOS_SECURITY #ifdef PEGASUS_ZOS_SECURITY
 bool isEnhancedSecurity = 99; bool isEnhancedSecurity = 99;

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