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.1.2.10 and 1.112.8.1

version 1.1.2.10, 2001/10/29 11:25:22

version 1.112.8.1, 2013/06/03 22:35:13

Line 1

//%/////////////////////////////////////////////////////////////////////////////

//%LICENSE////////////////////////////////////////////////////////////////

// Licensed to The Open Group (TOG) under one or more contributor license

// agreements. Refer to the OpenPegasusNOTICE.txt file distributed with

// this work for additional information regarding copyright ownership.

// Each contributor licenses this file to you under the OpenPegasus Open

// Source License; you may not use this file except in compliance with the

// License.

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

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

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

// deal in the Software without restriction, including without limitation the

// to deal in the Software without restriction, including without limitation

// rights to use, copy, modify, merge, publish, distribute, sublicense, and/or

// the rights to use, copy, modify, merge, publish, distribute, sublicense,

// sell copies of the Software, and to permit persons to whom the Software is

// and/or sell copies of the Software, and to permit persons to whom the

// furnished to do so, subject to the following conditions:

// Software is furnished to do so, subject to the following conditions:

// THE ABOVE COPYRIGHT NOTICE AND THIS PERMISSION NOTICE SHALL BE INCLUDED IN

// ALL COPIES OR SUBSTANTIAL PORTIONS OF THE SOFTWARE. THE SOFTWARE IS PROVIDED

// "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT

// LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR

// PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT

// HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN

// ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION

// WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

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

// The above copyright notice and this permission notice shall be included

// in all copies or substantial portions of the Software.

// Author: Mike Day (mdday@us.ibm.com)

// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS

// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF

// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.

// IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY

// CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,

// TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE

// SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

// Modified By:

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

//%/////////////////////////////////////////////////////////////////////////////

#include "Thread.h"

#include <Pegasus/Common/IPC.h>

#include <errno.h>

#include <exception>

#include <Pegasus/Common/Tracer.h>

#include <Pegasus/Common/AutoPtr.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"

#else

# error "Unsupported platform"

#endif

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

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

return return_value;

}

catch(IPCException& e)

void Thread::cancel()

{

delete cu;

pthread_cancel(_handle.thid.thread);

throw;

}

return;

}

void Thread::cleanup_pop(Boolean execute) throw(IPCException)

void Thread::thread_switch()

