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

//

// Copyright (c) 2005-2012, Industrial Light & Magic, a division of Lucas

// Digital Ltd. LLC

// 

// 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 Industrial Light & Magic 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 THE COPYRIGHT HOLDERS AND CONTRIBUTORS

// "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 THE COPYRIGHT

// OWNER OR CONTRIBUTORS 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.

//

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

//-----------------------------------------------------------------------------

//

//	class Task, class ThreadPool, class TaskGroup

//

//-----------------------------------------------------------------------------

#include "IlmThread.h"

#include "IlmThreadMutex.h"

#include "IlmThreadSemaphore.h"

#include "IlmThreadPool.h"

#include "Iex.h"

#include <list>

using namespace std;

ILMTHREAD_INTERNAL_NAMESPACE_SOURCE_ENTER

namespace {

class WorkerThread: public Thread

{

  public:

    WorkerThread (ThreadPool::Data* data);

    virtual void	run ();

  private:

    ThreadPool::Data *	_data;

};

} //namespace

struct TaskGroup::Data

{

     Data ();

    ~Data ();

    void	addTask () ;

    void	removeTask ();

    Semaphore	isEmpty;        // used to signal that the taskgroup is empty

    int         numPending;     // number of pending tasks to still execute

    Mutex       dtorMutex;      // used to work around the glibc bug:

                                // http://sources.redhat.com/bugzilla/show_bug.cgi?id=12674

};

struct ThreadPool::Data

{

     Data ();

    ~Data();

    void	finish ();

    bool	stopped () const;

    void	stop ();

    Semaphore taskSemaphore;        // threads wait on this for ready tasks

    Mutex taskMutex;                // mutual exclusion for the tasks list

    list<Task*> tasks;              // the list of tasks to execute

    size_t numTasks;                // fast access to list size

                                    //   (list::size() can be O(n))

    Semaphore threadSemaphore;      // signaled when a thread starts executing

    Mutex threadMutex;              // mutual exclusion for threads list

    list<WorkerThread*> threads;    // the list of all threads

    size_t numThreads;              // fast access to list size

    bool stopping;                  // flag indicating whether to stop threads

    Mutex stopMutex;                // mutual exclusion for stopping flag

};

//

// class WorkerThread

//

WorkerThread::WorkerThread (ThreadPool::Data* data):

    _data (data)

{

    start();

}

void

WorkerThread::run ()

{

    //

    // Signal that the thread has started executing

    //

    _data->threadSemaphore.post();

    while (true)

    {

	//

        // Wait for a task to become available

	//

        _data->taskSemaphore.wait();

        {

            Lock taskLock (_data->taskMutex);

	    //

            // If there is a task pending, pop off the next task in the FIFO

	    //

            if (_data->numTasks > 0)

            {

                Task* task = _data->tasks.front();

		TaskGroup* taskGroup = task->group();

                _data->tasks.pop_front();

                _data->numTasks--;

                taskLock.release();

                task->execute();

                taskLock.acquire();

                delete task;

                taskGroup->_data->removeTask();

            }

            else if (_data->stopped())

	    {

                break;

	    }

        }

    }

}

//

// struct TaskGroup::Data

//

TaskGroup::Data::Data (): isEmpty (1), numPending (0)

{

    // empty

}

TaskGroup::Data::~Data ()

{

    //

    // A TaskGroup acts like an "inverted" semaphore: if the count

    // is above 0 then waiting on the taskgroup will block.  This

    // destructor waits until the taskgroup is empty before returning.

    //

    isEmpty.wait ();

    // Alas, given the current bug in glibc we need a secondary

    // syncronisation primitive here to account for the fact that

    // destructing the isEmpty Semaphore in this thread can cause

    // an error for a separate thread that is issuing the post() call.

    // We are entitled to destruct the semaphore at this point, however,

    // that post() call attempts to access data out of the associated

    // memory *after* it has woken the waiting threads, including this one,

    // potentially leading to invalid memory reads.

    // http://sources.redhat.com/bugzilla/show_bug.cgi?id=12674

    Lock lock (dtorMutex);

}

void

TaskGroup::Data::addTask () 

{

    //

    // Any access to the taskgroup is protected by a mutex that is

    // held by the threadpool.  Therefore it is safe to access

    // numPending before we wait on the semaphore.

    //

    if (numPending++ == 0)

	isEmpty.wait ();

}

void

TaskGroup::Data::removeTask ()

