dlib C++ Library - threads_kernel_shared.cpp

    // Copyright (C) 2003  Davis E. King (davisking@users.sourceforge.net)
    // License: Boost Software License   See LICENSE.txt for the full license.
    #ifndef DLIB_THREADS_KERNEL_SHARED_CPp_
    #define DLIB_THREADS_KERNEL_SHARED_CPp_

    #include "threads_kernel_shared.h"
    #include "../assert.h"
    #include <iostream>

    #ifndef DLIB_THREAD_POOL_TIMEOUT
    // default to 30000 milliseconds
    #define DLIB_THREAD_POOL_TIMEOUT 30000
    #endif

    namespace dlib
    {

    // ----------------------------------------------------------------------------------------
    // ----------------------------------------------------------------------------------------
    // threader functions
    // ----------------------------------------------------------------------------------------
    // ----------------------------------------------------------------------------------------

        namespace threads_kernel_shared
        {

            threader& thread_pool (
            )
            {
                static threader* thread_pool = new threader;
                return *thread_pool;
            }

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

            struct threader_destruct_helper
            {
                // cause the thread pool to begin its destruction process when
                // global objects start to be destroyed
                ~threader_destruct_helper()
                {
                    thread_pool().destruct_when_ready();
                }
            };
            static threader_destruct_helper a;

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

            bool threader::
            is_dlib_thread (
                thread_id_type id
            )
            {
                auto_mutex M(data_mutex);
                return thread_ids.is_member(id);
            }

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

            threader::
            threader (
            ) :
                total_count(0),
                function_pointer(0),
                pool_count(0),
                data_ready(data_mutex),
                data_empty(data_mutex),
                destruct(false),
                destructed(data_mutex),
                should_destruct(false)
            {}

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

            threader::
            ~threader (
            )
            {
                data_mutex.lock();
                destruct = true;
                data_ready.broadcast();

                // wait for all the threads to end
                while (total_count > 0)
                    destructed.wait();
                data_mutex.unlock();
            }

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

            void threader::
            destruct_when_ready (
            )
            {
                data_mutex.lock();

                // if there aren't any active threads, just maybe some sitting around
                // in the pool then just destroy the threader
                if (total_count == pool_count)
                {
                    data_mutex.unlock();
                    delete this;
                }
                else
                {
                    // in this case we just let the thread pool know that it
                    // should self destruct whenever it gets a chance
                    should_destruct = true;
                    data_mutex.unlock();
                }
            }

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

            void threader::
            call_end_handlers (
            )
            {
                reg.m.lock();
                const thread_id_type id = get_thread_id();
                thread_id_type id_copy;
                unsigned long count = reg.reg.count(id);
                member_function_pointer<>::kernel_1a mfp;

                // Remove all the member function pointers for this thread from the tree
                // and call them.
                for (unsigned long i = 0; i < count; ++i)
                {
                    reg.reg.remove(id,id_copy,mfp);
                    reg.m.unlock();
                    mfp();
                    reg.m.lock();
                }
                reg.m.unlock();
            }

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

            bool threader::
            create_new_thread (
                void (*funct)(void*),
                void* param
            )
            {

                // get a lock on the data mutex
                auto_mutex M(data_mutex);

                // loop to ensure that the new function poitner is in the data
                while (true)
                {
                    // if the data is empty then add new data and quit loop
                    if (function_pointer == 0)
                    {
                        parameter = param;
                        function_pointer = funct;
                        break;
                    }
                    else
                    {
                        // wait for data to become empty
                        data_empty.wait();
                    }
                }

                // get a thread for this new data
                // if a new thread must be crated
                if (pool_count == 0)
                {
                    // make thread and add it to the pool
                    if ( threads_kernel_shared_helpers::spawn_thread(thread_starter, this) == false )
                    {
                        function_pointer = 0;
                        parameter = 0;
                        data_empty.signal();
                        return false;
                    }
                    ++total_count;
                }
                // wake up a thread from the pool
                else
                {
                    data_ready.signal();
                }

                return true;
            }

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

            void thread_starter (
                void* object
            )
            {
                // get a reference to the calling threader object
                threader& self = *reinterpret_cast<threader*>(object);

                bool should_destroy_threader = false;

                {
                auto_mutex M(self.data_mutex);

                // add this thread id
                thread_id_type thread_id = get_thread_id();
                self.thread_ids.add(thread_id);

                // indicate that this thread is now in the thread pool
                ++self.pool_count;

                while (true)
                {
                    // if data is ready then process it and launch the thread
                    // if its not ready then go back into the pool
                    while (self.function_pointer != 0)
                    {
                        // indicate that this thread is now out of the thread pool
                        --self.pool_count;

                        // get the data for the function call
                        void (*funct)(void*) = self.function_pointer;
                        void* param = self.parameter;
                        self.function_pointer = 0;

                        // signal that the data is now empty
                        self.data_empty.signal();

                        self.data_mutex.unlock();
                        // call funct with its intended parameter
                        try
                        {
                            funct(param);
                            self.call_end_handlers();
                        }
                        catch (std::exception& e)
                        {
                            std::cerr << "An exception was thrown in a thread and was not caught.  Its what() string is:\n"
                                 << e.what() << std::endl;

                            self.data_mutex.lock();
                            --self.total_count;
                            self.destructed.signal();
                            self.data_mutex.unlock();

                            abort();
                        }
                        catch (...)
                        {
                            std::cerr << "An exception was thrown in a thread and was not caught." << std::endl;

                            self.data_mutex.lock();
                            --self.total_count;
                            self.destructed.signal();
                            self.data_mutex.unlock();

                            abort();
                        }

                        self.data_mutex.lock();

                        // indicate that this thread is now back in the thread pool
                        ++self.pool_count;
                    }

                    if (self.destruct == true || self.should_destruct == true)
                        break;

                    // if we timed out and there isn't any work to do then
                    // this thread will quit this loop and end.
                    if (self.data_ready.wait_or_timeout(DLIB_THREAD_POOL_TIMEOUT) == false &&
                        self.function_pointer == 0)
                        break;

                }

                // remove this thread id from thread_ids
                thread_id = get_thread_id();
                self.thread_ids.destroy(thread_id);

                // indicate that this thread is now out of the thread pool
                --self.pool_count;
                --self.total_count;

                self.destructed.signal();

                if (self.should_destruct && self.total_count == 0)
                    should_destroy_threader = true;

                } // end of auto_mutex M(self.data_mutex) block

                if (should_destroy_threader)
                    delete &self;
            }

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

        }

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

        bool is_dlib_thread (
            thread_id_type id
        )
        {
            return threads_kernel_shared::thread_pool().is_dlib_thread(id);
        }

        bool is_dlib_thread (
        )
        {
            return is_dlib_thread(get_thread_id());
        }

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

    }

    #endif // DLIB_THREADS_KERNEL_SHARED_CPp_