threadpool.cpp

Pegasus is an open-source implementationof the DMTF CIM and WBEM standards. It is designed to be por

//%2006////////////////////////////////////////////////////////////////////////
//
// Copyright (c) 2000, 2001, 2002 BMC Software; Hewlett-Packard Development
// 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
// of this software and associated documentation files (the "Software"), to
// deal in the Software without restriction, including without limitation the
// rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
// sell copies of the Software, and to permit persons to whom the 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.
//
//==============================================================================
//
//%/////////////////////////////////////////////////////////////////////////////

#include "ThreadPool.h"
#include "Thread.h"
#include <exception>
#include <Pegasus/Common/Tracer.h>
#include "Time.h"

PEGASUS_USING_STD;

PEGASUS_NAMESPACE_BEGIN

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