pegasus/src/Pegasus/Common/Thread.cpp - diff

Return to Thread.cpp CVS log

Up to [Pegasus] / pegasus / src / Pegasus / Common

Diff for /pegasus/src/Pegasus/Common/Thread.cpp between version 1.87.4.1 and 1.105

version 1.87.4.1, 2006/01/18 17:37:56

version 1.105, 2008/03/20 01:06:41

Line 1

//%2005////////////////////////////////////////////////////////////////////////

//%2006////////////////////////////////////////////////////////////////////////

// Company, L.P.; IBM Corp.; The Open Group; Tivoli Systems.

Line 8

// IBM Corp.; EMC Corporation; VERITAS Software Corporation; The Open Group.

// EMC Corporation; VERITAS Software Corporation; The Open Group.

// EMC Corporation; Symantec Corporation; The Open Group.

// Permission is hereby granted, free of charge, to any person obtaining a copy

// of this software and associated documentation files (the "Software"), to

Line 27

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 <errno.h>

#include <exception>

#include <Pegasus/Common/IPC.h>

#include <Pegasus/Common/Tracer.h>

#include <Pegasus/Common/AutoPtr.h>

#if defined(PEGASUS_OS_TYPE_WINDOWS)

#include "Time.h"

# 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_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));

// Clean up dynamic memory now to prevent a leak if the thread is canceled.

}

StartWrapperArg arg;

}

arg.start = ((StartWrapperArg *) arg_)->start;

// l10n end

arg.arg = ((StartWrapperArg *) arg_)->arg;

delete (StartWrapperArg *) arg_;

Boolean Thread::_signals_blocked = false;

// establish cancelability of the thread

// l10n

pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);

#ifndef PEGASUS_OS_ZOS

pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL);

PEGASUS_THREAD_KEY_TYPE Thread::_platform_thread_key = PEGASUS_THREAD_KEY_TYPE(-1);

#else

PEGASUS_THREAD_KEY_TYPE Thread::_platform_thread_key;

#endif

Boolean Thread::_key_initialized = false;

Boolean Thread::_key_error = false;

void *return_value = (*arg.start) (arg.arg);

void Thread::cleanup_push( void (*routine)(void *), void *parm)

return return_value;

{

AutoPtr<cleanup_handler> cu(new cleanup_handler(routine, parm));

_cleanup.insert_first(cu.get());

cu.release();

return;

}

void Thread::cleanup_pop(Boolean execute)

void Thread::cancel()

{

AutoPtr<cleanup_handler> cu;

pthread_cancel(_handle.thid.thread);

try

{

cu.reset(_cleanup.remove_first());

}

catch(IPCException&)

{

PEGASUS_ASSERT(0);

}

if(execute == true)

cu->execute();

}

void Thread::thread_switch()

//thread_data *Thread::put_tsd(const Sint8 *key, void (*delete_func)(void *), Uint32 size, void *value)

#ifndef PEGASUS_THREAD_EXIT_NATIVE

void Thread::exit_self(PEGASUS_THREAD_RETURN exit_code)

{

// execute the cleanup stack and then return

while( _cleanup.count() )

{

try

{

cleanup_pop(true);

}

catch(IPCException&)

{

PEGASUS_ASSERT(0);

#if defined(PEGASUS_PLATFORM_ZOS_ZSERIES_IBM)

break;

pthread_yield(NULL);

}

#else

}

sched_yield();

_exit_code = exit_code;

exit_thread(exit_code);

_handle.thid = 0;

}

#endif

// l10n start

Sint8 Thread::initializeKey()

