(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.30 and 1.90.2.9

version 1.30, 2003/02/13 14:53:10 version 1.90.2.9, 2006/07/29 15:54:07
Line 1 
Line 1 
 //%/////////////////////////////////////////////////////////////////////////////  //%2006////////////////////////////////////////////////////////////////////////
 // //
 // Copyright (c) 2000, 2001, 2002 BMC Software, Hewlett-Packard Company, IBM,  // Copyright (c) 2000, 2001, 2002 BMC Software; Hewlett-Packard Development
 // The Open Group, Tivoli Systems  // 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.
   // Copyright (c) 2005 Hewlett-Packard Development Company, L.P.; IBM Corp.;
   // EMC Corporation; VERITAS Software Corporation; The Open Group.
   // Copyright (c) 2006 Hewlett-Packard Development Company, L.P.; IBM Corp.;
   // EMC Corporation; Symantec 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 33 
 // //
 // 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
   //              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 <Pegasus/Common/IPC.h>  #include <exception>
 #include <Pegasus/Common/Tracer.h> #include <Pegasus/Common/Tracer.h>
   #include "Time.h"
  
 #if defined(PEGASUS_OS_TYPE_WINDOWS)  PEGASUS_USING_STD;
 # include "ThreadWindows.cpp"  
 #elif defined(PEGASUS_OS_TYPE_UNIX)  
 # include "ThreadUnix.cpp"  
 #elif defined(PEGASUS_OS_TYPE_NSK)  
 # include "ThreadNsk.cpp"  
 #else  
 # error "Unsupported platform"  
 #endif  
  
 PEGASUS_NAMESPACE_BEGIN PEGASUS_NAMESPACE_BEGIN
  
 void thread_data::default_delete(void * data)  //==============================================================================
 {  //
    if( data != NULL)  // POSIX Threads Implementation:
       ::operator delete(data);  //
 }  //==============================================================================
  
 Boolean Thread::_signals_blocked = false;  #if defined(PEGASUS_HAVE_PTHREADS)
  
 // for non-native implementations  struct StartWrapperArg
 #ifndef PEGASUS_THREAD_CLEANUP_NATIVE  
 void Thread::cleanup_push( void (*routine)(void *), void *parm) throw(IPCException)  
 { {
     cleanup_handler *cu = new cleanup_handler(routine, parm);      void *(PEGASUS_THREAD_CDECL * start) (void *);
     try      void *arg;
   };
   
   extern "C" void *_start_wrapper(void *arg_)
     {     {
         _cleanup.insert_first(cu);      StartWrapperArg *arg = (StartWrapperArg *) arg_;
   
       void *return_value = (*arg->start) (arg->arg);
       delete arg;
   
       return return_value;
     }     }
     catch(IPCException&)  
   void Thread::cancel()
     {     {
         delete cu;      _cancelled = true;
         throw;      pthread_cancel(_handle.thid.tt_handle());
     }  
     return;  
 } }
  
 void Thread::cleanup_pop(Boolean execute) throw(IPCException)  void Thread::test_cancel()
 { {
     cleanup_handler *cu ;  #if defined(PEGASUS_PLATFORM_ZOS_ZSERIES_IBM)
     try      pthread_testintr();
   #else
       pthread_testcancel();
   #endif
   }
   
   Boolean Thread::is_cancelled(void)
     {     {
         cu = _cleanup.remove_first() ;      return _cancelled;
     }     }
     catch(IPCException&)  
   void Thread::thread_switch()
     {     {
         PEGASUS_ASSERT(0);  #if defined(PEGASUS_PLATFORM_ZOS_ZSERIES_IBM)
       pthread_yield(NULL);
   #else
       sched_yield();
   #endif
      }      }
     if(execute == true)  
         cu->execute();  /*
     delete cu;  ATTN: why are these missing on other platforms?
   */
   #if defined(PEGASUS_PLATFORM_LINUX_GENERIC_GNU)
   void Thread::suspend()
   {
       pthread_kill(_handle.thid.tt_handle(), SIGSTOP);
 } }
  
   void Thread::resume()
   {
       pthread_kill(_handle.thid.tt_handle(), SIGCONT);
   }
 #endif #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) throw(IPCException)  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.count() )  
    {    {
        try  #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(void)
        {        {
            cleanup_pop(true);      _is_detached = true;
       pthread_detach(_handle.thid.tt_handle());
        }        }
        catch(IPCException&)  
   ThreadStatus Thread::run()
        {        {
           PEGASUS_ASSERT(0);      StartWrapperArg *arg = new StartWrapperArg;
           break;      arg->start = _start;
       arg->arg = this;
   
       Threads::Type type = _is_detached ? Threads::DETACHED : 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 == 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;
    }    }
    _exit_code = exit_code;      return PEGASUS_THREAD_OK;
    exit_thread(exit_code);  
    _handle.thid = 0;  
 } }
  
   static sigset_t *block_signal_mask(sigset_t * sig)
   {
       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);
  
 #endif      sigaddset(sig, SIGABRT);
       sigaddset(sig, SIGALRM);
       sigaddset(sig, SIGPIPE);
  
 DQueue<ThreadPool> ThreadPool::_pools(true);  
  
   // 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;
   }
  
 void ThreadPool::kill_idle_threads(void)  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)
 { {
    static struct timeval now, last = {0, 0};      Threads::clear(_handle.thid);
   }
  
    pegasus_gettimeofday(&now);  Thread::~Thread()
    if(now.tv_sec - last.tv_sec > 5)  
    {  
       _pools.lock();  
       ThreadPool *p = _pools.next(0);  
       while(p != 0)  
       {       {
          try          try
          {          {
             p->kill_dead_threads();          join();
           empty_tsd();
          }          }
          catch(...)          catch(...)
          {          {
          }          // Do not allow the destructor to throw an exception
          p = _pools.next(p);  
       }  
       _pools.unlock();  
       pegasus_gettimeofday(&last);  
    }    }
 } }
  
   #endif /* PEGASUS_HAVE_PTHREADS */
  
 ThreadPool::ThreadPool(Sint16 initial_size,  //==============================================================================
                        const Sint8 *key,  //
                        Sint16 min,  // Windows Threads Implementation:
                        Sint16 max,  //
                        struct timeval & alloc_wait,  //==============================================================================
                        struct timeval & dealloc_wait,  
                        struct timeval & deadlock_detect)  
    : _max_threads(max), _min_threads(min),  
      _current_threads(0),  
      _pool(true), _running(true),  
      _dead(true), _dying(0)  
 {  
    _allocate_wait.tv_sec = alloc_wait.tv_sec;  
    _allocate_wait.tv_usec = alloc_wait.tv_usec;  
    _deallocate_wait.tv_sec = dealloc_wait.tv_sec;  
    _deallocate_wait.tv_usec = dealloc_wait.tv_usec;  
    _deadlock_detect.tv_sec = deadlock_detect.tv_sec;  
    _deadlock_detect.tv_usec = deadlock_detect.tv_usec;  
    memset(_key, 0x00, 17);  
    if(key != 0)  
       strncpy(_key, key, 16);  
    if(_max_threads > 0 && _max_threads < initial_size)  
       _max_threads = initial_size;  
    if(_min_threads > initial_size)  
       _min_threads = initial_size;  
  
    int i;  #if defined(PEGASUS_HAVE_WINDOWS_THREADS)
    for(i = 0; i < initial_size; i++)  
    {  
       _link_pool(_init_thread());  
    }  
    _pools.insert_last(this);  
  
 }  ThreadStatus Thread::run(void)
   {
       // 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;
  
 ThreadPool::~ThreadPool(void)      if (Threads::id(_handle.thid) == 0)
 { {
           if (errno == EAGAIN)
    _pools.remove(this);  
    _dying++;  
    Thread *th = 0;  
    th = _pool.remove_first();  
    while(th != 0)  
    {    {
       Semaphore *sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");              return PEGASUS_THREAD_INSUFFICIENT_RESOURCES;
           }
       if(sleep_sem == 0)          else
       {       {
          th->dereference_tsd();              return PEGASUS_THREAD_SETUP_FAILURE;
          throw NullPointer();  
       }       }
       }
       sleep_sem->signal();      return PEGASUS_THREAD_OK;
       sleep_sem->signal();  
       th->dereference_tsd();  
       // signal the thread's sleep semaphore  
       th->cancel();  
       th->join();  
       th->empty_tsd();  
       delete th;  
       th = _pool.remove_first();  
    }    }
  
    th = _running.remove_first();  void Thread::cancel(void)
    while(th != 0)  
    {    {
       // signal the thread's sleep semaphore      _cancelled = true;
       th->cancel();  
       th->join();  
       th->empty_tsd();  
       delete th;  
       th = _running.remove_first();  
    }    }
  
    th = _dead.remove_first();  void Thread::test_cancel(void)
    while(th != 0)  
    {    {
       // signal the thread's sleep semaphore      if (_cancel_enabled && _cancelled)
       th->cancel();      {
       th->join();          exit_self(0);
       th->empty_tsd();  
       delete th;  
       th = _dead.remove_first();  
    }    }
   
 } }
  
 // make this static to the class  Boolean Thread::is_cancelled(void)
 PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL ThreadPool::_loop(void *parm)  
 { {
    PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::_loop");      return _cancelled;
   }
  
    Thread *myself = (Thread *)parm;  void Thread::thread_switch(void)
    if(myself == 0)  
    {    {
       PEG_METHOD_EXIT();      Sleep(0);
       throw NullPointer();  
    }    }
    ThreadPool *pool = (ThreadPool *)myself->get_parm();  
    if(pool == 0 )  void Thread::sleep(Uint32 milliseconds)
    {    {
       PEG_METHOD_EXIT();      Sleep(milliseconds);
       throw NullPointer();  
    }    }
    Semaphore *sleep_sem = 0;  
    Semaphore *blocking_sem = 0;  
   
    struct timeval *deadlock_timer = 0;  
  
    try  void Thread::join(void)
    {    {
       sleep_sem = (Semaphore *)myself->reference_tsd("sleep sem");      if (Threads::id(_handle.thid) != 0)
       myself->dereference_tsd();  
       deadlock_timer = (struct timeval *)myself->reference_tsd("deadlock timer");  
       myself->dereference_tsd();  
    }  
    catch(IPCException &)  
    {    {
       PEG_METHOD_EXIT();          if (!_is_detached)
       myself->exit_self(0);  
    }  
    catch(...)  
    {    {
       PEG_METHOD_EXIT();              if (!_cancelled)
       myself->exit_self(0);  
    }  
   
    if(sleep_sem == 0 || deadlock_timer == 0)  
    {    {
       PEG_METHOD_EXIT();                  // Emulate the unix join api. Caller sleeps until thread is
       throw NullPointer();                  // done.
                   WaitForSingleObject(_handle.thid.handle, INFINITE);
    }    }
               else
    while(pool->_dying < 1)  
    {    {
       sleep_sem->wait();                  // Currently this is the only way to ensure this code does
                   // not
       // when we awaken we reside on the running queue, not the pool queue                  // hang forever.
       if(pool->_dying > 0)                  if (WaitForSingleObject(_handle.thid.handle, 10000) ==
          break;                      WAIT_TIMEOUT)
   
       PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *_work)(void *) = 0;  
       void *parm = 0;  
   
       try  
       {       {
          _work = (PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *)(void *)) \                      TerminateThread(_handle.thid.handle, 0);
             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(IPCException &)  
       {  
          PEG_METHOD_EXIT();  
          myself->exit_self(0);  
       }       }
  
       if(_work == 0)              DWORD exit_code = 0;
       {              GetExitCodeThread(_handle.thid.handle, &exit_code);
          PEG_METHOD_EXIT();              _exit_code = (ThreadReturnType) exit_code;
          throw NullPointer();  
       }       }
  
       if(_work ==          CloseHandle(_handle.thid.handle);
          (PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *)(void *)) &_undertaker)          Threads::clear(_handle.thid);
       }
   }
   
   void Thread::thread_init(void)
       {       {
          _work(parm);      _cancel_enabled = true;
       }       }
  
       gettimeofday(deadlock_timer, NULL);  void Thread::detach(void)
       try  
       {       {
          _work(parm);      _is_detached = true;
       }       }
       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)
       {       {
          gettimeofday(deadlock_timer, NULL);      Threads::clear(_handle.thid);
       }       }
       gettimeofday(deadlock_timer, NULL);  
       if( blocking_sem != 0 )  
          blocking_sem->signal();  
  
       // put myself back onto the available list  Thread::~Thread()
   {
       try       try
       {       {
          pool->_running.remove((void *)myself);          join();
          pool->_link_pool(myself);          empty_tsd();
       }       }
       catch(IPCException &)      catch(...)
       {       {
          PEG_METHOD_EXIT();  
          myself->exit_self(0);  
       }       }
    }    }
    // wait to be awakend by the thread pool destructor  
    sleep_sem->wait();  
    myself->test_cancel();  
  
    PEG_METHOD_EXIT();  #endif /* PEGASUS_HAVE_WINDOWS_THREADS */
    myself->exit_self(0);  
    return((PEGASUS_THREAD_RETURN)0);  
 }  
  
 void ThreadPool::allocate_and_awaken(void *parm,  //==============================================================================
                                      PEGASUS_THREAD_RETURN \  //
                                      (PEGASUS_THREAD_CDECL *work)(void *),  // Common implementation:
                                      Semaphore *blocking)  //
   //==============================================================================
  
    throw(IPCException)  void thread_data::default_delete(void *data)
 { {
    PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::allocate_and_awaken");      if (data != NULL)
    struct timeval start;          ::operator  delete(data);
    gettimeofday(&start, NULL);  }
   
    Thread *th = _pool.remove_first();  
  
    // wait for the right interval and try again  void language_delete(void *data)
    while(th == 0 && _dying < 1)  
    {    {
       _check_deadlock(&start) ;      if (data != NULL)
   
       if(_max_threads == 0 || _current_threads < _max_threads)  
       {       {
          th = _init_thread();          AutoPtr < AcceptLanguageList > al(static_cast <
          continue;                                            AcceptLanguageList * >(data));
       }       }
       pegasus_yield();  
       th = _pool.remove_first();  
    }    }
  
   Boolean Thread::_signals_blocked = false;
   #ifndef PEGASUS_OS_ZOS
   TSDKeyType Thread::_platform_thread_key = TSDKeyType(-1);
   #else
   TSDKeyType Thread::_platform_thread_key;
   #endif
   Boolean Thread::_key_initialized = false;
   Boolean Thread::_key_error = false;
  
    if(_dying < 1)  void Thread::cleanup_push(void (*routine) (void *), void *parm)
    {  
       // 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( 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  
       _running.insert_first(th);  
   
       // signal the thread's sleep semaphore to awaken it  
       Semaphore *sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");  
   
       if(sleep_sem == 0)  
       {       {
          th->dereference_tsd();      AutoPtr < cleanup_handler > cu(new cleanup_handler(routine, parm));
          PEG_METHOD_EXIT();      _cleanup.insert_front(cu.get());
          throw NullPointer();      cu.release();
       }      return;
       Tracer::trace(TRC_THREAD, Tracer::LEVEL4, "Signal thread to awaken");  
       sleep_sem->signal();  
       th->dereference_tsd();  
    }  
    else  
       _pool.insert_first(th);  
   
    PEG_METHOD_EXIT();  
 } }
  
 // caller is responsible for only calling this routine during slack periods  void Thread::cleanup_pop(Boolean execute)
 // but should call it at least once per _deadlock_detect with the running q  
 // and at least once per _deallocate_wait for the pool q  
   
 Uint32 ThreadPool::kill_dead_threads(void)  
          throw(IPCException)  
 { {
    struct timeval now;      AutoPtr < cleanup_handler > cu;
    gettimeofday(&now, NULL);      try
    Uint32 bodies = 0;  
   
    // first go thread the dead q and clean it up as much as possible  
    while(_dead.count() > 0)  
    {    {
       PEGASUS_STD(cout) << "ThreadPool:: removing and joining dead thread" << PEGASUS_STD(endl);          cu.reset(_cleanup.remove_front());
       Thread *dead = _dead.remove_first();  
       if(dead == 0)  
          throw NullPointer();  
       dead->join();  
       delete dead;  
    }    }
       catch(IPCException &)
    DQueue<Thread> * map[2] =  
       {       {
          &_pool, &_running          PEGASUS_ASSERT(0);
       };      }
       if (execute == true)
           cu->execute();
   }
  
  
    DQueue<Thread> *q = 0;  //thread_data *Thread::put_tsd(const Sint8 *key, void (*delete_func)(void *), Uint32 size, void *value)
    int i = 0;  
    AtomicInt needed(0);  
  
  
    for( ; i < 2; i++)  void Thread::exit_self(ThreadReturnType exit_code)
    {    {
       q = map[i];  #if defined(PEGASUS_PLATFORM_HPUX_ACC) || \
       if(q->count() > 0 )      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())
       {       {
          try          try
          {          {
             q->try_lock();              cleanup_pop(true);
          }          }
          catch(...)          catch(IPCException &)
          {          {
             return bodies;              PEGASUS_ASSERT(0);
               break;
          }          }
   
          struct timeval dt = { 0, 0 };  
          struct timeval *dtp;  
          Thread *th = 0;  
          th = q->next(th);  
          while (th != 0 )  
          {  
             try  
             {  
                dtp = (struct timeval *)th->try_reference_tsd("deadlock timer");  
             }             }
             catch(...)      _exit_code = exit_code;
             {      Threads::exit(exit_code);
                q->unlock();      Threads::clear(_handle.thid);
                return bodies;  #endif
             }             }
  
             if(dtp != 0)  Sint8 Thread::initializeKey()
             {             {
                memcpy(&dt, dtp, sizeof(struct timeval));      PEG_METHOD_ENTER(TRC_THREAD, "Thread::initializeKey");
             }      if (!Thread::_key_initialized)
             th->dereference_tsd();  
             struct timeval deadlock_timeout;  
             Boolean too_long;  
             if( i == 0)  
             {             {
                too_long = check_time(&dt, get_deallocate_wait(&deadlock_timeout));          if (Thread::_key_error)
             }  
             else  
             {             {
                too_long = check_time(&dt, get_deadlock_detect(&deadlock_timeout));              Tracer::trace(TRC_THREAD, Tracer::LEVEL4,
                             "Thread: ERROR - thread key error");
               return -1;
             }             }
  
             if( true == too_long)          if (TSDKey::create(&Thread::_platform_thread_key) == 0)
             {  
                // if we are deallocating from the pool, escape if we are  
                // down to the minimum thread count  
                _current_threads--;  
                if( _current_threads.value() < (Uint32)_min_threads )  
                {  
                   if( i == 0)  
                   {                   {
                      _current_threads++;              Tracer::trace(TRC_THREAD, Tracer::LEVEL4,
                      th = q->next(th);                            "Thread: able to create a thread key");
                      continue;              Thread::_key_initialized = true;
                   }                   }
                   else                   else
                   {                   {
                      // we are killing a hung thread and we will drop below the              Tracer::trace(TRC_THREAD, Tracer::LEVEL4,
                      // minimum. create another thread to make up for the one                            "Thread: ERROR - unable to create a thread key");
                      // we are about to kill              Thread::_key_error = true;
                      needed++;              return -1;
                   }                   }
                }                }
  
                th = q->remove_no_lock((void *)th);      PEG_METHOD_EXIT();
       return 0;
   }
  
                if(th != 0)  Thread *Thread::getCurrent()
                {  
                   if( i == 0 )  
                   {                   {
                      th->delete_tsd("work func");      PEG_METHOD_ENTER(TRC_THREAD, "Thread::getCurrent");
                      th->put_tsd("work func", NULL,      if (Thread::initializeKey() != 0)
                                  sizeof( PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *)(void *)),  
                                  (void *)&_undertaker);  
                      th->delete_tsd("work parm");  
                      th->put_tsd("work parm", NULL, sizeof(void *), th);  
   
                      // signal the thread's sleep semaphore to awaken it  
                      Semaphore *sleep_sem = (Semaphore *)th->reference_tsd("sleep sem");  
   
                      if(sleep_sem == 0)  
                      {                      {
                         q->unlock();          return NULL;
                         th->dereference_tsd();  
                         throw NullPointer();  
                      }                      }
       PEG_METHOD_EXIT();
                      bodies++;      return (Thread *) TSDKey::get_thread_specific(_platform_thread_key);
                      th->dereference_tsd();  
                      _dead.insert_first(th);  
                      sleep_sem->signal();  
                      th = 0;  
                   }                   }
                   else  
   void Thread::setCurrent(Thread * thrd)
                   {                   {
                      // deadlocked threads      PEG_METHOD_ENTER(TRC_THREAD, "Thread::setCurrent");
                      PEGASUS_STD(cout) << "Killing a deadlocked thread" << PEGASUS_STD(endl);      if (Thread::initializeKey() == 0)
                      th->cancel();      {
                      delete th;          if (TSDKey::
                   }              set_thread_specific(Thread::_platform_thread_key,
                }                                  (void *) thrd) == 0)
             }  
             th = q->next(th);  
             pegasus_sleep(1);  
          }  
          q->unlock();  
          while (needed.value() > 0)  
          {          {
             _link_pool(_init_thread());              Tracer::trace(TRC_THREAD, Tracer::LEVEL4,
             needed--;                  "Successful set Thread * into thread specific storage");
             pegasus_sleep(0);  
          }          }
           else
           {
               Tracer::trace(TRC_THREAD, Tracer::LEVEL4,
                   "ERROR: error setting Thread * into thread specific storage");
       }       }
    }    }
     return bodies;      PEG_METHOD_EXIT();
 } }
  
   AcceptLanguageList *Thread::getLanguages()
 Boolean ThreadPool::check_time(struct timeval *start, struct timeval *interval)  
 { {
    // never time out if the interval is zero      PEG_METHOD_ENTER(TRC_THREAD, "Thread::getLanguages");
    if(interval && interval->tv_sec == 0 && interval->tv_usec == 0)  
       return false;  
   
    struct timeval now, finish, remaining;  
    Uint32 usec;  
    gettimeofday(&now, NULL);  
   
    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;  
  
    if ( timeval_subtract(&remaining, &finish, &now) )      Thread *curThrd = Thread::getCurrent();
       return true;      if (curThrd == NULL)
    else          return NULL;
       return false;      AcceptLanguageList *acceptLangs =
 }          (AcceptLanguageList *) curThrd->reference_tsd("acceptLanguages");
       curThrd->dereference_tsd();
 PEGASUS_THREAD_RETURN ThreadPool::_undertaker( void *parm )      PEG_METHOD_EXIT();
 {      return acceptLangs;
    exit_thread((PEGASUS_THREAD_RETURN)1);  
    return (PEGASUS_THREAD_RETURN)1;  
 } }
  
   void Thread::setLanguages(AcceptLanguageList * langs)   // l10n
  void ThreadPool::_sleep_sem_del(void *p)  
 {  
    if(p != 0)  
    {    {
       delete (Semaphore *)p;      PEG_METHOD_ENTER(TRC_THREAD, "Thread::setLanguages");
    }  
 }  
  
  void ThreadPool::_check_deadlock(struct timeval *start) throw(Deadlock)      Thread *currentThrd = Thread::getCurrent();
       if (currentThrd != NULL)
 { {
    if (true == check_time(start, &_deadlock_detect))          // deletes the old tsd and creates a new one
       throw Deadlock(pegasus_thread_self());          currentThrd->put_tsd("acceptLanguages",
    return;                               language_delete,
                                sizeof (AcceptLanguageList *), langs);
 } }
  
       PEG_METHOD_EXIT();
  Boolean ThreadPool::_check_deadlock_no_throw(struct timeval *start)  
 {  
    return(check_time(start, &_deadlock_detect));  
 } }
  
  Boolean ThreadPool::_check_dealloc(struct timeval *start)  void Thread::clearLanguages()   // l10n
 { {
    return(check_time(start, &_deallocate_wait));      PEG_METHOD_ENTER(TRC_THREAD, "Thread::clearLanguages");
 }  
  
  Thread *ThreadPool::_init_thread(void) throw(IPCException)      Thread *currentThrd = Thread::getCurrent();
       if (currentThrd != NULL)
 { {
    Thread *th = (Thread *) new Thread(_loop, this, false);          // deletes the old tsd
    // allocate a sleep semaphore and pass it in the thread context          currentThrd->delete_tsd("acceptLanguages");
    // 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));  
    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  
    th->run();  
    _current_threads++;  
    pegasus_yield();  
   
    return th;  
 } }
  
  void ThreadPool::_link_pool(Thread *th) throw(IPCException)      PEG_METHOD_EXIT();
 {  
    if(th == 0)  
       throw NullPointer();  
    _pool.insert_first(th);  
 } }
  
   // ATTN: not sure where to put this!
   #ifdef PEGASUS_ZOS_SECURITY
   bool isEnhancedSecurity = 99;
   #endif
  
 PEGASUS_NAMESPACE_END PEGASUS_NAMESPACE_END
   


Legend:
Removed from v.1.30  
changed lines
  Added in v.1.90.2.9

No CVS admin address has been configured
Powered by
ViewCVS 0.9.2