C:/dev/de/differentialevolution/processors.hpp

/*
 * Copyright (c) 2011 Adrian Michel
 * http://www.amichel.com
 *
 * Permission to use, copy, modify, distribute and sell this 
 * software and its documentation for any purpose is hereby 
 * granted without fee, provided that both the above copyright 
 * notice and this permission notice appear in all copies and in 
 * the supporting documentation. 
 *  
 * This library is distributed in the hope that it will be 
 * useful. However, Adrian Michel makes no representations about
 * the suitability of this software for any purpose.  It is 
 * provided "as is" without any express or implied warranty. 
 * 
 * Should you find this library useful, please email 
 * info@amichel.com with a link or other reference 
 * to your work. 
*/

#ifndef DE_PROCESSORS_HPP_INCLUDED
#define DE_PROCESSORS_HPP_INCLUDED

// MS compatible compilers support #pragma once

#if defined(_MSC_VER) && (_MSC_VER >= 1020)
#pragma once
#endif

#include <queue>
#include <boost/thread.hpp>
#include <boost/scope_exit.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/utility.hpp>

#include "individual.hpp"
#include "population.hpp"

namespace de
{

class processor_listener
{
public:
        virtual ~processor_listener(){}
        virtual void start( size_t index ) = 0;
        virtual void start_of( size_t index, individual_ptr individual ) = 0;
        virtual void end_of( size_t index, individual_ptr individual ) = 0;
        virtual void end( size_t index ) = 0;
        virtual void error( size_t index, const std::string& message ) = 0;

};

class null_processor_listener : public processor_listener
{
public:
        virtual void start(size_t index)
        {
        }
        virtual void start_of( size_t index, individual_ptr individual )
        {
        }
        virtual void end_of( size_t index, individual_ptr individual )
        {
        }
        virtual void end(size_t index)
        {
        }
        virtual void error(size_t index, const std::string& message )
        {
        }
};


typedef boost::shared_ptr< processor_listener > processor_listener_ptr;

class objective_function_exception : public exception
{
public:
        objective_function_exception( const std::string& message )
        : exception( message.c_str() )
        {
        }
};

template< typename T >class objective_function_factory
{
public:
        typedef boost::shared_ptr< T > T_ptr;

        virtual ~objective_function_factory(){}

        virtual T_ptr make() = 0;
};

template< typename T >class processor_traits
{
public:
        // \cond
        typedef T  value_type;
        static double run( T t, de::DVectorPtr vars ) { return t( vars); }
        static T make( T t ) { return t; }
        // \endcond
};

template< typename T > class processor_traits< T* >
{
public:
        // \cond
        typedef T*  value_type;
        static double run( value_type t, de::DVectorPtr vars ) { return (*t)( vars ); }
        static value_type make( value_type t ) { return t; }
        // \endcond
};

template< typename T > class processor_traits< boost::shared_ptr< T > >
{
public:
        // \cond
        typedef boost::shared_ptr< T > value_type;
        static double run( value_type t, de::DVectorPtr vars ) { return (*t)( vars); }
        static value_type make( value_type t ) { return t; }
        // \endcond
};

template< typename T > class processor_traits< objective_function_factory< T >* >
{
public:
        // \cond
        typedef boost::shared_ptr< T > value_type;
        static double run( value_type t, de::DVectorPtr vars ) { return (*t)( vars); }
        static value_type make( objective_function_factory< T >* off ) { return off->make(); }
        // \endcond
};
template< typename T > class processor_traits< boost::shared_ptr< objective_function_factory< T > > >
{
public:
        // \cond
        typedef boost::shared_ptr< T > value_type;
        static double run( value_type t, de::DVectorPtr vars ) { return (*t)( vars); }
        static value_type make( boost::shared_ptr< objective_function_factory< T > > off ) { return off->make(); }
        // \endcond
};

template< typename T > class processor_traits< objective_function_factory< T >& >
{
public:
        // \cond
        typedef boost::shared_ptr< T > value_type;
        static double run( value_type t, de::DVectorPtr vars ) { return (*t)( vars ); }
        static value_type make( objective_function_factory< T >& off ) { return off.make(); }
        // \endcond
};

template < typename T > class processor : boost::noncopyable
{
private:
        typename processor_traits< T >::value_type m_of;
        individual_queue& m_indQueue;
        processor_listener_ptr m_listener;
        size_t m_index;

        bool m_result;

public:
        processor( size_t index, T of, individual_queue& indQueue, processor_listener_ptr listener )
        : m_of( processor_traits< T >::make( of ) ), m_indQueue( indQueue ), m_result( false ), m_listener( listener ), m_index( index )
        {
                assert( listener );
        }

        void operator()()
        {
                m_listener->start( m_index );
                m_result = false;
                try
                {
                        for( individual_ptr ind = m_indQueue.pop(); ind; ind = m_indQueue.pop() )
                        {
                                m_listener->start_of( m_index, ind );
                                double result = processor_traits< T >::run( m_of, ind->vars() );

                                ind->setCost( result );
                                m_listener->end_of( m_index, ind );
                        }
                        m_result = true;

                        BOOST_SCOPE_EXIT_TPL( (&m_index) (&m_listener) )
                        {
                                m_listener->end( m_index );
                        } 
                        BOOST_SCOPE_EXIT_END
                }
                catch( const objective_function_exception& e )
                {
                        m_result = false;
                        m_listener->error( m_index, e.what() );
                }
        }

        bool success() const { return m_result; }
};



class processors_exception : exception
{
public:
        processors_exception( const std::string& message )
        : exception( message.c_str() )
        {
        }
};


template< typename T > class processors
{
private:
        typedef boost::shared_ptr< boost::thread_group > thread_group_ptr;
        typedef boost::shared_ptr< processor< T > > processor_ptr;
        typedef std::vector< processor_ptr > processor_vector;
        typedef boost::shared_ptr< T > T_ptr;

private:
        individual_queue m_indQueue;
        processor_vector m_processors;
        thread_group_ptr m_threads;

public:
        processors( size_t count, T of, processor_listener_ptr listener )
        {
                assert( count > 0 );
                assert( listener );

                for( size_t n = 0; n < count; ++n )
                {
                        processor_ptr processor( boost::make_shared< processor< T > >( n, of, boost::ref( m_indQueue ), listener ) );
                        m_processors.push_back( processors< T >::processor_ptr( processor ) ) ;
                }
        }

        void push( individual_ptr ind ) { m_indQueue.push( ind ); }
        void start()
        {
                // create a new group every time, don't bother removing all individual threads
                m_threads = boost::make_shared< boost::thread_group >();

                for( typename processor_vector::size_type n = 0; n < m_processors.size(); ++n )
                {
                        processor_ptr p( m_processors[ n  ] );
                        boost::thread* th( new boost::thread( boost::ref( *p ) ) );
                        m_threads->add_thread( th );
                }
        }

        void wait()
        {
                m_threads->join_all();

                if( !m_indQueue.empty() )
                        throw processors_exception( "threads ended before emptying the queue");

                if( !success() )
                        throw processors_exception( "objective function error");
        }

        bool success()
        {
                for( typename processor_vector::size_type n = 0; n < m_processors.size(); ++n )
                {
                        processor_ptr processor( m_processors[ n  ] );
                        if( !processor->success() )
                                return false;
                }

                return true;
        }

        void push( population_ptr population )
        {
                std::copy( population->begin(), population->end(), std::back_inserter( m_indQueue ) );
        }

typedef boost::shared_ptr< processors< T > > processors_ptr;

};

}

#endif //DE_PROCESSORS_HPP_INCLUDED