{

PEG_METHOD_ENTER(TRC_THREAD, "Thread::initializeKey");

if (!Thread::_key_initialized)

{

if (Thread::_key_error)

{

Tracer::trace(TRC_THREAD, Tracer::LEVEL4,

"Thread: ERROR - thread key error");

return -1;

}

if (pegasus_key_create(&Thread::_platform_thread_key) == 0)

void Thread::sleep(Uint32 msec)

{

Tracer::trace(TRC_THREAD, Tracer::LEVEL4,

Threads::sleep(msec);

"Thread: able to create a thread key");

Thread::_key_initialized = true;

}

else

void Thread::join()

{

Tracer::trace(TRC_THREAD, Tracer::LEVEL4,

if (!_is_detached && !Threads::null(_handle.thid))

"Thread: ERROR - unable to create a thread key");

pthread_join(_handle.thid.thread, &_exit_code);

Thread::_key_error = true;

return -1;

}

PEG_METHOD_EXIT();

Threads::clear(_handle.thid);

return 0;

}

Thread * Thread::getCurrent()

void Thread::detach()

{

PEG_METHOD_ENTER(TRC_THREAD, "Thread::getCurrent");

_is_detached = true;

if (Thread::initializeKey() != 0)

#if defined(PEGASUS_PLATFORM_ZOS_ZSERIES_IBM)

{

pthread_t thread_id=_handle.thid.thread;

return NULL;

pthread_detach(&thread_id);

}

#else

PEG_METHOD_EXIT();

pthread_detach(_handle.thid.thread);

return (Thread *)pegasus_get_thread_specific(_platform_thread_key);

#endif

}

void Thread::setCurrent(Thread * thrd)

ThreadStatus Thread::run()

{

PEG_METHOD_ENTER(TRC_THREAD, "Thread::setCurrent");

StartWrapperArg *arg = new StartWrapperArg;

if (Thread::initializeKey() == 0)

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 == -1)

rc = errno;

if ((rc == EAGAIN) || (rc == ENOMEM))

{

if (pegasus_set_thread_specific(

Threads::clear(_handle.thid);

Thread::_platform_thread_key, (void *) thrd) == 0)

delete arg;

{

return PEGASUS_THREAD_INSUFFICIENT_RESOURCES;

Tracer::trace(TRC_THREAD, Tracer::LEVEL4,

"Successful set Thread * into thread specific storage");

}

else

else if (rc != 0)

{

Tracer::trace(TRC_THREAD, Tracer::LEVEL4,

Threads::clear(_handle.thid);

"ERROR: error setting Thread * into thread specific storage");

delete arg;

}

return PEGASUS_THREAD_SETUP_FAILURE;

}

PEG_METHOD_EXIT();

return PEGASUS_THREAD_OK;

}

AcceptLanguageList * Thread::getLanguages()

Thread::Thread(

ThreadReturnType(PEGASUS_THREAD_CDECL* start) (void*),

void* parameter,

Boolean detached)

: _is_detached(detached),

_start(start),

_cleanup(),

_tsd(),

_thread_parm(parameter),

_exit_code(0)

{

PEG_METHOD_ENTER(TRC_THREAD, "Thread::getLanguages");

Threads::clear(_handle.thid);

Thread * curThrd = Thread::getCurrent();

if (curThrd == NULL)

return NULL;

AcceptLanguageList * acceptLangs =

(AcceptLanguageList *)curThrd->reference_tsd("acceptLanguages");

curThrd->dereference_tsd();

PEG_METHOD_EXIT();

return acceptLangs;

}

void Thread::setLanguages(AcceptLanguageList *langs) //l10n

Thread::~Thread()

{

PEG_METHOD_ENTER(TRC_THREAD, "Thread::setLanguages");

try

Thread* currentThrd = Thread::getCurrent();

if (currentThrd != NULL)

{

// deletes the old tsd and creates a new one

join();

currentThrd->put_tsd("acceptLanguages",

empty_tsd();

language_delete,

sizeof(AcceptLanguageList *),

langs);

}

PEG_METHOD_EXIT();

}

catch (...)

void Thread::clearLanguages() //l10n

{

PEG_METHOD_ENTER(TRC_THREAD, "Thread::clearLanguages");

Thread * currentThrd = Thread::getCurrent();

if (currentThrd != NULL)

{

// deletes the old tsd

// Do not allow the destructor to throw an exception

currentThrd->delete_tsd("acceptLanguages");

}

PEG_METHOD_EXIT();

}

// l10n end

#endif /* PEGASUS_HAVE_PTHREADS */

///////////////////////////////////////////////////////////////////////////////

//==============================================================================

// ThreadPool

// Windows Threads Implementation:

///////////////////////////////////////////////////////////////////////////////

//==============================================================================

#if defined(PEGASUS_HAVE_WINDOWS_THREADS)

ThreadPool::ThreadPool(

ThreadStatus Thread::run()

Sint16 initialSize,

const char* key,

Sint16 minThreads,

Sint16 maxThreads,

struct timeval& deallocateWait)