{

cleanup_handler *cu ;

try

{

cu = _cleanup.remove_first() ;

#if defined(PEGASUS_OS_ZOS)

pthread_yield(NULL);

#else

sched_yield();

#endif

}

catch(IPCException& e)

void Thread::sleep(Uint32 msec)

{

PEGASUS_ASSERT(0);

Threads::sleep(msec);

}

if(execute == true)

cu->execute();

void Thread::join()

delete cu;

{

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

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

Threads::clear(_handle.thid);

}

void Thread::detach()

{

_is_detached = true;

#if defined(PEGASUS_OS_ZOS)

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;

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

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.

#ifndef PEGASUS_THREAD_EXIT_NATIVE

if (rc == -1)

void Thread::exit_self(PEGASUS_THREAD_RETURN exit_code)

rc = errno;

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

{

// execute the cleanup stack and then return

Threads::clear(_handle.thid);

while( _cleanup.count() )

delete arg;

return PEGASUS_THREAD_INSUFFICIENT_RESOURCES;

}

else if (rc != 0)

{

try

Threads::clear(_handle.thid);

delete arg;

return PEGASUS_THREAD_SETUP_FAILURE;

}

return PEGASUS_THREAD_OK;

}

Thread::Thread(

ThreadReturnType(PEGASUS_THREAD_CDECL* start) (void*),

void* parameter,

Boolean detached)

: _is_detached(detached),

_start(start),

_cleanup(),

_thread_parm(parameter),

_exit_code(0)

{

cleanup_pop(true);

Threads::clear(_handle.thid);

memset(_tsd, 0, sizeof(_tsd));

}

catch(IPCException& e)

Thread::~Thread()

{

PEGASUS_ASSERT(0);

try

break;

{

join();

empty_tsd();

}

catch (...)

{

// Do not allow the destructor to throw an exception

}

_exit_code = exit_code;

exit_thread(exit_code);

}

#endif /* PEGASUS_HAVE_PTHREADS */

#endif

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

// Windows Threads Implementation:

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

#if defined(PEGASUS_HAVE_WINDOWS_THREADS)

ThreadPool::ThreadPool(Sint16 initial_size,

ThreadStatus Thread::run()

Sint16 max,

{

Sint16 min,

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

Sint8 *key)

// Win32 threads have both a thread ID and a handle. The handle

: _max_threads(max), _min_threads(min),

// is used in the wait functions, etc.

_current_threads(0), _waiters(initial_size),

// So _handle.thid is actually the thread handle.

_pool_sem(0), _pool(true), _running(true),

_dying(0)

{

_allocate_wait.tv_sec = 1;

_allocate_wait.tv_usec = 0;

_deallocate_wait.tv_sec = 30;

_deallocate_wait.tv_usec = 0;

_deadlock_detect.tv_sec = 60;

_deadlock_detect.tv_usec = 0;

memset(_key, 0x00, 17);

if(key != 0)

strncpy(_key, key, 16);

if(_max_threads < initial_size)

_max_threads = initial_size;

if(_min_threads > initial_size)

_min_threads = initial_size;

int i;

for(i = 0; i < initial_size; i++)

{

_link_pool(_init_thread());

}

ThreadPool::~ThreadPool(void)

unsigned threadid = 0;

{

_dying++;

Thread *th = _pool.remove_first();

while(th != 0)

{

// signal the thread's sleep semaphore

th->cancel();

th->join();

th->empty_tsd();

delete th;

th = _pool.remove_first();

}

// make this static to the class

ThreadType tt;

PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL ThreadPool::_loop(void *parm)

tt.handle = (HANDLE) _beginthreadex(NULL, 0, _start, this, 0, &threadid);

{

_handle.thid = tt;

Thread *myself = (Thread *)parm;

if(myself == 0)

throw NullPointer();

ThreadPool *pool = (ThreadPool *)myself->get_parm();

if(pool == 0 )

throw NullPointer();

Semaphore *sleep_sem;

struct timeval *deadlock_timer;

try

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

{

sleep_sem = (Semaphore *)myself->reference_tsd("sleep sem");

if (errno == EAGAIN)

myself->dereference_tsd();

{

deadlock_timer = (struct timeval *)myself->reference_tsd("deadlock timer");

return PEGASUS_THREAD_INSUFFICIENT_RESOURCES;

myself->dereference_tsd();

}

catch(IPCException & e)

else

{

myself->exit_self(0);

return PEGASUS_THREAD_SETUP_FAILURE;

}

return PEGASUS_THREAD_OK;

}

if(sleep_sem == 0 || deadlock_timer == 0)

throw NullPointer();

while(pool->_dying < 1)

void Thread::cancel()

{

myself->test_cancel();

_cancelled = true;

sleep_sem->wait();

}

// when we awaken we reside on the running queue, not the pool queue

myself->test_cancel();

gettimeofday(deadlock_timer, NULL);

PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *_work)(void *);

void *parm;

try

void Thread::thread_switch()

{

_work = (PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *)(void *)) \

Sleep(0);

myself->reference_tsd("work func");

myself->dereference_tsd();

parm = myself->reference_tsd("work parm");

myself->dereference_tsd();

}

catch(IPCException & e)

void Thread::sleep(Uint32 milliseconds)

{

myself->exit_self(0);

Sleep(milliseconds);

}

if(_work == 0)

void Thread::join()

throw NullPointer();

{

_work(parm);

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

{

// put myself back onto the available list

if (!_is_detached)

try

{

if (!_cancelled)

{

pool->_running.remove((void *)myself);

// Emulate the unix join api. Caller sleeps until thread is

pool->_link_pool(myself);

// done.

WaitForSingleObject(_handle.thid.handle, INFINITE);

}

catch(IPCException & e)

else

{

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

// not

// hang forever.

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

WAIT_TIMEOUT)

{

myself->exit_self(0);

TerminateThread(_handle.thid.handle, 0);

}

myself->exit_self(0);

return((PEGASUS_THREAD_RETURN)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 ThreadPool::allocate_and_awaken(void *parm,

void Thread::detach()

PEGASUS_THREAD_RETURN \

(PEGASUS_THREAD_CDECL *work)(void *))

throw(IPCException)

{

struct timeval start;

_is_detached = true;

gettimeofday(&start, NULL);

}

Thread *th = _pool.remove_first();

Thread::Thread(

ThreadReturnType(PEGASUS_THREAD_CDECL* start)(void*),

void *parameter,

Boolean detached) :_is_detached(detached),

_cancelled(false),

_start(start),

_cleanup(),

_thread_parm(parameter),

_exit_code(0)

{

Threads::clear(_handle.thid);

memset(_tsd, 0, sizeof(_tsd));

}

while (th == 0 && _dying < 1)

Thread::~Thread()

{

try // we couldn't get a free thread from the pool

try

{

// wait for the right interval and try again

join();

while(th == 0 && _dying < 1)

empty_tsd();

}

catch (...)

{

_check_deadlock(&start);

Uint32 interval = _allocate_wait.tv_sec * 1000;

if(_allocate_wait.tv_usec > 0)

interval += (_deallocate_wait.tv_usec / 1000);

// will throw a timeout if no thread comes free

_pool_sem.time_wait(interval);

th = _pool.remove_first();

}

catch(TimeOut & to)

#endif /* PEGASUS_HAVE_WINDOWS_THREADS */

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

// Common implementation:

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

void language_delete(void *data)

{

if(_current_threads < _max_threads)

if (data != NULL)

{

th = _init_thread();

AutoPtr < AcceptLanguageList > al(static_cast <

break;

AcceptLanguageList * >(data));

}

// will throw a Deadlock Exception before falling out of the loop

_check_deadlock(&start);

} // while th == null

if(_dying < 1)

{

// initialize the thread data with the work function and parameters

th->remove_tsd("work func");

th->put_tsd("work func", NULL,

sizeof( PEGASUS_THREAD_RETURN (PEGASUS_THREAD_CDECL *)(void *)),

(void *)work);

th->remove_tsd("work parm");

th->put_tsd("work parm", NULL, sizeof(void *), parm);

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

throw NullPointer();

sleep_sem->signal();

}

else

_pool.insert_first(th);

}

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

Boolean Thread::_signals_blocked = false;

// but should call it at least once per _deadlock_detect with the running q

#ifndef PEGASUS_OS_ZOS

// and at least once per _deallocate_wait for the pool q

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

#else

TSDKeyType Thread::_platform_thread_key;

#endif

Boolean Thread::_key_initialized = false;

Boolean Thread::_key_error = false;

void ThreadPool::_kill_dead_threads(DQueue<Thread> *q, Boolean (*check)(struct timeval *))

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

throw(IPCException)

{

struct timeval now;

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

gettimeofday(&now, NULL);

_cleanup.insert_front(cu.get());

cu.release();

DQueue<Thread> dead ;

return;

}

if(q->count() > 0 )

void Thread::cleanup_pop(Boolean execute)

{

AutoPtr < cleanup_handler > cu;

try

{

q->try_lock();

cu.reset(_cleanup.remove_front());

}

catch(AlreadyLocked & a)

catch (...)

{

return;

PEGASUS_ASSERT(0);

}

if (execute == true)

{

cu->execute();

}

Thread *context = 0;

struct timeval dt = { 0, 0 };

void Thread::exit_self(ThreadReturnType exit_code)

struct timeval *dtp;

{

Thread *th = q->next(context);

#if !defined(PEGASUS_PLATFORM_AIX_RS_IBMCXX) \

while (th != 0 )

&& !defined(PEGASUS_PLATFORM_PASE_ISERIES_IBMCXX)

Threads::exit(exit_code);

#else

// execute the cleanup stack and then return

while (_cleanup.size())

{

try

{

dtp = (struct timeval *)th->try_reference_tsd("deadlock timer");

cleanup_pop(true);

}

catch(AlreadyLocked & a)

catch (...)

{

context = th;

PEGASUS_ASSERT(0);

th = q->next(context);

break;

continue;

}

_exit_code = exit_code;

Threads::exit(exit_code);

Threads::clear(_handle.thid);

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

if (Thread::_key_error)

if( true == check(&dt))

{

th = q->remove_no_lock((void *)th);

PEG_TRACE_CSTRING(TRC_THREAD, Tracer::LEVEL1,

"Thread: ERROR - thread key error");

return -1;

}

if(th != 0)

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

{

dead.insert_first(th);

PEG_TRACE_CSTRING(TRC_THREAD, Tracer::LEVEL4,

th = 0;

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

Thread::_key_initialized = true;

}

else

{

PEG_TRACE_CSTRING(TRC_THREAD, Tracer::LEVEL1,

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

Thread::_key_error = true;

return -1;

}

context = th;

th = q->next(context);

}

q->unlock();

PEG_METHOD_EXIT();

return 0;

}

if(dead.count())

Thread *Thread::getCurrent()

{

Thread *th = dead.remove_first();

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

while(th != 0)

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

{

th->cancel();

return NULL;

th->join();

delete th;

th = dead.remove_first();

}

PEG_METHOD_EXIT();

return;

return (Thread *) TSDKey::get_thread_specific(_platform_thread_key);

}

Boolean ThreadPool::_check_time(struct timeval *start, struct timeval *interval)

void Thread::setCurrent(Thread * thrd)

{

struct timeval now;

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

gettimeofday(&now, NULL);

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

if( (now.tv_sec - start->tv_sec) > interval->tv_sec ||

(((now.tv_sec - start->tv_sec) == interval->tv_sec) &&

((now.tv_usec - start->tv_usec) >= interval->tv_usec ) ) )

return true;

else

return false;

}

void ThreadPool::_sleep_sem_del(void *p)

{

if(p != 0)

if (TSDKey::

set_thread_specific(Thread::_platform_thread_key,

(void *) thrd) == 0)

{

delete (Semaphore *)p;

PEG_TRACE_CSTRING(TRC_THREAD, Tracer::LEVEL4,

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

}

else

{

PEG_TRACE_CSTRING(TRC_THREAD, Tracer::LEVEL1,

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

}

PEG_METHOD_EXIT();

}

inline void ThreadPool::_check_deadlock(struct timeval *start) throw(Deadlock)

AcceptLanguageList *Thread::getLanguages()

{

if (true == _check_time(start, &_deadlock_detect))

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

throw Deadlock(pegasus_thread_self());

return;

Thread *curThrd = Thread::getCurrent();

if (curThrd == NULL)

return NULL;

AcceptLanguageList *acceptLangs =

(AcceptLanguageList *) curThrd->reference_tsd(TSD_ACCEPT_LANGUAGES);

curThrd->dereference_tsd();

PEG_METHOD_EXIT();

return acceptLangs;

}

void Thread::setLanguages(const AcceptLanguageList& langs)

{

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

inline Boolean ThreadPool::_check_deadlock_no_throw(struct timeval *start)

Thread *currentThrd = Thread::getCurrent();

if (currentThrd != NULL)

{

return(_check_time(start, &_deadlock_detect));

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

// deletes the old tsd and creates a new one

currentThrd->put_tsd(

TSD_ACCEPT_LANGUAGES,

language_delete,

sizeof (AcceptLanguageList *),

langsCopy.get());

langsCopy.release();

}

inline Boolean ThreadPool::_check_dealloc(struct timeval *start)

PEG_METHOD_EXIT();

{

return(_check_time(start, &_deallocate_wait));

}

inline Thread *ThreadPool::_init_thread(void) throw(IPCException)

void Thread::clearLanguages()

{

Thread *th = (Thread *) new Thread(&_loop, this, false);

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

// allocate a sleep semaphore and pass it in the thread context

// initial count is zero, loop function will sleep until

// we signal the semaphore

Semaphore *sleep_sem = (Semaphore *) new Semaphore(0);

th->put_tsd("sleep sem", &_sleep_sem_del, sizeof(Semaphore), (void *)sleep_sem);

struct timeval *dldt = (struct timeval *) ::operator new(sizeof(struct timeval));

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

return th;

}

inline void ThreadPool::_link_pool(Thread *th) throw(IPCException)

Thread *currentThrd = Thread::getCurrent();

if (currentThrd != NULL)

{

if(th == 0)

// deletes the old tsd

throw NullPointer();

currentThrd->delete_tsd(TSD_ACCEPT_LANGUAGES);

_pool.insert_first(th);

_pool_sem.signal();

}

PEG_METHOD_EXIT();

}

PEGASUS_NAMESPACE_END

// ATTN: not sure where to put this!

#ifdef PEGASUS_ZOS_SECURITY

bool isEnhancedSecurity = 99;

#endif

PEGASUS_NAMESPACE_END

Legend:

Removed from v.1.1.2.10
changed lines
	Added in v.1.112.8.1

No CVS admin address has been configured