threadpool.cpp

Go to the documentation of this file.

/*
 * threadpool.cpp
 *
 * Copyright (C) 2007  Thomas A. Vaughan
 * All rights reserved.
 *
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *     * Redistributions of source code must retain the above copyright
 *       notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above copyright
 *       notice, this list of conditions and the following disclaimer in the
 *       documentation and/or other materials provided with the distribution.
 *     * Neither the name of the <organization> nor the
 *       names of its contributors may be used to endorse or promote products
 *       derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THOMAS A. VAUGHAN ''AS IS'' AND ANY
 * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL THOMAS A. VAUGHAN BE LIABLE FOR ANY
 * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 *
 * Implementation of a basic threadpool object.
 *
 * NOTE: at the moment, the implementation is *VERY* basic!  There is no pool at
 * all, and instead threads are created for every incoming request.  But I
 * wanted to at least set up the abstraction, and this can later be filled in
 * with real threadpool behavior.
 */

// includes --------------------------------------------------------------------
#include "threadpool.h"         // always include our own header first!

#include "common/wave_ex.h"
#include "datahash/datahash_util.h"
#include "perf/perf.h"
#include "threadsafe/smart_mutex.h"
#include "threadsafe/threadsafe.h"


namespace threadpool {


// interface destructor implementation
Pool::~Pool(void) throw() { }


////////////////////////////////////////////////////////////////////////////////
//
//      static helper methods
//
////////////////////////////////////////////////////////////////////////////////


////////////////////////////////////////////////////////////////////////////////
//
//      TPool - object that implements the threadpool::Pool interface
//
////////////////////////////////////////////////////////////////////////////////

class TPool : public Pool {
public:
        TPool(void) throw();
        ~TPool(void) throw() { }

        // public class methods ------------------------------------------------
        void initialize(IN const Datahash * params);

        // threadpool::Pool class interface methods ----------------------------
        bool addRequest(IN thread_work_fn work_fn,
                                IN void * context);

private:
        // private typedefs ----------------------------------------------------
        struct request_t {
                request_t(void) { bzero(this, sizeof(request_t)); }

                thread_work_fn          fn;
                void *                  context;
                TPool *                 pThis;
        };

        // private helper functions --------------------------------------------
        void updateThreadCount(IN int increment) throw();
        static void * poolThreadStart(IN void * ctx);

        // private data members ------------------------------------------------
        smart_mutex     m_mutex;
        int             m_maxThreadCount;
        int             m_threadCount;
};



TPool::TPool
(
void
)
throw()
{
        m_maxThreadCount = 0;
        m_threadCount = 0;
}



void
TPool::initialize
(
IN const Datahash * params
)
{
        ASSERT(params, "null");

        m_maxThreadCount = getInt(params, "maxThreads");
        ASSERT(m_maxThreadCount > 0, "Bad max thread count: %d",
            m_maxThreadCount);
}



bool
TPool::addRequest
(
IN thread_work_fn fn,
IN void * context
)
{
        perf::Timer timer("threadpool::addRequest");
        ASSERT(fn, "null");
        // ASSERT(context) -- we don't care!

        // NOTE: we aren't actually maintaining a thread pool!  But we at
        // least keep an eye on the thread count...

        smart_ptr<request_t> req = new request_t;
        ASSERT(req, "out of memory");

        req->fn = fn;
        req->context = context;
        req->pThis = this;

        thread_id_t thread_id_unused;
        ASSERT_THROW(!createThread(poolThreadStart, req, thread_id_unused),
            "Failed to create thread for threadpool?");
        req.disown();   // thread owns this now!

        // worked just fine
        return true;
}



void *
TPool::poolThreadStart
(
IN void * ctx
)
{
        request_t * preq = (request_t *) ctx;
        ASSERT(preq, "null thread context?");
        smart_ptr<request_t> req = preq;
        ASSERT(req, "failed to assign smart pointer?");
        ASSERT(req->pThis, "null");
        ASSERT(req->fn, "null thread work function");
        // ASSERT(req->context) - we don't care

        // increment thread count
        req->pThis->updateThreadCount(+1);

        // invoke provided work function and provide context
        try {
                req->fn(req->context);
        } catch (std::exception& e) {
                DPRINTF("Exception in thread pool while handling request: %s",
                    e.what());
        } catch (...) {
                DPRINTF("Unknown exception handling thread pool request!");
        }

        // decrement thread count
        req->pThis->updateThreadCount(-1);

        // all done
        return NULL;
}



void
TPool::updateThreadCount
(
IN int increment
)
throw()
{
        int count = 0;
        {
                mlock lock(m_mutex);
                m_threadCount += increment;
                count = m_threadCount;
        }

        if (count < 0) {
                DPRINTF("ERROR!  Negative thread count? %d", count);
        }
        if (count > m_maxThreadCount) {
                DPRINTF("WARNING!  Thread count is higher than requested size");
                DPRINTF("  Requested %d threads, have %d currently",
                    m_maxThreadCount, count);
        }
}



////////////////////////////////////////////////////////////////////////////////
//
//      public API
//
////////////////////////////////////////////////////////////////////////////////

smart_ptr<Pool>
Pool::create
(
IN const Datahash * params
)
{
        ASSERT(params, "null");

        smart_ptr<TPool> local = new TPool;
        ASSERT(local, "out of memory");

        local->initialize(params);

        return local;
}



};      // threadpool namespace