: _maxThreads(maxThreads),

_minThreads(minThreads),

_currentThreads(0),

_idleThreads(true),

_runningThreads(true),

_dying(0)

{

_deallocateWait.tv_sec = deallocateWait.tv_sec;

// Note: A Win32 thread ID is not the same thing as a pthread ID.

_deallocateWait.tv_usec = deallocateWait.tv_usec;

// 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;

memset(_key, 0x00, 17);

if (Threads::null(_handle.thid))

if (key != 0)

{

strncpy(_key, key, 16);

if (errno == EAGAIN)

{

return PEGASUS_THREAD_INSUFFICIENT_RESOURCES;

}

else

if ((_maxThreads > 0) && (_maxThreads < initialSize))

{

_maxThreads = initialSize;

return PEGASUS_THREAD_SETUP_FAILURE;

}

return PEGASUS_THREAD_OK;

}

if (_minThreads > initialSize)

void Thread::cancel()

{

_minThreads = initialSize;

_cancelled = true;

}

for (int i = 0; i < initialSize; i++)

void Thread::thread_switch()

{

_addToIdleThreadsQueue(_initializeThread());

Sleep(0);

}

ThreadPool::~ThreadPool()

void Thread::sleep(Uint32 milliseconds)

{

PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::~ThreadPool");

Sleep(milliseconds);

}

try

void Thread::join()

{

if (!Threads::null(_handle.thid))

{

// 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_first();

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

{

pegasus_yield();

// Currently this is the only way to ensure this code does

}

// not

}

// hang forever.

}

if (WaitForSingleObject(_handle.thid.handle, 10000) ==

catch (...)

WAIT_TIMEOUT)

{

TerminateThread(_handle.thid.handle, 0);

}

PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL ThreadPool::_loop(void* parm)

DWORD exit_code = 0;

{

GetExitCodeThread(_handle.thid.handle, &exit_code);

PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::_loop");

_exit_code = (ThreadReturnType) exit_code;

}

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;

CloseHandle(_handle.thid.handle);

struct timeval* lastActivityTime = 0;

Threads::clear(_handle.thid);

}

try

void Thread::detach()

{

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),

_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

{

sleep_sem->wait();

join();

empty_tsd();

}

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());

}

Boolean Thread::_signals_blocked = false;

#ifndef PEGASUS_OS_ZOS

TSDKeyType Thread::_platform_thread_key = TSDKeyType(-1);

#else

TSDKeyType Thread::_platform_thread_key;

#endif

catch (...)

Boolean Thread::_key_initialized = false;

Boolean Thread::_key_error = false;

void Thread::cleanup_push(void (*routine) (void *), void *parm)

{

PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,

AutoPtr < cleanup_handler > cu(new cleanup_handler(routine, parm));

"Unknown exception from work in ThreadPool::_loop.");

_cleanup.insert_front(cu.get());

cu.release();

return;

}

// put myself back onto the available list

void Thread::cleanup_pop(Boolean execute)

{

AutoPtr < cleanup_handler > cu;

try

{

gettimeofday(lastActivityTime, NULL);

cu.reset(_cleanup.remove_front());

if (blocking_sem != 0)

}

catch (IPCException &)

{

blocking_sem->signal();

PEGASUS_ASSERT(0);

}

if (execute == true)

cu->execute();

}

Boolean removed = pool->_runningThreads.remove((void *)myself);

PEGASUS_ASSERT(removed);

pool->_idleThreads.insert_first(myself);

void Thread::exit_self(ThreadReturnType exit_code)

}

catch (...)

{

Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,

#if !defined(PEGASUS_PLATFORM_AIX_RS_IBMCXX) \

"ThreadPool::_loop: Adding thread to idle pool failed.");

&& !defined(PEGASUS_PLATFORM_PASE_ISERIES_IBMCXX)

PEGASUS_ASSERT(false);

Threads::exit(exit_code);

pool->_currentThreads--;

#else

PEG_METHOD_EXIT();

// execute the cleanup stack and then return

return((PEGASUS_THREAD_RETURN)1);

while (_cleanup.size())

}

catch (const Exception& e)

{

PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,

try

"Caught exception: \"" + e.getMessage() + "\". Exiting _loop.");

{

cleanup_pop(true);

}

catch (...)

catch (IPCException &)