{

    // Alas, given the current bug in glibc we need a secondary

    // syncronisation primitive here to account for the fact that

    // destructing the isEmpty Semaphore in a separate thread can

    // cause an error. Issuing the post call here the current libc

    // implementation attempts to access memory *after* it has woken

    // waiting threads.

    // Since other threads are entitled to delete the semaphore the

    // access to the memory location can be invalid.

    // http://sources.redhat.com/bugzilla/show_bug.cgi?id=12674

    if (--numPending == 0)

    {

        Lock lock (dtorMutex);

        isEmpty.post ();

    }

}

//

// struct ThreadPool::Data

//

ThreadPool::Data::Data (): numTasks (0), numThreads (0), stopping (false)

{

    // empty

}

ThreadPool::Data::~Data()

{

    Lock lock (threadMutex);

    finish ();

}

void

ThreadPool::Data::finish ()

{

    stop();

    //

    // Signal enough times to allow all threads to stop.

    //

    // Wait until all threads have started their run functions.

    // If we do not wait before we destroy the threads then it's

    // possible that the threads have not yet called their run

    // functions.

    // If this happens then the run function will be called off

    // of an invalid object and we will crash, most likely with

    // an error like: "pure virtual method called"

    //

    for (size_t i = 0; i < numThreads; i++)

    {

	taskSemaphore.post();

	threadSemaphore.wait();

    }

    //

    // Join all the threads

    //

    for (list<WorkerThread*>::iterator i = threads.begin();

	 i != threads.end();

	 ++i)

    {

	delete (*i);

    }

    Lock lock1 (taskMutex);

    Lock lock2 (stopMutex);

    threads.clear();

    tasks.clear();

    numThreads = 0;

    numTasks = 0;

    stopping = false;

}

bool

ThreadPool::Data::stopped () const

{

    Lock lock (stopMutex);

    return stopping;

}

void

ThreadPool::Data::stop ()

{

    Lock lock (stopMutex);

    stopping = true;

}

//

// class Task

//

Task::Task (TaskGroup* g): _group(g)

{

    // empty

}

Task::~Task()

{

    // empty

}

TaskGroup*

Task::group ()

{

    return _group;

}

TaskGroup::TaskGroup ():

    _data (new Data())

{

    // empty

}

TaskGroup::~TaskGroup ()

{

    delete _data;

}

//

// class ThreadPool

//

ThreadPool::ThreadPool (unsigned nthreads):

    _data (new Data())

{

    setNumThreads (nthreads);

}

ThreadPool::~ThreadPool ()

{

    delete _data;

}

int

ThreadPool::numThreads () const

{

    Lock lock (_data->threadMutex);

    return _data->numThreads;

}

void

ThreadPool::setNumThreads (int count)

{

    if (count < 0)

        throw IEX_INTERNAL_NAMESPACE::ArgExc ("Attempt to set the number of threads "

			   "in a thread pool to a negative value.");

    //

    // Lock access to thread list and size

    //

    Lock lock (_data->threadMutex);

    if ((size_t)count > _data->numThreads)

    {

	//

        // Add more threads

	//

        while (_data->numThreads < (size_t)count)

        {

            _data->threads.push_back (new WorkerThread (_data));

            _data->numThreads++;

        }

    }

    else if ((size_t)count < _data->numThreads)

    {

	//

	// Wait until all existing threads are finished processing,

	// then delete all threads.

	//

        _data->finish ();

	//

        // Add in new threads

	//

        while (_data->numThreads < (size_t)count)

        {

            _data->threads.push_back (new WorkerThread (_data));

            _data->numThreads++;

        }

    }

}

void

ThreadPool::addTask (Task* task) 

{

    //

    // Lock the threads, needed to access numThreads

    //

    Lock lock (_data->threadMutex);

    if (_data->numThreads == 0)

    {

        task->execute ();

        delete task;

    }

    else

    {

	//

        // Get exclusive access to the tasks queue

	//

        {

            Lock taskLock (_data->taskMutex);

	    //

            // Push the new task into the FIFO

	    //

            _data->tasks.push_back (task);

            _data->numTasks++;

            task->group()->_data->addTask();

        }

	//

        // Signal that we have a new task to process

	//

        _data->taskSemaphore.post ();

    }

}

ThreadPool&

ThreadPool::globalThreadPool ()

{

    //

    // The global thread pool

    //

    static ThreadPool gThreadPool (0);

    return gThreadPool;

}

void

ThreadPool::addGlobalTask (Task* task)

{

    globalThreadPool().addTask (task);

}

ILMTHREAD_INTERNAL_NAMESPACE_SOURCE_EXIT