{

PEG_TRACE_STRING(TRC_DISCARDED_DATA, Tracer::LEVEL2,

PEGASUS_ASSERT(0);

"Caught unrecognized exception. Exiting _loop.");

break;

}

PEG_METHOD_EXIT();

_exit_code = exit_code;

return((PEGASUS_THREAD_RETURN)0);

Threads::exit(exit_code);

Threads::clear(_handle.thid);

#endif

}

ThreadStatus ThreadPool::allocate_and_awaken(

Sint8 Thread::initializeKey()

void* parm,

PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL* work)(void *),

Semaphore* blocking)

{

PEG_METHOD_ENTER(TRC_THREAD, "ThreadPool::allocate_and_awaken");

PEG_METHOD_ENTER(TRC_THREAD, "Thread::initializeKey");

if (!Thread::_key_initialized)

// 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())

if (Thread::_key_error)

{

Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,

PEG_TRACE_CSTRING(TRC_THREAD, Tracer::LEVEL4,

"ThreadPool::allocate_and_awaken: ThreadPool is dying(1).");

"Thread: ERROR - thread key error");

return PEGASUS_THREAD_UNAVAILABLE;

return -1;

}

struct timeval start;

gettimeofday(&start, NULL);

Thread* th = 0;

th = _idleThreads.remove_first();

if (th == 0)

if (TSDKey::create(&Thread::_platform_thread_key) == 0)

{

if ((_maxThreads == 0) ||

PEG_TRACE_CSTRING(TRC_THREAD, Tracer::LEVEL4,

(_currentThreads.get() < Uint32(_maxThreads)))

"Thread: able to create a thread key");

{

Thread::_key_initialized = true;

th = _initializeThread();

}

else

if (th == 0)

{

// ATTN-DME-P3-20031103: This trace message should not be

PEG_TRACE_CSTRING(TRC_THREAD, Tracer::LEVEL4,

// be labeled TRC_DISCARDED_DATA, because it does not

"Thread: ERROR - unable to create a thread key");

// necessarily imply that a failure has occurred. However,

Thread::_key_error = true;

// this label is being used temporarily to help isolate

return -1;

// 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.count(), _idleThreads.count());

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( 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_first(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;

return 0;

}

// caller is responsible for only calling this routine during slack periods

Thread *Thread::getCurrent()

// 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.count();

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_last();

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_last(thread);

set_thread_specific(Thread::_platform_thread_key,

break;

(void *) thrd) == 0)

}

{

PEG_TRACE_CSTRING(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

{

_idleThreads.insert_first(thread);

PEG_TRACE_CSTRING(TRC_THREAD, Tracer::LEVEL4,

"ERROR: error setting Thread * into thread specific storage");

}

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();

return acceptLangs;

}

Boolean ThreadPool::_timeIntervalExpired(

void Thread::setLanguages(const AcceptLanguageList& langs)

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;

Thread *currentThrd = Thread::getCurrent();

Uint32 usec;

if (currentThrd != NULL)

pegasus_gettimeofday(&now);

{

pegasus_gettimeofday(&remaining); // Avoid valgrind error

AutoPtr<AcceptLanguageList> langsCopy(new AcceptLanguageList(langs));

finish.tv_sec = start->tv_sec + interval->tv_sec;

// deletes the old tsd and creates a new one

usec = start->tv_usec + interval->tv_usec;

currentThrd->put_tsd(

finish.tv_sec += (usec / 1000000);

"acceptLanguages",

usec %= 1000000;

language_delete,

finish.tv_usec = usec;

sizeof (AcceptLanguageList *),

langsCopy.get());

return (timeval_subtract(&remaining, &finish, &now) != 0);

langsCopy.release();

}

void ThreadPool::_deleteSemaphore(void *p)

PEG_METHOD_EXIT();

{

delete (Semaphore *)p;

}

Thread* ThreadPool::_initializeThread()

void Thread::clearLanguages()

{

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();

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_first(th);

bool isEnhancedSecurity = 99;

}

#endif

catch (...)

{

Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,

"ThreadPool::_addToIdleThreadsQueue: _idleThreads.insert_first "

"failed.");

}

PEGASUS_NAMESPACE_END

Legend:

Removed from v.1.87.4.1
changed lines
	Added in v.1.105

No CVS admin address has been configured